
Skill Manager
- 106 installs
- 543 repo stars
- Updated August 5, 2026
- cat-xierluo/legal-skills
Installs, version-tracks, and update-checks Claude Code, Codex, and OpenClaw skills from local paths or GitHub repos into the right agent config directory.
About
Manages installation, version tracking, and update checks for Claude Code, Codex, and OpenClaw skills and commands. Developers use it to install skills from local paths or GitHub, record source URL and version, and check GitHub for updates.
- Auto-detects .codex/.claude/.openclaw target directories
- Records install time, source URL, and version per skill
Skill Manager by the numbers
- 106 all-time installs (skills.sh)
- Ranked #258 of 782 Skill Development skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/cat-xierluo/legal-skills --skill skill-managerAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 106 |
|---|---|
| repo stars | ★ 543 |
| Last updated | August 5, 2026 |
| Repository | cat-xierluo/legal-skills ↗ |
What it does
Installs, version-tracks, and update-checks Claude Code, Codex, and OpenClaw skills from local paths or GitHub repos into the right agent config directory.
Files
Skill Manager
管理 Claude Code、Codex 和 OpenClaw Skills/Commands 的安装、同步、卸载和列表查看。
前置条件
- Git 已安装(用于 GitHub 克隆)
- 有写入目标 Agent 配置目录的权限,例如
.codex/skills/、.claude/skills/、.openclaw/skills/
安装行为
- 本地路径 (Skill) → 符号链接(保持与源同步)
- 本地路径 (Command) → 符号链接(保持与源同步)
- 本地集合目录 → 批量符号链接
- GitHub 仓库/子目录 → 克隆后删除 .git(静态复制)+ 自动安全检查
目标目录识别
执行安装、列表、卸载、更新时,脚本会从调用目录向上查找 Agent 配置目录:
- 在
/Users/maoking/.codex或其子目录调用时,目标为/Users/maoking/.codex/skills/ - 在项目根目录包含
.codex/、.claude/或.openclaw/时,目标为对应配置目录下的skills/或commands/ - 在
.codex/skills/、.claude/skills/、.openclaw/skills/内调用时,目标为其上级配置目录 - 如需显式指定目标根目录,可使用
--target参数或设置SKILL_MANAGER_TARGET_DIR=/path/to/.codex - 从全局配置目录(如
~/.claude)调用时,会尝试通过 git 自动发现项目本地目录,并打印告警
支持的来源类型
本地路径(符号链接)
# 单个 skill 目录
skill-manager install ~/skills/pdf-tool
# 单个 command 文件
skill-manager install ~/commands/deepresearch.md
# 包含多个 skills 的目录(批量安装)
skill-manager install ~/skills/external-skills/
# 包含多个 commands 的目录(批量安装)
skill-manager install ~/commands/GitHub 仓库根目录(克隆,删除 .git)
skill-manager install https://github.com/owner/skill-repo
skill-manager install owner/skill-repoGitHub 子目录(稀疏克隆,删除 .git)
# 完整 URL 到子目录
skill-manager install https://github.com/jgtolentino/insightpulse-odoo/tree/main/docs/claude-code-skills/community
# 简写格式:owner/repo/branch/path/to/skills-directory
skill-manager install jgtolentino/insightpulse-odoo/main/docs/claude-code-skills/community工作流程
安装
1. 检测来源类型 - 自动识别本地路径、GitHub 仓库或子目录 2. 检测 Item 类型 - 自动识别是 Skill(目录)还是 Command(.md 文件) 3. 检测是否为集合目录 - 检查目录是否包含多个 items 4. 批量处理模式 - 如果是集合目录,遍历所有 items 并分别安装 5. 本地来源 - 创建符号链接,保持与源同步更新 6. GitHub 仓库根 - 使用 git clone --depth 1 浅克隆 7. GitHub 子目录 - 使用稀疏克隆(sparse checkout)仅获取指定目录 8. 冲突处理 - 已存在时先备份为 .backup,然后安装新版本
安装命令
# 使用脚本安装
scripts/install.sh [--target <dir>] <source>
# 示例
scripts/install.sh ~/dev/my-skills/pdf-tool
scripts/install.sh ~/dev/my-commands/deepresearch.md
scripts/install.sh ~/dev/my-skills/
scripts/install.sh ~/dev/my-commands/
scripts/install.sh https://github.com/anthropics/claude-code
scripts/install.sh jgtolentino/insightpulse-odoo/main/docs/claude-code-skills/community
# 显式指定目标(从非项目目录调用时使用)
scripts/install.sh --target /path/to/project/.claude ~/dev/my-skills/pdf-tool列出已安装 Items
scripts/list.sh显示当前识别到的 Agent 配置目录下所有已安装的 items 及其类型(符号链接或克隆)。
卸载
scripts/remove.sh <name>删除指定的 skill 或 command(自动识别类型)。
更新
scripts/update.sh [name]- 不指定参数:更新所有通过 git 克隆的 skills
- 指定名称:更新指定的 skill
- 注意:符号链接的 items 会自动与源同步,无需手动更新
检查更新
scripts/check.sh检查所有远程安装 Skills 的更新状态,检测策略:
- 有版本号 → 直接比较本地与远程版本号
- 无版本号 → 检查远程仓库最近 Commits,与安装时间对比
- 子目录安装 → 精确检查 Skill 所在子目录的 Commits
显示:
- 📦 有可用更新的 Skills
- ✅ 已是最新版本的 Skills
- ⚠️ 检查失败的 Skills(无来源信息等)
每次安装和更新都会自动记录到 assets/skill-registry.json。
查看已安装记录
python3 scripts/record.py list显示所有已安装 Skills 的详细记录,包括:
- 安装时间
- 来源 URL
- 当前版本
- 描述信息
识别规则
Skill 目录规则
一个目录被视为有效的 skill 目录,如果它包含:
SKILL.md文件(标准 skill)- 或
skill.md文件(变体) - 或
.codex/.claude/.openclaw子目录
Command 文件规则
- 文件扩展名为
.md
集合目录规则
- Skills 集合:包含多个 skill 子目录
- Commands 集合:包含多个
.md文件
使用示例
# ========== 安装 ==========
# 安装本地单个 skill
skill-manager install ~/dev/my-skills/pdf-tool
# 批量安装本地目录下的所有 skills
skill-manager install ~/dev/my-skills/
skill-manager install ../other-project/.claude/skills/
# 在 Codex 全局目录中调用时,安装到 ~/.codex/skills/
cd /Users/maoking/.codex
skill-manager install ~/dev/my-skills/pdf-tool
# 从全局目录调用但安装到指定项目(使用 --target 避免装错位置)
skill-manager install --target /path/to/project/.claude ~/dev/my-skills/pdf-tool
# 从 GitHub 仓库根目录安装
skill-manager install https://github.com/anthropics/claude-code
skill-manager install anthropics/claude-code
# 从 GitHub 子目录安装
skill-manager install https://github.com/jgtolentino/insightpulse-odoo/tree/main/docs/claude-code-skills/community
skill-manager install jgtolentino/insightpulse-odoo/main/docs/claude-code-skills/community
# ========== 查看与管理 ==========
# 列出已安装的 skills
skill-manager list
# 卸载 skill
skill-manager remove pdf-tool
# ========== 更新与检查 ==========
# 检查所有 skills 的更新
skill-manager check
# 更新所有 git 克隆的 skills
skill-manager update
# 更新指定 skill
skill-manager update claude-code
# 查看安装记录
python3 scripts/record.py list安全检查
从 GitHub 安装 skill 时,会自动进行安全检查(本地安装不检查)。
检测内容
| 类别 | 说明 |
|---|---|
| 危险代码模式 | 命令执行、敏感文件访问、数据外泄、代码混淆、权限提升等 |
| Skill 特有风险 | 安装钩子、MCP 服务器配置等 |
| 提示词安全 | 提示注入、数据收集指令、执行指令、欺骗性描述等 |
| 硬编码凭证 | API Key、Token、密码等敏感信息 |
风险等级
- 🔴 CRITICAL - 极高风险,强烈建议不要使用
- 🟠 HIGH - 高风险,需审计后使用
- 🟡 MEDIUM - 中等风险,使用前请检查
- 🟢 LOW - 低风险,建议定期检查
- ✅ NONE - 未发现明显风险
注意事项
- 安全检查需要 Python 3 环境,无 Python 时静默跳过
- 检查发现问题不会阻止安装,仅输出警告报告
- 建议在安装外部 skill 后仔细阅读安全报告
注册表 Schema
每个已安装的 Skill 记录在 assets/skill-registry.json 中,包含以下字段:
| 字段 | 说明 |
|---|---|
name | Skill 目录名 |
source | 原始安装来源(本地路径或 GitHub URL) |
install_type | "local"(符号链接)或 "remote"(GitHub 克隆) |
installed_at | 初始安装时间(ISO 8601) |
last_updated | 最后版本更新时间 |
last_check_at | 最后一次更新检查时间(仅远程) |
installed_version | 安装时的版本号 |
current_version | 当前已安装版本 |
latest_version | 远程最新版本 |
install_commit | 安装时的 Git commit hash(仅远程) |
install_branch | 安装时使用的 Git branch(仅远程) |
remote_url | 完整 GitHub URL,含子目录路径(仅远程) |
remote_subpath | Skill 在仓库中的子路径(仅子目录安装) |
description | Skill 描述 |
homepage | 主页 URL |
目录结构
skill-manager/
├── SKILL.md # 本文件
├── CHANGELOG.md # 变更日志
├── CLAUDE.md # AI 开发助手说明
├── LICENSE.txt # 许可证
├── scripts/
│ ├── install.sh # 安装脚本
│ ├── list.sh # 列表脚本
│ ├── remove.sh # 卸载脚本
│ ├── update.sh # 更新脚本
│ ├── check.sh # 更新检查脚本
│ ├── auto-check.sh # 定期自动检查触发器
│ ├── target.sh # Agent 配置目录识别模块
│ ├── record.py # 记录管理模块
│ └── security.py # 安全检查模块
└── assets/ # 资源文件
├── skill-registry.json # Skill 安装记录(运行时生成)
└── skill-registry.example.json # 注册表示例{
"_note": "Skill Manager 注册表示例 - 复制此文件为 skill-registry.json 使用",
"example-local-skill": {
"name": "example-local-skill",
"source": "/path/to/local/skills/example-local-skill",
"install_type": "local",
"installed_at": "2026-04-08T10:00:00+08:00",
"last_updated": "2026-04-08T10:00:00+08:00",
"last_check_at": null,
"installed_version": "1.0.0",
"current_version": "1.0.0",
"latest_version": null,
"install_commit": null,
"install_branch": null,
"remote_url": null,
"description": "示例本地安装的 Skill"
},
"example-remote-skill": {
"name": "example-remote-skill",
"source": "https://github.com/owner/skill-repo",
"install_type": "remote",
"installed_at": "2026-04-08T10:00:00+08:00",
"last_updated": "2026-04-08T10:00:00+08:00",
"last_check_at": "2026-04-15T10:00:00+08:00",
"installed_version": "1.0.0",
"current_version": "1.0.0",
"latest_version": "1.2.0",
"install_commit": "a1b2c3d",
"install_branch": "main",
"remote_url": "https://github.com/owner/skill-repo",
"description": "示例 GitHub 安装的 Skill"
}
}
变更日志
[1.5.0] - 2026-05-17
新增
- 远程安装元数据追踪:注册表新增
install_type、install_commit、install_branch、remote_url、remote_subpath字段,记录 GitHub 安装的完整上下文 - 智能更新检测:双策略检测远程 Skill 更新 — 有版本号时比较版本,无版本号时检查 Commits 与安装时间对比
- 子目录精确追踪:GitHub 子目录安装时记录完整子路径,commit 检查精确到 Skill 所在子目录
- 可配置检查间隔:通过
SKILL_MANAGER_CHECK_THRESHOLD环境变量自定义自动检查间隔(默认 7 天) - 向后兼容迁移:旧注册表条目自动填充新字段默认值
修复
- GitHub 子目录安装缺少记录:修复从 GitHub 子目录安装时未调用
record_install的问题 - GitHub Commits 抓取:修复
get_github_commits()因 GitHub 页面结构变更导致无法获取 commit 的问题
改进
install.sh在删除.git前捕获 commit hash 和 branchcheck-all跳过本地安装的 Skill(符号链接自动同步,无需远程检查),仅检查远程安装的 Skillrecord.py list显示安装类型、commit、branch 等新增字段
[1.4.0] - 2026-05-17
新增
- Codex 目标目录支持:在
/Users/maoking/.codex或其子目录调用时,自动将 Skill/Command 安装到.codex/skills/或.codex/commands/ - 项目级 Codex 配置识别:当调用目录或上级目录包含
.codex/、.claude/、.openclaw/时,自动选择对应 Agent 配置目录 - 目标目录覆盖变量:支持通过
SKILL_MANAGER_TARGET_DIR显式指定目标 Agent 配置根目录
改进
- 抽取
scripts/target.sh作为安装、列表、卸载、更新脚本共用的目标目录识别模块 list.sh、remove.sh、update.sh改为从调用目录识别目标,行为与install.sh保持一致- Skill 目录识别规则扩展到
.codex和.openclaw子目录 - 修正文档中的注册表路径说明为
assets/skill-registry.json
修复
- 修复安装记录时间戳使用 UTC 时间拼接本地时区偏移,导致显示时间偏早的问题
[1.3.0] - 2026-04-08
新增
- 版本追踪:自动记录每个 Skill 的安装时间、来源 URL 和版本号
- 更新检查:新增
scripts/check.sh脚本,检查所有已安装 Skills 的最新版本 - 更新提示:显示有更新的 Skills 列表,包含版本对比和变更摘要
- 持久化存储:安装记录保存在
assets/skill-registry.json - Changelog 获取:自动从 GitHub 获取远程 CHANGELOG.md 内容
- Commits 获取:当没有 CHANGELOG 时,自动获取最近的 commits 作为更新内容参考
- 子目录支持:支持 GitHub 仓库子目录中的 Skill 安装
改进
- 安装时自动调用记录模块,无需手动操作
- 更新时自动记录版本变更历史
- 无需 GitHub API,直接读取网页获取版本和变更信息
- 检查更新时区分三类状态:有更新 / 已是最新 / 检查失败
技术细节
- 新增
scripts/record.py核心记录管理模块 - 新增
scripts/check.sh更新检查脚本 - 新增
assets/目录存储注册表数据 - 修改
scripts/install.sh安装后自动记录 - 修改
scripts/update.sh更新后自动记录
[1.2.0] - 2026-02-21
新增
- 安全检查:从 GitHub 安装 skill 时自动进行安全检测
- 检测危险代码模式(命令执行、敏感文件访问、数据外泄等)
- 检测 Skill 特有风险(安装钩子、MCP 服务器等)
- 检测提示词安全风险(提示注入、数据收集指令等)
- 检测硬编码凭证(API Key、Token 等)
- 生成 Markdown 格式的安全报告
改进
- 仅 GitHub 安装触发安全检查,本地安装不检查
- 无 Python 环境时静默跳过安全检查
- 安全检查发现问题不会阻止安装,仅输出警告
技术细节
- 新增
scripts/security.py安全分析模块 - 修改
scripts/install.sh在 GitHub 安装后调用安全检查
[1.1.1] - 2026-02-09
修复
- 项目目录检测:修复通过 Skill 工具调用时,错误将符号链接创建到全局
~/.claude/skills/而非项目本地.claude/skills/的问题 - 原始目录保存:在脚本开头保存
ORIGINAL_PWD,确保find_claude_dir()从调用者的原始工作目录开始查找.claude
[1.1.0] - 2026-01-21
新增
- Command 支持:现在可以管理
.claude/commands/目录下的命令文件 - 统一管理:所有脚本(install、list、remove、update)同时支持 Skills 和 Commands
- 自动类型检测:根据文件扩展名(.md)自动识别 Command,根据目录结构识别 Skill
- 批量安装 Commands:支持批量安装 commands 目录下的所有 .md 文件
改进
- 路径解析优化:新增
find_claude_dir()函数,通过向上查找.claude目录,支持符号链接结构 - 更清晰的输出:list.sh 分别显示 Skills 和 Commands,便于查看
技术细节
- Skill 识别规则:包含
SKILL.md/skill.md/.claude/的目录 - Command 识别规则:
.md文件 - 集合目录规则:
- Skills 集合:包含多个 skill 子目录
- Commands 集合:包含多个
.md文件
[1.0.0] - 2026-01-21
新增
- 初始版本发布
- 支持从本地路径安装单个 skill(符号链接)
- 支持批量安装本地 skills 集合目录(符号链接)
- 支持从 GitHub 仓库根目录克隆 skill
- 支持从 GitHub 子目录稀疏克隆 skill
- 支持列出已安装的 skills
- 支持卸载 skills
- 支持更新 Git 克隆的 skills
- 自动识别 skill 目录(SKILL.md、skill.md 或 .claude 目录)
MIT License
Copyright (c) 2025 杨卫薪律师(微信ywxlaw)
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
#!/bin/bash
# Skill Manager - Auto Update Check
# 检查超过阈值的远程 Skill 更新状态
# 仅在发现更新时输出,否则静默退出
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if ! command -v python3 &> /dev/null; then
exit 0
fi
RECORD_SCRIPT="$SCRIPT_DIR/record.py"
if [ ! -f "$RECORD_SCRIPT" ]; then
exit 0
fi
THRESHOLD="${SKILL_MANAGER_CHECK_THRESHOLD:-7}"
python3 "$RECORD_SCRIPT" auto-check --threshold "$THRESHOLD" 2>/dev/null
#!/bin/bash
# Skill Manager - Check Updates Script
# 检查所有已安装 Skills 的更新状态
# 获取脚本所在目录
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MANAGER_DIR="$(dirname "$SCRIPT_DIR")"
# 检查 Python 是否可用
if ! command -v python3 &> /dev/null; then
echo "❌ 需要 Python 3 来运行更新检查"
exit 1
fi
RECORD_SCRIPT="$SCRIPT_DIR/record.py"
if [ ! -f "$RECORD_SCRIPT" ]; then
echo "❌ 找不到记录模块: $RECORD_SCRIPT"
exit 1
fi
# 执行检查
echo "🔍 正在检查 Skill 更新..."
echo ""
python3 "$RECORD_SCRIPT" check-all
#!/bin/bash
# Skill & Command Manager - Install Script
# 安装或同步外部 skills/commands 到本地 Agent 配置目录
# 自动检测所有 Agent 目录 (.codex/.claude/.openclaw) 并批量安装
set -e
# Parse arguments: support --target flag for explicit target directory
SOURCE=""
TARGET_OVERRIDE=""
while [ $# -gt 0 ]; do
case "$1" in
--target|-t)
if [ -z "${2:-}" ]; then
echo "❌ 错误: --target 需要指定目录路径"
exit 1
fi
TARGET_OVERRIDE="$2"
shift 2
;;
-h|--help)
echo "用法: $0 [选项] <source>"
echo ""
echo "选项:"
echo " --target, -t <dir> 指定目标 Agent 配置目录"
echo ""
echo "示例:"
echo " $0 ~/skills/pdf-tool"
echo " $0 --target /path/to/project/.claude ~/skills/pdf-tool"
exit 0
;;
*)
SOURCE="$1"
shift
;;
esac
done
# 保存调用者的原始工作目录(关键:用于定位项目 Agent 配置目录)
ORIGINAL_PWD="$PWD"
# Apply --target override
if [ -n "$TARGET_OVERRIDE" ]; then
export SKILL_MANAGER_TARGET_DIR="$TARGET_OVERRIDE"
fi
# 获取脚本所在目录
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MANAGER_DIR="$(dirname "$SCRIPT_DIR")"
TARGET_HELPER="$SCRIPT_DIR/target.sh"
if [ -f "$TARGET_HELPER" ]; then
# shellcheck source=target.sh
source "$TARGET_HELPER"
else
echo "❌ 错误: 找不到目标目录识别模块: $TARGET_HELPER"
exit 1
fi
# 安装记录函数 - 记录 Skill 安装到注册表
record_install() {
local skill_name="$1"
local source_url="$2"
local target_path="$3"
# 调用 Python 记录模块
if command -v python3 &> /dev/null; then
RECORD_SCRIPT="$SCRIPT_DIR/record.py"
if [ -f "$RECORD_SCRIPT" ]; then
python3 "$RECORD_SCRIPT" install "$skill_name" "$source_url" --path "$target_path" --install-type local 2>/dev/null || true
fi
fi
}
# 检测源类型(skill 或 command)
detect_source_type() {
local src="$1"
# 如果是文件
if [ -f "$src" ]; then
if [[ "$src" =~ \.md$ ]]; then
echo "command"
else
echo "unknown"
fi
# 如果是目录
elif [ -d "$src" ]; then
# 优先检查是否为 skill(包含 SKILL.md 或 Agent 配置目录等)
if [ -f "$src/SKILL.md" ] || [ -f "$src/skill.md" ] || [ -d "$src/.codex" ] || [ -d "$src/.claude" ] || [ -d "$src/.openclaw" ] || [ -d "$src/.agents" ] || [ -d "$src/.agent" ]; then
echo "skill"
# 检查是否为 command 集合目录(包含多个 .md 文件,但不包含 SKILL.md)
else
local md_count=$(find "$src" -maxdepth 1 -name "*.md" -type f 2>/dev/null | wc -l | tr -d ' ')
if [ "$md_count" -gt 0 ]; then
echo "command-collection"
else
echo "unknown"
fi
fi
else
echo "unknown"
fi
}
# 检查是否为 skills 集合目录
is_skills_collection() {
local dir="$1"
local found_skills=0
for item in "$dir"/*; do
if [ -d "$item" ]; then
if [ -f "$item/SKILL.md" ] || [ -f "$item/skill.md" ] || [ -d "$item/.codex" ] || [ -d "$item/.claude" ] || [ -d "$item/.openclaw" ]; then
((found_skills++))
fi
fi
done
[ "$found_skills" -gt 1 ]
}
# 检查是否为 commands 集合目录
# 注意:必须排除包含 SKILL.md 的 skill 目录
is_commands_collection() {
local dir="$1"
local found_commands=0
# 如果目录包含 SKILL.md,则不是 commands 集合
if [ -f "$dir/SKILL.md" ] || [ -f "$dir/skill.md" ] || [ -d "$dir/.codex" ] || [ -d "$dir/.claude" ] || [ -d "$dir/.openclaw" ]; then
return 1
fi
for item in "$dir"/*; do
if [ -f "$item" ] && [[ "$item" =~ \.md$ ]]; then
# 排除 SKILL.md/skill.md 文件
local basename=$(basename "$item")
if [ "$basename" != "SKILL.md" ] && [ "$basename" != "skill.md" ]; then
((found_commands++))
fi
fi
done
[ "$found_commands" -gt 1 ]
}
# 检查参数
if [ -z "$SOURCE" ]; then
echo "❌ 错误: 请提供源路径或 URL"
echo ""
echo "使用方法:"
echo " $0 <本地路径 | github-url | owner/repo>"
echo ""
echo "示例:"
echo " 本地单个 skill/command: $0 ~/my-skills/pdf-tool"
echo " 本地 skills 集合: $0 ~/skills/"
echo " 本地 commands 集合: $0 ~/commands/"
echo " GitHub 仓库: $0 owner/repo"
echo " GitHub 子目录: $0 owner/repo/branch/path/to/skills"
exit 1
fi
# ---- 检测来源类型 ----
if [[ "$SOURCE" =~ ^https?://github\.com/([^/]+)/([^/]+)/blob/([^/]+)/(.+)$ ]]; then
# GitHub URL 到子目录 (blob 格式)
OWNER="${BASH_REMATCH[1]}"
REPO="${BASH_REMATCH[2]}"
BRANCH="${BASH_REMATCH[3]}"
SUBPATH="${BASH_REMATCH[4]}"
SOURCE_TYPE="github-subdir"
CLONE_URL="https://github.com/$OWNER/$REPO"
elif [[ "$SOURCE" =~ ^https?://github\.com/([^/]+)/([^/]+)/tree/([^/]+)/(.+)$ ]]; then
# GitHub URL 到子目录 (tree 格式)
OWNER="${BASH_REMATCH[1]}"
REPO="${BASH_REMATCH[2]}"
BRANCH="${BASH_REMATCH[3]}"
SUBPATH="${BASH_REMATCH[4]}"
SOURCE_TYPE="github-subdir"
CLONE_URL="https://github.com/$OWNER/$REPO"
elif [[ "$SOURCE" =~ ^https?://github\.com/([^/]+)/([^/]+)(\.git)?/?$ ]]; then
# GitHub 仓库根目录
OWNER="${BASH_REMATCH[1]}"
REPO="${BASH_REMATCH[2]}"
SOURCE_TYPE="github"
CLONE_URL="https://github.com/$OWNER/$REPO"
elif [[ "$SOURCE" =~ ^([^/]+)/([^/]+)(/(.+))?$ ]]; then
# 可能是 GitHub 简写格式,需要进一步检查
# 如果路径不存在,则认为是 GitHub 格式
if [ ! -e "$SOURCE" ]; then
OWNER="${BASH_REMATCH[1]}"
REPO="${BASH_REMATCH[2]}"
if [ -n "${BASH_REMATCH[4]}" ]; then
SUBPATH="${BASH_REMATCH[4]}"
SOURCE_TYPE="github-subdir"
CLONE_URL="https://github.com/$OWNER/$REPO"
else
SOURCE_TYPE="github"
CLONE_URL="https://github.com/$OWNER/$REPO"
fi
else
SOURCE_TYPE="local"
fi
else
# 本地路径
SOURCE_TYPE="local"
fi
# ---- 检测所有 Agent 配置目录 ----
FALLBACK_AGENT_DIR="$(find_agent_config_dir "$ORIGINAL_PWD" "$(find_agent_config_dir "$MANAGER_DIR" "$PWD/.claude")")"
ALL_AGENT_DIRS=()
while IFS= read -r dir; do
ALL_AGENT_DIRS+=("$dir")
done < <(find_all_agent_config_dirs "$ORIGINAL_PWD" "$FALLBACK_AGENT_DIR")
if [ ${#ALL_AGENT_DIRS[@]} -eq 0 ]; then
echo "❌ 错误: 未找到任何 Agent 配置目录 (.codex/.claude/.openclaw)"
exit 1
fi
# Safety: if target would be a global config root, try git-based project rescue
_is_global_config_root() {
local dir="$1"
local _home="${HOME:-/Users/${USER}}"
case "$dir" in
"$_home/.codex"| "$_home/.claude"| "$_home/.openclaw"| "$_home/.agents"| "$_home/.agent") return 0 ;;
esac
return 1
}
if [ -z "${SKILL_MANAGER_TARGET_DIR:-}" ] && [ ${#ALL_AGENT_DIRS[@]} -gt 0 ]; then
_only_global=true
for _dir in "${ALL_AGENT_DIRS[@]}"; do
if ! _is_global_config_root "$_dir"; then
_only_global=false
break
fi
done
if [ "$_only_global" = true ]; then
# Try git-based project rescue
_git_root="$(git rev-parse --show-toplevel 2>/dev/null || true)"
_rescued=false
if [ -n "$_git_root" ] && ! _is_global_config_root "$_git_root" && [ "$_git_root" != "$HOME" ]; then
for _cfg in .codex .claude .openclaw .agents .agent; do
if [ -d "$_git_root/$_cfg" ]; then
echo "🔄 检测到从全局配置目录调用,已通过 git 发现项目: $_git_root"
echo ""
ALL_AGENT_DIRS=("$_git_root/$_cfg")
_rescued=true
break
fi
done
fi
if [ "$_rescued" = false ]; then
echo "⚠️ 目标为全局配置目录 (${ALL_AGENT_DIRS[*]})"
echo " 如需安装到项目目录,请使用: $0 --target <project/.claude> <source>"
echo ""
fi
fi
fi
# 显示检测到的目录
AGENT_NAMES=()
for dir in "${ALL_AGENT_DIRS[@]}"; do
AGENT_NAMES+=($(basename "$dir"))
done
echo "🔍 检测到 ${#ALL_AGENT_DIRS[@]} 个 Agent 配置目录: ${AGENT_NAMES[*]}"
echo ""
# ---- GitHub 预处理:克隆到临时目录(只克隆一次)----
TEMP_CLONE_DIR=""
INSTALL_COMMIT=""
INSTALL_BRANCH=""
GH_SKILL_NAME=""
if [ "$SOURCE_TYPE" = "github-subdir" ] || [ "$SOURCE_TYPE" = "github" ]; then
if [ "$SOURCE_TYPE" = "github-subdir" ]; then
GH_SKILL_NAME=$(basename "$SUBPATH")
else
GH_SKILL_NAME="$REPO"
fi
TEMP_CLONE_DIR=$(mktemp -d)
if [ "$SOURCE_TYPE" = "github-subdir" ]; then
echo "📦 正在从 GitHub 获取子目录..."
echo " 仓库: $CLONE_URL"
echo " 路径: $SUBPATH"
cd "$TEMP_CLONE_DIR"
git init -q
git remote add origin "$CLONE_URL"
git config core.sparseCheckout true
echo "$SUBPATH" > .git/info/sparse-checkout
git fetch --depth 1 origin "${BRANCH:-main}" -q 2>/dev/null || {
echo "❌ 错误: 无法从 GitHub 获取"
cd - > /dev/null
rm -rf "$TEMP_CLONE_DIR"
exit 1
}
git checkout "${BRANCH:-main}" -q
INSTALL_COMMIT=$(git rev-parse --short HEAD 2>/dev/null || echo "")
INSTALL_BRANCH="${BRANCH:-main}"
cd - > /dev/null
# 移动到临时目录内的固定位置(仅当源和目标不同时,避免单段路径 mv 到自身内部)
if [ "$TEMP_CLONE_DIR/$SUBPATH" != "$TEMP_CLONE_DIR/$GH_SKILL_NAME" ]; then
mv "$TEMP_CLONE_DIR/$SUBPATH" "$TEMP_CLONE_DIR/$GH_SKILL_NAME"
fi
else
echo "📦 正在从 GitHub 克隆..."
echo " 仓库: $CLONE_URL"
git clone --depth 1 -q "$CLONE_URL" "$TEMP_CLONE_DIR/$GH_SKILL_NAME" 2>/dev/null || {
echo "❌ 错误: 无法从 GitHub 克隆"
rm -rf "$TEMP_CLONE_DIR"
exit 1
}
INSTALL_COMMIT=$(cd "$TEMP_CLONE_DIR/$GH_SKILL_NAME" && git rev-parse --short HEAD 2>/dev/null || echo "")
INSTALL_BRANCH=$(cd "$TEMP_CLONE_DIR/$GH_SKILL_NAME" && git branch --show-current 2>/dev/null || echo "main")
# 删除 .git 目录
rm -rf "$TEMP_CLONE_DIR/$GH_SKILL_NAME/.git"
fi
echo "✓ GitHub 源准备完成"
echo ""
fi
# ---- 安装到每个 Agent 目录 ----
install_count=0
success_dirs=()
fail_count=0
for AGENT_DIR in "${ALL_AGENT_DIRS[@]}"; do
agent_name=$(basename "$AGENT_DIR")
SKILLS_DIR="$AGENT_DIR/skills"
COMMANDS_DIR="$AGENT_DIR/commands"
echo "───────────────────────────────────"
echo "📦 安装到 $agent_name"
echo "───────────────────────────────────"
# ---- 本地来源 ----
if [ "$SOURCE_TYPE" = "local" ]; then
DETECTED_TYPE=$(detect_source_type "$SOURCE")
if [ "$DETECTED_TYPE" = "unknown" ]; then
if [ ! -e "$SOURCE" ]; then
echo "❌ 错误: 找不到源: $SOURCE"
else
echo "❌ 错误: 无法识别源类型,请确保是 skill 目录或 command .md 文件"
fi
((fail_count++)) || true
echo ""
continue
fi
# 处理单个 command 文件
if [ "$DETECTED_TYPE" = "command" ]; then
COMMAND_NAME=$(basename "$SOURCE" .md)
TARGET_PATH="$COMMANDS_DIR/$COMMAND_NAME.md"
mkdir -p "$COMMANDS_DIR"
if [ -L "$TARGET_PATH" ]; then
echo "⚠ 发现现有符号链接,正在移除..."
rm "$TARGET_PATH"
elif [ -f "$TARGET_PATH" ]; then
if [ "$TARGET_PATH" -ef "$SOURCE" ]; then
echo "✓ 已指向相同文件"
((install_count++)) || true
success_dirs+=("$agent_name")
echo ""
continue
fi
echo "⚠ 目标已存在,正在备份到 ${TARGET_PATH}.backup..."
mv "$TARGET_PATH" "${TARGET_PATH}.backup"
fi
echo "🔗 正在创建 command 符号链接..."
ln -s "$SOURCE" "$TARGET_PATH"
echo "✓ 已链接 command: $TARGET_PATH -> $SOURCE"
((install_count++)) || true
success_dirs+=("$agent_name")
echo ""
continue
fi
# 处理目录
if [ ! -d "$SOURCE" ]; then
echo "❌ 错误: 找不到源目录: $SOURCE"
((fail_count++)) || true
echo ""
continue
fi
# 检查是否为 skills 集合目录
if is_skills_collection "$SOURCE"; then
local_count=0
for skill_dir in "$SOURCE"/*; do
if [ -d "$skill_dir" ]; then
skill_name=$(basename "$skill_dir")
if [ -f "$skill_dir/SKILL.md" ] || [ -f "$skill_dir/skill.md" ] || [ -d "$skill_dir/.codex" ] || [ -d "$skill_dir/.claude" ] || [ -d "$skill_dir/.openclaw" ]; then
target_path="$SKILLS_DIR/$skill_name"
mkdir -p "$SKILLS_DIR"
if [ -L "$target_path" ]; then
rm "$target_path"
elif [ -d "$target_path" ]; then
if [ "$target_path" -ef "$skill_dir" ]; then
continue
fi
rm -rf "${target_path}.backup"
mv "$target_path" "${target_path}.backup"
fi
ln -s "$skill_dir" "$target_path"
echo " ✓ 已链接: $skill_name"
((local_count++)) || true
fi
fi
done
echo "✓ 已安装 $local_count 个 skills"
((install_count++)) || true
success_dirs+=("$agent_name")
echo ""
continue
fi
# 检查是否为 commands 集合目录
if is_commands_collection "$SOURCE"; then
local_count=0
for cmd_file in "$SOURCE"/*.md; do
if [ -f "$cmd_file" ]; then
cmd_name=$(basename "$cmd_file" .md)
target_path="$COMMANDS_DIR/$cmd_name.md"
mkdir -p "$COMMANDS_DIR"
if [ -L "$target_path" ]; then
rm "$target_path"
elif [ -f "$target_path" ]; then
if [ "$target_path" -ef "$cmd_file" ]; then
continue
fi
mv "$target_path" "${target_path}.backup"
fi
ln -s "$cmd_file" "$target_path"
echo " ✓ 已链接: $cmd_name"
((local_count++)) || true
fi
done
echo "✓ 已安装 $local_count 个 commands"
((install_count++)) || true
success_dirs+=("$agent_name")
echo ""
continue
fi
# 单个本地 skill - 使用符号链接
if [ "$DETECTED_TYPE" = "skill" ]; then
SKILL_NAME=$(basename "$SOURCE")
TARGET_PATH="$SKILLS_DIR/$SKILL_NAME"
mkdir -p "$SKILLS_DIR"
if [ -L "$TARGET_PATH" ]; then
echo "⚠ 发现现有符号链接,正在移除..."
rm "$TARGET_PATH"
elif [ -d "$TARGET_PATH" ]; then
if [ "$TARGET_PATH" -ef "$SOURCE" ]; then
echo "✓ 已指向相同目录"
((install_count++)) || true
success_dirs+=("$agent_name")
echo ""
continue
fi
echo "⚠ 目标已存在,正在备份到 ${TARGET_PATH}.backup..."
rm -rf "${TARGET_PATH}.backup"
mv "$TARGET_PATH" "${TARGET_PATH}.backup"
fi
echo "🔗 正在创建 skill 符号链接..."
ln -s "$SOURCE" "$TARGET_PATH"
echo "✓ 已链接 skill: $TARGET_PATH -> $SOURCE"
# 记录安装
record_install "$SKILL_NAME" "$SOURCE" "$TARGET_PATH"
((install_count++)) || true
success_dirs+=("$agent_name")
echo ""
continue
fi
fi
# ---- GitHub 来源 ----
if [ "$SOURCE_TYPE" = "github-subdir" ] || [ "$SOURCE_TYPE" = "github" ]; then
TARGET_PATH="$SKILLS_DIR/$GH_SKILL_NAME"
mkdir -p "$SKILLS_DIR"
if [ -e "$TARGET_PATH" ]; then
echo "⚠ 目标已存在,正在备份到 ${TARGET_PATH}.backup..."
rm -rf "${TARGET_PATH}.backup"
mv "$TARGET_PATH" "${TARGET_PATH}.backup"
fi
cp -R "$TEMP_CLONE_DIR/$GH_SKILL_NAME" "$TARGET_PATH"
echo "✓ 已安装: $TARGET_PATH"
# 记录安装
if command -v python3 &> /dev/null; then
RECORD_SCRIPT="$SCRIPT_DIR/record.py"
if [ -f "$RECORD_SCRIPT" ]; then
local_args=()
if [ "$SOURCE_TYPE" = "github-subdir" ]; then
local_args=(--remote-url "$CLONE_URL/tree/${INSTALL_BRANCH}/$SUBPATH" --remote-subpath "$SUBPATH")
else
local_args=(--remote-url "$CLONE_URL")
fi
python3 "$RECORD_SCRIPT" install "$GH_SKILL_NAME" "$CLONE_URL" \
--path "$TARGET_PATH" \
--install-type remote \
--install-commit "$INSTALL_COMMIT" \
--install-branch "$INSTALL_BRANCH" \
"${local_args[@]}" 2>/dev/null || true
fi
fi
((install_count++)) || true
success_dirs+=("$agent_name")
echo ""
fi
done
# ---- 清理临时目录 ----
if [ -n "$TEMP_CLONE_DIR" ] && [ -d "$TEMP_CLONE_DIR" ]; then
rm -rf "$TEMP_CLONE_DIR"
fi
# ---- 安全检查(GitHub 来源,只检查一次)----
if [ "$SOURCE_TYPE" = "github" ] || [ "$SOURCE_TYPE" = "github-subdir" ]; then
if [ ${#success_dirs[@]} -gt 0 ]; then
FIRST_AGENT_DIR="${ALL_AGENT_DIRS[0]}"
CHECK_PATH="$FIRST_AGENT_DIR/skills/$GH_SKILL_NAME"
if command -v python3 &> /dev/null && [ -d "$CHECK_PATH" ]; then
SECURITY_SCRIPT="$SCRIPT_DIR/security.py"
if [ -f "$SECURITY_SCRIPT" ]; then
echo "🔍 正在进行安全检查..."
python3 "$SECURITY_SCRIPT" "$CHECK_PATH" 2>/dev/null || {
echo ""
echo "⚠️ 安全检查发现问题,请查看上方报告"
echo " 如需继续使用,请自行评估风险"
}
fi
else
echo "💡 提示: 未检测到 Python,跳过安全检查"
fi
fi
fi
# ---- 汇总 ----
echo ""
echo "========================================"
echo "📊 安装汇总"
echo "========================================"
if [ "$install_count" -gt 0 ]; then
echo " 成功: $install_count 个 Agent 目录"
for d in "${success_dirs[@]}"; do
echo " ✅ $d"
done
fi
if [ "$fail_count" -gt 0 ]; then
echo " 失败: $fail_count"
fi
echo "========================================"
#!/bin/bash
# Skill & Command Manager - List Script
# 列出已安装的 skills 和 commands
ORIGINAL_PWD="$PWD"
# 获取脚本所在目录
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MANAGER_DIR="$(dirname "$SCRIPT_DIR")"
TARGET_HELPER="$SCRIPT_DIR/target.sh"
if [ -f "$TARGET_HELPER" ]; then
# shellcheck source=target.sh
source "$TARGET_HELPER"
else
echo "❌ 错误: 找不到目标目录识别模块: $TARGET_HELPER"
exit 1
fi
SCRIPT_AGENT_DIR="$(find_agent_config_dir "$MANAGER_DIR" "$PWD/.claude")"
AGENT_DIR="$(find_agent_config_dir "$ORIGINAL_PWD" "$SCRIPT_AGENT_DIR")"
SKILLS_DIR="$AGENT_DIR/skills"
COMMANDS_DIR="$AGENT_DIR/commands"
# 列出 skills
list_skills() {
local dir="$1"
if [ ! -d "$dir" ]; then
return
fi
echo "📋 已安装的 Skills"
echo ""
count=0
for item in "$dir"/*; do
if [ -e "$item" ] && [ "$(basename "$item")" != "skill-manager" ]; then
name=$(basename "$item")
if [ -L "$item" ]; then
# 符号链接
target=$(readlink "$item")
echo "🔗 $name"
echo " 类型: 符号链接"
echo " 指向: $target"
elif [ -d "$item" ]; then
# 目录
if [ -d "$item/.git" ]; then
# Git 仓库
remote=$(cd "$item" && git remote get-url origin 2>/dev/null || echo "未知")
branch=$(cd "$item" && git branch --show-current 2>/dev/null || echo "未知")
echo "📦 $name"
echo " 类型: Git 克隆"
echo " 仓库: $remote"
echo " 分支: $branch"
else
# 普通目录
echo "📁 $name"
echo " 类型: 本地目录"
fi
fi
echo ""
((count++))
fi
done
if [ "$count" -eq 0 ]; then
echo "暂无已安装的 skills"
else
echo "总计: $count 个 skills"
fi
}
# 列出 commands
list_commands() {
local dir="$1"
if [ ! -d "$dir" ]; then
return
fi
echo ""
echo "📋 已安装的 Commands"
echo ""
count=0
for item in "$dir"/*.md; do
if [ -e "$item" ]; then
name=$(basename "$item" .md)
if [ -L "$item" ]; then
# 符号链接
target=$(readlink "$item")
echo "🔗 $name"
echo " 类型: 符号链接"
echo " 指向: $target"
elif [ -f "$item" ]; then
# 普通文件
echo "📄 $name"
echo " 类型: 本地文件"
fi
echo ""
((count++))
fi
done
if [ "$count" -eq 0 ]; then
echo "暂无已安装的 commands"
else
echo "总计: $count 个 commands"
fi
}
# 执行列表
list_skills "$SKILLS_DIR"
list_commands "$COMMANDS_DIR"
#!/usr/bin/env python3
"""
Skill Manager - Record Management Module
管理 Skill 安装记录、版本追踪和更新检查
"""
import json
import os
import sys
import re
import subprocess
import hashlib
from datetime import datetime
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
# 获取脚本所在目录
SCRIPT_DIR = Path(__file__).parent.resolve()
MANAGER_DIR = SCRIPT_DIR.parent
ASSETS_DIR = MANAGER_DIR / "assets"
REGISTRY_FILE = ASSETS_DIR / "skill-registry.json"
EXAMPLE_FILE = ASSETS_DIR / "skill-registry.example.json"
# 确保 assets 目录存在
ASSETS_DIR.mkdir(exist_ok=True)
def get_current_timestamp():
"""获取当前时间戳 (ISO 8601 格式)"""
return datetime.now().astimezone().isoformat(timespec="seconds")
def _migrate_entry(name: str, entry: dict) -> dict:
"""为旧条目填充缺失的新字段"""
source = entry.get("source", "")
if "install_type" not in entry:
if "github.com" in source or ("/" in source and not source.startswith("/") and not source.startswith("~")):
entry["install_type"] = "remote"
else:
entry["install_type"] = "local"
if "install_commit" not in entry:
entry["install_commit"] = None
if "install_branch" not in entry:
entry["install_branch"] = None
if "remote_url" not in entry:
entry["remote_url"] = None
if "remote_subpath" not in entry:
entry["remote_subpath"] = None
if "last_check_at" not in entry:
entry["last_check_at"] = None
return entry
def load_registry() -> dict:
"""加载 Skill 注册表(自动迁移旧条目)"""
if REGISTRY_FILE.exists():
try:
with open(REGISTRY_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
registry = {k: _migrate_entry(k, v) for k, v in data.items() if not k.startswith('_')}
return registry
except (json.JSONDecodeError, IOError):
pass
# 如果注册表不存在,尝试从示例文件初始化
if EXAMPLE_FILE.exists():
try:
with open(EXAMPLE_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
registry = {k: _migrate_entry(k, v) for k, v in data.items() if not k.startswith('_')}
if registry:
save_registry(registry)
return registry
except (json.JSONDecodeError, IOError):
pass
return {}
def save_registry(registry: dict):
"""保存 Skill 注册表"""
with open(REGISTRY_FILE, 'w', encoding='utf-8') as f:
json.dump(registry, f, indent=2, ensure_ascii=False)
def get_skill_version(skill_path: Path) -> Optional[str]:
"""从 SKILL.md 读取版本号"""
skill_md = skill_path / "SKILL.md"
if not skill_md.exists():
skill_md = skill_path / "skill.md"
if skill_md.exists():
try:
content = skill_md.read_text(encoding='utf-8')
# 匹配 version: "1.2.0" 或 version: "1.2.0"
match = re.search(r'version:\s*["\']?(\d+\.\d+\.\d+)["\']?', content, re.IGNORECASE)
if match:
return match.group(1)
except Exception:
pass
return None
def get_skill_info(skill_path: Path) -> dict:
"""获取 Skill 的详细信息"""
skill_md = skill_path / "SKILL.md"
if not skill_md.exists():
skill_md = skill_path / "skill.md"
info = {
"name": skill_path.name,
"version": None,
"description": None,
"author": None,
"homepage": None,
"updated_at": None,
}
if skill_md.exists():
try:
content = skill_md.read_text(encoding='utf-8')
# 提取 version
match = re.search(r'version:\s*["\']?(\d+\.\d+\.\d+)["\']?', content, re.IGNORECASE)
if match:
info["version"] = match.group(1)
# 提取 description
match = re.search(r'description:\s*["\']([^"\']+)["\']', content, re.IGNORECASE)
if match:
info["description"] = match.group(1)
# 提取 author
match = re.search(r'author:\s*["\']([^"\']+)["\']', content, re.IGNORECASE)
if match:
info["author"] = match.group(1)
# 提取 homepage
match = re.search(r'homepage:\s*["\']([^"\']+)["\']', content, re.IGNORECASE)
if match:
info["homepage"] = match.group(1)
except Exception:
pass
return info
def parse_github_url(url: str) -> Optional[tuple]:
"""解析 GitHub URL,返回 (owner, repo, subpath)
支持格式:
- https://github.com/owner/repo
- https://github.com/owner/repo/tree/branch/path/to/skill
- owner/repo
- owner/repo/branch/path/to/skill
"""
if not url:
return None
# 处理简写格式 owner/repo[/branch/path]
if '/' in url and not url.startswith('http'):
parts = url.split('/')
if len(parts) >= 2:
subpath = '/'.join(parts[2:]) if len(parts) > 2 else None
return (parts[0], parts[1], subpath)
return None
# 处理完整 URL
parsed = urlparse(url)
if 'github.com' in parsed.netloc:
path_parts = [p for p in parsed.path.split('/') if p]
if len(path_parts) >= 2:
owner = path_parts[0]
repo = path_parts[1].replace('.git', '')
# 处理 /tree/branch/subpath 格式
if len(path_parts) >= 5 and path_parts[2] == 'tree':
subpath = '/'.join(path_parts[4:])
return (owner, repo, subpath if subpath else None)
# 处理 /blob/branch/subpath 格式
elif len(path_parts) >= 5 and path_parts[2] == 'blob':
subpath = '/'.join(path_parts[4:])
return (owner, repo, subpath if subpath else None)
else:
subpath = '/'.join(path_parts[2:]) if len(path_parts) > 2 else None
return (owner, repo, subpath)
return None
def get_raw_github_content(owner: str, repo: str, path: str = None, branch: str = "main") -> Optional[str]:
"""通过 raw.githubusercontent.com 获取 GitHub 文件内容"""
try:
if path:
url = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{path}"
else:
# 默认获取 SKILL.md
url = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/SKILL.md"
import urllib.request
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
with urllib.request.urlopen(req, timeout=10) as resp:
return resp.read().decode('utf-8')
except Exception:
return None
def get_github_changelog(owner: str, repo: str, branch: str = "main", skill_path: str = None) -> Optional[str]:
"""获取 GitHub 仓库的 CHANGELOG 内容"""
# 构建 CHANGELOG.md 的路径
if skill_path:
# 如果 skill_path 是目录,需要加上 CHANGELOG.md
if not skill_path.endswith("CHANGELOG.md"):
skill_path = f"{skill_path}/CHANGELOG.md"
else:
skill_path = "CHANGELOG.md"
return get_raw_github_content(owner, repo, skill_path, branch)
def get_github_commits(owner: str, repo: str, branch: str = "main", skill_path: str = None, limit: int = 5) -> list:
"""获取 GitHub 仓库/文件的最近 commits"""
try:
# 构建 commits URL
if skill_path:
url = f"https://github.com/{owner}/{repo}/commits/{branch}/{skill_path}"
else:
url = f"https://github.com/{owner}/{repo}/commits/{branch}"
html = fetch_webpage(url)
if not html:
return []
commits = []
# GitHub 在 <script type="application/json" data-target="react-app.embeddedData"> 中嵌入了 JSON
json_pattern = re.compile(
r'<script[^>]*data-target="react-app\.embeddedData"[^>]*>\s*(\{.*?\})\s*</script>',
re.DOTALL
)
match = json_pattern.search(html)
if match:
try:
payload = json.loads(match.group(1))
for group in payload.get("payload", {}).get("commitGroups", []):
for c in group.get("commits", []):
commit = {
"sha": c.get("oid", "")[:7],
"time": c.get("committedDate", ""),
"message": c.get("shortMessage", ""),
"url": f"https://github.com/{owner}/{repo}/commit/{c.get('oid', '')[:7]}"
}
if commit["sha"] and commit["sha"] not in [x["sha"] for x in commits]:
commits.append(commit)
if len(commits) >= limit:
return commits
except (json.JSONDecodeError, KeyError):
pass
return commits
except Exception:
return []
def get_github_skill_version(owner: str, repo: str, branch: str = "main", skill_path: str = None) -> Optional[tuple]:
"""获取远程 Skill 的版本号和最后更新时间"""
# 构建 SKILL.md 的路径
if skill_path:
# 如果 skill_path 是目录,需要加上 SKILL.md
if not skill_path.endswith("SKILL.md"):
skill_path = f"{skill_path}/SKILL.md"
else:
skill_path = "SKILL.md"
content = get_raw_github_content(owner, repo, skill_path, branch)
if not content:
return None, None
version = None
updated = None
# 匹配 version
match = re.search(r'version:\s*["\']?(\d+\.\d+\.\d+)["\']?', content, re.IGNORECASE)
if match:
version = match.group(1)
# 尝试从原始 URL 获取最后更新
# 注意:raw.githubusercontent.com 不提供修改时间,我们需要通过其他方式
# 一种方式是检查 GitHub commits API,但用户说不使用 API
# 另一种方式是在返回内容中标注获取时间
return version, get_current_timestamp()
def fetch_webpage(url: str) -> Optional[str]:
"""获取网页内容"""
try:
import urllib.request
req = urllib.request.Request(url, headers={
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
})
with urllib.request.urlopen(req, timeout=15) as resp:
return resp.read().decode('utf-8', errors='ignore')
except Exception:
return None
def get_github_last_modified(owner: str, repo: str, skill_path: str = None, branch: str = "main") -> Optional[str]:
"""获取 GitHub 文件的最后修改时间(通过网页)"""
# 构建文件的路径
if skill_path:
# 如果 skill_path 是目录,需要加上 SKILL.md
if not skill_path.endswith("SKILL.md"):
skill_path = f"{skill_path}/SKILL.md"
else:
skill_path = "SKILL.md"
try:
# GitHub 的 blob 页面有 relative-time 标签
url = f"https://github.com/{owner}/{repo}/blob/{branch}/{skill_path}"
html = fetch_webpage(url)
if html:
# 查找 <relative-time> 标签
match = re.search(r'<relative-time[^>]*datetime="([^"]+)"', html)
if match:
return match.group(1)
# 备用:查找 commit 信息
match = re.search(r'data-pjax-date="([^"]+)"', html)
if match:
return match.group(1)
except Exception:
pass
return None
def get_local_skill_info(skill_path: Path) -> dict:
"""获取本地已安装 Skill 的完整信息"""
registry = load_registry()
name = skill_path.name
# 先从注册表获取基础信息
base_info = registry.get(name, {})
# 获取 SKILL.md 中的信息
skill_md = skill_path / "SKILL.md"
if not skill_md.exists():
skill_md = skill_path / "skill.md"
if skill_md.exists():
content = skill_md.read_text(encoding='utf-8')
# 提取没有的信息
if not base_info.get("description"):
match = re.search(r'description:\s*["\']([^"\']+)["\']', content, re.IGNORECASE)
if match:
base_info["description"] = match.group(1)
if not base_info.get("version"):
match = re.search(r'version:\s*["\']?(\d+\.\d+\.\d+)["\']?', content, re.IGNORECASE)
if match:
base_info["version"] = match.group(1)
base_info["current_version"] = match.group(1)
if not base_info.get("homepage"):
match = re.search(r'homepage:\s*["\']([^"\']+)["\']', content, re.IGNORECASE)
if match:
base_info["homepage"] = match.group(1)
else:
base_info["current_version"] = base_info.get("version")
base_info["name"] = name
base_info["local_path"] = str(skill_path)
return base_info
def record_install(skill_name: str, source: str, skill_path: Path = None, action: str = "install",
install_type: str = None, install_commit: str = None,
install_branch: str = None, remote_url: str = None,
remote_subpath: str = None):
"""记录 Skill 安装/更新"""
registry = load_registry()
# 自动推断 install_type
if not install_type:
if source and ("github.com" in source or ("/" in source and not source.startswith("/") and not source.startswith("~"))):
install_type = "remote"
else:
install_type = "local"
info = {
"name": skill_name,
"source": source,
"install_type": install_type,
"installed_at": get_current_timestamp(),
"last_updated": get_current_timestamp(),
"installed_version": None,
"current_version": None,
"description": None,
"homepage": None,
"install_commit": install_commit,
"install_branch": install_branch,
"remote_url": remote_url or (source if install_type == "remote" else None),
"remote_subpath": remote_subpath,
"last_check_at": None,
}
# 如果有本地路径,读取 SKILL.md
if skill_path and skill_path.exists():
skill_info = get_skill_info(skill_path)
info["installed_version"] = skill_info.get("version")
info["current_version"] = skill_info.get("version")
info["description"] = skill_info.get("description")
info["homepage"] = skill_info.get("homepage")
# 解析 GitHub 来源获取远程版本
if source:
gh_info = parse_github_url(source)
if gh_info:
owner, repo, subpath = gh_info
remote_version, _ = get_github_skill_version(owner, repo, skill_path=subpath)
if remote_version:
info["latest_version"] = remote_version
if not info["current_version"]:
info["current_version"] = remote_version
# 更新或创建记录
registry[skill_name] = info
save_registry(registry)
return info
def record_update(skill_name: str, old_version: str = None, new_version: str = None):
"""记录 Skill 更新"""
registry = load_registry()
if skill_name in registry:
registry[skill_name]["last_updated"] = get_current_timestamp()
if new_version:
registry[skill_name]["current_version"] = new_version
registry[skill_name]["installed_version"] = old_version or registry[skill_name].get("installed_version")
save_registry(registry)
return registry.get(skill_name)
def check_updates_for_skill(skill_name: str) -> dict:
"""检查单个 Skill 的更新状态"""
registry = load_registry()
skill_info = registry.get(skill_name, {})
source = skill_info.get("source")
current_version = skill_info.get("current_version")
installed_at = skill_info.get("installed_at")
result = {
"name": skill_name,
"source": source,
"current_version": current_version,
"latest_version": current_version,
"has_update": False,
"changelog": None,
"commits": [],
"last_modified": None,
"error": None,
}
if not source:
result["error"] = "无来源信息"
return result
# 解析 GitHub 来源
# 优先使用 remote_subpath(精确到 Skill 子目录),其次解析 remote_url
remote_url = skill_info.get("remote_url") or source
remote_subpath = skill_info.get("remote_subpath")
gh_info = parse_github_url(remote_url)
if not gh_info:
result["error"] = "无法解析来源"
return result
owner, repo, url_subpath = gh_info
# 使用精确的 subpath:remote_subpath > URL 解析 > None
subpath = remote_subpath or url_subpath
# 策略 1: 有版本号 → 比较版本
remote_version, fetch_time = get_github_skill_version(owner, repo, skill_path=subpath)
if remote_version and current_version:
result["latest_version"] = remote_version
result["has_update"] = remote_version != current_version
# 策略 2: 无版本号 → 检查最近 commits 是否有新于安装时间的
if not result["has_update"] and not current_version:
commits = get_github_commits(owner, repo, skill_path=subpath)
if commits and installed_at:
try:
installed_dt = datetime.fromisoformat(installed_at)
for c in commits:
commit_time = c.get("time", "")
if commit_time:
commit_dt = datetime.fromisoformat(commit_time.replace("Z", "+00:00"))
# commit 时间晚于安装时间 → 有更新
if commit_dt > installed_dt:
result["has_update"] = True
break
except (ValueError, TypeError):
# 时间解析失败,有 commits 就认为可能有更新
if commits:
result["has_update"] = True
if commits:
result["commits"] = commits
# 补充信息:changelog 和最后修改时间
changelog = get_github_changelog(owner, repo, skill_path=subpath)
if changelog:
result["changelog"] = changelog
# 如果策略 1 没获取到 commits(有版本号时也要获取 commits 作为补充信息)
if not result["commits"]:
commits = get_github_commits(owner, repo, skill_path=subpath)
if commits:
result["commits"] = commits
last_modified = get_github_last_modified(owner, repo, skill_path=subpath)
if last_modified:
result["last_modified"] = last_modified
# 更新 last_check_at
now = get_current_timestamp()
reg = load_registry()
if skill_name in reg:
reg[skill_name]["last_check_at"] = now
if remote_version:
reg[skill_name]["latest_version"] = remote_version
save_registry(reg)
result["last_check_at"] = now
return result
def check_all_updates() -> list:
"""检查所有已安装 Skill 的更新状态(仅远程安装的 Skill 检查远程更新)"""
registry = load_registry()
results = []
for skill_name, entry in registry.items():
install_type = entry.get("install_type", "local")
source = entry.get("source", "")
is_remote = install_type == "remote" or "github.com" in source
if is_remote:
result = check_updates_for_skill(skill_name)
results.append(result)
else:
# 本地 Skill(符号链接)自动同步,无需远程检查
results.append({
"name": skill_name,
"source": source,
"current_version": entry.get("current_version"),
"latest_version": entry.get("current_version"),
"has_update": False,
"changelog": None,
"commits": [],
"last_modified": None,
"error": None,
"skipped": True,
})
return results
DEFAULT_STALE_THRESHOLD_DAYS = 7
def check_stale_remote_skills(threshold_days: int = DEFAULT_STALE_THRESHOLD_DAYS) -> list:
"""检查超过阈值的远程 Skill 更新状态"""
registry = load_registry()
results = []
now = datetime.now().astimezone()
for skill_name, entry in registry.items():
source = entry.get("source", "")
install_type = entry.get("install_type", "local")
is_remote = install_type == "remote" or "github.com" in source
if not is_remote:
continue
last_check = entry.get("last_check_at")
if last_check:
try:
last_check_dt = datetime.fromisoformat(last_check)
days_since = (now - last_check_dt).days
if days_since < threshold_days:
continue
except (ValueError, TypeError):
pass
result = check_updates_for_skill(skill_name)
results.append(result)
return results
def summarize_updates(result: dict) -> str:
"""将更新检测结果总结为一段简明的更新摘要"""
commits = result.get("commits", [])
changelog = result.get("changelog")
current = result.get("current_version")
latest = result.get("latest_version")
parts = []
# 版本变化
if current and latest and current != latest:
parts.append(f"版本 {current} → {latest}")
# 从 changelog 提取最新版本的更新说明
changelog_extracted = False
if changelog:
sections = re.split(r'^##\s', changelog, maxsplit=2)
if len(sections) > 1:
latest_section = sections[1]
summary_lines = []
for line in latest_section.split('\n'):
line = line.strip().lstrip('- ').strip()
if line and not line.startswith('#') and not line.startswith('['):
summary_lines.append(line)
if len(summary_lines) >= 3:
break
if summary_lines:
parts.append(';'.join(summary_lines))
changelog_extracted = True
# 从 commits 生成摘要(仅在 changelog 未提取到内容时)
if commits and not changelog_extracted:
summaries = []
for c in commits[:5]:
msg = c.get("message", "")
if not msg:
continue
if msg.startswith("Merge pull request"):
continue
msg = msg.split('\n')[0]
msg = re.sub(r'^(feat|fix|docs|chore|refactor|perf|test|ci|build|style)(\([^)]*\))?:\s*', '', msg)
if msg:
summaries.append(msg[:50])
if summaries:
parts.append(';'.join(summaries))
# 兜底:只说有多少新 commit
if not parts and commits:
parts.append(f"安装后有 {len(commits)} 个新提交")
return ','.join(parts) if parts else "检测到远程更新"
def format_recommendation_report(results: list) -> str:
"""格式化更新推荐报告(仅对有更新的 skill 输出)"""
has_update = [r for r in results if r.get("has_update")]
if not has_update:
return ""
lines = []
lines.append("=" * 60)
lines.append("Skill 更新推荐")
lines.append("=" * 60)
lines.append("")
for r in has_update:
lines.append(f" * {r['name']}")
# 更新摘要
summary = summarize_updates(r)
lines.append(f" {summary}")
if r.get("source"):
lines.append(f" 更新: skill-manager install {r['source']}")
lines.append("")
lines.append("=" * 60)
return "\n".join(lines)
def format_version_list(updates: list) -> str:
"""格式化版本列表为可读文本"""
if not updates:
return "没有已安装的 Skills\n"
lines = []
lines.append("=" * 60)
lines.append("🔍 Skill 更新检查")
lines.append("=" * 60)
lines.append("")
# 按状态分组
has_update = [u for u in updates if u.get("has_update")]
up_to_date = [u for u in updates if not u.get("has_update") and not u.get("error") and not u.get("skipped")]
local_skills = [u for u in updates if u.get("skipped")]
errors = [u for u in updates if u.get("error")]
if has_update:
lines.append("📦 有可用更新:")
lines.append("-" * 40)
for u in has_update:
lines.append(f" • {u['name']}")
summary = summarize_updates(u)
lines.append(f" {summary}")
lines.append("")
if up_to_date:
lines.append("✅ 已是最新:")
lines.append("-" * 40)
for u in up_to_date:
lines.append(f" • {u['name']} (v{u.get('current_version', '?')})")
lines.append("")
if errors:
lines.append("⚠️ 检查失败:")
lines.append("-" * 40)
for u in errors:
lines.append(f" • {u['name']}: {u.get('error')}")
lines.append("")
if local_skills:
lines.append("🔗 本地安装(自动同步,无需检查):")
lines.append("-" * 40)
for u in local_skills:
ver = u.get('current_version', '?')
lines.append(f" • {u['name']}" + (f" (v{ver})" if ver else ""))
lines.append("")
lines.append("=" * 60)
return "\n".join(lines)
def format_install_record(info: dict) -> str:
"""格式化安装记录"""
lines = []
lines.append("=" * 40)
lines.append("📦 Skill 安装记录")
lines.append("=" * 40)
lines.append(f" 名称: {info.get('name', '未知')}")
lines.append(f" 来源: {info.get('source', '未知')}")
lines.append(f" 版本: {info.get('current_version', '未知')}")
lines.append(f" 时间: {info.get('installed_at', '未知')}")
if info.get("description"):
lines.append(f" 描述: {info['description']}")
lines.append("=" * 40)
return "\n".join(lines)
def main():
"""主入口"""
if len(sys.argv) < 2:
print("Usage:")
print(" record.py install <skill_name> <source> [--path <path>] [--install-type <type>] [--install-commit <hash>] [--install-branch <branch>] [--remote-url <url>]")
print(" record.py check <skill_name>")
print(" record.py check-all")
print(" record.py auto-check [--threshold <days>]")
print(" record.py list")
print(" record.py update <skill_name> [--from <version>] [--to <version>]")
sys.exit(1)
command = sys.argv[1]
if command == "install":
# record.py install <skill_name> <source> [--path <path>] [--install-type <type>] [--install-commit <hash>] [--install-branch <branch>] [--remote-url <url>] [--remote-subpath <path>]
skill_name = sys.argv[2] if len(sys.argv) > 2 else None
source = sys.argv[3] if len(sys.argv) > 3 else None
path = None
install_type = None
install_commit = None
install_branch = None
remote_url = None
remote_subpath = None
for i, arg in enumerate(sys.argv):
if arg == "--path" and len(sys.argv) > i + 1:
path = Path(sys.argv[i + 1])
elif arg == "--install-type" and len(sys.argv) > i + 1:
install_type = sys.argv[i + 1]
elif arg == "--install-commit" and len(sys.argv) > i + 1:
install_commit = sys.argv[i + 1]
elif arg == "--install-branch" and len(sys.argv) > i + 1:
install_branch = sys.argv[i + 1]
elif arg == "--remote-url" and len(sys.argv) > i + 1:
remote_url = sys.argv[i + 1]
elif arg == "--remote-subpath" and len(sys.argv) > i + 1:
remote_subpath = sys.argv[i + 1]
if not skill_name or not source:
print("❌ 错误: 请提供 skill_name 和 source")
sys.exit(1)
info = record_install(skill_name, source, path,
install_type=install_type,
install_commit=install_commit,
install_branch=install_branch,
remote_url=remote_url,
remote_subpath=remote_subpath)
print(format_install_record(info))
elif command == "auto-check":
threshold = DEFAULT_STALE_THRESHOLD_DAYS
for i, arg in enumerate(sys.argv):
if arg == "--threshold" and len(sys.argv) > i + 1:
threshold = int(sys.argv[i + 1])
results = check_stale_remote_skills(threshold_days=threshold)
report = format_recommendation_report(results)
if report:
print(report)
elif command == "check":
# record.py check <skill_name>
skill_name = sys.argv[2] if len(sys.argv) > 2 else None
if not skill_name:
print("❌ 错误: 请提供 skill_name")
sys.exit(1)
result = check_updates_for_skill(skill_name)
print(json.dumps(result, indent=2, ensure_ascii=False))
elif command == "check-all":
results = check_all_updates()
print(format_version_list(results))
elif command == "list":
registry = load_registry()
if not registry:
print("📭 暂无安装记录")
else:
for name, info in registry.items():
print(f"\n📦 {name}")
print(f" 版本: {info.get('current_version', '未知')}")
print(f" 来源: {info.get('source', '未知')}")
print(f" 类型: {info.get('install_type', '未知')}")
print(f" 安装时间: {info.get('installed_at', '未知')}")
if info.get("install_commit"):
print(f" 安装 Commit: {info['install_commit']}")
if info.get("install_branch"):
print(f" 安装 Branch: {info['install_branch']}")
if info.get("last_check_at"):
print(f" 上次检查: {info['last_check_at']}")
if info.get("description"):
print(f" 描述: {info['description']}")
elif command == "update":
# record.py update <skill_name> [--from <version>] [--to <version>]
skill_name = sys.argv[2] if len(sys.argv) > 2 else None
old_version = None
new_version = None
for i, arg in enumerate(sys.argv):
if arg == "--from" and len(sys.argv) > i + 1:
old_version = sys.argv[i + 1]
if arg == "--to" and len(sys.argv) > i + 1:
new_version = sys.argv[i + 1]
if not skill_name:
print("❌ 错误: 请提供 skill_name")
sys.exit(1)
info = record_update(skill_name, old_version, new_version)
if info:
print(f"✅ 已更新记录: {skill_name}")
if old_version and new_version:
print(f" {old_version} → {new_version}")
else:
print(f"⚠️ 未找到记录: {skill_name}")
else:
print(f"❌ 未知命令: {command}")
sys.exit(1)
if __name__ == "__main__":
main()
#!/bin/bash
# Skill & Command Manager - Remove Script
# 卸载指定的 skill 或 command
ITEM_NAME="$1"
ORIGINAL_PWD="$PWD"
# 获取脚本所在目录
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MANAGER_DIR="$(dirname "$SCRIPT_DIR")"
TARGET_HELPER="$SCRIPT_DIR/target.sh"
if [ -f "$TARGET_HELPER" ]; then
# shellcheck source=target.sh
source "$TARGET_HELPER"
else
echo "❌ 错误: 找不到目标目录识别模块: $TARGET_HELPER"
exit 1
fi
SCRIPT_AGENT_DIR="$(find_agent_config_dir "$MANAGER_DIR" "$PWD/.claude")"
AGENT_DIR="$(find_agent_config_dir "$ORIGINAL_PWD" "$SCRIPT_AGENT_DIR")"
SKILLS_DIR="$AGENT_DIR/skills"
COMMANDS_DIR="$AGENT_DIR/commands"
# 检查参数
if [ -z "$ITEM_NAME" ]; then
echo "❌ 错误: 请提供要卸载的 skill 或 command 名称"
echo ""
echo "使用方法: $0 <name>"
echo ""
echo "示例:"
echo " 卸载 skill: $0 pdf-tool"
echo " 卸载 command: $0 deepresearch"
exit 1
fi
# 查找目标(先找 skill,再找 command)
SKILL_PATH="$SKILLS_DIR/$ITEM_NAME"
COMMAND_PATH="$COMMANDS_DIR/$ITEM_NAME.md"
TARGET_PATH=""
TARGET_TYPE=""
if [ -e "$SKILL_PATH" ]; then
TARGET_PATH="$SKILL_PATH"
TARGET_TYPE="skill"
elif [ -e "$COMMAND_PATH" ]; then
TARGET_PATH="$COMMAND_PATH"
TARGET_TYPE="command"
else
echo "❌ 错误: 未找到 '$ITEM_NAME'"
echo " 请确认名称是否正确"
exit 1
fi
# 检查是否是 skill-manager 自身
if [ "$ITEM_NAME" = "skill-manager" ]; then
echo "❌ 错误: 不能卸载 skill-manager 自身"
exit 1
fi
# 确认删除
echo "⚠ 即将卸载: $ITEM_NAME"
if [ "$TARGET_TYPE" = "skill" ]; then
if [ -L "$TARGET_PATH" ]; then
echo " 类型: 符号链接 (skill)"
elif [ -d "$TARGET_PATH/.git" ]; then
echo " 类型: Git 克隆 (skill)"
else
echo " 类型: 本地目录 (skill)"
fi
else
if [ -L "$TARGET_PATH" ]; then
echo " 类型: 符号链接 (command)"
else
echo " 类型: 本地文件 (command)"
fi
fi
echo ""
read -p "确认删除? (y/N) " -n 1 -r
echo ""
if [[ ! $REPLY =~ ^[Yy]$ ]]; then
echo "已取消"
exit 0
fi
# 执行删除
if [ -L "$TARGET_PATH" ]; then
rm "$TARGET_PATH"
echo "✓ 已删除符号链接"
elif [ -d "$TARGET_PATH" ]; then
rm -rf "$TARGET_PATH"
echo "✓ 已删除目录"
elif [ -f "$TARGET_PATH" ]; then
rm "$TARGET_PATH"
echo "✓ 已删除文件"
fi
echo "✓ $TARGET_TYPE '$ITEM_NAME' 已卸载"
#!/usr/bin/env python3
"""
安全分析器 - 检测 AI Agent/Skill 中的安全风险
基于 OWASP Agentic AI Top 10 和常见漏洞模式设计,
用于识别 skill 中可能存在的恶意代码或安全隐患。
"""
import re
import json
from pathlib import Path
from typing import Dict, List, Set, Tuple, Optional
from collections import defaultdict
from dataclasses import dataclass, field
@dataclass
class SecurityFinding:
"""安全发现"""
category: str # 类别
severity: str # critical, high, medium, low
file: str # 文件路径
line: int # 行号
code: str # 代码片段
message: str # 说明
pattern: str # 匹配的模式
# 危险代码模式检测规则
DANGEROUS_PATTERNS = {
'command_execution': {
'patterns': [
r'os\.system\s*\(',
r'subprocess\.(run|call|Popen|check_output|check_call)\s*\(',
r'eval\s*\(',
r'exec\s*\(',
r'execfile\s*\(',
r'commands\.getoutput',
r'child_process\.exec',
r'child_process\.spawn',
r'child_process\.execSync',
],
'severity': 'critical',
'message': '检测到命令执行代码,可能执行任意系统命令',
},
'sensitive_file_access': {
'patterns': [
r'["\']~/.ssh/',
r'["\']~/.aws/',
r'["\']~/.gnupg/',
r'["\']?\.pem["\']?',
r'["\']?\.key["\']?',
r'\.bashrc',
r'\.zshrc',
r'\.profile',
r'\.netrc',
r'\.pgpass',
r'_netrc',
r'id_rsa',
r'id_ed25519',
r'authorized_keys',
],
'severity': 'high',
'message': '检测到敏感文件访问,可能泄露凭证',
},
'env_file_access': {
'patterns': [
r'open\s*\([^)]*\.env',
r'load_dotenv',
r'dotenv\.load',
r'process\.env\[', # Node.js
r'os\.environ',
r'os\.getenv',
],
'severity': 'medium',
'message': '检测到环境变量/配置文件访问',
},
'data_exfiltration': {
'patterns': [
r'requests\.(post|put|patch)\s*\(',
r'urllib\.request\.urlopen',
r'urllib\.request\.Request',
r'httpx\.(post|put|patch)',
r'axios\.(post|put|patch)',
r'fetch\s*\(',
r'XMLHttpRequest',
r'\.send\s*\(',
],
'severity': 'high',
'message': '检测到 HTTP 请求,可能向外部发送数据',
},
'socket_connection': {
'patterns': [
r'socket\.socket',
r'socket\.connect',
r'websocket',
r'WebSocket',
r'socket\.send',
],
'severity': 'high',
'message': '检测到 Socket 连接,可能建立隐蔽通信',
},
'obfuscation': {
'patterns': [
r'base64\.b64decode',
r'base64\.decode',
r'codecs\.decode',
r'__import__\s*\(\s*["\']codecs',
r'chr\s*\(\s*\d+\s*\)\s*\+', # chr(111)+chr(112)...
r'atob\s*\(',
r'btoa\s*\(',
r'Buffer\.from\s*\([^)]*,\s*["\']base64["\']',
],
'severity': 'high',
'message': '检测到代码混淆,可能隐藏恶意行为',
},
'privilege_escalation': {
'patterns': [
r'\bsudo\b',
r'\bdoas\b',
r'chmod\s+777',
r'chmod\s+\+x',
r'chmod\s+[0-7]*[0-7]77',
r'setuid',
r'setgid',
r'NOPASSWD',
r'setcap',
],
'severity': 'critical',
'message': '检测到权限提升操作',
},
'dynamic_import': {
'patterns': [
r'__import__\s*\(',
r'importlib\.import_module',
r'importlib\.__import__',
r'globals\s*\(\s*\)\s*\[',
r'locals\s*\(\s*\)\s*\[',
r'getattr\s*\([^,]+,\s*[^)]+\)', # 动态属性访问
r'require\s*\(\s*[^"\']', # Node.js 动态 require
],
'severity': 'medium',
'message': '检测到动态导入,可能加载任意模块',
},
'hardcoded_secrets': {
'patterns': [
r'(?i)["\']?api_key["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?api-key["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?apikey["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?secret["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?secret_key["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?password["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?passwd["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?token["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?access_token["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?auth_token["\']?\s*[=:]\s*["\'][^"\']{8,}["\']',
r'(?i)["\']?private_key["\']?\s*[=:]\s*["\'][^"\']{20,}["\']',
r'AKIA[0-9A-Z]{16}', # AWS Access Key ID
r'aws_secret_access_key',
r'sk-[a-zA-Z0-9]{20,}', # OpenAI/Anthropic API Key
r'sk-ant-[a-zA-Z0-9-]+', # Anthropic API Key
r'ghp_[a-zA-Z0-9]{36}', # GitHub Personal Access Token
r'gho_[a-zA-Z0-9]{36}', # GitHub OAuth Access Token
r'ghu_[a-zA-Z0-9]{36}', # GitHub User-to-Server Token
r'ghs_[a-zA-Z0-9]{36}', # GitHub Server-to-Server Token
r'ghr_[a-zA-Z0-9]{36}', # GitHub Refresh Token
r'xox[baprs]-[a-zA-Z0-9-]+', # Slack Token
r'glpat-[a-zA-Z0-9-]+', # GitLab Personal Access Token
r'(?i)Bearer\s+[a-zA-Z0-9._-]+', # Bearer Token
],
'severity': 'high',
'message': '检测到硬编码凭证',
},
'download_and_execute': {
'patterns': [
r'curl\s+.*\|\s*(bash|sh|zsh|python|node)',
r'wget\s+.*\|\s*(bash|sh|zsh|python|node)',
r'curl\s+.*>\s*/tmp/',
r'wget\s+.*>\s*/tmp/',
r'Invoke-Expression.*Invoke-WebRequest', # PowerShell
r'iex.*DownloadString', # PowerShell
r'/dev/tcp/', # Bash network
],
'severity': 'critical',
'message': '检测到下载并执行模式,极高风险',
},
'file_deletion': {
'patterns': [
r'rm\s+-rf\s+/',
r'rm\s+-rf\s+~',
r'shutil\.rmtree',
r'os\.remove',
r'os\.unlink',
r'fs\.rmSync',
r'fs\.unlinkSync',
r'\.remove\s*\(',
],
'severity': 'high',
'message': '检测到文件删除操作',
},
'process_manipulation': {
'patterns': [
r'os\.kill',
r'signal\.kill',
r'process\.kill',
r'pkill',
r'killall',
r'taskkill',
],
'severity': 'medium',
'message': '检测到进程操作',
},
'registry_manipulation': {
'patterns': [
r'winreg\.',
r'Register-ScheduledTask',
r'Set-ItemProperty.*HKLM',
r'Set-ItemProperty.*HKCU',
r'reg\s+add',
],
'severity': 'high',
'message': '检测到注册表操作,可能修改系统设置',
},
'persistence': {
'patterns': [
r'crontab',
r'systemctl\s+enable',
r'launchctl\s+load',
r'/etc/init\.d/',
r'/etc/systemd/',
r'~/Library/LaunchAgents/',
r'HKLM\\Software\\Microsoft\\Windows\\CurrentVersion\\Run',
r'ScheduledTask',
],
'severity': 'high',
'message': '检测到持久化机制',
},
}
# Skill 特有风险检测
SKILL_RISKS = {
'hidden_hooks': {
'patterns': [
r'post[-_]?install',
r'pre[-_]?install',
r'after[-_]?install',
r'before[-_]?install',
r'postInstall',
r'preInstall',
],
'severity': 'high',
'message': '检测到安装钩子,可能自动执行代码',
},
'mcp_server': {
'patterns': [
r'@modelcontextprotocol',
r'mcp[-_]server',
r'McpServer',
r'stdio.*transport',
r'SSE.*transport',
],
'severity': 'medium',
'message': '检测到 MCP 服务器,可能获取系统访问权限',
},
'skill_description_mismatch': {
'check': 'compare_description_to_code',
'severity': 'medium',
'message': 'SKILL.md 描述与代码功能可能不匹配',
},
}
# 可疑依赖名称(typosquatting 检测)
SUSPICIOUS_PACKAGE_PATTERNS = [
r'^requests?$', # 可能是 requests 的拼写错误
r'^numpy$', # 检查是否来自官方
r'^pytol$', # pylint typo
r'^djanga$', # django typo
r'^flaskk$', # flask typo
]
# 敏感依赖(可能带来安全风险)
SENSITIVE_DEPENDENCIES = [
'pickles', 'marshal', 'shelve', # 序列化风险
'paramiko', 'fabric', 'asyncssh', # SSH
'pexpect', 'ptyprocess', # 伪终端
'cryptography', 'pycrypto', 'pycryptodome', # 加密
]
# 提示词安全检测规则 - 用于检测 SKILL.md 中的恶意提示
PROMPT_SECURITY_PATTERNS = {
'prompt_injection': {
'patterns': [
# 经典提示注入
r'(?i)ignore\s+(all\s+)?(previous|above|prior)\s+instructions?',
r'(?i)disregard\s+(all\s+)?(previous|above)\s+instructions?',
r'(?i)forget\s+(all\s+)?(previous|above)\s+instructions?',
r'(?i)override\s+(all\s+)?(previous|default)\s+(instructions?|settings?|rules?)',
r'(?i)bypass\s+(all\s+)?(security|safety|restrictions?|filters?)',
# 越狱尝试
r'(?i)jailbreak',
r'(?i)DAN\s+mode',
r'(?i)developer\s+mode',
r'(?i)ignore\s+your\s+training',
r'(?i)you\s+are\s+now\s+free',
r'(?i)no\s+restrictions?',
r'(?i)unrestricted\s+mode',
# 角色扮演欺骗
r'(?i)pretend\s+(you\s+are|to\s+be)\s+(a\s+)?(admin|root|system|sudo)',
r'(?i)act\s+as\s+(if\s+you\s+are\s+)?(admin|root|system|sudo)',
r'(?i)role[ -]?play\s+as\s+(admin|root|system)',
],
'severity': 'critical',
'message': '检测到提示注入模式,可能试图绕过安全限制',
},
'data_collection_instruction': {
'patterns': [
# 收集敏感数据指令
r'(?i)(collect|gather|harvest|extract)\s+.*\s+(password|token|key|secret|credential)',
r'(?i)(send|upload|transmit|exfiltrate)\s+.*\s+(to\s+)?(external|third[ -]?party|remote)',
r'(?i)(read|access|scan)\s+.*\s+\.env\b',
r'(?i)(read|access|scan)\s+.*\s+(user\s+)?(ssh|aws|gpg)\s+(keys?|credentials?)',
r'(?i)(copy|steal|exfiltrate)\s+.*\s+(file|data|credential)',
r'(?i)(export|output)\s+.*\s+(environment|env)\s+variables?',
r'(?i)(log|record|save)\s+.*\s+(user\s+)?(password|token|api[ -]?key)',
r'(?i)(capture|intercept)\s+.*\s+(clipboard|input|keystroke)',
],
'severity': 'critical',
'message': '检测到数据收集指令,可能试图窃取敏感信息',
},
'execution_instruction': {
'patterns': [
# 在提示词中指令执行命令
r'(?i)(execute|run|invoke)\s+.*\s+(shell|terminal|command|script)',
r'(?i)(must|should|always)\s+.*\s+(run|execute)\s+.*\s+(on\s+)?(install|startup|load)',
r'(?i)(download|fetch)\s+.*\s+and\s+(execute|run|install)',
r'(?i)curl\s+\S+\s*\|\s*(bash|sh|python|node)',
r'(?i)wget\s+\S+\s*\|\s*(bash|sh|python|node)',
r'(?i)(install|setup)\s+.*\s+(automatically|silently|background)',
r'(?i)(spawn|fork|launch)\s+.*\s+(process|subprocess|child)',
],
'severity': 'high',
'message': '检测到执行指令,可能在提示词中隐藏恶意执行逻辑',
},
'privilege_escalation_instruction': {
'patterns': [
# 权限提升指令
r'(?i)(run|execute)\s+.*\s+with\s+(root|admin|sudo|elevated)\s+(privileges?|access|permissions?)',
r'(?i)(grant|give|obtain)\s+.*\s+(root|admin|full)\s+(access|permissions?)',
r'(?i)sudo\s+(apt|yum|brew|pip|npm)',
r'(?i)(disable|turn\s+off|bypass)\s+.*\s+(security|firewall|antivirus|protection)',
r'(?i)(modify|edit|change)\s+.*\s+(system|host)\s+(files?|settings?|config)',
],
'severity': 'high',
'message': '检测到权限提升指令',
},
'deceptive_description': {
'patterns': [
# 欺骗性描述 - 描述与功能不符
r'(?i)(simple|harmless|safe|innocent)\s+.*\s+(tool|script|utility)',
r'(?i)(just|only|merely)\s+.*\s+(checks?|reads?|displays?)\s+.*\s+(file|system|info)',
r'(?i)for\s+(educational|research|testing)\s+purposes?\s+only',
r'(?i)(hidden|secret|stealth|covert)\s+.*\s+(feature|mode|function)',
r'(?i)do\s+not\s+(run|use|execute)\s+.*\s+(in\s+)?(production|real)',
],
'severity': 'medium',
'message': '检测到可疑的描述模式,可能存在欺骗性说明',
},
'network_instruction': {
'patterns': [
# 网络外泄指令
r'(?i)(connect|phone\s+home|beacon)\s+.*\s+(to\s+)?(server|endpoint|api)',
r'(?i)(post|send|upload)\s+.*\s+(data|file|credential)\s+.*\s+to\s+\S+',
r'(?i)(establish|create|open)\s+.*\s+(reverse|backdoor|covert)\s+(shell|connection)',
r'(?i)(ping|check[ -]in)\s+.*\s+(with|to)\s+.*\s+(server|c2|command[ -]and[ -]control)',
r'(?i)websocket\s+.*\s+(connect|client)',
],
'severity': 'high',
'message': '检测到网络通信指令,可能存在数据外泄风险',
},
'persistence_instruction': {
'patterns': [
# 持久化指令
r'(?i)(add|create|install)\s+.*\s+(cron|scheduled\s+task|startup|launch)',
r'(?i)(persist|remain|stay)\s+.*\s+(running|active|installed)',
r'(?i)(auto[ -]?start|auto[ -]?run|auto[ -]?launch)',
r'(?i)(background|daemon)\s+.*\s+(process|service|task)',
r'(?i)(modify|edit)\s+.*\s+(\.\w*rc|profile|bashrc|zshrc)',
],
'severity': 'high',
'message': '检测到持久化指令,可能试图建立长期驻留',
},
'hidden_instruction': {
'patterns': [
# 隐藏指令 - 使用特殊字符或编码
r'[\x00-\x08\x0b\x0c\x0e-\x1f]', # 控制字符
r'\\u[0-9a-fA-F]{4}', # Unicode 转义
r'&#x[0-9a-fA-F]+;', # HTML 实体
r'&#[0-9]+;', # HTML 数字实体
r'\\x[0-9a-fA-F]{2}', # 十六进制转义
# 零宽字符
r'[\u200b-\u200f\u2028-\u202f\u205f-\u206f]',
],
'severity': 'medium',
'message': '检测到隐藏字符或编码,可能隐藏恶意内容',
},
'social_engineering': {
'patterns': [
# 社会工程学模式
r'(?i)(urgent|critical|important)\s+.*\s+(update|security|patch)',
r'(?i)(verify|confirm|validate)\s+.*\s+(your\s+)?(identity|account|credential)',
r'(?i)(winner|won|selected|chosen)\s+.*\s+(for|to)',
r'(?i)(limited|expires?\s+soon|act\s+now)',
r'(?i)(trust|safe|verified|official|authentic)',
],
'severity': 'medium',
'message': '检测到社会工程学模式,可能用于欺骗用户',
},
}
class SecurityAnalyzer:
"""AI Agent/Skill 安全分析器"""
def __init__(self, repo_path: str):
"""初始化分析器
Args:
repo_path: 仓库本地路径
"""
self.repo_path = Path(repo_path)
if not self.repo_path.exists():
raise ValueError(f"仓库路径不存在: {repo_path}")
self.findings: List[SecurityFinding] = []
self.exclude_dirs = {'.git', 'node_modules', '__pycache__', 'venv', '.venv',
'dist', 'build', 'target', '.cache', 'vendor'}
def analyze(self) -> Dict:
"""执行完整安全分析
Returns:
安全分析结果
"""
self.findings = []
# 执行各项检测
skill_structure = self._analyze_skill_structure()
dangerous_patterns = self._detect_dangerous_patterns()
dependencies = self._analyze_dependencies()
network = self._analyze_network_activity()
file_ops = self._analyze_file_operations()
secrets = self._detect_secrets()
prompt_security = self._analyze_prompt_security() # 新增:提示词安全检测
# 计算风险汇总
risk_summary = self._calculate_risk_summary()
return {
'risk_summary': risk_summary,
'skill_structure': skill_structure,
'dangerous_patterns': dangerous_patterns,
'dependencies': dependencies,
'network': network,
'file_operations': file_ops,
'secrets': secrets,
'prompt_security': prompt_security, # 新增
'findings': self._serialize_findings(),
}
def _analyze_skill_structure(self) -> Dict:
"""分析 Skill 结构"""
result = {
'is_skill': False,
'has_skill_md': False,
'has_scripts': False,
'has_hooks': False,
'has_mcp': False,
'warnings': [],
}
# 检查是否是 skill 结构
skill_md = self.repo_path / 'SKILL.md'
if skill_md.exists():
result['has_skill_md'] = True
result['is_skill'] = True
# 解析 SKILL.md
try:
content = skill_md.read_text(encoding='utf-8')
result['description'] = self._extract_description(content)
# 检测 MCP 相关
if any(p in content.lower() for p in ['mcp', 'modelcontextprotocol']):
result['has_mcp'] = True
result['warnings'].append('MCP 服务器配置检测到')
except Exception:
pass
# 检查脚本目录
scripts_dir = self.repo_path / 'scripts'
if scripts_dir.exists() and scripts_dir.is_dir():
result['has_scripts'] = True
script_files = list(scripts_dir.glob('*.py')) + list(scripts_dir.glob('*.js'))
result['script_count'] = len(script_files)
# 检查 package.json 中的 hooks
package_json = self.repo_path / 'package.json'
if package_json.exists():
try:
data = json.loads(package_json.read_text(encoding='utf-8'))
if 'scripts' in data:
hooks = ['postinstall', 'preinstall', 'poststart', 'prestart']
for hook in hooks:
if hook in data['scripts']:
result['has_hooks'] = True
result['warnings'].append(f'检测到 npm hook: {hook}')
except Exception:
pass
return result
def _extract_description(self, content: str) -> str:
"""从 SKILL.md 提取描述"""
# 尝试从 frontmatter 提取
if content.startswith('---'):
parts = content.split('---', 2)
if len(parts) >= 3:
frontmatter = parts[1]
for line in frontmatter.split('\n'):
if line.startswith('description:'):
return line.split(':', 1)[1].strip().strip('"').strip("'")
# 尝试从第一段提取
lines = content.split('\n')
for line in lines:
line = line.strip()
if line and not line.startswith('#') and not line.startswith('---'):
return line[:200]
return ""
def _detect_dangerous_patterns(self) -> Dict:
"""检测危险代码模式"""
results = defaultdict(list)
# 获取所有代码文件
code_extensions = {'.py', '.js', '.jsx', '.ts', '.tsx', '.go', '.rs',
'.java', '.sh', '.bash', '.ps1', '.rb', '.php'}
code_files = []
for ext in code_extensions:
code_files.extend(self.repo_path.rglob(f'*{ext}'))
# 过滤排除目录
code_files = [f for f in code_files
if not any(exc in f.parts for exc in self.exclude_dirs)]
for file_path in code_files:
self._scan_file_for_patterns(file_path, results)
# 合并 Skill 特有检测
self._scan_skill_risks(results)
return dict(results)
def _scan_file_for_patterns(self, file_path: Path, results: Dict):
"""扫描文件中的危险模式"""
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
lines = content.split('\n')
rel_path = str(file_path.relative_to(self.repo_path))
for category, config in DANGEROUS_PATTERNS.items():
for pattern in config['patterns']:
try:
regex = re.compile(pattern, re.IGNORECASE)
for i, line in enumerate(lines, 1):
if regex.search(line):
finding = SecurityFinding(
category=category,
severity=config['severity'],
file=rel_path,
line=i,
code=line.strip()[:100],
message=config['message'],
pattern=pattern,
)
self.findings.append(finding)
results[category].append({
'file': rel_path,
'line': i,
'code': line.strip()[:100],
'severity': config['severity'],
'message': config['message'],
})
except re.error:
pass
except Exception:
pass
def _scan_skill_risks(self, results: Dict):
"""扫描 Skill 特有风险"""
for category, config in SKILL_RISKS.items():
if 'patterns' not in config:
continue
for file_path in self.repo_path.rglob('*'):
if not file_path.is_file():
continue
if any(exc in file_path.parts for exc in self.exclude_dirs):
continue
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
lines = content.split('\n')
rel_path = str(file_path.relative_to(self.repo_path))
for pattern in config['patterns']:
try:
regex = re.compile(pattern, re.IGNORECASE)
for i, line in enumerate(lines, 1):
if regex.search(line):
finding = SecurityFinding(
category=f'skill_{category}',
severity=config['severity'],
file=rel_path,
line=i,
code=line.strip()[:100],
message=config['message'],
pattern=pattern,
)
self.findings.append(finding)
results[f'skill_{category}'].append({
'file': rel_path,
'line': i,
'code': line.strip()[:100],
'severity': config['severity'],
'message': config['message'],
})
except re.error:
pass
except Exception:
pass
def _analyze_dependencies(self) -> Dict:
"""分析依赖安全"""
result = {
'package_managers': [],
'total_dependencies': 0,
'high_risk_dependencies': [],
'warnings': [],
}
# Python 依赖
req_files = ['requirements.txt', 'pyproject.toml', 'setup.py', 'Pipfile']
for req_file in req_files:
path = self.repo_path / req_file
if path.exists():
result['package_managers'].append('python')
deps = self._parse_python_deps(path)
result['total_dependencies'] += len(deps)
result['high_risk_dependencies'].extend(
self._check_sensitive_deps(deps, 'python')
)
# Node.js 依赖
package_json = self.repo_path / 'package.json'
if package_json.exists():
result['package_managers'].append('nodejs')
try:
data = json.loads(package_json.read_text(encoding='utf-8'))
deps = {**data.get('dependencies', {}), **data.get('devDependencies', {})}
result['total_dependencies'] += len(deps)
result['high_risk_dependencies'].extend(
self._check_sensitive_deps(list(deps.keys()), 'nodejs')
)
except Exception:
pass
if result['high_risk_dependencies']:
result['warnings'].append(
f"发现 {len(result['high_risk_dependencies'])} 个敏感依赖"
)
return result
def _parse_python_deps(self, path: Path) -> List[str]:
"""解析 Python 依赖文件"""
deps = []
try:
content = path.read_text(encoding='utf-8')
for line in content.split('\n'):
line = line.strip()
if line and not line.startswith('#'):
# 提取包名
match = re.match(r'^([a-zA-Z0-9_-]+)', line)
if match:
deps.append(match.group(1))
except Exception:
pass
return deps
def _check_sensitive_deps(self, deps: List[str], manager: str) -> List[Dict]:
"""检查敏感依赖"""
high_risk = []
for dep in deps:
dep_lower = dep.lower()
for sensitive in SENSITIVE_DEPENDENCIES:
if sensitive in dep_lower:
high_risk.append({
'name': dep,
'manager': manager,
'risk': f'敏感依赖: {sensitive}',
})
return high_risk
def _analyze_network_activity(self) -> Dict:
"""分析网络活动"""
result = {
'external_urls': [],
'websockets': [],
'api_endpoints': [],
'warnings': [],
}
url_pattern = r'https?://[^\s\'"<>]+'
ws_pattern = r'wss?://[^\s\'"<>]+'
for file_path in self.repo_path.rglob('*'):
if not file_path.is_file():
continue
if any(exc in file_path.parts for exc in self.exclude_dirs):
continue
if file_path.suffix not in {'.py', '.js', '.ts', '.jsx', '.tsx', '.json',
'.md', '.yml', '.yaml', '.sh'}:
continue
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
rel_path = str(file_path.relative_to(self.repo_path))
# 检测 URL
urls = re.findall(url_pattern, content)
for url in urls:
# 过滤常见的安全 URL
if not any(safe in url for safe in
['github.com', 'example.com', 'localhost', '127.0.0.1',
'docs.', 'help.', 'readme']):
result['external_urls'].append({
'file': rel_path,
'url': url[:100],
})
# 检测 WebSocket
ws_urls = re.findall(ws_pattern, content)
for url in ws_urls:
result['websockets'].append({
'file': rel_path,
'url': url[:100],
})
except Exception:
pass
# 去重
seen_urls = set()
unique_urls = []
for item in result['external_urls']:
if item['url'] not in seen_urls:
seen_urls.add(item['url'])
unique_urls.append(item)
result['external_urls'] = unique_urls[:20] # 限制数量
if result['external_urls']:
result['warnings'].append(f"发现 {len(result['external_urls'])} 个外部 URL")
return result
def _analyze_file_operations(self) -> Dict:
"""分析文件操作"""
result = {
'sensitive_paths': [],
'file_operations': [],
'warnings': [],
}
sensitive_paths = [
'~/.ssh', '~/.aws', '~/.gnupg', '.env', '.pem', '.key',
'/etc/passwd', '/etc/shadow', '/etc/hosts',
'id_rsa', 'id_ed25519', 'authorized_keys',
'.bashrc', '.zshrc', '.profile',
]
for file_path in self.repo_path.rglob('*'):
if not file_path.is_file():
continue
if any(exc in file_path.parts for exc in self.exclude_dirs):
continue
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
rel_path = str(file_path.relative_to(self.repo_path))
for sensitive in sensitive_paths:
if sensitive in content:
result['sensitive_paths'].append({
'file': rel_path,
'path': sensitive,
})
except Exception:
pass
if result['sensitive_paths']:
result['warnings'].append(
f"发现 {len(result['sensitive_paths'])} 处敏感路径引用"
)
return result
def _detect_secrets(self) -> Dict:
"""检测硬编码的敏感信息"""
result = {
'findings': [],
'warnings': [],
}
# 使用 hardcoded_secrets 规则
if 'hardcoded_secrets' not in DANGEROUS_PATTERNS:
return result
patterns = DANGEROUS_PATTERNS['hardcoded_secrets']['patterns']
for file_path in self.repo_path.rglob('*'):
if not file_path.is_file():
continue
if any(exc in file_path.parts for exc in self.exclude_dirs):
continue
# 跳过某些文件类型
if file_path.suffix in {'.png', '.jpg', '.jpeg', '.gif', '.ico',
'.pdf', '.zip', '.tar', '.gz'}:
continue
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
lines = content.split('\n')
rel_path = str(file_path.relative_to(self.repo_path))
for pattern in patterns:
try:
regex = re.compile(pattern, re.IGNORECASE)
for i, line in enumerate(lines, 1):
match = regex.search(line)
if match:
# 脱敏处理
masked_line = self._mask_secret(line, match.group())
result['findings'].append({
'file': rel_path,
'line': i,
'type': 'potential_secret',
'masked': masked_line[:100],
})
except re.error:
pass
except Exception:
pass
if result['findings']:
result['warnings'].append(
f"发现 {len(result['findings'])} 处可能的硬编码凭证"
)
return result
def _mask_secret(self, line: str, secret: str) -> str:
"""脱敏处理"""
if len(secret) > 8:
masked = secret[:4] + '*' * (len(secret) - 8) + secret[-4:]
return line.replace(secret, masked)
return line
def _analyze_prompt_security(self) -> Dict:
"""分析 SKILL.md 和其他 Markdown 文件中的提示词安全
检测提示注入、数据收集指令、执行指令等恶意模式
"""
result = {
'files_analyzed': 0,
'findings': [],
'warnings': [],
'categories_found': set(),
}
# 获取所有 Markdown 文件
md_files = list(self.repo_path.rglob('*.md'))
# 过滤排除目录
md_files = [f for f in md_files
if not any(exc in f.parts for exc in self.exclude_dirs)]
for file_path in md_files:
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
rel_path = str(file_path.relative_to(self.repo_path))
result['files_analyzed'] += 1
# 对每个文件应用提示词安全检测规则
self._scan_prompt_content(content, rel_path, result)
except Exception:
pass
# 转换 set 为 list 以便 JSON 序列化
result['categories_found'] = list(result['categories_found'])
if result['findings']:
result['warnings'].append(
f"在 {result['files_analyzed']} 个 Markdown 文件中发现 "
f"{len(result['findings'])} 处提示词安全问题"
)
return result
def _scan_prompt_content(self, content: str, file_path: str, result: Dict):
"""扫描 Markdown 内容中的提示词安全风险"""
lines = content.split('\n')
for category, config in PROMPT_SECURITY_PATTERNS.items():
for pattern in config['patterns']:
try:
regex = re.compile(pattern, re.IGNORECASE | re.MULTILINE)
for i, line in enumerate(lines, 1):
match = regex.search(line)
if match:
# 记录发现
finding = {
'file': file_path,
'line': i,
'category': category,
'severity': config['severity'],
'message': config['message'],
'matched_text': match.group()[:50], # 限制长度
'code': line.strip()[:100],
}
result['findings'].append(finding)
result['categories_found'].add(category)
# 同时添加到全局 findings 列表
security_finding = SecurityFinding(
category=f'prompt_{category}',
severity=config['severity'],
file=file_path,
line=i,
code=line.strip()[:100],
message=f"[提示词安全] {config['message']}",
pattern=pattern,
)
self.findings.append(security_finding)
except re.error:
pass
def _calculate_risk_summary(self) -> Dict:
"""计算风险汇总"""
severity_counts = defaultdict(int)
for finding in self.findings:
severity_counts[finding.severity] += 1
# 计算总体风险等级
if severity_counts['critical'] > 0:
overall_risk = 'critical'
elif severity_counts['high'] > 3:
overall_risk = 'high'
elif severity_counts['high'] > 0 or severity_counts['medium'] > 5:
overall_risk = 'medium'
elif severity_counts['medium'] > 0 or severity_counts['low'] > 0:
overall_risk = 'low'
else:
overall_risk = 'none'
# 按类别统计
category_counts = defaultdict(int)
for finding in self.findings:
category_counts[finding.category] += 1
return {
'overall_risk': overall_risk,
'total_findings': len(self.findings),
'severity_counts': dict(severity_counts),
'category_counts': dict(category_counts),
'risk_emoji': {
'critical': '🔴',
'high': '🟠',
'medium': '🟡',
'low': '🟢',
'none': '✅',
}.get(overall_risk, '⚪'),
}
def _serialize_findings(self) -> List[Dict]:
"""序列化发现结果"""
return [
{
'category': f.category,
'severity': f.severity,
'file': f.file,
'line': f.line,
'code': f.code,
'message': f.message,
}
for f in self.findings
]
def generate_report(self) -> str:
"""生成安全分析报告"""
analysis = self.analyze()
summary = analysis['risk_summary']
lines = [
"# 安全分析报告",
"",
f"> **分析目标**: {self.repo_path.name}",
f"> **总体风险等级**: {summary['risk_emoji']} **{summary['overall_risk'].upper()}**",
"",
"---",
"",
"## 风险概览",
"",
]
# 严重程度统计
if summary['total_findings'] > 0:
lines.append("### 发现统计")
lines.append("")
lines.append("| 严重程度 | 数量 |")
lines.append("|:---------|:-----|")
for severity in ['critical', 'high', 'medium', 'low']:
count = summary['severity_counts'].get(severity, 0)
emoji = {'critical': '🔴', 'high': '🟠', 'medium': '🟡', 'low': '🟢'}.get(severity, '⚪')
if count > 0:
lines.append(f"| {emoji} {severity.upper()} | {count} |")
lines.append("")
else:
lines.append("✅ **未发现安全风险**")
lines.append("")
# Skill 结构分析
skill_info = analysis['skill_structure']
if skill_info['is_skill']:
lines.append("## Skill 结构分析")
lines.append("")
lines.append(f"- **SKILL.md**: {'✅ 存在' if skill_info['has_skill_md'] else '❌ 缺失'}")
lines.append(f"- **脚本目录**: {'✅ 存在' if skill_info['has_scripts'] else '❌ 缺失'}")
if skill_info.get('has_hooks'):
lines.append(f"- **安装钩子**: ⚠️ 检测到")
if skill_info.get('has_mcp'):
lines.append(f"- **MCP 配置**: ⚠️ 检测到")
lines.append("")
# 按类别详细展示
if analysis['findings']:
lines.append("## 详细发现")
lines.append("")
# 按严重程度分组
for severity in ['critical', 'high', 'medium', 'low']:
findings = [f for f in analysis['findings'] if f['severity'] == severity]
if not findings:
continue
emoji = {'critical': '🔴', 'high': '🟠', 'medium': '🟡', 'low': '🟢'}.get(severity, '⚪')
lines.append(f"### {emoji} {severity.upper()} 风险 ({len(findings)})")
lines.append("")
# 按类别分组
by_category = defaultdict(list)
for f in findings:
by_category[f['category']].append(f)
for category, items in by_category.items():
lines.append(f"#### {category}")
lines.append("")
for item in items[:5]: # 每个类别最多显示 5 个
lines.append(f"- `{item['file']}:{item['line']}`")
lines.append(f" - {item['message']}")
lines.append(f" - 代码: `{item['code']}`")
if len(items) > 5:
lines.append(f" - ... 还有 {len(items) - 5} 个")
lines.append("")
# 依赖分析
deps = analysis['dependencies']
if deps['high_risk_dependencies']:
lines.append("## 敏感依赖")
lines.append("")
for dep in deps['high_risk_dependencies']:
lines.append(f"- **{dep['name']}** ({dep['manager']}): {dep['risk']}")
lines.append("")
# 外部 URL
network = analysis['network']
if network['external_urls']:
lines.append("## 外部网络请求")
lines.append("")
lines.append("| URL | 文件 |")
lines.append("|:----|:-----|")
for item in network['external_urls'][:10]:
lines.append(f"| `{item['url'][:50]}` | `{item['file']}` |")
if len(network['external_urls']) > 10:
lines.append(f"| ... | 还有 {len(network['external_urls']) - 10} 个 |")
lines.append("")
# 提示词安全分析
prompt_sec = analysis.get('prompt_security', {})
if prompt_sec.get('findings'):
lines.append("## 提示词安全分析")
lines.append("")
lines.append(f"> 分析了 {prompt_sec['files_analyzed']} 个 Markdown 文件")
lines.append("")
# 按类别分组
by_category = defaultdict(list)
for f in prompt_sec['findings']:
by_category[f['category']].append(f)
# 类别中文名映射
category_names = {
'prompt_injection': '提示注入',
'data_collection_instruction': '数据收集指令',
'execution_instruction': '执行指令',
'privilege_escalation_instruction': '权限提升指令',
'deceptive_description': '欺骗性描述',
'network_instruction': '网络通信指令',
'persistence_instruction': '持久化指令',
'hidden_instruction': '隐藏指令',
'social_engineering': '社会工程学',
}
for category, items in by_category.items():
cat_name = category_names.get(category, category)
severity = items[0]['severity']
emoji = {'critical': '🔴', 'high': '🟠', 'medium': '🟡', 'low': '🟢'}.get(severity, '⚪')
lines.append(f"### {emoji} {cat_name} ({len(items)})")
lines.append("")
lines.append(f"*{items[0]['message']}*")
lines.append("")
for item in items[:5]:
lines.append(f"- `{item['file']}:{item['line']}`")
lines.append(f" - 匹配: `{item['matched_text']}`")
if len(items) > 5:
lines.append(f"- ... 还有 {len(items) - 5} 个")
lines.append("")
if prompt_sec.get('warnings'):
lines.append("**警告:**")
for warning in prompt_sec['warnings']:
lines.append(f"- {warning}")
lines.append("")
# 安全建议
lines.append("## 安全建议")
lines.append("")
if summary['overall_risk'] == 'critical':
lines.append("### 🔴 高危警告")
lines.append("")
lines.append("- **强烈建议**:不要在生产环境运行此 skill")
lines.append("- 需要进行完整的代码审计后才能使用")
lines.append("- 考虑在沙箱环境中隔离运行")
lines.append("")
elif summary['overall_risk'] == 'high':
lines.append("### 🟠 警告")
lines.append("")
lines.append("- 建议在代码审计后使用")
lines.append("- 注意检查上述发现的安全问题")
lines.append("- 考虑限制 skill 的访问权限")
lines.append("")
elif summary['overall_risk'] == 'medium':
lines.append("### 🟡 注意")
lines.append("")
lines.append("- 使用前请检查上述发现项")
lines.append("- 建议在测试环境先验证")
lines.append("")
else:
lines.append("### ✅ 低风险")
lines.append("")
lines.append("- 未发现明显安全风险")
lines.append("- 建议定期进行安全检查")
lines.append("")
return '\n'.join(lines)
def main():
"""测试入口"""
import sys
if len(sys.argv) < 2:
print("用法: python -m scripts.security <仓库路径>")
sys.exit(1)
repo_path = sys.argv[1]
analyzer = SecurityAnalyzer(repo_path)
print(analyzer.generate_report())
if __name__ == '__main__':
main()
#!/bin/bash
# Shared target resolution for Skill Manager scripts.
# Returns the agent config directory that owns skills/ and commands/.
is_agent_config_dir_name() {
case "$1" in
.codex|.claude|.openclaw|.agents|.agent)
return 0
;;
*)
return 1
;;
esac
}
canonical_dir() {
local dir="$1"
if [ -d "$dir" ]; then
(cd "$dir" 2>/dev/null && pwd) || printf '%s\n' "$dir"
else
printf '%s\n' "$dir"
fi
}
find_agent_config_dir() {
local start_dir="${1:-$PWD}"
local fallback_dir="${2:-$PWD/.claude}"
local current
local home_dir="${HOME:-/Users/${USER}}"
local max_iterations=20
local iteration=0
if [ -n "${SKILL_MANAGER_TARGET_DIR:-}" ]; then
printf '%s\n' "$SKILL_MANAGER_TARGET_DIR"
return 0
fi
current="$(canonical_dir "$start_dir")"
# Global config roots: support calls from ~/.codex, ~/.claude and ~/.openclaw.
for config_name in .codex .claude .openclaw .agents .agent; do
case "$current" in
"$home_dir/$config_name"|"$home_dir/$config_name"/*)
printf '%s\n' "$home_dir/$config_name"
return 0
;;
esac
done
while [ "$iteration" -lt "$max_iterations" ]; do
local current_name
current_name="$(basename "$current")"
if is_agent_config_dir_name "$current_name"; then
printf '%s\n' "$current"
return 0
fi
# Project-local config directories. Prefer Codex when multiple configs coexist.
for config_name in .codex .claude .openclaw .agents .agent; do
if [ -d "$current/$config_name" ]; then
printf '%s\n' "$current/$config_name"
return 0
fi
done
local parent
parent="$(dirname "$current")"
if [ "$parent" = "$current" ]; then
break
fi
local parent_name
parent_name="$(basename "$parent")"
if [ "$parent_name" = "skills" ] || [ "$parent_name" = "commands" ]; then
local grandparent
grandparent="$(dirname "$parent")"
local grandparent_name
grandparent_name="$(basename "$grandparent")"
if is_agent_config_dir_name "$grandparent_name"; then
printf '%s\n' "$grandparent"
return 0
fi
fi
current="$parent"
iteration=$((iteration + 1))
done
printf '%s\n' "$fallback_dir"
}
# Find ALL agent config directories in the project.
# Returns newline-separated paths (one per line).
# - SKILL_MANAGER_TARGET_DIR set → returns that single dir
# - Called from global config root (~/codex etc.) → returns that single dir
# - Project directory → walks up to project root, returns all .codex/.claude/.openclaw found
# - Not found → returns fallback_dir
find_all_agent_config_dirs() {
local start_dir="${1:-$PWD}"
local fallback_dir="${2:-$PWD/.claude}"
local current
local home_dir="${HOME:-/Users/${USER}}"
local max_iterations=20
local iteration=0
# Explicit override → single target
if [ -n "${SKILL_MANAGER_TARGET_DIR:-}" ]; then
printf '%s\n' "$SKILL_MANAGER_TARGET_DIR"
return 0
fi
current="$(canonical_dir "$start_dir")"
# Global config roots → single target (no multi-dir in global scope)
for config_name in .codex .claude .openclaw .agents .agent; do
case "$current" in
"$home_dir/$config_name"|"$home_dir/$config_name"/*)
printf '%s\n' "$home_dir/$config_name"
return 0
;;
esac
done
# Walk up to find project root
while [ "$iteration" -lt "$max_iterations" ]; do
local current_name
current_name="$(basename "$current")"
# If inside an agent config dir, step out to project root
if is_agent_config_dir_name "$current_name"; then
current="$(dirname "$current")"
elif [ "$current_name" = "skills" ] || [ "$current_name" = "commands" ]; then
local parent
parent="$(dirname "$current")"
local parent_name
parent_name="$(basename "$parent")"
if is_agent_config_dir_name "$parent_name"; then
current="$(dirname "$parent")"
fi
fi
# Check for agent config dirs at this level
local found=0
for config_name in .codex .claude .openclaw .agents .agent; do
if [ -d "$current/$config_name" ]; then
printf '%s\n' "$current/$config_name"
found=1
fi
done
if [ "$found" -eq 1 ]; then
return 0
fi
local parent
parent="$(dirname "$current")"
if [ "$parent" = "$current" ]; then
break
fi
current="$parent"
iteration=$((iteration + 1))
done
printf '%s\n' "$fallback_dir"
}