1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

#!/usr/bin/env python3
"""Create the Modal secret used by Grafana's Pocket ID login.

Pocket ID exposes public OIDC endpoint metadata through its discovery document,
but intentionally never exposes an OAuth client secret. This helper fetches the
public metadata and prompts locally for the credentials that must remain private.
Grafana also requires the separate ``atop-grafana-clickhouse-auth`` Secret and
its matching read-only ClickHouse user; this helper does not create either.
"""

import argparse
import getpass
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from urllib.parse import urlparse
from urllib.request import urlopen


DEFAULT_POCKET_ID_URL = "https://codegod100--pocket-id-serve.modal.run"
DEFAULT_GRAFANA_URL = "https://codegod100--atop-grafana-grafana.modal.run"
DEFAULT_CLICKHOUSE_HOST = "codegod100--atop-clickhouse-clickhouse.modal.run"
REQUIRED_DISCOVERY_FIELDS = {
    "authorization_endpoint": "GF_AUTH_GENERIC_OAUTH_AUTH_URL",
    "token_endpoint": "GF_AUTH_GENERIC_OAUTH_TOKEN_URL",
    "userinfo_endpoint": "GF_AUTH_GENERIC_OAUTH_API_URL",
    "jwks_uri": "GF_AUTH_GENERIC_OAUTH_JWK_SET_URL",
}


def https_url(value: str, name: str) -> str:
    parsed = urlparse(value)
    if parsed.scheme != "https" or not parsed.netloc:
        raise ValueError(f"{name} must be an HTTPS URL")
    return value.rstrip("/")


def get_value(prompt: str, environment_name: str, *, secret: bool = False) -> str:
    value = os.environ.get(environment_name)
    if not value:
        reader = getpass.getpass if secret else input
        value = reader(f"{prompt}: ").strip()
    if not value:
        raise ValueError(f"{prompt} is required")
    return value


def fetch_discovery(pocket_id_url: str) -> dict[str, str]:
    discovery_url = f"{pocket_id_url}/.well-known/openid-configuration"
    with urlopen(discovery_url, timeout=15) as response:  # noqa: S310: URL is HTTPS-validated above.
        document = json.load(response)
    missing = set(REQUIRED_DISCOVERY_FIELDS) - document.keys()
    if missing:
        raise ValueError(f"Pocket ID discovery document is missing: {', '.join(sorted(missing))}")
    return document


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--pocket-id-url", default=DEFAULT_POCKET_ID_URL)
    parser.add_argument("--grafana-url", default=DEFAULT_GRAFANA_URL)
    parser.add_argument("--clickhouse-host", default=DEFAULT_CLICKHOUSE_HOST)
    parser.add_argument("--secret-name", default="atop-grafana")
    parser.add_argument("--force", action="store_true", help="replace an existing Modal secret")
    parser.add_argument("--yes", action="store_true", help="skip the final confirmation")
    args = parser.parse_args()

    pocket_id_url = https_url(args.pocket_id_url, "Pocket ID URL")
    grafana_url = https_url(args.grafana_url, "Grafana URL")
    discovery = fetch_discovery(pocket_id_url)
    values = {
        "CLICKHOUSE_HOST": args.clickhouse_host,
        "MODAL_KEY": get_value("Modal token key (wk-...)", "MODAL_KEY", secret=True),
        "MODAL_SECRET": get_value("Modal token secret (ws-...)", "MODAL_SECRET", secret=True),
        "GF_SERVER_ROOT_URL": grafana_url,
        "GF_AUTH_GENERIC_OAUTH_CLIENT_ID": get_value("Pocket ID Grafana client ID", "GF_AUTH_GENERIC_OAUTH_CLIENT_ID"),
        "GF_AUTH_GENERIC_OAUTH_CLIENT_SECRET": get_value(
            "Pocket ID Grafana client secret", "GF_AUTH_GENERIC_OAUTH_CLIENT_SECRET", secret=True
        ),
    }
    values.update({environment_name: discovery[field] for field, environment_name in REQUIRED_DISCOVERY_FIELDS.items()})

    print(f"Creating Modal secret {args.secret_name!r} for {grafana_url}.")
    print(f"Pocket ID discovery: {pocket_id_url}/.well-known/openid-configuration")
    if not args.yes and input("Continue? [y/N] ").strip().lower() not in {"y", "yes"}:
        print("Cancelled.")
        return 0

    temporary_path: Path | None = None
    try:
        descriptor, filename = tempfile.mkstemp(prefix="atop-grafana-", suffix=".json")
        temporary_path = Path(filename)
        with os.fdopen(descriptor, "w", encoding="utf-8") as temporary_file:
            json.dump(values, temporary_file)
        command = ["modal", "secret", "create", args.secret_name, "--from-json", str(temporary_path)]
        if args.force:
            command.append("--force")
        subprocess.run(command, check=True)
    finally:
        if temporary_path:
            temporary_path.unlink(missing_ok=True)
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, subprocess.CalledProcessError) as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1)