Add live TUI, structured filters, WebSocket capture, and session diffs

Made-with: Cursor

Author: Rikkarrr

Cargo.lock

@@ -41,6 +41,15 @@ url = "2.5"
 uuid = { version = "1.11", features = ["v4", "serde"] }
 futures-util = "0.3"
 indicatif = "0.17"
+rcgen = "0.14.7"
+tokio-rustls = "0.26.4"
+rustls-pemfile = "2.2.0"
+rustls = "0.23.37"
+time = "0.3.47"
+ratatui = "0.30.0"
+crossterm = "0.29.0"
+tokio-tungstenite = { version = "0.29.0", features = ["native-tls"] }
+hex = "0.4.3"
 
 [dev-dependencies]
 axum = "0.7"

Cargo.toml

@@ -346,6 +346,7 @@ pub fn import_har(
             status,
             duration_ms,
             error,
+            ws_frames: vec![],
         };
         session.persist_capture(rec)?;
     }

src/capture/har.rs

@@ -4,6 +4,7 @@ pub mod proxy;
 pub mod redact;
 pub mod session;
 pub mod store;
+pub mod websocket;
 
 pub use model::{CaptureRecord, HttpMessageSnapshot, ReplayRecord, SessionMeta};
 pub use proxy::run_proxy;

src/capture/mod.rs

