117 lines
4.9 KiB
Python
117 lines
4.9 KiB
Python
"""Git-Connector (→ Konzept 7.5): lesender Zugriff auf beliebige Git-Server.
|
|
|
|
Sicherheit (→ Konzept 7.3):
|
|
- nur https:// zu erlaubten Hosts; file:// nur für lokale Test-Repositories
|
|
- keine Zugangsdaten in der URL; das Token kommt aus dem Secret Store und geht
|
|
über die Umgebung als HTTP-Header an git, nie über die Kommandozeile
|
|
- git wird ohne Shell aufgerufen, ohne System- und Nutzerkonfiguration,
|
|
ohne Rückfragen und mit Zeitlimit
|
|
"""
|
|
|
|
import base64
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
from app.core.config import Settings
|
|
from app.core.errors import Invalid
|
|
from app.data_sources.connectors import TestResult
|
|
|
|
BRANCH = re.compile(r"^(?!.*\.\.)(?![/-])(?!.*//)[A-Za-z0-9._/-]{1,200}(?<!/)(?<!\.lock)$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GitConfig:
|
|
url: str
|
|
branch: str
|
|
|
|
|
|
USERNAME = re.compile(r"^[^\s:]{1,100}$")
|
|
|
|
|
|
def auth_header(token: str, username: str = "") -> str:
|
|
"""Anmeldung am Git-Server: ohne Nutzername Bearer (Bitbucket Data Center, HTTP Access Token), mit Nutzername
|
|
Basic mit dem Token als Passwort (Gitea, Forgejo, GitLab, GitHub, klassische HTTPS-Server)."""
|
|
if username:
|
|
return "Basic " + base64.b64encode(f"{username}:{token}".encode()).decode()
|
|
return f"Bearer {token}"
|
|
|
|
|
|
def check_username(username: str) -> str:
|
|
username = username.strip()
|
|
if username and not USERNAME.match(username):
|
|
raise Invalid("Nutzername ohne Leerzeichen und Doppelpunkt, höchstens 100 Zeichen.")
|
|
return username
|
|
|
|
|
|
def check_config(url: str, branch: str, settings: Settings) -> GitConfig:
|
|
url, branch = url.strip(), (branch.strip() or "main")
|
|
parts = urlsplit(url)
|
|
if parts.scheme not in ("https", "file"):
|
|
raise Invalid("Nur https:// ist erlaubt (SSH folgt später).")
|
|
if parts.username or parts.password:
|
|
raise Invalid("Keine Zugangsdaten in der URL. Das Token gehört in den Secret Store.")
|
|
if parts.scheme == "https":
|
|
host = (parts.hostname or "").lower()
|
|
allowed = settings.allowed_git_hosts
|
|
if host not in allowed and parts.netloc.lower() not in allowed:
|
|
raise Invalid(f"Host {host or '(leer)'} ist nicht freigegeben (AIW_GIT_ALLOWED_HOSTS).")
|
|
else:
|
|
if not settings.git_allow_local:
|
|
raise Invalid("Lokale Repositories (file://) sind nicht freigegeben.")
|
|
if not parts.path.startswith("/"):
|
|
raise Invalid("file:// braucht einen absoluten Pfad.")
|
|
if not BRANCH.match(branch):
|
|
raise Invalid("Ungültiger Branch-Name.")
|
|
return GitConfig(url=url, branch=branch)
|
|
|
|
|
|
class GitConnector:
|
|
def __init__(self, config: GitConfig, settings: Settings, token: str | None = None, username: str = ""):
|
|
self.config, self.settings, self.token, self.username = config, settings, token, username
|
|
|
|
def environment(self) -> dict[str, str]:
|
|
env = {
|
|
"PATH": os.environ.get("PATH", "/usr/bin:/bin"),
|
|
"HOME": os.devnull,
|
|
"GIT_TERMINAL_PROMPT": "0",
|
|
"GIT_CONFIG_NOSYSTEM": "1",
|
|
"GIT_CONFIG_GLOBAL": os.devnull,
|
|
"GIT_ALLOW_PROTOCOL": "https:file" if self.settings.git_allow_local else "https",
|
|
}
|
|
if self.token:
|
|
# Als Header, nie in der URL; Bearer oder Basic je nach Quelle (auth_header)
|
|
env |= {"GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "http.extraHeader",
|
|
"GIT_CONFIG_VALUE_0": f"Authorization: {auth_header(self.token, self.username)}"}
|
|
return env
|
|
|
|
def git(self, *args: str) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(["git", *args], env=self.environment(), capture_output=True, text=True,
|
|
timeout=self.settings.git_timeout_seconds, check=False)
|
|
|
|
def test_connection(self) -> TestResult:
|
|
ref = f"refs/heads/{self.config.branch}"
|
|
try:
|
|
result = self.git("ls-remote", "--heads", "--", self.config.url, ref)
|
|
except subprocess.TimeoutExpired:
|
|
return TestResult(False, f"Keine Antwort nach {self.settings.git_timeout_seconds} Sekunden.")
|
|
except OSError as e:
|
|
return TestResult(False, f"git konnte nicht gestartet werden: {e.strerror}")
|
|
if result.returncode != 0:
|
|
return TestResult(False, self.clean(result.stderr) or f"git beendet mit Code {result.returncode}.")
|
|
if ref not in result.stdout:
|
|
return TestResult(False, f"Branch {self.config.branch} nicht gefunden.")
|
|
return TestResult(True, f"Erreichbar, Branch {self.config.branch} vorhanden.")
|
|
|
|
def discover_capabilities(self) -> list[dict[str, Any]]:
|
|
return []
|
|
|
|
def clean(self, text: str) -> str:
|
|
"""Fehlermeldung kürzen; ein Token darf nie in UI oder Audit landen."""
|
|
text = text.strip().splitlines()[-1] if text.strip() else ""
|
|
if self.token:
|
|
text = text.replace(self.token, "***")
|
|
return text[:300]
|