Add GitHub App authentication for git DAG bundles

providers/git/pyproject.toml

@@ -64,6 +64,13 @@ dependencies = [
     "GitPython>=3.1.44",
 ]
 
+# The optional dependencies should be modified in place in the generated file
+# Any change in the dependencies is preserved when the file is regenerated
+[project.optional-dependencies]
+github = [
+    "PyGithub>=2.1.1,<3",
+]
+
 [dependency-groups]
 dev = [
     "apache-airflow",

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

@@ -24,6 +24,7 @@
 import shlex
 import stat
 import tempfile
+from datetime import datetime, timedelta, timezone
 from typing import Any
 from urllib.parse import quote as urlquote
 
@@ -49,6 +50,10 @@ class GitHook(BaseHook):
     * ``ssh_config_file`` — path to a custom SSH config file.
     * ``host_proxy_cmd`` — SSH ProxyCommand string (e.g. for bastion/jump hosts).
     * ``ssh_port`` — non-default SSH port.
+    * ``github_app_id`` — GitHub App ID used for GitHub App authentication. Requires the GitHub App
+      private key to be provided as a PEM-encoded key via either ``private_key`` (inline) or
+      ``key_file`` (path to key file).
+    * ``github_installation_id`` — GitHub App installation ID used for GitHub App authentication.
     """
 
     conn_name_attr = "git_conn_id"
@@ -76,6 +81,8 @@ def get_ui_field_behaviour(cls) -> dict[str, Any]:
                         "ssh_config_file": "",
                         "host_proxy_cmd": "",
                         "ssh_port": "",
+                        "github_app_id": "",
+                        "github_installation_id": "",
                     }
                 )
             },
@@ -104,10 +111,36 @@ def __init__(
         self.host_proxy_cmd = extra.get("host_proxy_cmd")
         self.ssh_port: int | None = int(extra["ssh_port"]) if extra.get("ssh_port") else None
 
+        # GitHub App Auth Options
+        self.github_app_id = extra.get("github_app_id")
+        self.github_installation_id = extra.get("github_installation_id")
+        self.github_app_private_key: str | None = None
+        self.github_app_token_exp: datetime | None = None
+
         self.env: dict[str, str] = {}
 
         if self.key_file and self.private_key:
             raise AirflowException("Both 'key_file' and 'private_key' cannot be provided at the same time")
+        if (self.github_app_id is not None and self.github_installation_id is None) or (
+            self.github_app_id is None and self.github_installation_id is not None
+        ):
+            raise ValueError(
+                "Both 'github_app_id' and 'github_installation_id' must be provided to use GitHub App Authentication"
+            )
+        if self.github_app_id is not None and self.github_installation_id is not None:
+            if self.key_file and not self.private_key:
+                with open(self.key_file, encoding="utf-8") as key_file:
+                    self.private_key = key_file.read()
+            if not (self.repo_url or "").startswith(("https://", "http://")):
+                raise ValueError(
+                    f"GitHub App authentication requires an HTTPS repository URL, but got: {self.repo_url!r}"
+                )
+            # Store the PEM separately so configure_hook_env() does not treat it as an SSH key.
+            # Keep `private_key` populated for callers/tests that expect it to be available,
+            # but also keep a dedicated attribute so configure_hook_env() can avoid
+            # treating the GitHub App PEM as an SSH key.
+            self.github_app_private_key = self.private_key
+            self.auth_token = ""
         self._process_git_auth_url()
 
     _VALID_STRICT_HOST_KEY_CHECKING = frozenset({"yes", "no", "accept-new", "off", "ask"})
@@ -142,6 +175,82 @@ def _build_ssh_command(self, key_path: str | None = None) -> str:
 
         return " ".join(parts)
 
+    def _get_github_app_token(self):
+        try:
+            from github import Auth, GithubIntegration
+        except ImportError as exc:
+            raise ImportError(
+                "The PyGithub library is required for GitHub App authentication. Please install it with 'pip install apache-airflow-providers-git[github]'"
+            ) from exc
+
+        auth = Auth.AppAuth(self.github_app_id, self.github_app_private_key)
+        integration = GithubIntegration(auth=auth)
+        access_token = integration.get_access_token(installation_id=self.github_installation_id)
+        github_app_token_exp = access_token.expires_at
+        log.info(
+            "Successfully obtained GitHub App installation access token (expires at: %s)",
+            github_app_token_exp,
+        )
+
+        return "x-access-token", access_token.token, github_app_token_exp
+
+    def _ensure_github_app_token(self) -> None:
+        if self.github_app_id is None or self.github_installation_id is None:
+            return
+
+        TOKEN_REFRESH_BUFFER = timedelta(minutes=5)
+        if (
+            self.github_app_token_exp is None
+            or self.github_app_token_exp < datetime.now(timezone.utc) + TOKEN_REFRESH_BUFFER
+        ):
+            log.info(
+                "GitHub App token is missing or near expiry (expires at: %s). Refreshing token.",
+                self.github_app_token_exp,
+            )
+            self.user_name, self.auth_token, self.github_app_token_exp = self._get_github_app_token()
+
+    @contextlib.contextmanager
+    def _github_app_askpass_env(self):
+        if not self.auth_token:
+            yield
+            return
+
+        token = shlex.quote(self.auth_token)
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".sh", delete=True) as askpass_script:
+            askpass_script.write(
+                "#!/bin/sh\n"
+                'case "$1" in\n'
+                "  *Username*) echo x-access-token;;\n"
+                f"  *Password*) echo {token};;\n"
+                f"  *) echo {token};;\n"
+                "esac\n"
+            )
+            askpass_script.flush()
+            os.chmod(askpass_script.name, stat.S_IRWXU)
+
+            old_askpass = os.environ.get("GIT_ASKPASS")
+            old_terminal_prompt = os.environ.get("GIT_TERMINAL_PROMPT")
+            try:
+                os.environ["GIT_ASKPASS"] = askpass_script.name
+                os.environ["GIT_TERMINAL_PROMPT"] = "0"
+                self.env["GIT_ASKPASS"] = askpass_script.name
+                self.env["GIT_TERMINAL_PROMPT"] = "0"
+                yield
+            finally:
+                if old_askpass is None:
+                    self.env.pop("GIT_ASKPASS", None)
+                    os.environ.pop("GIT_ASKPASS", None)
+                else:
+                    self.env["GIT_ASKPASS"] = old_askpass
+                    os.environ["GIT_ASKPASS"] = old_askpass
+
+                if old_terminal_prompt is None:
+                    self.env.pop("GIT_TERMINAL_PROMPT", None)
+                    os.environ.pop("GIT_TERMINAL_PROMPT", None)
+                else:
+                    self.env["GIT_TERMINAL_PROMPT"] = old_terminal_prompt
+                    os.environ["GIT_TERMINAL_PROMPT"] = old_terminal_prompt
+
     def _process_git_auth_url(self):
         if not isinstance(self.repo_url, str):
             return
@@ -199,7 +308,16 @@ def _passphrase_askpass_env(self):
 
     @contextlib.contextmanager
     def configure_hook_env(self):
-        if self.private_key:
+        self._ensure_github_app_token()
+
+        if self.github_app_id is not None and self.github_installation_id is not None:
+            with self._github_app_askpass_env():
+                yield
+            return
+
+        # If a GitHub App PEM is present, it should not be treated as an SSH key
+        # for configuring `GIT_SSH_COMMAND`.
+        if self.private_key and not self.github_app_private_key:
             with tempfile.NamedTemporaryFile(mode="w", delete=True) as tmp_keyfile:
                 tmp_keyfile.write(self.private_key)
                 tmp_keyfile.flush()