mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-17 21:25:47 +02:00
Jack
a24f2e23c6
1. Added Mail Marketing Automation trigger tasks 2. Added Mail Marketing Groups Import, Export, Merge 3. Added Mail Marketing Subscribers to support paste import 4. Added Mail Marketing Template Import Export Duplicate 5. Added Mail server add timed automatic reply 6. Added HTTPS Protection function for Website (disables automatic HTTP to HTTPS redirection when enabled) 7. Optimize Website Interface Button Integration 8. Optimize the response speed of WP Toolkit interface
241 lines
10 KiB
Python
241 lines
10 KiB
Python
#coding: utf-8
|
|
# +-------------------------------------------------------------------
|
|
# | aaPanel
|
|
# +-------------------------------------------------------------------
|
|
# | Copyright (c) 2015-2099 aaPanel(www.aapanel.com) All rights reserved.
|
|
# +-------------------------------------------------------------------
|
|
# | Author: zhwen <zhw@aapanel.com>
|
|
# +-------------------------------------------------------------------
|
|
import os
|
|
os.chdir("/www/server/panel")
|
|
import public,db,json
|
|
|
|
|
|
class setPanelLets:
|
|
__vhost_cert_path = "/www/server/panel/vhost/cert/"
|
|
__panel_cert_path = "/www/server/panel/ssl/"
|
|
__tmp_key = ""
|
|
__tmp_cert = ""
|
|
def __init__(self):
|
|
pass
|
|
|
|
# 保存面板证书
|
|
def __save_panel_cert(self,cert,key):
|
|
keyPath = 'ssl/privateKey.pem'
|
|
certPath = 'ssl/certificate.pem'
|
|
checkCert = '/tmp/cert.pl'
|
|
public.writeFile(checkCert,cert)
|
|
if key:
|
|
public.writeFile(keyPath,key)
|
|
if cert:
|
|
public.writeFile(certPath,cert)
|
|
if not public.CheckCert(checkCert): return public.returnMsg(False, public.lang("Certificate error, please check!"))
|
|
public.writeFile('ssl/input.pl','True')
|
|
return public.returnMsg(True, public.lang("The certificate has been saved!"))
|
|
|
|
# 检查是否存在站点aapanel主机名站点
|
|
def __check_host_name(self, domain):
|
|
sql = db.Sql()
|
|
path = sql.table('sites').where('name=?', (domain,)).getField('path')
|
|
return path
|
|
|
|
# 创建证书使用的站点
|
|
def __create_site_of_panel_lets(self,get):
|
|
import panelSite
|
|
ps = panelSite.panelSite()
|
|
get.webname = json.dumps({"domain":get.domain,"domainlist":[],"count":0})
|
|
get.ps = "For panel Lets Encrypt certificate request and renewal, please do not delete"
|
|
get.path = "/www/wwwroot/panel_ssl_site"
|
|
get.ftp = "false"
|
|
get.sql = "false"
|
|
get.codeing = "utf8"
|
|
get.type = "PHP"
|
|
get.version = "00"
|
|
get.type_id = "0"
|
|
get.port = "80"
|
|
psa = ps.AddSite(get)
|
|
if "status" in psa.keys():
|
|
return psa
|
|
|
|
# 申请面板域名证书
|
|
def __create_lets(self,get):
|
|
from acme_v2 import acme_v2
|
|
site_id = str(public.M('sites').where('name=?',(get.domain,)).getField('id'))
|
|
get.auth_type = 'http'
|
|
get.auth_to = site_id
|
|
get.id = site_id
|
|
get.auto_wildcard = '0'
|
|
get.domains = json.dumps([get.domain])
|
|
get.siteName = get.domain
|
|
p = acme_v2()
|
|
cert_info = p.apply_cert_api(get)
|
|
if 'private_key' not in cert_info:
|
|
return public.returnMsg(False, public.lang("Failed to apply for a certificate, please try to manually apply for a certificate for the panel domain name on the site management page"))
|
|
get.key = cert_info['private_key']
|
|
get.csr = cert_info['cert'] + cert_info['root']
|
|
return public.returnMsg(True, self._deploy_cert(get))
|
|
|
|
# 部署证书
|
|
def _deploy_cert(self,get):
|
|
from panelSite import panelSite
|
|
return panelSite().SetSSL(get)
|
|
|
|
# 检查证书夹是否存在可用证书
|
|
def __check_cert_dir(self,get):
|
|
import panelSSL,time
|
|
pssl = panelSSL.panelSSL()
|
|
gcl = pssl.GetCertList(get)
|
|
for i in gcl:
|
|
if get.domain in i['dns'] or get.domain == i['subject']:
|
|
try:
|
|
time_stamp = int(i['notAfter'])
|
|
except:
|
|
time_array = time.strptime(i['notAfter'],"%Y-%m-%d")
|
|
time_stamp = int(time.mktime(time_array))
|
|
now = time.time()
|
|
if time_stamp > int(now):
|
|
return i
|
|
for d in i['dns']:
|
|
d = d.split('.')
|
|
if '*' in d and d[1:] == get.domain.split('.')[1:]:
|
|
try:
|
|
time_stamp = int(i['notAfter'])
|
|
except:
|
|
time_array = time.strptime(i['notAfter'], "%Y-%m-%d")
|
|
time_stamp = int(time.mktime(time_array))
|
|
now = time.time()
|
|
if time_stamp > int(now):
|
|
return i
|
|
|
|
# 读取可用站点证书
|
|
def __read_site_cert(self,domain_cert):
|
|
try:
|
|
key_file = "{path}{domain}/{key}".format(path=self.__vhost_cert_path,domain=domain_cert["subject"],key="privkey.pem")
|
|
cert_file = "{path}{domain}/{cert}".format(path=self.__vhost_cert_path, domain=domain_cert["subject"],
|
|
cert="fullchain.pem")
|
|
except:
|
|
key_file = "/www/server/panel/{}/privkey.pem".format(domain_cert['save_path'])
|
|
cert_file = "/www/server/panel/{}/fullchain.pem".format(domain_cert['save_path'])
|
|
if not os.path.exists(key_file):
|
|
key_file = "{path}{domain}/{key}".format(path="/www/server/panel/vhost/ssl/",domain=domain_cert["subject"],key="privkey.pem")
|
|
cert_file = "{path}{domain}/{cert}".format(path="/www/server/panel/vhost/ssl/", domain=domain_cert["subject"],
|
|
cert="fullchain.pem")
|
|
if not os.path.exists(key_file) and '*.' in key_file:
|
|
key_file = key_file.replace('*.','')
|
|
cert_file = cert_file.replace('*.','')
|
|
if not os.path.exists(key_file):
|
|
return public.returnMsg(False, public.lang("Can not found the ssl file! {}", key_file))
|
|
self.__tmp_key = public.readFile(key_file)
|
|
self.__tmp_cert = public.readFile(cert_file)
|
|
|
|
# 检查面板证书是否存在
|
|
def __check_panel_cert(self):
|
|
key = public.readFile(self.__panel_cert_path+"privateKey.pem")
|
|
cert = public.readFile(self.__panel_cert_path+"certificate.pem")
|
|
if key and cert:
|
|
return {"key":key,"cert":cert}
|
|
|
|
# 写面板证书
|
|
def __write_panel_cert(self):
|
|
public.writeFile(self.__panel_cert_path + "privateKey.pem", self.__tmp_key)
|
|
public.writeFile(self.__panel_cert_path + "certificate.pem", self.__tmp_cert)
|
|
|
|
# 记录证书源
|
|
def __save_cert_source(self,domain,email):
|
|
public.writeFile(self.__panel_cert_path+"lets.info",json.dumps({"domain":domain,"cert_type":"2","email":email}))
|
|
|
|
# 获取证书源
|
|
def get_cert_source(self):
|
|
data = public.readFile(self.__panel_cert_path+"lets.info")
|
|
if not data:
|
|
return {"cert_type":"","email":"","domain":""}
|
|
return json.loads(data)
|
|
|
|
# 检查面板是否绑定域名
|
|
def __check_panel_domain(self):
|
|
domain = public.readFile("/www/server/panel/data/domain.conf")
|
|
if not domain:
|
|
return False
|
|
return domain.split('\n')[0]
|
|
|
|
# 任务调用，检查面板是否有证书更新
|
|
def check_cert_update(self,sitename):
|
|
# 构造对象
|
|
from collections import namedtuple
|
|
get = namedtuple("get",['siteName','domain'])
|
|
get.siteName = sitename
|
|
get.domain = sitename
|
|
# 证书文件夹是否存在
|
|
cert_info = self.__check_cert_dir(get)
|
|
if cert_info:
|
|
return self.copy_cert(cert_info)
|
|
# 如果没有检查cert文件夹
|
|
else:
|
|
from panelSite import panelSite
|
|
p_s = panelSite().GetSSL(get)
|
|
if 'msg' in p_s:
|
|
return False
|
|
if not p_s['cert_data']:
|
|
return False
|
|
for i in p_s['cert_data']['dns']:
|
|
if i == sitename:
|
|
cert_info = {'issuer':p_s['cert_data']['issuer'],
|
|
'dns':p_s['cert_data']['dns'],
|
|
'notAfter':p_s['cert_data']['notAfter'],
|
|
'notBefore':p_s['cert_data']['notBefore'],
|
|
'subject':p_s['cert_data']['subject']}
|
|
return self.copy_cert(cert_info)
|
|
|
|
# 复制证书
|
|
def copy_cert(self,domain_cert):
|
|
res = self.__read_site_cert(domain_cert)
|
|
if res:
|
|
return res
|
|
panel_cert_data = self.__check_panel_cert()
|
|
if not panel_cert_data:
|
|
self.__write_panel_cert()
|
|
return public.returnMsg(True, public.lang("1"))
|
|
if panel_cert_data["key"] != self.__tmp_key and panel_cert_data["cert"] != self.__tmp_cert:
|
|
self.__write_panel_cert()
|
|
return public.returnMsg(True, public.lang("1"))
|
|
return public.returnMsg(True, public.lang(""))
|
|
|
|
# 设置lets证书
|
|
def set_lets(self,get):
|
|
"""
|
|
传入参数
|
|
get.domain 面板域名
|
|
get.email 管理员email
|
|
"""
|
|
create_site = ""
|
|
domain = self.__check_panel_domain()
|
|
get.domain = domain
|
|
if not domain:
|
|
return public.returnMsg(False, public.lang("You need to bind the domain name to the panel before you can apply for the Lets Encrypt certificate."))
|
|
if not self.__check_host_name(domain):
|
|
create_site = self.__create_site_of_panel_lets(get)
|
|
domain_cert = self.__check_cert_dir(get)
|
|
if domain_cert:
|
|
res = self.copy_cert(domain_cert)
|
|
if not res['status']:
|
|
return res
|
|
public.writeFile("/www/server/panel/data/ssl.pl", "True")
|
|
# public.writeFile("/www/server/panel/data/reload.pl","1")
|
|
self.__save_cert_source(domain,get.email)
|
|
return public.returnMsg(True, public.lang("Setup successfully!"))
|
|
if not create_site:
|
|
create_lets = self.__create_lets(get)
|
|
if not create_lets['status']:
|
|
return create_lets
|
|
if create_lets['msg']:
|
|
domain_cert = self.__check_cert_dir(get)
|
|
self.copy_cert(domain_cert)
|
|
public.writeFile("/www/server/panel/data/ssl.pl", "True")
|
|
# public.writeFile("/www/server/panel/data/reload.pl", "1")
|
|
self.__save_cert_source(domain, get.email)
|
|
return public.returnMsg(True, 'Setup successfully!')
|
|
else:
|
|
return public.returnMsg(False, create_lets)
|
|
else:
|
|
return public.returnMsg(False, create_site)
|