Compare commits
7
Commits
No files matched your search
Generated
+5
-5
@@ -3624,7 +3624,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "wakey"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"chrono",
|
||||
@@ -3645,7 +3645,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "wakey-agent"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"clap",
|
||||
@@ -3670,7 +3670,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "wakey-control-plane"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"axum",
|
||||
@@ -3699,7 +3699,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "wakey-core"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
dependencies = [
|
||||
"macaddr",
|
||||
"serde",
|
||||
@@ -3711,7 +3711,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "wakey-linux"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"futures",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "wakey"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
edition = "2024"
|
||||
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.
|
||||
|
||||
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.
|
||||
|
||||
The agent reports a normal exit status or terminating signal when available. The UI must distinguish an exited shell from a transport failure.
|
||||
|
||||
@@ -3,12 +3,16 @@ export type Agent = {
|
||||
connected: boolean;
|
||||
nickname?: string | null;
|
||||
capabilities: "terminal"[];
|
||||
capability_options?: {
|
||||
terminal?: { max_sessions: number };
|
||||
};
|
||||
};
|
||||
|
||||
export type TerminalSession = {
|
||||
terminal_id: string;
|
||||
agent_id: string;
|
||||
created_at_unix: number;
|
||||
expires_at_unix?: number | null;
|
||||
agent_attached: boolean;
|
||||
operator_attached: boolean;
|
||||
websocket_url: string;
|
||||
|
||||
@@ -67,18 +67,31 @@ function mergeTerminalSession(
|
||||
sessions: TerminalSession[],
|
||||
next: TerminalSession,
|
||||
): TerminalSession[] {
|
||||
const remaining = sessions.filter(
|
||||
(item) => item.terminal_id !== next.terminal_id,
|
||||
const existingIndex = sessions.findIndex(
|
||||
(item) => item.terminal_id === next.terminal_id,
|
||||
);
|
||||
return [...remaining, next].sort(
|
||||
(left, right) => left.created_at_unix - right.created_at_unix,
|
||||
if (existingIndex >= 0) {
|
||||
// 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[] {
|
||||
return [...sessions].sort(
|
||||
(left, right) => left.created_at_unix - right.created_at_unix,
|
||||
);
|
||||
return [...sessions].sort(compareTerminalSessions);
|
||||
}
|
||||
|
||||
function reconcileTerminalSessions(
|
||||
@@ -156,7 +169,12 @@ export function TerminalPage({
|
||||
const selectedAgentSessionCount = sessions.filter(
|
||||
(item) => item.agent_id === selectedAgentId,
|
||||
).length;
|
||||
const agentAtSessionLimit = selectedAgentSessionCount >= 2;
|
||||
const selectedAgentSessionLimit = Math.max(
|
||||
1,
|
||||
selectedAgent?.capability_options?.terminal?.max_sessions ?? 2,
|
||||
);
|
||||
const agentAtSessionLimit =
|
||||
selectedAgentSessionCount >= selectedAgentSessionLimit;
|
||||
const canRequestStart =
|
||||
selectedAgent?.connected &&
|
||||
selectedAgent.capabilities.includes("terminal") &&
|
||||
@@ -744,7 +762,7 @@ export function TerminalPage({
|
||||
</TooltipTrigger>
|
||||
<TooltipContent>
|
||||
{agentAtSessionLimit
|
||||
? "Two-session limit reached"
|
||||
? `${selectedAgentSessionLimit}-session limit reached`
|
||||
: "New session"}
|
||||
</TooltipContent>
|
||||
</Tooltip>
|
||||
|
||||
@@ -365,6 +365,9 @@
|
||||
gap: 0.4rem;
|
||||
height: 1.4rem;
|
||||
padding-inline: 0.45rem;
|
||||
border-color: rgb(255 255 255 / 14%);
|
||||
background: rgb(255 255 255 / 5%);
|
||||
color: #cbd5e1;
|
||||
font-size: 0.6875rem;
|
||||
text-transform: capitalize;
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "wakey-agent"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
edition = "2024"
|
||||
publish = ["gitea"]
|
||||
|
||||
|
||||
@@ -4,6 +4,11 @@ use std::fmt;
|
||||
use std::path::{Path, PathBuf};
|
||||
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_PID_FILE: &str = "/var/run/wakey-agent.pid";
|
||||
const WAKEY_DHCP_LEASES_ENV: &str = "WAKEY_DHCP_LEASES";
|
||||
@@ -40,6 +45,7 @@ pub struct AgentConfig {
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct TerminalConfig {
|
||||
#[serde(default)]
|
||||
pub enabled: bool,
|
||||
@@ -51,6 +57,9 @@ pub struct TerminalConfig {
|
||||
pub current_dir: Option<PathBuf>,
|
||||
#[serde(default = "default_terminal_max_sessions")]
|
||||
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 {
|
||||
@@ -61,6 +70,7 @@ impl Default for TerminalConfig {
|
||||
args: Vec::new(),
|
||||
current_dir: None,
|
||||
max_sessions: default_terminal_max_sessions(),
|
||||
session_ttl_seconds: default_terminal_session_ttl_seconds(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -142,7 +152,17 @@ fn default_terminal_shell() -> PathBuf {
|
||||
}
|
||||
|
||||
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 {
|
||||
@@ -167,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> {
|
||||
let content = std::fs::read_to_string(path)
|
||||
.with_context(|| format!("failed to read agent config {}", path.display()))?;
|
||||
toml::from_str(&content)
|
||||
.with_context(|| format!("failed to parse agent config {}", path.display()))
|
||||
let config: AgentConfig = toml::from_str(&content)
|
||||
.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<()> {
|
||||
@@ -258,6 +280,7 @@ mod tests {
|
||||
args: vec!["-l".into()],
|
||||
current_dir: Some("/tmp".into()),
|
||||
max_sessions: 2,
|
||||
session_ttl_seconds: 600,
|
||||
},
|
||||
};
|
||||
|
||||
@@ -311,5 +334,45 @@ max_sessions = 2
|
||||
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -157,6 +157,7 @@ mod tests {
|
||||
args: vec!["-l".into()],
|
||||
current_dir: Some("/root".into()),
|
||||
max_sessions: 2,
|
||||
session_ttl_seconds: 600,
|
||||
},
|
||||
};
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ use macaddr::MacAddr;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::fmt;
|
||||
use std::net::IpAddr;
|
||||
use std::time::{Duration, Instant};
|
||||
use wakey_core::parse::mac;
|
||||
use wakey_core::{
|
||||
Device, DeviceInventory, DhcpLeaseWithState, InterfaceSummary, InventoryQuery,
|
||||
@@ -65,10 +66,59 @@ pub enum AgentCapability {
|
||||
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)]
|
||||
pub struct AgentTerminalSession {
|
||||
pub terminal_id: TerminalId,
|
||||
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)]
|
||||
@@ -238,6 +288,8 @@ pub enum ClientMessage {
|
||||
agent_id: String,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
capabilities: Vec<AgentCapability>,
|
||||
#[serde(default, skip_serializing_if = "AgentCapabilityOptions::is_empty")]
|
||||
capability_options: AgentCapabilityOptions,
|
||||
},
|
||||
Auth {
|
||||
agent_id: String,
|
||||
@@ -384,6 +436,7 @@ mod tests {
|
||||
sessions: vec![AgentTerminalSession {
|
||||
terminal_id: TerminalId::new("term-1").expect("terminal id"),
|
||||
created_at_unix: 42,
|
||||
session_ttl_seconds: 600,
|
||||
}],
|
||||
};
|
||||
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");
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -8,7 +8,10 @@ use tracing::{debug, error, info, info_span, warn};
|
||||
|
||||
use crate::config::AgentConfig;
|
||||
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;
|
||||
|
||||
pub async fn run(config: AgentConfig) -> Result<()> {
|
||||
@@ -82,6 +85,15 @@ async fn run_once(
|
||||
&ClientMessage::Hello {
|
||||
agent_id: config.agent_id.clone(),
|
||||
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?;
|
||||
|
||||
@@ -52,6 +52,7 @@ struct ActiveTerminal {
|
||||
cancel: Option<oneshot::Sender<()>>,
|
||||
relay_credentials: mpsc::UnboundedSender<String>,
|
||||
created_at_unix: u64,
|
||||
session_ttl_seconds: u64,
|
||||
}
|
||||
|
||||
/// Owns PTY workers independently of any individual control-plane connection.
|
||||
@@ -91,6 +92,7 @@ impl TerminalManager {
|
||||
anyhow::bail!("terminal capability is disabled");
|
||||
}
|
||||
validate_size(rows, cols)?;
|
||||
let session_ttl = config.terminal.session_ttl()?;
|
||||
|
||||
let terminal_key = terminal_id.to_string();
|
||||
let (cancel_tx, cancel_rx) = oneshot::channel();
|
||||
@@ -113,6 +115,7 @@ impl TerminalManager {
|
||||
cancel: Some(cancel_tx),
|
||||
relay_credentials: relay_tx.clone(),
|
||||
created_at_unix,
|
||||
session_ttl_seconds: config.terminal.session_ttl_seconds,
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -122,7 +125,16 @@ impl TerminalManager {
|
||||
let active = Arc::downgrade(&self.active);
|
||||
let events = self.events.clone();
|
||||
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
|
||||
// notifying the control session, which may immediately resync it.
|
||||
remove_completed(&active, terminal_id.as_str());
|
||||
@@ -176,6 +188,7 @@ impl TerminalManager {
|
||||
.map(|terminal_id| AgentTerminalSession {
|
||||
terminal_id,
|
||||
created_at_unix: active.created_at_unix,
|
||||
session_ttl_seconds: active.session_ttl_seconds,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
@@ -210,6 +223,7 @@ async fn run_terminal(
|
||||
terminal_id: &TerminalId,
|
||||
rows: u16,
|
||||
cols: u16,
|
||||
session_ttl: Option<Duration>,
|
||||
mut cancel: oneshot::Receiver<()>,
|
||||
mut relay_credentials: mpsc::UnboundedReceiver<String>,
|
||||
) -> Result<()> {
|
||||
@@ -242,8 +256,10 @@ async fn run_terminal(
|
||||
let mut terminal_state = TerminalState::new(rows, cols);
|
||||
let mut output = [0_u8; 16 * 1024];
|
||||
let mut requested_close = false;
|
||||
let mut ttl_expired = false;
|
||||
let mut observed_status = 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 {
|
||||
tokio::select! {
|
||||
biased;
|
||||
@@ -251,6 +267,15 @@ async fn run_terminal(
|
||||
requested_close = true;
|
||||
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() => {
|
||||
observed_status = Some(status.context("failed waiting for terminal child")?);
|
||||
drain_deadline = Some(Box::pin(tokio::time::sleep(PTY_EXIT_DRAIN_TIMEOUT)));
|
||||
@@ -363,7 +388,7 @@ async fn run_terminal(
|
||||
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(())
|
||||
}
|
||||
|
||||
@@ -511,6 +536,7 @@ async fn run_terminal(
|
||||
_terminal_id: &TerminalId,
|
||||
_rows: u16,
|
||||
_cols: u16,
|
||||
_session_ttl: Option<Duration>,
|
||||
_cancel: oneshot::Receiver<()>,
|
||||
_relay_credentials: mpsc::UnboundedReceiver<String>,
|
||||
) -> Result<()> {
|
||||
@@ -607,6 +633,7 @@ mod tests {
|
||||
cancel: Some(cancel),
|
||||
relay_credentials,
|
||||
created_at_unix: 42,
|
||||
session_ttl_seconds: 600,
|
||||
},
|
||||
)]))),
|
||||
max_sessions: 2,
|
||||
@@ -660,6 +687,7 @@ mod tests {
|
||||
cancel: Some(cancel),
|
||||
relay_credentials,
|
||||
created_at_unix: 42,
|
||||
session_ttl_seconds: 600,
|
||||
},
|
||||
)])));
|
||||
let manager = TerminalManager {
|
||||
@@ -702,6 +730,34 @@ mod tests {
|
||||
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]
|
||||
fn snapshot_reconstructs_screen_content_and_cursor() {
|
||||
let mut state = TerminalState::new(24, 80);
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "wakey-control-plane"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
edition = "2024"
|
||||
publish = ["gitea"]
|
||||
|
||||
|
||||
@@ -6,8 +6,9 @@ use serde::{Deserialize, Serialize};
|
||||
use std::time::Instant;
|
||||
use tracing::{info, info_span, warn};
|
||||
use uuid::Uuid;
|
||||
use wakey_agent::protocol::AgentCapability;
|
||||
use wakey_agent::protocol::{AgentCommand, ErrorPayload, RequestId, ServerMessage};
|
||||
use wakey_agent::protocol::{
|
||||
AgentCapability, AgentCapabilityOptions, AgentCommand, ErrorPayload, RequestId, ServerMessage,
|
||||
};
|
||||
|
||||
use crate::api::ApiError;
|
||||
use crate::runtime::{AgentReply, AppState, SessionEvent};
|
||||
@@ -19,6 +20,7 @@ pub struct AgentStatus {
|
||||
pub connected: bool,
|
||||
pub nickname: Option<String>,
|
||||
pub capabilities: Vec<AgentCapability>,
|
||||
pub capability_options: AgentCapabilityOptions,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
@@ -49,6 +51,10 @@ pub async fn list_agents(State(state): State<AppState>) -> Result<impl IntoRespo
|
||||
.get(&agent_id)
|
||||
.map(|session| session.capabilities.clone())
|
||||
.unwrap_or_default(),
|
||||
capability_options: sessions
|
||||
.get(&agent_id)
|
||||
.map(|session| session.capability_options.clone())
|
||||
.unwrap_or_default(),
|
||||
agent_id,
|
||||
nickname,
|
||||
})
|
||||
|
||||
@@ -5,16 +5,16 @@ use axum::http::StatusCode;
|
||||
use axum::response::Response;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::time::Duration;
|
||||
use tracing::{info, warn};
|
||||
use wakey_agent::protocol::{
|
||||
AgentCapability, ServerMessage, TerminalAgentHandshake, TerminalControl, TerminalId,
|
||||
TerminalOperatorHandshake,
|
||||
AgentCapability, DEFAULT_TERMINAL_MAX_SESSIONS, DEFAULT_TERMINAL_SESSION_TTL_SECONDS,
|
||||
ServerMessage, TerminalAgentHandshake, TerminalControl, TerminalId, TerminalOperatorHandshake,
|
||||
};
|
||||
|
||||
use crate::api::ApiError;
|
||||
use crate::runtime::terminals::{
|
||||
TERMINAL_ABSOLUTE_TIMEOUT, TERMINAL_ATTACH_TIMEOUT, TERMINAL_MAX_FRAME_BYTES,
|
||||
TerminalRelayFrame, TerminalSummary,
|
||||
TERMINAL_ATTACH_TIMEOUT, TERMINAL_MAX_FRAME_BYTES, TerminalRelayFrame, TerminalSummary,
|
||||
};
|
||||
use crate::runtime::{AppState, SessionEvent};
|
||||
use crate::state::AuditEventInput;
|
||||
@@ -40,6 +40,7 @@ pub struct TerminalSessionResponse {
|
||||
pub terminal_id: String,
|
||||
pub agent_id: String,
|
||||
pub created_at_unix: u64,
|
||||
pub expires_at_unix: Option<u64>,
|
||||
pub agent_attached: bool,
|
||||
pub operator_attached: bool,
|
||||
pub websocket_url: String,
|
||||
@@ -52,7 +53,7 @@ pub async fn create_terminal(
|
||||
Json(request): Json<CreateTerminalRequest>,
|
||||
) -> Result<(StatusCode, Json<TerminalSessionResponse>), ApiError> {
|
||||
validate_size(request.rows, request.cols)?;
|
||||
let agent_tx = {
|
||||
let (agent_tx, max_sessions, session_ttl_seconds) = {
|
||||
let sessions = state.sessions.read().await;
|
||||
let session = sessions.get(&request.agent_id).ok_or_else(|| {
|
||||
ApiError::new(
|
||||
@@ -68,12 +69,25 @@ pub async fn create_terminal(
|
||||
"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
|
||||
.terminals
|
||||
.create(request.agent_id.clone())
|
||||
.create_with_limits(request.agent_id.clone(), max_sessions, session_ttl_seconds)
|
||||
.await
|
||||
.map_err(registry_error)?;
|
||||
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(
|
||||
&state,
|
||||
&request.agent_id,
|
||||
@@ -122,6 +141,7 @@ pub async fn create_terminal(
|
||||
terminal_id: terminal_id.to_string(),
|
||||
agent_id: request.agent_id,
|
||||
created_at_unix: created.created_at_unix,
|
||||
expires_at_unix: created.expires_at_unix,
|
||||
agent_attached: false,
|
||||
operator_attached: false,
|
||||
attachment_token: Some(created.attachment_token),
|
||||
@@ -133,7 +153,7 @@ pub async fn get_terminal(
|
||||
State(state): State<AppState>,
|
||||
Path(terminal_id): Path<String>,
|
||||
) -> Result<Json<TerminalSessionResponse>, ApiError> {
|
||||
let (agent_id, created_at_unix, agent_attached, operator_attached) = state
|
||||
let summary = state
|
||||
.terminals
|
||||
.summary(&terminal_id)
|
||||
.await
|
||||
@@ -141,10 +161,11 @@ pub async fn get_terminal(
|
||||
Ok(Json(TerminalSessionResponse {
|
||||
websocket_url: operator_ws_path(&terminal_id),
|
||||
terminal_id,
|
||||
agent_id,
|
||||
created_at_unix,
|
||||
agent_attached,
|
||||
operator_attached,
|
||||
agent_id: summary.agent_id,
|
||||
created_at_unix: summary.created_at_unix,
|
||||
expires_at_unix: summary.expires_at_unix,
|
||||
agent_attached: summary.agent_attached,
|
||||
operator_attached: summary.operator_attached,
|
||||
attachment_token: None,
|
||||
}))
|
||||
}
|
||||
@@ -171,7 +192,7 @@ pub async fn attach_terminal(
|
||||
.issue_attachment_token_for_operator(&terminal_id, &request.operator_id)
|
||||
.await
|
||||
.map_err(registry_error)?;
|
||||
let (agent_id, created_at_unix, agent_attached, operator_attached) = state
|
||||
let summary = state
|
||||
.terminals
|
||||
.summary(&terminal_id)
|
||||
.await
|
||||
@@ -179,10 +200,11 @@ pub async fn attach_terminal(
|
||||
Ok(Json(TerminalSessionResponse {
|
||||
websocket_url: operator_ws_path(&terminal_id),
|
||||
terminal_id,
|
||||
agent_id,
|
||||
created_at_unix,
|
||||
agent_attached,
|
||||
operator_attached,
|
||||
agent_id: summary.agent_id,
|
||||
created_at_unix: summary.created_at_unix,
|
||||
expires_at_unix: summary.expires_at_unix,
|
||||
agent_attached: summary.agent_attached,
|
||||
operator_attached: summary.operator_attached,
|
||||
attachment_token: Some(attachment_token),
|
||||
}))
|
||||
}
|
||||
@@ -343,10 +365,10 @@ async fn handle_operator_terminal_socket(
|
||||
warn!(terminal_id, code, "failed to request terminal snapshot");
|
||||
}
|
||||
let summary = state.terminals.summary(&terminal_id).await;
|
||||
if let Some((agent_id, _, _, _)) = &summary {
|
||||
if let Some(summary) = &summary {
|
||||
append_terminal_audit(
|
||||
&state,
|
||||
agent_id,
|
||||
&summary.agent_id,
|
||||
&terminal_id,
|
||||
TerminalAudit {
|
||||
actor_type: "admin_api",
|
||||
@@ -360,7 +382,7 @@ async fn handle_operator_terminal_socket(
|
||||
}
|
||||
|
||||
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)
|
||||
.expect("terminal ready control serializes");
|
||||
if send_relay_frame(&mut write, TerminalRelayFrame::Text(ready))
|
||||
@@ -414,6 +436,7 @@ fn terminal_response(summary: TerminalSummary) -> TerminalSessionResponse {
|
||||
terminal_id: summary.terminal_id,
|
||||
agent_id: summary.agent_id,
|
||||
created_at_unix: summary.created_at_unix,
|
||||
expires_at_unix: summary.expires_at_unix,
|
||||
agent_attached: summary.agent_attached,
|
||||
operator_attached: summary.operator_attached,
|
||||
attachment_token: None,
|
||||
@@ -549,9 +572,17 @@ async fn close_registered_terminal(
|
||||
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::time::sleep(TERMINAL_ABSOLUTE_TIMEOUT).await;
|
||||
tokio::time::sleep(Duration::from_secs(session_ttl_seconds)).await;
|
||||
if close_registered_terminal(
|
||||
&state,
|
||||
terminal_id.as_str(),
|
||||
|
||||
@@ -16,7 +16,7 @@ use tower_http::services::ServeFile;
|
||||
use tracing::info;
|
||||
#[cfg(unix)]
|
||||
use tracing::warn;
|
||||
use wakey_agent::protocol::{AgentCapability, ErrorPayload, ServerMessage};
|
||||
use wakey_agent::protocol::{AgentCapability, AgentCapabilityOptions, ErrorPayload, ServerMessage};
|
||||
|
||||
use crate::api;
|
||||
use crate::config;
|
||||
@@ -50,6 +50,7 @@ pub struct AgentSession {
|
||||
pub connection_id: String,
|
||||
pub tx: mpsc::UnboundedSender<SessionEvent>,
|
||||
pub capabilities: Vec<AgentCapability>,
|
||||
pub capability_options: AgentCapabilityOptions,
|
||||
}
|
||||
#[derive(Clone)]
|
||||
pub enum SessionEvent {
|
||||
|
||||
@@ -9,9 +9,7 @@ use wakey_agent::protocol::{AgentTerminalSession, TerminalId};
|
||||
pub const TERMINAL_RELAY_QUEUE: usize = 32;
|
||||
pub const TERMINAL_MAX_FRAME_BYTES: usize = 64 * 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_ABSOLUTE_TIMEOUT: Duration = Duration::from_secs(12 * 60 * 60);
|
||||
const TERMINAL_TOMBSTONE_TTL: Duration = Duration::from_secs(5 * 60);
|
||||
const TERMINAL_MAX_TOMBSTONES: usize = 1024;
|
||||
|
||||
@@ -31,7 +29,7 @@ pub struct TerminalRegistry {
|
||||
struct TerminalSession {
|
||||
agent_id: String,
|
||||
created_at_unix: u64,
|
||||
expires_at: Instant,
|
||||
expiry: TerminalExpiry,
|
||||
agent_confirmed: bool,
|
||||
relay_token: Option<String>,
|
||||
attachment_token: Option<String>,
|
||||
@@ -50,6 +48,7 @@ pub struct CreatedTerminal {
|
||||
pub relay_token: String,
|
||||
pub attachment_token: String,
|
||||
pub created_at_unix: u64,
|
||||
pub expires_at_unix: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
@@ -57,6 +56,7 @@ pub struct TerminalSummary {
|
||||
pub terminal_id: String,
|
||||
pub agent_id: String,
|
||||
pub created_at_unix: u64,
|
||||
pub expires_at_unix: Option<u64>,
|
||||
pub agent_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;
|
||||
prune_expired_sessions(&mut sessions);
|
||||
if sessions
|
||||
.values()
|
||||
.filter(|session| session.agent_id == agent_id)
|
||||
.count()
|
||||
>= TERMINAL_MAX_SESSIONS_PER_AGENT
|
||||
>= max_sessions.max(1)
|
||||
{
|
||||
return Err("agent_terminal_limit_reached");
|
||||
}
|
||||
@@ -94,12 +112,14 @@ impl TerminalRegistry {
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs();
|
||||
let expiry = expiry_for_created_at(created_at_unix, session_ttl_seconds)
|
||||
.ok_or("terminal_ttl_invalid")?;
|
||||
sessions.insert(
|
||||
terminal_id.clone(),
|
||||
TerminalSession {
|
||||
agent_id,
|
||||
created_at_unix,
|
||||
expires_at: Instant::now() + TERMINAL_ABSOLUTE_TIMEOUT,
|
||||
expiry,
|
||||
agent_confirmed: false,
|
||||
relay_token: Some(relay_token.clone()),
|
||||
attachment_token: Some(attachment_token.clone()),
|
||||
@@ -119,6 +139,7 @@ impl TerminalRegistry {
|
||||
relay_token,
|
||||
attachment_token,
|
||||
created_at_unix,
|
||||
expires_at_unix: expiry.unix(),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -169,7 +190,10 @@ impl TerminalRegistry {
|
||||
) -> Vec<(TerminalId, String)> {
|
||||
let reported_ids = reported
|
||||
.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())
|
||||
.collect::<std::collections::HashSet<_>>();
|
||||
let stale_ids = {
|
||||
@@ -192,8 +216,10 @@ impl TerminalRegistry {
|
||||
let mut credentials = Vec::new();
|
||||
let mut sessions = self.inner.lock().await;
|
||||
for reported_session in reported {
|
||||
let Some(expires_at) = expires_at_for_created_at(reported_session.created_at_unix)
|
||||
else {
|
||||
let Some(expiry) = expiry_for_created_at(
|
||||
reported_session.created_at_unix,
|
||||
reported_session.session_ttl_seconds,
|
||||
) else {
|
||||
continue;
|
||||
};
|
||||
let terminal_id = reported_session.terminal_id.as_str().to_string();
|
||||
@@ -202,7 +228,7 @@ impl TerminalRegistry {
|
||||
.or_insert_with(|| TerminalSession {
|
||||
agent_id: agent_id.to_string(),
|
||||
created_at_unix: reported_session.created_at_unix,
|
||||
expires_at,
|
||||
expiry,
|
||||
agent_confirmed: true,
|
||||
relay_token: None,
|
||||
attachment_token: None,
|
||||
@@ -422,17 +448,12 @@ impl TerminalRegistry {
|
||||
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;
|
||||
prune_expired_sessions(&mut sessions);
|
||||
sessions.get(terminal_id).map(|session| {
|
||||
(
|
||||
session.agent_id.clone(),
|
||||
session.created_at_unix,
|
||||
session.agent_tx.is_some(),
|
||||
session.operator_tx.is_some(),
|
||||
)
|
||||
})
|
||||
sessions
|
||||
.get(terminal_id)
|
||||
.map(|session| terminal_summary(terminal_id, session))
|
||||
}
|
||||
|
||||
pub async fn summaries(&self) -> Vec<TerminalSummary> {
|
||||
@@ -440,26 +461,31 @@ impl TerminalRegistry {
|
||||
prune_expired_sessions(&mut sessions);
|
||||
let mut summaries = sessions
|
||||
.iter()
|
||||
.map(|(terminal_id, session)| TerminalSummary {
|
||||
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(),
|
||||
})
|
||||
.map(|(terminal_id, session)| terminal_summary(terminal_id, session))
|
||||
.collect::<Vec<_>>();
|
||||
summaries.sort_by_key(|session| std::cmp::Reverse(session.created_at_unix));
|
||||
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>(
|
||||
sessions: &'a mut HashMap<String, TerminalSession>,
|
||||
terminal_id: &str,
|
||||
) -> Result<&'a mut TerminalSession, &'static str> {
|
||||
if sessions
|
||||
.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);
|
||||
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>) {
|
||||
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
|
||||
/// deadline. Reconciliation must not grant an old PTY a fresh twelve hours.
|
||||
fn expires_at_for_created_at(created_at_unix: u64) -> Option<Instant> {
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
enum TerminalExpiry {
|
||||
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()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs();
|
||||
let age = Duration::from_secs(now_unix.saturating_sub(created_at_unix));
|
||||
if age >= TERMINAL_ABSOLUTE_TIMEOUT {
|
||||
if expires_at_unix <= now_unix {
|
||||
return None;
|
||||
}
|
||||
let remaining = TERMINAL_ABSOLUTE_TIMEOUT - age;
|
||||
Some(Instant::now() + remaining)
|
||||
let remaining = Duration::from_secs(expires_at_unix - now_unix);
|
||||
Some(TerminalExpiry::Finite {
|
||||
deadline: Instant::now().checked_add(remaining)?,
|
||||
unix: expires_at_unix,
|
||||
})
|
||||
}
|
||||
|
||||
fn relay_frame_size(frame: &TerminalRelayFrame) -> usize {
|
||||
@@ -528,6 +580,7 @@ fn relay_frame_size(frame: &TerminalRelayFrame) -> usize {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use wakey_agent::protocol::DEFAULT_TERMINAL_SESSION_TTL_SECONDS;
|
||||
|
||||
const OPERATOR_A: &str = "browser-tab-a";
|
||||
const OPERATOR_B: &str = "browser-tab-b";
|
||||
@@ -643,6 +696,73 @@ mod tests {
|
||||
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]
|
||||
async fn removed_session_is_distinct_from_unknown_session() {
|
||||
let registry = TerminalRegistry::new();
|
||||
@@ -663,6 +783,7 @@ mod tests {
|
||||
let reported = AgentTerminalSession {
|
||||
terminal_id: terminal_id.clone(),
|
||||
created_at_unix,
|
||||
session_ttl_seconds: 0,
|
||||
};
|
||||
|
||||
let credentials = registry
|
||||
@@ -679,9 +800,9 @@ mod tests {
|
||||
.summary(terminal_id.as_str())
|
||||
.await
|
||||
.expect("adopted summary");
|
||||
assert_eq!(summary.0, "router");
|
||||
assert_eq!(summary.1, created_at_unix);
|
||||
assert!(summary.2);
|
||||
assert_eq!(summary.agent_id, "router");
|
||||
assert_eq!(summary.created_at_unix, created_at_unix);
|
||||
assert!(summary.agent_attached);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -695,7 +816,10 @@ mod tests {
|
||||
.await
|
||||
.get_mut(&first.terminal_id)
|
||||
.expect("first session")
|
||||
.expires_at = Instant::now();
|
||||
.expiry = TerminalExpiry::Finite {
|
||||
deadline: Instant::now(),
|
||||
unix: first.created_at_unix,
|
||||
};
|
||||
|
||||
registry
|
||||
.create("router".into())
|
||||
@@ -711,6 +835,7 @@ mod tests {
|
||||
let reported = AgentTerminalSession {
|
||||
terminal_id: terminal_id.clone(),
|
||||
created_at_unix: 0,
|
||||
session_ttl_seconds: DEFAULT_TERMINAL_SESSION_TTL_SECONDS,
|
||||
};
|
||||
|
||||
assert!(
|
||||
@@ -739,7 +864,7 @@ mod tests {
|
||||
.summary(&created.terminal_id)
|
||||
.await
|
||||
.expect("session remains after detach");
|
||||
assert!(!summary.3);
|
||||
assert!(!summary.operator_attached);
|
||||
registry
|
||||
.issue_attachment_token_for_operator(&created.terminal_id, OPERATOR_A)
|
||||
.await
|
||||
@@ -765,7 +890,13 @@ mod tests {
|
||||
.await
|
||||
.expect("handoff token");
|
||||
|
||||
assert!(!registry.summary(&previous.terminal_id).await.unwrap().3);
|
||||
assert!(
|
||||
!registry
|
||||
.summary(&previous.terminal_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.operator_attached
|
||||
);
|
||||
registry
|
||||
.attach_operator(&next.terminal_id, &next_token, OPERATOR_A)
|
||||
.await
|
||||
@@ -800,8 +931,20 @@ mod tests {
|
||||
.expect_err("occupied target remains protected");
|
||||
|
||||
assert_eq!(error, "terminal_operator_already_attached");
|
||||
assert!(registry.summary(&previous.terminal_id).await.unwrap().3);
|
||||
assert!(registry.summary(&occupied.terminal_id).await.unwrap().3);
|
||||
assert!(
|
||||
registry
|
||||
.summary(&previous.terminal_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.operator_attached
|
||||
);
|
||||
assert!(
|
||||
registry
|
||||
.summary(&occupied.terminal_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.operator_attached
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -829,7 +972,13 @@ mod tests {
|
||||
.await
|
||||
.is_none()
|
||||
);
|
||||
assert!(registry.summary(&created.terminal_id).await.unwrap().3);
|
||||
assert!(
|
||||
registry
|
||||
.summary(&created.terminal_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.operator_attached
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -9,8 +9,9 @@ use tokio::sync::mpsc;
|
||||
use tracing::{debug, info, info_span, warn};
|
||||
use uuid::Uuid;
|
||||
use wakey_agent::protocol::{
|
||||
AgentCapability, AgentTerminalSession, ErrorPayload, RequestId, ServerMessage, TerminalControl,
|
||||
TerminalId,
|
||||
AgentCapability, AgentCapabilityOptions, AgentTerminalSession, DEFAULT_TERMINAL_MAX_SESSIONS,
|
||||
DEFAULT_TERMINAL_SESSION_TTL_SECONDS, ErrorPayload, RequestId, ServerMessage,
|
||||
TerminalCapabilityOptions, TerminalControl, TerminalId, checked_terminal_session_ttl,
|
||||
};
|
||||
use wakey_core::Device;
|
||||
|
||||
@@ -24,6 +25,8 @@ enum IncomingClientMessage {
|
||||
agent_id: String,
|
||||
#[serde(default)]
|
||||
capabilities: Vec<AgentCapability>,
|
||||
#[serde(default)]
|
||||
capability_options: AgentCapabilityOptions,
|
||||
},
|
||||
Auth {
|
||||
agent_id: String,
|
||||
@@ -59,6 +62,7 @@ struct AgentConnectionState {
|
||||
hello_agent_id: Option<String>,
|
||||
hello_at: Option<Instant>,
|
||||
capabilities: Vec<AgentCapability>,
|
||||
capability_options: AgentCapabilityOptions,
|
||||
}
|
||||
|
||||
pub async fn agent_ws(ws: WebSocketUpgrade, State(state): State<AppState>) -> impl IntoResponse {
|
||||
@@ -193,6 +197,7 @@ async fn process_agent_text(
|
||||
IncomingClientMessage::Hello {
|
||||
agent_id,
|
||||
capabilities,
|
||||
capability_options,
|
||||
} => {
|
||||
if connection
|
||||
.hello_agent_id
|
||||
@@ -208,6 +213,27 @@ async fn process_agent_text(
|
||||
let connect_to_hello_ms = connected_at.elapsed().as_millis() as u64;
|
||||
info!(agent_id = %agent_id, connect_to_hello_ms, "agent hello received");
|
||||
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;
|
||||
}
|
||||
IncomingClientMessage::Auth {
|
||||
@@ -253,6 +279,7 @@ async fn process_agent_text(
|
||||
connection_id: connection_id.to_string(),
|
||||
tx: tx.clone(),
|
||||
capabilities: connection.capabilities.clone(),
|
||||
capability_options: connection.capability_options.clone(),
|
||||
},
|
||||
);
|
||||
connection.authed_agent_id = Some(agent_id.clone());
|
||||
@@ -424,10 +451,13 @@ mod tests {
|
||||
use std::collections::HashMap;
|
||||
|
||||
use tokio::sync::mpsc;
|
||||
use wakey_agent::protocol::AgentCapabilityOptions;
|
||||
|
||||
use crate::runtime::AgentSession;
|
||||
|
||||
use super::{AgentConnectionState, is_current_session, validate_auth_identity};
|
||||
use super::{
|
||||
AgentConnectionState, IncomingClientMessage, is_current_session, validate_auth_identity,
|
||||
};
|
||||
|
||||
#[test]
|
||||
fn current_session_check_rejects_stale_connection_ids() {
|
||||
@@ -439,6 +469,7 @@ mod tests {
|
||||
connection_id: "conn-new".to_string(),
|
||||
tx,
|
||||
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-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,6 +1,6 @@
|
||||
[package]
|
||||
name = "wakey-core"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
edition = "2024"
|
||||
publish = ["gitea"]
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "wakey-linux"
|
||||
version = "0.3.2"
|
||||
version = "0.3.3"
|
||||
edition = "2024"
|
||||
publish = ["gitea"]
|
||||
|
||||
|
||||
Reference in new issue
Block a user