scripts/final_env_wizard.py

#!/usr/bin/env python3
"""Create a local final live-proof env file without printing secrets."""

from __future__ import annotations

import argparse
from dataclasses import dataclass
import getpass
import json
import os
from pathlib import Path
import re
import subprocess
from typing import Callable, Mapping
from urllib.parse import urlparse


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_OUTPUT = ROOT / ".env.final.local"
DEFAULT_B2_SETUP = ROOT / "docs" / "assets" / "b2-live-setup.json"
SAFE_ENV_RE = re.compile(r"^[A-Za-z0-9_./:@%+=,-]+$")
PLACEHOLDERS = {"", "placeholder", "change-me", "changeme", "todo", "tbd", "none", "null", "..."}

NEXT_COMMANDS = [
    "python scripts/live_env_handoff.py --env-file .env.final.local --strict",
    "python scripts/run_final_live_proof.py --env-file .env.final.local --preflight-only",
    "python scripts/run_final_live_proof.py --env-file .env.final.local --genblaze-provider local --genblaze-image-model local-svg-v1 --preflight-only",
    "python scripts/run_final_live_proof.py --env-file .env.final.local --genblaze-provider openai --genblaze-image-model gpt-image-1 --preflight-only",
    "python scripts/run_final_live_proof.py --env-file .env.final.local --evidence-out docs/assets/final-live-proof-evidence.json",
]


@dataclass(frozen=True)
class EnvField:
    name: str
    prompt: str
    default: str = ""
    required: bool = True
    secret: bool = False
    aliases: tuple[str, ...] = ()
    mirror_to: tuple[str, ...] = ()


FIELDS = [
    EnvField("PROOFFRAME_STORAGE_BACKEND", "ProofFrame storage backend", default="b2"),
    EnvField("PROOFFRAME_GENERATION_BACKEND", "ProofFrame generation backend", default="genblaze"),
    EnvField("B2_ENDPOINT_URL", "Backblaze B2 S3 endpoint URL"),
    EnvField("B2_REGION", "Backblaze B2 region", required=False),
    EnvField("B2_BUCKET", "Backblaze B2 bucket name"),
    EnvField("B2_KEY_ID", "Backblaze B2 application key id"),
    EnvField(
        "B2_APPLICATION_KEY",
        "Backblaze B2 application key",
        secret=True,
        aliases=("B2_APP_KEY",),
        mirror_to=("B2_APP_KEY",),
    ),
    EnvField("B2_PUBLIC_BASE_URL", "Optional B2 public base URL", required=False),
    EnvField("GENBLAZE_PROVIDER", "Genblaze provider", default="gmicloud", required=False),
    EnvField(
        "GENBLAZE_API_KEY",
        "Genblaze provider API key",
        secret=True,
        aliases=("GMI_API_KEY",),
    ),
    EnvField("GMI_API_KEY", "Optional GMI API key alias", required=False, secret=True),
    EnvField("OPENAI_API_KEY", "Optional OpenAI API key alias", required=False, secret=True),
    EnvField("GENBLAZE_IMAGE_MODEL", "Genblaze image model", default="seedream-5.0-lite"),
    EnvField("GENBLAZE_ASPECT_RATIO", "Genblaze aspect ratio", default="16:9", required=False),
    EnvField("GENBLAZE_TIMEOUT_SECONDS", "Genblaze timeout seconds", default="180", required=False),
]


def normalize(value: str | None) -> str:
    return (value or "").strip()


def has_real_value(value: str | None) -> bool:
    normalized = normalize(value).strip('"').strip("'")
    if normalized.lower() in PLACEHOLDERS:
        return False
    if normalized.startswith("<") and normalized.endswith(">"):
        return False
    return bool(normalized)


def env_lookup(field: EnvField, environ: Mapping[str, str]) -> str:
    for name in (field.name, *field.aliases):
        value = normalize(environ.get(name))
        if has_real_value(value):
            return value
    return ""


