From c1224833d9b6d144afba0c336f0e0b96df4dd3d8 Mon Sep 17 00:00:00 2001 From: hashbk Date: Fri, 16 Oct 2026 23:44:01 +0800 Subject: [PATCH 1/2] feat(management): add dynamic config updates, pending-restart reporting and in-memory logs Promote the management API so configuration changes no longer require the external node agent to restart or rebuild containers. - PUT /v1/config: partial merge against the persisted desired config, validated against the schema; dynamic entries are applied live via a service-registered applier while restart-required entries are reported as pending_restart with current/pending values - GET /v1/config now returns live effective values plus the pending restart diff instead of a startup snapshot - schema version 2 marks every entry with an explicit dynamic flag; hbbr applies downgrade-threshold, downgrade-start-check, limit-speed, total-bandwidth and single-bandwidth live, hbbs applies always-use-relay and relay-servers live through the existing Data::RelayServers0 channel - config persistence follows the bans contract: atomically persist the merged desired state first, activate only after the write succeeds - GET /v1/logs: bounded in-memory ring buffer of recent log lines, captured from the standard log format in both binaries - secrets remain redacted in every API view; resubmitting the placeholder keeps the stored secret unchanged --- Cargo.lock | 1 + Cargo.toml | 3 + docs/management.md | 10 +- management-schema.json | 40 ++- src/common.rs | 14 + src/hbbr.rs | 2 +- src/main.rs | 2 +- src/management.rs | 529 ++++++++++++++++++++++++++++++-- src/relay_server/command.rs | 47 +++ src/relay_server/connection.rs | 3 + src/relay_server/mod.rs | 4 + src/relay_server/server.rs | 1 + src/rendezvous_server/server.rs | 13 + 13 files changed, 624 insertions(+), 45 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3e45003..73fdf29 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1561,6 +1561,7 @@ dependencies = [ "hbb_common", "headers", "http 0.2.12", + "hyper 0.14.32", "ipnetwork", "jsonwebtoken", "lazy_static", diff --git a/Cargo.toml b/Cargo.toml index 399a084..2821178 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -67,6 +67,9 @@ reqwest = { git = "https://github.com/rustdesk-org/reqwest", features = ["blocki hbb_common = { path = "libs/hbb_common" } openssl-sys = { version = "0.9", features = ["vendored"] } +[dev-dependencies] +hyper = "0.14" + [workspace] members = ["libs/hbb_common"] exclude = ["ui"] diff --git a/docs/management.md b/docs/management.md index 9d24d38..00453b3 100644 --- a/docs/management.md +++ b/docs/management.md @@ -2,10 +2,14 @@ The opt-in `/v1` HTTP API is independent of RustDesk's client protocol ports. Set `RD_MANAGEMENT_BIND` to a private socket address, `RD_MANAGEMENT_TOKEN` to an independent random Bearer credential of at least 32 characters, and `RD_MANAGEMENT_DIR` to a persistent directory shared with the node agent. Do not publish this listener to the client-facing network. -hbbs exposes `GET /v1/status`, `GET /v1/config`, `GET /v1/peers` and `GET/PUT /v1/bans`. hbbr additionally exposes `GET /v1/sessions` and `DELETE /v1/sessions/:uuid`; it has no peers endpoint. All endpoints require authorization. API version is 1. Session closure is asynchronous (`202`, `state: closing`). IP bans check both relay endpoints. Reported target IDs are not authenticated source identities. +hbbs exposes `GET /v1/status`, `GET /v1/config`, `PUT /v1/config`, `GET /v1/logs`, `GET /v1/peers` and `GET/PUT /v1/bans`. hbbr additionally exposes `GET /v1/sessions` and `DELETE /v1/sessions/:uuid`; it has no peers endpoint. All endpoints require authorization. API version is 1. Session closure is asynchronous (`202`, `state: closing`). IP bans check both relay endpoints. Reported target IDs are not authenticated source identities. -Files are service-specific: `hbbs-config.json`, `hbbr-config.json`, `hbbs-bans.json`, and `hbbr-bans.json`. Configurations use `{ "values": { "port": "21116" } }`, and policies use `{ "device_ids": [], "ips": [] }`. Saved bans are validated and restored before protocol startup even when `RD_MANAGEMENT_BIND` is empty. Disabling the HTTP listener does not clear admission policy; explicitly replace the bans with empty lists before disabling management if clearing is intended. Ban replacements serialize persistence outside the protocol admission lock and activate only after the write succeeds. +Files are service-specific: `hbbs-config.json`, `hbbr-config.json`, `hbbs-bans.json`, and `hbbr-bans.json`. Configurations use `{ "values": { "port": "21116" } }`, and policies use `{ "device_ids": [], "ips": [] }`. Saved bans are validated and restored before protocol startup even when `RD_MANAGEMENT_BIND` is empty. Disabling the HTTP listener does not clear admission policy; explicitly replace the bans with empty lists before disabling management if clearing is intended. Ban replacements serialize persistence outside the protocol admission lock and activate only after the write succeeds. Config updates follow the same contract: the merged desired configuration is persisted atomically first, and running state changes only after the write succeeds. -Managed settings override startup inputs; they require a restart. Schema and validation are provided by `management-schema.json`. The API returns key placeholders, never the private key. It exposes only `id_ed25519.pub` as the public key. +`PUT /v1/config` accepts a partial `{"values": {...}}` merge against the current desired configuration. Schema entries marked `"dynamic": true` take effect immediately without a restart: hbbr applies `downgrade-threshold`, `downgrade-start-check`, `limit-speed`, `total-bandwidth` and `single-bandwidth` live, and hbbs applies `always-use-relay` and `relay-servers` live. The response reports `applied` entries and `pending_restart` entries (with `current` and `pending` values); pending entries take effect on the next restart, when the saved file is loaded before the protocol starts. `GET /v1/config` returns the live effective values, the schema and the current `pending_restart` diff. Secret entries are stored in the clear in the shared directory but are always reported as `__REDACTED__`; submitting the placeholder keeps the stored value unchanged. A missing dynamic handler falls the entry back to restart-required handling. + +`GET /v1/logs?tail=N` returns up to 2000 recent in-memory log lines (default 200) captured from the standard log format; it requires no container log access. + +Managed settings override startup inputs. Schema and validation are provided by `management-schema.json`. The API returns key placeholders, never the private key. It exposes only `id_ed25519.pub` as the public key. `docker-managed/Dockerfile` builds the managed hbbs/hbbr image in CI. The Console repository supplies the node agent, combined Compose deployment, permission enforcement, UI, integration workflow and detailed lifecycle documentation in `docs/server-management.md`. Existing binaries remain compatible with deployments that do not enable management. diff --git a/management-schema.json b/management-schema.json index 4525c6a..f96c431 100644 --- a/management-schema.json +++ b/management-schema.json @@ -1,11 +1,12 @@ { - "version": 1, + "version": 2, "hbbs": [ { "name": "port", "default": "21116", "kind": "integer", "secret": false, + "dynamic": false, "restart_required": true, "minimum": 3, "maximum": 65533 @@ -15,6 +16,7 @@ "default": "", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true }, { @@ -22,13 +24,15 @@ "default": "", "kind": "string", "secret": false, - "restart_required": true + "dynamic": true, + "restart_required": false }, { "name": "key", "default": "-", "kind": "string", "secret": true, + "dynamic": false, "restart_required": true }, { @@ -36,6 +40,7 @@ "default": "0", "kind": "integer", "secret": false, + "dynamic": false, "restart_required": true, "minimum": 0, "maximum": 2147483647 @@ -45,6 +50,7 @@ "default": "0", "kind": "integer", "secret": false, + "dynamic": false, "restart_required": true, "minimum": 0, "maximum": 2147483647 @@ -54,6 +60,7 @@ "default": "", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true }, { @@ -61,6 +68,7 @@ "default": "", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true }, { @@ -68,6 +76,7 @@ "default": "", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true }, { @@ -75,6 +84,7 @@ "default": "", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true }, { @@ -82,6 +92,7 @@ "default": "./db_v2.sqlite3", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true }, { @@ -89,6 +100,7 @@ "default": "1", "kind": "integer", "secret": false, + "dynamic": false, "restart_required": true, "minimum": 1, "maximum": 1024 @@ -98,6 +110,7 @@ "default": "", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true }, { @@ -105,13 +118,15 @@ "default": "N", "kind": "relay-mode", "secret": false, - "restart_required": true + "dynamic": true, + "restart_required": false }, { "name": "log", "default": "info", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true } ], @@ -121,6 +136,7 @@ "default": "21117", "kind": "integer", "secret": false, + "dynamic": false, "restart_required": true, "minimum": 1, "maximum": 65533 @@ -130,6 +146,7 @@ "default": "", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true }, { @@ -137,6 +154,7 @@ "default": "", "kind": "string", "secret": true, + "dynamic": false, "restart_required": true }, { @@ -144,7 +162,8 @@ "default": "0.66", "kind": "number", "secret": false, - "restart_required": true, + "dynamic": true, + "restart_required": false, "minimum": 0.01, "maximum": 1 }, @@ -153,7 +172,8 @@ "default": "1800", "kind": "integer", "secret": false, - "restart_required": true, + "dynamic": true, + "restart_required": false, "minimum": 1, "maximum": 4294967 }, @@ -162,7 +182,8 @@ "default": "32", "kind": "number", "secret": false, - "restart_required": true, + "dynamic": true, + "restart_required": false, "minimum": 0.01, "maximum": 1000000 }, @@ -171,7 +192,8 @@ "default": "1024", "kind": "number", "secret": false, - "restart_required": true, + "dynamic": true, + "restart_required": false, "minimum": 0.01, "maximum": 1000000 }, @@ -180,7 +202,8 @@ "default": "128", "kind": "number", "secret": false, - "restart_required": true, + "dynamic": true, + "restart_required": false, "minimum": 0.01, "maximum": 1000000 }, @@ -189,6 +212,7 @@ "default": "info", "kind": "string", "secret": false, + "dynamic": false, "restart_required": true } ] diff --git a/src/common.rs b/src/common.rs index 5fa6161..2398804 100644 --- a/src/common.rs +++ b/src/common.rs @@ -11,6 +11,20 @@ use std::{ time::{Instant, SystemTime}, }; +// Mirrors the standard opt_format output while retaining a bounded in-memory +// copy so the management API can serve recent log lines without any container +// log access. +pub fn log_format( + w: &mut dyn std::io::Write, + now: &mut flexi_logger::DeferredNow, + record: &log::Record, +) -> Result<(), std::io::Error> { + let mut buffer = Vec::new(); + flexi_logger::opt_format(&mut buffer, now, record)?; + crate::management::push_log_line(&String::from_utf8_lossy(&buffer)); + w.write_all(&buffer) +} + pub fn parse_bind_address(value: &str) -> Result> { let value = value.trim(); if value.is_empty() { diff --git a/src/hbbr.rs b/src/hbbr.rs index b40fafa..7f6d3e0 100644 --- a/src/hbbr.rs +++ b/src/hbbr.rs @@ -13,7 +13,7 @@ fn main() -> ResultType<()> { managed.apply(); let _logger = Logger::try_with_env_or_str("info")? .log_to_stdout() - .format(opt_format) + .format(common::log_format) .write_mode(WriteMode::Async) .start()?; let args = format!( diff --git a/src/main.rs b/src/main.rs index e606a21..36d2899 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,7 +13,7 @@ fn main() -> ResultType<()> { managed.apply(); let _logger = Logger::try_with_env_or_str("info")? .log_to_stdout() - .format(opt_format) + .format(log_format) .write_mode(WriteMode::Async) .start()?; let args = format!( diff --git a/src/management.rs b/src/management.rs index 306ae28..def3bfc 100644 --- a/src/management.rs +++ b/src/management.rs @@ -1,7 +1,7 @@ //! Private, opt-in management API. Client protocol ports are never used for HTTP. use axum::{ body::Body, - extract::{Extension, Path}, + extract::{Extension, Path, Query}, http::{Request, StatusCode}, middleware::{self, Next}, response::{IntoResponse, Response}, @@ -11,9 +11,9 @@ use axum::{ use hbb_common::{anyhow::anyhow, log, tokio, ResultType}; use once_cell::sync::Lazy; use serde_derive::{Deserialize, Serialize}; -use serde_json::{json, Value}; +use serde_json::{json, Map, Value}; use std::{ - collections::{BTreeMap, HashMap}, + collections::{BTreeMap, HashMap, VecDeque}, net::{IpAddr, SocketAddr}, path::PathBuf, sync::RwLock, @@ -26,6 +26,39 @@ static PEERS: Lazy>> = Lazy::new(|| RwLock::new(Vec::new())); static SESSIONS: Lazy>> = Lazy::new(|| RwLock::new(HashMap::new())); static BANS: Lazy> = Lazy::new(|| RwLock::new(Bans::default())); static BAN_UPDATES: Lazy> = Lazy::new(|| Mutex::new(())); +static LOGS: Lazy>> = Lazy::new(|| RwLock::new(VecDeque::new())); + +// Runtime view of the managed configuration. `values` holds the effective +// running value of every schema entry (secrets stay REDACTED), `overrides` +// holds the persisted desired values as submitted, and `pending` holds the +// restart-required entries that differ from the running values. +#[derive(Clone, Default)] +struct ManagedState { + values: BTreeMap, + overrides: BTreeMap, + pending: BTreeMap, +} + +static MANAGED: Lazy> = Lazy::new(|| RwLock::new(ManagedState::default())); +static CONFIG_UPDATES: Lazy> = Lazy::new(|| Mutex::new(())); + +// Applied by the owning binary for entries marked as dynamic in the schema. +// Returning false falls the entry back to restart-required handling. +pub type DynamicApplier = Box bool + Send + Sync>; +static DYNAMIC_APPLY: Lazy>> = Lazy::new(|| RwLock::new(None)); + +pub fn set_dynamic_apply(applier: DynamicApplier) { + *DYNAMIC_APPLY.write().unwrap() = Some(applier); +} + +fn apply_dynamic(name: &str, value: &str) -> bool { + DYNAMIC_APPLY + .read() + .unwrap() + .as_ref() + .map(|applier| applier(name, value)) + .unwrap_or(false) +} #[derive(Clone, Default, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] @@ -161,6 +194,33 @@ pub fn schema(service: &str) -> Vec { schema[service].as_array().unwrap().clone() } +fn validate_value(definition: &Value, name: &str, value: &str) -> ResultType<()> { + if value.len() > 4096 || value.contains('\0') || value == REDACTED { + return Err(anyhow!("Invalid value for {name}")); + } + if definition["kind"] == "integer" || definition["kind"] == "number" { + if definition["kind"] == "integer" { + // Startup consumers parse integers rather than floating point strings. + value.parse::()?; + } + let number: f64 = value.parse()?; + if !number.is_finite() + || number < definition["minimum"].as_f64().unwrap() + || number > definition["maximum"].as_f64().unwrap() + || (definition["kind"] == "integer" && number.fract() != 0.0) + { + return Err(anyhow!("Out of range: {name}")); + } + } + if name == "bind" { + crate::common::parse_bind_address(value)?; + } + if name == "always-use-relay" && value != "Y" && value != "N" { + return Err(anyhow!("Expected Y or N")); + } + Ok(()) +} + pub fn load_config(service: &str) -> ResultType { let path = data_path(service, "config"); let config: ManagedConfig = match std::fs::read(&path) { @@ -174,29 +234,7 @@ pub fn load_config(service: &str) -> ResultType { .iter() .find(|d| d["name"].as_str() == Some(name.as_str())) .ok_or_else(|| anyhow!("Unknown managed configuration: {name}"))?; - if value.len() > 4096 || value.contains('\0') || value == REDACTED { - return Err(anyhow!("Invalid value for {name}")); - } - if definition["kind"] == "integer" || definition["kind"] == "number" { - if definition["kind"] == "integer" { - // Startup consumers parse integers rather than floating point strings. - value.parse::()?; - } - let number: f64 = value.parse()?; - if !number.is_finite() - || number < definition["minimum"].as_f64().unwrap() - || number > definition["maximum"].as_f64().unwrap() - || (definition["kind"] == "integer" && number.fract() != 0.0) - { - return Err(anyhow!("Out of range: {name}")); - } - } - if name == "bind" { - crate::common::parse_bind_address(value)?; - } - if name == "always-use-relay" && value != "Y" && value != "N" { - return Err(anyhow!("Expected Y or N")); - } + validate_value(definition, name, value)?; } Ok(config) } @@ -218,7 +256,7 @@ struct ApiState { service: &'static str, token: String, started: Instant, - values: BTreeMap, + } async fn authorize(request: Request, next: Next) -> Response { @@ -253,7 +291,193 @@ async fn status(Extension(state): Extension) -> Json { } async fn config(Extension(state): Extension) -> Json { - Json(json!({"schema":schema(state.service),"values":state.values})) + let managed = MANAGED.read().unwrap(); + let mut pending = Map::new(); + for (name, value) in &managed.pending { + pending.insert( + name.clone(), + json!({ + "current": managed.values.get(name).cloned().unwrap_or_default(), + "pending": value.clone(), + }), + ); + } + Json( + json!({"schema":schema(state.service),"values":managed.values.clone(), + "pending_restart":pending}), + ) +} + +#[derive(Clone, Default, Deserialize)] +#[serde(deny_unknown_fields)] +struct ConfigUpdate { + values: BTreeMap, +} + +fn definition_of(definitions: &[Value], name: &str) -> Option { + definitions + .iter() + .find(|d| d["name"].as_str() == Some(name)) + .cloned() +} + +fn bad_request(message: String) -> Response { + ( + StatusCode::BAD_REQUEST, + Json(json!({"error": message})), + ) + .into_response() +} + +async fn replace_config( + Extension(state): Extension, + Json(update): Json, +) -> Response { + // Validate the submission against the schema before touching any state. + let definitions = schema(state.service); + for (name, value) in &update.values { + let Some(definition) = definition_of(&definitions, name) else { + return bad_request(format!("Unknown managed configuration: {name}")); + }; + if definition["secret"] == true && value == REDACTED { + continue; + } + if let Err(err) = validate_value(&definition, name, value) { + return bad_request(err.to_string()); + } + } + let service = state.service; + // The task owns the update through persistence and activation even if the + // HTTP caller disconnects while the blocking filesystem work is in flight. + let result = tokio::spawn(async move { + let _update = CONFIG_UPDATES.lock().await; + let (mut values, mut overrides, mut pending) = { + let managed = MANAGED.read().unwrap(); + ( + managed.values.clone(), + managed.overrides.clone(), + managed.pending.clone(), + ) + }; + for (name, value) in &update.values { + let Some(definition) = definition_of(&definitions, name) else { + continue; + }; + if definition["secret"] == true && value == REDACTED { + continue; + } + overrides.insert(name.clone(), value.clone()); + } + // Persist before activation; a write failure keeps the running state. + let persisted = overrides.clone(); + tokio::task::spawn_blocking(move || persist_config(service, &persisted)).await??; + let mut applied = Map::new(); + let mut pending_report = Map::new(); + for (name, value) in &overrides { + let Some(definition) = definition_of(&definitions, name) else { + continue; + }; + let secret = definition["secret"] == true; + // Secrets cannot be compared against the running value because the + // effective view is redacted; only resubmitted secrets are reported. + if secret && !update.values.contains_key(name) { + continue; + } + let display = if secret { + REDACTED.to_owned() + } else { + value.clone() + }; + if !secret && values.get(name) == Some(&display) { + pending.remove(name); + continue; + } + if definition["dynamic"] == true && apply_dynamic(name, value) { + values.insert(name.clone(), display.clone()); + pending.remove(name); + applied.insert(name.clone(), json!(display)); + } else { + let current = values.get(name).cloned().unwrap_or_default(); + pending.insert(name.clone(), display.clone()); + pending_report.insert( + name.clone(), + json!({"current": current, "pending": display}), + ); + } + } + *MANAGED.write().unwrap() = ManagedState { + values, + overrides, + pending, + }; + Ok::<_, hbb_common::anyhow::Error>((applied, pending_report)) + }) + .await; + match result { + Ok(Ok((applied, pending_report))) => Json( + json!({"state":"saved","applied":applied,"pending_restart":pending_report}), + ) + .into_response(), + Ok(Err(err)) => { + log::error!("Unable to persist managed configuration: {err}"); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + } + Err(join) => { + log::error!("Managed configuration update task failed: {join}"); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + } + } +} + +fn persist_config(service: &str, overrides: &BTreeMap) -> ResultType<()> { + let path = data_path(service, "config"); + std::fs::create_dir_all(path.parent().unwrap())?; + let temporary = path.with_extension("tmp"); + let payload = serde_json::to_vec(&ManagedConfig { + values: overrides.clone(), + })?; + std::fs::write(&temporary, payload)?; + std::fs::rename(temporary, path)?; + Ok(()) +} + +pub const LOG_CAPACITY: usize = 2000; +const LOG_LINE_LIMIT: usize = 2048; + +// Called from the logging format hook in both binaries so management clients +// can read recent in-memory log lines without container log access. +pub fn push_log_line(line: &str) { + let mut owned; + let line = if line.len() > LOG_LINE_LIMIT { + let mut end = LOG_LINE_LIMIT; + while end > 0 && !line.is_char_boundary(end) { + end -= 1; + } + owned = String::from(&line[..end]); + owned.push_str("..."); + &owned + } else { + line + }; + let mut logs = LOGS.write().unwrap(); + if logs.len() >= LOG_CAPACITY { + logs.pop_front(); + } + logs.push_back(line.to_owned()); +} + +#[derive(Clone, Copy, Deserialize)] +struct LogsQuery { + #[serde(default)] + tail: Option, +} + +async fn logs(Query(query): Query) -> Response { + let tail = query.tail.unwrap_or(200).min(LOG_CAPACITY); + let logs = LOGS.read().unwrap(); + let skipped = logs.len().saturating_sub(tail); + let data: Vec = logs.iter().skip(skipped).cloned().collect(); + Json(json!({"data": data})).into_response() } async fn peers(Extension(state): Extension) -> Response { @@ -381,6 +605,7 @@ pub fn start(service: &'static str, port: u16) -> ResultType<()> { "RD_MANAGEMENT_TOKEN must contain at least 32 characters" )); } + let overrides = load_config(service)?.values; let mut values = BTreeMap::new(); for definition in schema(service) { let name = definition["name"].as_str().unwrap(); @@ -400,11 +625,17 @@ pub fn start(service: &'static str, port: u16) -> ResultType<()> { }, ); } + // Restart-required overrides took effect during startup, so nothing is + // pending right after boot; dynamic entries were applied by the owner too. + *MANAGED.write().unwrap() = ManagedState { + values, + overrides, + pending: BTreeMap::new(), + }; let state = ApiState { service, token, started: Instant::now(), - values, }; let router = router(state); let listener = std::net::TcpListener::bind(&bind)?; @@ -422,7 +653,8 @@ pub fn start(service: &'static str, port: u16) -> ResultType<()> { fn router(state: ApiState) -> Router { Router::new() .route("/v1/status", get(status)) - .route("/v1/config", get(config)) + .route("/v1/config", get(config).put(replace_config)) + .route("/v1/logs", get(logs)) .route("/v1/peers", get(peers)) .route("/v1/sessions", get(sessions)) .route("/v1/sessions/:uuid", axum::routing::delete(disconnect)) @@ -491,7 +723,7 @@ mod tests { service: "hbbs", token: "test".into(), started: Instant::now(), - values: BTreeMap::new(), + }; let a = Bans { device_ids: vec!["policy-a".into()], @@ -559,7 +791,7 @@ mod tests { service: "hbbs", token: token.into(), started: Instant::now(), - values: BTreeMap::new(), + }); let task = tokio::spawn( axum::Server::from_tcp(listener) @@ -632,4 +864,237 @@ mod tests { assert!(rx.try_recv().is_ok()); close_session("test-session"); } + + struct ConfigFixture { + directory: PathBuf, + previous_directory: Option, + previous_managed: ManagedState, + previous_applier: Option, + } + + impl ConfigFixture { + fn new() -> Self { + let directory = std::env::temp_dir().join(format!( + "rustdesk-config-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&directory).unwrap(); + let fixture = Self { + directory, + previous_directory: std::env::var("RD_MANAGEMENT_DIR").ok(), + previous_managed: MANAGED.read().unwrap().clone(), + previous_applier: DYNAMIC_APPLY.write().unwrap().take(), + }; + std::env::set_var("RD_MANAGEMENT_DIR", &fixture.directory); + std::env::remove_var("RD_MANAGEMENT_BIND"); + *MANAGED.write().unwrap() = ManagedState::default(); + fixture + } + + fn state(service: &'static str) -> ApiState { + ApiState { + service, + token: "test".into(), + started: Instant::now(), + } + } + } + + impl Drop for ConfigFixture { + fn drop(&mut self) { + *MANAGED.write().unwrap() = self.previous_managed.clone(); + *DYNAMIC_APPLY.write().unwrap() = self.previous_applier.take(); + match &self.previous_directory { + Some(value) => std::env::set_var("RD_MANAGEMENT_DIR", value), + None => std::env::remove_var("RD_MANAGEMENT_DIR"), + } + let _ = std::fs::remove_dir_all(&self.directory); + } + } + + fn managed_values(entries: &[(&str, &str)]) -> BTreeMap { + entries + .iter() + .map(|(name, value)| ((*name).to_owned(), (*value).to_owned())) + .collect() + } + + #[tokio::test] + async fn put_config_applies_dynamic_entries_and_reports_pending_restart() { + let _test = POLICY_TEST.lock().unwrap(); + let _fixture = ConfigFixture::new(); + *MANAGED.write().unwrap() = ManagedState { + values: managed_values(&[("limit-speed", "32"), ("port", "21117")]), + overrides: BTreeMap::new(), + pending: BTreeMap::new(), + }; + set_dynamic_apply(Box::new(|name, _| { + matches!(name, "limit-speed" | "single-bandwidth") + })); + let response = replace_config( + Extension(ConfigFixture::state("hbbr")), + Json(ConfigUpdate { + values: managed_values(&[("limit-speed", "64"), ("port", "21118")]), + }), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let body: Value = + serde_json::from_slice(&hyper::body::to_bytes(response.into_body()).await.unwrap()) + .unwrap(); + assert_eq!(body["applied"]["limit-speed"], "64"); + assert_eq!(body["pending_restart"]["port"]["current"], "21117"); + assert_eq!(body["pending_restart"]["port"]["pending"], "21118"); + let managed = MANAGED.read().unwrap(); + assert_eq!(managed.values.get("limit-speed").unwrap(), "64"); + assert_eq!(managed.values.get("port").unwrap(), "21117"); + assert_eq!(managed.pending.get("port").unwrap(), "21118"); + let persisted: ManagedConfig = serde_json::from_slice( + &std::fs::read(data_path("hbbr", "config")).unwrap(), + ) + .unwrap(); + assert_eq!(persisted.values, managed_values(&[("limit-speed", "64"), ("port", "21118")])); + // The live config view reflects the dynamic change and the pending entry. + let view = config(Extension(ConfigFixture::state("hbbr"))).await; + assert_eq!(view["values"]["limit-speed"], "64"); + assert_eq!(view["values"]["port"], "21117"); + assert_eq!(view["pending_restart"]["port"]["pending"], "21118"); + } + + #[tokio::test] + async fn put_config_write_failure_does_not_activate() { + let _test = POLICY_TEST.lock().unwrap(); + let fixture = ConfigFixture::new(); + *MANAGED.write().unwrap() = ManagedState { + values: managed_values(&[("limit-speed", "32")]), + overrides: BTreeMap::new(), + pending: BTreeMap::new(), + }; + set_dynamic_apply(Box::new(|name, _| name == "limit-speed")); + // Force the rename to fail; the running values must remain intact. + std::fs::create_dir_all(data_path("hbbr", "config").parent().unwrap()).unwrap(); + std::fs::create_dir(data_path("hbbr", "config")).unwrap(); + let response = replace_config( + Extension(ConfigFixture::state("hbbr")), + Json(ConfigUpdate { + values: managed_values(&[("limit-speed", "64")]), + }), + ) + .await; + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + MANAGED.read().unwrap().values.get("limit-speed").unwrap(), + "32" + ); + assert!(MANAGED.read().unwrap().pending.is_empty()); + drop(fixture); + } + + #[tokio::test] + async fn put_config_rejects_unknown_and_invalid_values() { + let _test = POLICY_TEST.lock().unwrap(); + let _fixture = ConfigFixture::new(); + let state = ConfigFixture::state("hbbr"); + let unknown = replace_config( + Extension(state.clone()), + Json(ConfigUpdate { + values: managed_values(&[("unknown-entry", "1")]), + }), + ) + .await; + assert_eq!(unknown.status(), StatusCode::BAD_REQUEST); + let out_of_range = replace_config( + Extension(state.clone()), + Json(ConfigUpdate { + values: managed_values(&[("limit-speed", "0")]), + }), + ) + .await; + assert_eq!(out_of_range.status(), StatusCode::BAD_REQUEST); + let malformed_bind = replace_config( + Extension(state), + Json(ConfigUpdate { + values: managed_values(&[("bind", "not-an-ip")]), + }), + ) + .await; + assert_eq!(malformed_bind.status(), StatusCode::BAD_REQUEST); + } + + #[tokio::test] + async fn put_config_keeps_secret_values_and_redacts_reports() { + let _test = POLICY_TEST.lock().unwrap(); + let _fixture = ConfigFixture::new(); + *MANAGED.write().unwrap() = ManagedState { + values: managed_values(&[("key", REDACTED), ("port", "21117")]), + overrides: BTreeMap::new(), + pending: BTreeMap::new(), + }; + let response = replace_config( + Extension(ConfigFixture::state("hbbr")), + Json(ConfigUpdate { + values: managed_values(&[("key", "replacement-secret")]), + }), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let body: Value = + serde_json::from_slice(&hyper::body::to_bytes(response.into_body()).await.unwrap()) + .unwrap(); + assert_eq!(body["pending_restart"]["key"]["pending"], REDACTED); + assert!(!format!("{body:?}").contains("replacement-secret")); + let persisted: ManagedConfig = serde_json::from_slice( + &std::fs::read(data_path("hbbr", "config")).unwrap(), + ) + .unwrap(); + assert_eq!( + persisted.values.get("key").unwrap(), + "replacement-secret" + ); + // Resubmitting the placeholder keeps the stored secret untouched. + let response = replace_config( + Extension(ConfigFixture::state("hbbr")), + Json(ConfigUpdate { + values: managed_values(&[("key", REDACTED), ("port", "21117")]), + }), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let persisted: ManagedConfig = serde_json::from_slice( + &std::fs::read(data_path("hbbr", "config")).unwrap(), + ) + .unwrap(); + assert_eq!( + persisted.values.get("key").unwrap(), + "replacement-secret" + ); + } + + #[tokio::test] + async fn log_ring_buffer_captures_tails_and_truncates() { + for _ in 0..8 { + push_log_line("short line"); + } + let long: String = "x".repeat(LOG_LINE_LIMIT + 100); + push_log_line(&long); + let response = logs(Query(LogsQuery { tail: Some(3) })).await; + assert_eq!(response.status(), StatusCode::OK); + let body: Value = + serde_json::from_slice(&hyper::body::to_bytes(response.into_body()).await.unwrap()) + .unwrap(); + let data = body["data"].as_array().unwrap(); + assert_eq!(data.len(), 3); + assert_eq!(data[0], "short line"); + assert_eq!(data[1], "short line"); + let truncated = data[2].as_str().unwrap(); + assert!(truncated.starts_with('x')); + assert!(truncated.ends_with("...")); + assert!(truncated.len() <= LOG_LINE_LIMIT + 3); + // Reset the ring for other tests. + LOGS.write().unwrap().clear(); + } } diff --git a/src/relay_server/command.rs b/src/relay_server/command.rs index 5f93581..424dedb 100644 --- a/src/relay_server/command.rs +++ b/src/relay_server/command.rs @@ -3,6 +3,53 @@ use async_speed_limit::Limiter; use hbb_common::log; use std::sync::atomic::Ordering; +// Applies schema entries marked as dynamic for the relay server. Values use +// schema units and are converted exactly like the startup parameters. +pub(crate) fn apply_dynamic(name: &str, value: &str) -> bool { + match name { + "downgrade-threshold" => match value.parse::() { + Ok(v) if v > 0. => { + DOWNGRADE_THRESHOLD_100.store((v * 100.) as _, Ordering::SeqCst); + true + } + _ => false, + }, + "downgrade-start-check" => match value.parse::() { + Ok(v) if v > 0 => { + DOWNGRADE_START_CHECK.store(v * 1000, Ordering::SeqCst); + true + } + _ => false, + }, + "limit-speed" => match value.parse::() { + Ok(v) if v > 0. => { + LIMIT_SPEED.store((v * 1024. * 1024.) as _, Ordering::SeqCst); + true + } + _ => false, + }, + "total-bandwidth" => match value.parse::() { + Ok(v) if v > 0. => { + let bits = (v * 1024. * 1024.) as usize; + TOTAL_BANDWIDTH.store(bits, Ordering::SeqCst); + if let Some(limiter) = &*TOTAL_LIMITER.read().unwrap() { + limiter.set_speed_limit(bits as _); + } + true + } + _ => false, + }, + "single-bandwidth" => match value.parse::() { + Ok(v) if v > 0. => { + SINGLE_BANDWIDTH.store((v * 1024. * 1024.) as _, Ordering::SeqCst); + true + } + _ => false, + }, + _ => false, + } +} + pub(crate) fn check_params() { let tmp = crate::common::get_arg("DOWNGRADE_THRESHOLD") .parse::() diff --git a/src/relay_server/connection.rs b/src/relay_server/connection.rs index 034af35..8be0abe 100644 --- a/src/relay_server/connection.rs +++ b/src/relay_server/connection.rs @@ -27,6 +27,9 @@ pub(crate) async fn io_loop( ) { command::check_params(); let limiter = ::new(TOTAL_BANDWIDTH.load(Ordering::SeqCst) as _); + // Publish the live limiter so dynamic total-bandwidth updates reach the + // running accept loop (clones share the same underlying budget). + *TOTAL_LIMITER.write().unwrap() = Some(limiter.clone()); loop { tokio::select! { res = listener.accept() => { diff --git a/src/relay_server/mod.rs b/src/relay_server/mod.rs index a853e0e..8d203b2 100644 --- a/src/relay_server/mod.rs +++ b/src/relay_server/mod.rs @@ -4,6 +4,7 @@ mod helper; mod server; mod stream; +use async_speed_limit::Limiter; use hbb_common::tokio::sync::{Mutex, RwLock}; use std::{ collections::{HashMap, HashSet}, @@ -19,6 +20,9 @@ lazy_static::lazy_static! { pub(crate) static ref USAGE: RwLock> = Default::default(); pub(crate) static ref BLACKLIST: RwLock> = Default::default(); pub(crate) static ref BLOCKLIST: RwLock> = Default::default(); + // The active shared limiter for TOTAL_BANDWIDTH, refreshed per accept loop + // so management updates can adjust the live limiter without restarts. + pub(crate) static ref TOTAL_LIMITER: std::sync::RwLock> = Default::default(); } pub(crate) static DOWNGRADE_THRESHOLD_100: AtomicUsize = AtomicUsize::new(66); // 0.66 diff --git a/src/relay_server/server.rs b/src/relay_server/server.rs index 6231f45..8aad251 100644 --- a/src/relay_server/server.rs +++ b/src/relay_server/server.rs @@ -51,6 +51,7 @@ pub async fn start_with_bind(bind_addr: Option, port: &str, key: &str) - let console = crate::common::listen_console(bind_addr, port).await?; if !management_started { crate::management::start("hbbr", port)?; + crate::management::set_dynamic_apply(Box::new(command::apply_dynamic)); management_started = true; } connection::io_loop(listener, websocket, console, &key).await; diff --git a/src/rendezvous_server/server.rs b/src/rendezvous_server/server.rs index 83e47dd..5b73fb6 100644 --- a/src/rendezvous_server/server.rs +++ b/src/rendezvous_server/server.rs @@ -91,6 +91,19 @@ impl RendezvousServer { log::info!("local-ip: {:?}", rs.inner.local_ip); std::env::set_var("PORT_FOR_API", port.to_string()); rs.parse_relay_servers(&get_arg("relay-servers")); + // Register the hbbs handlers for schema entries marked as dynamic. + { + let tx = rs.tx.clone(); + crate::management::set_dynamic_apply(Box::new(move |name, value| match name { + "always-use-relay" => { + let enabled = value.eq_ignore_ascii_case("Y"); + ALWAYS_USE_RELAY.store(enabled, std::sync::atomic::Ordering::SeqCst); + true + } + "relay-servers" => tx.send(Data::RelayServers0(value.to_owned())).is_ok(), + _ => false, + })); + } let mut listener = helper::create_tcp_listener(bind_addr, port).await?; let mut listener2 = helper::create_tcp_listener(bind_addr, nat_port).await?; let mut listener3 = helper::create_tcp_listener(bind_addr, ws_port).await?; From 515618cff877ec6478a0aa7de025a38a3e191833 Mon Sep 17 00:00:00 2001 From: hashbk Date: Sat, 17 Oct 2026 08:36:50 +0800 Subject: [PATCH 2/2] fix(hbbr): register the dynamic config applier before the listener starts A PUT /v1/config request arriving between management::start and set_dynamic_apply persisted dynamic entries but reported them as pending_restart instead of applying them immediately. Register the applier first, matching the hbbs startup order. --- src/relay_server/server.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/relay_server/server.rs b/src/relay_server/server.rs index 8aad251..702f14a 100644 --- a/src/relay_server/server.rs +++ b/src/relay_server/server.rs @@ -50,8 +50,11 @@ pub async fn start_with_bind(bind_addr: Option, port: &str, key: &str) - let websocket = crate::common::listen_tcp(bind_addr, port2).await?; let console = crate::common::listen_console(bind_addr, port).await?; if !management_started { - crate::management::start("hbbr", port)?; + // Register the dynamic applier before the listener starts so + // early config updates apply immediately instead of being + // reported as pending restart. crate::management::set_dynamic_apply(Box::new(command::apply_dynamic)); + crate::management::start("hbbr", port)?; management_started = true; } connection::io_loop(listener, websocket, console, &key).await;