宝塔续签SSL证书自动同步EdgeOne（国际站）

前言

• 众所周知腾讯云每个账号仅支持申请50张免费证书，同时不支持手动上传证书托管续签。
• 为了应对日复一日逐步收紧的SSL证书期限，域名太多手动上传SSL证书太麻烦，若忘记续签网站直接爆红（骗你的直接打不开）。
• 配合宝塔自动续签证书脚本，当有新的证书续签时，将同步上传并更新EdgeOne下的域名HTTPS。
• 功能介绍：本地证书上传至腾讯云 → 通知EdgeOne更新证书 → 60秒后统一删除旧证书。

参考API

• Tencent EdgeOne文档：https://edgeone.ai/zh/document/50539?product=api
• Tencent SSL API文档：https://www.tencentcloud.com/zh/document/product/1007/36571

实现教程

1. 在宝塔申请SSL证书，域名或站点较多建议DNS验证生成泛域名证书（如：baidu.com，*.baidu.com）证书路径在/www/server/panel/vhost/ssl/baidu.com目录下；若用单域名证书（如：a.baidu.com）证书路径在/www/server/panel/vhost/cert/a.baidu.com目录下；
2. 在腾讯云控制台生成访问密钥（API密钥管理），并记录SecretId和SecretKey。主账号的API密钥拥有完全控制权，注意密钥泄漏风险！如果不放心可创建有限权限的子账号（需申请的权限名我没试）；

3. 前往EdgeOne控制台，选择根域名获取站点ID：zone-xxxxxx，同时记录已加速的所有域名；

4. 下面展示完整的脚本，将以上步骤中的（证书目录、API密钥，站点ID）填写到脚本中：

# 作者：漫蓝梦坤（@mlmk6698）
# 来拿东西，别忘记感谢喵~

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
宝塔泛域名证书 → 腾讯云SSL + EdgeOne 自动更新脚本
支持自定义二级域名 + 批量删除旧证书（延迟60秒）

用法：
  export TENCENTCLOUD_SECRET_ID="your_id"
  export TENCENTCLOUD_SECRET_KEY="your_key"
  python3 update_ssl.py
