腾讯域名SSl免费证书自动续签

由于现在免费的ssl证书有效期才三个月，每次自己手动去申请和替换麻烦得要死，就自己弄了自动部署的。

我的是在腾讯上的，阿里和其它平台的不支持，但下述的原理可以相通。

设置访问秘钥

它的使用说明：https://cloud.tencent.com/document/product/400/41657

申请安全凭证

本文使用的安全凭证为密钥，密钥包括 SecretId 和 SecretKey。每个用户最多可以拥有两对密钥。
SecretId：用于标识 API 调用者身份，可以简单类比为用户名。
SecretKey：用于验证 API 调用者的身份，可以简单类比为密码。
用户必须严格保管安全凭证，避免泄露，否则将危及财产安全。如已泄露，请立刻禁用该安全凭证。
申请安全凭证的具体步骤如下：

登录 腾讯云管理中心控制台 。
前往 云API密钥 的控制台页面。
在 云API密钥 页面，单击【新建密钥】创建一对密钥。

直接访问：https://console.cloud.tencent.com/cam/capi

创建

有些账号还需要进行验证，验证成功后便申请成功

注意，上面的 SecretKey 和 SecretId 一定要自己保存好。

配置脚本文件

# -*- coding: utf-8 -*-
import hashlib
import hmac
import json
import time
from datetime import datetime
from http.client import HTTPSConnection
import base64
import zipfile
from io import BytesIO
import logging
import os

def load_write(data):
    """
    json通用文件写入
    :param data: 字典数据
    :return:
    """
    try:
        with open(config_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=4)
    except Exception as e:
        logging.info('文件创建失败！', e)


def load_config():
    """ 配置文件的初始化"""
    config_data = {}
    try:
        with open(config_path, "rb") as fp:
            config_data = json.load(fp)
    except FileNotFoundError:
        logging.info('创建文件，如果是权限不足给赋予读写文件权限。\n创建配置文件中...')
        config_template = {
            "SecretId": "AKID 腾讯云的 SecretId",
            "secret_key": "腾讯云的 SecretKey",
            "user_url": "证书绑定的域名。示例值：jiubanyipeng.com",
            "certificateId": "当前域名的证书ID，如果这里为空，会默认去帮你生成一个证书的。新申请后的证书得到的ID会替换这里的值",
            "ssl_key_path": "证书(PEM格式)文件的绝对路径名称，这里需要进行替换的证书文件",
            "ssl_pem_path": "密钥(KEY)文件的绝对路径名称，这里需要进行替换的证书文件",
            "MAX_DAY": "低于几天就开始申请证书，这里填写字符串的数字，示例值：5"
        }
        load_write(config_template)
        logging.info('文件创建完成，请填写配置文件信息！')

    except Exception as e:
        logging.error(f'文件可能有问题，请检查！以下是报错信息：\n {e}')
    return config_data


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


def days_difference(past_date_str):
    """
    输入时间计算相隔天数
    :param past_date_str:年月日时分秒的
    :return: 天数
    """
    past_datetime = datetime.strptime(past_date_str, "%Y-%m-%d %H:%M:%S")  # 将字符串转换为日期对象
    today_date = datetime.today()  # 获取今天的日期
    delta = past_datetime - today_date  # 计算天数差异

    # 返回天数差异
    return delta.days


