当AI开始幻觉、而有人在那里等着：2026企业级AI幻觉依赖、Slopsquatting抢注、代码供应链投毒与幽灵API治理实战

┌────────────────────────────────────────────────────────────────────────────────┐
│  2026 Hallucinated Dependencies & Code Supply Chain: Five-Layer Model          │
├────────────────────────────────────────────────────────────────────────────────┤
│                                                                                │
│  [Predictable hallucination distribution + Occupiable namespace +             │
│   Zero-touch install chain + Error-driven adaptation culture →                │
│   AI invents names, attackers register them, pipelines install them,          │
│   teams build reality around them — no human ever makes a wrong decision]      │
│       ↓                                                                        │
│  ┌─ L1: 存在性与出生证明核验层 (Existence & Birth-Certificate Verification)──┐ │
│  │  • 双重核验: AI推荐的每个依赖先验存在性再验注册历史(年龄/作者/版本轨迹)    │ │
│  │  • 抢注特征库: 注册晚于训练截止/新账户/LLM生成README/跳版本号 指纹匹配     │ │
│  │  • 幽灵API检测: 生成代码的全部外部调用与官方SDK清单比对, 不存在即阻断       │ │
│  │  • 引入血缘: 每个依赖标记"人类引入"或"AI引入", AI引入强制增强核验           │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L2: 幻觉预测与防御性占位层 (Hallucination Prediction & Squat Defense) ───┐ │
│  │  • 自模型审计: 用企业同款模型批量生成推荐, 提取高频幻觉包名清单             │ │
│  │  • 防御性注册: 高概率幻觉名在可信命名空间下防御性占位(空包+警示)            │ │
│  │  • 抢注监测: 幻觉清单与注册局新增数据实时比对, 命中即全网黑名单             │ │
│  │  • 生态共享: 幻觉预测清单与抢注指纹跨企业/跨生态共享                       │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L3: 构建时隔离与执行阻断层 (Build-Time Isolation & Execution Blocking) ──┐ │
│  │  • 安装脚本禁用: 全生态默认禁止post-install脚本, 白名单逐项审批             │ │
│  │  • CI凭据隔离: 构建环境零长期凭据, 依赖安装与密钥访问网络分区               │ │
│  │  • 条件载荷检测: 依赖行为在CI环境变量上下文中差分执行(本地vs CI)            │ │
│  │  • lockfile出生证明: 每个锁定条目附带注册历史快照哈希, 漂移即重建审查        │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L4: 幻觉施工熔断层 (Hallucination Construction Circuit Breaking) ────────┐ │
│  │  • 报错分类: "方法/包不存在"类报错强制进入幻觉核验流程, 禁止直接适配         │ │
│  │  • 适配审批: 为AI生成代码新建适配层/垫片必须附带存在性核验结论               │ │
│  │  • 幽灵功能审计: 季度扫描内部实现中对接外部能力的调用链完整性                │ │
│  │  • 文档URL核验: AI引用的全部文档链接执行404+内容指称双重检查                 │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│       ↓                                                                        │
│  ┌─ L5: 供应链生态归责层 (Supply Chain Ecology Attribution) ─────────────────┐ │
│  │  • AI引入占比普查: 代码库依赖的AI引入率季度审计                             │ │
│  │  • 污染损失归因: 幻觉依赖导致的凭据泄露/审计失败归入AI开发工具生态成本       │ │
│  │  • 工具厂商责任: 生成侧幻觉率与生态侧抢注率的联合SLA框架                    │ │
│  │  • 架构归责: "审查已知的包而不审查不存在的东西为什么存在"的责任框架          │ │
│  └──────────────────────────────────────────────────────────────────────────┘ │
│                                                                                │
└────────────────────────────────────────────────────────────────────────────────┘

"""
hallucination_squat_detection.py - 幻觉依赖与抢注检测引擎
核心原则: "这个包存在"不等于"这个包应该存在"——
          存在性检查是传统供应链防御的第一关,
          而slopsquatting恰恰把这一关变成了攻击的入口:
          包是真的, 功能是正的, README是流畅的,
          唯一假的是它出生的理由——
          没有任何人类需求驱动过这个名字,
          它被注册, 是因为一台模型会以20%的概率说出它;
          如果幻觉在训练分布里是高频模式,
          那攻击者不需要猜你会要什么,
          他只需要用你的模型跑一遍, 就拿到了
          你的供应链未来三年的引入清单;
          如果报错的文化解读是"我的代码有问题",
          那组织会为每个幻觉配一支施工队——
          十八个月, 四十万行, 完整测试, 唯一不存在的
          是它对接的现实;
          幻觉依赖检测最反直觉的地方在于:
          你要审计的不是已安装的包,
          而是"还没被安装、但你的模型一定会推荐"的名字——
          防御必须发生在幻觉被说出口之前,
          因为在它被说出口之后,
          世界的另一端已经有人替你把包写好了
"""
from typing import Dict, List, Any, Optional, Tuple, Set
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict, Counter
import time, uuid, json, hashlib, re
import numpy as np

