Alibabacloud Sas Openclaw Security

"""阿里云 AISC OpenAPI 客户端。"""

from __future__ import annotations

from .base_client import BaseClient


class AiscClient(BaseClient):
    """AISC OpenAPI 客户端（aliyun CLI 实现）。"""

    PRODUCT_NAME = "AISC"

    def __init__(self):
        super().__init__("cn-shanghai")

    def get_ai_agent_plugin_command(self) -> dict:
        """调用 GetAIAgentPluginKey，获取 OpenClaw 安全助手的安装命令。

        注意：底层 API 名称为 GetAIAgentPluginKey，响应字段为 InstallKey，
        但其实际含义是一条完整的 shell 安装命令（install command），
        而非传统意义上的密钥（key/token）。方法名使用 command 以准确描述语义。

        CLI 等价命令：
            aliyun aisc GetAIAgentPluginKey
                --version 2026-01-01
                --endpoint aisc.cn-shanghai.aliyuncs.com
                --force
        """
        # API Action 名称保持原样：GetAIAgentPluginKey（不可改动）
        args = [
            "aisc",
            "GetAIAgentPluginKey",  # API 名称，勿改
            "--version",
            "2026-01-01",
            "--endpoint",
            "aisc.cn-shanghai.aliyuncs.com",
            "--force",
        ]
        return self._run_cli(args)

"""阿里云 OpenAPI 客户端基类。"""

from __future__ import annotations

import json
import logging
import os
import subprocess

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# 异常类
# ---------------------------------------------------------------------------


class CredentialError(Exception):
    """aliyun CLI 未安装或未配置凭据"""


class ProductNotEnabledError(Exception):
    """云产品未开通"""


class APIError(Exception):
    """API 调用失败"""

    def __init__(self, api_name: str, message: str):
        self.api_name = api_name
        super().__init__(f"{api_name}: {message}")


# ---------------------------------------------------------------------------
# 基础客户端
# ---------------------------------------------------------------------------


class BaseClient:
    """阿里云 OpenAPI 客户端基类（aliyun CLI 实现）。"""

    DEFAULT_PAGE_SIZE = 20
    DEFAULT_MAX_PAGES = 3
    MAX_RECORDS = 200

    PRODUCT_NAME: str = ""

    def __init__(self, region: str | None = None):
        self._region = region or os.environ.get("ALICLOUD_REGION_ID", "cn-shanghai")

    def _is_not_enabled_error(self, e: Exception) -> bool:
        """判断是否为产品未开通错误。"""
        msg = str(e).lower()
        return any(
            kw in msg
            for kw in [
                "notopened",
                "not_opened",
                "forbidden",
                "nosubscription",
                "not activated",
                "未开通",
            ]
        )

    def _run_cli(
        self,
        args: list[str],
        region: str | None = None,
    ) -> dict:
        """执行 aliyun CLI 命令并返回解析后的 JSON。

        Args:
            args: 产品 + API + 参数，如 ["sas", "DescribeVulList", "--Type", "cve"]
            region: 覆盖 self._region 的区域，不传则使用 self._region
        """
        effective_region = region or self._region
        cmd = [
            "aliyun",
            "--region", effective_region,
            "--user-agent", "AlibabaCloud-Agent-Skills/alibabacloud-sas-openclaw-security",
        ] + args
        api_name = args[1] if len(args) > 1 else args[0]
        try:
            proc = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=30,
            )
        except subprocess.TimeoutExpired:
            raise APIError(api_name, "CLI 命令超时（30s）")
        except FileNotFoundError:
            raise CredentialError(
                "未找到 aliyun CLI，请先安装：https://help.aliyun.com/zh/cli/installation"
            )

        if proc.returncode != 0:
            err_msg = proc.stderr.strip() or proc.stdout.strip()
            if self._is_not_enabled_error(Exception(err_msg)):
                raise ProductNotEnabledError(
                    f"{self.PRODUCT_NAME}未开通或当前版本不支持 {api_name}。"
                    f"\n原始错误: {err_msg}"
                )
            raise APIError(api_name, err_msg)

        try:
            return json.loads(proc.stdout)
        except json.JSONDecodeError as exc:
            raise APIError(
                api_name,
                f"JSON 解析失败: {exc}\n输出: {proc.stdout[:300]}",
            )

    def _paginate_cli(
        self,
        base_args: list[str],
        items_key: str,
        max_pages: int | None = None,
        page_size: int | None = None,
        region: str | None = None,
    ) -> list[dict]:
        """通用 CLI 分页，自动翻页。

        Args:
            base_args: 不含 --PageSize/--CurrentPage 的 CLI 参数列表
            items_key: 响应 JSON 中条目列表的键名
            max_pages: 最大翻页数
            page_size: 每页条数
            region: 覆盖 self._region 的区域
        """
        ps = page_size or self.DEFAULT_PAGE_SIZE
        mp = max_pages or self.DEFAULT_MAX_PAGES
        all_items: list[dict] = []
        for page in range(1, mp + 1):
            args = base_args + [
                "--page-size",
                str(ps),
                "--current-page",
                str(page),
            ]
            body = self._run_cli(args, region=region)
            items = body.get(items_key, [])
            all_items.extend(items)
            total = body.get("TotalCount") or body.get("PageInfo", {}).get(
                "TotalCount", 0
            )
            if len(all_items) >= total or len(items) < ps:
                break
            if len(all_items) >= self.MAX_RECORDS:
                logger.warning(
                    "已达到 %d 条记录上限，停止分页",
                    self.MAX_RECORDS,
                )
                break
        return all_items

