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