thermograph/infra/.claude/skills/key-gaps/key_gaps.py

#!/usr/bin/env python3
"""Audit secret-key coverage across Thermograph environments.

Reports, per environment and across them:
  - CRITICAL gaps: a required, *self-generating* secret is missing. Missing here is
    the worst case — the app silently mints a new value on boot (AUTH_SECRET, VAPID,
    INDEXNOW), invalidating every session / push subscription.
  - Required gaps: a required secret is missing.
  - Feature gaps: a feature is *partially* configured (some keys present, some not),
    so it's silently broken. Fully-unset features are "off", not a gap.
  - Dependent-key gaps: an optional key is set but the OTHER key it needs to do
    anything isn't (e.g. a Discord channel id with no bot token) — asymmetric,
    unlike feature gaps: the prerequisite is fine set alone.
  - Drift: a key present in one environment but absent in another.

It reads only KEY NAMES, never values — safe to run anywhere. Sources per environment:
  * a SOPS-encrypted YAML (deploy/secrets/<env>.yaml) — keys are plaintext even when
    encrypted, so no age key or decryption is needed; or
  * a plain key-list file (one KEY per line, or KEY=... lines) — e.g. the output of
    `ssh <box> 'grep -oE "^[A-Z][A-Z0-9_]*=" /etc/thermograph.env'` for a live audit.

Usage:
  key_gaps.py prod=deploy/secrets/prod.yaml beta=deploy/secrets/beta.yaml
  key_gaps.py prod=/tmp/prod.keys           # a live key-list gathered over SSH
Exit status is non-zero if any CRITICAL or required gap is found (usable in CI).
"""
from __future__ import annotations

import re
import sys

# --- what counts as a secret/key, and how it behaves --------------------------
# category: "required" (app needs it) | "optional" (fine to omit)
# self_gen: missing => the app silently generates a replacement (data-loss hazard)
REQUIRED = {
    # key: self_generating?
    "POSTGRES_PASSWORD": False,
    "THERMOGRAPH_AUTH_SECRET": True,
    "THERMOGRAPH_VAPID_PRIVATE_KEY": True,
    "THERMOGRAPH_VAPID_PUBLIC_KEY": True,
    "REGISTRY_TOKEN": False,
}
OPTIONAL_SELF_GEN = {
    "THERMOGRAPH_INDEXNOW_KEY": True,   # regenerates, but persisted to a file — low risk
}
# Features that only work if ALL their keys are set together; some-but-not-all = broken.
FEATURE_GROUPS = {
    "discord-account-linking": ["THERMOGRAPH_DISCORD_APP_ID", "THERMOGRAPH_DISCORD_CLIENT_SECRET"],
    "mail-smtp": ["THERMOGRAPH_SMTP_USER", "THERMOGRAPH_SMTP_PASSWORD"],
}
# Optional keys that only do anything when a specific OTHER key is also set. Unlike
# FEATURE_GROUPS the relationship is asymmetric: the prerequisite is fine set alone
# (THERMOGRAPH_DISCORD_BOT_TOKEN alone already enables DMs), but the dependent key
# alone is a silent no-op (discord.py gates both channel posts on
# `bool(BOT_TOKEN and <channel>)`) — flagging it as a symmetric feature group would
# false-positive on every environment that has the bot token for DMs only.
DEPENDENT_OPTIONAL = {
    "THERMOGRAPH_DISCORD_WEATHER_CHANNEL": "THERMOGRAPH_DISCORD_BOT_TOKEN",
    "THERMOGRAPH_DISCORD_SUBSCRIPTION_CHANNEL": "THERMOGRAPH_DISCORD_BOT_TOKEN",
    # Gateway-bot enable flag: truthy alone does nothing (web/app.py only starts the
    # bot when discord_bot.enabled() sees the token too). One gateway connection per
    # token — set in exactly one environment.
    "THERMOGRAPH_DISCORD_BOT": "THERMOGRAPH_DISCORD_BOT_TOKEN",
}
# Standalone optional keys: reported present/absent, never a "gap" on their own.
STANDALONE_OPTIONAL = [
    "THERMOGRAPH_METRICS_TOKEN",        # ops metrics remote access (else loopback-only, not broken)
    "THERMOGRAPH_DISCORD_WEBHOOK",      # daily post
    "THERMOGRAPH_DISCORD_PUBLIC_KEY",   # slash-command interactions
    "THERMOGRAPH_DISCORD_BOT_TOKEN",    # bot DMs / gateway bot (prerequisite for the two below)
]

C = {"red": "\033[31m", "yellow": "\033[33m", "green": "\033[32m",
     "dim": "\033[2m", "bold": "\033[1m", "off": "\033[0m"}