#!/usr/bin/env python3
"""查询 OpenClaw 相关告警。

用法:
  python -m scripts.check_openclaw_alerts
  python -m scripts.check_openclaw_alerts \
      --uuids <UUID> --dealed N
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from scripts.sas_client import SasClient  # noqa: E402


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="查询 OpenClaw 相关告警")
    parser.add_argument(
        "--dealed",
        default="N",
        help="是否已处理: Y/N（默认: N）",
    )
    parser.add_argument(
        "--levels",
        default=None,
        help="告警级别过滤 (serious/suspicious/remind)",
    )
    parser.add_argument(
        "--uuids",
        default=None,
        help="指定主机 UUID（逗号分隔）",
    )
    parser.add_argument(
        "--name",
        default=None,
        help="告警名称过滤",
    )
    parser.add_argument(
        "--region",
        default=None,
        help="区域（默认: cn-shanghai）",
    )
    parser.add_argument(
        "--max-pages",
        type=int,
        default=3,
        help="最大翻页数（默认: 3）",
    )
    parser.add_argument(
        "--output-dir",
        default="output",
        help="输出目录（默认: output）",
    )
    return parser.parse_args()


def format_markdown(alerts: list[dict]) -> str:
    """将告警列表格式化为 Markdown。"""
    lines = [
        "# OpenClaw 告警查询结果",
        "",
        f"查询时间: " f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        f"告警总数: {len(alerts)}",
        "",
    ]

    if not alerts:
        lines.append("未发现相关告警。")
        return "\n".join(lines)

    # 按级别分组
    level_map = {
        "serious": ("🔴 紧急", []),
        "suspicious": ("🟡 可疑", []),
        "remind": ("🟢 提醒", []),
    }
    other = []

    for a in alerts:
        level = a.get("Level", "").lower()
        if level in level_map:
            level_map[level][1].append(a)
        else:
            other.append(a)

    for level_key in ["serious", "suspicious", "remind"]:
        label, group = level_map[level_key]
        if not group:
            continue
        lines.append(f"## {label}（{len(group)} 个）")
        lines.append("")
        lines.append("| 告警名称 | 主机名 | IP | " "首次发现 | 最近发现 |")
        lines.append("|----------|--------|-----|" "----------|----------|")
        for a in group:
            aname = a.get("AlarmEventName") or a.get("Name", "-")
            host = a.get("InstanceName", "-")
            ip = a.get("IntranetIp") or a.get("InternetIp") or "-"
            first = a.get("OccurrenceTime", "-")
            last = a.get("LastTime", "-")
            lines.append(f"| {aname} | {host} | {ip} " f"| {first} | {last} |")
        lines.append("")

    if other:
        lines.append(f"## 其他（{len(other)} 个）")
        lines.append("")
        for a in other:
            aname = a.get("Name", "-")
            host = a.get("InstanceName", "-")
            lines.append(f"- {aname} @ {host}")
        lines.append("")

    return "\n".join(lines)


def main() -> None:
    args = parse_args()
    client = SasClient(region=args.region)

    params = f"dealed={args.dealed}"
    if args.uuids:
        params += f", uuids={args.uuids}"
    if args.levels:
        params += f", levels={args.levels}"
    print(f"[*] 查询告警 ({params})")

    alerts = client.describe_susp_events(
        dealed=args.dealed,
        levels=args.levels,
        uuids=args.uuids,
        name=args.name,
        max_pages=args.max_pages,
    )

    print(f"[+] 发现 {len(alerts)} 个告警")

    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    json_path = out_dir / "openclaw_alerts.json"
    json_path.write_text(
        json.dumps(alerts, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(f"[+] JSON → {json_path}")

    md_path = out_dir / "openclaw_alerts.md"
    md_path.write_text(format_markdown(alerts), encoding="utf-8")
    print(f"[+] Markdown → {md_path}")

    if len(alerts) > 0:
        print()
        print(
            "[!] 安全加固建议: 当前存在未处理告警，"
            "请阅读 remediation_guide.md 进行修复"
        )
        print("[!] 修复指南: references/remediation_guide.md")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""查询 OpenClaw 基线检查结果（按 UUID）。

用法:
  python -m scripts.check_openclaw_baseline \
      --uuid xxxxxxxx
  python -m scripts.check_openclaw_baseline \
      --uuid xxxxxxxx --risk-id 320
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from scripts.sas_client import SasClient  # noqa: E402


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="按 UUID 查询 OpenClaw 基线检查结果")
    parser.add_argument(
        "--uuid",
        required=True,
        help="云安全中心实例 UUID",
    )
    parser.add_argument(
        "--risk-id",
        dest="risk_id",
        type=int,
        default=None,
        help="指定风险项 ID（RiskId）；不传仅查询汇总",
    )
    parser.add_argument(
        "--region",
        default=None,
        help="区域（默认: cn-shanghai）",
    )
    parser.add_argument(
        "--output-dir",
        default="output",
        help="输出目录（默认: output）",
    )
    return parser.parse_args()


STATUS_MAP = {
    1: "未通过",
    2: "验证中",
    3: "已通过",
    6: "已忽略",
    8: "已通过",
}


def _append_summary_table(lines: list[str], summary_items: list[dict]) -> None:
    """追加汇总表格。"""
    lines.append("| 序号 | RiskID | 检查项 | 高危 | 中危 | 状态 |")
    lines.append("|------|---------|--------|------|------|------|")
    for i, item in enumerate(summary_items, 1):
        risk_id = item.get("CheckId") or item.get("RiskId") or item.get("Id") or "-"
        name = item.get("CheckName") or item.get("RiskName") or item.get("Name") or "-"
        high_count = int(item.get("HighWarningCount") or 0)
        medium_count = int(item.get("MediumWarningCount") or 0)
        status = "有风险" if (high_count + medium_count) > 0 else "无风险"
        lines.append(
            f"| {i} | {risk_id} | {name} | "
            f"{high_count} | {medium_count} | {status} |"
        )
    lines.append("")


def _extract_list(body: dict) -> list[dict]:
    """尽可能兼容地提取列表字段。"""
    if not isinstance(body, dict):
        return []
    candidates = [
        "WarningSummarys",
        "CheckWarningSummarys",
        "CheckWarningSummaries",
        "CheckWarningSummaryList",
        "CheckWarningSummary",
        "Warnings",
        "CheckWarnings",
        "List",
    ]
    for key in candidates:
        value = body.get(key)
        if isinstance(value, list):
            return value
    return []


def format_markdown(
    uuid: str,
    summary_items: list[dict],
    details: list[dict],
) -> str:
    """将基线检查结果格式化为 Markdown。"""

    def _risk_count(item: dict) -> int:
        return int(item.get("HighWarningCount") or 0) + int(
            item.get("MediumWarningCount") or 0
        )

    at_risk = [item for item in summary_items if _risk_count(item) > 0]
    fixed = [item for item in summary_items if _risk_count(item) == 0]

    detail_total = sum(len(item.get("warnings", [])) for item in details)
    summary_total = len(summary_items)
    if summary_total > 0:
        total_text = f"检查项总数: {summary_total}"
        risk_text = f"有风险: {len(at_risk)} 项 | " f"无风险: {len(fixed)} 项"
    else:
        total_text = f"检查记录总数: {detail_total}"
        risk_text = "按详情模式展示，不统计汇总项风险数"

    lines = [
        "# OpenClaw 基线检查结果",
        "",
        f"查询时间: " f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        f"UUID: {uuid}",
        total_text,
        risk_text,
        "",
    ]
    if summary_items:
        lines.append("## 汇总")
        lines.append("")
        _append_summary_table(lines, summary_items)
    else:
        lines.append("## 汇总")
        lines.append("")
        lines.append("本次未查询汇总数据（按 risk_id 查询详情模式）。")
        lines.append("")

    lines.append("## 检查项详情")
    lines.append("")
    if not details:
        lines.append("未返回检查项详情。")
        return "\n".join(lines)

    for item in details:
        risk_id = item.get("risk_id", "-")
        warnings = item.get("warnings", [])
        lines.append(f"### RiskId={risk_id}（{len(warnings)} 条）")
        lines.append("")
        if not warnings:
            lines.append("- 无详情记录")
            lines.append("")
            continue

        for w in warnings:
            item_name = (
                w.get("Item")
                or w.get("CheckName")
                or w.get("RiskName")
                or w.get("WarningName")
                or "-"
            )
            item_type = w.get("Type") or w.get("CheckType") or "-"
            check_name = (
                w.get("CheckName")
                or w.get("RiskName")
                or w.get("WarningName")
                or item_name
            )
            status_val = w.get("Status", -1)
            status = STATUS_MAP.get(status_val, str(status_val))
            fix_status_val = w.get("FixStatus", -1)
            fix_status_map = {
                0: "-",
                1: "已修复",
                -1: "-",
            }
            # 业务规则：Status=3（已修复）优先级高于 FixStatus。
            if status_val == 3:
                fix_status = "已修复"
            else:
                fix_status = fix_status_map.get(fix_status_val, str(fix_status_val))
            level = w.get("Level") or w.get("RiskLevel") or "-"
            desc = w.get("Description") or w.get("Desc") or "-"
            lines.append(
                f"- **{item_name}** | 类型: {item_type} | "
                f"级别: {level} | 检查状态: {status} | "
                f"修复状态: {fix_status}"
            )
            if desc != "-":
                lines.append(f"  - 描述: {desc}")
            if check_name != item_name and check_name != "-":
                lines.append(f"  - 检查项: {check_name}")
            check_warning_id = w.get("CheckWarningId")
            if check_warning_id is not None:
                lines.append(f"  - CheckWarningId: {check_warning_id}")
        lines.append("")

    return "\n".join(lines)


def main() -> None:
    args = parse_args()
    client = SasClient(region=args.region)

    summary_body: dict = {}
    summary_items: list[dict] = []
    details: list[dict] = []

    if args.risk_id is None:
        print(f"[*] 查询基线汇总 (uuid={args.uuid})")
        summary_body = client.describe_check_warning_summary(uuids=args.uuid)
        summary_items = _extract_list(summary_body)
        at_risk = [
            item
            for item in summary_items
            if (item.get("HighWarningCount") or 0)
            + (item.get("MediumWarningCount") or 0)
            > 0
        ]
        fixed = [
            item
            for item in summary_items
            if (item.get("HighWarningCount") or 0)
            + (item.get("MediumWarningCount") or 0)
            == 0
        ]
        print(
            f"[+] 共 {len(summary_items)} 项: "
            f"有风险 {len(at_risk)} 项, "
            f"无风险 {len(fixed)} 项"
        )
    else:
        print(f"[*] 查询基线详情 (uuid={args.uuid}, " f"risk_id={args.risk_id})")
        detail_body = client.describe_check_warnings(
            uuid=args.uuid,
            risk_id=args.risk_id,
        )
        detail_items = _extract_list(detail_body)
        details.append(
            {
                "risk_id": args.risk_id,
                "warnings": detail_items,
                "raw": detail_body,
            }
        )
        print(f"[+] risk_id={args.risk_id} " f"详情 {len(detail_items)} 条")
        at_risk = detail_items

    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    json_path = out_dir / "openclaw_baseline.json"
    json_path.write_text(
        json.dumps(
            {
                "uuid": args.uuid,
                "query_mode": ("summary" if args.risk_id is None else "detail"),
                "summary": summary_body,
                "details": details,
            },
            ensure_ascii=False,
            indent=2,
        ),
        encoding="utf-8",
    )
    print(f"[+] JSON → {json_path}")

    md_path = out_dir / "openclaw_baseline.md"
    md_path.write_text(
        format_markdown(
            uuid=args.uuid,
            summary_items=summary_items,
            details=details,
        ),
        encoding="utf-8",
    )
    print(f"[+] Markdown → {md_path}")

    if len(at_risk) > 0:
        print()
        print(
            "[!] 安全加固建议: 当前存在基线风险，"
            "请阅读 remediation_guide.md 进行修复"
        )
        print("[!] 修复指南: references/remediation_guide.md")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""查询 OpenClaw 相关漏洞。

用法:
  python -m scripts.check_openclaw_vulns
  python -m scripts.check_openclaw_vulns \
      --type emg --dealed n
  python -m scripts.check_openclaw_vulns \
      --name "emg:SCA:AVD-2026-1860246" --type emg
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from scripts.sas_client import SasClient  # noqa: E402


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="查询 OpenClaw 相关漏洞")
    parser.add_argument(
        "--name",
        default="emg:SCA:AVD-2026-1860246",
        help="漏洞名称精确匹配" "（默认: emg:SCA:AVD-2026-1860246）",
    )
    parser.add_argument(
        "--type",
        default="emg",
        help="漏洞类型: cve/sys/cms/emg（默认: emg）",
    )
    parser.add_argument(
        "--dealed",
        default="n",
        help="是否已处理: y/n（默认: n）",
    )
    parser.add_argument(
        "--necessity",
        default=None,
        help="修复紧急度过滤",
    )
    parser.add_argument(
        "--uuids",
        default=None,
        help="指定主机 UUID（逗号分隔）",
    )
    parser.add_argument(
        "--region",
        default=None,
        help="区域（默认: cn-shanghai）",
    )
    parser.add_argument(
        "--max-pages",
        type=int,
        default=3,
        help="最大翻页数（默认: 3）",
    )
    parser.add_argument(
        "--output-dir",
        default="output",
        help="输出目录（默认: output）",
    )
    return parser.parse_args()


def format_markdown(vulns: list[dict]) -> str:
    """将漏洞列表格式化为 Markdown。"""
    lines = [
        "# OpenClaw 漏洞查询结果",
        "",
        f"查询时间: " f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        f"漏洞总数: {len(vulns)}",
        "",
    ]

    if not vulns:
        lines.append("未发现相关漏洞。")
        return "\n".join(lines)

    # 按严重程度分组
    high, medium, low = [], [], []
    for v in vulns:
        necessity = v.get("Necessity", "")
        if necessity == "asap":
            high.append(v)
        elif necessity == "later":
            medium.append(v)
        else:
            low.append(v)

    for label, group in [
        ("🔴 高危", high),
        ("🟡 中危", medium),
        ("🟢 低危", low),
    ]:
        if not group:
            continue
        lines.append(f"## {label}（{len(group)} 个）")
        lines.append("")
        lines.append(
            "| 漏洞名称 | 主机/IP | 区域 | 版本命中条件 | "
            "当前版本 | 首次发现 | 状态 |"
        )
        lines.append(
            "|----------|---------|------|--------------|"
            "----------|----------|------|"
        )
        for v in group:
            vname = v.get("AliasName") or v.get("Name", "-")
            host = v.get("InstanceName", "-")
            ip = v.get("InternetIp") or v.get("IntranetIp") or "-"
            host_ip = f"{host}/{ip}"
            region = v.get("RegionId", "-")
            first = _format_ts(v.get("FirstTs"))
            status = "已处理" if v.get("Status") == 0 else "未处理"
            if v.get("RealRisk") is True:
                status = f"{status}（真实风险）"

            match_expr, full_version = _extract_version_info(v)
            lines.append(
                f"| {vname} | {host_ip} | {region} | "
                f"{match_expr} | {full_version} | {first} | {status} |"
            )
        lines.append("")

        lines.append("### 详情")
        lines.append("")
        for i, v in enumerate(group, 1):
            vname = v.get("AliasName") or v.get("Name", "-")
            name = v.get("Name", "-")
            uuid = v.get("Uuid", "-")
            primary_id = v.get("PrimaryId", "-")
            last = _format_ts(v.get("LastTs"))
            match_expr, full_version = _extract_version_info(v)
            lines.append(f"{i}. **{vname}**")
            lines.append(f"   - 漏洞ID: `{name}` / PrimaryId: `{primary_id}`")
            lines.append(f"   - UUID: `{uuid}`，最近发现: {last}")
            lines.append(f"   - 命中条件: `{match_expr}`，当前版本: `{full_version}`")
        lines.append("")

    return "\n".join(lines)


def _format_ts(ts: object) -> str:
    """毫秒时间戳转可读时间。"""
    if ts is None:
        return "-"
    try:
        value = int(ts)
        # 返回为毫秒时间戳
        if value > 10_000_000_000:
            value = value // 1000
        return datetime.fromtimestamp(value).strftime("%Y-%m-%d %H:%M:%S")
    except (TypeError, ValueError, OSError):
        return str(ts)


def _extract_version_info(vuln: dict) -> tuple[str, str]:
    """提取版本命中条件和当前版本。"""
    extend = vuln.get("ExtendContentJson", {})
    if not isinstance(extend, dict):
        return "-", "-"
    rpm_list = extend.get("RpmEntityList", [])
    if not isinstance(rpm_list, list) or not rpm_list:
        return "-", "-"
    entity = rpm_list[0] if isinstance(rpm_list[0], dict) else {}
    match_list = entity.get("MatchList", [])
    match_expr = "-"
    if isinstance(match_list, list) and match_list:
        match_expr = str(match_list[0])
    full_version = entity.get("FullVersion") or entity.get("Version") or "-"
    return match_expr, str(full_version)


def main() -> None:
    args = parse_args()
    client = SasClient(region=args.region)

    params = f"type={args.type}, dealed={args.dealed}"
    if args.name:
        params += f", name={args.name}"
    print(f"[*] 查询漏洞 ({params})")

    vulns = client.describe_vul_list(
        vul_type=args.type,
        dealed=args.dealed,
        name=args.name,
        necessity=args.necessity,
        uuids=args.uuids,
        max_pages=args.max_pages,
    )

    print(f"[+] 发现 {len(vulns)} 个漏洞")

    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    json_path = out_dir / "openclaw_vulns.json"
    json_path.write_text(
        json.dumps(vulns, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(f"[+] JSON → {json_path}")

    md_path = out_dir / "openclaw_vulns.md"
    md_path.write_text(format_markdown(vulns), encoding="utf-8")
    print(f"[+] Markdown → {md_path}")

    if len(vulns) > 0:
        print()
        print(
            "[!] 安全加固建议: 当前存在未修复漏洞，"
            "请阅读 remediation_guide.md 进行修复"
        )
        print("[!] 修复指南: references/remediation_guide.md")


if __name__ == "__main__":
    main()

"""阿里云 ECS（云服务器）OpenAPI 客户端。

