waf_capture/
shipper.rs

1// SPDX-FileCopyrightText: 2026 0x00spor3
2// SPDX-License-Identifier: Apache-2.0
3
4//! The HTTP shipper — the decision-path half of the transport.
5//!
6//! Captures leave the ring on the (rare) denied path and are handed to a bounded channel; a single
7//! background task drains it and `POST`s each capture to the control-plane `/ingest/captures` with a
8//! short retry. The channel is **bounded** and the hand-off is non-blocking ([`CaptureSink::enqueue`]
9//! uses `try_send`): under a burst the shipper drops rather than back-pressuring the request path —
10//! availability of the proxy always wins over completeness of forensics. Dropped/failed counts are
11//! surfaced (atomic counter + periodic log) so the loss is visible, never silent.
12//!
13//! The reqwest client mirrors `waf-threatintel`'s `HttpListFeed`: `rustls-no-provider` + an explicit
14//! `ring` provider, so capture shipping pulls no second crypto stack (no aws-lc-rs/cmake).
15
16use std::sync::atomic::{AtomicU64, Ordering};
17use std::sync::Arc;
18use std::time::Duration;
19
20use tokio::sync::mpsc;
21use tokio_rustls::rustls::crypto::ring;
22use tokio_rustls::rustls::{ClientConfig, RootCertStore};
23use tracing::{debug, info, warn};
24
25use crate::{CaptureConfig, CapturedRequest};
26
27/// Max POST attempts per capture before it is given up (best-effort).
28const MAX_ATTEMPTS: u32 = 3;
29/// Upstream round-trip timeout — a stalled control-plane must not pin the shipper.
30const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
31
32/// The producer half held by the verdict-join layer: a non-blocking handle onto the shipper queue.
33/// Cloneable and cheap; dropping the last clone closes the channel and ends the shipper task.
34#[derive(Clone)]
35pub struct CaptureSink {
36    tx: mpsc::Sender<CapturedRequest>,
37    dropped: Arc<AtomicU64>,
38}
39
40impl CaptureSink {
41    /// Create a sink and its receiver. Shared by [`CaptureShipper::spawn`] and tests.
42    pub(crate) fn channel(capacity: usize) -> (Self, mpsc::Receiver<CapturedRequest>) {
43        let (tx, rx) = mpsc::channel(capacity.max(1));
44        (Self { tx, dropped: Arc::new(AtomicU64::new(0)) }, rx)
45    }
46
47    /// Enqueue a capture without blocking. On a full queue the capture is dropped and counted — the
48    /// request path is never stalled by a slow/unreachable control-plane.
49    pub fn enqueue(&self, capture: CapturedRequest) {
50        if self.tx.try_send(capture).is_err() {
51            self.dropped.fetch_add(1, Ordering::Relaxed);
52        }
53    }
54
55    /// Number of captures dropped because the queue was full (observability).
56    pub fn dropped(&self) -> u64 {
57        self.dropped.load(Ordering::Relaxed)
58    }
59}
60
61/// Spawns and owns the background shipping task.
62pub struct CaptureShipper;
63
64impl CaptureShipper {
65    /// Build the HTTPS-capable client, spawn the drain-and-POST task, and return the producer sink.
66    /// `token` (from the environment) is sent as a bearer on each POST when present. Errors only if
67    /// the TLS/client cannot be built (boot-fatal at wiring).
68    pub fn spawn(cfg: &CaptureConfig, token: Option<String>) -> Result<CaptureSink, String> {
69        let client = build_client()?;
70        let url = format!("{}/ingest/captures", cfg.controlplane_url.trim_end_matches('/'));
71        let (sink, rx) = CaptureSink::channel(cfg.queue_capacity);
72        let dropped = Arc::clone(&sink.dropped);
73        tokio::spawn(ship_loop(rx, client, url, token, dropped));
74        Ok(sink)
75    }
76}
77
78/// reqwest client with a `ring`-based rustls config (public CA roots). Mirrors `HttpListFeed`.
79fn build_client() -> Result<reqwest::Client, String> {
80    let mut roots = RootCertStore::empty();
81    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
82    let tls = ClientConfig::builder_with_provider(Arc::new(ring::default_provider()))
83        .with_safe_default_protocol_versions()
84        .map_err(|e| format!("rustls client config: {e}"))?
85        .with_root_certificates(roots)
86        .with_no_client_auth();
87    reqwest::Client::builder()
88        .use_preconfigured_tls(tls)
89        .timeout(REQUEST_TIMEOUT)
90        .build()
91        .map_err(|e| format!("http client: {e}"))
92}
93
94/// Drain the queue, POSTing each capture with a short retry. Ends when every sink clone is dropped.
95async fn ship_loop(
96    mut rx: mpsc::Receiver<CapturedRequest>,
97    client: reqwest::Client,
98    url: String,
99    token: Option<String>,
100    dropped: Arc<AtomicU64>,
101) {
102    let (mut shipped, mut failed) = (0u64, 0u64);
103    while let Some(capture) = rx.recv().await {
104        let mut ok = false;
105        for attempt in 1..=MAX_ATTEMPTS {
106            let mut rb = client.post(&url).json(&capture);
107            if let Some(t) = &token {
108                rb = rb.bearer_auth(t);
109            }
110            match rb.send().await {
111                Ok(resp) if resp.status().is_success() => {
112                    ok = true;
113                    break;
114                }
115                Ok(resp) => debug!(status = %resp.status(), attempt, "capture ship: non-2xx"),
116                Err(e) => debug!(error = %e, attempt, "capture ship: request error"),
117            }
118            if attempt < MAX_ATTEMPTS {
119                tokio::time::sleep(Duration::from_millis(200 * u64::from(attempt))).await;
120            }
121        }
122        if ok {
123            shipped += 1;
124        } else {
125            failed += 1;
126            warn!(request_id = %capture.request_id, "capture shipping failed after retries; dropping");
127        }
128        // Periodic visibility without a log line per request.
129        if (shipped + failed).is_multiple_of(50) {
130            info!(shipped, failed, dropped = dropped.load(Ordering::Relaxed), "capture shipper stats");
131        }
132    }
133}
waf_capture/shipper.rs

waf_capture/
shipper.rs
