mod common;
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::process::Output;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use base64::Engine as _;
use common::{commit, git};
use knot_runtime::{K256Signer, SeededEntropy, Signer as _};
use serde_json::{Value, json};
#[allow(dead_code)]
#[path = "common/cobs.rs"]
mod cobs;
const KNOT_OWNER: &str = "did:plc:knotknotknotknotknotknot";
const REPO_OWNER: &str = "did:plc:repoownerrepoownerrepoown";
const COLLABORATOR: &str = "did:plc:collabcollabcollabcollab";
const STRANGER: &str = "did:plc:strangerstrangerstrangers";
const KNOT_HOST: &str = "knot.phase0.test";
const MASTER_KEY_ENV: &str = "KNOT_STATIC_PHASE0_KEY";
const MASTER_KEY: [u8; 32] = [7; 32];
fn ssh_line(seed: u8) -> String {
let string = |bytes: &[u8]| [&(bytes.len() as u32).to_be_bytes()[..], bytes].concat();
let blob = [string(b"ssh-ed25519"), string(&[seed; 32])].concat();
format!(
"ssh-ed25519 {}",
base64::engine::general_purpose::STANDARD.encode(blob)
)
}
struct Stub {
url: String,
ca: PathBuf,
_dir: tempfile::TempDir,
published: Arc<Mutex<HashMap<String, Vec<String>>>>,
unreachable: Arc<Mutex<HashSet<String>>>,
slow: Arc<Mutex<HashSet<String>>>,
oauth: Arc<Oauth>,
}
#[derive(Default)]
struct Oauth {
on: AtomicBool,
states: Mutex<HashMap<String, String>>,
puts: Mutex<Vec<String>>,
}
impl Stub {
fn start() -> Self {
let signer = K256Signer::generate(&SeededEntropy::new(41));
let key = knot_types::crypto::multikey(0xe7, signer.public_key().as_bytes());
let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
let dir = tempfile::tempdir().unwrap();
let ca = dir.path().join("ca.pem");
std::fs::write(&ca, cert.cert.pem()).unwrap();
let chain = vec![cert.cert.der().clone()];
let private =
rustls::pki_types::PrivateKeyDer::try_from(cert.key_pair.serialize_der()).unwrap();
let tls = Arc::new(
rustls::ServerConfig::builder_with_provider(Arc::new(
rustls::crypto::ring::default_provider(),
))
.with_safe_default_protocol_versions()
.unwrap()
.with_no_client_auth()
.with_single_cert(chain, private)
.unwrap(),
);
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!(
"https://localhost:{}",
listener.local_addr().unwrap().port()
);
let stub = Self {
url: url.clone(),
ca,
_dir: dir,
published: Arc::default(),
unreachable: Arc::default(),
slow: Arc::default(),
oauth: Arc::default(),
};
let (published, unreachable, slow, oauth) = (
stub.published.clone(),
stub.unreachable.clone(),
stub.slow.clone(),
stub.oauth.clone(),
);
std::thread::spawn(move || {
for stream in listener.incoming().flatten() {
let (tls, url, key) = (tls.clone(), url.clone(), key.clone());
let (published, unreachable, slow, oauth) = (
published.clone(),
unreachable.clone(),
slow.clone(),
oauth.clone(),
);
std::thread::spawn(move || {
let connection = rustls::ServerConnection::new(tls).unwrap();
let stream = rustls::StreamOwned::new(connection, stream);
let _ =
Self::answer(stream, &url, &key, &published, &unreachable, &slow, &oauth);
});
}
});
stub
}
fn answer(
stream: impl std::io::Read + std::io::Write,
url: &str,
key: &str,
published: &Mutex<HashMap<String, Vec<String>>>,
unreachable: &Mutex<HashSet<String>>,
slow: &Mutex<HashSet<String>>,
oauth: &Oauth,
) -> std::io::Result<()> {
use std::io::{BufRead as _, Read as _};
let mut reader = std::io::BufReader::new(stream);
let mut request_line = String::new();
reader.read_line(&mut request_line)?;
let mut length = 0;
loop {
let mut header = String::new();
reader.read_line(&mut header)?;
if header.trim().is_empty() {
break;
}
if let Some(value) = header.to_ascii_lowercase().strip_prefix("content-length:") {
length = value.trim().parse().unwrap_or(0);
}
}
let mut body = vec![0; length];
reader.read_exact(&mut body)?;
let target = request_line.split_whitespace().nth(1).unwrap_or_default();
let form: HashMap<String, String> =
url::form_urlencoded::parse(&body).into_owned().collect();
let (status, answer) = if oauth.on.load(Ordering::SeqCst)
&& target.starts_with("/.well-known/oauth-authorization-server")
{
(
200,
json!({
"issuer": url,
"pushed_authorization_request_endpoint": format!("{url}/par"),
"authorization_endpoint": format!("{url}/authorize"),
"token_endpoint": format!("{url}/token"),
})
.to_string(),
)
} else if target.ends_with("/com.atproto.repo.putRecord") {
let put: Value = serde_json::from_slice(&body).unwrap_or_default();
oauth
.puts
.lock()
.unwrap()
.push(format!("{}/{}", put["repo"], put["rkey"]));
(200, "{}".to_string())
} else if target == "/par" {
oauth
.states
.lock()
.unwrap()
.insert(form["login_hint"].clone(), form["state"].clone());
(
201,
json!({ "request_uri": "urn:test:request" }).to_string(),
)
} else if target == "/token" {
// the code carries the account, which the token's `sub` takes.
let account = form["code"].strip_prefix("code-for-").unwrap_or_default();
(
200,
json!({
"access_token": "access", "refresh_token": "refresh",
"sub": account, "expires_in": 3600,
})
.to_string(),
)
} else if let Some(did) = target.strip_prefix("/did:plc:") {
let did = format!("did:plc:{did}");
(
200,
json!({
"id": did,
"alsoKnownAs": ["at://actor.test"],
"verificationMethod": [{
"id": format!("{did}#atproto"), "type": "Multikey",
"controller": did, "publicKeyMultibase": key,
}],
"service": [
{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": url },
{ "id": "#tangled_knot", "type": "TangledKnot", "serviceEndpoint": format!("https://{KNOT_HOST}") },
],
})
.to_string(),
)
} else if let Some((_, query)) = target.split_once("listRecords?") {
let did = query
.split('&')
.find_map(|pair| pair.strip_prefix("repo="))
.unwrap_or_default()
.replace("%3A", ":")
.replace("%3a", ":");
if slow.lock().unwrap().contains(&did) {
std::thread::sleep(std::time::Duration::from_secs(4));
}
if unreachable.lock().unwrap().contains(&did) {
(500, "{}".to_string())
} else {
let records: Vec<Value> = published
.lock()
.unwrap()
.get(&did)
.into_iter()
.flatten()
.map(|line| {
json!({
"uri": format!("at://{did}/org.tangled.key.sshKey/k"),
"value": { "$type": "org.tangled.key.sshKey", "publicKey": line, "name": "k", "createdAt": "2026-06-08T00:00:00Z" },
})
})
.collect();
(200, json!({ "records": records }).to_string())
}
} else {
(200, "{}".to_string())
};
let stream = reader.get_mut();
write!(
stream,
"HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{answer}",
answer.len()
)?;
stream.flush()
}
fn authorize(&self) {
self.oauth.on.store(true, Ordering::SeqCst);
}
fn publish(&self, account: &str, seeds: &[u8]) {
self.published.lock().unwrap().insert(
account.to_string(),
seeds.iter().map(|seed| ssh_line(*seed)).collect(),
);
}
}
struct Env {
_root: tempfile::TempDir,
stub: Stub,
config: PathBuf,
output: PathBuf,
source: PathBuf,
}
fn knot_static(env: &Env, args: &[&str]) -> Output {
common::knot_static_raw(
&env.config,
MASTER_KEY_ENV,
&MASTER_KEY,
args,
&[("KNOT_EXTRA_CA_FILE", env.stub.ca.to_str().unwrap())],
)
}
fn succeeds(output: Output) -> String {
common::assert_success(&output, "knot-static");
String::from_utf8(output.stdout).unwrap()
}
fn setup() -> Env {
let stub = Stub::start();
let root = tempfile::tempdir().unwrap();
let source = root.path().join("source");
let config = root.path().join("config.toml");
std::fs::create_dir_all(&source).unwrap();
git(&source, ["init", "-q", "-b", "main"]);
commit(&source, "README.md", "# sample\n", "initial commit");
std::fs::write(
&config,
format!(
r#"[knot]
hostname = "{KNOT_HOST}"
owner = "{KNOT_OWNER}"
[repo]
scan_path = "{}"
default_branch = "main"
object_format = "sha1"
[secrets]
sealed_key_file = "{}"
master_key_env = "{MASTER_KEY_ENV}"
[atproto]
plc_directory = "{}/"
allow_private_addresses = true
[output]
root = "{}"
[activate]
nginx_binary = "true"
[writes]
enabled = true
fastcgi_socket = "unix:/dev/null"
"#,
root.path().join("state").display(),
root.path().join("keys.sealed").display(),
stub.url,
root.path().join("output").display(),
),
)
.unwrap();
let env = Env {
output: root.path().join("output"),
_root: root,
stub,
config,
source,
};
succeeds(knot_static(&env, &["init"]));
for rkey in ["sample", "other"] {
succeeds(knot_static(
&env,
&[
"repo",
"import",
"--owner",
REPO_OWNER,
"--rkey",
rkey,
"--default-branch",
"main",
env.source.to_str().unwrap(),
],
));
}
env
}
impl Env {
fn repo(&self, rkey: &str) -> (String, PathBuf) {
let listed = succeeds(knot_static(self, &["repo", "list"]));
let row: Vec<&str> = listed
.lines()
.map(|line| line.split_whitespace().collect::<Vec<_>>())
.find(|row| row.contains(&rkey))
.unwrap_or_else(|| panic!("{rkey} in {listed}"));
(row[0].to_string(), PathBuf::from(row[row.len() - 1]))
}
fn collaborate(&self, rkey: &str, account: &str) {
let (did, path) = self.repo(rkey);
let signer = K256Signer::generate(&SeededEntropy::new(5));
let repo = knot_git::Repo::open(&path).unwrap();
let did = knot_types::RepoDid::new(did).unwrap();
cobs::Cobs {
repo: &repo,
did: &did,
signer: &signer,
}
.add_collaborator(
&knot_types::AccountDid::new(account).unwrap(),
&knot_types::AccountDid::new(REPO_OWNER).unwrap(),
10,
);
}
/// a grant writes the keys from a detached process, which is why this polls.
fn wait_for_keys(&self, account: &str, present: bool) {
for _ in 0..200 {
let stored = std::fs::read(self.output.join("ssh-keys.json"))
.ok()
.and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok());
if stored.is_some_and(|stored| stored.get(account).is_some() == present) {
return;
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
panic!("keys of {account} never became present={present}");
}
fn stored(&self) -> Value {
serde_json::from_slice(&std::fs::read(self.output.join("ssh-keys.json")).unwrap()).unwrap()
}
fn push(&self, rkey: &str, actors: &[&str], branch: &str) -> bool {
self.push_ref(rkey, actors, branch, &format!("refs/heads/{branch}"))
.status
.success()
}
fn push_ref(&self, rkey: &str, actors: &[&str], branch: &str, target: &str) -> Output {
let (mut push, _work) = self.push_command(rkey, actors, branch, target);
push.output().unwrap()
}
fn push_command(
&self,
rkey: &str,
actors: &[&str],
branch: &str,
target: &str,
) -> (std::process::Command, tempfile::TempDir) {
let dir = tempfile::tempdir().unwrap();
let work = dir.path().join("work");
git(
dir.path(),
["clone", "-q", self.source.to_str().unwrap(), "work"],
);
commit(&work, "pushed.txt", branch, "a push");
let actors: String = match actors {
[] => " --operator".to_string(),
_ => actors
.iter()
.map(|actor| format!(" --actor {actor}"))
.collect(),
};
let mut push = common::git_command(&work);
push.args([
"push",
&format!("git@host:{rkey}"),
&format!("HEAD:{target}"),
])
// git runs this through a shell, and the remote command is its last arg.
.env(
"GIT_SSH_COMMAND",
"f() { for last; do :; done; SSH_ORIGINAL_COMMAND=$last exec \"$KS\" --config \"$CFG\" git-ssh $ACTORS; }; f",
)
.env("GIT_SSH_VARIANT", "simple")
.env("KS", env!("CARGO_BIN_EXE_knot-static"))
.env("CFG", &self.config)
.env("ACTORS", actors)
.env(
MASTER_KEY_ENV,
base64::engine::general_purpose::STANDARD.encode(MASTER_KEY),
)
.env("KNOT_EXTRA_CA_FILE", &self.stub.ca);
(push, dir)
}
fn hosted(&self, rkey: &str) -> bool {
succeeds(knot_static(self, &["repo", "list"]))
.lines()
.any(|line| line.split_whitespace().any(|word| word == rkey))
}
fn has_branch(&self, rkey: &str, branch: &str) -> bool {
let (_, path) = self.repo(rkey);
common::git_raw(
&path,
[
"rev-parse",
"--verify",
"-q",
&format!("refs/heads/{branch}"),
],
)
.status
.success()
}
}
fn actors_of(listed: &str, key: &str) -> Vec<String> {
let line = listed
.lines()
.find(|line| line.ends_with(key))
.unwrap_or_else(|| panic!("{key} in {listed}"));
line.split("--actor ")
.skip(1)
.map(|rest| rest.split([' ', '"']).next().unwrap().to_string())
.collect()
}
#[test]
fn a_collaborators_key_pushes_where_it_was_granted_and_a_strangers_does_not() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
env.stub.publish(STRANGER, &[3]);
env.collaborate("sample", COLLABORATOR);
succeeds(knot_static(&env, &["keys", "sync"]));
let listed = succeeds(knot_static(
&env,
&["keys", "list", "--command", "/bin/git-ssh"],
));
assert_eq!(actors_of(&listed, &ssh_line(2)), [COLLABORATOR]);
assert!(
listed.contains("command=\"/bin/git-ssh --actor"),
"{listed}"
);
assert!(listed.contains("\",restrict ssh-ed25519 "), "{listed}");
assert!(
!listed.contains(&ssh_line(3)),
"a stranger's key isn't listed"
);
assert!(env.push("sample", &[COLLABORATOR], "collab"));
assert!(env.has_branch("sample", "collab"));
assert!(
!env.push("other", &[COLLABORATOR], "collab"),
"a grant on one repo isn't a grant on another"
);
assert!(!env.has_branch("other", "collab"));
assert!(!env.push("sample", &[STRANGER], "stranger"));
assert!(!env.has_branch("sample", "stranger"));
}
#[test]
fn a_sync_that_reads_the_same_keys_doesnt_rewrite_the_key_set() {
use std::os::unix::fs::MetadataExt as _;
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
env.collaborate("sample", COLLABORATOR);
let file = env.output.join("ssh-keys.json");
succeeds(knot_static(&env, &["keys", "sync"]));
let written = std::fs::metadata(&file).unwrap().ino();
succeeds(knot_static(&env, &["keys", "sync"]));
assert_eq!(std::fs::metadata(&file).unwrap().ino(), written);
env.stub.publish(COLLABORATOR, &[2, 4]);
succeeds(knot_static(&env, &["keys", "sync"]));
assert_ne!(std::fs::metadata(&file).unwrap().ino(), written);
}
#[test]
fn only_the_owner_provisions_a_repo_by_pushing_and_the_owners_own_keys_still_push() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
env.collaborate("sample", COLLABORATOR);
succeeds(knot_static(&env, &["keys", "sync"]));
assert!(!env.push("fresh", &[COLLABORATOR], "collab"));
assert!(!env.hosted("fresh"), "a collaborator doesn't provision");
assert!(env.push("fresh", &[KNOT_OWNER], "owner"));
assert!(env.hosted("fresh"));
assert!(
env.push("other", &[REPO_OWNER], "owner"),
"a repo's owner pushes it"
);
assert!(
env.push("other", &[], "operator"),
"a key in the ssh user's own authorized_keys names no actor and pushes anywhere"
);
}
#[test]
fn a_key_two_accounts_publish_names_both() {
let env = setup();
env.stub.publish(REPO_OWNER, &[1]);
env.stub.publish(COLLABORATOR, &[1]);
env.collaborate("sample", COLLABORATOR);
succeeds(knot_static(&env, &["keys", "sync"]));
let listed = succeeds(knot_static(&env, &["keys", "list", "--command", "x"]));
assert_eq!(listed.lines().count(), 1, "{listed}");
assert_eq!(actors_of(&listed, &ssh_line(1)), [COLLABORATOR, REPO_OWNER]);
}
#[test]
fn a_push_that_lands_nothing_says_nothing_of_the_declaration() {
let env = setup();
let notice = "the repo declaration cannot be published";
let stderr = |output: Output| String::from_utf8_lossy(&output.stderr).into_owned();
let landed = env.push_ref("other", &[REPO_OWNER], "owner", "refs/heads/owner");
assert!(landed.status.success());
let said = stderr(landed);
assert!(said.contains(notice), "{said}");
let active: Value =
serde_json::from_slice(&std::fs::read(env.output.join("active.json")).unwrap()).unwrap();
let generation = active["generation"].as_str().unwrap();
assert!(
said.contains(
"remote: push accepted, building...
"
),
"{said}"
);
assert!(
said.contains(&format!(
"remote: generation {} activated
",
&generation[..7]
)),
"{said}"
);
for leaked in [
env.output.to_str().unwrap(),
&generation[..8],
"completed generation build",
"generated ",
"firehose:",
"activated generation",
"gc:",
] {
assert!(
!said.contains(leaked),
"push leaked {leaked:?} in:
{said}"
);
}
let refused = env.push_ref("other", &[REPO_OWNER], "owner", "refs/hidden/x");
let refused = stderr(refused);
assert!(refused.contains("refs/hidden/x"), "{refused}");
assert!(!refused.contains(notice), "{refused}");
}
fn stderr_of(output: &Output) -> String {
String::from_utf8_lossy(&output.stderr).into_owned()
}
#[test]
fn a_push_that_isnt_the_owners_neither_waits_for_authorization_nor_prints_a_link() {
let env = setup();
env.stub.authorize();
env.stub.publish(COLLABORATOR, &[2]);
env.collaborate("sample", COLLABORATOR);
succeeds(knot_static(&env, &["keys", "sync"]));
let started = std::time::Instant::now();
for (actors, branch) in [(vec![COLLABORATOR], "collab"), (vec![], "operator")] {
let pushed = env.push_ref("sample", &actors, branch, &format!("refs/heads/{branch}"));
let stderr = stderr_of(&pushed);
assert!(pushed.status.success(), "{stderr}");
assert!(!stderr.contains("open this URL"), "{stderr}");
assert!(
stderr.contains(&format!("knot-static oauth login --owner {REPO_OWNER}")),
"{stderr}"
);
}
assert!(
started.elapsed().as_secs() < 60,
"a push waited for the owner"
);
}
#[test]
fn the_owners_own_push_asks_them_to_authorize_and_waits() {
use std::io::{BufRead as _, BufReader};
use std::os::unix::process::CommandExt as _;
use std::process::Stdio;
let env = setup();
env.stub.authorize();
for (actors, branch) in [(vec![REPO_OWNER], "owner"), (vec![], "operator-for-own")] {
// an empty actor list is the operator, who pushes with no session.
let rkey = if actors.is_empty() { "mine" } else { "sample" };
if actors.is_empty() {
succeeds(knot_static(
&env,
&[
"repo",
"import",
"--owner",
KNOT_OWNER,
"--rkey",
rkey,
"--default-branch",
"main",
env.source.to_str().unwrap(),
],
));
}
let (mut push, _work) = env.push_command(rkey, &actors, branch, "refs/heads/x");
let mut push = push
.stderr(Stdio::piped())
.process_group(0)
.spawn()
.unwrap();
let group = push.id().to_string();
let mut lines = BufReader::new(push.stderr.take().unwrap()).lines();
let mut seen = String::new();
let asked = lines.any(|line| {
let line = line.unwrap();
seen += &line;
line.contains("open this URL")
});
let _ = std::process::Command::new("kill")
.args(["-KILL", &format!("-{group}")])
.status();
let _ = push.wait();
assert!(asked, "{branch}: the push never asked to authorize: {seen}");
}
}
impl Env {
fn callback(&self, query: &str) -> String {
let output = common::knot_static_cmd(&self.config, MASTER_KEY_ENV, &MASTER_KEY, ["cgi"])
.env("REQUEST_METHOD", "GET")
.env("REQUEST_URI", format!("/oauth/callback?{query}"))
.output()
.unwrap();
String::from_utf8_lossy(&output.stdout).into_owned()
}
fn pending(&self) -> PathBuf {
self.output.join("oauth-pending")
}
fn pending_files(&self) -> Vec<String> {
let mut names: Vec<String> = std::fs::read_dir(self.pending())
.into_iter()
.flatten()
.map(|entry| entry.unwrap().file_name().into_string().unwrap())
.collect();
names.sort();
names
}
}
#[test]
fn two_pushes_waiting_to_authorize_each_get_their_own_code() {
let env = setup();
env.stub.authorize();
succeeds(knot_static(
&env,
&[
"repo",
"import",
"--owner",
KNOT_OWNER,
"--rkey",
"mine",
"--default-branch",
"main",
env.source.to_str().unwrap(),
],
));
let pushes: Vec<_> = [("sample", REPO_OWNER), ("mine", KNOT_OWNER)]
.into_iter()
.map(|(rkey, owner)| {
let (mut push, work) =
env.push_command(rkey, &[owner], "authorized", "refs/heads/authorized");
(
owner,
push.stderr(std::process::Stdio::piped()).spawn().unwrap(),
work,
)
})
.collect();
for _ in 0..300 {
if env.stub.oauth.states.lock().unwrap().len() == 2 {
break;
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
for (owner, ..) in &pushes {
let state = env.stub.oauth.states.lock().unwrap()[*owner].clone();
let answer = env.callback(&format!("code=code-for-{owner}&state={state}"));
assert!(answer.starts_with("Status: 200"), "{answer}");
}
for (owner, push, _work) in pushes {
let output = push.wait_with_output().unwrap();
let stderr = stderr_of(&output);
assert!(output.status.success(), "{stderr}");
assert!(
stderr.contains(&format!("Tokens saved for {owner}")),
"{stderr}"
);
}
assert_eq!(env.pending_files(), Vec::<String>::new());
}
#[test]
fn a_login_publishes_the_declarations_of_the_owners_repos() {
let env = setup();
env.stub.authorize();
let login = common::knot_static_cmd(
&env.config,
MASTER_KEY_ENV,
&MASTER_KEY,
["oauth", "login", "--owner", REPO_OWNER],
)
.env("KNOT_EXTRA_CA_FILE", &env.stub.ca)
.stderr(std::process::Stdio::piped())
.spawn()
.unwrap();
let state = loop {
if let Some(state) = env.stub.oauth.states.lock().unwrap().get(REPO_OWNER) {
break state.clone();
}
std::thread::sleep(std::time::Duration::from_millis(50));
};
env.callback(&format!("code=code-for-{REPO_OWNER}&state={state}"));
let output = login.wait_with_output().unwrap();
assert!(output.status.success(), "{}", stderr_of(&output));
let mut puts = env.stub.oauth.puts.lock().unwrap().clone();
puts.sort();
assert_eq!(
puts,
[
format!("\"{REPO_OWNER}\"/\"other\""),
format!("\"{REPO_OWNER}\"/\"sample\"")
]
);
}
#[test]
fn a_callback_that_names_no_waiting_flow_is_refused_and_writes_nothing() {
let env = setup();
let waiting = "0123456789abcdef0123456789abcdef";
let expired = "fedcba9876543210fedcba9876543210";
std::fs::create_dir_all(env.pending()).unwrap();
std::fs::File::create(env.pending().join(waiting)).unwrap();
let old = std::fs::File::create(env.pending().join(expired)).unwrap();
old.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(3600))
.unwrap();
let before = env.pending_files();
for query in [
"code=c".to_string(),
format!("code=c&state={}", "f".repeat(32)),
format!("code=c&state={expired}"),
format!("code=c&state={}", waiting.to_uppercase()),
format!("code=c&state={waiting}0"),
format!("code=c&state=../oauth-pending/{waiting}"),
format!("code=c&state={waiting}.code"),
format!("state={waiting}"),
format!("code={}&state={waiting}", "c".repeat(3000)),
format!("code=a%20b&state={waiting}"),
] {
let answer = env.callback(&query);
assert!(answer.starts_with("Status: 400"), "{query}: {answer}");
}
assert_eq!(env.pending_files(), [waiting.to_string()], "{before:?}");
let delivered = env.callback(&format!("code=first&state={waiting}"));
assert!(delivered.starts_with("Status: 200"), "{delivered}");
let again = env.callback(&format!("code=second&state={waiting}"));
assert!(again.starts_with("Status: 409"), "{again}");
assert_eq!(
std::fs::read_to_string(env.pending().join(format!("{waiting}.code"))).unwrap(),
"first"
);
assert_eq!(env.pending_files().len(), 2);
}
#[test]
fn an_account_that_published_a_key_twice_is_named_once() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2, 2, 2]);
env.collaborate("sample", COLLABORATOR);
succeeds(knot_static(&env, &["keys", "sync"]));
let listed = succeeds(knot_static(&env, &["keys", "list", "--command", "x"]));
assert_eq!(actors_of(&listed, &ssh_line(2)), [COLLABORATOR]);
}
#[test]
fn an_activation_reads_only_new_accounts_and_sync_reads_everyone() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
env.collaborate("sample", COLLABORATOR);
succeeds(knot_static(&env, &["keys", "sync"]));
let (did, _) = env.repo("sample");
let newcomer = "did:plc:newcomernewcomernewcomer";
env.stub.publish(COLLABORATOR, &[9]);
env.stub.publish(newcomer, &[8]);
env.collaborate("other", newcomer);
succeeds(knot_static(&env, &["post-activate", "--repo", &did]));
let stored = env.stored();
assert_eq!(stored[newcomer], json!([ssh_line(8)]));
assert_eq!(
stored[COLLABORATOR],
json!([ssh_line(2)]),
"a known account's keys wait for sync"
);
succeeds(knot_static(&env, &["keys", "sync"]));
assert_eq!(env.stored()[COLLABORATOR], json!([ssh_line(9)]));
}
#[test]
fn sync_forgets_accounts_that_no_longer_matter_and_keeps_keys_it_couldnt_read() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
env.collaborate("sample", COLLABORATOR);
succeeds(knot_static(&env, &["keys", "sync"]));
let mut stored = env.stored();
stored[STRANGER] = json!([ssh_line(3)]);
std::fs::write(env.output.join("ssh-keys.json"), stored.to_string()).unwrap();
env.stub.publish(COLLABORATOR, &[9]);
env.stub
.unreachable
.lock()
.unwrap()
.insert(COLLABORATOR.to_string());
let failed = knot_static(&env, &["keys", "sync"]);
assert!(!failed.status.success());
assert!(String::from_utf8_lossy(&failed.stderr).contains(COLLABORATOR));
let stored = env.stored();
assert_eq!(stored[COLLABORATOR], json!([ssh_line(2)]));
assert!(stored.get(STRANGER).is_none());
}
const GRANT: &str = "org.tangled.graph.collaborator";
fn service_jwt(account: &str, repo: &str, nsid: &str, jti: &str) -> String {
let b64 = |bytes: &[u8]| base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let claims = json!({ "iss": account, "aud": repo, "iat": now, "exp": now + 60, "jti": jti, "lxm": nsid });
let signing_input = format!(
"{}.{}",
b64(br#"{"alg":"ES256K","typ":"JWT"}"#),
b64(claims.to_string().as_bytes())
);
let signer = K256Signer::generate(&SeededEntropy::new(41));
format!(
"{signing_input}.{}",
b64(signer.sign(signing_input.as_bytes()).as_bytes())
)
}
impl Env {
fn grant_over_cgi(&self, account: &str, method: &str, subject: &str) -> (u16, Value) {
let (did, _) = self.repo("sample");
let nsid = format!("com.atproto.repo.{method}");
let mut body = json!({ "repo": did, "collection": GRANT, "rkey": subject });
if method == "createRecord" {
body["record"] = json!({ "$type": GRANT, "createdAt": "2026-01-01T00:00:00Z" });
}
let body = body.to_string();
let token = service_jwt(account, &did, &nsid, &format!("{method}-{subject}"));
let mut child = common::knot_static_cmd(&self.config, MASTER_KEY_ENV, &MASTER_KEY, ["cgi"])
.env("REQUEST_METHOD", "POST")
.env("REQUEST_URI", format!("/xrpc/{nsid}"))
.env("CONTENT_LENGTH", body.len().to_string())
.env("HTTP_AUTHORIZATION", format!("Bearer {token}"))
.env("KNOT_EXTRA_CA_FILE", &self.stub.ca)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.spawn()
.unwrap();
std::io::Write::write_all(&mut child.stdin.take().unwrap(), body.as_bytes()).unwrap();
let out = String::from_utf8(child.wait_with_output().unwrap().stdout).unwrap();
let (head, json) = out.split_once("\r\n\r\n").expect("a cgi answer");
let status = head
.lines()
.find_map(|line| line.strip_prefix("Status: "))
.and_then(|status| status.split(' ').next())
.and_then(|code| code.parse().ok())
.expect("a Status header");
(status, serde_json::from_str(json).unwrap())
}
}
#[test]
fn the_owner_adds_and_removes_a_collaborator_from_the_cli() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
assert!(!env.push("sample", &[COLLABORATOR], "before"));
let added = succeeds(knot_static(
&env,
&["repo", "collaborator", "add", "sample", COLLABORATOR],
));
assert_eq!(added.trim(), "{}");
env.wait_for_keys(COLLABORATOR, true);
assert_eq!(env.stored()[COLLABORATOR], json!([ssh_line(2)]));
assert!(env.push("sample", &[COLLABORATOR], "after"));
succeeds(knot_static(
&env,
&["repo", "collaborator", "rm", "sample", COLLABORATOR],
));
env.wait_for_keys(COLLABORATOR, false);
assert!(!env.push("sample", &[COLLABORATOR], "revoked"));
assert!(!env.has_branch("sample", "revoked"));
}
#[test]
fn only_the_owner_grants_a_collaborator_over_the_cgi_and_the_answer_is_empty() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
env.stub.publish(STRANGER, &[3]);
let (status, refused) = env.grant_over_cgi(STRANGER, "createRecord", STRANGER);
assert_eq!(status, 403, "{refused}");
assert!(
refused
.to_string()
.contains("only repository owner may manage collaborators")
);
assert!(!env.push("sample", &[STRANGER], "stranger"));
let (status, added) = env.grant_over_cgi(REPO_OWNER, "createRecord", COLLABORATOR);
assert_eq!((status, &added), (200, &json!({})));
env.wait_for_keys(COLLABORATOR, true);
assert!(env.push("sample", &[COLLABORATOR], "granted"));
let (status, refused) = env.grant_over_cgi(COLLABORATOR, "createRecord", STRANGER);
assert_eq!(status, 403, "a collaborator doesn't grant: {refused}");
let (status, refused) = env.grant_over_cgi(COLLABORATOR, "deleteRecord", COLLABORATOR);
assert_eq!(status, 403, "{refused}");
let (status, removed) = env.grant_over_cgi(REPO_OWNER, "deleteRecord", COLLABORATOR);
assert_eq!((status, &removed), (200, &json!({})));
assert!(!env.push("sample", &[COLLABORATOR], "revoked"));
}
#[test]
fn a_slow_full_sync_doesnt_overwrite_a_grant_made_while_it_runs() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
env.collaborate("sample", COLLABORATOR);
succeeds(knot_static(&env, &["keys", "sync"]));
env.stub
.slow
.lock()
.unwrap()
.insert(COLLABORATOR.to_string());
let mut slow =
common::knot_static_cmd(&env.config, MASTER_KEY_ENV, &MASTER_KEY, ["keys", "sync"])
.env("KNOT_EXTRA_CA_FILE", &env.stub.ca)
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
std::thread::sleep(std::time::Duration::from_millis(1000));
let newcomer = "did:plc:newcomernewcomernewcomer";
env.stub.publish(newcomer, &[8]);
succeeds(knot_static(
&env,
&["repo", "collaborator", "add", "sample", newcomer],
));
assert!(slow.wait().unwrap().success());
env.wait_for_keys(newcomer, true);
}
#[test]
fn git_ssh_refuses_to_guess_who_is_pushing() {
let env = setup();
let run = |flags: &[&str]| {
let mut args = vec!["git-ssh"];
args.extend(flags);
let mut cmd = common::knot_static_cmd(&env.config, MASTER_KEY_ENV, &MASTER_KEY, args);
cmd.env("SSH_ORIGINAL_COMMAND", "git-upload-pack 'sample'")
.env("KNOT_EXTRA_CA_FILE", &env.stub.ca)
.output()
.unwrap()
};
let bare = run(&[]);
assert!(!bare.status.success());
assert!(String::from_utf8_lossy(&bare.stderr).contains("--operator or --actor"));
assert!(
!run(&["--operator", "--actor", COLLABORATOR])
.status
.success()
);
let operator = run(&["--operator"]);
assert!(!String::from_utf8_lossy(&operator.stderr).contains("--operator or --actor"));
}
#[test]
fn a_push_finds_its_repo_by_did_or_by_owner_and_name() {
let env = setup();
env.stub.publish(COLLABORATOR, &[2]);
env.collaborate("sample", COLLABORATOR);
let (did, _) = env.repo("sample");
let (other, _) = env.repo("other");
assert!(env.push(&did, &[COLLABORATOR], "by-did"));
assert!(env.has_branch("sample", "by-did"));
assert!(env.push(
&format!("{REPO_OWNER}/sample.git"),
&[COLLABORATOR],
"by-owner"
));
assert!(env.has_branch("sample", "by-owner"));
assert!(!env.push(&other, &[COLLABORATOR], "ungranted"));
assert!(!env.push(&format!("{REPO_OWNER}/other"), &[COLLABORATOR], "ungranted"));
assert!(!env.has_branch("other", "ungranted"));
}
#[test]
fn a_path_that_names_a_repo_this_knot_doesnt_host_is_not_a_new_repo() {
let env = setup();
let missing = "did:plc:missingmissingmissingmiss";
assert!(!env.push(missing, &[KNOT_OWNER], "owner"));
assert!(!env.push(missing, &[], "operator"));
assert!(!env.push(
&format!("{REPO_OWNER}/fresh"),
&[KNOT_OWNER],
"someone-elses"
));
assert!(!env.hosted(missing));
assert!(!env.hosted("fresh"));
assert!(env.push(&format!("{KNOT_OWNER}/fresh"), &[KNOT_OWNER], "own"));
assert!(env.hosted("fresh"));
}
#[test]
fn a_bare_rkey_two_owners_share_is_refused_until_it_names_its_owner() {
let env = setup();
succeeds(knot_static(
&env,
&[
"repo",
"import",
"--owner",
KNOT_OWNER,
"--rkey",
"sample",
"--default-branch",
"main",
env.source.to_str().unwrap(),
],
));
assert!(!env.push("sample", &[], "ambiguous"));
assert!(env.push(&format!("{KNOT_OWNER}/sample"), &[], "named"));
assert!(env.push(&format!("{REPO_OWNER}/sample"), &[], "named-too"));
}