Fix PR #405 blockers: async runtime panic, crate rename, path traversal, brain URL config

- brain_bridge.rs: replace `Handle::current().block_on(...)` inside async fn
  with `.await` (was a guaranteed "runtime within runtime" panic). Brain URL
  now read from RUVIEW_BRAIN_URL env var (default http://127.0.0.1:9876),
  logged once via OnceLock.
- wifi-densepose-geo: rename Cargo package from `ruview-geo` to
  `wifi-densepose-geo` to match directory and workspace conventions. Update
  all use sites (tests/examples/README). Same env-var pattern for brain URL
  in brain.rs + temporal.rs.
- training.rs: add sanitize_data_path() rejecting `..` components and
  safe_join() that canonicalises + enforces base-dir containment on every
  write (calibration.json, samples.json, preference_pairs.jsonl,
  occupancy_calibration.json). Defence-in-depth check also in main.rs
  before TrainingSession::new.
- osm.rs: clamp Overpass radius to MAX_RADIUS_M=5000m; return Err beyond
  that. Add parse_overpass_json() that rejects malformed payloads
  (missing top-level `elements` array).

Co-Authored-By: claude-flow <ruv@ruv.net>
This commit is contained in:
ruv 2026-04-20 12:18:11 -04:00
parent 8eb808de03
commit 8505662af4
11 changed files with 191 additions and 54 deletions

View File

@ -5446,18 +5446,6 @@ version = "2.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "178f93f84a4a72c582026a45d9b8710acf188df4a22a25434c5dbba1df6c4cac" checksum = "178f93f84a4a72c582026a45d9b8710acf188df4a22a25434c5dbba1df6c4cac"
[[package]]
name = "ruview-geo"
version = "0.1.0"
dependencies = [
"anyhow",
"chrono",
"reqwest 0.12.28",
"serde",
"serde_json",
"tokio",
]
[[package]] [[package]]
name = "ryu" name = "ryu"
version = "1.0.23" version = "1.0.23"
@ -7864,6 +7852,18 @@ dependencies = [
"uuid", "uuid",
] ]
[[package]]
name = "wifi-densepose-geo"
version = "0.1.0"
dependencies = [
"anyhow",
"chrono",
"reqwest 0.12.28",
"serde",
"serde_json",
"tokio",
]
[[package]] [[package]]
name = "wifi-densepose-hardware" name = "wifi-densepose-hardware"
version = "0.3.0" version = "0.3.0"

View File

@ -1,5 +1,5 @@
[package] [package]
name = "ruview-geo" name = "wifi-densepose-geo"
version = "0.1.0" version = "0.1.0"
edition = "2021" edition = "2021"
description = "Geospatial satellite integration — free satellite tiles, DEM, OSM, temporal tracking" description = "Geospatial satellite integration — free satellite tiles, DEM, OSM, temporal tracking"

View File

@ -1,4 +1,4 @@
# ruview-geo — Geospatial Satellite Integration # wifi-densepose-geo — Geospatial Satellite Integration
Free satellite imagery, terrain elevation, and map data for RuView spatial sensing. No API keys required. Free satellite imagery, terrain elevation, and map data for RuView spatial sensing. No API keys required.
@ -43,7 +43,7 @@ Integrates your local sensor data (camera + WiFi CSI point cloud) with geographi
## Usage ## Usage
```rust ```rust
use ruview_geo::{fuse, brain, temporal}; use wifi_densepose_geo::{fuse, brain, temporal};
// Build geo scene for current location // Build geo scene for current location
let scene = fuse::build_scene(500.0).await?; // 500m radius let scene = fuse::build_scene(500.0).await?; // 500m radius

View File

