alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::net::IpAddr;
2use std::path::PathBuf;
3use std::sync::Arc;
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::time::Duration;
6
7use futures::StreamExt;
8use knot_acl::{KnotAcl, can_push};
9use knot_index::Resolved;
10use knot_lfs::TransferOp;
11use knot_pack::{PackError, PackLimits, RepoLookup};
12use knot_resource::SubjectKey;
13use knot_runtime::{Clock, HttpTransport};
14use knot_types::{AccountDid, ClonePath, ObjectFormat, OwnerDid, RepoDid};
15use russh::Channel;
16use russh::server::Msg;
17use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
18use tokio::runtime::Handle;
19use tokio::sync::mpsc;
20
21use crate::SshState;
22use crate::identity::Credential;
23
24const READ_CHUNK: usize = 64 * 1024;
25const MAX_UPLOAD_REQUEST: usize = 16 * 1024 * 1024;
26const RECEIVE_BODY_DEADLINE: Duration = Duration::from_secs(1800);
27const ARCHIVE_REQUEST_DEADLINE: Duration = Duration::from_secs(60);
28const LFS_PROGRESS_GRACE: Duration = Duration::from_secs(60);
29const LFS_PROGRESS_FLOOR_BYTES_PER_SEC: u64 = 1024;
30const LFS_STALL_TIMEOUT: Duration = Duration::from_secs(120);
31const CANDIDATE_FANOUT: usize = 4;
32const AUTHORIZED_NAMES_SHOWN: usize = 4;
33
34fn lfs_within_progress_budget(waited: Duration, moved_bytes: u64) -> bool {
35 waited
36 <= LFS_PROGRESS_GRACE + Duration::from_secs(moved_bytes / LFS_PROGRESS_FLOOR_BYTES_PER_SEC)
37}
38
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40enum Service {
41 Upload,
42 UploadArchive,
43 Receive,
44 Lfs(TransferOp),
45}
46
47enum ReadError {
48 Io(std::io::Error),
49 Pack(PackError),
50 TooLarge,
51 Truncated,
52 Deadline,
53}
54
55enum RepoRef {
56 Did(RepoDid),
57 OwnerPath(OwnerDid, ClonePath),
58 HandlePath(knot_types::Handle, ClonePath),
59}
60
61enum ResolvedRef {
62 Did(RepoDid),
63 OwnerPath(OwnerDid, ClonePath),
64}
65
66fn parse_exec(command: &[u8]) -> Option<(Service, RepoRef)> {
67 let text = std::str::from_utf8(command).ok()?.trim();
68 if let Some(rest) = text.strip_prefix("git-lfs-transfer ") {
69 let (path, op_token) = rest.trim().rsplit_once(' ')?;
70 let op = TransferOp::parse(op_token.trim())?;
71 return Some((Service::Lfs(op), parse_repo_path(path)?));
72 }
73 let (service, rest) = [
74 ("git-upload-pack ", Service::Upload),
75 ("git upload-pack ", Service::Upload),
76 ("git-upload-archive ", Service::UploadArchive),
77 ("git upload-archive ", Service::UploadArchive),
78 ("git-receive-pack ", Service::Receive),
79 ("git receive-pack ", Service::Receive),
80 ]
81 .into_iter()
82 .find_map(|(prefix, service)| text.strip_prefix(prefix).map(|rest| (service, rest)))?;
83 Some((service, parse_repo_path(rest)?))
84}
85
86fn parse_repo_path(raw: &str) -> Option<RepoRef> {
87 let path = raw
88 .trim()
89 .trim_matches('\'')
90 .trim_matches('"')
91 .trim_start_matches('/');
92 match path.split_once('/') {
93 Some((owner, name)) => {
94 let candidates = ClonePath::parse(name)?;
95 match knot_types::OwnerRef::parse(owner)? {
96 knot_types::OwnerRef::Did(owner) => Some(RepoRef::OwnerPath(owner, candidates)),
97 knot_types::OwnerRef::Handle(handle) => {
98 Some(RepoRef::HandlePath(handle, candidates))
99 }
100 }
101 }
102 None => Some(RepoRef::Did(RepoDid::new(path).ok()?)),
103 }
104}
105
106fn resolve_repo_ref<H: HttpTransport, C: Clock>(
107 state: &Arc<SshState<H, C>>,
108 repo_ref: ResolvedRef,
109) -> RepoLookup {
110 let candidate = match repo_ref {
111 ResolvedRef::Did(did) => RepoLookup::Hosted(did),
112 ResolvedRef::OwnerPath(owner, candidates) => RepoLookup::from_resolved(
113 state.index.resolve_clone_path(&owner, &candidates),
114 |found| found,
115 ),
116 };
117 match candidate {
118 RepoLookup::Hosted(did) => {
119 RepoLookup::from_resolved(state.index.owner_of(&did), |_| did.clone())
120 }
121 undecided => undecided,
122 }
123}
124
125pub(crate) async fn run_exec<H: HttpTransport, C: Clock>(
126 state: Arc<SshState<H, C>>,
127 credential: Credential,
128 channel: Channel<Msg>,
129 command: &[u8],
130 protocol_v2: bool,
131 peer: Option<IpAddr>,
132) {
133 let Some((service, repo_ref)) = parse_exec(command) else {
134 fail(channel, &state.catalog.ssh.unsupported_command.text()).await;
135 return;
136 };
137 let peer_limiter = match &service {
138 Service::Lfs(_) => state
139 .lfs
140 .as_ref()
141 .map_or(&state.peer_slots, |lfs| &lfs.peer_slots),
142 _ => &state.peer_slots,
143 };
144 let _peer_guard = match peer_limiter.admit(peer, state.atproto.now()) {
145 Ok(guard) => guard,
146 Err(refusal) => {
147 let reason = match refusal {
148 knot_resource::Refusal::RateLimited => "peer request rate exceeded",
149 knot_resource::Refusal::Saturated => "peer concurrency limit reached",
150 };
151 tracing::warn!(?peer, reason, "ssh exec rejected");
152 return fail(channel, &state.catalog.ssh.too_many_operations.text()).await;
153 }
154 };
155 let resolved_ref = match repo_ref {
156 RepoRef::Did(did) => ResolvedRef::Did(did),
157 RepoRef::OwnerPath(owner, candidates) => ResolvedRef::OwnerPath(owner, candidates),
158 RepoRef::HandlePath(owner_handle, candidates) => {
159 let Some(_lookup_permit) = state.lookup_slots.try_acquire() else {
160 tracing::warn!(
161 ?peer,
162 "ssh exec rejected, the lookup budget can't resolve another handle"
163 );
164 return fail(channel, &state.catalog.ssh.too_many_operations.text()).await;
165 };
166 match state
167 .atproto
168 .resolve_handle_to_did(&owner_handle)
169 .await
170 .ok()
171 {
172 Some(did) => ResolvedRef::OwnerPath(did.into(), candidates),
173 None => {
174 fail(channel, &state.catalog.ssh.repo_not_found.text()).await;
175 return;
176 }
177 }
178 }
179 };
180 let repo_did = match resolve_repo_ref(&state, resolved_ref) {
181 RepoLookup::Hosted(did) => did,
182 RepoLookup::Unhosted => {
183 fail(channel, &state.catalog.ssh.repo_not_found.text()).await;
184 return;
185 }
186 RepoLookup::Unavailable => {
187 fail(channel, &state.catalog.ssh.index_warming.text()).await;
188 return;
189 }
190 };
191 let layout = state.layout.clone();
192 let did = repo_did.clone();
193 let opened = tokio::task::spawn_blocking(move || layout.open(&did).is_ok())
194 .await
195 .unwrap_or(false);
196 if !opened {
197 fail(channel, &state.catalog.ssh.repo_not_found.text()).await;
198 return;
199 }
200 match service {
201 Service::Upload => serve_upload(state, channel, repo_did, protocol_v2).await,
202 Service::UploadArchive => serve_upload_archive(state, channel, repo_did).await,
203 Service::Receive => serve_receive(state, credential, channel, repo_did, peer).await,
204 Service::Lfs(op) => serve_lfs(state, credential, channel, repo_did, op, peer).await,
205 }
206}
207
208async fn serve_lfs<H: HttpTransport, C: Clock>(
209 state: Arc<SshState<H, C>>,
210 credential: Credential,
211 mut channel: Channel<Msg>,
212 repo_did: RepoDid,
213 op: TransferOp,
214 peer: Option<IpAddr>,
215) {
216 let Some(lfs) = state.lfs.clone() else {
217 return fail(channel, &state.catalog.ssh.lfs_disabled.text()).await;
218 };
219 if op == TransferOp::Upload
220 && let PushAuth::Refused { reason, message } =
221 authorize_push(&state, &credential, &repo_did, peer).await
222 {
223 tracing::warn!(repo = repo_did.as_str(), reason, "ssh lfs upload denied");
224 return fail(channel, &message).await;
225 }
226 let permit = match Arc::clone(&lfs.slots).acquire_owned().await {
227 Ok(permit) => permit,
228 Err(_) => return fail(channel, &state.catalog.ssh.shutting_down.text()).await,
229 };
230 let started = std::time::Instant::now();
231
232 let (tx, rx) = mpsc::channel::<Vec<u8>>(8);
233 let writer = Box::pin(channel.make_writer());
234 let handle = lfs.handle.clone();
235 let runtime = Handle::current();
236 let did = repo_did.clone();
237 let catalog = Arc::clone(&state.catalog);
238 let mut engine = tokio::task::spawn_blocking(move || {
239 let _permit = permit;
240 let output = std::io::BufWriter::new(MeteredWrite::new(runtime.clone(), writer));
241 knot_lfs::serve_transfer(
242 handle.store.as_ref(),
243 handle.admission.as_ref(),
244 &did,
245 op,
246 &catalog.lfs,
247 MpscRead::new(runtime, rx),
248 output,
249 )
250 });
251 let joined = {
252 let reader = channel.make_reader();
253 tokio::select! {
254 joined = &mut engine => joined,
255 () = pump_input(reader, tx) => engine.await,
256 }
257 };
258 let status = match joined {
259 Ok(Ok(())) => {
260 tracing::info!(
261 repo = repo_did.as_str(),
262 op = match op {
263 TransferOp::Upload => "upload",
264 TransferOp::Download => "download",
265 },
266 duration_ms = started.elapsed().as_millis() as u64,
267 "ssh lfs transfer finished"
268 );
269 0
270 }
271 Ok(Err(fault)) => {
272 tracing::warn!(repo = repo_did.as_str(), %fault, "ssh lfs transfer failed");
273 1
274 }
275 Err(join) => {
276 tracing::error!(repo = repo_did.as_str(), %join, "ssh lfs transfer task panicked");
277 1
278 }
279 };
280 finish(channel, status).await;
281}
282
283async fn pump_input<R: AsyncRead + Unpin>(reader: R, tx: mpsc::Sender<Vec<u8>>) {
284 use futures::TryStreamExt;
285 let _ = tokio_util::io::ReaderStream::with_capacity(reader, READ_CHUNK)
286 .map_err(|_| ())
287 .try_for_each(|chunk| {
288 let tx = &tx;
289 async move {
290 match chunk.is_empty() {
291 true => Ok(()),
292 false => tx.send(chunk.to_vec()).await.map_err(|_| ()),
293 }
294 }
295 })
296 .await;
297}
298
299fn stalled(direction: &'static str) -> std::io::Error {
300 std::io::Error::other(format!("lfs {direction} stalled past the idle timeout"))
301}
302
303struct MpscRead {
304 runtime: Handle,
305 rx: mpsc::Receiver<Vec<u8>>,
306 buffer: Vec<u8>,
307 offset: usize,
308 waited: Duration,
309 received: u64,
310}
311
312impl MpscRead {
313 fn new(runtime: Handle, rx: mpsc::Receiver<Vec<u8>>) -> Self {
314 Self {
315 runtime,
316 rx,
317 buffer: Vec::new(),
318 offset: 0,
319 waited: Duration::ZERO,
320 received: 0,
321 }
322 }
323}
324
325impl std::io::Read for MpscRead {
326 fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
327 if self.offset >= self.buffer.len() {
328 let started = std::time::Instant::now();
329 let rx = &mut self.rx;
330 let received = self
331 .runtime
332 // I know I know, but these aren't runtime workers here
333 .block_on(async { tokio::time::timeout(LFS_STALL_TIMEOUT, rx.recv()).await });
334 match received {
335 Ok(Some(chunk)) => {
336 self.waited += started.elapsed();
337 self.received += chunk.len() as u64;
338 if !lfs_within_progress_budget(self.waited, self.received) {
339 return Err(std::io::Error::other(
340 "lfs input trickles below the progress floor",
341 ));
342 }
343 self.buffer = chunk;
344 self.offset = 0;
345 }
346 Ok(None) => return Ok(0),
347 Err(_) => return Err(stalled("input")),
348 }
349 }
350 let take = out.len().min(self.buffer.len() - self.offset);
351 out[..take].copy_from_slice(&self.buffer[self.offset..self.offset + take]);
352 self.offset += take;
353 Ok(take)
354 }
355}
356
357struct MeteredWrite<W> {
358 runtime: Handle,
359 inner: W,
360 waited: Duration,
361 written: u64,
362}
363
364impl<W: AsyncWrite + Unpin> MeteredWrite<W> {
365 fn new(runtime: Handle, inner: W) -> Self {
366 Self {
367 runtime,
368 inner,
369 waited: Duration::ZERO,
370 written: 0,
371 }
372 }
373
374 fn charge(&mut self, started: std::time::Instant) -> std::io::Result<()> {
375 self.waited += started.elapsed();
376 match lfs_within_progress_budget(self.waited, self.written) {
377 true => Ok(()),
378 false => Err(std::io::Error::other(
379 "lfs output trickles below the progress floor",
380 )),
381 }
382 }
383}
384
385impl<W: AsyncWrite + Unpin> std::io::Write for MeteredWrite<W> {
386 fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
387 let started = std::time::Instant::now();
388 let inner = &mut self.inner;
389 let wrote = self
390 .runtime
391 .block_on(async { tokio::time::timeout(LFS_STALL_TIMEOUT, inner.write(buf)).await })
392 .map_err(|_| stalled("output"))??;
393 self.written += wrote as u64;
394 self.charge(started).map(|()| wrote)
395 }
396
397 fn flush(&mut self) -> std::io::Result<()> {
398 let started = std::time::Instant::now();
399 let inner = &mut self.inner;
400 self.runtime
401 .block_on(async { tokio::time::timeout(LFS_STALL_TIMEOUT, inner.flush()).await })
402 .map_err(|_| stalled("output"))??;
403 self.charge(started)
404 }
405}
406
407async fn serve_upload_archive<H: HttpTransport, C: Clock>(
408 state: Arc<SshState<H, C>>,
409 mut channel: Channel<Msg>,
410 repo_did: RepoDid,
411) {
412 let request = {
413 let mut reader = channel.make_reader();
414 tokio::time::timeout(ARCHIVE_REQUEST_DEADLINE, read_archive_request(&mut reader)).await
415 };
416 let request = match request {
417 Ok(Ok(request)) => request,
418 Ok(Err(())) => return fail(channel, &state.catalog.ssh.archive_malformed.text()).await,
419 Err(_) => return fail(channel, &state.catalog.ssh.archive_timeout.text()).await,
420 };
421
422 let permit = state.slots.pack.acquire().await;
423 let (tx, mut rx) = mpsc::channel::<Vec<u8>>(16);
424 let layout = state.layout.clone();
425 let did = repo_did.clone();
426 let archive_limit = state.archive_limit;
427 let handle = tokio::task::spawn_blocking(move || -> Result<(), PackError> {
428 let _permit = permit;
429 let repo = layout.open(&did)?;
430 let mut sink = |chunk: &[u8]| -> std::io::Result<()> {
431 tx.blocking_send(chunk.to_vec())
432 .map_err(|_| std::io::Error::other("client disconnected"))
433 };
434 knot_pack::upload_archive_streamed(&repo, &request, archive_limit, &mut sink)
435 });
436
437 let mut writer = channel.make_writer();
438 let mut forward = Ok(());
439 while let Some(chunk) = rx.recv().await {
440 if writer.write_all(&chunk).await.is_err() {
441 forward = Err(());
442 break;
443 }
444 }
445 drop(rx);
446 let produced = handle.await;
447 match &produced {
448 Ok(Err(error)) => {
449 tracing::warn!(repo = repo_did.as_str(), %error, "upload-archive failed")
450 }
451 Err(join) => {
452 tracing::error!(repo = repo_did.as_str(), %join, "upload-archive task panicked")
453 }
454 Ok(Ok(())) => {}
455 }
456 match (forward, produced) {
457 (Ok(()), Ok(Ok(()))) if writer.flush().await.is_ok() => finish(channel, 0).await,
458 _ => fail(channel, &state.catalog.ssh.archive_failed.text()).await,
459 }
460}
461
462async fn read_archive_request<R: AsyncRead + Unpin>(reader: &mut R) -> Result<Vec<u8>, ()> {
463 let mut buf = Vec::new();
464 loop {
465 if knot_pack::archive_request_complete(&buf).is_some() {
466 return Ok(buf);
467 }
468 match read_chunk(reader, &mut buf, MAX_UPLOAD_REQUEST).await {
469 Ok(true) => {}
470 Ok(false) | Err(_) => return Err(()),
471 }
472 }
473}
474
475async fn serve_upload<H: HttpTransport, C: Clock>(
476 state: Arc<SshState<H, C>>,
477 mut channel: Channel<Msg>,
478 repo_did: RepoDid,
479 protocol_v2: bool,
480) {
481 let advert = {
482 let layout = state.layout.clone();
483 let did = repo_did.clone();
484 tokio::task::spawn_blocking(move || -> Result<Vec<u8>, PackError> {
485 let repo = layout.open(&did)?;
486 if protocol_v2 {
487 knot_pack::advertise_upload_ssh(&repo)
488 } else {
489 knot_pack::advertise_upload_v0_ssh(&repo)
490 }
491 })
492 .await
493 };
494 let advert = match advert {
495 Ok(Ok(bytes)) => bytes,
496 _ => return fail(channel, &state.catalog.ssh.advertise_failed.text()).await,
497 };
498
499 let mut writer = channel.make_writer();
500 if writer.write_all(&advert).await.is_err() || writer.flush().await.is_err() {
501 return;
502 }
503
504 let outcome = {
505 let mut reader = channel.make_reader();
506 if protocol_v2 {
507 upload_loop_v2(&state, &repo_did, &mut reader, &mut writer).await
508 } else {
509 upload_loop_v0(&state, &repo_did, &mut reader, &mut writer).await
510 }
511 };
512 let status = match outcome {
513 Ok(()) => 0,
514 Err(()) => 1,
515 };
516 finish(channel, status).await;
517}
518
519async fn upload_loop_v2<H, C, R, W>(
520 state: &Arc<SshState<H, C>>,
521 repo_did: &RepoDid,
522 reader: &mut R,
523 writer: &mut W,
524) -> Result<(), ()>
525where
526 H: HttpTransport,
527 C: Clock,
528 R: AsyncRead + Unpin,
529 W: AsyncWriteExt + Unpin,
530{
531 let mut buf = Vec::new();
532 let mut framer = knot_pack::UploadFramer::new();
533 loop {
534 if let Some(len) = framer.advance(&buf) {
535 let request: Vec<u8> = buf.drain(..len).collect();
536 stream_upload(state, repo_did, request, writer).await?;
537 framer = knot_pack::UploadFramer::new();
538 continue;
539 }
540 match read_chunk(reader, &mut buf, MAX_UPLOAD_REQUEST).await {
541 Ok(true) => {}
542 Ok(false) => return Ok(()),
543 Err(_) => return Err(()),
544 }
545 }
546}
547
548async fn upload_loop_v0<H, C, R, W>(
549 state: &Arc<SshState<H, C>>,
550 repo_did: &RepoDid,
551 reader: &mut R,
552 writer: &mut W,
553) -> Result<(), ()>
554where
555 H: HttpTransport,
556 C: Clock,
557 R: AsyncRead + Unpin,
558 W: AsyncWriteExt + Unpin,
559{
560 let mut buf = Vec::new();
561 let mut framer = knot_pack::UploadFramer::new();
562 let mut naks_sent = 0usize;
563 loop {
564 if let Some(len) = framer.advance(&buf) {
565 let request: Vec<u8> = buf.drain(..len).collect();
566 return stream_upload(state, repo_did, request, writer).await;
567 }
568 let needed = framer.unanswered_flushes();
569 if naks_sent < needed {
570 let nak = knot_pack::upload_v0_nak();
571 if writer.write_all(&nak).await.is_err() || writer.flush().await.is_err() {
572 return Err(());
573 }
574 naks_sent += 1;
575 continue;
576 }
577 match read_chunk(reader, &mut buf, MAX_UPLOAD_REQUEST).await {
578 Ok(true) => {}
579 Ok(false) => return Ok(()),
580 Err(_) => return Err(()),
581 }
582 }
583}
584
585async fn stream_upload<H, C, W>(
586 state: &Arc<SshState<H, C>>,
587 repo_did: &RepoDid,
588 request: Vec<u8>,
589 writer: &mut W,
590) -> Result<(), ()>
591where
592 H: HttpTransport,
593 C: Clock,
594 W: AsyncWriteExt + Unpin,
595{
596 let permit = state.slots.pack.acquire().await;
597 let (tx, mut rx) = mpsc::channel::<Vec<u8>>(16);
598 let layout = state.layout.clone();
599 let did = repo_did.clone();
600 let catalog = Arc::clone(&state.catalog);
601 let knot = state.hostname.clone();
602 let handle = tokio::task::spawn_blocking(move || -> Result<(), PackError> {
603 let _permit = permit;
604 let repo = layout.open(&did)?;
605 let mut sink = |chunk: &[u8]| -> std::io::Result<()> {
606 tx.blocking_send(chunk.to_vec())
607 .map_err(|_| std::io::Error::other("client disconnected"))
608 };
609 knot_pack::upload_pack_streamed(&repo, &request, &catalog.fetch, &knot, &mut sink)
610 });
611
612 let mut forward = Ok(());
613 while let Some(chunk) = rx.recv().await {
614 if writer.write_all(&chunk).await.is_err() {
615 forward = Err(());
616 break;
617 }
618 }
619 drop(rx);
620 match (forward, handle.await) {
621 (Ok(()), Ok(Ok(()))) => writer.flush().await.map_err(|_| ()),
622 _ => Err(()),
623 }
624}
625
626async fn serve_receive<H: HttpTransport, C: Clock>(
627 state: Arc<SshState<H, C>>,
628 credential: Credential,
629 mut channel: Channel<Msg>,
630 repo_did: RepoDid,
631 peer: Option<IpAddr>,
632) {
633 let advert = {
634 let layout = state.layout.clone();
635 let did = repo_did.clone();
636 tokio::task::spawn_blocking(move || -> Result<(Vec<u8>, ObjectFormat), PackError> {
637 let repo = layout.open(&did)?;
638 let bytes = knot_pack::advertise_receive_ssh(&repo)?;
639 Ok((bytes, repo.object_format()))
640 })
641 .await
642 };
643 let (advert, object_format) = match advert {
644 Ok(Ok(pair)) => pair,
645 _ => return fail(channel, &state.catalog.ssh.advertise_failed.text()).await,
646 };
647
648 let mut writer = channel.make_writer();
649 if writer.write_all(&advert).await.is_err() || writer.flush().await.is_err() {
650 return;
651 }
652
653 let committer = match authorize_push(&state, &credential, &repo_did, peer).await {
654 PushAuth::Allowed(did) => did,
655 PushAuth::Refused { reason, message } => {
656 tracing::warn!(repo = repo_did.as_str(), reason, "ssh push denied");
657 return fail(channel, &message).await;
658 }
659 };
660
661 let _receive_permit = state.slots.receive.acquire().await;
662
663 let limits = state.limits;
664 let body = {
665 let mut reader = channel.make_reader();
666 let dir = state.layout.scratch_dir().to_path_buf();
667 match tokio::time::timeout(
668 RECEIVE_BODY_DEADLINE,
669 read_receive(
670 &mut reader,
671 dir,
672 state.max_pack_bytes,
673 limits,
674 object_format,
675 ),
676 )
677 .await
678 {
679 Ok(result) => result,
680 Err(_) => Err(ReadError::Deadline),
681 }
682 };
683 let body = match body {
684 Ok(body) => body,
685 Err(ReadError::TooLarge) => {
686 return fail(channel, &state.catalog.ssh.push_too_large.text()).await;
687 }
688 Err(ReadError::Deadline) => {
689 return fail(channel, &state.catalog.ssh.receive_deadline.text()).await;
690 }
691 Err(ReadError::Pack(error)) => {
692 tracing::warn!(repo = repo_did.as_str(), %error, "receive framing failed");
693 return fail(channel, &state.catalog.ssh.malformed_pack.text()).await;
694 }
695 Err(ReadError::Io(error)) => {
696 tracing::warn!(repo = repo_did.as_str(), %error, "receive read error");
697 return fail(channel, &state.catalog.ssh.receive_read_error.text()).await;
698 }
699 Err(ReadError::Truncated) => {
700 return fail(channel, &state.catalog.ssh.receive_ended_early.text()).await;
701 }
702 };
703 if body.is_empty() {
704 return finish(channel, 0).await;
705 }
706
707 let _pack_permit = state.slots.pack.acquire().await;
708 let landed = knot_receive::land(knot_receive::Push {
709 layout: &state.layout,
710 repo_did: &repo_did,
711 received: body,
712 limits: state.limits,
713 knot_actor: state.knot_actor.clone(),
714 committer,
715 events: Arc::clone(&state.events),
716 index: &state.index,
717 atproto: &state.atproto,
718 resolve_slots: &state.slots.resolve,
719 appview: &state.appview,
720 maintenance: &state.maintenance,
721 hostname: &state.hostname,
722 languages_push_budget: state.languages_push_budget,
723 catalog: Arc::clone(&state.catalog),
724 ci_logs: state.ci_logs.clone(),
725 })
726 .await;
727 match landed {
728 Ok(framed) => {
729 let _ = writer.write_all(&framed).await;
730 let _ = writer.flush().await;
731 finish(channel, 0).await;
732 }
733 Err(error) => {
734 tracing::warn!(repo = repo_did.as_str(), %error, "receive-pack failed");
735 fail(channel, &state.catalog.ssh.receive_failed.text()).await;
736 }
737 }
738}
739
740pub(crate) async fn run_greeting<H: HttpTransport, C: Clock>(
741 state: Arc<SshState<H, C>>,
742 credential: Credential,
743 channel: Channel<Msg>,
744) {
745 let knot = state.hostname.as_str().to_string();
746 let greeting = match greeting_visitor(&state, &credential).await {
747 Visitor::Named(who) => state.catalog.ssh.greeting.lines(|key| match key {
748 knot_messages::GreetingKey::User => who.clone(),
749 knot_messages::GreetingKey::Knot => knot.clone(),
750 }),
751 Visitor::Unknown => state
752 .catalog
753 .ssh
754 .greeting_unknown
755 .lines(|knot_messages::KnotKey::Knot| knot.clone()),
756 };
757 if greeting.is_empty() {
758 return finish(channel, 0).await;
759 }
760 let body = greeting.join("\r\n");
761 let _ = channel
762 .extended_data_bytes(1, format!("{body}\r\n").into_bytes())
763 .await;
764 finish(channel, 0).await;
765}
766
767async fn greeting_visitor<H: HttpTransport, C: Clock>(
768 state: &Arc<SshState<H, C>>,
769 credential: &Credential,
770) -> Visitor {
771 let did = match credential {
772 Credential::Identified(did) => did.clone(),
773 Credential::Offered(key) => {
774 match state.index.owner_of_key(key, state.atproto.now().seconds()) {
775 Resolved::Ready(Some(did)) => did,
776 _ => return Visitor::Unknown,
777 }
778 }
779 };
780 match knot_receive::resolve_handle(&state.atproto, &state.slots.resolve, &did).await {
781 Some(handle) => Visitor::Named(format!("@{}", handle.as_str())),
782 None => Visitor::Named(did.as_str().to_string()),
783 }
784}
785
786enum PusherLookup {
787 Matched(AccountDid),
788 Unmatched(Vec<AccountDid>),
789 Unavailable,
790}
791
792enum PushAuth {
793 Allowed(AccountDid),
794 Refused {
795 reason: &'static str,
796 message: String,
797 },
798}
799
800enum Visitor {
801 Named(String),
802 Unknown,
803}
804
805async fn authorize_push<H: HttpTransport, C: Clock>(
806 state: &Arc<SshState<H, C>>,
807 credential: &Credential,
808 repo: &RepoDid,
809 peer: Option<IpAddr>,
810) -> PushAuth {
811 match resolve_pusher(state, credential, repo, peer).await {
812 PusherLookup::Matched(did) => {
813 let acl = KnotAcl::new(&state.admins, state.admission, &state.index);
814 match can_push(&acl, &did, repo).is_allowed() {
815 true => PushAuth::Allowed(did),
816 false => PushAuth::Refused {
817 reason: "unauthorized",
818 message: state.catalog.ssh.push_denied.text(),
819 },
820 }
821 }
822 PusherLookup::Unavailable => PushAuth::Refused {
823 reason: "identity_unavailable",
824 message: state.catalog.ssh.identity_unavailable.text(),
825 },
826 PusherLookup::Unmatched(candidates) => {
827 let authorized = describe_authorized(state, &candidates).await;
828 PushAuth::Refused {
829 reason: "unregistered_key",
830 message: state
831 .catalog
832 .ssh
833 .key_not_registered
834 .line(|knot_messages::AuthorizedKey::Authorized| authorized.clone()),
835 }
836 }
837 }
838}
839
840async fn describe_authorized<H: HttpTransport, C: Clock>(
841 state: &Arc<SshState<H, C>>,
842 candidates: &[AccountDid],
843) -> String {
844 let names: Vec<String> = futures::stream::iter(
845 candidates
846 .iter()
847 .take(AUTHORIZED_NAMES_SHOWN)
848 .cloned()
849 .collect::<Vec<_>>(),
850 )
851 .map(|did| {
852 let state = Arc::clone(state);
853 async move {
854 match knot_receive::resolve_handle(&state.atproto, &state.slots.resolve, &did).await {
855 Some(handle) => format!("@{}", handle.as_str()),
856 None => did.as_str().to_string(),
857 }
858 }
859 })
860 .buffered(CANDIDATE_FANOUT)
861 .collect()
862 .await;
863 match (
864 names.as_slice(),
865 candidates.len().saturating_sub(AUTHORIZED_NAMES_SHOWN),
866 ) {
867 ([], _) => "nobody".to_string(),
868 (shown, 0) => shown.join(", "),
869 (shown, hidden) => format!("{}, and {hidden} more", shown.join(", ")),
870 }
871}
872
873fn probe_due<H: HttpTransport, C: Clock>(state: &Arc<SshState<H, C>>, did: &AccountDid) -> bool {
874 state
875 .probe_pace
876 .reserve_now(&SubjectKey::new(did.as_str()), state.atproto.now())
877}
878
879async fn resolve_pusher<H: HttpTransport, C: Clock>(
880 state: &Arc<SshState<H, C>>,
881 credential: &Credential,
882 repo: &RepoDid,
883 peer: Option<IpAddr>,
884) -> PusherLookup {
885 let key = match credential {
886 Credential::Identified(did) => return PusherLookup::Matched(did.clone()),
887 Credential::Offered(key) => key,
888 };
889 let owner = match state.index.owner_of(repo) {
890 Resolved::Ready(Some(owner)) => Some(AccountDid::from(owner)),
891 _ => None,
892 };
893 {
894 let index = Arc::clone(&state.index);
895 let target = repo.clone();
896 let _ = tokio::task::spawn_blocking(move || index.ensure_collaborators(&target)).await;
897 }
898 let collaborators = match state.index.collaborators_of(repo) {
899 Resolved::Ready(collaborators) => collaborators,
900 _ => Vec::new(),
901 };
902 let candidates: Vec<AccountDid> = owner.into_iter().chain(collaborators).collect();
903 let now = state.atproto.now().seconds();
904 if let Some(publisher) = state.index.keys().publisher_among(&candidates, key, now) {
905 return PusherLookup::Matched(publisher);
906 }
907 let unread: Vec<AccountDid> = candidates
908 .iter()
909 .filter(|did| !state.index.keys().is_fresh(did, now) || probe_due(state, did))
910 .cloned()
911 .collect();
912 if unread.is_empty() {
913 tracing::debug!(
914 ?peer,
915 repo = repo.as_str(),
916 candidates = candidates.len(),
917 "push check has every candidate's keys on file, and the candidates don't publish \
918 the offered key"
919 );
920 return PusherLookup::Unmatched(candidates);
921 }
922 let lease = state.key_ttl.lease_from(now);
923 let unresolved = Arc::new(AtomicBool::new(false));
924 let read: Vec<Option<AccountDid>> = futures::stream::iter(unread)
925 .map(|did| {
926 let state = Arc::clone(state);
927 let key = key.clone();
928 let unresolved = Arc::clone(&unresolved);
929 async move {
930 let _permit = state.slots.resolve.acquire().await;
931 match state.atproto.resolve_pubkeys(&did).await {
932 Ok(keys) => {
933 let matches = keys.contains(&key);
934 state.index.keys().record(&did, keys, lease);
935 matches.then_some(did)
936 }
937 Err(error) if error.is_gone() => {
938 tracing::debug!(
939 did = did.as_str(),
940 %error,
941 "push check records an empty key set for a candidate whose DID document is gone"
942 );
943 state.index.keys().record(&did, Vec::new(), lease);
944 None
945 }
946 Err(error) => {
947 tracing::debug!(
948 did = did.as_str(),
949 %error,
950 "push check couldn't read a candidate's records"
951 );
952 unresolved.store(true, Ordering::Relaxed);
953 None
954 }
955 }
956 }
957 })
958 .buffered(CANDIDATE_FANOUT)
959 .collect()
960 .await;
961 match read.into_iter().flatten().next() {
962 Some(did) => PusherLookup::Matched(did),
963 None if unresolved.load(Ordering::Relaxed) => PusherLookup::Unavailable,
964 None => PusherLookup::Unmatched(candidates),
965 }
966}
967
968async fn read_chunk<R: AsyncRead + Unpin>(
969 reader: &mut R,
970 buf: &mut Vec<u8>,
971 limit: usize,
972) -> Result<bool, ReadError> {
973 let mut chunk = [0u8; READ_CHUNK];
974 let read = reader.read(&mut chunk).await.map_err(ReadError::Io)?;
975 if read == 0 {
976 return Ok(false);
977 }
978 buf.extend_from_slice(&chunk[..read]);
979 if buf.len() > limit {
980 return Err(ReadError::TooLarge);
981 }
982 Ok(true)
983}
984
985async fn read_receive<R: AsyncRead + Unpin>(
986 reader: &mut R,
987 dir: PathBuf,
988 limit: knot_pack::MaxWireBytes,
989 limits: PackLimits,
990 format: ObjectFormat,
991) -> Result<knot_pack::ReceivedPack, ReadError> {
992 let (tx, rx) = mpsc::channel::<Vec<u8>>(8);
993 let mut framer =
994 tokio::task::spawn_blocking(move || frame_receive(rx, dir, limit, limits, format));
995 let mut chunk = [0u8; READ_CHUNK];
996 let mut io_error = None;
997 loop {
998 tokio::select! {
999 biased;
1000 framed = &mut framer => return join_framed(framed, io_error),
1001 read = reader.read(&mut chunk) => match read {
1002 Ok(0) => break,
1003 Ok(read) => {
1004 if tx.send(chunk[..read].to_vec()).await.is_err() {
1005 break;
1006 }
1007 }
1008 Err(error) => {
1009 io_error = Some(error);
1010 break;
1011 }
1012 },
1013 }
1014 }
1015 drop(tx);
1016 join_framed(framer.await, io_error)
1017}
1018
1019fn join_framed(
1020 framed: Result<Result<knot_pack::ReceivedPack, ReadError>, tokio::task::JoinError>,
1021 io_error: Option<std::io::Error>,
1022) -> Result<knot_pack::ReceivedPack, ReadError> {
1023 match framed {
1024 Ok(Ok(body)) => Ok(body),
1025 Ok(Err(ReadError::Truncated)) => {
1026 Err(io_error.map(ReadError::Io).unwrap_or(ReadError::Truncated))
1027 }
1028 Ok(Err(other)) => Err(other),
1029 Err(_) => Err(ReadError::Truncated),
1030 }
1031}
1032
1033fn read_error(error: knot_pack::ReceiveReadError) -> ReadError {
1034 match error {
1035 knot_pack::ReceiveReadError::Io(error) => ReadError::Io(error),
1036 knot_pack::ReceiveReadError::Pack(error) => ReadError::Pack(error),
1037 knot_pack::ReceiveReadError::TooLarge => ReadError::TooLarge,
1038 knot_pack::ReceiveReadError::Truncated => ReadError::Truncated,
1039 }
1040}
1041
1042fn frame_receive(
1043 mut rx: mpsc::Receiver<Vec<u8>>,
1044 dir: PathBuf,
1045 limit: knot_pack::MaxWireBytes,
1046 limits: PackLimits,
1047 format: ObjectFormat,
1048) -> Result<knot_pack::ReceivedPack, ReadError> {
1049 let mut receiver =
1050 knot_pack::PackReceiver::new(&dir, limit, limits, format.kind()).map_err(ReadError::Io)?;
1051 loop {
1052 match rx.blocking_recv() {
1053 Some(chunk) => {
1054 if receiver.write(&chunk).map_err(read_error)? {
1055 return receiver.finish().map_err(read_error);
1056 }
1057 }
1058 None => return receiver.finish().map_err(read_error),
1059 }
1060 }
1061}
1062
1063async fn fail(channel: Channel<Msg>, message: &str) {
1064 let _ = channel
1065 .extended_data_bytes(1, format!("{message}\n").into_bytes())
1066 .await;
1067 finish(channel, 1).await;
1068}
1069
1070async fn finish(channel: Channel<Msg>, status: u32) {
1071 let _ = channel.exit_status(status).await;
1072 let _ = channel.eof().await;
1073 let _ = channel.close().await;
1074}
1075
1076#[cfg(test)]
1077mod tests {
1078 use super::*;
1079
1080 #[test]
1081 fn the_lfs_progress_budget_spares_slow_links_and_cuts_trickles() {
1082 assert!(lfs_within_progress_budget(Duration::from_secs(59), 0));
1083 assert!(!lfs_within_progress_budget(Duration::from_secs(61), 0));
1084 assert!(lfs_within_progress_budget(
1085 Duration::from_secs(50_000),
1086 5 * 1024 * 1024 * 1024
1087 ));
1088 assert!(!lfs_within_progress_budget(Duration::from_secs(1_000), 10));
1089 }
1090
1091 #[test]
1092 fn the_repo_path_parser_separates_dids_from_handles() {
1093 assert!(matches!(
1094 parse_repo_path("did:plc:nel/squid"),
1095 Some(RepoRef::OwnerPath(..))
1096 ));
1097 assert!(matches!(
1098 parse_repo_path("nel.pet/squid"),
1099 Some(RepoRef::HandlePath(..))
1100 ));
1101 assert!(matches!(
1102 parse_repo_path("did:plc:barnacle"),
1103 Some(RepoRef::Did(_))
1104 ));
1105 assert!(parse_repo_path("did:nonsense/squid").is_none());
1106 assert!(parse_repo_path("nel.pet").is_none());
1107 }
1108}