74 lines
3.1 KiB
Python
Executable File
74 lines
3.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Read-only tenant-isolation pre-audit for CASAN state partitions.
|
|
|
|
It validates filesystem boundaries and records concrete findings. It is not an
|
|
independent audit; use its JSON output as input to an independent review.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
TENANT = re.compile(r"^[A-Za-z0-9_-]+$")
|
|
|
|
def mode(path: Path) -> int:
|
|
return stat.S_IMODE(path.lstat().st_mode)
|
|
|
|
def insecure(mode_value: int) -> bool:
|
|
return bool(mode_value & 0o077)
|
|
|
|
def review(root: Path) -> dict:
|
|
findings: list[dict] = []
|
|
tenants: list[str] = []
|
|
keys: dict[str, str] = {}
|
|
if not root.is_dir() or root.is_symlink():
|
|
findings.append({"severity": "critical", "code": "tenant_root_invalid", "path": str(root)})
|
|
return {"ok": False, "root": str(root), "tenants": tenants, "findings": findings}
|
|
if insecure(mode(root)):
|
|
findings.append({"severity": "high", "code": "tenant_root_permissions", "path": str(root), "mode": oct(mode(root))})
|
|
for candidate in sorted(root.iterdir()):
|
|
if not TENANT.fullmatch(candidate.name):
|
|
findings.append({"severity": "medium", "code": "unexpected_root_entry", "path": str(candidate)})
|
|
continue
|
|
tenants.append(candidate.name)
|
|
if candidate.is_symlink() or not candidate.is_dir():
|
|
findings.append({"severity": "critical", "code": "tenant_partition_not_directory", "tenant": candidate.name})
|
|
continue
|
|
if insecure(mode(candidate)):
|
|
findings.append({"severity": "high", "code": "tenant_permissions", "tenant": candidate.name, "mode": oct(mode(candidate))})
|
|
for current, dirs, files in os.walk(candidate, followlinks=False):
|
|
base = Path(current)
|
|
for name in dirs + files:
|
|
item = base / name
|
|
if item.is_symlink():
|
|
findings.append({"severity": "critical", "code": "tenant_symlink_forbidden", "tenant": candidate.name, "path": str(item)})
|
|
key = candidate / "keys" / "at-rest.key"
|
|
if key.exists():
|
|
if key.is_symlink() or mode(key) != 0o600:
|
|
findings.append({"severity": "high", "code": "tenant_key_permissions", "tenant": candidate.name, "path": str(key), "mode": oct(mode(key))})
|
|
else:
|
|
keys[candidate.name] = hashlib.sha256(key.read_bytes()).hexdigest()
|
|
values = list(keys.values())
|
|
if len(values) != len(set(values)):
|
|
findings.append({"severity": "critical", "code": "tenant_key_material_reused"})
|
|
return {"ok": not findings, "root": str(root), "tenants": tenants, "keyed_tenants": sorted(keys), "findings": findings}
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--root", required=True)
|
|
parser.add_argument("--out")
|
|
args = parser.parse_args()
|
|
result = review(Path(args.root).resolve())
|
|
payload = json.dumps(result, sort_keys=True, indent=2) + "\n"
|
|
if args.out:
|
|
Path(args.out).write_text(payload, encoding="utf8")
|
|
print(payload, end="")
|
|
return 0 if result["ok"] else 2
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|