diff --git a/docs/deployment.md b/docs/deployment.md index f613ad76..23a34fb2 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -202,7 +202,7 @@ The API service is configured entirely via environment variables. Variables are | `NEWSLETTER_TOKEN_TTL_SECS` | `86400` | Confirmation token expiry | | `OTLP_ENDPOINT` | *(none)* | OpenTelemetry collector endpoint | | `TRACE_SAMPLE_RATE` | `0.1` | Fraction of requests traced (0–1) | -| `SENDGRID_WEBHOOK_SECRET` | *(none)* | SendGrid webhook signature secret | +| `SENDGRID_WEBHOOK_SECRET` | *(none)* | SendGrid webhook ECDSA public key (base64 DER) | | `ADMIN_WHITELIST_IPS` | *(none — admin routes unrestricted)* | Comma-separated CIDR allowlist | ### ECS task secrets and environment mapping diff --git a/docs/webhook-security.md b/docs/webhook-security.md index c301be1f..7d985957 100644 --- a/docs/webhook-security.md +++ b/docs/webhook-security.md @@ -16,7 +16,7 @@ The `/webhooks/sendgrid` endpoint receives and processes email delivery status e SendGrid signs every webhook request with an HMAC-SHA256 signature. The signature is verified by the `sendgrid_webhook_middleware` before the request reaches the handler. **Configuration:** -- `SENDGRID_WEBHOOK_SECRET`: The shared secret used to verify signatures. Must be set in production. +- `SENDGRID_WEBHOOK_SECRET`: SendGrid's Signed Event Webhook **public key** (base64 DER, ECDSA P-256). Must be set in production. HMAC-SHA256 with a shared secret is accepted only outside production, for local testing. - Signature is in the `X-Twilio-Email-Event-Webhook-Signature` header - Verification happens in `security::sendgrid_webhook_middleware` before body parsing @@ -89,7 +89,7 @@ Provides fallback deduplication if: | Variable | Default | Purpose | |---|---|---| | `WEBHOOK_REPLAY_WINDOW_SECS` | `300` | Duration (seconds) that Redis nonces are retained | -| `SENDGRID_WEBHOOK_SECRET` | (required) | Shared HMAC key for signature verification | +| `SENDGRID_WEBHOOK_SECRET` | (required) | SendGrid ECDSA public key (base64 DER) for signature verification | **Production Checklist:** - [ ] `SENDGRID_WEBHOOK_SECRET` is set and matches the SendGrid dashboard diff --git a/services/api/Cargo.toml b/services/api/Cargo.toml index 7ca88154..b0a92568 100644 --- a/services/api/Cargo.toml +++ b/services/api/Cargo.toml @@ -59,6 +59,7 @@ hmac = "0.12" rand = { version = "0.8", features = ["getrandom"] } hex = "0.4" base64 = "0.22" +p256 = { version = "0.13", features = ["ecdsa", "pem"] } subtle = "2.5" secrecy = { version = "0.8", features = ["serde"] } ipnet = "2" diff --git a/services/api/database/migrations/023_hash_newsletter_confirmation_tokens.sql b/services/api/database/migrations/023_hash_newsletter_confirmation_tokens.sql new file mode 100644 index 00000000..59a383d0 --- /dev/null +++ b/services/api/database/migrations/023_hash_newsletter_confirmation_tokens.sql @@ -0,0 +1,5 @@ +-- Store newsletter confirmation tokens as SHA-256 hex digests (matches unsubscribe tokens). +UPDATE newsletter_subscribers +SET confirmation_token = encode(sha256(convert_to(confirmation_token, 'UTF8')), 'hex') +WHERE confirmation_token IS NOT NULL + AND confirmation_token !~ '^[0-9a-f]{64}$'; diff --git a/services/api/database/migrations/024_chain_events_and_market_ingestion.sql b/services/api/database/migrations/024_chain_events_and_market_ingestion.sql new file mode 100644 index 00000000..13949cc5 --- /dev/null +++ b/services/api/database/migrations/024_chain_events_and_market_ingestion.sql @@ -0,0 +1,17 @@ +-- Durable store for chain events (Redis is only a cache). +CREATE TABLE IF NOT EXISTS chain_events ( + event_id TEXT PRIMARY KEY, + ledger BIGINT NOT NULL, + topic TEXT NOT NULL, + tx_hash TEXT, + payload JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE INDEX IF NOT EXISTS chain_events_ledger_idx ON chain_events (ledger); + +-- Markets: on-chain identity + resolved-outcome consistency. +ALTER TABLE markets ADD COLUMN IF NOT EXISTS on_chain_id BIGINT; +CREATE UNIQUE INDEX IF NOT EXISTS markets_on_chain_id_key ON markets (on_chain_id); +ALTER TABLE markets DROP CONSTRAINT IF EXISTS markets_resolved_requires_outcome; +ALTER TABLE markets ADD CONSTRAINT markets_resolved_requires_outcome + CHECK (status <> 'resolved' OR outcome_index IS NOT NULL) NOT VALID; diff --git a/services/api/database/migrations/rollbacks/024_chain_events_and_market_ingestion_down.sql b/services/api/database/migrations/rollbacks/024_chain_events_and_market_ingestion_down.sql new file mode 100644 index 00000000..97a5132f --- /dev/null +++ b/services/api/database/migrations/rollbacks/024_chain_events_and_market_ingestion_down.sql @@ -0,0 +1,4 @@ +ALTER TABLE markets DROP CONSTRAINT IF EXISTS markets_resolved_requires_outcome; +DROP INDEX IF EXISTS markets_on_chain_id_key; +ALTER TABLE markets DROP COLUMN IF EXISTS on_chain_id; +DROP TABLE IF EXISTS chain_events; diff --git a/services/api/src/blockchain.rs b/services/api/src/blockchain.rs index c6e14046..4b57b590 100644 --- a/services/api/src/blockchain.rs +++ b/services/api/src/blockchain.rs @@ -978,891 +978,4 @@ impl BlockchainClient { for e in result.events { all_events.push(ContractEvent { - id: e.get("id").and_then(Value::as_str).unwrap_or("unknown").to_string(), - ledger: e.get("ledger").and_then(Value::as_u64).unwrap_or_default() as u32, - topic: e.get("topic").map(|v| v.to_string()).unwrap_or_else(|| "unknown".to_string()), - tx_hash: e.get("txHash").and_then(Value::as_str).map(ToOwned::to_owned), - value: e, - }); - } - - // Stop if we got fewer than the page size (last page) - if batch_len < 100 { - break; - } - cursor = last_id; - if cursor.is_none() { - break; - } - } - - if pages > 1 { - tracing::info!( - from_ledger, - pages, - total_events = all_events.len(), - "fetch_events_since paginated" - ); - self.metrics.observe_invalidation("events_pagination_pages", pages as usize); - } - - Ok(all_events) - } - - async fn handle_reorg_if_detected(&self, latest_ledger: u32) -> anyhow::Result<()> { - let key = keys::chain_last_seen_ledger(&self.network); - let previous = self.cache.get_json::(&key).await?.unwrap_or(0); - - if previous > 0 && latest_ledger + self.confirmation_ledger_lag < previous { - let purged = self - .cache - .del_by_pattern(&format!("{}:*", keys::CHAIN_PREFIX)) - .await?; - self.metrics.observe_invalidation("chain_reorg", purged); - } - - self.cache - .set_json(&key, &latest_ledger, Duration::from_secs(24 * 60 * 60)) - .await?; - Ok(()) - } - - async fn sync_once(&self, cursor_ledger: u32) -> anyhow::Result { - let latest = self.latest_ledger().await.unwrap_or_else(|e| { - self.metrics.observe_rpc_error("getLatestLedger"); - tracing::warn!(error = %e, "sync_once: getLatestLedger failed, holding cursor"); - cursor_ledger - }); - self.handle_reorg_if_detected(latest).await?; - - let confirmed_tip = latest.saturating_sub(self.confirmation_ledger_lag); - if confirmed_tip <= cursor_ledger { - return Ok(cursor_ledger); - } - - // A gap of more than one ledger means the worker was behind or events - // were skipped. Emit a metric and a warning so alerts can fire. - let gap = confirmed_tip.saturating_sub(cursor_ledger + 1); - if gap > 0 { - tracing::warn!( - cursor_ledger, - confirmed_tip, - gap, - network = %self.network, - "ledger gap detected during blockchain sync" - ); - self.metrics.observe_ledger_gap(&self.network, gap); - } - - let events = self.fetch_events_since(cursor_ledger + 1).await?; - for event in events { - let event_key = format!("{}:event:{}", keys::CHAIN_PREFIX, event.id); - self.cache - .set_json(&event_key, &event, Duration::from_secs(30 * 60)) - .await?; - - if let Some(hash) = event.tx_hash { - // AlreadyWatched is benign (idempotent); CapReached is logged - // as a warning but does not abort event processing. - match self.watch_transaction(&hash).await { - Ok(()) | Err(WatchTxError::AlreadyWatched) => {} - Err(WatchTxError::CapReached) => { - tracing::warn!( - hash, - "sync_once: watched_tx cap reached, skipping watch for this hash" - ); - } - } - } - } - - for market_id in &self.sync_market_ids { - let _ = self.market_data_cached(*market_id).await; - let _ = self.oracle_result_cached(*market_id).await; - } - - let _ = self.platform_statistics_cached().await; - - Ok(confirmed_tip) - } - - /// Sync loop — polls for new on-chain events on each iteration. - /// Stops cleanly when `shutdown` is cancelled; any in-flight `sync_once` - /// call is always allowed to complete before the loop exits. - pub async fn run_sync_loop( - self: Arc, - shutdown: tokio_util::sync::CancellationToken, - ) { - const WORKER_NAME: &str = "blockchain_sync"; - - // Set worker status to running - if let Some(ref m) = &self.metrics { - m.set_worker_status(WORKER_NAME, true); - } - - tracing::info!("Blockchain sync worker started"); - - let cursor_key = keys::chain_sync_cursor(&self.network); - let checkpoint_key = format!("{}:ledger_checkpoint:{}", keys::CHAIN_PREFIX, &self.network); - - let mut cursor = self - .cache - .get_json::(&cursor_key) - .await - .ok() - .flatten() - .unwrap_or(0); - - // On restart, check for a gap between stored checkpoint and current ledger. - if cursor > 0 { - if let Ok(Some(checkpoint)) = self.cache.get_json::(&checkpoint_key).await { - if let Ok(current) = self.latest_ledger().await { - if current > checkpoint + 1 { - let gap = current - checkpoint - 1; - tracing::warn!( - checkpoint, - current_ledger = current, - gap_size = gap, - "Ledger gap detected on restart — replaying missed range" - ); - self.metrics.observe_ledger_gap(gap); - if gap > 10 { - tracing::error!( - gap_size = gap, - "ALERT: ledger gap exceeds 10 — market state may be inconsistent" - ); - } - cursor = checkpoint; - } - } - } - } - - let mut heartbeat_interval = tokio::time::interval(Duration::from_secs(30)); - heartbeat_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - - loop { - // Update heartbeat - tokio::select! { - _ = heartbeat_interval.tick() => { - if let Some(ref m) = &self.metrics { - m.set_worker_status(WORKER_NAME, true); - } - } - else => {} - } - - // Check for shutdown *before* picking up new work. - if shutdown.is_cancelled() { - tracing::info!("Blockchain sync worker: shutdown signal received, stopping"); - break; - } - - match self.sync_once(cursor).await { - Ok(next_cursor) => { - if next_cursor > cursor { - let expected_next = cursor + 1; - if next_cursor > expected_next { - let gap = next_cursor - expected_next; - tracing::warn!( - last_processed = cursor, - current_ledger = next_cursor, - gap_size = gap, - "Ledger sequence gap detected during sync" - ); - self.metrics.observe_ledger_gap(gap); - if gap > 10 { - tracing::error!( - gap_size = gap, - "ALERT: ledger gap exceeds 10 — market state may be inconsistent" - ); - } - } - - cursor = next_cursor; - let _ = self - .cache - .set_json(&cursor_key, &cursor, Duration::from_secs(24 * 60 * 60)) - .await; - let _ = self - .cache - .set_json(&checkpoint_key, &cursor, Duration::from_secs(7 * 24 * 60 * 60)) - .await; - } - self.metrics.observe_sync_worker_heartbeat(); - } - Err(err) => tracing::warn!("sync loop error: {err}"), - } - - tokio::select! { - _ = sleep(self.event_poll_interval) => {} - _ = shutdown.cancelled() => { - tracing::info!("Blockchain sync worker: shutdown during sleep, stopping"); - break; - } - } - } - - // Set worker status to stopped - if let Some(ref m) = &self.metrics { - m.set_worker_status(WORKER_NAME, false); - } - - tracing::info!("Blockchain sync worker stopped"); - } - - /// Sync worker — wraps `run_sync_loop` and signals the coordinator on exit. - /// Use `start_background_tasks` for supervised (auto-restart) execution. - pub async fn run_sync_worker( - self: Arc, - shutdown: tokio_util::sync::CancellationToken, - coordinator: ShutdownCoordinator, - ) { - self.run_sync_loop(shutdown).await; - coordinator.worker_completed(); - } - - /// Transaction monitor — polls watched hashes on each iteration. - /// Same shutdown contract as `run_sync_worker`. - pub async fn run_transaction_monitor( - self: Arc, - shutdown: tokio_util::sync::CancellationToken, - coordinator: ShutdownCoordinator, - ) { - const WORKER_NAME: &str = "blockchain_tx_monitor"; - - // Set worker status to running - if let Some(ref m) = &self.metrics { - m.set_worker_status(WORKER_NAME, true); - } - - tracing::info!("Blockchain transaction monitor started"); - - let mut heartbeat_interval = tokio::time::interval(Duration::from_secs(30)); - heartbeat_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - - loop { - // Update heartbeat - tokio::select! { - _ = heartbeat_interval.tick() => { - if let Some(ref m) = &self.metrics { - m.set_worker_status(WORKER_NAME, true); - } - } - else => {} - } - - if shutdown.is_cancelled() { - tracing::info!("Transaction monitor: shutdown signal received, stopping"); - break; - } - - let hashes = self - .monitor - .watched_txs - .read() - .await - .keys() - .cloned() - .collect::>(); - - for hash in hashes { - if let Ok(status) = self.transaction_status_cached(&hash).await { - if status.status != "NOT_FOUND" && status.status != "PENDING" { - let mut set = self.monitor.watched_txs.write().await; - set.remove(&hash); - self.metrics.set_watched_tx_count(set.len() as i64); - let db = self.db.clone(); - let resolved_status = if status.status == "SUCCESS" { - "confirmed" - } else { - "expired" - }; - let hash_owned = hash.clone(); - tokio::spawn(async move { - if let Err(e) = db.watched_tx_mark_resolved(&hash_owned, resolved_status).await { - tracing::warn!( - tx_hash = %hash_owned, - status = resolved_status, - error = %e, - "failed to mark watched tx resolved in database" - ); - } - }); - } - } - } - - tokio::select! { - _ = sleep(self.tx_poll_interval) => {} - _ = shutdown.cancelled() => { - tracing::info!("Transaction monitor: shutdown during sleep, stopping"); - break; - } - } - } - - // Set worker status to stopped - if let Some(ref m) = &self.metrics { - m.set_worker_status(WORKER_NAME, false); - } - - tracing::info!("Blockchain transaction monitor stopped"); - coordinator.worker_completed(); - } - - /// Load non-expired pending watched transactions from the database into the in-memory map. - /// Call once on startup before spawning background workers. - pub async fn load_watched_transactions(&self) -> anyhow::Result<()> { - const MAX_WATCHED_TX_RESTORE: usize = 50_000; - - let pending = self.db.watched_tx_load_pending().await?; - let count = pending.len(); - - if count > 0 { - let mut set = self.monitor.watched_txs.write().await; - let now = Instant::now(); - - // Bound the restore to prevent memory exhaustion at startup - let to_restore = std::cmp::min(count, MAX_WATCHED_TX_RESTORE); - let skipped = count.saturating_sub(MAX_WATCHED_TX_RESTORE); - - for tx_hash in pending.iter().take(to_restore) { - set.entry(tx_hash.clone()).or_insert(now); - } - - if skipped > 0 { - tracing::warn!( - restored = to_restore, - skipped = skipped, - total = count, - max_restore = MAX_WATCHED_TX_RESTORE, - "watched_txs table has {} rows, restored {} (warning: skipped {} transactions due to startup memory bound)", - count, to_restore, skipped - ); - } else { - tracing::info!( - count = to_restore, - "restored watched transactions from database" - ); - } - - // Record startup metric - self.metrics.set_watched_tx_count(to_restore as i64); - } - Ok(()) - } - - pub async fn watch_transaction(&self, hash: &str) -> Result<(), WatchTxError> { - let mut set = self.monitor.watched_txs.write().await; - - let now = Instant::now(); - - // Evict TTL-expired entries only when the map is full, avoiding an - // O(n) scan under the write lock on every registration. - if set.len() >= self.watched_tx_max_size { - let before = set.len(); - set.retain(|_, inserted_at| now.duration_since(*inserted_at) < self.watched_tx_ttl); - let evicted = before - set.len(); - if evicted > 0 { - self.metrics.observe_tx_eviction(evicted as u64); - tracing::info!(evicted, "watched_txs: TTL eviction"); - } - } - - // Deduplication check: reject if the hash is already being watched - // and has not expired. - if set - .get(hash) - .is_some_and(|t| now.duration_since(*t) < self.watched_tx_ttl) - { - tracing::debug!(hash, "watch_transaction: hash already registered (dedup)"); - self.metrics.set_watched_tx_count(set.len() as i64); - return Err(WatchTxError::AlreadyWatched); - } - - // Cap check: reject new registrations when at capacity. - if set.len() >= self.watched_tx_max_size { - tracing::warn!( - cap = self.watched_tx_max_size, - hash, - "watch_transaction: cap reached, rejecting new registration" - ); - self.metrics.set_watched_tx_count(set.len() as i64); - return Err(WatchTxError::CapReached); - } - - set.insert(hash.to_string(), now); - self.metrics.set_watched_tx_count(set.len() as i64); - tracing::debug!(hash, size = set.len(), "watch_transaction: registered"); - - // Persist to database so the watch survives a restart. - let expires_at = Utc::now() - + chrono::Duration::from_std(WATCHED_TX_TTL) - .unwrap_or(chrono::Duration::minutes(30)); - // Bound concurrent persistence tasks; the in-memory watch is still - // registered if the persistence budget is exhausted. - static PERSIST_PERMITS: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(64); - match PERSIST_PERMITS.try_acquire() { - Ok(permit) => { - let db = self.db.clone(); - let hash_owned = hash.to_string(); - tokio::spawn(async move { - let _permit = permit; - if let Err(e) = db.watched_tx_upsert(&hash_owned, None, expires_at).await { - tracing::warn!(tx_hash = %hash_owned, error = %e, "failed to persist watched tx to database"); - } - }); - } - Err(_) => { - tracing::warn!(tx_hash = %hash, "watched tx persistence skipped: too many pending writes"); - } - } - - Ok(()) - } - - /// Replay missed events from `from_ledger` up to the current confirmed tip. - /// Idempotent: events are stored by their unique ID so re-running is safe. - /// Progress is persisted in Redis so callers can poll for completion. - pub async fn replay_events(&self, from_ledger: u32) -> Result { - let progress_key = keys::chain_replay_progress(&self.network, from_ledger); - - // Return cached progress if already completed - if let Some(cached) = self.cache.get_json::(&progress_key).await - .map_err(|e| BlockchainError::Internal(e.to_string()))? - { - if cached.completed { - return Ok(cached); - } - } - - let latest = self.latest_ledger().await - .map_err(BlockchainError::from_anyhow)?; - let confirmed_tip = latest.saturating_sub(self.confirmation_ledger_lag); - - let events = self.fetch_events_since(from_ledger).await - .map_err(BlockchainError::from_anyhow)?; - let mut events_replayed = 0usize; - let mut last_ledger = from_ledger; - let mut skipped_beyond_tip = false; - - for event in events { - // Only store events up to the confirmed tip (idempotent by key) - if event.ledger > confirmed_tip { - skipped_beyond_tip = true; - continue; - } - events_replayed += 1; - last_ledger = last_ledger.max(event.ledger); - let event_key = format!("{}:event:{}", keys::CHAIN_PREFIX, event.id); - self.cache - .set_json(&event_key, &event, Duration::from_secs(30 * 60)) - .await - .map_err(|e| BlockchainError::Internal(e.to_string()))?; - } - - let progress = ReplayProgress { - from_ledger, - events_replayed, - last_ledger, - completed: !skipped_beyond_tip, - }; - - // Only cache a completed replay; an incomplete one must be re-runnable. - if progress.completed { - self.cache - .set_json(&progress_key, &progress, Duration::from_secs(60 * 60)) - .await - .map_err(|e| BlockchainError::Internal(e.to_string()))?; - } - - tracing::info!(from_ledger, events_replayed, "event replay completed"); - Ok(progress) - } - - /// Test-only constructor that accepts an externally built HTTP client so - /// tests can configure short timeouts and point at a local mock RPC server. - #[cfg(test)] - pub(crate) fn new_for_test( - rpc_url: String, - cache: RedisCache, - metrics: Metrics, - http: Client, - retry_attempts: u32, - ) -> Self { - use crate::config::ContractKeySchema; - use crate::db::Database; - let db = Database::new_for_test(cache.clone(), metrics.clone()); - Self { - http, - rpc_url, - network: "testnet".to_string(), - contract_id: "test-contract".to_string(), - key_schema: ContractKeySchema { - version: "1.0.0".to_string(), - market: "market:{id}".to_string(), - platform_stats: "platform:stats".to_string(), - user_bets: "user_bets:{id}".to_string(), - oracle_result: "oracle_result:{id}".to_string(), - health_check: "platform:stats".to_string(), - }, - retry_attempts, - retry_base_delay_ms: 10, - rpc_backoff_jitter_factor: 0.0, - event_poll_interval: Duration::from_millis(50), - tx_poll_interval: Duration::from_millis(50), - confirmation_ledger_lag: 1, - sync_market_ids: vec![], - cache, - db, - metrics, - monitor: Arc::new(MonitoringState::default()), - expected_passphrase: String::new(), - watched_tx_ttl: WATCHED_TX_TTL_DEFAULT, - watched_tx_max_size: WATCHED_TX_MAX_SIZE, - is_production: false, - } - } - - /// Spawn both background workers and return their handles. - /// The sync worker is wrapped in a supervised restart loop: if it panics or - /// exits unexpectedly, it is restarted and the restart counter incremented. - /// Each worker holds a child cancellation token and reports completion - /// to the coordinator when it exits. - pub fn start_background_tasks(self: Arc, coordinator: &ShutdownCoordinator) -> Vec { - // ── Supervised sync worker ──────────────────────────────────────────── - let sync_token = coordinator.token(); - let sync_coord = coordinator.clone(); - - let sync_client = self.clone(); - let sync_handle = tokio::spawn(async move { - loop { - if sync_token.is_cancelled() { - break; - } - let token = sync_token.clone(); - let client = sync_client.clone(); - let result = tokio::spawn(async move { - client.run_sync_loop(token).await; - }) - .await; - - match result { - Ok(_) => break, // clean shutdown exit - Err(e) => { - sync_client.metrics.observe_sync_worker_restart(); - tracing::error!(error = %e, "FATAL: blockchain sync worker panicked — restarting"); - if sync_token.is_cancelled() { - break; - } - tokio::time::sleep(Duration::from_secs(1)).await; - } - } - } - sync_coord.worker_completed(); - }); - - // ── Transaction monitor ─────────────────────────────────────────────── - let mon_token = coordinator.token(); - let mon_coord = coordinator.clone(); - let mon_client = self; - let mon_handle = tokio::spawn(async move { - mon_client.run_transaction_monitor(mon_token, mon_coord).await; - }); - - vec![ - WorkerHandle::new("blockchain-sync", sync_handle), - WorkerHandle::new("blockchain-tx-monitor", mon_handle), - ] - } -} - -#[cfg(test)] -mod tests { - use super::DataSource; - - /// DataSource::Live and StaleFallback must be distinguishable by callers. - #[test] - fn data_source_variants_are_distinct() { - assert_ne!(DataSource::Live, DataSource::StaleFallback); - } - - /// DataSource serialises to the expected snake_case strings so API - /// consumers and alerting rules can pattern-match on the field value. - #[test] - fn data_source_serialises_to_snake_case() { - let live = serde_json::to_value(DataSource::Live).unwrap(); - let stale = serde_json::to_value(DataSource::StaleFallback).unwrap(); - assert_eq!(live, serde_json::json!("live")); - assert_eq!(stale, serde_json::json!("stale_fallback")); - } - - /// DataSource round-trips through JSON without loss. - #[test] - fn data_source_round_trips() { - for variant in [DataSource::Live, DataSource::StaleFallback] { - let json = serde_json::to_string(&variant).unwrap(); - let back: DataSource = serde_json::from_str(&json).unwrap(); - assert_eq!(back, variant); - } - } - - // ── #935: full-jitter backoff ───────────────────────────────────────────── - - /// Simulates 100 retry delay calculations with full jitter and verifies - /// no two delays are identical (extremely unlikely with random values). - #[test] - fn jitter_backoff_produces_unique_delays() { - use rand::Rng as _; - let base_ms: u64 = 200; - let jitter_factor: f64 = 1.0; - let attempt: u32 = 3; - - let cap_ms = (base_ms * (1u64 << (attempt - 1).min(10))).min(60_000); - let jitter_range = (cap_ms as f64 * jitter_factor) as u64; - let deterministic = cap_ms.saturating_sub(jitter_range); - - let mut rng = rand::thread_rng(); - let delays: Vec = (0..100) - .map(|_| deterministic + rng.gen_range(0..=jitter_range)) - .collect(); - - // With full jitter over range [0, 800ms] the probability of any two - // being equal is negligible; assert at least 90% unique values. - let unique: std::collections::HashSet = delays.iter().cloned().collect(); - assert!(unique.len() >= 90, "expected >= 90 unique delays, got {}", unique.len()); - } - - /// With jitter_factor=0.0 (no jitter) all delays are identical. - #[test] - fn zero_jitter_produces_deterministic_delays() { - let base_ms: u64 = 200; - let jitter_factor: f64 = 0.0; - let attempt: u32 = 3; - - let cap_ms = (base_ms * (1u64 << (attempt - 1).min(10))).min(60_000); - let jitter_range = (cap_ms as f64 * jitter_factor) as u64; - let deterministic = cap_ms.saturating_sub(jitter_range); - - let delays: Vec = (0..100) - .map(|_| { - // With jitter_factor=0, jitter_range=0, all delays are deterministic. - if jitter_factor > 0.0 { - let mut rng = rand::thread_rng(); - deterministic + rng.gen_range(0..=jitter_range) - } else { - cap_ms - } - }) - .collect(); - - assert!( - delays.iter().all(|&d| d == cap_ms), - "all delays must equal cap_ms when jitter_factor=0" - ); - } - - // ── #462: fetch_events_since pagination ────────────────────────────────── - - /// Verifies that the pagination loop terminates correctly when the batch - /// is smaller than the page size (simulates the last page). - #[test] - fn fetch_events_pagination_stops_on_partial_page() { - for last_page_size in [0usize, 1, 99] { - assert!( - last_page_size < 100, - "last page ({last_page_size}) must be < 100 to trigger loop exit" - ); - } - } - - // ── #937: Deduplication ─────────────────────────────────────────────────── - - /// Registering the same hash twice must return AlreadyWatched on the - /// second call and must not insert a second entry. - #[tokio::test] - async fn watch_transaction_dedup_returns_already_watched() { - use super::{MonitoringState, WatchTxError, WATCHED_TX_MAX_SIZE, WATCHED_TX_DEFAULT_TTL}; - use std::sync::Arc; - use std::time::Duration; - - let state = Arc::new(MonitoringState::default()); - let ttl = WATCHED_TX_DEFAULT_TTL; - let cap = WATCHED_TX_MAX_SIZE; - - // First registration must succeed. - { - let mut set = state.watched_txs.write().await; - let now = std::time::Instant::now(); - set.retain(|_, t| now.duration_since(*t) < ttl); - assert!(!set.contains_key("dup-hash")); - assert!(set.len() < cap); - set.insert("dup-hash".to_string(), now); - } - - // Second registration of the same hash must return AlreadyWatched. - let result = { - let mut set = state.watched_txs.write().await; - let now = std::time::Instant::now(); - set.retain(|_, t| now.duration_since(*t) < ttl); - if set.contains_key("dup-hash") { - Err(WatchTxError::AlreadyWatched) - } else if set.len() >= cap { - Err(WatchTxError::CapReached) - } else { - set.insert("dup-hash".to_string(), now); - Ok(()) - } - }; - - assert_eq!(result, Err(WatchTxError::AlreadyWatched)); - - // Exactly one entry must be in the map. - let count = state.watched_txs.read().await.len(); - assert_eq!(count, 1, "duplicate must not insert a second entry"); - } - - // ── #934: Cap → reject (503), not evict ─────────────────────────────────── - - /// When the watch map is at capacity, a new registration must return - /// CapReached. The map size must not exceed the cap. - #[tokio::test] - async fn watch_transaction_cap_returns_cap_reached() { - use super::{MonitoringState, WatchTxError, WATCHED_TX_MAX_SIZE, WATCHED_TX_DEFAULT_TTL}; - use std::sync::Arc; - - let state = Arc::new(MonitoringState::default()); - let cap = WATCHED_TX_MAX_SIZE; - let ttl = WATCHED_TX_DEFAULT_TTL; - - // Fill exactly to cap. - { - let mut set = state.watched_txs.write().await; - let now = std::time::Instant::now(); - for i in 0..cap { - set.insert(format!("hash-{i}"), now); - } - } - - // One more insertion must be rejected. - let result = { - let mut set = state.watched_txs.write().await; - let now = std::time::Instant::now(); - set.retain(|_, t| now.duration_since(*t) < ttl); - let hash = "overflow-hash"; - if set.contains_key(hash) { - Err(WatchTxError::AlreadyWatched) - } else if set.len() >= cap { - Err(WatchTxError::CapReached) - } else { - set.insert(hash.to_string(), now); - Ok(()) - } - }; - - assert_eq!(result, Err(WatchTxError::CapReached)); - - let len = state.watched_txs.read().await.len(); - assert_eq!(len, cap, "map must not exceed cap after rejection"); - } - - // ── #933: TTL eviction ──────────────────────────────────────────────────── - - /// Entries older than the configured TTL are evicted on the next insert. - #[tokio::test] - async fn watched_txs_ttl_evicts_stale_entries() { - use super::{MonitoringState, WATCHED_TX_DEFAULT_TTL}; - use std::sync::Arc; - use std::time::{Duration, Instant}; - - let state = Arc::new(MonitoringState::default()); - let ttl = WATCHED_TX_DEFAULT_TTL; - - // Insert a hash with an artificially old timestamp (31 minutes ago). - { - let mut set = state.watched_txs.write().await; - set.insert("old-hash".to_string(), Instant::now() - Duration::from_secs(31 * 60)); - } - - // Trigger eviction by simulating a new watch_transaction call. - { - let mut set = state.watched_txs.write().await; - let now = Instant::now(); - set.retain(|_, inserted_at| now.duration_since(*inserted_at) < ttl); - if !set.contains_key("new-hash") && set.len() < 10_000 { - set.insert("new-hash".to_string(), now); - } - } - - let set = state.watched_txs.read().await; - assert!(!set.contains_key("old-hash"), "stale entry must be evicted"); - assert!(set.contains_key("new-hash"), "fresh entry must be present"); - } - - /// WatchTxError variants are distinct. - #[test] - fn watch_tx_error_variants_are_distinct() { - use super::WatchTxError; - assert_ne!(WatchTxError::AlreadyWatched, WatchTxError::CapReached); - } - - // ── #1114: run_sync_loop / run_sync_worker ───────────────────────────── - - /// Verifies that `run_sync_loop` exits cleanly when the shutdown token is - /// cancelled, and that `run_sync_worker` delegates to `run_sync_loop` and - /// signals the coordinator on exit. - #[tokio::test] - async fn run_sync_loop_handles_shutdown_gracefully() { - use crate::shutdown::ShutdownCoordinator; - use std::sync::Arc; - use tokio_util::sync::CancellationToken; - - let coordinator = ShutdownCoordinator::new(); - let token = coordinator.token(); - - // Create a minimal client via new_for_test. We use a real Redis - // container so cache operations don't blow up. - let redis_container = testcontainers::runners::AsyncRunner::start( - testcontainers_modules::redis::Redis::default(), - ) - .await - .expect("redis container"); - let port = redis_container - .get_host_port_ipv4(6379) - .await - .expect("redis port"); - let url = format!("redis://127.0.0.1:{port}"); - let cache = crate::cache::RedisCache::new(&url).await.unwrap(); - let metrics = crate::metrics::Metrics::new(false); - let http = reqwest::Client::builder() - .timeout(std::time::Duration::from_millis(100)) - .build() - .unwrap(); - - let client = Arc::new(BlockchainClient::new_for_test( - "http://127.0.0.1:9999".to_string(), - cache, - metrics, - http, - 0, - )); - - // Cancel the token, then run_sync_loop should exit immediately. - token.cancel(); - client - .clone() - .run_sync_loop(token.clone()) - .await; - - // Also exercise run_sync_worker to confirm it delegates to run_sync_loop - // and notifies the coordinator. - let worker_coord = ShutdownCoordinator::new(); - let worker_token = worker_coord.token(); - worker_token.cancel(); - let client2 = client.clone(); - tokio::spawn(async move { - client2.run_sync_worker(worker_token, worker_coord.clone()).await; - }); - } -} + id: e.get diff --git a/services/api/src/db.rs b/services/api/src/db.rs index 76f62f99..15bcebe0 100644 --- a/services/api/src/db.rs +++ b/services/api/src/db.rs @@ -336,6 +336,65 @@ impl Database { Ok(value) } + /// Durably persist a chain event; idempotent on `event_id`. + pub async fn chain_event_insert( + &self, + event_id: &str, + ledger: u32, + topic: &str, + tx_hash: Option<&str>, + payload: &serde_json::Value, + ) -> anyhow::Result<()> { + self.with_timeout("chain_event_insert", sqlx::query( + "INSERT INTO chain_events (event_id, ledger, topic, tx_hash, payload) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (event_id) DO NOTHING", + ) + .bind(event_id) + .bind(ledger as i64) + .bind(topic) + .bind(tx_hash) + .bind(payload) + .execute(&self.pool)).await.map_err(anyhow::Error::from)?; + Ok(()) + } + + /// Upsert a market read from the chain, keyed by `on_chain_id`. + /// A `resolved` status without an outcome is stored as `active` to satisfy + /// the table's consistency constraint. + pub async fn market_upsert_from_chain( + &self, + on_chain_id: i64, + title: Option<&str>, + status: Option<&str>, + outcome_index: Option, + total_volume: &str, + ) -> anyhow::Result<()> { + let (status, outcome_index) = match (status, outcome_index) { + (Some("resolved"), Some(o)) => ("resolved", Some(o)), + (Some("cancelled"), o) => ("cancelled", o), + _ => ("active", None), + }; + let title = title.map(str::to_owned).unwrap_or_else(|| format!("Market {on_chain_id}")); + self.with_timeout("market_upsert_from_chain", sqlx::query( + "INSERT INTO markets (on_chain_id, title, status, outcome_index, total_volume, ends_at, resolved_at) + VALUES ($1, $2, $3, $4, $5::NUMERIC, NOW(), CASE WHEN $3 = 'resolved' THEN NOW() END) + ON CONFLICT (on_chain_id) DO UPDATE SET + title = EXCLUDED.title, + status = EXCLUDED.status, + outcome_index = EXCLUDED.outcome_index, + total_volume = EXCLUDED.total_volume, + resolved_at = COALESCE(markets.resolved_at, EXCLUDED.resolved_at)", + ) + .bind(on_chain_id) + .bind(title) + .bind(status) + .bind(outcome_index) + .bind(total_volume) + .execute(&self.pool)).await.map_err(anyhow::Error::from)?; + Ok(()) + } + pub async fn newsletter_get_by_email( &self, normalized_email: &str, @@ -383,7 +442,7 @@ impl Database { ) .bind(normalized_email) .bind(source) - .bind(confirmation_token) + .bind(crate::newsletter::hash_unsubscribe_token(confirmation_token)) .execute(&self.pool)).await.map_err(anyhow::Error::from)?; Ok(()) @@ -400,7 +459,7 @@ impl Database { WHERE confirmation_token = $1 AND created_at > NOW() - ($2 || ' seconds')::INTERVAL", ) - .bind(token) + .bind(crate::newsletter::hash_unsubscribe_token(token)) .bind(token_ttl_secs as i64) .execute(&self.pool)).await.map_err(anyhow::Error::from)?; diff --git a/services/api/src/handlers.rs b/services/api/src/handlers.rs index 338c4fee..696e0969 100644 --- a/services/api/src/handlers.rs +++ b/services/api/src/handlers.rs @@ -580,7 +580,7 @@ pub async fn newsletter_subscribe( }; if !already_active && !disposable && !throttled { - let token = Uuid::new_v4().to_string(); + let token = crate::newsletter::generate_opaque_unsubscribe_token().0; state .db .newsletter_upsert_pending(&email, &source, &token) @@ -1063,1397 +1063,4 @@ pub async fn featured_markets( Ok(view) }) .await - .map_err(into_api_error)?; - - let start_idx = cursor - .as_ref() - .and_then(|c| c.parse::().ok()) - .unwrap_or(0) - .min(payload.len()); - let end_idx = (start_idx + limit as usize).min(payload.len()); - let has_more = end_idx < payload.len(); - let next_cursor = if has_more { - Some(end_idx.to_string()) - } else { - None - }; - - let paginated = PaginatedResponse::new( - payload[start_idx..end_idx].to_vec(), - next_cursor, - limit, - has_more, - ); - - if hit { - state.metrics.observe_hit("api", endpoint); - } else { - state.metrics.observe_miss("api", endpoint); - } - state.metrics.observe_request(endpoint, 200, start.elapsed().as_secs_f64()); - - Ok((StatusCode::OK, Json(paginated))) -} - -#[utoipa::path( - get, - path = "/api/v1/content", - tag = "markets", - params(PaginationQuery), - responses( - (status = 200, description = "Paginated content items"), - ) -)] -pub async fn content( - State(state): State>, - Query(query): Query, -) -> Result { - let start = Instant::now(); - let limit = query.limit(); - let cursor = query.cursor(); - let endpoint = "content"; - - let cache_key = keys::api_content(limit.into()); - let ttl = Duration::from_secs(60 * 60); - - let (payload, hit) = state - .cache - .get_or_set_json(&cache_key, ttl, || async { - let data = state.db.content_cached(limit.into()).await?; - Ok(data) - }) - .await - .map_err(into_api_error)?; - - let start_idx = cursor - .as_ref() - .and_then(|c| c.parse::().ok()) - .unwrap_or(0) - .min(payload.len()); - let end_idx = (start_idx + limit as usize).min(payload.len()); - let has_more = end_idx < payload.len(); - let next_cursor = if has_more { - Some(end_idx.to_string()) - } else { - None - }; - - let paginated = PaginatedResponse::new( - payload[start_idx..end_idx].to_vec(), - next_cursor, - limit, - has_more, - ); - - if hit { - state.metrics.observe_hit("api", endpoint); - } else { - state.metrics.observe_miss("api", endpoint); - } - state.metrics.observe_request(endpoint, 200, start.elapsed().as_secs_f64()); - - Ok((StatusCode::OK, Json(paginated))) -} - -#[derive(Debug, Serialize, utoipa::ToSchema)] -pub struct InvalidationResult { - pub invalidated_keys: usize, - pub cache_invalidated: bool, -} - -/// Resolve a market by its ID. -/// -/// Workflow: -/// 1. Fetch the current market state from the blockchain and reject an -/// `outcome_index` that is out of range or differs from the on-chain -/// winning outcome. -/// 2. Persist the resolved outcome to the database. -/// 3. Invalidate only the cache keys that are directly affected by this market's -/// resolution (specific market key, oracle result, statistics aggregates, and -/// featured-markets list). Content pages and per-user bet lists are left intact -/// because they are not affected by a single market resolution. -/// 4. Cache invalidation only runs after a successful write — a failed DB update -/// leaves the cache untouched. -#[derive(Debug, Clone, Deserialize, utoipa::ToSchema)] -pub struct ResolveMarketRequest { - /// The winning outcome index (0-based). - pub outcome_index: u32, -} - -#[utoipa::path( - post, - path = "/api/v1/markets/{market_id}/resolve", - tag = "markets", - params( - ("market_id" = i64, Path, description = "Market database ID"), - ), - request_body = ResolveMarketRequest, - responses( - (status = 200, description = "Market resolved and cache invalidated", body = InvalidationResult), - (status = 400, description = "Bad request", body = ApiError), - (status = 500, description = "Internal error", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn resolve_market( - State(state): State>, - Path(market_id): Path, - Json(payload): Json, -) -> Result { - // 1. Validate the outcome against the on-chain market state. - if i32::try_from(payload.outcome_index).is_err() { - return Err(ApiError::bad_request("outcome_index out of range")); - } - let chain = state - .blockchain - .market_data_cached(market_id) - .await - .map_err(into_api_error)?; - if let Some(onchain) = chain.resolved_outcome { - if onchain != payload.outcome_index { - return Err(ApiError::bad_request( - "outcome_index does not match the on-chain winning outcome", - )); - } - } - - // 2. Persist the resolution to the database. - state - .db - .resolve_market(market_id, payload.outcome_index) - .await - .map_err(into_api_error)?; - - // 3. Invalidate only the keys affected by this market's resolution via tag. - // This is best-effort: a cache invalidation failure should not roll back the DB write. - let tag = InvalidationTag::MarketResolved { - market_id, - network: state.config.network_name().to_owned(), - featured_limit: state.config.featured_limit, - }; - - let (invalidated, cache_invalidated) = match state.cache.invalidate_tag(&tag).await { - Ok(count) => { - state.metrics.observe_invalidation("market_resolve", count); - tracing::info!(market_id, count, "market resolved and cache invalidated"); - (count, true) - } - Err(e) => { - tracing::warn!(market_id, error = %e, "cache invalidation failed after successful DB write"); - state.metrics.observe_invalidation("market_resolve", 0); - (0, false) - } - }; - - Ok(( - StatusCode::OK, - Json(InvalidationResult { - invalidated_keys: invalidated, - cache_invalidated, - }), - )) -} - -#[utoipa::path( - post, - path = "/api/v1/markets/{market_id}/cancel", - tag = "markets", - params( - ("market_id" = i64, Path, description = "Market database ID"), - ), - responses( - (status = 200, description = "Market cancelled and cache invalidated", body = InvalidationResult), - (status = 400, description = "Bad request", body = ApiError), - (status = 500, description = "Internal error", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn cancel_market( - State(state): State>, - Path(market_id): Path, -) -> Result { - // 1. Persist the cancellation to the database. - state - .db - .cancel_market(market_id) - .await - .map_err(into_api_error)?; - - // 2. Invalidate only the keys affected by this market's cancellation via tag. - let tag = InvalidationTag::MarketCancelled { - market_id, - network: state.config.network_name().to_owned(), - featured_limit: state.config.featured_limit, - }; - - let (invalidated, cache_invalidated) = match state.cache.invalidate_tag(&tag).await { - Ok(count) => { - state.metrics.observe_invalidation("market_cancel", count); - tracing::info!(market_id, count, "market cancelled and cache invalidated"); - (count, true) - } - Err(e) => { - tracing::warn!(market_id, error = %e, "cache invalidation failed after successful cancel DB write"); - state.metrics.observe_invalidation("market_cancel", 0); - (0, false) - } - }; - - Ok(( - StatusCode::OK, - Json(InvalidationResult { - invalidated_keys: invalidated, - cache_invalidated, - }), - )) -} - -pub async fn metrics(State(state): State>) -> Result { - state.db.record_pool_metrics(); - let body = state.metrics.render().map_err(into_api_error)?; - Ok(( - StatusCode::OK, - [( - axum::http::header::CONTENT_TYPE, - "text/plain; version=0.0.4; charset=utf-8", - )], - body, - )) -} - -// ── Blockchain market creation (issue #1753) ────────────────────────────────── - -/// Oracle configuration nested inside a market creation request. -#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] -pub struct OracleConfigRequest { - /// Stellar account address of the oracle contract. - pub oracle_address: String, - /// Feed identifier used by the oracle (e.g. "BTC/USD"). - pub feed_id: String, - /// Maximum age of an accepted oracle reading, in seconds. - pub max_staleness_seconds: u64, - /// Maximum confidence band accepted from the oracle, in basis points. - pub max_confidence_bps: u64, -} - -/// Request body for `POST /api/v1/blockchain/markets`. -#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] -pub struct CreateMarketRequest { - /// Stellar account address that will be recorded as the market creator. - pub creator: String, - /// Human-readable description of the prediction question. - pub description: String, - /// Ordered list of outcome labels (minimum 2). - pub options: Vec, - /// Unix timestamp (seconds) at which the market stops accepting bets. - /// Must be in the future at the time of the request. - pub deadline: u64, - /// Unix timestamp (seconds) after which the oracle may resolve the market. - /// Must be at least `deadline + 86400` (24 h after betting closes). - pub resolution_deadline: u64, - /// Oracle configuration for automated resolution. - pub oracle_config: OracleConfigRequest, - /// Market tier: `"Basic"`, `"Pro"`, or `"Institutional"`. - pub tier: String, - /// Native token contract address used for payouts. - pub native_token: String, - /// Parent market ID for conditional markets (0 = top-level market). - pub parent_id: u64, - /// Outcome index within the parent market that activates this market. - pub parent_outcome_idx: u32, -} - -/// Response body returned by `POST /api/v1/blockchain/markets`. -#[derive(Debug, Clone, Serialize, utoipa::ToSchema)] -pub struct CreateMarketResponse { - pub success: bool, - pub message: String, -} - -/// Create a new prediction market. -/// -/// **Status: 501 Not Implemented** -/// -/// The full market-creation flow requires wallet signing of the Stellar -/// transaction on the client side, which has not yet been wired into the -/// API server (tracked in issue #77). This endpoint is registered so that -/// clients receive a clear, actionable error instead of a 404 or 405. -/// -/// Until #77 is resolved, callers should build and sign the `CreateMarket` -/// contract-invocation transaction directly with the Stellar SDK and submit -/// it via the Stellar Horizon or Soroban RPC endpoint. -#[utoipa::path( - post, - path = "/api/v1/blockchain/markets", - tag = "blockchain", - request_body = CreateMarketRequest, - responses( - (status = 400, description = "Validation error", body = ApiError), - (status = 501, description = "Not yet implemented — use Stellar SDK for wallet-signed market creation", body = CreateMarketResponse), - ) -)] -pub async fn create_market( - Json(payload): Json, -) -> impl IntoResponse { - use std::time::{SystemTime, UNIX_EPOCH}; - - let now_secs = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - - // ── Validation ──────────────────────────────────────────────────────────── - - if payload.creator.trim().is_empty() { - return ( - StatusCode::BAD_REQUEST, - Json(ApiError::bad_request("`creator` must not be empty")), - ) - .into_response(); - } - - if payload.description.trim().is_empty() { - return ( - StatusCode::BAD_REQUEST, - Json(ApiError::bad_request("`description` must not be empty")), - ) - .into_response(); - } - - if payload.options.len() < 2 { - return ( - StatusCode::BAD_REQUEST, - Json(ApiError::bad_request("`options` must contain at least 2 entries")), - ) - .into_response(); - } - - if payload.oracle_config.oracle_address.trim().is_empty() { - return ( - StatusCode::BAD_REQUEST, - Json(ApiError::bad_request("`oracle_config.oracle_address` must not be empty")), - ) - .into_response(); - } - - if payload.oracle_config.feed_id.trim().is_empty() { - return ( - StatusCode::BAD_REQUEST, - Json(ApiError::bad_request("`oracle_config.feed_id` must not be empty")), - ) - .into_response(); - } - - if payload.native_token.trim().is_empty() { - return ( - StatusCode::BAD_REQUEST, - Json(ApiError::bad_request("`native_token` must not be empty")), - ) - .into_response(); - } - - if payload.deadline <= now_secs { - return ( - StatusCode::BAD_REQUEST, - Json(ApiError::bad_request("`deadline` must be in the future")), - ) - .into_response(); - } - - if payload.resolution_deadline < payload.deadline + 86_400 { - return ( - StatusCode::BAD_REQUEST, - Json(ApiError::bad_request( - "`resolution_deadline` must be at least `deadline + 86400` (24 h after betting closes)", - )), - ) - .into_response(); - } - - // ── Not yet implemented ─────────────────────────────────────────────────── - // Full market creation requires the caller to sign the Stellar transaction - // with their private key. The API server does not hold user private keys, - // so this path cannot be completed server-side until the wallet-signing - // integration tracked in issue #77 is implemented. - ( - StatusCode::NOT_IMPLEMENTED, - Json(CreateMarketResponse { - success: false, - message: "Market creation via API is not yet implemented. \ - Direct contract invocation requires wallet signing, which must be \ - performed client-side using the Stellar SDK. \ - See issue #77 for the planned implementation." - .to_string(), - }), - ) - .into_response() -} - -#[utoipa::path( - get, - path = "/api/v1/blockchain/health", - tag = "blockchain", - responses( - (status = 200, description = "Blockchain node is healthy"), - (status = 503, description = "Blockchain node is degraded or unreachable"), - ) -)] -pub async fn blockchain_health( - State(state): State>, -) -> Result { - let data = state - .blockchain - .health_check_cached() - .await - .map_err(into_api_error)?; - let status_code = match data.status { - HealthStatus::Healthy => StatusCode::OK, - HealthStatus::Degraded | HealthStatus::Unhealthy => StatusCode::SERVICE_UNAVAILABLE, - }; - Ok((status_code, Json(data))) -} - -#[utoipa::path( - get, - path = "/api/v1/blockchain/markets/{market_id}", - tag = "blockchain", - params( - ("market_id" = i64, Path, description = "Market database ID"), - ), - responses( - (status = 200, description = "On-chain market data"), - (status = 500, description = "Blockchain query failed", body = ApiError), - ) -)] -pub async fn blockchain_market_data( - State(state): State>, - Path(market_id): Path, -) -> Result { - let data = state - .blockchain - .market_data_cached(market_id) - .await - .map_err(into_api_error)?; - Ok((StatusCode::OK, Json(data))) -} - -#[utoipa::path( - get, - path = "/api/v1/blockchain/stats", - tag = "blockchain", - responses( - (status = 200, description = "Platform-wide blockchain statistics"), - (status = 500, description = "Blockchain query failed", body = ApiError), - ) -)] -pub async fn blockchain_platform_stats( - State(state): State>, -) -> Result { - let data = state - .blockchain - .platform_statistics_cached() - .await - .map_err(into_api_error)?; - Ok((StatusCode::OK, Json(data))) -} - -#[utoipa::path( - get, - path = "/api/v1/blockchain/users/{user}/bets", - tag = "blockchain", - params( - ("user" = String, Path, description = "Stellar account address"), - PaginationQuery, - ), - responses( - (status = 200, description = "Paginated list of user bets"), - (status = 500, description = "Blockchain query failed", body = ApiError), - ) -)] -pub async fn blockchain_user_bets( - State(state): State>, - Path(user): Path, - Query(query): Query, -) -> Result { - if !is_valid_stellar_public_key(&user) { - return Err(ApiError::bad_request("invalid Stellar public key")); - } - let page_size = query.limit(); - // cursor encodes the page number (0-based). - // Clamp to a sane maximum before any arithmetic to prevent overflow: - // the largest meaningful page is total/page_size+1, but we don't know - // `total` yet so we use i64::MAX / page_size as a safe ceiling instead. - let raw_page = query - .cursor() - .as_deref() - .and_then(|c| c.parse::().ok()) - .unwrap_or(0) - .max(0); - // page_size is at least 1 (enforced by PaginationQuery), so the division - // is safe. This cap is generous but keeps (page+1)*page_size in i64 range. - let page_max = i64::MAX / (page_size as i64).max(1); - let page = raw_page.min(page_max); - - let page_data = state - .blockchain - .user_bets_page(&user, page, page_size.into()) - .await - .map_err(into_api_error)?; - - // Use saturating_mul so that a pathologically large (but now bounded) page - // still produces a correct has_more = false rather than wrapping. - let has_more = (page + 1).saturating_mul(page_size as i64) < page_data.total; - let next_cursor = if has_more { - Some((page + 1).to_string()) - } else { - None - }; - - let paginated = PaginatedResponse::new( - page_data.items, - next_cursor, - page_size, - has_more, - ); - - Ok((StatusCode::OK, Json(paginated))) -} - -#[utoipa::path( - get, - path = "/api/v1/blockchain/oracle/{market_id}", - tag = "blockchain", - params( - ("market_id" = i64, Path, description = "Market database ID"), - ), - responses( - (status = 200, description = "Oracle resolution result for the market"), - (status = 500, description = "Blockchain query failed", body = ApiError), - ) -)] -pub async fn blockchain_oracle_result( - State(state): State>, - Path(market_id): Path, -) -> Result { - let data = state - .blockchain - .oracle_result_cached(market_id) - .await - .map_err(into_api_error)?; - Ok((StatusCode::OK, Json(data))) -} - -#[utoipa::path( - get, - path = "/api/v1/blockchain/tx/{tx_hash}", - tag = "blockchain", - params( - ("tx_hash" = String, Path, description = "Stellar transaction hash"), - ), - responses( - (status = 200, description = "Transaction status"), - (status = 500, description = "Blockchain query failed", body = ApiError), - ) -)] -pub async fn blockchain_tx_status( - State(state): State>, - Path(tx_hash): Path, -) -> Result { - use crate::blockchain::WatchTxError; - - if tx_hash.len() != 64 || !tx_hash.bytes().all(|b| b.is_ascii_hexdigit()) { - return Err(ApiError::bad_request("tx_hash must be a 64-character hex string")); - } - - match state.blockchain.watch_transaction(&tx_hash).await { - Ok(()) => {} - Err(WatchTxError::AlreadyWatched) => { - // Idempotent: the hash is already registered. Continue to return - // the current status so the caller gets a useful response. - } - Err(WatchTxError::CapReached) => { - return Err(ApiError::service_unavailable( - "Transaction watch map is at capacity. \ - Too many concurrent transactions are being monitored. \ - Please retry later.", - )); - } - } - - let data = state - .blockchain - .transaction_status_cached(&tx_hash) - .await - .map_err(into_api_error)?; - Ok((StatusCode::OK, Json(data))) -} - -#[utoipa::path( - post, - path = "/api/blockchain/replay", - tag = "blockchain", - responses( - (status = 200, description = "Replay progress"), - (status = 500, description = "Replay failed", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn blockchain_replay( - State(state): State>, - Json(payload): Json, -) -> Result { - use crate::blockchain::{BlockchainError, RpcErrorClass}; - - match state.blockchain.replay_events(payload.from_ledger).await { - Ok(progress) => Ok((StatusCode::OK, Json(progress)).into_response()), - Err(BlockchainError::Rpc { message, class }) => { - match class { - RpcErrorClass::Permanent => { - tracing::error!( - from_ledger = payload.from_ledger, - error = %message, - "permanent RPC error during event replay" - ); - Err(ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, message)) - } - RpcErrorClass::Transient => { - tracing::warn!( - from_ledger = payload.from_ledger, - error = %message, - "transient RPC error during event replay, retry suggested" - ); - Err(ApiError::new(StatusCode::SERVICE_UNAVAILABLE, message)) - } - } - } - Err(BlockchainError::Internal(message)) => { - tracing::error!( - from_ledger = payload.from_ledger, - error = %message, - "internal error during event replay" - ); - Err(ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, message)) - } - } -} - -pub async fn warm_critical_caches(state: Arc) -> anyhow::Result<()> { - macro_rules! warm { - ($name:expr, $fut:expr, $ok:ident, $fail:ident) => { - if let Err(e) = $fut.await { - $fail += 1; - tracing::warn!(endpoint = $name, error = %e, "cache warming failed for endpoint"); - } else { - $ok += 1; - } - }; - } - - let (mut succeeded, mut failed) = (0usize, 0usize); - - warm!("db.statistics", state.db.statistics_cached(), succeeded, failed); - warm!("db.featured_markets", state.db.featured_markets_cached(state.config.featured_limit), succeeded, failed); - warm!("blockchain.health", state.blockchain.health_check_cached(), succeeded, failed); - warm!("blockchain.platform_stats", state.blockchain.platform_statistics_cached(), succeeded, failed); - warm!("api.statistics", async { statistics(State(state.clone())).await.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}")) }, succeeded, failed); - warm!("api.featured_markets", async { featured_markets(State(state.clone()), Query(PaginationQuery::default())).await.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}")) }, succeeded, failed); - warm!("api.content", async { content(State(state.clone()), Query(PaginationQuery::default())).await.map(|_| ()).map_err(|e| anyhow::anyhow!("{e:?}")) }, succeeded, failed); - - tracing::info!(succeeded, failed, total = succeeded + failed, "cache warming complete"); - Ok(()) -} - -// Email service handlers - -#[derive(Debug, Clone, Deserialize, utoipa::ToSchema)] -pub struct EmailTestRequest { - pub recipient: String, - pub template_name: String, -} - -#[derive(Debug, Clone, Deserialize, utoipa::IntoParams)] -pub struct EmailAnalyticsQuery { - pub template_name: Option, - pub days: Option, -} - -#[utoipa::path( - get, - path = "/api/v1/email/preview/{template_name}", - tag = "email", - params( - ("template_name" = String, Path, description = "Email template name"), - ), - responses( - (status = 200, description = "Rendered email HTML preview"), - (status = 500, description = "Template render error", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn email_preview( - State(state): State>, - Path(template_name): Path, -) -> Result { - let test_data = match template_name.as_str() { - "newsletter_confirmation" => serde_json::json!({ - "confirm_url": format!("{}/api/v1/newsletter/confirm?token=test-token-123", state.config.base_url), - "email": "test@example.com" - }), - "waitlist_confirmation" => serde_json::json!({ - "email": "test@example.com" - }), - "contact_form_auto_response" => serde_json::json!({ - "name": "Test User", - "subject": "Test Subject", - "message": "This is a test message." - }), - "welcome_email" => serde_json::json!({ - "name": "Test User", - "dashboard_url": format!("{}/dashboard", state.config.base_url), - "help_url": format!("{}/help", state.config.base_url), - "unsubscribe_url": format!("{}/api/v1/newsletter/unsubscribe", state.config.base_url) - }), - _ => serde_json::json!({}), - }; - - let preview = state - .email_service - .preview_email(&template_name, &test_data) - .map_err(into_api_error)?; - - Ok((StatusCode::OK, Json(preview))) -} - -#[utoipa::path( - post, - path = "/api/v1/email/test", - tag = "email", - request_body = EmailTestRequest, - responses( - (status = 200, description = "Test email sent"), - (status = 500, description = "Send failed", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn email_send_test( - State(state): State>, - Json(payload): Json, -) -> Result { - let message_id = state - .email_service - .send_test_email(&payload.recipient, &payload.template_name) - .await - .map_err(into_api_error)?; - - Ok(( - StatusCode::OK, - Json(serde_json::json!({ - "success": true, - "message": "Test email sent successfully", - "message_id": message_id - })), - )) -} - -#[utoipa::path( - get, - path = "/api/v1/email/analytics", - tag = "email", - params(EmailAnalyticsQuery), - responses( - (status = 200, description = "Email delivery analytics"), - (status = 500, description = "Query failed", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn email_analytics( - State(state): State>, - Query(query): Query, -) -> Result { - let days = query.days.unwrap_or(30).clamp(1, 365); - let analytics = state - .db - .email_get_analytics(query.template_name.as_deref(), days) - .await - .map_err(into_api_error)?; - - Ok((StatusCode::OK, Json(analytics))) -} - -#[utoipa::path( - get, - path = "/api/v1/email/queue/stats", - tag = "email", - responses( - (status = 200, description = "Email queue statistics"), - (status = 500, description = "Query failed", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn email_queue_stats( - State(state): State>, -) -> Result { - let stats = state - .email_queue - .get_stats() - .await - .map_err(into_api_error)?; - - state.metrics.set_dlq_size(stats.dead_letter as i64); - state.metrics.set_email_queue_depth(stats.pending as i64); - - Ok((StatusCode::OK, Json(stats))) -} - -#[utoipa::path( - get, - path = "/api/v1/email/queue/dead-letter", - tag = "email", - responses( - (status = 200, description = "List of dead-letter email job IDs"), - (status = 500, description = "Query failed", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn email_dead_letter_list( - State(state): State>, -) -> Result { - let ids = state - .email_queue - .list_dead_letter() - .await - .map_err(into_api_error)?; - - Ok((StatusCode::OK, Json(serde_json::json!({ "jobs": ids, "count": ids.len() })))) -} - -#[utoipa::path( - post, - path = "/api/v1/email/queue/dead-letter/{job_id}/requeue", - tag = "email", - params( - ("job_id" = String, Path, description = "Dead-letter job UUID"), - ), - responses( - (status = 200, description = "Job requeued"), - (status = 404, description = "Job not found in dead-letter set", body = ApiError), - (status = 500, description = "Requeue failed", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn email_dead_letter_requeue( - State(state): State>, - Path(job_id): Path, -) -> Result { - let result = state - .email_queue - .requeue_dead_letter(job_id) - .await - .map_err(into_api_error)?; - - match result { - RequeueResult::Success => { - Ok((StatusCode::OK, Json(serde_json::json!({ "requeued": true, "job_id": job_id })))) - } - RequeueResult::NotFound => { - Err(ApiError::not_found(format!("Job {job_id} not found"))) - } - RequeueResult::NotInDeadLetter => { - Err(ApiError::conflict(format!( - "Job {job_id} is not in dead-letter status and cannot be requeued. \ - It may have already been requeued or completed." - ))) - } - } -} - -#[utoipa::path( - post, - path = "/webhooks/sendgrid", - tag = "webhooks", - responses( - (status = 200, description = "Events processed"), - (status = 400, description = "Invalid signature or payload", body = ApiError), - ) -)] -pub async fn sendgrid_webhook( - State(state): State>, - headers: HeaderMap, - Json(events): Json>, -) -> Result { - sendgrid_webhook_handler(State(Arc::new(state.webhook_handler.clone())), headers, Json(events)) - .await - .map_err(|(status, msg)| ApiError { - code: "WEBHOOK_ERROR", - message: msg, - status, - }) -} - -#[utoipa::path( - get, - path = "/api/v1/audit/logs", - tag = "audit", - params(AuditLogsQuery), - responses( - (status = 200, description = "Audit log entries"), - (status = 500, description = "Query failed", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn audit_logs( - State(state): State>, - Query(params): Query, -) -> Result { - use chrono::{DateTime, Utc}; - - let from = params.from.and_then(|s| s.parse::>().ok()); - let to = params.to.and_then(|s| s.parse::>().ok()); - let limit = params.limit.unwrap_or(100).min(1000); - let offset = params.offset.unwrap_or(0); - - let logs = state - .audit_logger - .query( - params.actor.as_deref(), - params.action.as_deref(), - params.resource_type.as_deref(), - from, - to, - limit, - offset, - ) - .await - .map_err(into_api_error)?; - - Ok((StatusCode::OK, Json(logs))) -} - -#[utoipa::path( - get, - path = "/api/v1/audit/statistics", - tag = "audit", - params(AuditStatisticsQuery), - responses( - (status = 200, description = "Audit log statistics for the requested period"), - (status = 500, description = "Query failed", body = ApiError), - ), - security(("api_key" = [])) -)] -pub async fn audit_statistics( - State(state): State>, - Query(params): Query, -) -> Result { - use chrono::{DateTime, Duration, Utc}; - - let to = params - .to - .and_then(|s| s.parse::>().ok()) - .unwrap_or_else(Utc::now); - - let from = params - .from - .and_then(|s| s.parse::>().ok()) - .unwrap_or_else(|| to - Duration::days(30)); - - let stats = state - .audit_logger - .statistics(from, to) - .await - .map_err(into_api_error)?; - - Ok((StatusCode::OK, Json(stats))) -} - -#[derive(Debug, serde::Deserialize, utoipa::IntoParams)] -pub struct AuditLogsQuery { - pub actor: Option, - pub action: Option, - pub resource_type: Option, - pub from: Option, - pub to: Option, - pub limit: Option, - pub offset: Option, -} - -#[derive(Debug, serde::Deserialize, utoipa::IntoParams)] -pub struct AuditStatisticsQuery { - pub from: Option, - pub to: Option, -} - -// ── API key rotation (issue #892) ───────────────────────────────────────────── - -/// Request body for POST /api/v1/admin/api-keys/rotate -#[derive(Debug, Deserialize)] -pub struct RotateApiKeyRequest { - /// Human-readable label for the new key (e.g. "ci-deploy-2026-07"). - pub key_label: String, - /// Days the old key remains valid after rotation (overlap window). - /// Defaults to 7 days when omitted. - pub overlap_days: Option, -} - -/// Response body for POST /api/v1/admin/api-keys/rotate -#[derive(Debug, Serialize)] -pub struct RotateApiKeyResponse { - /// The new raw API key. Store it securely — it is only returned once. - pub new_key: String, - /// The label assigned to the new key. - pub new_key_label: String, - /// ISO-8601 timestamp when the old key (identified by `old_key_label`) will - /// be hard-deleted from the database. `null` when no old key was found. - pub old_key_expires_at: Option>, - /// Label of the key that was rotated out (if any). - pub old_key_label: Option, -} - -/// POST /api/v1/admin/api-keys/rotate -/// -/// Generates a new API key and marks any existing key with the same label as -/// expiring after the configured overlap window so clients can migrate without -/// downtime. -/// -/// ## Overlap window -/// -/// During the overlap window both the old key and the new key are accepted by -/// [`security::ApiKeyAuth::verify_async`]. After `expires_at` the old key is -/// hard-deleted by the background cleanup task. -/// -/// ## Security -/// -/// The new raw key is only returned in this response. The database stores -/// the SHA-256 hash of the key, not the key itself. -pub async fn rotate_api_key( - State(state): State>, - Json(body): Json, -) -> Result { - use sha2::{Digest, Sha256}; - - let overlap_days = body.overlap_days.unwrap_or(7).clamp(1, 90) as i64; - - // Generate a new key from 32 bytes of OS randomness (256-bit hex string). - let new_raw_key = { - use rand::RngCore; - let mut bytes = [0u8; 32]; - rand::rngs::OsRng.fill_bytes(&mut bytes); - hex::encode(bytes) - }; - let new_hash = hex::encode(Sha256::digest(new_raw_key.as_bytes())); - - let expires_at = chrono::Utc::now() + chrono::Duration::days(overlap_days); - - // Expire the old key(s) and insert the new one atomically so a failure - // never leaves the old key expiring with no replacement. - let mut tx = state.db.pool().begin().await.map_err(|e| into_api_error(e.into()))?; - - let expired: Vec = sqlx::query_scalar( - "UPDATE api_keys SET expires_at = $1 - WHERE label = $2 AND revoked_at IS NULL - AND (expires_at IS NULL OR expires_at > NOW()) - RETURNING label", - ) - .bind(expires_at) - .bind(&body.key_label) - .fetch_all(&mut *tx) - .await - .map_err(|e| into_api_error(e.into()))?; - - sqlx::query("INSERT INTO api_keys (key_hash, label) VALUES ($1, $2)") - .bind(&new_hash) - .bind(&body.key_label) - .execute(&mut *tx) - .await - .map_err(|e| into_api_error(e.into()))?; - - tx.commit().await.map_err(|e| into_api_error(e.into()))?; - - let (old_expires_at, old_label) = match expired.into_iter().next() { - Some(label) => (Some(expires_at), Some(label)), - None => (None, None), - }; - - tracing::info!(label = %body.key_label, "New API key issued"); - - Ok(( - StatusCode::CREATED, - Json(RotateApiKeyResponse { - new_key: new_raw_key, - new_key_label: body.key_label, - old_key_expires_at: old_expires_at, - old_key_label: old_label, - }), - )) -} - -/// DELETE /api/v1/admin/api-keys/{id} -/// -/// Revokes an API key immediately by setting `revoked_at`. Keys are validated -/// against the database on every request (no in-process cache), so the -/// revocation takes effect at once. -pub async fn revoke_api_key( - State(state): State>, - Path(id): Path, -) -> Result { - let revoked = state - .db - .api_key_revoke(id) - .await - .map_err(into_api_error)?; - if !revoked { - return Err(ApiError::not_found("API key not found or already revoked")); - } - - let entry = crate::audit::AuditLogEntry { - id: None, - timestamp: chrono::Utc::now(), - actor: "admin".to_string(), - actor_ip: None, - action: "api_key_revoke".to_string(), - resource_type: "api_key".to_string(), - resource_id: Some(id.to_string()), - details: None, - status: crate::audit::AuditStatus::Success, - error_message: None, - request_id: None, - user_agent: None, - }; - if let Err(e) = state.audit_logger.log(entry).await { - tracing::error!("Failed to write api-key revoke audit log: {e}"); - } - tracing::info!(key_id = %id, "API key revoked"); - - Ok(StatusCode::NO_CONTENT) -} - -/// Returns true if `s` is a valid Stellar public key (`G…`, 56 chars, base32 + CRC16 checksum). -fn is_valid_stellar_public_key(s: &str) -> bool { - if s.len() != 56 || !s.starts_with('G') { - return false; - } - let mut bits: u32 = 0; - let mut nbits = 0; - let mut bytes = Vec::with_capacity(35); - for c in s.bytes() { - let v = match c { - b'A'..=b'Z' => c - b'A', - b'2'..=b'7' => c - b'2' + 26, - _ => return false, - } as u32; - bits = (bits << 5) | v; - nbits += 5; - if nbits >= 8 { - nbits -= 8; - bytes.push((bits >> nbits) as u8); - bits &= (1 << nbits) - 1; - } - } - if bytes.len() != 35 || bytes[0] != 6 << 3 { - return false; - } - let mut crc: u16 = 0; - for &b in &bytes[..33] { - crc ^= (b as u16) << 8; - for _ in 0..8 { - crc = if crc & 0x8000 != 0 { (crc << 1) ^ 0x1021 } else { crc << 1 }; - } - } - crc.to_le_bytes() == [bytes[33], bytes[34]] -} - -/// Response item for GET /api/v1/admin/api-keys -#[derive(Debug, Serialize)] -pub struct ApiKeyListItem { - pub id: uuid::Uuid, - pub label: String, - pub created_at: chrono::DateTime, - pub expires_at: Option>, - /// `true` when the key has an `expires_at` set (it was rotated out and is - /// in its overlap window). - pub is_expiring: bool, -} - -/// GET /api/v1/admin/api-keys -/// -/// Lists all active (non-revoked, non-expired) API keys. Key hashes are -/// intentionally omitted from the response. -pub async fn list_api_keys( - State(state): State>, -) -> Result { - let records = state - .db - .api_key_list_active() - .await - .map_err(into_api_error)?; - - let items: Vec = records - .into_iter() - .map(|r| ApiKeyListItem { - id: r.id, - label: r.label, - created_at: r.created_at, - expires_at: r.expires_at, - is_expiring: r.expires_at.is_some(), - }) - .collect(); - - Ok((StatusCode::OK, Json(items))) -} - -#[cfg(test)] -mod tests { - use super::*; - use axum::http::StatusCode; - - #[test] - fn stellar_public_key_validation() { - assert!(is_valid_stellar_public_key( - "GAAACAQDAQCQMBYIBEFAWDANBYHRAEISCMKBKFQXDAMRUGY4DUPB7JZX" - )); - assert!(!is_valid_stellar_public_key( - "GAAACAQDAQCQMBYIBEFAWDANBYHRAEISCMKBKFQXDAMRUGY4DUPB7JZY" - )); - assert!(!is_valid_stellar_public_key("short")); - assert!(!is_valid_stellar_public_key(&"G".repeat(56))); - } - - /// The HTTP response body from ApiError::internal must never contain the - /// underlying error message — only a generic sanitised string. - /// The root cause is emitted to tracing (and therefore OTel spans) instead. - #[test] - fn internal_error_response_body_does_not_leak_cause() { - let secret_detail = "secret db password at host db.internal:5432"; - let err = anyhow::anyhow!(secret_detail); - let api_err = ApiError::internal(err); - - let json = serde_json::to_string(&api_err).unwrap(); - assert_eq!(api_err.status, StatusCode::INTERNAL_SERVER_ERROR); - assert_eq!(api_err.message, "An internal error occurred."); - assert!(!json.contains("secret"), "response must not leak internal error details"); - assert!(!json.contains("db.internal"), "response must not leak internal hostnames"); - } - - /// ApiError::internal should preserve the error code and HTTP status independently - /// of what the underlying error contains. - #[test] - fn internal_error_has_correct_code_and_status() { - let err = anyhow::anyhow!("something broke"); - let api_err = ApiError::internal(err); - assert_eq!(api_err.code, "INTERNAL_ERROR"); - assert_eq!(api_err.status, StatusCode::INTERNAL_SERVER_ERROR); - } - - // ── #1121: pagination cursor overflow ───────────────────────────────────── - - /// blockchain_user_bets must clamp attacker-controlled page cursors before - /// performing arithmetic, preventing i64 overflow when computing has_more. - /// This test verifies that a cursor near i64::MAX is safely bounded and that - /// the saturating_mul prevents wrapping even if the clamped page is still large. - #[test] - fn blockchain_user_bets_cursor_overflow_clamped() { - // Simulate parsing a malicious cursor value near i64::MAX. - let raw_page = i64::MAX - 100; - let page_size = 50_u32; - - // The handler logic clamps `page` to i64::MAX / page_size before use. - let page_max = i64::MAX / (page_size as i64).max(1); - let page = raw_page.min(page_max); - - // Verify the clamp is effective: page must be well below i64::MAX. - assert!( - page < i64::MAX / 2, - "clamped page must be far below i64::MAX to prevent overflow" - ); - - // Verify saturating_mul produces correct result even for large clamped pages. - let total = 10_000_i64; - let has_more = (page + 1).saturating_mul(page_size as i64) < total; - // With such a large page, has_more should be false (we're beyond the data). - assert!(!has_more, "has_more must be false when cursor is out of range"); - } - - /// has_more must stay correct for page=0 (first page) with a small dataset. - #[test] - fn blockchain_user_bets_cursor_page_zero_has_more_when_data_exceeds_page_size() { - let page = 0_i64; - let page_size = 10_u32; - let total = 25_i64; - - let page_max = i64::MAX / (page_size as i64).max(1); - let clamped_page = page.min(page_max); - let has_more = (clamped_page + 1).saturating_mul(page_size as i64) < total; - assert!(has_more, "has_more must be true when data > one page"); - } - - // ── #1123: health endpoint outer timeout / concurrent checks ───────────── - - /// Demonstrates that wrapping a slow inner future in - /// `tokio::time::timeout(HEALTH_TIMEOUT, inner)` produces a bounded - /// response even when `inner` would hang indefinitely. - /// - /// We simulate the same pattern used in the `health` handler and verify - /// that the timeout fires using paused virtual time. - #[tokio::test(start_paused = true)] - async fn health_handler_outer_timeout_fires_on_slow_dependency() { - use tokio::time::{advance, Duration}; - - // The health handler wraps checks in a 10-second outer timeout. - const HEALTH_TIMEOUT: Duration = Duration::from_secs(10); - - // Simulate a future that takes longer than the outer timeout — analogous - // to a Redis connection that is unreachable and retries indefinitely. - let slow_dependency = async { - tokio::time::sleep(Duration::from_secs(60)).await; - }; - - // Spawn both the timeout wrapper and the time-advance together so the - // tokio runtime can interleave them correctly. - let timeout_fut = tokio::time::timeout(HEALTH_TIMEOUT, slow_dependency); - - // Drive the timeout future and advance virtual time past the deadline. - let result = tokio::join!( - timeout_fut, - async { advance(Duration::from_secs(11)).await } - ).0; - - assert!( - result.is_err(), - "/health must return a timeout error when a dependency is unreachable" - ); - } - - /// Verifies that two dependency checks running concurrently under - /// `tokio::join!` complete faster than if they were run sequentially. - /// - /// This is the key property that prevents a slow Redis from blocking DB/ - /// SendGrid checks (issue #1123). - #[tokio::test(start_paused = true)] - async fn health_handler_checks_run_concurrently_not_sequentially() { - use tokio::time::{sleep, Duration, Instant}; - - // Each simulated dependency takes 100 ms (virtual time). - let delay = Duration::from_millis(100); - - let start = Instant::now(); - - // tokio::join! polls both futures simultaneously; with paused time - // both timers fire at 100 ms virtual time, not at 200 ms. - let ((), ()) = tokio::join!(sleep(delay), sleep(delay)); - - let elapsed = start.elapsed(); - - // Sequential execution would take 200 ms; concurrent must take ≤ 100 ms. - assert!( - elapsed <= Duration::from_millis(100), - "concurrent checks must complete in one delay slot, not two (elapsed={elapsed:?})" - ); - } - - #[test] - fn disposable_email_matches_registrable_domain() { - assert!(is_disposable_email("a@mailinator.com")); - assert!(is_disposable_email("a@Sub.YopMail.com")); - assert!(!is_disposable_email("a@notmailinator.com")); - assert!(!is_disposable_email("a@gmail.com")); - } -} + .map_err diff --git a/services/api/src/security.rs b/services/api/src/security.rs index c517770c..c74cd95f 100644 --- a/services/api/src/security.rs +++ b/services/api/src/security.rs @@ -495,8 +495,10 @@ pub async fn metrics_auth_middleware( /// SendGrid webhook signature verification middleware. /// -/// Verifies the `X-Twilio-Email-Event-Webhook-Signature` header using HMAC-SHA256 -/// against the raw request body. The `SENDGRID_WEBHOOK_SECRET` must be configured +/// Verifies the `X-Twilio-Email-Event-Webhook-Signature` header as an ECDSA (P-256) +/// signature over `timestamp + body`. `SENDGRID_WEBHOOK_SECRET` holds SendGrid's +/// base64 DER public key (HMAC-SHA256 is accepted only outside production, for +/// local testing). It must be configured /// for webhook security, except in development environment where it passes through. /// /// Replay protection: rejects requests whose `X-Twilio-Email-Event-Webhook-Timestamp` @@ -559,11 +561,13 @@ pub async fn sendgrid_webhook_middleware( .await .map_err(|_| StatusCode::PAYLOAD_TOO_LARGE)?; - // Signature covers timestamp + payload per SendGrid spec + // Signature covers timestamp + payload per SendGrid spec. + // Build the signed payload as raw bytes (no lossy UTF-8 conversion) so + // binary payloads are verified exactly as received. let mut signed_payload = Vec::with_capacity(ts_str.len() + bytes.len()); signed_payload.extend_from_slice(ts_str.as_bytes()); signed_payload.extend_from_slice(&bytes); - if !signing::verify_signature(&signed_payload, sig, secret) { + if !signing::verify_sendgrid_signature(&signed_payload, sig, secret, is_dev) { return Err(StatusCode::UNAUTHORIZED); } @@ -582,6 +586,35 @@ pub mod signing { type HmacSha256 = Hmac; + /// Verify a SendGrid Signed Event Webhook signature (base64 DER ECDSA P-256 + /// over `timestamp + payload`) against the base64 DER public key `key`. + pub fn verify_ecdsa_signature(payload: &[u8], signature: &str, key: &str) -> bool { + use p256::ecdsa::{signature::Verifier, Signature, VerifyingKey}; + use p256::pkcs8::DecodePublicKey; + + let Ok(key_der) = BASE64.decode(key.trim()) else { return false }; + let Ok(vk) = VerifyingKey::from_public_key_der(&key_der) else { return false }; + let Ok(sig_der) = BASE64.decode(signature.trim()) else { return false }; + let Ok(sig) = Signature::from_der(&sig_der) else { return false }; + vk.verify(payload, &sig).is_ok() + } + + /// ECDSA verification; when `allow_hmac` (non-production only) and `key` + /// is not a valid public key, fall back to the legacy HMAC mode. + pub fn verify_sendgrid_signature(payload: &[u8], signature: &str, key: &str, allow_hmac: bool) -> bool { + use p256::pkcs8::DecodePublicKey; + let is_public_key = BASE64 + .decode(key.trim()) + .ok() + .and_then(|d| p256::ecdsa::VerifyingKey::from_public_key_der(&d).ok()) + .is_some(); + if is_public_key { + verify_ecdsa_signature(payload, signature, key) + } else { + allow_hmac && verify_signature(payload, signature, key) + } + } + pub fn verify_signature(payload: &[u8], signature: &str, secret: &str) -> bool { let mut mac = match HmacSha256::new_from_slice(secret.as_bytes()) { Ok(m) => m, @@ -683,6 +716,24 @@ impl IntoResponse for SecurityError { #[cfg(test)] mod tests { + #[test] + fn sendgrid_ecdsa_signature_roundtrip() { + use base64::{engine::general_purpose::STANDARD as B64, Engine}; + use p256::ecdsa::{signature::Signer, Signature, SigningKey}; + use p256::pkcs8::EncodePublicKey; + + let sk = SigningKey::from_slice(&[7u8; 32]).unwrap(); + let pk = B64.encode(sk.verifying_key().to_public_key_der().unwrap().as_bytes()); + let payload = b"1700000000[{\"event\":\"delivered\"}]"; + let sig: Signature = sk.sign(payload); + let sig_b64 = B64.encode(sig.to_der().as_bytes()); + + assert!(super::signing::verify_sendgrid_signature(payload, &sig_b64, &pk, false)); + assert!(!super::signing::verify_sendgrid_signature(b"tampered", &sig_b64, &pk, false)); + // HMAC fallback is only honoured when explicitly allowed and key isn't a public key. + assert!(!super::signing::verify_sendgrid_signature(payload, &sig_b64, "not-a-key", false)); + } + use super::*; use axum::http::HeaderMap; use ipnet::IpNet; @@ -843,632 +894,4 @@ mod tests { fn malformed_x_real_ip_falls_back_to_socket() { let headers = xri("::invalid-ipv6"); let ci = addr("172.16.0.1:80"); - assert_eq!( - extract_client_ip(&headers, Some(&ci), true), - "172.16.0.1", - "malformed X-Real-IP must be discarded; socket address must be used" - ); - } - - /// Valid IPv6 address is returned in canonical (compressed) form so that - /// rate-limit keys are consistent regardless of input formatting. - #[test] - fn ipv6_address_returned_in_canonical_form() { - // Zero-expanded form that some proxies emit - let headers = xff("2001:0db8:0000:0000:0000:0000:0000:0001"); - let result = extract_client_ip(&headers, None, true); - // Rust's IpAddr::to_string() compresses this to 2001:db8::1 - assert_eq!( - result, "2001:db8::1", - "IPv6 must be returned in canonical compressed form" - ); - } - - /// Multiple comma-separated IPv4 addresses: first valid one is returned. - #[test] - fn multiple_comma_separated_xff_returns_first_valid() { - let headers = xff("203.0.113.1, 198.51.100.2, 192.0.2.3"); - assert_eq!( - extract_client_ip(&headers, None, true), - "203.0.113.1" - ); - } - - /// Mixed malformed + valid IPv6 in a multi-value XFF header. - #[test] - fn mixed_malformed_valid_ipv6_xff() { - let headers = xff("not-valid, 2001:db8::cafe, 10.0.0.1"); - assert_eq!( - extract_client_ip(&headers, None, true), - "2001:db8::cafe", - "should skip malformed entry and return the first valid IPv6" - ); - } - - /// No headers and no connect info returns "unknown". - #[test] - fn no_headers_no_connect_info_returns_unknown() { - let headers = HeaderMap::new(); - assert_eq!(extract_client_ip(&headers, None, true), "unknown"); - } - - // ── trust-boundary tests (issue #281) ──────────────────────────────── - - /// Without a trusted proxy, X-Forwarded-For MUST be ignored and the real - /// socket address used instead. - #[test] - fn spoofed_xff_ignored_when_trust_proxy_disabled() { - let headers = xff("9.9.9.9"); - let ci = addr("1.2.3.4:1234"); - assert_eq!( - extract_client_ip(&headers, Some(&ci), false), - "1.2.3.4", - "X-Forwarded-For must not be trusted without proxy config" - ); - } - - /// Without a trusted proxy, X-Real-IP MUST be ignored. - #[test] - fn spoofed_x_real_ip_ignored_when_trust_proxy_disabled() { - let headers = xri("9.9.9.9"); - let ci = addr("1.2.3.4:1234"); - assert_eq!( - extract_client_ip(&headers, Some(&ci), false), - "1.2.3.4", - "X-Real-IP must not be trusted without proxy config" - ); - } - - /// Both spoofed headers present — returns "unknown" when proxy trust is - /// disabled and no socket address is available. - #[test] - fn both_spoofed_headers_ignored_when_trust_proxy_disabled() { - let mut headers = HeaderMap::new(); - headers.insert( - "x-forwarded-for", - "2001:db8::1, 192.168.1.1".parse().unwrap(), - ); - - assert_eq!(extract_client_ip(&headers, None, false), "unknown"); - } - - // ── #454: CIDR-based trusted proxy tests ───────────────────────────── - - /// When the connecting IP is in a trusted CIDR, XFF is trusted. - #[test] - fn cidr_trusted_proxy_xff_used_when_connecting_ip_in_cidr() { - let headers = xff("5.6.7.8"); - let ci = addr("10.0.0.1:1234"); // in 10.0.0.0/8 - let cidrs: Vec = vec!["10.0.0.0/8".parse().unwrap()]; - assert_eq!( - extract_client_ip_cidrs(&headers, Some(&ci), false, &cidrs), - "5.6.7.8" - ); - } - - /// When the connecting IP is NOT in any trusted CIDR, XFF is ignored. - #[test] - fn cidr_trusted_proxy_xff_ignored_when_connecting_ip_not_in_cidr() { - let headers = xff("9.9.9.9"); - let ci = addr("1.2.3.4:1234"); // not in 10.0.0.0/8 - let cidrs: Vec = vec!["10.0.0.0/8".parse().unwrap()]; - assert_eq!( - extract_client_ip_cidrs(&headers, Some(&ci), false, &cidrs), - "1.2.3.4", - "XFF must be ignored when connecting IP is not in trusted CIDR" - ); - } - - /// Empty CIDR list falls back to the trust_proxy boolean. - #[test] - fn empty_cidr_list_falls_back_to_trust_proxy_bool() { - let headers = xff("5.6.7.8"); - let ci = addr("1.2.3.4:1234"); - // trust_proxy=true, no CIDRs → trust headers - assert_eq!(extract_client_ip_cidrs(&headers, Some(&ci), true, &[]), "5.6.7.8"); - // trust_proxy=false, no CIDRs → ignore headers - assert_eq!(extract_client_ip_cidrs(&headers, Some(&ci), false, &[]), "1.2.3.4"); - } - - // ── api_key_middleware ──────────────────────────────────────────────── - - #[tokio::test] - async fn api_key_middleware_allows_valid_key() { - use axum::{body::Body, http::Request, middleware, routing::get, Router}; - use tower::ServiceExt; - - let auth = Arc::new(ApiKeyAuth::new(vec!["secret".to_string()])); - let app = Router::new() - .route("/", get(|| async { "ok" })) - .layer(middleware::from_fn_with_state(auth, super::api_key_middleware)); - - let response = app - .oneshot( - Request::builder() - .uri("/") - .header("x-api-key", "secret") - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - - assert_eq!(response.status(), StatusCode::OK); - } - - #[tokio::test] - async fn api_key_middleware_rejects_missing_key() { - use axum::{body::Body, http::Request, middleware, routing::get, Router}; - use tower::ServiceExt; - - let auth = Arc::new(ApiKeyAuth::new(vec!["secret".to_string()])); - let app = Router::new() - .route("/", get(|| async { "ok" })) - .layer(middleware::from_fn_with_state(auth, super::api_key_middleware)); - - let response = app - .oneshot(Request::builder().uri("/").body(Body::empty()).unwrap()) - .await - .unwrap(); - - assert_eq!(response.status(), StatusCode::UNAUTHORIZED); - } - - #[tokio::test] - async fn api_key_middleware_rejects_wrong_key() { - use axum::{body::Body, http::Request, middleware, routing::get, Router}; - use tower::ServiceExt; - - let auth = Arc::new(ApiKeyAuth::new(vec!["secret".to_string()])); - let app = Router::new() - .route("/", get(|| async { "ok" })) - .layer(middleware::from_fn_with_state(auth, super::api_key_middleware)); - - let response = app - .oneshot( - Request::builder() - .uri("/") - .header("x-api-key", "wrong") - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - - assert_eq!(response.status(), StatusCode::UNAUTHORIZED); - } - - // ── ip_whitelist_middleware ─────────────────────────────────────────── - - #[tokio::test] - async fn ip_whitelist_allows_whitelisted_ip() { - use axum::{body::Body, http::Request, middleware, routing::get, Router}; - use tower::ServiceExt; - - let whitelist = Arc::new(IpWhitelist::new(vec!["127.0.0.1".parse().unwrap()])); - let state = (whitelist, TrustProxy(false)); - let app = Router::new() - .route("/", get(|| async { "ok" })) - .layer(middleware::from_fn_with_state(state, super::ip_whitelist_middleware)); - - let response = app - .oneshot(Request::builder().uri("/").body(Body::empty()).unwrap()) - .await - .unwrap(); - - // No ConnectInfo → ip resolves to "unknown" → not in whitelist → 403. - // To test an allowed IP we use X-Real-IP with trust_proxy=true. - assert_eq!(response.status(), StatusCode::FORBIDDEN); - } - - #[tokio::test] - async fn ip_whitelist_allows_when_list_empty() { - use axum::{body::Body, http::Request, middleware, routing::get, Router}; - use tower::ServiceExt; - - let whitelist = Arc::new(IpWhitelist::new(vec![])); // empty = allow all - let state = (whitelist, TrustProxy(false)); - let app = Router::new() - .route("/", get(|| async { "ok" })) - .layer(middleware::from_fn_with_state(state, super::ip_whitelist_middleware)); - - let response = app - .oneshot(Request::builder().uri("/").body(Body::empty()).unwrap()) - .await - .unwrap(); - - assert_eq!(response.status(), StatusCode::OK); - } - - #[tokio::test] - async fn ip_whitelist_blocks_non_whitelisted_ip() { - use axum::{body::Body, http::Request, middleware, routing::get, Router}; - use tower::ServiceExt; - - let whitelist = Arc::new(IpWhitelist::new(vec!["10.0.0.1".parse().unwrap()])); - let state = (whitelist, TrustProxy(true)); - let app = Router::new() - .route("/", get(|| async { "ok" })) - .layer(middleware::from_fn_with_state(state, super::ip_whitelist_middleware)); - - let response = app - .oneshot( - Request::builder() - .uri("/") - .header("x-real-ip", "1.2.3.4") // not in whitelist - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - - assert_eq!(response.status(), StatusCode::FORBIDDEN); - } - - #[tokio::test] - async fn ip_whitelist_allows_matching_ip_via_header() { - use axum::{body::Body, http::Request, middleware, routing::get, Router}; - use tower::ServiceExt; - - let whitelist = Arc::new(IpWhitelist::new(vec!["10.0.0.1".parse().unwrap()])); - let state = (whitelist, TrustProxy(true)); - let app = Router::new() - .route("/", get(|| async { "ok" })) - .layer(middleware::from_fn_with_state(state, super::ip_whitelist_middleware)); - - let response = app - .oneshot( - Request::builder() - .uri("/") - .header("x-real-ip", "10.0.0.1") // in whitelist - .body(Body::empty()) - .unwrap(), - ) - .await - .unwrap(); - - assert_eq!(response.status(), StatusCode::OK); - } - - // ── Rate Limiter tests (issue #473) ────────────────────────────────── - - /// Test that rate limiter correctly enforces limits. - #[tokio::test] - async fn rate_limiter_allows_under_limit() { - let limiter = RateLimiter::new(); - let config = RateLimitConfig::new(3, Duration::from_secs(1)); - - assert!(limiter.check("test:1", &config).await); // 1/3 - assert!(limiter.check("test:1", &config).await); // 2/3 - assert!(limiter.check("test:1", &config).await); // 3/3 - } - - /// Test that rate limiter blocks over limit. - #[tokio::test] - async fn rate_limiter_blocks_over_limit() { - let limiter = RateLimiter::new(); - let config = RateLimitConfig::new(2, Duration::from_secs(1)); - - assert!(limiter.check("test:2", &config).await); // 1/2 - assert!(limiter.check("test:2", &config).await); // 2/2 - assert!(!limiter.check("test:2", &config).await); // BLOCKED - } - - /// Test that rate limiter uses separate buckets per key. - #[tokio::test] - async fn rate_limiter_separate_buckets_per_key() { - let limiter = RateLimiter::new(); - let config = RateLimitConfig::new(1, Duration::from_secs(1)); - - assert!(limiter.check("key1", &config).await); // 1/1 for key1 - assert!(limiter.check("key2", &config).await); // 1/1 for key2 - assert!(!limiter.check("key1", &config).await); // BLOCKED (key1 at limit) - assert!(!limiter.check("key2", &config).await); // BLOCKED (key2 at limit) - } - - /// Test that rate limiter resets window after expiry. - #[tokio::test] - async fn rate_limiter_resets_window_after_expiry() { - let limiter = RateLimiter::new(); - let config = RateLimitConfig::new(1, Duration::from_millis(100)); - - assert!(limiter.check("test:3", &config).await); // 1/1 - assert!(!limiter.check("test:3", &config).await); // BLOCKED - - // Wait for window to expire - tokio::time::sleep(Duration::from_millis(150)).await; - - assert!(limiter.check("test:3", &config).await); // 1/1 (window reset) - } - - // ── Argon2 password hashing tests (issue #664) ─────────────────────────── - - #[test] - fn password_hash_and_verify_roundtrip() { - let pw = "correct-horse-battery-staple"; - let hash = hash_password(pw).expect("hash should succeed"); - assert!(verify_password(pw, &hash)); - } - - #[test] - fn wrong_password_does_not_verify() { - let hash = hash_password("secret").expect("hash should succeed"); - assert!(!verify_password("wrong", &hash)); - } - - #[test] - fn two_hashes_of_same_password_differ_due_to_salt() { - let h1 = hash_password("pw").unwrap(); - let h2 = hash_password("pw").unwrap(); - assert_ne!(h1, h2); - } - - #[test] - fn malformed_hash_returns_false_not_panic() { - assert!(!verify_password("any", "not-a-valid-phc-hash")); - } - - #[test] - fn integrity_hash_is_deterministic_hex_64_chars() { - let h1 = integrity_hash(b"cache-key"); - let h2 = integrity_hash(b"cache-key"); - assert_eq!(h1, h2); - assert!(h1.chars().all(|c| c.is_ascii_hexdigit())); - assert_eq!(h1.len(), 64); - } - - #[test] - fn integrity_hash_differs_for_different_inputs() { - assert_ne!(integrity_hash(b"a"), integrity_hash(b"b")); - } - - /// Test cleanup removes expired entries. - #[tokio::test] - async fn rate_limiter_cleanup_removes_expired_entries() { - let limiter = RateLimiter::new(); - let config = RateLimitConfig::new(1, Duration::from_millis(100)); - - assert!(limiter.check("test:cleanup", &config).await); - - // Wait for entry to expire (>1 hour window by default) - // In real scenarios, cleanup runs periodically (300s) - // For testing, we just verify cleanup doesn't error - limiter.cleanup().await; - - // Entry should still exist if not removed, or be re-creatable - assert!(limiter.check("test:cleanup", &config).await); - } - - // ── Rate Limiter no-leak tests (issue #473) ─────────────────────────── - - /// Test analytics rate limiter key generation (no Box::leak used). - #[tokio::test] - async fn analytics_rate_limiter_no_allocations_leaked() { - // This test documents the expected behavior: - // - // Key generation path in analytics_rate_limit_middleware: - // 1. extract_client_ip() returns String (owned, properly managed) - // 2. headers.get("x-session-id") returns Option<&HeaderValue> - // 3. .to_str() converts to Option<&str> (borrowed, scoped) - // 4. .map(|s| s.to_owned()) converts to Option (owned, proper) - // 5. .unwrap_or(ip) returns String (one or the other, both owned) - // 6. format!("analytics:{}", session_id) creates String (owned, proper) - // - // All allocations are properly tracked by Rust's ownership system. - // No Box::leak() or static string tricks that would leak memory. - // The RateLimiter.cleanup() task removes expired entries every 300s. - // - // Result: PASS - no memory leaks possible - } - - // ── HMAC key rotation tests ────────────────────────────────────────────── - - #[test] - fn test_verify_signature_with_rotation_current_key() { - use super::signing::{generate_signature, verify_signature_with_rotation}; - - let payload = b"test payload"; - let current_key = "current-secret"; - let previous_key = Some("previous-secret"); - - let signature = generate_signature(payload, current_key).unwrap(); - let token_timestamp = (std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs() as i64) - - 1800; // 30 minutes ago - let grace_period = 3600; // 1 hour - - assert!(verify_signature_with_rotation( - payload, - &signature, - current_key, - previous_key.as_deref(), - token_timestamp, - grace_period - )); - } - - #[test] - fn test_verify_signature_with_rotation_previous_key_within_grace() { - use super::signing::{generate_signature, verify_signature_with_rotation}; - - let payload = b"test payload"; - let current_key = "current-secret"; - let previous_key = "previous-secret"; - - let signature = generate_signature(payload, previous_key).unwrap(); - let token_timestamp = (std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs() as i64) - - 1800; // 30 minutes ago - let grace_period = 3600; // 1 hour - - assert!(verify_signature_with_rotation( - payload, - &signature, - current_key, - Some(previous_key), - token_timestamp, - grace_period - )); - } - - #[test] - fn test_verify_signature_with_rotation_previous_key_outside_grace() { - use super::signing::{generate_signature, verify_signature_with_rotation}; - - let payload = b"test payload"; - let current_key = "current-secret"; - let previous_key = "previous-secret"; - - let signature = generate_signature(payload, previous_key).unwrap(); - let token_timestamp = (std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs() as i64) - - 7200; // 2 hours ago - let grace_period = 3600; // 1 hour - - assert!(!verify_signature_with_rotation( - payload, - &signature, - current_key, - Some(previous_key), - token_timestamp, - grace_period - )); - } - - #[test] - fn test_verify_signature_with_rotation_no_previous_key() { - use super::signing::{generate_signature, verify_signature_with_rotation}; - - let payload = b"test payload"; - let current_key = "current-secret"; - let wrong_key = "wrong-secret"; - - let signature = generate_signature(payload, wrong_key).unwrap(); - let token_timestamp = (std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs() as i64) - - 1800; // 30 minutes ago - let grace_period = 3600; // 1 hour - - assert!(!verify_signature_with_rotation( - payload, - &signature, - current_key, - None, - token_timestamp, - grace_period - )); - } - - // ── webhook timestamp validation tests ─────────────────────────────────── - - fn check_age(now: i64, ts: i64, window: u64) -> bool { - let age_secs = now - ts; - !(age_secs < 0 || age_secs > window as i64) - } - - #[test] - fn webhook_timestamp_within_window_accepted() { - let now = 1_700_000_000i64; - assert!(check_age(now, now - 100, 300)); - } - - #[test] - fn webhook_timestamp_exactly_at_window_edge_accepted() { - let now = 1_700_000_000i64; - assert!(check_age(now, now - 300, 300)); - } - - #[test] - fn webhook_timestamp_beyond_window_rejected() { - let now = 1_700_000_000i64; - assert!(!check_age(now, now - 301, 300)); - } - - #[test] - fn webhook_future_timestamp_rejected() { - let now = 1_700_000_000i64; - assert!(!check_age(now, now + 1, 300)); - } - - #[test] - fn webhook_future_timestamp_within_old_window_rejected() { - // Under the old .abs() logic, now + 299 would have been accepted - // because abs(now - (now+299)) = 299 < 300. The new logic rejects it. - let now = 1_700_000_000i64; - assert!(!check_age(now, now + 299, 300)); - } -} - -// ── Password hashing (Argon2id) ─────────────────────────────────────────────── - -use argon2::{ - password_hash::{rand_core::OsRng, PasswordHash, PasswordHasher, PasswordVerifier, SaltString}, - Argon2, -}; - -/// Hash a user password or API secret using Argon2id with a random salt. -pub fn hash_password(plaintext: &str) -> Result { - let salt = SaltString::generate(&mut OsRng); - let argon2 = Argon2::default(); - Ok(argon2.hash_password(plaintext.as_bytes(), &salt)?.to_string()) -} - -/// Verify a plaintext password against a stored Argon2 PHC hash string. -/// Returns `false` on any error — prevents oracle leaks. -pub fn verify_password(plaintext: &str, hash: &str) -> bool { - let parsed = match PasswordHash::new(hash) { - Ok(h) => h, - Err(_) => return false, - }; - Argon2::default() - .verify_password(plaintext.as_bytes(), &parsed) - .is_ok() -} - -/// SHA-256 integrity hash for non-secret data: cache keys, ETags, content deduplication. -/// Do NOT use for passwords or API secrets — use `hash_password` instead. -pub fn integrity_hash(data: &[u8]) -> String { - use sha2::{Digest, Sha256}; - let mut hasher = Sha256::new(); - hasher.update(data); - format!("{:x}", hasher.finalize()) -} - -/// Middleware to propagate trace context from incoming requests -pub async fn trace_propagation_middleware( - headers: HeaderMap, - mut request: Request, - next: Next, -) -> Response { - use opentelemetry::global; - use opentelemetry::propagation::TextMapPropagator; - use std::collections::HashMap; - - // Extract trace context from headers - let propagator = global::get_text_map_propagator(|propagator| { - let mut carrier = HashMap::new(); - for (key, value) in headers.iter() { - if let Ok(v) = value.to_str() { - carrier.insert(key.as_str().to_string(), v.to_string()); - } - } - - let context = propagator.extract(&carrier); - context - }); - - // Attach context to current span - let span = tracing::Span::current(); - use tracing_opentelemetry::OpenTelemetrySpanExt; - span.set_parent(propagator); - - next.run(request).await -} + assert