+33 -1

knot2/crates/knot-messages/src/lib.rs

View file Reviewed

··· 78 78 UrlKey { Url = "url" } 79 79 CiLogsKey { Host = "host", Port = "port", Repo = "repo", Sha = "sha" } 80 80 GreetingKey { User = "user", Knot = "knot" } 81 81 + AuthorizedKey { Authorized = "authorized" } 81 82 CountKey { Count = "count" } 82 83 RefKey { Ref = "ref" } 83 84 ErrorKey { Error = "error" } ··· 148 149 "This knot serves git over ssh, so there's no shell here. :P", 149 150 "Clone repo with: git clone {knot}:<repoDID>" 150 151 ], 152 152 + greeting_unknown: Lines<KnotKey> = [ 153 153 + "Hi there! This is the {knot} knot.", 154 154 + "This knot serves git over ssh, so there's no shell here. :P", 155 155 + "Clone repo with: git clone {knot}:<repoDID>", 156 156 + "Publish your ssh key to your atproto account so this knot can identify your pushes.", 157 157 + "Put your handle in the url, as in yourhandle@{knot}:<repoDID>, so your ssh client can find your registered key on its own." 158 158 + ], 151 159 unsupported_command: Line<NoKeys> = "knot: unsupported command", 152 160 too_many_operations: Line<NoKeys> = "knot: too many concurrent operations from your address, try again shortly", 153 161 repo_not_found: Line<NoKeys> = "knot: repository not found", 154 162 index_warming: Line<NoKeys> = "knot: repository index is warming, retry shortly", 155 163 lfs_disabled: Line<NoKeys> = "knot: LFS isn't enabled on this knot", 156 156 - key_not_registered: Line<NoKeys> = "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.", 164 164 + key_not_registered: Line<AuthorizedKey> = "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.", 165 165 + identity_unavailable: Line<NoKeys> = "knot: couldn't read the account records needed to check your ssh key, retry shortly", 157 166 push_denied: Line<NoKeys> = "knot: you aren't authorized to push to this repository.", 158 167 shutting_down: Line<NoKeys> = "knot: server is shutting down", 159 168 archive_malformed: Line<NoKeys> = "knot: malformed upload-archive request", ··· 304 313 }); 305 314 assert!(lines[0].contains("@nel.pet")); 306 315 assert!(lines.iter().any(|line| line.contains("oyster.cafe"))); 316 316 + } 317 317 + 318 318 + #[test] 319 319 + fn an_unidentified_visitor_is_greeted_and_shown_what_a_push_needs() { 320 320 + let catalog = Catalog::defaults(); 321 321 + let lines = catalog 322 322 + .ssh 323 323 + .greeting_unknown 324 324 + .lines(|KnotKey::Knot| "oyster.cafe".to_string()); 325 325 + assert!(lines[0].contains("oyster.cafe")); 326 326 + assert!( 327 327 + lines.iter().any(|line| line.contains("ssh key")), 328 328 + "a visitor the knot can't identify learns what a push needs: {lines:?}" 329 329 + ); 330 330 + 331 331 + let denial = catalog 332 332 + .ssh 333 333 + .key_not_registered 334 334 + .line(|AuthorizedKey::Authorized| "@nel.pet".to_string()); 335 335 + assert!( 336 336 + denial.contains("@nel.pet"), 337 337 + "the denial lists who may push instead: {denial}" 338 338 + ); 307 339 } 308 340 309 341 #[test]

+218 -73

knot2/crates/knot-ssh/src/exec.rs

View file Reviewed

