147 lines
5.0 KiB
Python
147 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
# /// script
|
|
# dependencies = [
|
|
# "boto3>=1.34.0",
|
|
# ]
|
|
# ///
|
|
"""
|
|
audit_hosting_boundaries.py — Automated Continuous Leak Prevention Guard (ADR 0005).
|
|
|
|
Verifies that no personal, familial, or physiological data files are exposed at the
|
|
unauthenticated root or public folders of MinIO bucket 'hosting'.
|
|
Enforces:
|
|
1. All dashboards for Mila, Moritz, family nutrition, and baby sleep must strictly reside under 'family/'.
|
|
2. Root and public directories must only contain public assets (/recipes/*, /showcase/*).
|
|
3. Any violating key found at root or public prefixes is detected and optionally quarantined/purged.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
DEFAULT_ENDPOINT = os.environ.get("MINIO_ENDPOINT", "http://minio.minio.svc.cluster.local:9000")
|
|
DEFAULT_ACCESS_KEY = os.environ.get("MINIO_ACCESS_KEY", "openclaw")
|
|
DEFAULT_SECRET_KEY = os.environ.get("MINIO_SECRET_KEY", "5uu23t69xC2Oigf8Bky0KTlI")
|
|
DEFAULT_BUCKET = os.environ.get("MINIO_BUCKET", "hosting")
|
|
|
|
SENSITIVE_PATTERNS = [
|
|
"mila",
|
|
"moritz",
|
|
"nutrition",
|
|
"sleep",
|
|
"schlaf",
|
|
"health",
|
|
"diaper",
|
|
"windel",
|
|
"fläschchen",
|
|
"flaeschchen",
|
|
"garmin",
|
|
"zwift",
|
|
"incyclist",
|
|
]
|
|
|
|
|
|
def get_s3_client():
|
|
import boto3
|
|
from botocore.client import Config
|
|
|
|
return boto3.client(
|
|
"s3",
|
|
endpoint_url=DEFAULT_ENDPOINT,
|
|
aws_access_key_id=DEFAULT_ACCESS_KEY,
|
|
aws_secret_access_key=DEFAULT_SECRET_KEY,
|
|
config=Config(signature_version="s3v4", connect_timeout=5, retries={"max_attempts": 2}),
|
|
region_name="us-east-1",
|
|
)
|
|
|
|
|
|
def audit_hosting_bucket(clean: bool = False) -> list[str]:
|
|
"""Scans MinIO bucket for sensitive personal files outside family/."""
|
|
s3 = get_s3_client()
|
|
violations: list[str] = []
|
|
|
|
paginator = s3.get_paginator("list_objects_v2")
|
|
for page in paginator.paginate(Bucket=DEFAULT_BUCKET):
|
|
for obj in page.get("Contents", []):
|
|
key = obj["Key"]
|
|
# Whitelisted protected prefixes
|
|
if key.startswith("family/") or key.startswith("private/"):
|
|
continue
|
|
|
|
# Check if key matches sensitive personal data
|
|
key_lower = key.lower()
|
|
if any(term in key_lower for term in SENSITIVE_PATTERNS):
|
|
violations.append(key)
|
|
|
|
if violations:
|
|
print(f"🚨 SECURITY VIOLATION: Found {len(violations)} sensitive personal file(s) outside /family/ in bucket '{DEFAULT_BUCKET}':", file=sys.stderr)
|
|
for v in violations:
|
|
print(f" - {v}", file=sys.stderr)
|
|
if clean:
|
|
try:
|
|
s3.delete_object(Bucket=DEFAULT_BUCKET, Key=v)
|
|
print(f" [CLEANED] Deleted '{v}' from public area.", file=sys.stderr)
|
|
except Exception as ex:
|
|
print(f" [ERROR] Failed to delete '{v}': {ex}", file=sys.stderr)
|
|
else:
|
|
print(f"✅ S3 Hosting Audit PASSED: Zero sensitive personal files found in public areas of bucket '{DEFAULT_BUCKET}'.")
|
|
|
|
return violations
|
|
|
|
|
|
def audit_local_public_vault(vault_dir: Path, clean: bool = False) -> list[Path]:
|
|
"""Scans local knowledge-wiki/public directory for sensitive files outside family/."""
|
|
violations: list[Path] = []
|
|
public_dir = vault_dir / "public"
|
|
if not public_dir.is_dir():
|
|
return violations
|
|
|
|
for f in public_dir.iterdir():
|
|
if f.is_file():
|
|
f_lower = f.name.lower()
|
|
if any(term in f_lower for term in SENSITIVE_PATTERNS):
|
|
violations.append(f)
|
|
if clean:
|
|
try:
|
|
f.unlink()
|
|
print(f" [CLEANED] Deleted local unauthenticated file: {f}")
|
|
except Exception as ex:
|
|
print(f" [ERROR] Failed to delete {f}: {ex}")
|
|
|
|
if violations and not clean:
|
|
print(f"🚨 Local Vault Public Warning: Found {len(violations)} personal file(s) in {public_dir} root.", file=sys.stderr)
|
|
elif not violations:
|
|
print(f"✅ Local Vault Public Audit PASSED: No sensitive files at public root.")
|
|
|
|
return violations
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Audit and enforce ADR 0005 hosting boundary segregation")
|
|
parser.add_argument("--clean", action="store_true", help="Automatically delete any violating files found")
|
|
parser.add_argument("--vault", help="Path to knowledge-wiki directory")
|
|
args = parser.parse_args()
|
|
|
|
vault_path = Path(args.vault).resolve() if args.vault else Path(__file__).resolve().parents[4] / "knowledge-wiki"
|
|
|
|
s3_violations = []
|
|
try:
|
|
s3_violations = audit_hosting_bucket(clean=args.clean)
|
|
except Exception as ex:
|
|
print(f"Warning: could not connect to S3/MinIO for audit: {ex}", file=sys.stderr)
|
|
|
|
local_violations = audit_local_public_vault(vault_path, clean=args.clean)
|
|
|
|
if s3_violations or local_violations:
|
|
if not args.clean:
|
|
sys.exit(1)
|
|
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|