diff --git a/Cargo.lock b/Cargo.lock index e31ea2b..43f1384 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3375,6 +3375,7 @@ dependencies = [ "data-encoding", "ed25519-dalek", "iroh-relay", + "rand 0.10.3", "rds-cli", "rds-client", "rds-core", @@ -3384,6 +3385,9 @@ dependencies = [ "rds-observe", "rds-sync", "rustix", + "serde", + "serde_json", + "thiserror 2.0.20", "tokio", "tracing", ] diff --git a/README.md b/README.md index 0d39dea..3737973 100644 --- a/README.md +++ b/README.md @@ -123,6 +123,12 @@ bind access to the controlled endpoint. `rds session renew --session --grant-file ` extends a live same-scope connection; automatic GDS issuance is still pending. Old grants must be reissued, and CLI/agent IPC v4 upgraded together. +The agent's role, data-plane services, authority posture and deadlines can +also live in one versioned JSON document (`--agent-config`), with +`--role`/`--service`/`--no-service` as the flag surface; disabled services +are refused before grant machinery runs and `Info` advertises exactly what +is served. See [agent configuration](docs/agent-configuration.md). + Endpoint publication also persists its revision counter and exact retry bytes (`rds-agent --record-state`, default beside the identity key). Records and deletes use a versioned signature contract; old directories require the explicit diff --git a/crates/rds-agent/Cargo.toml b/crates/rds-agent/Cargo.toml index f6cb66b..c8da597 100644 --- a/crates/rds-agent/Cargo.toml +++ b/crates/rds-agent/Cargo.toml @@ -25,12 +25,17 @@ rds-desktop = { workspace = true, optional = true } rds-net.workspace = true rds-discovery.workspace = true rds-sync.workspace = true +rustix.workspace = true +serde.workspace = true +serde_json.workspace = true +thiserror.workspace = true tokio.workspace = true tracing.workspace = true [dev-dependencies] rustix = { workspace = true, features = ["process"] } iroh-relay = { workspace = true, features = ["server"] } +rand = { workspace = true } rds-cli = { path = "../rds-cli" } [lints] diff --git a/crates/rds-agent/src/lib.rs b/crates/rds-agent/src/lib.rs index 0dd7f9c..a76c2c2 100644 --- a/crates/rds-agent/src/lib.rs +++ b/crates/rds-agent/src/lib.rs @@ -22,7 +22,7 @@ //! TCP forwarding is restricted to an explicit set of `(host, port)` //! targets; the default set is exactly the configured SSH socket. -use std::collections::HashSet; +use std::collections::{BTreeSet, HashSet}; use std::path::PathBuf; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -38,9 +38,13 @@ use tracing::{Instrument, debug, info, info_span, warn}; mod authz; mod limits; mod revocations; +pub mod settings; use authz::{ConnAuthz, ConnectionLifetime, authorize}; pub use limits::AgentLimits; pub use revocations::{RevocationFeed, RevocationPolicy, watch_revocations}; +pub use settings::{ + AgentConfigError, AgentOverrides, AgentSettings, ResolvedAgent, Role, ServiceName, +}; /// Session ids and the `rds.conn` span shape are minted by rds-observe so /// both sides of a connection share the correlation convention. @@ -53,6 +57,32 @@ const HELLO_TIMEOUT: Duration = Duration::from_secs(15); const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(15); const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); +/// Operational deadlines on the serving side. Deployments tune these at +/// the policy level; session and transfer internals keep their own +/// service-scoped budgets. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TimeoutPolicy { + /// Inbound connection-handshake budget. + pub handshake: Duration, + /// Per-stream `StreamHello` read deadline — and the reply deadline for + /// greeting refusals that must not outlive a parked peer task. + pub hello: Duration, +} + +impl Default for TimeoutPolicy { + fn default() -> Self { + Self { + handshake: HANDSHAKE_TIMEOUT, + hello: HELLO_TIMEOUT, + } + } +} + +/// The largest accepted timeout; guards against effectively-unbounded +/// deadline configuration (roughly one workday of idle handshake budget +/// is never the intent). +pub const MAX_TIMEOUT: Duration = Duration::from_secs(3600); + /// Mutex acquisition that survives a poisoned lock: every mutex here /// guards plain data (a state word, an `Option`, a `HashSet`) whose /// invariants a panic cannot corrupt, so refusing service forever @@ -89,6 +119,14 @@ pub struct AgentPolicy { /// Directory the `Sync` service may read/write under (WS6). `None` /// disables sync entirely. pub sync_dir: Option, + /// Data-plane services this agent answers. `None` keeps the implicit + /// set — `Tcp` plus `Desktop` when compiled and `Sync` when `sync_dir` + /// is configured. `Some` is the explicit set; `Ping`/`Info` are the + /// always-on control plane and are never gated. A disabled service is + /// refused before any grant or service work runs. + pub services: Option>, + /// Admission and greeting deadlines applied per connection/stream. + pub timeouts: TimeoutPolicy, } impl std::fmt::Debug for AgentPolicy { @@ -100,6 +138,8 @@ impl std::fmt::Debug for AgentPolicy { .field("issuers", &self.issuers.len()) .field("grant_max_ttl", &self.grant_max_ttl) .field("sync_dir", &self.sync_dir) + .field("services", &self.services) + .field("timeouts", &self.timeouts) .finish_non_exhaustive() } } @@ -115,6 +155,8 @@ impl AgentPolicy { denylist: watch::channel(Arc::new(RevocationPolicy::default())).0, active_grants: Arc::new(Mutex::new(HashSet::new())), sync_dir: None, + services: None, + timeouts: TimeoutPolicy::default(), } } @@ -137,6 +179,76 @@ impl AgentPolicy { !self.issuers.is_empty() } + /// Every service this agent answers: the always-on `Ping`/`Info` + /// control plane plus either the explicit `services` set or the + /// implicit set derived from build features and configuration. + /// `Audio` is wire-reserved but unimplemented, so it is never in the + /// effective set even if it slips into an explicit one. + pub fn effective_services(&self) -> BTreeSet { + let mut set = BTreeSet::from([ServiceKind::Ping, ServiceKind::Info]); + match &self.services { + Some(explicit) => set.extend(explicit.iter().copied().filter(|k| { + matches!( + k, + ServiceKind::Tcp | ServiceKind::Desktop | ServiceKind::Sync + ) + })), + None => { + set.insert(ServiceKind::Tcp); + if cfg!(feature = "desktop") { + set.insert(ServiceKind::Desktop); + } + if self.sync_dir.is_some() { + set.insert(ServiceKind::Sync); + } + } + } + set + } + + /// Whether a service stream is admitted by deployment policy, before + /// any grant scope or per-service check. + fn service_enabled(&self, kind: ServiceKind) -> bool { + self.effective_services().contains(&kind) + } + + /// Deployment preflight: an explicit `services` set may only name + /// implemented services whose prerequisites are configured, and the + /// timeout policy must stay inside sane bounds. Programmatic callers + /// should run this before binding; `Agent::run` runs it too. + pub fn validate(&self) -> Result<(), &'static str> { + if let Some(set) = &self.services { + for kind in set { + match kind { + ServiceKind::Tcp => {} + ServiceKind::Desktop if cfg!(feature = "desktop") => {} + ServiceKind::Desktop => { + return Err("desktop service requires the `desktop` build feature"); + } + ServiceKind::Sync if self.sync_dir.is_some() => {} + ServiceKind::Sync => { + return Err("sync service requires a configured sync directory"); + } + ServiceKind::Audio => { + return Err("audio service is reserved but not implemented"); + } + _ => { + return Err("service set may only contain tcp, desktop or sync; \ + ping and info are always served"); + } + } + } + } + if self.timeouts.handshake.is_zero() + || self.timeouts.hello.is_zero() + || self.timeouts.handshake > MAX_TIMEOUT + || self.timeouts.hello > MAX_TIMEOUT + { + return Err("timeouts must be between 1 and 3600 seconds"); + } + Ok(()) + } + /// Revoke a grant id — pushes onto the denylist and notifies every /// live connection watcher. The value is retained without subscribers. pub fn revoke(&self, id: GrantId) { @@ -263,6 +375,9 @@ impl Agent { !self.policy.grants_required() || self.limits.streams() >= 2, "grant mode requires at least two stream slots" ); + self.policy + .validate() + .map_err(|why| anyhow::anyhow!("invalid agent policy: {why}"))?; info!(id = %self.endpoint.id(), "agent listening"); rds_observe::emit(rds_observe::Event::ListenerReady); let mut connections = JoinSet::new(); @@ -292,7 +407,7 @@ impl Agent { let stream_counter = self.stream_counter.clone(); connections.spawn(async move { let _permit = permit; - match tokio::time::timeout(HANDSHAKE_TIMEOUT, incoming).await { + match tokio::time::timeout(policy.timeouts.handshake, incoming).await { Ok(Ok(conn)) => { let span = conn_span(next_session_id()); span.record("peer", tracing::field::display(conn.remote_id())); @@ -460,14 +575,14 @@ async fn serve_stream( authz: Arc, desktop: bool, ) -> anyhow::Result<()> { - let hello: StreamHello = match tokio::time::timeout(HELLO_TIMEOUT, read_frame(&mut recv)).await - { - Ok(h) => h?, - Err(_) => { - rds_observe::request_refused(Reason::Timeout); - anyhow::bail!("stream hello timed out"); - } - }; + let hello: StreamHello = + match tokio::time::timeout(policy.timeouts.hello, read_frame(&mut recv)).await { + Ok(h) => h?, + Err(_) => { + rds_observe::request_refused(Reason::Timeout); + anyhow::bail!("stream hello timed out"); + } + }; if let StreamHello::Authz(grant) = hello { return rds_observe::observe( rds_observe::Operation::GrantAuthorize, @@ -482,6 +597,21 @@ async fn serve_stream( ) .await; } + // Deployment policy answers first: a disabled service is refused before + // grant machinery or per-service work runs. + if let Some(kind) = service_kind(&hello) + && !policy.service_enabled(kind) + { + rds_observe::request_refused(Reason::Denied); + write_frame( + &mut send, + &HelloAck::Error { + message: format!("service {kind:?} not enabled on this agent"), + }, + ) + .await?; + anyhow::bail!("service {kind:?} not enabled"); + } let grant = match authz.service_scope(&policy).await { Ok(g) => g, Err(why) => { @@ -509,7 +639,7 @@ async fn serve_stream( let Some(_service_slot) = authz.try_service_slot() else { rds_observe::request_refused(Reason::BudgetExhausted); tokio::time::timeout( - HELLO_TIMEOUT, + policy.timeouts.hello, write_frame( &mut send, &HelloAck::Error { @@ -546,14 +676,17 @@ async fn serve_stream( version: env!("CARGO_PKG_VERSION").to_string(), hostname: hostname(), services: { - let mut s = vec![ServiceKind::Ping, ServiceKind::Info, ServiceKind::Tcp]; - if desktop { - s.push(ServiceKind::Desktop); - } - if policy.sync_dir.is_some() { - s.push(ServiceKind::Sync); - } - s + let enabled = policy.effective_services(); + [ + ServiceKind::Ping, + ServiceKind::Info, + ServiceKind::Tcp, + ServiceKind::Desktop, + ServiceKind::Sync, + ] + .into_iter() + .filter(|k| enabled.contains(k) && (*k != ServiceKind::Desktop || desktop)) + .collect() }, desktop: desktop_caps(desktop), }; diff --git a/crates/rds-agent/src/main.rs b/crates/rds-agent/src/main.rs index ae31e26..a27e4b1 100644 --- a/crates/rds-agent/src/main.rs +++ b/crates/rds-agent/src/main.rs @@ -1,10 +1,9 @@ //! `rds-agent`: daemon on a controlled device. -use std::str::FromStr; - use clap::Parser; -use rds_agent::{Agent, AgentLimits, AgentPolicy}; -use rds_net::EndpointId; +use rds_agent::{ + Agent, AgentLimits, AgentOverrides, AgentPolicy, AgentSettings, Role, ServiceName, +}; use rds_net::{ EndpointOverrides, EndpointSettings, Ticket, acquire_key, bind_endpoint, default_key_path, }; @@ -47,72 +46,94 @@ struct Cli { /// Local UDP bind address; repeatable on the owned backend. #[arg(long)] bind_address: Vec, + /// Versioned agent JSON configuration: role, services, peers, + /// authority, limits and timeouts. Explicit flags override it. + #[arg(long)] + agent_config: Option, + /// Deployment role preset: access | sync | desktop | full. + /// Mutually exclusive with --service. + #[arg(long, conflicts_with = "service")] + role: Option, + /// Data-plane service to serve: tcp | desktop | sync. Repeatable; + /// replaces the implicit set and any configured role. + #[arg(long = "service", conflicts_with = "role")] + service: Vec, + /// Remove a service from the role/implicit set. Repeatable. + #[arg(long = "no-service")] + no_service: Vec, /// Allowed peer EndpointId. Repeatable. #[arg(long = "allow")] allow: Vec, - /// SSH socket the TcpConnect service may reach. - #[arg(long, default_value = "127.0.0.1:22")] - ssh: rds_core::TcpTarget, + /// SSH socket the TcpConnect service may reach (default 127.0.0.1:22). + #[arg(long)] + ssh: Option, /// Permit TcpConnect to any host:port (development only). #[arg(long)] allow_any_tcp: bool, /// Pending handshakes and admitted connections; positive 16-bit limit. - #[arg(long, default_value = "32")] - max_connections: std::num::NonZeroU16, - /// Concurrent tasks per connection. Grant mode needs >=2; one slot is - /// reserved from service bodies for authorization/renewal. - #[arg(long, default_value = "64")] - max_streams: std::num::NonZeroU16, + #[arg(long)] + max_connections: Option, + /// Concurrent tasks per connection (default 64). Grant mode needs >=2; + /// one slot is reserved from service bodies for authorization/renewal. + #[arg(long)] + max_streams: Option, + /// Inbound connection handshake deadline in seconds (1..=3600). + #[arg(long)] + handshake_timeout: Option, + /// Per-stream greeting read deadline in seconds (1..=3600). + #[arg(long)] + hello_timeout: Option, /// Directory HTTP(S) origin or legacy IP:port; the agent publishes its /// signed record and keeps it fresh. #[arg(long)] directory: Option, /// PEM CA bundle for directory HTTPS; replaces the public root store. - #[arg(long, requires = "directory")] + #[arg(long)] directory_ca: Option, /// Trusted registry authority for local-manager device-name resolution. - #[arg(long, requires = "directory")] + #[arg(long)] registry_key: Option, - /// Bootstrap registry authority epoch. - #[arg(long, default_value = "1")] - registry_epoch: u64, + /// Bootstrap registry authority epoch (default 1). + #[arg(long)] + registry_epoch: Option, /// Durable name-trust state; default is beside --key-file. - #[arg(long, requires = "registry_key")] + #[arg(long)] registry_state: Option, /// Registry authority rotation receipt; repeat in epoch order. - #[arg(long, requires = "registry_key")] + #[arg(long)] registry_rotation: Vec, - /// Record TTL when `--directory` is set. - #[arg(long, default_value = "300")] - record_ttl: u64, + /// Record TTL when `--directory` is set (default 300). + #[arg(long)] + record_ttl: Option, /// Private durable publisher state; default is beside --key-file. - #[arg(long, requires = "directory")] + #[arg(long)] record_state: Option, /// Trusted grant issuer (base32 verifying key). Repeatable. When /// set, every connection must present a valid estate-signed grant /// before any service stream opens. #[arg(long = "issuer")] issuers: Vec, - /// Maximum grant lifetime accepted, in seconds. - #[arg(long, default_value = "300")] - grant_ttl: u64, + /// Maximum grant lifetime accepted, in seconds (default 300). + #[arg(long)] + grant_ttl: Option, /// Verifying key that signs the estate revocation snapshot /// (`GET /v1/revocations`). Required for denylist polling when /// `--directory` is set. - #[arg(long, requires_all = ["directory", "issuers"])] + #[arg(long)] revocations_key: Option, - /// Epoch of the independently provisioned bootstrap revocation authority. - #[arg(long, default_value = "1")] - revocations_epoch: u64, + /// Epoch of the independently provisioned bootstrap revocation + /// authority (default 1). + #[arg(long)] + revocations_epoch: Option, /// Private durable policy directory; default is beside --key-file. - #[arg(long, requires = "revocations_key")] + #[arg(long)] revocations_state: Option, /// Dual-signed authority rotation receipt. Repeat in epoch order. - #[arg(long, requires = "revocations_key")] + #[arg(long)] authority_rotation: Vec, - /// Revocation poll interval in seconds. - #[arg(long, default_value = "30")] - revocations_interval: u64, + /// Revocation poll interval in seconds (default 30). + #[arg(long)] + revocations_interval: Option, /// Directory the Sync service may read/write under. #[arg(long)] sync_dir: Option, @@ -125,10 +146,6 @@ async fn main() -> std::process::ExitCode { } async fn run(cli: Cli) -> anyhow::Result<()> { - anyhow::ensure!( - cli.issuers.is_empty() || cli.max_streams.get() >= 2, - "grant mode requires --max-streams at least 2" - ); let prepared_admin = cli.admin.bind().await?; let mut config = cli .endpoint_config @@ -145,6 +162,83 @@ async fn run(cli: Cli) -> anyhow::Result<()> { })? .into_endpoint()?; + // Role/service/authority policy resolves in the same preflight window: + // before an identity is created or a socket bound. + let merged = cli + .agent_config + .as_deref() + .map(AgentSettings::load) + .transpose()? + .unwrap_or_default() + .apply(AgentOverrides { + role: cli.role, + services: cli.service, + no_services: cli.no_service, + ssh: cli.ssh, + allow_any_tcp: cli.allow_any_tcp, + sync_dir: cli.sync_dir, + allow: cli.allow, + issuers: cli.issuers, + grant_ttl: cli.grant_ttl, + directory: cli.directory, + directory_ca: cli.directory_ca, + record_ttl: cli.record_ttl, + record_state: cli.record_state, + registry_key: cli.registry_key, + registry_epoch: cli.registry_epoch, + registry_state: cli.registry_state, + registry_rotations: cli.registry_rotation, + revocations_key: cli.revocations_key, + revocations_epoch: cli.revocations_epoch, + revocations_state: cli.revocations_state, + revocations_rotations: cli.authority_rotation, + revocations_interval: cli.revocations_interval, + max_connections: cli.max_connections, + max_streams: cli.max_streams, + handshake_timeout: cli.handshake_timeout, + hello_timeout: cli.hello_timeout, + }); + merged.validate()?; + // Budget and authority cross-checks run on the merged document before + // any string is decoded, matching the previous flag-only order. + let max_streams = merged + .limits + .max_streams + .unwrap_or(std::num::NonZeroU16::new(64).expect("positive limit")); + anyhow::ensure!( + merged.authority.issuers.is_empty() || max_streams.get() >= 2, + "grant mode requires --max-streams at least 2" + ); + if !merged.authority.issuers.is_empty() && merged.authority.revocations.is_none() { + anyhow::bail!("managed grants require --directory and --revocations-key"); + } + let resolved = merged.resolve()?; + + let ssh_target = resolved + .ssh_target + .unwrap_or_else(|| "127.0.0.1:22".parse().expect("static target parses")); + let (ssh_host, ssh_port) = ssh_target.into_parts(); + + let mut policy = AgentPolicy::ssh_only((ssh_host, ssh_port)); + for target in &resolved.tcp_targets { + let (host, port) = target.clone().into_parts(); + policy.tcp_targets.insert((host, port)); + } + for id in &resolved.allow { + policy.allow.insert(*id); + } + policy.allow_any_tcp = resolved.allow_any_tcp; + policy.grant_max_ttl = std::time::Duration::from_secs(resolved.grant_ttl.unwrap_or(300)); + policy.sync_dir = resolved.sync_dir; + policy.services = resolved.services; + if let Some(timeouts) = resolved.timeouts { + policy.timeouts = timeouts; + } + policy.issuers.extend(resolved.issuers.iter().copied()); + policy + .validate() + .map_err(|why| anyhow::anyhow!("invalid agent policy: {why}"))?; + let key_path = cli .key_file .or_else(default_key_path) @@ -163,47 +257,28 @@ async fn run(cli: Cli) -> anyhow::Result<()> { let identity = tokio::task::spawn_blocking(move || acquire_key(&key_load_path)).await??; let secret_key = identity.secret_key().clone(); - let (ssh_host, ssh_port) = cli.ssh.into_parts(); - - let mut policy = AgentPolicy::ssh_only((ssh_host, ssh_port)); - for id in &cli.allow { - policy.allow.insert(EndpointId::from_str(id)?); - } - policy.allow_any_tcp = cli.allow_any_tcp; - policy.grant_max_ttl = std::time::Duration::from_secs(cli.grant_ttl); - policy.sync_dir = cli.sync_dir; - for s in &cli.issuers { - let bytes = data_encoding::BASE32_NOPAD - .decode(s.to_uppercase().as_bytes()) - .map_err(|e| anyhow::anyhow!("--issuer not base32: {e}"))?; - let raw: [u8; 32] = bytes - .try_into() - .map_err(|_| anyhow::anyhow!("--issuer is not 32 bytes"))?; - policy.issuers.insert(raw); - } - config.secret_key = Some(secret_key.clone()); - let mut directory = cli + let mut directory = resolved .directory .as_deref() .map(|origin| -> anyhow::Result<_> { let mut client = rds_discovery::client::Client::from_endpoint(origin)?; - if let Some(path) = &cli.directory_ca { + if let Some(path) = &resolved.directory_ca { client = client.with_ca_pem(&std::fs::read(path)?)?; } Ok(client) }) .transpose()?; - if let Some(key) = cli.registry_key { + if let Some(key) = resolved.registry_key { let client = directory .take() .ok_or_else(|| anyhow::anyhow!("registry trust requires --directory"))?; - let path = cli + let path = resolved .registry_state .unwrap_or_else(|| key_path.with_extension("registry-state")); - let epoch = cli.registry_epoch; - let rotations = cli.registry_rotation; + let epoch = resolved.registry_epoch.unwrap_or(1); + let rotations = resolved.registry_rotations; directory = Some( tokio::task::spawn_blocking(move || -> anyhow::Result<_> { let authority = rds_discovery::authority::Authority::from_base32(&key, epoch)?; @@ -213,19 +288,18 @@ async fn run(cli: Cli) -> anyhow::Result<()> { .await??, ); } - if !policy.issuers.is_empty() && cli.revocations_key.is_none() { - anyhow::bail!("managed grants require --directory and --revocations-key"); - } let policy_handle = std::sync::Arc::new(policy.clone()); let _revocations = if let (Some(client), Some(key)) = - (directory.clone(), cli.revocations_key.as_deref()) + (directory.clone(), resolved.revocations_key.as_deref()) { - let authority = - rds_discovery::authority::Authority::from_base32(key, cli.revocations_epoch)?; - let path = cli + let authority = rds_discovery::authority::Authority::from_base32( + key, + resolved.revocations_epoch.unwrap_or(1), + )?; + let path = resolved .revocations_state .unwrap_or_else(|| key_path.with_extension("revocations-state")); - let rotations = cli.authority_rotation; + let rotations = resolved.revocations_rotations; let store = tokio::task::spawn_blocking(move || -> Result<_, rds_discovery::DiscoveryError> { let now = rds_discovery::clock::Reading::now()?; @@ -240,14 +314,14 @@ async fn run(cli: Cli) -> anyhow::Result<()> { client, store, policy_handle, - std::time::Duration::from_secs(cli.revocations_interval), + std::time::Duration::from_secs(resolved.revocations_interval.unwrap_or(30)), )?) } else { None }; let record_issuer = if directory.is_some() { - let path = cli + let path = resolved .record_state .unwrap_or_else(|| key_path.with_extension("publisher-state")); let key = ed25519_dalek::SigningKey::from_bytes(&secret_key.to_bytes()); @@ -266,16 +340,29 @@ async fn run(cli: Cli) -> anyhow::Result<()> { // develops independently; the announcer publishes address changes. let mut announce = if let (Some(client), Some(issuer)) = (directory.clone(), record_issuer) { + // The directory record advertises what the policy actually serves: + // Ping is the always-on liveness beacon; data-plane services map + // from the effective set and only when usable in this binary. + let services = policy + .effective_services() + .iter() + .filter_map(|kind| match kind { + rds_core::ServiceKind::Ping => Some(rds_discovery::Service::Ping), + rds_core::ServiceKind::Tcp => Some(rds_discovery::Service::TcpForward), + rds_core::ServiceKind::Desktop if cfg!(feature = "desktop") => { + Some(rds_discovery::Service::Desktop) + } + rds_core::ServiceKind::Sync => Some(rds_discovery::Service::Sync), + _ => None, + }) + .collect(); let announced = rds_net::announce( endpoint.clone(), rds_net::AnnounceConfig { issuer, directory: client, - services: vec![ - rds_discovery::Service::Ping, - rds_discovery::Service::TcpForward, - ], - ttl: std::time::Duration::from_secs(cli.record_ttl), + services, + ttl: std::time::Duration::from_secs(resolved.record_ttl.unwrap_or(300)), }, ); match announced { @@ -290,8 +377,12 @@ async fn run(cli: Cli) -> anyhow::Result<()> { }; let agent = std::sync::Arc::new( - Agent::new(endpoint, policy) - .with_limits(AgentLimits::new(cli.max_connections, cli.max_streams)), + Agent::new(endpoint, policy).with_limits(AgentLimits::new( + resolved + .max_connections + .unwrap_or(std::num::NonZeroU16::new(32).expect("positive limit")), + max_streams, + )), ); let metrics = agent.metrics(); let mut control = diff --git a/crates/rds-agent/src/settings.rs b/crates/rds-agent/src/settings.rs new file mode 100644 index 0000000..a634cbe --- /dev/null +++ b/crates/rds-agent/src/settings.rs @@ -0,0 +1,917 @@ +//! Versioned agent configuration: role, service, peer, authority, limit +//! and timeout policy as one checked document. +//! +//! `AgentSettings` mirrors `EndpointSettings`: a bounded regular file, +//! `schema_version` gated, unknown fields rejected, explicit flags +//! overriding file values. The file carries no secrets — issuers and +//! registry/revocation keys are verifying keys, and state paths are only +//! locations. + +use std::collections::BTreeSet; +use std::io::Read; +use std::num::NonZeroU16; +use std::path::{Path, PathBuf}; +use std::str::FromStr; +use std::time::Duration; + +use rds_core::{ServiceKind, TcpTarget}; +use rds_net::EndpointId; +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::{MAX_TIMEOUT, TimeoutPolicy}; + +pub const AGENT_CONFIG_VERSION: u32 = 1; +pub const MAX_AGENT_CONFIG_BYTES: usize = 16 * 1024; +const MAX_ALLOWLIST: usize = 256; +const MAX_TCP_TARGETS: usize = 64; +const MAX_GRANT_TTL_SECS: u64 = 86_400; + +#[derive(Debug, Error)] +pub enum AgentConfigError { + #[error("agent configuration I/O: {0}")] + Io(#[from] std::io::Error), + #[error("agent configuration JSON: {0}")] + Json(#[from] serde_json::Error), + #[error("invalid agent configuration: {0}")] + Invalid(&'static str), +} + +/// A deployment role preset: which data-plane services the agent serves. +/// `Ping`/`Info` are the always-on control plane in every role. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Role { + /// TCP forwarding only — the historical default. + Access, + /// File synchronization only. + Sync, + /// Desktop sessions only. + Desktop, + /// Every implemented service the binary can serve. + Full, +} + +impl Role { + fn services(self) -> BTreeSet { + match self { + Self::Access => BTreeSet::from([ServiceKind::Tcp]), + Self::Sync => BTreeSet::from([ServiceKind::Sync]), + Self::Desktop => BTreeSet::from([ServiceKind::Desktop]), + Self::Full => { + BTreeSet::from([ServiceKind::Tcp, ServiceKind::Desktop, ServiceKind::Sync]) + } + } + } +} + +impl FromStr for Role { + type Err = AgentConfigError; + + fn from_str(value: &str) -> Result { + match value { + "access" => Ok(Self::Access), + "sync" => Ok(Self::Sync), + "desktop" => Ok(Self::Desktop), + "full" => Ok(Self::Full), + _ => Err(AgentConfigError::Invalid( + "unknown role; expected access, sync, desktop or full", + )), + } + } +} + +/// A gateable data-plane service name in configuration. `ping` and `info` +/// are control-plane and always served; `audio` is wire-reserved but not +/// implemented and is rejected rather than silently advertised. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ServiceName { + Tcp, + Desktop, + Sync, + Audio, +} + +impl ServiceName { + pub fn kind(self) -> ServiceKind { + match self { + Self::Tcp => ServiceKind::Tcp, + Self::Desktop => ServiceKind::Desktop, + Self::Sync => ServiceKind::Sync, + Self::Audio => ServiceKind::Audio, + } + } +} + +impl FromStr for ServiceName { + type Err = AgentConfigError; + + fn from_str(value: &str) -> Result { + match value { + "tcp" => Ok(Self::Tcp), + "desktop" => Ok(Self::Desktop), + "sync" => Ok(Self::Sync), + "audio" => Ok(Self::Audio), + _ => Err(AgentConfigError::Invalid( + "unknown service; expected tcp, desktop, sync or audio", + )), + } + } +} + +/// Per-service knobs. `tcp_targets` extends the permitted destination set +/// beyond the single SSH socket the `--ssh` flag configures. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct ServiceSettings { + pub ssh_target: Option, + pub tcp_targets: Vec, + pub allow_any_tcp: bool, + pub sync_dir: Option, +} + +/// The membership allowlist: peers that may open any stream. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct PeerSettings { + pub allow: Vec, +} + +/// Registry authority (device-name trust) configuration. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct RegistrySettings { + pub key: Option, + pub epoch: Option, + pub state: Option, + pub rotations: Vec, +} + +/// Revocation snapshot authority configuration. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct RevocationSettings { + pub key: Option, + pub epoch: Option, + pub state: Option, + pub interval_secs: Option, + pub rotations: Vec, +} + +/// Grant issuers, TTL and the directory/revocation authority surfaces. +/// Directory-dependent fields require `directory` so a partial authority +/// posture fails loudly instead of being silently unused. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct AuthoritySettings { + /// Trusted grant issuers (base32 Ed25519 verifying keys). + pub issuers: Vec, + pub grant_ttl_secs: Option, + /// Directory HTTP(S) origin or legacy IP:port. + pub directory: Option, + pub directory_ca: Option, + pub record_ttl_secs: Option, + pub record_state: Option, + pub registry: Option, + pub revocations: Option, +} + +/// Admission budgets; absent values keep the built-in defaults. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct LimitSettings { + pub max_connections: Option, + pub max_streams: Option, +} + +/// Connection-admission and stream-greeting deadlines in seconds, each +/// 1..=3600. Service-internal budgets are separate. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct TimeoutSettings { + pub handshake_secs: Option, + pub hello_secs: Option, +} + +/// Agent-level file schema; holds policy identities, never secret key +/// material. Unknown fields and unsupported versions are refused. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AgentSettings { + pub schema_version: u32, + /// Service preset; mutually exclusive with `services`. + #[serde(default)] + pub role: Option, + /// Explicit data-plane service list; mutually exclusive with `role`. + #[serde(default)] + pub services: Option>, + /// Services removed from the role/implicit set after selection. + #[serde(default)] + pub disabled_services: Vec, + #[serde(default)] + pub service: ServiceSettings, + #[serde(default)] + pub peers: PeerSettings, + #[serde(default)] + pub authority: AuthoritySettings, + #[serde(default)] + pub limits: LimitSettings, + #[serde(default)] + pub timeouts: TimeoutSettings, +} + +/// Only explicitly supplied flags override a file; repeated flags replace +/// their whole corresponding list. `--no-service` replaces the file's +/// `disabled_services` list. +#[derive(Debug, Default)] +pub struct AgentOverrides { + pub role: Option, + pub services: Vec, + pub no_services: Vec, + pub ssh: Option, + pub allow_any_tcp: bool, + pub sync_dir: Option, + pub allow: Vec, + pub issuers: Vec, + pub grant_ttl: Option, + pub directory: Option, + pub directory_ca: Option, + pub record_ttl: Option, + pub record_state: Option, + pub registry_key: Option, + pub registry_epoch: Option, + pub registry_state: Option, + pub registry_rotations: Vec, + pub revocations_key: Option, + pub revocations_epoch: Option, + pub revocations_state: Option, + pub revocations_rotations: Vec, + pub revocations_interval: Option, + pub max_connections: Option, + pub max_streams: Option, + pub handshake_timeout: Option, + pub hello_timeout: Option, +} + +/// The merged, typed configuration the binary consumes. +#[derive(Debug)] +pub struct ResolvedAgent { + /// `None` keeps the implicit service set (`services`/`role` absent and + /// nothing disabled); `Some` is the resolved explicit set. + pub services: Option>, + pub ssh_target: Option, + pub tcp_targets: Vec, + pub allow_any_tcp: bool, + pub sync_dir: Option, + pub allow: Vec, + pub issuers: Vec<[u8; 32]>, + pub grant_ttl: Option, + pub directory: Option, + pub directory_ca: Option, + pub record_ttl: Option, + pub record_state: Option, + pub registry_key: Option, + pub registry_epoch: Option, + pub registry_state: Option, + pub registry_rotations: Vec, + pub revocations_key: Option, + pub revocations_epoch: Option, + pub revocations_state: Option, + pub revocations_rotations: Vec, + pub revocations_interval: Option, + pub max_connections: Option, + pub max_streams: Option, + pub timeouts: Option, +} + +fn decode_issuer(encoded: &str) -> Result<[u8; 32], AgentConfigError> { + let bytes = data_encoding::BASE32_NOPAD + .decode(encoded.to_uppercase().as_bytes()) + .map_err(|_| AgentConfigError::Invalid("issuer is not base32"))?; + bytes + .try_into() + .map_err(|_| AgentConfigError::Invalid("issuer is not a 32-byte key")) +} + +fn duplicate_services(list: &[ServiceName]) -> bool { + let mut seen = BTreeSet::new(); + list.iter().any(|s| !seen.insert(*s)) +} + +impl Default for AgentSettings { + fn default() -> Self { + Self { + schema_version: AGENT_CONFIG_VERSION, + role: None, + services: None, + disabled_services: Vec::new(), + service: ServiceSettings::default(), + peers: PeerSettings::default(), + authority: AuthoritySettings::default(), + limits: LimitSettings::default(), + timeouts: TimeoutSettings::default(), + } + } +} + +impl AgentSettings { + pub fn from_json(bytes: &[u8]) -> Result { + if bytes.len() > MAX_AGENT_CONFIG_BYTES { + return Err(AgentConfigError::Invalid("file exceeds 16 KiB")); + } + let settings: Self = serde_json::from_slice(bytes)?; + if settings.schema_version != AGENT_CONFIG_VERSION { + return Err(AgentConfigError::Invalid("unsupported schema_version")); + } + Ok(settings) + } + + pub fn load(path: &Path) -> Result { + use rustix::fs::{Mode, OFlags}; + // Same posture as EndpointSettings::load: a caller-selected path + // may be a symlink; NONBLOCK plus the regular-file check refuses + // FIFOs/devices before the size bound is applied. + let file = std::fs::File::from( + rustix::fs::open( + path, + OFlags::RDONLY | OFlags::NONBLOCK | OFlags::CLOEXEC, + Mode::empty(), + ) + .map_err(std::io::Error::from)?, + ); + if !file.metadata()?.is_file() { + return Err(AgentConfigError::Invalid( + "configuration must be a regular file", + )); + } + let mut bytes = Vec::new(); + file.take(MAX_AGENT_CONFIG_BYTES as u64 + 1) + .read_to_end(&mut bytes)?; + Self::from_json(&bytes) + } + + /// Merge flag overrides. Explicit scalars replace file values; + /// repeated flags replace their whole list. Role and explicit service + /// selection are one surface: whichever the flags supply wins. + pub fn apply(mut self, flags: AgentOverrides) -> Self { + if !flags.services.is_empty() { + self.services = Some(flags.services); + self.role = None; + } else if let Some(role) = flags.role { + self.role = Some(role); + self.services = None; + } + if !flags.no_services.is_empty() { + self.disabled_services = flags.no_services; + } + if let Some(ssh) = flags.ssh { + self.service.ssh_target = Some(ssh.to_string()); + } + self.service.allow_any_tcp |= flags.allow_any_tcp; + if flags.sync_dir.is_some() { + self.service.sync_dir = flags.sync_dir; + } + if !flags.allow.is_empty() { + self.peers.allow = flags.allow; + } + if !flags.issuers.is_empty() { + self.authority.issuers = flags.issuers; + } + if flags.grant_ttl.is_some() { + self.authority.grant_ttl_secs = flags.grant_ttl; + } + if flags.directory.is_some() { + self.authority.directory = flags.directory; + } + if flags.directory_ca.is_some() { + self.authority.directory_ca = flags.directory_ca; + } + if flags.record_ttl.is_some() { + self.authority.record_ttl_secs = flags.record_ttl; + } + if flags.record_state.is_some() { + self.authority.record_state = flags.record_state; + } + if flags.registry_key.is_some() + || flags.registry_epoch.is_some() + || flags.registry_state.is_some() + || !flags.registry_rotations.is_empty() + { + let registry = self.authority.registry.get_or_insert_with(Default::default); + if let Some(key) = flags.registry_key { + registry.key = Some(key); + } + if flags.registry_epoch.is_some() { + registry.epoch = flags.registry_epoch; + } + if flags.registry_state.is_some() { + registry.state = flags.registry_state; + } + if !flags.registry_rotations.is_empty() { + registry.rotations = flags.registry_rotations; + } + } + if flags.revocations_key.is_some() + || flags.revocations_epoch.is_some() + || flags.revocations_state.is_some() + || flags.revocations_interval.is_some() + || !flags.revocations_rotations.is_empty() + { + let revocations = self + .authority + .revocations + .get_or_insert_with(Default::default); + if let Some(key) = flags.revocations_key { + revocations.key = Some(key); + } + if flags.revocations_epoch.is_some() { + revocations.epoch = flags.revocations_epoch; + } + if flags.revocations_state.is_some() { + revocations.state = flags.revocations_state; + } + if flags.revocations_interval.is_some() { + revocations.interval_secs = flags.revocations_interval; + } + if !flags.revocations_rotations.is_empty() { + revocations.rotations = flags.revocations_rotations; + } + } + if flags.max_connections.is_some() { + self.limits.max_connections = flags.max_connections; + } + if flags.max_streams.is_some() { + self.limits.max_streams = flags.max_streams; + } + if flags.handshake_timeout.is_some() { + self.timeouts.handshake_secs = flags.handshake_timeout; + } + if flags.hello_timeout.is_some() { + self.timeouts.hello_secs = flags.hello_timeout; + } + self + } + + /// Structural checks on the merged document: mutual exclusion, + /// duplicates, bounds and cross-field requirements. Typed parsing of + /// addresses and keys happens in [`AgentSettings::resolve`]. + pub fn validate(&self) -> Result<(), AgentConfigError> { + if self.role.is_some() && self.services.is_some() { + return Err(AgentConfigError::Invalid( + "role and services are mutually exclusive", + )); + } + if let Some(services) = &self.services + && duplicate_services(services) + { + return Err(AgentConfigError::Invalid("duplicate service name")); + } + if duplicate_services(&self.disabled_services) { + return Err(AgentConfigError::Invalid("duplicate disabled service")); + } + if self + .services + .iter() + .flatten() + .chain(self.disabled_services.iter()) + .any(|s| *s == ServiceName::Audio) + { + return Err(AgentConfigError::Invalid( + "audio service is reserved but not implemented", + )); + } + if self.service.tcp_targets.len() > MAX_TCP_TARGETS { + return Err(AgentConfigError::Invalid( + "at most 64 permitted tcp targets", + )); + } + if self.peers.allow.len() > MAX_ALLOWLIST { + return Err(AgentConfigError::Invalid("at most 256 allowed peers")); + } + let authority = &self.authority; + if authority + .grant_ttl_secs + .is_some_and(|secs| secs == 0 || secs > MAX_GRANT_TTL_SECS) + { + return Err(AgentConfigError::Invalid( + "grant_ttl_secs must be 1..=86400", + )); + } + if authority.directory.is_none() + && (authority.directory_ca.is_some() + || authority.record_ttl_secs.is_some() + || authority.record_state.is_some() + || authority.registry.is_some() + || authority.revocations.is_some()) + { + return Err(AgentConfigError::Invalid( + "registry, revocations and record settings require a directory", + )); + } + if let Some(registry) = &authority.registry + && registry.key.is_none() + { + return Err(AgentConfigError::Invalid("registry requires a key")); + } + if let Some(revocations) = &authority.revocations { + if revocations.key.is_none() { + return Err(AgentConfigError::Invalid("revocations requires a key")); + } + if authority.issuers.is_empty() { + return Err(AgentConfigError::Invalid( + "revocations require at least one grant issuer", + )); + } + if matches!(revocations.interval_secs, Some(0)) { + return Err(AgentConfigError::Invalid( + "revocations interval_secs must be positive", + )); + } + } + for secs in [self.timeouts.handshake_secs, self.timeouts.hello_secs] + .into_iter() + .flatten() + { + if secs == 0 || secs > MAX_TIMEOUT.as_secs() { + return Err(AgentConfigError::Invalid( + "timeouts must be between 1 and 3600 seconds", + )); + } + } + Ok(()) + } + + /// Lower the validated document into typed values. The resolved set is + /// explicit when a role, a service list or a disabled entry selects it; + /// otherwise `None` preserves the implicit runtime set. + pub fn resolve(&self) -> Result { + let sync_dir = self.service.sync_dir.clone(); + let explicit: Option> = if let Some(services) = &self.services { + Some(services.iter().map(|s| s.kind()).collect()) + } else { + self.role.map(|role| role.services()) + }; + let disabled: BTreeSet = + self.disabled_services.iter().map(|s| s.kind()).collect(); + let services: Option> = match (explicit, disabled.is_empty()) { + (Some(set), _) => Some(set.difference(&disabled).copied().collect()), + (None, false) => { + let mut implicit = BTreeSet::from([ServiceKind::Tcp]); + if cfg!(feature = "desktop") { + implicit.insert(ServiceKind::Desktop); + } + if sync_dir.is_some() { + implicit.insert(ServiceKind::Sync); + } + Some(implicit.difference(&disabled).copied().collect()) + } + (None, true) => None, + }; + // An explicit enabled set must name services this build and + // configuration can actually serve. + if let Some(set) = &services { + if set.contains(&ServiceKind::Desktop) && !cfg!(feature = "desktop") { + return Err(AgentConfigError::Invalid( + "desktop service requires the `desktop` build feature", + )); + } + if set.contains(&ServiceKind::Sync) && sync_dir.is_none() { + return Err(AgentConfigError::Invalid( + "sync service requires a configured sync directory", + )); + } + } + + let ssh_target = self + .service + .ssh_target + .as_deref() + .map(|s| { + TcpTarget::from_str(s) + .map_err(|_| AgentConfigError::Invalid("invalid ssh_target destination")) + }) + .transpose()?; + let mut tcp_targets = Vec::with_capacity(self.service.tcp_targets.len()); + for target in &self.service.tcp_targets { + tcp_targets.push( + TcpTarget::from_str(target) + .map_err(|_| AgentConfigError::Invalid("invalid tcp_targets destination"))?, + ); + } + let mut allow = Vec::with_capacity(self.peers.allow.len()); + for peer in &self.peers.allow { + allow.push( + EndpointId::from_str(peer) + .map_err(|_| AgentConfigError::Invalid("invalid peer endpoint id"))?, + ); + } + let mut issuers = Vec::with_capacity(self.authority.issuers.len()); + for issuer in &self.authority.issuers { + issuers.push(decode_issuer(issuer)?); + } + let authority = &self.authority; + let registry = authority.registry.as_ref(); + let revocations = authority.revocations.as_ref(); + let timeouts = + if self.timeouts.handshake_secs.is_some() || self.timeouts.hello_secs.is_some() { + let default = TimeoutPolicy::default(); + Some(TimeoutPolicy { + handshake: self + .timeouts + .handshake_secs + .map_or(default.handshake, Duration::from_secs), + hello: self + .timeouts + .hello_secs + .map_or(default.hello, Duration::from_secs), + }) + } else { + None + }; + Ok(ResolvedAgent { + services, + ssh_target, + tcp_targets, + allow_any_tcp: self.service.allow_any_tcp, + sync_dir, + allow, + issuers, + grant_ttl: authority.grant_ttl_secs, + directory: authority.directory.clone(), + directory_ca: authority.directory_ca.clone(), + record_ttl: authority.record_ttl_secs, + record_state: authority.record_state.clone(), + registry_key: registry.and_then(|r| r.key.clone()), + registry_epoch: registry.and_then(|r| r.epoch), + registry_state: registry.and_then(|r| r.state.clone()), + registry_rotations: registry.map_or_else(Vec::new, |r| r.rotations.clone()), + revocations_key: revocations.and_then(|r| r.key.clone()), + revocations_epoch: revocations.and_then(|r| r.epoch), + revocations_state: revocations.and_then(|r| r.state.clone()), + revocations_rotations: revocations.map_or_else(Vec::new, |r| r.rotations.clone()), + revocations_interval: revocations.and_then(|r| r.interval_secs), + max_connections: self.limits.max_connections, + max_streams: self.limits.max_streams, + timeouts, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeSet; + + fn parse(json: &str) -> AgentSettings { + AgentSettings::from_json(json.as_bytes()).expect("parse") + } + + fn resolve_ok(json: &str) -> ResolvedAgent { + let settings = parse(json); + settings.validate().expect("validate"); + settings.resolve().expect("resolve") + } + + #[test] + fn version_and_unknown_fields_are_refused() { + assert!(AgentSettings::from_json(br#"{"schema_version":2}"#).is_err()); + assert!(AgentSettings::from_json(br#"{"schema_version":1,"mystery":1}"#).is_err()); + assert!(AgentSettings::from_json(br#"{"role":"access"}"#).is_err()); + assert!(AgentSettings::from_json(b"[]").is_err()); + } + + #[test] + fn role_and_services_are_mutually_exclusive() { + let settings = parse(r#"{"schema_version":1,"role":"sync","services":["tcp"]}"#); + assert!(settings.validate().is_err()); + } + + #[test] + fn duplicate_and_reserved_services_rejected() { + for json in [ + r#"{"schema_version":1,"services":["tcp","tcp"]}"#, + r#"{"schema_version":1,"services":["audio"]}"#, + r#"{"schema_version":1,"disabled_services":["audio"]}"#, + r#"{"schema_version":1,"role":"desktop","disabled_services":["sync","sync"]}"#, + ] { + let settings = parse(json); + assert!(settings.validate().is_err(), "{json}"); + } + } + + #[test] + fn roles_resolve_to_service_sets() { + let check = |json: &str| resolve_ok(json).services.expect("explicit set"); + assert_eq!( + check(r#"{"schema_version":1,"role":"access"}"#), + BTreeSet::from([ServiceKind::Tcp]) + ); + assert_eq!( + check(r#"{"schema_version":1,"role":"sync","service":{"sync_dir":"/tmp/x"}}"#), + BTreeSet::from([ServiceKind::Sync]) + ); + // Sync without its directory refuses to resolve. + assert!( + parse(r#"{"schema_version":1,"services":["sync"]}"#) + .resolve() + .is_err() + ); + // Nothing selected and nothing disabled keeps the implicit set. + assert_eq!(resolve_ok(r#"{"schema_version":1}"#).services, None); + } + + #[cfg(feature = "desktop")] + #[test] + fn desktop_roles_resolve_on_desktop_builds() { + let check = |json: &str| resolve_ok(json).services.expect("explicit set"); + assert_eq!( + check( + r#"{"schema_version":1,"role":"full","disabled_services":["sync"], + "service":{"sync_dir":"/tmp/x"}}"#, + ), + BTreeSet::from([ServiceKind::Tcp, ServiceKind::Desktop]) + ); + assert_eq!( + check(r#"{"schema_version":1,"services":["desktop","tcp"]}"#), + BTreeSet::from([ServiceKind::Desktop, ServiceKind::Tcp]) + ); + } + + #[cfg(not(feature = "desktop"))] + #[test] + fn desktop_roles_refuse_on_non_desktop_builds() { + for json in [ + r#"{"schema_version":1,"role":"desktop"}"#, + r#"{"schema_version":1,"services":["desktop","tcp"]}"#, + r#"{"schema_version":1,"role":"full","service":{"sync_dir":"/tmp/x"}}"#, + ] { + assert!(parse(json).resolve().is_err(), "{json}"); + } + } + + #[test] + fn disabled_carves_the_implicit_set() { + // With no role/services, disabled entries make the set explicit: + // implicit {tcp} on non-desktop builds; sync joins only when a + // sync_dir is configured. + let resolved = resolve_ok( + r#"{"schema_version":1,"disabled_services":["tcp"],"service":{"sync_dir":"/tmp/x"}}"#, + ); + let set = resolved.services.expect("explicit after disable"); + assert!(!set.contains(&ServiceKind::Tcp)); + assert!(set.contains(&ServiceKind::Sync)); + } + + #[test] + fn authority_requires_directory_and_issuers() { + let bad = [ + r#"{"schema_version":1,"authority":{"revocations":{"key":"ab"}}}"#, + r#"{"schema_version":1,"authority":{"directory":"http://localhost:9","revocations":{"key":"ab"}}}"#, + r#"{"schema_version":1,"authority":{"registry":{}}}"#, + r#"{"schema_version":1,"authority":{"directory_ca":"/ca.pem"}}"#, + r#"{"schema_version":1,"authority":{"record_ttl_secs":30}}"#, + ]; + for json in bad { + let settings = parse(json); + assert!(settings.validate().is_err(), "{json}"); + } + // A complete revocation authority section validates. + let ok = parse( + r#"{"schema_version":1,"authority":{"directory":"http://localhost:9", + "issuers":["aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"], + "revocations":{"key":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}}}"#, + ); + ok.validate().expect("complete authority validates"); + let resolved = ok.resolve().expect("resolve"); + assert_eq!(resolved.issuers.len(), 1); + assert_eq!( + resolved.revocations_key.as_deref(), + Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb") + ); + } + + #[test] + fn bounds_are_enforced() { + let bad = [ + r#"{"schema_version":1,"timeouts":{"handshake_secs":0}}"#, + r#"{"schema_version":1,"timeouts":{"hello_secs":3601}}"#, + r#"{"schema_version":1,"authority":{"grant_ttl_secs":0}}"#, + r#"{"schema_version":1,"authority":{"grant_ttl_secs":86401}}"#, + r#"{"schema_version":1,"authority":{"directory":"http://localhost:9","issuers":["a"],"revocations":{"key":"k","interval_secs":0}}}"#, + ]; + for json in bad { + let settings = parse(json); + assert!(settings.validate().is_err(), "{json}"); + } + // Zero limits decode as a JSON error through NonZero. + assert!( + AgentSettings::from_json(br#"{"schema_version":1,"limits":{"max_streams":0}}"#) + .is_err() + ); + } + + #[test] + fn typed_parsing_happens_in_resolve() { + for json in [ + r#"{"schema_version":1,"service":{"ssh_target":"host:0"}}"#, + r#"{"schema_version":1,"service":{"tcp_targets":[":22"]}}"#, + r#"{"schema_version":1,"peers":{"allow":["not-an-id"]}}"#, + r#"{"schema_version":1,"authority":{"directory":"http://localhost:9","issuers":["not-base32"],"revocations":{"key":"k"}}}"#, + ] { + let settings = parse(json); + settings.validate().expect("structure is valid"); + assert!(settings.resolve().is_err(), "{json}"); + } + let resolved = resolve_ok( + r#"{"schema_version":1,"service":{"ssh_target":"10.0.0.7:2222", + "tcp_targets":["127.0.0.1:8080"],"allow_any_tcp":true}}"#, + ); + assert_eq!(resolved.ssh_target.unwrap().to_string(), "10.0.0.7:2222"); + assert_eq!(resolved.tcp_targets.len(), 1); + assert!(resolved.allow_any_tcp); + } + + #[test] + fn flag_overrides_merge_over_file() { + // An explicit --service list replaces both file role and list. + let settings = parse(r#"{"schema_version":1,"role":"sync"}"#).apply(AgentOverrides { + services: vec![ServiceName::Tcp], + ..Default::default() + }); + let resolved = settings.resolve().unwrap(); + assert_eq!(resolved.services, Some(BTreeSet::from([ServiceKind::Tcp]))); + + // --no-service replaces the file's whole disabled list. + let settings = parse( + r#"{"schema_version":1,"role":"full","disabled_services":["desktop"]}"#, + ) + .apply(AgentOverrides { + no_services: vec![ServiceName::Sync], + ..Default::default() + }); + assert_eq!(settings.disabled_services, vec![ServiceName::Sync]); + + // Scalar flags only apply when present; bools are additive-true. + let settings = parse( + r#"{"schema_version":1,"service":{"ssh_target":"1.2.3.4:22","allow_any_tcp":false}, + "authority":{"grant_ttl_secs":60},"timeouts":{"hello_secs":5}}"#, + ) + .apply(AgentOverrides { + grant_ttl: Some(120), + ..Default::default() + }); + let resolved = settings.resolve().unwrap(); + assert_eq!(resolved.ssh_target.unwrap().to_string(), "1.2.3.4:22"); + assert!(!resolved.allow_any_tcp); + assert_eq!(resolved.grant_ttl, Some(120)); + assert_eq!(resolved.timeouts.unwrap().hello, Duration::from_secs(5)); + + // A scalar flag beats the file value. + let settings = parse(r#"{"schema_version":1,"service":{"ssh_target":"1.2.3.4:22"}}"#) + .apply(AgentOverrides { + ssh: Some("9.9.9.9:22".parse().unwrap()), + ..Default::default() + }); + assert_eq!( + settings.resolve().unwrap().ssh_target.unwrap().to_string(), + "9.9.9.9:22" + ); + } + + #[test] + fn nested_authority_flags_merge_per_field() { + // A flag registry key preserves the file's registry state path — + // the same partial-merge convention owned-relay limits use. + let settings = parse( + r#"{"schema_version":1,"authority":{"directory":"http://localhost:9", + "registry":{"key":"old","state":"/tmp/reg-state"}}}"#, + ) + .apply(AgentOverrides { + registry_key: Some("new".into()), + ..Default::default() + }); + let resolved = settings.resolve().unwrap(); + assert_eq!(resolved.registry_key.as_deref(), Some("new")); + assert_eq!( + resolved.registry_state.as_deref(), + Some(Path::new("/tmp/reg-state")) + ); + } + + #[test] + fn timeouts_resolve_to_policy() { + let resolved = + resolve_ok(r#"{"schema_version":1,"timeouts":{"handshake_secs":9,"hello_secs":3}}"#); + let policy = resolved.timeouts.unwrap(); + assert_eq!(policy.handshake, Duration::from_secs(9)); + assert_eq!(policy.hello, Duration::from_secs(3)); + // Partial timeout config preserves the other default. + let resolved = resolve_ok(r#"{"schema_version":1,"timeouts":{"hello_secs":3}}"#); + assert_eq!( + resolved.timeouts.unwrap().handshake, + TimeoutPolicy::default().handshake + ); + assert_eq!(resolve_ok(r#"{"schema_version":1}"#).timeouts, None); + } +} diff --git a/crates/rds-agent/tests/configuration.rs b/crates/rds-agent/tests/configuration.rs index 254b113..0330af9 100644 --- a/crates/rds-agent/tests/configuration.rs +++ b/crates/rds-agent/tests/configuration.rs @@ -36,3 +36,75 @@ fn grant_mode_requires_control_capacity_before_identity_or_bind() { assert!(String::from_utf8_lossy(&output.stderr).contains("--max-streams at least 2")); assert!(!dir.0.join("endpoint.key").exists()); } + +fn run_agent_config(dir: &Scratch, json: &str, args: &[&str]) -> std::process::Output { + let path = dir.0.join("agent.json"); + std::fs::write(&path, json).unwrap(); + let mut full = vec!["--no-relay", "--agent-config"]; + full.push(path.to_str().unwrap()); + full.extend_from_slice(args); + run(&full, dir) +} + +#[test] +fn agent_config_rejects_bad_schema_before_identity_or_bind() { + for json in [ + r#"{"schema_version":2}"#, + r#"{"schema_version":1,"mystery":true}"#, + r#"{"schema_version":1,"role":"sync","services":["tcp"]}"#, + r#"{"schema_version":1,"services":["audio"]}"#, + r#"{"schema_version":1,"services":["sync"]}"#, // no sync_dir + r#"{"schema_version":1,"timeouts":{"hello_secs":0}}"#, + r#"{"schema_version":1,"authority":{"grant_ttl_secs":0}}"#, + r#"{"schema_version":1,"authority":{"revocations":{"key":"ab"}}}"#, + r#"{"schema_version":1,"service":{"ssh_target":"host:0"}}"#, + r#"{"schema_version":1,"peers":{"allow":["not-an-id"]}}"#, + ] { + let dir = Scratch::new(); + let output = run_agent_config(&dir, json, &[]); + assert!(!output.status.success(), "{json}"); + assert!(!dir.0.join("endpoint.key").exists(), "{json}"); + } +} + +#[test] +fn service_flags_gate_before_identity_or_bind() { + // `audio` is reserved in every build; `sync` without a directory and + // out-of-bounds timeouts fail identically from flags or file. + for args in [ + &["--service", "audio"][..], + &["--service", "sync"][..], + &["--no-relay", "--role", "sync"][..], + &["--no-relay", "--hello-timeout", "0"][..], + &["--no-relay", "--handshake-timeout", "7200"][..], + ] { + let dir = Scratch::new(); + let output = run(args, &dir); + assert!(!output.status.success(), "{args:?}"); + assert!(!dir.0.join("endpoint.key").exists(), "{args:?}"); + } +} + +#[test] +fn flag_role_overrides_file_services() { + // The file enables tcp; the flag role reselects sync, which then fails + // its sync_dir prerequisite — proving the flag replaced the file set. + let dir = Scratch::new(); + let output = run_agent_config( + &dir, + r#"{"schema_version":1,"services":["tcp"]}"#, + &["--role", "sync"], + ); + assert!(!output.status.success()); + assert!(String::from_utf8_lossy(&output.stderr).contains("sync")); + assert!(!dir.0.join("endpoint.key").exists()); +} + +#[test] +fn role_and_service_flags_conflict() { + let output = run( + &["--no-relay", "--role", "access", "--service", "tcp"], + &Scratch::new(), + ); + assert!(!output.status.success()); +} diff --git a/crates/rds-agent/tests/grant_scopes.rs b/crates/rds-agent/tests/grant_scopes.rs index b4c328f..8c2e316 100644 --- a/crates/rds-agent/tests/grant_scopes.rs +++ b/crates/rds-agent/tests/grant_scopes.rs @@ -227,9 +227,15 @@ async fn directional_sync(backend: Backend) { let ack = read_frame::<_, rds_core::HelloAck>(&mut recv) .await .unwrap(); - assert!( - matches!(ack, rds_core::HelloAck::Error { message } if message == "service Desktop not granted") - ); + // The deployment gate answers first on builds where desktop is not + // compiled; on desktop builds the grant scope check produces the + // refusal. Either way the stream is refused by name. + let desktop_refusal = if cfg!(feature = "desktop") { + "service Desktop not granted" + } else { + "service Desktop not enabled on this agent" + }; + assert!(matches!(ack, rds_core::HelloAck::Error { message } if message == desktop_refusal)); // The per-session greeting must not bypass the same scope check. let (mut send, mut recv) = conn.open_bi().await.unwrap(); write_frame( @@ -249,9 +255,7 @@ async fn directional_sync(backend: Backend) { let ack = read_frame::<_, rds_core::HelloAck>(&mut recv) .await .unwrap(); - assert!( - matches!(ack, rds_core::HelloAck::Error { message } if message == "service Desktop not granted") - ); + assert!(matches!(ack, rds_core::HelloAck::Error { message } if message == desktop_refusal)); assert!( rds_client::open_sync(&conn) .await diff --git a/crates/rds-agent/tests/service_policy.rs b/crates/rds-agent/tests/service_policy.rs new file mode 100644 index 0000000..1f9b93a --- /dev/null +++ b/crates/rds-agent/tests/service_policy.rs @@ -0,0 +1,214 @@ +//! Deployment service policy enforced by a real agent over loopback QUIC: +//! the explicit service set gates streams, `Info` advertises only what is +//! enabled, and greeting deadlines come from the timeout policy. +use std::collections::BTreeSet; +use std::sync::Arc; +use std::time::Duration; +use std::{net::SocketAddr, path::PathBuf}; + +use rds_agent::{Agent, AgentPolicy, TimeoutPolicy}; +use rds_core::{HelloAck, ServiceKind, StreamHello}; +use rds_net::{Backend, Endpoint, EndpointConfig, read_frame, write_frame}; + +struct Scratch(PathBuf); +impl Scratch { + fn new() -> Self { + use std::os::unix::fs::DirBuilderExt; + static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let path = std::env::temp_dir().canonicalize().unwrap().join(format!( + "rds-service-policy-{}-{}", + std::process::id(), + NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + )); + std::fs::DirBuilder::new() + .mode(0o700) + .create(&path) + .unwrap(); + Self(path) + } +} +impl Drop for Scratch { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +struct Runner(tokio::task::JoinHandle<()>); +impl Drop for Runner { + fn drop(&mut self) { + self.0.abort(); + } +} + +async fn endpoint(backend: Backend) -> Endpoint { + rds_net::bind_endpoint(EndpointConfig { + backend, + discovery: false, + bind_addrs: vec!["127.0.0.1:0".parse().unwrap()], + ..Default::default() + }) + .await + .unwrap() +} + +fn backends() -> Vec { + #[cfg(feature = "transport-noq")] + { + vec![Backend::Iroh, Backend::Noq] + } + #[cfg(not(feature = "transport-noq"))] + { + vec![Backend::Iroh] + } +} + +async fn error_ack(conn: &rds_net::Connection, hello: StreamHello) -> String { + let (mut send, mut recv) = conn.open_bi().await.unwrap(); + write_frame(&mut send, &hello).await.unwrap(); + match tokio::time::timeout(Duration::from_secs(5), read_frame(&mut recv)) + .await + .unwrap() + .unwrap() + { + HelloAck::Error { message } => message, + other => panic!("expected refusal, got {other:?}"), + } +} + +/// The explicit service set is the only surface a client can reach: `Info` +/// advertises it honestly, enabled services work, everything else is +/// refused by name before grant or service machinery runs. +async fn explicit_service_set(backend: Backend) { + let root = Scratch::new(); + let client = endpoint(backend).await; + let server = endpoint(backend).await; + // A reachable TCP target proves the enabled service still works. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let tcp_port = match listener.local_addr().unwrap() { + SocketAddr::V4(addr) => addr.port(), + SocketAddr::V6(_) => panic!("expected v4"), + }; + + let mut policy = AgentPolicy::ssh_only(("127.0.0.1".into(), tcp_port)); + policy.allow.insert(client.id()); + // Configured but not enabled: the explicit set wins over inference. + policy.sync_dir = Some(root.0.clone()); + policy.services = Some(BTreeSet::from([ServiceKind::Tcp])); + + let agent = Arc::new(Agent::new(server.clone(), policy)); + let runner = agent.clone(); + let _task = Runner(tokio::spawn(async move { + runner.run().await.unwrap(); + })); + let conn = rds_client::connect(&client, server.addr()).await.unwrap(); + + // The Info advertisement is exactly the enabled set, never more. + let info = rds_client::info(&conn).await.unwrap(); + assert_eq!( + info.services, + vec![ServiceKind::Ping, ServiceKind::Info, ServiceKind::Tcp] + ); + + rds_client::ping(&conn, rand::random()).await.unwrap(); + let (send, _recv) = rds_client::open_tcp(&conn, "127.0.0.1", tcp_port) + .await + .expect("enabled tcp service is served"); + drop(send); + drop(listener); + + // Every disabled service refuses by name — including sync, which is + // configured on disk but not enabled, and audio, which is reserved. + for hello in [ + StreamHello::Sync, + StreamHello::SyncTransferV2 { id: rand::random() }, + StreamHello::Audio(rds_core::AudioHello { + codec: rds_core::AudioCodec::Opus, + sample_rate: 48_000, + channels: 2, + }), + ] { + let message = error_ack(&conn, hello).await; + assert!(message.contains("not enabled"), "{message}"); + } +} + +/// A disabled service is refused by deployment policy before the grant +/// machinery runs — the peer hears "not enabled", not "grant required". +async fn disabled_precedes_grants(backend: Backend) { + let client = endpoint(backend).await; + let server = endpoint(backend).await; + let mut policy = AgentPolicy::ssh_only(("127.0.0.1".into(), 9)); + policy.allow.insert(client.id()); + policy.issuers.insert([5; 32]); + policy.use_local_revocations(); + policy.services = Some(BTreeSet::from([ServiceKind::Tcp])); + + let agent = Arc::new(Agent::new(server.clone(), policy)); + let runner = agent.clone(); + let _task = Runner(tokio::spawn(async move { + runner.run().await.unwrap(); + })); + let conn = rds_client::connect(&client, server.addr()).await.unwrap(); + + let message = error_ack(&conn, StreamHello::Sync).await; + assert!(message.contains("not enabled"), "{message}"); + + // The enabled service still fails closed on the missing grant. + let message = error_ack( + &conn, + StreamHello::TcpConnect { + host: "127.0.0.1".into(), + port: 9, + }, + ) + .await; + assert!(message.contains("grant"), "{message}"); +} + +/// The greeting deadline is policy, not a fixed constant: a silent stream +/// is cut loose inside the configured hello timeout. +async fn hello_deadline_is_policy(backend: Backend) { + let client = endpoint(backend).await; + let server = endpoint(backend).await; + let mut policy = AgentPolicy::ssh_only(("127.0.0.1".into(), 9)); + policy.allow.insert(client.id()); + policy.timeouts = TimeoutPolicy { + hello: Duration::from_millis(200), + ..Default::default() + }; + + let agent = Arc::new(Agent::new(server.clone(), policy)); + let runner = agent.clone(); + let _task = Runner(tokio::spawn(async move { + runner.run().await.unwrap(); + })); + let conn = rds_client::connect(&client, server.addr()).await.unwrap(); + + let (_send, mut recv) = conn.open_bi().await.unwrap(); + // The server never writes a refusal for a silent greeting; the task + // dies and the stream ends well inside the default 15s. + let outcome = + tokio::time::timeout(Duration::from_secs(5), read_frame::<_, HelloAck>(&mut recv)).await; + assert!(matches!(outcome, Ok(Err(_)) | Err(_)), "{outcome:?}"); +} + +#[tokio::test] +async fn explicit_service_set_gates_streams() { + for backend in backends() { + explicit_service_set(backend).await; + } +} + +#[tokio::test] +async fn disabled_service_precedes_grant_requirement() { + for backend in backends() { + disabled_precedes_grants(backend).await; + } +} + +#[tokio::test] +async fn hello_deadline_comes_from_policy() { + for backend in backends() { + hello_deadline_is_policy(backend).await; + } +} diff --git a/crates/rds-core/src/lib.rs b/crates/rds-core/src/lib.rs index 92bf41a..ad8587f 100644 --- a/crates/rds-core/src/lib.rs +++ b/crates/rds-core/src/lib.rs @@ -132,7 +132,7 @@ pub struct AgentInfo { pub desktop: Option, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] pub enum ServiceKind { Ping, Info, diff --git a/docs/agent-configuration.md b/docs/agent-configuration.md new file mode 100644 index 0000000..d3193b2 --- /dev/null +++ b/docs/agent-configuration.md @@ -0,0 +1,101 @@ +# Agent configuration + +`rds-agent` accepts `--agent-config FILE`: a versioned JSON document covering +the agent's role, data-plane services, peer allowlist, authority posture, +admission limits and timeout policy. It is the policy counterpart of +[`--endpoint-config`](endpoint-configuration.md), which stays limited to the +endpoint transport. The file carries no secrets — issuers and +registry/revocation entries are verifying keys, and `state`/`ca` fields are +only filesystem locations. + +## Precedence and validation + +Precedence is built-in defaults, then the file, then explicitly supplied +flags — the same convention as the endpoint configuration. Repeated list +flags (`--allow`, `--issuer`, rotation receipts, `--service`, +`--no-service`) replace the file's whole corresponding list; an absent flag +preserves the file value. A flag selecting a role or service set replaces +whichever the file configured. + +`schema_version` is required and equals 1. Files are limited to 16 KiB, +must be regular files, and unknown fields, duplicate fields, unknown +variants and unsupported versions are rejected. Structural validation — +mutual exclusion, cross-field requirements, numeric bounds — runs before +identity creation or socket binding, alongside the endpoint preflight. + +## Sections + +| Field | Meaning / bound | +|---|---| +| `schema_version` | Required integer 1 | +| `role` | `access` \| `sync` \| `desktop` \| `full`; mutually exclusive with `services` | +| `services` | Explicit data-plane list `["tcp","desktop","sync"]` | +| `disabled_services` | Services subtracted from the resolved set | +| `service` | `ssh_target`, `tcp_targets`, `allow_any_tcp`, `sync_dir` | +| `peers` | `allow`: endpoint-id strings, at most 256 | +| `authority` | `issuers`, `grant_ttl_secs` (1–86400), `directory`, `directory_ca`, `record_ttl_secs`, `record_state`, `registry`, `revocations` | +| `limits` | `max_connections`, `max_streams`; positive 16-bit | +| `timeouts` | `handshake_secs`, `hello_secs`; each 1–3600 | + +`registry` holds `key` (required when present), `epoch`, `state` and +`rotations`. `revocations` holds `key` (required when present), `epoch`, +`state`, `interval_secs` and `rotations`. Registry, revocation and record +fields all require `authority.directory`; revocations additionally require +at least one issuer — the same requirements the flags carried, now +enforced on the merged document so file and flag sources mix freely. + +## Service enablement + +`Ping` and `Info` are the always-on control plane and are never gated. +The gateable data-plane services are `tcp`, `desktop` and `sync`; `audio` +is wire-reserved but unimplemented and is rejected rather than silently +accepted. + +- **No `role`/`services`/`disabled_services`** — the implicit set: `tcp`, + plus `desktop` when the build has the `desktop` feature, plus `sync` + when a `sync_dir` is configured. +- **`role`** — `access` = `{tcp}`, `sync` = `{sync}`, `desktop` = + `{desktop}`, `full` = `{tcp, desktop, sync}`. +- **`services`** — exactly the listed services. +- **`disabled_services`** — subtracted from whatever the role, list or + implicit set resolved to. + +An enabled service whose prerequisites are missing fails validation: +`desktop` requires the `desktop` build feature, `sync` requires a +configured `sync_dir`. A stream greeting for a disabled service is refused +with `service not enabled on this agent` before any grant or +per-service machinery runs — the deployment gate answers first, ahead of +grant scope checks. `Info` advertises exactly the served set, and the +directory record publishes the same list (Ping plus enabled data-plane +services). + +`service.tcp_targets` permits extra `TcpConnect` destinations beyond the +single SSH socket; the flag surface has no equivalent list flag. + +## Timeout policy + +`timeouts.handshake_secs` bounds the inbound connection handshake; +`timeouts.hello_secs` bounds the `StreamHello` read on every new stream. +Both default to 15 seconds and accept 1–3600. Flags `--handshake-timeout` +and `--hello-timeout` override file values. Per-service budgets (frame +streams, transfer deadlines, renewal windows) remain service-internal and +are not set from this file; global timeout classes are W2.6 work. + +## Example + +[Access role](../examples/agent-access.json) keeps the historical default: +TCP forwarding to the configured SSH socket, allowlisted peers, default +timeouts spelled out explicitly. A sync-only deployment looks like: + +```json +{ + "schema_version": 1, + "role": "sync", + "service": { "sync_dir": "/srv/sync" }, + "peers": { "allow": [""] }, + "timeouts": { "handshake_secs": 10 } +} +``` + +which is equivalent to `rds-agent --role sync --sync-dir /srv/sync --allow + --handshake-timeout 10`. diff --git a/docs/capability-matrix.md b/docs/capability-matrix.md index e06fb15..8efe23c 100644 --- a/docs/capability-matrix.md +++ b/docs/capability-matrix.md @@ -20,13 +20,19 @@ below; `README.md` must link this file. ## Services (agent `Info` advertisement and grant scopes) +The data-plane services are gated by deployment policy — `role`, `services` +and `disabled_services` in [agent configuration](agent-configuration.md), or +`--role`/`--service`/`--no-service` flags. A disabled service is refused by +name before grant machinery runs, and `Info`/directory announcements list +only what is actually served. `ping`/`info` are the always-on control plane. + | Capability | State | Runtime prerequisites | Evidence | |---|---|---|---| | service:ping | implemented | peer on `--allow` list | `rds-net` tests, CI | | service:info | implemented | peer on `--allow` list | `rds-cli` tests, CI | -| service:tcp | implemented | `--allow` + grant scope for tcp | `rds-ssh` e2e, CI | -| service:desktop | experimental | `desktop` build flag; usable capture backend; grant scope | capture→encode→decode→stats contract tests; no viewer | -| service:sync | implemented | `--sync-dir` configured | `rds-sync` tests, resumable transfer tests | +| service:tcp | implemented | enabled + `--allow` + grant scope for tcp | `rds-ssh` e2e, CI | +| service:desktop | experimental | enabled + `desktop` build flag; usable capture backend; grant scope | capture→encode→decode→stats contract tests; no viewer | +| service:sync | implemented | enabled + `--sync-dir` configured | `rds-sync` tests, resumable transfer tests | | service:audio | stub | no codec; wire shape reserved in v2 | `ServiceKind::Audio` variant only | | service:sync-read | implemented | grant scope on sync root | grant/scope tests | | service:sync-write | implemented | grant scope on sync root | grant/scope tests | diff --git a/docs/deployment.md b/docs/deployment.md index ee6b0ec..df9f4cb 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -45,6 +45,10 @@ local binaries together; old binaries and copied/manual-replaced keys are outsid this cooperative guarantee. See [migration and ownership](local-sessions.md). The relay's `--allow` covers every endpoint that may use the relay. +Agent role, service, peer, authority, limit and timeout policy can live in a +versioned JSON file instead of flags — `--agent-config /etc/rds/agent.json`; +see [agent configuration](agent-configuration.md). + ## Ports and firewall | Port | Proto | Service | Exposure | diff --git a/docs/endpoint-configuration.md b/docs/endpoint-configuration.md index 1004aca..0c8a0f1 100644 --- a/docs/endpoint-configuration.md +++ b/docs/endpoint-configuration.md @@ -1,10 +1,10 @@ # Endpoint configuration Both `rds` and `rds-agent` accept `--endpoint-config FILE`. This versioned JSON -file controls the endpoint transport. It does not contain the endpoint secret, -grant issuer, registry/revocation authority, directory credentials or service -policy; their existing provisioning options remain separate. Unified role-level -policy and negotiated session-budget configuration is still W2.1 work. +file controls the endpoint transport. It does not contain the endpoint secret +or service policy; role, service, peer, authority, limit and timeout policy +live in the versioned agent configuration — see +[agent configuration](agent-configuration.md). Connectivity commands now use the local agent by default. Put transport/directory configuration on that agent; managed CLI commands reject those flags. Explicit @@ -27,8 +27,8 @@ variants, trailing JSON and unsupported schema versions are rejected. The selected backend must be compiled into the binary; `noq` requires the `transport-noq` feature. Syntax is parsed before overrides, and the resulting combination is validated before creating a key file, opening policy state or -binding the endpoint. This preflight applies to endpoint settings, not yet every -role-specific service option. +binding the endpoint. The same preflight window applies to the merged agent +configuration (role/service/authority/timeout sections). | Field | Meaning / bound | |---|---| diff --git a/docs/remediation-progress.md b/docs/remediation-progress.md index 3b2e92c..7895d4e 100644 --- a/docs/remediation-progress.md +++ b/docs/remediation-progress.md @@ -45,7 +45,7 @@ Neither increment closes these product gaps or any wave. | W1.8 | Implemented; Linux checks passed | Directory-relative no-follow journal/destination I/O and a held source file replace path-check-then-open. Link planting and substitutions after open are tested. Native macOS verification remains pending. | | W1.9 | Partial | Pull path and Done-root binding, exact frame decoding, canonical Need, batch bounds, requested/unique chunks, verified completion, actual wire-byte accounting and absolute session budgets are implemented. Managed transfers now run the negotiated v2 session: per-transfer route IDs bound into every frame, Hello/HelloAck limit negotiation, typed Cancel both directions. Stronger physical cancellation barriers and native macOS qualification remain open. | | W1.10 | Partial; Linux transaction checks passed | Root and destination-parent locks cover overlapping roots and filesystem aliases. Reserved private staging, both-parent sync and bounded known-name recovery are implemented. 23 transaction/cleanup boundaries cover process exit and two returned-error classes. Physical power loss, native macOS, large-file campaign and inactive/legacy journal collection remain open. | -| W2.1 | Partial; endpoint settings checked on Linux | Shared version-1 endpoint JSON, explicit file/flag precedence, typed backend/relay validation, preflight before identity creation, owned-relay CLI/agent selection and canonical TCP targets shared with client and agent policy are implemented. Role-level service/authority settings and timeout policy remain open. | +| W2.1 | Implemented; endpoint + agent settings checked on Linux | Shared version-1 endpoint JSON, explicit file/flag precedence, typed backend/relay validation and preflight before identity creation are implemented. The version-1 agent JSON now carries role/service/peers/authority/limits/timeouts with the same precedence and preflight; `--role`/`--service`/`--no-service` select the gateable service set, disabled services are refused by name ahead of grant machinery, `Info` and directory announcements advertise exactly the served set, and handshake/hello deadlines come from `TimeoutPolicy`. See [agent configuration](agent-configuration.md). | | W2.2 | Partial; exact ALPN selection + sync/desktop session routing | Immutable per-protocol TLS offers prevent silent fallback and concurrent request interference. Managed single-file transfers use fresh control/uni routing IDs and negotiate version/limits in the `SyncTransferV2` session envelope before any filesystem operation. Desktop sessions mint random per-session IDs in `StreamHello::DesktopV2` and route frames through `UniHello::DesktopFrames { id }`, isolating stale streams and allowing concurrent sessions; the same display grant scope check covers both greetings. Service-wide capability negotiation remains open. | | W2.3 | Partial; destination-bound renewable grants and directional scopes | Grant v2 adds a strict signature domain, audience and stable session ID across positive lease revisions. Same-scope renewal preserves streams/revocation, retains one replay slot/watchdog and enforces wall/continuous expiry. Explicit managed renewal uses IPC v3 and a control-completion barrier. `SyncRead`/`SyncWrite` and `DesktopView`/`DesktopControl` are enforced before filesystem/input operations. Tenant/policy binding, per-path/account scopes and automatic GDS issuer integration remain open. See [contract](grant-leases.md). | | W2.4 | Partial; default connectivity manager | Agent local control is enabled by default; ordinary ticket/ping/info/SSH/forward/send/recv commands and keyless `rds session` reuse its endpoint. Same-UID IPC, pinned streams, cancellation and aggregate metrics are implemented. Agent/direct CLI/owned relay acquire exclusive ownership of a validated seed inode. Viewer manager APIs, coordinated installed-binary migration, native macOS and real multi-user/relay qualification remain open. See [contract](local-sessions.md) and [migration receipt](reports/rds-identity-migration-20260925.md). | @@ -1932,3 +1932,36 @@ Default-feature builds stay clean (`#[cfg(feature = "desktop")]` on the agent-side route derivation). Open: service-wide capability negotiation (W2.2 remainder), global media budgets (W2.5/W8), native macOS and real-network qualification. + +## Unified agent role/service/authority configuration (2026-09-27) + +W2.1 closes its remaining scope — role-level service/authority settings and +timeout policy — as a version-1 `AgentSettings` JSON loaded with +`--agent-config`, mirroring the endpoint configuration contract: bounded +regular file, `schema_version` gated, unknown fields rejected, explicit +flags overriding file values (repeated flags replacing whole lists), and +structural validation inside the same preflight window before identity or +sockets exist. + +`AgentPolicy::services` makes the served set explicit: `Ping`/`Info` remain +the always-on control plane; `tcp`, `desktop` and `sync` are gateable; +`audio` stays wire-reserved and is refused at configuration time rather +than advertised. With no selection, the implicit set matches prior +behavior (tcp + desktop-if-compiled + sync-if-configured). A stream for a +disabled service is refused by name before grant or per-service work runs; +`Info` and the directory announcement advertise exactly the effective set. +Role presets (`access`/`sync`/`desktop`/`full`), per-service sections, +authority posture (issuers, grant TTL, directory/registry/revocations) and +a bounded `TimeoutPolicy` (handshake and greeting deadlines, 1–3600 s) +round out the document; global timeout classes remain W2.6 scope. + +Coverage: `settings` unit tests pin the schema contract (version, unknown +fields, mutual exclusion, bounds, cross-field authority requirements) and +the merge semantics (flag-over-file scalars, list replacement, nested +authority field merge, implicit/explicit/disabled service resolution, +desktop build gating). Binary-level tests assert every failure mode exits +before identity or bind, including flag-file precedence. `service_policy` +e2e proves the gate on a real agent: an explicit `{tcp}` set refuses +sync/audio by name despite a configured sync root, `Info` lists exactly +the enabled set, disabled-service refusal precedes grant requirements, +and the greeting deadline follows the configured timeout. diff --git a/examples/agent-access.json b/examples/agent-access.json new file mode 100644 index 0000000..1f38596 --- /dev/null +++ b/examples/agent-access.json @@ -0,0 +1,7 @@ +{ + "schema_version": 1, + "role": "access", + "service": { "ssh_target": "127.0.0.1:22" }, + "peers": { "allow": [""] }, + "timeouts": { "handshake_secs": 15, "hello_secs": 15 } +}