synheart_sensor_agent/
server.rs

1//! HTTP server for receiving behavioral data from Chrome extension.
2//!
3//! This module provides an HTTP server that:
4//! - Accepts raw behavioral data via POST /collect
5//! - Returns acknowledgment to the caller
6//!
7//! Orchestration (forwarding, feature extraction, storage) is handled by
8//! synheart-core-rust, not by the sensor agent.
9
10use axum::{
11    extract::{DefaultBodyLimit, State},
12    http::{header::AUTHORIZATION, HeaderMap, StatusCode},
13    routing::{get, post},
14    Json, Router,
15};
16use serde::{Deserialize, Serialize};
17use std::net::SocketAddr;
18use std::path::PathBuf;
19use std::sync::Arc;
20use tokio::net::TcpListener;
21use tower_http::cors::{AllowOrigin, Any, CorsLayer};
22
23/// Maximum accepted `POST /collect` body size (256 KiB). Behavioral payloads are
24/// small interaction-metadata blobs; anything larger is rejected with `413`.
25const MAX_BODY_BYTES: usize = 256 * 1024;
26
27/// Server configuration
28#[derive(Debug, Clone)]
29pub struct ServerConfig {
30    /// Port to bind to (0 for random)
31    pub port: u16,
32    /// State directory for data
33    pub state_dir: PathBuf,
34    /// Optional bearer token required on `POST /collect`.
35    ///
36    /// When `Some`, requests must carry a matching `Authorization: Bearer <token>`
37    /// header or they are rejected with `401`. When `None`, the endpoint is open
38    /// (intended for trusted loopback-only use).
39    pub token: Option<String>,
40}
41
42impl ServerConfig {
43    /// Create a new server configuration without authentication.
44    pub fn new(port: u16, state_dir: PathBuf) -> Self {
45        Self {
46            port,
47            state_dir,
48            token: None,
49        }
50    }
51
52    /// Require a bearer token on `POST /collect`.
53    pub fn with_token(mut self, token: Option<String>) -> Self {
54        self.token = token.filter(|t| !t.is_empty());
55        self
56    }
57}
58
59/// Shared server state
60pub struct ServerState {
61    /// State directory
62    #[allow(dead_code)]
63    state_dir: PathBuf,
64    /// Expected bearer token, if authentication is enabled.
65    token: Option<String>,
66}
67
68impl ServerState {
69    /// Create new server state
70    pub fn new(config: &ServerConfig) -> Self {
71        Self {
72            state_dir: config.state_dir.clone(),
73            token: config.token.clone(),
74        }
75    }
76}
77
78/// Behavioral session data from Chrome extension
79#[derive(Debug, Clone, Serialize, Deserialize)]
80pub struct BehavioralSession {
81    /// The behavioral session payload
82    pub session: serde_json::Value,
83}
84
85/// Response from collect endpoint
86#[derive(Debug, Clone, Serialize)]
87pub struct CollectResponse {
88    /// Processing status.
89    pub status: String,
90    /// Human-readable description.
91    pub message: String,
92}
93
94/// Health check response
95#[derive(Serialize)]
96pub struct HealthResponse {
97    /// Service health status.
98    pub status: String,
99    /// Agent version string.
100    pub version: String,
101}
102
103/// Error response
104#[derive(Serialize)]
105pub struct ErrorResponse {
106    /// Human-readable error message.
107    pub error: String,
108    /// Machine-readable error code.
109    pub code: String,
110}
111
112/// GET /health
113async fn health() -> Json<HealthResponse> {
114    Json(HealthResponse {
115        status: "ok".to_string(),
116        version: env!("CARGO_PKG_VERSION").to_string(),
117    })
118}
119
120/// Verify the request's bearer token against the configured one.
121///
122/// Returns `Ok(())` when authentication is disabled (no token configured) or when
123/// a matching `Authorization: Bearer <token>` header is present; otherwise `401`.
124fn check_auth(
125    state: &ServerState,
126    headers: &HeaderMap,
127) -> Result<(), (StatusCode, Json<ErrorResponse>)> {
128    let Some(expected) = state.token.as_deref() else {
129        return Ok(());
130    };
131
132    let provided = headers
133        .get(AUTHORIZATION)
134        .and_then(|v| v.to_str().ok())
135        .and_then(|v| v.strip_prefix("Bearer "));
136
137    match provided {
138        Some(token) if constant_time_eq(token.as_bytes(), expected.as_bytes()) => Ok(()),
139        _ => Err((
140            StatusCode::UNAUTHORIZED,
141            Json(ErrorResponse {
142                error: "Missing or invalid bearer token".to_string(),
143                code: "UNAUTHORIZED".to_string(),
144            }),
145        )),
146    }
147}
148
149/// Compare two byte strings in constant time relative to their content.
150///
151/// Guards the bearer-token check against timing attacks that could otherwise
152/// recover the token byte-by-byte from response-time differences. The length is
153/// not secret (it short-circuits), but equal-length inputs are always fully
154/// scanned regardless of where they first differ.
155fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
156    if a.len() != b.len() {
157        return false;
158    }
159    let mut diff = 0u8;
160    for (x, y) in a.iter().zip(b.iter()) {
161        diff |= x ^ y;
162    }
163    diff == 0
164}
165
166/// POST /collect
167///
168/// Accepts raw behavioral data. Validates the bearer token (when configured) and
169/// the JSON structure, then acknowledges. The caller (synheart-core-rust) is
170/// responsible for feature extraction.
171async fn collect(
172    State(state): State<Arc<ServerState>>,
173    headers: HeaderMap,
174    Json(data): Json<BehavioralSession>,
175) -> Result<Json<CollectResponse>, (StatusCode, Json<ErrorResponse>)> {
176    check_auth(&state, &headers)?;
177
178    if !data.session.is_object() {
179        return Err((
180            StatusCode::BAD_REQUEST,
181            Json(ErrorResponse {
182                error: "Session payload must be a JSON object".to_string(),
183                code: "INVALID_SESSION".to_string(),
184            }),
185        ));
186    }
187
188    Ok(Json(CollectResponse {
189        status: "ok".to_string(),
190        message: "Data accepted".to_string(),
191    }))
192}
193
194/// Run the HTTP server
195pub async fn run(
196    config: ServerConfig,
197) -> anyhow::Result<(SocketAddr, tokio::sync::oneshot::Sender<()>)> {
198    let state = Arc::new(ServerState::new(&config));
199
200    let app = Router::new()
201        .route("/health", get(health))
202        .route("/collect", post(collect))
203        // Behavioral payloads are small metadata blobs; cap the body to reject
204        // oversized requests early (defense against memory-exhaustion attempts).
205        .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
206        .layer(
207            // Allow browser extensions and local pages only. Extension origins
208            // are `chrome-extension://<extension-id>` (the id varies per
209            // install, so a prefix match is required), and localhost origins
210            // carry a port (`http://localhost:<port>`), so exact-match values
211            // would never match a real request. The server binds to loopback
212            // and supports bearer-token auth; CORS is not the auth boundary.
213            CorsLayer::new()
214                .allow_origin(AllowOrigin::predicate(|origin, _| {
215                    let Ok(origin) = origin.to_str() else {
216                        return false;
217                    };
218                    origin.starts_with("chrome-extension://")
219                        || origin == "http://localhost"
220                        || origin.starts_with("http://localhost:")
221                        || origin == "http://127.0.0.1"
222                        || origin.starts_with("http://127.0.0.1:")
223                }))
224                .allow_methods(Any)
225                .allow_headers(Any),
226        )
227        .with_state(state);
228
229    let addr = SocketAddr::from(([127, 0, 0, 1], config.port));
230    let listener = TcpListener::bind(addr).await?;
231    let actual_addr = listener.local_addr()?;
232
233    tracing::info!("Sensor agent server listening on http://{}", actual_addr);
234
235    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
236
237    tokio::spawn(async move {
238        if let Err(e) = axum::serve(listener, app)
239            .with_graceful_shutdown(async {
240                let _ = shutdown_rx.await;
241                tracing::info!("Server shutdown signal received");
242            })
243            .await
244        {
245            tracing::error!("Server error: {}", e);
246        }
247    });
248
249    Ok((actual_addr, shutdown_tx))
250}
synheart_sensor_agent/server.rs

synheart_sensor_agent/
server.rs
