package serviceadapter import ( "context" "fmt" "net" "net/http" "net/http/httputil" "net/url" "strings" "time" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "forgejo.riotpiao.com/rock/homelab-frontend/internal/auth" "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 } // 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/", ) return &Dispatcher{ registry: registry, sqsJWTAuth: sqsValidator, } } // Matches returns true if the request should be dispatched based on 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 == "" { d.writeError(w, problem.BadRequest("X-Service header required")) 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) 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 } } if resource == nil { p := problem.NotFound(fmt.Sprintf("resource '%s' not found in service '%s'", resourceName, serviceName)) _ = p.Write(w) 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 } } if method == nil { p := problem.NotFound(fmt.Sprintf("method %s not defined for resource '%s'", r.Method, resourceName)) _ = p.Write(w) 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) 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. 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))) return } // Create reverse proxy proxy := httputil.NewSingleHostReverseProxy(parsedURL) proxy.Director = func(req *http.Request) { req.URL.Scheme = parsedURL.Scheme req.URL.Host = parsedURL.Host 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, 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")) 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")) return } // Set timeout timeout := adapter.Spec.Upstream.TimeoutSeconds if timeout <= 0 { timeout = 30 } 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 ), ) 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))) 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")) } func (d *Dispatcher) writeError(w http.ResponseWriter, p *problem.Problem) { _ = p.Write(w) }