diff --git a/authbridge/authlib/config/config.go b/authbridge/authlib/config/config.go index 39cefac23..d6d021161 100644 --- a/authbridge/authlib/config/config.go +++ b/authbridge/authlib/config/config.go @@ -18,9 +18,23 @@ type Config struct { Listener ListenerConfig `yaml:"listener" json:"listener"` Bypass BypassConfig `yaml:"bypass" json:"bypass"` Routes RoutesConfig `yaml:"routes" json:"routes"` + Pipeline PipelineConfig `yaml:"pipeline" json:"pipeline"` Stats StatsConfig `yaml:"stats" json:"stats"` } +// PipelineConfig holds the plugin pipeline composition. +// If omitted (empty), default pipelines are used: +// inbound=[jwt-validation], outbound=[token-exchange]. +type PipelineConfig struct { + Inbound PipelineStageConfig `yaml:"inbound" json:"inbound"` + Outbound PipelineStageConfig `yaml:"outbound" json:"outbound"` +} + +// PipelineStageConfig lists the plugins for a pipeline stage in execution order. +type PipelineStageConfig struct { + Plugins []string `yaml:"plugins" json:"plugins"` +} + // InboundConfig holds JWT validation settings. type InboundConfig struct { JWKSURL string `yaml:"jwks_url" json:"jwks_url"` diff --git a/authbridge/authlib/plugins/defaults.go b/authbridge/authlib/plugins/defaults.go new file mode 100644 index 000000000..7467f62d7 --- /dev/null +++ b/authbridge/authlib/plugins/defaults.go @@ -0,0 +1,27 @@ +package plugins + +import ( + "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/routing" +) + +var ( + DefaultInboundPlugins = []string{"jwt-validation"} + DefaultOutboundPlugins = []string{"token-exchange"} +) + +func DefaultInboundPipeline(a *auth.Auth) (*pipeline.Pipeline, error) { + return Build(DefaultInboundPlugins, a) +} + +func DefaultOutboundPipeline(a *auth.Auth) (*pipeline.Pipeline, error) { + return Build(DefaultOutboundPlugins, a) +} + +// WaypointInboundPipeline creates an inbound pipeline for waypoint mode +// where audience is derived per-request from the destination host. +func WaypointInboundPipeline(a *auth.Auth) (*pipeline.Pipeline, error) { + jwtPlugin := NewJWTValidation(a, WithAudienceDeriver(routing.ServiceNameFromHost)) + return pipeline.New([]pipeline.Plugin{jwtPlugin}) +} diff --git a/authbridge/authlib/plugins/jwtvalidation.go b/authbridge/authlib/plugins/jwtvalidation.go new file mode 100644 index 000000000..64565603a --- /dev/null +++ b/authbridge/authlib/plugins/jwtvalidation.go @@ -0,0 +1,59 @@ +package plugins + +import ( + "context" + + "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" +) + +// JWTValidation is a pipeline plugin that validates inbound JWTs. +// It delegates to auth.HandleInbound and populates pctx.Claims on success. +type JWTValidation struct { + auth *auth.Auth + audienceDeriver func(string) string +} + +// JWTValidationOption configures the JWTValidation plugin. +type JWTValidationOption func(*JWTValidation) + +// WithAudienceDeriver sets a function that derives the expected JWT audience +// from the request host. Used in waypoint mode where audience varies per-request. +func WithAudienceDeriver(f func(string) string) JWTValidationOption { + return func(j *JWTValidation) { j.audienceDeriver = f } +} + +func NewJWTValidation(a *auth.Auth, opts ...JWTValidationOption) *JWTValidation { + p := &JWTValidation{auth: a} + for _, opt := range opts { + opt(p) + } + return p +} + +func (p *JWTValidation) Name() string { return "jwt-validation" } + +func (p *JWTValidation) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{Writes: []string{"security"}} +} + +func (p *JWTValidation) OnRequest(ctx context.Context, pctx *pipeline.Context) pipeline.Action { + authHeader := pctx.Headers.Get("Authorization") + path := pctx.Path + + var audience string + if p.audienceDeriver != nil { + audience = p.audienceDeriver(pctx.Host) + } + + result := p.auth.HandleInbound(ctx, authHeader, path, audience) + if result.Action == auth.ActionDeny { + return pipeline.Action{Type: pipeline.Reject, Status: result.DenyStatus, Reason: result.DenyReason} + } + pctx.Claims = result.Claims + return pipeline.Action{Type: pipeline.Continue} +} + +func (p *JWTValidation) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} diff --git a/authbridge/authlib/plugins/plugins_test.go b/authbridge/authlib/plugins/plugins_test.go new file mode 100644 index 000000000..fa0d8b158 --- /dev/null +++ b/authbridge/authlib/plugins/plugins_test.go @@ -0,0 +1,303 @@ +package plugins + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/cache" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/exchange" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/routing" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/validation" +) + +type mockVerifier struct { + claims *validation.Claims + err error +} + +func (m *mockVerifier) Verify(_ context.Context, _ string, _ string) (*validation.Claims, error) { + return m.claims, m.err +} + +// --- JWTValidation Tests --- + +func TestJWTValidation_ValidToken(t *testing.T) { + a := auth.New(auth.Config{ + Verifier: &mockVerifier{claims: &validation.Claims{Subject: "user-1", ClientID: "agent"}}, + Identity: auth.IdentityConfig{Audience: "my-agent"}, + }) + plugin := NewJWTValidation(a) + + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Path: "/api/test", + Headers: http.Header{"Authorization": []string{"Bearer valid-token"}}, + } + action := plugin.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Continue { + t.Errorf("got %v, want Continue", action.Type) + } + if pctx.Claims == nil { + t.Fatal("expected pctx.Claims to be populated") + } + if pctx.Claims.Subject != "user-1" { + t.Errorf("subject = %q, want user-1", pctx.Claims.Subject) + } +} + +func TestJWTValidation_InvalidToken(t *testing.T) { + a := auth.New(auth.Config{ + Verifier: &mockVerifier{err: errorf("token expired")}, + Identity: auth.IdentityConfig{Audience: "my-agent"}, + }) + plugin := NewJWTValidation(a) + + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Path: "/api/test", + Headers: http.Header{"Authorization": []string{"Bearer bad-token"}}, + } + action := plugin.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Reject { + t.Fatalf("got %v, want Reject", action.Type) + } + if action.Status != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", action.Status) + } +} + +func TestJWTValidation_MissingHeader(t *testing.T) { + a := auth.New(auth.Config{ + Verifier: &mockVerifier{claims: &validation.Claims{Subject: "user"}}, + Identity: auth.IdentityConfig{Audience: "my-agent"}, + }) + plugin := NewJWTValidation(a) + + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Path: "/api/test", + Headers: http.Header{}, + } + action := plugin.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Reject { + t.Fatalf("got %v, want Reject", action.Type) + } + if action.Status != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", action.Status) + } +} + +func TestJWTValidation_WithAudienceDeriver(t *testing.T) { + var receivedAudience string + verifier := &mockVerifier{claims: &validation.Claims{Subject: "user"}} + a := auth.New(auth.Config{ + Verifier: &captureAudienceVerifier{inner: verifier, captured: &receivedAudience}, + Identity: auth.IdentityConfig{Audience: "default-aud"}, + }) + plugin := NewJWTValidation(a, WithAudienceDeriver(func(host string) string { + return "derived-from-" + host + })) + + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Host: "target-svc", + Path: "/api", + Headers: http.Header{"Authorization": []string{"Bearer token"}}, + } + action := plugin.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Continue { + t.Fatalf("got %v, want Continue", action.Type) + } + if receivedAudience != "derived-from-target-svc" { + t.Errorf("audience = %q, want derived-from-target-svc", receivedAudience) + } +} + +// --- TokenExchange Tests --- + +func TestTokenExchange_Passthrough(t *testing.T) { + router, _ := routing.NewRouter("passthrough", []routing.Route{}) + a := auth.New(auth.Config{Router: router}) + plugin := NewTokenExchange(a) + + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + Host: "some-host", + Headers: http.Header{"Authorization": []string{"Bearer user-token"}}, + } + action := plugin.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Continue { + t.Errorf("got %v, want Continue", action.Type) + } + if pctx.Headers.Get("Authorization") != "Bearer user-token" { + t.Error("headers should not be modified for passthrough") + } +} + +func TestTokenExchange_ExchangeSuccess(t *testing.T) { + exchangeSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "access_token": "new-token", + "token_type": "Bearer", + "expires_in": 300, + }) + })) + defer exchangeSrv.Close() + + router, _ := routing.NewRouter("exchange", []routing.Route{}) + exchanger := exchange.NewClient(exchangeSrv.URL, &exchange.ClientSecretAuth{ + ClientID: "agent", ClientSecret: "secret", + }) + a := auth.New(auth.Config{ + Router: router, + Exchanger: exchanger, + Cache: cache.New(), + }) + plugin := NewTokenExchange(a) + + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + Host: "target-svc", + Headers: http.Header{"Authorization": []string{"Bearer user-token"}}, + } + action := plugin.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Continue { + t.Fatalf("got %v, want Continue", action.Type) + } + if pctx.Headers.Get("Authorization") != "Bearer new-token" { + t.Errorf("token = %q, want Bearer new-token", pctx.Headers.Get("Authorization")) + } +} + +func TestTokenExchange_ExchangeFailure(t *testing.T) { + exchangeSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(`{"error":"invalid_grant"}`)) + })) + defer exchangeSrv.Close() + + router, _ := routing.NewRouter("exchange", []routing.Route{}) + exchanger := exchange.NewClient(exchangeSrv.URL, &exchange.ClientSecretAuth{ + ClientID: "agent", ClientSecret: "secret", + }) + a := auth.New(auth.Config{ + Router: router, + Exchanger: exchanger, + }) + plugin := NewTokenExchange(a) + + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + Host: "target-svc", + Headers: http.Header{"Authorization": []string{"Bearer user-token"}}, + } + action := plugin.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Reject { + t.Fatalf("got %v, want Reject", action.Type) + } + if action.Status != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503", action.Status) + } +} + +func TestTokenExchange_NoToken_Deny(t *testing.T) { + router, _ := routing.NewRouter("exchange", []routing.Route{}) + a := auth.New(auth.Config{ + Router: router, + NoTokenPolicy: auth.NoTokenPolicyDeny, + }) + plugin := NewTokenExchange(a) + + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + Host: "target-svc", + Headers: http.Header{}, + } + action := plugin.OnRequest(context.Background(), pctx) + if action.Type != pipeline.Reject { + t.Fatalf("got %v, want Reject", action.Type) + } + if action.Status != http.StatusUnauthorized { + t.Errorf("status = %d, want 401", action.Status) + } +} + +// --- Registry/Build Tests --- + +func TestBuild_ValidNames(t *testing.T) { + a := auth.New(auth.Config{}) + p, err := Build([]string{"jwt-validation", "token-exchange"}, a) + if err != nil { + t.Fatalf("Build returned error: %v", err) + } + if p == nil { + t.Fatal("expected non-nil pipeline") + } +} + +func TestBuild_UnknownName(t *testing.T) { + a := auth.New(auth.Config{}) + _, err := Build([]string{"nonexistent-plugin"}, a) + if err == nil { + t.Fatal("expected error for unknown plugin name") + } +} + +func TestBuild_EmptyList(t *testing.T) { + a := auth.New(auth.Config{}) + p, err := Build([]string{}, a) + if err != nil { + t.Fatalf("Build returned error: %v", err) + } + action := p.Run(context.Background(), &pipeline.Context{Headers: http.Header{}}) + if action.Type != pipeline.Continue { + t.Errorf("empty pipeline got %v, want Continue", action.Type) + } +} + +func TestDefaultInboundPipeline(t *testing.T) { + a := auth.New(auth.Config{}) + p, err := DefaultInboundPipeline(a) + if err != nil { + t.Fatalf("DefaultInboundPipeline returned error: %v", err) + } + if p == nil { + t.Fatal("expected non-nil pipeline") + } +} + +func TestDefaultOutboundPipeline(t *testing.T) { + a := auth.New(auth.Config{}) + p, err := DefaultOutboundPipeline(a) + if err != nil { + t.Fatalf("DefaultOutboundPipeline returned error: %v", err) + } + if p == nil { + t.Fatal("expected non-nil pipeline") + } +} + +// --- Helpers --- + +type errString string + +func errorf(s string) error { return errString(s) } + +func (e errString) Error() string { return string(e) } + +// captureAudienceVerifier wraps a verifier and captures the audience parameter. +type captureAudienceVerifier struct { + inner *mockVerifier + captured *string +} + +func (v *captureAudienceVerifier) Verify(ctx context.Context, token string, audience string) (*validation.Claims, error) { + *v.captured = audience + return v.inner.Verify(ctx, token, audience) +} diff --git a/authbridge/authlib/plugins/registry.go b/authbridge/authlib/plugins/registry.go new file mode 100644 index 000000000..0f6e69525 --- /dev/null +++ b/authbridge/authlib/plugins/registry.go @@ -0,0 +1,30 @@ +package plugins + +import ( + "fmt" + + "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" +) + +// PluginFactory creates a pipeline plugin from an auth.Auth instance. +type PluginFactory func(a *auth.Auth) pipeline.Plugin + +var registry = map[string]PluginFactory{ + "jwt-validation": func(a *auth.Auth) pipeline.Plugin { return NewJWTValidation(a) }, + "token-exchange": func(a *auth.Auth) pipeline.Plugin { return NewTokenExchange(a) }, +} + +// Build constructs a pipeline from an ordered list of plugin names. +// Returns an error if any name is not in the registry. +func Build(names []string, a *auth.Auth, opts ...pipeline.Option) (*pipeline.Pipeline, error) { + ps := make([]pipeline.Plugin, 0, len(names)) + for _, name := range names { + factory, ok := registry[name] + if !ok { + return nil, fmt.Errorf("unknown plugin %q", name) + } + ps = append(ps, factory(a)) + } + return pipeline.New(ps, opts...) +} diff --git a/authbridge/authlib/plugins/tokenexchange.go b/authbridge/authlib/plugins/tokenexchange.go new file mode 100644 index 000000000..9904f338c --- /dev/null +++ b/authbridge/authlib/plugins/tokenexchange.go @@ -0,0 +1,42 @@ +package plugins + +import ( + "context" + + "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" +) + +// TokenExchange is a pipeline plugin that performs outbound token exchange. +// It delegates to auth.HandleOutbound and mutates pctx.Headers on token replacement. +type TokenExchange struct { + auth *auth.Auth +} + +func NewTokenExchange(a *auth.Auth) *TokenExchange { + return &TokenExchange{auth: a} +} + +func (p *TokenExchange) Name() string { return "token-exchange" } + +func (p *TokenExchange) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} + +func (p *TokenExchange) OnRequest(ctx context.Context, pctx *pipeline.Context) pipeline.Action { + authHeader := pctx.Headers.Get("Authorization") + host := pctx.Host + + result := p.auth.HandleOutbound(ctx, authHeader, host) + switch result.Action { + case auth.ActionDeny: + return pipeline.Action{Type: pipeline.Reject, Status: result.DenyStatus, Reason: result.DenyReason} + case auth.ActionReplaceToken: + pctx.Headers.Set("Authorization", "Bearer "+result.Token) + } + return pipeline.Action{Type: pipeline.Continue} +} + +func (p *TokenExchange) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} diff --git a/authbridge/cmd/authbridge/listener/extauthz/server.go b/authbridge/cmd/authbridge/listener/extauthz/server.go index fa77503ea..773c1925e 100644 --- a/authbridge/cmd/authbridge/listener/extauthz/server.go +++ b/authbridge/cmd/authbridge/listener/extauthz/server.go @@ -6,6 +6,8 @@ package extauthz import ( "context" "encoding/json" + "net/http" + "strings" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" authv3 "github.com/envoyproxy/go-control-plane/envoy/service/auth/v3" @@ -14,14 +16,14 @@ import ( rpcstatus "google.golang.org/genproto/googleapis/rpc/status" - authpkg "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" - "github.com/kagenti/kagenti-extensions/authbridge/authlib/routing" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" ) // Server implements the Envoy ext_authz Authorization gRPC service. type Server struct { authv3.UnimplementedAuthorizationServer - Auth *authpkg.Auth + InboundPipeline *pipeline.Pipeline + OutboundPipeline *pipeline.Pipeline } // Check handles a single ext_authz authorization request. @@ -36,30 +38,54 @@ func (s *Server) Check(ctx context.Context, req *authv3.CheckRequest) (*authv3.C if host == "" { host = headers["host"] } - authHeader := headers["authorization"] path := httpReq.GetPath() - // Derive audience from destination host (waypoint pattern) - audience := routing.ServiceNameFromHost(host) + // Inbound validation via pipeline + inPctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Host: host, + Path: path, + Headers: mapToHTTPHeader(headers), + } + inAction := s.InboundPipeline.Run(ctx, inPctx) + if inAction.Type == pipeline.Reject { + return denied(codes.Unauthenticated, inAction.Status, inAction.Reason), nil + } + + // Outbound exchange via pipeline + outPctx := &pipeline.Context{ + Direction: pipeline.Outbound, + Host: host, + Path: path, + Headers: mapToHTTPHeader(headers), + } + originalAuth := outPctx.Headers.Get("Authorization") + outAction := s.OutboundPipeline.Run(ctx, outPctx) + if outAction.Type == pipeline.Reject { + return denied(codes.PermissionDenied, outAction.Status, outAction.Reason), nil + } - // Inbound validation - inResult := s.Auth.HandleInbound(ctx, authHeader, path, audience) - if inResult.Action == authpkg.ActionDeny { - return denied(codes.Unauthenticated, inResult.DenyStatus, inResult.DenyReason), nil + newAuth := outPctx.Headers.Get("Authorization") + if newAuth != originalAuth { + return allowedWithToken(extractBearer(newAuth)), nil } + return allowed(), nil +} - // Outbound exchange - outResult := s.Auth.HandleOutbound(ctx, authHeader, host) - switch outResult.Action { - case authpkg.ActionReplaceToken: - return allowedWithToken(outResult.Token), nil - case authpkg.ActionDeny: - return denied(codes.PermissionDenied, outResult.DenyStatus, outResult.DenyReason), nil - default: - return allowed(), nil +func mapToHTTPHeader(m map[string]string) http.Header { + h := make(http.Header) + for k, v := range m { + h.Set(k, v) } + return h } +func extractBearer(authHeader string) string { + if len(authHeader) > 7 && strings.EqualFold(authHeader[:7], "bearer ") { + return authHeader[7:] + } + return "" +} func denied(code codes.Code, httpStatus int, msg string) *authv3.CheckResponse { body, _ := json.Marshal(map[string]string{"error": msg}) diff --git a/authbridge/cmd/authbridge/listener/extauthz/server_test.go b/authbridge/cmd/authbridge/listener/extauthz/server_test.go index 5eb809dbc..d51697cac 100644 --- a/authbridge/cmd/authbridge/listener/extauthz/server_test.go +++ b/authbridge/cmd/authbridge/listener/extauthz/server_test.go @@ -14,6 +14,7 @@ import ( authpkg "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" "github.com/kagenti/kagenti-extensions/authbridge/authlib/cache" "github.com/kagenti/kagenti-extensions/authbridge/authlib/exchange" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins" "github.com/kagenti/kagenti-extensions/authbridge/authlib/routing" "github.com/kagenti/kagenti-extensions/authbridge/authlib/validation" ) @@ -29,6 +30,20 @@ func (m *mockVerifier) Verify(_ context.Context, _ string, audience string) (*va return m.claims, m.err } +func serverFromAuth(t *testing.T, a *authpkg.Auth) *Server { + t.Helper() + // ext_authz is waypoint mode — audience derived from host + inbound, err := plugins.WaypointInboundPipeline(a) + if err != nil { + t.Fatalf("building inbound pipeline: %v", err) + } + outbound, err := plugins.DefaultOutboundPipeline(a) + if err != nil { + t.Fatalf("building outbound pipeline: %v", err) + } + return &Server{InboundPipeline: inbound, OutboundPipeline: outbound} +} + func checkRequest(host, path, authHeader string) *authv3.CheckRequest { headers := map[string]string{ ":authority": host, @@ -76,7 +91,7 @@ func TestCheck_ValidToken_Exchange(t *testing.T) { Cache: cache.New(), Identity: authpkg.IdentityConfig{Audience: "default-aud"}, }) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) resp, err := srv.Check(context.Background(), checkRequest("auth-target-service.authbridge.svc:8081", "/api/test", "Bearer user-token")) @@ -107,7 +122,7 @@ func TestCheck_InvalidToken(t *testing.T) { Verifier: &mockVerifier{err: fmt.Errorf("bad token")}, Identity: authpkg.IdentityConfig{Audience: "aud"}, }) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) resp, _ := srv.Check(context.Background(), checkRequest("svc", "/api", "Bearer bad")) @@ -123,7 +138,7 @@ func TestCheck_InvalidToken(t *testing.T) { func TestCheck_MissingHTTPAttributes(t *testing.T) { a := authpkg.New(authpkg.Config{}) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) resp, _ := srv.Check(context.Background(), &authv3.CheckRequest{}) diff --git a/authbridge/cmd/authbridge/listener/extproc/server.go b/authbridge/cmd/authbridge/listener/extproc/server.go index 2774bfd4a..432917cfd 100644 --- a/authbridge/cmd/authbridge/listener/extproc/server.go +++ b/authbridge/cmd/authbridge/listener/extproc/server.go @@ -1,11 +1,11 @@ // Package extproc implements an Envoy ext_proc gRPC streaming listener. -// It translates ext_proc ProcessingRequests into auth.HandleInbound/HandleOutbound -// calls and maps the results back to ProcessingResponses. +// It translates ext_proc ProcessingRequests into pipeline runs and maps +// the results back to ProcessingResponses. package extproc import ( - "context" "encoding/json" + "net/http" "strings" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" @@ -14,13 +14,14 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/status" - "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" ) // Server implements the Envoy ext_proc ExternalProcessor gRPC service. type Server struct { extprocv3.UnimplementedExternalProcessorServer - Auth *auth.Auth + InboundPipeline *pipeline.Pipeline + OutboundPipeline *pipeline.Pipeline } // Process handles the bidirectional ext_proc stream. @@ -46,9 +47,9 @@ func (s *Server) Process(stream extprocv3.ExternalProcessor_ProcessServer) error direction := getHeader(headers, "x-authbridge-direction") if direction == "inbound" { - resp = s.handleInbound(ctx, headers) + resp = s.handleInbound(stream, headers) } else { - resp = s.handleOutbound(ctx, headers) + resp = s.handleOutbound(stream, headers) } case *extprocv3.ProcessingRequest_ResponseHeaders: @@ -68,38 +69,63 @@ func (s *Server) Process(stream extprocv3.ExternalProcessor_ProcessServer) error } } -func (s *Server) handleInbound(ctx context.Context, headers *corev3.HeaderMap) *extprocv3.ProcessingResponse { - authHeader := getHeader(headers, "authorization") - path := getHeader(headers, ":path") - - result := s.Auth.HandleInbound(ctx, authHeader, path, "") +func (s *Server) handleInbound(stream extprocv3.ExternalProcessor_ProcessServer, headers *corev3.HeaderMap) *extprocv3.ProcessingResponse { + ctx := stream.Context() + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Path: getHeader(headers, ":path"), + Headers: headerMapToHTTP(headers), + } - if result.Action == auth.ActionDeny { - return denyResponse(typev3.StatusCode_Unauthorized, - jsonError("unauthorized", result.DenyReason)) + action := s.InboundPipeline.Run(ctx, pctx) + if action.Type == pipeline.Reject { + return denyResponse(typev3.StatusCode(action.Status), + jsonError("unauthorized", action.Reason)) } return allowResponse() } -func (s *Server) handleOutbound(ctx context.Context, headers *corev3.HeaderMap) *extprocv3.ProcessingResponse { - authHeader := getHeader(headers, "authorization") - host := getHeader(headers, ":authority") - if host == "" { - host = getHeader(headers, "host") +func (s *Server) handleOutbound(stream extprocv3.ExternalProcessor_ProcessServer, headers *corev3.HeaderMap) *extprocv3.ProcessingResponse { + ctx := stream.Context() + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + Host: getHeader(headers, ":authority"), + Headers: headerMapToHTTP(headers), + } + if pctx.Host == "" { + pctx.Host = getHeader(headers, "host") } - result := s.Auth.HandleOutbound(ctx, authHeader, host) - - switch result.Action { - case auth.ActionReplaceToken: - return replaceTokenResponse(result.Token) - case auth.ActionDeny: + originalAuth := pctx.Headers.Get("Authorization") + action := s.OutboundPipeline.Run(ctx, pctx) + if action.Type == pipeline.Reject { return denyResponse(typev3.StatusCode_ServiceUnavailable, - jsonError("token_acquisition_failed", result.DenyReason)) - default: - return passResponse() + jsonError("token_acquisition_failed", action.Reason)) + } + + newAuth := pctx.Headers.Get("Authorization") + if newAuth != originalAuth { + return replaceTokenResponse(extractBearer(newAuth)) } + return passResponse() +} + +func headerMapToHTTP(headers *corev3.HeaderMap) http.Header { + h := make(http.Header) + if headers != nil { + for _, hdr := range headers.Headers { + h.Set(hdr.Key, string(hdr.RawValue)) + } + } + return h +} + +func extractBearer(authHeader string) string { + if len(authHeader) > 7 && strings.EqualFold(authHeader[:7], "bearer ") { + return authHeader[7:] + } + return "" } func allowResponse() *extprocv3.ProcessingResponse { diff --git a/authbridge/cmd/authbridge/listener/extproc/server_test.go b/authbridge/cmd/authbridge/listener/extproc/server_test.go index baadb6f65..0efb16020 100644 --- a/authbridge/cmd/authbridge/listener/extproc/server_test.go +++ b/authbridge/cmd/authbridge/listener/extproc/server_test.go @@ -16,6 +16,7 @@ import ( "github.com/kagenti/kagenti-extensions/authbridge/authlib/bypass" "github.com/kagenti/kagenti-extensions/authbridge/authlib/cache" "github.com/kagenti/kagenti-extensions/authbridge/authlib/exchange" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins" "github.com/kagenti/kagenti-extensions/authbridge/authlib/routing" "github.com/kagenti/kagenti-extensions/authbridge/authlib/validation" ) @@ -43,10 +44,10 @@ func (m *mockStream) Recv() (*extprocv3.ProcessingRequest, error) { return req, nil } func (m *mockStream) SetHeader(metadata.MD) error { return nil } -func (m *mockStream) SendHeader(metadata.MD) error { return nil } -func (m *mockStream) SetTrailer(metadata.MD) {} -func (m *mockStream) SendMsg(any) error { return nil } -func (m *mockStream) RecvMsg(any) error { return nil } +func (m *mockStream) SendHeader(metadata.MD) error { return nil } +func (m *mockStream) SetTrailer(metadata.MD) {} +func (m *mockStream) SendMsg(any) error { return nil } +func (m *mockStream) RecvMsg(any) error { return nil } type mockVerifier struct { claims *validation.Claims @@ -57,6 +58,19 @@ func (v *mockVerifier) Verify(_ context.Context, _ string, _ string) (*validatio return v.claims, v.err } +func serverFromAuth(t *testing.T, a *auth.Auth) *Server { + t.Helper() + inbound, err := plugins.DefaultInboundPipeline(a) + if err != nil { + t.Fatalf("building inbound pipeline: %v", err) + } + outbound, err := plugins.DefaultOutboundPipeline(a) + if err != nil { + t.Fatalf("building outbound pipeline: %v", err) + } + return &Server{InboundPipeline: inbound, OutboundPipeline: outbound} +} + func makeHeaders(kvs ...string) *corev3.HeaderMap { hm := &corev3.HeaderMap{} for i := 0; i < len(kvs); i += 2 { @@ -91,7 +105,7 @@ func TestExtProc_Inbound_ValidJWT(t *testing.T) { Verifier: &mockVerifier{claims: &validation.Claims{Subject: "user-1"}}, Identity: auth.IdentityConfig{Audience: "my-agent"}, }) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) stream := &mockStream{ ctx: context.Background(), @@ -135,7 +149,7 @@ func TestExtProc_Inbound_InvalidJWT(t *testing.T) { Verifier: &mockVerifier{err: fmt.Errorf("token expired")}, Identity: auth.IdentityConfig{Audience: "my-agent"}, }) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) stream := &mockStream{ ctx: context.Background(), @@ -168,7 +182,7 @@ func TestExtProc_Inbound_BypassPath(t *testing.T) { Verifier: &mockVerifier{err: fmt.Errorf("should not be called")}, Bypass: matcher, }) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) stream := &mockStream{ ctx: context.Background(), @@ -213,7 +227,7 @@ func TestExtProc_Outbound_Exchange(t *testing.T) { Exchanger: exchanger, Cache: cache.New(), }) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) stream := &mockStream{ ctx: context.Background(), @@ -246,7 +260,7 @@ func TestExtProc_Outbound_Exchange(t *testing.T) { func TestExtProc_Outbound_Passthrough(t *testing.T) { router, _ := routing.NewRouter("passthrough", []routing.Route{}) a := auth.New(auth.Config{Router: router}) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) stream := &mockStream{ ctx: context.Background(), @@ -279,7 +293,7 @@ func TestExtProc_Outbound_Deny(t *testing.T) { Router: router, NoTokenPolicy: auth.NoTokenPolicyDeny, }) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) stream := &mockStream{ ctx: context.Background(), @@ -309,7 +323,7 @@ func TestExtProc_Outbound_Deny(t *testing.T) { func TestExtProc_ResponseHeaders(t *testing.T) { a := auth.New(auth.Config{}) - srv := &Server{Auth: a} + srv := serverFromAuth(t, a) stream := &mockStream{ ctx: context.Background(), diff --git a/authbridge/cmd/authbridge/listener/forwardproxy/server.go b/authbridge/cmd/authbridge/listener/forwardproxy/server.go index 5d59f83a7..83f94811b 100644 --- a/authbridge/cmd/authbridge/listener/forwardproxy/server.go +++ b/authbridge/cmd/authbridge/listener/forwardproxy/server.go @@ -8,21 +8,22 @@ import ( "io" "log/slog" "net/http" + "strings" "time" - "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" ) // Server is an HTTP forward proxy that performs token exchange on outbound requests. type Server struct { - Auth *auth.Auth - Client *http.Client + OutboundPipeline *pipeline.Pipeline + Client *http.Client } // NewServer creates a forward proxy server with a default HTTP client. -func NewServer(a *auth.Auth) *Server { +func NewServer(outbound *pipeline.Pipeline) *Server { return &Server{ - Auth: a, + OutboundPipeline: outbound, Client: &http.Client{ Timeout: 30 * time.Second, CheckRedirect: func(_ *http.Request, _ []*http.Request) error { @@ -38,23 +39,32 @@ func (s *Server) Handler() http.Handler { } func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { - // Reject CONNECT (HTTPS tunneling) — only handle plain HTTP if r.Method == http.MethodConnect { http.Error(w, `{"error":"HTTPS CONNECT not supported — only HTTP proxy"}`, http.StatusMethodNotAllowed) return } - result := s.Auth.HandleOutbound(r.Context(), r.Header.Get("Authorization"), r.Host) + pctx := &pipeline.Context{ + Direction: pipeline.Outbound, + Host: r.Host, + Path: r.URL.Path, + Headers: r.Header.Clone(), + } + + originalAuth := pctx.Headers.Get("Authorization") + action := s.OutboundPipeline.Run(r.Context(), pctx) - switch result.Action { - case auth.ActionDeny: + if action.Type == pipeline.Reject { w.Header().Set("Content-Type", "application/json") - w.WriteHeader(result.DenyStatus) - body, _ := json.Marshal(map[string]string{"error": result.DenyReason}) + w.WriteHeader(action.Status) + body, _ := json.Marshal(map[string]string{"error": action.Reason}) w.Write(body) return - case auth.ActionReplaceToken: - r.Header.Set("Authorization", "Bearer "+result.Token) + } + + newAuth := pctx.Headers.Get("Authorization") + if newAuth != originalAuth { + r.Header.Set("Authorization", "Bearer "+extractBearer(newAuth)) } // Remove hop-by-hop headers @@ -88,3 +98,10 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { slog.Debug("response copy error", "host", r.Host, "error", err) } } + +func extractBearer(authHeader string) string { + if len(authHeader) > 7 && strings.EqualFold(authHeader[:7], "bearer ") { + return authHeader[7:] + } + return "" +} diff --git a/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go b/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go index 704c73d88..83ca497ea 100644 --- a/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go +++ b/authbridge/cmd/authbridge/listener/forwardproxy/server_test.go @@ -11,6 +11,8 @@ import ( "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" "github.com/kagenti/kagenti-extensions/authbridge/authlib/cache" "github.com/kagenti/kagenti-extensions/authbridge/authlib/exchange" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins" "github.com/kagenti/kagenti-extensions/authbridge/authlib/routing" "github.com/kagenti/kagenti-extensions/authbridge/authlib/validation" ) @@ -24,6 +26,15 @@ func (m *mockVerifier) Verify(_ context.Context, _ string, _ string) (*validatio return m.claims, m.err } +func outboundPipelineFromAuth(t *testing.T, a *auth.Auth) *pipeline.Pipeline { + t.Helper() + p, err := plugins.DefaultOutboundPipeline(a) + if err != nil { + t.Fatalf("building outbound pipeline: %v", err) + } + return p +} + func TestForwardProxy_Exchange(t *testing.T) { // Token exchange server exchangeSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { @@ -57,7 +68,7 @@ func TestForwardProxy_Exchange(t *testing.T) { Cache: cache.New(), }) - srv := &Server{Auth: a, Client: http.DefaultClient} + srv := &Server{OutboundPipeline: outboundPipelineFromAuth(t, a), Client: http.DefaultClient} proxy := httptest.NewServer(srv.Handler()) defer proxy.Close() @@ -83,7 +94,7 @@ func TestForwardProxy_Exchange(t *testing.T) { func TestForwardProxy_CONNECT_Rejected(t *testing.T) { a := auth.New(auth.Config{}) - srv := NewServer(a) + srv := NewServer(outboundPipelineFromAuth(t, a)) proxy := httptest.NewServer(srv.Handler()) defer proxy.Close() @@ -104,7 +115,7 @@ func TestForwardProxy_Deny(t *testing.T) { NoTokenPolicy: auth.NoTokenPolicyDeny, }) - srv := NewServer(a) + srv := NewServer(outboundPipelineFromAuth(t, a)) proxy := httptest.NewServer(srv.Handler()) defer proxy.Close() diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/server.go b/authbridge/cmd/authbridge/listener/reverseproxy/server.go index dbe782b71..aefccded9 100644 --- a/authbridge/cmd/authbridge/listener/reverseproxy/server.go +++ b/authbridge/cmd/authbridge/listener/reverseproxy/server.go @@ -1,5 +1,5 @@ // Package reverseproxy implements an HTTP reverse proxy listener. -// Inbound requests are validated via auth.HandleInbound before being +// Inbound requests are validated via the inbound pipeline before being // forwarded to a fixed backend. package reverseproxy @@ -9,26 +9,26 @@ import ( "net/http/httputil" "net/url" - "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" ) // Server is an HTTP reverse proxy with inbound JWT validation. type Server struct { - Auth *auth.Auth - proxy *httputil.ReverseProxy - backend string + InboundPipeline *pipeline.Pipeline + proxy *httputil.ReverseProxy + backend string } // NewServer creates a reverse proxy that forwards to the given backend URL. -func NewServer(a *auth.Auth, backendURL string) (*Server, error) { +func NewServer(inbound *pipeline.Pipeline, backendURL string) (*Server, error) { target, err := url.Parse(backendURL) if err != nil { return nil, err } return &Server{ - Auth: a, - proxy: httputil.NewSingleHostReverseProxy(target), - backend: backendURL, + InboundPipeline: inbound, + proxy: httputil.NewSingleHostReverseProxy(target), + backend: backendURL, }, nil } @@ -38,12 +38,18 @@ func (s *Server) Handler() http.Handler { } func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { - result := s.Auth.HandleInbound(r.Context(), r.Header.Get("Authorization"), r.URL.Path, "") + pctx := &pipeline.Context{ + Direction: pipeline.Inbound, + Host: r.Host, + Path: r.URL.Path, + Headers: r.Header.Clone(), + } - if result.Action == auth.ActionDeny { + action := s.InboundPipeline.Run(r.Context(), pctx) + if action.Type == pipeline.Reject { w.Header().Set("Content-Type", "application/json") - w.WriteHeader(result.DenyStatus) - body, _ := json.Marshal(map[string]string{"error": result.DenyReason}) + w.WriteHeader(action.Status) + body, _ := json.Marshal(map[string]string{"error": action.Reason}) w.Write(body) return } diff --git a/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go b/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go index b7f71a43e..8672dca92 100644 --- a/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go +++ b/authbridge/cmd/authbridge/listener/reverseproxy/server_test.go @@ -9,6 +9,8 @@ import ( "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" "github.com/kagenti/kagenti-extensions/authbridge/authlib/bypass" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins" "github.com/kagenti/kagenti-extensions/authbridge/authlib/validation" ) @@ -21,6 +23,15 @@ func (m *mockVerifier) Verify(_ context.Context, _ string, _ string) (*validatio return m.claims, m.err } +func inboundPipelineFromAuth(t *testing.T, a *auth.Auth) *pipeline.Pipeline { + t.Helper() + p, err := plugins.DefaultInboundPipeline(a) + if err != nil { + t.Fatalf("building inbound pipeline: %v", err) + } + return p +} + func TestReverseProxy_AllowedRequest(t *testing.T) { backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) @@ -32,7 +43,7 @@ func TestReverseProxy_AllowedRequest(t *testing.T) { Verifier: &mockVerifier{claims: &validation.Claims{Subject: "user"}}, Identity: auth.IdentityConfig{Audience: "my-app"}, }) - srv, err := NewServer(a, backend.URL) + srv, err := NewServer(inboundPipelineFromAuth(t, a), backend.URL) if err != nil { t.Fatal(err) } @@ -56,7 +67,7 @@ func TestReverseProxy_DeniedRequest(t *testing.T) { Verifier: &mockVerifier{err: fmt.Errorf("invalid token")}, Identity: auth.IdentityConfig{Audience: "my-app"}, }) - srv, _ := NewServer(a, "http://localhost:9999") + srv, _ := NewServer(inboundPipelineFromAuth(t, a), "http://localhost:9999") proxy := httptest.NewServer(srv.Handler()) defer proxy.Close() @@ -81,7 +92,7 @@ func TestReverseProxy_BypassPath(t *testing.T) { Verifier: &mockVerifier{err: fmt.Errorf("should not be called")}, Bypass: matcher, }) - srv, _ := NewServer(a, backend.URL) + srv, _ := NewServer(inboundPipelineFromAuth(t, a), backend.URL) proxy := httptest.NewServer(srv.Handler()) defer proxy.Close() diff --git a/authbridge/cmd/authbridge/main.go b/authbridge/cmd/authbridge/main.go index 0bafc26f9..f787b3f11 100644 --- a/authbridge/cmd/authbridge/main.go +++ b/authbridge/cmd/authbridge/main.go @@ -26,6 +26,8 @@ import ( "github.com/kagenti/kagenti-extensions/authbridge/authlib/auth" "github.com/kagenti/kagenti-extensions/authbridge/authlib/config" "github.com/kagenti/kagenti-extensions/authbridge/authlib/observe" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" + "github.com/kagenti/kagenti-extensions/authbridge/authlib/plugins" "github.com/kagenti/kagenti-extensions/authbridge/cmd/authbridge/listener/extauthz" "github.com/kagenti/kagenti-extensions/authbridge/cmd/authbridge/listener/extproc" "github.com/kagenti/kagenti-extensions/authbridge/cmd/authbridge/listener/forwardproxy" @@ -103,6 +105,16 @@ func main() { } handler := auth.New(*resolved) + // Build pipelines from config (falls back to defaults if pipeline section is omitted) + inboundPipeline, err := buildInboundPipeline(cfg, handler) + if err != nil { + log.Fatalf("building inbound pipeline: %v", err) + } + outboundPipeline, err := buildOutboundPipeline(cfg, handler) + if err != nil { + log.Fatalf("building outbound pipeline: %v", err) + } + // Track servers for graceful shutdown var grpcServers []*grpc.Server var httpServers []*http.Server @@ -110,19 +122,19 @@ func main() { // Start listeners FIRST — before credential resolution switch cfg.Mode { case config.ModeEnvoySidecar: - grpcServers = append(grpcServers, startGRPCExtProc(handler, cfg.Listener.ExtProcAddr)) + grpcServers = append(grpcServers, startGRPCExtProc(inboundPipeline, outboundPipeline, cfg.Listener.ExtProcAddr)) case config.ModeWaypoint: - grpcServers = append(grpcServers, startGRPCExtAuthz(handler, cfg.Listener.ExtAuthzAddr)) - httpServers = append(httpServers, startHTTPServer("forward-proxy", forwardproxy.NewServer(handler).Handler(), cfg.Listener.ForwardProxyAddr)) + grpcServers = append(grpcServers, startGRPCExtAuthz(inboundPipeline, outboundPipeline, cfg.Listener.ExtAuthzAddr)) + httpServers = append(httpServers, startHTTPServer("forward-proxy", forwardproxy.NewServer(outboundPipeline).Handler(), cfg.Listener.ForwardProxyAddr)) case config.ModeProxySidecar: - rpSrv, err := reverseproxy.NewServer(handler, cfg.Listener.ReverseProxyBackend) + rpSrv, err := reverseproxy.NewServer(inboundPipeline, cfg.Listener.ReverseProxyBackend) if err != nil { log.Fatalf("creating reverse proxy: %v", err) } httpServers = append(httpServers, startHTTPServer("reverse-proxy", rpSrv.Handler(), cfg.Listener.ReverseProxyAddr)) - httpServers = append(httpServers, startHTTPServer("forward-proxy", forwardproxy.NewServer(handler).Handler(), cfg.Listener.ForwardProxyAddr)) + httpServers = append(httpServers, startHTTPServer("forward-proxy", forwardproxy.NewServer(outboundPipeline).Handler(), cfg.Listener.ForwardProxyAddr)) default: log.Fatalf("unhandled mode %q", cfg.Mode) @@ -176,9 +188,29 @@ func main() { statSrv.Shutdown(shutdownCtx) } -func startGRPCExtProc(handler *auth.Auth, addr string) *grpc.Server { +func buildInboundPipeline(cfg *config.Config, handler *auth.Auth) (*pipeline.Pipeline, error) { + if len(cfg.Pipeline.Inbound.Plugins) > 0 { + return plugins.Build(cfg.Pipeline.Inbound.Plugins, handler) + } + if cfg.Mode == config.ModeWaypoint { + return plugins.WaypointInboundPipeline(handler) + } + return plugins.DefaultInboundPipeline(handler) +} + +func buildOutboundPipeline(cfg *config.Config, handler *auth.Auth) (*pipeline.Pipeline, error) { + if len(cfg.Pipeline.Outbound.Plugins) > 0 { + return plugins.Build(cfg.Pipeline.Outbound.Plugins, handler) + } + return plugins.DefaultOutboundPipeline(handler) +} + +func startGRPCExtProc(inbound, outbound *pipeline.Pipeline, addr string) *grpc.Server { srv := grpc.NewServer() - extprocv3.RegisterExternalProcessorServer(srv, &extproc.Server{Auth: handler}) + extprocv3.RegisterExternalProcessorServer(srv, &extproc.Server{ + InboundPipeline: inbound, + OutboundPipeline: outbound, + }) registerHealth(srv) reflection.Register(srv) @@ -195,9 +227,12 @@ func startGRPCExtProc(handler *auth.Auth, addr string) *grpc.Server { return srv } -func startGRPCExtAuthz(handler *auth.Auth, addr string) *grpc.Server { +func startGRPCExtAuthz(inbound, outbound *pipeline.Pipeline, addr string) *grpc.Server { srv := grpc.NewServer() - authv3.RegisterAuthorizationServer(srv, &extauthz.Server{Auth: handler}) + authv3.RegisterAuthorizationServer(srv, &extauthz.Server{ + InboundPipeline: inbound, + OutboundPipeline: outbound, + }) registerHealth(srv) reflection.Register(srv)