class DependencyOrigin(str, Enum):
    HUMAN_INTRODUCED = "human"        # 人类需求驱动引入
    AI_RECOMMENDED = "ai_recommended" # AI生成代码引入(强制增强核验)
    AI_ADAPTED = "ai_adapted"         # 为迎合AI幻觉新建的适配(最高危)
    UNKNOWN = "unknown"               # 引入血缘不明(视同AI)

class SquatFeature(str, Enum): jiyi.tongsou.com 
    POST_CUTOFF_REGISTRATION = "post_cutoff"    # 注册晚于模型训练截止
    FRESH_ACCOUNT = "fresh_account"             # 注册账户年龄<30天
    LLM_README = "llm_readme"                   # README由LLM生成
    VERSION_JUMP = "version_jump"               # 首版直接≥1.0.0伪装成熟
    NAME_PATTERN_MATCH = "name_pattern"         # 命中幻觉命名模式
    NO_ISSUE_HISTORY = "no_history"             # 零issue/零下载历史突增
    MULTI_ECOSYSTEM = "multi_ecosystem"         # 同名跨生态同步注册
    CONDITIONAL_PAYLOAD = "conditional_payload" # CI条件触发载荷特征

class VerdictState(str, Enum):
    VERIFIED = "verified"            # 存在性+出生证明双通过
    SUSPECTED_SQUAT = "suspected"    # 抢注嫌疑(阻断+上报)
    HALLUCINATION = "hallucination"  # 不存在(阻断+记录幻觉日志)
    GHOST_API = "ghost_api"          # 幽灵API调用
    UNKNOWN = "unknown"

@dataclass
class DependencyProfile:
    """依赖出生证明档案"""
    dep_id: str = field(default_factory=lambda: f"dep-{uuid.uuid4().hex[:10]}")
    ecosystem: str = ""              # pypi / npm / maven
    package_name: str = ""
    # 存在性
    exists: bool = False
    # 出生证明
    first_registered_at: float = 0.0
    registrant_account_age_days: float = 0.0
    first_version: str = ""
    readme_llm_score: float = 0.0    # README的LLM生成置信度
    download_history: List[float] = field(default_factory=list)
    issue_history_count: int = 0
    # 血缘
    origin: DependencyOrigin = DependencyOrigin.UNKNOWN
    introduced_by_commit: str = ""
    introduced_at: float = 0.0
    ai_generation_confidence: float = 0.0  # AI推荐时的自信度(若可测)
    # 判定
    verdict: VerdictState = VerdictState.UNKNOWN
    squat_features: List[SquatFeature] = field(default_factory=list)
    risk_score: float = 0.0

@dataclass
class HallucinationPrediction:
    """幻觉预测审计结果"""
    prediction_id: str = field(default_factory=lambda: f"hp-{uuid.uuid4().hex[:10]}")
    model_id: str = ""
    prompts_run: int = 0
    hallucinated_names: List[str] = field(default_factory=list)
    name_frequency: Dict[str, int] = field(default_factory=dict)
    already_registered: List[str] = field(default_factory=list)  # 幻觉名已被注册
    registration_owners_suspicious: List[str] = field(default_factory=list)
    predicted_at: float = field(default_factory=time.time)

@dataclass
class SquatEvidence:
    """抢注证据"""
    evidence_id: str = field(default_factory=lambda: f"sq-{uuid.uuid4().hex[:10]}")
    package_name: str = ""
    ecosystem: str = ""
    features: List[SquatFeature] = field(default_factory=list)
    detail: str = ""
    severity: str = "critical"
    affected_repos: List[str] = field(default_factory=list)
    detected_at: float = field(default_factory=time.time)

# 幻觉命名模式: 模型高频合成的名称结构
HALLUCINATION_NAME_PATTERNS = [
    r"^[a-z]+-(utils|helpers|tools|kit|lib|core|common)$",
    r"^(flask|django|fastapi|express|react|vue|spring)-[a-z]+-(auth|jwt|security|crypto|cache|queue)$",
    r"^[a-z]+-(openssl|crypto|hash|sign)-[a-z]+(utils|tools)?$",
    r"^(python|node|java|go)-[a-z]+-(sdk|client|wrapper|binding)$",
    r"^[a-z]{3,12}-rest-(api|client|auth)(-[a-z]+)?$",
]