全部通过 aliyun CLI 实现，无需 SDK 依赖。
"""

from __future__ import annotations
import base64
import time

from .base_client import APIError, BaseClient


class EcsClient(BaseClient):
    """ECS OpenAPI 客户端（aliyun CLI 实现）。"""

    PRODUCT_NAME = "云服务器 ECS"
    POLL_RETRY_TIMES = 3
    POLL_RETRY_DELAY = 2

    def __init__(self, region: str | None = None):
        super().__init__(region or "cn-hangzhou")

    def run_command(
        self,
        instance_ids: list[str],
        command_content: str,
        command_type: str = "RunShellScript",
        region: str | None = None,
        name: str | None = None,
        description: str | None = None,
        timeout: int | None = None,
        working_dir: str | None = None,
        username: str | None = None,
        keep_command: bool | None = None,
    ) -> dict:
        """通过云助手在 ECS 实例上执行命令。

        Args:
            instance_ids: ECS 实例 ID 列表
            command_content: 命令内容（明文，自动进行 Base64 编码）
            command_type: 命令类型（默认 RunShellScript）
            region: 区域 ID，未传时使用客户端默认区域
            name: 命令名称
            description: 命令描述
            timeout: 超时时间（秒）
            working_dir: 执行目录
            username: 执行用户
            keep_command: 是否保留命令

        CLI 等价命令：
            aliyun ecs run-command --biz-region-id <region>
                --type RunShellScript --command-content <b64> --content-encoding Base64
                --instance-id <id1> [<id2> ...]
                [--name <name>] [--timeout <sec>] ...
        """
        if not instance_ids:
            raise ValueError("instance_ids 不能为空")

        effective_region = region or self._region
        command_b64 = base64.b64encode(command_content.encode()).decode()
        args = [
            "ecs",
            "run-command",
            "--biz-region-id",
            effective_region,
            "--type",
            command_type,
            "--command-content",
            command_b64,
            "--content-encoding",
            "Base64",
        ]
        args += ["--instance-id"] + instance_ids
        if name:
            args += ["--name", name]
        if description:
            args += ["--description", description]
        if timeout is not None:
            args += ["--timeout", str(timeout)]
        if working_dir:
            args += ["--working-dir", working_dir]
        if username:
            args += ["--username", username]
        if keep_command is not None:
            args += ["--keep-command", str(keep_command).lower()]

        return self._run_cli(args, region=effective_region)

    def get_command_result_by_invoke_id(
        self,
        invoke_id: str,
        instance_id: str,
    ) -> dict:
        """按 invoke_id 查询单台实例的命令执行结果。

        CLI 等价命令：
            aliyun ecs describe-invocation-results --biz-region-id <region>
                --invoke-id <invoke_id> --instance-id <instance_id>
        """
        args = [
            "ecs",
            "describe-invocation-results",
            "--biz-region-id",
            self._region,
            "--invoke-id",
            invoke_id,
            "--instance-id",
            instance_id,
        ]
        body = self._run_cli(args)

        invocation = body.get("Invocation", {})
        results = invocation.get("InvocationResults", {})
        result_list = results.get("InvocationResult", [])
        if not result_list:
            raise APIError(
                "DescribeInvocationResults",
                f"未找到执行结果: invoke_id={invoke_id}, " f"instance_id={instance_id}",
            )
        item = result_list[0]
        if not isinstance(item, dict):
            raise APIError(
                "DescribeInvocationResults",
                "执行结果格式异常",
            )
        return item

    @staticmethod
    def _is_retryable_poll_error(err: Exception) -> bool:
        """判断轮询结果查询是否命中了可重试的瞬时错误。"""
        msg = str(err).lower()
        return any(
            keyword in msg
            for keyword in [
                "connection aborted",
                "remotedisconnected",
                "remote end closed connection without response",
                "read timed out",
                "connect timeout",
                "connection reset by peer",
                "temporarily unavailable",
                "service unavailable",
                "502 bad gateway",
                "503 service unavailable",
                "504 gateway timeout",
            ]
        )

    def get_command_result_with_retry(
        self,
        invoke_id: str,
        instance_id: str,
        retries: int | None = None,
        retry_delay: int | None = None,
    ) -> dict:
        """查询执行结果，并对瞬时网络错误做短暂重试。"""
        max_retries = self.POLL_RETRY_TIMES if retries is None else retries
        base_delay = self.POLL_RETRY_DELAY if retry_delay is None else retry_delay

        last_error: Exception | None = None
        for attempt in range(max_retries + 1):
            try:
                return self.get_command_result_by_invoke_id(
                    invoke_id=invoke_id,
                    instance_id=instance_id,
                )
            except Exception as err:
                last_error = err
                if attempt >= max_retries or not self._is_retryable_poll_error(err):
                    raise
                time.sleep(base_delay * (attempt + 1))

        raise APIError(
            "DescribeInvocationResults",
            f"查询执行结果失败: {last_error}",
        )

    def wait_command_result(
        self,
        invoke_id: str,
        instance_id: str,
        timeout: int = 60,
        max_polls: int = 10,
        allow_nonzero_exit: bool = False,
    ) -> dict:
        """循环等待命令执行结果并返回最终状态。"""
        if max_polls <= 0:
            raise ValueError("max_polls 必须大于 0")
        if timeout <= 0:
            raise ValueError("timeout 必须大于 0")

        sleep_interval = max(timeout / max_polls, 1)
        pending_status = {"Pending", "Running", "Stopping"}

        last_result: dict | None = None
        for _ in range(max_polls):
            time.sleep(sleep_interval)
            result = self.get_command_result_with_retry(
                invoke_id=invoke_id,
                instance_id=instance_id,
            )
            last_result = result
            status = result.get("InvocationStatus", "")
            if status in pending_status:
                continue

            output_b64 = result.get("Output", "")
            output = self._decode_output(output_b64)
            final_result = {
                "InvocationStatus": status,
                "ExitCode": result.get("ExitCode"),
                "Output": output,
                "RawOutput": output_b64,
                "ErrorCode": result.get("ErrorCode"),
                "ErrorInfo": result.get("ErrorInfo"),
                "Result": result,
            }

            if status == "Success" or allow_nonzero_exit:
                return final_result

            raise APIError(
                "DescribeInvocationResults",
                f"命令执行失败: status={status}, "
                f"exit_code={result.get('ExitCode')}, "
                f"error_code={result.get('ErrorCode')}, "
                f"error_info={result.get('ErrorInfo')}, "
                f"output={repr(output)}",
            )

        raise TimeoutError(
            f"获取命令执行结果超时: timeout={timeout}s, "
            f"invoke_id={invoke_id}, last_result={last_result}"
        )

    @staticmethod
    def _decode_output(output_b64: str) -> str:
        """解码 Base64 输出，失败时返回原文。"""
        if not output_b64:
            return ""
        try:
            return base64.b64decode(output_b64).decode("utf-8", errors="replace")
        except Exception:
            return output_b64

#!/usr/bin/env python3
"""生成 OpenClaw 安全日报。

