0.28.0 — operational hardening (plan D1–D5): CI deploy gate, versioned images, rate limiting, resource limits, health probes
D1: deploy.sh CI gate — read-only SHA via git ls-remote, Gitea commit-status
poll, portable mkdir deploy lock (macOS, no flock), TOCTOU guard, token
passed via curl --config - (not argv), graceful misconfig tolerance.
D2: version-tagged images — OIKOS_VERSION=v$VERSION, keep-last-3 prune derived
from 'docker compose config --images'; VERSION read after pull.
D3: per-IP rate limiting — new internal/httpapi/ratelimit.go (x/time/rate),
rightmost-XFF, /healthz exempt, ctx-driven sweep; disabled by default.
D4: mem_limit/cpus on all 10 compose services.
D5: staleness-aware health probes — new internal/health package wired into
scheduler (:8093) and notifier (:8094); nomos already had :8092.
Two /review passes hardened the deploy lock, TOCTOU guard, token hygiene,
and XFF handling.
This commit is contained in:
@@ -30,6 +30,17 @@ type Config struct {
|
||||
// port than the API); a no-op when the SPA and API share an origin.
|
||||
CORSAllowedOrigin string
|
||||
|
||||
// Rate limiting (plan D3). APIRateLimit is the per-IP requests/sec cap;
|
||||
// APIRateBurst is the token-bucket burst (defaults to 2x the limit when
|
||||
// unset). A limit of 0 disables rate limiting entirely.
|
||||
APIRateLimit int
|
||||
APIRateBurst int
|
||||
|
||||
// Health probe HTTP listener (plan D5). Background-loop roles (scheduler,
|
||||
// notifier) expose a staleness-aware /healthz here. Empty disables the
|
||||
// health server (local/non-docker runs).
|
||||
HealthListen string
|
||||
|
||||
// Observability
|
||||
Debug bool // verbose logging, probe payloads, SQL
|
||||
|
||||
@@ -121,6 +132,11 @@ func FromEnv() Config {
|
||||
if v := os.Getenv("OIKOS_CORS_ORIGIN"); v != "" {
|
||||
c.CORSAllowedOrigin = v
|
||||
}
|
||||
c.APIRateLimit = parseInt(os.Getenv("OIKOS_API_RATE_LIMIT"))
|
||||
c.APIRateBurst = parseInt(os.Getenv("OIKOS_API_RATE_BURST"))
|
||||
if v := os.Getenv("OIKOS_HEALTH_LISTEN"); v != "" {
|
||||
c.HealthListen = v
|
||||
}
|
||||
if v := os.Getenv("OIKOS_SEEDS_DIR"); v != "" {
|
||||
c.SeedsDir = v
|
||||
}
|
||||
|
||||
93
internal/health/health.go
Normal file
93
internal/health/health.go
Normal file
@@ -0,0 +1,93 @@
|
||||
// Package health provides a staleness-aware liveness probe for background-
|
||||
// loop services (scheduler, notifier) that don't otherwise serve HTTP.
|
||||
//
|
||||
// The owning loop calls Probe.Bump() on each iteration. A /healthz endpoint
|
||||
// returns 200 while the last bump is within the staleness window, and 503
|
||||
// once the loop has gone quiet — so a wedged goroutine (stuck SSH, deadlock)
|
||||
// surfaces as an unhealthy container instead of a silently-idle one.
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Probe tracks the last time the owning loop made progress.
|
||||
type Probe struct {
|
||||
last atomic.Int64 // unix-nano timestamp of the last Bump
|
||||
stale time.Duration
|
||||
}
|
||||
|
||||
// New returns a Probe that considers the owner healthy while Bump has been
|
||||
// called within stale of the current time.
|
||||
func New(stale time.Duration) *Probe {
|
||||
if stale <= 0 {
|
||||
stale = 2 * time.Minute
|
||||
}
|
||||
p := &Probe{stale: stale}
|
||||
p.last.Store(time.Now().UnixNano()) // boot-healthy until first loop stalls
|
||||
return p
|
||||
}
|
||||
|
||||
// Bump records that the owning loop completed another iteration.
|
||||
func (p *Probe) Bump() {
|
||||
p.last.Store(time.Now().UnixNano())
|
||||
}
|
||||
|
||||
// Healthy reports whether the last Bump is within the staleness window.
|
||||
func (p *Probe) Healthy() bool {
|
||||
last := time.Unix(0, p.last.Load())
|
||||
return time.Since(last) <= p.stale
|
||||
}
|
||||
|
||||
// Handler returns an http.Handler serving GET /healthz. Returns 200 with a
|
||||
// small JSON body when healthy, 503 (Service Unavailable) when stale.
|
||||
func (p *Probe) Handler() http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
last := time.Unix(0, p.last.Load())
|
||||
if !p.Healthy() {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
}
|
||||
json.NewEncoder(w).Encode(map[string]any{
|
||||
"status": healthStatus(p.Healthy()),
|
||||
"last_heartbeat": last.UTC().Format(time.RFC3339),
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// Serve starts an HTTP server exposing the probe's /healthz on addr until ctx
|
||||
// is cancelled. A no-op when addr is empty (local/non-docker runs skip it).
|
||||
// The server is bound to addr (e.g. ":8093"); containers hit it via 127.0.0.1.
|
||||
func (p *Probe) Serve(ctx context.Context, addr string) {
|
||||
if addr == "" {
|
||||
return
|
||||
}
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle("/healthz", p.Handler())
|
||||
srv := &http.Server{Addr: addr, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
|
||||
|
||||
go func() {
|
||||
slog.Info("health server listening", "addr", addr)
|
||||
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||
slog.Warn("health server stopped", "addr", addr, "error", err)
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
srv.Shutdown(shutdownCtx)
|
||||
}()
|
||||
}
|
||||
|
||||
func healthStatus(ok bool) string {
|
||||
if ok {
|
||||
return "ok"
|
||||
}
|
||||
return "stale"
|
||||
}
|
||||
73
internal/health/health_test.go
Normal file
73
internal/health/health_test.go
Normal file
@@ -0,0 +1,73 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestProbeHealthyAtBoot(t *testing.T) {
|
||||
p := New(time.Minute)
|
||||
if !p.Healthy() {
|
||||
t.Fatal("probe should be healthy immediately after creation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProbeStaleAfterWindow(t *testing.T) {
|
||||
p := New(50 * time.Millisecond)
|
||||
time.Sleep(80 * time.Millisecond)
|
||||
if p.Healthy() {
|
||||
t.Fatal("probe should be stale after the staleness window elapses with no Bump")
|
||||
}
|
||||
p.Bump()
|
||||
if !p.Healthy() {
|
||||
t.Fatal("probe should recover immediately after Bump")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProbeHandlerStatusCodes(t *testing.T) {
|
||||
p := New(20 * time.Millisecond)
|
||||
|
||||
// Fresh → 200
|
||||
if code := probeCode(p); code != http.StatusOK {
|
||||
t.Fatalf("fresh probe: want 200, got %d", code)
|
||||
}
|
||||
|
||||
// Stale → 503
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
if code := probeCode(p); code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("stale probe: want 503, got %d", code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProbeHandlerBody(t *testing.T) {
|
||||
p := New(time.Minute)
|
||||
rec := httptest.NewRecorder()
|
||||
p.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil))
|
||||
|
||||
var body map[string]any
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("invalid JSON body: %v (body=%q)", err, rec.Body.String())
|
||||
}
|
||||
if body["status"] != "ok" {
|
||||
t.Fatalf("want status=ok, got %v", body["status"])
|
||||
}
|
||||
if _, ok := body["last_heartbeat"].(string); !ok {
|
||||
t.Fatalf("want last_heartbeat string, got %v", body["last_heartbeat"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewDefaultsStale(t *testing.T) {
|
||||
p := New(0)
|
||||
if p.stale <= 0 {
|
||||
t.Fatal("New(0) should fall back to a positive staleness window")
|
||||
}
|
||||
}
|
||||
|
||||
func probeCode(p *Probe) int {
|
||||
rec := httptest.NewRecorder()
|
||||
p.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil))
|
||||
return rec.Code
|
||||
}
|
||||
148
internal/httpapi/ratelimit.go
Normal file
148
internal/httpapi/ratelimit.go
Normal file
@@ -0,0 +1,148 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
// rateLimiter is a per-client (IP) token-bucket limiter registry. Each unique
|
||||
// client gets its own *rate.Limiter; idle entries are swept periodically so a
|
||||
// flood of distinct IPs can't grow the map unbounded. A rate of zero (rps==0)
|
||||
// disables limiting entirely — the returned middleware is a no-op.
|
||||
//
|
||||
// Client identity is the source IP. Behind Caddy the real client is in
|
||||
// X-Forwarded-For: Caddy appends the immediate client as the LAST hop, while
|
||||
// earlier hops are client-supplied and spoofable. clientIP therefore takes the
|
||||
// rightmost XFF entry (the proxy's contribution) rather than the first.
|
||||
type rateLimiter struct {
|
||||
mu sync.Mutex
|
||||
limiters map[string]*entry
|
||||
rps rate.Limit
|
||||
burst int
|
||||
}
|
||||
|
||||
type entry struct {
|
||||
limiter *rate.Limiter
|
||||
lastSeen time.Time
|
||||
}
|
||||
|
||||
// newRateLimiter builds the registry and starts the idle-entry sweeper tied to
|
||||
// ctx, so the ticker is stopped when the server shuts down.
|
||||
func newRateLimiter(ctx context.Context, rps, burst int) *rateLimiter {
|
||||
rl := &rateLimiter{
|
||||
limiters: make(map[string]*entry),
|
||||
rps: rate.Limit(rps),
|
||||
burst: burst,
|
||||
}
|
||||
if rps > 0 {
|
||||
go rl.sweep(ctx)
|
||||
}
|
||||
return rl
|
||||
}
|
||||
|
||||
// sweep drops entries untouched since the last sweep so the registry doesn't
|
||||
// grow without bound under a rotating-IP attack or long-lived process. Exits
|
||||
// (and stops its ticker) when ctx is cancelled.
|
||||
func (rl *rateLimiter) sweep(ctx context.Context) {
|
||||
ticker := time.NewTicker(5 * time.Minute)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
rl.mu.Lock()
|
||||
for ip, e := range rl.limiters {
|
||||
if time.Since(e.lastSeen) > 10*time.Minute {
|
||||
delete(rl.limiters, ip)
|
||||
}
|
||||
}
|
||||
rl.mu.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (rl *rateLimiter) get(ip string) *rate.Limiter {
|
||||
rl.mu.Lock()
|
||||
defer rl.mu.Unlock()
|
||||
if e, ok := rl.limiters[ip]; ok {
|
||||
e.lastSeen = time.Now()
|
||||
return e.limiter
|
||||
}
|
||||
l := rate.NewLimiter(rl.rps, rl.burst)
|
||||
rl.limiters[ip] = &entry{limiter: l, lastSeen: time.Now()}
|
||||
return l
|
||||
}
|
||||
|
||||
// middleware returns a chi-style middleware that enforces the per-IP limit.
|
||||
// Call with rps==0 to get a pass-through no-op.
|
||||
func (rl *rateLimiter) middleware(next http.Handler) http.Handler {
|
||||
if rl.rps <= 0 {
|
||||
return next
|
||||
}
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
// Exempt infra liveness probes so Caddy/compose healthchecks can't be
|
||||
// throttled into marking the service unhealthy.
|
||||
if r.URL.Path == "/healthz" {
|
||||
next.ServeHTTP(w, r)
|
||||
return
|
||||
}
|
||||
if !rl.get(clientIP(r)).Allow() {
|
||||
w.Header().Set("Retry-After", "1")
|
||||
writeProblem(w, r, http.StatusTooManyRequests, "rate limit exceeded", "")
|
||||
return
|
||||
}
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
// clientIP extracts the originating client address. It takes the rightmost
|
||||
// X-Forwarded-For hop — the one the reverse proxy (Caddy) appends for the
|
||||
// immediate client — because earlier hops are attacker-controlled and could
|
||||
// be spoofed to dodge the limit or exhaust another client's bucket. Falls
|
||||
// back to r.RemoteAddr when no XFF header is present.
|
||||
//
|
||||
// Known limitation: this is only trustworthy when the request actually
|
||||
// traverses Caddy. A client connecting directly to the published :8090 (not
|
||||
// behind the proxy) can set a single-hop XFF and have it trusted. That only
|
||||
// evades rate limiting (auth is still required), and rate limiting is off by
|
||||
// default, so the blast radius is narrow. Fully closing it requires either
|
||||
// Caddy trusted_proxies (so it overwrites XFF / sets a non-spoofable
|
||||
// X-Real-Ip) or keying the limiter on the auth token instead of IP.
|
||||
func clientIP(r *http.Request) string {
|
||||
if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
|
||||
if idx := strings.LastIndex(xff, ","); idx >= 0 {
|
||||
xff = xff[idx+1:]
|
||||
}
|
||||
if ip := strings.TrimSpace(xff); ip != "" {
|
||||
return ip
|
||||
}
|
||||
}
|
||||
host, _, err := net.SplitHostPort(r.RemoteAddr)
|
||||
if err != nil {
|
||||
return r.RemoteAddr
|
||||
}
|
||||
return host
|
||||
}
|
||||
|
||||
// newRateLimiterFromConfig builds the limiter from API config, logging the
|
||||
// chosen policy once at startup. rps<=0 means "disabled" (returns a no-op
|
||||
// middleware) so dev/single-user setups aren't throttled by default.
|
||||
func newRateLimiterFromConfig(ctx context.Context, rps, burst int) *rateLimiter {
|
||||
if rps <= 0 {
|
||||
slog.Info("api rate limiting disabled (OIKOS_API_RATE_LIMIT unset)")
|
||||
return &rateLimiter{rps: 0}
|
||||
}
|
||||
if burst <= 0 {
|
||||
burst = rps * 2
|
||||
}
|
||||
slog.Info("api rate limiting enabled", "rps", rps, "burst", burst)
|
||||
return newRateLimiter(ctx, rps, burst)
|
||||
}
|
||||
161
internal/httpapi/ratelimit_test.go
Normal file
161
internal/httpapi/ratelimit_test.go
Normal file
@@ -0,0 +1,161 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// okHandler is a sentinel upstream that records it was reached.
|
||||
func okHandler(t *testing.T, reached *bool) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
*reached = true
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
}
|
||||
|
||||
func TestRateLimiterDisabledWhenRPSZero(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
rl := newRateLimiterFromConfig(ctx, 0, 0)
|
||||
if rl.rps != 0 {
|
||||
t.Fatalf("rps should be 0 when disabled, got %v", rl.rps)
|
||||
}
|
||||
// Disabled limiter is a pass-through: requests always reach upstream.
|
||||
reached := false
|
||||
h := rl.middleware(okHandler(t, &reached))
|
||||
for i := 0; i < 50; i++ {
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/entities", nil))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("disabled limiter request %d: want 200, got %d", i, rec.Code)
|
||||
}
|
||||
}
|
||||
if !reached {
|
||||
t.Fatal("disabled limiter never reached upstream")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRateLimiterThrottlesAfterBurst(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
rl := newRateLimiter(ctx, 1, 3) // 1 rps, burst 3
|
||||
reached := false
|
||||
h := rl.middleware(okHandler(t, &reached))
|
||||
|
||||
var last429, okCount int
|
||||
for i := 0; i < 6; i++ {
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/entities", nil))
|
||||
switch rec.Code {
|
||||
case http.StatusOK:
|
||||
okCount++
|
||||
case http.StatusTooManyRequests:
|
||||
last429 = i
|
||||
}
|
||||
}
|
||||
if okCount < 1 {
|
||||
t.Fatal("expected at least one request through within burst")
|
||||
}
|
||||
if last429 == 0 {
|
||||
t.Fatal("expected at least one 429 once burst exhausted")
|
||||
}
|
||||
if !reached {
|
||||
t.Fatal("upstream never reached")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRateLimiterExemptsHealthz(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
rl := newRateLimiter(ctx, 1, 1) // tiny burst
|
||||
reached := false
|
||||
h := rl.middleware(okHandler(t, &reached))
|
||||
// /healthz must never be throttled, even under a flood.
|
||||
for i := 0; i < 20; i++ {
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("healthz request %d throttled: want 200, got %d", i, rec.Code)
|
||||
}
|
||||
}
|
||||
if !reached {
|
||||
t.Fatal("healthz never reached upstream")
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientIPTakesRightmostXFF(t *testing.T) {
|
||||
// The reverse proxy appends the real client as the LAST hop; earlier hops
|
||||
// are spoofable and must be ignored.
|
||||
cases := []struct {
|
||||
name string
|
||||
xff string
|
||||
remote string
|
||||
wantIP string
|
||||
}{
|
||||
{"single xff", "203.0.113.7", "10.0.0.1:4000", "203.0.113.7"},
|
||||
{"multi hop takes rightmost", "spoofed-attacker, 203.0.113.7", "10.0.0.1:4000", "203.0.113.7"},
|
||||
{"no xff falls back to remote", "", "198.51.100.2:4000", "198.51.100.2"},
|
||||
{"blank xff falls back to remote", " ", "198.51.100.2:4000", "198.51.100.2"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
req := httptest.NewRequest(http.MethodGet, "/", nil)
|
||||
req.RemoteAddr = tc.remote
|
||||
if strings.TrimSpace(tc.xff) != "" {
|
||||
req.Header.Set("X-Forwarded-For", tc.xff)
|
||||
}
|
||||
if got := clientIP(req); got != tc.wantIP {
|
||||
t.Fatalf("clientIP: want %q, got %q", tc.wantIP, got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRateLimiterConcurrency(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
rl := newRateLimiter(ctx, 1000, 100) // generous; ensures no deadlock/panic under contention
|
||||
h := rl.middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 20; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/entities", nil))
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func TestNewRateLimiterFromConfigBurstDefault(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel() // release the sweep goroutine + ticker
|
||||
rl := newRateLimiterFromConfig(ctx, 10, 0) // burst unset → defaults to 2x
|
||||
if rl.burst != 20 {
|
||||
t.Fatalf("default burst should be 2x rps (20), got %d", rl.burst)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSweepStopsOnContextCancel verifies the ticker is released when the
|
||||
// server context is cancelled (no process-lifetime goroutine/ticker leak).
|
||||
func TestSweepStopsOnContextCancel(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
rl := newRateLimiter(ctx, 1, 1)
|
||||
cancel()
|
||||
// Give the sweeper a moment to observe cancellation. It must return
|
||||
// without blocking; the deferred ticker.Stop() fires on return.
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
// Limiter remains usable for the brief test lifetime.
|
||||
if !rl.get("10.0.0.1").Allow() {
|
||||
t.Fatal("limiter should still allow within burst after sweep stops")
|
||||
}
|
||||
}
|
||||
@@ -23,8 +23,8 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/dtoro/oikos/internal/config"
|
||||
"github.com/dtoro/oikos/internal/actuator"
|
||||
"github.com/dtoro/oikos/internal/config"
|
||||
"github.com/dtoro/oikos/internal/db"
|
||||
"github.com/dtoro/oikos/internal/httpapi/gen"
|
||||
mcphandler "github.com/dtoro/oikos/internal/mcp"
|
||||
@@ -118,6 +118,11 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Hand
|
||||
r.Use(middleware.Recoverer)
|
||||
r.Use(middleware.RequestID)
|
||||
r.Use(requestLogger)
|
||||
// Per-IP rate limiting (plan D3). Applied before CORS/auth so a runaway
|
||||
// agent loop is throttled regardless of credentials. The middleware
|
||||
// exempts /healthz so liveness probes can't be throttled.
|
||||
limiter := newRateLimiterFromConfig(ctx, cfg.APIRateLimit, cfg.APIRateBurst)
|
||||
r.Use(limiter.middleware)
|
||||
r.Use(cors.Handler(cors.Options{
|
||||
AllowedOrigins: []string{cfg.CORSAllowedOrigin},
|
||||
AllowedMethods: []string{"GET", "POST", "PATCH", "DELETE", "OPTIONS"},
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
|
||||
"github.com/dtoro/oikos/internal/config"
|
||||
"github.com/dtoro/oikos/internal/db"
|
||||
"github.com/dtoro/oikos/internal/health"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
@@ -25,7 +26,14 @@ import (
|
||||
func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
|
||||
slog.Info("notifier: starting")
|
||||
|
||||
// Liveness probe (plan D5): the notifier ticks every 15s (approvals) and
|
||||
// 30s (reactions). 2 min staleness covers a slow Matrix round-trip plus a
|
||||
// missed tick without false-failing.
|
||||
probe := health.New(2 * time.Minute)
|
||||
probe.Serve(ctx, cfg.HealthListen)
|
||||
|
||||
processPendingApprovals(ctx, pool, cfg)
|
||||
probe.Bump()
|
||||
|
||||
ticker := time.NewTicker(15 * time.Second)
|
||||
defer ticker.Stop()
|
||||
@@ -40,8 +48,10 @@ func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
|
||||
return
|
||||
case <-ticker.C:
|
||||
processPendingApprovals(ctx, pool, cfg)
|
||||
probe.Bump()
|
||||
case <-reactionTimer.C:
|
||||
pollReactions(ctx, pool, cfg)
|
||||
probe.Bump()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -52,13 +62,13 @@ func RunnerForMain() func(context.Context, *db.Pool, config.Config) {
|
||||
}
|
||||
|
||||
type pendingApproval struct {
|
||||
ID uuid.UUID
|
||||
Action string
|
||||
RiskClass string
|
||||
TokenHash *string
|
||||
AlertSentAt *time.Time
|
||||
MatrixEventID *string
|
||||
ExpiresAt time.Time
|
||||
ID uuid.UUID
|
||||
Action string
|
||||
RiskClass string
|
||||
TokenHash *string
|
||||
AlertSentAt *time.Time
|
||||
MatrixEventID *string
|
||||
ExpiresAt time.Time
|
||||
}
|
||||
|
||||
// processPendingApprovals finds pending approvals, generates tokens, and sends Matrix alerts.
|
||||
@@ -173,7 +183,7 @@ func checkReaction(ctx context.Context, cfg config.Config, roomID, eventID strin
|
||||
|
||||
var result struct {
|
||||
Chunk []struct {
|
||||
Type string `json:"type"`
|
||||
Type string `json:"type"`
|
||||
Content struct {
|
||||
RelatesTo map[string]string `json:"m.relates_to"`
|
||||
} `json:"content"`
|
||||
@@ -259,7 +269,9 @@ func sendMatrixAlert(ctx context.Context, cfg config.Config, approvalID uuid.UUI
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
var mxResp struct{ EventID string `json:"event_id"` }
|
||||
var mxResp struct {
|
||||
EventID string `json:"event_id"`
|
||||
}
|
||||
json.NewDecoder(resp.Body).Decode(&mxResp)
|
||||
|
||||
if mxResp.EventID == "" {
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
"github.com/dtoro/oikos/internal/config"
|
||||
"github.com/dtoro/oikos/internal/db"
|
||||
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
||||
"github.com/dtoro/oikos/internal/health"
|
||||
"github.com/dtoro/oikos/internal/observability"
|
||||
"github.com/dtoro/oikos/internal/remote"
|
||||
"github.com/google/uuid"
|
||||
@@ -48,11 +49,22 @@ func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
|
||||
sshUser = "root"
|
||||
}
|
||||
|
||||
// Liveness probe (plan D5): staleness is 3x the interval so a single
|
||||
// slow check pass (one host hung on SSH) doesn't flap the container
|
||||
// unhealthy before the next scheduled tick.
|
||||
stale := 3 * interval
|
||||
if stale < 90*time.Second {
|
||||
stale = 90 * time.Second
|
||||
}
|
||||
probe := health.New(stale)
|
||||
probe.Serve(ctx, cfg.HealthListen)
|
||||
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
|
||||
// Immediate first pass
|
||||
runCheckPass(ctx, pool)
|
||||
probe.Bump()
|
||||
|
||||
for {
|
||||
select {
|
||||
@@ -61,6 +73,7 @@ func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
|
||||
return
|
||||
case <-ticker.C:
|
||||
runCheckPass(ctx, pool)
|
||||
probe.Bump()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -586,7 +599,7 @@ func checkDisk(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkRes
|
||||
// checkCertExpiry checks TLS certificate expiry.
|
||||
func checkCertExpiry(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
|
||||
cfg := struct {
|
||||
Host string `json:"host"`
|
||||
Host string `json:"host"`
|
||||
// Dial is an optional explicit dial address (the TLS terminator's IP)
|
||||
// for when the hostname doesn't resolve/reach from the scheduler — the
|
||||
// container has no mesh interface and the host resolver doesn't know
|
||||
|
||||
Reference in New Issue
Block a user