acme-auto/backend/acme_service.py

"""
ACME 核心服务 - 集成 Let's Encrypt 证书申请与续签
支持 DNS-01 验证，通过 AliDNS API 自动添加 TXT 记录
"""
import json
import os
import shutil
import time
import hashlib
import base64
import logging
import subprocess
import tempfile
from pathlib import Path
from datetime import datetime, timedelta

import requests
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa

from backend.config import get_settings
from backend.models import AcmeConfig, AcmeLog, Domain

logger = logging.getLogger(__name__)


class AliDNSClient:
    """AliDNS API 客户端，用于添加/删除 TXT 记录"""

    def __init__(self, access_key: str, access_secret: str):
        self.access_key = access_key
        self.access_secret = access_secret
        self.api_url = "https://alidns.aliyuncs.com"

    def _sign_params(self, params: dict) -> dict:
        """生成阿里云 API 签名"""
        params.update({
            "Format": "JSON",
            "Version": "2015-01-09",
            "AccessKeyId": self.access_key,
            "SignatureMethod": "HMAC-SHA1",
            "Timestamp": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
            "SignatureVersion": "1.0",
            "SignatureNonce": str(int(time.time() * 1000)),
        })
        sorted_params = sorted(params.items())
        query_string = "&".join(
            f"{self._percent_encode(k)}={self._percent_encode(v)}" for k, v in sorted_params
        )
        string_to_sign = f"GET&{self._percent_encode('/')}&{self._percent_encode(query_string)}"
        import hmac
        sign = hmac.new(
            (self.access_secret + "&").encode(),
            string_to_sign.encode(),
            hashlib.sha1,
        ).digest()
        params["Signature"] = base64.b64encode(sign).decode()
        return params

    @staticmethod
    def _percent_encode(s: str) -> str:
        s = str(s)
        return requests.utils.quote(s, safe="")

    def _request(self, params: dict) -> dict:
        signed = self._sign_params(params)
        resp = requests.get(self.api_url, params=signed, timeout=10)
        resp.raise_for_status()
        return resp.json()

    def add_txt_record(self, domain: str, value: str) -> str:
        """添加 _acme-challenge TXT 记录，返回 RecordId"""
        # 提取主域名
        parts = domain.split(".")
        rr = f"_acme-challenge.{'.'.join(parts[:-2])}" if len(parts) > 2 else "_acme-challenge"
        main_domain = ".".join(parts[-2:])

        result = self._request({
            "Action": "AddDomainRecord",
            "DomainName": main_domain,
            "RR": rr,
            "Type": "TXT",
            "Value": value,
        })
        record_id = result.get("RecordId", "")
        logger.info(f"AliDNS: added TXT record {rr}.{main_domain} = {value}, RecordId={record_id}")
        return str(record_id)

    def delete_txt_record(self, record_id: str):
        """删除指定的 DNS 记录"""
        self._request({
            "Action": "DeleteDomainRecord",
            "RecordId": record_id,
        })
        logger.info(f"AliDNS: deleted record {record_id}")


