Fix SSH

Author: ptr-yudai

ptrlib/connection/ssh.py

@@ -123,6 +123,53 @@ def _ensure_windows_askpass_cmd() -> str:
     return str(path)
 
 
+def _ensure_posix_askpass_sh() -> str:
+    """Create a minimal SSH_ASKPASS helper for POSIX and return its path.
+
+    OpenSSH can read passwords via an external helper specified by SSH_ASKPASS.
+    We provide a tiny, non-interactive helper that prints the password from an
+    environment variable. This avoids fragile prompt parsing and works even
+    when stdin is not a TTY or when ssh suppresses prompts.
+    """
+    path = Path(tempfile.gettempdir()) / "ptrlib-ssh-askpass.sh"
+
+    content = (
+        "#!/bin/sh\n"
+        "# ptrlib askpass helper (POSIX)\n"
+        "# $1 is the prompt (ignored).\n"
+        "# Print the password from $PTRLIB_SSH_PASSWORD without a trailing newline.\n"
+        "if [ -n \"$PTRLIB_SSH_PASSWORD\" ]; then\n"
+        "  printf '%s' \"$PTRLIB_SSH_PASSWORD\"\n"
+        "fi\n"
+    )
+
+    try:
+        if path.is_file():
+            try:
+                if path.read_text(encoding="utf-8", errors="ignore") == content:
+                    # Ensure executable bit is set
+                    try:
+                        os.chmod(path, 0o700)
+                    except Exception:
+                        pass
+                    return str(path)
+            except Exception:
+                pass
+        path.write_text(content, encoding="utf-8")
+        try:
+            os.chmod(path, 0o700)
+        except Exception:
+            pass
+    except Exception:
+        # Fallback to a per-process path
+        path = Path(tempfile.gettempdir()) / f"ptrlib-ssh-askpass-{os.getpid()}.sh"
+        path.write_text(content, encoding="utf-8")
+        with contextlib.suppress(Exception):
+            os.chmod(path, 0o700)
+
+    return str(path)
+
+
 def _ensure_ssh_option(options: list[str], key: str, value: str) -> None:
     """Append -oKey=Value if the key isn't already present."""
     prefix = f"-o{key}="
