Spaces:

GGSheng
/

action

Running

App Files Files Community

action / bt-source /panel /class /acme_v2.py

GGSheng

feat: deploy Gemma 4 to hf space

020c337 verified about 17 hours ago

raw

history blame contribute delete

141 kB

	#!/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)?\sinclude\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)?\sinclude\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"\sinclude\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']))