diff --git a/Cargo.lock b/Cargo.lock index b836d650e..e9cc1ee9f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1808,9 +1808,11 @@ dependencies = [ "extenddb-storage", "futures", "hyper", + "hyper-util", "libc", "metrics", "rand 0.9.4", + "rcgen", "rustls", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 97e64e9e0..aef0fd967 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -59,8 +59,9 @@ futures = "0.3" # HTTP axum = { version = "0.8", features = ["macros"] } tower = { version = "0.5", features = ["util"] } -tower-http = { version = "0.6", features = ["compression-gzip", "cors", "set-header"] } +tower-http = { version = "0.6", features = ["compression-gzip", "cors", "set-header", "timeout"] } hyper = { version = "1" } +hyper-util = { version = "0.1", features = ["tokio"] } # URI parsing for the backup store's endpoint check; the same major hyper and # the AWS SDK already use. http = { version = "1" } diff --git a/crates/app/src/init_helpers.rs b/crates/app/src/init_helpers.rs index 8be891e82..c95992234 100755 --- a/crates/app/src/init_helpers.rs +++ b/crates/app/src/init_helpers.rs @@ -247,9 +247,10 @@ backend = "{backend}" [server] bind_addr = "{bind_addr}" # port = 18443 # HTTPS port -# region = "us-east-1" # AWS region for ARN generation +# region = "us-east-1" # AWS region for ARN generation; requests must be signed for this region # run_dir = "{run_dir}" # Directory for PID file # throttling_enabled = false # Enable provisioned throughput throttling +# request_timeout_secs = 30 # Close a connection whose headers or body stall this long (1 to 86400) [server.tls] # TLS is mandatory. The server refuses to start with enabled = false. diff --git a/crates/auth/src/lib.rs b/crates/auth/src/lib.rs old mode 100755 new mode 100644 index 7d6a0bb63..8904b56f0 --- a/crates/auth/src/lib.rs +++ b/crates/auth/src/lib.rs @@ -104,18 +104,26 @@ pub trait CredentialStore: Send + Sync { /// `SigV4` auth provider with local credential store. /// -/// Parses the `Authorization` header, looks up the access key, decrypts the -/// secret, verifies the `SigV4` signature, and validates the request timestamp. +/// Parses the `Authorization` header, validates the request timestamp, checks +/// the credential scope, looks up the access key, decrypts the secret and +/// verifies the `SigV4` signature, in that order (the service's order). /// Handles both long-lived (AKIA*) and temporary (ASIA* + X-Amz-Security-Token) -/// credentials. +/// credentials. The credential scope must name this server's region. pub struct BuiltinAuthProvider { credential_store: C, + region: String, } impl BuiltinAuthProvider { - /// Create a new `BuiltinAuthProvider` with the given credential store. - pub fn new(credential_store: C) -> Self { - Self { credential_store } + /// Create a new `BuiltinAuthProvider` over `credential_store` for a server + /// deployed as `region`. A request whose credential scope names another + /// region is rejected, as the service rejects a scope for a region other + /// than the endpoint's. + pub fn new(credential_store: C, region: impl Into) -> Self { + Self { + credential_store, + region: region.into(), + } } } @@ -137,8 +145,22 @@ impl AuthProvider for BuiltinAuthProvider { // Parse Authorization header let parsed = sigv4::parse::parse_authorization(auth_header)?; - // Look up credential before timestamp validation — real DynamoDB returns - // an invalid-key error even when the timestamp is also expired. + // The service checks the request timestamp, then the credential + // scope, then looks the key up (measured 2026-10-23: an unknown key + // with a stale date answers the timestamp error, an unknown key with + // a wrong region answers the scope message, and a stale date with a + // wrong region answers the timestamp error). Neither check reads the + // stored credential, so running them first leaves the S-5 timing + // argument below unchanged. + // + // Validate timestamp (±15 minute window). + sigv4::verify::validate_timestamp(headers)?; + + // The credential scope must name this server's region and the + // dynamodb service; the service reports both complaints in one + // message when both are wrong. + sigv4::verify::check_scope(&parsed, &self.region)?; + let credential = self .credential_store .lookup_credential(&parsed.access_key_id) @@ -150,15 +172,14 @@ impl AuthProvider for BuiltinAuthProvider { })?; // S-5: Track inactive status but do NOT return early. Continue through - // timestamp validation and signature verification to ensure constant-time - // failure paths — no timing difference between inactive, invalid, and - // absent keys. + // signature verification to ensure constant-time failure paths — no + // timing difference between inactive, invalid, and absent keys. let is_inactive = !credential.is_active; - // Validate timestamp (±15 minute window) after credential lookup. - sigv4::verify::validate_timestamp(headers)?; - - // For session credentials, verify X-Amz-Security-Token matches the stored token. + // Session tokens. For a session credential the presented token must + // match the stored one. For a long-term key no token may be presented: + // the service validates any presented token and a long-term key has + // none, so the request fails as an invalid token. // CB-12: Session expiration is enforced at the credential store layer (fail-closed). // Expired sessions are never returned — the storage layer returns // ExpiredTokenException on the cache-miss path, and CachedCredentialStore @@ -167,15 +188,19 @@ impl AuthProvider for BuiltinAuthProvider { // S-5: Compare the session token in constant time and defer the // rejection until after signature verification, so neither the token's // byte contents nor its presence changes the failure-path timing. + // A header value that is not visible ASCII cannot be compared with + // the stored token, and it is still a token the client presented, so + // presence is read from the header map and the value from its text. + let token_header_present = headers.contains_key("x-amz-security-token"); + let presented_token = headers + .get("x-amz-security-token") + .and_then(|v| v.to_str().ok()); let session_token_ok = if credential.is_session { - let token = headers - .get("x-amz-security-token") - .and_then(|v| v.to_str().ok()) - .unwrap_or(""); + let token = presented_token.unwrap_or(""); let expected = credential.session_token.as_deref().unwrap_or(""); sigv4::verify::constant_time_eq(token.as_bytes(), expected.as_bytes()) } else { - true + !token_header_present }; // Verify SigV4 signature — always, even for inactive keys (S-5). @@ -191,11 +216,19 @@ impl AuthProvider for BuiltinAuthProvider { // S-5: Reject inactive keys or a mismatched session token only after // full signature verification, to prevent timing side-channels. - if is_inactive || !session_token_ok { + // The service's message for a presented token that does not resolve + // has no trailing period; its message for an unknown key or a missing + // token does (both measured 2026-09-16). + if is_inactive || (credential.is_session && presented_token.is_none()) { return Err(DynamoDbError::UnrecognizedClientException( "The security token included in the request is invalid.".to_owned(), )); } + if !session_token_ok { + return Err(DynamoDbError::UnrecognizedClientException( + "The security token included in the request is invalid".to_owned(), + )); + } // Build identity from credential (clone fields because ZeroizeOnDrop // prevents moving out of the struct). @@ -215,112 +248,4 @@ impl AuthProvider for BuiltinAuthProvider { } #[cfg(test)] -mod tests { - use super::*; - use axum::http::{HeaderMap, HeaderValue}; - - /// In-memory credential store for unit tests. - struct MockCredentialStore { - /// `Some(credential)` for found credentials, `None` for not found. - credential: Option, - /// If set, `lookup_credential` returns this error instead. - error: Option<&'static str>, - } - - #[async_trait::async_trait] - impl CredentialStore for MockCredentialStore { - async fn lookup_credential( - &self, - _access_key_id: &str, - ) -> Result, DynamoDbError> { - if let Some(msg) = self.error { - return Err(DynamoDbError::ExpiredTokenException(msg.to_owned())); - } - Ok(self.credential.clone()) - } - } - - fn make_headers_with_auth(access_key: &str, token: Option<&str>) -> HeaderMap { - let mut headers = HeaderMap::new(); - // Use current time for the date to pass timestamp validation. - let now = time::OffsetDateTime::now_utc(); - let date_str = format!( - "{:04}{:02}{:02}T{:02}{:02}{:02}Z", - now.year(), - now.month() as u8, - now.day(), - now.hour(), - now.minute(), - now.second(), - ); - let date_short = &date_str[..8]; - // Signature is fake — these tests exercise error paths (expired token, - // unknown key) that trigger before signature verification in - // authenticate(). - let auth = format!( - "AWS4-HMAC-SHA256 Credential={}/{}/us-east-1/dynamodb/aws4_request, \ - SignedHeaders=content-type;host;x-amz-date;x-amz-target, \ - Signature=0000000000000000000000000000000000000000000000000000000000000000", - access_key, date_short - ); - headers.insert("authorization", HeaderValue::from_str(&auth).unwrap()); - headers.insert("x-amz-date", HeaderValue::from_str(&date_str).unwrap()); - headers.insert("host", HeaderValue::from_static("localhost:18443")); - headers.insert( - "content-type", - HeaderValue::from_static("application/x-amz-json-1.0"), - ); - headers.insert( - "x-amz-target", - HeaderValue::from_static("DynamoDB_20120810.ListTables"), - ); - if let Some(t) = token { - headers.insert("x-amz-security-token", HeaderValue::from_str(t).unwrap()); - } - headers - } - - #[tokio::test] - async fn expired_session_returns_expired_token_exception() { - // CB-12: The credential store returns ExpiredTokenException directly - // for expired sessions (fail-closed). The auth layer propagates it. - let store = MockCredentialStore { - credential: None, - error: Some("The security token included in the request is expired"), - }; - let provider = BuiltinAuthProvider::new(store); - let headers = make_headers_with_auth("ASIAEXTENDDB00000000", Some("test-token-value")); - - let result = provider.authenticate(&headers, b"{}").await; - match result { - Err(DynamoDbError::ExpiredTokenException(msg)) => { - assert!( - msg.contains("expired"), - "Expected 'expired' in message: {msg}" - ); - } - other => panic!("Expected ExpiredTokenException, got: {other:?}"), - } - } - - #[tokio::test] - async fn nonexistent_key_returns_unrecognized_client() { - let store = MockCredentialStore { - credential: None, - error: None, - }; - let provider = BuiltinAuthProvider::new(store); - let headers = make_headers_with_auth("AKIAXXXXXXXXXXXXXXXX", None); - - let result = provider.authenticate(&headers, b"{}").await; - match result { - Err(DynamoDbError::UnrecognizedClientException(msg)) => { - assert!( - msg.contains("invalid"), - "Expected 'invalid' in message: {msg}" - ); - } - other => panic!("Expected UnrecognizedClientException, got: {other:?}"), - } - } -} +mod tests; diff --git a/crates/auth/src/sigv4/canonical.rs b/crates/auth/src/sigv4/canonical.rs index c46cb6812..e46847fd6 100755 --- a/crates/auth/src/sigv4/canonical.rs +++ b/crates/auth/src/sigv4/canonical.rs @@ -39,8 +39,9 @@ pub fn canonical_request( // trusting the header would let the sender choose the hash that the // signature is checked against. When the client includes the header in // SignedHeaders it is covered as an ordinary header by - // build_canonical_headers. UNSIGNED-PAYLOAD is rejected by - // verify_signature() before this function is called. + // build_canonical_headers. A client that signed a literal such as + // UNSIGNED-PAYLOAD in place of the hash fails as a plain signature + // mismatch, which is what the service returns. let payload_hash = sha256_hex(body); format!( @@ -65,10 +66,11 @@ pub fn string_to_sign(timestamp: &str, scope: &str, canonical_request: &str) -> /// Build canonical headers from the signed header list. /// -/// Per the `SigV4` spec, header names are lowercased, values are trimmed, -/// and headers are sorted alphabetically. Each line ends with `\n`. -/// CB-7: Header names are explicitly lowercased to handle clients that -/// send mixed-case `SignedHeaders` values. +/// Per the `SigV4` spec, header names are lowercased, values are trimmed with +/// internal whitespace collapsed, and a header that appears more than once +/// contributes its values joined by commas in the order they were sent. +/// Each line ends with `\n`. CB-7: Header names are explicitly lowercased to +/// handle clients that send mixed-case `SignedHeaders` values. fn build_canonical_headers(headers: &HeaderMap, signed_headers: &str) -> String { let mut result = String::new(); // signed_headers is already sorted and semicolon-delimited. @@ -77,27 +79,32 @@ fn build_canonical_headers(headers: &HeaderMap, signed_headers: &str) -> String // does not require pre-lowercased input. for name in signed_headers.split(';') { let lower = name.to_ascii_lowercase(); - let value = headers - .get(lower.as_str()) - .and_then(|v| v.to_str().ok()) - .unwrap_or(""); + let mut value = headers + .get_all(lower.as_str()) + .iter() + // A value that is not visible ASCII cannot have been signed as + // anything; it contributes an empty value rather than disappearing, + // so a request carrying such a header never canonicalizes the same + // as one without it. + .map(|v| v.to_str().unwrap_or("")) + // Trim leading/trailing whitespace, collapse internal whitespace + .map(|v| v.split_whitespace().collect::>().join(" ")) + .collect::>() + .join(","); // HTTP/2 uses :authority instead of Host. If the host header is empty // or missing, fall back to the :authority pseudo-header value. // Defense-in-depth: handler.rs injects Host from URI authority for // HTTP/2 requests, so this fallback should rarely activate. - let value = if lower == "host" && value.is_empty() { - headers + if lower == "host" && value.is_empty() { + value = headers .get(":authority") .and_then(|v| v.to_str().ok()) - .unwrap_or(value) - } else { - value - }; - // Trim leading/trailing whitespace, collapse internal whitespace - let trimmed = value.split_whitespace().collect::>().join(" "); + .map(|v| v.split_whitespace().collect::>().join(" ")) + .unwrap_or_default(); + } result.push_str(&lower); result.push(':'); - result.push_str(&trimmed); + result.push_str(&value); result.push('\n'); } result @@ -202,4 +209,46 @@ mod tests { assert_eq!(lines[8], sha256_hex(received_body)); assert_ne!(lines[8], sha256_hex(signed_body)); } + + /// A header sent more than once contributes its values joined by commas, + /// which is how the service canonicalizes it. + #[test] + fn repeated_header_values_are_comma_joined_in_order() { + let mut headers = HeaderMap::new(); + headers.insert("host", "localhost:18443".parse().unwrap()); + headers.insert("x-amz-date", "20260415T120000Z".parse().unwrap()); + headers.append("x-amz-meta-dup", " a ".parse().unwrap()); + headers.append("x-amz-meta-dup", "b c".parse().unwrap()); + + let creq = canonical_request( + "POST", + "/", + "", + &headers, + "host;x-amz-date;x-amz-meta-dup", + b"{}", + ); + + let lines: Vec<&str> = creq.split('\n').collect(); + assert_eq!(lines[5], "x-amz-meta-dup:a,b c"); + } + + /// A duplicate value that is not visible ASCII cannot be signed as anything. + /// It must still change the canonical form, so a request carrying it does not + /// verify under a signature made without it. + #[test] + fn unsignable_duplicate_value_changes_the_canonical_form() { + let mut plain = HeaderMap::new(); + plain.insert("host", "localhost:18443".parse().unwrap()); + plain.insert("x-amz-meta-dup", "a".parse().unwrap()); + let mut with_extra = plain.clone(); + with_extra.append( + "x-amz-meta-dup", + axum::http::HeaderValue::from_bytes(b"\xff\xfe").unwrap(), + ); + let a = canonical_request("POST", "/", "", &plain, "host;x-amz-meta-dup", b"{}"); + let b = canonical_request("POST", "/", "", &with_extra, "host;x-amz-meta-dup", b"{}"); + assert_ne!(a, b); + assert!(b.contains("x-amz-meta-dup:a,\n"), "{b}"); + } } diff --git a/crates/auth/src/sigv4/parse.rs b/crates/auth/src/sigv4/parse.rs index 7dc969046..05152667f 100755 --- a/crates/auth/src/sigv4/parse.rs +++ b/crates/auth/src/sigv4/parse.rs @@ -73,9 +73,10 @@ pub fn parse_authorization(header: &str) -> Result = credential.splitn(5, '/').collect(); if parts.len() != 5 || parts[4] != "aws4_request" { - return Err(incomplete( - "Credential must follow the format: AKID/date/region/service/aws4_request", - )); + // Measured 2026-09-16: the service names the rule and echoes the value. + return Err(incomplete(&format!( + "Credential must have exactly 5 slash-delimited elements, e.g. keyid/date/region/service/term, got '{credential}'" + ))); } Ok(ParsedAuthorization { diff --git a/crates/auth/src/sigv4/verify.rs b/crates/auth/src/sigv4/verify.rs index 3444d9b3a..0f3f656be 100755 --- a/crates/auth/src/sigv4/verify.rs +++ b/crates/auth/src/sigv4/verify.rs @@ -17,6 +17,36 @@ use super::signing_key; /// Maximum allowed clock skew between client and server (±15 minutes). const MAX_CLOCK_SKEW_SECS: i64 = 15 * 60; +/// Check the credential scope's region and service against this server. +/// +/// The service reports every scope complaint in one message, region first, +/// each sentence ending with a space (measured 2026-09-16: a scope wrong in +/// both gives "Credential should be scoped to a valid region. Credential +/// should be scoped to correct service: 'dynamodb'. "). The region is not +/// named in the message. +/// +/// # Errors +/// +/// Returns `InvalidSignatureException` carrying the accumulated complaints +/// when the region is not `expected_region` or the service is not `dynamodb`. +pub fn check_scope( + parsed: &ParsedAuthorization, + expected_region: &str, +) -> Result<(), DynamoDbError> { + let mut complaints = String::new(); + if parsed.region != expected_region { + complaints.push_str("Credential should be scoped to a valid region. "); + } + if parsed.service != "dynamodb" { + complaints.push_str("Credential should be scoped to correct service: 'dynamodb'. "); + } + if complaints.is_empty() { + Ok(()) + } else { + Err(DynamoDbError::InvalidSignatureException(complaints)) + } +} + /// Verify a `SigV4` signature against the request. /// /// # Arguments @@ -62,16 +92,6 @@ pub fn verify_signature( )); } - // Reject UNSIGNED-PAYLOAD: real DynamoDB requires a computed body hash - // for all API calls. UNSIGNED-PAYLOAD is only valid for S3. - if let Some(content_sha) = headers.get("x-amz-content-sha256") - && content_sha.as_bytes() == b"UNSIGNED-PAYLOAD" - { - return Err(DynamoDbError::InvalidSignatureException( - "UNSIGNED-PAYLOAD is not supported for DynamoDB operations.".to_owned(), - )); - } - // Build canonical request let creq = canonical::canonical_request( method, @@ -295,23 +315,35 @@ mod tests { } } - /// Reject UNSIGNED-PAYLOAD for DynamoDB API calls. + /// A client that signs the payload line as the `UNSIGNED-PAYLOAD` literal + /// (an S3 convention) fails with the plain mismatch, which is what the + /// service returns (measured 2026-09-16): the payload line is always the + /// hash of the received body. #[test] - fn reject_unsigned_payload() { - let parsed = make_parsed( - "dynamodb", - "20260415", - "content-type;host;x-amz-content-sha256;x-amz-date", - ); + fn unsigned_payload_literal_is_a_plain_signature_mismatch() { let mut headers = make_headers("20260415T120000Z"); headers.insert("x-amz-content-sha256", "UNSIGNED-PAYLOAD".parse().unwrap()); - let err = verify_signature(&parsed, "secret", "POST", "/", "", &headers, b"{}"); - match err { + let signed = "content-type;host;x-amz-content-sha256;x-amz-date"; + // Sign the way such a client does: the literal in place of the body hash. + let creq = canonical::canonical_request("POST", "/", "", &headers, signed, SIGNED_BODY) + .rsplit_once('\n') + .map(|(head, _)| format!("{head}\nUNSIGNED-PAYLOAD")) + .unwrap(); + let sts = canonical::string_to_sign( + "20260415T120000Z", + "20260415/us-east-1/dynamodb/aws4_request", + &creq, + ); + let key = signing_key::derive_signing_key(SECRET, "20260415", "us-east-1", "dynamodb"); + let mut parsed = make_parsed("dynamodb", "20260415", signed); + parsed.signature = signing_key::compute_signature(&key, &sts); + match verify_signature(&parsed, SECRET, "POST", "/", "", &headers, SIGNED_BODY) { Err(DynamoDbError::InvalidSignatureException(msg)) => { assert!( - msg.contains("UNSIGNED-PAYLOAD"), - "Expected 'UNSIGNED-PAYLOAD' in: {msg}" + msg.starts_with("The request signature we calculated does not match"), + "{msg}" ); + assert!(!msg.contains("UNSIGNED-PAYLOAD"), "{msg}"); } other => panic!("Expected InvalidSignatureException, got: {other:?}"), } diff --git a/crates/auth/src/tests.rs b/crates/auth/src/tests.rs new file mode 100644 index 000000000..109fa6c34 --- /dev/null +++ b/crates/auth/src/tests.rs @@ -0,0 +1,405 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Unit tests for `BuiltinAuthProvider::authenticate`. + +use super::*; +use axum::http::{HeaderMap, HeaderValue}; + +/// In-memory credential store for unit tests. +struct MockCredentialStore { + /// `Some(credential)` for found credentials, `None` for not found. + credential: Option, + /// If set, `lookup_credential` returns this error instead. + error: Option<&'static str>, +} + +#[async_trait::async_trait] +impl CredentialStore for MockCredentialStore { + async fn lookup_credential( + &self, + _access_key_id: &str, + ) -> Result, DynamoDbError> { + if let Some(msg) = self.error { + return Err(DynamoDbError::ExpiredTokenException(msg.to_owned())); + } + Ok(self.credential.clone()) + } +} + +fn make_headers_with_auth(access_key: &str, token: Option<&str>) -> HeaderMap { + let mut headers = HeaderMap::new(); + // Use current time for the date to pass timestamp validation. + let now = time::OffsetDateTime::now_utc(); + let date_str = format!( + "{:04}{:02}{:02}T{:02}{:02}{:02}Z", + now.year(), + now.month() as u8, + now.day(), + now.hour(), + now.minute(), + now.second(), + ); + let date_short = &date_str[..8]; + // Signature is fake — these tests exercise error paths (expired token, + // unknown key) that trigger before signature verification in + // authenticate(). + let auth = format!( + "AWS4-HMAC-SHA256 Credential={}/{}/us-east-1/dynamodb/aws4_request, \ + SignedHeaders=content-type;host;x-amz-date;x-amz-target, \ + Signature=0000000000000000000000000000000000000000000000000000000000000000", + access_key, date_short + ); + headers.insert("authorization", HeaderValue::from_str(&auth).unwrap()); + headers.insert("x-amz-date", HeaderValue::from_str(&date_str).unwrap()); + headers.insert("host", HeaderValue::from_static("localhost:18443")); + headers.insert( + "content-type", + HeaderValue::from_static("application/x-amz-json-1.0"), + ); + headers.insert( + "x-amz-target", + HeaderValue::from_static("DynamoDB_20120810.ListTables"), + ); + if let Some(t) = token { + headers.insert("x-amz-security-token", HeaderValue::from_str(t).unwrap()); + } + headers +} + +#[tokio::test] +async fn expired_session_returns_expired_token_exception() { + // CB-12: The credential store returns ExpiredTokenException directly + // for expired sessions (fail-closed). The auth layer propagates it. + let store = MockCredentialStore { + credential: None, + error: Some("The security token included in the request is expired"), + }; + let provider = BuiltinAuthProvider::new(store, "us-east-1"); + let headers = make_headers_with_auth("ASIAEXTENDDB00000000", Some("test-token-value")); + + let result = provider.authenticate(&headers, b"{}").await; + match result { + Err(DynamoDbError::ExpiredTokenException(msg)) => { + assert!( + msg.contains("expired"), + "Expected 'expired' in message: {msg}" + ); + } + other => panic!("Expected ExpiredTokenException, got: {other:?}"), + } +} + +#[tokio::test] +async fn nonexistent_key_returns_unrecognized_client() { + let store = MockCredentialStore { + credential: None, + error: None, + }; + let provider = BuiltinAuthProvider::new(store, "us-east-1"); + let headers = make_headers_with_auth("AKIAXXXXXXXXXXXXXXXX", None); + + let result = provider.authenticate(&headers, b"{}").await; + match result { + Err(DynamoDbError::UnrecognizedClientException(msg)) => { + assert!( + msg.contains("invalid"), + "Expected 'invalid' in message: {msg}" + ); + } + other => panic!("Expected UnrecognizedClientException, got: {other:?}"), + } +} + +const SECRET: &str = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"; + +fn credential(is_session: bool) -> StoredCredential { + StoredCredential { + secret_key: SECRET.to_owned(), + account_id: "123456789012".to_owned(), + principal_name: "alice".to_owned(), + session_name: is_session.then(|| "sess".to_owned()), + is_session, + session_token: is_session.then(|| "stored-token".to_owned()), + is_active: true, + expires_at: None, + } +} + +/// Headers for a request correctly signed by a client whose credential +/// scope names `region`. `token`, when present, is sent and signed. +fn signed_headers(access_key: &str, region: &str, token: Option<&str>, body: &[u8]) -> HeaderMap { + let mut headers = make_headers_with_auth(access_key, token); + headers.remove("authorization"); + let date_str = headers["x-amz-date"].to_str().unwrap().to_owned(); + let date_short = date_str[..8].to_owned(); + let mut names = vec!["content-type", "host", "x-amz-date", "x-amz-target"]; + if token.is_some() { + names.push("x-amz-security-token"); + } + let signed = names.join(";"); + let creq = sigv4::canonical::canonical_request("POST", "/", "", &headers, &signed, body); + let scope = format!("{date_short}/{region}/dynamodb/aws4_request"); + let sts = sigv4::canonical::string_to_sign(&date_str, &scope, &creq); + let key = sigv4::signing_key::derive_signing_key(SECRET, &date_short, region, "dynamodb"); + let sig = sigv4::signing_key::compute_signature(&key, &sts); + let auth = format!( + "AWS4-HMAC-SHA256 Credential={access_key}/{scope}, SignedHeaders={signed}, Signature={sig}" + ); + headers.insert("authorization", HeaderValue::from_str(&auth).unwrap()); + headers +} + +fn provider(is_session: bool) -> BuiltinAuthProvider { + BuiltinAuthProvider::new( + MockCredentialStore { + credential: Some(credential(is_session)), + error: None, + }, + "us-east-1", + ) +} + +#[tokio::test] +async fn correctly_signed_long_term_key_authenticates() { + let headers = signed_headers("AKIAEXTENDDB00000000", "us-east-1", None, b"{}"); + let id = provider(false).authenticate(&headers, b"{}").await.unwrap(); + assert!(matches!(id, AuthIdentity::User { .. })); +} + +/// Measured 2026-09-16: a scope for another region is refused with this +/// exact message; the region is not named and the message ends with a space. +#[tokio::test] +async fn scope_for_another_region_is_rejected() { + let headers = signed_headers("AKIAEXTENDDB00000000", "us-west-2", None, b"{}"); + match provider(false).authenticate(&headers, b"{}").await { + Err(DynamoDbError::InvalidSignatureException(msg)) => { + assert_eq!(msg, "Credential should be scoped to a valid region. "); + } + other => panic!("Expected InvalidSignatureException, got: {other:?}"), + } +} + +/// A long-term key has no session token; presenting one is an invalid token. +#[tokio::test] +async fn session_token_with_a_long_term_key_is_rejected() { + let headers = signed_headers( + "AKIAEXTENDDB00000000", + "us-east-1", + Some("any-token"), + b"{}", + ); + match provider(false).authenticate(&headers, b"{}").await { + Err(DynamoDbError::UnrecognizedClientException(msg)) => { + assert_eq!(msg, "The security token included in the request is invalid"); + } + other => panic!("Expected UnrecognizedClientException, got: {other:?}"), + } +} + +#[tokio::test] +async fn session_key_with_matching_token_authenticates() { + let headers = signed_headers( + "ASIAEXTENDDB00000000", + "us-east-1", + Some("stored-token"), + b"{}", + ); + let id = provider(true).authenticate(&headers, b"{}").await.unwrap(); + assert!(matches!(id, AuthIdentity::RoleSession { .. })); +} + +/// Measured 2026-09-16: a presented token that does not resolve has no +/// trailing period; a missing token does. +#[tokio::test] +async fn session_key_token_messages_match_the_service() { + let wrong = signed_headers("ASIAEXTENDDB00000000", "us-east-1", Some("bogus"), b"{}"); + match provider(true).authenticate(&wrong, b"{}").await { + Err(DynamoDbError::UnrecognizedClientException(msg)) => { + assert_eq!(msg, "The security token included in the request is invalid"); + } + other => panic!("Expected UnrecognizedClientException, got: {other:?}"), + } + let missing = signed_headers("ASIAEXTENDDB00000000", "us-east-1", None, b"{}"); + match provider(true).authenticate(&missing, b"{}").await { + Err(DynamoDbError::UnrecognizedClientException(msg)) => { + assert_eq!( + msg, + "The security token included in the request is invalid." + ); + } + other => panic!("Expected UnrecognizedClientException, got: {other:?}"), + } +} + +/// The scope region is compared byte for byte: region names are lowercase +/// and the service does not fold case either. +#[tokio::test] +async fn scope_region_comparison_is_exact() { + for other in ["US-EAST-1", "us-east-1 ", ""] { + let headers = signed_headers("AKIAEXTENDDB00000000", other, None, b"{}"); + match provider(false).authenticate(&headers, b"{}").await { + Err(DynamoDbError::InvalidSignatureException(msg)) => { + assert_eq!( + msg, "Credential should be scoped to a valid region. ", + "{other:?}" + ); + } + other_result => { + panic!("{other:?}: expected InvalidSignatureException, got {other_result:?}") + } + } + } +} + +/// An empty token header on a long-term key is a presented token, and is +/// rejected like any other; only an absent header passes. +#[tokio::test] +async fn empty_session_token_header_with_a_long_term_key_is_rejected() { + let headers = signed_headers("AKIAEXTENDDB00000000", "us-east-1", Some(""), b"{}"); + match provider(false).authenticate(&headers, b"{}").await { + Err(DynamoDbError::UnrecognizedClientException(msg)) => { + assert_eq!(msg, "The security token included in the request is invalid"); + } + other => panic!("Expected UnrecognizedClientException, got: {other:?}"), + } +} + +/// A token header whose value is not visible ASCII is a presented token +/// too. It cannot be compared with a stored token, and for a long-term key +/// the rule is that only an absent header passes, so the request is +/// rejected rather than served as an unauthenticated-token request. +#[tokio::test] +async fn non_ascii_session_token_header_with_a_long_term_key_is_rejected() { + // Signed without a token, then the header is added unsigned: SigV4 + // covers the signed headers only, so the signature stays valid. + let mut headers = signed_headers("AKIAEXTENDDB00000000", "us-east-1", None, b"{}"); + headers.insert( + "x-amz-security-token", + HeaderValue::from_bytes(&[0xc3, 0xa9]).expect("a header value with a byte above 0x7f"), + ); + assert!( + headers["x-amz-security-token"].to_str().is_err(), + "the test header must not be readable as a string" + ); + match provider(false).authenticate(&headers, b"{}").await { + Err(DynamoDbError::UnrecognizedClientException(msg)) => { + assert_eq!(msg, "The security token included in the request is invalid"); + } + other => panic!("Expected UnrecognizedClientException, got: {other:?}"), + } +} + +/// Measured 2026-09-16: a scope wrong in both region and service reports +/// both sentences in one message, region first. +#[tokio::test] +async fn scope_wrong_in_region_and_service_reports_both() { + let mut headers = make_headers_with_auth("AKIAEXTENDDB00000000", None); + headers.remove("authorization"); + let date_str = headers["x-amz-date"].to_str().unwrap().to_owned(); + let date_short = date_str[..8].to_owned(); + let signed = "content-type;host;x-amz-date;x-amz-target"; + let creq = sigv4::canonical::canonical_request("POST", "/", "", &headers, signed, b"{}"); + let scope = format!("{date_short}/us-west-2/s3/aws4_request"); + let sts = sigv4::canonical::string_to_sign(&date_str, &scope, &creq); + let key = sigv4::signing_key::derive_signing_key(SECRET, &date_short, "us-west-2", "s3"); + let sig = sigv4::signing_key::compute_signature(&key, &sts); + let auth = format!( + "AWS4-HMAC-SHA256 Credential=AKIAEXTENDDB00000000/{scope}, SignedHeaders={signed}, Signature={sig}" + ); + headers.insert("authorization", HeaderValue::from_str(&auth).unwrap()); + match provider(false).authenticate(&headers, b"{}").await { + Err(DynamoDbError::InvalidSignatureException(msg)) => { + assert_eq!( + msg, + "Credential should be scoped to a valid region. Credential should be scoped to correct service: 'dynamodb'. " + ); + } + other => panic!("Expected InvalidSignatureException, got: {other:?}"), + } +} + +/// A provider whose store holds no credential at all. +fn provider_without_credentials() -> BuiltinAuthProvider { + BuiltinAuthProvider::new( + MockCredentialStore { + credential: None, + error: None, + }, + "us-east-1", + ) +} + +/// Move the request date one hour into the past, re-signing the scope +/// date in the `Authorization` header to match. +fn make_date_stale(headers: &mut HeaderMap) { + let fresh = headers["x-amz-date"].to_str().unwrap().to_owned(); + let stale = time::OffsetDateTime::now_utc() - time::Duration::hours(1); + let stale_str = format!( + "{:04}{:02}{:02}T{:02}{:02}{:02}Z", + stale.year(), + stale.month() as u8, + stale.day(), + stale.hour(), + stale.minute(), + stale.second(), + ); + let auth = headers["authorization"] + .to_str() + .unwrap() + .replacen(&fresh[..8], &stale_str[..8], 1); + headers.insert("authorization", HeaderValue::from_str(&auth).unwrap()); + headers.insert("x-amz-date", HeaderValue::from_str(&stale_str).unwrap()); +} + +/// Measured 2026-10-23 (R2-1 case F): an unknown access key scoped to +/// another region answers the scope message, so the scope check runs +/// before the key lookup. +#[tokio::test] +async fn unknown_key_with_a_wrong_region_reports_the_scope() { + let headers = signed_headers("AKIAEXAMPLEEXAMPLE01", "us-west-2", None, b"{}"); + match provider_without_credentials() + .authenticate(&headers, b"{}") + .await + { + Err(DynamoDbError::InvalidSignatureException(msg)) => { + assert_eq!(msg, "Credential should be scoped to a valid region. "); + } + other => panic!("Expected InvalidSignatureException, got: {other:?}"), + } +} + +/// Measured 2026-10-23 (R2-2 case V): an unknown access key with a date +/// one hour old answers "Signature expired", so the timestamp check runs +/// before the key lookup. The service's message continues with the dates +/// and its type is InvalidSignatureException; the server's timestamp +/// error shape predates this change and is not altered here. +#[tokio::test] +async fn unknown_key_with_a_stale_date_reports_the_timestamp() { + let mut headers = make_headers_with_auth("AKIAEXAMPLEEXAMPLE01", None); + make_date_stale(&mut headers); + match provider_without_credentials() + .authenticate(&headers, b"{}") + .await + { + Err(DynamoDbError::UnrecognizedClientException(msg)) => { + assert!(msg.starts_with("Signature expired: "), "{msg}"); + } + other => panic!("Expected the timestamp error, got: {other:?}"), + } +} + +/// Measured 2026-10-23 (R2-2 case D): a stale date scoped to another +/// region answers "Signature expired", so the timestamp check runs +/// before the scope check. +#[tokio::test] +async fn stale_date_with_a_wrong_region_reports_the_timestamp() { + let mut headers = signed_headers("AKIAEXTENDDB00000000", "us-west-2", None, b"{}"); + make_date_stale(&mut headers); + match provider(false).authenticate(&headers, b"{}").await { + Err(DynamoDbError::UnrecognizedClientException(msg)) => { + assert!(msg.starts_with("Signature expired: "), "{msg}"); + } + other => panic!("Expected the timestamp error, got: {other:?}"), + } +} diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index ccdb5833d..f42bdae07 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -78,6 +78,16 @@ pub struct ServerConfig { /// Enable provisioned throughput throttling via token buckets. /// When `None` or `false`, all requests are allowed regardless of capacity. pub throttling_enabled: Option, + /// Seconds a connection may take to deliver its request headers, and + /// seconds a request body may go without delivering data, before the + /// server closes the connection. Bounds how long a stalled or malicious + /// client can hold a connection open. Default 30. + #[serde(default = "default_request_timeout_secs")] + pub request_timeout_secs: u64, +} + +fn default_request_timeout_secs() -> u64 { + 30 } impl Default for ServerConfig { @@ -89,6 +99,7 @@ impl Default for ServerConfig { run_dir: default_run_dir(), tls: TlsConfig::default(), throttling_enabled: None, + request_timeout_secs: default_request_timeout_secs(), } } } diff --git a/crates/core/src/error/mod.rs b/crates/core/src/error/mod.rs index 071d15278..a2ebe4b5f 100755 --- a/crates/core/src/error/mod.rs +++ b/crates/core/src/error/mod.rs @@ -211,7 +211,7 @@ impl DynamoDbError { Self::InternalServerError(_) => "InternalServerError", Self::AccessDeniedException(_) => "AccessDeniedException", Self::MissingAuthenticationToken(_) => "MissingAuthenticationToken", - Self::IncompleteSignature(_) => "IncompleteSignature", + Self::IncompleteSignature(_) => "IncompleteSignatureException", Self::UnrecognizedClientException(_) => "UnrecognizedClientException", Self::ExpiredTokenException(_) => "ExpiredTokenException", Self::ExpiredIteratorException(_) => "ExpiredIteratorException", diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index e674e5c58..29c0ef917 100755 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -24,6 +24,7 @@ async-trait = { workspace = true } tower = { workspace = true } tower-http = { workspace = true } hyper = { workspace = true } +hyper-util = { workspace = true } tokio = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } @@ -52,5 +53,7 @@ libc = { workspace = true } [dev-dependencies] extenddb-cache = { workspace = true, features = ["test-util"] } +# Self-signed certificates for the TLS listener tests. +rcgen = { workspace = true } tempfile = "3" tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/server/src/handler.rs b/crates/server/src/handler.rs index aeef4ccd1..356395f36 100755 --- a/crates/server/src/handler.rs +++ b/crates/server/src/handler.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use axum::body::Bytes; use axum::extract::{ConnectInfo, State}; use axum::http::HeaderMap; -use axum::response::Response; +use axum::response::{IntoResponse, Response}; use extenddb_core::error::DynamoDbError; use extenddb_engine::OperationContext; use serde_json::Value; @@ -48,14 +48,31 @@ pub(crate) async fn handle_request( } let request_id = uuid::Uuid::new_v4().to_string(); - // SP-WIRE-007: body limit exceeded → 413 - let Ok(body) = body else { - return error_response( - &DynamoDbError::RequestEntityTooLargeException( - "Request size has exceeded the maximum allowed size".to_owned(), - ), - &request_id, - ); + // A repeated `host` or `authorization` header was refused before routing + // (see `repeated_headers`), so each is read as one value from here on. + + // SP-WIRE-007: body limit exceeded → 413. Any other failure to read the + // body is the body timeout (the client stopped sending) → 408. + let body = match body { + Ok(body) => body, + Err(axum::extract::rejection::BytesRejection::FailedToBufferBody( + axum::extract::rejection::FailedToBufferBody::LengthLimitError(_), + )) => { + return error_response( + &DynamoDbError::RequestEntityTooLargeException( + "Request size has exceeded the maximum allowed size".to_owned(), + ), + &request_id, + ); + } + Err(_) => { + return ( + axum::http::StatusCode::REQUEST_TIMEOUT, + [(axum::http::header::CONNECTION, "close")], + "request body was not received within the server's request timeout", + ) + .into_response(); + } }; // Extract operation from X-Amz-Target diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs old mode 100755 new mode 100644 index dd4cb480b..97c7abc57 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -18,10 +18,12 @@ mod handler; mod import_export_paths; pub mod key_info_cache; mod leading_keys; +mod listener; pub mod management; mod metrics_endpoint; pub mod rate_limit; mod registration; +mod repeated_headers; mod request_helpers; mod request_tags; mod response; @@ -53,6 +55,7 @@ pub use key_info_cache::CachedTableKeyInfoStore; use serde_json::json; use tower::ServiceBuilder; use tower_http::set_header::SetResponseHeaderLayer; +use tower_http::timeout::RequestBodyTimeoutLayer; /// Application state shared across all handlers. pub struct AppState { @@ -72,6 +75,8 @@ pub struct AppState { pub metrics: Arc, /// Whether TLS is enabled (affects cookie Secure flag). pub tls_enabled: bool, + /// Header read timeout and body inactivity timeout for every connection. + pub request_timeout: Duration, /// Developer mode: authorization is opened for authenticated callers (SigV4 /// is still verified). Only ever true in a `dev-mode` build bound to /// loopback; the bin's serve path enforces those preconditions. @@ -138,6 +143,7 @@ pub async fn start_server( const DYNAMODB_BODY_LIMIT: usize = 16 * 1024 * 1024; let tls_enabled = state.tls_enabled; + let request_timeout = state.request_timeout; let catalog_store = state.catalog_store.clone(); let auth_cache_for_mgmt = state.auth_cache.clone(); let auth_cache_for_console = state.auth_cache.clone(); @@ -228,8 +234,17 @@ pub async fn start_server( registration::refuse_while_unregistered, )); - // Apply security headers to all routes. - let app = app.layer(security_layers); + // Refuse a repeated host or authorization header on every route, then + // apply security headers to all routes (the refusal carries them too). + // A body that stops delivering data for `request_timeout` errors out, so a + // client that announces a Content-Length it never sends cannot hold the + // connection open. The handler answers that error with 408. + let app = app + .layer(axum::middleware::from_fn( + repeated_headers::refuse_repeated_headers, + )) + .layer(security_layers) + .layer(RequestBodyTimeoutLayer::new(request_timeout)); // Add HSTS header only when TLS is enabled. let app = if tls_enabled { @@ -241,66 +256,7 @@ pub async fn start_server( app }; - let local_addr = listener.local_addr()?; - - if let Some(tls_cfg) = tls { - // P57 Bug 2 fix: rustls 0.23 requires an explicit CryptoProvider. - // Install aws-lc-rs as the default before creating any TLS config. - // Ignore the error if a provider was already installed (e.g., by sqlx). - let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); - - tracing::info!("extenddb listening on {local_addr} (HTTPS)"); - let rustls_config = axum_server::tls_rustls::RustlsConfig::from_pem_file( - &tls_cfg.cert_path, - &tls_cfg.key_path, - ) - .await - // AI-3: The error intentionally does not include the file path to avoid - // leaking filesystem structure in logs. See docs/troubleshooting.md for - // the "Failed to load TLS certificates" entry. - .map_err(|e| anyhow::anyhow!("Failed to load TLS certificates: {e}"))?; - - let handle = axum_server::Handle::new(); - let shutdown_handle = handle.clone(); - let shutdown_pid = pid_file.clone(); - let stop = admission.clone(); - tokio::spawn(async move { - signal_or_stop(&stop).await; - shutdown_handle.graceful_shutdown(Some(Duration::from_secs(5))); - // P50: Log timeout but don't call std::process::exit — let the - // runtime shut down normally so destructors (including ZeroizeOnDrop) run. - tokio::time::sleep(Duration::from_secs(10)).await; - tracing::warn!("Graceful shutdown timed out, forcing PID file cleanup"); - cleanup_pid_file(shutdown_pid.as_deref()); - }); - - let std_listener = listener.into_std()?; - // AI-2: Use a custom acceptor that peeks the first byte of each - // connection. If it's plain HTTP (not 0x16 TLS ClientHello), write - // a 301 redirect to HTTPS and reject the connection before the TLS - // handshake. This gives users a helpful redirect instead of a - // confusing TLS handshake failure. The same acceptor sets - // TCP_NODELAY on every accepted socket before the handshake. - let redirect_acceptor = HttpsRedirectAcceptor { addr: local_addr }; - axum_server::from_tcp_rustls(std_listener, rustls_config)? - .map(|tls| tls.acceptor(redirect_acceptor)) - .handle(handle) - .serve(app.into_make_service_with_connect_info::()) - .await?; - } else { - tracing::info!("extenddb listening on {local_addr}"); - axum::serve( - nodelay_listener(listener), - app.into_make_service_with_connect_info::(), - ) - .with_graceful_shutdown(graceful_shutdown(pid_file.clone(), admission)) - .await?; - } - - // Normal shutdown path — clean up PID file after connections drain. - cleanup_pid_file(pid_file.as_deref()); - - Ok(()) + listener::serve_app(listener, app, request_timeout, tls, pid_file, admission).await } /// GET /health — REQ-OBS-006. @@ -337,264 +293,3 @@ async fn health_check( axum::Json(json!({ "status": status, "components": components })), ) } - -async fn graceful_shutdown(pid_file: Option, admission: registration::Admission) { - signal_or_stop(&admission).await; - // P50: Spawn a timeout task that cleans up the PID file if connections - // don't drain. Does NOT call std::process::exit — the runtime shuts down - // normally so destructors (including ZeroizeOnDrop) run. - tokio::spawn(async move { - tokio::time::sleep(Duration::from_secs(5)).await; - tracing::warn!("Graceful shutdown timed out after 5s, forcing PID file cleanup"); - cleanup_pid_file(pid_file.as_deref()); - }); -} - -/// Remove the PID file if it exists. Best-effort — log but don't fail. -fn cleanup_pid_file(pid_file: Option<&std::path::Path>) { - if let Some(path) = pid_file - && let Err(e) = std::fs::remove_file(path) - && e.kind() != std::io::ErrorKind::NotFound - { - tracing::warn!("Failed to remove PID file {}: {e}", path.display()); - } -} - -/// Set `TCP_NODELAY` on an accepted socket. -/// -/// With Nagle's algorithm on, the kernel holds a small write while an earlier -/// small write on the connection is unacknowledged, and the client's delayed -/// ACK leaves with its next request on that connection or when its timer -/// fires (about 40 ms on Linux). Once one response is written while the -/// previous one is unacknowledged, each response waits for the client's next -/// request, and the reply to that request is held behind it in turn. The -/// response latency then equals the gap between the client's requests on the -/// connection, for every gap under the delayed-ACK timer. `TCP_NODELAY` sends -/// each write as it is made. The option lives on the TCP socket, so under TLS -/// it is set before the handshake and applies below the TLS stream. A failure -/// to set it is logged at debug and the connection is served with the kernel -/// default. -fn set_tcp_nodelay(stream: &tokio::net::TcpStream) { - if let Err(e) = stream.set_nodelay(true) { - tracing::debug!("Failed to set TCP_NODELAY on accepted connection: {e}"); - } -} - -/// The plaintext listener: a `TcpListener` whose accepted sockets carry -/// `TCP_NODELAY`. -type NodelayListener = axum::serve::TapIo; - -/// Wrap the plaintext listener so every accepted socket carries `TCP_NODELAY`. -/// -/// `axum::serve` does not set the option itself; `ListenerExt::tap_io` is its -/// hook for per-connection socket options. -fn nodelay_listener(listener: tokio::net::TcpListener) -> NodelayListener { - use axum::serve::ListenerExt; - listener.tap_io(|stream| set_tcp_nodelay(stream)) -} - -/// AI-2: Acceptor that detects plain HTTP connections on the TLS port. -/// -/// Sets `TCP_NODELAY` on the accepted socket first (see [`set_tcp_nodelay`]). -/// Then peeks the first byte of the connection. If it's `0x16` (TLS -/// `ClientHello`), the connection passes through to the TLS acceptor. If it's -/// a plain HTTP verb, a 301 redirect to `https://` is written and the -/// connection is rejected with an IO error (which `axum_server` handles by -/// dropping it). -/// -/// `axum_server` 0.8 ships a `NoDelayAcceptor`, but it cannot be stacked under -/// this acceptor without a generic wrapper, so this acceptor sets the option -/// itself, and logs instead of rejecting on failure. -#[derive(Clone)] -struct HttpsRedirectAcceptor { - addr: std::net::SocketAddr, -} - -impl axum_server::accept::Accept - for HttpsRedirectAcceptor -{ - type Stream = tokio::net::TcpStream; - type Service = S; - type Future = std::pin::Pin< - Box< - dyn std::future::Future> + Send, - >, - >; - - fn accept(&self, stream: tokio::net::TcpStream, service: S) -> Self::Future { - let addr = self.addr; - Box::pin(async move { - set_tcp_nodelay(&stream); - let mut peek_buf = [0u8; 1]; - match stream.peek(&mut peek_buf).await { - Ok(0) => { - // Connection closed before sending data. - Err(std::io::Error::new( - std::io::ErrorKind::ConnectionReset, - "empty connection", - )) - } - Ok(_) if peek_buf[0] == 0x16 => { - // TLS ClientHello — pass through to TLS acceptor. - Ok((stream, service)) - } - Ok(_) => { - // Plain HTTP — send redirect and reject. - send_https_redirect(&stream, addr); - Err(std::io::Error::new( - std::io::ErrorKind::InvalidData, - "plain HTTP redirected to HTTPS", - )) - } - Err(e) => Err(e), - } - }) - } -} - -/// Write an HTTP 301 redirect response to HTTPS on the raw TCP stream. -fn send_https_redirect(stream: &tokio::net::TcpStream, addr: std::net::SocketAddr) { - let location = format!("https://{addr}/"); - let response = format!( - "HTTP/1.1 301 Moved Permanently\r\n\ - Location: {location}\r\n\ - Content-Length: 0\r\n\ - Connection: close\r\n\r\n" - ); - // Best-effort write — if it fails, the client already got a connection error. - let _ = stream.try_write(response.as_bytes()); -} - -/// Resolves on a shutdown signal, or once `admission` asks for a stop -/// because the serving registration could not be taken again. -async fn signal_or_stop(admission: ®istration::Admission) { - tokio::select! { - () = shutdown_signal() => admission.mark_signalled(), - () = admission.stop_requested() => { - tracing::info!( - "Stopping because the serving registration could not be taken again, draining \ - connections..." - ); - } - } -} - -#[cfg(unix)] -async fn shutdown_signal() { - let ctrl_c = tokio::signal::ctrl_c(); - let Ok(mut sigterm) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) - else { - tracing::error!("Failed to register SIGTERM handler, falling back to Ctrl+C only"); - let _ = ctrl_c.await; - tracing::info!("Shutdown signal received, draining connections..."); - return; - }; - tokio::select! { - _ = ctrl_c => {}, - _ = sigterm.recv() => {}, - } - tracing::info!("Shutdown signal received, draining connections..."); -} - -/// Non-unix: there is no SIGTERM; Ctrl+C (which tokio maps to the console -/// control handler on Windows) is the only shutdown signal — and it only -/// fires for an interactive console. Supervisors that terminate the process -/// directly (including the npm launcher's `stop()`, which is -/// `TerminateProcess` on Windows) hard-kill without reaching this path, so -/// the worker drain is skipped. That is acceptable: drain only cancels and -/// joins background workers — it flushes nothing, and storage commits -/// per-transaction. -#[cfg(not(unix))] -async fn shutdown_signal() { - let _ = tokio::signal::ctrl_c().await; - tracing::info!("Shutdown signal received, draining connections..."); -} - -#[cfg(test)] -mod tests { - use super::*; - use tokio::io::AsyncWriteExt; - use tokio::net::{TcpListener, TcpStream}; - - /// Bind a loopback listener and connect one client to it. - async fn bind_and_connect() -> (TcpListener, TcpStream) { - let listener = TcpListener::bind("127.0.0.1:0") - .await - .expect("bind loopback"); - let addr = listener.local_addr().expect("local addr"); - let client = TcpStream::connect(addr).await.expect("connect"); - (listener, client) - } - - /// The TLS path: the acceptor that runs before the TLS handshake sets - /// `TCP_NODELAY` on the accepted socket, read back from the socket the - /// acceptor hands on. - #[tokio::test] - async fn tls_acceptor_sets_tcp_nodelay_before_the_handshake() { - use axum_server::accept::Accept; - - let (listener, mut client) = bind_and_connect().await; - // A TLS ClientHello starts with the handshake record type. - client - .write_all(&[0x16]) - .await - .expect("write ClientHello byte"); - let (accepted, _) = listener.accept().await.expect("accept"); - assert!( - !accepted.nodelay().expect("read TCP_NODELAY"), - "a fresh accepted socket has Nagle on" - ); - - let acceptor = HttpsRedirectAcceptor { - addr: listener.local_addr().expect("local addr"), - }; - let (stream, ()) = acceptor - .accept(accepted, ()) - .await - .expect("a TLS ClientHello passes through to the TLS acceptor"); - assert!(stream.nodelay().expect("read TCP_NODELAY")); - } - - /// The plaintext path: the listener handed to `axum::serve` sets - /// `TCP_NODELAY` on every socket it accepts. - #[tokio::test] - async fn plaintext_listener_sets_tcp_nodelay_on_accepted_sockets() { - use axum::serve::Listener; - - let (listener, client) = bind_and_connect().await; - let mut listener = nodelay_listener(listener); - let (accepted, peer) = listener.accept().await; - assert_eq!(peer, client.local_addr().expect("client addr")); - assert!(accepted.nodelay().expect("read TCP_NODELAY")); - } - - /// A stop that the registration supervisor asks for ends the plaintext - /// server through the graceful shutdown a signal starts. - #[tokio::test] - async fn a_stop_after_a_lost_registration_ends_the_server_as_a_signal_does() { - use std::future::IntoFuture; - - let listener = TcpListener::bind("127.0.0.1:0") - .await - .expect("bind loopback"); - let admission = registration::Admission::default(); - let app = Router::new().route("/", get(|| async { "served" })); - let server = tokio::spawn( - axum::serve(listener, app) - .with_graceful_shutdown(graceful_shutdown(None, admission.clone())) - .into_future(), - ); - tokio::time::sleep(Duration::from_millis(100)).await; - assert!(!server.is_finished(), "the server stopped by itself"); - admission.stop("the registration could not be taken again".to_owned()); - tokio::time::timeout(Duration::from_secs(10), server) - .await - .expect("the server stops within 10 s of the stop") - .expect("the server task") - .expect("the server"); - assert_eq!( - admission.stop_reason(), - Some("the registration could not be taken again") - ); - } -} diff --git a/crates/server/src/listener.rs b/crates/server/src/listener.rs new file mode 100644 index 000000000..099ae3d84 --- /dev/null +++ b/crates/server/src/listener.rs @@ -0,0 +1,473 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The listener: serves the built router over TLS or plaintext, applies the +//! per-connection request timeouts, sets `TCP_NODELAY` on accepted sockets, +//! redirects plain HTTP arriving on the TLS port, and handles shutdown. + +use std::path::PathBuf; +use std::time::Duration; + +use axum::Router; + +use crate::ServerTlsConfig; +use crate::registration::Admission; + +/// Serve `app` on `listener` until a shutdown signal arrives, or until +/// `admission` asks for a stop after the serving registration was lost. +/// +/// When `tls` is `Some`, the server serves HTTPS using `axum-server` with +/// rustls. When `tls` is `None`, the server serves plaintext HTTP. Both arms +/// run on the same `axum-server` builder, so the request timeouts, the +/// shutdown handle and `TCP_NODELAY` apply to every listener mode, and a +/// signal and a stop drain the listener the same way. +pub(crate) async fn serve_app( + listener: tokio::net::TcpListener, + app: Router, + request_timeout: Duration, + tls: Option, + pid_file: Option, + admission: Admission, +) -> Result<(), anyhow::Error> { + let local_addr = listener.local_addr()?; + + let handle = axum_server::Handle::new(); + let shutdown_handle = handle.clone(); + let shutdown_pid = pid_file.clone(); + tokio::spawn(async move { + signal_or_stop(&admission).await; + shutdown_handle.graceful_shutdown(Some(Duration::from_secs(5))); + // P50: Log timeout but don't call std::process::exit — let the + // runtime shut down normally so destructors (including ZeroizeOnDrop) run. + tokio::time::sleep(Duration::from_secs(10)).await; + tracing::warn!("Graceful shutdown timed out, forcing PID file cleanup"); + cleanup_pid_file(shutdown_pid.as_deref()); + }); + + let std_listener = listener.into_std()?; + let make_service = app.into_make_service_with_connect_info::(); + + if let Some(tls_cfg) = tls { + // P57 Bug 2 fix: rustls 0.23 requires an explicit CryptoProvider. + // Install aws-lc-rs as the default before creating any TLS config. + // Ignore the error if a provider was already installed (e.g., by sqlx). + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + + tracing::info!("extenddb listening on {local_addr} (HTTPS)"); + let rustls_config = axum_server::tls_rustls::RustlsConfig::from_pem_file( + &tls_cfg.cert_path, + &tls_cfg.key_path, + ) + .await + // AI-3: The error intentionally does not include the file path to avoid + // leaking filesystem structure in logs. See docs/troubleshooting.md for + // the "Failed to load TLS certificates" entry. + .map_err(|e| anyhow::anyhow!("Failed to load TLS certificates: {e}"))?; + + // AI-2: Use a custom acceptor that peeks the first byte of each + // connection. If it's plain HTTP (not 0x16 TLS ClientHello), write + // a 301 redirect to HTTPS and reject the connection before the TLS + // handshake. This gives users a helpful redirect instead of a + // confusing TLS handshake failure. The same acceptor sets + // TCP_NODELAY on every accepted socket before the handshake, and + // bounds the wait for that first byte by `request_timeout`. + let redirect_acceptor = HttpsRedirectAcceptor { + addr: local_addr, + first_byte_timeout: request_timeout, + }; + let server = axum_server::from_tcp_rustls(std_listener, rustls_config)? + .map(|tls| FirstByteAcceptor::new(tls.acceptor(redirect_acceptor), request_timeout)) + .handle(handle); + with_request_timeouts(server, request_timeout) + .serve(make_service) + .await?; + } else { + tracing::info!("extenddb listening on {local_addr}"); + let server = axum_server::from_tcp(std_listener)? + .acceptor(FirstByteAcceptor::new(NodelayAcceptor, request_timeout)) + .handle(handle); + with_request_timeouts(server, request_timeout) + .serve(make_service) + .await?; + } + + // Normal shutdown path — clean up PID file after connections drain. + cleanup_pid_file(pid_file.as_deref()); + + Ok(()) +} + +/// Apply the per-connection request timeouts to a server, in either +/// listener mode. +/// +/// HTTP/1: a connection that has not finished sending its request headers +/// within `request_timeout` is closed; the same bound applies to the idle +/// wait for the next request on a keep-alive connection. +/// +/// HTTP/2: a connection that stops answering PING frames is closed. A client +/// that sends the preface and then stalls holds the connection for at most +/// interval plus timeout. +/// +/// Neither timer runs until the connection has delivered its first bytes: +/// `hyper_util`'s auto builder reads the stream to tell HTTP/2 from HTTP/1 +/// before either protocol's timer starts, and that read has no bound of its +/// own. [`FirstByteAcceptor`] bounds it. +fn with_request_timeouts( + mut server: axum_server::Server, + request_timeout: Duration, +) -> axum_server::Server { + server + .http_builder() + .http1() + .timer(hyper_util::rt::TokioTimer::new()) + .header_read_timeout(request_timeout); + server + .http_builder() + .http2() + .timer(hyper_util::rt::TokioTimer::new()) + .keep_alive_interval(Some(request_timeout)) + .keep_alive_timeout(request_timeout); + server +} + +/// Acceptor wrapper that bounds the wait for the start of a request. +/// +/// Runs the inner acceptor (TLS handshake, `TCP_NODELAY`) and hands the +/// resulting stream on inside a [`FirstByteDeadline`]. A client that +/// connects (and, on the TLS listener, completes the handshake) and then +/// sends nothing, or sends only a prefix of the 24-byte HTTP/2 preface, is +/// closed once `timeout` passes. +#[derive(Clone)] +struct FirstByteAcceptor { + inner: A, + timeout: Duration, +} + +impl FirstByteAcceptor { + fn new(inner: A, timeout: Duration) -> Self { + Self { inner, timeout } + } +} + +impl axum_server::accept::Accept for FirstByteAcceptor +where + A: axum_server::accept::Accept, + A::Future: Send + 'static, + A::Stream: Send + 'static, + A::Service: Send + 'static, +{ + type Stream = FirstByteDeadline; + type Service = A::Service; + type Future = std::pin::Pin< + Box< + dyn std::future::Future> + Send, + >, + >; + + fn accept(&self, stream: I, service: S) -> Self::Future { + let inner = self.inner.accept(stream, service); + let timeout = self.timeout; + Box::pin(async move { + let (stream, service) = inner.await?; + Ok((FirstByteDeadline::new(stream, timeout), service)) + }) + } +} + +/// The HTTP/2 client preface. `hyper_util`'s auto builder reads a new +/// connection until it holds these 24 bytes or a byte that differs from +/// them, and only then hands the stream to HTTP/1 or HTTP/2, whose own +/// timers start at that point. +const H2_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"; + +/// IO adapter that bounds the version sniff on a new connection. +/// +/// The deadline starts when the adapter is built (after accept, or after the +/// TLS handshake). It is dropped once the bytes read so far decide the +/// protocol: a byte that differs from the HTTP/2 preface, or the whole +/// preface. Until then a pending read also polls the deadline, so the +/// connection task wakes when it fires and the read fails with `TimedOut`; +/// `hyper_util` then drops the connection. Writes pass through unchanged. +pub(crate) struct FirstByteDeadline { + inner: S, + deadline: Option>>, + /// Bytes read so far that match the preface prefix, while undecided. + preface_matched: usize, +} + +impl FirstByteDeadline { + fn new(inner: S, timeout: Duration) -> Self { + Self { + inner, + deadline: Some(Box::pin(tokio::time::sleep(timeout))), + preface_matched: 0, + } + } + + /// Feed newly read bytes to the sniff; returns true once the protocol is + /// decided (the auto builder stops reading at the same point). + fn sniff_decided(&mut self, new_bytes: &[u8]) -> bool { + for &b in new_bytes { + if self.preface_matched >= H2_PREFACE.len() || b != H2_PREFACE[self.preface_matched] { + return true; + } + self.preface_matched += 1; + } + self.preface_matched >= H2_PREFACE.len() + } +} + +impl tokio::io::AsyncRead for FirstByteDeadline { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + use std::task::Poll; + + let before = buf.filled().len(); + let read = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf); + if self.deadline.is_none() { + return read; + } + match read { + Poll::Ready(Ok(())) => { + let new_bytes = buf.filled()[before..].to_vec(); + if self.sniff_decided(&new_bytes) { + self.deadline = None; + } + Poll::Ready(Ok(())) + } + Poll::Pending => { + let deadline = self.deadline.as_mut().expect("checked above"); + match deadline.as_mut().poll(cx) { + Poll::Ready(()) => Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "no request received within request_timeout_secs", + ))), + Poll::Pending => Poll::Pending, + } + } + other => other, + } + } +} + +impl tokio::io::AsyncWrite for FirstByteDeadline { + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_write(cx, buf) + } + + fn poll_flush( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_shutdown(cx) + } + + fn poll_write_vectored( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + bufs: &[std::io::IoSlice<'_>], + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.inner).poll_write_vectored(cx, bufs) + } + + fn is_write_vectored(&self) -> bool { + self.inner.is_write_vectored() + } +} + +/// Remove the PID file if it exists. Best-effort — log but don't fail. +fn cleanup_pid_file(pid_file: Option<&std::path::Path>) { + if let Some(path) = pid_file + && let Err(e) = std::fs::remove_file(path) + && e.kind() != std::io::ErrorKind::NotFound + { + tracing::warn!("Failed to remove PID file {}: {e}", path.display()); + } +} + +/// Set `TCP_NODELAY` on an accepted socket. +/// +/// With Nagle's algorithm on, the kernel holds a small write while an earlier +/// small write on the connection is unacknowledged, and the client's delayed +/// ACK leaves with its next request on that connection or when its timer +/// fires (about 40 ms on Linux). Once one response is written while the +/// previous one is unacknowledged, each response waits for the client's next +/// request, and the reply to that request is held behind it in turn. The +/// response latency then equals the gap between the client's requests on the +/// connection, for every gap under the delayed-ACK timer. `TCP_NODELAY` sends +/// each write as it is made. The option lives on the TCP socket, so under TLS +/// it is set before the handshake and applies below the TLS stream. A failure +/// to set it is logged at debug and the connection is served with the kernel +/// default. +fn set_tcp_nodelay(stream: &tokio::net::TcpStream) { + if let Err(e) = stream.set_nodelay(true) { + tracing::debug!("Failed to set TCP_NODELAY on accepted connection: {e}"); + } +} + +/// The plaintext acceptor: sets `TCP_NODELAY` on every accepted socket (see +/// [`set_tcp_nodelay`]) and hands it on unchanged. +/// +/// `axum_server` 0.8 ships a `NoDelayAcceptor`, but it rejects the +/// connection when the option cannot be set; this one logs and serves it, +/// as the TLS acceptor does. +#[derive(Clone)] +struct NodelayAcceptor; + +impl axum_server::accept::Accept for NodelayAcceptor { + type Stream = tokio::net::TcpStream; + type Service = S; + type Future = std::future::Ready>; + + fn accept(&self, stream: tokio::net::TcpStream, service: S) -> Self::Future { + set_tcp_nodelay(&stream); + std::future::ready(Ok((stream, service))) + } +} + +/// AI-2: Acceptor that detects plain HTTP connections on the TLS port. +/// +/// Sets `TCP_NODELAY` on the accepted socket first (see [`set_tcp_nodelay`]). +/// Then peeks the first byte of the connection. If it's `0x16` (TLS +/// `ClientHello`), the connection passes through to the TLS acceptor. If it's +/// a plain HTTP verb, a 301 redirect to `https://` is written and the +/// connection is rejected with an IO error (which `axum_server` handles by +/// dropping it). +/// +/// `axum_server` 0.8 ships a `NoDelayAcceptor`, but it cannot be stacked under +/// this acceptor without a generic wrapper, so this acceptor sets the option +/// itself, and logs instead of rejecting on failure. +/// +/// The peek waits at most `first_byte_timeout` for the first byte, so a +/// client that connects and sends nothing is dropped before the handshake. +#[derive(Clone)] +struct HttpsRedirectAcceptor { + addr: std::net::SocketAddr, + first_byte_timeout: Duration, +} + +impl axum_server::accept::Accept + for HttpsRedirectAcceptor +{ + type Stream = tokio::net::TcpStream; + type Service = S; + type Future = std::pin::Pin< + Box< + dyn std::future::Future> + Send, + >, + >; + + fn accept(&self, stream: tokio::net::TcpStream, service: S) -> Self::Future { + let addr = self.addr; + let first_byte_timeout = self.first_byte_timeout; + Box::pin(async move { + set_tcp_nodelay(&stream); + let mut peek_buf = [0u8; 1]; + let peeked = tokio::time::timeout(first_byte_timeout, stream.peek(&mut peek_buf)) + .await + .map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "no bytes received before the TLS handshake within request_timeout_secs", + ) + })?; + match peeked { + Ok(0) => { + // Connection closed before sending data. + Err(std::io::Error::new( + std::io::ErrorKind::ConnectionReset, + "empty connection", + )) + } + Ok(_) if peek_buf[0] == 0x16 => { + // TLS ClientHello — pass through to TLS acceptor. + Ok((stream, service)) + } + Ok(_) => { + // Plain HTTP — send redirect and reject. + send_https_redirect(&stream, addr); + Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "plain HTTP redirected to HTTPS", + )) + } + Err(e) => Err(e), + } + }) + } +} + +/// Write an HTTP 301 redirect response to HTTPS on the raw TCP stream. +fn send_https_redirect(stream: &tokio::net::TcpStream, addr: std::net::SocketAddr) { + let location = format!("https://{addr}/"); + let response = format!( + "HTTP/1.1 301 Moved Permanently\r\n\ + Location: {location}\r\n\ + Content-Length: 0\r\n\ + Connection: close\r\n\r\n" + ); + // Best-effort write — if it fails, the client already got a connection error. + let _ = stream.try_write(response.as_bytes()); +} + +/// Resolves on a shutdown signal, or once `admission` asks for a stop +/// because the serving registration could not be taken again. +async fn signal_or_stop(admission: &Admission) { + tokio::select! { + () = shutdown_signal() => admission.mark_signalled(), + () = admission.stop_requested() => { + tracing::info!( + "Stopping because the serving registration could not be taken again, draining \ + connections..." + ); + } + } +} + +#[cfg(unix)] +async fn shutdown_signal() { + let ctrl_c = tokio::signal::ctrl_c(); + let Ok(mut sigterm) = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + else { + tracing::error!("Failed to register SIGTERM handler, falling back to Ctrl+C only"); + let _ = ctrl_c.await; + tracing::info!("Shutdown signal received, draining connections..."); + return; + }; + tokio::select! { + _ = ctrl_c => {}, + _ = sigterm.recv() => {}, + } + tracing::info!("Shutdown signal received, draining connections..."); +} + +/// Non-unix: there is no SIGTERM; Ctrl+C (which tokio maps to the console +/// control handler on Windows) is the only shutdown signal — and it only +/// fires for an interactive console. Supervisors that terminate the process +/// directly (including the npm launcher's `stop()`, which is +/// `TerminateProcess` on Windows) hard-kill without reaching this path, so +/// the worker drain is skipped. That is acceptable: drain only cancels and +/// joins background workers — it flushes nothing, and storage commits +/// per-transaction. +#[cfg(not(unix))] +async fn shutdown_signal() { + let _ = tokio::signal::ctrl_c().await; + tracing::info!("Shutdown signal received, draining connections..."); +} + +#[cfg(test)] +mod tests; diff --git a/crates/server/src/listener/tests.rs b/crates/server/src/listener/tests.rs new file mode 100644 index 000000000..606fc1740 --- /dev/null +++ b/crates/server/src/listener/tests.rs @@ -0,0 +1,431 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Unit tests for the listener: request timeouts, the first-byte deadline, +//! `TCP_NODELAY` on both listener modes, and the stop a lost serving +//! registration asks for. + +use super::*; +use tokio::io::AsyncWriteExt; +use tokio::net::{TcpListener, TcpStream}; + +/// Bind a loopback listener and connect one client to it. +async fn bind_and_connect() -> (TcpListener, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind loopback"); + let addr = listener.local_addr().expect("local addr"); + let client = TcpStream::connect(addr).await.expect("connect"); + (listener, client) +} + +/// The plaintext path: a connection that sends part of its request +/// headers and then stalls is closed once `request_timeout` passes, +/// as it is on the TLS path. The client reads EOF (or a reset) well +/// before a bound of several times the timeout; without the timer the +/// read blocks until that bound. +#[tokio::test] +async fn plaintext_listener_closes_a_stalled_header_phase() { + use tokio::io::AsyncReadExt; + + let timeout = Duration::from_millis(300); + let (listener, mut client) = bind_and_connect().await; + let app = Router::new().route("/", axum::routing::get(|| async { "ok" })); + let server = tokio::spawn(serve_app( + listener, + app, + timeout, + None, + None, + Admission::default(), + )); + + // A request line and one header, no terminating blank line. + client + .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n") + .await + .expect("write partial headers"); + let mut buf = [0u8; 64]; + let outcome = tokio::time::timeout(timeout * 10, client.read(&mut buf)).await; + server.abort(); + match outcome { + Ok(Ok(0)) | Ok(Err(_)) => {} + Ok(Ok(n)) => panic!("server answered {n} bytes instead of closing the connection"), + Err(_) => panic!( + "server held a stalled header phase open for {:?} on the plaintext listener", + timeout * 10 + ), + } +} + +/// The TLS path: the acceptor that runs before the TLS handshake sets +/// `TCP_NODELAY` on the accepted socket, read back from the socket the +/// acceptor hands on. +#[tokio::test] +async fn tls_acceptor_sets_tcp_nodelay_before_the_handshake() { + use axum_server::accept::Accept; + + let (listener, mut client) = bind_and_connect().await; + // A TLS ClientHello starts with the handshake record type. + client + .write_all(&[0x16]) + .await + .expect("write ClientHello byte"); + let (accepted, _) = listener.accept().await.expect("accept"); + assert!( + !accepted.nodelay().expect("read TCP_NODELAY"), + "a fresh accepted socket has Nagle on" + ); + + let acceptor = HttpsRedirectAcceptor { + addr: listener.local_addr().expect("local addr"), + first_byte_timeout: Duration::from_secs(5), + }; + let (stream, ()) = acceptor + .accept(accepted, ()) + .await + .expect("a TLS ClientHello passes through to the TLS acceptor"); + assert!(stream.nodelay().expect("read TCP_NODELAY")); +} + +/// The plaintext path: the acceptor sets `TCP_NODELAY` on every socket +/// it accepts. +#[tokio::test] +async fn plaintext_listener_sets_tcp_nodelay_on_accepted_sockets() { + use axum_server::accept::Accept; + + let (listener, client) = bind_and_connect().await; + let (accepted, peer) = listener.accept().await.expect("accept"); + assert_eq!(peer, client.local_addr().expect("client addr")); + assert!( + !accepted.nodelay().expect("read TCP_NODELAY"), + "a fresh accepted socket has Nagle on" + ); + let (stream, ()) = NodelayAcceptor + .accept(accepted, ()) + .await + .expect("the plaintext acceptor hands the socket on"); + assert!(stream.nodelay().expect("read TCP_NODELAY")); +} + +/// Read from `client` until EOF, an error, or `bound`. Returns `Ok(())` for +/// EOF or a reset, `Err(n)` for `n` unexpected bytes, and panics with `what` +/// when the server holds the connection past `bound`. +async fn expect_closed_within(client: &mut TcpStream, bound: Duration, what: &str) { + use tokio::io::AsyncReadExt; + + let mut buf = [0u8; 64]; + match tokio::time::timeout(bound, client.read(&mut buf)).await { + Ok(Ok(0)) | Ok(Err(_)) => {} + Ok(Ok(n)) => panic!("{what}: server answered {n} bytes instead of closing the connection"), + Err(_) => panic!("{what}: server held the connection open for {bound:?}"), + } +} + +/// The plaintext path: a connection that sends nothing after connecting is +/// closed once `request_timeout` passes. The header read timer never starts +/// for this shape, because the version sniff runs first. +#[tokio::test] +async fn plaintext_listener_closes_a_connection_that_sends_nothing() { + let timeout = Duration::from_millis(300); + let (listener, mut client) = bind_and_connect().await; + let app = Router::new().route("/", axum::routing::get(|| async { "ok" })); + let server = tokio::spawn(serve_app( + listener, + app, + timeout, + None, + None, + Admission::default(), + )); + + expect_closed_within(&mut client, timeout * 10, "plaintext, zero bytes").await; + server.abort(); +} + +/// The plaintext path: a client that sends 16 bytes of the HTTP/2 preface +/// and stops is still in the version sniff; it is closed at the deadline. +#[tokio::test] +async fn plaintext_listener_closes_a_partial_h2_preface() { + let timeout = Duration::from_millis(300); + let (listener, mut client) = bind_and_connect().await; + let app = Router::new().route("/", axum::routing::get(|| async { "ok" })); + let server = tokio::spawn(serve_app( + listener, + app, + timeout, + None, + None, + Admission::default(), + )); + + client + .write_all(&H2_PREFACE[..16]) + .await + .expect("write partial preface"); + expect_closed_within(&mut client, timeout * 10, "plaintext, 16 preface bytes").await; + server.abort(); +} + +/// Write a self-signed certificate for `localhost` and its key into `dir`, +/// returning the server's TLS config and the certificate's DER bytes for the +/// client's trust store. +fn self_signed_localhost(dir: &std::path::Path) -> (ServerTlsConfig, Vec) { + let params = rcgen::CertificateParams::new(vec!["localhost".to_owned()]).expect("params"); + let key_pair = rcgen::KeyPair::generate().expect("key pair"); + let cert = params.self_signed(&key_pair).expect("self-signed cert"); + let cert_path = dir.join("cert.pem"); + let key_path = dir.join("key.pem"); + std::fs::write(&cert_path, cert.pem()).expect("write cert"); + std::fs::write(&key_path, key_pair.serialize_pem()).expect("write key"); + ( + ServerTlsConfig { + cert_path, + key_path, + }, + cert.der().to_vec(), + ) +} + +/// Complete a TLS handshake with the server at `addr` as a client that +/// trusts `cert_der`, send `after_handshake` as application data, then read +/// until EOF, an error, or `bound`. Runs rustls over a blocking socket so +/// the test needs no async TLS client. Returns `Ok(())` for EOF or an error +/// from the server side, `Err(n)` for `n` application bytes received, and +/// panics when the bound passes with the connection still open. +fn tls_client_expect_closed_within( + addr: std::net::SocketAddr, + cert_der: Vec, + after_handshake: &'static [u8], + bound: Duration, + what: &'static str, +) { + // The rustls client needs the process-level provider as well. serve_app + // installs it in the TLS arm, but this client can reach the builder + // before the server task gets there, and ClientConfig::builder panics + // when no provider is installed. Installing it twice is not an error + // here: the second call reports that one is already set. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let mut roots = rustls::RootCertStore::empty(); + roots + .add(rustls::pki_types::CertificateDer::from(cert_der)) + .expect("trust the test certificate"); + let config = rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + let server_name = rustls::pki_types::ServerName::try_from("localhost").expect("server name"); + let mut conn = rustls::ClientConnection::new(std::sync::Arc::new(config), server_name) + .expect("client connection"); + let mut sock = std::net::TcpStream::connect(addr).expect("connect"); + sock.set_read_timeout(Some(bound)).expect("read timeout"); + while conn.is_handshaking() { + conn.complete_io(&mut sock).expect("TLS handshake"); + } + let mut tls = rustls::Stream::new(&mut conn, &mut sock); + if !after_handshake.is_empty() { + std::io::Write::write_all(&mut tls, after_handshake).expect("write application data"); + std::io::Write::flush(&mut tls).expect("flush"); + } + let mut buf = [0u8; 64]; + match std::io::Read::read(&mut tls, &mut buf) { + Ok(0) => {} + Ok(n) => panic!("{what}: server answered {n} bytes instead of closing the connection"), + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) => + { + panic!("{what}: server held the connection open for {bound:?}") + } + // A reset, a close_notify reported as an error, or an unexpected + // EOF all mean the server dropped the connection. + Err(_) => {} + } +} + +/// Start the TLS listener on a loopback port with `timeout`, returning its +/// address, the certificate to trust, the temp dir holding the key pair and +/// the server task. +async fn start_tls_listener( + timeout: Duration, +) -> ( + std::net::SocketAddr, + Vec, + std::path::PathBuf, + tokio::task::JoinHandle>, +) { + let dir = std::env::temp_dir().join(format!( + "extenddb-listener-test-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock") + .as_nanos() + )); + std::fs::create_dir_all(&dir).expect("temp dir"); + let (tls, cert_der) = self_signed_localhost(&dir); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind loopback"); + let addr = listener.local_addr().expect("local addr"); + let app = Router::new().route("/", axum::routing::get(|| async { "ok" })); + let server = tokio::spawn(serve_app( + listener, + app, + timeout, + Some(tls), + None, + Admission::default(), + )); + (addr, cert_der, dir, server) +} + +/// The TLS path: a client that completes the handshake and then sends no +/// application data is closed once `request_timeout` passes. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn tls_listener_closes_a_connection_that_sends_nothing_after_the_handshake() { + let timeout = Duration::from_millis(300); + let (addr, cert_der, dir, server) = start_tls_listener(timeout).await; + tokio::task::spawn_blocking(move || { + tls_client_expect_closed_within( + addr, + cert_der, + b"", + timeout * 10, + "TLS, zero bytes after the handshake", + ) + }) + .await + .expect("client task"); + server.abort(); + let _ = std::fs::remove_dir_all(dir); +} + +/// The TLS path: 16 bytes of the HTTP/2 preface after the handshake, then +/// silence, is closed at the deadline. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn tls_listener_closes_a_partial_h2_preface() { + let timeout = Duration::from_millis(300); + let (addr, cert_der, dir, server) = start_tls_listener(timeout).await; + tokio::task::spawn_blocking(move || { + tls_client_expect_closed_within( + addr, + cert_der, + &H2_PREFACE[..16], + timeout * 10, + "TLS, 16 preface bytes", + ) + }) + .await + .expect("client task"); + server.abort(); + let _ = std::fs::remove_dir_all(dir); +} + +/// The TLS path: a connection that sends no byte at all (not even a +/// ClientHello) is dropped by the redirect acceptor's peek at the deadline. +#[tokio::test] +async fn tls_acceptor_drops_a_connection_that_sends_nothing_before_the_handshake() { + use axum_server::accept::Accept; + + let timeout = Duration::from_millis(300); + let (listener, _client) = bind_and_connect().await; + let (accepted, _) = listener.accept().await.expect("accept"); + let acceptor = HttpsRedirectAcceptor { + addr: listener.local_addr().expect("local addr"), + first_byte_timeout: timeout, + }; + let outcome = tokio::time::timeout(timeout * 10, acceptor.accept(accepted, ())).await; + match outcome { + Ok(Err(e)) => assert_eq!(e.kind(), std::io::ErrorKind::TimedOut, "{e}"), + Ok(Ok(_)) => panic!("acceptor passed a connection that sent nothing"), + Err(_) => panic!( + "acceptor waited more than {:?} for the first byte", + timeout * 10 + ), + } +} + +/// The deadline is dropped once the first byte arrives: a client that sends +/// a complete request late, but before the deadline, is served. +#[tokio::test] +async fn plaintext_listener_serves_a_request_that_arrives_before_the_deadline() { + use tokio::io::AsyncReadExt; + + let timeout = Duration::from_millis(600); + let (listener, mut client) = bind_and_connect().await; + let app = Router::new().route("/", axum::routing::get(|| async { "ok" })); + let server = tokio::spawn(serve_app( + listener, + app, + timeout, + None, + None, + Admission::default(), + )); + + tokio::time::sleep(timeout / 3).await; + client + .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .await + .expect("write request"); + let mut response = Vec::new(); + tokio::time::timeout(timeout * 10, client.read_to_end(&mut response)) + .await + .expect("response within the bound") + .expect("read response"); + server.abort(); + let text = String::from_utf8_lossy(&response); + assert!(text.starts_with("HTTP/1.1 200"), "{text}"); + assert!(text.ends_with("ok"), "{text}"); +} + +/// The deadline is dropped exactly where the auto builder stops sniffing: a +/// byte that differs from the preface, or the whole preface; a prefix of it +/// keeps the deadline armed. +#[tokio::test] +async fn deadline_tracks_the_version_sniff() { + let mut adapter = FirstByteDeadline::new(tokio::io::empty(), Duration::from_secs(1)); + assert!(!adapter.sniff_decided(&H2_PREFACE[..16])); + assert!(!adapter.sniff_decided(&H2_PREFACE[16..23])); + assert!(adapter.sniff_decided(&H2_PREFACE[23..])); + + let mut adapter = FirstByteDeadline::new(tokio::io::empty(), Duration::from_secs(1)); + assert!(adapter.sniff_decided(b"G")); + + let mut adapter = FirstByteDeadline::new(tokio::io::empty(), Duration::from_secs(1)); + assert!(!adapter.sniff_decided(b"PRI")); + assert!(adapter.sniff_decided(b" + HTTP")); +} + +/// A stop that the registration supervisor asks for ends the plaintext +/// server through the graceful shutdown a signal starts. +#[tokio::test] +async fn a_stop_after_a_lost_registration_ends_the_server_as_a_signal_does() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind loopback"); + let admission = Admission::default(); + let app = Router::new().route("/", axum::routing::get(|| async { "served" })); + let server = tokio::spawn(serve_app( + listener, + app, + Duration::from_secs(5), + None, + None, + admission.clone(), + )); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(!server.is_finished(), "the server stopped by itself"); + admission.stop("the registration could not be taken again".to_owned()); + tokio::time::timeout(Duration::from_secs(10), server) + .await + .expect("the server stops within 10 s of the stop") + .expect("the server task") + .expect("the server"); + assert_eq!( + admission.stop_reason(), + Some("the registration could not be taken again") + ); +} diff --git a/crates/server/src/repeated_headers.rs b/crates/server/src/repeated_headers.rs new file mode 100644 index 000000000..e803ff32c --- /dev/null +++ b/crates/server/src/repeated_headers.rs @@ -0,0 +1,155 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Refuse a request that repeats a header the server reads as one value. +//! +//! A request carrying more than one `host` or `authorization` header is +//! refused before routing, as the service's front end does (400, connection +//! closed, no request id). Either duplicate is a header-confusion shape a +//! proxy in front of this server could resolve differently from the server: +//! the SigV4 check binds the host it signs, the console binds a session to a +//! host, and the management API reads one Basic credential. Applied as a +//! layer on the whole router so the DynamoDB route, `/console`, +//! `/management`, `/health` and `/metrics` all answer the same way. + +use axum::extract::Request; +use axum::http::{HeaderMap, StatusCode, header}; +use axum::middleware::Next; +use axum::response::{IntoResponse, Response}; + +/// Whether `headers` carries more than one `host` or `authorization` value. +fn has_repeated_host_or_authorization(headers: &HeaderMap) -> bool { + headers.get_all(header::HOST).iter().count() > 1 + || headers.get_all(header::AUTHORIZATION).iter().count() > 1 +} + +/// The refusal: 400 with the connection closed and a plain text body. +fn refusal() -> Response { + ( + StatusCode::BAD_REQUEST, + [(header::CONNECTION, "close")], + "400 Bad Request", + ) + .into_response() +} + +/// Middleware: refuse a request with a repeated `host` or `authorization` +/// header; pass every other request on. +pub(crate) async fn refuse_repeated_headers(request: Request, next: Next) -> Response { + if has_repeated_host_or_authorization(request.headers()) { + return refusal(); + } + next.run(request).await +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::Router; + use axum::body::Body; + use axum::routing::{get, post}; + use tower::ServiceExt; + + /// A router shaped like the server's: a data-plane route, a nested + /// console and a nested management API, with the guard layered on the + /// whole thing after nesting. + fn router() -> Router { + let console = Router::new().route("/login", get(|| async { "login" })); + let management = Router::new().route("/accounts", get(|| async { "accounts" })); + Router::new() + .route("/", post(|| async { "data plane" })) + .route("/health", get(|| async { "healthy" })) + .nest("/console", console) + .nest("/management", management) + .layer(axum::middleware::from_fn(refuse_repeated_headers)) + } + + async fn send( + headers: &[(&str, &str)], + method: &str, + path: &str, + ) -> (StatusCode, Option, String) { + let mut builder = axum::http::Request::builder().method(method).uri(path); + for (name, value) in headers { + builder = builder.header(*name, *value); + } + let response = router() + .oneshot(builder.body(Body::empty()).expect("request")) + .await + .expect("response"); + let status = response.status(); + let connection = response + .headers() + .get(header::CONNECTION) + .map(|v| v.to_str().expect("ascii").to_owned()); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("body"); + ( + status, + connection, + String::from_utf8(body.to_vec()).expect("utf8"), + ) + } + + #[tokio::test] + async fn a_repeated_host_is_refused_on_every_route() { + for (method, path) in [ + ("POST", "/"), + ("GET", "/health"), + ("GET", "/console/login"), + ("GET", "/management/accounts"), + ("GET", "/no-such-route"), + ] { + let (status, connection, body) = send( + &[("host", "a.example"), ("host", "b.example")], + method, + path, + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{method} {path}"); + assert_eq!(connection.as_deref(), Some("close"), "{method} {path}"); + assert_eq!(body, "400 Bad Request", "{method} {path}"); + } + } + + #[tokio::test] + async fn a_repeated_authorization_is_refused_on_every_route() { + for (method, path) in [ + ("POST", "/"), + ("GET", "/console/login"), + ("GET", "/management/accounts"), + ] { + let (status, connection, body) = send( + &[ + ("host", "a.example"), + ("authorization", "Basic YWRtaW46eA=="), + ("authorization", "Basic YWRtaW46eQ=="), + ], + method, + path, + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{method} {path}"); + assert_eq!(connection.as_deref(), Some("close"), "{method} {path}"); + assert_eq!(body, "400 Bad Request", "{method} {path}"); + } + } + + #[tokio::test] + async fn single_headers_and_repeated_other_headers_pass() { + let (status, _, body) = send( + &[ + ("host", "a.example"), + ("authorization", "Basic YWRtaW46eA=="), + ("x-amz-date", "20260101T000000Z"), + ("x-amz-date", "20260102T000000Z"), + ], + "GET", + "/console/login", + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body, "login"); + } +} diff --git a/crates/server/src/serve.rs b/crates/server/src/serve.rs index dc4acc2f4..089c647a4 100644 --- a/crates/server/src/serve.rs +++ b/crates/server/src/serve.rs @@ -231,6 +231,38 @@ impl std::fmt::Display for RegistrationLost { impl std::error::Error for RegistrationLost {} +/// The largest `request_timeout_secs` the server accepts. +/// +/// hyper adds the HTTP/1 header read timeout to `Instant::now()` once per +/// connection, and that addition panics on overflow, so a value near +/// `u64::MAX` panics the task of every connection the server accepts. A day +/// is longer than any pause a client takes between connecting and sending a +/// request, and it leaves the addition far from the overflow point. +const MAX_REQUEST_TIMEOUT_SECS: u64 = 86_400; + +/// Check the `request_timeout_secs` value from the `[server]` section. +/// +/// Returns the startup error message when the value is outside the accepted +/// range, so misconfiguration fails fast instead of reaching the listener. +fn check_request_timeout_secs(secs: u64) -> Result<(), String> { + if secs == 0 { + return Err( + "Invalid [server] configuration: request_timeout_secs must be at least 1 \ + (0 would close every connection before it could send a request)." + .to_owned(), + ); + } + if secs > MAX_REQUEST_TIMEOUT_SECS { + return Err(format!( + "Invalid [server] configuration: request_timeout_secs must be at most \ + {MAX_REQUEST_TIMEOUT_SECS} (a day); the per-connection timers add it to the \ + current instant, and a larger value overflows that addition and panics the \ + connection task." + )); + } + Ok(()) +} + /// Run the ExtendDB server on the pre-bound listener in `params` until shutdown. /// /// On any error before the HTTP server starts, and on [`RegistrationLost`] @@ -465,6 +497,9 @@ async fn serve_inner(params: ServeParams, port: u16) -> anyhow::Result<()> { soft_ttl_seconds, negative_ttl_seconds, max_entries." ); } + if let Err(e) = check_request_timeout_secs(app_config.server.request_timeout_secs) { + anyhow::bail!("{e}"); + } if !cache_enabled { tracing::warn!( "auth.cache.enabled = false — auth/authz caches are in pass-through mode \ @@ -483,9 +518,11 @@ async fn serve_inner(params: ServeParams, port: u16) -> anyhow::Result<()> { make_cache_cfg("credential"), ) }); - let auth: Arc = Arc::new( - extenddb_auth::BuiltinAuthProvider::new((*cached_cred_store).clone()), - ); + let auth: Arc = + Arc::new(extenddb_auth::BuiltinAuthProvider::new( + (*cached_cred_store).clone(), + app_config.server.region.as_str(), + )); // Phase 3: Build the authorization cache. let authz_cache: Arc = { @@ -693,6 +730,7 @@ async fn serve_inner(params: ServeParams, port: u16) -> anyhow::Result<()> { ), metrics: metrics.clone(), tls_enabled, + request_timeout: Duration::from_secs(app_config.server.request_timeout_secs), dev_mode, import_paths, export_paths, @@ -1061,6 +1099,9 @@ pub fn log_to_syslog_raw(msg: &str) { eprintln!("{msg}"); } +#[cfg(test)] +mod request_timeout_tests; + #[cfg(test)] mod tests { use std::sync::atomic::{AtomicU32, Ordering}; diff --git a/crates/server/src/serve/request_timeout_tests.rs b/crates/server/src/serve/request_timeout_tests.rs new file mode 100644 index 000000000..ae58d031f --- /dev/null +++ b/crates/server/src/serve/request_timeout_tests.rs @@ -0,0 +1,47 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Unit tests for the startup configuration checks. + +use super::*; + +#[test] +fn a_request_timeout_of_zero_is_refused() { + let err = check_request_timeout_secs(0).expect_err("0 is refused"); + assert!(err.contains("must be at least 1"), "{err}"); +} + +#[test] +fn a_request_timeout_above_a_day_is_refused() { + let err = + check_request_timeout_secs(MAX_REQUEST_TIMEOUT_SECS + 1).expect_err("86401 is refused"); + assert!(err.contains("must be at most 86400"), "{err}"); + + // The value that panicked every connection task before the ceiling. + let err = check_request_timeout_secs(u64::MAX).expect_err("u64::MAX is refused"); + assert!(err.contains("must be at most 86400"), "{err}"); +} + +#[test] +fn the_accepted_range_runs_from_one_second_to_a_day() { + for secs in [1, 30, 3600, MAX_REQUEST_TIMEOUT_SECS] { + check_request_timeout_secs(secs).unwrap_or_else(|e| panic!("{secs} is accepted: {e}")); + } +} + +/// The ceiling keeps the per-connection timers clear of the overflow that +/// made every connection task panic: the deadline hyper builds for a +/// connection is the current instant plus the timeout. +#[test] +fn the_ceiling_added_to_the_current_instant_does_not_overflow() { + let now = std::time::Instant::now(); + assert!( + now.checked_add(Duration::from_secs(MAX_REQUEST_TIMEOUT_SECS)) + .is_some(), + "the ceiling must be addable to an instant" + ); + assert!( + now.checked_add(Duration::from_secs(u64::MAX)).is_none(), + "u64::MAX seconds overflows, which is what the ceiling refuses" + ); +} diff --git a/docs/manuals/05-admin-guide.md b/docs/manuals/05-admin-guide.md index 280d59fdc..a7950514c 100755 --- a/docs/manuals/05-admin-guide.md +++ b/docs/manuals/05-admin-guide.md @@ -114,7 +114,8 @@ These settings require a server restart to take effect. |-----|---------|-------------| | `bind_addr` | `127.0.0.1` | Network interface to bind | | `port` | `18443` | HTTP port | -| `region` | `us-east-1` | AWS region for ARN generation. Baked into every backup ARN the server issues, so it must stay within `[a-z0-9-]` (under a value outside that class `CreateBackup` is refused with a `ValidationException` naming `server.region`, since no backup ARN can be formed; backups taken before the upgrade under such a region keep an ARN `DescribeBackup`, `DeleteBackup`, and `RestoreTableFromBackup` refuse as "not valid"; the server logs a warning at startup). A backup ARN of the caller's account in another region is resolved in the server's region first, then as written; see the upgrade manual before changing this on a deployment with backups | +| `region` | `us-east-1` | AWS region for ARN generation. Requests must carry a credential scope for this region; any other region is rejected with `InvalidSignatureException`, as the service rejects a scope for a region other than the endpoint's. Baked into every backup ARN the server issues, so it must stay within `[a-z0-9-]` (under a value outside that class `CreateBackup` is refused with a `ValidationException` naming `server.region`, since no backup ARN can be formed; backups taken before the upgrade under such a region keep an ARN `DescribeBackup`, `DeleteBackup`, and `RestoreTableFromBackup` refuse as "not valid"; the server logs a warning at startup). A backup ARN of the caller's account in another region is resolved in the server's region first, then as written; see the upgrade manual before changing this on a deployment with backups | +| `request_timeout_secs` | `30` | Seconds a connection may take to deliver its request headers, and seconds a request body may stall without delivering data, before the server closes it (a stalled body gets HTTP 408). A connection that is accepted, or that completes the TLS handshake, and then sends no byte is closed after the same number of seconds, as is one that sends only part of the 24-byte HTTP/2 preface. The same value bounds how long an idle keep-alive connection waits for its next request, so SDK connection pools see idle connections closed after it (they reopen transparently), and it is the HTTP/2 keep-alive PING interval and timeout, so a stalled HTTP/2 connection is closed within twice the value. Must be at least 1 and at most 86400 (a day); the server refuses a larger value at startup, because the per-connection timers add it to the current instant and a value near `u64::MAX` overflows that addition | #### [storage] diff --git a/docs/manuals/10-security-model.md b/docs/manuals/10-security-model.md index 5f1f6b943..1e4c93ec8 100755 --- a/docs/manuals/10-security-model.md +++ b/docs/manuals/10-security-model.md @@ -34,15 +34,19 @@ extenddb uses SigV4 signature verification with a local credential store and IAM #### SigV4 Verification -1. Extract the `Authorization` header (AWS4-HMAC-SHA256 scheme) -2. Parse credential scope, signed headers, and signature -3. Look up the access key in the credential store -4. Decrypt the secret key (AES-256-GCM) -5. Compute the signing key: HMAC-SHA256 chain over date, region, service, "aws4_request" -6. Compute the string-to-sign from the canonical request. The hashed-payload line is the SHA-256 of the body the server received; a client-supplied `x-amz-content-sha256` header is not used for it, so a request whose body was altered after signing fails at step 7 -7. Compare computed signature against the provided signature (constant-time comparison) -8. On mismatch: return `InvalidSignatureException` (HTTP 400) -9. On unknown key: return `UnrecognizedClientException` (HTTP 400) +The steps run in this order; every rejection is HTTP 400. + +1. Extract the `Authorization` header (AWS4-HMAC-SHA256 scheme); absent: `MissingAuthenticationToken` +2. Parse credential scope, signed headers, and signature; malformed: `IncompleteSignatureException`. A request with more than one `host` or `authorization` header is refused before this point with a plain 400, as the service's front end does +3. Look up the access key in the credential store (through the credential cache, see below); unknown: `UnrecognizedClientException` +4. Check `X-Amz-Date` against the server clock: more than 15 minutes of skew is `UnrecognizedClientException` (signature expired) +5. Check the credential scope: the region must be this server's `[server] region` and the service must be `dynamodb`; otherwise `InvalidSignatureException` carrying one sentence per failing part, region first, as the service reports it +6. Compare the session token: a temporary key must present its stored token, and a long-term key must present none; the decision is applied at step 9 +7. Reconstruct the canonical request and the string-to-sign. `host` must be among the signed headers. Repeated headers contribute their values comma-joined. The hashed-payload line is the SHA-256 of the body the server received; a client-supplied `x-amz-content-sha256` header is not used for it, so a request whose body was altered after signing, or signed with a literal such as `UNSIGNED-PAYLOAD`, fails at step 8 +8. Derive the signing key (HMAC-SHA256 chain over date, region, service, "aws4_request") and compare the computed signature with the provided one in constant time; mismatch: `InvalidSignatureException` +9. Apply the deferred decisions from steps 3 and 6, after the signature check so that failure paths take the same time: an inactive key or a wrong or missing token is `UnrecognizedClientException` + +Request bodies are parsed before authentication (the service does the same), so a malformed body returns `SerializationException` whatever the signature. Headers not named in `SignedHeaders` do not take part in verification; the service behaves the same way, and clients sign every `x-amz-*` header they send. #### Credential Types diff --git a/tests/test_auth_error_fidelity.py b/tests/test_auth_error_fidelity.py index 6b6602dd6..ffe37fe70 100755 --- a/tests/test_auth_error_fidelity.py +++ b/tests/test_auth_error_fidelity.py @@ -20,6 +20,7 @@ import json import os from typing import Any +from urllib.parse import urlparse import boto3 import pytest @@ -241,3 +242,371 @@ def test_unchanged_body_is_accepted(self, dynamodb_client, table): TableName=table, Key={"pk": {"S": "intact"}}, ConsistentRead=True ) assert got["Item"]["pk"]["S"] == "intact" + + +def _env_credentials() -> tuple[str, str, str | None]: + """Credentials to sign with: the environment first, then the SDK's chain.""" + access_key = os.environ.get("AWS_ACCESS_KEY_ID", "") + secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY", "") + if access_key and secret_key: + return access_key, secret_key, os.environ.get("AWS_SESSION_TOKEN") or None + resolved = boto3.Session().get_credentials() + if resolved is None: + pytest.skip("no credentials available to sign the request") + frozen = resolved.get_frozen_credentials() + return frozen.access_key, frozen.secret_key, frozen.token + + +class TestRequestChecks: + """Checks on the signed request other than the signature itself. + + Each expected message was measured against the service on 2026-09-16. + """ + + @pytest.fixture(autouse=True) + def setup(self, endpoint_url): + if endpoint_url and not os.environ.get("EXTENDDB_ADMIN_USER", "").strip(): + pytest.fail( + "MISCONFIGURED: request check tests require auth-enabled extenddb " + "(set EXTENDDB_ADMIN_USER to signal builtin auth mode)." + ) + self.endpoint_url = endpoint_url + self.region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1") + self.url = endpoint_url or f"https://dynamodb.{self.region}.amazonaws.com/" + + def _post(self, body: dict, *, headers: dict | None = None, region: str | None = None, + token: str | None = "env") -> requests.Response: + """Sign and send one ListTables-style request with botocore's signer. + + ``headers`` are added before signing so they are covered. ``region`` + overrides the credential scope. ``token`` "env" uses the environment's + session token, if any; None sends no token; a string sends that token. + """ + access_key, secret_key, env_token = _env_credentials() + if token == "env": + token = env_token + body_bytes = json.dumps(body).encode("utf-8") + hdrs = { + "X-Amz-Target": "DynamoDB_20120810.ListTables", + "Content-Type": "application/x-amz-json-1.0", + } + hdrs.update(headers or {}) + aws_req = AWSRequest(method="POST", url=self.url, data=body_bytes, headers=hdrs) + SigV4Auth(Credentials(access_key, secret_key, token), "dynamodb", region or self.region).add_auth(aws_req) + return requests.post( + self.url, + data=body_bytes, + headers=dict(aws_req.headers), + verify=self.endpoint_url is None, + ) + + def test_scope_for_another_region_is_rejected(self): + other = "us-west-2" if self.region != "us-west-2" else "us-east-1" + resp = self._post({}, region=other) + assert resp.status_code == 400, resp.text + assert resp.json()["__type"].endswith("InvalidSignatureException"), resp.text + assert resp.json()["message"] == "Credential should be scoped to a valid region. " + + def test_scope_wrong_in_region_and_service_reports_both(self): + access_key, secret_key, token = _env_credentials() + other = "us-west-2" if self.region != "us-west-2" else "us-east-1" + body_bytes = json.dumps({}).encode("utf-8") + aws_req = AWSRequest( + method="POST", + url=self.url, + data=body_bytes, + headers={ + "X-Amz-Target": "DynamoDB_20120810.ListTables", + "Content-Type": "application/x-amz-json-1.0", + }, + ) + SigV4Auth(Credentials(access_key, secret_key, token), "s3", other).add_auth(aws_req) + resp = requests.post(self.url, data=body_bytes, headers=dict(aws_req.headers), verify=self.endpoint_url is None) + assert resp.status_code == 400, resp.text + assert resp.json()["__type"].endswith("InvalidSignatureException"), resp.text + assert resp.json()["message"] == ( + "Credential should be scoped to a valid region. " + "Credential should be scoped to correct service: 'dynamodb'. " + ) + + def test_unknown_key_with_another_region_reports_the_scope(self): + # Measured against the service on 2026-10-23: the scope is checked + # before the access key is looked up, so an unknown key scoped to + # another region gets the scope message, not the invalid-key error. + other = "us-west-2" if self.region != "us-west-2" else "us-east-1" + body_bytes = json.dumps({}).encode("utf-8") + aws_req = AWSRequest( + method="POST", + url=self.url, + data=body_bytes, + headers={ + "X-Amz-Target": "DynamoDB_20120810.ListTables", + "Content-Type": "application/x-amz-json-1.0", + }, + ) + SigV4Auth(Credentials("AKIAEXAMPLEEXAMPLE01", "x" * 40, None), "dynamodb", other).add_auth(aws_req) + resp = requests.post(self.url, data=body_bytes, headers=dict(aws_req.headers), verify=self.endpoint_url is None) + assert resp.status_code == 400, resp.text + assert resp.json()["__type"].endswith("InvalidSignatureException"), resp.text + assert resp.json()["message"] == "Credential should be scoped to a valid region. " + + def test_duplicate_host_header_is_refused(self): + # The service's front end refuses a request with two host headers + # before it reaches signing (measured 2026-10-23 on POST /: 400, + # Connection: close, and an HTML error page instead of a JSON + # error). The body assertion runs against extenddb only, and it is + # what separates the listener's refusal from the handler's + # MissingAuthenticationToken, which answers 400 as well. + u = urlparse(self.url) + status, text, connection = self._raw_request( + "POST", + "/", + [ + ("Host", u.netloc), + ("Host", "evil.example.com"), + ("X-Amz-Target", "DynamoDB_20120810.ListTables"), + ("Content-Type", "application/x-amz-json-1.0"), + ], + body=b"{}", + ) + assert status == 400, (status, text[:200]) + assert connection == "close", connection + if self.endpoint_url is not None: + assert text == "400 Bad Request", text[:200] + + def _raw_request(self, method: str, path: str, headers: list[tuple[str, str]], body: bytes = b""): + """Send one request with each header on its own line, repeats included. + + Returns (status, body text, Connection header). http.client sends the + headers in the order given, so a repeated name goes out twice. + """ + import http.client # noqa: PLC0415 + import ssl # noqa: PLC0415 + from urllib.parse import urlparse # noqa: PLC0415 + + u = urlparse(self.url) + ctx = ssl.create_default_context() + if self.endpoint_url is not None: + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + conn = http.client.HTTPSConnection(u.hostname, u.port or 443, context=ctx, timeout=30) + conn.putrequest(method, path, skip_host=True, skip_accept_encoding=True) + for k, v in headers: + conn.putheader(k, v) + if body or method == "POST": + conn.putheader("Content-Length", str(len(body))) + conn.endheaders() + if body: + conn.send(body) + resp = conn.getresponse() + text = resp.read().decode(errors="replace") + connection = resp.getheader("Connection") + conn.close() + return resp.status, text, connection + + @pytest.mark.parametrize("path", ["/console/login", "/management/accounts", "/health"]) + def test_duplicate_host_header_is_refused_on_every_route(self, path): + # The refusal is a property of the listener, not of the DynamoDB + # route: the console, the management API and the health check get + # the same 400 with the connection closed. + if self.endpoint_url is None: + pytest.skip("console and management routes exist on extenddb only") + u = urlparse(self.url) + status, text, connection = self._raw_request( + "GET", path, [("Host", u.netloc), ("Host", "evil.example.com")] + ) + assert status == 400, (path, status, text[:200]) + assert text == "400 Bad Request", (path, text[:200]) + assert connection == "close", (path, connection) + + @pytest.mark.parametrize("path", ["/console/login", "/management/accounts", "/health"]) + def test_duplicate_authorization_header_is_refused_on_every_route(self, path): + if self.endpoint_url is None: + pytest.skip("console and management routes exist on extenddb only") + u = urlparse(self.url) + status, text, connection = self._raw_request( + "GET", + path, + [ + ("Host", u.netloc), + ("Authorization", "Basic YWRtaW46eA=="), + ("Authorization", "Basic YWRtaW46eQ=="), + ], + ) + assert status == 400, (path, status, text[:200]) + assert text == "400 Bad Request", (path, text[:200]) + assert connection == "close", (path, connection) + + def test_duplicate_authorization_header_is_refused_on_the_data_plane(self): + # Measured against the service on 2026-10-23: two Authorization + # headers get the front end's 400 Bad Request with the connection + # closed, before authentication runs. + u = urlparse(self.url) + auth = ( + "AWS4-HMAC-SHA256 Credential=AKIAEXAMPLEEXAMPLE01/20260101/us-east-1/dynamodb/aws4_request, " + "SignedHeaders=host;x-amz-date;x-amz-target, Signature=" + "0" * 64 + ) + status, text, connection = self._raw_request( + "POST", + "/", + [ + ("Host", u.netloc), + ("X-Amz-Target", "DynamoDB_20120810.ListTables"), + ("Content-Type", "application/x-amz-json-1.0"), + ("X-Amz-Date", "20260101T000000Z"), + ("Authorization", auth), + ("Authorization", auth), + ], + body=b"{}", + ) + assert status == 400, (status, text[:200]) + assert connection == "close", connection + if self.endpoint_url is not None: + assert text == "400 Bad Request", text[:200] + + def test_session_token_with_a_long_term_key_is_rejected(self): + _, _, env_token = _env_credentials() + if env_token: + pytest.skip("the environment's credentials are temporary; this case needs a long-term key") + resp = self._post({}, token="AQoDYXdzEJr-not-a-real-token") + assert resp.status_code == 400, resp.text + assert resp.json()["__type"].endswith("UnrecognizedClientException"), resp.text + assert resp.json()["message"] == "The security token included in the request is invalid" + + def test_repeated_signed_header_is_canonicalized_comma_joined(self): + # botocore joins repeated header values with commas when signing; the + # server must canonicalize the same way, or every such request fails. + access_key, secret_key, token = _env_credentials() + body_bytes = json.dumps({}).encode("utf-8") + aws_req = AWSRequest( + method="POST", + url=self.url, + data=body_bytes, + headers={ + "X-Amz-Target": "DynamoDB_20120810.ListTables", + "Content-Type": "application/x-amz-json-1.0", + }, + ) + aws_req.headers.add_header("X-Amz-Meta-Dup", "first") + aws_req.headers.add_header("X-Amz-Meta-Dup", "second") + SigV4Auth(Credentials(access_key, secret_key, token), "dynamodb", self.region).add_auth(aws_req) + # Send every header as its own line, including both X-Amz-Meta-Dup values. + prepared_headers = list(aws_req.headers.items()) + import http.client # noqa: PLC0415 + import ssl # noqa: PLC0415 + from urllib.parse import urlparse # noqa: PLC0415 + + u = urlparse(self.url) + ctx = ssl.create_default_context() + if self.endpoint_url is not None: + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + conn = http.client.HTTPSConnection(u.hostname, u.port or 443, context=ctx, timeout=30) + conn.putrequest("POST", "/", skip_host=True, skip_accept_encoding=True) + conn.putheader("Host", u.netloc) + for k, v in prepared_headers: + conn.putheader(k, v) + conn.putheader("Content-Length", str(len(body_bytes))) + conn.endheaders() + conn.send(body_bytes) + resp = conn.getresponse() + text = resp.read().decode() + conn.close() + assert resp.status == 200, text + + @pytest.mark.parametrize( + "name,second_value", + [("X-Amz-Date", "20260101T000000Z"), ("X-Amz-Target", "DynamoDB_20120810.ListTables")], + ) + def test_signed_header_repeated_after_signing_is_a_signature_mismatch(self, name, second_value): + # Measured against the service on 2026-10-23: a request signed over one + # X-Amz-Date or X-Amz-Target value and sent with that header twice gets + # InvalidSignatureException with a request id. The front end does not + # refuse these; the canonical headers join both values, so the + # signature no longer matches. The second X-Amz-Target value is the + # same string as the first. + access_key, secret_key, token = _env_credentials() + body_bytes = json.dumps({"Limit": 1}).encode("utf-8") + aws_req = AWSRequest( + method="POST", + url=self.url, + data=body_bytes, + headers={ + "X-Amz-Target": "DynamoDB_20120810.ListTables", + "Content-Type": "application/x-amz-json-1.0", + }, + ) + SigV4Auth(Credentials(access_key, secret_key, token), "dynamodb", self.region).add_auth(aws_req) + u = urlparse(self.url) + headers = [("Host", u.netloc)] + headers += [(k, v) for k, v in aws_req.headers.items() if k.lower() != "host"] + headers.append((name, second_value)) + status, text, _ = self._raw_request("POST", "/", headers, body=body_bytes) + assert status == 400, (status, text[:300]) + payload = json.loads(text) + assert payload["__type"].endswith("InvalidSignatureException"), text + assert payload["message"].startswith("The request signature we calculated does not match"), text + + def test_unsigned_payload_literal_is_a_plain_signature_mismatch(self): + # A client that signs the payload line as the S3 literal fails with the + # ordinary mismatch: the service does not special-case the literal. + access_key, secret_key, token = _env_credentials() + body_bytes = json.dumps({}).encode("utf-8") + aws_req = AWSRequest( + method="POST", + url=self.url, + data=body_bytes, + headers={ + "X-Amz-Target": "DynamoDB_20120810.ListTables", + "Content-Type": "application/x-amz-json-1.0", + "X-Amz-Content-SHA256": "UNSIGNED-PAYLOAD", + }, + ) + signer = SigV4Auth(Credentials(access_key, secret_key, token), "dynamodb", self.region) + signer.payload = lambda request: "UNSIGNED-PAYLOAD" # sign the literal, as such a client does + signer.add_auth(aws_req) + resp = requests.post(self.url, data=body_bytes, headers=dict(aws_req.headers), verify=self.endpoint_url is None) + assert resp.status_code == 400, resp.text + assert resp.json()["__type"].endswith("InvalidSignatureException"), resp.text + assert resp.json()["message"].startswith("The request signature we calculated does not match"), resp.text + assert "UNSIGNED-PAYLOAD" not in resp.json()["message"] + + def test_stalled_body_is_closed_by_the_server(self): + """A Content-Length the client never delivers must not hold the connection. + + Runs against extenddb only. The server's `request_timeout_secs` bounds the + wait; the test allows that plus ten seconds, reading the value from + EXTENDDB_REQUEST_TIMEOUT_SECS (default 30). + """ + if self.endpoint_url is None: + pytest.skip("timeout behavior is a server property; not measured against the service") + import http.client # noqa: PLC0415 + import socket # noqa: PLC0415 + import ssl # noqa: PLC0415 + import time # noqa: PLC0415 + from urllib.parse import urlparse # noqa: PLC0415 + + timeout = int(os.environ.get("EXTENDDB_REQUEST_TIMEOUT_SECS", "30")) + u = urlparse(self.url) + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + conn = http.client.HTTPSConnection(u.hostname, u.port or 443, context=ctx, timeout=timeout + 10) + conn.putrequest("POST", "/", skip_host=True, skip_accept_encoding=True) + conn.putheader("Host", f"{u.hostname}:{u.port or 443}") + conn.putheader("X-Amz-Target", "DynamoDB_20120810.ListTables") + conn.putheader("Content-Type", "application/x-amz-json-1.0") + conn.putheader("Content-Length", "100") + conn.endheaders() + conn.send(b"{}") # 2 of the announced 100 bytes; nothing more follows + started = time.monotonic() + try: + resp = conn.getresponse() + status = resp.status + resp.read() + except (http.client.RemoteDisconnected, ConnectionResetError, socket.timeout) as e: + status = type(e).__name__ + elapsed = time.monotonic() - started + conn.close() + assert status in (408, "RemoteDisconnected", "ConnectionResetError"), status + assert elapsed < timeout + 10, f"server held the stalled connection for {elapsed:.0f}s"