355 lines
13 KiB
Python
355 lines
13 KiB
Python
"""
|
||
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]:
|
||
"""
|
||
申请证书（使用 certbot + DNS-01 验证）
|
||
返回 (成功?, 日志信息)
|
||
|
||
泛域名 *.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")
|
||
|
||
if success:
|
||
cert_path = domain_dir / "fullchain.pem"
|
||
if cert_path.exists():
|
||
cert_data = cert_path.read_bytes()
|
||
cert = x509.load_pem_x509_certificate(cert_data)
|
||
not_after = cert.not_valid_after_utc
|
||
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
|
||
return True, f"Certificate issued, expires: {not_after.strftime('%Y-%m-%d %H:%M:%S')}, SAN: {san}"
|
||
|
||
return success, output
|
||
|
||
def renew_certificate(self, domain: str) -> tuple[bool, 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
|