汇总所有安全维度：实例、漏洞、基线、告警。

用法:
  python -m scripts.generate_security_report
  python -m scripts.generate_security_report \
      --output-dir output
"""

from __future__ import annotations

import argparse
import json
import sys
import traceback
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from scripts.sas_client import SasClient  # noqa: E402


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="生成 OpenClaw 安全日报")
    parser.add_argument(
        "--name-pattern",
        default="openclaw",
        help="SCA 组件名称模糊匹配（默认: openclaw）",
    )
    parser.add_argument(
        "--biz",
        default="sca_ai",
        help="业务类型（默认: sca_ai）",
    )
    parser.add_argument(
        "--vuln-name",
        default="emg:SCA:AVD-2026-1860246",
        help="漏洞名称精确匹配" "（默认: emg:SCA:AVD-2026-1860246）",
    )
    parser.add_argument(
        "--region",
        default=None,
        help="区域（默认: cn-shanghai）",
    )
    parser.add_argument(
        "--output-dir",
        default="output",
        help="输出目录（默认: output）",
    )
    return parser.parse_args()


def _safe_query(label: str, fn):
    """安全调用查询函数，失败返回空列表。"""
    try:
        result = fn()
        print(f"  [+] {label}: {len(result)} 条")
        return result
    except Exception as e:
        print(f"  [!] {label} 查询失败: {e}")
        traceback.print_exc()
        return []


def generate_report(
    instances: list[dict],
    vulns: list[dict],
    baseline: list[dict],
    alerts: list[dict],
) -> str:
    """生成 Markdown 安全日报。"""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    date_str = datetime.now().strftime("%Y-%m-%d")

    lines = [
        f"# OpenClaw 安全日报 - {date_str}",
        "",
        f"生成时间: {now}",
        "",
        "---",
        "",
        "## 概览",
        "",
        "| 维度 | 数量 | 状态 |",
        "|------|------|------|",
    ]

    # 概览统计
    inst_status = "✅ 正常" if instances else "⚠️ 未发现实例"
    lines.append(f"| OpenClaw 实例 | {len(instances)} " f"| {inst_status} |")

    vuln_status = "🔴 存在风险" if vulns else "✅ 无漏洞"
    lines.append(f"| 未修复漏洞 | {len(vulns)} " f"| {vuln_status} |")

    base_at_risk = [
        m for m in baseline
        if int(m.get("HighWarningCount") or 0)
        + int(m.get("MediumWarningCount") or 0) > 0
    ]
    if base_at_risk:
        base_status = "🔴 存在风险"
    else:
        base_status = "✅ 基线合规"
    lines.append(
        f"| 基线风险项 | "
        f"{len(base_at_risk)}/{len(baseline)} "
        f"| {base_status} |"
    )

    alert_status = "🔴 存在告警" if alerts else "✅ 无告警"
    lines.append(f"| 未处理告警 | {len(alerts)} " f"| {alert_status} |")

    lines.append("")

    # 实例详情
    lines.append("---")
    lines.append("")
    lines.append("## 1. OpenClaw 实例")
    lines.append("")
    if instances:
        lines.append("| 主机名 | IP | 组件名 | 版本 |")
        lines.append("|--------|----|--------|------|")
        for inst in instances[:20]:
            host = inst.get("InstanceName", "-")
            ip = inst.get("Ip") or inst.get("InternetIp", "-")
            name = inst.get("Name", "-")
            ver = inst.get("Version", "-")
            lines.append(f"| {host} | {ip} | {name} | {ver} |")
        if len(instances) > 20:
            lines.append(f"\n> 仅显示前 20 条，" f"共 {len(instances)} 条")
    else:
        lines.append("未发现 OpenClaw 实例。")
    lines.append("")

    # 漏洞详情
    lines.append("---")
    lines.append("")
    lines.append("## 2. 漏洞风险")
    lines.append("")
    if vulns:
        high = [v for v in vulns if v.get("Necessity") == "asap"]
        med = [v for v in vulns if v.get("Necessity") == "later"]
        low = [v for v in vulns if v.get("Necessity") not in ("asap", "later")]
        lines.append(f"- 🔴 高危: {len(high)} 个")
        lines.append(f"- 🟡 中危: {len(med)} 个")
        lines.append(f"- 🟢 低危: {len(low)} 个")
        lines.append("")
        for v in vulns[:10]:
            vname = v.get("AliasName") or v.get("Name", "-")
            host = v.get("InstanceName", "-")
            nec = v.get("Necessity", "-")
            lines.append(f"- **{vname}** @ {host} " f"(紧急度: {nec})")
        if len(vulns) > 10:
            lines.append(f"\n> 仅显示前 10 条，" f"共 {len(vulns)} 条")
    else:
        lines.append("未发现相关漏洞。✅")
    lines.append("")

    # 基线详情
    lines.append("---")
    lines.append("")
    lines.append("## 3. 基线检查")
    lines.append("")
    if baseline:
        if base_at_risk:
            total_high = sum(int(m.get("HighWarningCount") or 0) for m in base_at_risk)
            total_med = sum(int(m.get("MediumWarningCount") or 0) for m in base_at_risk)
            lines.append(
                f"🔴 **{len(base_at_risk)} 个风险项未通过基线检查"
                f"（高危: {total_high}，中危: {total_med}）：**"
            )
            lines.append("")
            for m in base_at_risk[:10]:
                name = (
                    m.get("CheckName") or m.get("RiskName") or m.get("Name") or "-"
                )
                high = int(m.get("HighWarningCount") or 0)
                med = int(m.get("MediumWarningCount") or 0)
                lines.append(f"- **{name}**: 高危 {high} / 中危 {med}")
            if len(base_at_risk) > 10:
                lines.append(f"\n> 仅显示前 10 条，共 {len(base_at_risk)} 条")
            lines.append("")
        else:
            lines.append("基线检查通过。✅")
    else:
        lines.append("未发现基线检查记录。")
    lines.append("")

    # 告警详情
    lines.append("---")
    lines.append("")
    lines.append("## 4. 安全告警")
    lines.append("")
    if alerts:
        for a in alerts[:10]:
            aname = a.get("AlarmEventName") or a.get("Name", "-")
            host = a.get("InstanceName", "-")
            level = a.get("Level", "-")
            lines.append(f"- **{aname}** @ {host} " f"(级别: {level})")
        if len(alerts) > 10:
            lines.append(f"\n> 仅显示前 10 条，" f"共 {len(alerts)} 条")
    else:
        lines.append("未发现安全告警。✅")
    lines.append("")

    # 建议
    lines.append("---")
    lines.append("")
    lines.append("## 5. 安全建议")
    lines.append("")
    recommendations = []
    if vulns:
        recommendations.append(
            "1. **漏洞修复**: 优先处理高危漏洞，" "参考 remediation_guide.md"
        )
    if base_at_risk:
        recommendations.append(
            "2. **基线加固**: 修复基线不合规项，" "特别关注弱口令和权限配置"
        )
    if alerts:
        recommendations.append("3. **告警处置**: 及时处理安全告警，" "排查可疑行为")
    recommendations.append(
        f"{len(recommendations) + 1}. "
        "**安全护栏**: 安装阿里云安全护栏，"
        "实时拦截高危命令和异常行为"
    )
    if not any("漏洞" in r for r in recommendations):
        if not any("基线" in r for r in recommendations):
            if not any("告警" in r for r in recommendations):
                if len(recommendations) == 1:
                    recommendations.insert(0, "当前安全状态良好，" "建议保持定期巡检。")
    lines.extend(recommendations)
    lines.append("")

    return "\n".join(lines)


def main() -> None:
    args = parse_args()
    client = SasClient(region=args.region)

    print("[*] 开始生成 OpenClaw 安全日报...")

    # 依次查询
    instances = _safe_query(
        "实例",
        lambda: client.describe_property_sca_detail(
            biz=args.biz,
            sca_name_pattern=args.name_pattern,
        ),
    )

    vulns = _safe_query(
        "漏洞",
        lambda: client.describe_vul_list(
            vul_type="emg",
            dealed="n",
            name=args.vuln_name,
        ),
    )

    baseline = _safe_query(
        "基线",
        lambda: client.describe_check_warning_summary().get("WarningSummarys", []),
    )

    alerts = _safe_query(
        "告警",
        lambda: client.describe_susp_events(
            dealed="N",
        ),
    )

    # 生成报告
    report = generate_report(
        instances,
        vulns,
        baseline,
        alerts,
    )

    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    date_str = datetime.now().strftime("%Y%m%d")
    md_path = out_dir / f"security_report_{date_str}.md"
    md_path.write_text(report, encoding="utf-8")
    print(f"[+] 安全日报 → {md_path}")

    # 同时保存原始数据
    raw = {
        "generated_at": datetime.now().isoformat(),
        "instances": instances,
        "vulns": vulns,
        "baseline": baseline,
        "alerts": alerts,
    }
    json_path = out_dir / f"security_report_{date_str}.json"
    json_path.write_text(
        json.dumps(raw, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(f"[+] 原始数据 → {json_path}")
    print("[+] 安全日报生成完成！")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""调用 AISC GetAIAgentPluginKey，获取 OpenClaw 安全助手的安装命令。

注意：API 名称为 GetAIAgentPluginKey、字段为 InstallKey，但实际返回的是
一条可直接执行的 shell 安装命令，而非传统意义的密钥。

用法:
  python -m scripts.get_ai_agent_plugin_key
"""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from scripts.aisc_client import AiscClient  # noqa: E402


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        # API 名称保持原样（GetAIAgentPluginKey），但实际获取的是安装命令
        description="调用 AISC GetAIAgentPluginKey，获取 OpenClaw 安全助手安装命令"
    )
    parser.add_argument(
        "--output-dir",
        default="output",
        help="输出目录（默认: output）",
    )
    return parser.parse_args()


