diff --git a/CHANGELOG.md b/CHANGELOG.md index 45bfc86..1b62071 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,22 @@ ## Unreleased +- Add a public raw request API for tools that must reach every Server route + and report its response unchanged (#180). `Client::control_plane_request` + and `Client::worker_plane_request` take a `RawRequest` (path segments the + SDK percent-encodes, query pairs, a verbatim JSON body, a timeout and + optional version-header overrides) and return a `RawResponse` for every + HTTP status, with the exact body bytes and all headers. + `RawResponse::error_for_status` gives the same `Error::Http` or + `Error::Protocol` a typed call would, and `refusal()` decodes `reason`, + `rejection_reason`, `message`, `error` and `validation_errors`. Raw + requests make one attempt and never rewrite bodies. Worker-plane 2xx + responses must echo a same-major, equal-or-newer-minor protocol version in + both the header and the body. Typed methods remain the API for + applications. +- Add `ClientBuilder::danger_accept_invalid_certs` for self-hosted Servers + whose certificates cannot be verified. It defaults to `false`; prefer + `additional_root_certificates_pem` for a private CA. - Replay condition-wait histories in which the Server recorded a second open of a wait that was still open (#191, #192). A wait is now linked to the same logical wait by its state when the next open was recorded: if it was diff --git a/Cargo.toml b/Cargo.toml index bcdaa20..a784090 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -108,14 +108,15 @@ path = "src/lib.rs" [features] default = [] -telemetry = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:tracing"] +telemetry = ["dep:opentelemetry", "dep:opentelemetry_sdk"] metrics-prometheus = ["dep:prometheus"] [dependencies] prometheus = { version = "0.14", optional = true, default-features = false } opentelemetry = { version = "0.33", optional = true, default-features = false, features = ["trace", "logs"] } opentelemetry_sdk = { version = "0.33", optional = true, default-features = false, features = ["trace"] } -tracing = { version = "0.1", optional = true, default-features = false, features = ["std"] } +# Always present so security-relevant client settings can warn (#180). +tracing = { version = "0.1", default-features = false, features = ["std"] } apache-avro = { version = "0.21", default-features = false } base64 = "0.22" # Keep apache-avro's dependency resolution compatible with the Rust 1.86 MSRV. diff --git a/README.md b/README.md index a273795..6c7dfe0 100644 --- a/README.md +++ b/README.md @@ -123,6 +123,41 @@ Certificate-chain and hostname verification remain enabled. A private CA does not make a certificate for the wrong hostname valid. Client certificates (mTLS) are not configured by this option. +If a certificate cannot be verified at all, `.danger_accept_invalid_certs(true)` +turns off certificate and hostname verification for every request the client +makes. It defaults to `false`. With it on, anyone on the network path can read +and change traffic, including bearer tokens, so prefer a private CA root. + +### Raw Server requests + +Typed `Client` methods are the API for applications. Tools that must reach +every Server route and report the Server's response unchanged, such as the +`dw` CLI, can send raw requests over the same configured transport: + +```rust,no_run +use durable_workflow::{Client, RawRequest}; +# async fn list(client: &Client) -> Result<(), durable_workflow::Error> { +let response = client + .control_plane_request( + RawRequest::get(["workflows", "order/42", "runs"]).query("page_size", "50"), + ) + .await?; +if let Some(refusal) = response.refusal() { + eprintln!("refused: {:?}", refusal.reason); +} +let body = response.error_for_status()?.json()?; +# let _ = body; +# Ok(()) +# } +``` + +The SDK percent-encodes each path segment, keeps any path prefix in the base +URL, and sends the namespace, the plane's token and both version headers. +Every HTTP status comes back as a `RawResponse` with its exact body bytes and +headers. A raw request makes one attempt and never rewrites its body. +`worker_plane_request` also requires 2xx responses to echo a same-major, +equal-or-newer-minor worker protocol version. + ### Worker concurrency `Worker::max_concurrent_workflow_tasks(...)` and diff --git a/src/lib.rs b/src/lib.rs index c2f1c83..c18c49e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,6 +16,7 @@ pub mod metrics; mod nexus; pub mod pagination; mod query_authority; +mod raw_request; mod runtime_payloads; mod runtime_uploads; mod sticky_worker; @@ -73,6 +74,9 @@ use base64::{engine::general_purpose::STANDARD as BASE64, Engine as _}; use chrono::DateTime; use futures_util::{future::OptionFuture, task::noop_waker_ref}; pub use local_activity::LocalActivityOptions; +pub use raw_request::{ + worker_protocol_compatible, HttpRefusal, RawMethod, RawRequest, RawResponse, +}; use serde::{ de::DeserializeOwned, ser::{SerializeMap, SerializeSeq}, @@ -275,6 +279,8 @@ pub enum Error { InvalidBaseUrl, #[error("invalid additional root certificates: {0}")] InvalidRootCertificates(String), + #[error("invalid raw request: {0}")] + InvalidRawRequest(String), #[error(transparent)] FailureReporting(#[from] FailureReportError), #[error(transparent)] @@ -3063,6 +3069,7 @@ impl Client { max_external_payload_bytes: 64 * 1024 * 1024, worker_protocol_version: WorkerProtocolVersion::default(), additional_root_certificates: Vec::new(), + danger_accept_invalid_certs: false, metrics: metrics::Metrics::default(), } } @@ -5905,6 +5912,7 @@ pub struct ClientBuilder { max_external_payload_bytes: usize, worker_protocol_version: WorkerProtocolVersion, additional_root_certificates: Vec, + danger_accept_invalid_certs: bool, } impl ClientBuilder { @@ -5967,6 +5975,14 @@ impl ClientBuilder { for certificate in self.additional_root_certificates { http = http.add_root_certificate(certificate); } + if self.danger_accept_invalid_certs { + tracing::warn!( + target: "durable_workflow::tls", + endpoint = %tls_warning_endpoint(&base_url), + "TLS certificate and hostname verification is disabled for this client" + ); + http = http.danger_accept_invalid_certs(true); + } Ok(Client { metrics: self.metrics, http: http.build()?, @@ -10298,6 +10314,15 @@ fn default_worker_id() -> String { format!("rust-worker-{}-{millis}", std::process::id()) } +/// Only the endpoint origin, since base paths can contain namespace identities +/// or proxy credentials that must not appear in logs. +fn tls_warning_endpoint(base_url: &str) -> String { + match reqwest::Url::parse(base_url) { + Ok(url) => url.origin().ascii_serialization(), + Err(_) => "".to_string(), + } +} + fn percent_encode_path_segment(segment: &str) -> String { const HEX: &[u8; 16] = b"0123456789ABCDEF"; let mut encoded = String::with_capacity(segment.len()); diff --git a/src/metrics.rs b/src/metrics.rs index 230dc46..4a71e46 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -160,7 +160,7 @@ pub(super) struct ClientObservation { tags: MetricTags, } impl ClientObservation { - fn new( + pub(super) fn new( client: &Client, method: &reqwest::Method, path: &str, @@ -191,7 +191,7 @@ impl ClientObservation { pub(super) fn status(&self, status: reqwest::StatusCode) { self.status.store(status.as_u16(), Ordering::Relaxed); } - fn finish(&mut self, result: &Result) { + pub(super) fn finish(&mut self, result: &Result) { if self.metrics.0.is_none() { return; } diff --git a/src/raw_request.rs b/src/raw_request.rs new file mode 100644 index 0000000..057a621 --- /dev/null +++ b/src/raw_request.rs @@ -0,0 +1,439 @@ +//! Untyped Server requests over the client's configured transport. +//! +//! Typed `Client` methods remain the API for applications. These requests are +//! for tools such as the CLI that must reach every Server route and report the +//! Server's response unchanged. A raw request makes exactly one HTTP attempt +//! and never rewrites its body: no failure-report reduction, no external +//! payload handling, no activity-lease checks and no storage-admission retry. + +use std::time::Duration; + +use serde_json::Value; + +use crate::{ + metrics::ClientObservation, percent_encode_path_segment, protocol_failure, Client, Error, + ProtocolFailure, RequestProtocol, Result, CONTROL_PLANE_VERSION, +}; + +const PROTOCOL_VERSION_HEADER: &str = "X-Durable-Workflow-Protocol-Version"; +const CONTROL_PLANE_VERSION_HEADER: &str = "X-Durable-Workflow-Control-Plane-Version"; + +/// HTTP method of a [`RawRequest`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RawMethod { + Get, + Post, + Put, + Patch, + Delete, +} + +impl RawMethod { + fn as_reqwest(self) -> reqwest::Method { + match self { + Self::Get => reqwest::Method::GET, + Self::Post => reqwest::Method::POST, + Self::Put => reqwest::Method::PUT, + Self::Patch => reqwest::Method::PATCH, + Self::Delete => reqwest::Method::DELETE, + } + } +} + +/// A Server API request addressed by path segments under `{base_url}/api`. +/// +/// Each segment is percent-encoded like PHP `rawurlencode`: RFC 3986 +/// unreserved characters stay, everything else (including `/`) is escaped, so +/// an identifier stays inside its own segment. Empty, `.` and `..` segments +/// are refused before any request is sent. +#[derive(Clone, Debug)] +pub struct RawRequest { + method: RawMethod, + segments: Vec, + query: Vec<(String, String)>, + body: Option, + timeout: Option, + protocol_version: Option, + control_plane_version: Option, +} + +impl RawRequest { + pub fn new(method: RawMethod, segments: I) -> Self + where + I: IntoIterator, + S: Into, + { + Self { + method, + segments: segments.into_iter().map(Into::into).collect(), + query: Vec::new(), + body: None, + timeout: None, + protocol_version: None, + control_plane_version: None, + } + } + + pub fn get, S: Into>(segments: I) -> Self { + Self::new(RawMethod::Get, segments) + } + + pub fn post, S: Into>(segments: I) -> Self { + Self::new(RawMethod::Post, segments) + } + + pub fn put, S: Into>(segments: I) -> Self { + Self::new(RawMethod::Put, segments) + } + + pub fn patch, S: Into>(segments: I) -> Self { + Self::new(RawMethod::Patch, segments) + } + + pub fn delete, S: Into>(segments: I) -> Self { + Self::new(RawMethod::Delete, segments) + } + + /// Append one query pair. Pairs keep their order. + pub fn query(mut self, key: impl Into, value: impl Into) -> Self { + self.query.push((key.into(), value.into())); + self + } + + /// Send this JSON value as the request body, unchanged. + pub fn json(mut self, body: Value) -> Self { + self.body = Some(body); + self + } + + /// Override the client's whole-request timeout for this request. + pub fn timeout(mut self, timeout: Duration) -> Self { + self.timeout = Some(timeout); + self + } + + /// Override the `X-Durable-Workflow-Protocol-Version` header value. + /// Defaults to the client's worker protocol version. + pub fn protocol_version(mut self, version: impl Into) -> Self { + self.protocol_version = Some(version.into()); + self + } + + /// Override the `X-Durable-Workflow-Control-Plane-Version` header value. + /// Defaults to [`CONTROL_PLANE_VERSION`]. + pub fn control_plane_version(mut self, version: impl Into) -> Self { + self.control_plane_version = Some(version.into()); + self + } + + pub fn method(&self) -> RawMethod { + self.method + } + + /// The encoded path relative to `/api`, for example `/workflows/a%2Fb`. + pub fn path(&self) -> Result { + if self.segments.is_empty() { + return Err(invalid_request("at least one path segment is required")); + } + let mut path = String::new(); + for segment in &self.segments { + if segment.is_empty() || segment == "." || segment == ".." { + return Err(invalid_request( + "path segments must not be empty, \".\" or \"..\"", + )); + } + path.push('/'); + path.push_str(&percent_encode_path_segment(segment)); + } + Ok(path) + } +} + +fn invalid_request(reason: &str) -> Error { + Error::InvalidRawRequest(reason.to_string()) +} + +/// A Server response, unchanged, for any HTTP status. +#[derive(Clone, Debug)] +pub struct RawResponse { + status: u16, + headers: reqwest::header::HeaderMap, + body: Vec, +} + +impl RawResponse { + pub fn status(&self) -> u16 { + self.status + } + + pub fn is_success(&self) -> bool { + (200..300).contains(&self.status) + } + + /// A response header value, when present and valid visible ASCII. + pub fn header(&self, name: &str) -> Option<&str> { + self.headers.get(name)?.to_str().ok() + } + + /// Every response header as received. Names are lowercase. + pub fn headers(&self) -> impl Iterator { + self.headers + .iter() + .map(|(name, value)| (name.as_str(), value.as_bytes())) + } + + /// The worker protocol version the Server echoed, if any. + pub fn protocol_version(&self) -> Option<&str> { + self.header(PROTOCOL_VERSION_HEADER) + } + + /// The control-plane version the Server echoed, if any. + pub fn control_plane_version(&self) -> Option<&str> { + self.header(CONTROL_PLANE_VERSION_HEADER) + } + + /// The exact response body bytes. + pub fn body_bytes(&self) -> &[u8] { + &self.body + } + + /// The body as JSON. An empty body is `null`. + pub fn json(&self) -> Result { + if self.body.is_empty() { + return Ok(Value::Null); + } + Ok(serde_json::from_slice(&self.body)?) + } + + /// The decoded refusal body of a non-2xx response whose body is a JSON + /// object. `None` for 2xx responses and other bodies. + pub fn refusal(&self) -> Option { + if self.is_success() { + return None; + } + HttpRefusal::decode(self.status, serde_json::from_slice(&self.body).ok()?) + } + + /// Return the response for 2xx. Otherwise return the error a typed call + /// would: [`Error::Protocol`] for a protocol-version refusal, else + /// [`Error::Http`] with the exact body text. + pub fn error_for_status(self) -> Result { + if self.is_success() { + return Ok(self); + } + let status = reqwest::StatusCode::from_u16(self.status) + .map_err(|_| invalid_request("response status is not a valid HTTP status"))?; + let body = String::from_utf8_lossy(&self.body).into_owned(); + if let Some(failure) = protocol_failure(status, &body) { + return Err(Error::Protocol(failure)); + } + Err(Error::Http { status, body }) + } +} + +/// The decoded body of a refused (non-2xx) Server request. +#[derive(Clone, Debug, PartialEq)] +#[non_exhaustive] +pub struct HttpRefusal { + pub status: u16, + pub reason: Option, + pub rejection_reason: Option, + pub message: Option, + pub error: Option, + pub validation_errors: Option, + /// The full JSON body, unchanged. + pub body: Value, +} + +impl HttpRefusal { + fn decode(status: u16, body: Value) -> Option { + let object = body.as_object()?; + let text = |key: &str| object.get(key).and_then(Value::as_str).map(str::to_owned); + Some(Self { + status, + reason: text("reason"), + rejection_reason: text("rejection_reason"), + message: text("message"), + error: text("error"), + validation_errors: object + .get("validation_errors") + .or_else(|| object.get("errors")) + .cloned(), + body, + }) + } +} + +impl Error { + /// Decode a Server refusal from [`Error::Http`] or [`Error::Protocol`]. + /// + /// Returns `None` for other errors and for HTTP errors whose body is not + /// a JSON object. + pub fn refusal(&self) -> Option { + match self { + Self::Http { status, body } => { + HttpRefusal::decode(status.as_u16(), serde_json::from_str(body).ok()?) + } + Self::Protocol(failure) => HttpRefusal::decode(failure.status, failure.body.clone()), + _ => None, + } + } +} + +/// Whether a Server worker protocol version is compatible with the client's: +/// same major, and a Server minor at least the client's. Versions that are +/// not `major.minor` integers must match exactly. This is the CLI 2.x rule. +pub fn worker_protocol_compatible(client: &str, server: &str) -> bool { + fn split(version: &str) -> Option<(u64, u64)> { + let (major, minor) = version.split_once('.')?; + let digits = |part: &str| !part.is_empty() && part.bytes().all(|b| b.is_ascii_digit()); + if !digits(major) || !digits(minor) { + return None; + } + Some((major.parse().ok()?, minor.parse().ok()?)) + } + match (split(client), split(server)) { + (Some(client), Some(server)) => client.0 == server.0 && client.1 <= server.1, + _ => client == server, + } +} + +fn invalid_worker_response(status: u16, message: String, requested: &str, body: Value) -> Error { + Error::Protocol(ProtocolFailure { + status, + reason: "invalid_worker_protocol_response".to_string(), + message, + supported_version: None, + requested_version: Some(requested.to_string()), + body, + }) +} + +impl Client { + /// Send a control-plane request with this client's base URL (including + /// any path prefix), TLS settings, namespace, control token, timeout and + /// metrics recorder. + /// + /// Every HTTP response is returned as a [`RawResponse`], including 4xx + /// and 5xx. Use [`RawResponse::error_for_status`] or + /// [`RawResponse::refusal`] to classify a refusal. Transport failures, + /// including timeouts, are [`Error::Transport`]. + pub async fn control_plane_request(&self, request: RawRequest) -> Result { + self.raw_request(request, RequestProtocol::ControlPlane) + .await + } + + /// Send a worker-plane request with the worker token. Otherwise like + /// [`Client::control_plane_request`]. + /// + /// A 2xx response must echo a compatible protocol version (see + /// [`worker_protocol_compatible`]) in its header, and its JSON body must + /// carry the same `protocol_version`; otherwise the result is + /// [`Error::Protocol`] with reason `invalid_worker_protocol_response`. + /// A numeric body value is compared in its JSON spelling (`1.2` stays + /// `"1.2"`, `1.20` becomes `"1.2"`). Servers send it as a string. + /// Non-2xx responses are not checked. + pub async fn worker_plane_request(&self, request: RawRequest) -> Result { + self.raw_request(request, RequestProtocol::DefaultWorker) + .await + } + + async fn raw_request( + &self, + request: RawRequest, + protocol: RequestProtocol, + ) -> Result { + let path = request.path()?; + let method = request.method.as_reqwest(); + let mut observation = ClientObservation::new(self, &method, &path, protocol); + let result = self + .send_raw_request(request, method, &path, protocol, &observation) + .await; + observation.finish(&result); + result + } + + async fn send_raw_request( + &self, + request: RawRequest, + method: reqwest::Method, + path: &str, + protocol: RequestProtocol, + observation: &ClientObservation, + ) -> Result { + // Both version headers go on every request, on both planes, as the + // CLI 2.x client sends them. + // The configured version, with no negotiation or worker-loop checks. + let worker_version = request + .protocol_version + .clone() + .unwrap_or_else(|| self.worker_protocol_version.as_str().to_string()); + let control_version = request + .control_plane_version + .as_deref() + .unwrap_or(CONTROL_PLANE_VERSION); + let auth_token = self.auth_token(protocol)?; + let mut builder = self + .http + .request(method, format!("{}/api{}", self.base_url, path)) + .header(reqwest::header::ACCEPT, "application/json") + .header("X-Namespace", &self.namespace) + .header(PROTOCOL_VERSION_HEADER, &worker_version) + .header(CONTROL_PLANE_VERSION_HEADER, control_version); + if let Some(timeout) = request.timeout { + builder = builder.timeout(timeout); + } + if let Some(token) = auth_token { + builder = builder.bearer_auth(token); + } + if !request.query.is_empty() { + builder = builder.query(&request.query); + } + if let Some(body) = &request.body { + builder = builder.json(body); + } + + // Timeouts stay Error::Transport, as for typed calls. + let response = builder.send().await?; + let status = response.status(); + observation.status(status); + let response = RawResponse { + status: status.as_u16(), + headers: response.headers().clone(), + body: response.bytes().await?.to_vec(), + }; + + if matches!(protocol, RequestProtocol::ControlPlane) || !response.is_success() { + return Ok(response); + } + let body = response.json().unwrap_or(Value::Null); + let echoed = response.protocol_version().map(str::trim); + if !echoed.is_some_and(|echoed| worker_protocol_compatible(&worker_version, echoed)) { + return Err(invalid_worker_response( + response.status, + format!( + "worker response version {} is not same-major compatible with {worker_version}", + echoed.unwrap_or("missing") + ), + &worker_version, + body, + )); + } + let body_version = body + .get("protocol_version") + .and_then(|version| match version { + Value::String(text) => Some(text.trim().to_string()), + Value::Number(number) => Some(number.to_string()), + _ => None, + }); + if body_version.as_deref() != echoed { + return Err(invalid_worker_response( + response.status, + "worker response body protocol_version must equal the protocol header".into(), + &worker_version, + body, + )); + } + Ok(response) + } +} diff --git a/src/tls.rs b/src/tls.rs index 95d8b5c..801e2e7 100644 --- a/src/tls.rs +++ b/src/tls.rs @@ -18,4 +18,16 @@ impl ClientBuilder { self.additional_root_certificates.extend(certificates); Ok(self) } + + /// Disable TLS certificate and hostname verification for every request + /// this client makes, including worker polling and external payload + /// transfers. The default is `false` (verified). + /// + /// Any party on the network path can then read and change traffic, + /// including bearer tokens. Prefer + /// [`ClientBuilder::additional_root_certificates_pem`] for a private CA. + pub fn danger_accept_invalid_certs(mut self, accept: bool) -> Self { + self.danger_accept_invalid_certs = accept; + self + } } diff --git a/tests/private_ca.rs b/tests/private_ca.rs index 23f76f9..9234266 100644 --- a/tests/private_ca.rs +++ b/tests/private_ca.rs @@ -208,3 +208,116 @@ async fn adding_a_private_ca_keeps_bundled_public_roots_trusted() { Ok(_) | Err(Error::Http { .. }) | Err(Error::Json(_)) )); } + +#[tokio::test] +async fn certificate_verification_is_on_by_default_and_only_an_explicit_opt_out_disables_it() { + let server = TestServer::start().await; + for client in [ + Client::builder(server.url("127.0.0.1")).build().unwrap(), + Client::builder(server.url("127.0.0.1")) + .danger_accept_invalid_certs(false) + .build() + .unwrap(), + ] { + assert!(matches!(client.health().await, Err(Error::Transport(_)))); + } + assert!( + server.requests.lock().unwrap().is_empty(), + "verified TLS must fail before HTTP" + ); + + // Untrusted issuer and a hostname the certificate does not name. + let insecure = Client::builder(server.url("127.0.0.1")) + .danger_accept_invalid_certs(true) + .build() + .unwrap(); + assert_eq!(insecure.health().await.unwrap()["status"], "ok"); + assert_eq!(*server.requests.lock().unwrap(), ["/api/health"]); +} + +/// Records warn-level events from the SDK's TLS target. +#[derive(Default)] +struct WarningLog(Mutex>); + +struct FieldText<'a>(&'a mut String); + +impl tracing::field::Visit for FieldText<'_> { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0.push_str(&format!("{}={:?};", field.name(), value)); + } +} + +impl tracing::Subscriber for WarningLog { + fn enabled(&self, _: &tracing::Metadata<'_>) -> bool { + true + } + fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id { + tracing::span::Id::from_u64(1) + } + fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {} + fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {} + fn event(&self, event: &tracing::Event<'_>) { + let metadata = event.metadata(); + if *metadata.level() == tracing::Level::WARN && metadata.target() == "durable_workflow::tls" + { + let mut text = String::new(); + event.record(&mut FieldText(&mut text)); + self.0.lock().unwrap().push(text); + } + } + fn enter(&self, _: &tracing::span::Id) {} + fn exit(&self, _: &tracing::span::Id) {} +} + +#[test] +fn disabling_verification_warns_once_with_only_the_endpoint_origin() { + for (base_url, origin) in [ + ( + "https://operator:hunter2@workflow.internal:8443/api/runtime/v1/namespaces/customer-private/proxy-secret?query-secret=query-value#fragment-secret", + "https://workflow.internal:8443", + ), + ( + "https://operator:hunter2@[2001:db8::1]:9443/customer-private/proxy-secret?query-secret=query-value#fragment-secret", + "https://[2001:db8::1]:9443", + ), + ] { + let log = Arc::new(WarningLog::default()); + tracing::subscriber::with_default(log.clone(), || { + Client::builder(base_url) + .danger_accept_invalid_certs(true) + .build() + .unwrap(); + Client::builder("https://verified.internal") + .build() + .unwrap(); + Client::builder("https://verified.internal") + .danger_accept_invalid_certs(false) + .build() + .unwrap(); + }); + let warnings = log.0.lock().unwrap().clone(); + assert_eq!(warnings.len(), 1, "{warnings:?}"); + assert!( + warnings[0].contains(&format!("endpoint={origin};")), + "{}", + warnings[0] + ); + assert!( + warnings[0].contains("verification is disabled"), + "{}", + warnings[0] + ); + for sensitive in [ + "operator", + "hunter2", + "/api/runtime", + "customer-private", + "proxy-secret", + "query-secret", + "query-value", + "fragment-secret", + ] { + assert!(!warnings[0].contains(sensitive), "{}", warnings[0]); + } + } +} diff --git a/tests/raw_request.rs b/tests/raw_request.rs new file mode 100644 index 0000000..2e201f5 --- /dev/null +++ b/tests/raw_request.rs @@ -0,0 +1,497 @@ +use durable_workflow::{ + json, worker_protocol_compatible, Client, Error, RawMethod, RawRequest, Value, + CONTROL_PLANE_VERSION, WORKER_PROTOCOL_VERSION, +}; +use std::sync::{Arc, Mutex}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, +}; + +/// One captured HTTP request: request line, lowercase headers and body. +#[derive(Clone, Debug)] +struct Captured { + line: String, + headers: Vec<(String, String)>, + body: Vec, +} + +impl Captured { + fn header(&self, name: &str) -> Option<&str> { + self.headers + .iter() + .find(|(key, _)| key == name) + .map(|(_, value)| value.as_str()) + } + fn target(&self) -> &str { + self.line.split_whitespace().nth(1).unwrap_or_default() + } +} + +struct Reply { + status: &'static str, + headers: Vec<(&'static str, String)>, + body: String, +} + +fn reply(status: &'static str, headers: &[(&'static str, &str)], body: &str) -> Reply { + Reply { + status, + headers: headers.iter().map(|(k, v)| (*k, v.to_string())).collect(), + body: body.to_string(), + } +} + +struct Server { + url: String, + requests: Arc>>, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for Server { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// Serve the scripted replies in order, one per connection. +async fn serve(replies: Vec) -> Server { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let requests = Arc::new(Mutex::new(Vec::new())); + let observed = requests.clone(); + let task = tokio::spawn(async move { + for reply in replies { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut raw = Vec::new(); + let header_end = loop { + let mut buffer = [0; 4096]; + let count = stream.read(&mut buffer).await.unwrap(); + assert!(count > 0, "client closed before sending headers"); + raw.extend_from_slice(&buffer[..count]); + if let Some(end) = raw.windows(4).position(|w| w == b"\r\n\r\n") { + break end + 4; + } + }; + let head = String::from_utf8_lossy(&raw[..header_end]).to_string(); + let mut lines = head.lines(); + let line = lines.next().unwrap().to_string(); + let headers: Vec<(String, String)> = lines + .filter_map(|l| l.split_once(':')) + .map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string())) + .collect(); + let length = headers + .iter() + .find(|(k, _)| k == "content-length") + .map(|(_, v)| v.parse::().unwrap()) + .unwrap_or(0); + while raw.len() < header_end + length { + let mut buffer = [0; 4096]; + let count = stream.read(&mut buffer).await.unwrap(); + assert!(count > 0, "client closed before sending its body"); + raw.extend_from_slice(&buffer[..count]); + } + observed.lock().unwrap().push(Captured { + line, + headers, + body: raw[header_end..header_end + length].to_vec(), + }); + let mut response = format!( + "HTTP/1.1 {}\r\ncontent-length: {}\r\nconnection: close\r\n", + reply.status, + reply.body.len() + ); + for (name, value) in &reply.headers { + response.push_str(&format!("{name}: {value}\r\n")); + } + response.push_str("\r\n"); + response.push_str(&reply.body); + stream.write_all(response.as_bytes()).await.unwrap(); + stream.shutdown().await.unwrap(); + } + }); + Server { + url, + requests, + task, + } +} + +fn client(url: &str) -> Client { + Client::builder(url) + .control_token(Some("control-secret".into())) + .worker_token(Some("worker-secret".into())) + .namespace("orders") + .build() + .unwrap() +} + +#[test] +fn path_segments_are_rawurlencoded_and_traversal_is_refused() { + let request = RawRequest::get(["workflows", "a/b c%é?#", "runs", "run-1.2_~"]); + assert_eq!(request.method(), RawMethod::Get); + assert_eq!( + request.path().unwrap(), + "/workflows/a%2Fb%20c%25%C3%A9%3F%23/runs/run-1.2_~" + ); + for segments in [vec![], vec![""], vec!["workflows", "."], vec!["..", "x"]] { + assert!(matches!( + RawRequest::get(segments).path(), + Err(Error::InvalidRawRequest(_)) + )); + } +} + +#[tokio::test] +async fn invalid_paths_are_refused_before_any_connection() { + let server = serve(Vec::new()).await; + let error = client(&server.url) + .control_plane_request(RawRequest::delete(["workflows", ".."])) + .await + .unwrap_err(); + assert!(matches!(error, Error::InvalidRawRequest(_))); + assert!(server.requests.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn control_plane_request_sends_identity_query_and_body_and_returns_bytes_unchanged() { + // Unknown fields, key order, whitespace and float digits must survive. + let body = "{\"zeta\":1, \"unknown\":{\"nested\":[true,null]},\"score\":1.4820142857142857}"; + let server = serve(vec![reply( + "200 OK", + &[ + ("content-type", "application/json"), + ("x-durable-workflow-control-plane-version", "2"), + ], + body, + )]) + .await; + // A managed-namespace base URL keeps its path prefix. + let base = format!("{}/api/runtime/v1/namespaces/ns-1", server.url); + let response = client(&base) + .control_plane_request( + RawRequest::post(["workflows", "order/42", "signal", "approve"]) + .query("b", "2") + .query("a", "x y&z") + .json(json!({"input": [1, "two"], "extra": {"kept": true}})), + ) + .await + .unwrap(); + + assert_eq!(response.status(), 200); + assert!(response.is_success()); + assert_eq!(response.body_bytes(), body.as_bytes()); + assert_eq!(response.control_plane_version(), Some("2")); + assert_eq!(response.header("content-type"), Some("application/json")); + assert!(response + .headers() + .any(|(name, value)| name == "content-type" && value == b"application/json")); + let decoded = response.json().unwrap(); + assert_eq!(decoded["unknown"], json!({"nested": [true, null]})); + assert_eq!(decoded["score"].as_f64(), Some(1.482_014_285_714_285_7)); + assert!(response.refusal().is_none()); + + let captured = server.requests.lock().unwrap()[0].clone(); + assert_eq!( + captured.line, + "POST /api/runtime/v1/namespaces/ns-1/api/workflows/order%2F42/signal/approve?b=2&a=x+y%26z HTTP/1.1" + ); + assert_eq!( + captured.header("authorization"), + Some("Bearer control-secret") + ); + assert_eq!(captured.header("x-namespace"), Some("orders")); + assert_eq!(captured.header("accept"), Some("application/json")); + assert_eq!(captured.header("content-type"), Some("application/json")); + assert_eq!( + captured.header("x-durable-workflow-control-plane-version"), + Some(CONTROL_PLANE_VERSION) + ); + assert_eq!( + captured.header("x-durable-workflow-protocol-version"), + Some(WORKER_PROTOCOL_VERSION) + ); + let sent: Value = serde_json::from_slice(&captured.body).unwrap(); + assert_eq!(sent, json!({"input": [1, "two"], "extra": {"kept": true}})); +} + +#[tokio::test] +async fn per_request_version_overrides_replace_both_headers() { + let server = serve(vec![reply("204 No Content", &[], "")]).await; + let response = client(&server.url) + .control_plane_request( + RawRequest::get(["namespaces"]) + .protocol_version("1.17") + .control_plane_version("3"), + ) + .await + .unwrap(); + assert_eq!(response.status(), 204); + assert_eq!(response.json().unwrap(), Value::Null); + let captured = server.requests.lock().unwrap()[0].clone(); + assert_eq!(captured.target(), "/api/namespaces"); + assert!(captured.body.is_empty()); + assert_eq!( + captured.header("x-durable-workflow-protocol-version"), + Some("1.17") + ); + assert_eq!( + captured.header("x-durable-workflow-control-plane-version"), + Some("3") + ); +} + +#[tokio::test] +async fn refusals_keep_headers_and_decode_every_message_field() { + let refused = json!({ + "reason": "historical_run_command_rejected", + "rejection_reason": "run is not current", + "message": "Command rejected.", + "error": "conflict", + "validation_errors": {"run_id": ["must be current"]}, + "unknown": 7 + }) + .to_string(); + let protocol = json!({ + "reason": "unsupported_control_plane_version", + "message": "use 2", + "supported_version": "2" + }) + .to_string(); + let server = serve(vec![ + reply( + "409 Conflict", + &[("x-durable-workflow-control-plane-version", "2")], + &refused, + ), + reply("400 Bad Request", &[], &protocol), + reply("502 Bad Gateway", &[], "upstream"), + ]) + .await; + let client = client(&server.url); + + let response = client + .control_plane_request(RawRequest::post(["workflows", "w", "cancel"])) + .await + .unwrap(); + assert_eq!(response.status(), 409); + assert!(!response.is_success()); + assert_eq!(response.control_plane_version(), Some("2")); + let refusal = response.refusal().unwrap(); + assert_eq!(refusal.status, 409); + assert_eq!( + refusal.reason.as_deref(), + Some("historical_run_command_rejected") + ); + assert_eq!( + refusal.rejection_reason.as_deref(), + Some("run is not current") + ); + assert_eq!(refusal.message.as_deref(), Some("Command rejected.")); + assert_eq!(refusal.error.as_deref(), Some("conflict")); + assert_eq!( + refusal.validation_errors, + Some(json!({"run_id": ["must be current"]})) + ); + assert_eq!(refusal.body["unknown"], 7); + let error = response.error_for_status().unwrap_err(); + match &error { + Error::Http { status, body } => { + assert_eq!(status.as_u16(), 409); + assert_eq!(body, &refused); + } + other => panic!("expected Error::Http, got {other:?}"), + } + assert_eq!( + error.refusal().unwrap().reason.as_deref(), + Some("historical_run_command_rejected") + ); + + let response = client + .control_plane_request(RawRequest::get(["workflows"])) + .await + .unwrap(); + let error = response.error_for_status().unwrap_err(); + let Error::Protocol(failure) = &error else { + panic!("protocol refusal must stay Error::Protocol, got {error:?}"); + }; + assert_eq!(failure.reason, "unsupported_control_plane_version"); + assert_eq!( + error.refusal().unwrap().reason.as_deref(), + Some("unsupported_control_plane_version") + ); + + let response = client + .control_plane_request(RawRequest::get(["health"])) + .await + .unwrap(); + assert_eq!(response.body_bytes(), b"upstream"); + assert!(response.refusal().is_none()); + let error = response.error_for_status().unwrap_err(); + assert!(matches!(&error, Error::Http { body, .. } if body == "upstream")); + assert!(error.refusal().is_none()); +} + +#[tokio::test] +async fn worker_plane_accepts_compatible_versions_and_refuses_invalid_echoes() { + let ok = |version: &str| { + reply( + "200 OK", + &[("x-durable-workflow-protocol-version", version)], + &json!({"task": null, "protocol_version": version}).to_string(), + ) + }; + let server = serve(vec![ + ok("1.20"), + ok("1.21"), + ok("2.0"), + ok("1.19"), + reply( + "200 OK", + &[], + &json!({"protocol_version": "1.20"}).to_string(), + ), + reply( + "200 OK", + &[("x-durable-workflow-protocol-version", "1.20")], + &json!({"task": null}).to_string(), + ), + reply( + "200 OK", + &[("x-durable-workflow-protocol-version", "1.21")], + &json!({"protocol_version": "1.20"}).to_string(), + ), + reply( + "409 Conflict", + &[], + &json!({"reason": "task_not_leased"}).to_string(), + ), + ]) + .await; + let client = client(&server.url); + let poll = || RawRequest::post(["worker", "workflow-tasks", "poll"]).json(json!({})); + + for _ in 0..2 { + let response = client.worker_plane_request(poll()).await.unwrap(); + assert!(response.is_success()); + } + for expected in [ + "not same-major compatible", + "not same-major compatible", + "is not same-major compatible with 1.20", + "must equal the protocol header", + "must equal the protocol header", + ] { + let error = client.worker_plane_request(poll()).await.unwrap_err(); + let Error::Protocol(failure) = &error else { + panic!("expected Error::Protocol, got {error:?}"); + }; + assert_eq!(failure.reason, "invalid_worker_protocol_response"); + assert_eq!(failure.requested_version.as_deref(), Some("1.20")); + assert!(failure.message.contains(expected), "{}", failure.message); + } + // Non-2xx responses are returned unchecked. + let response = client.worker_plane_request(poll()).await.unwrap(); + assert_eq!(response.status(), 409); + assert_eq!( + response.refusal().unwrap().reason.as_deref(), + Some("task_not_leased") + ); + + let captured = server.requests.lock().unwrap().clone(); + assert_eq!(captured.len(), 8); + for request in &captured { + assert_eq!(request.target(), "/api/worker/workflow-tasks/poll"); + assert_eq!( + request.header("authorization"), + Some("Bearer worker-secret") + ); + assert_eq!( + request.header("x-durable-workflow-protocol-version"), + Some(WORKER_PROTOCOL_VERSION) + ); + assert_eq!( + request.header("x-durable-workflow-control-plane-version"), + Some(CONTROL_PLANE_VERSION) + ); + } +} + +#[test] +fn worker_protocol_compatibility_is_same_major_with_a_newer_or_equal_minor() { + for (client, server, compatible) in [ + ("1.20", "1.20", true), + ("1.20", "1.21", true), + ("1.20", "1.100", true), + ("1.20", "1.19", false), + ("1.20", "2.20", false), + ("1.20", "1.2x", false), + ("1.20", "", false), + ("v1", "v1", true), + ("1.20", " 1.20", false), + ] { + assert_eq!( + worker_protocol_compatible(client, server), + compatible, + "{client} vs {server}" + ); + } +} + +#[tokio::test] +async fn raw_requests_use_role_tokens_and_refuse_a_missing_role_credential() { + let server = serve(Vec::new()).await; + let worker_only = Client::builder(&server.url) + .worker_token(Some("worker-secret".into())) + .build() + .unwrap(); + assert!(matches!( + worker_only + .control_plane_request(RawRequest::get(["namespaces"])) + .await, + Err(Error::MissingRoleCredentials { + role: "control", + .. + }) + )); + assert!(server.requests.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn url_shaped_segments_stay_on_the_configured_origin_with_client_identity() { + let server = serve(vec![ + reply("200 OK", &[], "{}"), + reply("200 OK", &[], "{}"), + reply("200 OK", &[], "{}"), + ]) + .await; + let client = client(&server.url); + for segments in [ + vec!["https://evil.example/steal"], + vec!["//evil.example", "x"], + vec!["workflows", "@evil.example"], + ] { + let response = client + .control_plane_request(RawRequest::get(segments)) + .await + .unwrap(); + assert_eq!(response.status(), 200); + } + let captured = server.requests.lock().unwrap().clone(); + let targets: Vec<&str> = captured.iter().map(Captured::target).collect(); + assert_eq!( + targets, + [ + "/api/https%3A%2F%2Fevil.example%2Fsteal", + "/api/%2F%2Fevil.example/x", + "/api/workflows/%40evil.example", + ] + ); + for request in &captured { + assert_eq!( + request.header("authorization"), + Some("Bearer control-secret") + ); + assert_eq!(request.header("x-namespace"), Some("orders")); + } +}