Add a public raw control-plane and worker-plane request API

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

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.

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

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<reqwest::Certificate>,
+    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(_) => "<unparseable base URL>".to_string(),
+    }
+}
+
 fn percent_encode_path_segment(segment: &str) -> String {
     const HEX: &[u8; 16] = b"0123456789ABCDEF";
     let mut encoded = String::with_capacity(segment.len());

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<T>(&mut self, result: &Result<T>) {
+    pub(super) fn finish<T>(&mut self, result: &Result<T>) {
         if self.metrics.0.is_none() {
             return;
         }