From a4f3e719c9bf2c7d0312143c51a1ccaa5fabbd29 Mon Sep 17 00:00:00 2001 From: Christian Date: Tue, 10 Nov 2026 20:22:43 -0600 Subject: [PATCH 1/3] Extract native EC lifecycle and DynamoDB persistence --- Cargo.lock | 128 ++- crates/trusted-server-adapter-axum/Cargo.toml | 3 + crates/trusted-server-adapter-axum/src/app.rs | 370 +++++-- .../trusted-server-adapter-axum/src/ec_kv.rs | 696 +++++++++++++ .../src/ec_kv_tests.rs | 811 +++++++++++++++ crates/trusted-server-adapter-axum/src/lib.rs | 2 + .../src/startup.rs | 12 +- .../tests/ec_kv_configuration.rs | 105 ++ .../tests/native_ec.rs | 524 ++++++++++ .../src/app.rs | 1 + .../trusted-server-adapter-fastly/src/app.rs | 1 + crates/trusted-server-adapter-spin/src/app.rs | 1 + .../trusted-server-core/src/ec/kv_backend.rs | 6 + .../src/integrations/registry.rs | 15 +- crates/trusted-server-core/src/publisher.rs | 4 +- docs/guide/native-ec-persistence.md | 136 +++ scripts/test-native-ec-image.mjs | 981 ++++++++++++++++++ 17 files changed, 3666 insertions(+), 130 deletions(-) create mode 100644 crates/trusted-server-adapter-axum/src/ec_kv.rs create mode 100644 crates/trusted-server-adapter-axum/src/ec_kv_tests.rs create mode 100644 crates/trusted-server-adapter-axum/tests/ec_kv_configuration.rs create mode 100644 crates/trusted-server-adapter-axum/tests/native_ec.rs create mode 100644 docs/guide/native-ec-persistence.md create mode 100644 scripts/test-native-ec-image.mjs diff --git a/Cargo.lock b/Cargo.lock index cdc1a28f3..0ee33ae6e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -310,7 +310,7 @@ dependencies = [ "bytes", "fastrand", "hex", - "http", + "http 1.4.2", "sha1", "time", "tokio", @@ -371,7 +371,7 @@ dependencies = [ "bytes", "bytes-utils", "fastrand", - "http", + "http 1.4.2", "http-body", "percent-encoding", "pin-project-lite", @@ -379,6 +379,33 @@ dependencies = [ "uuid", ] +[[package]] +name = "aws-sdk-dynamodb" +version = "1.120.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f578638cafe182113da770e9798e7063d6de96cbbfe06f647d0026f197c9b97c" +dependencies = [ + "arc-swap", + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-observability", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema 0.2.1", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 0.2.12", + "http 1.4.2", + "regex-lite", + "tracing", + "url", +] + [[package]] name = "aws-sdk-secretsmanager" version = "1.120.0" @@ -399,7 +426,7 @@ dependencies = [ "aws-types", "bytes", "fastrand", - "http", + "http 1.4.2", "regex-lite", "tracing", ] @@ -424,7 +451,7 @@ dependencies = [ "aws-types", "bytes", "fastrand", - "http", + "http 1.4.2", "regex-lite", "tracing", ] @@ -449,7 +476,7 @@ dependencies = [ "aws-types", "bytes", "fastrand", - "http", + "http 1.4.2", "regex-lite", "tracing", ] @@ -475,7 +502,7 @@ dependencies = [ "aws-smithy-xml", "aws-types", "fastrand", - "http", + "http 1.4.2", "regex-lite", "tracing", ] @@ -494,7 +521,7 @@ dependencies = [ "form_urlencoded", "hex", "hmac 0.13.0", - "http", + "http 1.4.2", "percent-encoding", "sha2 0.11.1", "time", @@ -524,7 +551,7 @@ dependencies = [ "bytes-utils", "futures-core", "futures-util", - "http", + "http 1.4.2", "http-body", "http-body-util", "percent-encoding", @@ -542,7 +569,7 @@ dependencies = [ "aws-smithy-runtime-api", "aws-smithy-types", "h2", - "http", + "http 1.4.2", "hyper", "hyper-rustls", "hyper-util", @@ -604,7 +631,7 @@ dependencies = [ "aws-smithy-types", "bytes", "fastrand", - "http", + "http 1.4.2", "http-body", "http-body-util", "pin-project-lite", @@ -622,7 +649,7 @@ dependencies = [ "aws-smithy-runtime-api-macros", "aws-smithy-types", "bytes", - "http", + "http 1.4.2", "pin-project-lite", "tokio", "tracing", @@ -648,7 +675,7 @@ checksum = "e8f395d93304280b64b7632fea798d177e74897fe7f063416ce627cd6fa24829" dependencies = [ "aws-smithy-runtime-api", "aws-smithy-types", - "http", + "http 1.4.2", ] [[package]] @@ -659,7 +686,7 @@ checksum = "1d7a3b01a2a1b453176457b260b0ed55d33de61d35d21b74022c51ec62a42895" dependencies = [ "aws-smithy-runtime-api", "aws-smithy-types", - "http", + "http 1.4.2", ] [[package]] @@ -672,7 +699,7 @@ dependencies = [ "bytes", "bytes-utils", "futures-core", - "http", + "http 1.4.2", "http-body", "http-body-util", "itoa", @@ -722,7 +749,7 @@ dependencies = [ "bytes", "form_urlencoded", "futures-util", - "http", + "http 1.4.2", "http-body", "http-body-util", "hyper", @@ -753,7 +780,7 @@ checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" dependencies = [ "bytes", "futures-core", - "http", + "http 1.4.2", "http-body", "http-body-util", "mime", @@ -866,7 +893,7 @@ dependencies = [ "futures-util", "hex", "home", - "http", + "http 1.4.2", "http-body-util", "hyper", "hyper-named-pipe", @@ -1993,7 +2020,7 @@ dependencies = [ "futures", "futures-util", "getrandom 0.4.2", - "http", + "http 1.4.2", "http-body", "hyper", "hyper-util", @@ -2146,7 +2173,7 @@ dependencies = [ "flate2", "futures", "futures-util", - "http", + "http 1.4.2", "http-body", "log", "matchit 0.9.2", @@ -2386,7 +2413,7 @@ dependencies = [ "fastly-macros", "fastly-shared", "fastly-sys", - "http", + "http 1.4.2", "itertools 0.13.0", "mime", "serde", @@ -2418,7 +2445,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "191c2cf082f5d17f4ca0685cedaa0642ad4ade949fac8d1dbdd62ff756b13500" dependencies = [ "bitflags 1.3.2", - "http", + "http 1.4.2", ] [[package]] @@ -2429,7 +2456,7 @@ checksum = "239c77706cf9110b423a89b6b03b8011729b144390d2a7f641d95ef66b2768b8" dependencies = [ "bitflags 1.3.2", "fastly-shared", - "http", + "http 1.4.2", "wasip2", "wit-bindgen 0.51.0", ] @@ -3093,7 +3120,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c91d8cffac8849493a82233811bd02b2b183b8cf39bf704de0fa0841b737595" dependencies = [ "bstr", - "hashbrown 0.16.1", + "hashbrown 0.17.1", ] [[package]] @@ -3604,7 +3631,7 @@ dependencies = [ "fnv", "futures-core", "futures-sink", - "http", + "http 1.4.2", "indexmap 2.14.0", "slab", "tokio", @@ -3772,6 +3799,17 @@ dependencies = [ "match_token", ] +[[package]] +name = "http" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "601cbb57e577e2f5ef5be8e7b83f0f63994f25aa94d673e54a92d5c516d101f1" +dependencies = [ + "bytes", + "fnv", + "itoa", +] + [[package]] name = "http" version = "1.4.2" @@ -3789,7 +3827,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" dependencies = [ "bytes", - "http", + "http 1.4.2", ] [[package]] @@ -3800,7 +3838,7 @@ checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" dependencies = [ "bytes", "futures-core", - "http", + "http 1.4.2", "http-body", "pin-project-lite", ] @@ -3837,7 +3875,7 @@ dependencies = [ "futures-channel", "futures-core", "h2", - "http", + "http 1.4.2", "http-body", "httparse", "httpdate", @@ -3869,7 +3907,7 @@ version = "0.27.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" dependencies = [ - "http", + "http 1.4.2", "hyper", "hyper-util", "rustls", @@ -3903,7 +3941,7 @@ dependencies = [ "bytes", "futures-channel", "futures-util", - "http", + "http 1.4.2", "http-body", "hyper", "ipnet", @@ -5315,7 +5353,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck", - "itertools 0.10.5", + "itertools 0.13.0", "log", "multimap", "once_cell", @@ -5335,7 +5373,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.13.0", "proc-macro2", "quote", "syn 2.0.118", @@ -5348,7 +5386,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.13.0", "proc-macro2", "quote", "syn 2.0.118", @@ -5652,7 +5690,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "http", + "http 1.4.2", "http-body", "http-body-util", "hyper", @@ -5692,7 +5730,7 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "http", + "http 1.4.2", "http-body", "http-body-util", "hyper", @@ -6447,7 +6485,7 @@ dependencies = [ "anyhow", "bytes", "futures", - "http", + "http 1.4.2", "http-body", "http-body-util", "spin-macro", @@ -6941,7 +6979,7 @@ dependencies = [ "base64", "bytes", "h2", - "http", + "http 1.4.2", "http-body", "http-body-util", "hyper", @@ -7015,7 +7053,7 @@ dependencies = [ "bytes", "futures-core", "futures-util", - "http", + "http 1.4.2", "http-body", "http-body-util", "pin-project-lite", @@ -7076,6 +7114,8 @@ name = "trusted-server-adapter-axum" version = "0.1.0" dependencies = [ "async-trait", + "aws-config", + "aws-sdk-dynamodb", "axum", "base64", "edgezero-adapter-axum", @@ -7184,7 +7224,7 @@ dependencies = [ "futures", "gix", "glob", - "http", + "http 1.4.2", "http-body-util", "hyper", "hyper-util", @@ -7243,7 +7283,7 @@ dependencies = [ "glob", "hex", "hmac 0.12.1", - "http", + "http 1.4.2", "httpdate", "iab_gpp", "jose-jwk", @@ -7282,7 +7322,7 @@ dependencies = [ "edgezero-core", "env_logger", "error-stack", - "http", + "http 1.4.2", "http-body-util", "libc", "log", @@ -7341,7 +7381,7 @@ checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" dependencies = [ "bytes", "data-encoding", - "http", + "http 1.4.2", "httparse", "log", "rand 0.9.4", @@ -7461,7 +7501,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c" dependencies = [ "base64", - "http", + "http 1.4.2", "httparse", "log", ] @@ -7626,7 +7666,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "75e9ed91ee5a995b6fe16c89d4511a8c12eb76967be20b57557618e02e781bcc" dependencies = [ "bytes", - "http", + "http 1.4.2", "http-body", "thiserror 2.0.18", "wit-bindgen 0.57.1", @@ -8307,7 +8347,7 @@ dependencies = [ "chrono", "futures-channel", "futures-util", - "http", + "http 1.4.2", "http-body", "js-sys", "matchit 0.7.3", diff --git a/crates/trusted-server-adapter-axum/Cargo.toml b/crates/trusted-server-adapter-axum/Cargo.toml index e204ae90b..d31db3539 100644 --- a/crates/trusted-server-adapter-axum/Cargo.toml +++ b/crates/trusted-server-adapter-axum/Cargo.toml @@ -22,10 +22,13 @@ path = "src/main.rs" default = [] aws-appconfig-agent = ["edgezero-adapter-axum/aws-appconfig-agent"] aws-secrets-manager = ["edgezero-adapter-axum/aws-secrets-manager"] +aws-dynamodb-ec = ["dep:aws-config", "dep:aws-sdk-dynamodb"] geo-cloudfront = ["edgezero-adapter-axum/geo-cloudfront"] [dependencies] async-trait = { workspace = true } +aws-config = { version = "=1.12.0", default-features = false, features = ["default-https-client", "rt-tokio", "credentials-process", "sso"], optional = true } +aws-sdk-dynamodb = { version = "=1.120.0", default-features = false, features = ["default-https-client", "rt-tokio"], optional = true } edgezero-adapter-axum = { workspace = true, features = ["axum"] } edgezero-core = { workspace = true } error-stack = { workspace = true } diff --git a/crates/trusted-server-adapter-axum/src/app.rs b/crates/trusted-server-adapter-axum/src/app.rs index 3aeca800e..8126a4089 100644 --- a/crates/trusted-server-adapter-axum/src/app.rs +++ b/crates/trusted-server-adapter-axum/src/app.rs @@ -17,13 +17,24 @@ use trusted_server_core::auction::{ }; use trusted_server_core::cache_policy::EdgeCacheHeader; use trusted_server_core::config_payload::{DEFAULT_CONFIG_STORE_ID, DEFAULT_SECRET_STORE_ID}; -use trusted_server_core::ec::EcContext; +use trusted_server_core::constants::{COOKIE_SHAREDID, COOKIE_TS_EIDS}; use trusted_server_core::ec::admin::{ admin_ec_lookup_not_supported, deny_admin_diagnostic_fallback, handle_admin_eids_lookup, }; +use trusted_server_core::ec::consent::ec_consent_withdrawn; +use trusted_server_core::ec::device::DeviceSignals; +use trusted_server_core::ec::finalize::ec_finalize_response; +use trusted_server_core::ec::identify::{cors_preflight_identify, handle_identify}; +use trusted_server_core::ec::kv::KvIdentityGraph; +use trusted_server_core::ec::kv_backend::EcKvStore; use trusted_server_core::ec::registry::PartnerRegistry; +use trusted_server_core::ec::{EcContext, EidSyncSource}; use trusted_server_core::error::{IntoHttpResponse as _, TrustedServerError}; -use trusted_server_core::integrations::{IntegrationRegistry, ProxyDispatchInput}; +use trusted_server_core::http_util::is_navigation_request; +use trusted_server_core::integrations::{ + IntegrationRegistry, ProxyDispatchInput, RequestFilterRegistryInput, + RequestFilterRegistryOutcome, +}; use trusted_server_core::proxy::{ handle_first_party_click, handle_first_party_proxy, handle_first_party_proxy_rebuild, handle_first_party_proxy_sign, @@ -52,6 +63,8 @@ pub struct AppState { orchestrator: Arc, registry: Arc, services: Option, + partners: PartnerRegistry, + ec_graph: Option KvIdentityGraph + Send + Sync>>, } /// Build the application state from explicit settings. @@ -74,12 +87,15 @@ fn build_state_with_services( plan.validate_for_target(trusted_server_core::platform::AuctionTargetId::Axum)?; let orchestrator = build_orchestrator_with_plan(Arc::clone(&plan), &settings)?; let registry = IntegrationRegistry::with_plan(&settings, plan)?; + let partners = PartnerRegistry::from_config(&settings.ec.partners)?; Ok(Arc::new(AppState { settings: Arc::new(settings), orchestrator: Arc::new(orchestrator), registry: Arc::new(registry), services, + partners, + ec_graph: None, })) } @@ -90,6 +106,17 @@ impl AppState { }) } + pub(crate) fn set_ec_store(&mut self, store: S) + where + S: EcKvStore + Clone + Send + Sync + 'static, + { + self.ec_graph = Some(Arc::new(move || KvIdentityGraph::new(store.clone()))); + } + + fn identity_graph(&self) -> Option { + self.ec_graph.as_ref().map(|factory| factory()) + } + fn services_for_request(&self, ctx: &RequestContext) -> Result { if let Some(services) = &self.services { return Ok(services.clone()); @@ -188,9 +215,13 @@ where /// Unknown, which fails the auction consent gate closed even for consented /// users. Geo comes from the immutable managed request snapshot. Off, untrusted /// or unavailable results remain Unknown unless real consent supplies it. A -/// malformed consent string is logged and falls back to the default -/// (fail-closed) context rather than being silently swallowed. -fn build_ec_context(state: &AppState, services: &RuntimeServices, req: &Request) -> EcContext { +/// malformed EC cookie fails closed before dispatch rather than silently +/// substituting a context. Consent decoding remains owned by core. +fn build_ec_context( + state: &AppState, + services: &RuntimeServices, + req: &Request, +) -> Result> { let geo_info = services .geo() .lookup(services.client_info().client_ip) @@ -199,10 +230,89 @@ fn build_ec_context(state: &AppState, services: &RuntimeServices, req: &Request) None }); EcContext::read_from_request_with_geo(&state.settings, req, services, geo_info.as_ref()) - .unwrap_or_else(|e| { - log::warn!("EC context read failed: {e:?}"); - EcContext::default() +} + +/// Native has no trusted TLS/JA4 fingerprint from CloudFront/ALB. Classify only +/// obvious browser traffic from UA, then require checked client metadata and +/// shared navigation/consent gates before issuance. This is a heuristic, not +/// authentication, and forwarded fingerprint headers are deliberately ignored. +fn native_browser_signals(req: &Request) -> (DeviceSignals, bool) { + let ua = req + .headers() + .get(header::USER_AGENT) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default(); + let signals = DeviceSignals::derive(ua, None, None); + let browser = signals.platform_class.is_some() + && ["Firefox/", "Chrome/", "Safari/"] + .iter() + .any(|token| ua.contains(token)) + && !ua.to_ascii_lowercase().contains("bot"); + (signals, browser) +} + +fn cookie_value(req: &Request, name: &str) -> Option { + req.headers() + .get_all(header::COOKIE) + .iter() + .filter_map(|value| value.to_str().ok()) + .flat_map(|value| value.split(';')) + .filter_map(|pair| pair.trim().split_once('=')) + .find(|(key, _)| key.trim() == name) + .map(|(_, value)| value.trim().to_owned()) +} + +struct NativeEcRequest { + context: EcContext, + graph: Option, + browser: bool, + eids_cookie: Option, + sharedid_cookie: Option, +} + +impl NativeEcRequest { + fn read( + state: &AppState, + services: &RuntimeServices, + req: &Request, + ) -> Result> { + let mut context = build_ec_context(state, services, req)?; + let (signals, browser) = native_browser_signals(req); + context.set_device_signals(signals); + Ok(Self { + context, + graph: state.identity_graph(), + browser, + eids_cookie: cookie_value(req, COOKIE_TS_EIDS), + sharedid_cookie: cookie_value(req, COOKIE_SHAREDID), }) + } + + async fn finalize(mut self, state: &AppState, mut response: Response) -> Response { + // Revocation must remain writable even for a hardened/non-browser UA. + let graph = self + .graph + .as_ref() + .filter(|_| self.browser || ec_consent_withdrawn(self.context.consent())); + ec_finalize_response( + &state.settings, + &mut self.context, + graph, + &state.partners, + self.eids_cookie.as_deref(), + self.sharedid_cookie.as_deref(), + &mut response, + ) + .await; + response + } +} + +#[derive(Clone, Copy)] +enum EcRoute { + Publisher, + Auction, + PageBids, } // --------------------------------------------------------------------------- @@ -226,16 +336,77 @@ async fn dispatch_fallback( return handle_tsjs_dynamic(&req, &state.registry, EdgeCacheHeader::SMaxageFallback); } - if state.registry.has_route(&method, &path) { - let mut ec_context = EcContext::default(); - return state + dispatch_ec_route(state, services, req, EcRoute::Publisher).await +} + +async fn dispatch_ec_route( + state: &AppState, + services: &RuntimeServices, + mut req: Request, + route: EcRoute, +) -> Result> { + let mut ec = NativeEcRequest::read(state, services, &req)?; + let method = req.method().clone(); + let path = req.uri().path().to_owned(); + let effects = match state + .registry + .filter_request(RequestFilterRegistryInput { + settings: &state.settings, + services, + req: &mut req, + geo_info: ec.context.geo_info(), + }) + .await + { + Ok(RequestFilterRegistryOutcome::Continue(effects)) => effects, + Ok(RequestFilterRegistryOutcome::Respond { response, effects }) => { + let mut response = ec.finalize(state, *response).await; + effects.apply_to_response(&mut response); + return Ok(response); + } + Err(error) => return Ok(ec.finalize(state, http_error(&error)).await), + }; + // Borrows of the graph and context remain local to this request. Neither + // the non-Send graph nor a response snapshot enters shared App extensions. + let graph = ec.graph.as_ref().filter(|_| ec.browser); + let auction = AuctionDispatch { + orchestrator: &state.orchestrator, + slots: state.settings.creative_opportunity_slots(), + registry: Some(&state.partners), + }; + let result = match route { + EcRoute::Auction => { + handle_auction( + &state.settings, + &state.orchestrator, + graph, + Some(&state.partners), + &mut ec.context, + services, + req, + ) + .await + } + EcRoute::PageBids => { + handle_page_bids( + &state.settings, + services, + graph, + auction, + &mut ec.context, + req, + ) + .await + } + EcRoute::Publisher if state.registry.has_route(&method, &path) => state .registry .handle_proxy(ProxyDispatchInput { method: &method, path: &path, settings: &state.settings, - kv: None, - ec_context: &mut ec_context, + kv: graph, + allow_generation: false, + ec_context: &mut ec.context, services, req, }) @@ -244,38 +415,54 @@ async fn dispatch_fallback( Err(Report::new(TrustedServerError::BadRequest { message: format!("Unknown integration route: {path}"), })) - }); - } - - // Run the server-side auction with the configured creative-opportunity - // slots; `handle_publisher_request` matches them against the request path. - let mut ec_context = build_ec_context(state, services, &req); - let auction = AuctionDispatch { - orchestrator: &state.orchestrator, - slots: state.settings.creative_opportunity_slots(), - registry: None, + }), + EcRoute::Publisher => { + let navigation = method == Method::GET + && is_navigation_request(&req) + && state.settings.asset_route_for_path(&path).is_none(); + if navigation { + ec.context.set_eid_sync_source(EidSyncSource::Navigation); + } + let eligible = navigation && ec.browser && graph.is_some(); + if eligible + && let Err(error) = ec.context.generate_if_needed(&state.settings, graph).await + { + log::warn!("native EC generation failed: {error:?}"); + } + match handle_publisher_request( + &state.settings, + services, + graph, + &mut ec.context, + auction, + req, + EdgeCacheHeader::SMaxageFallback, + ) + .await + { + Ok(publisher_response) => { + ec.context.set_recovery_eligible(eligible); + buffer_publisher_response_async( + publisher_response, + &method, + &state.settings, + &state.registry, + &state.orchestrator, + services, + ) + .await + } + Err(error) => Err(error), + } + } }; - let publisher_response = handle_publisher_request( - &state.settings, - services, - None, - &mut ec_context, - auction, - req, - EdgeCacheHeader::SMaxageFallback, - ) - .await?; - // Async finalize so the dispatched auction is collected and its bids are - // injected before `` (the sync buffer path would drop them). - buffer_publisher_response_async( - publisher_response, - &method, - &state.settings, - &state.registry, - &state.orchestrator, - services, - ) - .await + // Convert handled route errors before finalization, so withdrawal is not + // lost when an origin/auction fails. There is exactly one finalize call. + let mut response = ec + .finalize(state, result.unwrap_or_else(|error| http_error(&error))) + .await; + effects.apply_to_response(&mut response); + Ok(response) } fn fallback_handler() -> impl Fn(RequestContext) -> HandlerFuture + Clone + Send + Sync + 'static { @@ -303,6 +490,7 @@ enum NamedRouteHandler { LegacyAdminDenied, Auction, PageBids, + Identify, FirstPartyProxy, FirstPartyClick, FirstPartySign, @@ -325,8 +513,13 @@ const LEGACY_ADMIN_DENY_METHODS: &[Method] = &[ Method::DELETE, ]; -fn named_routes() -> [NamedRoute; 17] { +fn named_routes() -> [NamedRoute; 18] { [ + NamedRoute { + path: "/_ts/api/v1/identify", + primary_methods: &[Method::GET, Method::OPTIONS], + handler: NamedRouteHandler::Identify, + }, NamedRoute { path: "/.well-known/trusted-server.json", primary_methods: &[Method::GET], @@ -487,31 +680,39 @@ fn named_route_handler( Ok(resp) } NamedRouteHandler::AdminEcNotSupported => { - // The EC identity graph is Fastly KV backed; the Axum - // dev server has no store to read. + // Enabling native operator EC diagnostics is separate + // from the browser/server lifecycle. Ok(admin_ec_lookup_not_supported()) } NamedRouteHandler::AdminEidsLookup => { - let partner_registry = - PartnerRegistry::from_config(&state.settings.ec.partners)?; - handle_admin_eids_lookup(&partner_registry, &req) + handle_admin_eids_lookup(&state.partners, &req) } NamedRouteHandler::LegacyAdminDenied => Ok(legacy_admin_alias_denied()), + NamedRouteHandler::Identify => { + // Read-only identity lookup never enters generation or + // response finalization, including cookie ingestion. + if req.method() == Method::OPTIONS { + cors_preflight_identify(&state.settings, &req) + } else { + let context = build_ec_context(&state, &services, &req)?; + let graph = state.identity_graph().ok_or_else(|| { + Report::new(TrustedServerError::KvStore { + store_name: "native EC".to_owned(), + message: "native EC provider is not selected".to_owned(), + }) + })?; + handle_identify( + &state.settings, + &graph, + &state.partners, + &req, + &context, + ) + .await + } + } NamedRouteHandler::Auction => { - // Build the geo-aware EC context so the auction consent - // gate sees the caller's jurisdiction — `EcContext::default()` - // fails it closed for consented users. - let mut ec_context = build_ec_context(&state, &services, &req); - handle_auction( - &state.settings, - &state.orchestrator, - None, - None, - &mut ec_context, - &services, - req, - ) - .await + dispatch_ec_route(&state, &services, req, EcRoute::Auction).await } NamedRouteHandler::PageBids => { // SPA re-auction endpoint. `OPTIONS` is a CORS preflight @@ -520,21 +721,7 @@ fn named_route_handler( if req.method() == Method::OPTIONS { Ok(page_bids_preflight_denied()) } else { - let mut ec_context = build_ec_context(&state, &services, &req); - let auction = AuctionDispatch { - orchestrator: &state.orchestrator, - slots: state.settings.creative_opportunity_slots(), - registry: None, - }; - handle_page_bids( - &state.settings, - &services, - None, - auction, - &mut ec_context, - req, - ) - .await + dispatch_ec_route(&state, &services, req, EcRoute::PageBids).await } } NamedRouteHandler::FirstPartyProxy => { @@ -662,6 +849,31 @@ impl TrustedServerApp { let state = build_state_with_services(settings, Some(services))?; Ok(build_router(Some(state))) } + + /// Builds a router with an explicit live conditional EC store. + /// + /// This seam uses the same request-local graph and lifecycle as production. + /// It never reads provider selection from the environment. + /// + /// # Errors + /// Returns an error when settings or runtime plans cannot initialize. + /// + /// # Panics + /// Panics if a newly built application state unexpectedly has another owner. + pub fn routes_with_settings_services_and_ec_store( + settings: Settings, + services: RuntimeServices, + store: S, + ) -> Result> + where + S: EcKvStore + Clone + Send + Sync + 'static, + { + let mut state = build_state_with_services(settings, Some(services))?; + Arc::get_mut(&mut state) + .expect("should uniquely own new app state") + .set_ec_store(store); + Ok(build_router(Some(state))) + } } fn build_router(state: Option>) -> RouterService { diff --git a/crates/trusted-server-adapter-axum/src/ec_kv.rs b/crates/trusted-server-adapter-axum/src/ec_kv.rs new file mode 100644 index 000000000..a71548370 --- /dev/null +++ b/crates/trusted-server-adapter-axum/src/ec_kv.rs @@ -0,0 +1,696 @@ +//! Native single-region DynamoDB storage for the EC graph. +//! +//! Rows and the retained generation allocator use separate namespace partitions. +//! Never delete or expire the allocator while a namespace can be reused: its +//! monotonic counter prevents stale CAS after hard deletion and recreation. +//! Lookup/list apply logical TTL. Exact existence includes expired physical rows +//! so withdrawal can fence conditional writes already in flight before expiry. + +use std::sync::Arc; + +use async_trait::async_trait; +use edgezero_core::EdgeError; +use error_stack::Report; +use trusted_server_core::ec::kv_backend::{EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteOutcome}; +use trusted_server_core::error::TrustedServerError; + +/// Cloneable live provider. Clones share the SDK connection pool, not a cache. +#[derive(Clone, Debug)] +pub struct DynamoDbEcKv { + #[cfg(feature = "aws-dynamodb-ec")] + client: aws_sdk_dynamodb::Client, + #[cfg(feature = "aws-dynamodb-ec")] + table: String, + #[cfg(feature = "aws-dynamodb-ec")] + partition: String, + #[cfg(feature = "aws-dynamodb-ec")] + allocator_partition: String, +} + +fn preparation_error() -> EdgeError { + EdgeError::internal(std::io::Error::other( + "native EC DynamoDB preparation failed", + )) +} + +fn store_error() -> Report { + Report::new(TrustedServerError::KvStore { + store_name: "native-ec-dynamodb".to_owned(), + message: "EC DynamoDB operation failed".to_owned(), + }) +} + +/// Prepare the explicitly selected EC backend before accepting requests. +/// +/// `TS_EC_DYNAMODB_ENABLED=true` selects this backend. Absent or `false` leaves +/// it off, regardless of AWS configuration. Selection requires REGION, TABLE, +/// and NAMESPACE under the same prefix, and the `aws-dynamodb-ec` build feature. +/// Credentials use the normal AWS SDK chain. For isolated local tests only, +/// `TS_EC_DYNAMODB_LOCAL_ENDPOINT` accepts an HTTP root URL at `127.0.0.1` with +/// an explicit port and bypasses the credential chain with dummy credentials. +/// Provider errors never include SDK responses, identifiers, endpoint URLs, +/// credentials, or record contents. +/// +/// # Errors +/// Returns a sanitized error on invalid selection or failed preparation. +pub async fn prepare_from_env() -> Result>, EdgeError> { + match std::env::var("TS_EC_DYNAMODB_ENABLED") { + Err(std::env::VarError::NotPresent) => return Ok(None), + Ok(value) if value == "false" => return Ok(None), + Ok(value) if value == "true" => {} + _ => return Err(preparation_error()), + } + #[cfg(feature = "aws-dynamodb-ec")] + { + enabled::prepare_from_env().await.map(Some) + } + #[cfg(not(feature = "aws-dynamodb-ec"))] + Err(EdgeError::internal(std::io::Error::other( + "native EC DynamoDB selected but aws-dynamodb-ec build feature is unavailable", + ))) +} + +#[cfg(not(feature = "aws-dynamodb-ec"))] +#[async_trait(?Send)] +impl EcKvStore for DynamoDbEcKv { + fn store_name(&self) -> &str { + "native-ec-dynamodb" + } + async fn lookup(&self, _: &str) -> Result, Report> { + Err(store_error()) + } + async fn key_exists(&self, _: &str) -> Result> { + Err(store_error()) + } + async fn insert( + &self, + _: &str, + _: EcKvWrite<'_>, + ) -> Result> { + Err(store_error()) + } + async fn list_keys_with_prefix( + &self, + _: &str, + _: u32, + ) -> Result, Report> { + Err(store_error()) + } + async fn delete(&self, _: &str) -> Result<(), Report> { + Err(store_error()) + } +} + +#[cfg(feature = "aws-dynamodb-ec")] +mod enabled { + use std::collections::HashMap; + use std::time::{Duration, SystemTime, UNIX_EPOCH}; + + use aws_config::BehaviorVersion; + use aws_sdk_dynamodb::config::{Region, retry::RetryConfig, timeout::TimeoutConfig}; + use aws_sdk_dynamodb::primitives::Blob; + use aws_sdk_dynamodb::types::{AttributeValue, ReturnValue, TableDescription}; + use trusted_server_core::ec::kv_backend::EcKvWriteMode; + + use super::{ + Arc, DynamoDbEcKv, EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteOutcome, EdgeError, Report, + TrustedServerError, async_trait, preparation_error, store_error, + }; + + // Query can stop at 1 MB or an empty filtered page. These bounds limit work, + // not completeness: budget exhaustion is an error, never a short success. + const MAX_LIST_PAGES: usize = 64; + const PAGE_SIZE: usize = 1_000; + const OPERATION_TIMEOUT: Duration = Duration::from_secs(3); + const PREPARATION_TIMEOUT: Duration = Duration::from_secs(10); + type Item = HashMap; + type StoreResult = Result>; + + pub(super) async fn prepare_from_env() -> Result, EdgeError> { + let required = |suffix: &str| { + std::env::var(format!("TS_EC_DYNAMODB_{suffix}")).map_err(|_| preparation_error()) + }; + let region = required("REGION")?; + let table = required("TABLE")?; + let namespace = required("NAMESPACE")?; + if region.is_empty() || region.chars().any(char::is_whitespace) { + return Err(preparation_error()); + } + validate_binding(&table, &namespace)?; + tokio::time::timeout(PREPARATION_TIMEOUT, async { + let builder = match std::env::var("TS_EC_DYNAMODB_LOCAL_ENDPOINT") { + Ok(endpoint) => { + validate_local_endpoint(&endpoint)?; + // Explicit emulator mode never reads local AWS files or + // sends real workload credentials to the test endpoint. + aws_sdk_dynamodb::Config::builder() + .behavior_version(BehaviorVersion::v2026_01_12()) + .region(Region::new(region)) + .credentials_provider(aws_sdk_dynamodb::config::Credentials::new( + "local", + "local", + None, + None, + "ec-dynamodb-local", + )) + .endpoint_url(endpoint) + } + Err(std::env::VarError::NotPresent) => { + let shared = aws_config::defaults(BehaviorVersion::v2026_01_12()) + .region(Region::new(region)) + .load() + .await; + let mut builder = aws_sdk_dynamodb::Config::from(&shared).to_builder(); + // Clear both global and service-specific endpoint URLs + // copied from the shared config. Resolve normal AWS endpoints. + builder.set_endpoint_url(None); + builder + } + Err(_) => return Err(preparation_error()), + }; + let config = builder + .retry_config(RetryConfig::standard().with_max_attempts(1)) + .timeout_config(timeout_config()) + .build(); + DynamoDbEcKv::prepare_with_client( + aws_sdk_dynamodb::Client::from_conf(config), + table, + namespace, + ) + .await + }) + .await + .map_err(|_| preparation_error())? + } + + pub(super) fn validate_local_endpoint(endpoint: &str) -> Result<(), EdgeError> { + let url = reqwest::Url::parse(endpoint).map_err(|_| preparation_error())?; + if url.scheme() != "http" + || url.host_str() != Some("127.0.0.1") + || url.port().is_none() + || !url.username().is_empty() + || url.password().is_some() + || url.path() != "/" + || url.query().is_some() + || url.fragment().is_some() + { + return Err(preparation_error()); + } + Ok(()) + } + + fn validate_binding(table: &str, namespace: &str) -> Result<(), EdgeError> { + if !(3..=255).contains(&table.len()) + || !table + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b"_.-".contains(&b)) + || namespace.is_empty() + || namespace.len() > 512 + || namespace.chars().any(char::is_control) + { + return Err(preparation_error()); + } + Ok(()) + } + + fn timeout_config() -> TimeoutConfig { + TimeoutConfig::builder() + .connect_timeout(Duration::from_secs(1)) + .operation_attempt_timeout(OPERATION_TIMEOUT) + .operation_timeout(OPERATION_TIMEOUT) + .build() + } + + impl DynamoDbEcKv { + /// Prepare a provider with an explicitly supplied SDK client. + /// + /// Useful for `DynamoDB` Local and deterministic loopback tests without + /// reading ambient credentials or changing process environment. The + /// caller owns endpoint/credential configuration; retries are disabled + /// for each operation even if the supplied client enables them. + /// + /// # Errors + /// Returns a sanitized error when bindings or the table are unsuitable. + pub async fn prepare_with_client( + client: aws_sdk_dynamodb::Client, + table: impl Into, + namespace: impl Into, + ) -> Result, EdgeError> { + let table = table.into(); + let namespace = namespace.into(); + validate_binding(&table, &namespace)?; + let provider = Self { + client, + table, + partition: format!("ec:{namespace}"), + allocator_partition: format!("ec-allocator:{namespace}"), + }; + let output = tokio::time::timeout( + OPERATION_TIMEOUT, + provider + .client + .describe_table() + .table_name(&provider.table) + .customize() + .config_override(Self::policy()) + .send(), + ) + .await + .map_err(|_| preparation_error())? + .map_err(|_| preparation_error())?; + validate_table( + output.table.as_ref().ok_or_else(preparation_error)?, + &provider.table, + )?; + Ok(Arc::new(provider)) + } + + fn policy() -> aws_sdk_dynamodb::config::Builder { + aws_sdk_dynamodb::Config::builder() + .retry_config(RetryConfig::standard().with_max_attempts(1)) + .timeout_config(timeout_config()) + } + + fn key(&self, key: &str) -> Item { + HashMap::from([ + ("pk".to_owned(), AttributeValue::S(self.partition.clone())), + ("sk".to_owned(), AttributeValue::S(key.to_owned())), + ]) + } + + async fn allocate_generation(&self) -> StoreResult { + let output = self + .client + .update_item() + .table_name(&self.table) + .key("pk", AttributeValue::S(self.allocator_partition.clone())) + .key("sk", AttributeValue::S("generation".to_owned())) + .update_expression("ADD #g :one") + .condition_expression("attribute_not_exists(#g) OR #g < :max") + .expression_attribute_names("#g", "generation") + .expression_attribute_values(":one", AttributeValue::N("1".to_owned())) + .expression_attribute_values(":max", AttributeValue::N(u64::MAX.to_string())) + .return_values(ReturnValue::UpdatedNew) + .customize() + .config_override(Self::policy()) + .send() + .await + .map_err(|_| store_error())?; + let generation = unsigned( + output.attributes.as_ref().ok_or_else(store_error)?, + "generation", + )?; + if generation == 0 { + return Err(store_error()); + } + Ok(generation) + } + + async fn get_physical(&self, key: &str) -> StoreResult> { + validate_key(key)?; + let output = self + .client + .get_item() + .table_name(&self.table) + .set_key(Some(self.key(key))) + .consistent_read(true) + .customize() + .config_override(Self::policy()) + .send() + .await + .map_err(|_| store_error())?; + match output.item { + None => Ok(None), + Some(item) => { + if row_key(&item, &self.partition)? != key { + return Err(store_error()); + } + // Malformed expiry is an error, not evidence of absence. + expiry(&item)?; + Ok(Some(item)) + } + } + } + } + + #[async_trait(?Send)] + impl EcKvStore for DynamoDbEcKv { + fn store_name(&self) -> &str { + "native-ec-dynamodb" + } + + async fn lookup(&self, key: &str) -> StoreResult> { + tokio::time::timeout(OPERATION_TIMEOUT, async { + let Some(item) = self.get_physical(key).await? else { + return Ok(None); + }; + if expiry(&item)? <= now_ms()? { + return Ok(None); + } + let binary = |name: &str| match item.get(name) { + Some(AttributeValue::B(value)) => Ok(value.as_ref().to_vec()), + _ => Err(store_error()), + }; + let generation = unsigned(&item, "generation")?; + if generation == 0 { + return Err(store_error()); + } + Ok(Some(EcKvLookup { + body: binary("body")?, + metadata: Some(binary("metadata")?), + generation, + })) + }) + .await + .map_err(|_| store_error())? + } + + async fn key_exists(&self, key: &str) -> StoreResult { + // Withdrawal must fence writes already in flight with an earlier + // client timestamp. An expired physical row can still satisfy such + // a CAS, so only physical absence proves there is nothing to revoke. + tokio::time::timeout(OPERATION_TIMEOUT, self.get_physical(key)) + .await + .map_err(|_| store_error())? + .map(|item| item.is_some()) + } + + async fn insert(&self, key: &str, write: EcKvWrite<'_>) -> StoreResult { + validate_key(key)?; + // DynamoDB caps complete items at 400 KiB. Include all names, + // partition/key bytes and conservative decimal numeric accounting. + let bytes = write + .body + .len() + .checked_add(write.metadata.len()) + .and_then(|n| n.checked_add(self.partition.len() + key.len() + 256)) + .ok_or_else(store_error)?; + if bytes > 409_600 || write.ttl.is_zero() { + return Err(store_error()); + } + tokio::time::timeout(OPERATION_TIMEOUT, async { + let generation = self.allocate_generation().await?; + let timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| store_error())?; + let now = u64::try_from(timestamp.as_millis()).map_err(|_| store_error())?; + let expires = expiry_from_duration(timestamp, write.ttl)?; + let mut item = self.key(key); + item.insert( + "body".to_owned(), + AttributeValue::B(Blob::new(write.body.as_bytes())), + ); + item.insert( + "metadata".to_owned(), + AttributeValue::B(Blob::new(write.metadata.as_bytes())), + ); + item.insert( + "generation".to_owned(), + AttributeValue::N(generation.to_string()), + ); + item.insert( + "expires_at_ms".to_owned(), + AttributeValue::N(expires.to_string()), + ); + item.insert( + "ttl_epoch_s".to_owned(), + AttributeValue::N(expires.div_ceil(1_000).to_string()), + ); + let mut request = self + .client + .put_item() + .table_name(&self.table) + .set_item(Some(item)); + request = match write.mode { + EcKvWriteMode::Add => request + .condition_expression("attribute_not_exists(#pk) OR #e <= :now") + .expression_attribute_names("#pk", "pk") + .expression_attribute_names("#e", "expires_at_ms") + .expression_attribute_values(":now", AttributeValue::N(now.to_string())), + EcKvWriteMode::IfGenerationMatch(expected) => request + .condition_expression("#g = :expected AND #e > :now") + .expression_attribute_names("#g", "generation") + .expression_attribute_names("#e", "expires_at_ms") + .expression_attribute_values( + ":expected", + AttributeValue::N(expected.to_string()), + ) + .expression_attribute_values(":now", AttributeValue::N(now.to_string())), + EcKvWriteMode::Overwrite => request, + }; + match request + .customize() + .config_override(Self::policy()) + .send() + .await + { + Ok(_) => Ok(EcKvWriteOutcome::Written), + Err(error) + if error + .as_service_error() + .is_some_and(aws_sdk_dynamodb::operation::put_item::PutItemError::is_conditional_check_failed_exception) => + { + Ok(EcKvWriteOutcome::PreconditionFailed) + } + Err(_) => Err(store_error()), + } + }) + .await + .map_err(|_| store_error())? + } + + async fn list_keys_with_prefix( + &self, + prefix: &str, + limit: u32, + ) -> StoreResult> { + if prefix.len() > 1_024 || prefix.chars().any(char::is_control) { + return Err(store_error()); + } + if limit == 0 { + return Ok(Vec::new()); + } + tokio::time::timeout(OPERATION_TIMEOUT, async { + let mut keys = Vec::new(); + let mut cursor: Option = None; + for _ in 0..MAX_LIST_PAGES { + let remaining = (limit as usize).saturating_sub(keys.len()).min(PAGE_SIZE); + let mut request = self + .client + .query() + .table_name(&self.table) + .consistent_read(true) + .limit(i32::try_from(remaining).map_err(|_| store_error())?) + .key_condition_expression(if prefix.is_empty() { + "#pk = :pk" + } else { + "#pk = :pk AND begins_with(#sk, :prefix)" + }) + .expression_attribute_names("#pk", "pk") + .expression_attribute_names("#sk", "sk") + .expression_attribute_names("#e", "expires_at_ms") + .expression_attribute_names("#t", "ttl_epoch_s") + .expression_attribute_values( + ":pk", + AttributeValue::S(self.partition.clone()), + ) + .projection_expression("#pk, #sk, #e, #t") + .set_exclusive_start_key(cursor.clone()); + if !prefix.is_empty() { + request = request.expression_attribute_values( + ":prefix", + AttributeValue::S(prefix.to_owned()), + ); + } + let output = request + .customize() + .config_override(Self::policy()) + .send() + .await + .map_err(|_| store_error())?; + let items = output.items.unwrap_or_default(); + if items.len() > remaining { + return Err(store_error()); + } + let mut previous = cursor + .as_ref() + .map(|c| row_key(c, &self.partition)) + .transpose()? + .map(str::to_owned); + let now = now_ms()?; + for item in &items { + let key = row_key(item, &self.partition)?; + if !key.starts_with(prefix) + || previous.as_deref().is_some_and(|last| key <= last) + { + return Err(store_error()); + } + previous = Some(key.to_owned()); + if expiry(item)? > now { + keys.push(key.to_owned()); + } + } + let next = output.last_evaluated_key.filter(|c| !c.is_empty()); + if let Some(next) = &next { + let key = row_key(next, &self.partition)?; + if next.len() != 2 + || !key.starts_with(prefix) + || cursor.as_ref().is_some_and(|last| { + row_key(last, &self.partition).is_ok_and(|last| key <= last) + }) + || previous.as_deref().is_some_and(|last| key < last) + { + return Err(store_error()); + } + } + if keys.len() == limit as usize || next.is_none() { + return Ok(keys); + } + cursor = next; + } + Err(store_error()) + }) + .await + .map_err(|_| store_error())? + } + + async fn delete(&self, key: &str) -> StoreResult<()> { + validate_key(key)?; + tokio::time::timeout( + OPERATION_TIMEOUT, + self.client + .delete_item() + .table_name(&self.table) + .set_key(Some(self.key(key))) + .customize() + .config_override(Self::policy()) + .send(), + ) + .await + .map_err(|_| store_error())? + .map_err(|_| store_error())?; + Ok(()) + } + } + + fn validate_key(key: &str) -> StoreResult<()> { + if key.is_empty() || key.len() > 1_024 || key.chars().any(char::is_control) { + return Err(store_error()); + } + Ok(()) + } + + fn now_ms() -> StoreResult { + u64::try_from( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|_| store_error())? + .as_millis(), + ) + .map_err(|_| store_error()) + } + + fn expiry_from_duration(now: Duration, ttl: Duration) -> StoreResult { + let expires = now.checked_add(ttl).ok_or_else(store_error)?; + let rounded = expires + .as_millis() + .checked_add(u128::from( + !expires.subsec_nanos().is_multiple_of(1_000_000), + )) + .ok_or_else(store_error)?; + u64::try_from(rounded).map_err(|_| store_error()) + } + + fn unsigned(item: &Item, name: &str) -> StoreResult { + let Some(AttributeValue::N(value)) = item.get(name) else { + return Err(store_error()); + }; + let number = value.parse::().map_err(|_| store_error())?; + if number.to_string() != *value { + return Err(store_error()); + } + Ok(number) + } + + fn expiry(item: &Item) -> StoreResult { + let expires = unsigned(item, "expires_at_ms")?; + if unsigned(item, "ttl_epoch_s")? != expires.div_ceil(1_000) { + return Err(store_error()); + } + Ok(expires) + } + + fn row_key<'a>(item: &'a Item, partition: &str) -> StoreResult<&'a str> { + match (item.get("pk"), item.get("sk")) { + (Some(AttributeValue::S(pk)), Some(AttributeValue::S(sk))) if pk == partition => { + validate_key(sk)?; + Ok(sk) + } + _ => Err(store_error()), + } + } + + fn validate_table(table: &TableDescription, expected: &str) -> Result<(), EdgeError> { + if table.table_name.as_deref() != Some(expected) + || table + .table_status + .as_ref() + .is_none_or(|status| status.as_str() != "ACTIVE") + || !table.replicas().is_empty() + || table.global_table_version.is_some() + || !table.global_secondary_indexes().is_empty() + || !table.local_secondary_indexes().is_empty() + || table + .vector_indexes + .as_ref() + .is_some_and(|indexes| !indexes.is_empty()) + || table.key_schema().len() != 2 + || table.attribute_definitions().len() != 2 + || ![("pk", "HASH"), ("sk", "RANGE")] + .into_iter() + .all(|(name, kind)| { + table + .key_schema() + .iter() + .any(|key| key.attribute_name() == name && key.key_type().as_str() == kind) + }) + || !["pk", "sk"].into_iter().all(|name| { + table.attribute_definitions().iter().any(|attr| { + attr.attribute_name() == name && attr.attribute_type().as_str() == "S" + }) + }) + { + return Err(preparation_error()); + } + Ok(()) + } + #[cfg(test)] + mod unit_tests { + use super::*; + + #[test] + fn absolute_expiry_rounding_never_shortens_ttl() { + assert_eq!( + expiry_from_duration(Duration::from_nanos(999_999), Duration::from_nanos(1)) + .expect("should round exact millisecond boundary"), + 1 + ); + assert_eq!( + expiry_from_duration(Duration::from_nanos(999_999), Duration::from_nanos(2)) + .expect("should round above millisecond boundary"), + 2 + ); + assert_eq!( + expiry_from_duration(Duration::from_nanos(1), Duration::from_secs(60)) + .expect("should retain fractional current timestamp"), + 60_001 + ); + assert!(expiry_from_duration(Duration::MAX, Duration::from_nanos(1)).is_err()); + assert!(expiry_from_duration(Duration::ZERO, Duration::MAX).is_err()); + } + } +} + +#[cfg(all(test, feature = "aws-dynamodb-ec"))] +#[path = "ec_kv_tests.rs"] +mod tests; diff --git a/crates/trusted-server-adapter-axum/src/ec_kv_tests.rs b/crates/trusted-server-adapter-axum/src/ec_kv_tests.rs new file mode 100644 index 000000000..abccc1016 --- /dev/null +++ b/crates/trusted-server-adapter-axum/src/ec_kv_tests.rs @@ -0,0 +1,811 @@ +//! Tests through the real SDK HTTP protocol, without ambient AWS credentials. + +use std::collections::VecDeque; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use aws_config::BehaviorVersion; +use aws_sdk_dynamodb::Client; +use aws_sdk_dynamodb::config::{Credentials, Region}; +use aws_sdk_dynamodb::types::{ + AttributeDefinition, AttributeValue, BillingMode, KeySchemaElement, KeyType, + ScalarAttributeType, +}; +use axum::body::Bytes; +use axum::extract::State; +use axum::http::{HeaderMap, Request, StatusCode}; +use axum::response::IntoResponse; +use edgezero_core::body::Body; +use edgezero_core::http::Response; +use futures::FutureExt as _; +use serde_json::{Value, json}; +use trusted_server_core::ec::EcContext; +use trusted_server_core::ec::finalize::ec_finalize_response; +use trusted_server_core::ec::kv::{KvIdentityGraph, UpsertResult}; +use trusted_server_core::ec::kv_backend::{EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome}; +use trusted_server_core::ec::kv_types::KvEntry; +use trusted_server_core::ec::registry::PartnerRegistry; +use trusted_server_core::platform::GeoInfo; +use trusted_server_core::test_support::nextjs_auction::{self, NextJsAuctionOrigin}; + +use super::DynamoDbEcKv; + +const TABLE: &str = "ec-fixture"; +const NS: &str = "fixture"; + +type Requests = Arc>>; +type Responses = Arc>>; + +struct Fixture { + provider: Arc, + requests: Requests, + responses: Responses, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.task.abort(); + } +} + +fn client(endpoint: &str) -> Client { + Client::from_conf( + aws_sdk_dynamodb::Config::builder() + .behavior_version(BehaviorVersion::v2026_01_12()) + .region(Region::new("us-east-1")) + .credentials_provider(Credentials::new( + "fixture", "fixture", None, None, "fixture", + )) + .endpoint_url(endpoint) + .build(), + ) +} + +fn table_description() -> Value { + json!({"Table": { + "TableName": TABLE, "TableStatus": "ACTIVE", + "KeySchema": [{"AttributeName":"pk", "KeyType":"HASH"}, {"AttributeName":"sk", "KeyType":"RANGE"}], + "AttributeDefinitions": [{"AttributeName":"pk", "AttributeType":"S"}, {"AttributeName":"sk", "AttributeType":"S"}] + }}) +} + +async fn fixture() -> Fixture { + let requests: Requests = Arc::new(Mutex::new(Vec::new())); + let responses: Responses = Arc::new(Mutex::new(VecDeque::from([( + StatusCode::OK, + table_description(), + )]))); + let router = axum::Router::new() + .fallback( + async |State((requests, responses)): State<(Requests, Responses)>, + headers: HeaderMap, + body: Bytes| { + let target = headers + .get("x-amz-target") + .expect("should receive SDK operation target") + .to_str() + .expect("should decode SDK operation target") + .to_owned(); + requests + .lock() + .expect("should lock recorded SDK requests") + .push(( + target, + serde_json::from_slice(&body).expect("should decode SDK JSON request"), + )); + let (status, body) = responses + .lock() + .expect("should lock queued SDK responses") + .pop_front() + .expect("unexpected SDK request"); + ( + status, + [("content-type", "application/x-amz-json-1.0")], + body.to_string(), + ) + .into_response() + }, + ) + .with_state((Arc::clone(&requests), Arc::clone(&responses))); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("should bind loopback wire fixture"); + let endpoint = format!( + "http://{}", + listener + .local_addr() + .expect("should read wire fixture address") + ); + let task = tokio::spawn(async move { + axum::serve(listener, router) + .await + .expect("should serve SDK wire fixture"); + }); + let provider = DynamoDbEcKv::prepare_with_client(client(&endpoint), TABLE, NS) + .await + .expect("should prepare SDK wire provider"); + Fixture { + provider, + requests, + responses, + task, + } +} + +impl Fixture { + fn queue(&self, value: Value) { + self.responses + .lock() + .expect("should lock queued wire response") + .push_back((StatusCode::OK, value)); + } + fn conflict(&self) { + self.responses.lock().expect("should queue conditional conflict response").push_back((StatusCode::BAD_REQUEST, json!({"__type":"com.amazonaws.dynamodb.v20120810#ConditionalCheckFailedException", "message":"private key and body"}))); + } + fn fail(&self) { + self.responses + .lock() + .expect("should queue denied response") + .push_back(( + StatusCode::FORBIDDEN, + json!({"__type":"AccessDeniedException", "message":"private credential and row"}), + )); + } + fn allocate(&self, generation: u64) { + self.queue(json!({"Attributes":{"generation":{"N":generation.to_string()}}})); + } + fn requests(&self) -> Vec<(String, Value)> { + self.requests + .lock() + .expect("should lock recorded requests") + .clone() + } +} + +fn row(key: &str, expires: u64) -> Value { + json!({"pk":{"S":"ec:fixture"}, "sk":{"S":key}, "body":{"B":"Ym9keQ=="}, "metadata":{"B":"bWV0YQ=="}, "generation":{"N":"7"}, "expires_at_ms":{"N":expires.to_string()}, "ttl_epoch_s":{"N":expires.div_ceil(1_000).to_string()}}) +} + +fn write(mode: EcKvWriteMode) -> EcKvWrite<'static> { + EcKvWrite { + body: "body", + metadata: "meta", + ttl: Duration::from_secs(60), + mode, + } +} + +#[tokio::test(flavor = "current_thread")] +async fn conditional_writes_use_real_sdk_expressions_and_allocator_partition() { + let f = fixture().await; + f.allocate(1); + f.queue(json!({})); + assert_eq!( + f.provider + .insert("key", write(EcKvWriteMode::Add)) + .await + .expect("should perform Add wire write"), + EcKvWriteOutcome::Written + ); + f.allocate(2); + f.conflict(); + assert_eq!( + f.provider + .insert("key", write(EcKvWriteMode::Add)) + .await + .expect("should classify Add conflict"), + EcKvWriteOutcome::PreconditionFailed + ); + f.allocate(3); + f.conflict(); + assert_eq!( + f.provider + .insert("key", write(EcKvWriteMode::IfGenerationMatch(1))) + .await + .expect("should classify stale CAS conflict"), + EcKvWriteOutcome::PreconditionFailed + ); + f.allocate(4); + f.queue(json!({})); + assert_eq!( + f.provider + .insert("key", write(EcKvWriteMode::Overwrite)) + .await + .expect("should perform overwrite wire write"), + EcKvWriteOutcome::Written + ); + let requests = f.requests(); + for (_, req) in requests + .iter() + .filter(|(target, _)| target.ends_with("UpdateItem")) + { + assert_eq!( + req["Key"], + json!({"pk":{"S":"ec-allocator:fixture"}, "sk":{"S":"generation"}}) + ); + assert_eq!(req["UpdateExpression"], "ADD #g :one"); + assert_eq!( + req["ConditionExpression"], + "attribute_not_exists(#g) OR #g < :max" + ); + assert_eq!( + req["ExpressionAttributeValues"][":max"]["N"], + u64::MAX.to_string() + ); + assert!(req.get("ttl_epoch_s").is_none()); + } + let puts: Vec<_> = requests + .iter() + .filter(|(target, _)| target.ends_with("PutItem")) + .map(|(_, r)| r) + .collect(); + assert_eq!( + puts[0]["ConditionExpression"], + "attribute_not_exists(#pk) OR #e <= :now" + ); + assert_eq!( + puts[2]["ConditionExpression"], + "#g = :expected AND #e > :now" + ); + assert_eq!(puts[2]["ExpressionAttributeValues"][":expected"]["N"], "1"); + assert!(puts[3].get("ConditionExpression").is_none()); + for (i, put) in puts.into_iter().enumerate() { + assert_eq!(put["Item"]["generation"]["N"], (i + 1).to_string()); + assert_eq!(put["Item"]["body"]["B"], "Ym9keQ=="); + assert_eq!(put["Item"]["metadata"]["B"], "bWV0YQ=="); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn lookup_expires_logically_but_exact_physical_existence_can_fence_withdrawal() { + let f = fixture().await; + f.queue(json!({"Item":row("key", u64::MAX)})); + let found = f + .provider + .lookup("key") + .await + .expect("should read SDK wire row") + .expect("should retain SDK wire row"); + assert_eq!(found.body, b"body"); + assert_eq!( + found.metadata.expect("should retain wire metadata"), + b"meta" + ); + assert_eq!(found.generation, 7); + f.queue(json!({"Item":row("key", 1)})); + assert!( + f.provider + .lookup("key") + .await + .expect("should hide logically expired body") + .is_none() + ); + f.queue(json!({"Item":row("key", 1)})); + assert!( + f.provider + .key_exists("key") + .await + .expect("should confirm physical expired row for withdrawal fencing") + ); + f.queue(json!({})); + assert!( + !f.provider + .key_exists("absent") + .await + .expect("should check exact absent key") + ); + for (_, req) in f + .requests() + .iter() + .filter(|(target, _)| target.ends_with("GetItem")) + { + assert_eq!(req["ConsistentRead"], true); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn prefix_listing_drives_empty_expired_and_short_pages_to_completion() { + let f = fixture().await; + let cursor = |key: &str| json!({"pk":{"S":"ec:fixture"}, "sk":{"S":key}}); + f.queue(json!({"Items":[], "LastEvaluatedKey":cursor("marker:0")})); + f.queue(json!({"Items":[row("marker:1",1)], "LastEvaluatedKey":cursor("marker:1")})); + f.queue(json!({"Items":[row("marker:2",u64::MAX)], "LastEvaluatedKey":cursor("marker:2")})); + f.queue(json!({"Items":[row("marker:3",u64::MAX)]})); + assert_eq!( + f.provider + .list_keys_with_prefix("marker:", 3) + .await + .expect("should complete prefix pagination"), + ["marker:2", "marker:3"] + ); + let requests = f.requests(); + let queries: Vec<_> = requests + .iter() + .filter(|(target, _)| target.ends_with("Query")) + .map(|(_, r)| r) + .collect(); + assert_eq!(queries.len(), 4); + for req in queries { + assert_eq!(req["ConsistentRead"], true); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn incomplete_listing_and_infrastructure_failures_never_look_like_absence_or_conflict() { + let f = fixture().await; + for n in 0..64 { + f.queue(json!({"Items":[], "LastEvaluatedKey":{"pk":{"S":"ec:fixture"}, "sk":{"S":format!("p:{n:03}")}}})); + } + assert!(f.provider.list_keys_with_prefix("p:", 2).await.is_err()); + f.fail(); + let error = f + .provider + .key_exists("private-key") + .await + .expect_err("denied read should remain an infrastructure error"); + assert!(!format!("{error:?}").contains("private")); + f.allocate(99); + f.fail(); + assert!( + f.provider + .insert("key", write(EcKvWriteMode::Add)) + .await + .is_err() + ); + f.fail(); + assert!(f.provider.delete("key").await.is_err()); + assert_eq!(f.requests().len(), 69); // describe + 64 query + get + allocation + put + delete +} + +#[tokio::test(flavor = "current_thread")] +async fn retryable_write_failure_is_sanitized_and_never_automatically_retried() { + let f = fixture().await; + f.allocate(50); + f.responses + .lock() + .expect("should queue retryable failure") + .push_back(( + StatusCode::INTERNAL_SERVER_ERROR, + json!({"__type":"InternalServerError", "message":"private row and credentials"}), + )); + let error = f + .provider + .insert("private-key", write(EcKvWriteMode::Add)) + .await + .expect_err("uncertain write should fail without an automatic retry"); + assert!(!format!("{error:?}").contains("private")); + assert_eq!( + f.requests().len(), + 3, + "should send DescribeTable, allocator and one PutItem only" + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn malformed_cursor_expiry_or_allocator_fails_closed() { + let f = fixture().await; + f.queue(json!({"Items":[], "LastEvaluatedKey":{"pk":{"S":"other"}, "sk":{"S":"p:1"}}})); + assert!(f.provider.list_keys_with_prefix("p:", 2).await.is_err()); + let mut malformed = row("key", u64::MAX); + malformed["ttl_epoch_s"] = json!({"N":"0"}); + f.queue(json!({"Item":malformed})); + assert!(f.provider.lookup("key").await.is_err()); + f.allocate(0); + assert!( + f.provider + .insert("key", write(EcKvWriteMode::Add)) + .await + .is_err() + ); +} + +/// Run explicitly against disposable `DynamoDB` Local, never AWS: +/// `TS_EC_DYNAMODB_LOCAL_ENDPOINT=http://127.0.0.1:PORT cargo test ... -- --ignored` +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires explicitly selected disposable DynamoDB Local loopback endpoint"] +async fn dynamodb_local_contract() { + let endpoint = std::env::var("TS_EC_DYNAMODB_LOCAL_ENDPOINT") + .expect("should explicitly select a Local endpoint"); + super::enabled::validate_local_endpoint(&endpoint) + .expect("should restrict DynamoDB Local tests to a loopback root URL"); + let client = client(&endpoint); + let table = format!( + "ec-contract-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("should obtain table name timestamp") + .as_nanos() + ); + client + .create_table() + .table_name(&table) + .billing_mode(BillingMode::PayPerRequest) + .key_schema( + KeySchemaElement::builder() + .attribute_name("pk") + .key_type(KeyType::Hash) + .build() + .expect("should build partition schema"), + ) + .key_schema( + KeySchemaElement::builder() + .attribute_name("sk") + .key_type(KeyType::Range) + .build() + .expect("should build sort schema"), + ) + .attribute_definitions( + AttributeDefinition::builder() + .attribute_name("pk") + .attribute_type(ScalarAttributeType::S) + .build() + .expect("should define partition type"), + ) + .attribute_definitions( + AttributeDefinition::builder() + .attribute_name("sk") + .attribute_type(ScalarAttributeType::S) + .build() + .expect("should define sort type"), + ) + .send() + .await + .expect("should create disposable Local table"); + let result = std::panic::AssertUnwindSafe(async { + let p = DynamoDbEcKv::prepare_with_client(client.clone(), &table, NS) + .await + .expect("should prepare Local provider"); + local_generation_contract(&client, &table, &p).await; + local_expiry_and_listing_contract(&client, &table, &p).await; + local_graph_withdrawal_contract(&p).await; + local_allocator_exhaustion_contract(&client, &table, &p).await; + }) + .catch_unwind() + .await; + let cleanup = client.delete_table().table_name(&table).send().await; + if let Err(panic) = result { + std::panic::resume_unwind(panic); + } + cleanup.expect("should delete disposable Local table"); +} + +async fn local_generation_contract(client: &Client, table: &str, p: &DynamoDbEcKv) { + let writes = (0..12).map(|_| p.insert("key", write(EcKvWriteMode::Add))); + let outcomes = futures::future::join_all(writes).await; + assert!( + outcomes.iter().all(Result::is_ok), + "concurrent Add must not hide infrastructure failures" + ); + assert_eq!( + outcomes + .into_iter() + .filter(|r| matches!(r, Ok(EcKvWriteOutcome::Written))) + .count(), + 1 + ); + let initial = p + .lookup("key") + .await + .expect("should read created row") + .expect("should retain created row"); + assert_eq!(initial.body, b"body"); + assert_eq!(initial.metadata.expect("should retain metadata"), b"meta"); + assert_eq!( + p.insert( + "key", + write(EcKvWriteMode::IfGenerationMatch(initial.generation)) + ) + .await + .expect("should execute matching CAS"), + EcKvWriteOutcome::Written + ); + assert_eq!( + p.insert( + "key", + write(EcKvWriteMode::IfGenerationMatch(initial.generation)) + ) + .await + .expect("should execute stale CAS"), + EcKvWriteOutcome::PreconditionFailed + ); + let before_delete = p + .lookup("key") + .await + .expect("should read before delete") + .expect("should retain row before delete") + .generation; + p.delete("key").await.expect("should hard-delete row"); + let recreated = DynamoDbEcKv::prepare_with_client(client.clone(), table, NS) + .await + .expect("should recreate provider"); + assert_eq!( + recreated + .insert("key", write(EcKvWriteMode::Add)) + .await + .expect("should recreate deleted row"), + EcKvWriteOutcome::Written + ); + assert!( + recreated + .lookup("key") + .await + .expect("should read recreated row") + .expect("should retain recreated row") + .generation + > before_delete + ); + assert_eq!( + p.insert( + "key", + write(EcKvWriteMode::IfGenerationMatch(before_delete)) + ) + .await + .expect("should reject pre-delete generation"), + EcKvWriteOutcome::PreconditionFailed + ); + assert!(!p.key_exists("ke").await.expect("should check exact key")); + let isolated = DynamoDbEcKv::prepare_with_client(client.clone(), table, "other") + .await + .expect("should prepare isolated namespace"); + assert!( + !isolated + .key_exists("key") + .await + .expect("should isolate exact lookup") + ); + assert!( + isolated + .list_keys_with_prefix("", 100) + .await + .expect("should isolate prefix listing") + .is_empty() + ); +} + +async fn local_expiry_and_listing_contract(client: &Client, table: &str, p: &DynamoDbEcKv) { + let expiring = EcKvWrite { + ttl: Duration::from_millis(1), + ..write(EcKvWriteMode::Add) + }; + p.insert("expired", expiring) + .await + .expect("should insert expiring row"); + tokio::time::sleep(Duration::from_millis(20)).await; + let physical = client + .get_item() + .table_name(table) + .key("pk", AttributeValue::S("ec:fixture".into())) + .key("sk", AttributeValue::S("expired".into())) + .consistent_read(true) + .send() + .await + .expect("should read physical expired row") + .item + .expect("Local TTL disabled should retain physical expired row"); + let expired_generation = physical["generation"] + .as_n() + .expect("should store numeric generation") + .parse::() + .expect("should store u64 generation"); + assert!( + p.lookup("expired") + .await + .expect("should check logical expiry") + .is_none() + ); + assert!( + p.key_exists("expired") + .await + .expect("should retain physical presence for withdrawal fencing") + ); + assert_eq!( + p.insert( + "expired", + write(EcKvWriteMode::IfGenerationMatch(expired_generation)) + ) + .await + .expect("should reject expired CAS"), + EcKvWriteOutcome::PreconditionFailed + ); + assert_eq!( + p.insert("expired", write(EcKvWriteMode::Add)) + .await + .expect("should replace logically absent row"), + EcKvWriteOutcome::Written + ); + p.insert( + "__ts_ec_withdrawal_complete__:key:a", + write(EcKvWriteMode::Add), + ) + .await + .expect("should support completion marker key"); + assert_eq!( + p.count_keys_with_prefix("__ts_ec_withdrawal_complete__:", 100) + .await + .expect("should count completion markers"), + 1 + ); + // Query Limit=2 first evaluates only expired rows. A later short page also + // has a continuation. The adapter must drive both before returning live keys. + for key in ["page:000", "page:001", "page:002"] { + p.insert(key, expiring) + .await + .expect("should insert expiring prefix row"); + } + for key in ["page:003", "page:004"] { + p.insert(key, write(EcKvWriteMode::Add)) + .await + .expect("should insert live prefix row"); + } + tokio::time::sleep(Duration::from_millis(20)).await; + assert_eq!( + p.list_keys_with_prefix("page:", 2) + .await + .expect("should traverse expired Local query pages"), + ["page:003", "page:004"] + ); + assert_eq!( + p.count_keys_with_prefix("page:", 3) + .await + .expect("should count complete live prefix"), + 2 + ); +} + +async fn local_allocator_exhaustion_contract(client: &Client, table: &str, p: &DynamoDbEcKv) { + client + .update_item() + .table_name(table) + .key("pk", AttributeValue::S("ec-allocator:fixture".to_owned())) + .key("sk", AttributeValue::S("generation".to_owned())) + .update_expression("SET #g = :near_max") + .expression_attribute_names("#g", "generation") + .expression_attribute_values(":near_max", AttributeValue::N((u64::MAX - 1).to_string())) + .send() + .await + .expect("should seed retained allocator boundary"); + assert_eq!( + p.insert("allocator:last", write(EcKvWriteMode::Add)) + .await + .expect("should allocate final u64 generation"), + EcKvWriteOutcome::Written + ); + assert_eq!( + p.lookup("allocator:last") + .await + .expect("should read final generation") + .expect("should retain final generation") + .generation, + u64::MAX + ); + assert!( + p.insert("allocator:overflow", write(EcKvWriteMode::Add)) + .await + .is_err() + ); + assert!( + !p.key_exists("allocator:overflow") + .await + .expect("failed allocation should not put a row") + ); + let recreated = DynamoDbEcKv::prepare_with_client(client.clone(), table, NS) + .await + .expect("should recreate provider with exhausted allocator"); + assert!( + recreated + .insert("allocator:recreated", write(EcKvWriteMode::Add)) + .await + .is_err() + ); +} + +async fn local_graph_withdrawal_contract(p: &DynamoDbEcKv) { + let ec = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.Ab1234"; + let settings = nextjs_auction::settings(); + let services = nextjs_auction::services(Arc::new(NextJsAuctionOrigin::default())); + let graph = KvIdentityGraph::new(p.clone()); + graph + .create(ec, &KvEntry::minimal("initial.example", "initial-uid", 1)) + .await + .expect("should create graph root"); + let (first, second) = futures::join!( + graph.upsert_partner_id_if_exists(ec, "first.example", "first-uid"), + graph.upsert_partner_id_if_exists(ec, "second.example", "second-uid"), + ); + assert_eq!( + first.expect("should merge first concurrent UID"), + UpsertResult::Written + ); + assert_eq!( + second.expect("should merge second concurrent UID"), + UpsertResult::Written + ); + let (merged, _) = graph + .get(ec) + .await + .expect("should read merged graph") + .expect("should retain merged graph"); + assert_eq!(merged.ids["first.example"].uid, "first-uid"); + assert_eq!(merged.ids["second.example"].uid, "second-uid"); + let before_withdrawal = p + .lookup(ec) + .await + .expect("should read pre-withdrawal row") + .expect("should retain pre-withdrawal row"); + let stale_body = std::str::from_utf8(&before_withdrawal.body).expect("should store JSON body"); + let stale_meta = std::str::from_utf8( + before_withdrawal + .metadata + .as_deref() + .expect("should store JSON metadata"), + ) + .expect("should encode metadata as UTF-8"); + let request = Request::builder() + .uri("https://test-publisher.example.com/article") + .header("cookie", format!("ts-ec={ec}")) + .header("sec-gpc", "1") + .body(Body::empty()) + .expect("should build synthetic withdrawal request"); + let geo = GeoInfo::country_region("US", Some("TN".to_owned())); + let mut context = + EcContext::read_from_request_with_geo(&settings, &request, &services, Some(&geo)) + .expect("should decode synthetic GPC withdrawal"); + let registry = PartnerRegistry::from_config(&[]).expect("should create empty fixture registry"); + let mut response = Response::new(Body::empty()); + let (_, raced) = futures::join!( + ec_finalize_response( + &settings, + &mut context, + Some(&graph), + ®istry, + None, + None, + &mut response + ), + graph.upsert_partner_id_if_exists(ec, "late.example", "late-uid"), + ); + assert!(matches!( + raced.expect("should resolve partner update racing withdrawal"), + UpsertResult::Written | UpsertResult::ConsentWithdrawn + )); + let (tombstone, _) = graph + .get(ec) + .await + .expect("should read authoritative tombstone") + .expect("should retain tombstone"); + assert!(!tombstone.consent.ok); + assert!(tombstone.ids.is_empty()); + assert_eq!( + graph + .upsert_partner_id_if_exists(ec, "late.example", "retry-uid") + .await + .expect("should reject post-withdrawal UID"), + UpsertResult::ConsentWithdrawn + ); + assert_eq!( + p.insert( + ec, + EcKvWrite { + body: stale_body, + metadata: stale_meta, + ttl: Duration::from_secs(60), + mode: EcKvWriteMode::IfGenerationMatch(before_withdrawal.generation) + } + ) + .await + .expect("should reject stale pre-withdrawal CAS"), + EcKvWriteOutcome::PreconditionFailed + ); + assert!( + response + .headers() + .get_all("set-cookie") + .iter() + .any(|value| value + .to_str() + .is_ok_and(|value| value.starts_with("ts-ec=;") && value.contains("Max-Age=0"))), + "withdrawal should expire browser EC cookie" + ); +} diff --git a/crates/trusted-server-adapter-axum/src/lib.rs b/crates/trusted-server-adapter-axum/src/lib.rs index e0fcf6912..69160fc62 100644 --- a/crates/trusted-server-adapter-axum/src/lib.rs +++ b/crates/trusted-server-adapter-axum/src/lib.rs @@ -6,6 +6,8 @@ /// Application routing and handler registration for the Axum dev server. pub mod app; +/// Native live DynamoDB conditional identity storage and provider selection. +pub mod ec_kv; /// Request middleware (auth, response finalisation). pub mod middleware; /// Explicit native Geo boot profile and provider composition. diff --git a/crates/trusted-server-adapter-axum/src/startup.rs b/crates/trusted-server-adapter-axum/src/startup.rs index 6495a24fd..9eed039d5 100644 --- a/crates/trusted-server-adapter-axum/src/startup.rs +++ b/crates/trusted-server-adapter-axum/src/startup.rs @@ -181,6 +181,9 @@ fn validate_snapshot_consumers(settings: &Settings) -> Result<(), EdgeError> { /// # Errors /// Returns an error for missing bindings/values, bounded-read failures, invalid /// envelopes or runtime settings, unsupported store consumers, or invalid plans. +/// +/// # Panics +/// Panics if newly constructed application state unexpectedly has another owner. pub async fn initialize_native(app: &mut App, stores: &PreparedStores) -> Result<(), EdgeError> { let limits = app.config_extraction_limits().validate()?; let clock = app.monotonic_clock(); @@ -257,7 +260,14 @@ pub async fn initialize_native(app: &mut App, stores: &PreparedStores) -> Result ) .map_err(settings_error)?; validate_snapshot_consumers(&settings)?; - let state = build_state_with_settings(settings).map_err(settings_error)?; + let mut state = build_state_with_settings(settings).map_err(settings_error)?; + // EC records stay live. Only the validated provider handle is shared, never + // mutable identity rows in the immutable config/secret snapshot. + if let Some(provider) = crate::ec_kv::prepare_from_env().await? { + Arc::get_mut(&mut state) + .expect("should uniquely own initializing app state") + .set_ec_store(provider.as_ref().clone()); + } if deadline.is_expired_at(clock.now()) { return Err(configuration_error( "native application initialization exceeds its extraction deadline", diff --git a/crates/trusted-server-adapter-axum/tests/ec_kv_configuration.rs b/crates/trusted-server-adapter-axum/tests/ec_kv_configuration.rs new file mode 100644 index 000000000..fbb5c229d --- /dev/null +++ b/crates/trusted-server-adapter-axum/tests/ec_kv_configuration.rs @@ -0,0 +1,105 @@ +//! Exercise real environment selection in fresh processes, never mutate the +//! parent test runner's environment or consult ambient AWS configuration. + +use std::process::Command; + +use trusted_server_adapter_axum::ec_kv::prepare_from_env; + +#[test] +fn ec_kv_configuration_is_process_isolated() { + let mut cases = vec!["absent", "false", "empty", "uppercase", "numeric"]; + #[cfg(not(feature = "aws-dynamodb-ec"))] + cases.push("missing-feature"); + #[cfg(feature = "aws-dynamodb-ec")] + cases.extend([ + "missing-binding", + "remote-endpoint", + "https-endpoint", + "endpoint-userinfo", + "endpoint-path", + "endpoint-query", + "endpoint-fragment", + "endpoint-no-port", + ]); + for case in cases { + let output = + Command::new(std::env::current_exe().expect("should locate config test executable")) + .args(["--exact", "ec_kv_configuration_subprocess", "--nocapture"]) + .env_clear() + .env("TS_EC_CONFIGURATION_CASE", case) + .env("AWS_EC2_METADATA_DISABLED", "true") + .envs(configuration_bindings(case)) + .output() + .expect("should execute isolated provider configuration test"); + assert!( + output.status.success(), + "configuration case {case} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + } +} + +fn configuration_bindings(case: &str) -> Vec<(&'static str, &'static str)> { + let mut vars = vec![ + ("TS_EC_DYNAMODB_TABLE", "ec-configuration"), + ("TS_EC_DYNAMODB_NAMESPACE", "configuration"), + ("TS_EC_DYNAMODB_REGION", "us-east-1"), + ("TS_EC_DYNAMODB_LOCAL_ENDPOINT", "invalid-unused-endpoint"), + ]; + match case { + "absent" => {} + "false" => vars.push(("TS_EC_DYNAMODB_ENABLED", "false")), + "empty" => vars.push(("TS_EC_DYNAMODB_ENABLED", "")), + "uppercase" => vars.push(("TS_EC_DYNAMODB_ENABLED", "TRUE")), + "numeric" => vars.push(("TS_EC_DYNAMODB_ENABLED", "1")), + "missing-feature" | "missing-binding" => vars.push(("TS_EC_DYNAMODB_ENABLED", "true")), + endpoint_case => { + let endpoint = match endpoint_case { + "remote-endpoint" => "http://example.com:8000", + "https-endpoint" => "https://127.0.0.1:8000", + "endpoint-userinfo" => "http://user:secret@127.0.0.1:8000", + "endpoint-path" => "http://127.0.0.1:8000/private", + "endpoint-query" => "http://127.0.0.1:8000/?secret=value", + "endpoint-fragment" => "http://127.0.0.1:8000/#secret", + "endpoint-no-port" => "http://127.0.0.1", + _ => unreachable!("unknown subprocess fixture case"), + }; + vars.retain(|(name, _)| *name != "TS_EC_DYNAMODB_LOCAL_ENDPOINT"); + vars.push(("TS_EC_DYNAMODB_ENABLED", "true")); + vars.push(("TS_EC_DYNAMODB_LOCAL_ENDPOINT", endpoint)); + } + } + if case == "missing-binding" { + vars.retain(|(name, _)| *name != "TS_EC_DYNAMODB_TABLE"); + } + vars +} + +#[test] +fn ec_kv_configuration_subprocess() { + let Ok(case) = std::env::var("TS_EC_CONFIGURATION_CASE") else { + return; + }; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("should build isolated configuration runtime"); + let result = runtime.block_on(prepare_from_env()); + if matches!(case.as_str(), "absent" | "false") { + assert!( + result + .expect("unselected backend should remain off") + .is_none() + ); + } else { + let error = result.expect_err("selected invalid backend should fail closed"); + let detail = format!("{error:?}"); + if case == "missing-feature" { + assert!(detail.contains("aws-dynamodb-ec build feature is unavailable")); + } + assert!( + !detail.contains("secret"), + "provider configuration error should not echo input" + ); + } +} diff --git a/crates/trusted-server-adapter-axum/tests/native_ec.rs b/crates/trusted-server-adapter-axum/tests/native_ec.rs new file mode 100644 index 000000000..4774d641a --- /dev/null +++ b/crates/trusted-server-adapter-axum/tests/native_ec.rs @@ -0,0 +1,524 @@ +//! Native lifecycle regressions with isolated conditional storage and origins. + +use std::collections::BTreeMap; +use std::net::IpAddr; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; +use edgezero_core::body::Body; +use edgezero_core::http::{Method, Request, Response, header}; +use edgezero_core::router::RouterService; +use error_stack::Report; +use serde_json::json; +use tokio::sync::Notify; +use trusted_server_adapter_axum::app::TrustedServerApp; +use trusted_server_adapter_axum::platform::AxumPlatformBackend; +use trusted_server_core::ec::kv_backend::{ + EcKvLookup, EcKvStore, EcKvWrite, EcKvWriteMode, EcKvWriteOutcome, +}; +use trusted_server_core::error::TrustedServerError; +use trusted_server_core::platform::{ + ClientInfo, GeoInfo, PlatformConfigStore, PlatformError, PlatformGeo, PlatformSecretStore, + RuntimeServices, StoreId, StoreName, UnavailableKvStore, +}; +use trusted_server_core::settings::Settings; +use trusted_server_core::test_support::nextjs_auction::{self, NextJsAuctionOrigin}; + +#[derive(Clone, Default)] +struct Store { + rows: Arc>>, + generation: Arc, + fail_writes: Arc, + pause_writes: Arc, + started: Arc, + resume: Arc, +} + +#[async_trait(?Send)] +impl EcKvStore for Store { + fn store_name(&self) -> &str { + "native-ec-fixture" + } + async fn lookup(&self, key: &str) -> Result, Report> { + Ok(self + .rows + .lock() + .expect("should lock fixture rows") + .get(key) + .cloned()) + } + async fn key_exists(&self, key: &str) -> Result> { + Ok(self.lookup(key).await?.is_some()) + } + async fn insert( + &self, + key: &str, + write: EcKvWrite<'_>, + ) -> Result> { + if self.pause_writes.load(Ordering::SeqCst) { + self.started.notify_one(); + self.resume.notified().await; + } + if self.fail_writes.load(Ordering::SeqCst) { + return Err(Report::new(TrustedServerError::KvStore { + store_name: self.store_name().to_owned(), + message: "isolated uncertain write".to_owned(), + })); + } + let mut rows = self.rows.lock().expect("should lock fixture rows"); + match write.mode { + EcKvWriteMode::Add if rows.contains_key(key) => { + return Ok(EcKvWriteOutcome::PreconditionFailed); + } + EcKvWriteMode::IfGenerationMatch(expected) + if rows.get(key).map(|row| row.generation) != Some(expected) => + { + return Ok(EcKvWriteOutcome::PreconditionFailed); + } + _ => {} + } + rows.insert( + key.to_owned(), + EcKvLookup { + body: write.body.as_bytes().to_vec(), + metadata: Some(write.metadata.as_bytes().to_vec()), + generation: self.generation.fetch_add(1, Ordering::SeqCst) + 1, + }, + ); + Ok(EcKvWriteOutcome::Written) + } + async fn list_keys_with_prefix( + &self, + prefix: &str, + limit: u32, + ) -> Result, Report> { + Ok(self + .rows + .lock() + .expect("should lock fixture rows") + .keys() + .filter(|key| key.starts_with(prefix)) + .take(limit as usize) + .cloned() + .collect()) + } + async fn delete(&self, key: &str) -> Result<(), Report> { + self.rows + .lock() + .expect("should lock fixture rows") + .remove(key); + Ok(()) + } +} + +struct UnusedStores; +impl PlatformConfigStore for UnusedStores { + fn get(&self, _: &StoreName, _: &str) -> Result> { + Err(Report::new(PlatformError::Unsupported)) + } + fn put(&self, _: &StoreId, _: &str, _: &str) -> Result<(), Report> { + Err(Report::new(PlatformError::Unsupported)) + } + fn delete(&self, _: &StoreId, _: &str) -> Result<(), Report> { + Err(Report::new(PlatformError::Unsupported)) + } +} +impl PlatformSecretStore for UnusedStores { + fn get_bytes(&self, _: &StoreName, _: &str) -> Result, Report> { + Err(Report::new(PlatformError::Unsupported)) + } + fn create(&self, _: &StoreId, _: &str, _: &str) -> Result<(), Report> { + Err(Report::new(PlatformError::Unsupported)) + } + fn delete(&self, _: &StoreId, _: &str) -> Result<(), Report> { + Err(Report::new(PlatformError::Unsupported)) + } +} +struct Geo(Option); +impl PlatformGeo for Geo { + fn lookup(&self, _: Option) -> Result, Report> { + Ok(self.0.clone()) + } +} + +fn settings() -> Settings { + let mut settings = nextjs_auction::settings(); + settings.auction.enabled = false; + settings.auction.providers.clear(); + settings.creative_opportunities = None; + settings.ec.partners = serde_json::from_value(json!([{ + "name": "Fixture partner", "source_domain": "partner.example.com", + "bidstream_enabled": true, "api_token": "native-fixture-partner-token-32-bytes-minimum" + }])) + .expect("should configure isolated partner"); + settings +} + +fn router(store: Store, geo: Option) -> RouterService { + let services = RuntimeServices::builder() + .config_store(Arc::new(UnusedStores)) + .secret_store(Arc::new(UnusedStores)) + .kv_store(Arc::new(UnavailableKvStore)) + .backend(Arc::new(AxumPlatformBackend)) + .http_client(Arc::new(NextJsAuctionOrigin::default())) + .geo(Arc::new(Geo(geo))) + .client_info(ClientInfo { + client_ip: Some("198.51.100.42".parse().expect("should parse test IP")), + ..ClientInfo::default() + }) + .build(); + TrustedServerApp::routes_with_settings_services_and_ec_store(settings(), services, store) + .expect("should construct native EC fixture router") +} +fn allowed_router(store: Store) -> RouterService { + router(store, Some(GeoInfo::country_region("AU", None))) +} +fn request(method: Method, path: &str, cookie: Option<&str>) -> Request { + let mut req = axum::http::Request::builder() + .method(method) + .uri(format!("https://test-publisher.example.com{path}")) + .header( + header::USER_AGENT, + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/146.0.0.0 Safari/537.36", + ) + .header("sec-fetch-dest", "document") + .header(header::ACCEPT, "text/html"); + if let Some(cookie) = cookie { + req = req.header(header::COOKIE, cookie); + } + req.body(Body::empty()) + .expect("should construct fixture request") +} +fn ec_cookie(response: &Response) -> Option { + response + .headers() + .get_all(header::SET_COOKIE) + .iter() + .filter_map(|value| value.to_str().ok()) + .find_map(|value| { + value.strip_prefix("ts-ec=").map(|value| { + value + .split(';') + .next() + .expect("should contain cookie value") + .to_owned() + }) + }) +} +async fn call(router: &RouterService, request: Request) -> Response { + router + .oneshot(request) + .await + .expect("should dispatch native fixture request") +} + +#[tokio::test(flavor = "current_thread")] +async fn persists_before_cookie_reuses_across_app_recreation_and_withdraws() { + let store = Store::default(); + let app = allowed_router(store.clone()); + let response = call(&app, request(Method::GET, "/article", None)).await; + assert_eq!(response.status(), 200); + let ec = ec_cookie(&response).expect("should issue EC after confirmed persistence"); + assert!( + store + .key_exists(&ec) + .await + .expect("should confirm authoritative row") + ); + assert_eq!( + response + .headers() + .get_all(header::SET_COOKIE) + .iter() + .filter(|value| value + .to_str() + .is_ok_and(|value| value.starts_with("ts-ec="))) + .count(), + 1, + "should finalize exactly once" + ); + assert!( + response.headers()[header::CACHE_CONTROL] + .to_str() + .expect("should read cache policy") + .contains("private") + ); + let eids = BASE64.encode( + json!([{"source":"partner.example.com","uids":[{"id":"fixture-uid","atype":1}]}]) + .to_string(), + ); + let cookie = format!("ts-ec={ec}; ts-eids={eids}"); + let recreated = allowed_router(store.clone()); + let returning = call(&recreated, request(Method::GET, "/article", Some(&cookie))).await; + assert_eq!(returning.status(), 200); + assert!( + ec_cookie(&returning).is_none(), + "should not refresh an unchanged returning cookie" + ); + let row: serde_json::Value = serde_json::from_slice( + &store + .lookup(&ec) + .await + .expect("should read row") + .expect("should retain row") + .body, + ) + .expect("should decode persisted entry"); + assert_eq!(row["ids"]["partner.example.com"]["uid"], "fixture-uid"); + let mut identify = request(Method::GET, "/_ts/api/v1/identify", Some(&cookie)); + identify.headers_mut().insert( + header::AUTHORIZATION, + "Bearer native-fixture-partner-token-32-bytes-minimum" + .parse() + .expect("should set fixture partner token"), + ); + let identified = call(&recreated, identify).await; + assert_eq!(identified.status(), 200); + assert!( + ec_cookie(&identified).is_none(), + "identify must not emit an identity cookie" + ); + let data: serde_json::Value = serde_json::from_slice( + &identified + .into_body() + .into_bytes_bounded(64 * 1024) + .await + .expect("should read identify JSON"), + ) + .expect("should decode identify JSON"); + assert_eq!( + data["uid"], "fixture-uid", + "native identify must use the live graph and partner registry" + ); + let mut withdrawal = request(Method::GET, "/article", Some(&cookie)); + withdrawal.headers_mut().insert( + "sec-gpc", + "1".parse().expect("should set genuine withdrawal input"), + ); + // Revocation remains writable even when the user agent no longer passes + // the ordinary browser heuristic. + withdrawal.headers_mut().insert( + header::USER_AGENT, + "privacy-client".parse().expect("should set hardened UA"), + ); + let regulated = router( + store.clone(), + Some(GeoInfo::country_region("US", Some("TN".to_owned()))), + ); + let withdrawn = call(®ulated, withdrawal).await; + assert_eq!(ec_cookie(&withdrawn).as_deref(), Some("")); + let row: serde_json::Value = serde_json::from_slice( + &store + .lookup(&ec) + .await + .expect("should read tombstone") + .expect("should retain tombstone") + .body, + ) + .expect("should decode tombstone"); + assert_eq!(row["consent"]["ok"], false); + assert!( + row.get("ids") + .is_none_or(|ids| ids.as_object().is_some_and(serde_json::Map::is_empty)), + "should clear all withdrawn partner IDs" + ); + let retry = call(&recreated, request(Method::GET, "/article", Some(&cookie))).await; + assert!( + ec_cookie(&retry).is_none(), + "should not resurrect a revoked identity" + ); + let row: serde_json::Value = serde_json::from_slice( + &store + .lookup(&ec) + .await + .expect("should read tombstone after retry") + .expect("should retain tombstone") + .body, + ) + .expect("should decode retained tombstone"); + assert_eq!(row["consent"]["ok"], false); +} + +#[tokio::test(flavor = "current_thread")] +async fn missing_consent_unknown_geo_storage_failure_and_non_navigation_do_not_mint() { + let store = Store::default(); + for geo in [ + None, + Some(GeoInfo::country_region("US", Some("TN".to_owned()))), + ] { + let response = call( + &router(store.clone(), geo), + request(Method::GET, "/article", None), + ) + .await; + assert!( + ec_cookie(&response).is_none(), + "should preserve missing-consent policy" + ); + } + let app = allowed_router(store.clone()); + for (method, path) in [ + (Method::GET, "/health"), + (Method::GET, "/static/tsjs=fixture"), + (Method::OPTIONS, "/article"), + (Method::OPTIONS, "/_ts/page-bids"), + (Method::GET, "/_ts/admin/ec"), + (Method::GET, "/_ts/api/v1/identify"), + (Method::HEAD, "/article"), + (Method::POST, "/article"), + ] { + let response = call(&app, request(method, path, None)).await; + assert!(ec_cookie(&response).is_none(), "should not mint on {path}"); + } + let mut asset = request(Method::GET, "/article", None); + asset.headers_mut().insert( + "sec-fetch-dest", + "image".parse().expect("should set subresource destination"), + ); + assert!(ec_cookie(&call(&app, asset).await).is_none()); + let mut bot = request(Method::GET, "/article", None); + bot.headers_mut().insert( + header::USER_AGENT, + "curl/8.0".parse().expect("should set bot UA"), + ); + assert!(ec_cookie(&call(&app, bot).await).is_none()); + let mut identify = request(Method::GET, "/_ts/api/v1/identify", None); + identify.headers_mut().insert( + header::AUTHORIZATION, + "Bearer native-fixture-partner-token-32-bytes-minimum" + .parse() + .expect("should set fixture partner token"), + ); + let identified = call(&app, identify).await; + assert_eq!( + identified.status(), + 204, + "authenticated identify without an identity stays read-only" + ); + assert!(ec_cookie(&identified).is_none()); + assert!( + store + .rows + .lock() + .expect("should read fixture rows") + .is_empty() + ); + store.fail_writes.store(true, Ordering::SeqCst); + assert!( + ec_cookie(&call(&app, request(Method::GET, "/article", None)).await).is_none(), + "uncertain writes must not issue phantom identities" + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn regulated_fixture_uses_decoded_signal_and_gpc_keeps_precedence() { + let store = Store::default(); + let app = router( + store.clone(), + Some(GeoInfo::country_region("US", Some("TN".to_owned()))), + ); + // Encoded, synthetic US Privacy input exercises the real core decoder. + // This is not evidence of consent capture from a real publisher CMP. + let accepted = call( + &app, + request(Method::GET, "/article", Some("us_privacy=1YNN")), + ) + .await; + let ec = ec_cookie(&accepted).expect("should honor the decoded qualifying fixture signal"); + assert!( + store + .key_exists(&ec) + .await + .expect("should confirm regulated fixture row") + ); + let mut denied = request(Method::GET, "/article", Some("us_privacy=1YNN")); + denied + .headers_mut() + .insert("sec-gpc", "1".parse().expect("should set GPC input")); + assert!( + ec_cookie(&call(&app, denied).await).is_none(), + "GPC must override the non-opt-out fixture signal" + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn delayed_storage_yields_to_health_and_cancels_without_cookie() { + let store = Store::default(); + store.pause_writes.store(true, Ordering::SeqCst); + let app = allowed_router(store.clone()); + let article = call(&app, request(Method::GET, "/article", None)); + tokio::pin!(article); + tokio::select! { + _ = store.started.notified() => {}, + _ = &mut article => panic!("storage should remain pending"), + } + let health = tokio::time::timeout( + Duration::from_millis(250), + call(&app, request(Method::GET, "/health", None)), + ) + .await + .expect("health should proceed on the same current thread while storage waits"); + assert_eq!(health.status(), 200); + assert!(ec_cookie(&health).is_none()); + assert!( + store + .rows + .lock() + .expect("should inspect paused rows") + .is_empty() + ); + // Dropping the pending request models cancellation before acknowledgement. + // No synchronous executor/worker must be joined before this runtime exits. +} + +#[tokio::test(flavor = "current_thread")] +async fn orphan_recovery_rotates_only_after_an_eligible_publisher_start() { + let store = Store::default(); + let app = allowed_router(store.clone()); + let created = call(&app, request(Method::GET, "/article", None)).await; + let original = ec_cookie(&created).expect("should issue initial identity"); + store + .delete(&original) + .await + .expect("should simulate expired identity row"); + let eids = BASE64.encode( + json!([{"source":"partner.example.com","uids":[{"id":"fixture-uid","atype":1}]}]) + .to_string(), + ); + let cookie = format!("ts-ec={original}; ts-eids={eids}"); + let named = call( + &app, + request(Method::GET, "/_ts/api/v1/identify", Some(&cookie)), + ) + .await; + assert!( + ec_cookie(&named).is_none(), + "read-only identify must not recover an orphan" + ); + assert!( + store + .rows + .lock() + .expect("should read fixture rows") + .is_empty() + ); + let recovered = call(&app, request(Method::GET, "/article", Some(&cookie))).await; + let replacement = + ec_cookie(&recovered).expect("should recover on eligible publisher navigation"); + assert_ne!(replacement, original); + assert!( + store + .key_exists(&replacement) + .await + .expect("should confirm recovered row before cookie") + ); + assert!( + !store + .key_exists(&original) + .await + .expect("should leave orphan absent") + ); +} diff --git a/crates/trusted-server-adapter-cloudflare/src/app.rs b/crates/trusted-server-adapter-cloudflare/src/app.rs index 78349f853..ae1e8e4ba 100644 --- a/crates/trusted-server-adapter-cloudflare/src/app.rs +++ b/crates/trusted-server-adapter-cloudflare/src/app.rs @@ -560,6 +560,7 @@ fn build_router(state: &Arc) -> RouterService { path: &path, settings: &state.settings, kv: None, + allow_generation: true, ec_context: &mut ec_context, services: &services, req, diff --git a/crates/trusted-server-adapter-fastly/src/app.rs b/crates/trusted-server-adapter-fastly/src/app.rs index 23b57561d..f4ccc2747 100644 --- a/crates/trusted-server-adapter-fastly/src/app.rs +++ b/crates/trusted-server-adapter-fastly/src/app.rs @@ -856,6 +856,7 @@ async fn dispatch_fallback( path: &path, settings: &state.settings, kv: ec.kv_graph.as_ref(), + allow_generation: true, ec_context: &mut ec.ec_context, services, req, diff --git a/crates/trusted-server-adapter-spin/src/app.rs b/crates/trusted-server-adapter-spin/src/app.rs index 64052a711..df5b74c14 100644 --- a/crates/trusted-server-adapter-spin/src/app.rs +++ b/crates/trusted-server-adapter-spin/src/app.rs @@ -705,6 +705,7 @@ fn build_router(state: Option>) -> RouterService { path: &path, settings: &state.settings, kv: None, + allow_generation: true, ec_context: &mut ec_context, services: &services, req, diff --git a/crates/trusted-server-core/src/ec/kv_backend.rs b/crates/trusted-server-core/src/ec/kv_backend.rs index 28fa6c653..a6f1830c2 100644 --- a/crates/trusted-server-core/src/ec/kv_backend.rs +++ b/crates/trusted-server-core/src/ec/kv_backend.rs @@ -91,6 +91,12 @@ pub trait EcKvStore { /// /// A completed issuance must be visible even when [`Self::lookup`] lags. /// Prefix matches are insufficient and an inconclusive check is an error. + /// Logical expiry alone does not prove absence when a previously authorized + /// conditional write can still commit against the expired row. Such backends + /// must include physically held expired rows in this check for withdrawal + /// and orphan protection, even when lookup and listing hide logical expiry. + /// Returning `false` must establish that an old generation cannot still + /// update the apparently absent row. /// /// # Implementing this method /// diff --git a/crates/trusted-server-core/src/integrations/registry.rs b/crates/trusted-server-core/src/integrations/registry.rs index 273172f68..85dd161ed 100644 --- a/crates/trusted-server-core/src/integrations/registry.rs +++ b/crates/trusted-server-core/src/integrations/registry.rs @@ -804,6 +804,9 @@ pub struct ProxyDispatchInput<'a> { pub path: &'a str, pub settings: &'a Settings, pub kv: Option<&'a KvIdentityGraph>, + /// Whether this adapter permits identity generation on integration routes. + /// Native keeps issuance on publisher navigations, not static JS proxies. + pub allow_generation: bool, pub ec_context: &'a mut EcContext, pub services: &'a RuntimeServices, pub req: Request, @@ -1037,6 +1040,7 @@ impl IntegrationRegistry { path, settings, kv, + allow_generation, ec_context, services, mut req, @@ -1045,14 +1049,12 @@ impl IntegrationRegistry { // Organic proxy handler: generate if needed (best effort). // Only generate for document navigations — subresource requests // may lack consent signals such as the Sec-GPC header. - if is_navigation_request(&req) { + if allow_generation && is_navigation_request(&req) { if let Err(err) = ec_context.generate_if_needed(settings, kv).await { log::warn!("EC generation failed for integration proxy: {err:?}"); } } else { - log::debug!( - "EC generation skipped for integration proxy: non-document request (path={path})", - ); + log::debug!("EC generation skipped for integration proxy (path={path})",); } // Remove any caller-supplied EC header rather than forwarding it. @@ -1688,6 +1690,7 @@ mod tests { path: "/integrations/test/echo", settings: &settings, kv: None, + allow_generation: true, ec_context: &mut ec_context, services: &noop_services(), req, @@ -2045,6 +2048,7 @@ mod tests { path: "/integrations/test/ec", settings: &settings, kv: None, + allow_generation: true, ec_context: &mut ec_context, services: &services, req, @@ -2095,6 +2099,7 @@ mod tests { path: "/integrations/test/ec", settings: &settings, kv: None, + allow_generation: true, ec_context: &mut ec_context, services: &services, req, @@ -2144,6 +2149,7 @@ mod tests { path: "/integrations/test/ec", settings: &settings, kv: None, + allow_generation: true, ec_context: &mut ec_context, services: &services, req, @@ -2190,6 +2196,7 @@ mod tests { path: "/integrations/test/ec", settings: &settings, kv: None, + allow_generation: true, ec_context: &mut ec_context, services: &services, req, diff --git a/crates/trusted-server-core/src/publisher.rs b/crates/trusted-server-core/src/publisher.rs index 424f03ea4..f42fd058f 100644 --- a/crates/trusted-server-core/src/publisher.rs +++ b/crates/trusted-server-core/src/publisher.rs @@ -6894,8 +6894,8 @@ fn normalize_page_bids_path(raw: &str) -> String { /// Called by the client-side SPA navigation hook after `pushState` / `popstate`. /// /// `kv` enriches the bid request with server-side EIDs from the EC identity -/// graph. Only the Fastly adapter has a KV identity store, so Axum, Cloudflare, -/// and Spin pass `None`; client cookie EIDs are still resolved and consent-gated +/// graph. Fastly and selected native EC providers supply an identity store. +/// Cloudflare and Spin pass `None`; client cookie EIDs are still resolved and consent-gated /// on every adapter, so no adapter forwards unconsented EIDs. /// /// # Errors diff --git a/docs/guide/native-ec-persistence.md b/docs/guide/native-ec-persistence.md new file mode 100644 index 000000000..d1bf79d8c --- /dev/null +++ b/docs/guide/native-ec-persistence.md @@ -0,0 +1,136 @@ +# Native EC persistence + +The native Axum adapter can use a dedicated DynamoDB EC backend. Build with `aws-dynamodb-ec` and select it explicitly at startup. This is separate from ordinary byte KV and from immutable AppConfig or secret snapshots. A local database file on an ephemeral ECS volume is not a substitute for retained identity state. + +Persistence does not grant consent. Keep the shared EC consent, bot, browser-navigation and withdrawal rules intact. A request can be eligible for an auction but ineligible for an EC. Missing qualifying consent in a jurisdiction that requires it must still prevent EC issuance. Consent capture and CMP integration are separate work. + +```mermaid +flowchart TD + Start[Native startup] --> Validate[Validate selected regional table] + Validate --> Ready[Application ready] + Ready --> Request[Checked request and real consent signals] + Request --> Gate{Eligible publisher navigation and EC consent?} + Gate -->|No| NoMint[Do not mint an identity] + Gate -->|Yes| Allocate[Atomically allocate a fresh generation] + Allocate --> Persist[Conditionally persist identity row] + Persist -->|Confirmed| Finalize[Await shared EC response finalization] + Persist -->|Failed or uncertain| NoMint + Finalize --> Cookie[Commit response headers and ts-ec cookie] +``` + +## Native eligibility and route boundaries + +Native eligibility uses checked effective client metadata from managed ingress and a user-agent platform/browser-engine heuristic. It does not invent Fastly TLS or JA4 headers. The browser heuristic is not spoof-resistant authentication. Only a GET publisher document navigation that passes integration request filters may issue a new identity. Health checks, static assets, preflights and rejected authentication do not mint ECs. Consent behavior is unchanged. + +Native supplies the live graph and partner registry to existing auction, page-bids and integration routes, and to read-only identify. EID cookies and IDs supplied in auction bodies are supported. Identify does not create identities. The adapter awaits shared EC response finalization exactly once, including handled errors, before committing response headers. + +Native admin EC management and lookup remain unsupported and require separate enablement. This work does not provide Fastly S2S batch or post-send pull-sync parity; those native routes and background tasks are not integrated here. + +## Build and select the backend + +From the repository root, add the EC feature to the features your native image already needs: + +```sh +docker build --platform linux/amd64 \ + --build-arg TRUSTED_SERVER__GIT_VERSION="$(git rev-parse HEAD)" \ + --build-arg NATIVE_FEATURES=aws-appconfig-agent,aws-secrets-manager,aws-dynamodb-ec \ + -t trusted-server-native-ec:local . +``` + +This command is an operator example, not evidence of a qualified image. Rebuild the final runtime image from the new binary and record its source, dependency and image digests. A historical AppConfig/Secrets Manager image qualification does not cover EC persistence merely because environment variables were added. + +Set these variables on the application container, not the AppConfig Agent: + +```sh +TS_EC_DYNAMODB_ENABLED=true +TS_EC_DYNAMODB_REGION=us-east-1 +TS_EC_DYNAMODB_TABLE=example-retained-ec +TS_EC_DYNAMODB_NAMESPACE=example-staging +``` + +`TS_EC_DYNAMODB_ENABLED=true` selects the provider. `false` or absence leaves it off. Region, table and namespace are required when selected. Use the same region as the table. The TS region setting is explicit; do not rely on an unrelated `AWS_REGION` value to select the EC table region. Selected-provider preparation must succeed before readiness. Do not treat a selected-provider failure as permission to fall back to memory or emit an unbacked cookie. + +The provider uses the standard AWS workload credential chain. In ECS, the application uses the task role. Keep credential values out of deployment inputs, logs, Terraform and container environment. The execution role still handles image pull, logs and task startup secrets, not application DynamoDB calls. Without a local-test selection, production preparation ignores configured endpoint overrides and uses the regional AWS endpoint. Local-only acceptance can set `TS_EC_DYNAMODB_LOCAL_ENDPOINT=http://127.0.0.1:8000`. This seam accepts only loopback HTTP with an explicit port and root path, bypasses the AWS config and credential chain, and uses dummy credentials. Never set it in a deployed task. In-process fixtures can also inject a client through the test preparation API. + +The table is a single-region base table. Do not add Global Tables or route writes to different regions and assume the same CAS guarantees. A single namespace partition and generation counter can become hot; `PAY_PER_REQUEST` is not a throughput qualification. + +## Table and namespace ownership + +The [self-contained Terraform example](../../deploy/examples/native-ec-dynamodb/README.md) creates a retained table and attaches exact-table permissions to a supplied task role. It does not create or deploy an ECS service. + +| Item | Required layout or owner | +| ------------------- | ----------------------------------------------------------------------------------------------------------------------- | +| Table | String hash key `pk`, string range key `sk`, `PAY_PER_REQUEST`, no indexes or replicas | +| Identity partition | `pk=ec:{namespace}`, `sk` is the exact EC or withdrawal-marker key | +| Allocator partition | `pk=ec-allocator:{namespace}`, `sk=generation`, numeric generation counter only, no TTL | +| Service TTL | Enabled on numeric `ttl_epoch_s` for expiring identity/marker rows | +| Logical expiry | `expires_at_ms` filters lookup/list results and expiry-sensitive write conditions; held-row existence stays physical | +| Retained owner | Named team/person owns state, table, namespace reservations, retention, allocator continuity, costs and decommissioning | +| Workload owner | Owns the application task role, provider-enabled image and task environment | + +The namespace must be nonempty, at most 512 UTF-8 bytes, and contain no Unicode control characters. The runtime does not trim or normalize it. Backend keys must be nonempty, at most 1024 bytes, and contain no control characters; withdrawal-marker keys are supported. + +Reserve both partition values together for exactly one app/environment. Reuse the namespace across task replacements that should share identities. Use a distinct reservation for a separate environment. Do not share a namespace between unrelated applications or use an ephemeral task/run ID as its identity. + +The internal allocator is not part of EC prefix listing. It has no `ttl_epoch_s` or logical expiration. Never delete, expire or reset it while the namespace can be used again. A delete/recreate cycle must not reuse a generation that an in-flight stale writer still holds. Failed conditional writes may consume unused generations; gaps are harmless. + +Retain the allocator with the graph through backup, restore and migration. Rewinding it independently of rows can break stale-CAS protection. Do not restore old state into an active namespace without an explicit continuity plan. This example does not implement migration or disaster recovery. + +The graph owns live-row, tombstone and withdrawal-marker lifetimes. DynamoDB TTL only collects expired rows eventually. It does not define when a row stops authorizing EC use. `expires_at_ms` uses the application's UTC wall clock, so logical expiry and expiry-dependent Add/CAS conditions assume adequately synchronized task clocks. The deployment owner must qualify clock synchronization and skew across tasks. The backend adds no runtime skew guard or clock configuration. Current graph defaults are 31,536,000 seconds for live rows and 86,400 seconds for tombstones and completion markers. Do not apply a blanket TTL or cleanup job to the allocator partition. + +Lookup and prefix listing hide logically expired rows. Authoritative `key_exists` instead uses a strongly consistent exact-key read to confirm physical row presence, including logically expired rows. This means the identity is still held, not that it is logically readable. The read validates the expiry format; malformed rows are errors, not absence. Physical presence never authorizes identity use, EIDs or cookie issuance. It lets the graph write a fresh-generation withdrawal tombstone even when a CAS captured `:now` before expiry but arrives afterward. Treating the expired, still-held root as absent would skip that tombstone and let the delayed CAS match the old generation and revive the identity or partner IDs. + +## Task permissions and the existing boundary + +The runtime needs the following permissions on the exact table ARN: + +| Actions | Scope | +| ------------------------------------------------------------------------------- | -------------------------------------------------------------------------- | +| `dynamodb:DescribeTable` | Exact table, no row condition; startup schema validation | +| `dynamodb:GetItem`, `dynamodb:PutItem`, `dynamodb:DeleteItem`, `dynamodb:Query` | Exact table and `dynamodb:LeadingKeys` equal to `ec:{namespace}` | +| `dynamodb:UpdateItem` | Exact table and `dynamodb:LeadingKeys` equal to `ec-allocator:{namespace}` | + +No task permission is needed for `Scan`, `CreateTable`, `DescribeTimeToLive`, `UpdateTimeToLive`, index creation or table deletion. Table creation and TTL configuration belong to the infrastructure operator. The allocator does not require `GetItem`. The example uses separate row and allocator statements, so ordinary identity deletion cannot delete the allocator through this policy. IAM partition scoping does not enforce a particular `sk` or stop a compromised task from modifying permitted rows. + +The public native-demo design has two relevant policy layers. Its task inline policy grants runtime reads. The retained bootstrap task boundary has an explicit `Deny` with `NotAction` based on a read-action ceiling that excludes DynamoDB. Adding an inline allow alone still fails. + +Before any deployment, the boundary and workload owners must review both changes: + +1. Extend the task boundary's action ceiling with only the six required DynamoDB actions. Also add constrained boundary allows for the exact table and the two reviewed partitions. Preserve existing unrelated restrictions. An action-ceiling edit alone does not grant permission. +2. Add the equivalent exact-table, partition-scoped allow to the application's task inline policy. Review all effective denies, including organization and endpoint policies, with the actual task credentials. + +The example's `task_policy_json` is the constrained allow input for that review, not a replacement boundary document. Do not replace the boundary, add administrator access, remove explicit denies indiscriminately, or move the application calls to the execution role. No execution-role change is needed. Existing Secrets Manager-only KMS statements do not authorize unrelated KMS use. The example uses default service-managed DynamoDB encryption and adds no task KMS calls; a customer-managed key needs a separate review. + +## Retained infrastructure, disposable workload + +Keep the EC Terraform root in its own retained state, separate from the service's disposable run state. Give it a unique locked backend key and an accountable retention owner. The example enables table deletion protection and Terraform `prevent_destroy`. Neither replaces a backup, retention or erasure policy. + +Existing disposable-workload writers, manifests, inventory and cleanup code must not be assumed to know about DynamoDB. Provision this retained root through a separately reviewed operator path. Do not quietly add a retained table to automated disposable cleanup. If that ownership model changes, review creation grants, manifests, inventory, state and destruction together. + +Stopping or replacing an ECS task must leave EC state available. Keep the selected backend live while issued identities and withdrawal obligations exist. Disabling it is not a safe persistence rollback; qualify a rollback that preserves those obligations before changing provider selection. Before deleting the task role, remove or transfer the retained root's policy attachment. Identity erasure, namespace retirement, backups and table removal require their own reviewed procedure. Never reset the generation allocator as part of routine row cleanup. + +The workload still needs working regional HTTPS access to DynamoDB. This guide does not require a new NAT gateway or endpoint without checking the existing routing and endpoint policies. A read-only application root and ephemeral bindings volume can stay unchanged; neither stores the identity graph. + +## Qualification before deployment + +No AWS apply or live acceptance is implied by this guide. Local fixtures cannot prove installed IAM policies, capacity or real browser consent transport. + +With an already-built `aws-dynamodb-ec,geo-cloudfront` native binary and DynamoDB Local running on loopback, run: + +```sh +node scripts/test-native-ec-image.mjs \ + --binary /absolute/path/to/trusted-server-axum \ + --dynamodb http://127.0.0.1:8000 +``` + +Alternatively, supply `--image LOCAL_IMAGE` instead of `--binary`. Linux image mode uses host networking and never pulls or builds an image. The fixture generates isolated envelopes and secrets, owns a unique local table, and writes JSON evidence. It calls the app executable directly, so it does not qualify a deployment bootstrap wrapper. Use `--help` for image executable and source-version assertions. + +Before authorizing a deployment, retain evidence for the rebuilt image and these checks: + +- Real conditional expressions against DynamoDB Local, including concurrent Add, stale CAS, delete/recreate, metadata, logical expiry before TTL deletion, marker keys and short or empty Query pages with continuation. A bounded list that cannot prove exhaustion must fail rather than return an unsafe short result. +- Native lifecycle tests proving persistence acknowledgement precedes new `Set-Cookie: ts-ec`, reuse survives application recreation, withdrawal writes a tombstone and expires the browser cookie, and an old EID update cannot undo withdrawal. +- Failure cases for denied, unavailable and uncertain writes. No phantom identity may be issued. Test readiness failure and current-thread runtime scheduling while storage is delayed. +- Existing route exclusions and consent rules. Health, static assets, preflight and rejected authentication must not mint identities. Missing qualifying consent must remain blocked. Synthetic local consent fixtures are not live-browser consent acceptance. +- The actual final image's provider selection, non-root/read-only execution, startup, readiness, normal HTTPS trust and shutdown behavior, plus affected Fastly/WASM compatibility after shared EC changes. + +After separate AWS authorization, qualify the real task role with its retained boundary and inline policy. Prove allowed operations on the reviewed table/partitions and denial elsewhere, allocator retention, logical TTL and persistence across task replacement. Separately verify a real browser request carrying genuine qualifying consent. Do not relax privacy configuration or fabricate a live signal to make that acceptance pass. diff --git a/scripts/test-native-ec-image.mjs b/scripts/test-native-ec-image.mjs new file mode 100644 index 000000000..292285b82 --- /dev/null +++ b/scripts/test-native-ec-image.mjs @@ -0,0 +1,981 @@ +#!/usr/bin/env node +// Local synthetic EC smoke, using an already-built native runner or local image. +// Never builds, pulls images, reads operator configuration, or contacts AWS. +// Requires aws-dynamodb-ec + geo-cloudfront and an existing DynamoDB Local. +import assert from 'node:assert/strict' +import { execFile, spawn } from 'node:child_process' +import { createHash, randomUUID } from 'node:crypto' +import { + chmod, + mkdir, + mkdtemp, + readFile, + rm, + writeFile, +} from 'node:fs/promises' +import http from 'node:http' +import net from 'node:net' +import { tmpdir } from 'node:os' +import { isAbsolute, join, resolve } from 'node:path' +import { setTimeout as delay } from 'node:timers/promises' +import { promisify } from 'node:util' + +const usage = `Usage: + node scripts/test-native-ec-image.mjs --binary /absolute/trusted-server-axum --dynamodb http://127.0.0.1:8000 + node scripts/test-native-ec-image.mjs --image LOCAL_IMAGE --dynamodb http://127.0.0.1:8000 +Options: + --image-binary PATH Image executable, default /usr/local/bin/app + --expected-version SHA Assert each document's X-TS-Version + --output PATH JSON evidence, default a unique file in the temp directory + +Linux image mode uses host networking for loopback-only synthetic services, +--pull=never, a read-only root, UID 10001 and no capabilities. It tests the app +executable directly, not a deployment bootstrap wrapper. Binary mode avoids +image creation and disk use. Supply a disposable DynamoDB Local instance, never +an AWS endpoint. The script creates and deletes only its own uniquely named +table. All config, origins, geography and credentials are synthetic. This is not +real consent transport, AWS IAM, production capacity or image security proof. +` +function options(args) { + const known = new Set([ + 'binary', + 'image', + 'dynamodb', + 'image-binary', + 'expected-version', + 'output', + ]) + const result = {} + for (let index = 0; index < args.length; index++) { + const key = args[index].replace(/^--/, '') + assert( + args[index].startsWith('--') && known.has(key), + `unknown option ${args[index]}` + ) + assert(!Object.hasOwn(result, key), `duplicate --${key}`) + const value = args[++index] + assert(value && !value.startsWith('--'), `missing --${key} value`) + result[key] = value + } + assert( + Boolean(result.binary) !== Boolean(result.image), + 'supply exactly one of --binary or --image' + ) + assert(result.dynamodb, 'supply --dynamodb pointing to DynamoDB Local') + if (result.binary) + assert(isAbsolute(result.binary), '--binary must be an absolute path') + if (result.image) + assert.equal( + process.platform, + 'linux', + 'image host-network mode requires Linux' + ) + return result +} +function localEndpoint(value) { + const url = new URL(value) + assert.equal(url.protocol, 'http:', 'local endpoint must use HTTP') + assert.equal( + url.hostname, + '127.0.0.1', + 'use literal 127.0.0.1, never an external endpoint or DNS name' + ) + assert( + url.port && + !url.username && + !url.password && + !url.search && + !url.hash && + url.pathname === '/', + 'supply a plain loopback HTTP endpoint with explicit port' + ) + return url.origin +} +function sorted(value) { + if (Array.isArray(value)) return value.map(sorted) + if (value !== null && typeof value === 'object') { + return Object.fromEntries( + Object.keys(value) + .sort() + .map((key) => [key, sorted(value[key])]) + ) + } + return value +} +function envelope(data) { + data = sorted(data) + return JSON.stringify({ + version: 1, + generated_at: '2026-10-27T00:00:00Z', + data, + sha256: createHash('sha256').update(JSON.stringify(data)).digest('hex'), + }) +} +// Direct HTTP, no proxy environment, redirects, DNS, or credential lookup. +function request( + endpoint, + path = '/', + { method = 'GET', body, headers = {}, timeout = 10000, onHeaders } = {} +) { + return new Promise((resolve, reject) => { + const req = http.request( + `${endpoint}${path}`, + { method, headers, agent: false }, + (res) => { + onHeaders?.(res) + const chunks = [] + let bytes = 0 + res.on('data', (chunk) => { + bytes += chunk.length + if (bytes > 2 * 1024 * 1024) + res.destroy(new Error('fixture response exceeds 2 MiB')) + else chunks.push(chunk) + }) + res.on('error', reject) + res.on('end', () => + resolve({ + status: res.statusCode, + headers: res.headers, + body: Buffer.concat(chunks), + }) + ) + } + ) + const timer = setTimeout( + () => req.destroy(new Error('fixture HTTP deadline')), + timeout + ) + req.on('close', () => clearTimeout(timer)) + req.on('error', reject) + req.end(body) + }) +} +async function listen(server) { + await new Promise((resolve, reject) => { + server.once('error', reject) + server.listen(0, '127.0.0.1', resolve) + }) + return `http://127.0.0.1:${server.address().port}` +} +async function freePort() { + const server = net.createServer() + await listen(server) + const port = server.address().port + await new Promise((resolve) => server.close(resolve)) + return port +} +function minted(response) { + return (response.headers['set-cookie'] ?? []) + .filter((value) => value.startsWith('ts-ec=')) + .map((value) => value.slice('ts-ec='.length).split(';')[0]) + .filter(Boolean) +} + +if (process.argv.includes('--help')) { + console.log(usage) + process.exit(0) +} +const opt = options(process.argv.slice(2)) +const dynamo = localEndpoint(opt.dynamodb) +const id = randomUUID() +const table = `ts-native-ec-fixture-${id}` +const namespace = `fixture-${id}` +const partition = `ec:${namespace}` +const allocatorPartition = `ec-allocator:${namespace}` +const work = await mkdtemp(join(tmpdir(), 'ts-native-ec-smoke-')) +const output = resolve( + opt.output ?? join(tmpdir(), `ts-native-ec-smoke-${id}.json`) +) +const exec = promisify(execFile) +const docker = async (...args) => + ( + await exec('docker', args, { timeout: 20000, maxBuffer: 1024 * 1024 }) + ).stdout.trim() +const proof = { + outcome: 'running', + startedAt: new Date().toISOString(), + table, + namespace, + dynamo, + checks: [], + events: [], + artifacts: {}, + limitations: [ + 'synthetic geography and local consent policy only', + 'no live AWS IAM or browser acceptance', + 'direct app executable, not deployment wrapper', + 'local harness image or binary smoke does not qualify production release build, linkage or Docker recipe', + ], +} +let interrupted = false +const interrupt = () => { + interrupted = true +} +process.on('SIGINT', interrupt) +process.on('SIGTERM', interrupt) +const record = (name) => { + proof.checks.push(name) + console.log(`PASS ${name}`) +} +const event = (name, extra = {}) => { + const entry = { + sequence: proof.events.length, + at: new Date().toISOString(), + name, + ...extra, + } + proof.events.push(entry) + return entry.sequence +} +async function bounded(operation, milliseconds, message) { + const deadline = Date.now() + milliseconds + while (Date.now() < deadline) { + assert(!interrupted, 'interrupted') + if (await operation()) return + await delay(25) + } + throw new Error(message) +} +// DynamoDB Local checks the SigV4-shaped authorization header, not AWS keys. +// Fake credentials are generated here and are never read from the environment. +async function ddb(operation, payload) { + const date = new Date().toISOString().replace(/[:-]|\.\d{3}/g, '') + const body = JSON.stringify(payload) + const response = await request(dynamo, '/', { + method: 'POST', + body, + headers: { + 'content-type': 'application/x-amz-json-1.0', + 'content-length': Buffer.byteLength(body), + 'x-amz-target': `DynamoDB_20120810.${operation}`, + 'x-amz-date': date, + authorization: `AWS4-HMAC-SHA256 Credential=DUMMYLOCAL/${date.slice(0, 8)}/us-east-1/dynamodb/aws4_request, SignedHeaders=host;x-amz-date, Signature=${'0'.repeat(64)}`, + }, + }) + assert.equal( + response.status, + 200, + `DynamoDB Local ${operation} failed: ${response.body.toString('utf8').slice(0, 500)}` + ) + return JSON.parse(response.body.toString('utf8')) +} +const key = (sk, pk = partition) => ({ pk: { S: pk }, sk: { S: sk } }) +const get = async (sk, pk) => + ( + await ddb('GetItem', { + TableName: table, + Key: key(sk, pk), + ConsistentRead: true, + }) + ).Item +const decode = (item) => + JSON.parse(Buffer.from(item.body.B, 'base64').toString('utf8')) +let fault = 'normal' +const pending = new Set() +const verifiedPuts = [] +const ownedApps = [] +let tableCreated = false +let origin, proxy +// Faults alter only this fixture's app-to-local-Dynamo traffic. Administration +// bypasses the proxy so it can prove a commit before returning the SDK's ack. +async function proxyRequest(req, res) { + try { + const chunks = [] + let bytes = 0 + for await (const chunk of req) { + bytes += chunk.length + assert(bytes <= 2 * 1024 * 1024, 'fixture request limit') + chunks.push(chunk) + } + const body = Buffer.concat(chunks) + const payload = JSON.parse(body.toString('utf8')) + const operation = String(req.headers['x-amz-target']).split('.').at(-1) + assert.equal(payload.TableName, table, 'app attempted a non-fixture table') + const write = operation === 'PutItem' + if (write) + assert.equal( + payload.Item.pk.S, + partition, + 'app attempted a non-fixture identity partition' + ) + if (operation === 'UpdateItem') + assert.equal(payload.Key.pk.S, allocatorPartition) + const selectedFault = fault + event('ddb-request', { + operation, + fault: selectedFault, + key: payload.Item?.sk?.S ?? payload.Key?.sk?.S, + condition: payload.ConditionExpression, + expectedGeneration: payload.ExpressionAttributeValues?.[':expected']?.N, + capturedNowMs: payload.ExpressionAttributeValues?.[':now']?.N, + }) + if (write && selectedFault === 'deny') { + res.writeHead(400, { + 'content-type': 'application/x-amz-json-1.0', + 'x-amzn-errortype': 'AccessDeniedException', + }) + res.end( + JSON.stringify({ + __type: 'com.amazonaws.dynamodb.v20120810#AccessDeniedException', + message: 'synthetic fixture denial', + }) + ) + return + } + if ( + write && + (selectedFault === 'pause' || + (selectedFault === 'pause-cas' && + payload.ConditionExpression?.includes(':expected'))) + ) { + await new Promise((resolve) => { + const entry = { + resolve, + key: payload.Item.sk.S, + condition: payload.ConditionExpression, + expectedGeneration: + payload.ExpressionAttributeValues?.[':expected']?.N, + capturedNowMs: payload.ExpressionAttributeValues?.[':now']?.N, + } + pending.add(entry) + res.once('close', () => { + pending.delete(entry) + resolve() + }) + entry.resolve = () => { + pending.delete(entry) + resolve() + } + }) + if (res.destroyed) return + } + const response = await request(dynamo, '/', { + method: 'POST', + body, + headers: { ...req.headers, host: new URL(dynamo).host }, + timeout: 5000, + }) + if (write && response.status === 200) { + const sk = payload.Item.sk.S + const row = await get(sk) + assert(row, 'DynamoDB acknowledged PutItem without a readable row') + assert.equal(row.generation.N, payload.Item.generation.N) + assert.equal(row.body.B, payload.Item.body.B) + verifiedPuts.push({ + key: sk, + sequence: event('put-verified-before-ack', { + key: sk, + generation: row.generation.N, + fault: selectedFault, + }), + }) + } + if (write && selectedFault === 'uncertain' && response.status === 200) { + // The row exists, but the caller never receives its successful ack. + event('commit-ack-lost', { key: payload.Item.sk.S }) + res.destroy() + return + } + res.writeHead(response.status, response.headers) + res.end(response.body) + } catch (error) { + proof.proxyFailure ??= error.message + if (!res.headersSent) res.writeHead(502, { 'content-type': 'text/plain' }) + res.end('synthetic fixture proxy failure') + } +} +async function state(app) { + if (app.child) + return { + Running: app.child.exitCode === null && app.child.signalCode === null, + ExitCode: app.child.exitCode, + Signal: app.child.signalCode, + } + return JSON.parse( + await docker('inspect', '--format={{json .State}}', app.name) + ) +} +async function logs(app) { + if (app.child) return app.log + const result = await exec('docker', ['logs', app.name], { + timeout: 5000, + maxBuffer: 1024 * 1024, + }) + return result.stdout + result.stderr +} +async function signal(app, value) { + if (app.child) app.child.kill(value) + else await docker('kill', `--signal=${value}`, app.name) +} +async function stop(app, graceful = true) { + if (!(await state(app)).Running) return + await signal(app, graceful ? 'SIGTERM' : 'SIGKILL') + await bounded( + async () => !(await state(app)).Running, + 7000, + 'runner shutdown deadline' + ) + if (graceful) + assert.equal((await state(app)).ExitCode, 0, 'runner must exit normally') +} +async function launch(label, originEndpoint, proxyEndpoint) { + const port = await freePort() + const dir = join(work, label) + await mkdir(dir, { mode: 0o755 }) + await chmod(dir, 0o755) + const fixture = { + ec: { + passphrase: 'ec_key', + partners: [ + { + name: 'synthetic fixture', + source_domain: 'partner.example.com', + bidstream_enabled: true, + pull_sync_enabled: false, + }, + ], + }, + handlers: [ + { path: '^/_ts/admin', username: 'admin', password: 'admin_key' }, + ], + publisher: { + domain: 'publisher.example', + cookie_domain: '.publisher.example', + origin_url: originEndpoint, + proxy_secret: 'proxy_key', + }, + auction: { enabled: false }, + request_signing: null, + } + await writeFile( + join(dir, 'local-config-trusted_server_config.json'), + JSON.stringify({ trusted_server_config: envelope(fixture) }) + ) + await writeFile( + join(dir, 'native-geo.toml'), + 'version = 1\n[geo]\nprovider = "cloudfront"\nalb_peer_cidrs = ["127.0.0.1/32"]\norigin_credential_header_name = "x-origin-synthetic-fixture"\n' + ) + await chmod(join(dir, 'local-config-trusted_server_config.json'), 0o444) + await chmod(join(dir, 'native-geo.toml'), 0o444) + const root = opt.binary ? dir : '/fixture' + const env = { + HOME: root, + PATH: '/usr/local/bin:/usr/bin:/bin', + AWS_CONFIG_FILE: `${root}/nonexistent-aws-config`, + AWS_SHARED_CREDENTIALS_FILE: `${root}/nonexistent-aws-credentials`, + AWS_EC2_METADATA_DISABLED: 'true', + AWS_REGION: 'us-east-1', + AWS_DEFAULT_REGION: 'us-east-1', + TS_EC_DYNAMODB_ENABLED: 'true', + TS_EC_DYNAMODB_REGION: 'us-east-1', + TS_EC_DYNAMODB_TABLE: table, + TS_EC_DYNAMODB_NAMESPACE: namespace, + TS_EC_DYNAMODB_LOCAL_ENDPOINT: proxyEndpoint, + TS_NATIVE_GEO_PROFILE_FILE: `${root}/native-geo.toml`, + EDGEZERO__ADAPTER__CONFIG_DIR: root, + EDGEZERO__ADAPTER__DATA_DIR: opt.binary ? dir : '/data', + EDGEZERO__ADAPTER__HOST: '127.0.0.1', + EDGEZERO__ADAPTER__PORT: String(port), + EDGEZERO__ADAPTER__SHUTDOWN_GRACE_SECONDS: '3', + EDGEZERO__ADAPTER__TRUSTED_PROXY_CIDRS: '127.0.0.1/32', + EDGEZERO__ADAPTER__FORWARDING_HEADER_FAMILY: 'x-forwarded', + EDGEZERO__SECRETS__EC_KEY: 'synthetic-ec-passphrase-32-bytes-minimum', + EDGEZERO__SECRETS__PROXY_KEY: 'synthetic-proxy-secret-32-bytes-minimum', + EDGEZERO__SECRETS__ADMIN_KEY: 'synthetic-admin-password-32-bytes-minimum', + RUST_LOG: 'info', + } + const app = { label, endpoint: `http://127.0.0.1:${port}`, log: '' } + ownedApps.push(app) + if (opt.binary) { + app.child = spawn(opt.binary, [], { + cwd: dir, + env, + stdio: ['ignore', 'pipe', 'pipe'], + }) + app.child.on('error', (error) => { + app.spawnError = error.message + }) + const capture = (chunk) => { + app.log = (app.log + chunk.toString('utf8')).slice(-256 * 1024) + } + app.child.stdout.on('data', capture) + app.child.stderr.on('data', capture) + } else { + app.name = `ts-native-ec-${label}-${id}` + await docker( + 'run', + '-d', + '--pull=never', + '--name', + app.name, + '--network=host', + '--read-only', + '--user=10001:10001', + '--cap-drop=ALL', + '--security-opt=no-new-privileges', + '--mount', + `type=bind,src=${dir},dst=/fixture,readonly`, + '--tmpfs', + '/data:rw,noexec,nosuid,size=16m,uid=10001,gid=10001,mode=0700', + '--tmpfs', + '/tmp:rw,noexec,nosuid,size=16m,uid=10001,gid=10001,mode=0700', + ...Object.entries(env).flatMap(([key, value]) => [ + '--env', + `${key}=${value}`, + ]), + '--entrypoint', + opt['image-binary'] ?? '/usr/local/bin/app', + opt.image + ) + } + await bounded( + async () => { + assert(!app.spawnError, `runner spawn failed: ${app.spawnError}`) + assert( + (await state(app)).Running, + `runner exited before readiness: ${await logs(app)}` + ) + const health = await request(app.endpoint, '/health', { + timeout: 300, + }).catch(() => null) + if (!health) return false + assert.equal(health.status, 200) + assert.equal(health.body.toString('utf8'), 'ok') + assert.equal(minted(health).length, 0) + return true + }, + 15000, + 'runner readiness deadline; confirm local endpoint seam and required features' + ) + return app +} +const headers = (country = 'AU', extra = {}) => ({ + host: 'publisher.example', + accept: 'text/html', + 'user-agent': + 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/146.0.0.0 Safari/537.36', + 'sec-fetch-dest': 'document', + 'sec-fetch-mode': 'navigate', + 'x-forwarded-host': 'publisher.example', + 'x-forwarded-proto': 'https', + 'x-forwarded-for': '198.51.100.24', + 'cloudfront-viewer-country': country, + 'x-origin-synthetic-fixture': 'synthetic-fixture-only', + ...extra, +}) +async function document( + app, + country = 'AU', + extra = {}, + allowStorageError = false +) { + let responseSequence + const result = await request(app.endpoint, '/article', { + headers: headers(country, extra), + onHeaders: () => { + responseSequence = event('document-response-headers', { + label: app.label, + country, + }) + }, + }) + if (allowStorageError) + assert( + result.status === 200 || result.status >= 500, + 'storage failure must not become an unrelated redirect or client rejection' + ) + else + assert.equal( + result.status, + 200, + 'synthetic publisher document must succeed' + ) + if (result.status === 200) + assert.equal( + result.headers['x-geo-country'], + country, + 'synthetic trusted Geo must be admitted' + ) + if (opt['expected-version']) + assert.equal(result.headers['x-ts-version'], opt['expected-version']) + return { ...result, responseSequence } +} +try { + await chmod(work, 0o755) + if (opt.binary) { + proof.artifacts = { + binary: opt.binary, + sha256: createHash('sha256') + .update(await readFile(opt.binary)) + .digest('hex'), + } + } else { + const inspect = JSON.parse(await docker('image', 'inspect', opt.image))[0] + proof.artifacts = { + image: opt.image, + imageId: inspect.Id, + repoDigests: inspect.RepoDigests, + executable: opt['image-binary'] ?? '/usr/local/bin/app', + } + } + // No table can pre-exist under this unpredictable UUID. Register ownership + // before CreateTable so an ambiguous local response still gets cleanup. + tableCreated = true + await ddb('CreateTable', { + TableName: table, + BillingMode: 'PAY_PER_REQUEST', + KeySchema: [ + { AttributeName: 'pk', KeyType: 'HASH' }, + { AttributeName: 'sk', KeyType: 'RANGE' }, + ], + AttributeDefinitions: [ + { AttributeName: 'pk', AttributeType: 'S' }, + { AttributeName: 'sk', AttributeType: 'S' }, + ], + }) + await bounded( + async () => + (await ddb('DescribeTable', { TableName: table })).Table.TableStatus === + 'ACTIVE', + 5000, + 'DynamoDB Local table activation' + ) + await ddb('UpdateTimeToLive', { + TableName: table, + TimeToLiveSpecification: { AttributeName: 'ttl_epoch_s', Enabled: true }, + }) + origin = http.createServer((req, res) => { + if (req.headers['x-origin-synthetic-fixture'] !== undefined) { + proof.originFailure = + 'protected-origin credential header reached synthetic origin' + res.writeHead(502) + res.end('fixture origin credential removal failed') + return + } + res.writeHead(200, { 'content-type': 'text/html' }) + res.end( + 'synthetic native EC fixturefixture article' + ) + }) + const originEndpoint = await listen(origin) + proxy = http.createServer((req, res) => { + void proxyRequest(req, res) + }) + const proxyEndpoint = await listen(proxy) + let app = await launch('initial', originEndpoint, proxyEndpoint) + const beforeMissing = verifiedPuts.length + assert.equal( + minted( + await document(app, 'US', { 'cloudfront-viewer-country-region': 'TN' }) + ).length, + 0 + ) + assert.equal( + verifiedPuts.length, + beforeMissing, + 'US TN missing consent must not create an identity' + ) + record( + 'US TN without qualifying consent does not mint or persist an identity' + ) + for (const path of ['/health', '/_ts/admin/ec']) { + const result = await request(app.endpoint, path, { headers: headers() }) + assert.equal(minted(result).length, 0) + } + record('health and rejected admin route do not mint') + const first = await document(app) + assert.equal( + minted(first).length, + 1, + 'eligible synthetic AU navigation must issue exactly one EC' + ) + const ec = minted(first)[0] + const create = verifiedPuts.find((put) => put.key === ec) + assert( + create && create.sequence < first.responseSequence, + 'real row must be verified before SDK ack and cookie response headers' + ) + const live = await get(ec) + assert.equal(decode(live).consent.ok, true) + assert(Number(live.expires_at_ms.N) > Date.now()) + assert.equal( + BigInt(live.ttl_epoch_s.N), + (BigInt(live.expires_at_ms.N) + 999n) / 1000n + ) + assert.match(first.headers['cache-control'], /private/) + const allocator = await get('generation', allocatorPartition) + assert(allocator?.generation?.N) + assert.equal(allocator.ttl_epoch_s, undefined) + assert.equal(allocator.expires_at_ms, undefined) + record( + 'real DynamoDB row and non-expiring allocator precede new cookie headers' + ) + await stop(app) + app = await launch('recreated', originEndpoint, proxyEndpoint) + const returning = await document(app, 'AU', { cookie: `ts-ec=${ec}` }) + assert.equal( + minted(returning).length, + 0, + 'ordinary returning document must not refresh the EC cookie' + ) + assert.equal((await get(ec)).generation.N, live.generation.N) + record( + 'recreated native process reuses the retained identity without cookie refresh' + ) + // AU grants initial creation without a consent string, but unchanged core + // policy only treats this GPC signal as explicit EC withdrawal in US TN. + const withdrawn = await document(app, 'US', { + 'cloudfront-viewer-country-region': 'TN', + cookie: `ts-ec=${ec}`, + 'sec-gpc': '1', + 'user-agent': 'privacy-client', + }) + assert.equal(minted(withdrawn).length, 0) + assert( + (withdrawn.headers['set-cookie'] ?? []).some( + (value) => /^ts-ec=;/.test(value) && /max-age=0/i.test(value) + ), + 'withdrawal must expire the browser cookie' + ) + const tombstone = await get(ec) + assert.equal(decode(tombstone).consent.ok, false) + assert.equal(Object.keys(decode(tombstone).ids ?? {}).length, 0) + assert(BigInt(tombstone.generation.N) > BigInt(live.generation.N)) + assert(Number(tombstone.expires_at_ms.N) > Date.now()) + assert(Number(tombstone.ttl_epoch_s.N) * 1000 < Date.now() + 2 * 86400000) + assert.equal( + minted(await document(app, 'AU', { cookie: `ts-ec=${ec}` })).length, + 0 + ) + assert.equal( + decode(await get(ec)).consent.ok, + false, + 'returning cookie must not revive a tombstone' + ) + record( + 'US TN GPC withdrawal expires cookie, persists a fresh-generation tombstone and resists AU reuse' + ) + // Hold a real EID CAS after the client captured :now but before DynamoDB + // evaluates it. Let the physical row expire logically, then withdraw it. + // Filtering key_exists by logical expiry would skip the protective tombstone + // and allow the delayed CAS to match the old physical generation afterward. + const raceCreated = await document(app) + assert.equal(minted(raceCreated).length, 1) + const raceEc = minted(raceCreated)[0] + const raceItem = await get(raceEc) + const raceExpiry = Date.now() + 500 + raceItem.expires_at_ms = { N: String(raceExpiry) } + raceItem.ttl_epoch_s = { N: String(Math.ceil(raceExpiry / 1000)) } + await ddb('PutItem', { TableName: table, Item: raceItem }) + const raceEids = Buffer.from( + JSON.stringify([ + { + source: 'partner.example.com', + uids: [{ id: 'synthetic-late-eid', atype: 1 }], + }, + ]) + ).toString('base64') + fault = 'pause-cas' + const eidUpdate = document(app, 'AU', { + cookie: `ts-ec=${raceEc}; ts-eids=${raceEids}`, + }).then( + (value) => ({ value }), + (error) => ({ error }) + ) + await bounded( + () => [...pending].some((entry) => entry.key === raceEc), + 2500, + 'EID update must reach the real conditional PutItem before logical expiry' + ) + const pausedCas = [...pending].find((entry) => entry.key === raceEc) + assert(pausedCas.condition.includes(':expected')) + assert.equal(pausedCas.expectedGeneration, raceItem.generation.N) + assert( + Number(pausedCas.capturedNowMs) < raceExpiry, + 'CAS must capture its expiry check while the row is live' + ) + proof.expiryWithdrawalRace = { + ec: raceEc, + expiresAtMs: raceExpiry, + expectedGeneration: pausedCas.expectedGeneration, + capturedNowMs: pausedCas.capturedNowMs, + } + await bounded( + () => Date.now() > raceExpiry + 50, + 1500, + 'wait for logical expiry while the physical CAS remains pending' + ) + const raceWithdrawal = await document(app, 'US', { + 'cloudfront-viewer-country-region': 'TN', + cookie: `ts-ec=${raceEc}`, + 'sec-gpc': '1', + 'user-agent': 'privacy-client', + }) + proof.expiryWithdrawalRace.withdrawalExpiredCookie = ( + raceWithdrawal.headers['set-cookie'] ?? [] + ).some((value) => /^ts-ec=;/.test(value) && /max-age=0/i.test(value)) + proof.expiryWithdrawalRace.afterWithdrawalItem = await get(raceEc) + assert( + proof.expiryWithdrawalRace.withdrawalExpiredCookie, + 'racing withdrawal must expire the browser cookie' + ) + pausedCas.resolve() + const racedUpdate = await eidUpdate + fault = 'normal' + assert( + !racedUpdate.error, + `delayed EID request failed: ${racedUpdate.error?.message}` + ) + const afterRace = await get(raceEc) + proof.expiryWithdrawalRace.finalItem = afterRace + assert(afterRace, 'racing withdrawal must retain an authoritative tombstone') + const afterRaceBody = decode(afterRace) + assert.equal( + afterRaceBody.consent.ok, + false, + 'expired physical row must remain withdrawn after a delayed stale-time CAS' + ) + assert.equal( + Object.keys(afterRaceBody.ids ?? {}).length, + 0, + 'delayed CAS must not restore partner IDs after withdrawal' + ) + assert.equal(minted(racedUpdate.value).length, 0) + record( + 'logical expiry plus delayed EID CAS cannot bypass US TN withdrawal or restore partner IDs' + ) + fault = 'deny' + const beforeDenied = verifiedPuts.length + assert.equal(minted(await document(app, 'AU', {}, true)).length, 0) + assert.equal(verifiedPuts.length, beforeDenied) + assert( + proof.events.some( + (entry) => entry.operation === 'PutItem' && entry.fault === 'deny' + ) + ) + record('denied PutItem never emits a phantom cookie') + fault = 'uncertain' + const beforeUncertain = verifiedPuts.length + assert.equal(minted(await document(app, 'AU', {}, true)).length, 0) + assert( + verifiedPuts.length > beforeUncertain, + 'uncertain fixture must actually commit before losing acknowledgement' + ) + assert(proof.events.some((entry) => entry.name === 'commit-ack-lost')) + record('post-commit acknowledgement loss never emits a phantom cookie') + fault = 'pause' + const article = document(app).then( + (value) => ({ value }), + (error) => ({ error }) + ) + await bounded(() => pending.size > 0, 3000, 'pending real DynamoDB PutItem') + const healthStarted = Date.now() + const health = await request(app.endpoint, '/health', { timeout: 750 }) + assert.equal(health.status, 200) + assert.equal(minted(health).length, 0) + proof.healthWhileStoragePausedMs = Date.now() - healthStarted + const shutdownStarted = Date.now() + await signal(app, 'SIGTERM') + await bounded( + async () => /phase=draining/.test(await logs(app)), + 1000, + 'runner must begin drain while real storage is paused' + ) + for (const entry of [...pending]) entry.resolve() + const completed = await article + assert( + !completed.error, + `in-flight request failed during drain: ${completed.error?.message}` + ) + assert.equal(minted(completed.value).length, 1) + const delayedEc = minted(completed.value)[0] + assert( + verifiedPuts.some( + (put) => + put.key === delayedEc && put.sequence < completed.value.responseSequence + ) + ) + await bounded( + async () => !(await state(app)).Running, + 7000, + 'current-thread runner drain deadline' + ) + assert.equal((await state(app)).ExitCode, 0) + proof.shutdownMs = Date.now() - shutdownStarted + const finalLogs = await logs(app) + assert.match(finalLogs, /phase=stopped/) + record( + 'actual runner serves health and starts shutdown while storage waits, then drains normally' + ) + assert(!proof.proxyFailure, `fixture proxy error: ${proof.proxyFailure}`) + assert(!proof.originFailure, proof.originFailure) + proof.outcome = 'pass' +} catch (error) { + proof.outcome = 'fail' + proof.failure = error.stack ?? error.message + process.exitCode = 1 + console.error(`FAIL ${error.message}`) +} finally { + fault = 'normal' + for (const entry of [...pending]) entry.resolve() + proof.logs = {} + for (const app of [...ownedApps].reverse()) { + try { + await stop(app) + } catch (error) { + proof.cleanupErrors ??= [] + proof.cleanupErrors.push(`stop ${app.label}: ${error.message}`) + try { + await signal(app, 'SIGKILL') + } catch { + /* Already stopped. */ + } + } + try { + proof.logs[app.label] = (await logs(app)).slice(-256 * 1024) + } catch { + /* Failed launch. */ + } + if (app.name) { + try { + await docker('rm', '-f', app.name) + } catch (error) { + proof.cleanupErrors ??= [] + proof.cleanupErrors.push(`container ${app.label}: ${error.message}`) + } + } + } + for (const server of [proxy, origin]) { + if (server) { + server.closeAllConnections() + await new Promise((resolve) => server.close(resolve)) + } + } + if (tableCreated) { + try { + await ddb('DeleteTable', { TableName: table }) + } catch (error) { + proof.cleanupErrors ??= [] + proof.cleanupErrors.push(`local table: ${error.message}`) + } + } + if (proof.cleanupErrors?.length) { + proof.outcome = 'fail' + process.exitCode = 1 + } + proof.finishedAt = new Date().toISOString() + try { + await writeFile(output, JSON.stringify(proof, null, 2) + '\n', { + mode: 0o600, + }) + console.log(`Evidence: ${output}`) + } catch (error) { + process.exitCode = 1 + console.error(`Evidence write failed: ${error.code ?? error.message}`) + console.log(JSON.stringify(proof)) + } finally { + await rm(work, { recursive: true, force: true }) + process.off('SIGINT', interrupt) + process.off('SIGTERM', interrupt) + } +} From 5dfab3843691a481a2d882b8ce1f8b9f720cfa4d Mon Sep 17 00:00:00 2001 From: Christian Date: Tue, 10 Nov 2026 20:55:35 -0600 Subject: [PATCH 2/3] Use a concrete local port in the retained DynamoDB test example --- crates/trusted-server-adapter-axum/src/ec_kv_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/trusted-server-adapter-axum/src/ec_kv_tests.rs b/crates/trusted-server-adapter-axum/src/ec_kv_tests.rs index abccc1016..9ef1f6f65 100644 --- a/crates/trusted-server-adapter-axum/src/ec_kv_tests.rs +++ b/crates/trusted-server-adapter-axum/src/ec_kv_tests.rs @@ -401,7 +401,7 @@ async fn malformed_cursor_expiry_or_allocator_fails_closed() { } /// Run explicitly against disposable `DynamoDB` Local, never AWS: -/// `TS_EC_DYNAMODB_LOCAL_ENDPOINT=http://127.0.0.1:PORT cargo test ... -- --ignored` +/// `TS_EC_DYNAMODB_LOCAL_ENDPOINT=http://127.0.0.1:8000 cargo test ... -- --ignored` #[tokio::test(flavor = "current_thread")] #[ignore = "requires explicitly selected disposable DynamoDB Local loopback endpoint"] async fn dynamodb_local_contract() { From 8b11dbad397ae455898ce064600284efd8800d18 Mon Sep 17 00:00:00 2001 From: Christian Date: Tue, 10 Nov 2026 21:06:09 -0600 Subject: [PATCH 3/3] Keep the external Terraform example out of site-local links --- docs/guide/native-ec-persistence.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/guide/native-ec-persistence.md b/docs/guide/native-ec-persistence.md index d1bf79d8c..a3fd6f702 100644 --- a/docs/guide/native-ec-persistence.md +++ b/docs/guide/native-ec-persistence.md @@ -56,7 +56,7 @@ The table is a single-region base table. Do not add Global Tables or route write ## Table and namespace ownership -The [self-contained Terraform example](../../deploy/examples/native-ec-dynamodb/README.md) creates a retained table and attaches exact-table permissions to a supplied task role. It does not create or deploy an ECS service. +The separate examples unit provides `deploy/examples/native-ec-dynamodb/`. This self-contained Terraform example creates a retained table and attaches exact-table permissions to a supplied task role. It does not create or deploy an ECS service. | Item | Required layout or owner | | ------------------- | ----------------------------------------------------------------------------------------------------------------------- |