From ef16d48a8c9d6f3c1faa1bf1a293e2b01ea7ae4a Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Thu, 16 Jul 2026 20:37:18 +0200 Subject: [PATCH] =?UTF-8?q?feat(interceptors):=20extend=20before/after=20h?= =?UTF-8?q?ooks=20to=20docs/files/queue/pubsub/http=20(=C2=A79.4=20M7-M11)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every non-KV mutating op now runs the same before/after interceptor machinery: - docs create/update/delete (+ group) — create/update honor the M4 data transform; - files create/update/delete (+ group) — allow/deny only (the blob is never surfaced to the hook, value = None); - queue enqueue (+ shared) — transform-capable; after reports the message id; - pubsub publish_durable (+ shared topic) — transform-capable; - http request — all verbs funnel through two svc.request sites; collection = method, key = url, body not surfaced. ictx threaded into the free-fn/handle paths that lacked it (http was _ictx; queue/pubsub per-app publish). validate_bundle_for generalized to a per-service allowed-ops map (kv set/delete; docs/files create/update/delete; queue enqueue; pubsub publish; http request) — unknown service/op still rejected. INVARIANT enforced + verified: every allowed (service, op) has a matching runtime hook, so no validated-but-unhooked fail-open. Pinned by a new docs-create deny journey (11 interceptor journeys green). Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/executor-core/src/sdk/docs.rs | 176 ++++++++++++++++++++--- crates/executor-core/src/sdk/files.rs | 140 +++++++++++++++--- crates/executor-core/src/sdk/http.rs | 107 +++++++++----- crates/executor-core/src/sdk/pubsub.rs | 66 +++++++-- crates/executor-core/src/sdk/queue.rs | 81 ++++++++--- crates/manager-core/src/apply_service.rs | 31 ++-- crates/picloud-cli/src/manifest.rs | 3 +- crates/picloud-cli/tests/interceptors.rs | 72 ++++++++++ 8 files changed, 564 insertions(+), 112 deletions(-) diff --git a/crates/executor-core/src/sdk/docs.rs b/crates/executor-core/src/sdk/docs.rs index 7e2c004..4a2ce1f 100644 --- a/crates/executor-core/src/sdk/docs.rs +++ b/crates/executor-core/src/sdk/docs.rs @@ -39,9 +39,7 @@ pub struct DocsHandle { collection: String, service: Arc, cx: Arc, - // §9.4 M1 plumbing: carried so `create`/`update`/`delete` can run a before/ - // after hook in a later milestone (M8). Unused until then. - #[allow(dead_code)] + // §9.4: carried so `create`/`update`/`delete` run the before/after hook (M8). ictx: InterceptorCtx, } @@ -54,8 +52,7 @@ pub struct GroupDocsHandle { collection: String, service: Arc, cx: Arc, - // §9.4 M1 plumbing (see `DocsHandle`). Unused until M8. - #[allow(dead_code)] + // §9.4: carried so shared `create`/`update`/`delete` run the hook too (M8). ictx: InterceptorCtx, } @@ -135,12 +132,23 @@ fn register_create(engine: &mut RhaiEngine) { engine.register_fn( "create", |handle: &mut DocsHandle, data: Map| -> Result> { - let h = handle.clone(); let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?; + // §9.4 before-op interceptor (allow/deny + M4 data-transform). + let write_val = run_before(handle, "create", "", Some(&json))?.unwrap_or(json); + let written = write_val.clone(); + let h = handle.clone(); let id = block_on("docs", async move { - h.service.create(&h.cx, &h.collection, json).await + h.service.create(&h.cx, &h.collection, write_val).await })?; - Ok(id.to_string()) + let id_str = id.to_string(); + run_after( + handle, + "create", + "", + Some(&written), + serde_json::Value::String(id_str.clone()), + )?; + Ok(id_str) }, ); } @@ -194,14 +202,23 @@ fn register_update(engine: &mut RhaiEngine) { engine.register_fn( "update", |handle: &mut DocsHandle, id: &str, data: Map| -> Result<(), Box> { - let h = handle.clone(); let parsed_id = parse_doc_id(id)?; let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?; + let write_val = run_before(handle, "update", id, Some(&json))?.unwrap_or(json); + let written = write_val.clone(); + let h = handle.clone(); block_on("docs", async move { h.service - .update(&h.cx, &h.collection, parsed_id, json) + .update(&h.cx, &h.collection, parsed_id, write_val) .await - }) + })?; + run_after( + handle, + "update", + id, + Some(&written), + serde_json::Value::Null, + ) }, ); } @@ -210,15 +227,62 @@ fn register_delete(engine: &mut RhaiEngine) { engine.register_fn( "delete", |handle: &mut DocsHandle, id: &str| -> Result> { - let h = handle.clone(); let parsed_id = parse_doc_id(id)?; - block_on("docs", async move { + // Delete carries no value, so the before-hook never transforms. + run_before(handle, "delete", id, None)?; + let h = handle.clone(); + let was_present = block_on("docs", async move { h.service.delete(&h.cx, &h.collection, parsed_id).await - }) + })?; + run_after( + handle, + "delete", + id, + None, + serde_json::Value::Bool(was_present), + )?; + Ok(was_present) }, ); } +/// §9.4: run the before-op interceptor for a per-app docs write. +fn run_before( + handle: &DocsHandle, + op: &'static str, + key: &str, + value: Option<&serde_json::Value>, +) -> Result, Box> { + super::interceptor::run_before( + &handle.ictx, + &handle.cx, + "docs", + op, + &handle.collection, + key, + value, + ) +} + +fn run_after( + handle: &DocsHandle, + op: &'static str, + key: &str, + value: Option<&serde_json::Value>, + result: serde_json::Value, +) -> Result<(), Box> { + super::interceptor::run_after( + &handle.ictx, + &handle.cx, + "docs", + op, + &handle.collection, + key, + value, + result, + ) +} + fn register_list(engine: &mut RhaiEngine) { // Zero-arg form: full page from the start. engine.register_fn( @@ -283,12 +347,22 @@ fn register_group_create(engine: &mut RhaiEngine) { engine.register_fn( "create", |handle: &mut GroupDocsHandle, data: Map| -> Result> { - let h = handle.clone(); let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?; + let write_val = group_run_before(handle, "create", "", Some(&json))?.unwrap_or(json); + let written = write_val.clone(); + let h = handle.clone(); let id = block_on("docs", async move { - h.service.create(&h.cx, &h.collection, json).await + h.service.create(&h.cx, &h.collection, write_val).await })?; - Ok(id.to_string()) + let id_str = id.to_string(); + group_run_after( + handle, + "create", + "", + Some(&written), + serde_json::Value::String(id_str.clone()), + )?; + Ok(id_str) }, ); } @@ -342,14 +416,23 @@ fn register_group_update(engine: &mut RhaiEngine) { engine.register_fn( "update", |handle: &mut GroupDocsHandle, id: &str, data: Map| -> Result<(), Box> { - let h = handle.clone(); let parsed_id = parse_doc_id(id)?; let json = dynamic_to_json_capped(&Dynamic::from(data), MAX_JSON_MATERIALIZE_BYTES)?; + let write_val = group_run_before(handle, "update", id, Some(&json))?.unwrap_or(json); + let written = write_val.clone(); + let h = handle.clone(); block_on("docs", async move { h.service - .update(&h.cx, &h.collection, parsed_id, json) + .update(&h.cx, &h.collection, parsed_id, write_val) .await - }) + })?; + group_run_after( + handle, + "update", + id, + Some(&written), + serde_json::Value::Null, + ) }, ); } @@ -358,15 +441,62 @@ fn register_group_delete(engine: &mut RhaiEngine) { engine.register_fn( "delete", |handle: &mut GroupDocsHandle, id: &str| -> Result> { - let h = handle.clone(); let parsed_id = parse_doc_id(id)?; - block_on("docs", async move { + group_run_before(handle, "delete", id, None)?; + let h = handle.clone(); + let was_present = block_on("docs", async move { h.service.delete(&h.cx, &h.collection, parsed_id).await - }) + })?; + group_run_after( + handle, + "delete", + id, + None, + serde_json::Value::Bool(was_present), + )?; + Ok(was_present) }, ); } +/// §9.4: before/after hooks for a shared (group) docs write — mirror the per-app +/// helpers but over `GroupDocsHandle`. +fn group_run_before( + handle: &GroupDocsHandle, + op: &'static str, + key: &str, + value: Option<&serde_json::Value>, +) -> Result, Box> { + super::interceptor::run_before( + &handle.ictx, + &handle.cx, + "docs", + op, + &handle.collection, + key, + value, + ) +} + +fn group_run_after( + handle: &GroupDocsHandle, + op: &'static str, + key: &str, + value: Option<&serde_json::Value>, + result: serde_json::Value, +) -> Result<(), Box> { + super::interceptor::run_after( + &handle.ictx, + &handle.cx, + "docs", + op, + &handle.collection, + key, + value, + result, + ) +} + fn register_group_list(engine: &mut RhaiEngine) { engine.register_fn( "list", diff --git a/crates/executor-core/src/sdk/files.rs b/crates/executor-core/src/sdk/files.rs index 2f44c45..08eed57 100644 --- a/crates/executor-core/src/sdk/files.rs +++ b/crates/executor-core/src/sdk/files.rs @@ -37,9 +37,9 @@ pub struct FilesHandle { collection: String, service: Arc, cx: Arc, - // §9.4 M1 plumbing: carried so `create`/`update`/`delete` can run a before/ - // after hook in a later milestone (M9). Unused until then. - #[allow(dead_code)] + // §9.4: carried so `create`/`update`/`delete` run the before/after hook (M9). + // Files carry bytes, so the hook sees service/op/collection/key only — the + // blob is never materialized into a `value`, and there is no data transform. ictx: InterceptorCtx, } @@ -52,8 +52,7 @@ pub struct GroupFilesHandle { collection: String, service: Arc, cx: Arc, - // §9.4 M1 plumbing (see `FilesHandle`). Unused until M9. - #[allow(dead_code)] + // §9.4: carried so shared `create`/`update`/`delete` run the hook too (M9). ictx: InterceptorCtx, } @@ -133,6 +132,9 @@ fn register_create(engine: &mut RhaiEngine) { let name = require_string(&meta, "name")?; let content_type = require_string(&meta, "content_type")?; let data = require_blob(&meta, "data")?; + // §9.4 before-op interceptor (allow/deny only — no transform for + // files, the blob is not surfaced to the hook). + run_before(handle, "create", "")?; let h = handle.clone(); let new = NewFile { name, @@ -142,7 +144,14 @@ fn register_create(engine: &mut RhaiEngine) { let id = block_on("files", async move { h.service.create(&h.cx, &h.collection, new).await })?; - Ok(id.to_string()) + let id_str = id.to_string(); + run_after( + handle, + "create", + "", + serde_json::Value::String(id_str.clone()), + )?; + Ok(id_str) }, ); } @@ -182,16 +191,18 @@ fn register_update(engine: &mut RhaiEngine) { let data = require_blob(&meta, "data")?; let name = optional_string(&meta, "name")?; let content_type = optional_string(&meta, "content_type")?; + run_before(handle, "update", id)?; let h = handle.clone(); - let id = id.to_string(); + let id_owned = id.to_string(); let upd = FileUpdate { data, name, content_type, }; block_on("files", async move { - h.service.update(&h.cx, &h.collection, &id, upd).await - }) + h.service.update(&h.cx, &h.collection, &id_owned, upd).await + })?; + run_after(handle, "update", id, serde_json::Value::Null) }, ); } @@ -200,15 +211,52 @@ fn register_delete(engine: &mut RhaiEngine) { engine.register_fn( "delete", |handle: &mut FilesHandle, id: &str| -> Result> { + run_before(handle, "delete", id)?; let h = handle.clone(); - let id = id.to_string(); - block_on("files", async move { - h.service.delete(&h.cx, &h.collection, &id).await - }) + let id_owned = id.to_string(); + let was_present = block_on("files", async move { + h.service.delete(&h.cx, &h.collection, &id_owned).await + })?; + run_after(handle, "delete", id, serde_json::Value::Bool(was_present))?; + Ok(was_present) }, ); } +/// §9.4: before/after hooks for a per-app files write. Files pass `value = None` +/// (the blob is never surfaced to the hook), so the before-hook only allows or +/// denies — it never transforms. +fn run_before(handle: &FilesHandle, op: &'static str, key: &str) -> Result<(), Box> { + super::interceptor::run_before( + &handle.ictx, + &handle.cx, + "files", + op, + &handle.collection, + key, + None, + ) + .map(|_| ()) +} + +fn run_after( + handle: &FilesHandle, + op: &'static str, + key: &str, + result: serde_json::Value, +) -> Result<(), Box> { + super::interceptor::run_after( + &handle.ictx, + &handle.cx, + "files", + op, + &handle.collection, + key, + None, + result, + ) +} + fn register_list(engine: &mut RhaiEngine) { engine.register_fn( "list", @@ -288,6 +336,7 @@ fn register_group_create(engine: &mut RhaiEngine) { let name = require_string(&meta, "name")?; let content_type = require_string(&meta, "content_type")?; let data = require_blob(&meta, "data")?; + group_run_before(handle, "create", "")?; let h = handle.clone(); let new = NewFile { name, @@ -297,7 +346,14 @@ fn register_group_create(engine: &mut RhaiEngine) { let id = block_on("files", async move { h.service.create(&h.cx, &h.collection, new).await })?; - Ok(id.to_string()) + let id_str = id.to_string(); + group_run_after( + handle, + "create", + "", + serde_json::Value::String(id_str.clone()), + )?; + Ok(id_str) }, ); } @@ -337,16 +393,18 @@ fn register_group_update(engine: &mut RhaiEngine) { let data = require_blob(&meta, "data")?; let name = optional_string(&meta, "name")?; let content_type = optional_string(&meta, "content_type")?; + group_run_before(handle, "update", id)?; let h = handle.clone(); - let id = id.to_string(); + let id_owned = id.to_string(); let upd = FileUpdate { data, name, content_type, }; block_on("files", async move { - h.service.update(&h.cx, &h.collection, &id, upd).await - }) + h.service.update(&h.cx, &h.collection, &id_owned, upd).await + })?; + group_run_after(handle, "update", id, serde_json::Value::Null) }, ); } @@ -355,15 +413,55 @@ fn register_group_delete(engine: &mut RhaiEngine) { engine.register_fn( "delete", |handle: &mut GroupFilesHandle, id: &str| -> Result> { + group_run_before(handle, "delete", id)?; let h = handle.clone(); - let id = id.to_string(); - block_on("files", async move { - h.service.delete(&h.cx, &h.collection, &id).await - }) + let id_owned = id.to_string(); + let was_present = block_on("files", async move { + h.service.delete(&h.cx, &h.collection, &id_owned).await + })?; + group_run_after(handle, "delete", id, serde_json::Value::Bool(was_present))?; + Ok(was_present) }, ); } +/// §9.4: before/after hooks for a shared (group) files write — mirror the +/// per-app helpers but over `GroupFilesHandle`. +fn group_run_before( + handle: &GroupFilesHandle, + op: &'static str, + key: &str, +) -> Result<(), Box> { + super::interceptor::run_before( + &handle.ictx, + &handle.cx, + "files", + op, + &handle.collection, + key, + None, + ) + .map(|_| ()) +} + +fn group_run_after( + handle: &GroupFilesHandle, + op: &'static str, + key: &str, + result: serde_json::Value, +) -> Result<(), Box> { + super::interceptor::run_after( + &handle.ictx, + &handle.cx, + "files", + op, + &handle.collection, + key, + None, + result, + ) +} + fn register_group_list(engine: &mut RhaiEngine) { engine.register_fn( "list", diff --git a/crates/executor-core/src/sdk/http.rs b/crates/executor-core/src/sdk/http.rs index 81b83f8..dcb4555 100644 --- a/crates/executor-core/src/sdk/http.rs +++ b/crates/executor-core/src/sdk/http.rs @@ -51,29 +51,31 @@ const MAX_REDIRECTS: i64 = 10; const ALLOWED_OPT_KEYS: [&str; 4] = ["headers", "timeout_ms", "follow_redirects", "max_redirects"]; -// `http` registers its verbs as module native fns (no handle struct), so there -// is nowhere to stash the interceptor ctx yet. The param is threaded uniformly -// from `register_all` (§9.4 M1) and consumed by M11, which adds the before/after -// hook around the outbound-request closures. Prefixed `_` until then. +// `http` registers its verbs as module native fns (no handle struct), so the +// interceptor ctx is captured into each request closure directly. Every verb +// funnels through `invoke` / `invoke_form`, which run the §9.4 before/after hook +// around `svc.request` (M11). The hook sees `service = "http"`, `op = +// "request"`, `collection = `, `key = `; the body is never +// surfaced, so there is no data transform. pub(super) fn register( engine: &mut RhaiEngine, services: &Services, cx: Arc, - _ictx: InterceptorCtx, + ictx: InterceptorCtx, ) { let svc = services.http.clone(); let mut module = Module::new(); // Bodyless verbs: (url) / (url, opts). for verb in ["get", "head"] { - register_bodyless(&mut module, verb, &svc, &cx); + register_bodyless(&mut module, verb, &svc, &cx, &ictx); } // Body verbs: (url) / (url, body) / (url, body, opts). for verb in ["post", "put", "patch", "delete"] { - register_body(&mut module, verb, &svc, &cx); + register_body(&mut module, verb, &svc, &cx, &ictx); } - register_post_form(&mut module, &svc, &cx); - register_request(&mut module, &svc, &cx); + register_post_form(&mut module, &svc, &cx, &ictx); + register_request(&mut module, &svc, &cx, &ictx); engine.register_static_module("http", module.into()); } @@ -83,17 +85,18 @@ fn register_bodyless( verb: &'static str, svc: &Arc, cx: &Arc, + ictx: &InterceptorCtx, ) { { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn(verb, move |url: &str| { - invoke(&svc, &cx, verb, url, None, None) + invoke(&ictx, &svc, &cx, verb, url, None, None) }); } { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn(verb, move |url: &str, opts: Map| { - invoke(&svc, &cx, verb, url, None, Some(&opts)) + invoke(&ictx, &svc, &cx, verb, url, None, Some(&opts)) }); } } @@ -103,61 +106,72 @@ fn register_body( verb: &'static str, svc: &Arc, cx: &Arc, + ictx: &InterceptorCtx, ) { { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn(verb, move |url: &str| { - invoke(&svc, &cx, verb, url, None, None) + invoke(&ictx, &svc, &cx, verb, url, None, None) }); } { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn(verb, move |url: &str, body: Dynamic| { - invoke(&svc, &cx, verb, url, Some(body), None) + invoke(&ictx, &svc, &cx, verb, url, Some(body), None) }); } { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn(verb, move |url: &str, body: Dynamic, opts: Map| { - invoke(&svc, &cx, verb, url, Some(body), Some(&opts)) + invoke(&ictx, &svc, &cx, verb, url, Some(body), Some(&opts)) }); } } -fn register_post_form(module: &mut Module, svc: &Arc, cx: &Arc) { +fn register_post_form( + module: &mut Module, + svc: &Arc, + cx: &Arc, + ictx: &InterceptorCtx, +) { { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn("post_form", move |url: &str, form: Map| { - invoke_form(&svc, &cx, url, &form, None) + invoke_form(&ictx, &svc, &cx, url, &form, None) }); } { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn("post_form", move |url: &str, form: Map, opts: Map| { - invoke_form(&svc, &cx, url, &form, Some(&opts)) + invoke_form(&ictx, &svc, &cx, url, &form, Some(&opts)) }); } } -fn register_request(module: &mut Module, svc: &Arc, cx: &Arc) { +fn register_request( + module: &mut Module, + svc: &Arc, + cx: &Arc, + ictx: &InterceptorCtx, +) { { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn("request", move |method: &str, url: &str| { - invoke(&svc, &cx, method, url, None, None) + invoke(&ictx, &svc, &cx, method, url, None, None) }); } { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn("request", move |method: &str, url: &str, body: Dynamic| { - invoke(&svc, &cx, method, url, Some(body), None) + invoke(&ictx, &svc, &cx, method, url, Some(body), None) }); } { - let (svc, cx) = (svc.clone(), cx.clone()); + let (svc, cx, ictx) = (svc.clone(), cx.clone(), ictx.clone()); module.set_native_fn( "request", move |method: &str, url: &str, body: Dynamic, opts: Map| { - invoke(&svc, &cx, method, url, Some(body), Some(&opts)) + invoke(&ictx, &svc, &cx, method, url, Some(body), Some(&opts)) }, ); } @@ -260,8 +274,9 @@ fn dispatch_body(body: Dynamic) -> Result> { Ok((Some(bytes), Some("application/json".to_string()))) } -#[allow(clippy::needless_pass_by_value)] +#[allow(clippy::needless_pass_by_value, clippy::too_many_arguments)] fn invoke( + ictx: &InterceptorCtx, svc: &Arc, cx: &Arc, method: &str, @@ -271,6 +286,9 @@ fn invoke( ) -> Result> { let opts = parse_opts(opts)?; let method_uc = method.to_ascii_uppercase(); + // §9.4 before-op interceptor (allow/deny only; the body is not surfaced to + // the hook). `collection` = the HTTP method, `key` = the URL. + super::interceptor::run_before(ictx, cx, "http", "request", &method_uc, url, None)?; let bodyless = matches!(method_uc.as_str(), "GET" | "HEAD"); let (encoded, content_type) = if bodyless { (None, None) @@ -281,7 +299,7 @@ fn invoke( }; let req = HttpRequest { - method: method_uc, + method: method_uc.clone(), url: url.to_string(), headers: opts.headers, body: encoded, @@ -296,11 +314,22 @@ fn invoke( let cx = cx.clone(); block_on("http", async move { svc.request(&cx, req).await })? }; + super::interceptor::run_after( + ictx, + cx, + "http", + "request", + &method_uc, + url, + None, + serde_json::Value::Null, + )?; Ok(response_to_dynamic(&resp)) } #[allow(clippy::needless_pass_by_value)] fn invoke_form( + ictx: &InterceptorCtx, svc: &Arc, cx: &Arc, url: &str, @@ -308,6 +337,8 @@ fn invoke_form( opts: Option<&Map>, ) -> Result> { let opts = parse_opts(opts)?; + // §9.4 before-op interceptor. `post_form` is always a POST. + super::interceptor::run_before(ictx, cx, "http", "request", "POST", url, None)?; let mut serializer = url::form_urlencoded::Serializer::new(String::new()); for (k, v) in form { serializer.append_pair(k.as_str(), &dyn_to_string(v)); @@ -330,6 +361,16 @@ fn invoke_form( let cx = cx.clone(); block_on("http", async move { svc.request(&cx, req).await })? }; + super::interceptor::run_after( + ictx, + cx, + "http", + "request", + "POST", + url, + None, + serde_json::Value::Null, + )?; Ok(response_to_dynamic(&resp)) } diff --git a/crates/executor-core/src/sdk/pubsub.rs b/crates/executor-core/src/sdk/pubsub.rs index 8a61c42..c098985 100644 --- a/crates/executor-core/src/sdk/pubsub.rs +++ b/crates/executor-core/src/sdk/pubsub.rs @@ -38,16 +38,39 @@ pub(super) fn register( { let svc = svc.clone(); let cx = cx.clone(); + let ictx = ictx.clone(); module.set_native_fn( "publish_durable", move |topic: &str, message: Dynamic| -> Result<(), Box> { crate::sdk::emit_budget::charge_emission("pubsub::publish_durable")?; let json = message_to_json(&message)?; - let svc = svc.clone(); - let cx = cx.clone(); + // §9.4 before-op interceptor (allow/deny + M4 data-transform). + let write_val = super::interceptor::run_before( + &ictx, + &cx, + "pubsub", + "publish", + topic, + "", + Some(&json), + )? + .unwrap_or(json); + let written = write_val.clone(); + let svc2 = svc.clone(); + let cx2 = cx.clone(); block_on("pubsub", async move { - svc.publish_durable(&cx, topic, json).await - }) + svc2.publish_durable(&cx2, topic, write_val).await + })?; + super::interceptor::run_after( + &ictx, + &cx, + "pubsub", + "publish", + topic, + "", + Some(&written), + Json::Null, + ) }, ); } @@ -112,9 +135,7 @@ pub struct GroupTopicHandle { namespace: String, service: Arc, cx: Arc, - // §9.4 M1 plumbing: carried so `publish` can run a before/after hook in a - // later milestone (M11). Unused until then. - #[allow(dead_code)] + // §9.4: carried so shared `publish` runs the before/after hook too (M11). ictx: InterceptorCtx, } @@ -125,16 +146,41 @@ fn shared_publish( ) -> Result<(), Box> { crate::sdk::emit_budget::charge_emission("pubsub::shared_topic")?; let json = message_to_json(&message)?; + // §9.4 before-op interceptor also guards shared-topic publishes. The + // collection is the namespace; the subtopic is the key. + let write_val = super::interceptor::run_before( + &h.ictx, + &h.cx, + "pubsub", + "publish", + &h.namespace, + subtopic, + Some(&json), + )? + .unwrap_or(json); + let written = write_val.clone(); let service = h.service.clone(); let cx = h.cx.clone(); let namespace = h.namespace.clone(); - let subtopic = subtopic.to_string(); + let subtopic_owned = subtopic.to_string(); block_on("pubsub", async move { - service.publish(&cx, &namespace, &subtopic, json).await + service + .publish(&cx, &namespace, &subtopic_owned, write_val) + .await }) // The fan-out count isn't surfaced to scripts (fire-and-forget, like the // per-app publish). - .map(|_n: u32| ()) + .map(|_n: u32| ())?; + super::interceptor::run_after( + &h.ictx, + &h.cx, + "pubsub", + "publish", + &h.namespace, + subtopic, + Some(&written), + Json::Null, + ) } fn register_shared_publish(engine: &mut RhaiEngine) { diff --git a/crates/executor-core/src/sdk/queue.rs b/crates/executor-core/src/sdk/queue.rs index b0ac3b9..42bedaf 100644 --- a/crates/executor-core/src/sdk/queue.rs +++ b/crates/executor-core/src/sdk/queue.rs @@ -43,12 +43,11 @@ pub(super) fn register( { let svc = svc.clone(); let cx = cx.clone(); + let ictx = ictx.clone(); module.set_native_fn( "enqueue", move |name: &str, message: Dynamic| -> Result<(), Box> { - crate::sdk::emit_budget::charge_emission("queue::enqueue")?; - let json = message_to_json(&message)?; - enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default()) + app_enqueue(&ictx, &svc, &cx, name, &message, EnqueueOpts::default()) }, ); } @@ -58,13 +57,12 @@ pub(super) fn register( { let svc = svc.clone(); let cx = cx.clone(); + let ictx = ictx.clone(); module.set_native_fn( "enqueue", move |name: &str, message: Dynamic, opts: Map| -> Result<(), Box> { - crate::sdk::emit_budget::charge_emission("queue::enqueue")?; - let json = message_to_json(&message)?; let opts = parse_opts(&opts)?; - enqueue_blocking(&svc, &cx, name, json, opts) + app_enqueue(&ictx, &svc, &cx, name, &message, opts) }, ); } @@ -134,12 +132,42 @@ pub struct GroupQueueHandle { collection: String, service: Arc, cx: Arc, - // §9.4 M1 plumbing: carried so `enqueue` can run a before/after hook in a - // later milestone (M10). Unused until then. - #[allow(dead_code)] + // §9.4: carried so shared `enqueue` runs the before/after hook too (M10). ictx: InterceptorCtx, } +/// Per-app enqueue with the §9.4 interceptor around it: charge the emission +/// budget, run the before-hook (allow/deny + M4 data-transform), enqueue the +/// (possibly rewritten) message, then run the after-hook with the new message +/// id as the result. `key` is `""` (a queue has no key). `collection` = the +/// queue name. +fn app_enqueue( + ictx: &InterceptorCtx, + svc: &Arc, + cx: &Arc, + name: &str, + message: &Dynamic, + opts: EnqueueOpts, +) -> Result<(), Box> { + crate::sdk::emit_budget::charge_emission("queue::enqueue")?; + let json = message_to_json(message)?; + let write_val = + super::interceptor::run_before(ictx, cx, "queue", "enqueue", name, "", Some(&json))? + .unwrap_or(json); + let written = write_val.clone(); + let id = enqueue_blocking(svc, cx, name, write_val, opts)?; + super::interceptor::run_after( + ictx, + cx, + "queue", + "enqueue", + name, + "", + Some(&written), + Json::String(id.0.to_string()), + ) +} + fn group_enqueue( h: &mut GroupQueueHandle, message: Dynamic, @@ -147,6 +175,18 @@ fn group_enqueue( ) -> Result<(), Box> { crate::sdk::emit_budget::charge_emission("queue::shared_collection")?; let json = message_to_json(&message)?; + // §9.4 before-op interceptor also guards shared-queue enqueues. + let write_val = super::interceptor::run_before( + &h.ictx, + &h.cx, + "queue", + "enqueue", + &h.collection, + "", + Some(&json), + )? + .unwrap_or(json); + let written = write_val.clone(); let handle = TokioHandle::try_current().map_err(|e| -> Box { EvalAltResult::ErrorRuntime( format!("queue: no tokio runtime available: {e}").into(), @@ -157,12 +197,21 @@ fn group_enqueue( let service = h.service.clone(); let cx = h.cx.clone(); let collection = h.collection.clone(); - handle - .block_on(async move { service.enqueue(&cx, &collection, json, opts).await }) - .map(|_id| ()) + let id = handle + .block_on(async move { service.enqueue(&cx, &collection, write_val, opts).await }) .map_err(|err| -> Box { EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into() - }) + })?; + super::interceptor::run_after( + &h.ictx, + &h.cx, + "queue", + "enqueue", + &h.collection, + "", + Some(&written), + Json::String(id.0.to_string()), + ) } fn group_depth(h: &mut GroupQueueHandle, f: F) -> Result> @@ -209,14 +258,15 @@ fn register_shared_queue_methods(engine: &mut RhaiEngine) { } /// Run the async enqueue inside the synchronous Rhai context. Mirrors -/// `pubsub::block_on` but for the queue's `Result<_, QueueError>`. +/// `pubsub::block_on` but for the queue's `Result<_, QueueError>`. Returns the +/// new message id (surfaced to the §9.4 after-hook, not to the script). fn enqueue_blocking( svc: &Arc, cx: &Arc, name: &str, payload: Json, opts: EnqueueOpts, -) -> Result<(), Box> { +) -> Result> { let handle = TokioHandle::try_current().map_err(|e| -> Box { EvalAltResult::ErrorRuntime( format!("queue: no tokio runtime available: {e}").into(), @@ -229,7 +279,6 @@ fn enqueue_blocking( let name = name.to_string(); handle .block_on(async move { svc.enqueue(&cx, &name, payload, opts).await }) - .map(|_id| ()) .map_err(|err| -> Box { EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into() }) diff --git a/crates/manager-core/src/apply_service.rs b/crates/manager-core/src/apply_service.rs index 9a273c5..13f783e 100644 --- a/crates/manager-core/src/apply_service.rs +++ b/crates/manager-core/src/apply_service.rs @@ -1292,22 +1292,37 @@ impl ApplyService { for w in &bundle.workflows { validate_workflow_definition(&w.name, &w.definition).map_err(ApplyError::Invalid)?; } - // §9.4 interceptors (MVP): `service = "kv"`, `op ∈ {set, delete}`, - // `phase ∈ {before, after}` (§9.4 M3), authored on app OR group. One + // §9.4 interceptors: a per-service allow-list of mutating ops + // (`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. + // + // INVARIANT: every `(service, op)` accepted here MUST have a matching + // runtime before/after hook in `executor-core::sdk::` — a + // validated-but-unhooked op would be a silent fail-open bypass. The + // hooks live in kv/docs/files/queue/pubsub/http (§9.4 M7–M11). + let allowed_ops: &[(&str, &[&str])] = &[ + ("kv", &["set", "delete"]), + ("docs", &["create", "update", "delete"]), + ("files", &["create", "update", "delete"]), + ("queue", &["enqueue"]), + ("pubsub", &["publish"]), + ("http", &["request"]), + ]; let mut seen_hooks: HashSet<(String, String, String)> = HashSet::new(); for i in &bundle.interceptors { - if i.service != "kv" { + let Some((_, ops)) = allowed_ops.iter().find(|(s, _)| *s == i.service) else { return Err(ApplyError::Invalid(format!( - "interceptor service `{}` is not supported — only `kv` (MVP)", + "interceptor service `{}` is not supported — one of kv / docs / files / queue / pubsub / http", i.service ))); - } - if i.op != "set" && i.op != "delete" { + }; + if !ops.contains(&i.op.as_str()) { return Err(ApplyError::Invalid(format!( - "interceptor op `{}` is not supported for kv — only `set` / `delete`", - i.op + "interceptor op `{}` is not supported for `{}` — one of {}", + i.op, + i.service, + ops.join(" / ") ))); } if i.phase != "before" && i.phase != "after" { diff --git a/crates/picloud-cli/src/manifest.rs b/crates/picloud-cli/src/manifest.rs index 32d499e..066fffc 100644 --- a/crates/picloud-cli/src/manifest.rs +++ b/crates/picloud-cli/src/manifest.rs @@ -78,7 +78,8 @@ pub struct Manifest { pub struct ManifestInterceptor { /// Name of the interceptor script (resolved on the node's chain at run time). pub script: String, - /// The guarded service. MVP: only `"kv"`. + /// The guarded service: one of `kv` / `docs` / `files` / `queue` / + /// `pubsub` / `http` (§9.4 M7–M11). Defaults to `kv`. #[serde(default = "default_interceptor_service")] pub service: String, /// The guarded operations, e.g. `["set", "delete"]`. diff --git a/crates/picloud-cli/tests/interceptors.rs b/crates/picloud-cli/tests/interceptors.rs index 6af1fd3..6313e35 100644 --- a/crates/picloud-cli/tests/interceptors.rs +++ b/crates/picloud-cli/tests/interceptors.rs @@ -398,6 +398,78 @@ fn a_malformed_verdict_fails_closed() { ); } +// --- §9.4 M8: docs (a representative non-KV service) ----------------------- + +/// A `(docs, create)` guard that denies creating a doc whose `data.kind` is +/// `"secret"`, reading the written value from `ctx.request.body.value`. Proves a +/// non-KV service funnels through the same before/after machinery and that the +/// created document's data is surfaced to the hook. +const DOCS_GUARD: &str = r#" +let op = ctx.request.body; +if op.action == "create" && op.value.kind == "secret" { + #{ allowed: false, reason: "secret docs are protected" } +} else { + #{ allowed: true } +} +"#; + +/// A writer that tries a denied `docs::create` (caught), then a permitted one, +/// and reports how many docs actually landed in the collection. +const DOCS_WRITER: &str = r#" +let denied = false; +try { docs::collection("c").create(#{ kind: "secret", n: 1 }); } catch(e) { denied = true; } +docs::collection("c").create(#{ kind: "public", n: 2 }); +let all = docs::collection("c").find(#{}); +#{ denied: denied, count: all.len() } +"#; + +/// §9.4 M8: `[[interceptors]] service = "docs" ops = ["create"]` guards +/// `docs::collection(...).create(...)` through the same before/after machinery +/// as KV — a guarded shape is denied (the doc is never created), an unguarded +/// one is allowed. +#[ignore = "needs DATABASE_URL pointing at a running Postgres"] +#[test] +fn an_interceptor_guards_docs_create() { + let Some(fx) = common::fixture_or_skip() else { + return; + }; + let env = common::admin_env(fx); + let app = common::unique_slug("icd-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"), DOCS_GUARD).unwrap(); + fs::write(dir.path().join("scripts/writer.rhai"), DOCS_WRITER).unwrap(); + fs::write( + dir.path().join("picloud.toml"), + format!( + "[app]\nslug = \"{app}\"\nname = \"ICD\"\n\n\ + [[scripts]]\nname = \"guard\"\nfile = \"scripts/guard.rhai\"\n\n\ + [[scripts]]\nname = \"writer\"\nfile = \"scripts/writer.rhai\"\n\n\ + [[interceptors]]\nscript = \"guard\"\nservice = \"docs\"\nops = [\"create\"]\n" + ), + ) + .unwrap(); + common::pic_as(&env) + .args(["apply", "--file"]) + .arg(dir.path().join("picloud.toml")) + .assert() + .success(); + + // The guarded (secret) create is denied; the public one lands — so exactly + // one doc exists in the collection. + let body = invoke_body(&env, &app_script_id(&env, &app, "writer")); + assert_eq!( + body, + serde_json::json!({ "denied": true, "count": 1 }), + "the `secret` docs::create must be denied; only the public doc is created" + ); +} + // --- §9.4 M2: ordered before-chain + cycle guard -------------------------- /// A group before-interceptor that denies only `kv::set` of key `"group-key"`.