证书透明度CT日志监控实战：从检测到告警的完整方案

┌─────────────┐
                    │  CT日志API  │
                    └──────┬──────┘
                           │ get-entries (by tree_size)
                           ▼
                    ┌─────────────┐
                    │  批量解析   │
                    │  证书/SCT   │
                    └──────┬──────┘
                           │
          ┌────────────────┼────────────────┐
          ▼                ▼                ▼
    ┌───────────┐   ┌───────────┐   ┌───────────┐
    │ 域名清单  │   │ CA白名单  │   │ 策略规则  │
    │ 比对      │   │ 比对      │   │ 检查      │
    └─────┬─────┘   └─────┬─────┘   └─────┬─────┘
          │               │               │
          └───────────────┼───────────────┘
                          ▼
                   ┌─────────────┐
                   │  告警输出   │
                   │  JSON/邮件  │
                   └─────────────┘

"""
CT Log Monitor - RFC 6962 Compatible CT Log Query and Audit Tool

Requirements:
  pip install requests pyOpenSSL cryptography
"""

import json
import time
import hashlib
import base64
from datetime import datetime, timezone
from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field

import requests
from cryptography import x509
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import padding


@dataclass
class LogState:
    """Current state of a CT log (STH - SignedTreeHead)."""
    tree_size: int
    tree_head_hash: bytes
    root_hash: bytes
    timestamp: int
    signature: bytes

    @property
    def next_entry_index(self) -> int:
        return self.tree_size


@dataclass
class LogEntry:
    """A single entry in the CT log."""
    index: int
    leaf_input: bytes
    extra_data: bytes
    timestamp: int
    certificate: Optional[x509.Certificate] = None
    scts: List[Dict[str, Any]] = field(default_factory=list)


@dataclass
class MerkleProof:
    """Merkle tree inclusion proof for a log entry."""
    leaf_hash: bytes
    node_hashes: List[bytes]
    tree_size: int


