diff --git a/Cargo.lock b/Cargo.lock index d8ff175f..94fba90e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -965,11 +965,13 @@ dependencies = [ name = "didbot-agentd" version = "0.1.0" dependencies = [ + "axum", "didbot-http", "reqwest", "serde", "serde_json", "thiserror 2.0.20", + "time", "tokio", "tracing", "tracing-subscriber", diff --git a/crates/didbot-agentd/Cargo.toml b/crates/didbot-agentd/Cargo.toml index 98ba27d3..748a6b2a 100644 --- a/crates/didbot-agentd/Cargo.toml +++ b/crates/didbot-agentd/Cargo.toml @@ -14,10 +14,16 @@ reqwest.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true +time.workspace = true tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true url.workspace = true +# A double for the four decision routes, in process, so the client is tested +# against real HTTP rather than against a trait it was written to satisfy. +[dev-dependencies] +axum.workspace = true + [lints] workspace = true diff --git a/crates/didbot-agentd/src/bin/didbot-agentd.rs b/crates/didbot-agentd/src/bin/didbot-agentd.rs index 7389bbbd..fdc97e1c 100644 --- a/crates/didbot-agentd/src/bin/didbot-agentd.rs +++ b/crates/didbot-agentd/src/bin/didbot-agentd.rs @@ -49,7 +49,11 @@ async fn main() -> ExitCode { } }; - let daemon = Arc::new(Daemon::new(Pds::new(&server, HARNESS)).confirming_at(&server)); + let daemon = Arc::new( + Daemon::new(Pds::new(&server, HARNESS)) + .confirming_at(&server) + .deciding_at(&server), + ); let err = daemon.run(listener).await; error!(error = %err, "stopped listening"); ExitCode::FAILURE diff --git a/crates/didbot-agentd/src/bin/didbot.rs b/crates/didbot-agentd/src/bin/didbot.rs index bd84f524..70e5606c 100644 --- a/crates/didbot-agentd/src/bin/didbot.rs +++ b/crates/didbot-agentd/src/bin/didbot.rs @@ -1,29 +1,42 @@ //! The command an agent runs. //! -//! One job so far: hand the daemon an authorize URL a client printed, so the -//! authorization is confirmed as the account this command names. +//! An app asks to sign in as an agent, the server turns that into a decision, +//! and the daemon puts it in front of the agent. These commands are the +//! agent's side of that: see what is waiting, say yes, or say no. //! -//! It names its own account, and nothing checks that it is that account. See -//! `didbot_agentd::protocol::Confirm` for what that means and why it is -//! where this development stack already stands. +//! Nothing here names an account. `approve` and `decline` carry a one-time +//! token and nothing else, and which account signs in is read off the record +//! the daemon is holding that token in — see +//! `didbot_agentd::protocol::Approve`. `confirm` is the older command that +//! did name one, kept for one release; see `didbot_agentd::protocol::Confirm` +//! for what taking that from the caller costs. use std::io::{BufRead, BufReader, Write}; use std::os::unix::net::UnixStream; use std::process::ExitCode; -use didbot_agentd::protocol::{Answer, Confirm, Message, VERSION}; +use didbot_agentd::protocol::{ + Answer, Approve, Confirm, DecisionForAgent, Decline, Message, Pending, VERSION, +}; use didbot_agentd::socket::default_socket_path; const USAGE: &str = "\ -didbot confirm --as confirm an authorization a client printed +didbot pending what has asked to sign in as you +didbot approve let one of them in +didbot decline [--reason WHY] + turn one of them down +didbot confirm --as superseded by `approve`; kept for one release -The client prints its authorize URL instead of opening it; this hands that URL -to the local daemon, which confirms it as the account named here. +Nothing signs in until you choose, and a request nobody answers expires on its +own. A token names one request: it is single use and it is not an account. "; fn main() -> ExitCode { let args: Vec = std::env::args().skip(1).collect(); match args.first().map(String::as_str) { + Some("pending") => pending(), + Some("approve") => approve(&args[1..]), + Some("decline") => decline(&args[1..]), Some("confirm") => confirm(&args[1..]), Some("--help") | Some("-h") | None => { print!("{USAGE}"); @@ -36,12 +49,160 @@ fn main() -> ExitCode { } } +/// The first argument that is not a flag or a flag's value. +fn positional(args: &[String]) -> Option<&String> { + let mut skip = false; + for arg in args { + if skip { + skip = false; + continue; + } + if arg.starts_with("--") { + skip = !arg.contains('='); + continue; + } + return Some(arg); + } + None +} + +/// The value of `--name`, written either way round. +fn flag<'a>(args: &'a [String], name: &str) -> Option<&'a str> { + let long = format!("--{name}"); + let joined = format!("--{name}="); + for (at, arg) in args.iter().enumerate() { + if arg == &long { + return args.get(at + 1).map(String::as_str); + } + if let Some(value) = arg.strip_prefix(&joined) { + return Some(value); + } + } + None +} + +fn pending() -> ExitCode { + match ask(&Message::Pending(Pending { version: VERSION })) { + Ok(answer) => { + if let Some(trouble) = answer.trouble { + eprintln!("didbot pending: {trouble}"); + return ExitCode::FAILURE; + } + for decision in answer.pending.unwrap_or_default() { + println!("{}", one_line(&decision)); + } + ExitCode::SUCCESS + } + Err(err) => { + eprintln!("didbot pending: {err}"); + ExitCode::FAILURE + } + } +} + +/// One decision on one line, with the token last. +/// +/// Last because it is the part that gets copied into the next command, and +/// because a line that is cut short by a narrow terminal should lose the +/// commentary rather than the thing being named. +fn one_line(decision: &DecisionForAgent) -> String { + let mut line = format!( + "{} {} asked={} verdict={}", + decision.client_origin, + if decision.first_time { + "first-time" + } else { + "seen-before" + }, + list(&decision.requested), + decision.verdict, + ); + if decision.granted != decision.requested { + line.push_str(&format!(" granted={}", list(&decision.granted))); + } + if !decision.cut.is_empty() { + line.push_str(&format!(" cut={}", list(&decision.cut))); + } + if let Some(rule) = &decision.rule { + line.push_str(&format!(" rule={rule:?}")); + } + line.push_str(&format!(" expires={}", decision.expires_at)); + match &decision.token { + Some(token) => line.push_str(&format!(" token={token}")), + None => line.push_str(" token=none"), + } + line +} + +/// A scope set, as one field of a line. +fn list(scopes: &[String]) -> String { + if scopes.is_empty() { + "none".to_owned() + } else { + scopes.join(",") + } +} + +fn approve(args: &[String]) -> ExitCode { + let Some(token) = positional(args) else { + eprintln!("didbot approve: needs the token from `didbot pending`"); + return ExitCode::FAILURE; + }; + + match ask(&Message::Approve(Approve { + version: VERSION, + token: token.clone(), + })) { + Ok(answer) => { + if let Some(trouble) = answer.trouble { + eprintln!("didbot approve: {trouble}"); + return ExitCode::FAILURE; + } + println!("{}", answer.done.as_deref().unwrap_or("approved")); + // What was actually granted, which is not always what was asked + // for: a narrowed request grants less, and saying so here is the + // last chance the agent has to notice. + if let Some(granted) = answer.granted { + println!("granted: {}", list(&granted)); + } + ExitCode::SUCCESS + } + Err(err) => { + eprintln!("didbot approve: {err}"); + ExitCode::FAILURE + } + } +} + +fn decline(args: &[String]) -> ExitCode { + let Some(token) = positional(args) else { + eprintln!("didbot decline: needs the token from `didbot pending`"); + return ExitCode::FAILURE; + }; + + match ask(&Message::Decline(Decline { + version: VERSION, + token: token.clone(), + reason: flag(args, "reason").map(str::to_owned), + })) { + Ok(answer) => { + if let Some(trouble) = answer.trouble { + eprintln!("didbot decline: {trouble}"); + return ExitCode::FAILURE; + } + println!("{}", answer.done.as_deref().unwrap_or("declined")); + ExitCode::SUCCESS + } + Err(err) => { + eprintln!("didbot decline: {err}"); + ExitCode::FAILURE + } + } +} + fn confirm(args: &[String]) -> ExitCode { - let url = args.iter().find(|arg| !arg.starts_with("--")); - let did = args - .iter() - .position(|arg| arg == "--as") - .and_then(|at| args.get(at + 1)); + let url = positional(args); + let did = flag(args, "as"); let (Some(url), Some(did)) = (url, did) else { eprintln!("didbot confirm: needs the URL the client printed and `--as `"); @@ -50,7 +211,7 @@ fn confirm(args: &[String]) -> ExitCode { let message = Message::Confirm(Confirm { version: VERSION, - did: did.clone(), + did: did.to_owned(), url: url.clone(), }); @@ -95,3 +256,90 @@ fn ask(message: &Message) -> std::io::Result { ) }) } + +#[cfg(test)] +mod tests { + use super::*; + + fn args(raw: &[&str]) -> Vec { + raw.iter().map(|arg| (*arg).to_owned()).collect() + } + + #[test] + fn a_token_is_found_whichever_side_of_the_flags_it_is_on() { + assert_eq!(positional(&args(&["k7f3"])).unwrap(), "k7f3"); + assert_eq!( + positional(&args(&["--reason", "no", "k7f3"])).unwrap(), + "k7f3" + ); + assert_eq!( + positional(&args(&["k7f3", "--reason", "no"])).unwrap(), + "k7f3" + ); + // A reason that reads like a token is still the reason. + assert_eq!( + positional(&args(&["--reason=nope", "k7f3"])).unwrap(), + "k7f3" + ); + assert!(positional(&args(&["--reason", "no"])).is_none()); + } + + #[test] + fn a_reason_may_be_written_either_way_round() { + assert_eq!(flag(&args(&["--reason", "no"]), "reason"), Some("no")); + assert_eq!(flag(&args(&["--reason=no"]), "reason"), Some("no")); + assert_eq!(flag(&args(&["k7f3"]), "reason"), None); + } + + fn decision() -> DecisionForAgent { + DecisionForAgent { + token: Some("k7f3".into()), + client_origin: "http://127.0.0.1:40831".into(), + first_time: true, + requested: vec!["atproto".into(), "repo:com.example.thing".into()], + granted: vec!["atproto".into()], + cut: vec!["repo:com.example.thing".into()], + rule: Some("ceiling".into()), + verdict: "narrow".into(), + expires_at: "2026-09-09T12:04:00Z".into(), + } + } + + #[test] + fn a_decision_is_one_line_and_the_token_is_the_end_of_it() { + let line = one_line(&decision()); + assert!(!line.contains('\n')); + assert!(line.ends_with("token=k7f3"), "{line}"); + assert!( + line.starts_with("http://127.0.0.1:40831 first-time"), + "{line}" + ); + for part in ["asked=", "granted=", "cut=", "rule=", "expires="] { + assert!(line.contains(part), "{part} missing from {line}"); + } + } + + #[test] + fn one_that_was_refused_says_so_rather_than_offering_a_token() { + let mut refused = decision(); + refused.token = None; + refused.verdict = "deny".into(); + refused.granted = Vec::new(); + let line = one_line(&refused); + assert!(line.contains("verdict=deny"), "{line}"); + assert!(line.contains("granted=none"), "{line}"); + assert!(line.ends_with("token=none"), "{line}"); + } + + #[test] + fn a_request_granted_in_full_does_not_repeat_itself() { + let mut whole = decision(); + whole.verdict = "allow".into(); + whole.granted = whole.requested.clone(); + whole.cut = Vec::new(); + whole.rule = None; + let line = one_line(&whole); + assert!(!line.contains("granted="), "{line}"); + assert!(!line.contains("cut="), "{line}"); + } +} diff --git a/crates/didbot-agentd/src/confirm.rs b/crates/didbot-agentd/src/confirm.rs index db4a8ee7..06b04f75 100644 --- a/crates/didbot-agentd/src/confirm.rs +++ b/crates/didbot-agentd/src/confirm.rs @@ -6,6 +6,10 @@ //! for what that costs and why it is where this development stack already //! stands. //! +//! Superseded by the approval path in [`crate::serve`]: a daemon holding the +//! decision the URL names approves it by token instead of reading the page. +//! This is what happens when it is not, and stays for one release. +//! //! Three things bound what this will fetch, because the URL came from a //! process the model started: //! diff --git a/crates/didbot-agentd/src/context.rs b/crates/didbot-agentd/src/context.rs index a0277843..63322ed3 100644 --- a/crates/didbot-agentd/src/context.rs +++ b/crates/didbot-agentd/src/context.rs @@ -203,6 +203,7 @@ mod tests { kind: None, asker: None, call: None, + seen_request_uris: Vec::new(), } } diff --git a/crates/didbot-agentd/src/decisions.rs b/crates/didbot-agentd/src/decisions.rs new file mode 100644 index 00000000..df1d207f --- /dev/null +++ b/crates/didbot-agentd/src/decisions.rs @@ -0,0 +1,572 @@ +//! Sign-in decisions, as the server states them and as this daemon reads them. +//! +//! A pushed authorization request naming an agent account becomes a record at +//! the server: which client asked, what it asked for, and what policy made of +//! that. The daemon does not decide any of it. It fetches records for the +//! accounts it issued, hands them to the agent to choose, and carries the +//! choice back — see `plan/oauth.md`. +//! +//! Four routes, all `bot.did.*` XRPC, all authenticated the same way every +//! other agent-scoped route on that server is: `Authorization: Bearer` with +//! the account's own agent token (`didbot_serve::auth::require_agent_token`). +//! The daemon holds that token because it was handed one at provisioning and +//! kept it ([`crate::secret`]); no new credential kind exists for this. +//! +//! The wire vocabulary is the server's. A field this daemon does not +//! understand is skipped and a `state` it has never heard of is carried +//! through as text, because a server that grows a word should not make a +//! whole page of records unreadable to a daemon that has not been updated. + +use std::time::Duration; + +use serde::{Deserialize, Serialize}; +use url::Url; + +use crate::protocol::DecisionForAgent; +use crate::secret::Secret; + +/// How long the server is asked to hold a poll open, in seconds. +/// +/// The spec's ceiling is 30. Twenty-five leaves room under it for the +/// server's own slack and under this client's timeout for the round trip. +pub const WAIT: u32 = 25; + +/// An account this daemon can act as, and what proves it. +#[derive(Debug, Clone)] +pub struct Account { + /// The account's DID. + pub did: String, + /// Its agent token, held in memory and never written anywhere. + pub token: Secret, +} + +/// What the server says about one pending authorization. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Record { + /// The pushed request this is about, and the key everything holds it by. + pub request_uri: String, + /// The account the request named. + pub account: String, + /// Who asked. + pub client: Client, + /// The scopes the client asked for. + #[serde(default)] + pub requested: Vec, + /// What policy made of the request. + pub verdict: Verdict, + /// The one-time approval token, present only when there is something to + /// approve: a refused request carries none. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub token: Option, + /// When the server made the record. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub created_at: Option, + /// When it stops being answerable, RFC 3339. + pub expires_at: String, + /// Where the record has got to. + /// + /// Text rather than an enumeration on purpose. This is the server's + /// vocabulary, and a daemon that refuses to parse a word it has not seen + /// would lose every other record in the same page along with it. + /// [`Record::settled`] is the only question this crate asks of it. + #[serde(default = "pending")] + pub state: String, +} + +/// The state a record has when the server did not name one. +fn pending() -> String { + "pending".to_owned() +} + +/// Who asked to sign in. +/// +/// The client's own description of itself — its name, its homepage, its logo +/// — is logged with the record at the server and deliberately not sent here. +/// What an agent gets is where the request came from and whether this account +/// has seen that client before. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Client { + /// The origin the request came from. + pub origin: String, + /// The content-keyed identifier for the client's metadata. + pub key: String, + /// Whether this account has seen this client before. + #[serde(default)] + pub first_time: bool, +} + +/// What policy made of a request. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum Verdict { + /// Everything asked for may be granted. + Allow, + /// Some of it may, and the rest was cut. Never silently: `cut` and `rule` + /// are what makes it possible to say so. + Narrow { + /// What may be granted. + granted: Vec, + /// What was taken out of the request. + cut: Vec, + /// The rule that took it. + rule: String, + }, + /// None of it may. + Deny { + /// Why not, in one sentence. + reason: String, + /// The rule that refused. + rule: String, + }, +} + +impl Record { + /// Whether the record has stopped being answerable. + /// + /// Anything but `pending`: approved, declined, expired, auto-confirmed, + /// or a word this daemon has never heard. All of them mean the same thing + /// here, which is that there is no longer a choice to put in front of an + /// agent. + pub fn settled(&self) -> bool { + self.state != "pending" + } + + /// The scopes that would actually be granted. + pub fn granted(&self) -> Vec { + match &self.verdict { + Verdict::Allow => self.requested.clone(), + Verdict::Narrow { granted, .. } => granted.clone(), + Verdict::Deny { .. } => Vec::new(), + } + } +} + +impl From<&Record> for DecisionForAgent { + fn from(record: &Record) -> Self { + let (verdict, cut, rule) = match &record.verdict { + Verdict::Allow => ("allow", Vec::new(), None), + Verdict::Narrow { cut, rule, .. } => ("narrow", cut.clone(), Some(rule.clone())), + Verdict::Deny { reason, rule } => { + ("deny", Vec::new(), Some(format!("{rule}: {reason}"))) + } + }; + Self { + token: record.token.clone(), + client_origin: record.client.origin.clone(), + first_time: record.client.first_time, + requested: record.requested.clone(), + granted: record.granted(), + cut, + rule, + verdict: verdict.to_owned(), + expires_at: record.expires_at.clone(), + } + } +} + +/// A page of pending records, and where to ask from next time. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Page { + /// The cursor to send on the next poll. + #[serde(default)] + pub cursor: Option, + /// What is new since the cursor that was sent. + #[serde(default)] + pub pending: Vec, +} + +/// What an approval produced. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Approved { + /// The request that was approved. + pub request_uri: String, + /// The scopes the code was issued at, which is the granted set and not + /// necessarily the requested one. + #[serde(default)] + pub granted: Vec, + /// The client's own callback, with the code on it, for the daemon to + /// fetch. The server cannot: the client listens on this host. + #[serde(default)] + pub redirect: Option, +} + +/// What a refusal produced. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Declined { + /// The request that was declined. + pub request_uri: String, +} + +/// Why a call did not produce an answer. +#[derive(Debug, thiserror::Error)] +pub enum Trouble { + /// Nothing answered, so nothing was decided. + #[error("could not reach the server: {0}")] + Unreachable(String), + /// The server answered, and the answer was no. + #[error("the server refused: {0}")] + Refused(String), + /// Something answered and it was not this route. + #[error("the server answered with something unexpected: {0}")] + Unexpected(String), +} + +/// The error body every refusal on this server carries. +#[derive(Deserialize)] +struct ErrorBody { + error: String, + message: String, +} + +/// The server, as the four decision routes. +#[derive(Debug, Clone)] +pub struct Pds { + base: String, + http: reqwest::Client, +} + +impl Pds { + /// `base` is the origin the daemon reaches the server on. + pub fn new(base: impl Into) -> Self { + Self { + base: base.into().trim_end_matches('/').to_string(), + // Long enough for a held poll, and redirects off for the same + // reason [`crate::confirm`] turns them off: nothing this daemon + // fetches gets to choose where the next request goes. + http: didbot_http::builder() + .timeout(Duration::from_secs(u64::from(WAIT) + 10)) + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap_or_default(), + } + } + + /// The pending records for one account, since a cursor. + /// + /// Long-polls: the server answers at once when there is anything newer + /// than `cursor`, and otherwise holds the request for up to `wait` + /// seconds. A held poll that returns nothing is the ordinary case and not + /// an error. + pub async fn list_pending( + &self, + account: &Account, + cursor: Option<&str>, + wait: u32, + ) -> Result { + let mut url = self.route("bot.did.listPendingAuthorizations")?; + { + let mut query = url.query_pairs_mut(); + query.append_pair("wait", &wait.to_string()); + if let Some(cursor) = cursor { + query.append_pair("cursor", cursor); + } + } + self.read(self.http.get(url), account).await + } + + /// One record, by the request it is about. + pub async fn get(&self, account: &Account, request_uri: &str) -> Result { + let mut url = self.route("bot.did.getAuthorization")?; + url.query_pairs_mut().append_pair("requestUri", request_uri); + self.read(self.http.get(url), account).await + } + + /// Redeem an approval token. + pub async fn approve(&self, account: &Account, token: &str) -> Result { + let url = self.route("bot.did.approveAuthorization")?; + let request = self + .http + .post(url) + .json(&serde_json::json!({ "token": token })); + self.read(request, account).await + } + + /// Record that the agent said no, so that its refusal is a decision + /// rather than an expiry nobody can tell from a timeout. + pub async fn decline( + &self, + account: &Account, + token: &str, + reason: Option<&str>, + ) -> Result { + let url = self.route("bot.did.declineAuthorization")?; + let mut body = serde_json::json!({ "token": token }); + if let Some(reason) = reason { + body["reason"] = serde_json::Value::String(reason.to_owned()); + } + self.read(self.http.post(url).json(&body), account).await + } + + /// Hand a code to the client waiting for it on this machine. + /// + /// The same bounded fetch [`crate::confirm`] makes, over the same client: + /// see [`crate::loopback`] for why the daemon is the one that makes it. + pub async fn deliver(&self, redirect: &str) -> Result<(), crate::loopback::Trouble> { + crate::loopback::deliver(&self.http, redirect).await + } + + /// The URL of one XRPC method on this server. + fn route(&self, method: &str) -> Result { + Url::parse(&format!("{}/xrpc/{method}", self.base)) + .map_err(|err| Trouble::Unreachable(format!("{} is not a URL: {err}", self.base))) + } + + /// Send one call as `account` and read what comes back. + async fn read( + &self, + request: reqwest::RequestBuilder, + account: &Account, + ) -> Result { + let response = request + .bearer_auth(account.token.reveal()) + .send() + .await + .map_err(|err| Trouble::Unreachable(err.to_string()))?; + + let status = response.status(); + let text = response + .text() + .await + .map_err(|err| Trouble::Unreachable(err.to_string()))?; + + if !status.is_success() { + // The server's own sentence, not a restatement of the status: + // "token unknown", "account mismatch" and "policy refused" are + // three different things behind one 400, and the agent that has + // to decide what to do next is the one reading this. + return Err(Trouble::Refused( + match serde_json::from_str::(&text) { + Ok(body) => format!("{}: {}", body.error, body.message), + Err(_) => format!("{status}: {}", text.trim()), + }, + )); + } + + serde_json::from_str(&text) + .map_err(|err| Trouble::Unexpected(format!("{err}: {}", text.trim()))) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn account() -> Account { + Account { + did: "did:web:one.example".into(), + token: Secret::new("agent-token"), + } + } + + fn record(verdict: Verdict, token: Option<&str>) -> Record { + Record { + request_uri: "urn:ietf:params:oauth:request_uri:r1".into(), + account: "did:web:one.example".into(), + client: Client { + origin: "http://127.0.0.1:40831".into(), + key: "abcd".into(), + first_time: true, + }, + requested: vec!["atproto".into(), "repo:com.example.thing".into()], + verdict, + token: token.map(Into::into), + created_at: None, + expires_at: "2026-09-09T12:04:00Z".into(), + state: pending(), + } + } + + #[test] + fn a_record_reads_the_shape_the_server_sends() { + let line = r#"{ + "requestUri": "urn:ietf:params:oauth:request_uri:r1", + "account": "did:web:one.example", + "client": { "origin": "http://127.0.0.1:40831", "key": "abcd", "firstTime": true }, + "requested": ["atproto"], + "verdict": { "kind": "narrow", "granted": ["atproto"], "cut": ["blob:*"], + "rule": "ceiling" }, + "token": "k7f3", + "createdAt": "2026-09-09T12:02:00Z", + "expiresAt": "2026-09-09T12:04:00Z", + "state": "pending" + }"#; + let record: Record = serde_json::from_str(line).unwrap(); + assert_eq!(record.granted(), vec!["atproto".to_owned()]); + assert!(!record.settled()); + assert!(record.client.first_time); + } + + #[test] + fn a_state_this_daemon_has_never_heard_of_is_still_a_record() { + let mut settled = record(Verdict::Allow, Some("k7f3")); + settled.state = "somethingNew".into(); + assert!(settled.settled(), "anything but pending is settled"); + } + + #[test] + fn a_page_with_nothing_in_it_is_an_ordinary_answer() { + let page: Page = serde_json::from_str(r#"{"cursor":"c1"}"#).unwrap(); + assert!(page.pending.is_empty()); + assert_eq!(page.cursor.as_deref(), Some("c1")); + } + + #[test] + fn what_an_agent_is_shown_carries_the_cut_and_the_rule() { + let narrowed = record( + Verdict::Narrow { + granted: vec!["atproto".into()], + cut: vec!["repo:com.example.thing".into()], + rule: "ceiling".into(), + }, + Some("k7f3"), + ); + let shown = DecisionForAgent::from(&narrowed); + assert_eq!(shown.verdict, "narrow"); + assert_eq!(shown.cut, vec!["repo:com.example.thing".to_owned()]); + assert_eq!(shown.rule.as_deref(), Some("ceiling")); + assert_eq!(shown.token.as_deref(), Some("k7f3")); + } + + #[tokio::test] + async fn a_poll_presents_the_account_and_reads_what_it_gets() { + let double = crate::double::start().await; + double.state.offer(crate::double::record( + "urn:ietf:params:oauth:request_uri:r1", + "did:web:one.example", + Some("k1"), + &crate::double::later(), + )); + let pds = Pds::new(&double.origin); + + let page = pds.list_pending(&account(), None, 1).await.unwrap(); + + assert_eq!(page.pending.len(), 1); + assert_eq!(page.cursor.as_deref(), Some("c1")); + // As the account, with the credential the daemon was handed at + // provisioning -- the same scheme every other agent-scoped route on + // that server takes. + assert_eq!(double.state.presented(), vec!["agent-token".to_owned()]); + } + + #[tokio::test] + async fn a_request_uri_survives_being_a_query_parameter() { + let double = crate::double::start().await; + // Colons and all: the pushed request identifier is a URN, and a + // client that pasted it into a query string unencoded would ask for + // a record nobody has. + let uri = "urn:ietf:params:oauth:request_uri:a/b+c"; + double.state.know(crate::double::record( + uri, + "did:web:one.example", + Some("k1"), + &crate::double::later(), + )); + let pds = Pds::new(&double.origin); + + let record = pds.get(&account(), uri).await.unwrap(); + assert_eq!(record.request_uri, uri); + } + + #[tokio::test] + async fn an_approval_answers_with_what_was_granted_and_where_to_take_it() { + let double = crate::double::start().await; + let mut record = crate::double::record( + "urn:ietf:params:oauth:request_uri:r1", + "did:web:one.example", + Some("k1"), + &crate::double::later(), + ); + record.client.origin = double.origin.clone(); + double.state.know(record); + let pds = Pds::new(&double.origin); + + let approved = pds.approve(&account(), "k1").await.unwrap(); + assert_eq!(approved.granted, vec!["atproto".to_owned()]); + let redirect = approved.redirect.expect("somewhere to take the code"); + + pds.deliver(&redirect).await.unwrap(); + assert_eq!(double.state.delivered(), vec!["abc".to_owned()]); + } + + #[tokio::test] + async fn a_refusal_arrives_as_the_server_wrote_it() { + let double = crate::double::start().await; + double + .state + .refuse(400, "InvalidToken", "that token has been used"); + let pds = Pds::new(&double.origin); + + let err = pds.approve(&account(), "k1").await.unwrap_err(); + + // The server's own name and sentence: "used", "expired" and "policy + // refused" are three different things behind one 400, and the agent + // reading this is the one deciding what to do next. + assert!(matches!(err, Trouble::Refused(_)), "{err:?}"); + let said = err.to_string(); + assert!(said.contains("InvalidToken"), "{said}"); + assert!(said.contains("that token has been used"), "{said}"); + } + + #[tokio::test] + async fn a_refusal_is_relayed_and_a_decline_is_recorded() { + let double = crate::double::start().await; + double.state.know(crate::double::record( + "urn:ietf:params:oauth:request_uri:r1", + "did:web:one.example", + Some("k1"), + &crate::double::later(), + )); + let pds = Pds::new(&double.origin); + + let declined = pds + .decline(&account(), "k1", Some("not something I asked for")) + .await + .unwrap(); + assert_eq!(declined.request_uri, "urn:ietf:params:oauth:request_uri:r1"); + assert_eq!( + double.state.declined(), + vec![( + "k1".to_owned(), + Some("not something I asked for".to_owned()) + )] + ); + } + + #[tokio::test] + async fn a_server_that_is_not_there_is_unreachable_rather_than_a_refusal() { + // Bound and dropped, so the port is one nothing is listening on. + let closed = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let origin = format!("http://{}", closed.local_addr().unwrap()); + drop(closed); + + let pds = Pds::new(origin); + let err = pds.list_pending(&account(), None, 1).await.unwrap_err(); + assert!(matches!(err, Trouble::Unreachable(_)), "{err:?}"); + } + + #[test] + fn a_refusal_reaches_the_agent_with_nothing_to_approve() { + let denied = record( + Verdict::Deny { + reason: "that client is not admitted".into(), + rule: "app-allowlist".into(), + }, + None, + ); + let shown = DecisionForAgent::from(&denied); + assert_eq!(shown.verdict, "deny"); + assert!(shown.granted.is_empty()); + // No token, because there is no choice: the agent is being told what + // happened, not asked. + assert!(shown.token.is_none()); + assert_eq!( + shown.rule.as_deref(), + Some("app-allowlist: that client is not admitted") + ); + } +} diff --git a/crates/didbot-agentd/src/double.rs b/crates/didbot-agentd/src/double.rs new file mode 100644 index 00000000..732129df --- /dev/null +++ b/crates/didbot-agentd/src/double.rs @@ -0,0 +1,289 @@ +//! A stand-in for the server's four decision routes, and for a client's +//! callback. +//! +//! Real HTTP on a real loopback port rather than a trait the client was +//! written to satisfy. The point of these tests is the wire — the bearer +//! scheme, the query encoding of a `request_uri`, an error body relayed +//! rather than flattened into a status — and none of that is exercised by a +//! double the client cannot get wrong. +//! +//! Test-only, and never compiled into the daemon. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; + +use axum::extract::{Query, State}; +use axum::http::{HeaderMap, StatusCode}; +use axum::routing::{get, post}; +use axum::{Json, Router}; +use serde_json::{json, Value}; + +use crate::decisions::Record; + +/// What the double was asked, and what it should answer. +#[derive(Default)] +pub struct Shared { + /// Records not yet handed out by a poll. + queue: Mutex>, + /// Every record it knows, for `getAuthorization`. + known: Mutex>, + /// The credentials presented, in order. + presented: Mutex>, + /// What `approveAuthorization` should refuse with, if anything. + refusal: Mutex>, + /// The callbacks a client was sent to. + delivered: Mutex>, + /// How many polls have been answered. + polls: AtomicUsize, + /// What was approved and declined. + approved: Mutex>, + declined: Mutex)>>, +} + +impl Shared { + /// Put a record where the next poll will find it. + pub fn offer(&self, record: Record) { + self.known + .lock() + .unwrap() + .insert(record.request_uri.clone(), record.clone()); + self.queue.lock().unwrap().push(record); + } + + /// Let the double know a record without offering it to a poll, so the + /// only way to it is `getAuthorization`. + pub fn know(&self, record: Record) { + self.known + .lock() + .unwrap() + .insert(record.request_uri.clone(), record); + } + + /// Make the next approval fail with this error body. + pub fn refuse(&self, status: u16, error: &str, message: &str) { + *self.refusal.lock().unwrap() = Some((status, error.to_owned(), message.to_owned())); + } + + /// The credentials presented, in order. + pub fn presented(&self) -> Vec { + self.presented.lock().unwrap().clone() + } + + /// The callbacks a client was sent to. + pub fn delivered(&self) -> Vec { + self.delivered.lock().unwrap().clone() + } + + /// The tokens approved. + pub fn approved(&self) -> Vec { + self.approved.lock().unwrap().clone() + } + + /// The tokens declined, and why. + pub fn declined(&self) -> Vec<(String, Option)> { + self.declined.lock().unwrap().clone() + } + + /// How many polls have been answered. + pub fn polls(&self) -> usize { + self.polls.load(Ordering::SeqCst) + } +} + +/// A running double, stopped when it is dropped. +pub struct Double { + /// Where it is listening, such as `http://127.0.0.1:41000`. + pub origin: String, + /// What it was asked and what it will answer. + pub state: Arc, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for Double { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// Start one on a port the operating system picks. +pub async fn start() -> Double { + let state = Arc::new(Shared::default()); + let router = Router::new() + .route("/xrpc/bot.did.listPendingAuthorizations", get(list)) + .route("/xrpc/bot.did.getAuthorization", get(one)) + .route("/xrpc/bot.did.approveAuthorization", post(approve)) + .route("/xrpc/bot.did.declineAuthorization", post(decline)) + // The client's own loopback callback, which is what a delivered code + // actually reaches. + .route("/cb", get(callback)) + .with_state(Arc::clone(&state)); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let origin = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { + let _ = axum::serve(listener, router).await; + }); + Double { + origin, + state, + task, + } +} + +/// The refusal every route gives a caller presenting nothing. +fn unauthenticated() -> (StatusCode, Json) { + ( + StatusCode::UNAUTHORIZED, + Json(json!({ + "error": "AuthenticationRequired", + "message": "this route is for the account itself", + })), + ) +} + +/// Records what the caller presented, and refuses a caller presenting +/// nothing — which is the check the real routes make. +fn presented(state: &Shared, headers: &HeaderMap) -> Option { + let token = headers + .get("authorization") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + .map(str::to_owned)?; + state.presented.lock().unwrap().push(token.clone()); + Some(token) +} + +async fn list( + State(state): State>, + Query(query): Query>, + headers: HeaderMap, +) -> Result, (StatusCode, Json)> { + presented(&state, &headers).ok_or_else(unauthenticated)?; + state.polls.fetch_add(1, Ordering::SeqCst); + + let waiting: Vec = std::mem::take(&mut *state.queue.lock().unwrap()); + if waiting.is_empty() { + // A real long poll holds the request open. Holding it for the whole + // `wait` would make every test that starts a poller wait too, so this + // holds it for long enough that a polling loop does not spin. + if query.contains_key("wait") { + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + } + Ok(Json(json!({ "cursor": "c1", "pending": waiting }))) +} + +async fn one( + State(state): State>, + Query(query): Query>, + headers: HeaderMap, +) -> Result, (StatusCode, Json)> { + presented(&state, &headers).ok_or_else(unauthenticated)?; + let wanted = query.get("requestUri").cloned().unwrap_or_default(); + match state.known.lock().unwrap().get(&wanted) { + Some(record) => Ok(Json(serde_json::to_value(record).unwrap())), + None => Err(( + StatusCode::BAD_REQUEST, + Json(json!({ "error": "NotFound", "message": "no such request" })), + )), + } +} + +async fn approve( + State(state): State>, + headers: HeaderMap, + Json(body): Json, +) -> Result, (StatusCode, Json)> { + presented(&state, &headers).ok_or_else(unauthenticated)?; + if let Some((status, error, message)) = state.refusal.lock().unwrap().take() { + return Err(( + StatusCode::from_u16(status).unwrap(), + Json(json!({ "error": error, "message": message })), + )); + } + let token = body["token"].as_str().unwrap_or_default().to_owned(); + let found = state + .known + .lock() + .unwrap() + .values() + .find(|record| record.token.as_deref() == Some(&token)) + .cloned(); + let Some(record) = found else { + return Err(( + StatusCode::BAD_REQUEST, + Json(json!({ "error": "InvalidToken", "message": "no such approval token" })), + )); + }; + state.approved.lock().unwrap().push(token); + + Ok(Json(json!({ + "requestUri": record.request_uri, + "granted": record.granted(), + "redirect": record.client.origin.clone() + "/cb?code=abc&state=xyz", + }))) +} + +async fn decline( + State(state): State>, + headers: HeaderMap, + Json(body): Json, +) -> Result, (StatusCode, Json)> { + presented(&state, &headers).ok_or_else(unauthenticated)?; + let token = body["token"].as_str().unwrap_or_default().to_owned(); + let reason = body["reason"].as_str().map(str::to_owned); + let found = state + .known + .lock() + .unwrap() + .values() + .find(|record| record.token.as_deref() == Some(&token)) + .cloned(); + let Some(record) = found else { + return Err(( + StatusCode::BAD_REQUEST, + Json(json!({ "error": "InvalidToken", "message": "no such approval token" })), + )); + }; + state.declined.lock().unwrap().push((token, reason)); + Ok(Json(json!({ "requestUri": record.request_uri }))) +} + +async fn callback(State(state): State>, Query(query): Query>) { + let code = query.get("code").cloned().unwrap_or_default(); + state.delivered.lock().unwrap().push(code); +} + +/// A record shaped the way the server sends them, for a test to adjust. +pub fn record(request_uri: &str, account: &str, token: Option<&str>, expires_at: &str) -> Record { + Record { + request_uri: request_uri.to_owned(), + account: account.to_owned(), + client: crate::decisions::Client { + origin: String::new(), + key: "abcd".to_owned(), + first_time: true, + }, + requested: vec!["atproto".to_owned()], + verdict: crate::decisions::Verdict::Allow, + token: token.map(str::to_owned), + created_at: None, + expires_at: expires_at.to_owned(), + state: "pending".to_owned(), + } +} + +/// An expiry far enough ahead that no test outlives it. +pub fn later() -> String { + (time::OffsetDateTime::now_utc() + time::Duration::hours(1)) + .format(&time::format_description::well_known::Rfc3339) + .unwrap() +} + +/// And one that has already passed. +pub fn earlier() -> String { + (time::OffsetDateTime::now_utc() - time::Duration::hours(1)) + .format(&time::format_description::well_known::Rfc3339) + .unwrap() +} diff --git a/crates/didbot-agentd/src/lib.rs b/crates/didbot-agentd/src/lib.rs index 9125ab7f..4d01890e 100644 --- a/crates/didbot-agentd/src/lib.rs +++ b/crates/didbot-agentd/src/lib.rs @@ -16,15 +16,23 @@ //! in another repository and is not written in Rust, so the wire format is //! the contract rather than these types. //! -//! [`secret`] is what does not cross it. `plan/cred-delivery.md` says the -//! model holds nothing, so this process holds the account credentials -//! instead, and that module is how they stay held. +//! [`decisions`] is the other direction: the sign-in requests an agent is +//! asked about. Those come from the server, so this daemon holds the account +//! credential ([`secret`]) it needs to fetch them, polls for them +//! ([`poll`]), keeps them ([`pending`]) and relays the agent's answer. It +//! decides none of it. #![forbid(unsafe_code)] +#[cfg(test)] +mod double; + pub mod confirm; pub mod context; +pub mod decisions; pub mod loopback; +pub mod pending; +pub mod poll; pub mod protocol; pub mod registrar; pub mod secret; diff --git a/crates/didbot-agentd/src/pending.rs b/crates/didbot-agentd/src/pending.rs new file mode 100644 index 00000000..c99a0d16 --- /dev/null +++ b/crates/didbot-agentd/src/pending.rs @@ -0,0 +1,268 @@ +//! The decisions this daemon is holding, and who each one belongs to. +//! +//! Memory only, and deliberately: a record lives as long as its pushed +//! request — two minutes at the server — and a daemon that restarted has +//! nothing useful to say about one that was already in flight. Nothing here +//! reaches the network; [`crate::poll`] does that and puts what it finds +//! here. +//! +//! Two lookups matter. A `report` answers one context, so records are found +//! by the context that owns the account. An `approve` arrives with nothing +//! but a token, so records are also found by that — and that lookup is what +//! decides which account the daemon approves as. The caller never says. + +use std::collections::HashMap; + +use time::OffsetDateTime; + +use crate::context::Key; +use crate::decisions::Record; + +/// One record and the context whose account it names. +#[derive(Debug, Clone)] +pub struct Holding { + /// The context that was issued the account this record is about. + pub key: Key, + /// The record itself. + pub record: Record, +} + +/// Every decision this daemon is holding, by the request it is about. +#[derive(Debug, Default)] +pub struct Held { + by_request: HashMap, +} + +impl Held { + /// A store holding nothing. + pub fn new() -> Self { + Self::default() + } + + /// Hold a record for the context that owns its account. + /// + /// A record that has already been settled or has already expired is not + /// held: there is no choice left in it, and putting one in front of an + /// agent would be asking for a decision that cannot be made. + pub fn keep(&mut self, key: Key, record: Record, now: OffsetDateTime) -> bool { + if record.settled() || expired(&record, now) { + self.by_request.remove(&record.request_uri); + return false; + } + self.by_request + .insert(record.request_uri.clone(), Holding { key, record }); + true + } + + /// Stop holding one record, whatever became of it. + pub fn forget(&mut self, request_uri: &str) -> Option { + self.by_request.remove(request_uri) + } + + /// Drop everything whose moment has passed, and say how many went. + pub fn sweep(&mut self, now: OffsetDateTime) -> usize { + let before = self.by_request.len(); + self.by_request + .retain(|_, holding| !holding.record.settled() && !expired(&holding.record, now)); + before - self.by_request.len() + } + + /// What one context is holding, oldest request first. + /// + /// A context is answered with its own and nothing else: two agents on + /// this machine are two accounts, and one has no business being offered + /// the other's sign-in. + pub fn for_context(&self, key: &Key) -> Vec { + let mut found: Vec<_> = self + .by_request + .values() + .filter(|holding| &holding.key == key) + .map(|holding| holding.record.clone()) + .collect(); + found.sort_by(|a, b| a.request_uri.cmp(&b.request_uri)); + found + } + + /// Everything held, oldest request first. + pub fn all(&self) -> Vec { + let mut found: Vec<_> = self + .by_request + .values() + .map(|holding| holding.record.clone()) + .collect(); + found.sort_by(|a, b| a.request_uri.cmp(&b.request_uri)); + found + } + + /// The record an approval token belongs to. + /// + /// This is the whole of the daemon's answer to "who is approving": the + /// account is the one named by the record the token is in, and a token + /// this daemon is not holding names nobody. + pub fn by_token(&self, token: &str) -> Option<&Holding> { + self.by_request + .values() + .find(|holding| holding.record.token.as_deref() == Some(token)) + } + + /// The record for one pushed request. + pub fn by_request_uri(&self, request_uri: &str) -> Option<&Holding> { + self.by_request.get(request_uri) + } + + /// How many records are held. + pub fn len(&self) -> usize { + self.by_request.len() + } + + /// Whether nothing is held. + pub fn is_empty(&self) -> bool { + self.by_request.is_empty() + } +} + +/// Whether a record's moment has passed. +/// +/// A record whose `expiresAt` cannot be read is treated as live. The server +/// is the authority on expiry and will refuse a stale token; dropping a +/// record here because this daemon could not parse a timestamp would take a +/// decision away from an agent for a reason that has nothing to do with it. +fn expired(record: &Record, now: OffsetDateTime) -> bool { + match OffsetDateTime::parse( + &record.expires_at, + &time::format_description::well_known::Rfc3339, + ) { + Ok(expires) => expires <= now, + Err(_) => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::decisions::{Client, Verdict}; + + fn at(rfc3339: &str) -> OffsetDateTime { + OffsetDateTime::parse(rfc3339, &time::format_description::well_known::Rfc3339).unwrap() + } + + fn key(context: &str) -> Key { + Key { + session: "s1".into(), + context: Some(context.into()), + } + } + + fn record(request_uri: &str, expires_at: &str, token: Option<&str>) -> Record { + Record { + request_uri: request_uri.into(), + account: "did:web:one.example".into(), + client: Client { + origin: "http://127.0.0.1:40831".into(), + key: "abcd".into(), + first_time: true, + }, + requested: vec!["atproto".into()], + verdict: Verdict::Allow, + token: token.map(Into::into), + created_at: None, + expires_at: expires_at.into(), + state: "pending".into(), + } + } + + #[test] + fn a_context_is_offered_its_own_decisions_and_no_others() { + let mut held = Held::new(); + let now = at("2026-09-09T12:00:00Z"); + held.keep( + key("a1"), + record("r1", "2026-09-09T12:04:00Z", Some("k1")), + now, + ); + held.keep( + key("a2"), + record("r2", "2026-09-09T12:04:00Z", Some("k2")), + now, + ); + + let mine = held.for_context(&key("a1")); + assert_eq!(mine.len(), 1); + assert_eq!(mine[0].request_uri, "r1"); + assert_eq!(held.all().len(), 2); + } + + #[test] + fn a_token_names_the_account_that_will_approve() { + let mut held = Held::new(); + let now = at("2026-09-09T12:00:00Z"); + held.keep( + key("a1"), + record("r1", "2026-09-09T12:04:00Z", Some("k1")), + now, + ); + + let holding = held.by_token("k1").expect("held"); + assert_eq!(holding.key, key("a1")); + assert_eq!(holding.record.account, "did:web:one.example"); + assert!(held.by_token("never-issued").is_none()); + } + + #[test] + fn a_record_whose_moment_has_passed_is_dropped() { + let mut held = Held::new(); + let now = at("2026-09-09T12:00:00Z"); + held.keep( + key("a1"), + record("r1", "2026-09-09T12:04:00Z", Some("k1")), + now, + ); + held.keep( + key("a1"), + record("r2", "2026-09-09T12:10:00Z", Some("k2")), + now, + ); + + assert_eq!(held.sweep(at("2026-09-09T12:05:00Z")), 1); + assert_eq!(held.len(), 1); + assert!(held.by_token("k1").is_none()); + } + + #[test] + fn one_that_has_already_expired_is_never_taken_in() { + let mut held = Held::new(); + let now = at("2026-09-09T12:00:00Z"); + assert!(!held.keep( + key("a1"), + record("r1", "2026-09-09T11:59:00Z", Some("k1")), + now + )); + assert!(held.is_empty()); + } + + #[test] + fn one_the_server_has_settled_replaces_nothing_and_leaves_nothing() { + let mut held = Held::new(); + let now = at("2026-09-09T12:00:00Z"); + held.keep( + key("a1"), + record("r1", "2026-09-09T12:04:00Z", Some("k1")), + now, + ); + + // The same request coming back approved is the server saying the + // choice was made, so the daemon stops offering it. + let mut approved = record("r1", "2026-09-09T12:04:00Z", Some("k1")); + approved.state = "approved".into(); + assert!(!held.keep(key("a1"), approved, now)); + assert!(held.is_empty()); + } + + #[test] + fn an_expiry_this_daemon_cannot_read_is_left_to_the_server() { + let mut held = Held::new(); + let now = at("2026-09-09T12:00:00Z"); + assert!(held.keep(key("a1"), record("r1", "whenever", Some("k1")), now)); + assert_eq!(held.sweep(at("2030-01-01T00:00:00Z")), 0); + } +} diff --git a/crates/didbot-agentd/src/poll.rs b/crates/didbot-agentd/src/poll.rs new file mode 100644 index 00000000..cc032d22 --- /dev/null +++ b/crates/didbot-agentd/src/poll.rs @@ -0,0 +1,90 @@ +//! Asking the server what is waiting, over and over, outbound only. +//! +//! `plan/node.md` puts this daemon on the agent host and nothing on the agent +//! host is reachable from the server, so the direction is fixed: the daemon +//! calls out and holds the call open, rather than being told. One task per +//! account, each long-polling `listPendingAuthorizations` with the cursor its +//! last answer gave it. +//! +//! Nothing here pushes anything at anyone. The hook still starts every +//! exchange over the socket, and what this task has collected rides the next +//! answer. An agent therefore learns of a sign-in at its next tool call, +//! which is the latency this design accepts rather than the one it hides. + +use std::sync::Arc; +use std::time::Duration; + +use time::OffsetDateTime; +use tokio::sync::Mutex; +use tracing::{debug, warn}; + +use crate::context::Key; +use crate::decisions::{Account, Pds, WAIT}; +use crate::pending::Held; + +/// How long to wait after the first failure. +const FIRST_REST: Duration = Duration::from_secs(1); + +/// And the longest this ever waits between attempts. +/// +/// A server that is down comes back, and a daemon that has stopped asking +/// leaves an agent unable to sign in with nothing on either side saying why. +/// `plan/oauth.md`'s "prefer recovering from a gap over never having one" is +/// the same argument the index makes about being offline. +const LONGEST_REST: Duration = Duration::from_secs(30); + +/// Long-poll for one account until this task is dropped. +pub async fn run(pds: Arc, held: Arc>, key: Key, account: Account) { + let mut cursor: Option = None; + let mut rest = FIRST_REST; + + loop { + match pds.list_pending(&account, cursor.as_deref(), WAIT).await { + Ok(page) => { + rest = FIRST_REST; + if page.cursor.is_some() { + cursor = page.cursor; + } + let found = page.pending.len(); + let now = OffsetDateTime::now_utc(); + let mut held = held.lock().await; + let mut kept = 0; + for record in page.pending { + if held.keep(key.clone(), record, now) { + kept += 1; + } + } + let swept = held.sweep(now); + if found > 0 || swept > 0 { + debug!( + did = %account.did, found, kept, swept, holding = held.len(), + "polled" + ); + } + } + Err(err) => { + warn!(did = %account.did, error = %err, rest = ?rest, "could not poll"); + tokio::time::sleep(rest).await; + rest = (rest * 2).min(LONGEST_REST); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resting_doubles_up_to_a_ceiling_and_stays_there() { + let mut rest = FIRST_REST; + let mut seen = vec![rest]; + for _ in 0..8 { + rest = (rest * 2).min(LONGEST_REST); + seen.push(rest); + } + assert_eq!(seen[0], Duration::from_secs(1)); + assert_eq!(seen[5], Duration::from_secs(30)); + assert_eq!(*seen.last().unwrap(), LONGEST_REST); + } +} diff --git a/crates/didbot-agentd/src/protocol.rs b/crates/didbot-agentd/src/protocol.rs index 5ddc8be4..86c71de1 100644 --- a/crates/didbot-agentd/src/protocol.rs +++ b/crates/didbot-agentd/src/protocol.rs @@ -7,7 +7,9 @@ //! importing this crate is a change that has put the two back together. //! //! One line of JSON per message, request and reply alike. A connection -//! carries one exchange and closes; nothing here is a session. +//! carries one exchange and closes; nothing here is a session — including the +//! sign-in decisions [`Answer::pending`] carries, which ride an exchange the +//! adapter started rather than arriving on one the daemon opened. use serde::{Deserialize, Serialize}; @@ -17,7 +19,14 @@ use serde::{Deserialize, Serialize}; /// updates when a plugin does and the daemon when a package does — so skew is /// a handled case rather than a surprise. A daemon that does not know a /// version answers rather than closing the connection. -pub const VERSION: u32 = 1; +/// +/// Version 2 added the sign-in decisions: [`Answer::pending`], +/// [`Report::seen_request_uris`], and the [`Message::Approve`], +/// [`Message::Decline`] and [`Message::Pending`] asks. Everything version 1 +/// sent is still read and answered — see `serve::Daemon::consider`, which +/// refuses a version newer than its own and nothing else — and an adapter +/// still on version 1 finds the fields it knows where they were. +pub const VERSION: u32 = 2; /// What each caller wants. /// @@ -31,7 +40,16 @@ pub enum Message { /// A hook saying what the harness just did. Report(Report), /// A tool asking for an authorization to be confirmed on its behalf. + /// + /// Superseded by [`Message::Approve`]; see [`Confirm`]. Confirm(Confirm), + /// An agent saying yes to a sign-in the daemon offered it. + Approve(Approve), + /// An agent saying no to one, so that the refusal is recorded rather + /// than left to look like a timeout. + Decline(Decline), + /// A caller asking what the daemon is holding. + Pending(Pending), } impl<'de> Deserialize<'de> for Message { @@ -45,6 +63,15 @@ impl<'de> Deserialize<'de> for Message { Some("confirm") => serde_json::from_value(value) .map(Message::Confirm) .map_err(D::Error::custom), + Some("approve") => serde_json::from_value(value) + .map(Message::Approve) + .map_err(D::Error::custom), + Some("decline") => serde_json::from_value(value) + .map(Message::Decline) + .map_err(D::Error::custom), + Some("pending") => serde_json::from_value(value) + .map(Message::Pending) + .map_err(D::Error::custom), Some(other) => Err(D::Error::custom(format!( "this daemon does not know how to `{other}`" ))), @@ -103,12 +130,30 @@ pub struct Report { /// The harness's identifier for this particular tool call. #[serde(default, skip_serializing_if = "Option::is_none")] pub call: Option, + /// Pushed requests the adapter saw go past in tool output. + /// + /// The fast path, and only that. [`Answer::pending`] is filled by what + /// the daemon has already been told; this is for the case where a client + /// printed its authorize URL in the very tool call being reported, which + /// is sooner than a poll can have finished. The daemon fetches any of + /// these it is not already holding. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub seen_request_uris: Vec, } /// A tool asking the daemon to confirm an authorization it started. /// /// The URL is the client's own authorize page, printed rather than opened. /// +/// # Deprecated, and kept for one release +/// +/// [`Message::Approve`] replaces this. An approval names a decision the +/// daemon is already holding, so the account is the one that decision names +/// and the caller supplies nothing but a one-time token — which is the whole +/// of the problem described below, gone. A daemon that is holding the record +/// this URL's `request_uri` names approves it that way; one that is not falls +/// back to reading the authorize page, exactly as it always did. +/// /// # The caller names its own account /// /// `did` is taken from the caller and nothing checks that the caller is it. @@ -133,6 +178,84 @@ pub struct Confirm { pub url: String, } +/// An agent approving a sign-in the daemon put in front of it. +/// +/// One field, and it is not an account. The token was minted by the server +/// against one pushed request for one account, and the daemon knows which +/// because it is holding the record the token came in. So there is nothing +/// here for a caller to name wrongly: a token this daemon is not holding is +/// refused, and one it is holding is approved as the account that record +/// says, never as an account the caller asked for. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Approve { + /// The wire version this message was written against. + pub version: u32, + /// The one-time approval token from the decision. + pub token: String, +} + +/// An agent refusing one. +/// +/// Worth sending rather than letting the request expire: an expiry and a +/// refusal look the same from the server, and only one of them is a decision +/// somebody made. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Decline { + /// The wire version this message was written against. + pub version: u32, + /// The one-time approval token from the decision. + pub token: String, + /// Why, in the agent's own words, for the record. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reason: Option, +} + +/// A caller asking what the daemon is holding. +/// +/// Answered with everything, not with one context's share. A `report` names +/// the context it is about and is answered with that context's decisions +/// only; this arrives from a command line, which names nothing the daemon +/// can check — and the ceiling on that is the user account, as +/// [`crate::socket`] already sets out. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Pending { + /// The wire version this message was written against. + pub version: u32, +} + +/// One sign-in decision, as an agent is shown it. +/// +/// The rendered subset of what the server said, and not the whole record: +/// the client's own description of itself never leaves the server, and the +/// account is not here because a decision is only ever shown to the context +/// that holds it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DecisionForAgent { + /// What to approve or decline with, absent when the verdict is `deny` + /// and there is therefore nothing to choose. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub token: Option, + /// Where the request came from. + pub client_origin: String, + /// Whether this account has seen that client before. + pub first_time: bool, + /// What was asked for. + pub requested: Vec, + /// What would actually be granted. + pub granted: Vec, + /// What was taken out of the request, empty unless it was narrowed. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub cut: Vec, + /// The rule that narrowed or refused it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub rule: Option, + /// `allow`, `narrow` or `deny`. + pub verdict: String, + /// When the request stops being answerable, RFC 3339. + pub expires_at: String, +} + /// What the daemon says back. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Answer { @@ -148,6 +271,13 @@ pub struct Answer { /// Why there is nothing to report, when there should have been. #[serde(default, skip_serializing_if = "Option::is_none")] pub trouble: Option, + /// Sign-ins waiting on a decision, for the context that asked. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pending: Option>, + /// What an approval actually granted, which is not always what was asked + /// for: a narrowed request grants less, and says so here. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub granted: Option>, } impl Answer { @@ -158,37 +288,53 @@ impl Answer { identity: None, done: None, trouble: None, + pending: None, + granted: None, } } /// A context's identity, to be told to it once. pub fn identity(did: impl Into) -> Self { Self { - version: VERSION, identity: Some(did.into()), - done: None, - trouble: None, + ..Self::quiet() } } /// Something the daemon carried out. pub fn done(what: impl Into) -> Self { Self { - version: VERSION, - identity: None, done: Some(what.into()), - trouble: None, + ..Self::quiet() } } /// Something a caller should surface rather than swallow. pub fn trouble(why: impl Into) -> Self { Self { - version: VERSION, - identity: None, - done: None, trouble: Some(why.into()), + ..Self::quiet() + } + } + + /// Carry sign-in decisions back with whatever else this answer says. + /// + /// An empty list is left off entirely rather than sent as `[]`: an + /// adapter reading this in another language should be able to test for + /// the field's presence and not have to distinguish two kinds of nothing. + #[must_use] + pub fn and_pending(mut self, pending: Vec) -> Self { + if !pending.is_empty() { + self.pending = Some(pending); } + self + } + + /// Carry what an approval granted. + #[must_use] + pub fn and_granted(mut self, granted: Vec) -> Self { + self.granted = Some(granted); + self } } @@ -198,7 +344,7 @@ mod tests { #[test] fn a_report_carrying_only_what_it_must_round_trips() { - let line = r#"{"asks":"report","version":1,"observed":"began","session":"s1"}"#; + let line = r#"{"asks":"report","version":2,"observed":"began","session":"s1"}"#; let message: Message = serde_json::from_str(line).unwrap(); let Message::Report(report) = message else { panic!("a report"); @@ -215,10 +361,10 @@ mod tests { } #[test] - fn a_quiet_answer_is_two_fields_wide() { + fn a_quiet_answer_is_one_field_wide() { assert_eq!( serde_json::to_string(&Answer::quiet()).unwrap(), - r#"{"version":1}"# + r#"{"version":2}"# ); } @@ -235,7 +381,7 @@ mod tests { #[test] fn something_this_daemon_cannot_do_is_named_rather_than_guessed() { - let line = r#"{"asks":"revoke","version":1}"#; + let line = r#"{"asks":"revoke","version":2}"#; let err = serde_json::from_str::(line) .unwrap_err() .to_string(); @@ -244,17 +390,150 @@ mod tests { #[test] fn an_unknown_observation_is_refused_rather_than_guessed() { - let line = r#"{"asks":"report","version":1,"observed":"vanished","session":"s1"}"#; + let line = r#"{"asks":"report","version":2,"observed":"vanished","session":"s1"}"#; assert!(serde_json::from_str::(line).is_err()); } #[test] fn a_confirmation_carries_the_account_the_caller_claims() { - let line = r#"{"asks":"confirm","version":1,"did":"did:web:a","url":"http://x/authorize"}"#; + let line = r#"{"asks":"confirm","version":2,"did":"did:web:a","url":"http://x/authorize"}"#; let message: Message = serde_json::from_str(line).unwrap(); let Message::Confirm(confirm) = message else { panic!("a confirmation"); }; assert_eq!(confirm.did, "did:web:a"); } + + #[test] + fn an_approval_names_a_token_and_nothing_else() { + let line = r#"{"asks":"approve","version":2,"token":"k7f3"}"#; + let message: Message = serde_json::from_str(line).unwrap(); + let Message::Approve(approve) = message else { + panic!("an approval"); + }; + assert_eq!(approve.token, "k7f3"); + assert_eq!( + serde_json::to_string(&Message::Approve(approve)).unwrap(), + line + ); + } + + #[test] + fn a_refusal_may_say_why_and_need_not() { + let bare = r#"{"asks":"decline","version":2,"token":"k7f3"}"#; + let Message::Decline(decline) = serde_json::from_str::(bare).unwrap() else { + panic!("a refusal"); + }; + assert!(decline.reason.is_none()); + assert_eq!( + serde_json::to_string(&Message::Decline(decline)).unwrap(), + bare + ); + + let spoken = r#"{"asks":"decline","version":2,"token":"k7f3","reason":"not mine"}"#; + let Message::Decline(decline) = serde_json::from_str::(spoken).unwrap() else { + panic!("a refusal"); + }; + assert_eq!(decline.reason.as_deref(), Some("not mine")); + assert_eq!( + serde_json::to_string(&Message::Decline(decline)).unwrap(), + spoken + ); + } + + #[test] + fn asking_what_is_held_carries_only_a_version() { + let line = r#"{"asks":"pending","version":2}"#; + let message: Message = serde_json::from_str(line).unwrap(); + let Message::Pending(pending) = message else { + panic!("a question"); + }; + assert_eq!( + serde_json::to_string(&Message::Pending(pending)).unwrap(), + line + ); + } + + #[test] + fn a_report_may_carry_what_the_adapter_saw_go_past() { + let line = r#"{"asks":"report","version":2,"observed":"acted","session":"s1","seen_request_uris":["urn:ietf:params:oauth:request_uri:r1"]}"#; + let Message::Report(report) = serde_json::from_str::(line).unwrap() else { + panic!("a report"); + }; + assert_eq!(report.seen_request_uris.len(), 1); + assert_eq!( + serde_json::to_string(&Message::Report(report)).unwrap(), + line + ); + } + + #[test] + fn an_answer_carrying_a_decision_is_the_shape_the_adapter_renders() { + let answer = Answer::quiet().and_pending(vec![DecisionForAgent { + token: Some("k7f3".into()), + client_origin: "http://127.0.0.1:40831".into(), + first_time: true, + requested: vec!["atproto".into(), "repo:com.example.thing".into()], + granted: vec!["atproto".into()], + cut: vec!["repo:com.example.thing".into()], + rule: Some("ceiling".into()), + verdict: "narrow".into(), + expires_at: "2026-09-09T12:04:00Z".into(), + }]); + assert_eq!( + serde_json::to_string(&answer).unwrap(), + r#"{"version":2,"pending":[{"token":"k7f3","clientOrigin":"http://127.0.0.1:40831","firstTime":true,"requested":["atproto","repo:com.example.thing"],"granted":["atproto"],"cut":["repo:com.example.thing"],"rule":"ceiling","verdict":"narrow","expiresAt":"2026-09-09T12:04:00Z"}]}"# + ); + } + + #[test] + fn an_empty_list_of_decisions_is_left_off_rather_than_sent() { + assert_eq!( + serde_json::to_string(&Answer::quiet().and_pending(Vec::new())).unwrap(), + r#"{"version":2}"# + ); + } + + #[test] + fn an_adapter_a_version_behind_is_still_understood() { + // What a version 1 adapter sends. Every field of it is still read, + // and the daemon's own version check accepts anything up to its own + // -- see `serve::Daemon::consider`. + let line = + r#"{"asks":"report","version":1,"observed":"acted","session":"s1","context":"a1"}"#; + let Message::Report(report) = serde_json::from_str::(line).unwrap() else { + panic!("a report"); + }; + assert_eq!(report.version, 1); + assert!(report.seen_request_uris.is_empty()); + } + + #[test] + fn and_reads_the_answer_it_gets_back() { + // The same adapter reading a version 2 answer. `pending` and + // `granted` are fields it has never heard of, and the fields it does + // know are where they were. + #[derive(Deserialize)] + struct AsVersionOne { + version: u32, + identity: Option, + } + + let answer = Answer::identity("did:web:one.example").and_pending(vec![DecisionForAgent { + token: Some("k7f3".into()), + client_origin: "http://127.0.0.1:40831".into(), + first_time: false, + requested: vec!["atproto".into()], + granted: vec!["atproto".into()], + cut: Vec::new(), + rule: None, + verdict: "allow".into(), + expires_at: "2026-09-09T12:04:00Z".into(), + }]); + let line = serde_json::to_string(&answer).unwrap(); + + let old: AsVersionOne = serde_json::from_str(&line).unwrap(); + assert_eq!(old.version, 2); + assert_eq!(old.identity.as_deref(), Some("did:web:one.example")); + } } diff --git a/crates/didbot-agentd/src/serve.rs b/crates/didbot-agentd/src/serve.rs index 8c8419eb..dfcbf065 100644 --- a/crates/didbot-agentd/src/serve.rs +++ b/crates/didbot-agentd/src/serve.rs @@ -3,16 +3,29 @@ //! A connection carries one exchange because the adapter is a program the //! harness spawns per event. Several run at once, so each is handled in its //! own task and the store is shared. +//! +//! Sign-in decisions ride these exchanges rather than getting a channel of +//! their own: [`crate::poll`] collects them in the background, and whatever +//! is waiting for the reporting context goes out on the next answer. So this +//! stays what it was -- one line in, one line out, no session -- and the +//! agent hears about a sign-in on its next tool call. +use std::collections::HashMap; use std::sync::Arc; +use time::OffsetDateTime; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; use tokio::sync::Mutex; +use tokio::task::JoinHandle; use tracing::{debug, info, warn}; use crate::context::{Key, Next, Store}; -use crate::protocol::{Answer, Confirm, Message, Report, VERSION}; +use crate::decisions::{Account, Record}; +use crate::pending::Held; +use crate::protocol::{ + Answer, Approve, Confirm, DecisionForAgent, Decline, Message, Observed, Report, VERSION, +}; use crate::registrar::{Registrar, Wanted}; use crate::socket::Listener; @@ -21,6 +34,9 @@ pub struct Daemon { contexts: Mutex, registrar: R, confirmer: Option, + decisions: Option>, + held: Arc>, + pollers: Mutex>>, } impl Daemon { @@ -30,6 +46,9 @@ impl Daemon { contexts: Mutex::new(Store::new()), registrar, confirmer: None, + decisions: None, + held: Arc::new(Mutex::new(Held::new())), + pollers: Mutex::new(HashMap::new()), } } @@ -39,21 +58,50 @@ impl Daemon { self } + /// Give it somewhere to fetch and answer sign-in decisions. + /// + /// Optional, like the confirmer: a daemon with no server to ask still + /// issues identities, and says plainly that it has nowhere to ask rather + /// than failing to start. + pub fn deciding_at(mut self, server: impl Into) -> Self { + self.decisions = Some(Arc::new(crate::decisions::Pds::new(server))); + self + } + /// Confirm an authorization for the account the caller names. /// - /// See [`Confirm`] for what taking that from the caller costs. + /// Deprecated; see [`Confirm`]. When this daemon is already holding the + /// decision the URL names, that record is approved by token and the + /// caller's `did` is checked against it rather than believed. Otherwise + /// this is what it always was: read the page, post the reference. pub async fn confirm(&self, confirm: Confirm) -> Answer { - if confirm.version != VERSION { - return Answer::trouble(format!( - "this daemon speaks version {VERSION}, the request is version {}", - confirm.version - )); + if let Some(refusal) = too_new(confirm.version) { + return refusal; } + + let did = confirm.did.clone(); + if let Some(held) = self.held_for_url(&confirm.url).await { + return match held.record.token.clone() { + Some(token) if held.record.account == did => { + info!(did = %did, "approving a decision this daemon holds"); + self.redeem(&token).await + } + Some(_) => Answer::trouble(format!( + "that request is for {}, and this asks to confirm it as {did}", + held.record.account + )), + // A refused request has no token, and reading the page would + // not produce one either. + None => Answer::trouble( + "that request was refused, so there is nothing to confirm".to_owned(), + ), + }; + } + let Some(confirmer) = self.confirmer.as_ref() else { return Answer::trouble("this daemon has nowhere to confirm against".to_owned()); }; - let did = confirm.did.clone(); match confirmer.run(&confirm.url, &did).await { Ok(redirect) => { info!(did = %did, "confirmed an authorization"); @@ -67,6 +115,127 @@ impl Daemon { } } + /// Approve a sign-in, as the account the decision itself names. + /// + /// The caller supplies a token and nothing else. Which account signs in + /// is read off the record the daemon is holding that token in, so a + /// caller cannot name one -- see [`Approve`]. + pub async fn approve(&self, approve: Approve) -> Answer { + if let Some(refusal) = too_new(approve.version) { + return refusal; + } + self.redeem(&approve.token).await + } + + /// Decline one, so the refusal is recorded rather than left to expire. + pub async fn decline(&self, decline: Decline) -> Answer { + if let Some(refusal) = too_new(decline.version) { + return refusal; + } + let Some((pds, account, holding)) = self.holder_of(&decline.token).await else { + return Answer::trouble(unknown_token(&decline.token)); + }; + + match pds + .decline(&account, &decline.token, decline.reason.as_deref()) + .await + { + Ok(declined) => { + info!(did = %account.did, "declined a sign-in"); + self.held.lock().await.forget(&declined.request_uri); + Answer::done(format!("declined {}", holding.record.client.origin)) + } + Err(err) => { + warn!(did = %account.did, error = %err, "could not decline"); + Answer::trouble(err.to_string()) + } + } + } + + /// Everything this daemon is holding, for a caller that cannot say which + /// context it is. See [`crate::protocol::Pending`]. + pub async fn pending(&self, pending: crate::protocol::Pending) -> Answer { + if let Some(refusal) = too_new(pending.version) { + return refusal; + } + self.held.lock().await.sweep(OffsetDateTime::now_utc()); + let held = self.held.lock().await.all(); + Answer::quiet().and_pending(held.iter().map(DecisionForAgent::from).collect()) + } + + /// Redeem an approval token and hand the client its code. + async fn redeem(&self, token: &str) -> Answer { + let Some((pds, account, holding)) = self.holder_of(token).await else { + return Answer::trouble(unknown_token(token)); + }; + + let approved = match pds.approve(&account, token).await { + Ok(approved) => approved, + Err(err) => { + warn!(did = %account.did, error = %err, "could not approve"); + return Answer::trouble(err.to_string()); + } + }; + + // Spent either way: the server has moved the record on, and holding + // it here would offer the agent a choice it has already made. + self.held.lock().await.forget(&approved.request_uri); + + // The server cannot reach the client -- it is listening on this host + // -- so the last hop is the daemon's, and it is bounded. + if let Some(redirect) = approved.redirect.as_deref() { + if let Err(err) = pds.deliver(redirect).await { + warn!(did = %account.did, error = %err, "the code did not reach the client"); + return Answer::trouble(format!( + "signed in as {}, but the code did not reach the client: {err}", + account.did + )); + } + debug!(redirect = %redirect, "delivered the code"); + } + + info!(did = %account.did, origin = %holding.record.client.origin, "approved a sign-in"); + Answer::done(format!( + "signed in to {} as {}", + holding.record.client.origin, account.did + )) + .and_granted(approved.granted) + } + + /// The server, the account and the record one approval token belongs to. + /// + /// All three or none: a token this daemon is not holding names no + /// account, and a daemon with no server to ask cannot act on one. + async fn holder_of( + &self, + token: &str, + ) -> Option<(Arc, Account, crate::pending::Holding)> { + let pds = Arc::clone(self.decisions.as_ref()?); + let holding = self.held.lock().await.by_token(token).cloned()?; + let account = self.account_of(&holding.record.account).await?; + Some((pds, account, holding)) + } + + /// The credential for one account, if a context here was issued it. + async fn account_of(&self, did: &str) -> Option { + let contexts = self.contexts.lock().await; + let context = contexts.by_did(did)?; + Some(Account { + did: did.to_owned(), + token: context.token.clone()?, + }) + } + + /// The decision an authorize URL names, when this daemon holds it. + async fn held_for_url(&self, url: &str) -> Option { + let request_uri = url::Url::parse(url) + .ok()? + .query_pairs() + .find(|(name, _)| name == "request_uri") + .map(|(_, value)| value.into_owned())?; + self.held.lock().await.by_request_uri(&request_uri).cloned() + } + /// Serve until the listener fails. pub async fn run(self: Arc, listener: Listener) -> std::io::Error where @@ -102,6 +271,9 @@ impl Daemon { let answer = match serde_json::from_str::(&line) { Ok(Message::Report(report)) => self.consider(report).await, Ok(Message::Confirm(confirm)) => self.confirm(confirm).await, + Ok(Message::Approve(approve)) => self.approve(approve).await, + Ok(Message::Decline(decline)) => self.decline(decline).await, + Ok(Message::Pending(pending)) => self.pending(pending).await, Err(err) => { warn!(error = %err, "unreadable message"); Answer::trouble(format!("unreadable message: {err}")) @@ -114,33 +286,50 @@ impl Daemon { } /// Decide what one report is owed, and mint if that is a name. + /// + /// Every answer also carries whatever sign-in decisions are waiting for + /// the context that reported, which is how an agent hears about one at + /// all: nothing here opens a connection to the adapter. pub async fn consider(&self, report: Report) -> Answer { - if report.version != VERSION { - // Answered rather than dropped: an adapter that gets silence - // cannot tell a version it is too old for from a daemon that is - // not running. - return Answer::trouble(format!( - "this daemon speaks version {VERSION}, the report is version {}", - report.version - )); + if let Some(refusal) = too_new(report.version) { + return refusal; } + let key = Key { + session: report.session.clone(), + context: report.context.clone(), + }; let next = self.contexts.lock().await.observe(&report); - let key = match next { - Next::Nothing => return Answer::quiet(), + + // A context that will do no more has no use for a poll, and its + // account has nothing left to sign into. + if report.observed == Observed::Ended { + self.stop_polling(&key).await; + return Answer::quiet(); + } + + let answer = match next { + Next::Nothing => Answer::quiet(), Next::Tell(did) => { - let key = Key { - session: report.session.clone(), - context: report.context.clone(), - }; self.contexts.lock().await.told(&key, asker(&report)); - return Answer::identity(did); + Answer::identity(did) } - Next::Reserve(key) => key, + Next::Reserve(key) => self.reserve(&key, &report).await, }; + if answer.trouble.is_some() { + return answer; + } + self.fetch_seen(&key, &report.seen_request_uris).await; + self.held.lock().await.sweep(OffsetDateTime::now_utc()); + let waiting = self.held.lock().await.for_context(&key); + answer.and_pending(waiting.iter().map(DecisionForAgent::from).collect()) + } + + /// Mint an account for a context and start following its sign-ins. + async fn reserve(&self, key: &Key, report: &Report) -> Answer { let wanted = Wanted { - agent_id: agent_id(&key), + agent_id: agent_id(key), kind: report.kind.clone(), parent: None, }; @@ -152,10 +341,20 @@ impl Daemon { info!(did = %identity.did, handle = %identity.handle, "reserved"); let mut contexts = self.contexts.lock().await; // The credential stays here, in memory, for as long as the - // context does: `plan/cred-delivery.md` puts it in the daemon - // rather than anywhere the model can read. - contexts.reserved(&key, &identity.did, identity.token); - contexts.told(&key, asker(&report)); + // context does. `plan/cred-delivery.md` puts it in the + // daemon rather than anywhere the model can read. + contexts.reserved(key, &identity.did, identity.token.clone()); + contexts.told(key, asker(report)); + drop(contexts); + + self.start_polling( + key.clone(), + Account { + did: identity.did.clone(), + token: identity.token, + }, + ) + .await; Answer::identity(identity.did) } Err(err) => { @@ -165,6 +364,87 @@ impl Daemon { } } + /// Fetch decisions the adapter saw go past that no poll has delivered. + /// + /// The fast path for the tool call currently being reported: a client + /// that printed its authorize URL a moment ago may not be in a poll's + /// answer yet. Anything already held costs nothing. + async fn fetch_seen(&self, key: &Key, seen: &[String]) { + if seen.is_empty() { + return; + } + let Some(pds) = self.decisions.as_ref() else { + return; + }; + let Some(did) = self + .contexts + .lock() + .await + .get(key) + .and_then(|c| c.did.clone()) + else { + return; + }; + let Some(account) = self.account_of(&did).await else { + return; + }; + + for request_uri in seen { + if self.held.lock().await.by_request_uri(request_uri).is_some() { + continue; + } + match pds.get(&account, request_uri).await { + Ok(record) => { + let now = OffsetDateTime::now_utc(); + self.held.lock().await.keep(key.clone(), record, now); + } + Err(err) => { + // Not the caller's trouble: the report itself succeeded, + // and a decision that could not be fetched arrives on the + // next poll or not at all. + debug!(%request_uri, error = %err, "could not fetch a request the adapter saw"); + } + } + } + } + + /// Start following one account's sign-ins, outbound. + async fn start_polling(&self, key: Key, account: Account) { + let Some(pds) = self.decisions.clone() else { + return; + }; + let held = Arc::clone(&self.held); + let task = tokio::spawn(crate::poll::run(pds, held, key.clone(), account)); + if let Some(replaced) = self.pollers.lock().await.insert(key, task) { + replaced.abort(); + } + } + + /// Stop following one, and drop what it collected. + async fn stop_polling(&self, key: &Key) { + let Some(task) = self.pollers.lock().await.remove(key) else { + return; + }; + task.abort(); + debug!(session = %key.session, "stopped following a context"); + } + + /// How many contexts this daemon is following. The invariant is that a + /// context that has ended is not one of them. + pub async fn following(&self) -> usize { + self.pollers + .lock() + .await + .values() + .filter(|task| !task.is_finished()) + .count() + } + + /// Every decision this daemon is holding. + pub async fn holding(&self) -> Vec { + self.held.lock().await.all() + } + /// Every context this daemon has seen. pub async fn contexts(&self) -> Vec { self.contexts @@ -177,6 +457,31 @@ impl Daemon { } } +/// Refuses a message written against a version this daemon cannot read. +/// +/// Newer only. An adapter and a daemon are installed months apart by +/// different packages, so an adapter one version behind is the ordinary case +/// and is answered: every field version 1 sent is still read, and the extra +/// fields in the answer are ones it ignores. A version from the future is +/// another matter, because this cannot know what it left out. +fn too_new(version: u32) -> Option { + (version > VERSION).then(|| { + // Answered rather than dropped: an adapter that gets silence cannot + // tell a version it is too new for from a daemon that is not running. + Answer::trouble(format!( + "this daemon speaks version {VERSION}, the message is version {version}" + )) + }) +} + +/// What to say about a token this daemon is not holding. +fn unknown_token(token: &str) -> String { + format!( + "this daemon is holding no sign-in with the token `{token}`, so there is no account to \ + approve or decline as" + ) +} + /// Which plugin a report came from, for the once-per-asker bookkeeping. fn asker(report: &Report) -> &str { report.asker.as_deref().unwrap_or("-") @@ -207,10 +512,11 @@ fn agent_id(key: &Key) -> String { #[cfg(test)] mod tests { use super::*; - use crate::protocol::Observed; + use crate::double; use crate::registrar::{Identity, Trouble}; use crate::secret::Secret; use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; /// A registrar that mints without a server, and counts how often it was /// asked. The count is the point: telling a context its name twice is @@ -244,9 +550,32 @@ mod tests { kind: None, asker: None, call: None, + seen_request_uris: Vec::new(), } } + /// The identity a `Counting` registrar hands the only context these + /// tests use. + const MINTED: &str = "did:web:a-1.example"; + + /// Wait for something the poller does in the background, or give up. + /// + /// A poll is a round trip to the double, so the alternative to waiting is + /// a test that passes only on a fast machine. + macro_rules! until { + ($what:literal, $ready:expr) => {{ + let mut arrived = false; + for _ in 0..200 { + if $ready { + arrived = true; + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(arrived, concat!("never: ", $what)); + }}; + } + #[tokio::test] async fn a_context_is_minted_once_and_told_once() { let daemon = Daemon::new(Counting::default()); @@ -264,24 +593,6 @@ mod tests { assert_eq!(daemon.registrar.minted.load(Ordering::SeqCst), 1); } - #[tokio::test] - async fn the_credential_is_kept_and_never_said() { - let daemon = Daemon::new(Counting::default()); - let answer = daemon.consider(report(Observed::Began, Some("a-1"))).await; - - // Held, so the daemon can act as the account later. - let contexts = daemon.contexts().await; - let held = contexts[0].token.as_ref().expect("the credential"); - assert_eq!(held.reveal(), "agent-token"); - - // And nowhere the model can see it: not in the answer that goes back - // over the socket, and not in what the daemon prints about itself. - let line = serde_json::to_string(&answer).unwrap(); - assert!(!line.contains("agent-token"), "{line}"); - let printed = format!("{contexts:?}"); - assert!(!printed.contains("agent-token"), "{printed}"); - } - #[tokio::test] async fn a_context_that_never_announced_itself_is_named_on_its_first_call() { let daemon = Daemon::new(Counting::default()); @@ -341,6 +652,267 @@ mod tests { assert_eq!(daemon.registrar.minted.load(Ordering::SeqCst), 0); } + #[tokio::test] + async fn a_report_from_a_version_behind_this_one_is_answered_normally() { + let daemon = Daemon::new(Counting::default()); + let mut behind = report(Observed::Began, Some("a-1")); + behind.version = 1; + let answer = daemon.consider(behind).await; + // A version 1 adapter is the ordinary case, not a skew to refuse: + // packages update at different times, and everything it sent is + // still read. + assert_eq!(answer.identity.as_deref(), Some(MINTED)); + assert!(answer.trouble.is_none()); + } + + #[tokio::test] + async fn the_credential_is_kept_and_never_said() { + let daemon = Daemon::new(Counting::default()); + let answer = daemon.consider(report(Observed::Began, Some("a-1"))).await; + + // Held, so the daemon can act as the account later. + let contexts = daemon.contexts().await; + let held = contexts[0].token.as_ref().expect("the credential"); + assert_eq!(held.reveal(), "agent-token"); + + // And nowhere the model can see it: not in the answer that goes back + // over the socket, and not in what the daemon prints about itself. + let line = serde_json::to_string(&answer).unwrap(); + assert!(!line.contains("agent-token"), "{line}"); + let printed = format!("{contexts:?}"); + assert!(!printed.contains("agent-token"), "{printed}"); + } + + /// A daemon with a server to ask, and a context already provisioned. + async fn daemon_with(double: &double::Double) -> Daemon { + let daemon = Daemon::new(Counting::default()).deciding_at(&double.origin); + daemon.consider(report(Observed::Began, Some("a-1"))).await; + daemon + } + + /// A record the double will hand out, for the account `Counting` mints. + fn offered(double: &double::Double, request_uri: &str, token: &str) -> Record { + let mut record = double::record(request_uri, MINTED, Some(token), &double::later()); + // The client's callback is served by the double itself, so a + // delivered code actually arrives somewhere. + record.client.origin = double.origin.clone(); + record + } + + #[tokio::test] + async fn a_decision_the_server_has_reaches_the_context_it_belongs_to() { + let double = double::start().await; + double.state.offer(offered(&double, "r1", "k1")); + let daemon = daemon_with(&double).await; + + until!("the poll found it", !daemon.holding().await.is_empty()); + + let answer = daemon.consider(report(Observed::Acted, Some("a-1"))).await; + let pending = answer.pending.expect("the decision"); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].token.as_deref(), Some("k1")); + assert!(pending[0].first_time); + + // And the poll presented the account's own credential to get it. + assert!( + double.state.presented().contains(&"agent-token".to_owned()), + "{:?}", + double.state.presented() + ); + + // Another context on this machine is offered nothing. + let other = daemon.consider(report(Observed::Began, Some("a-2"))).await; + assert!(other.pending.is_none()); + } + + #[tokio::test] + async fn one_whose_moment_has_passed_is_never_offered() { + let double = double::start().await; + let mut stale = offered(&double, "r1", "k1"); + stale.expires_at = double::earlier(); + double.state.offer(stale); + let daemon = daemon_with(&double).await; + + until!("the poll happened", double.state.polls() > 0); + assert!(daemon.holding().await.is_empty()); + let answer = daemon.consider(report(Observed::Acted, Some("a-1"))).await; + assert!(answer.pending.is_none()); + } + + #[tokio::test] + async fn a_context_that_has_ended_is_no_longer_followed() { + let double = double::start().await; + let daemon = daemon_with(&double).await; + assert_eq!(daemon.following().await, 1); + + daemon.consider(report(Observed::Ended, Some("a-1"))).await; + until!("the task stopped", daemon.following().await == 0); + } + + #[tokio::test] + async fn approving_signs_in_as_the_account_the_decision_names() { + let double = double::start().await; + double.state.offer(offered(&double, "r1", "k1")); + let daemon = daemon_with(&double).await; + until!("the poll found it", !daemon.holding().await.is_empty()); + + let answer = daemon + .approve(Approve { + version: VERSION, + token: "k1".into(), + }) + .await; + + assert!(answer.trouble.is_none(), "{answer:?}"); + assert_eq!(answer.granted, Some(vec!["atproto".to_owned()])); + assert_eq!(double.state.approved(), vec!["k1".to_owned()]); + // The code reached the client, which the server could not have done. + until!("the code arrived", !double.state.delivered().is_empty()); + assert_eq!(double.state.delivered(), vec!["abc".to_owned()]); + // And the decision is spent, so the agent is not asked twice. + assert!(daemon.holding().await.is_empty()); + } + + #[tokio::test] + async fn a_token_this_daemon_never_saw_names_no_account_to_approve_as() { + let double = double::start().await; + let daemon = daemon_with(&double).await; + until!("the poll happened", double.state.polls() > 0); + let before = double.state.approved().len(); + + let answer = daemon + .approve(Approve { + version: VERSION, + token: "never-issued".into(), + }) + .await; + + assert!(answer.trouble.unwrap().contains("never-issued")); + // Not passed on: the daemon has no account to present, so there is + // nothing to ask the server on behalf of. + assert_eq!(double.state.approved().len(), before); + } + + #[tokio::test] + async fn a_refusal_from_the_server_reaches_the_agent_in_its_own_words() { + let double = double::start().await; + double.state.offer(offered(&double, "r1", "k1")); + double + .state + .refuse(400, "InvalidToken", "that token has been used"); + let daemon = daemon_with(&double).await; + until!("the poll found it", !daemon.holding().await.is_empty()); + + let answer = daemon + .approve(Approve { + version: VERSION, + token: "k1".into(), + }) + .await; + + let trouble = answer.trouble.expect("a refusal"); + assert!(trouble.contains("InvalidToken"), "{trouble}"); + assert!(trouble.contains("that token has been used"), "{trouble}"); + assert!(double.state.delivered().is_empty()); + } + + #[tokio::test] + async fn declining_is_recorded_rather_than_left_to_expire() { + let double = double::start().await; + double.state.offer(offered(&double, "r1", "k1")); + let daemon = daemon_with(&double).await; + until!("the poll found it", !daemon.holding().await.is_empty()); + + let answer = daemon + .decline(Decline { + version: VERSION, + token: "k1".into(), + reason: Some("not something I asked for".into()), + }) + .await; + + assert!(answer.trouble.is_none(), "{answer:?}"); + assert_eq!( + double.state.declined(), + vec![( + "k1".to_owned(), + Some("not something I asked for".to_owned()) + )] + ); + assert!(daemon.holding().await.is_empty()); + } + + #[tokio::test] + async fn what_the_adapter_saw_go_past_is_fetched_before_a_poll_finds_it() { + let double = double::start().await; + // Known but never offered to a poll, so the only way to it is the + // fast path. + double.state.know(offered(&double, "r1", "k1")); + let daemon = daemon_with(&double).await; + + let mut seen = report(Observed::Acted, Some("a-1")); + seen.seen_request_uris = vec!["r1".into()]; + let answer = daemon.consider(seen).await; + + let pending = answer.pending.expect("the decision"); + assert_eq!(pending[0].token.as_deref(), Some("k1")); + } + + #[tokio::test] + async fn the_old_confirm_approves_a_decision_this_daemon_is_holding() { + let double = double::start().await; + double.state.offer(offered(&double, "r1", "k1")); + let daemon = daemon_with(&double).await; + until!("the poll found it", !daemon.holding().await.is_empty()); + + // No authorize page is read and no consent reference is scraped: the + // URL is only where the request_uri comes from. + let answer = daemon + .confirm(Confirm { + version: VERSION, + did: MINTED.into(), + url: format!("{}/oauth/authorize?request_uri=r1", double.origin), + }) + .await; + + assert!(answer.trouble.is_none(), "{answer:?}"); + assert_eq!(double.state.approved(), vec!["k1".to_owned()]); + } + + #[tokio::test] + async fn and_refuses_a_caller_naming_an_account_that_is_not_the_requests() { + let double = double::start().await; + double.state.offer(offered(&double, "r1", "k1")); + let daemon = daemon_with(&double).await; + until!("the poll found it", !daemon.holding().await.is_empty()); + + let answer = daemon + .confirm(Confirm { + version: VERSION, + did: "did:web:somebody.else".into(), + url: format!("{}/oauth/authorize?request_uri=r1", double.origin), + }) + .await; + + assert!(answer.trouble.is_some(), "{answer:?}"); + assert!(double.state.approved().is_empty()); + } + + #[tokio::test] + async fn asking_what_is_held_answers_from_memory() { + let double = double::start().await; + double.state.offer(offered(&double, "r1", "k1")); + let daemon = daemon_with(&double).await; + until!("the poll found it", !daemon.holding().await.is_empty()); + + let answer = daemon + .pending(crate::protocol::Pending { version: VERSION }) + .await; + let pending = answer.pending.expect("the decision"); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].client_origin, double.origin); + } + #[test] fn an_id_the_harness_invents_survives_becoming_a_label() { let key = Key { diff --git a/plan/oauth.md b/plan/oauth.md index 1a17c45d..36a7744c 100644 --- a/plan/oauth.md +++ b/plan/oauth.md @@ -2,7 +2,7 @@ id: oauth title: A third-party app signs in as an agent, with nobody at the consent screen status: open -crates: [didbot-serve, didbot-pds] +crates: [didbot-serve, didbot-pds, didbot-agentd] dependsOn: [pds-writes] exitCriterion: > An atproto client that has never heard of this project completes an OAuth @@ -30,32 +30,32 @@ authored from is [policy-store](policy-store.md)'s. ## How an ordinary client signs in as an agent -No browser, and no URL in the agent's hands: +A pushed authorization request naming an agent becomes a decision record at +the server, and the one-time consent reference minted with it is the approval +token. The daemon relays the record and carries the answer back; nothing here +needs a browser, and the agent is never handed a URL. + +Two tool calls, and no browser anywhere: 1. The agent starts an ordinary client in the background. It binds its - loopback listener and pushes an authorization request naming its own DID as - the account. That push is where the decision is minted. -2. The daemon holding that account's agent token is long-polling - `bot.did.listPendingAuthorizations` and sees the decision. The agent is - shown what was asked, what would be granted, and what was cut, and answers - `approve` or `decline` with the decision's own token. -3. On an approval this server issues the code at the granted scopes and - answers with the client's own callback URL. The daemon fetches it — this - server cannot reach the agent's host — and the client's listener gets the - code. + loopback listener, pushes an authorization request naming its own DID as + the account, and prints the authorize URL instead of opening it. +2. The pushed request becomes a decision record at the server. The daemon is + already long-polling for records about the accounts it issued, so it + collects that one and puts it in front of the agent on its next tool call. +3. The agent runs `didbot approve ` or `didbot decline `. The + daemon redeems the token as the account the record names, and fetches the + redirect the server built — which is what delivers the code. 4. The client exchanges it and holds tokens bound to its own key. -- [ ] **The daemon is the user-agent, and that is a request the model - influenced.** It must accept only this deployment's own authorize - endpoint over https, and follow only a loopback redirect. Nothing else, - no chains. This server's half is built: `approveAuthorization` answers - with the redirect it built from the pending consent's own `redirect_uri` - and `state`, so the daemon fetches a URL it was handed rather than one a - model composed. The bounds on that fetch are the daemon's. -- [ ] **One thing a client must support, and no more.** Accept an account - identifier to send as `login_hint`. It does not have to open a browser, - print a URL, or have heard of this project: the decision is answered out - of band and its own loopback listener gets the code. +The agent never handles a URL and never names its account. `didbot confirm + --as ` still works and is kept for one release; when the daemon is +already holding the record that URL's `request_uri` names, it approves that +record by token instead of reading the page. + +- [ ] **Two things a client must support, and no more.** Print the URL rather + than opening it, and accept an account identifier to resolve. Neither + requires it to have heard of this project. - [ ] **Replace the refuse-everything default with policy.** [policy](policy.md) decides which client is denied; absent a denial the agent approves what it asked for. @@ -132,6 +132,26 @@ No browser, and no URL in the agent's hands: with the record and never serialized — `app-allowlist`'s "record the client's own copy; do not show it". Driven end to end over the real router in `crates/didbot-serve/tests/oauth_agent_flow.rs`. +- [x] **The daemon is the user-agent, and that is a request the model + influenced.** Every fetch it makes on a request the model touched is + bounded: the deprecated page path accepts only the authorize endpoint + this deployment publishes in its own discovery document, the approval + path fetches only a redirect the server built, and both deliver only to + loopback (`didbot_agentd::loopback`), with redirects turned off on every + client so there are no chains. The server cannot make that last fetch + itself — the client is listening on the agent host and, by the + architecture rule, the server is not on it. +- [x] **The daemon relays, outbound only.** One long poll per account it + issued (`didbot_agentd::poll`, `listPendingAuthorizations?wait=25`), + backing off from a second to half a minute when the server is not + answering, started when a context is provisioned and stopped when the + harness says that context has ended. Nothing on the agent host listens + for the server. Decisions ride the answer to an exchange the hook + started, so an agent learns of one at its next tool call; that latency + is accepted rather than hidden. `seen_request_uris` on a report is the + fast path for a client that printed its URL in the tool call being + reported, fetched with `getAuthorization` when no poll has delivered it + yet. - [x] **The granular atproto scope grammar.** `crates/didbot-serve/src/oauth/scope.rs`: `Scope::parse`/`ScopeSet::parse` read the wire grammar directly, not a parallel one — `repo:`, `rpc:`, `blob:`, `identity:`, `account:`,