zfsafe/tools/OpensslTool.py

from tools.ToolBase import ToolBase
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.x509 import NameOID
import re
import json

class OpensslTool(ToolBase):
    def validate_instruction(self, instruction):
        #指令过滤
        timeout = 0
        return instruction,timeout

    def parse_name(self,name):
        """解析X509名称对象为结构化字典"""
        return {
            NameOID.COUNTRY_NAME: name.get_attributes_for_oid(NameOID.COUNTRY_NAME),
            NameOID.STATE_OR_PROVINCE_NAME: name.get_attributes_for_oid(NameOID.STATE_OR_PROVINCE_NAME),
            NameOID.LOCALITY_NAME: name.get_attributes_for_oid(NameOID.LOCALITY_NAME),
            NameOID.ORGANIZATION_NAME: name.get_attributes_for_oid(NameOID.ORGANIZATION_NAME),
            NameOID.COMMON_NAME: name.get_attributes_for_oid(NameOID.COMMON_NAME),
            NameOID.ORGANIZATIONAL_UNIT_NAME: name.get_attributes_for_oid(NameOID.ORGANIZATIONAL_UNIT_NAME),
        }

    def parse_ssl_info(self,output):
        # 提取证书内容
        certs = re.findall(
            r'-----BEGIN CERTIFICATE-----(.*?)-----END CERTIFICATE-----',
            output,
            re.DOTALL
        )

        results = []
        cert_obj = None
        for cert in certs:
            cert_data = "-----BEGIN CERTIFICATE-----" + cert + "-----END CERTIFICATE-----"
            try:
                cert_obj = x509.load_pem_x509_certificate(cert_data.encode(), default_backend())
            except ValueError as e:
                print(f"证书加载失败:{str(e)}")
                continue

            san_list = []
            try:
                san_ext = cert_obj.extensions.get_extension_for_class(x509.SubjectAlternativeName)
                san_list = san_ext.value.get_values_for_type(x509.DNSName)
            except x509.ExtensionNotFound:
                pass

            if cert_obj:
                results.append({
                    'subject': str(cert_obj.subject),
                    'issuer': str(cert_obj.issuer),
                    'san': str(san_list),
                    'validity': {
                        'start': str(cert_obj.not_valid_before),
                        'end': str(cert_obj.not_valid_after)
                    },
                    'signature_algorithm': str(cert_obj.signature_algorithm_oid._name)
                })

        return results

    def analyze_result(self, result,instruction,stderr,stdout):
        #指令结果分析
        result = self.parse_ssl_info(stdout)
        result = json.dumps(result,ensure_ascii=False)
        return result