vignette/scripts/provision-outcome-os-runtime-secrets.py

#!/usr/bin/env python3
"""Provision independently revocable Outcome OS runtime secrets.

Only the six G3-G8 internal ingestion tokens are managed. Existing unrelated
environment entries and comments are preserved, secret values are never
printed, and the file is replaced atomically after a complete candidate has
been written beside it.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import secrets
import tempfile
from pathlib import Path


MIN_TOKEN_LENGTH = 32
TOKEN_KEYS = (
    "VIGNETTE_RUPTURE_INTERNAL_TOKEN",
    "VIGNETTE_PRACTICE_INTERNAL_TOKEN",
    "VIGNETTE_CALIBRATION_TRANSFER_INTERNAL_TOKEN",
    "VIGNETTE_SUPERVISION_RESEARCH_INTERNAL_TOKEN",
    "VIGNETTE_MULTIMODAL_ALLIANCE_INTERNAL_TOKEN",
    "VIGNETTE_CONTINUOUS_IMPROVEMENT_INTERNAL_TOKEN",
)
PLACEHOLDER_FRAGMENTS = (
    "change-me",
    "changeme",
    "replace-with",
    "placeholder",
    "dummy",
    "example",
)
ASSIGNMENT = re.compile(r"^(?P<prefix>\s*(?:export\s+)?)"
                        r"(?P<key>[A-Za-z_][A-Za-z0-9_]*)\s*=.*$")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--env-file", type=Path, required=True)
    parser.add_argument(
        "--check",
        action="store_true",
        help="report whether provisioning is needed without changing the file",
    )
    return parser.parse_args()


def is_valid_token(value: str) -> bool:
    normalized = value.strip().strip('"').strip("'")
    lowered = normalized.lower()
    return len(normalized) >= MIN_TOKEN_LENGTH and not any(
        fragment in lowered for fragment in PLACEHOLDER_FRAGMENTS
    )


def assignment_value(line: str) -> str:
    return line.split("=", 1)[1].strip()


def provision_text(text: str) -> tuple[str, list[str], list[str]]:
    lines = text.splitlines()
    locations: dict[str, int] = {}
    for index, line in enumerate(lines):
        match = ASSIGNMENT.match(line)
        if match and match.group("key") in TOKEN_KEYS:
            locations[match.group("key")] = index

    updated: list[str] = []
    preserved: list[str] = []
    used_values: set[str] = set()
    for key in TOKEN_KEYS:
        index = locations.get(key)
        current = assignment_value(lines[index]) if index is not None else ""
        normalized = current.strip().strip('"').strip("'")
        if is_valid_token(current) and normalized not in used_values:
            used_values.add(normalized)
            preserved.append(key)
            continue

        candidate = secrets.token_urlsafe(48)
        while candidate in used_values:
            candidate = secrets.token_urlsafe(48)
        used_values.add(candidate)
        replacement = f"{key}={candidate}"
        if index is None:
            lines.append(replacement)
        else:
            prefix_match = ASSIGNMENT.match(lines[index])
            prefix = prefix_match.group("prefix") if prefix_match else ""
            lines[index] = f"{prefix}{replacement}"
        updated.append(key)

    rendered = "\n".join(lines)
    if rendered and not rendered.endswith("\n"):
        rendered += "\n"
    return rendered, updated, preserved


def atomic_write(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temp_name = tempfile.mkstemp(
        dir=path.parent,
        prefix=f".{path.name}.",
        suffix=".tmp",
        text=True,
    )
    temp_path = Path(temp_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream:
            stream.write(text)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temp_path, path)
    finally:
        if temp_path.exists():
            temp_path.unlink()


def main() -> int:
    args = parse_args()
    path = args.env_file.resolve()
    original = path.read_text(encoding="utf-8-sig") if path.exists() else ""
    rendered, updated, preserved = provision_text(original)
    report = {
        "ok": not updated,
        "env_file": str(path),
        "managed_keys": len(TOKEN_KEYS),
        "updated_keys": updated,
        "preserved_keys": preserved,
        "secret_values_emitted": False,
    }
    if args.check:
        print(json.dumps(report, ensure_ascii=False, sort_keys=True))
        return 0 if not updated else 1

    if updated:
        atomic_write(path, rendered)
    report["ok"] = True
    print(json.dumps(report, ensure_ascii=False, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())