diff --git a/AGENTS.md b/AGENTS.md index 8253070..f635330 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -149,7 +149,7 @@ ### Formats And Versions -* **Control messages:** any change to a `Request` or `Response` shape in `ally-core/src/control.rs` raises `control::VERSION`, currently 14. +* **Control messages:** any change to a `Request` or `Response` shape in `ally-core/src/control.rs` raises `control::VERSION`, currently 15. * **Menu bar lines:** if a field in `ally-cli/src/watch/event.rs` changes meaning or is removed, raise `watch::event::VERSION` and change `ally-cli/menubar/Sources/Model.swift` in the same commit. * **Pairing messages:** a change to the pairing messages in `ally-core/src/protocol.rs` means both machines must update before the next link. A change to `Specs` or `Health`, which are saved or sent after pairing, forces every user to link again. Say which one explicitly. diff --git a/crates/ally-agent/src/jobs.rs b/crates/ally-agent/src/jobs.rs index 14b8ee7..38aea3e 100644 --- a/crates/ally-agent/src/jobs.rs +++ b/crates/ally-agent/src/jobs.rs @@ -495,10 +495,29 @@ fn stop_run(job: &Job) { kill_remaining(&tree); } -fn stop_session(root: &Path, job: &Job) { - let target = format!("={}", job.id); - let panes: Vec = tmux(root) - .args(["list-panes", "-s", "-t", &target, "-F", "#{pane_pid}"]) +/// Every live process that belongs to an active job, with its start time. A run whose +/// leader no longer has the recorded start time owns nothing, so a reused PID never counts. +pub fn processes(root: &Path, job: &Job) -> HashMap { + match (job.kind, job.pid, job.process_start) { + (JobKind::Session, _, _) => process_tree(&session_panes(root, &job.id), None), + (JobKind::Run, Some(pid), Some(start)) if process_start(pid) == Some(start) => { + process_tree(&[pid], Some(pid)) + } + (JobKind::Run, _, _) => HashMap::new(), + } +} + +/// The shell process of each pane in a session. +fn session_panes(root: &Path, id: &str) -> Vec { + tmux(root) + .args([ + "list-panes", + "-s", + "-t", + &format!("={id}"), + "-F", + "#{pane_pid}", + ]) .output() .map(|out| { String::from_utf8_lossy(&out.stdout) @@ -506,7 +525,12 @@ fn stop_session(root: &Path, job: &Job) { .filter_map(|line| line.trim().parse().ok()) .collect() }) - .unwrap_or_default(); + .unwrap_or_default() +} + +fn stop_session(root: &Path, job: &Job) { + let target = format!("={}", job.id); + let panes = session_panes(root, &job.id); let tree = process_tree(&panes, None); let shells: HashMap = tree .iter() @@ -607,6 +631,18 @@ pub fn signal_group(pid: u32, start: u64, signal: i32) -> bool { mod tests { use super::*; + /// A run whose leader has another start time is a reused PID, which owns nothing. + #[test] + fn a_run_owns_its_processes_only_while_its_leader_matches() { + let me = std::process::id(); + let mut job = new_job(JobKind::Run, None, "test".to_string()); + job.pid = Some(me); + job.process_start = process_start(me); + assert!(processes(Path::new("/nonexistent"), &job).contains_key(&me)); + job.process_start = job.process_start.map(|start| start + 1); + assert!(processes(Path::new("/nonexistent"), &job).is_empty()); + } + #[test] fn sessions_get_color_mouse_and_the_login_shell() { let config = tmux_config("/usr/bin/bash", "tmux-256color"); diff --git a/crates/ally-agent/src/lib.rs b/crates/ally-agent/src/lib.rs index f4b5213..5c8c572 100644 --- a/crates/ally-agent/src/lib.rs +++ b/crates/ally-agent/src/lib.rs @@ -5,6 +5,7 @@ pub mod awake; pub mod boot; pub mod clients; pub mod jobs; +pub mod ports; pub mod projects; pub mod runner; pub mod service; diff --git a/crates/ally-agent/src/ports.rs b/crates/ally-agent/src/ports.rs new file mode 100644 index 0000000..3f5e793 --- /dev/null +++ b/crates/ally-agent/src/ports.rs @@ -0,0 +1,330 @@ +//! The TCP ports that Ally's work listens on, for port forwarding. Sockets come from `ss` +//! on Linux and `lsof` on macOS, and count only when a job's process holds them. Ports +//! that Docker containers publish come from `docker ps`, since Docker's own processes +//! hold them outside any job. + +use crate::jobs; +use ally_core::control::Listening; +use ally_core::preflight; +use ally_core::telemetry::is_installed; +use std::collections::HashMap; +use std::path::Path; +use std::process::Command; + +/// A listening TCP socket and the process that holds it. +#[derive(Debug, PartialEq)] +struct Socket { + address: String, + port: u16, + pid: u32, + program: String, +} + +/// Every port that an active job's processes or a Docker container listen on, one entry +/// per port. `ss` or `lsof` runs only while a job is active. +pub fn list(root: &Path) -> anyhow::Result> { + let mut owners = HashMap::new(); + for job in jobs::list(root, false)? { + for pid in jobs::processes(root, &job).into_keys() { + owners.insert(pid, job.id.clone()); + } + } + let mut found = Vec::new(); + if !owners.is_empty() { + for socket in sockets()? { + if let Some(job) = owners.get(&socket.pid) { + found.push(Listening { + port: socket.port, + address: socket.address, + program: socket.program, + job: Some(job.clone()), + }); + } + } + } + found.extend(docker()); + Ok(merged(found)) +} + +fn sockets() -> anyhow::Result> { + match std::env::consts::OS { + "macos" => Ok(from_lsof(&output( + "lsof", + &["-nP", "-iTCP", "-sTCP:LISTEN", "-F", "pcn"], + "lsof", + )?)), + _ => Ok(from_ss(&output("ss", &["-Hltnp"], "iproute2")?)), + } +} + +/// What `program` prints. lsof exits 1 when nothing listens, so the exit code is ignored. +fn output(program: &str, args: &[&str], package: &str) -> anyhow::Result { + match Command::new(program).args(args).output() { + Ok(out) => Ok(String::from_utf8_lossy(&out.stdout).into_owned()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => anyhow::bail!( + "{program} is not installed on the Agent, so its ports cannot be found. Fix: {}", + preflight::install_hint(package) + ), + Err(error) => Err(anyhow::Error::from(error).context(format!("Could not run {program}"))), + } +} + +/// Lines of `ss -Hltnp`, such as +/// `LISTEN 0 511 127.0.0.1:5173 0.0.0.0:* users:(("node",pid=7417,fd=23))`. +/// ss names processes only for this account, which are the only ones a job can own. +fn from_ss(text: &str) -> Vec { + let mut sockets = Vec::new(); + for line in text.lines() { + let Some((address, port)) = line.split_whitespace().nth(3).and_then(split_address) else { + continue; + }; + let users = line.split_once("users:(").map_or("", |(_, rest)| rest); + for entry in users.split("(\"").skip(1) { + let Some((program, rest)) = entry.split_once("\",") else { + continue; + }; + let digits: String = rest + .strip_prefix("pid=") + .unwrap_or("") + .chars() + .take_while(char::is_ascii_digit) + .collect(); + if let Ok(pid) = digits.parse() { + sockets.push(Socket { + address: address.clone(), + port, + pid, + program: program.to_string(), + }); + } + } + } + sockets +} + +/// The field output of `lsof -F pcn`: a `p` line with the process ID, a `c` line with its +/// name, then an `n` line for each listening address. +fn from_lsof(text: &str) -> Vec { + let mut sockets = Vec::new(); + let mut pid = None; + let mut program = String::new(); + for line in text.lines() { + match line.split_at_checked(1) { + Some(("p", value)) => pid = value.parse().ok(), + Some(("c", value)) => program = value.to_string(), + Some(("n", value)) => { + if let (Some(pid), Some((address, port))) = (pid, split_address(value)) { + sockets.push(Socket { + address, + port, + pid, + program: program.clone(), + }); + } + } + _ => {} + } + } + sockets +} + +/// Ports that running containers publish. Docker that is missing or refuses this account +/// means no containers to report, never an error. +fn docker() -> Vec { + if !is_installed("docker") { + return Vec::new(); + } + let Ok(out) = Command::new("docker") + .args(["ps", "--format", "{{.Names}}\t{{.Ports}}"]) + .output() + else { + return Vec::new(); + }; + from_docker(&String::from_utf8_lossy(&out.stdout)) +} + +/// Lines of `docker ps`, a name and its ports, such as +/// `db\t0.0.0.0:5432->5432/tcp, [::]:5432->5432/tcp`. A port without `->` is not +/// published, and UDP cannot be forwarded over ssh. +fn from_docker(text: &str) -> Vec { + let mut found = Vec::new(); + for line in text.lines() { + let Some((name, ports)) = line.split_once('\t') else { + continue; + }; + for published in ports.split(", ") { + let Some((host, inside)) = published.split_once("->") else { + continue; + }; + if !inside.ends_with("/tcp") { + continue; + } + if let Some((address, port)) = split_address(host) { + found.push(Listening { + port, + address, + program: name.to_string(), + job: None, + }); + } + } + } + found +} + +/// An address and port such as `127.0.0.1:3000`, `[::1]:3000`, `*:3000`, or +/// `127.0.0.53%lo:53`, as the address alone and the port. `*` means every address. +fn split_address(text: &str) -> Option<(String, u16)> { + let (host, port) = text.rsplit_once(':')?; + let port = port.parse().ok()?; + let host = host.trim_start_matches('[').trim_end_matches(']'); + let host = host.split('%').next().unwrap_or(host); + match host { + "*" | "" => Some(("0.0.0.0".to_string(), port)), + host => Some((host.to_string(), port)), + } +} + +/// One entry per port. A server often listens on both IPv4 and IPv6, and IPv4 is kept, +/// since it reaches a server that listens on every address either way. +fn merged(mut found: Vec) -> Vec { + found.sort_by_key(|listening| (listening.port, listening.address.contains(':'))); + found.dedup_by_key(|listening| listening.port); + found +} + +#[cfg(test)] +mod tests { + use super::*; + + fn socket(address: &str, port: u16, pid: u32, program: &str) -> Socket { + Socket { + address: address.to_string(), + port, + pid, + program: program.to_string(), + } + } + + fn listening(address: &str, port: u16, job: Option<&str>) -> Listening { + Listening { + port, + address: address.to_string(), + program: "node".to_string(), + job: job.map(str::to_string), + } + } + + /// Copied from an Arch Linux Agent. Sockets of other accounts carry no process. + /// Runs the real `ss` or `lsof`, which must name this test's own listening socket. + #[test] + fn this_machines_tool_finds_a_port_this_process_listens_on() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let found = sockets().unwrap(); + assert!( + found + .iter() + .any(|s| s.port == port && s.pid == std::process::id() && s.address == "127.0.0.1"), + "{found:?}" + ); + } + + #[test] + fn ss_lines_give_this_accounts_sockets_with_their_processes() { + let text = "\ +LISTEN 0 4096 100.67.90.119:48074 0.0.0.0:* +LISTEN 0 5 127.0.0.1:8765 0.0.0.0:* users:((\"python\",pid=7417,fd=3)) +LISTEN 0 511 *:5173 *:* users:((\"node\",pid=88,fd=20),(\"node\",pid=89,fd=20)) +LISTEN 0 511 [::1]:3000 [::]:* users:((\"next-server\",pid=12,fd=21)) +LISTEN 0 128 [::]:22 [::]:* +"; + assert_eq!( + from_ss(text), + [ + socket("127.0.0.1", 8765, 7417, "python"), + socket("0.0.0.0", 5173, 88, "node"), + socket("0.0.0.0", 5173, 89, "node"), + socket("::1", 3000, 12, "next-server"), + ] + ); + } + + #[test] + fn lsof_fields_give_each_process_with_its_addresses() { + let text = + "p7417\ncpython\nf3\nn127.0.0.1:8765\np90\ncnode\nf20\nn*:5173\nf21\nn[::1]:5173\n"; + assert_eq!( + from_lsof(text), + [ + socket("127.0.0.1", 8765, 7417, "python"), + socket("0.0.0.0", 5173, 90, "node"), + socket("::1", 5173, 90, "node"), + ] + ); + } + + #[test] + fn docker_gives_only_published_tcp_ports() { + let text = "db\t0.0.0.0:5432->5432/tcp, [::]:5432->5432/tcp\nweb\t127.0.0.1:8080->80/tcp, 443/tcp\ndns\t0.0.0.0:53->53/udp\nidle\t\n"; + let found = from_docker(text); + let ports: Vec<_> = found + .iter() + .map(|l| { + ( + l.program.as_str(), + l.address.as_str(), + l.port, + l.job.is_none(), + ) + }) + .collect(); + assert_eq!( + ports, + [ + ("db", "0.0.0.0", 5432, true), + ("db", "::", 5432, true), + ("web", "127.0.0.1", 8080, true), + ] + ); + } + + #[test] + fn addresses_lose_brackets_and_interfaces_and_a_star_means_every_address() { + for (text, address, port) in [ + ("127.0.0.1:3000", "127.0.0.1", 3000), + ("[::1]:3000", "::1", 3000), + ("*:3000", "0.0.0.0", 3000), + ("127.0.0.53%lo:53", "127.0.0.53", 53), + ("[fe80::1%eth0]:8080", "fe80::1", 8080), + (":::5432", "::", 5432), + ] { + assert_eq!( + split_address(text), + Some((address.to_string(), port)), + "{text}" + ); + } + assert_eq!(split_address("127.0.0.1:8000-8001"), None); + assert_eq!(split_address("nothing"), None); + } + + #[test] + fn each_port_appears_once_and_ipv4_wins() { + let found = merged(vec![ + listening("::", 5432, None), + listening("0.0.0.0", 5432, None), + listening("::1", 3000, Some("ab12")), + listening("127.0.0.1", 8765, Some("ab12")), + listening("127.0.0.1", 8765, Some("ab12")), + ]); + assert_eq!( + found, + [ + listening("::1", 3000, Some("ab12")), + listening("0.0.0.0", 5432, None), + listening("127.0.0.1", 8765, Some("ab12")), + ] + ); + } +} diff --git a/crates/ally-agent/src/service.rs b/crates/ally-agent/src/service.rs index 9b00ffb..d3cba44 100644 --- a/crates/ally-agent/src/service.rs +++ b/crates/ally-agent/src/service.rs @@ -4,7 +4,7 @@ //! through that SSH login, relays bytes between the connection and a Unix socket that //! only the Agent account can open. The Agent OS account is the trust boundary. -use crate::{jobs, projects}; +use crate::{jobs, ports, projects}; use ally_core::control::{self, AgentStatus, AgentTools, Request, Response}; use ally_core::{preflight, storage, telemetry, tools}; use anyhow::{Context, bail}; @@ -273,6 +273,7 @@ fn respond(root: &Path, name: &str, request: Request) -> anyhow::Result Response::Ports(ports::list(root)?), Request::Ping | Request::Status | Request::Begin { .. } diff --git a/crates/ally-core/src/control.rs b/crates/ally-core/src/control.rs index 208ca79..f74d652 100644 --- a/crates/ally-core/src/control.rs +++ b/crates/ally-core/src/control.rs @@ -14,7 +14,7 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; /// Wire version of this control protocol. Both machines must agree on it, so any change /// to a request or response shape has to raise it. -pub const VERSION: u32 = 14; +pub const VERSION: u32 = 15; /// Largest frame in either direction. Manifests for very large projects are the limit. pub const MAX_FRAME: u32 = 16 * 1024 * 1024; @@ -136,6 +136,8 @@ pub enum Request { RemoveProject { project: String, }, + /// The TCP ports that Ally's work, and Docker's containers, listen on. + Ports, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -158,6 +160,19 @@ pub enum Response { Status(AgentStatus), Projects(Vec), Cleaned { bytes: u64 }, + Ports(Vec), +} + +/// A TCP port that work on the Agent listens on, which port forwarding can reach. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Listening { + pub port: u16, + /// The address it listens on, such as `127.0.0.1`, `::1`, or `0.0.0.0` for every one. + pub address: String, + /// The program listening, such as `node`, or the container's name for Docker. + pub program: String, + /// The job it belongs to, or `None` for a port a Docker container publishes. + pub job: Option, } /// One project copy on the Agent, as `ally projects` lists it. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e2f2235..c32a41c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -280,7 +280,7 @@ ally/ | --- | --- | --- | --- | | `lib.rs` | Lists the modules. | none | the other crates | | `protocol.rs` | Pairing messages and the SPAKE2 handshake, hardware details, and health values. | none | `config`, `control`, `telemetry`, Agent `lib`, several Client commands | -| `control.rs` | Versioned control messages, job records, and message reading and writing. | `protocol`, `source`, `tools` | Agent `service`, `jobs`, `projects`, `runner`; Client `client`, `transfer`, most commands | +| `control.rs` | Versioned control messages, job records, and message reading and writing. | `protocol`, `source`, `tools` | Agent `service`, `jobs`, `ports`, `projects`, `runner`; Client `client`, `transfer`, most commands | | `config.rs` | The Client's saved Agents: loading, saving, and choosing one. | `network`, `protocol` | `keys` and nearly every Client file | | `network.rs` | Tells local, tailnet, and other addresses apart. | none | `config`, Agent `lib`, Client `route` | | `tunnel.rs` | What both machines share for iroh: the connection name and a saved identity. | `storage`, `source` | Agent `lib`, `clients`, `tunnel`; Client `route`, `tunnel`, `link` | @@ -290,8 +290,8 @@ ally/ | `sync.rs` | Three way sync: planning, conflicts, staging, and safe apply with rollback. | `source`, `storage` | Agent `projects`, Client `transfer` | | `stack.rs` | Recognizes Rust, Node, and Python projects. | none | `artifacts`, `tools` | | `artifacts.rs` | Where each kind of project keeps its build output on the Agent. | `stack` | Agent `projects`, Client `run` | -| `telemetry.rs` | Reads hardware details and live use with `sysinfo` and `nvidia-smi`. | `presentation`, `protocol` | `preflight`, `tools`, Agent `lib`, Client `health`, `transfer`, `watch` | -| `preflight.rs` | Setup checks that print the fix, and the package manager's install command. | `presentation`, `telemetry` | `tools`, Agent `lib`, `jobs`, `service`; Client `link`, `transfer` | +| `telemetry.rs` | Reads hardware details and live use with `sysinfo` and `nvidia-smi`. | `presentation`, `protocol` | `preflight`, `tools`, Agent `lib`, `ports`, Client `health`, `transfer`, `watch` | +| `preflight.rs` | Setup checks that print the fix, and the package manager's install command. | `presentation`, `telemetry` | `tools`, Agent `lib`, `jobs`, `ports`, `service`; Client `link`, `transfer` | | `tools.rs` | The tools Ally can set up on the Agent, how to find them, and their install commands. | `preflight`, `stack`, `telemetry` | `control`, Agent `service`, Client `tools` | | `presentation.rs` | The shared output style: symbols, colors, rows, and sizes. | none | `step`, `preflight`, `telemetry`, every command | | `step.rs` | A spinner for slow work, then a finished line with the time taken. | `presentation` | Agent `lib`; Client `client`, `transfer`, `run`, `link`, `ps`, `unlink` | @@ -301,9 +301,10 @@ ally/ | File | Purpose | Uses | Used By | | --- | --- | --- | --- | | `lib.rs` | `ally start`: setup checks, the start screen, pairing, and starting the other parts. `ally internal-daemon`: the same without pairing, for the boot service. | `awake`, `clients`, `service`, `tunnel` | Client `lib` | -| `service.rs` | The private control socket, the `internal-control` helper that reaches it, and the tools probe run in the login shell. | `clients`, `jobs`, `projects` | `lib`, `runner`, Client `lib` | +| `service.rs` | The private control socket, the `internal-control` helper that reaches it, and the tools probe run in the login shell. | `clients`, `jobs`, `ports`, `projects` | `lib`, `runner`, Client `lib` | | `projects.rs` | Project storage: source copies, sync locks, build output, and environment files. Lists, cleans, and removes copies. | `jobs` | `service`, `jobs`, `runner`, `uninstall` | -| `jobs.rs` | Job records, `tmux` sessions, and safe stopping. | `projects` | `service`, `projects`, `runner` | +| `jobs.rs` | Job records, `tmux` sessions, the processes each job owns, and safe stopping. | `projects` | `service`, `ports`, `projects`, `runner` | +| `ports.rs` | The TCP ports that jobs and Docker containers listen on, found with `ss` on Linux and `lsof` on macOS, for port forwarding. | `jobs` | `service` | | `runner.rs` | `ally internal-run`: runs one command in the project copy with a real terminal. | `jobs`, `projects`, `service` | Client `lib` | | `clients.rs` | The Clients allowed over iroh, and their names for the start screen. | none | `lib`, `service`, `tunnel` | | `tunnel.rs` | The iroh endpoint, which passes paired Clients to the local ssh server. | `clients` | `lib` |