diff --git a/crates/contextful-agent/src/http.rs b/crates/contextful-agent/src/http.rs index 27074f4e4..4ba956099 100644 --- a/crates/contextful-agent/src/http.rs +++ b/crates/contextful-agent/src/http.rs @@ -15,6 +15,7 @@ use crate::mcp::{build_identity, Caller, ReadRecord, Tools}; use contextful_context::read::Face; use contextful_core::ports::Clock; +use contextful_core::grant::{Action, TablePattern}; use contextful_core::read::{ReadError, Refusal}; use contextful_core::AuthorityError; use contextful_policy::enforce::refuse::payload; @@ -41,6 +42,10 @@ pub const MCP_PATH: &str = "/mcp"; /// The readiness route (`read.register.health`). pub const HEALTH_PATH: &str = "/health"; +/// The store-published workflow state and validated control claim routes. +pub const WORKFLOWS_PATH: &str = "/control/workflows"; +pub const APPLY_PATH: &str = "/control/apply"; + /// Largest request head the face reads; a head past it answers `431`. const REQUEST_HEAD_BYTES: usize = 64 * 1024; @@ -193,6 +198,7 @@ pub struct Admitting<'a, C> { pub struct HttpFace<'a, C> { tools: Tools<'a>, admitting: Admitting<'a, C>, + control: Option<&'a (dyn Fn(&HttpRequest) -> HttpResponse + Sync)>, ceiling: usize, in_flight: AtomicUsize, exchange: Option<&'a (dyn Fn(&HttpRequest) -> HttpResponse + Sync)>, @@ -243,7 +249,7 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> { let ceiling = ceiling(max_in_flight)?; audience(Some(admitting.audience))?; let tools = Tools::new(face, clock, record)?; - Ok(HttpFace { tools, admitting, ceiling, in_flight: AtomicUsize::new(0), exchange: None, exchange_unconfigured: false }) + Ok(HttpFace { tools, admitting, control: None, ceiling, in_flight: AtomicUsize::new(0), exchange: None, exchange_unconfigured: false }) } /// The binary's exchange route mints the reader credential without putting issuer @@ -254,6 +260,12 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> { self } + /// Attach the store's control API without adding tools to the closed read face. + pub fn with_control(mut self, control: &'a (dyn Fn(&HttpRequest) -> HttpResponse + Sync)) -> Self { + self.control = Some(control); + self + } + /// Accept connections on `listener` until it fails. Each accepted connection takes a /// slot before its thread starts and holds it until its answer is written, so at most /// the ceiling's count of connection threads run. One accepted with every slot held @@ -343,28 +355,68 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> { } ("/auth/exchange", "POST") if self.exchange.is_some() => self.exchange.expect("checked above")(request), ("/auth/exchange", _) if self.exchange.is_some() => HttpResponse::message(405, "`/auth/exchange` answers POST").with("Allow", "POST"), + (WORKFLOWS_PATH, "GET") | (APPLY_PATH, "POST") if self.control.is_some() => self.control(request), + (WORKFLOWS_PATH, _) if self.control.is_some() => HttpResponse::message(405, "`/control/workflows` answers GET").with("Allow", "GET"), + (APPLY_PATH, _) if self.control.is_some() => HttpResponse::message(405, "`/control/apply` answers POST").with("Allow", "POST"), (other, _) => HttpResponse::message(404, format!("no route `{other}`; the protocol endpoint is `{MCP_PATH}`")), } } /// Admit the request's credential, then answer its one message. fn message(&self, request: &HttpRequest) -> HttpResponse { + let (authority, revocation) = match self.admit(request) { + Ok(a) => a, + Err(response) => return response, + }; + let message: Value = match serde_json::from_slice(&request.body) { + Ok(m @ Value::Object(_)) => m, + Ok(_) => return HttpResponse::rpc_error(INVALID_REQUEST, "a request body holds one JSON-RPC message object"), + Err(e) => return HttpResponse::rpc_error(PARSE_ERROR, format!("the request body is not JSON: {e}")), + }; + let admission = Admission::new(self.tools.clock().now(), &revocation).expecting(self.admitting.audience); + let boundary = |a: &AdmittedAuthority| effect_boundary(a, &admission); + match self.tools.handle(Caller { authority: &authority, boundary: &boundary }, &message) { + Some(answer) => HttpResponse::json(200, &answer), + None => HttpResponse::empty(202), + } + } + + fn control(&self, request: &HttpRequest) -> HttpResponse { + let (authority, _) = match self.admit(request) { + Ok(a) => a, + Err(response) => return response, + }; + if !authority.grants().iter().any(|grant| grant.actions.contains(&Action::Admin) && grant.tables.contains(&TablePattern::All)) { + return HttpResponse::json(403, &json!({ "error": { "identifier": "ControlAdminGrantMissing" } })); + } + let revocation = match (self.admitting.revocation)() { + Ok(r) => r, + Err(why) => return HttpResponse::unavailable(format!("the revocation denylist is unreadable: {why}")), + }; + let boundary = Admission::new(self.tools.clock().now(), &revocation).expecting(self.admitting.audience); + if let Err(error) = effect_boundary(&authority, &boundary) { + return unadmitted(&error); + } + self.control.expect("the route is installed only with a control handler")(request) + } + + fn admit(&self, request: &HttpRequest) -> Result<(AdmittedAuthority, RevocationState), HttpResponse> { let Some(credential) = request.credential() else { let missing = ReadError::HttpCredentialMissing( "a request carries `Authorization: Bearer `, or `Authorization: DPoP ` with a `DPoP` proof from the credential's holder key".into(), ); - return HttpResponse::json(401, &payload(&Refusal::Read(missing))).with("WWW-Authenticate", CHALLENGE); + return Err(HttpResponse::json(401, &payload(&Refusal::Read(missing))).with("WWW-Authenticate", CHALLENGE)); }; let revocation = match (self.admitting.revocation)() { Ok(r) => r, - Err(why) => return HttpResponse::unavailable(format!("the revocation denylist is unreadable: {why}")), + Err(why) => return Err(HttpResponse::unavailable(format!("the revocation denylist is unreadable: {why}"))), }; let admission = Admission::new(self.tools.clock().now(), &revocation).expecting(self.admitting.audience); let covered = ProofRequest { method: &request.method, target: &request.target, body: &request.body }; let checkpoint = self.admitting.checkpoint; let keys = match checkpoint.keys() { Ok(k) => k, - Err(e) => return HttpResponse::unavailable(e.to_string()), + Err(e) => return Err(HttpResponse::unavailable(e.to_string())), }; let authority = verify_network(credential, &keys, &admission, |jkt| match request.header("DPoP") { Some(proof) => checkpoint.verify_request(jkt, proof, &covered), @@ -372,20 +424,10 @@ impl<'a, C: Clock + Sync> HttpFace<'a, C> { "the credential binds a holder key, and the request carries no `DPoP` proof".into(), ))), }); - let authority = match authority { - Ok(a) => a, - Err(ProofRefusal::NonceCacheFull) => return HttpResponse::unavailable(ProofRefusal::NonceCacheFull.to_string()), - Err(ProofRefusal::Refused(e)) => return unadmitted(&e), - }; - let message: Value = match serde_json::from_slice(&request.body) { - Ok(m @ Value::Object(_)) => m, - Ok(_) => return HttpResponse::rpc_error(INVALID_REQUEST, "a request body holds one JSON-RPC message object"), - Err(e) => return HttpResponse::rpc_error(PARSE_ERROR, format!("the request body is not JSON: {e}")), - }; - let boundary = |a: &AdmittedAuthority| effect_boundary(a, &admission); - match self.tools.handle(Caller { authority: &authority, boundary: &boundary }, &message) { - Some(answer) => HttpResponse::json(200, &answer), - None => HttpResponse::empty(202), + match authority { + Ok(a) => Ok((a, revocation)), + Err(ProofRefusal::NonceCacheFull) => Err(HttpResponse::unavailable(ProofRefusal::NonceCacheFull.to_string())), + Err(ProofRefusal::Refused(e)) => Err(unadmitted(&e)), } } } diff --git a/crates/contextful-agent/tests/integration/http.rs b/crates/contextful-agent/tests/integration/http.rs index d3874beee..e23c3ed76 100644 --- a/crates/contextful-agent/tests/integration/http.rs +++ b/crates/contextful-agent/tests/integration/http.rs @@ -101,10 +101,14 @@ fn holder(n: u8) -> SigningKey { /// A credential reading `tables` for `ttl` seconds, bound to `holder` when one is given. fn credential(signer: &SeedSigner, tables: &str, ttl: u64, holder: Option<&SigningKey>) -> String { + credential_with_action(signer, Action::Read, tables, ttl, holder) +} + +fn credential_with_action(signer: &SeedSigner, action: Action, tables: &str, ttl: u64, holder: Option<&SigningKey>) -> String { let policy = IssuancePolicy::parse(&format!("default_audience = \"{AUD}\"\nmax_lifetime_secs = 86400\n")).unwrap(); let subject = Subject { on_behalf_of: Some("user://dana@acme.example".into()), zone: Some("on-prem:hq".into()), ..Subject::default() }; let grant = Grant { - actions: vec![Action::Read], + actions: vec![action], tables: vec![TablePattern::parse(tables).unwrap()], tenant: None, aggregate: None, @@ -121,6 +125,33 @@ fn credential(signer: &SeedSigner, tables: &str, ttl: u64, holder: Option<&Signi mint(&plan, &MintClaims { confirmation, ..MintClaims::default() }, signer).unwrap() } +#[test] +fn control_apply_rechecks_revocation_before_dispatch() { + let f = fixture(); + let token = credential_with_action(&f.signer, Action::Admin, "*", 900, None); + let clock = FixedClock(at(NOW)); + let checks = std::sync::atomic::AtomicUsize::new(0); + let revocation = || { + if checks.fetch_add(1, Ordering::SeqCst) == 0 { Ok(RevocationState::default()) } + else { Err("revocation source changed".into()) } + }; + let admitting = Admitting { checkpoint: &f.checkpoint, audience: AUD, revocation: &revocation }; + let dispatched = AtomicBool::new(false); + let handler = |_: &HttpRequest| { + dispatched.store(true, Ordering::SeqCst); + contextful_agent::http::HttpResponse::json(200, &json!({ "applied": 2 })) + }; + let face = HttpFace::new(&f.face, &clock, &f.audit, admitting, Some(2)).unwrap().with_control(&handler); + let request = HttpRequest { + method: "POST".into(), target: "/control/apply".into(), + headers: vec![("Authorization".into(), format!("Bearer {token}"))], body: b"{}".to_vec(), + }; + let answer = face.answer(&request); + assert_eq!(answer.status, 503); + assert!(!dispatched.load(Ordering::SeqCst)); + assert_eq!(checks.load(Ordering::SeqCst), 2); +} + fn call(tool: &str, arguments: Value) -> Value { json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": tool, "arguments": arguments } }) } diff --git a/crates/contextful-cli/src/cadence.rs b/crates/contextful-cli/src/cadence.rs index 58d9f006c..bd5574f36 100644 --- a/crates/contextful-cli/src/cadence.rs +++ b/crates/contextful-cli/src/cadence.rs @@ -17,6 +17,7 @@ use crate::run::{boot_id, wire_at, ProjectArgs}; use anyhow::{bail, Context, Result}; use contextful_context::project::Project; use contextful_core::pipeline::declare::{collect, dependent_runs, ManifestFile, PipelineSpec}; +use contextful_core::pipeline::transform::TransformOp; use contextful_core::grant::Action; use contextful_core::ports::Clock; use contextful_core::run::derive::task::Tasks; @@ -36,7 +37,8 @@ use contextful_policy::verify::{effect_boundary, Admission, AdmittedAuthority}; use contextful_outbound::egress::{system, Outbound, Transport}; use serde::Serialize; use serde_json::json; -use std::collections::BTreeMap; +use serde_json::Value; +use std::collections::{BTreeMap, BTreeSet}; use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::io::Read; @@ -238,6 +240,53 @@ fn applied(snaps: &Source) -> Result<(Option, BTreeMap Result { + let text = std::fs::read_to_string(declaration)?; + let control = control_config(&text, project)?; + let (version, pipelines) = applied(&control.source)?; + let truncated = pipelines.len() > 1000; + let pipelines: Vec = pipelines.into_values().take(1000).collect(); + let selected: BTreeSet<&str> = pipelines.iter().map(|spec| spec.id.as_str()).collect(); + let store = contextful_context::Store::open(&project.dir, &project.name)?; + let states = contextful_sync::run_state::run_states(&store)?; + let mut runs: BTreeMap = BTreeMap::new(); + for state in states.into_values() { + for (id, mark) in state.runs { + if selected.contains(id.as_str()) && runs.get(&id).is_none_or(|earlier| mark.started_at > earlier.started_at) { + runs.insert(id, mark); + } + } + } + let wired = wire_at(project, &None)?; + for row in wired.engine.catalog.runs(None)? { + if selected.contains(row.pipeline_id.as_str()) && runs.get(&row.pipeline_id).is_none_or(|earlier| row.started_at > earlier.started_at) { + runs.insert(row.pipeline_id.clone(), contextful_sync::RunMark { + run_id: row.run_id, + table: row.table, + status: row.status, + started_at: row.started_at, + ended_at: row.ended_at, + }); + } + } + let pipelines: Vec = pipelines.into_iter().map(|spec| json!({ + "id": spec.id, + "schedule": spec.schedule, + "tables": spec.tables.iter().map(|table| table.name()).collect::>(), + "source": spec.source.name, + "after": spec.after, + "steps": spec.transforms.iter().map(|step| match step { + TransformOp::Select { .. } => "select", + TransformOp::Rename { .. } => "rename", + TransformOp::Cast { .. } => "cast", + TransformOp::Filter { .. } => "filter", + TransformOp::Extract { .. } => "extract", + }).collect::>(), + })).collect(); + Ok(json!({ "applied": version, "pipelines": pipelines, "runs": runs, "truncated": truncated, "declined": 0 })) +} + /// A locally applied descendant keeps precedence over an older bucket head after every /// receipt on the local path verifies under the same pinned issuer keys. fn local_descends(snaps: &SnapshotDir, current: u64, bucket: &ControlHead, project: &str, trusted: &[SignerKey]) -> Result { diff --git a/crates/contextful-cli/src/lib.rs b/crates/contextful-cli/src/lib.rs index 779bbbc23..dc64dce43 100644 --- a/crates/contextful-cli/src/lib.rs +++ b/crates/contextful-cli/src/lib.rs @@ -254,7 +254,7 @@ pub fn main_host(host: Host) { #[cfg(feature = "read-plane")] Cmd::Mcp(c) => mcp::run(c), #[cfg(feature = "read-plane")] - Cmd::Serve(c) => serve::run(c), + Cmd::Serve(c) => serve::run(c, &tasks), #[cfg(feature = "data-plane")] Cmd::Derive(c) => derive::run(c), #[cfg(feature = "read-plane")] diff --git a/crates/contextful-cli/src/serve.rs b/crates/contextful-cli/src/serve.rs index 526e6d560..2234b8cc6 100644 --- a/crates/contextful-cli/src/serve.rs +++ b/crates/contextful-cli/src/serve.rs @@ -7,6 +7,8 @@ //! network transport, which admits each one on its own credential. Every value is //! resolved before the listener binds, so a process that cannot serve binds nothing. +#[cfg(feature = "data-plane")] +use crate::admit::AdmitArgs; use crate::admit::{face, revocation_state, LedgerFile, LivePins, AUDIENCE_VAR, PUBKEY_VAR}; use crate::project::locate; use crate::root::root as project_root; @@ -18,12 +20,19 @@ use contextful_core::ports::{Clock, SigningPort}; use contextful_policy::exchange::answer as exchange_answer; use contextful_policy::issue::{SeedSigner, DEFAULT_SEED_PATH}; use contextful_policy::possession::ProofChecker; +#[cfg(feature = "data-plane")] +use contextful_agent::http::{HttpRequest, HttpResponse, APPLY_PATH, WORKFLOWS_PATH}; +#[cfg(feature = "data-plane")] +use contextful_core::surface::SurfaceError; +use contextful_core::run::derive::task::Tasks; use contextful_policy::audit::AuditLog; use contextful_policy::keyset::{KeyCheckpoint, StaticPins}; use contextful_policy::revoke::RevocationState; use std::net::TcpListener; use std::path::PathBuf; use std::sync::Arc; +#[cfg(feature = "data-plane")] +use serde_json::{json, Value}; /// The refusals of starting the network transport. `Display` begins with the identifier. #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] @@ -78,7 +87,9 @@ fn issuer_pins(flag: Option<&str>) -> Result { StaticPins::parse(text).map_err(|e| ServeError::IssuerKeyUnusable(format!("--public-key / {PUBKEY_VAR}: {e}"))) } -pub fn run(args: ServeArgs) -> Result<()> { +pub fn run(args: ServeArgs, tasks: &Tasks) -> Result<()> { + #[cfg(not(feature = "data-plane"))] + let _ = tasks; let clock = SystemClock; // The declarations are checked before anything opens (`read.register.serve-declaration`). let audience = audience(args.audience.as_deref()).map_err(anyhow::Error::msg)?; @@ -132,9 +143,43 @@ pub fn run(args: ServeArgs) -> Result<()> { let result = exchange_answer(exchange.as_ref(), &request.body, request.header("DPoP"), &proofs, issuance, &ctx); contextful_agent::http::HttpResponse::json(result.status, &result.body) }; + #[cfg(feature = "data-plane")] + let control_admit = AdmitArgs { public_key: args.public_key.clone(), audience: args.audience.clone(), denylist: args.denylist.clone(), keyset: args.keyset.clone(), holder_key: None }; + #[cfg(feature = "data-plane")] + let control = |request: &HttpRequest| -> HttpResponse { + let answer = match request.target.split('?').next().unwrap_or_default() { + WORKFLOWS_PATH => crate::cadence::published(&located.project, &located.declaration), + APPLY_PATH => { + let body: Value = match serde_json::from_slice(&request.body) { + Ok(Value::Object(body)) => Value::Object(body), + _ => return HttpResponse::json(400, &json!({ "error": { "identifier": "ControlRequestMalformed" } })), + }; + let Some(fields) = body.as_object() else { unreachable!() }; + if fields.len() > 1 || fields.keys().any(|field| field != "id") || fields.get("id").is_some_and(|id| !id.is_string()) { + return HttpResponse::json(400, &json!({ "error": { "identifier": "ControlRequestMalformed" } })); + } + let project = crate::run::ProjectArgs { project: Some(located.project.name.clone()), now: None }; + crate::cadence::apply(&project, Some(located.declaration.clone()), body["id"].as_str(), tasks, &control_admit, None) + .and_then(|()| crate::cadence::published(&located.project, &located.declaration)) + } + _ => unreachable!(), + }; + match answer { + Ok(state) => HttpResponse::json(200, &state), + Err(error) => { + let surface = error.chain().find_map(|part| part.downcast_ref::()); + let status = surface.map_or(503, SurfaceError::status); + let identifier = surface.map(|refusal| refusal.to_string().split(':').next().unwrap_or("ControlUnavailable").to_string()) + .unwrap_or_else(|| "ControlUnavailable".into()); + HttpResponse::json(status, &json!({ "error": { "identifier": identifier } })) + } + } + }; let http = HttpFace::new(&face, &clock, &audit, admitting, Some(ceiling)) .map_err(anyhow::Error::msg)? .with_exchange(&exchange_route, exchange.is_none()); + #[cfg(feature = "data-plane")] + let http = http.with_control(&control); let listener = TcpListener::bind(&args.http)?; eprintln!("listening on http://{}/mcp", listener.local_addr()?); http.serve(listener)?; diff --git a/crates/contextful-cli/tests/integration/derive.rs b/crates/contextful-cli/tests/integration/derive.rs index 40e9f59f6..1d2e891d7 100644 --- a/crates/contextful-cli/tests/integration/derive.rs +++ b/crates/contextful-cli/tests/integration/derive.rs @@ -175,7 +175,7 @@ fn the_verb_refuses_a_fetch_engine_and_runs_a_transcriber_over_one_file() { /// The embedding binary under `examples/host_derive.rs`, built on first request: the command /// line with the compiled `word-split` task registered. -fn host_binary() -> std::path::PathBuf { +pub(crate) fn host_binary() -> std::path::PathBuf { static BUILT: std::sync::Once = std::sync::Once::new(); BUILT.call_once(|| { let status = Command::new(env!("CARGO")).args(["build", "-q", "-p", "contextful-cli", "--example", "host_derive"]).status().unwrap(); @@ -195,7 +195,7 @@ fn ok(out: &Output) -> String { /// A `word-split` pipeline over `documents`: `words` retains versions, `stats` does not, and /// `units` holds the markers. -fn host_manifest(task: &str) -> String { +pub(crate) fn host_manifest(task: &str) -> String { format!( "[[pipeline]]\nid = \"split\"\ntables = [\n \ {{ name = \"words\", primary_key = [\"unit_ref\", \"derivation_key\", \"word_seq\"], retain_versions = true }},\n \ diff --git a/crates/contextful-cli/tests/integration/profile.rs b/crates/contextful-cli/tests/integration/profile.rs index 8d4727309..a26b4f60e 100644 --- a/crates/contextful-cli/tests/integration/profile.rs +++ b/crates/contextful-cli/tests/integration/profile.rs @@ -110,3 +110,51 @@ fn the_read_replica_reads_and_syncs_and_refuses_every_write_path() { } assert!(!dir.path().join(".contextful").join("context").join("research").exists(), "a refused subcommand created a store root"); } + +/// HTTP admission stays available on the read replica; control routes require the run path. +#[cfg(feature = "read-plane")] +#[test] +fn served_http_keeps_read_admission_and_limits_control_to_the_data_plane() { + use std::io::{BufRead, BufReader, Read, Write}; + use std::net::TcpStream; + use std::process::Stdio; + struct Listener(std::process::Child); + impl Drop for Listener { + fn drop(&mut self) { let _ = self.0.kill(); let _ = self.0.wait(); } + } + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("contextful.toml"), "authoring_posture = \"per_request\"\n").unwrap(); + let key = cf(dir.path(), &["token", "keygen", "--out", ".contextful/issuer.seed"]); + assert!(key.status.success(), "{}", String::from_utf8_lossy(&key.stderr)); + let public = String::from_utf8_lossy(&key.stdout).trim().to_owned(); + let mut child = Command::new(env!("CARGO_BIN_EXE_contextful")) + .args(["serve", "--http", "127.0.0.1:0", "--project", "research", "--audience", "contextful://profile-test", "--max-in-flight", "2", "--public-key", &public]) + .current_dir(dir.path()).stdout(Stdio::null()).stderr(Stdio::piped()).spawn().unwrap(); + let mut stderr = BufReader::new(child.stderr.take().unwrap()); + let _listener = Listener(child); + let address = loop { + let mut line = String::new(); + assert!(stderr.read_line(&mut line).unwrap() > 0, "HTTP profile face exits before binding"); + if let Some(address) = line.trim().strip_prefix("listening on http://").and_then(|value| value.strip_suffix("/mcp")) { + break address.to_owned(); + } + }; + let request = |method: &str, path: &str| { + let mut stream = TcpStream::connect(&address).unwrap(); + stream.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap(); + write!(stream, "{method} {path} HTTP/1.1\r\nHost: {address}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").unwrap(); + let mut answer = String::new(); + stream.read_to_string(&mut answer).unwrap(); + answer + }; + assert!(request("GET", "/health").starts_with("HTTP/1.1 200")); + let refused = request("POST", "/mcp"); + assert!(refused.starts_with("HTTP/1.1 401"), "{refused}"); + assert!(refused.to_ascii_lowercase().contains("www-authenticate:"), "{refused}"); + for (method, path) in [("GET", "/control/workflows"), ("POST", "/control/apply")] { + let answer = request(method, path); + let status = if cfg!(feature = "data-plane") { "401" } else { "404" }; + assert!(answer.starts_with(&format!("HTTP/1.1 {status}")), "{answer}"); + assert!(!answer.contains("filings-flow"), "{answer}"); + } +} diff --git a/crates/contextful-cli/tests/integration/serve.rs b/crates/contextful-cli/tests/integration/serve.rs index 01f436f66..799e3b167 100644 --- a/crates/contextful-cli/tests/integration/serve.rs +++ b/crates/contextful-cli/tests/integration/serve.rs @@ -59,7 +59,11 @@ impl Drop for Listener { /// Start the face; its address, reported on standard error. fn serve(dir: &Path, args: &[&str]) -> (Listener, String) { - let mut child = Command::new(env!("CARGO_BIN_EXE_contextful")) + serve_binary(Path::new(env!("CARGO_BIN_EXE_contextful")), dir, args) +} + +fn serve_binary(binary: &Path, dir: &Path, args: &[&str]) -> (Listener, String) { + let mut child = Command::new(binary) .args(args) .current_dir(dir) .env_remove("CONTEXTFUL_ISSUER_PUBKEY") @@ -80,6 +84,22 @@ fn serve(dir: &Path, args: &[&str]) -> (Listener, String) { panic!("the face never listened") } +#[test] +fn control_apply_uses_the_host_registered_task_set() { + let binary = super::derive::host_binary(); + let (dir, public) = project(); + let root = dir.path(); + std::fs::write(root.join("contextful.toml"), format!("authoring_posture = \"per_request\"\n{}", super::derive::host_manifest("word-split"))).unwrap(); + let imported = Command::new(&binary).args(["pipeline", "import", "--project", "research"]).current_dir(root).output().unwrap(); + stdout(&imported); + std::fs::write(root.join("contextful.toml"), format!("authoring_posture = \"per_request\"\n{}", super::derive::host_manifest("word-split").replace("[pipeline.source]", "schedule = \"every 1h\"\n[pipeline.source]"))).unwrap(); + let admin = stdout(&run(root, &["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", "user://dana@acme.example", "--zone", "on-prem:hq", "--action", "admin", "--table", "*", "--ttl", "900"])); + let (_listener, addr) = serve_binary(&binary, root, &["serve", "--http", "127.0.0.1:0", "--audience", AUD, "--max-in-flight", "2", "--project", "research", "--public-key", &public]); + let (status, state) = control(&addr, "POST", "/control/apply", Some(&admin), "{}"); + assert_eq!(status, 200, "{state}"); + assert_eq!(state["applied"], json!(2)); +} + static NONCE: AtomicU64 = AtomicU64::new(0); /// `POST /mcp` carrying `message`, with a proof from `key` or as a bare bearer. @@ -165,6 +185,86 @@ fn served_exchange_mints_a_reader_credential_for_the_read_face() { assert_eq!(refused["error"]["identifier"], "ExchangeAssertionInvalid"); } +fn control(addr: &str, method: &str, path: &str, token: Option<&str>, body: &str) -> (u16, Value) { + let mut stream = TcpStream::connect(addr).unwrap(); + let auth = token.map(|token| format!("Authorization: Bearer {token}\r\n")).unwrap_or_default(); + write!(stream, "{method} {path} HTTP/1.1\r\n{auth}Content-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}", body.len()).unwrap(); + let mut raw = String::new(); + stream.read_to_string(&mut raw).unwrap(); + let (head, answer) = raw.split_once("\r\n\r\n").unwrap(); + (head.split(' ').nth(1).unwrap().parse().unwrap(), serde_json::from_str(answer).unwrap()) +} + +#[test] +fn control_routes_require_admin_and_project_applied_workflows() { + let (dir, public) = project(); + let root = dir.path(); + let pipeline = "[[pipeline]]\nid = \"filings-flow\"\nschedule = \"every 1h\"\ntables = [\"research/notes\"]\n[pipeline.source]\nname = \"http\"\nconfig = { endpoint = \"https://example.test/filings\" }\n"; + std::fs::create_dir_all(root.join("pipelines")).unwrap(); + std::fs::write(root.join("pipelines/filings.toml"), pipeline).unwrap(); + stdout(&run(root, &["pipeline", "import", "--project", "research"])); + let read = stdout(&run(root, &["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", "user://dana@acme.example", "--zone", "on-prem:hq", "--action", "read", "--table", "*", "--ttl", "900"])); + let admin = stdout(&run(root, &["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", "user://dana@acme.example", "--zone", "on-prem:hq", "--action", "admin", "--table", "*", "--ttl", "900"])); + let narrow_admin = stdout(&run(root, &["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", "user://dana@acme.example", "--zone", "on-prem:hq", "--action", "admin", "--table", "unrelated/*", "--ttl", "900"])); + let (_listener, addr) = serve(root, &["serve", "--http", "127.0.0.1:0", "--audience", AUD, "--max-in-flight", "2", "--project", "research", "--public-key", &public]); + + assert_eq!(control(&addr, "GET", "/control/workflows", None, "").0, 401); + let (status, answer) = control(&addr, "GET", "/control/workflows", Some(&read), ""); + assert_eq!(status, 403, "{answer}"); + assert_eq!(control(&addr, "POST", "/control/apply", Some(&read), "{}").0, 403); + assert_eq!(control(&addr, "GET", "/control/workflows", Some(&narrow_admin), "").0, 403); + let (status, view) = control(&addr, "GET", "/control/workflows", Some(&admin), ""); + assert_eq!(status, 200, "{view}"); + assert_eq!(view["applied"], json!(1)); + assert_eq!(view["pipelines"][0]["id"], "filings-flow"); + assert_eq!(view["pipelines"][0]["tables"], json!(["research/notes"])); + + assert_eq!(control(&addr, "POST", "/control/apply", Some(&admin), "{\"id\":1}").0, 400); + + std::fs::write(root.join("pipelines/filings.toml"), pipeline.replace("every 1h", "every 1d")).unwrap(); + let (status, applied) = control(&addr, "POST", "/control/apply", Some(&admin), "{\"id\":\"filings-flow\"}"); + assert_eq!(status, 200, "{applied}"); + assert_eq!(applied["applied"], json!(2)); + let (_, view) = control(&addr, "GET", "/control/workflows", Some(&admin), ""); + assert_eq!(view["pipelines"][0]["schedule"], "every 1d"); +} + +#[test] +fn published_workflow_listing_caps_entries_and_flags_truncation() { + let (dir, public) = project(); + let root = dir.path(); + let snapshot = contextful_engine::control::SnapshotDir::open(&root.join(".contextful/control/research")); + let document = (0..1001).map(|index| format!( + "[[pipeline]]\nid = \"flow-{index:04}\"\ntables = [\"research/notes\"]\n[pipeline.source]\nname = \"http\"\nconfig = {{ endpoint = \"https://example.test/filings\" }}\n" + )).collect::>().join("\n"); + snapshot.import(&document).unwrap(); + let admin = stdout(&run(root, &["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", "user://dana@acme.example", "--zone", "on-prem:hq", "--action", "admin", "--table", "*", "--ttl", "900"])); + let (_listener, addr) = serve(root, &["serve", "--http", "127.0.0.1:0", "--audience", AUD, "--max-in-flight", "2", "--project", "research", "--public-key", &public]); + let (status, view) = control(&addr, "GET", "/control/workflows", Some(&admin), ""); + assert_eq!(status, 200, "{view}"); + assert_eq!(view["pipelines"].as_array().unwrap().len(), 1000); + assert_eq!(view["truncated"], true); + assert_eq!(view["declined"], 0); +} + +#[test] +fn published_workflows_include_completed_local_runs_before_sync_push() { + let vendor = super::pipeline::Vendor::start(|_| (200, "[{\"note_id\":\"filed\"}]".into())); + let (dir, public) = project(); + let root = dir.path(); + let pipeline = format!("[[pipeline]]\nid = \"filings-flow\"\ntables = [\"research/filings\"]\n[pipeline.source]\nname = \"http\"\nconfig = {{ endpoint = \"{}\" }}\n", vendor.url("/filings")); + std::fs::create_dir_all(root.join("pipelines")).unwrap(); + std::fs::write(root.join("pipelines/filings.toml"), pipeline).unwrap(); + stdout(&run(root, &["pipeline", "import", "--project", "research"])); + stdout(&run(root, &["pipeline", "run", "filings-flow", "--project", "research", "--run-id", "run-flow", "--site-id", "site-a"])); + let admin = stdout(&run(root, &["token", "mint", "--issuer-key", ".contextful/issuer.seed", "--on-behalf-of", "user://dana@acme.example", "--zone", "on-prem:hq", "--action", "admin", "--table", "*", "--ttl", "900"])); + let (_listener, addr) = serve(root, &["serve", "--http", "127.0.0.1:0", "--audience", AUD, "--max-in-flight", "2", "--project", "research", "--public-key", &public]); + let (status, view) = control(&addr, "GET", "/control/workflows", Some(&admin), ""); + assert_eq!(status, 200, "{view}"); + assert_eq!(view["runs"]["filings-flow"]["run_id"], "run-flow"); + assert_eq!(view["runs"]["filings-flow"]["status"], "success"); +} + /// The network transport refuses to start without its audience, its ceiling, or an issuer key that resolves and parses. // spec: topology.publish-hostname.issuer-key@a7736a7f #[test]