Apache Airflow Google provider: SSH host key verification disabled in ComputeEngineSSHHook (paramiko AutoAddPolicy default)

@@ -112,6 +112,13 @@ class ComputeEngineSSHHook(SSHHook):
     :param impersonation_chain: Optional. The service account email to impersonate using short-term
         credentials. The provided service account must grant the originating account
         the Service Account Token Creator IAM role and have the sufficient rights to perform the request
+    :param host_key_policy: Policy used by the underlying ``paramiko.SSHClient`` for unknown host keys.
+        Accepts the string aliases ``"auto_add"`` (default, historical behaviour — adds unknown keys
+        without prompting), ``"reject"`` (refuse to connect to hosts whose key is not in ``known_hosts``)
+        and ``"warning"`` (log a warning but still connect). Alternatively, pass an instance of
+        ``paramiko.MissingHostKeyPolicy`` (or a subclass) to plug in a custom policy — for example one
+        that loads pinned host keys from GCE guest attributes. Any other value raises ``ValueError``
+        when the connection is opened.
     """
 
     conn_name_attr = "gcp_conn_id"
@@ -141,6 +148,7 @@ def __init__(
         cmd_timeout: int | ArgNotSet = NOTSET,
         max_retries: int = 10,
         impersonation_chain: str | None = None,
+        host_key_policy: str | paramiko.MissingHostKeyPolicy = "auto_add",
         **kwargs,
     ) -> None:
         # Ignore original constructor
@@ -158,8 +166,47 @@ def __init__(
         self.cmd_timeout = cmd_timeout
         self.max_retries = max_retries
         self.impersonation_chain = impersonation_chain
+        self.host_key_policy = host_key_policy
         self._conn: Any | None = None
 
+    def _resolve_host_key_policy(self) -> paramiko.MissingHostKeyPolicy:
+        """
+        Resolve ``self.host_key_policy`` to a concrete paramiko policy instance.
+
+        Accepts:
+
+        - the string aliases ``"auto_add"``, ``"reject"`` or ``"warning"`` —
+          mapped to the matching ``paramiko`` policy class;
+        - an instance of ``paramiko.MissingHostKeyPolicy`` — used as-is, so
+          callers can plug in a custom policy (e.g. one that loads pinned
+          host keys from GCE guest attributes).
+
+        Any other value raises :class:`ValueError`.
+
+        The default value (``"auto_add"``) preserves the historical behaviour
+        of this hook. Callers that want the remote SSH server authenticated
+        before the session opens should pass ``"reject"`` together with a
+        populated ``known_hosts`` file, or supply a custom policy that
+        looks the remote host's key up from an out-of-band source.
+        """
+        if not isinstance(self.host_key_policy, str):
+            # Trust the caller: an explicit paramiko.MissingHostKeyPolicy
+            # instance, or a subclass instance with custom behaviour.
+            return self.host_key_policy
+        builtins = {
+            "auto_add": paramiko.AutoAddPolicy,
+            "reject": paramiko.RejectPolicy,
+            "warning": paramiko.WarningPolicy,
+        }
+        try:
+            return builtins[self.host_key_policy]()
+        except KeyError:
+            raise ValueError(
+                f"Unknown host_key_policy {self.host_key_policy!r}. "
+                "Expected one of 'auto_add', 'reject', 'warning', "
+                "or an instance of paramiko.MissingHostKeyPolicy."
+            ) from None
+
     @cached_property
     def _oslogin_hook(self) -> OSLoginHook:
         return OSLoginHook(gcp_conn_id=self.gcp_conn_id)
@@ -310,9 +357,12 @@ def _connect_to_instance(self, user, hostname, pkey, proxy_command) -> paramiko.
         for time_to_wait in range(max_time_to_wait + 1):
             try:
                 client = _GCloudAuthorizedSSHClient(self._compute_hook)
-                # Default is RejectPolicy
-                # No known host checking since we are not storing privatekey
-                client.set_missing_host_key_policy(paramiko.AutoAddPolicy())  # nosec B507
+                # Apply the policy configured via the `host_key_policy` constructor
+                # argument; default is `"auto_add"` (paramiko.AutoAddPolicy), which
+                # preserves the historical behaviour of this hook. Callers that need
+                # the remote host authenticated should pass `"reject"` with a
+                # populated known_hosts file or a custom MissingHostKeyPolicy.
+                client.set_missing_host_key_policy(self._resolve_host_key_policy())  # nosec B507
                 client.connect(
                     hostname=hostname,
                     username=user,

@@ -21,6 +21,7 @@
 from unittest import mock
 
 import httplib2
+import paramiko
 import pytest
 from googleapiclient.errors import HttpError
 from paramiko.ssh_exception import SSHException
@@ -560,3 +561,39 @@ def test__authorize_compute_engine_instance_metadata(
             mock_set_instance_metadata.call_args.kwargs["metadata"]["items"].sort(key=lambda x: x["key"])
             expected_metadata["items"].sort(key=lambda x: x["key"])
             assert mock_set_instance_metadata.call_args.kwargs["metadata"] == expected_metadata
+
+
+class TestHostKeyPolicyResolution:
+    """Tests for the ``host_key_policy`` constructor argument."""
+
+    def test_default_is_auto_add(self):
+        hook = ComputeEngineSSHHook()
+
+        assert hook.host_key_policy == "auto_add"
+        assert isinstance(hook._resolve_host_key_policy(), paramiko.AutoAddPolicy)
+
+    def test_string_aliases(self):
+        assert isinstance(
+            ComputeEngineSSHHook(host_key_policy="auto_add")._resolve_host_key_policy(),
+            paramiko.AutoAddPolicy,
+        )
+        assert isinstance(
+            ComputeEngineSSHHook(host_key_policy="reject")._resolve_host_key_policy(),
+            paramiko.RejectPolicy,
+        )
+        assert isinstance(
+            ComputeEngineSSHHook(host_key_policy="warning")._resolve_host_key_policy(),
+            paramiko.WarningPolicy,
+        )
+
+    def test_custom_policy_instance_is_returned_unchanged(self):
+        custom_policy = paramiko.RejectPolicy()
+        hook = ComputeEngineSSHHook(host_key_policy=custom_policy)
+
+        assert hook._resolve_host_key_policy() is custom_policy
+
+    def test_unknown_string_raises_value_error(self):
+        hook = ComputeEngineSSHHook(host_key_policy="strict")
+
+        with pytest.raises(ValueError, match=r"Unknown host_key_policy 'strict'"):
+            hook._resolve_host_key_policy()