"""Kontextbrücke für kontextübergreifenden Wissenstransfer. Ermöglicht das sichere Teilen von Wissensartefakten zwischen Arbeitskontexten unter Wahrung der Sicherheitsgrenzen. Implementiert: - Sensitive-Content-Filterung (Secrets, Endpoints, PII) via Regex - Explizite Nutzerbestätigung vor Freigabe - Read-only-Referenzen in Zielkontexten - Update-Propagierung beim nächsten Lesezugriff - Widerruf von Freigaben Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6 """ from __future__ import annotations import logging import re from dataclasses import dataclass, field from pathlib import Path from typing import Any import yaml from monorepo.knowledge.artifact import Artifact, parse_frontmatter from monorepo.knowledge.index import IndexEntry, YAMLIndex from monorepo.models import Context logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Datenmodelle # --------------------------------------------------------------------------- @dataclass class SensitiveMatch: """Ein gefundener sensitiver Inhalt in einem Artefakt. Attributes: pattern_name: Name des Musters, das getroffen hat (z.B. "credentials", "endpoint", "pii"). matched_text: Der gefundene Text-Ausschnitt. line_number: Zeilennummer im Inhalt (1-basiert). reason: Menschenlesbare Begründung, warum der Inhalt als sensitiv gilt. """ pattern_name: str matched_text: str line_number: int reason: str @dataclass class ShareResult: """Ergebnis einer Freigabe-Operation. Attributes: success: True wenn die Freigabe erfolgreich war. artifact_id: ID des betroffenen Artefakts. shared_to: Liste der Kontexte, in denen das Artefakt nun verfügbar ist. errors: Liste von Fehlermeldungen bei gescheiterter Freigabe. """ success: bool artifact_id: str shared_to: list[str] = field(default_factory=list) errors: list[str] = field(default_factory=list) # --------------------------------------------------------------------------- # Shared-Artifacts-Registry # --------------------------------------------------------------------------- @dataclass class SharedArtifactEntry: """Eintrag in der Shared-Artifacts-Registry. Attributes: artifact_id: ID des geteilten Artefakts. source_context: Ursprungskontext des Artefakts. shared_to: Zielkontexte, in denen das Artefakt verfügbar ist. content_hash: Hash des Inhalts zum Zeitpunkt der Freigabe (für Update-Erkennung). """ artifact_id: str source_context: str shared_to: list[str] = field(default_factory=list) content_hash: str = "" # --------------------------------------------------------------------------- # ContextBridge # --------------------------------------------------------------------------- class ContextBridge: """Ermöglicht kontextübergreifenden Wissenstransfer. Die ContextBridge prüft Artefakte auf sensitive Inhalte, erfordert explizite Nutzerbestätigung und stellt geteilte Artefakte als schreibgeschützte Lesereferenzen in Zielkontexten bereit. Update-Propagierung: Wird ein Artefakt im Quellkontext aktualisiert, sind die Änderungen beim nächsten Lesezugriff über get_shared_content() sichtbar, da immer vom Quell-Artefakt gelesen wird. Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6 """ SENSITIVE_PATTERNS: list[tuple[str, str, str]] = [ ( r"(?i)(api[_-]?key|token|password|secret)\s*[:=]", "credentials", "Enthält kontextspezifische Zugangsdaten (API-Key, Token, Passwort oder Secret)", ), ( r"(?i)(endpoint|url)\s*[:=]\s*https?://", "endpoint", "Enthält kontextspezifische Endpoint-URL", ), ( r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "pii", "Enthält personenbezogene Daten (E-Mail-Adresse)", ), ] def __init__( self, knowledge_base_path: Path, index: YAMLIndex, registry_path: Path | None = None, ) -> None: """Initialisiert die ContextBridge. Args: knowledge_base_path: Basispfad des Wissensspeichers. index: Der YAML-Index des Wissensspeichers. registry_path: Pfad zur Shared-Artifacts-Registry (YAML). Falls None, wird in-memory gearbeitet. """ self.knowledge_base_path = knowledge_base_path self.index = index self.registry_path = registry_path self._registry: dict[str, SharedArtifactEntry] = {} if registry_path and registry_path.exists(): self._load_registry() # --------------------------------------------------------------------------- # Öffentliche API # --------------------------------------------------------------------------- def share_artifact( self, artifact_id: str, user_confirmed: bool = False ) -> ShareResult: """Gibt ein Artefakt kontextübergreifend frei. Prüft: 1. Existenz des Artefakts im Index 2. Explizite Nutzerbestätigung (Req 5.4) 3. Sensitive-Content-Filter (Req 5.1, 5.3) Bei Erfolg wird das Artefakt als Lesereferenz in allen anderen Kontexten bereitgestellt (Req 5.2). Args: artifact_id: ID des freizugebenden Artefakts. user_confirmed: Muss True sein für explizite Nutzerbestätigung. Returns: ShareResult mit Erfolgs-/Fehlerstatus. """ # Req 5.4: Explizite Nutzerbestätigung erforderlich if not user_confirmed: return ShareResult( success=False, artifact_id=artifact_id, errors=[ "Explizite Nutzerbestätigung erforderlich. " "Setze user_confirmed=True um die Freigabe zu bestätigen." ], ) # Artefakt im Index finden entry = self.index.get_entry(artifact_id) if entry is None: return ShareResult( success=False, artifact_id=artifact_id, errors=[f"Artefakt nicht im Index gefunden: {artifact_id}"], ) # Inhalt laden und auf sensitive Inhalte prüfen (Req 5.1, 5.3) content = self._load_artifact_content(entry) if content is None: return ShareResult( success=False, artifact_id=artifact_id, errors=[ f"Artefakt-Datei nicht gefunden oder nicht lesbar: {entry.path}" ], ) sensitive_matches = self.check_sensitive_content(content) if sensitive_matches: reasons = [ f"Zeile {m.line_number}: {m.reason} ('{m.matched_text}')" for m in sensitive_matches ] return ShareResult( success=False, artifact_id=artifact_id, errors=[ "Freigabe verweigert: Artefakt enthält sensible Inhalte.", *reasons, ], ) # Zielkontexte ermitteln (alle Kontexte außer dem Quellkontext) source_context = entry.scope target_contexts = [ ctx.value for ctx in Context if ctx.value != source_context ] # In Registry eintragen self._registry[artifact_id] = SharedArtifactEntry( artifact_id=artifact_id, source_context=source_context, shared_to=target_contexts, content_hash=entry.content_hash, ) # Registry persistieren self._save_registry() logger.info( "Artefakt '%s' aus Kontext '%s' für Kontexte %s freigegeben.", artifact_id, source_context, target_contexts, ) return ShareResult( success=True, artifact_id=artifact_id, shared_to=target_contexts, ) def check_sensitive_content(self, content: str) -> list[SensitiveMatch]: """Prüft Inhalt auf kontextspezifische Geheimnisse und PII. Verwendet Regex-Muster zur Erkennung von: - Credentials (API-Keys, Tokens, Passwörter, Secrets) - Endpoint-URLs - Personenbezogene Daten (E-Mail-Adressen) Args: content: Der zu prüfende Textinhalt. Returns: Liste der gefundenen sensitiven Stellen. Leere Liste wenn kein sensitiver Inhalt gefunden wurde. Requirements: 5.1, 5.3 """ matches: list[SensitiveMatch] = [] lines = content.splitlines() for line_num, line in enumerate(lines, start=1): for pattern_str, pattern_name, reason in self.SENSITIVE_PATTERNS: pattern = re.compile(pattern_str) for match in pattern.finditer(line): matched_text = match.group(0) # Text auf sinnvolle Länge kürzen if len(matched_text) > 60: matched_text = matched_text[:57] + "..." matches.append( SensitiveMatch( pattern_name=pattern_name, matched_text=matched_text, line_number=line_num, reason=reason, ) ) return matches def revoke_share(self, artifact_id: str) -> None: """Widerruft die Freigabe eines geteilten Artefakts. Entfernt das Artefakt aus allen Zielkontexten als Lesereferenz. Nach dem Widerruf ist das Artefakt nur noch im Quellkontext sichtbar. Args: artifact_id: ID des Artefakts, dessen Freigabe widerrufen wird. Raises: ValueError: Wenn das Artefakt nicht in der Freigabe-Registry ist. Requirements: 5.6 """ if artifact_id not in self._registry: raise ValueError( f"Artefakt '{artifact_id}' ist nicht freigegeben " f"und kann daher nicht widerrufen werden." ) entry = self._registry.pop(artifact_id) self._save_registry() logger.info( "Freigabe von Artefakt '%s' widerrufen. " "Entfernt aus Kontexten: %s", artifact_id, entry.shared_to, ) def get_shared_content(self, artifact_id: str, requesting_context: str) -> str | None: """Liest den aktuellen Inhalt eines geteilten Artefakts. Liefert immer den aktuellen Stand aus dem Quellkontext, sodass Updates automatisch beim nächsten Lesezugriff sichtbar sind. Args: artifact_id: ID des geteilten Artefakts. requesting_context: Der Kontext, aus dem der Zugriff erfolgt. Returns: Den aktuellen Inhalt des Artefakts als String, oder None wenn der Zugriff nicht erlaubt ist. Requirements: 5.2, 5.5 """ # Prüfe ob Artefakt freigegeben ist registry_entry = self._registry.get(artifact_id) if registry_entry is None: return None # Prüfe ob der anfragende Kontext berechtigt ist if requesting_context not in registry_entry.shared_to: return None # Lade den aktuellen Inhalt aus dem Quellkontext (Req 5.5: Update-Propagierung) index_entry = self.index.get_entry(artifact_id) if index_entry is None: return None return self._load_artifact_content(index_entry) def is_shared(self, artifact_id: str) -> bool: """Prüft ob ein Artefakt aktuell freigegeben ist. Args: artifact_id: ID des zu prüfenden Artefakts. Returns: True wenn das Artefakt freigegeben ist. """ return artifact_id in self._registry def get_shared_artifacts(self, context: str) -> list[str]: """Liefert alle Artefakt-IDs, die in einem bestimmten Kontext sichtbar sind. Args: context: Der Kontext, für den die sichtbaren Artefakte ermittelt werden. Returns: Liste der Artefakt-IDs, die im angegebenen Kontext als Lesereferenz verfügbar sind. """ return [ entry.artifact_id for entry in self._registry.values() if context in entry.shared_to ] # --------------------------------------------------------------------------- # Interne Methoden # --------------------------------------------------------------------------- def _load_artifact_content(self, entry: IndexEntry) -> str | None: """Lädt den Textinhalt eines Artefakts aus dem Dateisystem. Args: entry: Der IndexEntry des Artefakts. Returns: Den Inhalt als String oder None bei Fehler. """ if not entry.path: return None artifact_path = self.knowledge_base_path / entry.path if not artifact_path.exists(): logger.warning("Artefakt-Datei nicht gefunden: %s", artifact_path) return None try: artifact = parse_frontmatter(artifact_path) return artifact.content except (ValueError, FileNotFoundError, OSError) as e: logger.warning( "Fehler beim Laden von Artefakt '%s': %s", entry.id, e ) return None def _load_registry(self) -> None: """Lädt die Shared-Artifacts-Registry von der Festplatte.""" if self.registry_path is None or not self.registry_path.exists(): return try: with open(self.registry_path, encoding="utf-8") as f: data = yaml.safe_load(f) except (yaml.YAMLError, OSError) as e: logger.warning("Fehler beim Laden der Registry: %s", e) return if not isinstance(data, dict): return for artifact_data in data.get("shared_artifacts", []): if isinstance(artifact_data, dict): entry = SharedArtifactEntry( artifact_id=str(artifact_data.get("artifact_id", "")), source_context=str(artifact_data.get("source_context", "")), shared_to=list(artifact_data.get("shared_to", [])), content_hash=str(artifact_data.get("content_hash", "")), ) if entry.artifact_id: self._registry[entry.artifact_id] = entry def _save_registry(self) -> None: """Persistiert die Shared-Artifacts-Registry auf die Festplatte.""" if self.registry_path is None: return data: dict[str, Any] = { "shared_artifacts": [ { "artifact_id": entry.artifact_id, "source_context": entry.source_context, "shared_to": entry.shared_to, "content_hash": entry.content_hash, } for entry in self._registry.values() ] } self.registry_path.parent.mkdir(parents=True, exist_ok=True) with open(self.registry_path, "w", encoding="utf-8") as f: yaml.dump( data, f, default_flow_style=False, allow_unicode=True, sort_keys=False, )