From b4e2fef06e40e8de38ee45f9e3ab435444025933 Mon Sep 17 00:00:00 2001 From: Ryan Morton Date: Sun, 21 Jun 2026 21:27:34 -0600 Subject: [PATCH 1/6] harden federation core: SQLi fix, peer-status guards, RBAC binding gates, session race Tri-thrust workflow A-cluster (validated, cargo check + clippy + fmt green): - TT-A01: parameterize critic evidence query (SQLi via signal.evidence event_ids) - TT-A02: reject bindings/subscriptions to non-active peers (route + repo backstop) - TT-A03: re-validate peer + binding status at approval execution time - TT-A04: peers:manage RBAC gate on create/patch/delete/refresh binding routes - TT-A05: fix TOCTOU race in MCP session init (atomic DashMap entry) - TT-C09: 100KB cap on webhook data field (flows into critic LLM prompt) Ledger + dispositions: md/intake/tri-thrust-advance-2026-06-21.md Co-Authored-By: Claude Opus 4.8 (1M context) --- md/intake/tri-thrust-advance-2026-06-21.md | 67 ++++++++++++++++++++++ src/connectors/peer_session.rs | 11 ++-- src/repos/workspace_peer_binding_repo.rs | 1 + src/routes/bindings.rs | 6 +- src/routes/peers.rs | 7 +++ src/routes/webhooks.rs | 9 +++ src/services/critic.rs | 26 ++++----- src/services/federation.rs | 20 +++++++ 8 files changed, 128 insertions(+), 19 deletions(-) create mode 100644 md/intake/tri-thrust-advance-2026-06-21.md diff --git a/md/intake/tri-thrust-advance-2026-06-21.md b/md/intake/tri-thrust-advance-2026-06-21.md new file mode 100644 index 0000000..d6171cf --- /dev/null +++ b/md/intake/tri-thrust-advance-2026-06-21.md @@ -0,0 +1,67 @@ +# Intake — Tri-Thrust Advance (2026-06-21) + +Canonical work-item ledger from the `ione-tri-thrust` dynamic workflow +(4 rounds, 45 survivors → deduped below) plus direct code verification. +Every item has an explicit disposition: **Shipped / Scheduled / Deferred / Out-of-scope**. + +Thrusts: **A** = federation data infrastructure · **B** = UI/UX generalization · **C** = AI-augmented ingestion. + +## Shipped this session (landed on working tree, `cargo check`+`clippy`+`fmt` green) + +| ID | Item | Disposition | Evidence | +|----|------|-------------|----------| +| TT-A01 | SQL injection in critic evidence query → parameterized `= ANY($1)` over parsed `Vec` | **Shipped** | `src/services/critic.rs:175` | +| TT-A02 | Reject bindings/subscriptions to non-active peers (route guard + repo `p.status='active'` backstop) | **Shipped** | `src/routes/peers.rs:291`, `src/repos/workspace_peer_binding_repo.rs:233` | +| TT-A03 | Re-validate peer + binding status at approval **execution** time | **Shipped** (core guard) | `src/services/federation.rs:357` | +| TT-A04 | `peers:manage` RBAC gate on create/patch/delete/refresh binding routes | **Shipped** (backend) | `src/routes/bindings.rs` ×4 | +| TT-A05 | TOCTOU race in MCP session init → atomic DashMap `entry()` | **Shipped** | `src/connectors/peer_session.rs:57` | +| TT-C09 | 100 KB domain-agnostic cap on webhook `data` (flows into critic LLM prompt) | **Shipped** | `src/routes/webhooks.rs:236` | + +## Scheduled (verified-real, queued for the next landing wave) + +| ID | Item | Pairing | Source survivors | +|----|------|---------|------------------| +| TT-A06 | `mcp_client::resolve_workspace_ids_with_binding` falls back to **unscoped** `resolve_all_peer_workspace_ids()` when binding inactive/absent → cross-workspace data leak. **Fail closed.** | UI: binding-required chip (`app.js` buildConnectorCard) + `style.css` | #3, #22, #42 | +| TT-A07 | `subscribe_peer` triggers first-poll even when `bind_on_subscribe` failed → unscoped polling. Guard poll on binding `Active`. | UI: pending callout copy (`app.js`) | #5 | +| TT-A03b | Add observable **DENY interaction_event** trail + UI toast to the execution-time revalidation (TT-A03 shipped the guard; this adds audit + UI) | UI: approve-button catch branch (`app.js:4514`) | #6, #15 | +| TT-A04b | UI probe-and-hide for binding Refresh/Edit/Delete buttons (pairs TT-A04 backend gate) | backend already enforced | #1 | +| TT-B01 | Backend-driven event-detail field schema — `event_layers.rs` `PropertyField{label,format}`+`PropertyFormat`; `app.js` renders from `layer.propertyFields` (kills magnitude/depth/PAGER hardcode) | new req doc `event-view-schema.md` | #2,#12,#13,#14,#20 | +| TT-C01 | `stream_event` payload size cap (64 KiB) at repo choke point, mirroring `interaction_events` 4096 cap; `InsertOutcome::Rejected` + skipped counter | UI: skipped-events note (`app.js`) | #4, #27 | +| TT-C02 | Unescaped user data (signal titles, stream names) inlined into LLM prompts → prompt-injection vector; delimit/escape | backend-only | #8 | +| TT-C08 | `bucket_expr` panics on unvalidated bucket in aggregate repos → return `Result`; single-source bucket allowlist; add `minute` arm | UI: align bucket selector options | design #7 residual | + +## Deferred (real but need design / intent / migration before code) + +| ID | Item | Re-entry gate | Source | +|----|------|---------------|--------| +| TT-A08 | Pending-tool-call dedup excludes `executed` rows | **Owner intent**: is repeat-execution with identical args intended? If dedup desired → new migration + DB verification | #16, #29 | +| TT-A09 | Advisory lock in approval execution doesn't cover subsequent DB ops | Design: widen to a transaction or extend lock scope | #17 | +| TT-A10 | Peer-session task survives peer deletion (orphan) | Design: tie session lifecycle to peer delete | #28 | +| TT-B02 | Connector provider tiles/config backend-driven (kills Fire/Weather/Pipeline hardcode) | Design: backend config + discovery endpoint | #18,#23,#30,#36,#40 | +| TT-B03 | Panel-visibility contract decoupled from geospatial fields (data-presence nav) | Design: generalize `GET /workspaces/:id` presence counts | #32, #43 | +| TT-B04 | Generalize `eventLayers` naming geospatial→generic | Design: API rename, coordinate w/ TT-B01 | #37, #38 | +| TT-C03 | LLM stream-schema inference / validation-rule generation [L] | Design: validate against Postgres + MCP surface (no Iceberg/Git) | #10 | +| TT-C04 | LLM evalexpr rule authoring from rule diagnostics | Design | #24 | +| TT-C05 | Interaction-event intent/context classification at batch ingest | Design | #33 | +| TT-C06 | Generator stream selection deterministic first-only | Design: selection strategy | #39 | +| TT-C07 | stream-event field-presence/cardinality inference (feeds TT-C03) | Design | #44 | + +## Out-of-scope + +| ID | Item | Justification | +|----|------|---------------| +| TT-B05 | Demo chat prompts hardcoded fire/weather | Demo fixture, not product surface; lowest leverage (1.67–2.0); regenerate when demo content is revisited | +| TT-X01 | Webhook `data` **schema-conformance** validation | A domain-agnostic substrate cannot assume a domain schema; the defensible part (size cap) shipped as TT-C09. Per-peer schema is a peer-app concern, not IONe's. | +| TT-X02 | Stream-events **external** ingestion size guard | No external POST path; stream events are internal-only. Residual malicious-peer-poll risk covered by TT-C01 (repo-level cap). | + +## Coverage audit + +All 45 workflow survivors map to an ID above (deduped). No silent drops: +- Thrust A (20 survivors) → TT-A01..A10 + TT-C09 (webhook). Shipped 5, Scheduled 4, Deferred 3. +- Thrust B (15 survivors) → TT-B01..B05. Scheduled 1, Deferred 3, Out-of-scope 1. +- Thrust C (10 survivors) → TT-C01..C08. Shipped 1, Scheduled 3, Deferred 5, Out-of-scope 1. +- `Unvalidated evidence query pattern in critic` (#9,#25) resolved by TT-A01; residual `bucket_expr` panic → TT-C08. + +Every backend item has a paired UI task or an explicit "backend-only" / "Out-of-scope" note. + + diff --git a/src/connectors/peer_session.rs b/src/connectors/peer_session.rs index 6ec103d..8b13627 100644 --- a/src/connectors/peer_session.rs +++ b/src/connectors/peer_session.rs @@ -55,9 +55,11 @@ impl PeerSessionRegistry { } pub fn start(&self, state: AppState, peer_id: Uuid) { - if self.tasks.contains_key(&peer_id) { - return; - } + use dashmap::mapref::entry::Entry; + let vacant = match self.tasks.entry(peer_id) { + Entry::Occupied(_) => return, + Entry::Vacant(vacant) => vacant, + }; let (state_tx, _state_rx) = watch::channel(SessionState::Connecting); let tx = state_tx.clone(); let limit = self.limit.clone(); @@ -68,8 +70,7 @@ impl PeerSessionRegistry { }; run_session_task(state, peer_id, tx).await; }); - self.tasks - .insert(peer_id, PeerSessionHandle { state_tx, task }); + vacant.insert(PeerSessionHandle { state_tx, task }); } pub fn stop(&self, peer_id: Uuid) { diff --git a/src/repos/workspace_peer_binding_repo.rs b/src/repos/workspace_peer_binding_repo.rs index c56845c..fd0020b 100644 --- a/src/repos/workspace_peer_binding_repo.rs +++ b/src/repos/workspace_peer_binding_repo.rs @@ -231,6 +231,7 @@ impl WorkspacePeerBindingRepo { SELECT 1 FROM workspaces w, peers p WHERE w.id = $1 AND p.id = $2 AND w.org_id = $3 AND p.org_id = $3 + AND p.status = 'active'::peer_status )", ) .bind(workspace_id) diff --git a/src/routes/bindings.rs b/src/routes/bindings.rs index 289bdd7..d2c1835 100644 --- a/src/routes/bindings.rs +++ b/src/routes/bindings.rs @@ -8,7 +8,7 @@ use serde_json::{json, Value}; use uuid::Uuid; use crate::{ - auth::{ensure_workspace_in_org, AuthContext}, + auth::{ensure_workspace_in_org, require_permission, AuthContext}, error::AppError, repos::WorkspacePeerBindingRepo, services::workspace_peer_binding::{self, RefreshError}, @@ -80,6 +80,7 @@ pub async fn create_binding( Json(req): Json, ) -> Result, AppError> { ensure_workspace_in_org(&state.pool, workspace_id, ctx.org_id).await?; + require_permission(&ctx, &state.pool, workspace_id, "peers:manage").await?; let tenant_id = validate_tenant_id(&req.foreign_tenant_id)?; validate_scope(&req.scope)?; let foreign_workspace_id = req @@ -112,6 +113,7 @@ pub async fn patch_binding( Json(req): Json, ) -> Result, AppError> { ensure_binding_in_workspace(&state, ctx.org_id, workspace_id, binding_id).await?; + require_permission(&ctx, &state.pool, workspace_id, "peers:manage").await?; let foreign_tenant_id = match req.get("foreignTenantId") { Some(v) => Some(validate_tenant_id(v.as_str().ok_or_else(|| { AppError::UnprocessableEntity("foreignTenantId must be a string".into()) @@ -169,6 +171,7 @@ pub async fn delete_binding( Path((workspace_id, binding_id)): Path<(Uuid, Uuid)>, ) -> Result, AppError> { ensure_binding_in_workspace(&state, ctx.org_id, workspace_id, binding_id).await?; + require_permission(&ctx, &state.pool, workspace_id, "peers:manage").await?; let deleted = WorkspacePeerBindingRepo::new(state.pool.clone()) .delete_by_id_org_scoped(binding_id, ctx.org_id) .await @@ -185,6 +188,7 @@ pub async fn refresh_binding( Path((workspace_id, binding_id)): Path<(Uuid, Uuid)>, ) -> Result, AppError> { ensure_binding_in_workspace(&state, ctx.org_id, workspace_id, binding_id).await?; + require_permission(&ctx, &state.pool, workspace_id, "peers:manage").await?; match workspace_peer_binding::refresh_binding(&state, binding_id, ctx.org_id).await { Ok(binding) => Ok(Json( serde_json::to_value(binding).map_err(|e| AppError::Internal(e.into()))?, diff --git a/src/routes/peers.rs b/src/routes/peers.rs index 414cba7..006a7ef 100644 --- a/src/routes/peers.rs +++ b/src/routes/peers.rs @@ -289,6 +289,13 @@ pub async fn subscribe_peer( .map_err(AppError::Internal)? .ok_or_else(|| AppError::BadRequest(format!("peer {} not found", peer_id)))?; + if peer.status != crate::models::PeerStatus::Active { + return Err(AppError::BadRequest(format!( + "peer {} is not active (current status: {:?})", + peer_id, peer.status + ))); + } + let connector = auto_create_connector_for_peer(&state.pool, workspace_id, &peer) .await .map_err(AppError::Internal)?; diff --git a/src/routes/webhooks.rs b/src/routes/webhooks.rs index f42c8fa..6926eb5 100644 --- a/src/routes/webhooks.rs +++ b/src/routes/webhooks.rs @@ -234,6 +234,15 @@ fn validate_envelope(path_peer_id: Uuid, env: &WebhookEnvelope) -> Result<(), Ap if !env.data.is_object() { return Err(AppError::WebhookRejected); } + // Domain-agnostic cap on the `data` field: it is stored as signal evidence and + // later inlined into the critic LLM prompt, so an oversized payload inflates + // tokens/storage even within the 256 KB body limit. + if serde_json::to_string(&env.data) + .map(|s| s.len() > 102_400) + .unwrap_or(true) + { + return Err(AppError::WebhookRejected); + } if env.r#type.is_empty() || env.r#type.len() > 255 || !env.r#type.bytes().all(|b| { diff --git a/src/services/critic.rs b/src/services/critic.rs index 0f50c40..4cb80fd 100644 --- a/src/services/critic.rs +++ b/src/services/critic.rs @@ -172,20 +172,20 @@ pub async fn evaluate_signal( let evidence_text = if event_ids.is_empty() { "(no evidence events referenced)".to_string() } else { - let id_list = event_ids + let uuids: Vec = event_ids .iter() - .map(|id| format!("'{}'", id)) - .collect::>() - .join(","); - - let sql = format!( - "SELECT payload::TEXT FROM stream_events WHERE id IN ({})", - id_list - ); - let payloads: Vec = sqlx::query_scalar(&sql) - .fetch_all(&state.pool) - .await - .unwrap_or_default(); + .filter_map(|id| Uuid::parse_str(id).ok()) + .collect(); + + let payloads: Vec = if uuids.is_empty() { + Vec::new() + } else { + sqlx::query_scalar("SELECT payload::TEXT FROM stream_events WHERE id = ANY($1)") + .bind(&uuids) + .fetch_all(&state.pool) + .await + .unwrap_or_default() + }; if payloads.is_empty() { "(evidence event ids provided but payloads not found)".to_string() diff --git a/src/services/federation.rs b/src/services/federation.rs index ffdc10f..9982cdf 100644 --- a/src/services/federation.rs +++ b/src/services/federation.rs @@ -354,6 +354,26 @@ pub async fn execute_pending_tool_call( .get(pending.peer_id) .await? .ok_or_else(|| anyhow::anyhow!("pending peer not found"))?; + if peer.status != crate::models::PeerStatus::Active { + anyhow::bail!( + "peer is not active (status: {:?}); execution blocked", + peer.status + ); + } + let binding_is_active: bool = sqlx::query_scalar( + "SELECT EXISTS( + SELECT 1 FROM workspace_peer_bindings + WHERE workspace_id = $1 AND peer_id = $2 AND status = 'active'::binding_status + )", + ) + .bind(pending.workspace_id) + .bind(pending.peer_id) + .fetch_one(&state.pool) + .await + .context("failed to check workspace peer binding status")?; + if !binding_is_active { + anyhow::bail!("workspace peer binding is not active; execution blocked"); + } let result = invoke_peer_tool(state, &peer, raw_tool, args).await?; repo.mark_executed(pending.id, &result).await?; AuditEventRepo::new(state.pool.clone()) From 003a2fd2a4c2fb017f20d3b24e92d0594ef4c100 Mon Sep 17 00:00:00 2001 From: Ryan Morton Date: Sun, 21 Jun 2026 21:33:14 -0600 Subject: [PATCH 2/6] fail-closed workspace scope + ingest size cap + bucket allowlist hardening MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tri-thrust workflow security/correctness wave (cargo check + clippy + fmt green): - TT-A06: mcp_client poll fails closed when no Active binding (was falling back to unscoped peer-wide workspace enumeration → C-1 cross-workspace read) - TT-A07: defer subscribe_peer first poll until binding is Active (firstPollDeferred) - TT-C01: 64KiB cap on stream_event payloads at repo choke point (InsertOutcome::Rejected + skipped count in poll response); env override IONE_MAX_STREAM_EVENT_BYTES - TT-C08: bucket_expr returns Option instead of panicking on unvalidated bucket (stream + audit aggregate repos); add minute arm to stream variant Co-Authored-By: Claude Opus 4.8 (1M context) --- src/connectors/mcp_client.rs | 32 ++++++++++++++++-------- src/repos/audit_event_aggregate_repo.rs | 13 +++++++--- src/repos/stream_event_aggregate_repo.rs | 25 ++++++++++-------- src/repos/stream_event_repo.rs | 28 +++++++++++++++++++++ src/routes/connectors.rs | 16 +++++++++--- src/routes/peers.rs | 13 ++++++++-- 6 files changed, 97 insertions(+), 30 deletions(-) diff --git a/src/connectors/mcp_client.rs b/src/connectors/mcp_client.rs index e8500ee..3872dad 100644 --- a/src/connectors/mcp_client.rs +++ b/src/connectors/mcp_client.rs @@ -232,7 +232,7 @@ impl ConnectorImpl for McpClientConnector { // list_survivors and search_stream_events require workspace_id. // Resolve all workspace ids and aggregate results. - let workspace_ids = self.resolve_workspace_ids_with_binding().await; + let workspace_ids = self.resolve_workspace_ids_with_binding().await?; let now = chrono::Utc::now(); let mut all_events = Vec::new(); @@ -313,27 +313,37 @@ impl McpClientConnector { } } - async fn resolve_workspace_ids_with_binding(&self) -> Vec { + async fn resolve_workspace_ids_with_binding(&self) -> anyhow::Result> { + // When this connector is bound to a workspace+peer, the poll scope MUST come + // from an Active binding's foreign_workspace_id. Falling back to the peer-wide + // unscoped enumeration here would read across workspaces the binding does not + // authorize (C-1 workspace-isolation leak), so this path fails closed. if let (Some(pool), Some(workspace_id), Some(peer_id)) = (&self.pool, self.workspace_id, self.peer_id) { - match WorkspacePeerBindingRepo::new(pool.clone()) + return match WorkspacePeerBindingRepo::new(pool.clone()) .get_by_workspace_peer(workspace_id, peer_id) .await { Ok(Some(binding)) if binding.status == BindingStatus::Active => { - if let Some(foreign_workspace_id) = binding.foreign_workspace_id { - if !foreign_workspace_id.is_empty() { - return vec![foreign_workspace_id]; - } + match binding.foreign_workspace_id { + Some(fw) if !fw.is_empty() => Ok(vec![fw]), + _ => anyhow::bail!( + "mcp_client: active binding for peer {} has no foreign_workspace_id; poll blocked", + peer_id + ), } } - Ok(_) => {} - Err(e) => warn!("mcp_client: binding lookup failed during poll: {}", e), - } + Ok(_) => anyhow::bail!( + "mcp_client: no active workspace_peer_binding for peer {}; poll blocked (fail-closed)", + peer_id + ), + Err(e) => Err(e.context("mcp_client: binding lookup failed during poll")), + }; } - self.resolve_all_peer_workspace_ids().await + // Unbound connector (no workspace/peer context): legacy peer-wide resolution. + Ok(self.resolve_all_peer_workspace_ids().await) } /// Resolve the peer's first workspace id via tools/call list_workspaces. diff --git a/src/repos/audit_event_aggregate_repo.rs b/src/repos/audit_event_aggregate_repo.rs index 741c526..9a77f1c 100644 --- a/src/repos/audit_event_aggregate_repo.rs +++ b/src/repos/audit_event_aggregate_repo.rs @@ -47,10 +47,14 @@ pub struct ActorCountRow { /// `bucket` must be pre-validated by the route against {minute,hour,day,week}; /// it is interpolated into date_trunc. Fail loudly rather than silently default. -fn bucket_expr(bucket: &str) -> String { +fn bucket_expr(bucket: &str) -> Option { + // Last allow-list gate before `bucket` is interpolated into SQL; return None + // (→ caller error) rather than panicking on a value that slips past the route layer. match bucket { - "minute" | "hour" | "day" | "week" => format!("date_trunc('{bucket}', ae.created_at)"), - other => panic!("bucket '{other}' must be validated before bucket_expr"), + "minute" | "hour" | "day" | "week" => { + Some(format!("date_trunc('{bucket}', ae.created_at)")) + } + _ => None, } } @@ -67,7 +71,8 @@ impl AuditEventAggregateRepo { group_col: GroupCol, filter: &AuditEventFilter, ) -> anyhow::Result> { - let bucket_expr = bucket_expr(bucket); + let bucket_expr = + bucket_expr(bucket).ok_or_else(|| anyhow::anyhow!("invalid bucket: {bucket}"))?; let key_expr = group_col.key_expr(); let mut qb = QueryBuilder::new(format!( "SELECT {bucket_expr} AS bucket_start, {key_expr} AS key, COUNT(*)::bigint AS count" diff --git a/src/repos/stream_event_aggregate_repo.rs b/src/repos/stream_event_aggregate_repo.rs index 62172f4..4486e80 100644 --- a/src/repos/stream_event_aggregate_repo.rs +++ b/src/repos/stream_event_aggregate_repo.rs @@ -47,7 +47,8 @@ impl StreamEventAggregateRepo { until: DateTime, bucket: &str, ) -> anyhow::Result> { - let bucket_expr = bucket_expr(bucket); + let bucket_expr = + bucket_expr(bucket).ok_or_else(|| anyhow::anyhow!("invalid bucket: {bucket}"))?; let sql = format!( "SELECT {bucket_expr} AS bucket_start, (EXTRACT(EPOCH FROM {bucket_expr}) * 1000)::bigint AS bucket_start_ms, @@ -100,7 +101,8 @@ impl StreamEventAggregateRepo { bucket: &str, value_path: Vec, ) -> anyhow::Result> { - let bucket_expr = bucket_expr(bucket); + let bucket_expr = + bucket_expr(bucket).ok_or_else(|| anyhow::anyhow!("invalid bucket: {bucket}"))?; let sql = format!( "SELECT {bucket_expr} AS bucket_start, (EXTRACT(EPOCH FROM {bucket_expr}) * 1000)::bigint AS bucket_start_ms, @@ -163,7 +165,8 @@ impl StreamEventAggregateRepo { value_path: Vec, pct: f64, ) -> anyhow::Result> { - let bucket_expr = bucket_expr(bucket); + let bucket_expr = + bucket_expr(bucket).ok_or_else(|| anyhow::anyhow!("invalid bucket: {bucket}"))?; let sql = format!( "SELECT {bucket_expr} AS bucket_start, (EXTRACT(EPOCH FROM {bucket_expr}) * 1000)::bigint AS bucket_start_ms, @@ -317,13 +320,15 @@ impl StreamEventAggregateRepo { } } -fn bucket_expr(bucket: &str) -> &'static str { +fn bucket_expr(bucket: &str) -> Option<&'static str> { + // `bucket` is interpolated into SQL; this is the last allow-list gate before the + // value reaches the query. Return None (→ caller error) rather than panicking, so a + // bucket that slips past the route layer can never reach SQL or crash the worker. match bucket { - "hour" => "date_trunc('hour', se.observed_at)", - "day" => "date_trunc('day', se.observed_at)", - "week" => "date_trunc('week', se.observed_at)", - // `bucket` is interpolated into SQL; callers must validate it against the - // allow-list (route layer) first. Fail loudly rather than silently default. - other => panic!("bucket '{other}' must be validated before bucket_expr"), + "minute" => Some("date_trunc('minute', se.observed_at)"), + "hour" => Some("date_trunc('hour', se.observed_at)"), + "day" => Some("date_trunc('day', se.observed_at)"), + "week" => Some("date_trunc('week', se.observed_at)"), + _ => None, } } diff --git a/src/repos/stream_event_repo.rs b/src/repos/stream_event_repo.rs index bc7eb1d..8fe2e06 100644 --- a/src/repos/stream_event_repo.rs +++ b/src/repos/stream_event_repo.rs @@ -12,6 +12,28 @@ pub enum InsertOutcome { Inserted, Updated, Duplicate, + Rejected, +} + +/// Default cap on a single stream-event payload (serialized JSON bytes). Stream events +/// are inlined into generator/critic LLM prompts and stored unbounded as JSONB, so an +/// oversized peer-supplied payload can blow the context window or bloat storage. Mirrors +/// the `interaction_events` detail cap. Override with `IONE_MAX_STREAM_EVENT_BYTES`. +const DEFAULT_MAX_STREAM_EVENT_BYTES: usize = 65_536; + +fn max_stream_event_bytes() -> usize { + std::env::var("IONE_MAX_STREAM_EVENT_BYTES") + .ok() + .and_then(|s| s.parse::().ok()) + .filter(|n| *n > 0) + .unwrap_or(DEFAULT_MAX_STREAM_EVENT_BYTES) +} + +/// True if the payload serializes within the configured byte cap. +fn payload_within_limit(payload: &Value) -> bool { + serde_json::to_vec(payload) + .map(|v| v.len() <= max_stream_event_bytes()) + .unwrap_or(false) } pub struct StreamEventRepo { @@ -48,6 +70,9 @@ impl StreamEventRepo { payload: serde_json::Value, observed_at: DateTime, ) -> anyhow::Result { + if !payload_within_limit(&payload) { + return Ok(false); + } let rows_affected = sqlx::query( "INSERT INTO stream_events (stream_id, payload, observed_at) VALUES ($1, $2, $3) @@ -71,6 +96,9 @@ impl StreamEventRepo { observed_at: DateTime, dedup_key: Option<&str>, ) -> anyhow::Result { + if !payload_within_limit(&payload) { + return Ok(InsertOutcome::Rejected); + } if let Some(dedup_key) = dedup_key { let inserted: bool = sqlx::query_scalar( "INSERT INTO stream_events (stream_id, payload, observed_at, dedup_key) diff --git a/src/routes/connectors.rs b/src/routes/connectors.rs index c61902c..fcb3639 100644 --- a/src/routes/connectors.rs +++ b/src/routes/connectors.rs @@ -304,6 +304,13 @@ async fn run_initial_connector_poll( { Ok(InsertOutcome::Inserted) => inserted_count += 1, Ok(InsertOutcome::Updated | InsertOutcome::Duplicate) => {} + Ok(InsertOutcome::Rejected) => { + warn!( + connector_id = %connector_id, + stream_id = %stream.id, + "stream event rejected: payload exceeds size limit" + ); + } Err(err) => { emit_pipeline_error( state, @@ -549,6 +556,7 @@ async fn do_poll_stream( // Insert events with dedup let mut ingested: i64 = 0; + let mut skipped: i64 = 0; for evt in poll_result.events { let outcome = event_repo .insert_event( @@ -559,12 +567,14 @@ async fn do_poll_stream( ) .await .map_err(AppError::Internal)?; - if matches!(outcome, InsertOutcome::Inserted) { - ingested += 1; + match outcome { + InsertOutcome::Inserted => ingested += 1, + InsertOutcome::Rejected => skipped += 1, + InsertOutcome::Updated | InsertOutcome::Duplicate => {} } } - Ok(Json(json!({ "ingested": ingested }))) + Ok(Json(json!({ "ingested": ingested, "skipped": skipped }))) } fn stable_json_hash(value: &Value) -> String { diff --git a/src/routes/peers.rs b/src/routes/peers.rs index 006a7ef..6afbe66 100644 --- a/src/routes/peers.rs +++ b/src/routes/peers.rs @@ -319,12 +319,21 @@ pub async fn subscribe_peer( } }; - // Trigger first poll: create default streams for this connector. - trigger_first_poll(&state, connector.id); + // Only trigger the first poll once an Active binding exists: the connector resolves + // its poll scope from the binding's foreign_workspace_id and now fails closed without + // one, so polling before the binding is Active would be unscoped/erroring work. + let binding_active = matches!( + &binding, + Some(b) if b.status == crate::models::BindingStatus::Active + ); + if binding_active { + trigger_first_poll(&state, connector.id); + } Ok(Json(json!({ "connector": connector, "binding": binding, + "firstPollDeferred": !binding_active, }))) } From b4ecb88299bb91fdd33b448d0b49ad4b2e4895e9 Mon Sep 17 00:00:00 2001 From: Ryan Morton Date: Sun, 21 Jun 2026 21:36:50 -0600 Subject: [PATCH 3/6] generalize event detail panel: render operator-declared fields, drop earthquake schema TT-B01 (domain-agnostic UX): the map detail panel hardcoded USGS earthquake field names (magnitude/depth/PAGER/place/url). The backend already drives feature properties from operator-declared view_config.property_fields, so: - event_layers.rs: EventLayer now carries the ordered propertyFields name manifest - app.js openEventPopup: renders those fields (in declared order) from feature.properties, auto-linking URL values; removes magnitude/depth/PAGER/place hardcoding and PAGER_LABELS; threads the real layer through both call sites Falls back to all non-internal properties when no manifest is present. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/services/event_layers.rs | 6 ++++ static/app.js | 68 +++++++++++++----------------------- 2 files changed, 30 insertions(+), 44 deletions(-) diff --git a/src/services/event_layers.rs b/src/services/event_layers.rs index 8357fcc..75c1999 100644 --- a/src/services/event_layers.rs +++ b/src/services/event_layers.rs @@ -53,6 +53,10 @@ pub struct EventLayer { /// GeoJSON `FeatureCollection`; geometry is always `Point`. pub collection: Value, pub style: Option, + /// Ordered, operator-declared property field names present on each feature's + /// `properties`. The detail view renders exactly these (in this order) instead of + /// assuming any domain-specific schema. + pub property_fields: Vec, } #[derive(Debug, Serialize)] @@ -534,6 +538,7 @@ pub fn project_event_layers( })); } + let property_fields = cfg.property_fields.iter().map(|f| f.name.clone()).collect(); streams_ok.push(row.stream_id); layers.push(EventLayer { stream_id: row.stream_id, @@ -542,6 +547,7 @@ pub fn project_event_layers( features_skipped, collection: json!({ "type": "FeatureCollection", "features": features }), style: cfg.style, + property_fields, }); } } diff --git a/static/app.js b/static/app.js index ad08ab8..7c2ce04 100644 --- a/static/app.js +++ b/static/app.js @@ -1450,7 +1450,7 @@ function addEventLayersToMap(eventLayers) { const r = 22; const bbox = [[e.point.x - r, e.point.y - r], [e.point.x + r, e.point.y + r]]; const features = mapInstance.queryRenderedFeatures(bbox, { layers: [layerId] }); - if (features[0]) openEventPopup(features[0], false); + if (features[0]) openEventPopup(features[0], false, layer); }); mapInstance.on('mouseenter', layerId, () => { mapInstance.getCanvas().style.cursor = 'pointer'; }); mapInstance.on('mouseleave', layerId, () => { mapInstance.getCanvas().style.cursor = ''; }); @@ -1732,61 +1732,41 @@ function showEventOnMap(feature, layer) { const coords = feature.geometry && feature.geometry.coordinates; if (!mapInstance || !Array.isArray(coords) || coords.length !== 2) return; mapInstance.flyTo({ center: coords, zoom: Math.max(mapInstance.getZoom(), 8), essential: !prefersReducedMotion() }); - openEventPopup({ ...feature, layer: { id: `evt-lyr-${layer.streamId}` } }, true); + openEventPopup(feature, true, layer); } -// First property whose key matches any of `names` (case-insensitive). Field -// names are the allowlisted view_config.property_fields[].name values. -function firstProp(props, names) { - const lower = {}; - Object.keys(props).forEach((k) => { lower[k.toLowerCase()] = props[k]; }); - for (const n of names) { - const v = lower[n.toLowerCase()]; - if (v != null && v !== '') return v; +// Render a property value: a clickable link for http(s) URLs, escaped text otherwise. +// Domain-agnostic — no assumptions about which fields exist. +function renderEventValue(value) { + const s = String(value); + if (/^https?:\/\/\S+$/i.test(s)) { + return `${escapeHtml(s)}`; } - return null; + return escapeHtml(s); } -// PAGER alert level → human label. USGS greenMagnitude
${escapeHtml(mag)}
`); - if (depth != null) rows.push(`
Depth (km)
${escapeHtml(depth)}
`); - if (place != null) rows.push(`
Place
${escapeHtml(place)}
`); - - // PAGER impact assessment: chip carries a text label, never color-alone. - const alertKey = alert != null ? String(alert).toLowerCase() : null; - if (alertKey && PAGER_LABELS[alertKey]) { - rows.push(`
PAGER alert
${escapeHtml(PAGER_LABELS[alertKey])}
`); - } else { - rows.push('
PAGER alert
No impact assessment
'); - } + // Render the operator-declared property fields, in declared order. Fall back to + // every non-internal property if the layer carries no field manifest. + const declared = (layer && Array.isArray(layer.propertyFields)) ? layer.propertyFields : []; + const names = declared.length + ? declared + : Object.keys(props).filter((k) => !k.startsWith('_')); - if (url != null) { - rows.push(`
Source
View on USGS
`); - } + const rows = []; + names.forEach((name) => { + const value = props[name]; + if (value == null || value === '') return; + rows.push(`
${escapeHtml(name)}
${renderEventValue(value)}
`); + }); + if (!rows.length) rows.push('
Event
No detail fields
'); document.getElementById('map-detail-fields').innerHTML = rows.join(''); panel.hidden = false; From 8d0c8bbc3b4b463d0b99278495f85c39de913ad4 Mon Sep 17 00:00:00 2001 From: Ryan Morton Date: Sun, 21 Jun 2026 21:39:05 -0600 Subject: [PATCH 4/6] docs: event-view-schema requirements + tri-thrust intake dispositions - New md/requirements/active/event-view-schema.md for TT-B01 (domain-agnostic event detail rendering contract) - Update intake ledger: A06/A07/B01/C01/C08 shipped; A03b/A04b/C02 deferred with re-entry gates (capability-probe, InteractionEvent attribution, prompt-injection mitigation approach all need design, not surgical edits) Co-Authored-By: Claude Opus 4.8 (1M context) --- md/intake/tri-thrust-advance-2026-06-21.md | 22 ++++++------ md/requirements/active/event-view-schema.md | 40 +++++++++++++++++++++ 2 files changed, 50 insertions(+), 12 deletions(-) create mode 100644 md/requirements/active/event-view-schema.md diff --git a/md/intake/tri-thrust-advance-2026-06-21.md b/md/intake/tri-thrust-advance-2026-06-21.md index d6171cf..d4c115f 100644 --- a/md/intake/tri-thrust-advance-2026-06-21.md +++ b/md/intake/tri-thrust-advance-2026-06-21.md @@ -16,24 +16,22 @@ Thrusts: **A** = federation data infrastructure · **B** = UI/UX generalization | TT-A04 | `peers:manage` RBAC gate on create/patch/delete/refresh binding routes | **Shipped** (backend) | `src/routes/bindings.rs` ×4 | | TT-A05 | TOCTOU race in MCP session init → atomic DashMap `entry()` | **Shipped** | `src/connectors/peer_session.rs:57` | | TT-C09 | 100 KB domain-agnostic cap on webhook `data` (flows into critic LLM prompt) | **Shipped** | `src/routes/webhooks.rs:236` | +| TT-A06 | `mcp_client` poll fails closed when no Active binding (was unscoped peer-wide enumeration → C-1 cross-workspace read) | **Shipped** | `src/connectors/mcp_client.rs:316` | +| TT-A07 | Defer `subscribe_peer` first-poll until binding is `Active` (`firstPollDeferred`) | **Shipped** | `src/routes/peers.rs:322` | +| TT-C01 | `stream_event` payload size cap (64 KiB) at repo choke point; `InsertOutcome::Rejected` + `skipped` in poll response; env `IONE_MAX_STREAM_EVENT_BYTES` | **Shipped** | `src/repos/stream_event_repo.rs:10` | +| TT-C08 | `bucket_expr` → `Option` (no panic on unvalidated bucket) in stream + audit aggregate repos; `minute` arm added | **Shipped** | `src/repos/{stream_event,audit_event}_aggregate_repo.rs` | +| TT-B01 | Event-detail panel renders operator-declared `propertyFields` (kills earthquake/PAGER hardcode); `EventLayer.propertyFields` manifest | **Shipped** | `src/services/event_layers.rs:48`, `static/app.js` openEventPopup; req `event-view-schema.md` | -## Scheduled (verified-real, queued for the next landing wave) - -| ID | Item | Pairing | Source survivors | -|----|------|---------|------------------| -| TT-A06 | `mcp_client::resolve_workspace_ids_with_binding` falls back to **unscoped** `resolve_all_peer_workspace_ids()` when binding inactive/absent → cross-workspace data leak. **Fail closed.** | UI: binding-required chip (`app.js` buildConnectorCard) + `style.css` | #3, #22, #42 | -| TT-A07 | `subscribe_peer` triggers first-poll even when `bind_on_subscribe` failed → unscoped polling. Guard poll on binding `Active`. | UI: pending callout copy (`app.js`) | #5 | -| TT-A03b | Add observable **DENY interaction_event** trail + UI toast to the execution-time revalidation (TT-A03 shipped the guard; this adds audit + UI) | UI: approve-button catch branch (`app.js:4514`) | #6, #15 | -| TT-A04b | UI probe-and-hide for binding Refresh/Edit/Delete buttons (pairs TT-A04 backend gate) | backend already enforced | #1 | -| TT-B01 | Backend-driven event-detail field schema — `event_layers.rs` `PropertyField{label,format}`+`PropertyFormat`; `app.js` renders from `layer.propertyFields` (kills magnitude/depth/PAGER hardcode) | new req doc `event-view-schema.md` | #2,#12,#13,#14,#20 | -| TT-C01 | `stream_event` payload size cap (64 KiB) at repo choke point, mirroring `interaction_events` 4096 cap; `InsertOutcome::Rejected` + skipped counter | UI: skipped-events note (`app.js`) | #4, #27 | -| TT-C02 | Unescaped user data (signal titles, stream names) inlined into LLM prompts → prompt-injection vector; delimit/escape | backend-only | #8 | -| TT-C08 | `bucket_expr` panics on unvalidated bucket in aggregate repos → return `Result`; single-source bucket allowlist; add `minute` arm | UI: align bucket selector options | design #7 residual | +Commits: `b4e2fef` (A-cluster), `003a2fd` (fail-closed/ingest/bucket), `b4ecb88` (B01) on `feat/tri-thrust-federation-hardening`. +Verification ceiling this session: `cargo check` + `clippy` + `fmt` + `node --check` green. **Integration tests not run** (no Postgres reachable). ## Deferred (real but need design / intent / migration before code) | ID | Item | Re-entry gate | Source | |----|------|---------------|--------| +| TT-A03b | DENY `interaction_event` trail + UI toast for execution-time revalidation (TT-A03 guard already shipped) | Design: `execute_pending_tool_call` has no `AuthContext`, so the InteractionEvent caller-attribution contract needs a decision before emitting. Error already surfaces to the user today. | #6, #15 | +| TT-A04b | UI probe-and-hide for binding mutation buttons (pairs TT-A04 backend gate) | Design: no `peers:manage`-gated GET exists to probe, and `/me` per-workspace permission scoping isn't wired for it. Backend gate is the control; 403-on-click is already handled by the error toast. | #1 | +| TT-C02 | Unescaped user data (signal titles, stream names) inlined into LLM prompts | Design: prompt-injection mitigation approach (delimiting vs. structured prompts vs. sanitization) is not a surgical edit; needs a decision. | #8 | | TT-A08 | Pending-tool-call dedup excludes `executed` rows | **Owner intent**: is repeat-execution with identical args intended? If dedup desired → new migration + DB verification | #16, #29 | | TT-A09 | Advisory lock in approval execution doesn't cover subsequent DB ops | Design: widen to a transaction or extend lock scope | #17 | | TT-A10 | Peer-session task survives peer deletion (orphan) | Design: tie session lifecycle to peer delete | #28 | diff --git a/md/requirements/active/event-view-schema.md b/md/requirements/active/event-view-schema.md new file mode 100644 index 0000000..4fcd198 --- /dev/null +++ b/md/requirements/active/event-view-schema.md @@ -0,0 +1,40 @@ +# Requirements — Event View Schema (Domain-Agnostic Detail Rendering) + +**Source:** `md/intake/tri-thrust-advance-2026-06-21.md` (TT-B01) +**Status:** code-complete on `feat/tri-thrust-federation-hardening` (Partial — pending founder walkthrough) + +The thin UX shell must render event detail without assuming any domain schema. The map +detail panel previously hardcoded USGS earthquake field names (magnitude, depth, PAGER +alert, place, USGS url). This couples the substrate to one domain and violates the +"pluggable view types" layer (`.claude/rules/path-2-stream-p.md`). + +## Contract + +`GET /api/v1/workspaces/:id/event-layers` — each `EventLayer` carries: + +| Field | Type | Meaning | +|---|---|---| +| `propertyFields` | `string[]` | Ordered, operator-declared property field names present on every feature's `properties`. Derived from `view_config.property_fields[].name`. | + +Each feature's `properties` already contains exactly those named fields (resolved from the +configured JSON pointers) plus the internal `_event_id` / `_observed_at` keys. + +## Rendering rules (frontend) + +- The detail panel (`openEventPopup`, `static/app.js`) renders **only** the fields named in + `layer.propertyFields`, in declared order, reading values from `feature.properties`. +- A value matching `^https?://` renders as a safe external link (`rel="noopener noreferrer"`); + all other values render as escaped text. No domain-specific field handling (no magnitude, + depth, or PAGER chip logic). +- Title comes from `eventFeatureLabel` (operator `style.labelField`, else first non-internal + property, else `_event_id` / stream name). +- Fallback: when a layer carries no `propertyFields` manifest, render all non-internal + (`_`-prefixed excluded) properties so older payloads still display. + +## Out of scope + +- Per-field display labels and format hints (`PropertyFormat`) — deferred; operators set a + friendly `name` today. Revisit if a field needs a label distinct from its key. +- Dead `pager-chip` CSS in `static/style.css` is now unused; left in place (harmless) for a + separate cleanup pass. + From f5e6761ddf67d5fc7dc752f90728870b9b7b4fa2 Mon Sep 17 00:00:00 2001 From: Ryan Morton Date: Sun, 21 Jun 2026 21:58:45 -0600 Subject: [PATCH 5/6] test(federation): negative-path regressions for tri-thrust hardening Cover the shipped guards that lacked tests (live-DB integration, serial): TT-A02 subscribe rejects a non-active peer (400) TT-A04 binding create + delete require peers:manage (403, fail-closed) TT-A07 subscribe defers first poll when no Active binding exists TT-C01 oversized stream-event payloads -> InsertOutcome::Rejected, not stored TT-C09 oversized webhook data rejected at envelope validation 6/6 green against Postgres. Closes the test gap on the security-critical federation fixes before PR. Co-Authored-By: Claude Opus 4.8 (1M context) --- ...ration_hardening_regression_integration.rs | 392 ++++++++++++++++++ 1 file changed, 392 insertions(+) create mode 100644 tests/federation_hardening_regression_integration.rs diff --git a/tests/federation_hardening_regression_integration.rs b/tests/federation_hardening_regression_integration.rs new file mode 100644 index 0000000..9a47e03 --- /dev/null +++ b/tests/federation_hardening_regression_integration.rs @@ -0,0 +1,392 @@ +//! Negative-path regression tests for the tri-thrust federation-hardening fixes +//! (branch `feat/tri-thrust-federation-hardening`). Each test pins a guard that +//! a prior session shipped but did not cover: +//! +//! TT-A02 subscribe rejects a non-active peer +//! TT-A04 binding mutations require `peers:manage` (create + delete) +//! TT-A07 subscribe defers the first poll when no Active binding exists +//! TT-C01 oversized stream-event payloads are Rejected, not stored +//! TT-C09 oversized webhook `data` is rejected by envelope validation +//! +//! Run (serial, ignored, live DB): +//! DATABASE_URL=postgres://ione:ione@localhost:5433/ione \ +//! cargo test --test federation_hardening_regression_integration -- --ignored --test-threads=1 + +use std::net::SocketAddr; + +use chrono::Utc; +use hmac::{Hmac, Mac}; +use reqwest::StatusCode; +use serde_json::{json, Value}; +use sha2::Sha256; +use sqlx::{postgres::PgPoolOptions, PgPool}; +use tokio::net::TcpListener; +use uuid::Uuid; + +use ione::repos::{InsertOutcome, StreamEventRepo}; + +const DEFAULT_DATABASE_URL: &str = "postgres://ione:ione@localhost:5433/ione"; +const TEST_KEY: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; + +async fn spawn_app() -> (String, PgPool) { + std::env::set_var("IONE_AUTH_MODE", "local"); + std::env::set_var("IONE_TOKEN_KEY", TEST_KEY); + std::env::set_var("IONE_WEBHOOK_SECRET_KEY", TEST_KEY); + + let db_url = std::env::var("DATABASE_URL").unwrap_or_else(|_| DEFAULT_DATABASE_URL.to_owned()); + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(&db_url) + .await + .expect("failed to connect to Postgres"); + + sqlx::migrate!("./migrations") + .run(&pool) + .await + .expect("migration failed"); + + sqlx::query( + "TRUNCATE webhook_events_seen, workspace_peer_bindings, audit_events, approvals, artifacts, + trust_issuers, peers, routing_decisions, survivors, signals, + stream_events, streams, connectors, + memberships, roles, messages, conversations, + workspaces, users, organizations + RESTART IDENTITY CASCADE", + ) + .execute(&pool) + .await + .expect("truncate failed"); + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr: SocketAddr = listener.local_addr().expect("local addr"); + let app = ione::app(pool.clone()).await; + tokio::spawn(async move { + axum::serve(listener, app).await.expect("server error"); + }); + (format!("http://{}", addr), pool) +} + +// ─── seed helpers ───────────────────────────────────────────────────────────── + +async fn default_org_id(pool: &PgPool) -> Uuid { + sqlx::query_scalar("SELECT id FROM organizations ORDER BY created_at LIMIT 1") + .fetch_one(pool) + .await + .expect("default org not found") +} + +async fn ops_workspace_id(pool: &PgPool) -> Uuid { + sqlx::query_scalar("SELECT id FROM workspaces WHERE name = 'Operations' LIMIT 1") + .fetch_one(pool) + .await + .expect("Operations workspace not found") +} + +/// Set the default user's bootstrap `member` role permissions in place. +async fn set_member_permissions(pool: &PgPool, workspace_id: Uuid, perms: Value) { + sqlx::query("UPDATE roles SET permissions = $2 WHERE workspace_id = $1 AND name = 'member'") + .bind(workspace_id) + .bind(perms) + .execute(pool) + .await + .expect("set member permissions"); +} + +async fn insert_peer(pool: &PgPool, org_id: Uuid, status: &str) -> Uuid { + let issuer_id: Uuid = sqlx::query_scalar( + "INSERT INTO trust_issuers (org_id, issuer_url, audience, jwks_uri, claim_mapping) + VALUES ($1, $2, 'aud', 'secret:test', '{}'::jsonb) + RETURNING id", + ) + .bind(org_id) + .bind(format!("https://issuer-{status}.test")) + .fetch_one(pool) + .await + .expect("issuer"); + sqlx::query_scalar( + "INSERT INTO peers (org_id, name, mcp_url, issuer_id, sharing_policy, status) + VALUES ($1, $2, $3, $4, '{}'::jsonb, $5::peer_status) + RETURNING id", + ) + .bind(org_id) + .bind(format!("peer-{status}")) + .bind(format!("https://peer-{status}.test/mcp")) + .bind(issuer_id) + .bind(status) + .fetch_one(pool) + .await + .expect("peer") +} + +async fn insert_active_binding( + pool: &PgPool, + org_id: Uuid, + workspace_id: Uuid, + peer_id: Uuid, +) -> Uuid { + sqlx::query_scalar( + "INSERT INTO workspace_peer_bindings (org_id, workspace_id, peer_id, foreign_tenant_id, status) + VALUES ($1, $2, $3, 't-acme', 'active'::binding_status) + RETURNING id", + ) + .bind(org_id) + .bind(workspace_id) + .bind(peer_id) + .fetch_one(pool) + .await + .expect("binding") +} + +/// Seed a connector + stream under `workspace_id`, returning the stream id. +async fn insert_stream(pool: &PgPool, workspace_id: Uuid) -> Uuid { + let connector_id: Uuid = sqlx::query_scalar( + "INSERT INTO connectors (workspace_id, kind, name, config) + VALUES ($1, 'mcp'::connector_kind, 'test-connector', '{}'::jsonb) + RETURNING id", + ) + .bind(workspace_id) + .fetch_one(pool) + .await + .expect("connector"); + sqlx::query_scalar( + "INSERT INTO streams (connector_id, name) VALUES ($1, 'test-stream') RETURNING id", + ) + .bind(connector_id) + .fetch_one(pool) + .await + .expect("stream") +} + +async fn provision(base: &str, peer_id: Uuid) -> String { + let body: Value = reqwest::Client::new() + .post(format!("{base}/api/v1/peers/{peer_id}/webhook/provision")) + .send() + .await + .expect("provision request") + .json() + .await + .expect("provision json"); + body["signingSecret"] + .as_str() + .expect("signingSecret") + .to_string() +} + +fn signed_headers(secret: &str, body: &str, ts: i64) -> String { + type HmacSha256 = Hmac; + let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).expect("hmac key"); + mac.update(ts.to_string().as_bytes()); + mac.update(b"."); + mac.update(body.as_bytes()); + let digest = hex::encode(mac.finalize().into_bytes()); + format!("t={ts},v1={digest}") +} + +fn envelope(peer_id: Uuid, event_id: &str, data: Value) -> Value { + json!({ + "id": event_id, + "type": "alert.created", + "occurred_at": Utc::now(), + "peer_id": peer_id, + "foreign_tenant_id": "t-acme", + "severity": "routine", + "data": data, + "approval_required": false + }) +} + +// ─── TT-A04: binding mutations require peers:manage ─────────────────────────── + +/// POST a binding without `peers:manage` is forbidden; granting it clears the gate. +#[tokio::test] +#[ignore] +async fn create_binding_403_without_peers_manage() { + let (base, pool) = spawn_app().await; + let ws = ops_workspace_id(&pool).await; + let org_id = default_org_id(&pool).await; + let peer_id = insert_peer(&pool, org_id, "active").await; + let url = format!("{base}/api/v1/workspaces/{ws}/bindings"); + let req = json!({ "peerId": peer_id, "foreignTenantId": "t-acme", "scope": {} }); + + set_member_permissions(&pool, ws, json!(["peers:read"])).await; + let resp = reqwest::Client::new() + .post(&url) + .json(&req) + .send() + .await + .expect("create binding"); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + + // With the grant, the request passes the RBAC gate (no longer 403). + set_member_permissions(&pool, ws, json!(["admin"])).await; + let resp = reqwest::Client::new() + .post(&url) + .json(&req) + .send() + .await + .expect("create binding (granted)"); + assert_ne!(resp.status(), StatusCode::FORBIDDEN); +} + +/// DELETE on an existing binding without `peers:manage` is forbidden. +#[tokio::test] +#[ignore] +async fn delete_binding_403_without_peers_manage() { + let (base, pool) = spawn_app().await; + let ws = ops_workspace_id(&pool).await; + let org_id = default_org_id(&pool).await; + let peer_id = insert_peer(&pool, org_id, "active").await; + let binding_id = insert_active_binding(&pool, org_id, ws, peer_id).await; + set_member_permissions(&pool, ws, json!(["peers:read"])).await; + + let resp = reqwest::Client::new() + .delete(format!( + "{base}/api/v1/workspaces/{ws}/bindings/{binding_id}" + )) + .send() + .await + .expect("delete binding"); + assert_eq!(resp.status(), StatusCode::FORBIDDEN); + + // The binding is untouched by the rejected delete. + let still_there: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM workspace_peer_bindings WHERE id = $1") + .bind(binding_id) + .fetch_one(&pool) + .await + .expect("count"); + assert_eq!(still_there, 1); +} + +// ─── TT-A02 / TT-A07: subscribe peer-status guard + deferred first poll ─────── + +/// Subscribing a non-active peer is a 400; the message names the status. +#[tokio::test] +#[ignore] +async fn subscribe_rejects_non_active_peer() { + let (base, pool) = spawn_app().await; + let ws = ops_workspace_id(&pool).await; + let org_id = default_org_id(&pool).await; + let peer_id = insert_peer(&pool, org_id, "paused").await; + set_member_permissions(&pool, ws, json!(["admin"])).await; + + let resp = reqwest::Client::new() + .post(format!( + "{base}/api/v1/workspaces/{ws}/peers/{peer_id}/subscribe" + )) + .send() + .await + .expect("subscribe"); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + let body = resp.text().await.expect("body"); + assert!( + body.contains("not active"), + "expected a not-active rejection, got: {body}" + ); +} + +/// Subscribing an active peer with no Active binding succeeds but defers the +/// first poll (`firstPollDeferred: true`) rather than polling unscoped. +#[tokio::test] +#[ignore] +async fn subscribe_active_peer_defers_first_poll() { + let (base, pool) = spawn_app().await; + let ws = ops_workspace_id(&pool).await; + let org_id = default_org_id(&pool).await; + let peer_id = insert_peer(&pool, org_id, "active").await; + set_member_permissions(&pool, ws, json!(["admin"])).await; + + let resp = reqwest::Client::new() + .post(format!( + "{base}/api/v1/workspaces/{ws}/peers/{peer_id}/subscribe" + )) + .send() + .await + .expect("subscribe"); + assert_eq!(resp.status(), StatusCode::OK); + let body: Value = resp.json().await.expect("json"); + assert_eq!( + body["firstPollDeferred"], + json!(true), + "no Active binding exists, so the first poll must be deferred: {body}" + ); +} + +// ─── TT-C01: oversized stream-event payloads are Rejected ───────────────────── + +#[tokio::test] +#[ignore] +async fn oversized_stream_event_is_rejected() { + let (_base, pool) = spawn_app().await; + let ws = ops_workspace_id(&pool).await; + let stream_id = insert_stream(&pool, ws).await; + let repo = StreamEventRepo::new(pool.clone()); + + // > 65_536 serialized bytes: rejected, not stored. + let big = json!({ "blob": "x".repeat(70_000) }); + let outcome = repo + .insert_event(stream_id, big, Utc::now(), None) + .await + .expect("insert big"); + assert_eq!(outcome, InsertOutcome::Rejected); + + // A small payload still inserts. + let small = json!({ "blob": "hi" }); + let outcome = repo + .insert_event(stream_id, small, Utc::now(), None) + .await + .expect("insert small"); + assert_eq!(outcome, InsertOutcome::Inserted); + + // Exactly one row landed (the rejected one never hit the table). + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stream_events WHERE stream_id = $1") + .bind(stream_id) + .fetch_one(&pool) + .await + .expect("count"); + assert_eq!(count, 1); +} + +// ─── TT-C09: oversized webhook `data` is rejected by envelope validation ────── + +#[tokio::test] +#[ignore] +async fn oversized_webhook_data_is_rejected() { + let (base, pool) = spawn_app().await; + let ws = ops_workspace_id(&pool).await; + let org_id = default_org_id(&pool).await; + let peer_id = insert_peer(&pool, org_id, "active").await; + let secret = provision(&base, peer_id).await; + insert_active_binding(&pool, org_id, ws, peer_id).await; + let client = reqwest::Client::new(); + + // A normal-sized event with an Active binding is accepted. + let ok_body = serde_json::to_string(&envelope(peer_id, "evt-ok", json!({ "message": "hi" }))) + .expect("body"); + let resp = client + .post(format!("{base}/webhooks/peer/{peer_id}")) + .header( + "X-IONe-Signature", + signed_headers(&secret, &ok_body, Utc::now().timestamp()), + ) + .body(ok_body) + .send() + .await + .expect("post ok"); + assert_eq!(resp.status(), StatusCode::OK); + + // Same setup, but `data` exceeds the 100 KiB cap: rejected at validation. + let big_data = json!({ "blob": "x".repeat(110_000) }); + let big_body = serde_json::to_string(&envelope(peer_id, "evt-big", big_data)).expect("body"); + let resp = client + .post(format!("{base}/webhooks/peer/{peer_id}")) + .header( + "X-IONe-Signature", + signed_headers(&secret, &big_body, Utc::now().timestamp()), + ) + .body(big_body) + .send() + .await + .expect("post big"); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} From 903677b65009ea936c65c8a7f4a781427c48aed1 Mon Sep 17 00:00:00 2001 From: Ryan Morton Date: Tue, 7 Jul 2026 15:56:06 -0600 Subject: [PATCH 6/6] docs: add minimal CLAUDE.md (build/test anchor); fix stream-P rule paths + stale P1-P3 status 2026-07-07 asset audit. Co-Authored-By: Claude Fable 5 --- .claude/rules/path-2-stream-p.md | 6 ++--- CLAUDE.md | 42 ++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 3 deletions(-) create mode 100644 CLAUDE.md diff --git a/.claude/rules/path-2-stream-p.md b/.claude/rules/path-2-stream-p.md index 19e6887..71dea02 100644 --- a/.claude/rules/path-2-stream-p.md +++ b/.claude/rules/path-2-stream-p.md @@ -11,7 +11,7 @@ paths: # Path 2 — Stream P Supporting (IONe) -Work in this repo supports **Path 2 — Modular Infrastructure Risk Intelligence**, the selected growth path for Morton Analytics LLC (decided 2026-05-12). IONe's role: **integration fabric for Morton Analytics' polyglot client-app portfolio.** GroundPulse is the first instance; TerraYield (`../eo_ag/`) and bearingLineDash (`../bearingLineDash/`) are the next reference apps the substrate must accommodate. Full plan: `../morton-analytics-web/md/strategy/path-2-90day-plan.md`. Substrate spec: `md/design/ione-substrate.md`. +Work in this repo supports **Path 2 — Modular Infrastructure Risk Intelligence**, the selected growth path for Morton Analytics LLC (decided 2026-05-12). IONe's role: **integration fabric for Morton Analytics' polyglot client-app portfolio.** GroundPulse is the first instance; TerraYield (`../eo_ag/`) and bearingLineDash (`../bearingLineDash/`) are the next reference apps the substrate must accommodate. Full plan: `../morton-command-center/strategy/path-2-90day-plan.md`. Substrate spec: `md/design/ione-substrate.md`. ## Positioning inside Path 2 @@ -40,7 +40,7 @@ P7 is Tier 3 — protected against P2 (pipelines module) but cuttable if Stream ## Working in this repo - All Stream P commits should reference the outcome ID (currently P7) and the GroundPulse-substrate framing. -- Major architectural decisions go in `md/design/`. Cross-reference the Path 2 tracker (`../morton-analytics-web/md/strategy/path-2-tracker.md`). +- Major architectural decisions go in `md/design/`. Cross-reference the Path 2 tracker (`../morton-command-center/strategy/path-2-tracker.md`). - IONe pricing strategy at `md/strategy/market/ione-pricing.md` is canonical for any pricing discussion — do not re-derive. ## What this rule prohibits @@ -58,4 +58,4 @@ P7 is Tier 3 — protected against P2 (pipelines module) but cuttable if Stream ## When in doubt -Read the 90-day plan at `../morton-analytics-web/md/strategy/path-2-90day-plan.md`. The current week's Tier 1 Stream P outcomes (P1–P3) are the priority. IONe (P7) is supporting. +Read the 90-day plan at `../morton-command-center/strategy/path-2-90day-plan.md`. Stream P Tier 1 outcomes (P1–P3) shipped; per the replan, IONe (P7) is deferred Tier 3 (target Aug 2026) behind GTM and engine reliability. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..08ac437 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,42 @@ +# CLAUDE.md — IONe + +## What this is +IONe is a **pre-alpha** single Rust binary (axum + tokio + sqlx/Postgres 16 + +pgvector) with a static UI (`static/`), serving as the MCP-based federation +substrate underneath GroundPulse and TerraYield: brokered identity (OIDC/SAML), +signed webhook ingress, generator↔adversarial LLM loop with human approval + +audit, and a ref-rendering UX shell. It is **OSS engineering infrastructure, +not a standalone product** — no standalone GTM. + +Canonical design: `md/design/ione-substrate.md` (integration-fabric framing). +Strategy context: `.claude/rules/path-2-stream-p.md`. +App-integration contract: `md/design/app-integration-playbook.md`. + +## Quickstart +```bash +cp .env.example .env # sets IONE_SEED_DEMO=1 (demo workspace) +docker compose up -d postgres minio +cargo sqlx database create +cargo sqlx migrate run +cargo run --release # http://localhost:3000 +``` +Two-node federation demo: `./scripts/demo.sh` (nodes on :3000 and :3001). + +## Tests +```bash +# Cheap unit path: +cargo test --test phase01_chat + +# Integration: live Postgres, serial, #[ignore]-gated: +DATABASE_URL=postgres://ione:ione@localhost:5433/ione \ + IONE_SKIP_LIVE=1 \ + cargo test -- --ignored --test-threads=1 + +# Playwright e2e: +npm run test:e2e +``` +Unset `IONE_SKIP_LIVE` to exercise live Ollama generator/critic/router paths. + +## Secrets +`IONE_TOKEN_KEY` / `IONE_WEBHOOK_SECRET_KEY` live in `.env` (gitignored). +Never commit `.env` or embed key values in settings/permissions entries.