企业证书资产管理实战：从Shadow证书发现到自动化清单构建

#!/usr/bin/env python3
"""
企业证书资产扫描器
功能：网络层TLS探测 + 主机层文件系统扫描
"""

import os
import ssl
import socket
from datetime import datetime
from dataclasses import dataclass
from typing import List, Optional
import json


@dataclass
class CertificateInfo:
    """证书信息数据类"""
    hostname: str
    port: int
    subject: str
    issuer: str
    not_before: datetime
    not_after: datetime
    serial_number: str
    fingerprint_sha256: str
    key_type: str
    key_size: int
    source: str  # 'network' or 'filesystem'
    
    def to_dict(self) -> dict:
        return {
            'hostname': self.hostname,
            'port': self.port,
            'subject': self.subject,
            'issuer': self.issuer,
            'not_before': self.not_before.isoformat(),
            'not_after': self.not_after.isoformat(),
            'serial_number': self.serial_number,
            'fingerprint_sha256': self.fingerprint_sha256,
            'key_type': self.key_type,
            'key_size': self.key_size,
            'source': self.source
        }


class CertificateScanner:
    """证书扫描器"""
    
    def __init__(self, timeout: int = 5):
        self.timeout = timeout
        self.certificates: List[CertificateInfo] = []
    
    def scan_network(self, hostname: str, ports: List[int] = None) -> List[CertificateInfo]:
        """网络层TLS握手探测"""
        if ports is None:
            ports = [443, 8443, 9500, 9501]
        
        results = []
        for port in ports:
            try:
                context = ssl.create_default_context()
                context.check_hostname = False
                context.verify_mode = ssl.CERT_NONE
                
                with socket.create_connection((hostname, port), timeout=self.timeout) as sock:
                    with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                        cert_der = ssock.getpeercert(binary_form=True)
                        cert = ssl.DER_cert_to_pem_cert(cert_der)
                        
                        # 解析证书信息
                        cert_info = self._parse_certificate(cert, hostname, port, 'network')
                        if cert_info:
                            results.append(cert_info)
                            self.certificates.append(cert_info)
                            
            except (socket.timeout, socket.error, ssl.SSLError):
                continue
        
        return results
    
    def scan_filesystem(self, search_paths: List[str] = None) -> List[CertificateInfo]:
        """主机层文件系统扫描"""
        if search_paths is None:
            search_paths = [
                '/etc/ssl/certs',
                '/etc/pki/tls/certs',
                '/etc/nginx/ssl',
                '/etc/ssl/private',
                '/opt/certs',
            ]
        
        results = []
        cert_extensions = {'.crt', '.pem', '.cer', '.der', '.p12', '.pfx'}
        
        for search_path in search_paths:
            if not os.path.exists(search_path):
                continue
            
            try:
                for root, dirs, files in os.walk(search_path):
                    for filename in files:
                        if any(filename.endswith(ext) for ext in cert_extensions):
                            filepath = os.path.join(root, filename)
                            try:
                                with open(filepath, 'r', errors='ignore') as f:
                                    content = f.read(1024 * 1024)
                                    
                                if '-----BEGIN CERTIFICATE-----' in content:
                                    cert_info = self._parse_certificate_file(filepath, content)
                                    if cert_info:
                                        results.append(cert_info)
                                        self.certificates.append(cert_info)
                                        
                            except (IOError, UnicodeDecodeError):
                                continue
                                
            except PermissionError:
                continue
        
        return results
    
    def _parse_certificate(self, pem_cert: str, hostname: str, port: int, 
                           source: str) -> Optional[CertificateInfo]:
        """解析PEM格式证书"""
        try:
            cert = ssl._ssl._test_decode_cert(pem_cert)
            
            # 提取基本信息
            subject_parts = cert.get('subject', [])
            issuer_parts = cert.get('issuer', [])
            
            subject = self._format_name(subject_parts)
            issuer = self._format_name(issuer_parts)
            
            not_before = datetime.strptime(cert['notBefore'], '%b %d %H:%M:%S %Y %Z')
            not_after = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
            
            # 计算指纹
            import hashlib
            fingerprint = hashlib.sha256(pem_cert.encode()).hexdigest()
            
            # 密钥信息
            pubkey = cert.get('publicKey', {})
            key_type = pubkey.get('type', 'unknown')
            key_size = pubkey.get('size', 0)
            
            return CertificateInfo(
                hostname=hostname,
                port=port,
                subject=subject,
                issuer=issuer,
                not_before=not_before,
                not_after=not_after,
                serial_number=cert.get('serialNumber', ''),
                fingerprint_sha256=fingerprint,
                key_type=key_type,
                key_size=key_size,
                source=source
            )
        except Exception as e:
            print(f"解析证书失败: {e}")
            return None
    
    def _parse_certificate_file(self, filepath: str, content: str) -> Optional[CertificateInfo]:
        """解析本地证书文件"""
        try:
            # 提取所有证书
            certs = content.split('-----BEGIN CERTIFICATE-----')
            certs = [c for c in certs if '-----END CERTIFICATE-----' in c]
            
            if not certs:
                return None
            
            # 解析第一个证书
            cert_pem = f"-----BEGIN CERTIFICATE-----{certs[0]}-----END CERTIFICATE-----"
            cert = ssl._ssl._test_decode_cert(cert_pem)
            
            subject_parts = cert.get('subject', [])
            subject = self._format_name(subject_parts)
            
            # 计算指纹
            import hashlib
            fingerprint = hashlib.sha256(cert_pem.encode()).hexdigest()
            
            not_after_str = cert.get('notAfter', 'unknown')
            try:
                not_after = datetime.strptime(not_after_str, '%b %d %H:%M:%S %Y %Z')
            except:
                not_after = datetime.min
            
            return CertificateInfo(
                hostname=filepath,
                port=0,
                subject=subject,
                issuer='',
                not_before=datetime.min,
                not_after=not_after,
                serial_number=cert.get('serialNumber', ''),
                fingerprint_sha256=fingerprint,
                key_type='unknown',
                key_size=0,
                source='filesystem'
            )
        except Exception as e:
            return None
    
    def _format_name(self, name_parts: list) -> str:
        """格式化证书名称"""
        if not name_parts:
            return ''
        
        parts = []
        for part in name_parts:
            if len(part) >= 2:
                key = part[0]
                value = part[1]
                parts.append(f"{key}={value}")
        
        return ', '.join(parts)
    
    def export_inventory(self, output_file: str = 'certificate-inventory.json'):
        """导出证书清单"""
        inventory = {
            'scan_time': datetime.now().isoformat(),
            'total_certificates': len(self.certificates),
            'certificates': [cert.to_dict() for cert in self.certificates]
        }
        
        with open(output_file, 'w') as f:
            json.dump(inventory, f, indent=2, ensure_ascii=False)
        
        print(f"证书清单已导出到: {output_file}")
        return inventory


