Harden GitHook SSH command building and URL construction (#64756)

- Quote user-controlled values in _build_ssh_command with shlex.quote
- Validate strict_host_key_checking against allowlist of valid SSH values
- URL-encode username and auth token when embedding in repo URLs
- Fix logic bug: `not X or not Y` → `not X and not Y` for git@/https:// check
- Limit str.replace to first occurrence

Author: potiuk

providers/git/src/airflow/providers/git/hooks/git.py

@@ -25,6 +25,7 @@
 import stat
 import tempfile
 from typing import Any
+from urllib.parse import quote as urlquote
 
 from airflow.providers.common.compat.sdk import AirflowException, BaseHook
 
@@ -109,25 +110,32 @@ def __init__(
             raise AirflowException("Both 'key_file' and 'private_key' cannot be provided at the same time")
         self._process_git_auth_url()
 
+    _VALID_STRICT_HOST_KEY_CHECKING = frozenset({"yes", "no", "accept-new", "off", "ask"})
+
     def _build_ssh_command(self, key_path: str | None = None) -> str:
         parts = ["ssh"]
 
         if key_path:
-            parts.append(f"-i {key_path}")
+            parts.append(f"-i {shlex.quote(key_path)}")
             parts.append("-o IdentitiesOnly=yes")
 
+        if self.strict_host_key_checking not in self._VALID_STRICT_HOST_KEY_CHECKING:
+            raise ValueError(
+                f"Invalid strict_host_key_checking value: {self.strict_host_key_checking!r}. "
+                f"Must be one of {sorted(self._VALID_STRICT_HOST_KEY_CHECKING)}"
+            )
         parts.append(f"-o StrictHostKeyChecking={self.strict_host_key_checking}")
 
         if self.known_hosts_file:
-            parts.append(f"-o UserKnownHostsFile={self.known_hosts_file}")
+            parts.append(f"-o UserKnownHostsFile={shlex.quote(self.known_hosts_file)}")
         elif self.strict_host_key_checking == "no":
             parts.append("-o UserKnownHostsFile=/dev/null")
 
         if self.ssh_config_file:
-            parts.append(f"-F {self.ssh_config_file}")
+            parts.append(f"-F {shlex.quote(self.ssh_config_file)}")
 
         if self.host_proxy_cmd:
-            parts.append(f'-o ProxyCommand="{self.host_proxy_cmd}"')
+            parts.append(f"-o ProxyCommand={shlex.quote(self.host_proxy_cmd)}")
 
         if self.ssh_port:
             parts.append(f"-p {self.ssh_port}")
@@ -138,13 +146,17 @@ def _process_git_auth_url(self):
         if not isinstance(self.repo_url, str):
             return
         if self.auth_token and self.repo_url.startswith("https://"):
-            self.repo_url = self.repo_url.replace("https://", f"https://{self.user_name}:{self.auth_token}@")
+            encoded_user = urlquote(self.user_name, safe="")
+            encoded_token = urlquote(self.auth_token, safe="")
+            self.repo_url = self.repo_url.replace("https://", f"https://{encoded_user}:{encoded_token}@", 1)
         elif self.auth_token and self.repo_url.startswith("http://"):
-            self.repo_url = self.repo_url.replace("http://", f"http://{self.user_name}:{self.auth_token}@")
+            encoded_user = urlquote(self.user_name, safe="")
+            encoded_token = urlquote(self.auth_token, safe="")
+            self.repo_url = self.repo_url.replace("http://", f"http://{encoded_user}:{encoded_token}@", 1)
         elif self.repo_url.startswith("http://"):
             # if no auth token, use the repo url as is
-            self.repo_url = self.repo_url
-        elif not self.repo_url.startswith("git@") or not self.repo_url.startswith("https://"):
+            pass
+        elif not self.repo_url.startswith("git@") and not self.repo_url.startswith("https://"):
             self.repo_url = os.path.expanduser(self.repo_url)
 
     def set_git_env(self, key: str | None = None) -> None:

providers/git/tests/unit/git/hooks/test_git.py

@@ -249,7 +249,7 @@ def test_proxy_command(self, create_connection_without_db):
         hook = GitHook(git_conn_id="git_with_proxy")
         with hook.configure_hook_env():
             cmd = hook.env["GIT_SSH_COMMAND"]
-            assert 'ProxyCommand="ssh -W %h:%p bastion.example.com"' in cmd
+            assert "ProxyCommand='ssh -W %h:%p bastion.example.com'" in cmd
 
     def test_known_hosts_file(self, create_connection_without_db):
         create_connection_without_db(