refactor(maestro): drop allow_inbound policy gate, keep JWT session binding (NAP-339)

The peer (T2) attach path had a per-pair access-control layer
(`[daemon.policy].allow_inbound`): a target-side allow-list of daemon ids
permitted to broker-attach, enforced inside JWT verification, hot-reloadable via
SIGHUP / `POST /api/policy/reload`, with a `maestro policy` CLI. On a flat-trust
perimeter (the operator's VPN/firewall is the boundary for which hosts connect)
that granularity is unused, and deny-by-default made it config friction —
cross-daemon attach silently failed until every peer was listed.

Remove the allow-list; keep the registry-minted pair-attach JWT and ALL its
binding checks (iss=registry, aud=target, sid=session, exp). The firewall gates
host->port reachability; it does not stop a reachable host from attaching to a
session it wasn't handed or replaying a token elsewhere — that's the JWT's job
and it stays. Only the `sub in allow_inbound` check is dropped.

Removed: DaemonPolicySection / [daemon.policy].allow_inbound (config + daemon),
policy.rs (SIGHUP listener + GET/POST /api/policy), the `maestro policy` CLI,
PairAttachVerifier's allow-list field/reload/snapshot + JwtError::NotAllowed,
and the e2e broker_rejects_when_target_doesnt_allow_inbound test. The NAP-265
pre-upgrade-502 path stays covered by broker_attach_returns_502_on_perimeter_mismatch.

Also dropped the unit's ExecReload=kill -HUP: with no SIGHUP handler left,
default SIGHUP disposition is terminate, so wiring `systemctl reload` to it would
kill the daemon. Re-introducing a per-pair gate later is a verify-time change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-19 07:07:55 +00:00
co-authored by Claude Opus 4.8
parent 02475191df
commit b6833ede43
20 changed files with 60 additions and 722 deletions
+2 -1
View File
@@ -84,7 +84,8 @@ pub fn attach_url(info: &ConnectInfo, session_id: Uuid) -> Result<String, Attach
/// The broker endpoint (`WS /api/attach/:id`, NAP-243) lives on the daemon's
/// `bind_addr`, so we take the HTTP base (`http://127.0.0.1:7070`), swap the
/// scheme to its WS equivalent, and append the route. The daemon picks the
/// transport and enforces `allow_inbound` on the target side; the CLI just
/// transport and dials the target (which verifies the registry-minted
/// pair-attach JWT — session binding, not a per-peer allow-list); the CLI just
/// pumps frames.
#[allow(clippy::result_large_err)]
pub fn broker_attach_url(
-32
View File
@@ -42,23 +42,6 @@ pub enum Command {
Kill(KillArgs),
/// Set the default registry URL persisted in the CLI config.
SetRegistry(SetRegistryArgs),
/// Inspect or hot-reload the local daemon's `[daemon.policy]` (NAP-237).
#[command(subcommand)]
Policy(PolicyCmd),
}
#[derive(Debug, Subcommand)]
pub enum PolicyCmd {
/// Re-read `~/.config/maestro/config.toml` and apply
/// `[daemon.policy].allow_inbound` without restarting the daemon. Affects
/// only new attach attempts — in-flight sessions keep their connection.
Reload,
/// Print the daemon's current effective `allow_inbound` list.
Show {
/// Output as JSON instead of one entry per line.
#[arg(long)]
json: bool,
},
}
#[derive(Debug, clap::Args)]
@@ -212,21 +195,6 @@ mod tests {
}
}
#[test]
fn parses_policy_reload() {
let cli = Cli::try_parse_from(["maestro", "policy", "reload"]).unwrap();
assert!(matches!(cli.command, Command::Policy(PolicyCmd::Reload)));
}
#[test]
fn parses_policy_show_json() {
let cli = Cli::try_parse_from(["maestro", "policy", "show", "--json"]).unwrap();
match cli.command {
Command::Policy(PolicyCmd::Show { json }) => assert!(json),
other => panic!("wrong subcommand: {other:?}"),
}
}
#[test]
fn rejects_env_without_equals() {
let err = Cli::try_parse_from([
+3 -16
View File
@@ -11,7 +11,8 @@
//! * `list` reads the daemon's local catalog mirror (own ∪ peer sessions).
//! * `spawn` / `kill` hit the daemon's local session endpoints.
//! * `attach` opens the daemon's `WS /api/attach/:id` broker, which picks the
//! transport and enforces `allow_inbound` on the target side.
//! transport and dials the target (the target verifies the registry-minted
//! pair-attach JWT — session binding, not a per-peer allow-list).
//!
//! If the home daemon is **unreachable** (connection refused / not running),
//! every command falls back to the legacy registry path (registry HTTP +
@@ -23,10 +24,9 @@
use std::path::{Path, PathBuf};
use crate::attach::{broker_attach_url, parse_detach_key, run_attach, run_attach_url};
use crate::cli::{AttachArgs, Command, KillArgs, ListArgs, PolicyCmd, SetRegistryArgs, SpawnArgs};
use crate::cli::{AttachArgs, Command, KillArgs, ListArgs, SetRegistryArgs, SpawnArgs};
use crate::config::{default_config_path, CliConfig, ConfigError};
use crate::http::{HttpError, RegistryClient};
use crate::policy::{run_reload as run_policy_reload, run_show as run_policy_show};
use crate::sessions::{fetch_connect_info, run_kill, run_list, run_spawn};
#[derive(Debug, thiserror::Error)]
@@ -40,8 +40,6 @@ pub enum CliError {
Attach(#[from] crate::attach::AttachError),
#[error("{0}")]
Io(#[from] std::io::Error),
#[error("{0}")]
Policy(#[from] crate::policy::PolicyError),
}
/// Dispatch a parsed top-level subcommand. The umbrella binary passes the
@@ -61,7 +59,6 @@ pub async fn dispatch(
match command {
Command::SetRegistry(args) => set_registry(&config_path, args),
Command::Policy(sub) => policy(&config_path, sub).await,
Command::List(args) => {
list(
&config_path,
@@ -109,16 +106,6 @@ fn set_registry(config_path: &Path, args: SetRegistryArgs) -> Result<(), CliErro
Ok(())
}
#[allow(clippy::result_large_err)]
async fn policy(config_path: &Path, sub: PolicyCmd) -> Result<(), CliError> {
let mut stdout = std::io::stdout().lock();
match sub {
PolicyCmd::Reload => run_policy_reload(config_path, &mut stdout).await?,
PolicyCmd::Show { json } => run_policy_show(config_path, json, &mut stdout).await?,
}
Ok(())
}
#[allow(clippy::result_large_err)]
async fn list(
config_path: &Path,
-1
View File
@@ -10,7 +10,6 @@ pub mod cli;
pub mod config;
pub mod dispatch;
pub mod http;
pub mod policy;
pub mod sessions;
pub use dispatch::{dispatch, CliError};
-127
View File
@@ -1,127 +0,0 @@
//! `maestro policy reload | show` — talk to the local daemon's
//! `/api/policy` endpoints (NAP-237).
//!
//! The local daemon's bind port is read from `[daemon].bind_addr` in the
//! unified config; the host is always `127.0.0.1` (these endpoints are
//! same-host by design — the daemon's bind may be `0.0.0.0` for attach,
//! but operators reload policy from the same machine).
use std::io::Write;
use std::path::Path;
use maestro_config::MaestroConfigFile;
use serde::Deserialize;
const DEFAULT_PORT: u16 = 7070;
#[derive(Debug, thiserror::Error)]
pub enum PolicyError {
#[error("config: {0}")]
Config(#[from] maestro_config::ConfigError),
#[error("http: {0}")]
Http(#[from] reqwest::Error),
#[error("daemon returned {status}: {body}")]
DaemonError {
status: reqwest::StatusCode,
body: String,
},
#[error("io: {0}")]
Io(#[from] std::io::Error),
}
#[derive(Debug, Deserialize)]
pub struct PolicyView {
pub allow_inbound: Vec<String>,
}
pub async fn run_reload(config_path: &Path, out: &mut impl Write) -> Result<(), PolicyError> {
let url = format!("{}/api/policy/reload", local_base_url(config_path)?);
let resp = crate::http::build_client().post(&url).send().await?;
let status = resp.status();
if !status.is_success() {
let body = resp.text().await.unwrap_or_default();
return Err(PolicyError::DaemonError { status, body });
}
let view: PolicyView = resp.json().await?;
writeln!(
out,
"reloaded — allow_inbound now has {} entr{}",
view.allow_inbound.len(),
if view.allow_inbound.len() == 1 {
"y"
} else {
"ies"
},
)?;
for entry in &view.allow_inbound {
writeln!(out, " {entry}")?;
}
Ok(())
}
pub async fn run_show(
config_path: &Path,
json: bool,
out: &mut impl Write,
) -> Result<(), PolicyError> {
let url = format!("{}/api/policy", local_base_url(config_path)?);
let resp = crate::http::build_client().get(&url).send().await?;
let status = resp.status();
if !status.is_success() {
let body = resp.text().await.unwrap_or_default();
return Err(PolicyError::DaemonError { status, body });
}
let view: PolicyView = resp.json().await?;
if json {
writeln!(
out,
"{}",
serde_json::to_string(&view.allow_inbound).unwrap()
)?;
} else if view.allow_inbound.is_empty() {
writeln!(
out,
"(allow_inbound is empty — every cross-daemon attach is denied)"
)?;
} else {
for entry in &view.allow_inbound {
writeln!(out, "{entry}")?;
}
}
Ok(())
}
/// `http://127.0.0.1:<port>` where `<port>` is `[daemon].bind_addr`'s port
/// or `7070` if unset / file missing.
fn local_base_url(config_path: &Path) -> Result<String, PolicyError> {
let file = MaestroConfigFile::load_or_default(config_path)?;
let port = file
.daemon
.bind_addr
.map(|a| a.port())
.unwrap_or(DEFAULT_PORT);
Ok(format!("http://127.0.0.1:{port}"))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn local_base_url_uses_bind_port_or_default() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
// missing file → default
assert_eq!(local_base_url(&path).unwrap(), "http://127.0.0.1:7070");
std::fs::write(
&path,
r#"
[daemon]
bind_addr = "0.0.0.0:9090"
"#,
)
.unwrap();
assert_eq!(local_base_url(&path).unwrap(), "http://127.0.0.1:9090");
}
}
-23
View File
@@ -164,9 +164,6 @@ pub struct DaemonSection {
/// enable `loginctl enable-linger`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tmux_tmpdir: Option<PathBuf>,
/// `[daemon.policy]` — per-daemon attach policy (NAP-247).
#[serde(default, skip_serializing_if = "DaemonPolicySection::is_empty")]
pub policy: DaemonPolicySection,
}
impl DaemonSection {
@@ -181,23 +178,6 @@ impl DaemonSection {
&& self.private_addr.is_none()
&& self.pty_replay_bytes.is_none()
&& self.tmux_tmpdir.is_none()
&& self.policy.is_empty()
}
}
/// `[daemon.policy]` — application-layer access control complementing network
/// ACLs. Default deny: an empty `allow_inbound` rejects every cross-daemon
/// attach. Local attach (no JWT, same daemon) always bypasses.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct DaemonPolicySection {
/// daemon-ids permitted to broker-attach into this daemon's sessions.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub allow_inbound: Vec<String>,
}
impl DaemonPolicySection {
fn is_empty(&self) -> bool {
self.allow_inbound.is_empty()
}
}
@@ -391,9 +371,6 @@ mod tests {
private_addr: Some("10.0.0.7:7070".into()),
pty_replay_bytes: Some(131_072),
tmux_tmpdir: Some(PathBuf::from("/run/user/1000")),
policy: DaemonPolicySection {
allow_inbound: vec!["konishi".into()],
},
},
registry: RegistrySection {
bind_addr: Some("0.0.0.0:8080".parse().unwrap()),
+5 -16
View File
@@ -15,9 +15,9 @@ use crate::sidecar::SidecarStore;
use crate::tmux::Tmux;
use crate::worktree::WorktreeManager;
use crate::{
attach, default_config_path, default_state_dir, load_config, local_api, policy, run_configure,
spawn_session_publisher, spawn_sighup_listener, ConfigureInputs, EnvOverrides, JwtVerifier,
LocalEventBus, PairAttachVerifier, PolicyState, RegistryClient, SessionManager,
attach, default_config_path, default_state_dir, load_config, local_api, run_configure,
spawn_session_publisher, ConfigureInputs, EnvOverrides, JwtVerifier, LocalEventBus,
PairAttachVerifier, RegistryClient, SessionManager,
};
#[derive(Debug, Subcommand)]
@@ -113,11 +113,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
manager.start_liveness_watcher(DEFAULT_LIVENESS_INTERVAL);
let verifier = JwtVerifier::new(&config.token, &config.daemon_id);
let pair_verifier = PairAttachVerifier::new(
&config.token,
&config.daemon_id,
config.allow_inbound.iter().cloned(),
);
let pair_verifier = PairAttachVerifier::new(&config.token, &config.daemon_id);
let events = LocalEventBus::new();
spawn_session_publisher(manager.clone(), events.clone());
@@ -151,16 +147,9 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
self_daemon_id: std::sync::Arc::new(config.daemon_id.clone()),
self_perimeter_id: config.perimeter_id.clone().map(std::sync::Arc::new),
};
let policy_state = PolicyState {
verifier: pair_verifier.clone(),
config_path: std::sync::Arc::new(config_path.clone()),
};
spawn_sighup_listener(policy_state.clone());
let router = attach::router(manager.clone(), verifier, pair_verifier)
.merge(local_api::router(local_state))
.merge(local_api::broker_router(broker_state))
.merge(policy::router(policy_state));
.merge(local_api::broker_router(broker_state));
let listener = TcpListener::bind(config.bind_addr).await?;
let bound = listener.local_addr()?;
-5
View File
@@ -90,9 +90,6 @@ pub struct DaemonConfig {
/// Daemon's reachable address on the declared perimeter. Only meaningful
/// when `perimeter_id` is set.
pub private_addr: Option<String>,
/// daemon-ids permitted to broker-attach to this daemon's sessions.
/// Default empty (deny all). Local attach bypasses. (NAP-247.)
pub allow_inbound: Vec<String>,
/// Per-session PTY replay buffer cap in bytes (NAP-229). Each new attach
/// receives this much of the prior PTY output before the live stream.
/// Default `pty::DEFAULT_REPLAY_BYTES`; `0` disables.
@@ -242,7 +239,6 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result<DaemonConfig, Confi
let perimeter_id = env.perimeter_id.or(daemon.perimeter_id);
let private_addr = env.private_addr.or(daemon.private_addr);
let allow_inbound = daemon.policy.allow_inbound;
let pty_replay_bytes = daemon
.pty_replay_bytes
.unwrap_or(crate::pty::DEFAULT_REPLAY_BYTES);
@@ -264,7 +260,6 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result<DaemonConfig, Confi
worktree_root,
perimeter_id,
private_addr,
allow_inbound,
pty_replay_bytes,
tmux_tmpdir,
})
-3
View File
@@ -125,9 +125,6 @@ pub fn run(
// are operator-edited in config.toml directly. Leave None on bootstrap.
perimeter_id: None,
private_addr: None,
// Default deny: empty allow_inbound means no peer can attach until the
// operator opts in. Set in config.toml under [daemon.policy].
policy: Default::default(),
// None ⇒ daemon falls back to pty::DEFAULT_REPLAY_BYTES. Override in
// config.toml as `[daemon] pty_replay_bytes = ...` (NAP-229).
pty_replay_bytes: None,
+24 -117
View File
@@ -2,9 +2,6 @@
//! validated locally by each daemon using the shared secret — no callback to
//! the registry on the attach hot path. Spec §Auth.
use std::collections::HashSet;
use std::sync::{Arc, RwLock};
use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation};
use maestro_protocol::{JwtClaims, PairAttachClaims, REGISTRY_ISSUER};
use uuid::Uuid;
@@ -17,8 +14,6 @@ pub enum JwtError {
WrongDaemon { wanted: String, ours: String },
#[error("session mismatch (token wants `{wanted}`, path is `{path}`)")]
WrongSession { wanted: Uuid, path: Uuid },
#[error("home daemon `{sub}` not in this daemon's allow_inbound")]
NotAllowed { sub: String },
}
#[derive(Clone)]
@@ -68,61 +63,41 @@ pub fn issue(secret: &str, claims: &JwtClaims) -> Result<String, jsonwebtoken::e
}
/// Verifier for pair-scoped attach JWTs (NAP-247). Checks signature, `exp`,
/// `aud == self.daemon_id`, and `sub in allow_inbound`. The `allow_inbound`
/// set lives behind an [`Arc<RwLock<_>>`] so it can be swapped at runtime by
/// [`PairAttachVerifier::reload`] (NAP-237 hot reload via SIGHUP /
/// `POST /api/policy/reload`). Clones share the same inner — any reload
/// observed by one clone is observed by all of them.
/// `iss == REGISTRY_ISSUER`, `aud == self.daemon_id`, and `sid == session_id`
/// — i.e. the token was minted by the registry for *this* daemon and *this*
/// session, and hasn't expired. This is the session-binding guarantee the
/// network perimeter can't provide: a host the firewall admits still can't
/// attach to a session it wasn't handed, or replay a token to another
/// session/daemon.
///
/// It does **not** gate on the home daemon's identity (no `allow_inbound`
/// allow-list, NAP-339): the operator's perimeter/firewall is the trust
/// boundary for *which hosts may connect*, so every daemon on the perimeter is
/// equally trusted to attach. Re-introducing a per-pair gate later is a
/// verify-time check, not an architecture change.
#[derive(Clone)]
pub struct PairAttachVerifier {
decoding: DecodingKey,
daemon_id: String,
allow_inbound: Arc<RwLock<HashSet<String>>>,
}
impl std::fmt::Debug for PairAttachVerifier {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PairAttachVerifier")
.field("daemon_id", &self.daemon_id)
.field("allow_inbound", &self.snapshot())
.field("decoding", &"<redacted>")
.finish()
}
}
impl PairAttachVerifier {
pub fn new(
secret: &str,
daemon_id: impl Into<String>,
allow_inbound: impl IntoIterator<Item = String>,
) -> Self {
pub fn new(secret: &str, daemon_id: impl Into<String>) -> Self {
Self {
decoding: DecodingKey::from_secret(secret.as_bytes()),
daemon_id: daemon_id.into(),
allow_inbound: Arc::new(RwLock::new(allow_inbound.into_iter().collect())),
}
}
/// Swap the effective `allow_inbound` set. Affects subsequent `verify`
/// calls; in-flight attaches keep their existing connection (per spec
/// — revocation requires killing the session).
pub fn reload(&self, allow_inbound: impl IntoIterator<Item = String>) {
let mut g = self
.allow_inbound
.write()
.unwrap_or_else(|e| e.into_inner());
*g = allow_inbound.into_iter().collect();
}
/// Sorted snapshot of the current effective `allow_inbound` set. Used by
/// `GET /api/policy` and debug formatting.
pub fn snapshot(&self) -> Vec<String> {
let g = self.allow_inbound.read().unwrap_or_else(|e| e.into_inner());
let mut v: Vec<String> = g.iter().cloned().collect();
v.sort();
v
}
/// Validate the token for the path-bound `session_id`. On success returns
/// the home daemon's id (`sub`) for logging/tracing.
pub fn verify(&self, token: &str, session_id: Uuid) -> Result<String, JwtError> {
@@ -148,13 +123,6 @@ impl PairAttachVerifier {
path: session_id,
});
}
let allowed = {
let g = self.allow_inbound.read().unwrap_or_else(|e| e.into_inner());
g.contains(&claims.sub)
};
if !allowed {
return Err(JwtError::NotAllowed { sub: claims.sub });
}
Ok(claims.sub)
}
}
@@ -230,42 +198,21 @@ mod tests {
}
#[test]
fn pair_verifier_accepts_allow_listed_home() {
fn pair_verifier_accepts_registry_minted_token() {
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
let v = PairAttachVerifier::new(secret, "konishi");
let sid = Uuid::new_v4();
// The home daemon's identity is no longer gated (NAP-339) — any
// registry-minted, correctly-bound token for this daemon+session passes.
let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
let sub = v.verify(&token, sid).unwrap();
assert_eq!(sub, "laptop");
}
#[test]
fn pair_verifier_rejects_unlisted_home() {
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", ["other".into()]);
let sid = Uuid::new_v4();
let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
let err = v.verify(&token, sid).unwrap_err();
assert!(matches!(err, JwtError::NotAllowed { .. }), "{err:?}");
}
#[test]
fn pair_verifier_rejects_empty_allow_inbound() {
// Default deny: no peer listed → every cross-daemon attach fails.
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", std::iter::empty::<String>());
let sid = Uuid::new_v4();
let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
assert!(matches!(
v.verify(&token, sid).unwrap_err(),
JwtError::NotAllowed { .. }
));
}
#[test]
fn pair_verifier_rejects_wrong_audience() {
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
let v = PairAttachVerifier::new(secret, "konishi");
let sid = Uuid::new_v4();
// Token addressed to a different target — must not pass our checks.
let token = issue_pair(secret, &pair_claims("laptop", "elsewhere", sid, 60));
@@ -278,7 +225,7 @@ mod tests {
// `iss` (e.g. minted for some other purpose) must not validate as a
// pair-attach credential.
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
let v = PairAttachVerifier::new(secret, "konishi");
let sid = Uuid::new_v4();
let mut claims = pair_claims("laptop", "konishi", sid, 60);
claims.iss = "somebody-else".into();
@@ -289,7 +236,7 @@ mod tests {
#[test]
fn pair_verifier_rejects_session_id_mismatch() {
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
let v = PairAttachVerifier::new(secret, "konishi");
let sid = Uuid::new_v4();
let path = Uuid::new_v4();
let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
@@ -302,57 +249,17 @@ mod tests {
#[test]
fn pair_verifier_rejects_expired() {
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]);
let v = PairAttachVerifier::new(secret, "konishi");
let sid = Uuid::new_v4();
let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, -120));
assert!(v.verify(&token, sid).is_err());
}
// NAP-237: hot reload. After `reload`, the same verifier (and any
// pre-existing clone) sees the new set on the next `verify`.
#[test]
fn pair_verifier_reload_admits_new_peer() {
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", std::iter::empty::<String>());
fn pair_verifier_rejects_wrong_secret() {
let v = PairAttachVerifier::new("right", "konishi");
let sid = Uuid::new_v4();
let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
let token = issue_pair("wrong", &pair_claims("laptop", "konishi", sid, 60));
assert!(v.verify(&token, sid).is_err());
v.reload(["laptop".into()]);
assert_eq!(v.verify(&token, sid).unwrap(), "laptop");
}
#[test]
fn pair_verifier_reload_revokes_removed_peer() {
let secret = "s";
let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into(), "phone".into()]);
let sid = Uuid::new_v4();
let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
assert_eq!(v.verify(&token, sid).unwrap(), "laptop");
v.reload(["phone".into()]); // drop laptop
assert!(matches!(
v.verify(&token, sid).unwrap_err(),
JwtError::NotAllowed { .. }
));
}
#[test]
fn pair_verifier_clone_shares_inner_set() {
// Cloning the verifier (e.g. to hand to axum state and to the SIGHUP
// task) must NOT detach the allow_inbound set — both halves observe
// every reload.
let secret = "s";
let v1 = PairAttachVerifier::new(secret, "konishi", std::iter::empty::<String>());
let v2 = v1.clone();
v1.reload(["laptop".into()]);
let sid = Uuid::new_v4();
let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60));
assert_eq!(v2.verify(&token, sid).unwrap(), "laptop");
}
#[test]
fn pair_verifier_snapshot_is_sorted() {
let v = PairAttachVerifier::new("s", "konishi", ["phone".into(), "laptop".into()]);
assert_eq!(v.snapshot(), vec!["laptop".to_string(), "phone".into()]);
}
}
-2
View File
@@ -16,7 +16,6 @@ pub mod dirlist;
pub mod jwt;
pub mod local_api;
pub mod local_events;
pub mod policy;
pub mod pty;
pub mod registry_client;
pub mod session;
@@ -30,6 +29,5 @@ pub use config::{default_config_path, default_state_dir, load_config, DaemonConf
pub use configure::{run as run_configure, ConfigureError, ConfigureInputs};
pub use jwt::{issue as issue_jwt, JwtError, JwtVerifier, PairAttachVerifier};
pub use local_events::{spawn_session_publisher, LocalEventBus};
pub use policy::{spawn_sighup_listener, PolicyState};
pub use registry_client::{ClientError, RegistryClient, HEARTBEAT_INTERVAL};
pub use session::{SessionError, SessionEvent, SessionManager};
-196
View File
@@ -1,196 +0,0 @@
//! Hot reload for `[daemon.policy] allow_inbound` (NAP-237).
//!
//! Two triggers, one execution path:
//!
//! - **SIGHUP** — `systemctl --user reload maestrod` (or any `kill -HUP`).
//! - **`POST /api/policy/reload`** — used by `maestro policy reload` so
//! operators don't need to know the PID. Also gives `maestro-web` and
//! other local-API consumers a uniform reload hook.
//!
//! Both re-read `~/.config/maestro/config.toml` and swap the verifier's
//! allow-list atomically; in-flight attaches keep their existing connection
//! (revoking those requires killing the session, per spec).
use std::path::{Path, PathBuf};
use std::sync::Arc;
use axum::extract::State;
use axum::http::StatusCode;
use axum::routing::{get, post};
use axum::{Json, Router};
use maestro_config::MaestroConfigFile;
use serde::Serialize;
use crate::jwt::PairAttachVerifier;
#[derive(Debug, Clone)]
pub struct PolicyState {
pub verifier: PairAttachVerifier,
pub config_path: Arc<PathBuf>,
}
#[derive(Debug, Serialize)]
pub struct PolicyView {
pub allow_inbound: Vec<String>,
}
pub fn router(state: PolicyState) -> Router {
Router::new()
.route("/api/policy", get(get_policy))
.route("/api/policy/reload", post(reload_policy))
.with_state(state)
}
async fn get_policy(State(s): State<PolicyState>) -> Json<PolicyView> {
Json(PolicyView {
allow_inbound: s.verifier.snapshot(),
})
}
async fn reload_policy(
State(s): State<PolicyState>,
) -> Result<Json<PolicyView>, (StatusCode, String)> {
match reload_from_file(&s.config_path, &s.verifier) {
Ok(list) => {
tracing::info!(?list, source = "http", "reloaded allow_inbound");
Ok(Json(PolicyView {
allow_inbound: list,
}))
}
Err(e) => {
tracing::error!(error = %e, source = "http", "policy reload failed");
Err((StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))
}
}
}
/// Re-read the unified config file and apply `[daemon.policy].allow_inbound`
/// to the live verifier. Returns the new effective list (sorted), or an
/// error if the file can't be read/parsed — in which case the existing
/// allow-list stays in place so a bad edit doesn't lock everyone out.
pub fn reload_from_file(
config_path: &Path,
verifier: &PairAttachVerifier,
) -> Result<Vec<String>, maestro_config::ConfigError> {
let file = MaestroConfigFile::load(config_path)?;
let list = file.daemon.policy.allow_inbound;
verifier.reload(list.iter().cloned());
Ok(verifier.snapshot())
}
/// Spawn a tokio task that re-applies the policy on every SIGHUP. Logs
/// success / failure but never exits — a transient bad config (e.g. mid-edit)
/// becomes a logged error, not a daemon kill.
pub fn spawn_sighup_listener(state: PolicyState) {
tokio::spawn(async move {
let mut sig = match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup()) {
Ok(s) => s,
Err(e) => {
tracing::error!(error = %e, "failed to install SIGHUP handler");
return;
}
};
while sig.recv().await.is_some() {
match reload_from_file(&state.config_path, &state.verifier) {
Ok(list) => {
tracing::info!(?list, source = "sighup", "reloaded allow_inbound");
}
Err(e) => {
tracing::error!(error = %e, source = "sighup", "policy reload failed");
}
}
}
});
}
#[cfg(test)]
mod tests {
use super::*;
fn write_config(path: &Path, allow_inbound: &[&str]) {
let body = if allow_inbound.is_empty() {
r#"
[shared]
registry_url = "http://r"
token = "t"
[daemon]
daemon_url = "ws://x"
"#
.to_string()
} else {
let list = allow_inbound
.iter()
.map(|s| format!("\"{s}\""))
.collect::<Vec<_>>()
.join(", ");
format!(
r#"
[shared]
registry_url = "http://r"
token = "t"
[daemon]
daemon_url = "ws://x"
[daemon.policy]
allow_inbound = [{list}]
"#
)
};
std::fs::write(path, body).unwrap();
}
#[test]
fn reload_from_file_picks_up_new_entries() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
write_config(&path, &["laptop"]);
let v = PairAttachVerifier::new("s", "konishi", ["laptop".into()]);
assert_eq!(v.snapshot(), vec!["laptop".to_string()]);
write_config(&path, &["laptop", "phone"]);
let list = reload_from_file(&path, &v).unwrap();
assert_eq!(list, vec!["laptop".to_string(), "phone".into()]);
assert_eq!(v.snapshot(), vec!["laptop".to_string(), "phone".into()]);
}
#[test]
fn reload_from_file_drops_removed_entries() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
write_config(&path, &["laptop", "phone"]);
let v = PairAttachVerifier::new("s", "konishi", ["laptop".into(), "phone".into()]);
write_config(&path, &["phone"]);
let list = reload_from_file(&path, &v).unwrap();
assert_eq!(list, vec!["phone".to_string()]);
assert_eq!(v.snapshot(), vec!["phone".to_string()]);
}
#[test]
fn reload_from_file_leaves_set_intact_on_parse_error() {
// A mid-edit broken config must not lock everyone out — keep the
// previously-effective list and surface the error to the caller.
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
write_config(&path, &["laptop"]);
let v = PairAttachVerifier::new("s", "konishi", ["laptop".into()]);
std::fs::write(&path, "this = is = not = toml").unwrap();
assert!(reload_from_file(&path, &v).is_err());
assert_eq!(v.snapshot(), vec!["laptop".to_string()]);
}
#[test]
fn reload_from_file_treats_missing_section_as_deny_all() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
write_config(&path, &["laptop"]);
let v = PairAttachVerifier::new("s", "konishi", ["laptop".into()]);
write_config(&path, &[]); // no [daemon.policy] at all
let list = reload_from_file(&path, &v).unwrap();
assert!(list.is_empty());
assert!(v.snapshot().is_empty());
}
}
+1 -1
View File
@@ -36,7 +36,7 @@ async fn spawn_daemon_ws(daemon_id: &str) -> Harness {
let fx = DaemonFixture::new(daemon_id);
let manager = fx.manager.clone();
let verifier = JwtVerifier::new(TOKEN, daemon_id);
let pair_verifier = PairAttachVerifier::new(TOKEN, daemon_id, std::iter::empty::<String>());
let pair_verifier = PairAttachVerifier::new(TOKEN, daemon_id);
let router = attach::router(manager.clone(), verifier, pair_verifier);
let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
.await
+2 -30
View File
@@ -27,13 +27,12 @@ pub struct Daemon {
token: String,
}
/// Optional knobs for `Daemon::spawn_with` — perimeter + allow_inbound for
/// Optional knobs for `Daemon::spawn_with` — perimeter advertisement for
/// NAP-247 broker tests.
#[derive(Debug, Default, Clone)]
pub struct DaemonOptions {
pub perimeter_id: Option<String>,
pub private_addr: Option<String>,
pub allow_inbound: Vec<String>,
}
impl Daemon {
@@ -50,7 +49,7 @@ impl Daemon {
.await
}
/// Spawn with NAP-247 options (perimeter advertisement, allow_inbound).
/// Spawn with NAP-247 options (perimeter advertisement).
pub async fn spawn_with(
daemon_id: &str,
registry_base_url: &str,
@@ -116,11 +115,6 @@ impl Daemon {
let state = TempDir::new().expect("state tmpdir");
let tmux = TempDir::new().expect("tmux tmpdir");
// Write a minimal config.toml so allow_inbound (and any future
// file-only fields) reach the daemon. Required fields keep coming
// through env vars set by `launch`.
write_config(home.path(), &opts);
let daemon_url = format!("ws://{bind}");
let child = launch(
daemon_id,
@@ -219,25 +213,3 @@ fn launch(
cmd.spawn()
.unwrap_or_else(|e| panic!("spawn {} failed: {e}", bin.display()))
}
/// Write a minimal `~/.config/maestro/config.toml` carrying the NAP-247
/// `[daemon.policy] allow_inbound` list. Everything else still comes from
/// env vars set by `launch`.
fn write_config(home: &std::path::Path, opts: &DaemonOptions) {
if opts.allow_inbound.is_empty() {
return;
}
let dir = home.join(".config/maestro");
std::fs::create_dir_all(&dir).unwrap();
let mut allow = String::new();
for (i, id) in opts.allow_inbound.iter().enumerate() {
if i > 0 {
allow.push_str(", ");
}
allow.push('"');
allow.push_str(id);
allow.push('"');
}
let body = format!("[daemon.policy]\nallow_inbound = [{allow}]\n");
std::fs::write(dir.join("config.toml"), body).unwrap();
}
-114
View File
@@ -513,7 +513,6 @@ async fn broker_t2_attach_through_home_daemon() {
DaemonOptions {
perimeter_id: Some("wg-test".into()),
private_addr: Some(beta_bind.to_string()),
allow_inbound: vec!["alpha".into()],
},
)
.await;
@@ -529,7 +528,6 @@ async fn broker_t2_attach_through_home_daemon() {
DaemonOptions {
perimeter_id: Some("wg-test".into()),
private_addr: Some(alpha_bind.to_string()),
allow_inbound: vec![],
},
)
.await;
@@ -621,116 +619,6 @@ async fn broker_t2_attach_through_home_daemon() {
drop(registry);
}
#[tokio::test(flavor = "multi_thread")]
async fn broker_rejects_when_target_doesnt_allow_inbound() {
// NAP-247: beta has an empty allow_inbound, so beta's
// /sessions/:id/stream rejects alpha's broker JWT even though the
// registry happily mints one (default deny is target-side only).
let registry = Registry::spawn("e2e-shared-token").await;
let beta_bind = std::net::SocketAddr::from((
std::net::Ipv4Addr::LOCALHOST,
maestro_e2e::util::free_port(),
));
let beta = Daemon::spawn_at(
"beta",
&registry.http_base(),
"e2e-shared-token",
beta_bind,
DaemonOptions {
perimeter_id: Some("wg-test".into()),
private_addr: Some(beta_bind.to_string()),
// Default deny — alpha not listed.
allow_inbound: vec![],
},
)
.await;
let alpha_bind = std::net::SocketAddr::from((
std::net::Ipv4Addr::LOCALHOST,
maestro_e2e::util::free_port(),
));
let alpha = Daemon::spawn_at(
"alpha",
&registry.http_base(),
"e2e-shared-token",
alpha_bind,
DaemonOptions {
perimeter_id: Some("wg-test".into()),
private_addr: Some(alpha_bind.to_string()),
allow_inbound: vec![],
},
)
.await;
assert_eq!(beta.bind_addr, beta_bind);
assert_eq!(alpha.bind_addr, alpha_bind);
let client = TestClient::new(&registry.http_base(), &registry.ws_base());
client.bootstrap().await.expect("bootstrap");
wait_for_async("both daemons connected", Duration::from_secs(10), || {
let client = client.clone();
async move {
client
.list_daemons()
.await
.map(|ds| ds.len() >= 2)
.unwrap_or(false)
}
})
.await;
let cwd = TempDir::new().unwrap();
let session = client
.spawn_session(&create_req("beta", cwd.path(), "sleep 30"))
.await
.expect("spawn beta");
let alpha_base = format!("http://{}", alpha.bind_addr);
let sid = session.id;
wait_for_async(
"alpha mirrors beta's session",
Duration::from_secs(5),
|| {
let base = alpha_base.clone();
async move {
let resp = match reqwest::get(format!("{base}/api/sessions")).await {
Ok(r) => r,
Err(_) => return false,
};
let list: Vec<maestro_protocol::Session> = match resp.json().await {
Ok(v) => v,
Err(_) => return false,
};
list.iter().any(|s| s.id == sid)
}
},
)
.await;
let broker_url = format!("ws://{}/api/attach/{}", alpha.bind_addr, sid);
// NAP-265: beta rejects the pair JWT (empty allow_inbound) pre-upgrade.
// alpha's broker now dials the peer *before* completing the upgrade, so the
// client's handshake fails with a structured 502 rather than succeeding and
// then closing opaquely.
let err = tokio_tungstenite::connect_async(broker_url)
.await
.expect_err("handshake must fail when the peer rejects the attach");
match err {
tokio_tungstenite::tungstenite::Error::Http(resp) => {
assert_eq!(
resp.status(),
502,
"expected BAD_GATEWAY from alpha's broker, got {}",
resp.status()
);
}
other => panic!("expected an HTTP 502 response, got {other:?}"),
}
client.kill_session(sid).await.expect("kill");
drop(alpha);
drop(beta);
drop(registry);
}
#[tokio::test(flavor = "multi_thread")]
async fn broker_attach_returns_502_on_perimeter_mismatch() {
// NAP-270: the asymmetric-VLAN boundary. alpha and beta are on DIFFERENT
@@ -749,7 +637,6 @@ async fn broker_attach_returns_502_on_perimeter_mismatch() {
DaemonOptions {
perimeter_id: Some("wg-beta".into()),
private_addr: Some(beta_bind.to_string()),
allow_inbound: vec!["alpha".into()],
},
)
.await;
@@ -766,7 +653,6 @@ async fn broker_attach_returns_502_on_perimeter_mismatch() {
// Different perimeter from beta → T2 unavailable.
perimeter_id: Some("wg-alpha".into()),
private_addr: Some(alpha_bind.to_string()),
allow_inbound: vec![],
},
)
.await;
+6 -5
View File
@@ -57,7 +57,6 @@ pub fn render_user_unit(exec_start: &Path) -> String {
[Service]\n\
Type=simple\n\
ExecStart={exec} daemon\n\
ExecReload=/bin/kill -HUP $MAINPID\n\
# Signal only the daemon on stop/restart, not its whole cgroup —\n\
# otherwise systemd kills the tmux server (and every session) the\n\
# daemon spawned, defeating tmux persistence. recover() re-adopts the\n\
@@ -165,11 +164,13 @@ mod tests {
}
#[test]
fn unit_wires_execreload_to_sighup() {
// NAP-237: `systemctl --user reload maestro` must trigger our
// SIGHUP handler so operators can hot-reload `[daemon.policy]`.
fn unit_has_no_execreload() {
// NAP-339 removed the daemon's SIGHUP handler (it only ever reloaded
// the now-deleted `[daemon.policy]`). Without a handler, SIGHUP's
// default disposition is to terminate, so the unit must NOT wire
// `systemctl reload` to it — that would kill the daemon.
let body = render_user_unit(Path::new("/x/maestro"));
assert!(body.contains("ExecReload=/bin/kill -HUP $MAINPID"));
assert!(!body.contains("ExecReload"));
}
#[test]
+4 -3
View File
@@ -252,8 +252,8 @@ pub enum DaemonInbound {
result: RpcResult<DirListing>,
},
/// Registry signals a target daemon that a home daemon wants to attach via
/// the registry-mediated relay (T3). Target evaluates `allow_inbound` and
/// replies with [`DaemonOutbound::RelayAck`].
/// the registry-mediated relay (T3). Target decides whether to accept the
/// relay and replies with [`DaemonOutbound::RelayAck`].
RelaySetup {
reply_to: ReplyTo,
home_daemon: String,
@@ -465,7 +465,8 @@ pub const REGISTRY_ISSUER: &str = "maestro-registry";
/// Pair-scoped attach credential. Issued by the registry inside
/// [`AttachCandidates::jwt`]; verified by the target daemon for direct
/// transports (T1/T2) and gated by `allow_inbound` after JWT verification.
/// transports (T1/T2). Binds the token to a target daemon and session; the
/// home daemon's identity (`sub`) is not gated against an allow-list (NAP-339).
///
/// "Pair-scoped" means: `sub` (home daemon) may initiate one attach to
/// `aud` (target daemon) for session `sid`, valid until `exp`. A stolen
+4 -2
View File
@@ -38,8 +38,10 @@ pub fn issue_attach_jwt(
/// Mint a pair-scoped attach JWT used by the home-daemon broker (NAP-247).
/// `sub` is the home daemon-id; `aud` is the target daemon-id. Target
/// validates signature, `aud == self.daemon_id`, `exp > now`, and
/// `sub in allow_inbound` before accepting the broker-attach.
/// validates signature, `iss == REGISTRY_ISSUER`, `aud == self.daemon_id`,
/// `sid == session`, and `exp > now` before accepting the broker-attach.
/// `sub` is no longer gated against an allow-list (NAP-339) — it's retained
/// for logging/tracing.
pub fn issue_pair_attach_jwt(
secret: &str,
home_daemon_id: &str,
-15
View File
@@ -69,10 +69,6 @@ enum Command {
/// Set the default registry URL persisted in the CLI config.
#[cfg(feature = "cli")]
SetRegistry(maestro_cli::cli::SetRegistryArgs),
/// Inspect or hot-reload the local daemon's `[daemon.policy]` (NAP-237).
#[cfg(feature = "cli")]
#[command(subcommand)]
Policy(maestro_cli::cli::PolicyCmd),
/// Per-host daemon: tmux wrapper + registry client.
#[cfg(feature = "daemon")]
@@ -160,16 +156,6 @@ async fn dispatch(cli: Cli) -> Result<(), Box<dyn std::error::Error>> {
)
.await?
}
#[cfg(feature = "cli")]
Command::Policy(sub) => {
cli_dispatch(
maestro_cli::cli::Command::Policy(sub),
cli.config,
cli.registry_url,
cli.home_daemon_url,
)
.await?
}
#[cfg(feature = "daemon")]
Command::Daemon { cmd } => maestro_daemon::cli::dispatch(cli.config, cmd).await?,
#[cfg(feature = "registry")]
@@ -220,7 +206,6 @@ fn is_cli_subcommand(command: &Command) -> bool {
| Command::Attach(_)
| Command::Kill(_)
| Command::SetRegistry(_)
| Command::Policy(_)
)
}