"""Property-basierte Tests für den ContextGuard: Kontextübergreifender Secret-Zugriff. **Validates: Requirements 2.1, 2.3, 2.5, 2.7** Property 3: Kontextübergreifender Secret-Zugriff wird verweigert - For any Kombination aus anfragendem Kontext A und Zielkontext B (wobei A ≠ B und B ≠ shared), muss der Zugriff auf .env-Dateien und Secret-Dateien von B verweigert werden, und es muss ein Audit-Log-Eintrag mit Zeitstempel, anfragendem Kontext, Zielkontext und Ressource erzeugt werden. """ from __future__ import annotations import tempfile from datetime import datetime from pathlib import Path from hypothesis import given, settings, assume from hypothesis import strategies as st from monorepo.audit import AuditLogger from monorepo.models import Context, SecurityEvent from monorepo.security import ContextGuard # --- Constants --- # Non-shared contexts that can be requesting or target contexts NON_SHARED_CONTEXTS = [c.value for c in Context if c != Context.SHARED] ALL_CONTEXTS = [c.value for c in Context] # Secret file patterns that should be protected SECRET_FILE_NAMES = [".env", "credentials.pem", "private.key", "api-token.txt", "secret-config.yaml"] # --- Strategies --- @st.composite def cross_context_pairs(draw: st.DrawFn) -> tuple[str, str]: """Generates pairs (requesting_context, target_context) where: - A ≠ B - B ≠ shared (target is NOT the shared context) """ context_a = draw(st.sampled_from(ALL_CONTEXTS)) context_b = draw(st.sampled_from(NON_SHARED_CONTEXTS)) assume(context_a != context_b) return context_a, context_b @st.composite def secret_file_paths(draw: st.DrawFn, target_context: str) -> Path: """Generates secret file paths within a given target context. Produces paths like: - {context}/.env - {context}/some-project/{secret_file} """ secret_name = draw(st.sampled_from(SECRET_FILE_NAMES)) # Either at context root or in a subdirectory use_subdir = draw(st.booleans()) if use_subdir: subdir = draw(st.sampled_from(["project-a", "module-x", "config", "keys"])) return Path(target_context) / subdir / secret_name return Path(target_context) / secret_name # --- Helpers --- def _create_test_environment() -> tuple[Path, ContextGuard, AuditLogger]: """Creates a temporary monorepo structure with ContextGuard and AuditLogger. Returns a tuple of (tmp_dir_path, guard, audit_logger). The caller does NOT need to clean up - Python's tempfile handles that. """ tmp_dir = Path(tempfile.mkdtemp(prefix="monorepo_test_")) # Create the monorepo directory structure for ctx in ALL_CONTEXTS: (tmp_dir / ctx).mkdir(parents=True, exist_ok=True) # Create access-config.yaml config_dir = tmp_dir / "shared" / "config" config_dir.mkdir(parents=True, exist_ok=True) config_file = config_dir / "access-config.yaml" config_file.write_text( "contexts:\n" " privat:\n" " env_file: privat/.env\n" " allowed_shared:\n" " - shared/tools/\n" " - shared/powers/\n" " - shared/config/\n" " dhive:\n" " env_file: dhive/.env\n" " allowed_shared:\n" " - shared/tools/\n" " - shared/powers/\n" " - shared/config/\n" " bahn:\n" " env_file: bahn/.env\n" " allowed_shared:\n" " - shared/tools/\n" " - shared/powers/\n" " - shared/config/\n" " - shared/knowledge-store/\n" " shared:\n" " env_file: shared/.env\n" ' allowed_shared: ["*"]\n', encoding="utf-8", ) guard = ContextGuard(root_path=tmp_dir, access_config_path=config_file) log_path = tmp_dir / ".audit" / "access.log" audit_logger = AuditLogger(log_path=log_path) return tmp_dir, guard, audit_logger # --- Tests --- class TestProperty3CrossContextSecretAccessDenied: """Property 3: Kontextübergreifender Secret-Zugriff wird verweigert. **Validates: Requirements 2.1, 2.3, 2.5, 2.7** For any combination of requesting context A and target context B (where A ≠ B and B ≠ shared), access to .env files and secret files of B must be denied, and an audit log entry must be generated with timestamp, requesting context, target context, and resource. """ @given(data=st.data()) @settings(max_examples=200) def test_cross_context_env_access_denied(self, data: st.DataObject) -> None: """Access to .env files of another non-shared context is always denied.""" context_a, context_b = data.draw(cross_context_pairs()) _, guard, _ = _create_test_environment() # Target: the .env file of context B target_path = Path(context_b) / ".env" result = guard.check_access(context_a, target_path) assert result is False, ( f"ContextGuard allowed cross-context .env access: " f"'{context_a}' accessing '{target_path}' should be denied" ) @given(data=st.data()) @settings(max_examples=200) def test_cross_context_secret_file_access_denied(self, data: st.DataObject) -> None: """Access to secret files of another non-shared context is always denied.""" context_a, context_b = data.draw(cross_context_pairs()) _, guard, _ = _create_test_environment() # Generate a secret file path in context B target_path = data.draw(secret_file_paths(target_context=context_b)) result = guard.check_access(context_a, target_path) assert result is False, ( f"ContextGuard allowed cross-context secret access: " f"'{context_a}' accessing '{target_path}' should be denied" ) @given(data=st.data()) @settings(max_examples=200) def test_cross_context_access_generates_audit_log(self, data: st.DataObject) -> None: """When cross-context secret access is denied, an audit log entry must be generated with timestamp, requesting_context, target_context, and resource.""" context_a, context_b = data.draw(cross_context_pairs()) _, guard, audit_logger = _create_test_environment() # Target: .env file of context B target_path = Path(context_b) / ".env" # Verify access is denied access_allowed = guard.check_access(context_a, target_path) assert access_allowed is False # Log the violation (as the integration layer would do on denied access) now = datetime.now() event = SecurityEvent( timestamp=now, requesting_context=context_a, target_context=context_b, resource=str(target_path), action="read", outcome="denied", ) audit_logger.log_violation(event) # Verify audit log entry was created with required fields log_content = audit_logger.log_path.read_text(encoding="utf-8") assert log_content.strip() != "", "Audit log should not be empty after violation" # Parse the last log entry and verify required fields last_line = log_content.strip().split("\n")[-1] # Verify timestamp is present (ISO format in brackets) assert "[" in last_line and "]" in last_line, ( f"Audit log entry missing timestamp brackets: {last_line}" ) # Verify requesting context is present assert context_a in last_line, ( f"Audit log entry missing requesting context '{context_a}': {last_line}" ) # Verify target context is present assert context_b in last_line, ( f"Audit log entry missing target context '{context_b}': {last_line}" ) # Verify resource path is present assert str(target_path) in last_line, ( f"Audit log entry missing resource '{target_path}': {last_line}" ) @given(data=st.data()) @settings(max_examples=100) def test_audit_log_entry_has_valid_timestamp(self, data: st.DataObject) -> None: """The audit log entry must contain a parseable ISO timestamp.""" context_a, context_b = data.draw(cross_context_pairs()) _, _, audit_logger = _create_test_environment() now = datetime.now() event = SecurityEvent( timestamp=now, requesting_context=context_a, target_context=context_b, resource=f"{context_b}/.env", action="read", outcome="denied", ) audit_logger.log_violation(event) log_content = audit_logger.log_path.read_text(encoding="utf-8") last_line = log_content.strip().split("\n")[-1] # Extract timestamp from brackets [timestamp] start = last_line.index("[") + 1 end = last_line.index("]") timestamp_str = last_line[start:end] # Verify it's a valid ISO timestamp parsed_ts = datetime.fromisoformat(timestamp_str) assert parsed_ts is not None, f"Could not parse timestamp: {timestamp_str}" # Timestamp should be very close to 'now' (within seconds) delta = abs((parsed_ts - now).total_seconds()) assert delta < 2.0, f"Timestamp drift too large: {delta}s"