class CTLogClient:
    """
    Client for querying RFC 6962 CT logs.
    
    Public log endpoints:
      - Google: https://ct.googleapis.com/logs/argon2024
      - Sectigo: https://ct.sectigo.com/logs/mercure2024
      - Cloudflare: https://ct.cloudflare.com/logs/nimbus2024
    """
    
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            'Accept': 'application/json',
            'User-Agent': 'CT-Log-Monitor/1.0'
        })
    
    def get_sth(self) -> LogState:
        """Get the current SignedTreeHead from the log."""
        resp = self.session.get(f'{self.base_url}/ct/v1/get-sth')
        resp.raise_for_status()
        data = resp.json()
        
        return LogState(
            tree_size=int(data['tree_size']),
            tree_head_hash=base64.b64decode(data['tree_head_sha256']),
            root_hash=base64.b64decode(data['root_hash']),
            timestamp=int(data['timestamp']),
            signature=base64.b64decode(data['signature'])
        )
    
    def get_entries(self, start: int, end: int) -> List[LogEntry]:
        """
        Fetch log entries by index range (inclusive).
        
        Args:
            start: Starting index (0-based)
            end: Ending index (inclusive)
        
        Returns:
            List of LogEntry objects
        """
        entries = []
        # RFC 6962 recommends fetching in batches of 256
        batch_size = 256
        
        for batch_start in range(start, end + 1, batch_size):
            batch_end = min(batch_start + batch_size - 1, end)
            resp = self.session.get(
                f'{self.base_url}/ct/v1/get-entries',
                params={'start': batch_start, 'end': batch_end}
            )
            resp.raise_for_status()
            data = resp.json()
            
            for entry_data in data.get('entries', []):
                entry = self._parse_entry(entry_data)
                entries.append(entry)
            
            # Rate limiting: respect log server constraints
            time.sleep(0.1)
        
        return entries
    
    def get_proof_by_hash(self, entry_hash: bytes) -> MerkleProof:
        """
        Get Merkle inclusion proof for an entry by its leaf hash.
        
        Args:
            entry_hash: SHA256 hash of the leaf_input (base64 encoded)
        """
        resp = self.session.get(
            f'{self.base_url}/ct/v1/get-proof-by-hash',
            params={'leaf_hash': entry_hash}
        )
        resp.raise_for_status()
        data = resp.json()
        
        return MerkleProof(
            leaf_hash=base64.b64decode(data['leaf_hash']),
            node_hashes=[base64.b64decode(h) for h in data.get('tree_size', [])],
            tree_size=int(data['tree_size'])
        )
    
    def _parse_entry(self, entry_data: Dict[str, str]) -> LogEntry:
        """Parse a raw entry from the API response."""
        leaf_input = base64.b64decode(entry_data['leaf_input'])
        extra_data = base64.b64decode(entry_data.get('extra_data', ''))
        
        # Parse timestamp (varuint encoding in extra_data)
        timestamp = self._parse_timestamp(extra_data)
        
        # Try to parse certificate from leaf_input
        certificate = None
        try:
            if leaf_input[:5] == b'\x93\x01\x00\x00\x00':  # TLS certificate opaque
                cert_der = leaf_input[5:]
                certificate = x509.load_x509_certificate(cert_der)
        except Exception:
            pass
        
        # Parse SCTs from extra_data
        scts = self._parse_scts(extra_data)
        
        return LogEntry(
            index=len(entry_data),  # Will be set by caller
            leaf_input=leaf_input,
            extra_data=extra_data,
            timestamp=timestamp,
            certificate=certificate,
            scts=scts
        )
    
    @staticmethod
    def _parse_timestamp(data: bytes) -> int:
        """Parse varuint timestamp from extra_data."""
        if not data:
            return 0
        # Simple varuint decode (first byte indicates length)
        if data[0] < 0x80:
            return int.from_bytes(data[:1], 'big')
        elif data[0] < 0xc0:
            return int.from_bytes(data[:2], 'big')
        elif data[0] < 0xe0:
            return int.from_bytes(data[:3], 'big')
        else:
            return int.from_bytes(data[:4], 'big')
    
    @staticmethod
    def _parse_scts(data: bytes) -> List[Dict[str, Any]]:
        """Parse SCTs from extra_data."""
        scts = []
        offset = 0
        
        while offset < len(data):
            # SCT list size
            if offset + 2 > len(data):
                break
            list_size = int.from_bytes(data[offset:offset+2], 'big')
            offset += 2
            
            for _ in range(list_size):
                sct = {}
                # Version
                version = data[offset]
                sct['version'] = version
                offset += 1
                
                if version == 0:  # v1 SCT
                    # Log ID (32 bytes)
                    sct['log_id'] = base64.b64encode(data[offset:offset+32]).decode()
                    offset += 32
                    
                    # Timestamp (8 bytes)
                    sct['timestamp'] = int.from_bytes(data[offset:offset+8], 'big')
                    offset += 8
                    
                    # Extensions length
                    ext_len = int.from_bytes(data[offset:offset+2], 'big')
                    offset += 2
                    sct['extensions'] = data[offset:offset+ext_len]
                    offset += ext_len
                    
                    # Signature
                    sig_len = int.from_bytes(data[offset:offset+2], 'big')
                    offset += 2
                    sct['signature'] = base64.b64encode(data[offset:offset+sig_len]).decode()
                    offset += sig_len
                
                scts.append(sct)
        
        return scts


