use std::collections::{BTreeSet, HashMap, HashSet}; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use axum::body::Bytes; use axum::extract::State; use axum::response::{IntoResponse, Response}; use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use futures::StreamExt; use http::{HeaderMap, HeaderValue, StatusCode, header::AUTHORIZATION}; use serde_json::json; use tempfile::TempDir; use knot_atproto::Atproto; use knot_git::Layout; use knot_index::{Index, Resolved}; use knot_runtime::{ FakeHttp, HttpRequest, HttpResponse, HttpTransport, K256Signer, ManualClock, NetworkError, OsEntropy, SeededEntropy, Signer, UnixMicros, }; use knot_secrets::{MasterKey, SealedStore}; use knot_types::{ AccountDid, AdmissionPolicy, AuthorName, Email, KnotHostname, KnotId, OwnerDid, RepoDid, RepoRkey, }; use crate::XrpcState; const KNOT_HOST: &str = "knot.nel.pet"; const ADMIN_HOST: &str = "admin.nel.pet"; const MEMBER_HOST: &str = "member.nel.pet"; const STRANGER_HOST: &str = "stranger.nel.pet"; const ADD_MEMBER: &str = "sh.tangled.knot.addMember"; const REMOVE_MEMBER: &str = "sh.tangled.knot.removeMember"; const BAN: &str = "sh.tangled.knot.ban"; const UNBAN: &str = "sh.tangled.knot.unban"; const CREATE: &str = "sh.tangled.repo.create"; const RESERVE: &str = "sh.tangled.repo.reserveKey"; const DELETE: &str = "sh.tangled.repo.delete"; const RENAME: &str = "sh.tangled.repo.rename"; const ADD_COLLAB: &str = "sh.tangled.repo.addCollaborator"; const REMOVE_COLLAB: &str = "sh.tangled.repo.removeCollaborator"; const SET_DEFAULT: &str = "sh.tangled.repo.setDefaultBranch"; const DELETE_BRANCH: &str = "sh.tangled.repo.deleteBranch"; type Responder = Box Result + Send + Sync>; type SharedState = Arc, ManualClock>>; static JTI: AtomicU64 = AtomicU64::new(0); fn knot_did() -> KnotId { KnotId::new(format!("did:web:{KNOT_HOST}")).unwrap() } fn account(host: &str) -> AccountDid { AccountDid::new(format!("did:web:{host}")).unwrap() } fn signer(seed: u64) -> K256Signer { K256Signer::generate(&SeededEntropy::new(seed)) } fn did_web_doc(did: &str, sec1: &[u8], pds: &str) -> Bytes { let multikey = knot_types::crypto::multikey(0xe7, sec1); Bytes::from( serde_json::to_vec(&json!({ "id": did, "alsoKnownAs": [], "verificationMethod": [{ "id": format!("{did}#atproto"), "type": "Multikey", "controller": did, "publicKeyMultibase": multikey }], "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": pds }] })) .unwrap(), ) } fn repo_did_doc(did: &str, multikey: &str) -> Bytes { Bytes::from( serde_json::to_vec(&json!({ "id": did, "verificationMethod": [{ "id": format!("{did}#repo"), "type": "Multikey", "controller": did, "publicKeyMultibase": multikey }] })) .unwrap(), ) } fn mint(signer: &K256Signer, issuer: &AccountDid, method: &str) -> String { let jti = JTI.fetch_add(1, Ordering::Relaxed); let header = URL_SAFE_NO_PAD.encode(br#"{"alg":"ES256K","typ":"JWT"}"#); let payload = URL_SAFE_NO_PAD.encode( serde_json::to_vec(&json!({ "iss": issuer.as_str(), "aud": format!("did:web:{KNOT_HOST}"), "exp": 1_100, "iat": 999, "jti": format!("nonce-{jti}"), "lxm": method, })) .unwrap(), ); let signing_input = format!("{header}.{payload}"); let signature = signer.sign(signing_input.as_bytes()); format!( "{signing_input}.{}", URL_SAFE_NO_PAD.encode(signature.as_bytes()) ) } fn bearer(token: &str) -> HeaderMap { let mut headers = HeaderMap::new(); headers.insert( AUTHORIZATION, HeaderValue::from_str(&format!("Bearer {token}")).unwrap(), ); headers } fn body(value: serde_json::Value) -> Bytes { Bytes::from(serde_json::to_vec(&value).unwrap()) } fn into_response(result: Result) -> Response { match result { Ok(response) => response, Err(error) => error.into_response(), } } async fn json_of(response: Response) -> serde_json::Value { let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) .await .unwrap(); serde_json::from_slice(&bytes).unwrap() } async fn call( world: &World, handler: F, signer: &K256Signer, host: &str, nsid: &str, value: serde_json::Value, ) -> Response where F: FnOnce(State, HeaderMap, crate::Method, Bytes) -> Fut, Fut: std::future::Future>, { let token = mint(signer, &account(host), nsid); into_response( handler( world.state(), bearer(&token), crate::Method::from_nsid(nsid), body(value), ) .await, ) } async fn as_member( world: &World, handler: F, nsid: &str, value: serde_json::Value, ) -> Response where F: FnOnce(State, HeaderMap, crate::Method, Bytes) -> Fut, Fut: std::future::Future>, { call(world, handler, &world.member, MEMBER_HOST, nsid, value).await } async fn as_admin( world: &World, handler: F, nsid: &str, value: serde_json::Value, ) -> Response where F: FnOnce(State, HeaderMap, crate::Method, Bytes) -> Fut, Fut: std::future::Future>, { call(world, handler, &world.admin, ADMIN_HOST, nsid, value).await } async fn as_stranger( world: &World, handler: F, nsid: &str, value: serde_json::Value, ) -> Response where F: FnOnce(State, HeaderMap, crate::Method, Bytes) -> Fut, Fut: std::future::Future>, { call(world, handler, &world.stranger, STRANGER_HOST, nsid, value).await } fn member_owner() -> OwnerDid { OwnerDid::new(format!("did:web:{MEMBER_HOST}")).unwrap() } fn resolve(world: &World, rkey: &str) -> Resolved> { world .state .index .resolve_repo(&member_owner(), &RepoRkey::new(rkey).unwrap()) } fn replay(world: &World) -> Vec> { world .state .events .replay( knot_events::EventCursor::START, knot_events::ReplayBounds::new( knot_events::ReplayEvents::new(64).unwrap(), knot_events::ReplayBytes::new(16 << 20).unwrap(), ), ) .events } fn event_count(world: &World) -> usize { replay(world).len() } fn last_event(world: &World, nsid: &str) -> std::sync::Arc { replay(world) .into_iter() .rev() .find(|event| event.nsid == nsid) .unwrap_or_else(|| panic!("{nsid} event is emitted")) } fn git_events(world: &World) -> Vec> { replay(world) .into_iter() .filter(|event| { !matches!( event.nsid, "sh.tangled.knot.memberUpdate" | "sh.tangled.repo.collaboratorUpdate" ) }) .collect() } fn only_git_event(world: &World) -> std::sync::Arc { let mut events = git_events(world); assert_eq!(events.len(), 1, "expected exactly one non-acl event"); events.remove(0) } fn bootstrap(dir: &TempDir, rebuild: bool) -> (Layout, Arc, PathBuf) { let scan_path = dir.path().join("repos"); std::fs::create_dir_all(&scan_path).unwrap(); let knot = knot_did(); let layout = Layout::new(&scan_path).reserving_meta(&knot).unwrap(); layout.bootstrap_meta(&knot).unwrap(); let meta_path = layout.meta_path(&knot).unwrap(); let index = Arc::new(Index::new(meta_path.clone(), layout.clone())); if rebuild { index.rebuild().unwrap(); } (layout, index, meta_path) } fn state_from( dir: &TempDir, boot: (Layout, Arc, PathBuf), responder: Responder, admission: AdmissionPolicy, reservations: Arc, git_http: Arc, ) -> SharedState { let (layout, index, meta_path) = boot; let knot = knot_did(); let knot_url = knot_types::KnotServiceUrl::new(format!("https://{KNOT_HOST}")).unwrap(); let atproto = Arc::new(Atproto::new( FakeHttp::new(responder), ManualClock::new(UnixMicros::new(1_000_000_000)), knot.clone(), knot_atproto::PlcDirectory::new(url::Url::parse("https://plc.directory/").unwrap()) .unwrap(), )); let secrets = Arc::new( SealedStore::open( dir.path().join("keys.sealed"), &MasterKey::new([7u8; 32]).unwrap(), Box::new(OsEntropy), ) .unwrap(), ); secrets.ensure(&knot).unwrap(); Arc::new(XrpcState { layout, index, atproto, secrets, entropy: Arc::new(OsEntropy), ci_logs: None, admins: BTreeSet::from([account(ADMIN_HOST)]), admission, knot_did: knot, knot_hostname: KnotHostname::new(KNOT_HOST).unwrap(), meta_path, knot_service_url: knot_url, limiter: Arc::new(crate::PreAuthLimiter::default()), cob_locks: Arc::new(crate::CobLocks::default()), reservations, trusted_proxy_header: None, committer: crate::Committer { name: AuthorName::new("Tangled"), email: Email::new("noreply@tangled.sh"), }, byte_limits: crate::ByteLimits::default(), budgets: crate::Budgets::default(), git_http, pack_limits: knot_pack::PackLimits::default(), service_owner: account(ADMIN_HOST), events: Arc::new(knot_events::EventLog::new( ManualClock::new(UnixMicros::new(1_000_000_000)), knot_events::ReplayBounds::new( knot_events::ReplayEvents::new(1024).unwrap(), knot_events::ReplayBytes::new(16 << 20).unwrap(), ), )), subscriber_gate: Arc::new(knot_events::SubscriberGate::new( knot_events::GlobalSubscriberLimit::new(16), knot_events::PerPeerSubscriberLimit::new(8), )), maintenance: knot_maintenance::MaintenanceHandle::disabled(), appview: knot_types::AppviewEndpoint::new("https://tangled.test").unwrap(), slots: knot_resource::Slots::testing(8), lfs: None, catalog: Arc::new(knot_messages::Catalog::defaults()), }) } fn world_responder( pubkeys: HashMap>, repo_docs: Arc>>, pds_records: Arc>>, ) -> Responder { let doc_url = format!("https://{KNOT_HOST}"); Box::new(move |request: &HttpRequest| { if request.method == http::Method::POST { return Ok(HttpResponse { status: StatusCode::OK, headers: http::HeaderMap::new(), body: Bytes::new(), }); } if request.url.path().ends_with("com.atproto.repo.getRecord") { let rkey = request .url .query_pairs() .find(|(key, _)| key == "rkey") .map(|(_, value)| value.into_owned()) .unwrap_or_default(); let present = pds_records.lock().unwrap().contains(&rkey); return Ok(HttpResponse { status: if present { StatusCode::OK } else { StatusCode::BAD_REQUEST }, headers: http::HeaderMap::new(), body: if present { Bytes::new() } else { Bytes::from_static(b"{\"error\":\"RecordNotFound\"}") }, }); } let host = request.url.host_str().unwrap_or_default(); if let Some(multikey) = repo_docs.lock().unwrap().get(host).cloned() { return Ok(HttpResponse { status: StatusCode::OK, headers: http::HeaderMap::new(), body: repo_did_doc(&format!("did:web:{host}"), &multikey), }); } match pubkeys.get(host) { Some(sec1) => Ok(HttpResponse { status: StatusCode::OK, headers: http::HeaderMap::new(), body: did_web_doc(&format!("did:web:{host}"), sec1, &doc_url), }), None => Ok(HttpResponse { status: StatusCode::NOT_FOUND, headers: http::HeaderMap::new(), body: Bytes::new(), }), } }) } struct World { _dir: TempDir, layout: Layout, state: SharedState, admin: K256Signer, member: K256Signer, stranger: K256Signer, repo_docs: Arc>>, pds_records: Arc>>, } impl World { fn new() -> Self { Self::build(256, 256, AdmissionPolicy::Closed, None) } fn open() -> Self { Self::build(256, 256, AdmissionPolicy::Open, None) } fn with_pending_limit(limit: usize) -> Self { Self::build(limit, limit, AdmissionPolicy::Closed, None) } fn with_limits(global: usize, per_actor: usize) -> Self { Self::build(global, per_actor, AdmissionPolicy::Closed, None) } fn with_git_http(git_http: Arc) -> Self { Self::build(256, 256, AdmissionPolicy::Closed, Some(git_http)) } fn build( global: usize, per_actor: usize, admission: AdmissionPolicy, git_http: Option>, ) -> Self { let dir = tempfile::tempdir().unwrap(); let (layout, index, meta_path) = bootstrap(&dir, true); let admin = signer(1); let member = signer(2); let stranger = signer(3); let repo_docs: Arc>> = Arc::new(Mutex::new(HashMap::new())); let pds_records: Arc>> = Arc::new(Mutex::new(HashSet::new())); let pubkeys = HashMap::from([ ( ADMIN_HOST.to_string(), admin.public_key().as_bytes().to_vec(), ), ( MEMBER_HOST.to_string(), member.public_key().as_bytes().to_vec(), ), ( STRANGER_HOST.to_string(), stranger.public_key().as_bytes().to_vec(), ), ]); let responder = world_responder(pubkeys, Arc::clone(&repo_docs), Arc::clone(&pds_records)); let state = state_from( &dir, (layout.clone(), index, meta_path), responder, admission, Arc::new(crate::Reservations::new( crate::ReservationTtl::new(1_000_000), crate::PerActorQuota::new(per_actor), crate::GlobalQuota::new(global), )), git_http.unwrap_or_else(no_git_upstream), ); Self { _dir: dir, layout, state, admin, member, stranger, repo_docs, pds_records, } } fn state(&self) -> State { State(Arc::clone(&self.state)) } fn publish_repo_doc(&self, host: &str, multikey: &str) { self.repo_docs .lock() .unwrap() .insert(host.to_string(), multikey.to_string()); } fn publish_pds_record(&self, rkey: &str) { self.pds_records.lock().unwrap().insert(rkey.to_string()); } } fn build_state(responder: Responder, rebuild: bool) -> (TempDir, SharedState) { let dir = tempfile::tempdir().unwrap(); let (layout, index, meta_path) = bootstrap(&dir, rebuild); let state = state_from( &dir, (layout, index, meta_path), responder, AdmissionPolicy::Closed, Arc::new(crate::Reservations::new( crate::ReservationTtl::new(1_000_000), crate::PerActorQuota::new(256), crate::GlobalQuota::new(256), )), no_git_upstream(), ); (dir, state) } fn no_git_upstream() -> Arc { Arc::new(FakeHttp::new(|_request: &HttpRequest| { Err(NetworkError::Connect( "no git upstream is served in this test".to_string(), )) })) } fn doc_responder( sec1: Vec, post_status: impl Fn() -> StatusCode + Send + Sync + 'static, ) -> Responder { let pds = format!("https://{KNOT_HOST}"); Box::new(move |request: &HttpRequest| { let post = request.method == http::Method::POST; let host = request.url.host_str().unwrap_or_default(); Ok(HttpResponse { status: if post { post_status() } else { StatusCode::OK }, headers: http::HeaderMap::new(), body: if post { Bytes::new() } else { did_web_doc(&format!("did:web:{host}"), &sec1, &pds) }, }) }) } async fn add_member_helper(world: &World) { assert_eq!( as_admin( world, crate::members::add_member, ADD_MEMBER, json!({ "subject": format!("did:web:{MEMBER_HOST}") }) ) .await .status(), StatusCode::OK ); } async fn create_repo_helper(world: &World, name: &str) -> RepoDid { assert_eq!( as_member( world, crate::repos::create_repo, CREATE, json!({ "rkey": name, "name": name }) ) .await .status(), StatusCode::OK ); match resolve(world, name) { Resolved::Ready(Some(did)) => did, other => panic!("repo {name} wasn't registered: {other:?}"), } } async fn create_status( world: &World, signer: &K256Signer, host: &str, value: serde_json::Value, ) -> StatusCode { call( world, crate::repos::create_repo, signer, host, CREATE, value, ) .await .status() } async fn reserve_status(world: &World, signer: &K256Signer, host: &str, did: &str) -> StatusCode { call( world, crate::repos::reserve_key, signer, host, RESERVE, json!({ "repoDid": did }), ) .await .status() } async fn reserve_repo_key(world: &World, did_web: &str) -> String { let response = as_member( world, crate::repos::reserve_key, RESERVE, json!({ "repoDid": did_web }), ) .await; assert_eq!(response.status(), StatusCode::OK); let key = json_of(response).await["key"].as_str().unwrap().to_string(); let host = did_web.strip_prefix("did:web:").unwrap(); world.publish_repo_doc(host, &key); key } async fn rename_repo_as( world: &World, signer: &K256Signer, host: &str, repo: &RepoDid, rkey: &str, ) -> StatusCode { call( world, crate::repos::rename_repo, signer, host, RENAME, json!({ "repo": repo.as_str(), "rkey": rkey, "name": rkey }), ) .await .status() } #[tokio::test] async fn admission_gates_repo_creation() { let closed = World::new(); let make = || json!({ "rkey": "anemone", "name": "anemone" }); assert_eq!( create_status(&closed, &closed.stranger, STRANGER_HOST, make()).await, StatusCode::FORBIDDEN, "a closed knot denies a stranger" ); let open = World::open(); assert_eq!( create_status(&open, &open.stranger, STRANGER_HOST, make()).await, StatusCode::OK ); assert!( matches!( open.state.index.resolve_repo( &OwnerDid::new(format!("did:web:{STRANGER_HOST}")).unwrap(), &RepoRkey::new("anemone").unwrap() ), Resolved::Ready(Some(_)) ), "an open knot registers the stranger's repo without membership" ); } #[tokio::test] async fn blocklist_lifecycle() { let world = World::open(); let subject = format!("did:web:{STRANGER_HOST}"); let make = || json!({ "rkey": "anemone", "name": "anemone" }); assert_eq!( as_admin( &world, crate::blocklist::ban, BAN, json!({ "subject": subject }) ) .await .status(), StatusCode::OK ); assert!(matches!( world.state.index.is_blocked(&account(STRANGER_HOST)), Resolved::Ready(true) )); assert_eq!( create_status(&world, &world.stranger, STRANGER_HOST, make()).await, StatusCode::FORBIDDEN, "a banned account cannot create" ); assert_eq!( as_admin( &world, crate::blocklist::unban, UNBAN, json!({ "subject": subject }) ) .await .status(), StatusCode::OK ); assert!(matches!( world.state.index.is_blocked(&account(STRANGER_HOST)), Resolved::Ready(false) )); assert_eq!( create_status(&world, &world.stranger, STRANGER_HOST, make()).await, StatusCode::OK, "an unban restores creation" ); assert_eq!( as_admin( &world, crate::blocklist::ban, BAN, json!({ "subject": format!("did:web:{ADMIN_HOST}") }) ) .await .status(), StatusCode::FORBIDDEN, "an admin cannot be banned" ); assert_eq!( as_stranger( &world, crate::blocklist::ban, BAN, json!({ "subject": format!("did:web:{MEMBER_HOST}") }) ) .await .status(), StatusCode::FORBIDDEN, "a non-admin cannot ban" ); } #[tokio::test] async fn member_lifecycle() { let world = World::new(); let subject = json!({ "subject": format!("did:web:{MEMBER_HOST}") }); assert_eq!( as_admin( &world, crate::members::add_member, ADD_MEMBER, subject.clone() ) .await .status(), StatusCode::OK ); assert_eq!( world.state.index.is_member(&account(MEMBER_HOST)), Resolved::Ready(true), "member is effective on the very next read, with no firehose" ); let events = replay(&world); let [added] = events.as_slice() else { panic!("expected exactly one event, got {}", events.len()); }; assert_eq!(added.nsid, "sh.tangled.knot.memberUpdate"); assert_eq!(added.payload["op"], "add"); assert_eq!(added.payload["subject"], account(MEMBER_HOST).to_string()); let baseline = event_count(&world); assert_eq!( as_admin( &world, crate::members::add_member, ADD_MEMBER, subject.clone() ) .await .status(), StatusCode::OK ); assert_eq!( event_count(&world), baseline, "a redundant add is a no-op and emits no event" ); assert_eq!( as_admin( &world, crate::members::remove_member, REMOVE_MEMBER, subject.clone() ) .await .status(), StatusCode::OK ); assert_eq!( world.state.index.is_member(&account(MEMBER_HOST)), Resolved::Ready(false), "removed member is gone on the very next read" ); let removed = last_event(&world, "sh.tangled.knot.memberUpdate"); assert_eq!(removed.payload["op"], "remove"); assert_eq!(removed.payload["subject"], account(MEMBER_HOST).to_string()); let baseline = event_count(&world); assert_eq!( as_admin( &world, crate::members::remove_member, REMOVE_MEMBER, subject ) .await .status(), StatusCode::OK ); assert_eq!( event_count(&world), baseline, "removing a non-member is a no-op and emits no event" ); } #[tokio::test] async fn add_member_auth_outcomes() { struct Case { headers: HeaderMap, body: serde_json::Value, status: StatusCode, why: &'static str, } let world = World::new(); let lowercase = { let token = mint(&world.admin, &account(ADMIN_HOST), ADD_MEMBER); let mut headers = HeaderMap::new(); headers.insert( AUTHORIZATION, HeaderValue::from_str(&format!("bearer {token}")).unwrap(), ); headers }; let cases = vec![ Case { headers: bearer(&mint(&world.member, &account(MEMBER_HOST), ADD_MEMBER)), body: json!({ "subject": "did:web:olaren.dev" }), status: StatusCode::FORBIDDEN, why: "a non-admin cannot add a member", }, Case { headers: HeaderMap::new(), body: json!({ "subject": format!("did:web:{MEMBER_HOST}") }), status: StatusCode::UNAUTHORIZED, why: "a request without a token is unauthorized", }, Case { headers: bearer(&mint(&world.admin, &account(ADMIN_HOST), ADD_MEMBER)), body: json!({ "subject": "not-a-did" }), status: StatusCode::BAD_REQUEST, why: "an invalid DID is rejected at decode by the newtype Deserialize, never reaching a handler", }, Case { headers: lowercase, body: json!({ "subject": "did:web:olaren.dev" }), status: StatusCode::OK, why: "the bearer scheme is matched case-insensitively per RFC 7235", }, ]; let world_ref = &world; futures::stream::iter(cases) .for_each(|case| async move { let status = into_response( crate::members::add_member( world_ref.state(), case.headers, crate::Method::from_nsid(ADD_MEMBER), body(case.body), ) .await, ) .status(); assert_eq!(status, case.status, "{}", case.why); }) .await; } #[tokio::test] async fn a_transient_upstream_identity_failure_is_a_503_named_upstream_unavailable() { let responder: Responder = Box::new(|_request: &HttpRequest| { Ok(HttpResponse { status: StatusCode::INTERNAL_SERVER_ERROR, headers: http::HeaderMap::new(), body: Bytes::new(), }) }); let (_dir, state) = build_state(responder, true); let admin = signer(1); let token = mint(&admin, &account(ADMIN_HOST), ADD_MEMBER); let response = into_response( crate::members::add_member( State(state), bearer(&token), crate::Method::from_nsid(ADD_MEMBER), body(json!({ "subject": format!("did:web:{MEMBER_HOST}") })), ) .await, ); assert_eq!( response.status(), StatusCode::SERVICE_UNAVAILABLE, "transient issuer-doc failure is 503, distinct from a bad token's 401" ); assert_eq!( json_of(response).await["error"], "UpstreamUnavailable", "transient upstream identity failure is named distinctly from a warming projection" ); } #[tokio::test] async fn concurrent_first_member_adds_converge_to_a_single_cob_object() { use knot_cob::CobStore; use knot_cobs::MembersCob; use knot_git::Repo; let world = World::new(); let (left, right) = tokio::join!( as_admin( &world, crate::members::add_member, ADD_MEMBER, json!({ "subject": "did:web:witchcraft.systems" }) ), as_admin( &world, crate::members::add_member, ADD_MEMBER, json!({ "subject": "did:web:isabelroses.com" }) ), ); assert_eq!(left.status(), StatusCode::OK); assert_eq!(right.status(), StatusCode::OK); let meta = Repo::open(world.state.meta_path.clone()).unwrap(); assert_eq!( CobStore::new(&meta).list::().unwrap().len(), 1, "concurrent first adds serialize onto one singleton members COB, never splitting it" ); assert_eq!( world.state.index.is_member(&account("witchcraft.systems")), Resolved::Ready(true) ); assert_eq!( world.state.index.is_member(&account("isabelroses.com")), Resolved::Ready(true) ); } #[tokio::test] async fn re_adding_a_member_under_warming_appends_no_redundant_change() { use knot_cob::CobStore; use knot_cobs::{Grant, MembersChange, MembersCob}; use knot_git::Repo; let admin = signer(1); let responder = doc_responder(admin.public_key().as_bytes().to_vec(), || StatusCode::OK); let (_dir, state) = build_state(responder, false); let now = state.now(); let knot_signer = state.secrets.signer(&state.knot_did).unwrap(); let subject = account(MEMBER_HOST); let meta = Repo::open(state.meta_path.clone()).unwrap(); let created = CobStore::new(&meta) .create( &knot_cob::CobHome::from(&state.knot_did), &MembersChange::Add(Grant { subject: subject.clone(), added_by: account(ADMIN_HOST), created_at: now, }), &knot_signer, now, ) .unwrap(); let token = mint(&admin, &account(ADMIN_HOST), ADD_MEMBER); assert_eq!( into_response( crate::members::add_member( State(Arc::clone(&state)), bearer(&token), crate::Method::from_nsid(ADD_MEMBER), body(json!({ "subject": subject.as_str() })), ) .await ) .status(), StatusCode::OK ); let meta = Repo::open(state.meta_path.clone()).unwrap(); let delta = CobStore::new(&meta) .changes_since::(created.object, Some(created.tip)) .unwrap(); assert!( delta.changes.is_empty(), "re-adding an existing member appends no change, even while projection is warming" ); } #[tokio::test] async fn create_mints_a_did_plc_repo_and_refuses_a_duplicate_name() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "anemone").await; assert!( repo_did.as_str().starts_with("did:plc:"), "knot minted a did:plc identity for the repo" ); assert!( world.layout.open(&repo_did).is_ok(), "bare repo exists on disk under its minted DID" ); assert_eq!( world.state.secrets.len(), 1, "only the shared knot key is sealed" ); assert_eq!( create_status( &world, &world.member, MEMBER_HOST, json!({ "rkey": "anemone", "name": "anemone" }) ) .await, StatusCode::CONFLICT, "a second repo of the same name is refused, never silently overwritten" ); assert_eq!( resolve(&world, "anemone"), Resolved::Ready(Some(repo_did.clone())), "the original repo still owns the name" ); assert!( world.layout.open(&repo_did).is_ok(), "the original repo is untouched on disk" ); } #[tokio::test] async fn create_and_reserve_reject_bad_identities() { let world = World::new(); add_member_helper(&world).await; assert_eq!( create_status( &world, &world.member, MEMBER_HOST, json!({ "rkey": "a", "name": "a", "repoDid": "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa" }) ) .await, StatusCode::BAD_REQUEST, "a did:plc cannot be brought; the knot mints those itself" ); assert_eq!( create_status( &world, &world.member, MEMBER_HOST, json!({ "rkey": "evil", "name": "evil", "repoDid": format!("did:web:{KNOT_HOST}") }) ) .await, StatusCode::BAD_REQUEST, "the knot's own DID as a repoDid is a client error instead of a 500" ); assert_eq!( world.state.index.is_member(&account(MEMBER_HOST)), Resolved::Ready(true), "the meta-repo is intact" ); assert_eq!( create_status( &world, &world.member, MEMBER_HOST, json!({ "rkey": "uni", "name": "uni", "repoDid": "did:web:uni.olaren.dev" }) ) .await, StatusCode::BAD_REQUEST, "a did:web create without a prior reserveKey is refused" ); assert_eq!( resolve(&world, "uni"), Resolved::Ready(None), "nothing was registered for the unproven did:web" ); assert_eq!( reserve_status( &world, &world.member, MEMBER_HOST, &format!("did:web:{KNOT_HOST}") ) .await, StatusCode::BAD_REQUEST, "the knot's own DID cannot have a repo key reserved against it" ); let unpublished = "did:web:conch.olaren.dev"; assert_eq!( reserve_status(&world, &world.member, MEMBER_HOST, unpublished).await, StatusCode::OK ); let impostor = knot_types::crypto::multikey(0xe7, signer(99).public_key().as_bytes()); world.publish_repo_doc("conch.olaren.dev", &impostor); assert_eq!( create_status( &world, &world.member, MEMBER_HOST, json!({ "rkey": "conch", "name": "conch", "repoDid": unpublished }) ) .await, StatusCode::BAD_REQUEST, "a did:web whose document publishes a different key fails the control proof" ); assert!( world .layout .open(&RepoDid::new(unpublished).unwrap()) .is_err(), "the repo was never created on disk" ); let victim = "did:web:victim.olaren.dev"; let reserved = reserve_repo_key(&world, victim).await; assert_eq!( reserve_repo_key(&world, victim).await, reserved, "re-reserving returns the same key so an already-published document stays valid" ); assert_eq!( create_status( &world, &world.admin, ADMIN_HOST, json!({ "rkey": "victim", "name": "victim", "repoDid": victim }) ) .await, StatusCode::BAD_REQUEST, "only the account that reserved the did:web may create it" ); assert!( world.layout.open(&RepoDid::new(victim).unwrap()).is_err(), "the hijack attempt created no repo on disk" ); let before = world.state.secrets.len(); assert_eq!( create_status( &world, &world.member, MEMBER_HOST, json!({ "rkey": "doomed", "name": "doomed", "defaultBranch": "bad..name" }) ) .await, StatusCode::BAD_REQUEST ); assert_eq!( world.state.secrets.len(), before, "an invalid branch is rejected before any key is minted, sealed, or DID submitted" ); } #[tokio::test] async fn a_byo_did_web_repo_is_accepted_and_its_key_is_returned() { let world = World::new(); add_member_helper(&world).await; let did = "did:web:nautilus.olaren.dev"; let reserved_key = reserve_repo_key(&world, did).await; let response = as_member( &world, crate::repos::create_repo, CREATE, json!({ "rkey": "nautilus", "name": "nautilus", "repoDid": did }), ) .await; assert_eq!(response.status(), StatusCode::OK); let output = json_of(response).await; assert_eq!(output["repoDid"].as_str(), Some(did)); let repo_did = RepoDid::new(did).unwrap(); assert!( world.layout.open(&repo_did).is_ok(), "bring-your-own did:web repo is on disk" ); let knot_public = world .state .secrets .public_key(&world.state.knot_did) .unwrap(); let expected_key = knot_types::crypto::multikey(0xe7, knot_public.as_bytes()); assert_eq!( expected_key, reserved_key, "reserve returns the knot key the owner publishes in their did:web document" ); assert_eq!( output["key"].as_str(), Some(expected_key.as_str()), "create returns the knot-held key the owner published in their own did:web document" ); assert_eq!( as_member( &world, crate::collaborators::add_collaborator, ADD_COLLAB, json!({ "repo": did, "subject": "did:web:witchcraft.systems" }) ) .await .status(), StatusCode::OK ); assert_eq!( world .state .index .is_collaborator(&repo_did, &account("witchcraft.systems")), Resolved::Ready(true), "the collaborator COB signed by the knot-held repo key lands and is visible" ); let squid = "did:web:squid.olaren.dev"; let victim = RepoDid::new(squid).unwrap(); world.layout.create(&victim).unwrap(); reserve_repo_key(&world, squid).await; assert_eq!( create_status( &world, &world.member, MEMBER_HOST, json!({ "rkey": "anemone", "name": "anemone", "repoDid": squid }) ) .await, StatusCode::CONFLICT ); assert!( world.layout.open(&victim).is_ok(), "a colliding create mustn't delete the repository already on disk" ); } #[tokio::test] async fn a_rejected_plc_submission_is_a_bad_gateway() { let admin = signer(1); let key = admin.public_key().as_bytes().to_vec(); let responder = doc_responder(key, || StatusCode::BAD_REQUEST); let (_dir, state) = build_state(responder, true); let token = mint(&admin, &account(ADMIN_HOST), CREATE); let status = into_response( crate::repos::create_repo( State(Arc::clone(&state)), bearer(&token), crate::Method::from_nsid(CREATE), body(json!({ "rkey": "conch", "name": "conch" })), ) .await, ) .status(); assert_eq!( status, StatusCode::BAD_GATEWAY, "non-transient PLC rejection surfaces as 502, distinct from an internal 500" ); assert_eq!( state.secrets.len(), 1, "rejected PLC submission rolls back the minted repo key, leaving only the knot's own. Unpublished did:plc orphans nothing" ); assert_eq!( state.index.resolve_repo( &OwnerDid::new(format!("did:web:{ADMIN_HOST}")).unwrap(), &RepoRkey::new("conch").unwrap() ), Resolved::Ready(None), "repo isn't registered after a rejected PLC submission" ); } #[tokio::test] async fn a_rejected_plc_submission_never_touches_the_registry() { let admin = signer(1); let key = admin.public_key().as_bytes().to_vec(); let reject_posts = Arc::new(std::sync::atomic::AtomicBool::new(false)); let reject = Arc::clone(&reject_posts); let responder = doc_responder(key, move || { if reject.load(Ordering::Relaxed) { StatusCode::BAD_REQUEST } else { StatusCode::OK } }); let (_dir, state) = build_state(responder, true); let owner = OwnerDid::new(format!("did:web:{ADMIN_HOST}")).unwrap(); let mint_create = || mint(&admin, &account(ADMIN_HOST), CREATE); let anemone = || body(json!({ "rkey": "anemone", "name": "anemone" })); assert_eq!( into_response( crate::repos::create_repo( State(Arc::clone(&state)), bearer(&mint_create()), crate::Method::from_nsid(CREATE), anemone(), ) .await ) .status(), StatusCode::OK ); let victim = match state .index .resolve_repo(&owner, &RepoRkey::new("anemone").unwrap()) { Resolved::Ready(Some(did)) => did, other => panic!("victim repo wasn't registered: {other:?}"), }; let token = mint(&admin, &account(ADMIN_HOST), RENAME); assert_eq!( into_response( crate::repos::rename_repo( State(Arc::clone(&state)), bearer(&token), crate::Method::from_nsid(RENAME), body(json!({ "repo": victim.as_str(), "rkey": "barnacle", "name": "barnacle" })), ) .await ) .status(), StatusCode::OK ); reject_posts.store(true, Ordering::Relaxed); assert_eq!( into_response( crate::repos::create_repo( State(Arc::clone(&state)), bearer(&mint_create()), crate::Method::from_nsid(CREATE), anemone(), ) .await ) .status(), StatusCode::BAD_GATEWAY ); assert_eq!( state .index .resolve_repo(&owner, &RepoRkey::new("anemone").unwrap()), Resolved::Ready(Some(victim.clone())), "stale alias still resolves to its prior holder because the failed create never registered" ); state.index.refresh_registry().unwrap(); assert_eq!( state .index .resolve_repo(&owner, &RepoRkey::new("anemone").unwrap()), Resolved::Ready(Some(victim.clone())), "durable registry COB has no trace of the failed create" ); assert_eq!( state.index.rkey_of(&victim), Resolved::Ready(Some(RepoRkey::new("barnacle").unwrap())), "victim's canonical rkey is unmoved" ); } #[tokio::test] async fn resolve_by_name_matches_the_rkey_case_sensitively() { let admin = signer(1); let key = admin.public_key().as_bytes().to_vec(); let responder = doc_responder(key, || StatusCode::OK); let (_dir, state) = build_state(responder, true); let owner = OwnerDid::new(format!("did:web:{ADMIN_HOST}")).unwrap(); assert_eq!( into_response( crate::repos::create_repo( State(Arc::clone(&state)), bearer(&mint(&admin, &account(ADMIN_HOST), CREATE)), crate::Method::from_nsid(CREATE), body(json!({ "rkey": "anemone", "name": "anemone" })), ) .await ) .status(), StatusCode::OK ); assert!( crate::merge::resolve_by_name(&*state, &owner, "anemone").is_ok(), "the exact rkey resolves" ); assert!( crate::merge::resolve_by_name(&*state, &owner, "Anemone").is_err(), "a differently-cased name must not resolve to a distinct rkey, atproto record keys are case-sensitive" ); } #[tokio::test] async fn reserve_key_refuses_once_the_pending_limit_is_reached() { let world = World::with_pending_limit(2); add_member_helper(&world).await; assert_eq!( reserve_status(&world, &world.member, MEMBER_HOST, "did:web:p0.olaren.dev").await, StatusCode::OK, "first reservation is within the limit" ); assert_eq!( reserve_status(&world, &world.member, MEMBER_HOST, "did:web:p1.olaren.dev").await, StatusCode::OK, "second reservation reaches the limit" ); assert_eq!( reserve_status(&world, &world.member, MEMBER_HOST, "did:web:p2.olaren.dev").await, StatusCode::TOO_MANY_REQUESTS, "member cannot grow the sealed store without bound past the pending-reservation limit" ); } #[tokio::test] async fn one_account_cannot_exhaust_the_global_reservation_budget() { let world = World::with_limits(256, 2); add_member_helper(&world).await; let world_ref = &world; futures::stream::iter(["did:web:m0.olaren.dev", "did:web:m1.olaren.dev"]) .for_each(|did| async move { assert_eq!( reserve_status(world_ref, &world_ref.member, MEMBER_HOST, did).await, StatusCode::OK ); }) .await; assert_eq!( reserve_status(&world, &world.member, MEMBER_HOST, "did:web:m2.olaren.dev").await, StatusCode::TOO_MANY_REQUESTS, "member is held to its per-actor reservation budget" ); assert_eq!( reserve_status( &world, &world.admin, ADMIN_HOST, "did:web:admin0.olaren.dev" ) .await, StatusCode::OK, "a different account keeps its own budget while global capacity remains" ); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn concurrent_reserve_key_calls_all_succeed() { let world = World::new(); add_member_helper(&world).await; let handles: Vec<_> = (0..24) .map(|i| { let state = world.state(); let token = mint(&world.member, &account(MEMBER_HOST), RESERVE); let payload = body(json!({ "repoDid": format!("did:web:r{i}.olaren.dev") })); tokio::spawn(async move { into_response( crate::repos::reserve_key( state, bearer(&token), crate::Method::from_nsid(RESERVE), payload, ) .await, ) .status() }) }) .collect(); futures::future::join_all(handles) .await .into_iter() .for_each(|result| { assert_eq!( result.unwrap(), StatusCode::OK, "concurrent reserveKey mustn't race the in-memory reservation map" ) }); } #[tokio::test] async fn collaborator_lifecycle() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "scallop").await; let subject = || json!({ "repo": repo_did.as_str(), "subject": "did:web:witchcraft.systems" }); assert_eq!( as_member( &world, crate::collaborators::add_collaborator, ADD_COLLAB, subject() ) .await .status(), StatusCode::OK ); assert_eq!( world .state .index .is_collaborator(&repo_did, &account("witchcraft.systems")), Resolved::Ready(true) ); let added = last_event(&world, "sh.tangled.repo.collaboratorUpdate"); assert_eq!(added.payload["op"], "add"); assert_eq!( added.payload["subject"], account("witchcraft.systems").to_string() ); assert_eq!(added.payload["repo"], repo_did.to_string()); assert_eq!( as_member( &world, crate::collaborators::remove_collaborator, REMOVE_COLLAB, subject() ) .await .status(), StatusCode::OK ); assert_eq!( world .state .index .is_collaborator(&repo_did, &account("witchcraft.systems")), Resolved::Ready(false), "removed collaborator is gone on the very next read" ); let removed = last_event(&world, "sh.tangled.repo.collaboratorUpdate"); assert_eq!(removed.payload["op"], "remove"); assert_eq!( removed.payload["subject"], account("witchcraft.systems").to_string() ); assert_eq!(removed.payload["repo"], repo_did.to_string()); let baseline = event_count(&world); assert_eq!( as_member( &world, crate::collaborators::remove_collaborator, REMOVE_COLLAB, subject() ) .await .status(), StatusCode::OK ); assert_eq!( event_count(&world), baseline, "removing a non-collaborator is a no-op and emits no event" ); } #[tokio::test] async fn repo_management_is_owner_or_collaborator_gated() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "squid").await; let at = format!("at://did:web:{MEMBER_HOST}/sh.tangled.repo/squid"); let did = format!("did:web:{MEMBER_HOST}"); assert_eq!( as_admin( &world, crate::collaborators::add_collaborator, ADD_COLLAB, json!({ "repo": repo_did.as_str(), "subject": "did:web:isabelroses.com" }) ) .await .status(), StatusCode::FORBIDDEN, "collaborator management is the repo owner's right instead of a knot admin's" ); assert_eq!( as_stranger( &world, crate::branches::set_default_branch, SET_DEFAULT, json!({ "repo": at, "defaultBranch": "trunk" }) ) .await .status(), StatusCode::FORBIDDEN, "a stranger cannot set the default branch" ); assert_eq!( as_stranger( &world, crate::branches::delete_branch, DELETE_BRANCH, json!({ "repo": at, "branch": "trunk" }) ) .await .status(), StatusCode::FORBIDDEN, "a stranger cannot delete a branch" ); assert_eq!( as_stranger( &world, crate::repos::delete_repo, DELETE, json!({ "did": did, "name": "squid", "rkey": "squid" }) ) .await .status(), StatusCode::FORBIDDEN, "neither owner nor a knot admin, so delete is refused" ); assert_eq!( rename_repo_as(&world, &world.stranger, STRANGER_HOST, &repo_did, "stolen").await, StatusCode::FORBIDDEN ); assert_eq!( world.state.index.rkey_of(&repo_did), Resolved::Ready(Some(RepoRkey::new("squid").unwrap())), "the canonical rkey is untouched by the rejected rename" ); assert_eq!( as_member( &world, crate::collaborators::add_collaborator, ADD_COLLAB, json!({ "repo": repo_did.as_str(), "subject": format!("did:web:{STRANGER_HOST}") }) ) .await .status(), StatusCode::OK ); assert_eq!( rename_repo_as( &world, &world.stranger, STRANGER_HOST, &repo_did, "periwinkle" ) .await, StatusCode::OK, "rename is gated by can_push, so a collaborator may rename" ); } #[tokio::test] async fn the_owner_sets_the_default_branch_through_an_at_uri() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "mussel").await; assert_eq!( as_member(&world, crate::branches::set_default_branch, SET_DEFAULT, json!({ "repo": format!("at://did:web:{MEMBER_HOST}/sh.tangled.repo/mussel"), "defaultBranch": "trunk" })).await.status(), StatusCode::OK ); let repo = world.layout.open(&repo_did).unwrap(); assert_eq!( repo.default_branch().map(|name| name.as_str().to_string()), Some("refs/heads/trunk".to_string()), "HEAD now points at the requested default branch" ); let event = only_git_event(&world); assert_eq!(event.nsid, "sh.tangled.git.refUpdate"); assert_eq!(event.payload["repo"], repo_did.to_string()); assert_eq!(event.payload["ownerDid"], account(MEMBER_HOST).to_string()); assert_eq!( event.payload["committerDid"], account(MEMBER_HOST).to_string(), "the actor who set the default branch is the committer on the wire" ); } #[tokio::test] async fn set_default_branch_rejections() { use knot_cob::CobStore; use knot_cobs::{CollaboratorsChange, Grant}; use knot_git::RefUpdate; use knot_types::RefName; let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "mussel").await; let git = world.layout.open(&repo_did).unwrap(); let now = world.state.now(); let signer = world.state.secrets.signer(&world.state.knot_did).unwrap(); let created = CobStore::new(&git) .create( &knot_cob::CobHome::from(&repo_did), &CollaboratorsChange::Add(Grant { subject: account("olaren.dev"), added_by: account("olaren.dev"), created_at: now, }), &signer, now, ) .unwrap(); git.update_ref(&RefUpdate::Create { name: RefName::new("refs/heads/main").unwrap(), new: created.tip.oid(), }) .unwrap(); assert_eq!( as_member(&world, crate::branches::set_default_branch, SET_DEFAULT, json!({ "repo": format!("at://did:web:{MEMBER_HOST}/sh.tangled.notrepo/mussel"), "defaultBranch": "trunk" })).await.status(), StatusCode::BAD_REQUEST, "an at-uri addressing a collection other than sh.tangled.repo is rejected" ); assert_eq!( as_member(&world, crate::branches::set_default_branch, SET_DEFAULT, json!({ "repo": format!("at://did:web:{MEMBER_HOST}/sh.tangled.repo/mussel"), "defaultBranch": "ghost" })).await.status(), StatusCode::NOT_FOUND, "a populated repo rejects a default pointing at a branch that doesn't exist" ); } #[tokio::test] async fn delete_branch_removes_a_branch_and_refuses_the_default() { use knot_cob::CobStore; use knot_cobs::{CollaboratorsChange, Grant}; use knot_git::RefUpdate; use knot_types::RefName; let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "periwinkle").await; let git = world.layout.open(&repo_did).unwrap(); let now = world.state.now(); let signer = world.state.secrets.signer(&world.state.knot_did).unwrap(); let created = CobStore::new(&git) .create( &knot_cob::CobHome::from(&repo_did), &CollaboratorsChange::Add(Grant { subject: account("olaren.dev"), added_by: account("olaren.dev"), created_at: now, }), &signer, now, ) .unwrap(); let oid = created.tip.oid(); ["refs/heads/main", "refs/heads/trunk"] .into_iter() .for_each(|name| { git.update_ref(&RefUpdate::Create { name: RefName::new(name).unwrap(), new: oid, }) .unwrap(); }); git.set_head(&RefName::new("refs/heads/main").unwrap()) .unwrap(); let at = format!("at://did:web:{MEMBER_HOST}/sh.tangled.repo/periwinkle"); assert_eq!( as_member( &world, crate::branches::delete_branch, DELETE_BRANCH, json!({ "repo": at, "branch": "trunk" }) ) .await .status(), StatusCode::OK, "a non-default branch is deleted" ); assert!( git.find_ref(&RefName::new("refs/heads/trunk").unwrap()) .unwrap() .is_none(), "trunk is gone" ); assert_eq!( as_member( &world, crate::branches::delete_branch, DELETE_BRANCH, json!({ "repo": at, "branch": "main" }) ) .await .status(), StatusCode::BAD_REQUEST, "the current default branch cannot be deleted" ); let event = only_git_event(&world); assert_eq!(event.nsid, "sh.tangled.git.refUpdate"); assert_eq!(event.payload["repo"], repo_did.to_string()); assert_eq!(event.payload["ref"], "refs/heads/trunk"); assert_eq!( event.payload["oldSha"], oid.to_string(), "the deletion event includes the branch's old tip" ); assert_eq!( event.payload["newSha"], git.object_format().null_oid().to_string(), "deletion reports the null oid as the new sha" ); assert_eq!( event.payload["committerDid"], account(MEMBER_HOST).to_string() ); } #[tokio::test] async fn delete_repo_lifecycle_and_guards() { let world = World::new(); add_member_helper(&world).await; let did = format!("did:web:{MEMBER_HOST}"); let plain = create_repo_helper(&world, "whelk").await; assert_eq!( as_member( &world, crate::repos::delete_repo, DELETE, json!({ "did": did, "name": "whelk", "rkey": "whelk" }) ) .await .status(), StatusCode::OK ); assert!( world.layout.open(&plain).is_err(), "bare repo is removed from disk" ); assert_eq!( resolve(&world, "whelk"), Resolved::Ready(None), "repo is deregistered" ); let guarded = create_repo_helper(&world, "conch").await; world.publish_pds_record("conch"); let delete_conch = || json!({ "did": did, "name": "conch", "rkey": "conch" }); assert_eq!( as_member(&world, crate::repos::delete_repo, DELETE, delete_conch()) .await .status(), StatusCode::CONFLICT, "the guard refuses while the sh.tangled.repo record is still on the owner's PDS" ); assert!( world.layout.open(&guarded).is_ok(), "a refused delete left the repo intact on disk" ); let force_conch = || json!({ "did": did, "name": "conch", "rkey": "conch", "force": true }); assert_eq!( as_member(&world, crate::repos::delete_repo, DELETE, force_conch()) .await .status(), StatusCode::FORBIDDEN, "force is an admin-only escape hatch instead of the owner's" ); assert_eq!( as_admin(&world, crate::repos::delete_repo, DELETE, force_conch()) .await .status(), StatusCode::OK, "a knot admin forces the delete past the lingering PDS record" ); assert!( world.layout.open(&guarded).is_err(), "forced delete removed the repo from disk" ); } #[tokio::test] async fn rename_alias_lifecycle() { let world = World::new(); add_member_helper(&world).await; let repo_a = create_repo_helper(&world, "alpha").await; assert_eq!( rename_repo_as(&world, &world.member, MEMBER_HOST, &repo_a, "alphanew").await, StatusCode::OK ); assert_eq!( resolve(&world, "alphanew"), Resolved::Ready(Some(repo_a.clone())), "the new rkey resolves on the very next read" ); assert_eq!( resolve(&world, "alpha"), Resolved::Ready(Some(repo_a.clone())), "the prior rkey keeps resolving as an alias" ); assert_eq!( world.state.index.rkey_of(&repo_a), Resolved::Ready(Some(RepoRkey::new("alphanew").unwrap())), "the new rkey is canonical" ); assert_eq!( as_member(&world, crate::branches::set_default_branch, SET_DEFAULT, json!({ "repo": format!("at://did:web:{MEMBER_HOST}/sh.tangled.repo/alpha"), "defaultBranch": "trunk" })).await.status(), StatusCode::OK, "an at-uri with the pre-rename rkey still reaches the repo" ); let repo_a2 = create_repo_helper(&world, "alpha").await; assert_ne!(repo_a2, repo_a, "a brand-new repo DID was minted"); assert_eq!( resolve(&world, "alpha"), Resolved::Ready(Some(repo_a2.clone())), "the reused rkey now resolves to the new repo" ); assert_eq!( resolve(&world, "alphanew"), Resolved::Ready(Some(repo_a.clone())), "the renamed repo keeps its canonical rkey" ); let repo_b = create_repo_helper(&world, "beta").await; assert_eq!( rename_repo_as(&world, &world.member, MEMBER_HOST, &repo_b, "alpha").await, StatusCode::CONFLICT, "a rename cannot take the canonical rkey of another live repo" ); assert_eq!( resolve(&world, "alpha"), Resolved::Ready(Some(repo_a2)), "the contested rkey still belongs to its original repo" ); let repo_g = create_repo_helper(&world, "gamma").await; assert_eq!( rename_repo_as(&world, &world.member, MEMBER_HOST, &repo_g, "gammanew").await, StatusCode::OK ); assert_eq!( as_member(&world, crate::repos::delete_repo, DELETE, json!({ "did": format!("did:web:{MEMBER_HOST}"), "name": "gammanew", "rkey": "gammanew" })).await.status(), StatusCode::OK ); assert!( world.layout.open(&repo_g).is_err(), "a renamed repo is deleted through its new rkey" ); assert_eq!( resolve(&world, "gamma"), Resolved::Ready(None), "deletion drops the retained alias along with the repo" ); let ghost = RepoDid::new("did:web:ghost.nel.pet").unwrap(); assert_eq!( rename_repo_as(&world, &world.member, MEMBER_HOST, &ghost, "kelp").await, StatusCode::NOT_FOUND, "renaming a repo this knot doesn't host is 404 instead of 403" ); } #[tokio::test] async fn a_rename_against_a_warming_registry_is_unavailable_not_forbidden() { let admin = signer(1); let responder = doc_responder(admin.public_key().as_bytes().to_vec(), || StatusCode::OK); let (_dir, state) = build_state(responder, false); let token = mint(&admin, &account(ADMIN_HOST), RENAME); assert_eq!( into_response( crate::repos::rename_repo( State(Arc::clone(&state)), bearer(&token), crate::Method::from_nsid(RENAME), body(json!({ "repo": "did:web:squid.nel.pet", "rkey": "kelp", "name": "kelp" })), ) .await ) .status(), StatusCode::SERVICE_UNAVAILABLE, "a warming registry is a retryable 503, never a permanent 403" ); } #[tokio::test] async fn the_router_sheds_a_pre_auth_flood_from_one_peer() { use axum::body::Body; use axum::extract::ConnectInfo; use std::net::SocketAddr; use tower::ServiceExt; let world = World::new(); let app = crate::router(Arc::clone(&world.state)); let peer = SocketAddr::from(([203, 0, 113, 7], 5555)); let statuses: Vec = futures::stream::iter(0..22) .then(|_| { let app = app.clone(); async move { let mut request = http::Request::builder() .method("POST") .uri(crate::members::ADD_ROUTE) .body(Body::empty()) .unwrap(); request.extensions_mut().insert(ConnectInfo(peer)); app.oneshot(request).await.unwrap().status() } }) .collect() .await; assert!( statuses[..20] .iter() .all(|status| *status == StatusCode::UNAUTHORIZED), "per-peer burst is admitted and then fails auth on the missing token, got {statuses:?}" ); assert!( statuses[20..] .iter() .all(|status| *status == StatusCode::TOO_MANY_REQUESTS), "past the burst the router sheds the flood before it can reach the resolver, got {statuses:?}" ); } #[tokio::test] async fn the_router_binds_each_route_to_its_matched_method_scope() { use axum::body::Body; use axum::extract::ConnectInfo; use std::net::SocketAddr; use tower::ServiceExt; let world = World::new(); let app = crate::router(Arc::clone(&world.state)); let peer = SocketAddr::from(([203, 0, 113, 23], 5555)); let post = |token: String| { let mut request = http::Request::builder() .method("POST") .uri(crate::members::ADD_ROUTE) .header(http::header::AUTHORIZATION, format!("Bearer {token}")) .body(Body::from( serde_json::to_vec(&json!({ "subject": format!("did:web:{MEMBER_HOST}") })) .unwrap(), )) .unwrap(); request.extensions_mut().insert(ConnectInfo(peer)); request }; let matched = mint(&world.admin, &account(ADMIN_HOST), ADD_MEMBER); assert_eq!( app.clone().oneshot(post(matched)).await.unwrap().status(), StatusCode::OK, "a token whose lxm is the route's own method authenticates" ); let sibling = mint(&world.admin, &account(ADMIN_HOST), REMOVE_MEMBER); assert_eq!( app.oneshot(post(sibling)).await.unwrap().status(), StatusCode::UNAUTHORIZED, "a token minted for a sibling method is rejected at the addMember route" ); } #[tokio::test] async fn the_http_push_surface_sheds_a_bogus_credential_flood_from_one_peer() { use axum::body::Body; use axum::extract::ConnectInfo; use base64::Engine as _; use std::net::SocketAddr; use tower::ServiceExt; let world = World::new(); add_member_helper(&world).await; let repo = create_repo_helper(&world, "kelp").await; let app = crate::router(Arc::clone(&world.state)); let peer = SocketAddr::from(([203, 0, 113, 11], 5555)); let credential = format!( "Basic {}", base64::engine::general_purpose::STANDARD.encode("git:not-a-service-jwt") ); let statuses: Vec = futures::stream::iter(0..22) .then(|_| { let app = app.clone(); let uri = format!("/{}/git-receive-pack", repo.as_str()); let credential = credential.clone(); async move { let mut request = http::Request::builder() .method("POST") .uri(uri) .header(http::header::AUTHORIZATION, credential) .body(Body::empty()) .unwrap(); request.extensions_mut().insert(ConnectInfo(peer)); app.oneshot(request).await.unwrap().status() } }) .collect() .await; assert!( statuses[..20] .iter() .all(|status| *status == StatusCode::UNAUTHORIZED), "bogus credentials inside the burst fail authentication, got {statuses:?}" ); assert!( statuses[20..] .iter() .all(|status| *status == StatusCode::TOO_MANY_REQUESTS), "past the burst the push surface sheds the flood before it can reach the resolver, got {statuses:?}" ); } #[tokio::test] async fn health_is_unauthenticated_and_exempt_from_shedding() { use axum::body::Body; use axum::extract::ConnectInfo; use std::net::SocketAddr; use tower::ServiceExt; let world = World::new(); let app = crate::router(Arc::clone(&world.state)); let peer = SocketAddr::from(([203, 0, 113, 9], 5555)); let health = || { let mut request = http::Request::builder() .method("GET") .uri(crate::service::HEALTH_ROUTE) .body(Body::empty()) .unwrap(); request.extensions_mut().insert(ConnectInfo(peer)); request }; let statuses = futures::future::join_all((0..30).map(|_| { let app = app.clone(); async move { app.oneshot(health()).await.unwrap() } })) .await; assert!( statuses .iter() .all(|response| response.status() == StatusCode::OK), "health stays 200 even past the pre-auth burst, got {:?}", statuses.iter().map(|r| r.status()).collect::>() ); let wire = json_of(app.oneshot(health()).await.unwrap()).await; assert!( wire["version"] .as_str() .is_some_and(|v| v.starts_with("knot ")), "health reports a knot version, got {wire}" ); } mod merge_endpoints { use super::*; use std::path::Path; use knot_git::{EntryKind, Identity, NewCommit, RefUpdate, StagedAction, StagedChange}; use knot_types::{Oid, RefName, UnixSeconds}; const EMPTY_TREE: &str = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"; const MERGE: &str = "sh.tangled.repo.merge"; const UNIFIED_PATCH: &str = concat!( "diff --git a/reef.txt b/reef.txt\n", "index 1111111..2222222 100644\n", "--- a/reef.txt\n", "+++ b/reef.txt\n", "@@ -1 +1 @@\n", "-old line\n", "+new line\n", ); const CONFLICTING_PATCH: &str = concat!( "diff --git a/reef.txt b/reef.txt\n", "index 1111111..2222222 100644\n", "--- a/reef.txt\n", "+++ b/reef.txt\n", "@@ -1 +1 @@\n", "-something else entirely\n", "+new line\n", ); fn seed_main(world: &World, repo_did: &RepoDid, files: &[(&str, &str)]) -> Oid { let repo = world.layout.open(repo_did).unwrap(); let staged: Vec = files .iter() .map(|(path, content)| StagedChange { path: knot_types::RepoPath::new(*path).unwrap(), action: StagedAction::Put { content: content.as_bytes().to_vec(), kind: EntryKind::Blob, }, }) .collect(); let tree = repo .write_staged_tree(Oid::from_hex(EMPTY_TREE).unwrap(), &staged) .unwrap(); let nel = Identity { name: AuthorName::new("nel"), email: Email::new("nel@oyster.cafe"), time: UnixSeconds::new(1_000), offset_seconds: 0, }; let commit = repo .write_commit(&NewCommit { tree, parents: Vec::new(), author: nel.clone(), committer: nel, message: "base".to_string(), extra_headers: Vec::new(), }) .unwrap(); repo.update_ref(&RefUpdate::Create { name: RefName::new("refs/heads/main").unwrap(), new: commit, }) .unwrap(); commit } fn main_tip(world: &World, repo_did: &RepoDid) -> Oid { world .layout .open(repo_did) .unwrap() .find_ref(&RefName::new("refs/heads/main").unwrap()) .unwrap() .unwrap() } fn file_count(dir: &Path) -> usize { std::fs::read_dir(dir) .map(|entries| { entries .flatten() .map(|entry| match entry.file_type() { Ok(kind) if kind.is_dir() => file_count(&entry.path()), _ => 1, }) .sum() }) .unwrap_or(0) } fn blob_at(repo: &knot_git::Repo, commit: Oid, path: &str) -> Vec { let entry = repo .entry_at(commit, &knot_types::RepoPath::new(path).unwrap()) .unwrap() .unwrap(); repo.read_blob(entry.oid).unwrap() } #[tokio::test] async fn the_owner_merges_a_unified_patch_natively_and_cleans_up() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "kelp").await; let base = seed_main(&world, &repo_did, &[("reef.txt", "old line\n")]); assert_eq!( as_member( &world, crate::merge::merge, MERGE, json!({ "did": format!("did:web:{MEMBER_HOST}"), "name": "kelp", "branch": "main", "patch": UNIFIED_PATCH, "commitMessage": "Merge tide", "commitBody": "body text", "authorName": "bailey", "authorEmail": "bailey@nel.pet", }) ) .await .status(), StatusCode::OK ); let repo = world.layout.open(&repo_did).unwrap(); let tip = main_tip(&world, &repo_did); assert_ne!(tip, base); let commit = repo.find_commit(tip).unwrap(); assert_eq!(commit.parents, vec![base]); assert_eq!(commit.author.name.as_str(), "bailey"); assert_eq!(commit.author.email.as_str(), "bailey@nel.pet"); assert_eq!(commit.committer.name.as_str(), "Tangled"); assert_eq!(commit.committer.email.as_str(), "noreply@tangled.sh"); assert_eq!(commit.message, "Merge tide\n\nbody text\n"); assert_eq!(blob_at(&repo, tip, "reef.txt"), b"new line\n"); let event = only_git_event(&world); assert_eq!(event.nsid, "sh.tangled.git.refUpdate"); assert_eq!(event.payload["repo"], repo_did.to_string()); assert_eq!(event.payload["ref"], "refs/heads/main"); assert_eq!(event.payload["oldSha"], base.to_string()); assert_eq!(event.payload["newSha"], tip.to_string()); assert_eq!( event.payload["committerDid"], account(MEMBER_HOST).to_string() ); let staging = std::fs::read_dir(repo.path()) .unwrap() .flatten() .filter(|entry| { entry .file_name() .to_str() .is_some_and(|name| name.starts_with(knot_git::INCOMING_PREFIX)) }) .count(); assert_eq!( staging, 0, "a completed merge cleans up its staging directory" ); } #[tokio::test] async fn a_native_merge_advances_the_branch_without_a_pipeline_event() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "kelp").await; seed_main( &world, &repo_did, &[ ("reef.txt", "old line\n"), ( ".tangled/workflows/ci.yml", "engine: nixery.dev/x\nwhen:\n - event: push\n branch: ['**']\n", ), ], ); assert_eq!( as_member( &world, crate::merge::merge, MERGE, json!({ "did": format!("did:web:{MEMBER_HOST}"), "name": "kelp", "branch": "main", "patch": UNIFIED_PATCH, "commitMessage": "Merge tide", "authorName": "bailey", "authorEmail": "bailey@nel.pet", }) ) .await .status(), StatusCode::OK ); let update = last_event(&world, "sh.tangled.git.refUpdate"); assert_eq!(update.payload["ref"], "refs/heads/main"); assert!( replay(&world) .iter() .all(|event| event.nsid != "sh.tangled.pipeline"), "the knot emits no pipeline record" ); } #[tokio::test] async fn a_format_patch_merge_creates_one_commit_per_patch_with_change_id() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "limpet").await; let base = seed_main(&world, &repo_did, &[("reef.txt", "one\n")]); let mbox = concat!( "From 1111111111111111111111111111111111111111 Mon Sep 17 00:00:00 2001\n", "From: olaren \n", "Date: Tue, 5 Sep 2023 12:00:00 +0530\n", "Subject: [PATCH 1/2] first step\n", "\n", "step one body\n", "---\n", " reef.txt | 2 +-\n", "\n", "diff --git a/reef.txt b/reef.txt\n", "index 1111111..2222222 100644\n", "--- a/reef.txt\n", "+++ b/reef.txt\n", "@@ -1 +1 @@\n", "-one\n", "+two\n", "-- \n2.43.0\n\n", "From 2222222222222222222222222222222222222222 Mon Sep 17 00:00:00 2001\n", "From: olaren \n", "Date: Tue, 5 Sep 2023 13:00:00 +0530\n", "Subject: [PATCH 2/2] second step\n", "Change-Id: Ifeedfacecafe\n", "\n", "---\n", "diff --git a/reef.txt b/reef.txt\n", "index 2222222..3333333 100644\n", "--- a/reef.txt\n", "+++ b/reef.txt\n", "@@ -1 +1 @@\n", "-two\n", "+three\n", ); assert_eq!( as_member( &world, crate::merge::merge, MERGE, json!({ "did": format!("did:web:{MEMBER_HOST}"), "name": "limpet", "branch": "main", "patch": mbox, }) ) .await .status(), StatusCode::OK ); let repo = world.layout.open(&repo_did).unwrap(); let tip = main_tip(&world, &repo_did); let second = repo.find_commit(tip).unwrap(); assert_eq!(second.message, "second step\n"); assert_eq!( second.change_id(), Some(knot_git::CommitChangeId::new("Ifeedfacecafe").unwrap()) ); assert_eq!(second.author.name.as_str(), "olaren"); assert_eq!(second.author.email.as_str(), "olaren@olaren.dev"); assert_eq!(second.author.time.get(), 1_693_899_000); assert_eq!(second.author.offset_seconds, 19_800); assert_eq!(second.committer.name.as_str(), "Tangled"); assert_eq!(second.committer.email.as_str(), "noreply@tangled.sh"); let first = repo.find_commit(second.parents[0]).unwrap(); assert_eq!(first.message, "first step\n\nstep one body\n"); assert_eq!(first.author.time.get(), 1_693_895_400); assert_eq!(first.parents, vec![base]); assert_eq!(blob_at(&repo, tip, "reef.txt"), b"three\n"); } #[tokio::test] async fn merge_rejections_move_nothing() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "scallop").await; let base = seed_main(&world, &repo_did, &[("reef.txt", "old line\n")]); let did = format!("did:web:{MEMBER_HOST}"); assert_eq!( as_stranger( &world, crate::merge::merge, MERGE, json!({ "did": did, "name": "scallop", "branch": "main", "patch": UNIFIED_PATCH }) ) .await .status(), StatusCode::FORBIDDEN, "a stranger cannot merge" ); assert_eq!( as_member( &world, crate::merge::merge, MERGE, json!({ "did": did, "name": "scallop", "branch": "main", "patch": UNIFIED_PATCH }) ) .await .status(), StatusCode::BAD_REQUEST, "a merge without a commit message is rejected" ); assert_eq!( main_tip(&world, &repo_did), base, "rejected merge mustn't move the branch" ); assert_eq!( as_member(&world, crate::merge::merge, MERGE, json!({ "did": did, "name": "scallop", "branch": "driftwood", "patch": UNIFIED_PATCH, "commitMessage": "tide" })).await.status(), StatusCode::BAD_REQUEST, "merging into a branch the repo lacks is an invalid request" ); let response = as_member(&world, crate::merge::merge, MERGE, json!({ "did": did, "name": "scallop", "branch": "main", "patch": CONFLICTING_PATCH, "commitMessage": "tide" })).await; assert_eq!(response.status(), StatusCode::CONFLICT); let json = json_of(response).await; assert_eq!(json["error"], "MergeConflict"); assert!( json["message"] .as_str() .unwrap() .starts_with("Merge failed due to conflicts"), ); assert_eq!( main_tip(&world, &repo_did), base, "a conflicted merge mustn't move the branch" ); } #[tokio::test] async fn merge_check_is_open_and_reports_clean_conflicted_and_broken() { let world = World::new(); add_member_helper(&world).await; let repo_did = create_repo_helper(&world, "scallop").await; let base = seed_main(&world, &repo_did, &[("reef.txt", "old line\n")]); let repo = world.layout.open(&repo_did).unwrap(); let objects_before = file_count(&repo.objects_dir()); let input = |patch: &str| { body(json!({ "did": format!("did:web:{MEMBER_HOST}"), "name": "scallop", "branch": "main", "patch": patch, })) }; let clean = json_of( crate::merge::merge_check(world.state(), input(UNIFIED_PATCH)) .await .unwrap(), ) .await; assert_eq!(clean["is_conflicted"], false); assert!(clean.get("conflicts").is_none()); let conflicted = json_of( crate::merge::merge_check(world.state(), input(CONFLICTING_PATCH)) .await .unwrap(), ) .await; assert_eq!(conflicted["is_conflicted"], true); assert_eq!(conflicted["conflicts"][0]["filename"], "reef.txt"); assert_eq!(conflicted["conflicts"][0]["reason"], "patch doesn't apply"); assert_eq!(conflicted["message"], "patch cannot be applied cleanly"); let broken = json_of( crate::merge::merge_check(world.state(), input("hello world\n")) .await .unwrap(), ) .await; assert_eq!(broken["is_conflicted"], true); assert!(broken["error"].as_str().is_some()); assert_eq!( file_count(&repo.objects_dir()), objects_before, "merge check must write nothing into the object database" ); assert_eq!(main_tip(&world, &repo_did), base); } } mod fork_endpoints { use super::*; use knot_git::{EntryKind, Identity, NewCommit, RefUpdate, Repo, StagedAction, StagedChange}; use knot_runtime::HttpTransport; use knot_types::{Oid, RefName, UnixSeconds}; const EMPTY_TREE: &str = "4b825dc642cb6eb9a060e54bf8d69288fbee4904"; fn ident(time: i64) -> Identity { Identity { name: AuthorName::new("nel"), email: Email::new("nel@oyster.cafe"), time: UnixSeconds::new(time), offset_seconds: 0, } } fn put_commit(repo: &Repo, parent: Option, path: &str, content: &str, time: i64) -> Oid { let base_tree = match parent { Some(parent) => repo.find_commit(parent).unwrap().tree, None => Oid::from_hex(EMPTY_TREE).unwrap(), }; let staged = vec![StagedChange { path: knot_types::RepoPath::new(path).unwrap(), action: StagedAction::Put { content: content.as_bytes().to_vec(), kind: EntryKind::Blob, }, }]; let tree = repo.write_staged_tree(base_tree, &staged).unwrap(); repo.write_commit(&NewCommit { tree, parents: parent.into_iter().collect(), author: ident(time), committer: ident(time), message: format!("put {path}"), extra_headers: Vec::new(), }) .unwrap() } fn advance(repo: &Repo, branch: &RefName, path: &str, content: &str, time: i64) -> Oid { let old = repo.find_ref(branch).unwrap(); let new = put_commit(repo, old, path, content, time); let update = match old { Some(old) => RefUpdate::Update { name: branch.clone(), old, new, }, None => RefUpdate::Create { name: branch.clone(), new, }, }; repo.update_ref(&update).unwrap(); new } fn main_ref() -> RefName { RefName::new("refs/heads/main").unwrap() } fn member_did() -> OwnerDid { OwnerDid::new(format!("did:web:{MEMBER_HOST}")).unwrap() } fn source_url(rkey: &str) -> String { format!("https://{KNOT_HOST}/did:web:{MEMBER_HOST}/{rkey}") } async fn fork_repo(world: &World, source: &str, rkey: &str) -> RepoDid { assert_eq!( as_member( world, crate::repos::create_repo, CREATE, json!({ "rkey": rkey, "name": rkey, "source": source }) ) .await .status(), StatusCode::OK ); match world .state .index .resolve_repo(&member_did(), &RepoRkey::new(rkey).unwrap()) { Resolved::Ready(Some(did)) => did, other => panic!("fork {rkey} wasn't registered: {other:?}"), } } struct ForkWorld { world: World, source_did: RepoDid, fork_did: RepoDid, tip: Oid, } async fn forked_world() -> ForkWorld { let world = World::new(); add_member_helper(&world).await; let source_did = create_repo_helper(&world, "kelp").await; let source = world.layout.open(&source_did).unwrap(); advance(&source, &main_ref(), "reef.txt", "kelp forest\n", 1_000); let tip = advance(&source, &main_ref(), "tide.txt", "rock pool\n", 1_001); [ "refs/tags/v1", "refs/cobs/sh.tangled.repo.collaborator/limpet", "refs/hidden/feature/main", ] .into_iter() .for_each(|name| { source .update_ref(&RefUpdate::Create { name: RefName::new(name).unwrap(), new: tip, }) .unwrap(); }); let fork_did = fork_repo(&world, &source_url("kelp"), "uni").await; ForkWorld { world, source_did, fork_did, tip, } } async fn sync_fork(world: &World, signer: &K256Signer, host: &str, branch: &str) -> StatusCode { call( world, crate::forks::fork_sync, signer, host, "sh.tangled.repo.forkSync", json!({ "did": format!("did:web:{MEMBER_HOST}"), "name": "uni", "source": format!("at://did:web:{MEMBER_HOST}/sh.tangled.repo/kelp"), "branch": branch, }), ) .await .status() } async fn track_hidden(world: &World, fork_ref: &str, remote_ref: &str) -> StatusCode { as_member( world, crate::forks::hidden_ref, "sh.tangled.repo.hiddenRef", json!({ "repo": format!("at://did:web:{MEMBER_HOST}/sh.tangled.repo/uni"), "forkRef": fork_ref, "remoteRef": remote_ref, }), ) .await .status() } async fn fork_status( world: &World, branch: &str, hidden_ref: &str, ) -> (StatusCode, Option) { let response = as_member( world, crate::forks::fork_status, "sh.tangled.repo.forkStatus", json!({ "did": format!("did:web:{MEMBER_HOST}"), "name": "uni", "source": source_url("kelp"), "branch": branch, "hiddenRef": hidden_ref, }), ) .await; let status = response.status(); let value = json_of(response).await; (status, value["status"].as_u64()) } #[tokio::test] async fn a_member_forks_a_repo_hosted_on_this_knot() { let setup = forked_world().await; assert_ne!(setup.fork_did, setup.source_did); let fork = setup.world.layout.open(&setup.fork_did).unwrap(); assert_eq!(fork.find_ref(&main_ref()).unwrap(), Some(setup.tip)); assert_eq!( fork.find_ref(&RefName::new("refs/tags/v1").unwrap()) .unwrap(), Some(setup.tip) ); assert_eq!(fork.default_branch().unwrap().as_str(), "refs/heads/main"); assert_eq!( fork.origin_url().as_deref(), Some(source_url("kelp").as_str()) ); assert!( fork.references().unwrap().iter().all(|record| { !record.name.as_str().starts_with("refs/cobs/") && !record.name.as_str().starts_with("refs/hidden/") }), "a fork must copy only heads and tags, never cob or hidden refs" ); let entry = fork .entry_at(setup.tip, &knot_types::RepoPath::new("tide.txt").unwrap()) .unwrap() .unwrap(); assert_eq!(fork.read_blob(entry.oid).unwrap(), b"rock pool\n"); } #[tokio::test] async fn forking_a_source_this_knot_does_not_host_is_not_found() { let world = World::new(); add_member_helper(&world).await; assert_eq!( as_member(&world, crate::repos::create_repo, CREATE, json!({ "rkey": "uni", "name": "uni", "source": format!("https://{KNOT_HOST}/did:plc:whelk/ghost") })).await.status(), StatusCode::NOT_FOUND ); assert!(matches!( world .state .index .resolve_repo(&member_did(), &RepoRkey::new("uni").unwrap()), Resolved::Ready(None) )); } #[tokio::test] async fn fork_sync_lifecycle() { let setup = forked_world().await; let source = setup.world.layout.open(&setup.source_did).unwrap(); let new_tip = advance(&source, &main_ref(), "spray.txt", "salt\n", 1_002); assert_eq!( sync_fork(&setup.world, &setup.world.member, MEMBER_HOST, "main").await, StatusCode::OK ); let fork = setup.world.layout.open(&setup.fork_did).unwrap(); assert_eq!(fork.find_ref(&main_ref()).unwrap(), Some(new_tip)); let event = only_git_event(&setup.world); assert_eq!(event.nsid, "sh.tangled.git.refUpdate"); assert_eq!(event.payload["repo"], setup.fork_did.to_string()); assert_eq!(event.payload["ref"], "refs/heads/main"); assert_eq!(event.payload["oldSha"], setup.tip.to_string()); assert_eq!(event.payload["newSha"], new_tip.to_string()); assert_eq!( event.payload["committerDid"], account(MEMBER_HOST).to_string() ); assert_eq!( sync_fork(&setup.world, &setup.world.member, MEMBER_HOST, "main").await, StatusCode::OK, "an up-to-date sync is a no-op" ); assert_eq!( git_events(&setup.world).len(), 1, "an up-to-date sync emits no further event" ); assert_eq!( sync_fork(&setup.world, &setup.world.stranger, STRANGER_HOST, "main").await, StatusCode::FORBIDDEN, "a stranger cannot sync a fork" ); assert_eq!( sync_fork(&setup.world, &setup.world.member, MEMBER_HOST, "driftwood").await, StatusCode::NOT_FOUND, "syncing a branch the upstream lacks isn't found" ); } #[tokio::test] async fn hidden_ref_tracks_the_upstream_branch_and_stays_hidden() { let setup = forked_world().await; let source = setup.world.layout.open(&setup.source_did).unwrap(); let new_tip = advance(&source, &main_ref(), "spray.txt", "salt\n", 1_002); assert_eq!( track_hidden(&setup.world, "feature", "main").await, StatusCode::OK ); let fork = setup.world.layout.open(&setup.fork_did).unwrap(); let hidden = RefName::new("refs/hidden/feature/main").unwrap(); assert_eq!(fork.find_ref(&hidden).unwrap(), Some(new_tip)); assert!( fork.advertised_refs() .unwrap() .iter() .all(|record| !record.name.as_str().starts_with("refs/hidden/")), "a hidden ref must stay out of the public advertisement" ); assert_eq!( track_hidden(&setup.world, "feature", "main").await, StatusCode::OK, "tracking an already-tracked ref is idempotent" ); } #[tokio::test] async fn fork_status_reports_up_to_date_fast_forwardable_and_conflict() { let setup = forked_world().await; assert_eq!( track_hidden(&setup.world, "feature", "main").await, StatusCode::OK ); assert_eq!( fork_status(&setup.world, "main", "refs/hidden/feature/main").await, (StatusCode::OK, Some(0)) ); let source = setup.world.layout.open(&setup.source_did).unwrap(); advance(&source, &main_ref(), "spray.txt", "salt\n", 1_002); assert_eq!( track_hidden(&setup.world, "feature", "main").await, StatusCode::OK ); assert_eq!( fork_status(&setup.world, "main", "refs/hidden/feature/main").await, (StatusCode::OK, Some(1)) ); let fork = setup.world.layout.open(&setup.fork_did).unwrap(); advance(&fork, &main_ref(), "wreck.txt", "barnacle\n", 1_003); assert_eq!( fork_status(&setup.world, "main", "refs/hidden/feature/main").await, (StatusCode::OK, Some(2)) ); assert_eq!( fork_status(&setup.world, "main", "refs/hidden/ghost/main").await, (StatusCode::BAD_REQUEST, None), "an unresolvable revision is an invalid request" ); } #[tokio::test] async fn fork_status_reports_up_to_date_when_the_fork_is_ahead() { let setup = forked_world().await; assert_eq!( track_hidden(&setup.world, "feature", "main").await, StatusCode::OK ); let fork = setup.world.layout.open(&setup.fork_did).unwrap(); advance(&fork, &main_ref(), "wreck.txt", "barnacle\n", 1_003); assert_eq!( fork_status(&setup.world, "main", "refs/hidden/feature/main").await, (StatusCode::OK, Some(0)) ); } #[tokio::test] async fn a_fork_from_a_remote_knot_fetches_over_http() { let upstream_dir = tempfile::tempdir().unwrap(); let upstream_path = upstream_dir.path().join("uni.git"); let upstream = Repo::create(&upstream_path).unwrap(); upstream .set_head(&RefName::new("refs/heads/main").unwrap()) .unwrap(); advance(&upstream, &main_ref(), "reef.txt", "kelp forest\n", 1_000); let tip = advance(&upstream, &main_ref(), "tide.txt", "rock pool\n", 1_001); let served = upstream_path.clone(); let git_http: Arc = Arc::new(FakeHttp::new(move |request: &HttpRequest| { let _keep = &upstream_dir; let repo = Repo::open(&served).unwrap(); let body = if request.url.path().ends_with("/info/refs") { knot_pack::advertise_upload(&repo).unwrap() } else { knot_pack::upload_pack(&repo, request.body.as_deref().unwrap_or_default()) .unwrap() }; Ok(HttpResponse { status: StatusCode::OK, headers: http::HeaderMap::new(), body: body.into(), }) })); let world = World::with_git_http(git_http); add_member_helper(&world).await; let remote = "https://barnacle.nel.pet/did:plc:squid/uni"; let fork_did = fork_repo(&world, remote, "uni").await; let fork = world.layout.open(&fork_did).unwrap(); assert_eq!(fork.find_ref(&main_ref()).unwrap(), Some(tip)); assert_eq!(fork.origin_url().as_deref(), Some(remote)); let new_tip = advance( &Repo::open(&upstream_path).unwrap(), &main_ref(), "spray.txt", "salt\n", 1_002, ); assert_eq!( sync_fork(&world, &world.member, MEMBER_HOST, "main").await, StatusCode::OK ); let fork = world.layout.open(&fork_did).unwrap(); assert_eq!(fork.find_ref(&main_ref()).unwrap(), Some(new_tip)); } } mod legacy_admin_route { use super::*; use crate::legacy_admin::{ADD_MEMBER_ROUTE, LegacyAdminSecret}; use tower::ServiceExt; const SECRET: &str = "nekomilk2"; async fn call(router: &axum::Router, user: &str, password: &str, subject: &str) -> StatusCode { let encoded = base64::engine::general_purpose::STANDARD .encode(format!("{user}:{password}").as_bytes()); let request = http::Request::builder() .method("POST") .uri(ADD_MEMBER_ROUTE) .header(AUTHORIZATION, format!("Basic {encoded}")) .header(http::header::CONTENT_TYPE, "application/json") .body(axum::body::Body::from( json!({ "subject": subject }).to_string(), )) .unwrap(); router.clone().oneshot(request).await.unwrap().status() } #[tokio::test] async fn the_legacy_route_admits_a_member_only_with_the_configured_credentials() { let world = World::new(); let router = crate::router(Arc::clone(&world.state)).merge(crate::legacy_admin::router( Arc::clone(&world.state), LegacyAdminSecret::new(SECRET).unwrap(), )); let subject = format!("did:web:{MEMBER_HOST}"); let version = http::Request::builder() .method("GET") .uri(crate::service::VERSION_ROUTE) .body(axum::body::Body::empty()) .unwrap(); assert_eq!( router.clone().oneshot(version).await.unwrap().status(), StatusCode::OK, "merging the legacy route leaves the xrpc routes reachable" ); let refused = futures::future::join_all( [("admin", "nope"), ("root", SECRET), ("admin", "")] .map(|(user, password)| call(&router, user, password, &subject)), ) .await; assert!( refused .iter() .all(|status| *status == StatusCode::UNAUTHORIZED), "the knot refuses a wrong user or secret, got {refused:?}" ); assert_eq!( call( &router, "admin", SECRET, &"n".repeat(world.state.byte_limits.body.get() + 1) ) .await, StatusCode::PAYLOAD_TOO_LARGE ); assert_eq!( world.state.index.is_member(&account(MEMBER_HOST)), Resolved::Ready(false), "a refused call grants nothing" ); assert_eq!( call(&router, "admin", SECRET, &subject).await, StatusCode::OK ); assert_eq!( world.state.index.is_member(&account(MEMBER_HOST)), Resolved::Ready(true) ); let added = last_event(&world, "sh.tangled.knot.memberUpdate"); assert_eq!(added.payload["op"], "add"); assert_eq!(added.payload["subject"], account(MEMBER_HOST).to_string()); let Resolved::Ready(members) = world.state.index.member_entries() else { panic!("the member roster is warm in this test"); }; assert_eq!( members .iter() .find(|grant| grant.subject == account(MEMBER_HOST)) .expect("the member is in the roster") .added_by, world.state.service_owner, "the legacy grant records the service owner as the granter" ); let baseline = event_count(&world); assert_eq!( call(&router, "admin", SECRET, &subject).await, StatusCode::OK, "the legacy route is idempotent, matching the Go knot" ); assert_eq!( event_count(&world), baseline, "re-adding an existing member emits no event" ); } #[tokio::test] async fn the_legacy_route_sheds_a_pre_auth_flood_from_one_peer() { use axum::extract::ConnectInfo; use std::net::SocketAddr; let world = World::new(); let router = crate::legacy_admin::router( Arc::clone(&world.state), LegacyAdminSecret::new(SECRET).unwrap(), ); let peer = SocketAddr::from(([203, 0, 113, 9], 5555)); let statuses: Vec = futures::stream::iter(0..22) .then(|_| { let router = router.clone(); async move { let mut request = http::Request::builder() .method("POST") .uri(ADD_MEMBER_ROUTE) .body(axum::body::Body::empty()) .unwrap(); request.extensions_mut().insert(ConnectInfo(peer)); router.oneshot(request).await.unwrap().status() } }) .collect() .await; assert!( statuses[..20] .iter() .all(|status| *status == StatusCode::UNAUTHORIZED), "the knot admits the per-peer burst and then fails it on the missing credentials, got {statuses:?}" ); assert!( statuses[20..] .iter() .all(|status| *status == StatusCode::TOO_MANY_REQUESTS), "past the burst the knot sheds the guess flood before it reaches the secret comparison, got {statuses:?}" ); } }