@@ -1,6 +1,8 @@
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 
+use crate::capture::websocket::WsFrame;
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct SessionMeta {
     pub id: String,
@@ -30,6 +32,9 @@ pub struct CaptureRecord {
     pub status: Option<u16>,
     pub duration_ms: Option<u64>,
     pub error: Option<String>,
+    /// WebSocket frames (populated only for upgraded WS connections).
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub ws_frames: Vec<WsFrame>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/capture/model.rs

@@ -1,5 +1,6 @@
-//! Local HTTP forward proxy: explicit HTTP requests and **CONNECT** tunnels (TLS end-to-end;
-//! payload is not decrypted). HTTP traffic is forwarded with `reqwest` and fully captured.
+//! Local HTTP forward proxy: explicit HTTP requests and **CONNECT** tunnels.
+//! With `--decrypt`: TLS MITM via per-host certificates signed by a local root CA.
+//! Without `--decrypt`: CONNECT is tunneled end-to-end (no decryption).
 
 use std::convert::Infallible;
 use std::io::{self, IsTerminal};
@@ -22,8 +23,10 @@ use tokio::net::{TcpListener, TcpStream};
 use crate::capture::model::{CaptureRecord, HttpMessageSnapshot};
 use crate::capture::redact;
 use crate::capture::session::SessionManager;
-use crate::error::{Error, Result};
 use crate::config::AppConfig;
+use crate::error::{Error, Result};
+use crate::tls::ca::CertificateAuthority;
+use crate::tls::cert::make_tls_acceptor_for_host;
 
 const HOP_BY_HOP: &[&str] = &[
     "connection",
@@ -43,9 +46,30 @@ pub struct ProxyState {
     pub unsafe_show_secrets: bool,
     /// Spinner + message when stdout is a TTY; otherwise plain `println!` per capture.
     pub progress: Option<ProgressBar>,
+    /// When set, CONNECT is intercepted with TLS MITM instead of tunneled.
+    pub ca: Option<Arc<CertificateAuthority>>,
+    /// Live feed for TUI mode.
+    pub capture_tx: Option<tokio::sync::broadcast::Sender<CaptureRecord>>,
 }
 
 impl ProxyState {
+    pub fn persist_and_broadcast(
+        &self,
+        capture: CaptureRecord,
+        count: &AtomicU64,
+    ) {
+        let summary = format!("{} {}", capture.method, capture.url);
+        let ms = capture.duration_ms.unwrap_or(0);
+        if let Err(e) = self.session.persist_capture(capture.clone()) {
+            tracing::error!(?e, "persist capture failed");
+        }
+        if let Some(ref tx) = self.capture_tx {
+            let _ = tx.send(capture);
+        }
+        let n = count.fetch_add(1, Ordering::Relaxed) + 1;
+        self.notify_capture(n, &summary, ms);
+    }
+
     pub fn notify_capture(&self, n: u64, summary: &str, ms: u64) {
         let line = format!("{summary} · {ms}ms");
         if let Some(pb) = &self.progress {
@@ -202,7 +226,8 @@ async fn handle_connect(
 ) -> std::result::Result<Response<Full<Bytes>>, Infallible> {
     let started = Instant::now();
     let created_at = Utc::now();
-    let authority = match req.uri().authority().map(|a| a.as_str()) {
+    let authority_owned = req.uri().authority().map(|a| a.as_str().to_string());
+    let authority = match authority_owned.as_deref() {
         Some(a) if !a.is_empty() => a,
         _ => {
             return Ok(Response::builder()
@@ -263,7 +288,7 @@ async fn handle_connect(
                 created_at,
                 format!("CONNECT upstream: {e}"),
             );
-            let _ = state.session.persist_capture(rec);
+            state.persist_and_broadcast(rec, &count);
             return Ok(Response::builder()
                 .status(StatusCode::BAD_GATEWAY)
                 .body(Full::new(Bytes::from(format!("CONNECT failed: {e}"))))
@@ -304,31 +329,100 @@ async fn handle_connect(
         status: Some(200),
         duration_ms: Some(connect_ms),
         error: None,
+        ws_frames: vec![],
     };
-    if let Err(e) = state.session.persist_capture(capture) {
-        tracing::error!(?e, "persist CONNECT capture failed");
-    }
-    let n = count.fetch_add(1, Ordering::Relaxed) + 1;
-    state.notify_capture(n, &format!("CONNECT {target}"), connect_ms);
-
-    tokio::task::spawn(async move {
-        match upgrade.await {
-            Ok(up) => {
-                // Client side: hyper `Upgraded` via TokioIo. Upstream: raw `TcpStream` (tokio I/O).
-                let mut client = TokioIo::new(up);
-                let mut server = upstream;
-                let _ = tokio::io::copy_bidirectional(&mut client, &mut server).await;
+    state.persist_and_broadcast(capture, &count);
+
+    if let Some(ref ca) = state.ca {
+        let ca = Arc::clone(ca);
+        let host_for_cert = authority.split(':').next().unwrap_or(authority).to_string();
+        let state2 = Arc::clone(&state);
+        let count2 = Arc::clone(&count);
+        tokio::task::spawn(async move {
+            match upgrade.await {
+                Ok(up) => {
+                    handle_mitm(up, upstream, host_for_cert, ca, state2, count2).await;
+                }
+                Err(e) => tracing::debug!("CONNECT upgrade dropped: {e}"),
             }
-            Err(e) => tracing::debug!("CONNECT upgrade dropped: {e}"),
-        }
-    });
+        });
+    } else {
+        tokio::task::spawn(async move {
+            match upgrade.await {
+                Ok(up) => {
+                    let mut client = TokioIo::new(up);
+                    let mut server = upstream;
+                    let _ = tokio::io::copy_bidirectional(&mut client, &mut server).await;
+                }
+                Err(e) => tracing::debug!("CONNECT upgrade dropped: {e}"),
+            }
+        });
+    }
 
     Ok(Response::builder()
         .status(StatusCode::OK)
         .body(Full::new(Bytes::new()))
         .unwrap())
 }
 
+/// After TLS termination the inner HTTP request has a relative URI (e.g. `/api/v1/foo`).
+/// We rewrite it to an absolute `https://host/…` URL so our forwarding logic works.
+fn rewrite_connect_request_uri(req: &mut Request<Incoming>, host: &str) {
+    let path = req
+        .uri()
+        .path_and_query()
+        .map(|pq| pq.as_str())
+        .unwrap_or("/");
+    if let Ok(new_uri) = format!("https://{host}{path}").parse::<hyper::Uri>() {
+        *req.uri_mut() = new_uri;
+    }
+}
+
+/// MITM handler: terminate TLS with the client using a per-host cert, read plaintext
+/// HTTP inside the tunnel, forward to the real server via reqwest, capture everything.
+async fn handle_mitm(
+    upgraded: hyper::upgrade::Upgraded,
+    _upstream_tcp: TcpStream,
+    hostname: String,
+    ca: Arc<CertificateAuthority>,
+    state: Arc<ProxyState>,
+    count: Arc<AtomicU64>,
+) {
+    let acceptor = match make_tls_acceptor_for_host(&ca, &hostname) {
+        Ok(a) => a,
+        Err(e) => {
+            tracing::debug!("TLS acceptor for {hostname}: {e}");
+            return;
+        }
+    };
+
+    let client_io = TokioIo::new(upgraded);
+    let tls_stream = match acceptor.accept(client_io).await {
+        Ok(s) => s,
+        Err(e) => {
+            tracing::debug!("TLS accept for {hostname}: {e}");
+            return;
+        }
+    };
+
+    let tls_io = TokioIo::new(tls_stream);
+
+    let service = hyper::service::service_fn(move |mut req: Request<Incoming>| {
+        let state = Arc::clone(&state);
+        let count = Arc::clone(&count);
+        let host = hostname.clone();
+        async move {
+            rewrite_connect_request_uri(&mut req, &host);
+            handle_forward(req, state, count).await
+        }
+    });
+
+    let _ = ConnBuilder::new(TokioExecutor::new())
+        .serve_connection(tls_io, service)
+        .await;
+}
+
+/// Top-level handler dispatching CONNECT vs plain HTTP.
 async fn handle_request(
     req: Request<Incoming>,
     state: Arc<ProxyState>,
@@ -337,7 +431,16 @@ async fn handle_request(
     if req.method() == hyper::Method::CONNECT {
         return handle_connect(req, state, count).await;
     }
+    handle_forward(req, state, count).await
+}
 
+/// Forward a non-CONNECT HTTP request to the upstream and capture it.
+/// Called both from `handle_request` (plain HTTP) and from the MITM TLS service.
+async fn handle_forward(
+    req: Request<Incoming>,
+    state: Arc<ProxyState>,
+    count: Arc<AtomicU64>,
+) -> std::result::Result<Response<Full<Bytes>>, Infallible> {
     let started = Instant::now();
     let created_at = Utc::now();
     let url = match build_target_url(&req) {
@@ -409,7 +512,7 @@ async fn handle_request(
                 created_at,
                 e.to_string(),
             );
-            let _ = state.session.persist_capture(rec);
+            state.persist_and_broadcast(rec, &count);
             return Ok(Response::builder()
                 .status(StatusCode::BAD_GATEWAY)
                 .body(Full::new(Bytes::from(format!("request build error: {e}"))))
@@ -452,12 +555,9 @@ async fn handle_request(
                         status: st,
                         duration_ms: Some(elapsed),
                         error: None,
+                        ws_frames: vec![],
                     };
-                    if let Err(e) = state.session.persist_capture(capture) {
-                        tracing::error!(?e, "persist capture failed");
-                    }
-                    let n = count.fetch_add(1, Ordering::Relaxed) + 1;
-                    state.notify_capture(n, &format!("{method} {url}"), elapsed);
+                    state.persist_and_broadcast(capture, &count);
 
                     let code = StatusCode::from_u16(status).unwrap_or(StatusCode::BAD_GATEWAY);
                     let mut res = Response::builder().status(code);
@@ -487,7 +587,7 @@ async fn handle_request(
                         created_at,
                         e.to_string(),
                     );
-                    let _ = state.session.persist_capture(rec);
+                    state.persist_and_broadcast(rec, &count);
                     Ok(Response::builder()
                         .status(StatusCode::BAD_GATEWAY)
                         .body(Full::new(Bytes::from(format!("response body error: {e}"))))
@@ -505,7 +605,7 @@ async fn handle_request(
                 created_at,
                 e.to_string(),
             );
-            let _ = state.session.persist_capture(rec);
+            state.persist_and_broadcast(rec, &count);
             Ok(Response::builder()
                 .status(StatusCode::BAD_GATEWAY)
                 .body(Full::new(Bytes::from(format!("upstream error: {e}"))))
@@ -534,6 +634,7 @@ fn error_capture(
         status: None,
         duration_ms: Some(started.elapsed().as_millis() as u64),
         error: Some(err),
+        ws_frames: vec![],
     }
 }
 
@@ -542,6 +643,8 @@ pub async fn run_proxy(
     session: SessionManager,
     bind: SocketAddr,
     unsafe_show_secrets: bool,
+    ca: Option<CertificateAuthority>,
+    capture_tx: Option<tokio::sync::broadcast::Sender<CaptureRecord>>,
 ) -> Result<()> {
     let client = reqwest::Client::builder()
         .timeout(Duration::from_secs(cfg.timeout_secs))
@@ -562,12 +665,15 @@ pub async fn run_proxy(
         None
     };
 
+    let decrypt = ca.is_some();
     let state = Arc::new(ProxyState {
         cfg: cfg.clone(),
         session: Arc::new(session),
         client,
         unsafe_show_secrets,
         progress,
+        ca: ca.map(Arc::new),
+        capture_tx,
     });
 
     let listener = TcpListener::bind(bind).await?;
@@ -579,11 +685,19 @@ pub async fn run_proxy(
         "proxy listening on".dimmed(),
         bind.to_string().yellow()
     );
-    eprintln!(
-        "{}",
-        "HTTP + HTTPS (CONNECT tunnel). TLS inside CONNECT is not decrypted."
-            .dimmed()
-    );
+    if decrypt {
+        eprintln!(
+            "{}",
+            "HTTP + HTTPS (TLS MITM decryption enabled). HTTPS traffic will be captured in plaintext."
+                .dimmed()
+        );
+    } else {
+        eprintln!(
+            "{}",
+            "HTTP + HTTPS (CONNECT tunnel). TLS inside CONNECT is not decrypted."
+                .dimmed()
+        );
+    }
     eprintln!(
         "{}",
         "Press Ctrl+C to stop capture.".dimmed()

src/capture/proxy.rs

@@ -167,6 +167,31 @@ impl CaptureStore {
         }
     }
 
+    /// Resolve a session id or prefix. `last` returns the most recent.
+    pub fn resolve_session_id(&self, id: &str) -> Result<String> {
+        if id.eq_ignore_ascii_case("last") {
+            let sessions = self.list_sessions()?;
+            return sessions
+                .first()
+                .map(|s| s.id.clone())
+                .ok_or_else(|| Error::SessionNotFound("last".into()));
+        }
+        let meta_path = self.session_meta_path(id);
+        if meta_path.exists() {
+            return Ok(id.to_string());
+        }
+        let sessions = self.list_sessions()?;
+        let matches: Vec<_> = sessions.iter().filter(|s| s.id.starts_with(id)).collect();
+        match matches.len() {
+            0 => Err(Error::SessionNotFound(id.to_string())),
+            1 => Ok(matches[0].id.clone()),
+            _ => Err(Error::msg(format!(
+                "ambiguous session id prefix {id:?}: {} matches",
+                matches.len()
+            ))),
+        }
+    }
+
     pub fn write_replay(&self, session_id: &str, rec: &ReplayRecord) -> Result<()> {
         let dir = self.session_dir(session_id).join("replays");
         crate::util::fs::ensure_dir(&dir)?;