Skip to content

feat: bind proxy IPC to sidecar process #54

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kev1n77 merged 2 commits into from
Jul 18, 2026
Merged

feat: bind proxy IPC to sidecar process #54

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
139059d
feat: bind proxy IPC to sidecar process
kev1n77 Jul 18, 2026
45a390f
fix: keep IPC peer connection alive for verification
kev1n77 Jul 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ cargo run -p codeischeap-desktop-api --bin export-desktop-contract

捕获范围与敏感字段由 `policies/capture-policy.v0.1.json` 定义，公开 schema 位于 `schemas/capture-policy/v0.1.schema.json`。Python sidecar 在 IPC 前执行策略，`crates/core` 在进入持久化前再次拒绝越界请求并删除遗漏凭据。

sidecar IPC 协议为 `0.3`：仅监听 loopback，使用每次 Proxy 会话重新生成的 256-bit token，认证帧限制为 1 KiB，并要求 `mitmproxy` 来源声明；已接受连接必须在 2 秒内完成认证与数据帧，否则释放连接。认证帧可携带严格校验的临时 loopback 端点用于进程归因，端点不会进入 CaptureEnvelope、持久化或导出。sidecar manifest 分别声明 IPC、Envelope 和 Policy 版本，不兼容 bundle 会在启动前被拒绝。
sidecar IPC 协议为 `0.4`：仅监听 loopback，使用每次 Proxy 会话重新生成的 256-bit token，认证帧限制为 1 KiB，并要求 `mitmproxy` 来源声明；认证、数据帧、Windows/macOS 首连接 sidecar PID 校验与服务端 ACK 共用 2 秒截止时间，sidecar 在收到 ACK 前保持连接。认证帧可携带严格校验的临时 loopback 端点用于进程归因，端点不会进入 CaptureEnvelope、持久化或导出。sidecar manifest 分别声明 IPC、Envelope 和 Policy 版本，不兼容 bundle 会在启动前被拒绝。

Gateway 和 Proxy 捕获会按显式客户端标签、User-Agent 规则或捕获模式回退生成带置信度的应用归因。Gateway 客户端可设置 `x-codeischeap-client: <application>` 提供高置信度标签；该内部请求头会在持久化和转发上游前删除。Windows 与 macOS 上的 Gateway 和 Proxy 会用操作系统 TCP 连接表精确匹配客户端 PID，查询失败时保持未知，不做启发式推断；sidecar 自报 PID 会被忽略，用于匹配的临时 socket 端点不会进入持久化或导出。未知客户端会明确显示为低置信度的 `Gateway client` 或 `Proxy client`。

Expand Down
179 changes: 143 additions & 36 deletions apps/desktop/src-tauri/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use std::os::unix::fs::{OpenOptionsExt as _, PermissionsExt as _};

use codeischeap_adapters::AdapterRegistry;
use codeischeap_capture_ipc::{CaptureEnvelope, CaptureTransport, receive_one_with_transport};
use codeischeap_capture_ipc::{
CaptureEnvelope, CaptureTransport, receive_one_with_transport,
receive_one_with_transport_verified,
};
use codeischeap_capture_policy::CapturePolicy;
use codeischeap_core::{
GatewayCaptureError, GatewayCaptureOutcome, IngestError, ingest_envelope, persist_capture,
Expand Down Expand Up @@ -114,6 +117,7 @@ struct ProxyCaptureContext {
listener: TcpListener,
token: String,
policy: CapturePolicy,
sidecar_process_id: u32,
}

#[derive(Default)]
Expand Down Expand Up @@ -149,6 +153,22 @@ impl ProcessIdCache {
}
}

struct ProxyCaptureSession {
process_ids: ProcessIdCache,
sidecar_process_id: u32,
peer_verified: bool,
}

impl ProxyCaptureSession {
fn new(sidecar_process_id: u32) -> Self {
Self {
process_ids: ProcessIdCache::default(),
sidecar_process_id,
peer_verified: false,
}
}
}

