~ knotless

chadtmiller.com/knotless

Git hosted in your atproto account, on any stock PDS, fast enough for agents.

git clone knotless://chadtmiller.com/knotless

needs the knotless git helper. set it up

Log in with OAuth; delegate to gateways as an OAuth client

knotless login opens the browser and logs in as a loopback OAuth client,
with DPoP on every request, refresh under a lock, and revocation on
logout. An app password still logs in.

A gateway is now a confidential OAuth client of its own, so delegating
no longer needs an app password: knotless delegate has the gateway push
an authorization request, the owner approves it in a browser, and the
gateway keeps a session scoped to knotless records.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Chad Miller committed on Oct 21, 2026 parent 4de1da8 commit 5037862
Showing 45 changed files with 1821 additions and 262 deletions
README.md+8−7
@@ -13,7 +13,7 @@ run.
```sh
cargo install --path crates/knotless # git-remote-knotless + knotless
knotless login alice.bsky.social # app password, or KNOTLESS_PASSWORD
knotless login alice.bsky.social # OAuth, in your browser
git push knotless://alice.bsky.social/my-project main
git clone knotless://alice.bsky.social/my-project
@@ -109,8 +109,8 @@ knotless members knotless://alice.test/secret
A private repository is a permissioned space,
`at://<did>/space/dev.knotless.space/<name>`, holding the same records. Its
members read with a space credential, which needs a full login rather than
an app password, and which is cached until shortly before it expires. Each
members read with a space credential, which needs an OAuth login rather
than an app password, and which is cached until shortly before it expires. Each
member writes their issues and comments into their own repo in the space,
so the conversation stays as private as the code. Spaces are in the
reference PDS's permissioned-data alpha; on a PDS without them, every
@@ -255,7 +255,8 @@ rebuilds the index.
- **An appview**: `dev.knotless.listRepos` and `dev.knotless.getForge`,
from a SQLite index of what Jetstream or the firehose reports.
- **A gateway**: `dev.knotless.pushRefs`, `dev.knotless.createCommit` and
`dev.knotless.putDelegation`, for collaborators.
`dev.knotless.putDelegation` (the gateway is an OAuth client the owner
approves in a browser), for collaborators.
- **A moderation service**, as an appview with a public URL:
`com.atproto.moderation.createReport` for anyone,
`com.atproto.admin.updateSubjectStatus` and `dev.knotless.listReports`
@@ -293,7 +294,9 @@ on one small VPS.
| `knotless compact <url>` | fold packs into one |
| `knotless mirror <url>` | sync the cache mirror and print its path |
Environment: `KNOTLESS_PASSWORD`, `KNOTLESS_IDENTIFIER`, `KNOTLESS_SERVICE`
Environment: `KNOTLESS_PASSWORD` (an app password, instead of OAuth),
`KNOTLESS_IDENTIFIER`, `KNOTLESS_BROWSER` (the command an OAuth login opens
its page with), `KNOTLESS_SERVICE`
(skip DID resolution), `KNOTLESS_PLC_URL`, `KNOTLESS_HANDLE_RESOLVER`,
`KNOTLESS_APPVIEW`, `KNOTLESS_MODERATOR` (the appview's),
`KNOTLESS_ADMIN_PASSWORD`, `KNOTLESS_DID_WEB_SCHEME` (`https`),
@@ -352,7 +355,5 @@ the PLC directory makes new for each world, are masked.
## Not yet
- OAuth: logins are app passwords, and reading a private repository takes a
full login.
- Verifying refs against signed repository commits.
- The `dev.knotless` NSIDs assume control of `knotless.dev`.
crates/knotless-serve/src/gateway.rs+152−15
@@ -1,7 +1,9 @@
//! The gateway: writes on owners' behalf, for collaborators and for batches.
//!
//! An owner hands over an app password (`dev.knotless.putDelegation`). After
//! that, anyone on a repository's collaborator list can push
//! The gateway is an OAuth client of its owners' PDSes. An owner delegates
//! (`dev.knotless.putDelegation`) by approving the gateway's authorization
//! request, which grants it the knotless records and blobs and nothing else
//! of the account, and the gateway keeps the session. After that, anyone on a repository's collaborator list can push
//! (`dev.knotless.pushRefs`) or commit (`dev.knotless.createCommit`) with a service
//! token from their own PDS, and the gateway writes as the owner.
//!
@@ -13,23 +15,26 @@
use crate::{App, AppError};
use axum::Json;
use axum::body::Bytes;
use axum::extract::State;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode, header};
use knotless::commit::{self, Committed, Spec};
use knotless::gateway::{
self, COMMIT, CommitRequest, CommitResponse, DELEGATE, PUSH, PushResults, RefResult,
self, AWAIT_DELEGATION, COMMIT, CommitRequest, CommitResponse, DELEGATE, PUSH, PushResults,
RefResult,
};
use knotless::mirror::Mirror;
use knotless::push::{self, Update};
use knotless::remote::{Remote, Snapshot};
use knotless::session::{self, Session};
use knotless::url::RepoUrl;
use knotless::{identity, oauth};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use tokio::sync::oneshot;
use std::time::{Duration, Instant};
use tokio::sync::{oneshot, watch};
/// How many times a batch is rebuilt on a branch that moved under it.
const RETRIES: u32 = 8;
@@ -41,10 +46,34 @@ pub struct Gateway {
pub writes: AtomicU64,
/// Batches rebuilt because the branch moved under them.
pub rebuilt: AtomicU64,
/// Where browsers and authorization servers reach the gateway as an
/// OAuth client: the endpoint, unless that is served elsewhere too.
pub oauth_url: String,
/// How long `getDelegation` waits for an approval before it says to
/// ask again.
pub wait: Duration,
repos: Mutex<HashMap<(String, String), Lock>>,
branches: Mutex<HashMap<(String, String, String), Arc<Branch>>>,
/// Delegations waiting for their owners' approval, by state.
delegations: Mutex<HashMap<String, Delegation>>,
}
type Outcome = watch::Sender<Option<Result<(), String>>>;
struct Delegation {
did: String,
started: Instant,
/// Taken by the browser's return.
request: Option<oauth::Request>,
done: Arc<Outcome>,
}
/// Where the gateway's OAuth client metadata is, its client ID.
pub const METADATA: &str = "/oauth-client-metadata.json";
const CALLBACK: &str = "/oauth/callback";
/// How long a delegation waits for approval before it is dropped.
const ABANDONED: Duration = Duration::from_secs(15 * 60);
/// Held while a repository is written.
type Lock = Arc<tokio::sync::Mutex<()>>;
@@ -76,10 +105,23 @@ impl Gateway {
endpoint: endpoint.to_string(),
writes: Default::default(),
rebuilt: Default::default(),
oauth_url: endpoint.to_string(),
wait: Duration::from_secs(25),
repos: Mutex::default(),
branches: Mutex::default(),
delegations: Mutex::default(),
})
}
/// The gateway as an OAuth client, signing with the key in its config
/// directory.
fn client(&self, ctx: &knotless::env::Ctx) -> Result<oauth::Client, AppError> {
let key = oauth::client_key(ctx).map_err(AppError::internal)?;
Ok(oauth::Client::confidential(
&format!("{}{METADATA}", self.oauth_url),
key,
))
}
}
fn forbidden(msg: impl Into<String>) -> AppError {
@@ -131,22 +173,117 @@ pub async fn did_document(State(app): State<Arc<App>>) -> Result<Json<Value>, Ap
})))
}
#[derive(Deserialize)]
pub struct Delegation {
password: String,
/// The gateway's OAuth client metadata, which its client ID names.
pub async fn client_metadata(State(app): State<Arc<App>>) -> Result<Json<Value>, AppError> {
let gw = gateway(&app)?;
let client = gw.client(&app.ctx)?;
Ok(Json(json!({
"client_id": client.id,
"client_name": "knotless gateway",
"application_type": "web",
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"redirect_uris": [format!("{}{CALLBACK}", gw.oauth_url)],
"scope": gateway::SCOPE,
"token_endpoint_auth_method": "private_key_jwt",
"token_endpoint_auth_signing_alg": "ES256",
"dpop_bound_access_tokens": true,
"jwks": client.jwks(),
})))
}
/// Takes an owner's app password, and keeps the session it opens.
/// Starts delegating the caller's writes here: asks their PDS for a grant,
/// and answers with the page where they approve it.
pub async fn delegate(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(body): Json<Delegation>,
) -> Result<Json<Value>, AppError> {
let did = caller(&app, &headers, DELEGATE).await?;
session::login(&app.ctx, &did, &body.password)
let gw = gateway(&app)?;
let id = identity::resolve(&app.ctx, &did)
.await
.map_err(AppError::upstream)?;
let back = format!("{}{CALLBACK}", gw.oauth_url);
let client = gw.client(&app.ctx)?;
let request = oauth::authorize(&app.ctx, &client, id, &did, &back, gateway::SCOPE)
.await
.map_err(|e| forbidden(format!("that password does not log in as {did}: {e:#}")))?;
Ok(Json(json!({})))
.map_err(AppError::upstream)?;
let out = json!({ "url": request.url, "state": request.state });
let mut delegations = gw.delegations.lock().unwrap();
delegations.retain(|_, d| d.started.elapsed() < ABANDONED);
delegations.insert(
request.state.clone(),
Delegation {
did,
started: Instant::now(),
request: Some(request),
done: Arc::new(watch::channel(None).0),
},
);
Ok(Json(out))
}
/// Where the owner's browser comes back to: trades the code for the
/// session the gateway writes with.
pub async fn callback(
State(app): State<Arc<App>>,
Query(params): Query<HashMap<String, String>>,
) -> Result<String, AppError> {
let gw = gateway(&app)?;
let state = params.get("state").map_or("", String::as_str);
let waiting = gw
.delegations
.lock()
.unwrap()
.get_mut(state)
.and_then(|d| Some((d.request.take()?, d.done.clone())));
let Some((request, done)) = waiting else {
return Err(bad("no delegation is waiting for this"));
};
let result = match request
.redeem(&app.ctx, &gw.client(&app.ctx)?, &params)
.await
{
Ok(session) => session::save(&app.ctx, &session),
Err(err) => Err(err),
};
let page = match &result {
Ok(()) => "The gateway now writes for you. You can close this tab.".to_string(),
Err(err) => format!("The gateway was not delegated to: {err:#}"),
};
done.send_replace(Some(result.map_err(|e| format!("{e:#}"))));
Ok(page)
}
#[derive(Deserialize)]
pub struct Awaited {
state: String,
}
/// Waits a while for the caller's delegation to be approved.
pub async fn await_delegation(
State(app): State<Arc<App>>,
headers: HeaderMap,
Query(query): Query<Awaited>,
) -> Result<Json<Value>, AppError> {
let did = caller(&app, &headers, AWAIT_DELEGATION).await?;
let gw = gateway(&app)?;
let mut done = gw
.delegations
.lock()
.unwrap()
.get(&query.state)
.filter(|d| d.did == did)
.map(|d| d.done.subscribe())
.ok_or_else(|| AppError(StatusCode::NOT_FOUND, "no such delegation".into()))?;
let waited = tokio::time::timeout(gw.wait, done.wait_for(Option::is_some)).await;
let Ok(Ok(outcome)) = waited else {
return Ok(Json(json!({ "done": false })));
};
let outcome = outcome.clone().expect("waited for an outcome");
gw.delegations.lock().unwrap().remove(&query.state);
outcome.map_err(forbidden)?;
Ok(Json(json!({ "done": true })))
}
/// The repository, as its owner, for a caller who may write to it.
@@ -245,7 +382,7 @@ pub async fn push(
&remote,
&snap,
&mirror.git,
&session.access_jwt,
&session.auth(),
&updates,
)
.await
@@ -368,7 +505,7 @@ async fn attempt(
&remote,
&snap,
&mirror.git,
&session.access_jwt,
&session.auth(),
&[update],
)
.await
crates/knotless-serve/src/lib.rs+6−0
@@ -383,6 +383,12 @@ pub fn router(app: Arc<App>) -> Router {
.route("/api/{authority}/{repo}/object/{sha}", get(api::object))
.route("/.well-known/did.json", get(gateway::did_document))
.route("/xrpc/dev.knotless.putDelegation", post(gateway::delegate))
.route(
"/xrpc/dev.knotless.getDelegation",
get(gateway::await_delegation),
)
.route(gateway::METADATA, get(gateway::client_metadata))
.route("/oauth/callback", get(gateway::callback))
.route(
"/xrpc/dev.knotless.pushRefs",
post(gateway::push).layer(axum::extract::DefaultBodyLimit::max(PUSH_LIMIT)),
crates/knotless-serve/src/web.rs+1−1
@@ -762,7 +762,7 @@ async fn home(
let mut body = format!(
r#"<section class="hero"><h1>Git that lives in your atproto account.</h1><p>Each repository is a few records on your own PDS: no git server to run, nothing to install on the PDS. Clone and push with plain git, and let agents read, commit, review and merge over MCP.</p></section>
<section class="steps" id="connect">
<div class="step"><h3><b>1</b>Install and log in</h3><p>From a checkout of knotless. Use an app password.</p>{}{}</div>
<div class="step"><h3><b>1</b>Install and log in</h3><p>From a checkout of knotless. Logging in opens your browser.</p>{}{}</div>
<div class="step"><h3><b>2</b>Push</h3><p>Any repository, to any account on any PDS.</p>{}{}</div>
<div class="step"><h3><b>3</b>Connect your agent</h3><p>The MCP server acts as the account you logged in as.</p>{}<details><summary>Other MCP clients</summary><p>Add this to the client's MCP settings, such as Claude Desktop's or Cursor's:</p>{}</details></div>
</section>"#,
crates/knotless-serve/tests/forge.rs+9−2
@@ -18,6 +18,7 @@ struct Appview {
did: String,
app: Arc<App>,
rt: tokio::runtime::Runtime,
_proxy: TlsProxy,
_sandbox: Sandbox,
}
@@ -31,8 +32,12 @@ fn appview() -> Appview {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let base = format!("http://localhost:{}", listener.local_addr().unwrap().port());
let gw = Gateway::new(&base).unwrap();
let port = listener.local_addr().unwrap().port();
let base = format!("http://localhost:{port}");
let mut gw = Gateway::new(&base).unwrap();
// Authorization servers reach a gateway, an OAuth client, over https.
let proxy = tls_proxy(port);
gw.oauth_url = proxy.url.clone();
let did = gw.did.clone();
let index = Index::open(&s.path("index.db")).unwrap();
let app = App::build(ctx(&s), Duration::ZERO, Some(gw), Some(index));
@@ -45,6 +50,7 @@ fn appview() -> Appview {
did,
app,
rt,
_proxy: proxy,
_sandbox: s,
}
}
@@ -54,6 +60,7 @@ fn person(account: Account, appview: &Appview) -> Sandbox {
s.set("KNOTLESS_PASSWORD", "");
s.set("KNOTLESS_DID_WEB_SCHEME", "http");
s.set("KNOTLESS_APPVIEW", &appview.base);
s.set("KNOTLESS_BROWSER", &browser(&s.account.clone(), ""));
let password = s.account.password.clone();
let handle = s.account.handle.clone();
assert_eq!(
crates/knotless-serve/tests/gateway.rs+86−13
@@ -17,6 +17,7 @@ struct Server {
did: String,
app: Arc<App>,
_rt: tokio::runtime::Runtime,
_proxy: TlsProxy,
_sandbox: Sandbox,
}
@@ -31,8 +32,14 @@ fn gateway() -> Server {
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
// atproto takes a did:web with a port only for localhost.
let base = format!("http://localhost:{}", listener.local_addr().unwrap().port());
let gw = Gateway::new(&base).unwrap();
let port = listener.local_addr().unwrap().port();
let base = format!("http://localhost:{port}");
let mut gw = Gateway::new(&base).unwrap();
// Authorization servers reach a gateway, an OAuth client, over https.
let proxy = tls_proxy(port);
gw.oauth_url = proxy.url.clone();
// Short, so that delegating asks more than once.
gw.wait = Duration::from_millis(100);
let did = gw.did.clone();
let app = App::build(ctx(&s), Duration::ZERO, Some(gw), None);
let router = router(app.clone());
@@ -42,6 +49,7 @@ fn gateway() -> Server {
did,
app,
_rt: rt,
_proxy: proxy,
_sandbox: s,
}
}
@@ -52,10 +60,11 @@ fn web(mut s: Sandbox) -> Sandbox {
}
/// A sandbox logged in as `account`, with no password in its environment,
/// so it acts only as itself.
/// so it acts only as itself, and a browser that approves as it.
fn person(account: Account) -> Sandbox {
let mut s = web(Sandbox::with(account));
s.set("KNOTLESS_PASSWORD", "");
s.set("KNOTLESS_BROWSER", &browser(&s.account.clone(), ""));
let password = s.account.password.clone();
let handle = s.account.handle.clone();
assert_eq!(
@@ -133,6 +142,28 @@ fn collaborators_push_and_commit_through_a_gateway() {
let (_, refs) = owner.cli(&["refs", &url], b"");
assert!(!refs.contains("feature"), "{refs}");
// The gateway refreshes its grant itself, as the client it was granted
// to, once the access token runs out.
let held = gw
._sandbox
.path("config/sessions")
.join(format!("{}.json", owner.account.did.replace(':', "_")));
let mut session: knotless::session::Session = knotless::store::read(&held).unwrap();
let oauth = session.oauth.as_mut().unwrap();
assert!(
oauth
.client_id
.as_ref()
.unwrap()
.ends_with("/oauth-client-metadata.json")
);
oauth.expires_at = 0;
let used = session.refresh_jwt.clone();
knotless::store::write(&held, &session, true).unwrap();
collab.git(&clone, &["push", "-q", "origin", "main:refs/heads/again"]);
let refreshed: knotless::session::Session = knotless::store::read(&held).unwrap();
assert_ne!(refreshed.refresh_jwt, used);
// git's rules hold at the gateway.
owner.commit_file(&work, "f", "moved on\n");
owner.git(&work, &["pull", "-q", "--rebase", &url, "main"]);
@@ -482,17 +513,58 @@ fn gateway_refusals() {
401
);
// A wrong password does not delegate; until one does, nothing writes.
let t = token(&owner, &gw.did, "dev.knotless.putDelegation");
let (status, _) = post(
&delegate,
Some(&t),
json!({"password": "nope"}).to_string().into_bytes(),
true,
);
assert_eq!(status, 403);
// A refused grant does not delegate; until one is approved, nothing
// writes.
let mut refuser = person(owner.account.clone());
refuser.set("KNOTLESS_BROWSER", &browser(&owner.account, "--reject"));
let (ok, _, err) = refuser.knotless(&["delegate", &owner.account.handle, &gw.did], b"");
assert!(!ok);
assert!(err.contains("rejected"), "{err}");
let t = token(&owner, &gw.did, "dev.knotless.pushRefs");
assert_eq!(post(&push, Some(&t), header(json!([])), false).0, 403);
// Nobody waits on a delegation nobody started, nor takes one back.
let t = token(&owner, &gw.did, "dev.knotless.getDelegation");
let res = reqwest::blocking::Client::new()
.get(format!(
"{}/xrpc/dev.knotless.getDelegation?state=nope",
gw.base
))
.bearer_auth(&t)
.send()
.unwrap();
assert_eq!(res.status().as_u16(), 404);
let res =
reqwest::blocking::get(format!("{}/oauth/callback?state=nope&code=x", gw.base)).unwrap();
assert_eq!(res.status().as_u16(), 400);
// Started, and the browser comes back with a code that is no good.
let t = token(&owner, &gw.did, "dev.knotless.putDelegation");
let (status, body) = post(&delegate, Some(&t), vec![], false);
assert_eq!(status, 200, "{body}");
let state = serde_json::from_str::<Value>(&body).unwrap()["state"]
.as_str()
.unwrap()
.to_string();
let issuer = net().pds.clone();
let page = reqwest::blocking::get(format!(
"{}/oauth/callback?state={state}&code=bogus&iss={issuer}",
gw.base
))
.unwrap()
.text()
.unwrap();
assert!(page.contains("was not delegated"), "{page}");
// The metadata the PDS reads the gateway's client from.
let metadata: Value = reqwest::blocking::get(format!("{}/oauth-client-metadata.json", gw.base))
.unwrap()
.json()
.unwrap();
assert!(
metadata["client_id"]
.as_str()
.unwrap()
.starts_with("https://")
);
assert_eq!(metadata["jwks"]["keys"].as_array().unwrap().len(), 1);
assert_eq!(
owner
.cli(&["delegate", &owner.account.handle, &gw.did], b"")
@@ -536,7 +608,8 @@ fn gateway_refusals() {
let missing = json!({"repo": owner.account.did, "name": "nothing-here", "updates": []});
let mut body = missing.to_string().into_bytes();
body.push(b'\n');
assert_eq!(post(&push, Some(&t), body, false).0, 404);
let (status, text) = post(&push, Some(&t), body, false);
assert_eq!(status, 404, "{text}");
let mut body = json!({"repo": owner.account.did, "name": "bad name!", "updates": []})
.to_string()
.into_bytes();
crates/knotless/src/auth.rs+1−1
@@ -21,7 +21,7 @@ pub async fn service_token(ctx: &Ctx, me: &Session, aud: &str, lxm: &str) -> Res
.query(
"com.atproto.server.getServiceAuth",
&[("aud", aud), ("lxm", lxm)],
Some(&me.access_jwt),
Some(&me.auth()),
)
.await?;
Ok(out.token)
crates/knotless/src/cli.rs+16−9
@@ -8,6 +8,7 @@ use crate::git::Git;
use crate::helper::{self, Knotless};
use crate::manage;
use crate::moderation;
use crate::oauth;
use crate::remote::Remote;
use crate::session;
use crate::url::RepoUrl;
@@ -56,7 +57,8 @@ pub fn helper_main(args: Vec<String>) -> i32 {
const USAGE: &str = "usage: knotless <command>
login <handle> [--password-stdin] store a session (else KNOTLESS_PASSWORD)
login <handle> [--password-stdin] log in with OAuth in a browser, or with
an app password (or KNOTLESS_PASSWORD)
logout <handle>
create <url> [--private] [--description <text>] [--default-branch <name>]
delegate <handle> <gateway did> let a gateway write for you, so
@@ -171,16 +173,21 @@ async fn command(
Some("--password-stdin") => {
let mut text = String::new();
stdin.read_to_string(&mut text)?;
text.trim().to_string()
Some(text.trim().to_string())
}
Some(other) => bail!("unexpected {other}"),
None => ctx
.env
.password
.clone()
.context("pass --password-stdin or set KNOTLESS_PASSWORD")?,
None => ctx.env.password.clone(),
};
let s = match password {
Some(password) => session::login(ctx, identifier, &password).await?,
None => {
let pending = oauth::start(ctx, identifier).await?;
writeln!(out, "approve in your browser, or open {}", pending.url)?;
out.flush()?;
oauth::open(ctx, &pending.url);
pending.finish(ctx).await?
}
};
let s = session::login(ctx, identifier, &password).await?;
session::remember(ctx, &s)?;
writeln!(out, "logged in as {} ({})", s.handle, s.did)?;
}
@@ -230,7 +237,7 @@ async fn command(
let id = crate::identity::resolve(ctx, args.next("<handle>")?).await?;
let gateway = args.next("<gateway did>")?;
let me = session::ensure(ctx, &id).await?;
crate::gateway::delegate(ctx, &me, gateway).await?;
crate::gateway::delegate(ctx, &me, gateway, out).await?;
writeln!(out, "delegated to {gateway}")?;
}
"share" | "unshare" => {
crates/knotless/src/commit.rs+1−1
@@ -102,7 +102,7 @@ pub async fn commit(ctx: &Ctx, url: &RepoUrl, spec: &Spec) -> Result<Committed>
dst: refname.clone(),
force: true,
};
let report = push::push(ctx, &remote, &snap, &git, &session.access_jwt, &[update]).await?;
let report = push::push(ctx, &remote, &snap, &git, &session.auth(), &[update]).await?;
match &report[0].1 {
Ok(()) => {
return Ok(Committed {
crates/knotless/src/compact.rs+2−2
@@ -66,7 +66,7 @@ pub async fn compact(ctx: &Ctx, remote: &Remote, session: &Session) -> Result<Co
.collect();
writes.extend(claims.iter().map(|rkey| remote::delete(nsid::CLAIM, rkey)));
for chunk in writes.chunks(200) {
remote.apply(chunk.to_vec(), &session.access_jwt).await?;
remote.apply(chunk.to_vec(), &session.auth()).await?;
}
Ok(Compacted {
before: packs.len(),
@@ -95,7 +95,7 @@ async fn fold(
let mut record = push::place(
ctx,
remote,
&session.access_jwt,
&session.auth(),
key,
pack,
tips,
crates/knotless/src/crypto.rs+14−0
@@ -131,6 +131,20 @@ impl P256Key {
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;
crates/knotless/src/env.rs+3−0
@@ -35,6 +35,8 @@ pub struct Env {
pub service: Option<String>,
pub identifier: Option<String>,
pub password: Option<String>,
/// The command an OAuth login opens its page with, the URL added last.
pub browser: Option<String>,
/// Packs up to this many bytes ride inside the pack record itself.
pub inline_max: usize,
/// Blob part size for larger packs; halved when a PDS refuses it.
@@ -106,6 +108,7 @@ impl Env {
service: get("KNOTLESS_SERVICE"),
identifier: get("KNOTLESS_IDENTIFIER"),
password: get("KNOTLESS_PASSWORD"),
browser: get("KNOTLESS_BROWSER").filter(|v| !v.is_empty()),
inline_max: num("KNOTLESS_INLINE_MAX", 192 * 1024) as usize,
chunk_size: num("KNOTLESS_CHUNK_SIZE", 4_000_000) as usize,
compact_threshold: num("KNOTLESS_COMPACT_THRESHOLD", 32) as usize,
crates/knotless/src/forge.rs+9−15
@@ -415,10 +415,10 @@ impl Target {
let page: Repos = self
.remote
.xrpc
.query_as(
.query(
"com.atproto.space.listRepos",
&params,
&space.for_repo(self.remote.did()),
Some(&space.for_repo(self.remote.did())),
)
.await?;
let done = page.repos.is_empty();
@@ -470,11 +470,8 @@ impl Target {
async fn write(&self, ctx: &Ctx, collection: &str, value: Value) -> Result<String> {
let (mine, me) = self.mine(ctx).await?;
let rkey = keys::tid();
mine.apply(
vec![remote::create(collection, &rkey, value)],
&me.access_jwt,
)
.await?;
mine.apply(vec![remote::create(collection, &rkey, value)], &me.auth())
.await?;
Ok(format!("at://{}/{collection}/{rkey}", me.did))
}
@@ -528,7 +525,7 @@ pub async fn hide(ctx: &Ctx, target: &RepoUrl, subject: &str, hidden: bool) -> R
Some(_) => remote::update(nsid::GATE, &reference.key, value),
None => remote::create(nsid::GATE, &reference.key, value),
};
mine.apply(vec![write], &me.access_jwt).await
mine.apply(vec![write], &me.auth()).await
}
pub async fn open_pull(
@@ -703,7 +700,7 @@ pub async fn merge(ctx: &Ctx, target: &RepoUrl, pull: &str) -> Result<String> {
force: false,
};
let report = if me.did == t.remote.did() {
push::push(ctx, &t.remote, &t.snap, &git, &me.access_jwt, &[update]).await?
push::push(ctx, &t.remote, &t.snap, &git, &me.auth(), &[update]).await?
} else {
gateway::push(ctx, &t.remote, &t.snap, &git, &me, &[update]).await?
};
@@ -762,11 +759,8 @@ pub async fn retarget(
}
record.head.branch = head.to_string();
let value = records::typed(nsid::PULL, &record);
mine.apply(
vec![remote::update(nsid::PULL, rkey, value)],
&me.access_jwt,
)
.await?;
mine.apply(vec![remote::update(nsid::PULL, rkey, value)], &me.auth())
.await?;
}
Ok(())
}
@@ -879,7 +873,7 @@ async fn copy(ctx: &Ctx, src: &Target, dest: &RepoUrl, new: NewRepo, me: &Sessio
force: false,
})
.collect();
accepted(push::push(ctx, &copy, &snap, &mirror.git, &me.access_jwt, &updates).await?)
accepted(push::push(ctx, &copy, &snap, &mirror.git, &me.auth(), &updates).await?)
}
/// Pulls or issues on a repository, newest first.
crates/knotless/src/gateway.rs+36−23
@@ -21,15 +21,22 @@ use crate::records::Bytes;
use crate::remote::{Remote, Snapshot};
use crate::session::Session;
use crate::store;
use crate::xrpc::Xrpc;
use crate::xrpc::{Auth, Xrpc};
use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::collections::BTreeSet;
use std::io::Write;
pub const PUSH: &str = "dev.knotless.pushRefs";
pub const COMMIT: &str = "dev.knotless.createCommit";
pub const DELEGATE: &str = "dev.knotless.putDelegation";
pub const AWAIT_DELEGATION: &str = "dev.knotless.getDelegation";
/// What an owner grants a gateway: the records a push writes and the blobs
/// that hold its packs, in the public repo and in the owner's spaces.
pub const SCOPE: &str = "atproto repo:dev.knotless.repo repo:dev.knotless.ref \
repo:dev.knotless.claim repo:dev.knotless.pack blob:*/* \
space:*?authority=self&collection=dev.knotless.repo&collection=dev.knotless.ref\
&collection=dev.knotless.claim&collection=dev.knotless.pack";
/// The DID document service a gateway is reached at.
pub const SERVICE: &str = "#knotless_gateway";
pub const PUSH_MIME: &str = "application/x-knotless-push";
@@ -52,12 +59,12 @@ fn gateway_of<'a>(remote: &Remote, snap: &'a Snapshot) -> Result<&'a str> {
}
}
async fn client(ctx: &Ctx, me: &Session, gateway: &str, lxm: &str) -> Result<(Xrpc, String)> {
async fn client(ctx: &Ctx, me: &Session, gateway: &str, lxm: &str) -> Result<(Xrpc, Auth)> {
let (endpoint, token) = tokio::try_join!(
endpoint(ctx, gateway),
auth::service_token(ctx, me, gateway, lxm)
)?;
Ok((Xrpc::new(ctx.http.clone(), &endpoint), token))
Ok((Xrpc::new(ctx.http.clone(), &endpoint), Auth::Bearer(token)))
}
/// What precedes the pack in a push to a gateway, on a line of its own.
@@ -278,28 +285,33 @@ pub async fn commit(ctx: &Ctx, remote: &Remote, me: &Session, spec: &Spec) -> Re
})
}
/// Delegates `me`'s writes to the gateway `did`: mints an app password for
/// it, hands it over, and remembers the gateway for `share --write`.
pub async fn delegate(ctx: &Ctx, me: &Session, did: &str) -> Result<()> {
/// Delegates `me`'s writes to the gateway `did`: the gateway asks `me`'s
/// PDS for a grant, which `me` approves in a browser, shown the page on
/// `out`. Remembers the gateway for `share --write`.
pub async fn delegate(ctx: &Ctx, me: &Session, did: &str, out: &mut dyn Write) -> Result<()> {
#[derive(Deserialize)]
struct AppPassword {
password: String,
struct Started {
url: String,
state: String,
}
#[derive(Deserialize)]
struct Done {
done: bool,
}
let created: AppPassword = Xrpc::new(ctx.http.clone(), &me.service)
.procedure(
"com.atproto.server.createAppPassword",
Some(&json!({ "name": format!("knotless gateway {}", crate::keys::tid()) })),
Some(&me.access_jwt),
)
.await?;
let (xrpc, token) = client(ctx, me, did, DELEGATE).await?;
xrpc.procedure::<serde::de::IgnoredAny>(
DELEGATE,
Some(&json!({ "password": created.password })),
Some(&token),
)
.await?;
store::write_bytes(&gateway_path(ctx, &me.did), did.as_bytes(), false)
let started: Started = xrpc.procedure(DELEGATE, None, Some(&token)).await?;
writeln!(out, "approve in your browser, or open {}", started.url)?;
out.flush()?;
crate::oauth::open(ctx, &started.url);
// The gateway answers once it has the grant, or after a while without.
loop {
let (xrpc, token) = client(ctx, me, did, AWAIT_DELEGATION).await?;
let params = [("state", started.state.as_str())];
let done: Done = xrpc.query(AWAIT_DELEGATION, &params, Some(&token)).await?;
if done.done {
return store::write_bytes(&gateway_path(ctx, &me.did), did.as_bytes(), false);
}
}
}
fn gateway_path(ctx: &Ctx, did: &str) -> std::path::PathBuf {
@@ -349,6 +361,7 @@ pub fn is_sha(text: &str) -> bool {
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn push_bodies() {
crates/knotless/src/helper.rs+1−1
@@ -250,7 +250,7 @@ impl Backend for Knotless {
&self.remote,
&snap,
&self.git,
&auth.access_jwt,
&auth.auth(),
&updates,
)
.await?;
crates/knotless/src/lib.rs+1−0
@@ -27,6 +27,7 @@ pub mod keys;
pub mod manage;
pub mod mirror;
pub mod moderation;
pub mod oauth;
pub mod push;
pub mod read;
pub mod records;
crates/knotless/src/manage.rs+6−6
@@ -66,7 +66,7 @@ pub async fn create(ctx: &Ctx, url: &RepoUrl, new: &NewRepo) -> Result<()> {
{
writes.push(remote::delete(nsid::REDIRECT, &url.name));
}
if let Err(err) = remote.apply(writes, &auth.access_jwt).await {
if let Err(err) = remote.apply(writes, &auth.auth()).await {
if remote.repo().await?.is_some() {
bail!("{url} already exists");
}
@@ -171,7 +171,7 @@ async fn set_access(ctx: &Ctx, url: &RepoUrl, who: &str, access: Option<bool>) -
&name,
records::typed(nsid::REPO, &record),
)],
&auth.access_jwt,
&auth.auth(),
)
.await
}
@@ -273,9 +273,9 @@ pub async fn rename(ctx: &Ctx, url: &RepoUrl, name: &str) -> Result<RepoUrl> {
writes.push(remote::update(nsid::REDIRECT, &at, value));
}
}
remote.apply(writes, &auth.access_jwt).await?;
remote.apply(writes, &auth.auth()).await?;
if let Some((there, write)) = pointer {
there.apply(vec![write], &auth.access_jwt).await?;
there.apply(vec![write], &auth.auth()).await?;
}
Ok(to)
}
@@ -352,7 +352,7 @@ pub async fn transfer(ctx: &Ctx, url: &RepoUrl, to: Option<&RepoUrl>) -> Result<
&repo.rkey,
records::typed(nsid::REPO, &record),
)],
&auth.access_jwt,
&auth.auth(),
)
.await
}
@@ -419,7 +419,7 @@ pub async fn delete(ctx: &Ctx, url: &RepoUrl) -> Result<()> {
}
}
for chunk in writes.chunks(200) {
remote.apply(chunk.to_vec(), &auth.access_jwt).await?;
remote.apply(chunk.to_vec(), &auth.auth()).await?;
}
Ok(())
}
crates/knotless/src/mirror.rs+2−2
@@ -12,12 +12,12 @@ use std::time::Duration;
/// A lock file. One left behind by a process that died is taken over once
/// it is older than any sync could be.
struct Lock(PathBuf);
pub(crate) struct Lock(PathBuf);
const STALE: Duration = Duration::from_secs(120);
impl Lock {
async fn acquire(path: PathBuf) -> Result<Lock> {
pub(crate) async fn acquire(path: PathBuf) -> Result<Lock> {
loop {
match std::fs::OpenOptions::new()
.write(true)
crates/knotless/src/moderation.rs+8−8
@@ -21,7 +21,7 @@ use crate::records::{BlockRecord, nsid};
use crate::remote::Remote;
use crate::session::{self, Session};
use crate::url::RepoUrl;
use crate::xrpc::Xrpc;
use crate::xrpc::{Auth, Xrpc};
use anyhow::{Context, Result, bail};
use base64::Engine;
use serde::Deserialize;
@@ -119,7 +119,7 @@ pub async fn report(ctx: &Ctx, target: &str, reason: &str, text: Option<&str>) -
body["reason"] = text.into();
}
Xrpc::new(ctx.http.clone(), &endpoint)
.procedure(REPORT, Some(&body), Some(&token))
.procedure(REPORT, Some(&body), Some(&Auth::Bearer(token)))
.await
}
@@ -169,7 +169,7 @@ pub async fn block(ctx: &Ctx, who: &str, repo: Option<&RepoUrl>, on: bool) -> Re
.collect(),
};
if !writes.is_empty() {
account.apply(writes, &me.access_jwt).await?;
account.apply(writes, &me.auth()).await?;
}
Ok(did)
}
@@ -203,7 +203,7 @@ pub async fn blocked_by(ctx: &Ctx, did: &str) -> Result<Vec<String>> {
/// (`KNOTLESS_ADMIN_PASSWORD`), at the appview.
pub struct Admin {
xrpc: Xrpc,
headers: Vec<(String, String)>,
auth: Auth,
}
impl Admin {
@@ -221,7 +221,7 @@ impl Admin {
let basic = base64::engine::general_purpose::STANDARD.encode(format!("admin:{password}"));
Ok(Admin {
xrpc: Xrpc::new(ctx.http.clone(), service),
headers: vec![("authorization".into(), format!("Basic {basic}"))],
auth: Auth::Headers(vec![("authorization".into(), format!("Basic {basic}"))]),
})
}
@@ -230,7 +230,7 @@ impl Admin {
let subject = subject(ctx, target).await?;
let body = json!({"subject": subject, "takedown": {"applied": applied}});
self.xrpc
.procedure_as(UPDATE_STATUS, Some(&body), &self.headers)
.procedure(UPDATE_STATUS, Some(&body), Some(&self.auth))
.await
}
@@ -245,7 +245,7 @@ impl Admin {
),
};
self.xrpc
.query_as(GET_STATUS, &[(param.0, &param.1)], &self.headers)
.query(GET_STATUS, &[(param.0, &param.1)], Some(&self.auth))
.await
}
@@ -258,7 +258,7 @@ impl Admin {
let limit = limit.to_string();
let out: Out = self
.xrpc
.query_as(LIST_REPORTS, &[("limit", &limit)], &self.headers)
.query(LIST_REPORTS, &[("limit", &limit)], Some(&self.auth))
.await?;
Ok(out.reports)
}
crates/knotless/src/oauth.rsnew+629−0
@@ -0,0 +1,629 @@
//! OAuth logins, as atproto profiles them: knotless is the loopback client
//! `http://localhost`, a public client that pushes its authorization
//! request with PKCE and holds access tokens bound to a DPoP key. The
//! browser comes back with the code to a listener on 127.0.0.1.
use crate::crypto::P256Key;
use crate::env::Ctx;
use crate::identity::{self, Identity};
use crate::keys;
use crate::mirror::Lock;
use crate::session::{self, OAuth, Session};
use crate::store;
use crate::xrpc::Xrpc;
use anyhow::{Context, Result, bail};
use base64::Engine;
use base64::engine::general_purpose::{STANDARD, URL_SAFE_NO_PAD};
use rand_core::RngCore;
use reqwest::Url;
use serde::Deserialize;
use serde::de::DeserializeOwned;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap};
use std::sync::Mutex;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
const CALLBACK: &str = "/callback";
/// What knotless asks for: what an app password allows, and the spaces
/// private repositories live in, which only a `space` scope reaches. A
/// PDS without spaces drops the scope it does not know. The scope names
/// every space type: naming [`crate::space::SPACE_TYPE`] alone takes its
/// lexicon, which is not published yet.
pub const SCOPE: &str = "atproto transition:generic \
space:*?authority=*&collection=*&manage=create&manage=update&manage=delete";
/// The loopback client, which names its scope and redirect in its ID. The
/// redirect leaves out the port, so any port the listener gets matches.
pub fn client_id() -> String {
let query = Url::parse_with_params(
"http://localhost/",
&[
("redirect_uri", format!("http://127.0.0.1{CALLBACK}")),
("scope", SCOPE.to_string()),
],
)
.expect("a static URL");
format!("http://localhost?{}", query.query().unwrap_or_default())
}
/// The DPoP nonce each server last sent, by origin.
static NONCES: Mutex<BTreeMap<String, String>> = Mutex::new(BTreeMap::new());
fn origin(url: &str) -> String {
Url::parse(url)
.map(|u| u.origin().ascii_serialization())
.unwrap_or_default()
}
fn random(bytes: usize) -> String {
let mut buf = vec![0; bytes];
rand_core::OsRng.fill_bytes(&mut buf);
URL_SAFE_NO_PAD.encode(buf)
}
fn sha256(text: &str) -> String {
URL_SAFE_NO_PAD.encode(Sha256::digest(text.as_bytes()))
}
/// A DPoP proof that `key` holds the request `method` `url`, for the
/// access token `token` when it carries one.
pub fn proof(key: &P256Key, method: &str, url: &str, token: Option<&str>) -> String {
let htu = Url::parse(url).map_or_else(
|_| url.to_string(),
|mut u| {
u.set_query(None);
u.set_fragment(None);
u.to_string()
},
);
let mut claims = json!({
"jti": random(16),
"htm": method,
"htu": htu,
"iat": keys::now_micros() / 1_000_000,
});
if let Some(nonce) = NONCES.lock().unwrap().get(&origin(url)) {
claims["nonce"] = nonce.as_str().into();
}
if let Some(token) = token {
claims["ath"] = sha256(token).into();
}
key.sign_jwt(&json!({"typ": "dpop+jwt", "jwk": key.jwk()}), &claims)
}
/// Keeps the nonce a server's answer to `url` brings.
fn keep_nonce(url: &str, res: &reqwest::Response) {
if let Some(nonce) = res
.headers()
.get("dpop-nonce")
.and_then(|v| v.to_str().ok())
{
NONCES
.lock()
.unwrap()
.insert(origin(url), nonce.to_string());
}
}
/// Keeps the nonce a resource server's answer brings, and says whether it
/// refused the request for want of it, so that sending it again may pass.
pub fn nonce_refused(url: &str, res: &reqwest::Response) -> bool {
keep_nonce(url, res);
res.status() == reqwest::StatusCode::UNAUTHORIZED
&& res
.headers()
.get(reqwest::header::WWW_AUTHENTICATE)
.and_then(|v| v.to_str().ok())
.is_some_and(|v| v.contains("use_dpop_nonce"))
}
/// An authorization server's endpoints.
#[derive(Deserialize)]
struct Server {
issuer: String,
authorization_endpoint: String,
token_endpoint: String,
pushed_authorization_request_endpoint: String,
revocation_endpoint: Option<String>,
}
async fn get_json<T: DeserializeOwned>(ctx: &Ctx, url: &str) -> Result<T> {
let res = ctx.http.get(url).send().await?.error_for_status()?;
Ok(res.json().await?)
}
/// The authorization server for accounts on `pds`.
async fn server(ctx: &Ctx, pds: &str) -> Result<Server> {
#[derive(Deserialize)]
struct Resource {
authorization_servers: Vec<String>,
}
let pds = pds.trim_end_matches('/');
let resource: Resource = get_json(ctx, &format!("{pds}/.well-known/oauth-protected-resource"))
.await
.with_context(|| format!("{pds} does not say where to log in with OAuth"))?;
let issuer = resource
.authorization_servers
.into_iter()
.next()
.with_context(|| format!("{pds} names no authorization server"))?;
let server: Server = get_json(
ctx,
&format!("{issuer}/.well-known/oauth-authorization-server"),
)
.await?;
if server.issuer != issuer {
bail!("{issuer} calls itself {}", server.issuer);
}
Ok(server)
}
/// A client of authorization servers: knotless's loopback client, or a
/// gateway, a confidential client that proves who it is with its key.
pub struct Client {
pub id: String,
key: Option<P256Key>,
}
/// The key ID a confidential client's assertions and JWKS name.
const KID: &str = "knotless-gateway";
impl Client {
pub fn loopback() -> Client {
Client {
id: client_id(),
key: None,
}
}
/// The confidential client whose metadata document is at `id`, signing
/// with `key`.
pub fn confidential(id: &str, key: P256Key) -> Client {
Client {
id: id.to_string(),
key: Some(key),
}
}
/// The client a session was granted to: a confidential one signs with
/// the key in this process's config directory.
fn of(ctx: &Ctx, oauth: &OAuth) -> Result<Client> {
Ok(match &oauth.client_id {
Some(id) => Client::confidential(id, client_key(ctx)?),
None => Client::loopback(),
})
}
/// The public key, as the JWKS in a confidential client's metadata.
pub fn jwks(&self) -> Value {
let keys: Vec<Value> = self
.key
.iter()
.map(|key| {
let mut jwk = key.jwk();
jwk["kid"] = KID.into();
jwk["alg"] = "ES256".into();
jwk["use"] = "sig".into();
jwk
})
.collect();
json!({ "keys": keys })
}
/// The form fields that name the client to the server `issuer`, with an
/// assertion signed by its key if it has one.
fn fields(&self, issuer: &str) -> Vec<(&'static str, String)> {
let mut fields = vec![("client_id", self.id.clone())];
if let Some(key) = &self.key {
let now = keys::now_micros() / 1_000_000;
let claims = json!({
"iss": self.id,
"sub": self.id,
"aud": issuer,
"jti": random(16),
"iat": now,
"exp": now + 60,
});
fields.push((
"client_assertion_type",
"urn:ietf:params:oauth:client-assertion-type:jwt-bearer".into(),
));
fields.push((
"client_assertion",
key.sign_jwt(&json!({ "kid": KID }), &claims),
));
}
fields
}
}
/// This process's key as a confidential client, made the first time it is
/// asked for and kept, readable only by the user, in the config directory.
pub fn client_key(ctx: &Ctx) -> Result<P256Key> {
let path = ctx.env.config_dir.join("oauth-client-key");
if let Ok(text) = std::fs::read_to_string(&path) {
return P256Key::from_bytes(&STANDARD.decode(text.trim())?);
}
let key = P256Key::generate();
store::write_bytes(&path, STANDARD.encode(key.to_bytes()).as_bytes(), true)?;
Ok(key)
}
/// Posts `form` from `client` to the server `issuer`, with a DPoP proof of
/// `key`. An empty answer reads as `null`.
async fn post<T: DeserializeOwned>(
ctx: &Ctx,
url: &str,
key: &P256Key,
client: &Client,
issuer: &str,
form: &[(&str, &str)],
) -> Result<T> {
let attempt = || async {
let mut fields: Vec<(&str, String)> =
form.iter().map(|(k, v)| (*k, v.to_string())).collect();
fields.extend(client.fields(issuer));
let res = ctx
.http
.post(url)
.header("dpop", proof(key, "POST", url, None))
.form(&fields)
.send()
.await?;
keep_nonce(url, &res);
let ok = res.status().is_success();
let body = res.bytes().await?;
anyhow::Ok((ok, serde_json::from_slice(&body).unwrap_or(Value::Null)))
};
let mut answer = attempt().await?;
if answer.1["error"] == "use_dpop_nonce" {
answer = attempt().await?;
}
let (ok, value) = answer;
if !ok {
bail!(
"{}: {}",
value["error"].as_str().unwrap_or("refused"),
value["error_description"].as_str().unwrap_or_default()
);
}
Ok(serde_json::from_value(value)?)
}
#[derive(Deserialize)]
struct Tokens {
access_token: String,
refresh_token: String,
expires_in: u64,
sub: String,
}
fn expires_at(tokens: &Tokens) -> u64 {
keys::now_micros() / 1_000_000 + tokens.expires_in
}
/// An authorization request pushed to the account's server, waiting for
/// the user to approve it.
pub struct Request {
/// The page where the user approves it.
pub url: String,
/// Names the request in the browser's return.
pub state: String,
id: Identity,
server: Server,
key: P256Key,
verifier: String,
redirect_uri: String,
}
/// Pushes a request for `client` to act as `id` with `scope`, the browser
/// coming back to `redirect_uri`. `hint` is who the user said they are.
pub async fn authorize(
ctx: &Ctx,
client: &Client,
id: Identity,
hint: &str,
redirect_uri: &str,
scope: &str,
) -> Result<Request> {
let server = server(ctx, &id.pds).await?;
let key = P256Key::generate();
let verifier = random(32);
let state = random(16);
#[derive(Deserialize)]
struct Pushed {
request_uri: String,
}
let pushed: Pushed = post(
ctx,
&server.pushed_authorization_request_endpoint,
&key,
client,
&server.issuer,
&[
("response_type", "code"),
("code_challenge", &sha256(&verifier)),
("code_challenge_method", "S256"),
("redirect_uri", redirect_uri),
("scope", scope),
("state", &state),
("login_hint", hint.trim_start_matches('@')),
],
)
.await?;
let mut url = Url::parse(&server.authorization_endpoint)?;
url.query_pairs_mut()
.append_pair("client_id", &client.id)
.append_pair("request_uri", &pushed.request_uri);
Ok(Request {
url: url.to_string(),
state,
id,
server,
key,
verifier,
redirect_uri: redirect_uri.to_string(),
})
}
impl Request {
/// Trades the code in `params`, the query the browser came back with,
/// for a session for `client`, which is not stored and has no handle.
pub async fn redeem(
self,
ctx: &Ctx,
client: &Client,
params: &HashMap<String, String>,
) -> Result<Session> {
let iss = params.get("iss").unwrap_or(&self.server.issuer);
if params.get("state") != Some(&self.state) || iss != &self.server.issuer {
bail!("the browser came back from a login this one did not start");
}
let Some(code) = params.get("code") else {
bail!(
"the login was refused: {}",
params
.get("error_description")
.or(params.get("error"))
.map_or("no reason given", String::as_str)
);
};
let tokens: Tokens = post(
ctx,
&self.server.token_endpoint,
&self.key,
client,
&self.server.issuer,
&[
("grant_type", "authorization_code"),
("code", code),
("redirect_uri", &self.redirect_uri),
("code_verifier", &self.verifier),
],
)
.await?;
if tokens.sub != self.id.did {
bail!("logged in as {}, not {}", tokens.sub, self.id.did);
}
Ok(Session {
did: tokens.sub.clone(),
handle: String::new(),
service: self.id.pds,
access_jwt: tokens.access_token.clone(),
refresh_jwt: tokens.refresh_token.clone(),
oauth: Some(OAuth {
key: STANDARD.encode(self.key.to_bytes()),
issuer: self.server.issuer,
token_endpoint: self.server.token_endpoint,
revocation_endpoint: self.server.revocation_endpoint,
client_id: client.key.as_ref().map(|_| client.id.clone()),
expires_at: expires_at(&tokens),
}),
})
}
}
/// A login waiting for the user to approve it in a browser.
pub struct Pending {
/// The page where the user approves the login.
pub url: String,
/// Where the browser comes back to.
pub redirect_uri: String,
listener: TcpListener,
request: Request,
}
/// Starts logging in as `identifier`: pushes the authorization request
/// and listens for the browser's return.
pub async fn start(ctx: &Ctx, identifier: &str) -> Result<Pending> {
let id = identity::resolve(ctx, identifier).await?;
let listener = TcpListener::bind("127.0.0.1:0").await?;
let redirect_uri = format!(
"http://127.0.0.1:{}{CALLBACK}",
listener.local_addr()?.port()
);
let request = authorize(
ctx,
&Client::loopback(),
id,
identifier,
&redirect_uri,
SCOPE,
)
.await?;
Ok(Pending {
url: request.url.clone(),
redirect_uri,
listener,
request,
})
}
/// What the browser shows once it is back.
const DONE: &str = "<!doctype html><meta charset=utf-8><title>knotless</title>\
<p>Back to knotless: you can close this tab and return to your terminal.";
impl Pending {
/// Waits for the browser, trades the code for tokens, and stores the
/// session.
pub async fn finish(self, ctx: &Ctx) -> Result<Session> {
let params = self.callback().await?;
let mut session = self
.request
.redeem(ctx, &Client::loopback(), &params)
.await?;
#[derive(Deserialize)]
struct Who {
handle: String,
}
let who: Who = Xrpc::new(ctx.http.clone(), &session.service)
.query("com.atproto.server.getSession", &[], Some(&session.auth()))
.await?;
session.handle = who.handle;
session::save(ctx, &session)?;
Ok(session)
}
/// The query the browser comes back with. Anything else it asks the
/// listener for, such as an icon, is not found.
async fn callback(&self) -> Result<HashMap<String, String>> {
loop {
let (mut stream, _) = self.listener.accept().await?;
let mut buf = vec![0; 16 * 1024];
let n = stream.read(&mut buf).await?;
let head = String::from_utf8_lossy(&buf[..n]);
let target = head.split(' ').nth(1).unwrap_or_default();
let url = Url::parse(&format!("http://127.0.0.1{target}"))
.ok()
.filter(|u| u.path() == CALLBACK);
let Some(url) = url else {
stream
.write_all(
b"HTTP/1.1 404 Not Found\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
)
.await?;
continue;
};
let reply = format!(
"HTTP/1.1 200 OK\r\ncontent-type: text/html; charset=utf-8\r\n\
content-length: {}\r\nconnection: close\r\n\r\n{DONE}",
DONE.len()
);
stream.write_all(reply.as_bytes()).await?;
return Ok(url.query_pairs().into_owned().collect());
}
}
}
#[cfg(target_os = "macos")]
const OPENER: &str = "open";
#[cfg(not(target_os = "macos"))]
const OPENER: &str = "xdg-open";
/// Opens `url` in the user's browser: `KNOTLESS_BROWSER`, a command the URL
/// is added to, else the system's opener. Whether it opens or not, the URL
/// has been shown.
pub fn open(ctx: &Ctx, url: &str) {
let command = ctx.env.browser.as_deref().unwrap_or(OPENER);
let mut words = command.split_whitespace();
if let Some(program) = words.next() {
let _ = tokio::process::Command::new(program)
.args(words)
.arg(url)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.spawn();
}
}
/// New tokens for `stored`, an OAuth session. Refresh tokens are used once,
/// so processes refresh one at a time, and one that finds the session
/// already refreshed takes that.
pub async fn refresh(ctx: &Ctx, stored: &Session, oauth: &OAuth) -> Result<Session> {
let _lock = Lock::acquire(session::path(ctx, &stored.did).with_extension("lock")).await?;
if let Some(now) = session::load(ctx, &stored.did)
&& now.refresh_jwt != stored.refresh_jwt
{
return Ok(now);
}
let tokens: Tokens = post(
ctx,
&oauth.token_endpoint,
&oauth.key()?,
&Client::of(ctx, oauth)?,
&oauth.issuer,
&[
("grant_type", "refresh_token"),
("refresh_token", &stored.refresh_jwt),
],
)
.await?;
let session = Session {
access_jwt: tokens.access_token.clone(),
refresh_jwt: tokens.refresh_token.clone(),
oauth: Some(OAuth {
expires_at: expires_at(&tokens),
..oauth.clone()
}),
..stored.clone()
};
session::save(ctx, &session)?;
Ok(session)
}
/// Revokes an OAuth session's tokens, as far as its server lets it.
pub async fn revoke(ctx: &Ctx, session: &Session, oauth: &OAuth) -> Result<()> {
let Some(endpoint) = &oauth.revocation_endpoint else {
return Ok(());
};
post::<serde::de::IgnoredAny>(
ctx,
endpoint,
&oauth.key()?,
&Client::of(ctx, oauth)?,
&oauth.issuer,
&[("token", &session.refresh_jwt)],
)
.await
.map(drop)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::crypto;
#[test]
fn proofs() {
let key = P256Key::generate();
let token = proof(&key, "GET", "http://pds.test:1/xrpc/x?y=1#z", Some("tok"));
let (header, claims) = crypto::jwt_parts(&token).unwrap();
assert_eq!(header["typ"], "dpop+jwt");
assert_eq!(header["jwk"], key.jwk());
assert_eq!(claims["htu"], "http://pds.test:1/xrpc/x");
assert_eq!(claims["htm"], "GET");
assert_eq!(claims["ath"], sha256("tok"));
assert!(claims.get("nonce").is_none());
let (signed, sig) = token.rsplit_once('.').unwrap();
key.public()
.verify(signed.as_bytes(), &URL_SAFE_NO_PAD.decode(sig).unwrap())
.unwrap();
// Not a URL: proved as it is.
let (_, claims) = crypto::jwt_parts(&proof(&key, "POST", "nowhere", None)).unwrap();
assert_eq!(claims["htu"], "nowhere");
assert!(claims.get("ath").is_none());
assert_eq!(origin("nowhere"), "");
}
#[test]
fn client() {
let id = client_id();
assert!(id.starts_with(
"http://localhost?redirect_uri=http%3A%2F%2F127.0.0.1%2Fcallback&scope=atproto+"
));
assert!(SCOPE.contains(" space:*?authority=*&"));
}
}
crates/knotless/src/push.rs+5−5
@@ -8,7 +8,7 @@ use crate::records::{
self, BlobRef, Bytes, ClaimRecord, PACK_MIME, PackRecord, RefRecord, RepoRecord, nsid,
};
use crate::remote::{self, Remote, Snapshot, Stored};
use crate::xrpc::XrpcError;
use crate::xrpc::{Auth, XrpcError};
use anyhow::{Result, bail};
use futures::{StreamExt, TryStreamExt};
use serde_json::Value;
@@ -63,7 +63,7 @@ pub async fn push(
remote: &Remote,
snap: &Snapshot,
git: &Git,
auth: &str,
auth: &Auth,
updates: &[Update],
) -> Result<Report> {
let mut report = Report::new();
@@ -294,7 +294,7 @@ async fn conflicts(
pub async fn place(
ctx: &Ctx,
remote: &Remote,
auth: &str,
auth: &Auth,
key: &str,
pack: Vec<u8>,
tips: Vec<String>,
@@ -322,7 +322,7 @@ pub async fn place(
/// Uploads `pack` as blobs of at most the chunk size, four at a time,
/// halving the size while the PDS calls a part too large.
pub async fn upload(ctx: &Ctx, remote: &Remote, auth: &str, pack: &[u8]) -> Result<Vec<BlobRef>> {
pub async fn upload(ctx: &Ctx, remote: &Remote, auth: &Auth, pack: &[u8]) -> Result<Vec<BlobRef>> {
let mut chunk = ctx.env.chunk_size.max(1);
loop {
// Owned parts, so the stream's future is Send for any caller.
@@ -330,7 +330,7 @@ pub async fn upload(ctx: &Ctx, remote: &Remote, auth: &str, pack: &[u8]) -> Resu
let res: Result<Vec<BlobRef>> = futures::stream::iter(parts)
.map(|part| {
let xrpc = remote.xrpc.clone();
let auth = auth.to_string();
let auth = auth.clone();
async move { xrpc.upload_blob(part, PACK_MIME, &auth).await }
})
.buffered(4)
crates/knotless/src/remote.rs+5−8
@@ -9,7 +9,7 @@ use crate::session;
use crate::space::{self, Space};
use crate::store;
use crate::url::RepoUrl;
use crate::xrpc::{Xrpc, XrpcError};
use crate::xrpc::{Auth, Xrpc, XrpcError};
use anyhow::Result;
use serde::Deserialize;
use serde::de::DeserializeOwned;
@@ -232,10 +232,6 @@ impl Remote {
}
}
fn headers(&self) -> &[(String, String)] {
self.space.as_ref().map_or(&[], |s| &s.headers)
}
/// A read from the public repo or the space, with `params` naming the
/// repo and, for a space, the space.
async fn read<T: DeserializeOwned>(&self, method: &str, params: &[(&str, &str)]) -> Result<T> {
@@ -248,14 +244,15 @@ impl Remote {
None => format!("com.atproto.repo.{method}"),
};
all.extend_from_slice(params);
self.xrpc.query_as(&nsid, &all, self.headers()).await
let auth = self.space.as_ref().map(|s| &s.auth);
self.xrpc.query(&nsid, &all, auth).await
}
pub async fn get_blob(&self, cid: &str) -> Result<Vec<u8>> {
match &self.space {
Some(space) => {
self.xrpc
.get_space_blob(&space.uri, self.did(), cid, &space.headers)
.get_space_blob(&space.uri, self.did(), cid, &space.auth)
.await
}
None => self.xrpc.get_blob(self.did(), cid).await,
@@ -481,7 +478,7 @@ impl Remote {
/// Applies writes made with [`create`], [`update`] and [`delete`],
/// atomically, to the public repo or the space.
pub async fn apply(&self, mut writes: Vec<Value>, auth: &str) -> Result<()> {
pub async fn apply(&self, mut writes: Vec<Value>, auth: &Auth) -> Result<()> {
let (nsid, body) = match &self.space {
Some(space) => {
for write in &mut writes {
crates/knotless/src/session.rs+70−7
@@ -1,14 +1,19 @@
//! App password sessions, stored per DID and refreshed in place.
//! Sessions, from an OAuth login or an app password, stored per DID and
//! refreshed in place.
use crate::crypto::P256Key;
use crate::env::Ctx;
use crate::identity::{self, Identity};
use crate::oauth;
use crate::store;
use crate::xrpc::Xrpc;
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")]
@@ -19,9 +24,57 @@ pub struct Session {
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>,
}
fn path(ctx: &Ctx, did: &str) -> PathBuf {
/// 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, a gateway's;
/// none for knotless's loopback client.
pub client_id: 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.
fn expiring(&self) -> bool {
match &self.oauth {
Some(oauth) => oauth.expires_at < crate::keys::now_micros() / 1_000_000 + 60,
None => expiring(&self.access_jwt),
}
}
}
pub(crate) fn path(ctx: &Ctx, did: &str) -> PathBuf {
ctx.env
.config_dir
.join("sessions")
@@ -32,7 +85,7 @@ pub fn load(ctx: &Ctx, did: &str) -> Option<Session> {
store::read(&path(ctx, did))
}
fn save(ctx: &Ctx, session: &Session) -> Result<()> {
pub fn save(ctx: &Ctx, session: &Session) -> Result<()> {
store::write(&path(ctx, &session.did), session, true)
}
@@ -50,9 +103,16 @@ pub async fn login(ctx: &Ctx, identifier: &str, password: &str) -> Result<Sessio
Ok(session)
}
/// Removes the stored session. Returns whether there was one.
/// 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
{
// A token the server cannot revoke now expires on its own.
let _ = oauth::revoke(ctx, &session, oauth).await;
}
Ok(std::fs::remove_file(path(ctx, &id.did)).is_ok())
}
@@ -61,7 +121,7 @@ pub async fn logout(ctx: &Ctx, identifier: &str) -> Result<bool> {
/// `KNOTLESS_PASSWORD`.
pub async fn ensure(ctx: &Ctx, id: &Identity) -> Result<Session> {
if let Some(stored) = load(ctx, &id.did) {
if !expiring(&stored.access_jwt) {
if !stored.expiring() {
return Ok(stored);
}
if let Ok(fresh) = refresh(ctx, &stored).await {
@@ -109,11 +169,14 @@ pub async fn reader(ctx: &Ctx, owner: &Identity) -> Option<Session> {
}
async fn refresh(ctx: &Ctx, stored: &Session) -> Result<Session> {
if let Some(oauth) = &stored.oauth {
return oauth::refresh(ctx, stored, oauth).await;
}
let mut session: Session = Xrpc::new(ctx.http.clone(), &stored.service)
.procedure(
"com.atproto.server.refreshSession",
None,
Some(&stored.refresh_jwt),
Some(&Auth::bearer(&stored.refresh_jwt)),
)
.await?;
session.service = stored.service.clone();
crates/knotless/src/space.rs+47−33
@@ -11,7 +11,7 @@ use crate::env::Ctx;
use crate::keys;
use crate::session::Session;
use crate::store;
use crate::xrpc::{Xrpc, XrpcError};
use crate::xrpc::{Auth, Xrpc, XrpcError};
use anyhow::{Context, Result};
use base64::Engine;
use base64::engine::general_purpose::STANDARD;
@@ -28,11 +28,11 @@ pub fn uri(did: &str, name: &str) -> String {
format!("at://{did}/space/{SPACE_TYPE}/{name}")
}
/// A space, and the headers that let requests read its owner's repo in it.
/// A space, and what lets requests read its owner's repo in it.
#[derive(Clone)]
pub struct Space {
pub uri: String,
pub headers: Vec<(String, String)>,
pub auth: Auth,
/// A credential and its key, to read any member's repo in the space.
credential: Option<(String, Arc<P256Key>)>,
}
@@ -46,7 +46,8 @@ struct Cached {
/// The errors that mean "this space is not yours to read", which a reader
/// cannot tell apart from "there is no such space", or that the PDS has no
/// spaces at all: one without them passes `com.atproto.space.*` on to an
/// appview it may not have.
/// appview it may not have, which an OAuth grant without a scope for that
/// may not reach.
pub fn refused(err: &anyhow::Error) -> bool {
XrpcError::of(err).is_some_and(|e| {
matches!(
@@ -57,6 +58,7 @@ pub fn refused(err: &anyhow::Error) -> bool {
| "NotAuthorized"
| "MethodNotImplemented"
| "UpstreamFailure"
| "ScopeMissingError"
)
})
}
@@ -66,7 +68,7 @@ impl Space {
pub fn owner(uri: &str, session: &Session) -> Space {
Space {
uri: uri.to_string(),
headers: crate::xrpc::bearer(Some(&session.access_jwt)),
auth: session.auth(),
credential: None,
}
}
@@ -75,7 +77,11 @@ impl Space {
pub fn member(uri: &str, audience: &str, credential: &str, key: P256Key) -> Space {
Space {
uri: uri.to_string(),
headers: signed(&key, &format!("Atproto-Space {credential}"), Some(audience)),
auth: Auth::Headers(signed(
&key,
&format!("Atproto-Space {credential}"),
Some(audience),
)),
credential: Some((credential.to_string(), Arc::new(key))),
}
}
@@ -83,20 +89,22 @@ impl Space {
/// The space as read from `did`'s repo in it.
pub fn at(&self, did: &str) -> Space {
Space {
headers: self.for_repo(did),
auth: self.for_repo(did),
..self.clone()
}
}
/// Headers to read `did`'s repo in the space, or to ask the authority
/// (`did` the space's) which repos it holds. Only a credential reaches
/// past its holder's own repo.
pub fn for_repo(&self, did: &str) -> Vec<(String, String)> {
/// What lets a request read `did`'s repo in the space, or ask the
/// authority (`did` the space's) which repos it holds. Only a
/// credential reaches past its holder's own repo.
pub fn for_repo(&self, did: &str) -> Auth {
match &self.credential {
Some((credential, key)) => {
signed(key, &format!("Atproto-Space {credential}"), Some(did))
}
None => self.headers.clone(),
Some((credential, key)) => Auth::Headers(signed(
key,
&format!("Atproto-Space {credential}"),
Some(did),
)),
None => self.auth.clone(),
}
}
@@ -164,19 +172,19 @@ async fn credential(
.query(
"com.atproto.space.getDelegationToken",
&[("space", uri)],
Some(&me.access_jwt),
Some(&me.auth()),
)
.await
.context(
"getting a delegation token, which takes a full login: an app password cannot \
read another account's private repository",
"getting a delegation token, which takes an OAuth login (`knotless login`): \
an app password cannot read another account's private repository",
)?;
let headers = signed(key, &format!("Bearer {}", token.token), None);
let auth = Auth::Headers(signed(key, &format!("Bearer {}", token.token), None));
let out: Credential = host
.procedure_as(
.procedure(
"com.atproto.space.getSpaceCredential",
Some(&json!({ "space": uri })),
&headers,
Some(&auth),
)
.await?;
Ok(out.credential)
@@ -221,8 +229,8 @@ pub struct Member {
}
impl Admin<'_> {
fn auth(&self) -> Option<&str> {
Some(&self.session.access_jwt)
fn auth(&self) -> Auth {
self.session.auth()
}
/// Creates the space for private repository `name`, readable and
@@ -241,7 +249,7 @@ impl Admin<'_> {
"writePolicy": member_list,
"appAccess": {"$type": "com.atproto.simplespace.defs#open"},
})),
self.auth(),
Some(&self.auth()),
)
.await;
let made = match res {
@@ -296,7 +304,11 @@ impl Admin<'_> {
}
let out: Out = self
.xrpc
.query("com.atproto.simplespace.listMembers", &params, self.auth())
.query(
"com.atproto.simplespace.listMembers",
&params,
Some(&self.auth()),
)
.await?;
let done = out.members.is_empty();
all.extend(out.members);
@@ -309,7 +321,7 @@ impl Admin<'_> {
async fn call(&self, nsid: &str, body: Value) -> Result<()> {
self.xrpc
.procedure::<serde::de::IgnoredAny>(nsid, Some(&body), self.auth())
.procedure::<serde::de::IgnoredAny>(nsid, Some(&body), Some(&self.auth()))
.await
.map(drop)
}
@@ -342,23 +354,25 @@ mod tests {
.verify(base.as_bytes(), &STANDARD.decode(sig).unwrap())
.unwrap();
let headers = |auth: &Auth| auth.headers("GET", "http://pds.test/xrpc/x");
let own = Space {
uri: "at://x/space/t/k".into(),
headers: vec![("authorization".into(), "Bearer t".into())],
auth: Auth::bearer("t"),
credential: None,
};
assert_eq!(own.for_repo("did:plc:m"), own.headers);
assert_eq!(headers(&own.for_repo("did:plc:m")), headers(&own.auth));
let space = Space::member("at://x/space/t/k", "did:plc:o", "cred", key);
assert_eq!(space.headers.len(), 4);
assert_eq!(space.for_repo("did:plc:o")[1], space.headers[1]);
assert_eq!(space.for_repo("did:plc:m")[1].1, "did:plc:m");
let space_headers = headers(&space.auth);
assert_eq!(space_headers.len(), 4);
assert_eq!(headers(&space.for_repo("did:plc:o"))[1], space_headers[1]);
assert_eq!(headers(&space.for_repo("did:plc:m"))[1].1, "did:plc:m");
let key = P256Key::from_bytes(&space.credential.as_ref().unwrap().1.to_bytes()).unwrap();
assert_eq!(
space.headers[1],
space_headers[1],
("atproto-space-audience".into(), "did:plc:o".into())
);
let sig = space.headers[3]
let sig = space_headers[3]
.1
.strip_prefix("atproto-space=:")
.unwrap()
crates/knotless/src/xrpc.rs+99−58
@@ -1,11 +1,15 @@
//! A minimal XRPC client over the shared HTTP connection pool.
use crate::crypto::P256Key;
use crate::oauth;
use crate::records::BlobRef;
use anyhow::Result;
use bytes::Bytes;
use serde::Deserialize;
use serde::de::DeserializeOwned;
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)]
@@ -46,6 +50,36 @@ impl XrpcError {
}
}
/// What lets a request in.
#[derive(Clone)]
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)>),
}
impl Auth {
pub fn bearer(token: &str) -> Auth {
Auth::Bearer(token.to_string())
}
/// The headers for a `method` request to `url`.
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(),
}
}
}
#[derive(Clone)]
pub struct Xrpc {
http: reqwest::Client,
@@ -68,20 +102,13 @@ impl Xrpc {
&self,
nsid: &str,
params: &[(&str, &str)],
auth: Option<&str>,
auth: Option<&Auth>,
) -> Result<T> {
self.query_as(nsid, params, &bearer(auth)).await
}
/// A query sent with `headers`, such as a space credential's.
pub async fn query_as<T: DeserializeOwned>(
&self,
nsid: &str,
params: &[(&str, &str)],
headers: &[(String, String)],
) -> Result<T> {
let req = self.http.get(self.url(nsid)).query(params);
json(with(req, headers).send().await?).await
let url = self.url(nsid);
let res = self
.send("GET", &url, auth, || self.http.get(&url).query(params))
.await?;
json(res).await
}
/// Calls a procedure. `body: None` sends none, which a procedure that
@@ -90,22 +117,19 @@ impl Xrpc {
&self,
nsid: &str,
body: Option<&Value>,
auth: Option<&str>,
) -> Result<T> {
self.procedure_as(nsid, body, &bearer(auth)).await
}
pub async fn procedure_as<T: DeserializeOwned>(
&self,
nsid: &str,
body: Option<&Value>,
headers: &[(String, String)],
auth: Option<&Auth>,
) -> Result<T> {
let mut req = self.http.post(self.url(nsid));
if let Some(body) = body {
req = req.json(body);
}
json(with(req, headers).send().await?).await
let url = self.url(nsid);
let res = self
.send("POST", &url, auth, || {
let req = self.http.post(&url);
match body {
Some(body) => req.json(body),
None => req,
}
})
.await?;
json(res).await
}
/// A procedure whose input is raw bytes of `mime`.
@@ -114,30 +138,51 @@ impl Xrpc {
nsid: &str,
body: Vec<u8>,
mime: &str,
auth: Option<&str>,
auth: Option<&Auth>,
) -> Result<T> {
let req = self
.http
.post(self.url(nsid))
.header(reqwest::header::CONTENT_TYPE, mime)
.body(body);
json(with(req, &bearer(auth)).send().await?).await
let url = self.url(nsid);
let body = Bytes::from(body);
let res = self
.send("POST", &url, auth, || {
self.http
.post(&url)
.header(reqwest::header::CONTENT_TYPE, mime)
.body(body.clone())
})
.await?;
json(res).await
}
pub async fn upload_blob(&self, bytes: Vec<u8>, mime: &str, auth: &str) -> Result<BlobRef> {
pub async fn upload_blob(&self, bytes: Vec<u8>, mime: &str, auth: &Auth) -> Result<BlobRef> {
#[derive(Deserialize)]
struct Out {
blob: BlobRef,
}
let res = self
.http
.post(self.url("com.atproto.repo.uploadBlob"))
.bearer_auth(auth)
.header(reqwest::header::CONTENT_TYPE, mime)
.body(bytes)
.send()
let out: Out = self
.procedure_bytes("com.atproto.repo.uploadBlob", bytes, mime, Some(auth))
.await?;
Ok(json::<Out>(res).await?.blob)
Ok(out.blob)
}
/// 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,
url: &str,
auth: Option<&Auth>,
make: impl Fn() -> reqwest::RequestBuilder,
) -> Result<reqwest::Response> {
let attempt = || {
let headers = auth.map(|a| a.headers(method, url)).unwrap_or_default();
with(make(), &headers).send()
};
let res = attempt().await?;
if matches!(auth, Some(Auth::Dpop { .. })) && oauth::nonce_refused(url, &res) {
return Ok(attempt().await?);
}
Ok(res)
}
pub async fn get_blob(&self, did: &str, cid: &str) -> Result<Vec<u8>> {
@@ -156,24 +201,20 @@ impl Xrpc {
space: &str,
did: &str,
cid: &str,
headers: &[(String, String)],
auth: &Auth,
) -> Result<Vec<u8>> {
let req = self
.http
.get(self.url("com.atproto.space.getBlob"))
.query(&[("space", space), ("repo", did), ("cid", cid)]);
Ok(ok(with(req, headers).send().await?).await?.to_vec())
let url = self.url("com.atproto.space.getBlob");
let res = self
.send("GET", &url, Some(auth), || {
self.http
.get(&url)
.query(&[("space", space), ("repo", did), ("cid", cid)])
})
.await?;
Ok(ok(res).await?.to_vec())
}
}
/// The headers that carry a bearer token, if there is one.
pub fn bearer(token: Option<&str>) -> Vec<(String, String)> {
token
.map(|t| ("authorization".to_string(), format!("Bearer {t}")))
.into_iter()
.collect()
}
fn with(mut req: reqwest::RequestBuilder, headers: &[(String, String)]) -> reqwest::RequestBuilder {
for (name, value) in headers {
req = req.header(name, value);
crates/knotless/tests/cli.rs+0−1
@@ -13,7 +13,6 @@ fn sessions() {
assert_eq!(code, 0);
assert!(out.contains(&s.account.did), "{out}");
assert_eq!(s.cli(&["login", &handle, "--bogus"], b"").0, 1);
assert_eq!(s.cli(&["login", &handle], b"").0, 1);
assert_eq!(s.cli(&["login"], b"").0, 1);
// The stored session pushes with no password around.
let work = s.work("w", 1);
crates/knotless/tests/common/mod.rs+55−0
@@ -23,9 +23,19 @@ pub struct Net {
/// Whether the PDS is on the permissioned spaces alpha.
#[serde(default)]
pub spaces: bool,
pub gateway: GatewayTls,
pub accounts: Vec<Account>,
}
/// The hostname the harness resolves to 127.0.0.1 for gateways, and the
/// certificate it trusts for it.
#[derive(Deserialize, Clone, Debug)]
pub struct GatewayTls {
pub host: String,
pub cert: String,
pub key: String,
}
static NET: OnceLock<Net> = OnceLock::new();
static CHILD: Mutex<Option<Child>> = Mutex::new(None);
@@ -180,6 +190,51 @@ pub fn bin_dir() -> PathBuf {
dir
}
/// A browser that signs in as `account` and approves, plus `extra` (such
/// as `--reject`): `harness/authorize.mjs`, as `KNOTLESS_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().gateway.cert,
)
}
/// An https front for a server on `port`, at the harness's gateway
/// 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().gateway;
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,
}
}
/// Config and cache directories of its own, and an account.
pub struct Sandbox {
pub dir: tempfile::TempDir,
crates/knotless/tests/library.rs+8−18
@@ -35,11 +35,8 @@ async fn connect(ctx: &Ctx, url: &str) -> Remote {
.unwrap()
}
async fn auth(ctx: &Ctx, remote: &Remote) -> String {
session::ensure(ctx, &remote.identity)
.await
.unwrap()
.access_jwt
async fn auth(ctx: &Ctx, remote: &Remote) -> knotless::xrpc::Auth {
session::ensure(ctx, &remote.identity).await.unwrap().auth()
}
fn put(path: &str, data: &str) -> Change {
@@ -194,7 +191,7 @@ fn xrpc_failures() {
);
assert_eq!(e.to_string(), "HTTP404 (404): gone");
assert!(
x.query::<serde_json::Value>("x.y", &[], Some("t"))
x.query::<serde_json::Value>("x.y", &[], Some(&knotless::xrpc::Auth::bearer("t")))
.await
.is_err()
);
@@ -513,7 +510,7 @@ fn push_edges() {
&r,
&snap,
&git,
"bad-token",
&knotless::xrpc::Auth::bearer("bad-token"),
&[up("HEAD", "refs/heads/n")]
)
.await
@@ -607,16 +604,9 @@ fn compaction_keeps_young_packs_and_drops_dead_refs() {
dst: "refs/heads/main".into(),
force: false,
};
let err = push::push(
&aged,
&r,
&snap,
&mirror.git,
&session.access_jwt,
&[drop_main],
)
.await
.unwrap_err();
let err = push::push(&aged, &r, &snap, &mirror.git, &session.auth(), &[drop_main])
.await
.unwrap_err();
assert!(err.to_string().contains("too long ago"), "{err}");
let c = compact::compact(&aged, &r, &session).await.unwrap();
assert_eq!((c.removed, c.claims), (0, 2));
@@ -634,7 +624,7 @@ fn compaction_keeps_young_packs_and_drops_dead_refs() {
dst: "refs/heads/main".into(),
force: false,
};
push::push(&ctx0, &r, &snap, &git, &session.access_jwt, &[del])
push::push(&ctx0, &r, &snap, &git, &session.auth(), &[del])
.await
.unwrap();
let c = compact::compact(&ctx0, &r, &session).await.unwrap();
crates/knotless/tests/oauth.rsnew+276−0
@@ -0,0 +1,276 @@
//! Logging in with OAuth, against the reference PDS's authorization server,
//! with `harness/authorize.mjs` as the user's browser.
mod common;
use common::*;
use axum::Router;
use axum::routing::{get, post};
use knotless::env::Ctx;
use knotless::session::Session;
use knotless::{oauth, store};
use serde_json::json;
/// A sandbox with no password around, whose logins go through `browser`.
fn oauth_sandbox(account: Account, browser: &str) -> Sandbox {
let mut s = Sandbox::with(account);
s.set("KNOTLESS_PASSWORD", "");
s.set("KNOTLESS_BROWSER", browser);
s
}
fn session_path(s: &Sandbox) -> std::path::PathBuf {
s.path("config/sessions")
.join(format!("{}.json", s.account.did.replace(':', "_")))
}
fn stored(s: &Sandbox) -> Session {
store::read(&session_path(s)).expect("a stored session")
}
fn rt() -> tokio::runtime::Runtime {
tokio::runtime::Runtime::new().unwrap()
}
#[test]
fn logs_in_pushes_refreshes_and_logs_out() {
let account = account();
let mut s = oauth_sandbox(account.clone(), &browser(&account, ""));
// Packs go over blobs, so uploads carry DPoP proofs too.
s.set("KNOTLESS_INLINE_MAX", "0");
let handle = s.account.handle.clone();
let (code, out) = s.cli(&["login", &handle], b"");
assert_eq!(code, 0, "{out}");
assert!(out.contains("approve in your browser"), "{out}");
assert!(
out.contains(&format!("logged in as {handle} ({})", s.account.did)),
"{out}"
);
let first = stored(&s);
assert!(first.oauth.is_some());
// git, in a process of its own, with the stored session.
let work = s.work("w", 2);
s.git(&work, &["push", "-q", &s.url("by-oauth"), "main"]);
let (_, refs) = s.cli(&["refs", &s.url("by-oauth")], b"");
assert!(refs.ends_with("refs/heads/main"), "{refs}");
// An access token about to expire is refreshed before it is used.
let mut expired = first.clone();
expired.oauth.as_mut().unwrap().expires_at = 0;
store::write(&session_path(&s), &expired, true).unwrap();
assert_eq!(s.cli(&["create", &s.url("refreshed")], b"").0, 0);
let refreshed = stored(&s);
assert_ne!(refreshed.refresh_jwt, first.refresh_jwt);
assert!(refreshed.oauth.as_ref().unwrap().expires_at > 0);
// A process that refreshes from a stale copy takes the one another
// process already refreshed, rather than spending a used token.
let ctx = Ctx::new(s.env());
let again = rt()
.block_on(oauth::refresh(
&ctx,
&expired,
expired.oauth.as_ref().unwrap(),
))
.unwrap();
assert_eq!(again, refreshed);
// Logging out revokes the tokens: put back, they no longer refresh.
assert_eq!(s.cli(&["logout", &handle], b""), (0, "logged out".into()));
let mut revoked = refreshed.clone();
revoked.oauth.as_mut().unwrap().expires_at = 0;
store::write(&session_path(&s), &revoked, true).unwrap();
assert_eq!(s.cli(&["create", &s.url("after")], b"").0, 1);
// A server that names no revocation endpoint: logging out only forgets.
let mut unrevocable = refreshed;
unrevocable.oauth.as_mut().unwrap().revocation_endpoint = None;
store::write(&session_path(&s), &unrevocable, true).unwrap();
assert_eq!(s.cli(&["logout", &handle], b""), (0, "logged out".into()));
}
#[test]
fn refused_and_crossed_logins() {
let account = account();
let s = oauth_sandbox(account.clone(), &browser(&account, "--reject"));
let ctx = Ctx::new(s.env());
let rt = rt();
let err = rt
.block_on(async {
let pending = oauth::start(&ctx, &account.handle).await?;
oauth::open(&ctx, &pending.url);
pending.finish(&ctx).await
})
.unwrap_err();
assert!(err.to_string().contains("refused"), "{err}");
assert_eq!(s.cli(&["login", &account.handle], b"").0, 1);
// The browser signs in as someone else than the login asked for.
let other = self::account();
let s = oauth_sandbox(account.clone(), &browser(&other, ""));
let ctx = Ctx::new(s.env());
let err = rt
.block_on(async {
let pending = oauth::start(&ctx, &account.handle).await?;
oauth::open(&ctx, &pending.url);
pending.finish(&ctx).await
})
.unwrap_err();
assert!(
err.to_string().contains(&format!("not {}", account.did)),
"{err}"
);
}
#[test]
fn the_listener_takes_only_its_own_callback() {
let s = oauth_sandbox(account(), "");
let ctx = Ctx::new(s.env());
let err = rt()
.block_on(async {
let pending = oauth::start(&ctx, &s.account.handle).await?;
let back = pending.redirect_uri.clone();
let base = back.trim_end_matches("/callback").to_string();
let poke = tokio::spawn(async move {
let http = reqwest::Client::new();
let icon = http
.get(format!("{base}/favicon.ico"))
.send()
.await
.unwrap();
assert_eq!(icon.status(), 404);
let page = http
.get(format!("{back}?state=forged&code=x"))
.send()
.await
.unwrap();
assert!(page.text().await.unwrap().contains("knotless"));
});
let res = pending.finish(&ctx).await;
poke.await.unwrap();
res
})
.unwrap_err();
assert!(err.to_string().contains("did not start"), "{err}");
}
/// Serves the router `routes` makes for the server's own address.
fn mock_server(rt: &tokio::runtime::Runtime, routes: impl FnOnce(String) -> Router) -> String {
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let router = routes(base.clone());
rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
base
}
#[test]
fn discovery_failures() {
let rt = rt();
let s = Sandbox::new();
// Each its own cache, as identities are cached with the PDS they name.
let start = |pds: &str| {
let mut env = s.env();
env.service = Some(pds.to_string());
env.cache_dir = s.path(&format!("cache-{}", pds.rsplit(':').next().unwrap()));
let ctx = Ctx::new(env);
let handle = s.account.handle.clone();
rt.block_on(async move { oauth::start(&ctx, &handle).await.map(drop) })
.unwrap_err()
.to_string()
};
let resource = |servers: serde_json::Value| {
get(move || {
let servers = servers.clone();
async move { axum::Json(json!({ "authorization_servers": servers })) }
})
};
let none = mock_server(&rt, |_| Router::new());
assert!(start(&none).contains("does not say where to log in"));
let nameless = mock_server(&rt, |_| {
Router::new().route("/.well-known/oauth-protected-resource", resource(json!([])))
});
assert!(start(&nameless).contains("names no authorization server"));
let impostor = mock_server(&rt, |base| {
Router::new()
.route(
"/.well-known/oauth-protected-resource",
resource(json!([base])),
)
.route(
"/.well-known/oauth-authorization-server",
get(|| async {
axum::Json(json!({
"issuer": "https://elsewhere.example",
"authorization_endpoint": "x",
"token_endpoint": "x",
"pushed_authorization_request_endpoint": "x",
}))
}),
)
});
assert!(start(&impostor).contains("calls itself"));
let refusing = mock_server(&rt, |base| {
let metadata = json!({
"issuer": base,
"authorization_endpoint": format!("{base}/authorize"),
"token_endpoint": format!("{base}/token"),
"pushed_authorization_request_endpoint": format!("{base}/par"),
});
Router::new()
.route(
"/.well-known/oauth-protected-resource",
resource(json!([base])),
)
.route(
"/.well-known/oauth-authorization-server",
get(move || async move { axum::Json(metadata) }),
)
.route(
"/par",
post(|| async {
(
axum::http::StatusCode::BAD_REQUEST,
axum::Json(json!({"error": "invalid_request", "error_description": "no"})),
)
}),
)
});
assert!(start(&refusing).contains("invalid_request: no"));
}
/// What an app password cannot do: read someone else's private
/// repository. Needs the spaces alpha (`KNOTLESS_DEV_ENV`).
#[test]
fn private_repositories() {
if !net().spaces {
eprintln!("skipped: the PDS is not on the spaces alpha");
return;
}
let login = |account: Account| {
let s = oauth_sandbox(account.clone(), &browser(&account, ""));
let handle = s.account.handle.clone();
assert_eq!(s.cli(&["login", &handle], b"").0, 0);
s
};
let owner = login(account());
let reader = login(account());
let url = owner.url("secret");
assert_eq!(owner.cli(&["create", &url, "--private"], b"").0, 0);
let work = owner.work("w", 2);
owner.git(&work, &["push", "-q", &url, "main"]);
let head = owner.git(&work, &["rev-parse", "HEAD"]);
assert_eq!(
owner.cli(&["share", &url, &reader.account.handle], b"").0,
0
);
let (ok, refs, err) = reader.git_raw(&reader.path("x"), &["ls-remote", &url]);
assert!(ok, "{err}");
assert!(refs.contains(&head), "{refs}");
}
crates/knotless/tests/spaces.rs+4−2
@@ -136,7 +136,7 @@ fn private_repositories() {
via_app.set("KNOTLESS_IDENTIFIER", &reader.account.handle);
let (ok, _, err) = via_app.git_raw(&via_app.path("x"), &["ls-remote", &url]);
assert!(!ok);
assert!(err.contains("takes a full login"), "{err}");
assert!(err.contains("takes an OAuth login"), "{err}");
// Pushing takes a gateway to write through, which needs delegating.
let (code, _) = owner.cli(&["share", &url, &reader.account.handle, "--write"], b"");
assert_eq!(code, 1);
@@ -248,7 +248,7 @@ fn app_password(s: &Sandbox) -> String {
.procedure(
"com.atproto.server.createAppPassword",
Some(&serde_json::json!({"name": knotless::keys::tid()})),
Some(&session.access_jwt),
Some(&session.auth()),
)
.await
.unwrap();
@@ -271,6 +271,7 @@ fn space_administration_failures() {
service: net().pds.clone(),
access_jwt: "bogus".into(),
refresh_jwt: "bogus".into(),
oauth: None,
};
let admin = knotless::space::Admin {
xrpc: &xrpc,
@@ -313,6 +314,7 @@ fn a_bad_owner_token_is_an_error_not_a_public_repository() {
service: net().pds.clone(),
access_jwt: jwt.clone(),
refresh_jwt: jwt,
oauth: None,
};
let path = s
.path("config/sessions")
docs/design.md+45−10
@@ -152,6 +152,30 @@ without a lock, because each lands under its own name. The ref transaction
runs under a lock file, and a lock older than two minutes is treated as
abandoned.
## Logging in
`knotless login <handle>` logs in with OAuth as atproto profiles it.
knotless is a loopback client: its client ID is `http://localhost` with
its scope and redirect in the query, so there is no client metadata to
host. It finds the account's authorization server from its PDS
(`/.well-known/oauth-protected-resource`), pushes the request with PKCE,
opens the authorization page in a browser, and takes the code back on a
listener on `127.0.0.1`. Tokens are bound to a P-256 DPoP key kept with
the session; every request to the PDS carries a proof, and one refused
for want of the server's nonce is sent again with it. Refresh tokens are
used once, so processes refresh under a lock file, and one that finds the
session already refreshed takes it. Logging out revokes the tokens.
The scope is `transition:generic`, what an app password allows, plus a
`space` scope for private repositories, which `transition:generic` does
not reach. It names every space type: naming `dev.knotless.space` alone
makes the authorization server resolve its lexicon, which is not
published. A PDS without spaces drops the scope it does not know.
An app password (`--password-stdin`, or `KNOTLESS_PASSWORD`) still logs
in. A loopback client's session lasts two weeks at most, after which
`knotless login` again.
## Private repositories
A private repository is the space
@@ -167,7 +191,7 @@ delegation token from their own PDS for a space credential from the
owner's, signing both requests with a fresh P-256 key (RFC 9421, as the
alpha specifies). Reads then carry the credential and an audience naming
the repo being read. App passwords cannot mint delegation tokens, so
reading someone else's private repository takes a full login. The
reading someone else's private repository takes an OAuth login. The
credential and its key are cached, readable only by the user, and reused
until a minute before they expire.
@@ -187,11 +211,21 @@ directly. A gateway is a `knotless-serve` with `--public-url`, identified as
`did:web` of its host and serving its own DID document with a
`#knotless_gateway` service.
- **Delegation.** `knotless delegate` creates an app password named for
the gateway and posts it to `dev.knotless.putDelegation`, authenticated with a
service token. The gateway logs in with it and keeps the session; the
owner revokes it like any app password. Sharing a repository for writing
records the gateway in its `repo` record.
- **Delegation.** The gateway is an OAuth client of its own, a
confidential one: its client ID is `/oauth-client-metadata.json` on its
public URL, and it authenticates to authorization servers with a P-256
key kept in its config directory (`oauth-client-key`), published in that
metadata. `knotless delegate` calls `dev.knotless.putDelegation` with a
service token; the gateway pushes an authorization request for the owner
and returns its URL, which the owner approves in a browser. The browser
comes back to the gateway's `/oauth/callback`, and the gateway keeps the
session, refreshing it as that client. `knotless delegate` waits on
`dev.knotless.getDelegation`. The grant is narrow: the four knotless
record types, blobs, and those collections in the owner's own spaces.
A confidential client's session lasts up to two years if it is used
every three months; the owner revokes it from their PDS's account
settings. Sharing a repository for writing records the gateway in its
`repo` record.
- **Authentication.** Every call carries a service token from
`com.atproto.server.getServiceAuth`, with the gateway as audience and the
method as `lxm`. The gateway checks it against the signing key in the
@@ -207,8 +241,9 @@ directly. A gateway is a `knotless-serve` with `--public-url`, identified as
eight times. A change to a path that also changed since its `base` is a
conflict (409) for that commit alone.
The gateway holds the owner's app password, so it is trusted with write
access to the owner's account: run your own, or use one you trust.
The gateway holds an OAuth session for the owner, so it is trusted with
writing knotless records to the owner's account: run your own, or use one
you trust.
## Pulls and issues
@@ -370,8 +405,8 @@ repo. `knotless` reads them with plain XRPC and does not check the commit
signature itself; `com.atproto.sync.getRecord` returns the proof a client
that wants to check it needs.
A gateway can write anything to an owner's account that its app password
allows; what it writes to a repository is still checked by every reader's
A gateway can write the knotless records its OAuth grant allows to an
owner's account; what it writes to a repository is still checked by every reader's
`index-pack`. Paths in commits are refused if they are empty, contain `.`,
`..` or `.git` segments, or contain NUL or newline characters.
harness/authorize.mjsnew+118−0
@@ -0,0 +1,118 @@
// Stands in for the user's browser on the reference PDS's authorization
// page: signs in, approves (or rejects) the request, and follows the
// redirect back to the client's loopback listener.
//
// node harness/authorize.mjs <handle> <password> [--reject] [--ca <cert>]
// <authorize-url>
//
// `knotless login` and `knotless delegate` run it as KNOTLESS_BROWSER, with
// the URL last. With --ca, it reaches the harness's gateway hostname as
// pds.mjs does: at 127.0.0.1, trusting that certificate. It speaks plain
// node:http(s), as fetch sets the sec-fetch-* headers itself.
import { request as http } from 'node:http'
import { request as https } from 'node:https'
import { readFileSync } from 'node:fs'
const args = process.argv.slice(2)
const reject = args.includes('--reject')
const at = args.indexOf('--ca')
const ca = at === -1 ? undefined : readFileSync(args.splice(at, 2)[1])
const [username, password, url] = args.filter((a) => a !== '--reject')
const authorize = new URL(url)
const origin = authorize.origin
const api = `${origin}/@atproto/oauth-provider/~api`
const jar = new Map()
const cookies = () =>
[...jar].map(([name, value]) => `${name}=${value}`).join('; ')
// The gateway hostname, as pds.mjs resolves it.
const local = (_hostname, options, callback) =>
options?.all
? callback(null, [{ address: '127.0.0.1', family: 4 }])
: callback(null, '127.0.0.1', 4)
// One request: its status, headers and body.
const send = (url, { method = 'GET', headers = {}, body } = {}) =>
new Promise((done, fail) => {
const secure = new URL(url).protocol === 'https:'
const tls = secure ? { ca, lookup: local } : {}
const req = (secure ? https : http)(url, { method, headers, ...tls }, (res) => {
const chunks = []
res.on('data', (c) => chunks.push(c))
res.on('end', () =>
done({ status: res.statusCode, headers: res.headers, body: Buffer.concat(chunks).toString() }),
)
})
req.on('error', fail)
req.end(body)
})
const keep = (res) => {
for (const cookie of res.headers['set-cookie'] ?? []) {
const [pair] = cookie.split(';')
const at = pair.indexOf('=')
jar.set(pair.slice(0, at), pair.slice(at + 1))
}
}
const page = await send(authorize, {
headers: {
'sec-fetch-mode': 'navigate',
'sec-fetch-dest': 'document',
'sec-fetch-site': 'cross-site',
},
})
keep(page)
// A request the server refuses goes straight back to the client.
if (page.headers.location) {
await send(page.headers.location)
process.exit(0)
}
if (page.status !== 200) throw new Error(`authorize page: ${page.status} ${page.body}`)
const call = async (endpoint, body) => {
const res = await send(`${api}${endpoint}`, {
method: 'POST',
headers: {
'content-type': 'application/json',
'sec-fetch-mode': 'same-origin',
'sec-fetch-site': 'same-origin',
origin,
referer: authorize.href,
'x-csrf-token': jar.get('csrf-token'),
cookie: cookies(),
},
body: JSON.stringify(body),
})
keep(res)
if (res.status !== 200) throw new Error(`${endpoint}: ${res.status} ${res.body}`)
return JSON.parse(res.body)
}
let done
if (reject) {
done = await call('/reject', {})
} else {
const { account } = await call('/sign-in', {
locale: 'en',
username,
password,
remember: true,
})
done = await call('/consent', { did: account.did })
}
const back = await send(done.url, {
headers: {
'sec-fetch-mode': 'navigate',
'sec-fetch-dest': 'document',
'sec-fetch-site': 'same-origin',
referer: authorize.href,
cookie: cookies(),
},
})
const location = back.headers.location
if (!location) throw new Error(`redirect: ${back.status} ${back.body}`)
await send(location)
harness/changed.mjs+1−0
@@ -67,6 +67,7 @@ const runs = (file) => {
}
if (m(/^crates\/knotless-serve\//)) return { runs: [['-p', 'knotless-serve']], pages: true }
if (m(/^harness\/shots\//)) return { runs: [], pages: true }
if (m(/^harness\/authorize\.mjs$/)) return { runs: [['-p', 'knotless', '--test', 'oauth']] }
return { runs: [] }
}
harness/pds.mjs+31−1
@@ -6,9 +6,19 @@
//
// KNOTLESS_DEV_ENV names a built dev-env package to use instead of the npm
// one, such as an atproto checkout's packages/dev-env on the spaces alpha.
//
// A gateway is an OAuth client, which the PDS reaches only over https at a
// hostname. So the harness makes a certificate for GATEWAY_HOST, trusts
// it, and resolves that name to 127.0.0.1; tests put harness/tls-proxy.mjs
// in front of a gateway with it.
import { fileURLToPath, pathToFileURL } from 'node:url'
import { dirname, resolve } from 'node:path'
import { dirname, join, resolve } from 'node:path'
import { execFileSync } from 'node:child_process'
import { mkdtempSync, readFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import dns from 'node:dns'
import tls from 'node:tls'
// Relative to the repository root.
const root = resolve(dirname(fileURLToPath(import.meta.url)), '..')
@@ -19,6 +29,25 @@ const { TestNetworkNoAppView } = await import(
: '@atproto/dev-env'
)
const GATEWAY_HOST = 'gateway.knotless.internal'
const certs = mkdtempSync(join(tmpdir(), 'knotless-tls-'))
const cert = join(certs, 'cert.pem')
const key = join(certs, 'key.pem')
execFileSync('openssl', [
'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:prime256v1',
'-nodes', '-keyout', key, '-out', cert, '-days', '7',
'-subj', `/CN=${GATEWAY_HOST}`, '-addext', `subjectAltName=DNS:${GATEWAY_HOST}`,
], { stdio: 'ignore' })
tls.setDefaultCACertificates([...tls.getCACertificates('default'), readFileSync(cert, 'utf8')])
const lookup = dns.lookup
dns.lookup = (hostname, options, callback) => {
if (typeof options === 'function') [callback, options] = [options, {}]
if (hostname !== GATEWAY_HOST) return lookup(hostname, options, callback)
return options?.all
? callback(null, [{ address: '127.0.0.1', family: 4 }])
: callback(null, '127.0.0.1', 4)
}
const arg = (name) => {
const i = process.argv.indexOf(name)
return i === -1 ? undefined : process.argv[i + 1]
@@ -61,6 +90,7 @@ console.log(
pds: net.pds.url,
plc: net.plc.url,
spaces: described.spacesAlpha === true,
gateway: { host: GATEWAY_HOST, cert, key },
accounts,
}),
)
harness/shots/baseline/darwin/home-desktop-dark.png+0−0
A binary file, not shown.
harness/shots/baseline/darwin/home-desktop-dark.sha256+1−1
@@ -1 +1 @@
7ebdd3b2b907ba4afa5e6ceff837c0ad9b15975f813b5b9299ab163807685c97
\ No newline at end of file
e590778aaa2564a14bb87e1a735d8d32a7eda34874212b6c591cf8bda5e7e6c5
\ No newline at end of file
harness/shots/baseline/darwin/home-desktop-light.png+0−0
A binary file, not shown.
harness/shots/baseline/darwin/home-desktop-light.sha256+1−1
@@ -1 +1 @@
7ebdd3b2b907ba4afa5e6ceff837c0ad9b15975f813b5b9299ab163807685c97
\ No newline at end of file
e590778aaa2564a14bb87e1a735d8d32a7eda34874212b6c591cf8bda5e7e6c5
\ No newline at end of file
harness/shots/baseline/darwin/home-phone-dark.png+0−0
A binary file, not shown.
harness/shots/baseline/darwin/home-phone-dark.sha256+1−1
@@ -1 +1 @@
7ebdd3b2b907ba4afa5e6ceff837c0ad9b15975f813b5b9299ab163807685c97
\ No newline at end of file
e590778aaa2564a14bb87e1a735d8d32a7eda34874212b6c591cf8bda5e7e6c5
\ No newline at end of file
harness/shots/baseline/darwin/home-phone-light.png+0−0
A binary file, not shown.
harness/shots/baseline/darwin/home-phone-light.sha256+1−1
@@ -1 +1 @@
7ebdd3b2b907ba4afa5e6ceff837c0ad9b15975f813b5b9299ab163807685c97
\ No newline at end of file
e590778aaa2564a14bb87e1a735d8d32a7eda34874212b6c591cf8bda5e7e6c5
\ No newline at end of file
harness/tls-proxy.mjsnew+29−0
@@ -0,0 +1,29 @@
// An https front for a gateway under test, at the harness's gateway
// hostname, passing every request on to it over plain http.
//
// node harness/tls-proxy.mjs <cert> <key> <target port>
//
// Prints the port it listens on, and exits when stdin closes.
import { readFileSync } from 'node:fs'
import { createServer } from 'node:https'
import { request } from 'node:http'
const [cert, key, target] = process.argv.slice(2)
const server = createServer({ cert: readFileSync(cert), key: readFileSync(key) }, (req, res) => {
const upstream = request(
{ host: '127.0.0.1', port: target, path: req.url, method: req.method, headers: req.headers },
(answer) => {
res.writeHead(answer.statusCode, answer.headers)
answer.pipe(res)
},
)
upstream.on('error', () => res.writeHead(502).end())
req.pipe(upstream)
})
server.listen(0, '127.0.0.1', () => console.log(server.address().port))
process.stdin.on('end', () => process.exit(0))
process.stdin.on('close', () => process.exit(0))
process.stdin.resume()
lexicons/dev/knotless/getDelegation.jsonnew+28−0
@@ -0,0 +1,28 @@
{
"lexicon": 1,
"id": "dev.knotless.getDelegation",
"description": "Waits, for a while, for a delegation the caller started with dev.knotless.putDelegation to be approved. Requires a service token from the caller's PDS, audience the gateway's DID, for this method.",
"defs": {
"main": {
"type": "query",
"description": "Fails with 403 when the authorization was refused or did not log in as the caller, and with 404 for a delegation the caller did not start or that has ended.",
"parameters": {
"type": "params",
"required": ["state"],
"properties": {
"state": { "type": "string", "maxLength": 64 }
}
},
"output": {
"encoding": "application/json",
"schema": {
"type": "object",
"required": ["done"],
"properties": {
"done": { "type": "boolean", "description": "Whether the gateway now writes for the caller. False means ask again." }
}
}
}
}
}
}
lexicons/dev/knotless/putDelegation.json+5−9
@@ -1,24 +1,20 @@
{
"lexicon": 1,
"id": "dev.knotless.putDelegation",
"description": "Delegates the caller's writes to a gateway by handing it an app password, replacing any it held. The gateway logs in with it and keeps the session, to write on the caller's behalf for collaborators and batched commits. Requires a service token from the caller's PDS, audience the gateway's DID, for this method.",
"description": "Starts delegating the caller's writes to a gateway. The gateway, an OAuth client of the caller's PDS, pushes an authorization request for the caller and answers with the page where the caller approves it; once they do, the gateway keeps the session, replacing any it held, to write on the caller's behalf for collaborators and batched commits. Requires a service token from the caller's PDS, audience the gateway's DID, for this method.",
"defs": {
"main": {
"type": "procedure",
"description": "Fails with 403 when the password does not log in as the caller.",
"input": {
"output": {
"encoding": "application/json",
"schema": {
"type": "object",
"required": ["password"],
"required": ["url", "state"],
"properties": {
"password": { "type": "string", "maxLength": 256, "description": "An app password for the caller's account." }
"url": { "type": "string", "format": "uri", "description": "The authorization page for the caller to open." },
"state": { "type": "string", "maxLength": 64, "description": "Names the delegation to dev.knotless.getDelegation." }
}
}
},
"output": {
"encoding": "application/json",
"schema": { "type": "object", "properties": {} }
}
}
}