bt-source/panel/class/acme_v2.py

#!/usr/bin/python
#coding: utf-8
# -------------------------------------------------------------------
# 宝塔Linux面板
# -------------------------------------------------------------------
# Copyright (c) 2015-2099 宝塔软件(http://bt.cn) All rights reserved.
# -------------------------------------------------------------------
# Author: hwliang <hwl@bt.cn> a
# -------------------------------------------------------------------

# -------------------------------------------------------------------
# ACME v2客户端
# -------------------------------------------------------------------
import warnings
warnings.filterwarnings("ignore", message=r".*doesn't\s+match\s+a\s+supported\s+version", module="requests")
import re
import fcntl
import datetime
import binascii
import hashlib
import base64
import json
import shutil
import time
import os
import sys
import uuid

os.chdir('/www/server/panel')
if not 'class/' in sys.path:
    sys.path.insert(0,'class/')
import http_requests as requests
# requests.DEFAULT_TYPE = 'curl'
import public

try:
    import OpenSSL
except:
    public.ExecShell("btpip install pyopenssl")
    import OpenSSL
try:
    import dns.resolver
except:
    public.ExecShell("btpip install dnspython")
    import dns.resolver

import ssl_info
####
# auth to 格式说明
# 旧版 auth to 格式：
# 文件验证：/www/server/xxxx/xxxx
# DNS->手动：dns  DNS->api: CloudFlareDns|XXXXXXXX|XXXXXXXXX
#
# 新版 auth to 格式：
# DNS->手动：dns DNS->api: dns#@api

url_dic = {
    "letsencrypt": "https://acme-v02.api.letsencrypt.org/directory",
    "letsencrypt_staging": "https://acme-staging-v02.api.letsencrypt.org/directory",
    "litessl": "https://acme.litessl.com/acme/v2/directory"
}

