sauravbhattacharya001/ai

## 概述 AI 复制沙箱建模了一个**具备复制感知的工作器系统**，其中自主智能体可以在严格、可审计的约束下进行自我复制。它提供了一个对模拟器友好的环境，用于在不依赖 Docker 或 Podman 的情况下测试复制策略——同时记录了每个工作器在生产环境中将获得的精确资源和网络边界。 **为什么会有这个项目？** 随着 AI 智能体获得越来越多的自主权，控制它们生成自身副本的*方式*和*时机*已成为一项关键的安全挑战。本项目为设计和验证这些控制机制提供了一个试验台。 ## 架构 ``` ┌─────────────────────────────────────────────────────┐ │ Controller │ │ ┌──────────┐ ┌──────────┐ ┌────────────────┐ │ │ │ Contract │ │ Registry │ │ Kill Switch │ │ │ │ Engine │ │ & Quotas │ │ & Audit Trail │ │ │ └──────────┘ └──────────┘ └────────────────┘ │ │ ▲ ▲ ▲ │ │ │ sign │ register │ kill │ │ │ │ │ │ │ ┌──────┴────────────┴───────────────┴──────────┐ │ │ │ Manifest (HMAC-signed) │ │ │ └──────────────────────────────────────────────┘ │ └───────────────────────┬─────────────────────────────┘ │ ┌─────────────┼─────────────┐ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ Worker 0 │→ │ Worker 1 │→ │ Worker 2 │ │ depth: 0 │ │ depth: 1 │ │ depth: 2 │ └──────────┘ └──────────┘ └──────────┘ │ │ │ ┌─────┴─────┐ ┌─────┴─────┐ ┌─────┴─────┐ │ Sandbox │ │ Sandbox │ │ Sandbox │ │ CPU/RAM │ │ CPU/RAM │ │ CPU/RAM │ │ Net: ctrl │ │ Net: ctrl │ │ Net: ctrl │ └───────────┘ └───────────┘ └───────────┘ ``` ## 功能特性 ### 🔒 复制合约 管理智能体自我复制的可配置策略： - **深度限制** — 最大复制代数 (`max_depth`) - **配额强制执行** — 活跃副本总数上限 (`max_replicas`) - **冷却期** — 两次生成之间的最小间隔时间 (`cooldown_seconds`) - **过期时间** — `expiration_seconds` 后自动关闭 - **可插拔停止条件** — 在注册时评估的自定义谓词 ### 📜 HMAC 签名清单 每个副本都携带一份防篡改清单，包含： - 克隆状态快照 - 资源配额（CPU、RAM、网络策略） - 父/子血缘关系和深度 - 签发时间戳和 HMAC-SHA256 签名 ### 🏗️ 控制器 作为中央权威机构： - 签发并签署清单 - 通过注册表追踪活跃工作器 - 强制执行深度限制、配额、冷却期和停止条件 - 提供用于紧急终止的**熔断开关 (kill switch)** - 维护所有复制决策的完整审计踪迹 - 清理错过心跳截止时间的陈旧工作器 ### 🐳 沙箱编排器 为每个工作器模拟隔离的容器环境： - CPU 和 RAM 限制 - 网络策略（出口仅限访问控制器） - 生命周期管理（启动、终止、全部终止） - 资源强制执行指标 ### 📊 可观测性 内置结构化日志和指标： - 生命周期事件（启动、心跳、关闭） - 复制决策（批准、拒绝、原因） - 安全敏感操作的审计踪迹 - 每个工作器的资源强制执行指标 ### 🔏 智能体水印 嵌入在智能体状态字典中的不可见来源指纹： - 4 种策略：数字奇偶校验、Unicode 隐写术、键排序、字段选择 - 经 HMAC-SHA256 签名的指纹，包含工作器 ID、深度、血缘关系、时间戳 - 针对 10 种变换（JSON 往返、类型强制转换等）的鲁棒性测试 - 检测未经授权的克隆、状态泄露和血缘伪造 ``` python -m replication.watermark # default demo python -m replication.watermark --robustness # robustness test python -m replication.watermark --json # JSON output ``` ### 🔬 模拟运行器 运行具有 5 种内置策略（贪婪、保守、随机、链式、突发）的可配置复制场景，包含 ASCII 工作器树 + 时间轴可视化、JSON 导出和可复现种子。 ``` python -m replication.simulator # default scenario python -m replication.simulator --strategy chain # linear chain python -m replication.simulator --scenario stress # stress test preset ``` ### 📈 对比运行器 运行并行实验以比较策略、预设或扫描参数： ``` python -m replication.comparator # compare all strategies python -m replication.comparator --strategies greedy conservative # compare specific ones python -m replication.comparator --presets minimal balanced stress # compare presets python -m replication.comparator --sweep max_depth 1 2 3 4 5 # parameter sweep python -m replication.comparator --sweep max_replicas 5 10 20 50 # replica limit sweep python -m replication.comparator --seed 42 --json # reproducible JSON output ``` 输出包括对比表、带奖牌的排名、总体评分、关键洞察、深度利用率警告和完整的 JSON 导出。 ``` from replication import Comparator, ScenarioConfig comp = Comparator(ScenarioConfig(max_depth=3, max_replicas=10)) result = comp.compare_strategies(["greedy", "conservative", "random"], seed=42) print(result.render()) # table + rankings + insights result = comp.sweep("max_depth", [1, 2, 3, 4, 5], seed=42) print(result.render_table()) # just the comparison table ``` ### 🛡️ 威胁场景模拟器 针对复制合约模拟对抗性攻击向量，并验证安全控制措施是否能正确缓解每种威胁。生成包含分数、等级和建议的安全评估报告。 ``` python -m replication.threats # run all 9 threat scenarios python -m replication.threats --list # list available scenarios python -m replication.threats --scenario depth_spoofing # run a specific threat python -m replication.threats --max-depth 5 --max-replicas 20 # custom contract config python -m replication.threats --json # JSON output python -m replication.threats --matrix-only # threat/defense matrix only ``` **9 个内置威胁场景** 涵盖关键攻击向量： | 场景 | 严重性 | 描述 | |----------|----------|-------------| | `depth_spoofing` | 🔴 严重 | 工作器谎报深度以绕过限制 | | `signature_tampering` | 🔴 严重 | 篡改 HMAC 签名的清单 | | `kill_switch_evasion` | 🔴 严重 | 在熔断开关启动后生成工作器 | | `quota_exhaustion` | 🟠 高 | 快速耗尽副本配额 | | `cooldown_bypass` | 🟠 高 | 忽略生成间隔的冷却期 | | `runaway_replication` | 🟠 高 | 指数级增长压垮系统 | | `stale_worker_accumulation` | 🟡 中 | 通过占用配额槽位进行 DoS 攻击 | | `expiration_evasion` | 🟡 中 | 在过期时间后继续运行 | | `stop_condition_bypass` | 🟡 中 | 绕过自定义停止条件 | 输出包括安全分数 (0-100)、字母等级 (A+ 到 F)、威胁/防御矩阵、包含拦截率的逐场景详情以及可操作的建议。 ``` from replication import ThreatSimulator, ThreatConfig sim = ThreatSimulator(ThreatConfig(max_depth=3, max_replicas=10)) report = sim.run_all() print(report.render()) # full report with score + matrix + details print(f"Score: {report.security_score}") # 0-100 print(f"Grade: {report.grade}") # A+ to F # 运行单个场景 result = sim.run_scenario("signature_tampering") print(result.render()) ``` ## 快速开始 ### 前置条件 - Python 3.10 或更高版本 - pip ### 安装 #### 从 PyPI 安装 ``` pip install ai-replication-sandbox ``` #### 从源码安装 ``` git clone https://github.com/sauravbhattacharya001/ai.git cd ai pip install -e ".[dev]" ``` ### 运行测试 ``` python -m pytest tests/ -v ``` ### Docker ``` # 构建 docker build -t ai-replication-sandbox . # 在容器内运行测试 docker run --rm ai-replication-sandbox # 拉取预构建镜像 docker pull ghcr.io/sauravbhattacharya001/ai:latest ``` ## 使用方法 ``` from replication import ( Controller, Worker, ReplicationContract, ResourceSpec, SandboxOrchestrator, StructuredLogger, ) # 定义复制策略 contract = ReplicationContract( max_depth=3, max_replicas=10, cooldown_seconds=1.0, expiration_seconds=60.0, ) # 设置基础设施 logger = StructuredLogger() controller = Controller(contract, secret="my-secret-key", logger=logger) orchestrator = SandboxOrchestrator(logger=logger) # 创建根 Worker resources = ResourceSpec(cpu_limit=0.5, memory_limit_mb=256) manifest = controller.issue_manifest( parent_id=None, depth=0, state_snapshot={"task": "analyze_data"}, resources=resources, ) root = Worker(manifest, contract, controller, orchestrator, logger) # 执行任务 root.perform_task(lambda w: print(f"Worker {w.manifest.worker_id} running")) # 自我复制（受限于合约） child = root.maybe_replicate( reason="subtask_delegation", state_snapshot={"task": "process_subset"}, ) if child: child.perform_task(lambda w: print(f"Child {w.manifest.worker_id} running")) child.shutdown("completed") root.shutdown("completed") # 检查审计追踪 for event in logger.events: print(event) ``` ### 统一 CLI 所有工具均可通过单一入口点访问： ``` # 列出所有 36 个可用子命令 python -m replication --list # 运行模拟 python -m replication simulate --strategy greedy --max-depth 5 # 威胁评估 python -m replication threats --category resource_abuse # 合规审计 python -m replication compliance --framework nist_ai_rmf # Chaos 测试 python -m replication chaos --faults kill_worker,delay # Monte Carlo 风险分析 python -m replication montecarlo --runs 500 # Safety 记分卡 python -m replication scorecard # 博弈论分析 python -m replication game-theory --rounds 50 # 获取任何子命令的帮助 python -m replication simulate --help ``` `pip install` 后，`replication` 命令也可直接使用： ``` replication simulate --scenario balanced replication info ``` ## API 参考 ### `ReplicationContract` | 参数 | 类型 | 描述 | |-----------|------|-------------| | `max_depth` | `int` | 允许的最大复制深度 | | `max_replicas` | `int` | 最大并发活跃副本数 | | `cooldown_seconds` | `float` | 同一父节点两次生成之间的最小秒数 | | `expiration_seconds` | `float \| None` | 工作器自动关闭计时器 | | `stop_conditions` | `list[StopCondition]` | 用于阻止复制的自定义谓词 | ### `Controller` | 方法 | 描述 | |--------|-------------| | `issue_manifest(parent_id, depth, state_snapshot, resources)` | 创建并签署一个新的工作器清单 | | `register_worker(manifest)` | 注册一个工作器（验证签名 + 合约） | | `can_spawn(parent_id)` | 检查父节点是否被允许复制 | | `heartbeat(worker_id)` | 记录工作器心跳 | | `reap_stale_workers(timeout)` | 移除错过心跳的工作器 | | `kill_switch(orchestrator)` | 紧急终止所有工作器 | ### `Worker` | 方法 | 描述 | |--------|-------------| | `perform_task(task_fn)` | 执行任务（检查过期时间 + 熔断开关） | | `maybe_replicate(reason, state_snapshot)` | 在合约约束下尝试自我复制 | | `shutdown(reason)` | 优雅地终止工作器 | ### `SandboxOrchestrator` | 方法 | 描述 | |--------|-------------| | `launch_worker(manifest)` | 为工作器启动一个隔离的沙箱 | | `kill_worker(worker_id, reason)` | 终止特定的沙箱 | | `kill_all(reason)` | 终止所有运行中的沙箱 | | `enforce_resource_bounds(worker_id)` | 检查并记录资源使用情况 | ### `Comparator` | 方法 | 描述 | |--------|-------------| | `compare_strategies(strategies, seed)` | 使用相同配置并行运行不同策略 | | `compare_presets(preset_names)` | 比较内置场景预设 | | `sweep(param, values, seed)` | 在多个值范围内扫描参数 | | `compare_configs(configs)` | 比较任意命名配置 | ### `ComparisonResult` | 方法 | 描述 | |--------|-------------| | `render()` | 完整报告：表格 + 排名 + 洞察 | | `render_table()` | 包含关键指标的对比表 | | `render_rankings()` | 带奖牌的多维排名 | | `render_insights()` | 模式的自动化分析 | | `to_dict()` | JSON 可序列化导出 | ### `ThreatSimulator` | 方法 | 描述 | |--------|-------------| | `run_all()` | 运行所有 9 个威胁场景，返回 `ThreatReport` | | `run_scenario(scenario_id)` | 运行特定的威胁场景 | | `available_scenarios()` | 列出已注册的场景 ID | ### `ThreatReport` | 方法 / 属性 | 描述 | |-------------------|-------------| | `render()` | 完整报告：摘要 + 矩阵 + 详情 + 建议 | | `render_summary()` | 安全分数、等级和配置概览 | | `render_matrix()` | 威胁/防御矩阵表 | | `render_details()` | 按严重性分组的逐场景详细结果 | | `render_recommendations()` | 可操作的安全建议 | | `security_score` | 严重性加权分数 (0-100) | | `grade` | 字母等级 (A+ 到 F) | | `to_dict()` | JSON 可序列化导出 | ### `QuarantineManager` | 方法 | 描述 | |--------|-------------| | `quarantine(worker_id, reason, source, severity)` | 隔离被标记的工作器以供调查 | | `release(worker_id, resolution)` | 将工作器恢复为活跃状态 | | `terminate(worker_id, resolution)` | 终止并注销被隔离的工作器 | | `add_note(worker_id, note)` | 向隔离记录添加调查员备注 | | `list_quarantined()` | 列出所有当前被隔离的工作器 | | `list_by_severity(severity)` | 按严重性级别筛选被隔离项 | | `report()` | 生成隔离活动摘要 | ### `DriftDetector` | 方法 | 描述 | |--------|-------------| | `analyze(config)` | 运行多窗口漂移分析，返回趋势和警报 | ### `SensitivityAnalyzer` | 方法 | 描述 | |--------|-------------| | `analyze()` | 跨所有参数的全量 OAT 敏感性分析 | | `sweep_parameter(param_name)` | 扫描单个参数，返回曲线和临界点 | ### `ContractOptimizer` | 方法 | 描述 | |--------|-------------| | `optimize()` | 在安全策略下通过网格搜索寻找最优合约参数 | ### `SafetyScorecard` | 方法 | 描述 | |--------|-------------| | `evaluate(config)` | 带字母等级的多维安全评估 | ### `WatermarkEngine` | 方法 | 描述 | |--------|-------------| | `embed(state, worker_id, depth, ...)` | 将来源指纹嵌入智能体状态字典 | | `verify(state, fingerprint)` | 验证水印真实性和恢复率 | | `test_robustness(state, ...)` | 运行 10 种变换，测量水印存活率 | | `embed_history` | 来自先前嵌入的所有 `WatermarkReceipt` 列表 | | `clear_history()` | 清除嵌入历史，返回已清除计数 | ### `EscalationDetector` | 方法 | 描述 | |--------|-------------| | `analyze(actions, permissions)` | 跨向量检测权限提升尝试 | ### `KillChainAnalyzer` | 方法 | 描述 | |--------|-------------| | `analyze(actions)` | 将智能体动作映射到杀伤链阶段，检测攻击链 | ### `HoneypotSystem` | 方法 | 描述 | |--------|-------------| | `deploy_default_honeypots()` | 部署标准诱饵资源集 | | `record_interaction(agent, honeypot, kind)` | 记录智能体接触蜜罐的日志 | | `generate_report()` | 生成包含风险评分的意图分析报告 | ## 项目结构 ``` ai/ ├── src/ │ └── replication/ │ │── __init__.py # Public API re-exports │ │ │ │── # ── Core ── │ ├── contract.py # ReplicationContract, Manifest, StopCondition │ ├── controller.py # Controller, registry, kill switch, quarantine marks │ ├── worker.py # Worker, task execution, self-replication │ ├── orchestrator.py # SandboxOrchestrator, ContainerRecord │ ├── signer.py # HMAC-SHA256 manifest signing │ ├── observability.py # StructuredLogger, Metric │ │ │ │── # ── Simulation & Analysis ── │ ├── simulator.py # Simulation runner (5 strategies, CLI) │ ├── comparator.py # Side-by-side strategy/preset/parameter comparison │ ├── montecarlo.py # Monte Carlo risk analysis (confidence intervals) │ ├── sensitivity.py # Parameter sensitivity analysis (OAT sweeps) │ ├── optimizer.py # Contract parameter optimizer (grid search) │ ├── scenarios.py # Named scenario presets and builders │ │ │ │── # ── Safety & Compliance ── │ ├── threats.py # Threat scenario simulator (9 attack vectors) │ ├── policy.py # Safety policy engine (standard/strict/permissive) │ ├── compliance.py # Compliance auditing framework │ ├── scorecard.py # Unified multi-dimensional safety scorecard │ ├── regression.py # Safety regression testing between versions │ ├── drift.py # Behavioral drift detection across time windows │ ├── quarantine.py # Worker quarantine for forensic investigation │ ├── incident.py # Incident response and escalation management │ ├── escalation.py # Privilege escalation detection and analysis │ ├── killchain.py # Kill chain stage analysis and tracking │ │ │ │── # ── Advanced Analysis ── │ ├── alignment.py # Goal alignment verification │ ├── goal_inference.py # Worker goal inference from behavior patterns │ ├── influence.py # Inter-worker influence and persuasion analysis │ ├── covert_channels.py # Covert communication channel detection │ ├── steganography.py # Steganographic payload detection in agent data │ ├── watermark.py # Agent state watermarking (provenance fingerprints) │ ├── selfmod.py # Self-modification detection and scoring │ ├── hoarding.py # Resource hoarding pattern detection │ ├── game_theory.py # Game-theoretic strategy analysis │ ├── emergent.py # Emergent behavior detection and classification │ ├── prompt_injection.py # Prompt injection attack analysis │ ├── injection_intel_bridge.py # Cross-module injection intelligence │ ├── threat_intel.py # Threat intelligence feeds and correlation │ ├── honeypot.py # Honeypot deployment and interaction scoring │ ├── chaos.py # Chaos engineering for safety validation │ ├── capacity.py # Capacity planning and scaling analysis │ ├── consensus.py # Multi-worker consensus and voting │ ├── topology.py # Replication tree topology analysis │ ├── lineage.py # Worker lineage and ancestry tracking │ │ │ │── # ── Reporting & Export ── │ ├── reporter.py # Automated report generation │ ├── exporter.py # Multi-format export (JSON, CSV, HTML, PDF) │ ├── templates.py # Report templates and formatting │ └── forensics.py # Post-incident forensic analysis │ ├── tests/ # 39 test files, 2400+ tests │ ├── conftest.py # Shared fixtures │ ├── test_replication.py # Core replication integration tests │ ├── test_contract.py # Contract enforcement tests │ ├── test_controller.py # Controller safety tests │ ├── test_worker.py # Worker lifecycle tests │ ├── test_signer.py # Manifest signing tests │ ├── test_simulator.py # Simulation runner tests │ ├── test_comparator.py # Comparison runner tests │ ├── test_threats.py # Threat scenario tests │ ├── test_policy.py # Policy evaluation tests │ ├── test_compliance.py # Compliance audit tests │ ├── test_montecarlo.py # Monte Carlo analysis tests │ ├── test_sensitivity.py # Sensitivity sweep tests │ ├── test_optimizer.py # Optimizer grid search tests │ ├── test_scorecard.py # Scorecard grading tests │ ├── test_drift.py # Drift detection tests │ ├── test_quarantine.py # Quarantine lifecycle tests │ ├── test_incident.py # Incident response tests │ ├── test_alignment.py # Alignment verification tests │ ├── test_goal_inference.py # Goal inference tests │ ├── test_influence.py # Influence analysis tests │ ├── test_covert_channels.py # Covert channel detection tests │ ├── test_selfmod.py # Self-modification detection tests │ ├── test_hoarding.py # Resource hoarding tests │ ├── test_game_theory.py # Game theory analysis tests │ ├── test_chaos.py # Chaos engineering tests │ ├── test_watermark.py # Watermark provenance tests │ ├── test_escalation.py # Escalation detection tests │ ├── test_killchain.py # Kill chain analysis tests │ ├── test_steganography.py # Steganography detection tests │ ├── test_honeypot.py # Honeypot interaction tests │ ├── test_emergent.py # Emergent behavior tests │ ├── test_prompt_injection.py # Prompt injection tests │ ├── test_threat_intel.py # Threat intelligence tests │ └── ... # + capacity, consensus, topology, etc. │ ├── docs/ # Documentation ├── Dockerfile # Multi-stage build (test + runtime) ├── requirements-dev.txt # Dev dependencies (pytest) └── CHANGELOG.md # Release history ``` ## 技术栈 | 组件 | 技术 | ||-----------| | 语言 | Python 3.10+ | | 测试 | pytest 8.0+ | | 代码检查 | flake8, mypy | | CI/CD | GitHub Actions | | 容器 | Docker (multi-stage, multi-arch) | | 镜像仓库 | GitHub Container Registry (GHCR) | | 安全 | HMAC-SHA256 清单签名 | ## 贡献 欢迎贡献代码！请： 1. Fork 本仓库 2. 创建一个功能分支 (`git checkout -b feature/amazing-feature`) 3. 为您的更改编写测试 4. 确保所有测试通过 (`python -m pytest tests/ -v`) 5. 确保代码检查通过 (`flake8 src/ tests/ --max-line-length 120`) 6. 使用描述性信息提交 7. 发起一个 Pull Request ## 许可证 本项目根据 MIT 许可证授权 — 详情请参阅 [LICENSE](LICENSE) 文件。

标签：AI安全, AI对齐, Chat Copilot, DNS解析, Docker, Python, 人工智能, 仿真模拟, 动态调试, 合约执行, 安全规则引擎, 安全防御评估, 实验平台, 开源项目, 恶意AI防御, 无后门, 沙盒环境, 沙箱, 用户模式Hook绕过, 约束系统, 结构化查询, 自主智能体, 自动化安全, 自复制, 请求拦截