diff --git a/AGENTS.md b/AGENTS.md index 4aaa4eff..bf7dc52f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,7 +41,7 @@ Sixteen internal packages under `internal/` (plus `internal/testutil/` for share - **`ingest/`** — Ingest worker pipeline (`worker.go`: JetStream input → per-table batch INSERT with DLQ output; over a `Sharded` queue, `claims.go`'s `ClaimShards` narrows the worker to the units this process is assigned — membership leases, capped rendezvous, halt-drain-then-release handover and stop, reset at takeover from a dead owner, each unit's share of a 10,000-row budget of unsettled rows). The pipeline is **insert-only**. The wire format `EventMessage` (`types.go`) carries `{table_name, scope, received_timestamp, format, columns, row}` and nothing else — `row` is one positional `JSONCompactEachRow` line and `columns` names its slots, the table's **insertable** columns (a `MATERIALIZED`/`ALIAS` column cannot be named in an `INSERT`); the worker batches per (tenant, table, column list), the tenant read off each message's `mq.Topic`, and inserts each batch into its tenant's own ClickHouse (`chconn.Pools.Target`); the worker accepts whatever table name the envelope carries (table existence was already checked by the HTTP ingest handler, which `404`s an unknown table before publish; the worker doesn't re-validate), then bulk-INSERTs. In the embedded-NATS deployment (the default), the server runs with `DontListen: true` (`internal/mq/embedded.go`), so the only Publishers reachable on the `ingest.>` subjects are in-process Go code — today, only the HTTP `/v1/ingest?table={table}` handler. Non-insert mutations (`DELETE`/`UPDATE`/`TRUNCATE`/…) must go through `POST /v1/ops/query` under the admin role (the same `RequireAdmin` gate as the rest of `/v1/ops/*`, so non-admin callers never reach the proxy) or through an operator-authored pipe that writes, gated only by its `allowed_roles` (#386). A request with no token (or an invalid one) resolves to the `default_role`, which in a production config is not the admin role (setting them equal is a loudly-warned dev-only setting), so it can't reach this endpoint. Plus `Sweeper` (Active Sweeper for NATS message lifecycle) + `EventMessage`/`BufferConsumerName` types (`types.go`) - **`keyenc/`** — the one escaping composite keys are built from: `Escape` keeps `[A-Za-z0-9_-]` (exactly the tenant-id grammar, so a tenant id is its own escaped form) and writes every other byte as `%XX`, `Unescape` is `url.PathUnescape` (lenient: either hex case, and a byte left unescaped reads as itself, so a `%2D` an earlier build wrote still reads), `Join`/`AppendJoin` escape each field and put a separator between them (they panic on no fields, and on a separator the escaping could write or one outside ASCII) and `Split` reverses them. The package that builds a key takes raw names and escapes them itself, so no caller has to and no field reaches a key unescaped: NATS subjects (`Join`/`Split` after the verbatim tenant) and the cache's keys — the version index and the shared backend's Redis keys (`internal/cache`) — and the dedupe keys (`//`) use it; changing what it keeps orphans every stored key (an orphaned dedupe key lets a seen id through again), and on the shared backend, whose keys every process builds for itself, splits them between builds for the length of a rolling upgrade (a bump one build makes misses the entries the other filed, served until their TTL) - **`mq/`** — the message-queue boundary: the **only** package that imports NATS/JetStream (Key Design Decision #20), and the only one that knows how the broker works. Everything else addresses events by `Topic{Tenant, Table, Scope}` (a validated tenant id and raw names — the tenant leads every subject, `ingest..
`, so one wildcard selects a tenant's traffic, and a topic without one is refused) and states intent through the interfaces — `Publisher` (`ErrQueueFull` is the backpressure signal, `ErrUnavailable` a broker that cannot be reached — both a `503`, with `Retry-After` `30` and `5`; `WithIdempotencyKey` makes a republish inside the queue's duplicate window a no-op), `Subscriber`, `ConsumerManager`/`Consumer`/`ConsumerConfig` (the ingest worker's durable consumer), `DeadLetterer` and `DeadLetterStats` (park a message, count what is parked), `Purger` (drop what is both acked and older than a cutoff — the sweeper), `Replayer` (SSE gap-fill) — composed into `Broker`, which adds each tenant's byte budget (`SetMaxBytes`/`MaxBytes`: the `mq.max_bytes_gb` reload, which opens a tenant's queue the first time) and `Stats` (the system gauges' source). A broker whose ingest queue is split into units one consumer at a time owns implements `Sharded` too (`IngestUnits`, `ResetOrphaned`, `UnitStates`; `ConsumerConfig.Units` narrows a consumer to some of them, and its consumer is a `Releaser` and a `Halter`, capped per unit by `ConsumerConfig.MaxHeld`, with `ErrConsumerMismatch` when an operator durable no longer fits), and `Message.OnSettled` runs a hook once, at the first ack or nak attempt, confirmed or not. Every interface speaks per tenant, never per stream: the embedded implementation gives each tenant a queue of its own (a stream pair, `INGEST_`/`DLQ_`), and nothing outside the package may assume that layout — an external implementation may keep one shared stream. Subjects, prefixes, wildcards, stream names, sequences, and ack floors are private to the implementations, whose subject tokens are escaped by the shared `internal/keyenc`: `EmbeddedNATS` (`embedded.go`, `subject.go`, `purge.go`, `deadletter.go`), which `internal/app` constructs and hands everything else as a `mq.Broker`, and `ExternalNATS` (`external.go`, `subject_nats.go`, `nats_topology.go`: an operator-owned cluster whose streams, durables and lease bucket it never creates, changes, purges or deletes; `lease.go` holds `coord.backend: nats`'s leases in that bucket), which `internal/app` constructs from the `mq.nats` block when `mq.backend` is `nats` ([#613](https://github.com/Wave-RF/WaveHouse/issues/613)). Every implementation passes the conformance suite in `internal/mq/mqtest` (`mqtest.Run`), which states the `Broker` contract as behavior; a new backend runs it from its own test, with `mqtest.Caps` only where its semantics legitimately differ -- **`observability/`** — OpenTelemetry pipeline: `InitProvider` wires trace/metric/log providers via OTLP gRPC (each signal independently gated). A top-level `Prometheus` config block drives an optional `/metrics` scrape endpoint that runs independently of OTLP push — standalone (Alloy/Mimir scrape, no collector), alongside OTLP, or off. `NewLogger` produces a slog handler that fans out to stdout AND OTLP (stdout always 100%, OTLP sample-rate-aware; both legs gated by the one `WH_LOG_LEVEL` `LevelVar`). `TraceHandler` injects trace_id/span_id from active spans, and `request_id`/`tenant` from the request scope (`request.go`: `WithRequest`, filled by `api`'s `requestID` middleware and its tenant resolution; on log records and span attributes, never a metric label). `Identity` (`identity.go`) is the process's name in its telemetry: one resource for every signal (service name, version, `instance_id`, roles, commit, build time; `OTEL_SERVICE_NAME`/`OTEL_RESOURCE_ATTRIBUTES` override it) and the `wavehouse_build_info` gauge (no instance label; the resource carries it). `PprofHandler` (`pprof.go`) is `net/http/pprof` on a mux of its own, served only on the `profiling.addr` listener (`internal/app`'s `wire_profiling.go`, off by default, loopback by default), never on the API router. `tracer.go` provides W3C trace context propagation over message headers (`InjectHeaders`/`ExtractHeaders` on a plain header map; `internal/mq` injects on every publish and extracts on the `Subscribe` and `Consumer` paths, so this package never sees a NATS type). The ingest worker continues the trace past the queue (`internal/ingest/trace.go`, [#774](https://github.com/Wave-RF/WaveHouse/issues/774)): `IngestWorker.Flush`, a root span linked to the span each row of the batch was published under (one batch holds many requests' rows, so links, not a parent), and `clickhouse.insert` as its child around every INSERT; `tenant` rides as a span attribute there, never a metric label. +- **`observability/`** — OpenTelemetry pipeline: `InitProvider` wires trace/metric/log providers via OTLP gRPC (each signal independently gated). A top-level `Prometheus` config block drives an optional `/metrics` scrape endpoint that runs independently of OTLP push — standalone (Alloy/Mimir scrape, no collector), alongside OTLP, or off. `NewLogger` produces a slog handler that fans out to stdout AND OTLP (stdout always 100%, OTLP sample-rate-aware; both legs gated by the one `WH_LOG_LEVEL` `LevelVar`). `TraceHandler` injects trace_id/span_id from active spans, and `request_id`/`client.address`/`tenant` from the request scope (`request.go`: `WithRequest`, filled by `api`'s `requestID` middleware, its `clientAddress` (`clientip.go`: `X-Forwarded-For` read right to left, and only from a peer in `server.trusted_proxies`, which is also the only peer whose `traceparent` the server span joins; any other is linked) and its tenant resolution; on log records and span attributes, never a metric label). `Identity` (`identity.go`) is the process's name in its telemetry: one resource for every signal (service name, version, `instance_id`, roles, commit, build time; `OTEL_SERVICE_NAME`/`OTEL_RESOURCE_ATTRIBUTES` override it) and the `wavehouse_build_info` gauge (no instance label; the resource carries it). `PprofHandler` (`pprof.go`) is `net/http/pprof` on a mux of its own, served only on the `profiling.addr` listener (`internal/app`'s `wire_profiling.go`, off by default, loopback by default), never on the API router. `tracer.go` provides W3C trace context propagation over message headers (`InjectHeaders`/`ExtractHeaders` on a plain header map; `internal/mq` injects on every publish and extracts on the `Subscribe` and `Consumer` paths, so this package never sees a NATS type). The ingest worker continues the trace past the queue (`internal/ingest/trace.go`, [#774](https://github.com/Wave-RF/WaveHouse/issues/774)): `IngestWorker.Flush`, a root span linked to the span each row of the batch was published under (one batch holds many requests' rows, so links, not a parent), and `clickhouse.insert` as its child around every INSERT; `tenant` rides as a span attribute there, never a metric label. - **`pipes/`** — Named query pipes: `NamedQuery` type + `BindParams` + `Source` (read per request; `settings.Store` in production, `Static(q...)` in tests) - **`policy/`** — Hasura-style access control, **role-first**: `TablePolicy` is `map[string]RolePermissions`, and a role's grant splits by operation into `SelectPermissions` (columns, row `filter`, aggregations, the `max_*` limits) and `InsertPermissions` (columns, `check`) — so a field only one side honors does not exist on the other. `Evaluate()` resolves ONE operation and leaves the other side **nil** (`Select *ResolvedSelect` / `Insert *ResolvedInsert`), which every accessor fails closed on — nil is "not resolved", distinct from an empty side, which is "unrestricted" (what the admin return builds). Claim templating (`{{ jwt.claim.path }}`) resolves during that call. Policies come from `Source`, a `func() *Policy` read per call (`settings.Store.Policy` in production, `Static(p)` in tests) - **`query/`** — Structured query AST types + SQL builder with schema validation, structural policy predicate/limit emission, timestamp bucketing diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ff721ea..e1f6a7bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - **Dropped SSE frames say why, on which table and on which connection** (`internal/stream/{metrics,subscriber,hub}.go` (+ tests, `drop_test.go` new), `internal/api/stream.go` (+ `stream_drop_test.go` new), `docs/src/content/docs/{deployment,architecture,configuration}.md*`, `AGENTS.md`): closes [#424](https://github.com/Wave-RF/WaveHouse/issues/424), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). `wavehouse_sse_dropped_frames_total` counted only frames dropped to a full queue, labeled by `kind` alone, and nothing was logged, so a burst of drops could not be tied to a connection or told apart from clients going away. It now also counts the frames a stream left unwritten when it ended, and carries `reason` (`buffer_full`, `client_gone`, `stream_closed`) and `table` (empty for a table the tenant's schema does not have, since the client names it), which `prometheus.per_table_metrics: false` drops like every other table label; a query that sums the counter now also sees the two new reasons, so filter on `reason="buffer_full"` to keep the old meaning. New `wavehouse_sse_queue_peak_frames` gauge: the deepest any open stream's queue has been since it opened, so a fill between scrapes still shows. Every stream logs `SSE stream opened` and `SSE stream closed` (how it ended, how long it lasted, what it dropped), and a slow consumer logs `SSE stream dropping frames` at `WARN`, at most every 10 seconds per stream, with its queue's depth, peak and capacity; each line carries the connection's `request_id` and `tenant` from the request scope. No metric carries a tenant label. +- **Logs and traces name the real client behind a trusted proxy, and only a trusted proxy's `traceparent` is joined** (`internal/config/proxies.go` (new, + tests), `internal/config/{config,backends}.go`, `internal/api/clientip.go` (new, + tests), `internal/api/router.go`, `internal/observability/{request,logger}.go` (+ tests), one line each in `internal/app/{wire,wire_ops}.go`, comments in `internal/auth/auth.go`, `config.yaml`, `docs/src/content/docs/{configuration,reverse-proxy}.mdx`, `{deployment,architecture}.md`, `AGENTS.md`): closes [#333](https://github.com/Wave-RF/WaveHouse/issues/333), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). Behind a proxy WaveHouse could log only the proxy's address, while the server span's `client.address`, which OpenTelemetry's HTTP instrumentation takes from the leftmost `X-Forwarded-For` entry, said whatever the client wrote there. `server.trusted_proxies` (`WH_SERVER_TRUSTED_PROXIES`, CIDRs or bare addresses, default empty) lists the proxies to believe: from a listed peer the client is the first `X-Forwarded-For` hop, read right to left, that is not itself a listed proxy; from any other peer it is the connection's address and the header is ignored. The result is `client.address` on the server span, replacing the instrumentation's value, and on every log record written through the request's context, the operator-key audit line included, next to `request_id` and `tenant`; it is never a metric label. An entry that does not parse, or an IPv4-mapped prefix shorter than `/96`, refuses boot, and a `/0` entry is a boot warning. An incoming `traceparent` used to be joined whoever sent it; it is now joined only from a listed proxy, and from any other peer starts a new trace linked to the one it named, so with the list empty (the default) a proxy's trace is linked rather than continued until you list the proxy. `r.RemoteAddr` is still never rewritten, so nothing reintroduces `middleware.RealIP`'s spoofing. + - **Settings reloads and the tenants they serve or reject are metrics** (`internal/settings/{stats,registry}.go` (+ tests), `internal/app/wire_settings_metrics.go` (new, + test), `internal/app/{app,wire}.go`, `docs/src/content/docs/{deployment,architecture}.md`, `settings-directory.mdx`, `AGENTS.md`): closes [#776](https://github.com/Wave-RF/WaveHouse/issues/776), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). Reload outcomes were log lines only, so a tenant rejected on one process of several had nothing to alert on but a log grep. Each process now exports `wavehouse_settings_reloads_total{trigger, result}` (`adopted`, `partial`, `rejected`), `wavehouse_settings_last_reload_successful`, `wavehouse_settings_last_reload_success_timestamp_seconds`, `wavehouse_settings_tenants{state}` (`served`, `rejected`), `wavehouse_settings_tenants_removed_total` and `wavehouse_settings_reconcile_failures_total{component}`, none with a tenant label ([#538](https://github.com/Wave-RF/WaveHouse/issues/538)). A reload used to count as adopted even when the work after it failed, since the `AfterAdopt` hooks return nothing: the ClickHouse pool hook only logged, and the dedupe and queue-budget hooks discarded their errors. Those three now report through `settings.Registry.Reconciled`, which makes the reload `partial`, and a failure while boot applies the settings counts against the boot reload. What `POST /v1/ops/settings/reload` returns is unchanged. In `internal/app/wire.go` the change is one call per hook, and `app.go` gains one call at the end of `New`, which open PRs [#712](https://github.com/Wave-RF/WaveHouse/pull/712) and [#770](https://github.com/Wave-RF/WaveHouse/pull/770) also edit. - **Go's pprof endpoints on a listener of their own** (`internal/config/profiling.go` (new, + tests), `internal/config/{config,backends}.go`, `internal/observability/pprof.go` (new, + tests), `internal/app/wire_profiling.go` (new, + tests), one call in `internal/app/app.go`, `config.yaml`, `Makefile`, `docs/src/content/docs/{configuration.mdx,deployment,development,architecture}.md`, `AGENTS.md`): closes [#143](https://github.com/Wave-RF/WaveHouse/issues/143), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). Diagnosing memory growth, a goroutine leak or a CPU hotspot in a running process needed a rebuild with instrumentation. `profiling.enabled` (`WH_PROFILING_ENABLED`, default `false`) now serves `net/http/pprof` under `/debug/pprof/` on `profiling.addr` (`WH_PROFILING_ADDR`, default `127.0.0.1:6060`) in every process, whatever its roles. It is never mounted on the API router, and nothing serves `http.DefaultServeMux`, where the package's import also registers it. The endpoints take no credential, so the default address is loopback, and an address that is not loopback is logged at `WARN` at boot. Boot refuses an address that is not `host:port`, or whose port is `server.port` or the Prometheus sidecar's. A CPU profile or trace in flight ends when the graceful stop begins. `make dev` turns it on unless `WH_PROFILING_ENABLED` is already set. @@ -25,6 +27,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Changed +- **An incoming `traceparent` is joined only from a trusted proxy** (`internal/api/router.go`, `docs/src/content/docs/reverse-proxy.mdx`): part of [#333](https://github.com/Wave-RF/WaveHouse/issues/333) (see the `server.trusted_proxies` entry under Added). Before, a request's server span became a child of whatever `traceparent` it carried, so any client could write spans into your traces and pick the trace id its requests were logged under. Now only a request from a peer in `server.trusted_proxies` joins the incoming trace; from any other peer it starts a new trace linked to the one it named. The list is empty by default, so **if a tracing proxy (nginx's `ngx_otel_module`, Caddy's `tracing`, Envoy) sends `traceparent` to WaveHouse, list that proxy to keep continuing its traces** ([Reverse proxy → Client address and trace context](https://wavehouse.dev/reverse-proxy#client-address-and-trace-context)). + - **`Wave-RF/cloudflare-preview` is pinned at v1.1.2** (`.github/workflows/ci.yml`, `.github/workflows/docs-preview-cleanup.yml`). The docs preview comment no longer turns an `@name` in a commit subject into a mention that notifies that person: the old `@` escape was decoded by GitHub, and live testing measured a real mention. URLs in commit subjects no longer autolink. A Cloudflare authentication error is no longer retried three times. ([Wave-RF/cloudflare-preview#2](https://github.com/Wave-RF/cloudflare-preview/issues/2)) - **The docs preview job no longer holds `issues: write`, and `Wave-RF/cloudflare-preview` is pinned at v1.1.1** (`.github/workflows/ci.yml`, `.github/workflows/docs-preview-cleanup.yml`): `pull-requests: write` alone lists, posts and edits the preview comment on a PR (measured: [Wave-RF/cloudflare-preview#1](https://github.com/Wave-RF/cloudflare-preview/issues/1)), and v1.1.1 lists each earlier commit once in the comment's history, where a re-run used to add a second row. - **The docs site and its CI run on Wave RF's public packages instead of in-repo copies** (`docs/astro.config.mjs`, `docs/package.json`, `docs/src/config/trademarks.ts`, `docs/src/components/{Footer,Hero,ScrollHints,PageFeedback}.astro`, `docs/src/styles/global.css`, `.github/workflows/ci.yml`, `.github/workflows/docs-preview-cleanup.yml` (new), `.github/workflows/README.md`, `docs/wrangler.jsonc`, `Makefile`, `AGENTS.md`, `docs/src/content/docs/development.md`; removes `docs/scripts/{dev,mermaid-audit,screenshot-diagrams}.mjs`, `docs/src/integrations/diagram-png.mjs`, `docs/src/components/{MermaidZoom,Trademarks}.astro`, `docs/src/plugins/rehype-trademarks.ts`, `scripts/ci/docs-preview-comment.sh`): the site's own implementations were extracted into packages, and this adopts them back. `make dev-docs` runs `cloudflare-md-router dev` from [`@wave-rf/cloudflare-md-router`](https://github.com/Wave-RF/cloudflare-md-router) (same loop; `DOCS_PORT` is now `PORT` and `DOCS_WATCH_STRICT` is now `MD_ROUTER_DEV_STRICT`). Diagram PNG export and the click-to-zoom lightbox are `diagramPng`/`diagramZoom` from [`@wave-rf/astro-themed-mermaid`](https://github.com/Wave-RF/astro-themed-mermaid) (PNG paths unchanged; `DOCS_SKIP_DIAGRAM_PNG` is now `ASTRO_THEMED_MERMAID_SKIP_PNG`), and the diagram audit and screenshots are `pnpm --filter wavehouse-docs run diagrams:audit` / `diagrams:screenshot`. Trademark symbols and the footer notices come from [`@wave-rf/astro-trademarks`](https://github.com/Wave-RF/astro-trademarks), which builds the notices from the same pass over the rendered page that places the symbols, so the footer lists exactly the marks the page marks: a mark named only in a heading or code no longer gets a notice ([#497](https://github.com/Wave-RF/WaveHouse/issues/497)). The docs preview and production deploy use [`Wave-RF/cloudflare-preview`](https://github.com/Wave-RF/cloudflare-preview): the PR comment now leads with a stable per-PR URL (`pr--…workers.dev`) beside the per-commit one, keeps earlier commits' URLs, and the preview is recorded as a GitHub Deployment (environment `preview/docs`), so `docs-preview` gains `deployments: write`; a closed PR's preview deployments are retired by the new `docs-preview-cleanup.yml`. Production deploys are now recorded too, in a `production` environment pointing at `wavehouse.dev`, so `docs-deploy` gains `deployments: write`. diff --git a/config.yaml b/config.yaml index dae52b82..2f253d78 100644 --- a/config.yaml +++ b/config.yaml @@ -23,6 +23,11 @@ server: # release of the stores (5s) and the telemetry flush (3s) are fixed on # top, so size the kill grace at this plus 8s. shutdown_timeout: 10 + # Reverse proxies (CIDRs or addresses) whose X-Forwarded-For names the + # client in logs and spans (client.address), and whose traceparent a + # request's trace joins. Empty trusts no peer: the client is the + # connection's own address, and an incoming traceparent is only linked. + trusted_proxies: [] otel: enabled: false # master switch — set true to export via OTLP gRPC diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index a4051ec6..70c953e3 100644 --- a/docs/src/content/docs/architecture.md +++ b/docs/src/content/docs/architecture.md @@ -79,7 +79,7 @@ pkg/ ### `api/` — HTTP Layer -The API layer uses [Chi](https://github.com/go-chi/chi) for routing with `requestID` (`correlation.go`: keeps an incoming `X-Request-Id` of 1–128 printable ASCII characters with no spaces, sent once, otherwise mints a random one, echoes it on the response, and opens the request's `observability.WithRequest` scope; `annotateSpan`, after the otelhttp middleware, copies it onto the server span; `TenantMW`, and on `/v1/ops` `opsRequestTenant` ahead of authentication, add the tenant through `observability.SetRequestTenant`, any other ops route naming none adding tenant `0` there too, a settings reload naming none acting on every tenant and adding none), a CORS middleware that decorates each response from the allowlist of the tenant the request names (`corsOrigins`; the tenant-exempt routes and a refused request read tenant `0`'s, and nothing when no tenant `0` is served), and a custom JSON recoverer (`jsonRecoverer`) that emits a JSON `500` on panic instead of chi's plain-text `middleware.Recoverer`. +The API layer uses [Chi](https://github.com/go-chi/chi) for routing with `requestID` (`correlation.go`: keeps an incoming `X-Request-Id` of 1–128 printable ASCII characters with no spaces, sent once, otherwise mints a random one, echoes it on the response, and opens the request's `observability.WithRequest` scope; `clientAddress` (`clientip.go`) adds the client address through `observability.SetRequestClient` — the connection's own, or, when the peer is in `server.trusted_proxies`, the first `X-Forwarded-For` hop from the right that is not itself a trusted proxy; the otelhttp middleware joins an incoming `traceparent` only from a trusted proxy and otherwise starts a new trace linked to it (`WithPublicEndpointFn`); `annotateSpan`, after it, copies the scope onto the server span, replacing the `client.address` otelhttp reads from the leftmost `X-Forwarded-For` entry; `TenantMW`, and on `/v1/ops` `opsRequestTenant` ahead of authentication, add the tenant through `observability.SetRequestTenant`, any other ops route naming none adding tenant `0` there too, a settings reload naming none acting on every tenant and adding none), a CORS middleware that decorates each response from the allowlist of the tenant the request names (`corsOrigins`; the tenant-exempt routes and a refused request read tenant `0`'s, and nothing when no tenant `0` is served), and a custom JSON recoverer (`jsonRecoverer`) that emits a JSON `500` on panic instead of chi's plain-text `middleware.Recoverer`. - **router.go** — Route definitions. Public: `/livez`, `/readyz`, and the content-free `/v1/health` SDK ping (plus the permanent `/healthz` alias and the deprecated `/health`, `/ready` aliases). Policy-gated: `/v1/ingest?table={table}`, `/v1/query?table={table}` (structured), `/v1/pipes/{name}` (named pipes), `/v1/stream`. Admin-only (`RequireAdmin` — role == `policy.admin_role`, or a request bearing the operator key's operator bit, which passes even under a nil policy; over a nested settings directory `NewRouter` mounts the gate with no policy at all, whatever `Dependencies.PolicySource` was wired, so the operator key alone passes): `/v1/ops/schema/*`, `/v1/ops/dlq/stats`, `GET /v1/ops/pipes[/{name}]`, `/v1/ops/settings/reload`, `/v1/ops/query` (raw SQL — same gate as the rest of `/v1/ops/*`). `NewOpsRouter` is the router of a process without the `api` role: the probes and their aliases, `/version` and the same-port metrics path (the part it shares with `NewRouter`, `newProbeRouter`), and `POST /v1/ops/settings/reload` behind `RequireAdmin(nil)`, so only the operator key passes; every other route is a 404, under `/v1/ops` only once that gate has passed. - **auth middleware** — the JWT/JWKS authentication middleware is its own package, [`auth/`](#auth--authentication); the router runs it on every `/v1/*` route. @@ -116,7 +116,7 @@ The SSE fan-out, factored out of `api/` so the delivery hot path ([#294](https:/ ### `auth/` — Authentication -- **auth.go** — `NewAuthenticator(cfg, tenantOf, policies)` owns one verifier per tenant: `cfg` is the boot-config half (the operator key, shared by every tenant), and `Reconfigure(id, wiring)` gives a tenant a verifier built from its own `auth` block (`auth.jwks_url` / `auth.jwt_secret` / `auth.role_claim` — the HMAC secret is the tenant's own, so a reload that rotates it rebuilds the verifier; see [Settings Directory — Authentication](/settings-directory#authentication)) — key source plus its pinned `alg` allowlist, swapped atomically after a reload that adopts the tenant and kept when the wiring did not change; `Prune` drops the verifiers of tenants that stopped being served (removed or rejected), `Close` stops every JWKS refresh. `Middleware()` reads the request tenant's verifier per request — the tenant of the store `TenantMW` resolved (`settings.Store.Tenant()`, through the injected `TenantSource`), tenant `0` on a tenant-exempt route — so a JWKS-issued token verifies only under the tenants whose `jwks_url` names its provider's key set; a tenant with no verifier fails closed. A JWKS key set is fetched off the calling goroutine, so neither boot nor a reload waits on the endpoint: the verifier is in place at once and *pending* until a set has been stored (`observedKeys`, the one success signal the library gives — a failed first fetch leaves it holding an empty set). That first fetch is `fetch`'s to retry, from one second doubling to a minute, since a pending verifier never reaches the library and so never triggers its unknown-key-id refetch; once a set is stored the library keeps it fresh on its own, hourly and, rate-limited, on an unknown key id. Each response is capped at 1 MiB. A token checked against a pending verifier records `auth.ErrVerifierPending`, which the router's `refuseUnverifiable` turns into `503` + `Retry-After: 30` on every `/v1` route rather than let the request run under the `default_role`. Verifies JWT tokens with HMAC **or** JWKS (never both), with the accepted `alg` pinned to the active verifier and checked before any key is consulted (rejects `alg: none` and cross-family confusion). Extracts the caller's role from a configurable dot-path claim (`auth.role_claim`, default `role`). Claims parse with `jwt.WithJSONNumber()`, so a numeric claim reaches the policy engine as its exact digits (`json.Number`, never a rounded float64) — part of the row-visibility guarantee (AGENTS.md invariant 12). It always runs and never rejects — a missing/invalid/expired token yields an empty role (resolved to `default_role` downstream), with the token error stashed in context so a denying gate can fail loud (`401`, not a bare `403`). Before the Bearer token it checks a non-JWT operator key (`auth.operator_key`): a constant-time match on the presented credential — an `Authorization: Operator ` header, or the `X-Operator-Key` alias — stamps the live admin role — the request tenant's, read from that tenant's policy (`policies`) — plus an operator bit (`auth.WithOperator`) that `RequireAdmin` honors even under a nil policy — a full-access break-glass credential, audit-logged at Info with no client IP (the audit line goes through the default logger). A presented-but-wrong operator key is logged at `WARN` and counted by `wavehouse_auth_operator_key_failures_total` — a probing signal on the most privileged credential — then falls through like any unauthenticated request (the middleware never rejects). +- **auth.go** — `NewAuthenticator(cfg, tenantOf, policies)` owns one verifier per tenant: `cfg` is the boot-config half (the operator key, shared by every tenant), and `Reconfigure(id, wiring)` gives a tenant a verifier built from its own `auth` block (`auth.jwks_url` / `auth.jwt_secret` / `auth.role_claim` — the HMAC secret is the tenant's own, so a reload that rotates it rebuilds the verifier; see [Settings Directory — Authentication](/settings-directory#authentication)) — key source plus its pinned `alg` allowlist, swapped atomically after a reload that adopts the tenant and kept when the wiring did not change; `Prune` drops the verifiers of tenants that stopped being served (removed or rejected), `Close` stops every JWKS refresh. `Middleware()` reads the request tenant's verifier per request — the tenant of the store `TenantMW` resolved (`settings.Store.Tenant()`, through the injected `TenantSource`), tenant `0` on a tenant-exempt route — so a JWKS-issued token verifies only under the tenants whose `jwks_url` names its provider's key set; a tenant with no verifier fails closed. A JWKS key set is fetched off the calling goroutine, so neither boot nor a reload waits on the endpoint: the verifier is in place at once and *pending* until a set has been stored (`observedKeys`, the one success signal the library gives — a failed first fetch leaves it holding an empty set). That first fetch is `fetch`'s to retry, from one second doubling to a minute, since a pending verifier never reaches the library and so never triggers its unknown-key-id refetch; once a set is stored the library keeps it fresh on its own, hourly and, rate-limited, on an unknown key id. Each response is capped at 1 MiB. A token checked against a pending verifier records `auth.ErrVerifierPending`, which the router's `refuseUnverifiable` turns into `503` + `Retry-After: 30` on every `/v1` route rather than let the request run under the `default_role`. Verifies JWT tokens with HMAC **or** JWKS (never both), with the accepted `alg` pinned to the active verifier and checked before any key is consulted (rejects `alg: none` and cross-family confusion). Extracts the caller's role from a configurable dot-path claim (`auth.role_claim`, default `role`). Claims parse with `jwt.WithJSONNumber()`, so a numeric claim reaches the policy engine as its exact digits (`json.Number`, never a rounded float64) — part of the row-visibility guarantee (AGENTS.md invariant 12). It always runs and never rejects — a missing/invalid/expired token yields an empty role (resolved to `default_role` downstream), with the token error stashed in context so a denying gate can fail loud (`401`, not a bare `403`). Before the Bearer token it checks a non-JWT operator key (`auth.operator_key`): a constant-time match on the presented credential — an `Authorization: Operator ` header, or the `X-Operator-Key` alias — stamps the live admin role — the request tenant's, read from that tenant's policy (`policies`) — plus an operator bit (`auth.WithOperator`) that `RequireAdmin` honors even under a nil policy — a full-access break-glass credential, audit-logged at Info (the audit line goes through the default logger, so it carries the request scope's `client.address`). A presented-but-wrong operator key is logged at `WARN` and counted by `wavehouse_auth_operator_key_failures_total` — a probing signal on the most privileged credential — then falls through like any unauthenticated request (the middleware never rejects). - **context.go** — request-context accessors and their setters for the role, claims, and token error (`RoleFromContext`, `ClaimsFromContext`, `AuthErrorFromContext`, and the matching `With*` helpers). ### `cache/` — Query Cache @@ -135,7 +135,7 @@ The SSE fan-out, factored out of `api/` so the delivery hot path ([#294](https:/ - **config.go** — Loads *boot* configuration from a YAML file with environment variable overrides (using [cleanenv](https://github.com/ilyakaznacheev/cleanenv)); every key has a `WH_`-prefixed env var. Boot config is only what can't change under a running process — the implementation each layer runs on, the process's `roles`, resource sizing, listeners, observability exporters, the settings-directory path, and the platform's own secrets (`cache.redis.password`, `auth.operator_key`). Everything tenant-tunable lives in the settings directory (`settings/`), a tenant's secrets included: `clickhouse.password` beside its user and `auth.jwt_secret` beside its verifier wiring, so each reloads with what it belongs to ([#529](https://github.com/Wave-RF/WaveHouse/issues/529)). Both sources are strict: `Load` refuses to boot naming every YAML key the struct doesn't declare (`strict.go`) and every `WH_*` environment variable no field binds (`check.go`), so a tunable that moved to the settings directory can't be read, ignored, and believed. Boot is the validator for this half — there is no dry-run command. See [Configuration Reference](/configuration). - **check.go** — `rejectUnboundEnv` is the environment half of the strict loader: `unboundEnv` walks the struct's `env` tags (plus the two process-level names, `WH_CONFIG` and `WH_LOG_LEVEL`) against the environment; `CheckDataDir` probes `data_dir` — run by `main` right after `Load` when `Config.NeedsDataDir` says a selected backend keeps state there, so an unusable `data_dir` refuses boot before anything dials out. It refuses an empty value (reachable through `WH_DATA_DIR=`) outright rather than probing the working directory; a path that exists and is not a directory; a dangling symlink at `data_dir` or any component above it (the walk to the nearest existing ancestor uses `Lstat`, so a failed mount is not skipped over as "does not exist"); and a directory the process cannot write to — or, when it does not exist, an unwritable nearest ancestor — probed by creating and removing one temp file. A permission denial, on the probe or on reaching the path through a parent without search permission, carries the UID-65532 hint, since a bind mount owned by root is the typical cause. -- **backends.go** — the `.backend` keys: one string type per layer (`MQBackend`, `CacheBackend`, `DedupeBackend`, `CoordBackend`), each with its list of the backends this build has, and a `validate` per layer block (`checkBackend`, run by `validateBackends` in `Validate`, between `validateRoles` and `validateTopology`), which refuses a value not on the list and names the ones that are. One rule spans two layers: while `mq.backend` is `embedded`, `dedupe.lease` plus its own ceiling to the next whole second (`ceilSecond`) plus one more second must fit the embedded MQ's 2m duplicate window (`embeddedDuplicateWindow`), a cap of 59s (`maxEmbeddedLease`), because a client obeying the in-flight `503`'s `Retry-After` can republish as late as the lease plus that ceiling plus a second after the claim, since DynamoDB rounds a claim's expiry up to the second. A backend's own settings go in a `.` sub-block that is its `validate`'s case to check. `Distributed` reports whether the MQ is shared with other processes, `NeedsDataDir` whether a selected backend keeps state under `data_dir`, and `Warnings` returns what a valid configuration is still likely to get wrong — the combinations that are harmless or correct for one replica only (a shared MQ over a local cache or Pebble dedupe; under `nats`, `mq.max_bytes_gb` not applied; an `mq.nats` or `coord.nats` block its layer's backend ignores), a `cache.redis` block that is not read, certificate verification turned off — which `app.New` logs at `WARN`. `mq.backend` has two values, `embedded` and `nats` (`MQNATS`), and `nats` reads the `mq.nats` sub-block, which `MQ.validate` checks only when it is selected. `coord.backend` takes `nats` (`CoordNATS`) too, whose `coord.nats` block holds only the bucket name: the leases ride `mq.nats`'s connection. +- **backends.go** — the `.backend` keys: one string type per layer (`MQBackend`, `CacheBackend`, `DedupeBackend`, `CoordBackend`), each with its list of the backends this build has, and a `validate` per layer block (`checkBackend`, run by `validateBackends` in `Validate`, between `validateRoles` and `validateTopology`), which refuses a value not on the list and names the ones that are. One rule spans two layers: while `mq.backend` is `embedded`, `dedupe.lease` plus its own ceiling to the next whole second (`ceilSecond`) plus one more second must fit the embedded MQ's 2m duplicate window (`embeddedDuplicateWindow`), a cap of 59s (`maxEmbeddedLease`), because a client obeying the in-flight `503`'s `Retry-After` can republish as late as the lease plus that ceiling plus a second after the claim, since DynamoDB rounds a claim's expiry up to the second. A backend's own settings go in a `.` sub-block that is its `validate`'s case to check. `Distributed` reports whether the MQ is shared with other processes, `NeedsDataDir` whether a selected backend keeps state under `data_dir`, and `Warnings` returns what a valid configuration is still likely to get wrong — the combinations that are harmless or correct for one replica only (a shared MQ over a local cache or Pebble dedupe; under `nats`, `mq.max_bytes_gb` not applied; an `mq.nats` or `coord.nats` block its layer's backend ignores), a `cache.redis` block that is not read, certificate verification turned off, a `profiling.addr` that is not loopback (`profiling.go`), a `/0` entry in `server.trusted_proxies` (`proxies.go`, which also parses that list) — which `app.New` logs at `WARN`. `mq.backend` has two values, `embedded` and `nats` (`MQNATS`), and `nats` reads the `mq.nats` sub-block, which `MQ.validate` checks only when it is selected. `coord.backend` takes `nats` (`CoordNATS`) too, whose `coord.nats` block holds only the bucket name: the leases ride `mq.nats`'s connection. - **cache_redis.go** — `CacheRedisConfig`, the `cache.redis` sub-block, and its checks: an address (exactly one in `standalone` mode, which dials only the first), each `host:port` with a port from 1 to 65535 (a URL or `user:password@` form refused without repeating it, since it may hold a password), a known mode (`sentinel` is refused until [#656](https://github.com/Wave-RF/WaveHouse/issues/656)), `db` 0 in cluster mode, positive timeouts and sizes, a `timeout` and `dial_timeout` of at most 1 s each (boot and `Close` each wait out a dial: a connect and a handshake bounded by `dial_timeout`, and a cluster's topology read bounded by the larger of the two), a `version_ttl` of at least 2 s, and a `compress_min_bytes` that is not negative (`0` never compresses); its defaults are in `defaults()` with the rest. `CacheRedisTLS.Config` builds the `tls.Config`, reading the files; `Validate` calls it so an unreadable file refuses boot, and `internal/app` calls it again to build the connection. A TLS key set while `tls.enabled` is off is an error rather than a plaintext connection. - **mq_nats.go** — the two nats blocks and their rules, apart from backends.go for the e2e coverage gate as `cache_redis.go` is: `MQNATSConfig` (URLs, file-path-only credentials, TLS, and the topology to expect; its defaults, `defaultMQNATS`, are in `defaults()`) and its checks, `CoordNATSConfig` (the bucket name), the nats lines of `Warnings` (`natsWarnings`) and the two cross-layer rules `validateTopology` delegates (`validateNATSTopology`: `coord.backend=nats` needs `mq.backend=nats`, a process running `ingest` on a shared queue needs `coord.backend=nats`, and a process running only `sweeper` there is refused). - **config.go**, roles — `roles` (`[]Role`: `api`, `ingest`, `sweeper`; `AllRoles` by default; `Has(Role)`) picks which components `internal/app` wires, and `instance_id` names the process (`-<8 hex>` when empty, resolved in `Load`; a field on every log line once loaded, `service.instance.id` in the telemetry, and the holder a NATS lease's value names). `validateRoles` refuses an empty list, an empty entry, an unknown or a repeated role; `validateTopology` refuses a role set the backends cannot serve: any split over the embedded MQ, `coord.backend=nats` without `mq.backend=nats` (the leases ride its connection), `mq.backend=nats` with `coord.backend=local` in a process running `ingest` (the ingest processes share the queue through leases), a process running only `sweeper` under `mq.backend=nats` (it would run nothing), and a process with exactly one of `api` and `ingest` over a local cache. `NeedsDataDir` counts Pebble only for a process running `api`, and the cache and dedupe warnings are skipped without `api`, since only that role opens a cache it reads or a dedupe store. @@ -197,8 +197,8 @@ The **only** package that imports NATS/JetStream — a `depguard` rule in `.gola - **provider.go** — `InitProvider(ctx, Identity, ProviderConfig)` wires the OTel pipeline, every signal over one resource built from the `Identity`. Each output is independently gated; the W3C TraceContext + Baggage propagator is always installed (cheap, harmless when traces are off). Returns `(shutdown, promHandler http.Handler, err)` — `promHandler` is non-nil only when `PrometheusEnabled` is true and reads from a *private* `prometheus.Registry` to avoid leaking the process/Go collectors that `prometheus.DefaultRegisterer` auto-registers. OTLP-metrics push (`MetricsEnabled`) and Prometheus exposition (`PrometheusEnabled`) are independent: either, both, or neither may be set, and any combination produces a single MeterProvider feeding the active readers. The OTLP exporters (traces / metrics-OTLP / logs) take their endpoint from the `OTEL_EXPORTER_OTLP_*` env vars, not from `ProviderConfig`, and are only constructed when their signal is enabled, so Prometheus-only operation dials nothing even with those vars set. Provider init in `internal/app` runs whenever `otel.enabled` OR `prometheus.enabled` is true, so Prometheus-only operation (Alloy/scrape, no collector) is a first-class mode. - **cardinality.go** — `tableView`, the view every `wavehouse_*` instrument passes through on the meter provider, bounding the `table` attribute (`TableAttr`, [#538](https://github.com/Wave-RF/WaveHouse/issues/538)): with `ProviderConfig.AggregateTables` (`prometheus.per_table_metrics: false`) it drops the attribute, so each metric reports one series summed over every table, on every reader; otherwise a `tableCounter` keeps it and logs one `WARN` naming the setting once the metrics have carried more than `TableWarnThreshold` (500) distinct tables. Call sites keep recording `table` unconditionally. - **identity.go** — `Identity`, the process's name in its telemetry ([#772](https://github.com/Wave-RF/WaveHouse/issues/772)): service name, version, commit, build time, `instance_id` and roles, built by `internal/app`'s `identity` (`wire_identity.go`). `newResource` turns it into the resource (`service.name`, `service.version`, `service.instance.id`, `wavehouse.roles` (sorted by `identity`), `wavehouse.git_commit`, `wavehouse.build_time`, each left out when empty), then merges `resource.WithFromEnv`, so `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` override it key by key; a malformed `OTEL_RESOURCE_ATTRIBUTES` is logged, not fatal. `registerBuildInfo` exports `wavehouse_build_info` (always `1`; `version`, `git_commit`, `build_time`, `roles`) on the meter provider, with no instance label, since the resource carries it. `LogAttrs` are the build fields and `instance_id` the OTLP-aware logger carries on every record; `cmd/wavehouse` puts `instance_id` on the boot logger once the config is loaded. -- **logger.go** — `NewLogger(component, level, isJSON, otlpSampleRate)` produces a slog logger that fans out to stdout (always 100%) and the OTLP log exporter (DEBUG/INFO sampled at `otlpSampleRate`, WARN/ERROR always 100% as a non-configurable safety floor). Both legs read the same `level` (`WH_LOG_LEVEL`): stdout through its handler options, OTLP through `levelGate`, which sits ahead of the sampler because neither the otelslog bridge nor the log SDK filters by level, so a record below the level is never built for either. `TraceHandler` injects `trace_id`/`span_id` from the active span when one exists, and `request_id` and `tenant` from the request scope; `cmd/wavehouse` wraps its boot logger in it too, so stdout carries them whether or not OTLP logs are on. `otlpSamplerFn` is exposed (lowercase) for unit testing the per-level rate logic without driving through the slogmulti middleware. -- **request.go** — The request scope: `WithRequest(ctx, id)` puts it in the context, `SetRequestTenant` fills in the tenant once it is resolved (in place, so a record logged through a context taken earlier, the panic recoverer's, carries it too) and sets it on the span in the context, and `AnnotateSpan` copies both onto the server span. `request_id` and `tenant` are the keys on records and spans alike; a record that names its own tenant keeps it. They are never metric labels. +- **logger.go** — `NewLogger(component, level, isJSON, otlpSampleRate)` produces a slog logger that fans out to stdout (always 100%) and the OTLP log exporter (DEBUG/INFO sampled at `otlpSampleRate`, WARN/ERROR always 100% as a non-configurable safety floor). Both legs read the same `level` (`WH_LOG_LEVEL`): stdout through its handler options, OTLP through `levelGate`, which sits ahead of the sampler because neither the otelslog bridge nor the log SDK filters by level, so a record below the level is never built for either. `TraceHandler` injects `trace_id`/`span_id` from the active span when one exists, and `request_id`, `client.address` and `tenant` from the request scope; `cmd/wavehouse` wraps its boot logger in it too, so stdout carries them whether or not OTLP logs are on. `otlpSamplerFn` is exposed (lowercase) for unit testing the per-level rate logic without driving through the slogmulti middleware. +- **request.go** — The request scope: `WithRequest(ctx, id)` puts it in the context, `SetRequestClient` and `SetRequestTenant` fill in the client address and the tenant once they are resolved (in place, so a record logged through a context taken earlier, the panic recoverer's, carries them too) and set them on the span in the context, and `AnnotateSpan` copies them onto the server span. `request_id`, `client.address` (OpenTelemetry's own name for it) and `tenant` are the keys on records and spans alike; a record that names its own tenant keeps it. They are never metric labels. - **metrics.go** — `RegisterSystemMetrics(mqStats, pebbleStats)` registers observable gauges for the MQ's NATS connections and in-msgs (the embedded server's, or under `nats` this process's client connection), and Pebble dedupe storage stats. Both are functions read on every scrape — `mqStats` a `func() (MQStats, error)` (`mq.Broker.Stats` in production; nil skips the MQ gauges), `pebbleStats` a `func() map[string]int64` (`dedupe.Embedded.Stats`, the one instance's figures; nil, or a nil map while it is closed, skips the Pebble gauges) — this package never holds the NATS server or a store. Wired in `internal/app` after the providers are up. - **pprof.go** — `PprofHandler`, `net/http/pprof`'s endpoints under `/debug/pprof/` on a mux of its own, mounted only on the profiling listener (`internal/app`'s `wire_profiling.go`). The package's import also registers them on `http.DefaultServeMux`, which nothing in WaveHouse serves. - **tracer.go** — W3C TraceContext propagation over message headers (`InjectHeaders` / `ExtractHeaders` on a plain `map[string][]string`, the shape NATS and HTTP headers share) — `internal/mq` injects on every publish and extracts onto the delivered `Message.Ctx` on both the `Subscribe` path and the worker's `Consumer` path. The ingest worker reads it (`internal/ingest/trace.go`: its flush span links to the span each row was published under, and its INSERT span is the flush span's child, [#774](https://github.com/Wave-RF/WaveHouse/issues/774)); the SSE hub bridge forwards the bytes and starts no span. diff --git a/docs/src/content/docs/configuration.mdx b/docs/src/content/docs/configuration.mdx index b115c6cd..439a7197 100644 --- a/docs/src/content/docs/configuration.mdx +++ b/docs/src/content/docs/configuration.mdx @@ -124,6 +124,7 @@ Boot logs each of these at `WARN` rather than refusing. The first two are right - **`mq.nats` set with `mq.backend=embedded`**, or **`coord.nats` set with `coord.backend=local`**: the block is ignored. - **`cache.redis.addrs` set with `cache.backend=local`**, in a process running `api`: the block is ignored. - **`cache.redis.tls.insecure_skip_verify` on**, in a process running `api`: the cache accepts any certificate (see [Cache](#cache)). +- **`server.trusted_proxies` with a `/0` entry** (`0.0.0.0/0`, `::/0`), in every process, one per entry: every peer of that entry's address family is trusted (`0.0.0.0/0` every IPv4 peer, `::/0` every IPv6 one), so a client of that family that reaches `server.port` directly picks its own `client.address` and joins its `traceparent` to your traces (see [Reverse proxy](/reverse-proxy#client-address-and-trace-context)). - **`profiling.enabled` with a `profiling.addr` that is not loopback** (an empty host, `0.0.0.0`, a pod address), in every process: the unauthenticated pprof endpoints are reachable beyond the host (see [Profiling](#profiling)). - **`mq.backend=nats` and a tenant with dedupe on whose finite `dedupe.retention` is under the partitions' `duplicate_window`**, in a process running `api`, at boot and after every reload: see [Deployment → Create the topology](/deployment#create-the-topology). @@ -158,6 +159,7 @@ Boot refuses a role set the selected backends cannot serve: | --- | --- | ------- | ----------- | | `server.port` | `WH_SERVER_PORT` | `8080` | HTTP server listen port. | | `server.shutdown_timeout` | `WH_SERVER_SHUTDOWN_TIMEOUT` | `10` | Seconds the drain phase of a stop may take: in-flight requests and ingest batches finish (open SSE streams end immediately). The release of the stores (5s) and the telemetry flush (3s) have fixed budgets on top, so a stop takes at most this plus 8s; see [Stopping](/deployment#stopping). | +| `server.trusted_proxies` | `WH_SERVER_TRUSTED_PROXIES` | *(empty)* | The reverse proxies whose `X-Forwarded-For` names the client and whose `traceparent` a request's trace joins: CIDRs or bare addresses, a YAML list or a comma-separated variable. Empty trusts no peer, so the client is the connection's own address and an incoming trace is linked, not joined. An entry that is neither, or an IPv4-mapped prefix (`::ffff:…`) shorter than `/96`, refuses boot; a mapped prefix of `/96` or longer is read as the IPv4 CIDR it maps. See [Reverse proxy → Client address and trace context](/reverse-proxy#client-address-and-trace-context). | The server speaks **plain HTTP** — there is no inbound-TLS setting. Terminate TLS at a [reverse proxy](/reverse-proxy#tls) for internet-facing deployments. (The settings directory's `clickhouse.http_scheme` and `clickhouse.tls` are the *outbound* WaveHouse → ClickHouse hops, unrelated to your clients' TLS.) @@ -359,6 +361,7 @@ instance_id: "" # on every log line and span; empty = -<8 hex server: port: 8080 shutdown_timeout: 10 + trusted_proxies: [] # CIDRs or addresses whose X-Forwarded-For and traceparent are believed; empty = none clickhouse: # addr, ports, database, username, password, query_timeout, tls, headers @@ -467,6 +470,8 @@ WH_INSTANCE_ID= WH_SERVER_PORT=8080 WH_SERVER_SHUTDOWN_TIMEOUT=10 +# Comma-separated CIDRs or addresses; empty = trust no peer. +WH_SERVER_TRUSTED_PROXIES= WH_CH_MAX_TOTAL_CONNS=0 diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index a7b14540..0f85108e 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -763,7 +763,7 @@ WaveHouse **pushes** to an OTel collector; scraping-style pipelines (Promtail/Gr ### Following one request or one tenant -Every log record written through a request's context (as the request path logs) carries `request_id`, and `tenant` once the request's tenant is known: the `X-Tenant-ID` a `/v1` route resolves, or the `?tenant=` an ops route names, from authentication on (tenant `0` for any other ops route that names none, reads, raw SQL and the schema refresh alike; a settings reload that names none acts on every tenant and carries none). The request's server span carries the same two attributes under the same names, so one query finds both (SSE streams on `/v1/stream`, the probes, `/v1/health` and the metrics path open no server span, so their correlation is on log lines only: `/v1/stream` and `/v1/health` carry `request_id` and `tenant`, while the bare probes and the metrics path resolve no tenant and carry `request_id` alone). A record that already names a tenant keeps its own. The request id is the `X-Request-Id` the request arrived with, when it was sent once and is 1 to 128 printable ASCII characters with no spaces, and otherwise a fresh random 32-hex-character id; either way the response carries it back in `X-Request-Id`, so a log line can be joined to a proxy's access log or to a client's report. Have the proxy set the header (see [Reverse proxy](/reverse-proxy#header-and-auth-forwarding)). Neither is a label on the HTTP request metrics: with many tenants per process that would be a series per tenant per route. The ingest worker's lines are not request-scoped and carry the tenant only where they name it. +Every log record written through a request's context (as the request path logs) carries `request_id` and `client.address`, and `tenant` once the request's tenant is known: the `X-Tenant-ID` a `/v1` route resolves, or the `?tenant=` an ops route names, from authentication on (tenant `0` for any other ops route that names none, reads, raw SQL and the schema refresh alike; a settings reload that names none acts on every tenant and carries none). The request's server span carries the same attributes under the same names, so one query finds both (SSE streams on `/v1/stream`, the probes, `/v1/health` and the metrics path open no server span, so their correlation is on log lines only: `/v1/stream` and `/v1/health` carry `request_id`, `client.address` and `tenant`, while the bare probes and the metrics path resolve no tenant and carry the other two). `client.address` is the connection's address unless it comes from a proxy listed in [`server.trusted_proxies`](/configuration#server), which makes it the client the proxy names in `X-Forwarded-For`; a request from a listed proxy also joins the proxy's trace when it sends a `traceparent` (see [Reverse proxy → Client address and trace context](/reverse-proxy#client-address-and-trace-context)). A connection whose own address cannot be read has none: its log records leave `client.address` out, and its server span's `client.address` is empty. A record that already names a tenant keeps its own. The request id is the `X-Request-Id` the request arrived with, when it was sent once and is 1 to 128 printable ASCII characters with no spaces, and otherwise a fresh random 32-hex-character id; either way the response carries it back in `X-Request-Id`, so a log line can be joined to a proxy's access log or to a client's report. Have the proxy set the header (see [Reverse proxy](/reverse-proxy#header-and-auth-forwarding)). Neither is a label on the HTTP request metrics: with many tenants per process that would be a series per tenant per route. The ingest worker's lines are not request-scoped and carry the tenant only where they name it. An ingest request's trace continues past the queue. The handler's span (`IngestHandler.Handle`) rides in the queue headers of every row the request published, and the worker's flush of a batch opens a span of its own, `IngestWorker.Flush`, linked to the span of each request whose rows it holds (a batch holds rows from many requests, so the link, not a parent, is what ties them; a request's trace shows no worker span inline, and the flush's `producers` attribute counts every request whose rows it holds, of which at most the first 128 are linked), with `tenant`, `table` and `rows`. Under it, `clickhouse.insert` wraps each `INSERT` with the standard database attributes (`db.system`, `db.name`, `db.operation`, `db.sql.table`, `server.address`, `server.port`) and `rows`, and records the error when the `INSERT` fails, refused by ClickHouse or ClickHouse unreachable or timed out. Each flush is a trace of its own, sampled at `otel.traces.sample_rate` independently of the requests it links to, so below `1.0` either end of a link may be missing; to keep them together, sample at 100% at the source and downsample at the collector. With the `api` and `ingest` roles in separate processes the link crosses them, so a slow or failed insert reads back to every request that produced its rows. diff --git a/docs/src/content/docs/reverse-proxy.mdx b/docs/src/content/docs/reverse-proxy.mdx index 7dfe9ac4..5ebd634a 100644 --- a/docs/src/content/docs/reverse-proxy.mdx +++ b/docs/src/content/docs/reverse-proxy.mdx @@ -1,6 +1,6 @@ --- title: "Behind a reverse proxy" -description: "Running WaveHouse behind nginx, Caddy, or a Cloudflare Tunnel: TLS, request-body limits, SSE buffering and timeouts, header/auth forwarding, and health probes." +description: "Running WaveHouse behind nginx, Caddy, or a Cloudflare Tunnel: TLS, request-body limits, SSE buffering and timeouts, header/auth forwarding, the client address and trace context, and health probes." cloudCta: body: "Proxy buffering silently breaking SSE is the single most common way a healthy WaveHouse looks broken in production. WaveHouse Cloud terminates TLS at an edge already configured for streaming, so the proxy configuration on this page is not yours to do." sidebar: @@ -9,7 +9,7 @@ sidebar: import { Tabs, TabItem } from "@astrojs/starlight/components"; -WaveHouse serves plain HTTP on `:8080` and is meant to run **behind a reverse proxy, CDN, or tunnel** for any internet-facing deployment. It deliberately does *not* terminate TLS or manage a server certificate, rate-limit by IP, or mitigate slow-connection attacks — those are the proxy's job, and pushing them out keeps WaveHouse from growing a second, redundant set of knobs you'd have to tune in two places. This page covers everything that behaves differently behind a proxy: TLS, request-body limits, Server-Sent Events, header forwarding, and health probes. +WaveHouse serves plain HTTP on `:8080` and is meant to run **behind a reverse proxy, CDN, or tunnel** for any internet-facing deployment. It deliberately does *not* terminate TLS or manage a server certificate, rate-limit by IP, or mitigate slow-connection attacks — those are the proxy's job, and pushing them out keeps WaveHouse from growing a second, redundant set of knobs you'd have to tune in two places. This page covers everything that behaves differently behind a proxy: TLS, request-body limits, Server-Sent Events, header forwarding, the client address and trace context, and health probes. ## Division of responsibility @@ -196,10 +196,12 @@ These limit the *whole* request regardless of traffic, so no keepalive extends t - **`X-Operator-Key`** (optional) — only relevant if you present the [operator key](/access-control#operator-key) via this alias header instead of `Authorization: Operator `. Custom request headers are forwarded by default, but confirm your proxy doesn't strip it — and note nginx silently drops header names containing **underscores** (this one uses hyphens, so it's fine as named). The `Authorization` form needs none of this. - **`X-Tenant-ID`** — **strip it at the edge unless you are deliberately using it.** It selects the [tenant](/deployment#multi-tenant-deployments) (absent means tenant `0`), and a settings directory that isn't nested defines no tenant other than `0`, so a gateway or mesh that stamps `X-Tenant-ID` with any other value turns every `/v1` route outside `/v1/ops/*` into a `404` (a `400` when the value is not a tenant id at all, a dotted hostname, say) — the SDK's `/v1/health` reachability ping included. It is resolved before authentication and never derived from the token, so a proxy that owns tenant selection must **replace** a client-supplied value, not add to it: two `X-Tenant-ID` lines are refused with `400`. That is a stricter version of the append-vs-replace hazard of `Content-Type` below — here even two agreeing lines are refused. Over a [nested settings directory](/deployment#the-nested-settings-directory) the proxy owns the tenant: derive it from the host or path and set it on **every** request, the CORS `OPTIONS` preflight included — the browser sends no custom header there, and WaveHouse reads the preflight's tenant the same way ([per-tenant CORS](/deployment#multi-tenant-deployments)). The name is hyphenated, so nginx's underscore rule does not apply. - **`X-Request-Id`** (optional) — set it to your proxy's request id (nginx `$request_id`, Envoy's `x-request-id`) and WaveHouse logs and traces the request under the same id and returns it on the response, so its log lines join your access log; **replace** a client-supplied value rather than appending, since two lines are ignored and a fresh id minted. An id longer than 128 characters, or with spaces or non-ASCII characters, is replaced the same way ([details](/deployment#following-one-request-or-one-tenant)). -- **`X-Forwarded-For` / `X-Forwarded-Proto` / `Host`** — set these for your own logs and any upstream that reads them. WaveHouse does not currently derive a client IP from `X-Forwarded-For` (see the caution below); forwarding it is good hygiene and is what the trusted-proxy client-IP work ([#333](https://github.com/Wave-RF/WaveHouse/issues/333)) will consume. +- **`X-Forwarded-For`** — have the proxy **append** the peer it received the request from (nginx `$proxy_add_x_forwarded_for`; Caddy's `reverse_proxy` sets it by default) and list the proxy in [`server.trusted_proxies`](/configuration#server): WaveHouse then logs and traces the request under the client's address, read as [Client address and trace context](#client-address-and-trace-context) describes. From a peer that is not listed the header is ignored. +- **`traceparent` / `tracestate`** — a request from a listed proxy joins the trace these name. Most proxies forward the client's, with or without tracing of their own (nginx's `ngx_otel_module` and Caddy's `tracing` continue it), so have the proxy drop the client's or replace them with its own. The nginx and Caddy [example configurations](#example-configurations) drop them; see [below](#client-address-and-trace-context) for each proxy. +- **`X-Forwarded-Proto` / `Host`** — set these for your own logs and any upstream that reads them. WaveHouse reads neither. :::caution[Don't expose `:8080` directly] -WaveHouse does **not** derive a client IP from forwarded headers — it does no per-IP logic (rate limiting and IP allow/deny are the proxy's job) and does not trust `X-Forwarded-For` / `X-Real-IP` / `True-Client-IP` to rewrite the connection's source address. So a forged forwarded header has no effect on WaveHouse, and `r.RemoteAddr` (what OpenTelemetry records as the peer) is the honest immediate peer — your proxy, when one is in front. Still, don't expose `:8080` to untrusted clients: bind WaveHouse to a private interface or firewall the port so the proxy is the only path in. Capturing the real client IP in WaveHouse's own traces and logs — trusted-proxy-aware, so it can't be spoofed — is tracked in [#333](https://github.com/Wave-RF/WaveHouse/issues/333). +WaveHouse never rewrites the connection's source address from forwarded headers, and reads `X-Forwarded-For` only from a peer in [`server.trusted_proxies`](/configuration#server); `X-Real-IP` and `True-Client-IP` are never read. It does no per-IP logic either: rate limiting and IP allow/deny are the proxy's job. Still, don't expose `:8080` to untrusted clients: bind WaveHouse to a private interface or firewall the port so the proxy is the only path in. A trusted range wide enough to cover a client's own address lets that client choose the address it is logged under. ::: - **`Content-Type`** — forward it **verbatim**; ingest reads the format from it and refuses anything it cannot read ([details](/api#post-v1ingesttabletable--ingest-data)). Appending rather than replacing is safe *only* if the proxy sends a second header **line** — those are resolved together and accepted when they agree. A proxy that **merges** duplicates into one comma-joined value (Envoy's `append: true`, and some WAF rewrite rules) produces `application/json, application/json`, which is a `415` on every ingest **even though both halves agree**: a comma-joined value is refused unless the value as a whole still parses as one media type. Replacing it is worse than merging, because it fails silently: an NDJSON batch declared `application/json` is read as the single object it starts with and the remaining lines are dropped behind a `200` ([#561](https://github.com/Wave-RF/WaveHouse/issues/561)) — no error to alert on. If ingest starts returning `415` fleet-wide after a proxy change, look here first. @@ -208,6 +210,31 @@ WaveHouse does **not** derive a client IP from forwarded headers — it does no - **Response caching** — don't let a shared cache (a CDN or a caching proxy) store responses from the authenticated routes. What a read returns depends on the caller's role and claims, and a pipe read — a `GET`, so cacheable by default — sends no `Cache-Control` and no `Vary: Authorization` to say so. A cache that follows RFC 9111 already refuses to store the response to a request carrying `Authorization`; a credential sent as `?token=` or `X-Operator-Key` is not covered by that rule, so exclude those routes explicitly. The `Vary: X-Tenant-ID` on a [tenant-resolved](/deployment#multi-tenant-deployments) response keys it on the tenant only and says nothing about the caller. +## Client address and trace context + +Behind a proxy, the connection WaveHouse sees comes from the proxy, so on its own it can only log the proxy's address. List your proxies in [`server.trusted_proxies`](/configuration#server) (`WH_SERVER_TRUSTED_PROXIES`), as CIDRs or single addresses, and WaveHouse recovers the client's: + +```yaml +server: + trusted_proxies: [192.0.2.10, 198.51.100.0/28] # your proxies' addresses, or subnets only proxies use +``` + +List only addresses your proxies use. A range that also holds clients lets a client in it send its own `X-Forwarded-For` and `traceparent` and have them believed, and lets a forged hop pass as one more proxy. + +**How the client is chosen.** When the connection comes from a listed proxy, WaveHouse reads `X-Forwarded-For` from **right to left**, the order the proxies appended it in, and takes the first hop that is not itself a listed proxy. Every hop to the right of that one was written by a proxy you trust; everything to its left came from the client and may be forged, so a client that sends `X-Forwarded-For: 1.2.3.4` gets nothing for it, provided each listed proxy appends the address it received the request from, as nginx's `$proxy_add_x_forwarded_for` and Caddy's `reverse_proxy` do. A listed proxy that forwards the client's header without appending to it lets the client pick its own `client.address`. When the connection comes from anyone else, the client is the connection's own address and the header is ignored. Repeated `X-Forwarded-For` lines are read as one list, in order. A hop may carry a port or, for IPv6, brackets; IPv4-mapped IPv6 addresses read as IPv4. A hop that is not an address (`unknown`, an obfuscated identifier) ends the walk at the nearest proxy, since nothing past it can be read, and a header of listed proxies alone yields the leftmost of them. `X-Real-IP`, `True-Client-IP` and `Forwarded` are not read. + +**Where it shows up.** The address is the `client.address` attribute of the request's server span and a `client.address` field on every log record written through the request's context, next to `request_id` and `tenant` (see [Deployment → Following one request or one tenant](/deployment#following-one-request-or-one-tenant)). It replaces the value OpenTelemetry's HTTP instrumentation records on its own, which is the leftmost `X-Forwarded-For` entry whoever sent it. The span's `network.peer.address` is still the connection's address, your proxy. It is never a metric label. A connection whose own address cannot be read has no client address: its log records leave `client.address` out and its span's is empty (WaveHouse listens on TCP, where the address is always readable). + +**Trace context.** A proxy with OpenTelemetry tracing sends a W3C `traceparent` (and `tracestate`), and a request from a listed proxy **joins that trace**: its server span is a child of the proxy's span, and its log records carry the proxy's `trace_id`. A `traceparent` from any other peer, a client calling WaveHouse directly or one whose proxy is not listed, starts a new trace that **links** to the one it named instead of joining it, so a client cannot write spans into your traces or pick the trace id its requests are logged under. With `server.trusted_proxies` empty, no incoming trace is joined. + +Listing a proxy trusts the `traceparent` it forwards as well as its `X-Forwarded-For`, and most proxies forward the client's: one without tracing passes it through, and most with tracing continue the client's trace rather than start their own. Either way a listed proxy lets a client choose the trace its request joins, so have the proxy drop the client's `traceparent` and `tracestate` before anything reads them: + +- **nginx** without `ngx_otel_module`: `proxy_set_header traceparent "";` and `proxy_set_header tracestate "";` (an empty value drops the header), as in the [example](#example-configurations). With the module, `otel_trace_context inject` starts nginx's own trace in place of the client's; its default, `ignore`, forwards the client's untouched and `propagate` continues it. +- **Caddy** without `tracing`: `header_up -traceparent` and `header_up -tracestate` on each `reverse_proxy`, as in the example. `tracing` reads the headers before `header_up` runs, so with it, remove them with `request_header -traceparent` and `request_header -tracestate` inside a `route` that lists them ahead of `tracing`. +- **Cloudflare Tunnel**: a request-header Transform Rule that removes both. + +List the address each proxy connects to WaveHouse from: `127.0.0.1` for nginx or Caddy on the same host, as in the [example configurations](#example-configurations) (nginx's `$proxy_add_x_forwarded_for` appends the peer; Caddy's `reverse_proxy` sets the header by default), and likewise for a `cloudflared` running beside WaveHouse, behind which Cloudflare's edge has put the visitor's address in the header. A proxy whose upstream is `localhost`, like the Cloudflare Tunnel example's `http://localhost:8080`, may connect over IPv6, so list `::1` too: `trusted_proxies: [127.0.0.1, ::1]`. A proxy you left out is not an error; `client.address` just reads as the proxy's own address. Behind a load balancer and then an ingress, list both: each appends the peer it saw. When WaveHouse runs in a container on a bridge network, the peer is not `127.0.0.1` (with host networking the container shares the host's loopback, so `127.0.0.1` still applies): a proxy on the host reaches a published port from the container network's gateway (often `172.17.0.1`), and a proxy container from the address it has on that network. To find the address to list, send one request and read the span's `network.peer.address`, or `client.address` while the proxy is not yet listed. On Docker Desktop, rootless Docker, or with Docker's userland proxy, every connection to a published port can arrive from that gateway, outside clients included, so listing it would trust them all: publish the port on loopback only (`-p 127.0.0.1:8080:8080`), confirm that a direct connection from outside the host now fails (before Docker Engine 28.0.0, a host on the same layer-2 network can still reach a port published on loopback, so upgrade or run that check from such a host), and only then list the `network.peer.address` of a request sent through the proxy. + ## Fencing the admin surface Every admin-gated endpoint — raw SQL, pipe inspection, settings reload, schema discovery, DLQ stats — lives under the single `/v1/ops/` prefix, so one proxy rule covers the entire management surface: an nginx `location /v1/ops/ { deny all; }` on the public vhost (serving it only on an internal listener), or an IP allowlist on that prefix. If you deny the prefix outright, keep an internal listener that still serves it — `POST /v1/ops/settings/reload` with the [operator key](/access-control#operator-key) is the one HTTP path that re-adopts a fixed `policies.json` without a restart, and a blanket fence with no internal route removes that break-glass too. Under a [path prefix](#path-prefixes), fence `/v1/ops/` instead. This is belt-and-braces, not a requirement — the server's `RequireAdmin` gate remains the authorization boundary and denies non-admin callers on its own — but keeping the management surface off the public vhost removes it from unauthenticated reach entirely. @@ -260,6 +287,8 @@ server { proxy_set_header Host $host; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; + proxy_set_header traceparent ""; + proxy_set_header tracestate ""; } location / { @@ -267,6 +296,10 @@ server { proxy_set_header Host $host; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; + # Drop the client's trace context: trusted_proxies would join it. + # With ngx_otel_module, use otel_trace_context inject instead. + proxy_set_header traceparent ""; + proxy_set_header tracestate ""; } } ``` @@ -288,9 +321,17 @@ wavehouse.example.com { # response timeout, so quiet streams already survive. reverse_proxy /v1/stream* 127.0.0.1:8080 { flush_interval -1 + header_up -traceparent + header_up -tracestate } - reverse_proxy 127.0.0.1:8080 + # Drop the client's trace context: trusted_proxies would join it. + # With Caddy's tracing directive, strip it ahead of tracing instead + # (see Client address and trace context). + reverse_proxy 127.0.0.1:8080 { + header_up -traceparent + header_up -tracestate + } } ``` diff --git a/internal/api/clientip.go b/internal/api/clientip.go new file mode 100644 index 00000000..d6ccfae3 --- /dev/null +++ b/internal/api/clientip.go @@ -0,0 +1,121 @@ +package api + +import ( + "net/http" + "net/netip" + "strings" + + "github.com/Wave-RF/WaveHouse/internal/observability" +) + +// forwardedForHeader lists the hops a request passed through, each proxy +// appending the peer it received the request from. +const forwardedForHeader = "X-Forwarded-For" + +// trustedProxies are the peers whose X-Forwarded-For and traceparent are +// believed (server.trusted_proxies). Empty trusts no peer. +type trustedProxies []netip.Prefix + +func (t trustedProxies) trusts(a netip.Addr) bool { + if !a.IsValid() { + return false + } + for _, p := range t { + if p.Contains(a) { + return true + } + } + return false +} + +// trustsPeer reports whether the request arrived from a trusted proxy. +func (t trustedProxies) trustsPeer(r *http.Request) bool { + return t.trusts(peerAddr(r)) +} + +// clientAddr is the address the request came from. From an untrusted peer +// it is the peer, whatever the headers say. From a trusted proxy it walks +// X-Forwarded-For right to left, the order the proxies appended it in, and +// takes the first hop that is not itself a trusted proxy: every hop to its +// right was written by a proxy that is trusted, and everything to its left +// by the client, which may have forged it. A hop that is not an address +// stops the walk at the nearest proxy, since nothing past it can be read; a +// header of trusted hops alone yields the leftmost of them. +func (t trustedProxies) clientAddr(r *http.Request) netip.Addr { + client := peerAddr(r) + if !t.trusts(client) { + return client + } + values := r.Header.Values(forwardedForHeader) + for i := len(values) - 1; i >= 0; i-- { + hops := strings.Split(values[i], ",") + for j := len(hops) - 1; j >= 0; j-- { + hop := strings.TrimSpace(hops[j]) + if hop == "" { + continue + } + a, ok := parseHop(hop) + if !ok { + return client + } + client = a + if !t.trusts(a) { + return client + } + } + } + return client +} + +// peerAddr is the connection's own address, r.RemoteAddr without its port; +// invalid when RemoteAddr is not an IP address (a Unix socket, say). +func peerAddr(r *http.Request) netip.Addr { + if ap, err := netip.ParseAddrPort(r.RemoteAddr); err == nil { + return normalizeAddr(ap.Addr()) + } + if a, err := netip.ParseAddr(r.RemoteAddr); err == nil { + return normalizeAddr(a) + } + return netip.Addr{} +} + +// parseHop reads one X-Forwarded-For entry: an address, optionally with a +// port (some proxies append one) or, for IPv6, in brackets. +func parseHop(hop string) (netip.Addr, bool) { + if a, err := netip.ParseAddr(hop); err == nil { + return normalizeAddr(a), true + } + if ap, err := netip.ParseAddrPort(hop); err == nil { + return normalizeAddr(ap.Addr()), true + } + if strings.HasPrefix(hop, "[") && strings.HasSuffix(hop, "]") { + if a, err := netip.ParseAddr(hop[1 : len(hop)-1]); err == nil { + return normalizeAddr(a), true + } + } + return netip.Addr{}, false +} + +// normalizeAddr reads an IPv4-mapped IPv6 address as IPv4, so an IPv4 +// prefix matches it, and drops an IPv6 zone, which a prefix never matches. +func normalizeAddr(a netip.Addr) netip.Addr { + return a.Unmap().WithZone("") +} + +// clientAddress records the request's client address in the request scope, +// so its log records and server span carry client.address. It runs after +// requestID, which opens the scope. An address it cannot resolve is recorded +// as empty, so the span does not keep the one otelhttp read from +// X-Forwarded-For. +func clientAddress(t trustedProxies) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var addr string + if a := t.clientAddr(r); a.IsValid() { + addr = a.String() + } + observability.SetRequestClient(r.Context(), addr) + next.ServeHTTP(w, r) + }) + } +} diff --git a/internal/api/clientip_test.go b/internal/api/clientip_test.go new file mode 100644 index 00000000..87821ec5 --- /dev/null +++ b/internal/api/clientip_test.go @@ -0,0 +1,160 @@ +package api + +import ( + "net/http" + "net/http/httptest" + "net/netip" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/propagation" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" + + "github.com/Wave-RF/WaveHouse/internal/observability" +) + +func prefixes(t *testing.T, cidrs ...string) trustedProxies { + t.Helper() + out := make(trustedProxies, 0, len(cidrs)) + for _, c := range cidrs { + out = append(out, netip.MustParsePrefix(c)) + } + return out +} + +func TestTrustedProxies_ClientAddr(t *testing.T) { + t.Parallel() + proxies := prefixes(t, "10.0.0.0/8", "172.16.0.0/12", "fd00::/8") + tests := []struct { + name string + trusted trustedProxies + remote string + xff []string // every X-Forwarded-For line sent, in order + want string // "" = no address + }{ + {name: "no proxies trusted: the peer, headers ignored", remote: "203.0.113.9:4000", xff: []string{"198.51.100.1"}, want: "203.0.113.9"}, + {name: "untrusted peer's header ignored", trusted: proxies, remote: "203.0.113.9:4000", xff: []string{"198.51.100.1, 10.0.0.1"}, want: "203.0.113.9"}, + {name: "trusted peer, no header: the peer", trusted: proxies, remote: "10.0.0.1:4000", want: "10.0.0.1"}, + {name: "trusted peer, empty header: the peer", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{""}, want: "10.0.0.1"}, + {name: "one proxy", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"198.51.100.1"}, want: "198.51.100.1"}, + {name: "spoofed leftmost entry is skipped", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"1.2.3.4, 198.51.100.1"}, want: "198.51.100.1"}, + {name: "spoofed trusted-looking entry left of the client", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"10.9.9.9, 198.51.100.1"}, want: "198.51.100.1"}, + {name: "chained proxies", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"1.2.3.4, 198.51.100.1, 172.16.5.5, 10.2.2.2"}, want: "198.51.100.1"}, + {name: "chained proxies across header lines", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"1.2.3.4, 198.51.100.1", "172.16.5.5", "10.2.2.2"}, want: "198.51.100.1"}, + {name: "every hop trusted: the leftmost", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"10.3.3.3, 10.2.2.2"}, want: "10.3.3.3"}, + {name: "empty elements skipped", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"198.51.100.1, , 10.2.2.2,"}, want: "198.51.100.1"}, + {name: "unreadable hop stops at the nearest proxy", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"198.51.100.1, unknown, 10.2.2.2"}, want: "10.2.2.2"}, + {name: "unreadable last hop: the peer", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"198.51.100.1, _hidden"}, want: "10.0.0.1"}, + {name: "hop with a port", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"198.51.100.1:5555"}, want: "198.51.100.1"}, + {name: "IPv6 peer and client", trusted: proxies, remote: "[fd00::1]:4000", xff: []string{"2001:db8::7"}, want: "2001:db8::7"}, + {name: "bracketed IPv6 hop", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"[2001:db8::7]"}, want: "2001:db8::7"}, + {name: "bracketed IPv6 hop with a port", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"[2001:db8::7]:443"}, want: "2001:db8::7"}, + {name: "IPv6 hop inside a trusted prefix is a proxy", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"2001:db8::7, fd00::9"}, want: "2001:db8::7"}, + {name: "IPv4-mapped peer matches an IPv4 prefix", trusted: proxies, remote: "[::ffff:10.0.0.1]:4000", xff: []string{"198.51.100.1"}, want: "198.51.100.1"}, + {name: "IPv4-mapped hop reads as IPv4", trusted: proxies, remote: "10.0.0.1:4000", xff: []string{"::ffff:198.51.100.1"}, want: "198.51.100.1"}, + {name: "zoned peer", trusted: proxies, remote: "[fd00::1%eth0]:4000", xff: []string{"198.51.100.1"}, want: "198.51.100.1"}, + {name: "peer without a port", trusted: proxies, remote: "10.0.0.1", xff: []string{"198.51.100.1"}, want: "198.51.100.1"}, + {name: "peer not an address", trusted: proxies, remote: "@", xff: []string{"198.51.100.1"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil) + req.RemoteAddr = tt.remote + for _, v := range tt.xff { + req.Header.Add(forwardedForHeader, v) + } + got := tt.trusted.clientAddr(req) + if tt.want == "" { + assert.False(t, got.IsValid(), got.String()) + return + } + assert.Equal(t, tt.want, got.String()) + }) + } +} + +func TestTrustedProxies_TrustsPeer(t *testing.T) { + t.Parallel() + proxies := prefixes(t, "10.0.0.0/8") + req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil) + req.RemoteAddr = "10.1.2.3:80" + assert.True(t, proxies.trustsPeer(req)) + assert.False(t, trustedProxies(nil).trustsPeer(req)) + req.RemoteAddr = "203.0.113.9:80" + assert.False(t, proxies.trustsPeer(req)) + req.RemoteAddr = "garbage" + assert.False(t, proxies.trustsPeer(req)) +} + +// The router stamps the resolved client on the server span, replacing the +// leftmost X-Forwarded-For entry otelhttp would record, and joins an incoming +// traceparent only from a trusted proxy; from anyone else it is a link. +// Not parallel: otelhttp reads the global tracer provider and propagator. +func TestNewOpsRouter_ClientAddressAndTraceContinuation(t *testing.T) { + rec := tracetest.NewSpanRecorder() + savedTP, savedProp := otel.GetTracerProvider(), otel.GetTextMapPropagator() + otel.SetTracerProvider(sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec))) + otel.SetTextMapPropagator(propagation.TraceContext{}) + t.Cleanup(func() { + otel.SetTracerProvider(savedTP) + otel.SetTextMapPropagator(savedProp) + }) + + const parentTrace, parentSpan = "4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7" + r := NewOpsRouter(OpsDependencies{ + Health: &HealthHandler{}, + Version: NewVersionHandler("test", "test", "test"), + TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")}, + }) + tests := []struct { + name string + remote string + wantClient string + wantJoined bool + }{ + {name: "trusted proxy", remote: "10.0.0.1:4000", wantClient: "198.51.100.1", wantJoined: true}, + {name: "direct client", remote: "203.0.113.9:4000", wantClient: "203.0.113.9"}, + // otelhttp reads 198.51.100.1 from the header; an unparseable peer + // is trusted for nothing, so the span must not keep it. + {name: "unparseable peer", remote: "not-an-address", wantClient: ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + rec.Reset() + req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/version", nil) + req.RemoteAddr = tt.remote + req.Header.Set(forwardedForHeader, "1.2.3.4, 198.51.100.1") + req.Header.Set("traceparent", "00-"+parentTrace+"-"+parentSpan+"-01") + r.ServeHTTP(httptest.NewRecorder(), req) + + ended := rec.Ended() + require.Len(t, ended, 1) + span := ended[0] + var clients []attribute.Value + for _, kv := range span.Attributes() { + if kv.Key == observability.ClientAddressKey { + clients = append(clients, kv.Value) + } + } + assert.Equal(t, []attribute.Value{attribute.StringValue(tt.wantClient)}, clients, "one client.address, the resolved one") + + parent, _ := trace.TraceIDFromHex(parentTrace) + if tt.wantJoined { + assert.Equal(t, parent, span.SpanContext().TraceID()) + assert.Equal(t, parentSpan, span.Parent().SpanID().String()) + assert.Empty(t, span.Links()) + return + } + assert.NotEqual(t, parent, span.SpanContext().TraceID(), "a new root") + assert.False(t, span.Parent().IsValid()) + require.Len(t, span.Links(), 1) + assert.Equal(t, parent, span.Links()[0].SpanContext.TraceID()) + }) + } +} diff --git a/internal/api/router.go b/internal/api/router.go index 44990c13..44bdf5d2 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -4,6 +4,7 @@ import ( "errors" "log/slog" "net/http" + "net/netip" "runtime/debug" "strings" @@ -53,11 +54,15 @@ type Dependencies struct { // exporter when observability.metrics.prometheus.enabled is true AND port is 0. MetricsHandler http.Handler MetricsPath string + // TrustedProxies are the peers whose X-Forwarded-For names the client + // and whose traceparent the server span joins (server.trusted_proxies). + // Empty trusts no peer. + TrustedProxies []netip.Prefix } // NewRouter creates the chi router with all routes. func NewRouter(deps Dependencies) http.Handler { - r := newProbeRouter(corsOrigins(deps.Tenants, deps.CORSOrigins), deps.Health, deps.Version, deps.MetricsHandler, deps.MetricsPath) + r := newProbeRouter(corsOrigins(deps.Tenants, deps.CORSOrigins), deps.TrustedProxies, deps.Health, deps.Version, deps.MetricsHandler, deps.MetricsPath) // API v1 endpoints. The JWT auth middleware always runs (no enable/disable // switch) on both halves: the tenant routes, which resolve their tenant @@ -161,15 +166,16 @@ func NewRouter(deps Dependencies) http.Handler { // newProbeRouter is what every listener serves, the API's and the ops-only // one alike: the middleware, the JSON 404/405, the probes, /version, and the // same-port metrics endpoint. -func newProbeRouter(origins func(*http.Request) []string, health *HealthHandler, version *VersionHandler, metrics http.Handler, metricsPath string) chi.Router { +func newProbeRouter(origins func(*http.Request) []string, proxies []netip.Prefix, health *HealthHandler, version *VersionHandler, metrics http.Handler, metricsPath string) chi.Router { r := chi.NewRouter() + trusted := trustedProxies(proxies) r.Use(requestID) - // No middleware.RealIP: it rewrites r.RemoteAddr from spoofable forwarded - // headers on every request (chi deprecated it for the IP-spoofing GHSAs), - // and nothing here reads RemoteAddr — WaveHouse does no per-IP logic (that's - // the reverse proxy's job). Trusted-proxy-aware client-IP capture for - // traces/logs is tracked in #333; don't re-add RealIP to get it. + // Not middleware.RealIP, which rewrote r.RemoteAddr from forwarded + // headers whoever sent them (chi deprecated it for the IP-spoofing + // GHSAs): clientAddress reads X-Forwarded-For from a trusted proxy only, + // for the logs and the span, and leaves RemoteAddr alone. + r.Use(clientAddress(trusted)) r.Use(jsonRecoverer) r.Use(corsMiddleware(origins)) @@ -184,7 +190,14 @@ func newProbeRouter(origins func(*http.Request) []string, health *HealthHandler, writeJSONError(w, http.StatusMethodNotAllowed, "method not allowed") }) + // A request from a peer that is not a trusted proxy starts a trace of its + // own, linked to any traceparent it sent rather than joining it, so a + // client cannot write itself into the proxy's traces. + traced := otelhttp.NewMiddleware("wavehouse-api", otelhttp.WithPublicEndpointFn(func(r *http.Request) bool { + return !trusted.trustsPeer(r) + })) r.Use(func(next http.Handler) http.Handler { + withSpan := traced(next) return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { // Skip span creation on infra/probe paths: // /v1/stream — long-lived streams (SSE); the standard @@ -208,7 +221,7 @@ func newProbeRouter(origins func(*http.Request) []string, health *HealthHandler, return } // Normal REST tracing for everything else - otelhttp.NewMiddleware("wavehouse-api")(next).ServeHTTP(w, r) + withSpan.ServeHTTP(w, r) }) }) r.Use(annotateSpan) @@ -249,6 +262,8 @@ type OpsDependencies struct { AuthMW func(http.Handler) http.Handler MetricsHandler http.Handler MetricsPath string + // TrustedProxies is Dependencies.TrustedProxies. + TrustedProxies []netip.Prefix } // NewOpsRouter creates the router of a process without the api role. Every @@ -256,7 +271,7 @@ type OpsDependencies struct { // handlers behind them are not wired here. The reload is gated by the // operator key alone, as the ops tree is over a nested directory. func NewOpsRouter(deps OpsDependencies) http.Handler { - r := newProbeRouter(nil, deps.Health, deps.Version, deps.MetricsHandler, deps.MetricsPath) + r := newProbeRouter(nil, deps.TrustedProxies, deps.Health, deps.Version, deps.MetricsHandler, deps.MetricsPath) if deps.Settings != nil { r.Route("/v1/ops", func(r chi.Router) { r.Use(opsRequestTenant) diff --git a/internal/app/wire.go b/internal/app/wire.go index 9f958496..75cf2229 100644 --- a/internal/app/wire.go +++ b/internal/app/wire.go @@ -1026,6 +1026,7 @@ func (a *App) wireHTTP(authMW func(http.Handler) http.Handler) { } deps.MetricsHandler, deps.MetricsPath = a.inlineMetrics() + deps.TrustedProxies = a.cfg.Server.TrustedProxyPrefixes() a.handler = api.NewRouter(deps) a.wireServers(func() { close(closing) }) } diff --git a/internal/app/wire_ops.go b/internal/app/wire_ops.go index 7bfddb27..2e618b50 100644 --- a/internal/app/wire_ops.go +++ b/internal/app/wire_ops.go @@ -40,6 +40,7 @@ func (a *App) wireOpsHTTP(authMW func(http.Handler) http.Handler) { AuthMW: authMW, } deps.MetricsHandler, deps.MetricsPath = a.inlineMetrics() + deps.TrustedProxies = a.cfg.Server.TrustedProxyPrefixes() a.handler = api.NewOpsRouter(deps) a.wireServers(nil) } diff --git a/internal/auth/auth.go b/internal/auth/auth.go index b3c599c2..2f999e5d 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -500,8 +500,8 @@ func middleware(current func(context.Context) (tenant.ID, *verifier), operatorKe // WARN emitted on an authz denial. Correlation fields are // deliberately NOT stamped per-call-site: the global // TraceHandler (internal/observability) adds request_id, - // tenant and, when a span is active, trace_id/span_id to - // every line; the trusted-proxy client IP is #333. + // client.address, tenant and, when a span is active, + // trace_id/span_id to every line. slog.LogAttrs(r.Context(), slog.LevelInfo, "operator key authenticated request", slog.String("path", r.URL.Path), slog.String("method", r.Method), @@ -523,8 +523,8 @@ func middleware(current func(context.Context) (tenant.ID, *verifier), operatorKe // other unauthenticated caller. WARN + a counter because a // mismatch is a strong probing/brute-force signal: nobody // sends a wrong operator key by accident. Same correlation-field - // handling as the audit line above: request_id and tenant from - // TraceHandler, the client IP in #333. + // handling as the audit line above: request_id, client.address + // and tenant from TraceHandler. operatorKeyFailures.Add(r.Context(), 1) slog.LogAttrs(r.Context(), slog.LevelWarn, "operator key authentication failed", slog.String("path", r.URL.Path), diff --git a/internal/config/backends.go b/internal/config/backends.go index 5fa39571..1631aa11 100644 --- a/internal/config/backends.go +++ b/internal/config/backends.go @@ -266,6 +266,7 @@ func (c *Config) NeedsDataDir() bool { // which one process cannot tell from many. func (c *Config) Warnings() []string { out := append(c.natsWarnings(), c.profilingWarnings()...) + out = append(out, c.trustedProxyWarnings()...) // The rest are the api role's: a process without it opens no cache it // reads and no dedupe store, and a split's Deployments differ only in // roles, so the API's warnings cover the others'. diff --git a/internal/config/config.go b/internal/config/config.go index bce11681..a1cc1e20 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -135,6 +135,11 @@ type OTelLogs struct { type Server struct { Port int `yaml:"port" env:"WH_SERVER_PORT"` ShutdownTimeout int `yaml:"shutdown_timeout" env:"WH_SERVER_SHUTDOWN_TIMEOUT"` + // TrustedProxies names the reverse proxies whose X-Forwarded-For and + // traceparent are believed: CIDRs or bare addresses, parsed by + // TrustedProxyPrefixes. Empty, the default, trusts no peer: the client is + // the connection's own address and an incoming trace is linked, not joined. + TrustedProxies []string `yaml:"trusted_proxies" env:"WH_SERVER_TRUSTED_PROXIES"` } // ClickHouse holds the connection ceiling. The wiring — address, HTTP port @@ -355,6 +360,9 @@ func (c *Config) Validate() error { if err := c.validateProfiling(); err != nil { return err } + if err := c.validateTrustedProxies(); err != nil { + return err + } if err := c.validateRoles(); err != nil { return err } diff --git a/internal/config/proxies.go b/internal/config/proxies.go new file mode 100644 index 00000000..dc5de143 --- /dev/null +++ b/internal/config/proxies.go @@ -0,0 +1,83 @@ +package config + +import ( + "errors" + "fmt" + "net/netip" + "strings" +) + +// ParseTrustedProxies reads server.trusted_proxies: each entry a CIDR, or a +// bare address standing for itself alone. Prefixes are masked, so +// 10.1.2.3/8 is 10.0.0.0/8, and IPv4-mapped IPv6 entries are read as IPv4. +func ParseTrustedProxies(entries []string) ([]netip.Prefix, error) { + out := make([]netip.Prefix, 0, len(entries)) + for _, e := range entries { + e = strings.TrimSpace(e) + if e == "" { + continue + } + p, err := parseTrustedProxy(e) + if err != nil { + return nil, fmt.Errorf("server.trusted_proxies: %q %w", e, err) + } + out = append(out, p) + } + return out, nil +} + +var errNotAddress = errors.New("is neither a CIDR nor an IP address") + +func parseTrustedProxy(e string) (netip.Prefix, error) { + if !strings.Contains(e, "/") { + a, err := netip.ParseAddr(e) + if err != nil || a.Zone() != "" { + return netip.Prefix{}, errNotAddress + } + a = a.Unmap() + return netip.PrefixFrom(a, a.BitLen()), nil + } + p, err := netip.ParsePrefix(e) + if err != nil { + return netip.Prefix{}, errNotAddress + } + // Peers are unmapped before matching, so a mapped prefix must become IPv4; + // one shorter than /96 reaches past ::ffff:0:0/96 and has no IPv4 form. + if a := p.Addr(); a.Is4In6() { + if p.Bits() < 96 { + return netip.Prefix{}, errors.New("is an IPv4-mapped prefix shorter than /96: write it as an IPv4 CIDR") + } + p = netip.PrefixFrom(a.Unmap(), p.Bits()-96) + } + return p.Masked(), nil +} + +// TrustedProxyPrefixes is server.trusted_proxies parsed. Validate has refused an +// entry that does not parse, so a loaded Config never drops one here. +func (s Server) TrustedProxyPrefixes() []netip.Prefix { + p, _ := ParseTrustedProxies(s.TrustedProxies) + return p +} + +func (c *Config) validateTrustedProxies() error { + _, err := ParseTrustedProxies(c.Server.TrustedProxies) + return err +} + +// trustedProxyWarnings flags each /0 entry: it trusts every peer of its address +// family (peers are unmapped, so ::/0 covers only IPv6 ones), so a client of +// that family reaching the port directly chooses the address it is logged under. +func (c *Config) trustedProxyWarnings() []string { + var w []string + for _, p := range c.Server.TrustedProxyPrefixes() { + if p.Bits() != 0 { + continue + } + family := "IPv4" + if p.Addr().Is6() { + family = "IPv6" + } + w = append(w, "server.trusted_proxies includes "+p.String()+": every "+family+" peer is trusted, so an "+family+" client that reaches server.port directly sets its own client.address through X-Forwarded-For and has its traceparent adopted; list only your proxies' addresses") + } + return w +} diff --git a/internal/config/proxies_test.go b/internal/config/proxies_test.go new file mode 100644 index 00000000..f29ce5fe --- /dev/null +++ b/internal/config/proxies_test.go @@ -0,0 +1,85 @@ +package config + +import ( + "net/netip" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseTrustedProxies(t *testing.T) { + t.Parallel() + tests := []struct { + name string + entries []string + want []string + wantErr string + }{ + {name: "none", entries: nil, want: []string{}}, + {name: "CIDRs", entries: []string{"10.0.0.0/8", "172.16.0.0/12", "fd00::/8"}, want: []string{"10.0.0.0/8", "172.16.0.0/12", "fd00::/8"}}, + {name: "bare addresses stand for themselves", entries: []string{"10.1.2.3", "2001:db8::1"}, want: []string{"10.1.2.3/32", "2001:db8::1/128"}}, + {name: "host bits masked", entries: []string{"10.1.2.3/8"}, want: []string{"10.0.0.0/8"}}, + {name: "IPv4-mapped read as IPv4", entries: []string{"::ffff:10.0.0.0/104", "::ffff:10.1.2.3"}, want: []string{"10.0.0.0/8", "10.1.2.3/32"}}, + {name: "spaces and empty entries", entries: []string{" 10.0.0.0/8 ", "", " "}, want: []string{"10.0.0.0/8"}}, + {name: "hostname refused", entries: []string{"proxy.internal"}, wantErr: `"proxy.internal"`}, + {name: "bad prefix length refused", entries: []string{"10.0.0.0/33"}, wantErr: `"10.0.0.0/33"`}, + {name: "zoned address refused", entries: []string{"fe80::1%eth0"}, wantErr: `"fe80::1%eth0"`}, + {name: "IPv4-mapped prefix shorter than /96 refused", entries: []string{"::ffff:0:0/80"}, wantErr: `"::ffff:0:0/80" is an IPv4-mapped prefix`}, + {name: "port refused", entries: []string{"10.0.0.1:80"}, wantErr: `"10.0.0.1:80"`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got, err := ParseTrustedProxies(tt.entries) + if tt.wantErr != "" { + require.Error(t, err) + assert.Contains(t, err.Error(), tt.wantErr) + assert.Contains(t, err.Error(), "server.trusted_proxies") + return + } + require.NoError(t, err) + strs := make([]string, 0, len(got)) + for _, p := range got { + strs = append(strs, p.String()) + } + assert.Equal(t, tt.want, strs) + }) + } +} + +func TestValidate_TrustedProxies(t *testing.T) { + t.Parallel() + cfg := defaults() + cfg.Settings.Dir = t.TempDir() + cfg.Server.TrustedProxies = []string{"10.0.0.0/8", "nope"} + require.ErrorContains(t, cfg.Validate(), `"nope"`) + + cfg.Server.TrustedProxies = []string{"10.0.0.0/8", "192.0.2.7"} + require.NoError(t, cfg.Validate()) + assert.Equal(t, []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8"), netip.MustParsePrefix("192.0.2.7/32")}, cfg.Server.TrustedProxyPrefixes()) + assert.Empty(t, cfg.Warnings()) +} + +func TestWarnings_TrustEveryPeer(t *testing.T) { + t.Parallel() + for all, family := range map[string]string{"0.0.0.0/0": "IPv4", "::/0": "IPv6"} { + cfg := defaults() + cfg.Server.TrustedProxies = []string{"10.0.0.0/8", all} + w := cfg.Warnings() + require.Len(t, w, 1, all) + assert.Contains(t, w[0], "every "+family+" peer is trusted") + } + + // Each /0 trusts only its own family, so both are named. + cfg := defaults() + cfg.Server.TrustedProxies = []string{"0.0.0.0/0", "::/0"} + assert.Len(t, cfg.Warnings(), 2) +} + +func TestLoad_TrustedProxiesFromEnv(t *testing.T) { + t.Setenv("WH_SERVER_TRUSTED_PROXIES", "10.0.0.0/8,172.16.0.0/12") + cfg, err := Load("nonexistent.yaml") + require.NoError(t, err) + assert.Equal(t, []string{"10.0.0.0/8", "172.16.0.0/12"}, cfg.Server.TrustedProxies) +} diff --git a/internal/observability/logger.go b/internal/observability/logger.go index 9b587b8f..80bf727a 100644 --- a/internal/observability/logger.go +++ b/internal/observability/logger.go @@ -19,7 +19,8 @@ type TraceHandler struct { } // Handle pulls the trace_id and span_id from the context if they exist, and -// request_id and tenant from the request scope (WithRequest) when there is one. +// request_id, client.address and tenant from the request scope (WithRequest) +// when there is one. func (h *TraceHandler) Handle(ctx context.Context, r slog.Record) error { addRequestAttrs(ctx, &r) if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() { diff --git a/internal/observability/request.go b/internal/observability/request.go index 2c0ef3f7..3b4366ef 100644 --- a/internal/observability/request.go +++ b/internal/observability/request.go @@ -14,15 +14,19 @@ import ( const ( RequestIDKey = "request_id" TenantKey = "tenant" + // ClientAddressKey is OpenTelemetry's semantic-convention name, which + // the HTTP server span already uses for the same value. + ClientAddressKey = "client.address" ) // requestScope is what a request's log records and spans say about it: its -// id, known when it arrives, and its tenant, known only once a later -// middleware resolves it. The tenant is filled in place, so a record logged -// through a context taken before resolution (the panic recoverer's, say) -// still carries it. +// id, known when it arrives, its client address, and its tenant, known only +// once a later middleware resolves it. Both are filled in place, so a record +// logged through a context taken before resolution (the panic recoverer's, +// say) still carries them. type requestScope struct { id string + client atomic.Pointer[string] tenant atomic.Pointer[string] } @@ -57,8 +61,21 @@ func SetRequestTenant(ctx context.Context, tenant string) { trace.SpanFromContext(ctx).SetAttributes(attribute.String(TenantKey, tenant)) } +// SetRequestClient records the address the request came from on its scope +// and on the span in ctx. Outside a request it still annotates the span. An +// empty addr marks the address unresolved: log records leave it out, and the +// span's client.address is blanked rather than left as otelhttp's. +func SetRequestClient(ctx context.Context, addr string) { + if s := scopeFrom(ctx); s != nil { + s.client.Store(&addr) + } + trace.SpanFromContext(ctx).SetAttributes(attribute.String(ClientAddressKey, addr)) +} + // AnnotateSpan copies the request scope in ctx onto the span in ctx. Call it -// once the server span exists: the scope is set up before it. +// once the server span exists: the scope is set up before it. Its +// client.address replaces the one otelhttp derives, which reads +// X-Forwarded-For whoever sent it. func AnnotateSpan(ctx context.Context) { s := scopeFrom(ctx) if s == nil { @@ -69,6 +86,9 @@ func AnnotateSpan(ctx context.Context) { return } attrs := []attribute.KeyValue{attribute.String(RequestIDKey, s.id)} + if c := s.client.Load(); c != nil { + attrs = append(attrs, attribute.String(ClientAddressKey, *c)) + } if t := s.tenant.Load(); t != nil { attrs = append(attrs, attribute.String(TenantKey, *t)) } @@ -83,6 +103,9 @@ func addRequestAttrs(ctx context.Context, r *slog.Record) { return } r.AddAttrs(slog.String(RequestIDKey, s.id)) + if c := s.client.Load(); c != nil && *c != "" { + r.AddAttrs(slog.String(ClientAddressKey, *c)) + } t := s.tenant.Load() if t == nil { return diff --git a/internal/observability/request_test.go b/internal/observability/request_test.go index d3936f5a..63b4d5cb 100644 --- a/internal/observability/request_test.go +++ b/internal/observability/request_test.go @@ -13,6 +13,7 @@ import ( "go.opentelemetry.io/otel/attribute" sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" ) // logLines runs log against a TraceHandler over a JSON handler and returns @@ -161,3 +162,46 @@ func TestAnnotateSpan_TenantAlreadyResolved(t *testing.T) { require.Len(t, rec.Ended(), 1) assert.Contains(t, rec.Ended()[0].Attributes(), attribute.String(TenantKey, "acme")) } + +func TestTraceHandler_ClientAddress(t *testing.T) { + t.Parallel() + bare := logLines(t, func(l *slog.Logger) { l.InfoContext(WithRequest(context.Background(), "req-1"), "hello") }) + require.Len(t, bare, 1) + assert.NotContains(t, bare[0], ClientAddressKey, "no address recorded, none logged") + + ctx := WithRequest(context.Background(), "req-1") + SetRequestClient(ctx, "198.51.100.1") + lines := logLines(t, func(l *slog.Logger) { l.InfoContext(ctx, "hello") }) + require.Len(t, lines, 1) + assert.Equal(t, "198.51.100.1", lines[0][ClientAddressKey]) +} + +// AnnotateSpan replaces a client.address the span already carries (otelhttp's, +// read from X-Forwarded-For) with the request scope's. +func TestAnnotateSpan_ReplacesClientAddress(t *testing.T) { + t.Parallel() + rec := tracetest.NewSpanRecorder() + tracer := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)).Tracer("test") + + scoped := WithRequest(context.Background(), "req-1") + SetRequestClient(scoped, "198.51.100.1") + ctx, span := tracer.Start(scoped, "server", trace.WithAttributes(attribute.String(ClientAddressKey, "1.2.3.4"))) + AnnotateSpan(ctx) + span.End() + + require.Len(t, rec.Ended(), 1) + assert.ElementsMatch(t, []attribute.KeyValue{ + attribute.String(RequestIDKey, "req-1"), + attribute.String(ClientAddressKey, "198.51.100.1"), + }, rec.Ended()[0].Attributes()) +} + +// An unresolved client is left out of log records. +func TestTraceHandler_UnresolvedClientAddress(t *testing.T) { + t.Parallel() + ctx := WithRequest(context.Background(), "req-1") + SetRequestClient(ctx, "") + lines := logLines(t, func(l *slog.Logger) { l.InfoContext(ctx, "hello") }) + require.Len(t, lines, 1) + assert.NotContains(t, lines[0], ClientAddressKey) +} diff --git a/tests/integration/otel_test.go b/tests/integration/otel_test.go index 8b93f274..d48a3d8a 100644 --- a/tests/integration/otel_test.go +++ b/tests/integration/otel_test.go @@ -421,10 +421,11 @@ func TestOTel_Headers_AppliedToAllSignals(t *testing.T) { // TestOTel_IngestTraceContinuity follows one ingest request's trace past the // queue through the wired app: the server span and the handler's span under -// the caller's traceparent, then, in the ingest process, a flush span linked -// to the handler's span and an INSERT span under the flush. The link is what -// ties a slow or failed insert back to the request that produced its rows -// once the api and ingest roles run in separate processes. +// the caller's traceparent (the suite trusts loopback as a proxy), then, in +// the ingest process, a flush span linked to the handler's span and an +// INSERT span under the flush. The link is what ties a slow or failed insert +// back to the request that produced its rows once the api and ingest roles +// run in separate processes. func TestOTel_IngestTraceContinuity(t *testing.T) { // No t.Parallel: initAndShutdown mutates OTel globals, which the running // app's handlers and worker read per request and per flush. diff --git a/tests/integration/setup_test.go b/tests/integration/setup_test.go index 3da5ca97..ecfefaf4 100644 --- a/tests/integration/setup_test.go +++ b/tests/integration/setup_test.go @@ -174,8 +174,10 @@ func setup() (int, func()) { dataDir := mustTempDir() cleanups.push(func() { _ = os.RemoveAll(dataDir) }) cfg := &config.Config{ - DataDir: dataDir, - Server: config.Server{ShutdownTimeout: 10}, + DataDir: dataDir, + // The suite's client dials from loopback and plays the proxy, so + // a traceparent it sends is joined (TestOTel_IngestTraceContinuity). + Server: config.Server{ShutdownTimeout: 10, TrustedProxies: []string{"127.0.0.1", "::1"}}, MQ: config.MQ{Backend: config.MQEmbedded}, Cache: config.Cache{Backend: config.CacheLocal, L1MaxCost: 1 << 30}, // 1 GB Dedupe: config.Dedupe{Backend: config.DedupePebble},