class acme_v2:
    _url = None
    _apis = None
    _config = {}
    _dns_domains = []
    _bits = 2048
    _acme_timeout = 30
    _dns_class = None
    _user_agent = "BaoTa/1.0 (+https://www.bt.cn)"
    _replay_nonce = None
    _verify = False
    _digest = "sha256"
    _max_check_num = 15
    _wait_time = 5
    _mod_index = {True: "Staging", False: "Production", "litessl": "litessl", "letsencrypt": "Production"}
    _debug = False
    _auto_wildcard = False
    _dnsapi_file = 'config/dns_api.json'
    _save_path = 'vhost/letsencrypt'
    _conf_file = 'config/letsencrypt.json'
    _conf_file_v2 = 'config/letsencrypt_v2.json'
    _request_type = 'curl'

    def __init__(self):
        if not os.path.exists(self._conf_file_v2) and os.path.exists(self._conf_file):
            shutil.copyfile(self._conf_file, self._conf_file_v2)
        if self._debug:
            self._url = 'https://acme-staging-v02.api.letsencrypt.org/directory'
            if self._debug == 'litessl':
                self._url = 'https://acme.litessl.com/acme/v2/directory'
        else:
            self._url = 'https://acme-v02.api.letsencrypt.org/directory'
        self._config = self.read_config()
        self._nginx_cache_file_auth = {}
        self._can_use_lua = None
        self._well_known_check_cache = {}

    def can_use_lua_module(self):
        if self._can_use_lua is None:
            # 查询lua_module 不为空
            self._can_use_lua = public.ExecShell("nginx -V 2>&1 |grep lua_nginx_module")[0].strip() != ''
        return self._can_use_lua

    # 返回是否能通过lua 做了文件验证处理， 如果返回True，则表示可以处理了验证文件， 不再走之前的 if 验证方式
    def can_use_lua_for_site(self, site_name: str, site_type: str):
        if self._can_use_lua is None:
            # 查询lua_module 不为空
            self._can_use_lua = public.ExecShell("nginx -V 2>&1 |grep lua_nginx_module")[0].strip() != ''

        if not self._can_use_lua:
            return False

        if site_type.lower() in ("php", "proxy", "wp2"):
            prefix = ""
        else:
            prefix = site_type.lower() + "_"

        ng_file = "{}/nginx/{}{}.conf".format(public.get_vhost_path(), prefix, site_name)
        ng_data = public.readFile(ng_file)
        if not ng_data:
            return False

        rep_well_known = re.compile(
            r"(#.*\n)?\s*include\s+/www/server/panel/vhost/nginx/well-known/.*\.conf;.*\n(#.*\n)?"
            r"(.*\n)*?\s*#error_page 404/404\.html;"
        )   # 匹配一下引入的外部配置文件，同时保证这个配置在SSL配置之前， 这样避免路由匹配问题

        if rep_well_known.search(ng_data):
            lua_file = "{}/nginx/well-known/{}.conf".format(public.get_vhost_path(), site_name)
            lua_data = public.readFile(lua_file)
            if not lua_data or "set_by_lua_block $well_known" not in lua_data:
                return False
            else:
                return True
        else:
            return False

    # 返回是否能通过if 判断方式做了文件验证处理， 如果返回True，则表示可以
    @staticmethod
    def can_use_if_for_file_check(site_name: str, site_type: str):
        if site_type.lower() in ("php", "proxy", "wp2"):
            prefix = ""
        else:
            prefix = site_type.lower() + "_"

        # if 方式的文件验证必须是可重载的情况
        if public.checkWebConfig() is not True:
            return False

        ng_file = "{}/nginx/{}{}.conf".format(public.get_vhost_path(), prefix, site_name)
        ng_data = public.readFile(ng_file)
        if not ng_data:
            return False

        rep_well_known = re.compile(
            r"(#.*\n)?\s*include\s+/www/server/panel/vhost/nginx/well-known/.*\.conf;.*\n(#.*\n)?"
            r"(.*\n)*?\s*#error_page 404/404\.html;"
        )   # 匹配一下引入的外部配置文件，同时保证这个配置在SSL配置之前， 这样避免路由匹配问题

        if rep_well_known.search(ng_data):
            return True
        else:
            return False

    # 返回配置文件是否支持使用普通的文件验证
    @staticmethod
    def can_use_base_file_check(site_name: str, site_type: str):
        if site_type.lower() in ("php", "proxy", "wp2"):
            prefix = ""
        else:
            prefix = site_type.lower() + "_"

        webserver = public.get_webserver()
        if webserver == "nginx":
            ng_file = "{}/nginx/{}{}.conf".format(public.get_vhost_path(), prefix, site_name)
            ng_data = public.readFile(ng_file)
            if not ng_data:
                return False
            rep_well_known = re.compile(r"location\s+([=~^]*\s*)?/?\\?\.well-known/?\s*{")
            if rep_well_known.search(ng_data):
                return True
            else:
                return False
        elif webserver == "apache" and prefix:  # PHP 不用检查
            ap_file = "{}/apache/{}{}.conf".format(public.get_vhost_path(), prefix, site_name)
            ap_data = public.readFile(ap_file)
            if not ap_data:
                return False
            rep_well_known_list = [
                re.compile(r"<IfModule\s+alias_module>\s+Alias\s+/\.well-known/\s+\S+\s+</IfModule>"),
                re.compile(r"\s*ProxyPass\s+/\.well-known/\s+!", re.M),
            ]
            for rep_well_known in rep_well_known_list:
                if rep_well_known.search(ap_data):
                    return True
            return False
        return True

    # 取接口目录
    def get_apis(self):
        # if not self._apis:
        # 尝试从配置文件中获取
        api_index = self._mod_index[self._debug]
        if not 'apis' in self._config:
            self._config['apis'] = {}
        if api_index in self._config['apis']:
            if 'expires' in self._config['apis'][api_index] and 'directory' in self._config['apis'][api_index]:
                if time.time() < self._config['apis'][api_index]['expires']:
                    self._apis = self._config['apis'][api_index]['directory']
                    return self._apis

        # 尝试从云端获取
        res = requests.get(self._url,s_type=self._request_type)
        if not res.status_code in [200, 201]:
            result = res.json()
            if "type" in result:
                if result['type'] == 'urn:acme:error:serverInternal':
                    raise Exception('服务因维护而关闭或发生内部错误，查看 <a href="https://letsencrypt.status.io/" target="_blank" class="btlink">https://letsencrypt.status.io/</a> 了解更多详细信息。')
            raise Exception(res.content)
        s_body = res.json()
        self._apis = {}
        self._apis['newAccount'] = s_body['newAccount']
        self._apis['newNonce'] = s_body['newNonce']
        self._apis['newOrder'] = s_body['newOrder']
        self._apis['revokeCert'] = s_body['revokeCert']
        self._apis['keyChange'] = s_body['keyChange']

        # 保存到配置文件
        self._config['apis'][api_index] = {}
        self._config['apis'][api_index]['directory'] = self._apis
        self._config['apis'][api_index]['expires'] = time.time() + \
            86400  # 24小时后过期
        self.save_config()
        return self._apis

    # 获取帐户信息
    def get_account_info(self, args):
        try:
            if not 'account' in self._config:
                return {}
            k = self._mod_index[self._debug]
            if not k in self._config['account']:
                self.get_apis()
                self.get_kid()
            account = self._config['account'][k]
            account['email'] = self._config['email']
            self.set_crond()
            return account
        except Exception as ex:
            return public.returnMsg(False,str(ex))

    # 设置帐户信息
    def set_account_info(self, args):
        if not 'account' in self._config:
            return public.returnMsg(False, '指定帐户不存在')
        account = json.loads(args.account)
        if 'email' in account:
            self._config['email'] = account['email']
            del(account['email'])
        self._config['account'][self._mod_index[self._debug]] = account
        self.save_config()
        return public.returnMsg(True, '帐户设置成功!')

    # 获取订单列表
    def get_orders(self, args):
        if not 'orders' in self._config:
            return []
        s_orders = []
        for index in self._config['orders'].keys():
            tmp_order = self._config['orders'][index]
            tmp_order['index'] = index
            s_orders.append(tmp_order)
        return s_orders

    # 删除订单
    def remove_order(self, args):
        if not 'orders' in self._config:
            return public.returnMsg(False, '指定订单不存在!')
        if not args.index in self._config['orders']:
            return public.returnMsg(False, '指定订单不存在!')
        del(self._config['orders'][args.index])
        self.save_config()
        return public.returnMsg(True, '订单删除成功!')

    # 取指定订单数据
    def get_order_find(self, args):
        if not 'orders' in self._config:
            return public.returnMsg(False, '指定订单不存在!')
        if not args.index in self._config['orders']:
            return public.returnMsg(False, '指定订单不存在!')
        result = self._config['orders'][args.index]
        result['cert'] = self.get_cert_info(args.index)
        return result

    # 获取证书信息
    def get_cert_info(self, index):
        cert = {}
        path = self._config['orders'][index]['save_path']
        if not os.path.exists(path):
            self.download_cert(index)
        cert['private_key'] = public.readFile(path + "/privkey.pem")
        cert['fullchain'] = public.readFile(path + "/fullchain.pem")
        return cert

    # 更新证书压缩包
    def update_zip(self, args):
        path = self._config['orders'][args.index]['save_path']
        if not os.path.exists(path):  # 尝试重新下载证书
            self.download_cert(args.index)
        if not os.path.exists(path):
            return public.returnMsg(False, '证书读取失败，目录不存在!')
        import panelTask
        bt_task = panelTask.bt_task()
        zip_file = path+'/cert.zip'
        result = bt_task._zip(path, '.', path+'/cert.zip', '/dev/null', 'zip')
        if not os.path.exists(zip_file):
            return result
        return public.returnMsg(True, zip_file)

    # 吊销证书
    def revoke_order(self, index):
        if type(index) != str:
            index = index.index
        if not index in self._config['orders']:
            raise Exception("指定订单不存在!")
        cert_path = self._config['orders'][index]['save_path']
        if not os.path.exists(cert_path):
            raise Exception("指定订单没有找到可用的证书!")
        cert = self.dump_der(cert_path)
        if not cert:
            raise Exception("证书读取失败!")
        payload = {
            "certificate": self.calculate_safe_base64(cert),
            "reason": 4
        }
        res = self.acme_request(self._apis['revokeCert'], payload)
        if res.status_code in [200, 201]:
            if os.path.exists(cert_path):
                public.ExecShell("rm -rf {}".format(cert_path))
            del(self._config['orders'][index])
            self.save_config()
            return public.returnMsg(True, "证书吊销成功!")
        return res.json()

    # 取根域名和记录值
    def extract_zone(self, domain_name):
        return public.split_domain_sld(domain_name)
        # top_domain_list = public.readFile('{}/config/domain_root.txt'.format(public.get_panel_path()))
        # if top_domain_list:
        #     top_domain_list = top_domain_list.strip().split('\n')
        # else:
        #     top_domain_list = []
        # old_domain_name = domain_name
        # top_domain = "."+".".join(domain_name.rsplit('.')[-2:])
        # new_top_domain = "." + top_domain.replace(".", "")
        # is_tow_top = False
        # if top_domain in top_domain_list:
        #     is_tow_top = True
        #     domain_name = domain_name[:-len(top_domain)] + new_top_domain
        #
        # if domain_name.count(".") > 1:
        #     zone, middle, last = domain_name.rsplit(".", 2)
        #     if is_tow_top:
        #         last = top_domain[1:]
        #     root = ".".join([middle, last])
        # else:
        #     zone = ""
        #     root = old_domain_name
        # return root, zone

    # 自动构造通配符
    def auto_wildcard(self, domains):
        if not domains:
            return domains
        domain_list = []
        for domain in domains:
            root, zone = self.extract_zone(domain)
            tmp_list = zone.split(".", 1)
            if len(tmp_list) == 1:
                if root not in domain_list:
                    domain_list.append(root)
                if not "*." + root in domain_list:
                    domain_list.append("*." + root)
            else:
                new_root = "{}.{}".format(tmp_list[1], root)
                if new_root not in domain_list:
                    domain_list.append(new_root)
                if not "*." + new_root in domain_list:
                    domain_list.append("*." + new_root)
        return domain_list

    # 构造域名列表
    def format_domains(self, domains):
        if type(domains) != list:
            return []
        # 是否自动构造通配符
        if self._auto_wildcard:
            domains = self.auto_wildcard(domains)
        wildcard = []
        tmp_domains = []
        for domain in domains:
            domain = domain.strip()
            if domain in tmp_domains:
                continue
            # 将通配符域名转为验证正则表达式
            f_index = domain.find("*.")
            if f_index not in [-1, 0]:
                continue
            if f_index == 0:
                wildcard.append(domain.replace(
                    "*", r"^[\w-]+").replace(".", r"\."))
            # 添加到申请列表
            tmp_domains.append(domain)

        # 处理通配符包含
        apply_domains = tmp_domains[:]
        for domain in tmp_domains:
            for w in wildcard:
                if re.match(w, domain):
                    apply_domains.remove(domain)

        return apply_domains

    # 创建订单
    def create_order(self, domains, auth_type, auth_to, index=None):
        domains = self.format_domains(domains)
        if not domains:
            raise Exception("至少需要有一个域名或者IP")
        # 判断是否全部为IP或域名
        # 构造标识
        identifiers = []
        all_is_ip = set()
        for domain_name in domains:
            is_ip = public.check_ip(domain_name)
            all_is_ip.add(is_ip)
            if is_ip:
                identifiers.append({ "type": "ip", "value": domain_name })
                continue
            identifiers.append({"type": 'dns', "value": domain_name})
        payload = {"identifiers": identifiers}
        if len(all_is_ip) != 1:
            raise Exception("域名和IP不能同时申请证书!")
        all_is_ip = next(iter(all_is_ip))
        if all_is_ip:
            payload = {"identifiers": identifiers, "profile": "shortlived"}
            if auth_type != 'http':
                raise Exception("IP地址只能使用HTTP文件验证方式申请证书!")

        # 请求创建订单
        res = self.acme_request(self._apis['newOrder'], payload)
        if not res.status_code in [201,200]:  # 如果创建失败
            e_body = res.json()
            if 'type' in e_body:
                # 如果随机数失效
                if e_body['type'].find('error:badNonce') != -1:
                    self.get_nonce(force=True)
                    res = self.acme_request(self._apis['newOrder'], payload)

                # 如果帐户失效
                if e_body['detail'].find('KeyID header contained an invalid account URL') != -1:
                    k = self._mod_index[self._debug]
                    del(self._config['account'][k])
                    self.get_kid()
                    self.get_nonce(force=True)
                    res = self.acme_request(self._apis['newOrder'], payload)
            if not res.status_code in [201,200]:
                a_auth = res.json()

                ret_title = self.get_error(str(a_auth))
                raise StopIteration(
                        "{0} >>>> {1}".format(
                            ret_title,
                            json.dumps(a_auth)
                        )
                    )

        # 返回验证地址和验证
        s_json = res.json()
        s_json['auth_type'] = auth_type
        s_json['url'] = res.headers.get("Location")
        s_json['ca'] = self._debug
        s_json['domains'] = domains
        s_json['auth_to'] = auth_to
        index = self.save_order(s_json, index)
        return index

    def get_site_run_path_byid(self,site_id):
        '''
            @name 通过site_id获取网站运行目录
            @author hwliang
            @param site_id<int> 网站标识
            @return None or string
        '''
        if public.M('sites').where('id=? and project_type=?', (site_id, 'PHP')).count()>=1:
            site_path = public.M('sites').where('id=?',site_id).getField('path')
            if not site_path: return None
            if not os.path.exists(site_path): return None
            args = public.dict_obj()
            args.id = site_id
            import panelSite
            run_path = panelSite.panelSite().GetRunPath(args)
            if run_path in ['/']: run_path = ''
            if run_path:
                if run_path[0] == '/': run_path = run_path[1:]
            site_run_path = os.path.join(site_path,run_path)
            if not os.path.exists(site_run_path): return site_path
            return site_run_path
        else:
            find = public.M('sites').where('id=?',site_id).find()
            if not find:
                return False
            count = public.M('sites').where(
                'id=? and project_type in (?,?,?,?)',
                (site_id, 'Java', 'Go', 'Other', "Python")
            ).count()
            if count:
                try:
                    project_info = json.loads(find['project_config'])
                    if 'ssl_path' not in project_info:
                        ssl_path = '/www/wwwroot/java_node_ssl'
                    else:
                        ssl_path = project_info['ssl_path']
                    if not os.path.exists(ssl_path):
                        os.makedirs(ssl_path)
                    return ssl_path
                except:
                    return False
            else:
                import panelSite
                auth_to = find['path'] + '/' + panelSite.panelSite().GetRunPath(public.to_dict_obj({"id": site_id}))
                auth_to = auth_to.replace("//", "/")
                if auth_to[-1] == '/':
                    auth_to = auth_to[:-1]
                if not os.path.exists(auth_to):
                    return False
                return auth_to


    def get_site_run_path(self,domains):
        '''
            @name 通过域名列表获取网站运行目录
            @author hwliang
            @param domains<list> 域名列表
            @return None or string
        '''
        site_id = 0
        for domain in domains:
            site_id = public.M('domain').where("name=?",domain).getField('pid')
            if site_id: break

        if not site_id: return None
        return self.get_site_run_path_byid(site_id)

    # 获取验证信息
    def get_auths(self, index):
        if not index in self._config['orders']:
            raise Exception('指定订单不存在!')

        # 检查是否已经获取过授权信息
        if 'auths' in self._config['orders'][index]:
            # 检查授权信息是否过期
            if time.time() < self._config['orders'][index]['auths'][0]['expires']:
                return self._config['orders'][index]['auths']
        if self._config['orders'][index]['auth_type'] != 'dns':
            site_run_path = self.get_site_run_path(self._config['orders'][index]['domains'])
            if site_run_path: self._config['orders'][index]['auth_to'] = site_run_path

        #清理旧验证
        self.claer_auth_file(index)

        auths = []
        for auth_url in self._config['orders'][index]['authorizations']:
            res = self.acme_request(auth_url, "")
            if res.status_code not in [200, 201]:
                raise Exception("获取授权失败: {}".format(res.json()))

            s_body = res.json()
            if 'status' in s_body:
                if s_body['status'] in ['invalid']:
                    raise Exception("无效订单，此订单当前为验证失败状态!")
                if s_body['status'] in ['valid']:  # 跳过无需验证的域名
                    continue

            s_body['expires'] = self.utc_to_time(s_body['expires'])
            identifier_auth = self.get_identifier_auth(index, auth_url, s_body)
            if not identifier_auth:
                raise Exception("验证信息构造失败!{}")

            acme_keyauthorization, auth_value = self.get_keyauthorization(
                identifier_auth['token'])
            identifier_auth['acme_keyauthorization'] = acme_keyauthorization
            identifier_auth['auth_value'] = auth_value
            identifier_auth['expires'] = s_body['expires']
            identifier_auth['auth_to'] = self._config['orders'][index]['auth_to']
            identifier_auth['type'] = self._config['orders'][index]['auth_type']

            # 设置验证信息
            set_auth = True
            for auth in auths:
                if (
                        auth['domain'].replace("*.", "") == identifier_auth['domain'].replace("*.", "")
                        and auth['auth_value'] == identifier_auth['auth_value']
                ):
                    set_auth = False
                    break
            if set_auth:
                self.set_auth_info(identifier_auth, index=index)
            auths.append(identifier_auth)
        self._config['orders'][index]['auths'] = auths
        self.save_config()
        if not self.check_config(index, "auths", auths):
            self.save_config()
        return auths

    # 更新随机数
    def update_replay_nonce(self, res):
        replay_nonce = res.headers.get('Replay-Nonce')
        if replay_nonce:
            self._replay_nonce = replay_nonce

    # 设置验证信息
    def set_auth_info(self, identifier_auth, index=None):

        #从云端验证
        if not self.cloud_check_domain(identifier_auth['domain']):
            self.err = "云端验证失败!"

        # 是否手动验证DNS
        if identifier_auth['auth_to'] == 'dns':
            return None

        # 是否文件验证
        if identifier_auth['type'] in ['http', 'tls']:
            self.write_auth_file(
                identifier_auth['auth_to'], identifier_auth['token'], identifier_auth['acme_keyauthorization'], index)
        else:
            # dnsapi验证
            self.create_dns_record(
                identifier_auth['auth_to'], identifier_auth['domain'], identifier_auth['auth_value'])

    #从云端验证域名是否可访问
    def cloud_check_domain(self,domain):
        try:
            result = requests.post('https://www.bt.cn/api/panel/check_domain',{"domain":domain,"ssl":1},s_type=self._request_type).json()
            return result['status']
        except: return False


    #清理验证文件
    def claer_auth_file(self,index):
        if not self._config['orders'][index]['auth_type'] in ['http','tls']:
            return True
        acme_path = '{}/.well-known/acme-challenge'.format(self._config['orders'][index]['auth_to'])
        acme_path = acme_path.replace("//",'/')
        write_log("|-验证目录：{}".format(acme_path))
        if os.path.exists(acme_path):
            public.ExecShell("rm -f {}/*".format(acme_path))

        acme_path = '/www/server/stop/.well-known/acme-challenge'
        if os.path.exists(acme_path):
            public.ExecShell("rm -f {}/*".format(acme_path))

    def change_well_known_mod(self, path_dir: str):
        path_dir = path_dir.rstrip("/")
        if not os.path.isdir(path_dir):
            return False
        if path_dir in self._well_known_check_cache:
            return True
        else:
            self._well_known_check_cache[path_dir] = True
        import stat

        try:
            import pwd
            uid_data = pwd.getpwnam("www")
            uid = uid_data.pw_uid
            gid = uid_data.pw_gid
        except:
            return

        # 逐级给最低访问权限
        while path_dir != "/":
            path_dir_stat = os.stat(path_dir)
            if path_dir_stat.st_uid == 0 and uid != 0:
                old_mod = stat.S_IMODE(path_dir_stat.st_mode)
                if not old_mod & (1 << 3):
                    os.chmod(path_dir, old_mod + (1 << 3))  # chmod g+x
            if path_dir_stat.st_uid == uid:
                old_mod = stat.S_IMODE(path_dir_stat.st_mode)
                if not old_mod & (1 << 6):
                    os.chmod(path_dir, old_mod + (1 << 6))  # chmod u+x
            elif path_dir_stat.st_gid == gid:
                old_mod = stat.S_IMODE(path_dir_stat.st_mode)
                if not old_mod & (1 << 3):
                    os.chmod(path_dir, old_mod + (1 << 6))  # chmod g+x
            elif path_dir_stat.st_uid != uid or path_dir_stat.st_gid != gid:
                old_mod = stat.S_IMODE(path_dir_stat.st_mode)
                if not old_mod & 1:
                    os.chmod(path_dir, old_mod+1)   # chmod o+x
            path_dir = os.path.dirname(path_dir)

    # 写验证文件
    def write_auth_file(self, auth_to, token, acme_keyauthorization, index):
        if public.get_webserver() == "nginx":
            # 如果是nginx尝试使用配置文件进行验证
            self.write_ngin_authx_file(auth_to, token, acme_keyauthorization, index)

        # 尝试写文件进行验证
        try:
            acme_path = '{}/.well-known/acme-challenge'.format(auth_to)
            acme_path = acme_path.replace("//",'/')
            if not os.path.exists(acme_path):
                os.makedirs(acme_path)
                public.set_own(acme_path, 'www')
            self.change_well_known_mod(acme_path)
            wellknown_path = '{}/{}'.format(acme_path, token)
            public.writeFile(wellknown_path, acme_keyauthorization)
            public.set_own(wellknown_path, 'www')

            acme_path = '/www/server/stop/.well-known/acme-challenge'
            if not os.path.exists(acme_path):
                os.makedirs(acme_path)
                public.set_own(acme_path, 'www')
            self.change_well_known_mod(acme_path)
            wellknown_path = '{}/{}'.format(acme_path,token)
            public.writeFile(wellknown_path,acme_keyauthorization)
            public.set_own(wellknown_path, 'www')
            return True
        except:
            err = public.get_error_info()
            print(err)
            raise Exception("写入验证文件失败: {}".format(err))

    def write_ngin_authx_file(self, auth_to, token, acme_keyauthorization, index):
        site_name, project_type = self.get_site_name_by_domains(self._config["orders"][index]["domains"])
        if site_name is None:
            return

        if self.can_use_lua_for_site(site_name, project_type):
            return

        if project_type.lower() in ("php", "proxy"):
            nginx_conf_path = "{}/vhost/nginx/{}.conf".format(public.get_panel_path(), site_name)
        else:
            nginx_conf_path = "{}/vhost/nginx/{}_{}.conf".format(public.get_panel_path(), project_type.lower(), site_name)
        nginx_conf = public.readFile(nginx_conf_path)
        if nginx_conf is False:
            return

        file_check_config_path = "/www/server/panel/vhost/nginx/well-known/{}.conf".format(site_name)
        if not os.path.exists("/www/server/panel/vhost/nginx/well-known"):
            os.makedirs("/www/server/panel/vhost/nginx/well-known", 0o755)

        # 如果主配置中，没有引用则尝试添加，添加失败就跳出
        if not re.search(r"\s*include\s+/www/server/panel/vhost/nginx/well-known/.*\.conf;", nginx_conf, re.M):
            ssl_line = re.search(r"(#.*\n\s*)?#error_page 404/404\.html;", nginx_conf)
            if ssl_line is None:
                return
            default_cert_apply_check = (
                "#CERT-APPLY-CHECK--START\n"
                "    # 用于SSL证书申请时的文件验证相关配置 -- 请勿删除\n"
                "    include /www/server/panel/vhost/nginx/well-known/{}.conf;\n"
                "    #CERT-APPLY-CHECK--END\n    "
            ).format(site_name)
            if not os.path.exists(file_check_config_path):
                public.writeFile(file_check_config_path, "")

            new_conf = nginx_conf.replace(ssl_line.group(), default_cert_apply_check + ssl_line.group(), 1)
            public.writeFile(nginx_conf_path, new_conf)
            isError = public.checkWebConfig()
            if isError is not True:
                public.writeFile(nginx_conf_path, nginx_conf)
                return

        # 如果主配置有引用， 不再检测位置关系，因为不能保证用户的自定义配置的优先级， 直接进行文件验证的 lua 方式和 if 方式的尝试
        if self.can_use_lua_module():
            self.write_lua_file_for_site(file_check_config_path)
            return

        # 开始尝试if 验证方式
        if auth_to not in self._nginx_cache_file_auth:
            self._nginx_cache_file_auth[auth_to] = []

        self._nginx_cache_file_auth[auth_to].append((token, acme_keyauthorization))

        tmp_data = []
        for token, acme_key in self._nginx_cache_file_auth[auth_to]:
            tmp_data.append((
                'if ($request_uri ~ "^/\.well-known/acme-challenge/{}.*"){{\n'
                '    return 200 "{}";\n'
                '}}\n'
            ).format(token, acme_key))

        public.writeFile(file_check_config_path, "\n".join(tmp_data))
        isError = public.checkWebConfig()
        if isError is True:
            public.serviceReload()
        else:
            public.writeFile(file_check_config_path, "")

    @staticmethod
    def write_lua_file_for_site(file_check_config_path: str):
        old_data = public.readFile(file_check_config_path)
        if isinstance(old_data, str) and "set_by_lua_block $well_known" in old_data:
            return

        lua_file_data = """
set $well_known '';
if ( $uri ~ "^/.well-known/" ) {
  set_by_lua_block $well_known { 
    -- 获取路径
    local m,err = ngx.re.match(ngx.var.uri,"/.well-known/(.*)","isjo")
    -- 如果路径匹配
    if m then
      -- 拼接文件路径
      local filename = ngx.var.document_root .. m[0]
      -- 判断文件路径中是否合法
      if not ngx.re.find(m[1],"\\\\./","isjo") then
        -- 判断文件是否存在
        local is_exists = io.open(filename, "r")
        if not is_exists then
            -- Java项目?
            filename = "/www/wwwroot/java_node_ssl" ..  m[0]
        end
        -- 释放
        if is_exists then is_exists:close() end
        -- 读取文件
        local fp = io.open(filename,'r')
        if fp then
          local file_body = fp:read("*a")
          fp:close()
          if file_body then
            ngx.header['content-type'] = 'text/plain'
            return file_body
          end
        end
      end
    end
    return ""
  }
}

if ( $well_known != "" ) {
  return 200 $well_known;
}
"""
        public.writeFile(file_check_config_path, lua_file_data)
        isError = public.checkWebConfig()
        if isError is True:
            public.serviceReload()
        else:
            public.writeFile(file_check_config_path, old_data)

    # 解析域名 新的 前端修改完用这个
    # def create_dns_record(self, auth_to, domain, dns_value):
    #     # 如果为手动解析
    #     if auth_to == 'dns':
    #         return None
    #
    #     from panelDnsapi import DnsMager
    #
    #     self._dns_class = DnsMager().get_dns_obj_by_domain(domain)
    #     self._dns_class.create_dns_record(public.de_punycode(domain), dns_value)
    #     self._dns_domains.append({"domain": domain, "dns_value": dns_value})

    # 解析域名
    def create_dns_record(self, auth_to, domain, dns_value):
        # 如果为手动解析
        if auth_to == 'dns':
            return None

        if auth_to.find('|') != -1:
            import panelDnsapi
            dns_name, key, secret = self.get_dnsapi(auth_to)
            self._dns_class = getattr(panelDnsapi, dns_name)(key, secret)
            self._dns_class.create_dns_record(public.de_punycode(domain), dns_value)
        # else:
        #     from panelDnsapi import DnsMager
        #     self._dns_class = DnsMager().get_dns_obj_by_domain(domain)
        # self._dns_class.create_dns_record(public.de_punycode(domain), dns_value)
        # self._dns_domains.append({"domain": domain, "dns_value": dns_value})
        else:
            from sslModel import dataModel
            dataModel.main().add_dns_value_by_domain(domain, dns_value, is_let_txt=True)
        self._dns_domains.append({"domain": domain, "dns_value": dns_value})
        return

    # 解析域名
    # def create_dns_record(self, auth_to, domain, dns_value):
    #     # 如果为手动解析
    #     if auth_to == 'dns' or auth_to.find('|') == -1:
    #         return None
    #
    #     import panelDnsapi
    #     dns_name, key, secret = self.get_dnsapi(auth_to)
    #     self._dns_class = getattr(panelDnsapi, dns_name)(key, secret)
    #     self._dns_class.create_dns_record(public.de_punycode(domain), dns_value)
    #     self._dns_domains.append({"domain": domain, "dns_value": dns_value})
    #     return

    # 解析DNSAPI信息  # 不再使用的
    def get_dnsapi(self, auth_to):
        tmp = auth_to.split('|')
        dns_name = tmp[0]
        key = "None"
        secret = "None"
        if len(tmp) < 3:
            try:
                dnsapi_config = json.loads(public.readFile(self._dnsapi_file))
                for dc in dnsapi_config:
                    if dc['name'] != dns_name:
                        continue
                    if not dc['data']:
                        continue
                    key = dc['data'][0]['value']
                    secret = dc['data'][1]['value']
            except:
                raise Exception("没有找到有效的DNSAPI密钥信息")
        else:
            key = tmp[1]
            secret = tmp[2]
        return dns_name, key, secret

    # 删除域名解析
    def remove_dns_record(self):
        if not self._dns_domains:
            return None
        for dns_info in self._dns_domains:
            try:
                if self._dns_class:
                    self._dns_class.delete_dns_record(
                        public.de_punycode(dns_info['domain']), dns_info['dns_value'])
                else:
                    from sslModel import dataModel
                    dataModel.main().del_dns_value_by_domain(dns_info['domain'], is_let_txt=True)
            except Exception as e:
                pass

    # 验证域名
    def auth_domain(self, index):
        self._config['orders'][index]['auth_tag'] = True
        self.save_config()
        if index not in self._config['orders']:
            raise Exception('指定订单不存在!')

        if "auths" not in self._config['orders'][index]:
            raise Exception('订单验证信息丢失，请尝试重新申请!')

        # 开始验证
        for auth in self._config['orders'][index]['auths']:
            res = self.check_auth_status(auth['url'])  # 检查是否需要验证
            if res.json()['status'] == 'invalid':
                raise Exception('域名验证失败，请尝试重新申请!')
            if res.json()['status'] == 'pending':
                if auth['type'] == 'dns':  # 尝试提前验证dns解析
                    self.check_dns(
                        "_acme-challenge.{}".format(
                            auth['domain'].replace('*.', '')),
                        auth['auth_value'],
                        "TXT"
                    )
                self.respond_to_challenge(auth)

        # 检查验证结果
        for i in range(len(self._config['orders'][index]['auths'])):
            self.check_auth_status(self._config['orders'][index]['auths'][i]['url'], [
                                   'valid', 'invalid'])
            self._config['orders'][index]['status'] = 'valid'

    def auth_domain_api(self, get):
        """
        验证单个域名
        """
        index = get.index
        domain = get.domain

        write_log("|-正在验证域名：{}".format(domain))
        if index not in self._config['orders']:
            return public.returnMsg(False, '指定订单不存在!')
        order = self._config['orders'][index]
        if "ca" in order:
            self._debug = order['ca']
            self._url = url_dic.get(order['ca'], url_dic['letsencrypt'])
        self.get_apis()

        if "auths" not in order:
            return public.returnMsg(False, '订单验证信息丢失，请尝试重新申请!')

        for auth in order['auths']:
            if auth.get("status") == "invalid":
                return public.returnMsg(False, "域名验证失败，请尝试重新申请！")
            if domain and auth['domain'] != domain:
                continue
            try:
                res = self.check_auth_status(auth['url'])  # 检查是否需要验证
                if res.json()['status'] == 'pending':
                    self.respond_to_challenge(auth)
                    _return = self.check_auth_status(auth['url'], ['valid', 'invalid']).json()
                    auth['status'] = _return['status']
                    return auth
                auth['status'] = res.json()['status']
                write_log("|-验证成功!")
                return auth
            except StopIteration as e:
                auth['status'] = 'invalid'
                ex = str(e)
                if ex.find(">>>>") != -1:
                    msg = ex.split(">>>>")
                    msg[1] = json.loads(msg[1])
                else:
                    msg = ex
                    write_log(public.get_error_info())
                auth['error'] = msg
                return public.returnMsg(False, msg)
            finally:
                self.save_config()
                if "pending" not in [i.get("status", "pending") for i in order['auths']]:
                    return self.apply_dns_auth(get)
        write_log("|-没有在订单中找到此域名：{}".format(domain))
        return public.returnMsg(False, "没有在订单中找到此域名：{}".format(domain))

    # 检查验证状态
    def check_auth_status(self, url, desired_status=None):
        desired_status = desired_status or ["pending", "valid", "invalid"]
        number_of_checks = 0
        while True:
            if desired_status == ['valid', 'invalid']:
                write_log("|-第{}次查询验证结果..".format(number_of_checks + 1))
                time.sleep(self._wait_time)
            check_authorization_status_response = self.acme_request(url, "")
            a_auth = check_authorization_status_response.json()
            if 'type' in a_auth and a_auth['type'].find('error:badNonce') != -1:
                # 如果随机数失效
                self.get_nonce(force=True)
                a_auth = self.acme_request(url, "").json()
            if not isinstance(a_auth, dict):
                write_log(a_auth)
                continue
            authorization_status = a_auth["status"]
            number_of_checks += 1
            if authorization_status in desired_status:
                if authorization_status == "invalid":
                    write_log("|-验证失败!")
                    try:
                        if 'error' in a_auth['challenges'][0]:
                            ret_title = a_auth['challenges'][0]['error']['detail']
                        elif 'error' in a_auth['challenges'][1]:
                            ret_title = a_auth['challenges'][1]['error']['detail']
                        elif 'error' in a_auth['challenges'][2]:
                            ret_title = a_auth['challenges'][2]['error']['detail']
                        else:
                            ret_title = str(a_auth)
                        ret_title = self.get_error(ret_title)
                    except:
                        ret_title = str(a_auth)
                    raise StopIteration(
                        "{0} >>>> {1}".format(
                            ret_title,
                            json.dumps(a_auth)
                        )
                    )
                break

            if number_of_checks == self._max_check_num:
                raise StopIteration(
                    "错误：已尝试验证{0}次. 最大验证次数为{1}. 验证时间间隔为{2}秒.".format(
                        number_of_checks,
                        self._max_check_num,
                        self._wait_time
                    )
                )
        if desired_status == ['valid', 'invalid']:
            write_log("|-验证成功!")
        return check_authorization_status_response

    # 格式化错误输出
    def get_error(self, error):
        write_log("error_result: " + str(error))
        if error.find("Max checks allowed") >= 0:
            return "CA无法验证您的域名，请检查域名解析是否正确，或等待5-10分钟后重试."
        elif error.find("Max retries exceeded with") >= 0 or error.find('status_code=0 ') != -1:
            return "CA服务器连接超时，请稍候重试."
        elif error.find("The domain name belongs") >= 0:
            return "域名不属于此DNS服务商，请确保域名填写正确."
        elif error.find("domains in the last 168 hours") != -1 and error.find("Error creating new order") != -1:
            return "签发失败,该域名%s的根域名超出了每周的最大签发次数限制!" % re.findall("hours:\s+(.+?),", error)
        elif error.find('login token ID is invalid') >= 0:
            return 'DNS服务器连接失败，请检查密钥是否正确.'
        elif error.find('Error getting validation data') != -1:
            return '数据验证失败，CA无法从验证连接中获到正确的验证码.'
        elif "too many certificates already issued for exact set of domains" in error:
            return '签发失败,该域名%s超出了每周的重复签发次数限制!' % re.findall("exact set of domains: (.+):", error)
        elif "Error creating new account :: too many registrations for this IP" in error:
            return '签发失败,当前服务器IP已达到每3小时最多创建10个帐户的限制.'
        elif "DNS problem: NXDOMAIN looking up A for" in error:
            return '验证失败,没有解析域名,或解析未生效!'
        elif "Invalid response from" in error:
            return '验证失败,域名解析错误或验证URL无法被访问!'
        elif error.find('TLS Web Server Authentication') != -1:
            return "连接CA服务器失败，请稍候重试."
        elif error.find('Name does not end in a public suffix') != -1:
            return "不支持的域名%s，请检查域名是否正确!" % re.findall("Cannot issue for \"(.+)\":", error)
        elif error.find('No valid IP addresses found for') != -1:
            return "域名%s没有找到解析记录，请检查域名是否解析生效!" % re.findall("No valid IP addresses found for (.+)", error)
        elif error.find('No TXT record found at') != -1:
            return "没有在域名%s中找到有效的TXT解析记录,请检查是否正确解析TXT记录,如果是DNSAPI方式申请的,请10分钟后重试!" % re.findall("No TXT record found at (.+)", error)
        elif error.find('Incorrect TXT record') != -1:
            return "在%s上发现错误的TXT记录:%s,请检查TXT解析是否正确,如果是DNSAPI方式申请的,请10分钟后重试!" % (re.findall("found at (.+)", error), re.findall("Incorrect TXT record \"(.+)\"", error))
        elif error.find('Domain not under you or your user') != -1:
            return "这个dnspod账户下面不存在这个域名，添加解析失败!"
        elif error.find('SERVFAIL looking up TXT for') != -1:
            return "没有在域名%s中找到有效的TXT解析记录,请检查是否正确解析TXT记录,如果是DNSAPI方式申请的,请10分钟后重试!" % re.findall("looking up TXT for (.+)", error)
        elif error.find('Timeout during connect') != -1:
            return "连接超时,CA服务器无法访问您的网站!"
        elif error.find("DNS problem: SERVFAIL looking up CAA for") != -1:
            return "域名%s当前被要求验证CAA记录，请手动解析CAA记录，或1小时后重新尝试申请!" % re.findall("looking up CAA for (.+)", error)
        elif error.find("Read timed out.") != -1:
            return "验证超时,请检查域名是否正确解析，若已正确解析，可能服务器与Let'sEncrypt连接异常，请稍候再重试!"
        elif error.find("The ACME server can not issue a certificate for an IP address") != -1:
            return "不能使用IP地址申请证书!"
        elif error.find('Cannot issue for') != -1:
            return "无法为{}颁发证书，不能直接用域名后缀申请通配符证书!".format(re.findall(r'for\s+"(.+)"',error))
        elif error.find('too many failed authorizations recently') != -1:
            return '该帐户1小时内失败的订单次数超过5次，请等待1小时再重试!'
        elif error.find("Error creating new order") != -1:
            return "订单创建失败，请稍候重试!"
        elif error.find("Too Many Requests") != -1:
            return "1小时内超过5次验证失败，暂时禁止申请，请稍候重试!"
        elif error.find('HTTP Error 400: Bad Request') != -1:
            return "CA服务器拒绝访问，请稍候重试!"
        elif error.find('Temporary failure in name resolution') != -1:
            return '服务器DNS故障，无法解析域名，请使用Linux工具箱检查dns配置'
        elif error.find('Too Many Requests') != -1:
            return '该域名请求申请次数过多，请3小时后重试'
        elif error.find('Only domain names are supported') != -1:
            return "Let's Encrypt仅支持使用域名申请证书"
        elif error.find('DNSSEC: DNSKEY Missing') != -1:
            return "CA无法找到或获取用于验证DNSSEC签名的DNSKEY记录"
        else:
            return error

    # 发送验证请求
    def respond_to_challenge(self, auth):
        payload = {"keyAuthorization": "{0}".format(
            auth['acme_keyauthorization'])}
        respond_to_challenge_response = self.acme_request(
            auth['dns_challenge_url'], payload)
        return respond_to_challenge_response

    # 发送CSR
    def send_csr(self, index):
        csr = self.create_csr(index)
        payload = {"csr": self.calculate_safe_base64(csr)}
        send_csr_response = self.acme_request(
            url=self._config['orders'][index]['finalize'], payload=payload)
        if send_csr_response.status_code not in [200, 201]:
            if send_csr_response.status_code == 0:
                raise ValueError("错误：提示Connection reset by peer,可能请求过程被意外拦截，如果只有此域名无法申请，则该域名可能存在异常!")
            raise ValueError(
                "错误： 发送CSR: 响应状态{status_code} 响应值:{response}".format(
                    status_code=send_csr_response.status_code,
                    response=send_csr_response.json(),
                )
            )
        # 轮询获取证书
        l=0
        while True:
            res = self.acme_request(
                url=self._config['orders'][index]['url'], payload=""
            ).json()
            public.print_log(res)
            if res["status"] == "valid":
                certificate_url = res["certificate"]
                break
            if l >= self._max_check_num:
                raise Exception(
                    "错误：已尝试获取证书{0}次. 最大尝试次数为{1}. 尝试间隔为{2}秒.".format(
                        l,
                        self._max_check_num,
                        self._wait_time
                    )
                )
            time.sleep(1)
            l+=1
        self._config['orders'][index]['certificate_url'] = certificate_url
        self.save_config()
        return certificate_url

    # 获取证书到期时间
    def get_cert_timeout(self, cret_data):

        info = ssl_info.ssl_info().load_ssl_info_by_data(cret_data)
        if not info:
            return int(time.time() + (86400 * 90))

        try:
            return public.to_date(times=info['notAfter'])
        except:
            return int(time.time() + (86400 * 90))

    # 下载证书
    def download_cert(self, index):
        res = self.acme_request(
            self._config['orders'][index]['certificate_url'], "")
        if res.status_code not in [200, 201]:
            raise Exception("下载证书失败: {}".format(res.json()))

        pem_certificate = res.content
        if type(pem_certificate) == bytes:
            pem_certificate = pem_certificate.decode('utf-8')
        cert = self.split_ca_data(pem_certificate)
        cert['cert_timeout'] = self.get_cert_timeout(cert['cert'])
        cert['private_key'] = self._config['orders'][index]['private_key']
        cert['domains'] = self._config['orders'][index]['domains']
        del(self._config['orders'][index]['private_key'])
        del(self._config['orders'][index]['auths'])
        del(self._config['orders'][index]['expires'])
        del(self._config['orders'][index]['authorizations'])
        del(self._config['orders'][index]['finalize'])
        del(self._config['orders'][index]['identifiers'])
        if 'cert' in self._config['orders'][index]:
            del(self._config['orders'][index]['cert'])
        self._config['orders'][index]['status'] = 'valid'
        self._config['orders'][index]['cert_timeout'] = cert['cert_timeout']
        domain_name = self._config['orders'][index]['domains'][0]
        self._config['orders'][index]['save_path'] = '{}/{}'.format(
            self._save_path, domain_name)
        cert['save_path'] = self._config['orders'][index]['save_path']
        self.save_config()
        self.save_cert(cert, index)
        return cert

    # 保存证书到文件
    def save_cert(self, cert, index):
        try:
            from ssl_manage import SSLManger
            SSLManger().save_by_data(cert['cert'] + cert['root'], cert['private_key'])

            domain_name = self._config['orders'][index]['domains'][0]
            path = self._config['orders'][index]['save_path']
            if not os.path.exists(path):
                os.makedirs(path, 384)

            # 存储证书
            key_file = path + "/privkey.pem"
            pem_file = path + "/fullchain.pem"
            public.writeFile(key_file, cert['private_key'])
            public.writeFile(pem_file, cert['cert'] + cert['root'])
            public.writeFile(path + "/cert.csr", cert['cert'])
            public.writeFile(path + "/root_cert.csr", cert['root'])

            self.set_exclude_hash(index, cert['cert'] + cert['root'])

            # 转为IIS证书
            try:
                pfx_buffer = self.dump_pkcs12(
                    cert['private_key'], cert['cert'] + cert['root'], cert['root'], domain_name)
            except:
                pfx_buffer = ssl_info.ssl_info().dump_pkcs12_new(
                    cert['private_key'], cert['cert'] + cert['root'], cert['root'], domain_name)
            public.writeFile(path + "/fullchain.pfx", pfx_buffer, 'wb+')

            ps = '''文件说明：
privkey.pem     证书私钥
fullchain.pem   包含证书链的PEM格式证书(nginx/apache)
root_cert.csr   根证书
cert.csr        域名证书
fullchain.pfx   用于IIS的证书格式

如何在宝塔面板使用：
privkey.pem         粘贴到密钥输入框
fullchain.pem       粘贴到证书输入框
'''
            public.writeFile(path+'/说明.txt', ps)
            self.sub_all_cert(key_file, pem_file)
        except:
            write_log(public.get_error_info())

    def set_exclude_hash(self, order, exclude_hash):
        try:
            path = '{}/data/exclude_hash.json'.format(public.get_panel_path())

            try:
                data = json.loads(public.readFile(path))
            except:
                data = self.get_exclude_hash(public.to_dict_obj({}))
            if "exclude_hash_let" not in data:
                data["exclude_hash_let"] = {}
            data['exclude_hash_let'].update({order: self._hash(certificate=exclude_hash)})
            public.writeFile(path, json.dumps(data))
        except:
            pass

    def get_exclude_hash(self, get):
        path = '{}/data/exclude_hash.json'.format(public.get_panel_path())

        import panelSSL
        exclude_data = panelSSL.panelSSL().get_exclude_hash(get)
        if exclude_data.get('version_let') == '1':
            return exclude_data
        if "exclude_hash_let" not in exclude_data:
            exclude_data["exclude_hash_let"] = {}

        data = self.read_config()

        if data.get('orders', {}):
            try:
                self.get_apis()
            except:
                return exclude_data
            for order in data.get('orders', {}).values():
                if order['status'] != "valid" or not order.get("certificate_url"):
                    continue
                try:
                    res = self.acme_request(
                        order['certificate_url'], "")
                    if res.status_code not in [200, 201]:
                        continue
                    pem_certificate = res.content
                    if type(pem_certificate) == bytes:
                        pem_certificate = pem_certificate.decode('utf-8')
                    cert = self.split_ca_data(pem_certificate)
                    exclude_data["exclude_hash_let"].update({order['index']: self._hash(certificate=cert['cert'] + cert['root'])})
                except:
                    pass
        exclude_data['version_let'] = '1'
        public.writeFile(path, json.dumps(exclude_data))
        return exclude_data

    def _hash(self, cert_filename: str = None, certificate: str = None, ignore_errors: bool = False):
        if cert_filename is not None and os.path.isfile(cert_filename):
            certificate = public.readFile(cert_filename)

        if not isinstance(certificate, str) or not certificate.startswith("-----BEGIN"):
            if ignore_errors:
                return None
            raise ValueError("证书格式错误")

        md5_obj = hashlib.md5()
        md5_obj.update(certificate.encode("utf-8"))
        return md5_obj.hexdigest()


    # 通过域名获取网站名称
    def get_site_name_by_domains(self,domains):
        sql = public.M('domain')
        site_sql = public.M('sites')
        siteName, project_type = None, None
        for domain in domains:
            pid = sql.where('name=?',domain).getField('pid')
            if pid:
                site_data = site_sql.where('id=?', pid).field('name,project_type').find()
                siteName, project_type = site_data["name"], site_data["project_type"]
                break
        return siteName, project_type

    # 替换服务器上的同域名同品牌证书
    def sub_all_cert(self, key_file, pem_file):
        cert_init = self.get_cert_init(pem_file)  # 获取新证书的基本信息
        paths = ['/www/server/panel/vhost/cert', '/www/server/panel/vhost/ssl','/www/server/panel']
        is_panel = False
        for path in paths:
            if not os.path.exists(path):
                continue
            for p_name in os.listdir(path):
                to_path = path + '/' + p_name
                to_pem_file = to_path + '/fullchain.pem'
                to_key_file = to_path + '/privkey.pem'
                to_info = to_path + '/info.json'
                # 判断目标证书是否存在
                if not os.path.exists(to_pem_file):
                    if not p_name in ['ssl']: continue
                    to_pem_file = to_path + '/certificate.pem'
                    to_key_file = to_path + '/privateKey.pem'
                    if not os.path.exists(to_pem_file):
                        continue
                # 获取目标证书的基本信息
                to_cert_init = self.get_cert_init(to_pem_file)
                # 判断证书品牌是否一致
                try:
                    if to_cert_init['issuer'] != cert_init['issuer'] and to_cert_init['issuer'].find("Let's Encrypt") == -1 and to_cert_init.get('issuer_O', '') != "Let's Encrypt":
                        continue
                except: continue
                # 判断目标证书的到期时间是否较早
                if to_cert_init['notAfter'] > cert_init['notAfter']:
                    continue
                # 判断认识名称是否一致
                if len(to_cert_init['dns']) != len(cert_init['dns']):
                    continue
                is_copy = True
                for domain in to_cert_init['dns']:
                    if not domain in cert_init['dns']:
                        is_copy = False
                if not is_copy:
                    continue

                # 替换新的证书文件和基本信息
                public.writeFile(
                    to_pem_file, public.readFile(pem_file, 'rb'), 'wb')
                public.writeFile(
                    to_key_file, public.readFile(key_file, 'rb'), 'wb')
                public.writeFile(to_info, json.dumps(cert_init))
                write_log(
                    "|-检测到{}下的证书与本次申请的证书重叠，且到期时间较早，已替换为新证书!".format(to_path))
                if path == paths[-1]: is_panel = True

        # 重载web服务
        public.serviceReload()
        # if is_panel: public.restart_panel()

    # 检查指定证书是否在订单列表
    def check_order_exists(self, pem_file):
        try:
            cert_init = self.get_cert_init(pem_file)
            if not cert_init:
                return None
            if not (cert_init['issuer'].find("Let's Encrypt") != -1 or cert_init['issuer'] in ('R3', 'R10', 'R11') or cert_init.get('issuer_O', '') == "Let's Encrypt"):
                return None
            for index in self._config['orders'].keys():
                if not 'save_path' in self._config['orders'][index]:
                    continue
                for domain in self._config['orders'][index]['domains']:
                    if domain in cert_init['dns']:
                        return index
            return pem_file
        except:
            return None

    # 取证书基本信息API
    def get_cert_init_api(self, args):
        if not os.path.exists(args.pem_file):
            args.pem_file = 'vhost/cert/{}/fullchain.pem'.format(args.siteName)
            if not os.path.exists(args.pem_file):
                return public.returnMsg(False, '指定证书文件不存在!')
        cert_init = self.get_cert_init(args.pem_file)
        if not cert_init:
            return public.returnMsg(False, '证书信息获取失败!')
        try:
            cert_init['dnsapi'] = json.loads(public.readFile(self._dnsapi_file))
        except:
            cert_init['dnsapi'] = []
        return cert_init

    # 获取指定证书基本信息
    def get_cert_init(self, pem_file):
        return ssl_info.ssl_info().load_ssl_info(pem_file)

    # 转换时间
    def strf_date(self, sdate):
        return time.strftime('%Y-%m-%d', time.strptime(sdate, '%Y%m%d%H%M%S'))

    # 证书转为DER
    def dump_der(self, cert_path):
        cert = OpenSSL.crypto.load_certificate(
            OpenSSL.crypto.FILETYPE_PEM, public.readFile(cert_path+'/cert.csr'))
        return OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_ASN1, cert)

    # 证书转为pkcs12
    def dump_pkcs12(self, key_pem=None, cert_pem=None, ca_pem=None, friendly_name=None):
        p12 = OpenSSL.crypto.PKCS12()
        if cert_pem:
            p12.set_certificate(OpenSSL.crypto.load_certificate(
                OpenSSL.crypto.FILETYPE_PEM, cert_pem.encode()))
        if key_pem:
            p12.set_privatekey(OpenSSL.crypto.load_privatekey(
                OpenSSL.crypto.FILETYPE_PEM, key_pem.encode()))
        if ca_pem:
            p12.set_ca_certificates((OpenSSL.crypto.load_certificate(
                OpenSSL.crypto.FILETYPE_PEM, ca_pem.encode()),))
        if friendly_name:
            p12.set_friendlyname(friendly_name.encode())
        return p12.export()

    # 拆分根证书
    def split_ca_data(self,cert):
        sp_key = '-----END CERTIFICATE-----\n'
        datas = cert.split(sp_key)
        return {"cert": datas[0] + sp_key, "root": sp_key.join(datas[1:])}

    # 构造可选名称
    def get_alt_names(self, index):
        domain_name = self._config['orders'][index]['domains'][0]
        domain_alt_names = []
        if len(self._config['orders'][index]['domains']) > 1:
            domain_alt_names = self._config['orders'][index]['domains'][1:]
        return domain_name, domain_alt_names

    # 检查DNS记录
    def check_dns(self, domain, value, s_type='TXT'):
        write_log("|-尝试本地验证DNS记录,域名: {} , 类型: {} 记录值: {}".format(domain, s_type, value))
        time.sleep(10)
        n = 0
        while n < 20:
            n += 1
            try:
                import dns.resolver
                ns = dns.resolver.query(domain, s_type)
                for j in ns.response.answer:
                    for i in j.items:
                        txt_value = i.to_text().replace('"', '').strip()
                        write_log("|-第 {} 次验证值: {}".format(n, txt_value))
                        if txt_value == value:
                            write_log("|-本地验证成功!")
                            return True
            except:
                try:
                    import dns.resolver
                except:
                    return False
            time.sleep(3)
        write_log("|-本地验证失败!")
        return True

    # 创建CSR
    def create_csr(self, index):
        if 'csr' in self._config['orders'][index]:
            return self._config['orders']['csr']
        domain_name, domain_alt_names = self.get_alt_names(index)
        X509Req = OpenSSL.crypto.X509Req()
        if not public.check_ip(domain_name):
            X509Req.get_subject().CN = domain_name
            if domain_alt_names:
                SAN = "DNS:{0}, ".format(domain_name).encode("utf8") + ", ".join(
                    "DNS:" + i for i in domain_alt_names
                ).encode("utf8")
            else:
                SAN = "DNS:{0}".format(domain_name).encode("utf8")
        else:
            if domain_alt_names:
                SAN = "IP:{0}, ".format(domain_name).encode("utf8") + ", ".join(
                    "IP:" + i for i in domain_alt_names
                ).encode("utf8")
            else:
                SAN = "IP:{0}".format(domain_name).encode("utf8")

        X509Req.add_extensions(
            [
                OpenSSL.crypto.X509Extension(
                    "subjectAltName".encode("utf8"), critical=False, value=SAN
                )
            ]
        )
        pk = OpenSSL.crypto.load_privatekey(
            OpenSSL.crypto.FILETYPE_PEM, self.create_certificate_key(
                index).encode()
        )
        X509Req.set_pubkey(pk)
        try:
            X509Req.set_version(2)
        except ValueError as e:  # pyOpenSSL 新版本需要必须设置版本为0
            X509Req.set_version(0)
        X509Req.sign(pk, self._digest)
        return OpenSSL.crypto.dump_certificate_request(OpenSSL.crypto.FILETYPE_ASN1, X509Req)

    def create_csr_new(self, index):
        from cryptography import x509
        from cryptography.x509.oid import NameOID
        from cryptography.hazmat.primitives.asymmetric import rsa
        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives import serialization
        import ipaddress
        # 如果已经生成了 CSR，直接返回
        if 'csr' in self._config['orders'][index]:
            return self._config['orders'][index]['csr']

        # 获取域名和备用域名
        domain_name, domain_alt_names = self.get_alt_names(index)

        # 生成私钥
        pk = self.create_certificate_key(index).encode()
        private_key = serialization.load_pem_private_key(pk, password=None)

        if public.check_ip(domain_name):
            # IP证书不需要CN
            csr_builder = x509.CertificateSigningRequestBuilder().subject_name(
                x509.Name([])
            )
            # 添加 subjectAltName 扩展
            if domain_alt_names:
                alt_names = [x509.IPAddress(ipaddress.ip_address(domain_name))] + [x509.IPAddress(ipaddress.ip_address(alt)) for alt in domain_alt_names]
            else:
                alt_names = [x509.IPAddress(ipaddress.ip_address(domain_name))]
        else:
            # 构建 CSR
            csr_builder = x509.CertificateSigningRequestBuilder().subject_name(
                x509.Name([
                    x509.NameAttribute(NameOID.COMMON_NAME, domain_name),
                ])
            )

            # 添加 subjectAltName 扩展
            if domain_alt_names:
                alt_names = [x509.DNSName(domain_name)] + [x509.DNSName(alt) for alt in domain_alt_names]
            else:
                alt_names = [x509.DNSName(domain_name)]

        csr_builder = csr_builder.add_extension(
            x509.SubjectAlternativeName(alt_names),
            critical=False
        )

        # 签署 CSR
        csr = csr_builder.sign(private_key, hashes.SHA256())

        # 返回 CSR (ASN1 格式)
        return csr.public_bytes(serialization.Encoding.DER)

    # 构造域名验证头和验证值
    def get_keyauthorization(self, token):
        acme_header_jwk_json = json.dumps(
            self.get_acme_header("GET_THUMBPRINT")["jwk"], sort_keys=True, separators=(",", ":")
        )
        acme_thumbprint = self.calculate_safe_base64(
            hashlib.sha256(acme_header_jwk_json.encode("utf8")).digest()
        )
        acme_keyauthorization = "{0}.{1}".format(token, acme_thumbprint)
        base64_of_acme_keyauthorization = self.calculate_safe_base64(
            hashlib.sha256(acme_keyauthorization.encode("utf8")).digest()
        )

        return acme_keyauthorization, base64_of_acme_keyauthorization

    # 构造验证信息
    def get_identifier_auth(self, index, url, auth_info):
        s_type = self.get_auth_type(index)
        write_log("|-验证类型：{}".format(s_type))
        domain = auth_info['identifier']['value']
        wildcard = False
        # 处理通配符
        if 'wildcard' in auth_info:
            wildcard = auth_info['wildcard']
        if wildcard:
            domain = "*." + domain

        for auth in auth_info['challenges']:
            if auth['type'] != s_type:
                continue
            identifier_auth = {
                "domain": domain,
                "url": url,
                "wildcard": wildcard,
                "token": auth['token'],
                "dns_challenge_url": auth['url'],
            }
            return identifier_auth
        return None

    # 获取域名验证方式
    def get_auth_type(self, index):
        if not index in self._config['orders']:
            raise Exception('指定订单不存在!')
        s_type = 'http-01'
        if 'auth_type' in self._config['orders'][index]:
            if self._config['orders'][index]['auth_type'] == 'dns':
                s_type = 'dns-01'
            elif self._config['orders'][index]['auth_type'] == 'tls':
                s_type = 'tls-alpn-01'
            else:
                s_type = 'http-01'
        return s_type

    # 保存订单
    def save_order(self, order_object, index):
        if not 'orders' in self._config:
            self._config['orders'] = {}
        renew = False
        if not index:
            # index = public.md5(json.dumps(order_object['identifiers']))
            index = public.md5(str(uuid.uuid4()))
        else:
            renew = True
            order_object['certificate_url'] = self._config['orders'][index]['certificate_url']
            order_object['save_path'] = self._config['orders'][index]['save_path']

        order_object['expires'] = self.utc_to_time(order_object['expires'])
        self._config['orders'][index] = order_object
        self._config['orders'][index]['index'] = index
        if not renew:
            self._config['orders'][index]['create_time'] = int(time.time())
            self._config['orders'][index]['renew_time'] = 0
        self.save_config()
        return index

    # UTC时间转时间戳
    def utc_to_time(self, utc_string):
        try:
            utc_string = utc_string.split('.')[0]
            utc_date = datetime.datetime.strptime(
                utc_string, "%Y-%m-%dT%H:%M:%S")
            # 按北京时间返回
            return int(time.mktime(utc_date.timetuple())) + (3600 * 8)
        except:
            return int(time.time() + 86400 * 7)

    # 获取kid
    def get_kid(self, force=False):
        #如果配置文件中不存在kid或force = True时则重新注册新的acme帐户
        if not 'account' in self._config:
            self._config['account'] = {}
        k = self._mod_index[self._debug]
        if not k in self._config['account']:
            self._config['account'][k] = {}

        if not 'kid' in self._config['account'][k]:
            self._config['account'][k]['kid'] = self.register()
            self.save_config()
            time.sleep(3)
            self._config = self.read_config()
        return self._config['account'][k]['kid']

    # 注册acme帐户
    def register(self, existing=False):
        if not 'email' in self._config:
            self._config['email'] = 'demo@bt.cn'
        if existing:
            payload = {"onlyReturnExisting": True}
        elif self._config['email']:
            payload = {
                "termsOfServiceAgreed": True,
                "contact": ["mailto:{0}".format(self._config['email'])],
            }
        else:
            payload = {"termsOfServiceAgreed": True}
        if self._debug == 'litessl':
            payload["externalAccountBinding"] = self.build_eab(
                account_jwk_pem=self.get_account_key(),
                new_account_url=self._apis['newAccount']
            )
            public.print_log(payload["externalAccountBinding"])

        res = self.acme_request(url=self._apis['newAccount'], payload=payload)

        if res.status_code not in [201, 200, 409]:
            raise Exception("注册ACME帐户失败: {}".format(res.json()))
        kid = res.headers["Location"]
        return kid

    def build_eab(self, account_jwk_pem: str, new_account_url: str):
        """
        构造 External Account Binding (EAB)
        account_jwk_pem: RSA 私钥的 PEM 字符串
        """
        import base64
        import json
        import hmac
        import hashlib
        from cryptography.hazmat.primitives import serialization

        # Base64URL 编码
        def b64url(data: bytes) -> str:
            return base64.urlsafe_b64encode(data).rstrip(b"=").decode("utf-8")

        # 将整数转成 Base64URL（big-endian）
        def b64url_uint(val: int) -> str:
            length = (val.bit_length() + 7) // 8
            return base64.urlsafe_b64encode(val.to_bytes(length, "big")).rstrip(b"=").decode("utf-8")

        # PEM 字符串 → RSA 私钥对象
        private_key = serialization.load_pem_private_key(
            account_jwk_pem.encode("utf-8"),
            password=None
        )

        # RSA 公钥 → JWK dict
        public_numbers = private_key.public_key().public_numbers()
        public_jwk = {
            "kty": "RSA",
            "e": b64url_uint(public_numbers.e),
            "n": b64url_uint(public_numbers.n)
        }

        # 1. 请求 EAB 参数
        eab_res = requests.post("https://www.bt.cn/api/v3/litessl/eab")
        eab_data = eab_res.json()
        self._config["eab_kid"] = eab_data["res"]["data"]["eab_kid"]
        self._config["eab_hmac_key"] = eab_data["res"]["data"]["eab_mac_key"]

        # 2. protected header
        protected = {
            "alg": "HS256",
            "kid": self._config["eab_kid"],
            "url": new_account_url
        }
        protected_b64 = b64url(json.dumps(protected, separators=(",", ":")).encode())

        # 3. payload = 公钥 JWK
        payload_b64 = b64url(json.dumps(public_jwk, separators=(",", ":")).encode())

        # 4. HMAC-SHA256 签名
        def base64url_decode(s: str) -> bytes:
            # 补全 '=' 使长度能被4整除
            s = s + "=" * (-len(s) % 4)
            return base64.urlsafe_b64decode(s)

        hmac_key = base64url_decode(self._config["eab_hmac_key"])
        signing_input = f"{protected_b64}.{payload_b64}".encode()
        signature = hmac.new(hmac_key, signing_input, hashlib.sha256).digest()
        signature_b64 = b64url(signature)

        # 5. 返回 EAB
        return {
            "protected": protected_b64,
            "payload": payload_b64,
            "signature": signature_b64
        }

    # 请求到ACME接口
    def acme_request(self, url, payload):
        headers = {"User-Agent": self._user_agent}
        payload = self.stringfy_items(payload)

        if payload == "":
            payload64 = payload
        else:
            payload64 = self.calculate_safe_base64(json.dumps(payload))
        protected = self.get_acme_header(url)
        protected64 = self.calculate_safe_base64(json.dumps(protected))
        signature = self.sign_message(
            message="{0}.{1}".format(protected64, payload64))  # bytes
        signature64 = self.calculate_safe_base64(signature)  # str
        data = json.dumps(
            {"protected": protected64, "payload": payload64,
                "signature": signature64}
        )
        headers.update({"Content-Type": "application/jose+json"})
        response = requests.post(
            url, data=data.encode("utf8"), timeout=self._acme_timeout, headers=headers, verify=self._verify,s_type=self._request_type
        )
        # 更新随机数
        self.update_replay_nonce(response)
        return response

    # 计算signature
    def sign_message(self, message):
        try:
            pk = OpenSSL.crypto.load_privatekey(
                OpenSSL.crypto.FILETYPE_PEM, self.get_account_key().encode())
            return OpenSSL.crypto.sign(pk, message.encode("utf8"), self._digest)
        except:
            return self.sign_message_new(message)

    def sign_message_new(self, message):
        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives.asymmetric import padding

        import ssl_info
        pk = ssl_info.ssl_info().analysis_private_key(self.get_account_key())
        return pk.sign(message.encode("utf8"), padding.PKCS1v15(), hashes.SHA256())

    # 系列化payload
    def stringfy_items(self, payload):
        if isinstance(payload, str):
            return payload

        for k, v in payload.items():
            if isinstance(k, bytes):
                k = k.decode("utf-8")
            if isinstance(v, bytes):
                v = v.decode("utf-8")
            payload[k] = v
        return payload

    # 获取随机数
    def get_nonce(self, force=False):
        # 如果没有保存上一次的随机数或force=True时则重新获取新的随机数
        if not self._replay_nonce or force:
            headers = {"User-Agent": self._user_agent}
            response = requests.get(
                self._apis['newNonce'],
                timeout=self._acme_timeout,
                headers=headers,
                verify=self._verify,
                s_type=self._request_type
            )
            self._replay_nonce = response.headers["Replay-Nonce"]
        return self._replay_nonce

    # 获请ACME请求头
    def get_acme_header(self, url):
        header = {"alg": "RS256", "nonce": self.get_nonce(), "url": url}
        if url in [self._apis['newAccount'], 'GET_THUMBPRINT']:
            from cryptography.hazmat.backends import default_backend
            from cryptography.hazmat.primitives import serialization
            private_key = serialization.load_pem_private_key(
                self.get_account_key().encode(),
                password=None,
                backend=default_backend(),
            )
            public_key_public_numbers = private_key.public_key().public_numbers()

            exponent = "{0:x}".format(public_key_public_numbers.e)
            exponent = "0{0}".format(exponent) if len(
                exponent) % 2 else exponent
            modulus = "{0:x}".format(public_key_public_numbers.n)
            jwk = {
                "kty": "RSA",
                "e": self.calculate_safe_base64(binascii.unhexlify(exponent)),
                "n": self.calculate_safe_base64(binascii.unhexlify(modulus)),
            }
            header["jwk"] = jwk
        else:
            header["kid"] = self.get_kid()
        return header

    # 转为无填充的Base64
    def calculate_safe_base64(self, un_encoded_data):
        if sys.version_info[0] == 3:
            if isinstance(un_encoded_data, str):
                un_encoded_data = un_encoded_data.encode("utf8")
        r = base64.urlsafe_b64encode(un_encoded_data).rstrip(b"=")
        return r.decode("utf8")

    # 获用户取密钥对
    def get_account_key(self):
        if not 'account' in self._config:
            self._config['account'] = {}
        k = self._mod_index[self._debug]
        if not k in self._config['account']:
            self._config['account'][k] = {}

        if not 'key' in self._config['account'][k]:
            try:
                self._config['account'][k]['key'] = self.create_key()
            except:
                self._config['account'][k]['key'] = self.create_key_new()
            if type(self._config['account'][k]['key']) == bytes:
                self._config['account'][k]['key'] = self._config['account'][k]['key'].decode()
            self.save_config()
        return self._config['account'][k]['key']

    # 获取证书密钥对
    def create_certificate_key(self, index):
        # 判断是否已经创建private_key
        if 'private_key' in self._config['orders'][index]:
            return self._config['orders'][index]['private_key']
        # 创建新的私钥
        try:
            private_key = self.create_key()
        except:
            private_key = self.create_key_new()
        if type(private_key) == bytes:
            private_key = private_key.decode()
        # 保存私钥到订单配置文件
        self._config['orders'][index]['private_key'] = private_key
        self.save_config()
        return private_key

    # 创建Key
    def create_key(self, key_type=None):
        if key_type not in [OpenSSL.crypto.TYPE_RSA, OpenSSL.crypto.TYPE_DSA, OpenSSL.crypto.TYPE_EC]:
            key_type = OpenSSL.crypto.TYPE_RSA
        key = OpenSSL.crypto.PKey()
        key.generate_key(key_type, self._bits)
        private_key = OpenSSL.crypto.dump_privatekey(
            OpenSSL.crypto.FILETYPE_PEM, key)
        return private_key

    def create_key_new(self, key_type='RSA'):
        from cryptography.hazmat.primitives import serialization
        from cryptography.hazmat.primitives.asymmetric import rsa, ec, ed25519

        if key_type == 'RSA':
            private_key = rsa.generate_private_key(
                public_exponent=65537,
                key_size=self._bits
            )
        elif key_type == 'EC':
            private_key = ec.generate_private_key(ec.SECP256R1())
        elif key_type == 'ED25519':
            private_key = ed25519.Ed25519PrivateKey.generate()
        else:
            raise ValueError(f"Unsupported key type: {key_type}")

        private_key_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        )
        return private_key_pem

    # 写配置文件
    def save_config(self):
        fp = open(self._conf_file_v2, 'w+')
        fcntl.flock(fp, fcntl.LOCK_EX)  # 加锁
        fp.write(json.dumps(self._config))
        fcntl.flock(fp, fcntl.LOCK_UN)  # 解锁
        fp.close()
        return True

    # 读配置文件
    def read_config(self):
        if not os.path.exists(self._conf_file_v2):
            self._config['orders'] = {}
            self._config['account'] = {}
            self._config['apis'] = {}
            self._config['email'] = public.M('config').where('id=?',(1,)).getField('email')
            if self._config['email'] in [public.en_hexb('4d6a67334f5459794e545932514846784c6d4e7662513d3d')]:
                self._config['email'] = None
            self.save_config()
            return self._config
        tmp_config = public.readFile(self._conf_file_v2)
        if not tmp_config:
            return self._config
        try:
            self._config = json.loads(tmp_config)
        except:
            self.save_config()
            return self._config
        return self._config

    def check_config(self, index, key, value) -> bool:
        tmp_config = public.readFile(self._conf_file_v2)
        if not tmp_config:
            return False
        try:
            config_data = json.loads(tmp_config)
        except:
            return False

        if index not in config_data.get("orders", {}):
            return False

        if key not in config_data['orders'][index]:
            return False
        if config_data['orders'][index][key] == value:
            return True
        else:
            return False

    # 申请证书
    def apply_cert(self, domains, auth_type='dns', auth_to='Dns_com|None|None', **args):
        index = ''
        write_log("", "wb+")
        for i in range(len(domains)):
            domains[i] = domains[i].replace("[", "").replace("]", "")
        try:
            if 'ca' in args:
                public.set_module_logs(args['ca'], "apply_cert_ca", 1)
                self._debug = args['ca']
                self._url = url_dic.get(self._debug, url_dic['letsencrypt'])
            self.get_apis()
            index = None
            if 'index' in args:
                index = args['index']
            if not index:  # 判断是否只想验证域名
                write_log("|-正在创建订单..")
                index = self.create_order(domains, auth_type, auth_to)
                write_log("|-正在获取验证信息..")
                self.get_auths(index)
                if auth_to == 'dns' and len(self._config['orders'][index]['auths']) > 0:
                    auth_domains = [i["domain"].replace("*.", "") for i in self._config['orders'][index]['auths']]
                    if len(auth_domains) != len(set(auth_domains)):
                        self._config['orders'][index]["error"] = True
                        self._config['orders'][index]["error_msg"] = "检测到解析记录存在冲突，请分别验证以下域名"
                    return self._config['orders'][index]
            else:
                if "ca" in self._config['orders'][index]:
                    self._debug = self._config['orders'][index]['ca']
                    self._url = url_dic.get(self._debug, url_dic['letsencrypt'])
                    self.get_apis()
            write_log("|-正在验证域名..")
            self.auth_domain(index)
            self.remove_dns_record()
            write_log("|-正在发送CSR..")
            self.send_csr(index)
            write_log("|-正在下载证书..")
            cert = self.download_cert(index)
            if domains and public.check_ip(domains[0]):
                public.set_module_logs("apply_ipssl_success", "apply_ipssl_success", 1)
            cert['status'] = True
            cert['msg'] = '申请成功!'
            write_log("|-申请成功，正在部署到站点..")
            return cert
        except Exception as ex:
            self.remove_dns_record()
            ex = str(ex)
            if ex.find(">>>>") != -1:
                msg = ex.split(">>>>")
                msg[1] = json.loads(msg[1])
            else:
                msg = ex
                write_log(public.get_error_info())
            _res = {"status": False, "msg": msg, "index": index}
            return _res

    # 申请证书 - api
    def apply_cert_api(self, args):
        """
        @name 申请证书
        @param domains: list 域名列表
        @param auth_type: str 认证方式
        @param auth_to: str 认证路径
        @param auto_wildcard: str 是否自动组合泛域名
        """
        if not 'id' in args:
            return public.returnMsg(False,'网站id不能为空!')

        if 'auto_wildcard' in args and args.auto_wildcard == '1':
            self._auto_wildcard = True

        find = public.M('sites').where('id=?', (args.id,)).find()
        if not find:
            return public.returnMsg(False, "网站丢失，无法继续申请证书")

        if args.auth_type in ['http', 'tls']:
            if find["status"] != "1":
                return public.returnMsg(False, "当前网站未开启不能使用文件验证")
            if not self.can_use_base_file_check(find["name"], find["project_type"]):
                webserver: str = public.get_webserver()
                msg = "当前项目的服务（{}）配置文件被修改不支持文件验证，请选择其他方式或还原配置文件".format(webserver.title())
                if webserver != 'nginx':
                    return public.returnMsg(False, msg)
                # nginx 检测其他两种方案的可行性
                if not self.can_use_lua_for_site(find["name"], find["project_type"]) and \
                        not self.can_use_if_for_file_check(find["name"], find["project_type"]):

                    return public.returnMsg(False, msg)

        # 是否为指定站点
        count = public.M('sites').where(
            'id=? and project_type in (?,?,?,?)',
            (args.id, 'Java', 'Go', 'Other', "Python")
        ).count()
        if count:
            try:
                project_info = json.loads(find['project_config'])
                if 'ssl_path' not in project_info:
                    ssl_path = '/www/wwwroot/java_node_ssl'
                else:
                    ssl_path = project_info['ssl_path']
                if not os.path.exists(ssl_path):
                    os.makedirs(ssl_path)

                args.auth_to = ssl_path
            except:
                return public.returnMsg(False, '当前项目配置文件存在问题,请重新建立')
        else:
            if re.match(r"^\d+$", args.auth_to):
                import panelSite
                args.auth_to = find['path'] + '/' + panelSite.panelSite().GetRunPath(args)
                args.auth_to = args.auth_to.replace("//", "/")
                if args.auth_to[-1] == '/':
                    args.auth_to = args.auth_to[:-1]

                if not os.path.exists(args.auth_to):
                    return public.returnMsg(False, '无效的站点目录，请检查指定站点是否存在!')

            #检查认证环境
            if args.auth_type in ['http', 'tls']:
                check_result = self.check_auth_env(args)
                if check_result:
                    return check_result

        if "ca" in args and args.ca:
            ca = args.ca
        else:
            ca = False

        return self.apply_cert(json.loads(args.domains), args.auth_type, args.auth_to,ca=ca)

    # 检查认证环境
    def check_auth_env(self,args):
        for domain in json.loads(args.domains):
            if public.checkIp(domain): continue
            if domain.find('*.') != -1 and args.auth_type in ['http','tls']:
                return public.returnMsg(False, '泛域名不能使用【文件验证】的方式申请证书!')

        data = public.M('sites').where('id=?', (args.id,)).find()
        if not data:
            return public.returnMsg(False, "网站丢失，无法继续申请证书")
        else:
            args.siteName = data['name']
            site_type = data["project_type"]

        use_nginx_conf_to_auth = False
        if args.auth_type in ['http', 'tls'] and public.get_webserver() == "nginx":  # nginx 在lua验证和可重启的
            if self.can_use_lua_for_site(args.siteName, site_type):
                use_nginx_conf_to_auth = True
            else:
                if self.can_use_if_for_file_check(args.siteName, site_type):
                    use_nginx_conf_to_auth = True

        import panelSite
        s = panelSite.panelSite()
        if args.auth_type in ['http', 'tls'] and use_nginx_conf_to_auth is False:
            try:
                args.sitename = args.siteName
                data = s.GetRedirectList(args)
                # 检查重定向是否开启
                if type(data) == list:
                    for x in data:
                        if x['type']: return public.returnMsg(False, 'SITE_SSL_ERR_301')
                data = s.GetProxyList(args)
                # 检查反向代理是否开启
                if type(data) == list:
                    for x in data:
                        if x['type']: return public.returnMsg(False,'已开启反向代理的站点无法申请SSL!')
                # 检查旧重定向是否开启
                data = s.Get301Status(args)
                if data['status']:
                    return public.returnMsg(False,'网站已经开启重定向，请关闭后再申请!')
                #判断是否强制HTTPS
                if s.IsToHttps(args.siteName):
                    return public.returnMsg(False, '配置强制HTTPS后无法使用【文件验证】的方式申请证书!')
            except:
                return False
        else:
            if args.auth_to.find('Dns_com') != -1:
                if not os.path.exists('plugin/dns/dns_main.py'):
                    return public.returnMsg(False, '请先到软件商店安装【云解析】，并完成域名NS绑定.')

        return False

    # DNS手动验证
    def apply_dns_auth(self, args):
        if not hasattr(args, "index") or not args.index:
            return public.returnMsg(False, "参数信息不完整，没有索引参数【index】")
        return self.apply_cert([], auth_type='dns', auth_to='dns', index=args.index)


    #创建计划任务
    def set_crond(self):
        try:
            echo = public.md5(public.md5('renew_lets_ssl_bt'))
            find = public.M('crontab').where('echo=?',(echo,)).find()
            cron_id = find['id'] if find else None

            import crontab
            import random
            args_obj = public.dict_obj()
            if not cron_id:
                # 计划任务不存在且未关闭自动续签则创建任务
                if os.path.exists("/www/server/panel/data/close_ssl_auto_renewal.pl"):
                    return
                cronPath = public.GetConfigValue('setup_path') + '/cron/' + echo
                shell = '{} -u /www/server/panel/class/acme_v2.py --renew_v2=1'.format(sys.executable)
                public.writeFile(cronPath,shell)

                # 使用随机时间
                hour = random.randint(0, 23)
                minute = random.randint(1, 59)
                args_obj.id = public.M('crontab').add('name,type,where1,where_hour,where_minute,echo,addtime,status,save,backupTo,sType,sName,sBody,urladdress',("续签Let's Encrypt证书",'day','',hour,minute,echo,time.strftime('%Y-%m-%d %X',time.localtime()),0,'','localhost','toShell','',shell,''))
                crontab.crontab().set_cron_status(args_obj)
            else:
                # 计划任务存在则检查是否关闭自动续签，关闭则删除任务
                if os.path.exists("/www/server/panel/data/close_ssl_auto_renewal.pl"):
                    # 删除任务
                    args_obj.id = cron_id
                    crontab.crontab().DelCrontab(args_obj)
                    return
                # 检查任务如果是0点10分执行，改为随机时间
                if find['where_hour'] == 0 and find['where_minute'] == 10:
                    print('修改任务时间')
                    # 使用随机时间
                    hour = random.randint(0, 23)
                    minute = random.randint(1, 59)
                    public.M('crontab').where('id=?',(cron_id,)).save('where_hour,where_minute,status',(hour,minute,0))

                    # 停用任务
                    args_obj.id = cron_id
                    crontab.crontab().set_cron_status(args_obj)

                    # 启用任务
                    public.M('crontab').where('id=?',(cron_id,)).setField('status',1)
                    crontab.crontab().set_cron_status(args_obj)

                cron_path = public.get_cron_path()
                if os.path.exists(cron_path):
                    cron_s = public.readFile(cron_path)
                    if cron_s.find(echo) == -1:
                        public.M('crontab').where('id=?',(cron_id,)).setField('status',0)
                        args_obj.id = cron_id
                        crontab.crontab().set_cron_status(args_obj)
        except:pass


    # 获取当前正在使用此证书的网站目录
    def get_ssl_used_site(self, index):
        hash_dic = self.get_exclude_hash(public.dict_obj())
        if not hash_dic: return False
        ssl_hash = hash_dic.get("exclude_hash_let", {}).get(index, '')
        if not ssl_hash: return False

        cert_paths = 'vhost/cert'
        import panelSite
        args = public.dict_obj()
        args.siteName = ''
        for c_name in os.listdir(cert_paths):
            skey_file = '{}/{}/fullchain.pem'.format(cert_paths,c_name)
            try:
                skey = self._hash(cert_filename=skey_file)
            except:
                continue
            if not skey: continue
            if skey == ssl_hash:
                args.siteName = c_name
                site_info = public.M('sites').where('name=?', c_name).find()
                if not site_info or isinstance(site_info, str):
                    return False
                if site_info["project_type"] not in ("PHP", "proxy"):
                    if not os.path.isdir(site_info["path"]):
                        return os.path.dirname(site_info["path"])
                    else:
                        return site_info["path"]

                run_path = panelSite.panelSite().GetRunPath(args)
                if not run_path:
                    continue
                sitePath = public.M('sites').where('name=?',c_name).getField('path')
                if not sitePath:
                    continue
                to_path = "{}/{}".format(sitePath,run_path)
                to_path = to_path.replace("//", "/")
                return to_path
        return False

    def get_index(self,domains):
        '''
            @name 获取标识
            @author hwliang<2022-02-10>
            @param domains<list> 域名列表
            @return string
        '''
        identifiers = []
        for domain_name in domains:
            identifiers.append({"type": 'dns', "value": domain_name})
        return public.md5(json.dumps(identifiers))


    # 续签同品牌其它证书
    def renew_cert_other(self):
        '''
            @name 续签同品牌其它证书
            @author hwliang<2022-02-10>
            @return void
        '''
        cert_path = "{}/vhost/cert".format(public.get_panel_path())
        if not os.path.exists(cert_path): return
        new_time = time.time() + (86400 * 30)
        n=0
        if not 'orders' in self._config: self._config['orders'] = {}
        import panelSite
        siteObj = panelSite.panelSite()
        args = public.dict_obj()
        for siteName in os.listdir(cert_path):
            try:
                cert_file = '{}/{}/fullchain.pem'.format(cert_path,siteName)
                if not os.path.exists(cert_file): continue # 无证书文件
                siteInfo = public.M('sites').where('name=?',siteName).find()
                if not siteInfo: continue # 无网站信息
                cert_init = self.get_cert_init(cert_file)
                if not cert_init: continue # 无法获取证书
                end_time = time.mktime(time.strptime(cert_init['notAfter'],'%Y-%m-%d'))
                if end_time > new_time: continue # 未到期
                try:
                    if not cert_init['issuer'] in ['R3',"Let's Encrypt"] and cert_init['issuer'].find("Let's Encrypt") == -1 and cert_init.get('issuer_O', "") != "Let's Encrypt":
                        continue # 非同品牌证书
                except: continue

                if isinstance(cert_init['dns'],str): cert_init['dns'] = [cert_init['dns']]
                index = self.get_index(cert_init['dns'])
                if index in self._config['orders'].keys(): continue # 已在订单列表

                n+=1
                write_log("|-正在续签第 {} 张其它证书，域名: {}..".format(n,cert_init['subject']))
                write_log("|-正在创建订单..")
                args.id = siteInfo['id']
                runPath = siteObj.GetRunPath(args)
                if runPath and not runPath in ['/']:
                    path = siteInfo['path'] + '/' + runPath
                else:
                    path = siteInfo['path']

                self.renew_cert_to(cert_init['dns'],'http',path.replace('//','/'))
            except:
                write_log("|-[{}]续签失败".format(siteName))

    # 关闭强制https
    def close_httptohttps(self,siteName):
        try:

            if not siteName: siteName
            import panelSite
            site_obj = panelSite.panelSite()
            if not site_obj.IsToHttps(siteName):
                return False
            get = public.dict_obj()
            get.siteName = siteName
            site_obj.CloseToHttps(get)
            return True
        except:
            return False

    # 恢复强制https
    def rep_httptohttps(self,siteName):
        try:
            if not siteName: return False
            import panelSite
            site_obj = panelSite.panelSite()
            if not site_obj.IsToHttps(siteName):
                get = public.dict_obj()
                get.siteName = siteName
                site_obj.HttpToHttps(get)
            return True
        except:
            return False


    def renew_cert_to(self,domains,auth_type,auth_to,index = None):
        siteName = None
        cert = {}
        if os.path.exists(auth_to):
            if public.M('sites').where('path=?',auth_to).count() == 1:
                site_id = public.M('sites').where('path=?',auth_to).getField('id')
                siteName = public.M('sites').where('path=?',auth_to).getField('name')
                r_status = public.M('sites').where('path=?',auth_to).getField('status')
                if r_status != '1':
                    write_log("|- 该证书使用了【文件验证】方式,但网站【{}】未启动，只能跳过续签。".format(siteName))
                    return public.returnMsg(False, "该证书使用了【文件验证】方式,但网站【{}】未启动，只能跳过续签。".format(siteName))
                import panelSite
                siteObj = panelSite.panelSite()
                args = public.dict_obj()
                args.id = site_id
                runPath = siteObj.GetRunPath(args)
                if runPath and not runPath in ['/']:
                    path = auth_to + '/' + runPath
                    if os.path.exists(path): auth_to = path.replace('//','/')

            else:
                siteName, _ = self.get_site_name_by_domains(domains)

            isError = public.checkWebConfig()
            if isError is not True and public.get_webserver() == "nginx":
                write_log("|- 该证书使用了【文件验证】方式,但目前无法重载nginx服务器配置文件，只能跳过续签。")
                write_log("|- 配置文件报错信息如下：")
                write_log(isError)

        is_rep = self.close_httptohttps(siteName)
        try:
            index = self.create_order(
                domains,
                auth_type,
                auth_to.replace('//','/'),
                index
            )

            write_log("|-正在获取验证信息..")
            self.get_auths(index)
            write_log("|-正在验证域名..")
            self.auth_domain(index)
            write_log("|-正在发送CSR..")
            self.remove_dns_record()
            self.send_csr(index)
            write_log("|-正在下载证书..")
            cert = self.download_cert(index)
            self._config['orders'][index]['renew_time'] = int(time.time())

            # 清理失败重试记录
            self._config['orders'][index]['retry_count'] = 0
            self._config['orders'][index]['next_retry_time'] = 0

            # 保存证书配置
            self.save_config()
            cert['status'] = True
            cert['msg'] = '续签成功!'
            write_log("|-续签成功!")
        except Exception as e:

            if str(e).find('请稍候重试') == -1: # 受其它证书影响和连接CA失败的的不记录重试次数
                if index:
                    # 设置下次重试时间
                    self._config['orders'][index]['next_retry_time'] = int(time.time() + (86400 * 2))
                    # 记录重试次数
                    if not 'retry_count' in self._config['orders'][index].keys():
                        self._config['orders'][index]['retry_count'] = 1
                    self._config['orders'][index]['retry_count'] += 1
                    # 保存证书配置
                    self.save_config()
            e = str(e)
            if e.find(">>>>") != -1:
                msg = e.split(">>>>")[0]
                err = json.loads(e.split(">>>>")[1])
            else:
                msg = e
                err = {}
            write_log("|-" + msg)
            return {"status": False, "msg": msg, "err": err}
        finally:
            if is_rep: self.rep_httptohttps(siteName)
        write_log("-" * 70)
        return cert

    def set_auto_renew_status(self, index, status, error_msg=''):
        """
        @name 设置自动续签状态
        @param index:
        @param status:
        @param error_msg:
        @return:
        """
        path = "{}/config/letsencrypt_auto_renew.json".format(public.get_panel_path())
        try:
            data = json.loads(public.readFile(path))
            data[index] = {"status": status, "error_msg": error_msg}
        except:
            data = {index: {"status": status, "error_msg": error_msg}}
        public.writeFile(path, json.dumps(data))


    # 续签证书
    def renew_cert(self, index, cycle=None):
        write_log("", "wb+")
        set_status = False
        index_info = None
        try:
            order_index = []
            if index:
                set_status = True
                if type(index) != str:
                    index = index.index
                if not index in self._config['orders']:
                    write_log("|-指定订单号不存在，无法续签!")
                    self.set_auto_renew_status(index, -1, "指定订单号不存在，无法续签!")
                    raise Exception("指定订单号不存在，无法续签!")
                if cycle:
                    s_time = time.time() + (int(cycle) * 86400)
                    if self._config['orders'][index]['cert_timeout'] > s_time:
                        self.set_auto_renew_status(index, 0, "|-过期时间大于{}天，跳过续签!".format(cycle))
                        write_log("|-过期时间大于{}天，跳过续签!".format(cycle))
                        return
                order_index.append(index)
            else:
                s_time = time.time() + (30 * 86400)
                if not 'orders' in self._config: self._config['orders'] = {}
                for i in self._config['orders'].keys():
                    if not 'save_path' in self._config['orders'][i]:
                        continue
                    if 'cert' in self._config['orders'][i]:
                        self._config['orders'][i]['cert_timeout'] = self._config['orders'][i]['cert']['cert_timeout']
                    if not 'cert_timeout' in self._config['orders'][i]:
                        self._config['orders'][i]['cert_timeout'] = int(time.time())
                    if self._config['orders'][i]['cert_timeout'] > s_time:
                        continue

                    # 已删除的网站直接跳过续签
                    # is_file_check = (self._config['orders'][i]['auth_to'].find('|') == -1 or
                    #                  not self._config['orders'][i]['auth_to'].startswith("dns")
                    #                 ) and self._config['orders'][i]['auth_to'].find('/') != -1
                    # if is_file_check:
                        # if not os.path.exists(self._config['orders'][i]['auth_to']):
                        # ^^^^^^^^^^——————————这个不能判断网站已被删除的情况下，文件夹未删除时的问题
                    # _auth_to = self.get_ssl_used_site(self._config['orders'][i]['save_path'])
                    _auth_to = self.get_ssl_used_site(i)
                    if not _auth_to:
                        continue

                    # 域名不存在？
                    for domain in self._config['orders'][i]['domains']:
                        if domain.find('*') != -1:
                            break
                        if not public.M('domain').where("name=?",(domain,)).count() and not public.M('binding').where("domain=?",domain).count():
                            _auth_to = None
                            write_log("|-跳过被删除的域名: {}".format(self._config['orders'][i]['domains']))
                    if not _auth_to: continue
                    self._config['orders'][i]['auth_to'] = _auth_to

                    # 是否到了允许重试的时间
                    if 'next_retry_time' in self._config['orders'][i]:
                        timeout = self._config['orders'][i]['next_retry_time'] - int(time.time())
                        if timeout > 0:
                            write_log('|-本次跳过域名:{}，因第上次续签失败，还需要等待{}小时后再重试'.format(self._config['orders'][i]['domains'],int(timeout / 60 / 60)))
                            continue

                    # # 是否到了最大重试次数
                    # if 'retry_count' in self._config['orders'][i]:
                    #     if self._config['orders'][i]['retry_count'] >= 5:
                    #         write_log('|-本次跳过域名:{}，因连续5次续签失败，不再续签此证书(可尝试手动续签此证书，成功后错误次数将被重置)'.format(self._config['orders'][i]['domains']))
                    #         continue

                    # 加入到续签订单
                    order_index.append(i)
            if not order_index:
                write_log("|-没有找到30天内到期的SSL证书，正在尝试去寻找其它可续签证书!")
                self.get_apis()
                self.renew_cert_other()
                write_log("|-所有任务已处理完成!")
                return
            write_log("|-共需要续签 {} 张证书".format(len(order_index)))
            n = 0
            self.get_apis()
            cert = None
            for index in order_index:
                n += 1
                domains = _test_domains(self._config['orders'][index]['domains'], self._config['orders'][index]['auth_to'],self._config['orders'][index]['auth_type'])
                if len(domains) == 0:
                    write_log("|-第 {} 张证书下的域名全部未使用(这些域名是:[{}])，已跳过。".format(n, ",".join(self._config['orders'][index]['domains'])))
                    err_msg = "域名全部未使用，已跳过。"
                    if set_status:
                        self.set_auto_renew_status(index, -1, err_msg)
                    continue
                else:
                    index_info = self._config['orders'][index]
                    self._config['orders'][index]['domains'] = domains
                    write_log("|-正在续签第 {} 张，域名: {}..".format(n,self._config['orders'][index]['domains']))
                    write_log("|-正在创建订单..")
                    cert = self.renew_cert_to(self._config['orders'][index]['domains'],self._config['orders'][index]['auth_type'],self._config['orders'][index]['auth_to'],index)
                    err_msg = "续签失败!"
            if not cert:
                return public.returnMsg(False, err_msg)
            if cert.get('status') is False and set_status:
                self.set_auto_renew_status(index, -1, cert.get('msg'))
                self._config['orders'][index] = index_info
                self.save_config()
            if (cert.get('status') is None or cert.get('status') is True) and set_status:
                self.set_auto_renew_status(index, 1, "续签成功!")
            return cert

        except Exception as ex:
            self.remove_dns_record()
            ex = str(ex)
            if ex.find(">>>>") != -1:
                msg = ex.split(">>>>")
                msg[1] = json.loads(msg[1])
            else:
                msg = ex
                write_log(public.get_error_info())
            if set_status:
                self.set_auto_renew_status(index, -1, msg)
            if index_info:
                self._config['orders'][index] = index_info
                self.save_config()
            return public.returnMsg(False, msg)

    def renew_cert_v2(self, index, cycle):
        if index:
            hash_list = [index]
        else:
            # 获取所有网站证书
            from sslModel import certModel
            certModel = certModel.main()
            use_cert_list = certModel.get_cert_to_site(True)
            hash_list = use_cert_list.keys()
        s = 0
        from sslModel import base
        dns_data = base.sslBase().get_dns_data(None)
        for ssl_hash in hash_list:
            s += 1
            write_log("|-正在续签第 {} 张证书，共 {} 张..".format(s, len(hash_list)))
            cert_data = public.M('ssl_info').where('hash=?', ssl_hash).find()
            if not cert_data:
                write_log("|-【{}】没有找到指定证书信息，无法续签!".format(ssl_hash))
                continue
            # 借用ps字段，记录证书续签状态
            if cert_data['ps'] == "renewed":
                write_log("|-【{}】证书已续签，无需续签!".format(ssl_hash))
                continue
            try:
                cert_info = json.loads(cert_data['info'])
            except:
                write_log(public.get_error_info())
                write_log("|-【{}】证书信息格式错误，无法续签，请尝试手动续签!".format(ssl_hash))
                continue
            ca = ""
            if cert_info.get('issuer') in ("R3", "R8", "R11", "R10", "R5") or cert_info.get('issuer_O', '') == "Let's Encrypt":
                ca = "Let's Encrypt"
                self._debug = False
                self._url = "https://acme-v02.api.letsencrypt.org/directory"
            if "LiteSSl" in cert_info.get('issuer') or cert_info.get('issuer_O', '') == "TrustAsia Technologies, Inc.":
                ca = "TrustAsia Technologies, Inc."
                self._debug = "litessl"
                self._url = "https://acme.litessl.com/acme/v2/directory"
            if ca not in ("Let's Encrypt", "TrustAsia Technologies, Inc."):
                write_log("|-【{}】非Let's Encrypt或TrustAsia证书，无法续签，请尝试手动续签!".format(ssl_hash))
                continue
            # 计算 30 天后的日期
            if cycle:
                cycle = int(cycle)
                # 计算 30 天后的日期
                future_date = (datetime.datetime.now().date() + datetime.timedelta(days=cycle)).strftime('%Y-%m-%d')
                if future_date < cert_info['notAfter']:
                    write_log("|-【{}】到期时间大于{}天，无需续签!".format(ssl_hash, cycle))
                    continue
            else:
                try:
                    notBefore = datetime.datetime.strptime(cert_info['notBefore'][:10], "%Y-%m-%d").timestamp()
                    notAfter = datetime.datetime.strptime(cert_info['notAfter'][:10], "%Y-%m-%d").timestamp()
                except Exception as e:
                    write_log("|-【{}】证书时间格式错误，无法续签，请尝试手动续签!".format(ssl_hash))
                    write_log(public.get_error_info())
                    continue
                total_time = notAfter - notBefore
                if total_time >= 90 * 86400:
                    renew_time = notAfter - (30 * 86400)
                else:
                    renew_time = notAfter - (total_time / 3)

                if renew_time > time.time():
                    write_log("|-【{}】未到续签时间，下次续签：{}".format(ssl_hash, time.strftime('%Y-%m-%d', time.localtime(renew_time))))
                    continue
            # 判断是否有泛域名
            wildcard = False
            if "*" in ",".join(cert_info['dns']):
                write_log("|-【{}】存在泛域名，只能用dns验证方式续签!".format(ssl_hash))
                wildcard = True
            # 判断是否绑定了dns-api
            auth_domains = []
            for i in cert_info['dns']:
                root_domain, _, _ = base.sslBase().extract_zone(i)
                root_domain_info = public.M('ssl_domains').where("domain=?", root_domain).find()
                if not root_domain_info or (root_domain_info.get("dns_id") not in dns_data.keys() and "cloud_id=" not in root_domain_info.get("dns_id")):
                    write_log("|-根域名【{}】未绑定dns-api，跳过域名: {}!".format(root_domain, i))
                    continue
                auth_domains.append(i)
            if not auth_domains:
                write_log("|-【{}】域名全部未绑定dns-api，无法使用dns验证续签!".format(ssl_hash))
                dns_auth = False
                if wildcard:
                    continue
            else:
                dns_auth = True
            if set(auth_domains) == set(cert_info['dns']):
                write_log("|-【{}】全部域名已绑定dns-api，正在尝试使用dns验证续签!".format(ssl_hash))
                self.get_apis()
                cert = self.renew_cert_to(auth_domains, "dns", "dns-api")
                if cert.get('status') is False:
                    continue
                public.M('ssl_info').where('hash=?', ssl_hash).update({"ps": "renewed"})
                continue
            else:
                http_auth = False
                write_log("|-【{}】正在检测文件验证是否可用".format(ssl_hash))
                sites = {}
                for domain in cert_info['dns']:
                    domain_info = public.M('domain').where("name=?", domain).find()
                    if not domain_info:
                        write_log("|-域名【{}】不存在，跳过!".format(domain))
                        continue
                    if not sites.get(domain_info["pid"]):
                        sites[domain_info["pid"]] = [domain]
                    else:
                        sites[domain_info["pid"]].append(domain)
                if not sites:
                    write_log("|-【{}】没有找到可用的文件验证站点，请尝试手动续签!".format(ssl_hash))
                # 暂时不做多网站文件验证
                if len(sites.keys()) > 1:
                    write_log("|-【{}】检测到验证域名分散在多个站点，暂时不支持多个站点文件验证!".format(ssl_hash))
                for site_id, domains in sites.items():
                    site_info = public.M('sites').where("id=?", site_id).find()
                    if not site_info:
                        write_log("|-站点【{}】不存在，跳过!".format(site_id))
                        break
                    if set(domains) == set(cert_info['dns']) or len(domains) > len(auth_domains):
                        http_auth = True
                        break
            if http_auth:
                write_log("|-【{}】正在尝试使用文件验证续签!".format(ssl_hash))
                self.get_apis()
                cert = self.renew_cert_to(domains, "http", site_info['path'])
                if cert.get('status') is False:
                    continue
                public.M('ssl_info').where('hash=?', ssl_hash).update({"ps": "renewed"})
                continue
            elif dns_auth:
                write_log("|-【{}】正在尝试使用dns验证续签!".format(ssl_hash))
                self.get_apis()
                cert = self.renew_cert_to(auth_domains, "dns", "dns-api")
                if cert.get('status') is False:
                    continue
                public.M('ssl_info').where('hash=?', ssl_hash).update({"ps": "renewed"})
                continue
            else:
                write_log("|-【{}】没有找到可用的验证方式，请尝试手动续签!".format(ssl_hash))
                continue
        return

    def get_order_list(self, get):
        """
        获取订单列表
        """
        self.get_exclude_hash(get)
        data = self.read_config()
        if not data.get('orders'):
            return []
        status_id = int(get.status_id) if hasattr(get, 'status_id') else -1

        del_list = []
        _return = []

        import panelSSL
        exclude_hash = panelSSL.panelSSL().get_exclude_hash(get)
        format_time_strs = ("%Y-%m-%d", "%Y-%m-%d %H:%M:%S")
        today_time = datetime.datetime.today().timestamp()

        orders = list(data['orders'].values())
        for i in range(len(orders) - 1, -1, -1):
            if orders[i]['status'] == "valid" and (not orders[i].get("save_path") or not orders[i].get("cert_timeout")):
                del_list.append(orders[i]['index'])
                del orders[i]
                continue
            if orders[i]['status'] == "valid":
                orders[i]['expires'] = orders[i]['cert_timeout']
            if orders[i]['status'] == "ready":
                orders[i]['status'] = "pending"
            try:
                end_time = int(
                    (orders[i]['expires'] - datetime.datetime.today().timestamp()) / (60 * 60 * 24)
                )
            except Exception as e:
                end_time = 90
            hash = exclude_hash.get("exclude_hash_let", {}).get(orders[i]['index'], '')
            if hash:
                cert_data = public.M('ssl_info').where('hash=?', hash).find()
                if cert_data:
                    for f_str in format_time_strs:
                        try:
                            end_time = int(
                                (datetime.datetime.strptime(cert_data["not_after"], f_str).timestamp() - today_time) / (60 * 60 * 24)
                            )
                        except Exception as e:
                            continue

            orders[i]['endDay'] = end_time

            if status_id == 0 and (orders[i]['status'] != "pending" or orders[i]['endDay'] <= 0):
                del orders[i]
                continue
            elif status_id == 1 and (orders[i]['endDay'] <= 0 or orders[i]['status'] == "pending"):
                del orders[i]
                continue
            elif status_id == 2 and (orders[i]['endDay'] > 30 or orders[i]['endDay'] <= 0):
                del orders[i]
                continue
            elif status_id == 3 and (orders[i]['endDay'] > 0):
                del orders[i]
                continue
        try:
            arg = public.to_dict_obj({"index": ",".join(del_list), "d": "1"})
            self.delete_order(arg)
        except:
            pass
        orders.sort(key=lambda x: x['endDay'], reverse=True)
        orders.sort(key=lambda x: x['status']=="valid", reverse=True)
        return orders

    def get_order_detail(self, get):
        """
        订单详情
        """
        order_index = get.index
        orders = self.read_config()

        data = orders["orders"].get(order_index)

        if not data:
            return public.returnMsg(False, "没有找到此订单信息")
        if not data.get('auths'):
            if not data.get('authorizations'):
                return public.ReturnMsg(False, "订单验证信息丢失，请尝试重新申请!")
            try:
                self.get_apis()
                data['auths'] = []
                for auth_url in data['authorizations']:
                    res = self.acme_request(auth_url, "")
                    if res.status_code not in [200, 201]:
                        return public.ReturnMsg(False, "订单验证信息丢失，请尝试重新申请!")
                    s_body = res.json()
                    identifier_auth = self.get_identifier_auth(order_index, auth_url, s_body)
                    acme_keyauthorization, auth_value = self.get_keyauthorization(
                        identifier_auth['token'])
                    identifier_auth['acme_keyauthorization'] = acme_keyauthorization
                    identifier_auth['auth_value'] = auth_value
                    identifier_auth['expires'] = s_body['expires']
                    identifier_auth['auth_to'] = self._config['orders'][order_index]['auth_to']
                    identifier_auth['type'] = self._config['orders'][order_index]['auth_type']
                    data['auths'].append(identifier_auth)
                self.save_config()
            except:
                return public.ReturnMsg(False, "订单验证信息丢失，请尝试重新申请!")

        endtime = ((data.get('expires', 0) or 0) - datetime.datetime.today().timestamp()) / (60 * 60 * 24)
        if endtime <= 0:
            return public.ReturnMsg(False, "订单已过期，请重新申请!")

        _return = {"auths": []}
        if data['auth_type'] == 'dns':
            auth_domains = [i["domain"].replace("*.", "") for i in data['auths']]
            if len(auth_domains) != len(set(auth_domains)):
                _return["error"] = True
                _return["error_msg"] = "检测到解析记录存在冲突，请分别验证以下域名"

            for auth in data['auths']:
                domain = auth['domain']
                domains = auth['domain'].split('.')
                if domains[0] == '*':
                    domain = ".".join(domains[1:])
                dns_auth = [{
                        "domain": "_acme-challenge.{}".format(domain),
                        "auth_value": auth["auth_value"],
                        "type": "TXT",
                        "must": "是"
                    }]
                if 'ca' in data and data['ca'] != 'litessl':
                    dns_auth.append({
                        "domain": domain,
                        "auth_value": '0 issue "letsencrypt.org"',
                        "type": "CAA",
                        "must": "否"})
                _return["auths"].append({
                    "domain": auth['domain'],
                    "status": auth.get('status', "pending"),
                    "data": dns_auth
                })
            return _return
        else:
            for auth in data['auths']:
                domain = auth['domain']
                _return["auths"].append({
                    "domain": domain,
                    "data": [{
                        "domain": auth['domain'],
                        "file_path": os.path.join(auth['auth_to'], ".well-known/acme-challenge/", auth['token']),
                        "content": auth['acme_keyauthorization'],
                        "must": "是"
                    }]
                })
            return _return

    def validate_domain(self, get):
        order_index = get.index

        data = self.read_config()
        data = data["orders"].get(order_index)

        if not data:
            return public.returnMsg(False, "没有找到此订单信息")
        if not data.get('auths'):
            return public.ReturnMsg(False, "订单验证信息丢失，请尝试重新申请!")
        endtime = ((data.get('expires', 0) or 0) - datetime.datetime.today().timestamp()) / (60 * 60 * 24)
        if endtime <= 0:
            return public.ReturnMsg(False, "订单已过期，请重新申请!")
        # for auth in data['auths']:
        #     self.set_auth_info(auth, order_index)
        return self.apply_cert([], "dns", "dns", index=order_index)

    def delete_order(self, get):
        return self._delete_order(get)["finish_list"][0]

    def _delete_order(self, get):
        from sslModel import certModel
        certModel = certModel.main()
        if ('index' not in get or not get.index) and ('ssl_hash' not in get or not get.ssl_hash):
            return {'status': False, 'msg': "缺少必填参数", 'finish_list': []}
        # 强制删除已经部署的证书
        force = False
        if hasattr(get, 'force'):
            force = get.force
        local = True if 'local' not in get else get.local
        cloud = True if 'cloud' not in get else get.cloud

        path = '{}/data/exclude_hash.json'.format(public.get_panel_path())
        exclude_data = self.get_exclude_hash(get)

        # 组合删除数据
        del_data = []
        if 'index' in get and get.index:
            index_list = get.index.split(',')
            for index in index_list:
                ssl_hash = exclude_data.get("exclude_hash_let", {}).get(index)
                del_data.append({"index": index, "ssl_hash": ssl_hash})
        if 'ssl_hash' in get and get.ssl_hash:
            ssl_hash_list = get.ssl_hash.split(',')
            for ssl_hash in ssl_hash_list:
                append_data = {"index": "", "ssl_hash": ssl_hash}
                for index, value in exclude_data.get("exclude_hash_let", {}).items():
                    if value == ssl_hash:
                        append_data["index"] = index
                        break
                del_data.append(append_data)

        data = self.read_config()
        finish_list = []
        for d in del_data:
            finish = {"name": "let's Encrypt", "status": True, "msg": "删除成功"}
            local_err = ""
            order_err = ""
            # 删除本地证书
            if d['ssl_hash']:
                try:
                    certModel.remove_cert(ssl_hash=d['ssl_hash'], local=local, cloud=cloud, force=force)
                except Exception as e:
                    local_err = str(e)
            # 删除订单
            if d['index'] and d['index'] in data['orders'] and local:
                try:
                    if d['index'] in exclude_data["exclude_hash_let"]:
                        del exclude_data["exclude_hash_let"][d['index']]
                    del data['orders'][d['index']]
                except Exception as e:
                    order_err = str(e)
            if local_err:
                finish["status"] = False
                finish["msg"] = "本地证书删除失败: {}".format(local_err)
            if order_err:
                finish["status"] = False
                if finish["msg"] == "删除成功":
                    finish["msg"] = "订单删除失败: {}".format(order_err)
                else:
                    finish["msg"] += "；订单删除失败: {}".format(order_err)

            finish_list.append(finish)

        public.writeFile(path, json.dumps(exclude_data))
        public.writeFile(self._conf_file_v2, json.dumps(data))
        if 'd' in get:
            return data
        return {'status': True, 'msg': "删除成功", 'finish_list': finish_list}

    def download_cert_to_local(self, get):
        index = get.index

        orders = self.read_config()
        order = orders['orders'].get(index)
        if not order:
            return public.returnMsg(False, "下载失败，未找到此订单")

        exclude_data = self.get_exclude_hash(get)

        ssl_hash = exclude_data.get("exclude_hash_let", {}).get(index)
        if not ssl_hash:
            return public.returnMsg(False, "订单未完成或订单信息有误，下载失败")
        from sslModel import certModel
        return certModel.main().download_cert(public.to_dict_obj({"ssl_hash": ssl_hash}))

    def SetCertToSite(self,get):
        import panelSSL
        exclude_data = self.get_exclude_hash(get)
        ssl_hash = exclude_data['exclude_hash_let'].get(get.index)
        if not ssl_hash:
            return public.returnMsg(False, "未找到此证书")
        get.ssl_hash = ssl_hash
        return panelSSL.panelSSL().SetCertToSite(get)



