diff --git a/cmd/fabric-api/gateway.go b/cmd/fabric-api/gateway.go index 8dcb8515..56ff9922 100644 --- a/cmd/fabric-api/gateway.go +++ b/cmd/fabric-api/gateway.go @@ -55,6 +55,7 @@ type gatewayOptions struct { maxInFlight int retention, cleanupGrace time.Duration serviceIdentities []string + denyPrefixes []string serviceConfigMap string aggregateBudgets bool limits limit.Limits @@ -108,6 +109,9 @@ func newGatewayCommand() *cobra.Command { f.StringSliceVar(&o.serviceIdentities, "service-identities", nil, "Names of the Looking Glass service identities (spiffe://fabric-api.datumapis.com/cell//service/) "+ "allowed to call this cell's gateway; empty allows none") + f.StringSliceVar(&o.denyPrefixes, "deny-prefixes", nil, + "Platform or management ranges no Looking Glass probe may target, refused before the snapshot; "+ + "set it to the nodes' --deny-prefixes") f.StringVar(&o.serviceConfigMap, "service-config-map", "", "ConfigMap in --namespace whose key \"service\" turns Looking Glass execution on (\"enabled\") or off "+ "(\"disabled\") while the gateway runs, read every second; empty, missing, unreadable or any other "+ @@ -212,6 +216,10 @@ func runGateway(parent context.Context, o *gatewayOptions) error { if err != nil { return fmt.Errorf("--service-identities: %w", err) } + deny, err := query.ParsePrefixes(o.denyPrefixes) + if err != nil { + return fmt.Errorf("--deny-prefixes: %w", err) + } ctx, stop := signalContext(parent) defer stop() defer setupTracing(ctx, "gateway")() @@ -361,12 +369,13 @@ func runGateway(parent context.Context, o *gatewayOptions) error { // enforces the aggregate budgets; otherwise it authenticates, validates // and reports capabilities, and refuses execution with ServiceDisabled. service := &gateway.Service{ - Cell: o.cell, - Switch: serviceSwitch, - Metrics: metrics, - Allowlist: allowlist, - Elected: debug.Elected, - Leader: debug, + Cell: o.cell, + Switch: serviceSwitch, + Metrics: metrics, + Allowlist: allowlist, + Destinations: query.DestinationPolicy{Deny: deny}, + Elected: debug.Elected, + Leader: debug, Discoverer: &gateway.Discoverer{ Reader: mgr.GetAPIReader(), Namespace: o.namespace, Selector: sel, CertsSelector: certsSel, }, diff --git a/cmd/fabric-api/node.go b/cmd/fabric-api/node.go index 7e52fba0..92659c09 100644 --- a/cmd/fabric-api/node.go +++ b/cmd/fabric-api/node.go @@ -25,7 +25,6 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" - permitv1 "go.datum.net/galactic/api/fabric/permit/v1" fabricv1 "go.datum.net/galactic/api/fabric/v1" "go.datum.net/galactic/internal/fabric/frr" "go.datum.net/galactic/internal/fabric/gateway/limit" @@ -299,14 +298,17 @@ func nodePermits(o *nodeOptions, creds *identity.Credentials, metrics *node.Metr } metrics.RegisterClock(limit.KernelClock) if o.gatewayAddress != "" { - conn, err := grpc.NewClient(o.gatewayAddress, - grpc.WithTransportCredentials(credentials.NewTLS(creds.ClientConfig(identity.Gateway(o.cell)))), - // A permit request charges budgets; it is never replayed. - grpc.WithDisableRetry()) + gw, err := node.NewGatewayConn(func() (*grpc.ClientConn, error) { + return grpc.NewClient(o.gatewayAddress, + grpc.WithTransportCredentials(credentials.NewTLS(creds.ClientConfig(identity.Gateway(o.cell)))), + // A permit request charges budgets; it is never replayed. + grpc.WithDisableRetry()) + }) if err != nil { return nil, fmt.Errorf("--gateway-address: %w", err) } - p.Gateway = permitv1.NewPermitServiceClient(conn) + creds.OnLoad(func(*x509.Certificate) { gw.Redial() }) + p.Gateway = gw } return p, nil } diff --git a/config/fabric-router/components/fabric-api-require-permits/kustomization.yaml b/config/fabric-router/components/fabric-api-require-permits/kustomization.yaml index 5a8a7db5..6933b615 100644 --- a/config/fabric-router/components/fabric-api-require-permits/kustomization.yaml +++ b/config/fabric-router/components/fabric-api-require-permits/kustomization.yaml @@ -22,6 +22,25 @@ apiVersion: kustomize.config.k8s.io/v1alpha1 kind: Component patches: - path: daemonset-patch.yaml + # A sidecar asks the gateway for a direct operator probe's permit as a TLS + # client, and the gateway requires client auth of its callers, so each + # node's certificate is issued for client auth too. This recreates the + # fabric-api-certs pods only, never FRR. + - patch: |- + apiVersion: apps/v1 + kind: DaemonSet + metadata: + name: fabric-api-certs + namespace: galactic-system + spec: + template: + spec: + volumes: + - name: fabric-api-csi + csi: + driver: csi.cert-manager.io + volumeAttributes: + csi.cert-manager.io/key-usages: client auth,server auth # Fails the build unless components/fabric-api-permits, which adds this # Role, came first; it changes nothing. - patch: |- diff --git a/docs/fabric-api/configuration.md b/docs/fabric-api/configuration.md index 4b4eb499..bd84ca45 100644 --- a/docs/fabric-api/configuration.md +++ b/docs/fabric-api/configuration.md @@ -118,6 +118,7 @@ FRR, so it follows [Rollout](operations.md#rollout). | `--retention` | | `1h` | Retention the stale-object count assumes; keep equal to the janitor's `--retention` | | `--cleanup-grace` | | `15m` | Cleanup grace the stale-object count assumes; keep equal to the janitor's `--grace` | | `--service-identities` | | | Comma-separated names of the Looking Glass service identities allowed to call this gateway. Empty allows none | +| `--deny-prefixes` | | | Comma-separated platform or management CIDRs no Looking Glass probe may target, refused before the snapshot. Set it to the nodes' `--deny-prefixes`; see [Destination](security.md#destination) | | `--service-config-map` | | | ConfigMap in `--namespace` whose key `service` turns Looking Glass execution on (`enabled`) or off while the gateway runs, read every second. Empty keeps it off. Needs `--aggregate-budgets`. See [Operating the Looking Glass service](operations.md#operating-the-looking-glass-service) | | `--aggregate-budgets` | | `false` | Enforce the cell's aggregate admission and probe packet budgets on every caller and issue probe permits. Requires `--cell-slots` at most 8. See [Aggregate probe budgets](#aggregate-probe-budgets) | | Limit flags | | The contract's | `--request-*`, `--packet-*`, `--principal-slots`, `--principal-queue`, `--service-streams`, `--project-streams` and `--queue-wait`: the aggregate limits, each only lowerable; need `--aggregate-budgets`. See [Limits and capacity](#limits-and-capacity) | diff --git a/docs/fabric-api/operations.md b/docs/fabric-api/operations.md index 38295983..f787be4b 100644 --- a/docs/fabric-api/operations.md +++ b/docs/fabric-api/operations.md @@ -140,7 +140,9 @@ each step cell by cell: after `components/fabric-api-permits`. It adds `--require-probe-permits` and `--gateway-address=fabric-api-gateway.galactic-system.svc:9346` (the gateway Service), so direct operator probes are charged too. Another - routing rollout. + routing rollout. It also issues each node's certificate for client auth + as well as server auth, since the sidecar asks the gateway for those + permits as a TLS client; that recreates the `fabric-api-certs` pods. 4. **The gateway serves the Looking Glass service, switched off.** Add `config/fabric-api/components/service` after `components/aggregate-budgets`. See diff --git a/docs/fabric-api/security.md b/docs/fabric-api/security.md index 13cb5cb3..15ea7434 100644 --- a/docs/fabric-api/security.md +++ b/docs/fabric-api/security.md @@ -211,6 +211,13 @@ every call from the gateway identity, a destination is refused - inside a range listed in `--deny-prefixes`: add your management and platform ranges here. +The gateway applies the same public policy to every Looking Glass service +probe before its snapshot, using its own `--deny-prefixes`, and refuses a +forbidden target with `DestinationNotAllowed` (`INVALID_ARGUMENT`) before it +charges any token, slot or packet. Set the gateway's `--deny-prefixes` to the +nodes' value: a range missing on the gateway is still refused by every node, +but only after the request is admitted and charged. + The **operator policy** applies to calls made directly to a node with an operator identity. It is the public policy plus `--operator-allow-prefixes`, and an allowed prefix bypasses every other rule, including `--deny-prefixes`. The lab uses it to diff --git a/internal/fabric/gateway/admission_test.go b/internal/fabric/gateway/admission_test.go index 1fb15ea0..ccc7f7bc 100644 --- a/internal/fabric/gateway/admission_test.go +++ b/internal/fabric/gateway/admission_test.go @@ -38,8 +38,8 @@ const ( testLease = "fabric-api-gateway" testHolder = "gateway-0_test" // probeTarget6 and probeTarget4 are probe destinations. - probeTarget6 = "2001:db8::1" - probeTarget4 = "192.0.2.1" + probeTarget6 = "2606:4700::1" + probeTarget4 = "9.9.9.9" // podA is router-a's fabric-router pod UID. podA = "pod-a" ) @@ -206,7 +206,7 @@ func TestServiceDuplicateAttempts(t *testing.T) { if errcode.Of(err) != errcode.DuplicateRequest { t.Errorf("retry with a later expiration: %v", err) } - _, err = collect(t, svc, requestFor("project-a", 1, func(r *runReq) { r.Query.Target = "2001:db8::2" })) + _, err = collect(t, svc, requestFor("project-a", 1, func(r *runReq) { r.Query.Target = "2606:4700::2" })) if errcode.Of(err) != errcode.DuplicateConflict { t.Errorf("changed arguments: %v", err) } diff --git a/internal/fabric/gateway/load_test.go b/internal/fabric/gateway/load_test.go index 1ee05d80..e172dfda 100644 --- a/internal/fabric/gateway/load_test.go +++ b/internal/fabric/gateway/load_test.go @@ -195,7 +195,7 @@ func TestDestinationPrefixAcrossProjects(t *testing.T) { l := roomyLimits() l.PacketPrefix = limit.Rate{PerSecond: 1, Burst: 2 * limit.TraceroutePackets} withAdmission(t, svc, l) - for i, target := range []string{probeTarget6, "2001:db8::2", "2001:db8::3"} { + for i, target := range []string{probeTarget6, "2606:4700::2", "2606:4700::3"} { events, err := collect(t, svc, requestFor(fmt.Sprintf("project-%d", i), i, func(r *runReq) { r.Query = traceroute(target) })) @@ -226,7 +226,7 @@ func TestAggregateBudgetsCoverEveryCaller(t *testing.T) { routers := []Node{{Name: routerA, PodUID: podA, Selected: true}} expires := time.Now().Add(30 * time.Second) - events, err := collect(t, svc, requestFor("project-a", 1, func(r *runReq) { r.Query = traceroute("2001:db8:1::1") })) + events, err := collect(t, svc, requestFor("project-a", 1, func(r *runReq) { r.Query = traceroute("2606:4700:1::1") })) if err != nil { t.Fatal(err) } @@ -237,7 +237,7 @@ func TestAggregateBudgetsCoverEveryCaller(t *testing.T) { return node.QueryFromProto(&fabricv1.Query{Type: fabricv1.QueryType_QUERY_TYPE_TRACEROUTE, Target: target, AddressFamily: fabricv1.AddressFamily_ADDRESS_FAMILY_IPV6}) } - out := svc.Runner.Run(t.Context(), "ns/cluster", "fq-1", trace("2001:db8:2::1"), routers, nil, expires) + out := svc.Runner.Run(t.Context(), "ns/cluster", "fq-1", trace("2606:4700:2::1"), routers, nil, expires) if out[0].Err != nil { t.Fatalf("FabricQuery: %v", out[0].Err) } @@ -248,7 +248,7 @@ func TestAggregateBudgetsCoverEveryCaller(t *testing.T) { t.Fatalf("direct operator: %v", err) } - out = svc.Runner.Run(t.Context(), "operator/bob", "op-1", trace("2001:db8:4::1"), routers, nil, expires) + out = svc.Runner.Run(t.Context(), "operator/bob", "op-1", trace("2606:4700:4::1"), routers, nil, expires) if errcode.Of(out[0].Err) != errcode.Throttled || errcode.Metadata(out[0].Err)[errcode.MetadataScope] != "cell" { t.Errorf("operator through the gateway past the cell budget: %v", out[0].Err) } diff --git a/internal/fabric/gateway/service.go b/internal/fabric/gateway/service.go index 9a79982b..e53abb2d 100644 --- a/internal/fabric/gateway/service.go +++ b/internal/fabric/gateway/service.go @@ -241,6 +241,10 @@ type Service struct { Revision string // Allowlist names the accepted service identities. Allowlist Allowlist + // Destinations is the public destination policy a probe target must + // pass before its snapshot. Its Deny should match every node's + // --deny-prefixes; its Allow must be empty. + Destinations query.DestinationPolicy // Elected reports whether this replica leads. Elected func() bool // Leader connects a follower to the leader. @@ -490,7 +494,7 @@ func (s *Service) validate( return nil, time.Time{}, errcode.Status(errcode.InvalidQuery, fmt.Sprintf("expires_at is more than %s after receipt", MaxServiceRequestLifetime)) } - q, err := serviceQuery(req) + q, err := serviceQuery(req, s.Destinations) if err != nil { return nil, time.Time{}, err } @@ -514,8 +518,8 @@ func (s *Service) validate( // serviceQuery canonicalizes a request's query and refuses any type the // service may not request, and any probe target that is not a numeric -// address. -func serviceQuery(req *servicev1.RunDiagnosticRequest) (query.Query, error) { +// address or that destinations refuses. +func serviceQuery(req *servicev1.RunDiagnosticRequest, destinations query.DestinationPolicy) (query.Query, error) { if req.GetQuery() == nil { return query.Query{}, errcode.Status(errcode.InvalidQuery, "query is required") } @@ -528,7 +532,11 @@ func serviceQuery(req *servicev1.RunDiagnosticRequest) (query.Query, error) { return query.Query{}, errcode.FromError(err) } if q.Type.Probe() { - if _, err := query.ResolvedProbe(q); err != nil { + dst, err := query.ResolvedProbe(q) + if err != nil { + return query.Query{}, errcode.FromError(err) + } + if err := destinations.Check(dst); err != nil { return query.Query{}, errcode.FromError(err) } } diff --git a/internal/fabric/gateway/service_execution_test.go b/internal/fabric/gateway/service_execution_test.go index 45cf6727..70f6d70c 100644 --- a/internal/fabric/gateway/service_execution_test.go +++ b/internal/fabric/gateway/service_execution_test.go @@ -115,7 +115,7 @@ func pingObservation(name string, replies uint32) *fabricv1.Observation { obs := &fabricv1.Observation{ Node: name, RouterId: "192.0.2.1", Asn: 64512, SampleTime: timestamppb.Now(), Result: &fabricv1.Observation_Ping{Ping: &fabricv1.PingResult{ - Source: "2001:db8::a", Destination: "2001:db8::1", Sent: 3, Received: replies, + Source: "2606:4700::a", Destination: "2606:4700::1", Sent: 3, Received: replies, }}, } for i := range uint32(3) { @@ -126,7 +126,7 @@ func pingObservation(name string, replies uint32) *fabricv1.Observation { func failAnswer(code errcode.Code) routerAnswer { return func(context.Context, *fabricv1.ExecuteRequest) (*fabricv1.Observation, error) { - return nil, errcode.Status(code, "raw failure naming 2001:db8::1") + return nil, errcode.Status(code, "raw failure naming 2606:4700::1") } } @@ -497,7 +497,7 @@ func TestServiceCoverage(t *testing.T) { } for n, reason := range tc.failures { f := outcomes[n].GetFailure() - if f.GetReason() != reason || f.GetMessage() == "" || f.GetMessage() == "raw failure naming 2001:db8::1" { + if f.GetReason() != reason || f.GetMessage() == "" || f.GetMessage() == "raw failure naming 2606:4700::1" { t.Errorf("router %s failure = %v, want sanitized %s", n, f, reason) } } diff --git a/internal/fabric/gateway/service_test.go b/internal/fabric/gateway/service_test.go index cd5e0c67..b8b6ee02 100644 --- a/internal/fabric/gateway/service_test.go +++ b/internal/fabric/gateway/service_test.go @@ -8,6 +8,7 @@ import ( "context" "crypto/tls" "net" + "net/netip" "path/filepath" "strconv" "strings" @@ -35,6 +36,7 @@ import ( "go.datum.net/galactic/internal/fabric/errcode" "go.datum.net/galactic/internal/fabric/identity" "go.datum.net/galactic/internal/fabric/node" + "go.datum.net/galactic/internal/fabric/query" "go.datum.net/galactic/internal/fabric/testpki" ) @@ -92,7 +94,7 @@ func validRequest(now time.Time) *servicev1.RunDiagnosticRequest { RequestId: testRequestID, ExpiresAt: timestamppb.New(now.Add(30 * time.Second)), Query: &fabricv1.Query{ - Type: fabricv1.QueryType_QUERY_TYPE_PING, Target: "2001:0DB8::1", + Type: fabricv1.QueryType_QUERY_TYPE_PING, Target: "2606:4700:0000::1", }, } } @@ -400,7 +402,7 @@ func TestServiceForwarding(t *testing.T) { got.GetCell() != testSite { t.Errorf("envelope attribution = %v", env) } - if got.GetQuery().GetTarget() != "2001:db8::1" || + if got.GetQuery().GetTarget() != "2606:4700::1" || got.GetQuery().GetAddressFamily() != fabricv1.AddressFamily_ADDRESS_FAMILY_IPV6 { t.Errorf("envelope query is not canonical: %v", got.GetQuery()) } @@ -492,7 +494,10 @@ func (s *fakeEventStream) Send(ev *servicev1.DiagnosticEvent) error { func TestServiceValidation(t *testing.T) { now := time.Now() - svc := &Service{Cell: testSite, Allowlist: testAllowlist, Elected: elected, Now: func() time.Time { return now }} + svc := &Service{ + Cell: testSite, Allowlist: testAllowlist, Elected: elected, Now: func() time.Time { return now }, + Destinations: query.DestinationPolicy{Deny: []netip.Prefix{netip.MustParsePrefix("2606:4700:ff::/48")}}, + } ctx := node.WithCaller(t.Context(), identity.Service(testSite, testService)) for name, tc := range map[string]struct { edit func(*runReq) @@ -524,7 +529,18 @@ func TestServiceValidation(t *testing.T) { "as-path": {func(r *runReq) { r.Query = &fabricv1.Query{Type: fabricv1.QueryType_QUERY_TYPE_AS_PATH, Target: "_64512_"} }, errcode.InvalidQuery}, - "hostname probe": {func(r *runReq) { r.Query.Target = "example.com" }, errcode.InvalidQuery}, + "hostname probe": {func(r *runReq) { r.Query.Target = "example.com" }, errcode.InvalidQuery}, + // The public destination policy refuses before the snapshot, so a + // forbidden probe costs no tokens, slots or packets. + "documentation probe": {func(r *runReq) { r.Query.Target = "2001:db8::1" }, errcode.DestinationNotAllowed}, + "unique local probe": {func(r *runReq) { r.Query.Target = "fc00::1" }, errcode.DestinationNotAllowed}, + "private probe": {func(r *runReq) { + r.Query = &fabricv1.Query{Type: fabricv1.QueryType_QUERY_TYPE_TRACEROUTE, Target: "10.0.0.1"} + }, errcode.DestinationNotAllowed}, + "multicast probe": {func(r *runReq) { + r.Query = &fabricv1.Query{Type: fabricv1.QueryType_QUERY_TYPE_PING, Target: "224.0.0.1"} + }, errcode.DestinationNotAllowed}, + "denied probe": {func(r *runReq) { r.Query.Target = "2606:4700:ff::1" }, errcode.DestinationNotAllowed}, "too many nodes": {func(r *runReq) { r.Nodes = make([]string, MaxServiceNodes+1) }, errcode.InvalidQuery}, "duplicate nodes": {func(r *runReq) { r.Nodes = []string{"a", "a"} }, errcode.InvalidQuery}, "empty node name": {func(r *runReq) { r.Nodes = []string{""} }, errcode.InvalidQuery}, @@ -546,6 +562,15 @@ func TestServiceValidation(t *testing.T) { } } + // With no deployment ranges configured, the public policy still + // refuses special-use targets. + open := &Service{Cell: testSite, Allowlist: testAllowlist, Elected: elected, Now: func() time.Time { return now }} + req := validRequest(now) + req.Query.Target = "fc00::1" + if err := open.RunDiagnostic(req, &fakeEventStream{ctx: ctx}); errcode.Of(err) != errcode.DestinationNotAllowed { + t.Errorf("unique local probe without --deny-prefixes: %v, want %s", err, errcode.DestinationNotAllowed) + } + // GetCapabilities checks the version and cell too. _, err := svc.GetCapabilities(ctx, &servicev1.GetCapabilitiesRequest{Cell: testSite, ApiMajor: 2}) if errcode.Of(err) != errcode.VersionUnsupported { diff --git a/internal/fabric/gateway/serviceswitch_test.go b/internal/fabric/gateway/serviceswitch_test.go index 9863369a..24dfbbdc 100644 --- a/internal/fabric/gateway/serviceswitch_test.go +++ b/internal/fabric/gateway/serviceswitch_test.go @@ -251,7 +251,7 @@ func TestServiceTelemetryCarriesNoPayload(t *testing.T) { t.Errorf("charged packets = %g", got) } - forbidden := []string{testProject, testRequestID[:8], "2001:db8", routerA, routerB, podA, "raw failure", + forbidden := []string{testProject, testRequestID[:8], "2606:4700", routerA, routerB, podA, "raw failure", testService} families, err := reg.Gather() if err != nil { @@ -270,7 +270,7 @@ func TestServiceTelemetryCarriesNoPayload(t *testing.T) { } out := logs.String() - for _, bad := range []string{"2001:db8", "raw failure", "replies", "192.0.2.1"} { + for _, bad := range []string{"2606:4700", "raw failure", "replies", "192.0.2.1"} { if strings.Contains(out, bad) { t.Errorf("logs carry %q:\n%s", bad, out) } diff --git a/internal/fabric/node/auth.go b/internal/fabric/node/auth.go index d47cba0d..f4e94ae1 100644 --- a/internal/fabric/node/auth.go +++ b/internal/fabric/node/auth.go @@ -8,6 +8,7 @@ import ( "context" "crypto/tls" "log/slog" + "strings" "time" "google.golang.org/grpc" @@ -16,6 +17,7 @@ import ( "google.golang.org/grpc/peer" "google.golang.org/grpc/status" + servicev1 "go.datum.net/galactic/api/fabric/service/v1" fabricv1 "go.datum.net/galactic/api/fabric/v1" "go.datum.net/galactic/internal/fabric/errcode" "go.datum.net/galactic/internal/fabric/identity" @@ -149,7 +151,7 @@ func (a *Authorizer) Unary() grpc.UnaryServerInterceptor { if r, ok := req.(interface{ GetRequestId() string }); ok { attrs = append(attrs, "requestID", r.GetRequestId()) } - logCall(attrs, err) + logCall(info.FullMethod, attrs, err) return resp, err } } @@ -165,17 +167,29 @@ func (a *Authorizer) Stream() grpc.StreamServerInterceptor { defer cancel() start := time.Now() err = handler(srv, &callerStream{ServerStream: ss, ctx: ctx}) - logCall([]any{"method", info.FullMethod, "caller", id.URI(), "duration", time.Since(start)}, err) + logCall(info.FullMethod, []any{"method", info.FullMethod, "caller", id.URI(), "duration", time.Since(start)}, err) return err } } -func logCall(attrs []any, err error) { - if err != nil { - slog.Info("rpc failed", append(attrs, "code", errcode.Of(err), "error", errcode.Message(err))...) - } else { +// serviceMethodPrefix prefixes every Looking Glass service method. Errors +// refusing a service request can quote its target, which the service +// contract keeps out of routine logs, so their failures are logged by code +// alone. +var serviceMethodPrefix = "/" + servicev1.LookingGlassService_ServiceDesc.ServiceName + "/" + +func logCall(method string, attrs []any, err error) { + if err == nil { slog.Debug("rpc", attrs...) + return + } + attrs = append(attrs, "code", errcode.Of(err)) + if !strings.HasPrefix(method, serviceMethodPrefix) { + attrs = append(attrs, "error", errcode.Message(err)) + } else if qc := errcode.Metadata(err)[errcode.MetadataQueryCode]; qc != "" { + attrs = append(attrs, "queryCode", qc) } + slog.Info("rpc failed", attrs...) } // callerStream is a server stream whose context carries the caller and diff --git a/internal/fabric/node/auth_test.go b/internal/fabric/node/auth_test.go index 34fb2ce8..7ff72f29 100644 --- a/internal/fabric/node/auth_test.go +++ b/internal/fabric/node/auth_test.go @@ -5,14 +5,17 @@ package node import ( + "bytes" "context" "crypto/tls" "crypto/x509" "encoding/pem" "errors" + "log/slog" "net" "os" "path/filepath" + "strings" "testing" "time" @@ -24,6 +27,7 @@ import ( "google.golang.org/grpc/peer" "google.golang.org/grpc/status" + servicev1 "go.datum.net/galactic/api/fabric/service/v1" "go.datum.net/galactic/internal/fabric/errcode" "go.datum.net/galactic/internal/fabric/identity" "go.datum.net/galactic/internal/fabric/testpki" @@ -220,3 +224,31 @@ func TestAuthorizerEndsStreamAtExpiry(t *testing.T) { t.Errorf("stream ended at %v, before the certificate's expiry %v", now, notAfter) } } + +// A failed Looking Glass service call is logged by its codes alone, since +// its message can quote the caller's target; other calls keep the message. +func TestLogCallOmitsServiceErrorText(t *testing.T) { + var logs bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewJSONHandler(&logs, nil))) + t.Cleanup(func() { slog.SetDefault(prev) }) + + const target = "192.0.2.77" + refusal := errcode.StatusWithMetadata(errcode.InvalidQuery, "InvalidTarget: "+target+" is not a plain address", + map[string]string{errcode.MetadataQueryCode: "InvalidTarget"}) + + logCall(servicev1.LookingGlassService_RunDiagnostic_FullMethodName, nil, refusal) + line := logs.String() + if strings.Contains(line, target) { + t.Errorf("service failure log names the target: %s", line) + } + if !strings.Contains(line, `"code":"InvalidQuery"`) || !strings.Contains(line, `"queryCode":"InvalidTarget"`) { + t.Errorf("service failure log lacks its codes: %s", line) + } + + logs.Reset() + logCall(testMethod, nil, refusal) + if !strings.Contains(logs.String(), target) { + t.Errorf("operator failure log lost its message: %s", logs.String()) + } +} diff --git a/internal/fabric/node/gatewayconn.go b/internal/fabric/node/gatewayconn.go new file mode 100644 index 00000000..b5508c77 --- /dev/null +++ b/internal/fabric/node/gatewayconn.go @@ -0,0 +1,107 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package node + +import ( + "context" + "log/slog" + "sync" + + "google.golang.org/grpc" + + permitv1 "go.datum.net/galactic/api/fabric/permit/v1" +) + +// GatewayConn is a PermitGateway over a connection to the cell gateway that +// Redial replaces. A TLS connection presents the certificate it was opened +// with for as long as it lasts, while the gateway re-verifies its caller on +// every call; a connection outliving the node's credentials (a renewal, or +// a replaced fabric-api-certs pod, which changes the node's identity) is +// refused until it happens to close. Redial on every credential load keeps +// the connection's identity current. +type GatewayConn struct { + dial func() (*grpc.ClientConn, error) + + mu sync.Mutex + cur *gatewayConn +} + +type gatewayConn struct { + conn *grpc.ClientConn + client permitv1.PermitServiceClient + calls sync.WaitGroup +} + +// NewGatewayConn returns a GatewayConn whose connections dial makes. +func NewGatewayConn(dial func() (*grpc.ClientConn, error)) (*GatewayConn, error) { + g := &GatewayConn{dial: dial} + c, err := g.open() + if err != nil { + return nil, err + } + g.cur = c + return g, nil +} + +func (g *GatewayConn) open() (*gatewayConn, error) { + conn, err := g.dial() + if err != nil { + return nil, err + } + return &gatewayConn{conn: conn, client: permitv1.NewPermitServiceClient(conn)}, nil +} + +// Redial replaces the connection for later calls. The previous one closes +// once the calls already running on it end, so a redial never interrupts a +// permit request. If dialing fails, the previous connection stays in use. +func (g *GatewayConn) Redial() { + next, err := g.open() + if err != nil { + slog.Warn("probe permit connection to the gateway not replaced", "error", err) + return + } + g.mu.Lock() + prev := g.cur + g.cur = next + g.mu.Unlock() + go func() { + prev.calls.Wait() + _ = prev.conn.Close() + }() +} + +// Close closes the current connection. +func (g *GatewayConn) Close() error { + g.mu.Lock() + defer g.mu.Unlock() + return g.cur.conn.Close() +} + +// acquire returns the current connection, which the caller releases with +// its calls.Done once its call ends. +func (g *GatewayConn) acquire() *gatewayConn { + g.mu.Lock() + defer g.mu.Unlock() + g.cur.calls.Add(1) + return g.cur +} + +// RequestPermit implements PermitGateway. +func (g *GatewayConn) RequestPermit( + ctx context.Context, in *permitv1.RequestPermitRequest, opts ...grpc.CallOption, +) (*permitv1.RequestPermitResponse, error) { + c := g.acquire() + defer c.calls.Done() + return c.client.RequestPermit(ctx, in, opts...) +} + +// ReleasePermit implements PermitGateway. +func (g *GatewayConn) ReleasePermit( + ctx context.Context, in *permitv1.ReleasePermitRequest, opts ...grpc.CallOption, +) (*permitv1.ReleasePermitResponse, error) { + c := g.acquire() + defer c.calls.Done() + return c.client.ReleasePermit(ctx, in, opts...) +} diff --git a/internal/fabric/node/gatewayconn_test.go b/internal/fabric/node/gatewayconn_test.go new file mode 100644 index 00000000..fd19d849 --- /dev/null +++ b/internal/fabric/node/gatewayconn_test.go @@ -0,0 +1,130 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package node + +import ( + "context" + "crypto/x509" + "net" + "path/filepath" + "sync" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + + permitv1 "go.datum.net/galactic/api/fabric/permit/v1" + "go.datum.net/galactic/internal/fabric/identity" + "go.datum.net/galactic/internal/fabric/testpki" +) + +// recordingPermits records the identity each permit request arrives with, +// and holds a request whose node is "hold" until release closes. +type recordingPermits struct { + permitv1.UnimplementedPermitServiceServer + mu sync.Mutex + callers []string + held chan struct{} + release chan struct{} +} + +func (r *recordingPermits) RequestPermit( + ctx context.Context, req *permitv1.RequestPermitRequest, +) (*permitv1.RequestPermitResponse, error) { + id, err := peerIdentity(ctx) + if err != nil { + return nil, err + } + r.mu.Lock() + r.callers = append(r.callers, id.Name) + r.mu.Unlock() + if req.GetNode() == "hold" { + close(r.held) + <-r.release + } + return &permitv1.RequestPermitResponse{}, nil +} + +func (r *recordingPermits) last() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.callers[len(r.callers)-1] +} + +// A node's permit requests present its current certificate after the +// credentials reload with a new certs pod's identity, and a request +// running across the reload completes on its own connection. +func TestGatewayConnFollowsCredentials(t *testing.T) { + dir := t.TempDir() + ca := testpki.NewCA(t, "dfw") + gwCreds := creds(t, ca, dir, identity.Gateway("dfw")) + permits := &recordingPermits{held: make(chan struct{}), release: make(chan struct{})} + srv := grpc.NewServer(grpc.Creds(credentials.NewTLS(gwCreds.ServerConfig(func(identity.ID) error { return nil })))) + permitv1.RegisterPermitServiceServer(srv, permits) + var lc net.ListenConfig + l, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + go func() { _ = srv.Serve(l) }() + t.Cleanup(srv.Stop) + + nodeDir := filepath.Join(dir, "node") + issue := func(pod string) { + id := identity.Node("dfw", "galactic-system", pod) + ca.Issue(t, nodeDir, testpki.Options{URIs: []string{id.URI()}}) + } + issue("fabric-api-certs-a") + nodeCreds := &identity.Credentials{ + CertFile: filepath.Join(nodeDir, "tls.crt"), KeyFile: filepath.Join(nodeDir, "tls.key"), + CAFile: filepath.Join(nodeDir, "ca.crt"), Self: identity.Node("dfw", "galactic-system", ""), AcceptAnyNode: true, + } + if err := nodeCreds.Reload(); err != nil { + t.Fatal(err) + } + gw, err := NewGatewayConn(func() (*grpc.ClientConn, error) { + return grpc.NewClient(l.Addr().String(), + grpc.WithTransportCredentials(credentials.NewTLS(nodeCreds.ClientConfig(identity.Gateway("dfw"))))) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = gw.Close() }) + nodeCreds.OnLoad(func(*x509.Certificate) { gw.Redial() }) + + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + if _, err := gw.RequestPermit(ctx, &permitv1.RequestPermitRequest{Node: "n"}); err != nil { + t.Fatal(err) + } + if got := permits.last(); got != "fabric-api-certs-a" { + t.Fatalf("first request from %q", got) + } + + held := make(chan error, 1) + go func() { + _, err := gw.RequestPermit(ctx, &permitv1.RequestPermitRequest{Node: "hold"}) + held <- err + }() + <-permits.held + + // The certs pod is replaced: the node's certificate names the new pod. + issue("fabric-api-certs-b") + if err := nodeCreds.Reload(); err != nil { + t.Fatal(err) + } + if _, err := gw.RequestPermit(ctx, &permitv1.RequestPermitRequest{Node: "n"}); err != nil { + t.Fatal(err) + } + if got := permits.last(); got != "fabric-api-certs-b" { + t.Errorf("request after the reload from %q, want the new certs pod", got) + } + + close(permits.release) + if err := <-held; err != nil { + t.Errorf("request running across the reload: %v", err) + } +} diff --git a/internal/fabric/node/permit.go b/internal/fabric/node/permit.go index 5a4027f8..f41522eb 100644 --- a/internal/fabric/node/permit.go +++ b/internal/fabric/node/permit.go @@ -43,7 +43,8 @@ const ( ) // PermitGateway obtains and releases permits from the cell's gateway on a -// direct operator's behalf. permitv1.PermitServiceClient satisfies it. +// direct operator's behalf. GatewayConn and permitv1.PermitServiceClient +// satisfy it. type PermitGateway interface { RequestPermit(ctx context.Context, in *permitv1.RequestPermitRequest, opts ...grpc.CallOption) (*permitv1.RequestPermitResponse, error)