lampac/scripts/config-secrets.py

#!/usr/bin/env python3
"""Вынос секретов Lampac из git: шаблон config/init.conf <-> боевой config/init.custom.conf.

В git (config/init.conf) секреты всегда ПУСТЫЕ — это делает рабочее дерево чистым,
поэтому git pull/merge никогда не конфликтует и секреты не уезжают в remote.
Для сервера собирается config/init.custom.conf (шаблон + config/secrets.env),
он монтируется в контейнер вместо init.conf (см. docker-compose.yaml) и
в .gitignore.

Команды:
  apply   config/init.conf + config/secrets.env -> config/init.custom.conf  (после git pull)
  export  config/init.custom.conf -> config/init.conf, секреты обнуляются    (перед git add)
  check   проверяет, что в config/init.conf (или в staged-версии) нет секретов
  status  показывает, что где лежит

Обычно руками ничего не нужно: scripts/install-git-hooks.sh вешает
post-merge/post-checkout на apply и pre-commit на check.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TEMPLATE = os.path.join(ROOT, "config", "init.conf")
LIVE = os.path.join(ROOT, "config", "init.custom.conf")
SECRETS = os.path.join(ROOT, "config", "secrets.env")

# (секция, поле) в init.conf  ->  ключ в config/secrets.env
MAP = [
    (("TelegramBot", "bot_token"), "TELEGRAM_NOTIFY_BOT_TOKEN"),
    (("TelegramAuthBot", "bot_token"), "TELEGRAM_AUTH_BOT_TOKEN"),
    (("TelegramAuth", "mutations_api_secret"), "TELEGRAM_AUTH_MUTATIONS_SECRET"),
    (("TelegramAuthBot", "mutations_api_secret"), "TELEGRAM_AUTH_MUTATIONS_SECRET"),
]


def load_secrets() -> dict[str, str] | None:
    """None, если secrets.env отсутствует (тогда apply просто предупреждает)."""
    if not os.path.exists(SECRETS):
        return None
    out: dict[str, str] = {}
    for line in open(SECRETS, encoding="utf-8"):
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        out[k.strip()] = v.strip()
    return out


def load_json(path: str) -> dict:
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def save_json(path: str, data: dict) -> None:
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
        f.write("\n")
    os.replace(tmp, path)


def cmd_apply() -> None:
    """Шаблон + секреты -> боевой файл, который монтируется в контейнер."""
    tpl = load_json(TEMPLATE)
    sec = load_secrets()
    if sec is None:
        print(f"apply: нет {os.path.relpath(SECRETS, ROOT)} — "
              f"{os.path.relpath(LIVE, ROOT)} не создан (секреты останутся пустыми)")
        return
    missing = []
    for (section, field), key in MAP:
        value = sec.get(key, "")
        if not value:
            missing.append(key)
        if section in tpl:
            tpl[section][field] = value
    save_json(LIVE, tpl)
    # контейнер пишет в боевой конфиг через AdminPanel (uid 1000)
    try:
        os.chown(LIVE, 1000, 1000)
    except PermissionError:
        pass
    os.chmod(LIVE, 0o644)
    print(f"apply: {os.path.relpath(LIVE, ROOT)} собран из шаблона + secrets.env")
    if missing:
        print(f"  внимание: нет значения для {', '.join(sorted(set(missing)))} — поле останется пустым")


def cmd_export() -> None:
    """Боевой файл -> шаблон, секреты обнуляются (для git add)."""
    if not os.path.exists(LIVE):
        print(f"export: {os.path.relpath(LIVE, ROOT)} нет — нечего экспортировать")
        return
    live = load_json(LIVE)
    changed = 0
    for (section, field), _ in MAP:
        if section in live and live.get(section, {}).get(field):
            live[section][field] = ""
            changed += 1
    save_json(TEMPLATE, live)
    print(f"export: шаблон обновлён из боевого конфига, секретов обнулено: {changed}")


def read_template_for_check(staged: bool) -> str:
    if staged:
        try:
            return subprocess.check_output(
                ["git", "show", ":config/init.conf"], cwd=ROOT, text=True
            )
        except subprocess.CalledProcessError:
            sys.exit("check: config/init.conf не в индексе")
    return open(TEMPLATE, encoding="utf-8").read()


def cmd_check(staged: bool = False) -> None:
    """Падает (exit 1), если в шаблоне есть непустой секрет."""
    data = json.loads(read_template_for_check(staged))
    found = []
    for (section, field), key in MAP:
        value = (data.get(section) or {}).get(field)
        if value:
            found.append(f"{section}.{field} ({key}) = {value[:6]}…")
    if found:
        sys.exit(
            "check: ОТКАЗ — в config/init.conf есть секреты, коммитить нельзя:\n  "
            + "\n  ".join(found)
            + "\nВыполните: python3 scripts/config-secrets.py export"
        )
    print("check: секретов в config/init.conf нет — ок")


def cmd_status() -> None:
    tpl = load_json(TEMPLATE)
    live = load_json(LIVE) if os.path.exists(LIVE) else None
    print(f"{'путь':28} {'есть?' :6} секреты")
    for (section, field), key in MAP:
        t = (tpl.get(section) or {}).get(field)
        l = (live.get(section) or {}).get(field) if live else None
        print(
            f"{section}.{field:24} шаблон={'заполнен' if t else 'пусто':8} "
            f"боевой={'заполнен' if l else ('нет файла' if live is None else 'пусто')}"
        )
    print(f"secrets.env: {'есть' if os.path.exists(SECRETS) else 'НЕТ'}")


def main() -> None:
    args = sys.argv[1:]
    cmd = args[0] if args else "status"
    if cmd == "apply":
        cmd_apply()
    elif cmd == "export":
        cmd_export()
    elif cmd == "check":
        cmd_check(staged="--staged" in args)
    elif cmd == "status":
        cmd_status()
    else:
        sys.exit(__doc__)


if __name__ == "__main__":
    main()