def parse_env_file(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        if key.startswith("export "):
            key = key.removeprefix("export ").strip()
        if key:
            values[key] = normalize(value).strip('"').strip("'")
    return values


def default_source(
    environ: Mapping[str, str],
    initial_values: Mapping[str, str] | None,
    *,
    environ_overrides: bool,
) -> dict[str, str]:
    values = dict(initial_values or {})
    if environ_overrides:
        values.update(environ)
    else:
        env_values = dict(environ)
        env_values.update(values)
        values = env_values
    return values


def prompt_for_field(
    field: EnvField,
    default: str,
    *,
    input_func: Callable[[str], str],
    secret_input: Callable[[str], str],
) -> str:
    if field.secret:
        suffix = " [set; press Enter to keep]" if default else ""
        value = secret_input(f"{field.prompt}{suffix}: ")
        return normalize(value) or default

    suffix = f" [{default}]" if default else ""
    value = input_func(f"{field.prompt}{suffix}: ")
    return normalize(value) or default


def collect_values(
    *,
    from_env: bool,
    environ: Mapping[str, str] | None = None,
    initial_values: Mapping[str, str] | None = None,
    input_func: Callable[[str], str] = input,
    secret_input: Callable[[str], str] = getpass.getpass,
) -> dict[str, str]:
    default_environ: Mapping[str, str] = os.environ if from_env else {}
    source = default_source(
        default_environ if environ is None else environ,
        initial_values,
        environ_overrides=from_env,
    )
    values: dict[str, str] = {}
    missing: list[str] = []

    for field in FIELDS:
        default = env_lookup(field, source) or field.default
        value = default if from_env else prompt_for_field(
            field,
            default,
            input_func=input_func,
            secret_input=secret_input,
        )
        if field.required and not value:
            missing.append(field.name)
            continue
        if value:
            values[field.name] = value
            for mirror_name in field.mirror_to:
                values[mirror_name] = value

    if missing:
        raise ValueError("Missing required final env values: " + ", ".join(missing))
    return values


def missing_required_fields(values: Mapping[str, str]) -> list[EnvField]:
    missing: list[EnvField] = []
    for field in FIELDS:
        if field.required and not env_lookup(field, values):
            missing.append(field)
    return missing


def collect_missing_values(
    *,
    initial_values: Mapping[str, str],
    environ: Mapping[str, str] | None = None,
    input_func: Callable[[str], str] = input,
    secret_input: Callable[[str], str] = getpass.getpass,
) -> tuple[dict[str, str], list[str]]:
    values = dict(initial_values)
    for field in missing_required_fields(values):
        env_value = env_lookup(field, environ or {})
        if env_value:
            values[field.name] = env_value
            for mirror_name in field.mirror_to:
                values[mirror_name] = env_value
    filled_names: list[str] = []

    for field in missing_required_fields(values):
        default = field.default
        value = prompt_for_field(
            field,
            default,
            input_func=input_func,
            secret_input=secret_input,
        )
        if not value:
            continue
        values[field.name] = value
        filled_names.append(field.name)
        for mirror_name in field.mirror_to:
            values[mirror_name] = value
            filled_names.append(mirror_name)

    still_missing = [field.name for field in missing_required_fields(values)]
    if still_missing:
        raise ValueError("Missing required final env values: " + ", ".join(still_missing))
    return values, filled_names


def load_b2_setup(path: Path = DEFAULT_B2_SETUP) -> dict[str, str]:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError):
        return {}
    if not data.get("safe_to_commit"):
        return {}
    endpoint = normalize(str(data.get("endpoint", "")))
    return {
        "PROOFFRAME_STORAGE_BACKEND": "b2",
        "B2_ENDPOINT_URL": endpoint,
        "B2_REGION": region_from_b2_endpoint(endpoint),
        "B2_BUCKET": normalize(str(data.get("bucket_name", ""))),
    }


def region_from_b2_endpoint(endpoint_url: str) -> str:
    if not endpoint_url:
        return ""
    parsed = urlparse(endpoint_url if "://" in endpoint_url else f"https://{endpoint_url}")
    host = parsed.netloc or parsed.path
    match = re.match(r"^s3[.-]([a-z0-9-]+)\.backblazeb2\.com$", host)
    return match.group(1) if match else ""


def non_secret_prefill_values(*, b2_setup_path: Path = DEFAULT_B2_SETUP) -> dict[str, str]:
    values = {
        "PROOFFRAME_STORAGE_BACKEND": "b2",
        "PROOFFRAME_GENERATION_BACKEND": "genblaze",
        "GENBLAZE_PROVIDER": "gmicloud",
        "GENBLAZE_IMAGE_MODEL": "seedream-5.0-lite",
        "GENBLAZE_ASPECT_RATIO": "16:9",
        "GENBLAZE_TIMEOUT_SECONDS": "180",
    }
    values.update({key: value for key, value in load_b2_setup(b2_setup_path).items() if value})
    return values


def quote_env_value(value: str) -> str:
    if SAFE_ENV_RE.fullmatch(value):
        return value
    return "'" + value.replace("'", "'\"'\"'") + "'"


