refactor(gitssh): refresh crate across handler, server, service

Modified lib/gitssh/handler/auth.rs +1 −5 View lib/gitssh/handler/auth.rs

@@ -16,11 +16,7 @@

1616}

1717

1818impl GitSshHandler {

19 /// Find the user whose stored SSH key matches the presented public key.

20 /// The workspace stores OpenSSH public key text (content column) and a

21 /// sha256_hex(content bytes) fingerprint; gitssh must parse the stored

22 /// content and compare with the presented key directly.

23 async fn find_user_by_public_key(

19 async fn find_user_by_public_key(

2420 &self,

2521 username: &str,

2622 public_key: &ssh_key::PublicKey,

Modified lib/gitssh/handler/channel.rs +81 −44 View lib/gitssh/handler/channel.rs

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

55use crate::handler::GitSshHandler;

66

77impl GitSshHandler {

8 /// Append incoming client data to the per-channel buffer. All buffered

9 /// bytes are flushed to the git subprocess in a single chunk on EOF.

10 pub async fn data_impl(&mut self, channel: ChannelId, data: &[u8]) -> Result<(), russh::Error> {

8 pub async fn data_impl(&mut self, channel: ChannelId, data: &[u8]) -> Result<(), russh::Error> {

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

1210 state.buffer.extend_from_slice(data);

1311 Ok(())

@@ -17,9 +15,7 @@

1715 self.channels.remove(&channel);

1816 }

1917

20 /// On EOF, dispatch the buffered bytes to the git subprocess

21 /// (asynchronously) and send the result back over the channel.

22 pub async fn channel_eof_impl(

18 pub async fn channel_eof_impl(

2319 &mut self,

2420 channel: ChannelId,

2521 session: &mut Session,

@@ -31,11 +27,22 @@

3127 };

3228 let op = state.git_op.take();

3329 let buffer = std::mem::take(&mut state.buffer);

30 let repo_id = state.repo_id;

31 let channel_id = state.channel_id;

32 let actor_id = state.actor_id;

3433 let push_coordinator = self.service.push_coordinator();

34 let queue = self.service.queue.clone();

35 let sync_settings = self.service.sync_settings.clone();

3536

3637 let handle = session.handle();

3738 tokio::spawn(async move {

3839 let is_write = op.as_ref().is_some_and(|o| o.is_write());

40 if is_write

41 && let Some(bare) = op.as_ref().map(|o| o.bare().to_path_buf())

42 && let Err(e) = gitsync::prepare_post_receive(&bare)

43 {

44 tracing::warn!(?e, "post-receive hook setup failed");

45 }

3946 let push_key = op

4047 .as_ref()

4148 .map(|o| format!("git_op:{}", o.bare().display()))

@@ -53,6 +60,32 @@

5360 } else {

5461 run.await

5562 };

63 if result.is_ok()

64 && is_write

65 && let (Some(repo_id), Some(channel_id), Some(bare)) = (

66 repo_id,

67 channel_id,

68 op.as_ref().map(|o| o.bare().to_path_buf()),

69 )

70 && let Some(queue) = &queue

71 {

72 match gitsync::enqueue_post_receive_push(

73 queue,

74 &sync_settings,

75 channel_id,

76 repo_id,

77 actor_id,

78 &bare,

79 )

80 .await

81 {

82 Ok(n) if n > 0 => {

83 tracing::debug!(count = n, "enqueued post-receive ref updates")

84 }

85 Ok(_) => tracing::debug!("no refs to enqueue"),

86 Err(e) => tracing::warn!(?e, "post-receive enqueue failed"),

87 }

88 }

5689 let code = match &result {

5790 Ok(_) => 0,

5891 Err(_) => 1,

@@ -83,8 +116,6 @@

83116 }

84117}

85118

86/// Run the git subprocess once: write the buffered input, read stdout, and

87/// forward it back to the client.

88119async fn run_git(

89120 op: Option<&crate::handler::git_op::GitOperation>,

90121 input: &[u8],

@@ -94,44 +125,50 @@

94125 let Some(op) = op else {

95126 return Ok(());

96127 };

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

98 cmd.current_dir(op.bare())

99 .arg(op.service_arg())

100 .arg("--stateless-rpc")

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

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

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

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

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

128 let _timer = crate::metrics::SubprocessTimer::start();

129 let result: Result<(), crate::error::SshError> = async {

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

131 cmd.current_dir(op.bare())

132 .arg(op.service_arg())

133 .arg("--stateless-rpc")

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

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

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

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

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

106139

107 let mut child = cmd.spawn().map_err(crate::error::SshError::Io)?;

108 use tokio::io::AsyncWriteExt;

109 if let Some(mut stdin) = child.stdin.take() {

110 stdin

111 .write_all(input)

140 let mut child = cmd.spawn().map_err(crate::error::SshError::Io)?;

141 use tokio::io::AsyncWriteExt;

142 if let Some(mut stdin) = child.stdin.take() {

143 stdin

144 .write_all(input)

145 .await

146 .map_err(crate::error::SshError::Io)?;

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

148 }

149 let output = child

150 .wait_with_output()

112151 .await

113152 .map_err(crate::error::SshError::Io)?;

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

115 }

116 let output = child

117 .wait_with_output()

118 .await

119 .map_err(crate::error::SshError::Io)?;

120 if !output.stdout.is_empty() {

121 let _ = handle.data(channel, Bytes::from(output.stdout)).await;

122 }

123 if !output.stderr.is_empty() && !output.status.success() {

124 let _ = handle

125 .extended_data(channel, 1, Bytes::from(output.stderr))

126 .await;

127 }

128 if output.status.success() {

129 Ok(())

130 } else {

131 Err(crate::error::SshError::Internal(format!(

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

133 op.service_arg(),

134 output.status.code()

135 )))

153 if !output.stdout.is_empty() {

154 let _ = handle.data(channel, Bytes::from(output.stdout)).await;

155 }

156 if !output.stderr.is_empty() && !output.status.success() {

157 let _ = handle

158 .extended_data(channel, 1, Bytes::from(output.stderr))

159 .await;

160 }

161 if output.status.success() {

162 Ok(())

163 } else {

164 Err(crate::error::SshError::Internal(format!(

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

166 op.service_arg(),

167 output.status.code()

168 )))

169 }

136170 }

171 .await;

172 crate::metrics::record_subprocess(op.service_arg(), result.is_ok());

173 result

137174}

Modified lib/gitssh/handler/command.rs +71 −69 View lib/gitssh/handler/command.rs

@@ -17,80 +17,84 @@

1717 data: &[u8],

1818 session: &mut Session,

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

20 let cmd = std::str::from_utf8(data)

21 .map_err(|_| invalid_command_error("Invalid command encoding", session, channel))?;

22 let cmd = cmd.trim();

23 tracing::info!(

24 "ssh exec from {:?}: {}",

25 self.client_addr,

26 sanitize_for_log(cmd)

27 );

20 let mut service_label: &'static str = "unknown";

21 let result: Result<(), russh::Error> = async {

22 let cmd = std::str::from_utf8(data)

23 .map_err(|_| invalid_command_error("Invalid command encoding", session, channel))?;

24 let cmd = cmd.trim();

25 tracing::info!(

26 "ssh exec from {:?}: {}",

27 self.client_addr,

28 sanitize_for_log(cmd)

29 );

2830

29 let Some((service, raw_path)) = parse_git_command(cmd) else {

30 return Err(invalid_command_error(

31 "Invalid git command",

32 session,

33 channel,

34 ));

35 };

36 let Some((space, repo)) = parse_repo_path(raw_path) else {

37 return Err(invalid_command_error(

38 "Invalid repository path",

39 session,

40 channel,

41 ));

42 };

31 let Some((service, raw_path)) = parse_git_command(cmd) else {

32 return Err(invalid_command_error(

33 "Invalid git command",

34 session,

35 channel,

36 ));

37 };

38 let Some((space, repo)) = parse_repo_path(raw_path) else {

39 return Err(invalid_command_error(

40 "Invalid repository path",

41 session,

42 channel,

43 ));

44 };

45 service_label = service.as_subcommand();

46

47 let actor = match &self.operator {

48 Some(u) => ActorContext::for_user(u.uid),

49 None => ActorContext::anonymous(),

50 };

51 let required = if service.is_write() {

52 RepoAccess::Write

53 } else {

54 RepoAccess::Read

55 };

56 let resolved: ResolvedRepo = match self

57 .service

58 .auth

59 .authorize(space, repo, &actor, required)

60 .await

61 {

62 Ok(r) => r,

63 Err(e) => {

64 let msg = auth_error_message(&e);

65 tracing::warn!("ssh {service:?} {space}/{repo} denied: {msg}");

66 return Err(invalid_command_error(msg.as_str(), session, channel));

67 }

68 };

4369

44 let actor = match &self.operator {

45 Some(u) => ActorContext::for_user(u.uid),

46 None => ActorContext::anonymous(),

47 };

48 let required = if service.is_write() {

49 RepoAccess::Write

50 } else {

51 RepoAccess::Read

52 };

53 let resolved: ResolvedRepo = match self

54 .service

55 .auth

56 .authorize(space, repo, &actor, required)

57 .await

58 {

59 Ok(r) => r,

60 Err(e) => {

61 let msg = auth_error_message(&e);

62 tracing::warn!("ssh {service:?} {space}/{repo} denied: {msg}");

70 let bare = self.service.bare_path(&resolved);

71 if !bare.is_dir() {

72 let msg = format!("fatal: repository '{space}/{repo}' does not exist");

6373 return Err(invalid_command_error(msg.as_str(), session, channel));

6474 }

65 };

6675

67 let bare = self.service.bare_path(&resolved);

68 if !bare.is_dir() {

69 let msg = format!("fatal: repository '{space}/{repo}' does not exist");

70 return Err(invalid_command_error(msg.as_str(), session, channel));

71 }

76 let adv = run_advertise(&bare, service).await.unwrap_or_default();

77 if !adv.is_empty() {

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

79 }

80 let _ = session.channel_success(channel);

81 let _ = session.flush();

82

83 let op = match service {

84 GitService::UploadPack => GitOperation::UploadPack { bare: bare.clone() },

85 _ => GitOperation::ReceivePack { bare: bare.clone() },

86 };

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

88 state.git_op = Some(op);

89 state.repo_id = Some(resolved.repo.id);

90 state.channel_id = Some(resolved.channel.id);

91 state.actor_id = actor.user_id;

7292

73 // Run `git <service> --advertise-refs` to produce the initial advertisement

74 // and send it to the client. Then channel_success tells the client it can

75 // begin sending its half of the protocol. The actual request bytes are

76 // buffered in `state.buffer` and flushed to a one-shot git subprocess on EOF.

77 // Advertise failure is non-fatal — we still send channel_success so the

78 // client can proceed; the real work happens on channel_eof.

79 let adv = run_advertise(&bare, service).await.unwrap_or_default();

80 if !adv.is_empty() {

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

93 Ok(())

8294 }

83 let _ = session.channel_success(channel);

84 let _ = session.flush();

85

86 let op = match service {

87 GitService::UploadPack => GitOperation::UploadPack { bare: bare.clone() },

88 _ => GitOperation::ReceivePack { bare: bare.clone() },

89 };

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

91 state.git_op = Some(op);

92

93 Ok(())

95 .await;

96 crate::metrics::record_exec(service_label, result.is_ok());

97 result

9498 }

9599}

96100

@@ -116,7 +120,6 @@

116120 }

117121}

118122

119/// Run `git <service> --advertise-refs` and return its stdout bytes.

120123async fn run_advertise(

121124 bare: &std::path::PathBuf,

122125 service: GitService,

@@ -130,7 +133,6 @@

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

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

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

133 // Drain stderr concurrently to avoid a deadlock on a full pipe.

134136 let stderr_task = if let Some(stderr) = child.stderr.take() {

135137 Some(tokio::spawn(async move {

136138 use tokio::io::AsyncReadExt;

Modified lib/gitssh/handler/mod.rs +4 −1 View lib/gitssh/handler/mod.rs

@@ -6,6 +6,7 @@

66use russh::server::{Auth, Msg, Session};

77use russh::{Channel, ChannelId};

88use tokio::sync::OwnedSemaphorePermit;

9use uuid::Uuid;

910

1011use db::user::users;

1112

@@ -23,6 +24,9 @@

2324pub struct ChannelState {

2425 pub buffer: Vec<u8>,

2526 pub git_op: Option<GitOperation>,

27 pub repo_id: Option<Uuid>,

28 pub channel_id: Option<Uuid>,

29 pub actor_id: Option<Uuid>,

2630}

2731

2832pub struct GitSshHandler {

@@ -197,6 +201,5 @@

197201 }

198202}

199203

200/// Unused PathBuf import kept for symmetry with sibling modules that may grow.

201204#[allow(dead_code)]

202205type _PathBuf = PathBuf;

Modified lib/gitssh/tests/ssh_roundtrip.rs +0 −7 View lib/gitssh/tests/ssh_roundtrip.rs

@@ -262,7 +262,6 @@

262262 )

263263 .await;

264264

265 // Generate client ed25519 keypair and register the public key.

266265 let mut rng = rand::rng();

267266 let client_key = russh::keys::PrivateKey::random(&mut rng, russh::keys::Algorithm::Ed25519)

268267 .expect("client key");

@@ -272,7 +271,6 @@

272271 .expect("client pub openssh");

273272 insert_ssh_key(&conn, alice, &client_pub).await;

274273

275 // Write client private key for ssh.

276274 let client_key_path = root.join("id_ed25519");

277275 let client_key_pem = client_key

278276 .to_openssh(ssh_key::LineEnding::LF)

@@ -285,7 +283,6 @@

285283 std::fs::create_dir_all(bare.parent().expect("bare parent")).expect("bare dirs");

286284 git(&root, &["init", "--bare", bare.to_str().expect("bare str")]).await;

287285

288 // Seed the bare repo with an initial commit via a throwaway working tree.

289286 let seed_dir = root.join("seed");

290287 std::fs::create_dir_all(&seed_dir).expect("seed dir");

291288 git(

@@ -319,7 +316,6 @@

319316 .await;

320317 git(&seed_dir, &["push", "-u", "origin", "main"]).await;

321318

322 // Host key for the server.

323319 let host_key = generate_host_key().expect("host key");

324320 let host_key_path = root.join("host_key");

325321 let host_key_pem = host_key

@@ -390,7 +386,6 @@

390386 let server = start_server().await;

391387 let url = format!("ssh://alice@127.0.0.1:{}/acme/widget.git", server.port);

392388

393 // Clone (upload-pack).

394389 let clone_dir = server._tmp.path().join("clone");

395390 git_ssh(

396391 server._tmp.path(),

@@ -399,7 +394,6 @@

399394 )

400395 .await;

401396

402 // Make a commit and push (receive-pack).

403397 std::fs::write(clone_dir.join("hello.txt"), "hello ssh\n").expect("write file");

404398 git(&clone_dir, &["add", "hello.txt"]).await;

405399 git(

@@ -422,7 +416,6 @@

422416 )

423417 .await;

424418

425 // Verify the object landed in the bare repo.

426419 let bare = server

427420 .git_root

428421 .join(format!("{}/{}.git", server.channel_id, server.repo_id));

Modified lib/gitssh/git_service.rs +0 −6 View lib/gitssh/git_service.rs

@@ -36,8 +36,6 @@

3636 }

3737}

3838

39/// Parse a git-over-ssh exec command like `git-upload-pack 'acme/widget.git'`.

40/// Returns the service and the raw (single-quoted) path.

4139pub fn parse_git_command(cmd: &str) -> Option<(GitService, &str)> {

4240 let (bin, path) = cmd.split_once(' ')?;

4341 let service = match bin {

@@ -53,8 +51,6 @@

5351 s.trim().trim_matches('\'')

5452}

5553

56/// Parse a repository path into (owner, repo). Both segments must be non-empty.

57/// Enforces a strict character gate to block path traversal and shell metacharacters.

5854pub fn parse_repo_path(path: &str) -> Option<(&str, &str)> {

5955 let path = strip_apostrophes(path).trim_matches('/');

6056 if path.is_empty() || path.len() > MAX_REPO_PATH_LEN {

@@ -81,7 +77,6 @@

8177 Some((owner, repo))

8278}

8379

84/// Sanitize untrusted input for safe logging.

8580pub fn sanitize_for_log(input: &str) -> String {

8681 let mut out = String::with_capacity(input.len().min(256));

8782 for &b in input.as_bytes().iter().take(256) {

@@ -94,7 +89,6 @@

9489 out

9590}

9691

97/// Build the git subprocess command for a bare repository dir.

9892pub fn build_git_command(service: GitService, bare: &PathBuf) -> tokio::process::Command {

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

10094 cmd.current_dir(bare)

Modified lib/gitssh/lib.rs +1 −0 View lib/gitssh/lib.rs

@@ -2,6 +2,7 @@

22pub mod error;

33pub mod git_service;

44mod handler;

5pub mod metrics;

56pub mod server;

67pub mod service;

78

Added lib/gitssh/metrics.rs +87 −0 View lib/gitssh/metrics.rs

@@ -0,0 +1,87 @@

1use std::sync::Once;

2use std::time::Instant;

3

4use metrics::{Unit, counter, describe_counter, describe_histogram, histogram};

5

6pub const OUTCOME_SUCCESS: &str = "success";

7pub const OUTCOME_ERROR: &str = "error";

8pub const OUTCOME_ACQUIRED: &str = "acquired";

9pub const OUTCOME_REJECTED: &str = "rejected";

10

11static DESCRIBE_ONCE: Once = Once::new();

12

13pub fn ensure_described() {

14 DESCRIBE_ONCE.call_once(describe_all);

15}

16

17fn describe_all() {

18 describe_counter!(

19 "gitssh_sessions_total",

20 "SSH client sessions partitioned by outcome (acquired, rejected) when the session semaphore is granted or full."

21 );

22 describe_counter!(

23 "gitssh_exec_total",

24 "SSH exec requests partitioned by git service and outcome (success, error)."

25 );

26 describe_counter!(

27 "gitssh_subprocess_total",

28 "Git subprocess runs partitioned by git service and status (success, error)."

29 );

30 describe_histogram!(

31 "gitssh_subprocess_duration_seconds",

32 Unit::Seconds,

33 "Duration of git subprocess runs."

34 );

35}

36

37pub(crate) fn outcome(ok: bool) -> &'static str {

38 if ok { OUTCOME_SUCCESS } else { OUTCOME_ERROR }

39}

40

41pub(crate) fn record_session(acquired: bool) {

42 ensure_described();

43 counter!(

44 "gitssh_sessions_total",

45 "outcome" => if acquired { OUTCOME_ACQUIRED } else { OUTCOME_REJECTED }

46 )

47 .increment(1);

48}

49

50pub(crate) fn record_exec(service: &'static str, ok: bool) {

51 ensure_described();

52 counter!(

53 "gitssh_exec_total",

54 "service" => service,

55 "outcome" => outcome(ok)

56 )

57 .increment(1);

58}

59

60pub(crate) fn record_subprocess(service: &'static str, ok: bool) {

61 ensure_described();

62 counter!(

63 "gitssh_subprocess_total",

64 "service" => service,

65 "status" => outcome(ok)

66 )

67 .increment(1);

68}

69

70pub(crate) struct SubprocessTimer {

71 start: Instant,

72}

73

74impl SubprocessTimer {

75 pub(crate) fn start() -> Self {

76 ensure_described();

77 Self {

78 start: Instant::now(),

79 }

80 }

81}

82

83impl Drop for SubprocessTimer {

84 fn drop(&mut self) {

85 histogram!("gitssh_subprocess_duration_seconds").record(self.start.elapsed());

86 }

87}

Modified lib/gitssh/server.rs +1 −3 View lib/gitssh/server.rs

@@ -32,6 +32,7 @@

3232

3333 fn new_client(&mut self, _peer_addr: Option<SocketAddr>) -> Self::Handler {

3434 let permit = self.session_slots.clone().try_acquire_owned().ok();

35 crate::metrics::record_session(permit.is_some());

3536 GitSshHandler::new(self.service.clone(), _peer_addr, permit)

3637 }

3738

@@ -78,7 +79,6 @@

7879 })

7980}

8081

81/// Run the SSH server on the configured address. Never returns normally.

8282pub async fn run_ssh(service: GitSshService) -> SshResult<()> {

8383 let bind_addr = service.config.bind_addr.clone();

8484 let port = service.config.port;

@@ -91,8 +91,6 @@

9191 Ok(())

9292}

9393

94/// Run the SSH server on an already-bound listener. The future owns the

95/// listener and server, so it can be spawned directly. Never returns normally.

9694pub async fn serve_on_listener(

9795 service: GitSshService,

9896 listener: tokio::net::TcpListener,

Modified lib/gitssh/service.rs +14 −0 View lib/gitssh/service.rs

@@ -12,6 +12,8 @@

1212 pub auth: GitAuthService,

1313 pub config: GitSshConfig,

1414 pub kv: Option<Arc<RedisCache>>,

15 pub queue: Option<Arc<queue::Queue>>,

16 pub sync_settings: gitsync::SyncSettings,

1517}

1618

1719impl GitSshService {

@@ -20,6 +22,8 @@

2022 auth,

2123 config,

2224 kv: None,

25 queue: None,

26 sync_settings: gitsync::SyncSettings::default(),

2327 }

2428 }

2529

@@ -28,6 +32,16 @@

2832 self

2933 }

3034

35 pub fn with_queue(mut self, queue: Option<Arc<queue::Queue>>) -> Self {

36 self.queue = queue;

37 self

38 }

39

40 pub fn with_sync_settings(mut self, settings: gitsync::SyncSettings) -> Self {

41 self.sync_settings = settings;

42 self

43 }

44

3145 pub fn push_coordinator(&self) -> PushCoordinator {

3246 match &self.kv {

3347 Some(kv) => {