//! Tests for [`op_server::ApiKeyAuthLayer`] and [`op_server::RateLimitLayer`]. //! //! Same `tower::ServiceExt::oneshot` strategy as `api.rs` — no socket, //! no listener, just direct router invocation through the middleware //! stack. The rate-limit tests inject a mock clock so refill behavior //! is observed without sleeping. use std::collections::HashSet; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use axum::body::Body; use axum::http::{Request, StatusCode}; use http_body_util::BodyExt; use op_server::{ApiKeyAuthLayer, AppState, RateLimitLayer, router_with_middleware}; use serde_json::Value; use tower::util::ServiceExt; async fn body_to_value(body: Body) -> Value { let bytes = body.collect().await.expect("collect body").to_bytes(); if bytes.is_empty() { Value::Null } else { serde_json::from_slice(&bytes).expect("body is json") } } fn req(method: &str, path: &str, headers: &[(&str, &str)]) -> Request { let mut b = Request::builder().method(method).uri(path); for (k, v) in headers { b = b.header(*k, *v); } b.body(Body::empty()).unwrap() } fn allow_keys(keys: &[&str]) -> ApiKeyAuthLayer { let set: HashSet = keys.iter().map(|s| (*s).to_owned()).collect(); ApiKeyAuthLayer::new(set).with_bypass_paths(vec!["/health".into(), "/readiness".into()]) } /// Convenience: a permissive rate limiter (high capacity) — used by /// auth tests so the rate limiter never trips by accident. fn permissive_limiter() -> RateLimitLayer { RateLimitLayer::new(10_000, 10_000.0) } // ==================================================================== // API-key auth // ==================================================================== #[tokio::test] async fn auth_missing_header_returns_401() { let app = router_with_middleware( AppState::new_in_memory(), allow_keys(&["sk-good"]), permissive_limiter(), ); let res = app .oneshot(req( "GET", "/v1/audit/report?start_tx=0&end_tx=1&generated_at_unix_secs=0", &[], )) .await .unwrap(); assert_eq!(res.status(), StatusCode::UNAUTHORIZED); let body = body_to_value(res.into_body()).await; assert_eq!(body["code"], "unauthorized"); assert!(body["details"].is_null()); } #[tokio::test] async fn auth_wrong_key_returns_401() { let app = router_with_middleware( AppState::new_in_memory(), allow_keys(&["sk-good"]), permissive_limiter(), ); let res = app .oneshot(req( "GET", "/v1/audit/report?start_tx=0&end_tx=1&generated_at_unix_secs=0", &[("authorization", "Bearer sk-bad")], )) .await .unwrap(); assert_eq!(res.status(), StatusCode::UNAUTHORIZED); } #[tokio::test] async fn auth_valid_key_passes_through() { let app = router_with_middleware( AppState::new_in_memory(), allow_keys(&["sk-good"]), permissive_limiter(), ); let res = app .oneshot(req( "GET", "/v1/audit/report?start_tx=0&end_tx=1&generated_at_unix_secs=0", &[("authorization", "Bearer sk-good")], )) .await .unwrap(); assert_eq!(res.status(), StatusCode::OK); } #[tokio::test] async fn auth_bypass_health_endpoint() { let app = router_with_middleware( AppState::new_in_memory(), allow_keys(&["sk-good"]), permissive_limiter(), ); let res = app .clone() .oneshot(req("GET", "/health", &[])) .await .unwrap(); assert_eq!(res.status(), StatusCode::OK); let res = app.oneshot(req("GET", "/readiness", &[])).await.unwrap(); assert_eq!(res.status(), StatusCode::OK); } #[tokio::test] async fn auth_custom_header_no_bearer_stripping() { let auth = ApiKeyAuthLayer::new({ let mut s = HashSet::new(); s.insert("raw-token".to_owned()); s }) .with_header("x-api-key") .with_bypass_paths(vec!["/health".into()]); let app = router_with_middleware(AppState::new_in_memory(), auth, permissive_limiter()); // Missing custom header → 401. let res = app .clone() .oneshot(req( "GET", "/v1/audit/report?start_tx=0&end_tx=1&generated_at_unix_secs=0", &[], )) .await .unwrap(); assert_eq!(res.status(), StatusCode::UNAUTHORIZED); // Present custom header, raw (no Bearer prefix) → 200. let res = app .oneshot(req( "GET", "/v1/audit/report?start_tx=0&end_tx=1&generated_at_unix_secs=0", &[("x-api-key", "raw-token")], )) .await .unwrap(); assert_eq!(res.status(), StatusCode::OK); } // ==================================================================== // Rate limiting // ==================================================================== #[tokio::test] async fn rate_limit_blocks_after_capacity() { // capacity=3, refill very slow so the bucket stays empty during // the burst. let limiter = RateLimitLayer::new(3, 0.01).with_clock(|| 1_000); let app = router_with_middleware(AppState::new_in_memory(), allow_keys(&["sk"]), limiter); let headers = [("authorization", "Bearer sk")]; let mut statuses = Vec::new(); let mut retry_after_headers = Vec::new(); for _ in 0..5 { let res = app .clone() .oneshot(req("GET", "/health", &headers)) .await .unwrap(); statuses.push(res.status()); retry_after_headers.push( res.headers() .get("retry-after") .map(|v| v.to_str().unwrap().to_owned()), ); } // First 3 succeed, next 2 are rate limited. assert_eq!(statuses[0], StatusCode::OK); assert_eq!(statuses[1], StatusCode::OK); assert_eq!(statuses[2], StatusCode::OK); assert_eq!(statuses[3], StatusCode::TOO_MANY_REQUESTS); assert_eq!(statuses[4], StatusCode::TOO_MANY_REQUESTS); // The 429 responses must carry a Retry-After header. assert!(retry_after_headers[3].is_some()); assert!(retry_after_headers[4].is_some()); let retry_secs: u64 = retry_after_headers[3] .as_ref() .unwrap() .parse() .expect("Retry-After is a number"); assert!(retry_secs >= 1); } #[tokio::test] async fn rate_limit_refills_over_time() { // capacity=2, refill 1 token/sec. Mock clock starts at 1000 and we // advance it manually. let clock = Arc::new(AtomicU64::new(1_000)); let clock_for_layer = Arc::clone(&clock); let limiter = RateLimitLayer::new(2, 1.0).with_clock(move || clock_for_layer.load(Ordering::SeqCst)); let app = router_with_middleware(AppState::new_in_memory(), allow_keys(&["sk"]), limiter); let headers = [("authorization", "Bearer sk")]; // Drain bucket: 2 OK, then 429. for _ in 0..2 { let res = app .clone() .oneshot(req("GET", "/health", &headers)) .await .unwrap(); assert_eq!(res.status(), StatusCode::OK); } let res = app .clone() .oneshot(req("GET", "/health", &headers)) .await .unwrap(); assert_eq!(res.status(), StatusCode::TOO_MANY_REQUESTS); // Advance clock by 2s → bucket refilled to 2 tokens. clock.fetch_add(2, Ordering::SeqCst); let res = app .clone() .oneshot(req("GET", "/health", &headers)) .await .unwrap(); assert_eq!(res.status(), StatusCode::OK); let res = app .clone() .oneshot(req("GET", "/health", &headers)) .await .unwrap(); assert_eq!(res.status(), StatusCode::OK); let res = app.oneshot(req("GET", "/health", &headers)).await.unwrap(); assert_eq!(res.status(), StatusCode::TOO_MANY_REQUESTS); } #[tokio::test] async fn rate_limit_buckets_are_per_source_key() { // capacity=1, slow refill so buckets only see one request. let limiter = RateLimitLayer::new(1, 0.01).with_clock(|| 1_000); let app = router_with_middleware( AppState::new_in_memory(), allow_keys(&["sk-a", "sk-b"]), limiter, ); // First key burns its bucket. let res = app .clone() .oneshot(req("GET", "/health", &[("authorization", "Bearer sk-a")])) .await .unwrap(); assert_eq!(res.status(), StatusCode::OK); let res = app .clone() .oneshot(req("GET", "/health", &[("authorization", "Bearer sk-a")])) .await .unwrap(); assert_eq!(res.status(), StatusCode::TOO_MANY_REQUESTS); // Second key has its own untouched bucket. let res = app .oneshot(req("GET", "/health", &[("authorization", "Bearer sk-b")])) .await .unwrap(); assert_eq!(res.status(), StatusCode::OK); }