def join_request(action, payload):
    """ 腾讯云请求公共验证
    :param action: 应用名称
    :param payload: 请求的数据字典（注：字符串形式的字典值）
    :return:  算法签名token值
    """
    timestamp = int(time.time())
    date = datetime.utcfromtimestamp(timestamp).strftime("%Y-%m-%d")
    method = "POST"
    uri = "/"
    querystring = ""

    # ************* 步骤 1：拼接规范请求串 *************
    ct = "application/json; charset=utf-8"
    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_request_payload = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    canonical_request = (method + "\n" + uri + "\n" + querystring + "\n" + canonical_headers + "\n" +  signed_headers + "\n" +  hashed_request_payload)

    # ************* 步骤 2：拼接待签名字符串 *************
    credential_scope = date + "/" + service + "/" + "tc3_request"
    hashed_canonical_request = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
    string_to_sign = (algorithm + "\n" + str(timestamp) + "\n" + credential_scope + "\n" + hashed_canonical_request)

    # ************* 步骤 3：计算签名 *************
    secret_date = sign(("TC3" + config['secret_key']).encode("utf-8"), date)
    secret_service = sign(secret_date, service)
    secret_signing = sign(secret_service, "tc3_request")
    signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    # ************* 步骤 4：拼接 Authorization *************
    authorization = (algorithm + " " + "Credential=" + config['SecretId']  + "/" + credential_scope + ", " +  "SignedHeaders=" + signed_headers + ", " + "Signature=" + signature)

    # ************* 步骤 5：构造并发起请求 *************
    headers = {
        "Authorization": authorization,
        "Content-Type": "application/json; charset=utf-8",
        "Host": host,
        "X-TC-Action": action,
        "X-TC-Timestamp": timestamp,
        "X-TC-Version": version,
        "X-TC-RequestClient": "APIExplorer" ,
    }
    return headers


def get_ssl_info(CertificateId):
    """
    获取证书信息详情
    :param CertificateId:
    :return: 数据字典 或 False
    """
    action = "DescribeCertificateDetail"
    payload = '{"CertificateId": "%s"}'  % CertificateId
    headers = join_request(action,payload)
    try:
        req = HTTPSConnection(host)
        req.request("POST", "/", headers=headers, body=payload.encode("utf-8"))
        resp = req.getresponse()
        data_bytes = resp.read()
        data_str = data_bytes.decode('utf-8')
        data = json.loads(data_str)
        if data['Response'].get('Error', False):
            logging.error(f"错误：{data['Response']['Error']['Message']}")
            return False
        return data
    except Exception as err:
        logging.debug(err)
        return False


def get_ssl_create(user_url, certificateid =''):
    """
    腾讯云ssl证书申请
    :param user_url: 用户域名
    :param certificateid: 证书ID
    :return: 相应结果
    """
    logging.info('开始申请证书！')
    action = "ApplyCertificate"
    payload = '{"DvAuthMethod": "DNS_AUTO",  "DomainName": "%s","OldCertificateId":"%s"}' % (user_url, certificateid)
    headers = join_request(action,payload)
    try:
        req = HTTPSConnection(host)
        req.request("POST", "/", headers=headers, body=payload.encode("utf-8"))
        resp = req.getresponse()   # b'{"Response":{"CertificateId":"xxxx","Data":[],"RequestId":"xxxxxxxx-xxxxxx-xxxxxxxx-xxxx"}}'
        data_bytes = resp.read()
        data_str = data_bytes.decode('utf-8')
        data = json.loads(data_str)
        if data['Response'].get('Error',False):
            logging.error(f"错误：{data['Response']['Error']['Message']}")
            return False
        return data
    except Exception as err:
        logging.debug(err)
        return False


def get_downloadCertificate(certificateId):
    """
    下载证书并替换原证书(默认先等待10分钟后再进行更新！)
    :param certificateId:
    :return:
    """
    logging.info('下载证书，等待10分钟')
    time.sleep(60 * 10)  # 现在一般申请证书的时间 十分钟左右就可以了，应该去通过查询接口查的，赖得查了
    action = "DownloadCertificate"
    payload = '{"CertificateId": "%s"}' % certificateId
    headers = join_request(action, payload)

    try:
        req = HTTPSConnection(host)
        req.request("POST", "/", headers=headers, body=payload.encode("utf-8"))
        resp = req.getresponse()
        data_bytes = resp.read()
        data_str = data_bytes.decode('utf-8')
        data = json.loads(data_str)
        if data['Response'].get('Error',False):
            logging.error(f"错误：{data['Response']['Error']['Message']}")
            return False
        base64_content = data['Response']['Content']
        decoded_data = base64.b64decode(base64_content)
        try:
            # 使用 BytesIO 创建一个内存中的文件对象
            with zipfile.ZipFile(BytesIO(decoded_data), 'r') as zip_ref:
                # 打印 ZIP 文件中的所有文件名
                # print(zip_ref.namelist())  # 默认情况下文件排序0为域名.csr，1为域名.pem，2为域名.key，后面的不考虑
                logging.info('证书获取成功，正在写入文件中...')
                file_name_pem = zip_ref.namelist()[1]
                file_name_key = zip_ref.namelist()[2]
                with zip_ref.open(file_name_pem) as file:
                    content = file.read().decode('utf-8').replace('\r\n', '\n')
                    with open(config['ssl_pem_path'],'w+',encoding='utf-8') as f:
                        f.write(content)
                with zip_ref.open(file_name_key) as file:
                    content = file.read().decode('utf-8').replace('\r\n', '\n')
                    with open(config['ssl_key_path'],'w+',encoding='utf-8') as f:
                        f.write(content)
            # 删除之前的旧证书  ?
            # 重启nginx服务 不重启服务无法加载已经更新后的证书
            logging.info("重启nginx服务")
            os.system('systemctl restart nginx')
            return True
        except zipfile.BadZipFile:
            logging.error("解码后的数据不是一个有效的 ZIP 文件。")
            return False
        except Exception as err:
            logging.debug(err)
            return False
    except Exception as err:
        logging.debug(err)
        return False