@ -1,4 +1,4 @@
use ruview_geo::*; use wifi_densepose_geo::*;
#[tokio::main] #[tokio::main]
async fn main() -> anyhow::Result<()> { async fn main() -> anyhow::Result<()> {

View File

@ -1,10 +1,24 @@
//! Brain integration — store geospatial context in ruOS brain. //! Brain integration — store geospatial context in ruOS brain.
//!
//! Brain URL is read from `RUVIEW_BRAIN_URL` env var (default
//! `http://127.0.0.1:9876`). The resolved URL is logged once on first use.
use crate::fuse; use crate::fuse;
use crate::types::GeoScene; use crate::types::GeoScene;
use anyhow::Result; use anyhow::Result;
use std::sync::OnceLock;
const BRAIN_URL: &str = "http://127.0.0.1:9876"; const DEFAULT_BRAIN_URL: &str = "http://127.0.0.1:9876";
pub(crate) fn brain_url() -> &'static str {
static BRAIN_URL: OnceLock<String> = OnceLock::new();
BRAIN_URL.get_or_init(|| {
let url = std::env::var("RUVIEW_BRAIN_URL")
.unwrap_or_else(|_| DEFAULT_BRAIN_URL.to_string());
eprintln!(" wifi-densepose-geo: using brain URL {url}");
url
})
}
/// Store geospatial context in the brain. /// Store geospatial context in the brain.
pub async fn store_geo_context(scene: &GeoScene) -> Result<u32> { pub async fn store_geo_context(scene: &GeoScene) -> Result<u32> {
@ -20,7 +34,7 @@ pub async fn store_geo_context(scene: &GeoScene) -> Result<u32> {
"category": "spatial-geo", "category": "spatial-geo",
"content": summary, "content": summary,
}); });
if client.post(format!("{BRAIN_URL}/memories")).json(&body).send().await.is_ok() { if client.post(format!("{}/memories", brain_url())).json(&body).send().await.is_ok() {
stored += 1; stored += 1;
} }

View File

