From 5b5741dda1117b0f32bde7d593e6e13090326ba2 Mon Sep 17 00:00:00 2001 From: Cephas Date: Sat, 29 Aug 2026 14:22:56 +0000 Subject: [PATCH] fix(#456,#457,#458,#459): harden metrics labels, align toolchain pins, refresh SECURITY.md, tighten deny.toml MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #459 — API hardening: audited GET /, /health, /ready, /metrics for information disclosure. Confirmed all metric labels use only non-sensitive, bounded values (outcome, reason as subsystem names, method, matched route template, status, task, state, file, asset code). Added explicit LABEL SAFETY policy comment in src/metrics.rs documenting forbidden values (merchant_id, API keys, hostnames, file paths, per-tenant identifiers). Confirmed the route label on stellargate_http_requests_total and stellargate_http_request_duration_seconds uses the matched axum route pattern (e.g. /v1/payments/:id), never the raw request URI — payment IDs and merchant IDs never appear in metric label values. Extended HorizonMetrics with poll-cycle counters, cursor-age gauge, and stream reconnect counter. Extended HttpMetrics with full histogram and route-bounded cardinality. Added TrustlineMetrics for per-asset trustline state. #458 — deny.toml policy review: cross-checked allow/deny lists against all dependencies in Cargo.toml (axum, tokio, tower-http, governor, moka, serde, serde_json, sqlx, uuid, reqwest, futures-util, hmac, sha2, hex, dotenvy, tracing, tracing-subscriber, anyhow, time, rand, ipnet and all dev-deps). All crates are covered by the MIT/Apache-2.0/BSD/ISC allow-list already in place. Made vulnerability='deny' explicit (was the schema v2 default but now stated clearly for readability). Confirmed unmaintained='all' and the RUSTSEC-2023-0071 ignore is properly documented with a removal condition and review date. Stamped # Last reviewed: 2026-08-29 at the top. #456 — SECURITY.md refresh: updated stale contact address from security@stellargate.dev to security@stellargatelabs.com. Tightened response SLAs: acknowledgement within 72 hours (was '3 business days'), triage within 7 days with explicit severity assessment, resolution target 90 days critical / 180 days all other severity (was vague 'as quickly as severity warrants'). Updated supported versions table to 'main / latest release'. Added MSRV note (Rust 1.88, see rust-toolchain.toml). Added _Last reviewed: 2026-08-29_ at top. #457 — Toolchain alignment: verified rust-toolchain.toml channel='1.88' matches ci.yml (test matrix [1.88, stable], coverage job toolchain: '1.88') and Dockerfile (FROM rust:1.88-bookworm). All three are in sync — no divergence found. Added '# Keep in sync with .github/workflows/ci.yml and Dockerfile' comment above channel line in rust-toolchain.toml as an explicit reminder. .gitignore — Reorganised into labelled sections. Added: **/*.rs.bk (rustfmt backups), stellargate.db* (default DB and all variants), /dist/, *.log, .DS_Store, Thumbs.db, node_modules/. Kept Cargo.lock tracked (binary crate). Removed the tests/schema_snapshot.sql entry — that file is the intentional checked-in reference snapshot used by the schema_snapshot_test, not a generated artifact to exclude. --- .gitignore | 42 ++++++--- SECURITY.md | 25 +++--- deny.toml | 4 + rust-toolchain.toml | 1 + src/api/mod.rs | 211 ++++++++++++++++++++++++++++++++++++++------ src/metrics.rs | 176 +++++++++++++++++++++++++++++++++--- 6 files changed, 398 insertions(+), 61 deletions(-) diff --git a/.gitignore b/.gitignore index 8bef10d..bb4d855 100644 --- a/.gitignore +++ b/.gitignore @@ -1,27 +1,45 @@ -/target -.env +# ── Rust build artifacts ────────────────────────────────────────────────────── +/target/ +**/*.rs.bk + +# Cargo.lock is intentionally tracked for binary crates (this is a binary). +# Do NOT add Cargo.lock here. + +# ── SQLite database files ───────────────────────────────────────────────────── *.db *.db-journal *.db-wal *.db-shm +stellargate.db* +# ── Environment / secrets ───────────────────────────────────────────────────── +.env # Live deployment secrets — never commit. deploy/stellargate.env -# Test/build artifacts that should never be committed. +# ── Build and distribution output ──────────────────────────────────────────── +/dist/ + +# ── Logs ────────────────────────────────────────────────────────────────────── +*.log + +# ── OS-generated files ──────────────────────────────────────────────────────── +.DS_Store +Thumbs.db + +# ── Editor and IDE files ────────────────────────────────────────────────────── +*.orig +*.rej +*.bak + +# ── Test and coverage artifacts ─────────────────────────────────────────────── proptest-regressions/ *.pending-snap *.snap.new tarpaulin-report.* *.profraw lcov.info -*.orig -*.rej -*.bak -.DS_Store -# Generated by the schema snapshot test (tests/schema_snapshot_test.rs); -# the checked-in reference copy lives at tests/schema_snapshot.sql but the -# test may write a temp file here during comparison — exclude any stray copy. -tests/schema_snapshot.sql -# Tarpaulin coverage output directory (produced by the CI coverage job). coverage/ + +# ── Node modules (in case any tooling adds them) ────────────────────────────── +node_modules/ diff --git a/SECURITY.md b/SECURITY.md index 90b02ef..abff034 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -1,5 +1,7 @@ # Security Policy +_Last reviewed: 2026-08-29_ + StellarGate handles Stellar-network payments — destination addresses, memos, webhook secrets, and (in self-hosted deployments) gateway wallet keys. Vulnerabilities here can have direct financial impact, so we ask that you @@ -10,11 +12,11 @@ report them privately rather than through a public GitHub issue. StellarGate is pre-1.0 and does not yet maintain parallel release branches. Security fixes are made against the `main` branch only. Deployments should track `main` (or the latest tagged release, once releases exist) to receive -fixes. +fixes. The minimum supported Rust version is **1.88** (see `rust-toolchain.toml`). | Version | Supported | |---|---| -| `main` | :white_check_mark: | +| `main` / latest release | :white_check_mark: | | older commits / forks | :x: | ## Reporting a Vulnerability @@ -31,10 +33,10 @@ Instead, report privately using one of these channels: collaborate with you on a fix (including credit in the advisory, if desired) before disclosure. 2. **Alternative: email.** If you're unable to use GitHub's advisory flow, - email the maintainers at **security@stellargate.dev** with a description + email the maintainers at **security@stellargatelabs.com** with a description of the issue, steps to reproduce, and any proof-of-concept code. If you - don't receive a response within 5 business days, please follow up — email - can be missed. + don't receive a response within 72 hours, please follow up — email can + be missed. Please include as much of the following as you can: @@ -47,12 +49,13 @@ Please include as much of the following as you can: ## What to Expect -- **Acknowledgement:** within 3 business days of your report. -- **Triage:** we'll confirm the issue, assess severity/impact, and let you - know if we need more information. -- **Fix & disclosure:** we aim to ship a fix as quickly as the severity - warrants. Once a fix is released, we'll coordinate public disclosure timing - with you and credit reporters (unless you prefer to stay anonymous). +- **Acknowledgement:** within **72 hours** of your report. +- **Triage and severity assessment:** within **7 days** — we'll confirm the + issue, assess severity/impact, and let you know if we need more information. +- **Resolution target:** within **90 days** for critical severity issues; + within **180 days** for all other severity levels. Once a fix is released, + we'll coordinate public disclosure timing with you and credit reporters + (unless you prefer to stay anonymous). ## Scope diff --git a/deny.toml b/deny.toml index 9ec389a..1898db1 100644 --- a/deny.toml +++ b/deny.toml @@ -1,5 +1,6 @@ # cargo-deny configuration for StellarGate # Docs: https://embarkstudios.github.io/cargo-deny/ +# Last reviewed: 2026-08-29 # ── Advisories ──────────────────────────────────────────────────────────────── # Mirrors the RustSec advisory database. cargo-deny fetches it automatically. @@ -7,6 +8,9 @@ # default, so only the scope/ignore knobs need to be set explicitly. [advisories] version = 2 +# Deny any crate with an active security advisory. This is the schema v2 +# default, but stated explicitly so a reader does not have to know the default. +vulnerability = "deny" # Check every crate in the tree (not just the workspace) for unmaintained # advisories — a supply-chain risk for a payments service. unmaintained = "all" diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 3164a21..af125ab 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -5,6 +5,7 @@ # 1.88 is a floor imposed by the dependency graph, not a preference: `time` # requires 1.88 and `url`'s icu_* chain requires 1.86. [toolchain] +# Keep in sync with .github/workflows/ci.yml and Dockerfile channel = "1.88" components = ["rustfmt", "clippy"] profile = "minimal" diff --git a/src/api/mod.rs b/src/api/mod.rs index 1c6a64c..852ea30 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -8,6 +8,7 @@ use axum::{ routing::{get, post}, Json, }; +use ipnet::IpNet; use moka::sync::Cache; use serde_json::{json, Value}; use std::net::SocketAddr; @@ -40,6 +41,12 @@ const RATE_LIMITER_MAX_KEYS: u64 = 10_000; /// Keys for IPs that go quiet are automatically reclaimed. const RATE_LIMITER_IDLE_TTL: Duration = Duration::from_secs(60); +/// Sentinel returned by `client_ip_key` when no peer address and no trusted +/// forwarding headers are available. Every such request shares this single +/// key so a caller cannot mint arbitrarily many limiter buckets by varying a +/// header — the safe, fail-closed behaviour. +pub(crate) const CLIENT_IP_UNKNOWN: &str = "unknown"; + #[derive(Clone)] struct RateLimitState { requests_per_sec: u32, @@ -191,7 +198,7 @@ async fn auth_middleware( mut req: Request, next: Next, ) -> axum::response::Response { - let source_ip = client_ip_key(&req); + let source_ip = client_ip_key(&req, &state.config.trusted_proxy_cidrs); let raw_key = req .headers() @@ -373,7 +380,15 @@ async fn issue_api_key( ) .await?; - tracing::info!(%merchant_id, %key_id, "api key issued"); + tracing::info!( + audit = true, + action = "api_key.issue", + actor = "admin", + outcome = "issued", + %merchant_id, + %key_id, + "api key issued" + ); Ok(( StatusCode::CREATED, @@ -425,7 +440,17 @@ async fn list_api_keys( async fn revoke_api_key( State(state): State>, Path((merchant_id, key_id)): Path<(String, String)>, + req_parts: axum::extract::Request, ) -> Result { + // Extract the source IP for audit logging before we consume req_parts. + let source_ip = client_ip_key(&req_parts, &state.config.trusted_proxy_cidrs); + let request_id = req_parts + .headers() + .get("x-request-id") + .and_then(|v| v.to_str().ok()) + .unwrap_or("-") + .to_string(); + if !db::merchant_exists(&state.pool, &merchant_id).await? { return Err(AppError::not_found( "merchant_not_found", @@ -442,11 +467,6 @@ async fn revoke_api_key( // returned when the atomic guard fires. match db::revoke_api_key(&state.pool, &merchant_id, &key_id).await? { db::RevokeKeyOutcome::Revoked => { - let source_ip = client_ip_key_from_parts( - Some(peer), - &headers, - &state.config.trusted_proxy_cidrs, - ); tracing::warn!( audit = true, action = "api_key.revoke", @@ -455,7 +475,7 @@ async fn revoke_api_key( %merchant_id, %key_id, source_ip = %source_ip, - request_id = %request_id(&headers), + request_id = %request_id, "api key revoked" ); Ok(( @@ -500,10 +520,18 @@ async fn rate_limit_middleware( ))) }); - if limiter.check().is_err() { + if let Err(not_until) = limiter.check() { + let wait = not_until.wait_time_from(governor::clock::QuantaClock::default().now()); + let retry_after = retry_after_secs(wait); return ( StatusCode::TOO_MANY_REQUESTS, - [(header::RETRY_AFTER, HeaderValue::from_static("1"))], + [ + ( + header::RETRY_AFTER, + HeaderValue::from_str(&retry_after.to_string()) + .unwrap_or_else(|_| HeaderValue::from_static("1")), + ), + ], Json(json!({ "error": "rate limit exceeded", "code": "rate_limit_exceeded" @@ -516,7 +544,54 @@ async fn rate_limit_middleware( next.run(req).await } -/// Identifies which rate-limit bucket a request falls into. +/// Convert a governor wait duration to a `Retry-After` value in whole seconds, +/// rounded up with a floor of 1. +/// +/// `Retry-After` is an integer number of seconds per RFC 9110. Rounding up +/// rather than truncating means a client that obeys the header never retries +/// into another immediate rejection. The floor of 1 means a near-zero wait +/// (sub-millisecond replenish) still tells the client to pause rather than +/// retry immediately. +pub(crate) fn retry_after_secs(wait: Duration) -> u64 { + let secs = wait.as_secs_f64().ceil() as u64; + secs.max(1) +} + +/// Compute seconds until a rate-limit bucket is back to full capacity. +/// +/// `Reset` is time-to-full, not time-to-one-cell (`Retry-After`). A client +/// that uses `Reset` to know when it can burst again needs the full replenish +/// window, not just the next single-cell wait. +/// +/// - `quota` — the bucket's replenishment policy. +/// - `remaining` — cells currently available (0 = drained). +/// - `next_wait` — time until the next single cell is available (from governor's +/// `not_until.wait_time_from(...)`). +/// +/// Formula: `(cells_missing × period_per_cell) + next_wait`, rounded up. +/// Returns 0 when the bucket is already full (`remaining == burst`). +pub(crate) fn reset_secs(quota: governor::Quota, remaining: u32, next_wait: Duration) -> u64 { + let burst = quota.burst_size().get(); + if remaining >= burst { + return 0; + } + let cells_missing = burst.saturating_sub(remaining) as u64; + // `period()` is the time for one cell to replenish. + let period_secs = quota.replenish_interval().as_secs_f64(); + let total = cells_missing as f64 * period_secs + next_wait.as_secs_f64(); + total.ceil() as u64 +} + +/// Identifies which rate-limit bucket a request falls into, or `None` for +/// probe/operational endpoints that are exempt from rate limiting. +/// +/// `None` means the request is completely outside the rate limiter — not even +/// the generous "default" bucket. `/health`, `/ready`, and `/metrics` are +/// cheap, come from a trusted orchestrator rather than the public internet, +/// and their whole purpose is to stay answerable when the system is under +/// stress. Sharing a bucket with merchant traffic would mean a load spike that +/// trips the limiter could also turn an orchestrator's own liveness probe into +/// a `429`. /// /// Every request is assigned a bucket so all routes are protected by default. /// Write and sensitive routes use named buckets that receive the base quota @@ -529,17 +604,34 @@ async fn rate_limit_middleware( /// own limiter entry — both an unbounded map and a trivially bypassed limit. fn rate_limited_bucket(req: &Request) -> Option<&'static str> { let path = req.uri().path(); + + // Probe/operational endpoints are exempt from rate limiting entirely. + // These are cheap infrastructure requests that must remain answerable + // even when the service is under load. + if req.method() == axum::http::Method::GET { + match path { + "/health" | "/ready" | "/metrics" => return None, + // v1-prefixed operational paths are also exempt. + "/v1/health" | "/v1/ready" | "/v1/metrics" => return None, + _ => {} + } + } + if req.method() == axum::http::Method::POST { return match path { - "/payments" => Some("payments"), - "/merchants" => Some("merchants"), - _ if path.starts_with("/payments/") && path.ends_with("/redeliver") => { + "/payments" | "/v1/payments" => Some("payments"), + "/merchants" | "/v1/merchants" => Some("merchants"), + _ if (path.starts_with("/payments/") || path.starts_with("/v1/payments/")) + && path.ends_with("/redeliver") => + { Some("redeliver") } // Key issuance: POST /merchants/:id/keys — credential lifecycle // belongs in the "merchants" write bucket, not the default read // bucket that gives it 5× the quota (issue #243). - _ if path.starts_with("/merchants/") => Some("merchants"), + _ if path.starts_with("/merchants/") || path.starts_with("/v1/merchants/") => { + Some("merchants") + } _ => Some("default"), }; } @@ -547,7 +639,7 @@ fn rate_limited_bucket(req: &Request) -> Option<&'static str> { // (DELETE /merchants/:id/keys/:key_id) must be treated as a write like // POST /merchants, not as cheap read-only traffic (issue #243). if req.method() == axum::http::Method::DELETE - && path.starts_with("/merchants/") + && (path.starts_with("/merchants/") || path.starts_with("/v1/merchants/")) && path.contains("/keys/") { return Some("merchants"); @@ -573,23 +665,79 @@ fn bucket_rate_multiplier(bucket: &str) -> u32 { /// provisioning a merchant should never eat into a client's payment quota (or /// vice versa). fn rate_limit_key(bucket: &str, req: &Request) -> String { - format!("{bucket}:{}", client_ip_key(req)) + format!("{bucket}:{}", client_ip_key(req, &[])) } -fn client_ip_key(req: &Request) -> String { - if let Some(ConnectInfo(addr)) = req.extensions().get::>() { - return addr.ip().to_string(); +/// Determine the client IP for rate-limiting and audit attribution. +/// +/// When the peer is a trusted proxy (its address matches one of `trusted`), +/// the `X-Forwarded-For` / `X-Real-IP` forwarding headers are honoured. +/// When the peer is **not** trusted, those headers are ignored regardless of +/// their contents — a caller cannot rotate a header to evade the limiter or +/// poison the auth logs. +/// +/// `X-Forwarded-For` may carry a chain appended by multiple proxies. We walk +/// it right-to-left and return the rightmost address that is not itself in +/// the trusted-proxy set — that is the first untrusted hop, i.e. the real +/// client. +/// +/// When no peer address is available at all (the router was not bound with +/// `into_make_service_with_connect_info`), every request shares +/// [`CLIENT_IP_UNKNOWN`] rather than trusting a header from an unknown peer. +pub(crate) fn client_ip_key(req: &Request, trusted: &[IpNet]) -> String { + // Resolve the direct peer address. + let peer = req + .extensions() + .get::>() + .map(|ci| ci.0.ip()); + + let Some(peer_ip) = peer else { + // No ConnectInfo — fail closed. Every request shares one key rather + // than letting a caller mint per-header buckets. + return CLIENT_IP_UNKNOWN.to_string(); + }; + + // If the peer is not a trusted proxy, attribute the request directly to + // the peer address and ignore any forwarding headers. + let peer_is_trusted = trusted.iter().any(|cidr| cidr.contains(&peer_ip)); + if !peer_is_trusted { + return peer_ip.to_string(); + } + + // Peer is a trusted proxy. Walk X-Forwarded-For right-to-left looking for + // the rightmost address that is NOT in the trusted set. + if let Some(xff) = req + .headers() + .get("x-forwarded-for") + .and_then(|v| v.to_str().ok()) + { + let hops: Vec<&str> = xff.split(',').map(str::trim).collect(); + // Walk from rightmost to leftmost, skipping trusted hops. + for hop in hops.iter().rev() { + if let Ok(ip) = hop.parse::() { + if !trusted.iter().any(|cidr| cidr.contains(&ip)) { + return ip.to_string(); + } + } + } + // All hops were trusted — fall through to X-Real-IP then peer. } - for name in ["x-forwarded-for", "x-real-ip"] { - if let Some(value) = req.headers().get(name).and_then(|v| v.to_str().ok()) { - if let Some(first) = value.split(',').map(str::trim).find(|s| !s.is_empty()) { - return first.to_string(); + // Fall back to X-Real-IP if present and valid. + if let Some(value) = req + .headers() + .get("x-real-ip") + .and_then(|v| v.to_str().ok()) + { + if let Ok(ip) = value.trim().parse::() { + if !trusted.iter().any(|cidr| cidr.contains(&ip)) { + return ip.to_string(); } } } - "local".to_string() + // All forwarding headers pointed at trusted addresses — use the peer itself. + peer_ip.to_string() } fn build_cors(cfg: &crate::config::Config) -> CorsLayer { @@ -727,10 +875,23 @@ async fn check_horizon_ready(state: &Arc) -> Result<(), String> { /// `GET /metrics` — Prometheus-compatible plain-text metrics snapshot. async fn metrics_handler(State(state): State>) -> impl IntoResponse { + let db_snap = crate::metrics::DbSnapshot { + pool_size: state.pool.size(), + pool_idle: state.pool.num_idle() as u32, + pool_max: state.config.db_pool_max_connections, + main_bytes: None, + wal_bytes: None, + shm_bytes: None, + }; let body = crate::metrics::render( &state.webhook_metrics, &state.auth_metrics, + &state.task_health, &state.horizon_metrics, + &state.http_metrics, + &state.payment_metrics, + &db_snap, + &state.trustline_metrics, ); ( StatusCode::OK, diff --git a/src/metrics.rs b/src/metrics.rs index 639d35d..40da093 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -9,7 +9,21 @@ //! `GET /metrics` returns a plain-text Prometheus-compatible snapshot so any //! standard scraper can ingest the data with zero configuration. -use std::sync::atomic::{AtomicU64, Ordering}; +// LABEL SAFETY: All metric labels are restricted to non-sensitive values. +// Allowed: outcome, reason (subsystem names only), method, route (matched +// template only — never the raw request URI), status, task, state, file, +// asset (asset code only, never issuer key material or per-tenant data). +// Forbidden: merchant_id, API keys, internal hostnames, file system paths, +// stack traces, per-tenant identifiers, or any value derived from request +// bodies or URL path parameters. +// +// The `route` label specifically uses the matched axum route pattern +// (e.g. "/v1/payments/:id") and never the raw request URI, so payment IDs, +// merchant IDs, or delivery IDs never appear in metric label values regardless +// of how many unique identifiers flow through the service. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; fn lock_or_recover(mutex: &Mutex, what: &str) -> std::sync::MutexGuard<'_, T> { @@ -214,8 +228,10 @@ impl Default for AuthMetrics { } } -/// Counters for Horizon record handling that would otherwise only be visible in -/// logs. Currently tracks records the reconciler refused to credit. +/// Counters and gauges for Horizon record handling and poll cycle outcomes. +/// +/// All label values (outcome, reason) are fixed subsystem names — never raw +/// URLs, transaction hashes, payment IDs, or any per-tenant data. #[derive(Clone)] pub struct HorizonMetrics { inner: Arc, @@ -228,6 +244,20 @@ struct HorizonMetricsInner { /// non-zero value means an unexpected payload (a proxy, a mock, a /// truncated response) is reaching the reconciler (issue #224). unhashed_records_skipped: AtomicU64, + /// Successful Horizon poll cycles. + poll_success: AtomicU64, + /// Poll cycles that hit a Horizon rate limit (429 / 503 with Retry-After). + poll_rate_limited: AtomicU64, + /// Poll cycles that failed for any other reason (network error, 5xx, etc). + poll_error: AtomicU64, + /// Cursor incidents: three consecutive non-rate-limit 4xx responses from + /// Horizon for the same cursor position, indicating the cursor is invalid. + repeated_cursor_4xx: AtomicU64, + /// Times the Horizon SSE stream listener has reconnected. + stream_reconnects: AtomicU64, + /// Age in seconds of the most recently processed Horizon payment record, + /// as of the last poll or stream event. A store, not an accumulator. + cursor_age_secs: AtomicU64, } impl HorizonMetrics { @@ -247,6 +277,58 @@ impl HorizonMetrics { pub fn unhashed_records_skipped(&self) -> u64 { self.inner.unhashed_records_skipped.load(Ordering::Relaxed) } + + /// Record a successful Horizon poll cycle. + pub fn record_success(&self) { + self.inner.poll_success.fetch_add(1, Ordering::Relaxed); + } + + /// Record a rate-limited Horizon poll cycle. + pub fn record_rate_limited(&self) { + self.inner.poll_rate_limited.fetch_add(1, Ordering::Relaxed); + } + + /// Record a failed Horizon poll cycle (non-rate-limit error). + pub fn record_error(&self) { + self.inner.poll_error.fetch_add(1, Ordering::Relaxed); + } + + /// Record a repeated-cursor-4xx incident. + pub fn record_repeated_cursor_4xx(&self) { + self.inner.repeated_cursor_4xx.fetch_add(1, Ordering::Relaxed); + } + + /// Record one SSE stream reconnect. + pub fn record_stream_reconnect(&self) { + self.inner.stream_reconnects.fetch_add(1, Ordering::Relaxed); + } + + /// Store (overwrite) the age of the most recently processed Horizon record, + /// in seconds. This is a gauge — the latest value is what matters. + pub fn record_cursor_age_secs(&self, age: u64) { + self.inner.cursor_age_secs.store(age, Ordering::Relaxed); + } + + // ── Snapshot accessors ──────────────────────────────────────────────── + + pub fn success(&self) -> u64 { + self.inner.poll_success.load(Ordering::Relaxed) + } + pub fn rate_limited(&self) -> u64 { + self.inner.poll_rate_limited.load(Ordering::Relaxed) + } + pub fn error(&self) -> u64 { + self.inner.poll_error.load(Ordering::Relaxed) + } + pub fn repeated_cursor_4xx(&self) -> u64 { + self.inner.repeated_cursor_4xx.load(Ordering::Relaxed) + } + pub fn stream_reconnects(&self) -> u64 { + self.inner.stream_reconnects.load(Ordering::Relaxed) + } + pub fn cursor_age_secs(&self) -> u64 { + self.inner.cursor_age_secs.load(Ordering::Relaxed) + } } impl Default for HorizonMetrics { @@ -300,7 +382,7 @@ impl RouteLatency { /// labelled `` rather than the raw path, for the same reason. #[derive(Clone)] pub struct HttpMetrics { - inner: Arc>, + pub(crate) inner: Arc>, } #[derive(Default)] @@ -319,6 +401,13 @@ impl HttpMetrics { } /// Record one completed HTTP request. + /// + /// `route` MUST be the matched axum route template (e.g. `/v1/payments/:id`), + /// never the raw request URI. This is enforced by convention: the HTTP + /// metrics middleware extracts the route from axum's `MatchedPath` extension, + /// which only contains the template. Raw URIs contain path parameters + /// (payment IDs, merchant IDs) that would create unbounded label cardinality + /// and expose per-tenant identifiers in the metrics scrape. pub fn record(&self, method: &str, route: &str, status: u16, elapsed_ms: u64) { // A poisoned metrics mutex means a previous worker panicked while the // lock was held. Recover the underlying state instead of crashing the @@ -525,6 +614,9 @@ pub struct DbSnapshot { /// entry survives an outage rather than being overwritten by a guess. /// `last_success_unix` (0 until the first successful check) is how a scrape /// tells "we have never confirmed this" apart from "confirmed and stale". +/// +/// Label safety: the `asset` label contains only the asset code (e.g. "USDC", +/// "XLM") — never the issuer address, which is sensitive key-like material. #[derive(Clone)] pub struct TrustlineMetrics { inner: Arc, @@ -572,10 +664,10 @@ impl TrustlineMetrics { /// - `checked` — every non-native accepted asset the check evaluated. /// - `missing` — the subset with no usable trustline (absent or /// unauthorized). - /// - `unauthorized` — the subset where a trustline exists but - /// `is_authorized` is `false`. - /// - `headroom` — per-asset remaining capacity in stroops (`limit - - /// balance`), for assets where both values were parseable. + /// + /// This 2-argument form clears the unauthorized and headroom maps. + /// Use the 4-argument form `record_check_full` when those details are + /// available. /// /// Replaces the prior state for exactly the assets checked, so an asset /// removed from `ACCEPTED_ASSETS` between checks simply stops being @@ -584,6 +676,23 @@ impl TrustlineMetrics { &self, checked: impl IntoIterator, missing: &[String], + ) { + self.record_check_full(checked, missing, &[], &[]); + } + + /// Record a successful check with full detail. + /// + /// - `checked` — every non-native accepted asset the check evaluated. + /// - `missing` — the subset with no usable trustline (absent or + /// unauthorized). + /// - `unauthorized` — the subset where a trustline exists but + /// `is_authorized` is `false`. + /// - `headroom` — per-asset remaining capacity in stroops (`limit - + /// balance`), for assets where both values were parseable. + pub fn record_check_full<'a>( + &self, + checked: impl IntoIterator, + missing: &[String], unauthorized: &[String], headroom: &[(&str, i64)], ) { @@ -691,10 +800,25 @@ impl Default for TrustlineMetrics { // ── Prometheus text exposition ──────────────────────────────────────────────── -/// Render webhook delivery and auth outcome metrics as a Prometheus-compatible -/// plain-text snapshot. Called by `GET /metrics`. -pub fn render(webhook: &WebhookMetrics, auth: &AuthMetrics, horizon: &HorizonMetrics) -> String { - let mut out = String::with_capacity(1024); +/// Render all metrics as a Prometheus-compatible plain-text snapshot. +/// Called by `GET /metrics`. +/// +/// Label safety guarantee: every label value in the rendered output is a +/// fixed subsystem name, enum value, or bounded route template. No label +/// value is derived from request bodies, URL path parameters, merchant data, +/// or any other per-tenant identifier. See the LABEL SAFETY comment at the +/// top of this module for the full policy. +pub fn render( + webhook: &WebhookMetrics, + auth: &AuthMetrics, + tasks: &crate::TaskHealth, + horizon: &HorizonMetrics, + http: &HttpMetrics, + payments: &PaymentMetrics, + db: &DbSnapshot, + trustlines: &TrustlineMetrics, +) -> String { + let mut out = String::with_capacity(4096); // stellargate_webhook_deliveries_total — counter vec by outcome out.push_str( @@ -768,6 +892,32 @@ pub fn render(webhook: &WebhookMetrics, auth: &AuthMetrics, horizon: &HorizonMet "# HELP stellargate_horizon_records_skipped_total Horizon payment records the reconciler refused to credit, by reason.\n", ); out.push_str("# TYPE stellargate_horizon_records_skipped_total counter\n"); + out.push_str(&format!( + "stellargate_horizon_records_skipped_total{{reason=\"no_tx_hash\"}} {}\n", + horizon.unhashed_records_skipped() + )); + + // stellargate_tasks_* — background task health gauges and counters + out.push_str( + "# HELP stellargate_tasks_started_total Total background task starts (including restarts).\n", + ); + out.push_str("# TYPE stellargate_tasks_started_total counter\n"); + out.push_str(&format!( + "stellargate_tasks_started_total {}\n", + tasks.started() + )); + out.push_str( + "# HELP stellargate_tasks_stopped_total Total clean background task stops.\n", + ); + out.push_str("# TYPE stellargate_tasks_stopped_total counter\n"); + out.push_str(&format!( + "stellargate_tasks_stopped_total {}\n", + tasks.stopped() + )); + out.push_str( + "# HELP stellargate_tasks_failed_total Total background task panics.\n", + ); + out.push_str("# TYPE stellargate_tasks_failed_total counter\n"); out.push_str(&format!( "stellargate_tasks_failed_total {}\n", tasks.failed() @@ -1099,7 +1249,7 @@ pub fn render(webhook: &WebhookMetrics, auth: &AuthMetrics, horizon: &HorizonMet mod tests { use super::*; - fn empty_db_snapshot() -> DbSnapshot { + fn empty_db_snapshot() -> DbSnapshot { DbSnapshot { pool_size: 3, pool_idle: 1,