diff --git a/Cargo.lock b/Cargo.lock index 39c62fa..add01ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -985,11 +985,15 @@ dependencies = [ "rucelium-store", "rucelium-transport", "rucelium-worldgraph", + "rufield-core", + "rufield-fusion", + "rufield-provenance", "rustls", "serde", "serde_json", "tokio", "tower", + "tower-http", ] [[package]] diff --git a/crates/rucelium-gateway/Cargo.toml b/crates/rucelium-gateway/Cargo.toml index ad792ad..eb44a7c 100644 --- a/crates/rucelium-gateway/Cargo.toml +++ b/crates/rucelium-gateway/Cargo.toml @@ -30,10 +30,19 @@ rucelium-worldgraph = { workspace = true } rucelium-federation = { workspace = true } rucelium-policy = { workspace = true } rucelium-store = { workspace = true } +rufield-core = { workspace = true } +rufield-fusion = { workspace = true } +rufield-provenance = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } axum = "0.7" +# CORS (dev-only, v0.1 posture): lets a same-machine dashboard served from a +# different origin/port (e.g. RuView's ui/) read this gateway's JSON. See +# `api::router`'s doc comment for scope/limits — matches this crate's +# existing "admin endpoints unauthenticated, bind to localhost" posture +# rather than adding a new one. +tower-http = { version = "0.6", features = ["cors"] } tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "time", "signal", "sync"] } reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] } diff --git a/crates/rucelium-gateway/src/api.rs b/crates/rucelium-gateway/src/api.rs index ba2b3d7..5c6923a 100644 --- a/crates/rucelium-gateway/src/api.rs +++ b/crates/rucelium-gateway/src/api.rs @@ -21,9 +21,10 @@ use crate::federation::{accept_artifact, ArtifactEffect, ArtifactRejection}; use crate::state::{now_ns, GatewayState}; use crate::transport::FederationArtifact; use axum::extract::{Path, Query, State}; -use axum::http::StatusCode; +use axum::http::{HeaderValue, Method, StatusCode}; use axum::routing::{get, post}; use axum::{Json, Router}; +use tower_http::cors::CorsLayer; use rucelium_core::EventKind; use rucelium_federation::{project_sample, SensorThingsBundle}; use rucelium_policy::ControlError; @@ -60,12 +61,50 @@ struct WindowParam { window_s: Option, } +/// CORS for the two dashboard read routes ONLY (`/api/rf-context`, +/// `/api/stats`) — never applied to the rest of the router. Read-only +/// (`GET`), no credentials, and origin-gated: `origins` is the +/// `--dashboard-origin` allowlist (empty by default), so no browser origin +/// gets a cross-origin read until an operator explicitly configures one +/// (e.g. `http://127.0.0.1:8090`, RuView's own `ui/` origin). An empty +/// allowlist means [`CorsLayer`] emits no `Access-Control-Allow-Origin` +/// header at all, so an unconfigured deployment behaves exactly as it did +/// before this existed: no cross-origin browser reads of anything. +/// +/// This is deliberately scoped to two routes, not the whole router: the +/// admin, federation, observation, and event surfaces stay same-origin-only +/// in the browser regardless of this setting. Every route is still +/// unauthenticated at the network layer in v0.1 (module docs) — this +/// setting only ever controls what an arbitrary *page* can read from inside +/// an arbitrary *browser tab*, not what curl/any non-browser client can +/// already read. +fn dashboard_cors_layer(origins: &[String]) -> Result { + let layer = CorsLayer::new().allow_methods([Method::GET]); + if origins.is_empty() { + return Ok(layer); + } + let mut allowed = Vec::with_capacity(origins.len()); + for origin in origins { + let value = HeaderValue::from_str(origin) + .map_err(|e| format!("invalid --dashboard-origin {origin:?}: {e}"))?; + allowed.push(value); + } + Ok(layer.allow_origin(allowed)) +} + /// Build the gateway's axum router (see module docs for the endpoint list -/// and the v0.1 admin-endpoint security posture). -pub fn router(state: GatewayState) -> Router { - Router::new() - .route("/health", get(health)) +/// and the v0.1 admin-endpoint security posture). `dashboard_origins` is the +/// `--dashboard-origin` CORS allowlist for `/api/rf-context` and +/// `/api/stats` only (see [`dashboard_cors_layer`]); pass `&[]` to disable +/// cross-origin browser reads entirely (the default). +pub fn router(state: GatewayState, dashboard_origins: &[String]) -> Result { + let dashboard = Router::new() + .route("/api/rf-context", get(rf_context)) .route("/api/stats", get(stats)) + .route_layer(dashboard_cors_layer(dashboard_origins)?); + + let router = Router::new() + .route("/health", get(health)) .route("/api/observations/recent", get(observations_recent)) .route("/api/events", get(events_recent)) .route("/api/sensorthings/Things", get(st_things)) @@ -78,7 +117,9 @@ pub fn router(state: GatewayState) -> Router { .route("/api/federation/announce", post(fed_announce)) .route("/api/admin/revoke/:node_id", post(admin_revoke)) .route("/api/admin/command", post(admin_command)) - .with_state(state) + .merge(dashboard) + .with_state(state); + Ok(router) } /// `GET /health` — liveness. @@ -124,6 +165,32 @@ async fn stats(State(state): State) -> Json { })) } +/// `GET /api/rf-context` — the RuView RF-context bridge's state (ADR-264 §8, +/// see `rf_bridge`): the configured upstream (if any), ingest counters, the +/// bounded recent ring, the current within-RF-domain fused inferences +/// (`rf_bridge::current_inferences` — RuField's own room-state rules over +/// this upstream's own verified events; needs no external pairing), and a +/// bounded history of past inference snapshots so a transient inference is +/// still visible after its own TTL expires (in-memory only — does not +/// survive a restart; see [`crate::rf_bridge::InferenceSnapshot`] docs for +/// why not durable yet). Each ring entry is honestly tagged `verified` — an unverified event +/// is visible here but excluded from fusion, never silently dropped or +/// upgraded. `fused_into_worldgraph` is always `false`: this bridge does not +/// (yet) correlate RF context against any biome environmental sample — see +/// `rf_bridge`'s module docs for why that's a separate, still-unbuilt +/// decision. Empty and idle when `--rf-upstream` was not set. +async fn rf_context(State(state): State) -> Json { + let inner = state.inner.lock().await; + Json(json!({ + "upstream": inner.rf_upstream, + "fused_into_worldgraph": false, + "stats": inner.rf_stats, + "recent": inner.rf_recent.iter().collect::>(), + "room_state_inferences": crate::rf_bridge::current_inferences(&inner.rf_fusion), + "inference_history": inner.rf_inference_history.iter().collect::>(), + })) +} + /// `GET /api/observations/recent?limit=50` — most recent stored samples in /// append order. async fn observations_recent( diff --git a/crates/rucelium-gateway/src/config.rs b/crates/rucelium-gateway/src/config.rs index faf2b44..5b93a9e 100644 --- a/crates/rucelium-gateway/src/config.rs +++ b/crates/rucelium-gateway/src/config.rs @@ -24,6 +24,9 @@ pub const DEFAULT_ACTUATOR_ID: &str = "sluice-gate-1"; /// Default durability mode for the durable stores: the daemon fsyncs every /// accepted append, so an accepted record survives power loss. pub const DEFAULT_FSYNC: bool = true; +/// Default poll interval for the RuView RF-context bridge (ADR-264 §8), +/// matching `rufield-viewer`'s own default `/api/field` poll cadence. +pub const DEFAULT_RF_POLL_MS: u64 = 500; /// Runtime configuration of one gateway daemon instance. #[derive(Debug, Clone, PartialEq, Eq)] @@ -61,6 +64,22 @@ pub struct GatewayConfig { /// Durability mode for `ObservationStore` / `EventStore`: `true` fsyncs /// every accepted append before it is acknowledged. pub fsync: bool, + /// Base URL of a RuView `wifi-densepose-sensing-server` to poll for RF + /// context (ADR-264 §8). `None` (default) disables the bridge entirely — + /// this is opt-in, not a dependency the gateway assumes exists. + pub rf_upstream: Option, + /// Poll interval for the RF-context bridge, in milliseconds. Unused when + /// `rf_upstream` is `None`. + pub rf_poll_ms: u64, + /// Browser origins allowed to read the two dashboard endpoints + /// (`/api/rf-context`, `/api/stats`) cross-origin (repeatable + /// `--dashboard-origin`, e.g. `http://127.0.0.1:8090`). Empty (default) + /// means no CORS headers are emitted at all — a browser page on any + /// other origin cannot read them, same as every other route on this + /// gateway. This is a browser-confidentiality control, not an auth + /// control: every route stays unauthenticated at the network layer + /// regardless of this setting (module docs). + pub dashboard_origins: Vec, } impl Default for GatewayConfig { @@ -79,6 +98,9 @@ impl Default for GatewayConfig { federation_backfill_ms: None, actuator_id: DEFAULT_ACTUATOR_ID.to_string(), fsync: DEFAULT_FSYNC, + rf_upstream: None, + rf_poll_ms: DEFAULT_RF_POLL_MS, + dashboard_origins: Vec::new(), } } } @@ -129,6 +151,13 @@ impl GatewayConfig { } "--actuator" => config.actuator_id = value("--actuator")?, "--fsync" => config.fsync = parse_num(&value("--fsync")?, "--fsync")?, + "--rf-upstream" => config.rf_upstream = Some(value("--rf-upstream")?), + "--rf-poll-ms" => { + config.rf_poll_ms = parse_num(&value("--rf-poll-ms")?, "--rf-poll-ms")?; + } + "--dashboard-origin" => config + .dashboard_origins + .push(value("--dashboard-origin")?), unknown => return Err(format!("unknown flag {unknown}")), } } @@ -169,6 +198,12 @@ mod tests { assert_eq!(c.federation_backfill_ms(), 30_000); assert_eq!(c.actuator_id, "sluice-gate-1"); assert!(c.fsync, "the daemon fsyncs accepted appends by default"); + assert_eq!(c.rf_upstream, None, "the RF bridge is opt-in, off by default"); + assert_eq!(c.rf_poll_ms, 500); + assert!( + c.dashboard_origins.is_empty(), + "no browser origin is CORS-allowed by default" + ); } #[test] @@ -210,6 +245,12 @@ mod tests { "weir-3", "--fsync", "false", + "--rf-upstream", + "http://127.0.0.1:8080", + "--rf-poll-ms", + "250", + "--dashboard-origin", + "http://127.0.0.1:8090", ])) .unwrap(); assert_eq!(c.biome_id, "biome/x"); @@ -223,6 +264,9 @@ mod tests { assert_eq!(c.federation_poll_ms, 200); assert_eq!(c.actuator_id, "weir-3"); assert!(!c.fsync); + assert_eq!(c.rf_upstream, Some("http://127.0.0.1:8080".to_string())); + assert_eq!(c.rf_poll_ms, 250); + assert_eq!(c.dashboard_origins, vec!["http://127.0.0.1:8090"]); } #[test] @@ -237,6 +281,21 @@ mod tests { assert_eq!(c.peers, vec!["http://a:7465", "http://b:7465"]); } + #[test] + fn dashboard_origin_is_repeatable_in_order() { + let c = GatewayConfig::from_args(args(&[ + "--dashboard-origin", + "http://127.0.0.1:8090", + "--dashboard-origin", + "http://localhost:8090", + ])) + .unwrap(); + assert_eq!( + c.dashboard_origins, + vec!["http://127.0.0.1:8090", "http://localhost:8090"] + ); + } + #[test] fn unknown_flag_is_an_error() { let err = GatewayConfig::from_args(args(&["--nope"])).unwrap_err(); diff --git a/crates/rucelium-gateway/src/lib.rs b/crates/rucelium-gateway/src/lib.rs index 04d4285..d6d9957 100644 --- a/crates/rucelium-gateway/src/lib.rs +++ b/crates/rucelium-gateway/src/lib.rs @@ -14,8 +14,19 @@ //! ──► /api/federation/{pubkey,summary,revocations,peers} //! ──► /api/federation/announce (push inbox, ADR-269 §3) //! ──► /api/admin/{revoke/:node_id,command} +//! ──► /api/rf-context (ADR-264 §8, see `rf_bridge`) //! ``` //! +//! ## RuView RF context (ADR-264 §8) +//! +//! `--rf-upstream ` starts [`rf_bridge::run_rf_ingest`]: it polls a +//! RuView `wifi-densepose-sensing-server`'s `/api/field` (ADR-262 P3), +//! verifies each event's provenance receipt, and keeps a bounded, honestly +//! verified/unverified-tagged ring at `GET /api/rf-context`. This is storage +//! only — it does **not** fuse into the WorldGraph yet; see [`rf_bridge`]'s +//! module docs for why that pairing decision is deliberately not guessed at +//! here. +//! //! ## Push federation (ADR-269 §3) //! //! Federation is push-first over a swappable [`transport::FederationTransport`]: @@ -61,6 +72,7 @@ pub mod federation; pub mod journal; pub mod net; pub mod pipeline; +pub mod rf_bridge; pub mod simulate; pub mod state; pub mod transport; @@ -144,7 +156,7 @@ pub async fn spawn_gateway_with_transport( let mut tasks = Vec::new(); tasks.push(tokio::spawn(net::run_udp(udp, state.clone()))); - let router = api::router(state.clone()); + let router = api::router(state.clone(), &config.dashboard_origins)?; tasks.push(tokio::spawn(async move { if let Err(e) = axum::serve(listener, router).await { eprintln!("gateway: http server error: {e}"); @@ -185,6 +197,14 @@ pub async fn spawn_gateway_with_transport( ))); } + if let Some(upstream) = config.rf_upstream.clone() { + tasks.push(tokio::spawn(rf_bridge::run_rf_ingest( + state.clone(), + upstream, + config.rf_poll_ms, + ))); + } + Ok(GatewayHandle { state, udp_port, diff --git a/crates/rucelium-gateway/src/main.rs b/crates/rucelium-gateway/src/main.rs index 8e8370c..4a0185f 100644 --- a/crates/rucelium-gateway/src/main.rs +++ b/crates/rucelium-gateway/src/main.rs @@ -14,7 +14,8 @@ async fn main() { "usage: rucelium-gateway [--biome-id ] [--udp ] [--http ] \ [--data-dir ] [--peer ]... [--simulate ] [--seed ] \ [--sim-interval-ms ] [--retention-check-secs ] \ - [--federation-poll-ms ] [--actuator ] [--fsync ]" + [--federation-poll-ms ] [--actuator ] [--fsync ] \ + [--rf-upstream ] [--rf-poll-ms ] [--dashboard-origin ]..." ); std::process::exit(2); } @@ -35,6 +36,17 @@ async fn main() { println!(" peer: {peer}"); } } + match &config.rf_upstream { + Some(url) => println!(" rf-context: polling {url} every {}ms (ADR-264 §8)", config.rf_poll_ms), + None => println!(" rf-context: disabled (set --rf-upstream to enable)"), + } + if config.dashboard_origins.is_empty() { + println!(" dashboard: CORS disabled (no browser origin can read /api/rf-context or /api/stats cross-origin)"); + } else { + for origin in &config.dashboard_origins { + println!(" dashboard: CORS allowed for {origin} (GET /api/rf-context, /api/stats only)"); + } + } println!(" WARNING: admin endpoints are UNAUTHENTICATED in v0.1 — bind"); println!(" the http port to localhost or firewall it."); diff --git a/crates/rucelium-gateway/src/rf_bridge.rs b/crates/rucelium-gateway/src/rf_bridge.rs new file mode 100644 index 0000000..c6069df --- /dev/null +++ b/crates/rucelium-gateway/src/rf_bridge.rs @@ -0,0 +1,553 @@ +//! RuView RF-context ingest (ADR-264 §8) — two *different* kinds of "fusion", +//! only one of which this module does. +//! +//! Polls an external RuView `wifi-densepose-sensing-server`'s `GET +//! /api/field` (ADR-262 P3), verifies each event's provenance receipt, keeps +//! a bounded ring of the resulting [`RfContext`]s, and runs verified events +//! through [`rufield_fusion::RuFieldFusion`] to produce real room-state +//! inferences (`person_present`, `motion`, …) — all exposed at +//! `GET /api/rf-context`. +//! +//! ## Two fusions, not one +//! +//! * **`rufield_fusion::RuFieldFusion` (this module, real, wired in below)** +//! — a temporal-window fusion *within* the RF domain: it combines this +//! upstream's own `FieldEvent`s over a short window into inferences using +//! RuField's own rule set. It needs nothing external — no environmental +//! sample, no biome, no zone mapping — so there is nothing to fabricate. +//! This is the same engine `rufield-viewer` already runs on the exact same +//! event stream; wiring it here just makes the gateway able to answer +//! "what does this RF upstream currently think is happening" on its own. +//! * **`rucelium_worldgraph::fuse_rf_context` (still NOT called here)** — a +//! *cross-domain* fusion: pairing an `RfContext` against a *specific* +//! existing environmental sample's WorldGraph node to compute a +//! [`rucelium_worldgraph::Plausibility`] verdict (does the RF context +//! support or contradict *that* biome sensor's reading). Nothing in this +//! codebase yet defines which biome zone/sensor a given RuView device is +//! supposed to corroborate, and inventing that pairing would be exactly +//! the fabricated correlation ADR-264 §8 exists to prevent. That +//! zone-pairing design is still deliberate follow-up work, not guessed at +//! — this module does not touch the WorldGraph. +//! +//! ## Honesty +//! +//! Every event is receipt-checked via [`rufield_provenance::is_fusable`] and +//! tagged `verified: bool` in the stored ring — an unverified event is kept +//! visible (like a forged-data ✗ badge) but is excluded from the fusion +//! engine entirely (mirrors `rufield-viewer`'s own rule: forged data is never +//! rendered as trusted). `/api/rf-context` reports +//! `"fused_into_worldgraph": false` explicitly, so a consumer never has to +//! infer the WorldGraph boundary from absence. + +use rucelium_worldgraph::RfContext; +use rufield_core::{FieldEvent, FusionEngine, InferenceQuery}; +use rufield_fusion::RuFieldFusion; +use serde::{Deserialize, Serialize}; +use std::collections::VecDeque; +use std::time::Duration; + +use crate::state::{now_ns, GatewayState}; + +/// Bounded ring size for recently ingested RF context (mirrors +/// `wifi-densepose-sensing-server`'s own `FIELD_RING_CAPACITY` convention — +/// a live tap, not a store). +pub const RF_RECENT_CAPACITY: usize = 64; + +/// Bounded ring size for retained inference snapshots. Smaller than +/// [`RF_RECENT_CAPACITY`] because each snapshot carries every currently-live +/// inference (typically a handful), not one event. +pub const INFERENCE_HISTORY_CAPACITY: usize = 32; + +/// Default poll interval for `GET /api/field`, matching +/// `rufield-viewer`'s own default. +pub const DEFAULT_RF_POLL_MS: u64 = 500; + +/// One ingested RF context plus its verification outcome and local receipt +/// time. +#[derive(Debug, Clone, Serialize)] +pub struct RfEntry { + /// The distilled RF context (never ground truth — ADR-264 §8). + pub context: RfContext, + /// Whether the source `FieldEvent`'s provenance receipt verified + /// (`rufield_provenance::is_fusable`). `false` means the bytes were + /// reached but the signature/hash did not check out — stored so a forged + /// or corrupted upstream is *visible*, never silently dropped. + pub verified: bool, + /// When this gateway received the event, ns since Unix epoch. + pub received_ns: u64, +} + +/// Ingest counters, exposed at `GET /api/rf-context`. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)] +pub struct RfIngestStats { + /// Successful polls of the upstream `/api/field` ring. + pub polls: u64, + /// Polls that failed (unreachable, non-2xx, or a malformed payload). + pub poll_failures: u64, + /// Total `FieldEvent`s received across all polls. + pub events_received: u64, + /// Received events whose provenance receipt verified. + pub events_verified: u64, + /// Received events whose provenance receipt did NOT verify (stored, + /// flagged, never trusted). + pub events_unverified: u64, + /// Received events that were not `Modality::WifiCsi` and so are not RF + /// context this bridge understands (`RfContext::from_field_event` + /// returns `None`). Not an error — RuView's ring may carry other + /// modalities over time. + pub events_other_modality: u64, +} + +/// One captured moment of the fusion engine's current inferences, retained +/// so a `person_present` (or any other) inference that fired between two +/// polls is not simply lost the moment its 2-second TTL expires. This is the +/// gap between how RF inferences behaved before this struct existed (only +/// ever computed live, on request, from whatever is in the engine's window +/// *right now*) and how every other durable thing in this gateway behaves +/// (`ObservationStore`, `EventStore`) — RF inferences are still NOT written +/// to either of those (they are not `EnvSample`/`EnvironmentalEvent`s and +/// have no biome `node_id`/`sequence` to cite — the same structural gap +/// `fuse_rf_context` ran into), so this is in-memory history, not disk +/// durability. It survives across polls; it does not survive a restart. +#[derive(Debug, Clone, Serialize)] +pub struct InferenceSnapshot { + /// When this snapshot was captured, ns since Unix epoch (the poll's + /// `received_ns`, not any individual inference's own `produced_ns`). + pub captured_ns: u64, + /// Every inference the engine reported live at capture time. + pub inferences: Vec, +} + +/// Shape of RuView's `GET /api/field` response that this bridge actually +/// needs. Deliberately declares **only** `events`: the response also carries +/// `signer_pubkey_hex` and `dev_signing_key`, but their exact shape is +/// RuView's to evolve (see `ruvnet/RuCelium#1`, where a *different* consumer +/// of this same endpoint broke because it over-specified those two +/// informational fields). Serde drops unknown fields by default — no +/// `deny_unknown_fields` — so this struct is inert to that class of drift. +#[derive(Debug, Deserialize)] +struct ApiFieldPayload { + events: Vec, +} + +/// Pure ingest step (no I/O — unit-testable): fold one batch of freshly +/// polled `FieldEvent`s into the bounded ring, the counters, and the +/// within-RF-domain fusion engine (module docs: this is the fusion that +/// needs no external pairing, so it runs unconditionally on every verified +/// event) — and, only when this batch actually fed the engine something new, +/// capture a snapshot of its current inferences into `history` so a +/// transient inference outlives its own TTL instead of vanishing the moment +/// nobody happens to be polling `/api/rf-context`. +pub fn ingest_events( + recent: &mut VecDeque, + stats: &mut RfIngestStats, + fusion: &mut RuFieldFusion, + history: &mut VecDeque, + events: &[FieldEvent], + received_ns: u64, +) { + stats.polls += 1; + let mut fed_engine = false; + for ev in events { + stats.events_received += 1; + let verified = rufield_provenance::is_fusable(ev); + if verified { + stats.events_verified += 1; + // Mirrors rufield-viewer's own rule: only verified events reach + // the fusion engine. `ingest` itself re-checks §11 fusability + // (belt-and-suspenders), so forged data cannot reach it either way. + let _ = fusion.ingest(ev.clone()); + fed_engine = true; + } else { + stats.events_unverified += 1; + } + match RfContext::from_field_event(ev) { + Some(context) => { + if recent.len() == RF_RECENT_CAPACITY { + recent.pop_front(); + } + recent.push_back(RfEntry { + context, + verified, + received_ns, + }); + } + None => stats.events_other_modality += 1, + } + } + // Only capture a snapshot when this batch actually changed the engine's + // state — an empty or all-unverified poll would just re-record the same + // stale inferences (or an empty vec) under a new timestamp, which is + // noise, not history. + if fed_engine { + if history.len() == INFERENCE_HISTORY_CAPACITY { + history.pop_front(); + } + history.push_back(InferenceSnapshot { + captured_ns: received_ns, + inferences: current_inferences(fusion), + }); + } +} + +/// Record a failed poll attempt (unreachable upstream, non-2xx, or a +/// malformed payload). Never fatal — the next tick tries again. +pub fn record_poll_failure(stats: &mut RfIngestStats) { + stats.poll_failures += 1; +} + +/// Current room-state inferences from the within-RF-domain fusion engine +/// (module docs: the fusion that needs no external pairing). Empty on a +/// query error — `infer` errors only on the engine's own internal +/// consistency checks, not on caller input, so this stays a best-effort read +/// rather than a `Result` the HTTP handler has to plumb through. +#[must_use] +pub fn current_inferences(fusion: &RuFieldFusion) -> Vec { + fusion.infer(&InferenceQuery::all()).unwrap_or_default() +} + +/// Fetch and decode one `GET /api/field` response. The only I/O in +/// this module — everything else is pure and covered by unit tests. +async fn poll_once(client: &reqwest::Client, url: &str) -> Result, String> { + let resp = client + .get(url) + .send() + .await + .map_err(|e| format!("request failed: {e}"))?; + if !resp.status().is_success() { + return Err(format!("http {}", resp.status())); + } + let body = resp + .text() + .await + .map_err(|e| format!("read body: {e}"))?; + let payload: ApiFieldPayload = + serde_json::from_str(&body).map_err(|e| format!("decode /api/field: {e}"))?; + Ok(payload.events) +} + +/// Run the RF ingest task forever: poll `/api/field` every +/// `poll_ms`, verify each event, and fold accepted RF context into +/// [`crate::state::Inner::rf_recent`]. Never fatal — an unreachable upstream +/// or a malformed payload just counts a failure and retries on the next +/// tick. +pub async fn run_rf_ingest(state: GatewayState, upstream: String, poll_ms: u64) { + let client = match reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + { + Ok(c) => c, + Err(_) => return, + }; + let url = format!("{}/api/field", upstream.trim_end_matches('/')); + let mut tick = tokio::time::interval(Duration::from_millis(poll_ms.max(50))); + loop { + tick.tick().await; + match poll_once(&client, &url).await { + Ok(events) => { + let received_ns = now_ns(); + let mut guard = state.inner.lock().await; + let inner = &mut *guard; + ingest_events( + &mut inner.rf_recent, + &mut inner.rf_stats, + &mut inner.rf_fusion, + &mut inner.rf_inference_history, + &events, + received_ns, + ); + } + Err(e) => { + eprintln!("gateway: rf-context poll of {url} failed: {e}"); + let mut inner = state.inner.lock().await; + record_poll_failure(&mut inner.rf_stats); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rufield_core::{ + FieldAxis, FieldTensor, Modality, Observation, PrivacyClass, ProvenanceRef, + SensorDescriptor, + }; + use rufield_provenance::Signer; + + fn wifi_csi_event(device_id: &str, signed: bool) -> FieldEvent { + let tensor = FieldTensor::new( + 1_000, + Modality::WifiCsi, + vec![FieldAxis::Frequency], + vec![2], + vec![0.1, 0.2], + 0.9, + 0.01, + Some("cal".into()), + PrivacyClass::P2, + ) + .unwrap(); + let mut observation = Observation::occupancy(0.8, PrivacyClass::P2); + observation + .features + .insert("motion_energy".to_string(), 0.7); + // "presence" is what `rufield-fusion`'s default `person_present` rule + // (rules/room_state.toml) actually keys on — without it the fusion + // regression tests below would pass trivially (no rule ever fires). + observation.features.insert("presence".to_string(), 0.8); + let mut ev = FieldEvent::new( + format!("ev-{device_id}"), + 1_000, + SensorDescriptor { + modality: "wifi_csi".into(), + vendor: "simulated".into(), + device_id: device_id.to_string(), + placement: "unknown".into(), + clock_domain: "local".into(), + }, + tensor, + observation, + ProvenanceRef { + raw_hash: "sha256:raw".into(), + firmware_hash: "sha256:fw".into(), + model_id: "m".into(), + calibration_id: "cal".into(), + // Deliberately non-synthetic: `is_fusable` short-circuits to + // `true` for synthetic events regardless of signature, which + // would make the "unsigned → unverified" test meaningless. + // Real verification only actually runs when this is `false`. + synthetic: false, + signature_hex: None, + signer_pubkey_hex: None, + }, + ); + if signed { + Signer::from_seed(b"rf-bridge-test-seed-32-bytes-ok!") + .sign_event(&mut ev) + .expect("test event signs"); + } + ev + } + + fn other_modality_event() -> FieldEvent { + let tensor = FieldTensor::new( + 1_000, + Modality::MmwaveRadar, + vec![FieldAxis::Frequency], + vec![1], + vec![0.5], + 0.5, + 0.01, + None, + PrivacyClass::P2, + ) + .unwrap(); + FieldEvent::new( + "ev-radar", + 1_000, + SensorDescriptor { + modality: "mmwave_radar".into(), + vendor: "sim".into(), + device_id: "radar-1".into(), + placement: "unknown".into(), + clock_domain: "local".into(), + }, + tensor, + Observation::occupancy(0.5, PrivacyClass::P2), + ProvenanceRef { + raw_hash: "sha256:raw".into(), + firmware_hash: "sha256:fw".into(), + model_id: "m".into(), + calibration_id: "cal".into(), + synthetic: true, + signature_hex: None, + signer_pubkey_hex: None, + }, + ) + } + + #[test] + fn verified_wifi_csi_event_is_stored_and_counted() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-1", true); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 5_000); + + assert_eq!(recent.len(), 1); + assert!(recent[0].verified); + assert_eq!(recent[0].context.device_id, "node-1"); + assert_eq!(recent[0].received_ns, 5_000); + assert_eq!(stats.polls, 1); + assert_eq!(stats.events_received, 1); + assert_eq!(stats.events_verified, 1); + assert_eq!(stats.events_unverified, 0); + assert_eq!(stats.events_other_modality, 0); + } + + #[test] + fn unsigned_event_is_stored_but_flagged_unverified() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-2", false); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 1); + + assert_eq!(recent.len(), 1, "unverified events are still visible, not dropped"); + assert!(!recent[0].verified); + assert_eq!(stats.events_unverified, 1); + assert_eq!(stats.events_verified, 0); + } + + #[test] + fn non_wifi_csi_event_is_not_stored_as_rf_context() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = other_modality_event(); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 1); + + assert!(recent.is_empty()); + assert_eq!(stats.events_other_modality, 1); + assert_eq!(stats.events_received, 1, "still counted as received"); + } + + #[test] + fn ring_is_bounded_and_drops_oldest() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + for i in 0..(RF_RECENT_CAPACITY + 10) { + let ev = wifi_csi_event(&format!("node-{i}"), true); + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), i as u64); + } + assert_eq!(recent.len(), RF_RECENT_CAPACITY); + // Oldest entries evicted first: the front should be from late in the + // sequence, not node-0. + assert_ne!(recent.front().unwrap().context.device_id, "node-0"); + assert_eq!( + recent.back().unwrap().context.device_id, + format!("node-{}", RF_RECENT_CAPACITY + 9) + ); + } + + #[test] + fn verified_events_actually_reach_the_fusion_engine() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-fused", true); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 1); + + let inferences = current_inferences(&fusion); + assert!( + inferences.iter().any(|i| i.label == "person_present"), + "a verified presence-bearing event should produce a person_present \ + inference from RuField's own default room-state rules, got: {inferences:?}" + ); + } + + #[test] + fn unverified_events_never_reach_the_fusion_engine() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-forged", false); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 1); + + let inferences = current_inferences(&fusion); + assert!( + inferences.is_empty(), + "an unverified event must never influence a fused inference, got: {inferences:?}" + ); + } + + #[test] + fn a_verified_batch_captures_an_inference_snapshot() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-hist", true); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 42); + + assert_eq!(history.len(), 1, "a batch that fed the engine must be captured"); + let snap = &history[0]; + assert_eq!(snap.captured_ns, 42); + assert!( + snap.inferences.iter().any(|i| i.label == "person_present"), + "the snapshot must carry the inference the batch actually produced" + ); + } + + #[test] + fn an_empty_or_all_unverified_batch_is_not_recorded_as_history_noise() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + + // Empty batch (e.g. RuView's ring had nothing new this poll). + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, &[], 1); + assert!(history.is_empty(), "an empty poll must not add a snapshot"); + + // All-unverified batch never fed the engine either. + let unsigned = wifi_csi_event("node-x", false); + ingest_events( + &mut recent, + &mut stats, + &mut fusion, + &mut history, + std::slice::from_ref(&unsigned), + 2, + ); + assert!( + history.is_empty(), + "a batch that never fed the engine must not add a snapshot either" + ); + } + + #[test] + fn inference_history_is_bounded_and_drops_oldest() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + for i in 0..(INFERENCE_HISTORY_CAPACITY + 5) { + let ev = wifi_csi_event(&format!("node-{i}"), true); + ingest_events( + &mut recent, + &mut stats, + &mut fusion, + &mut history, + std::slice::from_ref(&ev), + i as u64, + ); + } + assert_eq!(history.len(), INFERENCE_HISTORY_CAPACITY); + assert_eq!(history.back().unwrap().captured_ns, (INFERENCE_HISTORY_CAPACITY + 4) as u64); + } + + #[test] + fn poll_failure_is_counted_and_never_touches_the_ring() { + let recent: VecDeque = VecDeque::new(); + let mut stats = RfIngestStats::default(); + record_poll_failure(&mut stats); + record_poll_failure(&mut stats); + assert_eq!(stats.poll_failures, 2); + assert!(recent.is_empty()); + } +} diff --git a/crates/rucelium-gateway/src/state.rs b/crates/rucelium-gateway/src/state.rs index bad2f3c..684144f 100644 --- a/crates/rucelium-gateway/src/state.rs +++ b/crates/rucelium-gateway/src/state.rs @@ -9,6 +9,8 @@ use crate::config::GatewayConfig; use crate::journal; +use crate::rf_bridge::{InferenceSnapshot, RfEntry, RfIngestStats}; +use rufield_fusion::RuFieldFusion; use rucelium_calibration::{ AuthorityRegistry as CalibrationAuthorities, CalibrationAuthority, CalibrationError, CalibrationSigner, CalibrationStore, Calibrator, DriftDetector, @@ -24,7 +26,7 @@ use rucelium_store::{EventStore, ObservationStore}; use rucelium_transport::Reassembler; use rucelium_worldgraph::WorldGraph; use serde::Serialize; -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::path::PathBuf; use std::sync::Arc; use std::time::{Instant, SystemTime, UNIX_EPOCH}; @@ -180,6 +182,26 @@ pub struct Inner { pub control: ControlStats, /// Path of the durable command-phase journal (`commands.jsonl`). pub command_journal: PathBuf, + + // --- RuView RF-context bridge (ADR-264 §8, `rf_bridge`) --- + /// Bounded ring of recently ingested, verification-tagged RF context. + /// Empty and never populated unless `--rf-upstream` is configured. + pub rf_recent: VecDeque, + /// RF-context ingest counters. + pub rf_stats: RfIngestStats, + /// The configured RF upstream, if any — retained here (not just in + /// `GatewayConfig`, which the running daemon doesn't otherwise keep) so + /// `GET /api/rf-context` can report what it's actually polling. + pub rf_upstream: Option, + /// Within-RF-domain fusion engine (`rf_bridge` module docs): combines + /// this upstream's own verified `FieldEvent`s into room-state inferences. + /// Needs no biome/WorldGraph pairing — see `rf_bridge` for why that's a + /// deliberately different, still-unbuilt kind of fusion. + pub rf_fusion: RuFieldFusion, + /// Bounded in-memory history of inference snapshots (`rf_bridge` module + /// docs) — survives across polls, not across a restart; RF inferences + /// have no biome `node_id`/`sequence` to durably cite yet. + pub rf_inference_history: VecDeque, } impl Inner { @@ -284,6 +306,11 @@ impl Inner { receipts: Vec::new(), control: ControlStats::default(), command_journal, + rf_recent: VecDeque::new(), + rf_stats: RfIngestStats::default(), + rf_upstream: config.rf_upstream.clone(), + rf_fusion: RuFieldFusion::new(), + rf_inference_history: VecDeque::new(), }) } diff --git a/crates/rucelium-gateway/tests/cors.rs b/crates/rucelium-gateway/tests/cors.rs new file mode 100644 index 0000000..d13de96 --- /dev/null +++ b/crates/rucelium-gateway/tests/cors.rs @@ -0,0 +1,222 @@ +//! CORS + `/api/rf-context` contract guard for the read-only dashboard use +//! case (`api::dashboard_cors_layer` doc comment, `api::router`'s +//! `dashboard_origins` param). +//! +//! Addresses the two review rejections on the original split-PR shape +//! (RuView#1731 / RuCelium#3): +//! - the dashboard's `pollRuCelium()` contract (`upstream`, `stats`, +//! `room_state_inferences`) is asserted directly against a live +//! `/api/rf-context` response, not just unit-tested in `rf_bridge`; +//! - CORS is origin-gated and scoped to the two dashboard routes only — an +//! unconfigured/unknown origin gets no `Access-Control-Allow-Origin` at +//! all, a configured origin gets it on `/api/rf-context` and +//! `/api/stats`, and `POST` is never allowed regardless of origin. + +use rucelium_gateway::{spawn_gateway_with_state, GatewayConfig, GatewayState}; +use std::path::PathBuf; +use std::time::{SystemTime, UNIX_EPOCH}; + +const DASHBOARD_ORIGIN: &str = "http://127.0.0.1:8090"; +const UNKNOWN_ORIGIN: &str = "http://evil.example"; + +fn temp_dir(tag: &str) -> PathBuf { + let t = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock after epoch") + .as_nanos(); + std::env::temp_dir().join(format!("rucelium-gw-cors-{tag}-{}-{t}", std::process::id())) +} + +async fn spawn_with_dashboard_origin() -> (rucelium_gateway::GatewayHandle, PathBuf) { + let dir = temp_dir("dashboard"); + let cfg = GatewayConfig { + udp_port: 0, + http_port: 0, + data_dir: dir.clone(), + dashboard_origins: vec![DASHBOARD_ORIGIN.to_string()], + ..GatewayConfig::default() + }; + let state = GatewayState::open(&cfg).expect("open state"); + let handle = spawn_gateway_with_state(state, cfg) + .await + .expect("spawn gateway"); + (handle, dir) +} + +#[tokio::test(flavor = "multi_thread")] +async fn rf_context_returns_the_schema_the_dashboard_s_polling_depends_on() { + let (handle, dir) = spawn_with_dashboard_origin().await; + let base = format!("http://127.0.0.1:{}", handle.http_port); + + let body: serde_json::Value = reqwest::get(format!("{base}/api/rf-context")) + .await + .expect("GET /api/rf-context") + .json() + .await + .expect("valid JSON"); + + // The exact contract `ui/pipeline.html`'s `pollRuCelium()` reads — an + // idle bridge (no `--rf-upstream` configured here) still has to shape + // this correctly, not just when there happens to be live traffic. + assert!(body.get("upstream").is_some(), "missing `upstream`: {body}"); + assert!(body.get("stats").is_some(), "missing `stats`: {body}"); + assert!( + body.get("room_state_inferences").is_some(), + "missing `room_state_inferences`: {body}" + ); + assert!( + body["room_state_inferences"].is_array(), + "`room_state_inferences` must be an array: {body}" + ); + assert_eq!( + body["upstream"], + serde_json::Value::Null, + "no --rf-upstream was configured, so it must be honestly null, not fabricated" + ); + + std::fs::remove_dir_all(&dir).ok(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_configured_dashboard_origin_gets_cors_on_both_dashboard_routes() { + let (handle, dir) = spawn_with_dashboard_origin().await; + let base = format!("http://127.0.0.1:{}", handle.http_port); + let client = reqwest::Client::new(); + + for path in ["/api/rf-context", "/api/stats"] { + let resp = client + .get(format!("{base}{path}")) + .header("Origin", DASHBOARD_ORIGIN) + .send() + .await + .unwrap_or_else(|e| panic!("GET {path}: {e}")); + assert!(resp.status().is_success(), "GET {path} failed: {resp:?}"); + let allow_origin = resp + .headers() + .get("access-control-allow-origin") + .unwrap_or_else(|| panic!("{path}: CORS header must be present for the configured origin")) + .to_str() + .unwrap(); + assert_eq!( + allow_origin, DASHBOARD_ORIGIN, + "{path}: must echo the configured origin, not a wildcard" + ); + } + + std::fs::remove_dir_all(&dir).ok(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_unconfigured_origin_gets_no_cors_header_on_dashboard_routes() { + let (handle, dir) = spawn_with_dashboard_origin().await; + let base = format!("http://127.0.0.1:{}", handle.http_port); + let client = reqwest::Client::new(); + + let resp = client + .get(format!("{base}/api/rf-context")) + .header("Origin", UNKNOWN_ORIGIN) + .send() + .await + .expect("GET /api/rf-context"); + + // The server still answers (this was never an auth control — see the + // module docs), but the browser gets no ACAO for this origin, so the + // browser's own same-origin policy still blocks the read. + assert!(resp.status().is_success()); + assert!( + resp.headers().get("access-control-allow-origin").is_none(), + "an unconfigured origin must not receive an Access-Control-Allow-Origin header" + ); + + std::fs::remove_dir_all(&dir).ok(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn no_dashboard_origin_configured_means_no_cors_header_at_all() { + // The default posture (empty --dashboard-origin): CORS is fully + // disabled, matching the gateway's pre-dashboard behavior exactly. + let dir = temp_dir("no-origin"); + let cfg = GatewayConfig { + udp_port: 0, + http_port: 0, + data_dir: dir.clone(), + ..GatewayConfig::default() + }; + let state = GatewayState::open(&cfg).expect("open state"); + let handle = spawn_gateway_with_state(state, cfg) + .await + .expect("spawn gateway"); + let base = format!("http://127.0.0.1:{}", handle.http_port); + + let client = reqwest::Client::new(); + let resp = client + .get(format!("{base}/api/rf-context")) + .header("Origin", DASHBOARD_ORIGIN) + .send() + .await + .expect("GET /api/rf-context"); + + assert!(resp.status().is_success()); + assert!( + resp.headers().get("access-control-allow-origin").is_none(), + "dashboard_origins defaults to empty, so CORS must be off by default" + ); + + std::fs::remove_dir_all(&dir).ok(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn cors_never_applies_outside_the_two_dashboard_routes() { + let (handle, dir) = spawn_with_dashboard_origin().await; + let base = format!("http://127.0.0.1:{}", handle.http_port); + let client = reqwest::Client::new(); + + // Even the configured dashboard origin gets nothing on a route the + // CORS layer was never applied to — the admin/federation/observation + // surfaces stay same-origin-only in the browser no matter what. + let resp = client + .get(format!("{base}/api/observations/recent")) + .header("Origin", DASHBOARD_ORIGIN) + .send() + .await + .expect("GET /api/observations/recent"); + + assert!(resp.status().is_success()); + assert!( + resp.headers().get("access-control-allow-origin").is_none(), + "CORS must be scoped to /api/rf-context and /api/stats only" + ); + + std::fs::remove_dir_all(&dir).ok(); +} + +#[tokio::test(flavor = "multi_thread")] +async fn preflight_for_a_write_route_does_not_allow_post() { + let (handle, dir) = spawn_with_dashboard_origin().await; + let base = format!("http://127.0.0.1:{}", handle.http_port); + + let client = reqwest::Client::new(); + // A browser preflight (OPTIONS) for a cross-origin POST to an admin + // route must NOT come back allowing POST — the CORS policy is + // deliberately GET-only (dashboard reads, not cross-origin writes), + // and the admin route isn't even inside the CORS-scoped sub-router. + let resp = client + .request(reqwest::Method::OPTIONS, format!("{base}/api/admin/command")) + .header("Origin", DASHBOARD_ORIGIN) + .header("Access-Control-Request-Method", "POST") + .send() + .await + .expect("OPTIONS preflight"); + + let allow_methods = resp + .headers() + .get("access-control-allow-methods") + .map(|v| v.to_str().unwrap_or("").to_string()) + .unwrap_or_default(); + assert!( + !allow_methods.contains("POST"), + "CORS policy must stay GET-only, got allow-methods={allow_methods:?}" + ); + + std::fs::remove_dir_all(&dir).ok(); +}