diff --git a/README.md b/README.md index 4030de4..f2d2ca6 100644 --- a/README.md +++ b/README.md @@ -128,6 +128,27 @@ echo "export STAGING_DATABASE_URL='postgresql://...'" >> ~/.zshrc `pgterm list` shows every profile and whether its variable is currently set. +### Databases behind an SSH jump host + +If a database only answers from inside a bastion or private network, tell the +profile how to get there: + +```bash +pgterm add production --env PROD_DATABASE_URL --ssh deploy@bastion.example.com +``` + +`--ssh` takes `[user@]host[:port]` or a bare `~/.ssh/config` alias — your own +ssh resolves it, so `HostName`, `IdentityFile`, `ProxyJump`, the agent and +`known_hosts` all behave exactly as they do when you type `ssh bastion` +yourself. The DSN keeps naming the **real** database host: no local port is +opened, `sslmode=verify-full` still verifies that hostname, and `.pgpass` +still matches on it. + +Health checks hand the spec to pgbot, which tunnels natively. The SQL and +Data tabs ride an `ssh -W` child process in BatchMode — it never prompts, so +authenticate once (`ssh bastion`) or load your key into the agent first; a +refused login shows ssh's own reason in the tab. + ### Adding from inside the UI Press `a` (or click `+ Add database`). Name, then Stage (`←`/`→` cycles @@ -348,6 +369,7 @@ bell = false # ring the terminal bell with a toast name = "production" env = "PROD_DATABASE_URL" stage = "prod" # prod | staging | dev | local — inferred when absent +ssh = "deploy@bastion" # optional: reach it through this SSH jump host ``` When a database you are *not* looking at turns critical or unavailable, a diff --git a/src/app.rs b/src/app.rs index 6ab2faf..97e0fec 100644 --- a/src/app.rs +++ b/src/app.rs @@ -182,6 +182,7 @@ impl DbState { stage: None, pgrun_project: None, writes: false, + ssh: None, }); db.source = ConnSource::Session(url); db @@ -985,7 +986,7 @@ impl App { } }; if let Err(e) = cfg - .add_with_stage(name, env_name, stage) + .add_with_stage(name, env_name, stage, None) .and_then(|()| cfg.save()) { popup.message = Some(Err(SafeError::new( @@ -1031,7 +1032,7 @@ impl App { ConnSource::Session(_) => unreachable!("handled above"), }; if let Err(e) = cfg - .add_with_stage(name, &env_name, stage) + .add_with_stage(name, &env_name, stage, None) .and_then(|()| cfg.save()) { popup.message = Some(Err(SafeError::new( @@ -1049,6 +1050,7 @@ impl App { stage, pgrun_project: None, writes: false, + ssh: None, })); let idx = self.dbs.len() - 1; self.selected = idx; @@ -1752,6 +1754,7 @@ fn parse_export_assignment(s: &str) -> Option<(String, String)> { pub async fn run_effect( pgbot_bin: PathBuf, source: ConnSource, + ssh: Option, db: usize, cmd: PgbotCommand, kind: CmdKind, @@ -1759,7 +1762,7 @@ pub async fn run_effect( ) -> Action { let _permit = sem.acquire_owned().await.ok(); let timeout = runner::default_timeout(&cmd); - let result = runner::run_pgbot(&pgbot_bin, &source, &cmd, timeout) + let result = runner::run_pgbot(&pgbot_bin, &source, ssh.as_deref(), &cmd, timeout) .await .and_then(|out| decode_result(&cmd, &out)); Action::CheckFinished { db, kind, result } @@ -1779,6 +1782,7 @@ impl Connections { &mut self, db: usize, source: &ConnSource, + ssh: Option<&str>, ) -> Result<&mut tokio_postgres::Client, SafeError> { // A closed connection is indistinguishable from a working one until // it is used, so drop it and reconnect rather than fail the query. @@ -1786,7 +1790,7 @@ impl Connections { self.0.remove(&db); } if let std::collections::hash_map::Entry::Vacant(slot) = self.0.entry(db) { - slot.insert(crate::db::connect(source).await?); + slot.insert(crate::db::connect(source, ssh).await?); } Ok(self.0.get_mut(&db).expect("present or just inserted")) } @@ -1801,12 +1805,13 @@ pub async fn run_sql_effect( conns: Arc>, db: usize, source: ConnSource, + ssh: Option, target: SqlTarget, sql: String, policy: WritePolicy, ) -> Action { let mut guard = conns.lock().await; - let result = match guard.get(db, &source).await { + let result = match guard.get(db, &source, ssh.as_deref()).await { Ok(client) => crate::db::run_sql(client, &sql, policy).await.map(Box::new), Err(e) => Err(e), }; @@ -1844,9 +1849,16 @@ pub async fn run_probe( ) -> Action { let _permit = sem.acquire_owned().await.ok(); let cmd = PgbotCommand::Probe; - let result = runner::run_pgbot(&pgbot_bin, &source, &cmd, runner::default_timeout(&cmd)) - .await - .and_then(|out| decode_result(&cmd, &out)); + // The add popup has no ssh field (yet); its probes always dial direct. + let result = runner::run_pgbot( + &pgbot_bin, + &source, + None, + &cmd, + runner::default_timeout(&cmd), + ) + .await + .and_then(|out| decode_result(&cmd, &out)); Action::ProbeFinished { name, source, diff --git a/src/cli.rs b/src/cli.rs index f78ded2..a654c37 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -19,6 +19,9 @@ pub struct AddOptions { pub open: bool, /// The environment badge to save with the profile; None = infer it. pub stage: Option, + /// SSH jump host the database is reached through; validated and saved + /// with the profile. + pub ssh: Option, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -38,7 +41,8 @@ pub enum Invocation { } const USAGE: &str = "usage: pgterm [--interval ] [--no-monitor] - pgterm add [--env ] [--stage prod|staging|dev|local] [--open] + pgterm add [--env ] [--stage prod|staging|dev|local] + [--ssh [user@]host[:port]] [--open] pgterm list pgterm remove pgterm --default-config"; @@ -53,6 +57,7 @@ pub fn parse_args(args: &[String]) -> Invocation { let mut env = None; let mut open = false; let mut stage = None; + let mut ssh = None; while let Some(a) = it.next() { match a.as_str() { "--env" => match it.next() { @@ -60,6 +65,14 @@ pub fn parse_args(args: &[String]) -> Invocation { None => return Invocation::Usage("--env needs a value".into()), }, "--open" => open = true, + "--ssh" => match it.next() { + Some(v) => ssh = Some(v.clone()), + None => { + return Invocation::Usage( + "--ssh needs a jump host ([user@]host[:port])".into(), + ) + } + }, "--stage" => match it.next().and_then(|v| Stage::parse(v)) { Some(st) => stage = Some(st), None => { @@ -81,6 +94,7 @@ pub fn parse_args(args: &[String]) -> Invocation { env, open, stage, + ssh, }), None => Invocation::Usage("add needs a database name".into()), } @@ -188,7 +202,10 @@ pub async fn cmd_add(opts: &AddOptions) -> i32 { return EXIT_FAILED; } }; - if let Err(e) = cfg.clone().add(&opts.name, &env_name) { + if let Err(e) = + cfg.clone() + .add_with_stage(&opts.name, &env_name, opts.stage, opts.ssh.as_deref()) + { eprintln!("pgterm: {e}"); eprintln!("Nothing was saved."); return EXIT_FAILED; @@ -203,10 +220,14 @@ pub async fn cmd_add(opts: &AddOptions) -> i32 { println!("✓ Found {env_name}"); } - println!("Testing {}...\n", opts.name); + match &opts.ssh { + Some(spec) => println!("Testing {} (via ssh {spec})...\n", opts.name), + None => println!("Testing {}...\n", opts.name), + } let probe = runner::run_pgbot( &runner::pgbot_bin(), &ConnSource::Env(env_name.clone()), + opts.ssh.as_deref(), &PgbotCommand::Probe, runner::default_timeout(&PgbotCommand::Probe), ) @@ -250,7 +271,7 @@ pub async fn cmd_add(opts: &AddOptions) -> i32 { } } - if let Err(e) = cfg.add_with_stage(&opts.name, &env_name, opts.stage) { + if let Err(e) = cfg.add_with_stage(&opts.name, &env_name, opts.stage, opts.ssh.as_deref()) { eprintln!("pgterm: {e}\nNothing was saved."); return EXIT_FAILED; } @@ -361,7 +382,8 @@ mod tests { name: "prod".into(), env: None, open: false, - stage: None + stage: None, + ssh: None }) ); assert_eq!( @@ -370,7 +392,8 @@ mod tests { name: "prod".into(), env: Some("PROD_URL".into()), open: true, - stage: None + stage: None, + ssh: None }) ); assert!(matches!(parse_args(&s(&["add"])), Invocation::Usage(_))); @@ -455,6 +478,25 @@ mod tests { )); } + #[test] + fn add_takes_an_ssh_jump_host() { + match parse_args(&s(&[ + "add", + "prod", + "--env", + "P", + "--ssh", + "deploy@bastion", + ])) { + Invocation::Add(o) => assert_eq!(o.ssh.as_deref(), Some("deploy@bastion")), + other => panic!("{other:?}"), + } + match parse_args(&s(&["add", "p", "--ssh"])) { + Invocation::Usage(msg) => assert!(msg.contains("jump host"), "{msg}"), + other => panic!("{other:?}"), + } + } + #[test] fn default_config_flag_is_its_own_invocation() { assert_eq!( diff --git a/src/config.rs b/src/config.rs index 44e26a2..61ceaa1 100644 --- a/src/config.rs +++ b/src/config.rs @@ -132,6 +132,8 @@ pointer = true # stage = \"prod\" # prod | staging | dev | local \u{2014} badge; inferred from the name when absent # pgrun_project = \"acme-api\" # show this pgrun project's branches for the database # writes = false # true lets the SQL tab write (a PROD badge still asks first) +# ssh = \"user@bastion\" # reach the database through this SSH jump host +# # ([user@]host[:port] or a ~/.ssh/config alias) "; /// One monitored database: a friendly name and the environment variable that @@ -152,6 +154,11 @@ pub struct DatabaseProfile { /// READ ONLY transaction until this is set. #[serde(default, skip_serializing_if = "is_false")] pub writes: bool, + /// SSH jump host to reach this database through: `[user@]host[:port]`, or + /// a bare `~/.ssh/config` alias. Resolved by the user's own ssh — pgterm + /// stores the spec, never keys or passphrases. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ssh: Option, } fn is_false(b: &bool) -> bool { @@ -257,17 +264,27 @@ impl TerminalConfig { /// Validates and appends a profile. Names are what tabs display: short, /// shell-friendly, unique. pub fn add(&mut self, name: &str, env: &str) -> anyhow::Result<()> { - self.add_with_stage(name, env, None) + self.add_with_stage(name, env, None, None) } - /// `add`, with the environment badge the caller chose (None = infer). + /// `add`, with the environment badge the caller chose (None = infer) and + /// the SSH jump host, if the database is only reachable through one. pub fn add_with_stage( &mut self, name: &str, env: &str, stage: Option, + ssh: Option<&str>, ) -> anyhow::Result<()> { validate_name(name)?; + let ssh = match ssh { + None => None, + Some(spec) => match crate::ssh::Spec::parse(spec) { + // The original spelling is kept — it is what ssh will resolve. + Ok(_) => Some(spec.trim().to_string()), + Err(e) => bail!("--ssh: {e}"), + }, + }; if env.is_empty() { bail!("environment variable name is empty"); } @@ -290,6 +307,7 @@ impl TerminalConfig { stage, pgrun_project: None, writes: false, + ssh, }); Ok(()) } @@ -533,6 +551,7 @@ env = "STAGING_DATABASE_URL" stage: Some(Stage::Dev), pgrun_project: None, writes: false, + ssh: None, }; assert_eq!(p.badge(), Some(Stage::Dev), "explicit stage beats the name"); let p = DatabaseProfile { @@ -541,6 +560,7 @@ env = "STAGING_DATABASE_URL" stage: None, pgrun_project: None, writes: false, + ssh: None, }; assert_eq!(p.badge(), Some(Stage::Prod)); } @@ -548,7 +568,7 @@ env = "STAGING_DATABASE_URL" #[test] fn stage_and_ui_round_trip_and_old_files_still_load() { let mut cfg = TerminalConfig::default(); - cfg.add_with_stage("prod", "PROD_URL", Some(Stage::Prod)) + cfg.add_with_stage("prod", "PROD_URL", Some(Stage::Prod), None) .unwrap(); cfg.add("analytics", "AN_URL").unwrap(); cfg.ui.bell = true; @@ -571,6 +591,30 @@ env = "STAGING_DATABASE_URL" ); } + #[test] + fn ssh_round_trips_is_validated_and_old_files_load_without_it() { + let mut cfg = TerminalConfig::default(); + cfg.add_with_stage("prod", "P_URL", None, Some("deploy@bastion:2222")) + .unwrap(); + cfg.add("plain", "X_URL").unwrap(); + let text = toml::to_string_pretty(&cfg).unwrap(); + assert!(text.contains("ssh = \"deploy@bastion:2222\""), "{text}"); + let back: TerminalConfig = toml::from_str(&text).unwrap(); + assert_eq!(back, cfg); + assert_eq!(back.databases[1].ssh, None, "absent must stay absent"); + + // A spec that could read as an ssh option is refused at add time. + let err = cfg + .add_with_stage("evil", "E_URL", None, Some("-oProxyCommand=x")) + .unwrap_err() + .to_string(); + assert!(err.contains("--ssh"), "{err}"); + + let old = "version = 1\n[[databases]]\nname = \"p\"\nenv = \"P_URL\"\n"; + let cfg: TerminalConfig = toml::from_str(old).unwrap(); + assert_eq!(cfg.databases[0].ssh, None); + } + #[test] fn unknown_stage_is_an_error_naming_the_four() { let bad = diff --git a/src/db.rs b/src/db.rs index 09e0091..8a9a233 100644 --- a/src/db.rs +++ b/src/db.rs @@ -135,26 +135,21 @@ fn tls_config() -> Result { .with_no_client_auth()) } -/// Opens a connection. TLS follows the DSN's own `sslmode`, which -/// tokio-postgres parses — pgterm does not weaken it. -pub async fn connect(source: &ConnSource) -> Result { +/// Opens a connection, through the profile's SSH jump host when it has one. +/// TLS follows the DSN's own `sslmode`, which tokio-postgres parses — pgterm +/// does not weaken it, tunneled or not. +pub async fn connect(source: &ConnSource, ssh: Option<&str>) -> Result { let dsn = source.resolve()?; - // tokio-postgres' own Display is terse ("error connecting to server"); - // the reason lives in the source chain, and the reason is the useful part. - let fail = |e: tokio_postgres::Error| { - let mut msg = e.to_string(); - let mut src = std::error::Error::source(&e); - while let Some(cause) = src { - msg.push_str(&format!(": {cause}")); - src = cause.source(); - } - SafeError::new(ErrorKind::ConnectionFailed, &msg, Some(&dsn)) - }; + let fail = |e: tokio_postgres::Error| chain_error(e, &dsn); let config: tokio_postgres::Config = dsn.parse().map_err(|e: tokio_postgres::Error| { SafeError::new(ErrorKind::Usage, &e.to_string(), Some(&dsn)) })?; + if let Some(spec) = ssh { + return connect_ssh(&config, &dsn, spec).await; + } + let connect_tls = async { let tls = tokio_postgres_rustls::MakeRustlsConnect::new(tls_config()?); config.connect(tls).await.map_err(fail) @@ -184,10 +179,7 @@ pub async fn connect(source: &ConnSource) -> Result { Err(tls_err) => { // A server with TLS off refuses the handshake; fall back to plain // only when the DSN did not demand TLS. - let demanded = dsn.contains("sslmode=require") - || dsn.contains("sslmode=verify-ca") - || dsn.contains("sslmode=verify-full"); - if demanded { + if dsn_demands_tls(&dsn) { return Err(tls_err); } match tokio::time::timeout(CONNECT_TIMEOUT, config.connect(NoTls)).await { @@ -213,6 +205,139 @@ fn timeout_error() -> SafeError { ) } +/// tokio-postgres' own Display is terse ("error connecting to server"); +/// the reason lives in the source chain, and the reason is the useful part. +fn chain_error(e: tokio_postgres::Error, dsn: &str) -> SafeError { + let mut msg = e.to_string(); + let mut src = std::error::Error::source(&e); + while let Some(cause) = src { + msg.push_str(&format!(": {cause}")); + src = cause.source(); + } + SafeError::new(ErrorKind::ConnectionFailed, &msg, Some(dsn)) +} + +fn dsn_demands_tls(dsn: &str) -> bool { + dsn.contains("sslmode=require") + || dsn.contains("sslmode=verify-ca") + || dsn.contains("sslmode=verify-full") +} + +/// `connect`, through an SSH jump host: the TCP leg is an `ssh -W` child, and +/// the Postgres startup (TLS negotiation included) runs over its stdio via +/// `connect_raw`. The DSN's hostname is what TLS verifies — the tunnel never +/// rewrites it to a loopback address. +async fn connect_ssh( + config: &tokio_postgres::Config, + dsn: &str, + spec: &str, +) -> Result { + use tokio_postgres::config::Host; + use tokio_postgres::tls::MakeTlsConnect; + + let spec = crate::ssh::Spec::parse(spec) + .map_err(|e| SafeError::new(ErrorKind::Usage, &e, Some(dsn)))?; + let socket_refused = || { + SafeError::new( + ErrorKind::Usage, + "a unix-socket DSN cannot go through an SSH tunnel — name the host and port the jump host can reach", + Some(dsn), + ) + }; + let host = match config.get_hosts().first() { + // On non-unix targets a socket path parses as a Tcp "host"; it can + // never be tunneled, so refuse it the same way everywhere. + Some(Host::Tcp(h)) if h.starts_with('/') => return Err(socket_refused()), + Some(Host::Tcp(h)) => h.clone(), + #[cfg(unix)] + Some(Host::Unix(_)) => return Err(socket_refused()), + None => { + return Err(SafeError::new( + ErrorKind::Usage, + "the DSN names no host to tunnel to", + Some(dsn), + )) + } + }; + let port = config.get_ports().first().copied().unwrap_or(5432); + + // A failed login or refused forward surfaces as EOF on the stream; ssh's + // stderr has the actual reason, so it is folded into the error. + let fail = |e: tokio_postgres::Error, stderr: &std::sync::Arc>| { + let mut err = chain_error(e, dsn); + if let Ok(said) = stderr.lock() { + let said = said.trim(); + if !said.is_empty() { + err = SafeError::new( + err.kind, + &format!("{} — ssh: {said}", err.message), + Some(dsn), + ); + } + } + err + }; + + let connect_tls = async { + let stream = crate::ssh::open(&spec, &host, port)?; + let stderr = stream.stderr_handle(); + let mut mk = tokio_postgres_rustls::MakeRustlsConnect::new(tls_config()?); + let tls = >::make_tls_connect(&mut mk, &host) + .map_err(|e| SafeError::new(ErrorKind::ConnectionFailed, &e.to_string(), Some(dsn)))?; + match config.connect_raw(stream, tls).await { + Ok(pair) => Ok(pair), + Err(e) => { + // Give the stderr reader a beat to collect ssh's last words. + tokio::time::sleep(Duration::from_millis(150)).await; + Err(fail(e, &stderr)) + } + } + }; + let tls_attempt = match tokio::time::timeout(CONNECT_TIMEOUT, connect_tls).await { + Ok(r) => r, + Err(_) => Err(timeout_error()), + }; + match tls_attempt { + Ok((client, connection)) => { + tokio::spawn(async move { + let _ = connection.await; + }); + Ok(client) + } + Err(tls_err) => { + // Same fallback contract as the direct path: plain only when the + // DSN did not demand TLS — over a fresh tunnel, the first ssh died + // with its stream. + if dsn_demands_tls(dsn) { + return Err(tls_err); + } + let connect_plain = async { + let stream = crate::ssh::open(&spec, &host, port)?; + let stderr = stream.stderr_handle(); + match config.connect_raw(stream, NoTls).await { + Ok(pair) => Ok(pair), + Err(e) => { + tokio::time::sleep(Duration::from_millis(150)).await; + Err(fail(e, &stderr)) + } + } + }; + match tokio::time::timeout(CONNECT_TIMEOUT, connect_plain).await { + Ok(Ok((client, connection))) => { + tokio::spawn(async move { + let _ = connection.await; + }); + Ok(client) + } + Ok(Err(e)) => Err(e), + Err(_) => Err(timeout_error()), + } + } + } +} + /// Runs one buffer inside a bounded transaction and renders the result. pub async fn run_sql( client: &mut Client, @@ -520,8 +645,65 @@ mod tests { assert!(strip_sql_noise("SELECT x FROM t").contains("FROM t")); } + /// Both halves mutate PGTERM_SSH_BIN, and lib tests share the process + /// environment — one test, sequential, so they cannot race each other. + #[cfg(unix)] + #[test] + fn ssh_tunnel_failures_say_what_actually_went_wrong() { + use std::io::Write as _; + use std::os::unix::fs::PermissionsExt; + + let rt = tokio::runtime::Runtime::new().unwrap(); + let dsn = "postgres://u:pw@db.internal:5432/app?sslmode=disable"; + + // No ssh at all: the error names the requirement and the override. + std::env::set_var("PGTERM_SSH_BIN", "/nonexistent/pgterm-test-ssh"); + let err = + rt.block_on(async { connect(&ConnSource::Session(dsn.into()), Some("bastion")).await }); + let err = err.expect_err("no ssh, no tunnel"); + assert!(err.to_string().contains("OpenSSH"), "{err}"); + + // A stand-in ssh that refuses the way a real one does: reason on + // stderr, nothing on stdout. The error the SQL tab shows must carry + // that reason, not just "unexpected EOF". + let dir = std::env::temp_dir().join(format!("pgterm-ssh-test-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let fake = dir.join("ssh"); + let mut f = std::fs::File::create(&fake).unwrap(); + f.write_all(b"#!/bin/sh\necho 'Permission denied (publickey).' >&2\nexit 255\n") + .unwrap(); + drop(f); + std::fs::set_permissions(&fake, std::fs::Permissions::from_mode(0o755)).unwrap(); + std::env::set_var("PGTERM_SSH_BIN", &fake); + + let err = rt.block_on(async { + connect(&ConnSource::Session(dsn.into()), Some("deploy@bastion")).await + }); + std::env::remove_var("PGTERM_SSH_BIN"); + let _ = std::fs::remove_dir_all(&dir); + let err = err.expect_err("a refused ssh login cannot connect"); + assert!( + err.to_string().contains("Permission denied"), + "ssh's reason is missing: {err}" + ); + assert!(!err.to_string().contains(":pw@"), "password leaked: {err}"); + } + + #[test] + fn ssh_tunnel_refuses_a_unix_socket_dsn() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let err = rt.block_on(async { + let source = + ConnSource::Session("postgres://u:pw@%2Fvar%2Frun%2Fpostgresql/app".into()); + connect(&source, Some("bastion")).await + }); + let err = err.expect_err("a socket path cannot be tunneled"); + assert!(err.to_string().contains("unix-socket"), "{err}"); + } + /// Against a real database: /// `PGTERM_TEST_DATABASE_URL=postgres://... cargo test --lib live_ -- --ignored --nocapture` + /// Add `PGTERM_TEST_SSH_TUNNEL=[user@]host[:port]` to run the tunnel test. fn live_source() -> Option { std::env::var("PGTERM_TEST_DATABASE_URL") .ok() @@ -529,6 +711,26 @@ mod tests { .map(ConnSource::Session) } + #[test] + #[ignore] + fn live_ssh_tunnel_runs_a_query() { + let (Some(source), Ok(spec)) = (live_source(), std::env::var("PGTERM_TEST_SSH_TUNNEL")) + else { + println!("set PGTERM_TEST_DATABASE_URL and PGTERM_TEST_SSH_TUNNEL to run this"); + return; + }; + let rt = tokio::runtime::Runtime::new().unwrap(); + rt.block_on(async { + let mut c = connect(&source, Some(&spec)) + .await + .expect("tunneled connect"); + let r = run_sql(&mut c, "SELECT 1 AS one", WritePolicy::ReadOnly) + .await + .expect("select over the tunnel"); + assert_eq!(r.rows[0], vec!["1"]); + }); + } + #[test] #[ignore] fn live_read_only_really_is_read_only() { @@ -538,7 +740,7 @@ mod tests { }; let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async { - let mut c = connect(&source).await.expect("connect"); + let mut c = connect(&source, None).await.expect("connect"); let r = run_sql( &mut c, diff --git a/src/lib.rs b/src/lib.rs index d21a833..2d1c52e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,4 +19,5 @@ pub mod pointer; pub mod runner; pub mod sanitize; pub mod screens; +pub mod ssh; pub mod ui; diff --git a/src/main.rs b/src/main.rs index 6451bf3..909d10e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -174,11 +174,12 @@ fn perform( continue; }; let source = state.source.clone(); + let ssh = state.profile.ssh.clone(); let bin = app.pgbot_bin.clone(); let tx = tx.clone(); let sem = sem.clone(); tokio::spawn(async move { - let _ = tx.send(app::run_effect(bin, source, db, cmd, kind, sem).await); + let _ = tx.send(app::run_effect(bin, source, ssh, db, cmd, kind, sem).await); }); } Effect::SpawnSql { @@ -191,11 +192,13 @@ fn perform( continue; }; let source = state.source.clone(); + let ssh = state.profile.ssh.clone(); let tx = tx.clone(); let conns = conns.clone(); tokio::spawn(async move { - let _ = - tx.send(app::run_sql_effect(conns, db, source, target, sql, policy).await); + let _ = tx.send( + app::run_sql_effect(conns, db, source, ssh, target, sql, policy).await, + ); }); } Effect::SpawnPgrun { db, cmd, open } => { diff --git a/src/runner.rs b/src/runner.rs index 519abe6..3305975 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -134,6 +134,7 @@ pub struct RunOutcome { pub async fn run_pgbot( pgbot_bin: &Path, source: &ConnSource, + ssh: Option<&str>, cmd: &PgbotCommand, timeout: Duration, ) -> Result { @@ -149,6 +150,17 @@ pub async fn run_pgbot( .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true); + // pgbot tunnels natively. The profile is the only source of the spec: an + // ambient PGBOT_SSH_TUNNEL would silently reroute EVERY database, so it is + // stripped when the profile has none — same rule as PGBOT_DATABASE_URL. + match ssh { + Some(spec) => { + c.env("PGBOT_SSH_TUNNEL", spec); + } + None => { + c.env_remove("PGBOT_SSH_TUNNEL"); + } + } let child = match c.spawn() { Ok(ch) => ch, diff --git a/src/screens/branches.rs b/src/screens/branches.rs index 27828f6..4c1d4ce 100644 --- a/src/screens/branches.rs +++ b/src/screens/branches.rs @@ -166,6 +166,7 @@ mod tests { stage: None, pgrun_project: project.map(String::from), writes: false, + ssh: None, }); db.branches = branches; db diff --git a/src/screens/data.rs b/src/screens/data.rs index 9ba74fe..a3bc145 100644 --- a/src/screens/data.rs +++ b/src/screens/data.rs @@ -140,6 +140,7 @@ mod tests { stage: None, pgrun_project: None, writes: false, + ssh: None, }) } diff --git a/src/screens/sql.rs b/src/screens/sql.rs index 0d40a09..b10868e 100644 --- a/src/screens/sql.rs +++ b/src/screens/sql.rs @@ -347,6 +347,7 @@ mod tests { stage: None, pgrun_project: None, writes: false, + ssh: None, }); let render_hint = |db: &crate::app::DbState| { let mut term = diff --git a/src/ssh.rs b/src/ssh.rs new file mode 100644 index 0000000..29b0630 --- /dev/null +++ b/src/ssh.rs @@ -0,0 +1,290 @@ +//! Reaching a database through an SSH jump host. +//! +//! pgterm does not speak SSH itself — it runs the user's own `ssh` with `-W`, +//! which pipes the database connection over the child's stdin/stdout. That is +//! not an `ssh -L` port forward: no local port is opened, and the DSN keeps +//! naming the REAL database host all the way through, so `sslmode=verify-full` +//! still validates against that hostname and `.pgpass` still matches on it. +//! +//! Delegating to the binary is deliberate. `~/.ssh/config` (HostName, User, +//! Port, IdentityFile, ProxyJump, ControlMaster), the agent, hardware keys and +//! known_hosts all behave exactly the way the user's own `ssh` already does +//! for that host — none of it re-implemented, none of it subtly different. +//! BatchMode keeps a TUI-owned terminal safe: ssh fails with its reason on +//! stderr instead of prompting into a screen that cannot answer. +//! +//! The health checks never come through here: pgbot has native `--ssh-tunnel` +//! support, so the runner hands it the spec via PGBOT_SSH_TUNNEL instead. + +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; + +use tokio::io::{AsyncBufReadExt as _, AsyncRead, AsyncWrite, ReadBuf}; +use tokio::process::{Child, ChildStdin, ChildStdout}; + +use crate::sanitize::{ErrorKind, SafeError}; + +/// A validated `[user@]host[:port]` jump-host spec. A bare host may be an +/// ssh_config alias; ssh resolves it, not us. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Spec { + pub user: Option, + pub host: String, + pub port: Option, +} + +impl Spec { + /// Parses and validates. The parts land in argv, so anything that could + /// read as an ssh option or smuggle in whitespace is refused outright. + pub fn parse(spec: &str) -> Result { + let raw = spec.trim(); + if raw.is_empty() { + return Err("ssh spec is empty — want [user@]host[:port]".into()); + } + if raw.contains("://") { + return Err("ssh spec is [user@]host[:port], not a URL".into()); + } + let (user, rest) = match raw.rsplit_once('@') { + Some((u, r)) => (Some(u.to_string()), r), + None => (None, raw), + }; + // Bracketed IPv6 keeps its colons; a single colon splits off a port; a + // bare IPv6 address has no unambiguous port syntax, so it is left alone. + let (host, port) = if let Some(inner) = rest.strip_prefix('[') { + match inner.split_once(']') { + Some((h, "")) => (h.to_string(), None), + Some((h, tail)) => match tail.strip_prefix(':') { + Some(p) => (h.to_string(), Some(p)), + None => return Err(format!("malformed ssh spec {raw:?}")), + }, + None => return Err(format!("malformed ssh spec {raw:?}")), + } + } else if rest.matches(':').count() == 1 { + let (h, p) = rest.split_once(':').expect("counted one"); + (h.to_string(), Some(p)) + } else { + (rest.to_string(), None) + }; + let port = match port { + None => None, + Some(p) => Some( + p.parse::() + .ok() + .filter(|p| *p > 0) + .ok_or_else(|| format!("{p:?} is not a port number"))?, + ), + }; + for (what, s) in [("user", user.as_deref().unwrap_or("x")), ("host", &host)] { + if s.is_empty() { + return Err(format!("ssh spec has an empty {what}")); + } + if s.starts_with('-') { + return Err(format!("ssh {what} may not start with '-'")); + } + if !s + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_' | ':' | '%')) + { + return Err(format!( + "ssh {what} may only contain letters, digits, '.', '-', '_' and ':'" + )); + } + } + Ok(Spec { user, host, port }) + } +} + +/// $PGTERM_SSH_BIN override → `ssh` on PATH. The same escape hatch pgbot and +/// pgrun binaries get. +fn ssh_bin() -> std::path::PathBuf { + match std::env::var("PGTERM_SSH_BIN") { + Ok(p) if !p.is_empty() => std::path::PathBuf::from(p), + _ => std::path::PathBuf::from("ssh"), + } +} + +/// The database connection, riding an `ssh -W` child's stdio. Dropping the +/// stream kills the child (kill_on_drop), so a closed SQL tab leaves no ssh +/// behind. +pub struct SshStream { + stdin: ChildStdin, + stdout: ChildStdout, + stderr: Arc>, + _child: Child, +} + +impl SshStream { + /// Whatever ssh has said so far — "Permission denied (publickey)", "Host + /// key verification failed" — for the error path. Empty means silence. + pub fn stderr_handle(&self) -> Arc> { + self.stderr.clone() + } +} + +/// Opens the tunnel: `ssh -W db_host:db_port [-l user] [-p port] -- host`. +/// The child is spawned, not awaited — a refused login surfaces as EOF on the +/// stream, with the reason in `stderr_handle`. +pub fn open(spec: &Spec, db_host: &str, db_port: u16) -> Result { + let bin = ssh_bin(); + let mut c = tokio::process::Command::new(&bin); + // IPv6 database hosts are bracketed for -W, as ssh expects. + let target = if db_host.contains(':') { + format!("[{db_host}]:{db_port}") + } else { + format!("{db_host}:{db_port}") + }; + c.arg("-W").arg(target); + // BatchMode: fail with the reason rather than prompt into the TUI. + // -W already implies -N, -T, ExitOnForwardFailure and ClearAllForwardings. + for opt in [ + "BatchMode=yes", + "ConnectTimeout=10", + "ServerAliveInterval=30", + "ServerAliveCountMax=3", + ] { + c.arg("-o").arg(opt); + } + if let Some(user) = &spec.user { + c.arg("-l").arg(user); + } + if let Some(port) = spec.port { + c.arg("-p").arg(port.to_string()); + } + c.arg("--").arg(&spec.host); + c.stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .kill_on_drop(true); + + let mut child = c.spawn().map_err(|e| { + SafeError::new( + ErrorKind::ConnectionFailed, + &format!( + "cannot run {} for the ssh tunnel: {e} — an OpenSSH client is required (or set PGTERM_SSH_BIN)", + bin.display() + ), + None, + ) + })?; + let stdin = child.stdin.take().expect("piped"); + let stdout = child.stdout.take().expect("piped"); + let err_pipe = child.stderr.take().expect("piped"); + let stderr = Arc::new(Mutex::new(String::new())); + let sink = stderr.clone(); + // Line by line, not read_to_string: the reason must be there when the + // stream fails, not only once ssh has exited and closed the pipe. + tokio::spawn(async move { + let mut lines = tokio::io::BufReader::new(err_pipe).lines(); + while let Ok(Some(line)) = lines.next_line().await { + if let Ok(mut s) = sink.lock() { + if !s.is_empty() { + s.push('\n'); + } + s.push_str(&line); + } + } + }); + Ok(SshStream { + stdin, + stdout, + stderr, + _child: child, + }) +} + +impl AsyncRead for SshStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.stdout).poll_read(cx, buf) + } +} + +impl AsyncWrite for SshStream { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.stdin).poll_write(cx, buf) + } + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.stdin).poll_flush(cx) + } + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.stdin).poll_shutdown(cx) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn specs_parse_the_ssh_shapes() { + assert_eq!( + Spec::parse("bastion").unwrap(), + Spec { + user: None, + host: "bastion".into(), + port: None + } + ); + assert_eq!( + Spec::parse("deploy@bastion.example.com:2222").unwrap(), + Spec { + user: Some("deploy".into()), + host: "bastion.example.com".into(), + port: Some(2222) + } + ); + assert_eq!( + Spec::parse("[::1]:2222").unwrap(), + Spec { + user: None, + host: "::1".into(), + port: Some(2222) + } + ); + // Bare IPv6: the colons are the address, not a port. + assert_eq!(Spec::parse("fe80::1").unwrap().port, None); + // Whitespace around the spec is tolerated, inside it is not. + assert!(Spec::parse(" bastion ").is_ok()); + } + + #[test] + fn hostile_specs_are_refused() { + // Anything that could become an ssh option or extra argv. + for bad in [ + "", + "-oProxyCommand=evil", + "-J other", + "user@-host", + "host extra", + "host\targ", + "ssh://host", + "postgres://u:p@h/db", + "host:notaport", + "host:0", + "host:99999", + "@host", + "user@", + "[::1", + "$(whoami)@host", + "host;rm", + ] { + assert!(Spec::parse(bad).is_err(), "accepted {bad:?}"); + } + } + + #[test] + fn user_and_port_survive_an_at_sign_in_passwords_never_seen_here() { + // rsplit on '@': the LAST @ separates user from host, like ssh. + let s = Spec::parse("we.ird-user@host").unwrap(); + assert_eq!(s.user.as_deref(), Some("we.ird-user")); + assert_eq!(s.host, "host"); + } +} diff --git a/src/ui.rs b/src/ui.rs index bb97886..96d95ef 100644 --- a/src/ui.rs +++ b/src/ui.rs @@ -993,10 +993,12 @@ mod tests { } = e { let source = app.dbs[db].source.clone(); + let ssh = app.dbs[db].profile.ssh.clone(); let action = rt.block_on(crate::app::run_sql_effect( conns.clone(), db, source, + ssh, target, sql, policy, @@ -1019,10 +1021,12 @@ mod tests { } = e { let source = app.dbs[db].source.clone(); + let ssh = app.dbs[db].profile.ssh.clone(); let action = rt.block_on(crate::app::run_sql_effect( conns.clone(), db, source, + ssh, target, sql, policy, @@ -1058,10 +1062,12 @@ mod tests { } = e { let source = app.dbs[db].source.clone(); + let ssh = app.dbs[db].profile.ssh.clone(); let action = rt.block_on(crate::app::run_sql_effect( conns.clone(), db, source, + ssh, target, sql, policy, diff --git a/tests/bin/fake_pgbot.rs b/tests/bin/fake_pgbot.rs index 6cdbe7e..8907cd4 100644 --- a/tests/bin/fake_pgbot.rs +++ b/tests/bin/fake_pgbot.rs @@ -76,6 +76,12 @@ fn main() { &dir.join("invocations.log"), &format!("{} url={dsn}", args.join(" ")), ); + // Its own file, not another key on the line above: the url= parse in the + // tests must keep seeing the DSN as the tail. + append_line( + &dir.join("ssh.log"), + &std::env::var("PGBOT_SSH_TUNNEL").unwrap_or_else(|_| "-".into()), + ); let _ = std::fs::write(dir.join(format!("running.{}", std::process::id())), b""); append_line(&dir.join("peaks.log"), &live_markers(&dir).to_string()); diff --git a/tests/monitor.rs b/tests/monitor.rs index d45264f..265cec5 100644 --- a/tests/monitor.rs +++ b/tests/monitor.rs @@ -32,9 +32,11 @@ async fn sweep(app: &mut App, bin: &std::path::Path, permits: usize) { for e in effects { if let Effect::Spawn { db, cmd, kind } = e { let source = app.dbs[db].source.clone(); + let ssh = app.dbs[db].profile.ssh.clone(); joins.push(tokio::spawn(run_effect( bin.to_path_buf(), source, + ssh, db, cmd, kind, diff --git a/tests/runner_integration.rs b/tests/runner_integration.rs index 917a6aa..18c5fef 100644 --- a/tests/runner_integration.rs +++ b/tests/runner_integration.rs @@ -20,6 +20,7 @@ async fn healthy_run_returns_json_stdout() { let out = run_pgbot( &bin, &ConnSource::Env("IT_HEALTHY_URL".into()), + None, &PgbotCommand::Monitor, Duration::from_secs(10), ) @@ -41,6 +42,7 @@ async fn warn_exit_one_still_carries_json() { let out = run_pgbot( &bin, &ConnSource::Env("IT_WARN_URL".into()), + None, &PgbotCommand::Monitor, Duration::from_secs(10), ) @@ -61,6 +63,7 @@ async fn refused_connection_is_sanitized() { let err = run_pgbot( &bin, &ConnSource::Env("IT_REFUSE_URL".into()), + None, &PgbotCommand::Monitor, Duration::from_secs(10), ) @@ -88,6 +91,7 @@ async fn hang_is_killed_at_the_deadline() { let err = run_pgbot( &bin, &ConnSource::Env("IT_HANG_URL".into()), + None, &PgbotCommand::Monitor, Duration::from_millis(300), ) @@ -111,6 +115,7 @@ async fn missing_env_never_spawns_pgbot() { let err = run_pgbot( &bin, &ConnSource::Env("IT_DOES_NOT_EXIST".into()), + None, &PgbotCommand::Monitor, Duration::from_secs(5), ) @@ -131,6 +136,7 @@ async fn missing_binary_reports_pgbot_missing() { let err = run_pgbot( std::path::Path::new("/nonexistent/pgbot"), &ConnSource::Env("IT_BIN_URL".into()), + None, &PgbotCommand::Monitor, Duration::from_secs(5), ) @@ -149,6 +155,7 @@ async fn indexes_and_why_reach_their_own_reports() { let idx = run_pgbot( &bin, &ConnSource::Env("IT_REPORTS_URL".into()), + None, &PgbotCommand::Indexes, Duration::from_secs(10), ) @@ -160,6 +167,7 @@ async fn indexes_and_why_reach_their_own_reports() { let why = run_pgbot( &bin, &ConnSource::Env("IT_REPORTS_URL".into()), + None, &PgbotCommand::Why, Duration::from_secs(10), ) @@ -179,6 +187,7 @@ async fn dsn_travels_by_env_not_argv() { run_pgbot( &bin, &ConnSource::Env("IT_ENVONLY_URL".into()), + None, &PgbotCommand::Monitor, Duration::from_secs(10), ) @@ -194,3 +203,50 @@ async fn dsn_travels_by_env_not_argv() { "child env missing DATABASE_URL: {env_url}" ); } + +#[tokio::test] +async fn profile_ssh_reaches_pgbot_as_its_native_tunnel_env() { + let _env = common::env_lock(); + let dir = common::TempDir::new("run-ssh"); + let bin = common::write_fake_pgbot(dir.path()); + std::env::set_var("IT_SSH_URL", common::dsn("healthy")); + + run_pgbot( + &bin, + &ConnSource::Env("IT_SSH_URL".into()), + Some("deploy@bastion.internal:2222"), + &PgbotCommand::Monitor, + Duration::from_secs(10), + ) + .await + .unwrap(); + let log = std::fs::read_to_string(dir.path().join("ssh.log")).unwrap(); + assert_eq!(log.trim(), "deploy@bastion.internal:2222"); + // Never in argv either — the spec travels as env, like the DSN. + let argv = std::fs::read_to_string(dir.path().join("invocations.log")).unwrap(); + assert!(!argv.contains("bastion"), "spec leaked into argv: {argv}"); +} + +#[tokio::test] +async fn ambient_tunnel_env_is_stripped_when_the_profile_has_none() { + let _env = common::env_lock(); + let dir = common::TempDir::new("run-ssh-strip"); + let bin = common::write_fake_pgbot(dir.path()); + std::env::set_var("IT_SSH_STRIP_URL", common::dsn("healthy")); + // An exported PGBOT_SSH_TUNNEL would reroute EVERY database; only the + // profile may say so. + std::env::set_var("PGBOT_SSH_TUNNEL", "ambient@leak"); + + run_pgbot( + &bin, + &ConnSource::Env("IT_SSH_STRIP_URL".into()), + None, + &PgbotCommand::Monitor, + Duration::from_secs(10), + ) + .await + .unwrap(); + std::env::remove_var("PGBOT_SSH_TUNNEL"); + let log = std::fs::read_to_string(dir.path().join("ssh.log")).unwrap(); + assert_eq!(log.trim(), "-", "the ambient spec reached the child"); +}