feat(ssh): implement duplex git session handling for upload-pack and upload-archive

Modified lib/git/src/service/merge.rs +1 −6 View lib/git/src/service/merge.rs

@@ -526,12 +526,7 @@

526526 let head_oid = baby.rev_parse(head).await.map_err(map_baby)?;

527527 let max = limit.min(MAX_LISTED_COMMITS);

528528 let ids = baby

529 .commit_ids_between_reverse(

530 &hex(&base_oid),

531 &hex(&head_oid),

532 None,

533 Some(max),

534 )

529 .commit_ids_between_reverse(&hex(&base_oid), &hex(&head_oid), None, Some(max))

535530 .await

536531 .map_err(map_baby)?;

537532 let mut out = Vec::with_capacity(ids.len());

Modified lib/git/src/ssh/handler/channel.rs +199 −11 View lib/git/src/ssh/handler/channel.rs

@@ -12,7 +12,9 @@

1212use crate::smart::pre_receive::parse_receive_pack_refs;

1313use crate::ssh::error::SshError;

1414use crate::ssh::git_service::apply_hardened_env;

15use crate::ssh::handler::{ChannelState, GitSshHandler};

15use crate::ssh::handler::ChannelState;

16use crate::ssh::handler::GitSshHandler;

17use crate::ssh::handler::git_op::GitOperation;

1618use crate::sync::post_receive::enqueue_post_receive_push_with_ci_logged;

1719

1820pub const MAX_SSH_CHANNEL_BUF: usize = 64 * 1024 * 1024;

@@ -38,6 +40,12 @@

3840 if state.buffer_overflow {

3941 return Ok(());

4042 }

43 if state.dispatched {

44 if let Some(tx) = &state.stdin_tx {

45 let _ = tx.send(data.to_vec());

46 }

47 return Ok(());

48 }

4149 let projected = state.buffer.len().saturating_add(data.len());

4250 if projected > MAX_SSH_CHANNEL_BUF {

4351 tracing::warn!(

@@ -61,14 +69,11 @@

6169 "ssh: first data chunk on armed channel"

6270 );

6371 }

64 let is_receive = matches!(

65 state.git_op,

66 Some(crate::ssh::handler::git_op::GitOperation::ReceivePack { .. })

67 );

68 if !is_receive {

69 return Ok(());

70 }

