257 lines
9.1 KiB
Python
257 lines
9.1 KiB
Python
"""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"
|