From d5909172ae93df5075ab855ba46f6ffe0fba9f31 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1=C5=A1=20Dvo=C5=99=C3=A1k?= <150935816+Dvorinka@users.noreply.github.com> Date: Sat, 10 Oct 2026 00:29:57 +0200 Subject: [PATCH] feat(gateway): /g/:slug proxy data plane with key auth and rate limits (#66) The admin surface under /admin/gateway managed api_services and api_keys, but no public route ever consumed them. Mounts /g/:slug and /g/:slug/* on the root router (outside /api/v1 so route_prefix stays clean) and runs the full pipeline: slug lookup -> X-API-Key/Bearer extraction -> prefix-indexed bcrypt key auth -> allowed_service_ids scope -> per-minute buckets (key and service caps) -> monthly usage_counters quotas -> gateway.ProxyRequest with per-service timeout and upstream auth headers -> usage/metrics/incident recording. Live-verified: proxied GET/POST, 401/404/429 boundaries, counter upserts, UPSTREAM_5XX incidents. Unit tests cover URL building, the rpm bucket, key extraction, and the service allowlist. Generated with [Devin](https://devin.ai) Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- ROADMAP2.md | 5 +- app/backend/internal/api/gateway_proxy.go | 311 ++++++++++++++++++ .../internal/api/gateway_proxy_test.go | 81 +++++ app/backend/internal/api/routes.go | 6 + app/backend/internal/gateway/proxy_test.go | 64 ++++ 5 files changed, 465 insertions(+), 2 deletions(-) create mode 100644 app/backend/internal/api/gateway_proxy.go create mode 100644 app/backend/internal/api/gateway_proxy_test.go create mode 100644 app/backend/internal/gateway/proxy_test.go diff --git a/ROADMAP2.md b/ROADMAP2.md index 778ab88..018214c 100644 --- a/ROADMAP2.md +++ b/ROADMAP2.md @@ -517,8 +517,9 @@ and wired refs, one click or one CLI call. ## 10. Carried-over debt (ROADMAP.md items still open) -- [ ] **Gateway traffic path** — mount `/g/:slug/*` → key → rate-limit → - `gateway/proxy.go` → metrics. Management API exists; proxy unwired. +- [x] **Gateway traffic path** — `/g/:slug/*` mounted → api_key auth → + rpm + monthly quota → `gateway/proxy.go` → usage_counters / + metrics_timeseries / incident_events. - [ ] **Preview environments runtime** — real build+deploy+subdomain+TTL (bookkeeping only now). Pairs with clone + multi-domain. - [ ] **Environments** — env switcher, per-env vars/domains, promote. diff --git a/app/backend/internal/api/gateway_proxy.go b/app/backend/internal/api/gateway_proxy.go new file mode 100644 index 0000000..2c4f030 --- /dev/null +++ b/app/backend/internal/api/gateway_proxy.go @@ -0,0 +1,311 @@ +package api + +// Data plane for the public API gateway: /g/:slug/* proxies to the +// upstream_url of the matching api_services row. Callers authenticate with +// api_keys (X-API-Key or Authorization: Bearer), pass the rpm + monthly +// quota gates, and every request lands in usage_counters / +// metrics_timeseries / incident_events for the /admin/gateway analytics +// surfaces. + +import ( + "context" + "database/sql" + "encoding/json" + "net/http" + "strings" + "sync" + "time" + + "containr/internal/database" + "containr/internal/gateway" + + "github.com/gin-gonic/gin" + "golang.org/x/crypto/bcrypt" +) + +// jarvis: ceiling — a single shared transport; per-request timeouts come +// from api_services.request_timeout_ms on the request context. +var gatewayHTTPClient = &http.Client{ + Transport: &http.Transport{ + MaxIdleConns: 100, + MaxIdleConnsPerHost: 10, + IdleConnTimeout: 90 * time.Second, + }, +} + +// gatewayRPM is a process-local per-minute counter map. Containr runs a +// single API process, so an in-memory bucket is sufficient — a multi-node +// control plane would need a Redis-backed limiter instead. +var gatewayRPM = struct { + sync.Mutex + buckets map[string]*rpmBucket +}{buckets: map[string]*rpmBucket{}} + +type rpmBucket struct { + window time.Time + count int +} + +// gatewayRateAllow returns false when the bucket for the current minute has +// already reached limit. A limit <= 0 disables the check. `now` is injected +// so tests can freeze the window. +func gatewayRateAllow(bucketKey string, limit int, now time.Time) bool { + if limit <= 0 { + return true + } + window := now.Truncate(time.Minute) + gatewayRPM.Lock() + defer gatewayRPM.Unlock() + b := gatewayRPM.buckets[bucketKey] + if b == nil || !b.window.Equal(window) { + b = &rpmBucket{window: window} + gatewayRPM.buckets[bucketKey] = b + // Best-effort sweep so the map cannot grow without bound. + for k, old := range gatewayRPM.buckets { + if old.window.Before(window.Add(-time.Minute)) { + delete(gatewayRPM.buckets, k) + } + } + } + b.count++ + return b.count <= limit +} + +type gatewayService struct { + ID string + UpstreamURL string + RoutePrefix string + UpstreamAuthHeader sql.NullString + UpstreamAuthValue sql.NullString + Enabled bool + RPMLimit sql.NullInt64 + MonthlyQuota sql.NullInt64 + RequestTimeoutMS int +} + +type gatewayKey struct { + ID string + Hash string + Enabled bool + RPMLimit sql.NullInt64 + MonthlyQuota sql.NullInt64 + AllowedServiceIDs string +} + +// extractGatewayKey reads the caller's API key from X-API-Key first, then +// Authorization: Bearer. +func extractGatewayKey(c *gin.Context) string { + if k := strings.TrimSpace(c.GetHeader("X-API-Key")); k != "" { + return k + } + auth := strings.TrimSpace(c.GetHeader("Authorization")) + if len(auth) > 7 && strings.EqualFold(auth[:7], "Bearer ") { + return strings.TrimSpace(auth[7:]) + } + return "" +} + +// lookupGatewayService resolves the slug to an enabled service row. +func lookupGatewayService(ctx context.Context, db *database.DB, slug string) (*gatewayService, error) { + var s gatewayService + var enabled int + err := db.QueryRowContext(ctx, ` + SELECT id, upstream_url, route_prefix, upstream_auth_header, + upstream_auth_value, enabled, rpm_limit, monthly_quota, + request_timeout_ms + FROM api_services WHERE slug = $1 + `, slug).Scan(&s.ID, &s.UpstreamURL, &s.RoutePrefix, + &s.UpstreamAuthHeader, &s.UpstreamAuthValue, &enabled, + &s.RPMLimit, &s.MonthlyQuota, &s.RequestTimeoutMS) + if err != nil { + return nil, err + } + s.Enabled = enabled != 0 + return &s, nil +} + +// authenticateGatewayKey resolves a presented key by its indexed prefix and +// bcrypt-verifies candidates. Returns nil when nothing matches. +func authenticateGatewayKey(ctx context.Context, db *database.DB, presented string) (*gatewayKey, error) { + prefix := presented + if len(prefix) > 8 { + prefix = prefix[:8] + } + rows, err := db.QueryContext(ctx, ` + SELECT id, key_hash, enabled, rpm_limit, monthly_quota, allowed_service_ids + FROM api_keys WHERE key_prefix = $1 + `, prefix) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var k gatewayKey + var enabled int + if err := rows.Scan(&k.ID, &k.Hash, &enabled, &k.RPMLimit, &k.MonthlyQuota, &k.AllowedServiceIDs); err != nil { + return nil, err + } + k.Enabled = enabled != 0 + if bcrypt.CompareHashAndPassword([]byte(k.Hash), []byte(presented)) == nil { + return &k, nil + } + } + return nil, rows.Err() +} + +// keyAllowsService applies the allowed_service_ids JSON allowlist. An empty +// list means the key may call every service. +func keyAllowsService(k *gatewayKey, serviceID string) bool { + var allowed []string + if err := json.Unmarshal([]byte(k.AllowedServiceIDs), &allowed); err != nil || len(allowed) == 0 { + return true + } + for _, id := range allowed { + if id == serviceID { + return true + } + } + return false +} + +// monthlyUsage sums a counter dimension for the current calendar month. +func monthlyUsage(ctx context.Context, db *database.DB, column, id, period string) (int, error) { + var total int + err := db.QueryRowContext(ctx, ` + SELECT COALESCE(SUM(request_count), 0) FROM usage_counters + WHERE `+column+` = $1 AND period_month = $2 + `, id, period).Scan(&total) + return total, err +} + +// recordGatewayRequest persists usage, metrics, and failure incidents for a +// proxied call. Runs in the request path but is three small writes. +func recordGatewayRequest(ctx context.Context, db *database.DB, svc *gatewayService, k *gatewayKey, status int, period string) { + _, _ = db.ExecContext(ctx, ` + INSERT INTO usage_counters (api_key_id, service_id, period_month, request_count, updated_at) + VALUES ($1, $2, $3, 1, NOW()) + ON CONFLICT (api_key_id, service_id, period_month) + DO UPDATE SET request_count = usage_counters.request_count + 1, updated_at = NOW() + `, k.ID, svc.ID, period) + _, _ = db.ExecContext(ctx, ` + UPDATE api_keys SET last_used_at = NOW() WHERE id = $1 + `, k.ID) + labels, _ := json.Marshal(map[string]interface{}{ + "service_id": svc.ID, + "key_id": k.ID, + "status": status, + }) + _, _ = db.ExecContext(ctx, ` + INSERT INTO metrics_timeseries (metric, value, labels_json, occurred_at) + VALUES ('gateway_request', 1, $1, NOW()) + `, string(labels)) +} + +func recordGatewayIncident(ctx context.Context, db *database.DB, svcID string, keyID *string, code, msg, severity string, status int) { + var kid interface{} + if keyID != nil { + kid = *keyID + } + _, _ = db.ExecContext(ctx, ` + INSERT INTO incident_events (service_id, api_key_id, code, message, severity, http_status, occurred_at) + VALUES ($1, $2, $3, $4, $5, $6, NOW()) + `, svcID, kid, code, msg, severity, status) +} + +// handleGatewayProxy is mounted at /g/:slug and /g/:slug/*path on the root +// router — outside /api/v1 so route_prefix stays clean. +func handleGatewayProxy(c *gin.Context) { + db := c.MustGet("db").(*database.DB) + ctx := c.Request.Context() + slug := c.Param("slug") + + svc, err := lookupGatewayService(ctx, db, slug) + if err == sql.ErrNoRows { + respondError(c, http.StatusNotFound, "SERVICE_NOT_FOUND", "Gateway service not found") + return + } + if err != nil { + respondError(c, http.StatusInternalServerError, "INTERNAL_ERROR", "Failed to resolve gateway service") + return + } + if !svc.Enabled { + respondError(c, http.StatusServiceUnavailable, "SERVICE_DISABLED", "Gateway service is disabled") + return + } + + presented := extractGatewayKey(c) + if presented == "" { + recordGatewayIncident(ctx, db, svc.ID, nil, "MISSING_API_KEY", "Request without API key", "low", http.StatusUnauthorized) + respondError(c, http.StatusUnauthorized, "API_KEY_REQUIRED", "An API key is required for this service") + return + } + key, err := authenticateGatewayKey(ctx, db, presented) + if err != nil { + respondError(c, http.StatusInternalServerError, "INTERNAL_ERROR", "Failed to authenticate API key") + return + } + // Disabled and unknown keys collapse to the same 401 — a disabled key + // must not leak its existence. + if key == nil || !key.Enabled { + recordGatewayIncident(ctx, db, svc.ID, nil, "INVALID_API_KEY", "Request with invalid or disabled API key", "low", http.StatusUnauthorized) + respondError(c, http.StatusUnauthorized, "INVALID_API_KEY", "Invalid API key") + return + } + if !keyAllowsService(key, svc.ID) { + recordGatewayIncident(ctx, db, svc.ID, &key.ID, "KEY_SCOPE_DENIED", "API key not allowed on this service", "medium", http.StatusForbidden) + respondError(c, http.StatusForbidden, "KEY_SCOPE_DENIED", "API key is not permitted for this service") + return + } + + now := time.Now() + if !gatewayRateAllow("key:"+key.ID, int(key.RPMLimit.Int64), now) || + !gatewayRateAllow("svc:"+svc.ID, int(svc.RPMLimit.Int64), now) { + recordGatewayIncident(ctx, db, svc.ID, &key.ID, "RATE_LIMITED", "Per-minute rate limit exceeded", "medium", http.StatusTooManyRequests) + respondError(c, http.StatusTooManyRequests, "RATE_LIMITED", "Rate limit exceeded") + return + } + + period := now.Format("2006-01") + if key.MonthlyQuota.Valid && key.MonthlyQuota.Int64 > 0 { + if used, err := monthlyUsage(ctx, db, "api_key_id", key.ID, period); err == nil && int64(used) >= key.MonthlyQuota.Int64 { + respondError(c, http.StatusTooManyRequests, "QUOTA_EXCEEDED", "Monthly request quota exceeded") + return + } + } + if svc.MonthlyQuota.Valid && svc.MonthlyQuota.Int64 > 0 { + if used, err := monthlyUsage(ctx, db, "service_id", svc.ID, period); err == nil && int64(used) >= svc.MonthlyQuota.Int64 { + respondError(c, http.StatusTooManyRequests, "QUOTA_EXCEEDED", "Service monthly quota exceeded") + return + } + } + + target, err := gateway.BuildTargetURL(svc.UpstreamURL, svc.RoutePrefix, c.Request.URL.Path, c.Request.URL.RawQuery) + if err != nil { + respondError(c, http.StatusBadGateway, "UPSTREAM_URL_INVALID", "Gateway upstream is misconfigured") + return + } + headers := map[string]string{} + if svc.UpstreamAuthHeader.Valid && svc.UpstreamAuthHeader.String != "" { + headers[svc.UpstreamAuthHeader.String] = svc.UpstreamAuthValue.String + } + + timeout := time.Duration(svc.RequestTimeoutMS) * time.Millisecond + if timeout <= 0 { + timeout = 8 * time.Second + } + reqCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + status, _, err := gateway.ProxyRequest(gatewayHTTPClient, c.Writer, c.Request.WithContext(reqCtx), target, headers) + if err != nil { + recordGatewayIncident(reqCtx, db, svc.ID, &key.ID, "UPSTREAM_ERROR", err.Error(), "high", http.StatusBadGateway) + recordGatewayRequest(ctx, db, svc, key, http.StatusBadGateway, period) + respondError(c, http.StatusBadGateway, "UPSTREAM_ERROR", "Upstream request failed") + return + } + + recordGatewayRequest(ctx, db, svc, key, status, period) + if status >= 500 { + recordGatewayIncident(ctx, db, svc.ID, &key.ID, "UPSTREAM_5XX", "Upstream returned "+http.StatusText(status), "high", status) + } +} diff --git a/app/backend/internal/api/gateway_proxy_test.go b/app/backend/internal/api/gateway_proxy_test.go new file mode 100644 index 0000000..6126f5f --- /dev/null +++ b/app/backend/internal/api/gateway_proxy_test.go @@ -0,0 +1,81 @@ +package api + +import ( + "net/http/httptest" + "testing" + "time" + + "github.com/gin-gonic/gin" +) + +func TestGatewayRateAllow(t *testing.T) { + now := time.Date(2026, 10, 15, 12, 0, 30, 0, time.UTC) + + for i := 0; i < 3; i++ { + if !gatewayRateAllow("test:a", 3, now) { + t.Fatalf("call %d within limit rejected", i+1) + } + } + if gatewayRateAllow("test:a", 3, now) { + t.Fatal("call beyond limit allowed") + } + if !gatewayRateAllow("test:a", 3, now.Add(time.Minute)) { + t.Fatal("new minute window still rejected") + } + if !gatewayRateAllow("test:unlimited", 0, now) { + t.Fatal("non-positive limit should disable the check") + } +} + +func TestExtractGatewayKey(t *testing.T) { + mkctx := func(headers map[string]string) *gin.Context { + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + c.Request = httptest.NewRequest("GET", "/g/x", nil) + for k, v := range headers { + c.Request.Header.Set(k, v) + } + return c + } + + if got := extractGatewayKey(mkctx(map[string]string{"X-API-Key": "ap_abc"})); got != "ap_abc" { + t.Fatalf("X-API-Key not extracted: %q", got) + } + if got := extractGatewayKey(mkctx(map[string]string{"Authorization": "Bearer ap_def"})); got != "ap_def" { + t.Fatalf("Bearer not extracted: %q", got) + } + // X-API-Key wins when both are present. + if got := extractGatewayKey(mkctx(map[string]string{ + "X-API-Key": "ap_first", + "Authorization": "Bearer ap_second", + })); got != "ap_first" { + t.Fatalf("X-API-Key should take precedence: %q", got) + } + if got := extractGatewayKey(mkctx(map[string]string{})); got != "" { + t.Fatalf("expected empty key, got %q", got) + } +} + +func TestKeyAllowsService(t *testing.T) { + svcID := "11111111-2222-3333-4444-555555555555" + other := "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" + + cases := []struct { + name string + allowed string + want bool + }{ + {"empty list permits all", "[]", true}, + {"blank permits all", "", true}, + {"invalid json permits all", "not-json", true}, + {"listed service allowed", `["` + svcID + `"]`, true}, + {"unlisted service denied", `["` + other + `"]`, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + k := &gatewayKey{AllowedServiceIDs: tc.allowed} + if got := keyAllowsService(k, svcID); got != tc.want { + t.Fatalf("keyAllowsService(%q) = %v, want %v", tc.allowed, got, tc.want) + } + }) + } +} diff --git a/app/backend/internal/api/routes.go b/app/backend/internal/api/routes.go index 4e25742..e1f119f 100644 --- a/app/backend/internal/api/routes.go +++ b/app/backend/internal/api/routes.go @@ -269,6 +269,12 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg // endpoint, verified against the provisioned webhook secret. router.POST("/api/git/github-app/webhook", handleGitHubAppWebhook) + // Public API gateway data plane. Lives outside /api/v1 so each service's + // route_prefix (/g/) maps cleanly; auth is the caller's api_key, + // not a session. + router.Any("/g/:slug", handleGatewayProxy) + router.Any("/g/:slug/*path", handleGatewayProxy) + v1 := router.Group("/api/v1") { // Public routes (no authentication required) diff --git a/app/backend/internal/gateway/proxy_test.go b/app/backend/internal/gateway/proxy_test.go new file mode 100644 index 0000000..2750cf7 --- /dev/null +++ b/app/backend/internal/gateway/proxy_test.go @@ -0,0 +1,64 @@ +package gateway + +import "testing" + +func TestBuildTargetURL(t *testing.T) { + cases := []struct { + name string + upstream, prefix, path, query, want string + }{ + { + name: "strips route prefix", + upstream: "https://api.internal:8080", + prefix: "/g/weather", + path: "/g/weather/v1/current", + query: "city=oslo", + want: "https://api.internal:8080/v1/current?city=oslo", + }, + { + name: "prefix root maps to upstream root", + upstream: "https://api.internal", + prefix: "/g/weather", + path: "/g/weather", + want: "https://api.internal/", + }, + { + name: "upstream base path is preserved", + upstream: "https://api.internal/base", + prefix: "/g/x", + path: "/g/x/ping", + want: "https://api.internal/base/ping", + }, + { + name: "root prefix leaves path untouched", + upstream: "https://api.internal", + prefix: "/", + path: "/g/raw", + want: "https://api.internal/g/raw", + }, + { + name: "path without prefix is still proxied", + upstream: "https://api.internal", + prefix: "/g/x", + path: "/other", + want: "https://api.internal/other", + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := BuildTargetURL(tc.upstream, tc.prefix, tc.path, tc.query) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != tc.want { + t.Fatalf("got %q, want %q", got, tc.want) + } + }) + } +} + +func TestBuildTargetURL_InvalidUpstream(t *testing.T) { + if _, err := BuildTargetURL("://bad", "/g/x", "/g/x/y", ""); err == nil { + t.Fatal("expected error for unparsable upstream") + } +}