impl ProxyRuntime {
fn shutdown(mut self) -> Result<(), String> {
if let Some(shutdown) = self.shutdown.take() {
Expand Down Expand Up @@ -1153,6 +1173,9 @@ async fn ensure_proxy(app: &AppHandle, state: &DesktopState) -> Result<(), Strin
.await
.map_err(|error| format!("explicit proxy launch task failed: {error}"))?
.map_err(|error| error.to_string())?;
let sidecar_process_id = process
.process_id()
.ok_or_else(|| "explicit proxy process ID is unavailable".to_owned())?;
let endpoint = format!("http://{}", process.endpoint());
let generation = state.next_proxy_generation.fetch_add(1, Ordering::Relaxed);
state.proxy_session_event_count.store(0, Ordering::Release);
Expand All @@ -1174,6 +1197,7 @@ async fn ensure_proxy(app: &AppHandle, state: &DesktopState) -> Result<(), Strin
listener,
token,
policy,
sidecar_process_id,
},
shutdown_rx,
));
Expand Down Expand Up @@ -1348,7 +1372,7 @@ fn generate_ipc_token() -> Result<String, String> {
async fn process_proxy_events(context: ProxyCaptureContext, mut shutdown: oneshot::Receiver<()>) {
let adapters = AdapterRegistry::default();
let mut captures_since_retention = 0_usize;
let mut process_ids = ProcessIdCache::default();
let mut session = ProxyCaptureSession::new(context.sidecar_process_id);
loop {
let result = tokio::select! {
_ = &mut shutdown => break,
Expand All @@ -1359,7 +1383,7 @@ async fn process_proxy_events(context: ProxyCaptureContext, mut shutdown: onesho
&adapters,
&context.store,
&context.capture_active,
&mut process_ids,
&mut session,
) => result,
};
match result {
Expand Down Expand Up @@ -1404,16 +1428,33 @@ async fn receive_and_persist_proxy_capture(
adapters: &AdapterRegistry,
store: &SharedStore,
capture_active: &AtomicBool,
process_ids: &mut ProcessIdCache,
session: &mut ProxyCaptureSession,
) -> Result<Option<String>, ProxyCaptureError> {
let received = receive_one_with_transport(listener, token)
.await
.map_err(IngestError::from)?;
let received = if session.peer_verified {
receive_one_with_transport(listener, token).await
} else {
let expected_process_id = session.sidecar_process_id;
let received =
receive_one_with_transport_verified(listener, token, move |peer, server| async move {
sidecar_peer_matches(peer, server, expected_process_id).await
})
.await;
if received.is_ok() {
session.peer_verified = true;
}
received
}
.map_err(IngestError::from)?;
let mut envelope = received.envelope;
let capture_id = envelope.capture_id.clone();
let is_outcome = envelope.outcome.is_some();
let resolved_process_id = resolve_proxy_process(received.transport).await;
apply_proxy_process_id(&mut envelope, policy, process_ids, resolved_process_id);
apply_proxy_process_id(
&mut envelope,
policy,
&mut session.process_ids,
resolved_process_id,
);
let capture = ingest_envelope(envelope, policy)?;
if !capture_active.load(Ordering::Acquire) {
return Ok(None);
Expand All @@ -1427,11 +1468,35 @@ async fn receive_and_persist_proxy_capture(
.ok_or(ProxyCaptureError::StoreUninitialized)?;
persist_capture(store, &capture, parsed.prompt_ir.as_ref())?;
if is_outcome {
process_ids.finish(&capture_id);
session.process_ids.finish(&capture_id);
}
Ok(Some(capture_id))
}

#[cfg(any(windows, target_os = "macos"))]
async fn sidecar_peer_matches(
peer: SocketAddr,
server: SocketAddr,
expected_process_id: u32,
) -> bool {
resolve_proxy_process(Some(CaptureTransport {
client_addr: peer,
server_addr: server,
}))
.await
== Some(expected_process_id)
}

#[cfg(not(any(windows, target_os = "macos")))]
async fn sidecar_peer_matches(
peer: SocketAddr,
server: SocketAddr,
expected_process_id: u32,
) -> bool {
let _ = (peer, server, expected_process_id);
true
}

async fn resolve_proxy_process(transport: Option<CaptureTransport>) -> Option<u32> {
let transport = transport?;
tokio::task::spawn_blocking(move || {
Expand Down Expand Up @@ -1695,6 +1760,8 @@ fn show_main_window(app: &tauri::AppHandle) {

#[cfg(test)]
mod tests {
#[cfg(any(windows, target_os = "macos"))]
use codeischeap_capture_ipc::IpcError;
use codeischeap_capture_ipc::{
AttributionConfidence, AttributionSource, CaptureAttribution, CaptureOutcome, CapturedBody,
CapturedBodyState, CapturedResponse, ResponseCompleteness,
Expand All @@ -1703,7 +1770,7 @@ mod tests {
use codeischeap_storage::DatabaseKey;
use std::net::TcpStream as StdTcpStream;
use tempfile::tempdir;
use tokio::io::AsyncWriteExt;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpStream;

use super::*;
Expand Down Expand Up @@ -1960,7 +2027,7 @@ mod tests {
let policy = CapturePolicy::load_default().expect("capture policy must load");
let adapters = AdapterRegistry::default();
let active = AtomicBool::new(true);
let mut process_ids = ProcessIdCache::default();
let mut session = ProxyCaptureSession::new(std::process::id());
let process_listener = StdTcpListener::bind((Ipv4Addr::LOCALHOST, 0))
.expect("process attribution listener must bind");
let process_server_addr = process_listener
Expand All @@ -1980,21 +2047,48 @@ mod tests {
))
.expect("request fixture must parse");

send_ipc_capture(address, &token, &request, Some(transport)).await;
assert_eq!(
#[cfg(any(windows, target_os = "macos"))]
{
let mut rejected_session = ProxyCaptureSession::new(u32::MAX);
let (acknowledged, result) = tokio::join!(
send_ipc_capture(address, &token, &request, Some(transport)),
receive_and_persist_proxy_capture(
&listener,
&token,
&policy,
&adapters,
&store,
&active,
&mut rejected_session,
)
);
let error = result.expect_err("an unexpected IPC process must be rejected");
assert!(!acknowledged);
assert!(matches!(
error,
ProxyCaptureError::Ingest(IngestError::Ipc(IpcError::UnauthorizedPeer))
));
assert!(!rejected_session.peer_verified);
}

let (acknowledged, result) = tokio::join!(
send_ipc_capture(address, &token, &request, Some(transport)),
receive_and_persist_proxy_capture(
&listener,
&token,
&policy,
&adapters,
&store,
&active,
&mut process_ids,
&mut session,
)
.await
.expect("request must persist"),
);
assert!(acknowledged);
assert_eq!(
result.expect("request must persist"),
Some(request.capture_id.clone())
);
assert!(session.peer_verified);

let mut response = request.clone();
response.outcome = Some(CaptureOutcome::Response(CapturedResponse {
Expand All @@ -2007,35 +2101,39 @@ mod tests {
duration_ms: 42,
completeness: ResponseCompleteness::Complete,
}));
send_ipc_capture(address, &token, &response, None).await;
receive_and_persist_proxy_capture(
&listener,
&token,
&policy,
&adapters,
&store,
&active,
&mut process_ids,
)
.await
.expect("response must persist");
let (acknowledged, result) = tokio::join!(
send_ipc_capture(address, &token, &response, None),
receive_and_persist_proxy_capture(
&listener,
&token,
&policy,
&adapters,
&store,
&active,
&mut session,
)
);
assert!(acknowledged);
result.expect("response must persist");

active.store(false, Ordering::Release);
let mut paused = request.clone();
paused.capture_id = "paused_capture".to_owned();
send_ipc_capture(address, &token, &paused, None).await;
assert_eq!(
let (acknowledged, result) = tokio::join!(
send_ipc_capture(address, &token, &paused, None),
receive_and_persist_proxy_capture(
&listener,
&token,
&policy,
&adapters,
&store,
&active,
&mut process_ids,
&mut session,
)
.await
.expect("paused event must be accepted and discarded"),
);
assert!(acknowledged);
assert_eq!(
result.expect("paused event must be accepted and discarded"),
None
);

Expand All @@ -2062,13 +2160,13 @@ mod tests {
token: &str,
envelope: &CaptureEnvelope,
transport: Option<CaptureTransport>,
) {
) -> bool {
let mut stream = TcpStream::connect(address)
.await
.expect("IPC client must connect");
let auth = serde_json::json!({
"protocol": "codeischeap.capture-ipc",
"version": "0.3",
"version": "0.4",
"origin": "mitmproxy",
"token": token,
"transport": transport,
Expand All @@ -2077,7 +2175,16 @@ mod tests {
.write_all(format!("{auth}\n{}\n", serde_json::to_string(envelope).unwrap()).as_bytes())
.await
.expect("IPC frames must write");
stream.shutdown().await.expect("IPC client must close");
let mut acknowledgement = String::new();
if BufReader::new(stream)
.read_line(&mut acknowledgement)
.await
.is_err()
{
return false;
}
serde_json::from_str::<serde_json::Value>(&acknowledgement).ok()
== Some(serde_json::json!({"status": "accepted"}))
}

#[test]
Expand Down
Loading
Loading