10 Commits
23 changed files with 724 additions and 105 deletions

No files matched your search

Generated
+5 -5
View File
@@ -3624,7 +3624,7 @@ dependencies = [
[[package]] [[package]]
name = "wakey" name = "wakey"
version = "0.3.2" version = "0.3.3"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"chrono", "chrono",
@@ -3645,7 +3645,7 @@ dependencies = [
[[package]] [[package]]
name = "wakey-agent" name = "wakey-agent"
version = "0.3.2" version = "0.3.3"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"clap", "clap",
@@ -3670,7 +3670,7 @@ dependencies = [
[[package]] [[package]]
name = "wakey-control-plane" name = "wakey-control-plane"
version = "0.3.2" version = "0.3.3"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"axum", "axum",
@@ -3699,7 +3699,7 @@ dependencies = [
[[package]] [[package]]
name = "wakey-core" name = "wakey-core"
version = "0.3.2" version = "0.3.3"
dependencies = [ dependencies = [
"macaddr", "macaddr",
"serde", "serde",
@@ -3711,7 +3711,7 @@ dependencies = [
[[package]] [[package]]
name = "wakey-linux" name = "wakey-linux"
version = "0.3.2" version = "0.3.3"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"futures", "futures",
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "wakey" name = "wakey"
version = "0.3.2" version = "0.3.3"
edition = "2024" edition = "2024"
publish = ["gitea"] publish = ["gitea"]
@@ -80,6 +80,8 @@ The agent keeps its terminal manager outside the control-WebSocket reconnect loo
Closing a session must terminate the entire PTY process group, not only the shell process. Cleanup should send a hangup first, then escalate to termination and forced kill after bounded grace periods. Explicit operator close, absolute session timeout, agent process exit, and machine reboot close the session. Closing a session must terminate the entire PTY process group, not only the shell process. Cleanup should send a hangup first, then escalate to termination and forced kill after bounded grace periods. Explicit operator close, absolute session timeout, agent process exit, and machine reboot close the session.
The agent configuration sets `terminal.session_ttl_seconds` for newly created PTYs. The default is 43,200 seconds (12 hours); explicit zero disables automatic expiry. The agent owns and enforces this deadline locally because PTYs survive browser and control-plane disconnection. It advertises the policy to the control plane and includes the creation-time TTL in terminal inventory, allowing CC to mirror expiry and preserve it across CC restart. Missing fields from older agents retain the 12-hour default. Positive values are exact and values that cannot fit the platform timer are rejected rather than clamped.
An agent process restart or machine reboot still ends every terminal session because PTY file descriptors cannot be recovered by a new process. Surviving that boundary would require an external session host such as `tmux` or a separate terminal daemon and is not part of this design. An agent process restart or machine reboot still ends every terminal session because PTY file descriptors cannot be recovered by a new process. Surviving that boundary would require an external session host such as `tmux` or a separate terminal daemon and is not part of this design.
The agent reports a normal exit status or terminating signal when available. The UI must distinguish an exited shell from a transport failure. The agent reports a normal exit status or terminating signal when available. The UI must distinguish an exited shell from a transport failure.
+11
View File
@@ -110,6 +110,16 @@ update_existing_repo() {
git -C "$repo_dir" pull --ff-only origin "$ref" || fail 'git pull --ff-only failed' git -C "$repo_dir" pull --ff-only origin "$ref" || fail 'git pull --ff-only failed'
} }
update_submodules() {
repo_dir=$1
# Sync first so an updated .gitmodules URL is honored by an existing
# checkout, then initialize and pin every nested submodule to the commit
# recorded by the parent repository.
git -C "$repo_dir" submodule sync --recursive || fail 'git submodule sync failed'
git -C "$repo_dir" submodule update --init --recursive || fail 'git submodule update failed'
}
install_executable_force() { install_executable_force() {
src=$1 src=$1
dst=$2 dst=$2
@@ -186,6 +196,7 @@ main() {
git clone --depth 1 --branch "$REF" "$REPO_URL" "$WORKDIR" || git clone --depth 1 --branch "$REF" "$REPO_URL" "$WORKDIR" ||
fail 'git clone failed' fail 'git clone failed'
fi fi
update_submodules "$WORKDIR"
cd "$WORKDIR" cd "$WORKDIR"
+4
View File
@@ -3,12 +3,16 @@ export type Agent = {
connected: boolean; connected: boolean;
nickname?: string | null; nickname?: string | null;
capabilities: "terminal"[]; capabilities: "terminal"[];
capability_options?: {
terminal?: { max_sessions: number };
};
}; };
export type TerminalSession = { export type TerminalSession = {
terminal_id: string; terminal_id: string;
agent_id: string; agent_id: string;
created_at_unix: number; created_at_unix: number;
expires_at_unix?: number | null;
agent_attached: boolean; agent_attached: boolean;
operator_attached: boolean; operator_attached: boolean;
websocket_url: string; websocket_url: string;
+27 -9
View File
@@ -67,18 +67,31 @@ function mergeTerminalSession(
sessions: TerminalSession[], sessions: TerminalSession[],
next: TerminalSession, next: TerminalSession,
): TerminalSession[] { ): TerminalSession[] {
const remaining = sessions.filter( const existingIndex = sessions.findIndex(
(item) => item.terminal_id !== next.terminal_id, (item) => item.terminal_id === next.terminal_id,
); );
return [...remaining, next].sort( if (existingIndex >= 0) {
(left, right) => left.created_at_unix - right.created_at_unix, // Attaching updates session state, not identity or tab position. Removing
// and re-inserting here made same-second sessions swap places on click.
return sessions.map((item, index) =>
index === existingIndex ? next : item,
);
}
return orderTerminalSessions([...sessions, next]);
}
function compareTerminalSessions(
left: TerminalSession,
right: TerminalSession,
): number {
return (
left.created_at_unix - right.created_at_unix ||
left.terminal_id.localeCompare(right.terminal_id)
); );
} }
function orderTerminalSessions(sessions: TerminalSession[]): TerminalSession[] { function orderTerminalSessions(sessions: TerminalSession[]): TerminalSession[] {
return [...sessions].sort( return [...sessions].sort(compareTerminalSessions);
(left, right) => left.created_at_unix - right.created_at_unix,
);
} }
function reconcileTerminalSessions( function reconcileTerminalSessions(
@@ -156,7 +169,12 @@ export function TerminalPage({
const selectedAgentSessionCount = sessions.filter( const selectedAgentSessionCount = sessions.filter(
(item) => item.agent_id === selectedAgentId, (item) => item.agent_id === selectedAgentId,
).length; ).length;
const agentAtSessionLimit = selectedAgentSessionCount >= 2; const selectedAgentSessionLimit = Math.max(
1,
selectedAgent?.capability_options?.terminal?.max_sessions ?? 2,
);
const agentAtSessionLimit =
selectedAgentSessionCount >= selectedAgentSessionLimit;
const canRequestStart = const canRequestStart =
selectedAgent?.connected && selectedAgent?.connected &&
selectedAgent.capabilities.includes("terminal") && selectedAgent.capabilities.includes("terminal") &&
@@ -744,7 +762,7 @@ export function TerminalPage({
</TooltipTrigger> </TooltipTrigger>
<TooltipContent> <TooltipContent>
{agentAtSessionLimit {agentAtSessionLimit
? "Two-session limit reached" ? `${selectedAgentSessionLimit}-session limit reached`
: "New session"} : "New session"}
</TooltipContent> </TooltipContent>
</Tooltip> </Tooltip>
+3
View File
@@ -365,6 +365,9 @@
gap: 0.4rem; gap: 0.4rem;
height: 1.4rem; height: 1.4rem;
padding-inline: 0.45rem; padding-inline: 0.45rem;
border-color: rgb(255 255 255 / 14%);
background: rgb(255 255 255 / 5%);
color: #cbd5e1;
font-size: 0.6875rem; font-size: 0.6875rem;
text-transform: capitalize; text-transform: capitalize;
} }
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "wakey-agent" name = "wakey-agent"
version = "0.3.2" version = "0.3.3"
edition = "2024" edition = "2024"
publish = ["gitea"] publish = ["gitea"]
+94 -3
View File
@@ -4,6 +4,11 @@ use std::fmt;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::LazyLock; use std::sync::LazyLock;
use crate::protocol::{
DEFAULT_TERMINAL_MAX_SESSIONS, DEFAULT_TERMINAL_SESSION_TTL_SECONDS,
checked_terminal_session_ttl,
};
pub const DEFAULT_CONFIG_PATH: &str = "/etc/wakey-agent/config.toml"; pub const DEFAULT_CONFIG_PATH: &str = "/etc/wakey-agent/config.toml";
pub const DEFAULT_PID_FILE: &str = "/var/run/wakey-agent.pid"; pub const DEFAULT_PID_FILE: &str = "/var/run/wakey-agent.pid";
const WAKEY_DHCP_LEASES_ENV: &str = "WAKEY_DHCP_LEASES"; const WAKEY_DHCP_LEASES_ENV: &str = "WAKEY_DHCP_LEASES";
@@ -40,13 +45,21 @@ pub struct AgentConfig {
} }
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct TerminalConfig { pub struct TerminalConfig {
#[serde(default)] #[serde(default)]
pub enabled: bool, pub enabled: bool,
#[serde(default = "default_terminal_shell")] #[serde(default = "default_terminal_shell")]
pub shell: PathBuf, pub shell: PathBuf,
#[serde(default)]
pub args: Vec<String>,
#[serde(default)]
pub current_dir: Option<PathBuf>,
#[serde(default = "default_terminal_max_sessions")] #[serde(default = "default_terminal_max_sessions")]
pub max_sessions: usize, pub max_sessions: usize,
/// Maximum lifetime of a PTY process. Zero disables automatic expiry.
#[serde(default = "default_terminal_session_ttl_seconds")]
pub session_ttl_seconds: u64,
} }
impl Default for TerminalConfig { impl Default for TerminalConfig {
@@ -54,7 +67,10 @@ impl Default for TerminalConfig {
Self { Self {
enabled: false, enabled: false,
shell: default_terminal_shell(), shell: default_terminal_shell(),
args: Vec::new(),
current_dir: None,
max_sessions: default_terminal_max_sessions(), max_sessions: default_terminal_max_sessions(),
session_ttl_seconds: default_terminal_session_ttl_seconds(),
} }
} }
} }
@@ -136,7 +152,17 @@ fn default_terminal_shell() -> PathBuf {
} }
const fn default_terminal_max_sessions() -> usize { const fn default_terminal_max_sessions() -> usize {
2 DEFAULT_TERMINAL_MAX_SESSIONS
}
const fn default_terminal_session_ttl_seconds() -> u64 {
DEFAULT_TERMINAL_SESSION_TTL_SECONDS
}
impl TerminalConfig {
pub(crate) fn session_ttl(&self) -> Result<Option<std::time::Duration>> {
checked_terminal_session_ttl(self.session_ttl_seconds).map_err(anyhow::Error::msg)
}
} }
impl AgentConfig { impl AgentConfig {
@@ -161,8 +187,10 @@ pub fn apply_local_path_env_to_command(cmd: &mut std::process::Command, config:
pub fn load_config(path: &Path) -> Result<AgentConfig> { pub fn load_config(path: &Path) -> Result<AgentConfig> {
let content = std::fs::read_to_string(path) let content = std::fs::read_to_string(path)
.with_context(|| format!("failed to read agent config {}", path.display()))?; .with_context(|| format!("failed to read agent config {}", path.display()))?;
toml::from_str(&content) let config: AgentConfig = toml::from_str(&content)
.with_context(|| format!("failed to parse agent config {}", path.display())) .with_context(|| format!("failed to parse agent config {}", path.display()))?;
config.terminal.session_ttl()?;
Ok(config)
} }
pub fn save_config(path: &Path, config: &AgentConfig) -> Result<()> { pub fn save_config(path: &Path, config: &AgentConfig) -> Result<()> {
@@ -249,7 +277,10 @@ mod tests {
terminal: TerminalConfig { terminal: TerminalConfig {
enabled: true, enabled: true,
shell: "/bin/sh".into(), shell: "/bin/sh".into(),
args: vec!["-l".into()],
current_dir: Some("/tmp".into()),
max_sessions: 2, max_sessions: 2,
session_ttl_seconds: 600,
}, },
}; };
@@ -284,4 +315,64 @@ agent_token = "secret"
); );
assert_eq!(config.terminal, TerminalConfig::default()); assert_eq!(config.terminal, TerminalConfig::default());
} }
#[test]
fn existing_terminal_config_defaults_optional_command_fields() {
let config: AgentConfig = toml::from_str(
r#"
server_url = "https://example.com"
agent_id = "agent-1"
agent_token = "secret"
[terminal]
enabled = true
shell = "/bin/sh"
max_sessions = 2
"#,
)
.expect("existing terminal config should parse");
assert!(config.terminal.args.is_empty());
assert!(config.terminal.current_dir.is_none());
assert_eq!(
config.terminal.session_ttl_seconds,
DEFAULT_TERMINAL_SESSION_TTL_SECONDS
);
}
#[test]
fn terminal_config_rejects_misspelled_fields() {
let error = toml::from_str::<AgentConfig>(
r#"
server_url = "https://example.com"
agent_id = "agent-1"
agent_token = "secret"
[terminal]
enabled = true
max_session = 67
"#,
)
.expect_err("unknown terminal fields must not silently use defaults");
assert!(error.to_string().contains("max_session"));
}
#[test]
fn terminal_ttl_zero_is_unlimited_and_positive_is_exact() {
let mut terminal = TerminalConfig {
session_ttl_seconds: 0,
..TerminalConfig::default()
};
assert_eq!(terminal.session_ttl().expect("unlimited TTL"), None);
terminal.session_ttl_seconds = 67;
assert_eq!(
terminal.session_ttl().expect("finite TTL"),
Some(std::time::Duration::from_secs(67))
);
terminal.session_ttl_seconds = u64::MAX;
assert!(terminal.session_ttl().is_err());
}
} }
+3
View File
@@ -154,7 +154,10 @@ mod tests {
terminal: crate::config::TerminalConfig { terminal: crate::config::TerminalConfig {
enabled: true, enabled: true,
shell: "/bin/ash".into(), shell: "/bin/ash".into(),
args: vec!["-l".into()],
current_dir: Some("/root".into()),
max_sessions: 2, max_sessions: 2,
session_ttl_seconds: 600,
}, },
}; };
+102
View File
@@ -2,6 +2,7 @@ use macaddr::MacAddr;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::fmt; use std::fmt;
use std::net::IpAddr; use std::net::IpAddr;
use std::time::{Duration, Instant};
use wakey_core::parse::mac; use wakey_core::parse::mac;
use wakey_core::{ use wakey_core::{
Device, DeviceInventory, DhcpLeaseWithState, InterfaceSummary, InventoryQuery, Device, DeviceInventory, DhcpLeaseWithState, InterfaceSummary, InventoryQuery,
@@ -65,10 +66,59 @@ pub enum AgentCapability {
Terminal, Terminal,
} }
pub const DEFAULT_TERMINAL_MAX_SESSIONS: usize = 2;
pub const DEFAULT_TERMINAL_SESSION_TTL_SECONDS: u64 = 12 * 60 * 60;
/// Validates a terminal TTL against the platform's monotonic timer range.
/// Zero is the explicit unlimited policy.
pub fn checked_terminal_session_ttl(
session_ttl_seconds: u64,
) -> Result<Option<Duration>, &'static str> {
if session_ttl_seconds == 0 {
return Ok(None);
}
let ttl = Duration::from_secs(session_ttl_seconds);
Instant::now()
.checked_add(ttl)
.ok_or("terminal session TTL is too large for the platform timer")?;
Ok(Some(ttl))
}
const fn default_terminal_session_ttl_seconds() -> u64 {
DEFAULT_TERMINAL_SESSION_TTL_SECONDS
}
/// Optional parameters attached to advertised agent capabilities.
///
/// Keep this separate from `AgentCapability`: the capability list remains a
/// compact, backward-compatible feature check, while this object can grow as
/// individual capabilities gain configurable limits or modes.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct AgentCapabilityOptions {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub terminal: Option<TerminalCapabilityOptions>,
}
impl AgentCapabilityOptions {
fn is_empty(&self) -> bool {
self.terminal.is_none()
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TerminalCapabilityOptions {
pub max_sessions: usize,
#[serde(default = "default_terminal_session_ttl_seconds")]
pub session_ttl_seconds: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AgentTerminalSession { pub struct AgentTerminalSession {
pub terminal_id: TerminalId, pub terminal_id: TerminalId,
pub created_at_unix: u64, pub created_at_unix: u64,
/// The policy captured when this PTY was created. Zero means unlimited.
#[serde(default = "default_terminal_session_ttl_seconds")]
pub session_ttl_seconds: u64,
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -238,6 +288,8 @@ pub enum ClientMessage {
agent_id: String, agent_id: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")] #[serde(default, skip_serializing_if = "Vec::is_empty")]
capabilities: Vec<AgentCapability>, capabilities: Vec<AgentCapability>,
#[serde(default, skip_serializing_if = "AgentCapabilityOptions::is_empty")]
capability_options: AgentCapabilityOptions,
}, },
Auth { Auth {
agent_id: String, agent_id: String,
@@ -384,6 +436,7 @@ mod tests {
sessions: vec![AgentTerminalSession { sessions: vec![AgentTerminalSession {
terminal_id: TerminalId::new("term-1").expect("terminal id"), terminal_id: TerminalId::new("term-1").expect("terminal id"),
created_at_unix: 42, created_at_unix: 42,
session_ttl_seconds: 600,
}], }],
}; };
let json = serde_json::to_string(&inventory).expect("serialize terminal inventory"); let json = serde_json::to_string(&inventory).expect("serialize terminal inventory");
@@ -397,4 +450,53 @@ mod tests {
let json = serde_json::to_string(&resume).expect("serialize terminal resume"); let json = serde_json::to_string(&resume).expect("serialize terminal resume");
assert!(json.contains("\"type\":\"resume_terminal\"")); assert!(json.contains("\"type\":\"resume_terminal\""));
} }
#[test]
fn hello_serializes_typed_capability_options() {
let message = ClientMessage::Hello {
agent_id: "router".into(),
capabilities: vec![AgentCapability::Terminal],
capability_options: AgentCapabilityOptions {
terminal: Some(TerminalCapabilityOptions {
max_sessions: 3,
session_ttl_seconds: 600,
}),
},
};
let value = serde_json::to_value(message).expect("serialize hello");
assert_eq!(value["capability_options"]["terminal"]["max_sessions"], 3);
assert_eq!(
value["capability_options"]["terminal"]["session_ttl_seconds"],
600
);
}
#[test]
fn terminal_ttl_defaults_for_old_messages_and_preserves_explicit_zero() {
let old_options: TerminalCapabilityOptions =
serde_json::from_value(serde_json::json!({ "max_sessions": 2 }))
.expect("deserialize old capability options");
assert_eq!(
old_options.session_ttl_seconds,
DEFAULT_TERMINAL_SESSION_TTL_SECONDS
);
let old_session: AgentTerminalSession = serde_json::from_value(serde_json::json!({
"terminal_id": "term-old",
"created_at_unix": 42
}))
.expect("deserialize old terminal inventory");
assert_eq!(
old_session.session_ttl_seconds,
DEFAULT_TERMINAL_SESSION_TTL_SECONDS
);
let unlimited: TerminalCapabilityOptions = serde_json::from_value(serde_json::json!({
"max_sessions": 2,
"session_ttl_seconds": 0
}))
.expect("deserialize unlimited capability options");
assert_eq!(unlimited.session_ttl_seconds, 0);
}
} }
+13 -1
View File
@@ -8,7 +8,10 @@ use tracing::{debug, error, info, info_span, warn};
use crate::config::AgentConfig; use crate::config::AgentConfig;
use crate::dispatch::{dispatch_command, inventory_for_config}; use crate::dispatch::{dispatch_command, inventory_for_config};
use crate::protocol::{AgentCapability, AgentCommand, ClientMessage, ErrorPayload, ServerMessage}; use crate::protocol::{
AgentCapability, AgentCapabilityOptions, AgentCommand, ClientMessage, ErrorPayload,
ServerMessage, TerminalCapabilityOptions,
};
use crate::terminal::TerminalManager; use crate::terminal::TerminalManager;
pub async fn run(config: AgentConfig) -> Result<()> { pub async fn run(config: AgentConfig) -> Result<()> {
@@ -82,6 +85,15 @@ async fn run_once(
&ClientMessage::Hello { &ClientMessage::Hello {
agent_id: config.agent_id.clone(), agent_id: config.agent_id.clone(),
capabilities: agent_capabilities(config), capabilities: agent_capabilities(config),
capability_options: AgentCapabilityOptions {
terminal: config
.terminal
.enabled
.then_some(TerminalCapabilityOptions {
max_sessions: config.terminal.max_sessions.max(1),
session_ttl_seconds: config.terminal.session_ttl_seconds,
}),
},
}, },
) )
.await?; .await?;
+67 -3
View File
@@ -52,6 +52,7 @@ struct ActiveTerminal {
cancel: Option<oneshot::Sender<()>>, cancel: Option<oneshot::Sender<()>>,
relay_credentials: mpsc::UnboundedSender<String>, relay_credentials: mpsc::UnboundedSender<String>,
created_at_unix: u64, created_at_unix: u64,
session_ttl_seconds: u64,
} }
/// Owns PTY workers independently of any individual control-plane connection. /// Owns PTY workers independently of any individual control-plane connection.
@@ -91,6 +92,7 @@ impl TerminalManager {
anyhow::bail!("terminal capability is disabled"); anyhow::bail!("terminal capability is disabled");
} }
validate_size(rows, cols)?; validate_size(rows, cols)?;
let session_ttl = config.terminal.session_ttl()?;
let terminal_key = terminal_id.to_string(); let terminal_key = terminal_id.to_string();
let (cancel_tx, cancel_rx) = oneshot::channel(); let (cancel_tx, cancel_rx) = oneshot::channel();
@@ -113,6 +115,7 @@ impl TerminalManager {
cancel: Some(cancel_tx), cancel: Some(cancel_tx),
relay_credentials: relay_tx.clone(), relay_credentials: relay_tx.clone(),
created_at_unix, created_at_unix,
session_ttl_seconds: config.terminal.session_ttl_seconds,
}, },
); );
} }
@@ -122,7 +125,16 @@ impl TerminalManager {
let active = Arc::downgrade(&self.active); let active = Arc::downgrade(&self.active);
let events = self.events.clone(); let events = self.events.clone();
tokio::spawn(async move { tokio::spawn(async move {
let result = run_terminal(&config, &terminal_id, rows, cols, cancel_rx, relay_rx).await; let result = run_terminal(
&config,
&terminal_id,
rows,
cols,
session_ttl,
cancel_rx,
relay_rx,
)
.await;
// Inventory is authoritative. Remove the stopped worker before // Inventory is authoritative. Remove the stopped worker before
// notifying the control session, which may immediately resync it. // notifying the control session, which may immediately resync it.
remove_completed(&active, terminal_id.as_str()); remove_completed(&active, terminal_id.as_str());
@@ -176,6 +188,7 @@ impl TerminalManager {
.map(|terminal_id| AgentTerminalSession { .map(|terminal_id| AgentTerminalSession {
terminal_id, terminal_id,
created_at_unix: active.created_at_unix, created_at_unix: active.created_at_unix,
session_ttl_seconds: active.session_ttl_seconds,
}) })
}) })
.collect() .collect()
@@ -210,11 +223,14 @@ async fn run_terminal(
terminal_id: &TerminalId, terminal_id: &TerminalId,
rows: u16, rows: u16,
cols: u16, cols: u16,
session_ttl: Option<Duration>,
mut cancel: oneshot::Receiver<()>, mut cancel: oneshot::Receiver<()>,
mut relay_credentials: mpsc::UnboundedReceiver<String>, mut relay_credentials: mpsc::UnboundedReceiver<String>,
) -> Result<()> { ) -> Result<()> {
let terminal = match wakey::wakey_linux::terminal::TerminalPty::spawn( let terminal = match wakey::wakey_linux::terminal::TerminalPty::spawn(
Path::new(&config.terminal.shell), Path::new(&config.terminal.shell),
&config.terminal.args,
config.terminal.current_dir.as_deref(),
rows, rows,
cols, cols,
) { ) {
@@ -225,7 +241,13 @@ async fn run_terminal(
}; };
let (mut reader, mut writer, mut child) = terminal.into_parts(); let (mut reader, mut writer, mut child) = terminal.into_parts();
let process_group = child.id(); let process_group = child.id();
info!(terminal_id = %terminal_id, shell = %config.terminal.shell.display(), "terminal PTY ready"); info!(
terminal_id = %terminal_id,
program = %config.terminal.shell.display(),
args = ?config.terminal.args,
current_dir = ?config.terminal.current_dir,
"terminal PTY ready"
);
let (relay_input_tx, mut relay_input_rx) = mpsc::channel(RELAY_INPUT_QUEUE); let (relay_input_tx, mut relay_input_rx) = mpsc::channel(RELAY_INPUT_QUEUE);
let mut relay_output: Option<mpsc::Sender<Message>> = None; let mut relay_output: Option<mpsc::Sender<Message>> = None;
@@ -234,8 +256,10 @@ async fn run_terminal(
let mut terminal_state = TerminalState::new(rows, cols); let mut terminal_state = TerminalState::new(rows, cols);
let mut output = [0_u8; 16 * 1024]; let mut output = [0_u8; 16 * 1024];
let mut requested_close = false; let mut requested_close = false;
let mut ttl_expired = false;
let mut observed_status = None; let mut observed_status = None;
let mut drain_deadline: Option<std::pin::Pin<Box<tokio::time::Sleep>>> = None; let mut drain_deadline: Option<std::pin::Pin<Box<tokio::time::Sleep>>> = None;
let mut ttl_deadline = session_ttl.map(|ttl| Box::pin(tokio::time::sleep(ttl)));
loop { loop {
tokio::select! { tokio::select! {
biased; biased;
@@ -243,6 +267,15 @@ async fn run_terminal(
requested_close = true; requested_close = true;
break; break;
} }
_ = async {
match ttl_deadline.as_mut() {
Some(deadline) => deadline.as_mut().await,
None => std::future::pending().await,
}
} => {
ttl_expired = true;
break;
}
status = child.wait(), if observed_status.is_none() => { status = child.wait(), if observed_status.is_none() => {
observed_status = Some(status.context("failed waiting for terminal child")?); observed_status = Some(status.context("failed waiting for terminal child")?);
drain_deadline = Some(Box::pin(tokio::time::sleep(PTY_EXIT_DRAIN_TIMEOUT))); drain_deadline = Some(Box::pin(tokio::time::sleep(PTY_EXIT_DRAIN_TIMEOUT)));
@@ -355,7 +388,7 @@ async fn run_terminal(
let _ = output.try_send(Message::Text(text.into())); let _ = output.try_send(Message::Text(text.into()));
} }
} }
info!(terminal_id = %terminal_id, exit_code = ?status.code(), requested_close, "terminal worker exited"); info!(terminal_id = %terminal_id, exit_code = ?status.code(), requested_close, ttl_expired, "terminal worker exited");
Ok(()) Ok(())
} }
@@ -503,6 +536,7 @@ async fn run_terminal(
_terminal_id: &TerminalId, _terminal_id: &TerminalId,
_rows: u16, _rows: u16,
_cols: u16, _cols: u16,
_session_ttl: Option<Duration>,
_cancel: oneshot::Receiver<()>, _cancel: oneshot::Receiver<()>,
_relay_credentials: mpsc::UnboundedReceiver<String>, _relay_credentials: mpsc::UnboundedReceiver<String>,
) -> Result<()> { ) -> Result<()> {
@@ -599,6 +633,7 @@ mod tests {
cancel: Some(cancel), cancel: Some(cancel),
relay_credentials, relay_credentials,
created_at_unix: 42, created_at_unix: 42,
session_ttl_seconds: 600,
}, },
)]))), )]))),
max_sessions: 2, max_sessions: 2,
@@ -652,6 +687,7 @@ mod tests {
cancel: Some(cancel), cancel: Some(cancel),
relay_credentials, relay_credentials,
created_at_unix: 42, created_at_unix: 42,
session_ttl_seconds: 600,
}, },
)]))); )])));
let manager = TerminalManager { let manager = TerminalManager {
@@ -694,6 +730,34 @@ mod tests {
assert!(manager.active.lock().expect("terminal manager").is_empty()); assert!(manager.active.lock().expect("terminal manager").is_empty());
} }
#[cfg(unix)]
#[tokio::test]
async fn local_ttl_terminates_pty_without_control_plane_close() {
let mut config = crate::config::DEFAULT_CONFIG.clone();
config.terminal.enabled = true;
config.terminal.shell = "/bin/sh".into();
config.terminal.args = vec!["-c".into(), "sleep 30".into()];
let terminal_id = TerminalId::new("ttl-test").expect("terminal id");
let (_cancel, cancel_rx) = oneshot::channel();
let (_relay, relay_rx) = mpsc::unbounded_channel();
tokio::time::timeout(
Duration::from_secs(3),
run_terminal(
&config,
&terminal_id,
24,
80,
Some(Duration::from_millis(20)),
cancel_rx,
relay_rx,
),
)
.await
.expect("local TTL should stop the PTY promptly")
.expect("TTL cleanup should succeed");
}
#[test] #[test]
fn snapshot_reconstructs_screen_content_and_cursor() { fn snapshot_reconstructs_screen_content_and_cursor() {
let mut state = TerminalState::new(24, 80); let mut state = TerminalState::new(24, 80);
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "wakey-control-plane" name = "wakey-control-plane"
version = "0.3.2" version = "0.3.3"
edition = "2024" edition = "2024"
publish = ["gitea"] publish = ["gitea"]
+8 -2
View File
@@ -6,8 +6,9 @@ use serde::{Deserialize, Serialize};
use std::time::Instant; use std::time::Instant;
use tracing::{info, info_span, warn}; use tracing::{info, info_span, warn};
use uuid::Uuid; use uuid::Uuid;
use wakey_agent::protocol::AgentCapability; use wakey_agent::protocol::{
use wakey_agent::protocol::{AgentCommand, ErrorPayload, RequestId, ServerMessage}; AgentCapability, AgentCapabilityOptions, AgentCommand, ErrorPayload, RequestId, ServerMessage,
};
use crate::api::ApiError; use crate::api::ApiError;
use crate::runtime::{AgentReply, AppState, SessionEvent}; use crate::runtime::{AgentReply, AppState, SessionEvent};
@@ -19,6 +20,7 @@ pub struct AgentStatus {
pub connected: bool, pub connected: bool,
pub nickname: Option<String>, pub nickname: Option<String>,
pub capabilities: Vec<AgentCapability>, pub capabilities: Vec<AgentCapability>,
pub capability_options: AgentCapabilityOptions,
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
@@ -49,6 +51,10 @@ pub async fn list_agents(State(state): State<AppState>) -> Result<impl IntoRespo
.get(&agent_id) .get(&agent_id)
.map(|session| session.capabilities.clone()) .map(|session| session.capabilities.clone())
.unwrap_or_default(), .unwrap_or_default(),
capability_options: sessions
.get(&agent_id)
.map(|session| session.capability_options.clone())
.unwrap_or_default(),
agent_id, agent_id,
nickname, nickname,
}) })
+54 -23
View File
@@ -5,16 +5,16 @@ use axum::http::StatusCode;
use axum::response::Response; use axum::response::Response;
use futures_util::{SinkExt, StreamExt}; use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::time::Duration;
use tracing::{info, warn}; use tracing::{info, warn};
use wakey_agent::protocol::{ use wakey_agent::protocol::{
AgentCapability, ServerMessage, TerminalAgentHandshake, TerminalControl, TerminalId, AgentCapability, DEFAULT_TERMINAL_MAX_SESSIONS, DEFAULT_TERMINAL_SESSION_TTL_SECONDS,
TerminalOperatorHandshake, ServerMessage, TerminalAgentHandshake, TerminalControl, TerminalId, TerminalOperatorHandshake,
}; };
use crate::api::ApiError; use crate::api::ApiError;
use crate::runtime::terminals::{ use crate::runtime::terminals::{
TERMINAL_ABSOLUTE_TIMEOUT, TERMINAL_ATTACH_TIMEOUT, TERMINAL_MAX_FRAME_BYTES, TERMINAL_ATTACH_TIMEOUT, TERMINAL_MAX_FRAME_BYTES, TerminalRelayFrame, TerminalSummary,
TerminalRelayFrame, TerminalSummary,
}; };
use crate::runtime::{AppState, SessionEvent}; use crate::runtime::{AppState, SessionEvent};
use crate::state::AuditEventInput; use crate::state::AuditEventInput;
@@ -40,6 +40,7 @@ pub struct TerminalSessionResponse {
pub terminal_id: String, pub terminal_id: String,
pub agent_id: String, pub agent_id: String,
pub created_at_unix: u64, pub created_at_unix: u64,
pub expires_at_unix: Option<u64>,
pub agent_attached: bool, pub agent_attached: bool,
pub operator_attached: bool, pub operator_attached: bool,
pub websocket_url: String, pub websocket_url: String,
@@ -52,7 +53,7 @@ pub async fn create_terminal(
Json(request): Json<CreateTerminalRequest>, Json(request): Json<CreateTerminalRequest>,
) -> Result<(StatusCode, Json<TerminalSessionResponse>), ApiError> { ) -> Result<(StatusCode, Json<TerminalSessionResponse>), ApiError> {
validate_size(request.rows, request.cols)?; validate_size(request.rows, request.cols)?;
let agent_tx = { let (agent_tx, max_sessions, session_ttl_seconds) = {
let sessions = state.sessions.read().await; let sessions = state.sessions.read().await;
let session = sessions.get(&request.agent_id).ok_or_else(|| { let session = sessions.get(&request.agent_id).ok_or_else(|| {
ApiError::new( ApiError::new(
@@ -68,12 +69,25 @@ pub async fn create_terminal(
"agent has not advertised terminal capability", "agent has not advertised terminal capability",
)); ));
} }
session.tx.clone() let max_sessions = session
.capability_options
.terminal
.as_ref()
.map(|terminal| terminal.max_sessions)
.unwrap_or(DEFAULT_TERMINAL_MAX_SESSIONS)
.max(1);
let session_ttl_seconds = session
.capability_options
.terminal
.as_ref()
.map(|terminal| terminal.session_ttl_seconds)
.unwrap_or(DEFAULT_TERMINAL_SESSION_TTL_SECONDS);
(session.tx.clone(), max_sessions, session_ttl_seconds)
}; };
let created = state let created = state
.terminals .terminals
.create(request.agent_id.clone()) .create_with_limits(request.agent_id.clone(), max_sessions, session_ttl_seconds)
.await .await
.map_err(registry_error)?; .map_err(registry_error)?;
let terminal_id = TerminalId::new(created.terminal_id.clone()).map_err(|message| { let terminal_id = TerminalId::new(created.terminal_id.clone()).map_err(|message| {
@@ -100,7 +114,12 @@ pub async fn create_terminal(
)); ));
} }
spawn_absolute_timeout(state.clone(), terminal_id.clone(), request.agent_id.clone()); spawn_session_timeout(
state.clone(),
terminal_id.clone(),
request.agent_id.clone(),
session_ttl_seconds,
);
append_terminal_audit( append_terminal_audit(
&state, &state,
&request.agent_id, &request.agent_id,
@@ -122,6 +141,7 @@ pub async fn create_terminal(
terminal_id: terminal_id.to_string(), terminal_id: terminal_id.to_string(),
agent_id: request.agent_id, agent_id: request.agent_id,
created_at_unix: created.created_at_unix, created_at_unix: created.created_at_unix,
expires_at_unix: created.expires_at_unix,
agent_attached: false, agent_attached: false,
operator_attached: false, operator_attached: false,
attachment_token: Some(created.attachment_token), attachment_token: Some(created.attachment_token),
@@ -133,7 +153,7 @@ pub async fn get_terminal(
State(state): State<AppState>, State(state): State<AppState>,
Path(terminal_id): Path<String>, Path(terminal_id): Path<String>,
) -> Result<Json<TerminalSessionResponse>, ApiError> { ) -> Result<Json<TerminalSessionResponse>, ApiError> {
let (agent_id, created_at_unix, agent_attached, operator_attached) = state let summary = state
.terminals .terminals
.summary(&terminal_id) .summary(&terminal_id)
.await .await
@@ -141,10 +161,11 @@ pub async fn get_terminal(
Ok(Json(TerminalSessionResponse { Ok(Json(TerminalSessionResponse {
websocket_url: operator_ws_path(&terminal_id), websocket_url: operator_ws_path(&terminal_id),
terminal_id, terminal_id,
agent_id, agent_id: summary.agent_id,
created_at_unix, created_at_unix: summary.created_at_unix,
agent_attached, expires_at_unix: summary.expires_at_unix,
operator_attached, agent_attached: summary.agent_attached,
operator_attached: summary.operator_attached,
attachment_token: None, attachment_token: None,
})) }))
} }
@@ -171,7 +192,7 @@ pub async fn attach_terminal(
.issue_attachment_token_for_operator(&terminal_id, &request.operator_id) .issue_attachment_token_for_operator(&terminal_id, &request.operator_id)
.await .await
.map_err(registry_error)?; .map_err(registry_error)?;
let (agent_id, created_at_unix, agent_attached, operator_attached) = state let summary = state
.terminals .terminals
.summary(&terminal_id) .summary(&terminal_id)
.await .await
@@ -179,10 +200,11 @@ pub async fn attach_terminal(
Ok(Json(TerminalSessionResponse { Ok(Json(TerminalSessionResponse {
websocket_url: operator_ws_path(&terminal_id), websocket_url: operator_ws_path(&terminal_id),
terminal_id, terminal_id,
agent_id, agent_id: summary.agent_id,
created_at_unix, created_at_unix: summary.created_at_unix,
agent_attached, expires_at_unix: summary.expires_at_unix,
operator_attached, agent_attached: summary.agent_attached,
operator_attached: summary.operator_attached,
attachment_token: Some(attachment_token), attachment_token: Some(attachment_token),
})) }))
} }
@@ -343,10 +365,10 @@ async fn handle_operator_terminal_socket(
warn!(terminal_id, code, "failed to request terminal snapshot"); warn!(terminal_id, code, "failed to request terminal snapshot");
} }
let summary = state.terminals.summary(&terminal_id).await; let summary = state.terminals.summary(&terminal_id).await;
if let Some((agent_id, _, _, _)) = &summary { if let Some(summary) = &summary {
append_terminal_audit( append_terminal_audit(
&state, &state,
agent_id, &summary.agent_id,
&terminal_id, &terminal_id,
TerminalAudit { TerminalAudit {
actor_type: "admin_api", actor_type: "admin_api",
@@ -360,7 +382,7 @@ async fn handle_operator_terminal_socket(
} }
let (mut write, mut read) = socket.split(); let (mut write, mut read) = socket.split();
if summary.is_some_and(|(_, _, agent_attached, _)| agent_attached) { if summary.is_some_and(|summary| summary.agent_attached) {
let ready = serde_json::to_string(&TerminalControl::Ready) let ready = serde_json::to_string(&TerminalControl::Ready)
.expect("terminal ready control serializes"); .expect("terminal ready control serializes");
if send_relay_frame(&mut write, TerminalRelayFrame::Text(ready)) if send_relay_frame(&mut write, TerminalRelayFrame::Text(ready))
@@ -414,6 +436,7 @@ fn terminal_response(summary: TerminalSummary) -> TerminalSessionResponse {
terminal_id: summary.terminal_id, terminal_id: summary.terminal_id,
agent_id: summary.agent_id, agent_id: summary.agent_id,
created_at_unix: summary.created_at_unix, created_at_unix: summary.created_at_unix,
expires_at_unix: summary.expires_at_unix,
agent_attached: summary.agent_attached, agent_attached: summary.agent_attached,
operator_attached: summary.operator_attached, operator_attached: summary.operator_attached,
attachment_token: None, attachment_token: None,
@@ -549,9 +572,17 @@ async fn close_registered_terminal(
Some(agent_id) Some(agent_id)
} }
fn spawn_absolute_timeout(state: AppState, terminal_id: TerminalId, agent_id: String) { fn spawn_session_timeout(
state: AppState,
terminal_id: TerminalId,
agent_id: String,
session_ttl_seconds: u64,
) {
if session_ttl_seconds == 0 {
return;
}
tokio::spawn(async move { tokio::spawn(async move {
tokio::time::sleep(TERMINAL_ABSOLUTE_TIMEOUT).await; tokio::time::sleep(Duration::from_secs(session_ttl_seconds)).await;
if close_registered_terminal( if close_registered_terminal(
&state, &state,
terminal_id.as_str(), terminal_id.as_str(),
+61 -2
View File
@@ -68,14 +68,14 @@ pub fn write_init_config(args: &InitConfigArgs) -> Result<Option<PathBuf>> {
.clone() .clone()
.or(base.state_file) .or(base.state_file)
.unwrap_or_else(|| defaults.state_file.clone()); .unwrap_or_else(|| defaults.state_file.clone());
let state_file = resolve_path(&data_dir, state_file_raw); let state_file = path_for_rendered_config(&data_dir, state_file_raw);
let pid_file_raw = args let pid_file_raw = args
.pid_file .pid_file
.clone() .clone()
.or(base.pid_file) .or(base.pid_file)
.unwrap_or_else(|| defaults.pid_file.clone()); .unwrap_or_else(|| defaults.pid_file.clone());
let pid_file = resolve_path(&data_dir, pid_file_raw); let pid_file = path_for_rendered_config(&data_dir, pid_file_raw);
let ui_dist_dir = args let ui_dist_dir = args
.ui_dist_dir .ui_dist_dir
@@ -147,6 +147,18 @@ pub fn write_init_config(args: &InitConfigArgs) -> Result<Option<PathBuf>> {
Ok(None) Ok(None)
} }
/// Keep relative paths relative when the rendered config's `data_dir` is also
/// relative. The daemon resolves relative paths under `data_dir`; embedding
/// that directory into another relative path here would make it get joined a
/// second time when the generated config is loaded.
fn path_for_rendered_config(data_dir: &std::path::Path, candidate: PathBuf) -> PathBuf {
if data_dir.is_relative() && candidate.is_relative() {
candidate
} else {
resolve_path(data_dir, candidate)
}
}
pub fn bootstrap_config_if_missing(args: &ServeArgs) -> Result<bool> { pub fn bootstrap_config_if_missing(args: &ServeArgs) -> Result<bool> {
if args.config_file.exists() { if args.config_file.exists() {
return Ok(false); return Ok(false);
@@ -303,4 +315,51 @@ json_logs = true
let _ = std::fs::remove_dir_all(&dir); let _ = std::fs::remove_dir_all(&dir);
} }
#[test]
fn init_config_preserves_relative_paths_for_relative_data_dir() {
let dir = temp_test_dir("relative-data-dir");
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("create temp dir");
let base = dir.join("base.toml");
let out = dir.join("out.toml");
std::fs::write(
&base,
r#"
data_dir = "test-ddir/wakey-control-plane"
state_file = "state.sqlite3"
pid_file = "wakey-control-plane.pid"
"#,
)
.expect("write base config");
write_init_config(&InitConfigArgs {
config_file: Some(out.clone()),
from_config: Some(base),
stdout: false,
data_dir: None,
bind: None,
public_url: None,
state_file: None,
pid_file: None,
ui_dist_dir: None,
command_timeout_ms: None,
enroll_token_ttl_seconds: None,
observation_retention_seconds: None,
bootstrap_enroll_tokens: Vec::new(),
telemetry_otlp_endpoint: None,
telemetry_service_name: None,
telemetry_json_logs: None,
force: false,
})
.expect("write derived config");
let parsed: toml::Value =
toml::from_str(&std::fs::read_to_string(&out).expect("read derived config"))
.expect("parse derived config");
assert_eq!(parsed["state_file"].as_str(), Some("state.sqlite3"));
assert_eq!(parsed["pid_file"].as_str(), Some("wakey-control-plane.pid"));
let _ = std::fs::remove_dir_all(&dir);
}
} }
+2 -1
View File
@@ -16,7 +16,7 @@ use tower_http::services::ServeFile;
use tracing::info; use tracing::info;
#[cfg(unix)] #[cfg(unix)]
use tracing::warn; use tracing::warn;
use wakey_agent::protocol::{AgentCapability, ErrorPayload, ServerMessage}; use wakey_agent::protocol::{AgentCapability, AgentCapabilityOptions, ErrorPayload, ServerMessage};
use crate::api; use crate::api;
use crate::config; use crate::config;
@@ -50,6 +50,7 @@ pub struct AgentSession {
pub connection_id: String, pub connection_id: String,
pub tx: mpsc::UnboundedSender<SessionEvent>, pub tx: mpsc::UnboundedSender<SessionEvent>,
pub capabilities: Vec<AgentCapability>, pub capabilities: Vec<AgentCapability>,
pub capability_options: AgentCapabilityOptions,
} }
#[derive(Clone)] #[derive(Clone)]
pub enum SessionEvent { pub enum SessionEvent {
+193 -44
View File
@@ -9,9 +9,7 @@ use wakey_agent::protocol::{AgentTerminalSession, TerminalId};
pub const TERMINAL_RELAY_QUEUE: usize = 32; pub const TERMINAL_RELAY_QUEUE: usize = 32;
pub const TERMINAL_MAX_FRAME_BYTES: usize = 64 * 1024; pub const TERMINAL_MAX_FRAME_BYTES: usize = 64 * 1024;
pub const TERMINAL_PENDING_AGENT_BYTES: usize = 256 * 1024; pub const TERMINAL_PENDING_AGENT_BYTES: usize = 256 * 1024;
pub const TERMINAL_MAX_SESSIONS_PER_AGENT: usize = 2;
pub const TERMINAL_ATTACH_TIMEOUT: Duration = Duration::from_secs(10); pub const TERMINAL_ATTACH_TIMEOUT: Duration = Duration::from_secs(10);
pub const TERMINAL_ABSOLUTE_TIMEOUT: Duration = Duration::from_secs(12 * 60 * 60);
const TERMINAL_TOMBSTONE_TTL: Duration = Duration::from_secs(5 * 60); const TERMINAL_TOMBSTONE_TTL: Duration = Duration::from_secs(5 * 60);
const TERMINAL_MAX_TOMBSTONES: usize = 1024; const TERMINAL_MAX_TOMBSTONES: usize = 1024;
@@ -31,7 +29,7 @@ pub struct TerminalRegistry {
struct TerminalSession { struct TerminalSession {
agent_id: String, agent_id: String,
created_at_unix: u64, created_at_unix: u64,
expires_at: Instant, expiry: TerminalExpiry,
agent_confirmed: bool, agent_confirmed: bool,
relay_token: Option<String>, relay_token: Option<String>,
attachment_token: Option<String>, attachment_token: Option<String>,
@@ -50,6 +48,7 @@ pub struct CreatedTerminal {
pub relay_token: String, pub relay_token: String,
pub attachment_token: String, pub attachment_token: String,
pub created_at_unix: u64, pub created_at_unix: u64,
pub expires_at_unix: Option<u64>,
} }
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
@@ -57,6 +56,7 @@ pub struct TerminalSummary {
pub terminal_id: String, pub terminal_id: String,
pub agent_id: String, pub agent_id: String,
pub created_at_unix: u64, pub created_at_unix: u64,
pub expires_at_unix: Option<u64>,
pub agent_attached: bool, pub agent_attached: bool,
pub operator_attached: bool, pub operator_attached: bool,
} }
@@ -75,14 +75,32 @@ impl TerminalRegistry {
} }
} }
pub async fn create(&self, agent_id: String) -> Result<CreatedTerminal, &'static str> { #[cfg(test)]
async fn create(&self, agent_id: String) -> Result<CreatedTerminal, &'static str> {
self.create_with_limits(
agent_id,
wakey_agent::protocol::DEFAULT_TERMINAL_MAX_SESSIONS,
wakey_agent::protocol::DEFAULT_TERMINAL_SESSION_TTL_SECONDS,
)
.await
}
/// Creates a session using the limit advertised by the connected agent.
/// The caller snapshots the limit with the agent connection so UI hints and
/// server-side enforcement use the same value.
pub async fn create_with_limits(
&self,
agent_id: String,
max_sessions: usize,
session_ttl_seconds: u64,
) -> Result<CreatedTerminal, &'static str> {
let mut sessions = self.inner.lock().await; let mut sessions = self.inner.lock().await;
prune_expired_sessions(&mut sessions); prune_expired_sessions(&mut sessions);
if sessions if sessions
.values() .values()
.filter(|session| session.agent_id == agent_id) .filter(|session| session.agent_id == agent_id)
.count() .count()
>= TERMINAL_MAX_SESSIONS_PER_AGENT >= max_sessions.max(1)
{ {
return Err("agent_terminal_limit_reached"); return Err("agent_terminal_limit_reached");
} }
@@ -94,12 +112,14 @@ impl TerminalRegistry {
.duration_since(UNIX_EPOCH) .duration_since(UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(); .as_secs();
let expiry = expiry_for_created_at(created_at_unix, session_ttl_seconds)
.ok_or("terminal_ttl_invalid")?;
sessions.insert( sessions.insert(
terminal_id.clone(), terminal_id.clone(),
TerminalSession { TerminalSession {
agent_id, agent_id,
created_at_unix, created_at_unix,
expires_at: Instant::now() + TERMINAL_ABSOLUTE_TIMEOUT, expiry,
agent_confirmed: false, agent_confirmed: false,
relay_token: Some(relay_token.clone()), relay_token: Some(relay_token.clone()),
attachment_token: Some(attachment_token.clone()), attachment_token: Some(attachment_token.clone()),
@@ -119,6 +139,7 @@ impl TerminalRegistry {
relay_token, relay_token,
attachment_token, attachment_token,
created_at_unix, created_at_unix,
expires_at_unix: expiry.unix(),
}) })
} }
@@ -169,7 +190,10 @@ impl TerminalRegistry {
) -> Vec<(TerminalId, String)> { ) -> Vec<(TerminalId, String)> {
let reported_ids = reported let reported_ids = reported
.iter() .iter()
.filter(|session| expires_at_for_created_at(session.created_at_unix).is_some()) .filter(|session| {
expiry_for_created_at(session.created_at_unix, session.session_ttl_seconds)
.is_some()
})
.map(|session| session.terminal_id.as_str()) .map(|session| session.terminal_id.as_str())
.collect::<std::collections::HashSet<_>>(); .collect::<std::collections::HashSet<_>>();
let stale_ids = { let stale_ids = {
@@ -192,8 +216,10 @@ impl TerminalRegistry {
let mut credentials = Vec::new(); let mut credentials = Vec::new();
let mut sessions = self.inner.lock().await; let mut sessions = self.inner.lock().await;
for reported_session in reported { for reported_session in reported {
let Some(expires_at) = expires_at_for_created_at(reported_session.created_at_unix) let Some(expiry) = expiry_for_created_at(
else { reported_session.created_at_unix,
reported_session.session_ttl_seconds,
) else {
continue; continue;
}; };
let terminal_id = reported_session.terminal_id.as_str().to_string(); let terminal_id = reported_session.terminal_id.as_str().to_string();
@@ -202,7 +228,7 @@ impl TerminalRegistry {
.or_insert_with(|| TerminalSession { .or_insert_with(|| TerminalSession {
agent_id: agent_id.to_string(), agent_id: agent_id.to_string(),
created_at_unix: reported_session.created_at_unix, created_at_unix: reported_session.created_at_unix,
expires_at, expiry,
agent_confirmed: true, agent_confirmed: true,
relay_token: None, relay_token: None,
attachment_token: None, attachment_token: None,
@@ -422,17 +448,12 @@ impl TerminalRegistry {
Some(detached_at) Some(detached_at)
} }
pub async fn summary(&self, terminal_id: &str) -> Option<(String, u64, bool, bool)> { pub async fn summary(&self, terminal_id: &str) -> Option<TerminalSummary> {
let mut sessions = self.inner.lock().await; let mut sessions = self.inner.lock().await;
prune_expired_sessions(&mut sessions); prune_expired_sessions(&mut sessions);
sessions.get(terminal_id).map(|session| { sessions
( .get(terminal_id)
session.agent_id.clone(), .map(|session| terminal_summary(terminal_id, session))
session.created_at_unix,
session.agent_tx.is_some(),
session.operator_tx.is_some(),
)
})
} }
pub async fn summaries(&self) -> Vec<TerminalSummary> { pub async fn summaries(&self) -> Vec<TerminalSummary> {
@@ -440,26 +461,31 @@ impl TerminalRegistry {
prune_expired_sessions(&mut sessions); prune_expired_sessions(&mut sessions);
let mut summaries = sessions let mut summaries = sessions
.iter() .iter()
.map(|(terminal_id, session)| TerminalSummary { .map(|(terminal_id, session)| terminal_summary(terminal_id, session))
terminal_id: terminal_id.clone(),
agent_id: session.agent_id.clone(),
created_at_unix: session.created_at_unix,
agent_attached: session.agent_tx.is_some(),
operator_attached: session.operator_tx.is_some(),
})
.collect::<Vec<_>>(); .collect::<Vec<_>>();
summaries.sort_by_key(|session| std::cmp::Reverse(session.created_at_unix)); summaries.sort_by_key(|session| std::cmp::Reverse(session.created_at_unix));
summaries summaries
} }
} }
fn terminal_summary(terminal_id: &str, session: &TerminalSession) -> TerminalSummary {
TerminalSummary {
terminal_id: terminal_id.to_string(),
agent_id: session.agent_id.clone(),
created_at_unix: session.created_at_unix,
expires_at_unix: session.expiry.unix(),
agent_attached: session.agent_tx.is_some(),
operator_attached: session.operator_tx.is_some(),
}
}
fn active_session<'a>( fn active_session<'a>(
sessions: &'a mut HashMap<String, TerminalSession>, sessions: &'a mut HashMap<String, TerminalSession>,
terminal_id: &str, terminal_id: &str,
) -> Result<&'a mut TerminalSession, &'static str> { ) -> Result<&'a mut TerminalSession, &'static str> {
if sessions if sessions
.get(terminal_id) .get(terminal_id)
.is_some_and(|session| session.expires_at <= Instant::now()) .is_some_and(|session| session.expiry.is_expired(Instant::now()))
{ {
sessions.remove(terminal_id); sessions.remove(terminal_id);
return Err("terminal_expired"); return Err("terminal_expired");
@@ -499,22 +525,48 @@ fn prune_tombstones(closed: &mut HashMap<String, Instant>) {
fn prune_expired_sessions(sessions: &mut HashMap<String, TerminalSession>) { fn prune_expired_sessions(sessions: &mut HashMap<String, TerminalSession>) {
let now = Instant::now(); let now = Instant::now();
sessions.retain(|_, session| session.expires_at > now); sessions.retain(|_, session| !session.expiry.is_expired(now));
} }
/// Converts the agent's durable creation timestamp into CC's monotonic #[derive(Clone, Copy, Debug)]
/// deadline. Reconciliation must not grant an old PTY a fresh twelve hours. enum TerminalExpiry {
fn expires_at_for_created_at(created_at_unix: u64) -> Option<Instant> { Unlimited,
Finite { deadline: Instant, unix: u64 },
}
impl TerminalExpiry {
fn unix(self) -> Option<u64> {
match self {
Self::Unlimited => None,
Self::Finite { unix, .. } => Some(unix),
}
}
fn is_expired(self, now: Instant) -> bool {
matches!(self, Self::Finite { deadline, .. } if deadline <= now)
}
}
/// Converts the agent's creation-time policy into CC's monotonic deadline.
/// `None` rejects expired or unrepresentable sessions; `Unlimited` is the
/// explicit zero-TTL policy.
fn expiry_for_created_at(created_at_unix: u64, session_ttl_seconds: u64) -> Option<TerminalExpiry> {
if session_ttl_seconds == 0 {
return Some(TerminalExpiry::Unlimited);
}
let expires_at_unix = created_at_unix.checked_add(session_ttl_seconds)?;
let now_unix = SystemTime::now() let now_unix = SystemTime::now()
.duration_since(UNIX_EPOCH) .duration_since(UNIX_EPOCH)
.unwrap_or_default() .unwrap_or_default()
.as_secs(); .as_secs();
let age = Duration::from_secs(now_unix.saturating_sub(created_at_unix)); if expires_at_unix <= now_unix {
if age >= TERMINAL_ABSOLUTE_TIMEOUT {
return None; return None;
} }
let remaining = TERMINAL_ABSOLUTE_TIMEOUT - age; let remaining = Duration::from_secs(expires_at_unix - now_unix);
Some(Instant::now() + remaining) Some(TerminalExpiry::Finite {
deadline: Instant::now().checked_add(remaining)?,
unix: expires_at_unix,
})
} }
fn relay_frame_size(frame: &TerminalRelayFrame) -> usize { fn relay_frame_size(frame: &TerminalRelayFrame) -> usize {
@@ -528,6 +580,7 @@ fn relay_frame_size(frame: &TerminalRelayFrame) -> usize {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use wakey_agent::protocol::DEFAULT_TERMINAL_SESSION_TTL_SECONDS;
const OPERATOR_A: &str = "browser-tab-a"; const OPERATOR_A: &str = "browser-tab-a";
const OPERATOR_B: &str = "browser-tab-b"; const OPERATOR_B: &str = "browser-tab-b";
@@ -643,6 +696,73 @@ mod tests {
assert_eq!(third.err(), Some("agent_terminal_limit_reached")); assert_eq!(third.err(), Some("agent_terminal_limit_reached"));
} }
#[tokio::test]
async fn advertised_limit_controls_session_capacity() {
let registry = TerminalRegistry::new();
for _ in 0..3 {
registry
.create_with_limits("router".into(), 3, DEFAULT_TERMINAL_SESSION_TTL_SECONDS)
.await
.expect("within advertised limit");
}
let fourth = registry
.create_with_limits("router".into(), 3, DEFAULT_TERMINAL_SESSION_TTL_SECONDS)
.await;
assert_eq!(fourth.err(), Some("agent_terminal_limit_reached"));
}
#[tokio::test]
async fn session_ttl_is_captured_and_zero_is_unlimited() {
let registry = TerminalRegistry::new();
let finite = registry
.create_with_limits("router".into(), 2, 67)
.await
.expect("finite session");
assert_eq!(
finite.expires_at_unix,
finite.created_at_unix.checked_add(67)
);
registry.remove(&finite.terminal_id).await;
let unlimited = registry
.create_with_limits("router".into(), 2, 0)
.await
.expect("unlimited session");
assert_eq!(unlimited.expires_at_unix, None);
assert_eq!(
registry
.summary(&unlimited.terminal_id)
.await
.expect("unlimited summary")
.expires_at_unix,
None
);
}
#[tokio::test]
async fn unlimited_inventory_can_adopt_an_old_session() {
let registry = TerminalRegistry::new();
let terminal_id = TerminalId::new("old-unlimited").expect("terminal id");
let reported = AgentTerminalSession {
terminal_id: terminal_id.clone(),
created_at_unix: 0,
session_ttl_seconds: 0,
};
let credentials = registry
.reconcile_agent_sessions("router", &[reported])
.await;
assert_eq!(credentials.len(), 1);
assert_eq!(
registry
.summary(terminal_id.as_str())
.await
.expect("adopted unlimited session")
.expires_at_unix,
None
);
}
#[tokio::test] #[tokio::test]
async fn removed_session_is_distinct_from_unknown_session() { async fn removed_session_is_distinct_from_unknown_session() {
let registry = TerminalRegistry::new(); let registry = TerminalRegistry::new();
@@ -663,6 +783,7 @@ mod tests {
let reported = AgentTerminalSession { let reported = AgentTerminalSession {
terminal_id: terminal_id.clone(), terminal_id: terminal_id.clone(),
created_at_unix, created_at_unix,
session_ttl_seconds: 0,
}; };
let credentials = registry let credentials = registry
@@ -679,9 +800,9 @@ mod tests {
.summary(terminal_id.as_str()) .summary(terminal_id.as_str())
.await .await
.expect("adopted summary"); .expect("adopted summary");
assert_eq!(summary.0, "router"); assert_eq!(summary.agent_id, "router");
assert_eq!(summary.1, created_at_unix); assert_eq!(summary.created_at_unix, created_at_unix);
assert!(summary.2); assert!(summary.agent_attached);
} }
#[tokio::test] #[tokio::test]
@@ -695,7 +816,10 @@ mod tests {
.await .await
.get_mut(&first.terminal_id) .get_mut(&first.terminal_id)
.expect("first session") .expect("first session")
.expires_at = Instant::now(); .expiry = TerminalExpiry::Finite {
deadline: Instant::now(),
unix: first.created_at_unix,
};
registry registry
.create("router".into()) .create("router".into())
@@ -711,6 +835,7 @@ mod tests {
let reported = AgentTerminalSession { let reported = AgentTerminalSession {
terminal_id: terminal_id.clone(), terminal_id: terminal_id.clone(),
created_at_unix: 0, created_at_unix: 0,
session_ttl_seconds: DEFAULT_TERMINAL_SESSION_TTL_SECONDS,
}; };
assert!( assert!(
@@ -739,7 +864,7 @@ mod tests {
.summary(&created.terminal_id) .summary(&created.terminal_id)
.await .await
.expect("session remains after detach"); .expect("session remains after detach");
assert!(!summary.3); assert!(!summary.operator_attached);
registry registry
.issue_attachment_token_for_operator(&created.terminal_id, OPERATOR_A) .issue_attachment_token_for_operator(&created.terminal_id, OPERATOR_A)
.await .await
@@ -765,7 +890,13 @@ mod tests {
.await .await
.expect("handoff token"); .expect("handoff token");
assert!(!registry.summary(&previous.terminal_id).await.unwrap().3); assert!(
!registry
.summary(&previous.terminal_id)
.await
.unwrap()
.operator_attached
);
registry registry
.attach_operator(&next.terminal_id, &next_token, OPERATOR_A) .attach_operator(&next.terminal_id, &next_token, OPERATOR_A)
.await .await
@@ -800,8 +931,20 @@ mod tests {
.expect_err("occupied target remains protected"); .expect_err("occupied target remains protected");
assert_eq!(error, "terminal_operator_already_attached"); assert_eq!(error, "terminal_operator_already_attached");
assert!(registry.summary(&previous.terminal_id).await.unwrap().3); assert!(
assert!(registry.summary(&occupied.terminal_id).await.unwrap().3); registry
.summary(&previous.terminal_id)
.await
.unwrap()
.operator_attached
);
assert!(
registry
.summary(&occupied.terminal_id)
.await
.unwrap()
.operator_attached
);
} }
#[tokio::test] #[tokio::test]
@@ -829,7 +972,13 @@ mod tests {
.await .await
.is_none() .is_none()
); );
assert!(registry.summary(&created.terminal_id).await.unwrap().3); assert!(
registry
.summary(&created.terminal_id)
.await
.unwrap()
.operator_attached
);
} }
#[tokio::test] #[tokio::test]
+50 -3
View File
@@ -9,8 +9,9 @@ use tokio::sync::mpsc;
use tracing::{debug, info, info_span, warn}; use tracing::{debug, info, info_span, warn};
use uuid::Uuid; use uuid::Uuid;
use wakey_agent::protocol::{ use wakey_agent::protocol::{
AgentCapability, AgentTerminalSession, ErrorPayload, RequestId, ServerMessage, TerminalControl, AgentCapability, AgentCapabilityOptions, AgentTerminalSession, DEFAULT_TERMINAL_MAX_SESSIONS,
TerminalId, DEFAULT_TERMINAL_SESSION_TTL_SECONDS, ErrorPayload, RequestId, ServerMessage,
TerminalCapabilityOptions, TerminalControl, TerminalId, checked_terminal_session_ttl,
}; };
use wakey_core::Device; use wakey_core::Device;
@@ -24,6 +25,8 @@ enum IncomingClientMessage {
agent_id: String, agent_id: String,
#[serde(default)] #[serde(default)]
capabilities: Vec<AgentCapability>, capabilities: Vec<AgentCapability>,
#[serde(default)]
capability_options: AgentCapabilityOptions,
}, },
Auth { Auth {
agent_id: String, agent_id: String,
@@ -59,6 +62,7 @@ struct AgentConnectionState {
hello_agent_id: Option<String>, hello_agent_id: Option<String>,
hello_at: Option<Instant>, hello_at: Option<Instant>,
capabilities: Vec<AgentCapability>, capabilities: Vec<AgentCapability>,
capability_options: AgentCapabilityOptions,
} }
pub async fn agent_ws(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse { pub async fn agent_ws(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
@@ -193,6 +197,7 @@ async fn process_agent_text(
IncomingClientMessage::Hello { IncomingClientMessage::Hello {
agent_id, agent_id,
capabilities, capabilities,
capability_options,
} => { } => {
if connection if connection
.hello_agent_id .hello_agent_id
@@ -208,6 +213,27 @@ async fn process_agent_text(
let connect_to_hello_ms = connected_at.elapsed().as_millis() as u64; let connect_to_hello_ms = connected_at.elapsed().as_millis() as u64;
info!(agent_id = %agent_id, connect_to_hello_ms, "agent hello received"); info!(agent_id = %agent_id, connect_to_hello_ms, "agent hello received");
connection.hello_agent_id = Some(agent_id); connection.hello_agent_id = Some(agent_id);
let terminal = if capabilities.contains(&AgentCapability::Terminal) {
let max_sessions = capability_options
.terminal
.as_ref()
.map(|terminal| terminal.max_sessions)
.unwrap_or(DEFAULT_TERMINAL_MAX_SESSIONS)
.max(1);
let session_ttl_seconds = capability_options
.terminal
.as_ref()
.map(|terminal| terminal.session_ttl_seconds)
.unwrap_or(DEFAULT_TERMINAL_SESSION_TTL_SECONDS);
checked_terminal_session_ttl(session_ttl_seconds).map_err(anyhow::Error::msg)?;
Some(TerminalCapabilityOptions {
max_sessions,
session_ttl_seconds,
})
} else {
None
};
connection.capability_options = AgentCapabilityOptions { terminal };
connection.capabilities = capabilities; connection.capabilities = capabilities;
} }
IncomingClientMessage::Auth { IncomingClientMessage::Auth {
@@ -253,6 +279,7 @@ async fn process_agent_text(
connection_id: connection_id.to_string(), connection_id: connection_id.to_string(),
tx: tx.clone(), tx: tx.clone(),
capabilities: connection.capabilities.clone(), capabilities: connection.capabilities.clone(),
capability_options: connection.capability_options.clone(),
}, },
); );
connection.authed_agent_id = Some(agent_id.clone()); connection.authed_agent_id = Some(agent_id.clone());
@@ -424,10 +451,13 @@ mod tests {
use std::collections::HashMap; use std::collections::HashMap;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use wakey_agent::protocol::AgentCapabilityOptions;
use crate::runtime::AgentSession; use crate::runtime::AgentSession;
use super::{AgentConnectionState, is_current_session, validate_auth_identity}; use super::{
AgentConnectionState, IncomingClientMessage, is_current_session, validate_auth_identity,
};
#[test] #[test]
fn current_session_check_rejects_stale_connection_ids() { fn current_session_check_rejects_stale_connection_ids() {
@@ -439,6 +469,7 @@ mod tests {
connection_id: "conn-new".to_string(), connection_id: "conn-new".to_string(),
tx, tx,
capabilities: Vec::new(), capabilities: Vec::new(),
capability_options: AgentCapabilityOptions::default(),
}, },
); );
@@ -456,4 +487,20 @@ mod tests {
assert!(validate_auth_identity(&connection, "agent-a").is_ok()); assert!(validate_auth_identity(&connection, "agent-a").is_ok());
assert!(validate_auth_identity(&connection, "agent-b").is_err()); assert!(validate_auth_identity(&connection, "agent-b").is_err());
} }
#[test]
fn legacy_hello_without_capability_options_deserializes() {
let message: IncomingClientMessage = serde_json::from_str(
r#"{"type":"hello","agent_id":"agent-a","capabilities":["terminal"]}"#,
)
.expect("deserialize legacy hello");
let IncomingClientMessage::Hello {
capability_options, ..
} = message
else {
panic!("expected hello");
};
assert_eq!(capability_options, AgentCapabilityOptions::default());
}
} }
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "wakey-core" name = "wakey-core"
version = "0.3.2" version = "0.3.3"
edition = "2024" edition = "2024"
publish = ["gitea"] publish = ["gitea"]
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "wakey-linux" name = "wakey-linux"
version = "0.3.2" version = "0.3.3"
edition = "2024" edition = "2024"
publish = ["gitea"] publish = ["gitea"]
+20 -4
View File
@@ -27,8 +27,18 @@ pub struct TerminalWriter {
} }
impl TerminalPty { impl TerminalPty {
/// Starts `program` attached to a newly allocated PTY. /// Starts a configured program attached to a newly allocated PTY.
pub fn spawn(program: &Path, rows: u16, cols: u16) -> Result<Self> { ///
/// Wakey retains control of the PTY streams, terminal environment, and
/// kill-on-drop behavior; callers may only select arguments and a working
/// directory.
pub fn spawn(
program: &Path,
args: &[String],
current_dir: Option<&Path>,
rows: u16,
cols: u16,
) -> Result<Self> {
let (pty, pts) = pty_process::open().context("failed to open PTY")?; let (pty, pts) = pty_process::open().context("failed to open PTY")?;
pty.resize(Size::new(rows, cols)) pty.resize(Size::new(rows, cols))
.context("failed to set initial PTY size")?; .context("failed to set initial PTY size")?;
@@ -39,10 +49,14 @@ impl TerminalPty {
// The remote frontend is xterm.js regardless of the daemon's own // The remote frontend is xterm.js regardless of the daemon's own
// environment, so advertise the terminal the child actually receives. // environment, so advertise the terminal the child actually receives.
let command = Command::new(program) let mut command = Command::new(program)
.args(args)
.env("TERM", "xterm-256color") .env("TERM", "xterm-256color")
.env("COLORTERM", "truecolor") .env("COLORTERM", "truecolor")
.kill_on_drop(true); .kill_on_drop(true);
if let Some(current_dir) = current_dir {
command = command.current_dir(current_dir);
}
let child = command let child = command
.spawn(pts) .spawn(pts)
.with_context(|| format!("failed to spawn {} in PTY", program.display()))?; .with_context(|| format!("failed to spawn {} in PTY", program.display()))?;
@@ -114,8 +128,10 @@ mod tests {
/// ///
/// `./scripts/test_remote.ps1 -Package wakey-linux -Filter terminal::tests::pty_round_trip_and_resize -Exact -ShowOutput` /// `./scripts/test_remote.ps1 -Package wakey-linux -Filter terminal::tests::pty_round_trip_and_resize -Exact -ShowOutput`
#[tokio::test] #[tokio::test]
#[ignore = "live system check, requires PTY and shell"]
async fn pty_round_trip_and_resize() { async fn pty_round_trip_and_resize() {
let (mut reader, mut writer, mut child) = TerminalPty::spawn(Path::new("/bin/sh"), 24, 80) let (mut reader, mut writer, mut child) =
TerminalPty::spawn(Path::new("/bin/sh"), &[], None, 24, 80)
.expect("spawn PTY shell") .expect("spawn PTY shell")
.into_parts(); .into_parts();