From e3b1ba970c474758bbacf63eb2b9e694a62bd362 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 12 Jul 2026 02:35:02 +0000 Subject: [PATCH 1/5] feat(core,storage): add TraceQuery filtering, TraceView DTO, and store query/clear - phantom-core: TraceQuery (method/status/url/time-range/trace_id filters) with a pure matches() predicate; StatusRange parses "404", "4xx", "400-499"; TraceView + RenderOptions as the shared agent-facing JSON shape (max_body truncation, headers_only, header redaction, body size reporting); HttpMethod FromStr; SpanId/TraceId from_hex. - phantom-storage: TraceStore::query via bounded by_time range scan (by_trace_id prefix scan when trace_id is set) with offset applied after filtering; TraceStore::clear via batch removes. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01BjmmAVFHuaU4Fb46QEd6ky --- crates/phantom-core/src/lib.rs | 2 + crates/phantom-core/src/query.rs | 253 +++++++++++++++++++ crates/phantom-core/src/storage.rs | 9 + crates/phantom-core/src/trace.rs | 94 ++++++++ crates/phantom-core/src/view.rs | 281 ++++++++++++++++++++++ crates/phantom-storage/src/fjall_store.rs | 204 ++++++++++++++++ 6 files changed, 843 insertions(+) create mode 100644 crates/phantom-core/src/query.rs create mode 100644 crates/phantom-core/src/view.rs diff --git a/crates/phantom-core/src/lib.rs b/crates/phantom-core/src/lib.rs index 07f75f0..98055dc 100644 --- a/crates/phantom-core/src/lib.rs +++ b/crates/phantom-core/src/lib.rs @@ -1,4 +1,6 @@ pub mod capture; pub mod error; +pub mod query; pub mod storage; pub mod trace; +pub mod view; diff --git a/crates/phantom-core/src/query.rs b/crates/phantom-core/src/query.rs new file mode 100644 index 0000000..312b355 --- /dev/null +++ b/crates/phantom-core/src/query.rs @@ -0,0 +1,253 @@ +use std::str::FromStr; +use std::time::SystemTime; + +use crate::trace::{HttpMethod, HttpTrace, TraceId}; + +/// Error returned when parsing a status-range expression. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("invalid status range: {0:?} (expected e.g. \"404\", \"4xx\", or \"400-499\")")] +pub struct ParseStatusRangeError(pub String); + +/// An inclusive range of HTTP status codes. +/// +/// Parses from `"404"` (exact), `"4xx"` (class), or `"400-499"` (explicit range). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StatusRange { + pub min: u16, + pub max: u16, +} + +impl StatusRange { + pub fn contains(&self, status: u16) -> bool { + (self.min..=self.max).contains(&status) + } +} + +impl FromStr for StatusRange { + type Err = ParseStatusRangeError; + + fn from_str(s: &str) -> Result { + let err = || ParseStatusRangeError(s.to_string()); + + // "4xx" / "4XX" — status class + if s.len() == 3 && s[1..].eq_ignore_ascii_case("xx") { + let class = s[..1].parse::().map_err(|_| err())?; + if !(1..=5).contains(&class) { + return Err(err()); + } + return Ok(Self { + min: class * 100, + max: class * 100 + 99, + }); + } + + // "400-499" — explicit range + if let Some((lo, hi)) = s.split_once('-') { + let min = lo.parse::().map_err(|_| err())?; + let max = hi.parse::().map_err(|_| err())?; + if min > max { + return Err(err()); + } + return Ok(Self { min, max }); + } + + // "404" — exact code + let code = s.parse::().map_err(|_| err())?; + Ok(Self { + min: code, + max: code, + }) + } +} + +/// A filter over stored traces. All set fields must match (logical AND); +/// unset fields match everything. +#[derive(Debug, Clone, Default)] +pub struct TraceQuery { + /// Match any of these methods; empty = all methods. + pub methods: Vec, + /// Match status codes within this inclusive range. + pub status: Option, + /// Match URLs containing this substring (case-insensitive). + pub url_contains: Option, + /// Only traces with `timestamp >= since` (inclusive). + pub since: Option, + /// Only traces with `timestamp <= until` (inclusive). + pub until: Option, + /// Restrict to spans of this trace ID. + pub trace_id: Option, + /// Maximum number of traces to return; 0 means the caller's default. + pub limit: usize, + /// Number of matching traces to skip (applied after filtering). + pub offset: usize, +} + +impl TraceQuery { + /// Returns true when `trace` matches every set filter field. + /// Pure predicate — storage layers scan and call this per trace. + pub fn matches(&self, trace: &HttpTrace) -> bool { + if !self.methods.is_empty() && !self.methods.contains(&trace.method) { + return false; + } + if let Some(range) = &self.status + && !range.contains(trace.status_code) + { + return false; + } + if let Some(pattern) = &self.url_contains + && !trace.url.to_lowercase().contains(&pattern.to_lowercase()) + { + return false; + } + if let Some(since) = self.since + && trace.timestamp < since + { + return false; + } + if let Some(until) = self.until + && trace.timestamp > until + { + return false; + } + if let Some(trace_id) = &self.trace_id + && &trace.trace_id != trace_id + { + return false; + } + true + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::time::{Duration, UNIX_EPOCH}; + + use super::*; + use crate::trace::SpanId; + + fn make_trace(method: HttpMethod, url: &str, status: u16, ts_secs: u64) -> HttpTrace { + HttpTrace { + span_id: SpanId([1; 8]), + trace_id: TraceId([2; 16]), + parent_span_id: None, + method, + url: url.to_string(), + request_headers: HashMap::new(), + request_body: None, + status_code: status, + response_headers: HashMap::new(), + response_body: None, + timestamp: UNIX_EPOCH + Duration::from_secs(ts_secs), + duration: Duration::from_millis(10), + source_addr: None, + dest_addr: None, + protocol_version: "HTTP/1.1".to_string(), + } + } + + #[test] + fn test_status_range_parse_exact() { + let r: StatusRange = "404".parse().unwrap(); + assert_eq!(r, StatusRange { min: 404, max: 404 }); + assert!(r.contains(404)); + assert!(!r.contains(403)); + } + + #[test] + fn test_status_range_parse_class() { + let r: StatusRange = "4xx".parse().unwrap(); + assert_eq!(r, StatusRange { min: 400, max: 499 }); + let r: StatusRange = "5XX".parse().unwrap(); + assert_eq!(r, StatusRange { min: 500, max: 599 }); + assert!("0xx".parse::().is_err()); + assert!("6xx".parse::().is_err()); + } + + #[test] + fn test_status_range_parse_explicit() { + let r: StatusRange = "400-499".parse().unwrap(); + assert_eq!(r, StatusRange { min: 400, max: 499 }); + assert!("499-400".parse::().is_err()); + assert!("abc".parse::().is_err()); + assert!("4x".parse::().is_err()); + assert!("".parse::().is_err()); + } + + #[test] + fn test_query_default_matches_everything() { + let q = TraceQuery::default(); + assert!(q.matches(&make_trace(HttpMethod::Get, "http://a/x", 200, 100))); + } + + #[test] + fn test_query_by_method() { + let q = TraceQuery { + methods: vec![HttpMethod::Post, HttpMethod::Put], + ..Default::default() + }; + assert!(q.matches(&make_trace(HttpMethod::Post, "http://a", 200, 0))); + assert!(!q.matches(&make_trace(HttpMethod::Get, "http://a", 200, 0))); + } + + #[test] + fn test_query_by_status() { + let q = TraceQuery { + status: Some("4xx".parse().unwrap()), + ..Default::default() + }; + assert!(q.matches(&make_trace(HttpMethod::Get, "http://a", 404, 0))); + assert!(!q.matches(&make_trace(HttpMethod::Get, "http://a", 200, 0))); + } + + #[test] + fn test_query_by_url_case_insensitive() { + let q = TraceQuery { + url_contains: Some("API/Users".to_string()), + ..Default::default() + }; + assert!(q.matches(&make_trace(HttpMethod::Get, "http://x/api/users/1", 200, 0))); + assert!(!q.matches(&make_trace(HttpMethod::Get, "http://x/health", 200, 0))); + } + + #[test] + fn test_query_time_range_inclusive() { + let q = TraceQuery { + since: Some(UNIX_EPOCH + Duration::from_secs(100)), + until: Some(UNIX_EPOCH + Duration::from_secs(200)), + ..Default::default() + }; + assert!(!q.matches(&make_trace(HttpMethod::Get, "u", 200, 99))); + assert!(q.matches(&make_trace(HttpMethod::Get, "u", 200, 100))); + assert!(q.matches(&make_trace(HttpMethod::Get, "u", 200, 200))); + assert!(!q.matches(&make_trace(HttpMethod::Get, "u", 200, 201))); + } + + #[test] + fn test_query_by_trace_id() { + let q = TraceQuery { + trace_id: Some(TraceId([2; 16])), + ..Default::default() + }; + assert!(q.matches(&make_trace(HttpMethod::Get, "u", 200, 0))); + let q = TraceQuery { + trace_id: Some(TraceId([9; 16])), + ..Default::default() + }; + assert!(!q.matches(&make_trace(HttpMethod::Get, "u", 200, 0))); + } + + #[test] + fn test_query_combined_filters() { + let q = TraceQuery { + methods: vec![HttpMethod::Get], + status: Some("200".parse().unwrap()), + url_contains: Some("/api".to_string()), + ..Default::default() + }; + assert!(q.matches(&make_trace(HttpMethod::Get, "http://x/api/a", 200, 0))); + assert!(!q.matches(&make_trace(HttpMethod::Post, "http://x/api/a", 200, 0))); + assert!(!q.matches(&make_trace(HttpMethod::Get, "http://x/api/a", 500, 0))); + assert!(!q.matches(&make_trace(HttpMethod::Get, "http://x/other", 200, 0))); + } +} diff --git a/crates/phantom-core/src/storage.rs b/crates/phantom-core/src/storage.rs index 8fe3767..005d5d1 100644 --- a/crates/phantom-core/src/storage.rs +++ b/crates/phantom-core/src/storage.rs @@ -1,4 +1,5 @@ use crate::error::StorageError; +use crate::query::TraceQuery; use crate::trace::{HttpTrace, SpanId, TraceId}; /// Abstraction over trace storage backends. @@ -20,4 +21,12 @@ pub trait TraceStore: Send + Sync { /// Get total trace count. fn count(&self) -> Result; + + /// Filtered listing (newest first). See [`TraceQuery`] for filter semantics; + /// `query.offset` is applied after filtering, `query.limit` of 0 means + /// the implementation's default page size. + fn query(&self, query: &TraceQuery) -> Result, StorageError>; + + /// Delete all stored traces and their indices. + fn clear(&self) -> Result<(), StorageError>; } diff --git a/crates/phantom-core/src/trace.rs b/crates/phantom-core/src/trace.rs index 56bdd43..7e025e8 100644 --- a/crates/phantom-core/src/trace.rs +++ b/crates/phantom-core/src/trace.rs @@ -1,9 +1,24 @@ use std::collections::HashMap; use std::fmt; +use std::str::FromStr; use std::time::{Duration, SystemTime}; use serde::{Deserialize, Serialize}; +/// Decodes a fixed-length lowercase/uppercase hex string into a byte array. +fn decode_hex(s: &str) -> Option<[u8; N]> { + if s.len() != N * 2 || !s.is_ascii() { + return None; + } + let mut bytes = [0u8; N]; + for (i, chunk) in s.as_bytes().chunks_exact(2).enumerate() { + let hi = (chunk[0] as char).to_digit(16)?; + let lo = (chunk[1] as char).to_digit(16)?; + bytes[i] = ((hi << 4) | lo) as u8; + } + Some(bytes) +} + /// Unique identifier for a trace (W3C Trace Context compatible, 128-bit). #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct TraceId(pub [u8; 16]); @@ -12,6 +27,11 @@ impl TraceId { pub fn as_bytes(&self) -> &[u8; 16] { &self.0 } + + /// Parses a 32-character hex string (the `Display` format) back into a `TraceId`. + pub fn from_hex(s: &str) -> Option { + decode_hex(s).map(Self) + } } impl fmt::Display for TraceId { @@ -31,6 +51,11 @@ impl SpanId { pub fn as_bytes(&self) -> &[u8; 8] { &self.0 } + + /// Parses a 16-character hex string (the `Display` format) back into a `SpanId`. + pub fn from_hex(s: &str) -> Option { + decode_hex(s).map(Self) + } } impl fmt::Display for SpanId { @@ -74,6 +99,30 @@ impl fmt::Display for HttpMethod { } } +/// Error returned when parsing an unrecognized HTTP method string. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("unknown HTTP method: {0:?}")] +pub struct ParseMethodError(pub String); + +impl FromStr for HttpMethod { + type Err = ParseMethodError; + + fn from_str(s: &str) -> Result { + match s.to_ascii_uppercase().as_str() { + "GET" => Ok(Self::Get), + "POST" => Ok(Self::Post), + "PUT" => Ok(Self::Put), + "DELETE" => Ok(Self::Delete), + "PATCH" => Ok(Self::Patch), + "HEAD" => Ok(Self::Head), + "OPTIONS" => Ok(Self::Options), + "TRACE" => Ok(Self::Trace), + "CONNECT" => Ok(Self::Connect), + _ => Err(ParseMethodError(s.to_string())), + } + } +} + /// A complete HTTP request-response pair with timing metadata. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct HttpTrace { @@ -104,3 +153,48 @@ pub struct HttpTrace { pub dest_addr: Option, pub protocol_version: String, } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_span_id_from_hex_round_trip() { + let id = SpanId([0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef]); + let hex = id.to_string(); + assert_eq!(hex, "0123456789abcdef"); + assert_eq!(SpanId::from_hex(&hex), Some(id)); + } + + #[test] + fn test_trace_id_from_hex_round_trip() { + let id = TraceId([0xff; 16]); + let hex = id.to_string(); + assert_eq!(hex.len(), 32); + assert_eq!(TraceId::from_hex(&hex), Some(id)); + // uppercase also accepted + assert_eq!( + TraceId::from_hex(&hex.to_uppercase()), + Some(TraceId([0xff; 16])) + ); + } + + #[test] + fn test_from_hex_rejects_invalid() { + assert_eq!(SpanId::from_hex(""), None); + assert_eq!(SpanId::from_hex("0123456789abcde"), None); // too short + assert_eq!(SpanId::from_hex("0123456789abcdef00"), None); // too long + assert_eq!(SpanId::from_hex("0123456789abcdeg"), None); // non-hex char + assert_eq!(TraceId::from_hex("0123456789abcdef"), None); // span-length for trace + } + + #[test] + fn test_http_method_from_str() { + assert_eq!("GET".parse::().unwrap(), HttpMethod::Get); + assert_eq!("get".parse::().unwrap(), HttpMethod::Get); + assert_eq!("Post".parse::().unwrap(), HttpMethod::Post); + assert_eq!("DELETE".parse::().unwrap(), HttpMethod::Delete); + assert!("FETCH".parse::().is_err()); + assert!("".parse::().is_err()); + } +} diff --git a/crates/phantom-core/src/view.rs b/crates/phantom-core/src/view.rs new file mode 100644 index 0000000..1f7ace3 --- /dev/null +++ b/crates/phantom-core/src/view.rs @@ -0,0 +1,281 @@ +use std::collections::HashMap; +use std::time::UNIX_EPOCH; + +use serde::Serialize; + +use crate::trace::HttpTrace; + +/// Controls how much of a trace is included when rendering a [`TraceView`]. +/// +/// Defaults include everything: unlimited bodies, no redaction. +#[derive(Debug, Clone, Default)] +pub struct RenderOptions { + /// Maximum body bytes to include; bodies longer than this are truncated + /// at a UTF-8 character boundary and flagged. `None` = unlimited. + pub max_body: Option, + /// Omit request/response bodies entirely (original sizes still reported). + pub headers_only: bool, + /// Header names (lower-cased) whose values are replaced with `"[redacted]"`. + pub redact_headers: Vec, +} + +impl RenderOptions { + /// Headers redacted by default in agent-facing contexts (MCP tools). + pub fn sensitive_headers() -> Vec { + [ + "authorization", + "proxy-authorization", + "cookie", + "set-cookie", + ] + .into_iter() + .map(String::from) + .collect() + } +} + +/// Agent-friendly, fully serializable representation of an [`HttpTrace`]. +/// +/// This is the canonical JSON shape shared by the JSONL output stream, the +/// query CLI, and the MCP server. With default [`RenderOptions`] the emitted +/// fields are a superset of the historical JSONL schema (add-only: +/// `*_body_bytes` and `*_body_truncated`). +#[derive(Debug, Clone, Serialize)] +pub struct TraceView { + /// Unix timestamp of the request in milliseconds. + pub timestamp_ms: u64, + /// Round-trip duration in milliseconds. + pub duration_ms: u64, + /// HTTP method ("GET", "POST", …). + pub method: String, + /// Full request URL. + pub url: String, + /// HTTP response status code. + pub status_code: u16, + /// Request headers (lower-cased keys). + pub request_headers: HashMap, + /// Response headers (lower-cased keys). + pub response_headers: HashMap, + /// Request body decoded as UTF-8 (replacement chars for non-UTF-8 bytes). + #[serde(skip_serializing_if = "Option::is_none")] + pub request_body: Option, + /// Response body decoded as UTF-8 (replacement chars for non-UTF-8 bytes). + #[serde(skip_serializing_if = "Option::is_none")] + pub response_body: Option, + /// Original request body size in bytes (present iff a body existed). + #[serde(skip_serializing_if = "Option::is_none")] + pub request_body_bytes: Option, + /// Original response body size in bytes (present iff a body existed). + #[serde(skip_serializing_if = "Option::is_none")] + pub response_body_bytes: Option, + /// True when `request_body` was truncated by `max_body`. + #[serde(skip_serializing_if = "std::ops::Not::not")] + pub request_body_truncated: bool, + /// True when `response_body` was truncated by `max_body`. + #[serde(skip_serializing_if = "std::ops::Not::not")] + pub response_body_truncated: bool, + /// Source socket address, if available. + #[serde(skip_serializing_if = "Option::is_none")] + pub source_addr: Option, + /// Destination socket address, if available. + #[serde(skip_serializing_if = "Option::is_none")] + pub dest_addr: Option, + /// HTTP protocol version string (e.g. "HTTP/1.1"). + pub protocol_version: String, + /// 128-bit W3C trace ID (hex). + pub trace_id: String, + /// 64-bit span ID (hex). + pub span_id: String, +} + +/// Decodes a body as lossy UTF-8, applying `headers_only`/`max_body` policy. +/// Returns `(rendered_body, original_size, truncated)`. +fn render_body( + body: &Option>, + opts: &RenderOptions, +) -> (Option, Option, bool) { + let Some(bytes) = body.as_ref() else { + return (None, None, false); + }; + let size = Some(bytes.len() as u64); + if opts.headers_only { + return (None, size, false); + } + let mut text = String::from_utf8_lossy(bytes).into_owned(); + let mut truncated = false; + if let Some(max) = opts.max_body + && text.len() > max + { + let mut cut = max; + while !text.is_char_boundary(cut) { + cut -= 1; + } + text.truncate(cut); + truncated = true; + } + (Some(text), size, truncated) +} + +fn render_headers(headers: &HashMap, redact: &[String]) -> HashMap { + headers + .iter() + .map(|(k, v)| { + if redact.iter().any(|r| r.eq_ignore_ascii_case(k)) { + (k.clone(), "[redacted]".to_string()) + } else { + (k.clone(), v.clone()) + } + }) + .collect() +} + +impl TraceView { + pub fn render(trace: &HttpTrace, opts: &RenderOptions) -> Self { + let timestamp_ms = trace + .timestamp + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let (request_body, request_body_bytes, request_body_truncated) = + render_body(&trace.request_body, opts); + let (response_body, response_body_bytes, response_body_truncated) = + render_body(&trace.response_body, opts); + + Self { + timestamp_ms, + duration_ms: trace.duration.as_millis() as u64, + method: trace.method.to_string(), + url: trace.url.clone(), + status_code: trace.status_code, + request_headers: render_headers(&trace.request_headers, &opts.redact_headers), + response_headers: render_headers(&trace.response_headers, &opts.redact_headers), + request_body, + response_body, + request_body_bytes, + response_body_bytes, + request_body_truncated, + response_body_truncated, + source_addr: trace.source_addr.clone(), + dest_addr: trace.dest_addr.clone(), + protocol_version: trace.protocol_version.clone(), + trace_id: trace.trace_id.to_string(), + span_id: trace.span_id.to_string(), + } + } +} + +impl From<&HttpTrace> for TraceView { + fn from(trace: &HttpTrace) -> Self { + Self::render(trace, &RenderOptions::default()) + } +} + +#[cfg(test)] +mod tests { + use std::time::{Duration, SystemTime}; + + use super::*; + use crate::trace::{HttpMethod, SpanId, TraceId}; + + fn make_trace(request_body: Option>, response_body: Option>) -> HttpTrace { + let mut request_headers = HashMap::new(); + request_headers.insert("authorization".to_string(), "Bearer secret".to_string()); + request_headers.insert("accept".to_string(), "application/json".to_string()); + HttpTrace { + span_id: SpanId([1; 8]), + trace_id: TraceId([2; 16]), + parent_span_id: None, + method: HttpMethod::Post, + url: "http://example.com/api".to_string(), + request_headers, + request_body, + status_code: 200, + response_headers: HashMap::new(), + response_body, + timestamp: SystemTime::UNIX_EPOCH + Duration::from_millis(1500), + duration: Duration::from_millis(42), + source_addr: None, + dest_addr: None, + protocol_version: "HTTP/1.1".to_string(), + } + } + + #[test] + fn test_render_default_keeps_full_body() { + let t = make_trace(Some(b"hello".to_vec()), None); + let v = TraceView::render(&t, &RenderOptions::default()); + assert_eq!(v.request_body.as_deref(), Some("hello")); + assert_eq!(v.request_body_bytes, Some(5)); + assert!(!v.request_body_truncated); + assert_eq!(v.response_body, None); + assert_eq!(v.response_body_bytes, None); + assert_eq!(v.timestamp_ms, 1500); + assert_eq!(v.duration_ms, 42); + assert_eq!(v.span_id, "0101010101010101"); + } + + #[test] + fn test_render_max_body_truncates() { + let t = make_trace(None, Some(b"0123456789".to_vec())); + let opts = RenderOptions { + max_body: Some(4), + ..Default::default() + }; + let v = TraceView::render(&t, &opts); + assert_eq!(v.response_body.as_deref(), Some("0123")); + assert_eq!(v.response_body_bytes, Some(10)); + assert!(v.response_body_truncated); + } + + #[test] + fn test_render_truncation_respects_char_boundary() { + // "あ" is 3 bytes in UTF-8; cutting at byte 4 must back up to byte 3. + let t = make_trace(Some("ああ".as_bytes().to_vec()), None); + let opts = RenderOptions { + max_body: Some(4), + ..Default::default() + }; + let v = TraceView::render(&t, &opts); + assert_eq!(v.request_body.as_deref(), Some("あ")); + assert_eq!(v.request_body_bytes, Some(6)); + assert!(v.request_body_truncated); + } + + #[test] + fn test_render_headers_only_omits_bodies_reports_sizes() { + let t = make_trace(Some(b"req".to_vec()), Some(b"resp".to_vec())); + let opts = RenderOptions { + headers_only: true, + ..Default::default() + }; + let v = TraceView::render(&t, &opts); + assert_eq!(v.request_body, None); + assert_eq!(v.response_body, None); + assert_eq!(v.request_body_bytes, Some(3)); + assert_eq!(v.response_body_bytes, Some(4)); + assert!(!v.request_body_truncated); + } + + #[test] + fn test_render_redacts_headers() { + let t = make_trace(None, None); + let opts = RenderOptions { + redact_headers: RenderOptions::sensitive_headers(), + ..Default::default() + }; + let v = TraceView::render(&t, &opts); + assert_eq!(v.request_headers["authorization"], "[redacted]"); + assert_eq!(v.request_headers["accept"], "application/json"); + } + + #[test] + fn test_serialized_shape_skips_absent_fields() { + let t = make_trace(None, None); + let json = serde_json::to_value(TraceView::from(&t)).unwrap(); + let obj = json.as_object().unwrap(); + assert!(!obj.contains_key("request_body")); + assert!(!obj.contains_key("request_body_bytes")); + assert!(!obj.contains_key("request_body_truncated")); + assert_eq!(obj["method"], "POST"); + } +} diff --git a/crates/phantom-storage/src/fjall_store.rs b/crates/phantom-storage/src/fjall_store.rs index 1a57264..ae97ad8 100644 --- a/crates/phantom-storage/src/fjall_store.rs +++ b/crates/phantom-storage/src/fjall_store.rs @@ -2,6 +2,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use fjall::{Config, Keyspace, PartitionCreateOptions, PartitionHandle}; use phantom_core::error::StorageError; +use phantom_core::query::TraceQuery; use phantom_core::storage::TraceStore; use phantom_core::trace::{HttpTrace, SpanId, TraceId}; @@ -157,6 +158,72 @@ impl TraceStore for FjallTraceStore { fn count(&self) -> Result { Ok(self.traces.approximate_len() as u64) } + + fn query(&self, query: &TraceQuery) -> Result, StorageError> { + const DEFAULT_LIMIT: usize = 100; + let limit = if query.limit == 0 { + DEFAULT_LIMIT + } else { + query.limit + }; + + // With a trace_id filter, the by_trace_id prefix scan is far narrower + // than a time scan; otherwise scan by_time (newest first), bounded by + // the since/until key range when given. + let index_entries: Box>> = + if let Some(trace_id) = &query.trace_id { + Box::new(self.by_trace_id.prefix(trace_id.as_bytes())) + } else { + let start: [u8; 16] = query + .since + .map(|ts| time_key(&ts, &SpanId([0x00; 8]))) + .unwrap_or([0x00; 16]); + let end: [u8; 16] = query + .until + .map(|ts| time_key(&ts, &SpanId([0xff; 8]))) + .unwrap_or([0xff; 16]); + Box::new(self.by_time.range(start..=end).rev()) + }; + + let mut skipped = 0; + let mut results = Vec::new(); + for entry in index_entries { + if results.len() >= limit { + break; + } + let (_key, value) = entry.map_err(|e| StorageError::Read(e.to_string()))?; + let span_id_bytes: [u8; 8] = value[..8] + .try_into() + .map_err(|_| StorageError::Read("invalid span_id in index".into()))?; + if let Some(trace) = self.get_by_span_id(&SpanId(span_id_bytes))? + && query.matches(&trace) + { + if skipped < query.offset { + skipped += 1; + continue; + } + results.push(trace); + } + } + Ok(results) + } + + fn clear(&self) -> Result<(), StorageError> { + for partition in [&self.traces, &self.by_time, &self.by_trace_id] { + let keys: Vec<_> = partition + .keys() + .collect::>() + .map_err(|e| StorageError::Read(e.to_string()))?; + let mut batch = self.keyspace.batch(); + for key in keys { + batch.remove(partition, key); + } + batch + .commit() + .map_err(|e| StorageError::Write(e.to_string()))?; + } + Ok(()) + } } #[cfg(test)] @@ -265,4 +332,141 @@ mod tests { let results = store.search_by_url("/api/", 10).unwrap(); assert_eq!(results.len(), 2); } + + use phantom_core::trace::HttpMethod; + + fn make_trace_at(url: &str, status: u16, ts_secs: u64) -> HttpTrace { + let mut t = make_trace(url, status); + t.timestamp = std::time::UNIX_EPOCH + Duration::from_secs(ts_secs); + t + } + + #[test] + fn test_query_by_method() { + let dir = tempfile::tempdir().unwrap(); + let store = FjallTraceStore::open(dir.path()).unwrap(); + + let mut post = make_trace("http://a/create", 201); + post.method = HttpMethod::Post; + store.insert(&post).unwrap(); + store.insert(&make_trace("http://a/read", 200)).unwrap(); + + let results = store + .query(&TraceQuery { + methods: vec![HttpMethod::Post], + ..Default::default() + }) + .unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].url, "http://a/create"); + } + + #[test] + fn test_query_by_status_range() { + let dir = tempfile::tempdir().unwrap(); + let store = FjallTraceStore::open(dir.path()).unwrap(); + + store.insert(&make_trace("http://a/ok", 200)).unwrap(); + store.insert(&make_trace("http://a/missing", 404)).unwrap(); + store.insert(&make_trace("http://a/boom", 500)).unwrap(); + + let results = store + .query(&TraceQuery { + status: Some("4xx".parse().unwrap()), + ..Default::default() + }) + .unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].status_code, 404); + } + + #[test] + fn test_query_time_range_bounds_scan() { + let dir = tempfile::tempdir().unwrap(); + let store = FjallTraceStore::open(dir.path()).unwrap(); + + for ts in [100, 200, 300, 400] { + store + .insert(&make_trace_at(&format!("http://a/{ts}"), 200, ts)) + .unwrap(); + } + + // Inclusive on both ends; newest first. + let results = store + .query(&TraceQuery { + since: Some(std::time::UNIX_EPOCH + Duration::from_secs(200)), + until: Some(std::time::UNIX_EPOCH + Duration::from_secs(300)), + ..Default::default() + }) + .unwrap(); + let urls: Vec<_> = results.iter().map(|t| t.url.as_str()).collect(); + assert_eq!(urls, ["http://a/300", "http://a/200"]); + } + + #[test] + fn test_query_offset_applied_after_filter() { + let dir = tempfile::tempdir().unwrap(); + let store = FjallTraceStore::open(dir.path()).unwrap(); + + // Interleave matching (404) and non-matching (200) traces. + for i in 0..6u64 { + let status = if i % 2 == 0 { 404 } else { 200 }; + store + .insert(&make_trace_at(&format!("http://a/{i}"), status, 100 + i)) + .unwrap(); + } + + // Matching, newest first: /4, /2, /0. Offset 1 + limit 1 → /2. + let results = store + .query(&TraceQuery { + status: Some("404".parse().unwrap()), + offset: 1, + limit: 1, + ..Default::default() + }) + .unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].url, "http://a/2"); + } + + #[test] + fn test_query_by_trace_id_with_filters() { + let dir = tempfile::tempdir().unwrap(); + let store = FjallTraceStore::open(dir.path()).unwrap(); + + let shared = TraceId(rand_bytes_16()); + for status in [200, 404] { + let mut t = make_trace(&format!("http://a/{status}"), status); + t.trace_id = shared.clone(); + store.insert(&t).unwrap(); + } + store.insert(&make_trace("http://other/404", 404)).unwrap(); + + let results = store + .query(&TraceQuery { + trace_id: Some(shared), + status: Some("404".parse().unwrap()), + ..Default::default() + }) + .unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].url, "http://a/404"); + } + + #[test] + fn test_clear() { + let dir = tempfile::tempdir().unwrap(); + let store = FjallTraceStore::open(dir.path()).unwrap(); + + let trace = make_trace("http://a/x", 200); + let span_id = trace.span_id.clone(); + store.insert(&trace).unwrap(); + store.insert(&make_trace("http://a/y", 200)).unwrap(); + + store.clear().unwrap(); + + assert!(store.get_by_span_id(&span_id).unwrap().is_none()); + assert!(store.list_recent(10, 0).unwrap().is_empty()); + assert!(store.query(&TraceQuery::default()).unwrap().is_empty()); + } } From 564c947c5e713a76d26e4d85028d0c3b1c6d220f Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 12 Jul 2026 02:46:04 +0000 Subject: [PATCH 2/5] feat(cli)!: restructure into subcommands with query support and exit-code propagation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Breaking: the flat CLI is replaced by subcommands. phantom run -- capture (old default behavior; same flags) phantom list -- filtered listing (--method/--status 4xx/--url/--since 10m/ --until/--trace-id/--limit/--offset, --format jsonl|json|table, --max-body/--headers-only/--redact-header) phantom get -- single trace by span ID (exit 1 when not found) phantom search -- positional URL-substring shorthand for list phantom stats -- store statistics as JSON phantom clear -- delete all traces (requires --yes) phantom mcp -- reserved (stub, implemented next) Agent/scripting fixes: - Propagate the traced child's exit status (was discarded — phantom always exited 0); Unix signal deaths map to 128+signal. - tracing output now goes to stderr (was stdout, could corrupt JSONL). - Machine-readable end-of-run summary on stderr in jsonl mode: {"event":"exit","child_exit_code":N,"traces_captured":N} - Global --quiet suppresses status lines; global --data-dir. - run --max-body / --headers-only control JSONL body size. phantom-storage: FjallTraceStore::open now takes an advisory flock on phantom.lock — fjall itself does not lock across processes, so two phantom instances (e.g. a query during a capture) could previously open the same keyspace concurrently and risk corruption. Query commands surface a lock hint pointing at the running process. main.rs split into cli.rs, runner.rs, commands/{run,query}.rs. Existing integration tests updated to the run subcommand; new tests/cli_query_integration.rs covers exit codes, JSONL purity, query filters, truncation, and the lock UX. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01BjmmAVFHuaU4Fb46QEd6ky --- Cargo.lock | 7 + Cargo.toml | 2 + crates/phantom-storage/src/fjall_store.rs | 34 + src/cli.rs | 359 ++++++++++ src/commands/mod.rs | 2 + src/commands/query.rs | 154 +++++ src/commands/run.rs | 251 +++++++ src/main.rs | 766 +++------------------- src/runner.rs | 200 ++++++ tests/cli_query_integration.rs | 293 +++++++++ tests/fault_injection.rs | 4 + tests/integration/lib.sh | 2 +- tests/proxy_java_clients_integration.rs | 1 + tests/proxy_node_integration.rs | 2 + tests/proxy_php_integration.rs | 1 + tests/proxy_php_ldpreload_integration.rs | 9 +- 16 files changed, 1406 insertions(+), 681 deletions(-) create mode 100644 src/cli.rs create mode 100644 src/commands/mod.rs create mode 100644 src/commands/query.rs create mode 100644 src/commands/run.rs create mode 100644 src/runner.rs create mode 100644 tests/cli_query_integration.rs diff --git a/Cargo.lock b/Cargo.lock index 996bcca..5017392 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -982,6 +982,12 @@ dependencies = [ "tracing", ] +[[package]] +name = "humantime" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15" + [[package]] name = "hyper" version = "1.8.1" @@ -1608,6 +1614,7 @@ dependencies = [ "cargo-husky", "clap", "dirs", + "humantime", "phantom-capture", "phantom-core", "phantom-storage", diff --git a/Cargo.toml b/Cargo.toml index 72526db..cdb0179 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,10 +43,12 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } clap = { version = "4", features = ["derive"] } dirs = "6" anyhow = "1" +humantime = "2" [dev-dependencies] serde_json = { workspace = true } serde = { workspace = true } +phantom-storage = { workspace = true } tempfile = "3" rustls = "0.22" rcgen = "0.13" diff --git a/crates/phantom-storage/src/fjall_store.rs b/crates/phantom-storage/src/fjall_store.rs index ae97ad8..149c23c 100644 --- a/crates/phantom-storage/src/fjall_store.rs +++ b/crates/phantom-storage/src/fjall_store.rs @@ -11,10 +11,30 @@ pub struct FjallTraceStore { traces: PartitionHandle, by_time: PartitionHandle, by_trace_id: PartitionHandle, + /// Advisory exclusive lock on the data directory, released on drop. + /// fjall itself does not lock across processes, and two writers on one + /// keyspace would corrupt it — so we enforce single-process access here. + _lock: std::fs::File, } impl FjallTraceStore { pub fn open(path: impl AsRef) -> Result { + let path = path.as_ref(); + std::fs::create_dir_all(path).map_err(|e| StorageError::Open(e.to_string()))?; + let lock_path = path.join("phantom.lock"); + let lock = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&lock_path) + .map_err(|e| StorageError::Open(format!("{}: {e}", lock_path.display())))?; + lock.try_lock().map_err(|_| { + StorageError::Open(format!( + "data dir {} holds an active store lock (another phantom process is using it)", + path.display() + )) + })?; + let keyspace = Config::new(path) .open() .map_err(|e| StorageError::Open(e.to_string()))?; @@ -39,6 +59,7 @@ impl FjallTraceStore { traces, by_time, by_trace_id, + _lock: lock, }) } } @@ -453,6 +474,19 @@ mod tests { assert_eq!(results[0].url, "http://a/404"); } + #[test] + fn test_open_is_exclusive() { + let dir = tempfile::tempdir().unwrap(); + let store = FjallTraceStore::open(dir.path()).unwrap(); + + let second = FjallTraceStore::open(dir.path()); + assert!(matches!(second, Err(StorageError::Open(_)))); + + // Lock is released when the store is dropped. + drop(store); + assert!(FjallTraceStore::open(dir.path()).is_ok()); + } + #[test] fn test_clear() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/cli.rs b/src/cli.rs new file mode 100644 index 0000000..6455e88 --- /dev/null +++ b/src/cli.rs @@ -0,0 +1,359 @@ +use std::path::PathBuf; + +use clap::{Args, Parser, Subcommand, ValueEnum}; +use phantom_core::query::StatusRange; +use phantom_core::trace::HttpMethod; + +#[derive(Debug, Clone, ValueEnum)] +pub enum Backend { + /// MITM proxy — captures HTTP + HTTPS, cross-platform. Node.js HTTPS injected automatically. + Proxy, + /// LD_PRELOAD agent — captures HTTP + HTTPS, Linux only. No proxy config needed. + #[cfg(target_os = "linux")] + Ldpreload, +} + +#[derive(Debug, Clone, Default, ValueEnum)] +pub enum OutputMode { + /// Interactive terminal UI with trace list and detail view. + #[default] + Tui, + /// Stream traces as JSON Lines to stdout; auto-exits when child process finishes. + Jsonl, +} + +/// Output format for query subcommands (`list`, `search`, `get`). +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, ValueEnum)] +pub enum QueryFormat { + /// One compact JSON object per line (pipe to jq). + #[default] + Jsonl, + /// Pretty-printed JSON (array for list/search, object for get). + Json, + /// Human-readable fixed-width columns. + Table, +} + +#[derive(Parser)] +#[command( + name = "phantom", + about = "Zero-instrumentation HTTP/HTTPS API observability tool", + long_about = "phantom — Zero-instrumentation HTTP/HTTPS API observability\n\ +\n\ +Captures every HTTP and HTTPS request/response made by a target process\n\ +and displays them in an interactive TUI, streams them as JSON Lines, or\n\ +serves them to AI coding agents over MCP.\n\ +\n\ +Typical workflows:\n\ +\n\ + # Capture traffic from a command (TUI):\n\ + phantom run -- node app.js\n\ +\n\ + # Capture and stream JSONL for scripting / AI analysis:\n\ + phantom run --output jsonl -- node app.js\n\ +\n\ + # Query previously captured traces:\n\ + phantom list --status 5xx --since 10m\n\ + phantom get \n\ +\n\ + # Run as an MCP server (capture control + queries over stdio):\n\ + phantom mcp\n\ +\n\ +Note: the trace store is locked by a single phantom process at a time.\n\ +Query subcommands work while no capture is running; while `phantom run`\n\ +or `phantom mcp` is active, query through the MCP server instead.", + version +)] +pub struct Cli { + /// Directory where captured traces are persisted (Fjall key-value store). + /// Defaults to the platform data directory, e.g. ~/.local/share/phantom/data. + #[arg(short, long, global = true)] + pub data_dir: Option, + + /// Suppress status messages on stderr (machine-friendly output only). + #[arg(short, long, global = true)] + pub quiet: bool, + + #[command(subcommand)] + pub command: Commands, +} + +#[derive(Subcommand)] +pub enum Commands { + /// Capture HTTP(S) traffic, optionally spawning a command to trace. + Run(RunArgs), + /// List captured traces (newest first) with filters. + List(ListArgs), + /// Show a single trace by span ID. + Get(GetArgs), + /// Shorthand for `list --url `. + Search(SearchArgs), + /// Print trace store statistics as JSON. + Stats, + /// Delete all captured traces. + Clear(ClearArgs), + /// Run as an MCP (Model Context Protocol) server over stdio. + /// + /// Exposes capture control and trace queries as MCP tools for AI coding + /// agents. Register with e.g.: claude mcp add phantom -- phantom mcp + Mcp, +} + +#[derive(Args)] +#[command( + long_about = "Capture HTTP(S) traffic, optionally spawning a command to trace.\n\ +\n\ +━━━ CAPTURE BACKENDS ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n\ +\n\ + proxy (default, cross-platform)\n\ + Starts a MITM proxy on 127.0.0.1:. Intercepts HTTP and HTTPS.\n\ +\n\ + • Node.js (`phantom run -- node app.js`)\n\ + proxy-preload.js is injected automatically via --require. Both http://\n\ + and https:// are captured with zero application changes.\n\ +\n\ + • PHP (`phantom run -- php app.php`)\n\ + The MITM CA certificate is injected automatically via -d curl.cainfo=.\n\ + curl-based HTTP and HTTPS (incl. Guzzle's default handler) are captured\n\ + with zero application changes. Requires PHP >= 5.3.7 for curl.cainfo;\n\ + only the curl extension is covered (not PHP streams).\n\ +\n\ + • Other commands (`phantom run -- curl http://api.example.com/v1`)\n\ + HTTP_PROXY / HTTPS_PROXY (and lowercase variants) are set automatically.\n\ + Plain HTTP is captured; HTTPS is captured if the application honours\n\ + these env vars for CONNECT tunnelling (as libcurl does by default).\n\ +\n\ + • Manual (start phantom alone, then configure your app)\n\ + Set HTTP_PROXY=http://127.0.0.1:8080 in the target process yourself.\n\ +\n\ + ldpreload (Linux only)\n\ + Injects libphantom_agent.so via LD_PRELOAD. Hooks send/recv/close at\n\ + the libc level for plain HTTP, and OpenSSL SSL_write/SSL_read for HTTPS\n\ + (captured above the TLS layer, before encryption). No proxy config\n\ + required and no MITM certificate involved — works for any dynamically\n\ + linked process, language-agnostic (e.g. PHP's curl extension).\n\ +\n\ +━━━ OUTPUT MODES ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n\ +\n\ + tui (default) — Interactive terminal UI with trace list + detail view.\n\ +\n\ + jsonl — One JSON object per line on stdout. phantom exits automatically\n\ + when the child process exits, propagating its exit code (ideal\n\ + for scripting and AI agents).\n\ +\n\ + JSONL record schema (all fields always present unless marked optional):\n\ + trace_id string W3C-compatible 128-bit trace ID (hex, 32 chars)\n\ + span_id string 64-bit span ID (hex, 16 chars)\n\ + timestamp_ms number Unix epoch milliseconds — request start time\n\ + duration_ms number Round-trip latency in milliseconds\n\ + method string HTTP verb: \"GET\", \"POST\", \"PUT\", \"DELETE\", …\n\ + url string Full request URL (scheme + host + path + query)\n\ + status_code number HTTP response status code (200, 404, 500, …)\n\ + protocol_version string HTTP version string, e.g. \"HTTP/1.1\"\n\ + request_headers object Lower-cased header names → values\n\ + response_headers object Lower-cased header names → values\n\ + request_body string? UTF-8 decoded body; omitted when empty\n\ + response_body string? UTF-8 decoded body; omitted when empty\n\ + request_body_bytes number? Original body size; present when a body existed\n\ + response_body_bytes number? Original body size; present when a body existed\n\ + request_body_truncated bool? Present (true) when --max-body truncated the body\n\ + response_body_truncated bool? Present (true) when --max-body truncated the body\n\ + source_addr string? Client socket address, e.g. \"127.0.0.1:54321\"\n\ + dest_addr string? Server socket address, e.g. \"93.184.216.34:443\"", + after_long_help = "EXAMPLES\n\ +\n\ + # Trace a Node.js app — HTTP + HTTPS captured, zero app changes:\n\ + phantom run -- node app.js\n\ +\n\ + # Stream traces as JSONL for scripting / AI analysis:\n\ + phantom run --output jsonl -- node app.js\n\ +\n\ + # Truncate bodies to keep output small:\n\ + phantom run --output jsonl --max-body 256 -- node app.js\n\ +\n\ + # Filter errors with jq:\n\ + phantom run --output jsonl -- node app.js | jq 'select(.status_code >= 400)'\n\ +\n\ + # LD_PRELOAD mode (Linux only):\n\ + cargo build -p phantom-agent\n\ + phantom run --backend ldpreload \\\n\ + --agent-lib ./target/debug/libphantom_agent.so \\\n\ + -- curl http://api.example.com/v1/users" +)] +pub struct RunArgs { + /// Capture backend: 'proxy' (MITM, cross-platform) or 'ldpreload' (Linux, HTTP + HTTPS). + #[arg(short, long, value_enum, default_value = "proxy")] + pub backend: Backend, + + /// Output mode: 'tui' opens the interactive UI; 'jsonl' streams one trace + /// per line to stdout and exits with the child's exit code when it finishes. + #[arg(short, long, value_enum, default_value = "tui")] + pub output: OutputMode, + + /// TCP port the proxy listens on. + #[arg(short, long, default_value = "8080")] + pub port: u16, + + /// Disable TLS certificate verification for connections to backend servers. + /// Use when tracing apps that talk to servers with self-signed certificates. + #[arg(long, default_value = "false")] + pub insecure: bool, + + /// Path to libphantom_agent.so [required for --backend ldpreload] + /// + /// Build with: cargo build -p phantom-agent + /// Then pass: --agent-lib ./target/debug/libphantom_agent.so + #[arg(long, value_name = "PATH")] + pub agent_lib: Option, + + /// Inject faults into proxied requests (proxy backend only). + /// + /// SPEC formats: + /// delay:100ms fixed 100 ms delay on all requests + /// delay:100ms-500ms random delay in the given range + /// delay:200ms:/api delay only URLs containing "/api" + /// error:503 return HTTP 503 for all requests + /// error:503:0.5 return HTTP 503 with 50% probability + /// error:500:0.1:/api 10% chance of HTTP 500 on URLs containing "/api" + /// + /// Rules are applied in order; delays and errors can be combined. + /// Repeat the flag to add multiple rules: + /// --fault delay:50ms --fault error:500:0.1 + #[arg(long, value_name = "SPEC")] + pub fault: Vec, + + /// Truncate request/response bodies to N bytes in JSONL output + /// (0 = unlimited). Truncated records carry `*_body_truncated: true` + /// and the original size in `*_body_bytes`. + #[arg(long, value_name = "N", default_value = "0")] + pub max_body: usize, + + /// Omit request/response bodies from JSONL output entirely + /// (original sizes still reported in `*_body_bytes`). + #[arg(long)] + pub headers_only: bool, + + /// Command to spawn and trace (everything after `--`). + /// + /// proxy mode: HTTP_PROXY is set automatically; Node.js additionally + /// gets proxy-preload.js injected via --require (captures HTTPS too). + /// ldpreload mode: LD_PRELOAD + PHANTOM_SOCKET are set automatically. + #[arg(last = true, value_name = "CMD")] + pub command: Vec, +} + +/// Filter and rendering flags shared by `list` and `search`. +#[derive(Args)] +pub struct FilterArgs { + /// Only these HTTP methods (repeatable): --method GET --method POST + #[arg(long = "method", value_name = "METHOD")] + pub methods: Vec, + + /// Status code filter: exact ("404"), class ("4xx"), or range ("400-499"). + #[arg(long, value_name = "RANGE")] + pub status: Option, + + /// Only traces newer than this: RFC3339 ("2026-07-12T10:00:00Z") + /// or a relative duration ago ("30s", "10m", "2h"). + #[arg(long, value_name = "TIME")] + pub since: Option, + + /// Only traces older than this: RFC3339 or a relative duration ago. + #[arg(long, value_name = "TIME")] + pub until: Option, + + /// Only spans belonging to this 32-char hex trace ID. + #[arg(long, value_name = "HEX32")] + pub trace_id: Option, + + /// Maximum number of traces to return. + #[arg(long, default_value = "50")] + pub limit: usize, + + /// Number of matching traces to skip (for pagination). + #[arg(long, default_value = "0")] + pub offset: usize, + + /// Output format. + #[arg(long, value_enum, default_value = "jsonl")] + pub format: QueryFormat, + + /// Truncate bodies to N bytes (0 = unlimited). + #[arg(long, value_name = "N", default_value = "1024")] + pub max_body: usize, + + /// Omit bodies entirely (original sizes still reported). + #[arg(long)] + pub headers_only: bool, + + /// Replace this header's value with "[redacted]" (repeatable). + #[arg(long = "redact-header", value_name = "NAME")] + pub redact_headers: Vec, +} + +#[derive(Args)] +#[command(after_long_help = "EXAMPLES\n\ +\n\ + # Recent failures, compact:\n\ + phantom list --status 5xx --limit 10 --format table\n\ +\n\ + # POSTs to the users API in the last 10 minutes, as JSONL for jq:\n\ + phantom list --method POST --url /api/users --since 10m | jq .url\n\ +\n\ + # Everything from one distributed trace:\n\ + phantom list --trace-id 0123456789abcdef0123456789abcdef")] +pub struct ListArgs { + /// Only URLs containing this substring (case-insensitive). + #[arg(long, value_name = "SUBSTR")] + pub url: Option, + + #[command(flatten)] + pub filter: FilterArgs, +} + +#[derive(Args)] +pub struct SearchArgs { + /// URL substring to search for (case-insensitive). + pub pattern: String, + + #[command(flatten)] + pub filter: FilterArgs, +} + +#[derive(Args)] +pub struct GetArgs { + /// 16-character hex span ID (as shown by `phantom list`). + pub span_id: String, + + /// Output format ('json' is pretty-printed; 'jsonl' is one compact line). + #[arg(long, value_enum, default_value = "json")] + pub format: QueryFormat, + + /// Truncate bodies to N bytes (0 = unlimited). + #[arg(long, value_name = "N", default_value = "0")] + pub max_body: usize, + + /// Omit bodies entirely (original sizes still reported). + #[arg(long)] + pub headers_only: bool, +} + +#[derive(Args)] +pub struct ClearArgs { + /// Confirm deletion (required; refuses to run without it). + #[arg(long)] + pub yes: bool, +} + +/// Resolved global flags passed to command handlers. +pub struct GlobalOpts { + pub quiet: bool, + pub data_dir: PathBuf, +} + +pub fn default_data_dir() -> PathBuf { + dirs::data_local_dir() + .unwrap_or_else(|| PathBuf::from(".")) + .join("phantom") + .join("data") +} diff --git a/src/commands/mod.rs b/src/commands/mod.rs new file mode 100644 index 0000000..d4c26e9 --- /dev/null +++ b/src/commands/mod.rs @@ -0,0 +1,2 @@ +pub mod query; +pub mod run; diff --git a/src/commands/query.rs b/src/commands/query.rs new file mode 100644 index 0000000..878fa38 --- /dev/null +++ b/src/commands/query.rs @@ -0,0 +1,154 @@ +use std::time::SystemTime; + +use phantom_core::query::TraceQuery; +use phantom_core::storage::TraceStore; +use phantom_core::trace::{HttpTrace, SpanId, TraceId}; +use phantom_core::view::{RenderOptions, TraceView}; + +use crate::cli::{FilterArgs, GetArgs, ListArgs, QueryFormat, SearchArgs}; + +/// Parse a `--since`/`--until` value: RFC3339 timestamp, or a relative +/// duration meaning "that long ago" (e.g. "30s", "10m", "2h"). +fn parse_time(s: &str) -> anyhow::Result { + if let Ok(duration) = humantime::parse_duration(s) { + return SystemTime::now() + .checked_sub(duration) + .ok_or_else(|| anyhow::anyhow!("relative time {s:?} is before the epoch")); + } + humantime::parse_rfc3339(s).map_err(|e| { + anyhow::anyhow!( + "invalid time {s:?}: {e} (expected RFC3339 like \"2026-07-12T10:00:00Z\" \ + or a relative duration like \"10m\")" + ) + }) +} + +fn build_query(url: Option, filter: &FilterArgs) -> anyhow::Result { + let trace_id = filter + .trace_id + .as_deref() + .map(|s| { + TraceId::from_hex(s) + .ok_or_else(|| anyhow::anyhow!("invalid trace ID {s:?}: expected 32 hex chars")) + }) + .transpose()?; + + Ok(TraceQuery { + methods: filter.methods.clone(), + status: filter.status, + url_contains: url, + since: filter.since.as_deref().map(parse_time).transpose()?, + until: filter.until.as_deref().map(parse_time).transpose()?, + trace_id, + limit: filter.limit, + offset: filter.offset, + }) +} + +fn render_options(max_body: usize, headers_only: bool, redact: &[String]) -> RenderOptions { + RenderOptions { + max_body: (max_body > 0).then_some(max_body), + headers_only, + redact_headers: redact.iter().map(|h| h.to_lowercase()).collect(), + } +} + +fn print_table(views: &[TraceView]) { + println!( + "{:<24} {:<7} {:>6} {:>8} {:<17} URL", + "TIME", "METHOD", "STATUS", "DURATION", "SPAN_ID" + ); + for v in views { + let time = humantime::format_rfc3339_millis( + std::time::UNIX_EPOCH + std::time::Duration::from_millis(v.timestamp_ms), + ); + println!( + "{:<24} {:<7} {:>6} {:>6}ms {:<17} {}", + time, v.method, v.status_code, v.duration_ms, v.span_id, v.url + ); + } +} + +fn output_traces( + traces: &[HttpTrace], + format: QueryFormat, + opts: &RenderOptions, +) -> anyhow::Result<()> { + let views: Vec = traces.iter().map(|t| TraceView::render(t, opts)).collect(); + match format { + QueryFormat::Jsonl => { + for view in &views { + println!("{}", serde_json::to_string(view)?); + } + } + QueryFormat::Json => println!("{}", serde_json::to_string_pretty(&views)?), + QueryFormat::Table => print_table(&views), + } + Ok(()) +} + +pub fn list(store: &dyn TraceStore, args: ListArgs) -> anyhow::Result<()> { + let query = build_query(args.url, &args.filter)?; + let traces = store.query(&query)?; + let opts = render_options( + args.filter.max_body, + args.filter.headers_only, + &args.filter.redact_headers, + ); + output_traces(&traces, args.filter.format, &opts) +} + +pub fn search(store: &dyn TraceStore, args: SearchArgs) -> anyhow::Result<()> { + let query = build_query(Some(args.pattern), &args.filter)?; + let traces = store.query(&query)?; + let opts = render_options( + args.filter.max_body, + args.filter.headers_only, + &args.filter.redact_headers, + ); + output_traces(&traces, args.filter.format, &opts) +} + +/// Returns `false` (exit code 1) when the span ID is unknown. +pub fn get(store: &dyn TraceStore, args: GetArgs) -> anyhow::Result { + let span_id = SpanId::from_hex(&args.span_id).ok_or_else(|| { + anyhow::anyhow!("invalid span ID {:?}: expected 16 hex chars", args.span_id) + })?; + + let Some(trace) = store.get_by_span_id(&span_id)? else { + eprintln!("phantom: no trace found for span ID {}", args.span_id); + return Ok(false); + }; + + let opts = render_options(args.max_body, args.headers_only, &[]); + let view = TraceView::render(&trace, &opts); + match args.format { + QueryFormat::Jsonl => println!("{}", serde_json::to_string(&view)?), + _ => println!("{}", serde_json::to_string_pretty(&view)?), + } + Ok(true) +} + +pub fn stats(store: &dyn TraceStore, data_dir: &std::path::Path) -> anyhow::Result<()> { + println!( + "{}", + serde_json::json!({ + "total_traces": store.count()?, + "data_dir": data_dir.display().to_string(), + }) + ); + Ok(()) +} + +/// Returns `false` (exit code 1) when `--yes` was not passed. +pub fn clear(store: &dyn TraceStore, yes: bool, quiet: bool) -> anyhow::Result { + if !yes { + eprintln!("phantom: refusing to delete all traces without --yes"); + return Ok(false); + } + store.clear()?; + if !quiet { + eprintln!("phantom: all traces cleared"); + } + Ok(true) +} diff --git a/src/commands/run.rs b/src/commands/run.rs new file mode 100644 index 0000000..57c6f74 --- /dev/null +++ b/src/commands/run.rs @@ -0,0 +1,251 @@ +use std::process::ExitStatus; +use std::sync::Arc; + +use phantom_capture::ProxyCaptureBackend; +use phantom_core::capture::CaptureBackend; +use phantom_core::storage::TraceStore; +use phantom_core::trace::HttpTrace; +use phantom_core::view::{RenderOptions, TraceView}; +use phantom_storage::FjallTraceStore; + +use crate::cli::{GlobalOpts, OutputMode, RunArgs}; +use crate::runner::{TempScript, build_fault_config, spawn_proxy_child, wait_for_proxy}; + +/// Render options for the JSONL stream, from `run` flags. +fn jsonl_render_options(args: &RunArgs) -> RenderOptions { + RenderOptions { + max_body: (args.max_body > 0).then_some(args.max_body), + headers_only: args.headers_only, + redact_headers: Vec::new(), + } +} + +/// Runs the JSONL output loop: each captured trace is serialized and written to +/// stdout as a single JSON object followed by a newline. +/// +/// Exits when: +/// - The trace channel is closed (sender dropped), +/// - Ctrl-C is received, or +/// - The optional `child` process exits. +/// +/// Returns the child's exit status (when a child was spawned and exited) so +/// the caller can propagate its exit code. +async fn run_jsonl_output( + store: Arc, + mut trace_rx: tokio::sync::mpsc::Receiver, + child: Option, + opts: &RenderOptions, + quiet: bool, +) -> anyhow::Result> { + // Spawn a background thread to wait() on the child so we don't block the + // async executor. The child's exit status is sent through a oneshot. + let mut child_done: Option>> = + if let Some(mut c) = child { + let (tx, rx) = tokio::sync::oneshot::channel(); + std::thread::spawn(move || { + let _ = tx.send(c.wait()); + }); + Some(rx) + } else { + None + }; + + let ctrl_c = tokio::signal::ctrl_c(); + tokio::pin!(ctrl_c); + + let mut traces_captured: u64 = 0; + let mut child_status: Option = None; + + let mut emit = |t: &HttpTrace| -> anyhow::Result<()> { + store.insert(t).ok(); + println!("{}", serde_json::to_string(&TraceView::render(t, opts))?); + traces_captured += 1; + Ok(()) + }; + + loop { + tokio::select! { + maybe_trace = trace_rx.recv() => { + match maybe_trace { + Some(t) => emit(&t)?, + None => break, + } + } + _ = &mut ctrl_c => break, + // When the child exits, wait briefly for the backend to flush any + // in-flight datagrams, then drain whatever arrived. + status = async { + if let Some(rx) = child_done.as_mut() { + rx.await + } else { + std::future::pending().await + } + } => { + if let Ok(Ok(status)) = status { + child_status = Some(status); + } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + while let Ok(t) = trace_rx.try_recv() { + emit(&t)?; + } + break; + } + } + } + + if !quiet { + // Machine-readable end-of-run summary on stderr (stdout stays pure JSONL). + eprintln!( + "{}", + serde_json::json!({ + "event": "exit", + "child_exit_code": child_status.and_then(|s| s.code()), + "traces_captured": traces_captured, + }) + ); + } + + Ok(child_status) +} + +pub async fn run_proxy( + globals: &GlobalOpts, + args: RunArgs, + store: Arc, +) -> anyhow::Result> { + let fault_config = build_fault_config(&args.fault)?; + let mut backend = ProxyCaptureBackend::new(args.port, args.insecure).with_faults(fault_config); + let backend_name = backend.name().to_string(); + let trace_rx = backend.start().map_err(|e| anyhow::anyhow!("{e}"))?; + + // Optionally spawn a child command routed through the proxy. + let child_and_script: Option<(std::process::Child, Option)> = + if !args.command.is_empty() { + // Wait for the proxy to be ready before spawning the child. + wait_for_proxy(args.port).await?; + let ca_cert_pem = backend.ca_cert_pem(); + let (child, ts) = spawn_proxy_child(&args.command, args.port, ca_cert_pem.as_deref())?; + if !globals.quiet { + eprintln!( + "phantom: spawned PID {} → {}", + child.id(), + args.command.join(" ") + ); + } + Some((child, ts)) + } else { + None + }; + + let mut child_status = None; + match args.output { + OutputMode::Tui => { + if !globals.quiet { + if args.command.is_empty() { + eprintln!("phantom: proxy listening on 127.0.0.1:{}", args.port); + eprintln!(" Set your HTTP proxy to http://127.0.0.1:{}", args.port); + eprintln!( + " Example: curl -x http://127.0.0.1:{} http://httpbin.org/get", + args.port + ); + } + eprintln!("phantom: traces stored in {}", globals.data_dir.display()); + } + phantom_tui::run_tui(store, trace_rx, &backend_name).await?; + } + OutputMode::Jsonl => { + if !globals.quiet { + eprintln!( + "phantom: proxy listening on 127.0.0.1:{} [jsonl mode]", + args.port + ); + } + // Split into child and script guard separately so the TempScript + // is NOT dropped until after run_jsonl_output completes (the file + // must exist while node is loading it via --require). + let (child, _script_guard) = match child_and_script { + Some((c, ts)) => (Some(c), ts), + None => (None, None), + }; + let opts = jsonl_render_options(&args); + child_status = run_jsonl_output(store, trace_rx, child, &opts, globals.quiet).await?; + // _script_guard dropped here — temp file deleted after child exits. + } + } + + backend.stop().map_err(|e| anyhow::anyhow!("{e}"))?; + Ok(child_status) +} + +#[cfg(target_os = "linux")] +pub async fn run_ldpreload( + globals: &GlobalOpts, + args: RunArgs, + store: Arc, +) -> anyhow::Result> { + use phantom_capture::LdPreloadCaptureBackend; + + let agent_lib = args.agent_lib.clone().ok_or_else(|| { + anyhow::anyhow!( + "--agent-lib is required for --backend ldpreload\n\ + Example: --agent-lib ./target/debug/libphantom_agent.so" + ) + })?; + + if args.command.is_empty() { + anyhow::bail!( + "A command to trace is required for --backend ldpreload.\n\ + Usage: phantom run --backend ldpreload --agent-lib ./libphantom_agent.so -- curl http://example.com" + ); + } + + // Generate a unique socket path for this run. + let socket_path = std::env::temp_dir().join(format!( + "phantom-{}.sock", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis()) + .unwrap_or(0) + )); + + let mut backend = LdPreloadCaptureBackend::new(socket_path.clone()); + let backend_name = backend.name().to_string(); + let trace_rx = backend.start().map_err(|e| anyhow::anyhow!("{e}"))?; + + if !globals.quiet { + eprintln!("phantom: ldpreload backend active"); + eprintln!(" agent lib : {}", agent_lib.display()); + eprintln!(" socket : {}", socket_path.display()); + eprintln!(" command : {}", args.command.join(" ")); + eprintln!("phantom: traces stored in {}", globals.data_dir.display()); + } + + // Spawn the target process with LD_PRELOAD and PHANTOM_SOCKET set. + let child = std::process::Command::new(&args.command[0]) + .args(&args.command[1..]) + .env("LD_PRELOAD", &agent_lib) + .env("PHANTOM_SOCKET", &socket_path) + .spawn() + .map_err(|e| anyhow::anyhow!("failed to spawn {:?}: {e}", args.command[0]))?; + + if !globals.quiet { + eprintln!("phantom: spawned PID {}", child.id()); + } + + let mut child_status = None; + match args.output { + OutputMode::Tui => { + // In TUI mode the user quits manually; child runs in background. + phantom_tui::run_tui(store, trace_rx, &backend_name).await?; + } + OutputMode::Jsonl => { + // In JSONL mode we exit automatically when the child finishes. + let opts = jsonl_render_options(&args); + child_status = + run_jsonl_output(store, trace_rx, Some(child), &opts, globals.quiet).await?; + } + } + + backend.stop().map_err(|e| anyhow::anyhow!("{e}"))?; + Ok(child_status) +} diff --git a/src/main.rs b/src/main.rs index 1ecaaf9..04d3408 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,708 +1,116 @@ -use std::path::{Path, PathBuf}; +mod cli; +mod commands; +mod runner; + +use std::process::ExitCode; use std::sync::Arc; -use std::time::UNIX_EPOCH; -use clap::{Parser, ValueEnum}; -use phantom_capture::{FaultConfig, ProxyCaptureBackend}; -use phantom_core::capture::CaptureBackend; -use phantom_core::storage::TraceStore; -use phantom_core::trace::HttpTrace; +use clap::Parser; use phantom_storage::FjallTraceStore; -use serde::Serialize; - -// ───────────────────────────────────────────────────────────────────────────── -// Embedded proxy preload script (Node.js transparent injection) -// ───────────────────────────────────────────────────────────────────────────── - -/// The proxy-preload.js content, embedded at compile time. -/// Written to a temp file when tracing Node.js processes via `phantom -- node …`. -const NODE_PROXY_PRELOAD: &str = include_str!("../tests/apps/node-app/proxy-preload.js"); - -/// The Java Agent JAR, embedded at compile time. -/// Written to a temp file when tracing Java processes via `phantom -- java …`. -const JAVA_AGENT_JAR: &[u8] = include_bytes!("../crates/phantom-java-agent/phantom-java-agent.jar"); - -// ───────────────────────────────────────────────────────────────────────────── -// CLI -// ───────────────────────────────────────────────────────────────────────────── - -#[derive(Debug, Clone, ValueEnum)] -enum Backend { - /// MITM proxy — captures HTTP + HTTPS, cross-platform. Node.js HTTPS injected automatically. - Proxy, - /// LD_PRELOAD agent — captures HTTP + HTTPS, Linux only. No proxy config needed. - #[cfg(target_os = "linux")] - Ldpreload, -} - -#[derive(Debug, Clone, Default, ValueEnum)] -enum OutputMode { - /// Interactive terminal UI with trace list and detail view. - #[default] - Tui, - /// Stream traces as JSON Lines to stdout; auto-exits when child process finishes. - Jsonl, -} - -#[derive(Parser)] -#[command( - name = "phantom", - about = "Zero-instrumentation HTTP/HTTPS API observability tool", - long_about = "phantom — Zero-instrumentation HTTP/HTTPS API observability\n\ -\n\ -Captures every HTTP and HTTPS request/response made by a target process\n\ -and displays them in an interactive TUI or streams them as JSON Lines.\n\ -The target application requires NO code changes.\n\ -\n\ -━━━ CAPTURE BACKENDS ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n\ -\n\ - proxy (default, cross-platform)\n\ - Starts a MITM proxy on 127.0.0.1:. Intercepts HTTP and HTTPS.\n\ -\n\ - • Node.js (`phantom -- node app.js`)\n\ - proxy-preload.js is injected automatically via --require. Both http://\n\ - and https:// are captured with zero application changes.\n\ -\n\ - • PHP (`phantom -- php app.php`)\n\ - The MITM CA certificate is injected automatically via -d curl.cainfo=.\n\ - curl-based HTTP and HTTPS (incl. Guzzle's default handler) are captured\n\ - with zero application changes. Requires PHP >= 5.3.7 for curl.cainfo;\n\ - only the curl extension is covered (not PHP streams).\n\ -\n\ - • Other commands (`phantom -- curl http://api.example.com/v1`)\n\ - HTTP_PROXY / HTTPS_PROXY (and lowercase variants) are set automatically.\n\ - Plain HTTP is captured; HTTPS is captured if the application honours\n\ - these env vars for CONNECT tunnelling (as libcurl does by default).\n\ -\n\ - • Manual (start phantom alone, then configure your app)\n\ - Set HTTP_PROXY=http://127.0.0.1:8080 in the target process yourself.\n\ -\n\ - ldpreload (Linux only)\n\ - Injects libphantom_agent.so via LD_PRELOAD. Hooks send/recv/close at\n\ - the libc level for plain HTTP, and OpenSSL SSL_write/SSL_read for HTTPS\n\ - (captured above the TLS layer, before encryption). No proxy config\n\ - required and no MITM certificate involved — works for any dynamically\n\ - linked process, language-agnostic (e.g. PHP's curl extension).\n\ -\n\ -━━━ OUTPUT MODES ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n\ -\n\ - tui (default) — Interactive terminal UI with trace list + detail view.\n\ -\n\ - jsonl — One JSON object per line on stdout. phantom exits automatically\n\ - when the child process exits (ideal for scripting and AI agents).\n\ -\n\ - JSONL record schema (all fields always present unless marked optional):\n\ - trace_id string W3C-compatible 128-bit trace ID (hex, 32 chars)\n\ - span_id string 64-bit span ID (hex, 16 chars)\n\ - timestamp_ms number Unix epoch milliseconds — request start time\n\ - duration_ms number Round-trip latency in milliseconds\n\ - method string HTTP verb: \"GET\", \"POST\", \"PUT\", \"DELETE\", …\n\ - url string Full request URL (scheme + host + path + query)\n\ - status_code number HTTP response status code (200, 404, 500, …)\n\ - protocol_version string HTTP version string, e.g. \"HTTP/1.1\"\n\ - request_headers object Lower-cased header names → values\n\ - response_headers object Lower-cased header names → values\n\ - request_body string? UTF-8 decoded body; omitted when empty\n\ - response_body string? UTF-8 decoded body; omitted when empty\n\ - source_addr string? Client socket address, e.g. \"127.0.0.1:54321\"\n\ - dest_addr string? Server socket address, e.g. \"93.184.216.34:443\"", - after_long_help = "EXAMPLES\n\ -\n\ - ┌─ Proxy mode (default) ──────────────────────────────────────────────────┐\n\ -\n\ - # Trace a Node.js app — HTTP + HTTPS captured, zero app changes:\n\ - phantom -- node app.js\n\ -\n\ - # Stream traces as JSONL for scripting / AI analysis:\n\ - phantom --output jsonl -- node app.js\n\ -\n\ - # Allow self-signed TLS certs on backend servers:\n\ - phantom --insecure --output jsonl -- node app.js\n\ -\n\ - # Trace a PHP app — curl-based HTTP + HTTPS captured, zero app changes:\n\ - phantom -- php app.php\n\ -\n\ - # Trace any command (plain HTTP only, HTTPS if app uses HTTP_PROXY CONNECT):\n\ - phantom -- curl http://api.example.com/v1/users\n\ -\n\ - # Start proxy only, configure target app manually:\n\ - phantom\n\ - # then in another shell:\n\ - HTTP_PROXY=http://127.0.0.1:8080 node app.js\n\ -\n\ - # Custom port, custom data directory:\n\ - phantom --port 9090 --data-dir ./traces -- node app.js\n\ -\n\ - └─────────────────────────────────────────────────────────────────────────┘\n\ -\n\ - ┌─ LD_PRELOAD mode (Linux only) ──────────────────────────────────────────┐\n\ -\n\ - # Build the agent first:\n\ - cargo build -p phantom-agent\n\ -\n\ - # Trace a process (HTTP + HTTPS, no MITM certificate needed):\n\ - phantom --backend ldpreload \\\n\ - --agent-lib ./target/debug/libphantom_agent.so \\\n\ - -- curl http://api.example.com/v1/users\n\ -\n\ - └─────────────────────────────────────────────────────────────────────────┘\n\ -\n\ - ┌─ Consume JSONL from another process ────────────────────────────────────┐\n\ -\n\ - phantom --output jsonl -- node app.js \\\n\ - | jq 'select(.status_code >= 400)' # filter errors\n\ -\n\ - phantom --output jsonl -- node app.js \\\n\ - | jq '{method,url,status_code,duration_ms}' # compact summary\n\ -\n\ - └─────────────────────────────────────────────────────────────────────────┘", - version -)] -struct Cli { - /// Capture backend: 'proxy' (MITM, cross-platform) or 'ldpreload' (Linux, HTTP + HTTPS). - #[arg(short, long, value_enum, default_value = "proxy")] - backend: Backend, - - /// Output mode: 'tui' opens the interactive UI; 'jsonl' streams one trace - /// per line to stdout and exits when the child process finishes. - #[arg(short, long, value_enum, default_value = "tui")] - output: OutputMode, - - /// TCP port the proxy listens on. - #[arg(short, long, default_value = "8080")] - port: u16, - - /// Disable TLS certificate verification for connections to backend servers. - /// Use when tracing apps that talk to servers with self-signed certificates. - #[arg(long, default_value = "false")] - insecure: bool, - - /// Directory where captured traces are persisted (Fjall key-value store). - /// Defaults to the platform data directory, e.g. ~/.local/share/phantom/data. - #[arg(short, long)] - data_dir: Option, - /// Path to libphantom_agent.so [required for --backend ldpreload] - /// - /// Build with: cargo build -p phantom-agent - /// Then pass: --agent-lib ./target/debug/libphantom_agent.so - #[arg(long, value_name = "PATH")] - agent_lib: Option, +use cli::{Backend, Cli, Commands, GlobalOpts, default_data_dir}; - /// Inject faults into proxied requests (proxy backend only). - /// - /// SPEC formats: - /// delay:100ms fixed 100 ms delay on all requests - /// delay:100ms-500ms random delay in the given range - /// delay:200ms:/api delay only URLs containing "/api" - /// error:503 return HTTP 503 for all requests - /// error:503:0.5 return HTTP 503 with 50% probability - /// error:500:0.1:/api 10% chance of HTTP 500 on URLs containing "/api" - /// - /// Rules are applied in order; delays and errors can be combined. - /// Repeat the flag to add multiple rules: - /// --fault delay:50ms --fault error:500:0.1 - #[arg(long, value_name = "SPEC")] - fault: Vec, - - /// Command to spawn and trace (everything after `--`). - /// - /// proxy mode: HTTP_PROXY is set automatically; Node.js additionally - /// gets proxy-preload.js injected via --require (captures HTTPS too). - /// ldpreload mode: LD_PRELOAD + PHANTOM_SOCKET are set automatically. - #[arg(last = true, value_name = "CMD")] - command: Vec, -} - -fn default_data_dir() -> PathBuf { - dirs::data_local_dir() - .unwrap_or_else(|| PathBuf::from(".")) - .join("phantom") - .join("data") -} - -// ───────────────────────────────────────────────────────────────────────────── -// JSONL output -// ───────────────────────────────────────────────────────────────────────────── - -/// Human-friendly, fully serializable representation of an `HttpTrace`. -/// Emitted as one JSON object per line to stdout in `--output jsonl` mode. -#[derive(Serialize)] -struct JsonlTrace { - /// Unix timestamp of the request in milliseconds. - timestamp_ms: u64, - /// Round-trip duration in milliseconds. - duration_ms: u64, - /// HTTP method ("GET", "POST", …). - method: String, - /// Full request URL. - url: String, - /// HTTP response status code. - status_code: u16, - /// Request headers (lower-cased keys). - request_headers: std::collections::HashMap, - /// Response headers (lower-cased keys). - response_headers: std::collections::HashMap, - /// Request body decoded as UTF-8 (replacement chars for non-UTF-8 bytes). - #[serde(skip_serializing_if = "Option::is_none")] - request_body: Option, - /// Response body decoded as UTF-8 (replacement chars for non-UTF-8 bytes). - #[serde(skip_serializing_if = "Option::is_none")] - response_body: Option, - /// Source socket address, if available. - #[serde(skip_serializing_if = "Option::is_none")] - source_addr: Option, - /// Destination socket address, if available. - #[serde(skip_serializing_if = "Option::is_none")] - dest_addr: Option, - /// HTTP protocol version string (e.g. "HTTP/1.1"). - protocol_version: String, - /// 128-bit W3C trace ID (hex). - trace_id: String, - /// 64-bit span ID (hex). - span_id: String, -} - -fn body_to_str(body: &Option>) -> Option { - body.as_ref() - .map(|b| String::from_utf8_lossy(b).into_owned()) -} - -fn trace_to_jsonl(t: &HttpTrace) -> JsonlTrace { - let timestamp_ms = t - .timestamp - .duration_since(UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(0); - - JsonlTrace { - timestamp_ms, - duration_ms: t.duration.as_millis() as u64, - method: t.method.to_string(), - url: t.url.clone(), - status_code: t.status_code, - request_headers: t.request_headers.clone(), - response_headers: t.response_headers.clone(), - request_body: body_to_str(&t.request_body), - response_body: body_to_str(&t.response_body), - source_addr: t.source_addr.clone(), - dest_addr: t.dest_addr.clone(), - protocol_version: t.protocol_version.clone(), - trace_id: t.trace_id.to_string(), - span_id: t.span_id.to_string(), +/// Maps a child process's exit status onto our own exit code: +/// the child's code clamped to u8, or 128+signal on Unix signal death. +fn exit_code_from_status(status: std::process::ExitStatus) -> ExitCode { + if let Some(code) = status.code() { + return ExitCode::from(code.clamp(0, 255) as u8); } -} - -/// Runs the JSONL output loop: each captured trace is serialized and written to -/// stdout as a single JSON object followed by a newline. -/// -/// Exits when: -/// - The trace channel is closed (sender dropped), -/// - Ctrl-C is received, or -/// - The optional `child` process exits (ldpreload mode). -async fn run_jsonl_output( - store: Arc, - mut trace_rx: tokio::sync::mpsc::Receiver, - child: Option, -) -> anyhow::Result<()> { - // Spawn a background thread to wait() on the child so we don't block the - // async executor. Signal completion via a oneshot channel. - let mut child_done: Option> = if let Some(mut c) = child { - let (tx, rx) = tokio::sync::oneshot::channel(); - std::thread::spawn(move || { - let _ = c.wait(); - let _ = tx.send(()); - }); - Some(rx) - } else { - None - }; - - let ctrl_c = tokio::signal::ctrl_c(); - tokio::pin!(ctrl_c); - - loop { - tokio::select! { - maybe_trace = trace_rx.recv() => { - match maybe_trace { - Some(t) => { - store.insert(&t).ok(); - println!("{}", serde_json::to_string(&trace_to_jsonl(&t))?); - } - None => break, - } - } - _ = &mut ctrl_c => break, - // When the child exits, wait briefly for the backend to flush any - // in-flight datagrams, then drain whatever arrived. - _ = async { - if let Some(rx) = child_done.as_mut() { - let _ = rx.await; - } else { - std::future::pending::<()>().await; - } - } => { - tokio::time::sleep(std::time::Duration::from_millis(200)).await; - while let Ok(t) = trace_rx.try_recv() { - store.insert(&t).ok(); - println!("{}", serde_json::to_string(&trace_to_jsonl(&t))?); - } - break; - } + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + if let Some(signal) = status.signal() { + return ExitCode::from(128u8.saturating_add(signal.clamp(0, 127) as u8)); } } - - Ok(()) + ExitCode::FAILURE } -// ───────────────────────────────────────────────────────────────────────────── -// Main -// ───────────────────────────────────────────────────────────────────────────── +/// Opens the trace store for a query command, adding a hint about fjall's +/// single-process lock when another phantom instance holds it. +fn open_store_for_query(data_dir: &std::path::Path) -> anyhow::Result { + FjallTraceStore::open(data_dir).map_err(|e| { + anyhow::anyhow!( + "{e}\n\ + hint: another phantom process (run/mcp) may hold the store lock on\n\ + {}. Stop it first, or query through the running MCP server.", + data_dir.display() + ) + }) +} #[tokio::main] -async fn main() -> anyhow::Result<()> { +async fn main() -> anyhow::Result { + let cli = Cli::parse(); + + // All diagnostics go to stderr so stdout stays pure JSONL / JSON. + let default_directive = if cli.quiet { + "phantom=error" + } else { + "phantom=info" + }; tracing_subscriber::fmt() + .with_writer(std::io::stderr) .with_env_filter( tracing_subscriber::EnvFilter::from_default_env() - .add_directive("phantom=info".parse()?), + .add_directive(default_directive.parse()?), ) .init(); - let cli = Cli::parse(); - let data_dir = cli.data_dir.clone().unwrap_or_else(default_data_dir); std::fs::create_dir_all(&data_dir)?; + let globals = GlobalOpts { + quiet: cli.quiet, + data_dir: data_dir.clone(), + }; - let store = Arc::new(FjallTraceStore::open(&data_dir)?); - - match cli.backend { - Backend::Proxy => run_proxy(cli, store).await, - #[cfg(target_os = "linux")] - Backend::Ldpreload => run_ldpreload(cli, store).await, - } -} - -// ───────────────────────────────────────────────────────────────────────────── -// Fault injection -// ───────────────────────────────────────────────────────────────────────────── - -fn build_fault_config(specs: &[String]) -> anyhow::Result { - let mut rules = Vec::new(); - for spec in specs { - let rule = phantom_capture::parse_fault_spec(spec) - .map_err(|e| anyhow::anyhow!("--fault {spec:?}: {e}"))?; - rules.push(rule); - } - Ok(FaultConfig { rules }) -} - -// ───────────────────────────────────────────────────────────────────────────── -// Proxy backend -// ───────────────────────────────────────────────────────────────────────────── - -/// RAII guard that deletes a temporary script file on drop. -struct TempScript(PathBuf); - -impl Drop for TempScript { - fn drop(&mut self) { - let _ = std::fs::remove_file(&self.0); - } -} - -/// Returns `true` if `exe` (path or bare name) resolves to `node` or `nodejs`. -fn is_node_command(exe: &str) -> bool { - let base = Path::new(exe) - .file_name() - .and_then(|n| n.to_str()) - .unwrap_or(exe); - base == "node" || base == "nodejs" -} - -/// Returns `true` if `exe` (path or bare name) resolves to `java` or `javaw`. -fn is_java_command(exe: &str) -> bool { - let base = Path::new(exe) - .file_name() - .and_then(|n| n.to_str()) - .unwrap_or(exe); - base == "java" || base == "javaw" -} - -/// Returns `true` if `exe` (path or bare name) resolves to `php` or a -/// version-suffixed PHP binary (e.g. `php7.4`, `php8.2`, `php5.3`). -fn is_php_command(exe: &str) -> bool { - let base = Path::new(exe) - .file_name() - .and_then(|n| n.to_str()) - .unwrap_or(exe); - base == "php" - || (base.starts_with("php") - && !base[3..].is_empty() - && base[3..].chars().all(|c| c.is_ascii_digit() || c == '.')) -} - -/// Spawns `command` as a child process routed through the phantom proxy. -/// -/// * `HTTP_PROXY` / `HTTPS_PROXY` (and lowercase variants) are set so plain -/// HTTP and HTTP-client-honoured HTTPS (e.g. libcurl) are captured. -/// * For Node.js executables the embedded proxy-preload script is written to a -/// temp file and prepended as `--require ` so HTTPS is also captured -/// without touching the application source. -/// * For Java executables, the phantom-java-agent.jar is injected via -javaagent -/// to force proxy settings and bypass SSL verification globally. -/// * For PHP executables, the MITM CA certificate is written to a temp PEM -/// file and injected via `-d curl.cainfo=` so the curl extension -/// trusts phantom's HTTPS interception without any application changes. -/// -/// Returns `(child, Option)`. The `TempScript` must be kept alive -/// until after the child exits so the file is not deleted prematurely. -fn spawn_proxy_child( - command: &[String], - proxy_port: u16, - ca_cert_pem: Option<&str>, -) -> anyhow::Result<(std::process::Child, Option)> { - let exe = &command[0]; - let proxy_url = format!("http://127.0.0.1:{proxy_port}"); - - let mut temp_script: Option = None; - let mut actual_args = command[1..].to_vec(); - - if is_node_command(exe) { - // Write the embedded preload script to a temp file. - let script_path = - std::env::temp_dir().join(format!("phantom-preload-{}.js", std::process::id())); - std::fs::write(&script_path, NODE_PROXY_PRELOAD) - .map_err(|e| anyhow::anyhow!("failed to write proxy preload script: {e}"))?; - temp_script = Some(TempScript(script_path.clone())); - - // Prepend --require