use std::sync::Arc;
use std::time::Duration;
use aauth_protocol::discovery::{Fetch, JwksResolver, KeyResolver, NoKeys};
use aauth_protocol::httpsig::{self, Incoming, SignatureKey, VerifyOptions};
use aauth_protocol::identifier::{is_server_identifier, trim};
use aauth_protocol::jwk::{Algorithm, Jwks};
use aauth_protocol::metadata::{dwk, PersonServerMetadata};
use aauth_protocol::token::{mint, verify_token, TokenKind, VerifyTokenOptions};
use aauth_protocol::{jwt, Budget, SigningKey};
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine;
use bytes::Bytes;
use hmac::{Hmac, Mac};
use http::{HeaderValue, Method, Request, Response, StatusCode};
use percent_encoding::{percent_decode_str, utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC};
use serde::Serialize;
use serde_json::{json, Map, Value};
use sha2::Sha256;
use crate::directory::{Directory, Hooks, NoHooks};
use crate::error::{problem, AAuthError};
use crate::store::{MemoryStore, Spend, Store};
use crate::time::rfc3339;
use crate::types::{
Agent, Approval, ApprovalKind, ApprovalStatus, Decision, Issued, IssuedKind, LogEntry, Mission,
Person, Verdict,
};
type Clock = Arc<dyn Fn() -> i64 + Send + Sync>;
type Trust = Arc<dyn Fn(&str) -> bool + Send + Sync>;
#[derive(Clone, Debug)]
enum Tenancy {
Default(String),
Required,
}
pub struct Config {
issuer: String,
directory: Arc<dyn Directory>,
trusted: Trust,
signing_key: Option<SigningKey>,
subject_secret: Option<String>,
store: Arc<dyn Store>,
hooks: Arc<dyn Hooks>,
fetch: Option<Arc<dyn Fetch>>,
base_path: String,
tenancy: Tenancy,
admin_token: Option<String>,
name: Option<String>,
description: Option<String>,
token_ttl: Duration,
mission_ttl: Duration,
approval_ttl: Duration,
retry_after: Duration,
now: Clock,
}
pub const DEFAULT_TENANT: &str = "default";
impl Config {
pub fn new<I, S>(
issuer: &str,
directory: impl Directory + 'static,
trusted_agent_providers: I,
) -> Self
where
I: IntoIterator<Item = S>,
S: AsRef<str>,
{
let trusted: Vec<String> = trusted_agent_providers
.into_iter()
.map(|p| trim(p.as_ref()).to_string())
.collect();
Self {
issuer: trim(issuer).to_string(),
directory: Arc::new(directory),
trusted: Arc::new(move |iss| trusted.iter().any(|t| t == iss)),
signing_key: None,
subject_secret: None,
store: Arc::new(MemoryStore::new()),
hooks: Arc::new(NoHooks),
fetch: None,
base_path: String::new(),
tenancy: Tenancy::Default(DEFAULT_TENANT.to_string()),
admin_token: None,
name: None,
description: None,
token_ttl: Duration::from_secs(60 * 60),
mission_ttl: Duration::from_secs(14 * 24 * 60 * 60),
approval_ttl: Duration::from_secs(24 * 60 * 60),
retry_after: Duration::from_secs(5),
now: Arc::new(aauth_protocol::now_unix),
}
}
pub fn default_tenant(mut self, tenant: impl Into<String>) -> Self {
self.tenancy = Tenancy::Default(tenant.into());
self
}
pub fn require_tenant(mut self) -> Self {
self.tenancy = Tenancy::Required;
self
}
pub fn trust(mut self, trusted: impl Fn(&str) -> bool + Send + Sync + 'static) -> Self {
self.trusted = Arc::new(trusted);
self
}
pub fn signing_key(mut self, key: SigningKey) -> Self {
self.signing_key = Some(key);
self
}
pub fn subject_secret(mut self, secret: impl Into<String>) -> Self {
self.subject_secret = Some(secret.into());
self
}
pub fn store(mut self, store: Arc<dyn Store>) -> Self {
self.store = store;
self
}
pub fn base_path(mut self, base: &str) -> Self {
self.base_path = trim(base).to_string();
self
}
pub fn hooks(mut self, hooks: impl Hooks + 'static) -> Self {
self.hooks = Arc::new(hooks);
self
}
pub fn admin_token(mut self, token: impl Into<String>) -> Self {
self.admin_token = Some(token.into()).filter(|t: &String| !t.is_empty());
self
}
pub fn name(mut self, name: impl Into<String>) -> Self {
self.name = Some(name.into());
self
}
pub fn description(mut self, description: impl Into<String>) -> Self {
self.description = Some(description.into());
self
}
pub fn token_ttl(mut self, ttl: Duration) -> Self {
self.token_ttl = ttl;
self
}
pub fn mission_ttl(mut self, ttl: Duration) -> Self {
self.mission_ttl = ttl;
self
}
pub fn approval_ttl(mut self, ttl: Duration) -> Self {
self.approval_ttl = ttl;
self
}
pub fn retry_after(mut self, after: Duration) -> Self {
self.retry_after = after;
self
}
pub fn fetch(mut self, fetch: Arc<dyn Fetch>) -> Self {
self.fetch = Some(fetch);
self
}
pub fn clock(mut self, now: impl Fn() -> i64 + Send + Sync + 'static) -> Self {
self.now = Arc::new(now);
self
}
}
#[derive(Clone, Debug, Serialize)]
pub struct Endpoints {
pub person_token: String,
pub token: String,
pub mission: String,
pending_prefix: String,
}
impl Endpoints {
pub fn pending(&self, tenant: &str, id: &str) -> String {
format!(
"{}{}/{id}",
self.pending_prefix,
utf8_percent_encode(tenant, SEGMENT)
)
}
}
pub struct PersonServer {
config: Config,
key: SigningKey,
secret: String,
fetch: Arc<dyn Fetch>,
authority: String,
resolver: JwksResolver,
endpoints: Endpoints,
}
fn secs(duration: Duration) -> i64 {
i64::try_from(duration.as_secs()).unwrap_or(i64::MAX)
}
fn random_hex(bytes: usize) -> String {
let mut buf = vec![0u8; bytes];
getrandom::fill(&mut buf).expect("the operating system provides randomness");
buf.iter().map(|b| format!("{b:02x}")).collect()
}
fn read_body(body: &[u8]) -> Result<Map<String, Value>, AAuthError> {
match serde_json::from_slice(body) {
Ok(Value::Object(map)) => Ok(map),
_ => Err(AAuthError::invalid_request("body must be a JSON object")),
}
}
fn str_of<'a>(params: &'a Map<String, Value>, name: &str) -> Option<&'a str> {
params.get(name).and_then(Value::as_str)
}
fn json_response(status: StatusCode, body: &impl Serialize) -> Response<Bytes> {
let mut response = Response::new(Bytes::from(serde_json::to_vec(body).unwrap_or_default()));
*response.status_mut() = status;
response
.headers_mut()
.insert("content-type", HeaderValue::from_static("application/json"));
response
}
fn ok(body: &impl Serialize) -> Response<Bytes> {
json_response(StatusCode::OK, body)
}
const SEGMENT: &AsciiSet = &NON_ALPHANUMERIC.remove(b'-').remove(b'_').remove(b'.');
fn named(params: &Map<String, Value>) -> Option<&str> {
str_of(params, "tenant").filter(|t| !t.is_empty())
}
const GRANT_CLAIMS: &[&str] = &["scope", "r3_uri", "r3_s256"];
impl PersonServer {
pub fn new(mut config: Config) -> aauth_protocol::Result<Self> {
if !is_server_identifier(&config.issuer) {
return Err(aauth_protocol::Error::Invalid(format!(
"issuer {} must be https://host with no port or path",
config.issuer
)));
}
let fetch = match config.fetch.take() {
Some(fetch) => fetch,
None => default_fetch()?,
};
let authority = url::Url::parse(&config.issuer)
.ok()
.and_then(|u| u.host_str().map(str::to_string))
.unwrap_or_default();
let url = |path: &str| format!("{}{}{path}", config.issuer, config.base_path);
let endpoints = Endpoints {
person_token: url("/person-token"),
token: url("/token"),
mission: url("/mission"),
pending_prefix: url("/pending/"),
};
Ok(Self {
key: config
.signing_key
.take()
.unwrap_or_else(SigningKey::generate),
secret: config
.subject_secret
.take()
.unwrap_or_else(|| random_hex(32)),
resolver: JwksResolver::new(fetch.clone()),
fetch,
authority,
endpoints,
config,
})
}
pub fn issuer(&self) -> &str {
&self.config.issuer
}
pub fn endpoints(&self) -> &Endpoints {
&self.endpoints
}
pub fn store(&self) -> &Arc<dyn Store> {
&self.config.store
}
pub fn metadata(&self) -> PersonServerMetadata {
PersonServerMetadata {
issuer: self.config.issuer.clone(),
auth_token_endpoint: self.endpoints.token.clone(),
person_token_endpoint: Some(self.endpoints.person_token.clone()),
mission_endpoint: Some(self.endpoints.mission.clone()),
jwks_uri: Some(format!("{}/.well-known/jwks.json", self.config.issuer)),
name: self.config.name.clone(),
description: self.config.description.clone(),
}
}
pub fn jwks(&self) -> Jwks {
self.key.jwks()
}
fn now(&self) -> i64 {
(self.config.now)()
}
fn default_tenant(&self) -> Option<&str> {
match &self.config.tenancy {
Tenancy::Default(tenant) => Some(tenant),
Tenancy::Required => None,
}
}
fn tenant(&self, named: Option<&str>) -> Result<String, AAuthError> {
named
.map(str::to_string)
.or_else(|| self.default_tenant().map(str::to_string))
.ok_or_else(|| AAuthError::invalid_request("name the tenant"))
}
pub async fn handle(&self, request: Request<Bytes>) -> Response<Bytes> {
let path = request.uri().path().to_string();
self.handle_at(request, &path).await
}
pub async fn handle_at(&self, request: Request<Bytes>, pathname: &str) -> Response<Bytes> {
match self.route(&request, pathname).await {
Ok(Some(response)) => response,
Ok(None) => problem(StatusCode::NOT_FOUND, "not_found", None),
Err(error) => error.into_response(),
}
}
pub async fn route(
&self,
request: &Request<Bytes>,
pathname: &str,
) -> Result<Option<Response<Bytes>>, AAuthError> {
let method = request.method();
if method == Method::GET && pathname == format!("/.well-known/{}", dwk::PERSON) {
return Ok(Some(ok(&self.metadata())));
}
if method == Method::GET && pathname == "/.well-known/jwks.json" {
return Ok(Some(ok(&self.jwks())));
}
let Some(path) = pathname.strip_prefix(self.config.base_path.as_str()) else {
return Ok(None);
};
if !self.config.base_path.is_empty() && !path.starts_with('/') {
return Ok(None);
}
let segments: Vec<&str> = path.trim_start_matches('/').split('/').collect();
let response = match (method, segments.as_slice()) {
(&Method::POST, ["mission"]) => self.propose_mission(request, path).await?,
(&Method::POST, ["mission", s256]) if !s256.is_empty() => {
self.mission_action(request, path, s256).await?
}
(&Method::POST, ["person-token"]) => self.request_person_token(request, path).await?,
(&Method::POST, ["token"]) => self.request_auth_token(request, path).await?,
(&Method::GET | &Method::DELETE, ["pending", id]) if !id.is_empty() => {
let tenant = self.tenant(None)?;
self.pending_state(request, path, &tenant, id).await?
}
(&Method::GET | &Method::DELETE, ["pending", tenant, id])
if !tenant.is_empty() && !id.is_empty() =>
{
let tenant = percent_decode_str(tenant).decode_utf8_lossy();
self.pending_state(request, path, &tenant, id).await?
}
_ => return self.manage(request, &segments).await,
};
Ok(Some(response))
}
async fn verify_agent(
&self,
request: &Request<Bytes>,
path: &str,
) -> Result<Agent, AAuthError> {
let full_path = format!("{}{path}", self.config.base_path);
let body = request.body();
let verified = httpsig::verify(
&Incoming {
method: request.method().as_str(),
authority: &self.authority,
path: &full_path,
query: request.uri().query(),
headers: request.headers(),
body: (!body.is_empty()).then_some(body.as_ref()),
},
&VerifyOptions {
now: self.now(),
require_content_digest: true,
algorithms: vec![Algorithm::Ed25519, Algorithm::Es256],
..VerifyOptions::default()
},
&NoKeys,
)
.await
.map_err(|e| AAuthError::signature(e.code, e.detail))?;
let Some(raw) = verified.jwt() else {
return Err(AAuthError::signature(
"unsupported_scheme",
"sign with sig=jwt",
));
};
let decoded =
jwt::decode(raw).map_err(|_| AAuthError::signature("invalid_jwt", "not a JWT"))?;
let provider = trim(decoded.claim_str("iss").unwrap_or_default()).to_string();
if !(self.config.trusted)(&provider) {
return Err(AAuthError::signature(
"invalid_jwt",
format!("{provider} is not trusted here"),
));
}
let token = verify_token(
&VerifyTokenOptions {
jwt: raw,
http_signature_thumbprint: &verified.thumbprint,
resource: &self.config.issuer,
accept: &[TokenKind::Agent],
now: self.now(),
clock_tolerance: 60,
},
&self.resolver,
)
.await
.map_err(|e| {
let code = if e.code == "token_expired" {
"expired_jwt"
} else {
"invalid_jwt"
};
AAuthError::signature(code, e.detail)
})?;
if token.ps().is_some_and(|ps| trim(ps) != self.config.issuer) {
return Err(AAuthError::forbidden(
"this agent's Person Server is another",
));
}
Ok(Agent {
id: token.sub,
provider,
jkt: verified.thumbprint,
jwk: token.cnf_jwk,
exp: token.exp,
})
}
async fn person_for(
&self,
tenant: &str,
named: Option<&str>,
agent: &Agent,
login_hint: Option<&str>,
) -> Result<Person, AAuthError> {
let Some(hint) = login_hint else {
return self
.config
.store
.binding(tenant, &agent.jkt)
.await?
.ok_or_else(|| AAuthError::invalid_request("name the person: login_hint"));
};
let Some(hinted) = self.config.directory.person(hint, tenant).await else {
return Err(AAuthError::invalid_request(
"not someone this Person Server speaks for",
));
};
if hinted
.tenant
.as_deref()
.is_some_and(|theirs| theirs != tenant)
{
return Err(AAuthError::forbidden("this person is not in that tenant"));
}
let hinted = Person {
tenant: hinted.tenant.or_else(|| named.map(str::to_string)),
..hinted
};
let bound = self.config.store.bind(tenant, &agent.jkt, &hinted).await?;
if bound.id != hinted.id {
return Err(AAuthError::forbidden("this agent acts for another person"));
}
Ok(hinted)
}
fn directed(&self, person: &Person, audience: &str) -> String {
let mut mac = <Hmac<Sha256> as Mac>::new_from_slice(self.secret.as_bytes())
.expect("HMAC takes any key");
mac.update(format!("{}\n{audience}", person.id).as_bytes());
URL_SAFE_NO_PAD.encode(mac.finalize().into_bytes())
}
fn issue(
&self,
kind: TokenKind,
claims: Map<String, Value>,
not_after: i64,
person: &Person,
) -> (String, i64) {
let iat = self.now();
let exp = (iat + secs(self.config.token_ttl)).min(not_after);
let mut all = Map::new();
all.insert("iss".into(), json!(self.config.issuer));
all.insert("dwk".into(), json!(dwk::PERSON));
all.insert("jti".into(), json!(uuid::Uuid::new_v4().to_string()));
all.insert("iat".into(), json!(iat));
all.insert("exp".into(), json!(exp));
if let Some(tenant) = &person.tenant {
all.insert("tenant".into(), json!(tenant));
}
all.extend(claims);
(mint(&self.key, kind, &Value::Object(all)), exp - iat)
}
async fn person_token(
&self,
agent: &Agent,
person: &Person,
resource: &str,
mission: Option<&Mission>,
) -> Value {
let mut claims = Map::new();
claims.insert("aud".into(), json!(resource));
claims.insert("sub".into(), json!(self.directed(person, resource)));
claims.insert("cnf".into(), json!({ "jwk": agent.jwk }));
if let Some(mission) = mission {
claims.insert("mission_s256".into(), json!(mission.s256));
}
let not_after = agent.exp.min(mission.map_or(i64::MAX, |m| m.expires_at));
let (token, expires_in) = self.issue(TokenKind::Person, claims, not_after, person);
self.config
.hooks
.on_issued(&Issued {
kind: IssuedKind::PersonToken,
agent: agent.clone(),
person: person.clone(),
resource: resource.to_string(),
mission: mission.cloned(),
budget: None,
})
.await;
json!({ "person_token": token, "expires_in": expires_in })
}
async fn auth_token(
&self,
agent: &Agent,
mission: &Mission,
grant: &Map<String, Value>,
) -> Value {
let mut claims = Map::new();
claims.insert("ps".into(), json!(self.config.issuer));
claims.insert("cnf".into(), json!({ "jwk": agent.jwk }));
claims.insert("mission_s256".into(), json!(mission.s256));
claims.extend(grant.clone());
let (token, expires_in) = self.issue(
TokenKind::Auth,
claims,
agent.exp.min(mission.expires_at),
&mission.person,
);
self.config
.hooks
.on_issued(&Issued {
kind: IssuedKind::AuthToken,
agent: agent.clone(),
person: mission.person.clone(),
resource: grant
.get("aud")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string(),
mission: Some(mission.clone()),
budget: grant
.get("budget")
.and_then(|b| serde_json::from_value(b.clone()).ok()),
})
.await;
json!({ "auth_token": token, "expires_in": expires_in })
}
async fn request_person_token(
&self,
request: &Request<Bytes>,
path: &str,
) -> Result<Response<Bytes>, AAuthError> {
let agent = self.verify_agent(request, path).await?;
let params = read_body(request.body())?;
let resource = trim(str_of(¶ms, "resource").unwrap_or_default()).to_string();
if !resource.starts_with("https://") {
return Err(AAuthError::invalid_request("resource is an https origin"));
}
let tenant = self.tenant(named(¶ms))?;
let person = self
.person_for(
&tenant,
named(¶ms),
&agent,
str_of(¶ms, "login_hint"),
)
.await?;
let mission = match params.get("mission_s256") {
None | Some(Value::Null) => None,
Some(hash) => Some(self.live_mission(&tenant, &agent, hash.as_str()).await?),
};
if mission.as_ref().is_some_and(|m| m.person.id != person.id) {
return Err(AAuthError::mission_not_found());
}
Ok(ok(&self
.person_token(&agent, &person, &resource, mission.as_ref())
.await))
}
async fn live_mission(
&self,
tenant: &str,
agent: &Agent,
s256: Option<&str>,
) -> Result<Mission, AAuthError> {
let mission = match s256 {
Some(s256) => self.config.store.mission(tenant, s256).await?,
None => None,
};
let Some(mission) = mission.filter(|m| m.agent_id == agent.id) else {
return Err(AAuthError::mission_not_found());
};
if self
.config
.store
.binding(tenant, &agent.jkt)
.await?
.map(|p| p.id)
!= Some(mission.person.id.clone())
{
return Err(AAuthError::mission_not_found());
}
if mission.completed_at.is_some() {
return Err(AAuthError::mission_terminated("mission completed"));
}
if self.now() >= mission.expires_at {
return Err(AAuthError::mission_terminated("mission expired"));
}
Ok(mission)
}
async fn propose_mission(
&self,
request: &Request<Bytes>,
path: &str,
) -> Result<Response<Bytes>, AAuthError> {
let agent = self.verify_agent(request, path).await?;
let params = read_body(request.body())?;
let tenant = self.tenant(named(¶ms))?;
let person = self
.person_for(
&tenant,
named(¶ms),
&agent,
str_of(¶ms, "login_hint"),
)
.await?;
let description = str_of(¶ms, "description").unwrap_or_default();
if description.is_empty() {
return Err(AAuthError::invalid_request("describe the mission"));
}
let resources: Vec<Value> = params
.get("resources")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter_map(Value::as_str)
.map(|r| trim(r).to_string())
.filter(|r| r.starts_with("https://"))
.map(Value::from)
.collect();
let mut grant = Map::new();
grant.insert("resources".into(), Value::Array(resources));
let approval = self.approval(
&tenant,
ApprovalKind::Mission,
agent,
person,
description.to_string(),
);
self.ask(Approval { grant, ..approval }).await
}
async fn mission_action(
&self,
request: &Request<Bytes>,
path: &str,
s256: &str,
) -> Result<Response<Bytes>, AAuthError> {
let agent = self.verify_agent(request, path).await?;
let params = read_body(request.body())?;
let tenant = self.tenant(named(¶ms))?;
let tenant = tenant.as_str();
let mission = self.live_mission(tenant, &agent, Some(s256)).await?;
match str_of(¶ms, "action") {
Some("update") => {
let entry = str_of(¶ms, "description")
.unwrap_or_default()
.to_string();
let hash = aauth_protocol::s256(&entry);
let now = self.now();
if !self
.config
.store
.log_mission(tenant, &mission.s256, LogEntry { at: now, entry }, now)
.await?
{
return Err(AAuthError::mission_terminated("mission ended"));
}
Ok(ok(&json!({ "s256": hash })))
}
Some("completion") => {
let summary =
str_of(¶ms, "summary").unwrap_or("The agent says the mission is done");
let approval = self.approval(
tenant,
ApprovalKind::Completion,
agent,
mission.person.clone(),
summary.to_string(),
);
self.ask(Approval {
mission_s256: Some(mission.s256),
..approval
})
.await
}
_ => Err(AAuthError::invalid_request(
"action is update or completion",
)),
}
}
async fn verify_resource_token(
&self,
agent: &Agent,
params: &Map<String, Value>,
) -> Result<(Map<String, Value>, Mission, String), AAuthError> {
let (Some(resource_token), Some(presented_token)) = (
str_of(params, "resource_token"),
str_of(params, "presented_token"),
) else {
return Err(AAuthError::invalid_request(
"resource_token and presented_token are required",
));
};
let invalid = |detail: &str| {
AAuthError::new(StatusCode::BAD_REQUEST, "invalid_resource_token", detail)
};
let presented = jwt::decode(presented_token)
.ok()
.filter(|p| p.kid() == Some(self.key.thumbprint()) || p.kid().is_none())
.filter(|p| p.verify_signature(&self.key.public_jwk(), Algorithm::Ed25519))
.filter(|p| p.claim_str("iss") == Some(self.config.issuer.as_str()))
.filter(|p| {
[TokenKind::Person.typ(), TokenKind::Auth.typ()]
.contains(&p.typ().unwrap_or_default())
})
.filter(|p| {
p.claims
.get("cnf")
.and_then(|c| c.get("jwk"))
.and_then(|j| serde_json::from_value::<aauth_protocol::Jwk>(j.clone()).ok())
.and_then(|j| j.thumbprint().ok())
.as_deref()
== Some(agent.jkt.as_str())
})
.ok_or_else(|| {
AAuthError::new(
StatusCode::BAD_REQUEST,
"invalid_presented_token",
"not a live token this Person Server issued",
)
})?;
if presented
.claims
.get("exp")
.and_then(Value::as_i64)
.is_none_or(|exp| self.now() >= exp)
{
return Err(AAuthError::new(
StatusCode::BAD_REQUEST,
"expired_presented_token",
"not a live token this Person Server issued",
));
}
let unverified = jwt::decode(resource_token).map_err(|_| invalid("not a JWT"))?;
if unverified.typ() != Some(TokenKind::Resource.typ()) {
return Err(invalid("not a resource token"));
}
let resource = trim(unverified.claim_str("iss").unwrap_or_default()).to_string();
let tenant = self.tenant(named(params))?;
let tenant = tenant.as_str();
let issued_under = presented
.claim_str("tenant")
.map(str::to_string)
.or_else(|| self.default_tenant().map(str::to_string));
if issued_under.as_deref() != Some(tenant) {
return Err(invalid("tenant does not match the presented token"));
}
let mission = match unverified.claim_str("mission_s256") {
Some(s256) => self.live_mission(tenant, agent, Some(s256)).await?,
None => {
return Err(AAuthError::new(
StatusCode::FORBIDDEN,
"access_denied",
"an auth token is issued only within a mission",
))
}
};
if !mission.resources.contains(&resource) {
return Err(AAuthError::new(
StatusCode::FORBIDDEN,
"access_denied",
"this mission does not cover that resource",
));
}
let jwks = self
.resolver
.jwks(&resource, dwk::RESOURCE)
.await
.map_err(|e| invalid(&e.to_string()))?;
let alg = unverified
.alg()
.and_then(Algorithm::from_name)
.ok_or_else(|| invalid("unsupported alg"))?;
let key = match unverified.kid() {
Some(kid) => jwks.keys.iter().find(|k| k.kid.as_deref() == Some(kid)),
None if jwks.keys.len() == 1 => jwks.keys.first(),
None => None,
}
.ok_or_else(|| invalid("no matching key in the resource's JWKS"))?;
if key.alg.as_deref().is_some_and(|a| a != alg.as_str())
|| !unverified.verify_signature(key, alg)
{
return Err(invalid("signature verification failed"));
}
let rt = unverified.claims;
if rt
.get("exp")
.and_then(Value::as_i64)
.is_none_or(|exp| self.now() >= exp)
{
return Err(invalid("resource token has expired"));
}
let not_before = rt.get("nbf").and_then(Value::as_i64);
let issued_at = rt.get("iat").and_then(Value::as_i64);
if not_before.is_some_and(|nbf| self.now() < nbf)
|| issued_at.is_some_and(|iat| iat > self.now() + 60)
{
return Err(invalid("resource token is not yet valid"));
}
if str_of(&rt, "aud") != Some(self.config.issuer.as_str())
|| trim(str_of(&rt, "ps").unwrap_or_default()) != self.config.issuer
{
return Err(invalid("not for this Person Server"));
}
if str_of(&rt, "agent_jkt") != Some(agent.jkt.as_str()) {
return Err(invalid("agent_jkt is not the signer"));
}
let p = &presented.claims;
if p.get("jti") != rt.get("presented_jti")
|| p.get("sub") != rt.get("sub")
|| p.get("mission_s256") != rt.get("mission_s256")
|| p.get("tenant") != rt.get("tenant")
|| str_of(p, "aud") != Some(resource.as_str())
{
return Err(invalid("does not match the presented token"));
}
Ok((rt, mission, resource))
}
async fn proposal_summary(&self, rt: &Map<String, Value>) -> Option<String> {
let (uri, expected) = (str_of(rt, "r3_uri")?, str_of(rt, "r3_s256")?);
let resource = str_of(rt, "iss")?;
if !uri.starts_with("https://") || !aauth_protocol::identifier::same_origin(resource, uri) {
return None;
}
let signed = httpsig::sign(
&self.key,
&SignatureKey::JwksUri {
id: self.config.issuer.clone(),
dwk: dwk::PERSON.into(),
kid: self.key.thumbprint().into(),
},
&httpsig::Outgoing {
method: "GET",
url: uri,
content_type: None,
body: None,
created: self.now(),
},
)
.ok()?;
let mut headers = http::HeaderMap::new();
for (name, value) in signed {
headers.insert(
http::HeaderName::from_bytes(name.as_bytes()).ok()?,
HeaderValue::from_str(&value).ok()?,
);
}
let fetched = self.fetch.get(uri, headers).await.ok()?;
if !fetched.status.is_success() || aauth_protocol::s256(&fetched.body) != expected {
return None;
}
let document: Value = serde_json::from_slice(&fetched.body).ok()?;
document
.get("display")?
.get("summary")?
.as_str()
.map(str::to_string)
}
async fn request_auth_token(
&self,
request: &Request<Bytes>,
path: &str,
) -> Result<Response<Bytes>, AAuthError> {
let agent = self.verify_agent(request, path).await?;
let params = read_body(request.body())?;
let (rt, mission, resource) = self.verify_resource_token(&agent, ¶ms).await?;
let tenant = mission.tenant.clone();
let tenant = tenant.as_str();
let budget = Budget::from_value(rt.get("budget")).map_err(|_| {
AAuthError::new(
StatusCode::BAD_REQUEST,
"invalid_resource_token",
"budget is not a budget object",
)
})?;
let mut grant = Map::new();
grant.insert("aud".into(), json!(resource));
grant.insert("sub".into(), rt.get("sub").cloned().unwrap_or(Value::Null));
for claim in GRANT_CLAIMS {
if let Some(value) = rt.get(*claim) {
grant.insert((*claim).into(), value.clone());
}
}
let Some(budget) = budget else {
return Ok(ok(&self.auth_token(&agent, &mission, &grant).await));
};
grant.insert("budget".into(), json!(budget));
if mission
.budget
.as_ref()
.is_some_and(|b| b.same_scale(&budget))
{
if let Some(spent) = self
.config
.store
.spend(
tenant,
&mission.s256,
budget.amount,
Spend::WithinBudget,
self.now(),
)
.await?
{
return Ok(ok(&self.auth_token(&agent, &spent, &grant).await));
}
}
let summary = self.proposal_summary(&rt).await;
let description = str_of(&rt, "scope").unwrap_or("a request").to_string();
let approval = self.approval(
tenant,
ApprovalKind::AuthToken,
agent,
mission.person.clone(),
description,
);
self.ask(Approval {
mission_s256: Some(mission.s256),
resource: Some(resource),
budget: Some(budget),
grant,
summary,
..approval
})
.await
}
fn approval(
&self,
tenant: &str,
kind: ApprovalKind,
agent: Agent,
person: Person,
description: String,
) -> Approval {
let created = self.now();
Approval {
id: uuid::Uuid::now_v7().to_string(),
kind,
status: ApprovalStatus::Pending,
agent,
approver: person.clone(),
person,
description,
summary: None,
mission_s256: None,
resource: None,
budget: None,
grant: Map::new(),
decided_budget: None,
decided_by: None,
decided_at: None,
created_at: created,
expires_at: created + secs(self.config.approval_ttl),
tenant: tenant.to_string(),
}
}
async fn ask(&self, mut approval: Approval) -> Result<Response<Bytes>, AAuthError> {
if approval.kind != ApprovalKind::Completion {
if let Some(approver) = self
.config
.directory
.approver(approval.kind, &approval.person)
.await
{
approval.approver = approver;
}
}
let tenant = approval.tenant.as_str();
self.config.store.insert_approval(tenant, &approval).await?;
self.config.hooks.on_approval(&approval).await;
Ok(self.still_pending(tenant, &approval.id))
}
fn still_pending(&self, tenant: &str, id: &str) -> Response<Bytes> {
let mut response = json_response(StatusCode::ACCEPTED, &json!({ "status": "pending" }));
let headers = response.headers_mut();
for (name, value) in [
("location", self.endpoints.pending(tenant, id)),
("retry-after", self.config.retry_after.as_secs().to_string()),
("cache-control", "no-store".to_string()),
(
aauth_protocol::requirement::HEADER,
"requirement=approval".to_string(),
),
] {
if let Ok(value) = HeaderValue::from_str(&value) {
headers.insert(name, value);
}
}
response
}
async fn pending_state(
&self,
request: &Request<Bytes>,
path: &str,
tenant: &str,
id: &str,
) -> Result<Response<Bytes>, AAuthError> {
let agent = self.verify_agent(request, path).await?;
let Some(approval) = self
.config
.store
.approval(tenant, id)
.await?
.filter(|a| a.agent.jkt == agent.jkt)
else {
return Ok(problem(
StatusCode::NOT_FOUND,
"not_found",
Some("no such pending request"),
));
};
let collected = || {
problem(
StatusCode::GONE,
"collected",
Some("this approval's token was already collected"),
)
};
if request.method() == Method::DELETE {
if approval.status == ApprovalStatus::Pending {
let withdrawn = Approval {
status: ApprovalStatus::Denied,
decided_by: Some("agent".into()),
decided_at: Some(self.now()),
..approval
};
self.config
.store
.transition_approval(tenant, ApprovalStatus::Pending, &withdrawn)
.await?;
}
let mut response = Response::new(Bytes::new());
*response.status_mut() = StatusCode::NO_CONTENT;
return Ok(response);
}
match approval.status {
ApprovalStatus::Denied => {
let who = approval.approver.name.as_deref().unwrap_or("The approver");
return Ok(problem(
StatusCode::FORBIDDEN,
"denied",
Some(&format!("{who} declined")),
));
}
ApprovalStatus::Pending if self.now() >= approval.expires_at => {
return Ok(problem(
StatusCode::REQUEST_TIMEOUT,
"expired",
Some("nobody answered in time"),
));
}
ApprovalStatus::Pending => return Ok(self.still_pending(tenant, id)),
ApprovalStatus::Collected => return Ok(collected()),
ApprovalStatus::Approved => {}
}
if approval.kind == ApprovalKind::Completion {
return Ok(ok(&json!({})));
}
let mission = self
.live_mission(tenant, &agent, approval.mission_s256.as_deref())
.await?;
if approval.kind == ApprovalKind::Mission {
let mut tokens = Map::new();
for resource in &mission.resources {
let issued = self
.person_token(&agent, &mission.person, resource, Some(&mission))
.await;
tokens.insert(resource.clone(), issued["person_token"].clone());
}
return Ok(ok(&json!({
"s256": mission.s256,
"mission": URL_SAFE_NO_PAD.encode(mission.mission.as_bytes()),
"person_tokens": tokens,
})));
}
let grant = approval.grant.clone();
let asked = approval.budget.clone();
let collecting = Approval {
status: ApprovalStatus::Collected,
..approval.clone()
};
if !self
.config
.store
.transition_approval(tenant, ApprovalStatus::Approved, &collecting)
.await?
{
return Ok(collected());
}
let mut mission = mission;
if let Some(asked) =
asked.filter(|a| mission.budget.as_ref().is_some_and(|b| b.same_scale(a)))
{
let spent = self
.config
.store
.spend(
tenant,
&mission.s256,
asked.amount,
Spend::Approved,
self.now(),
)
.await;
match spent {
Ok(Some(spent)) => mission = spent,
unspent => {
let _ = self
.config
.store
.transition_approval(tenant, ApprovalStatus::Collected, &approval)
.await;
unspent?;
return Err(AAuthError::mission_terminated("the mission has ended"));
}
}
}
Ok(ok(&self.auth_token(&agent, &mission, &grant).await))
}
pub async fn decide(
&self,
tenant: &str,
id: &str,
decision: Decision,
) -> Result<Approval, AAuthError> {
let Some(approval) = self.config.store.approval(tenant, id).await? else {
return Err(AAuthError::new(
StatusCode::NOT_FOUND,
"not_found",
"no such approval",
));
};
let already =
|| AAuthError::new(StatusCode::CONFLICT, "invalid_request", "already decided");
if approval.status != ApprovalStatus::Pending {
return Err(already());
}
let now = self.now();
if now >= approval.expires_at {
return Err(AAuthError::new(
StatusCode::GONE,
"expired",
"this approval has expired",
));
}
if decision.budget.is_some() && approval.kind != ApprovalKind::Mission {
return Err(AAuthError::invalid_request("only a mission takes a budget"));
}
let approved = decision.status == Verdict::Approved;
let mut decided = Approval {
status: if approved {
ApprovalStatus::Approved
} else {
ApprovalStatus::Denied
},
decided_budget: decision.budget.clone(),
decided_by: decision.decided_by,
decided_at: Some(now),
..approval.clone()
};
let created = (approved && approval.kind == ApprovalKind::Mission)
.then(|| self.mission_from(&approval, decision.budget, now));
if let Some(mission) = &created {
decided.mission_s256 = Some(mission.s256.clone());
} else if approved {
let mission = match &approval.mission_s256 {
Some(s256) => self.config.store.mission(tenant, s256).await?,
None => None,
};
if !mission.is_some_and(|m| m.is_live(now)) {
return Err(AAuthError::new(
StatusCode::CONFLICT,
"mission_terminated",
"the mission has ended",
));
}
}
if let Some(mission) = &created {
if !self.config.store.insert_mission(tenant, mission).await? {
return Err(already());
}
}
if !self
.config
.store
.transition_approval(tenant, ApprovalStatus::Pending, &decided)
.await?
{
if let Some(mission) = &created {
self.config
.store
.end_mission(tenant, &mission.s256, now)
.await?;
}
return Err(already());
}
if let (true, ApprovalKind::Completion, Some(s256)) =
(approved, approval.kind, &approval.mission_s256)
{
self.config.store.end_mission(tenant, s256, now).await?;
}
self.config.hooks.on_decision(&decided).await;
Ok(decided)
}
fn mission_from(&self, approval: &Approval, budget: Option<Budget>, now: i64) -> Mission {
let resources: Vec<String> = approval
.grant
.get("resources")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter_map(|r| r.as_str().map(str::to_string))
.collect();
let expires_at = now + secs(self.config.mission_ttl);
let mut document = json!({
"approval": approval.id,
"agent": approval.agent.id,
"approved_at": rfc3339(now),
"expires_at": rfc3339(expires_at),
"description": approval.description,
"approved_resources": resources,
});
if let Some(budget) = &budget {
document["budget"] = json!(budget);
}
let bytes = document.to_string();
Mission {
s256: aauth_protocol::s256(&bytes),
mission: bytes,
agent_id: approval.agent.id.clone(),
person: approval.person.clone(),
resources,
budget,
spent: 0,
log: Vec::new(),
expires_at,
completed_at: None,
tenant: approval.tenant.clone(),
}
}
pub async fn terminate(&self, tenant: &str, s256: &str) -> Result<Mission, AAuthError> {
self.config
.store
.end_mission(tenant, s256, self.now())
.await?
.ok_or_else(AAuthError::mission_not_found)
}
fn admitted(&self, request: &Request<Bytes>) -> Result<(), AAuthError> {
let presented = request
.headers()
.get("authorization")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "));
match (&self.config.admin_token, presented) {
(Some(token), Some(presented))
if aauth_protocol::constant_time_eq(token, presented) =>
{
Ok(())
}
_ => Err(AAuthError::new(
StatusCode::UNAUTHORIZED,
"unauthorized",
"the management routes need the admin token",
)),
}
}
async fn manage(
&self,
request: &Request<Bytes>,
segments: &[&str],
) -> Result<Option<Response<Bytes>>, AAuthError> {
if !matches!(
segments,
["consent"] | ["consent", _, "decision"] | ["missions"] | ["missions", _]
) {
return Ok(None);
}
self.admitted(request)?;
let query = |name: &str| {
request.uri().query().and_then(|q| {
url::form_urlencoded::parse(q.as_bytes())
.find(|(k, _)| k == name)
.map(|(_, v)| v.into_owned())
})
};
let tenant = self.tenant(query("tenant").filter(|t| !t.is_empty()).as_deref())?;
let tenant = tenant.as_str();
let response = match (request.method(), segments) {
(&Method::GET, ["consent"]) => {
let status = query("status");
let approvals: Vec<Approval> = self
.config
.store
.approvals(tenant)
.await?
.into_iter()
.filter(|a| {
status.as_deref().is_none_or(|s| {
serde_json::to_value(a.status)
.ok()
.and_then(|v| v.as_str().map(|v| v == s))
== Some(true)
})
})
.collect();
ok(&approvals)
}
(&Method::POST, ["consent", id, "decision"]) => {
let params = read_body(request.body())?;
let verdict = str_of(¶ms, "decision").or_else(|| str_of(¶ms, "status"));
let status = match verdict {
Some("approve" | "approved") => Verdict::Approved,
Some("deny" | "denied") => Verdict::Denied,
_ => return Err(AAuthError::invalid_request("decision is approve or deny")),
};
let budget = Budget::from_value(params.get("budget"))
.map_err(|e| AAuthError::invalid_request(e.to_string()))?;
let decision = Decision {
status,
budget,
decided_by: str_of(¶ms, "decided_by").map(str::to_string),
};
ok(&self.decide(tenant, id, decision).await?)
}
(&Method::GET, ["missions"]) => ok(&self.config.store.missions(tenant).await?),
(&Method::GET, ["missions", s256]) => {
match self.config.store.mission(tenant, s256).await? {
Some(mission) => ok(&mission),
None => json_response(StatusCode::NOT_FOUND, &Value::Null),
}
}
(&Method::PATCH, ["missions", s256]) => ok(&self.terminate(tenant, s256).await?),
_ => problem(StatusCode::METHOD_NOT_ALLOWED, "method_not_allowed", None),
};
Ok(Some(response))
}
}
#[cfg(feature = "reqwest")]
fn default_fetch() -> aauth_protocol::Result<Arc<dyn Fetch>> {
use aauth_protocol::client::DeniedNetworks;
Ok(Arc::new(aauth_protocol::discovery::ReqwestFetch::guarded(
DeniedNetworks::default(),
)?))
}
#[cfg(not(feature = "reqwest"))]
fn default_fetch() -> aauth_protocol::Result<Arc<dyn Fetch>> {
Err(aauth_protocol::Error::Invalid(
"without the reqwest feature, give the Person Server a Fetch".into(),
))
}