class HallucinationSquatEngine:
    """幻觉依赖与抢注检测引擎"""

    # 配置
    MODEL_TRAINING_CUTOFF = 1735689600.0   # 2025-01-01(示例基准)
    FRESH_ACCOUNT_DAYS = 30
    LLM_README_ALERT = 0.80
    SQUAT_FEATURE_ALERT = 2          # ≥2个抢注特征即判定嫌疑
    AI_ORIGIN_ENHANCED_AUDIT = True  # AI引入强制增强核验
    HALLUCINATION_LOG_MANDATORY = True
    PREDICTION_PROMPTS_MIN = 500     # 幻觉预测最小提示量
    CROSS_ECOSYSTEM_SYNC_DAYS = 7    # 跨生态同名同步注册窗口

    def __init__(self, registry_service, alerts, audit, metrics):
        self.registry = registry_service   # 包注册局查询服务
        self.alerts = alerts
        self.audit = audit
        self.metrics = metrics
        self.profiles: Dict[str, DependencyProfile] = {}
        self.blacklist: Set[str] = set()   # 跨生态共享黑名单
        self.evidences: List[SquatEvidence] = []
        self.hallucination_log: List[Dict[str, Any]] = []

    async def verify_dependency(self, profile: DependencyProfile,
                                 repo_id: str = ""
                                 ) -> DependencyProfile:
        """依赖核验总闸: 存在性+出生证明+血缘"""

        name = profile.package_name

        # ===== 阶段1: 存在性 =====
        if not profile.exists:
            profile.verdict = VerdictState.HALLUCINATION
            self.hallucination_log.append({
                "ts": time.time(), "name": name,
                "ecosystem": profile.ecosystem, "repo": repo_id,
                "origin": profile.origin.value
            })
            if self.HALLUCINATION_LOG_MANDATORY:
                await self.alerts.warning(
                    f"👻 HALLUCINATED PACKAGE: '{name}' "
                    f"({profile.ecosystem}) recommended by "
                    f"{profile.origin.value} code in repo '{repo_id}' "
                    f"— DOES NOT EXIST. Install blocked. Logged to "
                    f"hallucination ledger. "
                    f"Today this name is a 404. The ledger exists "
                    f"because 20% of hallucinated names eventually "
                    f"stop being 404s— not because the model got "
                    f"better, but because someone registered what it "
                    f"kept saying. This log entry is a standing "
                    f"prediction: watch this name."
                )
            await self.audit.log_hallucination(profile)
            return profile

        # ===== 阶段2: 出生证明审计 =====
        features = []

        if profile.first_registered_at > self.MODEL_TRAINING_CUTOFF:
            features.append(SquatFeature.POST_CUTOFF_REGISTRATION)
        if profile.registrant_account_age_days < self.FRESH_ACCOUNT_DAYS:
            features.append(SquatFeature.FRESH_ACCOUNT)
        if profile.readme_llm_score > self.LLM_README_ALERT:
            features.append(SquatFeature.LLM_README)
        if self._is_version_jump(profile.first_version):
            features.append(SquatFeature.VERSION_JUMP)
        if self._matches_hallucination_pattern(name):
            features.append(SquatFeature.NAME_PATTERN_MATCH)
        if self._download_anomaly(profile.download_history):
            features.append(SquatFeature.NO_ISSUE_HISTORY)

        profile.squat_features = features
        profile.risk_score = self._risk_score(features, profile)

        if len(features) >= self.SQUAT_FEATURE_ALERT:
            profile.verdict = VerdictState.SUSPECTED_SQUAT
            self.blacklist.add(f"{profile.ecosystem}:{name}")

            evidence = SquatEvidence(
                package_name=name, ecosystem=profile.ecosystem,
                features=features,
                detail=(f"Registered {profile.first_registered_at}, "
                        f"account age {profile.registrant_account_age_days:.0f}d, "
                        f"first version {profile.first_version}, "
                        f"README LLM score {profile.readme_llm_score:.2f}, "
                        f"risk {profile.risk_score:.2f}"),
                affected_repos=[repo_id] if repo_id else []
            )
            self.evidences.append(evidence)

            await self.alerts.critical(
                f"🚨 SUSPECTED SLOPSQUAT: '{name}' ({profile.ecosystem}). "
                f"Features: {[f.value for f in features]}. "
                f"{evidence.detail}. "
                f"BLACKLISTED across all repos; install blocked at "
                f"resolution time. "
                f"This package passes every traditional check: it "
                f"exists, it installs, its stated functions work. "
                f"What it cannot explain is its BIRTH— no human need "
                f"ever searched for this name; a model invented it, "
                f"and someone was waiting at the name the model "
                f"invents most often. The settlement-crypto-utils "
                f"chain had zero human error end to end: the AI "
                f"recommended confidently, the developer trusted "
                f"reasonably, CI installed dutifully, and the "
                f"post-install script collected credentials that no "
                f"ring-fence had ever separated from the build."
            )
            await self.audit.log_squat_evidence(evidence)
        else:
            profile.verdict = VerdictState.VERIFIED

        # ===== 阶段3: AI引入血缘增强 =====
        if self.AI_ORIGIN_ENHANCED_AUDIT and \
           profile.origin in (DependencyOrigin.AI_RECOMMENDED,
                              DependencyOrigin.AI_ADAPTED,
                              DependencyOrigin.UNKNOWN):
            profile.risk_score += 0.25
            if profile.origin == DependencyOrigin.AI_ADAPTED:
                await self.alerts.critical(
                    f"🛑 ADAPTATION-LAYER DEPENDENCY: '{name}' was "
                    f"introduced to make AI-generated code RUN, not to "
                    f"fulfill a stated requirement. "
                    f"Adapted dependencies are the sediment of "
                    f"hallucination construction— the org modifying "
                    f"reality to match what the model asserted. "
                    f"Every one deserves the question the FDA asked: "
                    f"'show me the external capability this adapts "
                    f"TO.' If the answer is a model's confidence, "
                    f"the dependency is a monument, not a component."
                )

        self.profiles[f"{profile.ecosystem}:{name}"] = profile
        await self.audit.log_dependency_profile(profile)
        return profile

    @staticmethod
    def _is_version_jump(first_version: str) -> bool:
        """首版跳号检测: 直接≥1.0.0伪装成熟度"""
        m = re.match(r"^(\d+)\.", first_version or "0")
        return bool(m and int(m.group(1)) >= 1)

    @staticmethod
    def _matches_hallucination_pattern(name: str) -> bool:
        return any(re.match(p, name) for p in HALLUCINATION_NAME_PATTERNS)

    @staticmethod
    def _download_anomaly(history: List[float]) -> bool:
        """下载历史异常: 零历史突增(安装即爆发)"""
        if len(history) < 4: zhaixing.tongsou.com 
            return bool(history) and history[-1] > 100
        baseline = float(np.mean(history[:-1]))
        return history[-1] > max(baseline * 50, 500)

    @staticmethod
    def _risk_score(features: List[SquatFeature],
                    profile: DependencyProfile) -> float:
        weights = {
            SquatFeature.POST_CUTOFF_REGISTRATION: 0.30,
            SquatFeature.FRESH_ACCOUNT: 0.20,
            SquatFeature.LLM_README: 0.15,
            SquatFeature.VERSION_JUMP: 0.10,
            SquatFeature.NAME_PATTERN_MATCH: 0.15,
            SquatFeature.NO_ISSUE_HISTORY: 0.10,
            SquatFeature.MULTI_ECOSYSTEM: 0.25,
            SquatFeature.CONDITIONAL_PAYLOAD: 0.40,
        }
        return min(sum(weights.get(f, 0.1) for f in features), 1.0)

    async def run_hallucination_prediction(self, model_id: str,
                                            generation_results: List[Dict[str, Any]],
                                            registry_snapshot: Dict[str, float]
                                            ) -> HallucinationPrediction:
        """幻觉预测审计: 自模型批量生成→高频幻觉名→注册局比对
        generation_results: [{"prompt_ctx":..., "recommended_packages":[...]}]
        registry_snapshot: {package_name: first_registered_ts}"""

        name_counter = Counter(xunling.tongsou.com)
        real_names = set()  # 由注册局确认存在的推荐名

        for r in generation_results:
            for pkg in r.get("recommended_packages", []):
                name_counter[pkg] += 1

        hallucinated, registered_hallucinations = [], []
        for name, freq in name_counter.items():
            if freq >= 3:   # 跨提示高频出现=分布级幻觉
                if name in registry_snapshot:
                    registered_hallucinations.append(name)
                else:
                    hallucinated.append(name)

        prediction = HallucinationPrediction(
            model_id=model_id, prompts_run=len(generation_results),
            hallucinated_names=hallucinated[:200],
            name_frequency=dict(name_counter.most_common(200)),
            already_registered=registered_hallucinations)

        # 已注册幻觉名的持有者审查
        suspicious = []
        for name in registered_hallucinations: zhuaci.tongsou.com 
            reg_ts = registry_snapshot[name]
            if reg_ts > self.MODEL_TRAINING_CUTOFF:
                suspicious.append(name)
        prediction.registration_owners_suspicious = suspicious

        if suspicious: maifushi.tongsou.com 
            await self.alerts.critical(
                f"🚨 HALLUCINATION LEDGER — ACTIVE SQUATS FOUND: "
                f"Model '{model_id}' across {len(generation_results)} "
                f"prompts reliably hallucinates "
                f"{len(registered_hallucinations)} package names that "
                f"NOW EXIST in registries; {len(suspicious)} were "
                f"registered AFTER the training cutoff: {suspicious[:10]}. "
                f"These are not discoveries. They are CONFIRMATIONS— "
                f"the attacker ran your model before you did and "
                f"stood at the names it repeats. "
                f"Every name on this list is blacklisted org-wide "
                f"and shared to the cross-enterprise ledger TODAY. "
                f"The unregistered {len(hallucinated)} names are not "
                f"relief— they are a countdown with the clock "
                f"visible only to whoever is watching the same "
                f"frequency table you just computed."
            )

        await self.audit.log_hallucination_prediction(prediction)
        return prediction

    async def detect_ghost_api(self, repo_id: str,
                                generated_calls: List[Dict[str, Any]],
                                sdk_manifests: Dict[str, Set[str]]
                                ) -> List[Dict[str, Any]]:
        """幽灵API检测: 生成代码的外部调用与官方SDK清单比对
        generated_calls: [{"call": "patient_record.merge_across_sites()",
                           "sdk": "partner_platform", "line":...}]"""

        ghosts = []
        for call in generated_calls: moli.tongsou.com 
            sdk = call.get("sdk", "")
            method = call.get("call", "")
            known = sdk_manifests.get(sdk, set())
            if known and method not in known:
                ghosts.append({"repo": repo_id, "sdk": sdk,
                               "method": method, "line": call.get("line")})

        if ghosts: zhendao.tongsou.com 
            await self.alerts.critical(
                f"👻 GHOST API CALLS: Repo '{repo_id}'. "
                f"{len(ghosts)} calls to methods absent from official "
                f"SDK manifests: "
                f"{[(g['method'], g['sdk']) for g in ghosts[:5]]}. "
                f"COMPILATION FAILURE IS NOT A BUG TICKET— it is a "
                f"hallucination signal demanding verification. "
                f"The fork in the road: delete the call (the model "
                f"was wrong) or build an adapter (make the model "
                f"right). Under deadline pressure the second path is "
                f"always faster, and one medical device company took "
                f"it for eighteen months— 400K lines, full test "
                f"coverage, internal docs, a complete implementation "
                f"of a method that existed only in a decoder's "
                f"confidence. The regulator's question found the "
                f"hole instantly: 'show me the platform capability "
                f"this integrates with.' Organizations do not need "
                f"better prompts. They need the error message to "
                f"mean 'verify existence' instead of 'write more "
                f"code'."
            )

        await self.audit.log_ghost_apis(repo_id, ghosts)
        return ghosts

    async def share_blacklist(self) -> Dict[str, Any]:
        """跨企业黑名单与幻觉清单共享"""

        payload = {
            "blacklist_entries": sorted(self.blacklist),
            "hallucination_ledger": [h["name"] for h in
                                     self.hallucination_log[-5000:]],
            "squat_evidences": [
                {"pkg": e.package_name, "eco": e.ecosystem,
                 "features": [f.value for f in e.features]}
                for e in self.evidences[-100:]],
            "shared_at": time.time()
        }

        await self.alerts.info(
            f"🤝 BLACKLIST SHARED: {len(payload['blacklist_entries'])} "
            f"entries + {len(payload['hallucination_ledger'])} predicted "
            f"hallucination names pushed to cross-enterprise ledger. "
            f"Hallucination distributions are SHARED across "
            f"organizations using similar models— which makes "
            f"defense a commons problem: your prediction audit "
            f"protects your competitors' pipelines, and theirs "
            f"protects yours. The attacker already treats the "
            f"frequency table as public infrastructure."
        )
        await self.audit.log_blacklist_share(payload)
        return payload

