From 4a943e835bee9d60c6de0ebe301c9680b122af22 Mon Sep 17 00:00:00 2001 From: Lewis Date: Mon, 3 Aug 2026 22:38:58 +0300 Subject: [PATCH] knot2/ssh: verify login name client asserts against published keys Lewis: May this revision serve well! --- knot2/crates/knot-messages/src/lib.rs | 34 +- knot2/crates/knot-ssh/src/exec.rs | 291 +++++++++---- knot2/crates/knot-ssh/src/identity.rs | 152 +++++++ knot2/crates/knot-ssh/src/lib.rs | 50 ++- knot2/crates/knot-ssh/src/roster.rs | 522 ------------------------ knot2/crates/knot-ssh/src/server.rs | 90 ++-- knot2/crates/knot-ssh/tests/ssh_push.rs | 416 +++++++++++++++++-- knot2/example.toml | 10 +- 8 files changed, 907 insertions(+), 658 deletions(-) create mode 100644 knot2/crates/knot-ssh/src/identity.rs delete mode 100644 knot2/crates/knot-ssh/src/roster.rs diff --git a/knot2/crates/knot-messages/src/lib.rs b/knot2/crates/knot-messages/src/lib.rs index f46cadc0..d794f286 100644 --- a/knot2/crates/knot-messages/src/lib.rs +++ b/knot2/crates/knot-messages/src/lib.rs @@ -78,6 +78,7 @@ keys! { UrlKey { Url = "url" } CiLogsKey { Host = "host", Port = "port", Repo = "repo", Sha = "sha" } GreetingKey { User = "user", Knot = "knot" } + AuthorizedKey { Authorized = "authorized" } CountKey { Count = "count" } RefKey { Ref = "ref" } ErrorKey { Error = "error" } @@ -148,12 +149,20 @@ message_group! { "This knot serves git over ssh, so there's no shell here. :P", "Clone repo with: git clone {knot}:" ], + greeting_unknown: Lines = [ + "Hi there! This is the {knot} knot.", + "This knot serves git over ssh, so there's no shell here. :P", + "Clone repo with: git clone {knot}:", + "Publish your ssh key to your atproto account so this knot can identify your pushes.", + "Put your handle in the url, as in yourhandle@{knot}:, so your ssh client can find your registered key on its own." + ], unsupported_command: Line = "knot: unsupported command", too_many_operations: Line = "knot: too many concurrent operations from your address, try again shortly", repo_not_found: Line = "knot: repository not found", index_warming: Line = "knot: repository index is warming, retry shortly", lfs_disabled: Line = "knot: LFS isn't enabled on this knot", - key_not_registered: Line = "knot: your ssh key isn't registered to a user authorized to push here. If you offer several keys, make sure the registered one is offered first.", + key_not_registered: Line = "knot: this ssh key doesn't match any key published by the accounts that may push here. Authorized: {authorized}. If your agent offers several keys, add -o IdentitiesOnly=yes so it offers your registered key.", + identity_unavailable: Line = "knot: couldn't read the account records needed to check your ssh key, retry shortly", push_denied: Line = "knot: you aren't authorized to push to this repository.", shutting_down: Line = "knot: server is shutting down", archive_malformed: Line = "knot: malformed upload-archive request", @@ -306,6 +315,29 @@ mod tests { assert!(lines.iter().any(|line| line.contains("oyster.cafe"))); } + #[test] + fn an_unidentified_visitor_is_greeted_and_shown_what_a_push_needs() { + let catalog = Catalog::defaults(); + let lines = catalog + .ssh + .greeting_unknown + .lines(|KnotKey::Knot| "oyster.cafe".to_string()); + assert!(lines[0].contains("oyster.cafe")); + assert!( + lines.iter().any(|line| line.contains("ssh key")), + "a visitor the knot can't identify learns what a push needs: {lines:?}" + ); + + let denial = catalog + .ssh + .key_not_registered + .line(|AuthorizedKey::Authorized| "@nel.pet".to_string()); + assert!( + denial.contains("@nel.pet"), + "the denial lists who may push instead: {denial}" + ); + } + #[test] fn an_empty_lines_template_mutes_the_message() { let template: Template = diff --git a/knot2/crates/knot-ssh/src/exec.rs b/knot2/crates/knot-ssh/src/exec.rs index d559e297..4fcfc16b 100644 --- a/knot2/crates/knot-ssh/src/exec.rs +++ b/knot2/crates/knot-ssh/src/exec.rs @@ -1,6 +1,7 @@ use std::net::IpAddr; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use futures::StreamExt; @@ -8,8 +9,9 @@ use knot_acl::{KnotAcl, can_push}; use knot_index::Resolved; use knot_lfs::TransferOp; use knot_pack::{PackError, PackLimits, RepoLookup}; +use knot_resource::SubjectKey; use knot_runtime::{Clock, HttpTransport}; -use knot_types::{AccountDid, ClonePath, ObjectFormat, OfferedKey, OwnerDid, RepoDid}; +use knot_types::{AccountDid, ClonePath, ObjectFormat, OwnerDid, RepoDid}; use russh::Channel; use russh::server::Msg; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; @@ -17,6 +19,7 @@ use tokio::runtime::Handle; use tokio::sync::mpsc; use crate::SshState; +use crate::identity::Credential; const READ_CHUNK: usize = 64 * 1024; const MAX_UPLOAD_REQUEST: usize = 16 * 1024 * 1024; @@ -25,6 +28,8 @@ const ARCHIVE_REQUEST_DEADLINE: Duration = Duration::from_secs(60); const LFS_PROGRESS_GRACE: Duration = Duration::from_secs(60); const LFS_PROGRESS_FLOOR_BYTES_PER_SEC: u64 = 1024; const LFS_STALL_TIMEOUT: Duration = Duration::from_secs(120); +const CANDIDATE_FANOUT: usize = 4; +const AUTHORIZED_NAMES_SHOWN: usize = 4; fn lfs_within_progress_budget(waited: Duration, moved_bytes: u64) -> bool { waited @@ -119,7 +124,7 @@ fn resolve_repo_ref( pub(crate) async fn run_exec( state: Arc>, - key: Option, + credential: Credential, channel: Channel, command: &[u8], protocol_v2: bool, @@ -151,6 +156,13 @@ pub(crate) async fn run_exec( RepoRef::Did(did) => ResolvedRef::Did(did), RepoRef::OwnerPath(owner, candidates) => ResolvedRef::OwnerPath(owner, candidates), RepoRef::HandlePath(owner_handle, candidates) => { + let Some(_lookup_permit) = state.lookup_slots.try_acquire() else { + tracing::warn!( + ?peer, + "ssh exec rejected, the lookup budget can't resolve another handle" + ); + return fail(channel, &state.catalog.ssh.too_many_operations.text()).await; + }; match state .atproto .resolve_handle_to_did(&owner_handle) @@ -188,39 +200,28 @@ pub(crate) async fn run_exec( match service { Service::Upload => serve_upload(state, channel, repo_did, protocol_v2).await, Service::UploadArchive => serve_upload_archive(state, channel, repo_did).await, - Service::Receive => serve_receive(state, key, channel, repo_did).await, - Service::Lfs(op) => serve_lfs(state, key, channel, repo_did, op).await, + Service::Receive => serve_receive(state, credential, channel, repo_did, peer).await, + Service::Lfs(op) => serve_lfs(state, credential, channel, repo_did, op, peer).await, } } async fn serve_lfs( state: Arc>, - key: Option, + credential: Credential, mut channel: Channel, repo_did: RepoDid, op: TransferOp, + peer: Option, ) { let Some(lfs) = state.lfs.clone() else { return fail(channel, &state.catalog.ssh.lfs_disabled.text()).await; }; - if op == TransferOp::Upload { - let pusher = resolve_pusher(&state, key.as_ref(), &repo_did).await; - let allowed = pusher.as_ref().is_some_and(|did| { - let acl = KnotAcl::new(&state.admins, state.admission, &state.index); - can_push(&acl, did, &repo_did).is_allowed() - }); - if !allowed { - tracing::warn!( - repo = repo_did.as_str(), - registered = pusher.is_some(), - "ssh lfs upload denied" - ); - let message = match pusher { - None => state.catalog.ssh.key_not_registered.text(), - Some(_) => state.catalog.ssh.push_denied.text(), - }; - return fail(channel, &message).await; - } + if op == TransferOp::Upload + && let PushAuth::Refused { reason, message } = + authorize_push(&state, &credential, &repo_did, peer).await + { + tracing::warn!(repo = repo_did.as_str(), reason, "ssh lfs upload denied"); + return fail(channel, &message).await; } let permit = match Arc::clone(&lfs.slots).acquire_owned().await { Ok(permit) => permit, @@ -624,9 +625,10 @@ where async fn serve_receive( state: Arc>, - key: Option, + credential: Credential, mut channel: Channel, repo_did: RepoDid, + peer: Option, ) { let advert = { let layout = state.layout.clone(); @@ -648,28 +650,11 @@ async fn serve_receive( return; } - let pusher = resolve_pusher(&state, key.as_ref(), &repo_did).await; - let allowed = |did: &AccountDid| { - let acl = KnotAcl::new(&state.admins, state.admission, &state.index); - can_push(&acl, did, &repo_did).is_allowed() - }; - let committer = match pusher { - Some(did) if allowed(&did) => did, - Some(_) => { - tracing::warn!( - repo = repo_did.as_str(), - registered = true, - "ssh push denied" - ); - return fail(channel, &state.catalog.ssh.push_denied.text()).await; - } - None => { - tracing::warn!( - repo = repo_did.as_str(), - registered = false, - "ssh push denied" - ); - return fail(channel, &state.catalog.ssh.key_not_registered.text()).await; + let committer = match authorize_push(&state, &credential, &repo_did, peer).await { + PushAuth::Allowed(did) => did, + PushAuth::Refused { reason, message } => { + tracing::warn!(repo = repo_did.as_str(), reason, "ssh push denied"); + return fail(channel, &message).await; } }; @@ -754,14 +739,21 @@ async fn serve_receive( pub(crate) async fn run_greeting( state: Arc>, - key: Option, + credential: Credential, channel: Channel, ) { - let who = greeting_identity(&state, key.as_ref()).await; - let greeting = state.catalog.ssh.greeting.lines(|key| match key { - knot_messages::GreetingKey::User => who.clone(), - knot_messages::GreetingKey::Knot => state.hostname.as_str().to_string(), - }); + let knot = state.hostname.as_str().to_string(); + let greeting = match greeting_visitor(&state, &credential).await { + Visitor::Named(who) => state.catalog.ssh.greeting.lines(|key| match key { + knot_messages::GreetingKey::User => who.clone(), + knot_messages::GreetingKey::Knot => knot.clone(), + }), + Visitor::Unknown => state + .catalog + .ssh + .greeting_unknown + .lines(|knot_messages::KnotKey::Knot| knot.clone()), + }; if greeting.is_empty() { return finish(channel, 0).await; } @@ -772,25 +764,128 @@ pub(crate) async fn run_greeting( finish(channel, 0).await; } -async fn greeting_identity( +async fn greeting_visitor( state: &Arc>, - key: Option<&OfferedKey>, -) -> String { - let Some(did) = key.and_then(|key| state.roster.did_for(key)) else { - return "there".to_string(); + credential: &Credential, +) -> Visitor { + let did = match credential { + Credential::Identified(did) => did.clone(), + Credential::Offered(key) => { + match state.index.owner_of_key(key, state.atproto.now().seconds()) { + Resolved::Ready(Some(did)) => did, + _ => return Visitor::Unknown, + } + } }; match knot_receive::resolve_handle(&state.atproto, &state.slots.resolve, &did).await { - Some(handle) => format!("@{}", handle.as_str()), - None => did.as_str().to_string(), + Some(handle) => Visitor::Named(format!("@{}", handle.as_str())), + None => Visitor::Named(did.as_str().to_string()), } } +enum PusherLookup { + Matched(AccountDid), + Unmatched(Vec), + Unavailable, +} + +enum PushAuth { + Allowed(AccountDid), + Refused { + reason: &'static str, + message: String, + }, +} + +enum Visitor { + Named(String), + Unknown, +} + +async fn authorize_push( + state: &Arc>, + credential: &Credential, + repo: &RepoDid, + peer: Option, +) -> PushAuth { + match resolve_pusher(state, credential, repo, peer).await { + PusherLookup::Matched(did) => { + let acl = KnotAcl::new(&state.admins, state.admission, &state.index); + match can_push(&acl, &did, repo).is_allowed() { + true => PushAuth::Allowed(did), + false => PushAuth::Refused { + reason: "unauthorized", + message: state.catalog.ssh.push_denied.text(), + }, + } + } + PusherLookup::Unavailable => PushAuth::Refused { + reason: "identity_unavailable", + message: state.catalog.ssh.identity_unavailable.text(), + }, + PusherLookup::Unmatched(candidates) => { + let authorized = describe_authorized(state, &candidates).await; + PushAuth::Refused { + reason: "unregistered_key", + message: state + .catalog + .ssh + .key_not_registered + .line(|knot_messages::AuthorizedKey::Authorized| authorized.clone()), + } + } + } +} + +async fn describe_authorized( + state: &Arc>, + candidates: &[AccountDid], +) -> String { + let names: Vec = futures::stream::iter( + candidates + .iter() + .take(AUTHORIZED_NAMES_SHOWN) + .cloned() + .collect::>(), + ) + .map(|did| { + let state = Arc::clone(state); + async move { + match knot_receive::resolve_handle(&state.atproto, &state.slots.resolve, &did).await { + Some(handle) => format!("@{}", handle.as_str()), + None => did.as_str().to_string(), + } + } + }) + .buffered(CANDIDATE_FANOUT) + .collect() + .await; + match ( + names.as_slice(), + candidates.len().saturating_sub(AUTHORIZED_NAMES_SHOWN), + ) { + ([], _) => "nobody".to_string(), + (shown, 0) => shown.join(", "), + (shown, hidden) => format!("{}, and {hidden} more", shown.join(", ")), + } +} + +fn probe_due(state: &Arc>, did: &AccountDid) -> bool { + state + .probe_pace + .reserve_now(&SubjectKey::new(did.as_str()), state.atproto.now()) +} + async fn resolve_pusher( state: &Arc>, - key: Option<&OfferedKey>, + credential: &Credential, repo: &RepoDid, -) -> Option { - let key = key?; + peer: Option, +) -> PusherLookup { + let key = match credential { + Credential::Identified(did) => return PusherLookup::Matched(did.clone()), + Credential::Offered(key) => key, + }; let owner = match state.index.owner_of(repo) { Resolved::Ready(Some(owner)) => Some(AccountDid::from(owner)), _ => None, @@ -806,18 +901,68 @@ async fn resolve_pusher( }; let candidates: Vec = owner.into_iter().chain(collaborators).collect(); let now = state.atproto.now().seconds(); - if let Resolved::Ready(Some(cached)) = state.index.owner_of_key(key, now) - && candidates.contains(&cached) - { - return Some(cached); + if let Some(publisher) = state.index.keys().publisher_among(&candidates, key, now) { + return PusherLookup::Matched(publisher); + } + let unread: Vec = candidates + .iter() + .filter(|did| !state.index.keys().is_fresh(did, now) || probe_due(state, did)) + .cloned() + .collect(); + if unread.is_empty() { + tracing::debug!( + ?peer, + repo = repo.as_str(), + candidates = candidates.len(), + "push check has every candidate's keys on file, and the candidates don't publish \ + the offered key" + ); + return PusherLookup::Unmatched(candidates); + } + let lease = state.key_ttl.lease_from(now); + let unresolved = Arc::new(AtomicBool::new(false)); + let read: Vec> = futures::stream::iter(unread) + .map(|did| { + let state = Arc::clone(state); + let key = key.clone(); + let unresolved = Arc::clone(&unresolved); + async move { + let _permit = state.slots.resolve.acquire().await; + match state.atproto.resolve_pubkeys(&did).await { + Ok(keys) => { + let matches = keys.contains(&key); + state.index.keys().record(&did, keys, lease); + matches.then_some(did) + } + Err(error) if error.is_gone() => { + tracing::debug!( + did = did.as_str(), + %error, + "push check records an empty key set for a candidate whose DID document is gone" + ); + state.index.keys().record(&did, Vec::new(), lease); + None + } + Err(error) => { + tracing::debug!( + did = did.as_str(), + %error, + "push check couldn't read a candidate's records" + ); + unresolved.store(true, Ordering::Relaxed); + None + } + } + } + }) + .buffered(CANDIDATE_FANOUT) + .collect() + .await; + match read.into_iter().flatten().next() { + Some(did) => PusherLookup::Matched(did), + None if unresolved.load(Ordering::Relaxed) => PusherLookup::Unavailable, + None => PusherLookup::Unmatched(candidates), } - let _permit = state.slots.resolve.acquire().await; - let matches = futures::stream::iter(candidates).filter_map(|did| async move { - let keys = state.atproto.resolve_pubkeys(&did).await.ok()?; - keys.iter().any(|resolved| resolved == key).then_some(did) - }); - futures::pin_mut!(matches); - matches.next().await } async fn read_chunk( diff --git a/knot2/crates/knot-ssh/src/identity.rs b/knot2/crates/knot-ssh/src/identity.rs new file mode 100644 index 00000000..f2602f4b --- /dev/null +++ b/knot2/crates/knot-ssh/src/identity.rs @@ -0,0 +1,152 @@ +use std::net::IpAddr; +use std::sync::Arc; + +use knot_atproto::ClaimedKeys; +use knot_index::{Coverage, Resolved}; +use knot_runtime::{Clock, HttpTransport}; +use knot_types::{AccountDid, OfferedKey, OwnerRef}; + +use crate::SshState; + +#[derive(Clone)] +pub(crate) enum Credential { + Identified(AccountDid), + Offered(OfferedKey), +} + +pub(crate) struct Asserted { + claim: OwnerRef, + outcome: Claimed, +} + +enum Claimed { + Publishes { + did: AccountDid, + keys: Vec, + }, + Unreadable, +} + +pub(crate) enum Verdict { + Identified(AccountDid), + Offered, + Refused, +} + +pub(crate) async fn verify( + state: &Arc>, + claim: Option, + key: &OfferedKey, + peer: Option, + asserted: &mut Option, +) -> Verdict { + match claim { + Some(claim) => match against_claim(state, claim, key, peer, asserted).await { + Some(verdict) => verdict, + None => against_key_set(state, key, peer), + }, + None => against_key_set(state, key, peer), + } +} + +fn against_key_set( + state: &Arc>, + key: &OfferedKey, + peer: Option, +) -> Verdict { + let now = state.atproto.now().seconds(); + match ( + state.index.owner_of_key(key, now), + state.index.keys().coverage(), + ) { + (Resolved::Ready(Some(_)), _) => Verdict::Offered, + (_, Coverage::Warming) => Verdict::Offered, + (_, Coverage::Ready) => match state.index.keys().any_unheld() { + true => Verdict::Offered, + false => { + if miss_worth_a_reread(state, peer) { + state.index.keys().note_miss(); + } + Verdict::Refused + } + }, + } +} + +fn miss_worth_a_reread( + state: &Arc>, + peer: Option, +) -> bool { + peer.is_none_or(|peer| state.miss_pace.reserve_now(&peer, state.atproto.now())) +} + +async fn against_claim( + state: &Arc>, + claim: OwnerRef, + key: &OfferedKey, + peer: Option, + asserted: &mut Option, +) -> Option { + let known = match asserted.take().filter(|known| known.claim == claim) { + Some(known) => known, + None => Asserted { + outcome: resolve_claim(state, &claim, peer).await, + claim, + }, + }; + let verdict = match &known.outcome { + Claimed::Unreadable => None, + Claimed::Publishes { did, keys } if keys.contains(key) => { + Some(Verdict::Identified(did.clone())) + } + Claimed::Publishes { did, keys } => { + tracing::debug!( + ?peer, + did = did.as_str(), + published = keys.len(), + "ssh auth refused a key the asserted account doesn't publish" + ); + Some(Verdict::Refused) + } + }; + *asserted = Some(known); + verdict +} + +async fn resolve_claim( + state: &Arc>, + claim: &OwnerRef, + peer: Option, +) -> Claimed { + let Ok(_peer_guard) = state.lookup_peers.admit(peer, state.atproto.now()) else { + tracing::debug!(?peer, "ssh auth couldn't check a claim, peer budget spent"); + return Claimed::Unreadable; + }; + let did = match claim { + OwnerRef::Did(did) => AccountDid::from(did.clone()), + OwnerRef::Handle(handle) => match state.atproto.resolve_handle_to_did(handle).await { + Ok(did) => did, + Err(error) => { + tracing::warn!( + ?peer, + handle = handle.as_str(), + %error, + "ssh auth couldn't resolve the handle in the login name" + ); + return Claimed::Unreadable; + } + }, + }; + match state.atproto.claimed_pubkeys(&did).await { + ClaimedKeys::Published(keys) => Claimed::Publishes { did, keys }, + ClaimedKeys::Unread(error) => { + tracing::warn!( + ?peer, + did = did.as_str(), + %error, + "ssh auth couldn't read the asserted account's published keys" + ); + Claimed::Unreadable + } + } +} diff --git a/knot2/crates/knot-ssh/src/lib.rs b/knot2/crates/knot-ssh/src/lib.rs index a23dce52..64897917 100644 --- a/knot2/crates/knot-ssh/src/lib.rs +++ b/knot2/crates/knot-ssh/src/lib.rs @@ -1,5 +1,5 @@ mod exec; -mod roster; +mod identity; mod server; use std::borrow::Cow; @@ -12,7 +12,7 @@ use std::time::Duration; use knot_atproto::Atproto; use knot_events::EventLog; use knot_git::{ArchiveLimit, Layout}; -use knot_index::Index; +use knot_index::{Index, KeyTtl}; use knot_maintenance::MaintenanceHandle; use knot_pack::{MaxWireBytes, PackLimits}; use knot_postreceive::LanguagesPushBudget; @@ -26,11 +26,22 @@ use tokio::sync::Semaphore; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; -use knot_resource::{LimitConfig, PerPeerInflight, PreAuthLimiter, Slots}; -use roster::KeyRoster; +use knot_resource::{ + Burst, GlobalInflight, LimitConfig, PeerPacer, PerPeerInflight, PreAuthLimiter, RateLimit, + RefillMicros, ResolveSlots, Slots, SubjectPacer, +}; use server::KnotSshServer; const MAX_INFLIGHT_PER_PEER: usize = 4; +const MAX_INFLIGHT_LOOKUPS: usize = 16; +const MAX_PREAUTH_LOOKUPS: usize = 4; +const LOOKUP_BURST_PER_PEER: u32 = 8; +const LOOKUP_REFILL_MICROS: u64 = 500_000; +const LOOKUP_INFLIGHT_PER_PEER: usize = 2; +const PROBE_BURST_PER_ACCOUNT: u32 = 1; +const PROBE_REFILL_MICROS: u64 = 30_000_000; +const MISS_BURST_PER_PEER: u32 = 1; +const MISS_REFILL_MICROS: u64 = 120_000_000; const INACTIVITY_TIMEOUT: Duration = Duration::from_secs(120); const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30); const AUTH_REJECTION_TIME: Duration = Duration::from_millis(250); @@ -61,8 +72,12 @@ pub struct SshState { languages_push_budget: LanguagesPushBudget, ci_logs: Option, slots: Slots, + lookup_slots: ResolveSlots, + lookup_peers: Arc, + probe_pace: SubjectPacer, + miss_pace: PeerPacer, + key_ttl: KeyTtl, peer_slots: Arc, - roster: Arc, maintenance: MaintenanceHandle, lfs: Option, catalog: Arc, @@ -124,10 +139,27 @@ impl SshState { languages_push_budget, ci_logs, slots: Slots::for_machine(), + lookup_slots: ResolveSlots::new(MAX_INFLIGHT_LOOKUPS), + lookup_peers: Arc::new(PreAuthLimiter::with_config(LimitConfig { + rate: Some(RateLimit { + burst: Burst::new(LOOKUP_BURST_PER_PEER), + refill: RefillMicros::new(LOOKUP_REFILL_MICROS), + }), + per_peer_inflight: Some(PerPeerInflight::new(LOOKUP_INFLIGHT_PER_PEER)), + global_inflight: Some(GlobalInflight::new(MAX_PREAUTH_LOOKUPS)), + })), + probe_pace: SubjectPacer::new(RateLimit { + burst: Burst::new(PROBE_BURST_PER_ACCOUNT), + refill: RefillMicros::new(PROBE_REFILL_MICROS), + }), + miss_pace: PeerPacer::new(RateLimit { + burst: Burst::new(MISS_BURST_PER_PEER), + refill: RefillMicros::new(MISS_REFILL_MICROS), + }), + key_ttl: KeyTtl::DEFAULT, peer_slots: Arc::new(PreAuthLimiter::with_config(LimitConfig::per_peer_only( PerPeerInflight::new(MAX_INFLIGHT_PER_PEER), ))), - roster: Arc::new(KeyRoster::new()), maintenance: MaintenanceHandle::disabled(), lfs: None, catalog: Arc::new(knot_messages::Catalog::defaults()), @@ -144,6 +176,11 @@ impl SshState { self } + pub fn with_key_ttl(mut self, ttl: KeyTtl) -> Self { + self.key_ttl = ttl; + self + } + pub fn with_maintenance(mut self, maintenance: MaintenanceHandle) -> Self { self.maintenance = maintenance; self @@ -208,7 +245,6 @@ pub async fn serve_drained( ) -> Result<(), SshError> { let config = server_config(host_key); let tracker = TaskTracker::new(); - state.roster.prime(&state.index, &state.atproto); let mut server = KnotSshServer { state, tracker: tracker.clone(), diff --git a/knot2/crates/knot-ssh/src/roster.rs b/knot2/crates/knot-ssh/src/roster.rs deleted file mode 100644 index 03981088..00000000 --- a/knot2/crates/knot-ssh/src/roster.rs +++ /dev/null @@ -1,522 +0,0 @@ -use std::collections::{HashMap, HashSet}; -use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; -use std::sync::{Arc, Mutex}; -use std::time::Duration; - -use futures::StreamExt; -use knot_atproto::Atproto; -use knot_index::{Index, IndexGeneration, Resolved}; -use knot_runtime::{Clock, HttpTransport, UnixMicros}; -use knot_types::{AccountDid, OfferedKey}; - -const FRESH_TTL: Duration = Duration::from_secs(60); -const DEGRADED_TTL: Duration = Duration::from_secs(5); -const BACKOFF_SHIFT_LIMIT: u32 = 4; -const MISS_REVALIDATE_BUDGET: Duration = Duration::from_secs(30); -const RESOLVE_FANOUT: usize = 16; - -fn degraded_ttl(consecutive_failures: u32) -> Duration { - let secs = DEGRADED_TTL - .as_secs() - .saturating_mul(1u64 << consecutive_failures.min(BACKOFF_SHIFT_LIMIT)) - .min(FRESH_TTL.as_secs()); - Duration::from_secs(secs) -} - -struct Freshness { - due: UnixMicros, - generation: IndexGeneration, -} - -#[derive(Debug, PartialEq, Eq)] -enum Staleness { - Fresh, - Revalidate, - Cold, -} - -pub(crate) struct KeyRoster { - by_did: Mutex>>, - recognized: Mutex>, - freshness: Mutex>, - failures: AtomicU32, - refresh: tokio::sync::Mutex<()>, - refresh_in_flight: AtomicBool, -} - -impl KeyRoster { - pub(crate) fn new() -> Self { - Self { - by_did: Mutex::new(HashMap::new()), - recognized: Mutex::new(HashSet::new()), - freshness: Mutex::new(None), - failures: AtomicU32::new(0), - refresh: tokio::sync::Mutex::new(()), - refresh_in_flight: AtomicBool::new(false), - } - } - - pub(crate) fn recognizes(&self, key: &OfferedKey) -> bool { - self.recognized - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .contains(key) - } - - pub(crate) fn did_for(&self, key: &OfferedKey) -> Option { - self.by_did - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .iter() - .find(|(_, keys)| keys.contains(key)) - .map(|(did, _)| did.clone()) - } - - fn is_fresh(&self, now: UnixMicros, generation: IndexGeneration) -> bool { - self.freshness - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .as_ref() - .is_some_and(|fresh| now.get() < fresh.due.get() && fresh.generation == generation) - } - - pub(crate) fn prime( - self: &Arc, - index: &Arc, - atproto: &Arc>, - ) { - self.spawn_refresh(index, atproto); - } - - pub(crate) fn ensure_fresh( - self: &Arc, - index: &Arc, - atproto: &Arc>, - ) { - match self.staleness(atproto.now(), index.generation()) { - Staleness::Fresh => {} - Staleness::Revalidate | Staleness::Cold => self.spawn_refresh(index, atproto), - } - } - - pub(crate) async fn recognizes_fresh( - self: &Arc, - key: &OfferedKey, - index: &Arc, - atproto: &Arc>, - ) -> bool { - if self.recognizes(key) { - self.ensure_fresh(index, atproto); - return true; - } - if self.is_fresh(atproto.now(), index.generation()) { - return false; - } - let _ = tokio::time::timeout(MISS_REVALIDATE_BUDGET, self.refresh(index, atproto)).await; - self.recognizes(key) - } - - fn staleness(&self, now: UnixMicros, generation: IndexGeneration) -> Staleness { - match self - .freshness - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .as_ref() - { - None => Staleness::Cold, - Some(fresh) if fresh.generation != generation => Staleness::Revalidate, - Some(fresh) if now.get() < fresh.due.get() => Staleness::Fresh, - Some(_) => Staleness::Revalidate, - } - } - - fn spawn_refresh( - self: &Arc, - index: &Arc, - atproto: &Arc>, - ) { - if self - .refresh_in_flight - .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .is_err() - { - return; - } - let roster = Arc::clone(self); - let index = Arc::clone(index); - let atproto = Arc::clone(atproto); - tokio::spawn(async move { - let _in_flight = InFlightGuard(&roster.refresh_in_flight); - roster.refresh(&index, &atproto).await; - }); - } - - async fn refresh(&self, index: &Index, atproto: &Atproto) { - let _single_flight = self.refresh.lock().await; - if self.is_fresh(atproto.now(), index.generation()) { - return; - } - let generation = index.generation(); - let (dids, incomplete) = relevant_dids(index); - let resolved: Vec<(AccountDid, Option>)> = futures::stream::iter(dids) - .map(|did| async move { - let keys = atproto.resolve_pubkeys(&did).await.ok(); - (did, keys) - }) - .buffer_unordered(RESOLVE_FANOUT) - .collect() - .await; - let any_failed = resolved.iter().any(|(_, keys)| keys.is_none()); - let relevant: HashSet = resolved.iter().map(|(did, _)| did.clone()).collect(); - { - let mut by_did = self - .by_did - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()); - by_did.retain(|did, _| relevant.contains(did)); - resolved.into_iter().for_each(|(did, keys)| { - if let Some(keys) = keys { - by_did.insert(did, keys.into_iter().collect()); - } - }); - let union: HashSet = by_did.values().flatten().cloned().collect(); - *self - .recognized - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = union; - } - let ttl = if any_failed { - degraded_ttl(self.failures.fetch_add(1, Ordering::Relaxed)) - } else { - self.failures.store(0, Ordering::Relaxed); - if incomplete { DEGRADED_TTL } else { FRESH_TTL } - }; - let due = UnixMicros::new(atproto.now().get().saturating_add(ttl.as_micros() as u64)); - *self - .freshness - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(Freshness { due, generation }); - } -} - -struct InFlightGuard<'a>(&'a AtomicBool); - -impl Drop for InFlightGuard<'_> { - fn drop(&mut self) { - self.0.store(false, Ordering::Release); - } -} - -fn relevant_dids(index: &Index) -> (Vec, bool) { - let (mut dids, incomplete): (Vec, bool) = index - .hosted_repos() - .iter() - .map(|repo| { - let (owner, owner_warming) = match index.owner_of(repo) { - Resolved::Ready(Some(owner)) => (Some(AccountDid::from(owner)), false), - Resolved::Ready(None) => (None, false), - Resolved::Warming => (None, true), - }; - let (collaborators, collaborators_warming) = match index.collaborators_of(repo) { - Resolved::Ready(collaborators) => (collaborators, false), - Resolved::Warming => (Vec::new(), true), - }; - ( - owner.into_iter().chain(collaborators).collect::>(), - owner_warming || collaborators_warming, - ) - }) - .fold( - (Vec::new(), false), - |(mut acc, warming), (dids, repo_warming)| { - acc.extend(dids); - (acc, warming || repo_warming) - }, - ); - dids.sort(); - dids.dedup(); - (dids, incomplete) -} - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::Arc; - use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; - - use knot_atproto::Atproto; - use knot_cob::{CobHome, CobStore}; - use knot_cobs::{Registration, RegistryChange}; - use knot_git::{Layout, Repo}; - use knot_runtime::{ - FakeHttp, HttpRequest, HttpResponse, K256Signer, NetworkError, SeededEntropy, Signer, - }; - use knot_types::{KnotId, OwnerDid, RepoDid, RepoName, RepoRkey, UnixSeconds, crypto}; - use russh::keys::{Algorithm, PrivateKey}; - use url::Url; - - struct SharedClock(Arc); - impl Clock for SharedClock { - fn now_unix_micros(&self) -> UnixMicros { - UnixMicros::new(self.0.load(Ordering::SeqCst)) - } - } - - fn line_and_offered() -> (String, OfferedKey) { - let key = PrivateKey::random(&mut crate::EntropyRng, Algorithm::Ed25519).unwrap(); - let public = key.public_key(); - ( - public.to_openssh().unwrap(), - OfferedKey::from_bytes(public.to_bytes().unwrap()), - ) - } - - type Responder = Box Result + Send + Sync>; - - struct Harness { - index: Arc, - atproto: Arc, SharedClock>>, - published: Arc>>, - list_calls: Arc, - _dir: tempfile::TempDir, - } - - fn harness(initial: Vec) -> Harness { - let dir = tempfile::tempdir().unwrap(); - let meta_path = dir.path().join("meta"); - Repo::create(&meta_path).unwrap(); - let layout = Layout::new(dir.path().join("repos")); - let repo_did = RepoDid::new("did:plc:squid").unwrap(); - layout.create(&repo_did).unwrap(); - let cob_signer = K256Signer::generate(&SeededEntropy::new(2)); - { - let meta = Repo::open(&meta_path).unwrap(); - CobStore::new(&meta) - .create( - &CobHome::from(&KnotId::new("did:web:nel.pet").unwrap()), - &RegistryChange::Register(Registration { - owner: OwnerDid::new("did:plc:nel").unwrap(), - rkey: RepoRkey::new("anemone").unwrap(), - name: RepoName::new("anemone").unwrap(), - repo: repo_did.clone(), - created_at: UnixSeconds::new(1), - }), - &cob_signer, - UnixSeconds::new(1), - ) - .unwrap(); - } - let index = Arc::new(Index::new(meta_path, layout.clone())); - index.rebuild().unwrap(); - - let published = Arc::new(std::sync::Mutex::new(initial)); - let list_calls = Arc::new(AtomicUsize::new(0)); - let multikey = crypto::multikey( - 0xe7, - K256Signer::generate(&SeededEntropy::new(7)) - .public_key() - .as_bytes(), - ); - let clock = Arc::new(AtomicU64::new(1_000_000_000)); - - let responder: Responder = { - let published = Arc::clone(&published); - let list_calls = Arc::clone(&list_calls); - Box::new(move |request: &HttpRequest| { - let host = request.url.host_str().unwrap_or_default().to_string(); - let body = if host == "pds.oyster.cafe" { - list_calls.fetch_add(1, Ordering::SeqCst); - let records: Vec<_> = published - .lock() - .unwrap() - .iter() - .map(|line| { - serde_json::json!({ - "uri": "at://did:plc:nel/sh.tangled.publicKey/1", - "value": { - "$type": "sh.tangled.publicKey", - "key": line, - "name": "laptop", - "createdAt": "2026-06-08T00:00:00Z" - } - }) - }) - .collect(); - serde_json::to_vec(&serde_json::json!({ "records": records })).unwrap() - } else if host == "plc.directory" { - serde_json::to_vec(&serde_json::json!({ - "id": "did:plc:nel", - "alsoKnownAs": ["at://nel.pet"], - "verificationMethod": [{ - "id": "did:plc:nel#atproto", - "type": "Multikey", - "controller": "did:plc:nel", - "publicKeyMultibase": multikey - }], - "service": [{ - "id": "#atproto_pds", - "type": "AtprotoPersonalDataServer", - "serviceEndpoint": "https://pds.oyster.cafe" - }] - })) - .unwrap() - } else { - return Ok(HttpResponse { - status: http::StatusCode::NOT_FOUND, - headers: http::HeaderMap::new(), - body: bytes::Bytes::new(), - }); - }; - Ok(HttpResponse { - status: http::StatusCode::OK, - headers: http::HeaderMap::new(), - body: bytes::Bytes::from(body), - }) - }) - }; - - let atproto = Arc::new(Atproto::new( - FakeHttp::new(responder), - SharedClock(clock), - KnotId::new("did:web:nel.pet").unwrap(), - knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(), - )); - - Harness { - index, - atproto, - published, - list_calls, - _dir: dir, - } - } - - async fn wait_recognized(roster: &KeyRoster, key: &OfferedKey) { - for _ in 0..1000 { - if roster.recognizes(key) { - return; - } - tokio::task::yield_now().await; - } - } - - #[tokio::test] - async fn an_acl_write_makes_a_freshly_published_key_recognized_without_waiting_for_the_ttl() { - let (line1, offered1) = line_and_offered(); - let (line2, offered2) = line_and_offered(); - - let Harness { - index, - atproto, - published, - list_calls, - _dir, - } = harness(vec![line1]); - - let roster = Arc::new(KeyRoster::new()); - roster.ensure_fresh(&index, &atproto); - wait_recognized(&roster, &offered1).await; - assert!(roster.recognizes(&offered1)); - assert_eq!(list_calls.load(Ordering::SeqCst), 1); - - published.lock().unwrap().push(line2.clone()); - - roster.ensure_fresh(&index, &atproto); - assert!( - !roster.recognizes(&offered2), - "stable index and unexpired TTL still serves cached roster, no re-resolution" - ); - assert_eq!(list_calls.load(Ordering::SeqCst), 1); - - index.refresh_members().unwrap(); - roster.ensure_fresh(&index, &atproto); - wait_recognized(&roster, &offered2).await; - assert!( - roster.recognizes(&offered2), - "ACL write bumps generation, so roster revalidates off the auth path" - ); - assert_eq!( - list_calls.load(Ordering::SeqCst), - 2, - "exactly one async re-resolution off the auth path" - ); - } - - #[tokio::test] - async fn a_miss_against_a_stale_roster_blocks_bounded_to_revalidate_before_rejecting() { - let (line1, offered1) = line_and_offered(); - let (line2, offered2) = line_and_offered(); - let Harness { - index, - atproto, - published, - list_calls, - _dir, - } = harness(vec![line1]); - let roster = Arc::new(KeyRoster::new()); - - assert!( - roster.recognizes_fresh(&offered1, &index, &atproto).await, - "the first handshake blocks on the primed resolve and recognizes the published key" - ); - assert_eq!(list_calls.load(Ordering::SeqCst), 1); - - published.lock().unwrap().push(line2.clone()); - index.refresh_members().unwrap(); - - assert!( - roster.recognizes_fresh(&offered2, &index, &atproto).await, - "a generation-bumped miss blocks to revalidate and picks up the new key on the first attempt" - ); - assert_eq!( - list_calls.load(Ordering::SeqCst), - 2, - "the miss triggers exactly one bounded re-resolution" - ); - } - - #[test] - fn staleness_classifies_cold_fresh_and_revalidate() { - let roster = KeyRoster::new(); - assert_eq!( - roster.staleness(UnixMicros::new(0), IndexGeneration::new(0)), - Staleness::Cold, - "with no roster yet the first auth is cold and must revalidate before it can answer a miss" - ); - *roster - .freshness - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(Freshness { - due: UnixMicros::new(1_000), - generation: IndexGeneration::new(0), - }); - assert_eq!( - roster.staleness(UnixMicros::new(500), IndexGeneration::new(0)), - Staleness::Fresh - ); - assert_eq!( - roster.staleness(UnixMicros::new(500), IndexGeneration::new(1)), - Staleness::Revalidate, - "an ACL write moves the generation, so the cached roster is stale" - ); - assert_eq!( - roster.staleness(UnixMicros::new(2_000), IndexGeneration::new(0)), - Staleness::Revalidate, - "an expired ttl at the same generation is stale too" - ); - } - - #[test] - fn degraded_ttl_backs_off_from_the_short_retry_to_the_fresh_ceiling() { - assert_eq!(degraded_ttl(0), Duration::from_secs(5)); - assert_eq!(degraded_ttl(1), Duration::from_secs(10)); - assert_eq!(degraded_ttl(2), Duration::from_secs(20)); - assert_eq!(degraded_ttl(3), Duration::from_secs(40)); - assert_eq!(degraded_ttl(4), Duration::from_secs(60)); - assert_eq!( - degraded_ttl(50), - Duration::from_secs(60), - "a persistently unresolvable did clamps the retry to the fresh ttl instead of storming" - ); - } -} diff --git a/knot2/crates/knot-ssh/src/server.rs b/knot2/crates/knot-ssh/src/server.rs index c4b53a95..829ca390 100644 --- a/knot2/crates/knot-ssh/src/server.rs +++ b/knot2/crates/knot-ssh/src/server.rs @@ -3,7 +3,7 @@ use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use knot_runtime::{Clock, HttpTransport}; -use knot_types::OfferedKey; +use knot_types::{OfferedKey, OwnerRef}; use russh::keys::ssh_key; use russh::server::{Auth, Handler, Msg, Server, Session}; use russh::{Channel, ChannelId}; @@ -11,6 +11,7 @@ use tokio_util::task::TaskTracker; use crate::SshState; use crate::exec::run_exec; +use crate::identity::{self, Asserted, Credential, Verdict}; pub(crate) struct KnotSshServer { pub(crate) state: Arc>, @@ -32,18 +33,27 @@ impl Server for KnotSshServer { pub(crate) struct KnotSession { state: Arc>, tracker: TaskTracker, - key: Option, + credential: Option, + asserted: Option, channels: HashMap>, protocols: HashSet, peer: Option, } +fn reject() -> Auth { + Auth::Reject { + proceed_with_methods: None, + partial_success: false, + } +} + impl KnotSession { fn new(state: Arc>, tracker: TaskTracker, peer: Option) -> Self { Self { state, tracker, - key: None, + credential: None, + asserted: None, channels: HashMap::new(), protocols: HashSet::new(), peer, @@ -51,32 +61,54 @@ impl KnotSession { } } +impl KnotSession { + async fn decide( + &mut self, + user: &str, + public_key: &ssh_key::PublicKey, + ) -> Option<(Verdict, OfferedKey)> { + let key = OfferedKey::from_bytes(public_key.to_bytes().ok()?); + let verdict = identity::verify( + &self.state, + OwnerRef::parse(user), + &key, + self.peer, + &mut self.asserted, + ) + .await; + Some((verdict, key)) + } +} + impl Handler for KnotSession { type Error = russh::Error; + async fn auth_publickey_offered( + &mut self, + user: &str, + public_key: &ssh_key::PublicKey, + ) -> Result { + match self.decide(user, public_key).await { + Some((Verdict::Refused, _)) | None => Ok(reject()), + Some(_) => Ok(Auth::Accept), + } + } + async fn auth_publickey( &mut self, - _user: &str, + user: &str, public_key: &ssh_key::PublicKey, ) -> Result { - let reject = Auth::Reject { - proceed_with_methods: None, - partial_success: false, - }; - let Ok(blob) = public_key.to_bytes() else { - return Ok(reject); - }; - let key = OfferedKey::from_bytes(blob); - if self - .state - .roster - .recognizes_fresh(&key, &self.state.index, &self.state.atproto) - .await - { - self.key = Some(key); - Ok(Auth::Accept) - } else { - Ok(reject) + match self.decide(user, public_key).await { + Some((Verdict::Identified(did), _)) => { + self.credential = Some(Credential::Identified(did)); + Ok(Auth::Accept) + } + Some((Verdict::Offered, key)) => { + self.credential = Some(Credential::Offered(key)); + Ok(Auth::Accept) + } + Some((Verdict::Refused, _)) | None => Ok(reject()), } } @@ -146,15 +178,16 @@ impl Handler for KnotSession { channel: ChannelId, session: &mut Session, ) -> Result<(), Self::Error> { - let Some(handle) = self.channels.remove(&channel) else { + let (Some(handle), Some(credential)) = + (self.channels.remove(&channel), self.credential.clone()) + else { session.channel_failure(channel)?; return Ok(()); }; session.channel_success(channel)?; let state = Arc::clone(&self.state); - let key = self.key.clone(); self.tracker.spawn(async move { - crate::exec::run_greeting(state, key, handle).await; + crate::exec::run_greeting(state, credential, handle).await; }); Ok(()) } @@ -165,18 +198,19 @@ impl Handler for KnotSession { data: &[u8], session: &mut Session, ) -> Result<(), Self::Error> { - let Some(handle) = self.channels.remove(&channel) else { + let (Some(handle), Some(credential)) = + (self.channels.remove(&channel), self.credential.clone()) + else { session.channel_failure(channel)?; return Ok(()); }; session.channel_success(channel)?; let protocol_v2 = self.protocols.remove(&channel); let state = Arc::clone(&self.state); - let key = self.key.clone(); let peer = self.peer; let command = data.to_vec(); self.tracker.spawn(async move { - run_exec(state, key, handle, &command, protocol_v2, peer).await; + run_exec(state, credential, handle, &command, protocol_v2, peer).await; }); Ok(()) } diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs index 50195783..64afdc14 100644 --- a/knot2/crates/knot-ssh/tests/ssh_push.rs +++ b/knot2/crates/knot-ssh/tests/ssh_push.rs @@ -1,14 +1,15 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::process::Command; -use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; use futures::stream::StreamExt; use knot_atproto::Atproto; use knot_cob::{CobHome, CobStore}; use knot_cobs::{CollaboratorsChange, Grant, MembersChange, Registration, RegistryChange}; use knot_git::{ArchiveLimit, Layout, Repo}; -use knot_index::Index; +use knot_index::{Index, Resolved}; use knot_pack::MaxWireBytes; use knot_postreceive::LanguagesPushBudget; use knot_runtime::{ @@ -133,6 +134,18 @@ fn not_found() -> HttpResponse { } } +fn forever() -> knot_index::KeyLease { + knot_index::KeyTtl::from_secs(u32::MAX.into()).lease_from(UnixSeconds::new(0)) +} + +fn server_error() -> HttpResponse { + HttpResponse { + status: http::StatusCode::INTERNAL_SERVER_ERROR, + headers: http::HeaderMap::new(), + body: bytes::Bytes::new(), + } +} + fn fake_http(published_line: String) -> impl knot_runtime::HttpTransport { let signer = K256Signer::generate(&SeededEntropy::new(1)); let pds = format!("https://{PDS_HOST}"); @@ -154,7 +167,54 @@ fn fake_http(published_line: String) -> impl knot_runtime::HttpTransport { }) } -fn multi_http(identities: HashMap>) -> impl knot_runtime::HttpTransport { +#[derive(Default, Clone)] +struct Accounts { + identities: Arc>>>, + unreachable: Arc>>, + listings: Arc, +} + +impl Accounts { + fn publishing(identities: HashMap>) -> Self { + Self { + identities: Arc::new(Mutex::new(identities)), + ..Self::default() + } + } + + fn unreachable(self, dids: HashSet) -> Self { + *self.unreachable.lock().unwrap() = dids; + self + } + + fn restore(&self, did: &str) { + self.unreachable.lock().unwrap().remove(did); + } + + fn publish(&self, did: &str, line: String) { + self.identities + .lock() + .unwrap() + .entry(did.to_string()) + .or_default() + .push(line); + } + + fn published_by(&self, did: &str) -> Vec { + self.identities + .lock() + .unwrap() + .get(did) + .cloned() + .unwrap_or_default() + } + + fn listings(&self) -> usize { + self.listings.load(Ordering::SeqCst) + } +} + +fn multi_http(accounts: Accounts) -> impl knot_runtime::HttpTransport { let signer = K256Signer::generate(&SeededEntropy::new(77)); FakeHttp::new(move |request| { let host = request.url.host_str().unwrap_or_default().to_string(); @@ -169,7 +229,11 @@ fn multi_http(identities: HashMap>) -> impl knot_runtime::Ht .find(|(key, _)| key == "repo") .map(|(_, value)| value.into_owned()) .unwrap_or_default(); - let lines = identities.get(&repo).cloned().unwrap_or_default(); + accounts.listings.fetch_add(1, Ordering::SeqCst); + if accounts.unreachable.lock().unwrap().contains(&repo) { + return Ok(server_error()); + } + let lines = accounts.published_by(&repo); let refs: Vec<&str> = lines.iter().map(String::as_str).collect(); return Ok(ok_body(list_records_body(&refs))); } @@ -995,6 +1059,66 @@ async fn cob_ref_guard_lifecycle() { ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_filled_key_set_refuses_an_unregistered_key_and_an_acl_write_reopens_the_check() { + let fx = fixture().await; + let head = seed_work(&fx.work); + let head_oid = Oid::from_hex(&head).unwrap(); + + fx.index.keys().mark_ready(fx.index.generation()); + fx.index.refresh_members().unwrap(); + let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await; + assert!( + ok, + "a grant written after the key set was read must reopen the check, or whoever it \ + grants is refused at the handshake until the next fill pass:\n{out}" + ); + assert_eq!( + main_tip(&fx.server.layout, &fx.server.repo_did), + Some(head_oid) + ); + + fx.index.keys().record( + &AccountDid::new(OWNER_DID).unwrap(), + vec![knot_types::OfferedKey::from_bytes(registered_blob(&fx))], + forever(), + ); + fx.index.keys().mark_ready(fx.index.generation()); + + let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered"); + let two_ids = format!( + "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \ + -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes", + fx.key_path + ); + let (ok, out) = { + let (work, url) = (fx.work.clone(), fx.url.clone()); + tokio::task::spawn_blocking(move || { + git( + &work, + &[("GIT_SSH_COMMAND", &two_ids)], + &["push", "-q", &url, "main:refs/heads/second"], + ) + }) + .await + .unwrap() + }; + assert!( + ok, + "a filled key set refuses the unregistered key, so the client offers its registered key \ + without the url identifying anybody:\n{out}" + ); + assert_eq!( + fx.server + .layout + .open(&fx.server.repo_did) + .unwrap() + .find_ref(&RefName::new("refs/heads/second").unwrap()) + .unwrap(), + Some(head_oid) + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn key_recognition_edge_cases() { let fx = fixture().await; @@ -1020,9 +1144,26 @@ async fn key_recognition_edge_cases() { .unwrap() }; assert!( - ok, - "rejecting unregistered key must let client cycle to the registered one:\n{out}" + !ok, + "with the key set still filling, the push is checked against whichever key the client \ + offers first:\n{out}" + ); + assert!( + out.contains("@nel.pet"), + "refusal lists who may push, so the pusher knows which key to offer:\n{out}" + ); + assert!( + out.contains("IdentitiesOnly"), + "refusal states how a multi-key client can offer its registered key:\n{out}" ); + assert_eq!( + main_tip(&fx.server.layout, &fx.server.repo_did), + None, + "the refused push leaves the repo empty" + ); + + let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await; + assert!(ok, "offering only the registered key must succeed:\n{out}"); assert_eq!( main_tip(&fx.server.layout, &fx.server.repo_did), Some(head_oid) @@ -1035,9 +1176,9 @@ async fn key_recognition_edge_cases() { .to_bytes() .unwrap(); fx.index.keys().record( - &AccountDid::new("did:plc:whelk").unwrap(), + &AccountDid::new("did:plc:cuttle").unwrap(), vec![knot_types::OfferedKey::from_bytes(blob)], - knot_index::KeyTtl::from_secs(u32::MAX.into()).lease_from(knot_types::UnixSeconds::new(0)), + forever(), ); let (ok, out) = push( &fx.work, @@ -1081,14 +1222,9 @@ fn a_group_or_other_readable_host_key_is_refused_on_load() { ); } -async fn launch( - host_key_dir: &Path, - layout: Layout, - index: Arc, - identities: HashMap>, -) -> u16 { +async fn launch(host_key_dir: &Path, layout: Layout, index: Arc, accounts: Accounts) -> u16 { let atproto = Arc::new(Atproto::new( - multi_http(identities), + multi_http(accounts), ManualClock::new(UnixMicros::new(1_000_000_000)), KnotId::new("did:web:nel.pet").unwrap(), knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(), @@ -1126,8 +1262,209 @@ async fn launch( } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn a_collaborator_pushes_its_repo_but_a_recognized_key_is_denied_on_a_repo_it_has_no_grant_on() - { +async fn a_handle_in_the_url_identifies_a_visitor_and_lets_a_multi_key_client_find_its_key() { + let fx = fixture().await; + let head = seed_work(&fx.work); + let head_oid = Oid::from_hex(&head).unwrap(); + + let port = fx.server.port; + let greeted_key = fx.key_path.clone(); + let (_ok, out) = + tokio::task::spawn_blocking(move || ssh_bare_as(&greeted_key, "nel.pet", port)) + .await + .unwrap(); + assert!( + out.contains("@nel.pet"), + "an asserted handle identifies the visitor on first contact, with an empty cache:\n{out}" + ); + + let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered"); + let two_ids = format!( + "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \ + -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes", + fx.key_path + ); + let identified = format!("ssh://nel.pet@127.0.0.1:{}/{REPO_DID}", fx.server.port); + let (ok, out) = { + let (work, url) = (fx.work.clone(), identified.clone()); + tokio::task::spawn_blocking(move || { + git( + &work, + &[("GIT_SSH_COMMAND", &two_ids)], + &["push", "-q", &url, "main"], + ) + }) + .await + .unwrap() + }; + assert!( + ok, + "a handle in the url lets the knot refuse the unregistered key so the client offers the \ + next key:\n{out}" + ); + assert_eq!( + main_tip(&fx.server.layout, &fx.server.repo_did), + Some(head_oid) + ); + assert_eq!( + fx.index.owner_of_key( + &knot_types::OfferedKey::from_bytes(registered_blob(&fx)), + UnixSeconds::new(0), + ), + Resolved::Ready(None), + "an asserted handle is whatever the client typed, so the keys read for it mustn't enter \ + the set, or anyone can fill the key budget by asserting handles" + ); +} + +fn registered_blob(fx: &Fixture) -> Vec { + russh::keys::ssh_key::PublicKey::from_openssh( + &std::fs::read_to_string(fx.scratch.path().join("client.pub")).unwrap(), + ) + .unwrap() + .to_bytes() + .unwrap() +} + +fn registered_index( + scratch: &TempDir, + budget: knot_index::KeyBudget, +) -> (Layout, RepoDid, Arc) { + let meta_path = scratch.path().join("meta"); + Repo::create(&meta_path).unwrap(); + let layout = Layout::new(scratch.path().join("repos")); + let repo_did = RepoDid::new(REPO_DID).unwrap(); + layout.create(&repo_did).unwrap(); + + let signer = K256Signer::generate(&SeededEntropy::new(2)); + let meta = Repo::open(&meta_path).unwrap(); + CobStore::new(&meta) + .create( + &CobHome::from(&KnotId::new("did:web:nel.pet").unwrap()), + &RegistryChange::Register(Registration { + owner: OwnerDid::new(OWNER_DID).unwrap(), + rkey: RepoRkey::new(REPO_NAME).unwrap(), + name: RepoName::new(REPO_NAME).unwrap(), + repo: repo_did.clone(), + created_at: UnixSeconds::new(1), + }), + &signer, + UnixSeconds::new(1), + ) + .unwrap(); + + let index = Arc::new(Index::with_key_budget(meta_path, layout.clone(), budget)); + index.rebuild().unwrap(); + index.warm_collaborators(); + (layout, repo_did, index) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_unreachable_pds_reads_as_transient_and_its_keys_get_in_once_it_recovers() { + let scratch = tempfile::tempdir().unwrap(); + let (stale_key, stale_line) = keygen(scratch.path(), "stale"); + let (fresh_key, fresh_line) = keygen(scratch.path(), "fresh"); + let (layout, repo_did, index) = registered_index(&scratch, knot_index::KeyBudget::DEFAULT); + + let accounts = Accounts::publishing(HashMap::from([(OWNER_DID.to_string(), vec![stale_line])])) + .unreachable(HashSet::from([OWNER_DID.to_string()])); + let port = launch( + &scratch.path().join("hostkey"), + layout.clone(), + Arc::clone(&index), + accounts.clone(), + ) + .await; + let url = format!("ssh://git@127.0.0.1:{port}/{REPO_DID}"); + + let work = scratch.path().join("work"); + let head = seed_work(&work); + let (ok, out) = push(&work, &url, &stale_key, &["main"]).await; + assert!( + !ok, + "a push mustn't be accepted while the owner's records are unreadable:\n{out}" + ); + assert!( + out.contains("retry shortly"), + "an unreadable PDS must read as transient:\n{out}" + ); + assert!( + !out.contains("doesn't match"), + "a transient failure mustn't be reported to the pusher as a wrong key:\n{out}" + ); + assert_eq!( + main_tip(&layout, &repo_did), + None, + "the refused push leaves the repo empty" + ); + + accounts.restore(OWNER_DID); + let (ok, out) = push(&work, &url, &stale_key, &["main"]).await; + assert!( + ok, + "the key the owner publishes must push once its PDS answers again:\n{out}" + ); + + accounts.publish(OWNER_DID, fresh_line); + let (ok, out) = push(&work, &url, &fresh_key, &["main", "--force"]).await; + assert!( + ok, + "a key the owner published after the knot last read the account must get in on the next \ + push, or publishing a second key locks its owner out until a fill pass catches up:\n{out}" + ); + assert_eq!( + main_tip(&layout, &repo_did), + Some(Oid::from_hex(&head).unwrap()) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_account_the_budget_couldnt_fit_still_clears_the_handshake_and_pushes() { + let scratch = tempfile::tempdir().unwrap(); + let (owner_key, owner_line) = keygen(scratch.path(), "owner"); + let (layout, repo_did, index) = + registered_index(&scratch, knot_index::KeyBudget::from_bytes(200)); + + let blob = russh::keys::ssh_key::PublicKey::from_openssh(&owner_line) + .unwrap() + .to_bytes() + .unwrap(); + assert_eq!( + index.keys().record( + &AccountDid::new(OWNER_DID).unwrap(), + vec![knot_types::OfferedKey::from_bytes(blob)], + forever(), + ), + knot_index::KeyRecord::Unheld, + "a 200-byte budget records the read without keeping the key" + ); + index.keys().mark_ready(index.generation()); + + let port = launch( + &scratch.path().join("hostkey"), + layout.clone(), + Arc::clone(&index), + Accounts::publishing(HashMap::from([(OWNER_DID.to_string(), vec![owner_line])])), + ) + .await; + let url = format!("ssh://git@127.0.0.1:{port}/{REPO_DID}"); + + let work = scratch.path().join("work"); + let head = seed_work(&work); + let (ok, out) = push(&work, &url, &owner_key, &["main"]).await; + assert!( + ok, + "the set can't fit the owner's keys, so the handshake must defer to the push check \ + instead of refusing a key the accounts on file don't publish:\n{out}" + ); + assert_eq!( + main_tip(&layout, &repo_did), + Some(Oid::from_hex(&head).unwrap()) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_collaborator_pushes_its_repo_but_is_denied_on_a_repo_it_doesnt_collaborate_on() { const REPO_A: &str = "did:plc:squid"; const REPO_B: &str = "did:plc:clam"; const OWNER: &str = "did:plc:nel"; @@ -1211,11 +1548,12 @@ async fn a_collaborator_pushes_its_repo_but_a_recognized_key_is_denied_on_a_repo (OWNER.to_string(), vec![owner_line]), (COLLAB.to_string(), vec![collab_line]), ]); + let accounts = Accounts::publishing(identities); let port = launch( &scratch.path().join("hostkey"), layout.clone(), Arc::clone(&index), - identities, + accounts.clone(), ) .await; @@ -1239,14 +1577,24 @@ async fn a_collaborator_pushes_its_repo_but_a_recognized_key_is_denied_on_a_repo let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await; assert!( !denied, - "key recognized via repo A but with no grant on repo B must be denied, recognition is \ - not authorization:\n{out}" + "a key that pushes repo A must be denied on repo B, where its owner was never granted:\n{out}" ); assert!( main_tip(&layout, &repo_b).is_none(), "denied cross-repo push must land nothing on repo B" ); + let after_first_denial = accounts.listings(); + let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await; + assert!(!denied, "the second attempt is denied the same way:\n{out}"); + assert_eq!( + accounts.listings(), + after_first_denial, + "repo B's owner was read during the first denial and is on file, so retrying mustn't \ + read that PDS again, or anyone with a key can make the knot fetch from a third party \ + at will:\n{out}" + ); + let work_owner = scratch.path().join("work_owner_b"); let head_owner = seed_work(&work_owner); let (ok, out) = push(&work_owner, &url_b, &owner_key, &["main"]).await; @@ -1259,6 +1607,10 @@ async fn a_collaborator_pushes_its_repo_but_a_recognized_key_is_denied_on_a_repo } fn ssh_bare(key_path: &str, port: u16) -> (bool, String) { + ssh_bare_as(key_path, "git", port) +} + +fn ssh_bare_as(key_path: &str, user: &str, port: u16) -> (bool, String) { let out = Command::new("ssh") .args([ "-i", @@ -1275,7 +1627,7 @@ fn ssh_bare(key_path: &str, port: u16) -> (bool, String) { "BatchMode=yes", "-p", &port.to_string(), - "git@127.0.0.1", + &format!("{user}@127.0.0.1"), ]) .output() .expect("ssh runs"); @@ -1290,18 +1642,32 @@ fn ssh_bare(key_path: &str, port: u16) -> (bool, String) { } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn a_bare_ssh_session_greets_the_recognized_user() { +async fn a_bare_ssh_session_greets_a_visitor_then_identifies_them_once_they_have_pushed() { let fx = fixture().await; let port = fx.server.port; + + let key_path = fx.key_path.clone(); + let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port)) + .await + .unwrap(); + assert!(out.contains("knot.test"), "greeting names the knot:\n{out}"); + assert!( + out.contains("ssh key"), + "a visitor the knot can't identify yet learns what a push needs:\n{out}" + ); + + seed_work(&fx.work); + let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await; + assert!(ok, "seeding main must succeed:\n{out}"); + let key_path = fx.key_path.clone(); let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port)) .await .unwrap(); assert!( out.contains("@nel.pet"), - "greeting resolves and addresses the user by handle:\n{out}" + "a push teaches the knot the key, so the next greeting uses the handle:\n{out}" ); - assert!(out.contains("knot.test"), "greeting names the knot:\n{out}"); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] diff --git a/knot2/example.toml b/knot2/example.toml index 9d8482e3..b7ef7965 100644 --- a/knot2/example.toml +++ b/knot2/example.toml @@ -516,6 +516,9 @@ # Default value: ["Hi {user}! You're authenticated to {knot} knot.", "This knot serves git over ssh, so there's no shell here. :P", "Clone repo with: git clone {knot}:"] #greeting = ["Hi {user}! You're authenticated to {knot} knot.", "This knot serves git over ssh, so there's no shell here. :P", "Clone repo with: git clone {knot}:"] +# Default value: ["Hi there! This is the {knot} knot.", "This knot serves git over ssh, so there's no shell here. :P", "Clone repo with: git clone {knot}:", "Publish your ssh key to your atproto account so this knot can identify your pushes.", "Put your handle in the url, as in yourhandle@{knot}:, so your ssh client can find your registered key on its own."] +#greeting_unknown = ["Hi there! This is the {knot} knot.", "This knot serves git over ssh, so there's no shell here. :P", "Clone repo with: git clone {knot}:", "Publish your ssh key to your atproto account so this knot can identify your pushes.", "Put your handle in the url, as in yourhandle@{knot}:, so your ssh client can find your registered key on its own."] + # Default value: "knot: unsupported command" #unsupported_command = "knot: unsupported command" @@ -531,8 +534,11 @@ # Default value: "knot: LFS isn't enabled on this knot" #lfs_disabled = "knot: LFS isn't enabled on this knot" -# Default value: "knot: your ssh key isn't registered to a user authorized to push here. If you offer several keys, make sure the registered one is offered first." -#key_not_registered = "knot: your ssh key isn't registered to a user authorized to push here. If you offer several keys, make sure the registered one is offered first." +# Default value: "knot: this ssh key doesn't match any key published by the accounts that may push here. Authorized: {authorized}. If your agent offers several keys, add -o IdentitiesOnly=yes so it offers your registered key." +#key_not_registered = "knot: this ssh key doesn't match any key published by the accounts that may push here. Authorized: {authorized}. If your agent offers several keys, add -o IdentitiesOnly=yes so it offers your registered key." + +# Default value: "knot: couldn't read the account records needed to check your ssh key, retry shortly" +#identity_unavailable = "knot: couldn't read the account records needed to check your ssh key, retry shortly" # Default value: "knot: you aren't authorized to push to this repository." #push_denied = "knot: you aren't authorized to push to this repository." -- 2.51.2