Initial monorepo structure
This commit is contained in:
@@ -0,0 +1,256 @@
|
||||
"""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"
|
||||
Reference in New Issue
Block a user