use std::collections::HashMap;
use kdl::{KdlDocument, KdlNode};
use super::{auth, response};
pub trait BusModule: Send + Sync {
fn name(&self) -> &'static str;
fn handle(&self, module_node: &KdlNode, command_nodes: &[&KdlNode]) -> String;
}
pub struct BusRegistry {
modules: HashMap<&'static str, Box<dyn BusModule>>,
#[cfg(feature = "filter")]
filter_store: crate::filter::FilterStore,
}
impl BusRegistry {
#[must_use]
pub fn from_state(state: &crate::state::AppState) -> Self {
let mut registry = Self {
modules: HashMap::new(),
#[cfg(feature = "filter")]
filter_store: state.filter.clone(),
};
registry.register_modules(state);
registry
}
fn register_modules(&mut self, state: &crate::state::AppState) {
#[cfg(feature = "filter")]
self.register(Box::new(crate::filter::control::FilterModule::new(
state.filter.clone(),
)));
self.register(Box::new(crate::telemetry::control::TelemetryModule));
#[cfg(feature = "image-trust")]
self.register(Box::new(crate::artifacts::control::ImageTrustModule::new(
state.image_trust_attestation.is_some(),
)));
#[cfg(all(feature = "kwt", feature = "filter-control"))]
{
use std::sync::Arc;
let store = state
.kwt_access
.as_ref()
.map(|k| k.jti_store.clone())
.unwrap_or_else(|| Arc::new(crate::kwt_replay::JtiStore::disabled()));
self.register(Box::new(crate::kwt_replay::control::KwtReplayModule::new(
store.clone(),
)));
if let Some(runtime) = &state.aether_runtime {
self.register(Box::new(
crate::kwt_replay::aether_runtime_control::AetherRuntimeModule::new(
Arc::clone(runtime),
),
));
self.register(Box::new(
crate::kwt_replay::aetherdb_control::AetherDbModule::with_runtime(
store.clone(),
Arc::clone(runtime),
),
));
} else {
self.register(Box::new(
crate::kwt_replay::aetherdb_control::AetherDbModule::new(store),
));
}
}
if let Some(tls) = crate::tls::InboundTls::global() {
#[cfg(feature = "filter-control")]
self.register(Box::new(crate::tls::control::TlsModule::new(tls)));
}
#[cfg(all(feature = "kwt", feature = "filter-control"))]
self.register(Box::new(
crate::kwt_replay::wt_aether_control::WebTransportModule,
));
}
pub fn register(&mut self, module: Box<dyn BusModule>) {
self.modules.insert(module.name(), module);
}
fn module_names(&self) -> Vec<&'static str> {
let mut names: Vec<_> = self.modules.keys().copied().collect();
names.sort_unstable();
names
}
pub fn handle(&self, request: &str) -> String {
let trimmed = request.trim();
if trimmed.is_empty() {
return response::err("empty request");
}
if let Some(body) = self.try_legacy_text(trimmed) {
if auth::credentials_configured() {
return response::err(
"legacy bus commands require KDL auth when bus credentials are configured",
);
}
return body;
}
let doc: KdlDocument = match trimmed.parse() {
Ok(doc) => doc,
Err(e) => return response::err(&format!("KDL parse error: {e}")),
};
let nodes: Vec<_> = doc.nodes().iter().collect();
if nodes.is_empty() {
return response::err("empty KDL document");
}
let (auth, head, tail) = match auth::authenticate(&nodes) {
Ok(pair) => pair,
Err(e) => return response::err(&e),
};
match head.name().value() {
"ping" => {
if !auth.permits_global_probe() {
return response::err("bus token not permitted for ping");
}
response::ok("pong")
}
"modules" => {
if !auth.permits_global_probe() {
return response::err("bus token not permitted for modules");
}
response::modules(&self.module_names())
}
"reload" | "status" => {
if !auth.allows_module("filter") {
return response::err("bus token not permitted for module `filter`");
}
self.dispatch_filter_compat(head)
}
name => {
if !auth.allows_module(name) {
return response::err(&format!(
"bus token not permitted for module `{name}`"
));
}
self.dispatch_module(name, head, &tail)
}
}
}
fn dispatch_module(&self, name: &str, module_node: &KdlNode, trailing: &[&KdlNode]) -> String {
match self.modules.get(name) {
Some(module) => module.handle(module_node, trailing),
None => response::err(&format!("unknown module `{name}` (try `modules` for list)")),
}
}
#[cfg(feature = "filter")]
fn dispatch_filter_compat(&self, node: &KdlNode) -> String {
crate::filter::control::handle_node(&self.filter_store, node)
}
#[cfg(not(feature = "filter"))]
fn dispatch_filter_compat(&self, _node: &KdlNode) -> String {
response::err("filter module not enabled")
}
#[cfg(feature = "filter")]
fn try_legacy_text(&self, request: &str) -> Option<String> {
if request.eq_ignore_ascii_case("PING")
|| request.eq_ignore_ascii_case("STATUS")
|| request.starts_with("RELOAD")
{
return Some(crate::filter::control::handle_legacy(
&self.filter_store,
request,
));
}
None
}
#[cfg(not(feature = "filter"))]
fn try_legacy_text(&self, _request: &str) -> Option<String> {
None
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::state::AppState;
fn registry() -> BusRegistry {
BusRegistry::from_state(&AppState::empty())
}
fn registry_open() -> BusRegistry {
registry()
}
#[test]
fn global_ping() {
assert_eq!(registry_open().handle("ping"), "ok \"pong\"");
}
#[test]
fn lists_modules() {
let body = registry_open().handle("modules");
assert!(body.starts_with("modules "));
assert!(body.contains("\"filter\""));
assert!(body.contains("\"telemetry\""));
}
#[test]
fn filter_module_status() {
let body = registry_open().handle("filter {\n status\n}");
assert!(body.starts_with("status deny="));
}
#[test]
fn filter_module_reload() {
let body = registry_open().handle(
r#"filter {
reload {
deny "203.0.113.50"
}
}"#,
);
assert_eq!(body, "ok \"reloaded\"");
}
#[test]
fn legacy_reload_rejected_when_credentials_configured() {
unsafe {
std::env::set_var("NAUTALID_BUS_TOKEN", "configured");
}
let reg = registry();
let out = reg.handle("RELOAD deny=203.0.113.50");
assert!(out.starts_with("err "));
assert!(out.contains("legacy bus commands require KDL auth"));
unsafe {
std::env::remove_var("NAUTALID_BUS_TOKEN");
}
}
#[test]
fn legacy_text_reload() {
let reg = registry_open();
assert_eq!(reg.handle("RELOAD deny=203.0.113.50"), "ok \"reloaded\"");
}
#[test]
fn scoped_filter_env_token_limits_module_dispatch() {
unsafe {
std::env::remove_var("NAUTALID_BUS_TOKEN");
std::env::set_var("NAUTALID_BUS_TOKEN_FILTER", "filter-only");
}
if !auth::credentials_configured() {
unsafe {
std::env::remove_var("NAUTALID_BUS_TOKEN_FILTER");
}
return;
}
let reg = registry();
let denied = reg.handle("ping");
assert!(denied.starts_with("err "));
assert!(
denied.contains("auth wrapper")
|| denied.contains("legacy bus commands require KDL auth")
);
let allowed = reg.handle(
r#"auth token="filter-only" {
filter {
status
}
}"#,
);
assert!(allowed.starts_with("status deny="));
let telemetry_denied = reg.handle(
r#"auth token="filter-only" {
telemetry {
status
}
}"#,
);
assert!(telemetry_denied.contains("not permitted"));
let modules_denied = reg.handle(
r#"auth token="filter-only" {
modules
}"#,
);
assert!(modules_denied.contains("not permitted for modules"));
unsafe {
std::env::remove_var("NAUTALID_BUS_TOKEN_FILTER");
}
}
}