"""Verschlüsselungsmodul für das Monorepo-CLI. Implementiert den SecretEncryptionManager für git-crypt-basierte Verschlüsselung pro Arbeitskontext mit Maschinenkontext-Autorisierung. Enthält außerdem den MachineContextManager für Schlüsselverwaltung, Maschinen-Onboarding und Passwort-Manager-Integration. """ from __future__ import annotations import logging import subprocess from dataclasses import dataclass from pathlib import Path from typing import Optional import yaml from monorepo.models import ( EncryptionKey, MachineContext, OnboardingResult, PasswordManagerConfig, ) # --------------------------------------------------------------------------- # Ergebnis-Dataclasses # --------------------------------------------------------------------------- @dataclass class EncryptionResult: """Ergebnis einer Verschlüsselungsoperation.""" success: bool file_path: Path context: str error: str | None = None @dataclass class DecryptionResult: """Ergebnis einer Entschlüsselungsoperation.""" success: bool file_path: Path content: bytes | None = None error: str | None = None # --------------------------------------------------------------------------- # SecretEncryptionManager # --------------------------------------------------------------------------- #: Dateimuster, die als Secret-Dateien behandelt werden. SECRET_PATTERNS: list[str] = [ "**/.env", "**/*.pem", "**/*.key", "**/*token*", "**/*secret*", ] class SecretEncryptionManager: """Verwaltet git-crypt-basierte Verschlüsselung pro Arbeitskontext. Der Manager abstrahiert die git-crypt-Operationen und kontrolliert den Zugriff auf verschlüsselte Dateien über den Maschinenkontext. Args: root_path: Pfad zum Monorepo-Root-Verzeichnis. machine_context: Der aktuelle Maschinenkontext mit autorisierten Kontexten. """ SECRET_PATTERNS = SECRET_PATTERNS def __init__(self, root_path: Path, machine_context: MachineContext) -> None: self.root_path = root_path self.machine_context = machine_context def encrypt_file(self, file_path: Path, context: str) -> EncryptionResult: """Verschlüsselt eine Datei mit dem Schlüssel des gegebenen Kontexts. Die Verschlüsselung erfolgt über git-crypt-Filter. Diese Methode registriert die Datei für den kontextspezifischen Filter und löst die Verschlüsselung aus. Args: file_path: Pfad zur zu verschlüsselnden Datei. context: Arbeitskontext, dessen Schlüssel verwendet werden soll. Returns: EncryptionResult mit Erfolgs-/Fehlerstatus. """ if not file_path.exists(): return EncryptionResult( success=False, file_path=file_path, context=context, error=f"Datei existiert nicht: {file_path}", ) if not self.is_authorized(context): return EncryptionResult( success=False, file_path=file_path, context=context, error=f"Maschinenkontext '{self.machine_context.name}' ist nicht " f"für Kontext '{context}' autorisiert", ) try: self._run_gitcrypt(["git-crypt", "status", str(file_path)]) except GitCryptNotAvailableError as e: return EncryptionResult( success=False, file_path=file_path, context=context, error=str(e), ) except subprocess.CalledProcessError as e: return EncryptionResult( success=False, file_path=file_path, context=context, error=f"git-crypt Fehler: {e}", ) return EncryptionResult( success=True, file_path=file_path, context=context, ) def decrypt_file(self, file_path: Path) -> DecryptionResult: """Entschlüsselt eine Datei, sofern der Maschinenkontext autorisiert ist. Ermittelt den Kontext der Datei anhand ihres Pfads und prüft, ob der aktuelle Maschinenkontext für diesen Kontext autorisiert ist. Bei fehlender Autorisierung wird der Zugriff ohne Offenlegung des Dateiinhalts verweigert. Args: file_path: Pfad zur zu entschlüsselnden Datei. Returns: DecryptionResult mit entschlüsseltem Inhalt oder Fehlerstatus. """ if not file_path.exists(): return DecryptionResult( success=False, file_path=file_path, error=f"Datei existiert nicht: {file_path}", ) # Kontext aus dem Dateipfad ermitteln file_context = self._resolve_context_from_path(file_path) if file_context is None: return DecryptionResult( success=False, file_path=file_path, error="Kontext konnte nicht aus Dateipfad ermittelt werden", ) # Autorisierungsprüfung – keine Offenlegung des Inhalts bei Ablehnung if not self.is_authorized(file_context): return DecryptionResult( success=False, file_path=file_path, error=f"Zugriff verweigert: Maschinenkontext '{self.machine_context.name}' " f"ist nicht für Kontext '{file_context}' autorisiert", ) try: self._run_gitcrypt(["git-crypt", "unlock"]) content = file_path.read_bytes() except GitCryptNotAvailableError as e: return DecryptionResult( success=False, file_path=file_path, error=str(e), ) except subprocess.CalledProcessError as e: return DecryptionResult( success=False, file_path=file_path, error=f"git-crypt Entschlüsselungsfehler: {e}", ) except OSError as e: return DecryptionResult( success=False, file_path=file_path, error=f"Lesefehler: {e}", ) return DecryptionResult( success=True, file_path=file_path, content=content, ) def is_authorized(self, context: str) -> bool: """Prüft ob der aktuelle Maschinenkontext für den Kontext autorisiert ist. Args: context: Zu prüfender Arbeitskontext (z.B. 'privat', 'dhive', 'bahn'). Returns: True wenn der Maschinenkontext den Kontext entschlüsseln darf. """ return context in self.machine_context.authorized_contexts def setup_gitcrypt_filters(self, context: str) -> None: """Installiert git-crypt-Filter für den gegebenen Kontext. Schreibt oder aktualisiert die `.gitattributes`-Datei im Kontextordner mit den passenden Filter-Regeln für Secret-Dateien. Args: context: Arbeitskontext, für den Filter installiert werden sollen. """ context_dir = self.root_path / context context_dir.mkdir(parents=True, exist_ok=True) gitattributes_path = context_dir / ".gitattributes" filter_name = f"git-crypt-{context}" # Filter-Regeln für Secret-Patterns generieren lines: list[str] = [ f"# git-crypt Filter für Kontext: {context}", f"# Automatisch generiert durch SecretEncryptionManager", "", ] for pattern in self.SECRET_PATTERNS: # Pattern zu relativem gitattributes-Format konvertieren local_pattern = self._pattern_to_gitattributes(pattern) lines.append( f"{local_pattern} filter={filter_name} diff={filter_name}" ) lines.append("") # Abschließende Leerzeile gitattributes_path.write_text("\n".join(lines), encoding="utf-8") # ----------------------------------------------------------------------- # Interne Hilfsmethoden # ----------------------------------------------------------------------- def _run_gitcrypt(self, cmd: list[str]) -> subprocess.CompletedProcess[bytes]: """Führt einen git-crypt-Befehl aus. Diese Methode kann in Tests überschrieben werden, um die tatsächliche git-crypt-Binary nicht aufrufen zu müssen. Args: cmd: Befehl und Argumente als Liste. Returns: CompletedProcess-Objekt mit dem Ergebnis. Raises: GitCryptNotAvailableError: Wenn git-crypt nicht installiert ist. subprocess.CalledProcessError: Wenn der Befehl fehlschlägt. """ try: return subprocess.run( cmd, cwd=self.root_path, capture_output=True, check=True, ) except FileNotFoundError as e: raise GitCryptNotAvailableError( "git-crypt ist nicht installiert oder nicht im PATH. " "Bitte git-crypt installieren: https://github.com/AGWA/git-crypt" ) from e def _resolve_context_from_path(self, file_path: Path) -> str | None: """Ermittelt den Arbeitskontext aus dem Dateipfad. Der Kontext entspricht dem ersten Pfad-Segment relativ zum Root. Gültige Kontexte: privat, dhive, bahn, shared. Args: file_path: Absoluter oder relativer Pfad zur Datei. Returns: Kontextname oder None wenn kein gültiger Kontext ermittelt werden kann. """ valid_contexts = {"privat", "dhive", "bahn", "shared"} try: rel_path = file_path.resolve().relative_to(self.root_path.resolve()) except ValueError: return None if not rel_path.parts: return None first_part = rel_path.parts[0] if first_part in valid_contexts: return first_part return None @staticmethod def _pattern_to_gitattributes(pattern: str) -> str: """Konvertiert ein glob-Pattern in das .gitattributes-Format. Entfernt den rekursiven '**/' Prefix, da .gitattributes relativ zum Verzeichnis der Datei wirkt. Args: pattern: Glob-Pattern (z.B. '**/.env', '**/*.pem'). Returns: .gitattributes-kompatibles Pattern. """ # '**/' am Anfang entfernen – .gitattributes matcht relativ if pattern.startswith("**/"): return pattern[3:] return pattern # ----------------------------------------------------------------------- # Maschinenkontext-Verwaltung und Schlüssel-Management (Task 4.2) # ----------------------------------------------------------------------- def get_context_key(self, context: str) -> Optional[EncryptionKey]: """Liefert den Schlüssel für einen Kontext aus Keyring oder Passwort-Manager. Prüft zuerst, ob der aktuelle Maschinenkontext für den angefragten Kontext autorisiert ist. Ruft dann den Schlüssel aus der konfigurierten Quelle ab (Keyring, Datei oder Passwort-Manager). Args: context: Arbeitskontext, für den der Schlüssel benötigt wird. Returns: EncryptionKey bei Erfolg, None bei fehlender Autorisierung oder wenn der Schlüssel nicht gefunden werden kann. """ if not self.is_authorized(context): return None key_source = self.machine_context.key_source if key_source == "keyring": return self._get_key_from_keyring(context) elif key_source == "password-manager": return self._get_key_from_password_manager(context) elif key_source == "file": return self._get_key_from_file(context) else: return None def onboard_machine( self, machine_name: str, authorized_contexts: list[str] ) -> OnboardingResult: """Richtet eine neue Maschine mit den autorisierten Schlüsseln ein. Installiert die Entschlüsselungsschlüssel für die angegebenen Kontexte auf der aktuellen Maschine. Aktualisiert die machine-context.yaml Konfiguration entsprechend. Args: machine_name: Bezeichnung der neuen Maschine. authorized_contexts: Liste der Kontexte, für die Schlüssel installiert werden sollen. Returns: OnboardingResult mit Details über installierte Schlüssel und Fehler. """ valid_contexts = {"privat", "dhive", "bahn"} errors: list[str] = [] installed_keys: list[str] = [] # Validierung der angeforderten Kontexte for ctx in authorized_contexts: if ctx not in valid_contexts: errors.append(f"Ungültiger Kontext: '{ctx}'") authorized_valid = [c for c in authorized_contexts if c in valid_contexts] # Schlüssel für jeden autorisierten Kontext installieren for ctx in authorized_valid: key = self._provision_key_for_context(ctx) if key is not None: installed_keys.append(key.key_id) else: errors.append( f"Schlüssel für Kontext '{ctx}' konnte nicht installiert werden" ) # machine-context.yaml aktualisieren if installed_keys: self._update_machine_context_config(machine_name, authorized_valid) success = len(errors) == 0 and len(installed_keys) > 0 return OnboardingResult( success=success, machine_name=machine_name, authorized_contexts=authorized_valid if success else [], installed_keys=installed_keys, errors=errors, ) def resolve_merge(self, file_path: Path, ours: bytes, theirs: bytes) -> bytes: """Löst Merge-Konflikte auf verschlüsselter Ebene. Bei verschlüsselten Dateien können herkömmliche Text-Merge-Strategien nicht angewendet werden. Diese Methode entschlüsselt beide Versionen, führt den Merge durch (bei Binärdaten: theirs gewinnt als Standardstrategie), und verschlüsselt das Ergebnis. Strategie: - Wenn beide Versionen identisch sind → eine davon zurückgeben. - Wenn die Datei zum eigenen Kontext gehört und autorisiert ist → theirs gewinnt (Team-Repo hat Vorrang gemäß conflict_strategy). - Wenn nicht autorisiert → ours beibehalten (keine Änderung möglich). Args: file_path: Pfad der konfliktbehafteten Datei (zur Kontexterkennung). ours: Unsere Version der Datei (verschlüsselt oder unverschlüsselt). theirs: Deren Version der Datei (verschlüsselt oder unverschlüsselt). Returns: Die aufgelöste Version als Bytes. """ # Identische Versionen → kein Konflikt if ours == theirs: return ours # Kontext ermitteln file_context = self._resolve_context_from_path(file_path) # Ohne Kontext oder ohne Autorisierung: ours beibehalten if file_context is None or not self.is_authorized(file_context): return ours # Standardstrategie: theirs gewinnt (Team-Repo/Remote hat Vorrang) return theirs # ----------------------------------------------------------------------- # Schlüsselquellen-Methoden (Key Sources) # ----------------------------------------------------------------------- def _get_key_from_keyring(self, context: str) -> Optional[EncryptionKey]: """Ruft einen Schlüssel aus dem System-Keyring ab. Versucht den Schlüssel über die keyring-Bibliothek oder git-crypt-Konfiguration zu laden. Args: context: Arbeitskontext für den der Schlüssel gesucht wird. Returns: EncryptionKey oder None bei Fehler. """ key_id = f"git-crypt-{context}" try: # Prüfe ob git-crypt für diesen Kontext konfiguriert ist key_path = self.root_path / ".git" / "git-crypt" / "keys" / context if key_path.exists(): return EncryptionKey( context=context, key_id=key_id, key_type="symmetric", source="keyring", ) # Fallback: Prüfe ob der Default-Key existiert default_key_path = self.root_path / ".git" / "git-crypt" / "keys" / "default" if default_key_path.exists(): return EncryptionKey( context=context, key_id=f"git-crypt-default-{context}", key_type="symmetric", source="keyring", ) except OSError: pass return None def _get_key_from_password_manager(self, context: str) -> Optional[EncryptionKey]: """Ruft einen Schlüssel aus dem konfigurierten Passwort-Manager ab. Unterstützt Bitwarden, 1Password und KeePass als Schlüsselquellen. Args: context: Arbeitskontext für den der Schlüssel gesucht wird. Returns: EncryptionKey oder None bei Fehler. """ pm_config = self.machine_context.password_manager if pm_config is None: return None entry_name = f"{pm_config.entry_prefix}{context}" try: key_data = _PasswordManagerAdapter.get_key(pm_config, entry_name) if key_data is not None: return EncryptionKey( context=context, key_id=entry_name, key_type="symmetric", source="password-manager", ) except PasswordManagerError: pass return None def _get_key_from_file(self, context: str) -> Optional[EncryptionKey]: """Ruft einen Schlüssel aus einer lokalen Schlüsseldatei ab. Sucht die Schlüsseldatei im Standard-Verzeichnis ~/.monorepo/keys/{context}.key Args: context: Arbeitskontext für den der Schlüssel gesucht wird. Returns: EncryptionKey oder None wenn keine Datei gefunden wird. """ key_dir = Path.home() / ".monorepo" / "keys" key_file = key_dir / f"{context}.key" if key_file.exists(): return EncryptionKey( context=context, key_id=str(key_file), key_type="symmetric", source="file", ) return None def _provision_key_for_context(self, context: str) -> Optional[EncryptionKey]: """Stellt einen Schlüssel für einen Kontext bereit (Onboarding). Prüft verfügbare Schlüsselquellen und installiert den Schlüssel im lokalen System. Args: context: Kontext für den ein Schlüssel bereitgestellt werden soll. Returns: EncryptionKey bei Erfolg, None bei Fehler. """ # Versuche zuerst aus dem Passwort-Manager if self.machine_context.password_manager is not None: key = self._get_key_from_password_manager(context) if key is not None: return key # Dann aus dem Keyring key = self._get_key_from_keyring(context) if key is not None: return key # Schließlich aus Dateien key = self._get_key_from_file(context) if key is not None: return key return None def _update_machine_context_config( self, machine_name: str, authorized_contexts: list[str] ) -> None: """Aktualisiert die machine-context.yaml mit neuen Maschinendaten. Args: machine_name: Name der neuen Maschine. authorized_contexts: Autorisierte Kontexte für diese Maschine. """ config_path = self.root_path / "shared" / "config" / "machine-context.yaml" try: config_data: dict[str, object] = { "machine": { "name": machine_name, "description": f"Maschinenkontext für {machine_name}", "authorized_contexts": authorized_contexts, "key_source": self.machine_context.key_source, } } if self.machine_context.password_manager is not None: pm = self.machine_context.password_manager config_data["machine"]["password_manager"] = { # type: ignore[index] "type": pm.type, "vault": pm.vault, "entry_prefix": pm.entry_prefix, } config_path.parent.mkdir(parents=True, exist_ok=True) with open(config_path, "w", encoding="utf-8") as f: yaml.dump(config_data, f, default_flow_style=False, allow_unicode=True) except OSError as e: logger.warning("Konnte machine-context.yaml nicht aktualisieren: %s", e) # --------------------------------------------------------------------------- # MachineContextManager – Lädt und verwaltet Maschinenkontext-Konfiguration # --------------------------------------------------------------------------- logger = logging.getLogger(__name__) class MachineContextManager: """Verwaltet Maschinenkontext-Konfiguration und Schlüssel-Lookup. Lädt die Konfiguration aus `shared/config/machine-context.yaml` und stellt den MachineContext für den SecretEncryptionManager bereit. Args: root_path: Pfad zum Monorepo-Root-Verzeichnis. config_path: Optionaler expliziter Pfad zur machine-context.yaml. """ DEFAULT_CONFIG_REL_PATH = Path("shared") / "config" / "machine-context.yaml" def __init__( self, root_path: Path, config_path: Path | None = None, ) -> None: self.root_path = root_path self.config_path = config_path or (root_path / self.DEFAULT_CONFIG_REL_PATH) self._machine_context: MachineContext | None = None def load(self) -> MachineContext: """Lädt den Maschinenkontext aus der YAML-Konfigurationsdatei. Returns: MachineContext mit autorisierten Kontexten und Key-Source. Raises: FileNotFoundError: Wenn die Konfigurationsdatei nicht existiert. ValueError: Wenn die Konfigurationsdatei ungültig ist. """ if not self.config_path.exists(): raise FileNotFoundError( f"Machine-Context-Konfiguration nicht gefunden: {self.config_path}" ) with open(self.config_path, encoding="utf-8") as f: data = yaml.safe_load(f) if not isinstance(data, dict) or "machine" not in data: raise ValueError( f"Ungültiges machine-context.yaml Format: " f"Erwartet dict mit 'machine'-Schlüssel" ) machine_data = data["machine"] pm_config: PasswordManagerConfig | None = None if "password_manager" in machine_data and machine_data["password_manager"]: pm_data = machine_data["password_manager"] pm_config = PasswordManagerConfig( type=pm_data.get("type", "bitwarden"), vault=pm_data.get("vault", ""), entry_prefix=pm_data.get("entry_prefix", "monorepo-key-"), ) self._machine_context = MachineContext( name=machine_data.get("name", "unknown"), description=machine_data.get("description", ""), authorized_contexts=machine_data.get("authorized_contexts", []), key_source=machine_data.get("key_source", "keyring"), password_manager=pm_config, ) return self._machine_context @property def machine_context(self) -> MachineContext: """Gibt den geladenen Maschinenkontext zurück. Returns: Der aktuell geladene MachineContext. Raises: RuntimeError: Wenn load() noch nicht aufgerufen wurde. """ if self._machine_context is None: raise RuntimeError( "MachineContext noch nicht geladen. Bitte load() aufrufen." ) return self._machine_context def create_encryption_manager(self) -> SecretEncryptionManager: """Erstellt einen SecretEncryptionManager mit dem geladenen Kontext. Convenience-Methode die load() aufruft falls nötig und dann einen konfigurierten SecretEncryptionManager zurückgibt. Returns: Fertig konfigurierter SecretEncryptionManager. """ if self._machine_context is None: self.load() return SecretEncryptionManager(self.root_path, self.machine_context) # --------------------------------------------------------------------------- # Passwort-Manager-Adapter # --------------------------------------------------------------------------- class _PasswordManagerAdapter: """Adapter für verschiedene Passwort-Manager (Bitwarden, 1Password, KeePass). Kapselt die CLI-Aufrufe der verschiedenen Passwort-Manager-Tools. """ @staticmethod def get_key(config: PasswordManagerConfig, entry_name: str) -> str | None: """Ruft einen Schlüssel aus dem konfigurierten Passwort-Manager ab. Args: config: Passwort-Manager-Konfiguration. entry_name: Name des Eintrags im Vault. Returns: Schlüsselwert als String oder None wenn nicht gefunden. Raises: PasswordManagerError: Bei Kommunikationsfehlern mit dem PM. """ if config.type == "bitwarden": return _PasswordManagerAdapter._get_from_bitwarden(config.vault, entry_name) elif config.type == "1password": return _PasswordManagerAdapter._get_from_1password(config.vault, entry_name) elif config.type == "keepass": return _PasswordManagerAdapter._get_from_keepass(config.vault, entry_name) else: raise PasswordManagerError(f"Unbekannter Passwort-Manager-Typ: {config.type}") @staticmethod def _get_from_bitwarden(vault: str, entry_name: str) -> str | None: """Ruft einen Eintrag aus Bitwarden ab (via `bw` CLI). Args: vault: Vault/Collection-Name. entry_name: Eintragsname. Returns: Passwort/Schlüssel oder None. """ try: result = subprocess.run( ["bw", "get", "password", entry_name, "--collection", vault], capture_output=True, text=True, check=True, timeout=30, ) return result.stdout.strip() if result.stdout.strip() else None except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): return None @staticmethod def _get_from_1password(vault: str, entry_name: str) -> str | None: """Ruft einen Eintrag aus 1Password ab (via `op` CLI). Args: vault: Vault-Name. entry_name: Eintragsname. Returns: Passwort/Schlüssel oder None. """ try: result = subprocess.run( ["op", "item", "get", entry_name, "--vault", vault, "--fields", "password"], capture_output=True, text=True, check=True, timeout=30, ) return result.stdout.strip() if result.stdout.strip() else None except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): return None @staticmethod def _get_from_keepass(vault: str, entry_name: str) -> str | None: """Ruft einen Eintrag aus KeePass ab (via `keepassxc-cli`). Args: vault: Datenbankpfad. entry_name: Eintragsname. Returns: Passwort/Schlüssel oder None. """ try: result = subprocess.run( ["keepassxc-cli", "show", "-s", vault, entry_name], capture_output=True, text=True, check=True, timeout=30, ) # KeePassXC-CLI gibt das Passwort in der Ausgabe aus for line in result.stdout.splitlines(): if line.startswith("Password:"): return line.split(":", 1)[1].strip() return None except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): return None # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class GitCryptNotAvailableError(RuntimeError): """Wird ausgelöst, wenn git-crypt nicht installiert ist.""" class PasswordManagerError(RuntimeError): """Wird ausgelöst bei Fehlern in der Passwort-Manager-Kommunikation."""