feat(serviceadapter): enforce JWT auth on X-Service dispatch

Replace hardcoded SQS-only validator (pointed at non-existent sqs provider JWKS) with shared multi-issuer validator from proxy. Auth-required adapters now validate JWT and check capability (<service>:read for GET, <service>:write for POST). Identity headers injected on success. Auth-optional adapters unchanged. 10 new tests, 69% package coverage. Closes homelab#11 Co-authored-by: poimen <[email protected]>
2026-09-08 16:14:15 -07:00
parent 6f7b850193
commit 62c5cb0b86
3 changed files with 370 additions and 104 deletions
@@ -14,42 +14,31 @@ import (
 	"google.golang.org/grpc/credentials/insecure"

 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/auth"
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/identity"
 	"forgejo.riotpiao.com/rock/homelab-frontend/internal/problem"
 )

 // Dispatcher routes X-Service requests to upstreams.
-// Auth per service:
-//   SQS: Gateway validates JWT (kmsvc code unverified)
-//   MinIO, Temporal: Native JWT support (dumb pipe pass-through)
-//   Memory, IAM: Services validate JWTs themselves
 type Dispatcher struct {
-	registry   *Registry
-	sqsJWTAuth *auth.Validator
+	registry     *Registry
+	jwtValidator *auth.Validator
 }