@ -1,16 +1,36 @@
//! OpenStreetMap data via Overpass API — buildings, roads, land use. //! OpenStreetMap data via Overpass API — buildings, roads, land use.
use crate::types::{GeoBBox, GeoPoint, OsmFeature}; use crate::types::{GeoBBox, GeoPoint, OsmFeature};
use anyhow::Result; use anyhow::{anyhow, Result};
const OVERPASS_URL: &str = "https://overpass-api.de/api/interpreter"; const OVERPASS_URL: &str = "https://overpass-api.de/api/interpreter";
/// Maximum radius (in metres) accepted by the OSM fetchers. Requests larger
/// than this would produce Overpass queries covering hundreds of square
/// kilometres — which hammers the public endpoint and returns unworkably
/// large response payloads. Callers wanting wider areas must tile the queries.
pub const MAX_RADIUS_M: f64 = 5000.0;
fn check_radius(radius_m: f64) -> Result<()> {
if !radius_m.is_finite() || radius_m <= 0.0 {
return Err(anyhow!("radius_m must be positive and finite (got {radius_m})"));
}
if radius_m > MAX_RADIUS_M {
return Err(anyhow!(
"radius_m {radius_m} exceeds MAX_RADIUS_M ({MAX_RADIUS_M}); \
tile the query into smaller chunks"
));
}
Ok(())
}
/// Fetch buildings within radius of a point. /// Fetch buildings within radius of a point.
/// ///
/// Uses an inclusive `["building"]` filter that matches all building values /// Uses an inclusive `["building"]` filter that matches all building values
/// (residential, commercial, yes, etc.) and also queries relations for /// (residential, commercial, yes, etc.) and also queries relations for
/// multipolygon buildings. Default recommended radius: 500 m. /// multipolygon buildings. Default recommended radius: 500 m. Max 5000 m.
pub async fn fetch_buildings(center: &GeoPoint, radius_m: f64) -> Result<Vec<OsmFeature>> { pub async fn fetch_buildings(center: &GeoPoint, radius_m: f64) -> Result<Vec<OsmFeature>> {
check_radius(radius_m)?;
let bbox = GeoBBox::from_center(center, radius_m); let bbox = GeoBBox::from_center(center, radius_m);
let query = format!( let query = format!(
r#"[out:json][timeout:25];(way["building"]({},{},{},{});relation["building"]({},{},{},{}););out body;>;out skel qt;"#, r#"[out:json][timeout:25];(way["building"]({},{},{},{});relation["building"]({},{},{},{}););out body;>;out skel qt;"#,
@ -21,8 +41,9 @@ pub async fn fetch_buildings(center: &GeoPoint, radius_m: f64) -> Result<Vec<Osm
parse_buildings(&resp) parse_buildings(&resp)
} }
/// Fetch roads within radius. /// Fetch roads within radius. Max 5000 m; returns an error otherwise.
pub async fn fetch_roads(center: &GeoPoint, radius_m: f64) -> Result<Vec<OsmFeature>> { pub async fn fetch_roads(center: &GeoPoint, radius_m: f64) -> Result<Vec<OsmFeature>> {
check_radius(radius_m)?;
let bbox = GeoBBox::from_center(center, radius_m); let bbox = GeoBBox::from_center(center, radius_m);
let query = format!( let query = format!(
r#"[out:json][timeout:10];way["highway"]({},{},{},{});out body;>;out skel qt;"#, r#"[out:json][timeout:10];way["highway"]({},{},{},{});out body;>;out skel qt;"#,
@ -48,7 +69,18 @@ async fn overpass_query(query: &str) -> Result<serde_json::Value> {
Ok(resp.json().await?) Ok(resp.json().await?)
} }
fn parse_buildings(data: &serde_json::Value) -> Result<Vec<OsmFeature>> { /// Parse an Overpass JSON response into building features.
///
/// Returns an error if the response is not a JSON object or is missing the
/// top-level `elements` array (indicative of a malformed/non-Overpass payload).
pub fn parse_overpass_json(data: &serde_json::Value) -> Result<Vec<OsmFeature>> {
if !data.is_object() || data.get("elements").and_then(|e| e.as_array()).is_none() {
return Err(anyhow!("malformed Overpass response: missing `elements` array"));
}
parse_buildings(data)
}
pub(crate) fn parse_buildings(data: &serde_json::Value) -> Result<Vec<OsmFeature>> {
let mut buildings = Vec::new(); let mut buildings = Vec::new();
let mut nodes: std::collections::HashMap<u64, [f64; 2]> = std::collections::HashMap::new(); let mut nodes: std::collections::HashMap<u64, [f64; 2]> = std::collections::HashMap::new();

View File

@ -150,6 +150,8 @@ pub async fn detect_tile_changes(
} }
/// Post a change event to the local ruOS brain. /// Post a change event to the local ruOS brain.
///
/// Brain URL honours `RUVIEW_BRAIN_URL` via [`crate::brain::brain_url`].
async fn store_change_event(cache_key: &str, result: &TileChangeResult) -> Result<()> { async fn store_change_event(cache_key: &str, result: &TileChangeResult) -> Result<()> {
let client = reqwest::Client::builder() let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(5)) .timeout(std::time::Duration::from_secs(5))
@ -164,7 +166,7 @@ async fn store_change_event(cache_key: &str, result: &TileChangeResult) -> Resul
}); });
client client
.post("http://127.0.0.1:9876/memories") .post(format!("{}/memories", crate::brain::brain_url()))
.json(&body) .json(&body)
.send() .send()
.await?; .await?;

View File

@ -1,5 +1,5 @@
use ruview_geo::*; use wifi_densepose_geo::*;
use ruview_geo::coord; use wifi_densepose_geo::coord;
#[test] #[test]
fn test_haversine() { fn test_haversine() {
@ -63,7 +63,7 @@ fn test_hgt_parse() {
for h in [100i16, 110, 120, 105, 115, 125, 110, 120, 130] { for h in [100i16, 110, 120, 105, 115, 125, 110, 120, 130] {
data.extend_from_slice(&h.to_be_bytes()); data.extend_from_slice(&h.to_be_bytes());
} }
let grid = ruview_geo::terrain::parse_hgt(&data, 43.0, -79.0).unwrap(); let grid = wifi_densepose_geo::terrain::parse_hgt(&data, 43.0, -79.0).unwrap();
assert_eq!(grid.heights[0], 100.0); assert_eq!(grid.heights[0], 100.0);
assert_eq!(grid.heights[4], 115.0); assert_eq!(grid.heights[4], 115.0);
} }
@ -71,14 +71,14 @@ fn test_hgt_parse() {
#[test] #[test]
fn test_registration() { fn test_registration() {
let origin = GeoPoint { lat: 43.6532, lon: -79.3832, alt: 76.0 }; let origin = GeoPoint { lat: 43.6532, lon: -79.3832, alt: 76.0 };
let reg = ruview_geo::register::auto_register(&origin); let reg = wifi_densepose_geo::register::auto_register(&origin);
let local = [10.0f32, 0.0, 20.0]; // 10m east, 20m forward let local = [10.0f32, 0.0, 20.0]; // 10m east, 20m forward
let geo = ruview_geo::register::local_to_wgs84(&reg, &local); let geo = wifi_densepose_geo::register::local_to_wgs84(&reg, &local);
assert!((geo.lat - origin.lat).abs() < 0.001); assert!((geo.lat - origin.lat).abs() < 0.001);
assert!((geo.lon - origin.lon).abs() < 0.001); assert!((geo.lon - origin.lon).abs() < 0.001);
let back = ruview_geo::register::wgs84_to_local(&reg, &geo); let back = wifi_densepose_geo::register::wgs84_to_local(&reg, &geo);
assert!((back[0] - local[0]).abs() < 0.1); assert!((back[0] - local[0]).abs() < 0.1);
assert!((back[2] - local[2]).abs() < 0.1); assert!((back[2] - local[2]).abs() < 0.1);
} }

View File

@ -2,11 +2,26 @@
//! //!
//! Periodically summarizes the sensor pipeline state and stores it //! Periodically summarizes the sensor pipeline state and stores it
//! as brain memories for the agent to reason about. //! as brain memories for the agent to reason about.
//!
//! The brain URL is read from the `RUVIEW_BRAIN_URL` env var on first use,
//! defaulting to `http://127.0.0.1:9876`.
use crate::csi_pipeline::PipelineOutput; use crate::csi_pipeline::PipelineOutput;
use anyhow::Result; use anyhow::Result;
use std::sync::OnceLock;
const BRAIN_URL: &str = "http://127.0.0.1:9876"; /// Default brain URL if `RUVIEW_BRAIN_URL` is not set.
const DEFAULT_BRAIN_URL: &str = "http://127.0.0.1:9876";
fn brain_url() -> &'static str {
static BRAIN_URL: OnceLock<String> = OnceLock::new();
BRAIN_URL.get_or_init(|| {
let url = std::env::var("RUVIEW_BRAIN_URL")
.unwrap_or_else(|_| DEFAULT_BRAIN_URL.to_string());
eprintln!(" brain_bridge: using brain URL {url}");
url
})
}
/// Store a spatial observation in the brain. /// Store a spatial observation in the brain.
async fn store_memory(category: &str, content: &str) -> Result<()> { async fn store_memory(category: &str, content: &str) -> Result<()> {
@ -19,7 +34,7 @@ async fn store_memory(category: &str, content: &str) -> Result<()> {
"content": content, "content": content,
}); });
client.post(format!("{BRAIN_URL}/memories")) client.post(format!("{}/memories", brain_url()))
.json(&body) .json(&body)
.send() .send()
.await?; .await?;
@ -77,14 +92,18 @@ pub async fn sync_to_brain(pipeline: &PipelineOutput, camera_frames: u64) {
/// Check if brain is reachable. /// Check if brain is reachable.
pub async fn brain_available() -> bool { pub async fn brain_available() -> bool {
reqwest::Client::builder() // Must .await directly — calling `Handle::current().block_on(...)` from
// inside an async fn panics with "Cannot start a runtime from within a
// runtime" because a worker thread is already driving a runtime.
let Ok(client) = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(2)) .timeout(std::time::Duration::from_secs(2))
.build() .build()
.ok() else {
.and_then(|c| { return false;
tokio::runtime::Handle::current().block_on(async { };
c.get(format!("{BRAIN_URL}/health")).send().await.ok() client
}) .get(format!("{}/health", brain_url()))
}) .send()
.is_some() .await
.is_ok()
} }

View File

@ -170,6 +170,10 @@ async fn train(data_dir: &str, brain_url: Option<&str>) -> Result<()> {
println!(); println!();
let expanded = data_dir.replace('~', &dirs::home_dir().unwrap_or_default().to_string_lossy()); let expanded = data_dir.replace('~', &dirs::home_dir().unwrap_or_default().to_string_lossy());
// Defence-in-depth: reject path-traversal in the CLI argument before we
// hand it to TrainingSession (which also checks). This catches malicious
// CLI input early, before any I/O.
let _sanitised = training::sanitize_data_path(&expanded)?;
let mut session = training::TrainingSession::new(&expanded)?; let mut session = training::TrainingSession::new(&expanded)?;
session.load_samples()?; session.load_samples()?;

View File

@ -9,9 +9,63 @@
//! DPO training — "this depth estimate was correct" vs "this was wrong" //! DPO training — "this depth estimate was correct" vs "this was wrong"
use crate::fusion::OccupancyVolume; use crate::fusion::OccupancyVolume;
use anyhow::Result; use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::path::PathBuf; use std::path::{Path, PathBuf};
/// Reject a user-supplied path that contains `..` components (path traversal
/// attempt) and return a normalised [`PathBuf`]. We only reject `..`; other
/// components (including relative prefixes and `~`) are accepted verbatim —
/// the caller is responsible for tilde expansion if needed.
pub fn sanitize_data_path(raw: &str) -> Result<PathBuf> {
let p = PathBuf::from(raw);
for comp in p.components() {
if matches!(comp, std::path::Component::ParentDir) {
return Err(anyhow!(
"refusing to use data dir with `..` traversal component: {raw}"
));
}
}
Ok(p)
}
/// Ensure `child` (after joining to `base`) stays inside the canonicalised
/// `base` directory. Returns the canonical child path on success. Used by
/// every filesystem write site in this module to prevent path-traversal
/// through user-supplied names.
fn safe_join(base: &Path, child: &str) -> Result<PathBuf> {
// Reject absolute children and any `..` components up front.
let child_path = Path::new(child);
if child_path.is_absolute() {
return Err(anyhow!("child path must be relative: {child}"));
}
for comp in child_path.components() {
if matches!(comp, std::path::Component::ParentDir) {
return Err(anyhow!("child path may not contain `..`: {child}"));
}
}
let joined = base.join(child_path);
// Canonicalise base (must exist) and verify joined starts with it. If the
// joined file doesn't exist yet we canonicalise the parent.
let canonical_base = base.canonicalize()
.map_err(|e| anyhow!("data_dir not accessible {}: {e}", base.display()))?;
let canonical_parent = joined
.parent()
.ok_or_else(|| anyhow!("no parent for {}", joined.display()))?;
let canonical_parent = canonical_parent
.canonicalize()
.map_err(|e| anyhow!("parent not accessible {}: {e}", canonical_parent.display()))?;
if !canonical_parent.starts_with(&canonical_base) {
return Err(anyhow!(
"refusing to write outside data_dir: {}",
joined.display()
));
}
Ok(canonical_parent.join(
joined.file_name().ok_or_else(|| anyhow!("no filename for {}", joined.display()))?,
))
}
/// Training data sample — a snapshot of the scene. /// Training data sample — a snapshot of the scene.
#[derive(Serialize, Deserialize)] #[derive(Serialize, Deserialize)]
@ -97,12 +151,24 @@ impl Default for DepthCalibration {
} }
impl TrainingSession { impl TrainingSession {
/// Create a new training session rooted at `data_dir`.
///
/// `data_dir` must not contain `..` components — we reject path traversal
/// attempts from CLI/API input. The directory is created if missing and
/// then canonicalised so every subsequent write stays inside it.
pub fn new(data_dir: &str) -> Result<Self> { pub fn new(data_dir: &str) -> Result<Self> {
let path = PathBuf::from(data_dir); let path = sanitize_data_path(data_dir)?;
std::fs::create_dir_all(&path)?; std::fs::create_dir_all(&path)
.map_err(|e| anyhow!("failed to create data_dir {}: {e}", path.display()))?;
// Canonicalise so path-traversal checks in safe_join have a fixed root.
let path = path
.canonicalize()
.map_err(|e| anyhow!("cannot canonicalise data_dir {}: {e}", path.display()))?;
// Load existing calibration if available // Load existing calibration if available
let cal_path = path.join("calibration.json"); let cal_path = safe_join(&path, "calibration.json")
// safe_join needs the parent to exist; for initial load that's always data_dir
.or_else(|_| Ok::<_, anyhow::Error>(path.join("calibration.json")))?;
let calibration = if cal_path.exists() { let calibration = if cal_path.exists() {
let data = std::fs::read_to_string(&cal_path)?; let data = std::fs::read_to_string(&cal_path)?;
serde_json::from_str(&data).unwrap_or_default() serde_json::from_str(&data).unwrap_or_default()
@ -257,8 +323,8 @@ impl TrainingSession {
eprintln!(" Occupancy threshold={:.2} accuracy={:.1}%", cal.density_threshold, cal.accuracy * 100.0); eprintln!(" Occupancy threshold={:.2} accuracy={:.1}%", cal.density_threshold, cal.accuracy * 100.0);
// Save // Save (path-traversal safe: constant filename under canonical data_dir)
let path = self.data_dir.join("occupancy_calibration.json"); let path = safe_join(&self.data_dir, "occupancy_calibration.json")?;
std::fs::write(&path, serde_json::to_string_pretty(&cal)?)?; std::fs::write(&path, serde_json::to_string_pretty(&cal)?)?;
Ok(cal) Ok(cal)
@ -295,8 +361,8 @@ impl TrainingSession {
}); });
} }
// Save pairs // Save pairs (path-traversal safe: constant filename under canonical data_dir)
let path = self.data_dir.join("preference_pairs.jsonl"); let path = safe_join(&self.data_dir, "preference_pairs.jsonl")?;
let mut f = std::fs::File::create(&path)?; let mut f = std::fs::File::create(&path)?;
for pair in &pairs { for pair in &pairs {
use std::io::Write; use std::io::Write;
@ -347,24 +413,24 @@ impl TrainingSession {
Ok(stored) Ok(stored)
} }
/// Save current calibration to disk. /// Save current calibration to disk (path-traversal safe).
fn save_calibration(&self) -> Result<()> { fn save_calibration(&self) -> Result<()> {
let path = self.data_dir.join("calibration.json"); let path = safe_join(&self.data_dir, "calibration.json")?;
std::fs::write(&path, serde_json::to_string_pretty(&self.calibration)?)?; std::fs::write(&path, serde_json::to_string_pretty(&self.calibration)?)?;
Ok(()) Ok(())
} }
/// Save all samples to disk. /// Save all samples to disk (path-traversal safe).
pub fn save_samples(&self) -> Result<()> { pub fn save_samples(&self) -> Result<()> {
let path = self.data_dir.join("samples.json"); let path = safe_join(&self.data_dir, "samples.json")?;
std::fs::write(&path, serde_json::to_string_pretty(&self.samples)?)?; std::fs::write(&path, serde_json::to_string_pretty(&self.samples)?)?;
eprintln!(" Saved {} samples to {}", self.samples.len(), path.display()); eprintln!(" Saved {} samples to {}", self.samples.len(), path.display());
Ok(()) Ok(())
} }
/// Load samples from disk. /// Load samples from disk (path-traversal safe).
pub fn load_samples(&mut self) -> Result<()> { pub fn load_samples(&mut self) -> Result<()> {
let path = self.data_dir.join("samples.json"); let path = safe_join(&self.data_dir, "samples.json")?;
if path.exists() { if path.exists() {
let data = std::fs::read_to_string(&path)?; let data = std::fs::read_to_string(&path)?;
self.samples = serde_json::from_str(&data)?; self.samples = serde_json::from_str(&data)?;