def keys_from_source(path: str) -> set[str]:
    """Extract KEY names from a SOPS yaml or a plain key-list file (values ignored)."""
    with open(path, encoding="utf-8", errors="replace") as fh:
        text = fh.read()
    keys: set[str] = set()
    in_sops_block = False
    for raw in text.splitlines():
        line = raw.rstrip("\n")
        if line.startswith("sops:"):          # SOPS metadata block — skip its subkeys
            in_sops_block = True
            continue
        if in_sops_block:
            if line and not line[0].isspace():
                in_sops_block = False           # dedented back to a real key
            else:
                continue
        s = line.strip()
        if not s or s.startswith("#"):
            continue
        # Accept "KEY: ..." (yaml), "KEY=..." (dotenv), or a bare "KEY" (key-list).
        # The whole stripped line must be just an identifier optionally followed by
        # a : or = — so prose/nested lines don't get mistaken for keys.
        m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)\s*(?:[:=].*)?$", s)
        if m:
            keys.add(m.group(1))
    return keys


def audit(envs: dict[str, set[str]]) -> int:
    names = list(envs)
    print(f"{C['bold']}Key-gap audit — {', '.join(names)}{C['off']}")
    for n in names:
        print(f"  {C['dim']}{n}: {len(envs[n])} keys{C['off']}")
    print()
    critical = required = feature = dependent = 0

    def present(env, key):  # noqa: ANN001
        return key in envs[env]

    # Required + self-generating
    print(f"{C['bold']}Required secrets{C['off']}")
    for key, self_gen in {**REQUIRED, **OPTIONAL_SELF_GEN}.items():
        req = key in REQUIRED
        missing = [n for n in names if not present(n, key)]
        if not missing:
            print(f"  {C['green']}OK{C['off']}   {key}  (all environments)")
            continue
        if req and self_gen:
            critical += 1
            tag = f"{C['red']}CRITICAL{C['off']}"
            note = " — self-generates on boot: MISSING = every session/subscription invalidated"
        elif req:
            required += 1
            tag = f"{C['red']}MISSING {C['off']}"
            note = ""
        else:
            tag = f"{C['yellow']}note    {C['off']}"
            note = " — self-generates (persisted to a file; low risk)"
        print(f"  {tag} {key}  absent in: {', '.join(missing)}{C['dim']}{note}{C['off']}")
    print()

    # Feature groups — partial config = broken
    print(f"{C['bold']}Feature groups (all-or-nothing){C['off']}")
    for feat, keys in FEATURE_GROUPS.items():
        for n in names:
            have = [k for k in keys if present(n, k)]
            if not have:
                print(f"  {C['dim']}off  {feat} @ {n} (none set){C['off']}")
            elif len(have) == len(keys):
                print(f"  {C['green']}OK{C['off']}   {feat} @ {n} (enabled)")
            else:
                feature += 1
                miss = [k for k in keys if not present(n, k)]
                print(f"  {C['red']}GAP {C['off']} {feat} @ {n}: has {have}, MISSING {miss}")
    print()

    # Dependent optional — the key only does anything with its prerequisite also set
    print(f"{C['bold']}Dependent optional keys{C['off']}")
    for key, prereq in DEPENDENT_OPTIONAL.items():
        for n in names:
            if not present(n, key):
                print(f"  {C['dim']}off  {key} @ {n} (unset){C['off']}")
            elif present(n, prereq):
                print(f"  {C['green']}OK{C['off']}   {key} @ {n} (enabled, {prereq} set)")
            else:
                dependent += 1
                print(f"  {C['red']}GAP {C['off']} {key} @ {n}: set but {prereq} is missing — silently no-ops")
    print()

    # Standalone optional
    print(f"{C['bold']}Optional keys{C['off']}")
    for key in STANDALONE_OPTIONAL:
        where = [n for n in names if present(n, key)]
        state = f"set in: {', '.join(where)}" if where else "unset everywhere"
        print(f"  {C['dim']}{key}: {state}{C['off']}")
    print()

    # Cross-env drift
    if len(names) > 1:
        print(f"{C['bold']}Cross-environment drift{C['off']}")
        union = set().union(*envs.values())
        drift = False
        for key in sorted(union):
            where = [n for n in names if key in envs[n]]
            if len(where) != len(names) and (key in REQUIRED or key in OPTIONAL_SELF_GEN
                                             or any(key in g for g in FEATURE_GROUPS.values())
                                             or key in DEPENDENT_OPTIONAL
                                             or key in STANDALONE_OPTIONAL):
                drift = True
                absent = [n for n in names if n not in where]
                print(f"  {C['yellow']}drift{C['off']} {key}: in {', '.join(where)}; not in {', '.join(absent)}")
        if not drift:
            print(f"  {C['green']}none{C['off']} (tracked keys are consistent)")
        print()

    total = critical + required + feature + dependent
    color = C["red"] if total else C["green"]
    print(f"{color}{C['bold']}Summary: {critical} critical, {required} required, "
          f"{feature} feature gaps, {dependent} dependent-key gaps{C['off']}")
    return 1 if (critical or required) else 0


def main(argv: list[str]) -> int:
    args = [a for a in argv[1:] if "=" in a]
    if not args:
        print(__doc__)
        return 2
    envs: dict[str, set[str]] = {}
    for a in args:
        name, _, path = a.partition("=")
        try:
            envs[name] = keys_from_source(path)
        except OSError as e:
            print(f"!! {name}: cannot read {path}: {e}", file=sys.stderr)
            return 2
    return audit(envs)


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))