feat(edgraph): add an AccessController and separate identity from tenancy

audit/interceptor.go

@@ -175,6 +175,18 @@ func auditGrpc(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo)
 	var namespace uint64
 	var err error
 	extractUser := func(md metadata.MD) {
+		// Prefer the identity the interceptor already verified. Without this,
+		// auditing a request re-parses and re-verifies its access token, a second
+		// full JWT verification per RPC purely to recover a username the identity
+		// interceptor has already extracted.
+		//
+		// The fallback below is unchanged and still covers every case where no
+		// Principal exists: poor-man's auth, a token that failed to verify
+		// (UnknownUser), and no credential at all (UnauthorisedUser under ACL).
+		if p := x.PrincipalFrom(ctx); p != nil {
+			user = p.Subject
+			return
+		}
 		if t := md.Get("accessJwt"); len(t) > 0 {
 			user = getUser(t[0], false)
 		} else if t := md.Get("auth-token"); len(t) > 0 {
@@ -230,9 +242,24 @@ func auditHttp(w *ResponseWriter, r *http.Request) {
 		user = getUser("", false)
 	}
 
+	// Audit must never reject a request, so a resolver error becomes the explicit
+	// unknown sentinel rather than a failure.
+	//
+	// Worth knowing what this does and does not achieve today: the built-in resolver
+	// tolerates a token it cannot parse and reports no error, so a malformed
+	// credential is still recorded against the root namespace and this branch is
+	// reached only by an installed resolver that fails closed. Attributing a
+	// malformed token accurately needs the resolver to distinguish "no credential"
+	// from "unusable credential", which it cannot yet do: /admin and the login
+	// mutation legitimately carry no credential at all.
+	namespace, err := x.ResolveTenantHTTP(r)
+	if err != nil {
+		namespace = UnknownNamespace
+	}
+
 	auditor.Audit(&AuditEvent{
 		User:        user,
-		Namespace:   x.ExtractNamespaceHTTP(r),
+		Namespace:   namespace,
 		ServerHost:  x.WorkerConfig.MyAddr,
 		ClientHost:  r.RemoteAddr,
 		Endpoint:    r.URL.Path,

dgraph/cmd/alpha/hooks.go

@@ -21,6 +21,20 @@ var RegisterFlags = defaultRegisterFlags
 
 func defaultRegisterFlags(f *pflag.FlagSet) {}
 
+// ConfigureIdentity installs deployment-specific authentication and authorization
+// — an x.Authenticator, and any additional capability sources — from parsed
+// configuration. Called from run() after x.WorkerConfig.Parse and before any
+// listener starts serving, which is the only window where the config exists and
+// nothing is being served yet. Default: no-op (OSS builds).
+//
+// Separate from the service-registration hooks below on purpose. What it installs
+// is process-wide: it resolves the caller's identity for every API the Alpha
+// serves, so tying it to whether one optional service happens to be enabled would
+// misdescribe its reach.
+var ConfigureIdentity = defaultConfigureIdentity
+
+func defaultConfigureIdentity() {}
+
 // RegisterZanzibar wires the Zanzibar gRPC service onto s and bootstraps the
 // fixed predicate schema. Default: no-op (OSS builds).
 var RegisterZanzibar = defaultRegisterZanzibar

dgraph/cmd/alpha/http.go

@@ -652,10 +652,9 @@ func alterHandler(w http.ResponseWriter, r *http.Request) {
 		glog.Infof("The alter request is forwarded by %s\n", fwd)
 	}
 
-	// Pass in PoorMan's auth, ACL and IP information if present.
-	ctx := x.AttachAuthToken(context.Background(), r)
-	ctx = x.AttachAccessJwt(ctx, r)
-	ctx = x.AttachRemoteIP(ctx, r)
+	// Pass in PoorMan's auth, ACL and IP information if present, and resolve the
+	// caller's identity from it.
+	ctx := x.AttachRequestIdentity(context.Background(), r)
 	if _, err := (&edgraph.Server{}).Alter(ctx, op); err != nil {
 		x.SetStatus(w, x.Error, err.Error())
 		return
@@ -705,7 +704,12 @@ func graphqlProbeHandler(gqlHealthStore *admin.GraphQLHealthStore, globalEpoch m
 		w.Header().Set("Content-Type", "application/json")
 		// lazy load the schema so that just by making a probe request,
 		// one can boot up GraphQL for their namespace
-		namespace := x.ExtractNamespaceHTTP(r)
+		namespace, err := x.ResolveTenantHTTP(r)
+		if err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+			x.Check2(fmt.Fprintf(w, `{"error":"%s"}`, err))
+			return
+		}
 		if err := admin.LazyLoadSchema(namespace); err != nil {
 			w.WriteHeader(http.StatusInternalServerError)
 			x.Check2(w.Write([]byte(fmt.Sprintf(`{"error":"%s"}`, err))))
@@ -732,10 +736,11 @@ func resolveWithAdminServer(gqlReq *schema.Request, r *http.Request,
 	adminServer admin.IServeGraphQL) *schema.Response {
 	md := metadata.New(nil)
 	ctx := metadata.NewIncomingContext(context.Background(), md)
-	ctx = x.AttachAccessJwt(ctx, r)
-	ctx = x.AttachRemoteIP(ctx, r)
-	ctx = x.AttachAuthToken(ctx, r)
-	ctx = x.AttachJWTNamespace(ctx)
+	ctx = x.AttachRequestIdentity(ctx, r)
+	ctx, err := x.ResolveTenant(ctx)
+	if err != nil {
+		return schema.ErrorResponse(err)
+	}
 
 	return adminServer.ResolveWithNs(ctx, x.RootNamespace, gqlReq)
 }

dgraph/cmd/alpha/run.go

@@ -393,11 +393,15 @@ func serveGRPC(l net.Listener, tlsCfg *tls.Config, closer *z.Closer) {
 
 	x.RegisterExporters(Alpha.Conf, "dgraph.alpha")
 
-	unary := []grpc.UnaryServerInterceptor{audit.AuditRequestGRPC}
+	// Identity resolution runs first so every later interceptor and handler sees
+	// the same verified Principal instead of re-parsing the credential. It never
+	// rejects — see x.WithResolvedIdentity — so ordering it ahead of audit cannot
+	// suppress an audit record.
+	unary := []grpc.UnaryServerInterceptor{x.IdentityUnaryInterceptor(), audit.AuditRequestGRPC}
 	if zi := ZanzibarUnaryInterceptor(); zi != nil {
 		unary = append(unary, zi)
 	}
-	stream := []grpc.StreamServerInterceptor{audit.AuditStreamGRPC}
+	stream := []grpc.StreamServerInterceptor{x.IdentityStreamInterceptor(), audit.AuditStreamGRPC}
 	if zs := ZanzibarStreamInterceptor(); zs != nil {
 		stream = append(stream, zs)
 	}
@@ -521,7 +525,22 @@ func setupServer(closer *z.Closer, enableMcp bool) {
 	mainServer, adminServer, gqlHealthStore = admin.NewServers(introspection,
 		globalEpoch, closer)
 	baseMux.HandleFunc("/graphql", func(w http.ResponseWriter, r *http.Request) {
-		namespace := x.ExtractNamespaceHTTP(r)
+		// Strict here because this handler routes by the resolved namespace: it sets
+		// the resolver header that selects which namespace's GraphQL schema serves
+		// the request, so a token it cannot resolve must be refused rather than
+		// quietly served the root namespace's. A request with no token still
+		// resolves to the root namespace — see x.ResolveTenantHTTPStrict.
+		//
+		// Two limits, neither of which this closes. The strict variant only rejects a
+		// token that was *presented* and could not be resolved, and its "was one
+		// presented" probe reads the ACL accessJwt channel, so an installed
+		// non-ACL resolver is not covered. And the graphql-ws subscription handler
+		// routes by the same header while still using the lenient resolver.
+		namespace, err := x.ResolveTenantHTTPStrict(r)
+		if err != nil {
+			admin.WriteErrorResponse(w, r, err)
+			return
+		}
 		r.Header.Set("resolver", strconv.FormatUint(namespace, 10))
 		if err := admin.LazyLoadSchema(namespace); err != nil {
 			admin.WriteErrorResponse(w, r, err)
@@ -536,7 +555,12 @@ func setupServer(closer *z.Closer, enableMcp bool) {
 		r.Header.Set("resolver", "0")
 		// We don't need to load the schema for all the admin operations.
 		// Only a few like getUser, queryGroup require this. So, this can be optimized.
-		if err := admin.LazyLoadSchema(x.ExtractNamespaceHTTP(r)); err != nil {
+		namespace, err := x.ResolveTenantHTTP(r)
+		if err != nil {
+			admin.WriteErrorResponse(w, r, err)
+			return
+		}
+		if err := admin.LazyLoadSchema(namespace); err != nil {
 			admin.WriteErrorResponse(w, r, err)
 			return
 		}
@@ -706,6 +730,11 @@ func run() {
 	}
 	x.WorkerConfig.Parse(Alpha.Conf)
 
+	// Install deployment-specific authentication and authorization now: the config
+	// is parsed, and nothing is serving yet. A misconfiguration here is fatal, which
+	// is why it runs before any listener rather than lazily on the first request.
+	ConfigureIdentity()
+
 	// Set the directory for temporary buffers.
 	z.SetTmpDir(x.WorkerConfig.TmpDir)
 

dgraphtest/config.go

@@ -14,6 +14,13 @@ import (
 	"github.com/golang-jwt/jwt/v5"
 )
 
+// defaultWhitelist is the --security whitelist every dgraphtest cluster gets
+// unless a test overrides it with WithWhitelist. It has to stay wide open:
+// LocalCluster.Start() probes GraphQL with an admin mutation, and every admin
+// GraphQL operation carries IpWhitelistingMW, so a narrower default would break
+// every existing test rather than only the ones doing admin work.
+const defaultWhitelist = "0.0.0.0/0"
+
 // UpgradeCombo represents a version combination before and
 // after the upgrade, and the strategy for upgrading
 type UpgradeCombo struct {
@@ -112,6 +119,7 @@ type ClusterConfig struct {
 	repoDir               string
 	mcp                   bool
 	securityToken         string
+	whitelist             string
 }
 
 // NewClusterConfig generates a default ClusterConfig
@@ -132,6 +140,7 @@ func NewClusterConfig() ClusterConfig {
 		portOffset:     -1,
 		customPlugins:  false,
 		mcp:            false,
+		whitelist:      defaultWhitelist,
 	}
 }
 
@@ -174,6 +183,30 @@ func (cc ClusterConfig) WithSecurityToken(token string) ClusterConfig {
 	return cc
 }
 
+// WithWhitelist sets the Alpha's --security whitelist for admin operations,
+// REPLACING the wide-open default rather than adding to it. Zero is unaffected: it
+// reads its own whitelist from DGRAPH_ZERO_SECURITY, because older zero binaries used
+// in upgrade tests would fail to start on an unrecognized flag.
+//
+// The value is the flag's own syntax: a comma-separated list of IP addresses,
+// a.b.c.d:w.x.y.z ranges, CIDR blocks, or hostnames, e.g.
+// "192.168.0.0/16,host.docker.internal".
+//
+// Replacing is the point. The default is 0.0.0.0/0, so an additive option could never
+// express a cluster that denies anyone, which is the only configuration worth a test.
+// Note that loopback is admitted unconditionally by x.isIpWhitelisted regardless of
+// this setting, so a test that wants a denial has to reach the alpha over a
+// non-loopback source — which a published Docker port gives it on Linux, though not
+// necessarily on Docker Desktop.
+//
+// Setting a whitelist other than the default makes Start() wait on /probe/graphql
+// instead of the admin GraphQL mutation it normally probes with; see
+// LocalCluster.waitUntilGraphqlHealthCheck for why.
+func (cc ClusterConfig) WithWhitelist(spec string) ClusterConfig {
+	cc.whitelist = spec
+	return cc
+}
+
 // WithAcl enables ACL feature for Dgraph cluster
 func (cc ClusterConfig) WithACL(aclTTL time.Duration) ClusterConfig {
 	cc.acl = true

dgraphtest/dgraph.go

@@ -256,9 +256,9 @@ func (a *alpha) cmd(c *LocalCluster) []string {
 		"--bindall", "--logtostderr", fmt.Sprintf("-v=%d", c.conf.verbosity)}
 
 	if c.lowerThanV21 {
-		acmd = append(acmd, `--whitelist=0.0.0.0/0`, "--telemetry=false")
+		acmd = append(acmd, fmt.Sprintf(`--whitelist=%s`, c.conf.whitelist), "--telemetry=false")
 	} else {
-		security := `--security=whitelist=0.0.0.0/0`
+		security := fmt.Sprintf(`--security=whitelist=%s`, c.conf.whitelist)
 		if c.conf.securityToken != "" {
 			security += fmt.Sprintf(`;token=%s`, c.conf.securityToken)
 		}

dgraphtest/local_cluster.go

@@ -864,6 +864,15 @@ func (c *LocalCluster) waitUntilLogin() error {
 }
 
 func (c *LocalCluster) waitUntilGraphqlHealthCheck() error {
+	// The probes below are admin GraphQL operations, and every admin GraphQL
+	// operation carries IpWhitelistingMW — login included. On a cluster whose
+	// whitelist does not admit this process, neither can ever succeed, so wait on
+	// /probe/graphql instead: no whitelist middleware, no auth, and its whole
+	// purpose is answering "is GraphQL serving yet".
+	if c.conf.whitelist != defaultWhitelist {
+		return c.waitUntilGraphqlProbe()
+	}
+
 	hc, err := c.HTTPClient()
 	if err != nil {
 		return errors.Wrap(err, "error creating http client while graphql health check")
@@ -901,6 +910,33 @@ func (c *LocalCluster) waitUntilGraphqlHealthCheck() error {
 	return errors.Wrap(err, "error during graphql health check")
 }
 
+// waitUntilGraphqlProbe waits for GraphQL to serve using /probe/graphql, which
+// carries no IP-whitelist and no auth middleware. It is the probe to use when the
+// cluster's whitelist may not admit the test process.
+func (c *LocalCluster) waitUntilGraphqlProbe() error {
+	url, err := c.serverURL("alpha", "/probe/graphql")
+	if err != nil {
+		return errors.Wrap(err, "error getting graphql probe URL")
+	}
+
+	var lastErr error
+	for attempt := range 10 {
+		time.Sleep(waitDurBeforeRetry)
+		req, err := http.NewRequest(http.MethodGet, "http://"+url, nil)
+		if err != nil {
+			return errors.Wrap(err, "error building graphql probe request")
+		}
+		if _, lastErr = dgraphapi.DoReq(req); lastErr == nil {
+			log.Printf("[INFO] graphql probe succeeded for %v", c.conf.prefix)
+			return nil
+		}
+		if attempt > 5 {
+			log.Printf("[WARNING] problem during graphql probe: %v", lastErr)
+		}
+	}
+	return errors.Wrap(lastErr, "error during graphql probe")
+}
+
 // Upgrades the cluster to the provided dgraph version
 func (c *LocalCluster) Upgrade(version string, strategy UpgradeStrategy) error {
 	if version == c.conf.version {