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

RSmod.rs40.4 KBDownload

1	// The control plane.
2	//
3	// Owns accounts, provisions a container per account, and proxies every data-plane
4	// request to the right one. It holds no repositories itself.
5	//
6	// Two rules shape everything here:
7	//
8	// * Provisioning is never awaited by a request. Creating an account writes a
9	// `Provisioning` record and returns; a reconciler drives it to `Ready` or
10	// `Failed`. Waiting synchronously on container boot ties a user-facing request
11	// to a multi-second operation that can fail halfway.
12	// * **The tenant is told who is calling by a signed, single-use, request-bound
13	// assertion.** Not a shared secret: a tenant runs the account's own CI, so
14	// anything symmetric in there forges every other tenant.
15
16	pub mod accounts;
17	pub mod provision;
18	pub mod proxy;
19
20	use crate::error::Result;
21	use crate::git::validate::Name;
22	use accounts::{AccountRecord, AccountStore};
23	use provision::{Provisioner, State};
24	use std::sync::Arc;
25	use std::time::Duration;
26
27	/// How often the reconciler sweeps accounts that are not yet `Ready`.
28	const RECONCILE_INTERVAL: Duration = Duration::from_secs(10);
29
30	/// Tenants upgraded per sweep.
31	///
32	/// Staggered rather than all at once: an upgrade restarts the tenant, and restarting
33	/// every account simultaneously turns a routine deploy into an outage.
34	const UPGRADES_PER_SWEEP: usize = 2;
35
36	/// How long a tenant may sit unused before it is stopped.
37	///
38	/// Always-on would mean one permanently running container per account, including for
39	/// people who push twice a year. Stopping is cheap and starting is a few seconds,
40	/// which the caller sees as a `429` with a `Retry-After`.
41	const IDLE_STOP: Duration = Duration::from_secs(1800);
42
43	/// Drive every non-ready account forward, forever.
44	pub async fn reconcile_forever(store: AccountStore, provisioner: Arc<dyn Provisioner>) {
45	eprintln!("[control] reconciler running every {RECONCILE_INTERVAL:?}");
46	loop {
47	let store = store.clone();
48	let provisioner = Arc::clone(&provisioner);
49	let outcome =
50	tokio::task::spawn_blocking(move \|\| reconcile_once(&store, provisioner.as_ref())).await;
51
52	if let Err(e) = outcome {
53	// A panic in one sweep must not stop the reconciler; the next sweep is
54	// the retry.
55	eprintln!("[control] reconcile sweep failed: {e}");
56	}
57	tokio::time::sleep(RECONCILE_INTERVAL).await;
58	}
59	}
60
61	/// One reconciliation sweep. Returns how many accounts changed state.
62	pub fn reconcile_once(store: &AccountStore, provisioner: &dyn Provisioner) -> usize {
63	reconcile_at(store, provisioner, crate::account::token::now_secs())
64	}
65
66	/// As [`reconcile_once`], with the clock injected so idle behaviour is testable.
67	pub fn reconcile_at(store: &AccountStore, provisioner: &dyn Provisioner, now: u64) -> usize {
68	let mut changed = 0;
69	let mut upgraded = 0;
70
71	for mut record in store.list() {
72	if record.state == State::Ready {
73	// A ready record with no usable endpoint cannot serve anything: the
74	// proxy turns an empty URL into a 429 and a version-matched tenant
75	// would otherwise sit there forever because upgrades are the only
76	// ready-state probe. Re-observe Incus and repair the durable record.
77	if record.endpoint.as_deref().is_none_or(str::is_empty) {
78	let Ok(account) = crate::git::validate::name(&record.name) else {
79	continue;
80	};
81	let next = drive(&account, &record, provisioner);
82	if next.state != record.state
83	\|\| next.endpoint != record.endpoint
84	\|\| next.version != record.binary_version
85	{
86	record.state = next.state;
87	record.endpoint = next.endpoint;
88	record.detail = next.detail;
89	record.binary_version = next.version;
90	if store.put(&record).is_ok() {
91	changed += 1;
92	eprintln!("[control] {} endpoint repaired", record.name);
93	}
94	}
95	continue;
96	}
97
98	// A tenant running an older binary is upgraded in place. Stopped tenants
99	// are skipped: they pick the new binary up when they are next started,
100	// because `create` is idempotent and re-pushes.
101	if record.binary_version.as_deref() != Some(crate::brand::build())
102	&& upgraded < UPGRADES_PER_SWEEP
103	{
104	let Ok(account) = crate::git::validate::name(&record.name) else {
105	continue;
106	};
107	match provisioner.upgrade(&account) {
108	Ok(()) => {
109	record.binary_version = Some(crate::brand::build().to_string());
110	if store.put(&record).is_ok() {
111	upgraded += 1;
112	changed += 1;
113	eprintln!(
114	"[control] {} upgraded to {}",
115	record.name,
116	crate::brand::build()
117	);
118	}
119	}
120	// Left at the old version so the next sweep retries. A failed
121	// upgrade must not mark the tenant as upgraded.
122	Err(e) => eprintln!("[control] could not upgrade {}: {e}", record.name),
123	}
124	continue;
125	}
126
127	// Stop a tenant nobody has used. It comes back on the next request.
128	if now.saturating_sub(record.last_seen_at) >= IDLE_STOP.as_secs() {
129	let Ok(account) = crate::git::validate::name(&record.name) else {
130	continue;
131	};
132	match provisioner.stop(&account) {
133	Ok(()) => {
134	record.state = State::Stopped;
135	if store.put(&record).is_ok() {
136	changed += 1;
137	eprintln!("[control] {} -> Stopped (idle)", record.name);
138	}
139	}
140	Err(e) => eprintln!("[control] could not stop {}: {e}", record.name),
141	}
142	}
143	continue;
144	}
145
146	// A stopped tenant is only restarted once someone asks for it again.
147	if record.state == State::Stopped
148	&& now.saturating_sub(record.last_seen_at) >= IDLE_STOP.as_secs()
149	{
150	continue;
151	}
152	let Ok(account) = crate::git::validate::name(&record.name) else {
153	continue;
154	};
155
156	let next = drive(&account, &record, provisioner);
157	if next.state != record.state
158	\|\| next.endpoint != record.endpoint
159	\|\| next.version != record.binary_version
160	{
161	record.state = next.state;
162	record.endpoint = next.endpoint;
163	record.detail = next.detail;
164	record.binary_version = next.version;
165	if store.put(&record).is_ok() {
166	changed += 1;
167	eprintln!("[control] {} -> {:?}", record.name, record.state);
168	}
169	}
170	}
171	changed
172	}
173
174	/// What one account's record should become.
175	struct Next {
176	state: State,
177	endpoint: Option<String>,
178	detail: Option<String>,
179	version: Option<String>,
180	}
181
182	fn drive(account: &Name, record: &AccountRecord, provisioner: &dyn Provisioner) -> Next {
183	// A failed account is not retried automatically: repeatedly relaunching a
184	// container that cannot start burns the host. It is retried when someone asks,
185	// through `retry`.
186	if record.state == State::Failed {
187	return Next {
188	state: State::Failed,
189	endpoint: record.endpoint.clone(),
190	detail: record.detail.clone(),
191	version: record.binary_version.clone(),
192	};
193	}
194
195	match provisioner.status(account) {
196	// Already up: adopt whatever the provisioner reports.
197	Ok(Some(instance)) if instance.state == State::Ready => Next {
198	state: State::Ready,
199	endpoint: Some(instance.endpoint),
200	detail: None,
201	version: record.binary_version.clone(),
202	},
203	// Exists but stopped, and someone wants it: start it.
204	Ok(Some(_)) => match provisioner.start(account) {
205	Ok(instance) => Next {
206	state: State::Ready,
207	endpoint: Some(instance.endpoint),
208	detail: None,
209	version: record.binary_version.clone(),
210	},
211	Err(e) => Next {
212	state: State::Provisioning,
213	endpoint: record.endpoint.clone(),
214	detail: Some(e.detail_for_user()),
215	version: record.binary_version.clone(),
216	},
217	},
218	// Not there at all: create it.
219	// A fresh container gets the binary we are running, by construction.
220	Ok(None) => match provisioner.create(account) {
221	Ok(instance) => Next {
222	state: instance.state,
223	endpoint: Some(instance.endpoint),
224	detail: None,
225	version: Some(crate::brand::build().to_string()),
226	},
227	Err(e) => Next {
228	state: State::Failed,
229	endpoint: None,
230	detail: Some(e.detail_for_user()),
231	version: None,
232	},
233	},
234	Err(e) => Next {
235	state: State::Provisioning,
236	endpoint: record.endpoint.clone(),
237	detail: Some(e.detail_for_user()),
238	version: record.binary_version.clone(),
239	},
240	}
241	}
242
243	/// Move a failed account back into the queue.
244	pub fn retry(store: &AccountStore, account: &Name) -> Result<()> {
245	let mut record = store
246	.get(account)?
247	.ok_or(crate::error::Error::NotFound("account"))?;
248	record.state = State::Provisioning;
249	record.detail = None;
250	store.put(&record)
251	}
252
253	// ── serving ────────────────────────────────────────────────────────────────
254
255	use crate::account::assertion::Signer_;
256	use crate::config::Config;
257	use crate::http::response::{self, Body};
258	use hyper::body::Incoming;
259	use hyper::service::service_fn;
260	use hyper::{Method, Request, Response, StatusCode};
261	use hyper_util::rt::{TokioIo, TokioTimer};
262	use std::convert::Infallible;
263	use tokio::net::TcpListener;
264
265	/// What the control plane needs to answer a request.
266	pub struct ControlState {
267	pub config: Config,
268	pub accounts: AccountStore,
269	pub signer: Signer_,
270	pub client: reqwest::Client,
271	pub provisioner: Arc<dyn Provisioner>,
272	/// The control plane is the authentication authority.
273	///
274	/// A tenant holds no tokens and cannot check one — it trusts the signed
275	/// assertion instead. So the token store lives here, and this is where a
276	/// credential is turned into an account.
277	tokens: crate::store::cached::Cached<crate::account::token::TokenFile>,
278	}
279
280	impl ControlState {
281	/// Authorise an operator action: creating, listing or destroying accounts.
282	///
283	/// These are not tenant operations. Creating an account provisions a container
284	/// and deleting one destroys its repositories, so they cannot be reachable with
285	/// an ordinary account's credential — a tenant admin token must not be able to
286	/// delete a different tenant — and they certainly cannot be reachable with none.
287	///
288	/// The operator is named by `<PREFIX>OPERATOR`. When it is unset every account
289	/// endpoint is refused, because the alternative default is the one this replaced:
290	/// a public host where a stranger could enumerate every tenant and delete them.
291	fn require_operator(&self, headers: &hyper::HeaderMap) -> Result<()> {
292	let configured = std::env::var(crate::brand::env_name("OPERATOR")).ok();
293	let tokens = self.tokens.get();
294	let identity =
295	crate::http::auth::resolve_local(&tokens, &crate::http::auth::extract(headers))?;
296	operator_allows(configured.as_deref(), &identity)
297	}
298
299	/// Resolve a caller to the account whose tenant should serve them.
300	fn account_for(&self, headers: &hyper::HeaderMap) -> Result<crate::git::validate::Name> {
301	let tokens = self.tokens.get();
302	let identity =
303	crate::http::auth::resolve_local(&tokens, &crate::http::auth::extract(headers))?;
304	Ok(identity.account)
305	}
306	}
307
308	/// Whether an identity may perform an operator action.
309	///
310	/// Separated from the request so the rule can be tested without a running control
311	/// plane, an Incus host or a token file — the thing that must not be wrong here is
312	/// the decision, not the plumbing that reaches it.
313	fn operator_allows(configured: Option<&str>, identity: &crate::account::Identity) -> Result<()> {
314	let Some(operator) = configured.filter(\|v\| !v.is_empty()) else {
315	// No operator configured means nobody may manage accounts. The alternative
316	// default is the one this replaced: a public host on which a stranger could
317	// enumerate every tenant and delete them.
318	return Err(crate::error::Error::Forbidden);
319	};
320	let operator = crate::git::validate::name(operator)?;
321
322	// `require_account_admin` carries both remaining conditions: the identity must
323	// own the account, and a token confined to a repository list never qualifies —
324	// a repo-scoped token has no business provisioning containers.
325	identity.require_account_admin(&operator)
326	}
327
328	/// How the control plane should present a Git request to a tenant.
329	///
330	/// A valid credential for another account is not a rejected credential. Public Git
331	/// reads still work through credential helpers that happen to offer such a token,
332	/// but no assertion is minted for it. A missing or bad credential may only take this
333	/// anonymous path for an upload-pack read; receive-pack always requires a caller.
334	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
335	enum GitCaller {
336	Vouched,
337	Anonymous,
338	}
339
340	fn git_caller(
341	claimed: &crate::git::validate::Name,
342	caller: Result<crate::git::validate::Name>,
343	offered: bool,
344	writes: bool,
345	) -> Result<GitCaller> {
346	match caller {
347	Ok(account) if account == *claimed => Ok(GitCaller::Vouched),
348	Ok(_) if writes => Err(crate::error::Error::NotFound("repository")),
349	Ok(_) => Ok(GitCaller::Anonymous),
350	Err(_) if writes \|\| offered => Err(crate::error::Error::Unauthorized),
351	Err(_) => Ok(GitCaller::Anonymous),
352	}
353	}
354
355	/// Serve the control plane.
356	pub async fn serve(
357	config: Config,
358	accounts: AccountStore,
359	signer: Signer_,
360	provisioner: Arc<dyn Provisioner>,
361	) -> anyhow::Result<()> {
362	let bind = config.bind;
363	let client = proxy::client(config.proxy_timeout)?;
364	let tokens = crate::store::cached::Cached::load(
365	config.tokens_file(),
366	crate::account::token::TokenFile::load,
367	)?;
368	let state = Arc::new(ControlState {
369	config,
370	accounts,
371	signer,
372	client,
373	provisioner,
374	tokens,
375	});
376
377	let listener = TcpListener::bind(bind).await?;
378	eprintln!("[control] listening on http://{bind}");
379
380	loop {
381	let (stream, _peer) = listener.accept().await?;
382	let state = Arc::clone(&state);
383	tokio::spawn(async move {
384	let io = TokioIo::new(stream);
385	let service = service_fn(move \|req\| handle(req, Arc::clone(&state)));
386	if let Err(e) = hyper::server::conn::http1::Builder::new()
387	.timer(TokioTimer::new())
388	.header_read_timeout(state_header_timeout())
389	.serve_connection(io, service)
390	.await
391	{
392	let message = e.to_string();
393	if !message.contains("closed") && !message.contains("reset") {
394	eprintln!("[control] conn: {message}");
395	}
396	}
397	});
398	}
399	}
400
401	fn state_header_timeout() -> Duration {
402	Duration::from_secs(15)
403	}
404
405	async fn handle(
406	request: Request<Incoming>,
407	state: Arc<ControlState>,
408	) -> std::result::Result<Response<Body>, Infallible> {
409	let path = request.uri().path().to_string();
410	let method = request.method().clone();
411
412	let outcome = route(request, &state).await;
413	Ok(match outcome {
414	Ok(response) => response,
415	Err(error) => {
416	if let Some(cause) = error.cause() {
417	eprintln!("[control] {method} {path} -> {} {cause:#}", error.status());
418	}
419	response::problem(&error)
420	}
421	})
422	}
423
424	async fn route(request: Request<Incoming>, state: &ControlState) -> Result<Response<Body>> {
425	let path = request.uri().path().to_string();
426	let segments: Vec<&str> = path.split('/').filter(\|s\| !s.is_empty()).collect();
427
428	match (request.method(), segments.as_slice()) {
429	(&Method::GET, ["healthz"]) => Ok(response::json(
430	StatusCode::OK,
431	&serde_json::json!({
432	"status": "ok",
433	"mode": "control",
434	"version": crate::brand::VERSION,
435	"build": crate::brand::build(),
436	"accounts": state.accounts.list().len(),
437	}),
438	)),
439
440	// Accounts are the control plane's own resource; everything else is proxied.
441	(&Method::GET, ["v1", "accounts"]) => {
442	state.require_operator(request.headers())?;
443	Ok(response::json(
444	StatusCode::OK,
445	&serde_json::json!({ "items": state.accounts.list(), "truncated": false }),
446	))
447	}
448	(&Method::POST, ["v1", "accounts", name]) => {
449	state.require_operator(request.headers())?;
450	let account = crate::git::validate::name(name)?;
451	let record = state.accounts.create(&account)?;
452	Ok(response::json(StatusCode::ACCEPTED, &record))
453	}
454	(&Method::GET, ["v1", "accounts", name]) => {
455	state.require_operator(request.headers())?;
456	let account = crate::git::validate::name(name)?;
457	let record = state
458	.accounts
459	.get(&account)?
460	.ok_or(crate::error::Error::NotFound("account"))?;
461	Ok(response::json(StatusCode::OK, &record))
462	}
463	(&Method::DELETE, ["v1", "accounts", name]) => {
464	state.require_operator(request.headers())?;
465	let account = crate::git::validate::name(name)?;
466	if !state.accounts.delete(&account)? {
467	return Err(crate::error::Error::NotFound("account"));
468	}
469	// The container goes with the record. Provisioning is idempotent, so a
470	// failure here is retried by the next sweep rather than blocking the
471	// response.
472	let provisioner = Arc::clone(&state.provisioner);
473	let name = account.clone();
474	tokio::task::spawn_blocking(move \|\| {
475	if let Err(e) = provisioner.delete(&name) {
476	eprintln!("[control] could not remove {name}'s tenant: {e}");
477	}
478	});
479	Ok(response::no_content())
480	}
481	(&Method::POST, ["v1", "accounts", name, "retry"]) => {
482	state.require_operator(request.headers())?;
483	let account = crate::git::validate::name(name)?;
484	retry(&state.accounts, &account)?;
485	Ok(response::no_content())
486	}
487
488	// Everything else belongs to a tenant. Which one is decided by the
489	// credential, not by the path: a path segment is a claim, a token is not.
490	(_, ["v1", ..]) \| (_, ["mcp"]) => {
491	let account = state.account_for(request.headers())?;
492	let endpoint = state.accounts.endpoint(&account)?;
493	proxy::forward(
494	&state.client,
495	&state.signer,
496	&endpoint,
497	account.as_str(),
498	request,
499	state.config.max_body_bytes,
500	)
501	.await
502	}
503
504	// Git paths name the account. Writes must match the credential's account;
505	// anonymous upload-pack is the one public Git read path, so a public clone
506	// can reach the tenant without minting an assertion for a stranger. A bad
507	// presented credential stays rejected rather than quietly becoming a clone.
508	(_, [path_account, repo, ..]) if repo.ends_with(".git") => {
509	let claimed = crate::git::validate::name(path_account)?;
510	let route =
511	crate::git::transport::parse_route(&path, request.uri().query().unwrap_or(""))
512	.ok_or(crate::error::Error::NotFound("route"))?;
513	let offered = crate::http::auth::presented(request.headers());
514	match git_caller(
515	&claimed,
516	state.account_for(request.headers()),
517	offered,
518	route.writes,
519	)? {
520	GitCaller::Vouched => {
521	let endpoint = state.accounts.endpoint(&claimed)?;
522	proxy::forward(
523	&state.client,
524	&state.signer,
525	&endpoint,
526	claimed.as_str(),
527	request,
528	state.config.max_body_bytes,
529	)
530	.await
531	}
532	GitCaller::Anonymous => {
533	// A missing account must challenge like a private repository,
534	// or an unauthenticated upload-pack probe becomes an account
535	// enumeration endpoint. Existing provisioning/paused accounts
536	// still return their honest retry response.
537	let endpoint = match state.accounts.endpoint(&claimed) {
538	Ok(endpoint) => endpoint,
539	Err(crate::error::Error::NotFound(_)) => {
540	return Err(crate::error::Error::Unauthorized)
541	}
542	Err(error) => return Err(error),
543	};
544	proxy::forward_anonymous(
545	&state.client,
546	&endpoint,
547	request,
548	state.config.max_body_bytes,
549	)
550	.await
551	}
552	}
553	}
554
555	// The stylesheet is static and identical in every build, so the control
556	// plane answers it rather than waking a tenant for it.
557	(&Method::GET, _) if crate::web::is_style_path(&path) => {
558	Ok(response::css(crate::web::STYLE))
559	}
560
561	// `/` features one repository, which lives in a tenant like any other.
562	(&Method::GET, []) => match state.config.home_repo() {
563	Some((account, repo)) => Ok(response::redirect(&format!("/{account}/{repo}"))),
564	None => Ok(crate::web::welcome()),
565	},
566
567	// The machine-readable front door follows the same rule as `/`.
568	(&Method::GET, ["llms.txt"]) => match state.config.home_repo() {
569	Some((account, repo)) => Ok(response::redirect(&format!("/{account}/{repo}/llms.txt"))),
570	None => Ok(crate::web::service_llms()),
571	},
572
573	// The package origin's root belongs to the host, not to an account. It is
574	// the one quiet sentence on the package hostname, so do not try to extract
575	// a tenant account from the bare internal prefix below.
576	(&Method::GET, ["_pkg"]) => Ok(response::package_landing()),
577
578	// The package-only Caddy host rewrites an external module URL into this
579	// internal prefix. It names the account in its path like the browser, but
580	// unlike the browser it is always anonymous: package v1 serves public
581	// source only, so a bearer token can never accidentally become private
582	// package support or make a cache credential-dependent.
583	(&Method::GET, ["_pkg", ..]) => {
584	let claimed = crate::package::account_from_segments(&segments)?;
585	let endpoint = state.accounts.endpoint(&claimed)?;
586	proxy::forward_anonymous(
587	&state.client,
588	&endpoint,
589	request,
590	state.config.max_body_bytes,
591	)
592	.await
593	}
594
595	// The browser surface. Unlike the git and API paths, the account comes from
596	// the path rather than the credential, because the whole point is that a
597	// stranger with no credential can reach a public repository.
598	(&Method::GET, [path_account, _repo, ..]) if crate::web::owns(&path) => {
599	let claimed = crate::git::validate::name(path_account)?;
600	let endpoint = state.accounts.endpoint(&claimed)?;
601
602	// Vouch only when the credential belongs to the account being browsed.
603	// A token is worth nothing in another account's tenant, so presenting
604	// one must not turn into an assertion naming that other account — and
605	// an anonymous forward is exactly right for browsing someone else's
606	// public repository.
607	let vouch = match state.account_for(request.headers()) {
608	Ok(name) if name == claimed => true,
609	Ok(_) => false,
610	// No credential at all is the anonymous case. A credential that was
611	// presented and rejected is still an error.
612	Err(_) if !crate::http::auth::presented(request.headers()) => false,
613	Err(e) => return Err(e),
614	};
615
616	if vouch {
617	proxy::forward(
618	&state.client,
619	&state.signer,
620	&endpoint,
621	claimed.as_str(),
622	request,
623	state.config.max_body_bytes,
624	)
625	.await
626	} else {
627	proxy::forward_anonymous(
628	&state.client,
629	&endpoint,
630	request,
631	state.config.max_body_bytes,
632	)
633	.await
634	}
635	}
636
637	_ => Err(crate::error::Error::NotFound("route")),
638	}
639	}
640
641	#[cfg(test)]
642	mod tests {
643	use super::*;
644	use provision::FakeProvisioner;
645	use std::sync::atomic::Ordering;
646
647	fn setup() -> (tempfile::TempDir, AccountStore, Arc<FakeProvisioner>) {
648	let dir = tempfile::tempdir().unwrap();
649	let store = AccountStore::open(dir.path()).unwrap();
650	(dir, store, Arc::new(FakeProvisioner::new()))
651	}
652
653	fn identity_for(
654	account: &str,
655	scopes: Vec<crate::account::token::Scope>,
656	) -> crate::account::Identity {
657	crate::account::Identity {
658	account: crate::git::validate::name(account).unwrap(),
659	token_id: "t".into(),
660	scopes,
661	repos: Vec::new(),
662	}
663	}
664
665	#[test]
666	fn account_management_is_refused_when_no_operator_is_named() {
667	use crate::account::token::Scope;
668	// The default has to be closed. Creating an account provisions a container
669	// and deleting one destroys its repositories, and both were once reachable
670	// with no credential at all.
671	let admin = identity_for("alice", vec![Scope::Admin]);
672	assert!(operator_allows(None, &admin).is_err());
673	assert!(operator_allows(Some(""), &admin).is_err());
674	}
675
676	#[test]
677	fn only_the_named_operator_may_manage_accounts() {
678	use crate::account::token::Scope;
679	assert!(operator_allows(Some("ops"), &identity_for("ops", vec![Scope::Admin])).is_ok());
680
681	// A tenant's own admin token must not reach account management, or any
682	// account could destroy any other.
683	assert!(operator_allows(Some("ops"), &identity_for("alice", vec![Scope::Admin])).is_err());
684
685	// The operator account still needs admin scope.
686	assert!(operator_allows(Some("ops"), &identity_for("ops", vec![Scope::RepoRead])).is_err());
687	assert!(
688	operator_allows(Some("ops"), &identity_for("ops", vec![Scope::RepoWrite])).is_err()
689	);
690	}
691
692	#[test]
693	fn an_operator_token_confined_to_repositories_is_refused() {
694	use crate::account::token::Scope;
695	// Confinement must not be escapable by an operation that names no
696	// repository — provisioning containers is not repository work.
697	let confined = crate::account::Identity {
698	repos: vec![crate::git::validate::name("site").unwrap()],
699	..identity_for("ops", vec![Scope::Admin])
700	};
701	assert!(operator_allows(Some("ops"), &confined).is_err());
702	}
703
704	#[test]
705	fn a_public_git_read_can_be_anonymous_but_a_write_never_can() {
706	let worklyn = crate::git::validate::name("worklyn").unwrap();
707	let alice = crate::git::validate::name("alice").unwrap();
708
709	assert_eq!(
710	git_caller(
711	&worklyn,
712	Err(crate::error::Error::Unauthorized),
713	false,
714	false
715	)
716	.unwrap(),
717	GitCaller::Anonymous
718	);
719	// Credential helpers often offer a valid token for a different account.
720	// That must not block a public clone or mint that account's assertion.
721	assert_eq!(
722	git_caller(&worklyn, Ok(alice), true, false).unwrap(),
723	GitCaller::Anonymous
724	);
725	// A bad presented token remains a visible authentication failure.
726	assert!(git_caller(
727	&worklyn,
728	Err(crate::error::Error::Unauthorized),
729	true,
730	false
731	)
732	.is_err());
733	// Neither a missing credential nor another account can reach receive-pack.
734	assert!(git_caller(
735	&worklyn,
736	Err(crate::error::Error::Unauthorized),
737	false,
738	true
739	)
740	.is_err());
741	assert!(git_caller(
742	&worklyn,
743	Ok(crate::git::validate::name("alice").unwrap()),
744	true,
745	true
746	)
747	.is_err());
748	}
749
750	#[test]
751	fn a_new_account_is_driven_to_ready_without_a_request_waiting() {
752	let (_dir, store, fake) = setup();
753	let alice = crate::git::validate::name("alice").unwrap();
754
755	// Creation records intent and returns; nothing is provisioned yet.
756	store.create(&alice).unwrap();
757	assert_eq!(
758	store.get(&alice).unwrap().unwrap().state,
759	State::Provisioning
760	);
761
762	// First sweep creates it, second observes it running.
763	reconcile_once(&store, fake.as_ref());
764	reconcile_once(&store, fake.as_ref());
765
766	let record = store.get(&alice).unwrap().unwrap();
767	assert_eq!(record.state, State::Ready);
768	assert!(record.endpoint.is_some(), "a ready tenant must be routable");
769	}
770
771	#[test]
772	fn a_ready_record_with_a_lost_endpoint_repairs_itself() {
773	// A record can survive an old provisioning race with state=ready but an
774	// empty endpoint. Without this repair a matching-version tenant is never
775	// probed again, and every proxy request returns a misleading 429 forever.
776	let (_dir, store, fake) = setup();
777	settle(&store, &fake, "alice");
778	let alice = crate::git::validate::name("alice").unwrap();
779	let mut record = store.get(&alice).unwrap().unwrap();
780	record.endpoint = Some(String::new());
781	store.put(&record).unwrap();
782
783	assert_eq!(reconcile_once(&store, fake.as_ref()), 1);
784	let repaired = store.get(&alice).unwrap().unwrap();
785	assert_eq!(repaired.state, State::Ready);
786	assert_eq!(
787	repaired.endpoint.as_deref(),
788	Some("http://127.0.0.1:9/alice")
789	);
790	}
791
792	#[test]
793	fn a_provisioning_failure_is_recorded_with_its_reason() {
794	let (_dir, store, fake) = setup();
795	let alice = crate::git::validate::name("alice").unwrap();
796	store.create(&alice).unwrap();
797
798	fake.fail_next.store(true, Ordering::SeqCst);
799	reconcile_once(&store, fake.as_ref());
800
801	let record = store.get(&alice).unwrap().unwrap();
802	assert_eq!(record.state, State::Failed);
803	assert!(record.detail.is_some(), "a failure must say why");
804	}
805
806	#[test]
807	fn a_failed_account_is_not_retried_until_asked() {
808	let (_dir, store, fake) = setup();
809	let alice = crate::git::validate::name("alice").unwrap();
810	store.create(&alice).unwrap();
811
812	fake.fail_next.store(true, Ordering::SeqCst);
813	reconcile_once(&store, fake.as_ref());
814	assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Failed);
815
816	// Sweeping again must not relaunch: a container that cannot start would be
817	// relaunched forever.
818	assert_eq!(reconcile_once(&store, fake.as_ref()), 0);
819	assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Failed);
820
821	retry(&store, &alice).unwrap();
822	reconcile_once(&store, fake.as_ref());
823	reconcile_once(&store, fake.as_ref());
824	assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
825	}
826
827	#[test]
828	fn a_stopped_tenant_is_started_again() {
829	let (_dir, store, fake) = setup();
830	let alice = crate::git::validate::name("alice").unwrap();
831	store.create(&alice).unwrap();
832	reconcile_once(&store, fake.as_ref());
833	reconcile_once(&store, fake.as_ref());
834
835	fake.stop(&alice).unwrap();
836	let mut record = store.get(&alice).unwrap().unwrap();
837	record.state = State::Stopped;
838	store.put(&record).unwrap();
839
840	reconcile_once(&store, fake.as_ref());
841	assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
842	}
843
844	/// Drive an account all the way to Ready on the current version.
845	fn settle(store: &AccountStore, fake: &FakeProvisioner, name: &str) {
846	let account = crate::git::validate::name(name).unwrap();
847	store.create(&account).unwrap();
848	reconcile_once(store, fake);
849	reconcile_once(store, fake);
850	let mut record = store.get(&account).unwrap().unwrap();
851	record.last_seen_at = crate::account::token::now_secs();
852	store.put(&record).unwrap();
853	}
854
855	#[test]
856	fn a_new_tenant_records_the_version_it_was_built_with() {
857	let (_dir, store, fake) = setup();
858	settle(&store, &fake, "alice");
859
860	let record = store
861	.get(&crate::git::validate::name("alice").unwrap())
862	.unwrap()
863	.unwrap();
864	assert_eq!(record.state, State::Ready);
865	assert_eq!(
866	record.binary_version.as_deref(),
867	Some(crate::brand::build())
868	);
869	assert!(
870	fake.upgrades.lock().unwrap().is_empty(),
871	"a container built from the current binary needs no upgrade"
872	);
873	}
874
875	#[test]
876	fn a_tenant_on_an_old_binary_is_upgraded_in_place() {
877	let (_dir, store, fake) = setup();
878	settle(&store, &fake, "alice");
879	let alice = crate::git::validate::name("alice").unwrap();
880
881	// What a deploy looks like: the control plane is newer than the tenant.
882	let mut record = store.get(&alice).unwrap().unwrap();
883	record.binary_version = Some("0.0.1-old".into());
884	store.put(&record).unwrap();
885
886	reconcile_once(&store, fake.as_ref());
887
888	assert_eq!(fake.upgrades.lock().unwrap().as_slice(), ["alice"]);
889	assert_eq!(
890	store
891	.get(&alice)
892	.unwrap()
893	.unwrap()
894	.binary_version
895	.as_deref(),
896	Some(crate::brand::build())
897	);
898	// Upgrading must not have disturbed the tenant's state.
899	assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
900	}
901
902	#[test]
903	fn a_failed_upgrade_leaves_the_old_version_so_the_next_sweep_retries() {
904	use std::sync::atomic::Ordering;
905	let (_dir, store, fake) = setup();
906	settle(&store, &fake, "alice");
907	let alice = crate::git::validate::name("alice").unwrap();
908
909	let mut record = store.get(&alice).unwrap().unwrap();
910	record.binary_version = Some("0.0.1-old".into());
911	store.put(&record).unwrap();
912
913	fake.fail_next.store(true, Ordering::SeqCst);
914	reconcile_once(&store, fake.as_ref());
915	assert_eq!(
916	store
917	.get(&alice)
918	.unwrap()
919	.unwrap()
920	.binary_version
921	.as_deref(),
922	Some("0.0.1-old"),
923	"a failed upgrade must not be recorded as done"
924	);
925
926	reconcile_once(&store, fake.as_ref());
927	assert_eq!(
928	store
929	.get(&alice)
930	.unwrap()
931	.unwrap()
932	.binary_version
933	.as_deref(),
934	Some(crate::brand::build()),
935	"the next sweep must retry"
936	);
937	}
938
939	#[test]
940	fn upgrades_are_staggered_so_a_deploy_is_not_an_outage() {
941	let (_dir, store, fake) = setup();
942	let names = ["a1", "a2", "a3", "a4", "a5"];
943
944	// Settle every account first: `settle` reconciles, so backdating inside the
945	// loop would let those sweeps do the upgrading and hide the stagger.
946	for name in names {
947	settle(&store, &fake, name);
948	}
949	for name in names {
950	let account = crate::git::validate::name(name).unwrap();
951	let mut record = store.get(&account).unwrap().unwrap();
952	record.binary_version = Some("0.0.1-old".into());
953	store.put(&record).unwrap();
954	}
955	fake.upgrades.lock().unwrap().clear();
956
957	reconcile_once(&store, fake.as_ref());
958	assert_eq!(
959	fake.upgrades.lock().unwrap().len(),
960	UPGRADES_PER_SWEEP,
961	"restarting every tenant at once turns a deploy into an outage"
962	);
963
964	// Successive sweeps finish the rest.
965	for _ in 0..3 {
966	reconcile_once(&store, fake.as_ref());
967	}
968	assert_eq!(fake.upgrades.lock().unwrap().len(), 5);
969	}
970
971	#[test]
972	fn a_stopped_tenant_is_not_upgraded_until_it_is_needed() {
973	let (_dir, store, fake) = setup();
974	settle(&store, &fake, "alice");
975	let alice = crate::git::validate::name("alice").unwrap();
976
977	let mut record = store.get(&alice).unwrap().unwrap();
978	record.state = State::Stopped;
979	record.binary_version = Some("0.0.1-old".into());
980	store.put(&record).unwrap();
981
982	reconcile_once(&store, fake.as_ref());
983	assert!(
984	fake.upgrades.lock().unwrap().is_empty(),
985	"a stopped tenant picks up the new binary when it starts; \
986	restarting it now would wake it for nothing"
987	);
988	}
989
990	#[test]
991	fn an_idle_tenant_is_stopped_and_woken_by_the_next_request() {
992	let (_dir, store, fake) = setup();
993	let alice = crate::git::validate::name("alice").unwrap();
994	store.create(&alice).unwrap();
995	reconcile_once(&store, fake.as_ref());
996	reconcile_once(&store, fake.as_ref());
997	assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
998
999	// Long enough that nobody has used it.
1000	let later = crate::account::token::now_secs() + IDLE_STOP.as_secs() + 1;
1001	reconcile_at(&store, fake.as_ref(), later);
1002	assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Stopped);
1003
1004	// Asking for it refreshes the timestamp and reports "come back shortly".
1005	let err = store.endpoint(&alice).unwrap_err();
1006	assert_eq!(err.status(), 429);
1007
1008	reconcile_once(&store, fake.as_ref());
1009	assert_eq!(
1010	store.get(&alice).unwrap().unwrap().state,
1011	State::Ready,
1012	"a request must bring a stopped tenant back"
1013	);
1014	}
1015
1016	#[test]
1017	fn a_busy_tenant_is_not_stopped() {
1018	let (_dir, store, fake) = setup();
1019	let alice = crate::git::validate::name("alice").unwrap();
1020	store.create(&alice).unwrap();
1021	reconcile_once(&store, fake.as_ref());
1022	reconcile_once(&store, fake.as_ref());
1023
1024	// Used just now.
1025	let mut record = store.get(&alice).unwrap().unwrap();
1026	record.last_seen_at = crate::account::token::now_secs();
1027	store.put(&record).unwrap();
1028
1029	reconcile_once(&store, fake.as_ref());
1030	assert_eq!(store.get(&alice).unwrap().unwrap().state, State::Ready);
1031	}
1032
1033	#[test]
1034	fn a_ready_account_is_left_alone() {
1035	let (_dir, store, fake) = setup();
1036	let alice = crate::git::validate::name("alice").unwrap();
1037	store.create(&alice).unwrap();
1038	reconcile_once(&store, fake.as_ref());
1039	reconcile_once(&store, fake.as_ref());
1040
1041	assert_eq!(
1042	reconcile_once(&store, fake.as_ref()),
1043	0,
1044	"a settled account must not be touched every sweep"
1045	);
1046	}
1047	}