# entropy_features.py —— 统计特征层的最小可复现实现
import math
from collections import Counter

def shannon_entropy(s: str) -> float:
    """香农熵：衡量字符串字符分布的混乱程度"""
    if not s:
        return 0.0
    counts = Counter(s)                    # 统计每个字符出现次数
    n = len(s)
    return -sum(c / n * math.log2(c / n) for c in counts.values())

def stat_features(domain: str) -> dict:
    """提取经典词汇统计特征，熵值只是其中一维"""
    sld = domain.split('.')[0].lower()     # 取二级域主体（去掉 TLD）
    if not sld:
        sld = domain.lower()
    n = max(len(sld), 1)
    vowels = sum(c in 'aeiou' for c in sld)
    consonants = sum(c.isalpha() and c not in 'aeiou' for c in sld)
    return {
        'len': len(sld),
        'entropy': round(shannon_entropy(sld), 4),
        'digit_ratio': round(sum(c.isdigit() for c in sld) / n, 4),
        'vowel_ratio': round(vowels / n, 4),
        'uniq_ratio': round(len(set(sld)) / n, 4),
        'hyphens': sld.count('-'),
    }

if __name__ == '__main__':
    for d in ['google', 'xjqkwzbnmgspv', 'impressiveideas', 'chncietyazlbbi']:
        f = stat_features(d)
        print(f"{d:16s} entropy={f['entropy']:.2f} vowel={f['vowel_ratio']}")

# dict_dga_sim.py —— 字典型 DGA 的实验复现（仅用于生成检测语料，无任何外联逻辑）
import random

WORDLISTS = {                      # 词表来源参考公开 matsnu/gozi 家族研究，此处精简为示意
    'adj':  ['bright', 'silent', 'lucky', 'swift', 'golden', 'silent'],
    'noun': ['market', 'forest', 'window', 'signal', 'harbor', 'meadow'],
    'verb': ['share', 'build', 'track', 'lift'],
}

def dict_dga_sample(seed: int, n: int = 5) -> list[str]:
    """模拟字典型 DGA：从词表拼接出'像人话'的域名"""
    rng = random.Random(seed)                       # 固定种子保证实验可复现
    out = []
    for _ in range(n):
        parts = [rng.choice(WORDLISTS[k]) for k in ('adj', 'noun')]
        if rng.random() < 0.4:
            parts.append(rng.choice(WORDLISTS['verb']))   # 部分样本三词拼接
        out.append(''.join(parts) + rng.choice(['.com', '.net', '.biz']))
    return out

if __name__ == '__main__':
    for d in dict_dga_sample(seed=20261105, n=6):
        print(d)
        # 预期输出示例：luckyharbor.com / brightsignaltrack.net / swiftmeadow.com ...

# 1. 建立隔离环境
python -m venv dga-venv && source dga-venv/bin/activate   # Windows: dga-venv\Scripts\activate
# 2. 安装依赖（版本为 2026-10 实测可用组合）
pip install "scikit-learn>=1.5" "lightgbm>=4.5" "transformers>=4.46" requests pandas
# 3. 下载语料（示例：360 Netlab 每日 DGA feed 与 Bambenek 域名列表，均公开）
# curl -o dga_360.txt https://data.netlab.360.com/feeds/dga/dga.txt
# curl -o dga_bambenek.csv https://osint.bambenekconsulting.com/feeds/dga-feed.csv
# 白名单基线：curl -o tranco.csv https://tranco-list.eu/top-1m.csv 下载每日更新的 Tranco 列表
# 4. 运行本文流水线
python train_baseline.py && python serve_pipeline.py

# 错误示例：普通 K-Fold —— 同一 DGA 家族的样本被随机分进训练集和测试集
from sklearn.model_selection import KFold
kf = KFold(n_splits=5, shuffle=True)   # 模型在验证集上"作弊"：它见过同家族的其他域名
# 结果：测试准确率 98%+，上线后遇到训练集外的新家族召回率可能不到 30%

# train_baseline.py —— 家族隔离交叉验证（Family-Stratified K-Fold）
import pandas as pd
from sklearn.metrics import classification_report
from lightgbm import LGBMClassifier
from sklearn.model_selection import GroupKFold
from entropy_features import stat_features   # 复用 2.1 的特征提取器

def load_domains(path: str, label: int) -> pd.DataFrame:
    """读入域名样本，DGA 侧需带家族标签用于分组"""
    rows = []
    for line in open(path, encoding='utf-8'):
        parts = line.strip().split(',')
        if len(parts) >= 2:                      # DGA feed 格式：domain,family
            rows.append({'domain': parts[0], 'family': parts[1], 'label': label})
        elif parts[0]:
            rows.append({'domain': parts[0], 'family': 'benign', 'label': label})
    return pd.DataFrame(rows)

df = pd.concat([load_domains('dga_360.txt', 1), load_domains('tranco_sld.txt', 0)])
df = df.sample(frac=1.0, random_state=42).reset_index(drop=True)   # 打乱，固定种子
X = pd.DataFrame([stat_features(d) for d in df['domain']])

