diff --git a/PLAN.md b/PLAN.md index cb1e40ac..b2971977 100644 --- a/PLAN.md +++ b/PLAN.md @@ -70,7 +70,8 @@ Landed so far: | 1 | #71 | Nothing | | 2 | #73, and attempt keys for a claim and a completion in #77 | An attempt key on the other operations; an uncertain external effect blocking reassignment; waiting on an outside event as a state; the record's remaining fields (`parent`, `scores`, `implements`, `reversibility`, `artifacts_touched`, `policy_context`) | | 3 | #74 | A suite that gates a version of a definition; the data scope enforced; a definition required by default, once the system's own filings name one; tools offered a few at boot and the rest by search | -| 6 | #78 | `check_policy` selecting rules by the action; the unused-rule check over one review period; the approver of an exception checked against the layer breached; a new version of a rule keeping its place under the ceiling; a decision that survives an enforced module that fails; the author and the date of an imported rule not trusted; a session that reloads the policy when a rule changes; a verified rule that stays in force while its next version waits for review; a denial that names no policy counted for the module that made it, in the rates and the lint | +| 6 | #78 | `check_policy` selecting rules by the action; the unused-rule check over one review period; the approver of an exception checked against the layer breached; a new version of a rule keeping its place under the ceiling; a decision that survives an enforced module that fails; the author and the date of an imported rule not trusted; a verified rule that stays in force while its next version waits for review; a denial that names no policy counted for the module that made it, in the rates and the lint | +| 10 | #79 | A stop for only the sessions that a change to the grants affects; rendered files removed when a session stops; the harness of a stopped session stopped; an expiry on a grant; an expiry that reaches a read filter from a root policy, and the advisory rules | What stays unbuilt after these, and stays on README's list: federation across domains, authentication of people, logprob-based drift, and accelerator time, which only a gateway can count. So do the items on that list that need another machine or an outside service more than they need code: the limits on re-enactment, harness settings and decomposition templates in the platform domain, bytes stored and resource limits on the sandbox, the authoring of agent definitions, the search tool against its live provider, a harness under Seatbelt on a Mac, and the Gofannon bridge against a live deployment, whose agents make model calls that the egress does not meter. diff --git a/README.md b/README.md index b9e3c47e..287b3c3e 100644 --- a/README.md +++ b/README.md @@ -142,8 +142,12 @@ two weeks. It is also the reason that none of it is the production shape. - **Curation and audit, in part.** No curation task demotes an entry, checks an owner or a cross-link, or curates policy. No test on planted faults qualifies the auditor's model. The auditor does not read handoffs. -- **Access freshness.** A session keeps its read filter until it ends. A change to the - grants or to a policy does not stop a session. +- **Access freshness, in part.** Any change to the grants file stops each open session, + not only a session that the change affects. The files that a session rendered before + the change stay on disk. A harness that a stopped session started continues to run, + and each of its memory calls and model calls is refused. A read that waits does not + check again before it gives its answer. Some commands of the owner do not do the + check. A grant has no expiry of its own. - **The improvement loop.** The only gate on a platform artifact is that a person verifies it. There is no performance gate, no safety gate, and no check that a change adds no permission. diff --git a/crates/egress/src/lib.rs b/crates/egress/src/lib.rs index 80ac3ae7..af437118 100644 --- a/crates/egress/src/lib.rs +++ b/crates/egress/src/lib.rs @@ -235,6 +235,8 @@ struct Inner { /// How long a tunnel has for the TLS hellos the egress reads, [`hello::HELLO_WAIT`] /// outside a test. hello_wait: std::time::Duration, + /// How often an open tunnel asks whether its session is still valid. + stop_check: std::time::Duration, } /// How the egress resolves a tunnel's host and dials it: the system's resolver and TCP, @@ -377,6 +379,7 @@ impl Egress { telemetry, network: Arc::new(SystemNetwork), hello_wait: hello::HELLO_WAIT, + stop_check: std::time::Duration::from_secs(1), }), }) } @@ -399,6 +402,10 @@ impl Egress { body: Bytes, ) -> Result { let inner = &self.inner; + inner + .memory + .fresh() + .map_err(|e| Error::Refused(e.to_string()))?; let dest = inner .tools .get(tool) @@ -834,13 +841,23 @@ fn over_budget(inner: &Inner, destination: &str) -> Option { .telemetry .spent(destination, None, Some(day_ago)) .unwrap_or(0); + // a check that cannot be made is a refusal: a session that has stopped, or a + // policy that does not evaluate, sends nothing match inner.memory.check_spend(destination, session, day) { Ok(d) if !d.allow => Some(d.reason), - _ => None, + Ok(_) => None, + Err(e) => Some(e.to_string()), } } fn admit(inner: &Inner, destination: &str) -> Result> { + if let Err(e) = inner.memory.fresh() { + return Err(Box::new(error_json( + StatusCode::FORBIDDEN, + "permission_error", + &e.to_string(), + ))); + } let Some(dest) = inner.destinations.get(destination) else { return Err(Box::new(error_json( StatusCode::NOT_FOUND, @@ -1033,6 +1050,9 @@ async fn tunnel_for( message, )) }; + if let Err(e) = inner.memory.fresh() { + return Err(refuse(&e.to_string())); + } let Some((host, port)) = host_and_port(authority) else { return Err(Box::new(error_json( StatusCode::BAD_REQUEST, @@ -1110,7 +1130,24 @@ async fn connect_or_not_found(State(egress): State, mut req: Request) -> } Err(Stop::Ended) => return, }; - let _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound).await; + // an open tunnel is the session's still: it carries bytes only while the session + // is valid, and is closed when the session stops + let stopped = async { + loop { + tokio::time::sleep(inner.stop_check).await; + if inner.memory.fresh().is_err() { + break; + } + } + }; + tokio::select! { + _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound) => {} + () = stopped => { + eprintln!("egress: tunnel to {host}: the session stopped; closed"); + let _ = inbound.shutdown().await; + let _ = outbound.shutdown().await; + } + } let _ = inner.telemetry.record( started, &session, @@ -1883,6 +1920,56 @@ mod tests { )); } + /// A session whose grants changed has stopped: no model call of it leaves, by the + /// library or by the router. + #[tokio::test] + async fn a_stopped_session_sends_no_model_call() { + let fake = Router::new().route( + "/v1/chat/completions", + any(|| async { + ( + [(header::CONTENT_TYPE, "application/json")], + r#"{"model":"m","choices":[{"message":{"content":"ok"}}],"usage":{"prompt_tokens":4,"completion_tokens":2}}"#, + ) + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, fake).await.unwrap() }); + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().join("m"); + Memory::init(&root, "human:test:me", "dom-me").unwrap(); + let m = Arc::new(Memory::open(&root).unwrap().with_author("agent:a")); + let mut d = HashMap::new(); + d.insert( + "openai".to_string(), + Destination::forward(format!("http://{addr}"), Auth::PassThrough), + ); + let e = Egress::new(m, d, &dir.path().join("t.db")).unwrap(); + let ask = serde_json::json!({ "messages": [{ "role": "user", "content": "hi" }] }); + e.call("openai", "v1/chat/completions", &ask).await.unwrap(); + let grants = root.join("grants.toml"); + let text = std::fs::read_to_string(&grants).unwrap(); + std::fs::write(&grants, format!("{text}\n# the owner changed the grants\n")).unwrap(); + assert!(e.call("openai", "v1/chat/completions", &ask).await.is_err()); + let r = e + .router() + .oneshot( + Request::builder() + .uri("/openai/v1/chat/completions") + .method("POST") + .header("content-type", "application/json") + .body(Body::from(ask.to_string())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(r.status(), StatusCode::FORBIDDEN); + let body = to_bytes(r.into_body(), 1 << 20).await.unwrap(); + let said = String::from_utf8_lossy(&body); + assert!(said.contains("stopped"), "{said}"); + } + #[tokio::test] async fn a_model_call_over_the_spend_limit_does_not_leave() { let fake = Router::new().route( @@ -3555,6 +3642,33 @@ mod tests { (answer, c) } + /// A tunnel that is open when its session stops is closed: a harness that keeps a + /// connection alive sends nothing more through it. + #[tokio::test] + async fn an_open_tunnel_is_closed_when_its_session_stops() { + use tokio::io::AsyncWriteExt; + let (addr, _) = echo_upstream().await; + let entry = format!("127.0.0.1:{}", addr.port()); + let (d, mut e) = fixture(Some(("github", Destination::tunnel([entry])))).await; + Arc::get_mut(&mut e.inner).unwrap().stop_check = std::time::Duration::from_millis(50); + let egress = serve(&e).await; + let (head, mut c) = connect(egress, &addr.to_string()).await; + assert!(head.starts_with("HTTP/1.1 200"), "{head}"); + echoes(&mut c, "127.0.0.1").await; + let grants = d.path().join("m").join(dm_server::GRANTS_FILE); + let text = std::fs::read_to_string(&grants).unwrap(); + std::fs::write(&grants, format!("{text}\n# a change\n")).unwrap(); + let (rest, closed) = until_closed(&mut c).await; + assert!(closed, "the tunnel stayed open after the session stopped"); + assert_eq!(rest, b""); + // and nothing written now comes back + let _ = c.write_all(b"sent after the session stopped").await; + assert_eq!(until_closed(&mut c).await.0, b""); + // nor does a new tunnel open + let (head, _) = connect(egress, &addr.to_string()).await; + assert!(head.starts_with("HTTP/1.1 403"), "{head}"); + } + #[tokio::test] async fn a_connect_tunnel_opens_only_to_allowed_hosts_within_the_ceiling() { let (addr, _) = echo_upstream().await; diff --git a/crates/index/src/lib.rs b/crates/index/src/lib.rs index adf2b1a4..59d913f8 100644 --- a/crates/index/src/lib.rs +++ b/crates/index/src/lib.rs @@ -277,6 +277,22 @@ impl Index { Index::init(conn) } + /// An index that exists, opened to read: nothing is created or changed, and a + /// writer elsewhere is waited for only briefly. For a reader that may be called + /// while another handle of the same process is at work. + /// + /// # Errors + /// + /// When the file is not there or cannot be opened. + pub fn open_reader(path: &Path) -> Result { + let conn = Connection::open_with_flags( + path, + rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX, + )?; + conn.busy_timeout(std::time::Duration::from_millis(250))?; + Ok(Index { conn }) + } + /// An index that lives only in memory. /// /// # Errors diff --git a/crates/policy/src/decide.rs b/crates/policy/src/decide.rs index 49463045..35dc80cc 100644 --- a/crates/policy/src/decide.rs +++ b/crates/policy/src/decide.rs @@ -450,6 +450,10 @@ pub struct Policy { refused: Vec<(String, String)>, /// The entries of the rules the ceiling holds at advisory. held_back: Vec, + /// The data the record gave the engine, and when each exception in it lapses: an + /// expiry is read against the clock at each decision, never compiled in. + data: serde_json::Value, + lapses: Vec<(String, time::OffsetDateTime)>, /// One engine for each advisory rule: the base with that rule's module alone. What /// it denies is reported and never joins the decision, and a module that fails /// takes no other rule's report with it. @@ -487,16 +491,38 @@ impl Policy { /// A policy that will not compile, or a file that cannot be read. pub fn load(root: &Path, grants: &Grants) -> Result { let dir = root.join("policy"); - let mut files: Vec<_> = std::fs::read_dir(&dir) - .map(|d| { - d.filter_map(Result::ok) - .map(|e| e.path()) - .filter(|p| p.extension().is_some_and(|x| x == "rego")) - .collect() - }) - .unwrap_or_default(); + match Policy::root_text(root)? { + Some(text) => Policy::compile(&text, &dir.display().to_string(), grants), + None => Policy::builtin(grants), + } + } + + /// The text of a root's own policy, its `policy/*.rego` in the order of their + /// names; none when the root has no such file and the built-in policy stands. + /// + /// # Errors + /// + /// A policy file that cannot be read. + pub fn root_text(root: &Path) -> Result, Error> { + let dir = root.join("policy"); + // no directory is no policy of the root's own; one that cannot be read is an + // error, since the built-in policy may allow what the root's would not + let mut files: Vec<_> = match std::fs::read_dir(&dir) { + Ok(d) => d + .filter_map(Result::ok) + .map(|e| e.path()) + .filter(|p| p.extension().is_some_and(|x| x == "rego")) + .collect(), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(), + Err(e) => { + return Err(Error::Io { + path: dir.display().to_string(), + source: e, + }) + } + }; if files.is_empty() { - return Policy::builtin(grants); + return Ok(None); } files.sort(); let mut text = String::new(); @@ -507,7 +533,7 @@ impl Policy { })?); text.push('\n'); } - Policy::compile(&text, &dir.display().to_string(), grants) + Ok(Some(text)) } /// This policy with the record's policy entries: each verified entry's module is @@ -590,6 +616,12 @@ impl Policy { .map(|(id, engine)| Ok((id, Policy::with_data(engine, &data)?))) .collect::>()?; self.version = digest(&texts); + self.lapses = entries + .iter() + .filter(|e| e.in_force(now)) + .filter_map(|e| Some((e.meta.id.clone(), e.lapses()?))) + .collect(); + self.data = data; self.entries = entries; self.held_back = over; Ok(self) @@ -710,6 +742,8 @@ impl Policy { entries: Vec::new(), refused: Vec::new(), held_back: Vec::new(), + data: serde_json::Value::Null, + lapses: Vec::new(), watch: Vec::new(), }) } @@ -738,12 +772,35 @@ impl Policy { /// /// When the policy fails to evaluate or its answer is not a scope. pub fn read_scope(&self, input: &Input) -> Result { - let json = self.eval("data.amem.read_filter", input)?; + let json = self.eval( + "data.amem.read_filter", + input, + time::OffsetDateTime::now_utc(), + )?; serde_json::from_str(&json).map_err(|e| Error::Policy(format!("{e}: {json}"))) } - fn eval(&self, rule: &str, input: &Input) -> Result { - let mut engine = self.engine.clone(); + /// The engine as it stands at `now`: without the exceptions that have lapsed since + /// the policy was compiled. + fn engine_at(&self, now: time::OffsetDateTime) -> Result { + let lapsed: Vec<&str> = self + .lapses + .iter() + .filter(|(_, at)| *at <= now) + .map(|(id, _)| id.as_str()) + .collect(); + if lapsed.is_empty() { + return Ok(self.engine.clone()); + } + let mut data = self.data.clone(); + if let Some(list) = data["exceptions"].as_array_mut() { + list.retain(|e| !lapsed.contains(&e["id"].as_str().unwrap_or_default())); + } + Policy::with_data(self.engine.clone(), &data) + } + + fn eval(&self, rule: &str, input: &Input, now: time::OffsetDateTime) -> Result { + let mut engine = self.engine_at(now)?; let value = regorus::Value::from_json_str( &serde_json::to_string(input).map_err(|e| Error::Policy(e.to_string()))?, ) @@ -762,7 +819,17 @@ impl Policy { /// /// When the policy fails to evaluate or its decision is not the shape expected. pub fn decide(&self, input: &Input) -> Result { - let json = self.eval("data.amem.decision", input)?; + self.decide_at(input, time::OffsetDateTime::now_utc()) + } + + /// Decide one request as the clock reads `now`: an exception that has lapsed by + /// then lets nothing through, whenever the policy was compiled. + /// + /// # Errors + /// + /// As [`Policy::decide`]. + pub fn decide_at(&self, input: &Input, now: time::OffsetDateTime) -> Result { + let json = self.eval("data.amem.decision", input, now)?; let mut d: Decision = serde_json::from_str(&json).map_err(|e| Error::Policy(format!("{e}: {json}")))?; d.policy = self.version.clone(); @@ -1075,7 +1142,11 @@ mod tests { } fn supersede_decision(p: &Policy, held: Option<&str>) -> Decision { - p.decide(&Input { + p.decide(&supersede_input(held)).unwrap() + } + + fn supersede_input(held: Option<&str>) -> Input { + Input { action: Action::Write, principal: Principal { id: "agent:x".into(), @@ -1096,8 +1167,49 @@ mod tests { held_task: held.map(str::to_string), spend: None, fetch: None, - }) - .unwrap() + } + } + + /// An expiry is read against the clock at each decision: a policy compiled while + /// an exception was in force refuses once the exception has lapsed. + #[test] + fn an_exception_lapses_at_its_expiry_whenever_the_policy_was_compiled() { + use dm_core::TrustTier::Verified; + let g = grants(); + let rule = rule_entry("pol-L2-031", 2, Verified, "", true); + let exception = |id: &str, expires: &str| { + rule_entry( + id, + 2, + Verified, + &format!( + "breaches = \"pol-L2-031\"\nscope_task = \"task-t-1\"\nexpires = \"{expires}\"" + ), + false, + ) + }; + let compiled = time::macros::datetime!(2026-11-01 00:00 UTC); + let short = exception("exc-2026-010", "2026-11-02T00:00:00Z"); + let p = Policy::builtin(&g) + .unwrap() + .with_record(&g, vec![rule.clone(), short.clone()], compiled) + .unwrap(); + let input = supersede_input(Some("task-t-1")); + let before = time::macros::datetime!(2026-11-01 12:00 UTC); + let at = time::macros::datetime!(2026-11-02 00:00 UTC); + assert!(p.decide_at(&input, before).unwrap().allow); + let d = p.decide_at(&input, at).unwrap(); + assert!(!d.allow, "{d:?}"); + assert_eq!(d.cites, ["pol-L2-031"]); + // an exception that still runs is kept when another lapses + let long = exception("exc-2026-011", "2026-12-01T00:00:00Z"); + let p = Policy::builtin(&g) + .unwrap() + .with_record(&g, vec![rule, short, long], compiled) + .unwrap(); + let d = p.decide_at(&input, at).unwrap(); + assert!(d.allow, "{d:?}"); + assert_eq!(d.cites, ["exc-2026-011"]); } fn with(g: &Grants, entries: Vec) -> Policy { diff --git a/crates/policy/src/record.rs b/crates/policy/src/record.rs index 15510378..1dad9de1 100644 --- a/crates/policy/src/record.rs +++ b/crates/policy/src/record.rs @@ -243,6 +243,13 @@ impl Entry { self.state() == Status::Advisory && self.rego.is_some() && !self.meta.is_exception() } + /// When, as an exception, it lapses: its expiry, if that is a time. + #[must_use] + pub fn lapses(&self) -> Option { + let expires = self.meta.expires.as_deref()?; + time::OffsetDateTime::parse(expires, &time::format_description::well_known::Rfc3339).ok() + } + /// Whether, as an exception, it is in force at `now`: verified, not expired, and /// not retired. #[must_use] @@ -250,10 +257,7 @@ impl Entry { self.meta.is_exception() && self.meta.status != Some(Status::Retired) && self.tier == dm_core::TrustTier::Verified - && self.meta.expires.as_deref().is_some_and(|e| { - time::OffsetDateTime::parse(e, &time::format_description::well_known::Rfc3339) - .is_ok_and(|t| t > now) - }) + && self.lapses().is_some_and(|t| t > now) } } diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index ebb81a67..a5c2aa53 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -73,6 +73,23 @@ fn check_write_size(size: usize) -> Result<(), Error> { } /// The index, relative to the root. pub const INDEX_FILE: &str = "state/index.db"; +/// What changes whenever a session changes the policy: every other session compares it +/// with what it last loaded. +pub const POLICY_STAMP_FILE: &str = "state/policy.stamp"; + +/// The policy stamp as the root holds it now: what the last change to a policy entry +/// left, and the digest of the root's own policy files. A file that cannot be read +/// gives a stamp no session has seen, so the session loads the policy again and is +/// refused there. +fn policy_stamp(root: &Path) -> String { + let entries = std::fs::read_to_string(root.join(POLICY_STAMP_FILE)).unwrap_or_default(); + let own = match Policy::root_text(root) { + Ok(Some(text)) => dm_core::digest(&text), + Ok(None) => String::new(), + Err(e) => format!("unreadable: {e}"), + }; + format!("{entries}|{own}") +} /// The ledger, relative to the root. pub const LEDGER_FILE: &str = "state/ledger.db"; @@ -112,6 +129,10 @@ pub enum Error { /// The grants do not allow it. #[error("not allowed: {0}")] Denied(String), + /// The session's authority is no longer what it started with, or cannot be shown: + /// the session stops, and a new one starts under the grants as they are now. + #[error("stopped: {0}")] + Stale(String), /// Not there, or not for this session to see. The two are deliberately the same. #[error("not found: {0}")] NotFound(String), @@ -308,7 +329,15 @@ pub struct Memory { claims: Arc>>, /// What this session may read, as the policy computed it once and the index applies /// inside every query. - scope: OnceLock, + /// It is kept with the version of the policy that computed it, and is computed + /// again when the policy in force is another. + scope: Mutex>, + /// The digest of the grants file this session was opened from. The session is + /// valid only while the file reads the same. + grants_stamp: String, + /// The policy stamp this session last loaded its policy at: what + /// `state/policy.stamp` read then. Shared with the session's views. + policy_seen: Arc>, /// How many decision contexts this session has kept. contexts_kept: Arc, /// The tools this session may call, when the launcher hosts them. @@ -430,7 +459,9 @@ impl Memory { /// /// A missing or bad grants file, or a store, index, or ledger that will not open. pub fn open(root: &Path) -> Result { - let grants = Grants::load(&root.join(GRANTS_FILE))?; + let grants_text = std::fs::read_to_string(root.join(GRANTS_FILE)).map_err(io(root))?; + let grants = Grants::parse(&grants_text)?; + let policy_seen = policy_stamp(root); std::fs::create_dir_all(root.join("state")).map_err(io(root))?; let store = Store::open_dir(root)?; let index = Index::open(&root.join(INDEX_FILE))?; @@ -467,7 +498,9 @@ impl Memory { held: Arc::default(), superseded: Arc::new(AtomicUsize::new(0)), claims: Arc::default(), - scope: OnceLock::new(), + scope: Mutex::new(None), + grants_stamp: dm_core::digest(&grants_text), + policy_seen: Arc::new(Mutex::new(policy_seen)), contexts_kept: Arc::new(AtomicUsize::new(0)), tools: OnceLock::new(), playback: OnceLock::new(), @@ -494,6 +527,7 @@ impl Memory { /// When the policy fails to give the read scope. pub fn for_request(&self) -> Result { let scope = self.session_filter()?; + let version = self.policy().version().to_string(); let mut session = self.grants.session(); session.handling = session.handling.min(scope.handling); Ok(Memory { @@ -516,7 +550,9 @@ impl Memory { held: Arc::clone(&self.held), superseded: Arc::clone(&self.superseded), claims: Arc::clone(&self.claims), - scope: OnceLock::from(scope), + scope: Mutex::new(Some((version, scope))), + grants_stamp: self.grants_stamp.clone(), + policy_seen: Arc::clone(&self.policy_seen), contexts_kept: Arc::clone(&self.contexts_kept), tools: self.tools.clone(), playback: self.playback.clone(), @@ -530,8 +566,12 @@ impl Memory { /// away and never add them. The session's handling is capped at it as well. The /// decision is logged like any other. fn session_filter(&self) -> Result { - if let Some(f) = self.scope.get() { - return Ok(f.clone()); + self.fresh()?; + let policy = self.policy(); + if let Some((version, f)) = lock(&self.scope).as_ref() { + if version == policy.version() { + return Ok(f.clone()); + } } let input = Input { action: Action::Read, @@ -546,13 +586,14 @@ impl Memory { spend: None, fetch: None, }; - let f = self - .policy() + let f = policy .read_scope(&input)? .narrow(&self.grants.read_filter()); { + // from the grants each time, so a filter that widens again widens the + // session's handling with it let mut s = lock(&self.session); - s.handling = s.handling.min(f.handling); + s.handling = self.grants.session().handling.min(f.handling); } self.log_decision( &input, @@ -566,13 +607,72 @@ impl Memory { f.handling, f.min_tier ), - policy: self.policy().version().to_string(), + policy: policy.version().to_string(), cites: Vec::new(), denials: Vec::new(), advisories: Vec::new(), }, ); - Ok(self.scope.get_or_init(|| f).clone()) + // what a pack pushed under another filter is no note for this one + if lock(&self.scope).is_some() { + *lock(&self.cache) = recall::Cache::default(); + } + *lock(&self.scope) = Some((policy.version().to_string(), f.clone())); + Ok(f) + } + + /// Show that this session's authority is current, before anything is released + /// under it: the grants file reads as it did when the session opened, and the + /// policy is the one in force. A session whose grants changed stops, because a + /// request made after a revocation cannot use the old grant; a new session starts + /// under the grants as they are. A policy another session changed is loaded here, + /// and the read filter follows it. When neither can be shown, the answer is no. + /// + /// # Errors + /// + /// [`Error::Stale`] when the grants changed or cannot be read, or the policy + /// cannot be loaded again. + pub fn fresh(&self) -> Result<(), Error> { + let now = std::fs::read_to_string(self.root.join(GRANTS_FILE)).map_err(|e| { + Error::Stale(format!( + "the grants cannot be read ({e}), so this session's access cannot be shown" + )) + })?; + if dm_core::digest(&now) != self.grants_stamp { + return Err(Error::Stale( + "the grants changed after this session started. A session keeps no grant \ + that was changed under it: start a new session" + .into(), + )); + } + let stamp = policy_stamp(&self.root); + let mut seen = lock(&self.policy_seen); + if *seen != stamp { + self.load_policy_again().map_err(|e| { + Error::Stale(format!( + "the policy changed and cannot be loaded again ({e}), so this \ + session's access cannot be shown" + )) + })?; + *seen = stamp; + } + Ok(()) + } + + /// Compile the policy from the record as it is now, through an index handle of its + /// own: the caller may hold this session's. + fn load_policy_again(&self) -> Result<(), Error> { + let index = Index::open_reader(&self.root.join(INDEX_FILE))?; + // an index that cannot be read shows no rules, which is not the same as none + index.list(&self.grants.read_filter(), Some(EntryType::Policy), 1)?; + let entries = policy_entries(&index, &self.root, &self.grants); + let p = Policy::load(&self.root, &self.grants)?.with_record( + &self.grants, + entries, + OffsetDateTime::now_utc(), + )?; + *self.policy.write().unwrap_or_else(PoisonError::into_inner) = std::sync::Arc::new(p); + Ok(()) } /// The task this session runs on: a launcher sets it before the harness starts, @@ -612,7 +712,16 @@ impl Memory { entries, OffsetDateTime::now_utc(), )?; + let version = p.version().to_string(); *self.policy.write().unwrap_or_else(PoisonError::into_inner) = std::sync::Arc::new(p); + // every other session learns from the stamp that the policy is another now + let stamp = format!( + "{version} {}", + OffsetDateTime::now_utc().unix_timestamp_nanos() + ); + let path = self.root.join(POLICY_STAMP_FILE); + std::fs::write(&path, &stamp).map_err(io(&path))?; + *lock(&self.policy_seen) = policy_stamp(&self.root); Ok(()) } @@ -627,6 +736,7 @@ impl Memory { /// Ask the policy, and log the decision under the record's state: every allow and /// deny, with the policy version that produced it, as the paper has decision logs. fn decide(&self, input: &Input) -> Result { + self.fresh()?; let mut input = input.clone(); input.held_task = self.held_task(); let mut d = self.policy().decide(&input)?; @@ -692,6 +802,7 @@ impl Memory { }), fetch: None, }; + self.fresh()?; let mut d = self.policy().decide(&input)?; if !d.allow { self.log_decision(&input, &d); @@ -771,10 +882,18 @@ impl Memory { self.author = author.into(); self.reads.set_author(&self.author); // the read scope is the principal's; a new author is asked afresh - self.scope = OnceLock::new(); + self.scope = Mutex::new(None); self } + /// Take over the claims another session of the same holder made: for a session + /// opened in the place of one that stopped, so the tasks it held are still its own + /// to complete or release. + pub fn adopt_claims(&self, from: &Memory) { + let theirs = lock(&from.claims).clone(); + lock(&self.claims).extend(theirs); + } + /// The capabilities this session's manifest carries, matched against task /// requirements on claim. #[must_use] @@ -1106,7 +1225,11 @@ impl Memory { body: req.body, }; check_write_size(entry.render().len())?; - self.gate(&entry).await?; + let replaced = self.gate(&entry).await?; + // a policy entry written, or one replaced by an entry of any kind, changes what + // is in force + let policy_changed = entry.front.entry_type == EntryType::Policy + || replaced.is_some_and(|old| old.front.entry_type == EntryType::Policy); self.put_logged(&entry).await?; self.index_put(&entry).await?; if let Some(old) = &entry.front.supersedes { @@ -1124,7 +1247,7 @@ impl Memory { entry.front.supersedes.as_deref(), ); self.refresh_projection().await; - if entry.front.entry_type == EntryType::Policy { + if policy_changed { self.reload_policy().await?; } self.written.0.fetch_add(1, Ordering::Relaxed); @@ -1923,6 +2046,9 @@ impl Memory { self.put_logged(&entry).await?; self.index_put(&entry).await?; self.refresh_projection().await; + if entry.front.entry_type == EntryType::Policy || entry.front.supersedes.is_some() { + self.reload_policy().await?; + } Some(Written { id: entry.front.id.clone(), classification: entry.front.classification, @@ -3468,6 +3594,7 @@ impl Memory { src: &Path, into_domain: Option<&str>, ) -> Result { + self.fresh()?; let domain = self.resolve_domain(into_domain)?; let default = self .grants @@ -3508,6 +3635,8 @@ impl Memory { .await?; self.restore_promotions(&imported.promotions).await?; self.refresh_projection().await; + // an import can land a rule or an exception, or replace one + self.reload_policy().await?; Ok(imported) } @@ -3536,6 +3665,9 @@ impl Memory { let entries = self.store.all().await?; let n = self.index.lock().await.rebuild(entries)?; let m = self.ledger.lock().await.rebuild().await?; + // the policy is compiled from the index: a rebuilt index can hold rules the old + // one had lost, and every open session must decide under them + self.reload_policy().await?; Ok((n, m)) } } @@ -3948,6 +4080,245 @@ mod tests { ); } + /// The built-in policy with agents narrowed to public entries. + fn narrowed_policy() -> String { + dm_policy::decide::BUILTIN.replace( + "read_filter := base_filter\n", + "read_filter := object.union(base_filter, {\"handling\": \"public\"}) if class == \"agent\"\n\nread_filter := base_filter if class != \"agent\"\n", + ) + } + + fn is_stale(r: &Result) -> bool { + matches!(r, Err(Error::Stale(_))) + } + + /// A session keeps no grant that was changed under it: after a change to the + /// grants file, every read, write and claim of the session is refused, and a new + /// session works under the grants as they are. + #[tokio::test] + async fn a_session_stops_when_its_grants_change() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + owner.write(req("Backoff is exponential.\n")).await.unwrap(); + let t = filed(&owner, tspec("Tidy", 0, None, &[], &[])).await; + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); + assert_eq!(agent.search("backoff", None, 10).await.unwrap().len(), 1); + let view = agent.for_request().unwrap(); + let path = dir.path().join(GRANTS_FILE); + let before = std::fs::read_to_string(&path).unwrap(); + std::fs::write( + &path, + format!("{before}\n[spend.gateway]\nsession_tokens = 400\n"), + ) + .unwrap(); + assert!(is_stale(&agent.search("backoff", None, 10).await)); + assert!(is_stale(&agent.t0(None).await)); + assert!(is_stale(&agent.list(None, None, 10).await)); + assert!(is_stale(&agent.write(req("A late note.\n")).await)); + assert!(is_stale(&agent.claim(&t.id, Duration::minutes(5)).await)); + assert!(is_stale(&agent.check_spend("gateway", 0, 0))); + assert!( + is_stale(&view.search("backoff", None, 10).await), + "a view of it too" + ); + assert!(is_stale(&agent.for_request())); + let fresh = Memory::open(dir.path()).unwrap().with_author("agent:a"); + assert_eq!(fresh.search("backoff", None, 10).await.unwrap().len(), 1); + // a grants file that cannot be read shows nothing, so the answer is no + std::fs::remove_file(&path).unwrap(); + assert!(is_stale(&fresh.search("backoff", None, 10).await)); + // the file as it was is the authority the first session started with + std::fs::write(&path, before).unwrap(); + assert_eq!(agent.search("backoff", None, 10).await.unwrap().len(), 1); + } + + /// A rule that another session puts in force reaches a session that is already + /// open: its next decision is made under the policy as it is now. + #[tokio::test] + async fn an_open_session_decides_under_a_rule_verified_after_it_started() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let d1 = owner + .write(WriteRequest { + entry_type: EntryType::Decision, + ..req("Use SQLite on a laptop.\n") + }) + .await + .unwrap() + .id; + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); + let supersede = |of: &str| WriteRequest { + entry_type: EntryType::Decision, + supersedes: Some(of.to_string()), + ..req("Use something else on a laptop.\n") + }; + let d2 = agent.write(supersede(&d1)).await.unwrap().id; + let rule = policy_entry(&owner, &rule_text("pol-L2-070", 2, "")).await; + let curator = Memory::open(dir.path()) + .unwrap() + .with_author("agent:curator"); + verify(&owner, &curator, &rule).await; + let refused = agent.write(supersede(&d2)).await.unwrap_err().to_string(); + assert!(refused.contains("pol-L2-070"), "{refused}"); + } + + /// The policy is compiled from the index. An index that lost its entries gives a + /// policy with no rule; the rebuild brings the rules back for the sessions that are + /// open, not only for the next one. + #[tokio::test] + async fn a_rebuilt_index_brings_its_rules_to_the_open_sessions() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let d1 = owner + .write(WriteRequest { + entry_type: EntryType::Decision, + ..req("Use SQLite on a laptop.\n") + }) + .await + .unwrap() + .id; + let rule = policy_entry(&owner, &rule_text("pol-L2-071", 2, "")).await; + let curator = Memory::open(dir.path()) + .unwrap() + .with_author("agent:curator"); + verify(&owner, &curator, &rule).await; + owner.index.lock().await.rebuild(Vec::new()).unwrap(); + owner.reload_policy().await.unwrap(); + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); + assert!(agent.policy().entries().is_empty()); + owner.reindex().await.unwrap(); + let refused = agent + .write(WriteRequest { + entry_type: EntryType::Decision, + supersedes: Some(d1), + ..req("Use something else on a laptop.\n") + }) + .await + .unwrap_err() + .to_string(); + assert!(refused.contains("pol-L2-071"), "{refused}"); + } + + /// A change to the root's own policy reaches an open session's read filter: its + /// next search is made under the narrower filter, and under the wider one again + /// when the file is gone. + #[tokio::test] + async fn an_open_session_reads_under_the_root_s_policy_as_it_is_now() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + owner + .write(WriteRequest { + classification: Some(Label::Public), + ..req("Backoff is exponential, public.\n") + }) + .await + .unwrap(); + let internal = owner + .write(req("Backoff is exponential, internal.\n")) + .await + .unwrap() + .id; + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); + assert_eq!(agent.search("backoff", None, 10).await.unwrap().len(), 2); + std::fs::create_dir_all(dir.path().join("policy")).unwrap(); + std::fs::write(dir.path().join("policy/local.rego"), narrowed_policy()).unwrap(); + assert_eq!(agent.search("backoff", None, 10).await.unwrap().len(), 1); + assert!(agent.get(&internal).await.is_err()); + // a policy that will not compile shows nothing, so the answer is no + std::fs::write( + dir.path().join("policy/local.rego"), + "package amem\nnot rego {", + ) + .unwrap(); + assert!(is_stale(&agent.search("backoff", None, 10).await)); + std::fs::remove_dir_all(dir.path().join("policy")).unwrap(); + assert_eq!(agent.search("backoff", None, 10).await.unwrap().len(), 2); + // the session's handling widened with the filter: the entry is read by id again + agent.get(&internal).await.unwrap(); + } + + /// A root's policy directory that cannot be read is no reason to fall back to the + /// built-in policy: the session is refused. + #[cfg(unix)] + #[tokio::test] + async fn a_root_policy_that_cannot_be_read_refuses_the_session() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + owner.write(req("Backoff is exponential.\n")).await.unwrap(); + let policy = dir.path().join("policy"); + std::fs::create_dir_all(&policy).unwrap(); + std::fs::write(policy.join("local.rego"), narrowed_policy()).unwrap(); + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); + assert!(agent.search("backoff", None, 10).await.unwrap().is_empty()); + std::fs::set_permissions(&policy, std::fs::Permissions::from_mode(0o000)).unwrap(); + // a process that reads every directory whatever its mode cannot show this + let blind = std::fs::read_dir(&policy).is_err(); + let after = agent.search("backoff", None, 10).await; + std::fs::set_permissions(&policy, std::fs::Permissions::from_mode(0o755)).unwrap(); + if blind { + assert!(is_stale(&after), "{:?}", after.map(|h| h.len())); + } + } + + /// A stopped session imports nothing. An import, and an entry of any kind that + /// replaces a rule, change what is in force for the sessions that are open. + #[tokio::test] + async fn an_import_and_a_replaced_rule_reach_the_open_sessions() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let d1 = owner + .write(WriteRequest { + entry_type: EntryType::Decision, + ..req("Use SQLite on a laptop.\n") + }) + .await + .unwrap() + .id; + let curator = Memory::open(dir.path()) + .unwrap() + .with_author("agent:curator"); + let rule = policy_entry(&owner, &rule_text("pol-L2-071", 2, "")).await; + verify(&owner, &curator, &rule).await; + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); + let supersede = |of: &str| WriteRequest { + entry_type: EntryType::Decision, + supersedes: Some(of.to_string()), + ..req("Use something else on a laptop.\n") + }; + assert!( + agent.write(supersede(&d1)).await.is_err(), + "the rule refuses" + ); + // the owner replaces the rule with a note, which is no policy entry + let stamp = || std::fs::read_to_string(dir.path().join(POLICY_STAMP_FILE)).unwrap(); + let before = stamp(); + let owner = Memory::open(dir.path()).unwrap(); + owner + .write(WriteRequest { + supersedes: Some(rule.clone()), + ..req("The rule is withdrawn.\n") + }) + .await + .unwrap(); + assert_ne!(stamp(), before, "the stamp says the policy changed"); + agent.write(supersede(&d1)).await.unwrap(); + // an import stamps too, and a stopped session imports nothing + let bundle = dir.path().join("out"); + owner.export(None, Label::Internal, &bundle).await.unwrap(); + let before = stamp(); + owner.import(&bundle, None).await.unwrap(); + assert_ne!(stamp(), before); + let grants = dir.path().join(GRANTS_FILE); + let text = std::fs::read_to_string(&grants).unwrap(); + std::fs::write( + &grants, + format!("{text}\n[spend.gateway]\nsession_tokens = 4\n"), + ) + .unwrap(); + assert!(is_stale(&owner.import(&bundle, None).await)); + } + #[tokio::test] async fn a_roots_policy_narrows_what_an_agent_reads_inside_the_query() { let dir = tempfile::tempdir().unwrap(); @@ -7221,7 +7592,6 @@ mod tests { async fn the_lint_finds_what_the_lifecycle_asks_of_a_rule() { let dir = tempfile::tempdir().unwrap(); let owner = mem(dir.path()); - let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); let d1 = owner .write(WriteRequest { entry_type: EntryType::Decision, @@ -7239,6 +7609,7 @@ mod tests { }; std::fs::write(&grants, ceiling).unwrap(); let owner = Memory::open(dir.path()).unwrap(); + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); let first = policy_entry(&owner, &rule_text("pol-L2-001", 2, "")).await; verify(&owner, &agent, &first).await; let late = rule_text("pol-L2-002", 2, "review_by = \"2020-01-01\"\n") diff --git a/crates/server/src/log.rs b/crates/server/src/log.rs index 475bc963..82aeef92 100644 --- a/crates/server/src/log.rs +++ b/crates/server/src/log.rs @@ -316,6 +316,8 @@ impl Memory { entries.extend(unnamed); self.index.lock().await.rebuild_domain(&domain, entries)?; out.ledger_events = self.ledger.lock().await.rebuild().await?; + // the policy is compiled from the index, which this changed + self.reload_policy().await?; Ok(out) } } diff --git a/crates/server/src/mcp.rs b/crates/server/src/mcp.rs index c7cc5a7e..eb4479fe 100644 --- a/crates/server/src/mcp.rs +++ b/crates/server/src/mcp.rs @@ -34,6 +34,7 @@ const MAX_LIMIT: usize = 100; fn mcp_err(e: Error) -> McpError { match e { Error::Denied(_) + | Error::Stale(_) | Error::NotFound(_) | Error::Gone(_) | Error::Moved { .. } @@ -1857,6 +1858,10 @@ mod tests { assert_eq!(invalid.code, also.code); assert_eq!(invalid.code, ledger.code); assert_ne!(invalid.code, internal.code); + // a stopped session is told so in words it can act on, as a refusal is + let stopped = mcp_err(Error::Stale("the grants changed".into())); + assert_eq!(stopped.code, invalid.code); + assert!(stopped.message.contains("stopped: the grants changed")); assert_eq!(t0_uri("dom-a"), "memory://t0/dom-a"); } diff --git a/crates/server/src/recall.rs b/crates/server/src/recall.rs index 9d693be8..3b50069b 100644 --- a/crates/server/src/recall.rs +++ b/crates/server/src/recall.rs @@ -664,6 +664,7 @@ impl Memory { return self.recall_owed(question, domain, depth, budget).await; }; let deadline = Instant::now() + budget.wall; + let read_under = self.filter_for(domain)?; let mut pack = self .pack(&[question.to_string()], domain, Depth::Sections, budget) .await?; @@ -690,6 +691,15 @@ impl Memory { Err(_) => break, } } + // the deputy took time: the brief is released only if what this session may + // read is still what the pack was read under + if self.filter_for(domain)? != read_under { + return Err(Error::Stale( + "what this session may read changed while the brief was written, and the \ + brief is withheld: ask again" + .into(), + )); + } let in_pack: HashSet = pack.ids().into_iter().collect(); let mut citations: Vec = answer .citations @@ -1068,6 +1078,65 @@ mod tests { } } + /// A deputy during whose answer the root's policy narrows agents to public entries. + struct NarrowsMeanwhile { + root: std::path::PathBuf, + brief: String, + } + + impl Deputy for NarrowsMeanwhile { + fn model(&self) -> String { + "scripted-1".into() + } + + fn complete<'a>( + &'a self, + _prompt: &'a str, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + let text = dm_policy::decide::BUILTIN.replace( + "read_filter := base_filter\n", + "read_filter := object.union(base_filter, {\"handling\": \"public\"}) if class == \"agent\"\n\nread_filter := base_filter if class != \"agent\"\n", + ); + std::fs::create_dir_all(self.root.join("policy")).map_err(|e| e.to_string())?; + std::fs::write(self.root.join("policy/local.rego"), text) + .map_err(|e| e.to_string())?; + Ok(self.brief.clone()) + }) + } + } + + /// A brief written from what the session could read is withheld when the session + /// can no longer read it by the time the brief is ready. + #[tokio::test] + async fn a_brief_is_withheld_when_the_read_filter_changed_while_it_was_written() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let retry = write( + &owner, + "Retry policy", + "Operation R retries with backoff.\n", + vec![], + ) + .await; + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); + let deputy: Arc = Arc::new(NarrowsMeanwhile { + root: dir.path().to_path_buf(), + brief: format!( + "{{\"brief\": \"R retries [{retry}].\", \"citations\": [\"{retry}\"], \"missing\": []}}" + ), + }); + agent.set_deputy(&deputy); + let r = agent + .recall("how does R retry", None, None, Budget::default()) + .await; + assert!( + matches!(r, Err(Error::Stale(_))), + "{:?}", + r.map(|r| r.brief) + ); + } + #[tokio::test] async fn with_a_deputy_recall_synthesizes_takes_a_second_hop_and_returns_summaries() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/server/src/sources.rs b/crates/server/src/sources.rs index dc0c0d0d..9e2a0f08 100644 --- a/crates/server/src/sources.rs +++ b/crates/server/src/sources.rs @@ -218,6 +218,8 @@ impl Memory { /// [`Error::Tool`] when the tool fails or a request it needed was refused, with /// what it said withheld when the scan finds anything in that. pub async fn use_tool(self: &Arc, tool: &str, args: Value) -> Result { + // a stored source is released under the same check as a search + self.fresh()?; self.definition_allows(tool).await?; let (grant, label) = self.tool_grant(tool)?; let id = source_id(tool, &args); diff --git a/crates/web/src/lib.rs b/crates/web/src/lib.rs index 6f5aa7c4..be3dfb51 100644 --- a/crates/web/src/lib.rs +++ b/crates/web/src/lib.rs @@ -215,6 +215,10 @@ macro_rules! session { ($app:expr) => { match $app.session() { Ok(m) => m, + // a session that stopped refuses; that is the client's to know, not a fault + Err(e @ dm_server::Error::Stale(_)) => { + return (StatusCode::FORBIDDEN, e.to_string()).into_response(); + } Err(e) => { return (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(); } @@ -227,8 +231,9 @@ struct App { config: Config, /// The board's session for its life: the handles, the session id for the read and /// write logs, and the read scope. Each request gets a view of it with a label of - /// its own ([`App::session`]). - memory: Arc, + /// its own ([`App::session`]). When the grants change, that session stops, as + /// every session does, and the board opens a new one ([`App::memory`]). + memory: Arc>>, sessions: Arc>>>, /// Bundle exports run one at a time: each replaces a directory by renaming. exports: Arc>, @@ -502,7 +507,35 @@ impl App { /// only: a listing does not classify what a later request files or writes, as one /// long-lived label would. fn session(&self) -> Result { - self.memory.for_request() + self.memory().for_request() + } + + /// The board's session, under the grants as they are now. A session whose grants + /// changed has stopped; the board is the owner's, so it opens a new one in its + /// place. When a new one cannot be opened, the stopped one is given and refuses. + fn memory(&self) -> Arc { + let now = Arc::clone( + &self + .memory + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ); + if !matches!(now.fresh(), Err(dm_server::Error::Stale(_))) { + return now; + } + match open_board(&self.config.root) { + Ok(new) => { + // the tasks the board held are still the owner's to complete + new.adopt_claims(&now); + let new = Arc::new(new); + *self + .memory + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Arc::clone(&new); + new + } + Err(_) => now, + } } fn start(&self, req: &NewSession) -> Result, Error> { @@ -919,7 +952,7 @@ async fn board() -> Html<&'static str> { fn task_error(e: &dm_server::Error) -> Response { let status = match e { dm_server::Error::NotFound(_) => StatusCode::NOT_FOUND, - dm_server::Error::Denied(_) => StatusCode::FORBIDDEN, + dm_server::Error::Denied(_) | dm_server::Error::Stale(_) => StatusCode::FORBIDDEN, dm_server::Error::Ledger(_) => StatusCode::CONFLICT, _ => StatusCode::INTERNAL_SERVER_ERROR, }; @@ -1124,8 +1157,8 @@ async fn reader_asset(Path(name): Path) -> Response { } /// The domains the record grants, for the reader's picker. -async fn domains(State(app): State) -> Json> { - Json(app.memory.domains()) +async fn domains(State(app): State) -> Response { + Json(session!(app).domains()).into_response() } /// A bundle file name: letters, digits, dot, underscore and hyphen, not starting with a @@ -1159,13 +1192,17 @@ fn admit_bundle_request( if !plain_file_name(file) { return Err((StatusCode::NOT_FOUND, "no such file".into())); } - if !app.memory.domains().iter().any(|d| d == domain) { + let memory = app.memory(); + if let Err(stopped) = memory.fresh() { + return Err((StatusCode::FORBIDDEN, stopped.to_string())); + } + if !memory.domains().iter().any(|d| d == domain) { return Err(( StatusCode::FORBIDDEN, format!("domain {domain} is not granted"), )); } - let handling = app.memory.grants().handling; + let handling = memory.grants().handling; if label > handling { return Err(( StatusCode::FORBIDDEN, @@ -1181,6 +1218,7 @@ fn export_refusal(e: dm_server::Error) -> Refusal { let status = match e { dm_server::Error::Denied(_) | dm_server::Error::NotFound(_) + | dm_server::Error::Stale(_) | dm_server::Error::Bundle(dm_server::bundle::Error::Invalid(_)) => StatusCode::FORBIDDEN, _ => StatusCode::INTERNAL_SERVER_ERROR, }; @@ -1273,15 +1311,19 @@ async fn bundle_file( } } +/// The board's session on a record. The browser is the owner, a person: a task +/// addressed to a person, one that requires `human`, is one the board may claim. +fn open_board(root: &std::path::Path) -> Result { + Ok(Memory::open(root)?.with_manifest(vec!["human".into()])) +} + /// The router, for serving or for tests. /// /// # Errors /// /// When the record cannot be opened. pub fn router(config: Config) -> Result { - // the browser is the owner, a person: a task addressed to a person, one that - // requires `human`, is one the board may claim - let memory = Arc::new(Memory::open(&config.root)?.with_manifest(vec!["human".into()])); + let memory = Arc::new(std::sync::RwLock::new(Arc::new(open_board(&config.root)?))); let access = config.access.clone(); let app = App { config, @@ -2223,6 +2265,60 @@ mod tests { ); } + /// The board's session stops when the grants change, as every session does, and + /// the board opens a new one: it answers under the grants as they are now. + #[tokio::test] + async fn the_board_opens_a_new_session_when_the_grants_change() { + let dir = tempfile::tempdir().unwrap(); + let (addr, _h) = start(dir.path(), &["true"]).await; + let (status, _) = get_json(addr, "/api/tasks").await; + assert_eq!(status, 200); + let grants = dir.path().join("memory/grants.toml"); + let text = std::fs::read_to_string(&grants).unwrap(); + let more = format!("{text}\n[spend.gateway]\nsession_tokens = 400\n"); + std::fs::write(&grants, more).unwrap(); + let (status, tasks) = get_json(addr, "/api/tasks").await; + assert_eq!(status, 200, "the board reads under a new session: {tasks}"); + let (status, domains) = get_json(addr, "/api/domains").await; + assert_eq!((status, domains), (200, serde_json::json!(["dom-me"]))); + } + + /// When the grants cannot be read, no new session opens, and the stopped one serves + /// nothing: each request is refused as the client's, with the reason. + #[tokio::test] + async fn a_board_whose_session_stays_stopped_refuses() { + let dir = tempfile::tempdir().unwrap(); + let (addr, _h) = start(dir.path(), &["true"]).await; + let m = Memory::open(&dir.path().join("memory")).unwrap(); + let w = m + .write(dm_server::WriteRequest { + body: "Operation R retries with backoff.\n".into(), + ..dm_server::WriteRequest::default() + }) + .await + .unwrap(); + // exported once, so the entry's file is there to serve + let entry = format!("/bundle/dom-me/internal/{}.md", w.id); + let manifest = "/bundle/dom-me/internal/manifest.yaml"; + for path in ["/api/domains", manifest, entry.as_str()] { + let (status, body) = http(addr, "GET", path, None).await; + assert_eq!(status, 200, "{path}: {body}"); + } + let grants = dir.path().join("memory/grants.toml"); + std::fs::write(&grants, "this is = not [ grants").unwrap(); + for path in [ + "/api/tasks", + "/api/domains", + "/api/entries", + manifest, + &entry, + ] { + let (status, body) = http(addr, "GET", path, None).await; + assert_eq!(status, 403, "{path}: {body}"); + assert!(body.contains("stopped"), "{path}: {body}"); + } + } + #[tokio::test] async fn the_board_files_admits_claims_and_completes_a_person_s_task() { let dir = tempfile::tempdir().unwrap(); diff --git a/docs/demo-narrative.md b/docs/demo-narrative.md index dea0d3ba..5f24bd43 100644 --- a/docs/demo-narrative.md +++ b/docs/demo-narrative.md @@ -220,8 +220,14 @@ example.com. After `agent:reader` has read internal entries, it can reach only t hosts, because a URL could leak what it read. The repeated fetch comes from the stored copy. Nothing leaves. -**3.12 Reads.** A policy limits agents to public entries. When `agent:a` searches, it -finds one hit. Your own search reaches the cap of 50 results. +**3.12 A grant withdrawn.** `agent:a` has an open session and searches. You lower the +handling of the record to public. The next search of that session is refused, and so is +its next write: the session stops. A new session of `agent:a` starts under the new +grants and finds no hit, because the record has no public entry. Then the demo puts the +grants back. + +**3.13 Reads.** A policy limits agents to public entries. When `agent:a` searches, it +finds no hit. Your own search reaches the cap of 50 results. ## Act 4 ยท The task fabric diff --git a/docs/terms.md b/docs/terms.md index 0d503c3a..e866d5b6 100644 --- a/docs/terms.md +++ b/docs/terms.md @@ -50,6 +50,8 @@ term here in the same commit as the first document that needs it. | grant | One table in the grants: `[domains.]`, `[destinations.]` or `[taps.]` | permission | | policy | The Rego rules that decide each action: write, read, erase, promote, file a task, spend and fetch. The root's `policy/*.rego` replaces the built-in rules | rules engine | | perimeter | The policy, the grants and the parts that enforce them, as one part of the design | | +| stopped (session) | A session whose grants changed after it started. Each read, decision, tool call and model call of the session is refused, and a new session starts under the grants as they are | revoked session, stale session | +| policy stamp | `state/policy.stamp` and the digest of the root's `policy/*.rego`: what a session compares to know that the policy changed | | | rule | One policy entry in the record: an id, a layer, the rule, its intent, an owner and, where a machine can check it, a module | | | status (of a rule) | Where a rule is in its life: `proposed` (in no decision), `advisory` (reports, refuses nothing), `enforced` (refuses) or `retired` | state (of a rule) | | ceiling | `[gates] max_enforced_rules`: the largest number of enforced rules in one domain that need no sign-off | cap (for rules) | diff --git a/local/README.md b/local/README.md index ac53a188..8a582c86 100644 --- a/local/README.md +++ b/local/README.md @@ -563,9 +563,10 @@ each advisory module by itself. When an advisory module fails, the decision log for that rule, and the decision does not change. When the module of an enforced rule loads and then fails in a decision, that decision fails. -A session loads the policy when it starts, and again when it changes a policy entry -itself. A change that a different session makes applies to the sessions that start after -the change. +A session loads the policy when it starts. Before each read and each decision, the +session compares the policy stamp with the stamp that it loaded. When a different session +changed a policy entry, or the root's `policy/*.rego` changed, the session loads the +policy again and computes its read filter again. See "Access freshness" below. `[gates] max_enforced_rules` in `grants.toml` is a ceiling on the enforced rules of one domain. There is no ceiling by default. `amem` counts the enforced rules of a domain from @@ -644,6 +645,57 @@ With `--file`, `amem policy lint` files a task for the owner for each finding, e for a module that is not yet verified. Thus an unused rule goes to the owner. `amem` looks at the full decision log, not at one review period. +## Access freshness + +A session is valid only while its authority is what the session started with. `amem` +does these checks before each read through the read filter, each decision, each spend +check, each call of a hosted tool, each import, and each request through the egress: + +- **The grants.** `amem` reads `grants.toml` and compares its digest with the digest + from the start of the session. If the file changed, the session stops: each call + gives `stopped: the grants changed after this session started`. A new session starts + under the grants as they are. If the file has its first text again, the session is + valid again. If `amem` cannot read the file, it refuses the call. +- **The policy.** `state/policy.stamp` changes when a session writes, promotes or erases + a policy entry, replaces a policy entry with an entry of any type, imports a bundle, + resets the rules, or builds the index again (`amem reindex`, `amem materialize`). + The stamp also includes the digest of the root's `policy/*.rego`. When the stamp is + not the stamp that the session loaded, the session loads the policy again. The read + filter of the session has the version of the policy that computed it, and `amem` + computes it again for a new version. If the policy does not load, `amem` refuses the + call. +- **Expiry.** `amem` compares the expiry of an exception with the clock at each + decision. An exception that was in force when the policy loaded lets nothing through + after its expiry. + +`recall` with a deputy does one more check. When the brief is ready, `amem` computes the +read filter again. If the filter is not the filter that the pack was read under, `amem` +withholds the brief. + +A tunnel that is open when its session stops does not stay open. The egress asks each +second if the session is valid, and closes the tunnel when it is not. + +The board is the owner's session. When its session stops, the board opens a new session +under the grants as they are. When the board cannot open one, for example because +`grants.toml` does not parse, it answers each request with status 403 and the reason. +It also refuses the files of a bundle that it exported before. + +The read filter of a session is computed again only for a new version of the policy. +The expiry of an exception does not make a new version. Thus a root policy whose read +filter reads the exceptions keeps its filter after an expiry, until the stamp changes. +The built-in read filter does not read the exceptions. An advisory rule also reports +with the exceptions as they were when the policy loaded. + +`amem` does not look for which grant changed: any change to `grants.toml` stops each +open session of that record. + +The check is before the read. A read that waits, for example on the index, does not +check again before it gives its answer. Only `recall` with a deputy does. + +Some commands of the owner and of the launcher do not do the check: for example +`amem transcript`, `amem tape`, `amem source`, `amem quarantine`, `amem taint`, +`amem policy lint` and `amem policy reset`. No memory tool is in this group. + ## The states of a task, and its claim: `amem task` A task has one of six states, with the names that the paper gives: diff --git a/scripts/demo/3-perimeter.sh b/scripts/demo/3-perimeter.sh index cd7bb899..3cf5d563 100755 --- a/scripts/demo/3-perimeter.sh +++ b/scripts/demo/3-perimeter.sh @@ -160,9 +160,39 @@ for line in sys.stdin: ' fi +say "A grant withdrawn mid-session" "A session is valid only while its grants read as they did when it started. The owner lowers the handling of the record while an agent's session is open. The session's next read is refused, and so is its next write: a request made after a revocation cannot use the old grant. A new session starts under the grants as they are." +# the grants go back however the step ends, a signal included; a copy that an +# interrupted run left is put back first, so it is never overwritten with lowered grants +restore_grants() { + if [ -f "$root/grants.toml.before-revoke" ]; then + mv "$root/grants.toml.before-revoke" "$root/grants.toml" + fi +} +restore_grants +trap restore_grants EXIT +trap 'restore_grants; trap - EXIT; exit 130' INT TERM HUP +cp "$root/grants.toml" "$root/grants.toml.before-revoke" +sed 's/^handling = .*/handling = "public"/' "$root/grants.toml" > "$root/grants.toml.lowered" +agent agent:a steps "[ + {\"tool\": \"search\", \"arguments\": {\"query\": \"decision\", \"limit\": 50}}, + {\"run\": \"mv '$root/grants.toml.lowered' '$root/grants.toml'\"}, + {\"tool\": \"search\", \"arguments\": {\"query\": \"decision\", \"limit\": 50}}, + {\"tool\": \"write\", \"arguments\": {\"title\": \"A late note\", \"body\": \"Written after the grant was withdrawn.\"}} +]" | python3 -c " +import json, sys +for line in sys.stdin: + d = json.loads(line) + if d['ok']: + print(f\" {d['tool']}: {len(d['result']['hits'])} hits, under the grants the session started with\") + else: + print(f\" {d['tool']}: {d['refused'][:150]}\") +" +echo " a new session, under the grants as they are now: $(agent agent:a search '{"query": "decision", "limit": 50}' | j "len(d['hits'])") hits" +mv "$root/grants.toml.before-revoke" "$root/grants.toml" + say "Reads are the policy's too" "A root's policy can narrow what agents read, here to public entries, and the narrowing is compiled into the search query itself, not applied after." mkdir -p "$root/policy" sed 's/^read_filter := base_filter$/read_filter := object.union(base_filter, {"handling": "public"}) if class == "agent"\n\nread_filter := base_filter if class != "agent"/' \ "$here/crates/policy/rego/amem.rego" > "$root/policy/local.rego" -echo " an agent searching 'decision': $(agent agent:a search '{"query": "decision", "limit": 50}' | j 'len(d)') hits; the owner: $(a search decision --limit 50 | j 'len(d)')" +echo " an agent searching 'decision': $(agent agent:a search '{"query": "decision", "limit": 50}' | j "len(d['hits'])") hits; the owner: $(a search decision --limit 50 | j 'len(d)')" rm -r "$root/policy" diff --git a/scripts/demo/mcp.py b/scripts/demo/mcp.py index d6bde4c8..7cd24f19 100755 --- a/scripts/demo/mcp.py +++ b/scripts/demo/mcp.py @@ -8,7 +8,9 @@ Prints the tool's JSON result, or `refused: ` and exits 1. A batch is several calls in one session, as a harness makes them; it prints the last result. `steps` is a batch that goes on past a refusal and prints one JSON line per call, `{"tool", "ok", -"result"}` or `{"tool", "ok": false, "refused"}`, for an act that shows each.""" +"result"}` or `{"tool", "ok": false, "refused"}`, for an act that shows each. In +`steps`, an item `{"run": ""}` is not a tool call: the command runs here +between two calls, for an act that changes something under an open session.""" import json import os import subprocess @@ -49,6 +51,9 @@ def outcome(r): if tool == "steps": for n, c in enumerate(calls, start=2): + if "run" in c: + subprocess.run(c["run"], shell=True, check=True) + continue send({"jsonrpc": "2.0", "id": n, "method": "tools/call", "params": {"name": c["tool"], "arguments": c["arguments"]}}) ok, text = outcome(recv())