#!/usr/bin/env python3 """Verify the effective Leo service environment without exposing its values.""" from __future__ import annotations import argparse import json import sys from collections.abc import Mapping from pathlib import Path from typing import Any PROJECT_ID = "teleo-501523" PRIVATE_CLOUDSQL_HOST = "10.61.0.3" PRIVATE_CLOUDSQL_PORT = "5432" CANONICAL_DATABASE = "teleo_canonical" RUNTIME_DATABASE_ROLE = "leoclean_kb_runtime" SCOPED_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-leoclean-kb-runtime-password" ADMINISTRATOR_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-postgres-password" RUNTIME_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" FORBIDDEN_EXACT_FIELDS = frozenset( { "ALL_PROXY", "BASH_ENV", "BASHOPTS", "CDPATH", "CURL_CA_BUNDLE", "ENV", "GLOBIGNORE", "GOOGLE_APPLICATION_CREDENTIALS", "HTTPS_PROXY", "HTTP_PROXY", "IFS", "NO_PROXY", "PS4", "REQUESTS_CA_BUNDLE", "SHELLOPTS", "SSL_CERT_DIR", "SSL_CERT_FILE", "SSLKEYLOGFILE", } ) class VerificationError(RuntimeError): """A sanitized failure safe to serialize in a public receipt.""" def __init__(self, code: str, fields: tuple[str, ...] = ()) -> None: self.code = code self.fields = tuple(sorted(set(fields))) super().__init__(code) def as_dict(self) -> dict[str, object]: payload: dict[str, object] = {"code": self.code} if self.fields: payload["fields"] = list(self.fields) return payload class _SanitizedArgumentParser(argparse.ArgumentParser): def error(self, _message: str) -> None: raise VerificationError("invalid_arguments") def validate_pid(value: str | int) -> int: """Return a positive process id without retaining the input on failure.""" if isinstance(value, bool) or not isinstance(value, (str, int)): raise VerificationError("invalid_pid") if isinstance(value, str) and (not value.isascii() or not value.isdecimal() or value.startswith("0")): raise VerificationError("invalid_pid") try: pid = int(value) except (TypeError, ValueError): raise VerificationError("invalid_pid") from None if pid <= 0: raise VerificationError("invalid_pid") return pid def expected_environment() -> dict[str, str]: """Build the exact reviewed environment contract.""" return { "PATH": RUNTIME_PATH, "TELEO_CANONICAL_CLOUDSQL_DB": CANONICAL_DATABASE, "TELEO_CLOUDSQL_DB": CANONICAL_DATABASE, "TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK": "0", "TELEO_CLOUDSQL_HOST": PRIVATE_CLOUDSQL_HOST, "TELEO_CLOUDSQL_PASSWORD_SECRET": SCOPED_PASSWORD_SECRET, "TELEO_CLOUDSQL_PORT": PRIVATE_CLOUDSQL_PORT, "TELEO_CLOUDSQL_USER": RUNTIME_DATABASE_ROLE, "TELEO_GCP_METADATA_ONLY": "1", "TELEO_GCP_PROJECT": PROJECT_ID, "TELEO_KB_MODE": "cloudsql", } def parse_environment(raw: bytes) -> dict[str, str]: """Parse a Linux ``/proc/PID/environ`` block and reject ambiguity.""" if not isinstance(raw, bytes) or not raw or not raw.endswith(b"\0"): raise VerificationError("environment_malformed") records = raw[:-1].split(b"\0") if any(not record for record in records): raise VerificationError("environment_malformed") environment: dict[str, str] = {} try: for record in records: encoded_key, separator, encoded_value = record.partition(b"=") if not separator or not encoded_key: raise VerificationError("environment_malformed") key = encoded_key.decode("utf-8", "strict") value = encoded_value.decode("utf-8", "strict") if key in environment: raise VerificationError("environment_malformed") environment[key] = value except UnicodeDecodeError: raise VerificationError("environment_malformed") from None return environment def _forbidden_field_labels(environment: Mapping[str, str]) -> tuple[str, ...]: labels: set[str] = set() for key in environment: normalized = key.upper() if normalized.startswith("PG"): labels.add("PG*") if normalized.startswith("CLOUDSDK_"): labels.add("CLOUDSDK_*") if normalized.startswith("LD_"): labels.add("LD_*") if normalized.startswith("BASH_FUNC_"): labels.add("BASH_FUNC_*") if normalized.startswith("PYTHON") and normalized != "PYTHONUNBUFFERED": labels.add("PYTHON*") if normalized in FORBIDDEN_EXACT_FIELDS: labels.add(normalized) return tuple(sorted(labels)) def validate_environment(environment: Mapping[str, str]) -> tuple[str, ...]: """Validate the effective environment and return only reviewed field names.""" if not isinstance(environment, Mapping) or any( not isinstance(key, str) or not isinstance(value, str) for key, value in environment.items() ): raise VerificationError("environment_malformed") expected = expected_environment() administrator_secret = ADMINISTRATOR_PASSWORD_SECRET.casefold() if any(administrator_secret in value.casefold() for value in environment.values()): raise VerificationError("administrator_secret_reference") forbidden_fields = _forbidden_field_labels(environment) if forbidden_fields: raise VerificationError("forbidden_environment_fields", forbidden_fields) mismatched = tuple( sorted(key for key, expected_value in expected.items() if environment.get(key) != expected_value) ) if mismatched: raise VerificationError("environment_mismatch", mismatched) return tuple(sorted(expected)) def verify_process_environment( pid: int, *, proc_root: Path = Path("/proc"), ) -> dict[str, object]: """Read and verify one process environment, returning a sanitized receipt.""" validated_pid = validate_pid(pid) raw = (proc_root / str(validated_pid) / "environ").read_bytes() validated_fields = validate_environment(parse_environment(raw)) return { "runtime_environment": "scoped", "status": "pass", "validated_fields": list(validated_fields), } def failure_receipt(error: VerificationError) -> dict[str, object]: return {"error": error.as_dict(), "status": "fail"} def canonical_json(payload: Mapping[str, Any]) -> str: return json.dumps(payload, ensure_ascii=True, separators=(",", ":"), sort_keys=True) + "\n" def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = _SanitizedArgumentParser(description=__doc__, add_help=False) parser.add_argument("--pid", required=True) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: try: args = parse_args(argv) receipt = verify_process_environment(validate_pid(args.pid)) returncode = 0 except VerificationError as exc: receipt = failure_receipt(exc) returncode = 1 except Exception: receipt = failure_receipt(VerificationError("internal_error")) returncode = 1 sys.stdout.write(canonical_json(receipt)) return returncode if __name__ == "__main__": raise SystemExit(main())