fix UI 503s under checkpoint load

- Store: skip identical CT checkpoint writes (CertSpotter calls StoreLogState
  per batch across 64 logs; hashing the state avoids redundant writes).
- webui: cache /api/stats for 10s and raise API query timeouts; UI polls, so
  repeated aggregate queries no longer queue behind checkpoint bursts.
- reword discovery detail (no 'name enumeration' for S3/GCS; only Azure uses a
  container list).
This commit is contained in:
StasonJatham committed 2026-09-22 17:45:18 +02:00
1 parent 32035d0b9d
commit 1ce5700cec
6 files changed
+50 -14

No files matched your search

Binary file not shown.

After

Width:  |  Height:  |  Size: 130 KiB

+2 -2
View File
@@ -231,7 +231,7 @@ func (d *Discoverer) scanAzure(ctx context.Context, account string) {
Provider: model.ProviderAzure, Service: model.ServiceBlobStorage, Provider: model.ProviderAzure, Service: model.ServiceBlobStorage,
ResourceSpecific: true, Source: model.SourceDiscovery, ResourceSpecific: true, Source: model.SourceDiscovery,
} }
d.cfg.OnOpen(ctx, c, "azure public container(s): "+strings.Join(found, ","), entries) d.cfg.OnOpen(ctx, c, "azure anonymous listing allowed (container: "+strings.Join(found, ",")+")", entries)
} }
// validAzureAccount mirrors Azure's naming rules: 3-24 lowercase alphanumerics. // validAzureAccount mirrors Azure's naming rules: 3-24 lowercase alphanumerics.
@@ -299,7 +299,7 @@ func (d *Discoverer) test(ctx context.Context, t Target) {
if finalHost != "" && strings.HasSuffix(finalHost, ".amazonaws.com") { if finalHost != "" && strings.HasSuffix(finalHost, ".amazonaws.com") {
c.Hostname = finalHost c.Hostname = finalHost
} }
detail := "discovered open bucket via name enumeration (" + t.Hostname + ")" detail := "anonymous listing allowed"
if d.cfg.Metrics != nil { if d.cfg.Metrics != nil {
d.cfg.Metrics.BucketOpen.Add(1) d.cfg.Metrics.BucketOpen.Add(1)
} }
+1 -1
View File
@@ -90,7 +90,7 @@ func (m *Metrics) Handler() http.Handler {
counter("probe_public_total", "Probes classified as public read", m.ProbePublic.Load()) counter("probe_public_total", "Probes classified as public read", m.ProbePublic.Load())
counter("entries_indexed_total", "Metadata entries indexed from readable resources", m.EntriesIndexed.Load()) counter("entries_indexed_total", "Metadata entries indexed from readable resources", m.EntriesIndexed.Load())
counter("bucket_candidates_total", "Candidate bucket names tested", m.BucketCandidates.Load()) counter("bucket_candidates_total", "Candidate bucket names tested", m.BucketCandidates.Load())
counter("bucket_open_total", "Open (listable) buckets discovered by name enumeration", m.BucketOpen.Load()) counter("bucket_open_total", "Open (listable) buckets discovered from CT hostname labels", m.BucketOpen.Load())
for _, g := range m.gauges { for _, g := range m.gauges {
fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s gauge\n%s%s %g\n", g.name, g.help, g.name, g.name, g.extra, g.fn()) fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s gauge\n%s%s %g\n", g.name, g.help, g.name, g.name, g.extra, g.fn())
} }
+12 -1
View File
@@ -2,6 +2,7 @@ package store
import ( import (
"context" "context"
"crypto/sha256"
"encoding/json" "encoding/json"
"fmt" "fmt"
"log" "log"
@@ -35,11 +36,21 @@ func (s *Store) StoreLogState(ctx context.Context, logID monitor.LogID, state *m
if err != nil { if err != nil {
return err return err
} }
key := logID.Base64URLString()
hash := sha256.Sum256(data)
s.stateMu.Lock()
if prev, ok := s.lastState[key]; ok && prev == hash {
s.stateMu.Unlock()
return nil // unchanged; CertSpotter calls this per batch
}
s.lastState[key] = hash
s.stateMu.Unlock()
_, err = s.pool.Exec(ctx, ` _, err = s.pool.Exec(ctx, `
INSERT INTO ct_checkpoints (log_id, state, updated_at) INSERT INTO ct_checkpoints (log_id, state, updated_at)
VALUES ($1, $2, now()) VALUES ($1, $2, now())
ON CONFLICT (log_id) DO UPDATE SET state = EXCLUDED.state, updated_at = now()`, ON CONFLICT (log_id) DO UPDATE SET state = EXCLUDED.state, updated_at = now()`,
logID.Base64URLString(), data) key, data)
return err return err
} }
+5 -1
View File
@@ -5,6 +5,7 @@ package store
import ( import (
"context" "context"
"fmt" "fmt"
"sync"
"time" "time"
"github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5"
@@ -20,6 +21,9 @@ import (
type Store struct { type Store struct {
pool *pgxpool.Pool pool *pgxpool.Pool
metrics *metrics.Metrics metrics *metrics.Metrics
stateMu sync.Mutex
lastState map[string][32]byte // per-log checkpoint hash; skip identical writes
} }
// Open connects to PostgreSQL. The pool is bounded; the caller configures the // Open connects to PostgreSQL. The pool is bounded; the caller configures the
@@ -38,7 +42,7 @@ func Open(ctx context.Context, databaseURL string, maxConns int) (*Store, error)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return &Store{pool: pool}, nil return &Store{pool: pool, lastState: make(map[string][32]byte)}, nil
} }
func (s *Store) SetMetrics(m *metrics.Metrics) { s.metrics = m } func (s *Store) SetMetrics(m *metrics.Metrics) { s.metrics = m }
+30 -9
View File
@@ -12,6 +12,7 @@ import (
"net/http" "net/http"
"strconv" "strconv"
"strings" "strings"
"sync"
"sync/atomic" "sync/atomic"
"time" "time"
@@ -73,14 +74,16 @@ func New(s Store, cfg Config) http.Handler {
Limit: limit, Limit: limit,
Offset: offset, Offset: offset,
} }
assets, err := s.ListAssets(r.Context(), f) ctx, cancel := context.WithTimeout(r.Context(), 9*time.Second)
defer cancel()
assets, err := s.ListAssets(ctx, f)
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusServiceUnavailable)
return return
} }
total, err := s.CountAssets(r.Context(), f) total, err := s.CountAssets(ctx, f)
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusServiceUnavailable)
return return
} }
writeJSON(w, map[string]any{"assets": assets, "total": total, "limit": limit, "offset": offset}) writeJSON(w, map[string]any{"assets": assets, "total": total, "limit": limit, "offset": offset})
@@ -89,9 +92,11 @@ func New(s Store, cfg Config) http.Handler {
q := r.URL.Query() q := r.URL.Query()
limit, _ := strconv.Atoi(q.Get("limit")) limit, _ := strconv.Atoi(q.Get("limit"))
offset, _ := strconv.Atoi(q.Get("offset")) offset, _ := strconv.Atoi(q.Get("offset"))
entries, total, err := s.SearchEntries(r.Context(), strings.TrimSpace(q.Get("q")), limit, offset) ctx, cancel := context.WithTimeout(r.Context(), 9*time.Second)
defer cancel()
entries, total, err := s.SearchEntries(ctx, strings.TrimSpace(q.Get("q")), limit, offset)
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusServiceUnavailable)
return return
} }
writeJSON(w, map[string]any{"entries": entries, "total": total, "limit": limit, "offset": offset}) writeJSON(w, map[string]any{"entries": entries, "total": total, "limit": limit, "offset": offset})
@@ -143,12 +148,28 @@ func New(s Store, cfg Config) http.Handler {
writeJSON(w, map[string]any{"started": true, "exposure": exposure, "limit": limit}) writeJSON(w, map[string]any{"started": true, "exposure": exposure, "limit": limit})
}) })
} }
var (
statsMu sync.Mutex
statsVal store.Stats
statsExp time.Time
)
mux.HandleFunc("/api/stats", func(w http.ResponseWriter, r *http.Request) { mux.HandleFunc("/api/stats", func(w http.ResponseWriter, r *http.Request) {
st, err := s.Stats(r.Context()) // Cached for a few seconds: the UI polls, and each computation is several
if err != nil { // aggregate queries that can queue behind CertSpotter checkpoint bursts.
http.Error(w, err.Error(), http.StatusInternalServerError) statsMu.Lock()
defer statsMu.Unlock()
if time.Now().Before(statsExp) {
writeJSON(w, statsVal)
return return
} }
ctx, cancel := context.WithTimeout(r.Context(), 9*time.Second)
defer cancel()
st, err := s.Stats(ctx)
if err != nil {
http.Error(w, err.Error(), http.StatusServiceUnavailable)
return
}
statsVal, statsExp = st, time.Now().Add(10*time.Second)
writeJSON(w, st) writeJSON(w, st)
}) })