def ordered_names() -> list[str]:
    names: list[str] = []
    for field in FIELDS:
        names.append(field.name)
        names.extend(field.mirror_to)
    return names


def render_env_file(values: Mapping[str, str]) -> str:
    lines = [
        "# Generated by scripts/final_env_wizard.py.",
        "# Local secret file for final live proof. Do not commit.",
        "",
    ]
    for name in ordered_names():
        value = values.get(name, "")
        lines.append(f"{name}={quote_env_value(value)}")
    return "\n".join(lines) + "\n"


def render_env_file_preserving_extra(values: Mapping[str, str]) -> str:
    rendered_names = set(ordered_names())
    content = render_env_file(values).rstrip("\n")
    extra_lines = [
        f"{name}={quote_env_value(value)}"
        for name, value in values.items()
        if name not in rendered_names
    ]
    if extra_lines:
        content = "\n".join([content, "", "# Preserved local-only values.", *extra_lines])
    return content + "\n"


def path_for_git(root: Path, path: Path) -> str:
    try:
        return str(path.resolve().relative_to(root.resolve()))
    except ValueError:
        return str(path)


def fallback_gitignore_check(root: Path, path: Path) -> bool:
    gitignore = root / ".gitignore"
    if not gitignore.exists():
        return False
    name = path.name
    ignored_by_env_glob = False
    explicitly_unignored = False
    for raw_line in gitignore.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line == ".env.*" and name.startswith(".env."):
            ignored_by_env_glob = True
        if line.startswith("!") and line.removeprefix("!") == name:
            explicitly_unignored = True
    return ignored_by_env_glob and not explicitly_unignored


def target_is_git_ignored(root: Path, path: Path) -> bool:
    relative = path_for_git(root, path)
    try:
        completed = subprocess.run(
            ["git", "check-ignore", "--quiet", "--", relative],
            cwd=root,
            check=False,
        )
    except FileNotFoundError:
        return fallback_gitignore_check(root, path)
    if completed.returncode == 0:
        return True
    return fallback_gitignore_check(root, path)


def build_check(root: Path = ROOT, output: Path = DEFAULT_OUTPUT) -> dict[str, object]:
    output = output.resolve()
    ignored = target_is_git_ignored(root, output)
    env_file_present = output.exists()
    values = parse_env_file(output) if env_file_present else {}
    missing_required_names = [
        field.name for field in FIELDS if field.required and not env_lookup(field, values)
    ]
    placeholder_names: list[str] = []
    seen_placeholders: set[str] = set()
    for field in FIELDS:
        if not (field.required or field.secret):
            continue
        for name in (field.name, *field.aliases, *field.mirror_to):
            if name in values and not has_real_value(values.get(name)) and name not in seen_placeholders:
                placeholder_names.append(name)
                seen_placeholders.add(name)
    mode = output.stat().st_mode & 0o777 if env_file_present else None
    mode_ok = mode == 0o600 if env_file_present else False
    b2_endpoint = env_lookup(next(field for field in FIELDS if field.name == "B2_ENDPOINT_URL"), values)
    b2_region_value = env_lookup(next(field for field in FIELDS if field.name == "B2_REGION"), values)
    b2_region_derived = region_from_b2_endpoint(b2_endpoint)
    b2_region_ok = bool(b2_region_value or b2_region_derived)
    return {
        "schema": "proofframe.final_env_wizard_check.v1",
        "ok": ignored,
        "output": str(output),
        "git_ignored": ignored,
        "env_file_present": env_file_present,
        "env_file_mode": oct(mode) if mode is not None else None,
        "env_file_mode_ok": mode_ok,
        "required_fields": [field.name for field in FIELDS if field.required],
        "secret_fields": [
            name
            for field in FIELDS
            if field.secret
            for name in (field.name, *field.aliases, *field.mirror_to)
        ],
        "present_required_names": [
            field.name for field in FIELDS if field.required and env_lookup(field, values)
        ],
        "missing_required_names": missing_required_names,
        "placeholder_names": placeholder_names,
        "b2_region": {
            "present": bool(b2_region_value),
            "derived_from_endpoint": b2_region_derived,
            "ok": b2_region_ok,
        },
        "ready_for_live_entry": bool(
            ignored and env_file_present and mode_ok and not missing_required_names and b2_region_ok
        ),
        "next_commands": NEXT_COMMANDS,
        "policy": (
            "The wizard writes only to a git-ignored local env file and command output lists "
            "variable names only, never credential values."
        ),
    }


