dev-ops-agent/scripts/update_volcengine_cdn_cert_from_tencent.py

#!/usr/bin/env python3
"""Download a Tencent Cloud SSL cert and deploy it to a Volcengine CDN domain."""

from __future__ import annotations

import argparse
import base64
import json
import os
import shutil
import ssl
import subprocess
import sys
import tempfile
import zipfile
from pathlib import Path

import import_volcengine_certificate


def run_json(command: list[str]) -> dict:
    completed = subprocess.run(
        command,
        check=True,
        text=True,
        capture_output=True,
        encoding="utf-8",
        errors="replace",
    )
    return json.loads(completed.stdout)


def run_passthrough(command: list[str]) -> None:
    subprocess.run(command, check=True)


def find_tccli() -> str:
    tccli = shutil.which("tccli")
    if tccli:
        return tccli

    appdata = os.environ.get("APPDATA")
    if appdata:
        fallback = Path(appdata) / "Python" / "Python313" / "Scripts" / "tccli.exe"
        if fallback.exists():
            return str(fallback)

    raise RuntimeError("tccli was not found in PATH or the Python313 user Scripts directory")


def first_pem_certificate(bundle_text: str) -> str:
    begin = "-----BEGIN CERTIFICATE-----"
    end = "-----END CERTIFICATE-----"
    start = bundle_text.find(begin)
    stop = bundle_text.find(end, start)
    if start == -1 or stop == -1:
        raise RuntimeError("No PEM certificate block found in bundle")
    return bundle_text[start : stop + len(end)] + "\n"


def decode_leaf_certificate(bundle_path: Path) -> dict[str, str]:
    leaf_text = first_pem_certificate(bundle_path.read_text(encoding="utf-8"))
    with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".pem", delete=False) as handle:
        handle.write(leaf_text)
        leaf_path = handle.name
    try:
        decoded = ssl._ssl._test_decode_cert(leaf_path)  # noqa: SLF001 - stdlib has no public equivalent.
    finally:
        Path(leaf_path).unlink(missing_ok=True)

    return {
        "subject": "/".join("=".join(part) for row in decoded.get("subject", []) for part in row),
        "not_before": decoded.get("notBefore", ""),
        "not_after": decoded.get("notAfter", ""),
        "serial_number": decoded.get("serialNumber", ""),
    }


def download_tencent_certificate(tccli: str, cert_id: str, target_dir: Path) -> None:
    result = run_json([tccli, "ssl", "DownloadCertificate", "--CertificateId", cert_id])
    content = base64.b64decode(result["Content"])
    zip_path = target_dir / f"{cert_id}.zip"
    zip_path.write_bytes(content)
    with zipfile.ZipFile(zip_path) as archive:
        archive.extractall(target_dir)
    zip_path.unlink()


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--domain", required=True)
    parser.add_argument("--tencent-certificate-id", required=True)
    parser.add_argument("--volcengine-profile", default="intelligrow")
    parser.add_argument("--volcengine-profile-prefix", default="INTELLIGROW")
    parser.add_argument("--region", default="cn-beijing")
    parser.add_argument("--project-name", default="default")
    parser.add_argument("--env-path", default=".env")
    parser.add_argument("--keep-temp", action="store_true")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    temp_root = Path(tempfile.mkdtemp(prefix="volc-cert-update-"))
    try:
        domain_dir = temp_root / args.domain
        domain_dir.mkdir()
        tccli = find_tccli()

        print(f"Downloading Tencent Cloud certificate {args.tencent_certificate_id} for {args.domain}")
        download_tencent_certificate(tccli, args.tencent_certificate_id, domain_dir)

        certificate_path = domain_dir / "Nginx" / f"1_{args.domain}_bundle.crt"
        private_key_path = domain_dir / "Nginx" / f"2_{args.domain}.key"
        if not certificate_path.exists() or not private_key_path.exists():
            raise RuntimeError(f"Expected Nginx certificate/key files were not found under {domain_dir}")

        cert_info = decode_leaf_certificate(certificate_path)
        print(json.dumps({"downloaded_certificate": cert_info}, ensure_ascii=False, indent=2))

        import_args = argparse.Namespace(
            domain=args.domain,
            certificate_path=str(certificate_path),
            private_key_path=str(private_key_path),
            env_path=args.env_path,
            profile_prefix=args.volcengine_profile_prefix,
            region=args.region,
            project_name=args.project_name,
            timeout=30,
        )
        import_result = import_volcengine_certificate.import_certificate(import_args)
        volc_cert_id = import_result.get("Result", {}).get("InstanceId", "")
        if not volc_cert_id:
            raise RuntimeError("Volcengine ImportCertificate did not return Result.InstanceId")

        print(json.dumps({"volcengine_certificate_id": volc_cert_id}, ensure_ascii=False, indent=2))
        run_passthrough(
            [
                "ve",
                "cdn",
                "BatchDeployCert",
                "--Domain",
                args.domain,
                "--CertId",
                volc_cert_id,
                "---profile",
                args.volcengine_profile,
                "---region",
                args.region,
            ]
        )
        run_passthrough(
            [
                "ve",
                "cdn",
                "DescribeCdnConfig",
                "--Domain",
                args.domain,
                "---profile",
                args.volcengine_profile,
                "---region",
                args.region,
            ]
        )
        run_passthrough(
            [
                "ve",
                "cdn",
                "ListCdnDomains",
                "--Domain",
                args.domain,
                "--ExactMatch",
                "true",
                "---profile",
                args.volcengine_profile,
                "---region",
                args.region,
            ]
        )
        print(
            json.dumps(
                {
                    "domain": args.domain,
                    "tencent_certificate_id": args.tencent_certificate_id,
                    "volcengine_certificate_id": volc_cert_id,
                    "not_after": cert_info["not_after"],
                    "temp_directory": str(temp_root) if args.keep_temp else "",
                },
                ensure_ascii=False,
                indent=2,
            )
        )
        return 0
    finally:
        if args.keep_temp:
            print(f"Kept temporary certificate directory: {temp_root}", file=sys.stderr)
        else:
            shutil.rmtree(temp_root, ignore_errors=True)


if __name__ == "__main__":
    raise SystemExit(main())