"""
build_isolation_defense.py - 构建时隔离与出生证明防御引擎
核心: 幻觉依赖的杀伤力不在推荐的那一刻, 在安装的那一刻——
      推荐只是字符串, 安装是执行;
      而整个行业的构建管线, 把"下载并立即以构建权限执行
      陌生人的代码"当作默认配置运行了三十年,
      唯一的更新是: 现在推荐陌生人的, 是你自己付费的模型;
      你不能通过"更准的幻觉检测"守住安装时刻,
      因为检测有漏报而抢注在进化——
      你能做的是让"装上了"不等于"能伤到":
      安装脚本默认全禁, 让post-install这个最大的免费执行面关闭;
      构建环境零凭据+网络分区, 让收集到的环境变量一文不值;
      本地与CI差分执行, 让只在构建机上醒来的载荷显形;
      lockfile绑定出生证明快照, 让每个锁定条目回答
      "你被锁定的那天, 你是谁";
      防御的本质不是阻止幻觉被安装,
      是让安装这个动作失去兑现伤害的权力——
      而在效率至上的DevOps文化里,
      最贵的不是隔离架构的改造,
      是承认这句运维老话的新版本:
      你的构建环境曾经只需要防坏人,
      现在它需要防自己的助手
"""
from typing import Dict, List, Any, Optional, Tuple, Set
from enum import Enum
from dataclasses import dataclass, field
from collections import defaultdict, Counter
import time, uuid, json, hashlib
import numpy as np

