~ knotless Search/
↑↓ move↵ open⇧↵ all results

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

The Rust tests run against the reference PDS, under a coverage gate

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:harness

Only what this pull adds. It is built on grain runs on a Rust server modelled on knotless, without hatk.

.knotless/workflows/harness.ymlnew+57−0
View file
@@ -0,0 +1,57 @@
# The Rust tests against the reference PDS, under the coverage gate, as
# `just cov` runs them (harness/cov.mjs): every test binary, instrumented,
# against @atproto/pds and a PLC directory from harness/pds.mjs, failing
# under harness/cov.json's lines or functions. Then that every XRPC answer
# grain-serve's tests received is as its lexicon says (harness/answers.mjs),
# and that the check would name a wrong one (harness/answers-test.sh).
#
# Node, the harness's packages, cargo-llvm-cov and the build are kept
# between runs; each is made again only when what it is made from changes.
image: rust:1.98
runners: [runner.knotless.dev]
on:
push: { refs: [refs/heads/main] }
pull: { base: [refs/heads/main, refs/heads/feature/*] }
cache:
[
target,
/usr/local/cargo/registry,
/usr/local/rustup,
/opt/node,
/opt/tools,
harness/node_modules,
]
env:
NODE_VERSION: 24.22.1
LLVM_COV_VERSION: 0.9.1
steps:
- name: node
run: |
[ "$(/opt/node/bin/node --version 2>/dev/null)" = "v$NODE_VERSION" ] && exit 0
mkdir -p /opt/node && find /opt/node -mindepth 1 -delete
curl -fsSL "https://nodejs.org/dist/v$NODE_VERSION/node-v$NODE_VERSION-linux-x64.tar.xz" | tar -xJ -C /opt/node --strip-components 1
/opt/node/bin/node --version
- name: harness
run: |
cd harness
lock=$(sha256sum package-lock.json)
[ "$(cat node_modules/.lock 2>/dev/null)" = "$lock" ] && exit 0
PATH=/opt/node/bin:$PATH npm ci --no-audit --no-fund
echo "$lock" > node_modules/.lock
- name: tools
run: |
rustup component add llvm-tools-preview
[ "$(cat /opt/tools/.llvm-cov 2>/dev/null)" = "$LLVM_COV_VERSION" ] && exit 0
mkdir -p /opt/tools
curl -fsSL "https://github.com/taiki-e/cargo-llvm-cov/releases/download/v$LLVM_COV_VERSION/cargo-llvm-cov-x86_64-unknown-linux-gnu.tar.gz" | tar -xz -C /opt/tools
echo "$LLVM_COV_VERSION" > /opt/tools/.llvm-cov
- name: test
run: |
export PATH=/opt/node/bin:/opt/tools:$PATH
COV_JOBS=$(( $(nproc) / 2 )) COV_HARNESSES=2 COV_TOOLCHAIN=$(rustup show active-toolchain | cut -d' ' -f1) \
node harness/cov.mjs
- name: answers
run: |
export PATH=/opt/node/bin:$PATH
harness/answers-test.sh
node harness/answers.mjs target/llvm-cov-target/cov-report/answers
.knotless/workflows/rust.yml+2−4
View file
@@ -1,5 +1,5 @@
# The Rust workspace: formatting, clippy with warnings denied, and the
# tests, as `just lint` and `just t` run them. Kept apart from check.yml
# The Rust workspace: formatting and clippy with warnings denied, as `just
# lint` runs them; harness.yml runs the tests. Kept apart from check.yml
# and test.yml, which run the SvelteKit app and hatk's server in Node,
# until the move off hatk folds them together.
#
@@ -16,5 +16,3 @@ steps:
run: rustup component add rustfmt && cargo fmt --all --check
- name: clippy
run: rustup component add clippy && cargo clippy --workspace --all-targets --locked -- -D warnings
- name: test
run: cargo test --workspace --locked
crates/grain/tests/common/mod.rsnew+243−0
View file
@@ -0,0 +1,243 @@
//! A reference PDS for the process (`harness/pds.mjs`), accounts on it,
//! and contexts that resolve identities through it. As knotless's
//! `crates/knotless/tests/common`.
#![allow(dead_code)]
use grain::env::{Ctx, Env};
use serde::Deserialize;
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::{Mutex, OnceLock};
#[derive(Deserialize, Clone, Debug)]
pub struct Account {
pub handle: String,
pub did: String,
pub password: String,
}
#[derive(Deserialize, Clone, Debug)]
pub struct Net {
pub pds: String,
pub plc: String,
pub client: ClientTls,
pub accounts: Vec<Account>,
}
/// The hostname the harness resolves to 127.0.0.1 for confidential OAuth
/// clients, and the certificate it trusts for it.
#[derive(Deserialize, Clone, Debug)]
pub struct ClientTls {
pub host: String,
pub cert: String,
pub key: String,
}
static NET: OnceLock<Net> = OnceLock::new();
static CHILD: Mutex<Option<Child>> = Mutex::new(None);
pub fn root() -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR")).join("../..")
}
/// The harness `GRAIN_TEST_NET` names, if it is there and listening: one
/// `just pds` left running, or harness/cov.mjs's.
fn shared() -> Option<Net> {
let file = std::env::var_os("GRAIN_TEST_NET")?;
let net: Net = serde_json::from_str(&std::fs::read_to_string(file).ok()?).ok()?;
let addr = net
.pds
.trim_start_matches("http://")
.replace("localhost", "127.0.0.1");
std::net::TcpStream::connect(addr).ok()?;
Some(net)
}
/// The reference PDS: the one `GRAIN_TEST_NET` names, if it is up, else
/// one started for this test binary, which exits when this process does
/// and its stdin closes.
pub fn net() -> &'static Net {
NET.get_or_init(|| {
if let Some(net) = shared() {
return net;
}
// The dev-env picks free ports and then binds them, so harnesses
// starting side by side can collide: try again on a fresh one.
let mut failures = Vec::new();
for _ in 0..3 {
match start() {
Ok((net, child)) => {
*CHILD.lock().unwrap() = Some(child);
return net;
}
Err(log) => failures.push(log),
}
}
panic!("the harness exited:\n{}", failures.join("\n---\n"));
})
}
/// One harness, up and reporting, or why it died.
fn start() -> Result<(Net, Child), String> {
let mut child = Command::new("node")
.arg(root().join("harness/pds.mjs"))
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("node runs harness/pds.mjs (`just setup` installs its packages)");
// Kept to explain a harness that dies before it is up.
let log = std::sync::Arc::new(Mutex::new(String::new()));
let stderr = BufReader::new(child.stderr.take().unwrap());
let sink = log.clone();
let drained = std::thread::spawn(move || {
for line in stderr.lines().map_while(Result::ok) {
let mut log = sink.lock().unwrap();
if log.len() < 1 << 16 {
log.push_str(&line);
log.push('\n');
}
}
});
let mut lines = BufReader::new(child.stdout.take().unwrap()).lines();
let net: Net = loop {
let Some(Ok(line)) = lines.next() else {
let _ = child.wait();
let _ = drained.join();
return Err(log.lock().unwrap().clone());
};
if line.starts_with("{\"pds\"") {
break serde_json::from_str(&line).unwrap();
}
};
std::thread::spawn(move || for _ in lines {});
Ok((net, child))
}
static NEXT: Mutex<usize> = Mutex::new(0);
/// A handle no other test binary, in this run or an earlier one on the
/// same harness, has: a handle's first label is at most 18 characters.
pub fn fresh_handle() -> String {
let n = {
let mut next = NEXT.lock().unwrap();
*next += 1;
*next
};
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let tag = (nanos as u64 ^ (u64::from(std::process::id()) << 32)) as u32;
format!("g{tag:08x}{n}.test")
}
/// A new account, which no other test uses.
pub async fn account() -> Account {
let handle = fresh_handle();
let password = format!("{handle}-password");
let res: serde_json::Value = reqwest::Client::new()
.post(format!(
"{}/xrpc/com.atproto.server.createAccount",
net().pds
))
.json(&serde_json::json!({
"email": format!("{handle}@test.com"),
"handle": handle,
"password": password,
}))
.send()
.await
.and_then(|r| r.error_for_status())
.expect("the harness makes an account")
.json()
.await
.unwrap();
Account {
handle,
did: res["did"].as_str().unwrap().to_string(),
password,
}
}
/// A context whose config and cache are under `dir`, resolving handles at
/// the PDS and DIDs at the harness's PLC directory, with `extra` beside.
pub fn ctx(dir: &Path, extra: &[(&str, &str)]) -> Ctx {
let net = net();
let dir = dir.to_str().unwrap().to_string();
let extra: Vec<(String, String)> = extra
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
Ctx::new(Env::from_vars(|k| {
if let Some((_, v)) = extra.iter().find(|(key, _)| key == k) {
return Some(v.clone());
}
match k {
"GRAIN_CONFIG_DIR" => Some(format!("{dir}/config")),
"GRAIN_CACHE_DIR" => Some(format!("{dir}/cache")),
"GRAIN_HANDLE_RESOLVER" => Some(net.pds.clone()),
"GRAIN_PLC_URL" => Some(net.plc.clone()),
_ => None,
}
}))
}
/// A browser that signs in as `account` and approves, plus `extra` (such
/// as `--reject`): `harness/authorize.mjs`, as `GRAIN_BROWSER` runs it.
pub fn browser(account: &Account, extra: &str) -> String {
format!(
"node {} {} {} --ca {} {extra}",
root().join("harness/authorize.mjs").display(),
account.handle,
account.password,
net().client.cert,
)
}
/// Runs [`browser`] on `url`, as a user would approve it, and waits for it.
pub async fn approve(account: &Account, extra: &str, url: &str) {
let command = browser(account, extra);
let mut words = command.split_whitespace();
let status = tokio::process::Command::new(words.next().unwrap())
.args(words)
.arg(url)
.status()
.await
.unwrap();
assert!(status.success(), "harness/authorize.mjs failed on {url}");
}
/// An https front for a server on `port`, at the harness's client
/// hostname, until it is dropped.
pub struct TlsProxy {
pub url: String,
child: Child,
}
impl Drop for TlsProxy {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}
pub fn tls_proxy(port: u16) -> TlsProxy {
let tls = &net().client;
let mut child = Command::new("node")
.arg(root().join("harness/tls-proxy.mjs"))
.args([&tls.cert, &tls.key, &port.to_string()])
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.spawn()
.expect("node runs harness/tls-proxy.mjs");
let mut line = String::new();
BufReader::new(child.stdout.take().unwrap())
.read_line(&mut line)
.unwrap();
TlsProxy {
url: format!("https://{}:{}", tls.host, line.trim()),
child,
}
}
crates/grain/tests/cli.rsnew+26−0
View file
@@ -0,0 +1,26 @@
//! The `grain` command line: its version, and its usage for anything else.
use std::process::Command;
fn grain(args: &[&str]) -> (Option<i32>, String, String) {
let out = Command::new(env!("CARGO_BIN_EXE_grain"))
.args(args)
.output()
.unwrap();
let text = |b: &[u8]| String::from_utf8_lossy(b).into_owned();
(out.status.code(), text(&out.stdout), text(&out.stderr))
}
#[test]
fn names_its_version() {
let (code, out, _) = grain(&["--version"]);
assert_eq!(code, Some(0));
assert_eq!(out.trim(), format!("grain {}", env!("CARGO_PKG_VERSION")));
}
#[test]
fn anything_else_is_usage() {
let (code, _, err) = grain(&["nonsense"]);
assert_eq!(code, Some(2));
assert!(err.starts_with("usage: grain"), "{err}");
}
crates/grain/tests/oauth.rsnew+255−0
View file
@@ -0,0 +1,255 @@
//! Logging in with OAuth against the reference PDS's authorization server,
//! with `harness/authorize.mjs` as the user's browser: the command line's
//! loopback client, and a confidential client, as grain-serve is, behind
//! `harness/tls-proxy.mjs`. Each pushes its request (PAR), redeems the code,
//! calls the PDS with DPoP proofs under the nonces it hands out, refreshes
//! and revokes.
mod common;
use common::*;
use axum::Router;
use axum::extract::Query;
use axum::routing::get;
use grain::crypto::P256Key;
use grain::env::Ctx;
use grain::oauth::{self, Client};
use grain::session::{self, Session};
use grain::xrpc::{Auth, Xrpc};
use grain::{identity, keys};
use serde_json::{Value, json};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
/// A 1x1 PNG.
const PNG: &[u8] = &[
0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0x00, 0x00, 0x0d, 0x49, 0x48, 0x44, 0x52,
0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x08, 0x06, 0x00, 0x00, 0x00, 0x1f, 0x15, 0xc4,
0x89, 0x00, 0x00, 0x00, 0x0d, 0x49, 0x44, 0x41, 0x54, 0x78, 0x9c, 0x63, 0xf8, 0xcf, 0xc0, 0xf0,
0x1f, 0x00, 0x05, 0x00, 0x01, 0xff, 0x89, 0x99, 0x3d, 0x1d, 0x00, 0x00, 0x00, 0x00, 0x49, 0x45,
0x4e, 0x44, 0xae, 0x42, 0x60, 0x82,
];
/// Creates a record of `collection` in `me`'s repo.
async fn create(ctx: &Ctx, me: &Session, collection: &str, record: Value) -> anyhow::Result<Value> {
Xrpc::new(ctx.http.clone(), &me.service)
.procedure(
"com.atproto.repo.createRecord",
Some(&json!({ "repo": me.did, "collection": collection, "record": record })),
Some(&me.auth()),
)
.await
}
fn post() -> Value {
json!({ "$type": "app.bsky.feed.post", "text": "from grain", "createdAt": "2026-10-31T00:00:00.000Z" })
}
/// What an OAuth session may do, and what its scope keeps it from: the
/// post grain cross-posts goes through, a like does not; a picture goes up,
/// a text file does not. Each call carries a DPoP proof, sent again with
/// the nonce the PDS asks for.
async fn scoped(ctx: &Ctx, me: &Session) {
assert!(matches!(me.auth(), Auth::Dpop { .. }));
create(ctx, me, "app.bsky.feed.post", post()).await.unwrap();
let like = json!({
"$type": "app.bsky.feed.like",
"subject": { "uri": format!("at://{}/app.bsky.feed.post/3aaaaaaaaaaaa", me.did), "cid": "bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" },
"createdAt": "2026-10-31T00:00:00.000Z",
});
let why = create(ctx, me, "app.bsky.feed.like", like)
.await
.unwrap_err();
assert!(oauth::out_of_scope(&why), "{why:#}");
let pds = Xrpc::new(ctx.http.clone(), &me.service);
let blob = pds
.upload_blob(PNG.to_vec(), "image/png", &me.auth())
.await
.unwrap();
assert_eq!(blob.mime_type, "image/png");
let why = pds
.upload_blob(b"words".to_vec(), "text/plain", &me.auth())
.await
.unwrap_err();
assert!(oauth::out_of_scope(&why), "{why:#}");
}
#[tokio::test]
async fn a_command_line_login_refreshes_and_logs_out() {
let a = account().await;
let dir = tempfile::tempdir().unwrap();
let browser = browser(&a, "");
let ctx = ctx(dir.path(), &[("GRAIN_BROWSER", browser.as_str())]);
let pending = oauth::start(&ctx, &format!("@{}", a.handle)).await.unwrap();
assert!(pending.redirect_uri.starts_with("http://127.0.0.1:"));
oauth::open(&ctx, &pending.url);
let first = pending.finish(&ctx).await.unwrap();
assert_eq!(
(first.did.as_str(), first.handle.as_str()),
(a.did.as_str(), a.handle.as_str())
);
let o = first.oauth.clone().unwrap();
assert_eq!(o.loopback_id, Some(oauth::client_id()));
assert_eq!(o.client_id, None);
assert!(o.expires_at > keys::now_secs());
assert!(!oauth::behind(&first, oauth::SCOPE));
assert_eq!(session::load(&ctx, &a.did), Some(first.clone()));
scoped(&ctx, &first).await;
// About to expire, it is refreshed in place, under the client it was
// granted to.
let mut expiring = first.clone();
expiring.oauth.as_mut().unwrap().expires_at = 0;
session::save(&ctx, &expiring).unwrap();
let id = identity::resolve(&ctx, &a.handle).await.unwrap();
let fresh = session::ensure(&ctx, &id).await.unwrap();
assert_ne!(fresh.refresh_jwt, first.refresh_jwt);
assert_ne!(fresh.access_jwt, first.access_jwt);
create(&ctx, &fresh, "app.bsky.feed.post", post())
.await
.unwrap();
// The refresh token it held is spent.
let client = Client::granted(&o).unwrap();
assert!(oauth::refresh(&ctx, &client, &first, &o).await.is_err());
// Logging out revokes the tokens: put back, they no longer refresh.
assert!(session::logout(&ctx, &a.handle).await.unwrap());
let mut revoked = fresh;
revoked.oauth.as_mut().unwrap().expires_at = 0;
session::save(&ctx, &revoked).unwrap();
let why = session::ensure(&ctx, &id).await.unwrap_err();
assert!(why.to_string().contains("set GRAIN_PASSWORD"), "{why:#}");
}
#[tokio::test]
async fn a_refused_login_says_so() {
let a = account().await;
let dir = tempfile::tempdir().unwrap();
let ctx = ctx(dir.path(), &[]);
let pending = oauth::start(&ctx, &a.handle).await.unwrap();
let url = pending.url.clone();
let (finished, ()) = tokio::join!(pending.finish(&ctx), approve(&a, "--reject", &url));
let why = finished.unwrap_err();
assert!(why.to_string().contains("the login was refused"), "{why:#}");
assert!(session::load(&ctx, &a.did).is_none());
}
/// A confidential client at the harness's client hostname, which serves
/// its metadata document and keeps the query the browser comes back with.
struct Confidential {
client: Client,
redirect_uri: String,
back: Arc<Mutex<Option<HashMap<String, String>>>>,
_proxy: TlsProxy,
}
async fn confidential() -> Confidential {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let proxy = tls_proxy(listener.local_addr().unwrap().port());
let id = format!("{}/oauth-client-metadata.json", proxy.url);
let redirect_uri = format!("{}/callback", proxy.url);
let client = Client::confidential(&id, P256Key::generate());
let metadata = client.metadata("grain tests", &[&redirect_uri], oauth::SCOPE);
let back = Arc::new(Mutex::new(None));
let kept = back.clone();
let app = Router::new()
.route(
"/oauth-client-metadata.json",
get(move || async move { axum::Json(metadata) }),
)
.route(
"/callback",
get(move |Query(q): Query<HashMap<String, String>>| async move {
*kept.lock().unwrap() = Some(q);
"back"
}),
);
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
Confidential {
client,
redirect_uri,
back,
_proxy: proxy,
}
}
impl Confidential {
fn came_back(&self) -> HashMap<String, String> {
self.back
.lock()
.unwrap()
.take()
.expect("the browser came back")
}
}
#[tokio::test]
async fn a_confidential_client_logs_in_refreshes_and_revokes() {
let a = account().await;
let dir = tempfile::tempdir().unwrap();
let ctx = ctx(dir.path(), &[]);
let c = confidential().await;
let id = identity::resolve(&ctx, &a.handle).await.unwrap();
let request = oauth::authorize(
&ctx,
&c.client,
id,
&a.handle,
&c.redirect_uri,
oauth::SCOPE,
)
.await
.unwrap();
approve(&a, "", &request.url).await;
let params = c.came_back();
let session = request.redeem(&ctx, &c.client, &params).await.unwrap();
assert_eq!(session.did, a.did);
assert_eq!(session.handle, "");
let session = oauth::named(&ctx, session).await.unwrap();
assert_eq!(session.handle, a.handle);
let o = session.oauth.clone().unwrap();
assert_eq!(o.client_id.as_deref(), Some(c.client.id.as_str()));
assert_eq!(o.loopback_id, None);
// Only the client that holds the key refreshes it.
assert!(Client::granted(&o).is_err());
scoped(&ctx, &session).await;
let fresh = oauth::refresh(&ctx, &c.client, &session, &o).await.unwrap();
assert_ne!(fresh.refresh_jwt, session.refresh_jwt);
create(&ctx, &fresh, "app.bsky.feed.post", post())
.await
.unwrap();
let fo = fresh.oauth.clone().unwrap();
oauth::revoke(&ctx, &c.client, &fresh, &fo).await.unwrap();
assert!(oauth::refresh(&ctx, &c.client, &fresh, &fo).await.is_err());
}
#[tokio::test]
async fn an_account_made_at_sign_up_logs_in() {
let dir = tempfile::tempdir().unwrap();
let ctx = ctx(dir.path(), &[]);
let c = confidential().await;
let handle = fresh_handle();
let newcomer = Account {
password: format!("{handle}-password"),
handle,
did: String::new(),
};
let request = oauth::sign_up(&ctx, &c.client, &net().pds, &c.redirect_uri, oauth::SCOPE)
.await
.unwrap();
approve(&newcomer, "--create", &request.url).await;
let params = c.came_back();
let session = request.redeem(&ctx, &c.client, &params).await.unwrap();
let session = oauth::named(&ctx, session).await.unwrap();
assert_eq!(session.handle, newcomer.handle);
assert_eq!(
identity::resolve(&ctx, &newcomer.handle).await.unwrap().did,
session.did
);
}
crates/grain/tests/pds.rsnew+237−0
View file
@@ -0,0 +1,237 @@
//! The core against the reference PDS: identities resolved through its PLC
//! directory, app-password sessions kept and refreshed, service tokens it
//! signs checked against the account's key, and blobs put and fetched.
mod common;
use common::*;
use grain::identity::{self, NoSuchHandle};
use grain::session::{self, Session};
use grain::xrpc::{Auth, Xrpc, XrpcError};
use grain::{auth, crypto, keys};
use serde_json::{Value, json};
/// A 1x1 PNG, as a photo's smallest blob.
const PNG: &[u8] = &[
0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0x00, 0x00, 0x0d, 0x49, 0x48, 0x44, 0x52,
0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x08, 0x06, 0x00, 0x00, 0x00, 0x1f, 0x15, 0xc4,
0x89, 0x00, 0x00, 0x00, 0x0d, 0x49, 0x44, 0x41, 0x54, 0x78, 0x9c, 0x63, 0xf8, 0xcf, 0xc0, 0xf0,
0x1f, 0x00, 0x05, 0x00, 0x01, 0xff, 0x89, 0x99, 0x3d, 0x1d, 0x00, 0x00, 0x00, 0x00, 0x49, 0x45,
0x4e, 0x44, 0xae, 0x42, 0x60, 0x82,
];
#[tokio::test]
async fn an_account_resolves_through_plc() {
let a = account().await;
let dir = tempfile::tempdir().unwrap();
let ctx = ctx(dir.path(), &[]);
let id = identity::resolve(&ctx, &a.handle).await.unwrap();
assert_eq!(id.did, a.did);
assert!(a.did.starts_with("did:plc:"), "{}", a.did);
assert_eq!(id.pds, net().pds);
// By DID, and again from the cache.
let by_did = identity::resolve(&ctx, &a.did).await.unwrap();
assert_eq!((by_did.did, by_did.pds), (a.did.clone(), net().pds.clone()));
assert_eq!(identity::resolve(&ctx, &a.handle).await.unwrap().did, a.did);
let doc = identity::document(&ctx, &a.did).await.unwrap();
assert_eq!(doc["alsoKnownAs"][0], format!("at://{}", a.handle));
assert_eq!(
identity::service(&doc, "#atproto_pds"),
Some(net().pds.clone())
);
identity::signing_key(&doc).unwrap();
assert_eq!(
identity::resolve_pds(&ctx, &a.did).await.unwrap(),
net().pds
);
assert_eq!(
identity::verified_handle(&ctx, &a.did).await.unwrap(),
Some(a.handle.clone())
);
assert_eq!(identity::handle(&ctx, &a.did).await, Some(a.handle.clone()));
let handles = identity::handles(&ctx, [a.did.as_str(), a.did.as_str()]).await;
assert_eq!(handles.get(&a.did), Some(&a.handle));
// A handle no one has is no one's, as the PDS answers.
let nobody = format!("nobody-{}", fresh_handle());
let why = identity::resolve(&ctx, &nobody).await.unwrap_err();
assert!(why.downcast_ref::<NoSuchHandle>().is_some(), "{why:#}");
// A DID the directory never heard of.
let why = identity::resolve(&ctx, "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").await;
assert!(why.is_err());
}
/// The account's own password, then an app password, as the apps log in.
async fn sessions(ctx: &grain::env::Ctx, a: &Account) -> (Session, Session) {
let main = session::create(ctx, &a.handle, &a.password).await.unwrap();
#[derive(serde::Deserialize)]
struct Made {
password: String,
}
let made: Made = Xrpc::new(ctx.http.clone(), &main.service)
.procedure(
"com.atproto.server.createAppPassword",
Some(&json!({ "name": "grain-tests" })),
Some(&main.auth()),
)
.await
.unwrap();
let app = session::create(ctx, &a.handle, &made.password)
.await
.unwrap();
(main, app)
}
#[tokio::test]
async fn service_tokens_from_an_app_password() {
let a = account().await;
let dir = tempfile::tempdir().unwrap();
let ctx = ctx(dir.path(), &[]);
let (main, app) = sessions(&ctx, &a).await;
assert_eq!(
(main.did.as_str(), main.handle.as_str()),
(a.did.as_str(), a.handle.as_str())
);
assert_eq!(app.did, a.did);
assert_eq!(app.service, net().pds);
assert!(matches!(app.auth(), Auth::Bearer(_)));
assert!(!app.expiring());
// The account's password is wrong for anyone else.
let why = session::create(&ctx, &a.handle, "not-the-password")
.await
.unwrap_err();
assert_eq!(XrpcError::of(&why).map(|e| e.status), Some(401), "{why:#}");
let aud = "did:web:grain.test";
let lxm = "social.grain.unspecced.getCameras";
let token = auth::service_token(&ctx, &app, aud, lxm).await.unwrap();
assert_eq!(auth::verify(&ctx, &token, aud, lxm).await.unwrap(), a.did);
let (_, claims) = crypto::jwt_parts(&token).unwrap();
assert_eq!(
(claims["iss"].as_str(), claims["lxm"].as_str()),
(Some(a.did.as_str()), Some(lxm))
);
// Held to the service and the method it names.
assert!(
auth::verify(&ctx, &token, "did:web:other.test", lxm)
.await
.is_err()
);
assert!(
auth::verify(&ctx, &token, aud, "social.grain.other")
.await
.is_err()
);
let exp = keys::now_secs() + 600;
let lasting = auth::lasting_token(&ctx, &main, aud, lxm, exp)
.await
.unwrap();
let (_, claims) = crypto::jwt_parts(&lasting).unwrap();
assert_eq!(claims["exp"], exp);
assert_eq!(auth::verify(&ctx, &lasting, aud, lxm).await.unwrap(), a.did);
}
#[tokio::test]
async fn password_sessions_are_kept_refreshed_and_forgotten() {
let a = account().await;
let dir = tempfile::tempdir().unwrap();
let ctx = ctx(
dir.path(),
&[
("GRAIN_IDENTIFIER", a.handle.as_str()),
("GRAIN_PASSWORD", a.password.as_str()),
],
);
// No session kept: the password logs in, and the session is kept.
let first = session::me(&ctx).await.unwrap();
assert_eq!(first.did, a.did);
assert_eq!(session::load(&ctx, &a.did), Some(first.clone()));
session::remember(&ctx, &first).unwrap();
// An access token about to expire is refreshed with the refresh token.
let id = identity::resolve(&ctx, &a.handle).await.unwrap();
let stale = Session {
access_jwt: "expired".into(),
..first.clone()
};
session::save(&ctx, &stale).unwrap();
let fresh = session::ensure(&ctx, &id).await.unwrap();
assert_ne!(fresh.refresh_jwt, first.refresh_jwt);
assert!(!fresh.expiring());
assert_eq!(session::load(&ctx, &a.did), Some(fresh.clone()));
// The refreshed session lets calls in.
let who: Value = Xrpc::new(ctx.http.clone(), &fresh.service)
.query("com.atproto.server.getSession", &[], Some(&fresh.auth()))
.await
.unwrap();
assert_eq!(who["did"], a.did);
// A refresh token the PDS was told to forget, and no password: there
// is no session to be had.
let _: Value = Xrpc::new(ctx.http.clone(), &fresh.service)
.procedure(
"com.atproto.server.deleteSession",
None,
Some(&Auth::bearer(&fresh.refresh_jwt)),
)
.await
.unwrap();
let ended = Session {
access_jwt: "expired".into(),
..fresh
};
session::save(&ctx, &ended).unwrap();
let alone = common::ctx(dir.path(), &[("GRAIN_IDENTIFIER", a.handle.as_str())]);
let why = session::ensure(&alone, &id).await.unwrap_err();
assert!(why.to_string().contains("set GRAIN_PASSWORD"), "{why:#}");
assert!(session::logout(&ctx, &a.handle).await.unwrap());
assert!(session::load(&ctx, &a.did).is_none());
}
#[tokio::test]
async fn blobs_go_up_and_come_back() {
let a = account().await;
let dir = tempfile::tempdir().unwrap();
let ctx = ctx(dir.path(), &[]);
let me = session::create(&ctx, &a.handle, &a.password).await.unwrap();
let pds = Xrpc::new(ctx.http.clone(), &me.service);
let blob = pds
.upload_blob(PNG.to_vec(), "image/png", &me.auth())
.await
.unwrap();
assert_eq!(blob.kind, "blob");
assert_eq!(blob.mime_type, "image/png");
assert_eq!(blob.size, PNG.len() as u64);
// Referenced by a record, as a photo's is, it is the account's to serve.
let record = json!({
"$type": "app.bsky.feed.post",
"text": "",
"createdAt": "2026-10-31T00:00:00.000Z",
"embed": {
"$type": "app.bsky.embed.images",
"images": [{ "image": blob, "alt": "" }],
},
});
let _: Value = pds
.procedure(
"com.atproto.repo.createRecord",
Some(&json!({ "repo": me.did, "collection": "app.bsky.feed.post", "record": record })),
Some(&me.auth()),
)
.await
.unwrap();
let back = pds.get_blob(&me.did, &blob.link.cid).await.unwrap();
assert_eq!(back, PNG);
// A blob it does not have is refused as XRPC refuses.
let missing = "bafkreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let why = pds.get_blob(&me.did, missing).await.unwrap_err();
let status = XrpcError::of(&why).map(|e| e.status);
assert!(status.is_some_and(|s| (400..500).contains(&s)), "{why:#}");
}
crates/grain-serve/src/lib.rs+14−2
View file
@@ -2,12 +2,24 @@
//! index, the queries, ingest, login and push join it pull by pull on
//! `feature/rust`, as the plan orders them.
use axum::{Json, Router, routing::get};
use axum::http::StatusCode;
use axum::{Json, Router, routing::any, routing::get};
use serde_json::{Value, json};
/// The routes the server answers.
pub fn router() -> Router {
Router::new().route("/xrpc/_health", get(health))
Router::new()
.route("/xrpc/_health", get(health))
.route("/xrpc/{*method}", any(not_implemented))
}
/// A method the server does not serve, as atproto's XRPC servers refuse it.
async fn not_implemented() -> (StatusCode, Json<Value>) {
let refused = json!({
"error": "MethodNotImplemented",
"message": "Method Not Implemented",
});
(StatusCode::NOT_IMPLEMENTED, Json(refused))
}
/// `/xrpc/_health`, as atproto services answer it: the running version.
crates/grain-serve/tests/common/mod.rsnew+77−0
View file
@@ -0,0 +1,77 @@
//! grain-serve on a port of its own, and the XRPC calls the tests make to
//! it, each answer kept under `GRAIN_ANSWERS` when it is set, for
//! `harness/answers.mjs` to hold to its lexicon. Every test of what the
//! server answers asks through [`Served::get`] or [`Served::post`], so
//! nothing it answers goes unchecked.
#![allow(dead_code)]
use serde_json::{Value, json};
pub struct Served {
pub base: String,
http: reqwest::Client,
}
/// grain-serve, serving on a free port until the test's runtime ends.
pub async fn served() -> Served {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
tokio::spawn(grain_serve::serve(listener));
Served {
base,
http: reqwest::Client::new(),
}
}
impl Served {
/// A query: `path` is `/xrpc/<nsid>` and its parameters.
pub async fn get(&self, path: &str) -> (u16, Value) {
let res = self.http.get(format!("{}{path}", self.base));
self.answer(path, res).await
}
/// A procedure, with `body` as its JSON input.
pub async fn post(&self, path: &str, body: &Value) -> (u16, Value) {
let res = self.http.post(format!("{}{path}", self.base)).json(body);
self.answer(path, res).await
}
async fn answer(&self, path: &str, req: reqwest::RequestBuilder) -> (u16, Value) {
let res = req.send().await.unwrap();
let status = res.status().as_u16();
let json = res
.headers()
.get("content-type")
.is_some_and(|v| v.as_bytes().starts_with(b"application/json"));
let bytes = res.bytes().await.unwrap();
let body = if json {
serde_json::from_slice(&bytes).unwrap()
} else {
Value::Null
};
if let Some(dir) = std::env::var_os("GRAIN_ANSWERS") {
// An answer that is not JSON is kept with no body.
let mut kept = json!({ "path": path, "status": status });
if json {
kept["body"] = body.clone();
}
keep(&dir, &kept);
}
(status, body)
}
}
/// Adds `answer` to this process's file of them under `dir`.
fn keep(dir: &std::ffi::OsStr, answer: &Value) {
use std::io::Write;
static FILE: std::sync::Mutex<()> = std::sync::Mutex::new(());
let _held = FILE.lock().unwrap();
std::fs::create_dir_all(dir).unwrap();
let file = std::path::Path::new(dir).join(format!("serve-{}.jsonl", std::process::id()));
let mut out = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(file)
.unwrap();
writeln!(out, "{answer}").unwrap();
}
crates/grain-serve/tests/health.rs+19−14
View file
@@ -1,20 +1,25 @@
//! The server answers atproto's health check with its version.
//! The server answers atproto's health check with its version, and a
//! method it does not serve as XRPC refuses one.
mod common;
#[tokio::test]
async fn health_names_the_version() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(grain_serve::serve(listener));
let res = reqwest::get(format!("http://{addr}/xrpc/_health"))
.await
.unwrap();
assert_eq!(res.status(), 200);
let body: serde_json::Value = res.json().await.unwrap();
let s = common::served().await;
let (status, body) = s.get("/xrpc/_health").await;
assert_eq!(status, 200);
assert_eq!(body["version"], env!("CARGO_PKG_VERSION"));
}
let missing = reqwest::get(format!("http://{addr}/xrpc/social.grain.nothing"))
.await
.unwrap();
assert_eq!(missing.status(), 404);
#[tokio::test]
async fn a_method_not_served_is_not_implemented() {
let s = common::served().await;
for (status, body) in [
s.get("/xrpc/social.grain.nothing").await,
s.post("/xrpc/social.grain.nothing", &serde_json::json!({}))
.await,
] {
assert_eq!(status, 501);
assert_eq!(body["error"], "MethodNotImplemented");
}
}
crates/grain-serve/tests/main.rsnew+54−0
View file
@@ -0,0 +1,54 @@
//! The `grain-serve` binary: it serves where `GRAIN_LISTEN` says, stops
//! when told to, and says why when it cannot listen.
use std::process::{Command, Stdio};
use std::time::Duration;
/// A port free a moment ago.
fn free_port() -> u16 {
std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
}
#[tokio::test]
async fn serves_until_interrupted() {
let addr = format!("127.0.0.1:{}", free_port());
let mut child = Command::new(env!("CARGO_BIN_EXE_grain-serve"))
.env("GRAIN_LISTEN", &addr)
.stdout(Stdio::null())
.spawn()
.unwrap();
let url = format!("http://{addr}/xrpc/_health");
let mut up = false;
for _ in 0..100 {
if reqwest::get(&url).await.is_ok_and(|r| r.status() == 200) {
up = true;
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(up, "grain-serve never answered at {addr}");
// As ctrl-c stops it: gracefully, so it exits 0.
let interrupted = Command::new("kill")
.args(["-INT", &child.id().to_string()])
.status()
.unwrap();
assert!(interrupted.success());
assert!(child.wait().unwrap().success());
}
#[test]
fn says_why_it_cannot_listen() {
let taken = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let addr = taken.local_addr().unwrap().to_string();
let out = Command::new(env!("CARGO_BIN_EXE_grain-serve"))
.env("GRAIN_LISTEN", &addr)
.output()
.unwrap();
assert!(!out.status.success());
let said = String::from_utf8_lossy(&out.stderr);
assert!(said.contains(&format!("listening on {addr}")), "{said}");
}
harness/answers-test/lexicons/getThing.jsonnew+21−0
View file
harness/answers-test/right/answers.jsonlnew+4−0
View file
harness/answers-test/wrong/answers.jsonlnew+5−0
View file
harness/.node-versionnew+1−0
View file
harness/answers-test.shnew+21−0
View file
harness/answers.mjsnew+99−0
View file
harness/authorize.mjsnew+145−0
View file
harness/cov.jsonnew+1−0
View file
harness/cov.mjsnew+357−0
View file
harness/package-lock.jsonnew+9422−0
View file
A large diff, not shown.
harness/package.jsonnew+10−0
View file
harness/pds.mjsnew+114−0
View file
harness/tls-proxy.mjsnew+29−0
View file
AGENTS.md+30−0
View file
justfile+31−3
View file