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::time::Duration;
5
6use futures::StreamExt;
7use knot_acl::{KnotAcl, can_push};
8use knot_index::Resolved;
9use knot_lfs::TransferOp;
10use knot_pack::{PackError, PackLimits, RepoLookup};
11use knot_runtime::{Clock, HttpTransport};
12use knot_types::{AccountDid, ObjectFormat, OfferedKey, OwnerDid, RepoDid, RepoRkey};
13use russh::Channel;
14use russh::server::Msg;
15use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
16use tokio::runtime::Handle;
17use tokio::sync::mpsc;
18
19use crate::SshState;
20
21const READ_CHUNK: usize = 64 * 1024;
22const MAX_UPLOAD_REQUEST: usize = 16 * 1024 * 1024;
23const RECEIVE_BODY_DEADLINE: Duration = Duration::from_secs(1800);
24const ARCHIVE_REQUEST_DEADLINE: Duration = Duration::from_secs(60);
25const LFS_PROGRESS_GRACE: Duration = Duration::from_secs(60);
26const LFS_PROGRESS_FLOOR_BYTES_PER_SEC: u64 = 1024;
27const LFS_STALL_TIMEOUT: Duration = Duration::from_secs(120);
28
29fn lfs_within_progress_budget(waited: Duration, moved_bytes: u64) -> bool {
30 waited
31 <= LFS_PROGRESS_GRACE + Duration::from_secs(moved_bytes / LFS_PROGRESS_FLOOR_BYTES_PER_SEC)
32}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35enum Service {
36 Upload,
37 UploadArchive,
38 Receive,
39 Lfs(TransferOp),
40}
41
42enum ReadError {
43 Io(std::io::Error),
44 Pack(PackError),
45 TooLarge,
46 Truncated,
47 Deadline,
48}
49
50enum RepoRef {
51 Did(RepoDid),
52 OwnerRkey(OwnerDid, Vec<RepoRkey>),
53 HandleRkey(knot_types::Handle, Vec<RepoRkey>),
54}
55
56enum ResolvedRef {
57 Did(RepoDid),
58 OwnerRkey(OwnerDid, Vec<RepoRkey>),
59}
60
61fn parse_exec(command: &[u8]) -> Option<(Service, RepoRef)> {
62 let text = std::str::from_utf8(command).ok()?.trim();
63 if let Some(rest) = text.strip_prefix("git-lfs-transfer ") {
64 let (path, op_token) = rest.trim().rsplit_once(' ')?;
65 let op = TransferOp::parse(op_token.trim())?;
66 return Some((Service::Lfs(op), parse_repo_path(path)?));
67 }
68 let (service, rest) = [
69 ("git-upload-pack ", Service::Upload),
70 ("git upload-pack ", Service::Upload),
71 ("git-upload-archive ", Service::UploadArchive),
72 ("git upload-archive ", Service::UploadArchive),
73 ("git-receive-pack ", Service::Receive),
74 ("git receive-pack ", Service::Receive),
75 ]
76 .into_iter()
77 .find_map(|(prefix, service)| text.strip_prefix(prefix).map(|rest| (service, rest)))?;
78 Some((service, parse_repo_path(rest)?))
79}
80
81fn parse_repo_path(raw: &str) -> Option<RepoRef> {
82 let path = raw
83 .trim()
84 .trim_matches('\'')
85 .trim_matches('"')
86 .trim_start_matches('/');
87 match path.split_once('/') {
88 Some((owner, name)) => {
89 let candidates: Vec<RepoRkey> = RepoRkey::clone_path_candidates(name).collect();
90 if candidates.is_empty() {
91 return None;
92 }
93 match knot_types::OwnerRef::parse(owner)? {
94 knot_types::OwnerRef::Did(owner) => Some(RepoRef::OwnerRkey(owner, candidates)),
95 knot_types::OwnerRef::Handle(handle) => {
96 Some(RepoRef::HandleRkey(handle, candidates))
97 }
98 }
99 }
100 None => Some(RepoRef::Did(RepoDid::new(path).ok()?)),
101 }
102}
103
104fn resolve_repo_ref<H: HttpTransport, C: Clock>(
105 state: &Arc<SshState<H, C>>,
106 repo_ref: ResolvedRef,
107) -> RepoLookup {
108 let candidate = match repo_ref {
109 ResolvedRef::Did(did) => RepoLookup::Hosted(did),
110 ResolvedRef::OwnerRkey(owner, candidates) => RepoLookup::first(candidates, |rkey| {
111 RepoLookup::from_resolved(state.index.resolve_repo(&owner, &rkey), |found| found)
112 }),
113 };
114 match candidate {
115 RepoLookup::Hosted(did) => {
116 RepoLookup::from_resolved(state.index.owner_of(&did), |_| did.clone())
117 }
118 undecided => undecided,
119 }
120}
121
122pub(crate) async fn run_exec<H: HttpTransport, C: Clock>(
123 state: Arc<SshState<H, C>>,
124 key: Option<OfferedKey>,
125 channel: Channel<Msg>,
126 command: &[u8],
127 protocol_v2: bool,
128 peer: Option<IpAddr>,
129) {
130 let Some((service, repo_ref)) = parse_exec(command) else {
131 fail(channel, &state.catalog.ssh.unsupported_command.text()).await;
132 return;
133 };
134 let peer_limiter = match &service {
135 Service::Lfs(_) => state
136 .lfs
137 .as_ref()
138 .map_or(&state.peer_slots, |lfs| &lfs.peer_slots),
139 _ => &state.peer_slots,
140 };
141 let _peer_guard = match peer_limiter.admit(peer, state.atproto.now()) {
142 Ok(guard) => guard,
143 Err(refusal) => {
144 let reason = match refusal {
145 knot_resource::Refusal::RateLimited => "peer request rate exceeded",
146 knot_resource::Refusal::Saturated => "peer concurrency limit reached",
147 };
148 tracing::warn!(?peer, reason, "ssh exec rejected");
149 return fail(channel, &state.catalog.ssh.too_many_operations.text()).await;
150 }
151 };
152 let resolved_ref = match repo_ref {
153 RepoRef::Did(did) => ResolvedRef::Did(did),
154 RepoRef::OwnerRkey(owner, candidates) => ResolvedRef::OwnerRkey(owner, candidates),
155 RepoRef::HandleRkey(owner_handle, candidates) => {
156 match state
157 .atproto
158 .resolve_handle_to_did(&owner_handle)
159 .await
160 .ok()
161 {
162 Some(did) => ResolvedRef::OwnerRkey(did.into(), candidates),
163 None => {
164 fail(channel, &state.catalog.ssh.repo_not_found.text()).await;
165 return;
166 }
167 }
168 }
169 };
170 let repo_did = match resolve_repo_ref(&state, resolved_ref) {
171 RepoLookup::Hosted(did) => did,
172 RepoLookup::Unhosted => {
173 fail(channel, &state.catalog.ssh.repo_not_found.text()).await;
174 return;
175 }
176 RepoLookup::Unavailable => {
177 fail(channel, &state.catalog.ssh.index_warming.text()).await;
178 return;
179 }
180 };
181 let layout = state.layout.clone();
182 let did = repo_did.clone();
183 let opened = tokio::task::spawn_blocking(move || layout.open(&did).is_ok())
184 .await
185 .unwrap_or(false);
186 if !opened {
187 fail(channel, &state.catalog.ssh.repo_not_found.text()).await;
188 return;
189 }
190 match service {
191 Service::Upload => serve_upload(state, channel, repo_did, protocol_v2).await,
192 Service::UploadArchive => serve_upload_archive(state, channel, repo_did).await,
193 Service::Receive => serve_receive(state, key, channel, repo_did).await,
194 Service::Lfs(op) => serve_lfs(state, key, channel, repo_did, op).await,
195 }
196}
197
198async fn serve_lfs<H: HttpTransport, C: Clock>(
199 state: Arc<SshState<H, C>>,
200 key: Option<OfferedKey>,
201 mut channel: Channel<Msg>,
202 repo_did: RepoDid,
203 op: TransferOp,
204) {
205 let Some(lfs) = state.lfs.clone() else {
206 return fail(channel, &state.catalog.ssh.lfs_disabled.text()).await;
207 };
208 if op == TransferOp::Upload {
209 let pusher = resolve_pusher(&state, key.as_ref(), &repo_did).await;
210 let allowed = pusher.as_ref().is_some_and(|did| {
211 let acl = KnotAcl::new(&state.admins, state.admission, &state.index);
212 can_push(&acl, did, &repo_did).is_allowed()
213 });
214 if !allowed {
215 tracing::warn!(
216 repo = repo_did.as_str(),
217 registered = pusher.is_some(),
218 "ssh lfs upload denied"
219 );
220 let message = match pusher {
221 None => state.catalog.ssh.key_not_registered.text(),
222 Some(_) => state.catalog.ssh.push_denied.text(),
223 };
224 return fail(channel, &message).await;
225 }
226 }
227 let permit = match Arc::clone(&lfs.slots).acquire_owned().await {
228 Ok(permit) => permit,
229 Err(_) => return fail(channel, &state.catalog.ssh.shutting_down.text()).await,
230 };
231 let started = std::time::Instant::now();
232
233 let (tx, rx) = mpsc::channel::<Vec<u8>>(8);
234 let writer = Box::pin(channel.make_writer());
235 let handle = lfs.handle.clone();
236 let runtime = Handle::current();
237 let did = repo_did.clone();
238 let catalog = Arc::clone(&state.catalog);
239 let mut engine = tokio::task::spawn_blocking(move || {
240 let _permit = permit;
241 let output = std::io::BufWriter::new(MeteredWrite::new(runtime.clone(), writer));
242 knot_lfs::serve_transfer(
243 handle.store.as_ref(),
244 handle.admission.as_ref(),
245 &did,
246 op,
247 &catalog.lfs,
248 MpscRead::new(runtime, rx),
249 output,
250 )
251 });
252 let joined = {
253 let reader = channel.make_reader();
254 tokio::select! {
255 joined = &mut engine => joined,
256 () = pump_input(reader, tx) => engine.await,
257 }
258 };
259 let status = match joined {
260 Ok(Ok(())) => {
261 tracing::info!(
262 repo = repo_did.as_str(),
263 op = match op {
264 TransferOp::Upload => "upload",
265 TransferOp::Download => "download",
266 },
267 duration_ms = started.elapsed().as_millis() as u64,
268 "ssh lfs transfer finished"
269 );
270 0
271 }
272 Ok(Err(fault)) => {
273 tracing::warn!(repo = repo_did.as_str(), %fault, "ssh lfs transfer failed");
274 1
275 }
276 Err(join) => {
277 tracing::error!(repo = repo_did.as_str(), %join, "ssh lfs transfer task panicked");
278 1
279 }
280 };
281 finish(channel, status).await;
282}
283
284async fn pump_input<R: AsyncRead + Unpin>(reader: R, tx: mpsc::Sender<Vec<u8>>) {
285 use futures::TryStreamExt;
286 let _ = tokio_util::io::ReaderStream::with_capacity(reader, READ_CHUNK)
287 .map_err(|_| ())
288 .try_for_each(|chunk| {
289 let tx = &tx;
290 async move {
291 match chunk.is_empty() {
292 true => Ok(()),
293 false => tx.send(chunk.to_vec()).await.map_err(|_| ()),
294 }
295 }
296 })
297 .await;
298}
299
300fn stalled(direction: &'static str) -> std::io::Error {
301 std::io::Error::other(format!("lfs {direction} stalled past the idle timeout"))
302}
303
304struct MpscRead {
305 runtime: Handle,
306 rx: mpsc::Receiver<Vec<u8>>,
307 buffer: Vec<u8>,
308 offset: usize,
309 waited: Duration,
310 received: u64,
311}
312
313impl MpscRead {
314 fn new(runtime: Handle, rx: mpsc::Receiver<Vec<u8>>) -> Self {
315 Self {
316 runtime,
317 rx,
318 buffer: Vec::new(),
319 offset: 0,
320 waited: Duration::ZERO,
321 received: 0,
322 }
323 }
324}
325
326impl std::io::Read for MpscRead {
327 fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
328 if self.offset >= self.buffer.len() {
329 let started = std::time::Instant::now();
330 let rx = &mut self.rx;
331 let received = self
332 .runtime
333 // I know I know, but these aren't runtime workers here
334 .block_on(async { tokio::time::timeout(LFS_STALL_TIMEOUT, rx.recv()).await });
335 match received {
336 Ok(Some(chunk)) => {
337 self.waited += started.elapsed();
338 self.received += chunk.len() as u64;
339 if !lfs_within_progress_budget(self.waited, self.received) {
340 return Err(std::io::Error::other(
341 "lfs input trickles below the progress floor",
342 ));
343 }
344 self.buffer = chunk;
345 self.offset = 0;
346 }
347 Ok(None) => return Ok(0),
348 Err(_) => return Err(stalled("input")),
349 }
350 }
351 let take = out.len().min(self.buffer.len() - self.offset);
352 out[..take].copy_from_slice(&self.buffer[self.offset..self.offset + take]);
353 self.offset += take;
354 Ok(take)
355 }
356}
357
358struct MeteredWrite<W> {
359 runtime: Handle,
360 inner: W,
361 waited: Duration,
362 written: u64,
363}
364
365impl<W: AsyncWrite + Unpin> MeteredWrite<W> {
366 fn new(runtime: Handle, inner: W) -> Self {
367 Self {
368 runtime,
369 inner,
370 waited: Duration::ZERO,
371 written: 0,
372 }
373 }
374
375 fn charge(&mut self, started: std::time::Instant) -> std::io::Result<()> {
376 self.waited += started.elapsed();
377 match lfs_within_progress_budget(self.waited, self.written) {
378 true => Ok(()),
379 false => Err(std::io::Error::other(
380 "lfs output trickles below the progress floor",
381 )),
382 }
383 }
384}
385
386impl<W: AsyncWrite + Unpin> std::io::Write for MeteredWrite<W> {
387 fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
388 let started = std::time::Instant::now();
389 let inner = &mut self.inner;
390 let wrote = self
391 .runtime
392 .block_on(async { tokio::time::timeout(LFS_STALL_TIMEOUT, inner.write(buf)).await })
393 .map_err(|_| stalled("output"))??;
394 self.written += wrote as u64;
395 self.charge(started).map(|()| wrote)
396 }
397
398 fn flush(&mut self) -> std::io::Result<()> {
399 let started = std::time::Instant::now();
400 let inner = &mut self.inner;
401 self.runtime
402 .block_on(async { tokio::time::timeout(LFS_STALL_TIMEOUT, inner.flush()).await })
403 .map_err(|_| stalled("output"))??;
404 self.charge(started)
405 }
406}
407
408async fn serve_upload_archive<H: HttpTransport, C: Clock>(
409 state: Arc<SshState<H, C>>,
410 mut channel: Channel<Msg>,
411 repo_did: RepoDid,
412) {
413 let request = {
414 let mut reader = channel.make_reader();
415 tokio::time::timeout(ARCHIVE_REQUEST_DEADLINE, read_archive_request(&mut reader)).await
416 };
417 let request = match request {
418 Ok(Ok(request)) => request,
419 Ok(Err(())) => return fail(channel, &state.catalog.ssh.archive_malformed.text()).await,
420 Err(_) => return fail(channel, &state.catalog.ssh.archive_timeout.text()).await,
421 };
422
423 let permit = state.slots.pack.acquire().await;
424 let (tx, mut rx) = mpsc::channel::<Vec<u8>>(16);
425 let layout = state.layout.clone();
426 let did = repo_did.clone();
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, &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 key: Option<OfferedKey>,
629 mut channel: Channel<Msg>,
630 repo_did: RepoDid,
631) {
632 let advert = {
633 let layout = state.layout.clone();
634 let did = repo_did.clone();
635 tokio::task::spawn_blocking(move || -> Result<(Vec<u8>, ObjectFormat), PackError> {
636 let repo = layout.open(&did)?;
637 let bytes = knot_pack::advertise_receive_ssh(&repo)?;
638 Ok((bytes, repo.object_format()))
639 })
640 .await
641 };
642 let (advert, object_format) = match advert {
643 Ok(Ok(pair)) => pair,
644 _ => return fail(channel, &state.catalog.ssh.advertise_failed.text()).await,
645 };
646
647 let mut writer = channel.make_writer();
648 if writer.write_all(&advert).await.is_err() || writer.flush().await.is_err() {
649 return;
650 }
651
652 let pusher = resolve_pusher(&state, key.as_ref(), &repo_did).await;
653 let allowed = |did: &AccountDid| {
654 let acl = KnotAcl::new(&state.admins, state.admission, &state.index);
655 can_push(&acl, did, &repo_did).is_allowed()
656 };
657 let committer = match pusher {
658 Some(did) if allowed(&did) => did,
659 Some(_) => {
660 tracing::warn!(
661 repo = repo_did.as_str(),
662 registered = true,
663 "ssh push denied"
664 );
665 return fail(channel, &state.catalog.ssh.push_denied.text()).await;
666 }
667 None => {
668 tracing::warn!(
669 repo = repo_did.as_str(),
670 registered = false,
671 "ssh push denied"
672 );
673 return fail(channel, &state.catalog.ssh.key_not_registered.text()).await;
674 }
675 };
676
677 let _receive_permit = state.slots.receive.acquire().await;
678
679 let limits = state.limits;
680 let body = {
681 let mut reader = channel.make_reader();
682 let dir = state.layout.scratch_dir().to_path_buf();
683 match tokio::time::timeout(
684 RECEIVE_BODY_DEADLINE,
685 read_receive(
686 &mut reader,
687 dir,
688 state.max_pack_bytes,
689 limits,
690 object_format,
691 ),
692 )
693 .await
694 {
695 Ok(result) => result,
696 Err(_) => Err(ReadError::Deadline),
697 }
698 };
699 let body = match body {
700 Ok(body) => body,
701 Err(ReadError::TooLarge) => {
702 return fail(channel, &state.catalog.ssh.push_too_large.text()).await;
703 }
704 Err(ReadError::Deadline) => {
705 return fail(channel, &state.catalog.ssh.receive_deadline.text()).await;
706 }
707 Err(ReadError::Pack(error)) => {
708 tracing::warn!(repo = repo_did.as_str(), %error, "receive framing failed");
709 return fail(channel, &state.catalog.ssh.malformed_pack.text()).await;
710 }
711 Err(ReadError::Io(error)) => {
712 tracing::warn!(repo = repo_did.as_str(), %error, "receive read error");
713 return fail(channel, &state.catalog.ssh.receive_read_error.text()).await;
714 }
715 Err(ReadError::Truncated) => {
716 return fail(channel, &state.catalog.ssh.receive_ended_early.text()).await;
717 }
718 };
719 if body.is_empty() {
720 return finish(channel, 0).await;
721 }
722
723 let _pack_permit = state.slots.pack.acquire().await;
724 let landed = knot_receive::land(knot_receive::Push {
725 layout: &state.layout,
726 repo_did: &repo_did,
727 received: body,
728 limits: state.limits,
729 knot_actor: state.knot_actor.clone(),
730 committer,
731 events: Arc::clone(&state.events),
732 index: &state.index,
733 atproto: &state.atproto,
734 resolve_slots: &state.slots.resolve,
735 appview: &state.appview,
736 maintenance: &state.maintenance,
737 hostname: &state.hostname,
738 languages_push_budget: state.languages_push_budget,
739 catalog: Arc::clone(&state.catalog),
740 ci_logs: state.ci_logs.clone(),
741 })
742 .await;
743 match landed {
744 Ok(framed) => {
745 let _ = writer.write_all(&framed).await;
746 let _ = writer.flush().await;
747 finish(channel, 0).await;
748 }
749 Err(error) => {
750 tracing::warn!(repo = repo_did.as_str(), %error, "receive-pack failed");
751 fail(channel, &state.catalog.ssh.receive_failed.text()).await;
752 }
753 }
754}
755
756pub(crate) async fn run_greeting<H: HttpTransport, C: Clock>(
757 state: Arc<SshState<H, C>>,
758 key: Option<OfferedKey>,
759 channel: Channel<Msg>,
760) {
761 let who = greeting_identity(&state, key.as_ref()).await;
762 let greeting = state.catalog.ssh.greeting.lines(|key| match key {
763 knot_messages::GreetingKey::User => who.clone(),
764 knot_messages::GreetingKey::Knot => state.hostname.as_str().to_string(),
765 });
766 if greeting.is_empty() {
767 return finish(channel, 0).await;
768 }
769 let body = greeting.join("\r\n");
770 let _ = channel
771 .extended_data_bytes(1, format!("{body}\r\n").into_bytes())
772 .await;
773 finish(channel, 0).await;
774}
775
776async fn greeting_identity<H: HttpTransport, C: Clock>(
777 state: &Arc<SshState<H, C>>,
778 key: Option<&OfferedKey>,
779) -> String {
780 let Some(did) = key.and_then(|key| state.roster.did_for(key)) else {
781 return "there".to_string();
782 };
783 match knot_receive::resolve_handle(&state.atproto, &state.slots.resolve, &did).await {
784 Some(handle) => format!("@{}", handle.as_str()),
785 None => did.as_str().to_string(),
786 }
787}
788
789async fn resolve_pusher<H: HttpTransport, C: Clock>(
790 state: &Arc<SshState<H, C>>,
791 key: Option<&OfferedKey>,
792 repo: &RepoDid,
793) -> Option<AccountDid> {
794 let key = key?;
795 let owner = match state.index.owner_of(repo) {
796 Resolved::Ready(Some(owner)) => Some(AccountDid::from(owner)),
797 _ => None,
798 };
799 {
800 let index = Arc::clone(&state.index);
801 let target = repo.clone();
802 let _ = tokio::task::spawn_blocking(move || index.ensure_collaborators(&target)).await;
803 }
804 let collaborators = match state.index.collaborators_of(repo) {
805 Resolved::Ready(collaborators) => collaborators,
806 _ => Vec::new(),
807 };
808 let candidates: Vec<AccountDid> = owner.into_iter().chain(collaborators).collect();
809 if let Resolved::Ready(Some(cached)) = state.index.owner_of_key(key)
810 && candidates.contains(&cached)
811 {
812 return Some(cached);
813 }
814 let _permit = state.slots.resolve.acquire().await;
815 let matches = futures::stream::iter(candidates).filter_map(|did| async move {
816 let keys = state.atproto.resolve_pubkeys(&did).await.ok()?;
817 keys.iter()
818 .for_each(|resolved| state.index.cache_key(resolved.clone(), &did));
819 keys.iter().any(|resolved| resolved == key).then_some(did)
820 });
821 futures::pin_mut!(matches);
822 matches.next().await
823}
824
825async fn read_chunk<R: AsyncRead + Unpin>(
826 reader: &mut R,
827 buf: &mut Vec<u8>,
828 limit: usize,
829) -> Result<bool, ReadError> {
830 let mut chunk = [0u8; READ_CHUNK];
831 let read = reader.read(&mut chunk).await.map_err(ReadError::Io)?;
832 if read == 0 {
833 return Ok(false);
834 }
835 buf.extend_from_slice(&chunk[..read]);
836 if buf.len() > limit {
837 return Err(ReadError::TooLarge);
838 }
839 Ok(true)
840}
841
842async fn read_receive<R: AsyncRead + Unpin>(
843 reader: &mut R,
844 dir: PathBuf,
845 limit: knot_pack::MaxWireBytes,
846 limits: PackLimits,
847 format: ObjectFormat,
848) -> Result<knot_pack::ReceivedPack, ReadError> {
849 let (tx, rx) = mpsc::channel::<Vec<u8>>(8);
850 let mut framer =
851 tokio::task::spawn_blocking(move || frame_receive(rx, dir, limit, limits, format));
852 let mut chunk = [0u8; READ_CHUNK];
853 let mut io_error = None;
854 loop {
855 tokio::select! {
856 biased;
857 framed = &mut framer => return join_framed(framed, io_error),
858 read = reader.read(&mut chunk) => match read {
859 Ok(0) => break,
860 Ok(read) => {
861 if tx.send(chunk[..read].to_vec()).await.is_err() {
862 break;
863 }
864 }
865 Err(error) => {
866 io_error = Some(error);
867 break;
868 }
869 },
870 }
871 }
872 drop(tx);
873 join_framed(framer.await, io_error)
874}
875
876fn join_framed(
877 framed: Result<Result<knot_pack::ReceivedPack, ReadError>, tokio::task::JoinError>,
878 io_error: Option<std::io::Error>,
879) -> Result<knot_pack::ReceivedPack, ReadError> {
880 match framed {
881 Ok(Ok(body)) => Ok(body),
882 Ok(Err(ReadError::Truncated)) => {
883 Err(io_error.map(ReadError::Io).unwrap_or(ReadError::Truncated))
884 }
885 Ok(Err(other)) => Err(other),
886 Err(_) => Err(ReadError::Truncated),
887 }
888}
889
890fn read_error(error: knot_pack::ReceiveReadError) -> ReadError {
891 match error {
892 knot_pack::ReceiveReadError::Io(error) => ReadError::Io(error),
893 knot_pack::ReceiveReadError::Pack(error) => ReadError::Pack(error),
894 knot_pack::ReceiveReadError::TooLarge => ReadError::TooLarge,
895 knot_pack::ReceiveReadError::Truncated => ReadError::Truncated,
896 }
897}
898
899fn frame_receive(
900 mut rx: mpsc::Receiver<Vec<u8>>,
901 dir: PathBuf,
902 limit: knot_pack::MaxWireBytes,
903 limits: PackLimits,
904 format: ObjectFormat,
905) -> Result<knot_pack::ReceivedPack, ReadError> {
906 let mut receiver =
907 knot_pack::PackReceiver::new(&dir, limit, limits, format.kind()).map_err(ReadError::Io)?;
908 loop {
909 match rx.blocking_recv() {
910 Some(chunk) => {
911 if receiver.write(&chunk).map_err(read_error)? {
912 return receiver.finish().map_err(read_error);
913 }
914 }
915 None => return receiver.finish().map_err(read_error),
916 }
917 }
918}
919
920async fn fail(channel: Channel<Msg>, message: &str) {
921 let _ = channel
922 .extended_data_bytes(1, format!("{message}\n").into_bytes())
923 .await;
924 finish(channel, 1).await;
925}
926
927async fn finish(channel: Channel<Msg>, status: u32) {
928 let _ = channel.exit_status(status).await;
929 let _ = channel.eof().await;
930 let _ = channel.close().await;
931}
932
933#[cfg(test)]
934mod tests {
935 use super::*;
936
937 #[test]
938 fn the_lfs_progress_budget_spares_slow_links_and_cuts_trickles() {
939 assert!(lfs_within_progress_budget(Duration::from_secs(59), 0));
940 assert!(!lfs_within_progress_budget(Duration::from_secs(61), 0));
941 assert!(lfs_within_progress_budget(
942 Duration::from_secs(50_000),
943 5 * 1024 * 1024 * 1024
944 ));
945 assert!(!lfs_within_progress_budget(Duration::from_secs(1_000), 10));
946 }
947
948 #[test]
949 fn the_repo_path_parser_separates_dids_from_handles() {
950 assert!(matches!(
951 parse_repo_path("did:plc:nel/squid"),
952 Some(RepoRef::OwnerRkey(..))
953 ));
954 assert!(matches!(
955 parse_repo_path("nel.pet/squid"),
956 Some(RepoRef::HandleRkey(..))
957 ));
958 assert!(matches!(
959 parse_repo_path("did:plc:barnacle"),
960 Some(RepoRef::Did(_))
961 ));
962 assert!(parse_repo_path("did:nonsense/squid").is_none());
963 assert!(parse_repo_path("nel.pet").is_none());
964 }
965}