class AcmeService:
    """ACME 证书管理服务"""

    def __init__(self, config: AcmeConfig, cert_dir: str):
        self.config = config
        self.cert_dir = Path(cert_dir)
        self.cert_dir.mkdir(parents=True, exist_ok=True)

    def _get_dns_client(self) -> AliDNSClient:
        """根据配置创建 DNS 客户端"""
        creds = json.loads(self.config.dns_credentials)
        if self.config.dns_provider == "aliyun":
            return AliDNSClient(
                access_key=creds.get("access_key", ""),
                access_secret=creds.get("access_secret", ""),
            )
        raise ValueError(f"Unsupported DNS provider: {self.config.dns_provider}")

    def _generate_account_key(self) -> rsa.RSAPrivateKey:
        """生成 ACME 账户私钥"""
        return rsa.generate_private_key(public_exponent=65537, key_size=2048)

    def _load_or_create_account_key(self) -> rsa.RSAPrivateKey:
        """加载或创建账户私钥"""
        if self.config.account_key:
            return serialization.load_pem_private_key(
                self.config.account_key.encode(), password=None
            )
        key = self._generate_account_key()
        pem = key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        )
        self.config.account_key = pem.decode()
        return key

    def _generate_csr(self, domain: str, key_path: Path):
        """生成域名私钥和 CSR"""
        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)

        # 保存私钥
        key_pem = key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        )
        key_path.write_bytes(key_pem)

        # 生成 CSR
        csr = (
            x509.CertificateSigningRequestBuilder()
            .subject_name(x509.Name([x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, domain)]))
            .add_extension(
                x509.SubjectAlternativeName([x509.DNSName(domain)]),
                critical=False,
            )
            .sign(key, hashes.SHA256())
        )
        csr_pem = csr.public_bytes(serialization.Encoding.PEM)
        return csr_pem

    def _run_certbot(self, domain: str, action: str) -> tuple[bool, str]:
        """
        使用 certbot 执行 ACME 操作
        action: "certonly" (申请) 或 "renew" (续签)

        泛域名处理：
        - 输入 "*.zhzp.top" 自动同时申请 "*.zhzp.top" + "zhzp.top"
        - 证书存储目录使用裸域名 "zhzp.top"
        """
        creds = json.loads(self.config.dns_credentials)

        # 泛域名：同时申请 *.example.com 和 example.com
        # 存储目录使用裸域名
        if domain.startswith("*."):
            bare_domain = domain[2:]
            certbot_domains = ["-d", domain, "-d", bare_domain]
            store_dir = bare_domain
        else:
            certbot_domains = ["-d", domain]
            store_dir = domain

        domain_dir = self.cert_dir / store_dir
        domain_dir.mkdir(parents=True, exist_ok=True)

        # 写入临时凭据文件（certbot-dns-alicloud 需要 INI 格式）
        creds_content = (
            f"dns_alicloud_access_key = {creds.get('access_key', '')}\n"
            f"dns_alicloud_secret_key = {creds.get('access_secret', '')}\n"
            f"dns_alicloud_region = {creds.get('region', 'cn-hangzhou')}\n"
        )
        creds_file = tempfile.NamedTemporaryFile(
            mode="w", suffix=".ini", prefix="certbot-dns-", delete=False
        )
        creds_file.write(creds_content)
        creds_file.close()

        # 用当前 Python 环境对应的 certbot，避免调到系统装的
        import sys
        certbot_bin = str(Path(sys.prefix) / "bin" / "certbot")
        if not Path(certbot_bin).exists():
            certbot_bin = "certbot"  # fallback

        cmd = [
            certbot_bin, "certonly",
            "--non-interactive",
            "--agree-tos",
            "--email", self.config.email,
            "--authenticator", "dns-alicloud",
            "--dns-alicloud-credentials", creds_file.name,
            *certbot_domains,
            "--cert-path", str(domain_dir / "fullchain.pem"),
            "--key-path", str(domain_dir / "private.key"),
            "--work-dir", str(self.cert_dir / ".certbot-work"),
            "--config-dir", str(self.cert_dir / ".certbot-config"),
            "--logs-dir", str(self.cert_dir / ".certbot-logs"),
        ]

        if self.config.acme_server != "https://acme-v02.api.letsencrypt.org/directory":
            cmd.extend(["--server", self.config.acme_server])

        try:
            result = subprocess.run(
                cmd, capture_output=True, text=True, timeout=180,
            )
            output = result.stdout + "\n" + result.stderr
            success = result.returncode == 0

            # certbot 实际存储路径（可能忽略了 --cert-path）
            certbot_live = self.cert_dir / ".certbot-config" / "live" / store_dir
            if success and certbot_live.exists():
                import shutil
                src_fullchain = certbot_live / "fullchain.pem"
                src_privkey = certbot_live / "privkey.pem"
                dst_fullchain = domain_dir / "fullchain.pem"
                dst_privkey = domain_dir / "private.key"
                if src_fullchain.exists():
                    shutil.copy2(str(src_fullchain), str(dst_fullchain))
                if src_privkey.exists():
                    shutil.copy2(str(src_privkey), str(dst_privkey))

            return success, output
        except subprocess.TimeoutExpired:
            return False, "Certbot timeout after 180s"
        except FileNotFoundError:
            return False, "certbot not found, please install: pip install certbot certbot-dns-alicloud"
        finally:
            # 清理临时凭据文件
            try:
                os.unlink(creds_file.name)
            except OSError:
                pass

    def issue_certificate(self, domain: str) -> tuple[bool, str, dict, str]:
        """
        申请证书（使用 certbot + DNS-01 验证）
        返回 (成功?, 摘要信息, 证书详情dict, certbot原始输出)

        泛域名 *.example.com 会自动同时申请裸域名 example.com
        """
        # 泛域名使用裸域名路径
        store_dir = domain[2:] if domain.startswith("*.") else domain
        domain_dir = self.cert_dir / store_dir
        domain_dir.mkdir(parents=True, exist_ok=True)

        success, output = self._run_certbot(domain, "certonly")

        cert_info = {}
        if success:
            cert_path = domain_dir / "fullchain.pem"
            if cert_path.exists():
                try:
                    cert_data = cert_path.read_bytes()
                    cert = x509.load_pem_x509_certificate(cert_data)
                    not_before = getattr(cert, "not_valid_before_utc", None) or cert.not_valid_before
                    not_after = getattr(cert, "not_valid_after_utc", None) or cert.not_valid_after
                    san = []
                    try:
                        san_ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
                        san = san_ext.value.get_values_for_type(x509.DNSName)
                    except x509.ExtensionNotFound:
                        pass
                    cert_info = {
                        "not_before": not_before.isoformat(),
                        "not_after": not_after.isoformat(),
                        "serial_number": str(cert.serial_number),
                        "san": [str(s) for s in san],
                        "issuer": cert.issuer.rfc4514_string(),
                    }
                    msg = f"签发成功，过期时间: {not_after.strftime('%Y-%m-%d %H:%M:%S')}, SAN: {san}"
                    return True, msg, cert_info, output
                except Exception as e:
                    return True, f"签发成功但解析证书失败: {e}", cert_info, output

        return success, output, cert_info, output

    def renew_certificate(self, domain: str) -> tuple[bool, str, dict, str]:
        """续签证书"""
        return self.issue_certificate(domain)

    def check_expiry(self, domain: str) -> tuple[bool, int]:
        """检查证书到期天数，返回 (需要续签?, 剩余天数)"""
        # 泛域名使用裸域名路径
        store_dir = domain[2:] if domain.startswith("*.") else domain
        cert_path = self.cert_dir / store_dir / "fullchain.pem"
        if not cert_path.exists():
            return True, 0

        try:
            cert = x509.load_pem_x509_certificate(cert_path.read_bytes())
            not_after = cert.not_valid_after_utc
            days_left = (not_after - datetime.utcnow()).days
            return days_left <= self.config.renew_days, days_left
        except Exception:
            return True, 0

    def get_cert_info(self, domain: str) -> dict | None:
        """获取证书详细信息"""
        # 泛域名使用裸域名路径
        store_dir = domain[2:] if domain.startswith("*.") else domain
        cert_path = self.cert_dir / store_dir / "fullchain.pem"
        if not cert_path.exists():
            return None

        try:
            cert = x509.load_pem_x509_certificate(cert_path.read_bytes())
            # 兼容新旧版本 cryptography 库
            not_before = getattr(cert, "not_valid_before_utc", None) or cert.not_valid_before
            not_after = getattr(cert, "not_valid_after_utc", None) or cert.not_valid_after
            try:
                san = [name.value for name in cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value]
            except Exception:
                san = []
            return {
                "subject": cert.subject.rfc4514_string(),
                "issuer": cert.issuer.rfc4514_string(),
                "not_before": not_before.isoformat(),
                "not_after": not_after.isoformat(),
                "serial_number": str(cert.serial_number),
                "san": san,
            }
        except Exception as e:
            logger.error(f"解析证书失败 {cert_path}: {e}")
            return None

    def auto_renew_all(self, domains: list[Domain]) -> list[dict]:
        """自动续签所有即将过期的证书"""
        results = []
        for d in domains:
            need_renew, days_left = self.check_expiry(d.domain)
            if need_renew:
                success, msg = self.renew_certificate(d.domain)
                results.append({
                    "domain": d.domain,
                    "action": "renew",
                    "success": success,
                    "message": msg,
                    "days_left": days_left,
                })
            else:
                results.append({
                    "domain": d.domain,
                    "action": "skip",
                    "success": True,
                    "message": f"Expires in {days_left} days, no renewal needed",
                    "days_left": days_left,
                })
        return results