diff --git a/.env.example b/.env.example index 471d38c..2b679a4 100644 --- a/.env.example +++ b/.env.example @@ -27,8 +27,21 @@ SESSION_SECRET= # length is fine, but treat it as a real secret. SERVER_SECRET= -# Spotify Web API uses PKCE; no client secret is needed. +# PostgreSQL connection URL. Required: stores short-lived consumed OAuth +# states, replay nonces, and encrypted one-time credentials. +DATABASE_URL=postgres://postgres@localhost:5432/integration_proxy + +# Base64url-encoded, random 32-byte key for versioned XChaCha20-Poly1305 +# credential envelopes. Required. Generate one with, e.g.: +# openssl rand -base64 32 | tr '+/' '-_' | tr -d '=\n' +ENCRYPTION_KEY= + +# Spotify Web API uses PKCE; no client secret is needed. Its authorization +# server advertises only the public "none" client authentication method, so +# that must be set explicitly (the proxy otherwise defaults to +# client_secret_post, which requires a secret). OAUTH_SPOTIFY_CLIENT_ID= +OAUTH_SPOTIFY_CLIENT_AUTH_METHOD=none # Discord OAuth application; callback ${BASE_URL}/oauth/discord/callback OAUTH_DISCORD_CLIENT_ID= diff --git a/README.md b/README.md index ce5f7a1..a58d391 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,12 @@ # integration-proxy -A stateless Rust web server that lets a user log in with a configured OIDC -provider or a catalog-trusted authenticated API identity. Sign-in sets an encrypted session cookie; there is a logout button -to clear it. No database, no server-side session store — the cookie *is* -the session, so any number of instances can run behind a load balancer with -no shared session state. PostgreSQL stores only short-lived consumed OAuth -states, replay nonces, and encrypted one-time credentials. +A Rust web server that lets a user log in with a configured OIDC provider or +a catalog-trusted authenticated API identity. Sign-in sets an encrypted +session cookie; there is a logout button to clear it. There is no +server-side session store — the cookie *is* the session, so any number of +instances can run behind a load balancer with no shared session state. +PostgreSQL is required, though: it stores short-lived consumed OAuth states, +replay nonces, and encrypted one-time credentials. Built with [axum](https://github.com/tokio-rs/axum) and the [`oauth2`](https://docs.rs/oauth2) crate, following the OAuth 2.0 @@ -62,8 +63,17 @@ Configure an OIDC authorization-code client and register `/auth/callba ### 2. Configure environment variables -Copy `.env.example` to `.env` and fill it in (or export the variables -directly): +Copy `.env.example` to `.env` and fill it in, then load it into your shell +before running the server — nothing in the process reads `.env` files on its +own, only actual process environment variables: + +```sh +set -a +source .env +set +a +``` + +Or export the variables directly without a `.env` file. | Variable | Required | Description | | ----------------------| -------- | ---------------------------------------------------------------------------- | @@ -97,21 +107,26 @@ OAuth credentials are provider-specific. For a catalog platform named letters, digits, and hyphens, and are converted to uppercase with hyphens replaced by underscores for environment-variable names. -The server owns the OAuth endpoints and scopes. The built-in providers are -`google-calendar` (read-only Calendar scope) and `github-issues` (`repo` -scope); a request cannot supply a provider URL, token URL, or scope. +The server owns the OAuth endpoints and scopes for every platform in the +catalog, reading them from that platform's composed OpenAPI document; a +request cannot supply a provider URL, token URL, or scope. The PostgreSQL client validates the database TLS certificate. Heroku assigns `DATABASE_URL` automatically when its Postgres add-on is attached. -Use the `connection_code` returned by the OAuth redirect as the Bearer token -for `/proxy/{platform}/{path}`. Each successful proxy response includes a new -single-use value in `X-Connection-Code`; use that value for the next request. -The proxy refreshes an expired provider access token when a refresh token is -available, and rotates the handoff code after every request. GitHub sometimes -returns repository pagination links using its canonical numeric repository -path; the proxy rewrites that metadata to the current allowlisted owner/repo -path only when the collection suffix matches. +These two flows hand back a proxy credential differently. In the legacy +signed `/connect` and `/oauth/{platform}/start` flow, use the +`connection_code` returned directly by the OAuth redirect as the Bearer token +for `/proxy/{platform}/{path}`. In the browser bootstrap flow (`POST +/connect/authorize`), the OAuth callback instead returns a short-lived, +PKCE-bound handoff code that is **not** a proxy credential; redeem it first +at `POST /connect/redeem` (see above) to obtain the actual `connection_code`. +Either way, each successful proxy response includes a new single-use value +in `X-Connection-Code`; use that value as the Bearer token for the next +request. The proxy refreshes an expired provider access token when a refresh +token is available, and rotates the connection code after every request. +Pagination `Link` headers from the upstream are forwarded to the caller +unchanged. ## Trusted API identities @@ -143,8 +158,9 @@ Discord access tokens expire and use the existing refresh-token flow. Spotify uses `OAUTH_SPOTIFY_CLIENT_ID` and the callback `https://localthought.io/oauth/spotify/callback` in production. Register a Spotify Web API app with that exact redirect URI. The integration uses -Authorization Code with PKCE, so no client secret is required or transmitted. -It imports playlists with `playlist-read-private` and +Authorization Code with PKCE, so no client secret is required or transmitted; +set `OAUTH_SPOTIFY_CLIENT_AUTH_METHOD=none` so the proxy does not require or +send one. It imports playlists with `playlist-read-private` and `playlist-read-collaborative`; no write scopes are requested. No account ID parameter is needed. Development-mode access is subject to Spotify's Premium and app-user allowlist requirements. Access tokens refresh automatically; diff --git a/SECURITY.md b/SECURITY.md index c2528cd..b76b1c1 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -10,7 +10,7 @@ Provider OAuth state binds the platform, tenant, user, callback, PKCE verifier a A new callback returns only a random five-minute handoff code, never a token envelope or tenant secret. `/connect/redeem` requires the original PKCE verifier and atomically deletes the matching, unexpired handoff before issuing one rotating connection credential. Wrong verifiers do not burn a valid handoff; replays and concurrent second redemptions fail. The encrypted database payload carries platform/tenant/user identity and the exact requested grant. Revocation is checked again at redemption. Handoff codes cannot be used directly as proxy credentials. Redemption and consent responses are non-cacheable and use a no-referrer policy. -The browser clears callback parameters before further use, retains pending state outside graph resources, and removes its verifier when redeeming. A lost redemption response requires reconnecting rather than blindly retrying. The existing five-minute idle expiry and one-use rotation rules still apply to proxy credentials. +The browser clears callback parameters before further use, retains pending state outside graph resources, and removes its verifier when redeeming. A lost redemption response requires reconnecting rather than blindly retrying. The existing ten-minute idle expiry and one-use rotation rules still apply to proxy credentials (the handoff code above keeps its own, separate five-minute lifetime). The legacy tenant-proof endpoints remain for compatibility. They are not used by the new hub flow. The legacy tenant-secret redirect is deprecated; new callers must request the optional grant through PKCE redemption. No existing tenant or provider secrets are rotated by this deployment. @@ -27,6 +27,9 @@ subsequent consumption. ## OAuth (#9) +The following are the requirements this deployment targets; see "Release +gate" below for what is still outstanding rather than already implemented. + Register a distinct redirect URI per provider and validate it exactly. Keep the OAuth state and PKCE verifier in authenticated, short-lived, `Secure`, `HttpOnly`, `SameSite=Lax` cookies. Bind the state to the tenant and user id; @@ -56,9 +59,26 @@ credential. Do not forward the upstream's cookies or authorization headers. Rate-limit per tenant, audit token use without logging secrets, and return generic authentication errors. +Implemented today: the requested path is rejected if it contains a `.` or +`..` segment before catalog validation, so it cannot normalize to a +different path than the one authorized (see `proxy::contains_traversal_segment`). +Upstream requests use a bounded connect/read timeout and disable automatic +redirects, so a redirect cannot send a request to a destination the catalog +never validated. The upstream response body is read incrementally and capped +at 10 MiB instead of being buffered in full before the limit is checked. +Request validation against the OAD (`Catalog::validate_request`) is a bounded +subset: it checks that declared *required* query parameters are present, +that an enum-constrained query parameter's value is one of the declared +values, and that a request body's presence and content type match the +operation's declared `requestBody`. It does not validate full JSON Schema +for bodies or non-enum query parameter values. + ## Release gate Provider callback URLs and credential variable names are now deterministic -from catalog platform names. The remaining gate for #9 and #10 is provider -registration, OAD request validation, SSRF tests, and an external review of -the token envelope format before live credentials are handled. +from catalog platform names. Path-traversal rejection, upstream redirect +disabling, and the bounded OAD request validation above are implemented. +The remaining gate for #9 and #10 is provider registration, full JSON Schema +body validation, further SSRF hardening (e.g. blocking requests to internal +network ranges), rate limiting, and an external review of the token envelope +format before live credentials are handled. diff --git a/src/catalog.rs b/src/catalog.rs index 1fdc4b3..1d03734 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -203,6 +203,136 @@ impl Catalog { } Some(out) } + /// A bounded, explicit request validation against the composed OAD: + /// every declared *required* query parameter must be present, a + /// declared enum-constrained query parameter's value must be one of the + /// declared values, and a request body's presence and content type must + /// match the operation's declared `requestBody`. This does not validate + /// full JSON Schema for bodies or non-enum query parameter values; see + /// SECURITY.md for the documented scope. + pub fn validate_request( + &self, + platform: &str, + method: &str, + path: &str, + query: Option<&str>, + content_type: Option<&str>, + has_body: bool, + ) -> Result<(), &'static str> { + let document: Value = + serde_yaml::from_str(self.documents.get(platform).ok_or("unknown platform")?) + .map_err(|_| "invalid catalog document")?; + let server = document + .get("servers") + .and_then(Value::as_array) + .and_then(|servers| servers.first()) + .and_then(|server| server.get("url")) + .and_then(Value::as_str) + .ok_or("invalid catalog document")?; + let server_url = url::Url::parse(server).map_err(|_| "invalid catalog document")?; + let relative = path + .strip_prefix(server_url.path().trim_end_matches('/')) + .ok_or("method or path is not in the catalog")?; + let paths = document + .get("paths") + .and_then(Value::as_object) + .ok_or("invalid catalog document")?; + let template = paths + .keys() + .find(|t| path_matches(t, relative)) + .ok_or("method or path is not in the catalog")?; + let path_item = paths + .get(template) + .and_then(Value::as_object) + .ok_or("invalid catalog document")?; + let operation = path_item + .get(&method.to_ascii_lowercase()) + .and_then(Value::as_object) + .ok_or("method or path is not in the catalog")?; + + let query_pairs: Vec<(String, String)> = query + .map(|q| { + url::form_urlencoded::parse(q.as_bytes()) + .into_owned() + .collect() + }) + .unwrap_or_default(); + for parameter in operation + .get("parameters") + .into_iter() + .chain(path_item.get("parameters")) + .flat_map(Value::as_array) + .flatten() + { + let parameter = if let Some(reference) = parameter.get("$ref").and_then(Value::as_str) { + document + .pointer( + reference + .strip_prefix('#') + .ok_or("invalid catalog document")?, + ) + .ok_or("invalid catalog document")? + } else { + parameter + }; + if parameter.get("in").and_then(Value::as_str) != Some("query") { + continue; + } + let name = parameter + .get("name") + .and_then(Value::as_str) + .ok_or("invalid catalog document")?; + let values: Vec<&str> = query_pairs + .iter() + .filter(|(key, _)| key == name) + .map(|(_, value)| value.as_str()) + .collect(); + if values.is_empty() { + if parameter.get("required").and_then(Value::as_bool) == Some(true) { + return Err("missing required query parameter"); + } + continue; + } + if let Some(allowed) = parameter + .get("schema") + .and_then(|schema| schema.get("enum")) + .and_then(Value::as_array) + { + let allowed: Vec<&str> = allowed.iter().filter_map(Value::as_str).collect(); + if values.iter().any(|value| !allowed.contains(value)) { + return Err("query parameter value is not permitted"); + } + } + } + + match operation.get("requestBody") { + None => { + if has_body { + return Err("operation does not accept a request body"); + } + } + Some(request_body) => { + if !has_body { + if request_body.get("required").and_then(Value::as_bool) == Some(true) { + return Err("operation requires a request body"); + } + return Ok(()); + } + let content = request_body + .get("content") + .and_then(Value::as_object) + .ok_or("invalid catalog document")?; + let media_type = content_type + .and_then(|value| value.split(';').next()) + .map(str::trim) + .unwrap_or(""); + if !content.contains_key(media_type) { + return Err("unsupported content type"); + } + } + } + Ok(()) + } pub fn oauth_provider(&self, platform: &str) -> Result { let source = self.get(platform).ok_or("unknown catalog platform")?; let document = serde_yaml::from_str(source).map_err(|_| "invalid catalog document")?; @@ -529,6 +659,84 @@ mod tests { assert_eq!(headers(&doc).unwrap().len(), 2); } + #[test] + fn validate_request_checks_required_query_parameters_enums_and_body() { + use serde_json::json; + let doc = json!({"servers":[{"url":"https://example.com/v1"}], "paths":{"/items": { + "get":{"parameters":[ + {"in":"query","name":"state","required":true,"schema":{"enum":["open","closed"]}}, + {"in":"query","name":"page","schema":{"type":"integer"}} + ]}, + "post": {"requestBody": {"required": true, "content": {"application/json": {}}}} + }}}); + let catalog = super::Catalog { + documents: [("test".into(), serde_yaml::to_string(&doc).unwrap())].into(), + ..Default::default() + }; + + // Missing a required query parameter. + assert!(catalog + .validate_request("test", "GET", "/v1/items", None, None, false) + .is_err()); + // Required parameter present with a disallowed value. + assert!(catalog + .validate_request( + "test", + "GET", + "/v1/items", + Some("state=pending"), + None, + false + ) + .is_err()); + // Valid required value, plus an undeclared parameter that is simply ignored. + assert!(catalog + .validate_request( + "test", + "GET", + "/v1/items", + Some("state=open&page=2&undeclared=1"), + None, + false + ) + .is_ok()); + + // A body is rejected when the operation declares none. + assert!(catalog + .validate_request( + "test", + "GET", + "/v1/items", + Some("state=open"), + Some("application/json"), + true + ) + .is_err()); + // A required body that is missing is rejected. + assert!(catalog + .validate_request("test", "POST", "/v1/items", None, None, false) + .is_err()); + // An unsupported content type is rejected. + assert!(catalog + .validate_request("test", "POST", "/v1/items", None, Some("text/plain"), true) + .is_err()); + // A matching content type (with parameters, e.g. a charset) is accepted. + assert!(catalog + .validate_request( + "test", + "POST", + "/v1/items", + None, + Some("application/json; charset=utf-8"), + true + ) + .is_ok()); + // An unknown path/method is rejected the same way as `allows`. + assert!(catalog + .validate_request("test", "DELETE", "/v1/items", None, None, false) + .is_err()); + } + #[test] fn composed_notion_preserves_base_path_and_resolves_version_header() { let catalog = super::Catalog { @@ -550,6 +758,47 @@ mod tests { } use super::*; + /// Exercises the exact catalog revision the application loads by + /// default (`Config::from_env`'s `CATALOG_PATH` fallback), so a pass here + /// establishes that the deployed configuration actually supplies these + /// OAuth profiles. `pinned_catalog_selects_google_offline_and_spotify_pkce_profiles` + /// below covers only a separately identified, non-default revision. + #[tokio::test] + #[ignore = "downloads the pinned production catalog sources"] + async fn default_catalog_pin_selects_google_offline_and_spotify_pkce_profiles() { + let catalog = Catalog::load( + crate::config::DEFAULT_CATALOG_PATH, + &crate::build_http_client(), + ) + .await + .unwrap(); + for name in catalog.names() { + let provider = catalog + .oauth_provider(&name) + .unwrap_or_else(|error| panic!("{name}: {error}")); + // Notion's security scheme declares no per-operation OAuth + // scopes (it authorizes by integration capability instead), so + // it is the one platform excluded from the non-empty check. + if name != "notion" { + assert!(!provider.scopes.is_empty(), "{name}"); + } + } + let google = catalog.oauth_provider("google-calendar").unwrap(); + assert!(google.use_pkce); + assert!(google + .authorization_params + .contains(&("access_type".into(), "offline".into()))); + let spotify = catalog.oauth_provider("spotify").unwrap(); + assert!(spotify.use_pkce); + assert!(catalog + .allows("github-issues", "GET", "/repositories/123/issues") + .is_some()); + } + + /// Tests a specific, separately identified catalog revision (not the + /// application's default pin; see `TEST_CATALOG_PATH` below) for + /// regression coverage of canonical GitHub paths and Moneybird's + /// composed resource/throttling metadata. #[tokio::test] #[ignore = "downloads the pinned production catalog sources"] async fn pinned_catalog_supplies_oauth_and_canonical_pagination_paths() { @@ -606,6 +855,9 @@ mod tests { .is_none()); } + /// Covers a separately identified, non-default catalog revision; see + /// `default_catalog_pin_selects_google_offline_and_spotify_pkce_profiles` + /// above for coverage of the revision the application actually deploys. #[tokio::test] #[ignore = "downloads pinned OAuth authentication-details metadata"] async fn pinned_catalog_selects_google_offline_and_spotify_pkce_profiles() { diff --git a/src/config.rs b/src/config.rs index 2143229..c386b2d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -55,18 +55,28 @@ pub struct Config { pub revoked_subjects: Vec, } +/// Immutable pinned revision of `localthought/overlays`' `catalog.json` used +/// when `CATALOG_PATH` is not set. Shared with tests that need to validate +/// the exact catalog the application would load by default. +pub const DEFAULT_CATALOG_PATH: &str = "https://raw.githubusercontent.com/localthought/overlays/8f29d9973267b6b3877aa27a5ab50cd41b010e6c/catalog.json"; + +/// Reads a required environment variable and rejects it if unset or blank, +/// so a blank `.env` value fails configuration explicitly instead of being +/// silently accepted (e.g. as a reproducible empty secret). +fn require_env(name: &str) -> Result { + match env::var(name) { + Ok(value) if !value.is_empty() => Ok(value), + _ => Err(format!("{name} must be set")), + } +} + impl Config { pub fn from_env() -> Result { - let app_auth_client_id = env::var("APP_AUTH_CLIENT_ID") - .map_err(|_| "APP_AUTH_CLIENT_ID must be set".to_string())?; - let app_auth_client_secret = env::var("APP_AUTH_CLIENT_SECRET") - .map_err(|_| "APP_AUTH_CLIENT_SECRET must be set".to_string())?; - let app_auth_authorization_url = env::var("APP_AUTH_AUTHORIZATION_URL") - .map_err(|_| "APP_AUTH_AUTHORIZATION_URL must be set".to_string())?; - let app_auth_token_url = env::var("APP_AUTH_TOKEN_URL") - .map_err(|_| "APP_AUTH_TOKEN_URL must be set".to_string())?; - let app_auth_userinfo_url = env::var("APP_AUTH_USERINFO_URL") - .map_err(|_| "APP_AUTH_USERINFO_URL must be set".to_string())?; + let app_auth_client_id = require_env("APP_AUTH_CLIENT_ID")?; + let app_auth_client_secret = require_env("APP_AUTH_CLIENT_SECRET")?; + let app_auth_authorization_url = require_env("APP_AUTH_AUTHORIZATION_URL")?; + let app_auth_token_url = require_env("APP_AUTH_TOKEN_URL")?; + let app_auth_userinfo_url = require_env("APP_AUTH_USERINFO_URL")?; let app_auth_label = env::var("APP_AUTH_LABEL").unwrap_or_else(|_| "OIDC".to_string()); let app_auth_identity_namespace = identity_namespace(env::var("APP_AUTH_IDENTITY_NAMESPACE").ok())?; @@ -75,16 +85,18 @@ impl Config { .ok() .and_then(|p| p.parse().ok()) .unwrap_or(8080); - let session_secret = env::var("SESSION_SECRET").ok(); - let server_secret = - env::var("SERVER_SECRET").map_err(|_| "SERVER_SECRET must be set".to_string())?; - let catalog_path = env::var("CATALOG_PATH").unwrap_or_else(|_| { - "https://raw.githubusercontent.com/localthought/overlays/8f29d9973267b6b3877aa27a5ab50cd41b010e6c/catalog.json".to_string() - }); - let database_url = env::var("DATABASE_URL") + // An empty SESSION_SECRET is treated as unset, so it triggers random + // key generation instead of being hashed into a reproducible, + // guessable cookie-encryption key. + let session_secret = env::var("SESSION_SECRET") + .ok() + .filter(|value| !value.is_empty()); + let server_secret = require_env("SERVER_SECRET")?; + let catalog_path = + env::var("CATALOG_PATH").unwrap_or_else(|_| DEFAULT_CATALOG_PATH.to_string()); + let database_url = require_env("DATABASE_URL") .map_err(|_| "DATABASE_URL must be set for replay protection".to_string())?; - let encryption_key = - env::var("ENCRYPTION_KEY").map_err(|_| "ENCRYPTION_KEY must be set".to_string())?; + let encryption_key = require_env("ENCRYPTION_KEY")?; let revoked_subjects = env::var("REVOKED_SUBJECTS") .unwrap_or_default() .split(',') @@ -158,6 +170,18 @@ mod tests { assert!(Config::provider_env_prefix("Google Calendar").is_err()); } + #[test] + fn require_env_rejects_missing_and_blank_values() { + let name = "INTEGRATION_PROXY_TEST_REQUIRE_ENV_VAR"; + env::remove_var(name); + assert!(require_env(name).is_err()); + env::set_var(name, ""); + assert!(require_env(name).is_err()); + env::set_var(name, "a-value"); + assert_eq!(require_env(name).unwrap(), "a-value"); + env::remove_var(name); + } + #[test] fn legacy_identity_namespace_is_blank_or_fixed_https_only() { assert_eq!(identity_namespace(Some(" ".into())).unwrap(), None); diff --git a/src/identity.rs b/src/identity.rs index 5ee017b..b470a38 100644 --- a/src/identity.rs +++ b/src/identity.rs @@ -170,7 +170,7 @@ pub fn parse(document: &Value, selection: &Value) -> Result for Key { } fn build_http_client() -> reqwest::Client { - // GitHub's REST API requires a User-Agent on every request. + // GitHub's REST API requires a User-Agent on every request. Redirects are + // disabled because only the initial target is validated against the + // catalog allowlist; a followed redirect would escape that validation. + // A bounded timeout keeps a stalled or slow upstream from holding the + // connection (and the caller's request) open indefinitely. reqwest::Client::builder() .user_agent("LocalThought-integration-proxy") + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(std::time::Duration::from_secs(10)) + .timeout(std::time::Duration::from_secs(30)) .build() .expect("failed to build HTTP client") } @@ -185,6 +192,7 @@ fn browser_cors() -> tower_http::cors::CorsLayer { fn router(state: AppState) -> Router { Router::new() .route("/", get(home)) + .route("/healthz", get(healthz)) .route("/logo.png", get(logo)) .route("/auth/login", get(auth::login)) .route("/auth/login/:platform", get(api_login::start)) @@ -205,6 +213,19 @@ fn router(state: AppState) -> Router { .with_state(state) } +/// Reports whether the database is currently reachable, so an operator or +/// load balancer can detect a still-recovering connection instead of only +/// finding out from a failed request. +async fn healthz(State(state): State) -> impl axum::response::IntoResponse { + match &state.security { + Some(security) if security.is_ready().await => (axum::http::StatusCode::OK, "ok"), + _ => ( + axum::http::StatusCode::SERVICE_UNAVAILABLE, + "database unavailable", + ), + } +} + async fn logo() -> impl axum::response::IntoResponse { ( [ diff --git a/src/providers.rs b/src/providers.rs index 288662b..f6db384 100644 --- a/src/providers.rs +++ b/src/providers.rs @@ -188,17 +188,7 @@ impl Provider { None => ClientAuth::SecretPost, }, }; - if self - .token_operation - .as_ref() - .is_some_and(|op| op.requires_basic) - && client_auth != ClientAuth::SecretBasic - || self - .refresh_operation - .as_ref() - .is_some_and(|op| op.requires_basic) - && client_auth != ClientAuth::SecretBasic - { + if requires_basic && client_auth != ClientAuth::SecretBasic { return Err("OAuth token operation requires client_secret_basic".into()); } if self @@ -476,10 +466,10 @@ fn pkce_behavior(scheme: &Value) -> Result { .is_some_and(|values| { !values.is_empty() && values.iter().all(|value| value.is_string()) }) - || !pkce + || pkce .get("description") .and_then(Value::as_str) - .is_some_and(|value| !value.trim().is_empty()) + .is_none_or(|value| value.trim().is_empty()) { return Err("conditional PKCE requires requiredFor and description".into()); } diff --git a/src/proxy.rs b/src/proxy.rs index 800fabd..fe1f28a 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -251,6 +251,14 @@ async fn refresh_if_needed(state: &AppState, credential: &mut Credential) -> Res let refresh_token = credential.refresh_token.as_deref().ok_or(())?; let provider = crate::providers::Provider::configured(&state.catalog, &credential.provider) .map_err(|_| ())?; + #[cfg(test)] + let provider = { + let mut provider = provider; + if let Some(upstream) = &state.test_upstream { + provider.provider.token_url = format!("{}/token", upstream.trim_end_matches('/')); + } + provider + }; let response = provider .token_request( &state.http_client, @@ -288,6 +296,14 @@ pub async fn forward( ) .into_response(); }; + let request_path = format!("/{path}"); + if contains_traversal_segment(&request_path) { + return ( + StatusCode::BAD_REQUEST, + "path must not contain traversal segments", + ) + .into_response(); + } if body.len() > 1_048_576 { return (StatusCode::PAYLOAD_TOO_LARGE, "request body is too large").into_response(); } @@ -326,7 +342,6 @@ pub async fn forward( if refresh_if_needed(&state, &mut credential).await.is_err() { return (StatusCode::UNAUTHORIZED, "credential refresh failed").into_response(); } - let request_path = format!("/{path}"); let Some(required_headers) = state .catalog @@ -348,6 +363,18 @@ pub async fn forward( ) .into_response(); }; + if let Err(message) = state.catalog.validate_request( + platform, + method.as_str(), + &request_path, + query.as_deref(), + headers + .get(header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()), + !body.is_empty(), + ) { + return (StatusCode::BAD_REQUEST, message).into_response(); + } target.set_path(&request_path); target.set_query(query.as_deref()); let upstream = match upstream_request( @@ -367,9 +394,9 @@ pub async fn forward( }; let status = upstream.status(); let forwarded_headers = upstream_response_headers(upstream.headers()); - let bytes = match upstream.bytes().await { - Ok(bytes) if bytes.len() <= 10_485_760 => bytes, - _ => { + let bytes = match read_bounded_body(upstream, MAX_RESPONSE_BYTES).await { + Ok(bytes) => bytes, + Err(_) => { return ( StatusCode::BAD_GATEWAY, "upstream response failed or was too large", @@ -414,7 +441,14 @@ pub async fn forward( // Forward only representation/pagination metadata, never provider cookies or credentials. fn upstream_response_headers(headers: &HeaderMap) -> HeaderMap { let mut result = HeaderMap::new(); - for name in [header::CONTENT_TYPE, header::LINK, header::RETRY_AFTER] { + for name in [ + header::CONTENT_TYPE, + header::LINK, + header::RETRY_AFTER, + header::ETAG, + axum::http::HeaderName::from_static("x-total-count"), + axum::http::HeaderName::from_static("x-next-page"), + ] { for value in headers.get_all(&name) { result.append(name.clone(), value.clone()); } @@ -422,6 +456,36 @@ fn upstream_response_headers(headers: &HeaderMap) -> HeaderMap { result } +const MAX_RESPONSE_BYTES: usize = 10_485_760; + +/// Reads the upstream body incrementally so an oversized or slow response is +/// rejected as soon as the limit is crossed, instead of after it is fully +/// buffered in memory. +async fn read_bounded_body(mut upstream: reqwest::Response, limit: usize) -> Result { + if upstream + .content_length() + .is_some_and(|len| len > limit as u64) + { + return Err(()); + } + let mut buffer = Vec::new(); + while let Some(chunk) = upstream.chunk().await.map_err(|_| ())? { + if buffer.len() + chunk.len() > limit { + return Err(()); + } + buffer.extend_from_slice(&chunk); + } + Ok(Bytes::from(buffer)) +} + +/// Rejects any `.` or `..` path segment so a client cannot request a path +/// that, once assigned to the upstream URL, normalizes to a different path +/// than the one validated against the catalog allowlist. +fn contains_traversal_segment(path: &str) -> bool { + path.split('/') + .any(|segment| segment == "." || segment == "..") +} + fn upstream_request( client: &reqwest::Client, method: axum::http::Method, @@ -475,6 +539,7 @@ mod tests { use super::*; use axum::http::{header::AUTHORIZATION, HeaderValue}; use axum_extra::extract::cookie::Key; + use tower::ServiceExt; fn test_state(server_secret: &str) -> AppState { let config = crate::config::Config { @@ -611,6 +676,216 @@ mod tests { assert!(!forwarded.contains_key("x-connection-code")); } + #[test] + fn traversal_segments_are_rejected_including_encoded_forms() { + assert!(contains_traversal_segment("/repositories/../issues")); + assert!(contains_traversal_segment("/repositories/./issues")); + assert!(contains_traversal_segment("/../etc/passwd")); + assert!(!contains_traversal_segment("/repositories/123/issues")); + assert!(!contains_traversal_segment("/repositories/..foo/issues")); + } + + #[tokio::test] + async fn forward_rejects_a_path_containing_traversal_segments_before_touching_credentials() { + let mut state = test_state("server-secret"); + state.catalog = crate::catalog::Catalog::from_test_document( + "github-issues", + json!({ + "servers": [{"url": "https://api.example"}], + "paths": {"/repositories/{id}/issues": {"get": {}}} + }), + json!({}), + ); + // No Authorization header and no security service configured: if the + // traversal check did not run first, this would fail with 401/503 + // instead of 400. + let response = crate::router(state) + .oneshot( + axum::http::Request::builder() + .uri("/proxy/github-issues/repositories/../issues") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[tokio::test] + async fn read_bounded_body_rejects_a_response_over_the_limit() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let app = axum::Router::new().route("/big", axum::routing::get(|| async { vec![0u8; 20] })); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let client = crate::build_http_client(); + let response = client + .get(format!("http://{address}/big")) + .send() + .await + .unwrap(); + assert!(read_bounded_body(response, 10).await.is_err()); + + let response = client + .get(format!("http://{address}/big")) + .send() + .await + .unwrap(); + assert!(read_bounded_body(response, 20).await.is_ok()); + server.abort(); + } + + /// End-to-end coverage of `forward()` through the real router: an + /// expired credential is refreshed at the (mocked) provider token + /// endpoint, the resulting access token is used to call the (mocked) + /// provider API, the response is forwarded with its pagination/ETag + /// headers intact, and the one-time connection code is rotated. + #[tokio::test] + #[ignore = "requires TEST_DATABASE_URL and fixture OAuth credentials"] + async fn forward_refreshes_an_expired_credential_and_forwards_the_response() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let token_requests = std::sync::Arc::new(AtomicUsize::new(0)); + let item_requests = std::sync::Arc::new(AtomicUsize::new(0)); + let token_counter = token_requests.clone(); + let item_counter = item_requests.clone(); + let upstream = axum::Router::new() + .route( + "/token", + axum::routing::post(move |body: Bytes| { + let token_counter = token_counter.clone(); + async move { + token_counter.fetch_add(1, Ordering::SeqCst); + assert!(String::from_utf8(body.to_vec()) + .unwrap() + .contains("grant_type=refresh_token")); + Json(json!({"access_token": "refreshed-token", "expires_in": 3600})) + } + }), + ) + .route( + "/items", + axum::routing::get(move |headers: HeaderMap| { + let item_counter = item_counter.clone(); + async move { + item_counter.fetch_add(1, Ordering::SeqCst); + assert_eq!( + headers.get(AUTHORIZATION).unwrap(), + "Bearer refreshed-token" + ); + ( + [ + (header::ETAG, "\"v1\""), + (header::LINK, "; rel=\"next\""), + ], + Json(json!([{"id": 1}])), + ) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let upstream_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { axum::serve(listener, upstream).await.unwrap() }); + + let db = std::env::var("TEST_DATABASE_URL").unwrap(); + let security = crate::security::Security::connect( + &db, + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + vec![], + ) + .await + .unwrap(); + + let document = json!({ + "servers": [{"url": upstream_url}], + "components": {"securitySchemes": {"oauth": {"type": "oauth2", "flows": { + "authorizationCode": { + "authorizationUrl": "https://auth.example/authorize", + "tokenUrl": "https://auth.example/token", + "scopes": {"read": "Read items"} + } + }}}}, + "security": [{"oauth": ["read"]}], + "paths": {"/items": {"get": {}}} + }); + let mut state = test_state("server-secret"); + state.catalog = + crate::catalog::Catalog::from_test_document("github-issues", document, json!({})); + state.security = Some(security.clone()); + state.test_upstream = Some(upstream_url); + + let credential = Credential { + provider: "github-issues".into(), + tenant_id: "tenant".into(), + user_id: "user".into(), + access_token: "stale-token".into(), + refresh_token: Some("a-refresh-token".into()), + expires_at: Some(0), + }; + let envelope = security + .seal( + &serde_json::to_vec(&credential).unwrap(), + b"connection-credential-v1", + ) + .unwrap(); + let old_code = "e2e-test-connection-code"; + security + .store_connection_code(old_code, &envelope) + .await + .unwrap(); + + let response = crate::router(state) + .oneshot( + axum::http::Request::builder() + .uri("/proxy/github-issues/items") + .header(AUTHORIZATION, format!("Bearer {old_code}")) + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(token_requests.load(Ordering::SeqCst), 1); + assert_eq!(item_requests.load(Ordering::SeqCst), 1); + let new_code = response + .headers() + .get("x-connection-code") + .unwrap() + .to_str() + .unwrap() + .to_owned(); + assert_ne!(new_code, old_code); + assert_eq!(response.headers()[header::ETAG], "\"v1\""); + assert!(response.headers().get(header::LINK).is_some()); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!([{"id": 1}]) + ); + + // The old code is single-use and already consumed. + assert!(security + .take_connection_code(old_code) + .await + .unwrap() + .is_none()); + // The rotated code is valid and carries the refreshed credential. + let rotated_envelope = security + .take_connection_code(&new_code) + .await + .unwrap() + .unwrap(); + let plaintext = security + .open(&rotated_envelope, b"connection-credential-v1") + .unwrap(); + let rotated: Credential = serde_json::from_slice(&plaintext).unwrap(); + assert_eq!(rotated.access_token, "refreshed-token"); + + server.abort(); + } + fn logged_in_jar(key: Key) -> (PrivateCookieJar, crate::session::SessionUser) { let user = crate::session::SessionUser::new( "oidc-sub-123".to_string(), diff --git a/src/security.rs b/src/security.rs index 3da5acb..344d70d 100644 --- a/src/security.rs +++ b/src/security.rs @@ -1,4 +1,4 @@ -use std::{collections::HashSet, sync::Arc}; +use std::{collections::HashSet, pin::Pin, sync::Arc, time::Duration}; use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; use chacha20poly1305::{ @@ -6,8 +6,55 @@ use chacha20poly1305::{ XChaCha20Poly1305, XNonce, }; use rand::RngCore; +use tokio::sync::RwLock; use tokio_postgres::Client; +type ConnectionDriver = + Pin> + Send>>; + +async fn connect_once(database_url: &str) -> Result<(Client, ConnectionDriver), String> { + let tls = native_tls::TlsConnector::new().map_err(|e| e.to_string())?; + let tls = postgres_native_tls::MakeTlsConnector::new(tls); + let (client, connection) = tokio_postgres::connect(database_url, tls) + .await + .map_err(|e| e.to_string())?; + Ok((client, Box::pin(connection))) +} + +/// Drives the active connection to completion, then keeps reconnecting with +/// exponential backoff and swapping the shared client in on success, so a +/// dropped connection recovers without restarting the process. +fn spawn_reconnect_supervisor( + database_url: String, + current: Arc>>, + mut connection: ConnectionDriver, +) { + tokio::spawn(async move { + loop { + if let Err(error) = connection.await { + tracing::error!(%error, "postgres connection failed; reconnecting"); + } else { + tracing::warn!("postgres connection closed; reconnecting"); + } + let mut backoff = Duration::from_secs(1); + loop { + match connect_once(&database_url).await { + Ok((client, new_connection)) => { + *current.write().await = Arc::new(client); + connection = new_connection; + break; + } + Err(error) => { + tracing::error!(%error, ?backoff, "postgres reconnect attempt failed; retrying"); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(Duration::from_secs(30)); + } + } + } + } + }); +} + pub struct OAuthState { pub provider: String, pub redirect_uri: String, @@ -19,7 +66,7 @@ pub struct OAuthState { #[derive(Clone)] pub struct Security { - database: Arc, + database: Arc>>, encryption_key: [u8; 32], revoked_subjects: HashSet, } @@ -36,44 +83,62 @@ impl Security { let encryption_key: [u8; 32] = key .try_into() .map_err(|_| "ENCRYPTION_KEY must decode to exactly 32 bytes")?; - let tls = native_tls::TlsConnector::new().map_err(|e| e.to_string())?; - let tls = postgres_native_tls::MakeTlsConnector::new(tls); - let (mut database, connection) = tokio_postgres::connect(database_url, tls) - .await - .map_err(|e| e.to_string())?; - tokio::spawn(async move { - if let Err(error) = connection.await { - tracing::error!(%error, "postgres connection failed"); - } - }); - let transaction = database.transaction().await.map_err(|e| e.to_string())?; - transaction - .query_one( - "SELECT pg_advisory_xact_lock($1)", - &[&7_316_186_474_691_124_077_i64], - ) - .await - .map_err(|e| e.to_string())?; - transaction.batch_execute("CREATE TABLE IF NOT EXISTS used_challenges (nonce TEXT PRIMARY KEY, expires_at TIMESTAMPTZ NOT NULL); CREATE TABLE IF NOT EXISTS oauth_states (state TEXT PRIMARY KEY, provider TEXT NOT NULL, redirect_uri TEXT NOT NULL, tenant_id TEXT NOT NULL, user_id TEXT NOT NULL, verifier TEXT NOT NULL, expires_at TIMESTAMPTZ NOT NULL); CREATE TABLE IF NOT EXISTS connection_codes (code TEXT PRIMARY KEY, envelope TEXT NOT NULL, expires_at TIMESTAMPTZ NOT NULL); ALTER TABLE oauth_states ADD COLUMN IF NOT EXISTS context TEXT; CREATE TABLE IF NOT EXISTS connection_handoffs (code TEXT PRIMARY KEY, challenge TEXT NOT NULL, envelope TEXT NOT NULL, expires_at TIMESTAMPTZ NOT NULL)").await.map_err(|e| e.to_string())?; - transaction.commit().await.map_err(|e| e.to_string())?; + // Schema setup needs a transaction, which requires exclusive (&mut) + // access to a Client — incompatible with the shared, reconnectable + // client below. Run it on its own short-lived connection first. + { + let (mut setup, setup_connection) = connect_once(database_url).await?; + let driver = tokio::spawn(setup_connection); + let transaction = setup.transaction().await.map_err(|e| e.to_string())?; + transaction + .query_one( + "SELECT pg_advisory_xact_lock($1)", + &[&7_316_186_474_691_124_077_i64], + ) + .await + .map_err(|e| e.to_string())?; + transaction.batch_execute("CREATE TABLE IF NOT EXISTS used_challenges (nonce TEXT PRIMARY KEY, expires_at TIMESTAMPTZ NOT NULL); CREATE TABLE IF NOT EXISTS oauth_states (state TEXT PRIMARY KEY, provider TEXT NOT NULL, redirect_uri TEXT NOT NULL, tenant_id TEXT NOT NULL, user_id TEXT NOT NULL, verifier TEXT NOT NULL, expires_at TIMESTAMPTZ NOT NULL); CREATE TABLE IF NOT EXISTS connection_codes (code TEXT PRIMARY KEY, envelope TEXT NOT NULL, expires_at TIMESTAMPTZ NOT NULL); ALTER TABLE oauth_states ADD COLUMN IF NOT EXISTS context TEXT; CREATE TABLE IF NOT EXISTS connection_handoffs (code TEXT PRIMARY KEY, challenge TEXT NOT NULL, envelope TEXT NOT NULL, expires_at TIMESTAMPTZ NOT NULL); CREATE INDEX IF NOT EXISTS oauth_states_expires_at_idx ON oauth_states (expires_at); CREATE INDEX IF NOT EXISTS connection_codes_expires_at_idx ON connection_codes (expires_at); CREATE INDEX IF NOT EXISTS used_challenges_expires_at_idx ON used_challenges (expires_at); CREATE INDEX IF NOT EXISTS connection_handoffs_expires_at_idx ON connection_handoffs (expires_at)").await.map_err(|e| e.to_string())?; + transaction.commit().await.map_err(|e| e.to_string())?; + drop(setup); + let _ = driver.await; + } + + let (client, connection) = connect_once(database_url).await?; + let database = Arc::new(RwLock::new(Arc::new(client))); + spawn_reconnect_supervisor(database_url.to_string(), database.clone(), connection); Ok(Self { - database: Arc::new(database), + database, encryption_key, revoked_subjects: revoked_subjects.into_iter().collect(), }) } + /// The client currently backing this connection. Held only for the + /// duration of one query: a reconnect replaces the shared client without + /// invalidating a client already in hand. + async fn client(&self) -> Arc { + self.database.read().await.clone() + } + + /// Lightweight readiness probe for callers (health checks, startup + /// diagnostics) that want to know whether the database is currently + /// reachable without triggering the normal error-mapping of a query. + pub async fn is_ready(&self) -> bool { + self.client().await.simple_query("SELECT 1").await.is_ok() + } + pub fn is_revoked(&self, tenant_id: &str, user_id: &str) -> bool { self.revoked_subjects.contains(tenant_id) || self.revoked_subjects.contains(user_id) } /// Atomically records a nonce. A duplicate nonce is a replay. pub async fn consume_nonce(&self, nonce: &str) -> Result { - self.database + let database = self.client().await; + database .execute("DELETE FROM used_challenges WHERE expires_at <= NOW()", &[]) .await .map_err(|e| e.to_string())?; - let rows = self.database.execute("INSERT INTO used_challenges (nonce, expires_at) VALUES ($1, NOW() + INTERVAL '10 minutes') ON CONFLICT DO NOTHING", &[&nonce]).await.map_err(|e| e.to_string())?; + let rows = database.execute("INSERT INTO used_challenges (nonce, expires_at) VALUES ($1, NOW() + INTERVAL '10 minutes') ON CONFLICT DO NOTHING", &[&nonce]).await.map_err(|e| e.to_string())?; Ok(rows == 1) } @@ -122,12 +187,17 @@ impl Security { } pub async fn store_oauth_state(&self, state: &str, value: &OAuthState) -> Result<(), String> { - self.database.execute("INSERT INTO oauth_states (state, provider, redirect_uri, tenant_id, user_id, verifier, context, expires_at) VALUES ($1,$2,$3,$4,$5,$6,$7,NOW() + INTERVAL '10 minutes')", &[&state, &value.provider, &value.redirect_uri, &value.tenant_id, &value.user_id, &value.verifier, &value.context]).await.map_err(|e| e.to_string())?; + let database = self.client().await; + database + .execute("DELETE FROM oauth_states WHERE expires_at <= NOW()", &[]) + .await + .map_err(|e| e.to_string())?; + database.execute("INSERT INTO oauth_states (state, provider, redirect_uri, tenant_id, user_id, verifier, context, expires_at) VALUES ($1,$2,$3,$4,$5,$6,$7,NOW() + INTERVAL '10 minutes')", &[&state, &value.provider, &value.redirect_uri, &value.tenant_id, &value.user_id, &value.verifier, &value.context]).await.map_err(|e| e.to_string())?; Ok(()) } pub async fn take_oauth_state(&self, state: &str) -> Result, String> { - let row = self.database.query_opt("DELETE FROM oauth_states WHERE state = $1 AND expires_at > NOW() RETURNING provider, redirect_uri, tenant_id, user_id, verifier, context", &[&state]).await.map_err(|e| e.to_string())?; + let row = self.client().await.query_opt("DELETE FROM oauth_states WHERE state = $1 AND expires_at > NOW() RETURNING provider, redirect_uri, tenant_id, user_id, verifier, context", &[&state]).await.map_err(|e| e.to_string())?; Ok(row.map(|r| OAuthState { provider: r.get(0), redirect_uri: r.get(1), @@ -144,14 +214,15 @@ impl Security { challenge: &str, envelope: &str, ) -> Result<(), String> { - self.database + let database = self.client().await; + database .execute( "DELETE FROM connection_handoffs WHERE expires_at <= NOW()", &[], ) .await .map_err(|e| e.to_string())?; - self.database.execute("INSERT INTO connection_handoffs (code, challenge, envelope, expires_at) VALUES ($1,$2,$3,NOW() + INTERVAL '5 minutes')", &[&code, &challenge, &envelope]).await.map_err(|e| e.to_string())?; + database.execute("INSERT INTO connection_handoffs (code, challenge, envelope, expires_at) VALUES ($1,$2,$3,NOW() + INTERVAL '5 minutes')", &[&code, &challenge, &envelope]).await.map_err(|e| e.to_string())?; Ok(()) } @@ -161,19 +232,27 @@ impl Security { code: &str, challenge: &str, ) -> Result, String> { - self.database.query_opt("DELETE FROM connection_handoffs WHERE code = $1 AND challenge = $2 AND expires_at > NOW() RETURNING envelope", &[&code, &challenge]).await.map_err(|e| e.to_string()).map(|row| row.map(|r| r.get(0))) + self.client().await.query_opt("DELETE FROM connection_handoffs WHERE code = $1 AND challenge = $2 AND expires_at > NOW() RETURNING envelope", &[&code, &challenge]).await.map_err(|e| e.to_string()).map(|row| row.map(|r| r.get(0))) } // Allow a provider-mandated five-minute Retry-After without expiring the // next single-use credential while the client is waiting. OAuth handoffs // retain their separate five-minute lifetime. pub async fn store_connection_code(&self, code: &str, envelope: &str) -> Result<(), String> { - self.database.execute("INSERT INTO connection_codes (code, envelope, expires_at) VALUES ($1,$2,NOW() + INTERVAL '10 minutes')", &[&code, &envelope]).await.map_err(|e| e.to_string())?; + let database = self.client().await; + database + .execute( + "DELETE FROM connection_codes WHERE expires_at <= NOW()", + &[], + ) + .await + .map_err(|e| e.to_string())?; + database.execute("INSERT INTO connection_codes (code, envelope, expires_at) VALUES ($1,$2,NOW() + INTERVAL '10 minutes')", &[&code, &envelope]).await.map_err(|e| e.to_string())?; Ok(()) } pub async fn take_connection_code(&self, code: &str) -> Result, String> { - self.database.query_opt("DELETE FROM connection_codes WHERE code = $1 AND expires_at > NOW() RETURNING envelope", &[&code]).await.map_err(|e| e.to_string()).map(|row| row.map(|r| r.get(0))) + self.client().await.query_opt("DELETE FROM connection_codes WHERE code = $1 AND expires_at > NOW() RETURNING envelope", &[&code]).await.map_err(|e| e.to_string()).map(|row| row.map(|r| r.get(0))) } } @@ -236,4 +315,56 @@ mod tests { "concurrent schema initialization failed: {errors:?}" ); } + + #[tokio::test] + #[ignore = "requires TEST_DATABASE_URL; CI runs with --include-ignored"] + async fn recovers_after_the_database_connection_is_dropped() { + let database_url = std::env::var("TEST_DATABASE_URL") + .expect("set TEST_DATABASE_URL to an isolated test database"); + let tag = format!("security_reconnect_{:016x}", rand::random::()); + let mut tagged_url = url::Url::parse(&database_url).expect("parse TEST_DATABASE_URL"); + tagged_url + .query_pairs_mut() + .append_pair("application_name", &tag); + let security = Security::connect(tagged_url.as_ref(), TEST_KEY, vec![]) + .await + .expect("initial connection"); + assert!(security.is_ready().await); + + let (admin, admin_connection) = + tokio_postgres::connect(&database_url, tokio_postgres::NoTls) + .await + .expect("admin connection"); + tokio::spawn(async move { + admin_connection.await.expect("admin connection driver"); + }); + let terminated = admin + .execute( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE application_name = $1", + &[&tag], + ) + .await + .expect("terminate the security connection's backend"); + assert!( + terminated > 0, + "expected to terminate at least one backend for {tag}" + ); + + let mut healed = false; + for _ in 0..50 { + if security.is_ready().await { + healed = true; + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + assert!(healed, "connection did not recover after being terminated"); + + // A real write, not just the readiness probe, proves the reconnected + // client is fully usable and schema state survived the reconnect. + assert!(security + .consume_nonce(&format!("post-reconnect-{tag}")) + .await + .expect("query after reconnect")); + } } diff --git a/src/templates.rs b/src/templates.rs index d8a421f..e189b3e 100644 --- a/src/templates.rs +++ b/src/templates.rs @@ -59,7 +59,7 @@ pub fn render_platform_connect(

Connect {platform}

Continue with your {platform} account to establish your LocalThought identity and connect {platform} to this hub.

-

Target Atomic Data Hub:

+

Destination:

{tenant_consent}
@@ -97,11 +97,11 @@ pub fn render_platform_connect(

Connect {platform}

You are logged in with {auth_label}{identity}

-

Target Atomic Data Hub:

+

Destination:

{tenant_consent} - +
"#, @@ -358,7 +358,7 @@ mod tests { &[], ); assert!(html.contains("Google Calendar")); - assert!(html.contains("Use LocalThought to sync Google Calendar with your Atomic Data Hub")); + assert!(html.contains("Use LocalThought to sync Google Calendar with this destination")); assert_eq!(html.matches(r#"action="/connect/authorize""#).count(), 1); assert!(html.contains("name=\"csrf\" value=\"csrf&<"\"")); assert!(!html.contains("