diff --git a/crates/executor-core/src/sdk/interceptor.rs b/crates/executor-core/src/sdk/interceptor.rs index 6663166..bdf5abe 100644 --- a/crates/executor-core/src/sdk/interceptor.rs +++ b/crates/executor-core/src/sdk/interceptor.rs @@ -25,7 +25,7 @@ use std::cell::{Cell, RefCell}; use std::sync::Arc; -use picloud_shared::{InterceptorEntry, InterceptorService, ScriptId, SdkCallCx}; +use picloud_shared::{InterceptorChain, InterceptorEntry, InterceptorService, ScriptId, SdkCallCx}; use rhai::EvalAltResult; use serde_json::{json, Value as Json}; use tokio::runtime::Handle as TokioHandle; @@ -102,12 +102,20 @@ impl Drop for CycleGuard { } } +/// The parsed outcome of ONE allowing hook: the operation passes, optionally +/// carrying a `data` transform (M4) the caller writes instead of the original. +struct HookAllow { + /// `Some` iff the hook returned a `data` key alongside `allowed: true`. + data: Option, +} + /// Run the before-op interceptor CHAIN for `(service, op)` if any are -/// registered. Returns `Ok(())` to allow the operation (un-hooked, or every -/// before-interceptor explicitly allowed it) or an `Err` runtime error to deny -/// it (the caller must NOT then perform the write). A single deny -/// short-circuits — the write is blocked. `value` is the payload being written -/// (`None` for delete). +/// registered. Returns `Ok(Some(value))` when a before-hook rewrote the payload +/// via a `data` transform (M4) — the caller writes THAT instead of the original +/// — or `Ok(None)` to write the original unchanged (un-hooked, or every hook +/// allowed without transforming). An `Err` denies the operation (the caller +/// must NOT perform the write); the FIRST deny short-circuits. `value` is the +/// original payload (`None` for delete, which never transforms). #[allow(clippy::too_many_arguments)] pub(super) fn run_before( ictx: &InterceptorCtx, @@ -117,33 +125,103 @@ pub(super) fn run_before( collection: &str, key: &str, value: Option<&Json>, -) -> Result<(), Box> { +) -> Result, Box> { // Re-entrancy break: writes performed BY an interceptor (or anything it // invokes) are not themselves intercepted — otherwise an "allow + write" // hook would recurse to the depth cap and deny the original op. if currently_in_interceptor() { - return Ok(()); + return Ok(None); } + let chain = resolve_chain(ictx, cx, service, op)?; + if chain.before.is_empty() { + return Ok(None); + } + + // Thread the (possibly rewritten) value through the chain: each before-hook + // sees the prior hook's transform in `value`, and the write uses the final. + let mut current: Option = value.cloned(); + let mut transformed = false; + for entry in &chain.before { + let payload = op_payload(service, op, collection, key, current.as_ref(), cx, None); + let outcome = run_one_hook(ictx, cx, entry, service, op, &payload)?; + // M4 data-transform: honored ONLY on an allowing hook (the verdict parse + // already enforced fail-closed). Delete has no value to rewrite. + if let Some(data) = outcome.data { + if value.is_some() { + enforce_transform_size(service, op, &data)?; + current = Some(data); + transformed = true; + } + } + } + + Ok(transformed.then(|| current.unwrap_or(Json::Null))) +} + +/// Run the after-op interceptor CHAIN (M3) once the write has committed. +/// `result` is the write's outcome (serialized), surfaced to the hook so it can +/// observe/audit. After-hooks CANNOT roll back — the write already happened; a +/// deny here surfaces as an operation error to the caller but the write +/// persists. Returns `Ok(())` when every after-hook allowed (or none exist), an +/// `Err` on the first deny. +#[allow(clippy::too_many_arguments)] +pub(super) fn run_after( + ictx: &InterceptorCtx, + cx: &Arc, + service: &'static str, + op: &'static str, + collection: &str, + key: &str, + value: Option<&Json>, + result: Json, +) -> Result<(), Box> { + if currently_in_interceptor() { + return Ok(()); + } + let chain = resolve_chain(ictx, cx, service, op)?; + if chain.after.is_empty() { + return Ok(()); + } + for entry in &chain.after { + let payload = op_payload(service, op, collection, key, value, cx, Some(&result)); + // After-hooks observe/deny; a returned `data` is ignored (the write is + // already durable, so there is nothing to rewrite for the kv slice). + run_one_hook(ictx, cx, entry, service, op, &payload)?; + } + Ok(()) +} + +/// Resolve the before+after chain for `(service, op)` — every marker visible on +/// the calling app's chain, ordered per phase. An `Err` (no tokio runtime, or a +/// backend failure) makes the caller fail closed rather than silently allow. +fn resolve_chain( + ictx: &InterceptorCtx, + cx: &Arc, + service: &'static str, + op: &'static str, +) -> Result> { let handle = TokioHandle::try_current() .map_err(|e| runtime_err(&format!("{service} interceptor: no tokio runtime: {e}")))?; + let interceptors = ictx.interceptors.clone(); + let cx = cx.clone(); + handle + .block_on(async move { interceptors.resolve(&cx, service, op).await }) + .map_err(|e| runtime_err(&format!("{service}::{op} interceptor resolve: {e}"))) +} - // (1) Resolve the whole before+after chain, each entry sealed to its - // declaring owner. `before` is ordered ancestor→app (an outer/group guard - // runs first). `after` is unused this milestone. - let chain = { - let interceptors = ictx.interceptors.clone(); - let cx = cx.clone(); - handle - .block_on(async move { interceptors.resolve(&cx, service, op).await }) - .map_err(|e| runtime_err(&format!("{service}::{op} interceptor resolve: {e}")))? - }; - // Un-hooked: the ONLY path that allows without running anything. - if chain.before.is_empty() { - return Ok(()); - } - - let payload = json!({ +/// Build the operation-context payload handed to a hook. `result` is `Some` only +/// for an after-hook (the write's outcome). +fn op_payload( + service: &str, + op: &str, + collection: &str, + key: &str, + value: Option<&Json>, + cx: &SdkCallCx, + result: Option<&Json>, +) -> Json { + let mut m = json!({ "service": service, "action": op, "collection": collection, @@ -152,88 +230,114 @@ pub(super) fn run_before( "caller_script_id": cx.script_id.to_string(), "caller_execution_id": cx.execution_id.to_string(), }); + if let (Json::Object(obj), Some(r)) = (&mut m, result) { + obj.insert("result".to_string(), r.clone()); + } + m +} - // (2) Run each before-interceptor in order; the FIRST deny short-circuits. - for entry in &chain.before { - let resolved = match entry { - // A guard is declared but its script is missing/disabled — fail closed. - InterceptorEntry::Dangling(name) => { - return Err(runtime_err(&format!( - "{service}::{op} denied: interceptor script `{name}` is missing or disabled" - ))); - } - InterceptorEntry::Run(r) => &r.script, - }; - - // Identity cycle guard: if this interceptor's script is already on the - // run stack, running it again would recurse forever — DENY. - if CycleGuard::contains(resolved.script_id) { +/// Run ONE chain entry: fail-closed on a dangling marker, an identity cycle, a +/// missing engine back-reference, or a depth-limit breach; otherwise run the +/// sealed script and parse its verdict. Returns the allow outcome (with any +/// `data` transform) or an `Err` denying the operation. +fn run_one_hook( + ictx: &InterceptorCtx, + cx: &Arc, + entry: &InterceptorEntry, + service: &str, + op: &str, + payload: &Json, +) -> Result> { + let resolved = match entry { + // A guard is declared but its script is missing/disabled — fail closed. + InterceptorEntry::Dangling(name) => { return Err(runtime_err(&format!( - "{service}::{op} denied: interceptor cycle detected: `{}`", - resolved.name + "{service}::{op} denied: interceptor script `{name}` is missing or disabled" ))); } + InterceptorEntry::Run(r) => r, + }; + let script = &resolved.script; - // A guard is registered but the engine back-reference isn't installed (a - // bare test engine without `set_self_weak`): we cannot run it, so DENY — - // a resolvable guard we can't execute must not fail open. Production - // always installs the back-ref. - let Some(self_engine) = ictx.self_engine.as_ref() else { - return Err(runtime_err(&format!( - "{service}::{op} denied: interceptor `{}` cannot run (engine back-reference not installed)", - resolved.name - ))); - }; - - // Depth bound (shared with invoke / trigger fan-out). - if cx.trigger_depth + 1 > ictx.limits.trigger_depth_max { - return Err(runtime_err(&format!( - "{service}::{op} interceptor `{}`: depth limit exceeded (max {})", - resolved.name, ictx.limits.trigger_depth_max - ))); - } - - // Run the sealed script with the operation context as its request body, - // under the re-entrancy guard (its own writes bypass interception) and - // the identity cycle guard (its own script id is on the stack). - let ret = { - let _reentry = ReentryGuard::enter(); - let _cycle = CycleGuard::enter(resolved.script_id); - run_resolved_blocking( - self_engine, - cx, - resolved, - payload.clone(), - &format!("{service}::{op} interceptor `{}`", resolved.name), - )? - }; - - // Fail-closed verdict: allow ONLY on an explicit `#{ allowed: true }`. A - // bare bool, a typo'd key, a non-bool `allowed`, a bare unit, or any - // other shape DENIES — a control whose job is to deny must not allow by - // omission. - match &ret { - // Explicit allow → this entry passes; continue to the next. - Json::Object(m) if m.get("allowed") == Some(&Json::Bool(true)) => {} - Json::Object(m) => { - let reason = m - .get("reason") - .and_then(Json::as_str) - .unwrap_or("denied by interceptor"); - return Err(runtime_err(&format!( - "{service}::{op} denied by interceptor `{}`: {reason}", - resolved.name - ))); - } - _ => { - return Err(runtime_err(&format!( - "{service}::{op} denied by interceptor `{}`: no allow verdict returned", - resolved.name - ))); - } - } + // Identity cycle guard: if this interceptor's script is already on the run + // stack, running it again would recurse forever — DENY. + if CycleGuard::contains(script.script_id) { + return Err(runtime_err(&format!( + "{service}::{op} denied: interceptor cycle detected: `{}`", + script.name + ))); } - // Every before-interceptor allowed. + // A guard is registered but the engine back-reference isn't installed (a bare + // test engine without `set_self_weak`): we cannot run it, so DENY — a + // resolvable guard we can't execute must not fail open. Production always + // installs the back-ref. + let Some(self_engine) = ictx.self_engine.as_ref() else { + return Err(runtime_err(&format!( + "{service}::{op} denied: interceptor `{}` cannot run (engine back-reference not installed)", + script.name + ))); + }; + + // Depth bound (shared with invoke / trigger fan-out). + if cx.trigger_depth + 1 > ictx.limits.trigger_depth_max { + return Err(runtime_err(&format!( + "{service}::{op} interceptor `{}`: depth limit exceeded (max {})", + script.name, ictx.limits.trigger_depth_max + ))); + } + + // Run the sealed script with the operation context as its request body, under + // the re-entrancy guard (its own writes bypass interception) and the identity + // cycle guard (its own script id is on the stack). + let ret = { + let _reentry = ReentryGuard::enter(); + let _cycle = CycleGuard::enter(script.script_id); + run_resolved_blocking( + self_engine, + cx, + script, + payload.clone(), + &format!("{service}::{op} interceptor `{}`", script.name), + )? + }; + + // Fail-closed verdict: allow ONLY on an explicit `#{ allowed: true }`. A bare + // bool, a typo'd key, a non-bool `allowed`, a bare unit, or any other shape + // DENIES — a control whose job is to deny must not allow by omission. + match ret { + Json::Object(mut m) if m.get("allowed") == Some(&Json::Bool(true)) => Ok(HookAllow { + data: m.remove("data"), + }), + Json::Object(m) => { + let reason = m + .get("reason") + .and_then(Json::as_str) + .unwrap_or("denied by interceptor"); + Err(runtime_err(&format!( + "{service}::{op} denied by interceptor `{}`: {reason}", + script.name + ))) + } + _ => Err(runtime_err(&format!( + "{service}::{op} denied by interceptor `{}`: no allow verdict returned", + script.name + ))), + } +} + +/// Cap a `data` transform at the same byte ceiling the SDK uses for a Rhai→JSON +/// materialization, so an interceptor can't rewrite a small write into an +/// unbounded blob before it reaches the service's own value cap. +fn enforce_transform_size(service: &str, op: &str, data: &Json) -> Result<(), Box> { + let len = serde_json::to_vec(data) + .map(|v| v.len()) + .unwrap_or(usize::MAX); + if len > crate::sdk::bridge::MAX_JSON_MATERIALIZE_BYTES { + return Err(runtime_err(&format!( + "{service}::{op} interceptor transform too large ({len} bytes exceeds the {}-byte limit)", + crate::sdk::bridge::MAX_JSON_MATERIALIZE_BYTES + ))); + } Ok(()) } diff --git a/crates/executor-core/src/sdk/kv.rs b/crates/executor-core/src/sdk/kv.rs index 8a0f35d..7601cd2 100644 --- a/crates/executor-core/src/sdk/kv.rs +++ b/crates/executor-core/src/sdk/kv.rs @@ -172,9 +172,9 @@ fn register_set(engine: &mut RhaiEngine) { "set", |handle: &mut KvHandle, key: &str, value: Dynamic| -> Result<(), Box> { let json = dynamic_to_json_capped(&value, MAX_JSON_MATERIALIZE_BYTES)?; - // §9.4 before-op interceptor (allow/deny). A denial errors here and - // the write below never runs. - super::interceptor::run_before( + // §9.4 before-op interceptor (allow/deny + M4 data-transform). A + // denial errors here; a returned `data` rewrites the written value. + let write_val = super::interceptor::run_before( &handle.ictx, &handle.cx, "kv", @@ -182,11 +182,24 @@ fn register_set(engine: &mut RhaiEngine) { &handle.collection, key, Some(&json), - )?; + )? + .unwrap_or(json); + let written = write_val.clone(); let h = handle.clone(); block_on("kv", async move { - h.service.set(&h.cx, &h.collection, key, json).await - }) + h.service.set(&h.cx, &h.collection, key, write_val).await + })?; + // §9.4 M3 after-hook (observe/audit; the write already committed). + super::interceptor::run_after( + &handle.ictx, + &handle.cx, + "kv", + "set", + &handle.collection, + key, + Some(&written), + serde_json::Value::Null, + ) }, ); } @@ -228,6 +241,7 @@ fn register_delete(engine: &mut RhaiEngine) { engine.register_fn( "delete", |handle: &mut KvHandle, key: &str| -> Result> { + // Delete carries no value, so the before-hook never transforms. super::interceptor::run_before( &handle.ictx, &handle.cx, @@ -238,9 +252,20 @@ fn register_delete(engine: &mut RhaiEngine) { None, )?; let h = handle.clone(); - block_on("kv", async move { + let was_present = block_on("kv", async move { h.service.delete(&h.cx, &h.collection, key).await - }) + })?; + super::interceptor::run_after( + &handle.ictx, + &handle.cx, + "kv", + "delete", + &handle.collection, + key, + None, + serde_json::Value::Bool(was_present), + )?; + Ok(was_present) }, ); } @@ -252,7 +277,7 @@ fn group_run_before( op: &'static str, key: &str, value: Option<&serde_json::Value>, -) -> Result<(), Box> { +) -> Result, Box> { super::interceptor::run_before( &handle.ictx, &handle.cx, @@ -264,6 +289,25 @@ fn group_run_before( ) } +fn group_run_after( + handle: &GroupKvHandle, + op: &'static str, + key: &str, + value: Option<&serde_json::Value>, + result: serde_json::Value, +) -> Result<(), Box> { + super::interceptor::run_after( + &handle.ictx, + &handle.cx, + "kv", + op, + &handle.collection, + key, + value, + result, + ) +} + fn register_list(engine: &mut RhaiEngine) { // Zero-arg form — full page, no cursor. engine.register_fn( @@ -330,12 +374,15 @@ fn register_group_set(engine: &mut RhaiEngine) { "set", |handle: &mut GroupKvHandle, key: &str, value: Dynamic| -> Result<(), Box> { let json = dynamic_to_json_capped(&value, MAX_JSON_MATERIALIZE_BYTES)?; - // §9.4 before-op interceptor also guards shared-collection writes. - group_run_before(handle, "set", key, Some(&json))?; + // §9.4 before-op interceptor also guards shared-collection writes + // (allow/deny + M4 data-transform). + let write_val = group_run_before(handle, "set", key, Some(&json))?.unwrap_or(json); + let written = write_val.clone(); let h = handle.clone(); block_on("kv", async move { - h.service.set(&h.cx, &h.collection, key, json).await - }) + h.service.set(&h.cx, &h.collection, key, write_val).await + })?; + group_run_after(handle, "set", key, Some(&written), serde_json::Value::Null) }, ); } @@ -376,9 +423,17 @@ fn register_group_delete(engine: &mut RhaiEngine) { |handle: &mut GroupKvHandle, key: &str| -> Result> { group_run_before(handle, "delete", key, None)?; let h = handle.clone(); - block_on("kv", async move { + let was_present = block_on("kv", async move { h.service.delete(&h.cx, &h.collection, key).await - }) + })?; + group_run_after( + handle, + "delete", + key, + None, + serde_json::Value::Bool(was_present), + )?; + Ok(was_present) }, ); } diff --git a/crates/manager-core/src/apply_service.rs b/crates/manager-core/src/apply_service.rs index adb49e1..54791c2 100644 --- a/crates/manager-core/src/apply_service.rs +++ b/crates/manager-core/src/apply_service.rs @@ -121,9 +121,17 @@ pub struct Bundle { pub struct BundleInterceptor { pub service: String, pub op: String, + /// `before` (allow/deny + transform) or `after` (observe; §9.4 M3). + /// Defaults to `before` so a pre-M3 CLI (which omits it) still applies. + #[serde(default = "default_before_phase")] + pub phase: String, pub script: String, } +fn default_before_phase() -> String { + "before".to_string() +} + /// One declared workflow on the wire: a name + its (already-parsed) DAG. #[derive(Debug, Clone, Deserialize)] pub struct BundleWorkflow { @@ -820,8 +828,8 @@ pub struct CurrentState { /// v1.2 Workflows owned directly by this node (app-owned; empty for a group). pub workflows: Vec, /// §9.4 interceptor markers declared directly at this node, as - /// `(service, op, script)` triples. - pub interceptors: Vec<(String, String, String)>, + /// `(service, op, phase, script)` tuples. + pub interceptors: Vec<(String, String, String, String)>, } /// One row of the read-only extension-point report (§5.5). @@ -1281,9 +1289,10 @@ impl ApplyService { validate_workflow_definition(&w.name, &w.definition).map_err(ApplyError::Invalid)?; } // §9.4 interceptors (MVP): `service = "kv"`, `op ∈ {set, delete}`, - // authored on app OR group. One marker per `(service, op)` — a duplicate - // would collide on the reconcile key and the DB's partial-unique index. - let mut seen_hooks: HashSet<(String, String)> = HashSet::new(); + // `phase ∈ {before, after}` (§9.4 M3), authored on app OR group. One + // marker per `(service, op, phase)` — a duplicate would collide on the + // reconcile key and the DB's partial-unique index. + let mut seen_hooks: HashSet<(String, String, String)> = HashSet::new(); for i in &bundle.interceptors { if i.service != "kv" { return Err(ApplyError::Invalid(format!( @@ -1297,15 +1306,21 @@ impl ApplyService { i.op ))); } + if i.phase != "before" && i.phase != "after" { + return Err(ApplyError::Invalid(format!( + "interceptor phase `{}` is not supported — only `before` / `after`", + i.phase + ))); + } if i.script.trim().is_empty() { return Err(ApplyError::Invalid( "an interceptor must name a `script`".into(), )); } - if !seen_hooks.insert((i.service.clone(), i.op.clone())) { + if !seen_hooks.insert((i.service.clone(), i.op.clone(), i.phase.clone())) { return Err(ApplyError::Invalid(format!( - "duplicate interceptor for `{}/{}`", - i.service, i.op + "duplicate interceptor for `{}/{}/{}`", + i.service, i.op, i.phase ))); } } @@ -1594,7 +1609,7 @@ impl ApplyService { let bi = bundle .interceptors .iter() - .find(|i| format!("{}/{}", i.service, i.op) == ch.key) + .find(|i| format!("{}/{}/{}", i.service, i.op, i.phase) == ch.key) .ok_or_else(|| { ApplyError::Backend("internal: interceptor plan/bundle mismatch".into()) })?; @@ -1603,6 +1618,7 @@ impl ApplyService { owner.as_script_owner(), &bi.service, &bi.op, + &bi.phase, &bi.script, ) .await @@ -1746,16 +1762,20 @@ impl ApplyService { } // §9.4 interceptor markers are prunable config too. The delete keys - // by `(service, op)` (the marker's identity), so a script-change - // Update above is never clobbered here. + // by `(service, op, phase)` (the marker's identity), so a + // script-change Update above is never clobbered here. for ch in &plan.interceptors { if ch.op == Op::Delete { - let (service, op) = ch.key.split_once('/').unwrap_or((ch.key.as_str(), "")); + let mut parts = ch.key.splitn(3, '/'); + let service = parts.next().unwrap_or(""); + let op = parts.next().unwrap_or(""); + let phase = parts.next().unwrap_or("before"); crate::interceptor_repo::delete_interceptor_tx( &mut *tx, owner.as_script_owner(), service, op, + phase, ) .await .map_err(|e| ApplyError::Backend(e.to_string()))?; @@ -4243,7 +4263,7 @@ impl ApplyService { .await .map_err(|e| ApplyError::Backend(e.to_string()))? .into_iter() - .map(|m| (m.service, m.op, m.script)) + .map(|m| (m.service, m.op, m.phase, m.script)) .collect(); Ok(CurrentState { scripts, @@ -4331,15 +4351,15 @@ fn compute_diff_with_names( /// the same `(service, op)` is an `Update`; a live marker the manifest stops /// declaring is a `Delete` (applied only under `--prune`). fn diff_interceptors(current: &CurrentState, bundle: &Bundle) -> Vec { - let key = |service: &str, op: &str| format!("{service}/{op}"); + let key = |service: &str, op: &str, phase: &str| format!("{service}/{op}/{phase}"); let live: HashMap = current .interceptors .iter() - .map(|(s, o, script)| (key(s, o), script.as_str())) + .map(|(s, o, p, script)| (key(s, o, p), script.as_str())) .collect(); let mut out = Vec::new(); for bi in &bundle.interceptors { - let k = key(&bi.service, &bi.op); + let k = key(&bi.service, &bi.op, &bi.phase); match live.get(&k) { Some(cur) if *cur == bi.script => out.push(ResourceChange { op: Op::NoOp, @@ -4358,15 +4378,15 @@ fn diff_interceptors(current: &CurrentState, bundle: &Bundle) -> Vec Result, sqlx::Error> { - let rows: Vec<(String, String, String)> = match owner { + let rows: Vec<(String, String, String, String)> = match owner { ScriptOwner::App(a) => { sqlx::query_as( - "SELECT service, op, interceptor_script FROM interceptors \ - WHERE app_id = $1 ORDER BY service, op", + "SELECT service, op, phase, interceptor_script FROM interceptors \ + WHERE app_id = $1 ORDER BY service, op, phase", ) .bind(a.into_inner()) .fetch_all(pool) @@ -41,8 +43,8 @@ pub async fn list_for_owner( } ScriptOwner::Group(g) => { sqlx::query_as( - "SELECT service, op, interceptor_script FROM interceptors \ - WHERE group_id = $1 ORDER BY service, op", + "SELECT service, op, phase, interceptor_script FROM interceptors \ + WHERE group_id = $1 ORDER BY service, op, phase", ) .bind(g.into_inner()) .fetch_all(pool) @@ -51,9 +53,10 @@ pub async fn list_for_owner( }; Ok(rows .into_iter() - .map(|(service, op, script)| InterceptorMarker { + .map(|(service, op, phase, script)| InterceptorMarker { service, op, + phase, script, }) .collect()) @@ -66,21 +69,23 @@ pub async fn list_on_app_chain( pool: &PgPool, app_id: AppId, ) -> Result, sqlx::Error> { - let rows: Vec<(String, String, String)> = sqlx::query_as(&format!( + let rows: Vec<(String, String, String, String)> = sqlx::query_as(&format!( "{CHAIN_LEVELS_CTE} \ - SELECT DISTINCT ON (i.service, i.op) i.service, i.op, i.interceptor_script \ + SELECT DISTINCT ON (i.service, i.op, i.phase) \ + i.service, i.op, i.phase, i.interceptor_script \ FROM chain c \ JOIN interceptors i ON (i.app_id = c.app_owner OR i.group_id = c.group_owner) \ - ORDER BY i.service, i.op, c.depth ASC", + ORDER BY i.service, i.op, i.phase, c.depth ASC", )) .bind(app_id.into_inner()) .fetch_all(pool) .await?; Ok(rows .into_iter() - .map(|(service, op, script)| InterceptorMarker { + .map(|(service, op, phase, script)| InterceptorMarker { service, op, + phase, script, }) .collect()) @@ -218,37 +223,38 @@ pub async fn insert_interceptor_tx( owner: ScriptOwner, service: &str, op: &str, + phase: &str, script: &str, ) -> Result<(), sqlx::Error> { match owner { ScriptOwner::App(a) => { sqlx::query( - // `phase` defaults to 'before' (0074); the conflict arbiter must - // include it to match the per-(owner, service, op, phase) index. - "INSERT INTO interceptors (app_id, service, op, interceptor_script) \ - VALUES ($1, $2, $3, $4) \ + // The conflict arbiter includes `phase` to match the + // per-(owner, service, op, phase) index (§9.4 M3). + "INSERT INTO interceptors (app_id, service, op, phase, interceptor_script) \ + VALUES ($1, $2, $3, $4, $5) \ ON CONFLICT (app_id, service, op, phase) WHERE app_id IS NOT NULL \ DO UPDATE SET interceptor_script = EXCLUDED.interceptor_script, updated_at = NOW()", ) .bind(a.into_inner()) .bind(service) .bind(op) + .bind(phase) .bind(script) .execute(&mut **tx) .await?; } ScriptOwner::Group(g) => { sqlx::query( - // `phase` defaults to 'before' (0074); the conflict arbiter must - // include it to match the per-(owner, service, op, phase) index. - "INSERT INTO interceptors (group_id, service, op, interceptor_script) \ - VALUES ($1, $2, $3, $4) \ + "INSERT INTO interceptors (group_id, service, op, phase, interceptor_script) \ + VALUES ($1, $2, $3, $4, $5) \ ON CONFLICT (group_id, service, op, phase) WHERE group_id IS NOT NULL \ DO UPDATE SET interceptor_script = EXCLUDED.interceptor_script, updated_at = NOW()", ) .bind(g.into_inner()) .bind(service) .bind(op) + .bind(phase) .bind(script) .execute(&mut **tx) .await?; @@ -257,30 +263,37 @@ pub async fn insert_interceptor_tx( Ok(()) } -/// Delete a marker at `owner` (by `(service, op)`), in the apply transaction. -/// Used by `--prune` when the manifest stops declaring it. +/// Delete a marker at `owner` (by `(service, op, phase)`), in the apply +/// transaction. Used by `--prune` when the manifest stops declaring it. pub async fn delete_interceptor_tx( tx: &mut Transaction<'_, Postgres>, owner: ScriptOwner, service: &str, op: &str, + phase: &str, ) -> Result<(), sqlx::Error> { match owner { ScriptOwner::App(a) => { - sqlx::query("DELETE FROM interceptors WHERE app_id = $1 AND service = $2 AND op = $3") - .bind(a.into_inner()) - .bind(service) - .bind(op) - .execute(&mut **tx) - .await?; + sqlx::query( + "DELETE FROM interceptors \ + WHERE app_id = $1 AND service = $2 AND op = $3 AND phase = $4", + ) + .bind(a.into_inner()) + .bind(service) + .bind(op) + .bind(phase) + .execute(&mut **tx) + .await?; } ScriptOwner::Group(g) => { sqlx::query( - "DELETE FROM interceptors WHERE group_id = $1 AND service = $2 AND op = $3", + "DELETE FROM interceptors \ + WHERE group_id = $1 AND service = $2 AND op = $3 AND phase = $4", ) .bind(g.into_inner()) .bind(service) .bind(op) + .bind(phase) .execute(&mut **tx) .await?; } diff --git a/crates/picloud-cli/src/cmds/plan.rs b/crates/picloud-cli/src/cmds/plan.rs index 944e267..5776a33 100644 --- a/crates/picloud-cli/src/cmds/plan.rs +++ b/crates/picloud-cli/src/cmds/plan.rs @@ -263,7 +263,7 @@ pub fn build_bundle(manifest: &Manifest, base_dir: &Path) -> Result { .iter() .flat_map(|i| { i.ops.iter().map(move |op| { - json!({ "service": i.service, "op": op, "script": i.script }) + json!({ "service": i.service, "op": op, "phase": i.phase, "script": i.script }) }) }) .collect::>(), diff --git a/crates/picloud-cli/src/manifest.rs b/crates/picloud-cli/src/manifest.rs index f69fc07..53083ca 100644 --- a/crates/picloud-cli/src/manifest.rs +++ b/crates/picloud-cli/src/manifest.rs @@ -83,12 +83,21 @@ pub struct ManifestInterceptor { pub service: String, /// The guarded operations, e.g. `["set", "delete"]`. pub ops: Vec, + /// When the hook runs relative to the op: `"before"` (default — + /// allow/deny + data transform) or `"after"` (§9.4 M3 — observe/audit; + /// cannot roll back the write). + #[serde(default = "default_interceptor_phase")] + pub phase: String, } fn default_interceptor_service() -> String { "kv".to_string() } +fn default_interceptor_phase() -> String { + "before".to_string() +} + impl Manifest { /// Parse a manifest from TOML text. Enforces the app-XOR-group invariant. pub fn parse(text: &str) -> Result { diff --git a/crates/picloud-cli/tests/interceptors.rs b/crates/picloud-cli/tests/interceptors.rs index 76a1ade..f0604c9 100644 --- a/crates/picloud-cli/tests/interceptors.rs +++ b/crates/picloud-cli/tests/interceptors.rs @@ -594,3 +594,136 @@ fn a_self_referential_interceptor_does_not_recurse() { .unwrap(); assert!(x.contains('1'), "the original write must persist:\n{x}"); } + +// --- §9.4 M4: data transform (before) ------------------------------------- + +/// A before-interceptor that REWRITES the written value via `data` for the key +/// `"doc"` (allow + transform), passing everything else through unchanged. +const TRANSFORM_GUARD: &str = r#" +let op = ctx.request.body; +if op.action == "set" && op.key == "doc" { + #{ allowed: true, data: #{ redacted: true } } +} else { + #{ allowed: true } +} +"#; + +/// A writer that sets `doc` to a raw value and reads back what was stored. +const TRANSFORM_WRITER: &str = r#" +kv::collection("c").set("doc", #{ secret: "raw" }); +kv::collection("c").get("doc") +"#; + +/// M4: a before-interceptor that returns `#{ allowed: true, data: ... }` rewrites +/// the value actually written — the store holds the transform, not the original. +#[ignore = "needs DATABASE_URL pointing at a running Postgres"] +#[test] +fn a_before_interceptor_transforms_the_written_value() { + let Some(fx) = common::fixture_or_skip() else { + return; + }; + let env = common::admin_env(fx); + let app = common::unique_slug("xf-app"); + let _a = AppGuard::new(&env.url, &env.token, &app); + common::pic_as(&env) + .args(["apps", "create", &app]) + .assert() + .success(); + + let dir = manifest_dir(); + fs::write(dir.path().join("scripts/guard.rhai"), TRANSFORM_GUARD).unwrap(); + fs::write(dir.path().join("scripts/writer.rhai"), TRANSFORM_WRITER).unwrap(); + fs::write( + dir.path().join("picloud.toml"), + format!( + "[app]\nslug = \"{app}\"\nname = \"XF\"\n\n\ + [[scripts]]\nname = \"guard\"\nfile = \"scripts/guard.rhai\"\n\n\ + [[scripts]]\nname = \"writer\"\nfile = \"scripts/writer.rhai\"\n\n\ + [[interceptors]]\nscript = \"guard\"\nops = [\"set\"]\n" + ), + ) + .unwrap(); + common::pic_as(&env) + .args(["apply", "--file"]) + .arg(dir.path().join("picloud.toml")) + .assert() + .success(); + + // The writer reads back the STORED value — it must be the transform. + let body = invoke_body(&env, &app_script_id(&env, &app, "writer")); + assert_eq!( + body, + serde_json::json!({ "redacted": true }), + "the stored value must be the interceptor's `data` transform, not the original" + ); +} + +// --- §9.4 M3: after-hooks -------------------------------------------------- + +/// An AFTER-interceptor on `delete` that reads the write's outcome +/// (`op.result`, the was-present bool) and DENIES when a key was actually +/// removed. It cannot roll the delete back (the write already committed) — the +/// deny only surfaces as an error to the caller. +const AFTER_DELETE_GUARD: &str = r#" +let op = ctx.request.body; +if op.action == "delete" && op.result == true { + #{ allowed: false, reason: "an existing key was deleted" } +} else { + #{ allowed: true } +} +"#; + +/// A writer that seeds a key, deletes it (triggering the after-hook), then +/// checks whether the key survived. +const AFTER_DELETE_WRITER: &str = r#" +kv::collection("c").set("k", 1); +let after_denied = false; +try { kv::collection("c").delete("k"); } catch(e) { after_denied = true; } +let still = kv::collection("c").has("k"); +#{ after_denied: after_denied, still: still } +"#; + +/// M3: an after-hook runs with the write's `result` in its payload, can DENY +/// (surfacing an error to the caller), but CANNOT roll the write back. The +/// delete's `result` (was-present = true) drives the deny, and the key is gone +/// afterwards despite the error. +#[ignore = "needs DATABASE_URL pointing at a running Postgres"] +#[test] +fn an_after_hook_sees_the_result_and_cannot_roll_back() { + let Some(fx) = common::fixture_or_skip() else { + return; + }; + let env = common::admin_env(fx); + let app = common::unique_slug("after-app"); + let _a = AppGuard::new(&env.url, &env.token, &app); + common::pic_as(&env) + .args(["apps", "create", &app]) + .assert() + .success(); + + let dir = manifest_dir(); + fs::write(dir.path().join("scripts/guard.rhai"), AFTER_DELETE_GUARD).unwrap(); + fs::write(dir.path().join("scripts/writer.rhai"), AFTER_DELETE_WRITER).unwrap(); + fs::write( + dir.path().join("picloud.toml"), + format!( + "[app]\nslug = \"{app}\"\nname = \"After\"\n\n\ + [[scripts]]\nname = \"guard\"\nfile = \"scripts/guard.rhai\"\n\n\ + [[scripts]]\nname = \"writer\"\nfile = \"scripts/writer.rhai\"\n\n\ + [[interceptors]]\nscript = \"guard\"\nops = [\"delete\"]\nphase = \"after\"\n" + ), + ) + .unwrap(); + common::pic_as(&env) + .args(["apply", "--file"]) + .arg(dir.path().join("picloud.toml")) + .assert() + .success(); + + let body = invoke_body(&env, &app_script_id(&env, &app, "writer")); + assert_eq!( + body, + serde_json::json!({ "after_denied": true, "still": false }), + "the after-hook must see result==true and deny, but the delete must have persisted" + ); +}