synheart-sensor-agent 0.4.0

//! HTTP server for receiving behavioral data from Chrome extension.
//!
//! This module provides an HTTP server that:
//! - Accepts raw behavioral data via POST /collect
//! - Returns acknowledgment to the caller
//!
//! Orchestration (forwarding, feature extraction, storage) is handled by
//! synheart-core-rust, not by the sensor agent.

use axum::{
    extract::{DefaultBodyLimit, State},
    http::{header::AUTHORIZATION, HeaderMap, StatusCode},
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::net::TcpListener;
use tower_http::cors::{AllowOrigin, Any, CorsLayer};

/// Maximum accepted `POST /collect` body size (256 KiB). Behavioral payloads are
/// small interaction-metadata blobs; anything larger is rejected with `413`.
const MAX_BODY_BYTES: usize = 256 * 1024;

/// Server configuration
#[derive(Debug, Clone)]
pub struct ServerConfig {
    /// Port to bind to (0 for random)
    pub port: u16,
    /// State directory for data
    pub state_dir: PathBuf,
    /// Optional bearer token required on `POST /collect`.
    ///
    /// When `Some`, requests must carry a matching `Authorization: Bearer <token>`
    /// header or they are rejected with `401`. When `None`, the endpoint is open
    /// (intended for trusted loopback-only use).
    pub token: Option<String>,
}

impl ServerConfig {
    /// Create a new server configuration without authentication.
    pub fn new(port: u16, state_dir: PathBuf) -> Self {
        Self {
            port,
            state_dir,
            token: None,
        }
    }

    /// Require a bearer token on `POST /collect`.
    pub fn with_token(mut self, token: Option<String>) -> Self {
        self.token = token.filter(|t| !t.is_empty());
        self
    }
}

/// Shared server state
pub struct ServerState {
    /// State directory
    #[allow(dead_code)]
    state_dir: PathBuf,
    /// Expected bearer token, if authentication is enabled.
    token: Option<String>,
}

impl ServerState {
    /// Create new server state
    pub fn new(config: &ServerConfig) -> Self {
        Self {
            state_dir: config.state_dir.clone(),
            token: config.token.clone(),
        }
    }
}

/// Behavioral session data from Chrome extension
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BehavioralSession {
    /// The behavioral session payload
    pub session: serde_json::Value,
}

/// Response from collect endpoint
#[derive(Debug, Clone, Serialize)]
pub struct CollectResponse {
    /// Processing status.
    pub status: String,
    /// Human-readable description.
    pub message: String,
}

/// Health check response
#[derive(Serialize)]
pub struct HealthResponse {
    /// Service health status.
    pub status: String,
    /// Agent version string.
    pub version: String,
}

/// Error response
#[derive(Serialize)]
pub struct ErrorResponse {
    /// Human-readable error message.
    pub error: String,
    /// Machine-readable error code.
    pub code: String,
}

/// GET /health
async fn health() -> Json<HealthResponse> {
    Json(HealthResponse {
        status: "ok".to_string(),
        version: env!("CARGO_PKG_VERSION").to_string(),
    })
}

/// Verify the request's bearer token against the configured one.
///
/// Returns `Ok(())` when authentication is disabled (no token configured) or when
/// a matching `Authorization: Bearer <token>` header is present; otherwise `401`.
fn check_auth(
    state: &ServerState,
    headers: &HeaderMap,
) -> Result<(), (StatusCode, Json<ErrorResponse>)> {
    let Some(expected) = state.token.as_deref() else {
        return Ok(());
    };

    let provided = headers
        .get(AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "));

    match provided {
        Some(token) if constant_time_eq(token.as_bytes(), expected.as_bytes()) => Ok(()),
        _ => Err((
            StatusCode::UNAUTHORIZED,
            Json(ErrorResponse {
                error: "Missing or invalid bearer token".to_string(),
                code: "UNAUTHORIZED".to_string(),
            }),
        )),
    }
}

/// Compare two byte strings in constant time relative to their content.
///
/// Guards the bearer-token check against timing attacks that could otherwise
/// recover the token byte-by-byte from response-time differences. The length is
/// not secret (it short-circuits), but equal-length inputs are always fully
/// scanned regardless of where they first differ.
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// POST /collect
///
/// Accepts raw behavioral data. Validates the bearer token (when configured) and
/// the JSON structure, then acknowledges. The caller (synheart-core-rust) is
/// responsible for feature extraction.
async fn collect(
    State(state): State<Arc<ServerState>>,
    headers: HeaderMap,
    Json(data): Json<BehavioralSession>,
) -> Result<Json<CollectResponse>, (StatusCode, Json<ErrorResponse>)> {
    check_auth(&state, &headers)?;

    if !data.session.is_object() {
        return Err((
            StatusCode::BAD_REQUEST,
            Json(ErrorResponse {
                error: "Session payload must be a JSON object".to_string(),
                code: "INVALID_SESSION".to_string(),
            }),
        ));
    }

    Ok(Json(CollectResponse {
        status: "ok".to_string(),
        message: "Data accepted".to_string(),
    }))
}

/// Run the HTTP server
pub async fn run(
    config: ServerConfig,
) -> anyhow::Result<(SocketAddr, tokio::sync::oneshot::Sender<()>)> {
    let state = Arc::new(ServerState::new(&config));

    let app = Router::new()
        .route("/health", get(health))
        .route("/collect", post(collect))
        // Behavioral payloads are small metadata blobs; cap the body to reject
        // oversized requests early (defense against memory-exhaustion attempts).
        .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
        .layer(
            // Allow browser extensions and local pages only. Extension origins
            // are `chrome-extension://<extension-id>` (the id varies per
            // install, so a prefix match is required), and localhost origins
            // carry a port (`http://localhost:<port>`), so exact-match values
            // would never match a real request. The server binds to loopback
            // and supports bearer-token auth; CORS is not the auth boundary.
            CorsLayer::new()
                .allow_origin(AllowOrigin::predicate(|origin, _| {
                    let Ok(origin) = origin.to_str() else {
                        return false;
                    };
                    origin.starts_with("chrome-extension://")
                        || origin == "http://localhost"
                        || origin.starts_with("http://localhost:")
                        || origin == "http://127.0.0.1"
                        || origin.starts_with("http://127.0.0.1:")
                }))
                .allow_methods(Any)
                .allow_headers(Any),
        )
        .with_state(state);

    let addr = SocketAddr::from(([127, 0, 0, 1], config.port));
    let listener = TcpListener::bind(addr).await?;
    let actual_addr = listener.local_addr()?;

    tracing::info!("Sensor agent server listening on http://{}", actual_addr);

    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();

    tokio::spawn(async move {
        if let Err(e) = axum::serve(listener, app)
            .with_graceful_shutdown(async {
                let _ = shutdown_rx.await;
                tracing::info!("Server shutdown signal received");
            })
            .await
        {
            tracing::error!("Server error: {}", e);
        }
    });

    Ok((actual_addr, shutdown_tx))
}