From ec2b3841eed8ea1ba7f130faf778263695c0f55b Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Sun, 18 Oct 2026 02:17:08 -0400 Subject: [PATCH] mesh: a grant rides over the OpenAI-compatible HTTP binding, and a receipt can be shown back as a witness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The grant work (#294, #301, #306) was reachable only through the library. The reference HTTP server passed no grant to every request, had no way to learn which steward a grant must name, and no way to witness a receipt against the journal. Its binaries kept the journal in memory. It also read whatever request body length a client declared. - **Grants over HTTP.** A request's body carries its presentation in `jcp_presentation` (`presentation_in`). The presentation is signed for a request id, so a granted request is served as exactly the `ServeRequest` the body alone determines (`granted_serve_request`, plus a `_responses` variant). Its prompt-token figure stays the body's estimate rather than the server's tokenizer count, which a caller cannot reproduce. The body must name its own `metadata.joule_ceiling_uj`, since a server default would otherwise enter the id unseen. - **Endpoints.** `GET /v1/mesh/steward` returns the steward's did:key, the subject a grant must name. `POST /v1/mesh/witness`, with the `receipt` and `envelope_hex` a response carried, answers 200 when the journal holds the receipt and 409 when it does not. `Mesh::witness_wire` checks a receipt in that wire form: the envelope must verify under this steward and seal exactly that wire receipt. - **Statuses** (`http_status`): 401 when the caller is not the grant's issuer, 403 when the steward refuses the grant, 503 once it has stopped honouring grants, 500 for an unwritable journal, 400 otherwise. - **Journal in the binaries.** With `MESH_JOURNAL` set, the reference server example and serve_openai_falcon record to that file and replay it before serving, reporting any unresolved serve. - **Body cap.** A body declared over 4 MiB is refused (413) before it is read. - **Spec.** Mesh SPEC §2.3 describes the binding (informative). Checked on the build host: - mesh-rs: 81 tests pass. mesh-openai with the server feature passes 15, including an end-to-end socket test: learn the steward, grant to it, present over HTTP, witness the receipt, refuse a tampered one, refuse an oversize body. - The gguf and openai server features build, every Cargo.lock is current, and clippy is clean. - Mutations each fail tests: a server that drops the grant, a granted path that recounts tokens, a wire witness that skips the seal check, and no body cap. Co-Authored-By: Claude Opus 5.5 --- mesh/SPEC.md | 20 ++ mesh/mesh-rs/Cargo.lock | 2 + mesh/mesh-rs/crates/mesh-core/src/lib.rs | 54 ++++- .../src/bin/serve_openai_falcon.rs | 13 ++ mesh/mesh-rs/crates/mesh-openai/Cargo.toml | 2 + .../crates/mesh-openai/examples/serve.rs | 20 +- mesh/mesh-rs/crates/mesh-openai/src/lib.rs | 185 +++++++++++++++++- mesh/mesh-rs/crates/mesh-openai/src/server.rs | 168 ++++++++++++++-- 8 files changed, 440 insertions(+), 24 deletions(-) diff --git a/mesh/SPEC.md b/mesh/SPEC.md index 7b2b13d14..d169c3117 100644 --- a/mesh/SPEC.md +++ b/mesh/SPEC.md @@ -163,6 +163,26 @@ short and is dropped: an authorisation never recorded never started its serve, a debit never recorded leaves its serve unresolved, and a seal never recorded was never handed out. + +### §2.3 Grants over the OpenAI-compatible binding (informative) + +The reference binding (`mesh-openai`) serves `/v1/chat/completions` and +`/v1/responses`. A request rides on a grant by carrying its presentation (§2.1) +in the body field `jcp_presentation`. The presentation is signed for a request +id, so a caller must be able to compute the request a body is served as: a +granted request is served as exactly what the binding derives from the body +alone, with the prompt-token figure left as the body's estimate rather than +replaced by the server's tokenizer count, and it MUST name its own +`metadata.joule_ceiling_uj`, since a server default would otherwise enter the id +unseen. The binding exports that derivation as `granted_serve_request`. + +`GET /v1/mesh/steward` returns the steward's `did:key`, the subject a grant must +name. `POST /v1/mesh/witness`, with a body of the `receipt` and `envelope_hex` +a response's `usage.energy` carried, answers 200 when the journal holds that +receipt and 409 when it does not, after which grants are refused (§2.2). A +refused grant answers 401 when the caller is not its issuer, 403 when the +steward refuses the grant, and 503 once the steward has stopped honouring grants. + --- ## §3 The request and the energy record diff --git a/mesh/mesh-rs/Cargo.lock b/mesh/mesh-rs/Cargo.lock index 76be47197..9285c27cd 100644 --- a/mesh/mesh-rs/Cargo.lock +++ b/mesh/mesh-rs/Cargo.lock @@ -622,6 +622,8 @@ version = "0.1.0" dependencies = [ "ed25519-dalek", "hex", + "jcp-grant", + "jcp-schema", "jouleclaw-backend-apple", "mesh-core", "mesh-engine", diff --git a/mesh/mesh-rs/crates/mesh-core/src/lib.rs b/mesh/mesh-rs/crates/mesh-core/src/lib.rs index a3757f318..7f2ff66a4 100644 --- a/mesh/mesh-rs/crates/mesh-core/src/lib.rs +++ b/mesh/mesh-rs/crates/mesh-core/src/lib.rs @@ -275,10 +275,35 @@ impl<'m> Mesh<'m> { receipt .verify_envelope(envelope, self.log.steward(), &self.log.verifying_key()) .map_err(|e| MeshError::WitnessInvalid(e.to_string()))?; - if self.sealed.get(receipt.sequence as usize) == Some(&ContentId::of_bytes(envelope)) { + self.held(receipt.sequence, envelope) + } + + /// [`witness`](Self::witness) for a receipt in its `mesh/1` wire form, the + /// form the HTTP binding hands out: the envelope's signature must verify + /// under this steward, and its payload must be exactly the wire receipt's + /// signed core, whose `sequence` then names the record to look for. + pub fn witness_wire(&mut self, receipt: &serde_json::Value, envelope: &[u8]) -> Result<(), MeshError> { + let invalid = |e: String| MeshError::WitnessInvalid(e); + let payload = mesh_receipt::jcr1::verify_cose_sign1(envelope, self.log.steward(), &self.log.verifying_key()) + .map_err(|e| invalid(e.to_string()))?; + let core = mesh_receipt::jcr1::signed_core_bytes(receipt).map_err(|e| invalid(e.to_string()))?; + if payload != core { + return Err(invalid("the envelope does not seal this receipt".into())); + } + let sequence = receipt + .get("sequence") + .and_then(serde_json::Value::as_u64) + .ok_or_else(|| invalid("the receipt carries no sequence".into()))?; + self.held(sequence, envelope) + } + + /// Whether the journal holds the receipt sealed at `sequence` as + /// `envelope`; if not, the journal was cut, and grants are refused. + fn held(&mut self, sequence: u64, envelope: &[u8]) -> Result<(), MeshError> { + if self.sealed.get(sequence as usize) == Some(&ContentId::of_bytes(envelope)) { return Ok(()); } - let cut = (self.sealed.len() as u64, receipt.sequence); + let cut = (self.sealed.len() as u64, sequence); self.cut = Some(cut); Err(MeshError::JournalCut { holds: cut.0, witnessed: cut.1 }) } @@ -1195,6 +1220,31 @@ mod tests { assert!(resume(&recorded).is_none()); } + /// A receipt in its wire form, as the HTTP binding hands it out, is a + /// witness too: the envelope must seal exactly that wire receipt. + #[test] + fn a_wire_receipt_is_a_witness() { + let meter = SteppingMeter::new("test-rapl", Provenance::HwShunt, 1_000); + let req = request(8, Fidelity::F1, 4_000_000); + let journal = SharedJournal::default(); + let mut before = mesh_with(&meter).with_journal(journal.clone(), &[]).unwrap(); + let first = before.serve(&req, &[], None).unwrap(); + let second = before.serve(&req, &[], None).unwrap(); + let wire = |r: &ServeResponse| r.receipt.to_jcr1_wire(before.steward()); + let (w1, w2) = (wire(&first), wire(&second)); + before.witness_wire(&w1, &first.envelope).expect("its own wire receipt"); + let mut altered = w1.clone(); + altered["sequence"] = serde_json::json!(5); + assert!(matches!(before.witness_wire(&altered, &first.envelope), Err(MeshError::WitnessInvalid(_)))); + assert!(matches!(before.witness_wire(&w2, &first.envelope), Err(MeshError::WitnessInvalid(_)))); + drop(before); + + let recorded = journal.0.borrow().clone(); + let mut cut = mesh_with(&meter).with_journal(MemoryJournal::default(), &recorded[..1]).unwrap(); + cut.witness_wire(&w1, &first.envelope).expect("the journal holds the first"); + assert!(matches!(cut.witness_wire(&w2, &second.envelope), Err(MeshError::JournalCut { holds: 1, witnessed: 1 }))); + } + /// A witness must be a receipt this steward sealed, unaltered. #[test] fn a_witness_must_be_this_stewards_receipt() { diff --git a/mesh/mesh-rs/crates/mesh-engine-gguf/src/bin/serve_openai_falcon.rs b/mesh/mesh-rs/crates/mesh-engine-gguf/src/bin/serve_openai_falcon.rs index eafcacd22..d8c6c46fd 100644 --- a/mesh/mesh-rs/crates/mesh-engine-gguf/src/bin/serve_openai_falcon.rs +++ b/mesh/mesh-rs/crates/mesh-engine-gguf/src/bin/serve_openai_falcon.rs @@ -72,6 +72,19 @@ fn main() -> std::io::Result<()> { Box::new(engine), ); + // With `MESH_JOURNAL` set, grant decisions and seals are recorded in that + // file and replayed before serving, so a restart gives no budget back and + // the receipt chain continues (Mesh SPEC §2.2). + if let Some(path) = std::env::var_os("MESH_JOURNAL") { + let (journal, recorded) = mesh_core::FileJournal::open(&path)?; + mesh = mesh + .with_journal(journal, &recorded) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?; + for u in mesh.unresolved() { + eprintln!("unresolved: request {} on backend {}, debited {} µJ", u.request_id.to_hex(), u.backend, u.debited_uj); + } + } + let listener = TcpListener::bind(&addr)?; println!("OpenAI-compatible Mesh endpoint: http://{addr}/v1/chat/completions"); println!("steward: {} | backend: falcon-h1-0.5b (F3) | receipt in usage.energy", mesh.steward()); diff --git a/mesh/mesh-rs/crates/mesh-openai/Cargo.toml b/mesh/mesh-rs/crates/mesh-openai/Cargo.toml index fb3899a82..5e520d441 100644 --- a/mesh/mesh-rs/crates/mesh-openai/Cargo.toml +++ b/mesh/mesh-rs/crates/mesh-openai/Cargo.toml @@ -22,6 +22,8 @@ jouleclaw-backend-apple = { workspace = true, optional = true } [dev-dependencies] mesh-engine = { workspace = true } mesh-meter = { workspace = true } +jcp-schema = { workspace = true } +jcp-grant = { workspace = true } mesh-router = { workspace = true } mesh-receipt = { workspace = true } ed25519-dalek = { workspace = true } diff --git a/mesh/mesh-rs/crates/mesh-openai/examples/serve.rs b/mesh/mesh-rs/crates/mesh-openai/examples/serve.rs index 135848033..416d6689e 100644 --- a/mesh/mesh-rs/crates/mesh-openai/examples/serve.rs +++ b/mesh/mesh-rs/crates/mesh-openai/examples/serve.rs @@ -92,6 +92,24 @@ fn main() -> std::io::Result<()> { let listener = TcpListener::bind(&addr)?; let meter = meter(); eprintln!("mesh OpenAI server: http://{addr}/v1/chat/completions (meter: {})", meter.name()); - let mut mesh = mesh(meter.as_ref()); + let mut mesh = journalled(mesh(meter.as_ref()))?; server::run(&listener, &mut mesh, &ByteBackend, 100_000_000) } + +/// With `MESH_JOURNAL` set, record grant decisions and seals in that file and +/// replay what it already holds before serving, so a restart gives no budget +/// back and the receipt chain continues (Mesh SPEC §2.2). Without it the +/// journal is in memory, which a steward that restarts must not rely on. +pub fn journalled(mesh: mesh_core::Mesh<'_>) -> std::io::Result> { + let Some(path) = std::env::var_os("MESH_JOURNAL") else { + return Ok(mesh); + }; + let (journal, recorded) = mesh_core::FileJournal::open(&path)?; + let mesh = mesh + .with_journal(journal, &recorded) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?; + for u in mesh.unresolved() { + eprintln!("unresolved: request {} on backend {}, debited {} µJ", u.request_id.to_hex(), u.backend, u.debited_uj); + } + Ok(mesh) +} diff --git a/mesh/mesh-rs/crates/mesh-openai/src/lib.rs b/mesh/mesh-rs/crates/mesh-openai/src/lib.rs index 6037cd499..694d57069 100644 --- a/mesh/mesh-rs/crates/mesh-openai/src/lib.rs +++ b/mesh/mesh-rs/crates/mesh-openai/src/lib.rs @@ -23,6 +23,11 @@ //! served result returned as a Responses `output_text` item with the sealed //! receipt under `usage.energy`. The reference `server` exposes both routes. //! +//! - A request may ride on a JCP grant: its body carries a +//! [`Presentation`] in `jcp_presentation` ([`presentation_in`]), signed for +//! exactly the request [`granted_serve_request`] computes from the body, and +//! the steward decides it before any work (Mesh SPEC §2.1). +//! //! Ingest, not compete: the OpenAI shape is the surface; Mesh contributes the //! metered receipt. Token counts in/out are the engine's tokenizer's job; where //! a request omits them this crate makes an explicit ~chars/4 estimate (honestly @@ -47,6 +52,71 @@ pub enum OpenAiError { /// The mesh refused or failed to serve the request. #[error("serve failed: {0}")] Serve(#[from] MeshError), + /// A request that rides on a grant must name its own joule ceiling: the + /// request its caller signs for is computed from the body alone, and a + /// server default would enter it unseen. + #[error("a request that rides on a grant names its own `metadata.joule_ceiling_uj`")] + GrantedNeedsCeiling, + /// The body's `jcp_presentation` is not a presentation. + #[error("`{PRESENTATION_FIELD}` is not a presentation: {0}")] + BadPresentation(String), +} + +/// The body field a request carries its grant's [`Presentation`] in. +pub const PRESENTATION_FIELD: &str = "jcp_presentation"; + +/// The grant a request body carries, if any: its [`PRESENTATION_FIELD`]. +pub fn presentation_in(req: &Value) -> Result, OpenAiError> { + match req.get(PRESENTATION_FIELD) { + None => Ok(None), + Some(v) => serde_json::from_value(v.clone()).map(Some).map_err(|e| OpenAiError::BadPresentation(e.to_string())), + } +} + +/// The exact [`ServeRequest`] a granted `/v1/chat/completions` body is served +/// as: the request whose id the grant's capability names and the presentation +/// is signed for. It is computed from the body alone, so a caller can compute +/// it too. The body must name its joule ceiling, and the prompt-token figure +/// stays the body's chars/4 estimate rather than the server's tokenizer count, +/// which a caller cannot reproduce. That figure is only the router's cost +/// estimate: the energy is metered, and `tokens_in` is what the engine reports. +pub fn granted_serve_request(req: &Value) -> Result { + names_ceiling(req)?; + Ok(to_serve_request(req, 0)?.0) +} + +/// [`granted_serve_request`] for a `/v1/responses` body. +pub fn granted_serve_request_responses(req: &Value) -> Result { + names_ceiling(req)?; + Ok(to_serve_request_responses(req, 0)?.0) +} + +fn names_ceiling(req: &Value) -> Result<(), OpenAiError> { + match req.get("metadata").and_then(|m| m.get("joule_ceiling_uj")).and_then(Value::as_u64) { + Some(_) => Ok(()), + None => Err(OpenAiError::GrantedNeedsCeiling), + } +} + +/// The HTTP status a failed request answers with: 401 for a caller that is +/// not the grant's issuer, 403 for a grant the steward refuses, 503 for a +/// steward that has stopped honouring grants because its journal was cut, +/// 500 for a journal it cannot write, and 400 for everything else. +pub fn http_status(e: &OpenAiError) -> (u16, &'static str) { + match e { + OpenAiError::Serve(MeshError::CallerNotAuthenticated(_)) => (401, "Unauthorized"), + OpenAiError::Serve( + MeshError::Replayed + | MeshError::NotThisSteward + | MeshError::Unauthorized + | MeshError::GrantInvalid(_) + | MeshError::GrantDenied(_) + | MeshError::BudgetTooLow { .. }, + ) => (403, "Forbidden"), + OpenAiError::Serve(MeshError::JournalCut { .. }) => (503, "Service Unavailable"), + OpenAiError::Serve(MeshError::Journal(_)) => (500, "Internal Server Error"), + _ => (400, "Bad Request"), + } } /// Parse an OpenAI `/v1/chat/completions` request into a Mesh [`ServeRequest`] @@ -209,8 +279,13 @@ pub fn complete_chat( ) -> Result { let (mut serve, prompt) = to_serve_request(req, default_ceiling_uj)?; let ids = backend.encode(&prompt); - // Replace the chars/4 estimate with the tokenizer's real input count. - serve.prompt_tokens = ids.len() as u32; + if grant.is_some() { + // Served as exactly the request its caller signed for. + serve = granted_serve_request(req)?; + } else { + // Replace the chars/4 estimate with the tokenizer's real input count. + serve.prompt_tokens = ids.len() as u32; + } let model = req .get("model") @@ -415,7 +490,11 @@ pub fn complete_responses( ) -> Result { let (mut serve, prompt) = to_serve_request_responses(req, default_ceiling_uj)?; let ids = backend.encode(&prompt); - serve.prompt_tokens = ids.len() as u32; + if grant.is_some() { + serve = granted_serve_request_responses(req)?; + } else { + serve.prompt_tokens = ids.len() as u32; + } let model = req.get("model").and_then(Value::as_str).unwrap_or("mesh").to_string(); let resp = mesh.serve(&serve, &ids, grant)?; @@ -463,6 +542,106 @@ mod tests { assert_eq!(serve.max_tokens, 256); // default when omitted } + struct Bytes; + impl ChatBackend for Bytes { + fn encode(&self, text: &str) -> Vec { + text.bytes().map(u32::from).collect() + } + fn decode(&self, ids: &[u32]) -> String { + format!("{} tokens", ids.len()) + } + } + + fn steward_mesh(meter: &dyn mesh_meter::Meter) -> Mesh<'_> { + let mut mesh = Mesh::new(meter, ed25519_dalek::SigningKey::from_bytes(&[8u8; 32]), "did:example:mesh-test"); + mesh.register( + mesh_router::BackendSpec::new("tiny", "ssm-1b", Fidelity::F1, 50_000), + Box::new(mesh_engine::ReferenceEngine::new("tiny", "ssm-1b", Fidelity::F1, 50_000, 32)), + ); + mesh + } + + /// `body` with a presentation of a grant, issued by `caller` to `mesh`'s + /// steward over exactly the request the body is served as, signed by + /// `signer`. + fn granted(body: &Value, mesh: &Mesh, signer: &ed25519_dalek::SigningKey, nonce: u8) -> Value { + let caller = ed25519_dalek::SigningKey::from_bytes(&[42u8; 32]); + let req = granted_serve_request(body).unwrap(); + let g = jcp_schema::Grant { + issuer: jcp_schema::Did::from_verifying_key(&caller.verifying_key()), + subject: mesh.steward_did(), + capability: jcp_schema::Capability::new(req.id(), [mesh_core::INFER_ACTION]), + budget_uj: 1_000_000_000, + budget_funds: None, + caveats: Vec::new(), + expiry: 0, + parent: None, + delegate_depth: None, + nonce: [1u8; 16], + }; + let sg = jcp_grant::issue(&caller, g).unwrap(); + let mut out = body.clone(); + out[PRESENTATION_FIELD] = serde_json::to_value(Presentation::sign(vec![sg], &req, signer, [nonce; 16])).unwrap(); + out + } + + fn chat_body() -> Value { + json!({ + "model": "gpt-mini", + "messages": [{ "role": "user", "content": "capital of France?" }], + "max_tokens": 8, + "metadata": { "joule_ceiling_uj": 50_000_000, "fidelity_floor": 1 } + }) + } + + /// A request over HTTP can ride on a grant: its body carries the + /// presentation, signed for the request the body alone determines, and + /// the receipt names the grant it was spent on. + #[test] + fn a_granted_request_is_served_as_the_request_its_caller_signed_for() { + let meter = mesh_meter::EstimatorMeter::default(); + let mut mesh = steward_mesh(&meter); + let caller = ed25519_dalek::SigningKey::from_bytes(&[42u8; 32]); + let body = granted(&chat_body(), &mesh, &caller, 1); + let p = presentation_in(&body).unwrap().expect("a presentation"); + let resp = complete_chat(&mut mesh, &Bytes, Some(&p), &body, 1, 0, "id").unwrap(); + let receipt = &resp["usage"]["energy"]["receipt"]; + assert_eq!(receipt["grant_id"], p.leaf().unwrap().grant.id().to_hex()); + assert_eq!(receipt["request_id"], granted_serve_request(&body).unwrap().id().to_hex()); + // Without a grant the server swaps in its tokenizer's count, so the + // request it serves is not one a caller could have signed for. + let plain = complete_chat(&mut mesh, &Bytes, None, &chat_body(), 1, 0, "id").unwrap(); + assert_ne!(plain["usage"]["energy"]["receipt"]["request_id"], receipt["request_id"]); + } + + /// The ceiling enters the request id, so a granted request must name it. + #[test] + fn a_granted_request_must_name_its_ceiling() { + let mut body = chat_body(); + body["metadata"].as_object_mut().unwrap().remove("joule_ceiling_uj"); + assert!(matches!(granted_serve_request(&body), Err(OpenAiError::GrantedNeedsCeiling))); + assert!(matches!(presentation_in(&json!({ PRESENTATION_FIELD: 7 })), Err(OpenAiError::BadPresentation(_)))); + assert!(presentation_in(&chat_body()).unwrap().is_none()); + } + + /// A caller that is not the grant's issuer is 401, a replay 403. + #[test] + fn grant_refusals_carry_their_http_status() { + let meter = mesh_meter::EstimatorMeter::default(); + let mut mesh = steward_mesh(&meter); + let thief = ed25519_dalek::SigningKey::from_bytes(&[66u8; 32]); + let forged = granted(&chat_body(), &mesh, &thief, 1); + let e = complete_chat(&mut mesh, &Bytes, presentation_in(&forged).unwrap().as_ref(), &forged, 1, 0, "id").unwrap_err(); + assert_eq!(http_status(&e).0, 401, "{e}"); + + let caller = ed25519_dalek::SigningKey::from_bytes(&[42u8; 32]); + let body = granted(&chat_body(), &mesh, &caller, 2); + let p = presentation_in(&body).unwrap(); + complete_chat(&mut mesh, &Bytes, p.as_ref(), &body, 1, 0, "id").unwrap(); + let e = complete_chat(&mut mesh, &Bytes, p.as_ref(), &body, 1, 0, "id").unwrap_err(); + assert_eq!(http_status(&e).0, 403, "{e}"); + } + #[test] fn model_fidelity_heuristic() { assert_eq!(fidelity_for_model("gpt-mini"), Fidelity::F2); diff --git a/mesh/mesh-rs/crates/mesh-openai/src/server.rs b/mesh/mesh-rs/crates/mesh-openai/src/server.rs index e2254f72a..85fc7f2a4 100644 --- a/mesh/mesh-rs/crates/mesh-openai/src/server.rs +++ b/mesh/mesh-rs/crates/mesh-openai/src/server.rs @@ -6,6 +6,11 @@ //! `usage.energy`. It is single-connection-at-a-time by design, so the steward's //! receipt chain stays coherent (one ordered ledger of what was served). //! +//! A request rides on a JCP grant by carrying its presentation in the body's +//! `jcp_presentation` field (Mesh SPEC §2.1); `GET /v1/mesh/steward` names the +//! identity a grant must be issued to, and `POST /v1/mesh/witness` checks a +//! receipt this steward sealed against its journal (§2.2). +//! //! ```no_run //! # #[cfg(feature = "server")] { //! use std::net::TcpListener; @@ -26,7 +31,12 @@ use std::time::{SystemTime, UNIX_EPOCH}; use mesh_core::Mesh; use serde_json::{json, Value}; -use crate::{complete_chat, complete_responses, ChatBackend}; +use crate::{complete_chat, complete_responses, http_status, presentation_in, ChatBackend, OpenAiError}; + +/// The largest request body read, in bytes. A larger one is refused before +/// it is read, so a client cannot grow the server's memory by declaring a +/// length. +const MAX_BODY: usize = 4 << 20; /// Serve exactly `n` connections, then return. `run` is `serve_n(.., usize::MAX)`. /// Useful for tests and bounded demos. @@ -83,7 +93,12 @@ fn handle_conn( } }; - if req.method == "POST" && req.path.starts_with("/v1/chat/completions") { + if req.too_large { + write_json(stream, 413, "Payload Too Large", &error_json("request body over the server's limit")); + return; + } + let completion = req.path.starts_with("/v1/chat/completions"); + if req.method == "POST" && (completion || req.path.starts_with("/v1/responses")) { let body: Value = match serde_json::from_str(&req.body) { Ok(v) => v, Err(e) => { @@ -91,25 +106,28 @@ fn handle_conn( return; } }; - let id = format!("chatcmpl-mesh-{seq}"); - match complete_chat(mesh, backend, None, &body, default_ceiling_uj, now_secs(), &id) { - Ok(resp) => write_json(stream, 200, "OK", &resp), - // A refused serve (no route, budget, …) maps to a 400 with the reason. - Err(e) => write_json(stream, 400, "Bad Request", &error_json(&e.to_string())), - } - } else if req.method == "POST" && req.path.starts_with("/v1/responses") { - let body: Value = match serde_json::from_str(&req.body) { - Ok(v) => v, - Err(e) => { - write_json(stream, 400, "Bad Request", &error_json(&format!("invalid JSON: {e}"))); - return; + let served = presentation_in(&body).and_then(|grant| { + if completion { + let id = format!("chatcmpl-mesh-{seq}"); + complete_chat(mesh, backend, grant.as_ref(), &body, default_ceiling_uj, now_secs(), &id) + } else { + let id = format!("resp-mesh-{seq}"); + complete_responses(mesh, backend, grant.as_ref(), &body, default_ceiling_uj, now_secs(), &id) } - }; - let id = format!("resp-mesh-{seq}"); - match complete_responses(mesh, backend, None, &body, default_ceiling_uj, now_secs(), &id) { + }); + match served { Ok(resp) => write_json(stream, 200, "OK", &resp), - Err(e) => write_json(stream, 400, "Bad Request", &error_json(&e.to_string())), + // A refused serve answers with the status its reason calls for. + Err(e) => { + let (code, reason) = http_status(&e); + write_json(stream, code, reason, &error_json(&e.to_string())); + } } + } else if req.method == "GET" && req.path.starts_with("/v1/mesh/steward") { + write_json(stream, 200, "OK", &json!({ "steward": mesh.steward(), "did": mesh.steward_did().to_did_key() })); + } else if req.method == "POST" && req.path.starts_with("/v1/mesh/witness") { + let (code, reason, body) = witness(mesh, &req.body); + write_json(stream, code, reason, &body); } else if req.method == "GET" && req.path.starts_with("/v1/models") { write_json(stream, 200, "OK", &models_json(mesh)); } else { @@ -138,6 +156,8 @@ struct Request { method: String, path: String, body: String, + /// The declared body was over [`MAX_BODY`] and was not read. + too_large: bool, } /// Read one HTTP/1.1 request: the request line, headers up to the blank line, @@ -176,6 +196,9 @@ fn read_request(stream: &mut TcpStream) -> std::io::Result> { } } + if content_length > MAX_BODY { + return Ok(Some(Request { method, path, body: String::new(), too_large: true })); + } let body_start = header_end + 4; let mut body: Vec = buf[body_start..].to_vec(); while body.len() < content_length { @@ -191,9 +214,34 @@ fn read_request(stream: &mut TcpStream) -> std::io::Result> { method, path, body: String::from_utf8_lossy(&body).to_string(), + too_large: false, })) } +/// `POST /v1/mesh/witness`: a body `{"receipt": , +/// "envelope_hex": "..."}`, as a response's `usage.energy` carries them. 200 +/// if the journal holds the receipt; 409 if it does not, after which the +/// steward refuses grants; 400 if it is not a receipt this steward sealed. +fn witness(mesh: &mut Mesh, body: &str) -> (u16, &'static str, Value) { + let parsed: Result<(Value, Vec), String> = serde_json::from_str::(body) + .map_err(|e| format!("invalid JSON: {e}")) + .and_then(|v| { + let hex_env = v.get("envelope_hex").and_then(Value::as_str).ok_or("no `envelope_hex`")?; + let envelope = hex::decode(hex_env).map_err(|e| format!("`envelope_hex`: {e}"))?; + let receipt = v.get("receipt").cloned().ok_or("no `receipt`")?; + Ok((receipt, envelope)) + }); + let (receipt, envelope) = match parsed { + Ok(p) => p, + Err(e) => return (400, "Bad Request", error_json(&e)), + }; + match mesh.witness_wire(&receipt, &envelope) { + Ok(()) => (200, "OK", json!({ "held": true })), + Err(e @ mesh_core::MeshError::JournalCut { .. }) => (409, "Conflict", error_json(&OpenAiError::Serve(e).to_string())), + Err(e) => (400, "Bad Request", error_json(&e.to_string())), + } +} + fn write_json(stream: &mut TcpStream, code: u16, reason: &str, body: &Value) { let payload = serde_json::to_string(body).unwrap_or_else(|_| "{}".to_string()); let resp = format!( @@ -242,6 +290,90 @@ mod tests { } } + /// One HTTP exchange: the status code and the JSON body. + fn call(addr: std::net::SocketAddr, method: &str, path: &str, body: &str, declared: Option) -> (u16, serde_json::Value) { + let mut conn = TcpStream::connect(addr).unwrap(); + let req = format!( + "{method} {path} HTTP/1.1\r\nHost: x\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + declared.unwrap_or(body.len()), + ); + conn.write_all(req.as_bytes()).unwrap(); + // Nothing more is coming, so a server waiting on a declared length + // reads to the end rather than waiting forever. + conn.shutdown(std::net::Shutdown::Write).unwrap(); + let mut raw = String::new(); + conn.read_to_string(&mut raw).unwrap(); + let code = raw[9..12].parse().unwrap(); + let json = serde_json::from_str(raw.split("\r\n\r\n").nth(1).unwrap_or("null")).unwrap_or(serde_json::Value::Null); + (code, json) + } + + /// The whole grant flow over HTTP: learn the steward's identity, issue a + /// grant to it over exactly the request the body determines, present it in + /// the body, then show the receipt back as a witness. + #[test] + fn a_grant_rides_over_http_and_its_receipt_witnesses_the_journal() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + let meter = SteppingMeter::new("test-rapl", Provenance::HwShunt, 4_800_000); + let mut mesh = Mesh::new(&meter, SigningKey::from_bytes(&[8u8; 32]), "did:example:mesh-openai-srv"); + mesh.register( + BackendSpec::new("tiny", "ssm-1b", Fidelity::F2, 50_000), + Box::new(ReferenceEngine::new("tiny", "ssm-1b", Fidelity::F2, 50_000, 32)), + ); + super::serve_n(&listener, &mut mesh, &StubBackend, 100_000_000, 5).unwrap(); + }); + + let (code, steward) = call(addr, "GET", "/v1/mesh/steward", "", None); + assert_eq!(code, 200); + let holder = jcp_schema::Did::parse(steward["did"].as_str().unwrap()).unwrap(); + + let mut body = serde_json::json!({ + "model": "gpt-mini", + "messages": [{ "role": "user", "content": "hi" }], + "max_tokens": 16, + "metadata": { "joule_ceiling_uj": 50_000_000 } + }); + let req = crate::granted_serve_request(&body).unwrap(); + let caller = SigningKey::from_bytes(&[42u8; 32]); + let grant = jcp_grant::issue( + &caller, + jcp_schema::Grant { + issuer: jcp_schema::Did::from_verifying_key(&caller.verifying_key()), + subject: holder, + capability: jcp_schema::Capability::new(req.id(), [mesh_core::INFER_ACTION]), + budget_uj: 1_000_000_000, + budget_funds: None, + caveats: Vec::new(), + expiry: 0, + parent: None, + delegate_depth: None, + nonce: [1; 16], + }, + ) + .unwrap(); + let presented = mesh_core::Presentation::sign(vec![grant.clone()], &req, &caller, [3; 16]); + body[crate::PRESENTATION_FIELD] = serde_json::to_value(&presented).unwrap(); + let (code, resp) = call(addr, "POST", "/v1/chat/completions", &body.to_string(), None); + assert_eq!(code, 200, "{resp}"); + let energy = &resp["usage"]["energy"]; + assert_eq!(energy["receipt"]["grant_id"], grant.grant.id().to_hex()); + + let witness = serde_json::json!({ "receipt": energy["receipt"], "envelope_hex": energy["envelope_hex"] }); + let (code, held) = call(addr, "POST", "/v1/mesh/witness", &witness.to_string(), None); + assert_eq!((code, &held["held"]), (200, &serde_json::json!(true)), "{held}"); + + let mut tampered = witness.clone(); + tampered["receipt"]["sequence"] = serde_json::json!(9); + let (code, _) = call(addr, "POST", "/v1/mesh/witness", &tampered.to_string(), None); + assert_eq!(code, 400, "a receipt the envelope does not seal"); + + let (code, _) = call(addr, "POST", "/v1/chat/completions", "{}", Some(super::MAX_BODY + 1)); + assert_eq!(code, 413, "a body over the limit is refused unread"); + server.join().unwrap(); + } + #[test] fn http_roundtrip_returns_sealed_receipt() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); -- 2.54.0 (Apple Git-157)