blumeops/mise-tasks/dns-acme-cleanup

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = ["httpx==0.28.1", "rich==15.0.0", "typer==0.25.0"]
# ///
#MISE description="Delete orphaned ACME challenge TXT records in eblu.me"
#USAGE flag "--dry-run" help="List orphans without deleting"
"""Clean up orphaned _acme-challenge TXT records in the eblu.me zone.

Workaround for libdns/gandi v1.1.0: its DeleteRecords compares unquoted
certmagic values to Gandi-quoted stored values, so cleanup is a silent
no-op. Without this script, the rrset grows by ~2 values per successful
Caddy renewal cycle.

In healthy steady state these records should be absent. Run alongside
PAT rotation, or any time after Caddy ACME activity.
"""

import os
import subprocess
from typing import Annotated

import httpx
import typer
from rich.console import Console
from rich.table import Table

DOMAIN = "eblu.me"
RRSET = "_acme-challenge.ops"
GANDI_API = "https://api.gandi.net/v5/livedns"
OP_PAT_REF = "op://blumeops/gandi - blumeops/pat"


def resolve_token(console: Console) -> str:
    env_token = os.environ.get("GANDI_PERSONAL_ACCESS_TOKEN", "").strip()
    if env_token:
        return env_token
    console.print("[dim]Reading Gandi PAT from 1Password...[/dim]")
    try:
        result = subprocess.run(
            ["op", "read", OP_PAT_REF],
            capture_output=True,
            text=True,
            check=True,
        )
        return result.stdout.strip()
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        console.print(f"[red]Failed to read PAT from 1Password:[/red] {e}")
        raise typer.Exit(1)


app = typer.Typer(add_completion=False)


@app.command()
def main(
    dry_run: Annotated[
        bool,
        typer.Option("--dry-run", help="List orphans without deleting"),
    ] = False,
) -> None:
    """Delete orphan _acme-challenge TXT records in eblu.me."""
    console = Console()
    token = resolve_token(console)

    url = f"{GANDI_API}/domains/{DOMAIN}/records/{RRSET}/TXT"
    headers = {"Authorization": f"Bearer {token}"}

    with httpx.Client(timeout=15, headers=headers) as client:
        resp = client.get(url)
        if resp.status_code == 404:
            console.print(
                f"[green]Clean — {RRSET}.{DOMAIN} TXT rrset is absent.[/green]"
            )
            raise typer.Exit(0)
        resp.raise_for_status()
        values = resp.json().get("rrset_values", [])

        if not values:
            console.print(
                f"[green]Clean — {RRSET}.{DOMAIN} TXT rrset is empty.[/green]"
            )
            raise typer.Exit(0)

        table = Table(title=f"Orphan ACME challenge values: {RRSET}.{DOMAIN}")
        table.add_column("#", justify="right")
        table.add_column("Value")
        for i, v in enumerate(values, 1):
            table.add_row(str(i), v)
        console.print(table)
        console.print(f"\n[bold]{len(values)}[/bold] orphan(s).")

        if dry_run:
            console.print("\n[dim]Dry run — no records deleted.[/dim]")
            raise typer.Exit(0)

        del_resp = client.delete(url)
        if del_resp.status_code == 204:
            console.print(
                f"[green]Deleted {RRSET}.{DOMAIN} TXT "
                f"({len(values)} values).[/green]"
            )
        else:
            console.print(
                f"[red]Delete failed: HTTP {del_resp.status_code}[/red]\n"
                f"{del_resp.text[:300]}"
            )
            raise typer.Exit(1)


if __name__ == "__main__":
    app()