hns_browser_loopback_proxy/chromium/
backend.rs

1//! Platform-neutral request boundary used by the loopback HTTP server.
2
3use std::fmt;
4use std::io::{self, Read, Write};
5use std::sync::atomic::{AtomicBool, Ordering};
6use std::sync::{Arc, Condvar, Mutex};
7use std::time::Duration;
8use thiserror::Error;
9
10/// Cloneable cooperative cancellation shared by a running proxy and backend
11/// work started for that proxy generation.
12#[derive(Clone, Default)]
13pub struct CancellationToken {
14    inner: Arc<CancellationState>,
15}
16
17#[derive(Default)]
18struct CancellationState {
19    cancelled: AtomicBool,
20    wait_lock: Mutex<()>,
21    changed: Condvar,
22}
23
24impl CancellationToken {
25    pub fn new() -> Self {
26        Self::default()
27    }
28
29    /// Cancels all clones. Returns `true` only for the first cancellation.
30    pub(crate) fn cancel(&self) -> bool {
31        // Synchronize with the condition-variable handoff so cancellation
32        // cannot be notified in the gap between a waiter checking the atomic
33        // flag and going to sleep.
34        let _guard = self
35            .inner
36            .wait_lock
37            .lock()
38            .unwrap_or_else(|poisoned| poisoned.into_inner());
39        if self.inner.cancelled.swap(true, Ordering::AcqRel) {
40            return false;
41        }
42        self.inner.changed.notify_all();
43        true
44    }
45
46    pub fn is_cancelled(&self) -> bool {
47        self.inner.cancelled.load(Ordering::Acquire)
48    }
49
50    /// Waits until cancellation or the timeout. Returns whether cancellation
51    /// was observed.
52    pub fn wait_cancelled_timeout(&self, timeout: Duration) -> bool {
53        if self.is_cancelled() {
54            return true;
55        }
56        let guard = self
57            .inner
58            .wait_lock
59            .lock()
60            .unwrap_or_else(|poisoned| poisoned.into_inner());
61        if self.is_cancelled() {
62            return true;
63        }
64        let _guard = self
65            .inner
66            .changed
67            .wait_timeout_while(guard, timeout, |_| !self.is_cancelled())
68            .unwrap_or_else(|poisoned| poisoned.into_inner());
69        self.is_cancelled()
70    }
71}
72
73impl fmt::Debug for CancellationToken {
74    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
75        formatter
76            .debug_struct("CancellationToken")
77            .field("cancelled", &self.is_cancelled())
78            .finish()
79    }
80}
81
82#[derive(Clone, Eq, PartialEq)]
83pub struct ProxyHeader {
84    pub name: String,
85    pub value: String,
86}
87
88impl ProxyHeader {
89    pub fn new(name: impl Into<String>, value: impl Into<String>) -> Self {
90        Self {
91            name: name.into(),
92            value: value.into(),
93        }
94    }
95}
96
97impl fmt::Debug for ProxyHeader {
98    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
99        formatter.write_str("ProxyHeader(<redacted>)")
100    }
101}
102
103#[derive(Clone, Eq, PartialEq)]
104pub enum ProxyRequestBody {
105    Empty,
106    Bytes(Vec<u8>),
107}
108
109impl ProxyRequestBody {
110    pub fn len(&self) -> usize {
111        match self {
112            Self::Empty => 0,
113            Self::Bytes(bytes) => bytes.len(),
114        }
115    }
116
117    pub fn is_empty(&self) -> bool {
118        self.len() == 0
119    }
120
121    pub fn as_bytes(&self) -> &[u8] {
122        match self {
123            Self::Empty => &[],
124            Self::Bytes(bytes) => bytes,
125        }
126    }
127
128    pub fn into_bytes(self) -> Vec<u8> {
129        match self {
130            Self::Empty => Vec::new(),
131            Self::Bytes(bytes) => bytes,
132        }
133    }
134}
135
136impl fmt::Debug for ProxyRequestBody {
137    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
138        formatter
139            .debug_struct("ProxyRequestBody")
140            .field("len", &self.len())
141            .finish()
142    }
143}
144
145#[derive(Clone, Eq, PartialEq)]
146pub struct ProxyRequest {
147    pub method: String,
148    pub scheme: String,
149    pub host: String,
150    pub port: u16,
151    pub path_and_query: String,
152    pub headers: Vec<ProxyHeader>,
153    pub body: ProxyRequestBody,
154}
155
156impl fmt::Debug for ProxyRequest {
157    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
158        formatter
159            .debug_struct("ProxyRequest")
160            .field("port", &self.port)
161            .field("header_count", &self.headers.len())
162            .field("body", &self.body)
163            .finish_non_exhaustive()
164    }
165}
166
167#[derive(Clone, Eq, PartialEq)]
168pub struct ProxyResponseHead {
169    pub status_code: u16,
170    pub reason_phrase: String,
171    pub headers: Vec<ProxyHeader>,
172    /// Opaque backend-local correlation used only for typed status delivery.
173    /// It is never serialized into the proxy response.
174    pub observation_id: Option<u64>,
175}
176
177impl fmt::Debug for ProxyResponseHead {
178    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
179        formatter
180            .debug_struct("ProxyResponseHead")
181            .field("status_code", &self.status_code)
182            .field("header_count", &self.headers.len())
183            .field("observation_id_present", &self.observation_id.is_some())
184            .finish_non_exhaustive()
185    }
186}
187
188pub enum ProxyResponseBody {
189    Bytes(Vec<u8>),
190    /// A length-delimited response stream. Readers used by production
191    /// backends must have bounded I/O waits and must be released promptly
192    /// when the request's cancellation token is cancelled; proxy shutdown
193    /// joins rather than detaches client workers.
194    Stream {
195        expected_len: u64,
196        reader: Box<dyn Read + Send>,
197    },
198}
199
200impl ProxyResponseBody {
201    pub fn empty() -> Self {
202        Self::Bytes(Vec::new())
203    }
204
205    pub fn expected_len(&self) -> u64 {
206        match self {
207            Self::Bytes(bytes) => u64::try_from(bytes.len()).unwrap_or(u64::MAX),
208            Self::Stream { expected_len, .. } => *expected_len,
209        }
210    }
211
212    pub fn is_empty(&self) -> bool {
213        self.expected_len() == 0
214    }
215}
216
217impl fmt::Debug for ProxyResponseBody {
218    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
219        formatter
220            .debug_struct("ProxyResponseBody")
221            .field("expected_len", &self.expected_len())
222            .finish()
223    }
224}
225
226/// Authority hook which atomically validates and publishes one response head.
227///
228/// Implementations must hold the same exclusion boundary used by policy,
229/// lifecycle, and generation invalidation for the complete `publish`
230/// operation. The operation is intentionally limited to the sanitized
231/// response head; streaming bodies and tunnels retain their own cancellation
232/// and freshness checks.
233pub trait ProxyPublicationAuthority: Send + Sync + 'static {
234    fn publish(&self, operation: &mut dyn FnMut() -> io::Result<()>) -> io::Result<()>;
235}
236
237struct NoopProxyPublicationAuthority;
238
239impl ProxyPublicationAuthority for NoopProxyPublicationAuthority {
240    fn publish(&self, operation: &mut dyn FnMut() -> io::Result<()>) -> io::Result<()> {
241        operation()
242    }
243}
244
245/// Cloneable RAII permit carried from a backend result to loopback
246/// response-head publication.
247#[derive(Clone)]
248pub struct ProxyPublicationPermit {
249    authority: Arc<dyn ProxyPublicationAuthority>,
250}
251
252impl ProxyPublicationPermit {
253    pub fn new(authority: Arc<dyn ProxyPublicationAuthority>) -> Self {
254        Self { authority }
255    }
256
257    pub fn publish(&self, operation: impl FnOnce() -> io::Result<()>) -> io::Result<()> {
258        let mut operation = Some(operation);
259        self.authority.publish(&mut || {
260            operation
261                .take()
262                .ok_or_else(|| io::Error::other("publication operation was already consumed"))?(
263            )
264        })
265    }
266}
267
268impl Default for ProxyPublicationPermit {
269    fn default() -> Self {
270        Self::new(Arc::new(NoopProxyPublicationAuthority))
271    }
272}
273
274impl fmt::Debug for ProxyPublicationPermit {
275    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
276        formatter.write_str("ProxyPublicationPermit(<redacted authority>)")
277    }
278}
279
280pub struct ProxyResponse {
281    pub head: ProxyResponseHead,
282    pub body: ProxyResponseBody,
283    pub publication_permit: ProxyPublicationPermit,
284}
285
286/// One typed HTTP Upgrade response and its live origin connection. The proxy
287/// validates and sanitizes the response head before exposing it to a client.
288///
289/// Production streams must bound every read and write wait. In particular,
290/// an otherwise idle `read` must periodically return `TimedOut` or
291/// `WouldBlock` so the proxy can observe generation cancellation without
292/// detaching tunnel workers.
293pub struct ProxyTunnel {
294    pub head: ProxyResponseHead,
295    pub stream: Box<dyn ProxyTunnelIo>,
296    pub publication_permit: ProxyPublicationPermit,
297}
298
299/// Pre-TLS disposition for one authenticated outer CONNECT.
300pub enum ProxyConnectOpen {
301    /// Continue through the local certificate and HTTP gateway path.
302    Intercept,
303    /// Namespace selection required browser-owned WebPKI, but its exact
304    /// origin socket could not be opened. Never substitute local TLS.
305    WebPkiUnavailable,
306    /// Tunnel browser TLS unchanged to one backend-selected origin socket.
307    WebPkiPassthrough(ProxyConnectTunnel),
308}
309
310/// A raw, policy-bound CONNECT stream. The publication permit must validate
311/// before the proxy exposes `200 Connection Established`.
312pub struct ProxyConnectTunnel {
313    /// Origin-to-browser half. Reads must have bounded waits.
314    pub reader: Box<dyn Read + Send>,
315    /// Browser-to-origin half. Writes must have bounded waits.
316    pub writer: Box<dyn Write + Send>,
317    /// Wakes both halves when either direction finishes or is revoked.
318    pub shutdown: Arc<dyn ProxyConnectShutdown>,
319    pub observation_id: Option<u64>,
320    pub correlation_epoch: u64,
321    pub publication_permit: ProxyPublicationPermit,
322}
323
324/// Thread-safe, nonblocking, idempotent close which must wake both halves.
325///
326/// The CONNECT pump may invoke this concurrently from either direction.
327pub trait ProxyConnectShutdown: Send + Sync {
328    fn shutdown(&self);
329}
330
331impl fmt::Debug for ProxyConnectOpen {
332    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
333        match self {
334            Self::Intercept => formatter.write_str("Intercept"),
335            Self::WebPkiUnavailable => formatter.write_str("WebPkiUnavailable"),
336            Self::WebPkiPassthrough(tunnel) => formatter
337                .debug_tuple("WebPkiPassthrough")
338                .field(tunnel)
339                .finish(),
340        }
341    }
342}
343
344impl fmt::Debug for ProxyConnectTunnel {
345    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
346        formatter
347            .debug_struct("ProxyConnectTunnel")
348            .field("transport", &"<redacted split duplex stream>")
349            .field("observation_id_present", &self.observation_id.is_some())
350            .field("correlation_epoch", &self.correlation_epoch)
351            .finish()
352    }
353}
354
355/// Result of opening an Upgrade route. Policy and resolution failures can be
356/// returned as a normal bounded HTTP response without falsely switching the
357/// client into tunnel mode.
358pub enum ProxyTunnelOpen {
359    Tunnel(ProxyTunnel),
360    Response(ProxyResponse),
361}
362
363impl fmt::Debug for ProxyTunnelOpen {
364    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
365        match self {
366            Self::Tunnel(tunnel) => formatter.debug_tuple("Tunnel").field(tunnel).finish(),
367            Self::Response(response) => formatter.debug_tuple("Response").field(response).finish(),
368        }
369    }
370}
371
372impl fmt::Debug for ProxyTunnel {
373    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
374        formatter
375            .debug_struct("ProxyTunnel")
376            .field("head", &self.head)
377            .field("stream", &"<redacted duplex stream>")
378            .finish()
379    }
380}
381
382pub trait ProxyTunnelIo: Read + Write + Send {}
383
384impl<T: Read + Write + Send> ProxyTunnelIo for T {}
385
386impl fmt::Debug for ProxyResponse {
387    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
388        formatter
389            .debug_struct("ProxyResponse")
390            .field("head", &self.head)
391            .field("body", &self.body)
392            .finish()
393    }
394}
395
396#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
397pub enum BackendError {
398    #[error("backend operation was cancelled")]
399    Cancelled,
400    #[error("backend rejected an invalid request")]
401    InvalidRequest,
402    #[error("backend policy denied the request")]
403    PolicyDenied,
404    #[error("name resolution failed")]
405    ResolutionFailed,
406    #[error("upstream TLS validation failed")]
407    TlsValidationFailed,
408    #[error("upstream service is unavailable")]
409    UpstreamUnavailable,
410    #[error("upstream returned an invalid response")]
411    InvalidResponse,
412    #[error("upstream response exceeds the configured limit")]
413    ResponseTooLarge,
414    #[error("backend does not support HTTP Upgrade tunnelling")]
415    UnsupportedUpgrade,
416    #[error("internal backend failure")]
417    Internal,
418}
419
420pub trait ProxyBackend: Send + Sync + 'static {
421    /// Resolves the outer CONNECT before local TLS termination. The default
422    /// preserves interception. Implementations may return passthrough only
423    /// after making the same namespace and TLS policy decision used by normal
424    /// gateway requests.
425    fn open_connect(
426        &self,
427        _request: ProxyRequest,
428        _cancellation: &CancellationToken,
429    ) -> Result<ProxyConnectOpen, BackendError> {
430        Ok(ProxyConnectOpen::Intercept)
431    }
432
433    /// Executes one admitted request. Implementations must observe
434    /// `cancellation` and bound every network/storage wait so proxy shutdown
435    /// can join all work. Implementations must not panic: shipping profiles
436    /// abort on panic. Errors must remain payload-free at this boundary.
437    fn execute(
438        &self,
439        request: ProxyRequest,
440        cancellation: &CancellationToken,
441    ) -> Result<ProxyResponse, BackendError>;
442
443    /// Opens one admitted HTTP Upgrade tunnel. Implementations must apply the
444    /// same resolution, address, TLS, and DANE policy as [`Self::execute`],
445    /// observe `cancellation`, and satisfy the bounded-I/O contract documented
446    /// on [`ProxyTunnel`]. The default fails closed for compatibility with
447    /// backends that only implement request/response traffic.
448    fn open_tunnel(
449        &self,
450        _request: ProxyRequest,
451        _cancellation: &CancellationToken,
452    ) -> Result<ProxyTunnelOpen, BackendError> {
453        Err(BackendError::UnsupportedUpgrade)
454    }
455}
456
457#[cfg(test)]
458mod tests {
459    use super::*;
460    use std::io::Cursor;
461    use std::thread;
462    use std::time::Instant;
463
464    #[test]
465    fn cancellation_is_shared_idempotent_and_wakes_waiters() {
466        let token = CancellationToken::new();
467        let waiter = token.clone();
468        let handle = thread::spawn(move || waiter.wait_cancelled_timeout(Duration::from_secs(2)));
469        assert!(token.cancel());
470        assert!(!token.cancel());
471        assert!(handle.join().expect("waiter joins"));
472    }
473
474    #[test]
475    fn uncancelled_wait_times_out() {
476        let token = CancellationToken::new();
477        let started = Instant::now();
478        assert!(!token.wait_cancelled_timeout(Duration::from_millis(5)));
479        assert!(started.elapsed() >= Duration::from_millis(5));
480    }
481
482    #[test]
483    fn diagnostics_redact_request_secrets_and_payloads() {
484        let request = ProxyRequest {
485            method: "GET".to_owned(),
486            scheme: "https".to_owned(),
487            host: "user:secret@example".to_owned(),
488            port: 443,
489            path_and_query: "/private?token=secret".to_owned(),
490            headers: vec![ProxyHeader::new("Authorization", "secret")],
491            body: ProxyRequestBody::Bytes(b"secret body".to_vec()),
492        };
493        let diagnostic = format!("{request:?}");
494        for secret in [
495            "user",
496            "secret",
497            "example",
498            "private",
499            "token",
500            "Authorization",
501        ] {
502            assert!(!diagnostic.contains(secret));
503        }
504        assert!(diagnostic.contains("header_count"));
505        assert!(diagnostic.contains("len"));
506    }
507
508    #[test]
509    fn response_stream_declares_length_without_debugging_contents() {
510        let body = ProxyResponseBody::Stream {
511            expected_len: 12,
512            reader: Box::new(Cursor::new(b"secret bytes".to_vec())),
513        };
514        assert_eq!(body.expected_len(), 12);
515        assert_eq!(
516            format!("{body:?}"),
517            "ProxyResponseBody { expected_len: 12 }"
518        );
519    }
520}
hns_browser_loopback_proxy/chromium/backend.rs

hns_browser_loopback_proxy/chromium/
backend.rs
