client.rs 97.8%

87 of 89 lines run, 37 of 88 functions · run #11 on main · run missed

//! A client for the daemon over its Unix socket (spec §8.5), signing every
//! request with the caller's key: the shims and the CLI use it, and so do
//! the tests.

use anyhow::{Context, Result};
use http_body_util::{BodyExt, Full};
use hyper::body::Bytes;
use spectra_core::keys::KeyPair;
use spectra_core::proof::Proof;
use std::path::{Path, PathBuf};

/// A response, its body read whole.
#[derive(Debug)]
pub struct Response {
    pub status: u16,
    pub headers: Vec<(String, String)>,
    pub body: Vec<u8>,
}

impl Response {
    pub fn json(&self) -> Result<serde_json::Value> {
        serde_json::from_slice(&self.body).with_context(|| format!("not JSON: {}", String::from_utf8_lossy(&self.body)))
    }
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers.iter().find(|(k, _)| k.eq_ignore_ascii_case(name)).map(|(_, v)| v.as_str())
    }
}

pub struct Client {
    socket: PathBuf,
    signer: Option<KeyPair>,
}

impl Client {
    pub fn new(socket: &Path, signer: Option<KeyPair>) -> Client {
        Client { socket: socket.to_path_buf(), signer }
    }

    /// Whether a daemon answers on the socket.
    pub async fn alive(socket: &Path) -> bool {
        tokio::net::UnixStream::connect(socket).await.is_ok()
    }

    /// Send `method path` with `body` and extra headers, signed.
    pub async fn send(&self, method: &str, path: &str, headers: &[(&str, String)], body: Vec<u8>) -> Result<Response> {
        let stream = tokio::net::UnixStream::connect(&self.socket)
            .await
            .with_context(|| format!("connecting to {} (is `spectra serve` running?)", self.socket.display()))?;
        let (mut sender, conn) = hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await?;
        tokio::spawn(async move {
            let _ = conn.await;
        });
        let mut req = hyper::Request::builder()
            .method(method)
            .uri(path)
            .header("host", "spectra")
            .header("content-type", "application/json")
            .header("connection", "close");
        // One request a connection: the server closes it after answering, so
        // nothing lingers to hold up its shutdown.
        for (k, v) in headers {
            req = req.header(*k, v);
        }
        if let Some(pair) = &self.signer {
            for (k, v) in sign_headers(pair, &format!("{method} {path}"), &body) {
                req = req.header(k, v);
            }
        }
        let resp = sender.send_request(req.body(Full::new(Bytes::from(body)))?).await?;
        let status = resp.status().as_u16();
        let headers = resp.headers().iter().map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string())).collect();
        let body = resp.into_body().collect().await?.to_bytes().to_vec();
        Ok(Response { status, headers, body })
    }

    /// A JSON-RPC request to `/mcp` with the protocol headers of `version`.
    pub async fn mcp(&self, version: &str, request: &serde_json::Value) -> Result<Response> {
        let method = request.get("method").and_then(|m| m.as_str()).unwrap_or("").to_string();
        let mut headers = vec![
            ("accept", "application/json, text/event-stream".to_string()),
            ("mcp-protocol-version", version.to_string()),
            ("mcp-method", method.clone()),
        ];
        if let Some(name) = request.pointer("/params/name").or_else(|| request.pointer("/params/uri")).and_then(|n| n.as_str()) {
            headers.push(("mcp-name", name.to_string()));
        }
        self.send("POST", "/mcp", &headers, serde_json::to_vec(request)?).await
    }
}

/// The signature headers of spec §8.5 for `method_path` and `body`.
pub fn sign_headers(pair: &KeyPair, method_path: &str, body: &[u8]) -> Vec<(&'static str, String)> {
    let p = Proof::sign(pair, method_path, body);
    vec![
        ("spectra-key", p.sig.key.to_hex()),
        ("spectra-time", p.time_ms.to_string()),
        ("spectra-nonce", hex::encode(p.nonce)),
        ("spectra-signature", hex::encode(oxy_core::cbor::encode(&p.sig.to_value()))),
    ]
}

/// The JSON-RPC result inside a response, whether it came as JSON or as
/// the last `data:` of an SSE stream.
pub fn rpc_result(r: &Response) -> Result<serde_json::Value> {
    let is_sse = r.header("content-type").is_some_and(|c| c.starts_with("text/event-stream"));
    if !is_sse {
        return r.json();
    }
    let text = String::from_utf8_lossy(&r.body);
    let last = text
        .lines()
        .rev()
        .filter_map(|l| l.strip_prefix("data:"))
        .map(str::trim)
        .find(|l| !l.is_empty())
        .context("an SSE response with no data")?;
    Ok(serde_json::from_str(last)?)
}

