diff --git a/.gitattributes b/.gitattributes
index 67ab542ce..fdb3c5687 100644
--- a/.gitattributes
+++ b/.gitattributes
@@ -5,3 +5,4 @@ crates/yaak-templates/pkg/* linguist-generated=true
# Ensure consistent line endings for test files that check exact content
crates/yaak-http/tests/test.txt text eol=lf
+crates-proxy/yaak-proxy-lib/tests/fixtures/export.har text eol=lf
diff --git a/Cargo.lock b/Cargo.lock
index 0e9119bce..09736a5ce 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11763,6 +11763,7 @@ dependencies = [
"hyper",
"hyper-util",
"include_dir",
+ "libc",
"log 0.4.34",
"r2d2",
"r2d2_sqlite",
@@ -11774,9 +11775,12 @@ dependencies = [
"tempfile",
"tokio",
"ts-rs",
+ "url",
+ "yaak-common",
"yaak-database",
"yaak-filter",
"yaak-http",
+ "yaak-lifecycle",
"yaak-models",
"yaak-proxy",
"yaak-rpc",
diff --git a/apps/yaak-proxy/components/FlowDetail.tsx b/apps/yaak-proxy/components/FlowDetail.tsx
index 1dc644739..061166cac 100644
--- a/apps/yaak-proxy/components/FlowDetail.tsx
+++ b/apps/yaak-proxy/components/FlowDetail.tsx
@@ -73,7 +73,8 @@ export function FlowDetail({ style, sideBySide }: { style?: CSSProperties; sideB
function FlowPane({ flow }: { flow: Flow }) {
const [pretty, setPretty] = useState(true);
const responseBody = useFlowBody(flow, "response", flow.responseTruncated);
- const failed = flow.error != null && flow.status == null;
+ // Failed before any response; an error after the headers is a body cut short mid-way.
+ const noResponse = flow.error != null && flow.status == null;
const mime = mimeEssence(findHeader(flow.responseHeaders, "content-type"));
const queryCount = useMemo(() => queryParams(flow.url).length, [flow.url]);
@@ -153,7 +154,7 @@ function FlowPane({ flow }: { flow: Flow }) {
>
)}
- {!failed && (
+ {!noResponse && (
<>
•
@@ -178,7 +179,7 @@ function FlowPane({ flow }: { flow: Flow }) {
- {failed && (
+ {flow.error != null && (
{flow.error}
@@ -190,7 +191,7 @@ function FlowPane({ flow }: { flow: Flow }) {
tabListClassName="mt-0.5 -mb-1.5"
>
- {failed ? (
+ {noResponse ? (
No response
diff --git a/apps/yaak-proxy/components/FlowList.tsx b/apps/yaak-proxy/components/FlowList.tsx
index 84aae5820..10283af85 100644
--- a/apps/yaak-proxy/components/FlowList.tsx
+++ b/apps/yaak-proxy/components/FlowList.tsx
@@ -212,7 +212,9 @@ const FlowRow = memo(function FlowRow({
onSelect,
}: RowProps) {
const { host, path, query } = splitUrl(flow.url);
- const failed = flow.error != null && flow.status == null;
+ const failed = flow.error != null;
+ // Failed before any response; an error after the headers is a body cut short mid-way.
+ const noResponse = failed && flow.status == null;
const type = shortContentType(flow.contentType);
return (
@@ -261,7 +263,11 @@ const FlowRow = memo(function FlowRow({
flow.responseTruncated ? "text-notice" : "text-text-subtle",
)}
>
- {failed ?
-- : formatSize(flow.responseSize)}
+ {noResponse ? (
+
--
+ ) : (
+ formatSize(flow.responseSize)
+ )}
)}
diff --git a/crates-proxy/yaak-proxy-lib/Cargo.toml b/crates-proxy/yaak-proxy-lib/Cargo.toml
index aeb7233b2..6bfa3b9e6 100644
--- a/crates-proxy/yaak-proxy-lib/Cargo.toml
+++ b/crates-proxy/yaak-proxy-lib/Cargo.toml
@@ -7,25 +7,36 @@ publish = false
[dependencies]
base64 = "0.23.1"
+bytes = "1"
chrono = { workspace = true, features = ["serde"] }
log = { workspace = true }
+http-body-util = "0.1"
+hyper = { version = "1", features = ["http1", "server"] }
+hyper-util = { version = "0.1", features = ["server-graceful", "tokio"] }
include_dir = "0.7"
r2d2 = "0.8.10"
r2d2_sqlite = "0.32"
+reqwest = { workspace = true, features = ["json"] }
rusqlite = { version = "0.38", features = ["bundled", "chrono"] }
sea-query = { version = "1.0", features = ["with-chrono", "attr"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
-tokio = { workspace = true, features = ["io-util", "process"] }
+tokio = { workspace = true, features = ["io-util", "macros", "net", "process", "rt-multi-thread"] }
ts-rs = { workspace = true, features = ["chrono-impl"] }
+url = "2"
+yaak-common = { workspace = true }
yaak-database = { workspace = true }
yaak-filter = { workspace = true }
yaak-http = { workspace = true }
+yaak-lifecycle = { workspace = true }
yaak-models = { workspace = true }
yaak-proxy = { workspace = true }
yaak-rpc = { workspace = true }
yaak-templates = { workspace = true }
+[target.'cfg(unix)'.dependencies]
+libc = "0.2"
+
[dev-dependencies]
base64 = "0.23.1"
bytes = "1"
diff --git a/crates-proxy/yaak-proxy-lib/bindings/gen_models.ts b/crates-proxy/yaak-proxy-lib/bindings/gen_models.ts
index 7f4c19388..0fca90b9b 100644
--- a/crates-proxy/yaak-proxy-lib/bindings/gen_models.ts
+++ b/crates-proxy/yaak-proxy-lib/bindings/gen_models.ts
@@ -40,7 +40,11 @@ elapsedHeadersMs: number | null,
/**
* Milliseconds from `started_at` to the end of the response.
*/
-elapsedMs: number | null, error: string | null, };
+elapsedMs: number | null, error: string | null,
+/**
+ * The id of the Yaak request this flow was last saved as.
+ */
+savedRequestId: string | null, };
export type FlowBody = "request" | "response";
diff --git a/crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts b/crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts
index b440bebd0..5269b1192 100644
--- a/crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts
+++ b/crates-proxy/yaak-proxy-lib/bindings/gen_rpc.ts
@@ -69,10 +69,46 @@ export type ListModelsRequest = Record;
export type ListModelsResponse = { flows: Array, };
+/**
+ * The process that runs the engine.
+ */
+export type Owner = "app" | "cli";
+
+export type ProxySetting = { enabled: boolean, server: string, port: number, };
+
export type ProxyState = "running" | "stopped";
export type ProxyStatePayload = { state: ProxyState, };
export type RpcEventSchema = { model_write: ModelPayload, proxy_state_changed: ProxyStatePayload, };
-export type RpcSchema = { execute_action: [ActionInvocation, boolean], format_body: [FormatBodyRequest, string], get_proxy_state: [GetProxyStateRequest, GetProxyStateResponse], list_actions: [ListActionsRequest, ListActionsResponse], list_models: [ListModelsRequest, ListModelsResponse], ca_status: [CaStatusRequest, CaStatus], ca_trust: [CaTrustRequest, CaStatus], filter_body: [FilterBodyRequest, FilterBodyResponse], get_flow_body: [GetFlowBodyRequest, GetFlowBodyResponse], };
+export type RpcSchema = { execute_action: [ActionInvocation, boolean], format_body: [FormatBodyRequest, string], get_proxy_state: [GetProxyStateRequest, GetProxyStateResponse], list_actions: [ListActionsRequest, ListActionsResponse], list_models: [ListModelsRequest, ListModelsResponse], ca_status: [CaStatusRequest, CaStatus], ca_trust: [CaTrustRequest, CaStatus], filter_body: [FilterBodyRequest, FilterBodyResponse], get_flow_body: [GetFlowBodyRequest, GetFlowBodyResponse], system_proxy_set: [SystemProxySetRequest, SystemProxyStatus], system_proxy_status: [SystemProxyStatusRequest, SystemProxyStatus], };
+
+export type ServiceProxy = {
+/**
+ * The network service, as System Settings names it: "Wi-Fi", "USB 10/100/1000 LAN".
+ */
+name: string, http: ProxySetting, https: ProxySetting, };
+
+/**
+ * How a change gets administrator rights.
+ */
+export type SystemProxyBackend = "helper" | "admin_prompt" | "sudo";
+
+export type SystemProxySetRequest = { enabled: boolean, };
+
+export type SystemProxyStatus = {
+/**
+ * Every enabled network service sends HTTP and HTTPS to `127.0.0.1:`.
+ */
+enabled: boolean,
+/**
+ * The port checked against.
+ */
+port: number, services: Array, backend: SystemProxyBackend,
+/**
+ * This platform has a system proxy the app can set.
+ */
+supported: boolean, };
+
+export type SystemProxyStatusRequest = Record;
diff --git a/crates-proxy/yaak-proxy-lib/migrations/00000003_flow_saved_request.sql b/crates-proxy/yaak-proxy-lib/migrations/00000003_flow_saved_request.sql
new file mode 100644
index 000000000..ea584f355
--- /dev/null
+++ b/crates-proxy/yaak-proxy-lib/migrations/00000003_flow_saved_request.sql
@@ -0,0 +1,2 @@
+-- The Yaak request a flow was last saved as.
+ALTER TABLE flows ADD COLUMN saved_request_id TEXT;
diff --git a/crates-proxy/yaak-proxy-lib/src/bodies.rs b/crates-proxy/yaak-proxy-lib/src/bodies.rs
index 803b091a7..d0bd60731 100644
--- a/crates-proxy/yaak-proxy-lib/src/bodies.rs
+++ b/crates-proxy/yaak-proxy-lib/src/bodies.rs
@@ -20,12 +20,15 @@ pub struct DecodedBody {
pub data: Vec,
/// Bytes stored for the body, as they were on the wire.
pub stored_size: usize,
- /// The body is incomplete: cut on the wire at the capture limit, or decoded past `limit`.
+ /// The body is incomplete: cut on the wire at the capture limit, decoded past `limit`, or
+ /// decoded only as far as `decode_error`.
pub truncated: bool,
+ /// Why decoding stopped before the end of the stored bytes.
+ pub decode_error: Option,
}
-/// A flow's body decoded with its Content-Encoding, at most `limit` bytes of it. A body cut
-/// at the capture limit decodes as far as its stored bytes go.
+/// A flow's body decoded with its Content-Encoding, at most `limit` bytes of it. A body that
+/// fails to decode, as one cut short does, keeps what decoded before the failure.
pub async fn read_body(
blobs: &BlobManager,
flow: &Flow,
@@ -34,7 +37,9 @@ pub async fn read_body(
) -> Result {
let raw = read_raw_body(blobs, &flow.id, body)?;
let stored_size = raw.len();
- let wire_truncated = flow.truncated(body);
+ // A response that failed after its headers ended short on the wire.
+ let cut = body == FlowBody::Response && flow.status.is_some() && flow.error.is_some();
+ let wire_truncated = flow.truncated(body) || cut;
let encoding =
ContentEncoding::from_header(find_header(flow.headers(body), "content-encoding"));
// A HEAD or 304 response names its encoding but has no body to decode.
@@ -42,19 +47,50 @@ pub async fn read_body(
let truncated = wire_truncated || raw.len() > limit;
let mut data = raw;
data.truncate(limit);
- return Ok(DecodedBody { data, stored_size, truncated });
+ return Ok(DecodedBody { data, stored_size, truncated, decode_error: None });
}
// One byte past the limit tells a body that fills it from one that overflows it.
let take = u64::try_from(limit).unwrap_or(u64::MAX).saturating_add(1);
let mut data = Vec::new();
let mut decoder = streaming_decoder(Cursor::new(raw), encoding).take(take);
- match decoder.read_to_end(&mut data).await {
- Ok(_) => {}
- Err(_) if wire_truncated => {}
- Err(e) => return Err(format!("Failed to decode {encoding:?} body: {e}")),
- }
- let truncated = wire_truncated || data.len() > limit;
+ let decode_error = match decoder.read_to_end(&mut data).await {
+ Ok(_) => None,
+ // Expected of a body cut on the wire; anything else is worth saying.
+ Err(_) if wire_truncated => None,
+ Err(e) => Some(format!("Failed to decode {encoding:?} body: {e}")),
+ };
+ let truncated = wire_truncated || decode_error.is_some() || data.len() > limit;
data.truncate(limit);
- Ok(DecodedBody { data, stored_size, truncated })
+ Ok(DecodedBody { data, stored_size, truncated, decode_error })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use yaak_models::blob_manager::BodyChunkWriter;
+
+ #[tokio::test]
+ async fn response_failed_after_its_headers_is_truncated() {
+ let (_db, blobs, _rx) = yaak_models::init_in_memory().unwrap();
+ let mut writer =
+ BodyChunkWriter::new(blobs.clone(), Flow::body_id("fl_cut", FlowBody::Response));
+ writer.write(b"{\"rows\": [").unwrap();
+ writer.finish().unwrap();
+ let cut = Flow {
+ id: "fl_cut".into(),
+ status: Some(200),
+ error: Some("error reading a body from connection".into()),
+ ..Default::default()
+ };
+
+ let body = read_body(&blobs, &cut, FlowBody::Response, 1024).await.unwrap();
+ assert_eq!((body.data.as_slice(), body.truncated), (&b"{\"rows\": ["[..], true));
+ // The request went out whole; the error is the response's.
+ let request = read_body(&blobs, &cut, FlowBody::Request, 1024).await.unwrap();
+ assert!(!request.truncated);
+ // With no headers there was never a response body to cut.
+ let refused = Flow { status: None, ..cut };
+ assert!(!read_body(&blobs, &refused, FlowBody::Response, 1024).await.unwrap().truncated);
+ }
}
diff --git a/crates-proxy/yaak-proxy-lib/src/ca.rs b/crates-proxy/yaak-proxy-lib/src/ca.rs
index 3b57e145f..a36f164b1 100644
--- a/crates-proxy/yaak-proxy-lib/src/ca.rs
+++ b/crates-proxy/yaak-proxy-lib/src/ca.rs
@@ -1,8 +1,8 @@
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
use std::path::Path;
use ts_rs::TS;
-#[derive(Debug, Clone, Serialize, TS)]
+#[derive(Debug, Clone, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "gen_rpc.ts")]
pub struct CaStatus {
diff --git a/crates-proxy/yaak-proxy-lib/src/client.rs b/crates-proxy/yaak-proxy-lib/src/client.rs
new file mode 100644
index 000000000..2f714a6cc
--- /dev/null
+++ b/crates-proxy/yaak-proxy-lib/src/client.rs
@@ -0,0 +1,229 @@
+//! The other side of the control API, for the CLI and the MCP tools: find the owner through
+//! `control.json` and call it with its token, or, when nothing owns the engine, read `proxy.db`
+//! and `blobs.sqlite` read-only so captured traffic stays reachable.
+
+use crate::bodies::{self, DecodedBody};
+use crate::ca::CaStatus;
+use crate::control::{
+ self, ControlInfo, ControlStatus, ErrorBody, StartRequest, SystemProxyRequest, TRUNCATED_HEADER,
+};
+use crate::db::ProxyQueryManager;
+use crate::flows::{self, FlowDetail, FlowFilter};
+use crate::har::{self, Har};
+use crate::macos::SystemProxyStatus;
+use crate::models::{Flow, FlowBody};
+use crate::save::{SaveFlowRequest, SaveFlowResponse, WorkspaceSummary};
+use crate::settings::ProxySettings;
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use std::path::Path;
+use yaak_models::blob_manager::BlobManager;
+
+/// Where captured traffic is read from.
+pub enum ProxyConnection {
+ /// A process owns the engine; everything goes through its control API.
+ Live(ControlClient),
+ /// Nothing owns the engine; flows are read from disk.
+ Offline(OfflineStore),
+}
+
+impl ProxyConnection {
+ /// The live owner in `data_dir`, or its databases when there is none.
+ pub fn connect(data_dir: &Path) -> Result {
+ match control::live_owner(data_dir) {
+ Some(info) => Ok(Self::Live(ControlClient::new(info))),
+ None => Ok(Self::Offline(OfflineStore::open(data_dir)?)),
+ }
+ }
+
+ pub async fn flows(&self, filter: &FlowFilter) -> Result, String> {
+ match self {
+ Self::Live(c) => c.flows(filter).await,
+ Self::Offline(s) => s.flows(filter),
+ }
+ }
+
+ pub async fn flow(&self, id: &str, body_limit: usize) -> Result {
+ match self {
+ Self::Live(c) => c.flow(id, body_limit).await,
+ Self::Offline(s) => s.flow(id, body_limit).await,
+ }
+ }
+
+ pub async fn body(&self, id: &str, which: FlowBody) -> Result {
+ match self {
+ Self::Live(c) => c.body(id, which).await,
+ Self::Offline(s) => s.body(id, which).await,
+ }
+ }
+
+ pub async fn har(&self, filter: &FlowFilter) -> Result {
+ match self {
+ Self::Live(c) => c.har(filter).await,
+ Self::Offline(s) => s.har(filter).await,
+ }
+ }
+}
+
+/// Calls the control API of the owner `control.json` names.
+pub struct ControlClient {
+ info: ControlInfo,
+ http: reqwest::Client,
+}
+
+impl ControlClient {
+ pub fn new(info: ControlInfo) -> Self {
+ // Never through the system proxy, which may be the engine itself.
+ let http = reqwest::Client::builder().no_proxy().build().unwrap_or_default();
+ Self { info, http }
+ }
+
+ pub fn info(&self) -> &ControlInfo {
+ &self.info
+ }
+
+ pub async fn status(&self) -> Result {
+ self.get("/status").await
+ }
+
+ pub async fn start(&self, port: Option) -> Result {
+ self.post("/start", &StartRequest { port }).await
+ }
+
+ pub async fn stop(&self) -> Result {
+ self.post("/stop", &serde_json::json!({})).await
+ }
+
+ pub async fn flows(&self, filter: &FlowFilter) -> Result, String> {
+ self.get(&format!("/flows?{}", filter.to_query())).await
+ }
+
+ pub async fn flow(&self, id: &str, body_limit: usize) -> Result {
+ self.get(&format!("/flows/{}?bodyLimit={body_limit}", encode(id))).await
+ }
+
+ pub async fn body(&self, id: &str, which: FlowBody) -> Result {
+ let side = match which {
+ FlowBody::Request => "request-body",
+ FlowBody::Response => "response-body",
+ };
+ let resp =
+ self.send(self.http.get(self.url(&format!("/flows/{}/{side}", encode(id))))).await?;
+ let truncated = resp.headers().get(TRUNCATED_HEADER).is_some_and(|v| v == "true");
+ let data = resp.bytes().await.map_err(|e| e.to_string())?.to_vec();
+ Ok(DecodedBody { stored_size: data.len(), data, truncated, decode_error: None })
+ }
+
+ pub async fn har(&self, filter: &FlowFilter) -> Result {
+ self.get(&format!("/har?{}", filter.to_query())).await
+ }
+
+ pub async fn ca_pem(&self) -> Result {
+ let resp = self.send(self.http.get(self.url("/ca"))).await?;
+ resp.text().await.map_err(|e| e.to_string())
+ }
+
+ pub async fn trust_ca(&self) -> Result {
+ self.post("/ca/trust", &serde_json::json!({})).await
+ }
+
+ pub async fn system_proxy(&self) -> Result {
+ self.get("/system-proxy").await
+ }
+
+ pub async fn set_system_proxy(&self, enabled: bool) -> Result {
+ self.post("/system-proxy", &SystemProxyRequest { enabled }).await
+ }
+
+ pub async fn workspaces(&self) -> Result, String> {
+ self.get("/workspaces").await
+ }
+
+ pub async fn save(&self, id: &str, req: &SaveFlowRequest) -> Result {
+ self.post(&format!("/flows/{}/save", encode(id)), req).await
+ }
+
+ fn url(&self, path: &str) -> String {
+ format!("{}{path}", self.info.base_url())
+ }
+
+ async fn get(&self, path: &str) -> Result {
+ let resp = self.send(self.http.get(self.url(path))).await?;
+ resp.json().await.map_err(|e| format!("Invalid response from the proxy: {e}"))
+ }
+
+ async fn post(
+ &self,
+ path: &str,
+ body: &B,
+ ) -> Result {
+ let body = serde_json::to_vec(body).map_err(|e| e.to_string())?;
+ let req =
+ self.http.post(self.url(path)).header("content-type", "application/json").body(body);
+ let resp = self.send(req).await?;
+ resp.json().await.map_err(|e| format!("Invalid response from the proxy: {e}"))
+ }
+
+ async fn send(&self, req: reqwest::RequestBuilder) -> Result {
+ let resp = req
+ .bearer_auth(&self.info.token)
+ .send()
+ .await
+ .map_err(|e| format!("Failed to reach the proxy's control API: {e}"))?;
+ if resp.status().is_success() {
+ return Ok(resp);
+ }
+ let status = resp.status();
+ let text = resp.text().await.unwrap_or_default();
+ Err(serde_json::from_str::(&text)
+ .map(|e| e.error)
+ .unwrap_or_else(|_| format!("The proxy answered {status}: {text}")))
+ }
+}
+
+fn encode(segment: &str) -> String {
+ url::form_urlencoded::byte_serialize(segment.as_bytes()).collect()
+}
+
+/// `proxy.db` and `blobs.sqlite`, opened read-only while no process owns the engine.
+pub struct OfflineStore {
+ db: ProxyQueryManager,
+ blobs: BlobManager,
+ capture_body_limit: usize,
+}
+
+impl OfflineStore {
+ pub fn open(data_dir: &Path) -> Result {
+ let db_path = data_dir.join("proxy.db");
+ if !db_path.exists() {
+ return Err(format!(
+ "No captured traffic yet: {} does not exist. Start Yaak Proxy to capture some.",
+ db_path.display()
+ ));
+ }
+ let db = ProxyQueryManager::open_read_only(&db_path)
+ .map_err(|e| format!("Failed to open {}: {e}", db_path.display()))?;
+ let blobs = crate::db::open_blobs_read_only(&data_dir.join("blobs.sqlite"))
+ .map_err(|e| format!("Failed to open the proxy's blob store: {e}"))?;
+ let settings = ProxySettings::load(&data_dir.join("settings.json"))?;
+ Ok(Self { db, blobs, capture_body_limit: settings.capture_body_limit })
+ }
+
+ pub fn flows(&self, filter: &FlowFilter) -> Result, String> {
+ flows::list(&self.db, filter)
+ }
+
+ pub async fn flow(&self, id: &str, body_limit: usize) -> Result {
+ flows::detail(&self.db, &self.blobs, id, body_limit).await
+ }
+
+ pub async fn body(&self, id: &str, which: FlowBody) -> Result {
+ let flow = flows::get(&self.db, id)?;
+ bodies::read_body(&self.blobs, &flow, which, self.capture_body_limit).await
+ }
+
+ pub async fn har(&self, filter: &FlowFilter) -> Result {
+ let flows = flows::list(&self.db, filter)?;
+ har::export(&self.blobs, &flows, self.capture_body_limit).await
+ }
+}
diff --git a/crates-proxy/yaak-proxy-lib/src/control.rs b/crates-proxy/yaak-proxy-lib/src/control.rs
new file mode 100644
index 000000000..0eeb17969
--- /dev/null
+++ b/crates-proxy/yaak-proxy-lib/src/control.rs
@@ -0,0 +1,605 @@
+//! The control API: how the CLI, MCP tools and a second app launch reach the process that owns
+//! the engine. The owner serves JSON on `127.0.0.1:` behind a bearer token and writes
+//! where to find it to `control.json` in the proxy data directory (mode 0600). One owner at a
+//! time: a file whose process is gone, or whose port no longer answers, is stale and replaced.
+
+use crate::ca::{self, CaStatus};
+use crate::flows::{self, FlowFilter};
+use crate::har;
+use crate::macos::SystemProxyStatus;
+use crate::models::{FlowBody, find_header};
+use crate::save::SaveFlowRequest;
+use crate::{ProxyCtx, ProxyState, bodies};
+use bytes::Bytes;
+use chrono::{DateTime, Utc};
+use http_body_util::{BodyExt, Full, Limited};
+use hyper::body::Incoming;
+use hyper::header::{AUTHORIZATION, CONTENT_TYPE};
+use hyper::{Method, Request, Response, StatusCode};
+use hyper_util::rt::TokioIo;
+use hyper_util::server::graceful::GracefulShutdown;
+use log::{info, warn};
+use serde::{Deserialize, Serialize};
+use std::io::Write;
+use std::net::{SocketAddr, TcpStream};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::mpsc as std_mpsc;
+use std::time::Duration;
+use tokio::net::TcpListener;
+use ts_rs::TS;
+use yaak_proxy::CaStore;
+
+pub const CONTROL_FILE: &str = "control.json";
+const LOCK_FILE: &str = "control.lock";
+/// Body bytes `GET /flows/{id}` returns per side unless `bodyLimit` says otherwise.
+pub const DEFAULT_BODY_PREVIEW_LIMIT: usize = 1024 * 1024;
+const MAX_REQUEST_BODY: usize = 64 * 1024;
+const LIVENESS_TIMEOUT: Duration = Duration::from_millis(500);
+const SHUTDOWN_GRACE: Duration = Duration::from_secs(5);
+
+/// The process that runs the engine.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, TS)]
+#[serde(rename_all = "snake_case")]
+#[ts(export, export_to = "gen_rpc.ts")]
+pub enum Owner {
+ App,
+ Cli,
+}
+
+/// `control.json`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ControlInfo {
+ pub pid: u32,
+ pub control_port: u16,
+ /// The engine's port while it runs.
+ pub proxy_port: Option,
+ pub token: String,
+ pub owner: Owner,
+ pub started_at: DateTime,
+}
+
+impl ControlInfo {
+ pub fn base_url(&self) -> String {
+ format!("http://127.0.0.1:{}", self.control_port)
+ }
+}
+
+pub fn control_path(data_dir: &Path) -> PathBuf {
+ data_dir.join(CONTROL_FILE)
+}
+
+/// `control.json` as it is, live or stale. `None` when there is none or it does not parse.
+pub fn read(data_dir: &Path) -> Option {
+ let contents = std::fs::read_to_string(control_path(data_dir)).ok()?;
+ serde_json::from_str(&contents).ok()
+}
+
+/// Held while `control.json` is checked and then removed or claimed, across processes, so two
+/// of them cleaning up a stale file cannot delete each other's fresh claim. Released on drop.
+fn claim_lock(data_dir: &Path) -> std::io::Result {
+ std::fs::create_dir_all(data_dir)?;
+ let file = std::fs::OpenOptions::new()
+ .create(true)
+ .truncate(false)
+ .write(true)
+ .open(data_dir.join(LOCK_FILE))?;
+ file.lock()?;
+ Ok(file)
+}
+
+/// The owner running the engine now. A stale `control.json` is removed on the way, as is one
+/// that does not parse: it is only ever published whole, so that is never a claim in progress.
+pub fn live_owner(data_dir: &Path) -> Option {
+ let _lock = claim_lock(data_dir)
+ .inspect_err(|e| warn!("Failed to lock {LOCK_FILE}, checking without it: {e}"))
+ .ok();
+ live_owner_locked(data_dir)
+}
+
+/// [`live_owner`] for a caller holding the claim lock.
+fn live_owner_locked(data_dir: &Path) -> Option {
+ let path = control_path(data_dir);
+ let contents = std::fs::read_to_string(&path).ok()?;
+ let Ok(info) = serde_json::from_str::(&contents) else {
+ warn!("Removing unreadable {}", path.display());
+ let _ = std::fs::remove_file(&path);
+ return None;
+ };
+ if is_live(&info) {
+ return Some(info);
+ }
+ info!("Removing stale {CONTROL_FILE} of {:?} pid {}", info.owner, info.pid);
+ remove_if_token(data_dir, &info.token);
+ None
+}
+
+/// The process is alive and its control port answers. The port check catches a pid the OS
+/// has handed to some other process since the owner died.
+fn is_live(info: &ControlInfo) -> bool {
+ pid_alive(info.pid)
+ && TcpStream::connect_timeout(
+ &SocketAddr::from(([127, 0, 0, 1], info.control_port)),
+ LIVENESS_TIMEOUT,
+ )
+ .is_ok()
+}
+
+#[cfg(unix)]
+fn pid_alive(pid: u32) -> bool {
+ let Ok(pid) = libc::pid_t::try_from(pid) else {
+ return false;
+ };
+ // Signal 0 checks the process exists without signalling it; EPERM means it exists but is
+ // someone else's.
+ let alive = unsafe { libc::kill(pid, 0) } == 0;
+ alive || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
+}
+
+#[cfg(not(unix))]
+fn pid_alive(_pid: u32) -> bool {
+ // The control port check decides.
+ true
+}
+
+/// Removes `control.json` if it is still the one with `token`, so an owner never removes its
+/// successor's file. The caller holds the claim lock.
+fn remove_if_token(data_dir: &Path, token: &str) {
+ if read(data_dir).is_some_and(|info| info.token == token) {
+ let _ = std::fs::remove_file(control_path(data_dir));
+ }
+}
+
+/// Writes `info` to `path`, readable by its owner alone and never seen half-written. With
+/// `create_new`, fails when a file is already there, so two processes cannot both claim the
+/// engine.
+fn write_file(path: &Path, info: &ControlInfo, create_new: bool) -> std::io::Result<()> {
+ let contents = serde_json::to_vec_pretty(info)?;
+ let tmp = path.with_extension(format!("{}.tmp", std::process::id()));
+ let _ = std::fs::remove_file(&tmp);
+ let mut options = std::fs::OpenOptions::new();
+ options.write(true).create_new(true);
+ #[cfg(unix)]
+ std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
+ options.open(&tmp)?.write_all(&contents)?;
+ let published = if create_new {
+ // A link, unlike a rename, refuses to replace an existing file.
+ std::fs::hard_link(&tmp, path).and_then(|_| std::fs::remove_file(&tmp))
+ } else {
+ std::fs::rename(&tmp, path)
+ };
+ if published.is_err() {
+ let _ = std::fs::remove_file(&tmp);
+ }
+ published
+}
+
+/// The owner's `control.json`, rewritten as the engine starts and stops.
+pub(crate) struct ControlFile {
+ path: PathBuf,
+ pub(crate) info: ControlInfo,
+}
+
+impl ControlFile {
+ pub(crate) fn set_proxy_port(&mut self, port: Option) {
+ if self.info.proxy_port == port {
+ return;
+ }
+ self.info.proxy_port = port;
+ if let Err(e) = write_file(&self.path, &self.info, false) {
+ warn!("Failed to update {}: {e}", self.path.display());
+ }
+ }
+}
+
+// -- Responses --
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ControlStatus {
+ pub owner: Owner,
+ pub pid: u32,
+ pub started_at: DateTime,
+ pub state: ProxyState,
+ /// The engine's port while it runs.
+ pub proxy_port: Option,
+ /// The port the engine starts on.
+ pub configured_port: u16,
+ pub flow_count: usize,
+ pub ca: CaStatus,
+ /// `None` when it could not be read.
+ pub system_proxy: Option,
+}
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct StartRequest {
+ /// For this run only; `settings.json` keeps its port.
+ pub port: Option,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SystemProxyRequest {
+ pub enabled: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ErrorBody {
+ pub error: String,
+}
+
+/// Raw body headers: whether the body is cut short, at the capture limit or on the wire.
+pub const TRUNCATED_HEADER: &str = "x-yaak-truncated";
+
+// -- Server --
+
+/// The control server of the process that owns the engine. Dropping it stops serving and
+/// removes `control.json`.
+pub struct ControlServer {
+ info: ControlInfo,
+ data_dir: PathBuf,
+ ctx: ProxyCtx,
+ shutdown_tx: Option>,
+ thread: Option>,
+ stop_rx: Option>,
+}
+
+impl ControlServer {
+ /// Claims the engine for this process and serves the control API for `ctx`. Fails when
+ /// another live owner has it.
+ pub fn start(ctx: ProxyCtx) -> Result {
+ let data_dir = ctx.data_dir().to_path_buf();
+ let path = control_path(&data_dir);
+ if let Some(owner) = live_owner(&data_dir) {
+ return Err(already_owned(&owner));
+ }
+
+ let (ready_tx, ready_rx) = std_mpsc::channel();
+ let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
+ let (stop_tx, stop_rx) = std_mpsc::channel();
+ let token = yaak_database::generate_id_of_length(40);
+ let state = Arc::new(ServerState { ctx: ctx.clone(), token: token.clone(), stop_tx });
+
+ let thread = std::thread::spawn(move || {
+ let rt = match tokio::runtime::Builder::new_multi_thread()
+ .worker_threads(2)
+ .enable_all()
+ .thread_name("yaak-proxy-control")
+ .build()
+ {
+ Ok(rt) => rt,
+ Err(e) => {
+ let _ = ready_tx.send(Err(format!("Failed to create runtime: {e}")));
+ return;
+ }
+ };
+ rt.block_on(async move {
+ let listener = match TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0))).await
+ {
+ Ok(l) => l,
+ Err(e) => {
+ let _ = ready_tx.send(Err(format!("Failed to bind control port: {e}")));
+ return;
+ }
+ };
+ let _ = ready_tx
+ .send(listener.local_addr().map(|a| a.port()).map_err(|e| e.to_string()));
+ serve(listener, state, shutdown_rx).await;
+ });
+ });
+
+ let control_port = match ready_rx.recv() {
+ Ok(Ok(port)) => port,
+ Ok(Err(e)) => return Err(e),
+ Err(_) => return Err("Control thread died before binding".into()),
+ };
+ let info = ControlInfo {
+ pid: std::process::id(),
+ control_port,
+ proxy_port: ctx.port(),
+ token,
+ owner: ctx.owner(),
+ started_at: Utc::now(),
+ };
+ let mut server = Self {
+ info: info.clone(),
+ data_dir: data_dir.clone(),
+ ctx: ctx.clone(),
+ shutdown_tx: Some(shutdown_tx),
+ thread: Some(thread),
+ stop_rx: Some(stop_rx),
+ };
+
+ // Checked again under the lock: another process may have claimed it since.
+ let claimed = claim_lock(&data_dir)
+ .map_err(|e| format!("Failed to lock {LOCK_FILE}: {e}"))
+ .and_then(|_lock| match live_owner_locked(&data_dir) {
+ Some(owner) => Err(already_owned(&owner)),
+ None => write_file(&path, &info, true)
+ .map_err(|e| format!("Failed to write {}: {e}", path.display())),
+ });
+ if let Err(e) = claimed {
+ // Nothing of this server's to remove.
+ server.info.token.clear();
+ return Err(e);
+ }
+ ctx.set_control_file(Some(ControlFile { path, info }));
+ info!("Control API listening on 127.0.0.1:{control_port}");
+ Ok(server)
+ }
+
+ pub fn info(&self) -> &ControlInfo {
+ &self.info
+ }
+
+ /// Receives once per `POST /stop`. A CLI daemon exits on it; the app only stops its engine.
+ pub fn take_stop_rx(&mut self) -> Option> {
+ self.stop_rx.take()
+ }
+}
+
+pub(crate) fn already_owned(owner: &ControlInfo) -> String {
+ let who = match owner.owner {
+ Owner::App => "Yaak Proxy",
+ Owner::Cli => "the yaak CLI",
+ };
+ format!("The proxy is already running in {who} (pid {})", owner.pid)
+}
+
+impl Drop for ControlServer {
+ fn drop(&mut self) {
+ if let Some(tx) = self.shutdown_tx.take() {
+ let _ = tx.send(());
+ }
+ if let Some(thread) = self.thread.take() {
+ let _ = thread.join();
+ }
+ if !self.info.token.is_empty() {
+ self.ctx.set_control_file(None);
+ let _lock = claim_lock(&self.data_dir).ok();
+ remove_if_token(&self.data_dir, &self.info.token);
+ }
+ }
+}
+
+struct ServerState {
+ ctx: ProxyCtx,
+ token: String,
+ stop_tx: std_mpsc::Sender<()>,
+}
+
+async fn serve(
+ listener: TcpListener,
+ state: Arc,
+ mut shutdown_rx: tokio::sync::oneshot::Receiver<()>,
+) {
+ // Lets a request in flight at shutdown finish, such as the `POST /stop` a daemon exits on.
+ let graceful = GracefulShutdown::new();
+ loop {
+ tokio::select! {
+ result = listener.accept() => match result {
+ Ok((stream, _)) => {
+ let state = state.clone();
+ let service = hyper::service::service_fn(move |req| {
+ let state = state.clone();
+ async move { Ok::<_, std::convert::Infallible>(handle(&state, req).await) }
+ });
+ let conn = hyper::server::conn::http1::Builder::new()
+ .serve_connection(TokioIo::new(stream), service);
+ let conn = graceful.watch(conn);
+ tokio::spawn(async move {
+ let _ = conn.await;
+ });
+ }
+ Err(e) => warn!("Control accept error: {e}"),
+ },
+ _ = &mut shutdown_rx => break,
+ }
+ }
+ drop(listener);
+ if tokio::time::timeout(SHUTDOWN_GRACE, graceful.shutdown()).await.is_err() {
+ warn!("Control connections were still open after {SHUTDOWN_GRACE:?}");
+ }
+}
+
+type ControlResponse = Response>;
+
+struct ApiError {
+ status: StatusCode,
+ message: String,
+}
+
+impl ApiError {
+ fn new(status: StatusCode, message: impl Into) -> Self {
+ Self { status, message: message.into() }
+ }
+}
+
+impl From for ApiError {
+ fn from(message: String) -> Self {
+ Self::new(StatusCode::INTERNAL_SERVER_ERROR, message)
+ }
+}
+
+async fn handle(state: &ServerState, req: Request) -> ControlResponse {
+ if !authorized(&req, &state.token) {
+ return error_response(ApiError::new(
+ StatusCode::UNAUTHORIZED,
+ "Missing or wrong bearer token; it is in control.json",
+ ));
+ }
+ match route(state, req).await {
+ Ok(resp) => resp,
+ Err(e) => error_response(e),
+ }
+}
+
+fn authorized(req: &Request, token: &str) -> bool {
+ let Some(given) = req
+ .headers()
+ .get(AUTHORIZATION)
+ .and_then(|v| v.to_str().ok())
+ .and_then(|v| v.strip_prefix("Bearer "))
+ else {
+ return false;
+ };
+ // Compares every byte, so the time taken says nothing about how much matched.
+ given.len() == token.len()
+ && given.bytes().zip(token.bytes()).fold(0u8, |acc, (a, b)| acc | (a ^ b)) == 0
+}
+
+async fn route(state: &ServerState, req: Request) -> Result {
+ let ctx = &state.ctx;
+ let method = req.method().clone();
+ let path = req.uri().path().to_string();
+ let query = req.uri().query().unwrap_or_default().to_string();
+ let segments: Vec<&str> = path.trim_matches('/').split('/').collect();
+
+ match (&method, segments.as_slice()) {
+ (&Method::GET, ["status"]) => json(&status(ctx).await?),
+ (&Method::POST, ["start"]) => {
+ let body: StartRequest = read_json(req).await?;
+ ctx.start_on(body.port)?;
+ json(&status(ctx).await?)
+ }
+ (&Method::POST, ["stop"]) => {
+ ctx.stop()?;
+ let _ = state.stop_tx.send(());
+ json(&status(ctx).await?)
+ }
+ (&Method::GET, ["flows"]) => {
+ let filter = FlowFilter::from_query(&query).map_err(bad_request)?;
+ json(&flows::list(&ctx.db, &filter)?)
+ }
+ (&Method::GET, ["flows", id]) => {
+ found(ctx, id)?;
+ let limit = body_limit(&query)?;
+ json(&flows::detail(&ctx.db, &ctx.blobs, id, limit).await?)
+ }
+ (&Method::GET, ["flows", id, "request-body"]) => raw_body(ctx, id, FlowBody::Request).await,
+ (&Method::GET, ["flows", id, "response-body"]) => {
+ raw_body(ctx, id, FlowBody::Response).await
+ }
+ (&Method::POST, ["flows", id, "save"]) => {
+ found(ctx, id)?;
+ let id = id.to_string();
+ let body: SaveFlowRequest = read_json(req).await?;
+ json(&ctx.save_flow(&id, &body).await.map_err(bad_request)?)
+ }
+ (&Method::GET, ["har"]) => {
+ let filter = FlowFilter::from_query(&query).map_err(bad_request)?;
+ let flows = flows::list(&ctx.db, &filter)?;
+ json(&har::export(&ctx.blobs, &flows, ctx.capture_body_limit()?).await?)
+ }
+ (&Method::GET, ["ca"]) => {
+ let ca = CaStore::load_or_create(&ctx.ca_dir())
+ .map_err(|e| format!("Failed to load proxy CA: {e}"))?;
+ Ok(response(StatusCode::OK, "application/x-pem-file", ca.pem().to_string()))
+ }
+ (&Method::POST, ["ca", "trust"]) => {
+ let ca = CaStore::load_or_create(&ctx.ca_dir())
+ .map_err(|e| format!("Failed to load proxy CA: {e}"))?;
+ ca::trust(ca.cert_path()).await.map_err(bad_request)?;
+ json(&ca::status(ca.cert_path()).await?)
+ }
+ (&Method::GET, ["system-proxy"]) => {
+ let ctx = ctx.clone();
+ json(&blocking(move || ctx.system_proxy_status()).await?)
+ }
+ (&Method::POST, ["system-proxy"]) => {
+ let body: SystemProxyRequest = read_json(req).await?;
+ let ctx = ctx.clone();
+ json(&blocking(move || ctx.set_system_proxy(body.enabled)).await.map_err(bad_request)?)
+ }
+ (&Method::GET, ["workspaces"]) => json(&ctx.workspaces().map_err(bad_request)?),
+ _ => Err(ApiError::new(StatusCode::NOT_FOUND, format!("No route {method} {path}"))),
+ }
+}
+
+async fn status(ctx: &ProxyCtx) -> Result {
+ let control =
+ ctx.control_info().ok_or_else(|| "The control API is shutting down".to_string())?;
+ let proxy_port = ctx.port();
+ let ca = ca::status(&ctx.ca_dir().join("ca.pem")).await?;
+ let system_proxy = {
+ let ctx = ctx.clone();
+ blocking(move || ctx.system_proxy_status()).await.ok()
+ };
+ Ok(ControlStatus {
+ owner: control.owner,
+ pid: control.pid,
+ started_at: control.started_at,
+ state: if proxy_port.is_some() { ProxyState::Running } else { ProxyState::Stopped },
+ proxy_port,
+ configured_port: ctx.settings()?.port,
+ flow_count: flows::count(&ctx.db)?,
+ ca,
+ system_proxy,
+ })
+}
+
+fn found(ctx: &ProxyCtx, id: &str) -> Result<(), ApiError> {
+ match flows::find(&ctx.db, id) {
+ Some(_) => Ok(()),
+ None => Err(ApiError::new(StatusCode::NOT_FOUND, format!("No flow {id}"))),
+ }
+}
+
+async fn raw_body(ctx: &ProxyCtx, id: &str, which: FlowBody) -> Result {
+ let flow = flows::find(&ctx.db, id)
+ .ok_or_else(|| ApiError::new(StatusCode::NOT_FOUND, format!("No flow {id}")))?;
+ let body = bodies::read_body(&ctx.blobs, &flow, which, ctx.capture_body_limit()?).await?;
+ let content_type =
+ find_header(flow.headers(which), "content-type").unwrap_or("application/octet-stream");
+ let mut resp = response(StatusCode::OK, content_type, body.data);
+ resp.headers_mut().insert(TRUNCATED_HEADER, body.truncated.to_string().parse().unwrap());
+ Ok(resp)
+}
+
+fn body_limit(query: &str) -> Result {
+ for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
+ if key == "bodyLimit" {
+ return value.parse().map_err(|_| bad_request(format!("Invalid bodyLimit {value:?}")));
+ }
+ }
+ Ok(DEFAULT_BODY_PREVIEW_LIMIT)
+}
+
+async fn blocking(
+ f: impl FnOnce() -> Result + Send + 'static,
+) -> Result {
+ tokio::task::spawn_blocking(f).await.map_err(|e| e.to_string())?
+}
+
+/// The JSON request body; an empty one reads as `{}`.
+async fn read_json(req: Request) -> Result {
+ let bytes = Limited::new(req.into_body(), MAX_REQUEST_BODY)
+ .collect()
+ .await
+ .map_err(|e| bad_request(format!("Failed to read the request body: {e}")))?
+ .to_bytes();
+ let bytes: &[u8] = if bytes.iter().all(u8::is_ascii_whitespace) { b"{}" } else { &bytes };
+ serde_json::from_slice(bytes).map_err(|e| bad_request(format!("Invalid JSON body: {e}")))
+}
+
+fn bad_request(message: String) -> ApiError {
+ ApiError::new(StatusCode::BAD_REQUEST, message)
+}
+
+fn json(value: &T) -> Result {
+ let body = serde_json::to_vec(value).map_err(|e| e.to_string())?;
+ Ok(response(StatusCode::OK, "application/json", body))
+}
+
+fn error_response(e: ApiError) -> ControlResponse {
+ let body = serde_json::to_vec(&ErrorBody { error: e.message }).unwrap_or_default();
+ response(e.status, "application/json", body)
+}
+
+fn response(status: StatusCode, content_type: &str, body: impl Into) -> ControlResponse {
+ let mut resp = Response::new(Full::new(body.into()));
+ *resp.status_mut() = status;
+ if let Ok(value) = content_type.parse() {
+ resp.headers_mut().insert(CONTENT_TYPE, value);
+ }
+ resp
+}
diff --git a/crates-proxy/yaak-proxy-lib/src/convert.rs b/crates-proxy/yaak-proxy-lib/src/convert.rs
new file mode 100644
index 000000000..00c6f6622
--- /dev/null
+++ b/crates-proxy/yaak-proxy-lib/src/convert.rs
@@ -0,0 +1,469 @@
+//! A captured flow as a Yaak request. Headers Yaak sets itself when sending, and hop-by-hop
+//! headers that only meant something between the app and the proxy, are dropped and listed so
+//! the save dialog can say why. Authorization is kept as the plain header it was captured as.
+
+use crate::bodies::DecodedBody;
+use crate::models::{Flow, find_header};
+use serde::{Deserialize, Serialize};
+use serde_json::{Value, json};
+use std::collections::BTreeMap;
+use yaak_models::models::{HttpRequest, HttpRequestHeader, HttpUrlParameter};
+
+// The client's body types (`apps/yaak-client/lib/model_util.ts`).
+const BODY_TYPE_JSON: &str = "application/json";
+const BODY_TYPE_XML: &str = "text/xml";
+const BODY_TYPE_GRAPHQL: &str = "graphql";
+const BODY_TYPE_FORM_URLENCODED: &str = "application/x-www-form-urlencoded";
+const BODY_TYPE_OTHER: &str = "other";
+
+/// Removed between the app and the proxy; never forwarded as they are.
+const HOP_BY_HOP: &[&str] = &[
+ "connection",
+ "keep-alive",
+ "te",
+ "trailer",
+ "transfer-encoding",
+ "upgrade",
+];
+/// Yaak works these out itself when it sends the request.
+const AUTOMATIC: &[&str] = &["host", "content-length", "accept-encoding"];
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DroppedHeader {
+ pub name: String,
+ pub value: String,
+ pub reason: DropReason,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum DropReason {
+ /// Only meant for the hop between the app and the proxy.
+ HopByHop,
+ /// Yaak sets it when it sends the request.
+ SetBySending,
+ /// The body is saved decoded, so its encoding no longer applies.
+ BodyDecoded,
+}
+
+#[derive(Debug, Clone)]
+pub struct Conversion {
+ pub request: HttpRequest,
+ pub dropped_headers: Vec,
+ /// What the saved request does not carry over exactly, one sentence each.
+ pub warnings: Vec,
+}
+
+/// `flow` as a request in no workspace yet, with `body` its request body decoded.
+pub fn flow_to_request(flow: &Flow, body: &DecodedBody) -> Conversion {
+ let mut warnings = Vec::new();
+ let (url, url_parameters) = split_query(&flow.url);
+ let (headers, dropped_headers) = split_headers(flow, body);
+ let has_authorization = headers.iter().any(|h| h.name.eq_ignore_ascii_case("authorization"));
+ if has_authorization {
+ warnings.push("The Authorization header is saved as captured, in plain text.".into());
+ }
+ let (body_type, body_fields) = request_body(flow, body, &mut warnings);
+ if body.truncated {
+ warnings.push(format!(
+ "The body was cut at the capture limit, so only its first {} bytes are saved.",
+ body.data.len()
+ ));
+ }
+
+ let request = HttpRequest {
+ name: default_name(&url),
+ method: flow.method.clone(),
+ url,
+ url_parameters,
+ headers,
+ body_type,
+ body: body_fields,
+ // No auth rather than inherited, so a folder's or workspace's auth cannot replace the
+ // captured header when the request is sent.
+ authentication_type: has_authorization.then(|| "none".to_string()),
+ ..Default::default()
+ };
+ Conversion { request, dropped_headers, warnings }
+}
+
+/// The URL without its query, and the query as parameters.
+fn split_query(url: &str) -> (String, Vec) {
+ let (url, fragment) = match url.split_once('#') {
+ Some((url, fragment)) => (url, Some(fragment)),
+ None => (url, None),
+ };
+ let Some((base, query)) = url.split_once('?') else {
+ return (url.to_string() + &fragment.map(|f| format!("#{f}")).unwrap_or_default(), vec![]);
+ };
+ let params = url::form_urlencoded::parse(query.as_bytes())
+ .map(|(name, value)| HttpUrlParameter {
+ enabled: true,
+ name: name.into_owned(),
+ value: value.into_owned(),
+ id: None,
+ })
+ .collect();
+ (base.to_string() + &fragment.map(|f| format!("#{f}")).unwrap_or_default(), params)
+}
+
+fn split_headers(flow: &Flow, body: &DecodedBody) -> (Vec, Vec) {
+ // Connection names further headers that are hop-by-hop for this request alone.
+ let listed: Vec = flow
+ .request_headers
+ .iter()
+ .filter(|h| h.name.eq_ignore_ascii_case("connection"))
+ .flat_map(|h| h.value.split(','))
+ .map(|n| n.trim().to_ascii_lowercase())
+ .filter(|n| !n.is_empty())
+ .collect();
+ let encoding = find_header(&flow.request_headers, "content-encoding");
+ // A body that decodes to nothing was still decoded.
+ let decoded = body.stored_size > 0
+ && encoding.is_some_and(|e| !e.trim().eq_ignore_ascii_case("identity"));
+
+ let mut kept = Vec::new();
+ let mut dropped = Vec::new();
+ for h in &flow.request_headers {
+ let name = h.name.to_ascii_lowercase();
+ let reason = if HOP_BY_HOP.contains(&name.as_str())
+ || name.starts_with("proxy-")
+ || name.starts_with(':')
+ || listed.contains(&name)
+ {
+ Some(DropReason::HopByHop)
+ } else if AUTOMATIC.contains(&name.as_str()) {
+ Some(DropReason::SetBySending)
+ } else if name == "content-encoding" && decoded {
+ Some(DropReason::BodyDecoded)
+ } else {
+ None
+ };
+ match reason {
+ Some(reason) => {
+ dropped.push(DroppedHeader { name: h.name.clone(), value: h.value.clone(), reason })
+ }
+ None => kept.push(HttpRequestHeader {
+ enabled: true,
+ name: h.name.clone(),
+ value: h.value.clone(),
+ id: None,
+ }),
+ }
+ }
+ (kept, dropped)
+}
+
+fn request_body(
+ flow: &Flow,
+ body: &DecodedBody,
+ warnings: &mut Vec,
+) -> (Option, BTreeMap) {
+ if body.data.is_empty() {
+ return (None, BTreeMap::new());
+ }
+ let content_type = find_header(&flow.request_headers, "content-type").unwrap_or_default();
+ let mime = content_type.split(';').next().unwrap_or_default().trim().to_ascii_lowercase();
+ let Ok(text) = std::str::from_utf8(&body.data) else {
+ warnings.push(
+ "The body is binary and is saved as text, so bytes that are not UTF-8 are replaced."
+ .into(),
+ );
+ return text_body(BODY_TYPE_OTHER, &String::from_utf8_lossy(&body.data));
+ };
+
+ if mime == "application/json" || mime.ends_with("+json") {
+ if let Some(graphql) = graphql_body(flow, text) {
+ return (Some(BODY_TYPE_GRAPHQL.into()), graphql);
+ }
+ return text_body(BODY_TYPE_JSON, text);
+ }
+ if mime == "application/xml" || mime == "text/xml" || mime.ends_with("+xml") {
+ return text_body(BODY_TYPE_XML, text);
+ }
+ if mime == BODY_TYPE_FORM_URLENCODED {
+ let form: Vec = url::form_urlencoded::parse(text.as_bytes())
+ .map(|(name, value)| json!({ "enabled": true, "name": name, "value": value }))
+ .collect();
+ return (
+ Some(BODY_TYPE_FORM_URLENCODED.into()),
+ BTreeMap::from([("form".into(), json!(form))]),
+ );
+ }
+ if mime == "multipart/form-data" {
+ warnings.push(
+ "The multipart body is saved as text with its Content-Type, not as form fields.".into(),
+ );
+ }
+ text_body(BODY_TYPE_OTHER, text)
+}
+
+fn text_body(body_type: &str, text: &str) -> (Option, BTreeMap) {
+ (Some(body_type.into()), BTreeMap::from([("text".into(), json!(text))]))
+}
+
+/// A JSON body posted to a GraphQL endpoint, as the client's GraphQL body.
+fn graphql_body(flow: &Flow, text: &str) -> Option> {
+ let path = flow.url.split('?').next().unwrap_or_default();
+ if !path.to_ascii_lowercase().ends_with("/graphql") {
+ return None;
+ }
+ let Value::Object(body) = serde_json::from_str::(text).ok()? else {
+ return None;
+ };
+ // Anything else, such as persisted-query `extensions`, would be lost in a GraphQL body.
+ if body.keys().any(|k| !matches!(k.as_str(), "query" | "variables" | "operationName")) {
+ return None;
+ }
+ let query = body.get("query")?.as_str()?;
+ let mut fields = BTreeMap::from([("query".to_string(), json!(query))]);
+ if let Some(variables) = body.get("variables").filter(|v| !v.is_null()) {
+ let pretty = serde_json::to_string_pretty(variables).ok()?;
+ fields.insert("variables".into(), json!(pretty));
+ }
+ if let Some(name) = body.get("operationName").and_then(Value::as_str) {
+ fields.insert("operationName".into(), json!(name));
+ }
+ Some(fields)
+}
+
+/// The path, or the host for a request to the root.
+fn default_name(url: &str) -> String {
+ let rest = url.split_once("://").map_or(url, |(_, rest)| rest);
+ match rest.split_once('/') {
+ Some((_, path)) if !path.is_empty() => format!("/{path}"),
+ Some((host, _)) => host.to_string(),
+ None => rest.to_string(),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::models::ProxyHeader;
+
+ fn flow(method: &str, url: &str, headers: &[(&str, &str)]) -> Flow {
+ Flow {
+ method: method.into(),
+ url: url.into(),
+ request_headers: headers
+ .iter()
+ .map(|(n, v)| ProxyHeader { name: (*n).into(), value: (*v).into() })
+ .collect(),
+ ..Default::default()
+ }
+ }
+
+ fn body(data: &str) -> DecodedBody {
+ DecodedBody {
+ data: data.as_bytes().to_vec(),
+ stored_size: data.len(),
+ truncated: false,
+ decode_error: None,
+ }
+ }
+
+ fn names(request: &HttpRequest) -> Vec<&str> {
+ request.headers.iter().map(|h| h.name.as_str()).collect()
+ }
+
+ #[test]
+ fn get_without_body() {
+ let f = flow(
+ "GET",
+ "https://api.github.com/search/issues?q=is%3Aopen+label%3Abug&per_page=5",
+ &[
+ ("Host", "api.github.com"),
+ ("Accept", "application/json"),
+ ("Accept-Encoding", "gzip, br"),
+ ("Connection", "keep-alive, X-Debug"),
+ ("X-Debug", "1"),
+ ("Proxy-Authorization", "Basic Zm9v"),
+ ("Authorization", "Bearer ghp_secret"),
+ ],
+ );
+ let c = flow_to_request(&f, &body(""));
+ let r = &c.request;
+ assert_eq!(r.method, "GET");
+ assert_eq!(r.url, "https://api.github.com/search/issues");
+ assert_eq!(r.name, "/search/issues");
+ let params: Vec<(&str, &str)> =
+ r.url_parameters.iter().map(|p| (p.name.as_str(), p.value.as_str())).collect();
+ assert_eq!(params, vec![("q", "is:open label:bug"), ("per_page", "5")]);
+ assert_eq!(names(r), vec!["Accept", "Authorization"]);
+ assert_eq!(r.headers[1].value, "Bearer ghp_secret");
+ assert_eq!(r.authentication_type.as_deref(), Some("none"));
+ assert_eq!(r.body_type, None);
+ assert!(r.body.is_empty());
+
+ let dropped: Vec<(&str, DropReason)> =
+ c.dropped_headers.iter().map(|d| (d.name.as_str(), d.reason)).collect();
+ assert_eq!(
+ dropped,
+ vec![
+ ("Host", DropReason::SetBySending),
+ ("Accept-Encoding", DropReason::SetBySending),
+ ("Connection", DropReason::HopByHop),
+ ("X-Debug", DropReason::HopByHop),
+ ("Proxy-Authorization", DropReason::HopByHop),
+ ]
+ );
+ assert_eq!(
+ c.warnings,
+ vec!["The Authorization header is saved as captured, in plain text."]
+ );
+ }
+
+ #[test]
+ fn json_body() {
+ let f = flow(
+ "POST",
+ "https://api.stripe.com/v1/charges",
+ &[
+ ("Content-Type", "application/json; charset=utf-8"),
+ ("Content-Length", "13"),
+ ],
+ );
+ let c = flow_to_request(&f, &body(r#"{"amount":42}"#));
+ assert_eq!(c.request.body_type.as_deref(), Some("application/json"));
+ assert_eq!(c.request.body["text"], r#"{"amount":42}"#);
+ assert_eq!(names(&c.request), vec!["Content-Type"]);
+ assert!(c.warnings.is_empty());
+ // Without a captured Authorization, the request inherits its folder's auth.
+ assert_eq!(c.request.authentication_type, None);
+ }
+
+ #[test]
+ fn vendor_json_body() {
+ let f = flow("PATCH", "https://x.test/a", &[("Content-Type", "application/vnd.api+json")]);
+ let c = flow_to_request(&f, &body("{}"));
+ assert_eq!(c.request.body_type.as_deref(), Some("application/json"));
+ }
+
+ #[test]
+ fn graphql_body() {
+ let f =
+ flow("POST", "https://api.github.com/graphql", &[("Content-Type", "application/json")]);
+ let c = flow_to_request(
+ &f,
+ &body(
+ r#"{"query":"query($n:Int){viewer{login}}","variables":{"n":1},"operationName":"Me"}"#,
+ ),
+ );
+ assert_eq!(c.request.body_type.as_deref(), Some("graphql"));
+ assert_eq!(c.request.body["query"], "query($n:Int){viewer{login}}");
+ assert_eq!(c.request.body["variables"], "{\n \"n\": 1\n}");
+ assert_eq!(c.request.body["operationName"], "Me");
+
+ // Fields a GraphQL body cannot hold keep it plain JSON.
+ let persisted = r#"{"query":"q","extensions":{"persistedQuery":{"version":1}}}"#;
+ let c = flow_to_request(&f, &body(persisted));
+ assert_eq!(c.request.body_type.as_deref(), Some("application/json"));
+
+ // The same body anywhere else is plain JSON.
+ let f = flow("POST", "https://x.test/api", &[("Content-Type", "application/json")]);
+ let c = flow_to_request(&f, &body(r#"{"query":"q"}"#));
+ assert_eq!(c.request.body_type.as_deref(), Some("application/json"));
+ }
+
+ #[test]
+ fn xml_body() {
+ let f = flow("POST", "https://x.test/soap", &[("Content-Type", "application/soap+xml")]);
+ let c = flow_to_request(&f, &body(""));
+ assert_eq!(c.request.body_type.as_deref(), Some("text/xml"));
+ assert_eq!(c.request.body["text"], "");
+ }
+
+ #[test]
+ fn form_body() {
+ let f = flow(
+ "POST",
+ "http://localhost:3000/login",
+ &[("Content-Type", "application/x-www-form-urlencoded")],
+ );
+ let c = flow_to_request(&f, &body("user=greg¬e=hello+world%21"));
+ assert_eq!(c.request.body_type.as_deref(), Some("application/x-www-form-urlencoded"));
+ assert_eq!(
+ c.request.body["form"],
+ json!([
+ { "enabled": true, "name": "user", "value": "greg" },
+ { "enabled": true, "name": "note", "value": "hello world!" },
+ ])
+ );
+ assert_eq!(c.request.name, "/login");
+ }
+
+ #[test]
+ fn multipart_body() {
+ let content_type = "multipart/form-data; boundary=XyZ";
+ let f = flow("POST", "https://x.test/upload", &[("Content-Type", content_type)]);
+ let text = "--XyZ\r\nContent-Disposition: form-data; name=\"a\"\r\n\r\n1\r\n--XyZ--\r\n";
+ let c = flow_to_request(&f, &body(text));
+ assert_eq!(c.request.body_type.as_deref(), Some("other"));
+ assert_eq!(c.request.body["text"], text);
+ assert_eq!(c.request.headers[0].value, content_type);
+ assert_eq!(c.warnings.len(), 1);
+ }
+
+ #[test]
+ fn text_body() {
+ let f = flow("PUT", "https://x.test/", &[("Content-Type", "text/plain")]);
+ let c = flow_to_request(&f, &body("hello"));
+ assert_eq!(c.request.body_type.as_deref(), Some("other"));
+ assert_eq!(c.request.body["text"], "hello");
+ assert_eq!(c.request.name, "x.test");
+ }
+
+ #[test]
+ fn binary_body() {
+ let f = flow("POST", "https://x.test/bin", &[("Content-Type", "application/octet-stream")]);
+ let data = DecodedBody {
+ data: vec![b'a', 0xff, b'b'],
+ stored_size: 3,
+ truncated: false,
+ decode_error: None,
+ };
+ let c = flow_to_request(&f, &data);
+ assert_eq!(c.request.body_type.as_deref(), Some("other"));
+ assert_eq!(c.request.body["text"], "a\u{fffd}b");
+ assert_eq!(c.warnings.len(), 1);
+ }
+
+ #[test]
+ fn decoded_empty_body() {
+ let f = flow("POST", "https://x.test/", &[("Content-Encoding", "gzip")]);
+ // A gzip stream of nothing: stored bytes, but an empty body once decoded.
+ let data =
+ DecodedBody { data: Vec::new(), stored_size: 20, truncated: false, decode_error: None };
+ let c = flow_to_request(&f, &data);
+ assert!(c.request.headers.is_empty());
+ assert_eq!(c.dropped_headers[0].reason, DropReason::BodyDecoded);
+ assert_eq!(c.request.body_type, None);
+ }
+
+ #[test]
+ fn decoded_and_truncated_body() {
+ let f = flow(
+ "POST",
+ "https://x.test/logs",
+ &[
+ ("Content-Type", "application/json"),
+ ("Content-Encoding", "gzip"),
+ ],
+ );
+ let data = DecodedBody {
+ data: b"[1,2".to_vec(),
+ stored_size: 4,
+ truncated: true,
+ decode_error: None,
+ };
+ let c = flow_to_request(&f, &data);
+ assert_eq!(names(&c.request), vec!["Content-Type"]);
+ assert_eq!(c.dropped_headers[0].reason, DropReason::BodyDecoded);
+ assert_eq!(
+ c.warnings,
+ vec!["The body was cut at the capture limit, so only its first 4 bytes are saved."]
+ );
+ }
+}
diff --git a/crates-proxy/yaak-proxy-lib/src/db.rs b/crates-proxy/yaak-proxy-lib/src/db.rs
index 8e71c5e46..9bccc7e4e 100644
--- a/crates-proxy/yaak-proxy-lib/src/db.rs
+++ b/crates-proxy/yaak-proxy-lib/src/db.rs
@@ -1,10 +1,11 @@
use include_dir::{Dir, include_dir};
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
-use rusqlite::{Transaction, TransactionBehavior};
+use rusqlite::{OpenFlags, Transaction, TransactionBehavior};
use std::path::Path;
use std::time::Duration;
use yaak_database::{ConnectionOrTx, DbContext, Error, Result, SqlitePool, run_migrations};
+use yaak_models::blob_manager::BlobManager;
static MIGRATIONS: Dir<'static> = include_dir!("$CARGO_MANIFEST_DIR/migrations");
@@ -26,6 +27,13 @@ impl ProxyQueryManager {
Ok(Self { readers: file_pool(db_path, 5)?, writer })
}
+ /// `proxy.db` for reading while another process may be writing it, or none is. Writes
+ /// fail. No migrations run, so an older database reads as it is.
+ pub fn open_read_only(db_path: &Path) -> Result {
+ let readers = read_only_pool(db_path)?;
+ Ok(Self { writer: readers.clone(), readers })
+ }
+
pub fn with_conn(&self, func: F) -> T
where
F: FnOnce(&DbContext) -> T,
@@ -62,6 +70,23 @@ impl ProxyQueryManager {
}
}
+/// `blobs.sqlite` for reading, alongside [`ProxyQueryManager::open_read_only`].
+pub fn open_blobs_read_only(path: &Path) -> Result {
+ let readers = read_only_pool(path)?;
+ Ok(BlobManager::new(readers.clone(), readers))
+}
+
+fn read_only_pool(path: &Path) -> Result {
+ let manager = SqliteConnectionManager::file(path)
+ .with_flags(OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX)
+ .with_init(|conn| conn.busy_timeout(Duration::from_millis(5000)));
+ Pool::builder()
+ .max_size(2)
+ .connection_timeout(Duration::from_secs(10))
+ .build(manager)
+ .map_err(|e| Error::Database(e.to_string()))
+}
+
fn file_pool(path: &Path, max_size: u32) -> Result {
let manager = SqliteConnectionManager::file(path).with_init(|conn| {
conn.pragma_update(None, "journal_mode", "WAL")?;
@@ -85,6 +110,28 @@ mod tests {
stmt.query_map([], |r| r.get::<_, String>("name")).unwrap().map(|c| c.unwrap()).collect()
}
+ #[test]
+ fn reads_flows_read_only_before_migrations() {
+ use crate::models::Flow;
+ let dir = tempfile::tempdir().unwrap();
+ let path = dir.path().join("proxy.db");
+ drop(ProxyQueryManager::new(&path).unwrap());
+
+ // As a build from before the saved request column left it.
+ let conn = Connection::open(&path).unwrap();
+ conn.execute_batch(
+ "ALTER TABLE flows DROP COLUMN saved_request_id;
+ INSERT INTO flows (id, url, method) VALUES ('fl_old', 'https://example.com', 'GET');",
+ )
+ .unwrap();
+ drop(conn);
+
+ let db = ProxyQueryManager::open_read_only(&path).unwrap();
+ let flows = db.with_conn(|c| c.find_all::()).unwrap();
+ assert_eq!(flows.len(), 1);
+ assert_eq!((flows[0].id.as_str(), flows[0].saved_request_id.as_deref()), ("fl_old", None));
+ }
+
#[test]
fn migrates_an_old_schema_db() {
let dir = tempfile::tempdir().unwrap();
@@ -124,8 +171,7 @@ mod tests {
assert!(flow_columns.contains(&"response_truncated".to_string()));
assert!(!flow_columns.iter().any(|c| c.contains("body")), "{flow_columns:?}");
- let journal_mode: String =
- conn.query_row("PRAGMA journal_mode", [], |r| r.get(0)).unwrap();
+ let journal_mode: String = conn.query_row("PRAGMA journal_mode", [], |r| r.get(0)).unwrap();
assert_eq!(journal_mode, "wal");
}
}
diff --git a/crates-proxy/yaak-proxy-lib/src/flows.rs b/crates-proxy/yaak-proxy-lib/src/flows.rs
new file mode 100644
index 000000000..840ff7554
--- /dev/null
+++ b/crates-proxy/yaak-proxy-lib/src/flows.rs
@@ -0,0 +1,305 @@
+use crate::bodies;
+use crate::db::ProxyQueryManager;
+use crate::models::{Flow, FlowBody, FlowIden};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use serde::{Deserialize, Serialize};
+use yaak_models::blob_manager::BlobManager;
+
+/// Which flows a listing or a HAR export covers. Every field narrows it; none is all of them.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(default, rename_all = "camelCase")]
+pub struct FlowFilter {
+ /// Part of the host, any case: `stripe` matches `api.stripe.com`.
+ pub host: Option,
+ /// The method, any case.
+ pub method: Option,
+ /// A status (`404`), a class of them (`4xx`), or `error` for flows that failed.
+ pub status: Option,
+ /// At most this many flows, the newest.
+ pub limit: Option,
+ /// Only flows recorded after the flow with this id, for following new traffic. Flows are
+ /// recorded when they end, so a slow one that started earlier still comes through.
+ pub after: Option,
+}
+
+impl FlowFilter {
+ /// The filter from a query string's `host`, `method`, `status`, `limit` and `after`.
+ pub fn from_query(query: &str) -> Result {
+ let mut filter = Self::default();
+ for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
+ let value = Some(value.into_owned()).filter(|v| !v.is_empty());
+ match key.as_ref() {
+ "host" => filter.host = value,
+ "method" => filter.method = value,
+ "status" => filter.status = value,
+ "after" => filter.after = value,
+ "limit" => {
+ filter.limit = value
+ .map(|v| v.parse().map_err(|_| format!("Invalid limit {v:?}")))
+ .transpose()?
+ }
+ _ => {}
+ }
+ }
+ Ok(filter)
+ }
+
+ pub fn to_query(&self) -> String {
+ let mut query = url::form_urlencoded::Serializer::new(String::new());
+ let limit = self.limit.map(|l| l.to_string());
+ for (key, value) in [
+ ("host", &self.host),
+ ("method", &self.method),
+ ("status", &self.status),
+ ("limit", &limit),
+ ("after", &self.after),
+ ] {
+ if let Some(value) = value {
+ query.append_pair(key, value);
+ }
+ }
+ query.finish()
+ }
+
+ pub fn matches(&self, flow: &Flow) -> bool {
+ if let Some(host) = &self.host
+ && !host_of(&flow.url).to_ascii_lowercase().contains(&host.to_ascii_lowercase())
+ {
+ return false;
+ }
+ if let Some(method) = &self.method
+ && !flow.method.eq_ignore_ascii_case(method)
+ {
+ return false;
+ }
+ match self.status.as_deref().map(str::trim) {
+ None | Some("") => true,
+ Some(s) if s.eq_ignore_ascii_case("error") => flow.error.is_some(),
+ Some(s) => match (flow.status, s.to_ascii_lowercase().strip_suffix("xx")) {
+ (Some(status), Some(class)) => class.parse() == Ok(status / 100),
+ (Some(status), None) => s.parse() == Ok(status),
+ (None, _) => false,
+ },
+ }
+ }
+}
+
+/// `host[:port]` of a captured URL.
+fn host_of(url: &str) -> &str {
+ let rest = url.split_once("://").map_or(url, |(_, rest)| rest);
+ rest.split(['/', '?', '#']).next().unwrap_or_default()
+}
+
+/// The flows `filter` covers, newest first.
+pub fn list(db: &ProxyQueryManager, filter: &FlowFilter) -> Result, String> {
+ db.with_conn(|c| {
+ let flows = c.find_all::().map_err(|e| e.to_string())?;
+ let recorded_after = match &filter.after {
+ Some(id) => Some(recorded_after(c.conn().resolve(), id)?),
+ None => None,
+ };
+ Ok(flows
+ .into_iter()
+ .filter(|f| recorded_after.as_ref().is_none_or(|ids| ids.contains(&f.id)))
+ .filter(|f| filter.matches(f))
+ .take(filter.limit.unwrap_or(usize::MAX))
+ .collect())
+ })
+}
+
+/// Ids of the flows inserted after flow `id`, by rowid, which only grows as flows are written.
+fn recorded_after(
+ conn: &rusqlite::Connection,
+ id: &str,
+) -> Result, String> {
+ let rowid: i64 = conn
+ .query_row("SELECT rowid FROM flows WHERE id = ?", [id], |r| r.get(0))
+ .map_err(|_| format!("No flow {id}"))?;
+ let mut stmt =
+ conn.prepare("SELECT id FROM flows WHERE rowid > ?").map_err(|e| e.to_string())?;
+ let ids = stmt.query_map([rowid], |r| r.get(0)).map_err(|e| e.to_string())?;
+ ids.collect::>().map_err(|e| e.to_string())
+}
+
+pub fn find(db: &ProxyQueryManager, flow_id: &str) -> Option {
+ db.with_conn(|c| c.find_optional::(FlowIden::Id, flow_id))
+}
+
+pub fn get(db: &ProxyQueryManager, flow_id: &str) -> Result {
+ find(db, flow_id).ok_or_else(|| format!("No flow {flow_id}"))
+}
+
+pub fn count(db: &ProxyQueryManager) -> Result {
+ db.with_conn(|c| {
+ c.conn()
+ .resolve()
+ .query_row("SELECT COUNT(*) FROM flows", [], |r| r.get::<_, i64>(0))
+ .map(|n| n as usize)
+ .map_err(|e| e.to_string())
+ })
+}
+
+/// A body in JSON: text when it is UTF-8, base64 when it is not.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BodyPreview {
+ pub data: String,
+ pub encoding: BodyEncoding,
+ /// Bytes in `data` once decoded.
+ pub size: usize,
+ /// The body is longer than `data`: cut at the capture limit or at the preview cap.
+ pub truncated: bool,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum BodyEncoding {
+ Utf8,
+ Base64,
+}
+
+impl BodyPreview {
+ pub fn new(data: &[u8], truncated: bool) -> Self {
+ let (data, encoding) = match std::str::from_utf8(data) {
+ Ok(text) => (text.to_string(), BodyEncoding::Utf8),
+ // A cut can split a character; that alone does not make the body binary.
+ Err(e) if truncated && e.error_len().is_none() => {
+ (String::from_utf8_lossy(&data[..e.valid_up_to()]).into_owned(), BodyEncoding::Utf8)
+ }
+ Err(_) => (BASE64.encode(data), BodyEncoding::Base64),
+ };
+ let size = match encoding {
+ BodyEncoding::Utf8 => data.len(),
+ BodyEncoding::Base64 => BASE64.decode(&data).map(|d| d.len()).unwrap_or_default(),
+ };
+ Self { data, encoding, size, truncated }
+ }
+}
+
+/// One flow with its bodies, each decoded and cut at `body_cap` bytes.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FlowDetail {
+ pub flow: Flow,
+ pub request_body: Option,
+ pub response_body: Option,
+}
+
+pub async fn detail(
+ db: &ProxyQueryManager,
+ blobs: &BlobManager,
+ flow_id: &str,
+ body_cap: usize,
+) -> Result {
+ let flow = get(db, flow_id)?;
+ let request_body = preview(blobs, &flow, FlowBody::Request, body_cap).await?;
+ let response_body = preview(blobs, &flow, FlowBody::Response, body_cap).await?;
+ Ok(FlowDetail { flow, request_body, response_body })
+}
+
+async fn preview(
+ blobs: &BlobManager,
+ flow: &Flow,
+ which: FlowBody,
+ cap: usize,
+) -> Result