gkf = GroupKFold(n_splits=5)              # 按 family 分组整折隔离，杜绝家族泄漏
scores = []
for tr, te in gkf.split(X, df['label'], groups=df['family']):
    clf = LGBMClassifier(n_estimators=300, learning_rate=0.08, num_leaves=31)
    clf.fit(X.iloc[tr], df['label'].iloc[tr])
    scores.append(clf.score(X.iloc[te], df['label'].iloc[te]))
    print(f"fold score={scores[-1]:.4f}")
print(f"family-isolated CV mean = {sum(scores)/len(scores):.4f}")
# 预期输出：fold score ≈ 0.93~0.96（低于普通 KFold 的 0.98+，但这才是真实水位）

# 错误示例：全量域名直接丢给 LLM，自由文本提问
for domain in all_dns_queries:                  # 每天百万级 DNS 查询全部调用 API
    resp = llm.chat(f"这个域名 {domain} 是不是恶意的？请回答是或不是")
    # 问题1：成本失控（百万次调用 × 每次数百 token）
    # 问题2：域名是攻击者可控的输入，未经清洗直接拼进提示词，存在提示注入面
    # 问题3：自由文本输出无法被程序消费，告警系统解析不了"看起来有点可疑"

# llm_verdict.py —— 置信度门控 + 结构化输出的 LLM 研判层
import json, hashlib, time
import requests

CACHED: dict[str, dict] = {}          # 进程内 TTL 缓存（生产换 Redis）
TTL_SECONDS = 86400                   # 域名判定 24h 内缓存，重复查询零成本

JUDGE_PROMPT = """你是 DNS 安全分析师。判断该域名是否为算法生成（DGA）或仿冒域名。
只依据构词特征与你的语言知识判断，不要臆测无法验证的注册信息。
严格输出 JSON（不要任何多余文本）：
{{"verdict":"malicious|benign|unknown","confidence":0-100,
  "family_guess":"如 matsnu/gozi/未知","reasons":["一句话证据"]}}
域名：{domain}
初筛模型置信度：{ml_score}"""

def llm_judge(domain: str, ml_score: float) -> dict:
    key = hashlib.sha1(domain.encode()).hexdigest()          # 缓存键与提示词解耦
    hit = CACHED.get(key)
    if hit and time.time() - hit['ts'] < TTL_SECONDS:        # 命中缓存直接返回
        return hit['data']
    # 关键行1：只允许 [a-z0-9.-] 通过，剥掉一切可能构成提示注入的字符
    safe = ''.join(c for c in domain.lower() if c in 'abcdefghijklmnopqrstuvwxyz0123456789.-')
    if not safe or safe != domain.lower():
        return {'verdict': 'unknown', 'confidence': 0, 'reasons': ['域名含异常字符，转人工']}
    payload = {'model': 'qwen-plus', 'temperature': 0.1,     # 温度调低保证判定稳定
               'messages': [{'role': 'user',
                             'content': JUDGE_PROMPT.format(domain=safe, ml_score=round(ml_score, 3))}]}
    r = requests.post('http://llm-gw.internal/v1/chat/completions',
                      json=payload, timeout=20)              # 内部网关，走统一鉴权与限额
    text = r.json()['choices'][0]['message']['content']
    data = json.loads(text)                                  # 结构化输出：解析失败即降级
    data['reasons'] = data.get('reasons', [])[:3]            # 证据链截断，防 token 爆炸
    CACHED[key] = {'ts': time.time(), 'data': data}
    return data

# fuse.py —— 三层结果融合与 SIEM 告警输出
def fuse(domain: str, ml_score: float, llm_result: dict, ttl_anomaly: bool) -> dict:
    """统计分、语义判定、DNS 行为信号三源融合"""
    sem = 1 if llm_result['verdict'] == 'malicious' else 0
    # 融合规则：行为信号（NXDOMAIN 突发等）是强放大器，但单源不足以定罪
    if ml_score >= 0.9 and sem == 1:
        action, severity = 'block', 'high'      # 双源一致：可建议阻断
    elif ml_score >= 0.5 and (sem == 1 or ttl_anomaly):
        action, severity = 'alert', 'medium'     # 任一辅助源支持：告警人工复核
    else:
        action, severity = 'observe', 'low'      # 其余一律先观察，绝不误杀
    return {
        'domain': domain, 'action': action, 'severity': severity,
        'ml_score': round(ml_score, 3),
        'llm_family': llm_result.get('family_guess', 'n/a'),
        'evidence': llm_result['reasons'],      # LLM 给出的构词证据，供分析师复核
        'schema': 'dga-alert/1.0',              # 固定 schema 方便 SIEM 端解析
    }

if __name__ == '__main__':
    # 端到端示例：一个字典型 DGA 样本走完三层
    print(json.dumps(fuse('luckyharbor.com',
                         ml_score=0.42,                      # 统计层：低置信（字典型盲区）
                         llm_result=llm_judge('luckyharbor.com', 0.42),
                         ttl_anomaly=False), ensure_ascii=False, indent=2))

{
  "domain": "luckyharbor.com",
  "action": "alert",
  "severity": "medium",
  "ml_score": 0.42,
  "llm_family": "matsnu-like",
  "evidence": ["形容词+名词无品牌语义的机械拼接，符合字典型 DGA 构词模式"],
  "schema": "dga-alert/1.0"
}