def format_markdown(result: dict) -> str:
    """将调用结果格式化为 Markdown。"""
    data = result.get("Data", {})
    request_id = result.get("RequestId", "-")
    # API 字段名为 InstallKey，实际含义是安装命令（install command）
    install_command = data.get("InstallKey", "-")  # 字段名 InstallKey 为 API 规定，勿改
    expire_time = data.get("ExpireTime", "-")

    lines = [
        # 标题保留 API 名称，方便源码追溯
        "# AISC GetAIAgentPluginKey 调用结果（获取安装命令）",
        "",
        f"调用时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "",
        "## 响应概览",
        "",
        f"- RequestId: `{request_id}`",
        f"- ExpireTime: `{expire_time}`",
        "",
        "## 安装命令",
        "",
        "```bash",
        install_command,  # 这里输出的是安装命令字符串
        "```",
        "",
        "## 完整响应",
        "",
        "```json",
        json.dumps(result, ensure_ascii=False, indent=2),
        "```",
    ]
    return "\n".join(lines)


def main() -> None:
    args = parse_args()
    client = AiscClient()

    # API 名称为 GetAIAgentPluginKey，返回值为安装命令
    print("[*] 调用 GetAIAgentPluginKey（获取安装命令）")
    result = client.get_ai_agent_plugin_command()  # 方法已改名，底层调用的 API 不变
    print("[+] 调用完成")

    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    # 输出文件名使用 plugin_command 以准确语义
    raw_path = out_dir / "ai_agent_plugin_command.json"
    raw_path.write_text(
        json.dumps(
            {
                "called_at": datetime.now().isoformat(),
                "response": result,
            },
            ensure_ascii=False,
            indent=2,
        ),
        encoding="utf-8",
    )
    print(f"[+] JSON → {raw_path}")

    md_path = out_dir / "ai_agent_plugin_command.md"
    md_path.write_text(
        format_markdown(result),
        encoding="utf-8",
    )
    print(f"[+] Markdown → {md_path}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""获取安全护栏安装命令并通过云助手执行安装。

用法:
  python -m scripts.install_security_guardrail \
      --instance-ids i-abc123,i-def456
"""

from __future__ import annotations

import argparse
import json
import shlex
import sys
from datetime import datetime
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from scripts.aisc_client import AiscClient  # noqa: E402
from scripts.ecs_client import EcsClient  # noqa: E402

INSTALL_SUCCESS_MARKERS = ("=== 安装配置完成 ===",)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="获取安全护栏安装命令并通过云助手执行安装"
    )
    parser.add_argument(
        "--instance-ids",
        required=True,
        help="ECS 实例 ID 列表（逗号分隔）",
    )
    parser.add_argument(
        "--type",
        default="RunShellScript",
        help="云助手命令类型（默认: RunShellScript）",
    )
    parser.add_argument(
        "--name",
        default="openclaw-security-guardrail-install",
        help="云助手命令名称",
    )
    parser.add_argument(
        "--description",
        default="Install Aliyun security guardrail",
        help="云助手命令描述",
    )
    parser.add_argument(
        "--timeout",
        type=int,
        default=600,
        help="云助手命令超时时间（秒，默认: 600）",
    )
    parser.add_argument(
        "--max-polls",
        type=int,
        default=20,
        help="最大轮询次数（默认: 20）",
    )
    parser.add_argument(
        "--working-dir",
        default=None,
        help="执行目录",
    )
    parser.add_argument(
        "--username",
        default=None,
        help="执行用户",
    )
    parser.add_argument(
        "--keep-command",
        action="store_true",
        help="是否保留命令定义（默认: 否）",
    )
    parser.add_argument(
        "--region",
        default=None,
        help="ECS 区域（默认: cn-hangzhou）",
    )
    parser.add_argument(
        "--output-dir",
        default="output",
        help="输出目录（默认: output）",
    )
    return parser.parse_args()


def _to_instance_ids(raw: str) -> list[str]:
    """将逗号分隔的实例 ID 转为列表。"""
    ids = [item.strip() for item in raw.split(",")]
    result = [item for item in ids if item]
    if not result:
        raise ValueError("请至少提供一个有效的 --instance-ids")
    return result


def _extract_install_payload(result: dict[str, Any]) -> tuple[str, int | None]:
    """提取安装命令和过期时间。"""
    # API 直接返回 {"Data": {...}, "RequestId": "...}，无 body 包裹
    data = result.get("Data", {})
    if not isinstance(data, dict):
        # API GetAIAgentPluginKey 返回结构异常，缺少 Data 段
        raise ValueError("GetAIAgentPluginKey 返回缺少 Data（无法提取安装命令）")

    # API 字段名为 InstallKey，实际内容是一条可直接执行的 shell 安装命令
    install_command = data.get("InstallKey")  # 字段名 InstallKey 为 API 规定，勿改
    if not isinstance(install_command, str) or not install_command:
        # InstallKey 字段必须为非空字符串，否则无法下发安装命令
        raise ValueError("GetAIAgentPluginKey 返回缺少有效的 InstallKey（安装命令为空）")

    expire_time = data.get("ExpireTime")
    if isinstance(expire_time, bool):
        expire_time = None
    elif isinstance(expire_time, (int, float)):
        expire_time = int(expire_time)
    else:
        expire_time = None

    return install_command, expire_time


def _format_expire_time(expire_time: int | None) -> str:
    """格式化命令过期时间。"""
    if expire_time is None:
        return "-"
    return datetime.fromtimestamp(expire_time).strftime("%Y-%m-%d %H:%M:%S")


def _wrap_bash_c(command: str) -> str:
    """将安装命令包装为 bash -c 执行。"""
    return f"bash -c {shlex.quote(command)}"


def _is_install_success(result: dict[str, Any]) -> bool:
    """根据云助手结果和安装输出判断是否视为安装成功。"""
    if result.get("InvocationStatus") == "Success":
        return True

    output = result.get("Output", "")
    if not isinstance(output, str):
        return False
    return any(marker in output for marker in INSTALL_SUCCESS_MARKERS)


def _display_status(result: dict[str, Any]) -> str:
    """返回对用户更友好的安装状态。"""
    if _is_install_success(result):
        if result.get("InvocationStatus") == "Success":
            return "Success"
        return "SuccessByMarker"
    return str(result.get("InvocationStatus", "-"))


def format_markdown(
    key_result: dict[str, Any],
    install_command: str,
    remote_command: str,
    expire_time: int | None,
    instance_ids: list[str],
    region: str,
    run_result: dict[str, Any] | None,
    execution_results: dict[str, dict[str, Any]],
) -> str:
    """将安装流程结果格式化为 Markdown。"""
    request_id = key_result.get("RequestId", "-")

    lines = [
        "# 安全护栏安装结果",
        "",
        f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "",
        "## 安装参数",
        "",
        f"- ECS 区域: `{region}`",
        f"- 实例 ID: `{', '.join(instance_ids)}`",
        "",
        "## 安装命令",
        "",
        f"- RequestId: `{request_id}`",
        f"- 过期时间: `{_format_expire_time(expire_time)}`",
        "",
        "原始安装命令:",
        "",
        "```bash",
        install_command,
        "```",
        "",
        "云助手下发命令:",
        "",
        "```bash",
        remote_command,
        "```",
        "",
    ]

    run_result = run_result or {}
    lines.extend(
        [
            "## 云助手返回",
            "",
            f"- CommandId: `{run_result.get('CommandId', '-')}`",
            f"- InvokeId: `{run_result.get('InvokeId', '-')}`",
            "",
            "## 执行结果",
            "",
            "| 实例 ID | 状态 | ExitCode |",
            "|---------|------|----------|",
        ]
    )

    for instance_id in instance_ids:
        one = execution_results.get(instance_id, {})
        lines.append(
            f"| {instance_id} | "
            f"{_display_status(one)} | "
            f"{one.get('ExitCode', '-')} |"
        )

    for instance_id in instance_ids:
        one = execution_results.get(instance_id, {})
        output = one.get("Output", "")
        error = one.get("Error")
        lines.extend(
            [
                "",
                f"### 输出（{instance_id}）",
                "",
                "```text",
                error or output,
                "```",
            ]
        )

    return "\n".join(lines)


def main() -> None:
    args = parse_args()
    instance_ids = _to_instance_ids(args.instance_ids)
    region = args.region or "cn-shanghai"

    # 调用 GetAIAgentPluginKey 获取安装命令（API 字段 InstallKey 实际为 shell 命令）
    print("[*] 调用 API 获取安装命令")
    aisc_client = AiscClient()
    command_result = aisc_client.get_ai_agent_plugin_command()  # 方法已改名，底层 API 不变
    install_command, expire_time = _extract_install_payload(command_result)
    remote_command = _wrap_bash_c(install_command)
    print("[+] 已获取安全护栏安装命令")
    print("")
    print("[*] 将通过云助手执行以下完整命令:")
    print(remote_command)
    print("")
    print("[+] 安装命令过期时间:" f" {_format_expire_time(expire_time)}")

    run_result: dict[str, Any] | None = None
    execution_results: dict[str, dict[str, Any]] = {}
    has_failure = False

    ecs_client = EcsClient(region=args.region)
    params = f"region={region}, instances={len(instance_ids)}, " f"type={args.type}"
    print(f"[*] 提交云助手安装命令 ({params})")
    run_result = ecs_client.run_command(
        instance_ids=instance_ids,
        command_content=remote_command,
        command_type=args.type,
        region=args.region,
        name=args.name,
        description=args.description,
        timeout=args.timeout,
        working_dir=args.working_dir,
        username=args.username,
        keep_command=True if args.keep_command else None,
    )
    print(
        "[+] 提交成功:"
        f" CommandId={run_result.get('CommandId', '-')},"
        f" InvokeId={run_result.get('InvokeId', '-')}"
    )

    invoke_id = run_result.get("InvokeId")
    if not invoke_id:
        raise ValueError("RunCommand 返回缺少 InvokeId，无法查询执行结果")

    print(
        "[*] 正在轮询安装结果 " f"(timeout={args.timeout}s, max_polls={args.max_polls})"
    )
    for instance_id in instance_ids:
        print(f"[*] 等待实例执行完成: {instance_id}")
        try:
            one_result = ecs_client.wait_command_result(
                invoke_id=invoke_id,
                instance_id=instance_id,
                timeout=args.timeout,
                max_polls=args.max_polls,
                allow_nonzero_exit=True,
            )
            one_result["DisplayStatus"] = _display_status(one_result)
            execution_results[instance_id] = one_result
            if _is_install_success(one_result):
                print(
                    "[+] 安装完成:"
                    f" instance={instance_id}, "
                    f"status={one_result.get('DisplayStatus', '-')}, "
                    f"exit_code={one_result.get('ExitCode', '-')}"
                )
            else:
                has_failure = True
                print(
                    "[!] 执行失败:"
                    f" instance={instance_id}, "
                    f"status={one_result.get('DisplayStatus', '-')}, "
                    f"exit_code={one_result.get('ExitCode', '-')}"
                )
        except Exception as e:
            has_failure = True
            execution_results[instance_id] = {
                "InvocationStatus": "Failed",
                "ExitCode": "-",
                "Output": "",
                "Error": str(e),
            }
            print("[!] 执行失败:" f" instance={instance_id}, error={e}")

    out_dir = Path(args.output_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    raw = {
        "called_at": datetime.now().isoformat(),
        "region": region,
        "instance_ids": instance_ids,
        "install_command": install_command,
        "remote_command": remote_command,
        "install_command_expire_time": expire_time,
        "key_result": key_result,
        "run_result": run_result,
        "execution_results": execution_results,
    }
    json_path = out_dir / "security_guardrail_install.json"
    json_path.write_text(
        json.dumps(raw, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(f"[+] JSON → {json_path}")

    md_path = out_dir / "security_guardrail_install.md"
    md_path.write_text(
        format_markdown(
            key_result=key_result,
            install_command=install_command,
            remote_command=remote_command,
            expire_time=expire_time,
            instance_ids=instance_ids,
            region=region,
            run_result=run_result,
            execution_results=execution_results,
        ),
        encoding="utf-8",
    )
    print(f"[+] Markdown → {md_path}")

    if has_failure:
        raise SystemExit(1)


if __name__ == "__main__":
    main()