From 8d5e4106d6019fd933005e80c7644e8e36fbb2a8 Mon Sep 17 00:00:00 2001 From: Christian Mueller Date: Thu, 19 Mar 2026 23:01:52 +0100 Subject: [PATCH] feat(api): implement complete REST API with auth, CRUD, InfluxDB proxy, and user management Co-Authored-By: Claude Sonnet 4.6 --- LORAWEB_NEU/backend/api/src/alarms.rs | 99 +++++++ LORAWEB_NEU/backend/api/src/auth.rs | 147 ++++++++++ LORAWEB_NEU/backend/api/src/db.rs | 40 +++ LORAWEB_NEU/backend/api/src/main.rs | 87 +++++- LORAWEB_NEU/backend/api/src/sensors.rs | 377 +++++++++++++++++++++++++ LORAWEB_NEU/backend/api/src/topics.rs | 127 +++++++++ LORAWEB_NEU/backend/api/src/users.rs | 265 +++++++++++++++++ 7 files changed, 1136 insertions(+), 6 deletions(-) create mode 100644 LORAWEB_NEU/backend/api/src/alarms.rs create mode 100644 LORAWEB_NEU/backend/api/src/auth.rs create mode 100644 LORAWEB_NEU/backend/api/src/db.rs create mode 100644 LORAWEB_NEU/backend/api/src/sensors.rs create mode 100644 LORAWEB_NEU/backend/api/src/topics.rs create mode 100644 LORAWEB_NEU/backend/api/src/users.rs diff --git a/LORAWEB_NEU/backend/api/src/alarms.rs b/LORAWEB_NEU/backend/api/src/alarms.rs new file mode 100644 index 0000000..99c5373 --- /dev/null +++ b/LORAWEB_NEU/backend/api/src/alarms.rs @@ -0,0 +1,99 @@ +use axum::{ + extract::State, + http::StatusCode, + routing::get, + Json, Router, +}; +use serde::Serialize; +use crate::AppState; + +#[derive(Debug, Serialize)] +struct AlarmSummary { + active: i64, + warning: i64, + alarm: i64, + unknown: i64, + alarms: Vec, +} + +#[derive(Debug, Serialize, sqlx::FromRow)] +struct AlarmEntry { + topic_id: String, + topic: String, + alarm_status: String, + last_seen: Option, + sensor_name: Option, +} + +#[derive(Debug, sqlx::FromRow)] +struct StatusCount { + alarm_status: String, + count: i64, +} + +type ApiError = (StatusCode, Json); + +fn internal_err(msg: &str) -> ApiError { + (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": msg }))) +} + +pub fn routes() -> Router { + Router::new().route("/alarms", get(get_alarms)) +} + +async fn get_alarms( + State(state): State, +) -> Result, ApiError> { + // Get counts per alarm status + let counts = sqlx::query_as::<_, StatusCount>( + "SELECT alarm_status, COUNT(*) as count FROM topics GROUP BY alarm_status" + ) + .fetch_all(&state.pool) + .await + .map_err(|_| internal_err("Database error"))?; + + let mut active: i64 = 0; + let mut warning: i64 = 0; + let mut alarm: i64 = 0; + let mut unknown: i64 = 0; + + for row in &counts { + match row.alarm_status.as_str() { + "ok" => active += row.count, + "warning" => { + active += row.count; + warning += row.count; + } + "alarm" => { + active += row.count; + alarm += row.count; + } + _ => unknown += row.count, + } + } + + // Get alarm entries (warning + alarm status) with sensor name via LEFT JOIN + let alarms = sqlx::query_as::<_, AlarmEntry>( + r#"SELECT + t.id as topic_id, + t.topic, + t.alarm_status, + t.last_seen, + s.name as sensor_name + FROM topics t + LEFT JOIN sensors s ON s.topic_id = t.id + WHERE t.alarm_status IN ('warning', 'alarm') + ORDER BY t.alarm_status DESC, t.last_seen ASC"# + ) + .fetch_all(&state.pool) + .await + .map_err(|_| internal_err("Database error"))?; + + Ok(Json(AlarmSummary { + active, + warning, + alarm, + unknown, + alarms, + })) +} diff --git a/LORAWEB_NEU/backend/api/src/auth.rs b/LORAWEB_NEU/backend/api/src/auth.rs new file mode 100644 index 0000000..b55ac81 --- /dev/null +++ b/LORAWEB_NEU/backend/api/src/auth.rs @@ -0,0 +1,147 @@ +use axum::{ + extract::State, + http::{Request, StatusCode}, + middleware::Next, + response::{IntoResponse, Response}, + routing::post, + Json, Router, +}; +use argon2::{Argon2, PasswordHash, PasswordVerifier}; +use jsonwebtoken::{decode, encode, DecodingKey, EncodingKey, Header, Validation}; +use serde::{Deserialize, Serialize}; +use chrono::{Utc, Duration}; +use crate::AppState; + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct Claims { + pub sub: i64, + pub role: String, + pub exp: i64, + #[serde(default)] + pub purpose: Option, +} + +#[derive(Deserialize)] +struct LoginRequest { + username: Option, + password: Option, + totp_token: Option, + totp_code: Option, +} + +#[derive(Serialize)] +struct LoginResponse { + #[serde(skip_serializing_if = "Option::is_none")] + token: Option, + #[serde(skip_serializing_if = "Option::is_none")] + totp_required: Option, + #[serde(skip_serializing_if = "Option::is_none")] + totp_token: Option, +} + +pub fn routes() -> Router { + Router::new().route("/auth/login", post(login)) +} + +async fn login( + State(state): State, + Json(req): Json, +) -> Result, (StatusCode, Json)> { + let err = |msg: &str| (StatusCode::UNAUTHORIZED, Json(serde_json::json!({ "error": msg }))); + + // TOTP step 2 + if let (Some(totp_token), Some(totp_code)) = (&req.totp_token, &req.totp_code) { + let claims = decode_jwt(&state.jwt_secret, totp_token).map_err(|_| err("Invalid TOTP token"))?; + if claims.purpose.as_deref() != Some("totp") { + return Err(err("Invalid token purpose")); + } + + let user: Option<(i64, String, Option)> = sqlx::query_as( + "SELECT id, role, totp_secret FROM users WHERE id = ?" + ).bind(claims.sub).fetch_optional(&state.pool).await.map_err(|_| err("Database error"))?; + + let (user_id, role, totp_secret) = user.ok_or_else(|| err("User not found"))?; + let secret = totp_secret.ok_or_else(|| err("TOTP not configured"))?; + + let secret_bytes = totp_rs::Secret::Encoded(secret) + .to_bytes().map_err(|_| err("TOTP secret decode failed"))?; + let totp = totp_rs::TOTP::new( + totp_rs::Algorithm::SHA1, 6, 1, 30, secret_bytes, + ).map_err(|_| err("TOTP init failed"))?; + + if !totp.check_current(totp_code).map_err(|_| err("TOTP check failed"))? { + return Err(err("Invalid TOTP code")); + } + + let token = create_jwt(&state.jwt_secret, user_id, &role, None)?; + return Ok(Json(LoginResponse { token: Some(token), totp_required: None, totp_token: None })); + } + + // Password step 1 + let username = req.username.as_deref().ok_or_else(|| err("Username required"))?; + let password = req.password.as_deref().ok_or_else(|| err("Password required"))?; + + let user: Option<(i64, String, String, i32)> = sqlx::query_as( + "SELECT id, password_hash, role, totp_enabled FROM users WHERE username = ?" + ).bind(username).fetch_optional(&state.pool).await.map_err(|_| err("Database error"))?; + + let (user_id, hash, role, totp_enabled) = user.ok_or_else(|| err("Invalid credentials"))?; + + let parsed_hash = PasswordHash::new(&hash).map_err(|_| err("Invalid credentials"))?; + Argon2::default().verify_password(password.as_bytes(), &parsed_hash).map_err(|_| err("Invalid credentials"))?; + + if totp_enabled != 0 { + let totp_token = create_jwt(&state.jwt_secret, user_id, &role, Some("totp"))?; + return Ok(Json(LoginResponse { token: None, totp_required: Some(true), totp_token: Some(totp_token) })); + } + + let token = create_jwt(&state.jwt_secret, user_id, &role, None)?; + Ok(Json(LoginResponse { token: Some(token), totp_required: None, totp_token: None })) +} + +fn create_jwt(secret: &str, user_id: i64, role: &str, purpose: Option<&str>) -> Result)> { + let exp = if purpose.is_some() { Utc::now() + Duration::seconds(60) } else { Utc::now() + Duration::hours(24) }; + let claims = Claims { sub: user_id, role: role.to_string(), exp: exp.timestamp(), purpose: purpose.map(String::from) }; + encode(&Header::default(), &claims, &EncodingKey::from_secret(secret.as_bytes())) + .map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "Failed to create token" })))) +} + +fn decode_jwt(secret: &str, token: &str) -> Result { + let data = decode::(token, &DecodingKey::from_secret(secret.as_bytes()), &Validation::default())?; + Ok(data.claims) +} + +pub async fn require_auth( + State(state): State, + mut req: Request, + next: Next, +) -> Response { + let auth_header = req.headers().get("authorization") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")); + + let token = match auth_header { + Some(t) => t, + None => return (StatusCode::UNAUTHORIZED, Json(serde_json::json!({ "error": "Missing token" }))).into_response(), + }; + + let claims = match decode_jwt(&state.jwt_secret, token) { + Ok(c) => c, + Err(_) => return (StatusCode::UNAUTHORIZED, Json(serde_json::json!({ "error": "Invalid token" }))).into_response(), + }; + + if claims.purpose.is_some() { + return (StatusCode::UNAUTHORIZED, Json(serde_json::json!({ "error": "Invalid token type" }))).into_response(); + } + + req.extensions_mut().insert(claims); + next.run(req).await +} + +pub async fn require_admin(req: Request, next: Next) -> Response { + let claims = req.extensions().get::(); + match claims { + Some(c) if c.role == "admin" => next.run(req).await, + _ => (StatusCode::FORBIDDEN, Json(serde_json::json!({ "error": "Admin required" }))).into_response(), + } +} diff --git a/LORAWEB_NEU/backend/api/src/db.rs b/LORAWEB_NEU/backend/api/src/db.rs new file mode 100644 index 0000000..6014edc --- /dev/null +++ b/LORAWEB_NEU/backend/api/src/db.rs @@ -0,0 +1,40 @@ +use sqlx::SqlitePool; +use argon2::{Argon2, PasswordHasher, password_hash::SaltString}; +use rand::rngs::OsRng; +use chrono::Utc; + +pub async fn init_db(pool: &SqlitePool) -> Result<(), Box> { + sqlx::query("PRAGMA journal_mode=WAL").execute(pool).await?; + sqlx::query("PRAGMA busy_timeout=5000").execute(pool).await?; + + let migration = include_str!("../../../database/migrations/001_initial.sql"); + for statement in migration.split(';') { + let stmt = statement.trim(); + if !stmt.is_empty() { + sqlx::query(stmt).execute(pool).await?; + } + } + + let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM users") + .fetch_one(pool).await?; + + if count.0 == 0 { + let password = std::env::var("ADMIN_PASSWORD").unwrap_or_else(|_| "admin123".into()); + let salt = SaltString::generate(&mut OsRng); + let hash = Argon2::default() + .hash_password(password.as_bytes(), &salt)? + .to_string(); + let now = Utc::now().to_rfc3339(); + + sqlx::query( + "INSERT INTO users (username, password_hash, role, totp_enabled, created_at) VALUES ('admin', ?, 'admin', 0, ?)" + ) + .bind(&hash) + .bind(&now) + .execute(pool) + .await?; + tracing::info!("Default admin user created"); + } + + Ok(()) +} diff --git a/LORAWEB_NEU/backend/api/src/main.rs b/LORAWEB_NEU/backend/api/src/main.rs index 65902d7..5a9db12 100644 --- a/LORAWEB_NEU/backend/api/src/main.rs +++ b/LORAWEB_NEU/backend/api/src/main.rs @@ -1,7 +1,82 @@ -#[tokio::main] -async fn main() { - tracing_subscriber::fmt() - .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) - .init(); - tracing::info!("loraweb-api starting..."); +mod db; +mod auth; +mod sensors; +mod topics; +mod alarms; +mod users; + +use axum::Router; +use sqlx::sqlite::SqlitePoolOptions; +use tower_http::cors::{CorsLayer, Any}; + +#[derive(Clone)] +pub struct AppState { + pub pool: sqlx::SqlitePool, + pub jwt_secret: String, + pub influx_url: String, + pub influx_token: String, + pub influx_org: String, + pub influx_bucket: String, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::from_default_env() + .add_directive("loraweb_api=info".parse()?) + ) + .init(); + + let db_path = std::env::var("DATABASE_PATH").unwrap_or_else(|_| "/data/loraweb.db".into()); + let port = std::env::var("PORT").unwrap_or_else(|_| "3001".into()); + + let pool = SqlitePoolOptions::new() + .max_connections(10) + .connect(&format!("sqlite:{}?mode=rwc", db_path)) + .await?; + db::init_db(&pool).await?; + + let state = AppState { + pool, + jwt_secret: std::env::var("JWT_SECRET").unwrap_or_else(|_| "insecure-dev-secret".into()), + influx_url: std::env::var("INFLUX_URL").unwrap_or_else(|_| "http://influxdb:8086".into()), + influx_token: std::env::var("INFLUX_TOKEN").unwrap_or_default(), + influx_org: std::env::var("INFLUX_ORG").unwrap_or_else(|_| "loraweb".into()), + influx_bucket: std::env::var("INFLUX_BUCKET").unwrap_or_else(|_| "sensors".into()), + }; + + let cors = CorsLayer::new().allow_origin(Any).allow_methods(Any).allow_headers(Any); + + let app = Router::new() + .nest("/api", api_routes(state)) + .layer(cors); + + let addr = format!("0.0.0.0:{}", port); + tracing::info!("LoRaWEB API listening on {}", addr); + + let listener = tokio::net::TcpListener::bind(&addr).await?; + axum::serve(listener, app).await?; + Ok(()) +} + +fn api_routes(state: AppState) -> Router { + let public = Router::new().merge(auth::routes()); + + let protected = Router::new() + .merge(sensors::routes()) + .merge(topics::routes()) + .merge(alarms::routes()) + .route_layer(axum::middleware::from_fn_with_state(state.clone(), auth::require_auth)); + + let admin = Router::new() + .merge(users::routes()) + .route_layer(axum::middleware::from_fn(auth::require_admin)) + .route_layer(axum::middleware::from_fn_with_state(state.clone(), auth::require_auth)); + + Router::new() + .merge(public) + .merge(protected) + .merge(admin) + .with_state(state) } diff --git a/LORAWEB_NEU/backend/api/src/sensors.rs b/LORAWEB_NEU/backend/api/src/sensors.rs new file mode 100644 index 0000000..039cec3 --- /dev/null +++ b/LORAWEB_NEU/backend/api/src/sensors.rs @@ -0,0 +1,377 @@ +use axum::{ + extract::{Path, Query, State}, + http::StatusCode, + routing::get, + Json, Router, +}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use chrono::Utc; +use uuid::Uuid; +use crate::AppState; + +#[derive(Debug, Serialize, sqlx::FromRow)] +struct Sensor { + id: String, + name: String, + description: String, + location: String, + sensor_type: String, + active: i32, + topic_id: Option, + display_config: String, + created_at: String, +} + +#[derive(Deserialize)] +struct CreateSensorRequest { + name: String, + description: Option, + location: Option, + sensor_type: Option, + active: Option, + topic_id: Option, + display_config: Option, +} + +#[derive(Deserialize)] +struct UpdateSensorRequest { + name: Option, + description: Option, + location: Option, + sensor_type: Option, + active: Option, + topic_id: Option, + display_config: Option, +} + +#[derive(Deserialize)] +struct DataQuery { + start: Option, + stop: Option, + window: Option, +} + +#[derive(Deserialize)] +struct DisplayConfigRequest { + config: Value, +} + +type ApiError = (StatusCode, Json); + +fn internal_err(msg: &str) -> ApiError { + (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": msg }))) +} + +fn not_found(msg: &str) -> ApiError { + (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": msg }))) +} + +fn bad_request(msg: &str) -> ApiError { + (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": msg }))) +} + +pub fn routes() -> Router { + Router::new() + .route("/sensors", get(list_sensors).post(create_sensor)) + .route("/sensors/{id}", get(get_sensor).put(update_sensor).delete(delete_sensor)) + .route("/sensors/{id}/data", get(get_sensor_data)) + .route("/sensors/{id}/display-config", get(get_display_config).put(save_display_config)) +} + +async fn list_sensors( + State(state): State, +) -> Result>, ApiError> { + let sensors = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors ORDER BY name") + .fetch_all(&state.pool) + .await + .map_err(|_| internal_err("Database error"))?; + Ok(Json(sensors)) +} + +async fn create_sensor( + State(state): State, + Json(req): Json, +) -> Result<(StatusCode, Json), ApiError> { + let id = Uuid::new_v4().to_string(); + let now = Utc::now().to_rfc3339(); + let description = req.description.unwrap_or_default(); + let location = req.location.unwrap_or_default(); + let sensor_type = req.sensor_type.unwrap_or_default(); + let active = req.active.unwrap_or(1); + let display_config = req.display_config + .map(|v| serde_json::to_string(&v).unwrap_or_else(|_| "{}".into())) + .unwrap_or_else(|| "{}".into()); + + sqlx::query( + "INSERT INTO sensors (id, name, description, location, sensor_type, active, topic_id, display_config, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)" + ) + .bind(&id) + .bind(&req.name) + .bind(&description) + .bind(&location) + .bind(&sensor_type) + .bind(active) + .bind(&req.topic_id) + .bind(&display_config) + .bind(&now) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to create sensor"))?; + + let sensor = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?") + .bind(&id) + .fetch_one(&state.pool) + .await + .map_err(|_| internal_err("Failed to fetch created sensor"))?; + + Ok((StatusCode::CREATED, Json(sensor))) +} + +async fn get_sensor( + State(state): State, + Path(id): Path, +) -> Result, ApiError> { + let sensor = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?") + .bind(&id) + .fetch_optional(&state.pool) + .await + .map_err(|_| internal_err("Database error"))? + .ok_or_else(|| not_found("Sensor not found"))?; + Ok(Json(sensor)) +} + +async fn update_sensor( + State(state): State, + Path(id): Path, + Json(req): Json, +) -> Result, ApiError> { + let existing = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?") + .bind(&id) + .fetch_optional(&state.pool) + .await + .map_err(|_| internal_err("Database error"))? + .ok_or_else(|| not_found("Sensor not found"))?; + + let name = req.name.unwrap_or(existing.name); + let description = req.description.unwrap_or(existing.description); + let location = req.location.unwrap_or(existing.location); + let sensor_type = req.sensor_type.unwrap_or(existing.sensor_type); + let active = req.active.unwrap_or(existing.active); + let topic_id = if req.topic_id.is_some() { req.topic_id } else { existing.topic_id }; + let display_config = req.display_config + .map(|v| serde_json::to_string(&v).unwrap_or_else(|_| "{}".into())) + .unwrap_or(existing.display_config); + + sqlx::query( + "UPDATE sensors SET name=?, description=?, location=?, sensor_type=?, active=?, topic_id=?, display_config=? WHERE id=?" + ) + .bind(&name) + .bind(&description) + .bind(&location) + .bind(&sensor_type) + .bind(active) + .bind(&topic_id) + .bind(&display_config) + .bind(&id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to update sensor"))?; + + let sensor = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?") + .bind(&id) + .fetch_one(&state.pool) + .await + .map_err(|_| internal_err("Failed to fetch updated sensor"))?; + + Ok(Json(sensor)) +} + +async fn delete_sensor( + State(state): State, + Path(id): Path, +) -> Result { + let sensor = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?") + .bind(&id) + .fetch_optional(&state.pool) + .await + .map_err(|_| internal_err("Database error"))? + .ok_or_else(|| not_found("Sensor not found"))?; + + let mut tx = state.pool.begin().await.map_err(|_| internal_err("Transaction error"))?; + + if let Some(topic_id) = &sensor.topic_id { + sqlx::query("UPDATE topics SET approved=0 WHERE id=?") + .bind(topic_id) + .execute(&mut *tx) + .await + .map_err(|_| internal_err("Failed to reset topic"))?; + } + + sqlx::query("DELETE FROM sensors WHERE id=?") + .bind(&id) + .execute(&mut *tx) + .await + .map_err(|_| internal_err("Failed to delete sensor"))?; + + tx.commit().await.map_err(|_| internal_err("Transaction commit failed"))?; + + Ok(StatusCode::NO_CONTENT) +} + +async fn get_sensor_data( + State(state): State, + Path(id): Path, + Query(params): Query, +) -> Result>, ApiError> { + let row: Option<(Option, Option)> = sqlx::query_as( + "SELECT s.topic_id, t.topic FROM sensors s LEFT JOIN topics t ON s.topic_id = t.id WHERE s.id = ?" + ) + .bind(&id) + .fetch_optional(&state.pool) + .await + .map_err(|_| internal_err("Database error"))?; + + let (topic_id, topic_name) = row.ok_or_else(|| not_found("Sensor not found"))?; + + let _topic_id = topic_id.ok_or_else(|| bad_request("Sensor has no linked topic"))?; + let dev_eui = topic_name.ok_or_else(|| bad_request("Topic has no name"))?; + + let start = params.start.as_deref().unwrap_or("-1h"); + let stop = params.stop.as_deref().unwrap_or("now()"); + let window = params.window.as_deref().unwrap_or("5m"); + + let flux_query = format!( + r#"from(bucket: "{bucket}") + |> range(start: {start}, stop: {stop}) + |> filter(fn: (r) => r["dev_eui"] == "{dev_eui}" or r["topic"] == "{dev_eui}") + |> aggregateWindow(every: {window}, fn: mean, createEmpty: false) + |> yield(name: "mean")"#, + bucket = state.influx_bucket, + start = start, + stop = stop, + dev_eui = dev_eui, + window = window, + ); + + let client = reqwest::Client::new(); + let response = client + .post(format!("{}/api/v2/query", state.influx_url)) + .header("Authorization", format!("Token {}", state.influx_token)) + .header("Content-Type", "application/vnd.flux") + .query(&[("org", &state.influx_org)]) + .body(flux_query) + .send() + .await + .map_err(|_| internal_err("InfluxDB request failed"))?; + + if !response.status().is_success() { + let status = response.status().as_u16(); + let body = response.text().await.unwrap_or_default(); + tracing::warn!("InfluxDB error {}: {}", status, body); + return Err(( + StatusCode::BAD_GATEWAY, + Json(serde_json::json!({ "error": "InfluxDB query failed", "detail": body })), + )); + } + + let csv_text = response.text().await.map_err(|_| internal_err("Failed to read InfluxDB response"))?; + let data = parse_influx_csv(&csv_text); + + Ok(Json(data)) +} + +fn parse_influx_csv(csv: &str) -> Vec { + let mut results = Vec::new(); + let mut headers: Vec = Vec::new(); + + for line in csv.lines() { + if line.starts_with('#') || line.trim().is_empty() { + continue; + } + + let cols: Vec<&str> = line.split(',').collect(); + + if headers.is_empty() { + headers = cols.iter().map(|s| s.trim().to_string()).collect(); + continue; + } + + let mut obj = serde_json::Map::new(); + for (i, header) in headers.iter().enumerate() { + if header.is_empty() { + continue; + } + if header == "result" || header == "table" { + continue; + } + // Keep _time, _value, _field, _measurement; skip other _-prefixed + if header.starts_with('_') + && header != "_time" + && header != "_value" + && header != "_field" + && header != "_measurement" + { + continue; + } + + let value = cols.get(i).map(|s| s.trim()).unwrap_or(""); + + if let Ok(n) = value.parse::() { + obj.insert(header.clone(), serde_json::json!(n)); + } else if value == "true" { + obj.insert(header.clone(), serde_json::json!(true)); + } else if value == "false" { + obj.insert(header.clone(), serde_json::json!(false)); + } else { + obj.insert(header.clone(), serde_json::json!(value)); + } + } + + if !obj.is_empty() { + results.push(serde_json::Value::Object(obj)); + } + } + + results +} + +async fn get_display_config( + State(state): State, + Path(id): Path, +) -> Result, ApiError> { + let row: Option<(String,)> = sqlx::query_as("SELECT display_config FROM sensors WHERE id = ?") + .bind(&id) + .fetch_optional(&state.pool) + .await + .map_err(|_| internal_err("Database error"))?; + + let (config_str,) = row.ok_or_else(|| not_found("Sensor not found"))?; + let config: Value = serde_json::from_str(&config_str).unwrap_or(serde_json::json!({})); + + Ok(Json(config)) +} + +async fn save_display_config( + State(state): State, + Path(id): Path, + Json(req): Json, +) -> Result, ApiError> { + let config_str = serde_json::to_string(&req.config) + .map_err(|_| bad_request("Invalid config JSON"))?; + + let rows_affected = sqlx::query("UPDATE sensors SET display_config=? WHERE id=?") + .bind(&config_str) + .bind(&id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to save display config"))? + .rows_affected(); + + if rows_affected == 0 { + return Err(not_found("Sensor not found")); + } + + Ok(Json(req.config)) +} diff --git a/LORAWEB_NEU/backend/api/src/topics.rs b/LORAWEB_NEU/backend/api/src/topics.rs new file mode 100644 index 0000000..f8910a9 --- /dev/null +++ b/LORAWEB_NEU/backend/api/src/topics.rs @@ -0,0 +1,127 @@ +use axum::{ + extract::{Path, State}, + http::StatusCode, + routing::{get, put}, + Json, Router, +}; +use serde::{Deserialize, Serialize}; +use crate::AppState; + +#[derive(Debug, Serialize, sqlx::FromRow)] +struct Topic { + id: String, + topic: String, + approved: i32, + last_seen: Option, + alarm_status: String, + alarm_threshold_hours: f64, + created_at: String, +} + +#[derive(Deserialize)] +struct UpdateTopicRequest { + approved: Option, + sensor_id: Option, +} + +#[derive(Deserialize)] +struct ThresholdRequest { + threshold_minutes: f64, +} + +type ApiError = (StatusCode, Json); + +fn internal_err(msg: &str) -> ApiError { + (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": msg }))) +} + +fn not_found(msg: &str) -> ApiError { + (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": msg }))) +} + +pub fn routes() -> Router { + Router::new() + .route("/topics", get(list_topics)) + .route("/topics/{id}", put(update_topic)) + .route("/topics/{id}/threshold", put(set_threshold)) +} + +async fn list_topics( + State(state): State, +) -> Result>, ApiError> { + let topics = sqlx::query_as::<_, Topic>("SELECT * FROM topics ORDER BY created_at DESC") + .fetch_all(&state.pool) + .await + .map_err(|_| internal_err("Database error"))?; + Ok(Json(topics)) +} + +async fn update_topic( + State(state): State, + Path(id): Path, + Json(req): Json, +) -> Result, ApiError> { + // Verify topic exists + let existing = sqlx::query_as::<_, Topic>("SELECT * FROM topics WHERE id = ?") + .bind(&id) + .fetch_optional(&state.pool) + .await + .map_err(|_| internal_err("Database error"))? + .ok_or_else(|| not_found("Topic not found"))?; + + let approved = if let Some(sensor_id) = &req.sensor_id { + // Link sensor to this topic and auto-approve + sqlx::query("UPDATE sensors SET topic_id=? WHERE id=?") + .bind(&id) + .bind(sensor_id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to link sensor"))?; + 1 // auto-approve when sensor is linked + } else { + req.approved.map(|a| if a { 1 } else { 0 }).unwrap_or(existing.approved) + }; + + sqlx::query("UPDATE topics SET approved=? WHERE id=?") + .bind(approved) + .bind(&id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to update topic"))?; + + let topic = sqlx::query_as::<_, Topic>("SELECT * FROM topics WHERE id = ?") + .bind(&id) + .fetch_one(&state.pool) + .await + .map_err(|_| internal_err("Failed to fetch updated topic"))?; + + Ok(Json(topic)) +} + +async fn set_threshold( + State(state): State, + Path(id): Path, + Json(req): Json, +) -> Result, ApiError> { + let threshold_hours = req.threshold_minutes / 60.0; + + let rows_affected = sqlx::query("UPDATE topics SET alarm_threshold_hours=? WHERE id=?") + .bind(threshold_hours) + .bind(&id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to update threshold"))? + .rows_affected(); + + if rows_affected == 0 { + return Err(not_found("Topic not found")); + } + + let topic = sqlx::query_as::<_, Topic>("SELECT * FROM topics WHERE id = ?") + .bind(&id) + .fetch_one(&state.pool) + .await + .map_err(|_| internal_err("Failed to fetch updated topic"))?; + + Ok(Json(topic)) +} diff --git a/LORAWEB_NEU/backend/api/src/users.rs b/LORAWEB_NEU/backend/api/src/users.rs new file mode 100644 index 0000000..d62e5cf --- /dev/null +++ b/LORAWEB_NEU/backend/api/src/users.rs @@ -0,0 +1,265 @@ +use axum::{ + extract::{Path, State}, + http::StatusCode, + routing::{delete, get, post, put}, + Json, Router, +}; +use argon2::{Argon2, PasswordHasher, password_hash::SaltString}; +use rand::rngs::OsRng; +use serde::{Deserialize, Serialize}; +use chrono::Utc; +use crate::AppState; + +#[derive(Debug, Serialize, sqlx::FromRow)] +struct User { + id: i64, + username: String, + role: String, + totp_enabled: i32, + created_at: String, +} + +#[derive(Deserialize)] +struct CreateUserRequest { + username: String, + password: String, + role: Option, +} + +#[derive(Deserialize)] +struct UpdateUserRequest { + username: Option, + password: Option, + role: Option, +} + +#[derive(Serialize)] +struct TotpSetupResponse { + secret: String, + uri: String, +} + +type ApiError = (StatusCode, Json); + +fn internal_err(msg: &str) -> ApiError { + (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": msg }))) +} + +fn not_found(msg: &str) -> ApiError { + (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": msg }))) +} + +fn bad_request(msg: &str) -> ApiError { + (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": msg }))) +} + +fn conflict(msg: &str) -> ApiError { + (StatusCode::CONFLICT, Json(serde_json::json!({ "error": msg }))) +} + +pub fn routes() -> Router { + Router::new() + .route("/users", get(list_users).post(create_user)) + .route("/users/{id}", put(update_user).delete(delete_user)) + .route("/users/{id}/totp", post(enable_totp).delete(disable_totp)) +} + +async fn list_users( + State(state): State, +) -> Result>, ApiError> { + let users = sqlx::query_as::<_, User>( + "SELECT id, username, role, totp_enabled, created_at FROM users ORDER BY created_at ASC" + ) + .fetch_all(&state.pool) + .await + .map_err(|_| internal_err("Database error"))?; + Ok(Json(users)) +} + +async fn create_user( + State(state): State, + Json(req): Json, +) -> Result<(StatusCode, Json), ApiError> { + if req.username.trim().is_empty() { + return Err(bad_request("Username is required")); + } + if req.password.is_empty() { + return Err(bad_request("Password is required")); + } + + let role = req.role.unwrap_or_else(|| "user".into()); + let salt = SaltString::generate(&mut OsRng); + let hash = Argon2::default() + .hash_password(req.password.as_bytes(), &salt) + .map_err(|_| internal_err("Failed to hash password"))? + .to_string(); + let now = Utc::now().to_rfc3339(); + + let result = sqlx::query( + "INSERT INTO users (username, password_hash, role, totp_enabled, created_at) VALUES (?, ?, ?, 0, ?)" + ) + .bind(&req.username) + .bind(&hash) + .bind(&role) + .bind(&now) + .execute(&state.pool) + .await; + + let insert_result = match result { + Ok(r) => r, + Err(e) => { + let msg = e.to_string(); + if msg.contains("UNIQUE") || msg.contains("unique") { + return Err(conflict("Username already exists")); + } + return Err(internal_err("Failed to create user")); + } + }; + + let user_id = insert_result.last_insert_rowid(); + let user = sqlx::query_as::<_, User>( + "SELECT id, username, role, totp_enabled, created_at FROM users WHERE id = ?" + ) + .bind(user_id) + .fetch_one(&state.pool) + .await + .map_err(|_| internal_err("Failed to fetch created user"))?; + + Ok((StatusCode::CREATED, Json(user))) +} + +async fn update_user( + State(state): State, + Path(id): Path, + Json(req): Json, +) -> Result, ApiError> { + let existing = sqlx::query_as::<_, User>( + "SELECT id, username, role, totp_enabled, created_at FROM users WHERE id = ?" + ) + .bind(id) + .fetch_optional(&state.pool) + .await + .map_err(|_| internal_err("Database error"))? + .ok_or_else(|| not_found("User not found"))?; + + let username = req.username.unwrap_or(existing.username); + let role = req.role.unwrap_or(existing.role); + + if let Some(password) = req.password { + if password.is_empty() { + return Err(bad_request("Password cannot be empty")); + } + let salt = SaltString::generate(&mut OsRng); + let hash = Argon2::default() + .hash_password(password.as_bytes(), &salt) + .map_err(|_| internal_err("Failed to hash password"))? + .to_string(); + + sqlx::query("UPDATE users SET username=?, role=?, password_hash=? WHERE id=?") + .bind(&username) + .bind(&role) + .bind(&hash) + .bind(id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to update user"))?; + } else { + sqlx::query("UPDATE users SET username=?, role=? WHERE id=?") + .bind(&username) + .bind(&role) + .bind(id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to update user"))?; + } + + let user = sqlx::query_as::<_, User>( + "SELECT id, username, role, totp_enabled, created_at FROM users WHERE id = ?" + ) + .bind(id) + .fetch_one(&state.pool) + .await + .map_err(|_| internal_err("Failed to fetch updated user"))?; + + Ok(Json(user)) +} + +async fn delete_user( + State(state): State, + Path(id): Path, +) -> Result { + let rows_affected = sqlx::query("DELETE FROM users WHERE id=?") + .bind(id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Database error"))? + .rows_affected(); + + if rows_affected == 0 { + return Err(not_found("User not found")); + } + + Ok(StatusCode::NO_CONTENT) +} + +async fn enable_totp( + State(state): State, + Path(id): Path, +) -> Result, ApiError> { + // Verify user exists and get username for the TOTP URI + let row: Option<(String,)> = sqlx::query_as("SELECT username FROM users WHERE id = ?") + .bind(id) + .fetch_optional(&state.pool) + .await + .map_err(|_| internal_err("Database error"))?; + + let (username,) = row.ok_or_else(|| not_found("User not found"))?; + + let secret = totp_rs::Secret::generate_secret(); + let secret_encoded = secret.to_encoded().to_string(); + + let secret_bytes = totp_rs::Secret::Encoded(secret_encoded.clone()) + .to_bytes() + .map_err(|_| internal_err("Failed to decode TOTP secret"))?; + + let totp = totp_rs::TOTP::new( + totp_rs::Algorithm::SHA1, + 6, + 1, + 30, + secret_bytes, + ) + .map_err(|_| internal_err("Failed to create TOTP"))?; + + let uri = totp.get_url(&username, "LoRaWEB"); + + sqlx::query("UPDATE users SET totp_secret=?, totp_enabled=1 WHERE id=?") + .bind(&secret_encoded) + .bind(id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Failed to save TOTP secret"))?; + + Ok(Json(TotpSetupResponse { + secret: secret_encoded, + uri, + })) +} + +async fn disable_totp( + State(state): State, + Path(id): Path, +) -> Result { + let rows_affected = sqlx::query("UPDATE users SET totp_secret=NULL, totp_enabled=0 WHERE id=?") + .bind(id) + .execute(&state.pool) + .await + .map_err(|_| internal_err("Database error"))? + .rows_affected(); + + if rows_affected == 0 { + return Err(not_found("User not found")); + } + + Ok(StatusCode::NO_CONTENT) +}