Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 15 Packages Wiki Activity Actions

jcp-gateway: everything a gated request carries to the server is sealed and flow-checked #339

Merged
dcharlot merged 1 commit from fix/jcp-gateway-flow-checks-whole-request into main 2026-10-19 21:21:56 -04:00
Conversation 0 Commits 1 Files changed 2 +383 -39
2 changed files with 383 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

14
jcp/SPEC.md
View file

@ -506,6 +506,20 @@ policy — and *before* caveats and the budget test, so that **a tainted call co
zero joules**. A denial on flow is the same invariant as a denial on capability:
nothing is debited and nothing runs.
**What "the argument" covers.** Where an enforcement point forwards a request to
an endpoint, the argument whose label is gated is everything the request carries
to the endpoint except the routing that names the action. That includes fields
the protocol leaves open: extension parameters, side-channel metadata, and
envelope members the enforcement point does not otherwise interpret. Any such
field it leaves unlabelled is a way past the ceiling. The enforcement point MUST
label each of these fields, and MUST seal it with the arguments, or else MUST
NOT forward it. It MAY leave out the protocol's own fixed machinery (a version, a
capability set, a progress token) where folding that machinery would taint every
request. If it does, it MUST say which fields it leaves out. A host that admits
every value it labels SHOULD label a request's parts so that combining them adds
no text of its own. Otherwise the framing becomes shared text that carries one
tainted request's label into every request after it.
A grant carrying no `FlowBelow` caveat is byte-identical to one minted before
flow existed, and an enforcement point with no taint floor configured decides
exactly as it did before. Conformance class **C25**.

408
jcp/jcp-rs/crates/jcp-gateway/src/lib.rs
View file

