Skip to content
Open
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions pysqa/base/remote.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,9 +93,12 @@ def __init__(
)
self._ssh_host = config["ssh_host"]
self._ssh_username = config["ssh_username"]
self._ssh_known_hosts = os.path.abspath(
os.path.expanduser(config["known_hosts"])
)
if "known_hosts" in config:
self._ssh_known_hosts = os.path.abspath(
os.path.expanduser(config["known_hosts"])
)
else:
self._ssh_known_hosts = None
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue

Harden known_hosts config handling; avoid empty-string bug and add explicit opt-in for auto-accepting unknown hosts

  • Using "key in config" treats an empty string as a valid path and resolves it to the current working directory, causing paramiko to try to read a non-existent file.
  • Defaulting to AutoAddPolicy downstream without explicit opt-in is a security regression; prefer system host keys by default and make auto-accept a config flag.

Proposed fix:

  • Treat empty or missing known_hosts as None.
  • Validate file existence and warn if missing.
  • Introduce ssh_auto_add_unknown_hosts (default False).

Apply this diff:

-        if "known_hosts" in config:
-            self._ssh_known_hosts = os.path.abspath(
-                os.path.expanduser(config["known_hosts"])
-            )
-        else:
-            self._ssh_known_hosts = None
+        # Known hosts: allow None, but prefer system host keys when unspecified.
+        kh = config.get("known_hosts")
+        if kh:
+            path = os.path.abspath(os.path.expanduser(kh))
+            if os.path.exists(path):
+                self._ssh_known_hosts = path
+            else:
+                warnings.warn(
+                    f"known_hosts file not found at '{path}'. Falling back to system host keys.",
+                    stacklevel=2,
+                )
+                self._ssh_known_hosts = None
+        else:
+            self._ssh_known_hosts = None
+        # Opt-in flag to auto-accept unknown host keys (unsafe by default).
+        self._ssh_auto_add_unknown_hosts = bool(
+            config.get("ssh_auto_add_unknown_hosts", False)
+        )
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
if "known_hosts" in config:
self._ssh_known_hosts = os.path.abspath(
os.path.expanduser(config["known_hosts"])
)
else:
self._ssh_known_hosts = None
# Known hosts: allow None, but prefer system host keys when unspecified.
kh = config.get("known_hosts")
if kh:
path = os.path.abspath(os.path.expanduser(kh))
if os.path.exists(path):
self._ssh_known_hosts = path
else:
warnings.warn(
f"known_hosts file not found at '{path}'. Falling back to system host keys.",
stacklevel=2,
)
self._ssh_known_hosts = None
else:
self._ssh_known_hosts = None
# Opt-in flag to auto-accept unknown host keys (unsafe by default).
self._ssh_auto_add_unknown_hosts = bool(
config.get("ssh_auto_add_unknown_hosts", False)
)
🤖 Prompt for AI Agents
In pysqa/base/remote.py around lines 96 to 101, the current known_hosts handling
treats an empty string as a valid path and implicitly enables AutoAddPolicy
downstream; change it to treat missing or empty known_hosts as None, add a new
boolean config key ssh_auto_add_unknown_hosts defaulting to False, and only
enable auto-accept when that flag is true. Specifically: read
config.get("known_hosts") and if value is falsy (None or empty string) set
self._ssh_known_hosts = None; if a non-empty path is provided expand and abspath
it and then check os.path.exists(path) — if it does not exist set
self._ssh_known_hosts = None and emit a warning via the module logger; add
self._ssh_auto_add_unknown_hosts = bool(config.get("ssh_auto_add_unknown_hosts",
False)) so callers can decide whether to set AutoAddPolicy; ensure default
behavior prefers loading system host keys when known_hosts is None and only use
AutoAddPolicy when the new flag is true.

self._ssh_ask_for_password: Union[bool, str] = False
self._ssh_key = (
os.path.abspath(os.path.expanduser(config["ssh_key"]))
Expand Down Expand Up @@ -346,7 +349,10 @@ def _open_ssh_connection(self) -> paramiko.SSHClient:
paramiko.SSHClient: The SSH connection object.
"""
ssh = paramiko.SSHClient()
ssh.load_host_keys(self._ssh_known_hosts)
if self._ssh_known_hosts is not None:
ssh.load_host_keys(self._ssh_known_hosts)
else:
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Avoid silently trusting unknown hosts; load system host keys and gate AutoAddPolicy behind explicit opt-in

When _ssh_known_hosts is None, switching to AutoAddPolicy unconditionally trusts unknown hosts, enabling MITM risk. Prefer:

  • Always load system host keys.
  • Optionally load a user-provided known_hosts file.
  • Use RejectPolicy by default; only AutoAdd when explicitly configured.

Apply this diff:

-        if self._ssh_known_hosts is not None:
-            ssh.load_host_keys(self._ssh_known_hosts)
-        else:
-            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        # Load system-level keys first; optionally supplement with a custom file.
+        ssh.load_system_host_keys()
+        if self._ssh_known_hosts is not None:
+            try:
+                ssh.load_host_keys(self._ssh_known_hosts)
+            except FileNotFoundError:
+                warnings.warn(
+                    f"known_hosts file not found at '{self._ssh_known_hosts}'. Using system host keys only.",
+                    stacklevel=2,
+                )
+        # Decide how to handle unknown hosts
+        if getattr(self, "_ssh_auto_add_unknown_hosts", False):
+            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        else:
+            ssh.set_missing_host_key_policy(paramiko.RejectPolicy())

Committable suggestion skipped: line range outside the PR's diff.

🤖 Prompt for AI Agents
In pysqa/base/remote.py around lines 352-355, the code currently sets
AutoAddPolicy when self._ssh_known_hosts is None which silently trusts unknown
hosts; change it to always call ssh.load_system_host_keys(), then if
self._ssh_known_hosts is set load that file too, and set the missing host key
policy to paramiko.RejectPolicy() by default; introduce or use an explicit
boolean config flag (e.g. self._allow_auto_add_hosts defaulting to False) so
AutoAddPolicy() is only applied when that flag is True (otherwise use
RejectPolicy()).

if (
self._ssh_key is not None
and self._ssh_key_passphrase is not None
Expand Down
Loading