use crate::api::models::*;
use crate::api::{AppState, RouteRing};
use crate::commands::service::load_xbp_config;
use crate::commands::system_diag::get_system_metrics;
use crate::commands::{
install_package as install_package_cmd, pm2_delete, pm2_save, pm2_stop, run_config, run_ports,
run_redeploy, run_redeploy_service, run_service_command as run_service_cmd, run_setup,
};
use crate::sdk::network::{
add_floating_ip, list_floating_ips, list_network_config_sources, AddFloatingIpRequest,
};
use crate::strategies::get_all_services;
use crate::utils::command_exists;
use actix_web::{web, HttpRequest, HttpResponse, Responder};
use once_cell::sync::Lazy;
use prometheus::{Encoder, Gauge, IntCounterVec, IntGauge, IntGaugeVec, Registry, TextEncoder};
use reqwest;
use reqwest::header::{HeaderName as ReqHeaderName, HeaderValue as ReqHeaderValue};
use std::collections::HashSet;
use std::fs::{self, File};
use std::io::{BufRead, BufReader};
use std::path::PathBuf;
use std::sync::atomic::Ordering;
use std::time::Instant;
use tokio::process::Command;
use tracing::{error, info};
static REGISTRY: Lazy<Registry> = Lazy::new(Registry::new);
static PROXY_REQUESTS: Lazy<IntCounterVec> = Lazy::new(|| {
IntCounterVec::new(
prometheus::Opts::new("xbp_proxy_requests_total", "Total proxied requests"),
&["domain", "target"],
)
.expect("proxy counter")
});
static PROXY_FAILURES: Lazy<IntCounterVec> = Lazy::new(|| {
IntCounterVec::new(
prometheus::Opts::new("xbp_proxy_failures_total", "Proxy failures"),
&["domain", "target"],
)
.expect("proxy failure counter")
});
static HOST_CPU_USAGE_PERCENT: Lazy<Gauge> = Lazy::new(|| {
Gauge::new(
"xbp_host_cpu_usage_percent",
"CPU usage percentage across all CPUs",
)
.expect("host cpu gauge")
});
static HOST_MEMORY_TOTAL_BYTES: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new("xbp_host_memory_total_bytes", "Total host memory in bytes")
.expect("host memory total gauge")
});
static HOST_MEMORY_USED_BYTES: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new("xbp_host_memory_used_bytes", "Used host memory in bytes")
.expect("host memory used gauge")
});
static HOST_MEMORY_USAGE_PERCENT: Lazy<Gauge> = Lazy::new(|| {
Gauge::new(
"xbp_host_memory_usage_percent",
"Host memory usage percentage",
)
.expect("host memory percent gauge")
});
static HOST_DISK_TOTAL_BYTES: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new("xbp_host_disk_total_bytes", "Total host disk bytes")
.expect("host disk total gauge")
});
static HOST_DISK_USED_BYTES: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new("xbp_host_disk_used_bytes", "Used host disk bytes").expect("host disk used gauge")
});
static HOST_DISK_USAGE_PERCENT: Lazy<Gauge> = Lazy::new(|| {
Gauge::new("xbp_host_disk_usage_percent", "Host disk usage percentage")
.expect("host disk percent gauge")
});
static HOST_NETWORK_RX_BYTES_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_host_network_receive_bytes_total",
"Total received network bytes since boot",
)
.expect("host network rx gauge")
});
static HOST_NETWORK_TX_BYTES_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_host_network_transmit_bytes_total",
"Total transmitted network bytes since boot",
)
.expect("host network tx gauge")
});
static HOST_UPTIME_SECONDS: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new("xbp_host_uptime_seconds", "Host uptime in seconds").expect("host uptime gauge")
});
static HOST_PROCESS_COUNT: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_host_process_count",
"Total number of running processes",
)
.expect("host process gauge")
});
static HOST_LISTENING_PORTS_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_host_listening_ports_total",
"Number of unique listening TCP ports",
)
.expect("host listening ports gauge")
});
static HOST_EXPOSED_PORTS_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_host_exposed_ports_total",
"Number of unique listening TCP ports bound to all interfaces",
)
.expect("host exposed ports gauge")
});
static SYSTEMD_SERVICES_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_systemd_services_total",
"Total number of systemd services reported by systemctl",
)
.expect("systemd total gauge")
});
static SYSTEMD_SERVICES_RUNNING_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_systemd_services_running_total",
"Number of active systemd services",
)
.expect("systemd running gauge")
});
static SYSTEMD_SERVICES_FAILED_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_systemd_services_failed_total",
"Number of failed systemd services",
)
.expect("systemd failed gauge")
});
static SYSTEMD_SERVICE_ACTIVE: Lazy<IntGaugeVec> = Lazy::new(|| {
IntGaugeVec::new(
prometheus::Opts::new(
"xbp_systemd_service_active",
"Whether a systemd service is currently active (1 active, 0 otherwise)",
),
&["service"],
)
.expect("systemd service active gauge vec")
});
static NGINX_CONFIGS_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_nginx_configs_total",
"Number of Nginx configuration files in sites-available",
)
.expect("nginx config count gauge")
});
static NGINX_UP: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new("xbp_nginx_up", "Whether Nginx is active (1 up, 0 down)").expect("nginx up gauge")
});
static NGINX_CONFIG_VALID: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_nginx_config_valid",
"Whether `nginx -t` currently succeeds (1 valid, 0 invalid)",
)
.expect("nginx config valid gauge")
});
static NGINX_ACCESS_LOG_BYTES_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_nginx_access_log_bytes_total",
"Total bytes on disk for Nginx access logs",
)
.expect("nginx access log bytes gauge")
});
static NGINX_ERROR_LOG_BYTES_TOTAL: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_nginx_error_log_bytes_total",
"Total bytes on disk for Nginx error logs",
)
.expect("nginx error log bytes gauge")
});
static NGINX_RESPONSE_BYTES_TOTAL_FROM_LOGS: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_nginx_response_bytes_total_from_logs",
"Summed response bytes from Nginx access logs (best effort)",
)
.expect("nginx response bytes from logs gauge")
});
static NGINX_REQUEST_BYTES_TOTAL_FROM_LOGS: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_nginx_request_bytes_total_from_logs",
"Summed request bytes from Nginx access logs when request_length is logged (best effort)",
)
.expect("nginx request bytes from logs gauge")
});
static NGINX_REQUESTS_TOTAL_FROM_LOGS: Lazy<IntGauge> = Lazy::new(|| {
IntGauge::new(
"xbp_nginx_requests_total_from_logs",
"Total request lines parsed from Nginx access logs",
)
.expect("nginx requests from logs gauge")
});
const MAX_EXEC_ARGS: usize = 32;
const MAX_EXEC_ARG_LEN: usize = 1024;
const MAX_EXEC_OUTPUT_BYTES: usize = 64 * 1024;
fn ensure_metrics_registered() {
REGISTRY.register(Box::new(PROXY_REQUESTS.clone())).ok();
REGISTRY.register(Box::new(PROXY_FAILURES.clone())).ok();
REGISTRY
.register(Box::new(HOST_CPU_USAGE_PERCENT.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_MEMORY_TOTAL_BYTES.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_MEMORY_USED_BYTES.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_MEMORY_USAGE_PERCENT.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_DISK_TOTAL_BYTES.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_DISK_USED_BYTES.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_DISK_USAGE_PERCENT.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_NETWORK_RX_BYTES_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_NETWORK_TX_BYTES_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_UPTIME_SECONDS.clone()))
.ok();
REGISTRY.register(Box::new(HOST_PROCESS_COUNT.clone())).ok();
REGISTRY
.register(Box::new(HOST_LISTENING_PORTS_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(HOST_EXPOSED_PORTS_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(SYSTEMD_SERVICES_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(SYSTEMD_SERVICES_RUNNING_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(SYSTEMD_SERVICES_FAILED_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(SYSTEMD_SERVICE_ACTIVE.clone()))
.ok();
REGISTRY
.register(Box::new(NGINX_CONFIGS_TOTAL.clone()))
.ok();
REGISTRY.register(Box::new(NGINX_UP.clone())).ok();
REGISTRY.register(Box::new(NGINX_CONFIG_VALID.clone())).ok();
REGISTRY
.register(Box::new(NGINX_ACCESS_LOG_BYTES_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(NGINX_ERROR_LOG_BYTES_TOTAL.clone()))
.ok();
REGISTRY
.register(Box::new(NGINX_RESPONSE_BYTES_TOTAL_FROM_LOGS.clone()))
.ok();
REGISTRY
.register(Box::new(NGINX_REQUEST_BYTES_TOTAL_FROM_LOGS.clone()))
.ok();
REGISTRY
.register(Box::new(NGINX_REQUESTS_TOTAL_FROM_LOGS.clone()))
.ok();
}
pub async fn health() -> impl Responder {
HttpResponse::Ok().json(HealthResponse {
status: "ok".to_string(),
version: env!("CARGO_PKG_VERSION").to_string(),
})
}
pub async fn exec_command(payload: web::Json<ExecCommandRequest>) -> impl Responder {
let command = payload.command.trim().to_ascii_lowercase();
if !is_valid_exec_command_name(&command) {
return HttpResponse::BadRequest().json(ErrorResponse {
error: "Invalid command name".to_string(),
});
}
if !is_allowed_exec_command(&command) {
return HttpResponse::Forbidden().json(ErrorResponse {
error: format!("Command '{}' is not allowed via API", command),
});
}
if let Err(reason) = validate_exec_args(&payload.args) {
return HttpResponse::BadRequest().json(ErrorResponse { error: reason });
}
let executable = match std::env::current_exe() {
Ok(path) => path,
Err(e) => {
return HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("Failed to resolve xbp executable: {}", e),
})
}
};
info!(
"Executing API command '{}' with {} arg(s)",
command,
payload.args.len()
);
let started = Instant::now();
let output = Command::new(executable)
.arg(&command)
.args(&payload.args)
.env_remove("PORT_XBP_API")
.env_remove("XBP_API_BIND")
.output()
.await;
match output {
Ok(output) => {
let (stdout, truncated_stdout) = truncate_command_output(&output.stdout);
let (stderr, truncated_stderr) = truncate_command_output(&output.stderr);
HttpResponse::Ok().json(ExecCommandResponse {
success: output.status.success(),
command,
args: payload.args.clone(),
exit_code: output.status.code().unwrap_or(-1),
stdout,
stderr,
duration_ms: started.elapsed().as_millis(),
truncated_stdout,
truncated_stderr,
})
}
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("Failed to execute command: {}", e),
}),
}
}
pub async fn list_routes(state: web::Data<AppState>) -> impl Responder {
let routes = state.routes.read().await;
let items: Vec<RouteEntry> = routes
.iter()
.map(|(domain, ring)| RouteEntry {
domain: domain.clone(),
targets: ring.targets.clone(),
conditions: ring.conditions.clone(),
})
.collect();
HttpResponse::Ok().json(RoutesResponse { routes: items })
}
pub async fn create_route(
state: web::Data<AppState>,
payload: web::Json<CreateRouteRequest>,
) -> impl Responder {
ensure_metrics_registered();
if payload.targets.is_empty() {
return HttpResponse::BadRequest().json(ErrorResponse {
error: "At least one target is required".into(),
});
}
let entry = RouteEntry {
domain: payload.domain.clone(),
targets: payload.targets.clone(),
conditions: payload.conditions.clone(),
};
let mut routes = state.routes.write().await;
routes.insert(payload.domain.clone(), RouteRing::new(entry.clone()));
HttpResponse::Ok().json(CreateRouteResponse {
success: true,
domain: payload.domain.clone(),
target_count: payload.targets.len(),
})
}
pub async fn delete_route(state: web::Data<AppState>, path: web::Path<String>) -> impl Responder {
let domain = path.into_inner();
let mut routes = state.routes.write().await;
if routes.remove(&domain).is_some() {
HttpResponse::Ok().json(serde_json::json!({ "success": true }))
} else {
HttpResponse::NotFound().json(ErrorResponse {
error: format!("Route for {} not found", domain),
})
}
}
pub async fn list_ports() -> impl Responder {
match get_ports_data(None).await {
Ok(ports) => HttpResponse::Ok().json(PortsResponse { ports }),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn get_port(path: web::Path<u16>) -> impl Responder {
let port = path.into_inner();
match get_ports_data(Some(port)).await {
Ok(ports) => HttpResponse::Ok().json(PortsResponse { ports }),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn kill_port(path: web::Path<u16>) -> impl Responder {
let port = path.into_inner();
let args = vec!["-p".to_string(), port.to_string(), "--kill".to_string()];
match run_ports(&args, false).await {
Ok(_) => HttpResponse::Ok().json(serde_json::json!({"success": true, "message": format!("Killed processes on port {}", port)})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn list_network_floating_ips() -> impl Responder {
match list_floating_ips().await {
Ok(payload) => HttpResponse::Ok().json(payload),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse {
error: e.to_string(),
}),
}
}
pub async fn add_network_floating_ip(
payload: web::Json<AddFloatingIpApiRequest>,
) -> impl Responder {
let request = AddFloatingIpRequest {
ip: payload.ip.clone(),
cidr: payload.cidr,
interface: payload.interface.clone(),
label: payload.label.clone(),
apply: payload.apply,
dry_run: payload.dry_run,
};
match add_floating_ip(request).await {
Ok(response) => HttpResponse::Ok().json(response),
Err(e) => HttpResponse::BadRequest().json(ErrorResponse {
error: e.to_string(),
}),
}
}
pub async fn list_network_configs() -> impl Responder {
match list_network_config_sources().await {
Ok(payload) => HttpResponse::Ok().json(payload),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse {
error: e.to_string(),
}),
}
}
async fn get_ports_data(port_filter: Option<u16>) -> Result<Vec<PortInfo>, String> {
use netstat2::{get_sockets_info, AddressFamilyFlags, ProtocolFlags, ProtocolSocketInfo};
use std::collections::BTreeMap;
let af_flags = AddressFamilyFlags::IPV4 | AddressFamilyFlags::IPV6;
let proto_flags = ProtocolFlags::TCP;
let sockets = get_sockets_info(af_flags, proto_flags)
.map_err(|e| format!("Failed to get sockets info: {}", e))?;
let mut port_map: BTreeMap<u16, Vec<PortInfo>> = BTreeMap::new();
for socket in sockets {
if let ProtocolSocketInfo::Tcp(ref tcp_info) = socket.protocol_socket_info {
if let Some(filter) = port_filter {
if tcp_info.local_port != filter {
continue;
}
}
let pid = if !socket.associated_pids.is_empty() {
Some(socket.associated_pids[0].to_string())
} else {
None
};
let port_info = PortInfo {
port: tcp_info.local_port,
pid,
local_addr: tcp_info.local_addr.to_string(),
remote_addr: tcp_info.remote_addr.to_string(),
state: format!("{:?}", tcp_info.state),
process: "-".to_string(),
};
port_map
.entry(tcp_info.local_port)
.or_default()
.push(port_info);
}
}
let mut result = Vec::new();
for ports in port_map.values() {
result.extend(ports.iter().cloned());
}
Ok(result)
}
pub async fn list_systemctl() -> impl Responder {
match get_systemctl_data(None).await {
Ok(services) => HttpResponse::Ok().json(SystemctlResponse { services }),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn get_systemctl_service(path: web::Path<String>) -> impl Responder {
let service_name = path.into_inner();
match get_systemctl_data(Some(&service_name)).await {
Ok(services) => HttpResponse::Ok().json(SystemctlResponse { services }),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn systemctl_action(path: web::Path<(String, String)>) -> impl Responder {
let (service_name, action) = path.into_inner();
let allowed = ["start", "stop", "restart", "enable", "disable"];
if !allowed.contains(&action.as_str()) {
return HttpResponse::BadRequest().json(ErrorResponse {
error: format!("Unsupported systemctl action: {}", action),
});
}
match apply_systemctl_action(&service_name, &action).await {
Ok(message) => HttpResponse::Ok().json(serde_json::json!({
"success": true,
"service": service_name,
"action": action,
"message": message
})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
async fn get_systemctl_data(filter: Option<&str>) -> Result<Vec<SystemctlService>, String> {
if !cfg!(target_os = "linux") || !command_exists("systemctl") {
return Err(
"systemctl is only available on Linux hosts where systemd is installed.".to_string(),
);
}
let mut cmd = Command::new("systemctl");
cmd.arg("list-units");
cmd.arg("--type=service");
cmd.arg("--no-pager");
cmd.arg("--no-legend");
let output = cmd
.output()
.await
.map_err(|e| format!("Failed to run systemctl: {}", e))?;
if !output.status.success() {
return Err(String::from_utf8_lossy(&output.stderr).to_string());
}
let stdout = String::from_utf8_lossy(&output.stdout);
let mut services = Vec::new();
for line in stdout.lines() {
let parts: Vec<&str> = line.split_whitespace().collect();
if parts.len() < 4 {
continue;
}
let name = parts[0].to_string();
if let Some(filter_name) = filter {
if !name.contains(filter_name) {
continue;
}
}
let status = parts[2].to_string();
let active = status == "active";
let enabled = parts[3] == "enabled";
services.push(SystemctlService {
name,
status,
active,
enabled,
});
}
Ok(services)
}
async fn apply_systemctl_action(service: &str, action: &str) -> Result<String, String> {
if !cfg!(target_os = "linux") || !command_exists("systemctl") {
return Err(
"systemctl is only available on Linux hosts where systemd is installed.".to_string(),
);
}
let output = Command::new("systemctl")
.arg(action)
.arg(service)
.output()
.await
.map_err(|e| format!("Failed to run systemctl {}: {}", action, e))?;
if output.status.success() {
Ok(format!("systemctl {} {} succeeded", action, service))
} else {
let stderr = String::from_utf8_lossy(&output.stderr);
Err(format!(
"systemctl {} {} failed: {}",
action, service, stderr
))
}
}
pub async fn list_pm2() -> impl Responder {
match get_pm2_data().await {
Ok(processes) => HttpResponse::Ok().json(Pm2Response { processes }),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
async fn get_pm2_data() -> Result<Vec<Pm2Process>, String> {
let mut cmd = Command::new("pm2");
cmd.arg("jlist");
let output = cmd
.output()
.await
.map_err(|e| format!("Failed to run pm2 jlist: {}", e))?;
if !output.status.success() {
return Err(String::from_utf8_lossy(&output.stderr).to_string());
}
let stdout = String::from_utf8_lossy(&output.stdout);
let processes: Vec<serde_json::Value> =
serde_json::from_str(&stdout).map_err(|e| format!("Failed to parse pm2 output: {}", e))?;
let mut result = Vec::new();
for proc in processes {
let name = proc["name"].as_str().unwrap_or("unknown").to_string();
let pid = proc["pid"].as_u64().map(|p| p as u32);
let status = proc["pm2_env"]["status"]
.as_str()
.unwrap_or("unknown")
.to_string();
let cpu = proc["monit"]["cpu"].as_f64();
let memory = proc["monit"]["memory"]
.as_f64()
.map(|m| m / 1024.0 / 1024.0);
let uptime = proc["pm2_env"]["pm_uptime"].as_u64().map(|u| {
let seconds = u / 1000;
format!("{}s", seconds)
});
result.push(Pm2Process {
name,
pid,
status,
cpu,
memory,
uptime,
});
}
Ok(result)
}
pub async fn delete_pm2(path: web::Path<String>) -> impl Responder {
let name = path.into_inner();
match pm2_delete(&name, false).await {
Ok(_) => HttpResponse::Ok().json(serde_json::json!({"success": true, "message": format!("Deleted PM2 process: {}", name)})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn start_pm2(path: web::Path<String>) -> impl Responder {
let name = path.into_inner();
let mut cmd = Command::new("pm2");
cmd.arg("start").arg(&name);
match cmd.output().await {
Ok(output) => {
if output.status.success() {
HttpResponse::Ok().json(serde_json::json!({"success": true, "message": format!("Started PM2 process: {}", name)}))
} else {
let stderr = String::from_utf8_lossy(&output.stderr);
HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("PM2 start failed: {}", stderr),
})
}
}
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("Failed to start PM2 process: {}", e),
}),
}
}
pub async fn stop_pm2(path: web::Path<String>) -> impl Responder {
let name = path.into_inner();
match pm2_stop(&name, false).await {
Ok(_) => HttpResponse::Ok().json(serde_json::json!({"success": true, "message": format!("Stopped PM2 process: {}", name)})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn restart_pm2(path: web::Path<String>) -> impl Responder {
let name = path.into_inner();
let mut cmd = Command::new("pm2");
cmd.arg("restart").arg(&name);
match cmd.output().await {
Ok(output) => {
if output.status.success() {
HttpResponse::Ok().json(serde_json::json!({"success": true, "message": format!("Restarted PM2 process: {}", name)}))
} else {
let stderr = String::from_utf8_lossy(&output.stderr);
HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("PM2 restart failed: {}", stderr),
})
}
}
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("Failed to restart PM2 process: {}", e),
}),
}
}
pub async fn list_services() -> impl Responder {
match load_xbp_config().await {
Ok(config) => {
let services = get_all_services(&config);
let service_infos: Vec<ServiceInfo> = services
.iter()
.map(|s| ServiceInfo {
name: s.name.clone(),
target: s.target.clone(),
port: s.port,
branch: s.branch.clone(),
url: s.url.clone(),
})
.collect();
HttpResponse::Ok().json(ServicesResponse {
services: service_infos,
})
}
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn run_service_command(path: web::Path<(String, String)>) -> impl Responder {
let (name, command) = path.into_inner();
match run_service_cmd(&command, &name, false).await {
Ok(_) => HttpResponse::Ok().json(serde_json::json!({"success": true, "message": format!("Executed {} on service {}", command, name)})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn get_config() -> impl Responder {
match run_config(false).await {
Ok(_) => HttpResponse::Ok().json(serde_json::json!({"success": true})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn get_logs() -> impl Responder {
HttpResponse::Ok().json(serde_json::json!({"message": "Logs endpoint - use /logs?command=<command> for specific logs"}))
}
pub async fn install_package(path: web::Path<String>) -> impl Responder {
let package = path.into_inner();
match install_package_cmd(&package, false).await {
Ok(_) => HttpResponse::Ok().json(serde_json::json!({"success": true, "message": format!("Installed package: {}", package)})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn setup() -> impl Responder {
match run_setup(false).await {
Ok(_) => HttpResponse::Ok()
.json(serde_json::json!({"success": true, "message": "Setup completed"})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn redeploy() -> impl Responder {
match run_redeploy().await {
Ok(_) => HttpResponse::Ok()
.json(serde_json::json!({"success": true, "message": "Redeploy completed"})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn redeploy_service(path: web::Path<String>) -> impl Responder {
let service_name = path.into_inner();
match run_redeploy_service(&service_name, false).await {
Ok(_) => HttpResponse::Ok().json(serde_json::json!({"success": true, "message": format!("Redeployed service: {}", service_name)})),
Err(e) => HttpResponse::InternalServerError().json(ErrorResponse { error: e }),
}
}
pub async fn download_and_run_binary(req: web::Json<BinaryDownloadRequest>) -> impl Responder {
let download_req = req.into_inner();
info!("Downloading binary from: {}", download_req.url);
let client = reqwest::Client::new();
let response = match client.get(&download_req.url).send().await {
Ok(resp) => resp,
Err(e) => {
error!("Failed to download binary: {}", e);
return HttpResponse::BadRequest().json(ErrorResponse {
error: format!("Failed to download binary: {}", e),
});
}
};
let bytes = match response.bytes().await {
Ok(b) => b,
Err(e) => {
error!("Failed to read binary data: {}", e);
return HttpResponse::BadRequest().json(ErrorResponse {
error: format!("Failed to read binary data: {}", e),
});
}
};
let binary_path = format!("/tmp/{}", download_req.name);
match fs::write(&binary_path, &bytes) {
Ok(_) => {
info!("Binary saved to: {}", binary_path);
}
Err(e) => {
error!("Failed to save binary: {}", e);
return HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("Failed to save binary: {}", e),
});
}
}
let chmod_output = Command::new("chmod")
.arg("+x")
.arg(&binary_path)
.output()
.await;
if let Err(e) = chmod_output {
error!("Failed to make binary executable: {}", e);
return HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("Failed to make binary executable: {}", e),
});
}
let mut pm2_cmd = Command::new("pm2");
pm2_cmd.arg("start");
pm2_cmd.arg(&binary_path);
pm2_cmd.arg("--name");
pm2_cmd.arg(&download_req.name);
if let Some(ref args) = download_req.args {
for arg in args {
pm2_cmd.arg(arg);
}
}
match pm2_cmd.output().await {
Ok(output) => {
if output.status.success() {
let _ = pm2_save(false).await;
HttpResponse::Ok().json(BinaryDownloadResponse {
success: true,
message: format!(
"Binary downloaded and started as PM2 process: {}",
download_req.name
),
pm2_name: Some(download_req.name),
})
} else {
let stderr = String::from_utf8_lossy(&output.stderr);
error!("PM2 start failed: {}", stderr);
HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("PM2 start failed: {}", stderr),
})
}
}
Err(e) => {
error!("Failed to start PM2 process: {}", e);
HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("Failed to start PM2 process: {}", e),
})
}
}
}
pub async fn download_openapi() -> impl Responder {
let openapi_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("openapi.yaml");
match fs::read(&openapi_path) {
Ok(bytes) => HttpResponse::Ok()
.insert_header(("Content-Type", "application/yaml"))
.insert_header((
"Content-Disposition",
"attachment; filename=\"openapi.yaml\"",
))
.body(bytes),
Err(_) => HttpResponse::NotFound().json(ErrorResponse {
error: "OpenAPI spec not found".to_string(),
}),
}
}
pub async fn proxy_route(
path: web::Path<(String, String)>,
req: HttpRequest,
body: web::Bytes,
state: web::Data<AppState>,
) -> impl Responder {
ensure_metrics_registered();
let (domain, tail) = path.into_inner();
let routes = state.routes.read().await;
let ring = match routes.get(&domain) {
Some(r) => r.clone(),
None => {
return HttpResponse::NotFound().json(ErrorResponse {
error: format!("No route configured for {}", domain),
})
}
};
if let Some(cond) = &ring.conditions {
if let Some(prefix) = &cond.path_prefix {
if !tail.starts_with(prefix) {
return HttpResponse::NotFound().json(ErrorResponse {
error: "Path prefix condition not met".into(),
});
}
}
if let Some(header) = &cond.header {
let mut parts = header.splitn(2, ':');
if let (Some(name), Some(expected)) = (parts.next(), parts.next()) {
if req
.headers()
.get(name.trim())
.and_then(|v| v.to_str().ok())
.map(|v| v != expected.trim())
.unwrap_or(true)
{
return HttpResponse::NotFound().json(ErrorResponse {
error: "Header condition not met".into(),
});
}
}
}
}
let target = select_target(&ring).cloned();
drop(routes);
let target = match target {
Some(t) => t,
None => {
return HttpResponse::ServiceUnavailable().json(ErrorResponse {
error: "No targets available".into(),
})
}
};
let url = format!("{}/{}", target.url.trim_end_matches('/'), tail);
let method = match reqwest::Method::from_bytes(req.method().as_str().as_bytes()) {
Ok(m) => m,
Err(_) => {
return HttpResponse::MethodNotAllowed().json(ErrorResponse {
error: format!("Unsupported method {}", req.method()),
})
}
};
let mut builder = state.client.request(method, &url);
for (name, value) in req.headers().iter() {
let name_lower = name.as_str().to_ascii_lowercase();
if matches!(
name_lower.as_str(),
"host" | "content-length" | "connection" | "upgrade" | "proxy-connection"
) {
continue;
}
if let (Ok(hname), Ok(hval)) = (
ReqHeaderName::from_bytes(name.as_str().as_bytes()),
ReqHeaderValue::from_bytes(value.as_bytes()),
) {
builder = builder.header(hname, hval);
}
}
let response = match builder.body(body.clone()).send().await {
Ok(res) => res,
Err(e) => {
PROXY_FAILURES
.with_label_values(&[&domain, &target.url])
.inc();
return HttpResponse::BadGateway().json(ErrorResponse {
error: format!("Proxy request failed: {}", e),
});
}
};
PROXY_REQUESTS
.with_label_values(&[&domain, &target.url])
.inc();
let status = actix_web::http::StatusCode::from_u16(response.status().as_u16())
.unwrap_or(actix_web::http::StatusCode::BAD_GATEWAY);
let mut resp_builder = HttpResponse::build(status);
for (name, value) in response.headers() {
if name == "connection" || name == "content-length" {
continue;
}
if let (Ok(hname), Ok(hval)) = (
actix_web::http::header::HeaderName::from_bytes(name.as_str().as_bytes()),
actix_web::http::header::HeaderValue::from_bytes(value.as_bytes()),
) {
resp_builder.insert_header((hname, hval));
}
}
match response.bytes().await {
Ok(bytes) => resp_builder.body(bytes),
Err(e) => HttpResponse::BadGateway().json(ErrorResponse {
error: format!("Failed to read proxied body: {}", e),
}),
}
}
pub async fn metrics() -> impl Responder {
ensure_metrics_registered();
refresh_metrics_snapshot().await;
let metric_families = REGISTRY.gather();
let mut buffer = Vec::new();
let encoder = TextEncoder::new();
if let Err(e) = encoder.encode(&metric_families, &mut buffer) {
return HttpResponse::InternalServerError().json(ErrorResponse {
error: format!("Failed to encode metrics: {}", e),
});
}
HttpResponse::Ok()
.content_type(encoder.format_type())
.body(buffer)
}
async fn refresh_metrics_snapshot() {
refresh_host_metrics().await;
refresh_listening_port_metrics();
refresh_systemd_metrics().await;
refresh_nginx_metrics().await;
}
async fn refresh_host_metrics() {
let metrics = match get_system_metrics().await {
Ok(value) => value,
Err(_) => return,
};
HOST_CPU_USAGE_PERCENT.set(metrics.cpu_usage as f64);
HOST_MEMORY_TOTAL_BYTES.set(metrics.memory_total as i64);
HOST_MEMORY_USED_BYTES.set(metrics.memory_used as i64);
HOST_MEMORY_USAGE_PERCENT.set(metrics.memory_percent as f64);
HOST_DISK_TOTAL_BYTES.set(metrics.disk_total as i64);
HOST_DISK_USED_BYTES.set(metrics.disk_used as i64);
HOST_DISK_USAGE_PERCENT.set(metrics.disk_percent as f64);
HOST_NETWORK_RX_BYTES_TOTAL.set(metrics.network_rx as i64);
HOST_NETWORK_TX_BYTES_TOTAL.set(metrics.network_tx as i64);
HOST_UPTIME_SECONDS.set(metrics.uptime as i64);
HOST_PROCESS_COUNT.set(metrics.process_count as i64);
}
fn refresh_listening_port_metrics() {
use netstat2::{get_sockets_info, AddressFamilyFlags, ProtocolFlags, ProtocolSocketInfo};
let sockets = match get_sockets_info(
AddressFamilyFlags::IPV4 | AddressFamilyFlags::IPV6,
ProtocolFlags::TCP,
) {
Ok(value) => value,
Err(_) => return,
};
let mut listening_ports: HashSet<u16> = HashSet::new();
let mut exposed_ports: HashSet<u16> = HashSet::new();
for socket in sockets {
if let ProtocolSocketInfo::Tcp(tcp) = socket.protocol_socket_info {
let state = format!("{:?}", tcp.state);
if state != "Listen" && state != "LISTEN" {
continue;
}
listening_ports.insert(tcp.local_port);
if tcp.local_addr.is_unspecified() {
exposed_ports.insert(tcp.local_port);
}
}
}
HOST_LISTENING_PORTS_TOTAL.set(listening_ports.len() as i64);
HOST_EXPOSED_PORTS_TOTAL.set(exposed_ports.len() as i64);
}
async fn refresh_systemd_metrics() {
if !cfg!(target_os = "linux") || !command_exists("systemctl") {
return;
}
let output = match Command::new("systemctl")
.args([
"list-units",
"--type=service",
"--all",
"--no-pager",
"--no-legend",
"--plain",
])
.output()
.await
{
Ok(value) => value,
Err(_) => return,
};
if !output.status.success() {
return;
}
let stdout = String::from_utf8_lossy(&output.stdout);
let mut total = 0_i64;
let mut running = 0_i64;
let mut failed = 0_i64;
SYSTEMD_SERVICE_ACTIVE.reset();
for line in stdout.lines() {
let parts: Vec<&str> = line.split_whitespace().collect();
if parts.len() < 3 {
continue;
}
let service = parts[0];
let active_state = parts[2];
total += 1;
if active_state == "active" {
running += 1;
}
if active_state == "failed" {
failed += 1;
}
let active = if active_state == "active" { 1 } else { 0 };
SYSTEMD_SERVICE_ACTIVE
.with_label_values(&[service])
.set(active);
}
SYSTEMD_SERVICES_TOTAL.set(total);
SYSTEMD_SERVICES_RUNNING_TOTAL.set(running);
SYSTEMD_SERVICES_FAILED_TOTAL.set(failed);
}
async fn refresh_nginx_metrics() {
if !cfg!(target_os = "linux") {
return;
}
let config_count = fs::read_dir("/etc/nginx/sites-available")
.ok()
.map(|iter| iter.filter_map(|entry| entry.ok()).count())
.unwrap_or(0);
NGINX_CONFIGS_TOTAL.set(config_count as i64);
let is_up = Command::new("systemctl")
.args(["is-active", "nginx"])
.output()
.await
.map(|output| output.status.success())
.unwrap_or(false);
NGINX_UP.set(if is_up { 1 } else { 0 });
let config_valid = Command::new("nginx")
.arg("-t")
.output()
.await
.map(|output| output.status.success())
.unwrap_or(false);
NGINX_CONFIG_VALID.set(if config_valid { 1 } else { 0 });
let (access_size, error_size, req_count, req_bytes, resp_bytes) = collect_nginx_log_metrics();
NGINX_ACCESS_LOG_BYTES_TOTAL.set(access_size as i64);
NGINX_ERROR_LOG_BYTES_TOTAL.set(error_size as i64);
NGINX_REQUESTS_TOTAL_FROM_LOGS.set(req_count as i64);
NGINX_REQUEST_BYTES_TOTAL_FROM_LOGS.set(req_bytes as i64);
NGINX_RESPONSE_BYTES_TOTAL_FROM_LOGS.set(resp_bytes as i64);
}
fn collect_nginx_log_metrics() -> (u64, u64, u64, u64, u64) {
let mut access_log_size = 0_u64;
let mut error_log_size = 0_u64;
let mut request_count = 0_u64;
let mut request_bytes = 0_u64;
let mut response_bytes = 0_u64;
let entries = match fs::read_dir("/var/log/nginx") {
Ok(value) => value,
Err(_) => return (0, 0, 0, 0, 0),
};
for entry in entries.filter_map(|item| item.ok()) {
let path = entry.path();
if !path.is_file() {
continue;
}
let file_name = path
.file_name()
.and_then(|v| v.to_str())
.unwrap_or_default()
.to_ascii_lowercase();
let file_size = fs::metadata(&path).map(|meta| meta.len()).unwrap_or(0);
if file_name.contains("access") {
access_log_size = access_log_size.saturating_add(file_size);
let (count, req_sum, resp_sum) = parse_nginx_access_log(&path);
request_count = request_count.saturating_add(count);
request_bytes = request_bytes.saturating_add(req_sum);
response_bytes = response_bytes.saturating_add(resp_sum);
} else if file_name.contains("error") {
error_log_size = error_log_size.saturating_add(file_size);
}
}
(
access_log_size,
error_log_size,
request_count,
request_bytes,
response_bytes,
)
}
fn parse_nginx_access_log(path: &std::path::Path) -> (u64, u64, u64) {
let file = match File::open(path) {
Ok(value) => value,
Err(_) => return (0, 0, 0),
};
let reader = BufReader::new(file);
let mut request_count = 0_u64;
let mut request_bytes = 0_u64;
let mut response_bytes = 0_u64;
for line in reader.lines().map_while(Result::ok) {
if line.trim().is_empty() {
continue;
}
request_count = request_count.saturating_add(1);
let tokens: Vec<&str> = line.split_whitespace().collect();
if let Some(resp_token) = tokens.get(9) {
if *resp_token != "-" {
if let Ok(value) = resp_token.parse::<u64>() {
response_bytes = response_bytes.saturating_add(value);
}
}
}
if let Some(last) = tokens.last() {
if let Ok(value) = last.parse::<u64>() {
request_bytes = request_bytes.saturating_add(value);
}
}
}
(request_count, request_bytes, response_bytes)
}
fn is_valid_exec_command_name(command: &str) -> bool {
!command.is_empty()
&& command.len() <= 64
&& command
.chars()
.all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '-')
}
fn is_allowed_exec_command(command: &str) -> bool {
matches!(
command,
"ports"
| "services"
| "service"
| "nginx"
| "diag"
| "monitor"
| "logs"
| "list"
| "config"
| "version"
| "tail"
| "start"
| "stop"
| "flush"
| "redeploy"
| "redeploy-v2"
| "snapshot"
| "resurrect"
| "env"
| "curl"
| "docker"
| "network"
| "setup"
| "init"
)
}
fn validate_exec_args(args: &[String]) -> Result<(), String> {
if args.len() > MAX_EXEC_ARGS {
return Err(format!("Too many arguments (max {})", MAX_EXEC_ARGS));
}
for arg in args {
if arg.len() > MAX_EXEC_ARG_LEN {
return Err(format!(
"Argument exceeds max length of {} bytes",
MAX_EXEC_ARG_LEN
));
}
if arg.chars().any(|ch| ch == '\0' || ch.is_control()) {
return Err("Arguments may not contain control characters".to_string());
}
}
Ok(())
}
fn truncate_command_output(output: &[u8]) -> (String, bool) {
if output.len() <= MAX_EXEC_OUTPUT_BYTES {
return (String::from_utf8_lossy(output).to_string(), false);
}
let truncated = String::from_utf8_lossy(&output[..MAX_EXEC_OUTPUT_BYTES]).to_string();
(
format!(
"{}\n...[truncated {} bytes]",
truncated,
output.len() - MAX_EXEC_OUTPUT_BYTES
),
true,
)
}
fn select_target(ring: &RouteRing) -> Option<&RouteTarget> {
if ring.targets.is_empty() {
return None;
}
let idx = ring.cursor.fetch_add(1, Ordering::Relaxed);
Some(&ring.targets[idx % ring.targets.len()])
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn round_robin_targets() {
let ring = RouteRing {
targets: vec![
RouteTarget {
url: "http://a".into(),
weight: 1,
},
RouteTarget {
url: "http://b".into(),
weight: 1,
},
],
conditions: None,
cursor: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
};
let t1 = select_target(&ring).unwrap().url.clone();
let t2 = select_target(&ring).unwrap().url.clone();
let t3 = select_target(&ring).unwrap().url.clone();
assert_eq!(t1, "http://a");
assert_eq!(t2, "http://b");
assert_eq!(t3, "http://a");
}
#[test]
fn exec_command_allowlist_is_enforced() {
assert!(is_allowed_exec_command("nginx"));
assert!(is_allowed_exec_command("service"));
assert!(is_allowed_exec_command("network"));
assert!(!is_allowed_exec_command("api"));
assert!(!is_allowed_exec_command("generate"));
}
#[test]
fn exec_command_name_validation_rejects_invalid_tokens() {
assert!(is_valid_exec_command_name("redeploy-v2"));
assert!(!is_valid_exec_command_name("redeploy v2"));
assert!(!is_valid_exec_command_name("../setup"));
assert!(!is_valid_exec_command_name(""));
}
#[test]
fn exec_arg_validation_enforces_limits_and_chars() {
let too_many = vec!["a".to_string(); MAX_EXEC_ARGS + 1];
assert!(validate_exec_args(&too_many).is_err());
let bad_chars = vec!["line1\nline2".to_string()];
assert!(validate_exec_args(&bad_chars).is_err());
let valid = vec!["--json".to_string(), "service-name".to_string()];
assert!(validate_exec_args(&valid).is_ok());
}
#[test]
fn truncate_command_output_marks_truncation() {
let output = vec![b'a'; MAX_EXEC_OUTPUT_BYTES + 5];
let (text, truncated) = truncate_command_output(&output);
assert!(truncated);
assert!(text.contains("truncated 5 bytes"));
}
}