mod common;
use std::path::{Path, PathBuf};
use std::process::Output;
use common::{assert_success, commit, git};
const OWNER: &str = "did:web:owner.example";
const KNOT_HOST: &str = "knot.parity.test";
const MASTER_KEY_ENV: &str = "KNOT_STATIC_PARITY_MASTER_KEY";
const MASTER_KEY: [u8; 32] = [7; 32];
const PENDING_REF: &str = "refs/atproto/plc/pending";
fn knot_static(
config: &Path,
args: impl IntoIterator<Item = impl AsRef<std::ffi::OsStr>>,
) -> Output {
common::knot_static_raw(config, MASTER_KEY_ENV, &MASTER_KEY, args, &[])
}
fn seed_source(source: &Path) {
std::fs::create_dir_all(source).unwrap();
git(source, ["init", "-q", "-b", "main"]);
commit(source, "f.txt", "one\n", "one");
}
struct Fixture {
_root: tempfile::TempDir,
config: PathBuf,
meta_path: PathBuf,
}
fn setup_fixture(rkey: &str) -> Fixture {
let root = tempfile::tempdir().unwrap();
let scan = root.path().join("state");
let source = root.path().join("source");
let config = root.path().join("config.toml");
std::fs::create_dir_all(&scan).unwrap();
seed_source(&source);
// nothing listens on port 9, and import stops at the plc submission with its local state kept.
std::fs::write(
&config,
format!(
r#"[knot]
hostname = "{KNOT_HOST}"
owner = "{OWNER}"
[repo]
scan_path = "{}"
default_branch = "main"
object_format = "sha1"
[secrets]
sealed_key_file = "{}"
master_key_env = "{MASTER_KEY_ENV}"
[atproto]
plc_directory = "https://127.0.0.1:9"
[output]
root = "{}"
[activate]
nginx_binary = "true"
reload_command = "true"
"#,
scan.display(),
root.path().join("keys.sealed").display(),
root.path().join("output").display(),
),
)
.unwrap();
assert_success(&knot_static(&config, ["init"]), "init");
let imported = knot_static(
&config,
[
"repo",
"import",
"--owner",
OWNER,
"--rkey",
rkey,
source.to_str().unwrap(),
],
);
assert!(
!imported.status.success(),
"import must stop at the unreachable PLC step"
);
let meta_path = scan.join("web").join("kn").join("ot.parity.test");
assert!(meta_path.exists(), "meta.git exists under the scan path");
Fixture {
_root: root,
config,
meta_path,
}
}
#[test]
fn plc_sync_retry_preserves_journal_and_ops_imply_tangled_knot_doc() {
let f = setup_fixture("plcjournal");
let pending_oid = git(
&f.meta_path,
["rev-parse", "--verify", "--quiet", PENDING_REF],
);
assert!(
!pending_oid.is_empty(),
"pending ref resolves: got {pending_oid:?}"
);
let first_sync = knot_static(&f.config, ["plc", "sync"]);
assert!(
!first_sync.status.success(),
"plc sync against an unreachable directory must fail: stdout={} stderr={}",
String::from_utf8_lossy(&first_sync.stdout),
String::from_utf8_lossy(&first_sync.stderr),
);
let pending_after = git(&f.meta_path, ["rev-parse", PENDING_REF]);
assert_eq!(
pending_after, pending_oid,
"failed sync must not consume the pending genesis"
);
let blob = git(&f.meta_path, ["cat-file", "blob", &pending_oid]);
let entries: serde_json::Value = serde_json::from_str(&blob).unwrap();
let entry = entries.as_array().expect("journal is a list");
assert_eq!(entry.len(), 1, "one pending genesis");
let operation = &entry[0]["operation"];
let service = &operation["services"]["tangled_knot"];
assert_eq!(service["type"].as_str(), Some("TangledKnot"));
let endpoint = service["endpoint"].as_str().expect("service endpoint");
assert!(
endpoint.starts_with(&format!("https://{KNOT_HOST}/repo/")),
"TangledKnot endpoint must live on the configured host: {endpoint}"
);
let did = entry[0]["did"].as_str().expect("repo did");
assert!(
did.starts_with("did:plc:"),
"repo identity is a did:plc: {did}"
);
let pds = &operation["services"]["atproto_pds"];
assert_eq!(pds["type"].as_str(), Some("AtprotoPersonalDataServer"));
assert_eq!(
pds["endpoint"].as_str(),
Some(format!("https://{KNOT_HOST}").as_str())
);
assert_eq!(
operation["services"].as_object().map(|s| s.len()),
Some(2),
"genesis advertises tangled_knot and atproto_pds only"
);
let second_sync = knot_static(&f.config, ["plc", "sync"]);
assert!(
!second_sync.status.success(),
"retry hits the unreachable directory again"
);
let pending_final = git(&f.meta_path, ["rev-parse", PENDING_REF]);
assert_eq!(
pending_final, pending_oid,
"the journal survives repeated failed syncs"
);
}
#[test]
fn repo_record_emits_lexicon_conformant_payload() {
let f = setup_fixture("recordcheck");
let listed = knot_static(&f.config, ["repo", "list"]);
assert_success(&listed, "repo list");
let repo_did = String::from_utf8_lossy(&listed.stdout)
.split_whitespace()
.next()
.expect("imported repo DID")
.to_string();
let record = knot_static(&f.config, ["repo", "record", &repo_did]);
assert_success(&record, "repo record");
let payload: serde_json::Value =
serde_json::from_slice(&record.stdout).expect("record payload is JSON");
assert_eq!(payload["collection"], "org.tangled.repo.declaration");
assert_eq!(payload["rkey"], "recordcheck");
let inner = &payload["record"];
assert_eq!(inner["$type"], "org.tangled.repo.declaration");
assert_eq!(inner["repo"].as_str(), Some(repo_did.as_str()));
let created = inner["createdAt"].as_str().expect("createdAt required");
assert!(chrono_like(created), "createdAt must be RFC3339: {created}");
assert_eq!(inner.as_object().unwrap().len(), 3);
}
fn chrono_like(value: &str) -> bool {
let bytes = value.as_bytes();
bytes.len() == 20
&& matches!(
(
bytes[4], bytes[7], bytes[10], bytes[13], bytes[16], bytes[19]
),
(b'-', b'-', b'T', b':', b':', b'Z')
)
&& [0..4, 5..7, 8..10, 11..13, 14..16, 17..19]
.into_iter()
.all(|range| bytes[range].iter().all(|b| b.is_ascii_digit()))
}
#[test]
fn cross_knot_repo_migration_and_plc_handover_round_trip() {
use std::collections::HashMap;
use std::io::{BufRead as _, Read as _, Write as _};
use std::sync::{Arc, Mutex};
fn op_cid(op: &serde_json::Value) -> String {
use sha2::Digest as _;
let bytes = serde_ipld_dagcbor::to_vec(op).unwrap();
let digest = sha2::Sha256::digest(&bytes);
let mh = cid::multihash::Multihash::wrap(0x12, &digest).unwrap();
cid::Cid::new_v1(0x71, mh).to_string()
}
let cert = rcgen::generate_simple_self_signed(vec![
"localhost".to_string(),
"127.0.0.1".to_string(),
"old.knot.test".to_string(),
"new.knot.test".to_string(),
"third.knot.test".to_string(),
])
.unwrap();
let root = tempfile::tempdir().unwrap();
let ca_path = root.path().join("ca.pem");
std::fs::write(&ca_path, 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 third_signer = knot_runtime::K256Signer::generate(&knot_runtime::SeededEntropy::new(99));
let third_multibase = {
use knot_runtime::Signer as _;
knot_types::ActorId::from_secp256k1(third_signer.public_key().as_bytes()).to_string()
};
let plc_ops: Arc<Mutex<HashMap<String, Vec<serde_json::Value>>>> =
Arc::new(Mutex::new(HashMap::new()));
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let base_url = format!("https://localhost:{port}");
let ops_clone = plc_ops.clone();
let third_mb = third_multibase.clone();
let base_clone = base_url.clone();
std::thread::spawn(move || {
for stream in listener.incoming().flatten() {
let (tls, ops, third_mb, base) = (
tls.clone(),
ops_clone.clone(),
third_mb.clone(),
base_clone.clone(),
);
std::thread::spawn(move || {
let Ok(conn) = rustls::ServerConnection::new(tls) else {
return;
};
let mut stream = rustls::StreamOwned::new(conn, stream);
let mut reader = std::io::BufReader::new(&mut stream);
let mut req_line = String::new();
if reader.read_line(&mut req_line).unwrap_or(0) == 0 {
return;
}
let mut parts = req_line.split_whitespace();
let method = parts.next().unwrap_or_default().to_string();
let path = parts.next().unwrap_or_default().to_string();
let mut content_len = 0usize;
loop {
let mut hdr = String::new();
if reader.read_line(&mut hdr).unwrap_or(0) == 0 || hdr.trim().is_empty() {
break;
}
if let Some(v) = hdr.to_ascii_lowercase().strip_prefix("content-length:") {
content_len = v.trim().parse().unwrap_or(0);
}
}
let mut body = vec![0u8; content_len];
let _ = reader.read_exact(&mut body);
let (status, resp_body) = if method == "GET" && path == "/.well-known/did.json" {
let doc = serde_json::json!({
"id": format!("did:web:localhost%3A{port}"),
"verificationMethod": [{
"id": format!("did:web:localhost%3A{port}#atproto"),
"type": "Multikey",
"controller": format!("did:web:localhost%3A{port}"),
"publicKeyMultibase": third_mb,
}],
"service": [{
"id": "#tangled_knot",
"type": "TangledKnot",
"serviceEndpoint": base,
}]
});
(200, doc.to_string())
} else if method == "POST" && path.starts_with("/did:plc:") {
let did = path.trim_start_matches('/').to_string();
let op: serde_json::Value = serde_json::from_slice(&body).unwrap();
let mut guard = ops.lock().unwrap();
let log = guard.entry(did).or_default();
let expected_prev = log.last().map(op_cid);
if op["prev"].as_str().map(str::to_string) != expected_prev {
(400, r#"{"message":"stale prev"}"#.to_string())
} else {
log.push(op);
(200, "{}".to_string())
}
} else if method == "GET" && path.ends_with("/log/last") {
let did = path
.trim_start_matches('/')
.trim_end_matches("/log/last")
.to_string();
let guard = ops.lock().unwrap();
match guard.get(&did).and_then(|v| v.last()) {
Some(op) => (200, op.to_string()),
None => (404, "{}".to_string()),
}
} else if method == "GET" && path.starts_with("/did:plc:") {
let did = path.trim_start_matches('/').to_string();
let guard = ops.lock().unwrap();
match guard.get(&did).and_then(|v| v.last()) {
Some(op) => {
let knot_ep = op["services"]["tangled_knot"]["endpoint"]
.as_str()
.unwrap_or_default();
let pds_ep = op["services"]["atproto_pds"]["endpoint"]
.as_str()
.unwrap_or_default();
let doc = serde_json::json!({
"id": did,
"service": [
{"id": "#tangled_knot", "type": "TangledKnot", "serviceEndpoint": knot_ep},
{"id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": pds_ep},
]
});
(200, doc.to_string())
}
None => (404, "{}".to_string()),
}
} else {
(404, "{}".to_string())
};
let response = format!(
"HTTP/1.1 {status} OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{resp_body}",
resp_body.len()
);
let _ = reader.into_inner().write_all(response.as_bytes());
});
}
});
let ca_env = ca_path.display().to_string();
let run_ks = |cfg: &Path, key: &[u8; 32], args: &[&str]| -> Output {
common::knot_static_raw(
cfg,
MASTER_KEY_ENV,
key,
args,
&[("KNOT_EXTRA_CA_FILE", ca_env.as_str())],
)
};
let source = root.path().join("source");
seed_source(&source);
let old_cfg = root.path().join("old.toml");
let old_keys = root.path().join("old-keys.sealed");
let old_scan = root.path().join("old-state");
std::fs::create_dir_all(&old_scan).unwrap();
std::fs::write(
&old_cfg,
format!(
r#"[knot]
hostname = "old.knot.test"
owner = "{OWNER}"
[repo]
scan_path = "{}"
[secrets]
sealed_key_file = "{}"
master_key_env = "{MASTER_KEY_ENV}"
[atproto]
plc_directory = "{base_url}"
allow_private_addresses = true
[output]
root = "{}"
[activate]
nginx_binary = "true"
reload_command = "true"
"#,
old_scan.display(),
old_keys.display(),
root.path().join("old-out").display(),
),
)
.unwrap();
assert_success(&run_ks(&old_cfg, &[11; 32], &["init"]), "init old knot");
let minted_out = run_ks(
&old_cfg,
&[11; 32],
&[
"repo",
"import",
"--owner",
OWNER,
"--rkey",
"migrated",
"--name",
"Migrated-Repo",
source.to_str().unwrap(),
],
);
assert_success(&minted_out, "import into old knot");
let repo_did = String::from_utf8_lossy(&minted_out.stdout)
.trim()
.to_string();
assert!(repo_did.starts_with("did:plc:"));
assert_success(
&run_ks(
&old_cfg,
&[11; 32],
&[
"repo",
"settings",
&repo_did,
"--description",
"migrated across knots",
],
),
"settings on old knot",
);
assert_success(
&run_ks(
&old_cfg,
&[11; 32],
&[
"issue",
"new",
&repo_did,
"--title",
"first issue",
"--body",
"kept",
],
),
"issue on old knot",
);
let old_bare_repo = knot_git::Layout::new(&old_scan)
.guarded_path(&knot_types::RepoDid::new(&repo_did).unwrap())
.unwrap();
assert!(old_bare_repo.exists());
let new_cfg = root.path().join("new.toml");
let new_keys = root.path().join("new-keys.sealed");
let new_scan = root.path().join("new-state");
std::fs::create_dir_all(&new_scan).unwrap();
std::fs::write(
&new_cfg,
format!(
r#"[knot]
hostname = "new.knot.test"
owner = "{OWNER}"
[repo]
scan_path = "{}"
[secrets]
sealed_key_file = "{}"
master_key_env = "{MASTER_KEY_ENV}"
[atproto]
plc_directory = "{base_url}"
allow_private_addresses = true
[output]
root = "{}"
[activate]
nginx_binary = "true"
reload_command = "true"
"#,
new_scan.display(),
new_keys.display(),
root.path().join("new-out").display(),
),
)
.unwrap();
assert_success(&run_ks(&new_cfg, &[22; 32], &["init"]), "init new knot");
let old_master_b64 =
base64::Engine::encode(&base64::engine::general_purpose::STANDARD, [11u8; 32]);
let adopt_args = [
"repo",
"import",
"--from",
"knot2",
"--did",
&repo_did,
"--from-master-key-env",
"OLD_KNOT_MASTER_KEY",
"--path",
old_cfg.to_str().unwrap(),
];
let adopt_out = common::knot_static_raw(
&new_cfg,
MASTER_KEY_ENV,
&[22; 32],
adopt_args,
&[
("KNOT_EXTRA_CA_FILE", ca_env.as_str()),
("OLD_KNOT_MASTER_KEY", old_master_b64.as_str()),
],
);
assert_success(&adopt_out, "adopt repo into new knot via --from knot2");
assert_eq!(String::from_utf8_lossy(&adopt_out.stdout).trim(), repo_did);
assert!(
!old_bare_repo.exists(),
"source bare repo should be removed from knot2 scan_path"
);
let old_listed = run_ks(&old_cfg, &[11; 32], &["repo", "list"]);
assert_success(&old_listed, "repo list on old knot");
assert!(
!String::from_utf8_lossy(&old_listed.stdout).contains(&repo_did),
"repo should be deregistered from old knot's meta.git"
);
// rerunning adopt after plc converged and source repo was removed must still succeed
let adopt_retry = common::knot_static_raw(
&new_cfg,
MASTER_KEY_ENV,
&[22; 32],
adopt_args,
&[
("KNOT_EXTRA_CA_FILE", ca_env.as_str()),
("OLD_KNOT_MASTER_KEY", old_master_b64.as_str()),
],
);
assert_success(&adopt_retry, "adopt retry after convergence");
{
let log = plc_ops.lock().unwrap();
let ops = log.get(&repo_did).expect("plc ops for repo");
assert_eq!(ops.len(), 2, "genesis + migration update");
assert_eq!(ops[1]["prev"].as_str(), Some(op_cid(&ops[0]).as_str()));
assert_eq!(
ops[1]["services"]["atproto_pds"]["endpoint"].as_str(),
Some("https://new.knot.test")
);
}
let listed = run_ks(&new_cfg, &[22; 32], &["repo", "list"]);
assert_success(&listed, "repo list on new knot");
let list_line = String::from_utf8_lossy(&listed.stdout);
assert!(
list_line.contains(&format!("{repo_did} {OWNER} migrated Migrated-Repo")),
"unexpected repo list: {list_line}"
);
assert_success(
&run_ks(
&new_cfg,
&[22; 32],
&[
"issue",
"comment",
"migrated",
"1",
"--body",
"commented on new knot",
],
),
"comment on migrated issue on new knot",
);
let handover_out = run_ks(
&new_cfg,
&[22; 32],
&["plc", "handover", "migrated", "--to-url", &base_url],
);
assert_success(&handover_out, "plc handover to third knot");
{
let log = plc_ops.lock().unwrap();
let ops = log.get(&repo_did).expect("plc ops for repo");
assert_eq!(ops.len(), 3, "genesis + import update + handover update");
assert_eq!(ops[2]["prev"].as_str(), Some(op_cid(&ops[1]).as_str()));
assert_eq!(
ops[2]["rotationKeys"],
serde_json::json!([format!("did:key:{third_multibase}")])
);
assert_eq!(
ops[2]["services"]["atproto_pds"]["endpoint"].as_str(),
Some(base_url.as_str())
);
}
}