def _test_domains(domains, auth_to, auth_type):
    # 检查站点域名变更情况， 若有删除域名，则在续签时，删除已经不使用的域名，再执行续签任务
    # 是dns验证的跳过
    if auth_to.find("|") != -1 or auth_to.startswith("dns#@"):
        return domains
    # 是泛域名的跳过
    for domain in domains:
        if domain.find("*.") != -1:
            return domains
    sql = public.M('domain')
    site_sql = public.M('sites')
    for domain in domains:
        pid = sql.where('name=?', domain).getField('pid')
        if pid and site_sql.where('id=?',pid).find():
            site_domains = [i["name"] for i in sql.where('pid=?',(pid,)).field("name").select()]
            break
    else:
        site_id = site_sql.where('path=?', auth_to).getField('id')
        if bool(site_id) and str(site_id).isdigit():
            site_domains = [i["name"] for i in sql.where('pid=?',(site_id,)).field("name").select()]
        else:
            # 全都查询不到，认为这个站点已经被删除
            return []

    del_domains = list(set(domains) - set(site_domains))
    for i in del_domains:
        domains.remove(i)
    return domains


def echo_err(msg):
    write_log("\033[31m=" * 65)
    write_log("|-错误：{}\033[0m".format(msg))
    exit()


# 写日志
def write_log(log_str, mode="ab+"):
    if __name__ == "__main__":
        print(log_str)
        return
    _log_file = 'logs/letsencrypt.log'
    f = open(_log_file, mode)
    log_str += "\n"
    f.write(log_str.encode('utf-8'))
    f.close()
    return True

