362 lines
13 KiB
Python
362 lines
13 KiB
Python
"""Sicherheits-Guard für die Zugriffskontrolle zwischen Arbeitskontexten.
|
||
|
||
Implementiert den ContextGuard, der Sicherheitsgrenzen zwischen den
|
||
Arbeitskontexten (privat, dhive, bahn, shared) erzwingt. Die Zugriffskontrolle
|
||
basiert auf der zentralen access-config.yaml.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import yaml
|
||
|
||
from monorepo.models import Context
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ContextGuard:
|
||
"""Erzwingt Sicherheitsgrenzen zwischen Arbeitskontexten.
|
||
|
||
Prüft ob ein anfragender Kontext auf einen gegebenen Pfad zugreifen darf,
|
||
basierend auf den Regeln in access-config.yaml:
|
||
- Ein Kontext darf immer auf eigene Dateien zugreifen.
|
||
- Ein Kontext darf auf shared-Pfade zugreifen, die in allowed_shared gelistet sind.
|
||
- Der shared-Kontext mit ["*"] darf auf alles in shared zugreifen.
|
||
- Alle anderen kontextübergreifenden Zugriffe werden verweigert.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
root_path: Path,
|
||
access_config_path: Path | None = None,
|
||
encryption_manager: Any | None = None,
|
||
) -> None:
|
||
"""Initialisiert den ContextGuard.
|
||
|
||
Args:
|
||
root_path: Wurzelverzeichnis des Monorepos.
|
||
access_config_path: Pfad zur access-config.yaml.
|
||
Falls None, wird `root_path / shared/config/access-config.yaml` verwendet.
|
||
encryption_manager: Optionaler SecretEncryptionManager für die
|
||
Entschlüsselung von .env-Dateien. Falls None, werden .env-Dateien
|
||
direkt gelesen (unverschlüsselt oder bereits durch git-crypt entschlüsselt).
|
||
"""
|
||
self.root_path = root_path.resolve()
|
||
if access_config_path is None:
|
||
access_config_path = self.root_path / "shared" / "config" / "access-config.yaml"
|
||
self.access_config_path = access_config_path.resolve()
|
||
self.encryption_manager = encryption_manager
|
||
self._config: dict[str, Any] = self._load_access_config()
|
||
|
||
def _load_access_config(self) -> dict[str, Any]:
|
||
"""Lädt die access-config.yaml.
|
||
|
||
Returns:
|
||
Dict mit der Zugriffskonfiguration pro Kontext.
|
||
|
||
Raises:
|
||
FileNotFoundError: Wenn die Konfigurationsdatei nicht existiert.
|
||
ValueError: Wenn das YAML-Format ungültig ist.
|
||
"""
|
||
if not self.access_config_path.exists():
|
||
raise FileNotFoundError(
|
||
f"Access-Konfiguration nicht gefunden: {self.access_config_path}"
|
||
)
|
||
|
||
with open(self.access_config_path, encoding="utf-8") as f:
|
||
data = yaml.safe_load(f)
|
||
|
||
if not isinstance(data, dict) or "contexts" not in data:
|
||
raise ValueError(
|
||
f"Ungültiges access-config.yaml-Format: "
|
||
f"Erwartet dict mit 'contexts'-Schlüssel in {self.access_config_path}"
|
||
)
|
||
|
||
return data["contexts"]
|
||
|
||
def check_access(self, requesting_context: str, target_path: Path) -> bool:
|
||
"""Prüft ob der Zugriff auf target_path vom requesting_context erlaubt ist.
|
||
|
||
Regeln:
|
||
1. Ein Kontext darf immer auf eigene Dateien zugreifen (Pfad unter eigenem Kontextordner).
|
||
2. Ein Kontext darf auf shared-Pfade zugreifen, wenn diese in allowed_shared gelistet sind.
|
||
3. Der shared-Kontext mit allowed_shared: ["*"] darf auf alles in shared zugreifen.
|
||
4. Alle anderen kontextübergreifenden Zugriffe werden verweigert.
|
||
|
||
Args:
|
||
requesting_context: Name des anfragenden Kontexts (z.B. "privat", "dhive", "bahn", "shared").
|
||
target_path: Pfad auf den zugegriffen werden soll (absolut oder relativ zum root_path).
|
||
|
||
Returns:
|
||
True wenn der Zugriff erlaubt ist, False wenn verweigert.
|
||
"""
|
||
# Pfad relativ zum Root normalisieren
|
||
rel_path = self._resolve_relative_path(target_path)
|
||
rel_path_str = rel_path.as_posix()
|
||
|
||
# Kontext des Zielpfads ermitteln
|
||
target_context = self._get_path_context(rel_path_str)
|
||
|
||
# Regel 1: Zugriff auf eigenen Kontext immer erlaubt
|
||
if target_context == requesting_context:
|
||
return True
|
||
|
||
# Regel 2+3: Zugriff auf shared-Bereich prüfen
|
||
if target_context == Context.SHARED.value:
|
||
return self._check_shared_access(requesting_context, rel_path_str)
|
||
|
||
# Regel 4: Kontextübergreifender Zugriff verweigert
|
||
return False
|
||
|
||
def load_env(self, context: str) -> dict[str, str]:
|
||
"""Lädt die .env-Datei des gegebenen Kontexts (mit optionaler Entschlüsselung).
|
||
|
||
Wenn ein SecretEncryptionManager konfiguriert ist und der Maschinenkontext
|
||
für den angefragten Kontext autorisiert ist, wird die .env-Datei via
|
||
decrypt_file entschlüsselt. Andernfalls wird die Datei direkt gelesen
|
||
(unverschlüsselt oder bereits durch git-crypt entschlüsselt im Working Tree).
|
||
|
||
Parst eine Standard-.env-Datei (KEY=VALUE-Format) und gibt die
|
||
Umgebungsvariablen als Dict zurück. Kommentare (#) und Leerzeilen
|
||
werden übersprungen.
|
||
|
||
Args:
|
||
context: Name des Kontexts, dessen .env geladen werden soll.
|
||
|
||
Returns:
|
||
Dict mit den Umgebungsvariablen (Schlüssel → Wert).
|
||
|
||
Raises:
|
||
ValueError: Wenn der Kontext nicht in der Konfiguration definiert ist.
|
||
FileNotFoundError: Wenn die .env-Datei nicht existiert.
|
||
PermissionError: Wenn die Entschlüsselung fehlschlägt
|
||
(Maschinenkontext nicht autorisiert).
|
||
"""
|
||
if context not in self._config:
|
||
raise ValueError(
|
||
f"Unbekannter Kontext '{context}'. "
|
||
f"Gültige Kontexte: {list(self._config.keys())}"
|
||
)
|
||
|
||
env_file_rel = self._config[context].get("env_file")
|
||
if not env_file_rel:
|
||
raise ValueError(
|
||
f"Kein env_file für Kontext '{context}' in der Konfiguration definiert."
|
||
)
|
||
|
||
env_path = self.root_path / env_file_rel
|
||
|
||
if not env_path.exists():
|
||
raise FileNotFoundError(
|
||
f".env-Datei für Kontext '{context}' nicht gefunden: {env_path}"
|
||
)
|
||
|
||
# Integration mit SecretEncryptionManager für Entschlüsselung (Req 2.2, 9.3).
|
||
if self.encryption_manager is not None:
|
||
return self._load_env_with_decryption(env_path, context)
|
||
|
||
# Fallback: Datei direkt lesen (unverschlüsselt oder bereits entschlüsselt)
|
||
return self._parse_env_file(env_path)
|
||
|
||
def _load_env_with_decryption(
|
||
self, env_path: Path, context: str
|
||
) -> dict[str, str]:
|
||
"""Lädt und entschlüsselt eine .env-Datei via SecretEncryptionManager.
|
||
|
||
Prüft zuerst die Autorisierung des Maschinenkontexts. Wenn autorisiert,
|
||
wird die Datei entschlüsselt und geparst. Andernfalls wird ein
|
||
PermissionError ausgelöst.
|
||
|
||
Args:
|
||
env_path: Absoluter Pfad zur .env-Datei.
|
||
context: Arbeitskontext der .env-Datei.
|
||
|
||
Returns:
|
||
Dict mit den entschlüsselten Umgebungsvariablen.
|
||
|
||
Raises:
|
||
PermissionError: Wenn der Maschinenkontext nicht autorisiert ist.
|
||
"""
|
||
encryption_mgr = self.encryption_manager
|
||
|
||
# Autorisierungsprüfung
|
||
if not encryption_mgr.is_authorized(context):
|
||
raise PermissionError(
|
||
f"Maschinenkontext '{encryption_mgr.machine_context.name}' ist nicht "
|
||
f"für Kontext '{context}' autorisiert. Entschlüsselung verweigert."
|
||
)
|
||
|
||
# Entschlüsselung versuchen
|
||
result = encryption_mgr.decrypt_file(env_path)
|
||
|
||
if result.success and result.content is not None:
|
||
# Entschlüsselter Inhalt als Text parsen
|
||
content_text = result.content.decode("utf-8", errors="replace")
|
||
return self._parse_env_content(content_text)
|
||
|
||
# Entschlüsselung fehlgeschlagen – Fallback auf direkte Leseoperation.
|
||
# Dies tritt auf wenn die Datei bereits im Klartext vorliegt (z.B.
|
||
# nach git-crypt unlock) oder wenn git-crypt nicht verfügbar ist.
|
||
logger.debug(
|
||
"Entschlüsselung für '%s' nicht erfolgreich (%s). "
|
||
"Fallback auf direktes Lesen.",
|
||
env_path,
|
||
result.error or "unbekannter Fehler",
|
||
)
|
||
return self._parse_env_file(env_path)
|
||
|
||
def get_context_config(self, context: str) -> dict[str, Any]:
|
||
"""Gibt die Konfiguration für einen bestimmten Kontext zurück.
|
||
|
||
Args:
|
||
context: Name des Kontexts.
|
||
|
||
Returns:
|
||
Dict mit env_file und allowed_shared für den Kontext.
|
||
|
||
Raises:
|
||
ValueError: Wenn der Kontext nicht in der Konfiguration definiert ist.
|
||
"""
|
||
if context not in self._config:
|
||
raise ValueError(
|
||
f"Unbekannter Kontext '{context}'. "
|
||
f"Gültige Kontexte: {list(self._config.keys())}"
|
||
)
|
||
return self._config[context]
|
||
|
||
def _resolve_relative_path(self, target_path: Path) -> Path:
|
||
"""Löst einen Pfad relativ zum Monorepo-Root auf.
|
||
|
||
Args:
|
||
target_path: Absoluter oder relativer Pfad.
|
||
|
||
Returns:
|
||
Pfad relativ zum Root (immer mit Forward-Slashes via as_posix).
|
||
"""
|
||
resolved = Path(target_path).resolve() if target_path.is_absolute() else target_path
|
||
try:
|
||
return resolved.relative_to(self.root_path)
|
||
except ValueError:
|
||
# Pfad ist bereits relativ oder liegt außerhalb des Roots
|
||
return target_path
|
||
|
||
def _get_path_context(self, rel_path_str: str) -> str:
|
||
"""Ermittelt den Kontext eines relativen Pfads.
|
||
|
||
Der Kontext wird anhand des obersten Verzeichnisses bestimmt.
|
||
Gültige Kontexte: privat, dhive, bahn, shared.
|
||
|
||
Args:
|
||
rel_path_str: Relativer Pfad als String (mit Forward-Slashes).
|
||
|
||
Returns:
|
||
Name des Kontexts oder leerer String wenn nicht zuordenbar.
|
||
"""
|
||
parts = rel_path_str.split("/")
|
||
if not parts:
|
||
return ""
|
||
|
||
first_part = parts[0]
|
||
valid_contexts = {ctx.value for ctx in Context}
|
||
if first_part in valid_contexts:
|
||
return first_part
|
||
|
||
return ""
|
||
|
||
def _check_shared_access(self, requesting_context: str, rel_path_str: str) -> bool:
|
||
"""Prüft ob ein Kontext auf einen shared-Pfad zugreifen darf.
|
||
|
||
Args:
|
||
requesting_context: Name des anfragenden Kontexts.
|
||
rel_path_str: Relativer Pfad (beginnt mit "shared/").
|
||
|
||
Returns:
|
||
True wenn der Zugriff erlaubt ist.
|
||
"""
|
||
if requesting_context not in self._config:
|
||
return False
|
||
|
||
allowed_shared = self._config[requesting_context].get("allowed_shared", [])
|
||
|
||
# Wildcard: Zugriff auf alles in shared erlaubt
|
||
if allowed_shared == ["*"]:
|
||
return True
|
||
|
||
# Prüfe ob der Pfad unter einem der erlaubten shared-Pfade liegt
|
||
for allowed_path in allowed_shared:
|
||
# Normalisiere: Stelle sicher dass der Pfad mit / endet für Prefix-Matching
|
||
allowed_normalized = allowed_path.rstrip("/") + "/"
|
||
# Pfad liegt unter dem erlaubten Verzeichnis oder ist es selbst
|
||
if rel_path_str.startswith(allowed_normalized) or rel_path_str.rstrip("/") + "/" == allowed_normalized:
|
||
return True
|
||
|
||
return False
|
||
|
||
@staticmethod
|
||
def _parse_env_file(env_path: Path) -> dict[str, str]:
|
||
"""Parst eine .env-Datei im Standard-Format.
|
||
|
||
Format:
|
||
- KEY=VALUE (ein Paar pro Zeile)
|
||
- Zeilen die mit # beginnen sind Kommentare
|
||
- Leerzeilen werden übersprungen
|
||
- Führende/nachfolgende Leerzeichen werden entfernt
|
||
- Werte in Anführungszeichen (einfach oder doppelt) werden entquotet
|
||
|
||
Args:
|
||
env_path: Pfad zur .env-Datei.
|
||
|
||
Returns:
|
||
Dict mit geparsten Umgebungsvariablen.
|
||
"""
|
||
with open(env_path, encoding="utf-8") as f:
|
||
content = f.read()
|
||
return ContextGuard._parse_env_content(content)
|
||
|
||
@staticmethod
|
||
def _parse_env_content(content: str) -> dict[str, str]:
|
||
"""Parst .env-Inhalt im Standard-Format aus einem String.
|
||
|
||
Format:
|
||
- KEY=VALUE (ein Paar pro Zeile)
|
||
- Zeilen die mit # beginnen sind Kommentare
|
||
- Leerzeilen werden übersprungen
|
||
- Führende/nachfolgende Leerzeichen werden entfernt
|
||
- Werte in Anführungszeichen (einfach oder doppelt) werden entquotet
|
||
|
||
Args:
|
||
content: Textinhalt der .env-Datei.
|
||
|
||
Returns:
|
||
Dict mit geparsten Umgebungsvariablen.
|
||
"""
|
||
env_vars: dict[str, str] = {}
|
||
|
||
for line in content.splitlines():
|
||
line = line.strip()
|
||
|
||
# Leerzeilen und Kommentare überspringen
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
|
||
# KEY=VALUE aufteilen (nur beim ersten = splitten)
|
||
if "=" not in line:
|
||
continue
|
||
|
||
key, _, value = line.partition("=")
|
||
key = key.strip()
|
||
value = value.strip()
|
||
|
||
# Anführungszeichen entfernen
|
||
if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
|
||
value = value[1:-1]
|
||
|
||
if key:
|
||
env_vars[key] = value
|
||
|
||
return env_vars
|