mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-11 02:58:12 +00:00
Part A of #2128: optional, off-by-default user accounts. With the feature off nothing changes; with it on, visitors can register and log in, and admins manage users and use the operator actions without the API key. PR #2130 (settings sync) builds on this one. The two are meant to be merged together. ## The situation - Operator actions (geofilter save and prune, backup, perf reset) need the shared `apiKey`. There is no per-person right. - Nothing in CoreScope knows who a visitor is, so the requests in #2128 that need that (#1835, #2092, #1508, #730) have nothing to build on. ## What this PR adds **Two new Go modules** - `internal/users`: a separate `users.db` (SQLite through `modernc.org/sqlite`) with users, sessions, single-use tokens, an audit log and a mail log. Passwords use argon2id. - `internal/mailer`: a `Mailer` interface with a Brevo client (send, delivery events, webhook parsing) and an in-memory fake for tests. **Server (`cmd/server`)**, active only with `userManagement.enabled` - 24 routes, all documented in OpenAPI under the `users` tag ([`auth_routes.go`](https://github.com/efiten/CoreScope/blob/feat/user-management/cmd/server/auth_routes.go)): - auth: register, activate, login, logout, me, forgot, reset; - account: profile, password, email change with confirmation, sessions, self-delete; - admin: list, detail, disable, enable, delete, role, resend activation, manual activation, mail status refresh; - a Brevo webhook, registered only when `mail.webhookSecret` is set. - `requireAdmin` replaces `requireAPIKey` at the 7 operator call sites: the API key **or** an admin session. With the feature off it is the old API-key gate (`TestRequireAdminWithoutUserManagementIsAPIKeyGate`). - `/api/config/client` gets `userManagement: {enabled: true}` only when the service started; with the feature off the response is byte-identical. **Frontend** - `auth.js` (header account control, request helper that adds the CSRF header), `account.js` (login, register, activate, forgot, reset, confirm email, my account), `admin-users.js` (`#/admin/users`, deep-linked filters), `account.css` (theme tokens only). - On phones the top-bar control is hidden, so a conditional entry goes into the bottom-nav "More" sheet and the nav drawer. - The customizer geofilter tab and the Perf "Reset stats" button use the admin session when there is one. **Config.** A `userManagement` block (`config.example.json`, [`docs/user-guide/accounts.md`](https://github.com/efiten/CoreScope/blob/feat/user-management/docs/user-guide/accounts.md)). The Brevo key can come from `CORESCOPE_BREVO_API_KEY`. The server refuses to start when the block is enabled but incomplete. ## Security choices - Session cookie `cs_session`: HttpOnly, SameSite=Lax, Secure when `publicBaseUrl` is https. Every cookie-authenticated state change needs the `X-CS-CSRF` header and a matching Origin. - Activation needs the token **and** the account password. Without the password, an attacker who keeps re-registering a known address could get the owner to activate an account that carries the attacker's password. - Register, forgot and email change answer identically for known and unknown addresses. A password reset ends all sessions, a password change ends all other sessions, and both end outstanding email-change links. - Rate limits: login 10 per 15 minutes, register and forgot 5 per hour, per IP and per address. The bucket count is capped. `trustedProxies` makes the per-IP limits see real client IPs behind a proxy. - Server logs carry `#<user id>`, never addresses, tokens or passwords; mail-provider error texts are redacted before logging. ## Performance No change to an existing hot path with the feature off. With it on: - One `users.db` lookup per authenticated request (session by token hash). - The admin user table rebuilds its `tbody` on each filter change. `users.List` caps the result at 1000 rows (`internal/users/users.go`), which bounds the rebuild. - `map[string]interface{}` in `openapi.go`: 79 before, 78 after. ## Verification - `internal/users`, `internal/mailer` and `cmd/server`: `go vet` and `go test -race` pass locally. 121 new Go tests. - `cmd/server` with `-tags e2etest`: vet and the e2e hook tests pass. - `sh test-all.sh` exits 0. `tests/unit/test-user-management-ui.js`: 67 passing (vm, real modules). - `tests/e2e/test-user-management-e2e.js` (6 steps) passed locally against an `e2etest` build with the fake mailer and against a feature-off build. CI builds the `e2etest` binary and runs the suite on a second server (`deploy.yml`). - On a staging instance with a real Brevo key: register, activation mail delivered, activate, admin table, "Refresh status" showing sent, deferred, delivered, opened and clicked. ## Not in this PR - Settings sync (#2130), the admin dashboard, approval flows and notifications (parts B to E of #2128). - A `requireReadAuth` mode (#1835). Sessions from this PR are what such a mode would accept. - Binary size and build time with `modernc.org/sqlite` linked next to `mattn/go-sqlite3` were not measured. Their driver names do not collide. #1992 discusses the driver choice. - No Brevo webhook was configured on staging; delivery status there came from "Refresh status". --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
652 lines
24 KiB
Go
652 lines
24 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"flag"
|
||
"fmt"
|
||
"log"
|
||
"net/http"
|
||
_ "net/http/pprof"
|
||
"os"
|
||
"os/exec"
|
||
"os/signal"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"syscall"
|
||
"time"
|
||
|
||
"github.com/gorilla/mux"
|
||
"github.com/meshcore-analyzer/dbschema"
|
||
)
|
||
|
||
// Set via -ldflags at build time
|
||
var Version string
|
||
var Commit string
|
||
var BuildTime string
|
||
|
||
func resolveCommit() string {
|
||
if Commit != "" {
|
||
return Commit
|
||
}
|
||
// Try .git-commit file (baked by Docker / CI)
|
||
if data, err := os.ReadFile(".git-commit"); err == nil {
|
||
if c := strings.TrimSpace(string(data)); c != "" && c != "unknown" {
|
||
return c
|
||
}
|
||
}
|
||
// Try git rev-parse at runtime
|
||
if out, err := exec.Command("git", "rev-parse", "--short", "HEAD").Output(); err == nil {
|
||
return strings.TrimSpace(string(out))
|
||
}
|
||
return "unknown"
|
||
}
|
||
|
||
func resolveVersion() string {
|
||
// Issue #1807: the release fast-path re-tags :edge -> :vX.Y.Z without a
|
||
// rebuild, so the ldflags-baked Version still reports "edge" on tagged
|
||
// releases. Resolution order: operator env override, then the
|
||
// .image-version file appended by the fast-path retag (mirrors the
|
||
// .git-commit pattern in resolveCommit), then the baked Version.
|
||
if v := strings.TrimSpace(os.Getenv("CORESCOPE_VERSION")); v != "" {
|
||
return v
|
||
}
|
||
if data, err := os.ReadFile(".image-version"); err == nil {
|
||
if v := strings.TrimSpace(string(data)); v != "" && v != "unknown" {
|
||
return v
|
||
}
|
||
}
|
||
if Version != "" {
|
||
return Version
|
||
}
|
||
return "unknown"
|
||
}
|
||
|
||
func resolveBuildTime() string {
|
||
if BuildTime != "" {
|
||
return BuildTime
|
||
}
|
||
return "unknown"
|
||
}
|
||
|
||
func main() {
|
||
// pprof profiling — off by default, enable with ENABLE_PPROF=true
|
||
if os.Getenv("ENABLE_PPROF") == "true" {
|
||
pprofPort := os.Getenv("PPROF_PORT")
|
||
if pprofPort == "" {
|
||
pprofPort = "6060"
|
||
}
|
||
go func() {
|
||
log.Printf("[pprof] profiling UI at http://localhost:%s/debug/pprof/", pprofPort)
|
||
if err := http.ListenAndServe(":"+pprofPort, nil); err != nil {
|
||
log.Printf("[pprof] failed to start: %v (non-fatal)", err)
|
||
}
|
||
}()
|
||
}
|
||
|
||
var (
|
||
configDir string
|
||
port int
|
||
dbPath string
|
||
publicDir string
|
||
pollMs int
|
||
)
|
||
|
||
flag.StringVar(&configDir, "config-dir", ".", "Directory containing config.json")
|
||
flag.IntVar(&port, "port", 0, "HTTP port (overrides config)")
|
||
flag.StringVar(&dbPath, "db", "", "SQLite database path (overrides config/env)")
|
||
flag.StringVar(&publicDir, "public", "public", "Directory to serve static files from")
|
||
flag.IntVar(&pollMs, "poll-ms", 1000, "SQLite poll interval for WebSocket broadcast (ms)")
|
||
flag.Parse()
|
||
|
||
// Load config
|
||
cfg, err := LoadConfig(configDir)
|
||
if err != nil {
|
||
log.Printf("[config] warning: %v (using defaults)", err)
|
||
}
|
||
|
||
// CLI flags override config
|
||
if port > 0 {
|
||
cfg.Port = port
|
||
}
|
||
if cfg.Port == 0 {
|
||
cfg.Port = 3000
|
||
}
|
||
if dbPath != "" {
|
||
cfg.DBPath = dbPath
|
||
}
|
||
if cfg.APIKey == "" {
|
||
log.Printf("[security] WARNING: no apiKey configured — write endpoints are BLOCKED (set apiKey in config.json to enable them)")
|
||
} else if IsWeakAPIKey(cfg.APIKey) {
|
||
log.Printf("[security] WARNING: API key is weak or a known default — write endpoints are vulnerable")
|
||
}
|
||
|
||
// Apply Go runtime soft memory limit (#836, #1010).
|
||
// Precedence: GOMEMLIMIT env > runtime.maxMemoryMB > derived from packetStore.maxMemoryMB.
|
||
{
|
||
_, envSet := os.LookupEnv("GOMEMLIMIT")
|
||
runtimeMaxMB := 0
|
||
if cfg.Runtime != nil {
|
||
runtimeMaxMB = cfg.Runtime.MaxMemoryMB
|
||
}
|
||
maxMB := 0
|
||
if cfg.PacketStore != nil {
|
||
maxMB = cfg.PacketStore.MaxMemoryMB
|
||
}
|
||
// runtime.maxMemoryMB (explicit) wins over packetStore-derived (implicit).
|
||
effectiveMB := maxMB
|
||
usedRuntimeCfg := false
|
||
if !envSet && runtimeMaxMB > 0 {
|
||
effectiveMB = runtimeMaxMB
|
||
usedRuntimeCfg = true
|
||
}
|
||
limit, source := applyMemoryLimit(effectiveMB, envSet)
|
||
switch source {
|
||
case "env":
|
||
log.Printf("[memlimit] using GOMEMLIMIT from environment (%s)", os.Getenv("GOMEMLIMIT"))
|
||
case "derived":
|
||
if usedRuntimeCfg {
|
||
log.Printf("[memlimit] runtime.maxMemoryMB=%d → %d MiB (1.5x headroom)", runtimeMaxMB, limit/(1024*1024))
|
||
} else {
|
||
log.Printf("[memlimit] derived from packetStore.maxMemoryMB=%d → %d MiB (1.5x headroom)", maxMB, limit/(1024*1024))
|
||
}
|
||
default:
|
||
log.Printf("[memlimit] unset → default (no soft memory limit; recommend setting GOMEMLIMIT or runtime.maxMemoryMB to ≥1.5× working set to avoid OOM-kill)")
|
||
}
|
||
warnIfMemlimitUnderprovisioned(limit)
|
||
}
|
||
|
||
// Resolve DB path
|
||
resolvedDB := cfg.ResolveDBPath(configDir)
|
||
log.Printf("[config] port=%d db=%s public=%s", cfg.Port, resolvedDB, publicDir)
|
||
if len(cfg.NodeBlacklist) > 0 {
|
||
log.Printf("[config] nodeBlacklist: %d node(s) will be hidden from API", len(cfg.NodeBlacklist))
|
||
for _, pk := range cfg.NodeBlacklist {
|
||
if trimmed := strings.ToLower(strings.TrimSpace(pk)); trimmed != "" {
|
||
log.Printf("[config] blacklisted: %s", trimmed)
|
||
}
|
||
}
|
||
}
|
||
|
||
// Open database
|
||
database, err := OpenDB(resolvedDB)
|
||
if err != nil {
|
||
log.Fatalf("[db] failed to open %s: %v", resolvedDB, err)
|
||
}
|
||
var dbCloseOnce sync.Once
|
||
dbClose := func() error {
|
||
var err error
|
||
dbCloseOnce.Do(func() { err = database.Close() })
|
||
return err
|
||
}
|
||
defer dbClose()
|
||
|
||
// Verify DB has expected tables
|
||
var tableName string
|
||
err = database.conn.QueryRow("SELECT name FROM sqlite_master WHERE type='table' AND name='transmissions'").Scan(&tableName)
|
||
if err == sql.ErrNoRows {
|
||
log.Fatalf("[db] table 'transmissions' not found — is this a CoreScope database?")
|
||
}
|
||
|
||
stats, err := database.GetStats()
|
||
if err != nil {
|
||
log.Printf("[db] warning: could not read stats: %v", err)
|
||
} else {
|
||
log.Printf("[db] transmissions=%d observations=%d nodes=%d observers=%d",
|
||
stats.TotalTransmissions, stats.TotalObservations, stats.TotalNodes, stats.TotalObservers)
|
||
}
|
||
|
||
// auto_vacuum is checked + migrated by the ingestor (#1283). The
|
||
// server is read-only and must not race the writer for the lock.
|
||
|
||
// Assert all schema migrations the ingestor owns have already run
|
||
// (#1287). The server NEVER migrates — it only reads. If a required
|
||
// column/index/table is missing, the operator must restart the
|
||
// ingestor (which owns dbschema.Apply) before this server can start.
|
||
if err := dbschema.AssertReady(database.conn); err != nil {
|
||
log.Fatalf("[db] schema not ready (ingestor must run migrations first): %v", err)
|
||
}
|
||
|
||
// In-memory packet store
|
||
store := NewPacketStore(database, cfg.PacketStore, cfg.CacheTTL)
|
||
store.config = cfg
|
||
|
||
// Load the persisted neighbor graph BEFORE the packet load so the
|
||
// chunked loader can resolve relay-hop pubkeys from path_json. Since
|
||
// #1287 the ingestor persists relay data only as aggregate
|
||
// neighbor_edges — observations.resolved_path is never written — so
|
||
// without an available graph at load time a relay node's analytics
|
||
// history would rebuild only from post-restart live traffic (the
|
||
// "timeline empty after every restart" bug). neighbor_edges is small,
|
||
// so this adds negligible latency before the HTTP listener binds. The
|
||
// fresh-DB branch (no snapshot) still builds in-memory AFTER the load
|
||
// below, because BuildFromStore needs the loaded packets.
|
||
neighborEdgesPersisted := neighborEdgesTableExists(database.conn)
|
||
if neighborEdgesPersisted {
|
||
store.graph.Store(loadNeighborEdgesFromDB(database.conn))
|
||
log.Printf("[neighbor] loaded persisted neighbor graph")
|
||
}
|
||
|
||
// #1009: chunked Load with early HTTP readiness. RunStartupLoad runs
|
||
// asynchronously and signals FirstChunkReady after the first chunk
|
||
// is merged so the HTTP listener can bind without waiting for the
|
||
// full multi-minute scan to finish. loadStatusMiddleware (wired
|
||
// below) advertises loading|ready via X-CoreScope-Load-Status.
|
||
//
|
||
// #1809: the background fill loader (loadBackgroundChunks) used to
|
||
// be spawned here at FirstChunkReady, but at that point LoadChunked
|
||
// has not yet set s.oldestLoaded → bg loader read "" and bailed →
|
||
// coverage gate trips. RunStartupLoad now gates the bg loader on
|
||
// LoadChunked completion, preserving FirstChunkReady's parallelism
|
||
// for the HTTP listener bind.
|
||
chunkSize := cfg.DBLoadChunkSize()
|
||
// loadErrCh is buffered(1) and is the SOLE consumer slot for the
|
||
// RunStartupLoad goroutine's terminal error. We have exactly ONE
|
||
// reader path below: either the select consumes the error (fast
|
||
// empty-DB path) OR we hand it off to a single late-reader goroutine
|
||
// (slow path: bg loader still running after FirstChunkReady fired).
|
||
// The pre-#1811 code spawned a late-reader unconditionally, which
|
||
// leaked a goroutine forever after the select had already drained
|
||
// the channel. See PR #1811 adv #1.
|
||
loadErrCh := make(chan error, 1)
|
||
go func() {
|
||
loadErrCh <- store.RunStartupLoad(chunkSize)
|
||
}()
|
||
firstChunkPath := false
|
||
select {
|
||
case <-store.FirstChunkReady():
|
||
firstChunkPath = true
|
||
log.Printf("[store] first chunk ready (chunkSize=%d) — HTTP listener may bind", chunkSize)
|
||
case err := <-loadErrCh:
|
||
if err != nil {
|
||
log.Fatalf("[store] RunStartupLoad failed before first chunk: %v", err)
|
||
}
|
||
log.Printf("[store] RunStartupLoad completed before first-chunk signal (empty DB?)")
|
||
}
|
||
if store.hotStartupHours > 0 {
|
||
log.Printf("[store] hot-startup window configured: hotStartupHours=%gh, retentionHours=%gh (background fill loader runs after LoadChunked succeeds — see RunStartupLoad)",
|
||
store.hotStartupHours, store.retentionHours)
|
||
} else {
|
||
log.Printf("[store] hot-startup disabled (hotStartupHours=0) — no background fill loader will run")
|
||
}
|
||
if firstChunkPath {
|
||
// Only spawn the late-reader if the select did NOT already drain
|
||
// loadErrCh. Otherwise this goroutine would block forever on an
|
||
// empty channel (goroutine leak — adv #1).
|
||
go func() {
|
||
if err := <-loadErrCh; err != nil {
|
||
log.Printf("[store] RunStartupLoad background error: %v", err)
|
||
}
|
||
}()
|
||
}
|
||
|
||
// Neighbor graph: the persisted snapshot (if present) was already
|
||
// loaded above, before the packet load. Per #1287 schema migrations
|
||
// all live in the ingestor; the server only reads the snapshot and
|
||
// then refreshes it via the recompNeighborGraph slot every 60s.
|
||
dbPath = database.path
|
||
database.hasResolvedPath = true // dbschema.AssertReady above already verified observations.resolved_path exists
|
||
|
||
// WaitGroup for background init steps that gate /api/healthz readiness.
|
||
var initWg sync.WaitGroup
|
||
|
||
if !neighborEdgesPersisted {
|
||
// No persisted snapshot yet (e.g. fresh DB before the ingestor
|
||
// has run its first edge-build cycle). Build an in-memory graph
|
||
// from the packets we already have so reads aren't empty. We
|
||
// do NOT persist — the ingestor owns neighbor_edges writes per
|
||
// #1287; the recompNeighborGraph recomputer will pick up the
|
||
// real snapshot as soon as the ingestor populates it.
|
||
log.Printf("[neighbor] no persisted edges found, will build in-memory in background...")
|
||
store.graph.Store(NewNeighborGraph())
|
||
initWg.Add(1)
|
||
go func() {
|
||
defer initWg.Done()
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
log.Printf("[neighbor] graph build panic recovered: %v", r)
|
||
}
|
||
}()
|
||
built := BuildFromStore(store)
|
||
store.graph.Store(built)
|
||
log.Printf("[neighbor] in-memory graph build complete")
|
||
}()
|
||
}
|
||
|
||
// Initial pickBestObservation runs in background — doesn't need to block HTTP.
|
||
// API serves best-effort data until this completes (~10s for 100K txs).
|
||
// Processes in chunks of 5000, releasing the lock between chunks so API
|
||
// handlers remain responsive.
|
||
initWg.Add(1)
|
||
go func() {
|
||
defer initWg.Done()
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
log.Printf("[store] pickBestObservation panic recovered: %v", r)
|
||
}
|
||
}()
|
||
const chunkSize = 5000
|
||
store.mu.RLock()
|
||
totalPackets := len(store.packets)
|
||
store.mu.RUnlock()
|
||
|
||
for i := 0; i < totalPackets; i += chunkSize {
|
||
end := i + chunkSize
|
||
if end > totalPackets {
|
||
end = totalPackets
|
||
}
|
||
store.mu.Lock()
|
||
for j := i; j < end && j < len(store.packets); j++ {
|
||
pickBestObservation(store.packets[j])
|
||
store.trackedBytes += rechargeTx(store.packets[j])
|
||
}
|
||
store.mu.Unlock()
|
||
if end < totalPackets {
|
||
time.Sleep(10 * time.Millisecond) // yield to API handlers
|
||
}
|
||
}
|
||
log.Printf("[store] initial pickBestObservation complete (%d transmissions)", totalPackets)
|
||
}()
|
||
|
||
// Mark server ready once all background init completes.
|
||
go func() {
|
||
initWg.Wait()
|
||
readiness.Store(1)
|
||
log.Printf("[server] readiness: ready=true (background init complete)")
|
||
}()
|
||
|
||
// WebSocket hub
|
||
hub := NewHub()
|
||
hub.SetAllowedOrigins(cfg.CORSAllowedOrigins)
|
||
hub.ConfigureLimits(cfg.WSMaxConnsPerIP(), cfg.WSUpgradesPerMinPerIP(),
|
||
cfg.WSTrustedProxies(), cfg.WSDeny()) // #1794
|
||
hub.upgrader.EnableCompression = cfg.WSCompressionEnabled()
|
||
|
||
// HTTP server
|
||
srv := NewServer(database, cfg, hub)
|
||
srv.configDir = configDir
|
||
srv.store = store
|
||
// Optional user management (off by default). Fails startup on a bad
|
||
// config rather than running with registration that cannot work.
|
||
if err := srv.initUserManagement(resolvedDB); err != nil {
|
||
log.Fatalf("[users] %v", err)
|
||
}
|
||
router := mux.NewRouter()
|
||
srv.RegisterRoutes(router)
|
||
|
||
// WebSocket endpoint
|
||
router.HandleFunc("/ws", hub.ServeWS)
|
||
|
||
// Static files + SPA fallback
|
||
absPublic, _ := filepath.Abs(publicDir)
|
||
if _, err := os.Stat(absPublic); err == nil {
|
||
fs := http.FileServer(http.Dir(absPublic))
|
||
router.PathPrefix("/").Handler(wsOrStatic(hub, spaHandler(absPublic, fs)))
|
||
log.Printf("[static] serving %s", absPublic)
|
||
} else {
|
||
log.Printf("[static] directory %s not found — API-only mode", absPublic)
|
||
router.PathPrefix("/").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
w.Header().Set("Content-Type", "text/html")
|
||
w.Write([]byte(`<!DOCTYPE html><html><body><h1>CoreScope</h1><p>Frontend not found. API available at /api/</p></body></html>`))
|
||
})
|
||
}
|
||
|
||
// Start SQLite poller for WebSocket broadcast
|
||
poller := NewPoller(database, hub, time.Duration(pollMs)*time.Millisecond)
|
||
poller.store = store
|
||
go poller.Start()
|
||
|
||
// Start periodic eviction
|
||
stopEviction := store.StartEvictionTicker()
|
||
defer stopEviction()
|
||
|
||
// Steady-state analytics recomputers (issue #1240). Replaces the
|
||
// on-request compute-then-cache pattern for the default (region="",
|
||
// zero-window) analytics queries with a background refresh loop so
|
||
// reads always hit cache in <1ms.
|
||
stopAnalyticsRecomp := store.StartAnalyticsRecomputers(
|
||
cfg.AnalyticsDefaultRecomputeInterval(),
|
||
cfg.AnalyticsRecomputeIntervals(),
|
||
)
|
||
defer stopAnalyticsRecomp()
|
||
log.Printf("[analytics-recompute] background recompute enabled (default=%s)", cfg.AnalyticsDefaultRecomputeInterval())
|
||
|
||
// #1481 P0-1: background recomputer for the default-shape
|
||
// /api/analytics/neighbor-graph response (5 min cadence). Reads
|
||
// hit an atomic pointer; the rebuild path no longer runs on the
|
||
// request goroutine for the common filter shape.
|
||
stopNeighborGraphCache := make(chan struct{})
|
||
ngInterval := neighborGraphCacheInterval
|
||
if cfg.NeighborGraph != nil && cfg.NeighborGraph.CacheRecomputeIntervalSeconds > 0 {
|
||
ngInterval = time.Duration(cfg.NeighborGraph.CacheRecomputeIntervalSeconds) * time.Second
|
||
}
|
||
srv.startNeighborGraphRecomputer(ngInterval, stopNeighborGraphCache)
|
||
defer close(stopNeighborGraphCache)
|
||
log.Printf("[neighbor-graph-cache] background recompute enabled (interval=%s)", ngInterval)
|
||
|
||
// Known-channels catalogue cache (issue #1323). OPT-IN: an empty
|
||
// cfg.KnownChannelsURL leaves srv.knownChannels nil and starts no
|
||
// background fetch. The /api/known-channels endpoint then serves an
|
||
// empty snapshot. Operators who want the community catalogue must
|
||
// set knownChannelsUrl explicitly in config.json (see
|
||
// config.example.json for the pinned-SHA recommendation).
|
||
if cfg.KnownChannelsURL != "" {
|
||
kcRefresh := DefaultKnownChannelsRefresh
|
||
if cfg.KnownChannelsRefreshMs > 0 {
|
||
kcRefresh = time.Duration(cfg.KnownChannelsRefreshMs) * time.Millisecond
|
||
}
|
||
srv.knownChannels = newKnownChannelsCache(cfg.KnownChannelsURL, kcRefresh)
|
||
kcCtx, stopKnownChannels := context.WithCancel(context.Background())
|
||
srv.knownChannels.run(kcCtx)
|
||
defer stopKnownChannels()
|
||
log.Printf("[known-channels] background fetch enabled (url=%s, refresh=%s)", cfg.KnownChannelsURL, kcRefresh)
|
||
} else {
|
||
log.Printf("[known-channels] disabled (knownChannelsUrl unset in config)")
|
||
}
|
||
|
||
// Steady-state repeater-enrichment recomputer (issue #1262).
|
||
// Prewarms the bulk caches feeding handleNodes so the very first
|
||
// /api/nodes?limit=2000 from live.js's SPA bootstrap hits a
|
||
// populated cache instead of paying a 15.7s on-thread rebuild.
|
||
// Uses the configured RelayActiveHours window and the same
|
||
// default recompute interval as the other analytics caches.
|
||
relayWindowHours := cfg.GetHealthThresholds().RelayActiveHours
|
||
stopRepeaterEnrichRecomp := store.StartRepeaterEnrichmentRecomputer(
|
||
relayWindowHours,
|
||
cfg.AnalyticsDefaultRecomputeInterval(),
|
||
)
|
||
defer stopRepeaterEnrichRecomp()
|
||
log.Printf("[repeater-enrich-recompute] background recompute enabled (window=%.1fh, interval=%s)",
|
||
relayWindowHours, cfg.AnalyticsDefaultRecomputeInterval())
|
||
|
||
// Steady-state bridge-centrality recomputer (issue #672 axis 2).
|
||
// Computes betweenness centrality over the in-memory neighbor
|
||
// graph and stores the per-pubkey score map atomically. Read by
|
||
// handleNodes via a single atomic load.
|
||
stopBridgeRecomp := store.StartBridgeScoreRecomputer(
|
||
cfg.AnalyticsDefaultRecomputeInterval(),
|
||
)
|
||
defer stopBridgeRecomp()
|
||
log.Printf("[bridge-recompute] background recompute enabled (interval=%s)",
|
||
cfg.AnalyticsDefaultRecomputeInterval())
|
||
|
||
// Steady-state coverage + redundancy recomputer (issue #672 axes 3 & 4).
|
||
// Computes harmonic-reach coverage and articulation-point redundancy over
|
||
// the same neighbor graph as the bridge axis, storing both per-pubkey
|
||
// maps atomically. Read by handleNodes via single atomic loads.
|
||
stopUsefulnessAxesRecomp := store.StartUsefulnessAxesRecomputer(
|
||
cfg.AnalyticsDefaultRecomputeInterval(),
|
||
)
|
||
defer stopUsefulnessAxesRecomp()
|
||
log.Printf("[usefulness-axes-recompute] background recompute enabled (interval=%s)",
|
||
cfg.AnalyticsDefaultRecomputeInterval())
|
||
|
||
// Steady-state neighbor-graph snapshot recomputer (issue #1287).
|
||
// Per Option 4: the ingestor owns neighbor_edges; the server
|
||
// READS the snapshot every 60s and atomic-swaps it into s.graph.
|
||
// This is the ONLY path that updates s.graph at steady state.
|
||
stopNeighborRecomp := store.StartNeighborGraphRecomputer(NeighborGraphRecomputerDefaultInterval)
|
||
defer stopNeighborRecomp()
|
||
log.Printf("[neighbor-recompute] snapshot reload enabled (interval=%s)",
|
||
NeighborGraphRecomputerDefaultInterval)
|
||
|
||
// Packet / metrics / observer retention moved to the ingestor in
|
||
// #1283 (writes only belong on the writer process). Neighbor-edge
|
||
// pruning moved to the ingestor in #1287 for the same reason. The
|
||
// server no longer schedules any of these; the ingestor's tickers
|
||
// handle them.
|
||
_ = cfg.IncrementalVacuumPages() // kept reachable for config validation; not used here
|
||
_ = cfg.NeighborMaxAgeDays() // ditto — owned by ingestor now
|
||
|
||
// Graceful shutdown
|
||
var handler http.Handler = router
|
||
if cfg.GZipEnabled() {
|
||
handler = gzipMiddlewareWithConfig(cfg.Compression, router)
|
||
log.Printf("[server] HTTP gzip compression enabled")
|
||
}
|
||
// #1009: stamp X-CoreScope-Load-Status on every response so probes
|
||
// and dashboards can see when the chunked Load is still in flight.
|
||
// Outermost wrap so the header is set regardless of gzip/etc.
|
||
handler = loadStatusMiddleware(store, handler)
|
||
if cfg.WSCompressionEnabled() {
|
||
log.Printf("[server] WebSocket permessage-deflate compression enabled")
|
||
}
|
||
httpServer := &http.Server{
|
||
Addr: fmt.Sprintf(":%d", cfg.Port),
|
||
Handler: handler,
|
||
ReadTimeout: 30 * time.Second,
|
||
WriteTimeout: 60 * time.Second,
|
||
IdleTimeout: 120 * time.Second,
|
||
}
|
||
|
||
go func() {
|
||
sigCh := make(chan os.Signal, 1)
|
||
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
||
sig := <-sigCh
|
||
log.Printf("[server] received %v, shutting down...", sig)
|
||
|
||
// 1. Stop accepting new WebSocket/poll data
|
||
poller.Stop()
|
||
|
||
// 1b. Auto-prune tickers were all relocated to the ingestor in
|
||
// #1283/#1287 — nothing to stop here.
|
||
|
||
// 1c. Stop steady-state analytics recomputers (issue #1240).
|
||
// Must happen before dbClose so any in-flight compute that
|
||
// reaches into SQLite has finished.
|
||
if stopAnalyticsRecomp != nil {
|
||
stopAnalyticsRecomp()
|
||
}
|
||
|
||
// 2. Gracefully drain HTTP connections (up to 15s)
|
||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||
defer cancel()
|
||
if err := httpServer.Shutdown(ctx); err != nil {
|
||
log.Printf("[server] HTTP shutdown error: %v", err)
|
||
}
|
||
|
||
// 3. Close WebSocket hub
|
||
hub.Close()
|
||
|
||
// 3b. Close users.db (user management, opt-in; no-op when off).
|
||
srv.closeUserManagement()
|
||
|
||
// 4. Close database (release SQLite WAL lock)
|
||
if err := dbClose(); err != nil {
|
||
log.Printf("[server] DB close error: %v", err)
|
||
}
|
||
log.Println("[server] shutdown complete")
|
||
}()
|
||
|
||
log.Printf("[server] CoreScope (Go) listening on http://localhost:%d", cfg.Port)
|
||
|
||
// Backfills (resolved_path, from_pubkey) moved to the ingestor in
|
||
// #1287 — they are write operations and belong on the writer
|
||
// process. The server reads the results via the periodic
|
||
// recompNeighborGraph / fetchResolvedPathForObs paths.
|
||
|
||
// Migrate old content hashes in background (one-time, idempotent).
|
||
go migrateContentHashesAsync(store, 5000, 100*time.Millisecond)
|
||
|
||
if err := httpServer.ListenAndServe(); err != http.ErrServerClosed {
|
||
log.Fatalf("[server] %v", err)
|
||
}
|
||
}
|
||
|
||
// spaHandler serves static files, falling back to index.html for SPA routes.
|
||
// It reads index.html once at creation time and replaces the __BUST__ placeholder
|
||
// with a Unix timestamp so browsers fetch fresh JS/CSS after each server restart.
|
||
func spaHandler(root string, fs http.Handler) http.Handler {
|
||
// Pre-process index.html: replace __BUST__ with a cache-bust timestamp
|
||
indexPath := filepath.Join(root, "index.html")
|
||
rawHTML, err := os.ReadFile(indexPath)
|
||
if err != nil {
|
||
log.Printf("[static] warning: could not read index.html for cache-bust: %v", err)
|
||
rawHTML = []byte("<!DOCTYPE html><html><body><h1>CoreScope</h1><p>index.html not found</p></body></html>")
|
||
}
|
||
bustValue := fmt.Sprintf("%d", time.Now().Unix())
|
||
indexHTML := []byte(strings.ReplaceAll(string(rawHTML), "__BUST__", bustValue))
|
||
log.Printf("[static] cache-bust value: %s", bustValue)
|
||
|
||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
// Defense-in-depth: explicitly reject path-traversal attempts before
|
||
// we touch the filesystem. gorilla/mux + http.FileServer already clean
|
||
// most of these, but we don't want a future SkipClean(true) (or a
|
||
// different router) to silently expose the FS. See
|
||
// audit-input-vulns-20260603 (LOW — SPA static handler depends on
|
||
// default mux path-cleaning).
|
||
if !isSafeStaticPath(r.URL.Path, r.URL.RawPath) {
|
||
http.Error(w, "bad request", http.StatusBadRequest)
|
||
return
|
||
}
|
||
// Serve pre-processed index.html for root and /index.html
|
||
if r.URL.Path == "/" || r.URL.Path == "/index.html" {
|
||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||
w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate")
|
||
w.Write(indexHTML)
|
||
return
|
||
}
|
||
|
||
path := filepath.Join(root, r.URL.Path)
|
||
if _, err := os.Stat(path); os.IsNotExist(err) {
|
||
// SPA fallback — serve pre-processed index.html
|
||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||
w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate")
|
||
w.Write(indexHTML)
|
||
return
|
||
}
|
||
// Disable caching for JS/CSS/HTML
|
||
if filepath.Ext(path) == ".js" || filepath.Ext(path) == ".css" || filepath.Ext(path) == ".html" {
|
||
w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate")
|
||
}
|
||
fs.ServeHTTP(w, r)
|
||
})
|
||
}
|
||
|
||
// isSafeStaticPath rejects request paths that contain traversal sequences
|
||
// or backslashes — defense-in-depth for the SPA static handler so a future
|
||
// router with SkipClean(true) cannot expose the filesystem. Empty input is
|
||
// safe (root handled earlier).
|
||
//
|
||
// urlPath is the decoded path (r.URL.Path); rawPath is the raw, possibly
|
||
// percent-encoded path (r.URL.RawPath) used to catch encoded `..` / `\`.
|
||
func isSafeStaticPath(urlPath, rawPath string) bool {
|
||
for _, p := range []string{urlPath, rawPath} {
|
||
if p == "" {
|
||
continue
|
||
}
|
||
// Lowercase for case-insensitive percent-encoding checks.
|
||
lp := strings.ToLower(p)
|
||
// Block "..", any URL-encoded "%2e%2e" sequence, and backslashes
|
||
// (which Windows-style traversal exploits convert to "\").
|
||
if strings.Contains(p, "..") ||
|
||
strings.Contains(lp, "%2e%2e") ||
|
||
strings.Contains(p, "\\") ||
|
||
strings.Contains(lp, "%5c") {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|