# todo：兼容控制台，目前不兼容
if __name__ == "__main__":
    import argparse
    p = argparse.ArgumentParser(usage="必要的参数：--domain 域名列表，多个以逗号隔开!")
    p.add_argument('--domain', default=None,
                   help="请指定要申请证书的域名", dest="domains")
    p.add_argument('--type', default=None, help="请指定验证类型", dest="auth_type")
    p.add_argument('--path', default=None, help="请指定网站根目录", dest="path")
    p.add_argument('--dnsapi', default=None, help="请指定DNSAPI", dest="dnsapi")
    p.add_argument('--dns_key', default=None, help="请指定DNSAPI的key", dest="key")
    p.add_argument('--dns_secret', default=None,
                   help="请指定DNSAPI的secret", dest="secret")
    p.add_argument('--index', default=None, help="指定订单索引", dest="index")
    p.add_argument('--renew', default=None, help="续签证书", dest="renew")
    p.add_argument('--renew_v2', default=None, help="续签证书v2", dest="renew_v2")
    p.add_argument('--revoke', default=None, help="吊销证书", dest="revoke")
    p.add_argument('--cycle', default=None, help="到期时间小于等于时续签", dest="cycle")
    args = p.parse_args()
    cert = None
    if args.revoke:
        if not args.index:
            echo_err("请在--index参数中传入要被吊销的订单索引")
        p = acme_v2()
        result = p.revoke_order(args.index)
        write_log(result)
        exit()

    if args.renew_v2:
        sys.path.append(public.get_panel_path())
        p = acme_v2()
        if args.cycle:
            try:
                int(args.cycle)
            except:
                args.cycle = None
        p.renew_cert_v2(args.index, args.cycle)
        exit()

    if args.renew:
        sys.path.append(public.get_panel_path())
        p = acme_v2()
        if args.cycle:
            try:
                int(args.cycle)
            except:
                args.cycle = None
        p.renew_cert(args.index, args.cycle)
    else:
        try:
            if not args.index:
                if not args.domains:
                    echo_err("请在--domain参数中指定要申请证书的域名，多个以逗号(,)隔开")
                if not args.auth_type in ['http', 'tls', 'dns']:
                    echo_err("请在--type参数中指定正确的验证类型，支持dns和http")
                auth_to = ''
                if args.auth_type in ['http', 'tls']:
                    if not args.path:
                        echo_err("请在--path参数中指定网站根目录!")
                    if not os.path.exists(args.path):
                        echo_err("指定网站根目录不存在，请检查：{}".format(args.path))
                    auth_to = args.path
                else:
                    if args.dnsapi == '0':
                        auth_to = 'dns'
                    else:
                        if not args.key:
                            echo_err("使用dnsapi申请时请在--dns_key参数中指定dnsapi的key!")
                        if not args.secret:
                            echo_err(
                                "使用dnsapi申请时请在--dns_secret参数中指定dnsapi的secret!")
                        auth_to = "{}|{}|{}".format(
                            args.dnsapi, args.key, args.secret)

                domains = args.domains.strip().split(',')
                p = acme_v2()
                cert = p.apply_cert(
                    domains, auth_type=args.auth_type, auth_to=auth_to)
                if args.dnsapi == '0':
                    acme_txt = '_acme-challenge.'
                    acme_caa = '1 issue letsencrypt.org'
                    write_log("=" * 65)
                    write_log("\033[32m|-手动订单提交成功，请按以下提示解析DNS记录: \033[0m")
                    write_log("=" * 65)
                    write_log("|-订单索引: {}".format(cert['index']))
                    write_log(
                        "|-重试命令: ./acme_v2.py --index=\"{}\"".format(cert['index']))
                    write_log(
                        "|-共需要解析 \033[36m{}\033[0m 个域名记录。".format(len(cert['auths'])))
                    for i in range(len(cert['auths'])):
                        write_log('-' * 70)
                        write_log(
                            "|-第 \033[36m{}\033[0m 个域名为：{} ， 请解析以下信息：".format(i+1, cert['auths'][i]['domain']))
                        write_log("|-记录类型: TXT 记录名: \033[41m{}\033[0m  记录值: \033[41m{}\033[0m [必需]".format(
                            acme_txt + cert['auths'][i]['domain'].replace('*.', ''), cert['auths'][i]['auth_value']))
                        write_log("|-记录类型: CAA 记录名: \033[41m{}\033[0m 记录值: \033[41m{}\033[0m [可选]".format(
                            cert['auths'][i]['domain'].replace('*.', ''), acme_caa))
                    write_log('-' * 70)
                    input_data = ""
                    while input_data not in ['y', 'Y', 'n', 'N']:
                        input_msg = "请在完成解析后等待2-3分钟后输入Y按回车以继续验证域名: "
                        if sys.version_info[0] == 2:
                            input_data = raw_input(input_msg)
                        else:
                            input_data = input(input_msg)
                    if input_data in ['n', 'N']:
                        write_log("=" * 65)
                        write_log("|-用户放弃申请，退出程序!")
                        exit()
                    cert = p.apply_cert(
                        [], auth_type=args.auth_type, auth_to='dns', index=cert['index'])
            else:
                # 重新验证
                p = acme_v2()
                cert = p.apply_cert([], auth_type='dns',
                                    auth_to='dns', index=args.index)
        except Exception as ex:
            write_log("|-{}".format(public.get_error_info()))
            exit()
    if not cert:
        exit()
    write_log("=" * 65)
    write_log("|-证书获取成功!")
    write_log("=" * 65)
    write_log("证书到期时间: {}".format(
        public.format_date(times=cert['cert_timeout'])))
    write_log("证书已保存在: {}/".format(cert['save_path']))