def main():
    """主函数"""
    scanner = CertificateScanner(timeout=5)
    
    # 网络扫描示例
    print("正在扫描网络证书...")
    network_certs = scanner.scan_network('localhost', ports=[443, 8443])
    print(f"发现 {len(network_certs)} 个网络证书")
    
    # 文件系统扫描示例
    print("\n正在扫描本地证书文件...")
    fs_certs = scanner.scan_filesystem()
    print(f"发现 {len(fs_certs)} 个本地证书")
    
    # 导出清单
    scanner.export_inventory()
    
    # 统计信息
    print(f"\n=== 扫描统计 ===")
    print(f"总计发现证书: {len(scanner.certificates)}")
    
    # 统计国密证书（SM2/SM3/SM4标识）
    gm_certs = [c for c in scanner.certificates if any(kw in c.subject for kw in ['SM2', 'SM3', 'SM4', '国密'])]
    print(f"国密证书: {len(gm_certs)}")
    
    # 即将过期的证书
    expiring_soon = [c for c in scanner.certificates 
                     if c.not_after < datetime.now() + __import__('datetime').timedelta(days=30)]
    print(f"30天内过期: {len(expiring_soon)}")


if __name__ == '__main__':
    main()

class ShadowCertificateDetector:
    """影子证书检测器"""
    
    def __init__(self, known_certs: List[dict]):
        """
        known_certs: 已登记的合法证书列表
        每个证书包含: fingerprint, issuer, valid_from, valid_to
        """
        self.known_fingerprints = {c['fingerprint'] for c in known_certs}
    
    def detect_shadow(self, discovered_certs: List[CertificateInfo]) -> List[CertificateInfo]:
        """检测影子证书"""
        shadows = []
        
        for cert in discovered_certs:
            # 检查指纹是否在已知列表中
            if cert.fingerprint_sha256 not in self.known_fingerprints:
                shadows.append(cert)
                continue
            
            # 检查签发者是否为受信任CA
            if not self._is_trusted_issuer(cert.issuer):
                shadows.append(cert)
                continue
            
            # 检查存储位置是否安全
            if cert.source == 'filesystem':
                if not self._is_secure_storage(cert.hostname):
                    shadows.append(cert)
        
        return shadows
    
    def _is_trusted_issuer(self, issuer: str) -> bool:
        """检查签发者是否为受信任CA"""
        trusted_issuers = [
            '国家密码管理局',
            'BJCA',
            'ShenZhengCA',
            'DigiCert',
            "Let's Encrypt",
        ]
        return any(t in issuer for t in trusted_issuers)
    
    def _is_secure_storage(self, path: str) -> bool:
        """检查证书存储位置是否安全"""
        secure_paths = [
            '/etc/ssl/',
            '/etc/pki/',
            '/opt/certs/',
            'hsm://',
        ]
        return any(path.startswith(p) for p in secure_paths)

