homelab-frontend/internal/serviceadapter/router.go

package serviceadapter

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"net/http/httputil"
	"net/url"
	"strings"
	"time"

	"golang.org/x/net/http2"
	"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/identity"
	"forgejo.riotpiao.com/rock/homelab-frontend/internal/problem"
)

// Dispatcher routes X-Service requests to upstreams.
type Dispatcher struct {
	registry     *Registry
	jwtValidator *auth.Validator
}

// 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,
		jwtValidator: jwtValidator,
	}
}

// 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.
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
	}

	adapter := d.registry.Get(serviceName)
	if adapter == nil {
		d.writeError(w, problem.NotFound(fmt.Sprintf("service '%s' not found", serviceName)))
		return
	}

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

	resource := findResource(adapter, resourceName)
	if resource == nil {
		d.writeError(w, problem.NotFound(
			fmt.Sprintf("resource '%s' not found in service '%s'", resourceName, serviceName)))
		return
	}

	method := findMethod(resource, r.Method)
	if method == nil {
		d.writeError(w, problem.NotFound(
			fmt.Sprintf("method %s not defined for resource '%s'", r.Method, resourceName)))
		return
	}

	// 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
		}
	}

	upstreamURL := adapter.Spec.Upstream.URL
	if strings.HasPrefix(upstreamURL, "grpc://") {
		d.dispatchGRPC(w, r, upstreamURL, method, adapter)
	} else {
		d.dispatchHTTP(w, r, upstreamURL, method, adapter)
	}
}

// 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)))
		return
	}

	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
		
		// Preserve Authorization header for S3 SigV4 and other auth schemes
		// Note: httputil.ReverseProxy preserves most headers automatically,
		// but we need to ensure Authorization isn't lost when overriding Director
	}

	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,
	}

	proxy.ServeHTTP(w, r)
}

func (d *Dispatcher) dispatchGRPC(w http.ResponseWriter, r *http.Request, upstreamURL string, method *Method, adapter *ServiceAdapter) {
	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
	}

	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
	}

	timeout := adapter.Spec.Upstream.TimeoutSeconds
	if timeout <= 0 {
		timeout = 30
	}

	ctx, cancel := context.WithTimeout(r.Context(), time.Duration(timeout)*time.Second)
	defer cancel()

	conn, err := grpc.DialContext(ctx, host,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		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)))
		return
	}
	defer conn.Close()

	// Create HTTP/2 reverse proxy for gRPC
	// gRPC uses HTTP/2 protocol, so we need an HTTP/2-capable transport
	upstreamURLObj := &url.URL{
		Scheme: "http",
		Host:   host,
	}

	proxy := httputil.NewSingleHostReverseProxy(upstreamURLObj)
	proxy.Director = func(req *http.Request) {
		req.URL.Scheme = "http"
		req.URL.Host = host
		req.URL.Path = method.UpstreamPath
		req.RequestURI = ""
		req.Host = host
	}

	// Create HTTP/2 client transport for gRPC calls
	// gRPC requires HTTP/2 for proper message framing
	h2transport := &http2.Transport{
		AllowHTTP: true,
	}

	// Set the transport on the proxy
	proxy.Transport = h2transport

	// Serve the request through the proxy
	proxy.ServeHTTP(w, r)
}

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