Openhub
Sign in Sign up
samjeston / openhub Public
main
Switch branches/tags
Branches
main
Sam Jeston Working vertical 9aa3c45 · History
532 lines· 17.3 KB ·Rust
Raw
1//! Built-in SSH server for `git push` / `git fetch`.
2//!
3//! Users authenticate with the public keys registered in their settings.
4//! The only commands accepted are `git-upload-pack` (fetch) and
5//! `git-receive-pack` (push); each is mapped to a hosted repository after
6//! checking permissions and run with the system `git`.
7
8use std::collections::HashMap;
9use std::net::SocketAddr;
10use std::path::Path;
11use std::process::Stdio;
12use std::sync::Arc;
13use std::time::Duration;
14
15use anyhow::{Context, Result};
16use rand::RngCore;
17use russh::keys::ssh_key::private::Ed25519Keypair;
18use russh::keys::ssh_key::{HashAlg, LineEnding, PrivateKey, PublicKey};
19use russh::server::{self, Auth, Msg, Server as _, Session};
20use russh::{Channel, ChannelId, MethodKind, MethodSet, Pty, SshId};
21use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
22use tokio::process::Command;
23use tokio::sync::mpsc;
24
25use crate::db::{self, Repo, User};
26use crate::git;
27use crate::state::AppState;
28use crate::sync;
29
30pub async fn serve(state: AppState) -> Result<()> {
31 let key = load_or_create_host_key(&state.config.host_key_path())?;
32 tracing::info!(
33 "SSH host key fingerprint: {}",
34 key.public_key().fingerprint(HashAlg::Sha256)
35 );
36 let mut methods = MethodSet::empty();
37 methods.push(MethodKind::PublicKey);
38 let config = Arc::new(server::Config {
39 server_id: SshId::Standard("SSH-2.0-Openhub".into()),
40 methods,
41 auth_rejection_time: Duration::from_millis(300),
42 auth_rejection_time_initial: Some(Duration::ZERO),
43 keys: vec![key],
44 inactivity_timeout: Some(Duration::from_secs(3600)),
45 keepalive_interval: Some(Duration::from_secs(60)),
46 nodelay: true,
47 ..Default::default()
48 });
49 let addr = state.config.ssh_addr;
50 tracing::info!("SSH server listening on {addr}");
51 let mut server = SshServer { state };
52 server.run_on_address(config, addr).await?;
53 Ok(())
54}
55
56fn load_or_create_host_key(path: &Path) -> Result<PrivateKey> {
57 if path.exists() {
58 return PrivateKey::read_openssh_file(path)
59 .with_context(|| format!("reading SSH host key {}", path.display()));
60 }
61 let mut seed = [0u8; 32];
62 rand::rngs::OsRng.fill_bytes(&mut seed);
63 let key = PrivateKey::from(Ed25519Keypair::from_seed(&seed));
64 let pem = key.to_openssh(LineEnding::LF)?;
65 let mut opts = std::fs::OpenOptions::new();
66 opts.write(true).create_new(true);
67 #[cfg(unix)]
68 std::os::unix::fs::OpenOptionsExt::mode(&mut opts, 0o600);
69 std::io::Write::write_all(&mut opts.open(path)?, pem.as_bytes())?;
70 tracing::info!("generated new SSH host key at {}", path.display());
71 Ok(key)
72}
73
74struct SshServer {
75 state: AppState,
76}
77
78impl server::Server for SshServer {
79 type Handler = SshSession;
80
81 fn new_client(&mut self, peer: Option<SocketAddr>) -> SshSession {
82 SshSession {
83 state: self.state.clone(),
84 peer,
85 user: None,
86 channels: HashMap::new(),
87 }
88 }
89
90 fn handle_session_error(&mut self, error: anyhow::Error) {
91 tracing::debug!("ssh session ended with error: {error:#}");
92 }
93}
94
95#[derive(Default)]
96struct ChannelState {
97 /// Value of GIT_PROTOCOL sent by the client (enables protocol v2).
98 protocol: Option<String>,
99 /// Feeds client data to the running git process. Dropping it closes stdin.
100 stdin: Option<mpsc::UnboundedSender<Vec<u8>>>,
101}
102
103struct SshSession {
104 state: AppState,
105 peer: Option<SocketAddr>,
106 user: Option<User>,
107 channels: HashMap<ChannelId, ChannelState>,
108}
109
110#[derive(Debug, Clone, Copy, PartialEq, Eq)]
111enum Service {
112 UploadPack,
113 ReceivePack,
114}
115
116impl Service {
117 fn subcommand(self) -> &'static str {
118 match self {
119 Service::UploadPack => "upload-pack",
120 Service::ReceivePack => "receive-pack",
121 }
122 }
123}
124
125/// Parses `git-upload-pack '/owner/repo.git'` and similar into the service
126/// and the (owner, repo) pair.
127fn parse_command(command: &str) -> Option<(Service, String, String)> {
128 let command = command.trim();
129 let (verb, rest) = command.split_once(' ')?;
130 let (service, arg) = match verb {
131 "git-upload-pack" => (Service::UploadPack, rest),
132 "git-receive-pack" => (Service::ReceivePack, rest),
133 "git" => {
134 let (sub, arg) = rest.trim_start().split_once(' ')?;
135 match sub {
136 "upload-pack" => (Service::UploadPack, arg),
137 "receive-pack" => (Service::ReceivePack, arg),
138 _ => return None,
139 }
140 }
141 _ => return None,
142 };
143 let arg = arg.trim();
144 let unquoted = arg
145 .strip_prefix('\'')
146 .and_then(|a| a.strip_suffix('\''))
147 .or_else(|| arg.strip_prefix('"').and_then(|a| a.strip_suffix('"')))
148 .unwrap_or(arg);
149 if unquoted.contains(['\'', '"', '\\', '\0']) {
150 return None;
151 }
152 let path = unquoted.trim_matches('/');
153 let path = path.strip_suffix(".git").unwrap_or(path);
154 let (owner, name) = path.split_once('/')?;
155 if owner.is_empty() || name.is_empty() || name.contains('/') {
156 return None;
157 }
158 Some((service, owner.to_string(), name.to_string()))
159}
160
161fn branch_tips(path: &Path) -> HashMap<String, String> {
162 let Ok(repo) = git::open(path) else {
163 return HashMap::new();
164 };
165 git::branch_names(&repo)
166 .unwrap_or_default()
167 .into_iter()
168 .filter_map(|b| git::branch_tip(&repo, &b).map(|t| (b, t.to_string())))
169 .collect()
170}
171
172async fn forward<R, F, Fut>(mut reader: R, mut send: F)
173where
174 R: AsyncRead + Unpin,
175 F: FnMut(Vec<u8>) -> Fut,
176 Fut: std::future::Future<Output = bool>,
177{
178 let mut buf = vec![0u8; 32 * 1024];
179 loop {
180 match reader.read(&mut buf).await {
181 Ok(0) | Err(_) => break,
182 Ok(n) => {
183 if !send(buf[..n].to_vec()).await {
184 break;
185 }
186 }
187 }
188 }
189}
190
191impl SshSession {
192 fn reject(session: &mut Session, channel: ChannelId, message: &str) -> Result<()> {
193 session.channel_success(channel)?;
194 session.extended_data(channel, 1, format!("ERROR: {message}\n").into_bytes())?;
195 session.exit_status_request(channel, 1)?;
196 session.eof(channel)?;
197 session.close(channel)?;
198 Ok(())
199 }
200}
201
202/// Lines shown after a push, pointing at pull requests for pushed branches.
203async fn push_hints(
204 state: &AppState,
205 repo: &Repo,
206 before: &HashMap<String, String>,
207 after: &HashMap<String, String>,
208) -> Vec<String> {
209 let base = format!("{}/{}/{}", state.config.public_url, repo.owner, repo.name);
210 let default = db::repo_by_name(&state.db, &repo.owner, &repo.name)
211 .await
212 .ok()
213 .flatten()
214 .map(|r| r.default_branch)
215 .unwrap_or_else(|| repo.default_branch.clone());
216 let open = db::list_pulls(&state.db, repo.id, true)
217 .await
218 .unwrap_or_default();
219 let mut lines = Vec::new();
220 let mut updated: Vec<&String> = after
221 .iter()
222 .filter(|(b, tip)| before.get(*b) != Some(*tip) && **b != default)
223 .map(|(b, _)| b)
224 .collect();
225 updated.sort();
226 for branch in updated {
227 lines.push(String::new());
228 match open.iter().find(|p| &p.head_branch == branch) {
229 Some(pr) => {
230 lines.push(format!("View pull request for '{branch}':"));
231 lines.push(format!(" {base}/pull/{}", pr.number));
232 }
233 None => {
234 lines.push(format!(
235 "Create a pull request for '{branch}' on Openhub by visiting:"
236 ));
237 lines.push(format!(
238 " {base}/compare/{}...{}",
239 crate::web::fmt::urlpath(&default),
240 crate::web::fmt::urlpath(branch)
241 ));
242 }
243 }
244 }
245 if !lines.is_empty() {
246 lines.push(String::new());
247 }
248 lines
249}
250
251impl server::Handler for SshSession {
252 type Error = anyhow::Error;
253
254 async fn auth_publickey_offered(&mut self, _user: &str, key: &PublicKey) -> Result<Auth> {
255 let fingerprint = key.fingerprint(HashAlg::Sha256).to_string();
256 Ok(
257 if db::fingerprint_exists(&self.state.db, &fingerprint).await? {
258 Auth::Accept
259 } else {
260 Auth::reject()
261 },
262 )
263 }
264
265 async fn auth_publickey(&mut self, _user: &str, key: &PublicKey) -> Result<Auth> {
266 let fingerprint = key.fingerprint(HashAlg::Sha256).to_string();
267 match db::user_by_key_fingerprint(&self.state.db, &fingerprint).await? {
268 Some((user, key_id)) => {
269 let _ = db::touch_key(&self.state.db, key_id).await;
270 tracing::info!(
271 "ssh: {} authenticated with key {fingerprint} from {:?}",
272 user.username,
273 self.peer
274 );
275 self.user = Some(user);
276 Ok(Auth::Accept)
277 }
278 None => Ok(Auth::reject()),
279 }
280 }
281
282 async fn channel_open_session(
283 &mut self,
284 channel: Channel<Msg>,
285 reply: server::ChannelOpenHandle,
286 _session: &mut Session,
287 ) -> Result<()> {
288 self.channels.insert(channel.id(), ChannelState::default());
289 // Data arrives through the `data` callback; dropping the channel
290 // handle keeps russh from also buffering it there.
291 drop(channel);
292 reply.accept().await;
293 Ok(())
294 }
295
296 async fn env_request(
297 &mut self,
298 channel: ChannelId,
299 name: &str,
300 value: &str,
301 session: &mut Session,
302 ) -> Result<()> {
303 if name == "GIT_PROTOCOL" {
304 if let Some(state) = self.channels.get_mut(&channel) {
305 state.protocol = Some(value.to_string());
306 }
307 session.channel_success(channel)?;
308 } else {
309 session.channel_failure(channel)?;
310 }
311 Ok(())
312 }
313
314 async fn pty_request(
315 &mut self,
316 channel: ChannelId,
317 _term: &str,
318 _col_width: u32,
319 _row_height: u32,
320 _pix_width: u32,
321 _pix_height: u32,
322 _modes: &[(Pty, u32)],
323 session: &mut Session,
324 ) -> Result<()> {
325 session.channel_failure(channel)?;
326 Ok(())
327 }
328
329 async fn shell_request(&mut self, channel: ChannelId, session: &mut Session) -> Result<()> {
330 let name = self
331 .user
332 .as_ref()
333 .map(|u| u.username.as_str())
334 .unwrap_or("there");
335 session.channel_success(channel)?;
336 let msg = format!(
337 "Hi {name}! You've successfully authenticated, but Openhub does not provide shell access.\r\n"
338 );
339 session.extended_data(channel, 1, msg.into_bytes())?;
340 session.exit_status_request(channel, 1)?;
341 session.eof(channel)?;
342 session.close(channel)?;
343 Ok(())
344 }
345
346 async fn subsystem_request(
347 &mut self,
348 channel: ChannelId,
349 _name: &str,
350 session: &mut Session,
351 ) -> Result<()> {
352 session.channel_failure(channel)?;
353 Ok(())
354 }
355
356 async fn exec_request(
357 &mut self,
358 channel: ChannelId,
359 data: &[u8],
360 session: &mut Session,
361 ) -> Result<()> {
362 let Some(user) = self.user.clone() else {
363 return Self::reject(session, channel, "Not authenticated.");
364 };
365 let command = String::from_utf8_lossy(data).to_string();
366 let Some((service, owner, name)) = parse_command(&command) else {
367 tracing::info!(
368 "ssh: {} sent unsupported command {command:?}",
369 user.username
370 );
371 return Self::reject(
372 session,
373 channel,
374 "Openhub only supports git fetch and push over SSH.",
375 );
376 };
377 let repo = db::repo_by_name(&self.state.db, &owner, &name).await?;
378 let access = match &repo {
379 Some(r) => self.state.access(r, Some(&user)).await?,
380 None => Default::default(),
381 };
382 let Some(repo) = repo.filter(|_| access.read) else {
383 return Self::reject(session, channel, "Repository not found.");
384 };
385 if service == Service::ReceivePack && !access.write {
386 let msg = format!(
387 "You do not have permission to push to {}/{}.",
388 repo.owner, repo.name
389 );
390 return Self::reject(session, channel, &msg);
391 }
392
393 let path = self.state.config.repo_path(repo.id);
394 let before = if service == Service::ReceivePack {
395 let p = path.clone();
396 tokio::task::spawn_blocking(move || branch_tips(&p)).await?
397 } else {
398 HashMap::new()
399 };
400
401 let mut cmd = Command::new(&self.state.config.git_bin);
402 cmd.arg(service.subcommand())
403 .arg(&path)
404 .stdin(Stdio::piped())
405 .stdout(Stdio::piped())
406 .stderr(Stdio::piped())
407 .kill_on_drop(true);
408 if let Some(protocol) = self.channels.get(&channel).and_then(|c| c.protocol.clone()) {
409 cmd.env("GIT_PROTOCOL", protocol);
410 }
411 let mut child = match cmd.spawn() {
412 Ok(c) => c,
413 Err(e) => {
414 tracing::error!("could not start git: {e}");
415 return Self::reject(session, channel, "Internal server error.");
416 }
417 };
418 tracing::info!(
419 "ssh: {} running {} on {}/{}",
420 user.username,
421 service.subcommand(),
422 repo.owner,
423 repo.name
424 );
425 session.channel_success(channel)?;
426
427 let (tx, mut rx) = mpsc::unbounded_channel::<Vec<u8>>();
428 self.channels.entry(channel).or_default().stdin = Some(tx);
429 let mut stdin = child.stdin.take().context("git stdin")?;
430 let stdout = child.stdout.take().context("git stdout")?;
431 let stderr = child.stderr.take().context("git stderr")?;
432
433 tokio::spawn(async move {
434 while let Some(buf) = rx.recv().await {
435 if stdin.write_all(&buf).await.is_err() {
436 break;
437 }
438 }
439 });
440
441 let handle = session.handle();
442 let state = self.state.clone();
443 tokio::spawn(async move {
444 let out = forward(stdout, |buf| {
445 let h = handle.clone();
446 async move { h.data(channel, buf).await.is_ok() }
447 });
448 let err = forward(stderr, |buf| {
449 let h = handle.clone();
450 async move { h.extended_data(channel, 1, buf).await.is_ok() }
451 });
452 tokio::join!(out, err);
453 let code = match child.wait().await {
454 Ok(status) => status.code().unwrap_or(1),
455 Err(_) => 1,
456 };
457 if service == Service::ReceivePack && code == 0 {
458 if let Err(e) = sync::after_push(&state, &repo, user.id).await {
459 tracing::error!(
460 "post-push sync failed for {}/{}: {e:#}",
461 repo.owner,
462 repo.name
463 );
464 }
465 let p = path.clone();
466 let after = tokio::task::spawn_blocking(move || branch_tips(&p))
467 .await
468 .unwrap_or_default();
469 for line in push_hints(&state, &repo, &before, &after).await {
470 let text = format!("remote: {line}\n");
471 let _ = handle.extended_data(channel, 1, text.into_bytes()).await;
472 }
473 }
474 let _ = handle.exit_status_request(channel, code as u32).await;
475 let _ = handle.eof(channel).await;
476 let _ = handle.close(channel).await;
477 });
478 Ok(())
479 }
480
481 async fn data(
482 &mut self,
483 channel: ChannelId,
484 data: &[u8],
485 _session: &mut Session,
486 ) -> Result<()> {
487 if let Some(tx) = self.channels.get(&channel).and_then(|c| c.stdin.as_ref()) {
488 let _ = tx.send(data.to_vec());
489 }
490 Ok(())
491 }
492
493 async fn channel_eof(&mut self, channel: ChannelId, _session: &mut Session) -> Result<()> {
494 if let Some(state) = self.channels.get_mut(&channel) {
495 state.stdin = None;
496 }
497 Ok(())
498 }
499
500 async fn channel_close(&mut self, channel: ChannelId, _session: &mut Session) -> Result<()> {
501 self.channels.remove(&channel);
502 Ok(())
503 }
504}
505
506#[cfg(test)]
507mod tests {
508 use super::*;
509
510 #[test]
511 fn parses_git_commands() {
512 assert_eq!(
513 parse_command("git-upload-pack '/alice/project.git'"),
514 Some((Service::UploadPack, "alice".into(), "project".into()))
515 );
516 assert_eq!(
517 parse_command("git-receive-pack 'alice/project.git'"),
518 Some((Service::ReceivePack, "alice".into(), "project".into()))
519 );
520 assert_eq!(
521 parse_command("git upload-pack 'alice/project'"),
522 Some((Service::UploadPack, "alice".into(), "project".into()))
523 );
524 assert_eq!(
525 parse_command("git-upload-archive 'alice/project.git'"),
526 None
527 );
528 assert_eq!(parse_command("rm -rf /"), None);
529 assert_eq!(parse_command("git-upload-pack 'a/b/c.git'"), None);
530 assert_eq!(parse_command("git-upload-pack 'a/b'; ls'"), None);
531 }
532}