1336 lines
52 KiB
Python
1336 lines
52 KiB
Python
|
import json
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import shutil
|
|||
|
|
import time
|
|||
|
|
# import OpenSSL
|
|||
|
|
import re
|
|||
|
|
import warnings
|
|||
|
|
from hashlib import md5
|
|||
|
|
from datetime import datetime, timedelta
|
|||
|
|
from typing import Optional, Tuple, List, Dict, Union, Callable
|
|||
|
|
|
|||
|
|
from mod.base import json_response
|
|||
|
|
from .util import webserver, check_server_config, write_file, read_file, GET_CLASS, service_reload
|
|||
|
|
|
|||
|
|
if "/www/server/panel/class" not in sys.path:
|
|||
|
|
sys.path.insert(0, "/www/server/panel/class")
|
|||
|
|
import public
|
|||
|
|
import db
|
|||
|
|
from panelAes import AesCryptPy3
|
|||
|
|
warnings.filterwarnings("ignore", category=SyntaxWarning)
|
|||
|
|
|
|||
|
|
SSL_SAVE_PATH = "{}/vhost/ssl_saved".format(public.get_panel_path())
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _SSLDatabase:
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
db_path = "{}/data/db".format(public.get_panel_path())
|
|||
|
|
if not os.path.exists(db_path):
|
|||
|
|
os.makedirs(db_path, 0o600)
|
|||
|
|
self.db_file = '{}/data/db/ssl_data.db'.format(public.get_panel_path())
|
|||
|
|
if not os.path.exists(self.db_file):
|
|||
|
|
self.init_db()
|
|||
|
|
if not os.path.exists(SSL_SAVE_PATH):
|
|||
|
|
os.makedirs(SSL_SAVE_PATH, 0o600)
|
|||
|
|
|
|||
|
|
def init_db(self):
|
|||
|
|
tmp_db = db.Sql()
|
|||
|
|
setattr(tmp_db, "_Sql__DB_FILE", self.db_file)
|
|||
|
|
|
|||
|
|
create_sql_str = (
|
|||
|
|
"CREATE TABLE IF NOT EXISTS 'ssl_info' ("
|
|||
|
|
"'id' INTEGER PRIMARY KEY AUTOINCREMENT, "
|
|||
|
|
"'hash' TEXT NOT NULL UNIQUE, "
|
|||
|
|
"'path' TEXT NOT NULL, "
|
|||
|
|
"'dns' TEXT NOT NULL, "
|
|||
|
|
"'subject' TEXT NOT NULL, "
|
|||
|
|
"'info' TEXT NOT NULL DEFAULT '', "
|
|||
|
|
"'cloud_id' INTEGER NOT NULL DEFAULT -1, "
|
|||
|
|
"'not_after' TEXT NOT NULL, "
|
|||
|
|
"'use_for_panel' INTEGER NOT NULL DEFAULT 0, "
|
|||
|
|
"'use_for_site' TEXT NOT NULL DEFAULT '[]', "
|
|||
|
|
"'auth_info' TEXT NOT NULL DEFAULT '{}', "
|
|||
|
|
"'create_time' INTEGER NOT NULL DEFAULT (strftime('%s'))"
|
|||
|
|
");"
|
|||
|
|
)
|
|||
|
|
res = tmp_db.execute(create_sql_str)
|
|||
|
|
if isinstance(res, str) and res.startswith("error"):
|
|||
|
|
public.WriteLog("SSL管理", "建表ssl_info失败")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
index_sql_str = "CREATE INDEX IF NOT EXISTS 'hash_index' ON 'ssl_info' ('hash');"
|
|||
|
|
|
|||
|
|
res = tmp_db.execute(index_sql_str)
|
|||
|
|
if isinstance(res, str) and res.startswith("error"):
|
|||
|
|
public.WriteLog("SSL管理", "为ssl_info建立索引hash_index失败")
|
|||
|
|
return
|
|||
|
|
tmp_db.close()
|
|||
|
|
|
|||
|
|
def connection(self):
|
|||
|
|
tmp_db = db.Sql()
|
|||
|
|
setattr(tmp_db, "_Sql__DB_FILE", self.db_file)
|
|||
|
|
tmp_db.table("ssl_info")
|
|||
|
|
return tmp_db
|
|||
|
|
|
|||
|
|
|
|||
|
|
ssl_db = _SSLDatabase()
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _LocalSSLInfoTool:
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self._letsencrypt = self.get_letsencrypt_conf()
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def get_letsencrypt_conf():
|
|||
|
|
conf_file = "{}/config/letsencrypt_v2.json".format(public.get_panel_path())
|
|||
|
|
if not os.path.exists(conf_file):
|
|||
|
|
conf_file = "{}/config/letsencrypt.json".format(public.get_panel_path())
|
|||
|
|
if not os.path.exists(conf_file):
|
|||
|
|
return None
|
|||
|
|
tmp_config = public.readFile(conf_file)
|
|||
|
|
try:
|
|||
|
|
orders = json.loads(tmp_config)["orders"]
|
|||
|
|
except (json.JSONDecodeError, KeyError):
|
|||
|
|
return None
|
|||
|
|
return orders
|
|||
|
|
|
|||
|
|
def get_auth(self, domains):
|
|||
|
|
if self._letsencrypt is None:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
for _, data in self._letsencrypt.items():
|
|||
|
|
if 'save_path' not in data:
|
|||
|
|
continue
|
|||
|
|
for d in data['domains']:
|
|||
|
|
if d in domains:
|
|||
|
|
return {
|
|||
|
|
"auth_type": data.get('auth_type'),
|
|||
|
|
"auth_to": data.get('auth_to')
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
class RealSSLManger:
|
|||
|
|
_REFRESH_TIP = "{}/data/ssl_cloud_refresh.tip".format(public.get_panel_path())
|
|||
|
|
_OTHER_DATA_NAME = ("use_for_panel", "use_for_site",)
|
|||
|
|
|
|||
|
|
def __init__(self, conf_prefix=""):
|
|||
|
|
self._local_ssl_info_tool = None
|
|||
|
|
self._vhost_path = "/www/server/panel/vhost"
|
|||
|
|
self.conf_prefix = conf_prefix
|
|||
|
|
self._tls_v3 = None
|
|||
|
|
self._is_nginx_http3 = None
|
|||
|
|
|
|||
|
|
# 与letsencrypt对接
|
|||
|
|
@property
|
|||
|
|
def local_tool(self):
|
|||
|
|
if self._local_ssl_info_tool is None:
|
|||
|
|
self._local_ssl_info_tool = _LocalSSLInfoTool()
|
|||
|
|
return self._local_ssl_info_tool
|
|||
|
|
return self._local_ssl_info_tool
|
|||
|
|
|
|||
|
|
# 用于部署
|
|||
|
|
@classmethod
|
|||
|
|
def get_cert_for_deploy(cls, ssl_data: dict) -> Union[Dict, str]:
|
|||
|
|
data = {
|
|||
|
|
'privkey': public.readFile(ssl_data["path"] + '/privkey.pem'),
|
|||
|
|
'fullchain': public.readFile(ssl_data["path"] + '/fullchain.pem')
|
|||
|
|
}
|
|||
|
|
if not isinstance(data["privkey"], str) or not isinstance(data["fullchain"], str):
|
|||
|
|
return 'Certificate read error!'
|
|||
|
|
return data
|
|||
|
|
|
|||
|
|
# 是否刷新
|
|||
|
|
@classmethod
|
|||
|
|
def need_refresh(cls):
|
|||
|
|
now = int(time.time())
|
|||
|
|
if not os.path.isfile(cls._REFRESH_TIP):
|
|||
|
|
public.writeFile(cls._REFRESH_TIP, str(now))
|
|||
|
|
return True
|
|||
|
|
last_time = int(public.readFile(cls._REFRESH_TIP))
|
|||
|
|
if last_time + 60 * 60 * 4 < now:
|
|||
|
|
public.writeFile(cls._REFRESH_TIP, str(now))
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# 获取hash指纹
|
|||
|
|
@staticmethod
|
|||
|
|
def ssl_hash(cert_filename: str = None, certificate: str = None, ignore_errors: bool = False) -> Optional[str]:
|
|||
|
|
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 = md5()
|
|||
|
|
md5_obj.update(certificate.encode("utf-8"))
|
|||
|
|
return md5_obj.hexdigest()
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def strf_date(sdate):
|
|||
|
|
return time.strftime('%Y-%m-%d', time.strptime(sdate, '%Y%m%d%H%M%S'))
|
|||
|
|
|
|||
|
|
# 获取证书信息
|
|||
|
|
@classmethod
|
|||
|
|
def get_cert_info(cls, cert_filename: str = None, certificate: str = None):
|
|||
|
|
if cert_filename is not None and os.path.isfile(cert_filename):
|
|||
|
|
certificate = public.readFile(cert_filename)
|
|||
|
|
|
|||
|
|
if "/www/server/panel/class" not in sys.path:
|
|||
|
|
sys.path.insert(0, "/www/server/panel/class")
|
|||
|
|
import ssl_info
|
|||
|
|
return ssl_info.ssl_info().load_ssl_info_by_data(certificate)
|
|||
|
|
|
|||
|
|
# try:
|
|||
|
|
# result = {
|
|||
|
|
# "issuer": '',
|
|||
|
|
# "dns": [],
|
|||
|
|
# }
|
|||
|
|
# x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, certificate.encode("utf-8"))
|
|||
|
|
# # 取产品名称
|
|||
|
|
# issuer = x509.get_issuer()
|
|||
|
|
# result['issuer'] = ''
|
|||
|
|
# if hasattr(issuer, 'CN'):
|
|||
|
|
# result['issuer'] = issuer.CN
|
|||
|
|
# if not result['issuer']:
|
|||
|
|
# is_key = [b'0', '0']
|
|||
|
|
# issue_comp = issuer.get_components()
|
|||
|
|
# if len(issue_comp) == 1:
|
|||
|
|
# is_key = [b'CN', 'CN']
|
|||
|
|
# for iss in issue_comp:
|
|||
|
|
# if iss[0] in is_key:
|
|||
|
|
# result['issuer'] = iss[1].decode()
|
|||
|
|
# break
|
|||
|
|
# if not result['issuer']:
|
|||
|
|
# if hasattr(issuer, 'O'):
|
|||
|
|
# result['issuer'] = issuer.O
|
|||
|
|
# # 取到期时间
|
|||
|
|
# result['notAfter'] = cls.strf_date(x509.get_notAfter().decode("utf-8")[:-1])
|
|||
|
|
# # 取申请时间
|
|||
|
|
# result['notBefore'] = cls.strf_date(x509.get_notBefore().decode("utf-8")[:-1])
|
|||
|
|
# # 取可选名称
|
|||
|
|
# for i in range(x509.get_extension_count()):
|
|||
|
|
# s_name = x509.get_extension(i)
|
|||
|
|
# if s_name.get_short_name() in [b'subjectAltName', 'subjectAltName']:
|
|||
|
|
# s_dns = str(s_name).split(',')
|
|||
|
|
# for d in s_dns:
|
|||
|
|
# result['dns'].append(d.split(':')[1])
|
|||
|
|
# subject = x509.get_subject().get_components()
|
|||
|
|
# # 取主要认证名称
|
|||
|
|
# if len(subject) == 1:
|
|||
|
|
# result['subject'] = subject[0][1].decode()
|
|||
|
|
# else:
|
|||
|
|
# if not result['dns']:
|
|||
|
|
# for sub in subject:
|
|||
|
|
# if sub[0] == b'CN':
|
|||
|
|
# result['subject'] = sub[1].decode()
|
|||
|
|
# break
|
|||
|
|
# if 'subject' in result:
|
|||
|
|
# result['dns'].append(result['subject'])
|
|||
|
|
# else:
|
|||
|
|
# result['subject'] = result['dns'][0]
|
|||
|
|
# return result
|
|||
|
|
# except:
|
|||
|
|
# return None
|
|||
|
|
|
|||
|
|
# 通过文件名称检查并保存
|
|||
|
|
def save_by_file(self, cert_filename, private_key_filename, cloud_id=None, other_data: Optional[Dict] = None):
|
|||
|
|
if not os.path.isfile(cert_filename) or not os.path.isfile(private_key_filename):
|
|||
|
|
raise ValueError("不存在的证书")
|
|||
|
|
|
|||
|
|
certificate = public.readFile(cert_filename)
|
|||
|
|
private_key = public.readFile(private_key_filename)
|
|||
|
|
if not isinstance(certificate, str) or not isinstance(private_key, str):
|
|||
|
|
raise ValueError("证书格式错误")
|
|||
|
|
return self.save_by_data(certificate, private_key, cloud_id=cloud_id)
|
|||
|
|
|
|||
|
|
# 通过证书内容检查并保存
|
|||
|
|
def save_by_data(self, certificate: str,
|
|||
|
|
private_key: str,
|
|||
|
|
cloud_id: Optional[int] = None,
|
|||
|
|
other_data: Optional[Dict] = None) -> Dict:
|
|||
|
|
|
|||
|
|
if not certificate.startswith("-----BEGIN") or not private_key.startswith("-----BEGIN"):
|
|||
|
|
raise ValueError("证书格式检查错误")
|
|||
|
|
|
|||
|
|
if cloud_id is None:
|
|||
|
|
cloud_id = -1
|
|||
|
|
|
|||
|
|
hash_data = self.ssl_hash(certificate=certificate)
|
|||
|
|
db_data = self.get_ssl_info_by_hash(hash_data)
|
|||
|
|
if db_data is not None: # 已经保存过的
|
|||
|
|
# 检查 cloud_id 与 保存的 cloud_id 不同时，更新cloud_id
|
|||
|
|
if db_data['cloud_id'] != cloud_id and cloud_id != -1:
|
|||
|
|
ssl_db.connection().where("id = ?", (db_data["id"],)).update({"cloud_id": cloud_id})
|
|||
|
|
db_data['cloud_id'] = cloud_id
|
|||
|
|
db_data["dns"] = json.loads(db_data["dns"])
|
|||
|
|
db_data["info"] = json.loads(db_data["info"])
|
|||
|
|
db_data["auth_info"] = json.loads(db_data["auth_info"])
|
|||
|
|
db_data["use_for_site"] = json.loads(db_data["use_for_site"])
|
|||
|
|
return db_data
|
|||
|
|
info = self.get_cert_info(certificate=certificate)
|
|||
|
|
if info is None:
|
|||
|
|
raise ValueError("证书信息解析错误")
|
|||
|
|
|
|||
|
|
auth_info = self.local_tool.get_auth(info['dns'])
|
|||
|
|
if auth_info is None:
|
|||
|
|
auth_info = {}
|
|||
|
|
|
|||
|
|
pdata = {
|
|||
|
|
"hash": hash_data,
|
|||
|
|
"path": "{}/{}".format(SSL_SAVE_PATH, hash_data),
|
|||
|
|
"dns": json.dumps(info['dns']),
|
|||
|
|
"subject": info['subject'],
|
|||
|
|
"info": json.dumps(info),
|
|||
|
|
"cloud_id": cloud_id,
|
|||
|
|
"not_after": info["notAfter"],
|
|||
|
|
"auth_info": json.dumps(auth_info)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if other_data:
|
|||
|
|
for key, other_data in other_data.items():
|
|||
|
|
if key in self._OTHER_DATA_NAME:
|
|||
|
|
pdata[key] = other_data
|
|||
|
|
|
|||
|
|
res_id = ssl_db.connection().insert(pdata)
|
|||
|
|
if isinstance(res_id, str) and res_id.startswith("error"):
|
|||
|
|
raise ValueError("数据库写入错误：" + res_id)
|
|||
|
|
|
|||
|
|
pdata["id"] = res_id
|
|||
|
|
if not os.path.exists(pdata["path"]):
|
|||
|
|
os.makedirs(pdata["path"], 0o600)
|
|||
|
|
|
|||
|
|
public.writeFile("{}/privkey.pem".format(pdata["path"]), private_key)
|
|||
|
|
public.writeFile("{}/fullchain.pem".format(pdata["path"]), certificate)
|
|||
|
|
public.writeFile("{}/info.json".format(pdata["path"]), pdata["info"])
|
|||
|
|
|
|||
|
|
pdata["info"] = info
|
|||
|
|
return pdata
|
|||
|
|
|
|||
|
|
# 通过hash指纹获取ssl信息
|
|||
|
|
@staticmethod
|
|||
|
|
def get_ssl_info_by_hash(hash_data: str) -> Optional[dict]:
|
|||
|
|
data = ssl_db.connection().where("hash = ?", (hash_data,)).find()
|
|||
|
|
if isinstance(data, str):
|
|||
|
|
raise ValueError("数据库查询错误：" + data)
|
|||
|
|
if len(data) == 0:
|
|||
|
|
return None
|
|||
|
|
return data
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _get_cbc_key_and_iv(with_uer_info=True):
|
|||
|
|
uer_info_file = "{}/data/userInfo.json".format(public.get_panel_path())
|
|||
|
|
try:
|
|||
|
|
user_info = json.loads(public.readFile(uer_info_file))
|
|||
|
|
uid = user_info["uid"]
|
|||
|
|
except (json.JSONDecodeError, KeyError):
|
|||
|
|
return None, None, None
|
|||
|
|
|
|||
|
|
md5_obj = md5()
|
|||
|
|
md5_obj.update(str(uid).encode('utf8'))
|
|||
|
|
bytes_data = md5_obj.hexdigest()
|
|||
|
|
|
|||
|
|
key = ''
|
|||
|
|
iv = ''
|
|||
|
|
for i in range(len(bytes_data)):
|
|||
|
|
if i % 2 == 0:
|
|||
|
|
iv += bytes_data[i]
|
|||
|
|
else:
|
|||
|
|
key += bytes_data[i]
|
|||
|
|
|
|||
|
|
if with_uer_info:
|
|||
|
|
return key, iv, user_info
|
|||
|
|
|
|||
|
|
return key, iv, None
|
|||
|
|
|
|||
|
|
def get_cert_list(self,
|
|||
|
|
param: Optional[Tuple[str, List]] = None,
|
|||
|
|
force_refresh: bool = False,
|
|||
|
|
local_refresh: bool = False) -> List:
|
|||
|
|
if self.need_refresh() or force_refresh:
|
|||
|
|
self._refresh_ssl_info_by_cloud()
|
|||
|
|
self._get_ssl_by_local_data()
|
|||
|
|
elif local_refresh:
|
|||
|
|
self._get_ssl_by_local_data()
|
|||
|
|
|
|||
|
|
return self._get_cert_list(param)
|
|||
|
|
|
|||
|
|
# 获取证书列表
|
|||
|
|
@classmethod
|
|||
|
|
def _get_cert_list(cls, param: Optional[Tuple[str, List]]) -> List:
|
|||
|
|
db_conn = ssl_db.connection()
|
|||
|
|
if param is not None and len(param) == 2 and isinstance(param[0], str) and isinstance(param[1], (tuple, list)):
|
|||
|
|
db_conn.where(param[0], param[1])
|
|||
|
|
res = db_conn.select()
|
|||
|
|
if isinstance(res, str):
|
|||
|
|
raise ValueError("数据库查询错误：" + res)
|
|||
|
|
|
|||
|
|
for value in res:
|
|||
|
|
value["dns"] = json.loads(value["dns"])
|
|||
|
|
value["info"] = json.loads(value["info"])
|
|||
|
|
value["auth_info"] = json.loads(value["auth_info"])
|
|||
|
|
value["use_for_site"] = json.loads(value["use_for_site"])
|
|||
|
|
value['endtime'] = int((datetime.strptime(value['not_after'], "%Y-%m-%d").timestamp()
|
|||
|
|
- datetime.today().timestamp()) / (60 * 60 * 24))
|
|||
|
|
|
|||
|
|
res.sort(key=lambda x: x["not_after"], reverse=True)
|
|||
|
|
|
|||
|
|
return res
|
|||
|
|
|
|||
|
|
# 从云端收集证书
|
|||
|
|
def _refresh_ssl_info_by_cloud(self):
|
|||
|
|
if public.is_self_hosted():
|
|||
|
|
raise ValueError(public.lang('Certificate cloud sync is not available in self-hosted mode.'))
|
|||
|
|
key, iv, user_info = self._get_cbc_key_and_iv(with_uer_info=True)
|
|||
|
|
if key is None or iv is None:
|
|||
|
|
raise ValueError('面板未登录，无法链接云端!')
|
|||
|
|
|
|||
|
|
AES = AesCryptPy3(key, "CBC", iv, char_set="utf8")
|
|||
|
|
|
|||
|
|
# 对接云端
|
|||
|
|
url = "https://wafapi2.yakpanel.com/api/Cert_cloud_deploy/get_cert_list"
|
|||
|
|
try:
|
|||
|
|
res_text = public.httpPost(url, {
|
|||
|
|
"uid": user_info["uid"],
|
|||
|
|
"access_key": 'B' * 32,
|
|||
|
|
"serverid": user_info["server_id"],
|
|||
|
|
})
|
|||
|
|
res_data = json.loads(res_text)
|
|||
|
|
if res_data["status"] is False:
|
|||
|
|
raise ValueError("获取云端数据失败")
|
|||
|
|
|
|||
|
|
res_list = res_data['data']
|
|||
|
|
except:
|
|||
|
|
raise ValueError("链接云端失败")
|
|||
|
|
|
|||
|
|
change_set = set()
|
|||
|
|
for data in res_list:
|
|||
|
|
try:
|
|||
|
|
privateKey = AES.aes_decrypt(data["privateKey"])
|
|||
|
|
certificate = AES.aes_decrypt(data["certificate"])
|
|||
|
|
cloud_id = data["id"]
|
|||
|
|
change_data = self.save_by_data(certificate, privateKey, cloud_id)
|
|||
|
|
change_set.add(change_data.get("id"))
|
|||
|
|
except:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
all_ids = ssl_db.connection().field("id").select()
|
|||
|
|
for ssl_id in all_ids:
|
|||
|
|
if ssl_id["id"] not in change_set:
|
|||
|
|
ssl_db.connection().where("id = ?", (ssl_id["id"],)).update({"cloud_id": -1})
|
|||
|
|
|
|||
|
|
# 从本地收集证书
|
|||
|
|
def _get_ssl_by_local_data(self): # 从本地获取可用证书
|
|||
|
|
local_paths = ['/www/server/panel/vhost/cert', '/www/server/panel/vhost/ssl']
|
|||
|
|
for path in local_paths:
|
|||
|
|
if not os.path.exists(path):
|
|||
|
|
continue
|
|||
|
|
for p_name in os.listdir(path):
|
|||
|
|
pem_file = "{}/{}/fullchain.pem".format(path, p_name)
|
|||
|
|
key_file = "{}/{}/privkey.pem".format(path, p_name)
|
|||
|
|
if os.path.isfile(pem_file) and os.path.isfile(key_file):
|
|||
|
|
try:
|
|||
|
|
self.save_by_file(pem_file, key_file)
|
|||
|
|
except:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
panel_pem_file = "/www/server/panel/ssl/fullchain.pem"
|
|||
|
|
panel_key_file = "/www/server/panel/ssl/privkey.pem"
|
|||
|
|
if os.path.isfile(panel_pem_file) and os.path.isfile(panel_key_file):
|
|||
|
|
try:
|
|||
|
|
self.save_by_file(panel_pem_file, panel_key_file, other_data={"use_for_panel": 1})
|
|||
|
|
except:
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
# 从源储存位置删除
|
|||
|
|
@classmethod
|
|||
|
|
def _remove_ssl_from_local(cls, ssh_hash: str):
|
|||
|
|
local_path = '/www/server/panel/vhost/ssl'
|
|||
|
|
if not os.path.exists(local_path):
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
for p_name in os.listdir(local_path):
|
|||
|
|
pem_file = "{}/{}/fullchain.pem".format(local_path, p_name)
|
|||
|
|
|
|||
|
|
if os.path.isfile(pem_file):
|
|||
|
|
hash_data = cls.ssl_hash(cert_filename=pem_file)
|
|||
|
|
if hash_data == ssh_hash:
|
|||
|
|
shutil.rmtree("{}/{}".format(local_path, p_name))
|
|||
|
|
|
|||
|
|
# 查询证书
|
|||
|
|
@staticmethod
|
|||
|
|
def find_ssl_info(ssl_id=None, ssl_hash=None) -> Optional[dict]:
|
|||
|
|
tmp_conn = ssl_db.connection()
|
|||
|
|
if ssl_id is None and ssl_hash is None:
|
|||
|
|
raise ValueError("没有参数信息")
|
|||
|
|
if ssl_id is not None:
|
|||
|
|
tmp_conn.where("id = ?", (ssl_id,))
|
|||
|
|
else:
|
|||
|
|
tmp_conn.where("hash = ?", (ssl_hash,))
|
|||
|
|
|
|||
|
|
target = tmp_conn.find()
|
|||
|
|
if isinstance(target, str) and target.startswith("error"):
|
|||
|
|
raise ValueError("数据库查询错误：" + target)
|
|||
|
|
|
|||
|
|
if not bool(target):
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
target["auth_info"] = json.loads(target["auth_info"])
|
|||
|
|
target["use_for_site"] = json.loads(target["use_for_site"])
|
|||
|
|
target["dns"] = json.loads(target["dns"])
|
|||
|
|
target["info"] = json.loads(target["info"])
|
|||
|
|
target['endtime'] = int((datetime.strptime(target['not_after'], "%Y-%m-%d").timestamp()
|
|||
|
|
- datetime.today().timestamp()) / (60 * 60 * 24))
|
|||
|
|
return target
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def add_use_for_site(cls, site_id, ssl_id=None, ssl_hash=None) -> bool:
|
|||
|
|
return cls.change_use_for_site(site_id, ssl_id, ssl_hash, is_add=True)
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def remove_use_for_site(cls, site_id, ssl_id=None, ssl_hash=None):
|
|||
|
|
return cls.change_use_for_site(site_id, ssl_id, ssl_hash, is_add=False)
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def change_use_for_site(cls, site_id, ssl_id=None, ssl_hash=None, is_add=True):
|
|||
|
|
target = cls.find_ssl_info(ssl_id=ssl_id, ssl_hash=ssl_hash)
|
|||
|
|
if not target:
|
|||
|
|
return False
|
|||
|
|
try:
|
|||
|
|
site_ids = json.loads(target["use_for_site"])
|
|||
|
|
except:
|
|||
|
|
site_ids = []
|
|||
|
|
|
|||
|
|
if site_id in site_ids and is_add is False:
|
|||
|
|
site_ids.remove(site_id)
|
|||
|
|
up_res = ssl_db.connection().where("id = ?", (target["id"],)).update({"use_for_site": json.dumps(site_ids)})
|
|||
|
|
if isinstance(up_res, str) and up_res.startswith("error"):
|
|||
|
|
raise ValueError("数据库查询错误：" + up_res)
|
|||
|
|
|
|||
|
|
if site_id not in site_ids and is_add is True:
|
|||
|
|
site_ids.append(site_id)
|
|||
|
|
up_res = ssl_db.connection().where("id = ?", (target["id"],)).update({"use_for_site": json.dumps(site_ids)})
|
|||
|
|
if isinstance(up_res, str) and up_res.startswith("error"):
|
|||
|
|
raise ValueError("数据库查询错误：" + up_res)
|
|||
|
|
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
def get_all_site_ssl(self):
|
|||
|
|
all_sites = public.M("sites").select()
|
|||
|
|
self.clear_use_for_site()
|
|||
|
|
if isinstance(all_sites, str) and all_sites.startswith("error"):
|
|||
|
|
raise ValueError(all_sites)
|
|||
|
|
for site in all_sites:
|
|||
|
|
prefix = "" if site["project_type"] == "PHP" else site["project_type"].lower() + "_"
|
|||
|
|
tmp = self._get_site_ssl_info(site["name"], prefix=prefix)
|
|||
|
|
if tmp is None:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
hash_data = self.ssl_hash(cert_filename=tmp[0])
|
|||
|
|
self.add_use_for_site(site["id"], ssl_hash=hash_data)
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def clear_use_for_site():
|
|||
|
|
ssl_db.connection().update({"use_for_site": "[]"})
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def _get_site_ssl_info(site_name, prefix='') -> Optional[Tuple[str, str]]:
|
|||
|
|
path = os.path.join('/www/server/panel/vhost/cert/', site_name)
|
|||
|
|
|
|||
|
|
pem_file = os.path.join(path, "fullchain.pem")
|
|||
|
|
key_file = os.path.join(path, "privkey.pem")
|
|||
|
|
if not os.path.isfile(pem_file) or not os.path.isfile(key_file):
|
|||
|
|
path = os.path.join('/etc/letsencrypt/live/', site_name)
|
|||
|
|
pem_file = os.path.join(path, "fullchain.pem")
|
|||
|
|
key_file = os.path.join(path, "privkey.pem")
|
|||
|
|
if not os.path.isfile(pem_file) or not os.path.isfile(key_file):
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
webserver = public.get_webserver()
|
|||
|
|
if webserver == "nginx":
|
|||
|
|
conf_file = "{}/vhost/nginx/{}{}.conf".format(public.get_panel_path(), prefix, site_name)
|
|||
|
|
elif webserver == "apache":
|
|||
|
|
conf_file = "{}/vhost/apache/{}{}.conf".format(public.get_panel_path(), prefix, site_name)
|
|||
|
|
else:
|
|||
|
|
conf_file = "{}/vhost/openlitespeed/detail/{}.conf".format(public.get_panel_path(), site_name)
|
|||
|
|
|
|||
|
|
conf = public.readFile(conf_file)
|
|||
|
|
if not conf:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
if public.get_webserver() == 'nginx':
|
|||
|
|
keyText = 'ssl_certificate'
|
|||
|
|
elif public.get_webserver() == 'apache':
|
|||
|
|
keyText = 'SSLCertificateFile'
|
|||
|
|
else:
|
|||
|
|
keyText = 'openlitespeed/detail/ssl'
|
|||
|
|
|
|||
|
|
if conf.find(keyText) == -1:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
return pem_file, key_file
|
|||
|
|
|
|||
|
|
# 删除证书
|
|||
|
|
def remove_cert(self, ssl_id=None, ssl_hash=None, local: bool = False) -> Dict:
|
|||
|
|
_, _, user_info = self._get_cbc_key_and_iv(with_uer_info=True)
|
|||
|
|
if user_info is None:
|
|||
|
|
raise ValueError('面板未登录，无法上传云端!')
|
|||
|
|
|
|||
|
|
target = self.find_ssl_info(ssl_id=ssl_id, ssl_hash=ssl_hash)
|
|||
|
|
if not target:
|
|||
|
|
raise ValueError('没有指定的证书')
|
|||
|
|
|
|||
|
|
if local:
|
|||
|
|
shutil.rmtree(target["path"])
|
|||
|
|
self._remove_ssl_from_local(target["hash"]) # 把ssl下的也删除
|
|||
|
|
ssl_db.connection().delete(id=target["id"])
|
|||
|
|
|
|||
|
|
if target["cloud_id"] != -1 and not public.is_self_hosted():
|
|||
|
|
url = "https://wafapi2.yakpanel.com/api/Cert_cloud_deploy/del_cert"
|
|||
|
|
try:
|
|||
|
|
res_text = public.httpPost(url, {
|
|||
|
|
"cert_id": target["cloud_id"],
|
|||
|
|
"hashVal": target["hash"],
|
|||
|
|
"uid": user_info["uid"],
|
|||
|
|
"access_key": 'B' * 32,
|
|||
|
|
"serverid": user_info["server_id"],
|
|||
|
|
})
|
|||
|
|
res_data = json.loads(res_text)
|
|||
|
|
if res_data["status"] is False:
|
|||
|
|
return res_data
|
|||
|
|
except:
|
|||
|
|
if local:
|
|||
|
|
raise ValueError("本地以删除成功, 链接云端失败, 无法删除云端数据")
|
|||
|
|
raise ValueError("链接云端失败, 无法删除云端数据")
|
|||
|
|
|
|||
|
|
if not local:
|
|||
|
|
ssl_db.connection().where("id = ?", (target["id"],)).update({"cloud_id": -1})
|
|||
|
|
|
|||
|
|
elif target["cloud_id"] != -1 and public.is_self_hosted() and not local:
|
|||
|
|
ssl_db.connection().where("id = ?", (target["id"],)).update({"cloud_id": -1})
|
|||
|
|
|
|||
|
|
return public.returnMsg(True, "删除成功")
|
|||
|
|
|
|||
|
|
def mutil_remove_cert(self, ssl_id_list: List[int], local: bool = False):
|
|||
|
|
result = []
|
|||
|
|
for i in ssl_id_list:
|
|||
|
|
try:
|
|||
|
|
ssl_id = int(i)
|
|||
|
|
except:
|
|||
|
|
result.append({"status": False, "msg": "id信息解析错误"})
|
|||
|
|
continue
|
|||
|
|
res = self.remove_cert(ssl_id=ssl_id, local=local)
|
|||
|
|
result.append(res)
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
# 下载证书
|
|||
|
|
def upload_cert(self, ssl_id=None, ssl_hash=None) -> Dict:
|
|||
|
|
key, iv, user_info = self._get_cbc_key_and_iv()
|
|||
|
|
if key is None or iv is None:
|
|||
|
|
raise ValueError(False, '面板未登录，无法上传云端!')
|
|||
|
|
|
|||
|
|
target = self.find_ssl_info(ssl_id=ssl_id, ssl_hash=ssl_hash)
|
|||
|
|
if not target:
|
|||
|
|
raise ValueError("没有指定的证书信息")
|
|||
|
|
|
|||
|
|
data = {
|
|||
|
|
'privateKey': public.readFile(target["path"] + '/privkey.pem'),
|
|||
|
|
'certificate': public.readFile(target["path"] + '/fullchain.pem'),
|
|||
|
|
"encryptWay": "AES-128-CBC",
|
|||
|
|
"hashVal": target['hash'],
|
|||
|
|
"uid": user_info["uid"],
|
|||
|
|
"access_key": 'B' * 32,
|
|||
|
|
"serverid": user_info["server_id"],
|
|||
|
|
}
|
|||
|
|
if data["privateKey"] is False or data["certificate"] is False:
|
|||
|
|
raise ValueError('证书文件读取错误')
|
|||
|
|
|
|||
|
|
AES = AesCryptPy3(key, "CBC", iv, char_set="utf8")
|
|||
|
|
data["privateKey"] = AES.aes_encrypt(data["privateKey"])
|
|||
|
|
data["certificate"] = AES.aes_encrypt(data["certificate"])
|
|||
|
|
if public.is_self_hosted():
|
|||
|
|
raise ValueError(public.lang('Certificate cloud deploy is not available in self-hosted mode.'))
|
|||
|
|
# 对接云端
|
|||
|
|
url = "https://wafapi2.yakpanel.com/api/Cert_cloud_deploy/cloud_deploy"
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
res_text = public.httpPost(url, data)
|
|||
|
|
res_data = json.loads(res_text)
|
|||
|
|
if res_data["status"] is True:
|
|||
|
|
cloud_id = int(res_data["data"].get("id"))
|
|||
|
|
ssl_db.connection().where("id = ?", (target["id"],)).update({"cloud_id": cloud_id})
|
|||
|
|
|
|||
|
|
return res_data
|
|||
|
|
else:
|
|||
|
|
return res_data
|
|||
|
|
except:
|
|||
|
|
raise ValueError('链接云端失败')
|
|||
|
|
|
|||
|
|
# ssl_hash 证书储存记录的唯一值
|
|||
|
|
def set_site_ssl_conf(self, site_name: str, ssl_data: dict, mutil=False) -> Optional[str]:
|
|||
|
|
privkey = ssl_data["privkey"]
|
|||
|
|
fullchain = ssl_data["fullchain"]
|
|||
|
|
path = '/www/server/panel/vhost/cert/' + site_name
|
|||
|
|
if not os.path.exists(path):
|
|||
|
|
os.makedirs(path)
|
|||
|
|
|
|||
|
|
csrpath = path + "/fullchain.pem"
|
|||
|
|
keypath = path + "/privkey.pem"
|
|||
|
|
|
|||
|
|
# 清理旧的证书链
|
|||
|
|
remove_list = [keypath, csrpath, path + "/certOrderId", path + "/README"]
|
|||
|
|
for i in remove_list:
|
|||
|
|
if os.path.exists(i):
|
|||
|
|
os.remove(i)
|
|||
|
|
public.ExecShell('rm -rf ' + path + '-00*')
|
|||
|
|
public.ExecShell('rm -rf /etc/letsencrypt/archive/' + site_name)
|
|||
|
|
public.ExecShell('rm -rf /etc/letsencrypt/archive/' + site_name + '-00*')
|
|||
|
|
public.ExecShell('rm -f /etc/letsencrypt/renewal/' + site_name + '.conf')
|
|||
|
|
public.ExecShell('rm -f /etc/letsencrypt/renewal/' + site_name + '-00*.conf')
|
|||
|
|
|
|||
|
|
public.writeFile(keypath, privkey)
|
|||
|
|
public.writeFile(csrpath, fullchain)
|
|||
|
|
error_msg = self._set_ssl_conf_to_nginx(site_name, mutil)
|
|||
|
|
if error_msg is not None and webserver() == "nginx":
|
|||
|
|
return error_msg
|
|||
|
|
error_msg = self._set_ssl_conf_to_apache(site_name, mutil)
|
|||
|
|
if error_msg is not None and webserver() == "apache":
|
|||
|
|
return error_msg
|
|||
|
|
|
|||
|
|
# http3是否可用
|
|||
|
|
def is_nginx_http3(self):
|
|||
|
|
"""判断nginx是否可以使用http3"""
|
|||
|
|
if getattr(self, "_is_nginx_http3", None) is None:
|
|||
|
|
_is_nginx_http3 = public.ExecShell("nginx -V 2>&1| grep 'http_v3_module'")[0] != ''
|
|||
|
|
setattr(self, "_is_nginx_http3", _is_nginx_http3)
|
|||
|
|
return self._is_nginx_http3
|
|||
|
|
|
|||
|
|
# 在防火墙放行443
|
|||
|
|
@staticmethod
|
|||
|
|
def open_firewall_443() -> None:
|
|||
|
|
import firewalls
|
|||
|
|
get = GET_CLASS()
|
|||
|
|
get.port = '443'
|
|||
|
|
get.ps = 'HTTPS'
|
|||
|
|
firewalls.firewalls().AddAcceptPort(get)
|
|||
|
|
|
|||
|
|
# 在nginx配置文件中设置ssl信息
|
|||
|
|
def _set_ssl_conf_to_nginx(self, site_name, mutil=False) -> Optional[str]:
|
|||
|
|
# Nginx配置
|
|||
|
|
file = '{}/nginx/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
ng_conf = read_file(file)
|
|||
|
|
if not ng_conf:
|
|||
|
|
return "配置文件丢失，配置失败"
|
|||
|
|
|
|||
|
|
http3_header = ""
|
|||
|
|
if self.is_nginx_http3():
|
|||
|
|
http3_header = '''\n add_header Alt-Svc 'quic=":443"; h3=":443"; h3-29=":443"; h3-27=":443";h3-25=":443"; h3-T050=":443"; h3-Q050=":443";h3-Q049=":443";h3-Q048=":443"; h3-Q046=":443"; h3-Q043=":443"';'''
|
|||
|
|
|
|||
|
|
if ng_conf.find('ssl_certificate') == -1:
|
|||
|
|
sslStr = """#error_page 404/404.html;
|
|||
|
|
ssl_certificate /www/server/panel/vhost/cert/%s/fullchain.pem;
|
|||
|
|
ssl_certificate_key /www/server/panel/vhost/cert/%s/privkey.pem;
|
|||
|
|
ssl_protocols %s;
|
|||
|
|
ssl_ciphers EECDH+CHACHA20:EECDH+CHACHA20-draft:EECDH+AES128:RSA+AES128:EECDH+AES256:RSA+AES256:EECDH+3DES:RSA+3DES:!MD5;
|
|||
|
|
ssl_prefer_server_ciphers on;
|
|||
|
|
ssl_session_cache shared:SSL:10m;
|
|||
|
|
ssl_session_timeout 10m;
|
|||
|
|
add_header Strict-Transport-Security "max-age=31536000";%s
|
|||
|
|
error_page 497 https://$host$request_uri;""" % (
|
|||
|
|
site_name, site_name, self._get_tls_protocol(is_apache=False), http3_header
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
new_ng_conf = ng_conf.replace('#error_page 404/404.html;', sslStr)
|
|||
|
|
# 添加端口
|
|||
|
|
from .domain_tool import NginxDomainTool
|
|||
|
|
new_ng_conf = NginxDomainTool.nginx_add_port_by_config(new_ng_conf, "443", is_http3=self.is_nginx_http3())
|
|||
|
|
write_file(file, new_ng_conf)
|
|||
|
|
if webserver() == "nginx" and check_server_config() is not None:
|
|||
|
|
return "配置失败"
|
|||
|
|
if webserver() == "nginx" and not mutil:
|
|||
|
|
service_reload()
|
|||
|
|
self.open_firewall_443()
|
|||
|
|
|
|||
|
|
# 在apache配置文件中设置ssl信息
|
|||
|
|
def _set_ssl_conf_to_apache(self, site_name, mutil=False) -> Optional[str]:
|
|||
|
|
ap_file = '{}/apache/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
ap_conf = read_file(ap_file)
|
|||
|
|
if not ap_conf:
|
|||
|
|
return "配置文件丢失，配置失败"
|
|||
|
|
|
|||
|
|
tmp_template_res = re.search(r"<VirtualHost(.|\n)*?</VirtualHost>", ap_conf)
|
|||
|
|
if not tmp_template_res:
|
|||
|
|
return "配置文件丢失，配置失败"
|
|||
|
|
else:
|
|||
|
|
tmp_template = tmp_template_res.group()
|
|||
|
|
|
|||
|
|
rep_template_with_ports = re.compile(r"<VirtualHost +.*:(?P<port>\d+)+\s*>(.|\n)*?</VirtualHost>")
|
|||
|
|
target_vhost = None
|
|||
|
|
for tmp in rep_template_with_ports.finditer(ap_conf):
|
|||
|
|
if tmp.group("port") == "443":
|
|||
|
|
target_vhost = tmp.group()
|
|||
|
|
|
|||
|
|
if target_vhost and (target_vhost.find("SSLEngine On") or target_vhost.find("SSLCertificateFile")):
|
|||
|
|
return
|
|||
|
|
if not target_vhost:
|
|||
|
|
rep_ports = re.compile(r"<VirtualHost +.*:(?P<port>\d+)+\s*>")
|
|||
|
|
target_vhost = rep_ports.sub("<VirtualHost *:443>", tmp_template, 1)
|
|||
|
|
|
|||
|
|
# 添加SSL配置
|
|||
|
|
ssl_conf = """
|
|||
|
|
#SSL
|
|||
|
|
SSLEngine On
|
|||
|
|
SSLCertificateFile /www/server/panel/vhost/cert/%s/fullchain.pem
|
|||
|
|
SSLCertificateKeyFile /www/server/panel/vhost/cert/%s/privkey.pem
|
|||
|
|
SSLCipherSuite EECDH+CHACHA20:EECDH+CHACHA20-draft:EECDH+AES128:RSA+AES128:EECDH+AES256:RSA+AES256:EECDH+3DES:RSA+3DES:!MD5:ALL:!ADH:!EXPORT56:RC4+RSA:+HIGH:+MEDIUM:+LOW:+SSLv2:+EXP:+eNULL
|
|||
|
|
SSLProtocol All -SSLv2 -SSLv3 %s
|
|||
|
|
SSLHonorCipherOrder On
|
|||
|
|
""" % (site_name, site_name, self._get_tls_protocol(is_apache=True))
|
|||
|
|
|
|||
|
|
rep_list = [
|
|||
|
|
(re.compile(r"#DENY FILES"), True),
|
|||
|
|
(re.compile(r"CustomLog[^\n]*\n"), False),
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
# 使用正则匹配确定插入位置
|
|||
|
|
def set_by_rep_idx(tmp_rep: re.Pattern, use_start: bool) -> Optional[str]:
|
|||
|
|
tmp_res = tmp_rep.search(target_vhost)
|
|||
|
|
if not tmp_res:
|
|||
|
|
return None
|
|||
|
|
if use_start:
|
|||
|
|
new_conf = target_vhost[:tmp_res.start()] + ssl_conf + tmp_res.group() + target_vhost[tmp_res.end():]
|
|||
|
|
else:
|
|||
|
|
new_conf = target_vhost[:tmp_res.start()] + tmp_res.group() + ssl_conf + target_vhost[tmp_res.end():]
|
|||
|
|
return new_conf
|
|||
|
|
|
|||
|
|
ssl_vhost = None
|
|||
|
|
for r, s in rep_list:
|
|||
|
|
ssl_vhost = set_by_rep_idx(r, s)
|
|||
|
|
if ssl_vhost is not None:
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
if ssl_vhost is None:
|
|||
|
|
return "无法定位SSL配置文件位置，配置失败"
|
|||
|
|
|
|||
|
|
write_file(ap_file, ap_conf + "\n" + ssl_vhost)
|
|||
|
|
# 添加端口
|
|||
|
|
from .domain_tool import ApacheDomainTool
|
|||
|
|
ApacheDomainTool.apache_add_ports("443")
|
|||
|
|
web_server = webserver()
|
|||
|
|
if web_server == "apache" and check_server_config() is not None:
|
|||
|
|
write_file(ap_file, ap_conf)
|
|||
|
|
return "配置失败"
|
|||
|
|
|
|||
|
|
if web_server == "apache" and not mutil:
|
|||
|
|
service_reload()
|
|||
|
|
self.open_firewall_443()
|
|||
|
|
|
|||
|
|
def close_site_ssl_conf(self, site_name) -> Optional[str]:
|
|||
|
|
error_msg = self._close_ssl_conf_to_nginx(site_name)
|
|||
|
|
if error_msg is not None and webserver() == "nginx":
|
|||
|
|
return error_msg
|
|||
|
|
error_msg = self._close_ssl_conf_to_apache(site_name)
|
|||
|
|
if error_msg is not None and webserver() == "apache":
|
|||
|
|
return error_msg
|
|||
|
|
service_reload()
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _close_ssl_conf_to_nginx(self, site_name) -> Optional[str]:
|
|||
|
|
file = '{}/nginx/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
ng_conf = read_file(file)
|
|||
|
|
if not ng_conf:
|
|||
|
|
return "配置文件丢失，配置失败"
|
|||
|
|
rep_list = (
|
|||
|
|
re.compile(r"\n\s*#HTTP_TO_HTTPS_START(.|\n){1,300}#HTTP_TO_HTTPS_END"), # 关闭 强制https
|
|||
|
|
re.compile(r"\s*ssl_(certificate|certificate_key|protocols|"
|
|||
|
|
r"ciphers|prefer_server_ciphers|session_cache|session_timeout)[^;]*;"), # 关闭 强制https
|
|||
|
|
re.compile(r"\s*add_header\s+(Strict-Transport-Security|Alt-Svc)[^;]*;"), # 关闭 https 请求头配置
|
|||
|
|
re.compile(r"\s*error_page\s+497\s+[^;]*;"),
|
|||
|
|
re.compile(r"\s+listen\s+(\[::]:)?443.*;"), # 关闭端口监听
|
|||
|
|
)
|
|||
|
|
new_conf = ng_conf
|
|||
|
|
for rep in rep_list:
|
|||
|
|
new_conf = rep.sub("", new_conf)
|
|||
|
|
|
|||
|
|
write_file(file, new_conf)
|
|||
|
|
|
|||
|
|
def _close_ssl_conf_to_apache(self, site_name) -> Optional[str]:
|
|||
|
|
file = '{}/apache/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
ap_conf = read_file(file)
|
|||
|
|
if not ap_conf:
|
|||
|
|
return "配置文件丢失，配置失败"
|
|||
|
|
rep_list = (
|
|||
|
|
re.compile(r"\n\s*#HTTP_TO_HTTPS_START(.|\n){1,300}#HTTP_TO_HTTPS_END"),
|
|||
|
|
re.compile(r"\n<VirtualHost\s+\*:443\s*>(.|\n)*</VirtualHost>"),
|
|||
|
|
)
|
|||
|
|
new_conf = ap_conf
|
|||
|
|
for rep in rep_list:
|
|||
|
|
new_conf = rep.sub("", new_conf)
|
|||
|
|
|
|||
|
|
write_file(file, new_conf)
|
|||
|
|
|
|||
|
|
def _get_tls_protocol(self, is_apache=False):
|
|||
|
|
"""获取使用的协议
|
|||
|
|
@author baozi <202-04-18>
|
|||
|
|
@param:
|
|||
|
|
@return
|
|||
|
|
"""
|
|||
|
|
protocols = {
|
|||
|
|
"TLSv1": False,
|
|||
|
|
"TLSv1.1": True,
|
|||
|
|
"TLSv1.2": True,
|
|||
|
|
"TLSv1.3": False,
|
|||
|
|
}
|
|||
|
|
tls1_3 = self.get_tls13()
|
|||
|
|
file_path = public.get_panel_path() + "/data/ssl_protocol.json"
|
|||
|
|
if os.path.exists(file_path):
|
|||
|
|
data = public.readFile(file_path)
|
|||
|
|
if data is not False:
|
|||
|
|
protocols = json.loads(data)
|
|||
|
|
if protocols["TLSv1.3"] and tls1_3 == "":
|
|||
|
|
protocols["TLSv1.3"] = False
|
|||
|
|
if is_apache is False:
|
|||
|
|
return " ".join([p for p, v in protocols.items() if v is True])
|
|||
|
|
else:
|
|||
|
|
return " ".join(["-" + p for p, v in protocols.items() if v is False])
|
|||
|
|
else:
|
|||
|
|
if tls1_3 != "":
|
|||
|
|
protocols["TLSv1.3"] = True
|
|||
|
|
if is_apache is False:
|
|||
|
|
return " ".join([p for p, v in protocols.items() if v is True])
|
|||
|
|
else:
|
|||
|
|
return " ".join(["-" + p for p, v in protocols.items() if v is False])
|
|||
|
|
|
|||
|
|
# 获取TLS1.3标记
|
|||
|
|
def get_tls13(self):
|
|||
|
|
if self._tls_v3 is not None:
|
|||
|
|
return self._tls_v3
|
|||
|
|
nginx_bin = '/www/server/nginx/sbin/nginx'
|
|||
|
|
nginx_v = public.ExecShell(nginx_bin + ' -V 2>&1')[0]
|
|||
|
|
nginx_v_re = re.search(r"nginx/(?P<ng_ver>\d\.\d+).+OpenSSL\s+(?P<ssl_ver>\d\.\d+)", nginx_v)
|
|||
|
|
if nginx_v_re:
|
|||
|
|
ng_ver = nginx_v_re.group("ng_ver")
|
|||
|
|
ssl_ver = nginx_v_re.group("ssl_ver")
|
|||
|
|
if float(ng_ver) >= 1.15 and float(ssl_ver) >= 1.1:
|
|||
|
|
self._tls_v3 = 'TLSv1.3'
|
|||
|
|
else:
|
|||
|
|
can_ng_ver = re.search(r'nginx/1\.(1[5-9]|2\d)', nginx_v)
|
|||
|
|
openssl_v = public.ExecShell(nginx_bin + ' -V 2>&1|grep OpenSSL')[0].find('OpenSSL 1.1.') != -1
|
|||
|
|
if can_ng_ver and openssl_v:
|
|||
|
|
self._tls_v3 = 'TLSv1.3'
|
|||
|
|
|
|||
|
|
if self._tls_v3 is None:
|
|||
|
|
self._tls_v3 = ''
|
|||
|
|
return self._tls_v3
|
|||
|
|
|
|||
|
|
# HttpToHttps
|
|||
|
|
def set_http_to_https(self, site_name: str):
|
|||
|
|
# Nginx配置
|
|||
|
|
file = '{}/nginx/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
conf = read_file(file)
|
|||
|
|
if conf:
|
|||
|
|
if conf.find('ssl_certificate') == -1:
|
|||
|
|
return public.returnMsg(False, '当前未开启SSL')
|
|||
|
|
to_str = """#error_page 404/404.html;
|
|||
|
|
#HTTP_TO_HTTPS_START
|
|||
|
|
if ($server_port !~ 443){
|
|||
|
|
rewrite ^(/.*)$ https://$host$1 permanent;
|
|||
|
|
}
|
|||
|
|
#HTTP_TO_HTTPS_END
|
|||
|
|
"""
|
|||
|
|
conf = conf.replace('#error_page 404/404.html;', to_str)
|
|||
|
|
write_file(file, conf)
|
|||
|
|
|
|||
|
|
file = '{}/apache/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
conf = public.readFile(file)
|
|||
|
|
if conf:
|
|||
|
|
to_str = '''
|
|||
|
|
#HTTP_TO_HTTPS_START
|
|||
|
|
<IfModule mod_rewrite.c>
|
|||
|
|
RewriteEngine on
|
|||
|
|
RewriteCond %{SERVER_PORT} !^443$
|
|||
|
|
RewriteRule (.*) https://%{SERVER_NAME}$1 [L,R=301]
|
|||
|
|
</IfModule>
|
|||
|
|
#HTTP_TO_HTTPS_END
|
|||
|
|
SSLEngine On'''
|
|||
|
|
conf = re.sub('SSLEngine On', to_str, conf, 1)
|
|||
|
|
public.writeFile(file, conf)
|
|||
|
|
|
|||
|
|
service_reload()
|
|||
|
|
|
|||
|
|
# CloseToHttps
|
|||
|
|
def close_to_https(self, site_name):
|
|||
|
|
file = '{}/nginx/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
conf = public.readFile(file)
|
|||
|
|
if conf:
|
|||
|
|
rep_https = re.compile(r"(#HTTP_TO_HTTPS_START\s*)?if\s+\(\s*\$server_port\s+!~\s+443\s*\)"
|
|||
|
|
r"[^{]*\{[^}]*}\s*(#HTTP_TO_HTTPS_END\s*)?")
|
|||
|
|
new_conf = rep_https.sub('', conf)
|
|||
|
|
write_file(file, new_conf)
|
|||
|
|
|
|||
|
|
file = '{}/apache/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
conf = public.readFile(file)
|
|||
|
|
if conf:
|
|||
|
|
rep_https = re.compile(r"\n\s*#HTTP_TO_HTTPS_START(.|\n){1,300}#HTTP_TO_HTTPS_END")
|
|||
|
|
new_conf = rep_https.sub('', conf)
|
|||
|
|
write_file(file, new_conf)
|
|||
|
|
|
|||
|
|
service_reload()
|
|||
|
|
|
|||
|
|
# 是否有跳转到https
|
|||
|
|
def is_to_https(self, site_name) -> bool:
|
|||
|
|
file = '{}/nginx/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
conf = public.readFile(file)
|
|||
|
|
if conf:
|
|||
|
|
if conf.find('HTTP_TO_HTTPS_START') != -1:
|
|||
|
|
return True
|
|||
|
|
if conf.find('$server_port !~ 443') != -1:
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
def get_site_ssl_info(self, site_name: str) -> Optional[dict]:
|
|||
|
|
try:
|
|||
|
|
w_s = webserver()
|
|||
|
|
if w_s == 'nginx':
|
|||
|
|
conf_file = '{}/nginx/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
elif w_s == "apach":
|
|||
|
|
conf_file = '{}/apache/{}{}.conf'.format(self._vhost_path, self.conf_prefix, site_name)
|
|||
|
|
else:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
if not os.path.exists(conf_file):
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
s_conf = public.readFile(conf_file)
|
|||
|
|
if not s_conf:
|
|||
|
|
return None
|
|||
|
|
if w_s == "apach":
|
|||
|
|
s_tmp = re.findall(r"SSLCertificateFile\s+(.+\.pem)", s_conf)
|
|||
|
|
if not s_tmp:
|
|||
|
|
return None
|
|||
|
|
ssl_file = s_tmp[0]
|
|||
|
|
else:
|
|||
|
|
s_tmp = re.findall(r"ssl_certificate\s+(.+\.pem);", s_conf)
|
|||
|
|
if not s_tmp:
|
|||
|
|
return None
|
|||
|
|
ssl_file = s_tmp[0]
|
|||
|
|
|
|||
|
|
ssl_info = self.get_cert_info(cert_filename=ssl_file)
|
|||
|
|
if not ssl_info:
|
|||
|
|
return None
|
|||
|
|
ssl_info['endtime'] = int(
|
|||
|
|
int(time.mktime(time.strptime(ssl_info['notAfter'], "%Y-%m-%d")) - time.time()) / 86400)
|
|||
|
|
return ssl_info
|
|||
|
|
except:
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
class SSLManager:
|
|||
|
|
|
|||
|
|
def __init__(self, conf_prefix: str = ""):
|
|||
|
|
self.conf_prefix = conf_prefix
|
|||
|
|
|
|||
|
|
def set_site_ssl_conf(self, get):
|
|||
|
|
ssl_id = None
|
|||
|
|
ssl_hash = None
|
|||
|
|
try:
|
|||
|
|
if "ssl_id" in get:
|
|||
|
|
ssl_id = int(get.ssl_id)
|
|||
|
|
if "ssl_hash" in get:
|
|||
|
|
ssl_hash = get.ssl_hash.strip()
|
|||
|
|
site_name = get.site_name.strip()
|
|||
|
|
except (ValueError, AttributeError, KeyError):
|
|||
|
|
return public.ReturnMsg(False, "Parameter error")
|
|||
|
|
ssl_mgr = RealSSLManger(self.conf_prefix)
|
|||
|
|
try:
|
|||
|
|
info = ssl_mgr.find_ssl_info(ssl_id=ssl_id, ssl_hash=ssl_hash)
|
|||
|
|
if not info:
|
|||
|
|
return json_response(status=False, msg="未查询到证书信息")
|
|||
|
|
ssl_data = ssl_mgr.get_cert_for_deploy(info)
|
|||
|
|
if isinstance(ssl_data, str):
|
|||
|
|
return json_response(status=False, msg=ssl_data)
|
|||
|
|
except ValueError as e:
|
|||
|
|
return json_response(status=False, msg=str(e))
|
|||
|
|
|
|||
|
|
err_msg = ssl_mgr.set_site_ssl_conf(site_name=site_name, ssl_data=ssl_data)
|
|||
|
|
if err_msg:
|
|||
|
|
return json_response(status=False, msg=err_msg)
|
|||
|
|
return json_response(status=True, msg="部署成功")
|
|||
|
|
|
|||
|
|
def mutil_set_site_ssl_conf(self, get):
|
|||
|
|
ssl_id = None
|
|||
|
|
ssl_hash = None
|
|||
|
|
try:
|
|||
|
|
if "ssl_id" in get:
|
|||
|
|
ssl_id = int(get.ssl_id)
|
|||
|
|
if "ssl_hash" in get:
|
|||
|
|
ssl_hash = get.ssl_hash.strip()
|
|||
|
|
site_names = json.loads(get.site_names.strip())
|
|||
|
|
except (ValueError, AttributeError, KeyError, json.JSONDecodeError):
|
|||
|
|
return public.ReturnMsg(False, "Parameter error")
|
|||
|
|
ssl_mgr = RealSSLManger(self.conf_prefix)
|
|||
|
|
try:
|
|||
|
|
info = ssl_mgr.find_ssl_info(ssl_id=ssl_id, ssl_hash=ssl_hash)
|
|||
|
|
if not info:
|
|||
|
|
return json_response(status=False, msg="未查询到证书信息")
|
|||
|
|
ssl_data = ssl_mgr.get_cert_for_deploy(info)
|
|||
|
|
if isinstance(ssl_data, str):
|
|||
|
|
return json_response(status=False, msg=ssl_data)
|
|||
|
|
except ValueError as e:
|
|||
|
|
return json_response(status=False, msg=str(e))
|
|||
|
|
|
|||
|
|
result = {
|
|||
|
|
"total": len(site_names),
|
|||
|
|
"success": 0,
|
|||
|
|
"failed": 0,
|
|||
|
|
"success_list": [],
|
|||
|
|
"failed_list": [],
|
|||
|
|
"failed_msg": []
|
|||
|
|
}
|
|||
|
|
for i in site_names:
|
|||
|
|
err_msg = ssl_mgr.set_site_ssl_conf(site_name=i, ssl_data=ssl_data)
|
|||
|
|
if err_msg:
|
|||
|
|
result["failed"] += 1
|
|||
|
|
result["failed_list"].append(i)
|
|||
|
|
result["failed_msg"].append(err_msg)
|
|||
|
|
else:
|
|||
|
|
result["success"] += 1
|
|||
|
|
result["success_list"].append(i)
|
|||
|
|
|
|||
|
|
return json_response(status=True, data=result)
|
|||
|
|
|
|||
|
|
def close_site_ssl_conf(self, get):
|
|||
|
|
try:
|
|||
|
|
site_name = get.site_name.strip()
|
|||
|
|
except (ValueError, AttributeError, KeyError):
|
|||
|
|
return public.ReturnMsg(False, "Parameter error")
|
|||
|
|
|
|||
|
|
ssl_mgr = RealSSLManger(self.conf_prefix)
|
|||
|
|
try:
|
|||
|
|
err_msg = ssl_mgr.close_site_ssl_conf(site_name)
|
|||
|
|
if err_msg:
|
|||
|
|
return json_response(status=False, msg=err_msg)
|
|||
|
|
return json_response(status=True, msg="关闭成功")
|
|||
|
|
except Exception as e:
|
|||
|
|
return json_response(status=False, msg=str(e))
|
|||
|
|
|
|||
|
|
def upload_cert_to_cloud(self, get):
|
|||
|
|
ssl_id = None
|
|||
|
|
ssl_hash = None
|
|||
|
|
try:
|
|||
|
|
if "ssl_id" in get:
|
|||
|
|
ssl_id = int(get.ssl_id)
|
|||
|
|
if "ssl_hash" in get:
|
|||
|
|
ssl_hash = get.ssl_hash.strip()
|
|||
|
|
except (ValueError, AttributeError, KeyError):
|
|||
|
|
return public.ReturnMsg(False, "Parameter error")
|
|||
|
|
try:
|
|||
|
|
data = RealSSLManger(self.conf_prefix).upload_cert(ssl_id, ssl_hash)
|
|||
|
|
return json_response(status=True, data=data)
|
|||
|
|
except ValueError as e:
|
|||
|
|
return json_response(status=False, msg=str(e))
|
|||
|
|
except Exception as e:
|
|||
|
|
return json_response(status=False, msg="操作错误：" + str(e))
|
|||
|
|
|
|||
|
|
def remove_cloud_cert(self, get):
|
|||
|
|
ssl_id = None
|
|||
|
|
ssl_hash = None
|
|||
|
|
local = False
|
|||
|
|
try:
|
|||
|
|
if "ssl_id" in get:
|
|||
|
|
ssl_id = int(get.ssl_id)
|
|||
|
|
if "ssl_hash" in get:
|
|||
|
|
ssl_hash = get.ssl_hash.strip()
|
|||
|
|
|
|||
|
|
if "local" in get and get.local.strip() in ("1", 1, True, "true"):
|
|||
|
|
local = True
|
|||
|
|
|
|||
|
|
except (ValueError, AttributeError, KeyError):
|
|||
|
|
return public.ReturnMsg(False, "Parameter error")
|
|||
|
|
try:
|
|||
|
|
data = RealSSLManger(self.conf_prefix).remove_cert(ssl_id, ssl_hash, local=local)
|
|||
|
|
return json_response(status=data.get("status", True), msg=data.get("msg", ""), data=data)
|
|||
|
|
except ValueError as e:
|
|||
|
|
return json_response(status=False, msg=str(e))
|
|||
|
|
except Exception as e:
|
|||
|
|
return json_response(status=False, msg="操作错误：" + str(e))
|
|||
|
|
|
|||
|
|
def mutil_remove_cloud_cert(self, get):
|
|||
|
|
local = False
|
|||
|
|
try:
|
|||
|
|
ssl_id_list = json.loads(get.ssl_id_list.strip())
|
|||
|
|
if "local" in get and get.local.strip() in ("1", 1, True, "true"):
|
|||
|
|
local = True
|
|||
|
|
|
|||
|
|
except (ValueError, AttributeError, KeyError):
|
|||
|
|
return public.ReturnMsg(False, "Parameter error")
|
|||
|
|
try:
|
|||
|
|
data = RealSSLManger(self.conf_prefix).mutil_remove_cert(ssl_id_list, local=local)
|
|||
|
|
return json_response(status=True, data=data)
|
|||
|
|
except ValueError as e:
|
|||
|
|
return json_response(status=False, msg=str(e))
|
|||
|
|
except Exception as e:
|
|||
|
|
return json_response(status=False, msg="操作错误：" + str(e))
|
|||
|
|
|
|||
|
|
# 未使用
|
|||
|
|
def refresh_cert_list(self, get=None):
|
|||
|
|
try:
|
|||
|
|
data = RealSSLManger(self.conf_prefix).get_cert_list(force_refresh=True)
|
|||
|
|
return json_response(status=True, data=data)
|
|||
|
|
except ValueError as e:
|
|||
|
|
return json_response(status=False, msg=str(e))
|
|||
|
|
except Exception as e:
|
|||
|
|
return json_response(status=False, msg="操作错误：" + str(e))
|
|||
|
|
|
|||
|
|
def get_cert_info(self, get):
|
|||
|
|
ssl_id = None
|
|||
|
|
ssl_hash = None
|
|||
|
|
try:
|
|||
|
|
if "ssl_id" in get:
|
|||
|
|
ssl_id = int(get.ssl_id)
|
|||
|
|
if "ssl_hash" in get:
|
|||
|
|
ssl_hash = get.ssl_hash.strip()
|
|||
|
|
except (ValueError, AttributeError, KeyError):
|
|||
|
|
return public.ReturnMsg(False, "Parameter error")
|
|||
|
|
try:
|
|||
|
|
ssl_mager = RealSSLManger(self.conf_prefix)
|
|||
|
|
target = ssl_mager.find_ssl_info(ssl_id, ssl_hash)
|
|||
|
|
if target is None:
|
|||
|
|
return json_response(status=False, msg="未获取到证书信息")
|
|||
|
|
data = ssl_mager.get_cert_for_deploy(target)
|
|||
|
|
if isinstance(data, dict):
|
|||
|
|
target.update(data)
|
|||
|
|
return json_response(status=True, data=target)
|
|||
|
|
else:
|
|||
|
|
return json_response(status=False, msg=data)
|
|||
|
|
except ValueError as e:
|
|||
|
|
return json_response(status=False, msg=str(e))
|
|||
|
|
except Exception as e:
|
|||
|
|
return json_response(status=False, msg="操作错误：" + str(e))
|
|||
|
|
|
|||
|
|
def get_site_ssl_info(self, get):
|
|||
|
|
try:
|
|||
|
|
site_name = get.site_name.strip()
|
|||
|
|
except (ValueError, AttributeError, KeyError):
|
|||
|
|
return json_response(False, "Parameter error")
|
|||
|
|
ssl_info = RealSSLManger(self.conf_prefix).get_site_ssl_info(site_name)
|
|||
|
|
if ssl_info is None:
|
|||
|
|
return json_response(status=False, msg="未获取到证书信息")
|
|||
|
|
else:
|
|||
|
|
return json_response(status=True, data=ssl_info)
|
|||
|
|
|
|||
|
|
def get_cert_list(self, get):
|
|||
|
|
"""
|
|||
|
|
search_limit 0 -> 所有证书
|
|||
|
|
search_limit 1 -> 没有过期的证书
|
|||
|
|
search_limit 2 -> 有效期小于等于15天的证书 但未过期
|
|||
|
|
search_limit 3 -> 过期的证书
|
|||
|
|
search_limit 4 -> 过期时间1年以上的证书
|
|||
|
|
"""
|
|||
|
|
search_name = None
|
|||
|
|
search_limit = 0
|
|||
|
|
force_refresh = False
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
if "search_name" in get:
|
|||
|
|
search_name = get.search_name.strip()
|
|||
|
|
if "search_limit" in get:
|
|||
|
|
search_limit = int(get.search_limit.strip())
|
|||
|
|
if "force_refresh" in get and get.force_refresh.strip() in ("1", 1, "True", True):
|
|||
|
|
force_refresh = True
|
|||
|
|
|
|||
|
|
except (ValueError, AttributeError, KeyError):
|
|||
|
|
return json_response(status=False, msg="Parameter error")
|
|||
|
|
|
|||
|
|
param = None
|
|||
|
|
if search_name is not None:
|
|||
|
|
param = ['subject LIKE ?', ["%{}%".format(search_name)]]
|
|||
|
|
|
|||
|
|
now = datetime.now()
|
|||
|
|
filter_func: Callable[[dict, ], bool] = lambda x: True
|
|||
|
|
if search_limit == 1:
|
|||
|
|
date = now.strftime("%Y-%m-%d")
|
|||
|
|
filter_func: Callable[[dict, ], bool] = lambda x: x["not_after"] >= date
|
|||
|
|
elif search_limit == 2:
|
|||
|
|
date1 = now.strftime("%Y-%m-%d")
|
|||
|
|
date2 = (now + timedelta(days=15)).strftime("%Y-%m-%d")
|
|||
|
|
filter_func: Callable[[dict, ], bool] = lambda x: date1 <= x["not_after"] <= date2
|
|||
|
|
elif search_limit == 3:
|
|||
|
|
date = now.strftime("%Y-%m-%d")
|
|||
|
|
filter_func: Callable[[dict, ], bool] = lambda x: x["not_after"] < date
|
|||
|
|
elif search_limit == 4:
|
|||
|
|
date = (now + timedelta(days=366)).strftime("%Y-%m-%d")
|
|||
|
|
filter_func: Callable[[dict, ], bool] = lambda x: x["not_after"] > date
|
|||
|
|
try:
|
|||
|
|
res_list = RealSSLManger(self.conf_prefix).get_cert_list(param=param, force_refresh=force_refresh)
|
|||
|
|
res_list = list(filter(filter_func, res_list))
|
|||
|
|
res_list.sort(key=lambda x: x["not_after"])
|
|||
|
|
return json_response(status=True, data=res_list)
|
|||
|
|
except ValueError as e:
|
|||
|
|
return json_response(False, str(e))
|
|||
|
|
except Exception as e:
|
|||
|
|
return json_response(False, "操作错误：" + str(e))
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def set_ssl_protocol(get):
|
|||
|
|
""" 设置全局TLS版本
|
|||
|
|
@author baozi <202-04-18>
|
|||
|
|
@param:
|
|||
|
|
@return
|
|||
|
|
"""
|
|||
|
|
protocols = {
|
|||
|
|
"TLSv1": False,
|
|||
|
|
"TLSv1.1": False,
|
|||
|
|
"TLSv1.2": False,
|
|||
|
|
"TLSv1.3": False,
|
|||
|
|
}
|
|||
|
|
if "use_protocols" in get:
|
|||
|
|
use_protocols = getattr(get, "use_protocols", [])
|
|||
|
|
if isinstance(use_protocols, list):
|
|||
|
|
for protocol in use_protocols:
|
|||
|
|
if protocol in protocols:
|
|||
|
|
protocols[protocol] = True
|
|||
|
|
elif isinstance(use_protocols, str):
|
|||
|
|
for protocol in use_protocols.split(","):
|
|||
|
|
if protocol in protocols:
|
|||
|
|
protocols[protocol] = True
|
|||
|
|
else:
|
|||
|
|
protocols["TLSv1.1"] = True
|
|||
|
|
protocols["TLSv1.2"] = True
|
|||
|
|
protocols["TLSv1.3"] = True
|
|||
|
|
|
|||
|
|
else:
|
|||
|
|
protocols["TLSv1.1"] = True
|
|||
|
|
protocols["TLSv1.2"] = True
|
|||
|
|
protocols["TLSv1.3"] = True
|
|||
|
|
|
|||
|
|
public.print_log(protocols)
|
|||
|
|
public.WriteFile(public.get_panel_path() + "/data/ssl_protocol.json", json.dumps(protocols))
|
|||
|
|
return public.returnMsg(True, 'Successfully set')
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def get_ssl_protocol(get=None):
|
|||
|
|
""" 获取全局TLS版本
|
|||
|
|
@author baozi <202-04-18>
|
|||
|
|
@param:
|
|||
|
|
@return
|
|||
|
|
"""
|
|||
|
|
protocols = {
|
|||
|
|
"TLSv1": False,
|
|||
|
|
"TLSv1.1": True,
|
|||
|
|
"TLSv1.2": True,
|
|||
|
|
"TLSv1.3": False,
|
|||
|
|
}
|
|||
|
|
file_path = public.get_panel_path() + "/data/ssl_protocol.json"
|
|||
|
|
if os.path.exists(file_path):
|
|||
|
|
data = public.readFile(file_path)
|
|||
|
|
if data is not False:
|
|||
|
|
protocols = json.loads(data)
|
|||
|
|
return protocols
|
|||
|
|
|
|||
|
|
return protocols
|