def get_describecertificates():
    """
    获取证书列表(暂时无用，多域名准备的)
    :return:
    """
    action = "DescribeCertificates"
    payload = '{"CertificateId":""}'
    headers = join_request(action, payload)
    try:
        req = HTTPSConnection(host)
        req.request("POST", "/", headers=headers, body=payload.encode("utf-8"))
        resp = req.getresponse()
        data_bytes = resp.read()
        data_str = data_bytes.decode('utf-8')
        data = json.loads(data_str)
        if data['Response'].get('TotalCount', False):
            return data
        else:
            logging.error(f'证书ID： 删除失败！\n {data}')
            return False
    except Exception as err:
        logging.debug(f"未知bug： {err}")
        return False


def revoke_certificate(certificateid):
    """
    吊销证书（该功能未完善，只能提起吊销，后面还需要添加DNS的解析验证等等的后续操作）
    :param certificateid:  证书ID
    :return: 吊销结果
    """
    action = "RevokeCertificate"
    payload = '{"CertificateId":"%s"}' % certificateid
    headers = join_request(action, payload)
    try:
        req = HTTPSConnection(host)
        req.request("POST", "/", headers=headers, body=payload.encode("utf-8"))
        resp = req.getresponse()
        data_bytes = resp.read()
        data_str = data_bytes.decode('utf-8')
        data = json.loads(data_str)
        if data['Response'].get('RevokeDomainValidateAuths', False):
            logging.info(f'证书ID：{certificateid} 吊销成功')
        else:
            logging.error(f'证书ID：{certificateid} 吊销失败！\n {data}')
    except Exception as err:
        logging.debug(f"未知bug： {err}")


def del_certificate(certificateid):
    """
    删除证书，删除之前要保证证书是无效的，如证书已吊销或已过期等
    :param certificateid: 证书ID
    :return: 删除结果
    """
    action = "DeleteCertificate"
    payload = '{"CertificateId":"%s"}' % certificateid
    headers = join_request(action, payload)
    try:
        req = HTTPSConnection(host)
        req.request("POST", "/", headers=headers, body=payload.encode("utf-8"))
        resp = req.getresponse()
        data_bytes = resp.read()
        data_str = data_bytes.decode('utf-8')
        data = json.loads(data_str)
        if data['Response'].get('DeleteResult',False):
            logging.info(f'证书ID：{certificateid} 删除成功')
        else:
            logging.error(f'证书ID：{certificateid} 删除失败！\n {data}')
    except Exception as err:
        logging.debug(f"未知bug： {err}")