··· 1 1 use std::net::IpAddr; 2 2 use std::path::PathBuf; 3 3 use std::sync::Arc; 4 4 + use std::sync::atomic::{AtomicBool, Ordering}; 4 5 use std::time::Duration; 5 6 6 7 use futures::StreamExt; ··· 8 9 use knot_index::Resolved; 9 10 use knot_lfs::TransferOp; 10 11 use knot_pack::{PackError, PackLimits, RepoLookup}; 12 12 + use knot_resource::SubjectKey; 11 13 use knot_runtime::{Clock, HttpTransport}; 12 12 - use knot_types::{AccountDid, ClonePath, ObjectFormat, OfferedKey, OwnerDid, RepoDid}; 14 14 + use knot_types::{AccountDid, ClonePath, ObjectFormat, OwnerDid, RepoDid}; 13 15 use russh::Channel; 14 16 use russh::server::Msg; 15 17 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; ··· 17 19 use tokio::sync::mpsc; 18 20 19 21 use crate::SshState; 22 22 + use crate::identity::Credential; 20 23 21 24 const READ_CHUNK: usize = 64 * 1024; 22 25 const MAX_UPLOAD_REQUEST: usize = 16 * 1024 * 1024; ··· 25 28 const LFS_PROGRESS_GRACE: Duration = Duration::from_secs(60); 26 29 const LFS_PROGRESS_FLOOR_BYTES_PER_SEC: u64 = 1024; 27 30 const LFS_STALL_TIMEOUT: Duration = Duration::from_secs(120); 31 31 + const CANDIDATE_FANOUT: usize = 4; 32 32 + const AUTHORIZED_NAMES_SHOWN: usize = 4; 28 33 29 34 fn lfs_within_progress_budget(waited: Duration, moved_bytes: u64) -> bool { 30 35 waited ··· 119 124 120 125 pub(crate) async fn run_exec<H: HttpTransport, C: Clock>( 121 126 state: Arc<SshState<H, C>>, 122 122 - key: Option<OfferedKey>, 127 127 + credential: Credential, 123 128 channel: Channel<Msg>, 124 129 command: &[u8], 125 130 protocol_v2: bool, ··· 151 156 RepoRef::Did(did) => ResolvedRef::Did(did), 152 157 RepoRef::OwnerPath(owner, candidates) => ResolvedRef::OwnerPath(owner, candidates), 153 158 RepoRef::HandlePath(owner_handle, candidates) => { 159 159 + let Some(_lookup_permit) = state.lookup_slots.try_acquire() else { 160 160 + tracing::warn!( 161 161 + ?peer, 162 162 + "ssh exec rejected, the lookup budget can't resolve another handle" 163 163 + ); 164 164 + return fail(channel, &state.catalog.ssh.too_many_operations.text()).await; 165 165 + }; 154 166 match state 155 167 .atproto 156 168 .resolve_handle_to_did(&owner_handle) ··· 188 200 match service { 189 201 Service::Upload => serve_upload(state, channel, repo_did, protocol_v2).await, 190 202 Service::UploadArchive => serve_upload_archive(state, channel, repo_did).await, 191 191 - Service::Receive => serve_receive(state, key, channel, repo_did).await, 192 192 - Service::Lfs(op) => serve_lfs(state, key, channel, repo_did, op).await, 203 203 + Service::Receive => serve_receive(state, credential, channel, repo_did, peer).await, 204 204 + Service::Lfs(op) => serve_lfs(state, credential, channel, repo_did, op, peer).await, 193 205 } 194 206 } 195 207 196 208 async fn serve_lfs<H: HttpTransport, C: Clock>( 197 209 state: Arc<SshState<H, C>>, 198 198 - key: Option<OfferedKey>, 210 210 + credential: Credential, 199 211 mut channel: Channel<Msg>, 200 212 repo_did: RepoDid, 201 213 op: TransferOp, 214 214 + peer: Option<IpAddr>, 202 215 ) { 203 216 let Some(lfs) = state.lfs.clone() else { 204 217 return fail(channel, &state.catalog.ssh.lfs_disabled.text()).await; 205 218 }; 206 206 - if op == TransferOp::Upload { 207 207 - let pusher = resolve_pusher(&state, key.as_ref(), &repo_did).await; 208 208 - let allowed = pusher.as_ref().is_some_and(|did| { 209 209 - let acl = KnotAcl::new(&state.admins, state.admission, &state.index); 210 210 - can_push(&acl, did, &repo_did).is_allowed() 211 211 - }); 212 212 - if !allowed { 213 213 - tracing::warn!( 214 214 - repo = repo_did.as_str(), 215 215 - registered = pusher.is_some(), 216 216 - "ssh lfs upload denied" 217 217 - ); 218 218 - let message = match pusher { 219 219 - None => state.catalog.ssh.key_not_registered.text(), 220 220 - Some(_) => state.catalog.ssh.push_denied.text(), 221 221 - }; 222 222 - return fail(channel, &message).await; 223 223 - } 219 219 + if op == TransferOp::Upload 220 220 + && let PushAuth::Refused { reason, message } = 221 221 + authorize_push(&state, &credential, &repo_did, peer).await 222 222 + { 223 223 + tracing::warn!(repo = repo_did.as_str(), reason, "ssh lfs upload denied"); 224 224 + return fail(channel, &message).await; 224 225 } 225 226 let permit = match Arc::clone(&lfs.slots).acquire_owned().await { 226 227 Ok(permit) => permit, ··· 624 625 625 626 async fn serve_receive<H: HttpTransport, C: Clock>( 626 627 state: Arc<SshState<H, C>>, 627 627 - key: Option<OfferedKey>, 628 628 + credential: Credential, 628 629 mut channel: Channel<Msg>, 629 630 repo_did: RepoDid, 631 631 + peer: Option<IpAddr>, 630 632 ) { 631 633 let advert = { 632 634 let layout = state.layout.clone(); ··· 648 650 return; 649 651 } 650 652 651 651 - let pusher = resolve_pusher(&state, key.as_ref(), &repo_did).await; 652 652 - let allowed = |did: &AccountDid| { 653 653 - let acl = KnotAcl::new(&state.admins, state.admission, &state.index); 654 654 - can_push(&acl, did, &repo_did).is_allowed() 655 655 - }; 656 656 - let committer = match pusher { 657 657 - Some(did) if allowed(&did) => did, 658 658 - Some(_) => { 659 659 - tracing::warn!( 660 660 - repo = repo_did.as_str(), 661 661 - registered = true, 662 662 - "ssh push denied" 663 663 - ); 664 664 - return fail(channel, &state.catalog.ssh.push_denied.text()).await; 665 665 - } 666 666 - None => { 667 667 - tracing::warn!( 668 668 - repo = repo_did.as_str(), 669 669 - registered = false, 670 670 - "ssh push denied" 671 671 - ); 672 672 - return fail(channel, &state.catalog.ssh.key_not_registered.text()).await; 653 653 + let committer = match authorize_push(&state, &credential, &repo_did, peer).await { 654 654 + PushAuth::Allowed(did) => did, 655 655 + PushAuth::Refused { reason, message } => { 656 656 + tracing::warn!(repo = repo_did.as_str(), reason, "ssh push denied"); 657 657 + return fail(channel, &message).await; 673 658 } 674 659 }; 675 660 ··· 754 739 755 740 pub(crate) async fn run_greeting<H: HttpTransport, C: Clock>( 756 741 state: Arc<SshState<H, C>>, 757 757 - key: Option<OfferedKey>, 742 742 + credential: Credential, 758 743 channel: Channel<Msg>, 759 744 ) { 760 760 - let who = greeting_identity(&state, key.as_ref()).await; 761 761 - let greeting = state.catalog.ssh.greeting.lines(|key| match key { 762 762 - knot_messages::GreetingKey::User => who.clone(), 763 763 - knot_messages::GreetingKey::Knot => state.hostname.as_str().to_string(), 764 764 - }); 745 745 + let knot = state.hostname.as_str().to_string(); 746 746 + let greeting = match greeting_visitor(&state, &credential).await { 747 747 + Visitor::Named(who) => state.catalog.ssh.greeting.lines(|key| match key { 748 748 + knot_messages::GreetingKey::User => who.clone(), 749 749 + knot_messages::GreetingKey::Knot => knot.clone(), 750 750 + }), 751 751 + Visitor::Unknown => state 752 752 + .catalog 753 753 + .ssh 754 754 + .greeting_unknown 755 755 + .lines(|knot_messages::KnotKey::Knot| knot.clone()), 756 756 + }; 765 757 if greeting.is_empty() { 766 758 return finish(channel, 0).await; 767 759 } ··· 772 764 finish(channel, 0).await; 773 765 } 774 766 775 775 - async fn greeting_identity<H: HttpTransport, C: Clock>( 767 767 + async fn greeting_visitor<H: HttpTransport, C: Clock>( 776 768 state: &Arc<SshState<H, C>>, 777 777 - key: Option<&OfferedKey>, 778 778 - ) -> String { 779 779 - let Some(did) = key.and_then(|key| state.roster.did_for(key)) else { 780 780 - return "there".to_string(); 769 769 + credential: &Credential, 770 770 + ) -> Visitor { 771 771 + let did = match credential { 772 772 + Credential::Identified(did) => did.clone(), 773 773 + Credential::Offered(key) => { 774 774 + match state.index.owner_of_key(key, state.atproto.now().seconds()) { 775 775 + Resolved::Ready(Some(did)) => did, 776 776 + _ => return Visitor::Unknown, 777 777 + } 778 778 + } 781 779 }; 782 780 match knot_receive::resolve_handle(&state.atproto, &state.slots.resolve, &did).await { 783 783 - Some(handle) => format!("@{}", handle.as_str()), 784 784 - None => did.as_str().to_string(), 781 781 + Some(handle) => Visitor::Named(format!("@{}", handle.as_str())), 782 782 + None => Visitor::Named(did.as_str().to_string()), 783 783 + } 784 784 + } 785 785 + 786 786 + enum PusherLookup { 787 787 + Matched(AccountDid), 788 788 + Unmatched(Vec<AccountDid>), 789 789 + Unavailable, 790 790 + } 791 791 + 792 792 + enum PushAuth { 793 793 + Allowed(AccountDid), 794 794 + Refused { 795 795 + reason: &'static str, 796 796 + message: String, 797 797 + }, 798 798 + } 799 799 + 800 800 + enum Visitor { 801 801 + Named(String), 802 802 + Unknown, 803 803 + } 804 804 + 805 805 + async fn authorize_push<H: HttpTransport, C: Clock>( 806 806 + state: &Arc<SshState<H, C>>, 807 807 + credential: &Credential, 808 808 + repo: &RepoDid, 809 809 + peer: Option<IpAddr>, 810 810 + ) -> PushAuth { 811 811 + match resolve_pusher(state, credential, repo, peer).await { 812 812 + PusherLookup::Matched(did) => { 813 813 + let acl = KnotAcl::new(&state.admins, state.admission, &state.index); 814 814 + match can_push(&acl, &did, repo).is_allowed() { 815 815 + true => PushAuth::Allowed(did), 816 816 + false => PushAuth::Refused { 817 817 + reason: "unauthorized", 818 818 + message: state.catalog.ssh.push_denied.text(), 819 819 + }, 820 820 + } 821 821 + } 822 822 + PusherLookup::Unavailable => PushAuth::Refused { 823 823 + reason: "identity_unavailable", 824 824 + message: state.catalog.ssh.identity_unavailable.text(), 825 825 + }, 826 826 + PusherLookup::Unmatched(candidates) => { 827 827 + let authorized = describe_authorized(state, &candidates).await; 828 828 + PushAuth::Refused { 829 829 + reason: "unregistered_key", 830 830 + message: state 831 831 + .catalog 832 832 + .ssh 833 833 + .key_not_registered 834 834 + .line(|knot_messages::AuthorizedKey::Authorized| authorized.clone()), 835 835 + } 836 836 + } 837 837 + } 838 838 + } 839 839 + 840 840 + async fn describe_authorized<H: HttpTransport, C: Clock>( 841 841 + state: &Arc<SshState<H, C>>, 842 842 + candidates: &[AccountDid], 843 843 + ) -> String { 844 844 + let names: Vec<String> = futures::stream::iter( 845 845 + candidates 846 846 + .iter() 847 847 + .take(AUTHORIZED_NAMES_SHOWN) 848 848 + .cloned() 849 849 + .collect::<Vec<_>>(), 850 850 + ) 851 851 + .map(|did| { 852 852 + let state = Arc::clone(state); 853 853 + async move { 854 854 + match knot_receive::resolve_handle(&state.atproto, &state.slots.resolve, &did).await { 855 855 + Some(handle) => format!("@{}", handle.as_str()), 856 856 + None => did.as_str().to_string(), 857 857 + } 858 858 + } 859 859 + }) 860 860 + .buffered(CANDIDATE_FANOUT) 861 861 + .collect() 862 862 + .await; 863 863 + match ( 864 864 + names.as_slice(), 865 865 + candidates.len().saturating_sub(AUTHORIZED_NAMES_SHOWN), 866 866 + ) { 867 867 + ([], _) => "nobody".to_string(), 868 868 + (shown, 0) => shown.join(", "), 869 869 + (shown, hidden) => format!("{}, and {hidden} more", shown.join(", ")), 785 870 } 786 871 } 787 872 873 873 + fn probe_due<H: HttpTransport, C: Clock>(state: &Arc<SshState<H, C>>, did: &AccountDid) -> bool { 874 874 + state 875 875 + .probe_pace 876 876 + .reserve_now(&SubjectKey::new(did.as_str()), state.atproto.now()) 877 877 + } 878 878 + 788 879 async fn resolve_pusher<H: HttpTransport, C: Clock>( 789 880 state: &Arc<SshState<H, C>>, 790 790 - key: Option<&OfferedKey>, 881 881 + credential: &Credential, 791 882 repo: &RepoDid, 792 792 - ) -> Option<AccountDid> { 793 793 - let key = key?; 883 883 + peer: Option<IpAddr>, 884 884 + ) -> PusherLookup { 885 885 + let key = match credential { 886 886 + Credential::Identified(did) => return PusherLookup::Matched(did.clone()), 887 887 + Credential::Offered(key) => key, 888 888 + }; 794 889 let owner = match state.index.owner_of(repo) { 795 890 Resolved::Ready(Some(owner)) => Some(AccountDid::from(owner)), 796 891 _ => None, ··· 806 901 }; 807 902 let candidates: Vec<AccountDid> = owner.into_iter().chain(collaborators).collect(); 808 903 let now = state.atproto.now().seconds(); 809 809 - if let Resolved::Ready(Some(cached)) = state.index.owner_of_key(key, now) 810 810 - && candidates.contains(&cached) 811 811 - { 812 812 - return Some(cached); 904 904 + if let Some(publisher) = state.index.keys().publisher_among(&candidates, key, now) { 905 905 + return PusherLookup::Matched(publisher); 813 906 } 814 814 - let _permit = state.slots.resolve.acquire().await; 815 815 - let matches = futures::stream::iter(candidates).filter_map(|did| async move { 816 816 - let keys = state.atproto.resolve_pubkeys(&did).await.ok()?; 817 817 - keys.iter().any(|resolved| resolved == key).then_some(did) 818 818 - }); 819 819 - futures::pin_mut!(matches); 820 820 - matches.next().await 907 907 + let unread: Vec<AccountDid> = candidates 908 908 + .iter() 909 909 + .filter(|did| !state.index.keys().is_fresh(did, now) || probe_due(state, did)) 910 910 + .cloned() 911 911 + .collect(); 912 912 + if unread.is_empty() { 913 913 + tracing::debug!( 914 914 + ?peer, 915 915 + repo = repo.as_str(), 916 916 + candidates = candidates.len(), 917 917 + "push check has every candidate's keys on file, and the candidates don't publish \ 918 918 + the offered key" 919 919 + ); 920 920 + return PusherLookup::Unmatched(candidates); 921 921 + } 922 922 + let lease = state.key_ttl.lease_from(now); 923 923 + let unresolved = Arc::new(AtomicBool::new(false)); 924 924 + let read: Vec<Option<AccountDid>> = futures::stream::iter(unread) 925 925 + .map(|did| { 926 926 + let state = Arc::clone(state); 927 927 + let key = key.clone(); 928 928 + let unresolved = Arc::clone(&unresolved); 929 929 + async move { 930 930 + let _permit = state.slots.resolve.acquire().await; 931 931 + match state.atproto.resolve_pubkeys(&did).await { 932 932 + Ok(keys) => { 933 933 + let matches = keys.contains(&key); 934 934 + state.index.keys().record(&did, keys, lease); 935 935 + matches.then_some(did) 936 936 + } 937 937 + Err(error) if error.is_gone() => { 938 938 + tracing::debug!( 939 939 + did = did.as_str(), 940 940 + %error, 941 941 + "push check records an empty key set for a candidate whose DID document is gone" 942 942 + ); 943 943 + state.index.keys().record(&did, Vec::new(), lease); 944 944 + None 945 945 + } 946 946 + Err(error) => { 947 947 + tracing::debug!( 948 948 + did = did.as_str(), 949 949 + %error, 950 950 + "push check couldn't read a candidate's records" 951 951 + ); 952 952 + unresolved.store(true, Ordering::Relaxed); 953 953 + None 954 954 + } 955 955 + } 956 956 + } 957 957 + }) 958 958 + .buffered(CANDIDATE_FANOUT) 959 959 + .collect() 960 960 + .await; 961 961 + match read.into_iter().flatten().next() { 962 962 + Some(did) => PusherLookup::Matched(did), 963 963 + None if unresolved.load(Ordering::Relaxed) => PusherLookup::Unavailable, 964 964 + None => PusherLookup::Unmatched(candidates), 965 965 + } 821 966 } 822 967 823 968 async fn read_chunk<R: AsyncRead + Unpin>(

+152

knot2/crates/knot-ssh/src/identity.rs

View file Reviewed

··· 1 1 + use std::net::IpAddr; 2 2 + use std::sync::Arc; 3 3 + 4 4 + use knot_atproto::ClaimedKeys; 5 5 + use knot_index::{Coverage, Resolved}; 6 6 + use knot_runtime::{Clock, HttpTransport}; 7 7 + use knot_types::{AccountDid, OfferedKey, OwnerRef}; 8 8 + 9 9 + use crate::SshState; 10 10 + 11 11 + #[derive(Clone)] 12 12 + pub(crate) enum Credential { 13 13 + Identified(AccountDid), 14 14 + Offered(OfferedKey), 15 15 + } 16 16 + 17 17 + pub(crate) struct Asserted { 18 18 + claim: OwnerRef, 19 19 + outcome: Claimed, 20 20 + } 21 21 + 22 22 + enum Claimed { 23 23 + Publishes { 24 24 + did: AccountDid, 25 25 + keys: Vec<OfferedKey>, 26 26 + }, 27 27 + Unreadable, 28 28 + } 29 29 + 30 30 + pub(crate) enum Verdict { 31 31 + Identified(AccountDid), 32 32 + Offered, 33 33 + Refused, 34 34 + } 35 35 + 36 36 + pub(crate) async fn verify<H: HttpTransport, C: Clock>( 37 37 + state: &Arc<SshState<H, C>>, 38 38 + claim: Option<OwnerRef>, 39 39 + key: &OfferedKey, 40 40 + peer: Option<IpAddr>, 41 41 + asserted: &mut Option<Asserted>, 42 42 + ) -> Verdict { 43 43 + match claim { 44 44 + Some(claim) => match against_claim(state, claim, key, peer, asserted).await { 45 45 + Some(verdict) => verdict, 46 46 + None => against_key_set(state, key, peer), 47 47 + }, 48 48 + None => against_key_set(state, key, peer), 49 49 + } 50 50 + } 51 51 + 52 52 + fn against_key_set<H: HttpTransport, C: Clock>( 53 53 + state: &Arc<SshState<H, C>>, 54 54 + key: &OfferedKey, 55 55 + peer: Option<IpAddr>, 56 56 + ) -> Verdict { 57 57 + let now = state.atproto.now().seconds(); 58 58 + match ( 59 59 + state.index.owner_of_key(key, now), 60 60 + state.index.keys().coverage(), 61 61 + ) { 62 62 + (Resolved::Ready(Some(_)), _) => Verdict::Offered, 63 63 + (_, Coverage::Warming) => Verdict::Offered, 64 64 + (_, Coverage::Ready) => match state.index.keys().any_unheld() { 65 65 + true => Verdict::Offered, 66 66 + false => { 67 67 + if miss_worth_a_reread(state, peer) { 68 68 + state.index.keys().note_miss(); 69 69 + } 70 70 + Verdict::Refused 71 71 + } 72 72 + }, 73 73 + } 74 74 + } 75 75 + 76 76 + fn miss_worth_a_reread<H: HttpTransport, C: Clock>( 77 77 + state: &Arc<SshState<H, C>>, 78 78 + peer: Option<IpAddr>, 79 79 + ) -> bool { 80 80 + peer.is_none_or(|peer| state.miss_pace.reserve_now(&peer, state.atproto.now())) 81 81 + } 82 82 + 83 83 + async fn against_claim<H: HttpTransport, C: Clock>( 84 84 + state: &Arc<SshState<H, C>>, 85 85 + claim: OwnerRef, 86 86 + key: &OfferedKey, 87 87 + peer: Option<IpAddr>, 88 88 + asserted: &mut Option<Asserted>, 89 89 + ) -> Option<Verdict> { 90 90 + let known = match asserted.take().filter(|known| known.claim == claim) { 91 91 + Some(known) => known, 92 92 + None => Asserted { 93 93 + outcome: resolve_claim(state, &claim, peer).await, 94 94 + claim, 95 95 + }, 96 96 + }; 97 97 + let verdict = match &known.outcome { 98 98 + Claimed::Unreadable => None, 99 99 + Claimed::Publishes { did, keys } if keys.contains(key) => { 100 100 + Some(Verdict::Identified(did.clone())) 101 101 + } 102 102 + Claimed::Publishes { did, keys } => { 103 103 + tracing::debug!( 104 104 + ?peer, 105 105 + did = did.as_str(), 106 106 + published = keys.len(), 107 107 + "ssh auth refused a key the asserted account doesn't publish" 108 108 + ); 109 109 + Some(Verdict::Refused) 110 110 + } 111 111 + }; 112 112 + *asserted = Some(known); 113 113 + verdict 114 114 + } 115 115 + 116 116 + async fn resolve_claim<H: HttpTransport, C: Clock>( 117 117 + state: &Arc<SshState<H, C>>, 118 118 + claim: &OwnerRef, 119 119 + peer: Option<IpAddr>, 120 120 + ) -> Claimed { 121 121 + let Ok(_peer_guard) = state.lookup_peers.admit(peer, state.atproto.now()) else { 122 122 + tracing::debug!(?peer, "ssh auth couldn't check a claim, peer budget spent"); 123 123 + return Claimed::Unreadable; 124 124 + }; 125 125 + let did = match claim { 126 126 + OwnerRef::Did(did) => AccountDid::from(did.clone()), 127 127 + OwnerRef::Handle(handle) => match state.atproto.resolve_handle_to_did(handle).await { 128 128 + Ok(did) => did, 129 129 + Err(error) => { 130 130 + tracing::warn!( 131 131 + ?peer, 132 132 + handle = handle.as_str(), 133 133 + %error, 134 134 + "ssh auth couldn't resolve the handle in the login name" 135 135 + ); 136 136 + return Claimed::Unreadable; 137 137 + } 138 138 + }, 139 139 + }; 140 140 + match state.atproto.claimed_pubkeys(&did).await { 141 141 + ClaimedKeys::Published(keys) => Claimed::Publishes { did, keys }, 142 142 + ClaimedKeys::Unread(error) => { 143 143 + tracing::warn!( 144 144 + ?peer, 145 145 + did = did.as_str(), 146 146 + %error, 147 147 + "ssh auth couldn't read the asserted account's published keys" 148 148 + ); 149 149 + Claimed::Unreadable 150 150 + } 151 151 + } 152 152 + }

+43 -7

knot2/crates/knot-ssh/src/lib.rs

View file Reviewed

··· 1 1 mod exec; 2 2 - mod roster; 2 2 + mod identity; 3 3 mod server; 4 4 5 5 use std::borrow::Cow; ··· 12 12 use knot_atproto::Atproto; 13 13 use knot_events::EventLog; 14 14 use knot_git::{ArchiveLimit, Layout}; 15 15 - use knot_index::Index; 15 15 + use knot_index::{Index, KeyTtl}; 16 16 use knot_maintenance::MaintenanceHandle; 17 17 use knot_pack::{MaxWireBytes, PackLimits}; 18 18 use knot_postreceive::LanguagesPushBudget; ··· 26 26 use tokio_util::sync::CancellationToken; 27 27 use tokio_util::task::TaskTracker; 28 28 29 29 - use knot_resource::{LimitConfig, PerPeerInflight, PreAuthLimiter, Slots}; 30 30 - use roster::KeyRoster; 29 29 + use knot_resource::{ 30 30 + Burst, GlobalInflight, LimitConfig, PeerPacer, PerPeerInflight, PreAuthLimiter, RateLimit, 31 31 + RefillMicros, ResolveSlots, Slots, SubjectPacer, 32 32 + }; 31 33 use server::KnotSshServer; 32 34 33 35 const MAX_INFLIGHT_PER_PEER: usize = 4; 36 36 + const MAX_INFLIGHT_LOOKUPS: usize = 16; 37 37 + const MAX_PREAUTH_LOOKUPS: usize = 4; 38 38 + const LOOKUP_BURST_PER_PEER: u32 = 8; 39 39 + const LOOKUP_REFILL_MICROS: u64 = 500_000; 40 40 + const LOOKUP_INFLIGHT_PER_PEER: usize = 2; 41 41 + const PROBE_BURST_PER_ACCOUNT: u32 = 1; 42 42 + const PROBE_REFILL_MICROS: u64 = 30_000_000; 43 43 + const MISS_BURST_PER_PEER: u32 = 1; 44 44 + const MISS_REFILL_MICROS: u64 = 120_000_000; 34 45 const INACTIVITY_TIMEOUT: Duration = Duration::from_secs(120); 35 46 const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30); 36 47 const AUTH_REJECTION_TIME: Duration = Duration::from_millis(250); ··· 61 72 languages_push_budget: LanguagesPushBudget, 62 73 ci_logs: Option<CiLogsAddr>, 63 74 slots: Slots, 75 75 + lookup_slots: ResolveSlots, 76 76 + lookup_peers: Arc<PreAuthLimiter>, 77 77 + probe_pace: SubjectPacer, 78 78 + miss_pace: PeerPacer, 79 79 + key_ttl: KeyTtl, 64 80 peer_slots: Arc<PreAuthLimiter>, 65 65 - roster: Arc<KeyRoster>, 66 81 maintenance: MaintenanceHandle, 67 82 lfs: Option<LfsRuntime>, 68 83 catalog: Arc<knot_messages::Catalog>, ··· 124 139 languages_push_budget, 125 140 ci_logs, 126 141 slots: Slots::for_machine(), 142 142 + lookup_slots: ResolveSlots::new(MAX_INFLIGHT_LOOKUPS), 143 143 + lookup_peers: Arc::new(PreAuthLimiter::with_config(LimitConfig { 144 144 + rate: Some(RateLimit { 145 145 + burst: Burst::new(LOOKUP_BURST_PER_PEER), 146 146 + refill: RefillMicros::new(LOOKUP_REFILL_MICROS), 147 147 + }), 148 148 + per_peer_inflight: Some(PerPeerInflight::new(LOOKUP_INFLIGHT_PER_PEER)), 149 149 + global_inflight: Some(GlobalInflight::new(MAX_PREAUTH_LOOKUPS)), 150 150 + })), 151 151 + probe_pace: SubjectPacer::new(RateLimit { 152 152 + burst: Burst::new(PROBE_BURST_PER_ACCOUNT), 153 153 + refill: RefillMicros::new(PROBE_REFILL_MICROS), 154 154 + }), 155 155 + miss_pace: PeerPacer::new(RateLimit { 156 156 + burst: Burst::new(MISS_BURST_PER_PEER), 157 157 + refill: RefillMicros::new(MISS_REFILL_MICROS), 158 158 + }), 159 159 + key_ttl: KeyTtl::DEFAULT, 127 160 peer_slots: Arc::new(PreAuthLimiter::with_config(LimitConfig::per_peer_only( 128 161 PerPeerInflight::new(MAX_INFLIGHT_PER_PEER), 129 162 ))), 130 130 - roster: Arc::new(KeyRoster::new()), 131 163 maintenance: MaintenanceHandle::disabled(), 132 164 lfs: None, 133 165 catalog: Arc::new(knot_messages::Catalog::defaults()), ··· 141 173 142 174 pub fn with_slots(mut self, slots: Slots) -> Self { 143 175 self.slots = slots; 176 176 + self 177 177 + } 178 178 + 179 179 + pub fn with_key_ttl(mut self, ttl: KeyTtl) -> Self { 180 180 + self.key_ttl = ttl; 144 181 self 145 182 } 146 183 ··· 208 245 ) -> Result<(), SshError> { 209 246 let config = server_config(host_key); 210 247 let tracker = TaskTracker::new(); 211 211 - state.roster.prime(&state.index, &state.atproto); 212 248 let mut server = KnotSshServer { 213 249 state, 214 250 tracker: tracker.clone(),

-522

knot2/crates/knot-ssh/src/roster.rs

Reviewed

··· 1 1 - use std::collections::{HashMap, HashSet}; 2 2 - use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; 3 3 - use std::sync::{Arc, Mutex}; 4 4 - use std::time::Duration; 5 5 - 6 6 - use futures::StreamExt; 7 7 - use knot_atproto::Atproto; 8 8 - use knot_index::{Index, IndexGeneration, Resolved}; 9 9 - use knot_runtime::{Clock, HttpTransport, UnixMicros}; 10 10 - use knot_types::{AccountDid, OfferedKey}; 11 11 - 12 12 - const FRESH_TTL: Duration = Duration::from_secs(60); 13 13 - const DEGRADED_TTL: Duration = Duration::from_secs(5); 14 14 - const BACKOFF_SHIFT_LIMIT: u32 = 4; 15 15 - const MISS_REVALIDATE_BUDGET: Duration = Duration::from_secs(30); 16 16 - const RESOLVE_FANOUT: usize = 16; 17 17 - 18 18 - fn degraded_ttl(consecutive_failures: u32) -> Duration { 19 19 - let secs = DEGRADED_TTL 20 20 - .as_secs() 21 21 - .saturating_mul(1u64 << consecutive_failures.min(BACKOFF_SHIFT_LIMIT)) 22 22 - .min(FRESH_TTL.as_secs()); 23 23 - Duration::from_secs(secs) 24 24 - } 25 25 - 26 26 - struct Freshness { 27 27 - due: UnixMicros, 28 28 - generation: IndexGeneration, 29 29 - } 30 30 - 31 31 - #[derive(Debug, PartialEq, Eq)] 32 32 - enum Staleness { 33 33 - Fresh, 34 34 - Revalidate, 35 35 - Cold, 36 36 - } 37 37 - 38 38 - pub(crate) struct KeyRoster { 39 39 - by_did: Mutex<HashMap<AccountDid, HashSet<OfferedKey>>>, 40 40 - recognized: Mutex<HashSet<OfferedKey>>, 41 41 - freshness: Mutex<Option<Freshness>>, 42 42 - failures: AtomicU32, 43 43 - refresh: tokio::sync::Mutex<()>, 44 44 - refresh_in_flight: AtomicBool, 45 45 - } 46 46 - 47 47 - impl KeyRoster { 48 48 - pub(crate) fn new() -> Self { 49 49 - Self { 50 50 - by_did: Mutex::new(HashMap::new()), 51 51 - recognized: Mutex::new(HashSet::new()), 52 52 - freshness: Mutex::new(None), 53 53 - failures: AtomicU32::new(0), 54 54 - refresh: tokio::sync::Mutex::new(()), 55 55 - refresh_in_flight: AtomicBool::new(false), 56 56 - } 57 57 - } 58 58 - 59 59 - pub(crate) fn recognizes(&self, key: &OfferedKey) -> bool { 60 60 - self.recognized 61 61 - .lock() 62 62 - .unwrap_or_else(|poisoned| poisoned.into_inner()) 63 63 - .contains(key) 64 64 - } 65 65 - 66 66 - pub(crate) fn did_for(&self, key: &OfferedKey) -> Option<AccountDid> { 67 67 - self.by_did 68 68 - .lock() 69 69 - .unwrap_or_else(|poisoned| poisoned.into_inner()) 70 70 - .iter() 71 71 - .find(|(_, keys)| keys.contains(key)) 72 72 - .map(|(did, _)| did.clone()) 73 73 - } 74 74 - 75 75 - fn is_fresh(&self, now: UnixMicros, generation: IndexGeneration) -> bool { 76 76 - self.freshness 77 77 - .lock() 78 78 - .unwrap_or_else(|poisoned| poisoned.into_inner()) 79 79 - .as_ref() 80 80 - .is_some_and(|fresh| now.get() < fresh.due.get() && fresh.generation == generation) 81 81 - } 82 82 - 83 83 - pub(crate) fn prime<H: HttpTransport, C: Clock>( 84 84 - self: &Arc<Self>, 85 85 - index: &Arc<Index>, 86 86 - atproto: &Arc<Atproto<H, C>>, 87 87 - ) { 88 88 - self.spawn_refresh(index, atproto); 89 89 - } 90 90 - 91 91 - pub(crate) fn ensure_fresh<H: HttpTransport, C: Clock>( 92 92 - self: &Arc<Self>, 93 93 - index: &Arc<Index>, 94 94 - atproto: &Arc<Atproto<H, C>>, 95 95 - ) { 96 96 - match self.staleness(atproto.now(), index.generation()) { 97 97 - Staleness::Fresh => {} 98 98 - Staleness::Revalidate | Staleness::Cold => self.spawn_refresh(index, atproto), 99 99 - } 100 100 - } 101 101 - 102 102 - pub(crate) async fn recognizes_fresh<H: HttpTransport, C: Clock>( 103 103 - self: &Arc<Self>, 104 104 - key: &OfferedKey, 105 105 - index: &Arc<Index>, 106 106 - atproto: &Arc<Atproto<H, C>>, 107 107 - ) -> bool { 108 108 - if self.recognizes(key) { 109 109 - self.ensure_fresh(index, atproto); 110 110 - return true; 111 111 - } 112 112 - if self.is_fresh(atproto.now(), index.generation()) { 113 113 - return false; 114 114 - } 115 115 - let _ = tokio::time::timeout(MISS_REVALIDATE_BUDGET, self.refresh(index, atproto)).await; 116 116 - self.recognizes(key) 117 117 - } 118 118 - 119 119 - fn staleness(&self, now: UnixMicros, generation: IndexGeneration) -> Staleness { 120 120 - match self 121 121 - .freshness 122 122 - .lock() 123 123 - .unwrap_or_else(|poisoned| poisoned.into_inner()) 124 124 - .as_ref() 125 125 - { 126 126 - None => Staleness::Cold, 127 127 - Some(fresh) if fresh.generation != generation => Staleness::Revalidate, 128 128 - Some(fresh) if now.get() < fresh.due.get() => Staleness::Fresh, 129 129 - Some(_) => Staleness::Revalidate, 130 130 - } 131 131 - } 132 132 - 133 133 - fn spawn_refresh<H: HttpTransport, C: Clock>( 134 134 - self: &Arc<Self>, 135 135 - index: &Arc<Index>, 136 136 - atproto: &Arc<Atproto<H, C>>, 137 137 - ) { 138 138 - if self 139 139 - .refresh_in_flight 140 140 - .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) 141 141 - .is_err() 142 142 - { 143 143 - return; 144 144 - } 145 145 - let roster = Arc::clone(self); 146 146 - let index = Arc::clone(index); 147 147 - let atproto = Arc::clone(atproto); 148 148 - tokio::spawn(async move { 149 149 - let _in_flight = InFlightGuard(&roster.refresh_in_flight); 150 150 - roster.refresh(&index, &atproto).await; 151 151 - }); 152 152 - } 153 153 - 154 154 - async fn refresh<H: HttpTransport, C: Clock>(&self, index: &Index, atproto: &Atproto<H, C>) { 155 155 - let _single_flight = self.refresh.lock().await; 156 156 - if self.is_fresh(atproto.now(), index.generation()) { 157 157 - return; 158 158 - } 159 159 - let generation = index.generation(); 160 160 - let (dids, incomplete) = relevant_dids(index); 161 161 - let resolved: Vec<(AccountDid, Option<Vec<OfferedKey>>)> = futures::stream::iter(dids) 162 162 - .map(|did| async move { 163 163 - let keys = atproto.resolve_pubkeys(&did).await.ok(); 164 164 - (did, keys) 165 165 - }) 166 166 - .buffer_unordered(RESOLVE_FANOUT) 167 167 - .collect() 168 168 - .await; 169 169 - let any_failed = resolved.iter().any(|(_, keys)| keys.is_none()); 170 170 - let relevant: HashSet<AccountDid> = resolved.iter().map(|(did, _)| did.clone()).collect(); 171 171 - { 172 172 - let mut by_did = self 173 173 - .by_did 174 174 - .lock() 175 175 - .unwrap_or_else(|poisoned| poisoned.into_inner()); 176 176 - by_did.retain(|did, _| relevant.contains(did)); 177 177 - resolved.into_iter().for_each(|(did, keys)| { 178 178 - if let Some(keys) = keys { 179 179 - by_did.insert(did, keys.into_iter().collect()); 180 180 - } 181 181 - }); 182 182 - let union: HashSet<OfferedKey> = by_did.values().flatten().cloned().collect(); 183 183 - *self 184 184 - .recognized 185 185 - .lock() 186 186 - .unwrap_or_else(|poisoned| poisoned.into_inner()) = union; 187 187 - } 188 188 - let ttl = if any_failed { 189 189 - degraded_ttl(self.failures.fetch_add(1, Ordering::Relaxed)) 190 190 - } else { 191 191 - self.failures.store(0, Ordering::Relaxed); 192 192 - if incomplete { DEGRADED_TTL } else { FRESH_TTL } 193 193 - }; 194 194 - let due = UnixMicros::new(atproto.now().get().saturating_add(ttl.as_micros() as u64)); 195 195 - *self 196 196 - .freshness 197 197 - .lock() 198 198 - .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(Freshness { due, generation }); 199 199 - } 200 200 - } 201 201 - 202 202 - struct InFlightGuard<'a>(&'a AtomicBool); 203 203 - 204 204 - impl Drop for InFlightGuard<'_> { 205 205 - fn drop(&mut self) { 206 206 - self.0.store(false, Ordering::Release); 207 207 - } 208 208 - } 209 209 - 210 210 - fn relevant_dids(index: &Index) -> (Vec<AccountDid>, bool) { 211 211 - let (mut dids, incomplete): (Vec<AccountDid>, bool) = index 212 212 - .hosted_repos() 213 213 - .iter() 214 214 - .map(|repo| { 215 215 - let (owner, owner_warming) = match index.owner_of(repo) { 216 216 - Resolved::Ready(Some(owner)) => (Some(AccountDid::from(owner)), false), 217 217 - Resolved::Ready(None) => (None, false), 218 218 - Resolved::Warming => (None, true), 219 219 - }; 220 220 - let (collaborators, collaborators_warming) = match index.collaborators_of(repo) { 221 221 - Resolved::Ready(collaborators) => (collaborators, false), 222 222 - Resolved::Warming => (Vec::new(), true), 223 223 - }; 224 224 - ( 225 225 - owner.into_iter().chain(collaborators).collect::<Vec<_>>(), 226 226 - owner_warming || collaborators_warming, 227 227 - ) 228 228 - }) 229 229 - .fold( 230 230 - (Vec::new(), false), 231 231 - |(mut acc, warming), (dids, repo_warming)| { 232 232 - acc.extend(dids); 233 233 - (acc, warming || repo_warming) 234 234 - }, 235 235 - ); 236 236 - dids.sort(); 237 237 - dids.dedup(); 238 238 - (dids, incomplete) 239 239 - } 240 240 - 241 241 - #[cfg(test)] 242 242 - mod tests { 243 243 - use super::*; 244 244 - use std::sync::Arc; 245 245 - use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; 246 246 - 247 247 - use knot_atproto::Atproto; 248 248 - use knot_cob::{CobHome, CobStore}; 249 249 - use knot_cobs::{Registration, RegistryChange}; 250 250 - use knot_git::{Layout, Repo}; 251 251 - use knot_runtime::{ 252 252 - FakeHttp, HttpRequest, HttpResponse, K256Signer, NetworkError, SeededEntropy, Signer, 253 253 - }; 254 254 - use knot_types::{KnotId, OwnerDid, RepoDid, RepoName, RepoRkey, UnixSeconds, crypto}; 255 255 - use russh::keys::{Algorithm, PrivateKey}; 256 256 - use url::Url; 257 257 - 258 258 - struct SharedClock(Arc<AtomicU64>); 259 259 - impl Clock for SharedClock { 260 260 - fn now_unix_micros(&self) -> UnixMicros { 261 261 - UnixMicros::new(self.0.load(Ordering::SeqCst)) 262 262 - } 263 263 - } 264 264 - 265 265 - fn line_and_offered() -> (String, OfferedKey) { 266 266 - let key = PrivateKey::random(&mut crate::EntropyRng, Algorithm::Ed25519).unwrap(); 267 267 - let public = key.public_key(); 268 268 - ( 269 269 - public.to_openssh().unwrap(), 270 270 - OfferedKey::from_bytes(public.to_bytes().unwrap()), 271 271 - ) 272 272 - } 273 273 - 274 274 - type Responder = Box<dyn Fn(&HttpRequest) -> Result<HttpResponse, NetworkError> + Send + Sync>; 275 275 - 276 276 - struct Harness { 277 277 - index: Arc<Index>, 278 278 - atproto: Arc<Atproto<FakeHttp<Responder>, SharedClock>>, 279 279 - published: Arc<std::sync::Mutex<Vec<String>>>, 280 280 - list_calls: Arc<AtomicUsize>, 281 281 - _dir: tempfile::TempDir, 282 282 - } 283 283 - 284 284 - fn harness(initial: Vec<String>) -> Harness { 285 285 - let dir = tempfile::tempdir().unwrap(); 286 286 - let meta_path = dir.path().join("meta"); 287 287 - Repo::create(&meta_path).unwrap(); 288 288 - let layout = Layout::new(dir.path().join("repos")); 289 289 - let repo_did = RepoDid::new("did:plc:squid").unwrap(); 290 290 - layout.create(&repo_did).unwrap(); 291 291 - let cob_signer = K256Signer::generate(&SeededEntropy::new(2)); 292 292 - { 293 293 - let meta = Repo::open(&meta_path).unwrap(); 294 294 - CobStore::new(&meta) 295 295 - .create( 296 296 - &CobHome::from(&KnotId::new("did:web:nel.pet").unwrap()), 297 297 - &RegistryChange::Register(Registration { 298 298 - owner: OwnerDid::new("did:plc:nel").unwrap(), 299 299 - rkey: RepoRkey::new("anemone").unwrap(), 300 300 - name: RepoName::new("anemone").unwrap(), 301 301 - repo: repo_did.clone(), 302 302 - created_at: UnixSeconds::new(1), 303 303 - }), 304 304 - &cob_signer, 305 305 - UnixSeconds::new(1), 306 306 - ) 307 307 - .unwrap(); 308 308 - } 309 309 - let index = Arc::new(Index::new(meta_path, layout.clone())); 310 310 - index.rebuild().unwrap(); 311 311 - 312 312 - let published = Arc::new(std::sync::Mutex::new(initial)); 313 313 - let list_calls = Arc::new(AtomicUsize::new(0)); 314 314 - let multikey = crypto::multikey( 315 315 - 0xe7, 316 316 - K256Signer::generate(&SeededEntropy::new(7)) 317 317 - .public_key() 318 318 - .as_bytes(), 319 319 - ); 320 320 - let clock = Arc::new(AtomicU64::new(1_000_000_000)); 321 321 - 322 322 - let responder: Responder = { 323 323 - let published = Arc::clone(&published); 324 324 - let list_calls = Arc::clone(&list_calls); 325 325 - Box::new(move |request: &HttpRequest| { 326 326 - let host = request.url.host_str().unwrap_or_default().to_string(); 327 327 - let body = if host == "pds.oyster.cafe" { 328 328 - list_calls.fetch_add(1, Ordering::SeqCst); 329 329 - let records: Vec<_> = published 330 330 - .lock() 331 331 - .unwrap() 332 332 - .iter() 333 333 - .map(|line| { 334 334 - serde_json::json!({ 335 335 - "uri": "at://did:plc:nel/sh.tangled.publicKey/1", 336 336 - "value": { 337 337 - "$type": "sh.tangled.publicKey", 338 338 - "key": line, 339 339 - "name": "laptop", 340 340 - "createdAt": "2026-06-08T00:00:00Z" 341 341 - } 342 342 - }) 343 343 - }) 344 344 - .collect(); 345 345 - serde_json::to_vec(&serde_json::json!({ "records": records })).unwrap() 346 346 - } else if host == "plc.directory" { 347 347 - serde_json::to_vec(&serde_json::json!({ 348 348 - "id": "did:plc:nel", 349 349 - "alsoKnownAs": ["at://nel.pet"], 350 350 - "verificationMethod": [{ 351 351 - "id": "did:plc:nel#atproto", 352 352 - "type": "Multikey", 353 353 - "controller": "did:plc:nel", 354 354 - "publicKeyMultibase": multikey 355 355 - }], 356 356 - "service": [{ 357 357 - "id": "#atproto_pds", 358 358 - "type": "AtprotoPersonalDataServer", 359 359 - "serviceEndpoint": "https://pds.oyster.cafe" 360 360 - }] 361 361 - })) 362 362 - .unwrap() 363 363 - } else { 364 364 - return Ok(HttpResponse { 365 365 - status: http::StatusCode::NOT_FOUND, 366 366 - headers: http::HeaderMap::new(), 367 367 - body: bytes::Bytes::new(), 368 368 - }); 369 369 - }; 370 370 - Ok(HttpResponse { 371 371 - status: http::StatusCode::OK, 372 372 - headers: http::HeaderMap::new(), 373 373 - body: bytes::Bytes::from(body), 374 374 - }) 375 375 - }) 376 376 - }; 377 377 - 378 378 - let atproto = Arc::new(Atproto::new( 379 379 - FakeHttp::new(responder), 380 380 - SharedClock(clock), 381 381 - KnotId::new("did:web:nel.pet").unwrap(), 382 382 - knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(), 383 383 - )); 384 384 - 385 385 - Harness { 386 386 - index, 387 387 - atproto, 388 388 - published, 389 389 - list_calls, 390 390 - _dir: dir, 391 391 - } 392 392 - } 393 393 - 394 394 - async fn wait_recognized(roster: &KeyRoster, key: &OfferedKey) { 395 395 - for _ in 0..1000 { 396 396 - if roster.recognizes(key) { 397 397 - return; 398 398 - } 399 399 - tokio::task::yield_now().await; 400 400 - } 401 401 - } 402 402 - 403 403 - #[tokio::test] 404 404 - async fn an_acl_write_makes_a_freshly_published_key_recognized_without_waiting_for_the_ttl() { 405 405 - let (line1, offered1) = line_and_offered(); 406 406 - let (line2, offered2) = line_and_offered(); 407 407 - 408 408 - let Harness { 409 409 - index, 410 410 - atproto, 411 411 - published, 412 412 - list_calls, 413 413 - _dir, 414 414 - } = harness(vec![line1]); 415 415 - 416 416 - let roster = Arc::new(KeyRoster::new()); 417 417 - roster.ensure_fresh(&index, &atproto); 418 418 - wait_recognized(&roster, &offered1).await; 419 419 - assert!(roster.recognizes(&offered1)); 420 420 - assert_eq!(list_calls.load(Ordering::SeqCst), 1); 421 421 - 422 422 - published.lock().unwrap().push(line2.clone()); 423 423 - 424 424 - roster.ensure_fresh(&index, &atproto); 425 425 - assert!( 426 426 - !roster.recognizes(&offered2), 427 427 - "stable index and unexpired TTL still serves cached roster, no re-resolution" 428 428 - ); 429 429 - assert_eq!(list_calls.load(Ordering::SeqCst), 1); 430 430 - 431 431 - index.refresh_members().unwrap(); 432 432 - roster.ensure_fresh(&index, &atproto); 433 433 - wait_recognized(&roster, &offered2).await; 434 434 - assert!( 435 435 - roster.recognizes(&offered2), 436 436 - "ACL write bumps generation, so roster revalidates off the auth path" 437 437 - ); 438 438 - assert_eq!( 439 439 - list_calls.load(Ordering::SeqCst), 440 440 - 2, 441 441 - "exactly one async re-resolution off the auth path" 442 442 - ); 443 443 - } 444 444 - 445 445 - #[tokio::test] 446 446 - async fn a_miss_against_a_stale_roster_blocks_bounded_to_revalidate_before_rejecting() { 447 447 - let (line1, offered1) = line_and_offered(); 448 448 - let (line2, offered2) = line_and_offered(); 449 449 - let Harness { 450 450 - index, 451 451 - atproto, 452 452 - published, 453 453 - list_calls, 454 454 - _dir, 455 455 - } = harness(vec![line1]); 456 456 - let roster = Arc::new(KeyRoster::new()); 457 457 - 458 458 - assert!( 459 459 - roster.recognizes_fresh(&offered1, &index, &atproto).await, 460 460 - "the first handshake blocks on the primed resolve and recognizes the published key" 461 461 - ); 462 462 - assert_eq!(list_calls.load(Ordering::SeqCst), 1); 463 463 - 464 464 - published.lock().unwrap().push(line2.clone()); 465 465 - index.refresh_members().unwrap(); 466 466 - 467 467 - assert!( 468 468 - roster.recognizes_fresh(&offered2, &index, &atproto).await, 469 469 - "a generation-bumped miss blocks to revalidate and picks up the new key on the first attempt" 470 470 - ); 471 471 - assert_eq!( 472 472 - list_calls.load(Ordering::SeqCst), 473 473 - 2, 474 474 - "the miss triggers exactly one bounded re-resolution" 475 475 - ); 476 476 - } 477 477 - 478 478 - #[test] 479 479 - fn staleness_classifies_cold_fresh_and_revalidate() { 480 480 - let roster = KeyRoster::new(); 481 481 - assert_eq!( 482 482 - roster.staleness(UnixMicros::new(0), IndexGeneration::new(0)), 483 483 - Staleness::Cold, 484 484 - "with no roster yet the first auth is cold and must revalidate before it can answer a miss" 485 485 - ); 486 486 - *roster 487 487 - .freshness 488 488 - .lock() 489 489 - .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(Freshness { 490 490 - due: UnixMicros::new(1_000), 491 491 - generation: IndexGeneration::new(0), 492 492 - }); 493 493 - assert_eq!( 494 494 - roster.staleness(UnixMicros::new(500), IndexGeneration::new(0)), 495 495 - Staleness::Fresh 496 496 - ); 497 497 - assert_eq!( 498 498 - roster.staleness(UnixMicros::new(500), IndexGeneration::new(1)), 499 499 - Staleness::Revalidate, 500 500 - "an ACL write moves the generation, so the cached roster is stale" 501 501 - ); 502 502 - assert_eq!( 503 503 - roster.staleness(UnixMicros::new(2_000), IndexGeneration::new(0)), 504 504 - Staleness::Revalidate, 505 505 - "an expired ttl at the same generation is stale too" 506 506 - ); 507 507 - } 508 508 - 509 509 - #[test] 510 510 - fn degraded_ttl_backs_off_from_the_short_retry_to_the_fresh_ceiling() { 511 511 - assert_eq!(degraded_ttl(0), Duration::from_secs(5)); 512 512 - assert_eq!(degraded_ttl(1), Duration::from_secs(10)); 513 513 - assert_eq!(degraded_ttl(2), Duration::from_secs(20)); 514 514 - assert_eq!(degraded_ttl(3), Duration::from_secs(40)); 515 515 - assert_eq!(degraded_ttl(4), Duration::from_secs(60)); 516 516 - assert_eq!( 517 517 - degraded_ttl(50), 518 518 - Duration::from_secs(60), 519 519 - "a persistently unresolvable did clamps the retry to the fresh ttl instead of storming" 520 520 - ); 521 521 - } 522 522 - }

+62 -28

knot2/crates/knot-ssh/src/server.rs

View file Reviewed

··· 3 3 use std::sync::Arc; 4 4 5 5 use knot_runtime::{Clock, HttpTransport}; 6 6 - use knot_types::OfferedKey; 6 6 + use knot_types::{OfferedKey, OwnerRef}; 7 7 use russh::keys::ssh_key; 8 8 use russh::server::{Auth, Handler, Msg, Server, Session}; 9 9 use russh::{Channel, ChannelId}; ··· 11 11 12 12 use crate::SshState; 13 13 use crate::exec::run_exec; 14 14 + use crate::identity::{self, Asserted, Credential, Verdict}; 14 15 15 16 pub(crate) struct KnotSshServer<H, C> { 16 17 pub(crate) state: Arc<SshState<H, C>>, ··· 32 33 pub(crate) struct KnotSession<H, C> { 33 34 state: Arc<SshState<H, C>>, 34 35 tracker: TaskTracker, 35 35 - key: Option<OfferedKey>, 36 36 + credential: Option<Credential>, 37 37 + asserted: Option<Asserted>, 36 38 channels: HashMap<ChannelId, Channel<Msg>>, 37 39 protocols: HashSet<ChannelId>, 38 40 peer: Option<IpAddr>, 39 41 } 40 42 43 43 + fn reject() -> Auth { 44 44 + Auth::Reject { 45 45 + proceed_with_methods: None, 46 46 + partial_success: false, 47 47 + } 48 48 + } 49 49 + 41 50 impl<H, C> KnotSession<H, C> { 42 51 fn new(state: Arc<SshState<H, C>>, tracker: TaskTracker, peer: Option<IpAddr>) -> Self { 43 52 Self { 44 53 state, 45 54 tracker, 46 46 - key: None, 55 55 + credential: None, 56 56 + asserted: None, 47 57 channels: HashMap::new(), 48 58 protocols: HashSet::new(), 49 59 peer, ··· 51 61 } 52 62 } 53 63 64 64 + impl<H: HttpTransport, C: Clock> KnotSession<H, C> { 65 65 + async fn decide( 66 66 + &mut self, 67 67 + user: &str, 68 68 + public_key: &ssh_key::PublicKey, 69 69 + ) -> Option<(Verdict, OfferedKey)> { 70 70 + let key = OfferedKey::from_bytes(public_key.to_bytes().ok()?); 71 71 + let verdict = identity::verify( 72 72 + &self.state, 73 73 + OwnerRef::parse(user), 74 74 + &key, 75 75 + self.peer, 76 76 + &mut self.asserted, 77 77 + ) 78 78 + .await; 79 79 + Some((verdict, key)) 80 80 + } 81 81 + } 82 82 + 54 83 impl<H: HttpTransport, C: Clock> Handler for KnotSession<H, C> { 55 84 type Error = russh::Error; 56 85 86 86 + async fn auth_publickey_offered( 87 87 + &mut self, 88 88 + user: &str, 89 89 + public_key: &ssh_key::PublicKey, 90 90 + ) -> Result<Auth, Self::Error> { 91 91 + match self.decide(user, public_key).await { 92 92 + Some((Verdict::Refused, _)) | None => Ok(reject()), 93 93 + Some(_) => Ok(Auth::Accept), 94 94 + } 95 95 + } 96 96 + 57 97 async fn auth_publickey( 58 98 &mut self, 59 59 - _user: &str, 99 99 + user: &str, 60 100 public_key: &ssh_key::PublicKey, 61 101 ) -> Result<Auth, Self::Error> { 62 62 - let reject = Auth::Reject { 63 63 - proceed_with_methods: None, 64 64 - partial_success: false, 65 65 - }; 66 66 - let Ok(blob) = public_key.to_bytes() else { 67 67 - return Ok(reject); 68 68 - }; 69 69 - let key = OfferedKey::from_bytes(blob); 70 70 - if self 71 71 - .state 72 72 - .roster 73 73 - .recognizes_fresh(&key, &self.state.index, &self.state.atproto) 74 74 - .await 75 75 - { 76 76 - self.key = Some(key); 77 77 - Ok(Auth::Accept) 78 78 - } else { 79 79 - Ok(reject) 102 102 + match self.decide(user, public_key).await { 103 103 + Some((Verdict::Identified(did), _)) => { 104 104 + self.credential = Some(Credential::Identified(did)); 105 105 + Ok(Auth::Accept) 106 106 + } 107 107 + Some((Verdict::Offered, key)) => { 108 108 + self.credential = Some(Credential::Offered(key)); 109 109 + Ok(Auth::Accept) 110 110 + } 111 111 + Some((Verdict::Refused, _)) | None => Ok(reject()), 80 112 } 81 113 } 82 114 ··· 146 178 channel: ChannelId, 147 179 session: &mut Session, 148 180 ) -> Result<(), Self::Error> { 149 149 - let Some(handle) = self.channels.remove(&channel) else { 181 181 + let (Some(handle), Some(credential)) = 182 182 + (self.channels.remove(&channel), self.credential.clone()) 183 183 + else { 150 184 session.channel_failure(channel)?; 151 185 return Ok(()); 152 186 }; 153 187 session.channel_success(channel)?; 154 188 let state = Arc::clone(&self.state); 155 155 - let key = self.key.clone(); 156 189 self.tracker.spawn(async move { 157 157 - crate::exec::run_greeting(state, key, handle).await; 190 190 + crate::exec::run_greeting(state, credential, handle).await; 158 191 }); 159 192 Ok(()) 160 193 } ··· 165 198 data: &[u8], 166 199 session: &mut Session, 167 200 ) -> Result<(), Self::Error> { 168 168 - let Some(handle) = self.channels.remove(&channel) else { 201 201 + let (Some(handle), Some(credential)) = 202 202 + (self.channels.remove(&channel), self.credential.clone()) 203 203 + else { 169 204 session.channel_failure(channel)?; 170 205 return Ok(()); 171 206 }; 172 207 session.channel_success(channel)?; 173 208 let protocol_v2 = self.protocols.remove(&channel); 174 209 let state = Arc::clone(&self.state); 175 175 - let key = self.key.clone(); 176 210 let peer = self.peer; 177 211 let command = data.to_vec(); 178 212 self.tracker.spawn(async move { 179 179 - run_exec(state, key, handle, &command, protocol_v2, peer).await; 213 213 + run_exec(state, credential, handle, &command, protocol_v2, peer).await; 180 214 }); 181 215 Ok(()) 182 216 }

+391 -25

knot2/crates/knot-ssh/tests/ssh_push.rs

View file Reviewed

··· 1 1 - use std::collections::HashMap; 1 1 + use std::collections::{HashMap, HashSet}; 2 2 use std::path::{Path, PathBuf}; 3 3 use std::process::Command; 4 4 - use std::sync::Arc; 4 4 + use std::sync::atomic::{AtomicUsize, Ordering}; 5 5 + use std::sync::{Arc, Mutex}; 5 6 6 7 use futures::stream::StreamExt; 7 8 use knot_atproto::Atproto; 8 9 use knot_cob::{CobHome, CobStore}; 9 10 use knot_cobs::{CollaboratorsChange, Grant, MembersChange, Registration, RegistryChange}; 10 11 use knot_git::{ArchiveLimit, Layout, Repo}; 11 11 - use knot_index::Index; 12 12 + use knot_index::{Index, Resolved}; 12 13 use knot_pack::MaxWireBytes; 13 14 use knot_postreceive::LanguagesPushBudget; 14 15 use knot_runtime::{ ··· 133 134 } 134 135 } 135 136 137 137 + fn forever() -> knot_index::KeyLease { 138 138 + knot_index::KeyTtl::from_secs(u32::MAX.into()).lease_from(UnixSeconds::new(0)) 139 139 + } 140 140 + 141 141 + fn server_error() -> HttpResponse { 142 142 + HttpResponse { 143 143 + status: http::StatusCode::INTERNAL_SERVER_ERROR, 144 144 + headers: http::HeaderMap::new(), 145 145 + body: bytes::Bytes::new(), 146 146 + } 147 147 + } 148 148 + 136 149 fn fake_http(published_line: String) -> impl knot_runtime::HttpTransport { 137 150 let signer = K256Signer::generate(&SeededEntropy::new(1)); 138 151 let pds = format!("https://{PDS_HOST}"); ··· 154 167 }) 155 168 } 156 169 157 157 - fn multi_http(identities: HashMap<String, Vec<String>>) -> impl knot_runtime::HttpTransport { 170 170 + #[derive(Default, Clone)] 171 171 + struct Accounts { 172 172 + identities: Arc<Mutex<HashMap<String, Vec<String>>>>, 173 173 + unreachable: Arc<Mutex<HashSet<String>>>, 174 174 + listings: Arc<AtomicUsize>, 175 175 + } 176 176 + 177 177 + impl Accounts { 178 178 + fn publishing(identities: HashMap<String, Vec<String>>) -> Self { 179 179 + Self { 180 180 + identities: Arc::new(Mutex::new(identities)), 181 181 + ..Self::default() 182 182 + } 183 183 + } 184 184 + 185 185 + fn unreachable(self, dids: HashSet<String>) -> Self { 186 186 + *self.unreachable.lock().unwrap() = dids; 187 187 + self 188 188 + } 189 189 + 190 190 + fn restore(&self, did: &str) { 191 191 + self.unreachable.lock().unwrap().remove(did); 192 192 + } 193 193 + 194 194 + fn publish(&self, did: &str, line: String) { 195 195 + self.identities 196 196 + .lock() 197 197 + .unwrap() 198 198 + .entry(did.to_string()) 199 199 + .or_default() 200 200 + .push(line); 201 201 + } 202 202 + 203 203 + fn published_by(&self, did: &str) -> Vec<String> { 204 204 + self.identities 205 205 + .lock() 206 206 + .unwrap() 207 207 + .get(did) 208 208 + .cloned() 209 209 + .unwrap_or_default() 210 210 + } 211 211 + 212 212 + fn listings(&self) -> usize { 213 213 + self.listings.load(Ordering::SeqCst) 214 214 + } 215 215 + } 216 216 + 217 217 + fn multi_http(accounts: Accounts) -> impl knot_runtime::HttpTransport { 158 218 let signer = K256Signer::generate(&SeededEntropy::new(77)); 159 219 FakeHttp::new(move |request| { 160 220 let host = request.url.host_str().unwrap_or_default().to_string(); ··· 169 229 .find(|(key, _)| key == "repo") 170 230 .map(|(_, value)| value.into_owned()) 171 231 .unwrap_or_default(); 172 172 - let lines = identities.get(&repo).cloned().unwrap_or_default(); 232 232 + accounts.listings.fetch_add(1, Ordering::SeqCst); 233 233 + if accounts.unreachable.lock().unwrap().contains(&repo) { 234 234 + return Ok(server_error()); 235 235 + } 236 236 + let lines = accounts.published_by(&repo); 173 237 let refs: Vec<&str> = lines.iter().map(String::as_str).collect(); 174 238 return Ok(ok_body(list_records_body(&refs))); 175 239 } ··· 996 1060 } 997 1061 998 1062 #[tokio::test(flavor = "multi_thread", worker_threads = 4)] 1063 1063 + async fn a_filled_key_set_refuses_an_unregistered_key_and_an_acl_write_reopens_the_check() { 1064 1064 + let fx = fixture().await; 1065 1065 + let head = seed_work(&fx.work); 1066 1066 + let head_oid = Oid::from_hex(&head).unwrap(); 1067 1067 + 1068 1068 + fx.index.keys().mark_ready(fx.index.generation()); 1069 1069 + fx.index.refresh_members().unwrap(); 1070 1070 + let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await; 1071 1071 + assert!( 1072 1072 + ok, 1073 1073 + "a grant written after the key set was read must reopen the check, or whoever it \ 1074 1074 + grants is refused at the handshake until the next fill pass:\n{out}" 1075 1075 + ); 1076 1076 + assert_eq!( 1077 1077 + main_tip(&fx.server.layout, &fx.server.repo_did), 1078 1078 + Some(head_oid) 1079 1079 + ); 1080 1080 + 1081 1081 + fx.index.keys().record( 1082 1082 + &AccountDid::new(OWNER_DID).unwrap(), 1083 1083 + vec![knot_types::OfferedKey::from_bytes(registered_blob(&fx))], 1084 1084 + forever(), 1085 1085 + ); 1086 1086 + fx.index.keys().mark_ready(fx.index.generation()); 1087 1087 + 1088 1088 + let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered"); 1089 1089 + let two_ids = format!( 1090 1090 + "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \ 1091 1091 + -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes", 1092 1092 + fx.key_path 1093 1093 + ); 1094 1094 + let (ok, out) = { 1095 1095 + let (work, url) = (fx.work.clone(), fx.url.clone()); 1096 1096 + tokio::task::spawn_blocking(move || { 1097 1097 + git( 1098 1098 + &work, 1099 1099 + &[("GIT_SSH_COMMAND", &two_ids)], 1100 1100 + &["push", "-q", &url, "main:refs/heads/second"], 1101 1101 + ) 1102 1102 + }) 1103 1103 + .await 1104 1104 + .unwrap() 1105 1105 + }; 1106 1106 + assert!( 1107 1107 + ok, 1108 1108 + "a filled key set refuses the unregistered key, so the client offers its registered key \ 1109 1109 + without the url identifying anybody:\n{out}" 1110 1110 + ); 1111 1111 + assert_eq!( 1112 1112 + fx.server 1113 1113 + .layout 1114 1114 + .open(&fx.server.repo_did) 1115 1115 + .unwrap() 1116 1116 + .find_ref(&RefName::new("refs/heads/second").unwrap()) 1117 1117 + .unwrap(), 1118 1118 + Some(head_oid) 1119 1119 + ); 1120 1120 + } 1121 1121 + 1122 1122 + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] 999 1123 async fn key_recognition_edge_cases() { 1000 1124 let fx = fixture().await; 1001 1125 let head = seed_work(&fx.work); ··· 1020 1144 .unwrap() 1021 1145 }; 1022 1146 assert!( 1023 1023 - ok, 1024 1024 - "rejecting unregistered key must let client cycle to the registered one:\n{out}" 1147 1147 + !ok, 1148 1148 + "with the key set still filling, the push is checked against whichever key the client \ 1149 1149 + offers first:\n{out}" 1025 1150 ); 1151 1151 + assert!( 1152 1152 + out.contains("@nel.pet"), 1153 1153 + "refusal lists who may push, so the pusher knows which key to offer:\n{out}" 1154 1154 + ); 1155 1155 + assert!( 1156 1156 + out.contains("IdentitiesOnly"), 1157 1157 + "refusal states how a multi-key client can offer its registered key:\n{out}" 1158 1158 + ); 1159 1159 + assert_eq!( 1160 1160 + main_tip(&fx.server.layout, &fx.server.repo_did), 1161 1161 + None, 1162 1162 + "the refused push leaves the repo empty" 1163 1163 + ); 1164 1164 + 1165 1165 + let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await; 1166 1166 + assert!(ok, "offering only the registered key must succeed:\n{out}"); 1026 1167 assert_eq!( 1027 1168 main_tip(&fx.server.layout, &fx.server.repo_did), 1028 1169 Some(head_oid) ··· 1035 1176 .to_bytes() 1036 1177 .unwrap(); 1037 1178 fx.index.keys().record( 1038 1038 - &AccountDid::new("did:plc:whelk").unwrap(), 1179 1179 + &AccountDid::new("did:plc:cuttle").unwrap(), 1039 1180 vec![knot_types::OfferedKey::from_bytes(blob)], 1040 1040 - knot_index::KeyTtl::from_secs(u32::MAX.into()).lease_from(knot_types::UnixSeconds::new(0)), 1181 1181 + forever(), 1041 1182 ); 1042 1183 let (ok, out) = push( 1043 1184 &fx.work, ··· 1081 1222 ); 1082 1223 } 1083 1224 1084 1084 - async fn launch( 1085 1085 - host_key_dir: &Path, 1086 1086 - layout: Layout, 1087 1087 - index: Arc<Index>, 1088 1088 - identities: HashMap<String, Vec<String>>, 1089 1089 - ) -> u16 { 1225 1225 + async fn launch(host_key_dir: &Path, layout: Layout, index: Arc<Index>, accounts: Accounts) -> u16 { 1090 1226 let atproto = Arc::new(Atproto::new( 1091 1091 - multi_http(identities), 1227 1227 + multi_http(accounts), 1092 1228 ManualClock::new(UnixMicros::new(1_000_000_000)), 1093 1229 KnotId::new("did:web:nel.pet").unwrap(), 1094 1230 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(), ··· 1126 1262 } 1127 1263 1128 1264 #[tokio::test(flavor = "multi_thread", worker_threads = 4)] 1129 1129 - async fn a_collaborator_pushes_its_repo_but_a_recognized_key_is_denied_on_a_repo_it_has_no_grant_on() 1130 1130 - { 1265 1265 + async fn a_handle_in_the_url_identifies_a_visitor_and_lets_a_multi_key_client_find_its_key() { 1266 1266 + let fx = fixture().await; 1267 1267 + let head = seed_work(&fx.work); 1268 1268 + let head_oid = Oid::from_hex(&head).unwrap(); 1269 1269 + 1270 1270 + let port = fx.server.port; 1271 1271 + let greeted_key = fx.key_path.clone(); 1272 1272 + let (_ok, out) = 1273 1273 + tokio::task::spawn_blocking(move || ssh_bare_as(&greeted_key, "nel.pet", port)) 1274 1274 + .await 1275 1275 + .unwrap(); 1276 1276 + assert!( 1277 1277 + out.contains("@nel.pet"), 1278 1278 + "an asserted handle identifies the visitor on first contact, with an empty cache:\n{out}" 1279 1279 + ); 1280 1280 + 1281 1281 + let (unregistered_path, _unregistered_line) = keygen(fx.scratch.path(), "unregistered"); 1282 1282 + let two_ids = format!( 1283 1283 + "ssh -i {unregistered_path} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \ 1284 1284 + -o UserKnownHostsFile=/dev/null -o PreferredAuthentications=publickey -o BatchMode=yes", 1285 1285 + fx.key_path 1286 1286 + ); 1287 1287 + let identified = format!("ssh://nel.pet@127.0.0.1:{}/{REPO_DID}", fx.server.port); 1288 1288 + let (ok, out) = { 1289 1289 + let (work, url) = (fx.work.clone(), identified.clone()); 1290 1290 + tokio::task::spawn_blocking(move || { 1291 1291 + git( 1292 1292 + &work, 1293 1293 + &[("GIT_SSH_COMMAND", &two_ids)], 1294 1294 + &["push", "-q", &url, "main"], 1295 1295 + ) 1296 1296 + }) 1297 1297 + .await 1298 1298 + .unwrap() 1299 1299 + }; 1300 1300 + assert!( 1301 1301 + ok, 1302 1302 + "a handle in the url lets the knot refuse the unregistered key so the client offers the \ 1303 1303 + next key:\n{out}" 1304 1304 + ); 1305 1305 + assert_eq!( 1306 1306 + main_tip(&fx.server.layout, &fx.server.repo_did), 1307 1307 + Some(head_oid) 1308 1308 + ); 1309 1309 + assert_eq!( 1310 1310 + fx.index.owner_of_key( 1311 1311 + &knot_types::OfferedKey::from_bytes(registered_blob(&fx)), 1312 1312 + UnixSeconds::new(0), 1313 1313 + ), 1314 1314 + Resolved::Ready(None), 1315 1315 + "an asserted handle is whatever the client typed, so the keys read for it mustn't enter \ 1316 1316 + the set, or anyone can fill the key budget by asserting handles" 1317 1317 + ); 1318 1318 + } 1319 1319 + 1320 1320 + fn registered_blob(fx: &Fixture) -> Vec<u8> { 1321 1321 + russh::keys::ssh_key::PublicKey::from_openssh( 1322 1322 + &std::fs::read_to_string(fx.scratch.path().join("client.pub")).unwrap(), 1323 1323 + ) 1324 1324 + .unwrap() 1325 1325 + .to_bytes() 1326 1326 + .unwrap() 1327 1327 + } 1328 1328 + 1329 1329 + fn registered_index( 1330 1330 + scratch: &TempDir, 1331 1331 + budget: knot_index::KeyBudget, 1332 1332 + ) -> (Layout, RepoDid, Arc<Index>) { 1333 1333 + let meta_path = scratch.path().join("meta"); 1334 1334 + Repo::create(&meta_path).unwrap(); 1335 1335 + let layout = Layout::new(scratch.path().join("repos")); 1336 1336 + let repo_did = RepoDid::new(REPO_DID).unwrap(); 1337 1337 + layout.create(&repo_did).unwrap(); 1338 1338 + 1339 1339 + let signer = K256Signer::generate(&SeededEntropy::new(2)); 1340 1340 + let meta = Repo::open(&meta_path).unwrap(); 1341 1341 + CobStore::new(&meta) 1342 1342 + .create( 1343 1343 + &CobHome::from(&KnotId::new("did:web:nel.pet").unwrap()), 1344 1344 + &RegistryChange::Register(Registration { 1345 1345 + owner: OwnerDid::new(OWNER_DID).unwrap(), 1346 1346 + rkey: RepoRkey::new(REPO_NAME).unwrap(), 1347 1347 + name: RepoName::new(REPO_NAME).unwrap(), 1348 1348 + repo: repo_did.clone(), 1349 1349 + created_at: UnixSeconds::new(1), 1350 1350 + }), 1351 1351 + &signer, 1352 1352 + UnixSeconds::new(1), 1353 1353 + ) 1354 1354 + .unwrap(); 1355 1355 + 1356 1356 + let index = Arc::new(Index::with_key_budget(meta_path, layout.clone(), budget)); 1357 1357 + index.rebuild().unwrap(); 1358 1358 + index.warm_collaborators(); 1359 1359 + (layout, repo_did, index) 1360 1360 + } 1361 1361 + 1362 1362 + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] 1363 1363 + async fn an_unreachable_pds_reads_as_transient_and_its_keys_get_in_once_it_recovers() { 1364 1364 + let scratch = tempfile::tempdir().unwrap(); 1365 1365 + let (stale_key, stale_line) = keygen(scratch.path(), "stale"); 1366 1366 + let (fresh_key, fresh_line) = keygen(scratch.path(), "fresh"); 1367 1367 + let (layout, repo_did, index) = registered_index(&scratch, knot_index::KeyBudget::DEFAULT); 1368 1368 + 1369 1369 + let accounts = Accounts::publishing(HashMap::from([(OWNER_DID.to_string(), vec![stale_line])])) 1370 1370 + .unreachable(HashSet::from([OWNER_DID.to_string()])); 1371 1371 + let port = launch( 1372 1372 + &scratch.path().join("hostkey"), 1373 1373 + layout.clone(), 1374 1374 + Arc::clone(&index), 1375 1375 + accounts.clone(), 1376 1376 + ) 1377 1377 + .await; 1378 1378 + let url = format!("ssh://git@127.0.0.1:{port}/{REPO_DID}"); 1379 1379 + 1380 1380 + let work = scratch.path().join("work"); 1381 1381 + let head = seed_work(&work); 1382 1382 + let (ok, out) = push(&work, &url, &stale_key, &["main"]).await; 1383 1383 + assert!( 1384 1384 + !ok, 1385 1385 + "a push mustn't be accepted while the owner's records are unreadable:\n{out}" 1386 1386 + ); 1387 1387 + assert!( 1388 1388 + out.contains("retry shortly"), 1389 1389 + "an unreadable PDS must read as transient:\n{out}" 1390 1390 + ); 1391 1391 + assert!( 1392 1392 + !out.contains("doesn't match"), 1393 1393 + "a transient failure mustn't be reported to the pusher as a wrong key:\n{out}" 1394 1394 + ); 1395 1395 + assert_eq!( 1396 1396 + main_tip(&layout, &repo_did), 1397 1397 + None, 1398 1398 + "the refused push leaves the repo empty" 1399 1399 + ); 1400 1400 + 1401 1401 + accounts.restore(OWNER_DID); 1402 1402 + let (ok, out) = push(&work, &url, &stale_key, &["main"]).await; 1403 1403 + assert!( 1404 1404 + ok, 1405 1405 + "the key the owner publishes must push once its PDS answers again:\n{out}" 1406 1406 + ); 1407 1407 + 1408 1408 + accounts.publish(OWNER_DID, fresh_line); 1409 1409 + let (ok, out) = push(&work, &url, &fresh_key, &["main", "--force"]).await; 1410 1410 + assert!( 1411 1411 + ok, 1412 1412 + "a key the owner published after the knot last read the account must get in on the next \ 1413 1413 + push, or publishing a second key locks its owner out until a fill pass catches up:\n{out}" 1414 1414 + ); 1415 1415 + assert_eq!( 1416 1416 + main_tip(&layout, &repo_did), 1417 1417 + Some(Oid::from_hex(&head).unwrap()) 1418 1418 + ); 1419 1419 + } 1420 1420 + 1421 1421 + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] 1422 1422 + async fn an_account_the_budget_couldnt_fit_still_clears_the_handshake_and_pushes() { 1423 1423 + let scratch = tempfile::tempdir().unwrap(); 1424 1424 + let (owner_key, owner_line) = keygen(scratch.path(), "owner"); 1425 1425 + let (layout, repo_did, index) = 1426 1426 + registered_index(&scratch, knot_index::KeyBudget::from_bytes(200)); 1427 1427 + 1428 1428 + let blob = russh::keys::ssh_key::PublicKey::from_openssh(&owner_line) 1429 1429 + .unwrap() 1430 1430 + .to_bytes() 1431 1431 + .unwrap(); 1432 1432 + assert_eq!( 1433 1433 + index.keys().record( 1434 1434 + &AccountDid::new(OWNER_DID).unwrap(), 1435 1435 + vec![knot_types::OfferedKey::from_bytes(blob)], 1436 1436 + forever(), 1437 1437 + ), 1438 1438 + knot_index::KeyRecord::Unheld, 1439 1439 + "a 200-byte budget records the read without keeping the key" 1440 1440 + ); 1441 1441 + index.keys().mark_ready(index.generation()); 1442 1442 + 1443 1443 + let port = launch( 1444 1444 + &scratch.path().join("hostkey"), 1445 1445 + layout.clone(), 1446 1446 + Arc::clone(&index), 1447 1447 + Accounts::publishing(HashMap::from([(OWNER_DID.to_string(), vec![owner_line])])), 1448 1448 + ) 1449 1449 + .await; 1450 1450 + let url = format!("ssh://git@127.0.0.1:{port}/{REPO_DID}"); 1451 1451 + 1452 1452 + let work = scratch.path().join("work"); 1453 1453 + let head = seed_work(&work); 1454 1454 + let (ok, out) = push(&work, &url, &owner_key, &["main"]).await; 1455 1455 + assert!( 1456 1456 + ok, 1457 1457 + "the set can't fit the owner's keys, so the handshake must defer to the push check \ 1458 1458 + instead of refusing a key the accounts on file don't publish:\n{out}" 1459 1459 + ); 1460 1460 + assert_eq!( 1461 1461 + main_tip(&layout, &repo_did), 1462 1462 + Some(Oid::from_hex(&head).unwrap()) 1463 1463 + ); 1464 1464 + } 1465 1465 + 1466 1466 + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] 1467 1467 + async fn a_collaborator_pushes_its_repo_but_is_denied_on_a_repo_it_doesnt_collaborate_on() { 1131 1468 const REPO_A: &str = "did:plc:squid"; 1132 1469 const REPO_B: &str = "did:plc:clam"; 1133 1470 const OWNER: &str = "did:plc:nel"; ··· 1211 1548 (OWNER.to_string(), vec![owner_line]), 1212 1549 (COLLAB.to_string(), vec![collab_line]), 1213 1550 ]); 1551 1551 + let accounts = Accounts::publishing(identities); 1214 1552 let port = launch( 1215 1553 &scratch.path().join("hostkey"), 1216 1554 layout.clone(), 1217 1555 Arc::clone(&index), 1218 1218 - identities, 1556 1556 + accounts.clone(), 1219 1557 ) 1220 1558 .await; 1221 1559 ··· 1239 1577 let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await; 1240 1578 assert!( 1241 1579 !denied, 1242 1242 - "key recognized via repo A but with no grant on repo B must be denied, recognition is \ 1243 1243 - not authorization:\n{out}" 1580 1580 + "a key that pushes repo A must be denied on repo B, where its owner was never granted:\n{out}" 1244 1581 ); 1245 1582 assert!( 1246 1583 main_tip(&layout, &repo_b).is_none(), 1247 1584 "denied cross-repo push must land nothing on repo B" 1585 1585 + ); 1586 1586 + 1587 1587 + let after_first_denial = accounts.listings(); 1588 1588 + let (denied, out) = push(&work_b, &url_b, &collab_key, &["main"]).await; 1589 1589 + assert!(!denied, "the second attempt is denied the same way:\n{out}"); 1590 1590 + assert_eq!( 1591 1591 + accounts.listings(), 1592 1592 + after_first_denial, 1593 1593 + "repo B's owner was read during the first denial and is on file, so retrying mustn't \ 1594 1594 + read that PDS again, or anyone with a key can make the knot fetch from a third party \ 1595 1595 + at will:\n{out}" 1248 1596 ); 1249 1597 1250 1598 let work_owner = scratch.path().join("work_owner_b"); ··· 1259 1607 } 1260 1608 1261 1609 fn ssh_bare(key_path: &str, port: u16) -> (bool, String) { 1610 1610 + ssh_bare_as(key_path, "git", port) 1611 1611 + } 1612 1612 + 1613 1613 + fn ssh_bare_as(key_path: &str, user: &str, port: u16) -> (bool, String) { 1262 1614 let out = Command::new("ssh") 1263 1615 .args([ 1264 1616 "-i", ··· 1275 1627 "BatchMode=yes", 1276 1628 "-p", 1277 1629 &port.to_string(), 1278 1278 - "git@127.0.0.1", 1630 1630 + &format!("{user}@127.0.0.1"), 1279 1631 ]) 1280 1632 .output() 1281 1633 .expect("ssh runs"); ··· 1290 1642 } 1291 1643 1292 1644 #[tokio::test(flavor = "multi_thread", worker_threads = 4)] 1293 1293 - async fn a_bare_ssh_session_greets_the_recognized_user() { 1645 1645 + async fn a_bare_ssh_session_greets_a_visitor_then_identifies_them_once_they_have_pushed() { 1294 1646 let fx = fixture().await; 1295 1647 let port = fx.server.port; 1648 1648 + 1649 1649 + let key_path = fx.key_path.clone(); 1650 1650 + let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port)) 1651 1651 + .await 1652 1652 + .unwrap(); 1653 1653 + assert!(out.contains("knot.test"), "greeting names the knot:\n{out}"); 1654 1654 + assert!( 1655 1655 + out.contains("ssh key"), 1656 1656 + "a visitor the knot can't identify yet learns what a push needs:\n{out}" 1657 1657 + ); 1658 1658 + 1659 1659 + seed_work(&fx.work); 1660 1660 + let (ok, out) = push(&fx.work, &fx.url, &fx.key_path, &["main"]).await; 1661 1661 + assert!(ok, "seeding main must succeed:\n{out}"); 1662 1662 + 1296 1663 let key_path = fx.key_path.clone(); 1297 1664 let (_ok, out) = tokio::task::spawn_blocking(move || ssh_bare(&key_path, port)) 1298 1665 .await 1299 1666 .unwrap(); 1300 1667 assert!( 1301 1668 out.contains("@nel.pet"), 1302 1302 - "greeting resolves and addresses the user by handle:\n{out}" 1669 1669 + "a push teaches the knot the key, so the next greeting uses the handle:\n{out}" 1303 1670 ); 1304 1304 - assert!(out.contains("knot.test"), "greeting names the knot:\n{out}"); 1305 1671 } 1306 1672 1307 1673 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]

+8 -2

knot2/example.toml

View file Reviewed

··· 516 516 # 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}:<repoDID>"] 517 517 #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}:<repoDID>"] 518 518 519 519 + # 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}:<repoDID>", "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}:<repoDID>, so your ssh client can find your registered key on its own."] 520 520 + #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}:<repoDID>", "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}:<repoDID>, so your ssh client can find your registered key on its own."] 521 521 + 519 522 # Default value: "knot: unsupported command" 520 523 #unsupported_command = "knot: unsupported command" 521 524 ··· 531 534 # Default value: "knot: LFS isn't enabled on this knot" 532 535 #lfs_disabled = "knot: LFS isn't enabled on this knot" 533 536 534 534 - # 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." 535 535 - #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." 537 537 + # 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." 538 538 + #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." 539 539 + 540 540 + # Default value: "knot: couldn't read the account records needed to check your ssh key, retry shortly" 541 541 + #identity_unavailable = "knot: couldn't read the account records needed to check your ssh key, retry shortly" 536 542 537 543 # Default value: "knot: you aren't authorized to push to this repository." 538 544 #push_denied = "knot: you aren't authorized to push to this repository."