fix(flow): add mTLS authentication interceptor and recovery interceptor#4140
fix(flow): add mTLS authentication interceptor and recovery interceptor#4140jw-nvidia wants to merge 1 commit into
Conversation
Signed-off-by: Jin Wang <jinwan@nvidia.com>
Summary by CodeRabbit
WalkthroughFlow adds configurable SPIFFE service-identity authorization, unary and streaming gRPC enforcement, panic recovery interceptors, TLS-aware service validation, and CLI/environment configuration for authorization allowlists. ChangesFlow security controls
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant FlowClient
participant GRPCInterceptors
participant Authorizer
participant FlowHandler
FlowClient->>GRPCInterceptors: send unary or streaming RPC
GRPCInterceptors->>Authorizer: authorize TLS peer identity
Authorizer-->>GRPCInterceptors: allow, audit, or reject decision
GRPCInterceptors->>FlowHandler: invoke with propagated service identity
FlowHandler-->>FlowClient: return response or gRPC error
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
🔐 TruffleHog Secret Scan✅ No secrets or credentials found! Your code has been scanned for 700+ types of secrets and credentials. All clear! 🎉 🕐 Last updated: 2026-07-24 22:57:10 UTC | Commit: 0e4a342 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
rest-api/flow/internal/authz/interceptor.go (1)
65-91: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winFold
checkAuthorizationResultinto*Authorizeras a method.The function only ever consumes
authorizer.mode, threaded manually from both call sites (Lines 28, 50). Making it a receiver method removes the redundantmodeparameter and keeps decision-related behavior consistently owned byAuthorizer, alongsideauthorize().♻️ Proposed refactor
-func checkAuthorizationResult(method string, authorization *decision, mode Mode) error { +func (a *Authorizer) checkAuthorizationResult(method string, authorization *decision) error { if err := grpcError(authorization.rejection); err != nil { - enforce := mode.enforce() + enforce := a.mode.enforce() level := zerolog.WarnLevel if enforce { level = zerolog.ErrorLevel } log.WithLevel(level). Str("grpc_method", method). Str("grpc_code", status.Code(err).String()). - Str("authorization_mode", string(mode)). + Str("authorization_mode", string(a.mode)). Str("service_identity", authorization.identity). Msg(authorization.message)And update both call sites, e.g.:
- if err := checkAuthorizationResult( - info.FullMethod, - &authorization, - authorizer.mode, - ); err != nil { + if err := authorizer.checkAuthorizationResult(info.FullMethod, &authorization); err != nil {As per path instructions, "discourage scattered independent functions when a receiver method would make ownership and responsibilities clearer."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rest-api/flow/internal/authz/interceptor.go` around lines 65 - 91, Convert checkAuthorizationResult into an Authorizer method so it uses the receiver’s mode instead of accepting mode as a parameter. Update both callers to invoke the method through their Authorizer instance, remove the redundant mode argument, and preserve the existing authorization logging, identity handling, and enforcement behavior.Source: Path instructions
rest-api/flow/internal/authz/interceptor_test.go (1)
18-89: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winExpand interceptor test coverage: unary unauthenticated path and stream enforce/audit-denied paths.
TestUnaryServerInterceptornever asserts the "no TLS peer" (Unauthenticated) case, only "allowed"/"denied".TestStreamServerInterceptorPropagatesServiceIdentityonly exercises the happy path — there's no equivalent toTestUnaryServerInterceptorAuditModeor theModeEnforce-denied case forStreamServerInterceptor. Since both interceptors route through the samecheckAuthorizationResult, symmetric coverage would catch any future divergence between the two call sites.As per coding guidelines, "Endpoint tests must cover validation, authorization, ... status codes, response shape, ... timeout behavior, and secret absence where relevant."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rest-api/flow/internal/authz/interceptor_test.go` around lines 18 - 89, Expand interceptor tests to cover missing TLS peer authentication and stream authorization outcomes. In TestUnaryServerInterceptor, add a no-peer case expecting codes.Unauthenticated with handlerRuns false; add stream tests for ModeEnforce denial and audit-mode denial, asserting the expected status and whether the handler runs while preserving identity propagation for allowed requests. Reuse the existing TLS context helpers and symbols such as StreamServerInterceptor and checkAuthorizationResult’s expected behavior.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rest-api/flow/internal/common/grpcrecovery/recovery.go`:
- Around line 52-66: The service identity lookup in recoverPanic cannot observe
the context enriched by downstream interceptors, so remove this ineffective
authz.ServiceIdentityFromContext enrichment from the recovery logger and
implement the interceptor-chain fix described in service.go so the enriched
context is available when panic recovery logs execute. Keep panic status
conversion and existing log fields unchanged.
In `@rest-api/flow/internal/service/service.go`:
- Around line 273-284: Reorder the interceptor chains in the gRPC server setup
so authz.UnaryServerInterceptor and authz.StreamServerInterceptor are outermost,
followed by the corresponding grpcrecovery interceptors, allowing recovered
handler panics to observe the service identity added by authz. Add an
integration-style test covering an authenticated handler panic and assert the
recovery log includes service_identity.
- Around line 152-159: Move the cleanup variable declarations and defer block in
Service.Start above the certOption and newAuthorizer calls, ensuring failures
from newAuthorizer are covered while preserving the existing cleanup behavior
for later returns.
---
Nitpick comments:
In `@rest-api/flow/internal/authz/interceptor_test.go`:
- Around line 18-89: Expand interceptor tests to cover missing TLS peer
authentication and stream authorization outcomes. In TestUnaryServerInterceptor,
add a no-peer case expecting codes.Unauthenticated with handlerRuns false; add
stream tests for ModeEnforce denial and audit-mode denial, asserting the
expected status and whether the handler runs while preserving identity
propagation for allowed requests. Reuse the existing TLS context helpers and
symbols such as StreamServerInterceptor and checkAuthorizationResult’s expected
behavior.
In `@rest-api/flow/internal/authz/interceptor.go`:
- Around line 65-91: Convert checkAuthorizationResult into an Authorizer method
so it uses the receiver’s mode instead of accepting mode as a parameter. Update
both callers to invoke the method through their Authorizer instance, remove the
redundant mode argument, and preserve the existing authorization logging,
identity handling, and enforcement behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 4752123f-ac77-403f-b749-86750b925d1b
⛔ Files ignored due to path filters (1)
rest-api/go.sumis excluded by!**/*.sum
📒 Files selected for processing (13)
rest-api/flow/cmd/serve.gorest-api/flow/cmd/serve_test.gorest-api/flow/internal/authz/authorizer.gorest-api/flow/internal/authz/authorizer_test.gorest-api/flow/internal/authz/identity.gorest-api/flow/internal/authz/interceptor.gorest-api/flow/internal/authz/interceptor_test.gorest-api/flow/internal/common/grpcrecovery/recovery.gorest-api/flow/internal/common/grpcrecovery/recovery_test.gorest-api/flow/internal/service/config.gorest-api/flow/internal/service/config_test.gorest-api/flow/internal/service/service.gorest-api/go.mod
| func recoverPanic(ctx context.Context, method string, err *error) { | ||
| if recovered := recover(); recovered != nil { | ||
| event := log.Error(). | ||
| Str("grpc_method", method). | ||
| Interface("panic", recovered). | ||
| Str("stack", string(debug.Stack())) | ||
|
|
||
| if identity, ok := authz.ServiceIdentityFromContext(ctx); ok { | ||
| event = event.Str("service_identity", identity) | ||
| } | ||
|
|
||
| event.Msg("Recovered panic while handling Flow gRPC request") | ||
| *err = status.Error(codes.Internal, internalErrorMessage) | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Service identity is never attached to panic logs — dead code path.
authz.ServiceIdentityFromContext(ctx) reads the ctx captured at interceptor entry, but authz.UnaryServerInterceptor/StreamServerInterceptor (which run inside this interceptor per the chain in service.go) only attach the identity to a newly derived context that they pass further down to their own handler — it never propagates back up to this closure's ctx. Context values flow strictly downward through the call graph, so ok here will always be false and service_identity will never appear in a recovered-panic log entry, even though the feature is advertised as working.
See the consolidated comment on rest-api/flow/internal/service/service.go (interceptor chain ordering) for the root cause and proposed fix.
Separately, note that this cross-cutting recovery utility now depends on the domain-specific authz package purely for optional log enrichment — consider inverting this via an injectable enrichment hook if this pattern grows.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rest-api/flow/internal/common/grpcrecovery/recovery.go` around lines 52 - 66,
The service identity lookup in recoverPanic cannot observe the context enriched
by downstream interceptors, so remove this ineffective
authz.ServiceIdentityFromContext enrichment from the recovery logger and
implement the interceptor-chain fix described in service.go so the enriched
context is available when panic recovery logs execute. Keep panic status
conversion and existing log fields unchanged.
| func (s *Service) Start(ctx context.Context) (retErr error) { | ||
| log.Logger = log.With().Caller().Logger() | ||
|
|
||
| certOpt := s.certOption() | ||
| certOpt, secure := s.certOption() | ||
| authorizer, err := s.newAuthorizer(secure) | ||
| if err != nil { | ||
| return err | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
DB session leak if authorizer creation fails.
s.newAuthorizer(secure) can return a non-fatal error (line 156-159), but this return err happens before the cleanup defer (line 171) is registered. Since s.session is already open at this point (opened in New()), a newAuthorizer failure leaks the DB session/connection pool — none of the deferred cleanup runs.
Move the var (...)/defer block above the certOption()/newAuthorizer() calls so any early return is covered by cleanup.
🛠️ Proposed fix
func (s *Service) Start(ctx context.Context) (retErr error) {
log.Logger = log.With().Caller().Logger()
- certOpt, secure := s.certOption()
- authorizer, err := s.newAuthorizer(secure)
- if err != nil {
- return err
- }
-
// On any error return, shut down every resource that was started, in
// reverse start order. Boolean flags record which components are running
// because inventoryManager and taskManager are always non-nil (set in New)
// so a nil check cannot distinguish "created" from "started".
var (
invStarted bool
taskStarted bool
lis net.Listener
)
defer func() {
if retErr == nil {
return
}
...
s.session.Close()
}()
+ certOpt, secure := s.certOption()
+ authorizer, err := s.newAuthorizer(secure)
+ if err != nil {
+ return err
+ }
+
// Rule resolver is ready immediately (queries DB for rules)
log.Info().Msg("Rule resolver ready (will query DB for operation rules)")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func (s *Service) Start(ctx context.Context) (retErr error) { | |
| log.Logger = log.With().Caller().Logger() | |
| certOpt := s.certOption() | |
| certOpt, secure := s.certOption() | |
| authorizer, err := s.newAuthorizer(secure) | |
| if err != nil { | |
| return err | |
| } | |
| func (s *Service) Start(ctx context.Context) (retErr error) { | |
| log.Logger = log.With().Caller().Logger() | |
| // On any error return, shut down every resource that was started, in | |
| // reverse start order. Boolean flags record which components are running | |
| // because inventoryManager and taskManager are always non-nil (set in New) | |
| // so a nil check cannot distinguish "created" from "started". | |
| var ( | |
| invStarted bool | |
| taskStarted bool | |
| lis net.Listener | |
| ) | |
| defer func() { | |
| if retErr == nil { | |
| return | |
| } | |
| ... | |
| s.session.Close() | |
| }() | |
| certOpt, secure := s.certOption() | |
| authorizer, err := s.newAuthorizer(secure) | |
| if err != nil { | |
| return err | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rest-api/flow/internal/service/service.go` around lines 152 - 159, Move the
cleanup variable declarations and defer block in Service.Start above the
certOption and newAuthorizer calls, ensuring failures from newAuthorizer are
covered while preserving the existing cleanup behavior for later returns.
There was a problem hiding this comment.
+1
Low possibility to happen but better fix as suggested.
There was a problem hiding this comment.
@kunzhao-nv, agreed. Moving the cleanup setup before newAuthorizer is a small, production-safe change that ensures the already-open database session is closed on this early startup failure path. The finding remains open until the change is applied and verified.
🐇✅
| s.grpcServer = grpc.NewServer( | ||
| certOpt, | ||
| grpc.ChainUnaryInterceptor(grpclog.UnaryServerInterceptor()), | ||
| grpc.ChainUnaryInterceptor( | ||
| grpcrecovery.UnaryServerInterceptor(), | ||
| authz.UnaryServerInterceptor(authorizer), | ||
| grpclog.UnaryServerInterceptor(), | ||
| ), | ||
| grpc.ChainStreamInterceptor( | ||
| grpcrecovery.StreamServerInterceptor(), | ||
| authz.StreamServerInterceptor(authorizer), | ||
| ), | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Interceptor ordering makes grpcrecovery's service-identity log enrichment permanently ineffective.
grpcrecovery.UnaryServerInterceptor()/StreamServerInterceptor() are the outermost interceptors and capture ctx before authz.UnaryServerInterceptor attaches the service identity via withServiceIdentity on a newly derived context passed further down the chain. Because context values only flow downward, grpcrecovery's recoverPanic will never observe the identity attached by authz — see rest-api/flow/internal/common/grpcrecovery/recovery.go lines 52-66.
Swapping the order (authz outermost, grpcrecovery next) would let recovered-panic logs include service_identity, at the cost of no longer recovering panics that occur inside the authz interceptor itself. Given authorize()/checkAuthorizationResult() are pure, deterministic logic, this tradeoff is likely acceptable, but please confirm before reordering, and add an integration-style test that panics inside a handler with an authenticated identity to assert service_identity appears in the log.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rest-api/flow/internal/service/service.go` around lines 273 - 284, Reorder
the interceptor chains in the gRPC server setup so authz.UnaryServerInterceptor
and authz.StreamServerInterceptor are outermost, followed by the corresponding
grpcrecovery interceptors, allowing recovered handler panics to observe the
service identity added by authz. Add an integration-style test covering an
authenticated handler panic and assert the recovery log includes
service_identity.
There was a problem hiding this comment.
+1
if identity, ok := authz.ServiceIdentityFromContext(ctx); ok {
event = event.Str("service_identity", identity)
}
ok is always false here, service_identity will never show up.
There was a problem hiding this comment.
@kunzhao-nv, agreed. grpcrecovery captures the pre-authorization context, while authz adds service_identity only to a derived context passed inward. Consequently, ServiceIdentityFromContext(ctx) in recovery always returns ok == false for handler panics.
The interceptor order should be changed to make authz outermost, followed by recovery, for both unary and stream RPCs. We should also add coverage for an authenticated handler panic that verifies the recovery log includes service_identity.
🐇
| }) | ||
| require.NoError(t, err) | ||
|
|
||
| tests := map[string]struct { |
There was a problem hiding this comment.
nit: add one more case:
"no TLS peer": {ctx: context.Background(), wantCode: codes.Unauthenticated, handlerRuns: false}
| } | ||
| } | ||
|
|
||
| func TestStreamServerInterceptorPropagatesServiceIdentity(t *testing.T) { |
There was a problem hiding this comment.
nit:
add symmetric stream coverage for the ModeEnforce-denied and audit-mode-denied cases (asserting the returned code and whether the handler runs).
| } | ||
| } | ||
|
|
||
| func checkAuthorizationResult(method string, authorization *decision, mode Mode) error { |
There was a problem hiding this comment.
nit:
checkAuthorizationResult only ever uses authorizer.mode, which is threaded manually from both call sites. Consider making it a method on *Authorizer and dropping the mode parameter, so the decision logic stays owned by Authorizer alongside authorize():
func (a *Authorizer) checkAuthorizationResult(method string, d *decision) error
2 participants
Flow currently uses mutual TLS to authenticate gRPC connections, but authentication alone does not determine whether a trusted workload is allowed to call Flow. Any client certificate issued by a trusted certificate authority can reach the service, even when the certificate belongs to an unrelated workload. This leaves the Flow gRPC boundary broader than intended and makes it difficult to identify or restrict the services that depend on Flow.
This change adds authorization based on the caller's SPIFFE workload identity. Flow extracts the SPIFFE ID from the verified client certificate and compares the complete identity against an explicit service allowlist. The check applies to both unary and streaming RPCs, and an accepted identity is attached to the request context for consistent diagnostics and future request auditing.
The allowlist is deployment configuration rather than compiled policy because SPIFFE trust domains, Kubernetes namespaces, and service accounts vary between environments. Operators can provide identities through repeated --allowed-service-identity options or point FLOW_AUTHORIZATION_ALLOWED_SERVICE_IDENTITIES_FILE at a mounted plain-text file containing one SPIFFE ID per line. File-based configuration supports a Kubernetes ConfigMap mount without requiring Flow to understand Kubernetes manifests.
Authorization defaults to AUDIT mode so the feature can be deployed before all caller identities and ConfigMap mounts are available. Audit mode records the identity and the gRPC result that would be rejected, then allows the request to continue. After operators confirm the observed caller identities, setting FLOW_AUTHORIZATION_MODE=ENFORCE rejects callers that cannot present a verified SPIFFE identity or whose identity is absent from the allowlist.
Development environments may continue to run without TLS when ALLOW_INSECURE_GRPC=true is explicitly set. This path is limited to audit mode with an empty allowlist; staging and production still require TLS. Because an insecure caller has no authenticated identity, audit logs identify it with the audit-unidentified fallback.
The gRPC interceptor chain also gains panic recovery. A request-scoped panic in authorization, logging, or a handler is converted into a generic Internal response and logged server-side rather than terminating the Flow process or returning internal details to the caller.
Together, these changes provide a gradual path from observing current callers to enforcing least-privilege access at the Flow gRPC boundary without blocking deployment on the immediate availability of Kubernetes authorization manifests.
Related issues
Type of Change
Breaking Changes
Testing
Additional Notes