Nexus/scripts/generate_nexus_secrets.py

#!/usr/bin/env python3
"""Generate per-host Nexus secrets (install / Docker).

Each fresh install must get unique MYSQL_* and NEXUS_* keys. Matches install wizard:
  SECRET_KEY  = token_hex(32)
  API_KEY     = token_hex(16)
  ENCRYPTION_KEY = urlsafe_b64encode(token_bytes(32))
"""

from __future__ import annotations

import argparse
import base64
import re
import secrets
import shlex
import sys
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from server.utils.text_io import read_utf8_lf, write_utf8_lf  # noqa: E402


def generate_nexus_secrets() -> dict[str, str]:
    return {
        "MYSQL_ROOT_PASSWORD": secrets.token_urlsafe(24),
        "MYSQL_PASSWORD": secrets.token_urlsafe(24),
        "NEXUS_SECRET_KEY": secrets.token_hex(32),
        "NEXUS_API_KEY": secrets.token_hex(16),
        "NEXUS_ENCRYPTION_KEY": base64.urlsafe_b64encode(secrets.token_bytes(32)).decode(),
    }


def parse_env_file(path: Path) -> dict[str, str]:
    data: dict[str, str] = {}
    if not path.is_file():
        return data
    for line in read_utf8_lf(path).splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        data[key.strip()] = value.strip().strip('"').strip("'")
    return data


def import_secrets_from_env(path: Path) -> dict[str, str]:
    """Import NEXUS + MySQL passwords from an existing .env (migration)."""
    raw = parse_env_file(path)
    out = generate_nexus_secrets()
    for key in (
        "MYSQL_ROOT_PASSWORD",
        "MYSQL_PASSWORD",
        "NEXUS_SECRET_KEY",
        "NEXUS_API_KEY",
        "NEXUS_ENCRYPTION_KEY",
    ):
        val = raw.get(key, "").strip()
        if val:
            out[key] = val
    if not out.get("NEXUS_ENCRYPTION_KEY"):
        dburl = raw.get("NEXUS_DATABASE_URL", "")
        m = re.search(r"mysql\+aiomysql://[^:]+:([^@]+)@", dburl)
        if m and not out.get("MYSQL_PASSWORD"):
            from urllib.parse import unquote_plus

            out["MYSQL_PASSWORD"] = unquote_plus(m.group(1))
    return out


def apply_profile_pools(lines: list[str], profile_env: Path) -> list[str]:
    if not profile_env.is_file():
        return lines
    prof = parse_env_file(profile_env)
    pool = prof.get("NEXUS_DB_POOL_SIZE", "")
    overflow = prof.get("NEXUS_DB_MAX_OVERFLOW", "")
    out: list[str] = []
    for line in lines:
        if pool and line.startswith("NEXUS_DB_POOL_SIZE="):
            out.append(f"NEXUS_DB_POOL_SIZE={pool}")
        elif overflow and line.startswith("NEXUS_DB_MAX_OVERFLOW="):
            out.append(f"NEXUS_DB_MAX_OVERFLOW={overflow}")
        else:
            out.append(line)
    return out


def write_env_prod(
    template: Path,
    output: Path,
    *,
    domain: str,
    secrets_map: dict[str, str],
    profile_env: Path | None,
    wizard_pending: bool,
) -> None:
    if not template.is_file():
        raise SystemExit(f"Missing template: {template}")

    lines: list[str] = []
    for line in read_utf8_lf(template).splitlines():
        key = line.split("=", 1)[0].strip() if "=" in line and not line.strip().startswith("#") else ""
        if key in secrets_map:
            lines.append(f"{key}={secrets_map[key]}")
        elif key == "NEXUS_CORS_ORIGINS":
            if domain == "localhost":
                lines.append(line)
            else:
                lines.append(f"NEXUS_CORS_ORIGINS=https://{domain}")
        elif key == "NEXUS_API_BASE_URL":
            if domain == "localhost":
                lines.append(line)
            else:
                lines.append(f"NEXUS_API_BASE_URL=https://{domain}")
        else:
            lines.append(line)

    if wizard_pending:
        found = False
        new_lines: list[str] = []
        for line in lines:
            if line.startswith("NEXUS_INSTALL_WIZARD_PENDING="):
                new_lines.append("NEXUS_INSTALL_WIZARD_PENDING=1")
                found = True
            else:
                new_lines.append(line)
        if not found:
            new_lines.append("")
            new_lines.append("# Set 0 after /app/install.html completes (init-db writes /app/.env)")
            new_lines.append("NEXUS_INSTALL_WIZARD_PENDING=1")
        lines = new_lines

    if profile_env:
        lines = apply_profile_pools(lines, profile_env)

    header = [
        f"# Generated {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC",
        "# Unique secrets for this host — do not copy to other servers",
        "",
    ]
    text = "\n".join(header + lines) + "\n"
    write_utf8_lf(output, text)
    output.chmod(0o600)


