Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 28 additions & 1 deletion audit/interceptor.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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,
Expand Down
14 changes: 14 additions & 0 deletions dgraph/cmd/alpha/hooks.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
23 changes: 14 additions & 9 deletions dgraph/cmd/alpha/http.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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))))
Expand All @@ -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)
}
Expand Down
37 changes: 33 additions & 4 deletions dgraph/cmd/alpha/run.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
Expand All @@ -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
}
Expand Down Expand Up @@ -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)

Expand Down
33 changes: 33 additions & 0 deletions dgraphtest/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -112,6 +119,7 @@ type ClusterConfig struct {
repoDir string
mcp bool
securityToken string
whitelist string
}

// NewClusterConfig generates a default ClusterConfig
Expand All @@ -132,6 +140,7 @@ func NewClusterConfig() ClusterConfig {
portOffset: -1,
customPlugins: false,
mcp: false,
whitelist: defaultWhitelist,
}
}

Expand Down Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions dgraphtest/dgraph.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down
36 changes: 36 additions & 0 deletions dgraphtest/local_cluster.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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 {
Expand Down
Loading
Loading