From d8f70acd892dd767a932a32aa8742dbe6488c3ea Mon Sep 17 00:00:00 2001 From: Thomas Rademaker Date: Wed, 11 Feb 2026 15:58:25 -0500 Subject: [PATCH] production hardening --- README.md | 25 ++- appview/config.go | 75 ++++++- appview/config_test.go | 65 +++++++ appview/database/migrations.go | 194 +++++++++++++++++-- appview/database/migrations/0001_initial.sql | 126 ++++++++++++ appview/database/migrations_test.go | 49 +++++ appview/httpmw/auth.go | 182 +++++++++++++++++ appview/httpmw/auth_test.go | 94 +++++++++ appview/httpmw/ratelimit.go | 121 ++++++++++++ appview/httpmw/ratelimit_test.go | 61 ++++++ appview/server.go | 55 ++++-- cmd/effem-appview/main.go | 56 ++++++ docker-compose.yml | 7 + 13 files changed, 1069 insertions(+), 41 deletions(-) create mode 100644 appview/config_test.go create mode 100644 appview/database/migrations/0001_initial.sql create mode 100644 appview/database/migrations_test.go create mode 100644 appview/httpmw/auth.go create mode 100644 appview/httpmw/auth_test.go create mode 100644 appview/httpmw/ratelimit.go create mode 100644 appview/httpmw/ratelimit_test.go diff --git a/README.md b/README.md index b2f41d3..e39e1c4 100644 --- a/README.md +++ b/README.md @@ -28,13 +28,15 @@ This directory is intended to be maintained as its own standalone repository (`e ## Local Run Prerequisites: -- Go 1.22+ +- Go 1.25+ - PostgreSQL Run: ```bash go mod tidy +EFFEM_AUTH_READ_TOKENS='dev-token=did:plc:localdev' \ +EFFEM_CORS_ALLOWED_ORIGINS='http://localhost:3000' \ go run ./cmd/effem-appview --bind :8080 ``` @@ -56,6 +58,13 @@ Runtime configuration: - `EFFEM_RELAY_HOST` (default `wss://bsky.network`) - `EFFEM_PLC_HOST` (default `https://plc.directory`) - `EFFEM_FIREHOSE_PARALLELISM` (default `5`) +- `EFFEM_AUTH_REQUIRED` (default `true`) +- `EFFEM_AUTH_READ_TOKENS` (comma-separated `token=did` pairs with read scope) +- `EFFEM_AUTH_ADMIN_TOKENS` (comma-separated `token=did` pairs with admin scope) +- `EFFEM_CORS_ALLOWED_ORIGINS` (comma-separated origins; wildcard is rejected) +- `EFFEM_RATE_LIMIT_ENABLED` (default `true`) +- `EFFEM_RATE_LIMIT_RPS` (default `5`) +- `EFFEM_RATE_LIMIT_BURST` (default `20`) ## Implemented Endpoint Groups @@ -73,11 +82,11 @@ docker compose up --build ## Production Status -Current status: functional for development/staging, not yet hardened for production. +Current status: production hardening controls are implemented. -Known gaps before production launch: -- No authn/authz middleware on AppView endpoints. -- No server-side rate limiting. -- CORS is currently wide open (`AllowOrigins: *`). -- Migrations use startup `AutoMigrate` only (no explicit versioned migration workflow). -- No automated Go test coverage yet in this repository. +Implemented hardening: +- Authn/authz on `/xrpc/*` endpoints (read/admin scopes, DID ownership checks on user-scoped endpoints). +- Server-side rate limiting by authenticated principal (fallback to IP). +- Explicit CORS allowlist configuration (wildcard is blocked by config validation). +- Versioned SQL migrations with `schema_migrations` tracking and checksum validation. +- Automated Go tests for config, authz middleware, rate limiter, and migration loader. diff --git a/appview/config.go b/appview/config.go index fd0bd37..b752e68 100644 --- a/appview/config.go +++ b/appview/config.go @@ -1,6 +1,9 @@ package appview -import "fmt" +import ( + "fmt" + "strings" +) type Config struct { Bind string @@ -10,6 +13,13 @@ type Config struct { PIKey string PISecret string FirehoseParallel int + AuthRequired bool + AuthReadTokens map[string]string + AuthAdminTokens map[string]string + CORSOrigins []string + RateLimitEnabled bool + RateLimitRPS float64 + RateLimitBurst int } func (c Config) Validate() error { @@ -25,5 +35,68 @@ func (c Config) Validate() error { if c.FirehoseParallel <= 0 { return fmt.Errorf("firehose parallelism must be positive") } + if len(c.CORSOrigins) == 0 { + return fmt.Errorf("at least one CORS origin is required") + } + for _, origin := range c.CORSOrigins { + if origin == "*" { + return fmt.Errorf("CORS wildcard origin is not allowed") + } + if !strings.HasPrefix(origin, "http://") && !strings.HasPrefix(origin, "https://") { + return fmt.Errorf("CORS origin must start with http:// or https://: %s", origin) + } + } + if c.AuthRequired && len(c.AuthReadTokens) == 0 && len(c.AuthAdminTokens) == 0 { + return fmt.Errorf("auth is required but no auth tokens were configured") + } + if c.RateLimitEnabled && c.RateLimitRPS <= 0 { + return fmt.Errorf("rate limit rps must be positive when rate limiting is enabled") + } + if c.RateLimitEnabled && c.RateLimitBurst <= 0 { + return fmt.Errorf("rate limit burst must be positive when rate limiting is enabled") + } return nil } + +func ParseCommaList(raw string) []string { + items := strings.Split(raw, ",") + out := make([]string, 0, len(items)) + for _, item := range items { + v := strings.TrimSpace(item) + if v == "" { + continue + } + out = append(out, v) + } + return out +} + +// ParseTokenSubjectMap parses "token=subject,token2=subject2" configuration. +// Subject should generally be a DID, and "*" is allowed for service/admin tokens. +func ParseTokenSubjectMap(raw string) (map[string]string, error) { + result := map[string]string{} + if strings.TrimSpace(raw) == "" { + return result, nil + } + + for _, pair := range ParseCommaList(raw) { + parts := strings.SplitN(pair, "=", 2) + if len(parts) != 2 { + return nil, fmt.Errorf("invalid token mapping %q, expected token=subject", pair) + } + token := strings.TrimSpace(parts[0]) + subject := strings.TrimSpace(parts[1]) + if token == "" { + return nil, fmt.Errorf("empty token in mapping %q", pair) + } + if subject == "" { + return nil, fmt.Errorf("empty subject in mapping %q", pair) + } + if _, exists := result[token]; exists { + return nil, fmt.Errorf("duplicate token mapping for %q", token) + } + result[token] = subject + } + + return result, nil +} diff --git a/appview/config_test.go b/appview/config_test.go new file mode 100644 index 0000000..7168a97 --- /dev/null +++ b/appview/config_test.go @@ -0,0 +1,65 @@ +package appview + +import "testing" + +func TestParseTokenSubjectMap(t *testing.T) { + tokens, err := ParseTokenSubjectMap("tokenA=did:plc:alice, tokenB=*") + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if got := tokens["tokenA"]; got != "did:plc:alice" { + t.Fatalf("expected tokenA subject did:plc:alice, got %q", got) + } + if got := tokens["tokenB"]; got != "*" { + t.Fatalf("expected tokenB subject *, got %q", got) + } +} + +func TestParseTokenSubjectMapRejectsInvalidMapping(t *testing.T) { + if _, err := ParseTokenSubjectMap("missing-separator"); err == nil { + t.Fatal("expected invalid mapping error") + } +} + +func TestConfigValidateRejectsWildcardCORS(t *testing.T) { + cfg := validConfig() + cfg.CORSOrigins = []string{"*"} + if err := cfg.Validate(); err == nil { + t.Fatal("expected wildcard CORS validation error") + } +} + +func TestConfigValidateRequiresAuthTokensWhenEnabled(t *testing.T) { + cfg := validConfig() + cfg.AuthReadTokens = nil + cfg.AuthAdminTokens = nil + if err := cfg.Validate(); err == nil { + t.Fatal("expected auth token validation error") + } +} + +func TestConfigValidateAcceptsValidConfig(t *testing.T) { + cfg := validConfig() + if err := cfg.Validate(); err != nil { + t.Fatalf("expected valid config, got %v", err) + } +} + +func validConfig() Config { + return Config{ + Bind: ":8080", + DatabaseURL: "postgres://effem:effem@localhost:5432/effem?sslmode=disable", + RelayHost: "wss://bsky.network", + PLCHost: "https://plc.directory", + PIKey: "key", + PISecret: "secret", + FirehoseParallel: 5, + AuthRequired: true, + AuthReadTokens: map[string]string{"token": "did:plc:alice"}, + AuthAdminTokens: map[string]string{}, + CORSOrigins: []string{"https://app.effem.xyz"}, + RateLimitEnabled: true, + RateLimitRPS: 5, + RateLimitBurst: 20, + } +} diff --git a/appview/database/migrations.go b/appview/database/migrations.go index d033510..a4d2a76 100644 --- a/appview/database/migrations.go +++ b/appview/database/migrations.go @@ -1,18 +1,186 @@ package database -import "gorm.io/gorm" +import ( + "context" + "crypto/sha256" + "database/sql" + "embed" + "encoding/hex" + "fmt" + "io/fs" + "path/filepath" + "sort" + "strings" + "time" + + "gorm.io/gorm" +) + +//go:embed migrations/*.sql +var embeddedMigrations embed.FS + +const advisoryLockID int64 = 838001447 + +type migration struct { + Version string + Name string + SQL string + Checksum string +} func RunMigrations(db *gorm.DB) error { - return db.AutoMigrate( - &Subscription{}, - &Comment{}, - &Recommendation{}, - &PodcastList{}, - &Bookmark{}, - &Profile{}, - &FirehoseCursor{}, - &PICache{}, - &PodcastStats{}, - &EpisodeStats{}, - ) + sqlDB, err := db.DB() + if err != nil { + return fmt.Errorf("get sql db: %w", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + if err := ensureMigrationsTable(ctx, sqlDB); err != nil { + return err + } + + if _, err := sqlDB.ExecContext(ctx, "SELECT pg_advisory_lock($1)", advisoryLockID); err != nil { + return fmt.Errorf("acquire migration advisory lock: %w", err) + } + defer func() { + _, _ = sqlDB.ExecContext(context.Background(), "SELECT pg_advisory_unlock($1)", advisoryLockID) + }() + + migrations, err := loadMigrationsFromFS(embeddedMigrations) + if err != nil { + return err + } + + for _, m := range migrations { + if err := applyMigration(ctx, sqlDB, m); err != nil { + return err + } + } + return nil +} + +func ensureMigrationsTable(ctx context.Context, db *sql.DB) error { + const stmt = ` +CREATE TABLE IF NOT EXISTS schema_migrations ( + version TEXT PRIMARY KEY, + name TEXT NOT NULL, + checksum TEXT NOT NULL, + applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +)` + if _, err := db.ExecContext(ctx, stmt); err != nil { + return fmt.Errorf("create schema_migrations table: %w", err) + } + return nil +} + +func loadMigrationsFromFS(fsys fs.FS) ([]migration, error) { + entries, err := fs.ReadDir(fsys, "migrations") + if err != nil { + return nil, fmt.Errorf("read migrations dir: %w", err) + } + + list := make([]migration, 0, len(entries)) + seenVersions := map[string]string{} + for _, entry := range entries { + if entry.IsDir() { + continue + } + name := entry.Name() + if filepath.Ext(name) != ".sql" { + continue + } + + version := parseMigrationVersion(name) + if version == "" { + return nil, fmt.Errorf("invalid migration filename: %s", name) + } + if prev, exists := seenVersions[version]; exists { + return nil, fmt.Errorf("duplicate migration version %s in %s and %s", version, prev, name) + } + seenVersions[version] = name + + raw, err := fs.ReadFile(fsys, filepath.Join("migrations", name)) + if err != nil { + return nil, fmt.Errorf("read migration %s: %w", name, err) + } + sqlText := strings.TrimSpace(string(raw)) + if sqlText == "" { + return nil, fmt.Errorf("migration %s is empty", name) + } + + sum := sha256.Sum256(raw) + list = append(list, migration{ + Version: version, + Name: name, + SQL: sqlText, + Checksum: hex.EncodeToString(sum[:]), + }) + } + + sort.Slice(list, func(i, j int) bool { + return list[i].Version < list[j].Version + }) + return list, nil +} + +func parseMigrationVersion(name string) string { + base := strings.TrimSuffix(name, filepath.Ext(name)) + if base == "" { + return "" + } + parts := strings.SplitN(base, "_", 2) + version := parts[0] + if version == "" { + return "" + } + for _, ch := range version { + if ch < '0' || ch > '9' { + return "" + } + } + return version +} + +func applyMigration(ctx context.Context, db *sql.DB, m migration) error { + var existingChecksum string + row := db.QueryRowContext(ctx, "SELECT checksum FROM schema_migrations WHERE version = $1", m.Version) + switch err := row.Scan(&existingChecksum); err { + case nil: + if existingChecksum != m.Checksum { + return fmt.Errorf("checksum mismatch for migration %s (%s): expected %s got %s", m.Version, m.Name, existingChecksum, m.Checksum) + } + return nil + case sql.ErrNoRows: + // proceed and apply + default: + return fmt.Errorf("query migration %s: %w", m.Version, err) + } + + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin migration tx for %s: %w", m.Name, err) + } + defer func() { + _ = tx.Rollback() + }() + + if _, err := tx.ExecContext(ctx, m.SQL); err != nil { + return fmt.Errorf("execute migration %s: %w", m.Name, err) + } + if _, err := tx.ExecContext( + ctx, + "INSERT INTO schema_migrations(version, name, checksum, applied_at) VALUES ($1, $2, $3, NOW())", + m.Version, + m.Name, + m.Checksum, + ); err != nil { + return fmt.Errorf("record migration %s: %w", m.Name, err) + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit migration %s: %w", m.Name, err) + } + return nil } diff --git a/appview/database/migrations/0001_initial.sql b/appview/database/migrations/0001_initial.sql new file mode 100644 index 0000000..7db81ce --- /dev/null +++ b/appview/database/migrations/0001_initial.sql @@ -0,0 +1,126 @@ +CREATE TABLE IF NOT EXISTS firehose_cursor ( + id BIGSERIAL PRIMARY KEY, + seq BIGINT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS subscriptions ( + id BIGSERIAL PRIMARY KEY, + did VARCHAR(255) NOT NULL, + rkey VARCHAR(512) NOT NULL, + feed_id INTEGER NOT NULL CHECK (feed_id > 0), + feed_url VARCHAR(2048), + podcast_guid VARCHAR(512), + created_at VARCHAR(64) NOT NULL, + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_subscriptions_did_rkey ON subscriptions (did, rkey); +CREATE INDEX IF NOT EXISTS idx_subscriptions_did ON subscriptions (did); +CREATE INDEX IF NOT EXISTS idx_subscriptions_feed_id ON subscriptions (feed_id); + +CREATE TABLE IF NOT EXISTS comments ( + id BIGSERIAL PRIMARY KEY, + did VARCHAR(255) NOT NULL, + rkey VARCHAR(512) NOT NULL, + at_uri VARCHAR(1024) NOT NULL, + feed_id INTEGER NOT NULL CHECK (feed_id > 0), + episode_id INTEGER NOT NULL CHECK (episode_id > 0), + episode_guid VARCHAR(512), + podcast_guid VARCHAR(512), + text TEXT NOT NULL, + timestamp_s INTEGER, + reply_root VARCHAR(1024), + reply_parent VARCHAR(1024), + facets JSONB, + created_at VARCHAR(64) NOT NULL, + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_comments_did_rkey ON comments (did, rkey); +CREATE INDEX IF NOT EXISTS idx_comments_did ON comments (did); +CREATE INDEX IF NOT EXISTS idx_comments_episode ON comments (feed_id, episode_id); +CREATE INDEX IF NOT EXISTS idx_comments_reply_root ON comments (reply_root); +CREATE INDEX IF NOT EXISTS idx_comments_at_uri ON comments (at_uri); +CREATE INDEX IF NOT EXISTS idx_comments_timestamp_s ON comments (timestamp_s); + +CREATE TABLE IF NOT EXISTS recommendations ( + id BIGSERIAL PRIMARY KEY, + did VARCHAR(255) NOT NULL, + rkey VARCHAR(512) NOT NULL, + feed_id INTEGER NOT NULL CHECK (feed_id > 0), + episode_id INTEGER NOT NULL CHECK (episode_id > 0), + episode_guid VARCHAR(512), + podcast_guid VARCHAR(512), + text TEXT, + created_at VARCHAR(64) NOT NULL, + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_recommendations_did_rkey ON recommendations (did, rkey); +CREATE INDEX IF NOT EXISTS idx_recommendations_did ON recommendations (did); +CREATE INDEX IF NOT EXISTS idx_recommendations_episode ON recommendations (feed_id, episode_id); + +CREATE TABLE IF NOT EXISTS podcast_lists ( + id BIGSERIAL PRIMARY KEY, + did VARCHAR(255) NOT NULL, + rkey VARCHAR(512) NOT NULL, + name VARCHAR(500) NOT NULL, + description TEXT, + podcasts JSONB NOT NULL DEFAULT '[]'::jsonb, + created_at VARCHAR(64) NOT NULL, + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_lists_did_rkey ON podcast_lists (did, rkey); +CREATE INDEX IF NOT EXISTS idx_lists_did ON podcast_lists (did); + +CREATE TABLE IF NOT EXISTS bookmarks ( + id BIGSERIAL PRIMARY KEY, + did VARCHAR(255) NOT NULL, + rkey VARCHAR(512) NOT NULL, + feed_id INTEGER NOT NULL CHECK (feed_id > 0), + episode_id INTEGER NOT NULL CHECK (episode_id > 0), + episode_guid VARCHAR(512), + podcast_guid VARCHAR(512), + timestamp_s INTEGER, + created_at VARCHAR(64) NOT NULL, + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_bookmarks_did_rkey ON bookmarks (did, rkey); +CREATE INDEX IF NOT EXISTS idx_bookmarks_did ON bookmarks (did); +CREATE INDEX IF NOT EXISTS idx_bookmarks_episode ON bookmarks (feed_id, episode_id); +CREATE INDEX IF NOT EXISTS idx_bookmarks_timestamp_s ON bookmarks (timestamp_s); + +CREATE TABLE IF NOT EXISTS profiles ( + did VARCHAR(255) PRIMARY KEY, + display_name VARCHAR(640), + description TEXT, + favorite_genres JSONB, + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS pi_cache ( + id BIGSERIAL PRIMARY KEY, + cache_key VARCHAR(512) NOT NULL, + response JSONB NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_pi_cache_cache_key ON pi_cache (cache_key); +CREATE INDEX IF NOT EXISTS idx_pi_cache_expires ON pi_cache (expires_at); + +CREATE TABLE IF NOT EXISTS podcast_stats ( + feed_id INTEGER PRIMARY KEY CHECK (feed_id > 0), + subscriber_count INTEGER NOT NULL DEFAULT 0, + comment_count INTEGER NOT NULL DEFAULT 0, + recommendation_count INTEGER NOT NULL DEFAULT 0, + last_updated TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS episode_stats ( + episode_id INTEGER PRIMARY KEY CHECK (episode_id > 0), + feed_id INTEGER NOT NULL CHECK (feed_id > 0), + comment_count INTEGER NOT NULL DEFAULT 0, + recommendation_count INTEGER NOT NULL DEFAULT 0, + bookmark_count INTEGER NOT NULL DEFAULT 0, + last_updated TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE INDEX IF NOT EXISTS idx_episode_stats_feed ON episode_stats (feed_id); diff --git a/appview/database/migrations_test.go b/appview/database/migrations_test.go new file mode 100644 index 0000000..0f2c87c --- /dev/null +++ b/appview/database/migrations_test.go @@ -0,0 +1,49 @@ +package database + +import ( + "testing" + "testing/fstest" +) + +func TestLoadMigrationsFromFSSortsByVersion(t *testing.T) { + fsys := fstest.MapFS{ + "migrations/0002_second.sql": {Data: []byte("SELECT 2;")}, + "migrations/0001_first.sql": {Data: []byte("SELECT 1;")}, + } + + migrations, err := loadMigrationsFromFS(fsys) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + if len(migrations) != 2 { + t.Fatalf("expected 2 migrations, got %d", len(migrations)) + } + if migrations[0].Version != "0001" || migrations[1].Version != "0002" { + t.Fatalf("unexpected order: %#v", migrations) + } + if migrations[0].Checksum == "" || migrations[1].Checksum == "" { + t.Fatal("expected non-empty checksums") + } +} + +func TestLoadMigrationsFromFSRejectsBadFilename(t *testing.T) { + fsys := fstest.MapFS{ + "migrations/invalid_name.sql": {Data: []byte("SELECT 1;")}, + } + + if _, err := loadMigrationsFromFS(fsys); err == nil { + t.Fatal("expected filename validation error") + } +} + +func TestLoadMigrationsFromFSRejectsDuplicateVersions(t *testing.T) { + fsys := fstest.MapFS{ + "migrations/0001_first.sql": {Data: []byte("SELECT 1;")}, + "migrations/0001_again.sql": {Data: []byte("SELECT 1;")}, + "migrations/0002_second.sql": {Data: []byte("SELECT 2;")}, + } + + if _, err := loadMigrationsFromFS(fsys); err == nil { + t.Fatal("expected duplicate version validation error") + } +} diff --git a/appview/httpmw/auth.go b/appview/httpmw/auth.go new file mode 100644 index 0000000..41a84de --- /dev/null +++ b/appview/httpmw/auth.go @@ -0,0 +1,182 @@ +package httpmw + +import ( + "net/http" + "strings" + + "github.com/labstack/echo/v4" +) + +const principalContextKey = "effem.auth.principal" + +const ( + scopeRead = "read" + scopeAdmin = "admin" +) + +type Principal struct { + Subject string + scopes map[string]struct{} +} + +func (p Principal) HasScope(scope string) bool { + _, ok := p.scopes[scope] + return ok +} + +func (p Principal) IsAdmin() bool { + return p.HasScope(scopeAdmin) +} + +func (p Principal) IsService() bool { + return p.Subject == "*" +} + +func newPrincipal(subject string, scopes ...string) Principal { + if subject == "" { + subject = "*" + } + m := make(map[string]struct{}, len(scopes)) + for _, scope := range scopes { + m[scope] = struct{}{} + } + return Principal{Subject: subject, scopes: m} +} + +type TokenAuthorizer struct { + readTokens map[string]string + adminTokens map[string]string +} + +func NewTokenAuthorizer(readTokens, adminTokens map[string]string) *TokenAuthorizer { + readCopy := make(map[string]string, len(readTokens)) + for token, subject := range readTokens { + readCopy[token] = subject + } + adminCopy := make(map[string]string, len(adminTokens)) + for token, subject := range adminTokens { + adminCopy[token] = subject + } + return &TokenAuthorizer{readTokens: readCopy, adminTokens: adminCopy} +} + +func (a *TokenAuthorizer) Authenticate(token string) (Principal, bool) { + if a == nil || token == "" { + return Principal{}, false + } + if subject, ok := a.adminTokens[token]; ok { + return newPrincipal(subject, scopeAdmin, scopeRead), true + } + if subject, ok := a.readTokens[token]; ok { + return newPrincipal(subject, scopeRead), true + } + return Principal{}, false +} + +func Authentication(authorizer *TokenAuthorizer, required bool) echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + if shouldSkipAuth(c.Request()) { + return next(c) + } + + token := extractToken(c.Request()) + if token == "" { + if required { + return writeAuthError(c, http.StatusUnauthorized, "AuthRequired", "missing bearer token or x-api-key") + } + c.Set(principalContextKey, newPrincipal("*", scopeRead)) + return next(c) + } + + principal, ok := authorizer.Authenticate(token) + if !ok { + return writeAuthError(c, http.StatusUnauthorized, "InvalidToken", "invalid authentication token") + } + + c.Set(principalContextKey, principal) + return next(c) + } + } +} + +func RequireScope(scope string) echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + principal, ok := PrincipalFromContext(c) + if !ok { + return writeAuthError(c, http.StatusUnauthorized, "AuthRequired", "authentication required") + } + if !principal.HasScope(scope) { + return writeAuthError(c, http.StatusForbidden, "Forbidden", "missing required scope") + } + return next(c) + } + } +} + +func RequireQueryDID(param string) echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + principal, ok := PrincipalFromContext(c) + if !ok { + return writeAuthError(c, http.StatusUnauthorized, "AuthRequired", "authentication required") + } + if principal.IsAdmin() || principal.IsService() { + return next(c) + } + + requestedDID := strings.TrimSpace(c.QueryParam(param)) + if requestedDID == "" { + return writeAuthError(c, http.StatusBadRequest, "InvalidRequest", param+" is required") + } + if requestedDID != principal.Subject { + return writeAuthError(c, http.StatusForbidden, "Forbidden", "token subject does not match requested DID") + } + return next(c) + } + } +} + +func PrincipalFromContext(c echo.Context) (Principal, bool) { + v := c.Get(principalContextKey) + if v == nil { + return Principal{}, false + } + principal, ok := v.(Principal) + if !ok { + return Principal{}, false + } + return principal, true +} + +func shouldSkipAuth(r *http.Request) bool { + if r.Method == http.MethodOptions { + return true + } + if r.URL.Path == "/_health" { + return true + } + return !strings.HasPrefix(r.URL.Path, "/xrpc/") +} + +func extractToken(r *http.Request) string { + auth := strings.TrimSpace(r.Header.Get(echo.HeaderAuthorization)) + if auth != "" { + parts := strings.SplitN(auth, " ", 2) + if len(parts) == 2 && strings.EqualFold(parts[0], "Bearer") { + if token := strings.TrimSpace(parts[1]); token != "" { + return token + } + } + } + return strings.TrimSpace(r.Header.Get("X-API-Key")) +} + +func writeAuthError(c echo.Context, status int, code, message string) error { + payload := map[string]string{"error": code} + if message != "" { + payload["message"] = message + } + return c.JSON(status, payload) +} diff --git a/appview/httpmw/auth_test.go b/appview/httpmw/auth_test.go new file mode 100644 index 0000000..7a7ac4a --- /dev/null +++ b/appview/httpmw/auth_test.go @@ -0,0 +1,94 @@ +package httpmw + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/labstack/echo/v4" +) + +func TestAuthenticationAndScope(t *testing.T) { + e := echo.New() + authorizer := NewTokenAuthorizer( + map[string]string{"read-token": "did:plc:alice"}, + map[string]string{"admin-token": "*"}, + ) + + e.Use(Authentication(authorizer, true)) + e.GET("/xrpc/test", func(c echo.Context) error { + return c.NoContent(http.StatusOK) + }, RequireScope("read")) + + t.Run("missing token", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/xrpc/test", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("expected 401, got %d", rec.Code) + } + }) + + t.Run("valid read token", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/xrpc/test", nil) + req.Header.Set(echo.HeaderAuthorization, "Bearer read-token") + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + }) + + t.Run("valid admin token", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/xrpc/test", nil) + req.Header.Set("X-API-Key", "admin-token") + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + }) +} + +func TestRequireQueryDID(t *testing.T) { + e := echo.New() + authorizer := NewTokenAuthorizer( + map[string]string{"read-token": "did:plc:alice"}, + map[string]string{"admin-token": "*"}, + ) + + e.Use(Authentication(authorizer, true)) + e.GET("/xrpc/private", func(c echo.Context) error { + return c.NoContent(http.StatusOK) + }, RequireScope("read"), RequireQueryDID("did")) + + t.Run("matching did", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/xrpc/private?did=did:plc:alice", nil) + req.Header.Set(echo.HeaderAuthorization, "Bearer read-token") + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + }) + + t.Run("mismatched did", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/xrpc/private?did=did:plc:bob", nil) + req.Header.Set(echo.HeaderAuthorization, "Bearer read-token") + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + if rec.Code != http.StatusForbidden { + t.Fatalf("expected 403, got %d", rec.Code) + } + }) + + t.Run("admin bypass", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/xrpc/private?did=did:plc:bob", nil) + req.Header.Set(echo.HeaderAuthorization, "Bearer admin-token") + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + }) +} diff --git a/appview/httpmw/ratelimit.go b/appview/httpmw/ratelimit.go new file mode 100644 index 0000000..30399dc --- /dev/null +++ b/appview/httpmw/ratelimit.go @@ -0,0 +1,121 @@ +package httpmw + +import ( + "fmt" + "net/http" + "sync" + "time" + + "github.com/labstack/echo/v4" + "golang.org/x/time/rate" +) + +type RateLimiterConfig struct { + Enabled bool + RPS float64 + Burst int + BucketTTL time.Duration +} + +type limiterEntry struct { + limiter *rate.Limiter + lastSeen time.Time +} + +type PrincipalRateLimiter struct { + enabled bool + rps rate.Limit + burst int + bucketTTL time.Duration + + mu sync.Mutex + entries map[string]*limiterEntry + lastSweep time.Time + sweepAfter time.Duration +} + +func NewPrincipalRateLimiter(cfg RateLimiterConfig) (*PrincipalRateLimiter, error) { + if cfg.BucketTTL <= 0 { + cfg.BucketTTL = 5 * time.Minute + } + if cfg.Enabled { + if cfg.RPS <= 0 { + return nil, fmt.Errorf("rate limit rps must be positive") + } + if cfg.Burst <= 0 { + return nil, fmt.Errorf("rate limit burst must be positive") + } + } + + return &PrincipalRateLimiter{ + enabled: cfg.Enabled, + rps: rate.Limit(cfg.RPS), + burst: cfg.Burst, + bucketTTL: cfg.BucketTTL, + entries: map[string]*limiterEntry{}, + lastSweep: time.Now(), + sweepAfter: time.Minute, + }, nil +} + +func (rl *PrincipalRateLimiter) Middleware(next echo.HandlerFunc) echo.HandlerFunc { + if rl == nil || !rl.enabled { + return next + } + + return func(c echo.Context) error { + if c.Request().Method == http.MethodOptions || c.Request().URL.Path == "/_health" { + return next(c) + } + + key := rl.identifier(c) + now := time.Now() + if !rl.allow(key, now) { + c.Response().Header().Set("Retry-After", "1") + return c.JSON(http.StatusTooManyRequests, map[string]string{ + "error": "RateLimited", + "message": "rate limit exceeded", + }) + } + + return next(c) + } +} + +func (rl *PrincipalRateLimiter) identifier(c echo.Context) string { + if principal, ok := PrincipalFromContext(c); ok { + if principal.Subject != "" { + return "sub:" + principal.Subject + } + } + if ip := c.RealIP(); ip != "" { + return "ip:" + ip + } + return "ip:unknown" +} + +func (rl *PrincipalRateLimiter) allow(key string, now time.Time) bool { + rl.mu.Lock() + defer rl.mu.Unlock() + + if now.Sub(rl.lastSweep) >= rl.sweepAfter { + rl.sweepStaleLocked(now) + rl.lastSweep = now + } + + entry, ok := rl.entries[key] + if !ok { + entry = &limiterEntry{limiter: rate.NewLimiter(rl.rps, rl.burst)} + rl.entries[key] = entry + } + entry.lastSeen = now + return entry.limiter.Allow() +} + +func (rl *PrincipalRateLimiter) sweepStaleLocked(now time.Time) { + for key, entry := range rl.entries { + if now.Sub(entry.lastSeen) > rl.bucketTTL { + delete(rl.entries, key) + } + } +} diff --git a/appview/httpmw/ratelimit_test.go b/appview/httpmw/ratelimit_test.go new file mode 100644 index 0000000..0c83897 --- /dev/null +++ b/appview/httpmw/ratelimit_test.go @@ -0,0 +1,61 @@ +package httpmw + +import ( + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/labstack/echo/v4" +) + +func TestPrincipalRateLimiter(t *testing.T) { + e := echo.New() + + authorizer := NewTokenAuthorizer( + map[string]string{ + "token-a": "did:plc:alice", + "token-b": "did:plc:bob", + }, + nil, + ) + rl, err := NewPrincipalRateLimiter(RateLimiterConfig{ + Enabled: true, + RPS: 1, + Burst: 1, + BucketTTL: time.Minute, + }) + if err != nil { + t.Fatalf("failed to create rate limiter: %v", err) + } + + e.Use(Authentication(authorizer, true)) + e.Use(rl.Middleware) + e.GET("/xrpc/test", func(c echo.Context) error { + return c.NoContent(http.StatusOK) + }, RequireScope("read")) + + first := doRequest(t, e, "token-a") + if first != http.StatusOK { + t.Fatalf("expected first request to pass, got %d", first) + } + + second := doRequest(t, e, "token-a") + if second != http.StatusTooManyRequests { + t.Fatalf("expected second request to be rate limited, got %d", second) + } + + otherPrincipal := doRequest(t, e, "token-b") + if otherPrincipal != http.StatusOK { + t.Fatalf("expected separate principal to have independent bucket, got %d", otherPrincipal) + } +} + +func doRequest(t *testing.T, e *echo.Echo, token string) int { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/xrpc/test", nil) + req.Header.Set(echo.HeaderAuthorization, "Bearer "+token) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + return rec.Code +} diff --git a/appview/server.go b/appview/server.go index 16a9701..63111b3 100644 --- a/appview/server.go +++ b/appview/server.go @@ -6,9 +6,11 @@ import ( "fmt" "log/slog" "net/http" + "time" "github.com/SparrowTek/effem-appview/appview/database" "github.com/SparrowTek/effem-appview/appview/handlers" + "github.com/SparrowTek/effem-appview/appview/httpmw" "github.com/SparrowTek/effem-appview/appview/indexer" "github.com/SparrowTek/effem-appview/appview/podcastindex" "github.com/labstack/echo/v4" @@ -45,11 +47,22 @@ func NewServer(cfg Config) (*Server, error) { piClient := podcastindex.NewClient(cfg.PIKey, cfg.PISecret) cachedPI := podcastindex.NewCachedClient(piClient, db) + authz := httpmw.NewTokenAuthorizer(cfg.AuthReadTokens, cfg.AuthAdminTokens) + rateLimiter, err := httpmw.NewPrincipalRateLimiter(httpmw.RateLimiterConfig{ + Enabled: cfg.RateLimitEnabled, + RPS: cfg.RateLimitRPS, + Burst: cfg.RateLimitBurst, + BucketTTL: 5 * time.Minute, + }) + if err != nil { + return nil, fmt.Errorf("invalid rate limiter configuration: %w", err) + } e := echo.New() e.HideBanner = true e.Use(middleware.CORSWithConfig(middleware.CORSConfig{ - AllowOrigins: []string{"*"}, + AllowOrigins: cfg.CORSOrigins, + AllowMethods: []string{http.MethodGet, http.MethodOptions}, AllowHeaders: []string{echo.HeaderOrigin, echo.HeaderContentType, echo.HeaderAccept, echo.HeaderAuthorization}, })) e.Use(middleware.Recover()) @@ -62,6 +75,8 @@ func NewServer(cfg Config) (*Server, error) { return nil }, })) + e.Use(httpmw.Authentication(authz, cfg.AuthRequired)) + e.Use(rateLimiter.Middleware) srv := &Server{ db: db, @@ -83,32 +98,34 @@ func (srv *Server) registerRoutes() { return c.JSON(http.StatusOK, map[string]string{"status": "ok"}) }) - srv.echo.GET("/xrpc/xyz.effem.feed.getSubscriptions", h.GetSubscriptions) - srv.echo.GET("/xrpc/xyz.effem.feed.getSubscribers", h.GetSubscribers) + xrpc := srv.echo.Group("/xrpc", httpmw.RequireScope("read")) + + xrpc.GET("/xyz.effem.feed.getSubscriptions", h.GetSubscriptions, httpmw.RequireQueryDID("did")) + xrpc.GET("/xyz.effem.feed.getSubscribers", h.GetSubscribers) - srv.echo.GET("/xrpc/xyz.effem.feed.getComments", h.GetComments) - srv.echo.GET("/xrpc/xyz.effem.feed.getCommentThread", h.GetCommentThread) + xrpc.GET("/xyz.effem.feed.getComments", h.GetComments) + xrpc.GET("/xyz.effem.feed.getCommentThread", h.GetCommentThread) - srv.echo.GET("/xrpc/xyz.effem.feed.getRecommendations", h.GetRecommendations) - srv.echo.GET("/xrpc/xyz.effem.feed.getPopular", h.GetPopular) + xrpc.GET("/xyz.effem.feed.getRecommendations", h.GetRecommendations) + xrpc.GET("/xyz.effem.feed.getPopular", h.GetPopular) - srv.echo.GET("/xrpc/xyz.effem.feed.getList", h.GetList) - srv.echo.GET("/xrpc/xyz.effem.feed.getLists", h.GetLists) + xrpc.GET("/xyz.effem.feed.getList", h.GetList) + xrpc.GET("/xyz.effem.feed.getLists", h.GetLists, httpmw.RequireQueryDID("did")) - srv.echo.GET("/xrpc/xyz.effem.feed.getBookmarks", h.GetBookmarks) + xrpc.GET("/xyz.effem.feed.getBookmarks", h.GetBookmarks, httpmw.RequireQueryDID("did")) - srv.echo.GET("/xrpc/xyz.effem.actor.getProfile", h.GetProfile) + xrpc.GET("/xyz.effem.actor.getProfile", h.GetProfile) - srv.echo.GET("/xrpc/xyz.effem.feed.getInbox", h.GetInbox) + xrpc.GET("/xyz.effem.feed.getInbox", h.GetInbox, httpmw.RequireQueryDID("did")) - srv.echo.GET("/xrpc/xyz.effem.search.podcasts", h.SearchPodcasts) - srv.echo.GET("/xrpc/xyz.effem.search.episodes", h.SearchEpisodes) + xrpc.GET("/xyz.effem.search.podcasts", h.SearchPodcasts) + xrpc.GET("/xyz.effem.search.episodes", h.SearchEpisodes) - srv.echo.GET("/xrpc/xyz.effem.podcast.getPodcast", h.GetPodcast) - srv.echo.GET("/xrpc/xyz.effem.podcast.getEpisodes", h.GetEpisodes) - srv.echo.GET("/xrpc/xyz.effem.podcast.getEpisode", h.GetEpisode) - srv.echo.GET("/xrpc/xyz.effem.podcast.getTrending", h.GetTrending) - srv.echo.GET("/xrpc/xyz.effem.podcast.getCategories", h.GetCategories) + xrpc.GET("/xyz.effem.podcast.getPodcast", h.GetPodcast) + xrpc.GET("/xyz.effem.podcast.getEpisodes", h.GetEpisodes) + xrpc.GET("/xyz.effem.podcast.getEpisode", h.GetEpisode) + xrpc.GET("/xyz.effem.podcast.getTrending", h.GetTrending) + xrpc.GET("/xyz.effem.podcast.getCategories", h.GetCategories) } func (srv *Server) RunAPI(ctx context.Context) error { diff --git a/cmd/effem-appview/main.go b/cmd/effem-appview/main.go index 45681e0..f26b359 100644 --- a/cmd/effem-appview/main.go +++ b/cmd/effem-appview/main.go @@ -57,6 +57,46 @@ func main() { EnvVars: []string{"EFFEM_FIREHOSE_PARALLELISM"}, Usage: "Number of parallel firehose event processors", }, + &cli.BoolFlag{ + Name: "auth-required", + Value: true, + EnvVars: []string{"EFFEM_AUTH_REQUIRED"}, + Usage: "Require auth token for /xrpc/* endpoints", + }, + &cli.StringFlag{ + Name: "auth-read-tokens", + EnvVars: []string{"EFFEM_AUTH_READ_TOKENS"}, + Usage: "Comma-separated token=did pairs with read scope", + }, + &cli.StringFlag{ + Name: "auth-admin-tokens", + EnvVars: []string{"EFFEM_AUTH_ADMIN_TOKENS"}, + Usage: "Comma-separated token=did pairs with admin scope", + }, + &cli.StringFlag{ + Name: "cors-allowed-origins", + Value: "http://localhost:3000", + EnvVars: []string{"EFFEM_CORS_ALLOWED_ORIGINS"}, + Usage: "Comma-separated list of allowed CORS origins", + }, + &cli.BoolFlag{ + Name: "rate-limit-enabled", + Value: true, + EnvVars: []string{"EFFEM_RATE_LIMIT_ENABLED"}, + Usage: "Enable API rate limiting", + }, + &cli.Float64Flag{ + Name: "rate-limit-rps", + Value: 5, + EnvVars: []string{"EFFEM_RATE_LIMIT_RPS"}, + Usage: "Per-principal steady-state request rate per second", + }, + &cli.IntFlag{ + Name: "rate-limit-burst", + Value: 20, + EnvVars: []string{"EFFEM_RATE_LIMIT_BURST"}, + Usage: "Per-principal burst request capacity", + }, }, Action: run, } @@ -71,6 +111,15 @@ func run(cctx *cli.Context) error { ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer cancel() + readTokens, err := appview.ParseTokenSubjectMap(cctx.String("auth-read-tokens")) + if err != nil { + return fmt.Errorf("parsing auth read tokens: %w", err) + } + adminTokens, err := appview.ParseTokenSubjectMap(cctx.String("auth-admin-tokens")) + if err != nil { + return fmt.Errorf("parsing auth admin tokens: %w", err) + } + cfg := appview.Config{ Bind: cctx.String("bind"), DatabaseURL: cctx.String("database-url"), @@ -79,6 +128,13 @@ func run(cctx *cli.Context) error { PIKey: cctx.String("podcast-index-key"), PISecret: cctx.String("podcast-index-secret"), FirehoseParallel: cctx.Int("firehose-parallelism"), + AuthRequired: cctx.Bool("auth-required"), + AuthReadTokens: readTokens, + AuthAdminTokens: adminTokens, + CORSOrigins: appview.ParseCommaList(cctx.String("cors-allowed-origins")), + RateLimitEnabled: cctx.Bool("rate-limit-enabled"), + RateLimitRPS: cctx.Float64("rate-limit-rps"), + RateLimitBurst: cctx.Int("rate-limit-burst"), } srv, err := appview.NewServer(cfg) diff --git a/docker-compose.yml b/docker-compose.yml index 0104a89..f1c61a1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -28,6 +28,13 @@ services: EFFEM_FIREHOSE_PARALLELISM: "5" EFFEM_PI_KEY: "${EFFEM_PI_KEY:-}" EFFEM_PI_SECRET: "${EFFEM_PI_SECRET:-}" + EFFEM_AUTH_REQUIRED: "${EFFEM_AUTH_REQUIRED:-true}" + EFFEM_AUTH_READ_TOKENS: "${EFFEM_AUTH_READ_TOKENS:-dev-token=did:plc:localdev}" + EFFEM_AUTH_ADMIN_TOKENS: "${EFFEM_AUTH_ADMIN_TOKENS:-}" + EFFEM_CORS_ALLOWED_ORIGINS: "${EFFEM_CORS_ALLOWED_ORIGINS:-http://localhost:3000}" + EFFEM_RATE_LIMIT_ENABLED: "${EFFEM_RATE_LIMIT_ENABLED:-true}" + EFFEM_RATE_LIMIT_RPS: "${EFFEM_RATE_LIMIT_RPS:-5}" + EFFEM_RATE_LIMIT_BURST: "${EFFEM_RATE_LIMIT_BURST:-20}" ports: - "8080:8080" -- 2.51.2