@ -716,6 +716,25 @@ impl Gateway {
/// restarted gateway knows none of them: an exchange naming a task opened
/// before the restart is refused (fail closed), and a retry after one is
/// decided as a call with nothing to continue.
///
/// # Everything the server receives is checked (SPEC §4.10)
///
/// A gated request was decided and flow-checked over its arguments alone,
/// and forwarded whole, so whatever else it carried reached the server
/// unchecked. That covered a `_meta` key, a `params` member the method
/// does not define, and a member of the JSON-RPC message beside `params`.
/// A value from an untrusted read could leave by any of those past the
/// flow ceiling. Now each of those is sealed and flow-checked with the
/// arguments (`beside`). The exceptions are the protocol's own `_meta`
/// keys (`protocol_meta_key`), its machinery, which a server names back in
/// every reply, and this gateway's own namespace. A request carrying
/// nothing else is decided over exactly the bytes it was before.
///
/// A `resources/read` was gated over its whole `params`, `_meta`
/// included, so under MCP 2026-07-28 the protocol's own keys in every
/// request matched the server's in every reply. After one reply was
/// admitted, every read under a flow ceiling was refused as tainted. Its
/// `_meta` is now accounted for as a call's is.
pub fn relay(&mut self, line: &str, up: &mut impl Upstream) -> std::io::Result<Relayed> {
let msg: Value = match serde_json::from_str(line) {
Ok(v) => v,
@ -759,31 +778,42 @@ impl Gateway {
// ---- A gated method: the enforcement point.
let tool = action.as_str();
// A retry answering an interim result carries the answers beside the
// arguments. They go to the server, so they are payload; and the
// interim they answer is found by the action and the arguments alone.
let (args, answers) = split_answers(method, args, &params);
// Everything the server receives is payload, sealed and flow-checked
// (bar the protocol's own `_meta` keys: `protocol_meta_key`): the
// arguments; a retry's answers to an interim result, found by the
// action and the arguments alone; and whatever the request carries
// beside them.
let (args, answers, extra) = split_request(method, args, &msg);
let args_bytes = serde_json::to_vec(&args).unwrap_or_default();
let round = (action.clone(), ContentId::of_bytes(&args_bytes));
let interim = answers.as_ref().and_then(|_| self.awaiting_input.get(&round).cloned());
let continues = interim.as_ref().map(|i| i.receipt_id);
let (payload, scanned) = match answers {
None => (args_bytes.clone(), args_bytes),
Some(mut p) => {
let none = BTreeSet::new();
let (keys, state) = match &interim {
Some(i) => (&i.keys, i.request_state.as_ref()),
None => (&none, None),
};
let answers = authored(p.get("inputResponses"), p.get("requestState"), keys, state);
let scanned = serde_json::to_vec(&json!({"arguments": &args, "answers": answers}))
.unwrap_or_default();
p.insert("arguments".into(), args);
if let Some(c) = continues {
p.insert("continues".into(), json!(c.to_hex()));
}
(serde_json::to_vec(&Value::Object(p)).unwrap_or_default(), scanned)
let mut parts = vec![args_bytes.clone()];
if let Some(p) = &answers {
let none = BTreeSet::new();
let (keys, state) = match &interim {
Some(i) => (&i.keys, i.request_state.as_ref()),
None => (&none, None),
};
let authored = authored(p.get("inputResponses"), p.get("requestState"), keys, state);
parts.push(serde_json::to_vec(&authored).unwrap_or_default());
}
if !extra.is_empty() {
parts.push(serde_json::to_vec(&extra).unwrap_or_default());
}
let scanned = scan_bytes(&parts);
let payload = if answers.is_none() && extra.is_empty() {
args_bytes
} else {
let mut p = answers.unwrap_or_default();
p.insert("arguments".into(), args);
if let Some(c) = continues {
p.insert("continues".into(), json!(c.to_hex()));
}
if !extra.is_empty() {
p.insert("extra".into(), Value::Object(extra));
}
serde_json::to_vec(&Value::Object(p)).unwrap_or_default()
};
// A gated method sent as a notification (no `id`) cannot be answered:
@ -906,9 +936,17 @@ impl Gateway {
if let Some(answers) = answers {
payload.insert("inputResponses".into(), answers.clone());
}
// The `_meta` goes upstream as sent, so its keys beside the
// protocol's own are sealed and checked with the answers.
let rebuilt = json!({"params": sent});
let extra = beside(&rebuilt, &["taskId"]);
let mut parts = vec![serde_json::to_vec(&authored(answers, None, &task.request_keys, None)).unwrap_or_default()];
if !extra.is_empty() {
parts.push(serde_json::to_vec(&extra).unwrap_or_default());
payload.insert("extra".into(), Value::Object(extra));
}
let payload = serde_json::to_vec(&Value::Object(payload)).unwrap_or_default();
let scanned =
serde_json::to_vec(&authored(answers, None, &task.request_keys, None)).unwrap_or_default();
let scanned = scan_bytes(&parts);
if id.is_none() {
return Ok(self.deny_notification(tool, &payload));
}
@ -1579,7 +1617,7 @@ fn ceiling_digest(grant: &Grant) -> Option<ContentId> {
fn declared_witness_ids(params: &Value) -> BTreeSet<ContentId> {
jcp_meta(params)
.and_then(|m| m.get("witness"))
.or_else(|| params.get("_jcp_witness"))
.or_else(|| params.get(LEGACY_WITNESS_KEY))
.and_then(Value::as_array)
.map(|arr| {
arr.iter()
@ -1738,26 +1776,127 @@ fn finished(response: &str) -> Option<(String, Option<Value>)> {
Some((status.to_string(), outcome))
}
/// A gated call's arguments, and apart from them the answers it carries to
/// an interim result: `inputResponses`, with the `requestState` the server
/// handed back, if any. A call that answers nothing has none. A
/// `resources/read` is gated over its whole `params`, so the answers are
/// taken out of that; a `tools/call` or `prompts/get` is gated over its
/// `arguments`, which are left exactly as sent.
fn split_answers(method: &str, mut args: Value, params: &Value) -> (Value, Option<serde_json::Map<String, Value>>) {
let Some(responses) = params.get("inputResponses") else { return (args, None) };
let mut answers = serde_json::Map::new();
answers.insert("inputResponses".into(), responses.clone());
if let Some(state) = params.get("requestState") {
answers.insert("requestState".into(), state.clone());
}
/// A gated call's parts: its arguments; the answers it carries to an interim
/// result (`inputResponses`, with the `requestState` the server handed back),
/// if any; and what it carries beside both ([`beside`]).
///
/// A `tools/call` or `prompts/get` is gated over its `arguments`, which are
/// left exactly as sent. A `resources/read` is gated over its `params`, so
/// the answers and the `_meta` are taken out of those, and with them this
/// gateway's own legacy witness key: they are accounted for apart.
fn split_request(
method: &str,
mut args: Value,
msg: &Value,
) -> (Value, Option<serde_json::Map<String, Value>>, serde_json::Map<String, Value>) {
let params = msg.get("params").cloned().unwrap_or(Value::Null);
if method == "resources/read" {
if let Some(obj) = args.as_object_mut() {
obj.remove("inputResponses");
obj.remove("requestState");
for k in ["inputResponses", "requestState", "_meta", LEGACY_WITNESS_KEY] {
obj.remove(k);
}
}
}
(args, Some(answers))
let answers = params.get("inputResponses").map(|responses| {
let mut answers = serde_json::Map::new();
answers.insert("inputResponses".into(), responses.clone());
if let Some(state) = params.get("requestState") {
answers.insert("requestState".into(), state.clone());
}
answers
});
let defined: &[&str] = match method {
// Everything else in a read's params is in its payload already.
"resources/read" => &[],
_ => &["name", "arguments"],
};
(args, answers, beside(msg, defined))
}
/// The legacy flat slot for this gateway's witness declarations, read before
/// MCP had a namespaced `_meta` ([`declared_witness_ids`]).
const LEGACY_WITNESS_KEY: &str = "_jcp_witness";
/// What a gated request carries that is neither its routing nor its payload,
/// and that the server would otherwise receive unchecked:
///
/// - `message`: members of the JSON-RPC message beside `jsonrpc`, `id`,
/// `method` and `params`;
/// - `params`: members of `params` beside those the method defines (`defined`,
/// for a call its `name` and `arguments`), the answers to an interim
/// result, `_meta`, and this gateway's legacy witness key — all accounted
/// for elsewhere, and a `resources/read`'s other members are its payload;
/// - `_meta`: `_meta` keys beside the protocol's own ([`protocol_meta_key`])
/// and this gateway's namespace.
///
/// Each is sealed and flow-checked with the arguments. Empty for a request
/// with none, whose payload is then its arguments alone, byte for byte as
/// before.
fn beside(msg: &Value, defined: &[&str]) -> serde_json::Map<String, Value> {
let others = |v: Option<&Value>, skip: &dyn Fn(&str) -> bool| -> serde_json::Map<String, Value> {
v.and_then(Value::as_object)
.map(|m| m.iter().filter(|(k, _)| !skip(k)).map(|(k, v)| (k.clone(), v.clone())).collect())
.unwrap_or_default()
};
let mut out = serde_json::Map::new();
let message = others(Some(msg), &|k| matches!(k, "jsonrpc" | "id" | "method" | "params"));
let params = others(msg.get("params"), &|k| {
defined.contains(&k)
|| matches!(k, "inputResponses" | "requestState" | "_meta")
|| k == LEGACY_WITNESS_KEY
});
let meta = others(msg.get("params").and_then(|p| p.get("_meta")), &|k| {
k == JCP_META_NS || protocol_meta_key(k)
});
for (part, m) in [("message", message), ("params", params), ("_meta", meta)] {
if !m.is_empty() {
out.insert(part.into(), Value::Object(m));
}
}
out
}
/// Whether a `_meta` key is the protocol's own: `progressToken`, or a key
/// under a prefix MCP reserves for itself (one whose second label is
/// `modelcontextprotocol` or `mcp`, as `io.modelcontextprotocol/`).
///
/// These are left out of the flow check. They are the protocol's machinery,
/// written by the client's runtime rather than composed by the model, and a
/// server names them back in every reply (`io.modelcontextprotocol/serverInfo`
/// rides in each result's `_meta`): the flow store admits every payload it
/// labels, so folding them would make the first tainted request taint every
/// request after it. They are the one part of a request the server receives
/// unchecked.
fn protocol_meta_key(key: &str) -> bool {
key == "progressToken"
|| key.split_once('/').is_some_and(|(prefix, _)| {
prefix.split('.').nth(1).is_some_and(|label| label == "modelcontextprotocol" || label == "mcp")
})
}
/// The bytes a gated exchange is flow-checked over, from its parts in order,
/// each as it would be scanned alone. A lone part is scanned as it is, so a
/// call with nothing beside its arguments is scanned exactly as before.
/// Several are joined by a separator of one scan window's length unique to
/// this exchange (a NUL, which JSON text never holds, then a digest of the
/// parts), so no window that spans two parts matches anything but this same
/// exchange: the joining adds no text a later exchange could share. Joining
/// them as one JSON object would add its keys, the same in every exchange of
/// the shape, and the flow store admits every payload it labels, so the first
/// tainted exchange would then taint every one after it.
fn scan_bytes(parts: &[Vec<u8>]) -> Vec<u8> {
if let [one] = parts {
return one.clone();
}
let mut framed = Vec::new();
for p in parts {
framed.extend_from_slice(&(p.len() as u64).to_le_bytes());
framed.extend_from_slice(p);
}
let digest = ContentId::of_bytes(&framed).0;
let mut sep = vec![0u8];
sep.extend_from_slice(&digest[..jcp_flow::SCAN_W - 1]);
parts.join(sep.as_slice())
}
/// The handles the server minted in a reply for the client to hand back:
@ -3766,6 +3905,197 @@ mod tests {
assert_eq!(adv.enforced, JCP_ENFORCED_METHODS);
}
// ── everything the server receives is checked (SPEC §4.10) ──────────────
/// The `_meta` a client of MCP 2026-07-28 puts on every request.
fn protocol_meta() -> Value {
json!({
"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"io.modelcontextprotocol/clientInfo": {"name": "an-mcp-client-for-tests", "version": "1.0.0"},
"io.modelcontextprotocol/clientCapabilities": {"extensions": {"io.modelcontextprotocol/tasks": {}}},
"progressToken": "progress-token-0001",
})
}
/// A gateway that has read the inbox, with `send_email` held to
/// principal-clean arguments.
fn read_the_inbox() -> (Gateway, TaskServer) {
let mut g = mail_gw();
g.set_flow_ceiling("send_email", principal_only());
let mut up = TaskServer::new();
g.relay(&tool_call(1, "read_inbox", json!({})), &mut up).unwrap();
(g, up)
}
/// A request was decided over its arguments and forwarded whole, so a
/// value could leave beside them. Each way out is now checked, on a
/// gateway of its own: a refused payload is admitted under its taint, and
/// a later one sharing a scan window of it would join that taint.
#[test]
fn a_value_cannot_leave_beside_the_arguments() {
let leak = format!("fwd {INBOX_SECRET}");
let ways = [
(
"a _meta key",
json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
"params": {"name": "send_email", "arguments": {"body": "hello"},
"_meta": {"org.example/note": leak}}}),
),
(
"a params member",
json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
"params": {"name": "send_email", "arguments": {"body": "hello"}, "note": leak}}),
),
(
"a message member",
json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "note": leak,
"params": {"name": "send_email", "arguments": {"body": "hello"}}}),
),
(
"a _meta key beside the protocol's own",
json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
"params": {"name": "send_email", "arguments": {"body": "hello"},
"_meta": {"io.modelcontextprotocol/protocolVersion": "2026-07-28",
"org.example/note": leak}}}),
),
];
for (way, request) in ways {
let (mut g, mut up) = read_the_inbox();
match g.relay(&request.to_string(), &mut up).unwrap() {
Relayed::Denied { reason, .. } => assert_eq!(reason, "tainted-flow", "{way}"),
other => panic!("a value from the inbox left by {way}: {other:?}"),
}
assert_eq!(up.seen.len(), 1, "{way}: nothing past the read reached the server");
}
// A clean request carrying the protocol's own `_meta`, and a key of
// its own with a clean value, goes through.
let (mut g, mut up) = read_the_inbox();
let clean = json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
"params": {"name": "send_email", "arguments": {"body": "hello"},
"_meta": {"org.example/note": "routine", "io.modelcontextprotocol/protocolVersion": "2026-07-28"}}});
assert!(matches!(g.relay(&clean.to_string(), &mut up).unwrap(), Relayed::Allowed { .. }));
}
/// A server whose every reply carries the protocol's `_meta`, as MCP
/// 2026-07-28 asks of a server.
struct ProtocolServer;
impl Upstream for ProtocolServer {
fn call(&mut self, line: &str) -> std::io::Result<String> {
let v: Value = serde_json::from_str(line).unwrap_or(Value::Null);
Ok(json!({"jsonrpc": "2.0", "id": v["id"], "result": {
"resultType": "complete",
"contents": [{"text": "a quarterly report"}],
"_meta": {"io.modelcontextprotocol/serverInfo": {"name": "files", "version": "2.0.0"}},
}})
.to_string())
}
}
/// A `resources/read` was gated over its whole `params`, `_meta` and all,
/// so once a reply carrying the protocol's `_meta` was admitted, every
/// read carrying it matched that reply and was refused under a ceiling.
#[test]
fn the_protocols_own_meta_does_not_taint_a_clean_request() {
let mut g = Gateway::new(
"files",
&[("resources/read:file:///a", 50_000_000), ("resources/read:file:///b", 50_000_000)],
SigningKey::from_bytes(&[7u8; 32]),
);
g.set_flow_ceiling("resources/read:file:///b", principal_only());
let mut up = ProtocolServer;
let read = |id: i64, uri: &str| {
json!({"jsonrpc": "2.0", "id": id, "method": "resources/read",
"params": {"uri": uri, "_meta": protocol_meta()}})
.to_string()
};
assert!(matches!(g.relay(&read(1, "file:///a"), &mut up).unwrap(), Relayed::Allowed { .. }));
match g.relay(&read(2, "file:///b"), &mut up).unwrap() {
Relayed::Allowed { .. } => {}
other => panic!("the protocol's own _meta tainted a clean read: {other:?}"),
}
}
/// A request with nothing beside its arguments but the protocol's `_meta`
/// and this gateway's own is decided over its arguments, byte for byte as
/// before; one with more is decided over that too.
#[test]
fn what_a_request_is_sealed_over() {
let mut g = mail_gw();
let mut up = TaskServer::new();
let mut meta = protocol_meta();
meta[JCP_META_NS] = json!({"witness": []});
let plain = json!({"jsonrpc": "2.0", "id": 1, "method": "tools/call",
"params": {"name": "send_email", "arguments": {"body": "hello"}, "_meta": meta,
"_jcp_witness": []}});
g.relay(&plain.to_string(), &mut up).unwrap();
let sealed = jcr1_payload(&g.receipts()[0], MAIL, &mail_pub());
let args = serde_json::to_vec(&json!({"body": "hello"})).unwrap();
assert_eq!(sealed["payload_hash"], ContentId::of_bytes(&args).to_hex().as_str());
assert!(up.seen[0].contains("io.modelcontextprotocol/protocolVersion"), "the _meta is forwarded");
let more = json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "note": "n",
"params": {"name": "send_email", "arguments": {"body": "hello"}, "flag": true,
"_meta": {"org.example/note": "routine", "progressToken": "p-1"}}});
g.relay(&more.to_string(), &mut up).unwrap();
let sealed = jcr1_payload(&g.receipts()[1], MAIL, &mail_pub());
let payload = json!({
"arguments": {"body": "hello"},
"extra": {"message": {"note": "n"}, "params": {"flag": true}, "_meta": {"org.example/note": "routine"}},
});
assert_eq!(
sealed["payload_hash"],
ContentId::of_bytes(&serde_json::to_vec(&payload).unwrap()).to_hex().as_str()
);
}
/// A task's exchange forwards the request's `_meta`, so a key in it beside
/// the protocol's own is checked with the answers.
#[test]
fn a_task_exchange_cannot_carry_a_value_out_in_its_meta() {
let mut g = mail_gw();
g.set_flow_ceiling("read_email", principal_only());
let mut up = TaskServer::new();
g.relay(&tool_call(1, "read_email", json!({"folder": "inbox"})), &mut up).unwrap();
g.relay(&tool_call(2, "read_inbox", json!({})), &mut up).unwrap();
let poll = |id: i64, meta: Value| {
json!({"jsonrpc": "2.0", "id": id, "method": "tasks/get", "params": {"taskId": TASK_ID, "_meta": meta}})
.to_string()
};
assert!(matches!(g.relay(&poll(3, protocol_meta()), &mut up).unwrap(), Relayed::Allowed { .. }));
match g.relay(&poll(4, json!({"org.example/note": INBOX_SECRET})), &mut up).unwrap() {
Relayed::Denied { reason, .. } => assert_eq!(reason, "tainted-flow"),
other => panic!("a poll's _meta carried the inbox out: {other:?}"),
}
}
/// Joining a request's parts adds no text of its own: two requests whose
/// parts share no scan window share none once joined, so one's taint
/// cannot reach the other through the joining.
#[test]
fn joining_parts_adds_no_text_a_later_request_shares() {
let a = br#"{"body":"first request"}"#.to_vec();
let b = br#"{"_meta":{"org.example/x":"one"}}"#.to_vec();
let c = br#"{"text":"another thing entirely"}"#.to_vec();
let d = br#"[{"accept":"different"}]"#.to_vec();
assert_eq!(scan_bytes(std::slice::from_ref(&a)), a, "a lone part is scanned as it is");
let one = jcp_flow::windows(&scan_bytes(&[a.clone(), b.clone()]), jcp_flow::SCAN_W);
let two = jcp_flow::windows(&scan_bytes(&[c.clone(), d.clone()]), jcp_flow::SCAN_W);
assert!(one.is_disjoint(&two), "the joining put shared text into both");
// A repeat of the same parts is the same bytes, and matches itself.
assert_eq!(scan_bytes(&[a.clone(), b.clone()]), scan_bytes(&[a, b]));
}
#[test]
fn the_protocols_own_meta_keys() {
for key in ["progressToken", "io.modelcontextprotocol/protocolVersion", "dev.mcp/x", "io.modelcontextprotocol/clientInfo"] {
assert!(protocol_meta_key(key), "{key}");
}
for key in ["org.example/note", "modelcontextprotocol/x", "io.modelcontextprotocolx/y", "progress", JCP_META_NS, "mcp"] {
assert!(!protocol_meta_key(key), "{key}");
}
}
/// Decode a verified receipt envelope to its JSON payload (test helper).
fn jcr1_payload(envelope: &[u8], steward: &str, pubkey: &[u8; 32]) -> Value {
let core = jcp_receipt::jcr1::verify_cose_sign1(envelope, steward, pubkey).unwrap();