-// NewDispatcher creates a new service adapter dispatcher.
-func NewDispatcher(registry *Registry) *Dispatcher {
-	// Create JWT validator for SQS
-	// Issuer and JWKS URL should match Authentik application config
-	sqsValidator := auth.NewValidator(
-		"https://authentik.riotpiao.com/application/o/sqs/",
-		"sqs",
-		"https://authentik.riotpiao.com/application/o/sqs/jwks/",
-	)
-
+// NewDispatcher creates a dispatcher with a shared multi-issuer JWT validator.
+// Pass nil to disable auth enforcement (all requests pass through).
+func NewDispatcher(registry *Registry, jwtValidator *auth.Validator) *Dispatcher {
 	return &Dispatcher{
-		registry:   registry,
-		sqsJWTAuth: sqsValidator,
+		registry:     registry,
+		jwtValidator: jwtValidator,
 	}
 }

-// Matches returns true if the request should be dispatched based on X-Service header.
+// Matches returns true if the request has an X-Service header.
 func (d *Dispatcher) Matches(r *http.Request) bool {
 	return r.Header.Get("X-Service") != ""
 }

 // Dispatch routes a request to the appropriate adapter.
-// Returns a problem document if the adapter or resource is not found.
 func (d *Dispatcher) Dispatch(w http.ResponseWriter, r *http.Request) {
 	serviceName := r.Header.Get("X-Service")
 	if serviceName == "" {
@@ -57,102 +46,117 @@ func (d *Dispatcher) Dispatch(w http.ResponseWriter, r *http.Request) {
 		return
 	}

-	// Look up service adapter
 	adapter := d.registry.Get(serviceName)
 	if adapter == nil {
-		p := problem.NotFound(fmt.Sprintf("service '%s' not found", serviceName))
-		_ = p.Write(w)
+		d.writeError(w, problem.NotFound(fmt.Sprintf("service '%s' not found", serviceName)))
 		return
 	}

-	// Get resource and method from request
 	resourceName := r.Header.Get("X-Resource")
 	if resourceName == "" {
 		d.writeError(w, problem.BadRequest("X-Resource header required"))
 		return
 	}

-	// Find resource
-	var resource *Resource
-	for i := range adapter.Spec.Resources {
-		if adapter.Spec.Resources[i].Name == resourceName {
-			resource = &adapter.Spec.Resources[i]
-			break
-		}
-	}
-
+	resource := findResource(adapter, resourceName)
 	if resource == nil {
-		p := problem.NotFound(fmt.Sprintf("resource '%s' not found in service '%s'", resourceName, serviceName))
-		_ = p.Write(w)
+		d.writeError(w, problem.NotFound(
+			fmt.Sprintf("resource '%s' not found in service '%s'", resourceName, serviceName)))
 		return
 	}

-	// Find method matching HTTP verb
-	var method *Method
-	for i := range resource.Methods {
-		if resource.Methods[i].Verb == r.Method {
-			method = &resource.Methods[i]
-			break
-		}
-	}
-
+	method := findMethod(resource, r.Method)
 	if method == nil {
-		p := problem.NotFound(fmt.Sprintf("method %s not defined for resource '%s'", r.Method, resourceName))
-		_ = p.Write(w)
+		d.writeError(w, problem.NotFound(
+			fmt.Sprintf("method %s not defined for resource '%s'", r.Method, resourceName)))
 		return
 	}

-	// Gateway-level JWT validation for SQS (code unverified in kmsvc)
-	// MinIO, Temporal, Memory, IAM have native JWT support - pass through
-	if adapter.Spec.Auth.Required && serviceName == "sqs" {
-		authHeader := r.Header.Get("Authorization")
-		if authHeader == "" {
-			p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
-				"Forbidden", "SQS requires Authorization header")
-			_ = p.Write(w)
-			return
-		}
-
-		// Validate JWT signature against Authentik JWKS
-		claims, err := d.sqsJWTAuth.ValidateBearerToken(authHeader)
-		if err != nil {
-			p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
-				"Forbidden", fmt.Sprintf("JWT validation failed: %v", err))
-			_ = p.Write(w)
-			return
-		}
-
-		// Check required permissions (sqs:read or sqs:write or *)
-		hasPermission := d.sqsJWTAuth.CheckPermissions(claims, "sqs:read", "sqs:write", "*")
-		if !hasPermission {
-			p := problem.NewProblem(http.StatusForbidden, "about:blank#forbidden",
-				"Forbidden", "Insufficient permissions for SQS")
-			_ = p.Write(w)
+	// JWT auth enforcement for adapters that require it
+	if adapter.Spec.Auth.Required && d.jwtValidator != nil {
+		if !d.authenticate(w, r, serviceName, method.Verb) {
 			return
 		}
 	}

-	// Detect protocol from upstream URL scheme
 	upstreamURL := adapter.Spec.Upstream.URL
 	if strings.HasPrefix(upstreamURL, "grpc://") {
-		// gRPC upstream (Temporal, etc.)
 		d.dispatchGRPC(w, r, upstreamURL, method, adapter)
 	} else {
-		// HTTP upstream (MinIO, Authentik, etc.)
 		d.dispatchHTTP(w, r, upstreamURL, method, adapter)
 	}
 }

-// dispatchHTTP forwards HTTP requests to upstream, passing Authorization header through.
+// authenticate validates the JWT and checks service-level capability.
+// Returns false (and writes error response) if auth fails.
+func (d *Dispatcher) authenticate(w http.ResponseWriter, r *http.Request, serviceName, verb string) bool {
+	authHeader := r.Header.Get("Authorization")
+	if authHeader == "" {
+		d.writeError(w, problem.NewProblem(http.StatusUnauthorized,
+			"about:blank#unauthorized", "Unauthorized",
+			fmt.Sprintf("service '%s' requires Authorization header", serviceName)))
+		return false
+	}
+
+	claims, err := d.jwtValidator.ValidateBearerToken(authHeader)
+	if err != nil {
+		d.writeError(w, problem.NewProblem(http.StatusForbidden,
+			"about:blank#forbidden", "Forbidden",
+			fmt.Sprintf("JWT validation failed: %v", err)))
+		return false
+	}
+
+	// Check capability: <service>:read for GET/HEAD, <service>:write for mutating verbs
+	required := capabilityForVerb(serviceName, verb)
+	if !d.jwtValidator.CheckPermissions(claims, required, "*") {
+		d.writeError(w, problem.NewProblem(http.StatusForbidden,
+			"about:blank#insufficient-permissions", "Insufficient Permissions",
+			fmt.Sprintf("required capability: %s", required)))
+		return false
+	}
+
+	// Inject identity headers for downstream
+	identity.Inject(r, claims)
+	return true
+}
+
+// capabilityForVerb maps HTTP verbs to <service>:read or <service>:write.
+func capabilityForVerb(serviceName, verb string) string {
+	switch verb {
+	case "GET", "HEAD", "OPTIONS":
+		return serviceName + ":read"
+	default:
+		return serviceName + ":write"
+	}
+}
+
+func findResource(adapter *ServiceAdapter, name string) *Resource {
+	for i := range adapter.Spec.Resources {
+		if adapter.Spec.Resources[i].Name == name {
+			return &adapter.Spec.Resources[i]
+		}
+	}
+	return nil
+}
+
+func findMethod(resource *Resource, verb string) *Method {
+	for i := range resource.Methods {
+		if resource.Methods[i].Verb == verb {
+			return &resource.Methods[i]
+		}
+	}
+	return nil
+}
+
 func (d *Dispatcher) dispatchHTTP(w http.ResponseWriter, r *http.Request, upstreamURL string, method *Method, adapter *ServiceAdapter) {
 	parsedURL, err := url.Parse(upstreamURL)
 	if err != nil {
-		d.writeError(w, problem.NewProblem(http.StatusInternalServerError, "about:blank#server-error",
-			"Internal Server Error", fmt.Sprintf("invalid upstream URL: %v", err)))
+		d.writeError(w, problem.NewProblem(http.StatusInternalServerError,
+			"about:blank#server-error", "Internal Server Error",
+			fmt.Sprintf("invalid upstream URL: %v", err)))
 		return
 	}

-	// Create reverse proxy
 	proxy := httputil.NewSingleHostReverseProxy(parsedURL)
 	proxy.Director = func(req *http.Request) {
 		req.URL.Scheme = parsedURL.Scheme
@@ -160,42 +164,36 @@ func (d *Dispatcher) dispatchHTTP(w http.ResponseWriter, r *http.Request, upstre
 		req.URL.Path = method.UpstreamPath
 		req.RequestURI = ""
 		req.Host = parsedURL.Host
-		// Authorization header passes through unchanged
 	}

-	// Set timeout
 	timeout := adapter.Spec.Upstream.TimeoutSeconds
 	if timeout <= 0 {
 		timeout = 30
 	}
 	proxy.Transport = &http.Transport{
-		DialContext:         (&net.Dialer{Timeout: time.Duration(timeout) * time.Second}).DialContext,
+		DialContext:          (&net.Dialer{Timeout: time.Duration(timeout) * time.Second}).DialContext,
 		TLSHandshakeTimeout: time.Duration(timeout) * time.Second,
 	}

-	// Forward the request
 	proxy.ServeHTTP(w, r)
 }

-// dispatchGRPC forwards gRPC requests to upstream.
-// gRPC URL format: grpc://host:port
 func (d *Dispatcher) dispatchGRPC(w http.ResponseWriter, r *http.Request, upstreamURL string, method *Method, adapter *ServiceAdapter) {
-	// Extract host:port from grpc://host:port
 	host := strings.TrimPrefix(upstreamURL, "grpc://")
 	if host == upstreamURL {
-		d.writeError(w, problem.NewProblem(http.StatusInternalServerError, "about:blank#server-error",
-			"Internal Server Error", "invalid gRPC URL format"))
+		d.writeError(w, problem.NewProblem(http.StatusInternalServerError,
+			"about:blank#server-error", "Internal Server Error",
+			"invalid gRPC URL format"))
 		return
 	}

-	// Validate that this is a gRPC request
 	if !strings.HasPrefix(r.Header.Get("Content-Type"), "application/grpc") {
-		d.writeError(w, problem.NewProblem(http.StatusBadRequest, "about:blank#bad-request",
-			"Bad Request", "gRPC service requires application/grpc content-type"))
+		d.writeError(w, problem.NewProblem(http.StatusBadRequest,
+			"about:blank#bad-request", "Bad Request",
+			"gRPC service requires application/grpc content-type"))
 		return
 	}

-	// Set timeout
 	timeout := adapter.Spec.Upstream.TimeoutSeconds
 	if timeout <= 0 {
 		timeout = 30
@@ -204,25 +202,21 @@ func (d *Dispatcher) dispatchGRPC(w http.ResponseWriter, r *http.Request, upstre
 	ctx, cancel := context.WithTimeout(r.Context(), time.Duration(timeout)*time.Second)
 	defer cancel()

-	// Dial gRPC upstream
 	conn, err := grpc.DialContext(ctx, host,
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithDefaultCallOptions(
-			grpc.MaxCallRecvMsgSize(100 * 1024 * 1024), // 100MB
-		),
+		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(100*1024*1024)),
 	)
 	if err != nil {
-		d.writeError(w, problem.NewProblem(http.StatusBadGateway, "about:blank#bad-gateway",
-			"Bad Gateway", fmt.Sprintf("failed to dial gRPC upstream: %v", err)))
+		d.writeError(w, problem.NewProblem(http.StatusBadGateway,
+			"about:blank#bad-gateway", "Bad Gateway",
+			fmt.Sprintf("failed to dial gRPC upstream: %v", err)))
 		return
 	}
 	defer conn.Close()

-	// Forward gRPC request
-	// Note: Full gRPC forwarding requires grpcproxy or custom middleware.
-	// For now, return unimplemented (Temporal support coming in Phase 9)
-	d.writeError(w, problem.NewProblem(http.StatusNotImplemented, "about:blank#not-implemented",
-		"Not Implemented", "gRPC forwarding not yet implemented - use in-cluster gRPC clients directly"))
+	d.writeError(w, problem.NewProblem(http.StatusNotImplemented,
+		"about:blank#not-implemented", "Not Implemented",
+		"gRPC forwarding not yet implemented"))
 }

 func (d *Dispatcher) writeError(w http.ResponseWriter, p *problem.Problem) {