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
branchmain ▾
main All branches →
zuka/src/main.rs
RSmain.rs17.8 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 mut run = ci::run::Run::queued(account.as_str(), repo.as_str(), git_ref, new);
292 run.number = runs.next_number(&account, &repo);
293 match runs.put(&account, &repo, &run) {
294 Ok(()) => println!("{}", brand::say(format_args!("queued run {}", run.id))),
295 Err(e) => eprintln!("{}", brand::say(format_args!("could not queue a run: {e}"))),
296 }
297 }
298 Ok(())
299}
300
301/// Recover `(account, repo)` from a bare repository path under the git root.
302fn account_and_repo(
303 config: &Config,
304 git_dir: &str,
305) -> Option<(git::validate::Name, git::validate::Name)> {
306 let path = std::path::Path::new(git_dir).canonicalize().ok()?;
307 let root = config.data_dir.join("git").canonicalize().ok()?;
308 let rest = path.strip_prefix(&root).ok()?;
309
310 let mut parts = rest.components();
311 let account = parts.next()?.as_os_str().to_str()?;
312 let repo = parts.next()?.as_os_str().to_str()?.strip_suffix(".git")?;
313
314 Some((
315 git::validate::name(account).ok()?,
316 git::validate::name(repo).ok()?,
317 ))
318}
319
320/// One-shot maintenance, for an operator or a timer.
321#[derive(Debug, Clone, Copy)]
322enum Job {
323 Gc,
324 Sweep,
325 Fsck,
326}
327
328/// `<binary> gc | sweep | fsck [--repair]`
329fn run_maintenance(job: Job) -> Result<()> {
330 let config = Config::from_env().context("read configuration")?;
331
332 match job {
333 Job::Gc => {
334 let outcome = jobs::gc::run(&config);
335 eprintln!(
336 "[gc] repacked {} repositories, reclaimed {} bytes",
337 outcome.packed, outcome.reclaimed
338 );
339 }
340 Job::Sweep => {
341 let outcome = jobs::sweep::run(&config);
342 eprintln!(
343 "[sweep] removed {} deleted repositories, reclaimed {} bytes",
344 outcome.removed, outcome.bytes
345 );
346 }
347 Job::Fsck => {
348 let repair = std::env::args().any(|a| a == "--repair");
349 let outcome = jobs::fsck::run(&config, repair);
350 println!("{}", serde_json::to_string_pretty(&outcome)?);
351 eprintln!(
352 "[fsck] checked {} repositories, {} orphan(s), {} repaired",
353 outcome.checked,
354 outcome.orphans.len(),
355 outcome.repaired
356 );
357 // A non-zero exit lets a timer or a health check notice.
358 if !outcome.orphans.is_empty() && !repair {
359 std::process::exit(1);
360 }
361 }
362 }
363 Ok(())
364}
365
366/// `<binary> jobs` — the maintenance daemon, for its own service unit.
367async fn run_jobs() -> Result<()> {
368 let config = Config::from_env().context("read configuration")?;
369 eprintln!(
370 "[boot] {} {} jobs data={}",
371 brand::NAME,
372 brand::VERSION,
373 config.data_dir.display()
374 );
375 jobs::serve(Arc::new(config)).await;
376 Ok(())
377}
378
379/// `<binary> hook <name> [args...]` — invoked by the installed git hooks.
380///
381/// The hooks carry no policy; each execs back into this binary so the rule has one
382/// implementation shared with the API path.
383async fn run_hook() -> Result<()> {
384 let name = std::env::args()
385 .nth(2)
386 .with_context(|| format!("usage: {} hook <name>", brand::NAME))?;
387
388 match name.as_str() {
389 "update" => {
390 // git calls: update <ref> <old> <new>
391 let args: Vec<String> = std::env::args().skip(3).collect();
392 let [ref_name, old, new] = args.as_slice() else {
393 bail!("update hook expects <ref> <old> <new>");
394 };
395
396 let git_dir = std::env::var("GIT_DIR").unwrap_or_else(|_| ".".into());
397
398 match git::store::check_ref_move(std::path::Path::new(&git_dir), ref_name, old, new) {
399 Ok(()) => Ok(()),
400 Err(e) => {
401 // Printed to the pusher's terminal by receive-pack.
402 eprintln!("{}", brand::say(e.detail_for_user()));
403 std::process::exit(1);
404 }
405 }
406 }
407 "post-receive" => enqueue_runs().await,
408 other => bail!("unknown hook {other:?}"),
409 }
410}
411
412/// `<binary> token <account> [--name n] [--scopes a,b] [--repos a,b] [--days n]`
413fn mint_token() -> Result<()> {
414 let config = Config::from_env().context("read configuration")?;
415 let args: Vec<String> = std::env::args().skip(2).collect();
416
417 let account = args
418 .first()
419 .filter(|a| !a.starts_with("--"))
420 .with_context(|| {
421 format!(
422 "usage: {} token <account> [--scopes admin] [--days 90]",
423 brand::NAME
424 )
425 })?
426 .clone();
427 let account = git::validate::name(&account)?;
428
429 let name = flag(&args, "--name").unwrap_or_else(|| "default".into());
430 let scopes = parse_scopes(&flag(&args, "--scopes").unwrap_or_else(|| "admin".into()))?;
431 let repos = flag_repos(&args)?;
432
433 // A forever-token is total account compromise, so expiry is opt-out and loud.
434 let expires_at = if args.iter().any(|a| a == "--no-expiry") {
435 eprintln!("[token] warning: this token never expires");
436 None
437 } else {
438 let days = flag(&args, "--days")
439 .map(|v| v.parse::<u64>().context("--days must be a number"))
440 .transpose()?
441 .unwrap_or(config.token_ttl_days);
442 Some(token::now_secs() + days * 24 * 60 * 60)
443 };
444
445 let (secret, record) = token::mint(account.as_str(), &name, scopes, repos, expires_at);
446
447 let path = config.tokens_file();
448 let mut file = token::TokenFile::load(&path)?;
449 file.tokens.push(record);
450 file.save(&path)?;
451
452 eprintln!("[token] wrote {}", path.display());
453 eprintln!("[token] shown once and not recoverable:");
454 println!("{secret}");
455 Ok(())
456}
457
458/// `<binary> key <account> <path-to-pubkey> [--title t] [--read-only] [--repos a,b]`
459fn add_key() -> Result<()> {
460 let config = Config::from_env().context("read configuration")?;
461 let args: Vec<String> = std::env::args().skip(2).collect();
462
463 let account = args
464 .first()
465 .with_context(|| format!("usage: {} key <account> <path-to-public-key>", brand::NAME))?;
466 let account = git::validate::name(account)?;
467
468 let key_path = args
469 .get(1)
470 .filter(|a| !a.starts_with("--"))
471 .with_context(|| format!("usage: {} key <account> <path-to-public-key>", brand::NAME))?;
472 let line = std::fs::read_to_string(key_path).with_context(|| format!("read {key_path}"))?;
473
474 let (algorithm, blob) = account::key::parse_authorized_key(&line)?;
475
476 let repos = flag_repos(&args)?;
477
478 let record = account::KeyRecord {
479 id: uuid::Uuid::new_v4().simple().to_string(),
480 account: account.as_str().to_string(),
481 title: flag(&args, "--title").unwrap_or_else(|| "cli".into()),
482 public_key: account::key::encode_public_key(&algorithm, &blob),
483 fingerprint: account::key::fingerprint(&blob),
484 read_only: args.iter().any(|a| a == "--read-only"),
485 repos,
486 created_at: token::now_secs(),
487 last_used_at: None,
488 };
489
490 let path = config.keys_file();
491 let mut file = account::KeyFile::load(&path)?;
492 if file.keys.iter().any(|k| k.public_key == record.public_key) {
493 bail!("that public key is already registered");
494 }
495 eprintln!("[key] {} -> {}", record.fingerprint, record.account);
496 file.keys.push(record);
497 file.save(&path)?;
498 eprintln!("[key] wrote {}", path.display());
499 Ok(())
500}
501
502/// A comma-separated `--repos` list, validated into canonical names.
503fn flag_repos(args: &[String]) -> Result<Vec<String>> {
504 let Some(raw) = flag(args, "--repos") else {
505 return Ok(Vec::new());
506 };
507 raw.split(',')
508 .filter(|s| !s.is_empty())
509 .map(|s| Ok(git::validate::name(s)?.as_str().to_string()))
510 .collect()
511}
512
513fn flag(args: &[String], name: &str) -> Option<String> {
514 args.iter()
515 .position(|a| a == name)
516 .and_then(|i| args.get(i + 1))
517 .cloned()
518}
519
520fn parse_scopes(raw: &str) -> Result<Vec<Scope>> {
521 let scopes = raw
522 .split(',')
523 .filter(|s| !s.is_empty())
524 .map(|s| match s.trim() {
525 "repo:read" => Ok(Scope::RepoRead),
526 "repo:write" => Ok(Scope::RepoWrite),
527 "admin" => Ok(Scope::Admin),
528 "ci" => Ok(Scope::Ci),
529 other => bail!("unknown scope {other:?} (repo:read | repo:write | admin | ci)"),
530 })
531 .collect::<Result<Vec<_>>>()?;
532
533 if scopes.is_empty() {
534 bail!("at least one scope is required");
535 }
536 Ok(scopes)
537}