1		//! A client for the daemon over its Unix socket (spec §8.5), signing every
2		//! request with the caller's key: the shims and the CLI use it, and so do
3		//! the tests.
4
5		use anyhow::{Context, Result};
6		use http_body_util::{BodyExt, Full};
7		use hyper::body::Bytes;
8		use spectra_core::keys::KeyPair;
9		use spectra_core::proof::Proof;
10		use std::path::{Path, PathBuf};
11
12		/// A response, its body read whole.
13		#[derive(Debug)]
14		pub struct Response {
15		pub status: u16,
16		pub headers: Vec<(String, String)>,
17		pub body: Vec<u8>,
18		}
19
20		impl Response {
21	24×	pub fn json(&self) -> Result<serde_json::Value> {
22	24×	serde_json::from_slice(&self.body).with_context(\|\| format!("not JSON: {}", String::from_utf8_lossy(&self.body)))
23	24×	}
24	23×	pub fn header(&self, name: &str) -> Option<&str> {
25	23×	self.headers.iter().find(\|(k, _)\| k.eq_ignore_ascii_case(name)).map(\|(_, v)\| v.as_str())
26	23×	}
27		}
28
29		pub struct Client {
30		socket: PathBuf,
31		signer: Option<KeyPair>,
32		}
33
34		impl Client {
35	14×	pub fn new(socket: &Path, signer: Option<KeyPair>) -> Client {
36	14×	Client { socket: socket.to_path_buf(), signer }
37	14×	}
38
39		/// Whether a daemon answers on the socket.
40	31×	pub async fn alive(socket: &Path) -> bool {
41	31×	tokio::net::UnixStream::connect(socket).await.is_ok()
42	31×	}
43
44		/// Send `method path` with `body` and extra headers, signed.
45	39×	pub async fn send(&self, method: &str, path: &str, headers: &[(&str, String)], body: Vec<u8>) -> Result<Response> {
46	39×	let stream = tokio::net::UnixStream::connect(&self.socket)
47	39×	.await
48	39×	.with_context(\|\| format!("connecting to {} (is `spectra serve` running?)", self.socket.display()))?;
49	39×	let (mut sender, conn) = hyper::client::conn::http1::handshake(hyper_util::rt::TokioIo::new(stream)).await?;
50	39×	tokio::spawn(async move {
51	39×	let _ = conn.await;
52	39×	});
53	39×	let mut req = hyper::Request::builder()
54	39×	.method(method)
55	39×	.uri(path)
56	39×	.header("host", "spectra")
57	39×	.header("content-type", "application/json")
58	39×	.header("connection", "close");
59		// One request a connection: the server closes it after answering, so
60		// nothing lingers to hold up its shutdown.
61	129×	for (k, v) in headers {
62	129×	req = req.header(*k, v);
63	129×	}
64	39×	if let Some(pair) = &self.signer {
65	152×	for (k, v) in sign_headers(pair, &format!("{method} {path}"), &body) {
66	152×	req = req.header(k, v);
67	152×	}
68	1×	}
69	39×	let resp = sender.send_request(req.body(Full::new(Bytes::from(body)))?).await?;
70	39×	let status = resp.status().as_u16();
71	164×	let headers = resp.headers().iter().map(\|(k, v)\| (k.to_string(), v.to_str().unwrap_or("").to_string())).collect();
72	39×	let body = resp.into_body().collect().await?.to_bytes().to_vec();
73	39×	Ok(Response { status, headers, body })
74	39×	}
75
76		/// A JSON-RPC request to `/mcp` with the protocol headers of `version`.
77	33×	pub async fn mcp(&self, version: &str, request: &serde_json::Value) -> Result<Response> {
78	33×	let method = request.get("method").and_then(\|m\| m.as_str()).unwrap_or("").to_string();
79	33×	let mut headers = vec![
80	33×	("accept", "application/json, text/event-stream".to_string()),
81	33×	("mcp-protocol-version", version.to_string()),
82	33×	("mcp-method", method.clone()),
83		];
84	33×	if let Some(name) = request.pointer("/params/name").or_else(\|\| request.pointer("/params/uri")).and_then(\|n\| n.as_str()) {
85	24×	headers.push(("mcp-name", name.to_string()));
86	24×	}
87	33×	self.send("POST", "/mcp", &headers, serde_json::to_vec(request)?).await
88	33×	}
89		}
90
91		/// The signature headers of spec §8.5 for `method_path` and `body`.
92	38×	pub fn sign_headers(pair: &KeyPair, method_path: &str, body: &[u8]) -> Vec<(&'static str, String)> {
93	38×	let p = Proof::sign(pair, method_path, body);
94	38×	vec![
95	38×	("spectra-key", p.sig.key.to_hex()),
96	38×	("spectra-time", p.time_ms.to_string()),
97	38×	("spectra-nonce", hex::encode(p.nonce)),
98	38×	("spectra-signature", hex::encode(oxy_core::cbor::encode(&p.sig.to_value()))),
99		]
100	38×	}
101
102		/// The JSON-RPC result inside a response, whether it came as JSON or as
103		/// the last `data:` of an SSE stream.
104	23×	pub fn rpc_result(r: &Response) -> Result<serde_json::Value> {
105	23×	let is_sse = r.header("content-type").is_some_and(\|c\| c.starts_with("text/event-stream"));
106	23×	if !is_sse {
107	18×	return r.json();
108	5×	}
109	5×	let text = String::from_utf8_lossy(&r.body);
110	5×	let last = text
111	5×	.lines()
112	5×	.rev()
113	10×	.filter_map(\|l\| l.strip_prefix("data:"))
114	5×	.map(str::trim)
115	5×	.find(\|l\| !l.is_empty())
116	5×	.context("an SSE response with no data")?;
117	5×	Ok(serde_json::from_str(last)?)
118	23×	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/spectra-mcp/src/client.rs 97.8%

Keyboard