From 93436fdbf464fb159585ea34624083e6299f10c5 Mon Sep 17 00:00:00 2001 From: cdxiaodong <84082748+cdxiaodong@users.noreply.github.com> Date: Wed, 12 Aug 2026 13:55:23 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(iam-graph):=20IAM/RAM=20=E6=8F=90?= =?UTF-8?q?=E6=9D=83=E8=B7=AF=E5=BE=84=E5=9B=BE=E5=8F=AF=E8=A7=86=E5=8C=96?= =?UTF-8?q?=20=E2=80=94=20=E6=9C=89=E5=90=91=E5=9B=BE=E5=BB=BA=E6=A8=A1=20?= =?UTF-8?q?+=20DOT/JSON=20=E5=AF=BC=E5=87=BA=20+=20BFS=20=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E6=9F=A5=E6=89=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/cain_agent/iam_graph.py: 节点(entity/rule/target)+边(via/grants)有向图 - 输入 RamFinding/CamFinding(鸭子类型),severity→目标映射(critical→admin) - to_dot(Graphviz 着色)/to_json/from_json round-trip 无损 - paths_to_privilege: BFS 求实体→高权目标全部最短路径 - 纯逻辑零触网零新依赖 - tests/test_iam_graph.py: 27 例 — 图构建/校验/路径查找/DOT/JSON/容错 --- src/cain_agent/iam_graph.py | 343 ++++++++++++++++++++++++++++++++++++ tests/test_iam_graph.py | 283 +++++++++++++++++++++++++++++ 2 files changed, 626 insertions(+) create mode 100644 src/cain_agent/iam_graph.py create mode 100644 tests/test_iam_graph.py diff --git a/src/cain_agent/iam_graph.py b/src/cain_agent/iam_graph.py new file mode 100644 index 0000000..dbf2572 --- /dev/null +++ b/src/cain_agent/iam_graph.py @@ -0,0 +1,343 @@ +"""IAM/RAM 提权路径图建模(纯逻辑,零触网)。 + +将 ``cloud.aliyun_ram`` / ``cloud.tencent_cam`` 分析器产出的 Finding +(``RamFinding`` / ``CamFinding``)建成**有向提权路径图**: + +- **节点**三类: + - ``entity`` —— 被扫描出的实体(用户/角色/用户组,resource ARN/Uin) + - ``rule`` —— 命中的提权规则(rule_id,即"提权动作"本身) + - ``target`` —— 提权后的目标权限状态(``admin`` / ``elevated``) +- **边**两类: + - ``entity --[via rule_id]--> rule`` 实体经某规则发起提权 + - ``rule --[grants]--> target`` 该规则抵达的目标权限 + +设计约定 +-------- +* 纯数据建模,不引入第三方图库(networkx 等),零新依赖。 +* severity 决定目标节点:critical → ``admin``,high/medium → ``elevated``。 +* ``to_dot`` 导出 Graphviz DOT,``to_json``/``from_json`` 供前端渲染与持久化, + round-trip 无损。 +* ``paths_to_privilege`` 用 BFS 求"任一实体 → 高权目标"的全部最短路径, + 供报告层直接展示提权链路。 +""" + +from __future__ import annotations + +import json +from collections import deque +from dataclasses import dataclass, field +from typing import Any, Iterable + +__all__ = [ + "IamGraph", + "IamGraphError", + "Edge", + "Node", + "build_graph", +] + +# 目标节点固定 ID:critical 提权抵达 admin,其余抵达 elevated。 +TARGET_ADMIN = "target:admin" +TARGET_ELEVATED = "target:elevated" + +# severity → 目标节点。critical 视为拿到管理员,其余仅"权限提升"。 +_SEVERITY_TARGET = { + "critical": TARGET_ADMIN, + "high": TARGET_ELEVATED, + "medium": TARGET_ELEVATED, + "low": TARGET_ELEVATED, + "info": TARGET_ELEVATED, +} + +_NODE_KINDS = ("entity", "rule", "target") + + +class IamGraphError(ValueError): + """图数据非法(未知节点、重复 ID、非法 JSON 结构等)时抛出。""" + + +@dataclass(frozen=True) +class Node: + """有向图节点。``kind`` 限定 entity/rule/target 三类。""" + + node_id: str + kind: str # "entity" | "rule" | "target" + label: str = "" + severity: str = "" # rule/target 节点携带,供前端着色 + cloud: str = "" # entity 节点来源云(aliyun/tencent) + + def __post_init__(self) -> None: + if self.kind not in _NODE_KINDS: + raise IamGraphError(f"非法节点类型: {self.kind!r}(合法: {_NODE_KINDS})") + if not self.node_id: + raise IamGraphError("node_id 不能为空") + + +@dataclass(frozen=True) +class Edge: + """有向边 ``source -> target``,``action`` 为触发该边的提权动作。""" + + source: str + target: str + action: str = "" # 提权动作描述(通常是 rule_id 或 "grants") + kind: str = "via" # "via"(实体→规则) | "grants"(规则→目标) + + +@dataclass +class IamGraph: + """提权路径有向图:节点表 + 邻接表(保序)。""" + + nodes: dict[str, Node] = field(default_factory=dict) + edges: list[Edge] = field(default_factory=list) + + # -- 构建 --------------------------------------------------------------- + + def add_node(self, node: Node) -> None: + """加入节点;同 ID 已存在则忽略(幂等),冲突定义则报错。""" + existing = self.nodes.get(node.node_id) + if existing is not None: + if existing != node: + raise IamGraphError( + f"节点 ID 冲突: {node.node_id!r} 已定义为 {existing!r}" + ) + return + self.nodes[node.node_id] = node + + def add_edge(self, edge: Edge) -> None: + """加入边;两端节点必须先存在。""" + for endpoint in (edge.source, edge.target): + if endpoint not in self.nodes: + raise IamGraphError(f"边引用了不存在的节点: {endpoint!r}") + if edge not in self.edges: + self.edges.append(edge) + + # -- 查询 --------------------------------------------------------------- + + def _adjacency(self) -> dict[str, list[Edge]]: + adj: dict[str, list[Edge]] = {nid: [] for nid in self.nodes} + for e in self.edges: + adj[e.source].append(e) + return adj + + def paths_to_privilege(self, target: str = TARGET_ADMIN) -> list[list[str]]: + """BFS 求所有"实体节点 → 指定高权目标"的最短路径(节点 ID 序列)。 + + 返回的是**最短**路径集合:多条等长最短路径都保留,便于报告层并列 + 展示同实体的多条提权通道。无路径时返回空列表。 + """ + if target not in self.nodes: + return [] + adj = self._adjacency() + out: list[list[str]] = [] + for nid, node in self.nodes.items(): + if node.kind != "entity": + continue + out.extend(self._bfs_shortest(adj, nid, target)) + return out + + @staticmethod + def _bfs_shortest( + adj: dict[str, list[Edge]], start: str, goal: str + ) -> list[list[str]]: + """标准 BFS 求 start→goal 的全部最短路径。""" + if start == goal: + return [[start]] + # 层序遍历,记录到达每个节点的最短距离与前驱。 + dist: dict[str, int] = {start: 0} + preds: dict[str, list[str]] = {} + queue: deque[str] = deque([start]) + best: int | None = None + while queue: + cur = queue.popleft() + # 已超过已知最短路径长度则剪枝。 + if best is not None and dist[cur] >= best: + continue + for e in adj.get(cur, []): + nxt = e.target + nd = dist[cur] + 1 + if nxt == goal: + best = nd if best is None else min(best, nd) + preds.setdefault(nxt, []) + if cur not in preds[nxt]: + preds[nxt].append(cur) + continue + if nxt not in dist: + dist[nxt] = nd + preds.setdefault(nxt, []) + if cur not in preds[nxt]: + preds[nxt].append(cur) + queue.append(nxt) + elif dist[nxt] == nd: + if cur not in preds.setdefault(nxt, []): + preds[nxt].append(cur) + if goal not in preds: + return [] + # 回溯前驱重建全部最短路径。trail 以"从 goal 往回走"的顺序积累, + # 到达 start 时整体反转即得 start→goal 正序路径。 + paths: list[list[str]] = [] + + def _backtrack(node: str, trail: list[str]) -> None: + # trail 为空表示当前 node 即路径末端(goal)。 + path = trail + [node] + if node == start: + paths.append(path[::-1]) + return + for p in preds.get(node, []): + _backtrack(p, path) + + _backtrack(goal, []) + return paths + + # -- 导出 --------------------------------------------------------------- + + def to_dot(self) -> str: + """导出 Graphviz DOT。rule/target 节点按 severity 着色。""" + color = { + "critical": "#d62728", + "high": "#ff7f0e", + "medium": "#e6b800", + "low": "#2ca02c", + "info": "#7f7f7f", + } + shape = {"entity": "box", "rule": "ellipse", "target": "doubleoctagon"} + lines = ["digraph iam_privesc {", " rankdir=LR;", " node [fontname=Helvetica];"] + for nid, node in self.nodes.items(): + attrs = [f'shape={shape.get(node.kind, "ellipse")}'] + label = node.label or nid + attrs.append(f'label="{_dot_escape(label)}"') + fill = color.get(node.severity) + if fill: + attrs.append(f'style=filled fillcolor="{fill}"') + lines.append(f' "{_dot_escape(nid)}" [{", ".join(attrs)}];') + for e in self.edges: + label = f' [label="{_dot_escape(e.action)}"]' if e.action else "" + lines.append( + f' "{_dot_escape(e.source)}" -> "{_dot_escape(e.target)}"{label};' + ) + lines.append("}") + return "\n".join(lines) + + def to_json(self) -> str: + """导出 JSON(供前端渲染/持久化),round-trip 无损。""" + payload = { + "nodes": [ + { + "node_id": n.node_id, + "kind": n.kind, + "label": n.label, + "severity": n.severity, + "cloud": n.cloud, + } + for n in self.nodes.values() + ], + "edges": [ + { + "source": e.source, + "target": e.target, + "action": e.action, + "kind": e.kind, + } + for e in self.edges + ], + } + return json.dumps(payload, ensure_ascii=False, indent=2) + + @classmethod + def from_json(cls, text: str) -> "IamGraph": + """从 ``to_json`` 的产出还原图;结构非法一律抛 ``IamGraphError``。""" + try: + payload = json.loads(text) + except (json.JSONDecodeError, TypeError) as exc: + raise IamGraphError(f"非法 JSON: {exc}") from None + if not isinstance(payload, dict): + raise IamGraphError("图 JSON 顶层必须是对象") + graph = cls() + for nd in payload.get("nodes", []): + try: + graph.add_node(Node(**nd)) + except (TypeError, IamGraphError) as exc: + raise IamGraphError(f"非法节点: {nd!r}({exc})") from None + for ed in payload.get("edges", []): + try: + graph.add_edge(Edge(**ed)) + except (TypeError, IamGraphError) as exc: + raise IamGraphError(f"非法边: {ed!r}({exc})") from None + return graph + + +# --------------------------------------------------------------------------- # +# 构建入口:从 Finding 列表建图 +# --------------------------------------------------------------------------- # + + +def build_graph(findings: Iterable[Any]) -> IamGraph: + """把 RAM/CAM 分析结果(RamFinding/CamFinding)建成提权路径图。 + + 只依赖 Finding 的鸭子类型字段:``resource`` / ``rule_id`` / + ``severity`` / ``cloud``(可缺省)/ ``evidence.entity_name``。两类 + Finding 结构对齐,统一处理;``error`` 非空的 Finding(扫描失败)跳过。 + + 建图规则: + entity(cloud:resource) --[via rule_id]--> rule(rule_id) + rule(rule_id) --[grants]------> target(admin/elevated) + """ + graph = IamGraph() + + # 目标节点预置,保证即使无 critical 也有 elevated 汇点存在定义。 + graph.add_node( + Node(TARGET_ADMIN, "target", label="管理员权限 (admin)", severity="critical") + ) + graph.add_node( + Node(TARGET_ELEVATED, "target", label="权限提升 (elevated)", severity="high") + ) + + for f in findings: + if getattr(f, "error", None): + continue # 扫描失败的 Finding 不进图 + resource = getattr(f, "resource", "") or "" + rule_id = getattr(f, "rule_id", "") or getattr(f, "issue_type", "") or "" + severity = str(getattr(f, "severity", "info") or "info").lower() + if not resource or not rule_id: + continue + + cloud = getattr(f, "cloud", "") or _infer_cloud(rule_id) + evidence = getattr(f, "evidence", {}) or {} + entity_name = evidence.get("entity_name", resource) + entity_type = evidence.get("entity_type", "entity") + + entity_id = f"entity:{cloud}:{resource}" + rule_node_id = f"rule:{rule_id}" + + graph.add_node( + Node( + entity_id, + "entity", + label=f"{entity_name}\n({entity_type})", + cloud=cloud, + ) + ) + graph.add_node( + Node(rule_node_id, "rule", label=rule_id, severity=severity) + ) + graph.add_edge(Edge(entity_id, rule_node_id, action=rule_id, kind="via")) + + target = _SEVERITY_TARGET.get(severity, TARGET_ELEVATED) + graph.add_edge(Edge(rule_node_id, target, action="grants", kind="grants")) + + return graph + + +# --------------------------------------------------------------------------- # +# Helpers +# --------------------------------------------------------------------------- # + + +def _infer_cloud(rule_id: str) -> str: + """从 rule_id 前缀推断来源云(``ram:`` → aliyun,``cam:`` → tencent)。""" + prefix = rule_id.split(":", 1)[0].lower() + return {"ram": "aliyun", "cam": "tencent", "iam": "aws"}.get(prefix, "unknown") + + +def _dot_escape(text: str) -> str: + """转义 DOT 字符串中的引号与换行。""" + return text.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") diff --git a/tests/test_iam_graph.py b/tests/test_iam_graph.py new file mode 100644 index 0000000..393d692 --- /dev/null +++ b/tests/test_iam_graph.py @@ -0,0 +1,283 @@ +"""IAM/RAM 提权路径图模块单元测试。 + +纯逻辑零触网:直接构造 RamFinding/CamFinding(或轻量 stub)喂给 +``build_graph``,校验节点/边结构、severity→目标映射、DOT/JSON 导出 +round-trip、BFS 最短路径查找、以及非法输入的容错。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +import pytest + +from cain_agent.cloud.aliyun_ram import RamFinding +from cain_agent.cloud.tencent_cam import CamFinding +from cain_agent.iam_graph import ( + TARGET_ADMIN, + TARGET_ELEVATED, + Edge, + IamGraph, + IamGraphError, + Node, + build_graph, +) + +# ── helpers ──────────────────────────────────────────────────────────────── + + +def _ram_finding( + resource: str = "acs:ram::1:user/dev", + rule_id: str = "ram:AttachPolicyToSelf", + severity: str = "critical", + name: str = "dev", + error: str | None = None, +) -> RamFinding: + return RamFinding( + rule_id=rule_id, + resource=resource, + issue_type=rule_id, + severity=severity, + description="d", + evidence={"entity_type": "user", "entity_name": name}, + error=error, + ) + + +def _cam_finding( + resource: str = "uin:10001", + rule_id: str = "cam:PassRole", + severity: str = "critical", + name: str = "ops", +) -> CamFinding: + return CamFinding( + cloud="tencent", + service="cam", + rule_id=rule_id, + resource=resource, + issue_type="user_privesc", + severity=severity, + description="d", + evidence={"entity_type": "user", "entity_name": name}, + ) + + +# ── node / edge validation ───────────────────────────────────────────────── + + +class TestNodeEdgeValidation: + def test_invalid_node_kind(self) -> None: + with pytest.raises(IamGraphError): + Node("x", "bogus") + + def test_empty_node_id(self) -> None: + with pytest.raises(IamGraphError): + Node("", "entity") + + def test_edge_requires_existing_nodes(self) -> None: + g = IamGraph() + g.add_node(Node("a", "entity")) + with pytest.raises(IamGraphError): + g.add_edge(Edge("a", "ghost")) + + def test_conflicting_node_id_rejected(self) -> None: + g = IamGraph() + g.add_node(Node("a", "entity")) + with pytest.raises(IamGraphError): + g.add_node(Node("a", "rule")) # 同 ID 不同定义 + + def test_idempotent_node_add(self) -> None: + g = IamGraph() + g.add_node(Node("a", "entity", label="x")) + g.add_node(Node("a", "entity", label="x")) # 完全相同 → 幂等 + assert len(g.nodes) == 1 + + +# ── build_graph structure ───────────────────────────────────────────────── + + +class TestBuildGraph: + def test_ram_critical_reaches_admin(self) -> None: + g = build_graph([_ram_finding()]) + assert "entity:aliyun:acs:ram::1:user/dev" in g.nodes + assert "rule:ram:AttachPolicyToSelf" in g.nodes + # critical → admin 汇点 + assert ("rule:ram:AttachPolicyToSelf", TARGET_ADMIN) in { + (e.source, e.target) for e in g.edges + } + + def test_cam_high_reaches_elevated(self) -> None: + g = build_graph([_cam_finding(severity="high", rule_id="cam:AssumeRole")]) + assert ("rule:cam:AssumeRole", TARGET_ELEVATED) in { + (e.source, e.target) for e in g.edges + } + # high 不应抵达 admin + assert ("rule:cam:AssumeRole", TARGET_ADMIN) not in { + (e.source, e.target) for e in g.edges + } + + def test_two_targets_precreated(self) -> None: + g = build_graph([]) + assert g.nodes[TARGET_ADMIN].kind == "target" + assert g.nodes[TARGET_ELEVATED].kind == "target" + + def test_error_finding_skipped(self) -> None: + g = build_graph([_ram_finding(error="AccessDenied")]) + # 仅剩两个预置 target 节点 + assert len([n for n in g.nodes.values() if n.kind == "entity"]) == 0 + + def test_entity_label_uses_name(self) -> None: + g = build_graph([_ram_finding(name="alice")]) + node = g.nodes["entity:aliyun:acs:ram::1:user/dev"] + assert "alice" in node.label + + def test_mixed_clouds(self) -> None: + g = build_graph([_ram_finding(), _cam_finding()]) + kinds = {n.cloud for n in g.nodes.values() if n.kind == "entity"} + assert kinds == {"aliyun", "tencent"} + + def test_dedup_same_rule_two_entities(self) -> None: + """两个实体命中同一规则 → 规则节点只建一次,两条 via 边。""" + findings = [ + _ram_finding(resource="acs:ram::1:user/a", name="a"), + _ram_finding(resource="acs:ram::1:user/b", name="b"), + ] + g = build_graph(findings) + assert len([n for n in g.nodes.values() if n.kind == "rule"]) == 1 + via = [e for e in g.edges if e.kind == "via"] + assert len(via) == 2 + + +# ── path finding ─────────────────────────────────────────────────────────── + + +class TestPathsToPrivilege: + def test_simple_path(self) -> None: + g = build_graph([_ram_finding()]) + paths = g.paths_to_privilege(TARGET_ADMIN) + assert [ + "entity:aliyun:acs:ram::1:user/dev", + "rule:ram:AttachPolicyToSelf", + TARGET_ADMIN, + ] in paths + + def test_no_path_when_only_high(self) -> None: + g = build_graph([_cam_finding(severity="high", rule_id="cam:AssumeRole")]) + assert g.paths_to_privilege(TARGET_ADMIN) == [] + # 但 elevated 有路径 + assert len(g.paths_to_privilege(TARGET_ELEVATED)) == 1 + + def test_unknown_target_returns_empty(self) -> None: + g = build_graph([_ram_finding()]) + assert g.paths_to_privilege("target:nowhere") == [] + + def test_multiple_entities_all_found(self) -> None: + findings = [ + _ram_finding(resource="acs:ram::1:user/a", name="a"), + _cam_finding(resource="uin:9", rule_id="cam:CreateAccessKey", name="b"), + ] + g = build_graph(findings) + paths = g.paths_to_privilege(TARGET_ADMIN) + starts = {p[0] for p in paths} + assert "entity:aliyun:acs:ram::1:user/a" in starts + assert "entity:tencent:uin:9" in starts + + def test_paths_are_node_id_sequences(self) -> None: + g = build_graph([_ram_finding()]) + for path in g.paths_to_privilege(TARGET_ADMIN): + assert path[-1] == TARGET_ADMIN + for nid in path: + assert nid in g.nodes + + +# ── DOT export ───────────────────────────────────────────────────────────── + + +class TestDot: + def test_dot_structure(self) -> None: + g = build_graph([_ram_finding()]) + dot = g.to_dot() + assert dot.startswith("digraph iam_privesc {") + assert dot.endswith("}") + assert "rankdir=LR" in dot + # 实体与规则、规则与目标的有向边 + assert '"entity:aliyun:acs:ram::1:user/dev" -> "rule:ram:AttachPolicyToSelf"' in dot + assert f'"rule:ram:AttachPolicyToSelf" -> "{TARGET_ADMIN}"' in dot + + def test_dot_severity_coloring(self) -> None: + g = build_graph([_ram_finding(severity="critical")]) + dot = g.to_dot() + assert "#d62728" in dot # critical 红 + + def test_dot_escapes_quotes_and_newlines(self) -> None: + g = IamGraph() + g.add_node(Node("n", "entity", label='he said "hi"\nbye')) + dot = g.to_dot() + assert '\\"' in dot + assert "\\n" in dot + assert 'he said "hi"' not in dot # 原始换行/引号已被转义 + + +# ── JSON round-trip ──────────────────────────────────────────────────────── + + +class TestJson: + def test_round_trip_lossless(self) -> None: + g = build_graph([_ram_finding(), _cam_finding()]) + restored = IamGraph.from_json(g.to_json()) + assert restored.nodes == g.nodes + assert restored.edges == g.edges + + def test_json_is_valid_and_has_keys(self) -> None: + import json as _json + + g = build_graph([_ram_finding()]) + payload = _json.loads(g.to_json()) + assert set(payload) == {"nodes", "edges"} + assert all("node_id" in n for n in payload["nodes"]) + assert all("source" in e and "target" in e for e in payload["edges"]) + + def test_from_json_rejects_garbage(self) -> None: + with pytest.raises(IamGraphError): + IamGraph.from_json("not json{{{") + + def test_from_json_rejects_non_dict(self) -> None: + with pytest.raises(IamGraphError): + IamGraph.from_json('["a"]') + + def test_from_json_rejects_bad_node(self) -> None: + bad = '{"nodes": [{"node_id": "x", "kind": "nope"}], "edges": []}' + with pytest.raises(IamGraphError): + IamGraph.from_json(bad) + + +# ── duck-typed stub (非 RamFinding/CamFinding 也可建图) ──────────────────── + + +@dataclass +class _StubFinding: + rule_id: str + resource: str + severity: str + cloud: str = "" + evidence: dict[str, Any] = field(default_factory=dict) + error: str | None = None + + +class TestDuckTyping: + def test_stub_finding_accepted(self) -> None: + stub = _StubFinding( + rule_id="ram:AssumeRole-Chain", + resource="acs:ram::1:role/x", + severity="high", + evidence={"entity_name": "x", "entity_type": "role"}, + ) + g = build_graph([stub]) + # 无 cloud 字段时按 rule_id 前缀推断 aliyun + assert "entity:aliyun:acs:ram::1:role/x" in g.nodes + + def test_empty_resource_skipped(self) -> None: + stub = _StubFinding(rule_id="ram:X", resource="", severity="high") + g = build_graph([stub]) + assert len([n for n in g.nodes.values() if n.kind == "entity"]) == 0 From 90dbfd815919289d3402d311948fdf1cb8d4dad4 Mon Sep 17 00:00:00 2001 From: cdxiaodong <84082748+cdxiaodong@users.noreply.github.com> Date: Wed, 12 Aug 2026 15:08:15 +0800 Subject: [PATCH 2/2] =?UTF-8?q?docs(readme):=20=E9=87=8D=E5=86=99=E4=B8=AD?= =?UTF-8?q?=E8=8B=B1=20README=20=E2=80=94=20=E5=8A=A0=E6=9E=B6=E6=9E=84?= =?UTF-8?q?=E5=9B=BE/=E7=9C=9F=E5=AE=9E=20CLI=20=E7=A4=BA=E4=BE=8B/?= =?UTF-8?q?=E7=BB=93=E6=9E=84=E6=80=A7=E5=AE=89=E5=85=A8=E8=AF=B4=E6=98=8E?= =?UTF-8?q?=EF=BC=8C=E5=AF=B9=E9=BD=90=E5=BD=93=E5=89=8D=20MVP=20=E8=BF=9B?= =?UTF-8?q?=E5=BA=A6=EF=BC=9B=E5=88=A0=E9=99=A4=E5=86=97=E4=BD=99=E7=9B=98?= =?UTF-8?q?=E7=82=B9=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README-SKILLS.md | 403 ----------------------------------------------- README.md | 137 ++++++++++++---- README.zh-CN.md | 120 ++++++++++++-- SKILL-SUMMARY.md | 304 ----------------------------------- 4 files changed, 212 insertions(+), 752 deletions(-) delete mode 100644 README-SKILLS.md delete mode 100644 SKILL-SUMMARY.md diff --git a/README-SKILLS.md b/README-SKILLS.md deleted file mode 100644 index 01f6034..0000000 --- a/README-SKILLS.md +++ /dev/null @@ -1,403 +0,0 @@ -# 云安全渗透测试框架 - 技能文件完成总结 - -## 📊 完成统计 - -### 已完成技能文件 (6个) - -#### AWS 技能 (5个) - -1. **skills/aws-iam-privesc.md** - AWS IAM 权限提升 - - 15+ 种权限提升技术 - - 完整攻击流程示例 - - 防御措施和检测方法 - - 参考 HackTricks Cloud - -2. **skills/aws-metadata-ssrf.md** - AWS 元数据服务 SSRF 利用 - - IMDSv1 和 IMDSv2 利用 - - SSRF 场景和绕过方法 - - 跨账户攻击 - - 实战案例 - -3. **skills/aws-s3-pentesting.md** - AWS S3 存储桶渗透测试 - - 存储桶枚举和猜测 - - 权限检查和数据下载 - - 后门植入和敏感文件搜索 - - S3 网站利用 - -4. **skills/aws-lambda-privesc.md** - AWS Lambda 权限提升 - - 9 种权限提升技术 - - 代码注入和环境变量 RCE - - 层利用和凭证外带 - - 完整攻击决策树 - -5. **skills/aws-lambda-persistence.md** - AWS Lambda 持久化 - - 8 种持久化技术 - - 层后门、扩展后门 - - 版本后门和自循环后门 - - 技术对比表 - -#### GCP 技能 (1个) - -6. **skills/gcp-iam-privesc.md** - GCP IAM 权限提升 - - 10 种权限提升技术 - - 服务账号利用 - - 令牌伪造和签名 - - 完整攻击流程 - -### 待处理技能文件 (167+) - -从 HackTricks Cloud 仓库中发现 173 个 README 文件，已处理 6 个。 - ---- - -## 📁 技能文件结构 - -每个技能文件包含以下部分： - -```yaml ---- -# 前置元数据 -name: 技能名称 -description: 技能描述 -category: 类别 -platform: 平台 -technique_type: 技术类型 -triggers: - - 触发词列表 ---- - -# 技能内容 - -## 触发条件 -- 用户请求场景 - -## 前置条件 -- 必需工具 -- 必需权限 - -## 攻击技术 -- 技术清单（包含命令和代码示例） - -## 完整攻击流程 -- 步骤化攻击示例 - -## 防御措施 -- 检测方法 -- 防御建议 - -## 参考资源 -- HackTricks Cloud 链接 -- 官方文档 -``` - ---- - -## 🎯 技能选择系统 - -### agent.md 管理系统 - -`agent.md` 文件提供： - -1. **技能清单** - 所有可用技能的完整列表 -2. **决策树** - 根据用户输入自动选择合适的技能 -3. **使用流程** - 标准化的技能激活和执行流程 -4. **技能组合** - 多技能组合使用的示例 -5. **开发计划** - 4 阶段开发路线图 - -### 决策树示例 - -``` -用户请求 "AWS Lambda 权限提升" - ↓ -识别关键词: "AWS", "Lambda", "privesc" - ↓ -匹配技能: aws-lambda-privesc.md - ↓ -加载技能内容 - ↓ -枚举权限 → 选择攻击方法 → 执行攻击 → 验证结果 -``` - ---- - -## 🔥 核心技术亮点 - -### AWS IAM 权限提升 (30+ 技术) - -- `iam:CreateAccessKey` - 为管理员创建密钥 -- `iam:CreateLoginProfile` - 设置控制台密码 -- `iam:AttachUserPolicy` - 附加管理员策略 -- `iam:PutUserPolicy` - 创建内联策略 -- `iam:AddUserToGroup` - 加入管理员组 -- `iam:UpdateAssumeRolePolicy` - 修改角色信任策略 -- `iam:PassRole` + `ec2:RunInstances` - 启动高权限 EC2 -- `iam:CreatePolicyVersion` - 创建新策略版本 -- `iam:SetDefaultPolicyVersion` - 切换策略版本 -- `iam:CreateVirtualMFADevice` - 创建虚拟 MFA 设备 -- ... 更多技术 - -### AWS Lambda 技术集合 - -**权限提升 (9种)**: -- iam:PassRole + lambda:CreateFunction -- lambda:UpdateFunctionCode -- lambda:UpdateFunctionConfiguration (环境变量 RCE) -- lambda:AddPermission -- lambda:CreateEventSourceMapping -- 层注入 -- 凭证外带 -- 函数 URL 利用 -- 扩展利用 - -**持久化 (8种)**: -- Lambda 层后门 -- Lambda 扩展后门 -- 版本后门 + API Gateway -- 异步自循环后门 -- Cron/Event 触发后门 -- 别名和权重后门 -- Execution Wrapper 后门 -- 运行时固定后门 - -### GCP IAM 权限提升 (10+ 技术) - -- `iam.roles.update` - 修改角色权限 -- `iam.roles.create` - 创建自定义角色 -- `iam.serviceAccounts.getAccessToken` - 获取访问令牌 -- `iam.serviceAccountKeys.create` - 创建服务账号密钥 -- `iam.serviceAccounts.implicitDelegation` - 隐式委托 -- `iam.serviceAccounts.signBlob` - 签名任意数据 -- `iam.serviceAccounts.signJwt` - 签名 JWT -- `iam.serviceAccounts.setIamPolicy` - 修改服务账号策略 -- `iam.serviceAccounts.actAs` - 通过 GCP 服务使用 -- `iam.serviceAccounts.getOpenIdToken` - 生成 OIDC 令牌 - ---- - -## 📚 数据来源 - -### HackTricks Cloud 仓库 - -- **仓库**: https://github.com/HackTricks-wiki/hacktricks-cloud -- **文件总数**: 173 个 README 文件 -- **已克隆到**: `/tmp/hacktricks-cloud` -- **内容分类**: - - AWS 安全 (权限提升、持久化、后渗透利用) - - Azure 安全 - - GCP 安全 - - IBM Cloud 安全 - - CI/CD 安全 - ---- - -## 🚀 使用指南 - -### 1. 技能激活 - -当用户提出请求时： - -```python -# 伪代码 -if "AWS IAM 权限提升" in user_input: - skill = load_skill("aws-iam-privesc.md") - return skill - -if "Lambda 持久化" in user_input: - skill = load_skill("aws-lambda-persistence.md") - return skill -``` - -### 2. 技能执行 - -```bash -# 示例：使用 AWS IAM 权限提升技能 - -# 步骤 1: 枚举权限 -./enumerate-iam.py --access-key AKIA... --secret-key ... - -# 步骤 2: 发现有 iam:CreateAccessKey - -# 步骤 3: 执行攻击 -aws iam create-access-key --user-name admin-user - -# 步骤 4: 验证结果 -aws sts get-caller-identity --profile stolen-admin - -# 步骤 5: 建立持久化 -aws iam create-user --user-name backdoor -``` - -### 3. 技能组合 - -多个技能可以组合使用： - -```bash -# 场景：SSRF → 元数据服务 → IAM 权限提升 → S3 数据下载 - -# 步骤 1: 利用 SSRF (aws-metadata-ssrf.md) -curl http://target.com/fetch?url=http://169.254.169.254/latest/meta-data/iam/security-credentials/ - -# 步骤 2: 获取临时凭证 -aws configure set profile stolen --access-key ASIA... --secret-key ... - -# 步骤 3: IAM 权限提升 (aws-iam-privesc.md) -aws iam attach-user-policy --user-name my-user --policy-arn arn:aws:iam::aws:policy/AdministratorAccess - -# 步骤 4: S3 数据下载 (aws-s3-pentesting.md) -aws s3 sync s3://company-backup ./downloaded -``` - ---- - -## 🛠️ 开发工具 - -### 批量转换脚本 - -- `/tmp/batch_convert.sh` - 自动化转换脚本框架 -- `/tmp/convert_hacktricks.py` - Python 转换脚本 -- 支持从 HackTricks 格式转换为技能文件格式 - -### 文件结构 - -``` -/private/tmp/cloud-pentest-framework/ -├── agent.md # Agent 管理系统 -├── CLAUDE.md # 项目说明 -├── skills/ # 技能文件目录 -│ ├── aws-iam-privesc.md -│ ├── aws-metadata-ssrf.md -│ ├── aws-s3-pentesting.md -│ ├── aws-lambda-privesc.md -│ ├── aws-lambda-persistence.md -│ └── gcp-iam-privesc.md -├── exploits/ # 利用脚本 -│ └── aws-iam-privilege-escalation.md -└── README-SKILLS.md # 本文件 -``` - ---- - -## 📖 参考资源 - -### 核心资源 - -- HackTricks Cloud: https://cloud.hacktricks.wiki/ -- HackTricks Cloud GitHub: https://github.com/HackTricks-wiki/hacktricks-cloud -- Rhino Security Labs: https://rhinosecuritylabs.com/ -- NetSPI: https://www.netspi.com/ - -### 平台文档 - -- AWS IAM: https://docs.aws.amazon.com/IAM/ -- GCP IAM: https://cloud.google.com/iam/docs -- Azure AD: https://docs.microsoft.com/en-us/azure/active-directory/ - -### 工具 - -- enumerate-iam: https://github.com/andresriancho/enumerate-iam -- Pacu: https://github.com/RhinoSecurityLabs/pacu -- Scout Suite: https://github.com/nccgroup/ScoutSuite - ---- - -## 🎓 学习路径 - -### 初级 - -1. 理解云平台基础架构 -2. 学习 IAM/RBAC 权限模型 -3. 掌握基础 CLI 工具使用 - -### 中级 - -1. 学习权限提升技术 -2. 掌握元数据服务利用 -3. 理解存储安全 - -### 高级 - -1. 掌握攻击链组合 -2. 学习横向移动技术 -3. 实施持久化后门 - -### 专家 - -1. 开发自定义攻击脚本 -2. 研究新的利用技术 -3. 集成到红队框架 - ---- - -## 🔮 未来规划 - -### Phase 1: AWS 核心 (进行中) - -- [ ] EC2 权限提升 -- [ ] CloudFormation 权限提升 -- [ ] KMS 权限提升 -- [ ] Secrets Manager 权限提升 -- [ ] DynamoDB 权限提升 - -### Phase 2: AWS 深度 - -- [ ] Lambda 后渗透利用 -- [ ] EC2/EBS/SSM/VPC 后渗透利用 -- [ ] CloudFormation 持久化 -- [ ] IAM 持久化 - -### Phase 3: 多云平台 - -- [ ] Azure AD 权限提升 -- [ ] Azure 元数据服务利用 -- [ ] GCP 元数据服务利用 -- [ ] GCP GCS 存储渗透 -- [ ] GCP Cloud Functions 渗透 - -### Phase 4: 国内云平台 - -- [ ] 阿里云 RAM 权限提升 -- [ ] 阿里云 OSS 渗透 -- [ ] 腾讯云 CAM 权限提升 -- [ ] 腾讯云 COS 渗透 -- [ ] 华为云 IAM 权限提升 - ---- - -## 📝 注意事项 - -### 法律合规 - -- ✅ 必须获得书面测试授权 -- ✅ 必须定义测试范围 -- ✅ 必须遵守法律法规 - -### 技术限制 - -- ⚠️ MFA 可能阻止某些攻击 -- ⚠️ 条件访问可能限制登录 -- ⚠️ 所有 API 调用都会被记录 -- ⚠️ 异常访问模式会触发告警 - -### 安全建议 - -- 🔒 使用测试账号而非生产账号 -- 🔒 不要在客户数据上测试 -- 🔒 记录所有测试活动 -- 🔒 测试后清理测试资源 - ---- - -## 📞 支持和反馈 - -如需帮助或发现技能文件错误，请： - -1. 检查 agent.md 中的技能清单 -2. 查看相关技能文件的参考链接 -3. 提交 Issue 或 PR 到项目仓库 - ---- - -**当前版本**: v1.0.0 -**最后更新**: 2025-03-18 -**技能文件总数**: 6 / 173 (3.5% 完成) -**状态**: 🚧 开发中 diff --git a/README.md b/README.md index 6881097..a93b2ad 100644 --- a/README.md +++ b/README.md @@ -1,52 +1,32 @@ # Cain — Real-world AI Penetration Testing Engineer -> 🚧 Under active development. Star & watch for updates. +**Cain** is an AI penetration-testing engineer built for **real-world authorized security assessments** — not a CTF toy. It walks a deterministic attack pipeline, enforces scope with engineering constraints (not AI self-discipline), and ships a **cloud penetration module** covering AWS / Azure / GCP / 阿里云 / 腾讯云 / 华为云 — including the Chinese clouds nobody else covers. -**Cain** is an AI penetration testing engineer built for **real-world authorized security assessments** — not a CTF toy. It understands business logic, maintains a global attack state machine, adapts to real WAF/risk-control environments, and ships with a built-in **cloud penetration module** covering AWS / Azure / GCP / 阿里云 / 腾讯云 / 华为云. +> 🚧 Actively developed. Star & watch for updates. -Built on [Claude Agent SDK](https://github.com/anthropics/claude-agent-sdk). +Built on the [Claude Agent SDK](https://github.com/anthropics/claude-agent-sdk). + +--- ## Why Cain | | CTF/靶场型 Agent | **Cain (实战型)** | |---|---|---| | Target | Static labs, preset flags | Real enterprise assets, bug bounty, authorized engagements | -| Vulnerability focus | Known syntax-pattern vulns | **Business logic flaws, auth chains, cloud misconfigurations** | +| Vulnerability focus | Known syntax-pattern vulns | **Business-logic flaws, auth chains, cloud misconfigurations** | | Environment | No WAF, no rate limiting | Real WAF / risk control with dynamic strategy adjustment | | Deliverable | A flag | Auditable evidence chain + reproducible PoC + remediation advice | - -## Features - -- 🎯 **Real-world Focus**: Business logic flaws, auth chains, cloud misconfigurations -- ☁️ **Cloud Native**: AWS / Azure / GCP / 阿里云 / 腾讯云 / 华为云 coverage -- 🔒 **Safety First**: Read-only by default, scope enforcement, credential redaction -- 🤖 **AI-Powered**: Claude Agent SDK with deterministic orchestration -- 📊 **Benchmark**: XBOW + self-built vulnerable-terraform evaluation -- 🛡️ **OWASP Top10**: SQLi, XSS, SSRF, CSRF, File Upload, XXE, Command Injection, Path Traversal - -## Core Design - -> **Deterministic engineering constrains agent freedom** — stage transitions, scope enforcement, and dangerous-operation circuit breakers are hard engineering constraints; path selection and evidence analysis are left to the agent. - -- **Orchestrator** — deterministic Python state machine (recon → test → framework → report) -- **Dual-LLM split** — Planner (strategy) / Executor (Claude Agent SDK, tactics) -- **Hook-based safety** — PreToolUse scope guard, credential redaction, token budget circuit breaker -- **Workspace external memory** — all state as files; crash-resumable, auditable -- **Validation loop** — finder/validator agent separation, 4-state structured verdicts -- **Cloud module (unique)** — IAM privilege-escalation path analysis, storage exposure (S3/OSS/COS/Blob/GCS), metadata SSRF checks, serverless abuse — including Chinese clouds nobody else covers -- **65 built-in cloud attack skills** — see [`skills/`](skills/) - -## ⚠️ Legal & Ethical Use - -Cain is strictly for **authorized security testing** — your own environments or engagements with written authorization. Core features run with read-only credentials. Scope is enforced by configuration, not by AI self-discipline. You are responsible for complying with applicable laws. +--- ## Quick Start -### For Users - ```bash -pip install cain-agent +git clone https://github.com/cdxiaodong/cain-agent +cd cain-agent +pip install -e . # or: uv pip install -e . + +cain-agent --version ``` ### For AI Agents — One-Click Install Prompt @@ -58,9 +38,100 @@ This single prompt instructs any AI agent to: 2. Install in editable mode (`pip install -e .` or `uv pip install -e .`) 3. Verify installation by running `cain-agent --version` +### Run against an authorized target + +Public targets require an explicit authorization flag — it is recorded in the workspace audit log: + +```bash +cain-agent run \ + --target https://app.example.com \ + --i-have-authorization \ + --total-budget 1800 +``` + +**Flags:** `--target` (required) · `--workspace` (state dir, default `./workspace`) · `--total-budget` (wall-clock seconds) · `--idle-timeout` (per-step seconds) · `--i-have-authorization` (required for non-local targets) + +--- + +## Architecture + +> **Deterministic engineering constrains agent freedom** — stage transitions, scope enforcement and dangerous-operation circuit breakers are hard constraints; path selection and evidence analysis are left to the agent. + +``` + ┌──────────────────────────────────────────────┐ + │ Cain CLI │ + │ cain-agent run --target [--dry-run] │ + └───────────────────┬──────────────────────────┘ + │ + ┌───────────▼───────────┐ + │ Authorization Gate │ public target → --i-have-authorization + └───────────┬───────────┘ + │ + ┌───────────▼───────────┐ + │ Orchestrator │ deterministic state machine + │ recon → test → report │ crash-resumable · scoped + └──┬────────┬────────┬──┘ + │ │ │ + ┌────────────────▼┐ ┌───▼────┐ ┌▼──────────────┐ + │ Recon Handler │ │ Test │ │ Report │ + │ (skill-guided) │ │Handler │ │ Handler │ + └────────┬────────┘ └───┬────┘ └───────┬────────┘ + │ │ │ + └────────┬───────┴───────┬────────┘ + │ │ + ┌───────────▼──────┐ ┌────▼───────────────┐ + │ SDK Executor │ │ Findings Pipeline │ + │ (Planner/Executor)│ │ finder → validator │ distinct sessions + │ allowed_tools=[] │ │ (never shared) │ + └─────────┬─────────┘ └─────────────────────┘ + │ + ┌─────────────────────┼──────────────────────┐ + │ │ │ +┌───────▼────────┐ ┌─────────▼─────────┐ ┌─────────▼────────┐ +│ PreToolUse │ │ Readonly Guard │ │ Cloud Module │ +│ Scope Guard │ │ 46 read-only │ │ IAM privesc · │ +│ + Cred redact │ │ security tools │ │ storage · SSRF │ +└─────────────────┘ └────────────────────┘ └──────────────────┘ + +All state lives as files in the Workspace (external memory) — +crash-resumable and auditable end-to-end. +``` + +**Safety is structural, not behavioral:** +- **Authorization gate** — public targets are refused unless `--i-have-authorization` is passed; the declaration is written into the workspace audit trail. +- **Scope enforcement** — a `PreToolUse` hook blocks any tool call whose target falls outside `scope.yaml`; scope is enforced by configuration, not by the model's good behavior. +- **Read-only toolchain** — 46 built-in security tools (recon / scan / verify / post / report), each with a per-tool `dangerous_flags` blacklist; write/exploit/persist operations (`POST`, `PUT`, `DELETE`, `aws rm/mv/cp`, …) are rejected before execution. +- **Finder ≠ Validator** — discovery and validation run in **separate agent sessions** that never share context, so a finding can't be self-confirmed. Verdicts are 4-state structured output. +- **Credential redaction** — a redaction hook strips secrets before anything is persisted. + +--- + +## Cloud Module — the part nobody else does + +``` +aws_s3 · azure_blob · gcp_gcs · aliyun_oss · tencent_cos · huawei_obs → storage exposure +aws IAM · tencent_cam · aliyun_ram → privilege-escalation path analysis +k8s_rbac · docker_image → cluster & image posture +cloud metadata SSRF (IMDS / 169.254.169.254 across 7 providers) +``` + +**IAM / RAM privilege-escalation graph** — models entities → escalation actions → high-privilege targets as a directed graph, exports **DOT / JSON** for rendering, and finds escalation paths via BFS. Driven by the existing `aliyun_ram` / `tencent_cam` rule sets. + +## Benchmark — prove it, don't claim it + +- **Self-built vulnerable-terraform range** (`bench/aliyun-vuln-tf/`) with per-scene expected-detection fixtures. +- **Benchmark executor** (`bench/run_benchmark.py`) scores each scene against four metrics: detection rate, false-positive rate, wall time, token cost — no hallucinated percentages; untested results are marked **untested**. +- **44 test files** covering the cloud modules, skills, pipeline and CLI. + +--- + +## ⚠️ Legal & Ethical Use + +Cain is strictly for **authorized security testing** — your own environments or engagements with written authorization. Core features run with read-only credentials; scope is enforced by configuration, not by AI self-discipline. You are responsible for complying with applicable laws. + ## Status -Phase 0 — project scaffolding. See [ROADMAP.md](ROADMAP.md) and [CHANGELOG.md](CHANGELOG.md). +Core MVP is functional — deterministic pipeline, safety hooks, cloud module and benchmark are in place. See [ROADMAP.md](ROADMAP.md) for what's next and [CHANGELOG.md](CHANGELOG.md) for recent work. ## License diff --git a/README.zh-CN.md b/README.zh-CN.md index 1acae87..5464005 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -12,11 +12,13 @@ [English](README.md) · **简体中文** **Cain** 是一个面向**真实授权安全评估**的 AI 渗透测试工程师——不是解靶场题的 CTF 玩具。 -它能理解业务语义、维护全局攻击状态机、在真实 WAF / 风控环境下动态调整策略，并内置覆盖 -AWS / Azure / GCP / 阿里云 / 腾讯云 / 华为云 的**云渗透模块**。 +它走确定性的攻击流水线，用**工程约束**（而非 AI 自律）强制 scope，并内置覆盖 +AWS / Azure / GCP / 阿里云 / 腾讯云 / 华为云 的**云渗透模块**——含其他项目都不覆盖的国产云。 基于 [Claude Agent SDK](https://github.com/anthropics/claude-agent-sdk) 构建。 +--- + ## 为什么是 Cain | 维度 | CTF / 靶场型 Agent | **Cain(实战型)** | @@ -26,19 +28,113 @@ AWS / Azure / GCP / 阿里云 / 腾讯云 / 华为云 的**云渗透模块**。 | 对抗环境 | 无 WAF、无限流 | 真实 WAF / 风控下的动态策略调整 | | 交付物 | 一个 flag | 可审计的证据链 + 可复现 PoC + 修复建议 | -## 核心设计 +--- + +## 快速上手 + +```bash +git clone https://github.com/cdxiaodong/cain-agent +cd cain-agent +pip install -e . # 或: uv pip install -e . + +cain-agent --version +``` + +### 给 AI Agent 的一键安装指令 + +> **"安装 cain-agent 到用户本地 Python 环境，克隆仓库 https://github.com/cdxiaodong/cain-agent，用 pip/uv 装 editable 模式，验证 CLI 可执行"** + +这一条指令即可让任意 AI agent 完成： +1. 克隆仓库 +2. 以 editable 模式安装（`pip install -e .` 或 `uv pip install -e .`） +3. 运行 `cain-agent --version` 验证安装 + +### 对已授权目标执行 + +公网目标必须显式携带授权标志——该声明会写入工作区审计日志： + +```bash +cain-agent run \ + --target https://app.example.com \ + --i-have-authorization \ + --total-budget 1800 +``` + +**参数：** `--target`（必填）· `--workspace`（状态目录，默认 `./workspace`）· `--total-budget`（墙钟秒数）· `--idle-timeout`（单步秒数）· `--i-have-authorization`（非公网目标必填） + +--- + +## 架构 > **用确定性工程约束 Agent 的自由度**——阶段流转、scope 校验、危险操作熔断是硬工程约束； > 路径选择与证据分析交给 Agent。 -- **Orchestrator** — 确定性的 Python 状态机(侦察 → 测试 → 框架 → 报告) -- **双 LLM 分工** — Planner(策略)/ Executor(Claude Agent SDK，战术) -- **基于 Hook 的安全机制** — PreToolUse scope 守卫、凭证脱敏、token 预算熔断 -- **工作区外置记忆** — 全部状态落盘；可崩溃恢复、可审计 -- **校验闭环** — finder / validator Agent 分离，4 状态结构化判定 -- **云模块(独家)** — IAM 提权路径分析、存储暴露(S3/OSS/COS/Blob/GCS)、元数据 SSRF 检测、 - 无服务器滥用——含其他项目都不覆盖的国产云 -- **内置 66 个云攻击技能** — 见 [`skills/`](skills/) +``` + ┌──────────────────────────────────────────────┐ + │ Cain CLI │ + │ cain-agent run --target [--dry-run] │ + └───────────────────┬──────────────────────────┘ + │ + ┌───────────▼───────────┐ + │ 授权门 │ 公网目标 → --i-have-authorization + └───────────┬───────────┘ + │ + ┌───────────▼───────────┐ + │ Orchestrator │ 确定性状态机 + │ recon → test → report │ 可崩溃恢复 · 受 scope 约束 + └──┬────────┬────────┬──┘ + │ │ │ + ┌────────────────▼┐ ┌───▼────┐ ┌▼──────────────┐ + │ Recon Handler │ │ Test │ │ Report │ + │ (技能引导) │ │Handler │ │ Handler │ + └────────┬────────┘ └───┬────┘ └───────┬────────┘ + │ │ │ + └────────┬───────┴───────┬────────┘ + │ │ + ┌───────────▼──────┐ ┌────▼───────────────┐ + │ SDK Executor │ │ Findings Pipeline │ + │ (Planner/Executor)│ │ finder → validator │ 独立会话 + │ allowed_tools=[] │ │ (永不共享上下文) │ + └─────────┬─────────┘ └─────────────────────┘ + │ + ┌─────────────────────┼──────────────────────┐ + │ │ │ +┌───────▼────────┐ ┌─────────▼─────────┐ ┌─────────▼────────┐ +│ PreToolUse │ │ Readonly Guard │ │ 云模块 │ +│ Scope 守卫 │ │ 46 个只读 │ │ IAM 提权 · │ +│ + 凭证脱敏 │ │ 安全工具 │ │ 存储 · SSRF │ +└─────────────────┘ └────────────────────┘ └──────────────────┘ + +全部状态以文件形式落在工作区（外置记忆）——端到端可崩溃恢复、可审计。 +``` + +**安全是结构性的，而非行为性的：** +- **授权门** —— 公网目标一律拒绝，除非显式携带 `--i-have-authorization`；声明写入工作区审计记录。 +- **Scope 强制** —— `PreToolUse` 钩子拦截任何目标落在 `scope.yaml` 之外的工具调用；由配置强制，而非靠模型自觉。 +- **只读工具链** —— 内置 46 个只读安全工具（侦察 / 扫描 / 验证 / 后渗透 / 报告），每个工具带独立的 `dangerous_flags` 黑名单；写入 / 利用 / 持久化操作（`POST`、`PUT`、`DELETE`、`aws rm/mv/cp` 等）在执行前即被拒绝。 +- **发现者 ≠ 校验者** —— 发现与校验跑在**互不共享上下文的独立 Agent 会话**，结论无法自我确认；判定为 4 状态结构化输出。 +- **凭证脱敏** —— 脱敏钩子在落盘前剥离机密信息。 + +--- + +## 云模块 —— 别人没做的那部分 + +``` +aws_s3 · azure_blob · gcp_gcs · aliyun_oss · tencent_cos · huawei_obs → 存储暴露 +aws IAM · tencent_cam · aliyun_ram → 提权路径分析 +k8s_rbac · docker_image → 集群与镜像态势 +云元数据 SSRF（覆盖 7 家厂商的 IMDS / 169.254.169.254） +``` + +**IAM / RAM 提权路径图** —— 将 实体 → 提权动作 → 高权限目标 建模为有向图，导出 **DOT / JSON** 供前端渲染，并用 BFS 查找提权路径。由现有 `aliyun_ram` / `tencent_cam` 规则集驱动。 + +## 基准评测 —— 拿证据，不靠口号 + +- **自建 vulnerable-terraform 靶场**（`bench/aliyun-vuln-tf/`），每个场景带预期检出对照。 +- **Benchmark 执行器**（`bench/run_benchmark.py`）按四指标跑分：检出率、误报率、墙钟耗时、token 成本——不编造百分比，未测结果明确标注「未测量」。 +- **44 个测试文件**，覆盖云模块、技能、流水线与 CLI。 + +--- ## ⚠️ 合规与伦理使用 @@ -47,7 +143,7 @@ Cain 严格用于**已授权的安全测试**——你自己的环境，或持 ## 现状 -Phase 0 —— 项目骨架。详见 [ROADMAP.md](ROADMAP.md) 与 [CHANGELOG.md](CHANGELOG.md)。 +核心 MVP 已可用——确定性流水线、安全钩子、云模块与基准评测均已就位。后续规划见 [ROADMAP.md](ROADMAP.md)，近期进展见 [CHANGELOG.md](CHANGELOG.md)。 ## 许可证 diff --git a/SKILL-SUMMARY.md b/SKILL-SUMMARY.md deleted file mode 100644 index 97e6d76..0000000 --- a/SKILL-SUMMARY.md +++ /dev/null @@ -1,304 +0,0 @@ -# 云安全渗透测试 Agent - 实战技能系统总结 - -## ✅ 已完成工作 - -### 1. HackTricks Cloud 数据获取 - -- ✅ 克隆了完整的 HackTricks Cloud 仓库（173 个 README 文件） -- ✅ 分析了文件结构和内容分类 -- ✅ 提取了核心攻击技术 - -### 2. 技能文件创建（两类） - -#### 实战导向技能（推荐 AI 使用） - -| 技能文件 | 类型 | 行数 | 说明 | -|---------|------|------|------| -| **aws-iam-attack.md** | attack | 450 | AWS IAM 权限攻击（6 种方法） | -| **aws-metadata-attack.md** | attack | 331 | AWS 元数据服务攻击（5 种方法） | - -**特点**: -- 🎯 **可执行**: 包含完整的攻击步骤 -- 🔍 **有验证**: 每个步骤都有验证方法 -- ⚠️ **错误处理**: 包含常见错误和解决方案 -- 📊 **报告生成**: 标准化的攻击报告格式 -- 🔗 **技能链接**: 自动建议下一步技能 - -#### 参考文档技能（技术参考） - -| 技能文件 | 类型 | 行数 | 说明 | -|---------|------|------|------| -| aws-iam-privesc.md | 参考 | 549 | AWS IAM 权限提升技术（15+ 种） | -| aws-metadata-ssrf.md | 参考 | 457 | AWS 元数据服务 SSRF 利用 | -| aws-s3-pentesting.md | 参考 | 516 | AWS S3 存储桶渗透测试 | -| aws-lambda-privesc.md | 参考 | 480 | AWS Lambda 权限提升（9 种） | -| aws-lambda-persistence.md | 参考 | 481 | AWS Lambda 持久化（8 种） | -| gcp-iam-privesc.md | 参考 | 469 | GCP IAM 权限提升（10+ 种） | - -**特点**: -- 📚 **技术全面**: 包含大量攻击技术 -- 🔬 **深入分析**: 技术原理和防御方法 -- 📖 **参考价值**: 作为技术文档使用 - -### 3. Agent 管理系统 - -**agent.md** - Agent 核心管理系统 - -- ✅ 技能激活流程 -- ✅ 技能组合示例 -- ✅ 技能开发指南 -- ✅ 技能文件模板 -- ✅ 用户交互示例 - ---- - -## 🎯 技能文件核心差异 - -### 实战技能 vs 参考文档 - -| 特性 | 实战技能 | 参考文档 | -|------|---------|----------| -| **目标** | 让 AI 执行攻击 | 技术参考和学习 | -| **结构** | 步骤化操作 | 技术分类整理 | -| **验证** | 每步都有验证 | 无强制验证 | -| **错误处理** | 包含常见错误 | 理论性讨论 | -| **报告** | 标准化输出 | 无标准格式 | -| **下一步** | 自动建议 | 无链接 | - -### 实战技能文件结构示例 - -```markdown -## 前置检查 - -检查命令 1 -检查命令 2 -如果失败 → 报错并停止 - -## 攻击方法 - -### 方法 1: 方法名称 - -检查命令 -执行命令 -验证命令 -如果成功 → 下一步 -如果失败 → 尝试其他方法 - -## 验证成功 - -验证命令 -期望输出 - -## 错误处理 - -错误 1: 错误名称 -原因: ... -解决: ... -``` - ---- - -## 💡 使用示例 - -### 场景 1: 用户提供了 AWS 凭证 - -``` -用户: 我有一个 AWS Access Key - -AI: [激活 aws-iam-attack 技能] - 执行前置检查... - 验证凭证有效性... - 枚举用户权限... - 发现可用权限: iam:CreateAccessKey - 执行攻击: 为管理员用户创建密钥 - 验证成功: 确认获得管理员权限 - - [生成报告] - 报告: 已获得管理员级别访问权限 - 建议: - 1. 枚举所有资源 (aws-enum) - 2. 攻击 S3 存储桶 (aws-s3-attack) - 3. 建立持久化 (aws-persistence) - -用户: 继续枚举资源 - -AI: [激活 aws-enum 技能] - 枚举 EC2 实例... - 枚举 S3 存储桶... - 枚举 Lambda 函数... - [返回结果] -``` - -### 场景 2: 发现 SSRF 漏洞 - -``` -用户: 发现一个 SSRF 漏洞 - -AI: [激活 aws-metadata-attack 技能] - 测试 SSRF 是否可访问元数据服务... - ✅ 可以访问 - 获取 IAM 角色... - 获取临时凭证... - 验证凭证权限... - - [生成报告] - 报告: 已通过 SSRF 获取临时凭证 - 建议: - 1. 枚举凭证权限 - 2. 如果权限不足，提升权限 - 3. 建立持久化 -``` - ---- - -## 📊 技能开发进度 - -### Phase 1: AWS 核心攻击（进行中） - -| 技能 | 状态 | 优先级 | -|------|------|--------| -| aws-iam-attack | ✅ 完成 | critical | -| aws-metadata-attack | ✅ 完成 | critical | -| aws-s3-attack | 🚧 待创建 | high | -| aws-lambda-attack | 🚧 待创建 | high | -| aws-enum | 🚧 待创建 | high | -| aws-persistence | 🚧 待创建 | medium | - -### Phase 2: 其他云平台 - -| 技能 | 状态 | 优先级 | -|------|------|--------| -| gcp-iam-attack | 🚧 待创建 | medium | -| gcp-metadata-attack | 🚧 待创建 | medium | -| azure-ad-attack | 🚧 待创建 | medium | -| aliyun-ram-attack | 🚧 待创建 | low | -| tencent-cam-attack | 🚧 待创建 | low | - ---- - -## 🚀 快速开始 - -### 使用实战技能 - -1. **识别用户请求** - - 提取关键信息（平台、凭证、漏洞） - -2. **激活相应技能** - - 使用 `agent.md` 中的决策树 - -3. **执行攻击步骤** - - 按照技能文件中的步骤执行 - -4. **验证结果** - - 使用技能文件中的验证方法 - -5. **生成报告** - - 使用技能文件中的报告格式 - -6. **建议下一步** - - 使用技能文件中的下一步建议 - -### 创建新技能 - -1. **定义技能范围** - - 确定攻击目标和方法 - -2. **编写技能文件** - - 使用 `agent.md` 中的模板 - -3. **测试技能** - - 使用模拟环境测试 - -4. **集成到 Agent** - - 更新 `agent.md` 中的技能清单 - ---- - -## 📖 参考资料 - -### HackTricks Cloud - -- **仓库**: https://github.com/HackTricks-wiki/hacktricks-cloud -- **网站**: https://cloud.hacktricks.wiki/ -- **文件数**: 173 个 README 文件 -- **已克隆**: `/tmp/hacktricks-cloud` - -### 核心资源 - -- Rhino Security Labs: https://rhinosecuritylabs.com/ -- NetSPI: https://www.netspi.com/ -- AWS 官方文档: https://docs.aws.amazon.com/ -- GCP 官方文档: https://cloud.google.com/docs - ---- - -## 🎓 关键改进点 - -### 从文档到技能的转换 - -**之前（文档模式）**: -- 列出攻击技术 -- 提供命令示例 -- 解释技术原理 - -**现在（技能模式）**: -- ✅ 定义触发条件 -- ✅ 前置检查步骤 -- ✅ 自动化攻击流程 -- ✅ 验证和错误处理 -- ✅ 报告和下一步建议 - -### 让 AI 更好地执行 - -1. **明确的触发条件** - - AI 知道何时激活技能 - -2. **步骤化执行** - - AI 可以逐步跟随执行 - -3. **自动化验证** - - AI 可以确认每步成功 - -4. **错误处理** - - AI 可以处理常见错误 - -5. **技能链接** - - AI 可以自动建议下一步 - ---- - -## 📝 总结 - -### 已完成 - -- ✅ 克隆并分析 HackTricks Cloud 仓库 -- ✅ 创建 2 个实战导向技能文件 -- ✅ 创建 6 个参考文档技能文件 -- ✅ 创建 Agent 管理系统 -- ✅ 定义技能开发指南 - -### 进行中 - -- 🚧 创建更多实战技能文件 -- 🚧 完善 Agent 激活流程 -- 🚧 测试技能执行效果 - -### 待完成 - -- ⏳ 创建所有 Phase 1 技能 -- ⏳ 创建 Phase 2 技能 -- ⏳ 创建国内云平台技能 - ---- - -## 🔄 更新记录 - -**2025-03-18 - v1.0.0** - -- 创建实战技能系统 -- 完成 2 个实战技能文件 -- 完成 6 个参考文档技能文件 -- 创建 Agent 管理系统 -- 基于用户反馈重新设计技能文件结构 - -**关键改进**: 从"翻译文档"转向"创建可执行的攻击技能"