From 30491753038704aee110676299f65de9514955fe Mon Sep 17 00:00:00 2001 From: Orual Date: Sat, 1 Aug 2026 14:46:44 -0400 Subject: [PATCH] GC Unit G: exercise URI guards with valid CIDs --- justfile | 3 + src/appview/proxy.rs | 77 ++++++++++++----- src/appview/proxy_tests.rs | 144 +++++++++++++++++++++++++++++++ tools/check-proxy-response.py | 156 ++++++++++++++++++++++++++++++++++ 4 files changed, 360 insertions(+), 20 deletions(-) create mode 100644 tools/check-proxy-response.py diff --git a/justfile b/justfile index da1f1a7..cd27fca 100644 --- a/justfile +++ b/justfile @@ -121,6 +121,9 @@ validate-gc-jj-state *ARGS: check-content-type *ARGS: python3 tools/check-content-type.py {{ ARGS }} +# Validate the borrowed proxy response-reference parser and raw forwarding boundary. +check-proxy-response *ARGS: + python3 tools/check-proxy-response.py {{ ARGS }} # Create a sibling jj workspace for a Jira ticket or feature slug, then start a fresh change. workspace-create name: root="$(jj workspace root)" && workspace_path="$(dirname "$root")/{{ name }}" && jj workspace add "$workspace_path" -r @ diff --git a/src/appview/proxy.rs b/src/appview/proxy.rs index fa60274..2c40e66 100644 --- a/src/appview/proxy.rs +++ b/src/appview/proxy.rs @@ -33,7 +33,10 @@ use jacquard::oauth::client::OAuthSession; use jacquard_axum::ExtractXrpc; use jacquard_axum::oauth::ExtractOAuthSession; use jacquard_common::error::{ClientError, ClientErrorKind, HttpError}; -use jacquard_common::types::string::Did; +use jacquard_common::types::ident::AtIdentifier; +use jacquard_common::types::nsid::Nsid; +use jacquard_common::types::recordkey::Rkey; +use jacquard_common::types::string::{AtUri, Cid, Did}; use jacquard_common::types::value::to_data; use jacquard_common::xrpc::{Response as XrpcResponse, XrpcClient, XrpcResp}; use polymodel_api::com_atproto::identity::resolve_handle::ResolveHandleRequest; @@ -99,9 +102,7 @@ pub(super) async fn repo_get_record( // pending write still wins when a lagging source reports not-found. let pds_cid = match &result { Ok(resp) if resp.status().is_success() => { - serde_json::from_slice::(resp.buffer()) - .ok() - .and_then(|v| v.get("cid").and_then(|c| c.as_str()).map(str::to_owned)) + parse_record_cid(resp.buffer()).map(|cid| cid.as_str().to_owned()) } _ => None, }; @@ -143,12 +144,12 @@ pub(super) async fn repo_create_record( ) -> AppResult { let agent = Agent::from(session); let did = actor_did(&agent).await?; - let collection = req.collection.as_ref().to_string(); + let collection = req.collection.clone(); let record = serde_json::to_value(&req.record).ok(); let result = agent.send(req).await; if let Ok(resp) = &result && resp.status().is_success() - && collection.starts_with(POLYMODEL_PREFIX) + && collection.as_str().starts_with(POLYMODEL_PREFIX) && let Some(record) = record.as_ref() { eager_project(&state, &did, &collection, "create", record, resp.buffer()).await; @@ -163,12 +164,12 @@ pub(super) async fn repo_put_record( ) -> AppResult { let agent = Agent::from(session); let did = actor_did(&agent).await?; - let collection = req.collection.as_ref().to_string(); + let collection = req.collection.clone(); let record = serde_json::to_value(&req.record).ok(); let result = agent.send(req).await; if let Ok(resp) = &result && resp.status().is_success() - && collection.starts_with(POLYMODEL_PREFIX) + && collection.as_str().starts_with(POLYMODEL_PREFIX) && let Some(record) = record.as_ref() { eager_project(&state, &did, &collection, "update", record, resp.buffer()).await; @@ -311,12 +312,13 @@ async fn actor_did(agent: &UserAgent) -> AppResult { async fn eager_project( state: &AppState, did: &Did, - collection: &str, + collection: &Nsid, action: &str, record: &Value, output_buffer: &[u8], ) { - let Some((rkey, cid)) = parse_record_ref(output_buffer) else { + let Some((rkey, cid)) = parse_record_ref(output_buffer, did, collection) else { + tracing::warn!(%collection, "invalid proxied record reference; skipping eager projection"); return; }; let record_data = match to_data(record) { @@ -329,11 +331,11 @@ async fn eager_project( let event = ProjectionEvent { seq: 0, did: did.as_ref().to_string(), - collection: collection.to_string(), - rkey, + collection: collection.as_str().to_string(), + rkey: rkey.as_str().to_string(), action: action.to_string(), record: Some(record_data), - cid: Some(cid), + cid: Some(cid.as_str().to_owned()), }; // Serialize against PM-28's decision-making writes (which read local state // then mutate): the proxied write itself makes no decision, but its local @@ -345,13 +347,48 @@ async fn eager_project( } } -/// Parse `(rkey, cid)` from a createRecord/putRecord output's raw JSON. -fn parse_record_ref(output: &[u8]) -> Option<(String, String)> { - let out: Value = serde_json::from_slice(output).ok()?; - let uri = out.get("uri")?.as_str()?; - let cid = out.get("cid")?.as_str()?.to_string(); - let rkey = uri.rsplit('/').next()?.to_string(); - Some((rkey, cid)) +#[derive(serde::Deserialize)] +struct RecordRef<'a> { + #[serde(borrow)] + uri: AtUri<&'a str>, + #[serde(borrow)] + cid: Cid<&'a str>, +} + +#[derive(serde::Deserialize)] +struct RecordCid<'a> { + #[serde(borrow)] + cid: Cid<&'a str>, +} + +/// Parse and validate `(rkey, cid)` from a createRecord/putRecord response. +pub(super) fn parse_record_ref<'a>( + output: &'a [u8], + expected_did: &Did, + expected_collection: &Nsid, +) -> Option<(Rkey<&'a str>, Cid<&'a str>)> { + let out: RecordRef<'a> = serde_json::from_slice(output).ok()?; + if !out.cid.is_valid() { + return None; + } + match out.uri.authority() { + AtIdentifier::Did(did) if did.as_ref() == expected_did.as_ref() => {} + _ => return None, + } + if out.uri.collection()?.as_ref() != expected_collection.as_str() { + return None; + } + let rkey = out.uri.rkey()?; + if rkey.as_str().is_empty() { + return None; + } + Some((rkey, out.cid)) +} + +/// Parse and semantically validate the CID from a getRecord response. +pub(super) fn parse_record_cid<'a>(output: &'a [u8]) -> Option> { + let out: RecordCid<'a> = serde_json::from_slice(output).ok()?; + out.cid.is_valid().then_some(out.cid) } /// Read-your-writes override for a single `space.polymodel.*` getRecord. diff --git a/src/appview/proxy_tests.rs b/src/appview/proxy_tests.rs index a26b7d5..3f802cb 100644 --- a/src/appview/proxy_tests.rs +++ b/src/appview/proxy_tests.rs @@ -6,6 +6,150 @@ use jacquard_common::xrpc::Response as XrpcResponse; use polymodel_api::com_atproto::repo::get_record::GetRecordResponse; use crate::indexing::projection::{ProjectionEvent, ProjectionInput, project_eager_record, project_event}; +#[test] +fn parse_record_ref_accepts_valid_create_response() { + let did = did(DID_A); + let collection = + jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); + let output = format!( + r#"{{"uri":"at://{DID_A}/space.polymodel.library.thing/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}}"# + ); + let (rkey, cid) = super::proxy::parse_record_ref(output.as_bytes(), &did, &collection) + .expect("valid typed response"); + assert_eq!(rkey.as_str(), "t1"); + assert!(cid.is_valid()); +} + +#[test] +fn parse_record_ref_rejects_malformed_json() { + let did = did(DID_A); + let collection = + jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); + assert!(super::proxy::parse_record_ref(b"not json", &did, &collection).is_none()); +} + +#[test] +fn parse_record_ref_rejects_missing_fields() { + let did = did(DID_A); + let collection = + jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); + assert!( + super::proxy::parse_record_ref( + br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/t1"}"#, + &did, + &collection, + ) + .is_none() + ); +} + +#[test] +fn parse_record_ref_rejects_trailing_slash() { + let did = did(DID_A); + let collection = + jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); + assert!(super::proxy::parse_record_ref( + br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, + &did, + &collection, + ).is_none()); +} + +#[test] +fn parse_record_ref_rejects_mismatched_authority() { + let did = did(DID_A); + let collection = + jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); + assert!(super::proxy::parse_record_ref( + br#"{"uri":"at://did:plc:bbbbbbbbbbbbbbbbbbbbbbbb/space.polymodel.library.thing/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, + &did, + &collection, + ).is_none()); +} + +#[test] +fn parse_record_ref_rejects_mismatched_collection() { + let did = did(DID_A); + let collection = + jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); + assert!(super::proxy::parse_record_ref( + br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.model/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, + &did, + &collection, + ).is_none()); +} + +#[test] +fn parse_record_ref_rejects_empty_rkey() { + let did = did(DID_A); + let collection = + jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); + assert!(super::proxy::parse_record_ref( + br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, + &did, + &collection, + ).is_none()); +} + +#[test] +fn parse_record_ref_rejects_invalid_cid() { + let did = did(DID_A); + let collection = + jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); + assert!(super::proxy::parse_record_ref( + br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/t1","cid":"not-a-cid"}"#, + &did, + &collection, + ).is_none()); +} + +#[test] +fn parse_record_cid_accepts_valid_get_record_response() { + let cid = super::proxy::parse_record_cid( + br#"{"cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, + ) + .expect("valid cid"); + assert!(cid.is_valid()); +} + +#[test] +fn parse_record_cid_rejects_malformed_json() { + assert!(super::proxy::parse_record_cid(b"not json").is_none()); +} + +#[test] +fn parse_record_cid_rejects_missing_cid() { + assert!(super::proxy::parse_record_cid(br#"{"uri":"at://did:plc:a"}"#).is_none()); +} + +#[test] +fn parse_record_cid_rejects_invalid_cid() { + assert!(super::proxy::parse_record_cid(br#"{"cid":"not-a-cid"}"#).is_none()); +} + +#[tokio::test] +async fn forward_xrpc_preserves_success_body_and_json_content_type() { + let body = b"{\"uri\":\"at://x\"}"; + let resp = super::proxy::forward_xrpc::(Ok(XrpcResponse::new( + Bytes::from_static(body), + StatusCode::OK, + ))); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(resp.headers()["content-type"], "application/json"); + assert_eq!(body_text(resp).await, std::str::from_utf8(body).unwrap()); +} + +#[tokio::test] +async fn forward_xrpc_preserves_error_status_body_and_json_content_type() { + let body = b"{\"error\":\"InvalidSwap\"}"; + let resp = super::proxy::forward_xrpc::(Err(ClientError::http( + StatusCode::CONFLICT, + Some(Bytes::from_static(body)), + ))); + assert_eq!(resp.status(), StatusCode::CONFLICT); + assert_eq!(resp.headers()["content-type"], "application/json"); + assert_eq!(body_text(resp).await, std::str::from_utf8(body).unwrap()); +} // --------------------------------------------------------------------------- // PM-47 passthrough proxy: faithful forwarding, read-your-writes override, diff --git a/tools/check-proxy-response.py b/tools/check-proxy-response.py new file mode 100644 index 0000000..6caa468 --- /dev/null +++ b/tools/check-proxy-response.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 +"""Check the typed response-reference and raw proxy forwarding boundaries.""" +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +PROXY = ROOT / "src" / "appview" / "proxy.rs" + + +class CheckError(ValueError): + pass + + +def block_from(source: str, marker: str, next_marker: str) -> str: + start = source.find(marker) + if start < 0: + raise CheckError(f"missing source marker: {marker}") + end = source.find(next_marker, start + len(marker)) + if end < 0: + raise CheckError(f"missing source marker: {next_marker}") + return source[start:end] + + +def parser_region(source: str) -> str: + return block_from(source, "struct RecordRef", "/// Read-your-writes override") + + +def repo_get_record_region(source: str) -> str: + return block_from(source, "pub(super) async fn repo_get_record", "pub(super) async fn repo_list_records") + + +def forwarding_region(source: str) -> str: + return block_from(source, "pub(super) fn forward_xrpc", "async fn actor_did") + + +def raw_response_region(source: str) -> str: + return block_from(source, "fn raw_response", "async fn actor_did") + + +def proxy_response_parser_has_borrowed_typed_envelopes(source: str) -> None: + region = parser_region(source) + required = ( + "struct RecordRef<'a>", "uri: AtUri<&'a str>", "cid: Cid<&'a str>", + "struct RecordCid<'a>", "fn parse_record_ref<'a>", "fn parse_record_cid<'a>", + "expected_did: &Did", "expected_collection: &Nsid", "Rkey<&'a str>", + "out.uri.authority()", "out.uri.collection()?", "out.uri.rkey()?", + "out.cid.is_valid()", + ) + for needle in required: + if needle not in region: + raise CheckError(f"borrowed response parser is missing {needle}") + for needle in ("serde_json::from_slice::", ".get(\"cid\")", "Cid::new_owned", ".rsplit('/')"): + if needle in region: + raise CheckError(f"response parser retains forbidden extraction: {needle}") + + +def proxy_get_record_uses_borrowed_cid_parser(source: str) -> None: + region = repo_get_record_region(source) + if "parse_record_cid(resp.buffer())" not in region: + raise CheckError("repo_get_record does not use parse_record_cid") + for needle in ("serde_json::from_slice::", ".get(\"cid\")"): + if needle in region: + raise CheckError(f"repo_get_record retains manual CID extraction: {needle}") + + +def proxy_forwarding_contract_is_unchanged(source: str) -> None: + region = forwarding_region(source) + for needle in ("Ok(resp) => raw_response(resp.status(), resp.buffer().to_vec())", "ClientErrorKind::Http { status }", "raw_response(*status, body)"): + if needle not in region: + raise CheckError(f"forward_xrpc no longer preserves {needle}") + if 'header::CONTENT_TYPE, "application/json"' not in raw_response_region(source): + raise CheckError("raw_response no longer synthesizes application/json") + + +def run_checks(source: str) -> None: + proxy_response_parser_has_borrowed_typed_envelopes(source) + proxy_get_record_uses_borrowed_cid_parser(source) + proxy_forwarding_contract_is_unchanged(source) + + +def positive_fixture() -> str: + return ''' + pub(super) async fn repo_get_record() { + let pds_cid = parse_record_cid(resp.buffer()).map(|cid| cid.as_str().to_owned()); + } + pub(super) async fn repo_list_records() {} + pub(super) fn forward_xrpc(result: Result, ClientError>) -> Response { + match result { + Ok(resp) => raw_response(resp.status(), resp.buffer().to_vec()), + Err(err) => match err.kind() { + ClientErrorKind::Http { status } => raw_response(*status, body), + }, + } + } + fn raw_response(status: StatusCode, body: Vec) -> Response { + Response::builder().header(header::CONTENT_TYPE, "application/json").body(body).unwrap() + } + async fn actor_did() {} + #[derive(serde::Deserialize)] + struct RecordRef<'a> { uri: AtUri<&'a str>, cid: Cid<&'a str> } + #[derive(serde::Deserialize)] + struct RecordCid<'a> { cid: Cid<&'a str> } + fn parse_record_ref<'a>(output: &'a [u8], expected_did: &Did, expected_collection: &Nsid) -> Option<(Rkey<&'a str>, Cid<&'a str>)> { + let out: RecordRef<'a> = serde_json::from_slice(output).ok()?; + if !out.cid.is_valid() { return None; } + match out.uri.authority() { AtIdentifier::Did(did) if did.as_ref() == expected_did.as_ref() => {}, _ => return None } + if out.uri.collection()?.as_ref() != expected_collection.as_str() { return None; } + let rkey = out.uri.rkey()?; + if rkey.as_str().is_empty() { return None; } + Some((rkey, out.cid)) + } + fn parse_record_cid<'a>(output: &'a [u8]) -> Option> { + let out: RecordCid<'a> = serde_json::from_slice(output).ok()?; + out.cid.is_valid().then_some(out.cid) + } + /// Read-your-writes override + ''' + + +def self_test() -> None: + run_checks(positive_fixture()) + cases = { + "parser": positive_fixture().replace("uri: AtUri<&'a str>", "uri: Value"), + "get_record": positive_fixture().replace("parse_record_cid(resp.buffer()).map(|cid| cid.as_str().to_owned())", "serde_json::from_slice::(resp.buffer()).ok()"), + "forwarding": positive_fixture().replace("Ok(resp) => raw_response(resp.status(), resp.buffer().to_vec()),", "Ok(resp) => Response::new(Body::from(resp.buffer().to_vec())),"), + } + for name, fixture in cases.items(): + try: + run_checks(fixture) + except CheckError: + continue + raise CheckError(f"negative self-test fixture unexpectedly passed: {name}") + print("check-proxy-response self-test passed") + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--self-test", action="store_true") + args = parser.parse_args() + try: + if args.self_test: + self_test() + else: + run_checks(PROXY.read_text()) + print("proxy response checks passed") + except (CheckError, OSError) as exc: + print(f"check-proxy-response failed: {exc}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) -- knotmirror 0.0.1