Handle ACP and P2P

sourcenetwork · Jul 18, 2024 · fb4e41f · fb4e41f
1 parent 80c7c98
commit fb4e41f
Show file tree

Hide file tree

Showing 8 changed files with 259 additions and 45 deletions.
diff --git a/acp/README.md b/acp/README.md
@@ -458,9 +458,10 @@ If authentication fails for any reason a `403` forbidden response will be return
 ## _FAC Usage: (coming soon)_
 
 ## Warning / Caveats
+- If using Local ACP, P2P will only work with collections that do not have a policy assigned.  If you wish to use ACP
+on collections connected to a multi-node network, please use SourceHub ACP.
+
 The following features currently don't work with ACP, they are being actively worked on.
-- [P2P: Adding a replicator with permissioned collection](https://github.com/sourcenetwork/defradb/issues/2366)
-- [P2P: Subscription to a permissioned collection](https://github.com/sourcenetwork/defradb/issues/2366)
 - [Adding Secondary Indexes](https://github.com/sourcenetwork/defradb/issues/2365)
 - [Backing/Restoring Private Documents](https://github.com/sourcenetwork/defradb/issues/2430)
 

diff --git a/acp/acp.go b/acp/acp.go
@@ -98,4 +98,7 @@ type ACP interface {
 		resourceName string,
 		docID string,
 	) (bool, error)
+
+	// SupportsP2P returns true if the implementation supports ACP across a peer network.
+	SupportsP2P() bool
 }
diff --git a/acp/source_hub_client.go b/acp/source_hub_client.go
@@ -335,6 +335,11 @@ func (a *sourceHubBridge) CheckDocAccess(
 	}
 }
 
+func (a *sourceHubBridge) SupportsP2P() bool {
+	_, ok := a.client.(*acpSourceHub)
+	return ok
+}
+
 func (a *sourceHubBridge) Close() error {
 	return a.client.Close()
 }
diff --git a/internal/db/p2p_replicator.go b/internal/db/p2p_replicator.go
@@ -45,8 +45,6 @@ func (db *db) SetReplicator(ctx context.Context, rep client.Replicator) error {
 		return ErrSelfTargetForReplicator
 	}
 
-	// TODO-ACP: Support ACP <> P2P - https://github.com/sourcenetwork/defradb/issues/2366
-	// ctx = db.SetContextIdentity(ctx, identity)
 	ctx = SetContextTxn(ctx, txn)
 
 	storedRep := client.Replicator{}
@@ -90,20 +88,19 @@ func (db *db) SetReplicator(ctx context.Context, rep client.Replicator) error {
 		}
 
 	default:
-		// default to all collections (unless a collection contains a policy).
-		// TODO-ACP: default to all collections after resolving https://github.com/sourcenetwork/defradb/issues/2366
 		allCollections, err := db.GetCollections(ctx, client.CollectionFetchOptions{})
 		if err != nil {
 			return NewErrReplicatorCollections(err)
 		}
 
-		for _, col := range allCollections {
-			// Can not default to all collections if any collection has a policy.
-			// TODO-ACP: remove this check/loop after https://github.com/sourcenetwork/defradb/issues/2366
-			if col.Description().Policy.HasValue() {
-				return ErrReplicatorSomeColsHavePolicy
+		if db.acp.HasValue() && !db.acp.Value().SupportsP2P() {
+			for _, col := range allCollections {
+				if col.Description().Policy.HasValue() {
+					return ErrReplicatorSomeColsHavePolicy
+				}
 			}
 		}
+
 		collections = allCollections
 	}
 

diff --git a/internal/db/p2p_schema_root.go b/internal/db/p2p_schema_root.go
@@ -31,8 +31,6 @@ func (db *db) AddP2PCollections(ctx context.Context, collectionIDs []string) err
 	}
 	defer txn.Discard(ctx)
 
-	// TODO-ACP: Support ACP <> P2P - https://github.com/sourcenetwork/defradb/issues/2366
-	// ctx = db.SetContextIdentity(ctx, identity)
 	ctx = SetContextTxn(ctx, txn)
 
 	// first let's make sure the collections actually exists
@@ -53,11 +51,11 @@ func (db *db) AddP2PCollections(ctx context.Context, collectionIDs []string) err
 		storeCollections = append(storeCollections, storeCol...)
 	}
 
-	// Ensure none of the collections have a policy on them, until following is implemented:
-	// TODO-ACP: ACP <> P2P https://github.com/sourcenetwork/defradb/issues/2366
-	for _, col := range storeCollections {
-		if col.Description().Policy.HasValue() {
-			return ErrP2PColHasPolicy
+	if db.acp.HasValue() && !db.acp.Value().SupportsP2P() {
+		for _, col := range storeCollections {
+			if col.Description().Policy.HasValue() {
+				return ErrP2PColHasPolicy
+			}
 		}
 	}
 
@@ -98,8 +96,6 @@ func (db *db) RemoveP2PCollections(ctx context.Context, collectionIDs []string)
 	}
 	defer txn.Discard(ctx)
 
-	// TODO-ACP: Support ACP <> P2P - https://github.com/sourcenetwork/defradb/issues/2366
-	// ctx = db.SetContextIdentity(ctx, identity)
 	ctx = SetContextTxn(ctx, txn)
 
 	// first let's make sure the collections actually exists
@@ -211,7 +207,6 @@ func (db *db) loadAndPublishP2PCollections(ctx context.Context) error {
 		if _, ok := colMap[col.SchemaRoot()]; ok {
 			continue
 		}
-		// TODO-ACP: Support ACP <> P2P - https://github.com/sourcenetwork/defradb/issues/2366
 		docIDChan, err := col.GetAllDocIDs(ctx)
 		if err != nil {
 			return err

diff --git a/internal/db/p2p_schema_root_test.go b/internal/db/p2p_schema_root_test.go
@@ -249,9 +249,6 @@ func TestGetAllP2PCollections_WithMultipleValidCollections_ShouldSucceed(t *test
 	require.Equal(t, []string{schema2.Root, schema1.Root}, cols)
 }
 
-// This test documents that we don't allow adding p2p collections that have a policy
-// until the following is implemented:
-// TODO-ACP: ACP <> P2P https://github.com/sourcenetwork/defradb/issues/2366
 func TestAddP2PCollectionsWithPermissionedCollection_Error(t *testing.T) {
 	ctx := context.Background()
 	rootstore := memory.NewDatastore(ctx)

diff --git a/tests/integration/acp/p2p/replicator_test.go b/tests/integration/acp/p2p/replicator_test.go
@@ -18,23 +18,18 @@ import (
 	testUtils "github.com/sourcenetwork/defradb/tests/integration"
 )
 
-// This test documents that we don't allow setting replicator with a collections that has a policy
-// until the following is implemented:
-// TODO-ACP: ACP <> P2P https://github.com/sourcenetwork/defradb/issues/2366
-func TestACP_P2POneToOneReplicatorWithPermissionedCollection_Error(t *testing.T) {
+func TestACP_P2POneToOneReplicatorWithPermissionedCollection_LocalACP(t *testing.T) {
 	test := testUtils.TestCase{
-
-		Description: "Test acp, with p2p replicator with permissioned collection, error",
-
+		SupportedACPTypes: immutable.Some(
+			[]testUtils.ACPType{
+				testUtils.LocalACPType,
+			},
+		),
 		Actions: []any{
-
 			testUtils.RandomNetworkingConfig(),
 			testUtils.RandomNetworkingConfig(),
-
 			testUtils.AddPolicy{
-
 				Identity: immutable.Some(1),
-
 				Policy: `
                     name: test
                     description: a test policy which marks a collection in a database as a resource
@@ -63,10 +58,8 @@ func TestACP_P2POneToOneReplicatorWithPermissionedCollection_Error(t *testing.T)
                             types:
                               - actor
                 `,
-
 				ExpectedPolicyID: "94eb195c0e459aa79e02a1986c7e731c5015721c18a373f2b2a0ed140a04b454",
 			},
-
 			testUtils.SchemaUpdate{
 				Schema: `
 					type Users @policy(
@@ -78,7 +71,6 @@ func TestACP_P2POneToOneReplicatorWithPermissionedCollection_Error(t *testing.T)
 					}
 				`,
 			},
-
 			testUtils.ConfigureReplicator{
 				SourceNodeID:  0,
 				TargetNodeID:  1,
@@ -89,3 +81,113 @@ func TestACP_P2POneToOneReplicatorWithPermissionedCollection_Error(t *testing.T)
 
 	testUtils.ExecuteTestCase(t, test)
 }
+
+func TestACP_P2POneToOneReplicatorWithPermissionedCollection_SourceHubACP(t *testing.T) {
+	test := testUtils.TestCase{
+		SupportedACPTypes: immutable.Some(
+			[]testUtils.ACPType{
+				testUtils.SourceHubACPType,
+			},
+		),
+		Actions: []any{
+			testUtils.RandomNetworkingConfig(),
+			testUtils.RandomNetworkingConfig(),
+			testUtils.AddPolicy{
+				Identity: immutable.Some(1),
+				Policy: `
+                    name: test
+                    description: a test policy which marks a collection in a database as a resource
+
+                    actor:
+                      name: actor
+
+                    resources:
+                      users:
+                        permissions:
+                          read:
+                            expr: owner + reader
+                          write:
+                            expr: owner
+
+                        relations:
+                          owner:
+                            types:
+                              - actor
+                          reader:
+                            types:
+                              - actor
+                          admin:
+                            manages:
+                              - reader
+                            types:
+                              - actor
+                `,
+				ExpectedPolicyID: "94eb195c0e459aa79e02a1986c7e731c5015721c18a373f2b2a0ed140a04b454",
+			},
+			testUtils.SchemaUpdate{
+				Schema: `
+					type Users @policy(
+						id: "94eb195c0e459aa79e02a1986c7e731c5015721c18a373f2b2a0ed140a04b454",
+						resource: "users"
+					) {
+						name: String
+						age: Int
+					}
+				`,
+			},
+			testUtils.ConfigureReplicator{
+				SourceNodeID: 0,
+				TargetNodeID: 1,
+			},
+			testUtils.CreateDoc{
+				NodeID:   immutable.Some(0),
+				Identity: immutable.Some(1),
+				DocMap: map[string]any{
+					"name": "John",
+				},
+			},
+			testUtils.WaitForSync{},
+			testUtils.Request{
+				// Ensure that the document is accessible on all nodes to authorized actors
+				Identity: immutable.Some(1),
+				Request: `
+					query {
+						Users {
+							name
+						}
+					}
+				`,
+				Results: []map[string]any{
+					{
+						"name": "John",
+					},
+				},
+			},
+			testUtils.Request{
+				// Ensure that the document is hidden on all nodes to unidentified actors
+				Request: `
+					query {
+						Users {
+							name
+						}
+					}
+				`,
+				Results: []map[string]any{},
+			},
+			testUtils.Request{
+				// Ensure that the document is hidden on all nodes to unauthorized actors
+				Identity: immutable.Some(2),
+				Request: `
+					query {
+						Users {
+							name
+						}
+					}
+				`,
+				Results: []map[string]any{},
+			},
+		},
+	}
+
+	testUtils.ExecuteTestCase(t, test)
+}