fabrica

hanna/fabrica

main
Switch branch
mainView all branches
Clone
HTTPS
SSH
18550 bytes
Raw
1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4
5//! The russh server: publickey auth, exec dispatch into pack processes.
6
7use std::sync::Arc;
8use std::time::Duration;
9
10use config::Config;
11use model::KeyKind;
12use russh::keys::{HashAlg, PublicKey};
13use russh::server::{Auth, ChannelOpenHandle, Handler, Msg, Server as _, Session};
14use russh::{Channel, ChannelId, MethodKind, MethodSet};
15use store::Store;
16use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
17use tokio::process::ChildStdin;
18use tokio::sync::Mutex;
19
20use crate::SshError;
21use crate::{command, resolve};
22
23/// State shared across every connection.
24struct Shared {
25 store: Store,
26 config: Arc<Config>,
27 /// The `--config` path to hand hooks via `FABRICA_CONFIG`.
28 config_path: Option<String>,
29 /// The instance signing secret, for minting short-lived LFS bearer tokens.
30 secret: String,
31}
32
33/// The russh `Server` factory.
34struct GitServer {
35 shared: Arc<Shared>,
36}
37
38impl russh::server::Server for GitServer {
39 type Handler = GitHandler;
40
41 fn new_client(&mut self, _peer: Option<std::net::SocketAddr>) -> GitHandler {
42 GitHandler {
43 shared: Arc::clone(&self.shared),
44 identity: None,
45 stdin: Arc::new(Mutex::new(None)),
46 }
47 }
48}
49
50/// The authenticated identity, bound after publickey auth.
51#[derive(Clone)]
52struct Identity {
53 user_id: String,
54 username: String,
55 is_admin: bool,
56 key_id: String,
57}
58
59/// One connection's handler.
60struct GitHandler {
61 shared: Arc<Shared>,
62 identity: Option<Identity>,
63 /// The running pack subprocess's stdin, written to as client data arrives.
64 stdin: Arc<Mutex<Option<ChildStdin>>>,
65}
66
67impl Handler for GitHandler {
68 type Error = russh::Error;
69
70 async fn auth_publickey(
71 &mut self,
72 _user: &str,
73 public_key: &PublicKey,
74 ) -> Result<Auth, Self::Error> {
75 // Identify by key fingerprint, not the SSH username (everyone is `git@`).
76 let fingerprint = public_key.fingerprint(HashAlg::Sha256).to_string();
77 let Ok(Some(key)) = self
78 .shared
79 .store
80 .key_by_fingerprint(KeyKind::Ssh, &fingerprint)
81 .await
82 else {
83 return Ok(Auth::reject());
84 };
85 let Ok(Some(user)) = self.shared.store.user_by_id(&key.user_id).await else {
86 return Ok(Auth::reject());
87 };
88 if user.disabled_at.is_some() {
89 return Ok(Auth::reject());
90 }
91 let _ = self.shared.store.touch_key_used(&key.id).await;
92 // A successful public-key auth proves possession of the private key, so
93 // the key is now verified.
94 let _ = self.shared.store.set_key_verified(&key.id).await;
95 self.identity = Some(Identity {
96 user_id: user.id,
97 username: user.username,
98 is_admin: user.is_admin,
99 key_id: key.id,
100 });
101 Ok(Auth::Accept)
102 }
103
104 async fn channel_open_session(
105 &mut self,
106 _channel: Channel<Msg>,
107 reply: ChannelOpenHandle,
108 _session: &mut Session,
109 ) -> Result<(), Self::Error> {
110 // Dropping the handle rejects the channel, so accept it explicitly.
111 reply.accept().await;
112 Ok(())
113 }
114
115 async fn exec_request(
116 &mut self,
117 channel: ChannelId,
118 data: &[u8],
119 session: &mut Session,
120 ) -> Result<(), Self::Error> {
121 session.channel_success(channel)?;
122 let command = String::from_utf8_lossy(data).into_owned();
123 if let Err(banner) = self.spawn_exec(channel, &command, session).await {
124 let _ = session.extended_data(channel, 1, banner.into_bytes());
125 let _ = session.exit_status_request(channel, 1);
126 let _ = session.eof(channel);
127 let _ = session.close(channel);
128 }
129 Ok(())
130 }
131
132 async fn shell_request(
133 &mut self,
134 channel: ChannelId,
135 session: &mut Session,
136 ) -> Result<(), Self::Error> {
137 session.channel_success(channel)?;
138 // Use the async handle so the banner/close actually flush before return.
139 let handle = session.handle();
140 let _ = handle
141 .extended_data(channel, 1, self.no_shell_banner().into_bytes())
142 .await;
143 let _ = handle.exit_status_request(channel, 1).await;
144 let _ = handle.eof(channel).await;
145 let _ = handle.close(channel).await;
146 Ok(())
147 }
148
149 async fn data(
150 &mut self,
151 _channel: ChannelId,
152 data: &[u8],
153 _session: &mut Session,
154 ) -> Result<(), Self::Error> {
155 if let Some(stdin) = self.stdin.lock().await.as_mut() {
156 let _ = stdin.write_all(data).await;
157 }
158 Ok(())
159 }
160
161 async fn channel_eof(
162 &mut self,
163 _channel: ChannelId,
164 _session: &mut Session,
165 ) -> Result<(), Self::Error> {
166 // Close the child's stdin so upload-pack/receive-pack sees EOF.
167 *self.stdin.lock().await = None;
168 Ok(())
169 }
170}
171
172impl GitHandler {
173 /// The friendly shell-access banner.
174 fn no_shell_banner(&self) -> String {
175 let who = self
176 .identity
177 .as_ref()
178 .map_or("there", |i| i.username.as_str());
179 format!("Hi {who}! fabrica does not provide shell access.\n")
180 }
181
182 /// Authorize the exec command and, on success, spawn the pack process and start
183 /// streaming. On failure returns the banner to write to stderr.
184 async fn spawn_exec(
185 &mut self,
186 channel: ChannelId,
187 command: &str,
188 session: &mut Session,
189 ) -> Result<(), String> {
190 let Some(identity) = self.identity.clone() else {
191 return Err("Authentication required.".to_string());
192 };
193 // git-lfs-authenticate: hand back an HTTP endpoint + bearer token, then exit.
194 if let Some(lfs) = command::parse_lfs(command) {
195 return self
196 .lfs_authenticate(channel, &identity, &lfs, session)
197 .await;
198 }
199 let parsed = command::parse(command).ok_or_else(|| self.no_shell_banner())?;
200 let (owner, repo_path) = resolve::normalize(&parsed.path)
201 .ok_or_else(|| "Invalid repository path.".to_string())?;
202
203 let repo = self
204 .resolve_repo(&owner, &repo_path, &identity, parsed.service)
205 .await?;
206
207 let repo_dir = git::repo_path(&self.shared.config.storage.repo_dir, &repo.id)
208 .map_err(|_| "Repository is unavailable.".to_string())?;
209
210 let env = git::pack::PackEnv {
211 home: self.shared.config.storage.data_dir.join("tmp"),
212 config_path: self.shared.config_path.clone(),
213 user_id: Some(identity.user_id),
214 user_name: Some(identity.username),
215 repo_id: Some(repo.id),
216 key_id: Some(identity.key_id),
217 protocol: None,
218 };
219 let mut child = git::pack::spawn(
220 &self.shared.config.git.binary,
221 parsed.service,
222 &[],
223 &repo_dir,
224 &env,
225 )
226 .map_err(|e| format!("Failed to start git: {e}"))?;
227
228 let (Some(stdin), Some(mut stdout), Some(mut stderr)) =
229 (child.stdin.take(), child.stdout.take(), child.stderr.take())
230 else {
231 return Err("Internal error.".to_string());
232 };
233 *self.stdin.lock().await = Some(stdin);
234
235 let handle = session.handle();
236 tokio::spawn(async move {
237 let mut obuf = vec![0u8; 32 * 1024];
238 let mut ebuf = vec![0u8; 32 * 1024];
239 let (mut out_done, mut err_done) = (false, false);
240 while !(out_done && err_done) {
241 tokio::select! {
242 r = stdout.read(&mut obuf), if !out_done => match r {
243 Ok(0) | Err(_) => out_done = true,
244 Ok(n) => { let _ = handle.data(channel, obuf[..n].to_vec()).await; }
245 },
246 r = stderr.read(&mut ebuf), if !err_done => match r {
247 Ok(0) | Err(_) => err_done = true,
248 Ok(n) => { let _ = handle.extended_data(channel, 1, ebuf[..n].to_vec()).await; }
249 },
250 }
251 }
252 let code = child
253 .wait()
254 .await
255 .ok()
256 .and_then(|s| s.code())
257 .and_then(|c| u32::try_from(c).ok())
258 .unwrap_or(0);
259 let _ = handle.exit_status_request(channel, code).await;
260 let _ = handle.eof(channel).await;
261 let _ = handle.close(channel).await;
262 });
263 Ok(())
264 }
265
266 /// Handle `git-lfs-authenticate`: authorize the repo for the operation, mint a
267 /// short-lived bearer token, and write the LFS endpoint JSON to stdout. The
268 /// actual object transfer then happens over HTTP against that endpoint.
269 async fn lfs_authenticate(
270 &mut self,
271 channel: ChannelId,
272 identity: &Identity,
273 lfs: &command::LfsAuth,
274 session: &mut Session,
275 ) -> Result<(), String> {
276 let (owner, repo_path) =
277 resolve::normalize(&lfs.path).ok_or_else(|| "Invalid repository path.".to_string())?;
278 let not_found = || "Repository not found.".to_string();
279 let owner_user = self
280 .shared
281 .store
282 .user_by_username(&owner)
283 .await
284 .ok()
285 .flatten()
286 .ok_or_else(not_found)?;
287 let repo = self
288 .shared
289 .store
290 .repo_by_owner_path(&owner_user.id, &repo_path)
291 .await
292 .ok()
293 .flatten()
294 .ok_or_else(not_found)?;
295 let collaborator = self
296 .shared
297 .store
298 .effective_permission(&repo.id, &identity.user_id)
299 .await
300 .ok()
301 .flatten()
302 .and_then(|p| auth::Permission::parse(&p));
303 let viewer = auth::Viewer {
304 id: identity.user_id.clone(),
305 is_admin: identity.is_admin,
306 };
307 let access = auth::access(
308 Some(&viewer),
309 &repo,
310 collaborator,
311 self.shared.config.instance.allow_anonymous,
312 );
313 let required = if lfs.operation == "upload" {
314 auth::Access::Write
315 } else {
316 auth::Access::Read
317 };
318 if access < required {
319 return Err(if access >= auth::Access::Read {
320 "You do not have write access to this repository.".to_string()
321 } else {
322 not_found()
323 });
324 }
325
326 // A one-hour bearer token the HTTP LFS endpoints accept (verified by
327 // signature + expiry, not the API-token revocation path).
328 let now_s = i64::try_from(
329 std::time::SystemTime::now()
330 .duration_since(std::time::UNIX_EPOCH)
331 .map_or(0, |d| d.as_secs()),
332 )
333 .unwrap_or(0);
334 let claims = auth::Claims {
335 sub: identity.user_id.clone(),
336 jti: format!("lfs-{}", repo.id),
337 scopes: "lfs".to_string(),
338 iat: now_s,
339 exp: now_s + 3600,
340 };
341 let token = auth::mint_jwt(&self.shared.secret, &claims)
342 .map_err(|_| "Could not issue an LFS token.".to_string())?;
343 let href = format!(
344 "{}/{}/{}.git/info/lfs",
345 self.shared.config.instance.url.trim_end_matches('/'),
346 owner,
347 repo_path
348 );
349 // git-lfs reads this JSON from stdout. href/token contain only URL- and
350 // JWT-safe characters, so no escaping is needed.
351 let json = format!(
352 "{{\"href\":\"{href}\",\"header\":{{\"Authorization\":\"Bearer {token}\"}},\"expires_in\":3600}}\n"
353 );
354 let handle = session.handle();
355 let _ = handle.data(channel, json.into_bytes()).await;
356 let _ = handle.exit_status_request(channel, 0).await;
357 let _ = handle.eof(channel).await;
358 let _ = handle.close(channel).await;
359 Ok(())
360 }
361
362 /// Resolve and authorize the repo for the requested service.
363 async fn resolve_repo(
364 &self,
365 owner: &str,
366 repo_path: &str,
367 identity: &Identity,
368 service: git::pack::Service,
369 ) -> Result<model::Repo, String> {
370 let not_found = || "Repository not found.".to_string();
371 let owner_user = self
372 .shared
373 .store
374 .user_by_username(owner)
375 .await
376 .ok()
377 .flatten()
378 .ok_or_else(not_found)?;
379 let existing = self
380 .shared
381 .store
382 .repo_by_owner_path(&owner_user.id, repo_path)
383 .await
384 .ok()
385 .flatten();
386 // Push-to-create: a push (receive-pack) to a non-existent repo owned by
387 // the pushing user (or by an admin) auto-creates it, at the instance
388 // default visibility. Any other case is a 404 so a stranger cannot probe
389 // for repositories.
390 let repo = if let Some(repo) = existing {
391 repo
392 } else if matches!(service, git::pack::Service::ReceivePack)
393 && (identity.user_id == owner_user.id || identity.is_admin)
394 {
395 self.create_on_push(&owner_user, repo_path).await?
396 } else {
397 return Err(not_found());
398 };
399
400 // Archived repositories are read-only: refuse any push.
401 if matches!(service, git::pack::Service::ReceivePack) && repo.archived_at.is_some() {
402 return Err("This repository is archived and is read-only.".to_string());
403 }
404
405 let collaborator = self
406 .shared
407 .store
408 .effective_permission(&repo.id, &identity.user_id)
409 .await
410 .ok()
411 .flatten()
412 .and_then(|p| auth::Permission::parse(&p));
413 let viewer = auth::Viewer {
414 id: identity.user_id.clone(),
415 is_admin: identity.is_admin,
416 };
417 let access = auth::access(
418 Some(&viewer),
419 &repo,
420 collaborator,
421 self.shared.config.instance.allow_anonymous,
422 );
423
424 let required = match service {
425 git::pack::Service::ReceivePack => auth::Access::Write,
426 _ => auth::Access::Read,
427 };
428 if access >= required {
429 Ok(repo)
430 } else if access >= auth::Access::Read {
431 Err("You do not have write access to this repository.".to_string())
432 } else {
433 // No read access at all: do not confirm the repo exists.
434 Err(not_found())
435 }
436 }
437
438 /// Create a repository on first push: any missing group prefix, the database
439 /// row (visibility = instance default), and the bare repo on disk. Rolls the
440 /// row back if the on-disk creation fails.
441 async fn create_on_push(
442 &self,
443 owner: &model::User,
444 repo_path: &str,
445 ) -> Result<model::Repo, String> {
446 let (group_path, leaf) = match repo_path.rsplit_once('/') {
447 Some((group, leaf)) => (Some(group), leaf),
448 None => (None, repo_path),
449 };
450 let group_id = match group_path {
451 Some(gp) => Some(
452 self.shared
453 .store
454 .ensure_group_path(&owner.id, gp)
455 .await
456 .map_err(|e| e.to_string())?
457 .id,
458 ),
459 None => None,
460 };
461 let branch = self.shared.config.instance.default_branch.clone();
462 let visibility = model::Visibility::from_token(
463 self.shared.config.instance.default_visibility.as_token(),
464 )
465 .unwrap_or(model::Visibility::Private);
466 let repo = self
467 .shared
468 .store
469 .create_repo(store::NewRepo {
470 owner_id: owner.id.clone(),
471 group_id,
472 name: leaf.to_string(),
473 path: repo_path.to_string(),
474 description: None,
475 visibility,
476 default_branch: branch.clone(),
477 })
478 .await
479 .map_err(|e| e.to_string())?;
480
481 let hook = std::env::current_exe().unwrap_or_else(|_| std::path::PathBuf::from("fabrica"));
482 if let Err(err) = git::create_bare(
483 &self.shared.config.storage.repo_dir,
484 &repo.id,
485 &branch,
486 &hook,
487 ) {
488 let _ = self.shared.store.delete_repo(&repo.id).await;
489 return Err(format!("Could not create repository: {err}"));
490 }
491 tracing::info!(owner = %owner.username, path = %repo_path, "push-to-create: repository created");
492 Ok(repo)
493 }
494}
495
496/// Run the SSH server until the process exits. A no-op when `ssh.enabled` is false.
497///
498/// # Errors
499///
500/// Returns [`SshError`] if the host key cannot be prepared or the server fails to
501/// bind or run.
502pub async fn serve(
503 config: Arc<Config>,
504 store: Store,
505 config_path: Option<String>,
506 secret: String,
507) -> Result<(), SshError> {
508 if !config.ssh.enabled {
509 tracing::info!("ssh disabled (ssh.enabled = false)");
510 return Ok(());
511 }
512
513 let host_pem = crate::hostkey::load_or_generate(&config.ssh.host_key_path)?;
514 let host_key = russh::keys::PrivateKey::from_openssh(host_pem.as_str())
515 .map_err(|e| SshError::HostKey(format!("parsing host key: {e}")))?;
516 let russh_config = Arc::new(russh::server::Config {
517 methods: MethodSet::from(&[MethodKind::PublicKey][..]),
518 keys: vec![host_key],
519 inactivity_timeout: Some(Duration::from_secs(3600)),
520 auth_rejection_time: Duration::from_secs(2),
521 ..russh::server::Config::default()
522 });
523
524 let shared = Arc::new(Shared {
525 store,
526 config: Arc::clone(&config),
527 config_path,
528 secret,
529 });
530 let addr = (config.ssh.address.clone(), config.ssh.port);
531 tracing::info!(address = %config.ssh.address, port = config.ssh.port, "ssh server listening");
532
533 let mut server = GitServer { shared };
534 server
535 .run_on_address(russh_config, addr)
536 .await
537 .map_err(|e| SshError::Server(e.to_string()))
538}