Skip to main content

agent_first_psql/
container_transport.rs

1use crate::conn::{
2    PostgresEndpoint, make_supported_tls, postgres_tls_server_name, resolve_pg_config,
3    resolve_single_postgres_endpoint,
4};
5use crate::db::ConnectError;
6use crate::types::{ContainerConfig, ContainerDriver, SessionConfig};
7use std::pin::Pin;
8use std::process::Stdio;
9use std::sync::atomic::{AtomicU64, Ordering};
10use std::sync::{Arc, Mutex};
11use std::task::{Context, Poll};
12use std::time::{Duration, SystemTime, UNIX_EPOCH};
13use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, ReadBuf};
14use tokio::process::{ChildStderr, ChildStdin, ChildStdout};
15use tokio::sync::oneshot;
16use tokio_postgres::Client;
17use tokio_postgres::tls::MakeTlsConnect;
18
19const STDERR_CAPTURE_LIMIT: usize = 8 * 1024;
20const STDERR_HINT_BYTES: usize = 512;
21const BRIDGE_READY_PREFIX: &str = "AFPSQL_BRIDGE_OK";
22const BRIDGE_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(2);
23
24static BRIDGE_NONCE_COUNTER: AtomicU64 = AtomicU64::new(0);
25
26fn generate_bridge_nonce() -> String {
27    let nanos = SystemTime::now()
28        .duration_since(UNIX_EPOCH)
29        .map(|d| d.as_nanos() as u64)
30        .unwrap_or(0);
31    let counter = BRIDGE_NONCE_COUNTER.fetch_add(1, Ordering::Relaxed);
32    let mixed = nanos
33        .wrapping_mul(0x9E37_79B9_7F4A_7C15)
34        .wrapping_add(counter);
35    format!("{mixed:016x}")
36}
37
38const PYTHON_BRIDGE: &str = r#"import os,select,socket,sys
39mode=sys.argv[1]
40if mode=="tcp":
41    s=socket.create_connection((sys.argv[2], int(sys.argv[3])))
42elif mode=="unix":
43    s=socket.socket(socket.AF_UNIX)
44    s.connect(sys.argv[2])
45else:
46    sys.stderr.write("unsupported bridge mode: "+mode+"\n")
47    sys.exit(2)
48stdin_obj=getattr(sys.stdin,"buffer",sys.stdin)
49stdout_obj=getattr(sys.stdout,"buffer",sys.stdout)
50stdin_fd=stdin_obj.fileno()
51stdout_fd=stdout_obj.fileno()
52stdin_open=True
53while True:
54    readers=[s]
55    if stdin_open:
56        readers.append(stdin_fd)
57    ready,_,_=select.select(readers,[],[])
58    if stdin_fd in ready:
59        data=os.read(stdin_fd,65536)
60        if data:
61            s.sendall(data)
62        else:
63            stdin_open=False
64            try:
65                s.shutdown(socket.SHUT_WR)
66            except OSError:
67                pass
68    if s in ready:
69        data=s.recv(65536)
70        if data:
71            os.write(stdout_fd,data)
72        else:
73            break
74"#;
75
76const PERL_BRIDGE: &str = r#"use strict; use warnings; use IO::Socket::INET; use IO::Socket::UNIX; use IO::Select; use Socket qw(SOCK_STREAM);
77my $mode = shift @ARGV;
78my $sock;
79if ($mode eq "tcp") {
80    my ($host, $port) = @ARGV;
81    $sock = IO::Socket::INET->new(PeerHost => $host, PeerPort => $port, Proto => "tcp") or die "connect tcp failed: $!";
82} elsif ($mode eq "unix") {
83    my ($path) = @ARGV;
84    $sock = IO::Socket::UNIX->new(Type => SOCK_STREAM, Peer => $path) or die "connect unix failed: $!";
85} else {
86    die "unsupported bridge mode: $mode";
87}
88binmode STDIN; binmode STDOUT; binmode $sock;
89my $select = IO::Select->new($sock, \*STDIN);
90while (1) {
91    for my $fh ($select->can_read) {
92        if ($fh == \*STDIN) {
93            my $buf = "";
94            my $n = sysread(STDIN, $buf, 65536);
95            die "read stdin failed: $!" unless defined $n;
96            if ($n == 0) {
97                $select->remove(\*STDIN);
98                shutdown($sock, 1);
99            } else {
100                write_all($sock, $buf);
101            }
102        } else {
103            my $buf = "";
104            my $n = sysread($sock, $buf, 65536);
105            die "read socket failed: $!" unless defined $n;
106            exit 0 if $n == 0;
107            write_all(\*STDOUT, $buf);
108        }
109    }
110}
111sub write_all {
112    my ($fh, $buf) = @_;
113    my $off = 0;
114    my $len = length($buf);
115    while ($off < $len) {
116        my $n = syswrite($fh, $buf, $len - $off, $off);
117        die "write failed: $!" unless defined $n;
118        $off += $n;
119    }
120}
121"#;
122
123const SHELL_BRIDGE_BODY: &str = r#"if command -v python3 >/dev/null 2>&1; then
124  echo "AFPSQL_BRIDGE_OK $AFPSQL_BRIDGE_NONCE" >&2
125  exec python3 -c "$AFPSQL_CONTAINER_PY_BRIDGE" "$@"
126fi
127if command -v python >/dev/null 2>&1; then
128  echo "AFPSQL_BRIDGE_OK $AFPSQL_BRIDGE_NONCE" >&2
129  exec python -c "$AFPSQL_CONTAINER_PY_BRIDGE" "$@"
130fi
131if command -v perl >/dev/null 2>&1; then
132  echo "AFPSQL_BRIDGE_OK $AFPSQL_BRIDGE_NONCE" >&2
133  exec perl -e "$AFPSQL_CONTAINER_PERL_BRIDGE" "$@"
134fi
135echo "afpsql container bridge requires python3, python, or perl in the container" >&2
136exit 127
137"#;
138
139pub struct ContainerBridgeGuard {
140    child: Option<tokio::process::Child>,
141    connection_task: Option<tokio::task::JoinHandle<()>>,
142    stderr_task: Option<tokio::task::JoinHandle<()>>,
143}
144
145impl ContainerBridgeGuard {
146    pub fn is_finished(&self) -> bool {
147        self.connection_task
148            .as_ref()
149            .map(|task| task.is_finished())
150            .unwrap_or(true)
151    }
152
153    pub async fn shutdown(mut self, timeout: Duration) {
154        if let Some(task) = self.connection_task.take() {
155            let mut task = task;
156            if tokio::time::timeout(timeout, &mut task).await.is_err() {
157                task.abort();
158            }
159        }
160        if let Some(task) = self.stderr_task.take() {
161            let mut task = task;
162            if tokio::time::timeout(timeout, &mut task).await.is_err() {
163                task.abort();
164            }
165        }
166        if let Some(mut child) = self.child.take() {
167            let _ = child.start_kill();
168            let _ = tokio::time::timeout(timeout, child.wait()).await;
169        }
170    }
171}
172
173impl Drop for ContainerBridgeGuard {
174    fn drop(&mut self) {
175        if let Some(task) = self.connection_task.as_ref() {
176            task.abort();
177        }
178        if let Some(task) = self.stderr_task.as_ref() {
179            task.abort();
180        }
181        if let Some(child) = self.child.as_mut() {
182            let _ = child.start_kill();
183        }
184    }
185}
186
187struct ContainerStdioStream {
188    stdout: ChildStdout,
189    stdin: ChildStdin,
190}
191
192impl AsyncRead for ContainerStdioStream {
193    fn poll_read(
194        mut self: Pin<&mut Self>,
195        cx: &mut Context<'_>,
196        buf: &mut ReadBuf<'_>,
197    ) -> Poll<std::io::Result<()>> {
198        Pin::new(&mut self.stdout).poll_read(cx, buf)
199    }
200}
201
202impl AsyncWrite for ContainerStdioStream {
203    fn poll_write(
204        mut self: Pin<&mut Self>,
205        cx: &mut Context<'_>,
206        buf: &[u8],
207    ) -> Poll<std::io::Result<usize>> {
208        Pin::new(&mut self.stdin).poll_write(cx, buf)
209    }
210
211    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
212        Pin::new(&mut self.stdin).poll_flush(cx)
213    }
214
215    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
216        Pin::new(&mut self.stdin).poll_shutdown(cx)
217    }
218}
219
220/// The driver-shaped view the argv builders work from.
221///
222/// The external surface is one flag per (driver, option) pair; this is what
223/// those flags collapse to once the driver has been inferred, so each builder
224/// reads one field instead of re-deriving the family.
225#[derive(Debug, Clone, PartialEq, Eq)]
226struct ContainerSettings {
227    driver: ContainerDriver,
228    runtime: String,
229    target: String,
230    user: Option<String>,
231    namespace: Option<String>,
232    context: Option<String>,
233    compose_files: Vec<String>,
234    compose_project: Option<String>,
235    pod_container: Option<String>,
236    ssh_destination: Option<String>,
237    ssh_options: Vec<String>,
238}
239
240#[derive(Debug, Clone, PartialEq, Eq)]
241enum ContainerTarget {
242    Tcp { host: String, port: u16 },
243    UnixSocket { path: String },
244}
245
246pub async fn connect_stdio_bridge(
247    cfg: &SessionConfig,
248) -> Result<(Client, ContainerBridgeGuard), ConnectError> {
249    let settings = resolve_container_settings(cfg)?;
250    let target = resolve_container_target(cfg)?;
251    let nonce = generate_bridge_nonce();
252    let (program, args) =
253        build_bridge_process(&settings, &target, &nonce).map_err(ConnectError::new)?;
254    let mut child = tokio::process::Command::new(&program)
255        .args(&args)
256        .stdin(Stdio::piped())
257        .stdout(Stdio::piped())
258        .stderr(Stdio::piped())
259        .kill_on_drop(true)
260        .spawn()
261        .map_err(|e| ConnectError::new(format!("start container bridge failed: {e}")))?;
262    let stderr = child.stderr.take();
263    let stderr_capture = Arc::new(Mutex::new(Vec::new()));
264    let (stderr_task, handshake_rx) = stderr
265        .map(|stderr| spawn_stderr_capture(stderr, Arc::clone(&stderr_capture), nonce.clone()))
266        .map(|(task, rx)| (Some(task), Some(rx)))
267        .unwrap_or((None, None));
268
269    let stdin = child.stdin.take().ok_or_else(|| {
270        ConnectError::new("start container bridge failed: stdin pipe unavailable")
271    })?;
272    let stdout = child.stdout.take().ok_or_else(|| {
273        ConnectError::new("start container bridge failed: stdout pipe unavailable")
274    })?;
275    if let Err(mut err) = wait_bridge_handshake(&mut child, &stderr_capture, handshake_rx).await {
276        // If the runtime could not find/exec the target, list the real container
277        // names instead of telling the agent to "check the target name".
278        let stderr_text = captured_stderr(&stderr_capture);
279        if stderr_indicates_missing_target(&stderr_text)
280            && let Some(list) = list_container_targets(&settings).await
281        {
282            err.hint = Some(match err.hint.take() {
283                Some(base) => format!("{base}; available containers: {list}"),
284                None => format!("available containers: {list}"),
285            });
286        }
287        return Err(err);
288    }
289    let stream = ContainerStdioStream { stdout, stdin };
290    let pg_cfg = resolve_pg_config(cfg)
291        .map_err(|e| ConnectError::new(format!("invalid container connection config: {e}")))?;
292    // The container transport parses a full DSN, so it must honor the same
293    // sslmode range as every other path; NoTls here made sslmode=require fail.
294    let endpoint = resolve_single_postgres_endpoint(&pg_cfg, "container transport")
295        .map_err(ConnectError::new)?;
296    let mut tls_connector = make_supported_tls()
297        .map_err(|e| ConnectError::new(format!("create TLS connector failed: {e}")))?;
298    let tls = <postgres_native_tls::MakeTlsConnector as MakeTlsConnect<ContainerStdioStream>>::make_tls_connect(
299        &mut tls_connector,
300        &postgres_tls_server_name(&endpoint),
301    )
302    .map_err(|e| ConnectError::new(format!("create PostgreSQL TLS stream failed: {e}")))?;
303    let (client, connection) = match pg_cfg.connect_raw(stream, tls).await {
304        Ok(connection) => connection,
305        Err(e) => {
306            let status = match tokio::time::timeout(Duration::from_millis(100), child.wait()).await
307            {
308                Ok(Ok(status)) => Some(status),
309                _ => child.try_wait().ok().flatten(),
310            };
311            tokio::time::sleep(Duration::from_millis(20)).await;
312            let stderr_text = captured_stderr(&stderr_capture);
313            return Err(enrich_container_connect_error(
314                ConnectError::from_pg_error("connect through container bridge failed", e),
315                status,
316                &stderr_text,
317            ));
318        }
319    };
320    let connection_task = tokio::spawn(async move {
321        let _ = connection.await;
322    });
323
324    Ok((
325        client,
326        ContainerBridgeGuard {
327            child: Some(child),
328            connection_task: Some(connection_task),
329            stderr_task,
330        },
331    ))
332}
333
334fn spawn_stderr_capture(
335    mut stderr: ChildStderr,
336    capture: Arc<Mutex<Vec<u8>>>,
337    nonce: String,
338) -> (tokio::task::JoinHandle<()>, oneshot::Receiver<bool>) {
339    let (handshake_tx, handshake_rx) = oneshot::channel();
340    let task = tokio::spawn(async move {
341        let mut buf = [0u8; 1024];
342        let mut pending = Vec::new();
343        let mut handshake_tx = Some(handshake_tx);
344        while let Ok(n) = stderr.read(&mut buf).await {
345            if n == 0 {
346                break;
347            }
348            pending.extend_from_slice(&buf[..n]);
349            while let Some(pos) = pending.iter().position(|b| *b == b'\n') {
350                let line = pending.drain(..=pos).collect::<Vec<_>>();
351                handle_stderr_line(line, &capture, &mut handshake_tx, &nonce);
352            }
353        }
354        if !pending.is_empty() {
355            handle_stderr_line(pending, &capture, &mut handshake_tx, &nonce);
356        }
357        if let Some(tx) = handshake_tx.take() {
358            let _ = tx.send(false);
359        }
360    });
361    (task, handshake_rx)
362}
363
364fn handle_stderr_line(
365    line: Vec<u8>,
366    capture: &Arc<Mutex<Vec<u8>>>,
367    handshake_tx: &mut Option<oneshot::Sender<bool>>,
368    nonce: &str,
369) {
370    if stderr_line_is_banner(&line, nonce) {
371        if let Some(tx) = handshake_tx.take() {
372            let _ = tx.send(true);
373        }
374        return;
375    }
376    append_stderr_capture(capture, &line);
377}
378
379fn stderr_line_is_banner(line: &[u8], nonce: &str) -> bool {
380    let line = line.strip_suffix(b"\n").unwrap_or(line);
381    let line = line.strip_suffix(b"\r").unwrap_or(line);
382    let expected = format!("{BRIDGE_READY_PREFIX} {nonce}");
383    line == expected.as_bytes()
384}
385
386fn append_stderr_capture(capture: &Arc<Mutex<Vec<u8>>>, bytes: &[u8]) {
387    if let Ok(mut captured) = capture.lock() {
388        let remaining = STDERR_CAPTURE_LIMIT.saturating_sub(captured.len());
389        if remaining > 0 {
390            captured.extend_from_slice(&bytes[..bytes.len().min(remaining)]);
391        }
392    }
393}
394
395async fn wait_bridge_handshake(
396    child: &mut tokio::process::Child,
397    stderr_capture: &Arc<Mutex<Vec<u8>>>,
398    handshake_rx: Option<oneshot::Receiver<bool>>,
399) -> Result<(), ConnectError> {
400    let Some(handshake_rx) = handshake_rx else {
401        return Err(handshake_connect_error(
402            child,
403            stderr_capture,
404            "container bridge stderr pipe unavailable for startup handshake",
405        )
406        .await);
407    };
408
409    match tokio::time::timeout(BRIDGE_HANDSHAKE_TIMEOUT, handshake_rx).await {
410        Ok(Ok(true)) => Ok(()),
411        Ok(Ok(false)) => Err(handshake_connect_error(
412            child,
413            stderr_capture,
414            "container bridge exited before startup handshake",
415        )
416        .await),
417        Ok(Err(_)) => Err(handshake_connect_error(
418            child,
419            stderr_capture,
420            "container bridge startup handshake channel closed",
421        )
422        .await),
423        Err(_) => Err(handshake_connect_error(
424            child,
425            stderr_capture,
426            "container bridge did not emit startup handshake before timeout",
427        )
428        .await),
429    }
430}
431
432async fn handshake_connect_error(
433    child: &mut tokio::process::Child,
434    stderr_capture: &Arc<Mutex<Vec<u8>>>,
435    message: &str,
436) -> ConnectError {
437    let status = child.try_wait().ok().flatten();
438    let stderr_text = captured_stderr(stderr_capture);
439    let mut err = ConnectError::new(message);
440    err.hint = Some(if stderr_text.is_empty() {
441        format!(
442            "the container bridge never reported {BRIDGE_READY_PREFIX}; check target name, runtime access, /bin/sh, and bridge interpreter prerequisites"
443        )
444    } else {
445        format!(
446            "the container bridge wrote diagnostics before {BRIDGE_READY_PREFIX}; container bridge stderr: {stderr_text}"
447        )
448    });
449    enrich_container_connect_error(err, status, &stderr_text)
450}
451
452fn captured_stderr(capture: &Arc<Mutex<Vec<u8>>>) -> String {
453    capture
454        .lock()
455        .ok()
456        .map(|captured| sanitize_diagnostic(&captured))
457        .filter(|text| !text.is_empty())
458        .unwrap_or_default()
459}
460
461fn sanitize_diagnostic(bytes: &[u8]) -> String {
462    let text = String::from_utf8_lossy(bytes);
463    let mut out = String::with_capacity(STDERR_HINT_BYTES);
464    for ch in text.chars() {
465        let mapped = if matches!(ch, '\n' | '\t') {
466            ch
467        } else if ch.is_control() {
468            ' '
469        } else {
470            ch
471        };
472        if out.len() + mapped.len_utf8() > STDERR_HINT_BYTES {
473            break;
474        }
475        out.push(mapped);
476    }
477    out.trim().to_string()
478}
479
480fn enrich_container_connect_error(
481    mut err: ConnectError,
482    status: Option<std::process::ExitStatus>,
483    stderr: &str,
484) -> ConnectError {
485    if let Some(hint) = container_bridge_hint(status, stderr) {
486        err.hint = Some(match err.hint.take() {
487            Some(base) => format!("{base}; {hint}"),
488            None => hint,
489        });
490    }
491    err
492}
493
494fn container_bridge_hint(status: Option<std::process::ExitStatus>, stderr: &str) -> Option<String> {
495    let trimmed = stderr.trim();
496    let lower = trimmed.to_ascii_lowercase();
497    let base = if lower.contains("requires python3, python, or perl") {
498        "the container bridge started but the container has no supported interpreter; install python3/python/perl, use a sidecar, or connect through the host instead"
499    } else if lower.contains("sh:") && lower.contains("not found") {
500        "the container bridge requires /bin/sh or compatible shell in the target container"
501    } else if lower.contains("no such container")
502        || lower.contains("not found")
503        || lower.contains("is not running")
504    {
505        "the container runtime could not exec into the target; check the container target name and running state"
506    } else if lower.contains("error from server") || lower.contains("pods") {
507        "kubectl could not exec into the target; check context, namespace, pod name, and cluster access"
508    } else if matches!(status.and_then(|s| s.code()), Some(125..=127)) {
509        "the container runtime or bridge command exited before PostgreSQL handshake; check runtime access, target name, shell, and bridge prerequisites"
510    } else if !trimmed.is_empty() {
511        "the container bridge wrote diagnostics before PostgreSQL handshake failed"
512    } else {
513        return None;
514    };
515
516    if trimmed.is_empty() {
517        Some(base.to_string())
518    } else {
519        Some(format!("{base}; container bridge stderr: {trimmed}"))
520    }
521}
522
523fn stderr_indicates_missing_target(stderr: &str) -> bool {
524    let lower = stderr.to_ascii_lowercase();
525    lower.contains("no such container")
526        || lower.contains("is not running")
527        || (lower.contains("not found") && lower.contains("container"))
528}
529
530/// A best-effort diagnostic listing is not worth blocking the error path on.
531const CONTAINER_LIST_TIMEOUT: Duration = Duration::from_secs(3);
532
533/// List the runtime's actual container names so a wrong-target error can name the
534/// real options. Best-effort; `None` on any failure or unsupported driver.
535async fn list_container_targets(settings: &ContainerSettings) -> Option<String> {
536    if settings.ssh_destination.is_some() {
537        return None;
538    }
539    let mut args = Vec::new();
540    match settings.driver {
541        ContainerDriver::Docker | ContainerDriver::Podman | ContainerDriver::Nerdctl => {
542            if settings.driver == ContainerDriver::Docker
543                && let Some(context) = settings.context.as_ref()
544            {
545                args.push(format!("--context={context}"));
546            }
547            // Running only — you can only exec into a running container, and the
548            // generic hint already covers the stopped-container case.
549            args.push("ps".to_string());
550            args.push("--format".to_string());
551            args.push("{{.Names}}".to_string());
552        }
553        // Compose service names and kubectl pods need different listings; skip
554        // for now and fall back to the generic hint.
555        ContainerDriver::Compose | ContainerDriver::Kubectl => return None,
556    }
557    let output = tokio::time::timeout(
558        CONTAINER_LIST_TIMEOUT,
559        tokio::process::Command::new(&settings.runtime)
560            .args(&args)
561            .stdin(Stdio::null())
562            .stdout(Stdio::piped())
563            .stderr(Stdio::null())
564            .kill_on_drop(true)
565            .output(),
566    )
567    .await
568    .ok()?
569    .ok()?;
570    let names: Vec<String> = String::from_utf8_lossy(&output.stdout)
571        .lines()
572        .map(|line| line.trim())
573        .filter(|line| !line.is_empty())
574        .map(|line| line.to_string())
575        .collect();
576    if names.is_empty() {
577        return None;
578    }
579    // Cap the list but never hide truncation — a silently cut list reads as
580    // "these are all of them" when the real target was dropped.
581    const MAX_LISTED: usize = 30;
582    if names.len() > MAX_LISTED {
583        let shown = names[..MAX_LISTED].join(", ");
584        let extra = names.len() - MAX_LISTED;
585        Some(format!("{shown} (+{extra} more)"))
586    } else {
587        Some(names.join(", "))
588    }
589}
590
591/// Fill each unset container field from its environment variable.
592///
593/// A pinned profile reads no environment at all: its endpoint and transport are
594/// the administrator's, and a variable must not redirect them.
595pub fn container_config_with_env(
596    container: &ContainerConfig,
597    profile_pinned: bool,
598) -> ContainerConfig {
599    if profile_pinned {
600        return container.clone();
601    }
602    let env = crate::runtime_env::nonempty;
603    ContainerConfig {
604        docker_name: container
605            .docker_name
606            .clone()
607            .or_else(|| env("AFPSQL_CONTAINER_DOCKER_NAME")),
608        docker_user: container
609            .docker_user
610            .clone()
611            .or_else(|| env("AFPSQL_CONTAINER_DOCKER_USER")),
612        docker_context: container
613            .docker_context
614            .clone()
615            .or_else(|| env("AFPSQL_CONTAINER_DOCKER_CONTEXT")),
616        docker_runtime: container
617            .docker_runtime
618            .clone()
619            .or_else(|| env("AFPSQL_CONTAINER_DOCKER_RUNTIME")),
620        podman_name: container
621            .podman_name
622            .clone()
623            .or_else(|| env("AFPSQL_CONTAINER_PODMAN_NAME")),
624        podman_user: container
625            .podman_user
626            .clone()
627            .or_else(|| env("AFPSQL_CONTAINER_PODMAN_USER")),
628        podman_runtime: container
629            .podman_runtime
630            .clone()
631            .or_else(|| env("AFPSQL_CONTAINER_PODMAN_RUNTIME")),
632        nerdctl_name: container
633            .nerdctl_name
634            .clone()
635            .or_else(|| env("AFPSQL_CONTAINER_NERDCTL_NAME")),
636        nerdctl_user: container
637            .nerdctl_user
638            .clone()
639            .or_else(|| env("AFPSQL_CONTAINER_NERDCTL_USER")),
640        nerdctl_runtime: container
641            .nerdctl_runtime
642            .clone()
643            .or_else(|| env("AFPSQL_CONTAINER_NERDCTL_RUNTIME")),
644        compose_service: container
645            .compose_service
646            .clone()
647            .or_else(|| env("AFPSQL_CONTAINER_COMPOSE_SERVICE")),
648        compose_user: container
649            .compose_user
650            .clone()
651            .or_else(|| env("AFPSQL_CONTAINER_COMPOSE_USER")),
652        compose_files: if container.compose_files.is_empty() {
653            crate::runtime_env::colon_list("AFPSQL_CONTAINER_COMPOSE_FILE")
654        } else {
655            container.compose_files.clone()
656        },
657        compose_project: container
658            .compose_project
659            .clone()
660            .or_else(|| env("AFPSQL_CONTAINER_COMPOSE_PROJECT")),
661        compose_runtime: container
662            .compose_runtime
663            .clone()
664            .or_else(|| env("AFPSQL_CONTAINER_COMPOSE_RUNTIME")),
665        kubectl_pod: container
666            .kubectl_pod
667            .clone()
668            .or_else(|| env("AFPSQL_CONTAINER_KUBECTL_POD")),
669        kubectl_container: container
670            .kubectl_container
671            .clone()
672            .or_else(|| env("AFPSQL_CONTAINER_KUBECTL_CONTAINER")),
673        kubectl_namespace: container
674            .kubectl_namespace
675            .clone()
676            .or_else(|| env("AFPSQL_CONTAINER_KUBECTL_NAMESPACE")),
677        kubectl_context: container
678            .kubectl_context
679            .clone()
680            .or_else(|| env("AFPSQL_CONTAINER_KUBECTL_CONTEXT")),
681        kubectl_runtime: container
682            .kubectl_runtime
683            .clone()
684            .or_else(|| env("AFPSQL_CONTAINER_KUBECTL_RUNTIME")),
685    }
686}
687
688fn resolve_container_settings(cfg: &SessionConfig) -> Result<ContainerSettings, String> {
689    let container = container_config_with_env(&cfg.container, cfg.profile_pinned);
690    let ssh_destination = cfg.ssh.destination.clone().or_else(|| {
691        if cfg.profile_pinned {
692            None
693        } else {
694            crate::runtime_env::nonempty("AFPSQL_SSH")
695        }
696    });
697
698    let Some(driver) = container.selected_driver()? else {
699        return Err(
700            "container transport requires one of --container-docker-name, --container-podman-name, --container-nerdctl-name, --container-compose-service, or --container-kubectl-pod"
701                .to_string(),
702        );
703    };
704    // An option flag on its own picks a driver but names nothing to exec into,
705    // so the missing-target error has to speak that family's vocabulary:
706    // a compose service, a kubectl pod, a container name everywhere else.
707    let target_flag = driver.target_flag();
708    let (target, user, runtime) = match driver {
709        ContainerDriver::Docker => (
710            container.docker_name.clone(),
711            container.docker_user.clone(),
712            container.docker_runtime.clone(),
713        ),
714        ContainerDriver::Podman => (
715            container.podman_name.clone(),
716            container.podman_user.clone(),
717            container.podman_runtime.clone(),
718        ),
719        ContainerDriver::Nerdctl => (
720            container.nerdctl_name.clone(),
721            container.nerdctl_user.clone(),
722            container.nerdctl_runtime.clone(),
723        ),
724        ContainerDriver::Compose => (
725            container.compose_service.clone(),
726            container.compose_user.clone(),
727            container.compose_runtime.clone(),
728        ),
729        // No kubectl user: `kubectl exec` has no exec-as-user option, so the
730        // flag that would set one does not exist.
731        ContainerDriver::Kubectl => (
732            container.kubectl_pod.clone(),
733            None,
734            container.kubectl_runtime.clone(),
735        ),
736    };
737    let Some(target) = target else {
738        return Err(format!(
739            "{target_flag} is required when other {} options are set",
740            driver.flag_family()
741        ));
742    };
743    if target.trim().is_empty() {
744        return Err(format!("{target_flag} requires a non-empty value"));
745    }
746    if let Some(destination) = ssh_destination.as_ref() {
747        if destination.trim().is_empty() {
748            return Err("--ssh requires a non-empty USER@HOST destination".to_string());
749        }
750    } else if !cfg.ssh.options.is_empty() {
751        return Err(
752            "--ssh is required when --ssh-option is combined with container transport".to_string(),
753        );
754    }
755    if cfg.ssh.has_tunnel_or_bridge_options() {
756        return Err("container transport with --ssh supports only --ssh and --ssh-option; SSH tunnel and sudo bridge options are for non-container SSH transport".to_string());
757    }
758
759    Ok(ContainerSettings {
760        runtime: runtime.unwrap_or_else(|| driver.default_runtime().to_string()),
761        driver,
762        target,
763        user,
764        namespace: container.kubectl_namespace.clone(),
765        context: match driver {
766            ContainerDriver::Docker => container.docker_context.clone(),
767            ContainerDriver::Kubectl => container.kubectl_context.clone(),
768            // Podman, nerdctl, and Compose select no context, so no flag in
769            // those families can name one.
770            ContainerDriver::Podman | ContainerDriver::Nerdctl | ContainerDriver::Compose => None,
771        },
772        compose_files: container.compose_files.clone(),
773        compose_project: container.compose_project.clone(),
774        pod_container: container.kubectl_container.clone(),
775        ssh_destination,
776        ssh_options: cfg.ssh.options.clone(),
777    })
778}
779
780fn resolve_container_target(cfg: &SessionConfig) -> Result<ContainerTarget, String> {
781    let pg_cfg =
782        resolve_pg_config(cfg).map_err(|e| format!("invalid container connection config: {e}"))?;
783    match resolve_single_postgres_endpoint(&pg_cfg, "container transport")? {
784        PostgresEndpoint::Tcp { host, port } => Ok(ContainerTarget::Tcp { host, port }),
785        PostgresEndpoint::UnixSocket { directory, port } => Ok(ContainerTarget::UnixSocket {
786            path: socket_file_from_dir(&directory, port),
787        }),
788    }
789}
790
791fn socket_file_from_dir(dir: &str, port: u16) -> String {
792    format!("{}/.s.PGSQL.{port}", dir.trim_end_matches('/'))
793}
794
795fn build_bridge_process(
796    settings: &ContainerSettings,
797    target: &ContainerTarget,
798    nonce: &str,
799) -> Result<(String, Vec<String>), String> {
800    if settings.ssh_destination.is_some() {
801        Ok((
802            "ssh".to_string(),
803            build_bridge_ssh_args(settings, target, nonce)?,
804        ))
805    } else {
806        Ok((
807            settings.runtime.clone(),
808            build_container_exec_args(settings, target, nonce)?,
809        ))
810    }
811}
812
813fn build_container_exec_args(
814    settings: &ContainerSettings,
815    target: &ContainerTarget,
816    nonce: &str,
817) -> Result<Vec<String>, String> {
818    let command = bridge_command_args(target, nonce);
819    let mut args = Vec::new();
820    match settings.driver {
821        ContainerDriver::Docker | ContainerDriver::Podman | ContainerDriver::Nerdctl => {
822            if settings.driver == ContainerDriver::Docker
823                && let Some(context) = settings.context.as_ref()
824            {
825                args.push(format!("--context={context}"));
826            }
827            args.push("exec".to_string());
828            args.push("-i".to_string());
829            if let Some(user) = settings.user.as_ref() {
830                args.push("--user".to_string());
831                args.push(user.clone());
832            }
833            args.push("--".to_string());
834            args.push(settings.target.clone());
835            args.extend(command);
836        }
837        ContainerDriver::Compose => {
838            if !is_docker_compose_runtime(&settings.runtime) {
839                args.push("compose".to_string());
840            }
841            for file in &settings.compose_files {
842                args.push("-f".to_string());
843                args.push(file.clone());
844            }
845            if let Some(project) = settings.compose_project.as_ref() {
846                args.push("-p".to_string());
847                args.push(project.clone());
848            }
849            args.push("exec".to_string());
850            args.push("-T".to_string());
851            if let Some(user) = settings.user.as_ref() {
852                args.push("--user".to_string());
853                args.push(user.clone());
854            }
855            args.push("--".to_string());
856            args.push(settings.target.clone());
857            args.extend(command);
858        }
859        ContainerDriver::Kubectl => {
860            if let Some(context) = settings.context.as_ref() {
861                args.push(format!("--context={context}"));
862            }
863            if let Some(namespace) = settings.namespace.as_ref() {
864                args.push(format!("--namespace={namespace}"));
865            }
866            args.push("exec".to_string());
867            args.push("-i".to_string());
868            args.push(settings.target.clone());
869            if let Some(container) = settings.pod_container.as_ref() {
870                args.push("-c".to_string());
871                args.push(container.clone());
872            }
873            args.push("--".to_string());
874            args.extend(command);
875        }
876    }
877    Ok(args)
878}
879
880fn is_docker_compose_runtime(runtime: &str) -> bool {
881    runtime
882        .rsplit(['/', '\\'])
883        .next()
884        .is_some_and(|name| name == "docker-compose")
885}
886
887fn bridge_command_args(target: &ContainerTarget, nonce: &str) -> Vec<String> {
888    let mut args = vec![
889        "sh".to_string(),
890        "-c".to_string(),
891        shell_bridge_script(nonce),
892        "afpsql-container-bridge".to_string(),
893    ];
894    match target {
895        ContainerTarget::Tcp { host, port } => {
896            args.push("tcp".to_string());
897            args.push(host.clone());
898            args.push(port.to_string());
899        }
900        ContainerTarget::UnixSocket { path } => {
901            args.push("unix".to_string());
902            args.push(path.clone());
903        }
904    }
905    args
906}
907
908fn shell_bridge_script(nonce: &str) -> String {
909    format!(
910        "AFPSQL_BRIDGE_NONCE={}; AFPSQL_CONTAINER_PY_BRIDGE={}; AFPSQL_CONTAINER_PERL_BRIDGE={}; {}",
911        shell_quote(nonce),
912        shell_quote(PYTHON_BRIDGE),
913        shell_quote(PERL_BRIDGE),
914        SHELL_BRIDGE_BODY
915    )
916}
917
918fn build_bridge_ssh_args(
919    settings: &ContainerSettings,
920    target: &ContainerTarget,
921    nonce: &str,
922) -> Result<Vec<String>, String> {
923    let mut args = vec![
924        "-T".to_string(),
925        "-o".to_string(),
926        "BatchMode=yes".to_string(),
927    ];
928    for option in &settings.ssh_options {
929        args.push("-o".to_string());
930        args.push(option.clone());
931    }
932    if let Some(destination) = settings.ssh_destination.as_ref() {
933        args.push(destination.clone());
934    }
935    args.push(remote_container_command(settings, target, nonce)?);
936    Ok(args)
937}
938
939fn remote_container_command(
940    settings: &ContainerSettings,
941    target: &ContainerTarget,
942    nonce: &str,
943) -> Result<String, String> {
944    Ok(std::iter::once(settings.runtime.clone())
945        .chain(build_container_exec_args(settings, target, nonce)?)
946        .map(|arg| shell_quote(&arg))
947        .collect::<Vec<_>>()
948        .join(" "))
949}
950
951fn shell_quote(value: &str) -> String {
952    if value.is_empty() {
953        return "''".to_string();
954    }
955    if value.bytes().all(|b| {
956        b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'.' | b'/' | b':' | b'@' | b'=')
957    }) {
958        return value.to_string();
959    }
960    format!("'{}'", value.replace('\'', "'\\''"))
961}
962
963#[cfg(test)]
964mod tests {
965    use super::*;
966    use crate::types::ContainerConfig;
967
968    const TEST_NONCE: &str = "deadbeefcafef00d";
969
970    fn settings(driver: ContainerDriver) -> ContainerSettings {
971        ContainerSettings {
972            driver,
973            runtime: driver.default_runtime().to_string(),
974            target: "pg".to_string(),
975            user: Some("postgres".to_string()),
976            namespace: None,
977            context: None,
978            compose_files: vec![],
979            compose_project: None,
980            pod_container: None,
981            ssh_destination: None,
982            ssh_options: vec![],
983        }
984    }
985
986    #[test]
987    fn typed_drivers_select_fixed_default_runtimes() {
988        for (driver, runtime) in [
989            (ContainerDriver::Docker, "docker"),
990            (ContainerDriver::Podman, "podman"),
991            (ContainerDriver::Nerdctl, "nerdctl"),
992            (ContainerDriver::Compose, "docker"),
993            (ContainerDriver::Kubectl, "kubectl"),
994        ] {
995            assert_eq!(driver.default_runtime(), runtime);
996        }
997    }
998
999    #[test]
1000    fn missing_target_detection() {
1001        assert!(stderr_indicates_missing_target(
1002            "Error response from daemon: No such container: pg-typo"
1003        ));
1004        assert!(stderr_indicates_missing_target("container is not running"));
1005        assert!(!stderr_indicates_missing_target(
1006            "sh: python3: not found in PATH"
1007        ));
1008    }
1009
1010    #[test]
1011    fn bridge_args_target_tcp_for_container_cli_like_driver() -> Result<(), String> {
1012        let args = build_container_exec_args(
1013            &settings(ContainerDriver::Docker),
1014            &ContainerTarget::Tcp {
1015                host: "127.0.0.1".to_string(),
1016                port: 5432,
1017            },
1018            TEST_NONCE,
1019        )?;
1020        assert_eq!(args[0], "exec");
1021        assert!(args.contains(&"-i".to_string()));
1022        assert!(args.contains(&"--user".to_string()));
1023        assert!(args.contains(&"postgres".to_string()));
1024        assert!(args.contains(&"pg".to_string()));
1025        assert!(args.contains(&"tcp".to_string()));
1026        assert!(args.contains(&"127.0.0.1".to_string()));
1027        assert!(args.contains(&"5432".to_string()));
1028        assert!(!args.contains(&"-e".to_string()));
1029        Ok(())
1030    }
1031
1032    #[test]
1033    fn bridge_args_target_unix_socket() -> Result<(), String> {
1034        let args = build_container_exec_args(
1035            &ContainerSettings {
1036                user: None,
1037                ..settings(ContainerDriver::Docker)
1038            },
1039            &ContainerTarget::UnixSocket {
1040                path: "/var/run/postgresql/.s.PGSQL.5432".to_string(),
1041            },
1042            TEST_NONCE,
1043        )?;
1044        assert!(args.contains(&"unix".to_string()));
1045        assert!(args.contains(&"/var/run/postgresql/.s.PGSQL.5432".to_string()));
1046        Ok(())
1047    }
1048
1049    #[test]
1050    fn compose_driver_uses_compose_exec_without_tty() -> Result<(), String> {
1051        let settings = ContainerSettings {
1052            compose_files: vec!["compose.yml".to_string(), "compose.prod.yml".to_string()],
1053            compose_project: Some("demo".to_string()),
1054            ..settings(ContainerDriver::Compose)
1055        };
1056        let (program, args) = build_bridge_process(
1057            &settings,
1058            &ContainerTarget::Tcp {
1059                host: "db".to_string(),
1060                port: 5432,
1061            },
1062            TEST_NONCE,
1063        )?;
1064        assert_eq!(program, "docker");
1065        assert_eq!(
1066            &args[0..8],
1067            [
1068                "compose",
1069                "-f",
1070                "compose.yml",
1071                "-f",
1072                "compose.prod.yml",
1073                "-p",
1074                "demo",
1075                "exec"
1076            ]
1077        );
1078        assert_eq!(args[8], "-T");
1079        assert!(args.contains(&"pg".to_string()));
1080        Ok(())
1081    }
1082
1083    #[test]
1084    fn compose_runtime_docker_compose_skips_subcommand_prefix() -> Result<(), String> {
1085        let settings = ContainerSettings {
1086            runtime: "/usr/local/bin/docker-compose".to_string(),
1087            ..settings(ContainerDriver::Compose)
1088        };
1089        let (program, args) = build_bridge_process(
1090            &settings,
1091            &ContainerTarget::Tcp {
1092                host: "db".to_string(),
1093                port: 5432,
1094            },
1095            TEST_NONCE,
1096        )?;
1097        assert_eq!(program, "/usr/local/bin/docker-compose");
1098        assert_eq!(&args[0..2], ["exec", "-T"]);
1099        Ok(())
1100    }
1101
1102    #[test]
1103    fn kubectl_driver_uses_exec_separator() -> Result<(), String> {
1104        let settings = ContainerSettings {
1105            user: None,
1106            namespace: Some("prod".to_string()),
1107            context: Some("cluster-a".to_string()),
1108            ..settings(ContainerDriver::Kubectl)
1109        };
1110        let (program, args) = build_bridge_process(
1111            &settings,
1112            &ContainerTarget::Tcp {
1113                host: "127.0.0.1".to_string(),
1114                port: 5432,
1115            },
1116            TEST_NONCE,
1117        )?;
1118        assert_eq!(program, "kubectl");
1119        assert_eq!(
1120            &args[0..6],
1121            [
1122                "--context=cluster-a",
1123                "--namespace=prod",
1124                "exec",
1125                "-i",
1126                "pg",
1127                "--"
1128            ]
1129        );
1130        assert!(args.contains(&"sh".to_string()));
1131        Ok(())
1132    }
1133
1134    #[test]
1135    fn kubectl_driver_inserts_pod_container_before_separator() -> Result<(), String> {
1136        let settings = ContainerSettings {
1137            user: None,
1138            pod_container: Some("postgres".to_string()),
1139            ..settings(ContainerDriver::Kubectl)
1140        };
1141        let (_, args) = build_bridge_process(
1142            &settings,
1143            &ContainerTarget::Tcp {
1144                host: "127.0.0.1".to_string(),
1145                port: 5432,
1146            },
1147            TEST_NONCE,
1148        )?;
1149        assert_eq!(
1150            &args[0..7],
1151            ["exec", "-i", "pg", "-c", "postgres", "--", "sh"]
1152        );
1153        Ok(())
1154    }
1155
1156    #[test]
1157    fn bridge_banner_line_is_not_captured_as_diagnostic() {
1158        let capture = Arc::new(Mutex::new(Vec::new()));
1159        let (tx, mut rx) = oneshot::channel();
1160        let mut tx = Some(tx);
1161        let line = format!("AFPSQL_BRIDGE_OK {TEST_NONCE}\n").into_bytes();
1162        handle_stderr_line(line, &capture, &mut tx, TEST_NONCE);
1163        assert!(matches!(rx.try_recv(), Ok(true)));
1164        assert!(captured_stderr(&capture).is_empty());
1165    }
1166
1167    #[test]
1168    fn bridge_banner_without_matching_nonce_is_treated_as_diagnostic() {
1169        let capture = Arc::new(Mutex::new(Vec::new()));
1170        let (tx, mut rx) = oneshot::channel();
1171        let mut tx = Some(tx);
1172        let line = b"AFPSQL_BRIDGE_OK 0000000000000000\n".to_vec();
1173        handle_stderr_line(line, &capture, &mut tx, TEST_NONCE);
1174        assert!(rx.try_recv().is_err());
1175        assert!(captured_stderr(&capture).contains("AFPSQL_BRIDGE_OK"));
1176    }
1177
1178    #[test]
1179    fn bridge_banner_without_nonce_is_treated_as_diagnostic() {
1180        let capture = Arc::new(Mutex::new(Vec::new()));
1181        let (tx, mut rx) = oneshot::channel();
1182        let mut tx = Some(tx);
1183        handle_stderr_line(
1184            b"AFPSQL_BRIDGE_OK\n".to_vec(),
1185            &capture,
1186            &mut tx,
1187            TEST_NONCE,
1188        );
1189        assert!(rx.try_recv().is_err());
1190        assert!(captured_stderr(&capture).contains("AFPSQL_BRIDGE_OK"));
1191    }
1192
1193    #[test]
1194    fn sanitize_diagnostic_strips_control_chars_and_truncates() {
1195        let input = b"\x1b[31mboom\x1b[0m\nnext line\x00trailing";
1196        let cleaned = sanitize_diagnostic(input);
1197        assert!(!cleaned.contains('\x1b'));
1198        assert!(!cleaned.contains('\x00'));
1199        assert!(cleaned.contains("boom"));
1200        assert!(cleaned.contains("next line"));
1201
1202        let big = vec![b'A'; 4096];
1203        let cleaned = sanitize_diagnostic(&big);
1204        assert!(cleaned.len() <= STDERR_HINT_BYTES);
1205    }
1206
1207    #[test]
1208    fn bridge_process_uses_remote_ssh_container_command() -> Result<(), String> {
1209        let settings = ContainerSettings {
1210            driver: ContainerDriver::Podman,
1211            runtime: "podman".to_string(),
1212            target: "pg remote".to_string(),
1213            user: Some("postgres".to_string()),
1214            namespace: None,
1215            context: None,
1216            compose_files: vec![],
1217            compose_project: None,
1218            pod_container: None,
1219            ssh_destination: Some("root@example.com".to_string()),
1220            ssh_options: vec!["ProxyJump=bastion".to_string()],
1221        };
1222        let (program, args) = build_bridge_process(
1223            &settings,
1224            &ContainerTarget::Tcp {
1225                host: "host.containers.internal".to_string(),
1226                port: 5432,
1227            },
1228            TEST_NONCE,
1229        )?;
1230
1231        assert_eq!(program, "ssh");
1232        assert!(args.contains(&"BatchMode=yes".to_string()));
1233        assert!(args.contains(&"ProxyJump=bastion".to_string()));
1234        assert_eq!(
1235            args.iter().rev().nth(1),
1236            Some(&"root@example.com".to_string())
1237        );
1238
1239        let command = args.last().cloned().unwrap_or_default();
1240        assert!(command.starts_with("podman exec -i "));
1241        assert!(command.contains("'pg remote'"));
1242        assert!(command.contains("host.containers.internal"));
1243        Ok(())
1244    }
1245
1246    #[test]
1247    fn remote_container_command_quotes_shell_torture_values() -> Result<(), String> {
1248        let target = "pg 'quoted' \"$USER\" `whoami`\nnext".to_string();
1249        let user = "postgres '$HOME' `id`\nnext".to_string();
1250        let context = "ctx '$VAR' `cmd`\nnext".to_string();
1251        let settings = ContainerSettings {
1252            driver: ContainerDriver::Docker,
1253            runtime: "docker".to_string(),
1254            target: target.clone(),
1255            user: Some(user.clone()),
1256            namespace: None,
1257            context: Some(context.clone()),
1258            compose_files: vec![],
1259            compose_project: None,
1260            pod_container: None,
1261            ssh_destination: Some("root@example.com".to_string()),
1262            ssh_options: vec![],
1263        };
1264
1265        let command = remote_container_command(
1266            &settings,
1267            &ContainerTarget::Tcp {
1268                host: "127.0.0.1".to_string(),
1269                port: 5432,
1270            },
1271            TEST_NONCE,
1272        )?;
1273
1274        assert!(command.contains(&shell_quote(&format!("--context={context}"))));
1275        assert!(command.contains(&shell_quote(&target)));
1276        assert!(command.contains(&shell_quote(&user)));
1277        Ok(())
1278    }
1279
1280    #[test]
1281    fn target_from_dsn_tcp() -> Result<(), String> {
1282        let cfg = SessionConfig {
1283            container: crate::types::ContainerConfig {
1284                docker_name: Some("pg".to_string()),
1285                ..Default::default()
1286            },
1287            dsn_secret: Some("postgresql://u:p@127.0.0.1:6543/db".to_string()),
1288            ..Default::default()
1289        };
1290        assert_eq!(
1291            resolve_container_target(&cfg)?,
1292            ContainerTarget::Tcp {
1293                host: "127.0.0.1".to_string(),
1294                port: 6543,
1295            }
1296        );
1297        Ok(())
1298    }
1299
1300    #[test]
1301    fn target_from_unix_socket_dir() -> Result<(), String> {
1302        let cfg = SessionConfig {
1303            container: crate::types::ContainerConfig {
1304                docker_name: Some("pg".to_string()),
1305                ..Default::default()
1306            },
1307            host: Some("/var/run/postgresql".to_string()),
1308            port: Some(5433),
1309            ..Default::default()
1310        };
1311        assert_eq!(
1312            resolve_container_target(&cfg)?,
1313            ContainerTarget::UnixSocket {
1314                path: "/var/run/postgresql/.s.PGSQL.5433".to_string(),
1315            }
1316        );
1317        Ok(())
1318    }
1319
1320    #[test]
1321    fn settings_require_family_name_when_family_options_set() {
1322        let cfg = SessionConfig {
1323            container: crate::types::ContainerConfig {
1324                docker_user: Some("postgres".to_string()),
1325                ..Default::default()
1326            },
1327            ..Default::default()
1328        };
1329        let err = resolve_container_settings(&cfg);
1330        assert!(matches!(err, Err(message) if message
1331                == "--container-docker-name is required when other --container-docker-* options are set"));
1332    }
1333
1334    #[test]
1335    fn settings_require_ssh_destination_when_ssh_options_set() {
1336        let cfg = SessionConfig {
1337            container: crate::types::ContainerConfig {
1338                docker_name: Some("pg".to_string()),
1339                ..Default::default()
1340            },
1341            ssh: crate::types::SshConfig {
1342                options: vec!["ProxyJump=bastion".to_string()],
1343                ..Default::default()
1344            },
1345            ..Default::default()
1346        };
1347        let err = resolve_container_settings(&cfg);
1348        assert!(matches!(err, Err(message) if message.contains("--ssh is required")));
1349    }
1350
1351    #[test]
1352    fn settings_reject_tunnel_only_ssh_options_for_container_transport() {
1353        let cfg = SessionConfig {
1354            container: crate::types::ContainerConfig {
1355                docker_name: Some("pg".to_string()),
1356                ..Default::default()
1357            },
1358            ssh: crate::types::SshConfig {
1359                destination: Some("user@example.com".to_string()),
1360                local_port: Some(15432),
1361                ..Default::default()
1362            },
1363            ..Default::default()
1364        };
1365        let err = resolve_container_settings(&cfg);
1366        assert!(matches!(err, Err(message) if message.contains("supports only --ssh")));
1367    }
1368
1369    /// `kubectl exec` has no exec-as-user option, so the kubectl family has no
1370    /// user flag to carry one and the settings it resolves have no user at all.
1371    #[test]
1372    fn kubectl_family_cannot_carry_an_exec_user() {
1373        let cfg = SessionConfig {
1374            container: crate::types::ContainerConfig {
1375                kubectl_pod: Some("pod/app".to_string()),
1376                kubectl_namespace: Some("prod".to_string()),
1377                ..Default::default()
1378            },
1379            ..Default::default()
1380        };
1381        let settings = resolve_container_settings(&cfg).expect("resolve kubectl settings");
1382        assert_eq!(settings.driver, ContainerDriver::Kubectl);
1383        assert!(settings.user.is_none());
1384    }
1385
1386    #[test]
1387    fn settings_reject_two_driver_families() {
1388        let cfg = SessionConfig {
1389            container: crate::types::ContainerConfig {
1390                docker_name: Some("pg".to_string()),
1391                kubectl_pod: Some("pod/app".to_string()),
1392                ..Default::default()
1393            },
1394            ..Default::default()
1395        };
1396        let err = resolve_container_settings(&cfg);
1397        assert!(matches!(err, Err(message) if message
1398                == "--container-docker-name cannot be combined with --container-kubectl-pod; each container driver has its own flag family"));
1399    }
1400
1401    /// A driver that has no context selection has no flag able to name one, so
1402    /// the podman family plus a context is just the two-family rejection.
1403    #[test]
1404    fn settings_reject_context_from_another_family() {
1405        let cfg = SessionConfig {
1406            container: crate::types::ContainerConfig {
1407                podman_name: Some("pg".to_string()),
1408                docker_context: Some("prod".to_string()),
1409                ..Default::default()
1410            },
1411            ..Default::default()
1412        };
1413        let err = resolve_container_settings(&cfg);
1414        assert!(matches!(err, Err(message) if message
1415                == "--container-docker-context cannot be combined with --container-podman-name; each container driver has its own flag family"));
1416    }
1417
1418    #[test]
1419    fn settings_accept_compose_file_env_fallback() {
1420        let _guard = crate::test_env::env_lock();
1421        let old = std::env::var("AFPSQL_CONTAINER_COMPOSE_FILE").ok();
1422        // SAFETY: this test module's environment lock is held for the mutation.
1423        unsafe {
1424            std::env::set_var("AFPSQL_CONTAINER_COMPOSE_FILE", "base.yml:prod.yml");
1425        }
1426        let cfg = SessionConfig {
1427            container: crate::types::ContainerConfig {
1428                compose_service: Some("pg".to_string()),
1429                ..Default::default()
1430            },
1431            ..Default::default()
1432        };
1433        let settings = resolve_container_settings(&cfg);
1434        match old {
1435            // SAFETY: this test module's environment lock is still held here.
1436            Some(value) => unsafe { std::env::set_var("AFPSQL_CONTAINER_COMPOSE_FILE", value) },
1437            // SAFETY: this test module's environment lock is still held here.
1438            None => unsafe { std::env::remove_var("AFPSQL_CONTAINER_COMPOSE_FILE") },
1439        }
1440        assert!(matches!(
1441            settings,
1442            Ok(ContainerSettings {
1443                compose_files,
1444                ..
1445            }) if compose_files == vec!["base.yml".to_string(), "prod.yml".to_string()]
1446        ));
1447    }
1448
1449    #[test]
1450    fn pinned_profile_ignores_all_container_environment_fallbacks() {
1451        let _guard = crate::test_env::env_lock();
1452        // Every container variable, including the families the pinned profile
1453        // did not choose: a hostile value must not be able to redirect the
1454        // endpoint, nor to smuggle in a second driver family.
1455        let names = [
1456            "AFPSQL_CONTAINER_DOCKER_NAME",
1457            "AFPSQL_CONTAINER_DOCKER_USER",
1458            "AFPSQL_CONTAINER_DOCKER_CONTEXT",
1459            "AFPSQL_CONTAINER_DOCKER_RUNTIME",
1460            "AFPSQL_CONTAINER_PODMAN_NAME",
1461            "AFPSQL_CONTAINER_PODMAN_USER",
1462            "AFPSQL_CONTAINER_PODMAN_RUNTIME",
1463            "AFPSQL_CONTAINER_NERDCTL_NAME",
1464            "AFPSQL_CONTAINER_NERDCTL_USER",
1465            "AFPSQL_CONTAINER_NERDCTL_RUNTIME",
1466            "AFPSQL_CONTAINER_COMPOSE_SERVICE",
1467            "AFPSQL_CONTAINER_COMPOSE_USER",
1468            "AFPSQL_CONTAINER_COMPOSE_FILE",
1469            "AFPSQL_CONTAINER_COMPOSE_PROJECT",
1470            "AFPSQL_CONTAINER_COMPOSE_RUNTIME",
1471            "AFPSQL_CONTAINER_KUBECTL_POD",
1472            "AFPSQL_CONTAINER_KUBECTL_CONTAINER",
1473            "AFPSQL_CONTAINER_KUBECTL_NAMESPACE",
1474            "AFPSQL_CONTAINER_KUBECTL_CONTEXT",
1475            "AFPSQL_CONTAINER_KUBECTL_RUNTIME",
1476            "AFPSQL_SSH",
1477        ];
1478        let prior = names
1479            .iter()
1480            .map(|name| ((*name).to_string(), std::env::var(name).ok()))
1481            .collect::<Vec<_>>();
1482        for name in names {
1483            // SAFETY: the shared test environment lock serializes mutations.
1484            unsafe { std::env::set_var(name, "hostile-value") };
1485        }
1486
1487        let cfg = SessionConfig {
1488            profile_pinned: true,
1489            container: ContainerConfig {
1490                docker_name: Some("trusted-container".to_string()),
1491                ..Default::default()
1492            },
1493            ..Default::default()
1494        };
1495        let settings = resolve_container_settings(&cfg).expect("resolve pinned profile");
1496        assert_eq!(settings.driver, ContainerDriver::Docker);
1497        assert_eq!(settings.runtime, "docker");
1498        assert_eq!(settings.target, "trusted-container");
1499        assert!(settings.user.is_none());
1500        assert!(settings.context.is_none());
1501        assert!(settings.compose_files.is_empty());
1502        assert!(settings.pod_container.is_none());
1503        assert!(settings.ssh_destination.is_none());
1504
1505        for (name, value) in prior {
1506            match value {
1507                // SAFETY: the shared test environment lock is still held.
1508                Some(value) => unsafe { std::env::set_var(name, value) },
1509                // SAFETY: the shared test environment lock is still held.
1510                None => unsafe { std::env::remove_var(name) },
1511            }
1512        }
1513    }
1514
1515    #[test]
1516    fn container_bridge_hint_classifies_interpreter_failure() {
1517        let hint = container_bridge_hint(
1518            None,
1519            "afpsql container bridge requires python3, python, or perl in the container",
1520        )
1521        .unwrap_or_default();
1522        assert!(hint.contains("no supported interpreter"));
1523        assert!(hint.contains("container bridge stderr"));
1524    }
1525}