ParrotXray/Mantis
1
0
Fork 0
mirror of https://github.com/ParrotXray/Mantis.git synced 2026-08-24 18:50:28 +09:00
Code Issues Packages Projects Releases Wiki Activity

feat: add SQLCipher account DB, Argon2 password hashing, and JWT auth

Browse Source
This commit is contained in:
ParrotXray 2026-05-23 03:13:21 +00:00
parent 511e1259f2
commit 48919b9c58
33 changed files with 1419 additions and 1211 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1101
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

7
config.toml
View File

@ -40,6 +40,13 @@ ml_cpu = 7
ae_threshold_method = "94"
# Auth system. Remove this entire section to disable auth.
[Config.auth]
jwt_secret = "change-me-jwt-secret-must-be-32-bytes-min"
db_key = "change-me-db-key-must-be-32-bytes-min-x"
token_ttl_secs = 86400
default_admin_password = "admin"
# Suricata rule engine. Remove this entire section to disable.
[Config.suricata]
home_net = "140.130.34.0/24"

2
mantis-frontend

@ -1 +1 @@
Subproject commit f1fa091f4c1abc90049b32c36f2d339400090713
Subproject commit 66ae9de967702ef02ab47baa9597ab8f16e3ab91

12
mantis/Cargo.toml
View File

@ -7,10 +7,9 @@ edition = "2024"
common = { path = "../common", features = ["user"] }
macros = { path = "../macros" }
actix = "0.13.5"
actix-cors = "0.7.1"
actix-web = "4.11.0"
actix-ws = "0.4.0"
axum = { version = "0.8", features = ["ws", "macros"] }
tower = { version = "0.5", features = ["util"] }
tower-http = { version = "0.6", features = ["cors"] }
aya = { workspace = true }
aya-log = { workspace = true }
network-types = { workspace = true }
@ -25,7 +24,6 @@ serde_json = "1.0.143"
sysinfo = "0.38.2"
thiserror = "2.0.3"
tokio = { version = "1.40.0", features = ["full", "macros"] }
tokio-tungstenite = "0.28.0"
toml = "1.0.3"
tracing = "0.1.41"
tracing-appender = "0.2.3"
@ -37,6 +35,10 @@ lru = "0.16.2"
futures = "0.3.31"
tract-onnx = "0.22.1"
chrono = "0.4"
rusqlite = { version = "0.31", features = ["bundled-sqlcipher-vendored-openssl"] }
argon2 = "0.5"
jsonwebtoken = "9"
uuid = { version = "1", features = ["v4"] }
ort-tract = { version = "0.3.0+0.22", optional = true }
ort = { version = "=2.0.0-rc.12", default-features = false, features = ["std", "ndarray"] }

2
mantis/build.rs
View File

