Harden the surface: 0600 databases, checksum-verified installs, rate-limited login and key exchange

coordinator/internal/storage/sqlite/sqlite.go

@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"os"
 	"time"
 
 	_ "modernc.org/sqlite"
@@ -37,9 +38,29 @@ func Open(path string) (*sql.DB, error) {
 		_ = db.Close()
 		return nil, fmt.Errorf("ping sqlite database: %w", err)
 	}
+	if err := lockDownDatabase(path); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("lock down sqlite database: %w", err)
+	}
 	return db, nil
 }
 
+// lockDownDatabase restricts the database files to the owner: sqlite creates
+// them with the process umask (0644), which would let any local user read job
+// metadata and password hashes. WAL/SHM siblings inherit the main file's mode,
+// so existing ones are corrected too. Best-effort: failures only warn callers
+// via the returned error, never corrupt state.
+func lockDownDatabase(path string) error {
+	for _, candidate := range []string{path, path + "-wal", path + "-shm"} {
+		if info, err := os.Stat(candidate); err == nil && !info.IsDir() {
+			if err := os.Chmod(candidate, 0o600); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // querier is satisfied by both *sql.DB and *sql.Tx, letting every repository
 // method run identically inside or outside a transaction.
 type querier interface {

coordinator/internal/storage/sqlite/sqlite_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"os"
 	"path/filepath"
 	"testing"
 	"time"
@@ -347,3 +348,19 @@ func TestCancelByJobInvalidatesTasks(t *testing.T) {
 		t.Errorf("cancelled task = %+v", got)
 	}
 }
+
+func TestOpenRestrictsDatabasePermissions(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "locked.db")
+	db, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = db.Close()
+	info, err := os.Stat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if perm := info.Mode().Perm(); perm != 0o600 {
+		t.Errorf("db perms = %o, want 600", perm)
+	}
+}

coordinator/internal/transport/http/ratelimit.go

@@ -0,0 +1,117 @@
+package http
+
+import (
+	"net"
+	"net/http"
+	"sync"
+	"time"
+)
+
+// documentedLimits describes the default policy for the two public surfaces;
+// keep in sync with loginRatePerMinute and exchangeRatePerMinute below.
+const (
+	loginRatePerMinute    = 10
+	loginBurst            = 5
+	exchangeRatePerMinute = 30
+	exchangeBurst         = 10
+)
+
+// tokenBucket is a fixed-rate token bucket for one client address.
+type tokenBucket struct {
+	mu     sync.Mutex
+	tokens float64
+	last   time.Time
+	rate   float64 // tokens per second
+	burst  float64
+}
+
+func newTokenBucket(ratePerMinute, burst float64) *tokenBucket {
+	return &tokenBucket{
+		tokens: burst,
+		last:   time.Now(),
+		rate:   ratePerMinute / 60,
+		burst:  burst,
+	}
+}
+
+// allow consumes one token when available; the bucket refills continuously.
+func (b *tokenBucket) allow() bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	now := time.Now()
+	b.tokens += now.Sub(b.last).Seconds() * b.rate
+	if b.tokens > b.burst {
+		b.tokens = b.burst
+	}
+	b.last = now
+	if b.tokens < 1 {
+		return false
+	}
+	b.tokens--
+	return true
+}
+
+// ipLimiter tracks one bucket per client address and prunes stale entries.
+type ipLimiter struct {
+	mu            sync.Mutex
+	buckets       map[string]*tokenBucket
+	ratePerMinute float64
+	burst         float64
+}
+
+func newIPLimiter(ratePerMinute, burst float64) *ipLimiter {
+	return &ipLimiter{
+		buckets:       make(map[string]*tokenBucket),
+		ratePerMinute: ratePerMinute,
+		burst:         burst,
+	}
+}
+
+// Allow reports whether the caller's address may proceed. It also sweeps
+// entries idle for more than ten minutes so the map stays bounded.
+func (l *ipLimiter) Allow(r *http.Request) bool {
+	ip := remoteIP(r)
+	l.mu.Lock()
+	if len(l.buckets) > 1024 {
+		cutoff := time.Now().Add(-10 * time.Minute)
+		for addr, bucket := range l.buckets {
+			bucket.mu.Lock()
+			idle := bucket.last.Before(cutoff)
+			bucket.mu.Unlock()
+			if idle {
+				delete(l.buckets, addr)
+			}
+		}
+	}
+	bucket, ok := l.buckets[ip]
+	if !ok {
+		bucket = newTokenBucket(l.ratePerMinute, l.burst)
+		l.buckets[ip] = bucket
+	}
+	l.mu.Unlock()
+	return bucket.allow()
+}
+
+func remoteIP(r *http.Request) string {
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		return r.RemoteAddr
+	}
+	return host
+}
+
+// rateLimited wraps a handler with a per-address limiter; exhausted callers
+// receive 429 with a Retry-After header.
+func rateLimited(limiter *ipLimiter, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !limiter.Allow(r) {
+			w.Header().Set("Retry-After", "60")
+			writeJSON(w, http.StatusTooManyRequests, map[string]string{
+				"error":      "too many requests, try again shortly",
+				"request_id": requestIDFrom(r.Context()),
+			})
+			return
+		}
+		next.ServeHTTP(w, r)
+	})
+}

