传统静态黑名单过滤示例
blocklist = [
    "malware-domain.com",
    "phishing-site.net", 
    "c2-server.org"
]

def traditional_dns_filter(domain):
    if domain in blocklist:
        return "NXDOMAIN"   直接阻断
    else:
        return resolve_upstream(domain)   放行

DNS查询到达 → AI引擎提取特征（域名熵值、注册时间、词汇相似度等）
    ↓
模型评分（0.0-1.0）
    ↓
┌─────────────────────────────────────┐
│ 评分 > 0.85（高恶意）→ 直接阻断    │
│ 0.60 < 评分 ≤ 0.85 → 引导至蜜罐    │
│ 评分 ≤ 0.60（低风险）→ 正常解析    │
└─────────────────────────────────────┘

import re
import math
import random
from typing import Tuple
from datetime import datetime

 ============================================
 1. 特征提取模块
 ============================================
class FeatureExtractor:
    """从域名中提取用于AI判断的特征向量"""
    
    @staticmethod
    def calculate_entropy(domain: str) -> float:
        """计算域名的字符熵（信息熵越高越可疑）"""
        from collections import Counter
        chars = list(domain.replace('.', ''))
        if not chars:
            return 0.0
        counter = Counter(chars)
        entropy = 0.0
        for count in counter.values():
            prob = count / len(chars)
            entropy -= prob  math.log2(prob)
        return entropy
    
    @staticmethod
    def count_suspicious_keywords(domain: str) -> int:
        """统计可疑关键词数量"""
        keywords = ['verify', 'secure', 'login', 'account', 'confirm', 
                    'update', 'alert', 'security', 'billing', 'support']
        domain_lower = domain.lower()
        return sum(1 for kw in keywords if kw in domain_lower)
    
    @staticmethod
    def check_numeric_ratio(domain: str) -> float:
        """计算域名中数字字符占比"""
        domain_clean = domain.replace('.', '')
        if not domain_clean:
            return 0.0
        num_count = sum(1 for c in domain_clean if c.isdigit())
        return num_count / len(domain_clean)
    
    def extract_features(self, domain: str) -> dict:
        """提取多维特征向量"""
        return {
            'length': len(domain),
            'entropy': self.calculate_entropy(domain),
            'suspicious_keywords': self.count_suspicious_keywords(domain),
            'numeric_ratio': self.check_numeric_ratio(domain),
            'subdomain_count': domain.count('.')
        }

 ============================================
 2. AI评分模型（轻量级分类器）
 ============================================
class ThreatScorer:
    """基于特征的威胁评分模型"""
    
     模型权重（实际应用中通过机器学习训练得到）
    WEIGHTS = {
        'entropy': 1.5,            高熵值倾向恶意
        'suspicious_keywords': 0.8,
        'numeric_ratio': 1.2,
        'length': 0.3,             长域名倾向钓鱼
        'subdomain_count': 0.5
    }
    
    def __init__(self, threshold: float = 0.75):
        self.threshold = threshold
    
    def score(self, features: dict) -> Tuple[float, str]:
        """
        计算威胁评分 (0.0-1.0)
        返回值: (评分, 威胁等级)
        """
        raw_score = 0.0
        raw_score += features['entropy'] / 5.0  self.WEIGHTS['entropy']
        raw_score += features['suspicious_keywords']  self.WEIGHTS['suspicious_keywords']
        raw_score += features['numeric_ratio']  self.WEIGHTS['numeric_ratio']
        raw_score += min(features['length'] / 100, 1.0)  self.WEIGHTS['length']
        raw_score += min(features['subdomain_count'] / 5, 1.0)  self.WEIGHTS['subdomain_count']
        
         归一化到 [0, 1]
        max_possible = sum(self.WEIGHTS.values())
        score = min(raw_score / max_possible, 1.0)
        
         判定威胁等级
        if score >= 0.85:
            level = "CRITICAL"
        elif score >= 0.60:
            level = "SUSPICIOUS"
        else:
            level = "SAFE"
        
        return score, level

 ============================================
 3. DNS解析器 + AI拦截核心
 ============================================
class RootTrafficInterceptor:
    """根截流AI助手核心类"""
    
     模拟的正常域名白名单
    WHITELIST = {'google.com', 'github.com', 'stackoverflow.com', 'python.org'}
    
    def __init__(self):
        self.feature_extractor = FeatureExtractor()
        self.threat_scorer = ThreatScorer(threshold=0.75)
        self.stats = {'blocked': 0, 'passed': 0}
    
    def handle_dns_query(self, domain: str) -> dict:
        """
        处理DNS查询请求
        返回: {'action': 'block'/'pass'/'honeypot', 'reason': str, 'score': float}
        """
         Step 1: 白名单优先放行
        if domain in self.WHITELIST:
            return {'action': 'pass', 'reason': 'whitelist', 'score': 0.0}
        
         Step 2: 提取特征向量
        features = self.feature_extractor.extract_features(domain)
        
         Step 3: AI模型评分
        score, level = self.threat_scorer.score(features)
        
         Step 4: 根据评分执行处置策略
        if score >= 0.85:
            self.stats['blocked'] += 1
            return {'action': 'block', 'reason': f'AI判定: {level} (评分{score:.2f})', 'score': score}
        elif score >= 0.60:
            self.stats['blocked'] += 1
            return {'action': 'honeypot', 'reason': f'AI判定: {level} (评分{score:.2f})，转入蜜罐', 'score': score}
        else:
            self.stats['passed'] += 1
            return {'action': 'pass', 'reason': f'AI判定: {level} (评分{score:.2f})', 'score': score}
    
    def get_statistics(self) -> dict:
        """获取拦截统计"""
        total = self.stats['blocked'] + self.stats['passed']
        return {
            self.stats,
            'total': total,
            'block_rate': self.stats['blocked'] / total if total > 0 else 0
        }

 ============================================
 4. 模拟运行测试
 ============================================
if __name__ == "__main__":
    print("="  60)
    print("根截流AI助手 - 实时拦截测试")
    print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("="  60)
    
    interceptor = RootTrafficInterceptor()
    
     测试样本
    test_domains = [
         正常域名
        ("google.com", "正常访问"),
        ("myblog.example.com", "个人博客"),
        
         AI生成的钓鱼域名
        ("paypal-secure-login-verify.com", "AI生成钓鱼域名 - 模仿PayPal"),
        ("netflix-account-restore-online.net", "AI生成钓鱼域名 - 模仿Netflix"),
        ("apple-support-alert-2026.xyz", "AI生成钓鱼域名 - 高熵值+可疑词"),
        ("x9k2m4p1q7.xyz", "纯随机字符串域名"),
        ("zoom-meeting-help-urgent.org", "可疑关键词密集"),
        
         边界测试
        ("support.microsoft.com", "合法子域名（含support关键词）")
    ]
    
    print("\n【查询处理日志】")
    print("-"  60)
    
    for domain, desc in test_domains:
        result = interceptor.handle_dns_query(domain)
        action_icon = "✅ 放行" if result['action'] == 'pass' else "🚫 阻断"
        print(f"{action_icon} | {desc:<25} | {domain:<35} | {result['reason']}")
    
    print("\n" + "="  60)
    print("【拦截统计报告】")
    stats = interceptor.get_statistics()
    print(f"  总查询数: {stats['total']}")
    print(f"  放行次数: {stats['passed']}")
    print(f"  阻断次数: {stats['blocked']}")
    print(f"  拦截率: {stats['block_rate']:.1%}")
    print("="  60)