def write_env_file(path: Path, content: str, *, force: bool) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    flags = os.O_WRONLY | os.O_CREAT
    flags |= os.O_TRUNC if force else os.O_EXCL
    try:
        fd = os.open(path, flags, 0o600)
    except FileExistsError as exc:
        raise FileExistsError(
            f"{path} already exists. Pass --force only after confirming it is safe to overwrite."
        ) from exc
    with os.fdopen(fd, "w", encoding="utf-8") as handle:
        handle.write(content)
    os.chmod(path, 0o600)


def build_success(output: Path, values: Mapping[str, str]) -> dict[str, object]:
    return {
        "ok": True,
        "mode": "final_env_written",
        "output": str(output),
        "chmod": "0600",
        "written_names": [name for name in ordered_names() if name in values],
        "next_commands": NEXT_COMMANDS,
        "secret_policy": "Credential values were written locally and were not printed.",
    }


def build_missing_only_success(
    output: Path,
    values: Mapping[str, str],
    filled_names: list[str],
) -> dict[str, object]:
    return {
        "ok": True,
        "mode": "final_env_missing_values_filled",
        "output": str(output),
        "chmod": "0600",
        "filled_names": filled_names,
        "written_names": [name for name in ordered_names() if name in values],
        "next_commands": NEXT_COMMANDS,
        "secret_policy": "Only missing local values were prompted and written; credential values were not printed.",
    }


def build_prefill_success(output: Path, values: Mapping[str, str]) -> dict[str, object]:
    missing_secret_names = [
        name
        for field in FIELDS
        if field.secret
        for name in (field.name, *field.mirror_to)
        if name not in values
    ]
    missing_required_names = [
        field.name for field in FIELDS if field.required and field.name not in values
    ]
    return {
        "ok": True,
        "mode": "final_env_non_secret_prefilled",
        "output": str(output),
        "chmod": "0600",
        "written_names": [name for name in ordered_names() if name in values],
        "missing_required_names": missing_required_names,
        "missing_secret_names": missing_secret_names,
        "next_commands": [
            "python scripts/final_env_wizard.py --output .env.final.local --missing-only --force",
            *NEXT_COMMANDS,
        ],
        "secret_policy": "Only non-secret values were written. Credential values were not printed or stored.",
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Safely create .env.final.local for ProofFrame live proof."
    )
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument("--force", action="store_true", help="Overwrite an existing output file.")
    parser.add_argument(
        "--from-env",
        action="store_true",
        help="Read values from the current process environment instead of interactive prompts.",
    )
    parser.add_argument(
        "--check-only",
        action="store_true",
        help="Validate the output target is git-ignored without prompting or writing secrets.",
    )
    parser.add_argument(
        "--prefill-non-secret",
        action="store_true",
        help="Write known non-secret B2/default values to the local env file and leave secrets empty.",
    )
    parser.add_argument(
        "--missing-only",
        action="store_true",
        help="Prompt only for missing required values, preserving existing local values.",
    )
    parser.add_argument("--b2-setup", type=Path, default=DEFAULT_B2_SETUP)
    return parser


def main() -> None:
    args = build_parser().parse_args()
    check = build_check(ROOT, args.output)
    if args.check_only:
        print(json.dumps(check, indent=2))
        raise SystemExit(0 if check["ok"] else 2)
    if not check["ok"]:
        print(json.dumps(check, indent=2))
        raise SystemExit(2)
    if args.prefill_non_secret:
        values = non_secret_prefill_values(b2_setup_path=args.b2_setup)
        try:
            write_env_file(args.output, render_env_file(values), force=args.force)
        except FileExistsError as exc:
            print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
            raise SystemExit(2) from exc
        print(json.dumps(build_prefill_success(args.output, values), indent=2))
        raise SystemExit(0)

    try:
        initial_values = parse_env_file(args.output) if args.output.exists() else {}
        if args.missing_only:
            values, filled_names = collect_missing_values(
                initial_values=initial_values,
                environ=os.environ if args.from_env else None,
            )
            write_env_file(args.output, render_env_file_preserving_extra(values), force=args.force)
            print(json.dumps(build_missing_only_success(args.output, values, filled_names), indent=2))
            raise SystemExit(0)
        values = collect_values(from_env=args.from_env, initial_values=initial_values)
        write_env_file(args.output, render_env_file(values), force=args.force)
    except (FileExistsError, ValueError) as exc:
        print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
        raise SystemExit(2) from exc
    print(json.dumps(build_success(args.output, values), indent=2))


if __name__ == "__main__":
    main()