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