def run():
    """
    代码运行
    :return: 还有多少天运行
    """
    # 验证是否存在证书ID，目前默认长度为 8
    if len(config.get('certificateId','')) != 8:
        # 申请证书
        get_ssl_res = get_ssl_create(config['user_url'])
        if get_ssl_res:
            certificateId = get_ssl_res['Response']['CertificateId']
            config['certificateId'] = certificateId
            load_write(config)  # 更新字典文件
            # 更新证书
            update_res = get_downloadCertificate(config['certificateId'])
            if update_res:
                return 90 - int(config.get('MAX_DAY', '5').strip())
            else:
                return False
        else:
            logging.error(f'证书申请失败！\n {get_ssl_res}')
            return False

    ssl_info = get_ssl_info(config.get('certificateId', ''))
    CertEndTime = ssl_info['Response']['CertEndTime']  # 证书失效时间，可能为null
    day = days_difference(CertEndTime)
    if day > int(config.get('MAX_DAY', '5').strip()):
        logging.info(f'距离要更新证书的时间还有 {day} 天')
        return day
    else:
        # 准备过期，申请证书
        get_ssl_res = get_ssl_create(config['user_url'])
        if get_ssl_res:
            certificateId = get_ssl_res['Response']['CertificateId']
            config['certificateId'] = certificateId
            load_write(config)  # 更新字典文件
            # 更新证书
            update_res = get_downloadCertificate(config['certificateId'])
            if update_res:
                return 90 - int(config.get('MAX_DAY', '5').strip())
            else:
                return False
        else:
            logging.error(f'证书申请失败！\n {get_ssl_res}')
            return False


#####  腾讯云请求通用配置  ##########
algorithm = "TC3-HMAC-SHA256"   # 腾讯云签名算法 勿动！
host = "ssl.tencentcloudapi.com"  # 请求地址
service = "ssl"   #  请求服务
version = "2019-12-05"   # 加密算法版本

# 配置文件路径，默认为当前 脚本 目录下的config.json
config_path = os.path.join(os.path.dirname(__file__), 'config.json')


if __name__ == "__main__":
    # 初始化日志记录,不保存日志
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        encoding='utf-8',
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    config = load_config()
    if not config:
        logging.error("配置加载失败！")
        time.sleep(5)
        exit(1)  # 使用非零状态码表示异常退出

    if config.get('is_sustain', False):  # 是否一直持续运行
        while True:
            res = run()
            if not res:
                break
            time.sleep(60*60*24)  # 24小时后重新运行
    else:
        run()

首次运行该会在当前的运行目录生成一个 “config.json” 的配置文件，相关的配置说明：

{
    "SecretId": "AKID 腾讯云的 SecretId",
    "secret_key": "腾讯云的 SecretKey",
    "user_url": "证书绑定的域名。示例值：jiubanyipeng.com",
    "certificateId": "当前域名的证书ID，如果这里为空，会默认去帮你生成一个证书的。新申请后的证书得到的ID会替换这里的值",
    "ssl_key_path": "证书(PEM格式)文件的绝对路径名称，这里需要进行替换的证书文件",
    "ssl_pem_path": "密钥(KEY)文件的绝对路径名称，这里需要进行替换的证书文件",
    "MAX_DAY": "低于几天就开始申请证书，这里填写字符串的数字，示例值：5"
}

配置文件请根据自己的需要进行填写，到这里腾讯的部分已经完成了。

自动化原理

脚本会定期（PS：我这里使用别的工具的定时任务管理器，可以自己修改脚本定时去检测），去检测证书 离有效期 还有多少天（PS：可以自己优化脚本，直接定多少天后去执行就可以了，但我这里为了方便就都设置了）便进行重新申请证书，发起证书申请后，一般15分钟内会申请下来，查询到刚申请的证书进行与旧的替换（PS：这里很重要，设置的路径以及对应的公私钥证书要对应上），替换成功后重新启动中间件nginx服务。

在配置文件中，还有两个比较关键的值且很有可能会导致你失败的原因是”ssl_key_path” 和”ssl_pem_path” 配置的路径不对。

使用宝塔的定时管理器进行定时检测

也可以自行修改脚本，定时去检测。添加好定时任务后，等待即可。

执行结果如下：

注意：我在脚本中，配置了当更新替换证书后，会重启web中间件Nginx，如果是Linux版本不对或者Nginx的版本问题，需要自行在命令行环境中，在能够正确执行中间件重启的命令，进行替换，该命令在脚本的第211行代码中。