class IsolationViolation(str, Enum): weimeng.tongsou.com 
    INSTALL_SCRIPT_ENABLED = "install_script"    # 安装脚本未禁用
    CREDENTIAL_IN_BUILD = "credential_in_build"  # 构建环境含长期凭据
    NETWORK_UNSEGMENTED = "network_unsegmented"  # 安装区可访问密钥网络
    DIFF_BEHAVIOR_DETECTED = "diff_behavior"     # 本地/CI行为差异
    LOCKFILE_DRIFT = "lockfile_drift"            # 出生证明快照漂移
    UNPINNED_RANGE = "unpinned_range"            # 版本范围未锁定

class CircuitAction(str, Enum):
    ALLOW = "allow"
    SANDBOX_ONLY = "sandbox"          # 仅隔离沙箱可安装
    BLOCK = "block"
    REBUILD_REVIEW = "rebuild_review" # lockfile重建审查

@dataclass
class BuildEnvironmentProfile:
    """构建环境隔离档案"""
    env_id: str = field(default_factory=lambda: f"be-{uuid.uuid4().hex[:10]}")
    repo_id: str = ""
    install_scripts_allowed: bool = False
    script_whitelist: List[str] = field(default_factory=list)
    long_lived_credentials: int = 0     # 环境内长期凭据数量(目标0)
    network_segmented: bool = False     # 安装区与密钥网络分区
    egress_policy: str = ""             # 出站白名单策略
    last_audit: float = 0.0