coordinator/internal/transport/http/ratelimit_test.go

@@ -0,0 +1,54 @@
+package http
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestTokenBucketBurstThenThrottles(t *testing.T) {
+	bucket := newTokenBucket(10, 3)
+	for i := 0; i < 3; i++ {
+		if !bucket.allow() {
+			t.Fatalf("request %d must pass within the burst", i)
+		}
+	}
+	if bucket.allow() {
+		t.Error("fourth request within the burst must be throttled")
+	}
+}
+
+func TestRateLimitedReturns429(t *testing.T) {
+	limiter := newIPLimiter(10, 2)
+	next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+	handler := rateLimited(limiter, next)
+
+	req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/ui/login", nil)
+	req.RemoteAddr = "10.0.0.5:5555"
+	// Burst is 2: the first two pass, the third is throttled.
+	for i := 0; i < 2; i++ {
+		rec := httptest.NewRecorder()
+		handler.ServeHTTP(rec, req)
+		if rec.Code != http.StatusOK {
+			t.Fatalf("request %d: got %d", i, rec.Code)
+		}
+	}
+	rec := httptest.NewRecorder()
+	handler.ServeHTTP(rec, req)
+	if rec.Code != http.StatusTooManyRequests {
+		t.Fatalf("third request: got %d, want 429", rec.Code)
+	}
+	if rec.Header().Get("Retry-After") == "" {
+		t.Error("429 must carry Retry-After")
+	}
+	// A different address is not throttled by the same bucket.
+	other := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/ui/login", nil)
+	other.RemoteAddr = "10.0.0.6:5555"
+	rec = httptest.NewRecorder()
+	handler.ServeHTTP(rec, other)
+	if rec.Code != http.StatusOK {
+		t.Errorf("other client: got %d, want 200", rec.Code)
+	}
+}

coordinator/internal/transport/http/server.go

@@ -135,7 +135,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
 	// Worker-key exchange is fronted by the coordinator when the userservice
 	// is embedded (serve mode): the key itself is the credential.
 	if s.userserviceURL != "" {
-		mux.HandleFunc("POST /worker-tokens/exchange", s.handleWorkerTokenExchangeProxy)
+		mux.Handle("POST /worker-tokens/exchange", rateLimited(newIPLimiter(exchangeRatePerMinute, exchangeBurst), http.HandlerFunc(s.handleWorkerTokenExchangeProxy)))
 	}
 
 	hasBasicAuth := len(uiToken) > 0 && uiToken[0] != ""