class CTLogMonitor:
    """
    Enterprise CT log monitor for anomaly detection.
    
    Usage:
        monitor = CTLogMonitor(
            log_urls=['https://ct.googleapis.com/logs/argon2024'],
            domain_whitelist=['example.com', '*.example.com'],
            ca_whitelist=[
                'DigiCert Inc',
                'Let\'s Encrypt',
                'Sectigo Limited'
            ],
            max_cert_lifetime_days=397
        )
        monitor.run()
    """
    
    def __init__(
        self,
        log_urls: List[str],
        domain_whitelist: Optional[List[str]] = None,
        ca_whitelist: Optional[List[str]] = None,
        max_cert_lifetime_days: int = 397,
        alert_hooks: Optional[List[Any]] = None
    ):
        self.log_urls = log_urls
        self.domains = self._parse_domain_whitelist(domain_whitelist or [])
        self.ca_list = ca_whitelist or []
        self.max_lifetime_days = max_cert_lifetime_days
        self.alerts = []
        self.alert_hooks = alert_hooks or []
        
        self.clients = [CTLogClient(url) for url in log_urls]
        self.last_synced_size: Dict[str, int] = {}
    
    def _parse_domain_whitelist(self, domains: List[str]) -> set:
        """Parse domain whitelist, handling wildcard patterns."""
        parsed = set()
        for d in domains:
            parsed.add(d.lower())
            if d.startswith('*.'):
                # Add base domain for wildcard matching
                parsed.add(d[2:])
        return parsed
    
    def _is_domain_owned(self, domain: str) -> bool:
        """Check if domain is in the whitelist."""
        domain = domain.lower()
        for owned in self.domains:
            if domain == owned or domain.endswith('.' + owned):
                return True
        return False
    
    def _check_certificate(self, cert: x509.Certificate, entry_index: int) -> List[Dict[str, Any]]:
        """
        Check a single certificate for policy violations.
        
        Returns list of alert dictionaries.
        """
        alerts = []
        
        # Extract subject fields
        subject = cert.subject
        issuer = cert.issuer
        
        # Get SANs
        try:
            san_ext = cert.extensions.get_extension_for_class(
                x509.SubjectAlternativeName
            )
            sans = san_ext.value.get_values_for_type(x509.DNSName)
        except x509.ExtensionNotFound:
            sans = []
        
        # Check 1: Unauthorized domain issuance
        unauthorized = [d for d in sans if not self._is_domain_owned(d)]
        if unauthorized:
            alerts.append({
                'type': 'UNAUTHORIZED_DOMAIN',
                'severity': 'HIGH',
                'entry_index': entry_index,
                'domains': unauthorized,
                'subject': str(subject),
                'issuer': str(issuer),
                'not_before': cert.not_valid_before_utc.isoformat(),
                'not_after': cert.not_valid_after_utc.isoformat()
            })
        
        # Check 2: CA whitelist violation
        if self.ca_list:
            issuer_name = str(issuer).replace('/CN=', '').split(',')[0]
            if issuer_name not in self.ca_list:
                alerts.append({
                    'type': 'UNAUTHORIZED_CA',
                    'severity': 'HIGH',
                    'entry_index': entry_index,
                    'issuer': issuer_name,
                    'subject': str(subject)
                })
        
        # Check 3: Certificate lifetime violation
        lifetime_days = (cert.not_valid_after_utc - cert.not_valid_before_utc).days
        if lifetime_days > self.max_lifetime_days:
            alerts.append({
                'type': 'EXCESSIVE_LIFETIME',
                'severity': 'MEDIUM',
                'entry_index': entry_index,
                'lifetime_days': lifetime_days,
                'max_allowed': self.max_cert_lifetime_days,
                'subject': str(subject)
            })
        
        # Check 4: Missing SCTs (if we have entry data)
        # This is checked at the log level, not per-certificate
        
        return alerts
    
    def _verify_merkle_proof(
        self, 
        entry_hash: bytes, 
        proof: MerkleProof,
        sth: LogState
    ) -> bool:
        """
        Verify a Merkle tree inclusion proof.
        
        Returns True if proof is valid.
        """
        if len(proof.node_hashes) == 0:
            # Leaf is root (single entry log)
            return hashlib.sha256(entry_hash).digest() == sth.root_hash
        
        # Compute leaf hash
        current_hash = hashlib.sha256(entry_hash).digest()
        position = proof.leaf_hash  # Not used directly, position tracked by index
        
        # Walk up the tree
        for i, node_hash in enumerate(proof.node_hashes):
            # Determine if current node is left or right child
            if (position >> i) & 1:
                # Current is left child
                combined = node_hash + current_hash
            else:
                # Current is right child
                combined = current_hash + node_hash
            current_hash = hashlib.sha256(combined).digest()
        
        return current_hash == sth.root_hash
    
    def sync_log(self, client: CTLogClient, last_size: int = 0) -> List[Dict[str, Any]]:
        """
        Sync a single CT log and check for new entries.
        
        Returns list of alerts found.
        """
        all_alerts = []
        
        try:
            sth = client.get_sth()
        except requests.RequestException as e:
            print(f'[ERROR] Failed to get STH from {client.base_url}: {e}')
            return all_alerts
        
        current_size = sth.tree_size
        new_entries = current_size - last_size
        
        if new_entries <= 0:
            return all_alerts
        
        print(f'[INFO] Syncing {new_entries} new entries from {client.base_url}')
        
        try:
            entries = client.get_entries(last_size, current_size - 1)
        except requests.RequestException as e:
            print(f'[ERROR] Failed to fetch entries: {e}')
            return all_alerts
        
        for i, entry in enumerate(entries):
            entry.index = last_size + i
            
            if entry.certificate:
                alerts = self._check_certificate(entry.certificate, entry.index)
                all_alerts.extend(alerts)
                
                for alert in alerts:
                    print(f"[ALERT] {alert['type']}: {alert.get('domains', alert.get('subject', 'Unknown'))}")
        
        self.last_synced_size[client.base_url] = current_size
        return all_alerts
    
    def run_once(self) -> List[Dict[str, Any]]:
        """Run a single monitoring cycle across all configured logs."""
        all_alerts = []
        
        for client in self.clients:
            last_size = self.last_synced_size.get(client.base_url, 0)
            alerts = self.sync_log(client, last_size)
            all_alerts.extend(alerts)
        
        # Trigger alert hooks
        for hook in self.alert_hooks:
            try:
                hook(all_alerts)
            except Exception as e:
                print(f'[WARN] Alert hook failed: {e}')
        
        return all_alerts
    
    def run_continuous(self, interval_seconds: int = 300):
        """
        Run monitoring continuously at specified interval.
        
        Args:
            interval_seconds: Seconds between monitoring cycles (default: 5 minutes)
        """
        print(f'[INFO] Starting continuous CT monitoring (interval: {interval_seconds}s)')
        print(f'[INFO] Tracking logs: {[c.base_url for c in self.clients]}')
        print(f'[INFO] Monitoring domains: {self.domains}')
        
        while True:
            try:
                alerts = self.run_once()
                if alerts:
                    print(f'[ALERT] {len(alerts)} anomalies detected')
                else:
                    print('[OK] No anomalies detected')
            except Exception as e:
                print(f'[ERROR] Monitoring cycle failed: {e}')
            
            time.sleep(interval_seconds)