@ -13,6 +13,7 @@ fn main() {
let manifest_dir = PathBuf::from(env::var("CARGO_MANIFEST_DIR").unwrap());
let artifact_dir = manifest_dir.join("static").join("artifacts");
let rule_dir = manifest_dir.join("static").join("rules");
let db_dir = manifest_dir.join("static").join("db");
let csv_dir = manifest_dir.parent().unwrap().join("records");
let lib_dir = manifest_dir.parent().unwrap().join("lib");
@ -33,6 +34,7 @@ fn main() {
println!("cargo:rustc-env=INGRESS_PATH={}", ingress_edpf_dir.display());
println!("cargo:rustc-env=EGRESS_PATH={}", egress_edpf_dir.display());
println!("cargo:rustc-env=RULE_PATH={}", rule_dir.display());
println!("cargo:rustc-env=DB_PATH={}", db_dir.display());
println!("cargo:rustc-env=RULE_EVE_PATH={}", suricata_eve_socket.display());
for item in &[

22
mantis/src/core/app_state.rs Normal file
View File

@ -0,0 +1,22 @@
use std::sync::Arc;
use crate::core::ebpf::access_control::AccessControl;
use crate::core::ebpf::service::Service;
use crate::core::ebpf::statistics::Statistics;
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::infrastructure::app_db::AppDB;
use crate::core::infrastructure::detection_alert::DetectionAlert;
use crate::core::infrastructure::health::SystemHealth;
use crate::detection::ml::config_loader::InferenceConfig;
#[derive(Clone)]
pub struct AppState {
pub app_config: Arc<AppConfig>,
pub inference_config: Arc<InferenceConfig>,
pub access_control: Arc<AccessControl>,
pub service: Arc<Service>,
pub statistics: Arc<Statistics>,
pub health: Arc<SystemHealth>,
pub detection_alert: Arc<DetectionAlert>,
pub app_db: Option<Arc<AppDB>>,
}

115
mantis/src/core/infrastructure/app_db.rs Normal file
View File

@ -0,0 +1,115 @@
use std::path::Path;
use std::sync::Mutex;
use argon2::password_hash::{rand_core::OsRng, PasswordHash, PasswordHasher, PasswordVerifier, SaltString};
use argon2::Argon2;
use macros::log;
use rusqlite::{params, Connection};
use crate::model::error::auth::AuthError;
use crate::model::error::Error;
use crate::model::log::auth::AuthLog;
pub struct Account {
pub id: String,
pub username: String,
pub password_hash: String,
pub role: String,
}
pub struct AppDB {
conn: Mutex<Connection>,
}
impl AppDB {
pub fn open(path: impl AsRef<Path>, key: &str) -> Result<Self, Error> {
let path = path.as_ref();
if let Some(parent) = Path::new(path).parent() {
std::fs::create_dir_all(parent)
.map_err(|e| AuthError::DBError { msg: e.to_string() })?;
}
let conn = Connection::open(path)
.map_err(|e| AuthError::DBError { msg: e.to_string() })?;
// Must be the first statement on the connection to unlock the encrypted DB.
conn.execute_batch(&format!("PRAGMA key = '{}';", key.replace('\'', "''")))
.map_err(|e| AuthError::DBError { msg: e.to_string() })?;
conn.execute_batch(
"PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS accounts (
id TEXT PRIMARY KEY,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
role TEXT NOT NULL DEFAULT 'viewer',
created_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS sessions (
token TEXT PRIMARY KEY,
account_id TEXT NOT NULL,
expires_at INTEGER NOT NULL,
FOREIGN KEY (account_id) REFERENCES accounts(id) ON DELETE CASCADE
);",
)
.map_err(|e| AuthError::DBError { msg: e.to_string() })?;
log!(AuthLog::DbInitialized { path: path.to_string_lossy().to_string() });
Ok(Self { conn: Mutex::new(conn) })
}
pub fn ensure_default_admin(&self, default_password: &str) -> Result<(), Error> {
let conn = self.conn.lock().unwrap();
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM accounts", [], |row| row.get(0))
.map_err(|e| AuthError::DBError { msg: e.to_string() })?;
if count == 0 {
let hash = hash_password(default_password)?;
let now = chrono::Utc::now().timestamp();
conn.execute(
"INSERT INTO accounts (id, username, password_hash, role, created_at) VALUES (?1, ?2, ?3, ?4, ?5)",
params!["admin", "admin", hash, "admin", now],
)
.map_err(|e| AuthError::DBError { msg: e.to_string() })?;
log!(AuthLog::DefaultAdminCreated);
}
Ok(())
}
pub fn find_account_by_username(&self, username: &str) -> Result<Option<Account>, Error> {
let conn = self.conn.lock().unwrap();
let result = conn.query_row(
"SELECT id, username, password_hash, role FROM accounts WHERE username = ?1",
params![username],
|row| {
Ok(Account {
id: row.get(0)?,
username: row.get(1)?,
password_hash: row.get(2)?,
role: row.get(3)?,
})
},
);
match result {
Ok(account) => Ok(Some(account)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(AuthError::DBError { msg: e.to_string() }.into()),
}
}
}
pub fn hash_password(password: &str) -> Result<String, Error> {
let salt = SaltString::generate(&mut OsRng);
Argon2::default()
.hash_password(password.as_bytes(), &salt)
.map(|h| h.to_string())
.map_err(|_| AuthError::HashError.into())
}
pub fn verify_password(password: &str, stored_hash: &str) -> bool {
PasswordHash::new(stored_hash)
.map(|h| Argon2::default().verify_password(password.as_bytes(), &h).is_ok())
.unwrap_or(false)
}

13
mantis/src/core/infrastructure/mod.rs
View File

@ -1,4 +1,5 @@
pub mod app_config;
pub mod app_db;
pub mod detection_alert;
pub mod health;
pub mod geoip;
@ -13,6 +14,7 @@ use macros::log;
use tokio::sync::oneshot;
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::infrastructure::app_db::AppDB;
use crate::core::infrastructure::detection_alert::DetectionAlert;
use crate::core::infrastructure::health::SystemHealth;
use crate::detection::fusion::{FusionEngine, FusionMode};
@ -34,6 +36,7 @@ pub struct AppServices {
pub ml_models: Arc<MLModels>,
pub ml_engine: Arc<Engine>,
pub suricata_engine: Option<Arc<SuricataEngine>>,
pub app_db: Option<Arc<AppDB>>,
shutdowns: SegQueue<oneshot::Sender<()>>,
}
@ -85,6 +88,15 @@ impl AppServices {
None
};
let app_db = if let Some(ref auth) = app_config.auth {
let db_path = PathBuf::from(env!("DB_PATH")).join("app.db");
let db = AppDB::open(db_path, &auth.db_key)?;
db.ensure_default_admin(&auth.default_admin_password)?;
Some(Arc::new(db))
} else {
None
};
Ok(Self {
health: Arc::new(health),
detection_alert,
@ -92,6 +104,7 @@ impl AppServices {
ml_models,
ml_engine,
suricata_engine,
app_db,
shutdowns: SegQueue::new(),
})
}

1
mantis/src/core/mod.rs
View File

@ -1,3 +1,4 @@
pub mod app_state;
pub mod ebpf;
pub mod infrastructure;
pub mod system;

71
mantis/src/core/system.rs
View File

@ -1,14 +1,16 @@
use std::sync::Arc;
use actix_web::web::route;
use actix_web::{web, App, HttpServer};
use aya::maps::{MapData, ProgramArray};
use aya::programs::{Xdp, XdpFlags};
use aya::Ebpf;
use aya_log::EbpfLogger;
use axum::Router;
use common::define::program_array::*;
use macros::log;
use tokio::net::TcpListener;
use tower_http::cors::CorsLayer;
use crate::core::app_state::AppState;
use crate::core::ebpf::EbpfServices;
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::infrastructure::AppServices;
@ -20,7 +22,8 @@ use crate::model::error::Error;
use crate::model::log::ml::MLLog;
use crate::model::log::system::SystemLog;
use crate::utils::logging::Logging;
use crate::web::api::{control, default, detection_alert, health, misc};
use crate::web::api::{auth, control, detection_alert, health, misc};
use crate::web::api::default::default_route;
pub struct System {
pub app_config: Arc<AppConfig>,
@ -139,40 +142,36 @@ impl System {
}
async fn run_http_server(&self) -> Result<(), Error> {
let app_config = self.app_config.clone();
let inference_config = self.inference_config.clone();
let access_control = self.ebpf_services.access_control.clone();
let service = self.ebpf_services.service.clone();
let statistics = self.ebpf_services.statistics.clone();
let health = self.app_services.health.clone();
let detection_alert = self.app_services.detection_alert.clone();
let state = AppState {
app_config: self.app_config.clone(),
inference_config: self.inference_config.clone(),
access_control: self.ebpf_services.access_control.clone(),
service: self.ebpf_services.service.clone(),
statistics: self.ebpf_services.statistics.clone(),
health: self.app_services.health.clone(),
detection_alert: self.app_services.detection_alert.clone(),
app_db: self.app_services.app_db.clone(),
};
let app = Router::new()
.nest("/ebpf", control::router())
.nest("/detection", detection_alert::router())
.nest("/health", health::router())
.nest("/misc", misc::router())
.nest("/auth", auth::router())
.fallback(default_route)
.layer(CorsLayer::permissive())
.with_state(state);
let port = self.app_config.http_server_bind_port;
HttpServer::new(move || {
let cors = actix_cors::Cors::default()
.allow_any_origin()
.allow_any_method()
.allow_any_header()
.max_age(3600);
App::new()
.wrap(cors)
.app_data(web::Data::from(app_config.clone()))
.app_data(web::Data::from(inference_config.clone()))
.app_data(web::Data::from(access_control.clone()))
.app_data(web::Data::from(service.clone()))
.app_data(web::Data::from(statistics.clone()))
.app_data(web::Data::from(health.clone()))
.app_data(web::Data::from(detection_alert.clone()))
.service(control::initialize())
.service(detection_alert::initialize())
.service(health::initialize())
.service(misc::initialize())
.default_service(route().to(default::default_route))
})
.bind(format!("0.0.0.0:{}", port))
.map_err(HttpError::BindPortError)?
.run()
.await
.map_err(HttpError::ServerPanic)?;
let listener = TcpListener::bind(format!("0.0.0.0:{}", port))
.await
.map_err(HttpError::BindPortError)?;
axum::serve(listener, app)
.await
.map_err(HttpError::ServerPanic)?;
Ok(())
}

2
mantis/src/main.rs
View File

@ -7,7 +7,7 @@ mod detection;
use crate::core::system::System;
use crate::model::error::Error;
#[actix_web::main]
#[tokio::main]
async fn main() -> Result<(), Error> {
let mut system = System::new().await?;
system.run().await?;

15
mantis/src/model/config.rs
View File

@ -61,8 +61,23 @@ pub struct Config {
/// Suricata rule engine config. If absent, the rule engine is disabled.
pub suricata: Option<SuricataConfig>,
/// Auth system config. If absent, auth is disabled and all routes are public.
pub auth: Option<AuthConfig>,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct AuthConfig {
pub jwt_secret: String,
pub db_key: String,
#[serde(default = "default_token_ttl")]
pub token_ttl_secs: u64,
#[serde(default = "default_admin_password")]
pub default_admin_password: String,
}
fn default_token_ttl() -> u64 { 86400 }
fn default_admin_password() -> String { "admin".to_string() }
fn default_fusion_mode() -> String {
"or".to_string()
}

21
mantis/src/model/error/auth.rs Normal file
View File

@ -0,0 +1,21 @@
use macros::traceable;
traceable! {
AuthError {
#[no_source]
#[error("Database error: {msg}")]
DBError { msg: String } => tracing::Level::ERROR,
#[no_source]
#[error("Invalid credentials")]
InvalidCredentials => tracing::Level::WARN,
#[no_source]
#[error("Token expired or invalid")]
InvalidToken => tracing::Level::WARN,
#[no_source]
#[error("Password hashing failed")]
HashError => tracing::Level::ERROR,
}
}

5
mantis/src/model/error/http.rs
View File

@ -8,7 +8,8 @@ traceable! {
#[error("Http Server panic")]
ServerPanic => tracing::Level::ERROR,
#[error("WebSocket error")]
WebSocketError => tracing::Level::ERROR,
#[no_source]
#[error("WebSocket error: {msg}")]
WebSocketError { msg: String } => tracing::Level::ERROR,
}
}

10
mantis/src/model/error/mod.rs
View File

@ -1,3 +1,4 @@
pub mod auth;
pub mod ebpf;
pub mod http;
pub mod io;
@ -8,6 +9,7 @@ pub mod suricata;
use serde::{Deserialize, Serialize};
use crate::model::error::auth::AuthError;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::http::HttpError;
use crate::model::error::io::IOError;
@ -18,6 +20,8 @@ use crate::model::error::system::SystemError;
#[derive(Clone, Debug, thiserror::Error, Serialize, Deserialize)]
pub enum Error {
#[error("{0}")]
Auth(AuthError),
#[error("{0}")]
Ebpf(EbpfError),
#[error("{0}")]
@ -34,6 +38,12 @@ pub enum Error {
System(SystemError),
}
impl From<AuthError> for Error {
fn from(error: AuthError) -> Self {
Self::Auth(error)
}
}
impl From<EbpfError> for Error {
fn from(error: EbpfError) -> Self {
Self::Ebpf(error)

18
mantis/src/model/log/auth.rs Normal file
View File

@ -0,0 +1,18 @@
use macros::loggable;
use tracing;
loggable! {
AuthLog {
#[error("Auth DB initialized at {path}")]
DbInitialized { path: String } => tracing::Level::INFO,
#[error("Default admin account created")]
DefaultAdminCreated => tracing::Level::INFO,
#[error("Login successful for user: {username}")]
LoginSuccess { username: String } => tracing::Level::INFO,
#[error("Login failed for user: {username}")]
LoginFailed { username: String } => tracing::Level::WARN,
}
}

1
mantis/src/model/log/mod.rs
View File

@ -1,3 +1,4 @@
pub mod auth;
pub mod ebpf;
pub mod http;
pub mod ml;

99
mantis/src/web/api/auth.rs Normal file
View File

@ -0,0 +1,99 @@
use axum::{Router, routing::{get, post}};
use axum::extract::State;
use axum::http::StatusCode;
use axum::response::IntoResponse;
use axum::Json;
use chrono::Utc;
use jsonwebtoken::{encode, EncodingKey, Header};
use macros::log;
use serde::{Deserialize, Serialize};
use crate::core::app_state::AppState;
use crate::core::infrastructure::app_db::verify_password;
use crate::model::log::auth::AuthLog;
use crate::web::middleware::auth::{AuthenticatedUser, Claims};
pub fn router() -> Router<AppState> {
Router::new()
.route("/login", post(login))
.route("/me", get(me))
.route("/logout", post(logout))
}
#[derive(Deserialize)]
struct LoginRequest {
username: String,
password: String,
}
#[derive(Serialize)]
struct LoginResponse {
token: String,
}
#[derive(Serialize)]
struct MeResponse {
id: String,
username: String,
role: String,
}
async fn login(
State(state): State<AppState>,
Json(body): Json<LoginRequest>,
) -> impl IntoResponse {
let auth_cfg = match state.app_config.auth.as_ref() {
Some(c) => c,
None => return (StatusCode::NOT_IMPLEMENTED, "Auth not configured").into_response(),
};
let db = match &state.app_db {
Some(db) => db,
None => return (StatusCode::INTERNAL_SERVER_ERROR, "Database not available").into_response(),
};
let account = match db.find_account_by_username(&body.username) {
Ok(Some(a)) => a,
Ok(None) => {
log!(AuthLog::LoginFailed { username: body.username.clone() });
return (StatusCode::UNAUTHORIZED, "Invalid credentials").into_response();
}
Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, "Database error").into_response(),
};
if !verify_password(&body.password, &account.password_hash) {
log!(AuthLog::LoginFailed { username: body.username.clone() });
return (StatusCode::UNAUTHORIZED, "Invalid credentials").into_response();
}
let exp = (Utc::now().timestamp() as usize) + (auth_cfg.token_ttl_secs as usize);
let claims = Claims {
sub: account.id.clone(),
username: account.username.clone(),
role: account.role.clone(),
exp,
};
match encode(
&Header::default(),
&claims,
&EncodingKey::from_secret(auth_cfg.jwt_secret.as_bytes()),
) {
Ok(token) => {
log!(AuthLog::LoginSuccess { username: account.username });
Json(LoginResponse { token }).into_response()
}
Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Token generation failed").into_response(),
}
}
async fn me(user: AuthenticatedUser) -> impl IntoResponse {
Json(MeResponse {
id: user.0.sub,
username: user.0.username,
role: user.0.role,
})
}
async fn logout(_user: AuthenticatedUser) -> impl IntoResponse {
StatusCode::OK
}

118
mantis/src/web/api/control/access_control.rs
View File

@ -1,93 +1,81 @@
use std::net::{SocketAddrV4, SocketAddrV6};
use actix_web::{delete, get, put, web, HttpResponse, Responder, Scope};
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use axum::routing::{delete, get, put};
use axum::{Json, Router};
use crate::core::ebpf::access_control::AccessControl;
use crate::core::app_state::AppState;
use crate::model::direction::FlowDirection;
use crate::model::list_type::ListType;
pub fn initialize() -> Scope {
web::scope("/access_control")
.service(get_ipv4_list)
.service(get_ipv6_list)
.service(add_ipv4_list)
.service(add_ipv6_list)
.service(remove_ipv4_list)
.service(remove_ipv6_list)
pub fn router() -> Router<AppState> {
Router::new()
.route(
"/ipv4/{direction}/{list_type}",
get(get_ipv4_list).put(add_ipv4_list).delete(remove_ipv4_list),
)
.route(
"/ipv6/{direction}/{list_type}",
get(get_ipv6_list).put(add_ipv6_list).delete(remove_ipv6_list),
)
}
#[get("/ipv4/{direction}/{list_type}")]
async fn get_ipv4_list(
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let (direction, list_type) = path.into_inner();
let list = access_control.get_ipv4_list(direction, list_type).await;
HttpResponse::Ok().json(list)
Path((direction, list_type)): Path<(FlowDirection, ListType)>,
State(state): State<AppState>,
) -> impl IntoResponse {
Json(state.access_control.get_ipv4_list(direction, list_type).await)
}
#[get("/ipv6/{direction}/{list_type}")]
async fn get_ipv6_list(
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let (direction, list_type) = path.into_inner();
let list = access_control.get_ipv6_list(direction, list_type).await;
HttpResponse::Ok().json(list)
Path((direction, list_type)): Path<(FlowDirection, ListType)>,
State(state): State<AppState>,
) -> impl IntoResponse {
Json(state.access_control.get_ipv6_list(direction, list_type).await)
}
#[put("/ipv4/{direction}/{list_type}")]
async fn add_ipv4_list(
address: web::Json<SocketAddrV4>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
match access_control.add_ipv4_list(direction, list_type, address).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
Path((direction, list_type)): Path<(FlowDirection, ListType)>,
State(state): State<AppState>,
Json(address): Json<SocketAddrV4>,
) -> impl IntoResponse {
match state.access_control.add_ipv4_list(direction, list_type, address).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[put("/ipv6/{direction}/{list_type}")]
async fn add_ipv6_list(
address: web::Json<SocketAddrV6>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
match access_control.add_ipv6_list(direction, list_type, address).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
Path((direction, list_type)): Path<(FlowDirection, ListType)>,
State(state): State<AppState>,
Json(address): Json<SocketAddrV6>,
) -> impl IntoResponse {
match state.access_control.add_ipv6_list(direction, list_type, address).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv4/{direction}/{list_type}")]
async fn remove_ipv4_list(
address: web::Json<SocketAddrV4>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
match access_control.remove_ipv4_list(direction, list_type, address).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
Path((direction, list_type)): Path<(FlowDirection, ListType)>,
State(state): State<AppState>,
Json(address): Json<SocketAddrV4>,
) -> impl IntoResponse {
match state.access_control.remove_ipv4_list(direction, list_type, address).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv6/{direction}/{list_type}")]
async fn remove_ipv6_list(
address: web::Json<SocketAddrV6>,
path: web::Path<(FlowDirection, ListType)>,
access_control: web::Data<AccessControl>,
) -> impl Responder {
let address = address.into_inner();
let (direction, list_type) = path.into_inner();
match access_control.remove_ipv6_list(direction, list_type, address).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
Path((direction, list_type)): Path<(FlowDirection, ListType)>,
State(state): State<AppState>,
Json(address): Json<SocketAddrV6>,
) -> impl IntoResponse {
match state.access_control.remove_ipv6_list(direction, list_type, address).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
}

14
mantis/src/web/api/control/mod.rs
View File

@ -2,11 +2,13 @@ pub mod access_control;
pub mod service;
pub mod statistics;
use actix_web::{web, Scope};
use axum::Router;
pub fn initialize() -> Scope {
web::scope("/ebpf")
.service(access_control::initialize())
.service(service::initialize())
.service(statistics::initialize())
use crate::core::app_state::AppState;
pub fn router() -> Router<AppState> {
Router::new()
.nest("/access_control", access_control::router())
.nest("/service", service::router())
.nest("/statistics", statistics::router())
}

286
mantis/src/web/api/control/service.rs
View File

@ -1,251 +1,191 @@
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use actix_web::{delete, get, post, put, web, HttpResponse, Responder, Scope};
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use axum::routing::{delete, get, post, put};
use axum::{Json, Router};
use common::model::http_method::HttpMethod;
use crate::core::ebpf::service::Service;
use crate::core::app_state::AppState;
pub fn initialize() -> Scope {
web::scope("/service")
.service(get_ipv4_http_service)
.service(get_ipv6_http_service)
.service(add_ipv4_http_service)
.service(add_ipv6_http_service)
.service(remove_ipv4_http_service)
.service(remove_ipv6_http_service)
.service(is_ssh_white_list_enable)
.service(enable_ssh_white_list)
.service(disable_ssh_white_list)
.service(get_ipv4_ssh_service)
.service(get_ipv6_ssh_service)
.service(add_ipv4_ssh_service)
.service(add_ipv6_ssh_service)
.service(remove_ipv4_ssh_service)
.service(remove_ipv6_ssh_service)
.service(get_ipv4_ssh_white_list)
.service(get_ipv6_ssh_white_list)
.service(add_ipv4_ssh_white_list)
.service(add_ipv6_ssh_white_list)
.service(remove_ipv4_ssh_white_list)
.service(remove_ipv6_ssh_white_list)
.service(get_ipv4_ssh_black_list)
.service(get_ipv6_ssh_black_list)
.service(add_ipv4_ssh_black_list)
.service(add_ipv6_ssh_black_list)
.service(remove_ipv4_ssh_black_list)
.service(remove_ipv6_ssh_black_list)
pub fn router() -> Router<AppState> {
Router::new()
.route("/ipv4/http_service", get(get_ipv4_http_service).put(add_ipv4_http_service).delete(remove_ipv4_http_service))
.route("/ipv6/http_service", get(get_ipv6_http_service).put(add_ipv6_http_service).delete(remove_ipv6_http_service))
.route("/ssh_white_list", get(is_ssh_white_list_enable))
.route("/ssh_white_list/enable", post(enable_ssh_white_list))
.route("/ssh_white_list/disable", post(disable_ssh_white_list))
.route("/ipv4/ssh_service", get(get_ipv4_ssh_service).put(add_ipv4_ssh_service).delete(remove_ipv4_ssh_service))
.route("/ipv6/ssh_service", get(get_ipv6_ssh_service).put(add_ipv6_ssh_service).delete(remove_ipv6_ssh_service))
.route("/ipv4/ssh_white_list", get(get_ipv4_ssh_white_list).put(add_ipv4_ssh_white_list).delete(remove_ipv4_ssh_white_list))
.route("/ipv6/ssh_white_list", get(get_ipv6_ssh_white_list).put(add_ipv6_ssh_white_list).delete(remove_ipv6_ssh_white_list))
.route("/ipv4/ssh_black_list", get(get_ipv4_ssh_black_list).put(add_ipv4_ssh_black_list).delete(remove_ipv4_ssh_black_list))
.route("/ipv6/ssh_black_list", get(get_ipv6_ssh_black_list).put(add_ipv6_ssh_black_list).delete(remove_ipv6_ssh_black_list))
}
#[get("/ipv4/http_service")]
async fn get_ipv4_http_service(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv4_http_service().await;
HttpResponse::Ok().json(web::Json(list))
async fn get_ipv4_http_service(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.get_ipv4_http_service().await)
}
#[get("/ipv6/http_service")]
async fn get_ipv6_http_service(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv6_http_service().await;
HttpResponse::Ok().json(web::Json(list))
async fn get_ipv6_http_service(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.get_ipv6_http_service().await)
}
#[put("/ipv4/http_service")]
async fn add_ipv4_http_service(
payload: web::Json<(SocketAddrV4, Vec<HttpMethod>)>,
service: web::Data<Service>,
) -> impl Responder {
let (addr, methods) = payload.into_inner();
match service.add_ipv4_http_service(addr, methods).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn add_ipv4_http_service(State(state): State<AppState>, Json(payload): Json<(SocketAddrV4, Vec<HttpMethod>)>) -> impl IntoResponse {
let (addr, methods) = payload;
match state.service.add_ipv4_http_service(addr, methods).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[put("/ipv6/http_service")]
async fn add_ipv6_http_service(
payload: web::Json<(SocketAddrV6, Vec<HttpMethod>)>,
service: web::Data<Service>,
) -> impl Responder {
let (addr, methods) = payload.into_inner();
match service.add_ipv6_http_service(addr, methods).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn add_ipv6_http_service(State(state): State<AppState>, Json(payload): Json<(SocketAddrV6, Vec<HttpMethod>)>) -> impl IntoResponse {
let (addr, methods) = payload;
match state.service.add_ipv6_http_service(addr, methods).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv4/http_service")]
async fn remove_ipv4_http_service(
payload: web::Json<(SocketAddrV4, Vec<HttpMethod>)>,
service: web::Data<Service>,
) -> impl Responder {
let (addr, methods) = payload.into_inner();
match service.remove_ipv4_http_service(addr, methods).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn remove_ipv4_http_service(State(state): State<AppState>, Json(payload): Json<(SocketAddrV4, Vec<HttpMethod>)>) -> impl IntoResponse {
let (addr, methods) = payload;
match state.service.remove_ipv4_http_service(addr, methods).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv6/http_service")]
async fn remove_ipv6_http_service(
payload: web::Json<(SocketAddrV6, Vec<HttpMethod>)>,
service: web::Data<Service>,
) -> impl Responder {
let (addr, methods) = payload.into_inner();
match service.remove_ipv6_http_service(addr, methods).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn remove_ipv6_http_service(State(state): State<AppState>, Json(payload): Json<(SocketAddrV6, Vec<HttpMethod>)>) -> impl IntoResponse {
let (addr, methods) = payload;
match state.service.remove_ipv6_http_service(addr, methods).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[get("/ssh_white_list")]
async fn is_ssh_white_list_enable(service: web::Data<Service>) -> impl Responder {
let enabled = service.is_ssh_white_list_enable().await;
HttpResponse::Ok().json(enabled)
async fn is_ssh_white_list_enable(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.is_ssh_white_list_enable().await)
}
#[post("/ssh_white_list/enable")]
async fn enable_ssh_white_list(service: web::Data<Service>) -> impl Responder {
match service.enable_ssh_white_list().await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn enable_ssh_white_list(State(state): State<AppState>) -> impl IntoResponse {
match state.service.enable_ssh_white_list().await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[post("/ssh_white_list/disable")]
async fn disable_ssh_white_list(service: web::Data<Service>) -> impl Responder {
match service.disable_ssh_white_list().await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn disable_ssh_white_list(State(state): State<AppState>) -> impl IntoResponse {
match state.service.disable_ssh_white_list().await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[get("/ipv4/ssh_service")]
async fn get_ipv4_ssh_service(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv4_ssh_service().await;
HttpResponse::Ok().json(web::Json(list))
async fn get_ipv4_ssh_service(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.get_ipv4_ssh_service().await)
}
#[get("/ipv6/ssh_service")]
async fn get_ipv6_ssh_service(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv6_ssh_service().await;
HttpResponse::Ok().json(web::Json(list))
async fn get_ipv6_ssh_service(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.get_ipv6_ssh_service().await)
}
#[put("/ipv4/ssh_service")]
async fn add_ipv4_ssh_service(ip_addr: web::Json<SocketAddrV4>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv4_ssh_service(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn add_ipv4_ssh_service(State(state): State<AppState>, Json(addr): Json<SocketAddrV4>) -> impl IntoResponse {
match state.service.add_ipv4_ssh_service(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[put("/ipv6/ssh_service")]
async fn add_ipv6_ssh_service(ip_addr: web::Json<SocketAddrV6>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv6_ssh_service(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn add_ipv6_ssh_service(State(state): State<AppState>, Json(addr): Json<SocketAddrV6>) -> impl IntoResponse {
match state.service.add_ipv6_ssh_service(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv4/ssh_service")]
async fn remove_ipv4_ssh_service(ip_addr: web::Json<SocketAddrV4>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv4_ssh_service(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn remove_ipv4_ssh_service(State(state): State<AppState>, Json(addr): Json<SocketAddrV4>) -> impl IntoResponse {
match state.service.remove_ipv4_ssh_service(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv6/ssh_service")]
async fn remove_ipv6_ssh_service(ip_addr: web::Json<SocketAddrV6>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv6_ssh_service(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn remove_ipv6_ssh_service(State(state): State<AppState>, Json(addr): Json<SocketAddrV6>) -> impl IntoResponse {
match state.service.remove_ipv6_ssh_service(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[get("/ipv4/ssh_white_list")]
async fn get_ipv4_ssh_white_list(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv4_ssh_white_list().await;
HttpResponse::Ok().json(web::Json(list))
async fn get_ipv4_ssh_white_list(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.get_ipv4_ssh_white_list().await)
}
#[get("/ipv6/ssh_white_list")]
async fn get_ipv6_ssh_white_list(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv6_ssh_white_list().await;
HttpResponse::Ok().json(web::Json(list))
async fn get_ipv6_ssh_white_list(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.get_ipv6_ssh_white_list().await)
}
#[put("/ipv4/ssh_white_list")]
async fn add_ipv4_ssh_white_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv4_ssh_white_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn add_ipv4_ssh_white_list(State(state): State<AppState>, Json(addr): Json<Ipv4Addr>) -> impl IntoResponse {
match state.service.add_ipv4_ssh_white_list(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[put("/ipv6/ssh_white_list")]
async fn add_ipv6_ssh_white_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv6_ssh_white_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn add_ipv6_ssh_white_list(State(state): State<AppState>, Json(addr): Json<Ipv6Addr>) -> impl IntoResponse {
match state.service.add_ipv6_ssh_white_list(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv4/ssh_white_list")]
async fn remove_ipv4_ssh_white_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv4_ssh_white_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn remove_ipv4_ssh_white_list(State(state): State<AppState>, Json(addr): Json<Ipv4Addr>) -> impl IntoResponse {
match state.service.remove_ipv4_ssh_white_list(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv6/ssh_white_list")]
async fn remove_ipv6_ssh_white_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv6_ssh_white_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn remove_ipv6_ssh_white_list(State(state): State<AppState>, Json(addr): Json<Ipv6Addr>) -> impl IntoResponse {
match state.service.remove_ipv6_ssh_white_list(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[get("/ipv4/ssh_black_list")]
async fn get_ipv4_ssh_black_list(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv4_ssh_black_list().await;
HttpResponse::Ok().json(web::Json(list))
async fn get_ipv4_ssh_black_list(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.get_ipv4_ssh_black_list().await)
}
#[get("/ipv6/ssh_black_list")]
async fn get_ipv6_ssh_black_list(service: web::Data<Service>) -> impl Responder {
let list = service.get_ipv6_ssh_black_list().await;
HttpResponse::Ok().json(web::Json(list))
async fn get_ipv6_ssh_black_list(State(state): State<AppState>) -> impl IntoResponse {
Json(state.service.get_ipv6_ssh_black_list().await)
}
#[put("/ipv4/ssh_black_list")]
async fn add_ipv4_ssh_black_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv4_ssh_black_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn add_ipv4_ssh_black_list(State(state): State<AppState>, Json(addr): Json<Ipv4Addr>) -> impl IntoResponse {
match state.service.add_ipv4_ssh_black_list(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[put("/ipv6/ssh_black_list")]
async fn add_ipv6_ssh_black_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<Service>) -> impl Responder {
match service.add_ipv6_ssh_black_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn add_ipv6_ssh_black_list(State(state): State<AppState>, Json(addr): Json<Ipv6Addr>) -> impl IntoResponse {
match state.service.add_ipv6_ssh_black_list(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv4/ssh_black_list")]
async fn remove_ipv4_ssh_black_list(ip_addr: web::Json<Ipv4Addr>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv4_ssh_black_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn remove_ipv4_ssh_black_list(State(state): State<AppState>, Json(addr): Json<Ipv4Addr>) -> impl IntoResponse {
match state.service.remove_ipv4_ssh_black_list(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
#[delete("/ipv6/ssh_black_list")]
async fn remove_ipv6_ssh_black_list(ip_addr: web::Json<Ipv6Addr>, service: web::Data<Service>) -> impl Responder {
match service.remove_ipv6_ssh_black_list(ip_addr.into_inner()).await {
Ok(_) => HttpResponse::Ok().finish(),
Err(e) => HttpResponse::InternalServerError().body(e.to_string()),
async fn remove_ipv6_ssh_black_list(State(state): State<AppState>, Json(addr): Json<Ipv6Addr>) -> impl IntoResponse {
match state.service.remove_ipv6_ssh_black_list(addr).await {
Ok(_) => StatusCode::OK.into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}

90
mantis/src/web/api/control/statistics.rs
View File

@ -1,76 +1,66 @@
use std::sync::Arc;
use actix_web::{get, web, HttpRequest, HttpResponse, Responder, Scope};
use axum::extract::{Path, State};
use axum::extract::ws::WebSocketUpgrade;
use axum::http::StatusCode;
use axum::response::IntoResponse;
use axum::routing::get;
use axum::{Json, Router};
use macros::log;
use crate::core::ebpf::statistics::Statistics;
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::app_state::AppState;
use crate::model::direction::{Direction, FlowDirection};
use crate::model::time_type::TimeType;
use crate::web::websocket::flow_websocket;
pub fn initialize() -> Scope {
web::scope("/statistics")
.service(get_ipv4_flow)
.service(get_ipv6_flow)
.service(websocket_ipv4)
.service(websocket_ipv6)
pub fn router() -> Router<AppState> {
Router::new()
.route("/get/ipv4/{direction}/{flow_direction}/{time_type}", get(get_ipv4_flow))
.route("/get/ipv6/{direction}/{flow_direction}/{time_type}", get(get_ipv6_flow))
.route("/websocket/ipv4/{direction}/{flow_direction}/{time_type}", get(websocket_ipv4))
.route("/websocket/ipv6/{direction}/{flow_direction}/{time_type}", get(websocket_ipv6))
}
#[get("/get/ipv4/{direction}/{flow_direction}/{time_type}")]
async fn get_ipv4_flow(
path: web::Path<(Direction, FlowDirection, TimeType)>,
statistics: web::Data<Arc<Statistics>>,
) -> impl Responder {
let (direction, flow_direction, time_type) = path.into_inner();
match statistics.get_ipv4_flow_data(direction, flow_direction, time_type).await {
Ok(flow_data) => HttpResponse::Ok().json(web::Json(flow_data)),
Path((direction, flow_direction, time_type)): Path<(Direction, FlowDirection, TimeType)>,
State(state): State<AppState>,
) -> impl IntoResponse {
match state.statistics.get_ipv4_flow_data(direction, flow_direction, time_type).await {
Ok(data) => Json(data).into_response(),
Err(e) => {
log!(e);
HttpResponse::InternalServerError().finish()
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}
}
#[get("/get/ipv6/{direction}/{flow_direction}/{time_type}")]
async fn get_ipv6_flow(
path: web::Path<(Direction, FlowDirection, TimeType)>,
statistics: web::Data<Arc<Statistics>>,
) -> impl Responder {
let (direction, flow_direction, time_type) = path.into_inner();
match statistics.get_ipv6_flow_data(direction, flow_direction, time_type).await {
Ok(flow_data) => HttpResponse::Ok().json(web::Json(flow_data)),
Path((direction, flow_direction, time_type)): Path<(Direction, FlowDirection, TimeType)>,
State(state): State<AppState>,
) -> impl IntoResponse {
match state.statistics.get_ipv6_flow_data(direction, flow_direction, time_type).await {
Ok(data) => Json(data).into_response(),
Err(e) => {
log!(e);
HttpResponse::InternalServerError().finish()
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}
}
#[get("/websocket/ipv4/{direction}/{flow_direction}/{time_type}")]
async fn websocket_ipv4(
req: HttpRequest,
stream: web::Payload,
path: web::Path<(Direction, FlowDirection, TimeType)>,
app_config: web::Data<AppConfig>,
statistics: web::Data<Statistics>,
) -> impl Responder {
match flow_websocket::websocket_ipv4_flow(req, stream, path, app_config, statistics).await {
Ok(response) => response,
Err(err) => HttpResponse::InternalServerError().body(format!("WebSocket error: {}", err)),
}
Path((direction, flow_direction, time_type)): Path<(Direction, FlowDirection, TimeType)>,
State(state): State<AppState>,
ws: WebSocketUpgrade,
) -> impl IntoResponse {
ws.on_upgrade(move |socket| {
flow_websocket::handle_ipv4_flow(socket, state, direction, flow_direction, time_type)
})
}
#[get("/websocket/ipv6/{direction}/{flow_direction}/{time_type}")]
async fn websocket_ipv6(
req: HttpRequest,
stream: web::Payload,
path: web::Path<(Direction, FlowDirection, TimeType)>,
app_config: web::Data<AppConfig>,
statistics: web::Data<Statistics>,
) -> impl Responder {
match flow_websocket::websocket_ipv6_flow(req, stream, path, app_config, statistics).await {
Ok(response) => response,
Err(err) => HttpResponse::InternalServerError().body(format!("WebSocket error: {}", err)),
}
}
Path((direction, flow_direction, time_type)): Path<(Direction, FlowDirection, TimeType)>,
State(state): State<AppState>,
ws: WebSocketUpgrade,
) -> impl IntoResponse {
ws.on_upgrade(move |socket| {
flow_websocket::handle_ipv6_flow(socket, state, direction, flow_direction, time_type)
})
}

48
mantis/src/web/api/default.rs
View File

@ -1,10 +1,10 @@
use actix_web::{HttpRequest, HttpResponse, Responder};
use mime_guess::from_path;
use axum::http::{StatusCode, Uri};
use axum::response::{IntoResponse, Response};
use crate::utils::static_files::StaticFiles;
pub async fn default_route(req: HttpRequest) -> impl Responder {
let request_path = req.path();
pub async fn default_route(uri: Uri) -> Response {
let request_path = uri.path();
let file_system_path = if request_path == "/" {
"web/index.html".to_string()
@ -13,30 +13,40 @@ pub async fn default_route(req: HttpRequest) -> impl Responder {
};
if let Some(content) = StaticFiles::get(&file_system_path) {
let mime_type = from_path(&file_system_path).first_or_octet_stream();
return HttpResponse::Ok()
.content_type(mime_type.as_ref())
.body(content.data.into_owned());
let mime_type = mime_guess::from_path(&file_system_path)
.first_or_octet_stream();
return (
[("content-type", mime_type.as_ref().to_string())],
content.data.into_owned(),
)
.into_response();
}
let html_path = format!("{}.html", file_system_path);
if let Some(content) = StaticFiles::get(&html_path) {
return HttpResponse::Ok()
.content_type("text/html")
.body(content.data.into_owned());
return (
[("content-type", "text/html")],
content.data.into_owned(),
)
.into_response();
}
let index_path = format!("{}/index.html", file_system_path);
if let Some(content) = StaticFiles::get(&index_path) {
return HttpResponse::Ok()
.content_type("text/html")
.body(content.data.into_owned());
return (
[("content-type", "text/html")],
content.data.into_owned(),
)
.into_response();
}
match StaticFiles::get("web/404.html") {
Some(page) => HttpResponse::NotFound()
.content_type("text/html")
.body(page.data.into_owned()),
None => HttpResponse::NotFound().body("404 Not Found"),
Some(page) => (
StatusCode::NOT_FOUND,
[("content-type", "text/html")],
page.data.into_owned(),
)
.into_response(),
None => (StatusCode::NOT_FOUND, "404 Not Found").into_response(),
}
}
}

27
mantis/src/web/api/detection_alert.rs
View File

@ -1,23 +1,18 @@
use actix_web::{get, web, HttpRequest, HttpResponse, Responder, Scope};
use axum::{Router, routing::get};
use axum::extract::State;
use axum::response::IntoResponse;
use crate::core::infrastructure::detection_alert::DetectionAlert;
use crate::core::app_state::AppState;
use crate::web::websocket::alert_websocket;
pub fn initialize() -> Scope {
web::scope("/detection")
.service(websocket_alert)
pub fn router() -> Router<AppState> {
Router::new().route("/websocket/alert", get(websocket_alert))
}
#[get("/websocket/alert")]
async fn websocket_alert(
req: HttpRequest,
stream: web::Payload,
da: web::Data<DetectionAlert>,
) -> impl Responder {
match alert_websocket::websocket_alert(req, stream, da).await {
Ok(response) => response,
Err(err) => {
HttpResponse::InternalServerError().body(format!("WebSocket error: {}", err))
}
}
ws: axum::extract::ws::WebSocketUpgrade,
State(state): State<AppState>,
) -> impl IntoResponse {
let rx = state.detection_alert.subscribe();
ws.on_upgrade(|socket| alert_websocket::handle_alert(socket, rx))
}

45
mantis/src/web/api/health.rs
View File

@ -1,35 +1,30 @@
use actix_web::{get, web, HttpRequest, HttpResponse, Responder, Scope};
use axum::{Router, routing::get};
use axum::extract::State;
use axum::response::IntoResponse;
use axum::Json;
use crate::core::infrastructure::health::SystemHealth;
use crate::core::app_state::AppState;
use crate::web::websocket::health_websocket;
pub fn initialize() -> Scope {
web::scope("/health")
.service(get_current_metrics)
.service(get_health_status)
.service(websocket_metrics)
pub fn router() -> Router<AppState> {
Router::new()
.route("/metrics", get(get_current_metrics))
.route("/status", get(get_health_status))
.route("/websocket/metrics", get(websocket_metrics))
}
#[get("/metrics")]
async fn get_current_metrics(health: web::Data<SystemHealth>) -> impl Responder {
let metrics = health.get_current_metrics().await;
HttpResponse::Ok().json(metrics)
async fn get_current_metrics(State(state): State<AppState>) -> impl IntoResponse {
Json(state.health.get_current_metrics().await)
}
#[get("/status")]
async fn get_health_status(health: web::Data<SystemHealth>) -> impl Responder {
let status = health.is_system_healthy().await;
HttpResponse::Ok().json(status)
async fn get_health_status(State(state): State<AppState>) -> impl IntoResponse {
Json(state.health.is_system_healthy().await)
}
#[get("/websocket/metrics")]
async fn websocket_metrics(
req: HttpRequest,
stream: web::Payload,
health: web::Data<SystemHealth>,
) -> impl Responder {
match health_websocket::websocket_system_health(req, stream, health).await {
Ok(response) => response,
Err(err) => HttpResponse::InternalServerError().body(format!("WebSocket error: {}", err)),
}
}
ws: axum::extract::ws::WebSocketUpgrade,
State(state): State<AppState>,
) -> impl IntoResponse {
let rx = state.health.subscribe_to_metrics();
ws.on_upgrade(|socket| health_websocket::handle_health(socket, rx))
}

16
mantis/src/web/api/misc.rs
View File

@ -1,14 +1,14 @@
use actix_web::{get, web, HttpResponse, Responder, Scope};
use axum::{Router, routing::get};
use axum::response::IntoResponse;
use axum::Json;
use crate::core::app_state::AppState;
use crate::utils::boot_time::boot_time;
pub fn initialize() -> Scope {
web::scope("/misc")
.service(get_boot_time)
pub fn router() -> Router<AppState> {
Router::new().route("/boot_time", get(get_boot_time))
}
#[get("/boot_time")]
async fn get_boot_time() -> impl Responder {
let boot_time = boot_time();
HttpResponse::Ok().json(boot_time)
async fn get_boot_time() -> impl IntoResponse {
Json(boot_time())
}

1
mantis/src/web/api/mod.rs
View File

@ -1,3 +1,4 @@
pub mod auth;
pub mod control;
pub mod default;
pub mod detection_alert;

46
mantis/src/web/middleware/auth.rs Normal file
View File

@ -0,0 +1,46 @@
use axum::extract::FromRequestParts;
use axum::http::{request::Parts, StatusCode};
use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation};
use serde::{Deserialize, Serialize};
use crate::core::app_state::AppState;
#[derive(Serialize, Deserialize, Clone)]
pub struct Claims {
pub sub: String,
pub username: String,
pub role: String,
pub exp: usize,
}
pub struct AuthenticatedUser(pub Claims);
impl FromRequestParts<AppState> for AuthenticatedUser {
type Rejection = (StatusCode, &'static str);
async fn from_request_parts(
parts: &mut Parts,
state: &AppState,
) -> Result<Self, Self::Rejection> {
let auth_cfg = state
.app_config
.auth
.as_ref()
.ok_or((StatusCode::NOT_IMPLEMENTED, "Auth not configured"))?;
let token = parts
.headers
.get("Authorization")
.and_then(|v| v.to_str().ok())
.and_then(|s| s.strip_prefix("Bearer "))
.ok_or((StatusCode::UNAUTHORIZED, "Unauthorized"))?;
decode::<Claims>(
token,
&DecodingKey::from_secret(auth_cfg.jwt_secret.as_bytes()),
&Validation::new(Algorithm::HS256),
)
.map(|d| AuthenticatedUser(d.claims))
.map_err(|_| (StatusCode::UNAUTHORIZED, "Unauthorized"))
}
}

1
mantis/src/web/middleware/mod.rs Normal file
View File

@ -0,0 +1 @@
pub mod auth;

1
mantis/src/web/mod.rs
View File

@ -1,2 +1,3 @@
pub mod api;
pub mod middleware;
pub mod websocket;

102
mantis/src/web/websocket/alert_websocket.rs
View File

@ -1,91 +1,47 @@
use actix_web::{web, HttpRequest, HttpResponse, Result};
use actix_ws::{handle, Message, MessageStream, Session};
use futures_util::StreamExt;
use axum::extract::ws::{Message, WebSocket};
use futures_util::{SinkExt, StreamExt};
use macros::log;
use tokio::sync::broadcast;
use crate::core::infrastructure::detection_alert::DetectionAlert;
use crate::model::ml_detection::UnifiedAlert;
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::log::http::HttpLog;
pub async fn websocket_alert(
req: HttpRequest,
body: web::Payload,
da: web::Data<DetectionAlert>,
) -> Result<HttpResponse> {
let (response, session, msg_stream) = handle(&req, body)?;
pub async fn handle_alert(socket: WebSocket, mut broadcast_rx: broadcast::Receiver<UnifiedAlert>) {
let (mut sender, mut receiver) = socket.split();
let broadcast_rx = da.subscribe();
actix_web::rt::spawn(async move {
handle_alert_connection(session, msg_stream, broadcast_rx).await;
});
Ok(response)
}
async fn handle_alert_connection(
mut session: Session,
mut msg_stream: MessageStream,
mut broadcast_rx: broadcast::Receiver<UnifiedAlert>,
) {
loop {
tokio::select! {
msg_result = msg_stream.next() => {
if !handle_client_message(&mut session, msg_result).await {
break;
}
},
broadcast_result = broadcast_rx.recv() => {
match broadcast_result {
Ok(alert) => {
if !send_alert(&mut session, &alert).await {
break;
}
msg = receiver.next() => {
match msg {
Some(Ok(Message::Ping(data))) => {
if sender.send(Message::Pong(data)).await.is_err() { break; }
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
log!(HttpLog::WebSocketLaged(skipped));
continue;
}
Err(broadcast::error::RecvError::Closed) => {
Some(Ok(Message::Close(_))) | None => break,
Some(Err(e)) => {
log!(HttpError::WebSocketError { msg: e.to_string() });
break;
}
_ => {}
}
},
}
}
let _ = session.close(None).await;
}
async fn handle_client_message(
session: &mut Session,
msg_result: Option<Result<Message, actix_ws::ProtocolError>>,
) -> bool {
match msg_result {
Some(Ok(Message::Text(_))) => true,
Some(Ok(Message::Ping(bytes))) => session.pong(&bytes).await.is_ok(),
Some(Ok(Message::Close(reason))) => {
let _ = (session.clone()).close(reason).await;
false
}
Some(Err(err)) => {
log!(HttpError::WebSocketError(err));
false
}
None => false,
_ => true,
}
}
async fn send_alert(session: &mut Session, alert: &UnifiedAlert) -> bool {
match serde_json::to_string(alert) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
log!(MiscError::SerializeError(err));
false
}
result = broadcast_rx.recv() => {
match result {
Ok(alert) => {
match serde_json::to_string(&alert) {
Ok(json) => {
if sender.send(Message::Text(json.into())).await.is_err() { break; }
}
Err(e) => { log!(MiscError::SerializeError(e)); }
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {
log!(HttpLog::WebSocketLaged { skipped: n });
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
}
}
}

214
mantis/src/web/websocket/flow_websocket.rs
View File

@ -1,192 +1,96 @@
use std::sync::Arc;
use actix_web::{web, HttpRequest, HttpResponse, Result};
use actix_ws::{handle, Message, MessageStream, Session};
use futures_util::StreamExt;
use axum::extract::ws::{Message, WebSocket};
use futures_util::{SinkExt, StreamExt};
use macros::log;
use tokio::time::{interval, Duration};
use crate::core::ebpf::statistics::Statistics;
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::app_state::AppState;
use crate::model::direction::{Direction, FlowDirection};
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::time_type::TimeType;
pub async fn websocket_ipv4_flow(
req: HttpRequest,
body: web::Payload,
path: web::Path<(Direction, FlowDirection, TimeType)>,
app_config: web::Data<AppConfig>,
statistics: web::Data<Statistics>,
) -> Result<HttpResponse> {
let app_config = app_config.into_inner();
let statistics = statistics.into_inner();
let (direction, flow_direction, time_type) = path.into_inner();
let (response, session, msg_stream) = handle(&req, body)?;
actix_web::rt::spawn(async move {
handle_ipv4_flow_connection(
app_config,
statistics,
session,
msg_stream,
direction,
flow_direction,
time_type,
)
.await;
});
Ok(response)
}
pub async fn websocket_ipv6_flow(
req: HttpRequest,
body: web::Payload,
path: web::Path<(Direction, FlowDirection, TimeType)>,
app_config: web::Data<AppConfig>,
statistics: web::Data<Statistics>,
) -> Result<HttpResponse> {
let app_config = app_config.into_inner();
let statistics = statistics.into_inner();
let (direction, flow_direction, time_type) = path.into_inner();
let (response, session, msg_stream) = handle(&req, body)?;
actix_web::rt::spawn(async move {
handle_ipv6_flow_connection(
app_config,
statistics,
session,
msg_stream,
direction,
flow_direction,
time_type,
)
.await;
});
Ok(response)
}
async fn handle_ipv4_flow_connection(
app_config: Arc<AppConfig>,
statistics: Arc<Statistics>,
mut session: Session,
mut msg_stream: MessageStream,
pub async fn handle_ipv4_flow(
socket: WebSocket,
state: AppState,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) {
let config = app_config.config.clone();
let refresh_interval = Duration::from_secs(config.refresh_interval);
let refresh_interval = Duration::from_secs(state.app_config.refresh_interval);
let (mut sender, mut receiver) = socket.split();
let mut data_interval = interval(refresh_interval);
loop {
tokio::select! {
msg_result = msg_stream.next() => {
if !handle_client_message(&mut session, msg_result).await {
break;
msg = receiver.next() => {
match msg {
Some(Ok(Message::Ping(data))) => {
if sender.send(Message::Pong(data)).await.is_err() { break; }
}
Some(Ok(Message::Close(_))) | None => break,
Some(Err(e)) => {
log!(HttpError::WebSocketError { msg: e.to_string() });
break;
}
_ => {}
}
},
}
_ = data_interval.tick() => {
if !send_ipv4_flow_data(&statistics, &mut session, direction, flow_direction, time_type).await {
break;
match state.statistics.get_ipv4_flow_data(direction, flow_direction, time_type).await {
Ok(data) => {
match serde_json::to_string(&data) {
Ok(json) => {
if sender.send(Message::Text(json.into())).await.is_err() { break; }
}
Err(e) => { log!(MiscError::SerializeError(e)); }
}
}
Err(e) => { log!(e); break; }
}
},
}
}
}
let _ = session.close(None).await;
}
async fn handle_ipv6_flow_connection(
app_config: Arc<AppConfig>,
statistics: Arc<Statistics>,
mut session: Session,
mut msg_stream: MessageStream,
pub async fn handle_ipv6_flow(
socket: WebSocket,
state: AppState,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) {
let config = app_config.config.clone();
let refresh_interval = Duration::from_secs(config.refresh_interval);
let refresh_interval = Duration::from_secs(state.app_config.refresh_interval);
let (mut sender, mut receiver) = socket.split();
let mut data_interval = interval(refresh_interval);
loop {
tokio::select! {
msg_result = msg_stream.next() => {
if !handle_client_message(&mut session, msg_result).await {
break;
msg = receiver.next() => {
match msg {
Some(Ok(Message::Ping(data))) => {
if sender.send(Message::Pong(data)).await.is_err() { break; }
}
Some(Ok(Message::Close(_))) | None => break,
Some(Err(e)) => {
log!(HttpError::WebSocketError { msg: e.to_string() });
break;
}
_ => {}
}
},
}
_ = data_interval.tick() => {
if !send_ipv6_flow_data(&statistics, &mut session, direction, flow_direction, time_type).await {
break;
match state.statistics.get_ipv6_flow_data(direction, flow_direction, time_type).await {
Ok(data) => {
match serde_json::to_string(&data) {
Ok(json) => {
if sender.send(Message::Text(json.into())).await.is_err() { break; }
}
Err(e) => { log!(MiscError::SerializeError(e)); }
}
}
Err(e) => { log!(e); break; }
}
},
}
}
let _ = session.close(None).await;
}
async fn handle_client_message(
session: &mut Session,
msg_result: Option<Result<Message, actix_ws::ProtocolError>>,
) -> bool {
match msg_result {
Some(Ok(Message::Text(_))) => true,
Some(Ok(Message::Ping(bytes))) => session.pong(&bytes).await.is_ok(),
Some(Ok(Message::Close(reason))) => {
let _ = (session.clone()).close(reason).await;
false
}
Some(Err(err)) => {
log!(HttpError::WebSocketError(err));
false
}
None => false,
_ => true,
}
}
async fn send_ipv4_flow_data(
statistics: &Arc<Statistics>,
session: &mut Session,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) -> bool {
let flow_data = match statistics.get_ipv4_flow_data(direction, flow_direction, time_type).await {
Ok(data) => data,
Err(e) => { log!(e); return false; }
};
match serde_json::to_string(&flow_data) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
log!(MiscError::SerializeError(err));
true
}
}
}
async fn send_ipv6_flow_data(
statistics: &Arc<Statistics>,
session: &mut Session,
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) -> bool {
let flow_data = match statistics.get_ipv6_flow_data(direction, flow_direction, time_type).await {
Ok(data) => data,
Err(e) => { log!(e); return false; }
};
match serde_json::to_string(&flow_data) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
log!(MiscError::SerializeError(err));
true
}
}
}
}

104
mantis/src/web/websocket/health_websocket.rs
View File

@ -1,91 +1,47 @@
use actix_web::{web, HttpRequest, HttpResponse, Result};
use actix_ws::{handle, Message, MessageStream, Session};
use futures_util::StreamExt;
use axum::extract::ws::{Message, WebSocket};
use futures_util::{SinkExt, StreamExt};
use macros::log;
use tokio::sync::broadcast;
use crate::core::infrastructure::health::SystemHealth;
use crate::model::health::SystemHealthMetrics;
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::log::http::HttpLog;
use crate::model::health::SystemHealthMetrics;
pub async fn websocket_system_health(
req: HttpRequest,
body: web::Payload,
health: web::Data<SystemHealth>,
) -> Result<HttpResponse> {
let (response, session, msg_stream) = handle(&req, body)?;
pub async fn handle_health(socket: WebSocket, mut broadcast_rx: broadcast::Receiver<SystemHealthMetrics>) {
let (mut sender, mut receiver) = socket.split();
let broadcast_rx = health.subscribe_to_metrics();
actix_web::rt::spawn(async move {
handle_health_connection(session, msg_stream, broadcast_rx).await;
});
Ok(response)
}
async fn handle_health_connection(
mut session: Session,
mut msg_stream: MessageStream,
mut broadcast_rx: broadcast::Receiver<SystemHealthMetrics>,
) {
loop {
tokio::select! {
msg_result = msg_stream.next() => {
if !handle_client_message(&mut session, msg_result).await {
break;
}
},
broadcast_result = broadcast_rx.recv() => {
match broadcast_result {
Ok(metrics) => {
if !send_metrics(&mut session, &metrics).await {
break;
}
msg = receiver.next() => {
match msg {
Some(Ok(Message::Ping(data))) => {
if sender.send(Message::Pong(data)).await.is_err() { break; }
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
log!(HttpLog::WebSocketLaged(skipped));
continue;
}
Err(broadcast::error::RecvError::Closed) => {
Some(Ok(Message::Close(_))) | None => break,
Some(Err(e)) => {
log!(HttpError::WebSocketError { msg: e.to_string() });
break;
}
_ => {}
}
},
}
result = broadcast_rx.recv() => {
match result {
Ok(metrics) => {
match serde_json::to_string(&metrics) {
Ok(json) => {
if sender.send(Message::Text(json.into())).await.is_err() { break; }
}
Err(e) => { log!(MiscError::SerializeError(e)); }
}
}
Err(broadcast::error::RecvError::Lagged(n)) => {
log!(HttpLog::WebSocketLaged { skipped: n });
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
}
}
let _ = session.close(None).await;
}
async fn handle_client_message(
session: &mut Session,
msg_result: Option<Result<Message, actix_ws::ProtocolError>>,
) -> bool {
match msg_result {
Some(Ok(Message::Text(_))) => true,
Some(Ok(Message::Ping(bytes))) => session.pong(&bytes).await.is_ok(),
Some(Ok(Message::Close(reason))) => {
let _ = (session.clone()).close(reason).await;
false
}
Some(Err(err)) => {
log!(HttpError::WebSocketError(err));
false
}
None => false,
_ => true,
}
}
async fn send_metrics(session: &mut Session, metrics: &SystemHealthMetrics) -> bool {
match serde_json::to_string(metrics) {
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
log!(MiscError::SerializeError(err));
false
}
}
}