Updates and Deletes protected

Author: matthewmcneely

edgraph/server.go

@@ -16,6 +16,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 	"unicode"
@@ -573,6 +574,11 @@ func (s *Server) doMutate(ctx context.Context, qc *queryContext, resp *api.Respo
 	if err != nil {
 		return errors.Wrapf(err, "While doing mutations:")
 	}
+
+	// Check program authorization for modifying existing data (after namespace is available)
+	if err := checkMutationProgramAuth(ctx, edges, ns); err != nil {
+		return err
+	}
 	predHints := make(map[string]pb.Metadata_HintType)
 	for _, gmu := range qc.gmuList {
 		for pred, hint := range gmu.Metadata.GetPredHints() {
@@ -660,6 +666,10 @@ func (s *Server) doMutate(ctx context.Context, qc *queryContext, resp *api.Respo
 	resp.Txn.Keys = resp.Txn.Keys[:0]
 	resp.Txn.CommitTs = cts
 	calculateMutationMetrics()
+
+	// Update the program facet cache for future authorization checks
+	updateProgramFacetCache(ns, edges)
+
 	return nil
 }
 
@@ -2245,6 +2255,148 @@ func parseSubject(predSubject string) (uint64, error) {
 	}
 }
 
+// programFacetCache stores recently committed program facets for authorization checks.
+//
+// WHY THIS EXISTS:
+// Dgraph's MemoryLayer is a read-through cache - it only populates on reads, not writes.
+// When a mutation commits, the data goes to badger asynchronously. The MemoryLayer's
+// updateItemInCache() only updates keys that are ALREADY cached; it doesn't add new ones.
+// (See posting/mvcc.go:589-591)
+//
+// For program authorization, we need to check if a user can modify existing data BEFORE
+// the Raft proposal. But for recently created data, the MemoryLayer won't have it cached
+// and badger won't have synced it yet, so GetNoStoreSafe returns 0 postings.
+//
+// This cache provides write-through semantics specifically for program facets:
+// - Updated after successful CommitOverNetwork in doMutate()
+// - Checked in checkMutationProgramAuth() before the posting store fallback
+//
+// FUTURE CONSIDERATION:
+// A more holistic approach might store program data directly in the Posting protobuf
+// rather than as facets, which would integrate naturally with the existing caching.
+// However, that would require protobuf changes and migration strategy.
+var programFacetCache = struct {
+	sync.RWMutex
+	// Key: "namespace-predicate-entity", Value: comma-separated programs
+	data map[string]string
+}{data: make(map[string]string)}
+
+func programFacetCacheKey(ns uint64, attr string, entity uint64) string {
+	return fmt.Sprintf("%d-%s-%d", ns, attr, entity)
+}
+
+// updateProgramFacetCache updates the cache with program facets from committed mutations.
+func updateProgramFacetCache(ns uint64, edges []*pb.DirectedEdge) {
+	programFacetCache.Lock()
+	defer programFacetCache.Unlock()
+
+	for _, edge := range edges {
+		if edge.Entity == 0 {
+			continue
+		}
+		for _, f := range edge.Facets {
+			if f.Key == x.ProgramFacetKey {
+				key := programFacetCacheKey(ns, edge.Attr, edge.Entity)
+				programFacetCache.data[key] = string(f.Value)
+				glog.V(1).Infof("updateProgramFacetCache: cached programs=%s for key=%s", f.Value, key)
+				break
+			}
+		}
+	}
+}
+
+// checkMutationProgramAuth verifies that the user is authorized to modify existing data
+// protected by program facets. This check happens before Raft proposal.
+func checkMutationProgramAuth(ctx context.Context, edges []*pb.DirectedEdge, ns uint64) error {
+	authCtx := auth.ExtractOrNil(ctx)
+	if authCtx == nil || authCtx.IsNil {
+		// No auth context means no program restrictions - allow mutation
+		glog.V(1).Infof("checkMutationProgramAuth: no auth context, allowing mutation")
+		return nil
+	}
+
+	glog.V(1).Infof("checkMutationProgramAuth: auth context has programs=%v", authCtx.Programs)
+
+	// Build a set of user's programs for fast lookup
+	userPrograms := make(map[string]bool)
+	for _, p := range authCtx.Programs {
+		userPrograms[p] = true
+	}
+
+	for _, edge := range edges {
+		// Only check existing entities (non-zero UID)
+		if edge.Entity == 0 {
+			continue
+		}
+
+		// First check the in-memory cache for recently committed program facets
+		cacheKey := programFacetCacheKey(ns, edge.Attr, edge.Entity)
+		programFacetCache.RLock()
+		cachedPrograms, found := programFacetCache.data[cacheKey]
+		programFacetCache.RUnlock()
+
+		if found {
+			glog.V(1).Infof("checkMutationProgramAuth: found cached programs=%s for Entity=%d, Attr=%s",
+				cachedPrograms, edge.Entity, edge.Attr)
+			programs := strings.Split(cachedPrograms, ",")
+			hasMatch := false
+			for _, prog := range programs {
+				if userPrograms[prog] {
+					hasMatch = true
+					break
+				}
+			}
+			if !hasMatch && len(programs) > 0 && programs[0] != "" {
+				glog.V(1).Infof("checkMutationProgramAuth: DENYING - no program match (from cache)")
+				return errors.Errorf("not authorized to modify program-protected data")
+			}
+			continue // Authorized via cache, move to next edge
+		}
+
+		// Fall back to reading from posting store for older data
+		namespacedAttr := x.NamespaceAttr(ns, edge.Attr)
+		key := x.DataKey(namespacedAttr, edge.Entity)
+		glog.V(1).Infof("checkMutationProgramAuth: checking store for Entity=%d, Attr=%s", edge.Entity, edge.Attr)
+
+		pl, err := posting.GetNoStoreSafe(key, math.MaxUint64)
+		if err != nil || pl == nil {
+			continue
+		}
+
+		// Check each existing posting for program authorization
+		var authErr error
+		pl.Iterate(math.MaxUint64, 0, func(p *pb.Posting) error {
+			for _, f := range p.Facets {
+				if f.Key == x.ProgramFacetKey {
+					programs := strings.Split(string(f.Value), ",")
+					glog.V(1).Infof("checkMutationProgramAuth: store has programs=%v, user has=%v", programs, authCtx.Programs)
+					hasMatch := false
+					for _, prog := range programs {
+						if userPrograms[prog] {
+							hasMatch = true
+							break
+						}
+					}
+					if !hasMatch && len(programs) > 0 && programs[0] != "" {
+						glog.V(1).Infof("checkMutationProgramAuth: DENYING - no program match (from store)")
+						authErr = errors.Errorf("not authorized to modify program-protected data")
+						return posting.ErrStopIteration
+					}
+					break
+				}
+			}
+			return nil
+		})
+
+		if authErr != nil {
+			return authErr
+		}
+	}
+
+	glog.V(1).Infof("checkMutationProgramAuth: allowing mutation")
+	return nil
+}
+
 // injectProgramFacets adds the dgraph.programs facet to edges based on the auth context.
 // If the user has programs in their auth context, those programs are attached to each edge
 // being mutated. This enables server-side filtering during queries.

posting/lists.go

@@ -57,6 +57,15 @@ func GetNoStore(key []byte, readTs uint64) (rlist *List, err error) {
 	return getNew(key, pstore, readTs, false)
 }
 
+// GetNoStoreSafe is like GetNoStore but returns nil if the store is not initialized.
+// This is safe to call even before the posting package is fully initialized.
+func GetNoStoreSafe(key []byte, readTs uint64) (rlist *List, err error) {
+	if pstore == nil {
+		return nil, nil
+	}
+	return getNew(key, pstore, readTs, false)
+}
+
 // LocalCache stores a cache of posting lists and deltas.
 // This doesn't sync, so call this only when you don't care about dirty posting lists in
 // memory(for example before populating snapshot) or after calling syncAllMarks

systest/label/label_auth_test.go

@@ -833,6 +833,149 @@ func TestProgramAuthUidTraversal(t *testing.T) {
 	t.Log("Test passed: UID traversal respects program auth")
 }
 
+// TestProgramAuthUpdateBlocked tests that users cannot update program-protected data they don't have access to.
+func TestProgramAuthUpdateBlocked(t *testing.T) {
+	t.Log("=== TestProgramAuthUpdateBlocked: Users cannot update data they can't access ===")
+	setupProgramAuthTest(t)
+	dg := waitForCluster(t)
+
+	// Insert data with ALPHA program
+	ctxAlpha := auth.AttachToOutgoingContext(context.Background(), &auth.AuthContext{
+		Programs: []string{"ALPHA"},
+	})
+	resp, err := dg.NewTxn().Mutate(ctxAlpha, &api.Mutation{
+		CommitNow: true,
+		SetNquads: []byte(`_:p1 <name> "Alpha Secret" .`),
+	})
+	require.NoError(t, err)
+	uid := resp.Uids["p1"]
+	t.Logf("Created ALPHA-protected node with UID: %s", uid)
+
+	// Try to update with BRAVO program - should fail
+	ctxBravo := auth.AttachToOutgoingContext(context.Background(), &auth.AuthContext{
+		Programs: []string{"BRAVO"},
+	})
+	t.Log("Attempting to update ALPHA data with BRAVO credentials...")
+	_, err = dg.NewTxn().Mutate(ctxBravo, &api.Mutation{
+		CommitNow: true,
+		SetNquads: []byte(fmt.Sprintf(`<%s> <name> "Hacked by BRAVO" .`, uid)),
+	})
+	require.Error(t, err, "BRAVO user should not be able to update ALPHA data")
+	require.Contains(t, err.Error(), "not authorized", "error should indicate authorization failure")
+	t.Log("Test passed: update correctly blocked")
+}
+
+// TestProgramAuthDeleteBlocked tests that users cannot delete program-protected data they don't have access to.
+func TestProgramAuthDeleteBlocked(t *testing.T) {
+	t.Log("=== TestProgramAuthDeleteBlocked: Users cannot delete data they can't access ===")
+	setupProgramAuthTest(t)
+	dg := waitForCluster(t)
+
+	// Insert data with ALPHA program
+	ctxAlpha := auth.AttachToOutgoingContext(context.Background(), &auth.AuthContext{
+		Programs: []string{"ALPHA"},
+	})
+	resp, err := dg.NewTxn().Mutate(ctxAlpha, &api.Mutation{
+		CommitNow: true,
+		SetNquads: []byte(`_:p1 <name> "Alpha Secret" .`),
+	})
+	require.NoError(t, err)
+	uid := resp.Uids["p1"]
+	t.Logf("Created ALPHA-protected node with UID: %s", uid)
+
+	// Try to delete with BRAVO program - should fail
+	ctxBravo := auth.AttachToOutgoingContext(context.Background(), &auth.AuthContext{
+		Programs: []string{"BRAVO"},
+	})
+	t.Log("Attempting to delete ALPHA data with BRAVO credentials...")
+	_, err = dg.NewTxn().Mutate(ctxBravo, &api.Mutation{
+		CommitNow: true,
+		DelNquads: []byte(fmt.Sprintf(`<%s> <name> * .`, uid)),
+	})
+	require.Error(t, err, "BRAVO user should not be able to delete ALPHA data")
+	require.Contains(t, err.Error(), "not authorized", "error should indicate authorization failure")
+
+	// Verify data still exists with ALPHA credentials
+	queryResp, err := dg.NewTxn().Query(ctxAlpha, fmt.Sprintf(`
+		{
+			node(func: uid(%s)) {
+				name
+			}
+		}
+	`, uid))
+	require.NoError(t, err)
+	require.Contains(t, string(queryResp.GetJson()), "Alpha Secret", "data should still exist")
+	t.Log("Test passed: delete correctly blocked")
+}
+
+// TestProgramAuthUpdateAllowed tests that authorized users can update their own data.
+func TestProgramAuthUpdateAllowed(t *testing.T) {
+	t.Log("=== TestProgramAuthUpdateAllowed: Users can update data they have access to ===")
+	setupProgramAuthTest(t)
+	dg := waitForCluster(t)
+
+	// Insert data with ALPHA program
+	ctxAlpha := auth.AttachToOutgoingContext(context.Background(), &auth.AuthContext{
+		Programs: []string{"ALPHA"},
+	})
+	resp, err := dg.NewTxn().Mutate(ctxAlpha, &api.Mutation{
+		CommitNow: true,
+		SetNquads: []byte(`_:p1 <name> "Original Name" .`),
+	})
+	require.NoError(t, err)
+	uid := resp.Uids["p1"]
+
+	// Update with same ALPHA program - should succeed
+	t.Log("Updating with ALPHA credentials...")
+	_, err = dg.NewTxn().Mutate(ctxAlpha, &api.Mutation{
+		CommitNow: true,
+		SetNquads: []byte(fmt.Sprintf(`<%s> <name> "Updated Name" .`, uid)),
+	})
+	require.NoError(t, err, "ALPHA user should be able to update ALPHA data")
+
+	// Verify update
+	queryResp, err := dg.NewTxn().Query(ctxAlpha, fmt.Sprintf(`
+		{
+			node(func: uid(%s)) {
+				name
+			}
+		}
+	`, uid))
+	require.NoError(t, err)
+	require.Contains(t, string(queryResp.GetJson()), "Updated Name")
+	t.Log("Test passed: authorized update succeeded")
+}
+
+// TestProgramAuthNoAuthCannotModifyProtected tests that users without auth cannot modify protected data.
+func TestProgramAuthNoAuthCannotModifyProtected(t *testing.T) {
+	t.Log("=== TestProgramAuthNoAuthCannotModifyProtected: No-auth users cannot modify protected data ===")
+	setupProgramAuthTest(t)
+	dg := waitForCluster(t)
+
+	// Insert data with ALPHA program
+	ctxAlpha := auth.AttachToOutgoingContext(context.Background(), &auth.AuthContext{
+		Programs: []string{"ALPHA"},
+	})
+	resp, err := dg.NewTxn().Mutate(ctxAlpha, &api.Mutation{
+		CommitNow: true,
+		SetNquads: []byte(`_:p1 <name> "Protected Data" .`),
+	})
+	require.NoError(t, err)
+	uid := resp.Uids["p1"]
+
+	// Try to update with empty auth context (auth active but no programs)
+	ctxEmpty := auth.AttachToOutgoingContext(context.Background(), &auth.AuthContext{
+		Programs: []string{},
+	})
+	t.Log("Attempting to update protected data with empty auth...")
+	_, err = dg.NewTxn().Mutate(ctxEmpty, &api.Mutation{
+		CommitNow: true,
+		SetNquads: []byte(fmt.Sprintf(`<%s> <name> "Attempted Overwrite" .`, uid)),
+	})
+	require.Error(t, err, "user with no programs should not be able to modify protected data")
+	t.Log("Test passed: empty auth cannot modify protected data")
+}
+
 // TestProgramAuthComparisonOps tests program auth with comparison operators (lt, gt, le, ge).
 func TestProgramAuthComparisonOps(t *testing.T) {
 	t.Log("=== TestProgramAuthComparisonOps: comparison operators should respect program auth ===")