class GMDualCertManager:
    """国密双证书管理器"""
    
    def __init__(self, cert_dir: str = '/etc/gm-certs'):
        self.cert_dir = cert_dir
        self.signature_certs = []
        self.encryption_certs = []
    
    def scan_dual_certs(self):
        """扫描双证书配对"""
        import glob
        
        # 扫描签名证书（文件名含 sig 或 subject 含签名用途）
        sig_pattern = os.path.join(self.cert_dir, '*sig*.crt')
        for cert_path in glob.glob(sig_pattern):
            cert_info = self._parse_cert_file(cert_path)
            if cert_info:
                self.signature_certs.append(cert_info)
        
        # 扫描加密证书（文件名含 enc 或 subject 含加密用途）
        enc_pattern = os.path.join(self.cert_dir, '*enc*.crt')
        for cert_path in glob.glob(enc_pattern):
            cert_info = self._parse_cert_file(cert_path)
            if cert_info:
                self.encryption_certs.append(cert_info)
        
        return self._find_pairs()
    
    def _find_pairs(self) -> List[dict]:
        """查找配对证书"""
        pairs = []
        
        for sig_cert in self.signature_certs:
            paired = False
            for enc_cert in self.encryption_certs:
                # 检查是否属于同一主体
                if sig_cert['subject'] == enc_cert['subject']:
                    pairs.append({
                        'signature_cert': sig_cert,
                        'encryption_cert': enc_cert,
                        'paired': True
                    })
                    paired = True
                    break
            if not paired:
                pairs.append({
                    'signature_cert': sig_cert,
                    'encryption_cert': None,
                    'paired': False
                })
        
        return pairs
    
    def _parse_cert_file(self, filepath: str) -> Optional[dict]:
        """解析证书文件"""
        try:
            with open(filepath, 'r') as f:
                content = f.read()
            
            if '-----BEGIN CERTIFICATE-----' not in content:
                return None
            
            import ssl
            cert = ssl._ssl._test_decode_cert(content)
            
            subject_parts = cert.get('subject', [])
            subject = self._format_name(subject_parts)
            
            not_after_str = cert.get('notAfter', '')
            try:
                not_after = datetime.strptime(not_after_str, '%b %d %H:%M:%S %Y %Z')
            except:
                not_after = datetime.min
            
            return {
                'path': filepath,
                'subject': subject,
                'not_after': not_after,
                'fingerprint': cert.get('serialNumber', '')
            }
        except Exception as e:
            return None
    
    def _format_name(self, name_parts: list) -> str:
        """格式化名称"""
        if not name_parts:
            return ''
        parts = []
        for part in name_parts:
            if len(part) >= 2:
                parts.append(f"{part[0]}={part[1]}")
        return ', '.join(parts)
    
    def generate_report(self) -> str:
        """生成管理报告"""
        pairs = self.scan_dual_certs()
        
        report = []
        report.append("# 国密双证书管理报告")
        report.append(f"生成时间: {datetime.now().isoformat()}")
        report.append(f"\n签名证书数量: {len(self.signature_certs)}")
        report.append(f"加密证书数量: {len(self.encryption_certs)}")
        
        # 配对情况
        paired = [p for p in pairs if p['paired']]
        unpaired = [p for p in pairs if not p['paired']]
        
        report.append(f"\n## 配对情况")
        report.append(f"已配对: {len(paired)}")
        report.append(f"未配对: {len(unpaired)}")
        
        # 即将过期
        expiring = []
        for cert in self.signature_certs + self.encryption_certs:
            if cert['not_after'] < datetime.now() + __import__('datetime').timedelta(days=90):
                expiring.append(cert)
        
        report.append(f"\n## 即将过期（90天内）")
        report.append(f"数量: {len(expiring)}")
        for cert in expiring:
            report.append(f"- {cert['subject']}: {cert['not_after'].date()}")
        
        return '\n'.join(report)

