From 9a0c823c60be74261a0a78282e11a086beb5a076 Mon Sep 17 00:00:00 2001 From: "x.qntx.eth" Date: Sun, 9 Aug 2026 14:54:53 +0800 Subject: [PATCH 1/3] feat: production-hardening for hop admission, reload, and defaults Enforce target policy and circuit accounting on every redirect hop with CircuitHop RAII so cancelled probes cannot stick half-open. Retain circuit/rate state across SIGHUP when those sections are unchanged. Move inflight to limits, enable GCRA by default, cap response body and keyed map cardinality, and stamp CORS outside timeout/body layers. --- .gitignore | 1 + CHANGELOG.md | 25 ++- Cargo.lock | 1 + charts/corx/values.yaml | 6 +- corx.example.toml | 66 +++--- crates/corx-cli/src/main.rs | 10 +- crates/corx-core/src/config/circuit.rs | 11 +- crates/corx-core/src/config/limits.rs | 22 +- crates/corx-core/src/config/rate_limit.rs | 43 ++-- crates/corx-core/src/config/ssrf.rs | 2 +- crates/corx-core/src/config/target.rs | 2 +- crates/corx-core/src/config/upstream.rs | 2 +- crates/corx-core/src/config/validate.rs | 16 +- crates/corx-core/src/error.rs | 31 ++- crates/corx-core/src/policy/circuit.rs | 202 ++++++++++++++++-- crates/corx-core/src/policy/mod.rs | 2 +- crates/corx-core/src/policy/target.rs | 52 ++++- crates/corx-core/src/proxy/headers.rs | 33 ++- crates/corx-core/src/proxy/upstream.rs | 100 ++++++--- crates/corx-server/Cargo.toml | 1 + crates/corx-server/src/handlers/proxy.rs | 116 +++++----- crates/corx-server/src/hot_reload.rs | 51 ++++- crates/corx-server/src/lib.rs | 8 +- .../corx-server/src/middleware/load_shed.rs | 36 ++-- .../corx-server/src/middleware/rate_limit.rs | 141 ++++++++++-- .../src/middleware/request_guard.rs | 7 + .../corx-server/src/observability/metering.rs | 73 +++++++ crates/corx-server/src/observability/mod.rs | 2 +- crates/corx-server/src/router.rs | 28 +-- crates/corx-server/src/state.rs | 107 ++++++---- crates/corx-server/tests/integration_proxy.rs | 163 ++++++++++++++ crates/corx/src/lib.rs | 2 +- docs/architecture.md | 31 ++- docs/configuration.md | 19 +- docs/migration.md | 33 ++- docs/observability.md | 4 +- docs/operations.md | 11 +- docs/security.md | 7 +- 38 files changed, 1153 insertions(+), 314 deletions(-) diff --git a/.gitignore b/.gitignore index 7ffcc09..63e2551 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,4 @@ target 3rdparty/ audit/ plan/ +crates/.DS_Store diff --git a/CHANGELOG.md b/CHANGELOG.md index 75ce7db..51a24f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,13 +19,32 @@ binary CLI surface. - `[security.auth]` — optional `bearer` shared-secret authentication. - `security.require_client_binding` — hard-fail startup without origin whitelist, bearer auth, or mTLS. -- `[target]` — host/scheme admission (`any_public` | `allowlist` | `denylist`). -- `[circuit_breaker]` — process-local per-host circuit breaker. -- `limits.redirect_policy` — `follow` | `block` | `rewrite`. +- `[target]` — host/scheme admission (`any_public` | `allowlist` | `denylist`), + enforced on **every redirect hop**. +- `[circuit_breaker]` — process-local per-host circuit breaker (`max_hosts` + soft cap, default 8192). +- `limits.redirect_policy` — `follow` | `block` | `rewrite` (rewrite stamps + proxy path-prefix `Location`). +- `limits.inflight_max` — process concurrency load-shed (moved from + `rate_limit.global`). +- `limits.max_response_body_bytes` — streaming response size cap (default + 50 MiB; `0` = unlimited). +- `rate_limit.max_keys` — fail-closed cardinality cap for keyed GCRA maps. - `cors.origins` + `cors.allow_any_origin` (unified origin list). +- CatchPanic layer; load-shed inflight RAII permit; hot-reload retains + circuit / rate-limit maps when those config sections are unchanged. ### Changed (BREAKING) +- `rate_limit.global.inflight_max` **removed** — use `limits.inflight_max`. +- `rate_limit.enabled` default is now **`true`** (GCRA on by default). +- Client error JSON `message` is kind-stable (no internal DNS/connect detail). +- `CircuitDecision` public enum removed; `CircuitBreaker::check` returns + `Result<(), ProxyError>`. +- CORS middleware sits outside auth/load-shed/header-limit so 401/503/431 + include CORS headers; success path no longer double-stamps CORS in the + handler. + - Cleartext listeners honour `server.graceful_shutdown` with a force-abort deadline after SIGTERM/Ctrl+C (previously only the TLS path used the duration). diff --git a/Cargo.lock b/Cargo.lock index 5766200..704ef28 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -467,6 +467,7 @@ dependencies = [ "axum-server", "bytes", "corx-core", + "dashmap", "figment", "foldhash 0.1.5", "futures", diff --git a/charts/corx/values.yaml b/charts/corx/values.yaml index e01a5fe..451d3f9 100644 --- a/charts/corx/values.yaml +++ b/charts/corx/values.yaml @@ -83,6 +83,8 @@ config: | [limits] max_request_body_bytes = 10485760 max_request_header_bytes = 32768 + max_response_body_bytes = 52428800 + inflight_max = 1024 request_timeout = "60s" connect_timeout = "5s" max_redirects = 5 @@ -138,6 +140,7 @@ config: | open_duration = "30s" half_open_max = 1 count_5xx = true + max_hosts = 8192 [ssrf] mode = { kind = "strict" } @@ -152,6 +155,7 @@ config: | [rate_limit] enabled = true + max_keys = 16384 [rate_limit.origin] rps = 50 @@ -170,12 +174,10 @@ config: | [rate_limit.global] rps = 5000 burst = 10000 - inflight_max = 1024 [upstream] pool_max_idle_per_host = 32 pool_idle_timeout = "90s" - user_agent = "corx/1.0.0" [observability] log_format = "json" diff --git a/corx.example.toml b/corx.example.toml index e31dfbd..58a365c 100644 --- a/corx.example.toml +++ b/corx.example.toml @@ -1,7 +1,6 @@ -# corx configuration reference. All fields are required unless marked optional. -# The effective configuration layers this file, the CORX_* environment variables -# (double underscores separate nested keys, e.g. CORX_SERVER__BIND=...) and CLI -# flags, in increasing order of precedence. +# corx configuration reference. +# Layer order (increasing precedence): defaults → this file / CORX_CONFIG → CORX_* env → CLI. +# Nested env keys use `__`, e.g. CORX_SERVER__BIND=0.0.0.0:9000 [server] bind = "0.0.0.0:8080" @@ -15,16 +14,19 @@ http2 = true [limits] max_request_body_bytes = 10_485_760 # 10 MiB max_request_header_bytes = 32_768 # 32 KiB +max_response_body_bytes = 52_428_800 # 50 MiB; 0 = unlimited +inflight_max = 1000 # process concurrency cap; 0 = off request_timeout = "60s" connect_timeout = "10s" max_redirects = 5 -allow_https_to_http_downgrade = false # reject https → http redirects by default +allow_https_to_http_downgrade = false redirect_policy = "follow" # follow | block | rewrite [cors] policy = "reflect" # wildcard | reflect | explicit -origins = [] # gate for reflect / list for explicit -allow_any_origin = true # demo: echo any Origin; production: false + origins +origins = [] # production: list trusted origins +# Production: keep false and populate `origins`. Demo only: +allow_any_origin = false allowed_methods = ["GET", "HEAD", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"] allowed_headers = [ "accept", @@ -38,12 +40,12 @@ allowed_headers = [ exposed_headers = ["x-corx-status", "x-corx-target-url", "x-request-id"] max_age = "600s" allow_credentials = false -allow_private_network = false # PNA handshake (Chromium et al.) +allow_private_network = false [forwarded] -inject = true # inject X-Forwarded-* + RFC 7239 Forwarded -trust_inbound_xff = false # do not trust client-supplied XFF chains -inject_request_id = true # generate UUID v7 when missing +inject = true +trust_inbound_xff = false +inject_request_id = true [security] require_header = ["origin"] @@ -52,16 +54,15 @@ remove_request_headers = ["cookie", "cookie2"] remove_response_headers = ["set-cookie", "set-cookie2"] origin_blacklist = [] origin_whitelist = [] +# require_client_binding = true # refuse start without whitelist / bearer / mTLS [security.preflight] -mode = "enforce" # enforce | open (open = cors-anywhere: preflight before guards) -rate_limit = true # charge multi-dimensional rate limiter on OPTIONS +mode = "enforce" # enforce | open +rate_limit = true [security.auth] mode = "none" # none | bearer -bearer_tokens = [] # required when mode = bearer - -# require_client_binding = false # true → must set origin_whitelist, bearer, or mTLS +bearer_tokens = [] [target] mode = "any_public" # any_public | allowlist | denylist @@ -76,52 +77,51 @@ window = "30s" open_duration = "30s" half_open_max = 1 count_5xx = true +max_hosts = 8192 [ssrf] -# mode = { kind = "strict" } # production default (fail-closed) -# mode = { kind = "permissive", allow_private = false } # only RFC 1918 + loopback admitted mode = { kind = "strict" } allow_ipv6 = true -extra_blocked_cidrs = [] # add to the built-in block list -extra_allowed_cidrs = [] # punch holes for trusted internal CIDRs +extra_blocked_cidrs = [] +extra_allowed_cidrs = [] [rate_limit] -enabled = false # master switch +enabled = true +max_keys = 16384 [rate_limit.origin] rps = 50 burst = 100 -unlimited_patterns = [] # regex on the Origin header +unlimited_patterns = [] [rate_limit.ip] rps = 30 burst = 60 -trusted_cidrs = [] # CIDRs exempt from per-IP limits +trusted_cidrs = [] [rate_limit.target_host] rps = 100 burst = 200 [rate_limit.global] -rps = 5000 # global RPS cap -burst = 10000 -inflight_max = 1000 # 0 disables the load-shed layer +rps = 5000 +burst = 10000 [upstream] pool_max_idle_per_host = 32 pool_idle_timeout = "90s" -user_agent = "corx/1.0.0" +# Defaults to corx/ when omitted from a minimal config. [observability] -log_format = "json" # json | pretty +log_format = "json" log_level = "info" metrics_endpoint = "/metrics" [observability.otel] -enabled = false # requires the `otel` cargo feature +enabled = false endpoint = "http://otel-collector:4317" -protocol = "grpc" # grpc | http +protocol = "grpc" service_name = "corx" -service_namespace = "" # optional logical grouping -resource_attributes = [] # ["deployment.environment=prod", "team=platform"] -sample_ratio = 0.1 # 1.0 keeps every span; 0.1 keeps 10% +service_namespace = "" +resource_attributes = [] +sample_ratio = 0.1 diff --git a/crates/corx-cli/src/main.rs b/crates/corx-cli/src/main.rs index c86ff71..0146986 100644 --- a/crates/corx-cli/src/main.rs +++ b/crates/corx-cli/src/main.rs @@ -60,11 +60,11 @@ async fn serve(config_path: Option<&Path>) -> anyhow::Result<()> { let build = ServerBuild::from_config(config.clone(), metrics)?; let ready = std::sync::Arc::clone(&build.ready); - // Hot-reload watcher (Unix only): SIGHUP triggers a fresh load of the - // same config path. Hot-swappable fields are atomically replaced via - // `arc-swap`; immutable fields (bind address, TLS material) only log a - // warning and are otherwise ignored. The watcher exits with the - // server. + // Hot-reload watcher (Unix only): SIGHUP reloads the same config path. + // Hot-swappable policy is swapped via `arc-swap`. Attempts to change + // immutable fields (bind, TLS, body/header limits, timeouts, metrics + // path, inflight/response caps) are rejected and the previous snapshot + // stays active. #[cfg(unix)] { let owned_path = config_path.map(Path::to_path_buf); diff --git a/crates/corx-core/src/config/circuit.rs b/crates/corx-core/src/config/circuit.rs index b54cc4f..034c6eb 100644 --- a/crates/corx-core/src/config/circuit.rs +++ b/crates/corx-core/src/config/circuit.rs @@ -10,7 +10,7 @@ use super::default_true; /// /// State is not shared across replicas; pair with external rate limiting when /// multi-instance consistency is required. -#[derive(Debug, Clone, Copy, Deserialize, Serialize)] +#[derive(Debug, Clone, Copy, Deserialize, Serialize, Eq, PartialEq)] #[serde(deny_unknown_fields)] pub struct CircuitBreakerConfig { /// Master switch. Default: enabled with conservative thresholds. @@ -31,6 +31,10 @@ pub struct CircuitBreakerConfig { /// Count upstream HTTP 5xx as failures (in addition to connect/timeout). #[serde(default = "default_true")] pub count_5xx: bool, + /// Soft cap on tracked host keys. Idle closed entries are evicted when + /// the map exceeds this size (cardinality / abuse defence). + #[serde(default = "default_max_hosts")] + pub max_hosts: usize, } const fn default_failure_threshold() -> u32 { @@ -49,6 +53,10 @@ const fn default_open_duration() -> Duration { Duration::from_secs(30) } +const fn default_max_hosts() -> usize { + 8192 +} + impl Default for CircuitBreakerConfig { fn default() -> Self { Self { @@ -58,6 +66,7 @@ impl Default for CircuitBreakerConfig { open_duration: default_open_duration(), half_open_max: default_half_open_max(), count_5xx: true, + max_hosts: default_max_hosts(), } } } diff --git a/crates/corx-core/src/config/limits.rs b/crates/corx-core/src/config/limits.rs index 121480c..aeb7c6f 100644 --- a/crates/corx-core/src/config/limits.rs +++ b/crates/corx-core/src/config/limits.rs @@ -11,7 +11,7 @@ const MIB: u64 = 1024 * 1024; #[serde(rename_all = "lowercase")] pub enum RedirectPolicy { /// Follow redirects in-proxy (up to [`LimitsConfig::max_redirects`]), - /// re-validating SSRF on every hop. **Default.** + /// re-validating SSRF and target policy on every hop. **Default.** #[default] Follow, /// Do not follow; surface a proxy error instead of leaking Location. @@ -22,13 +22,21 @@ pub enum RedirectPolicy { } /// Size- and time-based limits applied to every request. -#[derive(Debug, Clone, Copy, Deserialize, Serialize)] +#[derive(Debug, Clone, Copy, Deserialize, Serialize, Eq, PartialEq)] #[serde(deny_unknown_fields)] pub struct LimitsConfig { /// Maximum inbound request body size, in bytes. pub max_request_body_bytes: u64, /// Maximum inbound header size, in bytes. pub max_request_header_bytes: u32, + /// Maximum upstream response body size, in bytes. `0` disables the cap. + /// When exceeded the stream is aborted mid-transfer. + #[serde(default = "default_max_response_body_bytes")] + pub max_response_body_bytes: u64, + /// Maximum concurrent in-flight requests process-wide. `0` disables + /// load-shed. Independent of GCRA rate limiting. + #[serde(default = "default_inflight_max")] + pub inflight_max: u32, /// Total allowable duration of a single proxied request, end-to-end. #[serde(with = "humantime_serde")] pub request_timeout: Duration, @@ -45,11 +53,21 @@ pub struct LimitsConfig { pub redirect_policy: RedirectPolicy, } +const fn default_max_response_body_bytes() -> u64 { + 50 * MIB +} + +const fn default_inflight_max() -> u32 { + 1_000 +} + impl Default for LimitsConfig { fn default() -> Self { Self { max_request_body_bytes: 10 * MIB, max_request_header_bytes: 32 * 1024, + max_response_body_bytes: default_max_response_body_bytes(), + inflight_max: default_inflight_max(), request_timeout: Duration::from_mins(1), connect_timeout: Duration::from_secs(10), max_redirects: 5, diff --git a/crates/corx-core/src/config/rate_limit.rs b/crates/corx-core/src/config/rate_limit.rs index 88ce8c1..2d2bccd 100644 --- a/crates/corx-core/src/config/rate_limit.rs +++ b/crates/corx-core/src/config/rate_limit.rs @@ -3,17 +3,27 @@ use ipnet::IpNet; use serde::{Deserialize, Serialize}; -/// Rate-limit configuration covering four orthogonal dimensions. +use super::default_true; + +/// Rate-limit configuration covering four orthogonal GCRA dimensions. /// /// Each sub-limiter is independent: setting any of `origin.rps`, `ip.rps`, /// `target_host.rps` or `global.rps` to `0` disables that dimension while /// leaving the others active. Setting [`RateLimitConfig::enabled`] to -/// `false` disables every dimension at once. -#[derive(Debug, Clone, Deserialize, Serialize)] +/// `false` disables every GCRA dimension at once. +/// +/// Process-wide **inflight** concurrency is configured under +/// [`crate::config::LimitsConfig::inflight_max`], not here. +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct RateLimitConfig { - /// Master switch. When `false`, every dimension below is bypassed. + /// Master switch for GCRA dimensions. Default: enabled. + #[serde(default = "default_true")] pub enabled: bool, + /// Soft cap on distinct keys per keyed dimension (origin / ip / host). + /// New keys are rejected with 429 when the map is full. + #[serde(default = "default_max_keys")] + pub max_keys: usize, /// Per-`Origin`-header limiting. #[serde(default)] pub origin: OriginLimitConfig, @@ -24,15 +34,20 @@ pub struct RateLimitConfig { /// caller targeting a popular destination). #[serde(default)] pub target_host: HostLimitConfig, - /// Process-wide concurrency limiter that drives the load-shed layer. + /// Process-wide GCRA token bucket. #[serde(default)] pub global: GlobalLimitConfig, } +const fn default_max_keys() -> usize { + 16_384 +} + impl Default for RateLimitConfig { fn default() -> Self { Self { - enabled: false, + enabled: true, + max_keys: default_max_keys(), origin: OriginLimitConfig { rps: 50, burst: 100, @@ -50,14 +65,13 @@ impl Default for RateLimitConfig { global: GlobalLimitConfig { rps: 5_000, burst: 10_000, - inflight_max: 1_000, }, } } } /// Per-`Origin` rate-limit configuration. -#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct OriginLimitConfig { /// Steady-state requests-per-second; `0` disables this dimension. @@ -73,7 +87,7 @@ pub struct OriginLimitConfig { } /// Per-client-IP rate-limit configuration. -#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct IpLimitConfig { /// Steady-state requests-per-second; `0` disables this dimension. @@ -89,7 +103,7 @@ pub struct IpLimitConfig { } /// Per-target-host rate-limit configuration. -#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize)] +#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, Eq, PartialEq)] #[serde(deny_unknown_fields)] pub struct HostLimitConfig { /// Steady-state requests-per-second; `0` disables this dimension. @@ -100,8 +114,8 @@ pub struct HostLimitConfig { pub burst: u32, } -/// Process-wide global limits that drive the load-shed layer. -#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize)] +/// Process-wide GCRA token bucket (not inflight concurrency). +#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, Eq, PartialEq)] #[serde(deny_unknown_fields)] pub struct GlobalLimitConfig { /// Steady-state requests-per-second across the entire proxy. `0` @@ -111,9 +125,4 @@ pub struct GlobalLimitConfig { /// Token-bucket burst budget on top of `rps`. #[serde(default)] pub burst: u32, - /// Maximum number of in-flight requests. Exceeding this triggers the - /// load-shed layer which immediately answers `503 Service Unavailable` - /// with a `Retry-After` header. `0` disables the load-shed layer. - #[serde(default)] - pub inflight_max: u32, } diff --git a/crates/corx-core/src/config/ssrf.rs b/crates/corx-core/src/config/ssrf.rs index 490562f..3872dd3 100644 --- a/crates/corx-core/src/config/ssrf.rs +++ b/crates/corx-core/src/config/ssrf.rs @@ -42,7 +42,7 @@ impl SsrfMode { } /// SSRF protection. -#[derive(Debug, Clone, Deserialize, Serialize)] +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct SsrfConfig { /// Operating mode. **Strict** is the production default. diff --git a/crates/corx-core/src/config/target.rs b/crates/corx-core/src/config/target.rs index b17b3b6..f73705a 100644 --- a/crates/corx-core/src/config/target.rs +++ b/crates/corx-core/src/config/target.rs @@ -17,7 +17,7 @@ pub enum TargetMode { /// Target URL admission (host + scheme) applied after path extraction and /// before rate limiting / upstream connect. -#[derive(Debug, Clone, Deserialize, Serialize)] +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct TargetConfig { /// Host admission mode. diff --git a/crates/corx-core/src/config/upstream.rs b/crates/corx-core/src/config/upstream.rs index cdf13bd..424a735 100644 --- a/crates/corx-core/src/config/upstream.rs +++ b/crates/corx-core/src/config/upstream.rs @@ -5,7 +5,7 @@ use std::time::Duration; use serde::{Deserialize, Serialize}; /// Upstream HTTP client tuning. -#[derive(Debug, Clone, Deserialize, Serialize)] +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] pub struct UpstreamConfig { /// Max idle connections retained per host in the connection pool. diff --git a/crates/corx-core/src/config/validate.rs b/crates/corx-core/src/config/validate.rs index f0153a7..e195c7e 100644 --- a/crates/corx-core/src/config/validate.rs +++ b/crates/corx-core/src/config/validate.rs @@ -210,6 +210,7 @@ fn validate_limits(cfg: &Config, report: &mut ValidationReport) { "must be > 0", )); } + // 0 means "unlimited" for response body — intentional opt-out. } fn validate_rate_limit(cfg: &RateLimitConfig, report: &mut ValidationReport) { @@ -217,17 +218,23 @@ fn validate_rate_limit(cfg: &RateLimitConfig, report: &mut ValidationReport) { return; } + if cfg.max_keys == 0 { + report.errors.push(ConfigError::new( + "rate_limit.max_keys", + "must be > 0 when rate limiting is enabled", + )); + } + let any_enabled = cfg.origin.rps > 0 || cfg.ip.rps > 0 || cfg.target_host.rps > 0 - || cfg.global.rps > 0 - || cfg.global.inflight_max > 0; + || cfg.global.rps > 0; if !any_enabled { report.errors.push(ConfigError::new( "rate_limit", - "rate_limit.enabled = true but every dimension is at 0; either \ + "rate_limit.enabled = true but every GCRA dimension is at 0; either \ disable rate-limiting or set at least one of origin.rps / ip.rps / \ - target_host.rps / global.rps / global.inflight_max", + target_host.rps / global.rps (inflight is limits.inflight_max)", )); } @@ -364,7 +371,6 @@ mod tests { cfg.rate_limit.ip.rps = 0; cfg.rate_limit.target_host.rps = 0; cfg.rate_limit.global.rps = 0; - cfg.rate_limit.global.inflight_max = 0; let err = cfg.validate().unwrap_err(); assert_eq!(err.path, "rate_limit"); } diff --git a/crates/corx-core/src/error.rs b/crates/corx-core/src/error.rs index e88382c..f8fdd0f 100644 --- a/crates/corx-core/src/error.rs +++ b/crates/corx-core/src/error.rs @@ -238,6 +238,10 @@ impl ProxyError { /// Renders the error into the wire-level pieces required to construct a /// HTTP response: status code and the JSON-serialisable payload. /// + /// `message` is a **client-safe** short phrase derived from + /// [`ErrorKind`]. Full diagnostic detail stays on the server log via + /// [`Display`](std::fmt::Display) / `tracing`. + /// /// Adapters typically: /// /// 1. Set the response status to `payload.0`. @@ -251,8 +255,33 @@ impl ProxyError { kind.status(), ErrorPayload { error: kind.as_str(), - message: self.to_string(), + message: kind.client_message().to_owned(), }, ) } } + +impl ErrorKind { + /// Stable, non-sensitive phrase safe to return to untrusted clients. + #[must_use] + pub const fn client_message(self) -> &'static str { + match self { + Self::InvalidUrl => "invalid target url", + Self::MissingRequiredHeader => "missing required header", + Self::OriginNotAllowed => "origin not allowed", + Self::SsrfBlocked => "target address blocked by ssrf policy", + Self::DnsFailure => "dns lookup failed", + Self::UpstreamUnreachable => "upstream unreachable", + Self::UpstreamTimeout => "upstream timed out", + Self::TooManyRedirects => "too many redirects", + Self::TlsFailure => "tls handshake failed", + Self::PayloadTooLarge => "payload too large", + Self::RateLimited => "rate limited", + Self::TargetNotAllowed => "target not allowed", + Self::Unauthorized => "unauthorized", + Self::CircuitOpen => "circuit open for target host", + Self::RedirectBlocked => "redirect blocked by policy", + Self::Io | Self::Internal => "internal error", + } + } +} diff --git a/crates/corx-core/src/policy/circuit.rs b/crates/corx-core/src/policy/circuit.rs index 08521a8..4f8dd8e 100644 --- a/crates/corx-core/src/policy/circuit.rs +++ b/crates/corx-core/src/policy/circuit.rs @@ -10,20 +10,12 @@ use crate::config::CircuitBreakerConfig; use crate::error::ProxyError; use crate::observability; -/// Outcome of a circuit check before dispatching upstream. -#[derive(Debug, Clone, Copy, Eq, PartialEq)] -pub enum CircuitDecision { - /// Proceed with the request. - Closed, - /// Proceed as a limited half-open probe. - HalfOpenProbe, -} - #[derive(Debug, Clone, Copy)] enum State { Closed, Open { until: Instant }, - HalfOpen { probes: u32 }, + /// `since` bounds half-open so cancelled probes cannot lock a host forever. + HalfOpen { probes: u32, since: Instant }, } struct HostCircuit { @@ -39,6 +31,18 @@ impl HostCircuit { failures: Vec::new(), } } + + const fn is_idle_closed(&self) -> bool { + matches!(self.state, State::Closed) && self.failures.is_empty() + } + + const fn is_closed(&self) -> bool { + matches!(self.state, State::Closed) + } + + fn is_expired_open(&self, now: Instant) -> bool { + matches!(self.state, State::Open { until } if now >= until) + } } /// Per-host circuit breaker shared across requests. @@ -54,6 +58,7 @@ struct Inner { open_duration: Duration, half_open_max: u32, count_5xx: bool, + max_hosts: usize, hosts: DashMap, } @@ -62,6 +67,7 @@ impl std::fmt::Debug for CircuitBreaker { f.debug_struct("CircuitBreaker") .field("enabled", &self.inner.enabled) .field("hosts", &self.inner.hosts.len()) + .field("max_hosts", &self.inner.max_hosts) .finish_non_exhaustive() } } @@ -78,6 +84,7 @@ impl CircuitBreaker { open_duration: cfg.open_duration, half_open_max: cfg.half_open_max.max(1), count_5xx: cfg.count_5xx, + max_hosts: cfg.max_hosts.max(1), hosts: DashMap::with_hasher(RandomState::default()), }), } @@ -91,26 +98,35 @@ impl CircuitBreaker { /// Check whether a request to `host` may proceed. /// + /// Half-open probe budgeting is enforced inside this method. Callers must + /// always pair a successful `check` with [`Self::record_success`] or + /// [`Self::record_failure`] (or a drop-guard that records failure) so a + /// cancelled probe cannot leave the host stuck in half-open forever. + /// /// # Errors /// /// Returns [`ProxyError::CircuitOpen`] when the breaker is open. - pub fn check(&self, host: &str) -> Result { + pub fn check(&self, host: &str) -> Result<(), ProxyError> { if !self.inner.enabled { - return Ok(CircuitDecision::Closed); + return Ok(()); } let now = Instant::now(); let half_open_max = self.inner.half_open_max; + let open_duration = self.inner.open_duration; let mut entry = self .inner .hosts .entry(host.to_owned()) .or_insert_with(HostCircuit::new); - let decision = transition_on_check(&mut entry, now, half_open_max); + let allowed = transition_on_check(&mut entry, now, half_open_max, open_duration); drop(entry); - decision.ok_or_else(|| { + if allowed { + self.evict_if_needed(now); + Ok(()) + } else { metrics::counter!(observability::CIRCUIT_REJECTS).increment(1); - ProxyError::CircuitOpen(host.to_owned()) - }) + Err(ProxyError::CircuitOpen(host.to_owned())) + } } /// Record a successful upstream response (or client-error that is not a @@ -145,6 +161,43 @@ impl CircuitBreaker { metrics::counter!(observability::CIRCUIT_OPENS).increment(1); tracing::warn!(host, "circuit breaker opened"); } + self.evict_if_needed(now); + } + + /// Number of tracked hosts (test / ops helper). + #[must_use] + pub fn tracked_hosts(&self) -> usize { + self.inner.hosts.len() + } + + fn evict_if_needed(&self, now: Instant) { + let max = self.inner.max_hosts; + if self.inner.hosts.len() <= max { + return; + } + // Prefer reclaiming entries that no longer need protection state. + self.evict_matching(max, HostCircuit::is_idle_closed); + if self.inner.hosts.len() > max { + self.evict_matching(max, |c| c.is_expired_open(now)); + } + if self.inner.hosts.len() > max { + self.evict_matching(max, HostCircuit::is_closed); + } + } + + fn evict_matching(&self, max: usize, pred: impl Fn(&HostCircuit) -> bool) { + let excess = self.inner.hosts.len().saturating_sub(max); + if excess == 0 { + return; + } + let mut removed = 0usize; + self.inner.hosts.retain(|_, circuit| { + if removed >= excess || !pred(circuit) { + return true; + } + removed = removed.saturating_add(1); + false + }); } } @@ -152,20 +205,35 @@ fn transition_on_check( entry: &mut HostCircuit, now: Instant, half_open_max: u32, -) -> Option { + open_duration: Duration, +) -> bool { match entry.state { - State::Closed => Some(CircuitDecision::Closed), + State::Closed => true, State::Open { until } if now >= until => { - entry.state = State::HalfOpen { probes: 1 }; - Some(CircuitDecision::HalfOpenProbe) + entry.state = State::HalfOpen { + probes: 1, + since: now, + }; + true } - State::HalfOpen { probes } if probes < half_open_max => { + State::HalfOpen { probes, since } if probes < half_open_max => { entry.state = State::HalfOpen { probes: probes.saturating_add(1), + since, + }; + true + } + // Probe budget exhausted: if the half-open window elapsed (cancelled + // probes never settled), open a new probe window instead of locking + // the host permanently. + State::HalfOpen { since, .. } if now.duration_since(since) >= open_duration => { + entry.state = State::HalfOpen { + probes: 1, + since: now, }; - Some(CircuitDecision::HalfOpenProbe) + true } - State::Open { .. } | State::HalfOpen { .. } => None, + State::Open { .. } | State::HalfOpen { .. } => false, } } @@ -191,6 +259,58 @@ fn record_failure_on( true } +/// RAII guard: records a failure if the hop is abandoned (cancel / panic) +/// without an explicit success or failure settlement. +#[derive(Debug)] +pub struct CircuitHop<'a> { + circuit: &'a CircuitBreaker, + host: String, + settled: bool, +} + +impl<'a> CircuitHop<'a> { + /// Admit `host` and return a guard that must be settled. + /// + /// # Errors + /// + /// Propagates [`CircuitBreaker::check`]. + pub fn admit(circuit: &'a CircuitBreaker, host: impl Into) -> Result { + let host = host.into(); + circuit.check(&host)?; + Ok(Self { + circuit, + host, + settled: false, + }) + } + + /// Host this hop was admitted for. + #[must_use] + pub fn host(&self) -> &str { + &self.host + } + + /// Mark the hop successful (disarms drop-failure). + pub fn success(mut self) { + self.circuit.record_success(&self.host); + self.settled = true; + } + + /// Mark the hop failed (disarms drop-failure). + pub fn failure(mut self) { + self.circuit.record_failure(&self.host); + self.settled = true; + } +} + +impl Drop for CircuitHop<'_> { + fn drop(&mut self) { + if !self.settled { + self.circuit.record_failure(&self.host); + } + } +} + #[cfg(test)] mod tests { use std::time::Duration; @@ -206,6 +326,7 @@ mod tests { open_duration: Duration::from_millis(50), half_open_max: 1, count_5xx: true, + max_hosts: 8192, } } @@ -238,4 +359,39 @@ mod tests { cb.record_failure("h.test"); assert!(cb.check("h.test").is_ok()); } + + #[test] + fn max_hosts_evicts_idle_closed() { + let mut c = cfg(10); + c.max_hosts = 2; + let cb = CircuitBreaker::from_config(&c); + assert!(cb.check("a.test").is_ok()); + assert!(cb.check("b.test").is_ok()); + assert!(cb.check("c.test").is_ok()); + assert!(cb.tracked_hosts() <= 2); + } + + #[test] + fn hop_guard_records_failure_on_drop() { + let cb = CircuitBreaker::from_config(&cfg(1)); + { + let hop = CircuitHop::admit(&cb, "h.test").expect("admit"); + assert_eq!(hop.host(), "h.test"); + // drop without settle + } + assert!( + cb.check("h.test").is_err(), + "unsettled hop must count as failure and trip threshold=1" + ); + } + + #[test] + fn hop_guard_success_disarms_drop() { + let cb = CircuitBreaker::from_config(&cfg(1)); + { + let hop = CircuitHop::admit(&cb, "h.test").expect("admit"); + hop.success(); + } + assert!(cb.check("h.test").is_ok()); + } } diff --git a/crates/corx-core/src/policy/mod.rs b/crates/corx-core/src/policy/mod.rs index ab420bc..79ea950 100644 --- a/crates/corx-core/src/policy/mod.rs +++ b/crates/corx-core/src/policy/mod.rs @@ -6,5 +6,5 @@ mod circuit; mod target; -pub use self::circuit::{CircuitBreaker, CircuitDecision}; +pub use self::circuit::{CircuitBreaker, CircuitHop}; pub use self::target::TargetPolicy; diff --git a/crates/corx-core/src/policy/target.rs b/crates/corx-core/src/policy/target.rs index eae27b5..2ecbe9f 100644 --- a/crates/corx-core/src/policy/target.rs +++ b/crates/corx-core/src/policy/target.rs @@ -1,10 +1,15 @@ //! Target host / scheme admission. +use http::Uri; + use crate::config::{TargetConfig, TargetMode}; use crate::error::ProxyError; use crate::proxy::url_parser::TargetUrl; /// Compiled target admission policy. +/// +/// Applied on the **initial** proxy target and on **every** redirect hop so +/// allowlists / denylists / `https_only` cannot be bypassed via 3xx. #[derive(Debug, Clone)] pub struct TargetPolicy { mode: TargetMode, @@ -45,7 +50,21 @@ impl TargetPolicy { /// Returns [`ProxyError::TargetNotAllowed`] when the host or scheme is /// outside policy. pub fn check(&self, target: &TargetUrl) -> Result<(), ProxyError> { - let scheme = target.url.scheme().to_ascii_lowercase(); + self.check_authority(target.url.scheme(), &target.host) + } + + /// Admit or reject a hop identified by scheme and host (redirects). + /// + /// Host comparison is case-insensitive. The host must already be in + /// ASCII / punycode form when it comes from the URL parser; redirect + /// `Location` hosts are lowercased here for matching. + /// + /// # Errors + /// + /// Returns [`ProxyError::TargetNotAllowed`] when the host or scheme is + /// outside policy. + pub fn check_authority(&self, scheme: &str, host: &str) -> Result<(), ProxyError> { + let scheme = scheme.to_ascii_lowercase(); if self.https_only && scheme != "https" { return Err(ProxyError::TargetNotAllowed(format!( "scheme `{scheme}` rejected (https_only)" @@ -57,7 +76,7 @@ impl TargetPolicy { ))); } - let host = target.host.to_ascii_lowercase(); + let host = host.to_ascii_lowercase(); let matched = self .hosts .iter() @@ -90,6 +109,22 @@ impl TargetPolicy { } } } + + /// Admit or reject a hop [`Uri`] (used on every redirect continue). + /// + /// # Errors + /// + /// Returns [`ProxyError::TargetNotAllowed`] or [`ProxyError::InvalidUrl`] + /// when the URI lacks a usable scheme/host or fails policy. + pub fn check_uri(&self, uri: &Uri) -> Result<(), ProxyError> { + let scheme = uri.scheme_str().ok_or_else(|| { + ProxyError::InvalidUrl("hop URI lacks a scheme".to_owned()) + })?; + let host = uri.host().ok_or_else(|| { + ProxyError::InvalidUrl("hop URI lacks a host".to_owned()) + })?; + self.check_authority(scheme, host) + } } fn host_matches(pattern: &str, host: &str) -> bool { @@ -153,4 +188,17 @@ mod tests { assert!(pol.check(&target("https://bad.test/x")).is_err()); assert!(pol.check(&target("https://good.test/x")).is_ok()); } + + #[test] + fn check_uri_enforces_allowlist_on_redirect_hop() { + let pol = TargetPolicy::from_config(&TargetConfig { + mode: TargetMode::Allowlist, + hosts: vec!["allowed.test".into()], + ..TargetConfig::default() + }); + let ok: Uri = "https://allowed.test/next".parse().unwrap(); + let bad: Uri = "https://evil.test/next".parse().unwrap(); + assert!(pol.check_uri(&ok).is_ok()); + assert!(pol.check_uri(&bad).is_err()); + } } diff --git a/crates/corx-core/src/proxy/headers.rs b/crates/corx-core/src/proxy/headers.rs index 392c14c..2404ca9 100644 --- a/crates/corx-core/src/proxy/headers.rs +++ b/crates/corx-core/src/proxy/headers.rs @@ -40,17 +40,32 @@ pub struct HeaderFilter { } impl HeaderFilter { - /// Compile a filter from the configured header names. Names that fail to - /// parse are silently dropped so a typo in the config cannot wedge the - /// proxy at startup. + /// Compile a filter from the configured header names. + /// + /// # Errors + /// + /// Returns an error when any name fails to parse as a valid HTTP header + /// name (fail-closed: typos must not silently disable stripping). + pub fn try_new(extra_deny: &[String]) -> Result { + let mut names = Vec::with_capacity(extra_deny.len()); + for raw in extra_deny { + let name = raw.parse::().map_err(|err| { + format!("invalid header name `{raw}`: {err}") + })?; + names.push(name); + } + Ok(Self { extra_deny: names }) + } + + /// Compile a filter, panicking only in tests that pass known-good names. #[must_use] pub fn new(extra_deny: &[String]) -> Self { - Self { - extra_deny: extra_deny - .iter() - .filter_map(|raw| raw.parse::().ok()) - .collect(), - } + Self::try_new(extra_deny).unwrap_or_else(|err| { + tracing::error!(%err, "invalid header filter config; using empty deny list"); + Self { + extra_deny: Vec::new(), + } + }) } /// Strip every header that must not survive a proxy hop: the diff --git a/crates/corx-core/src/proxy/upstream.rs b/crates/corx-core/src/proxy/upstream.rs index 925a34a..9a3e952 100644 --- a/crates/corx-core/src/proxy/upstream.rs +++ b/crates/corx-core/src/proxy/upstream.rs @@ -28,6 +28,7 @@ use tower_service::Service; use crate::config::RedirectPolicy; use crate::error::ProxyError; use crate::observability; +use crate::policy::{CircuitBreaker, CircuitHop, TargetPolicy}; use crate::proxy::redirect::{self, NextHop}; use crate::proxy::ssrf::SsrfGuard; @@ -58,7 +59,7 @@ pub struct ClientConfig { pub connect_timeout: Duration, /// Maximum number of redirects to follow when policy is `Follow`. pub max_redirects: u8, - /// Allow `https \u2192 http` redirect downgrades. Default: `false`. + /// Allow `https → http` redirect downgrades. Default: `false`. pub allow_https_to_http_downgrade: bool, /// How 3xx responses are handled. pub redirect_policy: RedirectPolicy, @@ -72,12 +73,14 @@ pub struct Upstream { client: HyperClient, guard: Arc, config: Arc, + target_policy: TargetPolicy, } impl std::fmt::Debug for Upstream { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Upstream") .field("config", &self.config) + .field("target_policy", &self.target_policy) .finish_non_exhaustive() } } @@ -87,13 +90,18 @@ impl Upstream { /// /// The supplied [`SsrfGuard`] is consulted inside the DNS resolver used /// by this client: no upstream connection is made to an address that - /// violates SSRF policy. + /// violates SSRF policy. [`TargetPolicy`] is enforced on the initial + /// request URI and on every redirect hop before connect. /// /// # Errors /// /// Returns an error when the platform TLS root store cannot be loaded /// (`rustls-platform-verifier`). - pub fn new(config: ClientConfig, guard: SsrfGuard) -> Result { + pub fn new( + config: ClientConfig, + guard: SsrfGuard, + target_policy: TargetPolicy, + ) -> Result { let guard = Arc::new(guard); let mut http = HttpConnector::new_with_resolver(GuardedResolver { @@ -122,68 +130,76 @@ impl Upstream { client, guard, config: Arc::new(config), + target_policy, }) } /// Executes a request against the upstream, following redirects up to /// the configured budget. /// + /// Each hop is admitted by [`TargetPolicy`] and the per-host + /// [`CircuitBreaker`] before the TCP connect. A [`CircuitHop`] guard + /// records failure if the hop is cancelled without settlement. + /// /// # Errors /// - /// Surfaces SSRF violations, DNS failures, upstream connection errors, - /// TLS failures and redirect loops as [`ProxyError`] variants. + /// Surfaces SSRF violations, target policy rejections, circuit opens, + /// DNS failures, upstream connection errors, TLS failures and redirect + /// loops as [`ProxyError`] variants. pub async fn execute( &self, request: Request, + circuit: &CircuitBreaker, ) -> Result, ProxyError> { let max_redirects = self.config.max_redirects; let allow_downgrade = self.config.allow_https_to_http_downgrade; let (mut state, first_request) = split_initial(request); - let target_host = state - .uri - .host() - .map_or_else(|| "unknown".to_owned(), str::to_owned); let mut hops: u8 = 0; let mut next_request = first_request; + let metric_host = host_of(&state.uri); loop { - let response = self - .client - .request(next_request) - .await - .map_err(|err| ProxyError::Upstream(Box::new(err)))?; + self.target_policy.check_uri(next_request.uri())?; + let hop = CircuitHop::admit(circuit, host_of(next_request.uri()))?; + + let response = match self.client.request(next_request).await { + Ok(response) => response, + Err(err) => { + hop.failure(); + return Err(ProxyError::Upstream(Box::new(err))); + } + }; if !redirect::is_redirect(response.status()) { - metrics::histogram!( - observability::REDIRECT_HOPS, - "target_host" => target_host.clone(), - ) - .record(f64::from(hops)); + settle_terminal(hop, circuit, response.status()); + record_redirect_hops(&metric_host, hops); return Ok(response); } match self.config.redirect_policy { RedirectPolicy::Block => { + hop.success(); return Err(ProxyError::RedirectBlocked( "redirect policy is block".to_owned(), )); } RedirectPolicy::Rewrite => { - // Return the 3xx as-is; the server layer rewrites Location - // to the proxy path-prefix form when desired. - metrics::histogram!( - observability::REDIRECT_HOPS, - "target_host" => target_host.clone(), - ) - .record(f64::from(hops)); + hop.success(); + record_redirect_hops(&metric_host, hops); return Ok(response); } RedirectPolicy::Follow => {} } if hops >= max_redirects { + hop.failure(); return Err(ProxyError::TooManyRedirects(max_redirects)); } + + // Transport hop succeeded (got 3xx). Settle before prepare_next so + // a Location parse/policy error cannot leave half-open unaccounted. + hop.success(); + match redirect::prepare_next(&mut state, &response, allow_downgrade)? { NextHop::Continue(req) => { next_request = *req; @@ -191,11 +207,7 @@ impl Upstream { } NextHop::Stop(reason) => { tracing::debug!(reason, hops, "stopping redirect chain"); - metrics::histogram!( - observability::REDIRECT_HOPS, - "target_host" => target_host.clone(), - ) - .record(f64::from(hops)); + record_redirect_hops(&metric_host, hops); return Ok(response); } } @@ -214,6 +226,32 @@ impl Upstream { pub fn user_agent(&self) -> &str { &self.config.user_agent } + + /// Target admission policy enforced on every hop. + #[must_use] + pub const fn target_policy(&self) -> &TargetPolicy { + &self.target_policy + } +} + +fn host_of(uri: &http::Uri) -> String { + uri.host().map_or_else(|| "unknown".to_owned(), str::to_owned) +} + +fn settle_terminal(hop: CircuitHop<'_>, circuit: &CircuitBreaker, status: http::StatusCode) { + if circuit.count_5xx() && status.is_server_error() { + hop.failure(); + } else { + hop.success(); + } +} + +fn record_redirect_hops(metric_host: &str, hops: u8) { + metrics::histogram!( + observability::REDIRECT_HOPS, + "target_host" => metric_host.to_owned(), + ) + .record(f64::from(hops)); } fn split_initial( diff --git a/crates/corx-server/Cargo.toml b/crates/corx-server/Cargo.toml index c850a33..6e236a2 100644 --- a/crates/corx-server/Cargo.toml +++ b/crates/corx-server/Cargo.toml @@ -56,6 +56,7 @@ regex = { workspace = true } ipnet = { workspace = true } arc-swap = { workspace = true } +dashmap = { workspace = true } foldhash = { workspace = true } tracing = { workspace = true } diff --git a/crates/corx-server/src/handlers/proxy.rs b/crates/corx-server/src/handlers/proxy.rs index 4f3ae5f..59dd8ea 100644 --- a/crates/corx-server/src/handlers/proxy.rs +++ b/crates/corx-server/src/handlers/proxy.rs @@ -7,7 +7,7 @@ use std::time::Instant; use axum::body::Body as AxumBody; use axum::extract::{ConnectInfo, State}; use axum::response::Response; -use corx_core::config::PreflightMode; +use corx_core::config::{PreflightMode, RedirectPolicy}; use corx_core::proxy::{self, InboundContext, TargetUrl, is_preflight}; use http::{HeaderMap, HeaderValue, Request, header}; use http_body_util::BodyExt as _; @@ -17,8 +17,7 @@ use super::outbound::{ set_host_from_target, }; use crate::error::ServerError; -use crate::observability::CountingBody; -use crate::observability::metrics as stats; +use crate::observability::{CountingBody, LimitingBody}; use crate::router::AppState; /// Primary proxy handler, bound to any path matching `/*path`. @@ -28,8 +27,8 @@ pub(crate) async fn proxy( request: Request, ) -> Result { let started = Instant::now(); - metrics::counter!(stats::REQUESTS_TOTAL).increment(1); - metrics::gauge!(stats::INFLIGHT_REQUESTS).increment(1.0); + metrics::counter!(crate::observability::metrics::REQUESTS_TOTAL).increment(1); + metrics::gauge!(crate::observability::metrics::INFLIGHT_REQUESTS).increment(1.0); let _decrement_inflight = InflightGuard; let outcome = serve(state, request, peer.ip()).await; @@ -38,16 +37,20 @@ pub(crate) async fn proxy( match &outcome { Ok(response) => { metrics::histogram!( - stats::REQUEST_DURATION, + crate::observability::metrics::REQUEST_DURATION, "status" => response.status().as_u16().to_string() ) .record(elapsed); } Err(error) => { let kind = error.kind(); - metrics::counter!(stats::UPSTREAM_ERRORS, "kind" => kind.as_str()).increment(1); + metrics::counter!( + crate::observability::metrics::UPSTREAM_ERRORS, + "kind" => kind.as_str() + ) + .increment(1); metrics::histogram!( - stats::REQUEST_DURATION, + crate::observability::metrics::REQUEST_DURATION, "status" => kind.status().as_u16().to_string() ) .record(elapsed); @@ -62,7 +65,7 @@ struct InflightGuard; impl Drop for InflightGuard { fn drop(&mut self) { - metrics::gauge!(stats::INFLIGHT_REQUESTS).decrement(1.0); + metrics::gauge!(crate::observability::metrics::INFLIGHT_REQUESTS).decrement(1.0); } } @@ -75,8 +78,7 @@ async fn serve( // CORS preflight: by default runs the same origin (and optional rate) // guards as real traffic so blacklisted origins cannot harvest 204s - // and OPTIONS cannot bypass the limiter. Set - // `security.preflight.mode = "open"` for classic cors-anywhere behaviour. + // and OPTIONS cannot bypass the limiter. if is_preflight(&request) { let preflight = &policies.config.security.preflight; if preflight.mode == PreflightMode::Enforce { @@ -94,21 +96,16 @@ async fn serve( return Ok(Response::from_parts(parts, axum_body)); } - // Stage 1: cheap origin / method / required-header guards. policies.guard.check_origin(&request)?; - // Extract and validate the upstream target URL. let target = proxy::extract_target(request.uri())?; + // First-hop admission (redirect hops re-check inside Upstream::execute). policies.target_policy.check(&target)?; - // Stage 2: multi-dimensional rate limiting now that we know the target. policies .guard .check_rate(&request, client_ip, Some(target.host.as_str()))?; - // Stage 3: per-host circuit breaker. - policies.circuit.check(&target.host)?; - drop(policies); execute_proxy(state, request, target, client_ip).await } @@ -119,9 +116,6 @@ async fn execute_proxy( target: TargetUrl, client_ip: IpAddr, ) -> Result { - // The listener (and therefore the inbound scheme/port) is locked at - // startup; pull it from the immutable snapshot so a SIGHUP-triggered - // reload mid-request cannot change it underneath us. let tls_on = state.build.immutable_server.tls.is_some(); let local_port = state.build.immutable_server.bind.port(); @@ -142,8 +136,6 @@ async fn execute_proxy( let policies = state.build.policies(); let (mut parts, body) = request.into_parts(); - // Preserve inbound headers so that CORS can still reflect the original - // Origin after the request has been consumed by the upstream client. let inbound_headers = parts.headers.clone(); policies.request_filter.apply(&mut parts.headers); @@ -163,54 +155,49 @@ async fn execute_proxy( parts.uri = target.to_uri()?; let outbound = Request::from_parts(parts, axum_to_upstream_body(body)); - let host = target.host.clone(); - let count_5xx = policies.circuit.count_5xx(); - let upstream_started = Instant::now(); - let upstream_response = policies.upstream.execute(outbound).await; + let upstream_response = policies + .upstream + .execute(outbound, &policies.circuit) + .await; let upstream_elapsed = upstream_started.elapsed().as_secs_f64(); let response = match upstream_response { Ok(response) => { - if count_5xx && response.status().is_server_error() { - policies.circuit.record_failure(&host); - } else { - policies.circuit.record_success(&host); - } metrics::histogram!( - stats::UPSTREAM_DURATION, + crate::observability::metrics::UPSTREAM_DURATION, "status" => response.status().as_u16().to_string() ) .record(upstream_elapsed); response } Err(error) => { - // Transport / upstream failures trip the breaker; client policy - // errors (SSRF, invalid URL) are not host health signals. - if matches!( - error.kind(), - corx_core::error::ErrorKind::UpstreamUnreachable - | corx_core::error::ErrorKind::UpstreamTimeout - | corx_core::error::ErrorKind::TlsFailure - | corx_core::error::ErrorKind::DnsFailure - ) { - policies.circuit.record_failure(&host); - } let kind = error.kind(); - metrics::histogram!(stats::UPSTREAM_DURATION, "status" => kind.as_str()) - .record(upstream_elapsed); + metrics::histogram!( + crate::observability::metrics::UPSTREAM_DURATION, + "status" => kind.as_str() + ) + .record(upstream_elapsed); return Err(error.into()); } }; - Ok(shape_response(response, &state, &target, &inbound_headers)) + let redirect_policy = policies.config.limits.redirect_policy; + Ok(shape_response( + response, + &state, + &target, + &inbound_headers, + redirect_policy, + )) } fn shape_response( response: hyper::Response, state: &AppState, target: &TargetUrl, - request_headers: &HeaderMap, + _request_headers: &HeaderMap, + redirect_policy: RedirectPolicy, ) -> Response { let policies = state.build.policies(); @@ -222,11 +209,34 @@ fn shape_response( } append_via_header(&mut response_parts.headers); - let mut reassembled = hyper::Response::from_parts(response_parts, response_body); - proxy::apply_to_response(&mut reassembled, request_headers, policies.cors.as_ref()); + if redirect_policy == RedirectPolicy::Rewrite { + rewrite_location_header(&mut response_parts.headers); + } + + // CORS is applied solely by `cors_layer` so every response path (errors, + // load-shed, success) shares one owner. + + let max_response = state.build.immutable_limits.max_response_body_bytes; + let counted = CountingBody::new(response_body, "response"); + let limited = LimitingBody::new(counted, max_response); + let axum_body = AxumBody::new(limited.map_err(axum::Error::new)); + Response::from_parts(response_parts, axum_body) +} - let (final_parts, final_body) = reassembled.into_parts(); - let counted = CountingBody::new(final_body, "response"); - let axum_body = AxumBody::new(counted.map_err(axum::Error::new)); - Response::from_parts(final_parts, axum_body) +/// Rewrite absolute `Location` values into the cors-anywhere path-prefix form +/// so the browser stays on the proxy for the next hop. +fn rewrite_location_header(headers: &mut HeaderMap) { + let Some(location) = headers.get(header::LOCATION) else { + return; + }; + let Ok(raw) = location.to_str() else { + return; + }; + if !(raw.starts_with("http://") || raw.starts_with("https://")) { + return; + } + let rewritten = format!("/{raw}"); + if let Ok(value) = HeaderValue::from_str(&rewritten) { + headers.insert(header::LOCATION, value); + } } diff --git a/crates/corx-server/src/hot_reload.rs b/crates/corx-server/src/hot_reload.rs index 2c25361..270fa9f 100644 --- a/crates/corx-server/src/hot_reload.rs +++ b/crates/corx-server/src/hot_reload.rs @@ -39,7 +39,10 @@ pub struct ReloadHandle { immutable_server: Arc, immutable_metrics_endpoint: String, immutable_max_body_bytes: u64, + immutable_max_request_header_bytes: u32, immutable_request_timeout: std::time::Duration, + immutable_inflight_max: u32, + immutable_max_response_body_bytes: u64, } impl ReloadHandle { @@ -55,7 +58,8 @@ impl ReloadHandle { pub fn reload(&self, path: Option<&Path>) -> anyhow::Result<()> { let config = config_loader::load(path)?; self.assert_immutable(&config)?; - let policies = LivePolicies::build(config)?; + let previous = self.policies.load_full(); + let policies = LivePolicies::build_from(config, Some(previous.as_ref()))?; self.policies.store(Arc::new(policies)); Ok(()) } @@ -79,9 +83,21 @@ impl ReloadHandle { if new.limits.max_request_body_bytes != self.immutable_max_body_bytes { anyhow::bail!("limits.max_request_body_bytes cannot be changed by reload"); } + if new.limits.max_request_header_bytes != self.immutable_max_request_header_bytes { + anyhow::bail!("limits.max_request_header_bytes cannot be changed by reload"); + } if new.limits.request_timeout != self.immutable_request_timeout { anyhow::bail!("limits.request_timeout cannot be changed by reload"); } + // inflight_max / max_response_body_bytes are taken from immutable_limits + // at serve time for load-shed and response capping; reject mid-flight + // changes so operators restart for those knobs. + if new.limits.inflight_max != self.immutable_inflight_max { + anyhow::bail!("limits.inflight_max cannot be changed by reload"); + } + if new.limits.max_response_body_bytes != self.immutable_max_response_body_bytes { + anyhow::bail!("limits.max_response_body_bytes cannot be changed by reload"); + } if new.observability.metrics_endpoint != self.immutable_metrics_endpoint { anyhow::bail!("observability.metrics_endpoint cannot be changed by reload"); } @@ -116,7 +132,10 @@ impl ServerBuild { immutable_server: Arc::clone(&self.immutable_server), immutable_metrics_endpoint: self.immutable_metrics_endpoint.clone(), immutable_max_body_bytes: self.immutable_limits.max_request_body_bytes, + immutable_max_request_header_bytes: self.immutable_limits.max_request_header_bytes, immutable_request_timeout: self.immutable_limits.request_timeout, + immutable_inflight_max: self.immutable_limits.inflight_max, + immutable_max_response_body_bytes: self.immutable_limits.max_response_body_bytes, } } } @@ -199,7 +218,10 @@ mod tests { let immutable_server = Arc::new(config.server.clone()); let immutable_metrics_endpoint = config.observability.metrics_endpoint.clone(); let immutable_max_body_bytes = config.limits.max_request_body_bytes; + let immutable_max_request_header_bytes = config.limits.max_request_header_bytes; let immutable_request_timeout = config.limits.request_timeout; + let immutable_inflight_max = config.limits.inflight_max; + let immutable_max_response_body_bytes = config.limits.max_response_body_bytes; let policies = LivePolicies::build(config.clone()).expect("policies build"); let handle = ReloadHandle { @@ -207,7 +229,10 @@ mod tests { immutable_server, immutable_metrics_endpoint, immutable_max_body_bytes, + immutable_max_request_header_bytes, immutable_request_timeout, + immutable_inflight_max, + immutable_max_response_body_bytes, }; (handle, config) } @@ -251,6 +276,30 @@ mod tests { .expect("mutable field change should pass"); } + #[test] + fn rebuild_retains_circuit_when_only_cors_changes() { + ensure_crypto_provider(); + let config = Config::default(); + let first = LivePolicies::build(config.clone()).expect("build"); + first.circuit.record_failure("h.test"); + first.circuit.record_failure("h.test"); + first.circuit.record_failure("h.test"); + first.circuit.record_failure("h.test"); + first.circuit.record_failure("h.test"); + assert!( + first.circuit.check("h.test").is_err(), + "breaker should be open before reload" + ); + + let mut next = config; + next.cors.allow_any_origin = !next.cors.allow_any_origin; + let second = LivePolicies::build_from(next, Some(&first)).expect("rebuild"); + assert!( + second.circuit.check("h.test").is_err(), + "open circuit must survive reload when circuit_breaker config is unchanged" + ); + } + #[test] fn tls_eq_compares_field_wise() { assert!(tls_eq(None, None)); diff --git a/crates/corx-server/src/lib.rs b/crates/corx-server/src/lib.rs index 8a5d842..f4eb314 100644 --- a/crates/corx-server/src/lib.rs +++ b/crates/corx-server/src/lib.rs @@ -15,14 +15,10 @@ #![forbid(unsafe_code)] #![deny(missing_docs)] -// Several workspace dependencies are wired in via Cargo.toml ahead of the -// modules that consume them (M2 forwarded headers, M3 access log + load shed, -// M4 OpenTelemetry, M5 config hot reload). The lint will trip until those -// modules land; revisit and remove the allow once the remaining milestones -// are implemented. #![allow( unused_crate_dependencies, - reason = "transitive deps wired ahead of M2-M5 modules" + reason = "Workspace deps shared across optional features (tls/otel) \ + and integration tests; lib target alone would false-positive." )] pub mod config_loader; diff --git a/crates/corx-server/src/middleware/load_shed.rs b/crates/corx-server/src/middleware/load_shed.rs index 0669a24..574a3c5 100644 --- a/crates/corx-server/src/middleware/load_shed.rs +++ b/crates/corx-server/src/middleware/load_shed.rs @@ -1,17 +1,15 @@ //! Process-wide load shedding. //! -//! When [`GlobalLimitConfig::inflight_max`] is non-zero, this layer rejects +//! When [`LimitsConfig::inflight_max`] is non-zero, this layer rejects //! every additional request with `503 Service Unavailable` once the //! configured concurrency cap is reached. Rejections carry a `Retry-After` -//! hint and bump the `corx_rate_limited_total{dimension="global"}` counter -//! so existing dashboards light up identically to other shed paths. +//! hint and bump the `corx_rate_limited_total{dimension="inflight"}` counter. //! -//! `inflight_max = 0` keeps the layer on the request path but as a no-op, -//! avoiding the cost of swapping layer stacks on configuration reload. +//! `inflight_max = 0` keeps the layer on the request path but as a no-op. //! -//! [`GlobalLimitConfig::inflight_max`]: corx_core::config::GlobalLimitConfig::inflight_max +//! [`LimitsConfig::inflight_max`]: corx_core::config::LimitsConfig::inflight_max -use std::sync::atomic::Ordering; +use std::sync::atomic::{AtomicU64, Ordering}; use axum::extract::{Request, State}; use axum::middleware::Next; @@ -27,23 +25,33 @@ pub async fn load_shed_layer( request: Request, next: Next, ) -> Response { - let max = u64::from(state.build.policies().config.rate_limit.global.inflight_max); + let max = u64::from(state.build.immutable_limits.inflight_max); if max == 0 { return next.run(request).await; } - let counter = &state.build.inflight; + let counter = state.build.inflight.as_ref(); let prev = counter.fetch_add(1, Ordering::AcqRel); if prev >= max { - // Restore the counter; we never actually entered service. counter.fetch_sub(1, Ordering::AcqRel); - metrics::counter!(observability::RATE_LIMITED, "dimension" => "global").increment(1); + metrics::counter!(observability::RATE_LIMITED, "dimension" => "inflight").increment(1); return shed_response(); } - let response = next.run(request).await; - counter.fetch_sub(1, Ordering::AcqRel); - response + // RAII permit: decrements on panic, cancel, or normal return. + let _permit = InflightPermit { counter }; + next.run(request).await +} + +/// Holds one inflight slot until drop. +struct InflightPermit<'a> { + counter: &'a AtomicU64, +} + +impl Drop for InflightPermit<'_> { + fn drop(&mut self) { + self.counter.fetch_sub(1, Ordering::AcqRel); + } } fn shed_response() -> Response { diff --git a/crates/corx-server/src/middleware/rate_limit.rs b/crates/corx-server/src/middleware/rate_limit.rs index 569f851..05fa9d5 100644 --- a/crates/corx-server/src/middleware/rate_limit.rs +++ b/crates/corx-server/src/middleware/rate_limit.rs @@ -1,13 +1,18 @@ //! Multi-dimensional rate limiter (per-Origin / per-IP / per-Target-Host / //! Global) backed by `governor`'s GCRA token bucket. //! -//! All four dimensions are independent and can be enabled \u00e0 la carte. The +//! All four dimensions are independent and can be enabled à la carte. The //! hot path is lock-free: keyed limiters use `dashmap`, the global limiter is //! a single atomic. When a dimension trips, the //! `corx_rate_limited_total{dimension}` counter is incremented before the //! request is rejected so operators can see *which* dimension is shedding //! load. +//! +//! Keyed dimensions enforce [`RateLimitConfig::max_keys`]: when the key +//! registry is full, unknown keys are rejected fail-closed rather than +//! growing without bound. +use std::hash::Hash; use std::net::IpAddr; use std::num::NonZeroU32; use std::sync::Arc; @@ -17,6 +22,8 @@ use corx_core::config::{ }; use corx_core::error::ProxyError; use corx_core::observability; +use dashmap::DashMap; +use foldhash::fast::RandomState; use governor::clock::QuantaClock; use governor::state::keyed::DashMapStateStore; use governor::state::{InMemoryState, NotKeyed}; @@ -47,6 +54,7 @@ pub struct RateLimiter { struct Inner { enabled: bool, + max_keys: usize, origin: Option, ip: Option, host: Option, @@ -55,22 +63,26 @@ struct Inner { struct OriginDimension { limiter: KeyedLimiter, + keys: DashMap, unlimited: RegexSet, } struct IpDimension { limiter: KeyedLimiter, + keys: DashMap, trusted: Vec, } struct HostDimension { limiter: KeyedLimiter, + keys: DashMap, } impl std::fmt::Debug for RateLimiter { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("RateLimiter") .field("enabled", &self.inner.enabled) + .field("max_keys", &self.inner.max_keys) .field("origin_enabled", &self.inner.origin.is_some()) .field("ip_enabled", &self.inner.ip.is_some()) .field("target_host_enabled", &self.inner.host.is_some()) @@ -91,6 +103,7 @@ impl RateLimiter { return Ok(Self { inner: Arc::new(Inner { enabled: false, + max_keys: cfg.max_keys.max(1), origin: None, ip: None, host: None, @@ -102,11 +115,12 @@ impl RateLimiter { let origin = build_origin(&cfg.origin)?; let ip = build_ip(&cfg.ip)?; let host = build_host(cfg.target_host)?; - let global = build_global(&cfg.global)?; + let global = build_global(cfg.global)?; Ok(Self { inner: Arc::new(Inner { enabled: true, + max_keys: cfg.max_keys.max(1), origin, ip, host, @@ -122,32 +136,32 @@ impl RateLimiter { /// /// # Errors /// - /// Returns [`ProxyError::RateLimited`] when any dimension is exhausted. + /// Returns [`ProxyError::RateLimited`] when any dimension is exhausted + /// or when a keyed store is full and the key is new. pub fn check(&self, ctx: &RateContext<'_>) -> Result<(), ProxyError> { if !self.inner.enabled { return Ok(()); } + let max_keys = self.inner.max_keys; + if let Some(dim) = self.inner.origin.as_ref() && let Some(origin) = ctx.origin && !dim.unlimited.is_match(origin) - && dim.limiter.check_key(&origin.to_owned()).is_err() { - return reject("origin"); + admit_and_check_str(&dim.limiter, &dim.keys, origin, max_keys, "origin")?; } if let Some(dim) = self.inner.ip.as_ref() && !is_trusted(&dim.trusted, ctx.client_ip) - && dim.limiter.check_key(&ctx.client_ip).is_err() { - return reject("ip"); + admit_and_check(&dim.limiter, &dim.keys, &ctx.client_ip, max_keys, "ip")?; } if let Some(dim) = self.inner.host.as_ref() && let Some(host) = ctx.target_host - && dim.limiter.check_key(&host.to_owned()).is_err() { - return reject("target_host"); + admit_and_check_str(&dim.limiter, &dim.keys, host, max_keys, "target_host")?; } if let Some(global) = self.inner.global.as_ref() @@ -160,6 +174,63 @@ impl RateLimiter { } } +fn admit_and_check_str( + limiter: &KeyedLimiter, + keys: &DashMap, + key: &str, + max_keys: usize, + dimension: &'static str, +) -> Result<(), ProxyError> { + let owned = key.to_owned(); + admit_key(keys, &owned, max_keys, dimension)?; + if limiter.check_key(&owned).is_err() { + return reject(dimension); + } + Ok(()) +} + +fn admit_and_check( + limiter: &KeyedLimiter, + keys: &DashMap, + key: &K, + max_keys: usize, + dimension: &'static str, +) -> Result<(), ProxyError> +where + K: Clone + Eq + Hash, +{ + admit_key(keys, key, max_keys, dimension)?; + if limiter.check_key(key).is_err() { + return reject(dimension); + } + Ok(()) +} + +/// Register a new key if needed. On concurrent overshoot past `max_keys`, +/// remove the insertion and reject (fail-closed). +fn admit_key( + keys: &DashMap, + key: &K, + max_keys: usize, + dimension: &'static str, +) -> Result<(), ProxyError> +where + K: Clone + Eq + Hash, +{ + if keys.contains_key(key) { + return Ok(()); + } + if keys.len() >= max_keys { + return reject(dimension); + } + keys.insert(key.clone(), ()); + if keys.len() > max_keys { + keys.remove(key); + return reject(dimension); + } + Ok(()) +} + fn reject(dimension: &'static str) -> Result<(), ProxyError> { metrics::counter!(observability::RATE_LIMITED, "dimension" => dimension).increment(1); Err(ProxyError::RateLimited) @@ -172,6 +243,10 @@ fn quota(rps: u32, burst: u32) -> anyhow::Result { Ok(Quota::per_second(rps).allow_burst(burst)) } +fn empty_key_map() -> DashMap { + DashMap::with_hasher(RandomState::default()) +} + fn build_origin(cfg: &OriginLimitConfig) -> anyhow::Result> { if cfg.rps == 0 { return Ok(None); @@ -181,7 +256,11 @@ fn build_origin(cfg: &OriginLimitConfig) -> anyhow::Result anyhow::Result> { @@ -193,6 +272,7 @@ fn build_ip(cfg: &IpLimitConfig) -> anyhow::Result> { GovRateLimiter::dashmap_with_clock(q, QuantaClock::default()); Ok(Some(IpDimension { limiter, + keys: empty_key_map(), trusted: cfg.trusted_cidrs.clone(), })) } @@ -204,10 +284,13 @@ fn build_host(cfg: HostLimitConfig) -> anyhow::Result> { let q = quota(cfg.rps, cfg.burst)?; let limiter: KeyedLimiter = GovRateLimiter::dashmap_with_clock(q, QuantaClock::default()); - Ok(Some(HostDimension { limiter })) + Ok(Some(HostDimension { + limiter, + keys: empty_key_map(), + })) } -fn build_global(cfg: &GlobalLimitConfig) -> anyhow::Result> { +fn build_global(cfg: GlobalLimitConfig) -> anyhow::Result> { if cfg.rps == 0 { return Ok(None); } @@ -235,6 +318,7 @@ mod tests { fn cfg() -> RateLimitConfig { RateLimitConfig { enabled: true, + max_keys: 16_384, origin: OriginLimitConfig { rps: 0, burst: 0, @@ -246,11 +330,7 @@ mod tests { trusted_cidrs: vec![], }, target_host: HostLimitConfig { rps: 0, burst: 0 }, - global: GlobalLimitConfig { - rps: 0, - burst: 0, - inflight_max: 0, - }, + global: GlobalLimitConfig { rps: 0, burst: 0 }, } } @@ -318,7 +398,6 @@ mod tests { let lim = RateLimiter::from_config(&c).unwrap(); assert!(lim.check(&ctx(None, "1.2.3.4", "x.test")).is_ok()); assert!(lim.check(&ctx(None, "1.2.3.4", "x.test")).is_err()); - // Different IP has its own bucket. assert!(lim.check(&ctx(None, "1.2.3.5", "x.test")).is_ok()); } @@ -342,7 +421,6 @@ mod tests { let lim = RateLimiter::from_config(&c).unwrap(); assert!(lim.check(&ctx(None, "1.1.1.1", "popular.test")).is_ok()); assert!(lim.check(&ctx(None, "2.2.2.2", "popular.test")).is_err()); - // Different upstream uses an independent bucket. assert!(lim.check(&ctx(None, "3.3.3.3", "rare.test")).is_ok()); } @@ -359,8 +437,6 @@ mod tests { #[test] fn first_failing_dimension_wins() { let mut c = cfg(); - // Origin is the most specific; an over-quota origin must not consume - // a token from the more permissive dimensions below. c.origin.rps = 1; c.origin.burst = 1; c.global.rps = 1_000; @@ -369,4 +445,27 @@ mod tests { assert!(lim.check(&ctx(Some("o"), "1.1.1.1", "a.test")).is_ok()); assert!(lim.check(&ctx(Some("o"), "1.1.1.1", "a.test")).is_err()); } + + #[test] + fn max_keys_rejects_new_keys_when_full() { + let mut c = cfg(); + c.max_keys = 1; + c.origin.rps = 100; + c.origin.burst = 100; + let lim = RateLimiter::from_config(&c).unwrap(); + assert!( + lim.check(&ctx(Some("https://a.test"), "1.1.1.1", "h")) + .is_ok() + ); + assert!( + lim.check(&ctx(Some("https://b.test"), "1.1.1.1", "h")) + .is_err(), + "second distinct origin must be rejected when max_keys=1" + ); + assert!( + lim.check(&ctx(Some("https://a.test"), "1.1.1.1", "h")) + .is_ok(), + "existing key remains admissible" + ); + } } diff --git a/crates/corx-server/src/middleware/request_guard.rs b/crates/corx-server/src/middleware/request_guard.rs index 52178df..eddecaf 100644 --- a/crates/corx-server/src/middleware/request_guard.rs +++ b/crates/corx-server/src/middleware/request_guard.rs @@ -27,6 +27,13 @@ impl RequestGuard { } } + /// Clone of the compiled multi-dimensional rate limiter (for hot-reload + /// state retention when `rate_limit` config is unchanged). + #[must_use] + pub fn rate_limiter(&self) -> RateLimiter { + self.rate_limit.clone() + } + /// Origin policy / required headers / blocked methods. Cheap, runs /// before URL extraction so that obviously-bad requests never reach the /// parser. diff --git a/crates/corx-server/src/observability/metering.rs b/crates/corx-server/src/observability/metering.rs index e7d087c..6275274 100644 --- a/crates/corx-server/src/observability/metering.rs +++ b/crates/corx-server/src/observability/metering.rs @@ -10,6 +10,7 @@ use std::pin::Pin; use std::task::{Context, Poll}; use bytes::Buf; +use corx_core::error::ProxyError; use corx_core::observability::BYTES_TRANSFERRED; use http_body::{Body, Frame, SizeHint}; use pin_project_lite::pin_project; @@ -69,3 +70,75 @@ where self.inner.size_hint() } } + +pin_project! { + /// Caps total bytes read from an inner body. When the budget is exhausted + /// the stream ends with [`ProxyError::PayloadTooLarge`]. + /// + /// `max_bytes = 0` disables the cap (pass-through). + pub struct LimitingBody { + #[pin] + inner: B, + max_bytes: u64, + seen: u64, + } +} + +impl LimitingBody { + /// Wrap `inner`, aborting after `max_bytes` of data frames. + pub const fn new(inner: B, max_bytes: u64) -> Self { + Self { + inner, + max_bytes, + seen: 0, + } + } +} + +impl Body for LimitingBody +where + B: Body, + B::Data: Buf, + B::Error: std::error::Error + Send + Sync + 'static, +{ + type Data = B::Data; + type Error = Box; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + let this = self.project(); + if over_budget(*this.max_bytes, *this.seen) { + return Poll::Ready(Some(Err(Box::new(ProxyError::PayloadTooLarge)))); + } + + match this.inner.poll_frame(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(None) => Poll::Ready(None), + Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(Box::new(err)))), + Poll::Ready(Some(Ok(frame))) => { + if let Some(data) = frame.data_ref() { + let n = u64::try_from(data.remaining()).unwrap_or(u64::MAX); + *this.seen = this.seen.saturating_add(n); + } + if over_budget(*this.max_bytes, *this.seen) { + return Poll::Ready(Some(Err(Box::new(ProxyError::PayloadTooLarge)))); + } + Poll::Ready(Some(Ok(frame))) + } + } + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.inner.size_hint() + } +} + +const fn over_budget(max_bytes: u64, seen: u64) -> bool { + max_bytes > 0 && seen > max_bytes +} diff --git a/crates/corx-server/src/observability/mod.rs b/crates/corx-server/src/observability/mod.rs index 813bcbe..7483002 100644 --- a/crates/corx-server/src/observability/mod.rs +++ b/crates/corx-server/src/observability/mod.rs @@ -7,6 +7,6 @@ pub mod metrics; pub mod otel; mod tracing; -pub use self::metering::CountingBody; +pub use self::metering::{CountingBody, LimitingBody}; pub use self::metrics::{MetricsHandle, active_features, init_metrics}; pub use self::tracing::init_tracing; diff --git a/crates/corx-server/src/router.rs b/crates/corx-server/src/router.rs index 79d45a9..c982ee5 100644 --- a/crates/corx-server/src/router.rs +++ b/crates/corx-server/src/router.rs @@ -7,6 +7,7 @@ use axum::extract::DefaultBodyLimit; use axum::middleware; use axum::routing::{any, get}; use http::StatusCode; +use tower_http::catch_panic::CatchPanicLayer; use tower_http::timeout::TimeoutLayer; use tower_http::trace::TraceLayer; @@ -34,10 +35,17 @@ impl AppState { } /// Builds the `axum` router with every middleware layer registered. +/// +/// Layer order (outer → inner): +/// +/// ```text +/// Trace → access_log → cors → Timeout → BodyLimit → CatchPanic +/// → header_limit → load_shed → auth → handler +/// ``` +/// +/// CORS sits outside timeout / body-limit / auth / load-shed so 401/503/431/ +/// 504/413 responses still carry browser-readable ACAO headers. pub fn build_router(state: AppState) -> Router<()> { - // Body size and timeout are baked into the router at startup; they - // come from the immutable snapshot for the same reason a SIGHUP reload - // cannot change them mid-flight. let max_body = state.build.immutable_limits.max_request_body_bytes; let request_timeout = state.build.immutable_limits.request_timeout; let metrics_path = state.build.immutable_metrics_endpoint.clone(); @@ -61,15 +69,11 @@ pub fn build_router(state: AppState) -> Router<()> { router = router.route(&metrics_path, get(handlers::prometheus_metrics)); } + let cors_state = state.clone(); + router .fallback(any(handlers::proxy)) .method_not_allowed_fallback(handlers::not_found) - // CORS runs first (innermost), so even responses from later layers - // gain the headers; load-shed sits just outside it so 503s also - // leave with valid CORS metadata. Bearer auth is outside the - // handler but inside load-shed so rejected auths still count as - // load for abuse control. - .layer(middleware::from_fn_with_state(state.clone(), cors_layer)) .layer(middleware::from_fn_with_state(state.clone(), auth_layer)) .layer(middleware::from_fn_with_state( state.clone(), @@ -80,14 +84,14 @@ pub fn build_router(state: AppState) -> Router<()> { header_limit_layer, )) .with_state(state) + .layer(CatchPanicLayer::new()) .layer(DefaultBodyLimit::max(body_limit)) .layer(TimeoutLayer::with_status_code( StatusCode::GATEWAY_TIMEOUT, request_timeout, )) - // Access log sits at the outermost level so it observes the *final* - // status (including timeouts and load-shed responses) and the wall - // clock duration the client actually saw. + // Outside timeout/body so 504/413 also get CORS headers. + .layer(middleware::from_fn_with_state(cors_state, cors_layer)) .layer(middleware::from_fn(access_log_layer)) .layer(TraceLayer::new_for_http()) } diff --git a/crates/corx-server/src/state.rs b/crates/corx-server/src/state.rs index 344d2d8..8be55f7 100644 --- a/crates/corx-server/src/state.rs +++ b/crates/corx-server/src/state.rs @@ -4,17 +4,20 @@ //! //! Fields fall into two camps: //! -//! * **Hot-swappable** \u2014 CORS, header filters, request guards (origin lists + +//! * **Hot-swappable** — CORS, header filters, request guards (origin lists + //! rate limiter), the upstream HTTP client and the source [`Config`] are //! bundled into [`LivePolicies`] and stored behind an //! [`ArcSwap`](arc_swap::ArcSwap). Each request loads a single snapshot so //! the policy view is consistent for the whole handler chain even if a //! reload races in mid-request. -//! * **Frozen-at-startup** \u2014 the listener (bind address, TLS, HTTP/2 toggle), +//! * **Process state retained across reload when config is unchanged** — +//! `circuit` and `rate` (`RateLimiter`) keep their in-memory maps so a +//! SIGHUP that only tweaks CORS does not reset open breakers or GCRA +//! budgets. `upstream` (connection pool + SSRF + target policy) is rebuilt +//! when ssrf / target / upstream / connect / redirect settings change. +//! * **Frozen-at-startup** — the listener (bind address, TLS, HTTP/2 toggle), //! request body limits and timeouts that are baked into the `axum::Router`, -//! and the metrics endpoint path. These are recorded under `immutable_*` -//! so a reload can detect attempts to change them and reject the new -//! configuration with a clear log message. +//! and the metrics endpoint path. use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64}; @@ -28,10 +31,6 @@ use crate::middleware::{OriginPolicy, RateLimiter, RequestGuard}; use crate::observability::MetricsHandle; /// Atomically-replaceable bundle of policy state. -/// -/// Every request handler dereferences exactly one snapshot so views are -/// internally consistent even while a SIGHUP-driven reload swaps the -/// pointer underneath. #[derive(Debug)] pub struct LivePolicies { /// Source configuration this snapshot was built from. @@ -45,14 +44,13 @@ pub struct LivePolicies { /// Inbound guards (origin allow/deny, multi-dimensional rate limiter, /// required-header check). pub guard: RequestGuard, - /// Target host / scheme admission. + /// Target host / scheme admission (also enforced inside `upstream` on + /// every redirect hop). pub target_policy: TargetPolicy, - /// Per-host circuit breaker (process-local). + /// Per-host circuit breaker (process-local; retained across reload when + /// circuit config is unchanged). pub circuit: CircuitBreaker, - /// Upstream HTTP client. Rebuilt on reload, so SIGHUP discards the - /// existing connection pool. Reloads are deliberate and rare, so this - /// trade-off is acceptable in exchange for picking up fresh SSRF, - /// timeout and pool-tuning settings without process restart. + /// Upstream HTTP client (pool + SSRF + hop target policy). pub upstream: Upstream, } @@ -64,30 +62,57 @@ impl LivePolicies { /// Returns an error when any component fails to compile (invalid /// regex, malformed CIDR, missing TLS material, etc.). pub fn build(config: Config) -> anyhow::Result { + Self::build_from(config, None) + } + + /// Build a new snapshot, reusing process state from `previous` when the + /// corresponding config sections are unchanged. + /// + /// # Errors + /// + /// Returns an error when any component fails to compile. + pub fn build_from(config: Config, previous: Option<&Self>) -> anyhow::Result { let cors = CorsPolicy::from_config(&config.cors); - let request_filter = HeaderFilter::new(&config.security.remove_request_headers); - let response_filter = HeaderFilter::new(&config.security.remove_response_headers); - - let resolver = corx_core::proxy::build_resolver(); - let ssrf = SsrfGuard::new(&config.ssrf, resolver); - - let client_config = corx_core::proxy::ClientConfig { - pool_max_idle_per_host: config.upstream.pool_max_idle_per_host, - pool_idle_timeout: config.upstream.pool_idle_timeout, - connect_timeout: config.limits.connect_timeout, - max_redirects: config.limits.max_redirects, - allow_https_to_http_downgrade: config.limits.allow_https_to_http_downgrade, - redirect_policy: config.limits.redirect_policy, - user_agent: config.upstream.user_agent.clone(), + let request_filter = HeaderFilter::try_new(&config.security.remove_request_headers) + .map_err(|err| anyhow::anyhow!("security.remove_request_headers: {err}"))?; + let response_filter = HeaderFilter::try_new(&config.security.remove_response_headers) + .map_err(|err| anyhow::anyhow!("security.remove_response_headers: {err}"))?; + + let target_policy = TargetPolicy::from_config(&config.target); + + let rate_limiter = match previous { + Some(prev) if prev.config.rate_limit == config.rate_limit => prev.guard.rate_limiter(), + _ => RateLimiter::from_config(&config.rate_limit)?, }; - let upstream = Upstream::new(client_config, ssrf) - .map_err(|err| anyhow::anyhow!("upstream client: {err}"))?; let origin_policy = OriginPolicy::from_config(&config.security); - let rate_limiter = RateLimiter::from_config(&config.rate_limit)?; let guard = RequestGuard::new(origin_policy, rate_limiter); - let target_policy = TargetPolicy::from_config(&config.target); - let circuit = CircuitBreaker::from_config(&config.circuit_breaker); + + let circuit = match previous { + Some(prev) if prev.config.circuit_breaker == config.circuit_breaker => { + prev.circuit.clone() + } + _ => CircuitBreaker::from_config(&config.circuit_breaker), + }; + + let upstream = match previous { + Some(prev) if upstream_config_eq(&prev.config, &config) => prev.upstream.clone(), + _ => { + let resolver = corx_core::proxy::build_resolver(); + let ssrf = SsrfGuard::new(&config.ssrf, resolver); + let client_config = corx_core::proxy::ClientConfig { + pool_max_idle_per_host: config.upstream.pool_max_idle_per_host, + pool_idle_timeout: config.upstream.pool_idle_timeout, + connect_timeout: config.limits.connect_timeout, + max_redirects: config.limits.max_redirects, + allow_https_to_http_downgrade: config.limits.allow_https_to_http_downgrade, + redirect_policy: config.limits.redirect_policy, + user_agent: config.upstream.user_agent.clone(), + }; + Upstream::new(client_config, ssrf, target_policy.clone()) + .map_err(|err| anyhow::anyhow!("upstream client: {err}"))? + } + }; Ok(Self { config: Arc::new(config), @@ -102,10 +127,18 @@ impl LivePolicies { } } +/// Fields that force a full upstream client rebuild (pool + SSRF + hop policy). +fn upstream_config_eq(a: &Config, b: &Config) -> bool { + a.ssrf == b.ssrf + && a.target == b.target + && a.upstream == b.upstream + && a.limits.connect_timeout == b.limits.connect_timeout + && a.limits.max_redirects == b.limits.max_redirects + && a.limits.allow_https_to_http_downgrade == b.limits.allow_https_to_http_downgrade + && a.limits.redirect_policy == b.limits.redirect_policy +} + /// All dependencies required by the server. -/// -/// Cheap to clone (everything is `Arc`-shared); cloned once per request via -/// the `axum` extractor so the hot path is wait-free. #[derive(Clone, Debug)] pub struct ServerBuild { /// Hot-swappable policy snapshot. diff --git a/crates/corx-server/tests/integration_proxy.rs b/crates/corx-server/tests/integration_proxy.rs index 256c190..bb6263a 100644 --- a/crates/corx-server/tests/integration_proxy.rs +++ b/crates/corx-server/tests/integration_proxy.rs @@ -100,6 +100,9 @@ fn make_stack(mutator: impl FnOnce(&mut Config)) -> (axum::Router, MockServer) { // Circuit breaker would trip under tight unit failure bursts; keep on // but with a high threshold so happy-path tests are not flaky. config.circuit_breaker.failure_threshold = 10_000; + // Isolation: disable GCRA so concurrent tests do not share process + // budgets (limiter state is per ServerBuild, but keep headroom clear). + config.rate_limit.enabled = false; mutator(&mut config); @@ -358,6 +361,166 @@ async fn bearer_auth_accepts_valid_token() { assert_eq!(response.status(), StatusCode::OK); } +#[tokio::test] +async fn redirect_follow_rejects_hop_outside_allowlist() { + use corx_core::config::TargetMode; + // Two-phase setup so the allowlist can include the mock's host. + ensure_crypto_provider(); + let mock = MockServer::start().await; + let host = mock_host(&mock); + + let mut config = Config::default(); + config.ssrf.mode = SsrfMode::Strict; + config + .ssrf + .extra_allowed_cidrs + .push("127.0.0.0/8".parse().unwrap()); + config.cors.allow_any_origin = true; + config.circuit_breaker.failure_threshold = 10_000; + config.rate_limit.enabled = false; + config.target.mode = TargetMode::Allowlist; + config.target.hosts = vec![host]; + config.limits.redirect_policy = corx_core::config::RedirectPolicy::Follow; + + let metrics = MetricsHandle::for_test(); + let build = ServerBuild::from_config(config, metrics).expect("build server"); + let router = build_router(AppState::new(build)); + + Mock::given(method("GET")) + .and(path("/start")) + .respond_with( + ResponseTemplate::new(302).insert_header("location", "https://evil.example/secret"), + ) + .mount(&mock) + .await; + + let response = router + .oneshot(with_peer( + Request::builder().uri(upstream_url(&mock, "/start")), + )) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::FORBIDDEN, + "redirect hop outside allowlist must be rejected" + ); +} + +#[tokio::test] +async fn redirect_follow_admits_hop_on_allowlist() { + use corx_core::config::TargetMode; + // Two mock servers on loopback — both allowlisted by suffix. + let mock_b = futures::executor::block_on(MockServer::start()); + let (router, mock_a) = make_stack(|cfg| { + cfg.target.mode = TargetMode::Allowlist; + cfg.target.hosts = vec!["127.0.0.1".into()]; + cfg.limits.redirect_policy = corx_core::config::RedirectPolicy::Follow; + cfg.ssrf + .extra_allowed_cidrs + .push("127.0.0.0/8".parse().unwrap()); + }); + + Mock::given(method("GET")) + .and(path("/final")) + .respond_with(ResponseTemplate::new(200).set_body_string("landed")) + .mount(&mock_b) + .await; + + let location = format!("{}/final", mock_b.uri().trim_end_matches('/')); + Mock::given(method("GET")) + .and(path("/start")) + .respond_with(ResponseTemplate::new(302).insert_header("location", location.as_str())) + .mount(&mock_a) + .await; + + let response = router + .oneshot(with_peer( + Request::builder().uri(upstream_url(&mock_a, "/start")), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = response.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(&body[..], b"landed"); +} + +fn mock_host(mock: &MockServer) -> String { + // mock.uri() is like http://127.0.0.1:PORT + let uri = mock.uri(); + let without_scheme = uri + .strip_prefix("http://") + .or_else(|| uri.strip_prefix("https://")) + .unwrap_or(&uri); + without_scheme + .split(':') + .next() + .unwrap_or(without_scheme) + .to_owned() +} + +#[tokio::test] +async fn error_payload_is_client_safe() { + let (router, _mock) = make_stack(|cfg| { + cfg.ssrf.extra_allowed_cidrs.clear(); + }); + let response = router + .oneshot(with_peer( + Request::builder().uri("/http://localhost/private"), + )) + .await + .unwrap(); + // Forbidden (SSRF) or bad gateway depending on resolution path. + assert!( + matches!( + response.status(), + StatusCode::FORBIDDEN | StatusCode::BAD_GATEWAY + ), + "unexpected status {}", + response.status() + ); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let text = String::from_utf8_lossy(&body); + // Client payload must not echo raw resolver / OS error strings. + assert!( + !text.contains("os error") && !text.contains("ConnectError"), + "leaked internal detail: {text}" + ); +} + +#[tokio::test] +async fn redirect_rewrite_rewrites_location_to_proxy_path() { + let (router, mock) = make_stack(|cfg| { + cfg.limits.redirect_policy = corx_core::config::RedirectPolicy::Rewrite; + }); + Mock::given(method("GET")) + .and(path("/bounce")) + .respond_with( + ResponseTemplate::new(302) + .insert_header("location", "https://next.example/path?q=1"), + ) + .mount(&mock) + .await; + + let response = router + .oneshot(with_peer( + Request::builder().uri(upstream_url(&mock, "/bounce")), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::FOUND); + let location = response + .headers() + .get(header::LOCATION) + .expect("Location") + .to_str() + .unwrap(); + assert_eq!( + location, "/https://next.example/path?q=1", + "rewrite policy must prefix absolute Location with /" + ); +} + #[tokio::test] async fn proxy_injects_forwarded_and_request_id_headers() { let (router, mock) = make_stack(|_| {}); diff --git a/crates/corx/src/lib.rs b/crates/corx/src/lib.rs index a474658..8160ee8 100644 --- a/crates/corx/src/lib.rs +++ b/crates/corx/src/lib.rs @@ -47,7 +47,7 @@ pub use corx_core::config::{ TargetConfig, TargetMode, TlsConfig, UpstreamConfig, ValidationReport, }; pub use corx_core::error::{ErrorKind, ErrorPayload, ProxyError, STATUS_HEADER}; -pub use corx_core::policy::{CircuitBreaker, CircuitDecision, TargetPolicy}; +pub use corx_core::policy::{CircuitBreaker, CircuitHop, TargetPolicy}; pub use corx_core::proxy::{ ClientConfig, CorsPolicy, HeaderFilter, SsrfGuard, TargetUrl, Upstream, UpstreamBody, apply_to_response, build_preflight_response, extract_target, is_preflight, diff --git a/docs/architecture.md b/docs/architecture.md index 6c3747f..1e66e5d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -39,20 +39,26 @@ client request → access_log_layer (outermost; sees final status) → TimeoutLayer → DefaultBodyLimit + → cors_layer (stamps ACAO on every response, incl. 504/413/errors) + → TimeoutLayer + → DefaultBodyLimit + → CatchPanicLayer → header_limit_layer (max_request_header_bytes → 431) - → load_shed_layer (global inflight ceiling) - → cors_layer (stamps headers on every response) + → load_shed_layer (limits.inflight_max) + → auth_layer (optional bearer) → proxy fallback handler: ├── policies = ServerBuild.policies.load() // ArcSwap snapshot ├── if preflight: │ ├── (default) origin guard + optional rate limit │ └── build_preflight_response → return ├── policies.guard.check_origin - ├── extract_target // URL parser + IDN punycode + scheme normaliser + ├── extract_target + target_policy.check // first hop ├── policies.guard.check_rate ├── inject Forwarded / X-Forwarded-* / X-Request-Id - ├── policies.upstream.execute // hyper-rustls + GuardedResolver - └── apply_cors + via header → return + ├── upstream.execute(circuit): + │ └── each hop: target_policy + circuit + SSRF resolver + └── shape_response (via, optional Location rewrite, body limit) + // CORS applied by cors_layer only ``` The whole chain is wait-free: every middleware reads from `Arc`-shared @@ -63,11 +69,16 @@ state or `ArcSwap` snapshots, never holds a lock. `ServerBuild` splits its state into: - `policies: Arc>` - Atomically replaceable on `SIGHUP`; carries `cors`, `request_filter`, - `response_filter`, `guard`, `upstream`, and the source `Config`. + Atomically replaceable on `SIGHUP`. Pure policy fields (`cors`, header + filters, origin policy, `target_policy`, source `Config`) always rebuild. + **Process state is retained when the matching config section is + unchanged:** `circuit`, GCRA `RateLimiter` maps, and the `Upstream` + connection pool (unless ssrf/target/upstream/redirect/connect knobs + change). - `immutable_server` / `immutable_limits` / `immutable_metrics_endpoint` Compared against incoming reloads; mismatches are rejected and the - previous snapshot stays active. + previous snapshot stays active. This includes `inflight_max` and + `max_response_body_bytes`. Every handler grabs exactly one snapshot via `state.build.policies()` so the policy view is internally consistent for the duration of the request. @@ -79,6 +90,10 @@ wrapping `SsrfGuard`. The guard returns *every* admissible address from a single DNS lookup so happy-eyeballs IPv4/IPv6 fallback works naturally while still blocking each candidate against the policy CIDRs. +`TargetPolicy` and the per-host `CircuitBreaker` run on **every** hop +inside `Upstream::execute` (initial request and each redirect continue), +so allowlists cannot be bypassed via 3xx. + ## Errors `ProxyError` (in `corx-core`) is the canonical error type. Every variant diff --git a/docs/configuration.md b/docs/configuration.md index cd91e29..d3b9ff8 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -17,12 +17,14 @@ that the binary is actually using. | ------------------------ | -------------------------------------------------- | | `[server]` | Listener address, HTTP/2 toggle, graceful shutdown | | `[server.tls]` | Optional inbound TLS / mTLS material | -| `[limits]` | Body size, header size, timeouts, redirects | +| `[limits]` | Body/header/response size, inflight, timeouts, redirects | | `[cors]` | Allow-origin policy, methods, headers, credentials | -| `[security]` | Required headers, origin allow/block lists | +| `[security]` | Required headers, origin allow/block, auth | +| `[target]` | Host/scheme admission (every hop including redirects) | +| `[circuit_breaker]` | Per-host circuit breaker | | `[ssrf]` | DNS-aware SSRF guard | | `[forwarded]` | RFC 7239 / X-Forwarded-* / X-Request-Id | -| `[rate_limit.*]` | Multi-dimensional rate limit | +| `[rate_limit.*]` | Multi-dimensional GCRA rate limit | | `[upstream]` | HTTP client tuning | | `[observability]` | Logging, metrics, OTLP traces | @@ -31,9 +33,16 @@ that the binary is actually using. A `SIGHUP` triggers a configuration reload. Fields fall into two camps: - **Hot-swappable** (replaced atomically via `arc-swap`): `cors`, - `security`, `ssrf`, `forwarded`, `rate_limit`, `upstream`, `observability.otel`. + `security`, `target`, `ssrf`, `forwarded`, `rate_limit`, + `circuit_breaker`, `upstream`, and related policy. When + `rate_limit` / `circuit_breaker` sections are **unchanged**, in-memory + GCRA buckets and open circuits are **retained** across reload. The + upstream connection pool is retained unless ssrf/target/upstream/ + connect/redirect knobs change. - **Immutable** (require a process restart): `server.bind`, `server.http2`, - `server.tls`, `limits.max_request_body_bytes`, `limits.request_timeout`, + `server.tls`, `limits.max_request_body_bytes`, + `limits.max_request_header_bytes`, `limits.max_response_body_bytes`, + `limits.inflight_max`, `limits.request_timeout`, `observability.metrics_endpoint`. Reload outcomes are reported via `corx_config_reload_total{result}` with diff --git a/docs/migration.md b/docs/migration.md index 5d8f2d3..975cd40 100644 --- a/docs/migration.md +++ b/docs/migration.md @@ -29,10 +29,10 @@ is now binary-only. omit it pick up `inject = true`, `inject_request_id = true`, `trust_inbound_xff = false`. -### Rate limit +### Rate limit and concurrency -The flat single-bucket schema was replaced with an explicit -multi-dimensional layout: +The flat single-bucket schema was replaced with multi-dimensional GCRA +plus a separate process inflight cap: ```toml # OLD (0.1) @@ -40,14 +40,19 @@ multi-dimensional layout: rps = 10 burst = 20 -# NEW (0.2) +# NEW (0.2+) +[limits] +inflight_max = 1024 # load-shed (was rate_limit.global.inflight_max) +max_response_body_bytes = 52428800 # streaming response cap; 0 = unlimited + [rate_limit] -enabled = true +enabled = true # default true +max_keys = 16384 # fail-closed cardinality for keyed dims [rate_limit.origin] rps = 50 burst = 100 -unlimited_origins = [] +unlimited_patterns = [] [rate_limit.ip] rps = 20 @@ -59,14 +64,20 @@ rps = 100 burst = 200 [rate_limit.global] -rps = 0 -burst = 0 -inflight_max = 1024 +rps = 5000 +burst = 10000 +# inflight_max removed — use limits.inflight_max ``` -Set any sub-section's `rps` to `0` to disable that dimension while +Set any sub-section's `rps` to `0` to disable that GCRA dimension while leaving the others active. Set `[rate_limit].enabled = false` to disable -every dimension at once. +every GCRA dimension at once (inflight load-shed is independent). + +### Target admission on redirects + +`[target]` allowlist / denylist / `https_only` apply on the **first hop +and every redirect hop**. Configs that relied on following 3xx to hosts +outside the allowlist will now get `403` / `target_not_allowed`. ### CORS diff --git a/docs/observability.md b/docs/observability.md index 49cbb98..ba1c9f0 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -41,7 +41,9 @@ layout shared across every duration metric. | `corx_upstream_errors_total` | counter | `kind` | | `corx_inflight_requests` | gauge | | | `corx_bytes_transferred_total` | counter | `direction = request \| response` | -| `corx_rate_limited_total` | counter | `dimension` | +| `corx_rate_limited_total` | counter | `dimension` (`origin` \| `ip` \| `target_host` \| `global` \| `inflight`) | +| `corx_circuit_opens_total` | counter | | +| `corx_circuit_rejects_total` | counter | | | `corx_ssrf_blocks_total` | counter | `cidr` | | `corx_dns_lookups_total` | counter | `result = literal \| ok \| error` | | `corx_redirect_hops` | histogram | `target_host` | diff --git a/docs/operations.md b/docs/operations.md index ee406a7..1aa3ceb 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -42,11 +42,16 @@ Three knobs to tune in production: 1. **`limits.max_request_body_bytes`** — the proxy buffers nothing, so this is mostly a denial-of-service guard. 10 MiB suits API gateways; bump to 100 MiB for upload-heavy workloads. -2. **`rate_limit.global.inflight_max`** — caps concurrent requests - process-wide and powers the load-shed layer. Set 2-3x p99 concurrency. -3. **`upstream.pool_max_idle_per_host`** — bigger means fewer TLS +2. **`limits.inflight_max`** — caps concurrent requests process-wide and + powers the load-shed layer (metric dimension `inflight`). Set + 2–3× p99 concurrency; `0` disables load-shed. +3. **`limits.max_response_body_bytes`** — streaming response size cap + (default 50 MiB; `0` = unlimited). Guards bandwidth amplification. +4. **`upstream.pool_max_idle_per_host`** — bigger means fewer TLS handshakes per upstream; right-size against the connection cap of the backend. +5. **`rate_limit.enabled` / `rate_limit.max_keys`** — GCRA dimensions and + keyed-map cardinality cap (fail-closed when full). ## Subcommands diff --git a/docs/security.md b/docs/security.md index 9dfe513..fe91397 100644 --- a/docs/security.md +++ b/docs/security.md @@ -47,7 +47,8 @@ classic cors-anywhere behaviour (preflight before guards). ## Target admission -`[target]` filters hosts and schemes **before** DNS/SSRF: +`[target]` filters hosts and schemes **before** DNS/SSRF on the first hop, +and again on **every redirect hop** before connect: - `any_public` (default) — any host; SSRF still applies to resolved IPs - `allowlist` / `denylist` — exact hosts or DNS suffixes (`.example.com`) @@ -108,7 +109,9 @@ is the one Prometheus attributes the rejection to via - `origin` (per `Origin` header) - `ip` (per remote address; CIDRs in `trusted_cidrs` are exempted) - `target_host` (per validated upstream host) -- `global` (process-wide), backed by an inflight gauge for load shed +- `global` (process-wide GCRA) +- process **inflight** concurrency lives under `limits.inflight_max` + (metric dimension `inflight`, independent of GCRA `enabled`) ## TLS / mTLS / FIPS From e9c8ff90e9c789876c468bc710d71f21ca6878b8 Mon Sep 17 00:00:00 2001 From: "x.qntx.eth" Date: Sun, 9 Aug 2026 14:59:59 +0800 Subject: [PATCH 2/3] docs: add AGENTS.md with development principles and guidelines --- AGENTS.md | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100644 AGENTS.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..4dbba1b --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,9 @@ +# AGENTS.md + +- Do not preserve backward compatibility. Remove obsolete paths instead of adding compatibility layers, fallbacks, or migrations. +- Choose the simplest implementation that fully meets the current requirements. Avoid speculative abstractions, configuration, and indirection. +- Grow the system in layers. Start from the smallest version that works end to end, and add each new capability on top of a product that already works. Never trade a working product for unfinished complexity. +- Keep components modular and concerns clearly separated. +- Prefer established, well-maintained libraries when they reduce overall complexity or improve reliability. Do not reimplement common functionality without a clear reason. +- Lean on the dependencies already in the project before writing your own implementation or adding packages. Do not assume a library lacks a capability without checking its documentation and types. +- Make architectural decisions for the long term. Do not accept a stopgap that only works for now and is meant to be replaced later. From 1657fd7c89870ae38909b6dfea44119669a0bacb Mon Sep 17 00:00:00 2001 From: "x.qntx.eth" Date: Sun, 9 Aug 2026 15:00:52 +0800 Subject: [PATCH 3/3] style: apply rustfmt for CI fmt check --- crates/corx-core/src/config/validate.rs | 6 ++---- crates/corx-core/src/policy/circuit.rs | 9 +++++++-- crates/corx-core/src/policy/target.rs | 12 ++++++------ crates/corx-core/src/proxy/headers.rs | 6 +++--- crates/corx-core/src/proxy/upstream.rs | 3 ++- crates/corx-server/src/handlers/proxy.rs | 5 +---- crates/corx-server/tests/integration_proxy.rs | 3 +-- 7 files changed, 22 insertions(+), 22 deletions(-) diff --git a/crates/corx-core/src/config/validate.rs b/crates/corx-core/src/config/validate.rs index e195c7e..a496b52 100644 --- a/crates/corx-core/src/config/validate.rs +++ b/crates/corx-core/src/config/validate.rs @@ -225,10 +225,8 @@ fn validate_rate_limit(cfg: &RateLimitConfig, report: &mut ValidationReport) { )); } - let any_enabled = cfg.origin.rps > 0 - || cfg.ip.rps > 0 - || cfg.target_host.rps > 0 - || cfg.global.rps > 0; + let any_enabled = + cfg.origin.rps > 0 || cfg.ip.rps > 0 || cfg.target_host.rps > 0 || cfg.global.rps > 0; if !any_enabled { report.errors.push(ConfigError::new( "rate_limit", diff --git a/crates/corx-core/src/policy/circuit.rs b/crates/corx-core/src/policy/circuit.rs index 4f8dd8e..40c0961 100644 --- a/crates/corx-core/src/policy/circuit.rs +++ b/crates/corx-core/src/policy/circuit.rs @@ -13,9 +13,14 @@ use crate::observability; #[derive(Debug, Clone, Copy)] enum State { Closed, - Open { until: Instant }, + Open { + until: Instant, + }, /// `since` bounds half-open so cancelled probes cannot lock a host forever. - HalfOpen { probes: u32, since: Instant }, + HalfOpen { + probes: u32, + since: Instant, + }, } struct HostCircuit { diff --git a/crates/corx-core/src/policy/target.rs b/crates/corx-core/src/policy/target.rs index 2ecbe9f..f5b9fad 100644 --- a/crates/corx-core/src/policy/target.rs +++ b/crates/corx-core/src/policy/target.rs @@ -117,12 +117,12 @@ impl TargetPolicy { /// Returns [`ProxyError::TargetNotAllowed`] or [`ProxyError::InvalidUrl`] /// when the URI lacks a usable scheme/host or fails policy. pub fn check_uri(&self, uri: &Uri) -> Result<(), ProxyError> { - let scheme = uri.scheme_str().ok_or_else(|| { - ProxyError::InvalidUrl("hop URI lacks a scheme".to_owned()) - })?; - let host = uri.host().ok_or_else(|| { - ProxyError::InvalidUrl("hop URI lacks a host".to_owned()) - })?; + let scheme = uri + .scheme_str() + .ok_or_else(|| ProxyError::InvalidUrl("hop URI lacks a scheme".to_owned()))?; + let host = uri + .host() + .ok_or_else(|| ProxyError::InvalidUrl("hop URI lacks a host".to_owned()))?; self.check_authority(scheme, host) } } diff --git a/crates/corx-core/src/proxy/headers.rs b/crates/corx-core/src/proxy/headers.rs index 2404ca9..0c6b10f 100644 --- a/crates/corx-core/src/proxy/headers.rs +++ b/crates/corx-core/src/proxy/headers.rs @@ -49,9 +49,9 @@ impl HeaderFilter { pub fn try_new(extra_deny: &[String]) -> Result { let mut names = Vec::with_capacity(extra_deny.len()); for raw in extra_deny { - let name = raw.parse::().map_err(|err| { - format!("invalid header name `{raw}`: {err}") - })?; + let name = raw + .parse::() + .map_err(|err| format!("invalid header name `{raw}`: {err}"))?; names.push(name); } Ok(Self { extra_deny: names }) diff --git a/crates/corx-core/src/proxy/upstream.rs b/crates/corx-core/src/proxy/upstream.rs index 9a3e952..fd5e368 100644 --- a/crates/corx-core/src/proxy/upstream.rs +++ b/crates/corx-core/src/proxy/upstream.rs @@ -235,7 +235,8 @@ impl Upstream { } fn host_of(uri: &http::Uri) -> String { - uri.host().map_or_else(|| "unknown".to_owned(), str::to_owned) + uri.host() + .map_or_else(|| "unknown".to_owned(), str::to_owned) } fn settle_terminal(hop: CircuitHop<'_>, circuit: &CircuitBreaker, status: http::StatusCode) { diff --git a/crates/corx-server/src/handlers/proxy.rs b/crates/corx-server/src/handlers/proxy.rs index 59dd8ea..0cb6e62 100644 --- a/crates/corx-server/src/handlers/proxy.rs +++ b/crates/corx-server/src/handlers/proxy.rs @@ -156,10 +156,7 @@ async fn execute_proxy( let outbound = Request::from_parts(parts, axum_to_upstream_body(body)); let upstream_started = Instant::now(); - let upstream_response = policies - .upstream - .execute(outbound, &policies.circuit) - .await; + let upstream_response = policies.upstream.execute(outbound, &policies.circuit).await; let upstream_elapsed = upstream_started.elapsed().as_secs_f64(); let response = match upstream_response { diff --git a/crates/corx-server/tests/integration_proxy.rs b/crates/corx-server/tests/integration_proxy.rs index bb6263a..e7f3a6e 100644 --- a/crates/corx-server/tests/integration_proxy.rs +++ b/crates/corx-server/tests/integration_proxy.rs @@ -496,8 +496,7 @@ async fn redirect_rewrite_rewrites_location_to_proxy_path() { Mock::given(method("GET")) .and(path("/bounce")) .respond_with( - ResponseTemplate::new(302) - .insert_header("location", "https://next.example/path?q=1"), + ResponseTemplate::new(302).insert_header("location", "https://next.example/path?q=1"), ) .mount(&mock) .await;