dev-ops-agent/scripts/import_volcengine_certificate.py

#!/usr/bin/env python3
"""Import a local certificate/key pair into Volcengine Certificate Service.

This script deliberately reads certificate material from files at runtime so
private keys do not need to appear in shell arguments.
"""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import hmac
import json
import sys
import urllib.error
import urllib.request
from pathlib import Path


def read_dotenv(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        name, value = line.split("=", 1)
        value = value.strip()
        if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
            value = value[1:-1]
        values[name.strip()] = value
    return values


def sha256_hex(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def hmac_bytes(key: bytes, text: str) -> bytes:
    return hmac.new(key, text.encode("utf-8"), hashlib.sha256).digest()


def hmac_hex(key: bytes, text: str) -> str:
    return hmac.new(key, text.encode("utf-8"), hashlib.sha256).hexdigest()


def sign_headers(
    *,
    access_key: str,
    secret_key: str,
    region: str,
    service: str,
    host: str,
    method: str,
    uri: str,
    query: str,
    body: str,
) -> dict[str, str]:
    payload_hash = sha256_hex(body)
    now = dt.datetime.now(dt.UTC)
    x_date = now.strftime("%Y%m%dT%H%M%SZ")
    short_date = now.strftime("%Y%m%d")

    canonical_headers = (
        f"content-type:application/json\n"
        f"host:{host}\n"
        f"x-content-sha256:{payload_hash}\n"
        f"x-date:{x_date}\n"
    )
    signed_headers = "content-type;host;x-content-sha256;x-date"
    canonical_request = "\n".join(
        [
            method,
            uri,
            query,
            canonical_headers,
            signed_headers,
            payload_hash,
        ]
    )
    scope = f"{short_date}/{region}/{service}/request"
    string_to_sign = "\n".join(
        [
            "HMAC-SHA256",
            x_date,
            scope,
            sha256_hex(canonical_request),
        ]
    )

    # Volcengine OpenAPI uses the raw SK as the first HMAC key for this API.
    k_date = hmac_bytes(secret_key.encode("utf-8"), short_date)
    k_region = hmac_bytes(k_date, region)
    k_service = hmac_bytes(k_region, service)
    k_signing = hmac_bytes(k_service, "request")
    signature = hmac_hex(k_signing, string_to_sign)

    authorization = (
        f"HMAC-SHA256 Credential={access_key}/{scope}, "
        f"SignedHeaders={signed_headers}, Signature={signature}"
    )
    return {
        "Authorization": authorization,
        "Content-Type": "application/json",
        "Host": host,
        "X-Content-Sha256": payload_hash,
        "X-Date": x_date,
    }


def import_certificate(args: argparse.Namespace) -> dict:
    env = read_dotenv(Path(args.env_path))
    access_key = env.get(f"{args.profile_prefix}_VOLCENGINE_ACCESS_KEY", "")
    secret_key = env.get(f"{args.profile_prefix}_VOLCENGINE_SECRET_KEY", "")
    if not access_key or not secret_key:
        raise RuntimeError(
            f"Missing {args.profile_prefix}_VOLCENGINE_ACCESS_KEY or "
            f"{args.profile_prefix}_VOLCENGINE_SECRET_KEY in {args.env_path}"
        )

    certificate = Path(args.certificate_path).read_text(encoding="utf-8")
    private_key = Path(args.private_key_path).read_text(encoding="utf-8")
    body = json.dumps(
        {
            "CertificateInfo": {
                "CertificateChain": certificate,
                "PrivateKey": private_key,
            },
            "ProjectName": args.project_name,
            "Repeatable": True,
            "Tag": args.domain,
        },
        ensure_ascii=False,
        separators=(",", ":"),
    )

    service = "certificate_service"
    host = "open.volcengineapi.com"
    query = "Action=ImportCertificate&Version=2024-10-01"
    headers = sign_headers(
        access_key=access_key,
        secret_key=secret_key,
        region=args.region,
        service=service,
        host=host,
        method="POST",
        uri="/",
        query=query,
        body=body,
    )

    request = urllib.request.Request(
        f"https://{host}/?{query}",
        data=body.encode("utf-8"),
        headers=headers,
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=args.timeout) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        # Do not print Authorization, AK, SK, certificate, or private key.
        raise RuntimeError(f"Volcengine API request failed with HTTP status {exc.code}") from None


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--domain", required=True)
    parser.add_argument("--certificate-path", required=True)
    parser.add_argument("--private-key-path", required=True)
    parser.add_argument("--env-path", default=".env")
    parser.add_argument("--profile-prefix", default="INTELLIGROW")
    parser.add_argument("--region", default="cn-beijing")
    parser.add_argument("--project-name", default="default")
    parser.add_argument("--timeout", type=int, default=30)
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    result = import_certificate(args)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


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