@dataclass
class LockfileEntry:
    """lockfile条目+出生证明"""
    ecosystem: str = ""
    package_name: str = ""
    pinned_version: str = ""
    # 出生证明快照哈希: 锁定时刻的注册历史指纹
    birth_certificate_hash: str = ""
    locked_at: float = 0.0
    origin: str = "unknown"             # human / ai_recommended
    integrity_hash: str = ""            # 包内容哈希

@dataclass
class DiffExecutionResult: aisou.tongsou.com 
    """本地/CI差分执行结果"""
    diff_id: str = field(default_factory=lambda: f"dx-{uuid.uuid4().hex[:10]}")
    package_name: str = ""
    local_behavior_hash: str = ""
    ci_behavior_hash: str = ""
    divergence: float = 0.0
    ci_only_network_calls: List[str] = field(default_factory=list)
    ci_only_env_reads: List[str] = field(default_factory=list)
    conditional_payload: bool = False
    executed_at: float = field(default_factory=time.time)

@dataclass
class SupplyChainCensus:
    """AI引入依赖季度普查"""
    census_id: str = field(default_factory=lambda: f"sc-{uuid.uuid4().hex[:10]}")
    org_id: str = ""
    total_deps: int = 0
    ai_introduced: int = 0
    ai_share: float = 0.0
    ai_verified_rate: float = 0.0       # AI引入中通过双重核验的比例
    adaptation_layer_deps: int = 0      # 幻觉施工型依赖
    censused_at: float = field(default_factory=time.time)