71 if !state.nak_sent && Self::wants_section_complete(&state.buffer) {

72 // Push-only: receive-pack runs after EOF, so its negotiation reply has

73 // to be synthesized here. Upload services must not receive it, their

74 // own process emits the real NAK/ACK.

75 let is_receive = matches!(state.git_op, Some(GitOperation::ReceivePack { .. }));

76 if is_receive && !state.nak_sent && Self::wants_section_complete(&state.buffer) {

7277 // Over SSH the request and response share one stream: send-pack

7378 // stops after the wants section and waits for NAK before it sends

7479 // the pack. We only spawn receive-pack after channel EOF, so

@@ -110,7 +115,11 @@

110115

111116 pub async fn channel_close_impl(&mut self, channel: ChannelId) {

112117 tracing::debug!(?channel, "ssh channel closed");

113 self.channels.remove(&channel);

118 if let Some(state) = self.channels.remove(&channel)

119 && let Some(abort) = state.stream_abort

120 {

121 abort.abort();

122 }

114123 }

115124

116125 pub async fn channel_eof_impl(

@@ -118,11 +127,22 @@

118127 channel: ChannelId,

119128 session: &mut Session,

120129 ) -> Result<(), russh::Error> {

121 let Some(state) = self.channels.remove(&channel) else {

130 let Some(state) = self.channels.get_mut(&channel) else {

122131 let _ = session.close(channel);

123132 let _ = session.flush();

124133 return Ok(());

125134 };

135 if state.dispatched {

136 // Half-close only: the git process may still have data to write

137 // back (upload-pack answers with the packfile after `done`), so the

138 // session is torn down when the pump task finishes, not here.

139 if let Some(tx) = state.stdin_tx.take() {

140 drop(tx);

141 }

142 tracing::debug!(?channel, "ssh: client half-closed duplex channel");

143 return Ok(());

144 }

145 let state = self.channels.remove(&channel).expect("channel state");

126146 self.dispatch_state(channel, state, session.handle(), "eof");

127147 Ok(())

128148 }

@@ -353,6 +373,173 @@

353373 }

354374}

355375

376/// Start a duplex git session for `upload-pack` / `upload-archive`: stdin is fed

377/// from `rx` and stdout is streamed back as it is produced. The client keeps the

378/// channel open while it waits for the packfile, so the response cannot be

379/// deferred until EOF the way the push path does.

380pub(crate) fn spawn_streaming_git(

381 channel: russh::ChannelId,

382 state: ChannelState,

383 handle: russh::server::Handle,

384 rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,

385) -> tokio::task::AbortHandle {

386 let op = state.git_op.clone();

387 let service = state.git_op.as_ref().map(|o| o.service_arg().to_string());

388 let timeout_secs = state.rpc_timeout_secs;

389 let task = tokio::spawn(async move {

390 let Some(op) = op else {

391 return;

392 };

393 let result = streaming_git(&op, &handle, channel, rx, timeout_secs).await;

394 match &result {

395 Ok(()) => tracing::debug!(?channel, "ssh streaming git finished"),

396 Err(e) => tracing::warn!(?channel, error = %e, "ssh streaming git failed"),

397 }

398 let code = if result.is_ok() { 0 } else { 1 };

399 _ = handle.exit_status_request(channel, code).await;

400 // Give the client a moment to drain the final bytes before the

401 // channel is torn down (the same race the push path guards against).

402 tokio::time::sleep(std::time::Duration::from_millis(250)).await;

403 _ = handle.eof(channel).await;

404 _ = handle.close(channel).await;

405 });

406 let abort = task.abort_handle();

407 if let Some(name) = service {

408 tracing::debug!(?channel, service = %name, "ssh streaming git started");

409 }

410 abort

411}

412

413/// Drive one duplex git process: pump `rx` into stdin, forward stdout to the

414/// SSH channel, and collect stderr for diagnostics.

415async fn streaming_git(

416 op: &crate::ssh::handler::git_op::GitOperation,

417 handle: &russh::server::Handle,

418 channel: russh::ChannelId,

419 mut rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,

420 timeout_secs: u64,

421) -> Result<(), SshError> {

422 // Stateful, like a real sshd: advertise refs, then answer the request

423 // arriving on stdin. Must not be gated on stdin EOF.

424 let mut cmd = tokio::process::Command::new("git");

425 cmd.current_dir(op.bare()).arg(op.service_arg());

426 cmd.arg(op.bare())

427 .env("GIT_DIR", op.bare())

428 .env("GIT_TERMINAL_PROMPT", "0")

429 .stdin(std::process::Stdio::piped())

430 .stdout(std::process::Stdio::piped())

431 .stderr(std::process::Stdio::piped());

432 apply_hardened_env(&mut cmd);

433

434 cmd.kill_on_drop(true);

435 let mut child = cmd.spawn().map_err(SshError::Io)?;

436 let mut stdin = child

437 .stdin

438 .take()

439 .ok_or_else(|| SshError::Internal("git stdin missing".to_string()))?;

440 let mut stdout = child

441 .stdout

442 .take()

443 .ok_or_else(|| SshError::Internal("git stdout missing".to_string()))?;

444 let stderr = child

445 .stderr

446 .take()

447 .ok_or_else(|| SshError::Internal("git stderr missing".to_string()))?;

448

449 tracing::debug!(

450 service = op.service_arg(),

451 bare = %op.bare().display(),

452 "spawned git process for duplex ssh operation"

453 );

454

455 let writer = tokio::spawn(async move {

456 while let Some(chunk) = rx.recv().await {

457 if stdin.write_all(&chunk).await.is_err() {

458 break;

459 }

460 }

461 let _ = stdin.shutdown().await;

462 });

463

464 let reader = {

465 let handle = handle.clone();

466 tokio::spawn(async move {

467 let mut buf = vec![0u8; 64 * 1024];

468 loop {

469 match stdout.read(&mut buf).await {

470 Ok(0) => break,

471 Ok(n) => {

472 if handle

473 .data(channel, Bytes::copy_from_slice(&buf[..n]))

474 .await

475 .is_err()

476 {

477 break;

478 }

479 }

480 Err(e) => {

481 tracing::warn!(?channel, error = %e, "ssh: git stdout read failed");

482 break;

483 }

484 }

485 }

486 })

487 };

488

489 let stderr_task = tokio::spawn(async move {

490 let mut buf = Vec::new();

491 let _ = stderr

492 .take(MAX_GIT_OUTPUT_BUF as u64 + 1)

493 .read_to_end(&mut buf)

494 .await;

495 buf

496 });

497

498 let status = match tokio::time::timeout(

499 std::time::Duration::from_secs(timeout_secs),

500 child.wait(),

501 )

502 .await

503 {

504 Ok(Ok(s)) => s,

505 _ => {

506 let _ = child.start_kill();

507 let _ = child.wait().await;

508 return Err(SshError::Internal(format!(

509 "git {} timed out",

510 op.service_arg()

511 )));

512 }

513 };

514

515 // The writer only stops on client half-close, so awaiting it here would

516 // hang once the process has already finished.

517 let _ = reader.await;

518 writer.abort();

519 let _ = writer.await;

520 let stderr_bytes = stderr_task.await.unwrap_or_default();

521 tracing::debug!(

522 service = op.service_arg(),

523 exit = ?status.code(),

524 stderr_bytes = stderr_bytes.len(),

525 "git process for duplex ssh operation finished"

526 );

527 if !stderr_bytes.is_empty() && !status.success() {

528 let _ = handle

529 .extended_data(channel, 1, Bytes::from(stderr_bytes.clone()))

530 .await;

531 }

532 if status.success() {

533 Ok(())

534 } else {

535 Err(SshError::Internal(format!(

536 "git {} failed: exit {:?}",

537 op.service_arg(),

538 status.code()

539 )))

540 }

541}

542

356543async fn run_git(

357544 op: Option<&crate::ssh::handler::git_op::GitOperation>,

358545 input: &[u8],

@@ -380,6 +567,7 @@

380567 .stderr(std::process::Stdio::piped());

381568 apply_hardened_env(&mut cmd);

382569

570 cmd.kill_on_drop(true);

383571 let mut child = cmd.spawn().map_err(SshError::Io)?;

384572 tracing::debug!(

385573 service = op.service_arg(),

Modified lib/git/src/ssh/handler/command.rs +56 −28 View lib/git/src/ssh/handler/command.rs

@@ -94,29 +94,35 @@

9494 return Ok(());

9595 }

9696

97 let adv = match run_advertise(

98 &bare,

99 service,

100 self.service.smart.git.state.config.git.git_rpc_timeout_secs,

101 )

102 .await

103 {

104 Ok(adv) => adv,

105 Err(e) => {

106 tracing::warn!(

107 service = service.as_subcommand(),

108 bare = %bare.display(),

109 error = %e,

110 "ssh exec: ref advertisement failed"

111 );

112 let msg = format!("fatal: unable to read repository advertisement: {e}");

113 invalid_command(&msg, session, channel).await;

114 return Ok(());

97 // Push is driven from a fully buffered request, so receive-pack only

98 // runs after EOF and its advertisement is produced up front here.

99 // Upload services are duplex and emit the advertisement themselves.

100 let mut adv_len = 0usize;

101 if service.is_write() {

102 let adv = match run_advertise(

103 &bare,

104 service,

105 self.service.smart.git.state.config.git.git_rpc_timeout_secs,

106 )

107 .await

108 {

109 Ok(adv) => adv,

110 Err(e) => {

111 tracing::warn!(

112 service = service.as_subcommand(),

113 bare = %bare.display(),

114 error = %e,

115 "ssh exec: ref advertisement failed"

116 );

117 let msg = format!("fatal: unable to read repository advertisement: {e}");

118 invalid_command(&msg, session, channel).await;

119 return Ok(());

120 }

121 };

122 adv_len = adv.len();

123 if !adv.is_empty() {

124 let _ = session.data(channel, Bytes::from(adv));

115125 }

116 };

117 let adv_len = adv.len();

118 if !adv.is_empty() {

119 let _ = session.data(channel, Bytes::from(adv));

120126 }

121127 let _ = session.channel_success(channel);

122128 let _ = session.flush();

@@ -124,7 +130,7 @@

124130 service = service.as_subcommand(),

125131 repo_id = %ctx.model.id,

126132 adv_bytes = adv_len,

127 "ssh exec: advertisement sent, channel armed"

133 "ssh exec: channel armed"

128134 );

129135

130136 let op = match service {

@@ -132,11 +138,33 @@

132138 GitService::UploadArchive => GitOperation::UploadArchive { bare: bare.clone() },

133139 GitService::ReceivePack => GitOperation::ReceivePack { bare: bare.clone() },

134140 };

135 let state = self.channels.entry(channel).or_default();

136 state.git_op = Some(op);

137 state.repo_id = Some(ctx.model.id);

138 state.space_id = Some(space_id);

139 state.actor_id = actor_id;

141 // Fetch/clone/archive is duplex: waiting for EOF before starting git,

142 // as the push path does, would deadlock against a client that is

143 // waiting for the packfile.

144 let streaming = !op.is_write();

145 {

146 let state = self.channels.entry(channel).or_default();

147 state.git_op = Some(op);

148 state.repo_id = Some(ctx.model.id);

149 state.space_id = Some(space_id);

150 state.actor_id = actor_id;

151 }

152 if streaming {

153 let state = self.channels.remove(&channel).expect("channel state");

154 let mut state = state;

155 state.rpc_timeout_secs = self.service.smart.git.state.config.git.git_rpc_timeout_secs;

156 let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();

157 let abort = crate::ssh::handler::channel::spawn_streaming_git(

158 channel,

159 state,

160 session.handle(),

161 rx,

162 );

163 let state = self.channels.entry(channel).or_default();

164 state.dispatched = true;

165 state.stdin_tx = Some(tx);

166 state.stream_abort = Some(abort);

167 }

140168 let _ = required;

141169 Ok(())

142170 }

Modified lib/git/src/ssh/handler/mod.rs +10 −1 View lib/git/src/ssh/handler/mod.rs

@@ -15,7 +15,7 @@

1515pub mod git_op;

1616

1717mod auth;

18mod channel;

18pub(crate) mod channel;

1919mod command;

2020

2121#[derive(Default)]

@@ -27,6 +27,15 @@

2727 pub actor_id: Option<Uuid>,

2828 pub buffer_overflow: bool,

2929 pub nak_sent: bool,

30 /// Set when a duplex git session (upload-pack / upload-archive) is running

31 /// for this channel; client data is then relayed to its stdin instead of

32 /// being buffered.

33 pub dispatched: bool,

34 pub stdin_tx: Option<tokio::sync::mpsc::UnboundedSender<Vec<u8>>>,

35 /// Abort handle for the task that owns the streaming git child process.

36 pub stream_abort: Option<tokio::task::AbortHandle>,

37 /// Wall-clock budget for the duplex git process, captured at exec time.

38 pub rpc_timeout_secs: u64,

3039}

3140

3241pub struct GitSshHandler {

Modified lib/git/src/sync/ai_task.rs +1 −3 View lib/git/src/sync/ai_task.rs

@@ -5,9 +5,7 @@

55use models::git::git_tags;

66use models::repo::{repo_release, repos};

77use sea_orm::PaginatorTrait;

8use sea_orm::{

9 ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set,

10};

8use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set};

119use time::OffsetDateTime;

1210use uuid::Uuid;

1311