def main():
    """Example usage of CTLogMonitor."""
    import argparse
    
    parser = argparse.ArgumentParser(description='CT Log Monitor')
    parser.add_argument('--domain', action='append', help='Domain to monitor (can be repeated)')
    parser.add_argument('--ca', action='append', help='Authorized CA (can be repeated)')
    parser.add_argument('--max-lifetime', type=int, default=397, help='Max certificate lifetime in days')
    parser.add_argument('--interval', type=int, default=300, help='Monitoring interval in seconds')
    parser.add_argument('--once', action='store_true', help='Run once and exit')
    parser.add_argument('--logs', nargs='*', default=[
        'https://ct.googleapis.com/logs/argon2024',
        'https://ct.sectigo.com/logs/mercure2024',
        'https://ct.cloudflare.com/logs/nimbus2024'
    ], help='CT log URLs to monitor')
    
    args = parser.parse_args()
    
    # Configure monitor
    monitor = CTLogMonitor(
        log_urls=args.logs,
        domain_whitelist=args.domain or [],
        ca_whitelist=args.ca or [],
        max_cert_lifetime_days=args.max_lifetime
    )
    
    if args.once:
        alerts = monitor.run_once()
        if alerts:
            print(f'\n[RESULT] {len(alerts)} alerts:')
            for a in alerts:
                print(f'  - {a["type"]}: {json.dumps(a, indent=2, default=str)}')
            exit(1)
        else:
            print('[RESULT] No anomalies detected')
            exit(0)
    else:
        monitor.run_continuous(args.interval)