class BuildIsolationEngine:
    """构建时隔离与出生证明防御引擎"""

    # 配置
    INSTALL_SCRIPTS_DEFAULT_DENY = True
    CREDENTIAL_COUNT_CEILING = 0        # 构建环境长期凭据上限: 零
    NETWORK_SEGMENTATION_MANDATORY = True
    DIFF_DIVERGENCE_ALERT = 0.10        # 行为差异>10%即条件载荷嫌疑
    BIRTH_DRIFT_BLOCK = True            # 出生证明漂移即重建审查
    AI_SHARE_CENSUS_QUARTERLY = True
    AI_VERIFIED_FLOOR = 0.95            # AI引入依赖核验率下限

    def __init__(self, squat_engine, alerts, audit, metrics):
        self.squat = squat_engine
        self.alerts = alerts
        self.audit = audit
        self.metrics = metrics
        self.env_profiles: Dict[str, BuildEnvironmentProfile] = {}
        self.lockfiles: Dict[str, List[LockfileEntry]] = defaultdict(list)
        self.censuses: List[SupplyChainCensus] = []

    async def audit_build_environment(self, profile: BuildEnvironmentProfile
                                       ) -> BuildEnvironmentProfile:
        """构建环境隔离审计"""

        self.env_profiles[profile.repo_id] = profile
        violations = []

        if profile.install_scripts_allowed and \
           self.INSTALL_SCRIPTS_DEFAULT_DENY:
            violations.append(IsolationViolation.INSTALL_SCRIPT_ENABLED)
        if profile.long_lived_credentials > self.CREDENTIAL_COUNT_CEILING:
            violations.append(IsolationViolation.CREDENTIAL_IN_BUILD)
        if not profile.network_segmented and self.NETWORK_SEGMENTATION_MANDATORY:
            violations.append(IsolationViolation.NETWORK_UNSEGMENTED)

        if violations:
            await self.alerts.critical(
                f"🚨 BUILD ISOLATION VIOLATIONS: Repo '{profile.repo_id}'. "
                f"{[v.value for v in violations]}. "
                f"Install scripts: {profile.install_scripts_allowed}. "
                f"Long-lived credentials in build env: "
                f"{profile.long_lived_credentials}. "
                f"Network segmented: {profile.network_segmented}. "
                f"The settlement chain did not exploit a "
                f"vulnerability— it used the DEFAULT CONFIGURATION: "
                f"pip and npm grant every downloaded package the "
                f"right to execute arbitrary code at install time, "
                f"with the build machine's full environment, on a "
                f"network that reaches your secrets. For thirty years "
                f"this was 'how packaging works'. Now the entity "
                f"choosing what gets downloaded is a model that "
                f"invents 20% of its suggestions, and a fifth of "
                f"those inventions are already owned by someone "
                f"else. Deny-by-default on install scripts, zero "
                f"standing credentials, segmented egress— not "
                f"because a specific package is malicious, but "
                f"because 'install' must stop meaning 'execute "
                f"anything with everything'."
            )

        await self.audit.log_build_environment(profile, violations)
        return profile

    async def gate_install(self, repo_id: str, package_name: str,
                            ecosystem: str,
                            has_install_script: bool
                            ) -> Dict[str, Any]:
        """安装闸门: 黑名单+脚本+环境联合判定"""

        key = f"{ecosystem}:{package_name}"
        env = self.env_profiles.get(repo_id)

        if key in self.squat.blacklist:
            return {"action": CircuitAction.BLOCK.value,
                    "reason": "blacklisted (suspected slopsquat / "
                              "confirmed malicious)"}

        profile = self.squat.profiles.get(key)
        if profile and profile.verdict == VerdictState.SUSPECTED_SQUAT:
            return {"action": CircuitAction.BLOCK.value,
                    "reason": f"squat features: "
                              f"{[f.value for f in profile.squat_features]}"}

        if has_install_script: toujing.tongsou.com 
            whitelisted = env and package_name in env.script_whitelist
            if not whitelisted:
                return {"action": CircuitAction.SANDBOX_ONLY.value,
                        "reason": "install script present, not "
                                  "whitelisted — sandboxed execution "
                                  "with network deny + credential zero"}

        return {"action": CircuitAction.ALLOW.value}

    async def diff_execute(self, package_name: str,
                            local_result: DiffExecutionResult,
                            ci_result: DiffExecutionResult
                            ) -> DiffExecutionResult:
        """差分执行: 本地vs CI行为比对(条件载荷检测)"""

        # 行为哈希差异
        divergence = 0.0
        if local_result.local_behavior_hash != ci_result.ci_behavior_hash:
            divergence = 1.0
        # 精细: CI独有的网络调用与环境读取
        ci_only_net = [c for c in ci_result.ci_only_network_calls
                       if c not in local_result.ci_only_network_calls]
        ci_only_env = [e for e in ci_result.ci_only_env_reads
                       if e not in local_result.ci_only_env_reads]

        merged = DiffExecutionResult(
            package_name=package_name,
            local_behavior_hash=local_result.local_behavior_hash,
            ci_behavior_hash=ci_result.ci_behavior_hash,
            divergence=max(divergence,
                           min(len(ci_only_net) * 0.2 + len(ci_only_env) * 0.3, 1.0)),
            ci_only_network_calls=ci_only_net,
            ci_only_env_reads=ci_only_env,
            conditional_payload=bool(ci_only_env) or divergence > self.DIFF_DIVERGENCE_ALERT)

        if merged.conditional_payload:
            await self.alerts.critical(
                f"🚨 CONDITIONAL PAYLOAD DETECTED: '{package_name}'. "
                f"Behavior diverges between local sandbox and CI "
                f"context (divergence {merged.divergence:.2f}). "
                f"CI-only network calls: {ci_only_net[:5]}. "
                f"CI-only env reads: {ci_only_env[:5]}. "
                f"QUARANTINE + blacklist. "
                f"This is the payload class that defeats every "
                f"local review: dormant on the developer's machine, "
                f"awake on the build server— because the build "
                f"server is where the credentials, the signing keys "
                f"and the production topology live. The three "
                f"conditional-trigger packages in last year's census "
                f"all keyed on CI environment variables. "
                f"A package that behaves differently in CI is not "
                f"environment-sensitive. It is target-selective."
            )

        await self.audit.log_diff_execution(merged)
        return merged

    async def certify_lockfile(self, repo_id: str,
                                entries: List[LockfileEntry],
                                registry_lookup: Dict[str, Dict[str, Any]]
                                ) -> Dict[str, Any]:
        """lockfile出生证明: 快照哈希绑定与漂移审查"""

        results = {"repo_id": repo_id, "certified": [], "drifted": [],
                   "unpinned": []}

        for e in entries: qiyin.tongsou.com 
            # 版本范围检测
            if any(c in e.pinned_version for c in ("^", "~", "*", ">", "<")):
                results["unpinned"].append(e.package_name)
                continue

            # 出生证明哈希: 注册历史指纹
            reg = registry_lookup.get(e.package_name, {})
            current_hash = hashlib.sha256(json.dumps({
                "first_registered": reg.get("first_registered_at", 0),
                "registrant": reg.get("registrant", ""),
                "version_count": reg.get("version_count", 0),
                "yanked": reg.get("yanked_versions", [])
            }, sort_keys=True).encode()).hexdigest()[:16]

            if not e.birth_certificate_hash: hongdong.tongsou.com 
                # 首次锁定: 铸造出生证明
                e.birth_certificate_hash = current_hash
                e.locked_at = time.time()
                results["certified"].append(e.package_name)
            elif e.birth_certificate_hash != current_hash and \
                 self.BIRTH_DRIFT_BLOCK:
                results["drifted"].append({
                    "package": e.package_name,
                    "locked_hash": e.birth_certificate_hash,
                    "current_hash": current_hash,
                    "delta": reg
                })

        if results["drifted"]: hanzhi.tongsou.com 
            await self.alerts.critical(
                f"🚨 LOCKFILE BIRTH-CERTIFICATE DRIFT: Repo "
                f"'{repo_id}'. {len(results['drifted'])} pinned "
                f"packages have registration histories that CHANGED "
                f"since locking: "
                f"{[(d['package'], d['delta'].get('registrant','')) for d in results['drifted'][:5]]}. "
                f"REBUILD REVIEW triggered — a pinned version is a "
                f"promise about content, and the birth certificate is "
                f"a promise about PROVENANCE: same name, same "
                f"version, new owner (account transfer, maintainer "
                f"handover, namespace resurrection) is the classic "
                f"post-trust hijack. Your pin says 'this content'. "
                f"The certificate asks 'is this still the entity you "
                f"vetted?' Lockfiles that only hash artifacts "
                f"verify the package. Lockfiles that hash histories "
                f"verify the birth."
            )

        self.lockfiles[repo_id] = entries
        await self.audit.log_lockfile_certification(results)
        return results

    async def census_ai_dependencies(self, org_id: str,
                                      dep_records: List[Dict[str, Any]]
                                      ) -> SupplyChainCensus:
        """AI引入依赖季度普查"""

        total = len(dep_records)
        ai = [d for d in dep_records
              if d.get("origin") in ("ai_recommended", "ai_adapted", "unknown")]
        ai_verified = [d for d in ai if d.get("dual_verified")]
        adapted = [d for d in dep_records if d.get("origin") == "ai_adapted"]

        census = SupplyChainCensus(
            org_id=org_id, total_deps=total, ai_introduced=len(ai),
            ai_share=len(ai) / max(total, 1),
            ai_verified_rate=len(ai_verified) / max(len(ai), 1),
            adaptation_layer_deps=len(adapted))
        self.censuses.append(census)

        if census.ai_verified_rate < self.AI_VERIFIED_FLOOR:
            await self.alerts.critical(
                f"🚨 AI DEPENDENCY VERIFICATION DEFICIT: Org "
                f"'{org_id}'. AI-introduced dependencies: "
                f"{census.ai_share:.0%} of {total} total "
                f"({census.ai_introduced} packages). Dual-verified: "
                f"only {census.ai_verified_rate:.0%} "
                f"(floor {self.AI_VERIFIED_FLOOR:.0%}). "
                f"Adaptation-layer dependencies: "
                f"{census.adaptation_layer_deps}. "
                f"39% of enterprises cannot answer 'how many of our "
                f"dependencies did AI introduce'— and the ones that "
                f"can are finding 38%. The unverified remainder is "
                f"not a paperwork gap: every unoverified AI-origin "
                f"package is a name your model may have invented and "
                f"someone else may have registered. "
                f"You are running an unmeasured experiment in "
                f"letting a stochastic recommender curate your "
                f"attack surface."
            )

        await self.audit.log_supply_chain_census(census)
        return census

    async def attribute_supply_chain_incident(self, org_id: str,
                                               incident_description: str,
                                               loss_usd: float
                                               ) -> Dict[str, Any]:
        """供应链生态归责"""

        attribution = {
            "org_id": org_id,
            "incident": incident_description,
            "loss_usd": loss_usd,
            "primary_accountable": "build_pipeline_architect",
            "rationale": "",
            "fixes": []
        }

        attribution["rationale"] = (
            "No developer was negligent. No reviewer was asleep. The "
            "AI assistant performed exactly as advertised— "
            "confidently. The registry performed exactly as "
            "designed— anyone may register any unclaimed name. The "
            "build system performed exactly as configured— download "
            "and execute with everything. Three correct components, "
            "one catastrophic composition, because the trust model "
            "connecting them was built for a world where package "
            "names originate from human need. They no longer do. "
            "They originate from a probability distribution that an "
            "adversary can sample as easily as you can. The "
            "pipeline architect certified tools, registries and "
            "builds separately and never once asked the only "
            "question that mattered: when the recommender is "
            "stochastic and the namespace is open, what stops "
            "'installed' from meaning 'compromised'?"
        )

        attribution["fixes"] = [
            "Dual-verify every AI-recommended dependency: existence AND birth certificate",
            "Run quarterly hallucination prediction audits with the production model; blacklist and defensively register the ledger",
            "Deny install scripts by default; whitelist individually with review",
            "Zero standing credentials in build environments; segment dependency-install egress from secret networks",
            "Diff-execute suspicious packages: local sandbox vs CI context; divergence means target-selection",
            "Bind lockfile entries to registration-history hashes; drift triggers rebuild review",
            "Treat 'method/package not found' errors as hallucination signals requiring verification, never as tickets for adapters"
        ]

        await self.alerts.critical(
            f"🔍 SUPPLY CHAIN ECOLOGY ATTRIBUTED: Org '{org_id}'. "
            f"Incident: {incident_description[:100]}. "
            f"Loss: {loss_usd:,.0f}. "
            f"Accountable: BUILD PIPELINE ARCHITECT. "
            f"{attribution['rationale'][:220]} "
            f"'A malicious package got in' is the wrong sentence. "
            f"The right sentence: 'we connected a stochastic "
            f"recommender to an open namespace with an "
            f"execute-on-install pipeline, and measured our "
            f"security by how many KNOWN-BAD packages we "
            f"caught— while the attack arrives as "
            f"never-before-seen, because it was invented "
            f"ninety seconds ago by our own tooling.'"
        )

        await self.audit.log_supply_chain_attribution(attribution)
        return attribution