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

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

Core resolves identities, verifies atproto keys and calls XRPC

Merged grain.social asked to merge into feature/rust from grain.social/grain:core/identity

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

crates/grain/src/crypto.rsnew+359−0
View file
@@ -0,0 +1,359 @@
//! The keys atproto uses: secp256k1 and P-256, as multikeys and `did:key`s,
//! and the JWTs they sign.
use anyhow::{Context, Result, bail};
use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use serde::Serialize;
use serde::de::DeserializeOwned;
use serde_json::Value;
const K256_CODEC: [u8; 2] = [0xe7, 0x01];
const P256_CODEC: [u8; 2] = [0x80, 0x24];
/// A public key a DID document or `did:key` names.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PublicKey {
K256(k256::ecdsa::VerifyingKey),
P256(p256::ecdsa::VerifyingKey),
}
impl PublicKey {
/// Parses a multibase (`z`, base58btc) multicodec key.
pub fn from_multibase(text: &str) -> Result<PublicKey> {
let Some(b58) = text.strip_prefix('z') else {
bail!("a multikey is base58btc, starting with z");
};
let bytes = bs58::decode(b58)
.into_vec()
.context("bad base58 in a multikey")?;
let (codec, key) = bytes.split_at(2.min(bytes.len()));
match <[u8; 2]>::try_from(codec) {
Ok(K256_CODEC) => Ok(PublicKey::K256(
k256::ecdsa::VerifyingKey::from_sec1_bytes(key).context("bad secp256k1 key")?,
)),
Ok(P256_CODEC) => Ok(PublicKey::P256(
p256::ecdsa::VerifyingKey::from_sec1_bytes(key).context("bad P-256 key")?,
)),
_ => bail!("unsupported key type in a multikey"),
}
}
/// Parses a DID document verification method: a `Multikey`, or one of
/// the older types whose multibase holds the bare key with no codec.
pub fn from_method(kind: &str, multibase: &str) -> Result<PublicKey> {
let raw = || -> Result<Vec<u8>> {
let Some(b58) = multibase.strip_prefix('z') else {
bail!("a multikey is base58btc, starting with z");
};
bs58::decode(b58).into_vec().context("bad base58 in a key")
};
match kind {
"EcdsaSecp256k1VerificationKey2019" => Ok(PublicKey::K256(
k256::ecdsa::VerifyingKey::from_sec1_bytes(&raw()?).context("bad secp256k1 key")?,
)),
"EcdsaSecp256r1VerificationKey2019" => Ok(PublicKey::P256(
p256::ecdsa::VerifyingKey::from_sec1_bytes(&raw()?).context("bad P-256 key")?,
)),
_ => PublicKey::from_multibase(multibase),
}
}
pub fn from_did_key(did: &str) -> Result<PublicKey> {
match did.strip_prefix("did:key:") {
Some(multibase) => PublicKey::from_multibase(multibase),
None => bail!("{did} is not a did:key"),
}
}
pub fn to_multibase(&self) -> String {
let codec = match self {
PublicKey::K256(_) => K256_CODEC,
PublicKey::P256(_) => P256_CODEC,
};
format!(
"z{}",
bs58::encode([&codec[..], &self.sec1()].concat()).into_string()
)
}
/// The key as SEC1 bytes, compressed.
pub fn sec1(&self) -> Vec<u8> {
match self {
PublicKey::K256(k) => k.to_encoded_point(true).as_bytes().to_vec(),
PublicKey::P256(k) => k.to_encoded_point(true).as_bytes().to_vec(),
}
}
pub fn to_did_key(&self) -> String {
format!("did:key:{}", self.to_multibase())
}
/// The JWT `alg` this key signs with.
pub fn alg(&self) -> &'static str {
match self {
PublicKey::K256(_) => "ES256K",
PublicKey::P256(_) => "ES256",
}
}
/// Checks a 64-byte `r||s` signature over SHA-256 of `message`. High-S
/// signatures are refused, as atproto requires.
pub fn verify(&self, message: &[u8], signature: &[u8]) -> Result<()> {
use k256::ecdsa::signature::Verifier;
let ok = match self {
PublicKey::K256(key) => k256::ecdsa::Signature::from_slice(signature)
.ok()
.filter(|s| s.normalize_s().is_none())
.is_some_and(|s| key.verify(message, &s).is_ok()),
PublicKey::P256(key) => p256::ecdsa::Signature::from_slice(signature)
.ok()
.filter(|s| s.normalize_s().is_none())
.is_some_and(|s| key.verify(message, &s).is_ok()),
};
if ok { Ok(()) } else { bail!("bad signature") }
}
}
/// A P-256 key grain signs with: a DPoP key that OAuth tokens are bound
/// to, or a confidential client's key that proves who the client is.
pub struct P256Key(p256::ecdsa::SigningKey);
impl P256Key {
pub fn generate() -> P256Key {
P256Key(p256::ecdsa::SigningKey::random(&mut rand_core::OsRng))
}
/// The 32-byte secret scalar, for a session that keeps the key with
/// the tokens bound to it.
pub fn to_bytes(&self) -> Vec<u8> {
self.0.to_bytes().to_vec()
}
pub fn from_bytes(bytes: &[u8]) -> Result<P256Key> {
Ok(P256Key(
p256::ecdsa::SigningKey::from_slice(bytes).context("bad P-256 secret key")?,
))
}
pub fn public(&self) -> PublicKey {
PublicKey::P256(*self.0.verifying_key())
}
/// The public key as a JWK, as a DPoP proof carries it.
pub fn jwk(&self) -> Value {
let point = self.0.verifying_key().to_encoded_point(false);
let coordinate = |c: Option<&p256::FieldBytes>| {
URL_SAFE_NO_PAD.encode(c.expect("an uncompressed point"))
};
serde_json::json!({
"kty": "EC",
"crv": "P-256",
"x": coordinate(point.x()),
"y": coordinate(point.y()),
})
}
/// A 64-byte, low-S `r||s` signature over SHA-256 of `message`.
pub fn sign(&self, message: &[u8]) -> Vec<u8> {
use p256::ecdsa::signature::Signer;
let sig: p256::ecdsa::Signature = self.0.sign(message);
sig.normalize_s().unwrap_or(sig).to_bytes().to_vec()
}
/// A compact JWT signed with this key, `alg` ES256.
pub fn sign_jwt(&self, header: &Value, claims: &impl Serialize) -> String {
let mut header = header.clone();
header["alg"] = "ES256".into();
let body = format!(
"{}.{}",
URL_SAFE_NO_PAD.encode(header.to_string()),
URL_SAFE_NO_PAD.encode(serde_json::to_vec(claims).expect("claims serialize"))
);
let sig = URL_SAFE_NO_PAD.encode(self.sign(body.as_bytes()));
format!("{body}.{sig}")
}
}
/// A JWT's header and claims, unverified.
pub fn jwt_parts(token: &str) -> Result<(Value, Value)> {
let mut parts = token.split('.');
let mut decode = |what: &str| -> Result<Value> {
let part = parts
.next()
.with_context(|| format!("a JWT without a {what}"))?;
let bytes = URL_SAFE_NO_PAD
.decode(part)
.with_context(|| format!("bad base64 in a JWT {what}"))?;
serde_json::from_slice(&bytes).with_context(|| format!("bad JSON in a JWT {what}"))
};
Ok((decode("header")?, decode("payload")?))
}
/// Verifies `token` against `key` and returns its claims. Checks the
/// signature, that `alg` matches the key, and `exp` against the clock.
pub fn verify_jwt<T: DeserializeOwned>(token: &str, key: &PublicKey) -> Result<T> {
let (header, claims) = jwt_parts(token)?;
if header["alg"].as_str() != Some(key.alg()) {
bail!(
"a JWT signed with {} where {} was expected",
header["alg"],
key.alg()
);
}
let (signed, sig) = token
.rsplit_once('.')
.context("a JWT without a signature")?;
let sig = URL_SAFE_NO_PAD
.decode(sig)
.context("bad base64 in a JWT signature")?;
key.verify(signed.as_bytes(), &sig)?;
let now = crate::keys::now_secs();
match claims["exp"].as_u64() {
Some(exp) if exp > now => {}
_ => bail!("an expired JWT"),
}
Ok(serde_json::from_value(claims)?)
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn multikeys_round_trip() {
let p = P256Key::generate().public();
assert_eq!(PublicKey::from_did_key(&p.to_did_key()).unwrap(), p);
assert!(p.to_multibase().starts_with("zDn"));
let k = PublicKey::K256(
*k256::ecdsa::SigningKey::random(&mut rand_core::OsRng).verifying_key(),
);
assert_eq!(PublicKey::from_multibase(&k.to_multibase()).unwrap(), k);
assert!(k.to_multibase().starts_with("zQ3s"));
assert_eq!((k.alg(), p.alg()), ("ES256K", "ES256"));
for bad in [
"x",
"z0OIl",
"z",
"z2",
&format!("z{}", bs58::encode([0xe7, 0x01, 1, 2]).into_string()),
&format!("z{}", bs58::encode([0x80, 0x24, 1, 2]).into_string()),
] {
assert!(PublicKey::from_multibase(bad).is_err(), "{bad}");
}
assert!(PublicKey::from_did_key("did:web:x").is_err());
// The older verification method types hold the bare key.
let bare = |k: &PublicKey| match k {
PublicKey::K256(k) => format!(
"z{}",
bs58::encode(k.to_encoded_point(false).as_bytes()).into_string()
),
PublicKey::P256(k) => format!(
"z{}",
bs58::encode(k.to_encoded_point(true).as_bytes()).into_string()
),
};
assert_eq!(
PublicKey::from_method("EcdsaSecp256k1VerificationKey2019", &bare(&k)).unwrap(),
k
);
assert_eq!(
PublicKey::from_method("EcdsaSecp256r1VerificationKey2019", &bare(&p)).unwrap(),
p
);
assert_eq!(
PublicKey::from_method("Multikey", &p.to_multibase()).unwrap(),
p
);
assert!(PublicKey::from_method("EcdsaSecp256k1VerificationKey2019", "x").is_err());
assert!(PublicKey::from_method("EcdsaSecp256k1VerificationKey2019", "z0").is_err());
assert!(PublicKey::from_method("EcdsaSecp256k1VerificationKey2019", "z2").is_err());
assert!(PublicKey::from_method("EcdsaSecp256r1VerificationKey2019", "z2").is_err());
}
#[test]
fn signatures() {
use k256::ecdsa::signature::Signer;
let key = P256Key::generate();
let again = P256Key::from_bytes(&key.to_bytes()).unwrap();
assert_eq!(again.public(), key.public());
assert!(P256Key::from_bytes(&[0; 32]).is_err());
let sig = key.sign(b"hello");
assert_eq!(sig.len(), 64);
key.public().verify(b"hello", &sig).unwrap();
assert!(key.public().verify(b"other", &sig).is_err());
assert!(key.public().verify(b"hello", b"short").is_err());
let k = k256::ecdsa::SigningKey::random(&mut rand_core::OsRng);
let s: k256::ecdsa::Signature = k.sign(b"m");
let low = s.normalize_s().unwrap_or(s);
let public = PublicKey::K256(*k.verifying_key());
public.verify(b"m", &low.to_bytes()).unwrap();
// The same signature with s flipped high verifies in ECDSA, and is refused.
let (r, s_low) = (low.r(), low.s());
let high = k256::ecdsa::Signature::from_scalars(r, -*s_low).unwrap();
assert!(public.verify(b"m", &high.to_bytes()).is_err());
}
#[test]
fn jwts() {
let key = P256Key::generate();
let exp = crate::keys::now_secs() + 60;
let token = key.sign_jwt(&json!({"typ": "JWT"}), &json!({"iss": "me", "exp": exp}));
let claims: Value = verify_jwt(&token, &key.public()).unwrap();
assert_eq!(claims["iss"], "me");
let (header, _) = jwt_parts(&token).unwrap();
assert_eq!(header["alg"], "ES256");
let other = P256Key::generate();
assert!(verify_jwt::<Value>(&token, &other.public()).is_err());
let k = PublicKey::K256(
*k256::ecdsa::SigningKey::random(&mut rand_core::OsRng).verifying_key(),
);
assert!(verify_jwt::<Value>(&token, &k).is_err());
let expired = key.sign_jwt(&json!({}), &json!({"exp": 1}));
assert!(verify_jwt::<Value>(&expired, &key.public()).is_err());
let (signed, _) = token.rsplit_once('.').unwrap();
assert!(verify_jwt::<Value>(&format!("{signed}.!!"), &key.public()).is_err());
assert!(verify_jwt::<Value>(signed, &key.public()).is_err());
for bad in ["", "a", "e30", "!!.x", "e30.!!", "bm90IGpzb24.e30"] {
assert!(jwt_parts(bad).is_err(), "{bad}");
}
}
/// A JWT signed with secp256k1, as most PDSes sign service tokens:
/// grain verifies them, though it signs none.
#[test]
fn k256_jwts() {
use k256::ecdsa::signature::Signer;
let k = k256::ecdsa::SigningKey::random(&mut rand_core::OsRng);
let public = PublicKey::K256(*k.verifying_key());
let sign = |header: Value, claims: Value| {
let body = format!(
"{}.{}",
URL_SAFE_NO_PAD.encode(header.to_string()),
URL_SAFE_NO_PAD.encode(claims.to_string())
);
let sig: k256::ecdsa::Signature = k.sign(body.as_bytes());
let sig = sig.normalize_s().unwrap_or(sig);
format!("{body}.{}", URL_SAFE_NO_PAD.encode(sig.to_bytes()))
};
let exp = crate::keys::now_secs() + 60;
let token = sign(
json!({"alg": "ES256K"}),
json!({"iss": "did:plc:a", "exp": exp}),
);
let claims: Value = verify_jwt(&token, &public).unwrap();
assert_eq!(claims["iss"], "did:plc:a");
// Its key named as a did:key verifies it too.
let named = PublicKey::from_did_key(&public.to_did_key()).unwrap();
verify_jwt::<Value>(&token, &named).unwrap();
// Signed by the key, but claiming ES256: the alg is the key's.
let mislabelled = sign(json!({"alg": "ES256"}), json!({"exp": exp}));
assert!(verify_jwt::<Value>(&mislabelled, &public).is_err());
let no_exp = sign(json!({"alg": "ES256K"}), json!({"iss": "did:plc:a"}));
assert!(verify_jwt::<Value>(&no_exp, &public).is_err());
}
}
crates/grain/src/env.rsnew+218−0
View file
@@ -0,0 +1,218 @@
//! Process configuration, read once from the environment.
use std::path::PathBuf;
use std::time::Duration;
/// How a handle becomes a DID, tried in order.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum HandleResolver {
/// `<scheme>://<handle>/.well-known/atproto-did`.
WellKnown { scheme: String },
/// `com.atproto.identity.resolveHandle` on a service.
Xrpc { service: String },
}
#[derive(Clone, Debug)]
pub struct Env {
pub cache_dir: PathBuf,
pub plc_url: String,
pub handle_resolvers: Vec<HandleResolver>,
/// Scheme for did:web documents; tests serve them over plain http.
pub did_web_scheme: String,
/// Certificates to trust beside the usual roots, PEM: a local network
/// behind a proxy with a CA of its own.
pub ca_certs: Option<PathBuf>,
/// Skips DID resolution: every identity is served by this PDS.
pub service: Option<String>,
pub identity_ttl: Duration,
pub timeout: Duration,
}
impl Env {
pub fn from_process() -> Env {
Env::from_vars(|name| std::env::var(name).ok())
}
pub fn from_vars(get: impl Fn(&str) -> Option<String>) -> Env {
let home = get("HOME").unwrap_or_else(|| ".".into());
let dir = |own: &str, xdg: &str, fallback: &str| {
get(own).map(PathBuf::from).unwrap_or_else(|| {
get(xdg)
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from(&home).join(fallback))
.join("grain")
})
};
let num = |name: &str, default: u64| {
get(name)
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(default)
};
let handle_resolvers = match get("GRAIN_HANDLE_RESOLVER") {
Some(list) => list
.split(',')
.filter(|s| !s.is_empty())
.map(|s| match s.strip_prefix("well-known:") {
Some(scheme) => HandleResolver::WellKnown {
scheme: scheme.into(),
},
None => HandleResolver::Xrpc { service: s.into() },
})
.collect(),
None => vec![
HandleResolver::WellKnown {
scheme: "https".into(),
},
HandleResolver::Xrpc {
service: "https://public.api.bsky.app".into(),
},
],
};
Env {
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,
did_web_scheme: get("GRAIN_DID_WEB_SCHEME").unwrap_or_else(|| "https".into()),
ca_certs: get("GRAIN_CA_CERTS")
.filter(|v| !v.is_empty())
.map(PathBuf::from),
service: get("GRAIN_SERVICE").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)),
}
}
}
/// Configuration plus the one HTTP client a process shares, so every request
/// to a PDS reuses its connection.
#[derive(Clone)]
pub struct Ctx {
pub env: Env,
pub http: reqwest::Client,
}
impl Ctx {
pub fn new(env: Env) -> Ctx {
let mut http = reqwest::Client::builder();
for cert in env.ca_certs.iter().flat_map(|path| extra_roots(path)) {
http = http.add_root_certificate(cert);
}
let http = http
.user_agent(concat!("grain/", env!("CARGO_PKG_VERSION")))
.connect_timeout(Duration::from_secs(10))
.timeout(env.timeout)
// Under the 5s Node servers keep an idle connection by default,
// as a PLC directory does: past it, one can close under the
// request reusing it. A PDS keeps one 90s, which ties with any
// longer wait here.
.pool_idle_timeout(Duration::from_secs(4))
.tcp_nodelay(true)
.build()
.expect("the HTTP client configuration is static");
Ctx { env, http }
}
}
/// The certificates in the PEM file at `path`; none, said once, if it
/// can't be read, so a mistyped path fails as an untrusted server would.
fn extra_roots(path: &std::path::Path) -> Vec<reqwest::Certificate> {
let roots = std::fs::read(path)
.map_err(anyhow::Error::from)
.and_then(|pem| Ok(reqwest::Certificate::from_pem_bundle(&pem)?));
roots.unwrap_or_else(|err| {
eprintln!("grain: GRAIN_CA_CERTS {}: {err}", path.display());
Vec::new()
})
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
fn env(vars: &[(&str, &str)]) -> Env {
let map: HashMap<String, String> = vars
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
Env::from_vars(|k| map.get(k).cloned())
}
#[test]
fn defaults() {
let e = env(&[("HOME", "/home/a")]);
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");
assert_eq!(
e.handle_resolvers,
vec![
HandleResolver::WellKnown {
scheme: "https".into()
},
HandleResolver::Xrpc {
service: "https://public.api.bsky.app".into()
}
]
);
assert_eq!(e.identity_ttl, Duration::from_secs(3600));
assert_eq!(e.timeout, Duration::from_secs(60));
assert!(e.service.is_none());
assert!(e.ca_certs.is_none());
let bare = env(&[]);
assert_eq!(bare.cache_dir, PathBuf::from("./.cache/grain"));
}
#[test]
fn overrides() {
let e = env(&[
("XDG_CACHE_HOME", "/x"),
("GRAIN_HANDLE_RESOLVER", "well-known:http,http://pds,"),
("GRAIN_PLC_URL", "http://plc"),
("GRAIN_DID_WEB_SCHEME", "http"),
("GRAIN_SERVICE", "http://pds"),
("GRAIN_IDENTITY_TTL_SECS", "5"),
("GRAIN_TIMEOUT_SECS", "soon"),
]);
assert_eq!(e.cache_dir, PathBuf::from("/x/grain"));
assert_eq!(e.plc_url, "http://plc");
assert_eq!(e.did_web_scheme, "http");
assert_eq!(e.service.as_deref(), Some("http://pds"));
assert_eq!(e.identity_ttl, Duration::from_secs(5));
assert_eq!(e.timeout, Duration::from_secs(60));
assert_eq!(
e.handle_resolvers,
vec![
HandleResolver::WellKnown {
scheme: "http".into()
},
HandleResolver::Xrpc {
service: "http://pds".into()
}
]
);
assert_eq!(
env(&[("GRAIN_CACHE_DIR", "/c")]).cache_dir,
PathBuf::from("/c")
);
assert!(env(&[("GRAIN_SERVICE", "")]).service.is_none());
let _ = Ctx::new(e);
}
/// A throwaway self-signed CA, only for reading back.
const TEST_CA: &str = "-----BEGIN CERTIFICATE-----\nMIIBjTCCATOgAwIBAgIUW5Z0OJcr5A6Uzy+D+Jc6OM3cDt8wCgYIKoZIzj0EAwIw\nGzEZMBcGA1UEAwwQa25vdGxlc3MgdGVzdCBDQTAgFw0yNjEwMjQwMDIzMTRaGA8y\nMTI2MDkzMDAwMjMxNFowGzEZMBcGA1UEAwwQa25vdGxlc3MgdGVzdCBDQTBZMBMG\nByqGSM49AgEGCCqGSM49AwEHA0IABJyz7he0x9D3F54sUTSlH1cmYOJFqVKhOxg/\nhms5rbXMyiARmltNCbVSSq//HJd/YNouFdBOXNnOuot4qukk3iajUzBRMB0GA1Ud\nDgQWBBQGfpQL8IXKrgaOIycZIqo3E3pdDzAfBgNVHSMEGDAWgBQGfpQL8IXKrgaO\nIycZIqo3E3pdDzAPBgNVHRMBAf8EBTADAQH/MAoGCCqGSM49BAMCA0gAMEUCIBeZ\nk3lRccBXvvSJxmcoAKKnalqoI/600ObPgeVxRkSyAiEAsBLjRbTAWH5dOVShu82A\n++dP/h0UHb5o4K49o69ITM8=\n-----END CERTIFICATE-----\n";
#[test]
fn extra_roots_are_read_from_a_pem_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("ca.pem");
std::fs::write(&path, TEST_CA).unwrap();
assert_eq!(extra_roots(&path).len(), 1);
// A mistyped path trusts nothing more.
assert!(extra_roots(&dir.path().join("missing.pem")).is_empty());
let e = env(&[("GRAIN_CA_CERTS", path.to_str().unwrap())]);
assert_eq!(e.ca_certs.as_deref(), Some(path.as_path()));
Ctx::new(e);
assert!(env(&[("GRAIN_CA_CERTS", "")]).ca_certs.is_none());
}
}
crates/grain/src/identity.rsnew+636−0
View file
636 lines changed: not loaded until asked for.
crates/grain/src/keys.rsnew+60−0
View file
@@ -0,0 +1,60 @@
//! The clock, and record keys made from it: TIDs, the timestamp keys
//! atproto names records with.
use crate::base32;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
static LAST_MICROS: AtomicU64 = AtomicU64::new(0);
pub fn now_micros() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("the clock is after 1970")
.as_micros() as u64
}
/// Seconds since the epoch, as a JWT's `iat` and `exp` count them.
pub fn now_secs() -> u64 {
now_micros() / 1_000_000
}
/// A TID that sorts after every TID this process made before it.
pub fn tid() -> String {
let now = now_micros();
let micros = LAST_MICROS
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |last| {
Some(now.max(last + 1))
})
.map(|last| now.max(last + 1))
.expect("the update closure always returns Some");
tid_from(micros, u64::from(std::process::id()))
}
pub fn tid_from(micros: u64, clock: u64) -> String {
let micros = micros & ((1 << 53) - 1);
base32::encode_u64(base32::SORTABLE, (micros << 10) | (clock & 1023))
}
/// The timestamp in a TID, or `None` if `tid` is not one.
pub fn tid_micros(tid: &str) -> Option<u64> {
base32::decode_u64(base32::SORTABLE, tid).map(|v| v >> 10)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn tids_increase_and_decode() {
let a = tid();
let b = tid();
assert!(b > a);
assert_eq!(a.len(), 13);
let micros = tid_micros(&a).unwrap();
assert!(micros.abs_diff(now_micros()) < 10_000_000);
assert_eq!(tid_micros(&tid_from(42, 7)), Some(42));
assert_eq!(tid_micros("nope"), None);
assert!(now_secs().abs_diff(now_micros() / 1_000_000) <= 1);
}
}
crates/grain/src/lib.rs+7−0
View file
@@ -5,4 +5,11 @@
//! over as grain needs them rather than all at once.
pub mod base32;
pub mod crypto;
pub mod drisl;
pub mod env;
pub mod identity;
pub mod keys;
pub mod store;
pub mod telemetry;
pub mod xrpc;
crates/grain/src/store.rsnew+75−0
View file
@@ -0,0 +1,75 @@
//! Small JSON files under the config and cache directories.
use anyhow::{Context, Result};
use serde::Serialize;
use serde::de::DeserializeOwned;
use std::path::Path;
/// 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.
pub fn read<T: DeserializeOwned>(path: &Path) -> Option<T> {
let bytes = std::fs::read(path).ok()?;
serde_json::from_slice(&bytes).ok()
}
/// Writes through a temporary file and a rename, so a reader never sees half
/// a file. `private` restricts it to the owner.
pub fn write<T: Serialize>(path: &Path, value: &T, private: bool) -> Result<()> {
write_bytes(path, &serde_json::to_vec_pretty(value)?, private)
}
pub fn write_bytes(path: &Path, bytes: &[u8], private: bool) -> Result<()> {
let dir = path.parent().context("a store path has a parent")?;
std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
let tmp = dir.join(format!(
".{}.{}.tmp",
path.file_name().and_then(|n| n.to_str()).unwrap_or("store"),
crate::keys::tid()
));
std::fs::write(&tmp, bytes).with_context(|| format!("writing {}", tmp.display()))?;
#[cfg(unix)]
if private {
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o600))?;
}
#[cfg(not(unix))]
let _ = private;
std::fs::rename(&tmp, path).with_context(|| format!("writing {}", path.display()))?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
#[test]
fn round_trip_and_failures() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("a/b.json");
assert_eq!(read::<BTreeMap<String, u8>>(&path), None);
let mut m = BTreeMap::new();
m.insert("x".to_string(), 1u8);
write(&path, &m, true).unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mode = std::fs::metadata(&path).unwrap().permissions().mode();
assert_eq!(mode & 0o777, 0o600);
}
write(&path, &m, false).unwrap();
assert_eq!(read::<BTreeMap<String, u8>>(&path), Some(m.clone()));
std::fs::write(&path, b"{").unwrap();
assert_eq!(read::<BTreeMap<String, u8>>(&path), None);
assert!(write(Path::new("/"), &m, false).is_err());
let blocked = dir.path().join("file");
std::fs::write(&blocked, b"").unwrap();
assert!(write(&blocked.join("x.json"), &m, false).is_err());
// A directory where the file should go: the rename fails.
let occupied = dir.path().join("occupied");
std::fs::create_dir_all(occupied.join("inside")).unwrap();
assert!(write(&occupied, &m, false).is_err());
// A name too long for the file system: writing the copy fails.
assert!(write(&dir.path().join("n".repeat(300)), &m, false).is_err());
}
}
crates/grain/src/telemetry.rsnew+322−0
View file
@@ -0,0 +1,322 @@
//! The calls grain makes to other servers, each logged as it is answered,
//! and the trace a request being served carries into them.
//!
//! Field names follow OpenTelemetry's semantic conventions
//! (`server.address`, `http.response.status_code`), so the lines can be
//! shipped to a backend as they are. A call made while a request is served
//! carries its `trace_id`, handed on as a W3C `traceparent`.
//!
//! Nothing here logs a URL's query, a header, or a body.
use rand_core::RngCore;
use reqwest::header::{HeaderName, HeaderValue};
use std::fmt::Write as _;
use std::time::{Duration, Instant};
use tracing::Level;
/// Milliseconds, as a log line gives a duration.
pub fn millis(d: Duration) -> u64 {
u64::try_from(d.as_millis()).unwrap_or(u64::MAX)
}
/// The slowest a call is before it is logged at warn.
pub const SLOW: Duration = Duration::from_secs(1);
tokio::task_local! {
static TRACE: String;
}
/// The trace the task is in: the request being served.
pub fn trace_id() -> Option<String> {
TRACE.try_with(String::clone).ok()
}
/// Runs `f` in the trace `id`.
pub async fn traced<F: Future>(id: String, f: F) -> F::Output {
TRACE.scope(id, f).await
}
/// The trace a `traceparent` header names, if it is a valid one; else a
/// new trace.
pub fn trace_from(traceparent: Option<&str>) -> String {
let given = traceparent.and_then(trace_of);
given.unwrap_or_else(|| hex(16))
}
fn trace_of(traceparent: &str) -> Option<String> {
let id = traceparent.split('-').nth(1)?;
let hex = id.len() == 32 && id.bytes().all(|b| b.is_ascii_hexdigit());
(hex && id.bytes().any(|b| b != b'0')).then(|| id.to_ascii_lowercase())
}
fn hex(bytes: usize) -> String {
let mut buf = vec![0u8; bytes];
rand_core::OsRng.fill_bytes(&mut buf);
buf.iter().fold(String::new(), |mut s, b| {
let _ = write!(s, "{b:02x}");
s
})
}
/// A path as a log line gives it: no query, and not too long.
pub fn short(path: &str) -> String {
path.chars().take(120).collect()
}
/// Sends `req`, the `resend`th time it is sent again, and logs it: at
/// debug, or warn if it failed or was slow, by host and path, never its
/// query or headers. In a trace, it hands the trace on as `traceparent`.
pub async fn send(req: reqwest::RequestBuilder, resend: u32) -> reqwest::Result<reqwest::Response> {
let (client, req) = req.build_split();
let mut req = req?;
if let Some(id) = trace_id() {
let parent = format!("00-{id}-{}-01", hex(8));
let parent = HeaderValue::try_from(parent).expect("hex is a header value");
let name = HeaderName::from_static("traceparent");
req.headers_mut().insert(name, parent);
}
let call = Call {
host: req.url().host_str().unwrap_or_default().to_string(),
method: req.method().to_string(),
path: short(req.url().path()),
resend,
};
let started = Instant::now();
let res = client.execute(req).await;
let status = res.as_ref().ok().map(|r| r.status().as_u16());
let error = res.as_ref().err().map(error_type);
call.done(status, error.as_deref(), started.elapsed());
res
}
struct Call {
host: String,
method: String,
path: String,
resend: u32,
}
impl Call {
fn done(&self, status: Option<u16>, error: Option<&str>, took: Duration) {
let ms = millis(took);
let server_error = status.filter(|s| *s >= 500).map(|s| s.to_string());
let error = error.or(server_error.as_deref());
let slow = took >= SLOW;
let rpc = self.path.strip_prefix("/xrpc/");
let resent = (self.resend > 0).then_some(self.resend);
let trace = trace_id();
macro_rules! line {
($level:expr) => {
tracing::event!(
target: "grain::outgoing",
$level,
server.address = self.host.as_str(),
http.request.method = self.method.as_str(),
url.path = self.path.as_str(),
rpc.method = rpc,
http.response.status_code = status,
error.type = error,
http.request.resend_count = resent,
duration_ms = ms,
trace_id = trace.as_deref(),
"outgoing"
)
};
}
match error.is_some() || slow {
true => line!(Level::WARN),
false => line!(Level::DEBUG),
}
}
}
/// What kind of failure `err` is, and why, without the URL it names.
fn error_type(err: &reqwest::Error) -> String {
let kind = match () {
_ if err.is_timeout() => "timeout",
_ if err.is_connect() => "connect",
_ if err.is_body() || err.is_decode() => "body",
_ => "request",
};
let mut said = kind.to_string();
let mut source = std::error::Error::source(err);
while let Some(cause) = source {
let _ = write!(said, ": {cause}");
source = cause.source();
}
said
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use std::sync::{Arc, Mutex};
use tracing_subscriber::layer::SubscriberExt;
/// What is logged while `f` runs, at every level, as JSON lines.
pub(crate) fn logged(f: impl FnOnce()) -> Vec<serde_json::Value> {
let lines = Arc::new(Mutex::new(Vec::new()));
let kept = lines.clone();
let writer = move || Kept(kept.clone());
let layer = tracing_subscriber::fmt::layer()
.json()
.flatten_event(true)
.with_writer(writer);
let subscriber = tracing_subscriber::registry()
.with(tracing_subscriber::EnvFilter::new("trace"))
.with(layer);
tracing::subscriber::with_default(subscriber, f);
let text = String::from_utf8(lines.lock().unwrap().clone()).unwrap();
text.lines()
.map(|l| serde_json::from_str(l).unwrap())
.collect()
}
struct Kept(Arc<Mutex<Vec<u8>>>);
impl std::io::Write for Kept {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
#[test]
fn settings() {
assert_eq!(millis(Duration::from_secs(2)), 2000);
assert_eq!(short(&"a".repeat(500)).len(), 120);
std::io::Write::flush(&mut Kept(Arc::default())).unwrap();
}
#[test]
fn traces() {
let id = "4bf92f3577b34da6a3ce929d0e0e4736";
let parent = format!("00-{id}-00f067aa0ba902b7-01");
assert_eq!(trace_from(Some(&parent)), id);
assert_eq!(trace_from(Some(&parent.to_uppercase())), id);
for bad in ["junk", "00-xyz-00f067aa0ba902b7-01"] {
assert_ne!(trace_from(Some(bad)), id);
}
let zeros = format!("00-{}-00f067aa0ba902b7-01", "0".repeat(32));
assert_eq!(trace_from(Some(&zeros)).len(), 32);
assert_eq!(trace_from(None).len(), 32);
assert_eq!(trace_id(), None);
let rt = tokio::runtime::Builder::new_current_thread()
.build()
.unwrap();
let within = rt.block_on(traced(id.into(), async { trace_id() }));
assert_eq!(within.as_deref(), Some(id));
}
/// A server that answers each request with `status`, and the headers
/// of the last it was sent.
pub(crate) fn answering(status: &'static str) -> (String, Arc<Mutex<String>>) {
use std::io::{Read, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let asked = Arc::new(Mutex::new(String::new()));
let kept = asked.clone();
std::thread::spawn(move || {
for stream in listener.incoming() {
let mut stream = stream.unwrap();
let mut buf = [0u8; 4096];
let n = stream.read(&mut buf).unwrap();
*kept.lock().unwrap() = String::from_utf8_lossy(&buf[..n]).into_owned();
let answer =
format!("HTTP/1.1 {status}\r\ncontent-length: 0\r\nconnection: close\r\n\r\n");
let _ = stream.write_all(answer.as_bytes());
}
});
(url, asked)
}
#[test]
fn outgoing_calls() {
let (ok, asked) = answering("200 OK");
let (broken, _) = answering("503 Service Unavailable");
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let http = reqwest::Client::new();
let lines = logged(|| {
rt.block_on(async {
let xrpc = format!("{ok}/xrpc/com.atproto.sync.getRepo?did=did:plc:x&secret=1");
let id = "4bf92f3577b34da6a3ce929d0e0e4736".to_string();
let sent = traced(id, send(http.get(&xrpc), 0)).await;
assert_eq!(sent.unwrap().status(), 200);
let sent = send(http.get(&broken), 1).await;
assert_eq!(sent.unwrap().status(), 503);
let refused = send(http.get("http://127.0.0.1:9/x"), 0).await;
assert!(refused.is_err());
let invalid = send(http.get("not a url"), 0).await;
assert!(invalid.is_err());
})
});
assert!(
asked
.lock()
.unwrap()
.contains("traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-")
);
let calls: Vec<&serde_json::Value> = lines
.iter()
.filter(|l| l["message"] == "outgoing")
.collect();
assert_eq!(calls.len(), 3);
assert_eq!(calls[0]["level"], "DEBUG");
assert_eq!(calls[0]["rpc.method"], "com.atproto.sync.getRepo");
assert_eq!(calls[0]["url.path"], "/xrpc/com.atproto.sync.getRepo");
assert_eq!(calls[0]["server.address"], "127.0.0.1");
assert_eq!(calls[0]["trace_id"], "4bf92f3577b34da6a3ce929d0e0e4736");
assert!(!lines.iter().any(|l| l.to_string().contains("secret")));
assert_eq!(calls[1]["level"], "WARN");
assert_eq!(calls[1]["error.type"], "503");
assert_eq!(calls[1]["http.request.resend_count"], 1);
assert!(
calls[2]["error.type"]
.as_str()
.unwrap()
.starts_with("connect")
);
}
#[test]
fn slow_calls_warn() {
let call = Call {
host: "pds.test".into(),
method: "GET".into(),
path: "/x".into(),
resend: 0,
};
let lines = logged(|| call.done(Some(200), None, SLOW));
assert_eq!(lines[0]["level"], "WARN");
assert_eq!(lines[0]["duration_ms"], 1000);
assert!(lines[0].get("rpc.method").is_none());
}
#[test]
fn error_types() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
let (url, _) = answering("200 OK");
let http = reqwest::Client::new();
// Asked of a listener that never answers, so the timeout always
// comes first: a local 200 could beat even a zero one.
let silent = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let never = format!("http://{}/", silent.local_addr().unwrap());
let timeout = rt.block_on(async { http.get(&never).timeout(Duration::ZERO).send().await });
let timeout = timeout.unwrap_err();
assert!(error_type(&timeout).starts_with("timeout"));
let built = http.get(&url).header("bad\nname", "x").build().unwrap_err();
assert!(error_type(&built).starts_with("request"));
let res = rt.block_on(http.get(&url).send()).unwrap();
let body = rt.block_on(res.json::<serde_json::Value>()).unwrap_err();
assert!(error_type(&body).starts_with("body"));
}
}
crates/grain/src/xrpc.rsnew+450−0
View file
450 lines changed: not loaded until asked for.
crates/grain/Cargo.toml+15−0
View file
@@ -7,6 +7,21 @@ license.workspace = true
rust-version.workspace = true
[dependencies]
anyhow.workspace = true
base64.workspace = true
bs58.workspace = true
bytes.workspace = true
ciborium.workspace = true
futures.workspace = true
k256.workspace = true
p256.workspace = true
rand_core.workspace = true
reqwest.workspace = true
serde.workspace = true
serde_json.workspace = true
tokio.workspace = true
tracing.workspace = true
[dev-dependencies]
tempfile.workspace = true
tracing-subscriber.workspace = true
Cargo.lock+400−5
View file
405 lines changed: not loaded until asked for.
Cargo.toml+7−0
View file
@@ -13,7 +13,13 @@ grain = { path = "crates/grain" }
anyhow = "1"
axum = "0.8"
base64 = "0.22"
bs58 = "0.5"
bytes = "1"
ciborium = "0.2"
futures = "0.3"
k256 = { version = "0.13", features = ["ecdsa"] }
p256 = { version = "0.13", features = ["ecdsa"] }
rand_core = { version = "0.6", features = ["getrandom"] }
reqwest = { version = "0.12", default-features = false, features = [
"json",
"rustls-tls",
@@ -22,6 +28,7 @@ reqwest = { version = "0.12", default-features = false, features = [
] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tempfile = "3"
tokio = { version = "1", features = ["rt", "rt-multi-thread", "macros", "net", "signal", "time"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", default-features = false, features = [