@@ -163,7 +163,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
 		if s.uiSessionMode() {
 			// Public auth pages — reachable without a session so a user can log in.
 			ui.HandleFunc("GET /ui/login", s.handleUILoginForm)
-			ui.HandleFunc("POST /ui/login", s.handleUILogin)
+			ui.Handle("POST /ui/login", rateLimited(newIPLimiter(loginRatePerMinute, loginBurst), http.HandlerFunc(s.handleUILogin)))
 			ui.HandleFunc("GET /ui/logout-form", s.handleUILogoutForm)
 			ui.HandleFunc("GET /ui/register", s.handleUIRegisterForm)
 			ui.HandleFunc("POST /ui/register", s.handleUIRegister)

coordinator/internal/userservice/storage/sqlite/sqlite.go

@@ -7,6 +7,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"os"
 	"time"
 
 	_ "modernc.org/sqlite"
@@ -290,6 +291,17 @@ func indexOf(haystack, needle string) int {
 }
 
 // Open opens (and creates when missing) the userservice database file.
+func lockDownDatabase(path string) error {
+	for _, candidate := range []string{path, path + "-wal", path + "-shm"} {
+		if info, err := os.Stat(candidate); err == nil && !info.IsDir() {
+			if err := os.Chmod(candidate, 0o600); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 func Open(path string) (*sql.DB, error) {
 	dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(10000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)&_pragma=synchronous(NORMAL)", path)
 	db, err := sql.Open("sqlite", dsn)
@@ -300,5 +312,9 @@ func Open(path string) (*sql.DB, error) {
 		_ = db.Close()
 		return nil, fmt.Errorf("ping userservice database: %w", err)
 	}
+	if err := lockDownDatabase(path); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("lock down userservice database: %w", err)
+	}
 	return db, nil
 }

coordinator/internal/userservice/transport/http/ratelimit.go

@@ -0,0 +1,97 @@
+package http
+
+import (
+	"net"
+	"net/http"
+	"sync"
+	"time"
+)
+
+// The same per-address token-bucket policy as the coordinator transport:
+// login is the credential brute-force surface, the exchange the only public
+// token-minting one.
+const (
+	loginRatePerMinute    = 10
+	loginBurst            = 5
+	exchangeRatePerMinute = 30
+	exchangeBurst         = 10
+)
+
+type tokenBucket struct {
+	mu     sync.Mutex
+	tokens float64
+	last   time.Time
+	rate   float64
+	burst  float64
+}
+
+func newTokenBucket(ratePerMinute, burst float64) *tokenBucket {
+	return &tokenBucket{tokens: burst, last: time.Now(), rate: ratePerMinute / 60, burst: burst}
+}
+
+func (b *tokenBucket) allow() bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	now := time.Now()
+	b.tokens += now.Sub(b.last).Seconds() * b.rate
+	if b.tokens > b.burst {
+		b.tokens = b.burst
+	}
+	b.last = now
+	if b.tokens < 1 {
+		return false
+	}
+	b.tokens--
+	return true
+}
+
+type ipLimiter struct {
+	mu            sync.Mutex
+	buckets       map[string]*tokenBucket
+	ratePerMinute float64
+	burst         float64
+}
+
+func newIPLimiter(ratePerMinute, burst float64) *ipLimiter {
+	return &ipLimiter{buckets: map[string]*tokenBucket{}, ratePerMinute: ratePerMinute, burst: burst}
+}
+
+func (l *ipLimiter) Allow(r *http.Request) bool {
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		host = r.RemoteAddr
+	}
+	l.mu.Lock()
+	if len(l.buckets) > 1024 {
+		cutoff := time.Now().Add(-10 * time.Minute)
+		for addr, bucket := range l.buckets {
+			bucket.mu.Lock()
+			idle := bucket.last.Before(cutoff)
+			bucket.mu.Unlock()
+			if idle {
+				delete(l.buckets, addr)
+			}
+		}
+	}
+	bucket, ok := l.buckets[host]
+	if !ok {
+		bucket = newTokenBucket(l.ratePerMinute, l.burst)
+		l.buckets[host] = bucket
+	}
+	l.mu.Unlock()
+	return bucket.allow()
+}
+
+func rateLimited(limiter *ipLimiter, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !limiter.Allow(r) {
+			w.Header().Set("Retry-After", "60")
+			writeJSON(w, http.StatusTooManyRequests, errorResponse{
+				Error:     "too many requests, try again shortly",
+				RequestID: requestIDFrom(r.Context()),
+			})
+			return
+		}
+		next.ServeHTTP(w, r)
+	})
+}

coordinator/internal/userservice/transport/http/server.go

@@ -51,14 +51,14 @@ func NewServer(log *slog.Logger, uc UseCases, issuer auth.Issuer) http.Handler {
 	// Method-aware patterns (Go 1.22+): a GET to /register is a 405, not a match.
 	mux.HandleFunc("GET /health", h.handleHealth)
 	mux.HandleFunc("POST /register", h.handleRegister)
-	mux.HandleFunc("POST /login", h.handleLogin)
+	mux.Handle("POST /login", rateLimited(newIPLimiter(loginRatePerMinute, loginBurst), http.HandlerFunc(h.handleLogin)))
 	// /me proves a token round-trips; it sits behind JWT auth.
 	mux.Handle("GET /me", chain(http.HandlerFunc(h.handleMe), withJWT(issuer)))
 
 	// Worker keys: a user mints a long-lived key (JWT-protected), and a worker
 	// trades it for a short-lived JWT on the public exchange endpoint — the key
 	// itself is the credential there, so no prior token is required.
-	mux.HandleFunc("POST /worker-tokens/exchange", h.handleExchangeWorkerKey)
+	mux.Handle("POST /worker-tokens/exchange", rateLimited(newIPLimiter(exchangeRatePerMinute, exchangeBurst), http.HandlerFunc(h.handleExchangeWorkerKey)))
 	mux.Handle("POST /worker-keys", chain(http.HandlerFunc(h.handleCreateWorkerKey), withJWT(issuer)))
 	mux.Handle("GET /worker-keys", chain(http.HandlerFunc(h.handleListWorkerKeys), withJWT(issuer)))
 	mux.Handle("DELETE /worker-keys/{id}", chain(http.HandlerFunc(h.handleRevokeWorkerKey), withJWT(issuer)))