grain.social/grain
photography-focused photo sharing platform
grain.social
1 collaborator
- 1 collaborator
0
2
Core logs in with OAuth, keeps sessions and checks service auth
Only what this pull adds. It is built on grain runs on a Rust server modelled on knotless, without hatk.
crates/grain/src/auth.rsnew+212−0
| @@ -0,0 +1,212 @@ | ||
//! Service auth: short-lived JWTs an account's PDS signs with the account's | ||
//! key, naming the service they are for (`aud`) and the one method they | ||
//! allow (`lxm`). The mobile apps and the command line send grain-serve | ||
//! these, and grain asks a PDS for them to call another service, such as a | ||
//! group's host, as the account. | ||
| ||
use crate::crypto::{self, PublicKey}; | ||
use crate::env::Ctx; | ||
use crate::identity; | ||
use crate::session::Session; | ||
use crate::xrpc::Xrpc; | ||
use anyhow::{Result, bail}; | ||
use serde::Deserialize; | ||
| ||
/// A token from `me`'s PDS for calling `lxm` on the service `aud`. | ||
pub async fn service_token(ctx: &Ctx, me: &Session, aud: &str, lxm: &str) -> Result<String> { | ||
token(ctx, me, &[("aud", aud), ("lxm", lxm)]).await | ||
} | ||
| ||
/// [`service_token`], good until `exp`, in seconds since the epoch: an | ||
/// hour from now at most. | ||
pub async fn lasting_token( | ||
ctx: &Ctx, | ||
me: &Session, | ||
aud: &str, | ||
lxm: &str, | ||
exp: u64, | ||
) -> Result<String> { | ||
let exp = exp.to_string(); | ||
token(ctx, me, &[("aud", aud), ("lxm", lxm), ("exp", &exp)]).await | ||
} | ||
| ||
async fn token(ctx: &Ctx, me: &Session, params: &[(&str, &str)]) -> Result<String> { | ||
#[derive(Deserialize)] | ||
struct Out { | ||
token: String, | ||
} | ||
let out: Out = Xrpc::new(ctx.http.clone(), &me.service) | ||
.query( | ||
"com.atproto.server.getServiceAuth", | ||
params, | ||
Some(&me.auth()), | ||
) | ||
.await?; | ||
Ok(out.token) | ||
} | ||
| ||
#[derive(Deserialize)] | ||
struct Claims { | ||
iss: String, | ||
aud: String, | ||
lxm: Option<String>, | ||
} | ||
| ||
/// Checks a service token for calling `lxm` on `aud`, and returns the DID | ||
/// that sent it. The signature is checked against the key the issuer's DID | ||
/// document names ([`signed_by`]), and `exp` against the clock. | ||
pub async fn verify(ctx: &Ctx, token: &str, aud: &str, lxm: &str) -> Result<String> { | ||
let (_, claims) = crypto::jwt_parts(token)?; | ||
let claims: Claims = serde_json::from_value(claims)?; | ||
let did = claims.iss.split('#').next().unwrap_or_default().to_string(); | ||
if claims.aud != aud { | ||
bail!("a token for {}, not {aud}", claims.aud); | ||
} | ||
if claims.lxm.as_deref() != Some(lxm) { | ||
bail!("a token for {:?}, not {lxm}", claims.lxm); | ||
} | ||
signed_by(ctx, token, &did).await?; | ||
Ok(did) | ||
} | ||
| ||
/// Checks `token`'s signature against the key `did`'s DID document names: | ||
/// as read a few minutes ago, and as it is now when that one turns it | ||
/// away, as it does once the key rotated. | ||
async fn signed_by(ctx: &Ctx, token: &str, did: &str) -> Result<()> { | ||
let kept = identity::signing_key(&identity::kept_document(ctx, did).await?)?; | ||
if crypto::verify_jwt::<serde_json::Value>(token, &kept).is_ok() { | ||
return Ok(()); | ||
} | ||
let key: PublicKey = identity::signing_key(&identity::document(ctx, did).await?)?; | ||
crypto::verify_jwt::<serde_json::Value>(token, &key)?; | ||
Ok(()) | ||
} | ||
| ||
#[cfg(test)] | ||
mod tests { | ||
use super::*; | ||
use crate::crypto::P256Key; | ||
use crate::env::Env; | ||
use crate::xrpc::tests::{read_request, serving}; | ||
use serde_json::{Value, json}; | ||
use std::sync::{Arc, Mutex}; | ||
| ||
#[tokio::test] | ||
async fn tokens_are_asked_of_the_accounts_pds() { | ||
let (pds, asked) = serving("200 OK", "", r#"{"token":"signed"}"#); | ||
let ctx = Ctx::new(Env::from_vars(|_| None)); | ||
let me = Session { | ||
did: "did:plc:alice".into(), | ||
handle: "alice.test".into(), | ||
service: pds, | ||
access_jwt: "access".into(), | ||
refresh_jwt: "refresh".into(), | ||
oauth: None, | ||
}; | ||
let aud = "did:web:groups.test"; | ||
let lxm = "fyi.opensocial.requestJoin"; | ||
assert_eq!(service_token(&ctx, &me, aud, lxm).await.unwrap(), "signed"); | ||
let exp = 1_700_000_000; | ||
assert_eq!( | ||
lasting_token(&ctx, &me, aud, lxm, exp).await.unwrap(), | ||
"signed" | ||
); | ||
let asked = asked.lock().unwrap(); | ||
let first = "GET /xrpc/com.atproto.server.getServiceAuth?aud=did%3Aweb%3Agroups.test&lxm=fyi.opensocial.requestJoin "; | ||
assert!(asked[0].starts_with(first), "{}", asked[0]); | ||
assert!(asked[0].contains("authorization: Bearer access")); | ||
assert!(asked[1].contains("&exp=1700000000 "), "{}", asked[1]); | ||
} | ||
| ||
/// A server that answers every request with 200 and whatever `body` | ||
/// holds when it is asked: a DID document that can change. | ||
fn documents() -> (u16, Arc<Mutex<String>>) { | ||
use std::io::Write; | ||
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); | ||
let port = listener.local_addr().unwrap().port(); | ||
let body = Arc::new(Mutex::new(String::new())); | ||
let held = body.clone(); | ||
std::thread::spawn(move || { | ||
for stream in listener.incoming() { | ||
let mut stream = stream.unwrap(); | ||
read_request(&mut stream); | ||
let body = held.lock().unwrap().clone(); | ||
let answer = format!( | ||
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", | ||
body.len() | ||
); | ||
let _ = stream.write_all(answer.as_bytes()); | ||
} | ||
}); | ||
(port, body) | ||
} | ||
| ||
fn document(did: &str, key: &PublicKey) -> String { | ||
json!({ | ||
"id": did, | ||
"verificationMethod": [{ | ||
"id": format!("{did}#atproto"), | ||
"type": "Multikey", | ||
"controller": did, | ||
"publicKeyMultibase": key.to_multibase(), | ||
}], | ||
}) | ||
.to_string() | ||
} | ||
| ||
#[tokio::test] | ||
async fn tokens_are_verified_against_the_issuers_key() { | ||
let (port, served) = documents(); | ||
let did = format!("did:web:127.0.0.1%3A{port}"); | ||
let ctx = Ctx::new(Env::from_vars(|k| { | ||
(k == "GRAIN_DID_WEB_SCHEME").then(|| "http".into()) | ||
})); | ||
let key = P256Key::generate(); | ||
*served.lock().unwrap() = document(&did, &key.public()); | ||
let aud = "did:web:grain.test"; | ||
let lxm = "social.grain.feed.getTimeline"; | ||
let exp = crate::keys::now_secs() + 60; | ||
let sign = |key: &P256Key, claims: Value| key.sign_jwt(&json!({"typ": "JWT"}), &claims); | ||
let claims = |iss: &str| json!({"iss": iss, "aud": aud, "lxm": lxm, "exp": exp}); | ||
| ||
let token = sign(&key, claims(&did)); | ||
assert_eq!(verify(&ctx, &token, aud, lxm).await.unwrap(), did); | ||
// An issuer naming one of the account's services is the account. | ||
let service = sign(&key, claims(&format!("{did}#atproto_labeler"))); | ||
assert_eq!(verify(&ctx, &service, aud, lxm).await.unwrap(), did); | ||
| ||
let why = verify(&ctx, &token, "did:web:other.test", lxm).await; | ||
assert!( | ||
why.unwrap_err() | ||
.to_string() | ||
.starts_with("a token for did:web:grain.test") | ||
); | ||
let why = verify(&ctx, &token, aud, "social.grain.other").await; | ||
assert!( | ||
why.unwrap_err() | ||
.to_string() | ||
.contains("not social.grain.other") | ||
); | ||
let any = sign(&key, json!({"iss": did, "aud": aud, "exp": exp})); | ||
assert!( | ||
verify(&ctx, &any, aud, lxm).await.is_err(), | ||
"no lxm, no method" | ||
); | ||
let expired = sign(&key, json!({"iss": did, "aud": aud, "lxm": lxm, "exp": 1})); | ||
assert!(verify(&ctx, &expired, aud, lxm).await.is_err()); | ||
let stranger = sign(&P256Key::generate(), claims(&did)); | ||
assert!(verify(&ctx, &stranger, aud, lxm).await.is_err()); | ||
assert!(verify(&ctx, "not.a.token", aud, lxm).await.is_err()); | ||
| ||
// The account's key rotates: a token signed with the new key turns | ||
// the document kept away, and is checked against it read again. | ||
let rotated = P256Key::generate(); | ||
*served.lock().unwrap() = document(&did, &rotated.public()); | ||
let token = sign(&rotated, claims(&did)); | ||
assert_eq!(verify(&ctx, &token, aud, lxm).await.unwrap(), did); | ||
| ||
// An issuer whose document cannot be read is no one's. | ||
let nowhere = sign(&key, claims("did:web:127.0.0.1%3A1")); | ||
assert!(verify(&ctx, &nowhere, aud, lxm).await.is_err()); | ||
} | ||
} | ||
crates/grain/src/env.rs+28−0
| @@ -14,6 +14,8 @@ pub enum HandleResolver { | ||
| ||
#[derive(Clone, Debug)] | ||
pub struct Env { | ||
/// Where sessions are kept, and the account this machine acts as. | ||
pub config_dir: PathBuf, | ||
pub cache_dir: PathBuf, | ||
pub plc_url: String, | ||
pub handle_resolvers: Vec<HandleResolver>, | ||
| @@ -24,6 +26,13 @@ pub struct Env { | ||
pub ca_certs: Option<PathBuf>, | ||
/// Skips DID resolution: every identity is served by this PDS. | ||
pub service: Option<String>, | ||
/// The account to act as, when not the one last logged in. | ||
pub identifier: Option<String>, | ||
/// An app password for [`Env::identifier`], to log in with when no | ||
/// session is kept. | ||
pub password: Option<String>, | ||
/// The command an OAuth login opens its page with, the URL added last. | ||
pub browser: Option<String>, | ||
pub identity_ttl: Duration, | ||
pub timeout: Duration, | ||
} | ||
| @@ -69,6 +78,7 @@ impl Env { | ||
], | ||
}; | ||
Env { | ||
config_dir: dir("GRAIN_CONFIG_DIR", "XDG_CONFIG_HOME", ".config"), | ||
cache_dir: dir("GRAIN_CACHE_DIR", "XDG_CACHE_HOME", ".cache"), | ||
plc_url: get("GRAIN_PLC_URL").unwrap_or_else(|| "https://plc.directory".into()), | ||
handle_resolvers, | ||
| @@ -77,6 +87,9 @@ impl Env { | ||
.filter(|v| !v.is_empty()) | ||
.map(PathBuf::from), | ||
service: get("GRAIN_SERVICE").filter(|v| !v.is_empty()), | ||
identifier: get("GRAIN_IDENTIFIER").filter(|v| !v.is_empty()), | ||
password: get("GRAIN_PASSWORD").filter(|v| !v.is_empty()), | ||
browser: get("GRAIN_BROWSER").filter(|v| !v.is_empty()), | ||
identity_ttl: Duration::from_secs(num("GRAIN_IDENTITY_TTL_SECS", 3600)), | ||
timeout: Duration::from_secs(num("GRAIN_TIMEOUT_SECS", 60)), | ||
} | ||
| @@ -141,6 +154,7 @@ mod tests { | ||
#[test] | ||
fn defaults() { | ||
let e = env(&[("HOME", "/home/a")]); | ||
assert_eq!(e.config_dir, PathBuf::from("/home/a/.config/grain")); | ||
assert_eq!(e.cache_dir, PathBuf::from("/home/a/.cache/grain")); | ||
assert_eq!(e.plc_url, "https://plc.directory"); | ||
assert_eq!(e.did_web_scheme, "https"); | ||
| @@ -159,14 +173,20 @@ mod tests { | ||
assert_eq!(e.timeout, Duration::from_secs(60)); | ||
assert!(e.service.is_none()); | ||
assert!(e.ca_certs.is_none()); | ||
assert!(e.identifier.is_none() && e.password.is_none() && e.browser.is_none()); | ||
let bare = env(&[]); | ||
assert_eq!(bare.cache_dir, PathBuf::from("./.cache/grain")); | ||
assert_eq!(bare.config_dir, PathBuf::from("./.config/grain")); | ||
} | ||
| ||
#[test] | ||
fn overrides() { | ||
let e = env(&[ | ||
("XDG_CACHE_HOME", "/x"), | ||
("XDG_CONFIG_HOME", "/y"), | ||
("GRAIN_IDENTIFIER", "alice.test"), | ||
("GRAIN_PASSWORD", "hunter2"), | ||
("GRAIN_BROWSER", "firefox --new-tab"), | ||
("GRAIN_HANDLE_RESOLVER", "well-known:http,http://pds,"), | ||
("GRAIN_PLC_URL", "http://plc"), | ||
("GRAIN_DID_WEB_SCHEME", "http"), | ||
| @@ -175,6 +195,10 @@ mod tests { | ||
("GRAIN_TIMEOUT_SECS", "soon"), | ||
]); | ||
assert_eq!(e.cache_dir, PathBuf::from("/x/grain")); | ||
assert_eq!(e.config_dir, PathBuf::from("/y/grain")); | ||
assert_eq!(e.identifier.as_deref(), Some("alice.test")); | ||
assert_eq!(e.password.as_deref(), Some("hunter2")); | ||
assert_eq!(e.browser.as_deref(), Some("firefox --new-tab")); | ||
assert_eq!(e.plc_url, "http://plc"); | ||
assert_eq!(e.did_web_scheme, "http"); | ||
assert_eq!(e.service.as_deref(), Some("http://pds")); | ||
| @@ -196,6 +220,10 @@ mod tests { | ||
PathBuf::from("/c") | ||
); | ||
assert!(env(&[("GRAIN_SERVICE", "")]).service.is_none()); | ||
assert_eq!( | ||
env(&[("GRAIN_CONFIG_DIR", "/k")]).config_dir, | ||
PathBuf::from("/k") | ||
); | ||
let _ = Ctx::new(e); | ||
} | ||
| ||
crates/grain/src/lib.rs+3−0
| @@ -4,12 +4,15 @@ | ||
//! Its modules start from knotless's (`crates/knotless/src` there), taken | ||
//! over as grain needs them rather than all at once. | ||
| ||
pub mod auth; | ||
pub mod base32; | ||
pub mod crypto; | ||
pub mod drisl; | ||
pub mod env; | ||
pub mod identity; | ||
pub mod keys; | ||
pub mod oauth; | ||
pub mod session; | ||
pub mod store; | ||
pub mod telemetry; | ||
pub mod xrpc; | ||
crates/grain/src/oauth.rsnew+1239−0
1239 lines changed: not loaded until asked for.
crates/grain/src/session.rsnew+370−0
| @@ -0,0 +1,370 @@ | ||
//! Sessions, from an OAuth login or an app password: what lets calls to an | ||
//! account's PDS in, and, for the command line, kept per DID under the | ||
//! config directory and refreshed in place. | ||
| ||
use crate::crypto::P256Key; | ||
use crate::env::Ctx; | ||
use crate::identity::{self, Identity}; | ||
use crate::oauth; | ||
use crate::store::{self, Lock}; | ||
use crate::xrpc::{Auth, Xrpc}; | ||
use anyhow::{Result, bail}; | ||
use base64::Engine; | ||
use base64::engine::general_purpose::STANDARD; | ||
use serde::{Deserialize, Serialize}; | ||
use serde_json::json; | ||
use std::path::PathBuf; | ||
use std::sync::Arc; | ||
| ||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] | ||
#[serde(rename_all = "camelCase")] | ||
pub struct Session { | ||
pub did: String, | ||
pub handle: String, | ||
#[serde(default)] | ||
pub service: String, | ||
pub access_jwt: String, | ||
pub refresh_jwt: String, | ||
/// Set for an OAuth session, whose tokens are bound to a DPoP key. | ||
#[serde(default, skip_serializing_if = "Option::is_none")] | ||
pub oauth: Option<OAuth>, | ||
} | ||
| ||
/// What an OAuth session needs beyond its tokens. | ||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] | ||
#[serde(rename_all = "camelCase")] | ||
pub struct OAuth { | ||
/// The DPoP key's secret scalar, base64. | ||
pub key: String, | ||
/// The authorization server, which a confidential client's assertions | ||
/// are addressed to. | ||
pub issuer: String, | ||
pub token_endpoint: String, | ||
pub revocation_endpoint: Option<String>, | ||
/// The confidential client the session was granted to, grain-serve's; | ||
/// none for the command line's loopback client. | ||
pub client_id: Option<String>, | ||
/// The loopback client ID the session was granted to, its scope and | ||
/// all: a grain that asks for more scope since still refreshes it | ||
/// under the ID it was granted to, which the server holds it to. None | ||
/// for a confidential client's. | ||
#[serde(default, skip_serializing_if = "Option::is_none")] | ||
pub loopback_id: Option<String>, | ||
/// The scope the session was asked for, which its server may have | ||
/// granted less of: one asked for less than grain asks for now is | ||
/// behind ([`oauth::behind`]), and logging in again brings it up. | ||
#[serde(default, skip_serializing_if = "Option::is_none")] | ||
pub scope: Option<String>, | ||
/// When the access token expires, in seconds since the epoch. | ||
pub expires_at: u64, | ||
} | ||
| ||
impl OAuth { | ||
pub fn key(&self) -> Result<P256Key> { | ||
P256Key::from_bytes(&STANDARD.decode(&self.key)?) | ||
} | ||
} | ||
| ||
impl Session { | ||
/// What lets this session's requests in. | ||
pub fn auth(&self) -> Auth { | ||
match self.oauth.as_ref().map(OAuth::key) { | ||
Some(Ok(key)) => Auth::Dpop { | ||
token: self.access_jwt.clone(), | ||
key: Arc::new(key), | ||
}, | ||
_ => Auth::bearer(&self.access_jwt), | ||
} | ||
} | ||
| ||
/// Whether the access token expires within a minute. | ||
pub fn expiring(&self) -> bool { | ||
match &self.oauth { | ||
Some(oauth) => oauth.expires_at < crate::keys::now_secs() + 60, | ||
None => expiring(&self.access_jwt), | ||
} | ||
} | ||
} | ||
| ||
fn path(ctx: &Ctx, did: &str) -> PathBuf { | ||
ctx.env | ||
.config_dir | ||
.join("sessions") | ||
.join(format!("{}.json", did.replace(':', "_"))) | ||
} | ||
| ||
pub fn load(ctx: &Ctx, did: &str) -> Option<Session> { | ||
store::read(&path(ctx, did)) | ||
} | ||
| ||
/// Forgets `did`'s session without a word to its PDS: for an account gone | ||
/// from it. | ||
pub fn remove(ctx: &Ctx, did: &str) { | ||
let _ = std::fs::remove_file(path(ctx, did)); | ||
} | ||
| ||
pub fn save(ctx: &Ctx, session: &Session) -> Result<()> { | ||
store::write(&path(ctx, &session.did), session, true) | ||
} | ||
| ||
pub async fn login(ctx: &Ctx, identifier: &str, password: &str) -> Result<Session> { | ||
let session = create(ctx, identifier, password).await?; | ||
save(ctx, &session)?; | ||
Ok(session) | ||
} | ||
| ||
/// A session for `identifier` from its password, kept nowhere. | ||
pub async fn create(ctx: &Ctx, identifier: &str, password: &str) -> Result<Session> { | ||
let id = identity::resolve(ctx, identifier).await?; | ||
let mut session: Session = Xrpc::new(ctx.http.clone(), &id.pds) | ||
.procedure( | ||
"com.atproto.server.createSession", | ||
Some(&json!({ "identifier": identifier, "password": password })), | ||
None, | ||
) | ||
.await?; | ||
session.service = id.pds; | ||
Ok(session) | ||
} | ||
| ||
/// Removes the stored session, revoking an OAuth one's tokens. Returns | ||
/// whether there was one. | ||
pub async fn logout(ctx: &Ctx, identifier: &str) -> Result<bool> { | ||
let id = identity::resolve(ctx, identifier).await?; | ||
if let Some(session) = load(ctx, &id.did) | ||
&& let Some(oauth) = &session.oauth | ||
&& let Ok(client) = oauth::Client::granted(oauth) | ||
{ | ||
// A token the server cannot revoke now expires on its own. | ||
let _ = oauth::revoke(ctx, &client, &session, oauth).await; | ||
} | ||
Ok(std::fs::remove_file(path(ctx, &id.did)).is_ok()) | ||
} | ||
| ||
/// A session for `id` with an access token that will last the call: the | ||
/// stored one, refreshed if it is about to expire, or a fresh login from | ||
/// `GRAIN_PASSWORD`. | ||
pub async fn ensure(ctx: &Ctx, id: &Identity) -> Result<Session> { | ||
if let Some(stored) = load(ctx, &id.did) { | ||
if !stored.expiring() { | ||
return Ok(stored); | ||
} | ||
if let Ok(fresh) = refresh(ctx, &stored).await { | ||
return Ok(fresh); | ||
} | ||
} | ||
// The password is the identifier's: for another account it would log | ||
// in as the identifier, and save that over the identifier's session, | ||
// an OAuth one among them. | ||
if let Some(password) = &ctx.env.password { | ||
let identifier = ctx.env.identifier.clone().unwrap_or_else(|| id.did.clone()); | ||
let whose = identity::resolve(ctx, &identifier).await?; | ||
if whose.did == id.did { | ||
return login(ctx, &identifier, password).await; | ||
} | ||
} | ||
bail!("not logged in as {}: log in, or set GRAIN_PASSWORD", id.did) | ||
} | ||
| ||
/// Marks `session` as the account this machine last logged in as. | ||
pub fn remember(ctx: &Ctx, session: &Session) -> Result<()> { | ||
store::write_bytes( | ||
&ctx.env.config_dir.join("me"), | ||
session.did.as_bytes(), | ||
false, | ||
) | ||
} | ||
| ||
/// The account this machine acts as: `GRAIN_IDENTIFIER`, else the one it | ||
/// last logged in as. | ||
pub async fn me(ctx: &Ctx) -> Result<Session> { | ||
let identifier = match &ctx.env.identifier { | ||
Some(identifier) => identifier.clone(), | ||
None => std::fs::read_to_string(ctx.env.config_dir.join("me")) | ||
.map_err(|_| anyhow::anyhow!("not logged in"))?, | ||
}; | ||
let id = identity::resolve(ctx, identifier.trim()).await?; | ||
ensure(ctx, &id).await | ||
} | ||
| ||
/// `stored` with new tokens, kept in its place. Refresh tokens are used | ||
/// once, so processes refresh one at a time, and one that finds the | ||
/// session already refreshed takes that. | ||
async fn refresh(ctx: &Ctx, stored: &Session) -> Result<Session> { | ||
let _lock = Lock::acquire(path(ctx, &stored.did).with_extension("lock")).await?; | ||
if let Some(now) = load(ctx, &stored.did) | ||
&& now.refresh_jwt != stored.refresh_jwt | ||
{ | ||
return Ok(now); | ||
} | ||
let session = match &stored.oauth { | ||
Some(oauth) => { | ||
let client = oauth::Client::granted(oauth)?; | ||
oauth::refresh(ctx, &client, stored, oauth).await? | ||
} | ||
None => { | ||
let mut session: Session = Xrpc::new(ctx.http.clone(), &stored.service) | ||
.procedure( | ||
"com.atproto.server.refreshSession", | ||
None, | ||
Some(&Auth::bearer(&stored.refresh_jwt)), | ||
) | ||
.await?; | ||
session.service = stored.service.clone(); | ||
session | ||
} | ||
}; | ||
save(ctx, &session)?; | ||
Ok(session) | ||
} | ||
| ||
/// Whether a JWT expires within a minute, or cannot be read. | ||
pub fn expiring(jwt: &str) -> bool { | ||
#[derive(Deserialize)] | ||
struct Claims { | ||
exp: u64, | ||
} | ||
let claims = jwt | ||
.split('.') | ||
.nth(1) | ||
.and_then(|p| { | ||
base64::engine::general_purpose::URL_SAFE_NO_PAD | ||
.decode(p) | ||
.ok() | ||
}) | ||
.and_then(|b| serde_json::from_slice::<Claims>(&b).ok()); | ||
match claims { | ||
Some(c) => c.exp < crate::keys::now_secs() + 60, | ||
None => true, | ||
} | ||
} | ||
| ||
#[cfg(test)] | ||
mod tests { | ||
use super::*; | ||
use crate::oauth::tests::{authorization_server, ctx_on}; | ||
| ||
fn jwt(exp: u64) -> String { | ||
let body = | ||
base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(format!("{{\"exp\":{exp}}}")); | ||
format!("h.{body}.s") | ||
} | ||
| ||
#[test] | ||
fn expiry() { | ||
let now = crate::keys::now_secs(); | ||
assert!(!expiring(&jwt(now + 3600))); | ||
assert!(expiring(&jwt(now + 10))); | ||
assert!(expiring("garbage")); | ||
assert!(expiring("a.!!.b")); | ||
} | ||
| ||
/// An OAuth session for `did:plc:alice` on `pds`, logged in at its | ||
/// stand-in server, whose access token expires at `expires_at`. | ||
fn oauth_session(pds: &str, expires_at: u64) -> Session { | ||
Session { | ||
did: "did:plc:alice".into(), | ||
handle: "alice.test".into(), | ||
service: pds.into(), | ||
access_jwt: "old".into(), | ||
refresh_jwt: "refresh-old".into(), | ||
oauth: Some(OAuth { | ||
key: STANDARD.encode(P256Key::generate().to_bytes()), | ||
issuer: pds.into(), | ||
token_endpoint: format!("{pds}/oauth/token"), | ||
revocation_endpoint: Some(format!("{pds}/oauth/revoke")), | ||
client_id: None, | ||
loopback_id: Some(oauth::client_id()), | ||
scope: Some(oauth::SCOPE.into()), | ||
expires_at, | ||
}), | ||
} | ||
} | ||
| ||
#[test] | ||
fn auth_and_expiry_of_sessions() { | ||
let now = crate::keys::now_secs(); | ||
let session = oauth_session("http://pds.test", now + 3600); | ||
assert!(matches!(session.auth(), Auth::Dpop { .. })); | ||
assert!(!session.expiring()); | ||
assert!(oauth_session("http://pds.test", now).expiring()); | ||
let password = Session { | ||
oauth: None, | ||
access_jwt: jwt(now + 3600), | ||
..session.clone() | ||
}; | ||
assert!(matches!(password.auth(), Auth::Bearer(_))); | ||
assert!(!password.expiring()); | ||
// A key that cannot be read leaves only the bare token. | ||
let mut broken = session.clone(); | ||
broken.oauth.as_mut().unwrap().key = "!!".into(); | ||
assert!(matches!(broken.auth(), Auth::Bearer(_))); | ||
assert!(broken.oauth.unwrap().key().is_err()); | ||
| ||
// Kept as camelCase, as knotless's and hatk's sessions are. | ||
let kept = serde_json::to_value(&session).unwrap(); | ||
assert_eq!(kept["accessJwt"], "old"); | ||
assert!(kept["oauth"]["expiresAt"].is_u64()); | ||
assert!(kept["oauth"].get("loopbackId").is_some()); | ||
assert!( | ||
serde_json::to_value(&password) | ||
.unwrap() | ||
.get("oauth") | ||
.is_none() | ||
); | ||
} | ||
| ||
#[tokio::test] | ||
async fn kept_refreshed_and_forgotten() { | ||
let (base, seen) = authorization_server().await; | ||
let dir = tempfile::tempdir().unwrap(); | ||
let ctx = ctx_on(&base, dir.path()); | ||
let alice = identity::resolve(&ctx, "did:plc:alice").await.unwrap(); | ||
assert!(load(&ctx, &alice.did).is_none()); | ||
assert!(me(&ctx).await.is_err(), "no one logged in yet"); | ||
let why = ensure(&ctx, &alice).await.unwrap_err(); | ||
assert!(why.to_string().contains("set GRAIN_PASSWORD"), "{why}"); | ||
| ||
// Lasting, it is the session kept. | ||
let lasting = oauth_session(&base, crate::keys::now_secs() + 3600); | ||
save(&ctx, &lasting).unwrap(); | ||
remember(&ctx, &lasting).unwrap(); | ||
assert_eq!(me(&ctx).await.unwrap(), lasting); | ||
assert_eq!(seen.lock().unwrap().refreshes, 0); | ||
| ||
// About to expire, it is refreshed in place. | ||
let expiring = oauth_session(&base, 0); | ||
save(&ctx, &expiring).unwrap(); | ||
let fresh = ensure(&ctx, &alice).await.unwrap(); | ||
assert_eq!(fresh.refresh_jwt, "refresh-0"); | ||
assert_eq!(load(&ctx, &alice.did), Some(fresh.clone())); | ||
// Asked again with the token it held, another process already | ||
// refreshed it: that one is taken. | ||
assert_eq!(refresh(&ctx, &expiring).await.unwrap(), fresh); | ||
assert_eq!(seen.lock().unwrap().refreshes, 1); | ||
| ||
// A refresh the server refuses leaves a login to do. | ||
let spent = Session { | ||
refresh_jwt: "spent".into(), | ||
..oauth_session(&base, 0) | ||
}; | ||
save(&ctx, &spent).unwrap(); | ||
assert!(ensure(&ctx, &alice).await.is_err()); | ||
// A confidential client's session is that client's to refresh. | ||
let mut theirs = oauth_session(&base, 0); | ||
theirs.oauth.as_mut().unwrap().client_id = Some("https://grain.test/client".into()); | ||
theirs.refresh_jwt = "refresh-theirs".into(); | ||
save(&ctx, &theirs).unwrap(); | ||
let why = refresh(&ctx, &theirs).await.unwrap_err(); | ||
assert!(why.to_string().contains("only that client"), "{why}"); | ||
| ||
save(&ctx, &fresh).unwrap(); | ||
assert!(logout(&ctx, "did:plc:alice").await.unwrap()); | ||
assert_eq!(seen.lock().unwrap().revoked, ["refresh-0"]); | ||
assert!(load(&ctx, &alice.did).is_none()); | ||
assert!(!logout(&ctx, "did:plc:alice").await.unwrap()); | ||
save(&ctx, &fresh).unwrap(); | ||
remove(&ctx, &alice.did); | ||
assert!(load(&ctx, &alice.did).is_none()); | ||
} | ||
} | ||
crates/grain/src/store.rs+45−1
| @@ -3,7 +3,8 @@ | ||
use anyhow::{Context, Result}; | ||
use serde::Serialize; | ||
use serde::de::DeserializeOwned; | ||
use std::path::Path; | ||
use std::path::{Path, PathBuf}; | ||
use std::time::Duration; | ||
| ||
/// The file's contents, or `None` when it is missing or unreadable: every | ||
/// file here is a cache or can be recreated by logging in again. | ||
| @@ -38,6 +39,35 @@ pub fn write_bytes(path: &Path, bytes: &[u8], private: bool) -> Result<()> { | ||
Ok(()) | ||
} | ||
| ||
/// An exclusive lock on the file at `path`, held until dropped: for | ||
/// processes that take turns at a file, as refreshing a session. | ||
pub struct Lock { | ||
_file: std::fs::File, | ||
} | ||
| ||
impl Lock { | ||
/// Waits for the lock, making the file (and its directory) if need be. | ||
pub async fn acquire(path: PathBuf) -> Result<Lock> { | ||
if let Some(dir) = path.parent() { | ||
std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?; | ||
} | ||
let file = std::fs::OpenOptions::new() | ||
.write(true) | ||
.create(true) | ||
.truncate(false) | ||
.open(&path) | ||
.with_context(|| format!("opening {}", path.display()))?; | ||
loop { | ||
match file.try_lock() { | ||
Err(std::fs::TryLockError::WouldBlock) => { | ||
tokio::time::sleep(Duration::from_millis(10)).await; | ||
} | ||
done => return Ok(done.map(|()| Lock { _file: file })?), | ||
} | ||
} | ||
} | ||
} | ||
| ||
#[cfg(test)] | ||
mod tests { | ||
use super::*; | ||
| @@ -72,4 +102,18 @@ mod tests { | ||
// A name too long for the file system: writing the copy fails. | ||
assert!(write(&dir.path().join("n".repeat(300)), &m, false).is_err()); | ||
} | ||
| ||
#[tokio::test] | ||
async fn locks_take_turns() { | ||
let dir = tempfile::tempdir().unwrap(); | ||
let path = dir.path().join("sessions/a.lock"); | ||
let held = Lock::acquire(path.clone()).await.unwrap(); | ||
let waiter = tokio::spawn(Lock::acquire(path.clone())); | ||
tokio::time::sleep(Duration::from_millis(50)).await; | ||
assert!(!waiter.is_finished()); | ||
drop(held); | ||
waiter.await.unwrap().unwrap(); | ||
// A directory where the file should be cannot be locked. | ||
assert!(Lock::acquire(dir.path().to_path_buf()).await.is_err()); | ||
} | ||
} | ||
crates/grain/src/xrpc.rs+33−4
| @@ -1,5 +1,7 @@ | ||
//! A minimal XRPC client over the shared HTTP connection pool. | ||
| ||
use crate::crypto::P256Key; | ||
use crate::oauth; | ||
use crate::telemetry; | ||
use anyhow::Result; | ||
use bytes::Bytes; | ||
| @@ -7,6 +9,7 @@ use serde::de::DeserializeOwned; | ||
use serde::{Deserialize, Serialize}; | ||
use serde_json::Value; | ||
use std::fmt; | ||
use std::sync::Arc; | ||
| ||
/// An XRPC call that reached the server and was refused. | ||
#[derive(Debug, Clone, PartialEq)] | ||
| @@ -56,6 +59,9 @@ impl XrpcError { | ||
pub enum Auth { | ||
/// An app password session's access token, or a service token. | ||
Bearer(String), | ||
/// An OAuth access token, bound to the key that signs a DPoP proof for | ||
/// each request. | ||
Dpop { token: String, key: Arc<P256Key> }, | ||
/// Headers made once, such as a space credential's. | ||
Headers(Vec<(String, String)>), | ||
} | ||
| @@ -66,9 +72,13 @@ impl Auth { | ||
} | ||
| ||
/// The headers for a `method` request to `url`. | ||
pub fn headers(&self, _method: &str, _url: &str) -> Vec<(String, String)> { | ||
pub fn headers(&self, method: &str, url: &str) -> Vec<(String, String)> { | ||
match self { | ||
Auth::Bearer(token) => vec![("authorization".into(), format!("Bearer {token}"))], | ||
Auth::Dpop { token, key } => vec![ | ||
("authorization".into(), format!("DPoP {token}")), | ||
("dpop".into(), oauth::proof(key, method, url, Some(token))), | ||
], | ||
Auth::Headers(headers) => headers.clone(), | ||
} | ||
} | ||
| @@ -192,7 +202,9 @@ impl Xrpc { | ||
Ok(out.blob) | ||
} | ||
| ||
/// Sends the request `make` builds, with `auth`'s headers. | ||
/// Sends the request `make` builds, with `auth`'s headers. A DPoP | ||
/// request the server refuses for want of a fresh nonce is sent again | ||
/// once, with the nonce it named. | ||
async fn send( | ||
&self, | ||
method: &str, | ||
| @@ -200,8 +212,15 @@ impl Xrpc { | ||
auth: Option<&Auth>, | ||
make: impl Fn() -> reqwest::RequestBuilder, | ||
) -> Result<reqwest::Response> { | ||
let headers = auth.map(|a| a.headers(method, url)).unwrap_or_default(); | ||
Ok(telemetry::send(with(make(), &headers), 0).await?) | ||
let attempt = |resend| { | ||
let headers = auth.map(|a| a.headers(method, url)).unwrap_or_default(); | ||
telemetry::send(with(make(), &headers), resend) | ||
}; | ||
let res = attempt(0).await?; | ||
if matches!(auth, Some(Auth::Dpop { .. })) && oauth::nonce_refused(url, &res) { | ||
return Ok(attempt(1).await?); | ||
} | ||
Ok(res) | ||
} | ||
| ||
pub async fn get_blob(&self, did: &str, cid: &str) -> Result<Vec<u8>> { | ||
| @@ -352,6 +371,16 @@ pub(crate) mod tests { | ||
); | ||
let fixed = vec![("x-credential".to_string(), "c".to_string())]; | ||
assert_eq!(Auth::Headers(fixed.clone()).headers("POST", "u"), fixed); | ||
let key = Arc::new(P256Key::generate()); | ||
let dpop = Auth::Dpop { | ||
token: "tok".into(), | ||
key: key.clone(), | ||
}; | ||
let headers = dpop.headers("GET", "http://pds.test/xrpc/x.y.z?a=1"); | ||
assert_eq!(headers[0], ("authorization".into(), "DPoP tok".into())); | ||
let (_, claims) = crate::crypto::jwt_parts(&headers[1].1).unwrap(); | ||
assert_eq!(claims["htu"], "http://pds.test/xrpc/x.y.z"); | ||
assert_eq!(claims["ath"], oauth::sha256("tok")); | ||
} | ||
| ||
#[tokio::test] | ||
crates/grain/Cargo.toml+2−0
| @@ -19,9 +19,11 @@ rand_core.workspace = true | ||
reqwest.workspace = true | ||
serde.workspace = true | ||
serde_json.workspace = true | ||
sha2.workspace = true | ||
tokio.workspace = true | ||
tracing.workspace = true | ||
| ||
[dev-dependencies] | ||
axum.workspace = true | ||
tempfile.workspace = true | ||
tracing-subscriber.workspace = true | ||
Cargo.lock+2−0
| @@ -554,6 +554,7 @@ name = "grain" | ||
version = "0.1.0" | ||
dependencies = [ | ||
"anyhow", | ||
"axum", | ||
"base64 0.22.1", | ||
"bs58", | ||
"bytes", | ||
| @@ -565,6 +566,7 @@ dependencies = [ | ||
"reqwest", | ||
"serde", | ||
"serde_json", | ||
"sha2", | ||
"tempfile", | ||
"tokio", | ||
"tracing", | ||
Cargo.toml+11−1
| @@ -28,8 +28,18 @@ reqwest = { version = "0.12", default-features = false, features = [ | ||
] } | ||
serde = { version = "1", features = ["derive"] } | ||
serde_json = "1" | ||
sha2 = "0.10" | ||
tempfile = "3" | ||
tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "net", "signal", "time"] } | ||
tokio = { version = "1", features = [ | ||
"rt", | ||
"rt-multi-thread", | ||
"macros", | ||
"net", | ||
"signal", | ||
"time", | ||
"io-util", | ||
"process", | ||
] } | ||
tracing = "0.1" | ||
tracing-subscriber = { version = "0.3", default-features = false, features = [ | ||
"std", | ||