cratefield_module_privacy/
provider.rs

1//! External subject-data providers (issue #653): the data a venture holds
2//! **outside** the harness — a warehouse, a CRM, another product's database —
3//! reached over one signed HTTP contract.
4//!
5//! A provider answers three POSTs, each carrying the subject and a shared
6//! request id and each signed with HMAC-SHA256 under a secret read from the
7//! config port at call time:
8//!
9//! - `{url}/export` -> `{"sections":[{"name": str, "description"?: str, "data": any}]}`
10//! - `{url}/erase/plan` -> `{"sections":[{"name": str, "action": "delete"|"anonymise"|"retain", "reason"?: str}]}`
11//! - `{url}/erase/apply` -> any 2xx means applied
12//!
13//! Nothing a provider sends back reaches a caller: a failure is one of four
14//! words ([`ProviderError`]), because an upstream's status text or body may
15//! quote the subject's data, and an erasure receipt is not the place for it.
16
17use cratefield_core::{Config, HttpClient, HttpError, HttpPolicy};
18use hmac::{Hmac, KeyInit, Mac};
19use http::header::CONTENT_TYPE;
20use serde_json::{Value, json};
21use sha2::{Digest, Sha256};
22use std::sync::Arc;
23use std::time::Duration;
24
25use crate::handlers::unix_now;
26
27/// The header every provider call carries, under the in-repo signing idiom
28/// (`module-webhooks` signs deliveries the same way, so one receiver-side
29/// verifier covers both).
30pub const SIGNATURE_HEADER: &str = "Cratefield-Signature";
31
32/// The deadline for one provider call when the builder sets none; clamped to
33/// the port ceiling at send time.
34const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
35
36/// The response cap for one provider call when the builder sets none: a
37/// section list about one subject, not a file download.
38const DEFAULT_MAX_RESPONSE_BYTES: usize = 1024 * 1024;
39
40/// Why a provider call did not succeed, in the four words a caller renders.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub(crate) enum ProviderError {
43    /// Transport failure, timeout, or a 5xx.
44    Unavailable,
45    /// A 4xx: the provider refused the call.
46    Rejected,
47    /// An oversize or malformed body, or a schema mismatch.
48    InvalidResponse,
49    /// No signing secret is configured.
50    NotConfigured,
51}
52
53impl ProviderError {
54    pub(crate) const fn as_str(self) -> &'static str {
55        match self {
56            Self::Unavailable => "unavailable",
57            Self::Rejected => "rejected",
58            Self::InvalidResponse => "invalid_response",
59            Self::NotConfigured => "not_configured",
60        }
61    }
62}
63
64/// One external system holding data about a subject; built with
65/// [`HttpProvider::new`] and registered with [`Privacy::provider`](crate::Privacy::provider).
66#[derive(Debug, Clone)]
67pub struct HttpProvider {
68    name: String,
69    url: String,
70    secret_env: Option<String>,
71    timeout: Duration,
72    max_response_bytes: usize,
73    /// Whether this provider holds the **identity** — the account row itself,
74    /// the record everything else is keyed on. See [`HttpProvider::account`].
75    account: bool,
76}
77
78impl HttpProvider {
79    /// A provider named `name`, reached at `url`, whose `/export`,
80    /// `/erase/plan` and `/erase/apply` paths are appended to `url`.
81    #[must_use]
82    pub fn new(name: impl Into<String>, url: impl Into<String>) -> Self {
83        Self {
84            name: name.into(),
85            url: url.into(),
86            secret_env: None,
87            timeout: DEFAULT_TIMEOUT,
88            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
89            account: false,
90        }
91    }
92
93    /// The config/env variable holding this provider's HMAC secret, read
94    /// through the config port at call time. None, missing or empty reports
95    /// `not_configured` — a call this module cannot sign is not one it sends.
96    #[must_use]
97    pub fn secret_env(mut self, name: impl Into<String>) -> Self {
98        self.secret_env = Some(name.into());
99        self
100    }
101
102    /// The deadline for one call, clamped up to the port ceiling at send time.
103    #[must_use]
104    pub fn timeout(mut self, timeout: Duration) -> Self {
105        self.timeout = timeout;
106        self
107    }
108
109    /// The most bytes of a response this module will read, clamped to the
110    /// port ceiling. A larger response is `invalid_response`.
111    #[must_use]
112    pub fn max_response_bytes(mut self, max_response_bytes: usize) -> Self {
113        self.max_response_bytes = max_response_bytes;
114        self
115    }
116
117    /// Marks this provider as the one holding the **identity itself** — the
118    /// accounts table, the CRM's contact record, the identity provider the
119    /// others key on.
120    ///
121    /// **Why it must be erased last.** Erasure identifies its subject by a
122    /// value (`acct-1`). Every provider after the account provider is reached
123    /// *through* that value, and the provider that holds the identity can
124    /// delete it. Called in registration order, a CRM holding only
125    /// `account_id` might be asked to erase after the accounts table is
126    /// already gone — and a row keyed on a subject the calling deployment no
127    /// longer holds an identifier for cannot be found, let alone erased. The
128    /// row survives and the erasure reports itself complete.
129    ///
130    /// So marked providers sort after every unmarked one, in all three loops
131    /// (`in_erasure_order`): export, plan and apply. Within each group the
132    /// registration order is kept, so the answer is stable and a reader can
133    /// see the sequence a build produced rather than a sort's opinion of it.
134    ///
135    /// A build is expected to mark **at most one** provider this way — the
136    /// account is one record however many systems mirror it — but nothing
137    /// breaks if it marks several: they go last, in the order they were
138    /// registered.
139    #[must_use]
140    pub fn account(mut self) -> Self {
141        self.account = true;
142        self
143    }
144
145    /// The provider's name, as a caller's response reports it.
146    #[must_use]
147    pub fn name(&self) -> &str {
148        &self.name
149    }
150
151    /// The signing secret, read from `config` now.
152    pub(crate) fn secret(&self, config: &dyn Config) -> Result<String, ProviderError> {
153        let name = self
154            .secret_env
155            .as_deref()
156            .ok_or(ProviderError::NotConfigured)?;
157        config
158            .get(name)
159            .filter(|secret| !secret.trim().is_empty())
160            .ok_or(ProviderError::NotConfigured)
161    }
162
163    /// `POST {url}/export`, signed.
164    pub(crate) async fn export(
165        &self,
166        http: &Arc<dyn HttpClient>,
167        config: &dyn Config,
168        subject: &str,
169        request_id: &str,
170    ) -> Result<Vec<Value>, ProviderError> {
171        let secret = self.secret(config)?;
172        let body = request_body(subject, request_id);
173        let response = self.post(http, "export", &secret, &body).await?;
174        provider_sections(response.body())
175    }
176
177    /// `POST {url}/erase/plan`, signed.
178    pub(crate) async fn erase_plan(
179        &self,
180        http: &Arc<dyn HttpClient>,
181        config: &dyn Config,
182        subject: &str,
183        request_id: &str,
184    ) -> Result<Vec<Value>, ProviderError> {
185        let secret = self.secret(config)?;
186        let body = request_body(subject, request_id);
187        let response = self.post(http, "erase/plan", &secret, &body).await?;
188        plan_sections(response.body())
189    }
190
191    /// `POST {url}/erase/apply`, signed, with a secret already resolved (the
192    /// retry path reuses this so it re-signs with a fresh timestamp).
193    pub(crate) async fn apply_signed(
194        &self,
195        http: &Arc<dyn HttpClient>,
196        secret: &str,
197        body: &bytes::Bytes,
198    ) -> Result<(), ProviderError> {
199        self.post(http, "erase/apply", secret, body).await?;
200        Ok(())
201    }
202
203    async fn post(
204        &self,
205        http: &Arc<dyn HttpClient>,
206        path: &str,
207        secret: &str,
208        body: &bytes::Bytes,
209    ) -> Result<http::Response<bytes::Bytes>, ProviderError> {
210        let policy = HttpPolicy {
211            max_response_bytes: self.max_response_bytes,
212            timeout: self.timeout,
213        }
214        .clamped();
215        let url = format!("{}/{path}", self.url.trim_end_matches('/'));
216        let mut request = http::Request::builder()
217            .method(http::Method::POST)
218            .uri(&url)
219            .header(CONTENT_TYPE, "application/json")
220            .header(SIGNATURE_HEADER, signature_header(secret, unix_now(), body))
221            .body(body.clone())
222            .map_err(|_| ProviderError::Unavailable)?;
223        request.extensions_mut().insert(policy);
224        let response = http.send(request).await.map_err(|err| match err {
225            // A body over the cap is a malformed answer, not an outage.
226            HttpError::ResponseTooLarge { .. } => ProviderError::InvalidResponse,
227            _ => ProviderError::Unavailable,
228        })?;
229        check_status(&response)?;
230        // The runtime wraps `ports.http` in `BoundedHttpClient`, but this
231        // module does not rely on the wrapper being there: an oversize body
232        // is refused here too, before anything parses it.
233        if response.body().len() > policy.max_response_bytes {
234            return Err(ProviderError::InvalidResponse);
235        }
236        Ok(response)
237    }
238}
239
240/// The providers in the order every loop must call them: unmarked providers
241/// in registration order, then the [`account`](HttpProvider::account) ones,
242/// also in registration order.
243///
244/// **Stable, and shared.** All three loops — export, plan, apply — read this
245/// one answer, because the order that finds a subject's rows is the order
246/// that erases them, and an export that reads the account provider last while
247/// an apply reads it first describes a deployment that does not exist. Within
248/// each group the registration order survives, so the sequence a build
249/// produced is legible rather than re-sorted behind the reader's back.
250pub(crate) fn in_erasure_order(providers: &[HttpProvider]) -> Vec<&HttpProvider> {
251    let (mut rest, account): (Vec<_>, Vec<_>) =
252        providers.iter().partition(|provider| !provider.account);
253    rest.extend(account);
254    rest
255}
256
257/// The shared body both sides sign.
258pub(crate) fn request_body(subject: &str, request_id: &str) -> bytes::Bytes {
259    bytes::Bytes::from(
260        serde_json::to_vec(&json!({ "subject": subject, "request_id": request_id }))
261            .expect("a JSON object serialises"),
262    )
263}
264
265/// `t=<unix secs>,v1=<lowercase hex HMAC-SHA256(secret, "{t}.{body}")>` — the
266/// layout `cratefield_core::StripeStyle` verifies.
267fn signature_header(secret: &str, signed_at: u64, body: &[u8]) -> String {
268    let mut mac =
269        Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("HMAC accepts any key length");
270    mac.update(signed_at.to_string().as_bytes());
271    mac.update(b".");
272    mac.update(body);
273    format!(
274        "t={signed_at},v1={}",
275        hex::encode(mac.finalize().into_bytes())
276    )
277}
278
279/// 2xx is success; a 4xx is the provider refusing the call; anything else (a
280/// 5xx, a stray 3xx) is an outage.
281fn check_status(response: &http::Response<bytes::Bytes>) -> Result<(), ProviderError> {
282    let status = response.status();
283    if status.is_success() {
284        Ok(())
285    } else if status.is_client_error() {
286        Err(ProviderError::Rejected)
287    } else {
288        Err(ProviderError::Unavailable)
289    }
290}
291
292/// Each entry of a response's `sections` array, rendered by `render`.
293fn sections(
294    body: &[u8],
295    render: impl Fn(&Value) -> Result<Value, ProviderError>,
296) -> Result<Vec<Value>, ProviderError> {
297    let value: Value = serde_json::from_slice(body).map_err(|_| ProviderError::InvalidResponse)?;
298    let Value::Object(mut object) = value else {
299        return Err(ProviderError::InvalidResponse);
300    };
301    let Some(Value::Array(sections)) = object.remove("sections") else {
302        return Err(ProviderError::InvalidResponse);
303    };
304    sections.iter().map(render).collect()
305}
306
307/// An export response's sections: `{name, description?, data}`, rebuilt so no
308/// unrequested field rides through to the caller.
309fn provider_sections(body: &[u8]) -> Result<Vec<Value>, ProviderError> {
310    sections(body, |section| {
311        let name = section
312            .get("name")
313            .and_then(Value::as_str)
314            .ok_or(ProviderError::InvalidResponse)?;
315        let data = section.get("data").ok_or(ProviderError::InvalidResponse)?;
316        let mut rendered = json!({ "name": name, "data": data });
317        match section.get("description") {
318            None => {}
319            Some(Value::String(description)) => rendered["description"] = json!(description),
320            Some(_) => return Err(ProviderError::InvalidResponse),
321        }
322        Ok(rendered)
323    })
324}
325
326/// A plan response's sections: `{name, action, reason?}`, with `reason`
327/// required when the action is `retain` — "we are keeping this" without a
328/// reason is the one thing a subject is most entitled to.
329fn plan_sections(body: &[u8]) -> Result<Vec<Value>, ProviderError> {
330    sections(body, |section| {
331        let name = section
332            .get("name")
333            .and_then(Value::as_str)
334            .ok_or(ProviderError::InvalidResponse)?;
335        match section.get("action").and_then(Value::as_str) {
336            Some(action @ ("delete" | "anonymise")) => {
337                Ok(json!({ "name": name, "action": action }))
338            }
339            Some("retain") => {
340                let reason = section
341                    .get("reason")
342                    .and_then(Value::as_str)
343                    .filter(|reason| !reason.is_empty())
344                    .ok_or(ProviderError::InvalidResponse)?;
345                Ok(json!({ "name": name, "action": "retain", "reason": reason }))
346            }
347            _ => Err(ProviderError::InvalidResponse),
348        }
349    })
350}
351
352/// The request id an erasure shares between plan, confirm and every retry,
353/// derived from the confirm token so all three agree with no stored state.
354pub(crate) fn erase_request_id(token: &str) -> String {
355    format!("erase_{}", &digest_hex(token.as_bytes())[..32])
356}
357
358/// The request id one export carries, derived from the subject and the
359/// moment it ran.
360pub(crate) fn export_request_id(subject: &str, now: u64) -> String {
361    let seed = format!("{subject}:{now}");
362    format!("export_{}", &digest_hex(seed.as_bytes())[..32])
363}
364
365/// Lowercase hex of `SHA-256(bytes)`.
366fn digest_hex(bytes: &[u8]) -> String {
367    hex::encode(Sha256::digest(bytes))
368}
369
370#[cfg(test)]
371mod tests {
372    use super::*;
373
374    #[test]
375    fn a_signature_is_t_and_lowercase_hex_v1() {
376        let header = signature_header("secret", 1_800_000_000, b"{}");
377        let (t, sig) = header.split_once(",v1=").expect("t=…,v1=…");
378        assert_eq!(t, "t=1800000000");
379        assert_eq!(sig.len(), 64);
380        assert!(
381            sig.bytes()
382                .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
383        );
384    }
385
386    #[test]
387    fn response_sections_are_validated_and_rebuilt() {
388        // Malformed envelopes, an export section with no `data`, a non-string
389        // description.
390        for body in [
391            b"not json".as_slice(),
392            b"{}".as_slice(),
393            br#"{"sections":[{"name":"a"}]}"#.as_slice(),
394            br#"{"sections":[{"name":"a","data":1,"description":2}]}"#.as_slice(),
395        ] {
396            assert_eq!(provider_sections(body), Err(ProviderError::InvalidResponse));
397        }
398        // An unknown action, and a `retain` with no (or an empty) reason.
399        for body in [
400            br#"{"sections":[{"name":"x","action":"shred"}]}"#.as_slice(),
401            br#"{"sections":[{"name":"invoices","action":"retain"}]}"#.as_slice(),
402            br#"{"sections":[{"name":"invoices","action":"retain","reason":""}]}"#.as_slice(),
403        ] {
404            assert_eq!(plan_sections(body), Err(ProviderError::InvalidResponse));
405        }
406
407        // A valid plan renders action and reason, dropping unknown fields.
408        let body = br#"{"sections":[
409            {"name":"profile","action":"delete"},
410            {"name":"invoices","action":"retain","reason":"Tax law."}
411        ],"extra":"dropped"}"#;
412        let sections = plan_sections(body).expect("valid");
413        assert_eq!(sections[0], json!({"name":"profile","action":"delete"}));
414        assert_eq!(
415            sections[1],
416            json!({"name":"invoices","action":"retain","reason":"Tax law."})
417        );
418    }
419}
cratefield_module_privacy/provider.rs

cratefield_module_privacy/
provider.rs