@@ -194,6 +241,18 @@ def SSH(host: str,
             env["SSH_ASKPASS_REQUIRE"] = "force"
             # Some builds still require DISPLAY to be non-empty to invoke askpass.
             env.setdefault("DISPLAY", "1")
+        else:
+            # On POSIX, prefer SSH_ASKPASS to avoid fragile prompt parsing and to
+            # work even when OpenSSH chooses not to echo prompts to the PTY.
+            askpass = _ensure_posix_askpass_sh()
+            env = os.environ.copy()
+            env["PTRLIB_SSH_PASSWORD"] = password
+            # Ensure no controlling TTY to force askpass on older OpenSSH builds
+            env["PTRLIB_START_NEW_SESSION"] = "1"
+            env["SSH_ASKPASS"] = askpass
+            env["SSH_ASKPASS_REQUIRE"] = "force"
+            # Historically OpenSSH only invoked askpass when DISPLAY was set; keep it.
+            env.setdefault("DISPLAY", "1")
 
     argv = _build_ssh_argv(
         host,
@@ -205,34 +264,39 @@ def SSH(host: str,
         command=command,
     )
 
-    # On POSIX, use a PTY for interactive sessions and/or password prompts.
-    need_tty = (not command) or (password is not None)
+    # On POSIX, use a PTY for interactive shells when no programmatic password is used.
+    # If a password is provided, prefer pipes (no controlling TTY) so SSH_ASKPASS is
+    # reliably used even on older OpenSSH that ignore SSH_ASKPASS_REQUIRE=force.
     if os.name != 'nt':
-        sess = Process(argv, use_tty=need_tty, env=env)
+        use_local_pty = (password is None) and (not command)
+        sess = Process(argv, use_tty=use_local_pty, env=env)
     else:
         sess = Process(argv, env=env)
 
     sess.prompt = ""
 
     if password is not None and os.name != 'nt':
-        # Handle common prompts robustly (case-insensitive, includes hostkey prompt).
-        hostkey_re = re.compile(br"(?i)are you sure you want to continue connecting")
-        password_re = re.compile(br"(?i)password:\s*")
-        denied_re = re.compile(br"(?i)permission denied")
-
-        # Best-effort prompt dance: accept host key prompt, then send password.
-        # If nothing matches within a short time, continue and let the user drive.
-        for _ in range(3):
-            with contextlib.suppress(Exception):
-                m = sess.recvregex([hostkey_re, password_re, denied_re], timeout=10)
-                if m.re is hostkey_re:
-                    sess.sendline("yes")
-                    continue
-                if m.re is password_re:
-                    sess.sendline(password)
-                    break
-                if m.re is denied_re:
-                    raise PermissionError("SSH authentication failed (Permission denied)")
+        # If we're using askpass, prompts won't appear on the TTY. Skip prompt parsing.
+        using_askpass = (env is not None) and (env.get("SSH_ASKPASS_REQUIRE") == "force")
+        if not using_askpass:
+            # Handle common prompts robustly (case-insensitive, includes hostkey prompt).
+            hostkey_re = re.compile(br"(?i)are you sure you want to continue connecting")
+            password_re = re.compile(br"(?i)password:\s*")
+            denied_re = re.compile(br"(?i)permission denied")
+
+            # Best-effort prompt dance: accept host key prompt, then send password.
+            # If nothing matches within a short time, continue and let the user drive.
+            for _ in range(3):
+                with contextlib.suppress(Exception):
+                    m = sess.recvregex([hostkey_re, password_re, denied_re], timeout=10)
+                    if m.re is hostkey_re:
+                        sess.sendline("yes")
+                        continue
+                    if m.re is password_re:
+                        sess.sendline(password)
+                        break
+                    if m.re is denied_re:
+                        raise PermissionError("SSH authentication failed (Permission denied)")
 
     # Windows: if we forced BatchMode (password=None), fail fast on auth errors.
     if os.name == 'nt' and password is None:

ptrlib/connection/tube.py

@@ -83,7 +83,25 @@ def __init__(self,
         else:
             path = Path(pcap)
 
-        self._pcap: PcapFile = PcapFile(path)
+        # pcap logging must NEVER crash the application; guard constructor
+        try:
+            self._pcap: PcapFile = PcapFile(path)
+        except Exception:
+            # Provide a dummy object with no-op methods
+            class _NoopPcap:
+                def __init__(self):
+                    self.udp = False
+                    self.remote = "0.0.0.0"
+                    self.remote_port = 0
+                def send(self, *_a, **_k):
+                    pass
+                def recv(self, *_a, **_k):
+                    pass
+                def close(self, *_a, **_k):
+                    pass
+                def close_send(self, *_a, **_k):
+                    pass
+            self._pcap = _NoopPcap()  # type: ignore[assignment]
 
         # Defer `after` / `sendafter` / `sendlineafter`
         self._defer_depth: int = 0

ptrlib/connection/unixproc.py

@@ -449,6 +449,16 @@ def setup_tty():
         else:
             popen_args = self._args
 
+        # Determine if we should detach from the current session (no controlling TTY).
+        # SSH with SSH_ASKPASS needs this on older OpenSSH builds where a controlling
+        # TTY (accessible via /dev/tty) suppresses askpass. The SSH wrapper sets
+        # PTRLIB_START_NEW_SESSION=1 when a password is provided on POSIX.
+        start_new_sess = False
+        try:
+            start_new_sess = (not use_tty) and (str(self._env.get("PTRLIB_START_NEW_SESSION", "0")) == "1")
+        except Exception:
+            start_new_sess = False
+
         # pylint: disable-next=subprocess-popen-preexec-fn
         self._proc = subprocess.Popen(
             popen_args,
@@ -461,6 +471,7 @@ def setup_tty():
             stderr=stderr,
             close_fds=True,
             bufsize=0,
+            start_new_session=start_new_sess,
         )
 
         with contextlib.suppress(OSError):

ptrlib/filestruct/pcap/emulator.py

@@ -397,10 +397,15 @@ def advance(self, seconds: float) -> None:
 
     def flush(self) -> None:
         """Flush the file handle and synchronize the file system.
+        
+        Best-effort: errors are suppressed so pcap logging never breaks
+        application logic (e.g., when the FD has been invalidated by the
+        environment/tests).
         """
         if self._fh is None:
             return
-        self._fh.flush()
+        with contextlib.suppress(Exception):
+            self._fh.flush()
 
     def __enter__(self) -> "PcapFile":
         return self
@@ -520,14 +525,28 @@ def _emit_udp(self,
         self._write(ts, frame)
 
     def _write(self, ts: float, frame: bytes) -> None:
-        assert self._fh is not None, "Pcap file already closed"
-        if ts <= self._last_ts:
-            ts = self._last_ts + 0.000001
-        self._last_ts = ts
-        hdr = _pack_pcap_packet_header(ts, len(frame), len(frame))
-        self._fh.write(hdr)
-        self._fh.write(frame)
-        self._fh.flush()
+        """Best-effort writer; swallow I/O errors to avoid affecting callers.
+        
+        In some environments, file descriptors may get closed or invalidated
+        unexpectedly (e.g., heavy mocking, embedded REPLs). Any error during
+        pcap emission is suppressed and the pcap stream is disabled.
+        """
+        fh = self._fh
+        if fh is None:
+            return
+        try:
+            if ts <= self._last_ts:
+                ts = self._last_ts + 0.000001
+            self._last_ts = ts
+            hdr = _pack_pcap_packet_header(ts, len(frame), len(frame))
+            fh.write(hdr)
+            fh.write(frame)
+            fh.flush()
+        except Exception:
+            # Disable further logging on any failure
+            with contextlib.suppress(Exception):
+                fh.close()
+            self._fh = None
 
     def _alloc_ip_id(self) -> int:
         self._ip_id = (self._ip_id + 1) & 0xFFFF