zuka
zuka/src/main.rs

worklyn / zukapublic

Agent-first git hosting. One Rust binary: git over HTTP and SSH, a REST API, MCP, CI, and multi-tenant isolation.

Get a copy: git clone https://zuka.worklyn.com/worklyn/zuka.git
branch73d1f1251cf621d50e001993f3b6792d6b657255 ▾
main All branches →
zuka/src/main.rs
RSmain.rs17.7 KBDownload
1// Agent-first git hosting.
2//
3// Boot and operator commands only. Everything else lives in the modules below.
4// The product name lives in `brand`; nothing here should spell it out.
5
6mod account;
7mod api;
8mod brand;
9mod ci;
10mod config;
11mod control;
12mod core;
13mod error;
14mod git;
15mod http;
16mod jobs;
17mod mcp;
18mod setup;
19mod ssh;
20mod store;
21
22use account::token;
23use account::Scope;
24use anyhow::{bail, Context, Result};
25use config::{Config, Mode};
26use hyper::service::service_fn;
27use hyper_util::rt::{TokioIo, TokioTimer};
28use std::sync::Arc;
29use tokio::net::TcpListener;
30
31#[tokio::main(flavor = "multi_thread")]
32async fn main() -> Result<()> {
33 match std::env::args().nth(1).as_deref() {
34 Some("hook") => return run_hook().await,
35 Some("jobs") => return run_jobs().await,
36 Some("setup") => return run_setup(),
37 Some("gc") => return run_maintenance(Job::Gc),
38 Some("sweep") => return run_maintenance(Job::Sweep),
39 Some("fsck") => return run_maintenance(Job::Fsck),
40 Some("token") => return mint_token(),
41 Some("key") => return add_key(),
42 Some("version") => {
43 println!("{} {}", brand::NAME, brand::build());
44 return Ok(());
45 }
46 Some(other) => bail!(
47 "unknown command {other:?} \
48 (setup | token | key | jobs | gc | sweep | fsck | hook | version)"
49 ),
50 None => {}
51 }
52
53 serve().await
54}
55
56async fn serve() -> Result<()> {
57 let config = Config::from_env().context("read configuration")?;
58 eprintln!(
59 "[boot] {} {} mode={} data={}",
60 brand::NAME,
61 brand::VERSION,
62 config.mode.as_str(),
63 config.data_dir.display()
64 );
65
66 if config.mode == Mode::Control {
67 return serve_control(config).await;
68 }
69
70 // Accepting a security-relevant setting and then ignoring it is worse than not
71 // accepting it: the operator believes hosted auth is enforcing revocation.
72 if config.me_url.is_some() {
73 bail!(
74 "{} is not implemented yet; unset it to use the local token file",
75 brand::env_name("ME_URL")
76 );
77 }
78 if config.mode == Mode::Tenant {
79 eprintln!(
80 "[boot] tenant mode: callers are identified by a signed assertion from the \
81 control plane ({} key(s) trusted)",
82 config.control_public_keys.len()
83 );
84 }
85 if config.import_allow_private {
86 eprintln!("[boot] import: PRIVATE ADDRESSES ALLOWED — only safe on a trusted network");
87 }
88 if !config.ci_enabled {
89 eprintln!("[boot] ci: disabled");
90 } else {
91 // Said loudly because standalone CI is not a security boundary: anyone who
92 // can push can run code as this user.
93 eprintln!("[boot] ci: ENABLED — steps run as this user with rlimits but no container");
94 }
95
96 // Resolved once here rather than per request, where it was a synchronous
97 // fork+exec on a runtime worker for every clone and push.
98 git::exec::http_backend()?;
99
100 let bind = config.bind;
101 let ssh_bind = config.ssh_bind;
102 let header_timeout = config.header_timeout;
103 let host_key_path = config.ssh_host_key.clone();
104 let state = Arc::new(http::AppState::new(config)?);
105
106 eprintln!(
107 "[boot] auth: {} token(s), {} ssh key(s)",
108 state.tokens().tokens.len(),
109 state.ssh_keys().keys.len()
110 );
111
112 if let Some(addr) = ssh_bind {
113 let host_key = ssh::load_host_key(&host_key_path)?;
114 let idle = state.config.ssh_idle_timeout;
115 let server = ssh::SshServer::new(Arc::clone(&state));
116 tokio::spawn(async move {
117 if let Err(e) = server.run(addr, host_key, idle).await {
118 eprintln!("[ssh] server stopped: {e:#}");
119 }
120 });
121 } else {
122 eprintln!("[boot] ssh: disabled");
123 }
124
125 tokio::spawn(ci::serve(Arc::clone(&state)));
126
127 let listener = TcpListener::bind(bind)
128 .await
129 .with_context(|| format!("bind {bind}"))?;
130 eprintln!("[boot] listening on http://{bind}");
131
132 loop {
133 let (stream, peer) = listener.accept().await.context("accept")?;
134 let state = Arc::clone(&state);
135 tokio::spawn(async move {
136 let io = TokioIo::new(stream);
137 let service = service_fn(move |req| http::handle(req, Arc::clone(&state), peer));
138 if let Err(e) = hyper::server::conn::http1::Builder::new()
139 // hyper panics at runtime if a timeout is set without a timer.
140 .timer(TokioTimer::new())
141 .header_read_timeout(header_timeout)
142 .serve_connection(io, service)
143 .await
144 {
145 // A client hanging up mid-clone is ordinary, not a fault.
146 let message = e.to_string();
147 if !message.contains("closed") && !message.contains("reset") {
148 eprintln!("[conn] {message}");
149 }
150 }
151 });
152 }
153}
154
155/// `<binary> setup --standalone|--isolated` — install onto this host.
156fn run_setup() -> Result<()> {
157 let args: Vec<String> = std::env::args().skip(2).collect();
158 let options = setup::Options::from_args(&args)?;
159
160 let checks = setup::preflight(options.shape);
161 let missing: Vec<&(String, bool, String)> = checks.iter().filter(|(_, ok, _)| !ok).collect();
162
163 for (name, ok, detail) in &checks {
164 eprintln!(
165 "[setup] {} {name}: {detail}",
166 if *ok { "ok " } else { "MISS" }
167 );
168 }
169 if !missing.is_empty() && !options.dry_run {
170 bail!(
171 "missing prerequisites: {}. Install them, or re-run with --dry-run to see \
172 what would be written.",
173 missing
174 .iter()
175 .map(|(n, _, _)| n.as_str())
176 .collect::<Vec<_>>()
177 .join(", ")
178 );
179 }
180 eprintln!();
181
182 setup::apply(&options)
183}
184
185/// The control plane: accounts, provisioning, and proxying to tenants.
186///
187/// Holds no repositories. Every data-plane request is forwarded to the account's own
188/// container with a freshly signed, single-use assertion.
189async fn serve_control(config: Config) -> Result<()> {
190 use control::provision::{IncusProvisioner, Provisioner};
191
192 let accounts = control::accounts::AccountStore::open(config.accounts_dir())?;
193 let signer = load_control_key(&config)?;
194 eprintln!(
195 "[control] tenants need {}={}",
196 brand::env_name("CONTROL_PUBLIC_KEYS"),
197 signer.public_key()
198 );
199
200 let provisioner: Arc<dyn Provisioner> = Arc::new(IncusProvisioner::new(
201 config.incus_image.clone(),
202 config.incus_network.clone(),
203 config.tenant_port,
204 config.tenant_binary.clone(),
205 signer.public_key(),
206 std::env::var(brand::env_name("PUBLIC_URL")).ok(),
207 std::env::var(brand::env_name("PUBLIC_SSH")).ok(),
208 ));
209
210 tokio::spawn(control::reconcile_forever(
211 accounts.clone(),
212 Arc::clone(&provisioner),
213 ));
214
215 control::serve(config, accounts, signer, provisioner).await
216}
217
218/// Load the control plane's signing key, generating one on first boot.
219///
220/// The private half never leaves this host: a tenant runs the account's own CI, so
221/// anything a tenant holds must be forgery-proof.
222fn load_control_key(config: &Config) -> Result<account::assertion::Signer_> {
223 use base64::Engine;
224
225 let path = &config.control_key_file;
226 if path.exists() {
227 let encoded =
228 std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
229 let bytes = base64::engine::general_purpose::STANDARD
230 .decode(encoded.trim())
231 .context("control key is not valid base64")?;
232 let seed: [u8; 32] = bytes
233 .try_into()
234 .map_err(|_| anyhow::anyhow!("control key must be 32 bytes"))?;
235 return Ok(account::assertion::Signer_::from_seed(&seed));
236 }
237
238 let mut seed = [0u8; 32];
239 getrandom::fill(&mut seed).map_err(|e| anyhow::anyhow!("read OS entropy: {e}"))?;
240 account::token::write_private(
241 path,
242 base64::engine::general_purpose::STANDARD
243 .encode(seed)
244 .as_bytes(),
245 )?;
246 eprintln!("[control] generated a signing key at {}", path.display());
247 Ok(account::assertion::Signer_::from_seed(&seed))
248}
249
250/// `post-receive` — queue a CI run for each updated ref.
251///
252/// git feeds `<old> <new> <ref>` per line on stdin. This never fails the push: a
253/// broken queue must not stop code being accepted.
254async fn enqueue_runs() -> Result<()> {
255 use std::io::Read;
256
257 let config = Config::from_env().context("read configuration")?;
258 if !config.ci_enabled {
259 return Ok(());
260 }
261
262 let git_dir = std::env::var("GIT_DIR").unwrap_or_else(|_| ".".into());
263 let Some((account, repo)) = account_and_repo(&config, &git_dir) else {
264 return Ok(());
265 };
266
267 let mut stdin = String::new();
268 if std::io::stdin().read_to_string(&mut stdin).is_err() {
269 return Ok(());
270 }
271
272 let runs = match ci::run::RunStore::open(config.runs_dir()) {
273 Ok(runs) => runs,
274 Err(e) => {
275 eprintln!("{}", brand::say(format_args!("could not queue a run: {e}")));
276 return Ok(());
277 }
278 };
279
280 for line in stdin.lines() {
281 let fields: Vec<&str> = line.split_whitespace().collect();
282 let [_old, new, git_ref] = fields.as_slice() else {
283 continue;
284 };
285 // A deletion has nothing to build.
286 if git::store::is_null_oid(new) {
287 continue;
288 }
289
290 let run = ci::run::Run::queued(account.as_str(), repo.as_str(), git_ref, new);
291 match runs.put(&account, &repo, &run) {
292 Ok(()) => println!("{}", brand::say(format_args!("queued run {}", run.id))),
293 Err(e) => eprintln!("{}", brand::say(format_args!("could not queue a run: {e}"))),
294 }
295 }
296 Ok(())
297}
298
299/// Recover `(account, repo)` from a bare repository path under the git root.
300fn account_and_repo(
301 config: &Config,
302 git_dir: &str,
303) -> Option<(git::validate::Name, git::validate::Name)> {
304 let path = std::path::Path::new(git_dir).canonicalize().ok()?;
305 let root = config.data_dir.join("git").canonicalize().ok()?;
306 let rest = path.strip_prefix(&root).ok()?;
307
308 let mut parts = rest.components();
309 let account = parts.next()?.as_os_str().to_str()?;
310 let repo = parts.next()?.as_os_str().to_str()?.strip_suffix(".git")?;
311
312 Some((
313 git::validate::name(account).ok()?,
314 git::validate::name(repo).ok()?,
315 ))
316}
317
318/// One-shot maintenance, for an operator or a timer.
319#[derive(Debug, Clone, Copy)]
320enum Job {
321 Gc,
322 Sweep,
323 Fsck,
324}
325
326/// `<binary> gc | sweep | fsck [--repair]`
327fn run_maintenance(job: Job) -> Result<()> {
328 let config = Config::from_env().context("read configuration")?;
329
330 match job {
331 Job::Gc => {
332 let outcome = jobs::gc::run(&config);
333 eprintln!(
334 "[gc] repacked {} repositories, reclaimed {} bytes",
335 outcome.packed, outcome.reclaimed
336 );
337 }
338 Job::Sweep => {
339 let outcome = jobs::sweep::run(&config);
340 eprintln!(
341 "[sweep] removed {} deleted repositories, reclaimed {} bytes",
342 outcome.removed, outcome.bytes
343 );
344 }
345 Job::Fsck => {
346 let repair = std::env::args().any(|a| a == "--repair");
347 let outcome = jobs::fsck::run(&config, repair);
348 println!("{}", serde_json::to_string_pretty(&outcome)?);
349 eprintln!(
350 "[fsck] checked {} repositories, {} orphan(s), {} repaired",
351 outcome.checked,
352 outcome.orphans.len(),
353 outcome.repaired
354 );
355 // A non-zero exit lets a timer or a health check notice.
356 if !outcome.orphans.is_empty() && !repair {
357 std::process::exit(1);
358 }
359 }
360 }
361 Ok(())
362}
363
364/// `<binary> jobs` — the maintenance daemon, for its own service unit.
365async fn run_jobs() -> Result<()> {
366 let config = Config::from_env().context("read configuration")?;
367 eprintln!(
368 "[boot] {} {} jobs data={}",
369 brand::NAME,
370 brand::VERSION,
371 config.data_dir.display()
372 );
373 jobs::serve(Arc::new(config)).await;
374 Ok(())
375}
376
377/// `<binary> hook <name> [args...]` — invoked by the installed git hooks.
378///
379/// The hooks carry no policy; each execs back into this binary so the rule has one
380/// implementation shared with the API path.
381async fn run_hook() -> Result<()> {
382 let name = std::env::args()
383 .nth(2)
384 .with_context(|| format!("usage: {} hook <name>", brand::NAME))?;
385
386 match name.as_str() {
387 "update" => {
388 // git calls: update <ref> <old> <new>
389 let args: Vec<String> = std::env::args().skip(3).collect();
390 let [ref_name, old, new] = args.as_slice() else {
391 bail!("update hook expects <ref> <old> <new>");
392 };
393
394 let git_dir = std::env::var("GIT_DIR").unwrap_or_else(|_| ".".into());
395
396 match git::store::check_ref_move(std::path::Path::new(&git_dir), ref_name, old, new) {
397 Ok(()) => Ok(()),
398 Err(e) => {
399 // Printed to the pusher's terminal by receive-pack.
400 eprintln!("{}", brand::say(e.detail_for_user()));
401 std::process::exit(1);
402 }
403 }
404 }
405 "post-receive" => enqueue_runs().await,
406 other => bail!("unknown hook {other:?}"),
407 }
408}
409
410/// `<binary> token <account> [--name n] [--scopes a,b] [--repos a,b] [--days n]`
411fn mint_token() -> Result<()> {
412 let config = Config::from_env().context("read configuration")?;
413 let args: Vec<String> = std::env::args().skip(2).collect();
414
415 let account = args
416 .first()
417 .filter(|a| !a.starts_with("--"))
418 .with_context(|| {
419 format!(
420 "usage: {} token <account> [--scopes admin] [--days 90]",
421 brand::NAME
422 )
423 })?
424 .clone();
425 let account = git::validate::name(&account)?;
426
427 let name = flag(&args, "--name").unwrap_or_else(|| "default".into());
428 let scopes = parse_scopes(&flag(&args, "--scopes").unwrap_or_else(|| "admin".into()))?;
429 let repos = flag_repos(&args)?;
430
431 // A forever-token is total account compromise, so expiry is opt-out and loud.
432 let expires_at = if args.iter().any(|a| a == "--no-expiry") {
433 eprintln!("[token] warning: this token never expires");
434 None
435 } else {
436 let days = flag(&args, "--days")
437 .map(|v| v.parse::<u64>().context("--days must be a number"))
438 .transpose()?
439 .unwrap_or(config.token_ttl_days);
440 Some(token::now_secs() + days * 24 * 60 * 60)
441 };
442
443 let (secret, record) = token::mint(account.as_str(), &name, scopes, repos, expires_at);
444
445 let path = config.tokens_file();
446 let mut file = token::TokenFile::load(&path)?;
447 file.tokens.push(record);
448 file.save(&path)?;
449
450 eprintln!("[token] wrote {}", path.display());
451 eprintln!("[token] shown once and not recoverable:");
452 println!("{secret}");
453 Ok(())
454}
455
456/// `<binary> key <account> <path-to-pubkey> [--title t] [--read-only] [--repos a,b]`
457fn add_key() -> Result<()> {
458 let config = Config::from_env().context("read configuration")?;
459 let args: Vec<String> = std::env::args().skip(2).collect();
460
461 let account = args
462 .first()
463 .with_context(|| format!("usage: {} key <account> <path-to-public-key>", brand::NAME))?;
464 let account = git::validate::name(account)?;
465
466 let key_path = args
467 .get(1)
468 .filter(|a| !a.starts_with("--"))
469 .with_context(|| format!("usage: {} key <account> <path-to-public-key>", brand::NAME))?;
470 let line = std::fs::read_to_string(key_path).with_context(|| format!("read {key_path}"))?;
471
472 let (algorithm, blob) = account::key::parse_authorized_key(&line)?;
473
474 let repos = flag_repos(&args)?;
475
476 let record = account::KeyRecord {
477 id: uuid::Uuid::new_v4().simple().to_string(),
478 account: account.as_str().to_string(),
479 title: flag(&args, "--title").unwrap_or_else(|| "cli".into()),
480 public_key: account::key::encode_public_key(&algorithm, &blob),
481 fingerprint: account::key::fingerprint(&blob),
482 read_only: args.iter().any(|a| a == "--read-only"),
483 repos,
484 created_at: token::now_secs(),
485 last_used_at: None,
486 };
487
488 let path = config.keys_file();
489 let mut file = account::KeyFile::load(&path)?;
490 if file.keys.iter().any(|k| k.public_key == record.public_key) {
491 bail!("that public key is already registered");
492 }
493 eprintln!("[key] {} -> {}", record.fingerprint, record.account);
494 file.keys.push(record);
495 file.save(&path)?;
496 eprintln!("[key] wrote {}", path.display());
497 Ok(())
498}
499
500/// A comma-separated `--repos` list, validated into canonical names.
501fn flag_repos(args: &[String]) -> Result<Vec<String>> {
502 let Some(raw) = flag(args, "--repos") else {
503 return Ok(Vec::new());
504 };
505 raw.split(',')
506 .filter(|s| !s.is_empty())
507 .map(|s| Ok(git::validate::name(s)?.as_str().to_string()))
508 .collect()
509}
510
511fn flag(args: &[String], name: &str) -> Option<String> {
512 args.iter()
513 .position(|a| a == name)
514 .and_then(|i| args.get(i + 1))
515 .cloned()
516}
517
518fn parse_scopes(raw: &str) -> Result<Vec<Scope>> {
519 let scopes = raw
520 .split(',')
521 .filter(|s| !s.is_empty())
522 .map(|s| match s.trim() {
523 "repo:read" => Ok(Scope::RepoRead),
524 "repo:write" => Ok(Scope::RepoWrite),
525 "admin" => Ok(Scope::Admin),
526 "ci" => Ok(Scope::Ci),
527 other => bail!("unknown scope {other:?} (repo:read | repo:write | admin | ci)"),
528 })
529 .collect::<Result<Vec<_>>>()?;
530
531 if scopes.is_empty() {
532 bail!("at least one scope is required");
533 }
534 Ok(scopes)
535}