def write_secrets_backup(path: Path, secrets_map: dict[str, str], *, domain: str) -> None:
    lines = [
        f"# Nexus install secrets backup — {datetime.now(timezone.utc).isoformat()}",
        f"# Domain: {domain}",
        "# Store only on this server (chmod 600). Do NOT commit to Git.",
        "",
    ]
    for key in (
        "MYSQL_ROOT_PASSWORD",
        "MYSQL_PASSWORD",
        "NEXUS_SECRET_KEY",
        "NEXUS_API_KEY",
        "NEXUS_ENCRYPTION_KEY",
    ):
        if key in secrets_map:
            lines.append(f"{key}={secrets_map[key]}")
    write_utf8_lf(path, "\n".join(lines) + "\n")
    path.chmod(0o600)


def cmd_shell(secrets_map: dict[str, str]) -> None:
    for key, val in secrets_map.items():
        print(f"export {key}={shlex.quote(val)}")


def main() -> None:
    parser = argparse.ArgumentParser(description="Generate unique Nexus install secrets")
    sub = parser.add_subparsers(dest="cmd", required=True)

    p_shell = sub.add_parser("shell", help="Print export statements for bash eval")
    p_shell.add_argument("--from-env", type=Path, help="Import keys from existing .env")

    p_write = sub.add_parser("write-prod", help="Write docker/.env.prod from template")
    p_write.add_argument("--template", type=Path, required=True)
    p_write.add_argument("--output", type=Path, required=True)
    p_write.add_argument("--domain", required=True)
    p_write.add_argument("--profile-env", type=Path, default=None)
    p_write.add_argument("--from-env", type=Path, default=None, help="Migration: import NEXUS keys")
    p_write.add_argument(
        "--fresh",
        action="store_true",
        help="New host: always generate new secrets (ignore process env)",
    )
    p_write.add_argument("--backup", type=Path, default=None, help="Write chmod 600 backup file")
    p_write.add_argument(
        "--wizard-pending",
        action="store_true",
        help="Set NEXUS_INSTALL_WIZARD_PENDING=1 in .env.prod",
    )

    args = parser.parse_args()

    if args.cmd == "shell":
        if args.from_env:
            secrets_map = import_secrets_from_env(args.from_env)
        else:
            secrets_map = generate_nexus_secrets()
        cmd_shell(secrets_map)
        return

    if args.cmd == "write-prod":
        if args.from_env and args.from_env.is_file():
            secrets_map = import_secrets_from_env(args.from_env)
            wizard = False
        elif args.fresh:
            secrets_map = generate_nexus_secrets()
            wizard = bool(args.wizard_pending)
        else:
            secrets_map = generate_nexus_secrets()
            wizard = bool(args.wizard_pending)

        write_env_prod(
            args.template,
            args.output,
            domain=args.domain,
            secrets_map=secrets_map,
            profile_env=args.profile_env,
            wizard_pending=wizard,
        )
        if args.backup:
            write_secrets_backup(args.backup, secrets_map, domain=args.domain)
        print(f"Wrote {args.output}")
        if args.backup:
            print(f"Backup {args.backup}")


if __name__ == "__main__":
    main()