85 lines
4.5 KiB
Python
85 lines
4.5 KiB
Python
"""Privacy, artifact, and runtime health helpers for V2R1C7."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
from deployment_state import RuntimeDeploymentState
|
|
|
|
_SECRET_KEY = re.compile(r"(?i)(?:token|secret|password|private[_-]?key|api[_-]?key)")
|
|
_SECRET_VALUE = re.compile(r"(?i)(?:bearer\s+)[A-Za-z0-9._~+/=-]+|(?:token|secret|password|private[_-]?key|api[_-]?key)\s*[:=]\s*[^\s,;]+")
|
|
|
|
def redact_text(value: str) -> str:
|
|
if not isinstance(value, str): raise TypeError("text_required")
|
|
return _SECRET_VALUE.sub("[REDACTED]", value)
|
|
|
|
def redact_value(value: Any, *, key: str = "") -> Any:
|
|
if isinstance(value, Mapping):
|
|
return {str(k): ("[REDACTED]" if _SECRET_KEY.search(str(k)) else redact_value(v, key=str(k))) for k,v in value.items()}
|
|
if isinstance(value, list): return [redact_value(v, key=key) for v in value]
|
|
if isinstance(value, tuple): return [redact_value(v, key=key) for v in value]
|
|
if isinstance(value, str): return "[REDACTED]" if _SECRET_KEY.search(key) else redact_text(value)
|
|
return value
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
if path.is_symlink() or not path.is_file(): raise ValueError("regular_file_required")
|
|
digest=hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024*1024), b""): digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
def _safe_relative(root: Path, name: str) -> Path:
|
|
relative=Path(name)
|
|
if relative.is_absolute() or ".." in relative.parts or name in {"", "."}: raise ValueError("unsafe_manifest_path")
|
|
candidate=(root/relative).resolve(); candidate.relative_to(root.resolve()); return candidate
|
|
|
|
def verify_manifest(root: Path, manifest: Path | None = None) -> dict[str, Any]:
|
|
root=root.resolve(); manifest=manifest or root/"artifacts"/"MANIFEST.sha256"
|
|
if manifest.is_symlink() or not manifest.is_file(): raise ValueError("manifest_missing")
|
|
checked=0
|
|
for raw in manifest.read_text(encoding="utf-8").splitlines():
|
|
line=raw.strip()
|
|
if not line: continue
|
|
parts=line.split(maxsplit=1)
|
|
if len(parts)!=2 or not re.fullmatch(r"[0-9a-f]{64}", parts[0]): raise ValueError("manifest_entry_invalid")
|
|
expected,name=parts; name=name[1:] if name.startswith("*") else name
|
|
if sha256_file(_safe_relative(root,name))!=expected: raise ValueError("manifest_hash_mismatch")
|
|
checked+=1
|
|
if not checked: raise ValueError("manifest_empty")
|
|
return {"status":"PASS","entries":checked}
|
|
|
|
def load_release_metadata(path: Path) -> dict[str, Any]:
|
|
metadata=json.loads(path.read_text(encoding="utf-8"))
|
|
if metadata.get("version") not in {"V2R1C6","V2R1C7"}: raise ValueError("version_mismatch")
|
|
if any(metadata.get(k) is not False for k in ("production_deployed","tag_created","baseline_changed")): raise ValueError("production_change_not_allowed")
|
|
return metadata
|
|
|
|
def health_payload(state_provider: RuntimeDeploymentState | None = None) -> dict[str, Any]:
|
|
return (state_provider or RuntimeDeploymentState.from_environment()).health_payload()
|
|
|
|
class _HealthHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None:
|
|
if self.path not in {"/","/health","/healthz"}: self.send_error(404); return
|
|
body=json.dumps(health_payload(), sort_keys=True).encode()
|
|
self.send_response(200); self.send_header("Content-Type","application/json"); self.send_header("Content-Length",str(len(body))); self.end_headers(); self.wfile.write(body)
|
|
def log_message(self, _format: str, *_args: object) -> None: return
|
|
|
|
def serve(host: str, port: int) -> None:
|
|
if host not in {"127.0.0.1","::1","localhost"}: raise ValueError("loopback_only_listener_required")
|
|
if not 1024 <= port <= 65535: raise ValueError("invalid_listener_port")
|
|
server=ThreadingHTTPServer((host,port),_HealthHandler)
|
|
try: server.serve_forever()
|
|
finally: server.server_close()
|
|
|
|
def main() -> int:
|
|
parser=argparse.ArgumentParser(); parser.add_argument("--root",type=Path,default=Path(".")); parser.add_argument("--manifest",type=Path); parser.add_argument("--serve",action="store_true"); parser.add_argument("--host",default="127.0.0.1"); parser.add_argument("--port",type=int,default=18927); args=parser.parse_args()
|
|
if args.serve: serve(args.host,args.port); return 0
|
|
print(json.dumps(verify_manifest(args.root,args.manifest),sort_keys=True)); return 0
|
|
|
|
if __name__ == "__main__": raise SystemExit(main())
|