企业证书资产发现与清单管理：从 Shadow IT 到全面可见性的 PKI 运维实战

#!/usr/bin/env python3
"""
企业证书发现工具 — 网络扫描版
扫描指定网段或域名列表的 TLS 证书
"""

import ssl
import socket
import csv
import json
import logging
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Optional

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)

def get_certificate(hostname: str, port: int = 443, timeout: int = 5) -> Optional[dict]:
    """获取远程主机的 TLS 证书信息"""
    try:
        context = ssl.create_default_context()
        # 允许自签证书（企业内部常见）
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
        
        with socket.create_connection((hostname, port), timeout=timeout) as sock:
            with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                cert = ssock.getpeercert(binary_form=False)
                if not cert:
                    return None
                
                # 解析证书信息
                subject = dict(x[0] for x in cert.get('subject', []))
                issuer = dict(x[0] for x in cert.get('issuer', []))
                
                not_before = cert.get('notBefore', '')
                not_after = cert.get('notAfter', '')
                
                # 解析 SAN
                san = cert.get('subjectAltName', [])
                san_list = [entry[1] for entry in san if entry[0] == 'DNS']
                
                # 获取证书二进制以计算指纹
                der_cert = ssock.getpeercert(binary_form=True)
                import hashlib
                fingerprint = hashlib.sha256(der_cert).hexdigest()
                
                # 获取 TLS 版本和密码套件
                tls_version = ssock.version()
                cipher = ssock.cipher()
                
                return {
                    'hostname': hostname,
                    'port': port,
                    'subject_cn': subject.get('commonName', ''),
                    'issuer_cn': issuer.get('commonName', ''),
                    'not_before': not_before,
                    'not_after': not_after,
                    'san': san_list,
                    'fingerprint_sha256': fingerprint,
                    'tls_version': tls_version,
                    'cipher': cipher[0] if cipher else '',
                    'scan_time': datetime.utcnow().isoformat()
                }
    except (socket.timeout, ConnectionRefusedError, OSError) as e:
        logger.debug(f"连接失败 {hostname}:{port} - {e}")
        return None
    except Exception as e:
        logger.error(f"扫描异常 {hostname}:{port} - {e}")
        return None

def scan_network(targets: list, max_workers: int = 20) -> list:
    """并发扫描多个目标"""
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {}
        for target in targets:
            if isinstance(target, tuple):
                host, port = target
            else:
                host, port = target, 443
            futures[executor.submit(get_certificate, host, port)] = (host, port)
        
        for future in as_completed(futures):
            result = future.result()
            if result:
                results.append(result)
                remaining = None
                if result.get('not_after'):
                    try:
                        # 解析 ASN.1 时间格式
                        expiry = datetime.strptime(result['not_after'], '%b %d %H:%M:%S %Y %Z')
                        remaining = (expiry - datetime.utcnow()).days
                    except:
                        pass
                logger.info(
                    f"发现证书: {result['hostname']} | "
                    f"CN={result['subject_cn']} | "
                    f"剩余 {remaining} 天"
                )
    
    return results

def load_targets_from_file(filepath: str) -> list:
    """从文件加载扫描目标"""
    targets = []
    with open(filepath) as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith('#'):
                continue
            if ':' in line:
                host, port = line.rsplit(':', 1)
                targets.append((host, int(port)))
            else:
                targets.append((line, 443))
    return targets

def export_results(results: list, output_prefix: str):
    """导出扫描结果"""
    if not results:
        logger.warning("未发现任何证书")
        return
    
    # 导出 CSV
    csv_path = f"{output_prefix}.csv"
    with open(csv_path, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=[
            'hostname', 'port', 'subject_cn', 'issuer_cn',
            'not_after', 'san', 'fingerprint_sha256', 'tls_version'
        ])
        writer.writeheader()
        for r in results:
            r['san'] = '; '.join(r.get('san', []))
            writer.writerow(r)
    logger.info(f"CSV 导出: {csv_path}")
    
    # 导出 JSON
    json_path = f"{output_prefix}.json"
    with open(json_path, 'w') as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
    logger.info(f"JSON 导出: {json_path}")
    
    # 打印摘要
    print(f"\n{'='*60}")
    print(f"证书发现报告")
    print(f"{'='*60}")
    print(f"扫描时间: {datetime.utcnow().isoformat()}")
    print(f"发现证书总数: {len(results)}")
    
    # 按签发机构统计
    issuer_stats = {}
    for r in results:
        issuer = r.get('issuer_cn', 'Unknown')
        issuer_stats[issuer] = issuer_stats.get(issuer, 0) + 1
    print(f"\n按签发机构:")
    for issuer, count in sorted(issuer_stats.items(), key=lambda x: -x[1]):
        print(f"  {issuer}: {count}")
    
    # 即将过期的证书（30天内）
    expiring = []
    for r in results:
        try:
            expiry = datetime.strptime(r['not_after'], '%b %d %H:%M:%S %Y %Z')
            remaining = (expiry - datetime.utcnow()).days
            if remaining <= 30:
                expiring.append((r['hostname'], remaining))
        except:
            pass
    if expiring:
        print(f"\n⚠️ 即将过期（30天内）: {len(expiring)} 张")
        for hostname, days in sorted(expiring, key=lambda x: x[1]):
            print(f"  {hostname}: {days} 天")
    
    print(f"{'='*60}\n")

if __name__ == '__main__':
    import sys
    if len(sys.argv) < 2:
        print("用法: python3 cert_discovery.py <targets.txt>")
        print("targets.txt 格式: 每行一个域名或 IP:端口")
        sys.exit(1)
    
    targets = load_targets_from_file(sys.argv[1])
    results = scan_network(targets)
    export_results(results, 'cert-discovery-report')

# Lemur 的证书清单查询 API 示例
import requests

LEmur_URL = "https://lemur.example.com"
HEADERS = {"Authorization": "Bearer $LEMUR_API_TOKEN"}

# 获取所有证书
r = requests.get(f"{LEmur_URL}/api/1/certificates", headers=HEADERS)
certs = r.json()

# 查询即将过期的证书
from datetime import datetime, timedelta
threshold = datetime.utcnow() + timedelta(days=30)
for cert in certs:
    if cert['validityEnd'] < threshold.isoformat():
        print(f"⚠️ 即将过期: {cert['name']} | 到期: {cert['validityEnd']}")

# 识别国密证书
from cryptography import x509
from cryptography.hazmat.backends import default_backend

GM_OIDS = {
    '1.2.156.10197.1.301': 'SM2 公钥算法',
    '1.2.156.10197.1.104': 'SM4 分组密码',
    '1.2.156.10197.1.501': 'SM3 杂凑算法',
}

def is_gm_certificate(cert_path: str) -> dict:
    """检查证书是否使用国密算法"""
    with open(cert_path, 'rb') as f:
        cert = x509.load_pem_x509_certificate(f.read(), default_backend())
    
    public_key = cert.public_key()
    algorithm = type(public_key).__name__
    
    result = {
        'is_sm2': 'SM2' in algorithm,
        'algorithm': algorithm,
        'signature_oid': cert.signature_algorithm_oid.dotted_string,
        'is_gm_signature': cert.signature_algorithm_oid.dotted_string in GM_OIDS,
    }
    
    return result