330 lines
13 KiB
Python
330 lines
13 KiB
Python
import os,sys,re,time,subprocess
|
||
|
||
panelPath = '/www/server/panel/'
|
||
os.chdir(panelPath)
|
||
|
||
sys.path.insert(0, panelPath + "class/")
|
||
import public
|
||
from datetime import date, datetime
|
||
|
||
is_openssl = True
|
||
try:
|
||
import OpenSSL
|
||
except:
|
||
is_openssl = False
|
||
|
||
class ssl_info:
|
||
|
||
def __init__(self) -> None:
|
||
pass
|
||
|
||
|
||
def create_key(self,bits=2048):
|
||
"""
|
||
@name 创建RSA密钥
|
||
@param bits 密钥长度
|
||
"""
|
||
if is_openssl:
|
||
|
||
key = OpenSSL.crypto.PKey()
|
||
key.generate_key(OpenSSL.crypto.TYPE_RSA, bits)
|
||
private_key = OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM, key)
|
||
return private_key
|
||
else:
|
||
tmp_pk_file = "/tmp/private_key_{}.pem".format(int(time.time()))
|
||
cmd = ["openssl", "genpkey", "-algorithm", "RSA", "-out", tmp_pk_file, "-pkeyopt", f"rsa_keygen_bits:{bits}"]
|
||
subprocess.run(cmd, check=True)
|
||
with open(tmp_pk_file, "r") as f:
|
||
private_key = f.read()
|
||
try:
|
||
os.remove("private_key.pem")
|
||
except:
|
||
pass
|
||
return private_key
|
||
|
||
def load_ssl_info_by_data(self, pem_data: str):
|
||
if not isinstance(pem_data, (str, bytes)):
|
||
return None
|
||
|
||
if is_openssl:
|
||
return self.__get_cert_info(pem_data)
|
||
else:
|
||
result = {}
|
||
pem_file = "/tmp/fullchain_{}.pem".format(int(time.time()))
|
||
public.writeFile(pem_file, pem_data)
|
||
res = public.ExecShell("openssl x509 -in {} -noout -text".format(pem_file))[0]
|
||
issuer_match = re.search(r"Issuer: (.*)", res)
|
||
if issuer_match:
|
||
data = {}
|
||
issuer = issuer_match.group(1)
|
||
for key,val in re.findall(r"(\w+)\s*=([^,]+)", issuer):
|
||
data[key] = val.strip()
|
||
|
||
result["issuer"] = data['CN']
|
||
if "CN" in data:
|
||
s = data['CN'].encode().decode('unicode_escape')
|
||
result["issuer"] = bytes(s, 'latin1').decode('utf-8')
|
||
result["issuer_O"] = data['O']
|
||
if "O" in data:
|
||
s = data['O'].encode().decode('unicode_escape')
|
||
result["issuer_O"] = bytes(s, 'latin1').decode('utf-8')
|
||
|
||
validity_match = re.search(r"Not After\s*:\s*(.*)", res)
|
||
if validity_match:
|
||
not_after = validity_match.group(1)
|
||
dt_after = datetime.strptime(not_after, "%b %d %H:%M:%S %Y %Z")
|
||
result['notAfter'] = dt_after.strftime("%Y-%m-%d %H:%M:%S")
|
||
result['endtime'] = (dt_after - datetime.now()).days
|
||
else:
|
||
result['endtime'] = 0
|
||
|
||
validity_match = re.search(r"Not Before\s*:\s*(.*)", res)
|
||
if validity_match:
|
||
not_befoer = validity_match.group(1)
|
||
dt_befoer = datetime.strptime(not_befoer, "%b %d %H:%M:%S %Y %Z")
|
||
result['notBefore'] = dt_befoer.strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
subject_match = re.search(r"Subject: (.*)", res)
|
||
if subject_match:
|
||
subject = subject_match.group(1)
|
||
for key,val in re.findall(r"(\w+)\s*=([^,]+)", subject):
|
||
if key == 'CN':
|
||
s = val.encode().decode('unicode_escape')
|
||
result["subject"] = bytes(s, 'latin1').decode('utf-8').strip()
|
||
# 取可选名称
|
||
result['dns'] = []
|
||
dns_match = re.findall(r"DNS:([^\s,]+)", res)
|
||
for dns in dns_match:
|
||
result['dns'].append(dns)
|
||
if os.path.exists(pem_file):
|
||
os.remove(pem_file)
|
||
return result
|
||
|
||
def load_ssl_info(self,pem_file):
|
||
"""
|
||
@name 获取证书详情
|
||
"""
|
||
if not os.path.exists(pem_file):
|
||
return None
|
||
|
||
pem_data = public.readFile(pem_file)
|
||
if not pem_data:
|
||
return None
|
||
return self.load_ssl_info_by_data(pem_data)
|
||
|
||
def __get_cert_info(self,pem_data):
|
||
"""
|
||
@name 通过python的openssl模块获取证书信息
|
||
@param pem_data 证书内容
|
||
"""
|
||
result = {}
|
||
try:
|
||
x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, pem_data)
|
||
except: # 证书格式可能是错的，就没有办法读取证书内容
|
||
return None
|
||
|
||
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 hasattr(issuer, 'O'):
|
||
result['issuer_O'] = issuer.O
|
||
# 取到期时间
|
||
result['notAfter'] = self.strf_date(
|
||
bytes.decode(x509.get_notAfter())[:-1])
|
||
# 取申请时间
|
||
result['notBefore'] = self.strf_date(
|
||
bytes.decode(x509.get_notBefore())[:-1])
|
||
# 取可选名称
|
||
result['dns'] = []
|
||
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]
|
||
result['endtime'] = int(int(time.mktime(time.strptime(result['notAfter'], "%Y-%m-%d")) - time.time()) / 86400)
|
||
return result
|
||
|
||
# 转换时间
|
||
def strf_date(self, sdate):
|
||
return time.strftime('%Y-%m-%d', time.strptime(sdate, '%Y%m%d%H%M%S'))
|
||
|
||
#转换时间
|
||
def strfToTime(self,sdate):
|
||
import time
|
||
return time.strftime('%Y-%m-%d',time.strptime(sdate,'%b %d %H:%M:%S %Y %Z'))
|
||
|
||
|
||
def dump_pkcs12_new(self, key_pem=None, cert_pem=None, ca_pem=None, friendly_name=""):
|
||
try:
|
||
from cryptography.hazmat.backends import default_backend
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.x509 import load_pem_x509_certificate
|
||
private_key = self.analysis_private_key(key_pem)
|
||
cert = self.analysis_certificate(cert_pem)
|
||
if not private_key or not cert:
|
||
return None
|
||
# 加载CA证书
|
||
cas = None
|
||
if ca_pem:
|
||
cas = [load_pem_x509_certificate(ca_pem.encode(), default_backend())]
|
||
# 将证书和私钥组合成PKCS12格式的文件
|
||
from cryptography.hazmat.primitives.serialization.pkcs12 import serialize_key_and_certificates
|
||
p12 = serialize_key_and_certificates(
|
||
name=friendly_name.encode() if friendly_name else None,
|
||
key=private_key,
|
||
cert=cert,
|
||
encryption_algorithm=serialization.NoEncryption(),
|
||
cas=cas
|
||
)
|
||
return p12
|
||
except:
|
||
return None
|
||
|
||
def analysis_private_key(self, key_pem, password=None):
|
||
"""
|
||
解析私钥
|
||
:param key_pem: 私钥内容
|
||
:param password: 私钥密码
|
||
:return: 私钥对象
|
||
"""
|
||
try:
|
||
from cryptography.hazmat.backends import default_backend
|
||
from cryptography.hazmat.primitives import serialization
|
||
private_key = serialization.load_pem_private_key(
|
||
key_pem.encode(),
|
||
password=password,
|
||
backend=default_backend()
|
||
)
|
||
return private_key
|
||
except:
|
||
return None
|
||
|
||
def analysis_certificate(self, cert_pem):
|
||
"""
|
||
解析证书
|
||
:param cert_pem: 证书内容
|
||
:return: 有效返回True，无效返回False
|
||
"""
|
||
try:
|
||
from cryptography.hazmat.backends import default_backend
|
||
from cryptography.x509 import load_pem_x509_certificate
|
||
cert = load_pem_x509_certificate(cert_pem.encode(), default_backend())
|
||
return cert
|
||
except Exception as e:
|
||
public.print_log(public.get_error_info())
|
||
return None
|
||
|
||
def verify_certificate_and_key_match(self, key_pem, cert_pem, password=None):
|
||
"""
|
||
验证证书和私钥是否匹配(通过签名验证)
|
||
:param key_pem: 私钥内容
|
||
:param cert_pem: 证书内容
|
||
:param password: 私钥密码
|
||
:return: 验证成功返回True，失败返回False
|
||
"""
|
||
try:
|
||
from cryptography.hazmat.primitives import hashes
|
||
from cryptography.hazmat.primitives.asymmetric import padding, ec
|
||
message = b"test message"
|
||
private_key = self.analysis_private_key(key_pem, password)
|
||
cert = self.analysis_certificate(cert_pem)
|
||
if not private_key:
|
||
return False, "Key error, please check if it is the correct PEM format private key"
|
||
if not cert:
|
||
return False, "Certificate error, please check if it is the correct PEM format certificate"
|
||
# 使用私钥对消息进行签名
|
||
try:
|
||
signature = private_key.sign(
|
||
message,
|
||
padding.PKCS1v15(),
|
||
hashes.SHA256()
|
||
)
|
||
except Exception as e:
|
||
return True, str(e)
|
||
# 使用证书中的公钥验证签名
|
||
public_key = cert.public_key()
|
||
try:
|
||
if isinstance(public_key, ec.EllipticCurvePublicKey):
|
||
public_key.verify(
|
||
signature,
|
||
message,
|
||
ec.ECDSA(hashes.SHA256())
|
||
)
|
||
else:
|
||
public_key.verify(
|
||
signature,
|
||
message,
|
||
padding.PKCS1v15(),
|
||
hashes.SHA256()
|
||
)
|
||
return True, "Verification successful"
|
||
except Exception as e:
|
||
return False, "The key and certificate do not match."
|
||
except Exception as e:
|
||
public.print_log(public.get_error_info())
|
||
return True, str(e)
|
||
|
||
def verify_certificate_chain(self, cert_pem):
|
||
"""
|
||
验证证书链是否完整
|
||
:param cert_pem: 证书链内容
|
||
:return: 验证成功返回True，失败返回False
|
||
"""
|
||
try:
|
||
from cryptography.hazmat.primitives.asymmetric import padding, ec
|
||
cert_chain = [(i + "-----END CERTIFICATE-----").strip() for i in
|
||
cert_pem.strip().split("-----END CERTIFICATE-----") if i]
|
||
for i in range(len(cert_chain) - 1):
|
||
cert = self.analysis_certificate(cert_chain[i])
|
||
issuer_cert = self.analysis_certificate(cert_chain[i + 1])
|
||
# 验证当前证书的签名是否由上一级签发
|
||
try:
|
||
issuer_public_key = issuer_cert.public_key()
|
||
# 检查 issuer_public_key 是否为 ECC 公钥或 RSA 公钥
|
||
if isinstance(issuer_public_key, ec.EllipticCurvePublicKey):
|
||
# 使用 ECDSA 验证 ECC 证书签名
|
||
issuer_public_key.verify(
|
||
cert.signature,
|
||
cert.tbs_certificate_bytes,
|
||
ec.ECDSA(cert.signature_hash_algorithm),
|
||
)
|
||
else:
|
||
# 使用 PKCS1v15 验证 RSA 证书签名
|
||
issuer_public_key.verify(
|
||
cert.signature,
|
||
cert.tbs_certificate_bytes,
|
||
padding.PKCS1v15(),
|
||
cert.signature_hash_algorithm,
|
||
)
|
||
except Exception as e:
|
||
return False, "Certificate chain verification failed, please check if the certificate chain is complete."
|
||
return True, "Certificate chain verification successful."
|
||
except Exception as e:
|
||
public.print_log(public.get_error_info())
|
||
return True, str(e)
|