mod common;
use std::path::PathBuf;
use std::process::{Output, Stdio};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use base64::Engine as _;
use common::{commit, git};
use knot_runtime::{K256Signer, PublicKeyBytes, SeededEntropy, Signature, 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 MASTER_KEY_ENV: &str = "KNOT_STATIC_PHASE0_KEY";
const MASTER_KEY: [u8; 32] = [7; 32];
const KNOT_HOST: &str = "knot.phase0.test";
const KNOT_DID: &str = "did:web:knot.phase0.test";
const NOTIFY: &str = "org.tangled.event.notifyEvent";
const ZEROS: &str = "0000000000000000000000000000000000000000";
fn b64() -> base64::engine::GeneralPurpose {
base64::engine::general_purpose::URL_SAFE_NO_PAD
}
#[derive(Default)]
struct Spindle {
down: AtomicBool,
attempts: AtomicUsize,
arrived: Mutex<Vec<Instant>>,
taken: Mutex<Vec<Value>>,
knot_key: Mutex<Option<PublicKeyBytes>>,
did: Mutex<String>,
served: Mutex<Option<(PathBuf, String)>>,
withhold_ms: AtomicU64,
first_poll: Mutex<Option<Instant>>,
served_at: Mutex<Option<Instant>>,
crawls: AtomicUsize,
advertised: Mutex<Vec<bool>>,
}
impl Spindle {
/// workers can lag a reload. this makes `listRepos` return rev 0 for `ms` after the next poll.
fn withhold(&self, ms: u64) {
self.withhold_ms.store(ms, Ordering::SeqCst);
*self.first_poll.lock().unwrap() = None;
*self.served_at.lock().unwrap() = None;
}
/// 503 when down, 401 unless the jwt is the knot's, for this spindle and method.
fn notify(&self, authorization: &str, body: &[u8]) -> (u16, String) {
self.attempts.fetch_add(1, Ordering::SeqCst);
self.arrived.lock().unwrap().push(Instant::now());
if self.down.load(Ordering::SeqCst) {
return (503, String::new());
}
if !self.authorized(authorization) {
return (
401,
json!({ "error": "AuthenticationRequired" }).to_string(),
);
}
let body: Value = serde_json::from_slice(body).unwrap();
let reference = body["event"]["ref"].as_str().unwrap().to_string();
let shows_tip = self.shows_tip(&body["event"]);
self.advertised.lock().unwrap().push(shows_tip);
self.taken.lock().unwrap().push(body);
(
200,
json!({ "ci": { "pipelines": ["3kpipeline"], "message": format!("follow {reference}") } })
.to_string(),
)
}
fn generation(&self) -> Option<(PathBuf, String)> {
let (output, did) = self.served.lock().unwrap().clone()?;
let active = std::fs::read(output.join("active.json")).ok()?;
let active: Value = serde_json::from_slice(&active).ok()?;
let directory = output
.join("generations")
.join(active["generation"].as_str()?);
Some((directory, common::repo_token(&did)))
}
fn shows_tip(&self, event: &Value) -> bool {
let Some((directory, token)) = self.generation() else {
return false;
};
let advertisement = directory
.join("www/_git")
.join(token)
.join("smart/advertisement");
let tip = format!(
"{} {}",
event["newSha"].as_str().unwrap(),
event["ref"].as_str().unwrap()
);
std::fs::read(advertisement).is_ok_and(|bytes| {
let tip = tip.as_bytes();
bytes.windows(tip.len()).any(|window| window == tip)
})
}
fn list_repos(&self) -> String {
let (Some((directory, token)), Some((_, did))) =
(self.generation(), self.served.lock().unwrap().clone())
else {
return "{}".into();
};
let first = *self
.first_poll
.lock()
.unwrap()
.get_or_insert_with(Instant::now);
let rev =
if first.elapsed() < Duration::from_millis(self.withhold_ms.load(Ordering::SeqCst)) {
"0".to_string()
} else {
self.served_at
.lock()
.unwrap()
.get_or_insert_with(Instant::now);
let latest = std::fs::read(
directory
.join("www/_atproto")
.join(token)
.join("latest.json"),
)
.unwrap();
serde_json::from_slice::<Value>(&latest).unwrap()["rev"]
.as_str()
.unwrap()
.to_string()
};
json!({ "repos": [{ "did": did, "rev": rev }] }).to_string()
}
fn authorized(&self, authorization: &str) -> bool {
let Some(token) = authorization.strip_prefix("Bearer ") else {
return false;
};
let parts: Vec<&str> = token.split('.').collect();
let [header, payload, signature] = parts[..] else {
return false;
};
let Ok(claims) = b64().decode(payload) else {
return false;
};
let claims: Value = serde_json::from_slice(&claims).unwrap();
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let key = self
.knot_key
.lock()
.unwrap()
.clone()
.expect("the knot's key");
let signed = b64().decode(signature).map(Signature::from_bytes);
claims["iss"] == KNOT_DID
&& claims["aud"] == self.did.lock().unwrap().as_str()
&& claims["lxm"] == NOTIFY
&& claims["exp"].as_u64().is_some_and(|exp| exp > now)
&& signed.is_ok_and(|signature| {
knot_runtime::verify(&key, format!("{header}.{payload}").as_bytes(), &signature)
})
}
}
/// one loopback https server is both the plc directory and the spindle. every did:plc gets the same actor key.
struct Stub {
url: String,
proxy: String,
ca: PathBuf,
_dir: tempfile::TempDir,
spindle: Arc<Spindle>,
}
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(),
KNOT_HOST.to_string(),
])
.unwrap();
let dir = tempfile::tempdir().unwrap();
let ca = dir.path().join("ca.pem");
std::fs::write(&ca, cert.cert.pem()).unwrap();
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(vec![cert.cert.der().clone()], private)
.unwrap(),
);
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let url = format!("https://localhost:{}", address.port());
let spindle = Arc::new(Spindle::default());
*spindle.did.lock().unwrap() = format!("did:web:localhost%3A{}", address.port());
let served = (url.clone(), spindle.clone());
std::thread::spawn(move || {
for stream in listener.incoming().flatten() {
let (tls, key, (url, spindle)) = (tls.clone(), key.clone(), served.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, &spindle);
});
}
});
Self {
proxy: Self::proxy(address),
url,
ca,
_dir: dir,
spindle,
}
}
/// the knot's host only resolves in production, and this proxy tunnels its `CONNECT`s to the stub.
fn proxy(to: std::net::SocketAddr) -> String {
use std::io::BufRead as _;
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
std::thread::spawn(move || {
for client in listener.incoming().flatten() {
std::thread::spawn(move || {
let mut requested = std::io::BufReader::new(client.try_clone().unwrap());
loop {
let mut line = String::new();
if requested.read_line(&mut line).unwrap_or(0) == 0
|| line.trim().is_empty()
{
break;
}
}
let Ok(upstream) = std::net::TcpStream::connect(to) else {
return;
};
let mut client = client;
let _ = std::io::Write::write_all(
&mut client,
b"HTTP/1.1 200 Connection established\r\n\r\n",
);
let (mut up, mut down) =
(upstream.try_clone().unwrap(), client.try_clone().unwrap());
let (mut upstream, mut client) = (upstream, client);
std::thread::spawn(move || std::io::copy(&mut client, &mut up));
let _ = std::io::copy(&mut upstream, &mut down);
});
}
});
url
}
fn env(&self) -> Vec<(&'static str, String)> {
vec![
("KNOT_EXTRA_CA_FILE", self.ca.display().to_string()),
("HTTPS_PROXY", self.proxy.clone()),
("NO_PROXY", "localhost,127.0.0.1".to_string()),
]
}
fn answer(
stream: impl std::io::Read + std::io::Write,
url: &str,
key: &str,
spindle: &Spindle,
) -> 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, mut authorization) = (0, String::new());
loop {
let mut header = String::new();
reader.read_line(&mut header)?;
if header.trim().is_empty() {
break;
}
let lowered = header.to_ascii_lowercase();
if let Some(value) = lowered.strip_prefix("content-length:") {
length = value.trim().parse().unwrap_or(0);
}
if lowered.starts_with("authorization:") {
authorization = header["authorization:".len()..].trim().to_string();
}
}
let mut body = vec![0; length];
reader.read_exact(&mut body)?;
let target = request_line.split_whitespace().nth(1).unwrap_or_default();
if target.contains("requestCrawl") {
spindle.crawls.fetch_add(1, Ordering::SeqCst);
}
let (status, answer) = if target.ends_with(NOTIFY) {
spindle.notify(&authorization, &body)
} else if target == "/.well-known/did.json" {
let did = spindle.did.lock().unwrap().clone();
let document = json!({
"id": did,
"service": [{ "id": "#tangled_spindle", "type": "TangledSpindle", "serviceEndpoint": url }],
});
(200, document.to_string())
} else if target.contains("listRepos") {
(200, spindle.list_repos())
} else if target.contains("listRecords?") {
(200, json!({ "records": [] }).to_string())
} else if let Some(did) = target.strip_prefix("/did:plc:") {
let did = format!("did:plc:{did}");
let document = 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}") },
],
});
(200, document.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 attempts(&self) -> usize {
self.spindle.attempts.load(Ordering::SeqCst)
}
fn taken(&self) -> Vec<Value> {
self.spindle.taken.lock().unwrap().clone()
}
fn wait_for_event(&self, reference: &str, new: &str) -> Value {
let deadline = Instant::now() + Duration::from_secs(20);
while Instant::now() < deadline {
let taken = self.taken();
let found = (taken.iter())
.find(|body| body["event"]["ref"] == reference && body["event"]["newSha"] == new);
if let Some(body) = found {
return body["event"].clone();
}
std::thread::sleep(Duration::from_millis(100));
}
panic!(
"the spindle never took {reference} at {new}: {:?}",
self.taken()
);
}
fn wait_for_taken(&self, count: usize) -> Vec<Value> {
let deadline = Instant::now() + Duration::from_secs(20);
while Instant::now() < deadline {
let taken = self.taken();
if taken.len() >= count {
return taken;
}
std::thread::sleep(Duration::from_millis(100));
}
panic!("the spindle never took {count} events: {:?}", self.taken());
}
}
struct Env {
_root: tempfile::TempDir,
stub: Stub,
config: PathBuf,
output: PathBuf,
work: PathBuf,
repo_did: String,
}
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 crawler = stub.url.clone();
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"
[crawl]
urls = ["{crawler}"]
[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 mut env = Env {
output: root.path().join("output"),
work: root.path().join("work"),
_root: root,
stub,
config,
repo_did: String::new(),
};
succeeds(env.knot(&["init"]));
succeeds(env.knot(&[
"repo",
"import",
"--owner",
REPO_OWNER,
"--rkey",
"sample",
"--default-branch",
"main",
source.to_str().unwrap(),
]));
let listed = succeeds(env.knot(&["repo", "list"]));
env.repo_did = listed.split_whitespace().next().unwrap().to_string();
git(
env.work.parent().unwrap(),
["clone", "-q", source.to_str().unwrap(), "work"],
);
*env.stub.spindle.served.lock().unwrap() = Some((env.output.clone(), env.repo_did.clone()));
let sealed = knot_secrets::SealedStore::open(
env.work.parent().unwrap().join("keys.sealed"),
&knot_secrets::MasterKey::new(MASTER_KEY.to_vec()).unwrap(),
Box::new(knot_runtime::OsEntropy),
)
.unwrap();
let knot = knot_types::KnotId::new(KNOT_DID).unwrap();
*env.stub.spindle.knot_key.lock().unwrap() = Some(sealed.public_key(&knot).unwrap());
env
}
impl Env {
fn knot(&self, args: &[&str]) -> Output {
let env = self.stub.env();
let env: Vec<_> = env
.iter()
.map(|(name, value)| (*name, value.as_str()))
.collect();
common::knot_static_raw(&self.config, MASTER_KEY_ENV, &MASTER_KEY, args, &env)
}
fn spindle_did(&self) -> String {
self.stub.spindle.did.lock().unwrap().clone()
}
fn set_spindle(&self) {
succeeds(self.knot(&[
"repo",
"settings",
"sample",
"--spindle",
&self.spindle_did(),
]));
}
fn collaborate(&self, account: &str) {
let listed = succeeds(self.knot(&["repo", "list"]));
let path = listed.split_whitespace().last().unwrap().to_string();
let signer = K256Signer::generate(&SeededEntropy::new(5));
let repo = knot_git::Repo::open(std::path::Path::new(&path)).unwrap();
let did = knot_types::RepoDid::new(&self.repo_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,
);
}
fn commit(&self, what: &str) -> String {
commit(&self.work, "pushed.txt", what, what);
git(&self.work, ["rev-parse", "HEAD"])
}
fn push(&self, actors: &[&str], refspecs: &[&str]) -> Output {
self.push_with(actors, refspecs, &[])
}
fn push_with(&self, actors: &[&str], refspecs: &[&str], extra: &[(&str, &str)]) -> Output {
let forced: String = match actors {
[] => " --operator".to_string(),
_ => actors
.iter()
.map(|actor| format!(" --actor {actor}"))
.collect(),
};
common::git_command(&self.work)
.args(["push", "git@host:sample"])
.args(refspecs)
// git runs `GIT_SSH_COMMAND` through a shell, with the host and remote command after it.
.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", forced)
.env(
MASTER_KEY_ENV,
base64::engine::general_purpose::STANDARD.encode(MASTER_KEY),
)
.envs(self.stub.env())
.envs(extra.iter().copied())
.output()
.unwrap()
}
fn outbox(&self) -> Option<Value> {
let bytes = std::fs::read(self.output.join("spindle-outbox.json")).ok()?;
serde_json::from_slice(&bytes).ok()
}
fn queued(&self) -> usize {
self.outbox()
.map_or(0, |outbox| outbox["entries"].as_array().unwrap().len())
}
fn wait_for_empty_outbox(&self) {
let deadline = Instant::now() + Duration::from_secs(20);
while self.queued() > 0 {
assert!(Instant::now() < deadline, "the outbox never drained");
std::thread::sleep(Duration::from_millis(100));
}
}
}
fn event<'a>(taken: &'a [Value], reference: &str) -> &'a Value {
&taken
.iter()
.find(|body| body["event"]["ref"] == reference)
.unwrap_or_else(|| panic!("an event for {reference} in {taken:?}"))["event"]
}
fn stderr(output: &Output) -> String {
String::from_utf8_lossy(&output.stderr).into_owned()
}
#[test]
fn an_ssh_push_tells_the_spindle_once_per_moved_public_ref_as_the_pusher() {
let env = setup();
env.collaborate(COLLABORATOR);
env.set_spindle();
let first = env.commit("one");
let pushed = env.push(
&[COLLABORATOR],
&["HEAD:refs/heads/feat", "HEAD:refs/tags/v1"],
);
assert!(pushed.status.success(), "{}", stderr(&pushed));
// the push only answers for the events taken by then. the rest go out in post-activate.
let answered = env.stub.taken();
let taken = env.stub.wait_for_taken(2);
assert_eq!(taken.len(), 2, "{taken:?}");
for reference in ["refs/heads/feat", "refs/tags/v1"] {
assert_eq!(
taken
.iter()
.find(|body| body["event"]["ref"] == reference)
.unwrap(),
&json!({
"repo": env.repo_did,
"event": {
"$type": "org.tangled.event.push",
"repo": env.repo_did,
"ref": reference,
"oldSha": ZEROS,
"newSha": first,
"actor": COLLABORATOR,
},
})
);
}
let said = stderr(&pushed);
for body in &answered {
let reference = body["event"]["ref"].as_str().unwrap();
assert!(
said.contains(&format!("remote: follow {reference}")),
"{said}"
);
}
let second = env.commit("two");
assert!(
env.push(&[COLLABORATOR], &["HEAD:refs/heads/feat"])
.status
.success()
);
let moved = env.stub.wait_for_event("refs/heads/feat", &second);
assert_eq!(moved["oldSha"], json!(first));
assert!(
env.push(&[COLLABORATOR], &[":refs/heads/feat"])
.status
.success()
);
let gone = env.stub.wait_for_event("refs/heads/feat", ZEROS);
assert_eq!(gone["oldSha"], json!(second));
env.wait_for_empty_outbox();
}
#[test]
fn the_operator_pushes_as_the_repos_owner_and_a_shared_key_as_the_account_that_may_push() {
let env = setup();
env.collaborate(COLLABORATOR);
env.set_spindle();
env.commit("operator");
assert!(
env.push(&[], &["HEAD:refs/heads/by-operator"])
.status
.success()
);
env.commit("shared");
let shared = env.push(&[STRANGER, COLLABORATOR], &["HEAD:refs/heads/by-shared"]);
assert!(shared.status.success(), "{}", stderr(&shared));
let taken = env.stub.wait_for_taken(2);
assert_eq!(event(&taken, "refs/heads/by-operator")["actor"], REPO_OWNER);
assert_eq!(event(&taken, "refs/heads/by-shared")["actor"], COLLABORATOR);
}
#[test]
fn a_repo_without_a_spindle_sends_nothing() {
let env = setup();
env.commit("no spindle yet");
assert!(env.push(&[], &["HEAD:refs/heads/a"]).status.success());
assert_eq!(env.stub.attempts(), 0);
assert!(env.outbox().is_none());
env.set_spindle();
env.commit("a spindle");
assert!(env.push(&[], &["HEAD:refs/heads/b"]).status.success());
env.stub.wait_for_taken(1);
env.wait_for_empty_outbox();
succeeds(env.knot(&["repo", "settings", "sample", "--no-spindle"]));
env.commit("no spindle again");
assert!(env.push(&[], &["HEAD:refs/heads/c"]).status.success());
assert_eq!(
env.stub.attempts(),
1,
"clearing the spindle stops the events"
);
}
#[test]
fn only_a_push_to_a_repo_with_a_spindle_waits_to_be_served() {
let env = setup();
let unserved = [
("HTTPS_PROXY", "http://127.0.0.1:9"),
("RUST_LOG", "knot_static::crawl=debug"),
];
env.commit("no spindle");
let pushed = env.push_with(&[], &["HEAD:refs/heads/a"], &unserved);
assert!(pushed.status.success(), "{}", stderr(&pushed));
assert!(
!stderr(&pushed).contains("listRepos poll failed"),
"{}",
stderr(&pushed)
);
env.set_spindle();
env.commit("a spindle");
let pushed = env.push_with(&[], &["HEAD:refs/heads/b"], &unserved);
assert!(pushed.status.success(), "{}", stderr(&pushed));
assert!(
stderr(&pushed).contains("listRepos poll failed"),
"{}",
stderr(&pushed)
);
assert_eq!(env.stub.attempts(), 0, "no attempt before it is served");
}
#[test]
fn a_change_that_moves_no_public_ref_sends_nothing() {
let env = setup();
env.set_spindle();
env.commit("aimed at a cob ref");
let refused = env.push(&[], &["HEAD:refs/cobs/org.tangled.track.ticket/1"]);
assert!(!refused.status.success(), "a push can't write cob refs");
let mut ticket = common::knot_static_cmd(
&env.config,
MASTER_KEY_ENV,
&MASTER_KEY,
["write", "createRecord", "--repo", &env.repo_did],
);
ticket.stdin(Stdio::piped()).stdout(Stdio::piped());
let mut child = ticket.envs(env.stub.env()).spawn().unwrap();
std::io::Write::write_all(
&mut child.stdin.take().unwrap(),
json!({
"repo": env.repo_did,
"collection": "org.tangled.track.ticket",
"record": { "title": "an issue", "body": "it breaks", "createdAt": "2026-01-01T00:00:00Z" },
})
.to_string()
.as_bytes(),
)
.unwrap();
common::assert_success(&child.wait_with_output().unwrap(), "a cob write");
assert_eq!(env.stub.attempts(), 0);
assert!(env.outbox().is_none());
}
#[test]
fn a_spindle_that_is_down_keeps_its_events_for_post_activate() {
let env = setup();
env.set_spindle();
env.stub.spindle.down.store(true, Ordering::SeqCst);
let head = env.commit("while it is down");
let pushed = env.push(&[], &["HEAD:refs/heads/feat"]);
assert!(pushed.status.success(), "{}", stderr(&pushed));
assert!(!stderr(&pushed).contains("follow"), "{}", stderr(&pushed));
assert_eq!(env.queued(), 1);
assert!(env.stub.taken().is_empty());
env.stub.spindle.down.store(false, Ordering::SeqCst);
succeeds(env.knot(&["post-activate"]));
env.wait_for_empty_outbox();
let taken = env.stub.taken();
assert_eq!(taken.len(), 1, "delivered once: {taken:?}");
assert_eq!(taken[0]["event"]["newSha"], head);
std::thread::sleep(Duration::from_secs(2));
assert_eq!(
env.stub.taken().len(),
1,
"the detached drain doesn't repeat it"
);
}
fn main_push(env: &Env, hosted: &std::path::Path) -> (String, String, Vec<u8>) {
let old = git(hosted, ["rev-parse", "refs/heads/main"]);
let new = env.commit("over http");
let mut pack = common::git_command(&env.work)
.args(["pack-objects", "--stdout", "--revs", "-q"])
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.spawn()
.unwrap();
std::io::Write::write_all(
&mut pack.stdin.take().unwrap(),
format!("{new}\n^{old}\n").as_bytes(),
)
.unwrap();
let command = format!("{old} {new} refs/heads/main\0report-status atomic");
let mut body = format!("{:04x}{command}", command.len() + 4).into_bytes();
body.extend(b"0000");
body.extend(pack.wait_with_output().unwrap().stdout);
(old, new, body)
}
fn service_jwt(signer: &K256Signer, knot_did: &str, account: &str) -> String {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let claims = json!({
"iss": account, "aud": knot_did, "iat": now, "exp": now + 60,
"jti": "jti-http-push", "lxm": "org.tangled.git.receivePack",
});
let signing_input = format!(
"{}.{}",
b64().encode(br#"{"alg":"ES256K","typ":"JWT"}"#),
b64().encode(claims.to_string().as_bytes())
);
format!(
"{signing_input}.{}",
b64().encode(signer.sign(signing_input.as_bytes()).as_bytes())
)
}
fn http_push(env: &Env) -> (Instant, String, String) {
use std::io::Write as _;
env.collaborate(COLLABORATOR);
let listed = succeeds(env.knot(&["repo", "list"]));
let hosted = PathBuf::from(listed.split_whitespace().last().unwrap());
let (old, new, body) = main_push(env, &hosted);
let signer = K256Signer::generate(&SeededEntropy::new(41));
let token = service_jwt(&signer, &KNOT_DID, COLLABORATOR);
let mut cgi = common::knot_static_cmd(&env.config, MASTER_KEY_ENV, &MASTER_KEY, ["cgi"]);
cgi.env("REQUEST_METHOD", "POST")
.env("REQUEST_URI", format!("/{}/git-receive-pack", env.repo_did))
.env("CONTENT_LENGTH", body.len().to_string())
.env("HTTP_AUTHORIZATION", format!("Bearer {token}"))
.envs(env.stub.env())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let mut child = cgi.spawn().unwrap();
let mut stdin = child.stdin.take().unwrap();
let feeder = std::thread::spawn(move || {
let _ = stdin.write_all(&body);
});
let answer = child.wait_with_output().unwrap();
let answered = Instant::now();
feeder.join().unwrap();
let said = String::from_utf8_lossy(&answer.stdout).into_owned();
assert!(
said.contains("ok refs/heads/main"),
"{said} {}",
stderr(&answer)
);
(answered, old, new)
}
#[test]
fn an_http_push_leaves_the_spindle_to_a_detached_process_that_still_tells_it() {
let env = setup();
env.set_spindle();
let (answered, old, new) = http_push(&env);
let taken = env.stub.wait_for_taken(1);
assert!(
env.stub.spindle.arrived.lock().unwrap()[0] > answered,
"the spindle heard of it after the answer went out"
);
assert_eq!(
taken[0],
json!({
"repo": env.repo_did,
"event": {
"$type": "org.tangled.event.push",
"repo": env.repo_did,
"ref": "refs/heads/main",
"oldSha": old,
"newSha": new,
"actor": COLLABORATOR,
},
})
);
}
#[test]
fn an_http_push_answers_before_its_crawl_waits_for_it_to_be_served() {
let env = setup();
env.set_spindle();
let crawled = env.stub.spindle.crawls.load(Ordering::SeqCst);
env.stub.spindle.withhold(3000);
http_push(&env);
let crawls = || env.stub.spindle.crawls.load(Ordering::SeqCst);
assert_eq!(crawls(), crawled, "the answer waited for the crawl");
let deadline = Instant::now() + Duration::from_secs(20);
while crawls() == crawled && Instant::now() < deadline {
std::thread::sleep(Duration::from_millis(50));
}
assert_eq!(crawls(), crawled + 1);
}
#[test]
fn repo_settings_set_and_clear_the_spindle() {
let env = setup();
let car = || {
let generation = std::fs::read_to_string(env.output.join("active.json")).unwrap();
let generation: Value = serde_json::from_str(&generation).unwrap();
std::fs::read(
env.output
.join("generations")
.join(generation["generation"].as_str().unwrap())
.join("www/_atproto")
.join(common::repo_token(&env.repo_did))
.join("repo.car"),
)
.unwrap()
};
let holds = |car: &[u8], needle: &str| {
car.windows(needle.len())
.any(|window| window == needle.as_bytes())
};
env.set_spindle();
assert!(holds(&car(), &env.spindle_did()));
succeeds(env.knot(&["repo", "settings", "sample", "--description", "kept"]));
assert!(holds(&car(), &env.spindle_did()), "other settings leave it");
succeeds(env.knot(&["repo", "settings", "sample", "--no-spindle"]));
let cleared = car();
assert!(!holds(&cleared, &env.spindle_did()));
assert!(holds(&cleared, "kept"));
let refused = env.knot(&["repo", "settings", "sample", "--spindle", "not a did"]);
assert!(!refused.status.success());
}
#[test]
fn an_ssh_push_tells_the_spindle_only_once_its_commit_is_served() {
let env = setup();
env.set_spindle();
env.stub.spindle.withhold(1000);
env.commit("slow to serve");
let pushed = env.push(&[], &["HEAD:refs/heads/feat"]);
assert!(pushed.status.success(), "{}", stderr(&pushed));
assert!(
stderr(&pushed).contains("follow refs/heads/feat"),
"{}",
stderr(&pushed)
);
let served = env
.stub
.spindle
.served_at
.lock()
.unwrap()
.expect("it was served");
assert!(
env.stub.spindle.arrived.lock().unwrap()[0] > served,
"the spindle heard of it before the new rev was served"
);
assert_eq!(*env.stub.spindle.advertised.lock().unwrap(), [true]);
}
#[test]
fn a_push_that_isnt_served_in_time_is_queued_for_the_detached_post_activate() {
let env = setup();
env.set_spindle();
env.stub.spindle.withhold(60_000);
env.commit("never served");
let pushed = env.push(&[], &["HEAD:refs/heads/feat"]);
assert!(pushed.status.success(), "{}", stderr(&pushed));
assert!(!stderr(&pushed).contains("follow"), "{}", stderr(&pushed));
assert_eq!(env.stub.attempts(), 0, "no attempt before it is served");
assert_eq!(env.queued(), 1);
env.stub.wait_for_taken(1);
assert_eq!(*env.stub.spindle.advertised.lock().unwrap(), [true]);
env.wait_for_empty_outbox();
}
#[test]
fn a_push_that_failed_to_publish_queues_its_event_for_the_next_one_that_publishes() {
let env = setup();
env.set_spindle();
let config = std::fs::read_to_string(&env.config).unwrap();
std::fs::write(
&env.config,
config.replace(r#"nginx_binary = "true""#, r#"nginx_binary = "false""#),
)
.unwrap();
env.commit("unpublished");
let pushed = env.push(&[], &["HEAD:refs/heads/first"]);
assert!(
stderr(&pushed).contains("the push wasn't published"),
"{}",
stderr(&pushed)
);
assert_eq!(env.queued(), 1);
std::thread::sleep(Duration::from_secs(1));
assert_eq!(
env.stub.attempts(),
0,
"nothing is sent for a push that isn't live"
);
std::fs::write(&env.config, config).unwrap();
env.commit("published");
let pushed = env.push(&[], &["HEAD:refs/heads/second"]);
assert!(pushed.status.success(), "{}", stderr(&pushed));
let taken = env.stub.wait_for_taken(2);
assert_eq!(taken[0]["event"]["ref"], "refs/heads/first");
assert_eq!(taken[1]["event"]["ref"], "refs/heads/second");
assert_eq!(*env.stub.spindle.advertised.lock().unwrap(), [true, true]);
env.wait_for_empty_outbox();
}
#[test]
fn post_activate_without_repos_only_drains_and_crawls_nothing() {
let env = setup();
env.set_spindle();
let crawled = env.stub.spindle.crawls.load(Ordering::SeqCst);
succeeds(env.knot(&["post-activate"]));
assert_eq!(env.stub.spindle.crawls.load(Ordering::SeqCst), crawled);
succeeds(env.knot(&["post-activate", "--repo", &env.repo_did]));
assert_eq!(env.stub.spindle.crawls.load(Ordering::SeqCst), crawled + 1);
}
#[test]
fn a_push_whose_merge_rebuild_fails_is_not_told_to_the_spindle_until_one_publishes() {
let env = setup();
env.set_spindle();
env.commit("stale and unrebuildable");
let pushed = env.push_with(
&[],
&["HEAD:refs/heads/first"],
&[("KNOT_FAILPOINT", "merge_rebuild")],
);
let said = stderr(&pushed);
assert!(said.contains("the push wasn't published"), "{said}");
assert_eq!(env.queued(), 1);
assert_eq!(
env.stub.attempts(),
0,
"a push that isn't live isn't announced"
);
env.commit("published");
assert!(env.push(&[], &["HEAD:refs/heads/second"]).status.success());
let taken = env.stub.wait_for_taken(2);
assert_eq!(taken[0]["event"]["ref"], "refs/heads/first");
assert_eq!(taken[1]["event"]["ref"], "refs/heads/second");
}