if __name__ == '__main__':
    main()

# 监控指定域名的CT日志
python3 ct_log_monitor.py \
  --domain example.com \
  --domain "*.example.com" \
  --ca "DigiCert Inc" \
  --ca "Let's Encrypt" \
  --once

# 输出示例：
# [INFO] Syncing 1523 new entries from https://ct.googleapis.com/logs/argon2024
# [ALERT] UNAUTHORIZED_DOMAIN: ['evil.example.com']
# [ALERT] EXCESSIVE_LIFETIME: lifetime_days=730, max_allowed=397
# [RESULT] 2 alerts

# 示例：Slack Webhook告警
import urllib.request
import json

def slack_alert(alerts: List[Dict]):
    if not alerts:
        return
    
    # Format alert message
    lines = ['⚠️ CT日志异常检测', '']
    for alert in alerts:
        lines.append(f'**{alert["type"]}** (严重: {alert["severity"]})')
        if 'domains' in alert:
            lines.append(f'  域名: {", ".join(alert["domains"])}')
        if 'issuer' in alert:
            lines.append(f'  签发机构: {alert["issuer"]}')
        lines.append('')
    
    payload = json.dumps({'text': '\n'.join(lines)}).encode('utf-8')
    req = urllib.request.Request(
        'https://hooks.slack.com/services/YOUR/WEBHOOK/URL',
        data=payload,
        headers={'Content-Type': 'application/json'}
    )
    urllib.request.urlopen(req)

# 注册告警钩子
monitor = CTLogMonitor(
    log_urls=['https://ct.googleapis.com/logs/argon2024'],
    domain_whitelist=['example.com'],
    alert_hooks=[slack_alert]
)

# 持久化同步状态
import sqlite3

def save_sync_state(log_url: str, tree_size: int, db_path: str = 'ct_monitor.db'):
    conn = sqlite3.connect(db_path)
    c = conn.cursor()
    c.execute('''
        INSERT INTO sync_state (log_url, tree_size, updated_at)
        VALUES (?, ?, datetime('now'))
        ON CONFLICT(log_url) DO UPDATE SET tree_size = excluded.tree_size
    ''')
    conn.commit()
    conn.close()

def load_sync_state(log_url: str, db_path: str = 'ct_monitor.db') -> int:
    conn = sqlite3.connect(db_path)
    c = conn.cursor()
    c.execute('SELECT tree_size FROM sync_state WHERE log_url = ?', (log_url,))
    row = c.fetchone()
    conn.close()
    return row[0] if row else 0

from concurrent.futures import ThreadPoolExecutor, as_completed

def sync_all_logs_parallel(self) -> List[Dict]:
    """Parallel sync across all configured logs."""
    all_alerts = []
    
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = {
            executor.submit(self.sync_log, client, self.last_synced_size.get(client.base_url, 0)): client
            for client in self.clients
        }
        
        for future in as_completed(futures):
            client = futures[future]
            try:
                alerts = future.result()
                all_alerts.extend(alerts)
            except Exception as e:
                print(f'[ERROR] Log sync failed for {client.base_url}: {e}')
    
    return all_alerts

def generate_audit_report(alerts: List[Dict], period: str = 'daily') -> Dict:
    """Generate structured audit report."""
    report = {
        'generated_at': datetime.now(timezone.utc).isoformat(),
        'period': period,
        'summary': {
            'total_alerts': len(alerts),
            'high_severity': sum(1 for a in alerts if a['severity'] == 'HIGH'),
            'medium_severity': sum(1 for a in alerts if a['severity'] == 'MEDIUM'),
            'low_severity': sum(1 for a in alerts if a['severity'] == 'LOW'),
        },
        'alerts': alerts,
        'compliance_status': 'PASS' if not alerts else 'FAIL'
    }
    return report