1use std::collections::{BTreeMap, BTreeSet, VecDeque};
5use std::path::{Path, PathBuf};
6use std::sync::atomic::{AtomicU64, Ordering};
7use std::sync::{Arc, Condvar, Mutex, MutexGuard};
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9
10use base64::Engine;
11use lonca_core::canonical::{digest_of, random_id};
12use lonca_core::keyed;
13use lonca_core::keys::KeyPair;
14use lonca_core::oplog::{OPLOG_REF, OpLog, SignedOp};
15use lonca_core::ops::{
16 Actor, Attestation, BindChecks, CheckResult, Claim, HumanKey, LeaseItem, MAX_UNPROPOSED_PATHS,
17 ObsKind, Observation, ObservationLevel, Op, Phase,
18};
19use lonca_core::store::CommitSpec;
20use lonca_core::{ChangeId, Error, FileMap, GitId, Result, Store};
21use lonca_policy::chain::Submission;
22use lonca_policy::config::{Profile, UnparsedFiles, Verification};
23use lonca_policy::decide::{evidence_problems, level_for, quota_problems, valid_token};
24use lonca_policy::events::{Event as SessionEvent, for_session, tell};
25use lonca_policy::harness::{Event, render, stop_message};
26use lonca_policy::human::{MAX_AGE_MS, MAX_SKEW_MS, check_fresh, verify_login};
27use lonca_policy::state::{ChangeRecord, ChangeStatus, Joins, State};
28use lonca_policy::{details, refs};
29use lonca_symbols::engine::ENGINE_CURRENT;
30use lonca_symbols::integrate::{facts_keyed, integrate_joining};
31use lonca_verify::{Sandbox, Verifier};
32use serde::de::DeserializeOwned;
33use serde_json::{Value, json};
34
35use crate::home::{Home, TokenEntry, Tokens};
36use crate::hooks;
37use crate::limits::{Bucket, Counters, Head};
38use crate::protocol::*;
39use crate::repo::{HUMAN_HEADER, POLICY_REF, RepoHandle, SESSION_HEADER, TRUNK_REF};
40
41pub type Clock = Arc<dyn Fn() -> i64 + Send + Sync>;
42
43pub fn system_clock() -> Clock {
44 Arc::new(|| {
45 SystemTime::now()
46 .duration_since(UNIX_EPOCH)
47 .map(|d| d.as_millis() as i64)
48 .unwrap_or(0)
49 })
50}
51
52#[derive(Clone, Debug)]
53pub struct Settings {
54 pub managed_settings: Option<PathBuf>,
57 pub require_root_settings: bool,
59 pub git: String,
60 pub git_timeout: std::time::Duration,
62 pub notify: crate::security::Notify,
64 pub local_roots: Option<Vec<PathBuf>>,
68 pub import_dir: Option<PathBuf>,
73 pub line_cap: usize,
76 pub proxy: Option<String>,
78 pub hardened: bool,
81}
82
83impl Settings {
84 pub fn for_daemon(
87 allow_user_owned: bool,
88 managed_settings: Option<PathBuf>,
89 proxy: Option<String>,
90 hardened: bool,
91 ) -> Self {
92 Self {
93 require_root_settings: !allow_user_owned,
94 managed_settings,
95 proxy,
96 hardened,
97 notify: crate::security::Notify::Desktop,
98 ..Self::default()
99 }
100 }
101
102 pub fn proxy_port(&self) -> Option<u16> {
104 let addr = self.proxy.as_deref()?.parse::<std::net::SocketAddr>();
105 addr.ok().map(|a| a.port())
106 }
107}
108
109impl Default for Settings {
110 fn default() -> Self {
111 Self {
112 managed_settings: None,
113 require_root_settings: true,
114 git: "git".into(),
115 git_timeout: crate::gitcmd::DEFAULT_TIMEOUT,
116 notify: crate::security::Notify::Off,
117 local_roots: None,
118 import_dir: None,
119 line_cap: MAX_LINE,
120 proxy: None,
121 hardened: false,
122 }
123 }
124}
125
126pub fn sandbox_at_start(
130 hardened: bool,
131 sandbox: Sandbox,
132 allowed: bool,
133) -> std::result::Result<Option<&'static str>, &'static str> {
134 match (sandbox, hardened, allowed) {
135 (Sandbox::None, true, false) => Err(
136 "no sandbox found: a hardened daemon does not run verification without one; \
137 install bubblewrap, or start it with --allow-unsandboxed",
138 ),
139 (Sandbox::None, true, true) => Ok(Some(
140 "--allow-unsandboxed: no sandbox found, verification commands run unsandboxed",
141 )),
142 (Sandbox::None, false, _) => Ok(Some(
143 "no sandbox found: verification commands run unsandboxed",
144 )),
145 _ => Ok(None),
146 }
147}
148
149pub fn local_roots_at_start(
155 hardened: bool,
156 roots: Vec<PathBuf>,
157) -> std::result::Result<Option<Vec<PathBuf>>, &'static str> {
158 match (hardened, roots.is_empty()) {
159 (true, false) => Err(
160 "--local-root is not for a hardened daemon: leave bundles in its import directory (--import-dir)",
161 ),
162 (true, true) => Ok(Some(roots)),
163 (false, true) => Ok(None),
164 (false, false) => Ok(Some(roots)),
165 }
166}
167
168fn log_page(limit: Option<u64>) -> usize {
170 limit.unwrap_or(LOG_PAGE).min(LOG_PAGE) as usize
171}
172
173pub fn page_budget(line_cap: usize) -> u64 {
176 (line_cap as u64 / 4) * 3
177}
178
179fn encoded_len(bytes: u64) -> u64 {
181 bytes.div_ceil(3) * 4
182}
183
184type Hr = std::result::Result<Value, (String, String)>;
185
186enum Source {
189 Given(String),
190 Copied(crate::import::Copy),
191}
192
193impl Source {
194 fn arg(&self) -> String {
197 match self {
198 Source::Given(r) => r.clone(),
199 Source::Copied(c) => c.path().to_string_lossy().into_owned(),
200 }
201 }
202}
203
204struct LeftBehind {
206 paths: Vec<String>,
208 total: u64,
209 round: u32,
211 told: bool,
213}
214
215fn showable(path: &str) -> bool {
218 lonca_core::paths::validate_repo_path(path).is_ok() && !path.chars().any(char::is_control)
219}
220
221fn unproposed(r: &RepoHandle, session: &str, files: Vec<LocalFile>) -> Result<Vec<String>> {
225 let state = &r.chain.state;
226 let standing = state.changes.values().filter(|c| {
227 matches!(
228 c.status,
229 ChangeStatus::Pending { .. } | ChangeStatus::Queued | ChangeStatus::Integrated { .. }
230 ) && c
231 .actor
232 .session()
233 .is_some_and(|s| state.continues(session, s))
234 });
235 let mut left = files;
236 for change in standing {
237 if left.is_empty() {
238 break;
239 }
240 let tree = r.store.commit_files(&change.commit)?;
241 left.retain(|f| tree.get(&f.path) != f.entry.as_ref());
242 }
243 Ok(left.into_iter().map(|f| f.path).collect())
244}
245
246fn report_unproposed(
250 r: &mut RepoHandle,
251 session: &str,
252 files: Vec<LocalFile>,
253 limit: u32,
254 now: i64,
255) -> std::result::Result<LeftBehind, (String, String)> {
256 let left = unproposed(r, session, files).or_fail()?;
257 let total = left.len() as u64;
258 let paths: Vec<String> = left
259 .into_iter()
260 .filter(|p| showable(p))
261 .take(MAX_UNPROPOSED_PATHS)
262 .collect();
263 let mark = r
264 .chain
265 .state
266 .sessions
267 .get(session)
268 .and_then(|s| s.unproposed.as_ref());
269 let rounds = mark.map_or(0, |m| m.rounds);
270 let told = total > 0 && rounds < limit;
271 let unchanged = mark.is_some_and(|m| m.total == total && m.paths == paths);
274 if told || !unchanged {
275 let op = Op::Observe {
276 session: session.to_string(),
277 events: vec![Observation {
278 phase: Phase::Post,
279 kind: ObsKind::Unproposed {
280 paths: paths.clone(),
281 total,
282 },
283 external: false,
284 }],
285 };
286 r.submit(Submission::new(Actor::Daemon, op), now)
287 .or_fail()?;
288 }
289 Ok(LeftBehind {
290 paths,
291 total,
292 round: rounds + u32::from(told),
293 told,
294 })
295}
296
297pub use lonca_client::PROTOCOL;
299
300pub const LOG_WAIT_MAX_MS: i64 = 30_000;
302pub const LOG_POLL_MS: u64 = 2_000;
306
307pub const LOG_PAGE: u64 = 1000;
309
310pub const DIFF_CAP_BYTES: usize = 2 * 1024 * 1024;
312
313pub const EVIDENCE_LOG_CAP_BYTES: usize = 32 * 1024;
317
318pub const HUMAN_SESSION_DEFAULT_MS: i64 = 60 * 60 * 1000;
320pub const HUMAN_SESSION_MAX_MS: i64 = 24 * 60 * 60 * 1000;
322pub const LOGIN_NONCE_WINDOW_MS: i64 = MAX_AGE_MS + MAX_SKEW_MS;
326
327#[derive(Clone, Copy, Debug, PartialEq, Eq)]
330pub enum Cred {
331 Public,
333 Login,
335 Agent,
337 Human,
339 AnyHuman,
341 Bootstrap,
345 Read,
347 Any,
349}
350
351impl Cred {
352 pub fn name(self) -> &'static str {
354 match self {
355 Cred::Public => "public",
356 Cred::Login => "login",
357 Cred::Agent => "agent",
358 Cred::Human => "human",
359 Cred::AnyHuman => "any human",
360 Cred::Bootstrap => "bootstrap",
361 Cred::Read => "read",
362 Cred::Any => "any",
363 }
364 }
365}
366
367const PUSH_REFS_PER_CALL: usize = 500;
369
370const LAUNCHER_METHODS: [&str; 2] = ["session.heartbeat", "session.end"];
373
374pub const METHODS: &[(&str, Cred)] = &[
378 ("ping", Cred::Public),
379 ("human.login", Cred::Login),
380 ("human.logout", Cred::Public),
381 ("whoami", Cred::Any),
382 ("repo.list", Cred::AnyHuman),
383 ("repo.prepare_init", Cred::Bootstrap),
384 ("repo.init", Cred::Bootstrap),
385 ("repo.restore", Cred::Bootstrap),
386 ("repo.backup", Cred::Human),
387 ("daemon.prepare_rotate", Cred::AnyHuman),
388 ("status", Cred::Read),
389 ("log", Cred::Read),
390 ("tree", Cred::Read),
391 ("blobs", Cred::Read),
392 ("change.get", Cred::Read),
393 ("change.diff", Cred::Read),
394 ("change.resolve", Cred::Read),
395 ("trunk.bundle", Cred::Read),
396 ("trunk.distance", Cred::Read),
397 ("human.prepare_change", Cred::Human),
398 ("human.prepare_policy", Cred::Human),
399 ("human.prepare_import", Cred::Human),
400 ("human.prepare_push", Cred::Human),
401 ("human.prepare_isolate", Cred::Human),
402 ("human.submit", Cred::Human),
403 ("lease.acquire", Cred::Agent),
404 ("lease.release", Cred::Agent),
405 ("lease.renew", Cred::Agent),
406 ("session.heartbeat", Cred::Agent),
407 ("change.propose", Cred::Agent),
408 ("change.withdraw", Cred::Agent),
409 ("facts.preview", Cred::Agent),
410 ("negotiate", Cred::Agent),
411 ("cap.attenuate", Cred::Agent),
412 ("lease.interest", Cred::Agent),
413 ("events", Cred::Agent),
414 ("inbox", Cred::Agent),
415 ("events.watch", Cred::Human),
416 ("session.end", Cred::Agent),
417 ("session.rotate", Cred::Agent),
418 ("hook.event", Cred::Agent),
419 ("hook.input", Cred::Agent),
420 ("hook.bind", Cred::Agent),
421 ("hook.stop", Cred::Agent),
422 ("mcp.call", Cred::Agent),
423 ("harness.setup", Cred::Agent),
424 ("queue.run", Cred::AnyHuman),
425 ("security.events", Cred::AnyHuman),
426];
427
428pub const LONG_LINES: &[&str] = &[
431 "tree",
432 "blobs",
433 "human.prepare_change",
434 "human.prepare_policy",
435 "human.submit",
436 "lease.acquire",
437 "lease.interest",
438 "change.propose",
439 "facts.preview",
440 "session.end",
441 "hook.event",
442 "hook.input",
443 "hook.stop",
444];
445
446const NOT_THROTTLED: [&str; 6] = [
450 "facts.preview",
451 "events",
452 "inbox",
453 "harness.setup",
454 "session.end",
455 "mcp.call",
456];
457
458#[derive(Clone, Debug)]
460enum Caller {
461 Anonymous,
462 Agent {
463 repo: String,
464 actor: Actor,
465 },
466 Human {
468 human: String,
469 key: String,
470 },
471}
472
473struct Auth {
475 caller: Caller,
476 repo: Option<String>,
478}
479
480impl Auth {
481 fn repo(&self) -> std::result::Result<String, (String, String)> {
482 self.repo
483 .clone()
484 .ok_or_else(|| ("invalid".into(), "missing repo".into()))
485 }
486
487 fn same_human(&self, signer: &str) -> std::result::Result<(), (String, String)> {
489 match &self.caller {
490 Caller::Human { human, .. } if human != signer => fail(
491 "forbidden",
492 format!("a session of {human} can not submit a signature by {signer}"),
493 ),
494 _ => Ok(()),
495 }
496 }
497}
498
499type SessionKey = (String, String);
501
502#[derive(Default)]
505struct ProxyBook {
506 open: BTreeMap<std::net::SocketAddr, Vec<(SessionKey, u64)>>,
509 carried: BTreeMap<SessionKey, u64>,
511 unattributed: u64,
514}
515
516impl ProxyBook {
517 fn open_of(&self, key: &SessionKey) -> usize {
518 let held = self.open.values().flatten();
519 held.filter(|(k, _)| k == key).count()
520 }
521}
522
523fn host_verdict(cfg: &lonca_policy::config::Observation, host: &str) -> (bool, bool) {
525 if cfg.agent_hosts.iter().any(|h| h == host) {
529 return (true, false);
530 }
531 if let Some(rule) = cfg.host_rule(host) {
532 return (true, rule.taint);
533 }
534 (cfg.network == "taint", true)
535}
536
537#[derive(Clone, Debug, PartialEq, Eq)]
538pub struct ProxyVerdict {
539 pub allow: bool,
540 pub allow_private: bool,
541 pub reason: String,
542}
543
544const AMBIGUOUS_SHOWN: usize = 5;
546
547fn resolve_prefix<'a>(ids: impl IntoIterator<Item = &'a ChangeId>, prefix: &str) -> Hr {
550 let found: Vec<&ChangeId> = ids.into_iter().filter(|c| c.starts_with(prefix)).collect();
551 match found.as_slice() {
552 [] => fail("not_found", format!("no change starts with {prefix}")),
553 [one] => Ok(json!({"change": one})),
554 many => {
555 let shown: Vec<String> = many
556 .iter()
557 .take(AMBIGUOUS_SHOWN)
558 .map(|c| c.to_string())
559 .collect();
560 let more = if many.len() > AMBIGUOUS_SHOWN {
561 ", ..."
562 } else {
563 ""
564 };
565 fail(
566 "invalid",
567 format!(
568 "{} changes start with {prefix}: {}{more}; give more letters",
569 many.len(),
570 shown.join(", ")
571 ),
572 )
573 }
574 }
575}
576
577fn fail<T>(code: &str, msg: impl Into<String>) -> std::result::Result<T, (String, String)> {
578 Err((code.into(), msg.into()))
579}
580
581fn from_err(e: Error) -> (String, String) {
582 let code = match e {
583 Error::Invalid(_) => "invalid",
584 Error::Conflict(_) => "conflict",
585 Error::Integrity(_) => "integrity",
586 Error::Unsupported(_) => "unsupported",
587 Error::Git(_) | Error::Io(_) | Error::Json(_) => "internal",
588 };
589 (code.into(), e.to_string())
590}
591
592trait OrFail<T> {
593 fn or_fail(self) -> std::result::Result<T, (String, String)>;
594}
595
596impl<T> OrFail<T> for Result<T> {
597 fn or_fail(self) -> std::result::Result<T, (String, String)> {
598 self.map_err(from_err)
599 }
600}
601
602fn params<T: DeserializeOwned>(v: &Value) -> std::result::Result<T, (String, String)> {
603 serde_json::from_value(v.clone()).map_err(|e| ("invalid".into(), format!("params: {e}")))
604}
605
606fn logged(op: &SignedOp) -> Value {
607 json!({"seq": op.entry.seq, "decision": op.entry.decision})
608}
609
610struct DaemonKeys {
613 current: KeyPair,
614 next: Option<KeyPair>,
615}
616
617impl DaemonKeys {
618 fn for_new(&self) -> &KeyPair {
620 self.next.as_ref().unwrap_or(&self.current)
621 }
622
623 fn all(&self) -> Vec<&KeyPair> {
624 std::iter::once(self.for_new())
625 .chain(self.next.as_ref().map(|_| &self.current))
626 .collect()
627 }
628}
629
630struct Inner {
631 keys: DaemonKeys,
632 repos: BTreeMap<String, RepoHandle>,
633 foreign: BTreeMap<String, RepoHandle>,
636 tokens: Tokens,
637 buckets: BTreeMap<(String, String), Bucket>,
640 login_nonces: BTreeMap<String, i64>,
642 prepared: BTreeMap<String, Op>,
643 workspaces: BTreeMap<(String, String), PathBuf>,
645}
646
647const VERIFIED_TREES: usize = 256;
649
650const VERIFIED_FOR_MS: i64 = 3_600_000;
653
654type VerifiedKey = (String, GitId, String);
655
656#[derive(Default)]
658struct Verified {
659 runs: BTreeMap<VerifiedKey, (i64, Vec<Attestation>)>,
660 order: VecDeque<VerifiedKey>,
662}
663
664impl Verified {
665 fn get(&self, key: &VerifiedKey, now: i64) -> Option<Vec<Attestation>> {
666 let fresh = self
667 .runs
668 .get(key)
669 .filter(|(at, _)| now - at < VERIFIED_FOR_MS);
670 fresh.map(|(_, run)| run.clone())
671 }
672
673 fn insert(&mut self, key: VerifiedKey, run: &[Attestation], now: i64) {
674 let entry = (now, run.to_vec());
675 if self.runs.insert(key.clone(), entry).is_none() {
676 self.order.push_back(key);
677 }
678 if self.order.len() > VERIFIED_TREES
679 && let Some(oldest) = self.order.pop_front()
680 {
681 self.runs.remove(&oldest);
682 }
683 }
684}
685
686fn environment(details: &BTreeMap<String, String>) -> Vec<Value> {
690 (0..lonca_policy::config::MAX_IDENTIFY)
691 .map_while(|n| {
692 let get = |k: &str| details.get(&format!("env.{n}.{k}"));
693 let command = get("command")?;
694 Some(json!({"command": command, "output": get("output"), "error": get("error")}))
695 })
696 .collect()
697}
698
699fn reused(mut a: Attestation) -> Attestation {
700 a.details.insert("reused".into(), "true".into());
701 a
702}
703
704fn watched(
708 state: &State,
709 human: &str,
710 named: Option<&[String]>,
711 now: i64,
712) -> std::result::Result<Vec<String>, (String, String)> {
713 let mine = |id: &str| {
714 state
715 .sessions
716 .get(id)
717 .is_some_and(|s| s.root_human == human)
718 };
719 match named {
720 Some(named) => match named.iter().find(|id| !mine(id)) {
721 Some(id) => fail("forbidden", format!("session {id} is not one you started")),
722 None => Ok(named.to_vec()),
723 },
724 None => {
725 let live = state.sessions.values().filter(|s| s.root_human == human);
726 let live = live.filter(|s| state.session_active(&s.id, now));
727 Ok(live.map(|s| s.id.clone()).collect())
728 }
729 }
730}
731
732fn ended_of(state: &State, named: Option<&[String]>, now: i64) -> Vec<EndedSession> {
736 let reason = |id: &str| {
737 let mut cur = state.sessions.get(id);
738 while let Some(s) = cur {
739 if let Some(r) = &s.ended {
740 return r.clone();
741 }
742 if now >= s.expires_ms {
743 return "expired".into();
744 }
745 cur = s.parent.as_deref().and_then(|p| state.sessions.get(p));
746 }
747 "expired".into()
748 };
749 let named = named.unwrap_or_default().iter();
750 let gone = named.filter(|id| !state.session_active(id, now));
751 gone.map(|id| EndedSession {
752 session: id.clone(),
753 reason: reason(id),
754 })
755 .collect()
756}
757
758fn ops_for_events(store: &Store, from: u64) -> Result<Vec<SignedOp>> {
762 let log = OpLog::new(store);
763 let ops = log.read_from(from)?;
764 let targets = ops.iter().filter_map(|o| match &o.entry.op {
765 Op::Approve { target_seq, .. } | Op::Reject { target_seq, .. } => Some(*target_seq),
766 _ => None,
767 });
768 match targets.min() {
769 Some(oldest) if oldest < from => log.read_from(oldest),
770 _ => Ok(ops),
771 }
772}
773
774fn watched_event(session: &str, event: &SessionEvent) -> WatchedEvent {
777 let mut out = WatchedEvent {
778 session: session.to_string(),
779 kind: String::new(),
780 seq: None,
781 change: None,
782 from: None,
783 reservation: None,
784 until_ms: None,
785 };
786 let (kind, seq) = match event {
787 SessionEvent::Integrated { seq, change } => {
788 out.change = Some(*change);
789 ("integrated", Some(*seq))
790 }
791 SessionEvent::Conflicted { seq, change, .. } => {
792 out.change = Some(*change);
793 ("conflicted", Some(*seq))
794 }
795 SessionEvent::Failed { seq, change, .. } => {
796 out.change = Some(*change);
797 ("failed", Some(*seq))
798 }
799 SessionEvent::Approved { seq, .. } => ("approved", Some(*seq)),
800 SessionEvent::Rejected { seq, .. } => ("rejected", Some(*seq)),
801 SessionEvent::Message { seq, from, .. } => {
802 out.from.clone_from(from);
803 ("message", Some(*seq))
804 }
805 SessionEvent::LeaseReceived { seq, from, .. } => {
806 out.from.clone_from(from);
807 ("lease_received", Some(*seq))
808 }
809 SessionEvent::SymbolsChanged { seq, change, .. } => {
810 out.change = Some(*change);
811 ("symbols_changed", Some(*seq))
812 }
813 SessionEvent::Reserved { until_ms, .. } => {
814 out.until_ms = Some(*until_ms);
815 ("reserved", None)
816 }
817 };
818 out.kind = kind.into();
819 out.seq = seq;
820 out
821}
822
823fn reservation_id(session: &str, items: &[LeaseItem]) -> String {
826 let mut id = digest_of(&(session, items));
827 id.truncate(16);
828 id
829}
830
831fn same_proposal(state: &State, change: ChangeId, commit: &GitId, base: &GitId) -> bool {
834 let rec = state.changes.get(&change);
835 rec.is_some_and(|c| c.commit == *commit && c.base == *base)
836}
837
838struct Running<'a> {
841 set: &'a Mutex<BTreeSet<ChangeId>>,
842 held: Vec<ChangeId>,
843}
844
845impl Running<'_> {
846 fn hold(&mut self, changes: impl IntoIterator<Item = ChangeId>) {
847 let mut set = self.set.lock().unwrap_or_else(|e| e.into_inner());
848 for c in changes {
849 set.insert(c);
850 self.held.push(c);
851 }
852 }
853}
854
855impl Drop for Running<'_> {
856 fn drop(&mut self) {
857 let mut set = self.set.lock().unwrap_or_else(|e| e.into_inner());
858 for c in &self.held {
859 set.remove(c);
860 }
861 }
862}
863
864pub struct Service {
865 pub home: Home,
866 pub counters: Counters,
867 clock: Clock,
868 verifier: Verifier,
869 verified: Mutex<Verified>,
872 proxy: Mutex<ProxyBook>,
873 security: crate::security::SecurityLog,
874 flagged: Mutex<BTreeSet<String>>,
876 verification_runs: AtomicU64,
877 running: Mutex<BTreeSet<ChangeId>>,
879 settings: Settings,
880 started_ms: i64,
883 inner: Mutex<Inner>,
884 wake: (Mutex<u64>, Condvar),
885 appended: (Mutex<()>, Condvar),
888}
889
890impl Service {
891 pub fn open(home: Home, settings: Settings, clock: Clock) -> Result<Arc<Self>> {
892 home.ensure()?;
893 let _ = std::fs::remove_dir_all(home.imports());
895 let (current, next) = home.load_keys()?;
896 let keys = DaemonKeys { current, next };
897 let tokens = Tokens::load(&home.tokens_path())?;
898 let mut repos = BTreeMap::new();
899 let mut foreign = BTreeMap::new();
900 for entry in std::fs::read_dir(home.repos())? {
901 let path = entry?.path();
902 let Some(name) = path
903 .file_name()
904 .and_then(|n| n.to_str())
905 .and_then(|n| n.strip_suffix(".git"))
906 else {
907 continue;
908 };
909 let r = RepoHandle::open(name, &path, &keys.all())?;
910 if r.foreign() {
911 eprintln!(
912 "lonca: repository {name} is on another daemon key; \
913 recover it with `lonca daemon-key recover --repo {name}`"
914 );
915 foreign.insert(name.to_string(), r);
916 } else {
917 repos.insert(name.to_string(), r);
918 }
919 }
920 let svc = Arc::new(Self {
921 verifier: Verifier::new(home.scratch()),
922 verified: Mutex::default(),
923 verification_runs: AtomicU64::new(0),
924 proxy: Mutex::default(),
925 security: crate::security::SecurityLog::new(&home.root, settings.notify.clone()),
926 flagged: Mutex::default(),
927 running: Mutex::default(),
928 home,
929 counters: Counters::default(),
930 started_ms: clock(),
931 clock,
932 settings,
933 inner: Mutex::new(Inner {
934 keys,
935 repos,
936 foreign,
937 tokens,
938 buckets: BTreeMap::new(),
939 login_nonces: BTreeMap::new(),
940 prepared: BTreeMap::new(),
941 workspaces: BTreeMap::new(),
942 }),
943 wake: (Mutex::new(0), Condvar::new()),
944 appended: (Mutex::new(()), Condvar::new()),
945 });
946 {
947 let mut inner = svc.lock();
948 let now = svc.now();
949 for r in inner.repos.values_mut() {
950 if r.initialized() {
951 Self::reconcile(r, now)
952 .map_err(|e| Error::Integrity(format!("{}: {}", e.0, e.1)))?;
953 }
954 }
955 svc.promote_if_done(&mut inner)?;
956 Self::sweep_tokens(&mut inner, now)
957 .map_err(|e| Error::Invalid(format!("{}: {}", e.0, e.1)))?;
958 }
959 Ok(svc)
960 }
961
962 fn git_runner(&self) -> crate::gitcmd::Git {
965 crate::gitcmd::Git {
966 bin: self.settings.git.clone(),
967 home: self.home.root.join("git-home"),
968 timeout: self.settings.git_timeout,
969 }
970 }
971
972 fn git(
973 &self,
974 git_dir: &std::path::Path,
975 args: &[&str],
976 ) -> std::result::Result<String, (String, String)> {
977 use crate::gitcmd::Failure;
978 self.git_runner().run(git_dir, args).map_err(|f| match f {
979 Failure::Start(e) => ("internal".to_string(), format!("git: {e}")),
980 Failure::Timeout(t) => (
981 "git".to_string(),
982 format!(
983 "git did not finish within {} s and was stopped",
984 t.as_secs()
985 ),
986 ),
987 Failure::Exit(said) => ("git".to_string(), said),
988 })
989 }
990
991 fn check_remote(&self, r: &str) -> std::result::Result<(), (String, String)> {
998 validate_remote(r)?;
999 if !r.starts_with('/') {
1000 return Ok(());
1001 }
1002 let refuse = |why: &str| fail("forbidden", format!("local remote {r}: {why}"));
1003 let Ok(path) = std::fs::canonicalize(r) else {
1004 return refuse("it does not exist");
1005 };
1006 let canon = |p: &Path| std::fs::canonicalize(p).unwrap_or_else(|_| p.to_path_buf());
1007 let backups = canon(&self.home.backups());
1008 if path.starts_with(&backups) {
1009 return Ok(());
1010 }
1011 if let Some(dir) = &self.settings.import_dir
1012 && path.starts_with(canon(dir))
1013 {
1014 return refuse("the import directory is only read, from a bundle copied first");
1015 }
1016 if path.starts_with(canon(&self.home.root)) {
1017 return refuse("inside the daemon home");
1018 }
1019 match &self.settings.local_roots {
1020 None => Ok(()),
1021 Some(roots) if roots.iter().any(|root| path.starts_with(canon(root))) => Ok(()),
1022 Some(_) => {
1023 refuse("outside the directories this daemon may read (`lonca daemon --local-root`)")
1024 }
1025 }
1026 }
1027
1028 fn source(&self, r: &str) -> std::result::Result<Source, (String, String)> {
1032 validate_remote(r)?;
1033 let Some(dir) = self.import_dir_given(r) else {
1034 self.check_remote(r)?;
1035 return Ok(Source::Given(r.to_string()));
1036 };
1037 let refuse = |why: &str| fail("forbidden", format!("local remote {r}: {why}"));
1038 let path = Path::new(r);
1039 let canon = |p: &Path| std::fs::canonicalize(p).ok();
1040 let (Some(name), Some(parent)) = (path.file_name(), path.parent()) else {
1041 return refuse("name a bundle directly in the import directory");
1042 };
1043 if canon(parent).as_deref() != Some(dir.as_path()) {
1044 return refuse("name a bundle directly in the import directory");
1045 }
1046 let into = self.home.imports();
1047 match crate::import::copy_bundle(
1048 &dir,
1049 name,
1050 &into,
1051 &random_id("imp"),
1052 crate::import::MAX_BUNDLE,
1053 ) {
1054 Ok(copy) => Ok(Source::Copied(copy)),
1055 Err(why) => refuse(&why),
1056 }
1057 }
1058
1059 fn import_dir_given(&self, r: &str) -> Option<PathBuf> {
1061 let dir = std::fs::canonicalize(self.settings.import_dir.as_ref()?).ok()?;
1062 let path = Path::new(r);
1063 let below = |p: &Path| p.starts_with(&dir);
1064 let parent = path.parent().and_then(|p| std::fs::canonicalize(p).ok());
1065 (below(path) || parent.as_deref().is_some_and(below)).then_some(dir)
1066 }
1067
1068 fn check_branch(&self, b: &str) -> std::result::Result<(), (String, String)> {
1070 validate_branch(b)?;
1071 if !crate::gitcmd::branch_ok(&self.git_runner(), b) {
1072 return fail("invalid", "invalid branch name");
1073 }
1074 Ok(())
1075 }
1076
1077 fn lock(&self) -> MutexGuard<'_, Inner> {
1078 self.inner.lock().unwrap_or_else(|e| e.into_inner())
1079 }
1080
1081 pub fn now(&self) -> i64 {
1082 (self.clock)()
1083 }
1084
1085 pub fn public_key(&self) -> String {
1087 self.lock().keys.current.public_hex()
1088 }
1089
1090 pub fn next_key(&self) -> Option<String> {
1092 self.lock().keys.next.as_ref().map(KeyPair::public_hex)
1093 }
1094
1095 fn promote_if_done(&self, inner: &mut Inner) -> Result<()> {
1098 let Some(next) = inner.keys.next.clone() else {
1099 return Ok(());
1100 };
1101 let current = inner.keys.current.public_hex();
1102 if inner
1103 .repos
1104 .values()
1105 .any(|r| r.initialized() && r.chain.public_key() == current)
1106 {
1107 return Ok(());
1108 }
1109 self.home.promote_next_key()?;
1110 inner.keys = DaemonKeys {
1111 current: next.clone(),
1112 next: None,
1113 };
1114 for r in inner.repos.values_mut().filter(|r| !r.initialized()) {
1116 r.chain.set_key(next.clone());
1117 inner.prepared.remove(&r.name);
1118 }
1119 Ok(())
1120 }
1121
1122 fn notify_queue(&self) {
1123 let (m, c) = &self.wake;
1124 *m.lock().unwrap_or_else(|e| e.into_inner()) += 1;
1125 c.notify_all();
1126 }
1127
1128 fn notify_appended(&self) {
1133 let (m, c) = &self.appended;
1134 let _held = m.lock().unwrap_or_else(|e| e.into_inner());
1135 c.notify_all();
1136 }
1137
1138 pub fn handle(&self, req: Request) -> Response {
1140 self.handle_from(req, &|| false)
1141 }
1142
1143 pub fn handle_from(&self, req: Request, in_sandbox: &dyn Fn() -> bool) -> Response {
1146 let r = match self.dispatch(&req, in_sandbox) {
1147 Ok(v) => Response::ok(req.id, v),
1148 Err((code, msg)) => {
1149 self.refusal(&req, &code, &msg);
1150 Response::err(req.id, &code, msg)
1151 }
1152 };
1153 self.take_alarms();
1154 if !matches!(
1155 req.method.as_str(),
1156 "ping"
1157 | "status"
1158 | "log"
1159 | "tree"
1160 | "change.get"
1161 | "whoami"
1162 | "repo.list"
1163 | "events.watch"
1164 ) {
1165 self.notify_appended();
1166 }
1167 r
1168 }
1169
1170 fn refusal(&self, req: &Request, code: &str, message: &str) {
1173 use crate::security::kind;
1174 let k = match code {
1175 "unauthenticated" | "forbidden" => kind::AUTH_FAILURE,
1176 "rate_limited" => kind::RATE_LIMITED,
1177 "not_in_sandbox" => kind::OUTSIDE_SANDBOX,
1178 _ => return,
1179 };
1180 let mut e = crate::security::Event::new(self.now(), k)
1181 .method(&req.method)
1182 .code(code)
1183 .detail(message);
1184 if let Some(repo) = &req.repo {
1185 e = e.repo(repo);
1186 }
1187 self.security.record(e);
1188 }
1189
1190 pub fn security_event(&self, event: crate::security::Event) {
1192 self.security.record(event);
1193 }
1194
1195 fn take_alarms(&self) {
1197 let taken: Vec<crate::security::Event> = {
1198 let mut guard = self.lock();
1199 let inner = &mut *guard;
1200 let repos = inner.repos.values_mut().chain(inner.foreign.values_mut());
1201 repos.flat_map(|r| std::mem::take(&mut r.alarms)).collect()
1202 };
1203 for e in taken {
1204 self.security.record(e);
1205 }
1206 }
1207
1208 fn security_events(&self, auth: &Auth, p: SecurityEvents) -> Hr {
1212 let Caller::Human { human, key } = &auth.caller else {
1213 return fail("forbidden", "a human session is required");
1214 };
1215 let mine = {
1216 let inner = self.lock();
1217 Self::enrolled_repos(&inner, human, key)
1218 };
1219 let since = p.since_ms.unwrap_or_else(|| self.now() - 86_400_000);
1220 let events: Vec<crate::security::Event> = self
1221 .security
1222 .read_since(since)
1223 .into_iter()
1224 .filter(|e| e.repo.as_ref().is_none_or(|r| mine.contains(r)))
1225 .collect();
1226 let counts = crate::security::summary(&events);
1227 let latest = &events[events.len().saturating_sub(100)..];
1228 Ok(json!({"since_ms": since, "counts": counts, "events": latest}))
1229 }
1230
1231 fn hold_to_sandbox(
1235 &self,
1236 req: &Request,
1237 in_sandbox: &dyn Fn() -> bool,
1238 ) -> std::result::Result<(), (String, String)> {
1239 if LAUNCHER_METHODS.contains(&req.method.as_str()) {
1240 return Ok(());
1241 }
1242 let mut guard = self.lock();
1243 let now = self.now();
1244 let (repo, actor) = Self::agent(&guard, req, now)?;
1245 let session = actor.session().expect("agent").to_string();
1246 let r = Self::repo(&mut guard, &repo)?;
1247 let held = r.chain.state.sessions.get(&session);
1248 let Some(s) = held.filter(|s| s.level == ObservationLevel::Isolated) else {
1249 return Ok(());
1250 };
1251 let clean = s.taint.is_none();
1252 if in_sandbox() {
1253 return Ok(());
1254 }
1255 if clean {
1256 let t = Op::Taint {
1257 session,
1258 reason: "its token was used outside its sandbox".into(),
1259 };
1260 r.submit(Submission::new(Actor::Daemon, t), now).or_fail()?;
1261 }
1262 fail(
1263 "not_in_sandbox",
1264 "this session runs in a sandbox, and this request came from outside it",
1265 )
1266 }
1267
1268 fn throttle(&self, req: &Request) -> std::result::Result<(), (String, String)> {
1273 let mut guard = self.lock();
1274 let inner = &mut *guard;
1275 let now = self.now();
1276 let (repo, actor) = Self::agent(inner, req, now)?;
1277 let session = actor.session().expect("agent");
1278 let r = Self::repo(inner, &repo)?;
1279 let rate = r.policy().or_fail()?.config.sessions.ops_per_minute as u64;
1280 let root = r.chain.state.root_session(session).to_string();
1281 let bucket = inner.buckets.entry((repo, root));
1282 let bucket = bucket.or_insert_with(|| Bucket::full(rate, now));
1283 bucket.take(rate, now).map_err(|wait_ms| {
1284 Counters::add(&self.counters.rate_limited);
1285 let msg = format!("this session is over its rate; try again in {wait_ms} ms");
1286 ("rate_limited".to_string(), msg)
1287 })
1288 }
1289
1290 pub fn long_line(&self, head: &Head) -> std::result::Result<(), (String, String)> {
1293 let method = head.method.as_deref().unwrap_or_default();
1294 let found = METHODS.iter().find(|(m, _)| *m == method);
1295 let Some(&(_, cred)) = found.filter(|_| LONG_LINES.contains(&method)) else {
1296 return fail("too_large", "this request line is too long");
1297 };
1298 if head.token.is_none() {
1299 return fail("too_large", "a long line names its method and token first");
1300 }
1301 let req = Request {
1302 id: head.id,
1303 method: method.into(),
1304 repo: head.repo.clone(),
1305 token: head.token.clone(),
1306 params: Value::Null,
1307 };
1308 self.authorize(&req, cred).map(|_| ())
1309 }
1310
1311 fn dispatch(&self, req: &Request, in_sandbox: &dyn Fn() -> bool) -> Hr {
1312 let p = &req.params;
1313 let Some(&(_, cred)) = METHODS.iter().find(|(m, _)| *m == req.method) else {
1314 return fail("unknown_method", format!("unknown method {}", req.method));
1315 };
1316 let by_recovery_key = p["op"]["op"] == "human.recover"
1320 || (p["op"]["op"] == "daemon.recover" && !p["op"]["recover"].is_null());
1321 let cred = if req.method == "human.submit" && by_recovery_key {
1322 Cred::Public
1323 } else {
1324 cred
1325 };
1326 let auth = self.authorize(req, cred)?;
1327 if cred == Cred::Agent {
1328 self.hold_to_sandbox(req, in_sandbox)?;
1329 if !NOT_THROTTLED.contains(&req.method.as_str()) {
1330 self.throttle(req)?;
1331 }
1332 }
1333 match req.method.as_str() {
1334 "ping" => Ok(json!({
1335 "version": env!("CARGO_PKG_VERSION"),
1336 "protocol": PROTOCOL,
1337 "daemon_key": self.public_key(),
1338 "next_key": self.next_key(),
1339 "proxy": self.settings.proxy,
1340 "hardened": self.settings.hardened,
1341 "refused": self.counters.to_json(),
1342 })),
1343 "human.login" => self.login(req, params(p)?),
1344 "human.logout" => self.logout(req),
1345 "daemon.prepare_rotate" => self.prepare_rotate(),
1346 "repo.list" => self.repo_list(&auth),
1347 "security.events" => self.security_events(&auth, params(p)?),
1348 "repo.prepare_init" => self.prepare_init(params(p)?),
1349 "repo.init" => self.init(&auth, params(p)?),
1350 "repo.backup" => self.backup(&auth),
1351 "repo.restore" => self.restore(params(p)?),
1352 "status" => self.status(&auth),
1353 "log" => self.log(&auth, params(p)?),
1354 "tree" => self.tree(&auth, params(p)?),
1355 "blobs" => self.blobs(&auth, params(p)?),
1356 "change.get" => self.change_get(&auth, params(p)?),
1357 "change.diff" => self.change_diff(&auth, params(p)?),
1358 "change.resolve" => self.change_resolve(&auth, params(p)?),
1359 "trunk.bundle" => self.trunk_bundle(&auth, params(p)?),
1360 "trunk.distance" => self.trunk_distance(&auth, params(p)?),
1361 "whoami" => self.whoami(&auth),
1362 "human.prepare_change" => self.prepare_change(&auth, params(p)?),
1363 "human.prepare_policy" => self.prepare_policy(&auth, params(p)?),
1364 "human.prepare_import" => self.prepare_import(&auth, params(p)?),
1365 "human.prepare_push" => self.prepare_push(&auth, params(p)?),
1366 "human.prepare_isolate" => self.prepare_isolate(&auth, params(p)?),
1367 "human.submit" => self.human_submit(&auth, params(p)?),
1368 "lease.acquire" => self.lease_acquire(req, params(p)?),
1369 "lease.release" => {
1370 let s: SetRef = params(p)?;
1371 self.agent_op(req, |_| Ok(Op::LeaseRelease { set: s.set.clone() }))
1372 }
1373 "lease.renew" => {
1374 let s: SetRef = params(p)?;
1375 let ttl = s.ttl_ms.unwrap_or(10 * 60_000);
1376 self.agent_op(req, |_| {
1377 Ok(Op::LeaseRenew {
1378 set: s.set.clone(),
1379 ttl_ms: ttl,
1380 })
1381 })
1382 }
1383 "change.propose" => self.propose(req, params(p)?),
1384 "change.withdraw" => {
1385 let c: ChangeRef = params(p)?;
1386 self.agent_op(req, |_| {
1388 if self.running_lock().contains(&c.change) {
1389 return Err((
1390 "verifying".into(),
1391 "its checks are running; withdraw it once they end".into(),
1392 ));
1393 }
1394 Ok(Op::ChangeWithdraw { change: c.change })
1395 })
1396 }
1397 "facts.preview" => self.facts_preview(req, params(p)?),
1398 "negotiate" => {
1399 let n: Negotiate = params(p)?;
1400 self.agent_op(req, |_| {
1401 Ok(Op::Negotiate {
1402 thread: n.thread.clone(),
1403 to_session: n.to_session.clone(),
1404 message: n.message.clone(),
1405 })
1406 })
1407 }
1408 "cap.attenuate" => self.attenuate(req, params(p)?),
1409 "lease.interest" => self.lease_interest(req, params(p)?),
1410 "events" => self.events(req, params(p)?),
1411 "inbox" => self.inbox(req),
1412 "events.watch" => self.watch(&auth, params(p)?),
1413 "session.end" => self.agent_end(req, params(p)?),
1414 "session.heartbeat" => self.heartbeat(req),
1415 "session.rotate" => self.rotate(req),
1416 "hook.event" => self.hook_event(req, params(p)?),
1417 "hook.input" => self.hook_input(req, params(p)?),
1418 "hook.bind" => self.hook_bind(req, params(p)?),
1419 "hook.stop" => self.hook_stop(req, params(p)?),
1420 "mcp.call" => self.mcp_call(req, params(p)?),
1421 "harness.setup" => self.harness_setup(req, params(p)?),
1422 "queue.run" => {
1423 self.notify_queue();
1424 Ok(json!({"integrated": 0, "woken": true}))
1425 }
1426 m => fail("unknown_method", format!("unknown method {m}")),
1427 }
1428 }
1429
1430 fn repo_name(req: &Request) -> std::result::Result<String, (String, String)> {
1431 req.repo
1432 .clone()
1433 .ok_or_else(|| ("invalid".into(), "missing repo".into()))
1434 }
1435
1436 fn repo<'a>(
1437 inner: &'a mut Inner,
1438 name: &str,
1439 ) -> std::result::Result<&'a mut RepoHandle, (String, String)> {
1440 if let Some(r) = inner.foreign.get(name) {
1441 return fail(
1442 "foreign_key",
1443 format!(
1444 "repository {name} is on daemon key {}, which this daemon does not hold; \
1445 recover it with `lonca daemon-key recover --repo {name}`, or, if your \
1446 key was lost too, `lonca human recover --repo {name} --recovery-key <file>`",
1447 r.chain.state.daemon_key.get(..16).unwrap_or_default()
1448 ),
1449 );
1450 }
1451 match inner.repos.get_mut(name) {
1452 Some(r) if r.initialized() => Ok(r),
1453 _ => fail("not_found", format!("no repository {name}")),
1454 }
1455 }
1456
1457 fn caller(
1461 inner: &Inner,
1462 req: &Request,
1463 now: i64,
1464 ) -> std::result::Result<Caller, (String, String)> {
1465 let Some(token) = req.token.as_deref() else {
1466 return Ok(Caller::Anonymous);
1467 };
1468 match inner.tokens.lookup(token).cloned() {
1469 None if token.starts_with("lonca_h_") => fail(
1470 "unauthenticated",
1471 "unknown human session token: it expired or was ended; log in again (`lonca human login`)",
1472 ),
1473 None => fail(
1474 "unauthenticated",
1475 "unknown session token: its session ended or expired, or it has a new token; \
1476 a human continues the workspace with `lonca session start --continue`",
1477 ),
1478 Some(TokenEntry::Agent {
1479 repo,
1480 session,
1481 expires_ms,
1482 }) => {
1483 let past_end = expires_ms.is_some_and(|end| end <= now);
1484 let Some(r) = inner.repos.get(&repo) else {
1485 return fail("unauthenticated", "token for unknown repository");
1486 };
1487 let Some(s) = r.chain.state.sessions.get(&session) else {
1488 return fail("unauthenticated", "unknown session");
1489 };
1490 if s.ended.is_some() {
1491 return fail("unauthenticated", "session ended");
1492 }
1493 if (past_end || !r.chain.state.session_active(&session, now))
1497 && !LAUNCHER_METHODS.contains(&req.method.as_str())
1498 {
1499 return fail(
1500 "unauthenticated",
1501 "session expired; a human continues the workspace with \
1502 `lonca session start --continue`",
1503 );
1504 }
1505 Ok(Caller::Agent {
1506 repo,
1507 actor: Actor::Agent {
1508 name: s.agent.clone(),
1509 session,
1510 },
1511 })
1512 }
1513 Some(TokenEntry::Human {
1514 human,
1515 key,
1516 expires_ms,
1517 }) => {
1518 if expires_ms <= now {
1519 return fail("unauthenticated", "human session expired");
1520 }
1521 Ok(Caller::Human { human, key })
1522 }
1523 }
1524 }
1525
1526 fn agent(
1528 inner: &Inner,
1529 req: &Request,
1530 now: i64,
1531 ) -> std::result::Result<(String, Actor), (String, String)> {
1532 match Self::caller(inner, req, now)? {
1533 Caller::Agent { repo, actor } => Ok((repo, actor)),
1534 Caller::Human { .. } => fail("forbidden", "an agent session token is required"),
1535 Caller::Anonymous => fail("unauthenticated", "missing session token"),
1536 }
1537 }
1538
1539 fn enrolled(inner: &Inner, human: &str, key: &str, repo: &str) -> bool {
1542 inner
1543 .repos
1544 .get(repo)
1545 .or_else(|| inner.foreign.get(repo))
1546 .and_then(|r| r.chain.state.humans.get(human))
1547 .is_some_and(|k| k.public == key)
1548 }
1549
1550 fn enrolled_repos(inner: &Inner, human: &str, key: &str) -> Vec<String> {
1552 inner
1553 .repos
1554 .keys()
1555 .chain(inner.foreign.keys())
1556 .filter(|r| Self::enrolled(inner, human, key, r))
1557 .cloned()
1558 .collect()
1559 }
1560
1561 fn bootstrap(inner: &Inner) -> bool {
1563 inner.repos.values().all(|r| !r.initialized()) && inner.foreign.is_empty()
1564 }
1565
1566 fn authorize(&self, req: &Request, cred: Cred) -> std::result::Result<Auth, (String, String)> {
1569 let inner = self.lock();
1570 let caller = Self::caller(&inner, req, self.now())?;
1571 const AGENT: &str = "an agent session token is required";
1572 const HUMAN: &str = "a human session is required (lonca human login)";
1573 const EITHER: &str =
1574 "a session token is required: an agent's, or a human session (lonca human login)";
1575 let repo = match (cred, &caller) {
1576 (Cred::Public | Cred::Login, _) => req.repo.clone(),
1577 (Cred::Any, Caller::Anonymous) => return fail("unauthenticated", EITHER),
1578 (Cred::Any, Caller::Human { .. }) => req.repo.clone(),
1579 (Cred::Agent | Cred::Read | Cred::Any, Caller::Agent { repo, .. }) => {
1580 if req.repo.as_ref().is_some_and(|r| r != repo) {
1581 return fail("forbidden", "this session token is for another repository");
1582 }
1583 Some(repo.clone())
1584 }
1585 (Cred::Agent, Caller::Human { .. }) => return fail("forbidden", AGENT),
1586 (Cred::Agent, Caller::Anonymous) => return fail("unauthenticated", AGENT),
1587 (Cred::Human | Cred::Read, Caller::Human { human, key }) => {
1588 let name = Self::repo_name(req)?;
1589 if !Self::enrolled(&inner, human, key, &name) {
1590 return fail(
1591 "forbidden",
1592 format!("this human session is not enrolled in repository {name}"),
1593 );
1594 }
1595 Some(name)
1596 }
1597 (Cred::Read, Caller::Anonymous) => return fail("unauthenticated", EITHER),
1598 (Cred::Human | Cred::AnyHuman | Cred::Bootstrap, Caller::Agent { .. }) => {
1599 return fail("forbidden", HUMAN);
1600 }
1601 (Cred::AnyHuman | Cred::Bootstrap, Caller::Human { human, key }) => {
1602 if Self::enrolled_repos(&inner, human, key).is_empty() {
1603 return fail(
1604 "forbidden",
1605 "this human session is not enrolled in any repository",
1606 );
1607 }
1608 req.repo.clone()
1609 }
1610 (Cred::Bootstrap, Caller::Anonymous) if Self::bootstrap(&inner) => req.repo.clone(),
1611 (Cred::Human | Cred::AnyHuman | Cred::Bootstrap, Caller::Anonymous) => {
1612 return fail("unauthenticated", HUMAN);
1613 }
1614 };
1615 Ok(Auth { caller, repo })
1616 }
1617
1618 fn login(&self, req: &Request, p: HumanLogin) -> Hr {
1624 let ttl = p.ttl_ms.unwrap_or(HUMAN_SESSION_DEFAULT_MS);
1625 if ttl <= 0 || ttl > HUMAN_SESSION_MAX_MS {
1626 return fail(
1627 "invalid",
1628 format!("ttl_ms must be between 1 and {HUMAN_SESSION_MAX_MS}"),
1629 );
1630 }
1631 let now = self.now();
1632 check_fresh(&p.sig, now).map_err(|e| ("unauthenticated".to_string(), e.to_string()))?;
1633 let mut guard = self.lock();
1634 let inner = &mut *guard;
1635 inner
1636 .login_nonces
1637 .retain(|_, issued| *issued >= now - LOGIN_NONCE_WINDOW_MS);
1638 if inner.login_nonces.contains_key(&p.sig.nonce) {
1639 return fail("unauthenticated", "login signature already used");
1640 }
1641 let human = &p.sig.human;
1642 let repo = req.repo.as_deref();
1643 let mut keys: Vec<HumanKey> = Vec::new();
1644 if valid_token(human, 64) {
1645 let handles = inner.repos.values().chain(inner.foreign.values());
1646 for r in handles.filter(|r| repo.is_none_or(|n| n == r.name)) {
1647 if let Some(k) = r.chain.state.humans.get(human)
1648 && !keys.contains(k)
1649 {
1650 keys.push(k.clone());
1651 }
1652 }
1653 }
1654 let Some(key) = keys
1655 .iter()
1656 .find(|k| verify_login(k, repo, ttl, &p.sig, now).is_ok())
1657 else {
1658 return fail(
1659 "unauthenticated",
1660 "login failed: no enrolled key of that human verifies this signature",
1661 );
1662 };
1663 inner
1664 .login_nonces
1665 .insert(p.sig.nonce.clone(), p.sig.issued_ms);
1666 Self::sweep_tokens(inner, now)?;
1667 let expires_ms = now + ttl;
1668 let token = inner
1669 .tokens
1670 .issue_human(human, &key.public, expires_ms)
1671 .or_fail()?;
1672 Ok(json!({"token": token, "human": human, "expires_ms": expires_ms}))
1673 }
1674
1675 fn logout(&self, req: &Request) -> Hr {
1678 let Some(token) = req.token.as_deref() else {
1679 return fail("invalid", "missing token");
1680 };
1681 let mut inner = self.lock();
1682 if matches!(inner.tokens.lookup(token), Some(TokenEntry::Agent { .. })) {
1683 return fail("forbidden", "agents end their session with session.end");
1684 }
1685 inner.tokens.revoke(token).or_fail()?;
1686 Ok(json!({"ended": true}))
1687 }
1688
1689 fn repo_list(&self, auth: &Auth) -> Hr {
1693 let Caller::Human { human, key } = &auth.caller else {
1694 return fail("forbidden", "a human session is required");
1695 };
1696 let inner = self.lock();
1697 let repos: Vec<&String> = inner
1698 .repos
1699 .keys()
1700 .filter(|r| Self::enrolled(&inner, human, key, r))
1701 .collect();
1702 Ok(json!(repos))
1703 }
1704
1705 fn prepare_init(&self, p: PrepareInit) -> Hr {
1708 if !lonca_policy::decide::valid_token(&p.repo, 64)
1709 || !lonca_policy::decide::valid_token(&p.human, 64)
1710 {
1711 return fail(
1712 "invalid",
1713 "repository and human names: letters, digits, '.', '_', '-'",
1714 );
1715 }
1716 let (profile, presence) = self.starting_profile(&p)?;
1717 let src = p
1720 .import_from
1721 .as_deref()
1722 .map(|s| self.source(s))
1723 .transpose()?;
1724 let mut guard = self.lock();
1725 let inner = &mut *guard;
1726 if inner.repos.get(&p.repo).is_some_and(|r| r.initialized())
1727 || inner.foreign.contains_key(&p.repo)
1728 {
1729 return fail("conflict", format!("repository {} exists", p.repo));
1730 }
1731 let now = self.now();
1732 if !inner.repos.contains_key(&p.repo) {
1733 let path = self.home.repo_path(&p.repo);
1734 let r = if path.exists() {
1735 RepoHandle::open(&p.repo, &path, &inner.keys.all())
1736 } else {
1737 RepoHandle::create(&p.repo, &path, inner.keys.for_new().clone())
1738 }
1739 .or_fail()?;
1740 inner.repos.insert(p.repo.clone(), r);
1741 }
1742 let r = inner.repos.get_mut(&p.repo).expect("inserted");
1743 r.chain.set_key(inner.keys.for_new().clone());
1744 let trunk = match &src {
1745 Some(src) => {
1746 let branch = p.branch.clone().unwrap_or_else(|| "main".into());
1747 self.check_branch(&branch)?;
1748 self.git(
1749 r.store.path(),
1750 &[
1751 "fetch",
1752 "--no-tags",
1753 "--",
1754 &src.arg(),
1755 &format!("+{branch}:refs/lonca/import/initial"),
1756 ],
1757 )?;
1758 r.store
1759 .ref_get("refs/lonca/import/initial")
1760 .or_fail()?
1761 .ok_or(("invalid".into(), "import fetched nothing".into()))?
1762 }
1763 None => {
1764 let tree = r.store.empty_tree().or_fail()?;
1765 r.store
1766 .write_commit(&CommitSpec {
1767 tree,
1768 parents: vec![],
1769 author_name: p.human.clone(),
1770 author_email: format!("{}@humans.lonca.invalid", p.human),
1771 time_ms: now,
1772 message: "lonca: initial commit\n".into(),
1773 headers: vec![],
1774 })
1775 .or_fail()?
1776 }
1777 };
1778 let (config, rules) = profile.files(presence);
1779 let policy = r.write_policy(&config, &rules, None, now).or_fail()?;
1780 let max_file = lonca_policy::config::Config::default()
1784 .changes
1785 .max_file_bytes;
1786 let budget = page_budget(self.settings.line_cap);
1787 let mut large_files = Vec::new();
1788 let mut tree_bytes = 0u64;
1789 for (path, e) in r.store.commit_files(&trunk).or_fail()? {
1790 if e.mode == lonca_core::FileMode::Submodule {
1791 continue;
1792 }
1793 let size = r.store.blob_size(&e.blob).or_fail()?;
1794 tree_bytes += size;
1795 if size > max_file {
1796 large_files.push(json!({"path": path, "size": size}));
1797 }
1798 }
1799 let mut warnings: Vec<String> = large_files
1800 .iter()
1801 .map(|f| {
1802 format!(
1803 "{} is {} bytes, above the change cap of {max_file}: a workspace gets it, a proposal can not change it",
1804 f["path"].as_str().unwrap_or_default(),
1805 f["size"]
1806 )
1807 })
1808 .collect();
1809 if encoded_len(tree_bytes) > budget {
1810 warnings.push(format!(
1811 "the tree is {tree_bytes} bytes, more than one page ({budget} encoded): workspaces fetch it in pages"
1812 ));
1813 }
1814 let op = Op::RepoInit {
1815 repo: p.repo.clone(),
1816 daemon_key: r.chain.public_key(),
1817 human: p.human,
1818 human_key: p.human_key,
1819 policy,
1820 trunk,
1821 };
1822 inner.prepared.insert(p.repo, op.clone());
1823 Ok(json!({
1824 "op": op, "warnings": warnings, "large_files": large_files, "tree_bytes": tree_bytes,
1825 "profile": profile.name(),
1826 }))
1827 }
1828
1829 fn starting_profile(
1834 &self,
1835 p: &PrepareInit,
1836 ) -> std::result::Result<(Profile, bool), (String, String)> {
1837 let own = if self.settings.hardened {
1838 Profile::Hardened
1839 } else {
1840 Profile::Cooperative
1841 };
1842 let profile = p.profile.unwrap_or(own);
1843 if profile == Profile::Cooperative {
1844 return match p.without_presence {
1845 true => fail(
1846 "invalid",
1847 "--without-presence goes with the hardened profile",
1848 ),
1849 false => Ok((profile, false)),
1850 };
1851 }
1852 if !p.without_presence && !lonca_policy::human::proves_presence(&p.human_key.alg) {
1853 return fail(
1854 "invalid",
1855 "a hardened repository requires presence and this key can not prove it: \
1856 use `lonca human keygen --secure-enclave` (Mac) or `--security-key` (a \
1857 FIDO security key), or with neither pass --without-presence",
1858 );
1859 }
1860 Ok((profile, !p.without_presence))
1861 }
1862
1863 fn init(&self, auth: &Auth, p: HumanSubmit) -> Hr {
1864 let Op::RepoInit {
1865 repo,
1866 human,
1867 trunk,
1868 policy,
1869 ..
1870 } = &p.op
1871 else {
1872 return fail("invalid", "expected repo.init");
1873 };
1874 auth.same_human(&p.sig.human)?;
1875 let mut inner = self.lock();
1876 if inner.prepared.get(repo) != Some(&p.op) {
1877 return fail(
1878 "invalid",
1879 "this init was not prepared by this daemon; call repo.prepare_init",
1880 );
1881 }
1882 let now = self.now();
1883 let r = inner.repos.get_mut(repo).expect("prepared repos exist");
1884 let mut sub = Submission::new(
1885 Actor::Human {
1886 name: human.clone(),
1887 },
1888 p.op.clone(),
1889 );
1890 sub.human = Some(p.sig);
1891 let dry = r.dry_run(sub.clone(), now).or_fail()?;
1892 if !dry.entry.decision.is_allow() {
1893 return fail("denied", dry.entry.decision.reasons.join(", "));
1894 }
1895 let op = r.submit(sub, now).or_fail()?;
1896 r.store.ref_cas(TRUNK_REF, trunk, None).or_fail()?;
1897 r.store.ref_cas(POLICY_REF, policy, None).or_fail()?;
1898 inner.prepared.remove(repo);
1899 Ok(logged(&op))
1900 }
1901
1902 fn status(&self, auth: &Auth) -> Hr {
1905 let name = auth.repo()?;
1906 let mut inner = self.lock();
1907 let now = self.now();
1908 let r = Self::repo(&mut inner, &name)?;
1909 let policy = r.policy().ok();
1910 let unparsed = policy.as_ref().map(|p| p.config.leases.unparsed_files);
1911 let keyed: Option<serde_json::Map<String, Value>> = policy.as_ref().map(|p| {
1913 let dirs = p.config.paths.keyed.keys();
1914 dirs.map(|d| (d.clone(), json!(lonca_core::keyed::container(d))))
1915 .collect()
1916 });
1917 let policy_checks: Option<Vec<&String>> = policy
1919 .as_ref()
1920 .map(|p| p.config.verification.checks.keys().collect());
1921 let below_hardened = match &policy {
1922 Some(p) if self.settings.hardened => p.below_hardened(),
1923 _ => vec![],
1924 };
1925 let s = &r.chain.state;
1926 let runs = |c: &ChangeRecord| policy.as_ref().map(|p| runs_for(&p.config.verification, c));
1927 let sessions: Vec<Value> = s
1928 .sessions
1929 .values()
1930 .map(|x| {
1931 json!({
1932 "id": x.id, "agent": x.agent, "task": x.task, "parent": x.parent, "level": x.level,
1933 "harness": x.harness,
1934 "taint": x.taint, "active": s.session_active(&x.id, now), "expires_ms": x.expires_ms,
1935 "last_seen_ms": x.heartbeat.map(|h| h.seen_ms),
1936 "verification_ms": s.session_spent(&x.id),
1937 "continued_from": x.continued_from, "continued_by": x.continued_by,
1938 "unproposed": x.unproposed,
1939 "bound_ms": x.bound_ms, "unhooked": x.unhooked,
1940 "last_hook_ms": r.hooked.get(&x.id).and_then(|h| h.last_ms),
1941 })
1942 })
1943 .collect();
1944 let leases: Vec<Value> = s
1945 .active_leases(now)
1946 .map(|a| {
1947 let l = a.lease;
1948 let pinned: Vec<Value> = l
1949 .pins
1950 .iter()
1951 .map(|(change, items)| json!({"change": change, "items": items}))
1952 .collect();
1953 json!({"set": l.set, "session": a.holder, "items": a.items, "intent": l.intent, "expires_ms": l.expires_ms, "pinned": pinned})
1954 })
1955 .collect();
1956 let changes: Vec<Value> = s
1957 .changes
1958 .values()
1959 .map(|c| {
1960 json!({
1961 "change": c.change, "actor": c.actor.label(), "status": c.status, "provenance": c.provenance,
1962 "commit": c.commit, "base": c.base, "intent": c.claim.intent, "created_ms": c.created_ms,
1963 "version": c.version, "paths": touched_paths(&c.facts),
1964 "checks": c.checks, "runs": runs(c),
1965 })
1966 })
1967 .collect();
1968 let pending: Vec<Value> = s
1971 .pending
1972 .values()
1973 .map(|p| {
1974 let session = p.actor.session().and_then(|id| s.sessions.get(id));
1975 let (change, base, intent) = match &p.op {
1976 Op::ChangePropose {
1977 change, base, claim, ..
1978 } => (Some(*change), Some(base.clone()), Some(claim.intent.clone())),
1979 _ => (None, None, None),
1980 };
1981 let record = change.and_then(|c| s.changes.get(&c));
1982 json!({
1983 "seq": p.seq, "created_ms": p.time_ms, "actor": p.actor.label(), "op": p.op.kind().as_str(),
1984 "reasons": p.reasons, "approvals": p.approvals, "required": p.required,
1985 "change": change, "base": base, "intent": intent,
1986 "agent": session.map(|x| &x.agent), "task": session.map(|x| &x.task),
1987 "paths": p.facts.as_ref().map(touched_paths).unwrap_or_default(),
1988 "checks": p.checks, "runs": record.and_then(runs),
1989 })
1990 })
1991 .collect();
1992 let humans: serde_json::Map<String, Value> = s
1995 .humans
1996 .iter()
1997 .map(|(n, k)| {
1998 (
1999 n.clone(),
2000 json!({"alg": k.alg, "public": k.public, "recovery": s.recovery.get(n).map(|r| &r.public)}),
2001 )
2002 })
2003 .collect();
2004 let interests: Vec<Value> = s
2005 .interests
2006 .iter()
2007 .map(|(seq, i)| json!({"seq": seq, "session": i.session, "items": i.items, "intent": i.intent, "blocks_until_ms": i.blocks_until_ms}))
2008 .collect();
2009 Ok(json!({
2010 "interests": interests, "reservations": s.reservations_at(now),
2011 "repo": s.repo, "trunk": s.trunk, "policy": s.policy, "policy_checks": policy_checks,
2012 "frozen": s.frozen, "queue": s.queue,
2013 "verification_day_ms": s.day_spent(now),
2014 "unparsed_files": unparsed, "keyed": keyed, "below_hardened": below_hardened,
2015 "humans": humans, "sessions": sessions, "leases": leases, "changes": changes, "pending": pending,
2016 "next_seq": s.next_seq, "head_hash": s.head_hash, "now_ms": now, "daemon_key": s.daemon_key,
2017 "proxy_unattributed": self.proxy_unattributed(),
2018 }))
2019 }
2020
2021 pub fn clear_left_over_verifications(&self) -> usize {
2026 self.verifier.clear_left_over()
2027 }
2028
2029 pub fn sandbox_warnings(&self, sandbox: Sandbox) -> Vec<String> {
2032 if sandbox != Sandbox::None {
2033 return vec![];
2034 }
2035 let mut inner = self.lock();
2036 let mut out = Vec::new();
2037 for (name, r) in &mut inner.repos {
2038 if r.policy()
2039 .is_ok_and(|p| p.config.verification.require_sandbox)
2040 {
2041 out.push(format!(
2042 "{name} requires a sandbox and there is none: \
2043 its changes will end in unsandboxed_verification"
2044 ));
2045 }
2046 }
2047 out
2048 }
2049
2050 fn log(&self, auth: &Auth, q: LogQuery) -> Hr {
2054 let name = auth.repo()?;
2055 let wait = q.wait_ms.unwrap_or(0).clamp(0, LOG_WAIT_MAX_MS);
2056 let deadline = std::time::Instant::now() + Duration::from_millis(wait as u64);
2057 let store = loop {
2058 let mut inner = self.lock();
2059 let r = Self::repo(&mut inner, &name)?;
2060 if r.chain.state.next_seq > q.from {
2061 break r.store.clone();
2062 }
2063 drop(inner);
2064 let left = deadline.saturating_duration_since(std::time::Instant::now());
2065 if left.is_zero() {
2066 return Ok(json!([]));
2067 }
2068 let (m, c) = &self.appended;
2069 let held = m.lock().unwrap_or_else(|e| e.into_inner());
2070 let _ = c
2071 .wait_timeout(held, left.min(Duration::from_millis(LOG_POLL_MS)))
2072 .unwrap_or_else(|e| e.into_inner());
2073 };
2074 let ops = OpLog::new(&store).read_from(q.from).or_fail()?;
2075 Ok(json!(ops[..log_page(q.limit).min(ops.len())]))
2076 }
2077
2078 fn tree(&self, auth: &Auth, q: TreeQuery) -> Hr {
2079 let name = auth.repo()?;
2080 let mut inner = self.lock();
2081 let r = Self::repo(&mut inner, &name)?;
2082 let store = r.store.clone();
2083 let redacted = r.chain.state.redacted.clone();
2084 drop(inner);
2085 let files = store.commit_files(&q.commit).or_fail()?;
2086 let mut out = Vec::new();
2087 for (path, e) in files {
2088 if let Some(only) = &q.paths
2089 && !only.contains(&path)
2090 {
2091 continue;
2092 }
2093 let size = if e.mode == lonca_core::FileMode::Submodule {
2094 0
2095 } else {
2096 store.blob_size(&e.blob).or_fail()?
2097 };
2098 let content_b64 = if q.with_content
2099 && e.mode != lonca_core::FileMode::Submodule
2100 && !redacted.contains(&e.blob)
2101 {
2102 Some(
2103 base64::engine::general_purpose::STANDARD
2104 .encode(store.read_blob(&e.blob).or_fail()?),
2105 )
2106 } else {
2107 None
2108 };
2109 out.push(TreeFile {
2110 path,
2111 mode: e.mode,
2112 blob: e.blob,
2113 size,
2114 content_b64,
2115 });
2116 }
2117 Ok(json!(out))
2118 }
2119
2120 fn blobs(&self, auth: &Auth, q: BlobsQuery) -> Hr {
2125 let name = auth.repo()?;
2126 let mut inner = self.lock();
2127 let r = Self::repo(&mut inner, &name)?;
2128 let store = r.store.clone();
2129 let redacted = r.chain.state.redacted.clone();
2130 drop(inner);
2131 if q.paths.is_empty() {
2132 return fail("invalid", "blobs: no paths");
2133 }
2134 let files = store.commit_files(&q.commit).or_fail()?;
2135 let budget = page_budget(self.settings.line_cap);
2136 let mut total = 0u64;
2137 let mut wanted = Vec::new();
2138 for path in &q.paths {
2139 let Some(e) = files.get(path) else {
2140 return fail("not_found", format!("blobs: {path} is not in {}", q.commit));
2141 };
2142 let size = if e.mode == lonca_core::FileMode::Submodule {
2143 0
2144 } else {
2145 store.blob_size(&e.blob).or_fail()?
2146 };
2147 let encoded = encoded_len(size);
2148 if encoded > budget {
2149 return fail(
2150 "too_large",
2151 format!(
2152 "blobs: {path} is {size} bytes, more than one page holds ({budget} encoded)"
2153 ),
2154 );
2155 }
2156 total += encoded;
2157 if total > budget {
2158 return fail(
2159 "too_large",
2160 format!(
2161 "blobs: this page would exceed {budget} encoded bytes; ask for fewer files"
2162 ),
2163 );
2164 }
2165 wanted.push((path.clone(), e.clone(), size));
2166 }
2167 let mut out = Vec::with_capacity(wanted.len());
2168 for (path, e, size) in wanted {
2169 let content_b64 =
2170 if e.mode != lonca_core::FileMode::Submodule && !redacted.contains(&e.blob) {
2171 Some(
2172 base64::engine::general_purpose::STANDARD
2173 .encode(store.read_blob(&e.blob).or_fail()?),
2174 )
2175 } else {
2176 None
2177 };
2178 out.push(TreeFile {
2179 path,
2180 mode: e.mode,
2181 blob: e.blob,
2182 size,
2183 content_b64,
2184 });
2185 }
2186 Ok(json!(out))
2187 }
2188
2189 fn change_get(&self, auth: &Auth, c: ChangeRef) -> Hr {
2193 let name = auth.repo()?;
2194 let mut inner = self.lock();
2195 let r = Self::repo(&mut inner, &name)?;
2196 let Some(rec) = r.chain.state.changes.get(&c.change).cloned() else {
2197 return fail("not_found", "unknown change");
2198 };
2199 let store = r.store.clone();
2200 let redacted = r.chain.state.redacted.clone();
2201 let runs = r
2202 .policy()
2203 .ok()
2204 .map(|p| runs_for(&p.config.verification, &rec));
2205 drop(inner);
2206 let mut out = json!(rec);
2207 out["runs"] = json!(runs);
2208 let ops = OpLog::new(&store).read_all().or_fail()?;
2209 let names_it = |o: &&SignedOp| match &o.entry.op {
2210 Op::ChangeIntegrate { change, .. } => *change == c.change,
2211 Op::ChangeIntegrateBatch { merged, .. } => merged.iter().any(|m| m.0 == c.change),
2212 _ => false,
2213 };
2214 let last = ops.iter().rev().find(names_it);
2215 let evidence: Vec<Value> = match last {
2216 Some(o) => o
2217 .entry
2218 .evidence
2219 .iter()
2220 .map(|a| {
2221 let (log, log_truncated) = match &a.log {
2222 Some(id) if !redacted.contains(id) => match store.read_blob(id) {
2223 Ok(bytes) => {
2224 let start = bytes.len().saturating_sub(EVIDENCE_LOG_CAP_BYTES);
2225 (
2226 Value::String(
2227 String::from_utf8_lossy(&bytes[start..]).into_owned(),
2228 ),
2229 start > 0,
2230 )
2231 }
2232 Err(_) => (Value::Null, false),
2233 },
2234 _ => (Value::Null, false),
2235 };
2236 let d = &a.details;
2237 json!({
2238 "check": a.check, "result": a.result,
2239 "exit": d.get("exit").and_then(|x| x.parse::<i64>().ok()),
2240 "duration_ms": d.get("duration_ms").and_then(|x| x.parse::<i64>().ok()),
2241 "attempts": d.get("attempts").and_then(|x| x.parse::<u32>().ok()),
2242 "sandbox": d.get("sandbox"), "error": d.get("error"),
2243 "env": a.env, "environment": environment(d),
2244 "details": d, "log": log, "log_truncated": log_truncated,
2245 })
2246 })
2247 .collect(),
2248 None => vec![],
2249 };
2250 out["evidence"] = json!(evidence);
2251 out["integrate_seq"] = json!(last.map(|o| o.entry.seq));
2252 out["integrate_decision"] = json!(last.map(|o| &o.entry.decision));
2253 Ok(out)
2254 }
2255
2256 fn trunk_bundle(&self, auth: &Auth, p: TrunkBundle) -> Hr {
2263 let name = auth.repo()?;
2264 let mut inner = self.lock();
2265 let r = Self::repo(&mut inner, &name)?;
2266 let store = r.store.clone();
2267 let trunk = r.chain.state.trunk.clone().expect("initialized");
2268 let redacted = r.chain.state.redacted.clone();
2269 let have: Vec<String> = p
2270 .have
2271 .iter()
2272 .filter(|h| store.read_commit(h).is_ok())
2273 .map(|h| format!("^{h}"))
2274 .collect();
2275 let mut args: Vec<&str> = vec!["rev-list", "--objects"];
2276 args.extend(have.iter().map(String::as_str));
2277 args.push("refs/heads/main");
2278 let objects = self.git(store.path(), &args)?;
2279 let mut hidden: Vec<&str> = Vec::new();
2280 for line in objects.lines() {
2281 let (id, path) = line.split_once(' ').unwrap_or((line, ""));
2282 if GitId::parse(id).is_ok_and(|g| redacted.contains(&g)) {
2283 hidden.push(path);
2284 }
2285 }
2286 if !hidden.is_empty() {
2287 return fail(
2288 "redacted",
2289 format!(
2290 "trunk history carries redacted content ({}); git needs whole objects, work from a workspace",
2291 hidden.join(", ")
2292 ),
2293 );
2294 }
2295 if objects.trim().is_empty() {
2297 return Ok(json!({
2298 "trunk": trunk, "bundle_b64": "", "bytes": 0, "thin": true, "up_to_date": true,
2299 }));
2300 }
2301 let dir = self.home.scratch();
2302 std::fs::create_dir_all(&dir).map_err(|e| from_err(e.into()))?;
2303 let tmp = dir.join(format!("{name}-{}.bundle", random_id("b")));
2304 let tmp_s = tmp.to_string_lossy().into_owned();
2305 let mut args: Vec<&str> = vec!["bundle", "create", "-q", &tmp_s];
2306 args.extend(have.iter().map(String::as_str));
2307 args.push("refs/heads/main");
2308 let made = self.git(store.path(), &args);
2309 let bytes = made.and_then(|_| std::fs::read(&tmp).map_err(|e| from_err(e.into())));
2310 let _ = std::fs::remove_file(&tmp);
2311 let bytes = bytes?;
2312 let budget = page_budget(self.settings.line_cap);
2313 if encoded_len(bytes.len() as u64) > budget {
2314 return fail(
2315 "too_large",
2316 format!(
2317 "the bundle is {} bytes, more than one page holds; fetch from a closer point (have) or restore a backup",
2318 bytes.len()
2319 ),
2320 );
2321 }
2322 Ok(json!({
2323 "trunk": trunk,
2324 "bundle_b64": base64::engine::general_purpose::STANDARD.encode(&bytes),
2325 "bytes": bytes.len(),
2326 "thin": !have.is_empty(),
2327 "up_to_date": false,
2328 }))
2329 }
2330
2331 fn trunk_distance(&self, auth: &Auth, p: TrunkDistance) -> Hr {
2334 let name = auth.repo()?;
2335 let mut inner = self.lock();
2336 let r = Self::repo(&mut inner, &name)?;
2337 let s = &r.chain.state;
2338 Ok(json!({"trunk": s.trunk, "base": p.base, "behind": s.behind(&p.base)}))
2339 }
2340
2341 fn change_resolve(&self, auth: &Auth, p: ChangeResolve) -> Hr {
2347 if !lonca_core::changeid::is_prefix(&p.prefix) {
2348 return fail(
2349 "invalid",
2350 format!(
2351 "a change id prefix is {} to 32 letters k-z",
2352 lonca_core::changeid::MIN_PREFIX
2353 ),
2354 );
2355 }
2356 let name = auth.repo()?;
2357 let mut inner = self.lock();
2358 let r = Self::repo(&mut inner, &name)?;
2359 resolve_prefix(r.chain.state.changes.keys(), &p.prefix)
2360 }
2361
2362 fn change_diff(&self, auth: &Auth, p: ChangeDiff) -> Hr {
2363 if let Some(f) = &p.format
2364 && f != "unified"
2365 {
2366 return fail("invalid", "format: only unified");
2367 }
2368 let name = auth.repo()?;
2369 let mut inner = self.lock();
2370 let r = Self::repo(&mut inner, &name)?;
2371 let Some(rec) = r.chain.state.changes.get(&p.change).cloned() else {
2372 return fail("not_found", "unknown change");
2373 };
2374 let store = r.store.clone();
2375 let redacted = r.chain.state.redacted.clone();
2376 drop(inner);
2377 let before = store.commit_files(&rec.base).or_fail()?;
2378 let after = store.commit_files(&rec.commit).or_fail()?;
2379 let mut paths: Vec<String> = Vec::new();
2380 let mut redacted_paths: Vec<String> = Vec::new();
2381 let mut status: BTreeMap<String, &str> = BTreeMap::new();
2382 let mut seen = std::collections::BTreeSet::new();
2383 for path in before.keys().chain(after.keys()) {
2384 if !seen.insert(path.clone()) {
2385 continue;
2386 }
2387 let (b, a) = (before.get(path), after.get(path));
2388 let st = match (b, a) {
2389 (Some(x), Some(y)) if x == y => continue,
2390 (Some(_), Some(_)) => "modified",
2391 (None, Some(_)) => "added",
2392 (Some(_), None) => "deleted",
2393 (None, None) => continue,
2394 };
2395 if [b, a]
2396 .into_iter()
2397 .flatten()
2398 .any(|e| redacted.contains(&e.blob))
2399 {
2400 redacted_paths.push(path.clone());
2401 continue;
2402 }
2403 status.insert(path.clone(), st);
2404 paths.push(path.clone());
2405 }
2406 let by_file = |items: &[String]| -> BTreeMap<String, Vec<String>> {
2408 let mut m: BTreeMap<String, Vec<String>> = BTreeMap::new();
2409 for s in items {
2410 if let Some((p, _)) = s.split_once('#') {
2411 m.entry(p.to_string()).or_default().push(s.clone());
2412 }
2413 }
2414 m
2415 };
2416 let edits = by_file(&rec.facts.edits);
2417 let adds = by_file(&rec.facts.adds);
2418 let deletes = by_file(&rec.facts.deletes);
2419 let sigs = by_file(&rec.facts.signature_changes);
2420 type Counts = BTreeMap<String, (Option<u64>, Option<u64>)>;
2421 let (mut diff, mut counts): (String, Counts) = (String::new(), BTreeMap::new());
2422 if !paths.is_empty() {
2423 let (base, commit) = (rec.base.to_string(), rec.commit.to_string());
2424 let mut args: Vec<&str> = vec![
2425 "diff",
2426 "--no-color",
2427 "--no-ext-diff",
2428 "--numstat",
2429 &base,
2430 &commit,
2431 "--",
2432 ];
2433 args.extend(paths.iter().map(String::as_str));
2434 for line in self.git(store.path(), &args)?.lines() {
2435 let mut it = line.split('\t');
2436 let (a, d, path) = (it.next(), it.next(), it.next());
2437 if let (Some(a), Some(d), Some(path)) = (a, d, path) {
2438 counts.insert(path.to_string(), (a.parse().ok(), d.parse().ok()));
2439 }
2440 }
2441 args[3] = "--patch";
2442 diff = self.git(store.path(), &args)?;
2443 }
2444 let (diff, truncated) = truncate_at_line(diff, DIFF_CAP_BYTES);
2445 let files: Vec<Value> = paths
2446 .iter()
2447 .map(|p| {
2448 let (added, removed) = counts.get(p).copied().unwrap_or((None, None));
2449 json!({
2450 "path": p, "status": status[p], "added": added, "removed": removed,
2451 "symbols": {
2452 "edits": edits.get(p).cloned().unwrap_or_default(),
2453 "adds": adds.get(p).cloned().unwrap_or_default(),
2454 "deletes": deletes.get(p).cloned().unwrap_or_default(),
2455 "signature_changes": sigs.get(p).cloned().unwrap_or_default(),
2456 },
2457 })
2458 })
2459 .collect();
2460 Ok(json!({
2461 "change": rec.change, "base": rec.base, "commit": rec.commit, "format": "unified",
2462 "files": files, "redacted": redacted_paths, "diff": diff, "truncated": truncated,
2463 }))
2464 }
2465
2466 fn whoami(&self, auth: &Auth) -> Hr {
2467 let inner = self.lock();
2468 match &auth.caller {
2469 Caller::Agent { repo, actor } => {
2470 let r = inner.repos.get(repo).expect("checked");
2471 let s = r
2472 .chain
2473 .state
2474 .sessions
2475 .get(actor.session().unwrap_or_default());
2476 Ok(json!({"repo": repo, "actor": actor.label(), "session": s}))
2477 }
2478 Caller::Human { human, key } => Ok(json!({
2479 "actor": Actor::Human { name: human.clone() }.label(),
2480 "human": human,
2481 "key": key,
2482 "repos": Self::enrolled_repos(&inner, human, key),
2483 })),
2484 Caller::Anonymous => fail("unauthenticated", "missing session token"),
2485 }
2486 }
2487
2488 fn prepare_change(&self, auth: &Auth, p: PrepareChange) -> Hr {
2491 let name = auth.repo()?;
2492 let mut inner = self.lock();
2493 let r = Self::repo(&mut inner, &name)?;
2494 let now = self.now();
2495 let change = p.change.unwrap_or_else(ChangeId::random);
2496 let max = r.policy().or_fail()?.config.changes.max_file_bytes;
2497 let commit = r
2498 .build_change_commit(
2499 &p.base,
2500 &p.files,
2501 change,
2502 &p.author,
2503 &format!("{}@humans.lonca.invalid", p.author),
2504 (HUMAN_HEADER, &p.author),
2505 &p.intent,
2506 now,
2507 max,
2508 )
2509 .or_fail()?;
2510 let keyed = &r.policy().or_fail()?.keyed;
2511 let a = facts_keyed(&r.store, ENGINE_CURRENT, &p.base, &commit, keyed).or_fail()?;
2512 let f = &a.facts;
2513 let claim = Claim {
2514 intent: p.intent,
2515 edits: f.edits.clone(),
2516 adds: f.adds.clone(),
2517 deletes: f.deletes.clone(),
2518 deleted_files: f.deleted_files.clone(),
2519 signature_changes: f.signature_changes.clone(),
2520 tests: f.tests.clone(),
2521 revert_of: p.revert_of,
2522 };
2523 let op = Op::ChangePropose {
2524 change,
2525 commit,
2526 base: p.base,
2527 claim,
2528 };
2529 Ok(json!({"op": op, "facts": a.facts, "syntax_errors": a.syntax_errors}))
2530 }
2531
2532 fn prepare_policy(&self, auth: &Auth, p: PreparePolicy) -> Hr {
2533 let name = auth.repo()?;
2534 let mut inner = self.lock();
2535 let r = Self::repo(&mut inner, &name)?;
2536 let parent = r.chain.state.policy.clone();
2537 let commit = r
2538 .write_policy(&p.config, &p.policy, parent.as_ref(), self.now())
2539 .or_fail()?;
2540 Ok(json!({"op": Op::PolicyUpdate { policy: commit }}))
2541 }
2542
2543 fn prepare_import(&self, auth: &Auth, p: PrepareRemote) -> Hr {
2544 let name = auth.repo()?;
2545 let src = self.source(&p.remote)?;
2546 let branch = p.branch.unwrap_or_else(|| "main".into());
2547 self.check_branch(&branch)?;
2548 let store = {
2549 let mut inner = self.lock();
2550 Self::repo(&mut inner, &name)?.store.clone()
2551 };
2552 self.git(
2553 store.path(),
2554 &[
2555 "fetch",
2556 "--no-tags",
2557 "--",
2558 &src.arg(),
2559 &format!("+{branch}:refs/lonca/incoming"),
2560 ],
2561 )?;
2562 let mut inner = self.lock();
2563 let r = Self::repo(&mut inner, &name)?;
2564 let commit = r
2565 .store
2566 .ref_get("refs/lonca/incoming")
2567 .or_fail()?
2568 .ok_or(("invalid".into(), "nothing fetched".into()))?;
2569 let trunk = r.chain.state.trunk.clone().expect("initialized");
2570 if commit == trunk {
2571 return fail("up_to_date", "trunk already contains the remote branch");
2572 }
2573 if !is_ancestor(&r.store, &trunk, &commit).or_fail()? {
2574 return fail(
2575 "not_fast_forward",
2576 "the remote branch does not contain trunk; diverged history is not imported",
2577 );
2578 }
2579 Ok(json!({"op": Op::ImportExternal { remote: p.remote, commit, trunk_before: trunk }}))
2580 }
2581
2582 fn prepare_push(&self, auth: &Auth, p: PrepareRemote) -> Hr {
2583 let name = auth.repo()?;
2584 self.check_remote(&p.remote)?;
2585 let mut inner = self.lock();
2586 let r = Self::repo(&mut inner, &name)?;
2587 let s = &r.chain.state;
2588 Ok(
2589 json!({"op": Op::BridgePush { remote: p.remote, oplog_seq: s.next_seq - 1, oplog_hash: s.head_hash.clone() }}),
2590 )
2591 }
2592
2593 fn prepare_isolate(&self, auth: &Auth, p: PrepareIsolate) -> Hr {
2597 let name = auth.repo()?;
2598 let mut inner = self.lock();
2599 let r = Self::repo(&mut inner, &name)?;
2600 let Some(s) = r.chain.state.sessions.get(&p.session) else {
2601 return fail("not_found", "no such session");
2602 };
2603 let base_known = r.chain.state.trunk_history.contains(&p.base);
2604 let scratch = s.scope.scratch.clone();
2605 let harness = s.harness.clone();
2606 let policy = r.policy().or_fail()?;
2607 let profile = harness.as_deref().and_then(|h| policy.config.harness(h));
2608 Ok(json!({
2609 "scratch": scratch,
2610 "proxy": self.settings.proxy,
2611 "home": self.home.root,
2612 "base_known": base_known,
2613 "harness": harness,
2614 "setup": profile.map(|p| &p.setup),
2615 }))
2616 }
2617
2618 fn backup(&self, auth: &Auth) -> Hr {
2625 let name = auth.repo()?;
2626 let dir = self.home.backups();
2627 std::fs::create_dir_all(&dir).map_err(|e| from_err(e.into()))?;
2628 set_mode(&dir, 0o750)?;
2629 let mut inner = self.lock();
2630 let r = Self::repo(&mut inner, &name)?;
2631 let seq = r.chain.state.next_seq - 1;
2632 let file = dir.join(format!("{name}-{seq:06}.bundle"));
2633 let tmp = dir.join(format!("{name}-{seq:06}.bundle.tmp"));
2634 self.git(
2635 r.store.path(),
2636 &["bundle", "create", "-q", &tmp.to_string_lossy(), "--all"],
2637 )?;
2638 set_mode(&tmp, 0o640)?;
2639 std::fs::rename(&tmp, &file).map_err(|e| from_err(e.into()))?;
2640 Ok(json!({
2641 "repo": name,
2642 "path": file,
2643 "seq": seq,
2644 "head_hash": r.chain.state.head_hash,
2645 "daemon_key": r.chain.state.daemon_key,
2646 }))
2647 }
2648
2649 fn restore(&self, p: Restore) -> Hr {
2657 if !lonca_policy::decide::valid_token(&p.repo, 64) {
2658 return fail(
2659 "invalid",
2660 "repository names: letters, digits, '.', '_', '-'",
2661 );
2662 }
2663 let src = self.source(&p.from)?;
2664 let mut guard = self.lock();
2665 let inner = &mut *guard;
2666 let path = self.home.repo_path(&p.repo);
2667 if inner.repos.contains_key(&p.repo) || inner.foreign.contains_key(&p.repo) || path.exists()
2668 {
2669 return fail(
2670 "conflict",
2671 format!(
2672 "repository {} exists; a restore never replaces a repository. \
2673 Move its store away (as the daemon's user) or restore under another name",
2674 p.repo
2675 ),
2676 );
2677 }
2678 let fetched = Store::init_bare(&path).or_fail().and_then(|store| {
2679 self.git(
2680 store.path(),
2681 &[
2682 "fetch",
2683 "-q",
2684 "--no-tags",
2685 "--",
2686 &src.arg(),
2687 "+refs/heads/*:refs/heads/*",
2688 "+refs/lonca/*:refs/lonca/*",
2689 ],
2690 )?;
2691 let r = RepoHandle::open(&p.repo, &path, &inner.keys.all()).or_fail()?;
2692 if !r.initialized() {
2693 return fail("invalid", "the source has no operation log");
2694 }
2695 if r.chain.state.repo != p.repo {
2697 return fail(
2698 "invalid",
2699 format!(
2700 "the source is a backup of repository {}; restore it under that name",
2701 r.chain.state.repo
2702 ),
2703 );
2704 }
2705 Ok(r)
2706 });
2707 let mut r = match fetched {
2708 Ok(r) => r,
2709 Err(e) => {
2710 let _ = std::fs::remove_dir_all(&path);
2711 return Err(e);
2712 }
2713 };
2714 let foreign = r.foreign();
2715 let out = json!({
2716 "repo": p.repo,
2717 "seq": r.chain.state.next_seq - 1,
2718 "head_hash": r.chain.state.head_hash,
2719 "daemon_key": r.chain.state.daemon_key,
2720 "foreign": foreign,
2721 "path": path,
2722 });
2723 if foreign {
2724 inner.foreign.insert(p.repo, r);
2725 } else {
2726 Self::reconcile(&mut r, self.now())?;
2729 inner.repos.insert(p.repo, r);
2730 }
2731 Ok(out)
2732 }
2733
2734 fn prepare_rotate(&self) -> Hr {
2740 let mut guard = self.lock();
2741 let inner = &mut *guard;
2742 if inner.keys.next.is_none() {
2743 inner.keys.next = Some(self.home.load_or_create_next_key().or_fail()?);
2744 }
2745 let next = inner.keys.for_new().clone();
2746 let current = inner.keys.current.public_hex();
2747 let repos: Vec<&String> = inner
2748 .repos
2749 .iter()
2750 .filter(|(_, r)| r.initialized() && r.chain.public_key() == current)
2751 .map(|(n, _)| n)
2752 .collect();
2753 Ok(json!({"key": next.public_hex(), "repos": repos}))
2754 }
2755
2756 fn daemon_recover(&self, name: &str, p: HumanSubmit) -> Hr {
2760 let mut guard = self.lock();
2761 let inner = &mut *guard;
2762 let now = self.now();
2763 let Some(r) = inner.foreign.get_mut(name) else {
2764 if inner.repos.get(name).is_some_and(|r| r.initialized()) {
2765 return fail(
2766 "invalid",
2767 format!(
2768 "this daemon holds the key of repository {name}; \
2769 use `lonca daemon-key rotate`"
2770 ),
2771 );
2772 }
2773 return fail("not_found", format!("no repository {name}"));
2774 };
2775 let key = inner.keys.for_new().clone();
2776 if p.op.daemon_key_switch() != Some(key.public_hex().as_str()) {
2777 return fail(
2778 "invalid",
2779 format!(
2780 "a recovery here must name this daemon's key {}",
2781 key.public_hex()
2782 ),
2783 );
2784 }
2785 r.chain.set_key(key);
2786 let mut sub = Submission::new(
2787 Actor::Human {
2788 name: p.sig.human.clone(),
2789 },
2790 p.op,
2791 );
2792 sub.human = Some(p.sig);
2793 let dry = r.dry_run(sub.clone(), now).or_fail()?;
2794 if !dry.entry.decision.is_allow() {
2795 return fail("denied", dry.entry.decision.reasons.join(", "));
2796 }
2797 let op = r.submit(sub, now).or_fail()?;
2798 let mut r = inner.foreign.remove(name).expect("present");
2799 Self::reconcile(&mut r, now)?;
2800 inner.repos.insert(name.to_string(), r);
2801 Self::revoke_ended_tokens(inner, name)?;
2802 Ok(logged(&op))
2803 }
2804
2805 fn human_submit(&self, auth: &Auth, p: HumanSubmit) -> Hr {
2806 let name = auth.repo()?;
2807 auth.same_human(&p.sig.human)?;
2808 if matches!(p.op, Op::DaemonRecover { .. }) {
2809 return self.daemon_recover(&name, p);
2810 }
2811 let mut guard = self.lock();
2812 let inner = &mut *guard;
2813 let now = self.now();
2814 let next = inner.keys.next.clone();
2815 let r = Self::repo(inner, &name)?;
2816 Self::reconcile(r, now)?;
2817 let mut sub = Submission::new(
2818 Actor::Human {
2819 name: p.sig.human.clone(),
2820 },
2821 p.op.clone(),
2822 );
2823 sub.human = Some(p.sig);
2824 let workspace = p.workspace.filter(|w| Path::new(w).is_absolute());
2825 let mut push: Option<(PathBuf, String, Vec<String>)> = None;
2827 match &p.op {
2828 Op::RepoInit { .. } => return fail("invalid", "use repo.init"),
2829 Op::ChangePropose {
2830 change,
2831 commit,
2832 base,
2833 ..
2834 } => {
2835 r.check_proposal(commit, base, *change).or_fail()?;
2836 let keyed = &r.policy().or_fail()?.keyed;
2837 sub.facts = Some(
2838 facts_keyed(&r.store, ENGINE_CURRENT, base, commit, keyed)
2839 .or_fail()?
2840 .facts,
2841 );
2842 sub.symbols_engine = Some(ENGINE_CURRENT);
2843 }
2844 Op::PolicyUpdate { policy } => {
2845 RepoHandle::load_policy(&r.store, policy).or_fail()?;
2846 }
2847 Op::DaemonRotate { key }
2850 if next.as_ref().map(KeyPair::public_hex).as_ref() != Some(key) =>
2851 {
2852 return fail("invalid", "rotate to the key from daemon.prepare_rotate");
2853 }
2854 Op::ImportExternal {
2855 commit,
2856 trunk_before,
2857 ..
2858 } if !is_ancestor(&r.store, trunk_before, commit).or_fail()? => {
2859 return fail("not_fast_forward", "import must contain trunk");
2860 }
2861 _ => {}
2862 }
2863 let old_policy = r.chain.state.policy.clone();
2864 let seq = r.chain.state.next_seq;
2865 if let Op::ChangePropose { change, commit, .. } = &p.op {
2866 r.store
2867 .ref_cas(&format!("refs/lonca/changes/{change}/{seq}"), commit, None)
2868 .or_fail()?;
2869 }
2870 let op = r.submit(sub, now).or_fail()?;
2871 let mut out = logged(&op);
2872 if !op.entry.decision.is_allow() {
2873 if op.entry.decision.outcome == lonca_core::ops::Outcome::Escalate {
2874 self.notify_queue();
2875 }
2876 return Ok(out);
2877 }
2878 match &op.entry.op {
2879 Op::SessionStart {
2880 session, continues, ..
2881 } => {
2882 let token = Self::issue_token(inner, &name, session)?;
2883 out["token"] = json!(token);
2884 if let Some(c) = continues {
2885 Self::revoke_tokens(inner, &name, &c.session)?;
2886 }
2887 }
2888 Op::PolicyUpdate { policy } => {
2889 let r = Self::repo(inner, &name)?;
2890 r.store
2891 .ref_cas(POLICY_REF, policy, old_policy.as_ref())
2892 .or_fail()?;
2893 }
2894 Op::ImportExternal {
2895 commit,
2896 trunk_before,
2897 ..
2898 } => {
2899 let r = Self::repo(inner, &name)?;
2900 r.store
2901 .ref_cas(TRUNK_REF, commit, Some(trunk_before))
2902 .or_fail()?;
2903 }
2904 Op::BridgePush { remote, .. } => {
2907 let r = Self::repo(inner, &name)?;
2908 let changes = r.chain.state.proposed.iter();
2909 let changes = changes.map(|(seq, (c, _))| refs::change_ref_name(c, *seq));
2910 push = Some((
2911 r.store.path().to_path_buf(),
2912 remote.clone(),
2913 changes.collect(),
2914 ));
2915 }
2916 Op::SessionIsolate { session, .. } => {
2919 let old = std::slice::from_ref(session);
2920 inner.tokens.revoke_sessions(&name, old).or_fail()?;
2921 out["token"] = json!(Self::issue_token(inner, &name, session)?);
2922 if let Some(ws) = workspace {
2923 let key = (name.clone(), session.clone());
2924 inner.workspaces.insert(key, PathBuf::from(ws));
2925 }
2926 }
2927 Op::CapRevoke { session, .. } | Op::SessionEnd { session, .. } => {
2928 Self::revoke_tokens(inner, &name, session)?;
2929 }
2930 Op::HumanRotate { .. } | Op::HumanRevoke { .. } | Op::HumanRecover { .. } => {
2933 Self::revoke_ended_tokens(inner, &name)?;
2934 }
2935 Op::Approve { .. } => {
2936 Self::revoke_ended_tokens(inner, &name)?;
2937 self.notify_queue();
2938 }
2939 Op::DaemonRotate { .. } => {
2940 let r = Self::repo(inner, &name)?;
2941 r.chain.set_key(next.expect("checked above"));
2942 self.promote_if_done(inner).or_fail()?;
2943 }
2944 Op::ChangePropose { .. } => self.notify_queue(),
2945 _ => {}
2946 }
2947 drop(guard);
2948 if let Some((dir, remote, changes)) = push {
2949 let mut pushed = String::new();
2950 let mine = ["refs/heads/main", "refs/lonca/oplog", "refs/lonca/policy"];
2951 let all: Vec<String> = mine.iter().map(|r| r.to_string()).chain(changes).collect();
2952 for part in all.chunks(PUSH_REFS_PER_CALL) {
2954 let specs: Vec<String> = part.iter().map(|r| format!("{r}:{r}")).collect();
2955 let mut args = vec!["push", "--porcelain", "--", remote.as_str()];
2956 args.extend(specs.iter().map(String::as_str));
2957 pushed.push_str(&self.git(&dir, &args)?);
2958 }
2959 out["push"] = json!(pushed);
2960 }
2961 Ok(out)
2962 }
2963
2964 fn issue_token(
2966 inner: &mut Inner,
2967 repo: &str,
2968 session: &str,
2969 ) -> std::result::Result<String, (String, String)> {
2970 let r = Self::repo(inner, repo)?;
2971 let Some(s) = r.chain.state.sessions.get(session) else {
2972 return fail("internal", "a token for a session the log does not have");
2973 };
2974 let expires_ms = s.expires_ms;
2975 inner.tokens.issue(repo, session, expires_ms).or_fail()
2976 }
2977
2978 fn sweep_tokens(inner: &mut Inner, now: i64) -> std::result::Result<(), (String, String)> {
2981 let Inner { tokens, repos, .. } = inner;
2982 let live = |repo: &str, session: &str| {
2983 repos
2984 .get(repo)
2985 .is_some_and(|r| r.chain.state.session_active(session, now))
2986 };
2987 tokens.sweep_expired(now, live).or_fail()
2988 }
2989
2990 fn revoke_tokens(
2991 inner: &mut Inner,
2992 repo: &str,
2993 root: &str,
2994 ) -> std::result::Result<(), (String, String)> {
2995 let r = Self::repo(inner, repo)?;
2996 let ended: Vec<String> = r
2997 .chain
2998 .state
2999 .sessions
3000 .values()
3001 .filter(|s| s.ended.is_some() && r.chain.state.descends_from(&s.id, root))
3002 .map(|s| s.id.clone())
3003 .collect();
3004 inner.tokens.revoke_sessions(repo, &ended).or_fail()
3005 }
3006
3007 fn revoke_ended_tokens(
3009 inner: &mut Inner,
3010 repo: &str,
3011 ) -> std::result::Result<(), (String, String)> {
3012 let r = Self::repo(inner, repo)?;
3013 let ended: Vec<String> = r
3014 .chain
3015 .state
3016 .sessions
3017 .values()
3018 .filter(|s| s.ended.is_some())
3019 .map(|s| s.id.clone())
3020 .collect();
3021 inner.tokens.revoke_sessions(repo, &ended).or_fail()
3022 }
3023
3024 fn agent_op(
3028 &self,
3029 req: &Request,
3030 make: impl FnOnce(&Actor) -> std::result::Result<Op, (String, String)>,
3031 ) -> Hr {
3032 let mut guard = self.lock();
3033 let now = self.now();
3034 let (repo, actor) = Self::agent(&guard, req, now)?;
3035 let op = make(&actor)?;
3036 let r = Self::repo(&mut guard, &repo)?;
3037 Self::reconcile(r, now)?;
3038 let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3039 Ok(logged(&signed))
3040 }
3041
3042 fn lease_acquire(&self, req: &Request, p: LeaseAcquire) -> Hr {
3043 let set = p.set.clone().unwrap_or_else(|| random_id("lse"));
3044 let mut guard = self.lock();
3045 let now = self.now();
3046 let (repo, actor) = Self::agent(&guard, req, now)?;
3047 let session = actor.session().expect("agent").to_string();
3048 let r = Self::repo(&mut guard, &repo)?;
3049 Self::reconcile(r, now)?;
3050 let mut items = p.items;
3051 let keys = Self::next_keys(r, &mut items, now)?;
3052 let op = Op::LeaseAcquire {
3053 set: set.clone(),
3054 items: items.clone(),
3055 intent: p.intent,
3056 ttl_ms: p.ttl_ms,
3057 };
3058 let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3059 let mut out = logged(&signed);
3060 out["set"] = json!(set);
3061 if !keys.is_empty() {
3062 out["keys"] = json!(keys);
3063 }
3064 let details = Self::lease_details(r, &session, &items, now);
3065 out["advice"] = json!(details::advice(&details));
3066 out["details"] = json!(details);
3067 Ok(out)
3068 }
3069
3070 fn next_keys(
3076 r: &mut RepoHandle,
3077 items: &mut [LeaseItem],
3078 now: i64,
3079 ) -> std::result::Result<BTreeMap<String, Vec<String>>, (String, String)> {
3080 let wants = |i: &LeaseItem| i.names.iter().any(|n| n == keyed::NEXT);
3081 if !items.iter().any(wants) {
3082 return Ok(BTreeMap::new());
3083 }
3084 let policy = r.policy().or_fail()?;
3085 let trunk = match &r.chain.state.trunk {
3086 Some(t) => r.store.commit_files(t).or_fail()?,
3087 None => FileMap::new(),
3088 };
3089 let state = &r.chain.state;
3090 let mut out = BTreeMap::new();
3091 for item in items.iter_mut().filter(|i| wants(i)) {
3092 let Some(dir) = item
3093 .symbol
3094 .strip_suffix(keyed::DIR)
3095 .and_then(|d| d.strip_suffix('#'))
3096 else {
3097 return fail(
3098 "invalid",
3099 format!("`{}` is a key: only for dir#<dir>", keyed::NEXT),
3100 );
3101 };
3102 let held: Vec<String> = state
3103 .active_leases(now)
3104 .flat_map(|l| l.items)
3105 .filter(|i| i.symbol == item.symbol)
3106 .flat_map(|i| i.names.iter().cloned())
3107 .collect();
3108 let mut taken: Vec<String> = held;
3109 taken.extend(item.names.iter().filter(|n| *n != keyed::NEXT).cloned());
3110 let mut names = Vec::new();
3111 for n in &item.names {
3112 if n != keyed::NEXT {
3113 names.push(n.clone());
3114 continue;
3115 }
3116 let tree = trunk.keys().map(String::as_str);
3117 let reserved = taken.iter().map(String::as_str);
3118 let Some(key) = policy.keyed.next_key(dir, tree, reserved) else {
3119 return fail(
3120 "invalid",
3121 format!(
3122 "`{}`: {dir} is not a keyed directory whose keys are numbers",
3123 keyed::NEXT
3124 ),
3125 );
3126 };
3127 taken.push(key.clone());
3128 names.push(key);
3129 }
3130 let got: Vec<String> = names
3131 .iter()
3132 .filter(|n| !item.names.contains(n))
3133 .cloned()
3134 .collect();
3135 out.entry(item.symbol.clone())
3136 .or_insert_with(Vec::new)
3137 .extend(got);
3138 item.names = names;
3139 }
3140 Ok(out)
3141 }
3142
3143 fn lease_details(
3144 r: &RepoHandle,
3145 session: &str,
3146 items: &[LeaseItem],
3147 now: i64,
3148 ) -> Vec<details::Detail> {
3149 let state = &r.chain.state;
3150 let mut out = details::lease_held(state, session, items, now);
3151 out.extend(details::lease_reserved(state, session, items, now));
3152 out.extend(details::lease_wanted(state, session, items, now));
3153 out
3154 }
3155
3156 fn lease_interest(&self, req: &Request, p: LeaseInterest) -> Hr {
3157 let mut guard = self.lock();
3158 let now = self.now();
3159 let (repo, actor) = Self::agent(&guard, req, now)?;
3160 let session = actor.session().expect("agent").to_string();
3161 let r = Self::repo(&mut guard, &repo)?;
3162 Self::reconcile(r, now)?;
3163 let op = Op::LeaseInterest {
3164 items: p.items.clone(),
3165 intent: p.intent,
3166 };
3167 let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3168 let mut out = logged(&signed);
3169 out["details"] = json!(Self::lease_details(r, &session, &p.items, now));
3170 Ok(out)
3171 }
3172
3173 fn events(&self, req: &Request, p: Events) -> Hr {
3175 let mut guard = self.lock();
3176 let now = self.now();
3177 let (repo, actor) = Self::agent(&guard, req, now)?;
3178 let session = actor.session().expect("agent").to_string();
3179 let r = Self::repo(&mut guard, &repo)?;
3180 let ops = OpLog::new(&r.store).read_all().or_fail()?;
3181 let state = &r.chain.state;
3182 let events = for_session(state, &session, &ops, p.from, now);
3183 let next = state.next_seq;
3184 r.told
3185 .entry(session)
3186 .or_default()
3187 .saw(p.from, &events, next);
3188 Ok(json!({"events": events, "next": next}))
3189 }
3190
3191 fn unread(
3193 r: &mut RepoHandle,
3194 session: &str,
3195 now: i64,
3196 ) -> std::result::Result<Vec<SessionEvent>, (String, String)> {
3197 let told = r.told.entry(session.to_string()).or_default();
3198 let from = told.from();
3199 let ops = OpLog::new(&r.store).read_from(from).or_fail()?;
3200 let events = for_session(&r.chain.state, session, &ops, from, now);
3201 Ok(told.unread(events, r.chain.state.next_seq))
3202 }
3203
3204 fn inbox(&self, req: &Request) -> Hr {
3207 let mut guard = self.lock();
3208 let now = self.now();
3209 let (repo, actor) = Self::agent(&guard, req, now)?;
3210 let session = actor.session().expect("agent").to_string();
3211 let r = Self::repo(&mut guard, &repo)?;
3212 let events = Self::unread(r, &session, now)?;
3213 let text = tell(&events, now, "");
3214 Ok(json!({"events": events, "text": text}))
3215 }
3216
3217 fn watch(&self, auth: &Auth, p: Watch) -> Hr {
3223 let Caller::Human { human, .. } = &auth.caller else {
3224 return fail(
3225 "forbidden",
3226 "a human session is required (lonca human login)",
3227 );
3228 };
3229 let name = auth.repo()?;
3230 let wait = p.wait_ms.unwrap_or(0).clamp(0, LOG_WAIT_MAX_MS);
3231 let deadline = std::time::Instant::now() + Duration::from_millis(wait as u64);
3232 let known: BTreeSet<&String> = p.reserved.iter().collect();
3233 let mut from = p.from;
3234 loop {
3235 let mut inner = self.lock();
3236 let now = self.now();
3237 let r = Self::repo(&mut inner, &name)?;
3238 let state = &r.chain.state;
3239 let next = state.next_seq;
3240 let start = *from.get_or_insert(next);
3241 let sessions = watched(state, human, p.sessions.as_deref(), now)?;
3242 let ended = ended_of(state, p.sessions.as_deref(), now);
3243 let mut events = Vec::new();
3244 if next > start && !sessions.is_empty() {
3245 let ops = ops_for_events(&r.store, start).or_fail()?;
3246 for s in &sessions {
3247 let all = for_session(state, s, &ops, start, now);
3248 let logged = all
3249 .iter()
3250 .filter(|e| !matches!(e, SessionEvent::Reserved { .. }));
3251 events.extend(logged.map(|e| watched_event(s, e)));
3252 }
3253 }
3254 let mut reserved = Vec::new();
3255 for res in state.reservations_at(now) {
3256 if !sessions.contains(&res.session) {
3257 continue;
3258 }
3259 let id = reservation_id(&res.session, &res.items);
3260 if !known.contains(&id) {
3261 events.push(WatchedEvent {
3262 session: res.session.clone(),
3263 kind: "reserved".into(),
3264 seq: None,
3265 change: None,
3266 from: None,
3267 reservation: Some(id.clone()),
3268 until_ms: Some(res.until_ms),
3269 });
3270 }
3271 reserved.push(id);
3272 }
3273 let left = deadline.saturating_duration_since(std::time::Instant::now());
3274 if !events.is_empty() || !ended.is_empty() || left.is_zero() {
3275 let out = Watched {
3276 events,
3277 next,
3278 reserved,
3279 ended,
3280 };
3281 return Ok(json!(out));
3282 }
3283 from = Some(next);
3285 drop(inner);
3286 let (m, c) = &self.appended;
3287 let held = m.lock().unwrap_or_else(|e| e.into_inner());
3288 let _ = c
3289 .wait_timeout(held, left.min(Duration::from_millis(LOG_POLL_MS)))
3290 .unwrap_or_else(|e| e.into_inner());
3291 }
3292 }
3293
3294 fn news_for_hook(
3297 &self,
3298 req: &Request,
3299 tool_prefix: &str,
3300 ) -> std::result::Result<Option<String>, (String, String)> {
3301 let mut guard = self.lock();
3302 let now = self.now();
3303 let (repo, actor) = Self::agent(&guard, req, now)?;
3304 let session = actor.session().expect("agent").to_string();
3305 let r = Self::repo(&mut guard, &repo)?;
3306 let events = Self::unread(r, &session, now)?;
3307 Ok(tell(&events, now, tool_prefix))
3308 }
3309
3310 fn propose(&self, req: &Request, p: ProposeChange) -> Hr {
3311 let mut guard = self.lock();
3312 let now = self.now();
3313 let (repo, actor) = Self::agent(&guard, req, now)?;
3314 let r = Self::repo(&mut guard, &repo)?;
3315 Self::reconcile(r, now)?;
3316 let session = actor.session().expect("agent").to_string();
3317 let change = p.change.unwrap_or_else(ChangeId::random);
3318 let max = r.policy().or_fail()?.config.changes.max_file_bytes;
3319 let commit = r
3320 .build_change_commit(
3321 &p.base,
3322 &p.files,
3323 change,
3324 &actor.label(),
3325 &format!("{session}@agents.lonca.invalid"),
3326 (SESSION_HEADER, &session),
3327 &p.claim.intent,
3328 now,
3329 max,
3330 )
3331 .or_fail()?;
3332 let keyed = &r.policy().or_fail()?.keyed;
3333 let analysis = facts_keyed(&r.store, ENGINE_CURRENT, &p.base, &commit, keyed).or_fail()?;
3334 let seq = r.chain.state.next_seq;
3335 r.store
3336 .ref_cas(&format!("refs/lonca/changes/{change}/{seq}"), &commit, None)
3337 .or_fail()?;
3338 let mut sub = Submission::new(
3339 actor,
3340 Op::ChangePropose {
3341 change,
3342 commit: commit.clone(),
3343 base: p.base,
3344 claim: p.claim,
3345 },
3346 );
3347 sub.facts = Some(analysis.facts.clone());
3348 sub.symbols_engine = Some(ENGINE_CURRENT);
3349 let op = r.submit(sub, now).or_fail()?;
3350 let unobserved = op
3351 .entry
3352 .decision
3353 .reasons
3354 .iter()
3355 .any(|x| x.starts_with("unobserved_write:"));
3356 if unobserved
3357 && r.chain
3358 .state
3359 .sessions
3360 .get(&session)
3361 .is_some_and(|s| s.taint.is_none())
3362 {
3363 let t = Op::Taint {
3364 session: session.clone(),
3365 reason: "changed files that no observed tool wrote".into(),
3366 };
3367 r.submit(Submission::new(Actor::Daemon, t), now).or_fail()?;
3368 }
3369 if op.entry.decision.is_allow() {
3370 self.notify_queue();
3371 }
3372 let mut out = logged(&op);
3373 out["change"] = json!(change);
3374 out["commit"] = json!(commit);
3375 out["facts"] = json!(analysis.facts);
3376 out["syntax_errors"] = json!(analysis.syntax_errors);
3377 let policy = r.policy().or_fail()?;
3378 let unparsed = policy.config.leases.unparsed_files;
3379 let state = &r.chain.state;
3380 let mut missing = details::lease_missing(state, &session, &analysis.facts, unparsed, now);
3381 let keys = details::key_report(state, &session, &analysis.facts, &policy.keyed, now);
3382 let keys = keys.into_iter().filter(|d| d.code != details::HELD_BY_YOU);
3383 missing.extend(keys.map(|d| details::Detail {
3384 code: if d.code == details::FREE {
3385 details::LEASE_MISSING
3386 } else {
3387 d.code
3388 },
3389 ..d
3390 }));
3391 out["advice"] = json!(details::advice(&missing));
3392 out["details"] = json!(missing);
3393 Ok(out)
3394 }
3395
3396 fn facts_preview(&self, req: &Request, p: FactsPreview) -> Hr {
3397 let mut guard = self.lock();
3398 let now = self.now();
3399 let (repo, actor) = Self::agent(&guard, req, now)?;
3400 let r = Self::repo(&mut guard, &repo)?;
3401 let max = r.policy().or_fail()?.config.changes.max_file_bytes;
3402 let change = ChangeId::random();
3403 let commit = r
3404 .build_change_commit(
3405 &p.base,
3406 &p.files,
3407 change,
3408 &actor.label(),
3409 "preview@lonca.invalid",
3410 ("lonca-preview", "1"),
3411 "",
3412 now,
3413 max,
3414 )
3415 .or_fail()?;
3416 let keyed = &r.policy().or_fail()?.keyed;
3417 let a = facts_keyed(&r.store, ENGINE_CURRENT, &p.base, &commit, keyed).or_fail()?;
3418 let session = actor.session().expect("agent");
3419 let unparsed = r.policy().or_fail()?.config.leases.unparsed_files;
3420 let mut leases = details::lease_report(&r.chain.state, session, &a.facts, unparsed, now);
3421 leases.extend(details::key_report(
3422 &r.chain.state,
3423 session,
3424 &a.facts,
3425 keyed,
3426 now,
3427 ));
3428 let no_lease = match unparsed {
3429 UnparsedFiles::Optimistic => a.facts.unparsed_files.clone(),
3430 UnparsedFiles::File => Vec::new(),
3431 };
3432 Ok(json!({
3433 "facts": a.facts, "syntax_errors": a.syntax_errors,
3434 "acquire": details::to_acquire(&leases), "advice": details::advice(&leases),
3435 "leases": leases, "no_lease": no_lease,
3436 }))
3437 }
3438
3439 fn attenuate(&self, req: &Request, p: Attenuate) -> Hr {
3440 let child = random_id("ses");
3441 let mut guard = self.lock();
3442 let now = self.now();
3443 let (repo, actor) = Self::agent(&guard, req, now)?;
3444 let parent = actor.session().expect("agent").to_string();
3445 let inner = &mut *guard;
3446 let r = Self::repo(inner, &repo)?;
3447 let op = Op::CapAttenuate {
3448 parent,
3449 child: child.clone(),
3450 agent: p.agent,
3451 scope: p.scope,
3452 expires_ms: p.expires_ms,
3453 leases: p.leases,
3454 };
3455 let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3456 let mut out = logged(&signed);
3457 if signed.entry.decision.is_allow() {
3458 out["session"] = json!(child);
3459 out["token"] = json!(Self::issue_token(inner, &repo, &child)?);
3460 }
3461 Ok(out)
3462 }
3463
3464 fn rotate(&self, req: &Request) -> Hr {
3468 let mut guard = self.lock();
3469 let now = self.now();
3470 let (repo, actor) = Self::agent(&guard, req, now)?;
3471 let session = actor.session().unwrap_or_default();
3472 let r = Self::repo(&mut guard, &repo)?;
3473 let level = r.chain.state.sessions.get(session).map(|s| s.level);
3474 if level == Some(ObservationLevel::Isolated) {
3475 return fail(
3476 "forbidden",
3477 "an isolated session's token is its launcher's; `lonca run` replaces it",
3478 );
3479 }
3480 let old = req.token.as_deref().unwrap_or_default();
3481 match guard.tokens.rotate(old).or_fail()? {
3482 Some(token) => {
3483 drop(guard);
3484 use crate::security::{Event, kind};
3485 let e = Event::new(now, kind::TOKEN_ROTATED)
3486 .repo(&repo)
3487 .session(session);
3488 self.security.record(e);
3489 Ok(json!({"token": token}))
3490 }
3491 None => fail("unauthenticated", "unknown session token"),
3492 }
3493 }
3494
3495 fn agent_end(&self, req: &Request, end: SessionEnd) -> Hr {
3496 let mut guard = self.lock();
3497 let now = self.now();
3498 let (repo, actor) = Self::agent(&guard, req, now)?;
3499 let session = actor.session().expect("agent").to_string();
3500 let inner = &mut *guard;
3501 let r = Self::repo(inner, &repo)?;
3502 let left = match end.files {
3504 Some(files) => Some(report_unproposed(r, &session, files, 0, now)?),
3505 None => None,
3506 };
3507 let op = Op::SessionEnd {
3508 session: session.clone(),
3509 reason: "ended by agent".into(),
3510 };
3511 let signed = r.submit(Submission::new(actor, op), now).or_fail()?;
3512 if signed.entry.decision.is_allow() {
3513 Self::revoke_tokens(inner, &repo, &session)?;
3514 } else {
3515 Self::sweep_tokens(inner, now)?;
3517 }
3518 let mut out = logged(&signed);
3519 if let Some(left) = left {
3520 out["unproposed"] = json!(left.paths);
3521 out["unproposed_total"] = json!(left.total);
3522 }
3523 Ok(out)
3524 }
3525
3526 fn heartbeat(&self, req: &Request) -> Hr {
3531 let mut guard = self.lock();
3532 let now = self.now();
3533 let (repo, actor) = Self::agent(&guard, req, now)?;
3534 let session = actor.session().expect("agent").to_string();
3535 let r = Self::repo(&mut guard, &repo)?;
3536 if !r.chain.state.session_active(&session, now) {
3539 return fail("session_inactive", "this session has ended or expired");
3540 }
3541 let limits = r.policy().or_fail()?.config.leases.clone();
3542 let beat = Observation {
3543 phase: Phase::Pre,
3544 kind: ObsKind::Heartbeat,
3545 external: false,
3546 };
3547 let op = Op::Observe {
3548 session: session.clone(),
3549 events: vec![beat],
3550 };
3551 let signed = r
3552 .submit(Submission::new(Actor::Daemon, op), now)
3553 .or_fail()?;
3554 let leases: Vec<Value> = r
3555 .chain
3556 .state
3557 .active_leases(now)
3558 .filter(|a| a.holder == session)
3559 .map(|a| json!({"set": a.lease.set, "expires_ms": a.lease.expires_ms}))
3560 .collect();
3561 let mut out = logged(&signed);
3562 out["now_ms"] = json!(now);
3563 out["timeout_ms"] = json!(limits.heartbeat_timeout_ms);
3564 out["max_ttl_ms"] = json!(limits.max_ttl_minutes * 60_000);
3565 out["leases"] = json!(leases);
3566 Ok(out)
3567 }
3568
3569 fn hook_event(&self, req: &Request, ev: HookEvent) -> Hr {
3570 let mut guard = self.lock();
3571 let now = self.now();
3572 let (repo, actor) = Self::agent(&guard, req, now)?;
3573 let session = actor.session().expect("agent").to_string();
3574 let r = Self::repo(&mut guard, &repo)?;
3575 if !std::path::Path::new(&ev.workspace).is_absolute() {
3576 return fail("invalid", "workspace must be an absolute path");
3577 }
3578 let policy = r.policy().or_fail()?;
3579 let scratch = r
3580 .chain
3581 .state
3582 .sessions
3583 .get(&session)
3584 .map(|s| s.scope.scratch.clone())
3585 .unwrap_or_default();
3586 let (events, verdict) = hooks::translate(&ev, &policy, &scratch);
3587 let heard = Self::heard(r, &session, now);
3590 let harness = ev.harness.as_deref().unwrap_or("claude-code");
3591 if ev.hook_event_name == "PreToolUse"
3592 && verdict.decision != "deny"
3593 && let Some(profile) = policy.config.harness(harness)
3594 && let Some(tool) = ev
3595 .tool_name
3596 .as_deref()
3597 .and_then(|t| t.strip_prefix(&profile.tool_prefix()))
3598 {
3599 heard.before(tool, now);
3600 }
3601 if !events.is_empty() {
3602 r.submit(
3603 Submission::new(
3604 Actor::Daemon,
3605 Op::Observe {
3606 session: session.clone(),
3607 events,
3608 },
3609 ),
3610 now,
3611 )
3612 .or_fail()?;
3613 }
3614 if verdict.decision == "deny" {
3615 drop(guard);
3616 use crate::security::{Event, kind};
3617 let tool = ev.tool_name.as_deref().unwrap_or("?");
3618 let e = Event::new(now, kind::HOOK_DENIED)
3619 .repo(&repo)
3620 .session(&session)
3621 .method("hook.event")
3622 .detail(&format!("{tool}: {}", verdict.reason));
3623 self.security.record(e);
3624 }
3625 Ok(json!(verdict))
3626 }
3627
3628 fn hook_input(&self, req: &Request, h: HookInput) -> Hr {
3632 let profile = {
3633 let mut guard = self.lock();
3634 let (repo, _) = Self::agent(&guard, req, self.now())?;
3635 let policy = Self::repo(&mut guard, &repo)?.policy().or_fail()?;
3636 policy.config.harness(&h.harness).cloned()
3637 };
3638 let Some(profile) = profile else {
3639 return fail("invalid", format!("unknown harness {}", h.harness));
3640 };
3641 let field = |name: &str| h.input.get(name).and_then(Value::as_str);
3642 let name = field(&profile.fields.event).unwrap_or("");
3643 let phase = match profile.event_of(name) {
3644 Some(Event::SessionStart) => {
3645 return Ok(json!({
3646 "phase": "session_start",
3647 "session": field(&profile.fields.session).unwrap_or(""),
3648 "context": profile.context,
3649 "tool_prefix": profile.tool_prefix(),
3650 }));
3651 }
3652 Some(Event::Stop) => return Ok(json!({"phase": "stop"})),
3655 Some(Event::BeforeTool) => "PreToolUse",
3656 Some(Event::AfterTool) => "PostToolUse",
3657 None => {
3658 let message = format!("{name:?} is not a {} hook event", h.harness);
3659 return fail("invalid", message);
3660 }
3661 };
3662 let ev = HookEvent {
3663 hook_event_name: phase.to_string(),
3664 tool_name: field(&profile.fields.tool).map(str::to_string),
3665 tool_input: h
3666 .input
3667 .get(&profile.fields.input)
3668 .cloned()
3669 .unwrap_or(Value::Null),
3670 cwd: field(&profile.fields.cwd).map(str::to_string),
3671 workspace: h.workspace.clone(),
3672 harness: Some(h.harness.clone()),
3673 };
3674 let mut out = self.hook_event(req, ev)?;
3675 if out["decision"] == "deny" {
3676 let reason = out["reason"].as_str().unwrap_or("").to_string();
3677 out["output"] = render(&profile.deny, "reason", &reason);
3678 } else if let Some(answer) = (phase == "PostToolUse")
3679 .then_some(profile.tool_context.as_ref())
3680 .flatten()
3681 && let Some(news) = self.news_for_hook(req, &profile.tool_prefix())?
3682 {
3683 out["output"] = render(answer, "context", &news);
3684 }
3685 Ok(out)
3686 }
3687
3688 fn heard<'a>(r: &'a mut RepoHandle, session: &str, now: i64) -> &'a mut crate::hooked::Hooked {
3691 if !r.hooked.contains_key(session) {
3692 let state = &r.chain.state;
3693 r.hooked.retain(|s, _| state.session_active(s, now));
3694 }
3695 let h = r.hooked.entry(session.to_string()).or_default();
3696 h.heard(now);
3697 h
3698 }
3699
3700 fn mcp_call(&self, req: &Request, call: McpCall) -> Hr {
3706 use crate::hooked::{START_GRACE_MS, valid_tool};
3707 if !valid_tool(&call.tool) {
3708 return fail("invalid", "tool must be a tool name of lonca mcp");
3709 }
3710 let mut guard = self.lock();
3711 let now = self.now();
3712 let (repo, actor) = Self::agent(&guard, req, now)?;
3713 let session = actor.session().expect("agent").to_string();
3714 let r = Self::repo(&mut guard, &repo)?;
3715 let policy = r.policy().or_fail()?;
3716 let Some(s) = r.chain.state.sessions.get(&session) else {
3717 return fail("unauthenticated", "unknown session");
3718 };
3719 let checked = s.level != ObservationLevel::Isolated
3722 && s.harness
3723 .as_deref()
3724 .and_then(|h| policy.config.harness(h))
3725 .is_some_and(|p| p.events.before_tool_mcp);
3726 if !checked {
3727 return Ok(json!({"checked": false, "hooked": null}));
3728 }
3729 let marked = s.unhooked.is_some();
3730 let fresh =
3733 s.created_ms < self.started_ms && now.saturating_sub(self.started_ms) < START_GRACE_MS;
3734 let took = r
3735 .hooked
3736 .get_mut(&session)
3737 .is_some_and(|h| h.take(&call.tool, now));
3738 let hooked = took || fresh;
3739 if !hooked && !marked {
3740 let op = Op::Observe {
3741 session: session.clone(),
3742 events: vec![Observation {
3743 phase: Phase::Pre,
3744 kind: ObsKind::Unhooked {
3745 tool: call.tool.clone(),
3746 },
3747 external: false,
3748 }],
3749 };
3750 r.submit(Submission::new(Actor::Daemon, op), now)
3751 .or_fail()?;
3752 }
3753 Ok(json!({"checked": true, "hooked": hooked}))
3754 }
3755
3756 fn hook_stop(&self, req: &Request, h: HookStop) -> Hr {
3759 let mut guard = self.lock();
3760 let now = self.now();
3761 let (repo, actor) = Self::agent(&guard, req, now)?;
3762 let session = actor.session().expect("agent").to_string();
3763 let r = Self::repo(&mut guard, &repo)?;
3764 let policy = r.policy().or_fail()?;
3765 let limit = policy.config.sessions.stop_rounds;
3766 let Some(profile) = policy.config.harness(&h.harness).cloned() else {
3767 return fail("invalid", format!("unknown harness {}", h.harness));
3768 };
3769 let Some(answer) = &profile.stop_deny else {
3770 return fail("invalid", format!("{} has no stop event", h.harness));
3771 };
3772 Self::heard(r, &session, now);
3773 let left = report_unproposed(r, &session, h.files, limit, now)?;
3774 let mut out = json!({
3775 "unproposed": left.paths, "total": left.total, "round": left.round, "limit": limit,
3776 });
3777 if left.told {
3778 let reason = stop_message(&profile.tool_prefix(), &left.paths, left.total);
3779 out["output"] = render(answer, "reason", &reason);
3780 }
3781 Ok(out)
3782 }
3783
3784 fn harness_setup(&self, req: &Request, h: HarnessSetup) -> Hr {
3788 let mut guard = self.lock();
3789 let (repo, _) = Self::agent(&guard, req, self.now())?;
3790 let policy = Self::repo(&mut guard, &repo)?.policy().or_fail()?;
3791 Ok(json!({"setup": policy.config.harness(&h.harness).map(|p| &p.setup)}))
3792 }
3793
3794 fn hook_bind(&self, req: &Request, b: HookBind) -> Hr {
3795 let mut guard = self.lock();
3796 let now = self.now();
3797 let (repo, actor) = Self::agent(&guard, req, now)?;
3798 let session = actor.session().expect("agent").to_string();
3799 let r = Self::repo(&mut guard, &repo)?;
3800 Self::heard(r, &session, now);
3801 let s = r.chain.state.sessions.get(&session);
3804 if s.is_some_and(|s| s.level == ObservationLevel::Isolated) {
3805 let scratch = s.map(|s| s.scope.scratch.clone()).unwrap_or_default();
3806 return Ok(json!({
3807 "level": ObservationLevel::Isolated,
3808 "notes": ["the session runs in a Lonca sandbox"],
3809 "scratch": scratch,
3810 }));
3811 }
3812 let named = s.and_then(|s| s.harness.clone());
3815 let policy = r.policy().or_fail()?;
3816 let locked = named
3817 .as_deref()
3818 .filter(|n| *n == b.harness)
3819 .and_then(|n| policy.config.locked(n));
3820 let file = locked.and_then(|l| {
3821 let named = || l.file_here().map(PathBuf::from);
3822 self.settings.managed_settings.clone().or_else(named)
3823 });
3824 let ins = match (locked, file) {
3825 (Some(l), Some(file)) => hooks::inspect(
3826 l,
3827 &file,
3828 self.settings.require_root_settings,
3829 self.settings.proxy_port(),
3830 ),
3831 _ => hooks::Inspection {
3832 notes: vec![match named {
3833 None => "the session was started with no harness named".to_string(),
3834 Some(n) if n != b.harness => {
3835 format!("the session was started for {n}, not {}", b.harness)
3836 }
3837 Some(n) => format!("no locked profile for harness {n}"),
3838 }],
3839 ..Default::default()
3840 },
3841 };
3842 let checks = BindChecks {
3843 harness: b.harness,
3844 harness_session: b.harness_session,
3845 managed_hooks: ins.managed_hooks,
3846 locked_sandbox: ins.locked_sandbox,
3847 sudo_without_password: b.sudo_without_password,
3848 config_digest: ins.config_digest,
3849 hook_binary_protected: ins.hook_binary_protected,
3850 };
3851 let level = level_for(&checks);
3852 if checks.sudo_without_password {
3853 use crate::security::{Event, kind};
3854 let e = Event::new(now, kind::SUDO_WITHOUT_PASSWORD)
3855 .repo(&repo)
3856 .session(&session)
3857 .detail("the hook reports that sudo runs without a password");
3858 self.security.record(e);
3859 }
3860 let signed = r
3861 .submit(
3862 Submission::new(
3863 Actor::Daemon,
3864 Op::SessionBind {
3865 session: session.clone(),
3866 checks,
3867 level,
3868 },
3869 ),
3870 now,
3871 )
3872 .or_fail()?;
3873 let mut out = logged(&signed);
3874 let held = r.chain.state.sessions.get(&session).map(|s| s.level);
3876 out["level"] = json!(held.unwrap_or(ObservationLevel::Unobserved));
3877 out["notes"] = json!(ins.notes);
3878 out["scratch"] = json!(
3881 r.chain
3882 .state
3883 .sessions
3884 .get(&session)
3885 .map(|s| s.scope.scratch.clone())
3886 .unwrap_or_default()
3887 );
3888 if let Some(ws) = b
3889 .workspace
3890 .filter(|w| std::path::Path::new(w).is_absolute())
3891 {
3892 let session = signed.entry.op.clone();
3893 if let Op::SessionBind { session, .. } = session {
3894 guard.workspaces.insert((repo, session), PathBuf::from(ws));
3895 }
3896 }
3897 Ok(out)
3898 }
3899
3900 pub fn proxy_connect(
3912 &self,
3913 cwd: Option<&std::path::Path>,
3914 host: &str,
3915 port: u16,
3916 peer: std::net::SocketAddr,
3917 ) -> ProxyVerdict {
3918 let now = self.now();
3919 let mut guard = self.lock();
3920 let inner = &mut *guard;
3921 let mut matched: Vec<(String, String)> = match cwd {
3922 Some(c) => inner
3923 .workspaces
3924 .iter()
3925 .filter(|(_, root)| c.starts_with(root))
3926 .map(|(k, _)| k.clone())
3927 .collect(),
3928 None => vec![],
3929 };
3930 matched.retain(|(repo, session)| {
3933 let r = inner.repos.get(repo);
3934 r.is_some_and(|r| r.chain.state.session_active(session, now))
3935 });
3936 let attributed = !matched.is_empty();
3937 if !attributed {
3938 for (name, r) in inner.repos.iter().filter(|(_, r)| r.initialized()) {
3939 let s = &r.chain.state;
3940 for x in s.sessions.values() {
3941 if x.level >= ObservationLevel::HooksSandbox && s.session_active(&x.id, now) {
3942 matched.push((name.clone(), x.id.clone()));
3943 }
3944 }
3945 }
3946 }
3947 if matched.is_empty() {
3948 return ProxyVerdict {
3949 allow: false,
3950 allow_private: false,
3951 reason: "no Lonca session uses this proxy".into(),
3952 };
3953 }
3954 let mut book = self.proxy.lock().unwrap_or_else(|e| e.into_inner());
3955 if !attributed {
3956 book.unattributed += 1;
3957 }
3958 let shown = match port {
3959 80 | 443 => host.to_string(),
3960 p => format!("{host}:{p}"),
3961 };
3962 let mut allow = true;
3963 let mut allow_private = true;
3964 let mut reason = None;
3965 let mut budgets = Vec::new();
3966 for key in &matched {
3967 let Some(r) = inner.repos.get_mut(&key.0) else {
3968 continue;
3969 };
3970 let Ok(policy) = r.policy() else {
3971 allow = false;
3972 continue;
3973 };
3974 let cfg = &policy.config.observation;
3975 let (passes, external) = host_verdict(cfg, host);
3976 let refused = if !cfg.allow_ports.contains(&port) {
3977 Some(format!("port {port} is not in observation.allow_ports"))
3978 } else if !passes {
3979 Some("network access is denied by repository policy".to_string())
3980 } else if book.open_of(key) >= cfg.max_connections as usize {
3981 Some(format!(
3982 "this session has {} connections open through the proxy (observation.max_connections)",
3983 cfg.max_connections
3984 ))
3985 } else if book.carried.get(key).copied().unwrap_or(0) >= cfg.max_bytes {
3986 Some("this session has moved observation.max_bytes through the proxy".to_string())
3987 } else {
3988 None
3989 };
3990 if let Some(why) = refused {
3991 allow = false;
3992 reason.get_or_insert(why);
3993 }
3994 allow_private &= cfg.allow_private_network;
3995 budgets.push((key.clone(), cfg.max_bytes));
3996 let ev = Observation {
3997 phase: Phase::Pre,
3998 kind: ObsKind::Fetch {
3999 host: shown.clone(),
4000 },
4001 external,
4002 };
4003 let op = Op::Observe {
4004 session: key.1.clone(),
4005 events: vec![ev],
4006 };
4007 if r.submit(Submission::new(Actor::Daemon, op), now).is_err() {
4008 allow = false;
4009 reason.get_or_insert("the attempt could not be recorded".to_string());
4010 }
4011 }
4012 if allow {
4013 book.open.insert(peer, budgets);
4014 }
4015 drop(book);
4016 drop(guard);
4017 self.notify_appended();
4018 ProxyVerdict {
4019 allow,
4020 allow_private,
4021 reason: reason.unwrap_or_default(),
4022 }
4023 }
4024
4025 pub fn proxy_finished(&self, peer: std::net::SocketAddr) {
4027 let mut book = self.proxy.lock().unwrap_or_else(|e| e.into_inner());
4028 book.open.remove(&peer);
4029 }
4030
4031 pub fn proxy_carried(&self, peer: std::net::SocketAddr, bytes: u64) -> bool {
4034 let mut book = self.proxy.lock().unwrap_or_else(|e| e.into_inner());
4035 let Some(budgets) = book.open.get(&peer).cloned() else {
4036 return false;
4037 };
4038 let mut within = true;
4039 for (key, max) in budgets {
4040 let used = book.carried.entry(key).or_insert(0);
4041 *used = used.saturating_add(bytes);
4042 within &= *used <= max;
4043 }
4044 within
4045 }
4046
4047 pub fn proxy_unattributed(&self) -> u64 {
4050 self.proxy.lock().map(|b| b.unattributed).unwrap_or(0)
4051 }
4052
4053 fn reconcile(r: &mut RepoHandle, now: i64) -> std::result::Result<(), (String, String)> {
4059 let state = &r.chain.state;
4060 let (Some(trunk), Some(policy)) = (state.trunk.clone(), state.policy.clone()) else {
4061 return Ok(());
4062 };
4063 let on = r.policy().or_fail()?.config.integrity.on_unmediated.clone();
4064 for (refname, expected, history) in [(TRUNK_REF, trunk, true), (POLICY_REF, policy, false)]
4065 {
4066 let found = r.store.ref_get(refname).or_fail()?;
4067 if found.as_ref() == Some(&expected) {
4068 continue;
4069 }
4070 let recovering = history
4071 && found
4072 .as_ref()
4073 .is_some_and(|f| r.chain.state.trunk_history.contains(f));
4074 if !recovering {
4075 let op = Op::ImportUnmediated {
4076 refname: refname.into(),
4077 found: found.clone(),
4078 expected: Some(expected.clone()),
4079 };
4080 r.submit(Submission::new(Actor::Daemon, op), now)
4081 .or_fail()?;
4082 if on == "freeze" || refname == POLICY_REF {
4083 let reason = format!("{refname} changed outside Lonca");
4084 if r.chain.state.frozen.is_none() {
4085 r.submit(Submission::new(Actor::Daemon, Op::Freeze { reason }), now)
4086 .or_fail()?;
4087 }
4088 }
4089 }
4090 r.store.ref_force(refname, &expected).or_fail()?;
4091 }
4092 Self::reconcile_oplog(r, now)?;
4093 Self::reconcile_others(r, now, on == "freeze")
4094 }
4095
4096 fn reconcile_oplog(r: &mut RepoHandle, now: i64) -> std::result::Result<(), (String, String)> {
4101 let Some(head) = r.head.clone() else {
4102 return Ok(());
4103 };
4104 let found = r.store.ref_get(OPLOG_REF).or_fail()?;
4105 if found.as_ref() == Some(&head) {
4106 return Ok(());
4107 }
4108 r.store.ref_force(OPLOG_REF, &head).or_fail()?;
4109 let op = Op::ImportUnmediated {
4110 refname: OPLOG_REF.into(),
4111 found,
4112 expected: Some(head),
4113 };
4114 r.submit(Submission::new(Actor::Daemon, op), now)
4115 .or_fail()?;
4116 Self::freeze_for(r, OPLOG_REF, now)
4117 }
4118
4119 fn reconcile_others(
4126 r: &mut RepoHandle,
4127 now: i64,
4128 freeze: bool,
4129 ) -> std::result::Result<(), (String, String)> {
4130 use lonca_policy::refs::{RefCheck, check, missing_changes};
4131 let refs = r.store.refs_with_prefix("refs/").or_fail()?;
4132 let present: BTreeSet<String> = refs.iter().map(|(n, _)| n.clone()).collect();
4133 let mut undone = Vec::new();
4134 for (name, found) in refs {
4135 if [TRUNK_REF, POLICY_REF, OPLOG_REF].contains(&name.as_str()) {
4136 continue;
4137 }
4138 match check(&r.chain.state, &name, &found) {
4139 RefCheck::Fine => {}
4140 RefCheck::NotLogged => r.store.ref_delete(&name, &found).or_fail()?,
4141 RefCheck::Differs(expected) => {
4142 match &expected {
4143 Some(e) => r.store.ref_force(&name, e).or_fail()?,
4144 None => r.store.ref_delete(&name, &found).or_fail()?,
4145 }
4146 undone.push((name, Some(found), expected));
4147 }
4148 RefCheck::Unknown => {
4149 r.store.ref_delete(&name, &found).or_fail()?;
4150 undone.push((name, Some(found), None));
4151 }
4152 }
4153 }
4154 let missing: Vec<(String, GitId)> = missing_changes(&r.chain.state, &present)
4155 .map(|(n, c)| (n, c.clone()))
4156 .collect();
4157 for (name, commit) in missing {
4158 if r.store.has_object(&commit) {
4159 r.store.ref_force(&name, &commit).or_fail()?;
4160 }
4161 undone.push((name, None, Some(commit)));
4162 }
4163 for (refname, found, expected) in undone {
4164 let op = Op::ImportUnmediated {
4165 refname: refname.clone(),
4166 found,
4167 expected,
4168 };
4169 r.submit(Submission::new(Actor::Daemon, op), now)
4170 .or_fail()?;
4171 if freeze {
4172 Self::freeze_for(r, &refname, now)?;
4173 }
4174 }
4175 Ok(())
4176 }
4177
4178 fn freeze_for(
4179 r: &mut RepoHandle,
4180 refname: &str,
4181 now: i64,
4182 ) -> std::result::Result<(), (String, String)> {
4183 if r.chain.state.frozen.is_none() {
4184 let reason = format!("{refname} changed outside Lonca");
4185 r.submit(Submission::new(Actor::Daemon, Op::Freeze { reason }), now)
4186 .or_fail()?;
4187 }
4188 Ok(())
4189 }
4190
4191 pub fn reconcile_all(&self) {
4193 let now = self.now();
4194 let mut inner = self.lock();
4195 let before: Vec<u64> = inner
4196 .repos
4197 .values()
4198 .map(|r| r.chain.state.next_seq)
4199 .collect();
4200 for r in inner.repos.values_mut().filter(|r| r.initialized()) {
4201 if let Err((code, message)) = Self::reconcile(r, now) {
4202 use crate::security::{Event, kind};
4203 let e = Event::new(now, kind::RECONCILE_ERROR)
4204 .repo(&r.name)
4205 .code(&code)
4206 .detail(&message);
4207 self.security.record(e);
4208 }
4209 let flag = self.home.freeze_flag(&r.name);
4211 if let Ok(text) = std::fs::read_to_string(&flag) {
4212 let reason: String = text.chars().filter(|c| !c.is_control()).take(200).collect();
4213 let reason = if reason.is_empty() {
4214 "auditor finding".to_string()
4215 } else {
4216 reason
4217 };
4218 if !self
4219 .flagged
4220 .lock()
4221 .unwrap_or_else(|e| e.into_inner())
4222 .contains(&r.name)
4223 {
4224 use crate::security::{Event, kind};
4225 let e = Event::new(now, kind::INTEGRITY_ALARM)
4226 .repo(&r.name)
4227 .detail(&reason);
4228 self.security.record(e);
4229 }
4230 self.flagged
4231 .lock()
4232 .unwrap_or_else(|e| e.into_inner())
4233 .insert(r.name.clone());
4234 if r.chain.state.frozen.is_none()
4235 && r.submit(Submission::new(Actor::Daemon, Op::Freeze { reason }), now)
4236 .is_err()
4237 {
4238 continue;
4239 }
4240 let _ = std::fs::remove_file(&flag);
4241 self.flagged
4242 .lock()
4243 .unwrap_or_else(|e| e.into_inner())
4244 .remove(&r.name);
4245 }
4246 }
4247 let after: Vec<u64> = inner
4248 .repos
4249 .values()
4250 .map(|r| r.chain.state.next_seq)
4251 .collect();
4252 drop(inner);
4253 if before != after {
4254 self.notify_appended();
4255 }
4256 }
4257
4258 pub fn process_one(&self, repo: &str) -> Result<Option<SignedOp>> {
4264 match self.process_batch(repo)? {
4265 Some(done) => Ok(Some(done)),
4266 None => self.process_head(repo),
4267 }
4268 }
4269
4270 fn verify(
4272 &self,
4273 repo: &str,
4274 store: &Store,
4275 vcfg: &Verification,
4276 (files, touched): (&FileMap, &[String]),
4277 tip: &GitId,
4278 reuse: bool,
4279 ) -> Result<Vec<Attestation>> {
4280 let tree = store.read_commit(tip)?.tree;
4281 let mut evidence = vec![self.verifier.syntax(store, files, touched, &tree)?];
4282 let key = (repo.to_string(), tree.clone(), digest_of(vcfg));
4283 let now = self.now();
4284 let known = self.verified_lock().get(&key, now).filter(|_| reuse);
4285 if let Some(known) = known {
4286 evidence.extend(known.into_iter().map(reused));
4287 return Ok(evidence);
4288 }
4289 self.verification_runs.fetch_add(1, Ordering::Relaxed);
4290 let run = self.verifier.run(store, files, &tree, vcfg)?;
4291 if reuse && !run.is_empty() && run.iter().all(|a| a.result == CheckResult::Pass) {
4292 self.verified_lock().insert(key, &run, now);
4293 }
4294 evidence.extend(run);
4295 Ok(evidence)
4296 }
4297
4298 fn process_batch(&self, repo: &str) -> Result<Option<SignedOp>> {
4306 let mut running = self.running();
4307 let (store, jobs, reaches, trunk, vcfg, joins, (keyed, digest)) = {
4308 let mut inner = self.lock();
4309 let Some(r) = inner.repos.get_mut(repo).filter(|r| r.initialized()) else {
4310 return Ok(None);
4311 };
4312 if r.chain.state.frozen.is_some() {
4313 return Ok(None);
4314 }
4315 let policy = r.policy()?;
4316 let vcfg = policy.config.verification.clone();
4317 let size = policy.config.queue.batch_size as usize;
4318 let t = r.chain.entry_time(self.now());
4319 let state = &r.chain.state;
4320 let quota = &policy.config.quota;
4321 let free = |c: &&ChangeId| {
4322 quota_problems(state, &state.changes[*c].actor, t, quota).is_empty()
4323 };
4324 let job = |c: &ChangeId| {
4325 (
4326 *c,
4327 state.changes[c].commit.clone(),
4328 state.changes[c].base.clone(),
4329 )
4330 };
4331 let reach = |c: &ChangeId| {
4332 let rec = &state.changes[c];
4333 (rec.facts.reaches_others(), rec.owed_checks(&vcfg))
4334 };
4335 let held: Vec<ChangeId> = state
4336 .queue
4337 .iter()
4338 .take(size)
4339 .take_while(free)
4340 .copied()
4341 .collect();
4342 let joins: Vec<_> = (0..held.len())
4344 .map(|k| state.joins(held[k], &held[..k], &policy.config))
4345 .collect();
4346 let jobs: Vec<_> = held.iter().map(job).collect();
4347 let reaches: Vec<_> = state.queue.iter().take(size).map(reach).collect();
4348 let trunk = state.trunk.clone().expect("initialized");
4349 if jobs.len() >= 2 {
4350 running.hold(jobs.iter().map(|j| j.0));
4351 }
4352 let keyed = (policy.keyed.clone(), policy.digest.clone());
4353 (r.store.clone(), jobs, reaches, trunk, vcfg, joins, keyed)
4354 };
4355 if jobs.len() < 2 {
4357 return Ok(None);
4358 }
4359 let mut stack: Vec<(ChangeId, GitId, FileMap, Vec<String>)> = Vec::new();
4361 let mut tip = trunk.clone();
4362 for ((change, commit, base), joins) in jobs.iter().zip(&joins) {
4363 let may_join = |s: &str| joins.allows(s);
4364 let i = integrate_joining(
4365 &store,
4366 ENGINE_CURRENT,
4367 base,
4368 &tip,
4369 commit,
4370 &may_join,
4371 &keyed,
4372 );
4373 let Some((merged, i)) = i.ok().and_then(|i| Some((i.merged.clone()?, i))) else {
4374 break;
4375 };
4376 tip = merged.clone();
4377 stack.push((*change, merged, i.files, i.touched));
4378 }
4379 let mut length = stack.len();
4380 while length >= 2 {
4381 let prefix = &stack[..length];
4382 let (_, tip, files, _) = &prefix[length - 1];
4383 let touched: BTreeSet<String> =
4384 prefix.iter().flat_map(|p| p.3.iter().cloned()).collect();
4385 let touched: Vec<String> = touched.into_iter().collect();
4386 let reach = reaches[..length].iter().any(|r| r.0);
4387 let named = reaches[..length].iter().flat_map(|r| r.1.iter().cloned());
4388 let owed = vcfg.for_change(reach, &named.collect());
4389 let evidence = self.verify(repo, &store, &owed, (files, &touched), tip, true);
4390 if let Some(evidence) = evidence
4391 .ok()
4392 .filter(|e| evidence_problems(&owed, e).is_empty())
4393 {
4394 let merged = prefix.iter().map(|p| (p.0, p.1.clone())).collect();
4395 let joins = &joins[..length];
4396 return self.log_batch(repo, trunk, merged, (&jobs, joins), &digest, evidence);
4397 }
4398 length /= 2;
4399 }
4400 Ok(None)
4401 }
4402
4403 fn log_batch(
4404 &self,
4405 repo: &str,
4406 trunk: GitId,
4407 merged: Vec<(ChangeId, GitId)>,
4408 (jobs, joins): (&[(ChangeId, GitId, GitId)], &[Joins]),
4409 digest: &str,
4410 evidence: Vec<Attestation>,
4411 ) -> Result<Option<SignedOp>> {
4412 let mut inner = self.lock();
4413 let r = inner.repos.get_mut(repo).expect("exists");
4414 if r.policy()?.digest != digest {
4417 return Ok(None);
4418 }
4419 let config = r.policy()?.config.clone();
4420 let state = &r.chain.state;
4421 let still = jobs.iter().take(merged.len()).zip(&state.queue);
4422 let same = still.filter(|(job, queued)| job.0 == **queued);
4423 let same = same.filter(|(job, _)| same_proposal(state, job.0, &job.1, &job.2));
4424 if same.count() != merged.len() || state.trunk.as_ref() != Some(&trunk) {
4425 return Ok(None);
4426 }
4427 let members: Vec<ChangeId> = merged.iter().map(|m| m.0).collect();
4430 let now = (0..members.len()).map(|k| state.joins(members[k], &members[..k], &config));
4431 if !now.eq(joins.iter().cloned()) {
4432 return Ok(None);
4433 }
4434 let tip = merged.last().map(|(_, tip)| tip.clone()).expect("a batch");
4435 let op = Op::ChangeIntegrateBatch {
4436 trunk_before: trunk.clone(),
4437 merged,
4438 };
4439 let mut sub = Submission::new(Actor::Daemon, op);
4440 sub.evidence = evidence;
4441 sub.symbols_engine = Some(ENGINE_CURRENT);
4442 let signed = r.submit(sub, self.now())?;
4443 if signed.entry.decision.is_allow() {
4444 r.store.ref_cas(TRUNK_REF, &tip, Some(&trunk))?;
4445 }
4446 drop(inner);
4447 self.notify_appended();
4448 Ok(Some(signed))
4449 }
4450
4451 fn process_head(&self, repo: &str) -> Result<Option<SignedOp>> {
4453 let mut running = self.running();
4454 let (store, change, commit, base, trunk, vcfg, reuse, past, joins, (keyed, digest)) = {
4455 let mut inner = self.lock();
4456 let Some(r) = inner.repos.get_mut(repo).filter(|r| r.initialized()) else {
4457 return Ok(None);
4458 };
4459 if r.chain.state.frozen.is_some() {
4460 return Ok(None);
4461 }
4462 let Some(change) = r.chain.state.queue.first().copied() else {
4463 return Ok(None);
4464 };
4465 let rec = &r.chain.state.changes[&change];
4466 let (commit, base) = (rec.commit.clone(), rec.base.clone());
4467 let reaches = rec.facts.reaches_others();
4468 let trunk = r.chain.state.trunk.clone().expect("initialized");
4469 let policy = r.policy()?;
4470 let named = r.chain.state.changes[&change].owed_checks(&policy.config.verification);
4471 let vcfg = policy.config.verification.for_change(reaches, &named);
4472 let reuse = policy.config.queue.batch_size > 1;
4473 let t = r.chain.entry_time(self.now());
4474 let state = &r.chain.state;
4475 let actor = &state.changes[&change].actor;
4476 let past = !quota_problems(state, actor, t, &policy.config.quota).is_empty();
4477 let joins = state.joins(change, &[], &policy.config);
4478 running.hold([change]);
4479 (
4480 r.store.clone(),
4481 change,
4482 commit,
4483 base,
4484 trunk,
4485 vcfg,
4486 reuse,
4487 past,
4488 joins,
4489 (policy.keyed.clone(), policy.digest.clone()),
4490 )
4491 };
4492 let may_join = |s: &str| joins.allows(s);
4493 let integ = integrate_joining(
4494 &store,
4495 ENGINE_CURRENT,
4496 &base,
4497 &trunk,
4498 &commit,
4499 &may_join,
4500 &keyed,
4501 )?;
4502 let mut evidence = Vec::new();
4503 if let Some(m) = integ.merged.as_ref().filter(|_| !past) {
4504 let changed = (&integ.files, integ.touched.as_slice());
4505 evidence = self.verify(repo, &store, &vcfg, changed, m, reuse)?;
4506 }
4507 let mut inner = self.lock();
4508 let r = inner.repos.get_mut(repo).expect("exists");
4509 if r.chain.state.queue.first() != Some(&change)
4510 || !same_proposal(&r.chain.state, change, &commit, &base)
4511 || r.chain.state.trunk.as_ref() != Some(&trunk)
4512 {
4513 return Ok(None);
4514 }
4515 let config = &r.policy()?.config;
4518 if r.chain.state.joins(change, &[], config) != joins || r.policy()?.digest != digest {
4519 drop(inner);
4520 return self.process_head(repo);
4521 }
4522 let now = self.now();
4525 let t = r.chain.entry_time(now);
4526 let quota = r.policy()?.config.quota.clone();
4527 let actor = &r.chain.state.changes[&change].actor;
4528 if past && quota_problems(&r.chain.state, actor, t, "a).is_empty() {
4529 drop(inner);
4530 return self.process_head(repo);
4531 }
4532 let op = Op::ChangeIntegrate {
4533 change,
4534 trunk_before: trunk.clone(),
4535 merged: integ.merged.clone(),
4536 conflicts: integ.conflicts,
4537 };
4538 let mut sub = Submission::new(Actor::Daemon, op);
4539 sub.evidence = evidence;
4540 sub.symbols_engine = Some(ENGINE_CURRENT);
4541 let signed = r.submit(sub, now)?;
4542 if signed.entry.decision.is_allow()
4543 && let Some(m) = &integ.merged
4544 {
4545 r.store.ref_cas(TRUNK_REF, m, Some(&trunk))?;
4546 }
4547 drop(inner);
4548 self.notify_appended();
4549 Ok(Some(signed))
4550 }
4551
4552 pub fn verification_runs(&self) -> u64 {
4555 self.verification_runs.load(Ordering::Relaxed)
4556 }
4557
4558 fn running_lock(&self) -> std::sync::MutexGuard<'_, BTreeSet<ChangeId>> {
4559 self.running.lock().unwrap_or_else(|e| e.into_inner())
4560 }
4561
4562 fn running(&self) -> Running<'_> {
4563 Running {
4564 set: &self.running,
4565 held: Vec::new(),
4566 }
4567 }
4568
4569 fn verified_lock(&self) -> std::sync::MutexGuard<'_, Verified> {
4570 self.verified.lock().unwrap_or_else(|e| e.into_inner())
4571 }
4572
4573 pub fn process_all(&self) -> usize {
4574 let names: Vec<String> = self.lock().repos.keys().cloned().collect();
4575 let mut n = 0;
4576 let mut work: VecDeque<String> = names.into();
4577 while let Some(name) = work.pop_front() {
4578 match self.process_one(&name) {
4579 Ok(Some(done)) => {
4580 n += match &done.entry.op {
4581 Op::ChangeIntegrateBatch { merged, .. } => merged.len(),
4582 _ => 1,
4583 };
4584 work.push_back(name);
4585 }
4586 Ok(None) => {}
4587 Err(e) => eprintln!("lonca: queue {name}: {e}"),
4588 }
4589 }
4590 n
4591 }
4592
4593 fn sweep_all(&self) {
4596 let mut inner = self.lock();
4597 let now = self.now();
4598 if let Err((code, message)) = Self::sweep_tokens(&mut inner, now) {
4599 eprintln!("lonca: sweeping tokens: {code}: {message}");
4600 }
4601 }
4602
4603 pub fn spawn_worker(self: &Arc<Self>) -> std::thread::JoinHandle<()> {
4605 let svc = self.clone();
4606 std::thread::spawn(move || {
4607 let mut seen = 0;
4608 loop {
4609 svc.process_all();
4610 svc.reconcile_all();
4611 svc.sweep_all();
4612 svc.take_alarms();
4613 let (m, c) = &svc.wake;
4614 let guard = m.lock().unwrap_or_else(|e| e.into_inner());
4615 if *guard == seen {
4616 let (g, _) = c
4617 .wait_timeout(guard, Duration::from_secs(2))
4618 .unwrap_or_else(|e| e.into_inner());
4619 seen = *g;
4620 } else {
4621 seen = *guard;
4622 }
4623 }
4624 })
4625 }
4626}
4627
4628fn truncate_at_line(mut text: String, cap: usize) -> (String, bool) {
4632 if text.len() <= cap {
4633 return (text, false);
4634 }
4635 let mut cut = cap;
4636 while !text.is_char_boundary(cut) {
4637 cut -= 1;
4638 }
4639 text.truncate(cut);
4640 match text.rfind('\n') {
4641 Some(nl) => text.truncate(nl + 1),
4642 None => text.clear(),
4643 }
4644 (text, true)
4645}
4646
4647fn runs_for(vc: &Verification, c: &ChangeRecord) -> Vec<String> {
4653 let owed = vc.for_change(c.facts.reaches_others(), &c.owed_checks(vc));
4654 let builtin = [
4655 ("build", &owed.build),
4656 ("tests", &owed.test),
4657 ("extended", &owed.on_signature_change),
4658 ];
4659 let builtin = builtin.into_iter().filter(|(_, c)| c.is_some());
4660 let builtin = builtin.map(|(name, _)| name.to_string());
4661 builtin.chain(owed.checks.into_keys()).collect()
4662}
4663
4664fn touched_paths(f: &lonca_core::ops::Facts) -> Vec<String> {
4665 let mut paths: Vec<String> = f
4666 .added_files
4667 .iter()
4668 .chain(&f.modified_files)
4669 .chain(&f.deleted_files)
4670 .cloned()
4671 .collect();
4672 paths.sort();
4673 paths.dedup();
4674 paths
4675}
4676
4677fn validate_remote(r: &str) -> std::result::Result<(), (String, String)> {
4681 let ok_scheme = r.starts_with("https://") || r.starts_with("ssh://") || r.starts_with('/');
4682 let scp_like = !r.contains("://") && r.contains('@') && r.contains(':') && !r.starts_with('/');
4683 if r.starts_with('-')
4684 || r.contains("::")
4685 || r.chars().any(char::is_control)
4686 || !(ok_scheme || scp_like)
4687 {
4688 return fail(
4689 "invalid",
4690 "remote must be an absolute path, https://, ssh:// or user@host:path",
4691 );
4692 }
4693 Ok(())
4694}
4695
4696fn validate_branch(b: &str) -> std::result::Result<(), (String, String)> {
4697 let ok = !b.is_empty()
4698 && !b.starts_with('-')
4699 && !b.contains("..")
4700 && b.chars()
4701 .all(|c| c.is_ascii_alphanumeric() || matches!(c, '/' | '_' | '-' | '.'));
4702 if ok {
4703 Ok(())
4704 } else {
4705 fail("invalid", "invalid branch name")
4706 }
4707}
4708
4709#[cfg(unix)]
4710fn set_mode(p: &std::path::Path, mode: u32) -> std::result::Result<(), (String, String)> {
4711 use std::os::unix::fs::PermissionsExt;
4712 std::fs::set_permissions(p, std::fs::Permissions::from_mode(mode))
4713 .map_err(|e| from_err(e.into()))
4714}
4715
4716#[cfg(not(unix))]
4717fn set_mode(_p: &std::path::Path, _mode: u32) -> std::result::Result<(), (String, String)> {
4718 Ok(())
4719}
4720
4721pub fn is_ancestor(store: &Store, ancestor: &GitId, commit: &GitId) -> Result<bool> {
4723 let mut seen = std::collections::BTreeSet::new();
4724 let mut todo = vec![commit.clone()];
4725 while let Some(c) = todo.pop() {
4726 if &c == ancestor {
4727 return Ok(true);
4728 }
4729 if !seen.insert(c.clone()) || seen.len() > 1_000_000 {
4730 continue;
4731 }
4732 todo.extend(store.read_commit(&c)?.parents);
4733 }
4734 Ok(false)
4735}
4736
4737#[cfg(test)]
4738mod tests {
4739 #[test]
4742 fn a_watcher_gets_the_kind_and_never_the_words() {
4743 use super::{ChangeId, SessionEvent as E, watched_event};
4744 use lonca_core::ops::{LeaseItem, NegotiationMsg};
4745 let c = ChangeId::random();
4746 let words = "WORDS-OF-ANOTHER-AGENT";
4747 let cases = [
4748 (
4749 E::Integrated { seq: 1, change: c },
4750 "integrated",
4751 Some(c),
4752 None,
4753 ),
4754 (
4755 E::Conflicted {
4756 seq: 1,
4757 change: c,
4758 conflicts: vec![words.into()],
4759 },
4760 "conflicted",
4761 Some(c),
4762 None,
4763 ),
4764 (
4765 E::Failed {
4766 seq: 1,
4767 change: c,
4768 reasons: vec![words.into()],
4769 },
4770 "failed",
4771 Some(c),
4772 None,
4773 ),
4774 (
4775 E::Approved {
4776 seq: 1,
4777 target_seq: 0,
4778 },
4779 "approved",
4780 None,
4781 None,
4782 ),
4783 (
4784 E::Rejected {
4785 seq: 1,
4786 target_seq: 0,
4787 reason: words.into(),
4788 },
4789 "rejected",
4790 None,
4791 None,
4792 ),
4793 (
4794 E::Message {
4795 seq: 1,
4796 thread: words.into(),
4797 from: Some("s9".into()),
4798 message: NegotiationMsg::Counter { note: words.into() },
4799 },
4800 "message",
4801 None,
4802 Some("s9"),
4803 ),
4804 (
4805 E::LeaseReceived {
4806 seq: 1,
4807 set: words.into(),
4808 from: Some("s9".into()),
4809 },
4810 "lease_received",
4811 None,
4812 Some("s9"),
4813 ),
4814 (
4815 E::SymbolsChanged {
4816 seq: 1,
4817 change: c,
4818 from: words.into(),
4819 sets: vec![words.into()],
4820 symbols: vec![words.into()],
4821 },
4822 "symbols_changed",
4823 Some(c),
4824 None,
4825 ),
4826 ];
4827 for (event, kind, change, from) in cases {
4828 let w = watched_event("s1", &event);
4829 assert_eq!(w.session, "s1");
4830 assert_eq!(w.kind, kind);
4831 assert_eq!(w.seq, Some(1), "{kind}");
4832 assert_eq!(w.change, change, "{kind}");
4833 assert_eq!(w.from.as_deref(), from, "{kind}");
4834 let said = serde_json::to_string(&w).unwrap();
4835 assert!(!said.contains(words), "{said}");
4836 }
4837 let r = E::Reserved {
4838 items: vec![LeaseItem {
4839 symbol: words.into(),
4840 kind: lonca_core::ops::LeaseKind::Edit,
4841 names: vec![],
4842 }],
4843 until_ms: 7,
4844 };
4845 let w = watched_event("s1", &r);
4846 assert_eq!(
4847 (w.kind.as_str(), w.seq, w.until_ms),
4848 ("reserved", None, Some(7))
4849 );
4850 assert!(!serde_json::to_string(&w).unwrap().contains(words));
4851 }
4852
4853 #[test]
4854 fn a_reservation_s_name_follows_its_session_and_items() {
4855 use super::reservation_id;
4856 use lonca_core::ops::{LeaseItem, LeaseKind};
4857 let item = |s: &str| LeaseItem {
4858 symbol: s.into(),
4859 kind: LeaseKind::Edit,
4860 names: vec![],
4861 };
4862 let a = reservation_id("s1", &[item("x")]);
4863 assert_eq!(a.len(), 16);
4864 assert_eq!(a, reservation_id("s1", &[item("x")]));
4865 assert_ne!(a, reservation_id("s2", &[item("x")]));
4866 assert_ne!(a, reservation_id("s1", &[item("y")]));
4867 }
4868
4869 #[test]
4870 fn a_prefix_names_one_change_or_says_why_not() {
4871 use super::{ChangeId, resolve_prefix};
4872 let id = |s: &str| ChangeId::from_reverse_hex(s).unwrap();
4873 let a = id("zzzzkkkkzzzzkkkkzzzzkkkkzzzzkkkk");
4874 let b = id("zzzzykkkzzzzkkkkzzzzkkkkzzzzkkkk");
4875 let ids = [a, b];
4876 assert_eq!(
4877 resolve_prefix(&ids, "zzzzk").unwrap(),
4878 serde_json::json!({"change": a.to_string()})
4879 );
4880 assert_eq!(
4881 resolve_prefix(&ids, &a.to_string()).unwrap()["change"],
4882 a.to_string()
4883 );
4884 let (code, msg) = resolve_prefix(&ids, "zzzz").unwrap_err();
4885 assert_eq!(code, "invalid");
4886 assert_eq!(
4887 msg,
4888 format!("2 changes start with zzzz: {a}, {b}; give more letters")
4889 );
4890 let (code, msg) = resolve_prefix(&ids, "yyyy").unwrap_err();
4891 assert_eq!(
4892 (code.as_str(), msg.as_str()),
4893 ("not_found", "no change starts with yyyy")
4894 );
4895 let many: Vec<ChangeId> = ["k", "l", "m", "n", "o", "p"]
4897 .iter()
4898 .map(|c| id(&format!("kkkk{}", c.repeat(28))))
4899 .collect();
4900 let (_, msg) = resolve_prefix(&many, "kkkk").unwrap_err();
4901 assert!(msg.starts_with("6 changes start with kkkk: "), "{msg}");
4902 assert!(msg.ends_with(", ...; give more letters"), "{msg}");
4903 assert!(
4904 msg.contains("kkkkoooo") && !msg.contains("kkkkpppp"),
4905 "{msg}"
4906 );
4907 let (_, msg) = resolve_prefix(&many[..5], "kkkk").unwrap_err();
4908 assert!(
4909 msg.ends_with("kkkkoooooooooooooooooooooooooooo; give more letters"),
4910 "{msg}"
4911 );
4912 }
4913
4914 use super::*;
4915
4916 fn verified_key(i: usize) -> VerifiedKey {
4917 let tree = GitId::of_blob(i.to_string().as_bytes());
4918 ("demo".to_string(), tree, "checks".to_string())
4919 }
4920
4921 #[test]
4923 fn the_daemon_trusts_only_root_s_files_unless_told_otherwise() {
4924 let file = Some(PathBuf::from("/m"));
4925 let strict = Settings::for_daemon(false, file.clone(), Some("p".into()), true);
4926 assert!(strict.require_root_settings);
4927 assert_eq!(strict.managed_settings, file);
4928 assert_eq!(strict.proxy.as_deref(), Some("p"));
4929 assert!(strict.hardened);
4930 assert!(!Settings::for_daemon(true, None, None, false).require_root_settings);
4931 assert!(!Settings::for_daemon(true, None, None, false).hardened);
4932 }
4933
4934 #[test]
4935 fn the_proxy_s_port_is_read_from_its_address() {
4936 let port = |proxy: Option<&str>| {
4937 Settings::for_daemon(false, None, proxy.map(String::from), false).proxy_port()
4938 };
4939 assert_eq!(port(Some("127.0.0.1:8471")), Some(8471));
4940 assert_eq!(port(Some("[::1]:8472")), Some(8472));
4941 assert_eq!(port(Some("p")), None);
4942 assert_eq!(port(None), None);
4943 }
4944
4945 #[test]
4946 fn a_verified_tree_is_known_until_the_hour() {
4947 let mut verified = Verified::default();
4948 verified.insert(verified_key(0), &[], 1_000);
4949 let at = |ms| verified.get(&verified_key(0), 1_000 + ms).is_some();
4950 assert!(at(VERIFIED_FOR_MS - 1));
4951 assert!(!at(VERIFIED_FOR_MS));
4952 }
4953
4954 #[test]
4956 fn the_oldest_verified_tree_goes_first() {
4957 let mut verified = Verified::default();
4958 for i in 0..VERIFIED_TREES {
4959 verified.insert(verified_key(i), &[], 0);
4960 }
4961 let known = |v: &Verified, i| v.get(&verified_key(i), 0).is_some();
4962 assert!((0..VERIFIED_TREES).all(|i| known(&verified, i)));
4963 verified.insert(verified_key(VERIFIED_TREES), &[], 0);
4964 assert!(!known(&verified, 0));
4965 assert!((1..=VERIFIED_TREES).all(|i| known(&verified, i)));
4966 }
4967
4968 #[test]
4969 fn queue_run_wakes_the_worker() {
4970 let dir = tempfile::tempdir().unwrap();
4971 let home = Home::new(dir.path().join("h"));
4972 let svc = Service::open(home, Settings::default(), system_clock()).unwrap();
4973 let before = *svc.wake.0.lock().unwrap();
4974 svc.notify_queue();
4975 assert_eq!(*svc.wake.0.lock().unwrap(), before + 1);
4976 }
4977
4978 #[test]
4979 fn a_log_page_is_what_was_asked_up_to_the_cap() {
4980 assert_eq!(log_page(Some(3)), 3);
4981 assert_eq!(log_page(None), LOG_PAGE as usize);
4982 assert_eq!(log_page(Some(u64::MAX)), LOG_PAGE as usize);
4983 }
4984
4985 fn documented_methods() -> BTreeMap<String, String> {
4987 let doc = include_str!("../../../docs/decisions.md");
4988 let section = doc
4989 .split("\n## ")
4990 .find(|s| s.starts_with("Daemon authentication"))
4991 .expect("a Daemon authentication section");
4992 section
4993 .lines()
4994 .filter_map(|l| l.strip_prefix("| `"))
4995 .map(|l| {
4996 let (method, rest) = l.split_once("` | ").expect("| `method` | cred |");
4997 let cred = rest.split(" |").next().unwrap_or_default().trim();
4998 (method.to_string(), cred.to_string())
4999 })
5000 .collect()
5001 }
5002
5003 #[test]
5008 fn the_protocol_doc_lists_every_method_and_states_the_number() {
5009 let doc = include_str!("../../../docs/protocol.md");
5010 let sections: Vec<&str> = doc
5011 .lines()
5012 .filter_map(|l| l.strip_prefix("### `"))
5013 .map(|l| l.trim_end_matches('`'))
5014 .collect();
5015 for (m, _) in METHODS {
5016 assert!(
5017 sections.contains(m),
5018 "{m} has no section in docs/protocol.md"
5019 );
5020 }
5021 assert_eq!(sections.len(), METHODS.len(), "{sections:?}");
5022 assert_eq!(PROTOCOL, 1);
5023 assert!(
5024 doc.contains(&format!("**Protocol number: {PROTOCOL}.**")),
5025 "docs/protocol.md must state the current protocol number"
5026 );
5027 assert!(doc.contains(&format!("\n- **{PROTOCOL}** (")));
5028 assert!(doc.contains("capped at 30 000") && LOG_WAIT_MAX_MS == 30_000);
5029 }
5030
5031 #[test]
5032 fn the_decisions_doc_lists_every_method_with_its_credential() {
5033 let documented = documented_methods();
5034 for (m, c) in METHODS {
5035 assert_eq!(
5036 documented.get(*m).map(String::as_str),
5037 Some(c.name()),
5038 "{m} in docs/decisions.md"
5039 );
5040 }
5041 assert_eq!(documented.len(), METHODS.len(), "{documented:?}");
5042 }
5043
5044 #[test]
5048 fn every_listed_method_is_dispatched_and_nothing_else() {
5049 let dir = tempfile::tempdir().unwrap();
5050 let svc = Service::open(
5051 Home::new(dir.path().join("h")),
5052 Settings {
5053 managed_settings: Some(dir.path().join("m.json")),
5054 require_root_settings: false,
5055 git: "git".into(),
5056 line_cap: MAX_LINE,
5057 hardened: false,
5058 proxy: None,
5059 ..Settings::default()
5060 },
5061 Arc::new(|| 1_760_000_000_000),
5062 )
5063 .unwrap();
5064 let call = |method: &str, token: Option<&str>| {
5065 let r = svc.handle(Request {
5066 id: 1,
5067 method: method.into(),
5068 repo: Some("demo".into()),
5069 token: token.map(String::from),
5070 params: json!({}),
5071 });
5072 r.error.map(|e| e.code).unwrap_or_default()
5073 };
5074 for (m, cred) in METHODS {
5075 let code = call(m, None);
5076 assert_ne!(code, "unknown_method", "{m}");
5077 match cred {
5078 Cred::Public | Cred::Login | Cred::Bootstrap => {
5079 assert_ne!(code, "unauthenticated", "{m}: {code}")
5080 }
5081 _ => assert_eq!(code, "unauthenticated", "{m}: {code}"),
5082 }
5083 assert_eq!(call(m, Some("lonca_nobody")), "unauthenticated", "{m}");
5085 }
5086 assert_eq!(call("admin.everything", None), "unknown_method");
5087 assert_eq!(call("", None), "unknown_method");
5088 }
5089
5090 #[test]
5093 fn a_diff_is_cut_on_a_line_inside_the_cap_never_inside_a_character() {
5094 assert_eq!(DIFF_CAP_BYTES, 2_097_152);
5095 assert_eq!(EVIDENCE_LOG_CAP_BYTES, 32_768);
5096 let short = "a\nb\n".to_string();
5097 assert_eq!(truncate_at_line(short.clone(), 4), (short.clone(), false));
5098 assert_eq!(truncate_at_line(short, 3), ("a\n".to_string(), true));
5099 let text: String = "xé\n".repeat(100);
5103 let (cut, truncated) = truncate_at_line(text, 202);
5104 assert!(truncated);
5105 assert_eq!(cut.len(), 200, "{}", cut.len());
5106 assert!(cut.ends_with("xé\n") && cut.is_char_boundary(cut.len()));
5107 assert_eq!(
5109 truncate_at_line("xé\n".to_string(), 2),
5110 (String::new(), true)
5111 );
5112 }
5113
5114 #[test]
5115 fn session_constants_are_what_the_document_says() {
5116 assert_eq!(HUMAN_SESSION_DEFAULT_MS, 3_600_000);
5117 assert_eq!(HUMAN_SESSION_MAX_MS, 86_400_000);
5118 assert_eq!(LOGIN_NONCE_WINDOW_MS, 660_000);
5119 assert_eq!(LOGIN_NONCE_WINDOW_MS, MAX_AGE_MS + MAX_SKEW_MS);
5120 }
5121
5122 #[test]
5123 fn every_error_keeps_its_message_under_a_stable_code() {
5124 let cases = [
5125 (Error::Invalid("a".into()), "invalid", "invalid: a"),
5126 (Error::Conflict("b".into()), "conflict", "conflict: b"),
5127 (Error::Integrity("c".into()), "integrity", "integrity: c"),
5128 (
5129 Error::Unsupported("op 3: log format 9".into()),
5130 "unsupported",
5131 "unsupported: op 3: log format 9",
5132 ),
5133 (Error::Git("d".into()), "internal", "git: d"),
5134 (Error::Io(std::io::Error::other("e")), "internal", "io: e"),
5135 ];
5136 for (err, code, msg) in cases {
5137 assert_eq!(from_err(err), (code.to_string(), msg.to_string()));
5138 }
5139 let json = serde_json::from_str::<Value>("{").unwrap_err();
5140 assert_eq!(from_err(Error::Json(json)).0, "internal");
5141 }
5142}