"""

import os
import ssl
import hashlib
import hmac
import json
import sys
import time
import datetime

if sys.version_info[0] <= 2:
    from httplib import HTTPSConnection
else:
    from http.client import HTTPSConnection

_SSL_CTX = ssl._create_unverified_context()

# ============================================================
# 配置区
# ============================================================

SECRET_ID  = os.getenv("TENCENTCLOUD_SECRET_ID",  "你的API密钥SecretId")
SECRET_KEY = os.getenv("TENCENTCLOUD_SECRET_KEY", "你的API密钥SecretKey")

# 记录旧证书：
STATE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".cert_sync_state.json")

DOMAINS = [
    {
        "name":     "baidu.com",
        "cert_dir": "/www/server/panel/vhost/ssl/你的证书目录baidu.com", #证书目录（当宝塔续签后证书变更，哈希值变化触发上传）
        "zone_id":  "zone-xxxxxx", #EdgeOne控制台的站点ID
        "subdomains": ["baidu.com", "www.baidu.com", "cpdd.baidu.com"]  #加速域名，包含根域（若是单域名证书，此处只可填写证书对应的二级域名）
    },
    # 同上，域名/站点多的，多复制几个填写
    # {
    #     "name":     "tencent.com",
    #     "cert_dir": "/www/server/panel/vhost/ssl/tencent.com",
    #     "zone_id":  "zone-xxxxxx",
    #     "subdomains": ["tencent.com", "cloud.tencent.com", "cloud.tencent.com"]
    # },
    # 同上，这个是单域名证书模板
    # {
    #     "name":     "bilibili.com",
    #     "cert_dir": "/www/server/panel/vhost/cert/www.bilibili.com",
    #     "zone_id":  "zone-xxxxxx",
    #     "subdomains": ["www.bilibili.com"]
    # },
]

# ============================================================
# 腾讯云签名 v3（通用）
# ============================================================

def _sign(key, msg):
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()

def call_api(host, action, service, version, params, region=""):
    CT        = "application/json; charset=utf-8"
    payload   = json.dumps(params, separators=(",", ":"))
    timestamp = int(time.time())
    date      = datetime.datetime.fromtimestamp(timestamp, datetime.timezone.utc).strftime("%Y-%m-%d")

    canonical_headers = "content-type:%s\nhost:%s\nx-tc-action:%s\n" % (CT, host, action.lower())
    signed_headers    = "content-type;host;x-tc-action"
    hashed_payload    = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    canonical_request = "\n".join(["POST", "/", "", canonical_headers, signed_headers, hashed_payload])

    algorithm        = "TC3-HMAC-SHA256"
    credential_scope = "%s/%s/tc3_request" % (date, service)
    hashed_cr        = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
    string_to_sign   = "\n".join([algorithm, str(timestamp), credential_scope, hashed_cr])

    sk  = _sign(_sign(_sign(("TC3" + SECRET_KEY).encode("utf-8"), date), service), "tc3_request")
    sig = hmac.new(sk, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    authorization = "%s Credential=%s/%s, SignedHeaders=%s, Signature=%s" % (
        algorithm, SECRET_ID, credential_scope, signed_headers, sig)

    headers = {
        "Authorization":  authorization,
        "Content-Type":   CT,
        "Host":           host,
        "X-TC-Action":    action,
        "X-TC-Timestamp": timestamp,
        "X-TC-Version":   version,
    }
    if region:
        headers["X-TC-Region"] = region

    conn = HTTPSConnection(host, context=_SSL_CTX)
    conn.request("POST", "/", headers=headers, body=payload.encode("utf-8"))
    data     = json.loads(conn.getresponse().read())
    response = data.get("Response", {})

    if "Error" in response:
        err = response["Error"]
        raise RuntimeError("%s - %s" % (err.get("Code"), err.get("Message")))

    return response

# ============================================================
# 业务函数
# ============================================================

def upload_certificate(public_key, private_key, alias):
    resp = call_api(
        host    = "ssl.intl.tencentcloudapi.com",
        action  = "UploadCertificate",
        service = "ssl",
        version = "2019-12-05",
        params  = {
            "CertificatePublicKey":  public_key,
            "CertificatePrivateKey": private_key,
            "CertificateType":       "SVR",
            "Alias":                 alias,
            "Repeatable":            True,
        },
    )
    return resp.get("CertificateId") or resp.get("RepeatCertId")

def push_cert_to_edgeone(zone_id, cert_id, subdomains):
    if not subdomains:
        print("⚠️ 没有指定任何域名，跳过推送")
        return []

    call_api(
        host    = "teo.intl.tencentcloudapi.com",
        action  = "ModifyHostsCertificate",
        service = "teo",
        version = "2022-09-01",
        params  = {
            "ZoneId": zone_id,
            "Hosts":  subdomains,
            "Mode":   "sslcert",
            "ServerCertInfo": [{"CertId": cert_id}],
        },
    )
    return subdomains

def delete_old_certificate(cert_id, domain_name):
    if not cert_id:
        return
    try:
        call_api(
            host    = "ssl.intl.tencentcloudapi.com",
            action  = "DeleteCertificate",
            service = "ssl",
            version = "2019-12-05",
            params  = {"CertificateId": cert_id},
        )
        print(f"[{domain_name}] ✅ 旧证书已删除：{cert_id}")
    except Exception as e:
        print(f"[{domain_name}] ⚠️ 删除旧证书失败：{cert_id}，原因：{e}")

# ============================================================
# 状态缓存
# ============================================================

def load_state():
    if os.path.isfile(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f)
    return {}

def save_state(state):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f, indent=2)

def cert_hash(cert_dir):
    fc = open(os.path.join(cert_dir, "fullchain.pem"), "rb").read()
    pk = open(os.path.join(cert_dir, "privkey.pem"),   "rb").read()
    return hashlib.sha256(fc + pk).hexdigest()

# ============================================================
# 核心逻辑
# ============================================================

def process_domain(domain, state, certs_to_delete):
    name     = domain["name"]
    cert_dir = domain["cert_dir"]
    zone_id  = domain["zone_id"]
    subdomains = domain.get("subdomains", [])

    print("-"*60)
    print(f"[{name}] 开始处理域名任务")

    fullchain_path = os.path.join(cert_dir, "fullchain.pem")
    privkey_path   = os.path.join(cert_dir, "privkey.pem")

    for p in (fullchain_path, privkey_path):
        if not os.path.isfile(p):
            print(f"[{name}] ⚠️ 文件不存在，跳过：{p}")
            return False

    current_hash = cert_hash(cert_dir)
    if current_hash == state.get(name, {}).get("hash"):
        print(f"[{name}] ✅ 证书无变动，跳过")
        return False

    print(f"[{name}] 🔄 证书已变动，开始更新…")

    public_key  = open(fullchain_path).read().strip()
    private_key = open(privkey_path).read().strip()

    # Step 1: 上传证书
    try:
        alias       = "auto-%s-%s" % (name, datetime.date.today())
        new_cert_id = upload_certificate(public_key, private_key, alias)
        print(f"[{name}] ✅ 证书上传成功，新ID: {new_cert_id}")
    except Exception as e:
        print(f"[{name}] ❌ UploadCertificate 失败：{e}")
        return False

    # Step 2: 推送到 EdgeOne
    try:
        hosts = push_cert_to_edgeone(zone_id, new_cert_id, subdomains)
        print(f"[{name}] ✅ EdgeOne 推送成功，域名：{hosts}")
    except Exception as e:
        print(f"[{name}] ❌ ModifyHostsCertificate 失败：{e}")
        print(f"[{name}] ℹ️ 证书已上传（ID: {new_cert_id}），请手动在 EdgeOne 控制台更新")
        state[name] = {"hash": current_hash, "cert_id": new_cert_id}
        return False

    # Step 3: 记录旧证书，统一删除
    old_cert_id = state.get(name, {}).get("cert_id")
    if old_cert_id and old_cert_id != new_cert_id:
        certs_to_delete.append((name, old_cert_id))

    state[name] = {"hash": current_hash, "cert_id": new_cert_id}

    print(f"[{name}] ✅ 域名任务完成")
    print("-"*60)
    return True

# ============================================================
# 主程序
# ============================================================

def main():
    print("=" * 60)
    print("  腾讯云 SSL + EdgeOne 证书同步  %s" % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    print("=" * 60)

    state   = load_state()
    updated = 0
    certs_to_delete = []

    # 上传 + EdgeOne 推送
    for domain in DOMAINS:
        try:
            if process_domain(domain, state, certs_to_delete):
                updated += 1
        except Exception as e:
            print(f"[{domain['name']}] ❌ 未预期错误：{e}")
        finally:
            save_state(state)

    # 等待60秒再删除旧证书
    if certs_to_delete:
        print("\n等待60秒再删除旧证书...")
        time.sleep(60)
        for domain_name, cert_id in certs_to_delete:
            delete_old_certificate(cert_id, domain_name)

    print("-" * 60)
    print("完成。共 %d 个域名，%d 个成功更新。" % (len(DOMAINS), updated))

if __name__ == "__main__":
    main()

5. 设置计划任务，可在宝塔已有的计划任务（续签Let’s Encrypt证书）下面加上你的脚本文件位置，如下：

/www/server/panel/pyenv/bin/python3 -u /www/server/panel/class/acme_v2.py --renew_v2=1
cd /www/update_ssl
/www/server/pyporject_evn/versions/3.13.7/bin/python3 update_ssl.py

6. 大功告成！