@dataclass
class CertificateAsset:
    """证书资产数据类"""
    # 基本信息
    fingerprint: str
    subject: str
    issuer: str
    serial_number: str
    
    # 有效期
    not_before: datetime
    not_after: datetime
    
    # 分类信息
    cert_type: str  # 'signature', 'encryption', 'server', 'client'
    key_algorithm: str  # 'SM2', 'RSA', 'ECDSA'
    key_size: int
    
    # 来源信息
    source: str  # 'network', 'filesystem', 'cloud'
    owner: str  # 责任人
    business_system: str  # 业务系统
    
    # 合规信息
    is_registered: bool  # 是否已登记
    compliance_status: str  # 'compliant', 'non-compliant', 'pending'
    
    def risk_score(self) -> int:
        """计算风险评分（0-100）"""
        score = 0
        
        # 未登记证书高风险
        if not self.is_registered:
            score += 30
        
        # 即将过期
        days_until_expiry = (self.not_after - datetime.now()).days
        if days_until_expiry < 0:
            score += 40  # 已过期
        elif days_until_expiry < 30:
            score += 20
        elif days_until_expiry < 90:
            score += 10
        
        # 弱密钥
        if self.key_algorithm == 'RSA' and self.key_size < 2048:
            score += 20
        elif self.key_algorithm not in ['SM2', 'RSA', 'ECDSA']:
            score += 10
        
        return min(score, 100)

import schedule
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def scheduled_scan():
    """定时扫描任务"""
    logger.info("开始定时证书扫描...")
    
    scanner = CertificateScanner()
    
    # 扫描已知主机列表
    hostnames = [
        'www.example.com',
        'api.example.com',
        'internal.example.com',
    ]
    
    for hostname in hostnames:
        certs = scanner.scan_network(hostname)
        logger.info(f"{hostname}: 发现 {len(certs)} 个证书")
    
    # 扫描本地文件系统
    fs_certs = scanner.scan_filesystem()
    logger.info(f"本地证书: 发现 {len(fs_certs)} 个")
    
    # 导出清单
    scanner.export_inventory('latest-inventory.json')
    
    # 检测影子证书
    detector = ShadowCertificateDetector(known_certs=[])  # 从数据库加载
    shadows = detector.detect_shadow(scanner.certificates)
    
    if shadows:
        logger.warning(f"发现 {len(shadows)} 个影子证书！")
        for cert in shadows:
            logger.warning(f"  - {cert.hostname}:{cert.port}")
    
    logger.info("定时扫描完成")


# 每天凌晨2点执行扫描
schedule.every().day.at("02:00").do(scheduled_scan)

# 每周日导出详细报告
schedule.every().sunday.at("03:00").do(lambda: generate_weekly_report())

import asyncio
from concurrent.futures import ThreadPoolExecutor


async def parallel_scan(hostnames: List[str], max_concurrent: int = 10):
    """并行扫描"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def scan_host(hostname: str):
        async with semaphore:
            scanner = CertificateScanner()
            return await asyncio.to_thread(scanner.scan_network, hostname)
    
    tasks = [scan_host(h) for h in hostnames]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    all_certs = []
    for result in results:
        if isinstance(result, list):
            all_certs.extend(result)
    
    return all_certs