diff --git a/.gitignore b/.gitignore index ac1be05..01b5efc 100644 --- a/.gitignore +++ b/.gitignore @@ -91,6 +91,8 @@ tools/ctf-website/_downloads/ # Local agent / IDE config (not part of public release) .codex/ +.codex_local/ +.open-reverselab-local/ .claude/settings.local.json .serena/ diff --git a/CODEX.bat b/CODEX.bat new file mode 100644 index 0000000..059be8a --- /dev/null +++ b/CODEX.bat @@ -0,0 +1,15 @@ +@echo off +setlocal +chcp 65001 >nul +cd /d "%~dp0" + +powershell.exe -NoProfile -ExecutionPolicy Bypass -File "%~dp0\scripts\codex\codex_entry.ps1" +set "EXIT_CODE=%ERRORLEVEL%" + +echo. +if not "%EXIT_CODE%"=="0" ( + echo CODEX.bat exited with code: %EXIT_CODE% +) +echo. +pause +exit /b %EXIT_CODE% diff --git a/LAUNCHER.bat b/LAUNCHER.bat new file mode 100644 index 0000000..3d8f284 --- /dev/null +++ b/LAUNCHER.bat @@ -0,0 +1,15 @@ +@echo off +setlocal +chcp 65001 >nul +cd /d "%~dp0" + +powershell.exe -NoProfile -ExecutionPolicy Bypass -File "%~dp0\scripts\codex\launcher_entry.ps1" +set "EXIT_CODE=%ERRORLEVEL%" + +echo. +if not "%EXIT_CODE%"=="0" ( + echo LAUNCHER.bat exited with code: %EXIT_CODE% +) +echo. +pause +exit /b %EXIT_CODE% diff --git a/README.md b/README.md index b4329f6..def6da0 100644 --- a/README.md +++ b/README.md @@ -108,6 +108,19 @@ python scripts/misc/ai_toolcheck.py --board misc 6. Create a task: `python scripts/misc/new_task.py --board ctf-website --name `. 7. After moving machines or changing MCP settings, run `uv run --project tools/skills/mcp/ReverseLabToolsMCP python scripts/misc/mcp_smoke_check.py --write-report` and confirm MCP tool calls pass. +## Optional Codex Launcher + +If you use Codex and want a dedicated entrypoint, use `CODEX.bat`. It installs +or repairs the global `open_reverselab_codex` MCP adapter for Codex and lets +you bind other projects to project-scoped open-reverselab mode. + +`LAUNCHER.bat` is an optional umbrella menu. Its `Codex` entry opens the same +Codex adapter workflow. Its `Claude` and `Other CLI` entries currently +delegate back to the upstream `START_HERE` flow rather than installing a new +multi-CLI adapter. + +For details, see [docs/codex/CODEX_ONLY.md](docs/codex/CODEX_ONLY.md). + Post-install verification: ```powershell diff --git a/README.zh.md b/README.zh.md index c49e303..2420018 100644 --- a/README.zh.md +++ b/README.zh.md @@ -107,6 +107,18 @@ python scripts/misc/ai_toolcheck.py --board misc 6. 创建任务：`python scripts/misc/new_task.py --board ctf-website --name `。 7. 每次换机器或重配 MCP 后，运行 `uv run --project tools/skills/mcp/ReverseLabToolsMCP python scripts/misc/mcp_smoke_check.py --write-report`，确认 MCP 真实可调用。 +## 可选 Codex 入口 + +如果你主要使用 Codex，并且想要单独的 Codex 入口，优先使用 `CODEX.bat`。 +它会为 Codex 安装或修复全局 `open_reverselab_codex` MCP 适配层，并允许你把 +其他项目绑定到项目级 open-reverselab 模式。 + +`LAUNCHER.bat` 是可选总入口。它的 `Codex` 菜单会进入同一套 Codex 适配流 +程；`Claude` 和 `其他 CLI` 菜单目前只是回到原版 `START_HERE` 流程，并不 +代表本仓库已经额外实现了一套新的多 CLI 适配层。 + +详细说明见 [docs/codex/CODEX_USAGE.zh-CN.md](docs/codex/CODEX_USAGE.zh-CN.md)。 + ## 迭代模式 ``` diff --git a/docs/codex/CODEX_ONLY.md b/docs/codex/CODEX_ONLY.md new file mode 100644 index 0000000..79da0a3 --- /dev/null +++ b/docs/codex/CODEX_ONLY.md @@ -0,0 +1,45 @@ +# Codex-only local tooling + +This local toolset is for Codex only. + +Use these root-level launchers: + +- `CODEX.bat` +- `LAUNCHER.bat` + +What it does: + +- registers the global Codex MCP server `open_reverselab_codex` +- merges a managed global Codex developer-instructions block +- preserves any existing workspace `.codex/config.toml` +- lets you bind other projects to open-reverselab Codex mode +- creates per-project `.codex/` runtime files when a target project is bound +- stores local adapter state, reports, and backups under `.open-reverselab-local/` + +What it does not do: + +- it does not replace the upstream `START_HERE.bat` +- it does not modify the repo-local upstream `.mcp.json` +- it does not manage `.claude/` +- `LAUNCHER.bat` routes its `Claude` and `Other CLI` entries back to the + upstream `START_HERE` flow; this adapter only adds a real Codex integration + path + +Recommended flow: + +- run `CODEX.bat` +- choose `1. 安装/修复 全局 Codex MCP` +- restart Codex App and Codex CLI +- open the real target project +- say `启用 open-reverselab Codex 模式` +- confirm the previewed changes when asked + +Project outputs: + +- `.codex/` +- `.open-reverselab-codex/` +- `notes/open-reverselab/` +- `reports/open-reverselab/` +- `exports/open-reverselab/` +- `patches/open-reverselab/` on demand +- `projects/open-reverselab/` on demand diff --git a/docs/codex/CODEX_USAGE.zh-CN.md b/docs/codex/CODEX_USAGE.zh-CN.md new file mode 100644 index 0000000..3c3d9ab --- /dev/null +++ b/docs/codex/CODEX_USAGE.zh-CN.md @@ -0,0 +1,233 @@ +# open-reverselab Codex 使用说明 + +## 1. 这套东西现在怎么理解 + +- `START_HERE.bat` + - 原作者入口。 + - 继续保留。 + - 主要给原版流程 / 其他 CLI 用。 + +- `LAUNCHER.bat` + - 总工作台入口。 + - 用于在 `Codex / Claude / 其他 CLI / 全部` 之间分流。 + - 其中 `Claude` 和 `其他 CLI` 当前只是回到原版 `START_HERE` 流程。 + +- `CODEX.bat` + - 现在唯一的 Codex 专用入口。 + - 启动时会先做一轮快速检查。 + - 关于 Codex 的安装、修复、项目管理、备份恢复，都从这里进入。 + +## 2. 第一次使用 + +按这个顺序来： + +1. 双击 `CODEX.bat` +2. 选 `1. 安装/修复 全局 Codex MCP` +3. 等它完成 +4. 完全关闭并重新打开 `Codex App` +5. 如果你也用 `Codex CLI`，也重新开一次 +6. 打开你真正要工作的目标项目 +7. 在那个项目的 Codex 会话里输入： + +```text +启用 open-reverselab Codex 模式 +``` + +首次接入时： + +- 会先预览要改哪些文件 +- 会要求你输入精确确认短语 +- 确认后才会真正写入 +- 会在目标项目生成 `.codex/` 受管运行层 + +## 3. 日常使用 + +### 3.1 启用当前项目 + +在目标项目会话里输入： + +```text +启用 open-reverselab Codex 模式 +``` + +英文也可以： + +```text +enable open-reverselab codex mode +``` + +### 3.2 查看当前项目状态 + +```text +查看 open-reverselab Codex 状态 +``` + +或： + +```text +show open-reverselab codex status +``` + +### 3.3 停用当前项目 + +```text +停用 open-reverselab Codex 模式 +``` + +或： + +```text +disable open-reverselab codex mode +``` + +## 4. 产物会写到哪里 + +绑定成功后，当前项目会生成这些目录： + +- `.codex/` +- `.open-reverselab-codex/` +- `notes/open-reverselab/` +- `reports/open-reverselab/` +- `exports/open-reverselab/` + +按需才会出现这些目录： + +- `patches/open-reverselab/` +- `projects/open-reverselab/` +- `samples/open-reverselab/` +- `cases/open-reverselab/` + +这表示： + +- 产物写回当前项目 +- 不再默认写回 `open-reverselab` 仓库本身 + +## 5. 如果你要手动管理别的项目 + +运行 `CODEX.bat`，进入： + +- `4. 项目管理` + +里面可以做： + +- 绑定指定项目 +- 查看指定项目状态 +- 修复指定项目接入 +- 解绑指定项目 + +适合这几种情况： + +- 某个项目的接入状态乱了 +- 你想手工给另一个项目接入 +- 你想不用会话命令，直接从菜单修复 + +## 6. 备份和恢复 + +运行 `CODEX.bat`，进入： + +- `5. 备份 / 恢复` + +这里恢复的是： + +- 接入配置 +- 接入状态 +- 受管块 +- 本地注册表 + +这里不会恢复或覆盖： + +- 分析报告 +- 导出文件 +- 业务产物 + +也就是说： + +- 恢复接入状态 +- 不碰你的分析结果 + +## 7. 当前会改哪些全局内容 + +安装后会改： + +- 用户目录下的 `.codex/config.toml` + +会加入： + +- 全局 MCP：`open_reverselab_codex` +- 全局说明层：识别启用 / 停用 / 状态 这几句固定短语 + +不会再依赖： + +- 工作区 `.codex/config.toml` 的删除式改动 + +并且： + +- 如果当前仓库本身已经有 `.codex/config.toml`，会保留，不会被 `CODEX.bat` 删除 + +## 8. 当前项目里会写哪些说明文件 + +项目第一次绑定后，会在项目里生成： + +- `.open-reverselab-codex/README.md` +- `.open-reverselab-codex/QUICK_START.md` +- `.open-reverselab-codex/TASK_TEMPLATE.md` +- `.codex/config.toml` +- `.codex/open-reverselab.config.toml` +- `.codex/open-reverselab.ctf.config.toml` +- `.codex/open-reverselab.ctf_optimized.md` + +如果项目里没有 `AGENTS.md`，会创建最小版。 + +如果有 `AGENTS.md`，会在文件末尾追加一个受管块。 + +停用时： + +- 会移除 `.codex/config.toml` 里的 open-reverselab 受管块 +- 但默认保留 `.codex/` 里的这些文件，方便后续重新启用或排查 + +## 9. 最短使用口诀 + +只记这一套就够了： + +1. `CODEX.bat` +2. `1. 安装/修复 全局 Codex MCP` +3. 重启 Codex +4. 打开目标项目 +5. 会话里输入：`启用 open-reverselab Codex 模式` + +## 10. 常见问题 + +### Q1. 以后还要先启动很多 BAT 吗？ + +不用。 + +关于 Codex，只认： + +- `CODEX.bat` +- 如果你想通过总菜单来选入口，再用： + - `LAUNCHER.bat` + +### Q2. `START_HERE.bat` 还要不要用？ + +如果你是走原作者原版流程 / 其他 CLI，可以继续用。 + +如果你现在只是在做 Codex 这套，就优先用： + +- `CODEX.bat` + +### Q3. 如果 Codex 里还显示旧 MCP 怎么办？ + +先做这几个动作： + +1. 运行 `CODEX.bat` +2. 选 `2. 启动前自检 / 状态查看` +3. 完全关闭并重新打开 Codex App + +### Q4. 怎么知道当前项目已经接入成功？ + +看两处： + +1. 会话里输入： + - `查看 open-reverselab Codex 状态` +2. 当前项目里是否存在： + - `.open-reverselab-codex/` diff --git a/scripts/codex/codex_entry.ps1 b/scripts/codex/codex_entry.ps1 new file mode 100644 index 0000000..05735a8 --- /dev/null +++ b/scripts/codex/codex_entry.ps1 @@ -0,0 +1,593 @@ +param( + [switch]$NonInteractive +) + +$ErrorActionPreference = "Stop" +[Console]::InputEncoding = [System.Text.UTF8Encoding]::new() +[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new() +$OutputEncoding = [Console]::OutputEncoding + +function Resolve-PythonCommand { + $python = Get-Command python -ErrorAction SilentlyContinue + if ($python) { + return @{ Exe = $python.Source; Args = @() } + } + + $py = Get-Command py -ErrorAction SilentlyContinue + if ($py) { + return @{ Exe = $py.Source; Args = @("-3") } + } + + return $null +} + +function Get-LocalContext { + $repoRoot = Split-Path -Parent (Split-Path -Parent $PSScriptRoot) + $localRoot = Join-Path $repoRoot ".open-reverselab-local" + return [ordered]@{ + RepoRoot = $repoRoot + LocalRoot = $localRoot + PythonRoot = Join-Path $repoRoot "scripts\codex\python" + ReportsRoot = Join-Path $localRoot "reports" + ManagerScript = Join-Path (Join-Path $repoRoot "scripts\codex\python") "open_reverselab_codex_manager.py" + FirstRunScript = Join-Path $repoRoot "scripts\misc\first_run_check.py" + StartHereScript = Join-Path $repoRoot "scripts\misc\start_here.ps1" + GlobalCodexConfig = Join-Path (Join-Path $HOME ".codex") "config.toml" + } +} + +function Invoke-ManagerJson { + param( + [hashtable]$Context, + [string[]]$ManagerArgs + ) + + $python = Resolve-PythonCommand + if (-not $python) { + throw "Python 3.10+ 未找到。请先安装 Python，并确保 PATH 中可以使用 python 或 py -3。" + } + + $commandArgs = @() + $commandArgs += $python.Args + $commandArgs += $Context.ManagerScript + $commandArgs += $ManagerArgs + + $output = & $python.Exe @commandArgs 2>&1 | Out-String + $exitCode = $LASTEXITCODE + $payload = $null + try { + $payload = $output | ConvertFrom-Json -ErrorAction Stop + } catch { + $payload = [ordered]@{ + raw_output = $output.Trim() + parse_error = $_.Exception.Message + } + } + + return [ordered]@{ + ExitCode = $exitCode + Output = $output.Trim() + Payload = $payload + } +} + +function Invoke-PythonJson { + param( + [hashtable]$Context, + [string]$ScriptPath, + [string[]]$ScriptArgs = @() + ) + + $python = Resolve-PythonCommand + if (-not $python) { + throw "Python 3.10+ 未找到。请先安装 Python，并确保 PATH 中可以使用 python 或 py -3。" + } + + $commandArgs = @() + $commandArgs += $python.Args + $commandArgs += $ScriptPath + $commandArgs += $ScriptArgs + + $output = & $python.Exe @commandArgs 2>&1 | Out-String + $exitCode = $LASTEXITCODE + $payload = $null + try { + $payload = $output | ConvertFrom-Json -ErrorAction Stop + } catch { + $payload = [ordered]@{ + raw_output = $output.Trim() + parse_error = $_.Exception.Message + } + } + + return [ordered]@{ + ExitCode = $exitCode + Output = $output.Trim() + Payload = $payload + } +} + +function Pause-Continue { + Write-Host "" + Read-Host "按回车继续" +} + +function Show-Title { + Clear-Host + Write-Host "============================================================" -ForegroundColor Cyan + Write-Host " open-reverselab Codex 工作台" -ForegroundColor Cyan + Write-Host "============================================================" -ForegroundColor Cyan + Write-Host "" +} + +function Get-StartupSnapshot { + param([hashtable]$Context) + + $status = Invoke-ManagerJson -Context $Context -ManagerArgs @("status") + $firstRun = Invoke-PythonJson -Context $Context -ScriptPath $Context.FirstRunScript -ScriptArgs @("--json") + + return [ordered]@{ + Status = $status + FirstRun = $firstRun + } +} + +function Show-StartupSummary { + param([hashtable]$Context) + + $snapshot = Get-StartupSnapshot -Context $Context + $statusPayload = $snapshot.Status.Payload + $firstRunPayload = $snapshot.FirstRun.Payload + + Write-Host "启动快速检查" -ForegroundColor Green + + if ($statusPayload.global_install_ready) { + Write-Host "- Codex 全局接入：PASS" -ForegroundColor Green + } else { + Write-Host "- Codex 全局接入：FAIL" -ForegroundColor Red + } + + if ($statusPayload.wrapper_exists -and $statusPayload.manager_exists) { + Write-Host "- Codex 本地桥接：PASS" -ForegroundColor Green + } else { + Write-Host "- Codex 本地桥接：FAIL" -ForegroundColor Red + } + + $failCount = 0 + $warnCount = 0 + if ($firstRunPayload.summary) { + $failCount = [int]$firstRunPayload.summary.fail + $warnCount = [int]$firstRunPayload.summary.warn + } + if ($failCount -gt 0) { + Write-Host "- 原版首检关键项：FAIL（FAIL=$failCount, WARN=$warnCount）" -ForegroundColor Red + } elseif ($warnCount -gt 0) { + Write-Host "- 原版首检关键项：WARN（WARN=$warnCount）" -ForegroundColor Yellow + } else { + Write-Host "- 原版首检关键项：PASS" -ForegroundColor Green + } + + Write-Host "- 当前仓库 .codex/config.toml：" -NoNewline + if ($statusPayload.workspace_codex_config_exists) { + Write-Host " 已存在（已保留，不会被本工作台删除）" -ForegroundColor Yellow + } else { + Write-Host " 未发现" -ForegroundColor DarkYellow + } + + Write-Host "- 已绑定项目数：$($statusPayload.bound_project_count)" + Write-Host "- 公开版适配版本：$($statusPayload.adapter_version)" + if ($statusPayload.replaceable_related_mcp_sections -and $statusPayload.replaceable_related_mcp_sections.Count -gt 0) { + Write-Host "- 检测到待替换旧 MCP 条目：$($statusPayload.replaceable_related_mcp_sections.Count) 个" -ForegroundColor Yellow + } + if ($statusPayload.available_upgrade_batches -and $statusPayload.available_upgrade_batches.Count -gt 0) { + Write-Host "- 可用升级批次：$($statusPayload.available_upgrade_batches.Count) 个" + } + + if ($failCount -gt 0) { + Write-Host "" + Write-Host "提示：原版首检有缺项，但不会阻止你进入 Codex 菜单。" -ForegroundColor Yellow + Write-Host "如需生成原版报告，请选 3. 完整首检 / 生成原版报告。" -ForegroundColor Yellow + } + Write-Host "" +} + +function Show-FirstUseGuide { + param([hashtable]$Context) + + Write-Host "初次使用流程" -ForegroundColor Yellow + Write-Host "1. 先选 1. 安装/修复 全局 Codex MCP。" + Write-Host "2. 安装完成后，完全关闭并重新打开 Codex App；Codex CLI 也建议重开。" + Write-Host "3. 打开你真正要工作的目标项目。" + Write-Host "4. 在新项目会话里输入：启用 open-reverselab Codex 模式。" + Write-Host "5. 首次启用会先预览改动，再要求你输入精确确认短语。" + Write-Host "6. 以后版本升级时，先关闭相关 Codex 会话，再运行 1. 安装/修复。" + Write-Host "" + Write-Host "当前仓库：" -NoNewline + Write-Host " $($Context.RepoRoot)" -ForegroundColor Green + Write-Host "" +} + +function Show-MainMenu { + Write-Host "主菜单" -ForegroundColor Green + Write-Host "1. 安装/修复 全局 Codex MCP" + Write-Host " 说明：写入用户目录下的 .codex/config.toml，注册 open_reverselab_codex，并自动升级已登记公开版项目。" + Write-Host "2. 启动前快速自检 / 状态查看" + Write-Host " 说明：查看 Codex 启动关键项，以及原版 first-run 关键摘要。" + Write-Host "3. 完整首检 / 生成原版报告" + Write-Host " 说明：调用原版 START_HERE 流程，生成 first-run-report.json 和 mcp-smoke-report.json。" + Write-Host "4. 项目管理" + Write-Host " 说明：给其他项目做绑定、状态、修复、解绑，并管理项目级 .codex 运行层。" + Write-Host "5. 备份 / 恢复" + Write-Host " 说明：优先按升级批次查看和回退，也支持底层 backup_id 恢复；不覆盖分析产物。" + Write-Host "6. 使用说明" + Write-Host " 说明：告诉你该先启动哪个 BAT、接着做什么、会话里输入什么。" + Write-Host "0. 退出" + Write-Host "" +} + +function Show-UsageGuide { + param([hashtable]$Context) + + Write-Host "推荐使用顺序" -ForegroundColor Green + Write-Host "1. 只用 Codex 时，双击 CODEX.bat。" + Write-Host "2. 如果你想看总入口，双击 LAUNCHER.bat。" + Write-Host "3. 在 CODEX.bat 里先选 1. 安装/修复 全局 Codex MCP。" + Write-Host "4. 安装后关闭并重新打开 Codex App；Codex CLI 也重新打开。" + Write-Host "5. 打开你真正要工作的目标项目。" + Write-Host "6. 在目标项目会话里输入：启用 open-reverselab Codex 模式。" + Write-Host "7. 查看状态时输入：查看 open-reverselab Codex 状态。" + Write-Host "8. 取消接入时输入：停用 open-reverselab Codex 模式。" + Write-Host "" + Write-Host "当前结构说明" -ForegroundColor Yellow + Write-Host "- START_HERE.bat：原作者原版入口，保持不动。" + Write-Host "- LAUNCHER.bat：总工作台入口，用于 Codex / Claude / 其他 CLI 分流。" + Write-Host "- CODEX.bat：Codex 专用入口，只做 Codex 相关操作。" + Write-Host "- 项目首次接入后，会在目标项目生成 .codex/ 受管文件；停用时默认保留这些文件。" + Write-Host "- 本地运行状态会写到 .open-reverselab-local/，这是本地目录，不提交到仓库。" + Write-Host "" + Write-Host "目标项目常用目录" -ForegroundColor Yellow + Write-Host "- notes/open-reverselab" + Write-Host "- reports/open-reverselab" + Write-Host "- exports/open-reverselab" + Write-Host "- .open-reverselab-codex" + Write-Host "- .codex" +} + +function Confirm-And-Install { + param([hashtable]$Context) + + Write-Host "正在执行全局安装/修复..." -ForegroundColor Green + $managerArgs = @("install") + while ($true) { + $result = Invoke-ManagerJson -Context $Context -ManagerArgs $managerArgs + if ($result.ExitCode -ne 2) { + break + } + + if ($result.Payload.confirmation_kind -eq "replace_related") { + Write-Host "" + Write-Host "检测到旧的相关 MCP 条目，需要你明确确认后才能替换。" -ForegroundColor Yellow + if ($result.Payload.related_entries) { + foreach ($entry in $result.Payload.related_entries) { + Write-Host "" + Write-Host "旧条目：" -ForegroundColor Yellow + Write-Host $entry.header + Write-Host $entry.text + } + } + Write-Host "" + Write-Host "新条目将注册为：" -ForegroundColor Green + $desired = $result.Payload.desired_server + Write-Host "名称：$($desired.name)" + Write-Host "命令：$($desired.command)" + Write-Host "参数：$([string]::Join(' ', $desired.args))" + Write-Host "" + $confirm = Read-Host "输入 YES 才继续替换旧条目" + if ($confirm -ne "YES") { + Write-Host "已取消替换。" -ForegroundColor Yellow + Pause-Continue + return + } + if ("--replace-related" -notin $managerArgs) { + $managerArgs += "--replace-related" + } + continue + } + + if ($result.Payload.confirmation_kind -eq "project_upgrades") { + Write-Host "" + Write-Host "检测到已登记公开版项目需要升级。" -ForegroundColor Yellow + Write-Host "请先关闭相关 Codex App / Codex CLI / 其他会话，再继续。" -ForegroundColor Yellow + Write-Host "" + Write-Host "升级摘要：" -ForegroundColor Green + Write-Host "- 需要升级：$($result.Payload.project_upgrade_summary.upgrade_needed)" + Write-Host "- 离线/缺失：$($result.Payload.project_upgrade_summary.offline_missing)" + Write-Host "- 标记缺失：$($result.Payload.project_upgrade_summary.marker_missing)" + Write-Host "- 已是最新：$($result.Payload.project_upgrade_summary.already_current)" + if ($result.Payload.project_upgrade_candidates) { + Write-Host "" + Write-Host "将升级这些项目：" -ForegroundColor Green + foreach ($item in $result.Payload.project_upgrade_candidates) { + Write-Host "- $($item.project_root) (当前版本 $($item.current_adapter_version))" + } + } + Write-Host "" + $confirm = Read-Host "确认你已经关闭相关会话，并输入 YES 继续" + if ($confirm -ne "YES") { + Write-Host "已取消升级。" -ForegroundColor Yellow + Pause-Continue + return + } + if ("--confirm-project-upgrades" -notin $managerArgs) { + $managerArgs += "--confirm-project-upgrades" + } + continue + } + + break + } + + Write-Host "" + Write-Host $result.Output + Write-Host "" + if ($result.ExitCode -eq 0) { + Write-Host "安装/修复完成。" -ForegroundColor Green + if ($result.Payload.project_upgrade_batch) { + $summary = $result.Payload.project_upgrade_batch.summary + Write-Host "项目升级摘要：" -ForegroundColor Green + Write-Host "- 成功升级：$($summary.upgraded)" + Write-Host "- 已是最新：$($summary.already_current)" + Write-Host "- 离线跳过：$($summary.offline_missing)" + Write-Host "- 标记缺失：$($summary.marker_missing)" + Write-Host "- 升级失败：$($summary.upgrade_failed)" + } + Write-Host "请现在重启 Codex App；Codex CLI 也建议重新打开。" -ForegroundColor Green + } else { + Write-Host "安装/修复失败，请先看上面的 JSON 结果。" -ForegroundColor Red + } + Pause-Continue +} + +function Show-DoctorStatus { + param([hashtable]$Context) + + $snapshot = Get-StartupSnapshot -Context $Context + $status = $snapshot.Status + $firstRun = $snapshot.FirstRun + + Write-Host "Codex 状态摘要" -ForegroundColor Green + Write-Host $status.Output + Write-Host "" + Write-Host "原版 first-run 摘要" -ForegroundColor Green + Write-Host $firstRun.Output + Pause-Continue +} + +function Invoke-FullCheck { + param([hashtable]$Context) + + Write-Host "即将运行原版 START_HERE 首检流程..." -ForegroundColor Green + Write-Host "这会生成原版报告，并真实执行 MCP smoke check。" -ForegroundColor Yellow + Write-Host "" + & powershell.exe -NoProfile -ExecutionPolicy Bypass -File $Context.StartHereScript + $exitCode = $LASTEXITCODE + Write-Host "" + if ($exitCode -eq 0) { + Write-Host "原版完整首检已完成。" -ForegroundColor Green + } else { + Write-Host "原版完整首检返回非零退出码：$exitCode" -ForegroundColor Yellow + } + Pause-Continue +} + +function Read-ProjectPath { + $projectPath = Read-Host "请输入目标项目完整路径" + return $projectPath.Trim() +} + +function Show-PreviewAndConfirm { + param( + [hashtable]$Context, + [string[]]$CommandArgs, + [string]$ActionLabel + ) + + $result = Invoke-ManagerJson -Context $Context -ManagerArgs $CommandArgs + if ($result.ExitCode -eq 2) { + $preview = $result.Payload.preview + Write-Host "" + Write-Host "$ActionLabel 预览" -ForegroundColor Yellow + Write-Host "项目路径：$($preview.project_root)" + Write-Host "将执行改动：" + foreach ($item in $preview.changes) { + Write-Host "- $item" + } + Write-Host "" + Write-Host "精确确认短语：" -ForegroundColor Green + Write-Host $result.Payload.confirmation_text + $confirmText = Read-Host "请原样输入确认短语；留空则取消" + if ([string]::IsNullOrWhiteSpace($confirmText)) { + Write-Host "已取消。" -ForegroundColor Yellow + Pause-Continue + return + } + $retryArgs = @($CommandArgs + @("--confirm-text", $confirmText)) + $result = Invoke-ManagerJson -Context $Context -ManagerArgs $retryArgs + } + + Write-Host "" + Write-Host $result.Output + if ($result.ExitCode -eq 0) { + Write-Host "$ActionLabel 完成。" -ForegroundColor Green + } else { + Write-Host "$ActionLabel 失败。" -ForegroundColor Red + } + Pause-Continue +} + +function Show-ProjectMenu { + param([hashtable]$Context) + + while ($true) { + Show-Title + Write-Host "项目管理" -ForegroundColor Green + Write-Host "1. 绑定/启用 指定项目" + Write-Host "2. 查看 指定项目状态" + Write-Host "3. 修复 指定项目接入" + Write-Host "4. 解绑/停用 指定项目" + Write-Host "0. 返回上一级" + Write-Host "" + $choice = Read-Host "请选择" + switch ($choice) { + "1" { + $path = Read-ProjectPath + if ($path) { + Show-PreviewAndConfirm -Context $Context -CommandArgs @("bind-project", $path) -ActionLabel "项目绑定" + } + } + "2" { + $path = Read-ProjectPath + if ($path) { + $result = Invoke-ManagerJson -Context $Context -ManagerArgs @("project-status", $path) + Write-Host "" + Write-Host $result.Output + Pause-Continue + } + } + "3" { + $path = Read-ProjectPath + if ($path) { + Show-PreviewAndConfirm -Context $Context -CommandArgs @("repair-project", $path) -ActionLabel "项目修复" + } + } + "4" { + $path = Read-ProjectPath + if ($path) { + Show-PreviewAndConfirm -Context $Context -CommandArgs @("unbind-project", $path) -ActionLabel "项目解绑" + } + } + "0" { return } + default { + Write-Host "无效选项。" -ForegroundColor Yellow + Start-Sleep -Milliseconds 700 + } + } + } +} + +function Show-BackupMenu { + param([hashtable]$Context) + + while ($true) { + Show-Title + Write-Host "备份 / 恢复" -ForegroundColor Green + Write-Host "1. 查看升级批次" + Write-Host "2. 按升级批次恢复单个项目" + Write-Host "3. 查看底层备份列表" + Write-Host "4. 按 backup_id 直接恢复" + Write-Host "0. 返回上一级" + Write-Host "" + $choice = Read-Host "请选择" + switch ($choice) { + "1" { + $result = Invoke-ManagerJson -Context $Context -ManagerArgs @("upgrade-batch-list") + Write-Host "" + Write-Host $result.Output + Pause-Continue + } + "2" { + $batchId = Read-Host "请输入升级批次 batch_id" + if ([string]::IsNullOrWhiteSpace($batchId)) { + continue + } + $projectId = Read-Host "请输入该批次里的 project_id" + if ([string]::IsNullOrWhiteSpace($projectId)) { + continue + } + $confirm = Read-Host "输入 RESTORE $batchId $projectId 才继续恢复" + if ($confirm -ne "RESTORE $batchId $projectId") { + Write-Host "已取消恢复。" -ForegroundColor Yellow + Pause-Continue + continue + } + $result = Invoke-ManagerJson -Context $Context -ManagerArgs @("upgrade-batch-restore", $batchId, $projectId) + Write-Host "" + Write-Host $result.Output + if ($result.ExitCode -eq 0) { + Write-Host "项目恢复完成。" -ForegroundColor Green + } else { + Write-Host "项目恢复失败。" -ForegroundColor Red + } + Pause-Continue + } + "3" { + $result = Invoke-ManagerJson -Context $Context -ManagerArgs @("backup-list") + Write-Host "" + Write-Host $result.Output + Pause-Continue + } + "4" { + $backupId = Read-Host "请输入要恢复的 backup_id" + if ([string]::IsNullOrWhiteSpace($backupId)) { + continue + } + $confirm = Read-Host "输入 RESTORE $backupId 才继续恢复" + if ($confirm -ne "RESTORE $backupId") { + Write-Host "已取消恢复。" -ForegroundColor Yellow + Pause-Continue + continue + } + $result = Invoke-ManagerJson -Context $Context -ManagerArgs @("backup-restore", $backupId) + Write-Host "" + Write-Host $result.Output + if ($result.ExitCode -eq 0) { + Write-Host "恢复完成。" -ForegroundColor Green + } else { + Write-Host "恢复失败。" -ForegroundColor Red + } + Pause-Continue + } + "0" { return } + default { + Write-Host "无效选项。" -ForegroundColor Yellow + Start-Sleep -Milliseconds 700 + } + } + } +} + +$context = Get-LocalContext + +if ($NonInteractive) { + Show-Title + Show-StartupSummary -Context $context + Show-FirstUseGuide -Context $context + Show-MainMenu + return +} + +while ($true) { + Show-Title + Show-StartupSummary -Context $context + Show-FirstUseGuide -Context $context + Show-MainMenu + $mainChoice = Read-Host "请选择" + switch ($mainChoice) { + "1" { Confirm-And-Install -Context $context } + "2" { Show-DoctorStatus -Context $context } + "3" { Invoke-FullCheck -Context $context } + "4" { Show-ProjectMenu -Context $context } + "5" { Show-BackupMenu -Context $context } + "6" { + Show-Title + Show-UsageGuide -Context $context + Pause-Continue + } + "0" { break } + default { + Write-Host "无效选项。" -ForegroundColor Yellow + Start-Sleep -Milliseconds 700 + } + } +} + diff --git a/scripts/codex/launcher_entry.ps1 b/scripts/codex/launcher_entry.ps1 new file mode 100644 index 0000000..5f57391 --- /dev/null +++ b/scripts/codex/launcher_entry.ps1 @@ -0,0 +1,105 @@ +param() + +$ErrorActionPreference = "Stop" +[Console]::InputEncoding = [System.Text.UTF8Encoding]::new() +[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new() +$OutputEncoding = [Console]::OutputEncoding + +$repoRoot = Split-Path -Parent (Split-Path -Parent $PSScriptRoot) +$codexEntry = Join-Path $PSScriptRoot "codex_entry.ps1" +$startHere = Join-Path $repoRoot "scripts\misc\start_here.ps1" + +function Pause-Continue { + Write-Host "" + Read-Host "按回车继续" +} + +function Show-Title { + Clear-Host + Write-Host "============================================================" -ForegroundColor Cyan + Write-Host " open-reverselab 启动工作台" -ForegroundColor Cyan + Write-Host "============================================================" -ForegroundColor Cyan + Write-Host "" +} + +function Show-Intro { + Write-Host "你现在只需要记这三个入口：" -ForegroundColor Yellow + Write-Host "- CODEX.bat：Codex 专用入口。" + Write-Host "- LAUNCHER.bat：总工作台入口。" + Write-Host "- START_HERE.bat：原作者原版入口，保持不动。" + Write-Host "" +} + +function Show-Menu { + Write-Host "主菜单" -ForegroundColor Green + Write-Host "1. Codex" + Write-Host " 说明：进入 Codex 专用工作台，只处理 Codex 的安装、检查、项目管理。" + Write-Host "2. Claude" + Write-Host " 说明：运行原版 START_HERE 首检流程，适合 Claude Code。" + Write-Host "3. 其他 CLI" + Write-Host " 说明：运行原版 START_HERE 首检流程，适合 Opencode / 其他兼容 CLI。" + Write-Host "4. 全部" + Write-Host " 说明：先跑原版 START_HERE 首检，再进入 Codex 工作台。" + Write-Host "5. 使用说明" + Write-Host " 说明：告诉你不同入口分别在什么场景使用。" + Write-Host "0. 退出" + Write-Host "" +} + +function Invoke-StartHere { + Write-Host "正在运行原版 START_HERE 流程..." -ForegroundColor Green + Write-Host "" + & powershell.exe -NoProfile -ExecutionPolicy Bypass -File $startHere + $exitCode = $LASTEXITCODE + Write-Host "" + if ($exitCode -eq 0) { + Write-Host "原版 START_HERE 流程已完成。" -ForegroundColor Green + } else { + Write-Host "原版 START_HERE 返回非零退出码：$exitCode" -ForegroundColor Yellow + } + Pause-Continue +} + +function Show-Usage { + Write-Host "入口分工" -ForegroundColor Green + Write-Host "- 如果你现在只做 Codex：直接用 CODEX.bat。" + Write-Host "- 如果你想通过一个总菜单来分流：用 LAUNCHER.bat。" + Write-Host "- 如果你只想保留原作者原版流程：用 START_HERE.bat。" + Write-Host "" + Write-Host "推荐习惯" -ForegroundColor Yellow + Write-Host "- Codex 用户：以后优先双击 CODEX.bat。" + Write-Host "- Claude / 其他 CLI 用户：优先走 START_HERE.bat 或 LAUNCHER.bat 里的对应入口。" + Write-Host "- 需要同时核对原版首检和 Codex 接入：选 4. 全部。" + Pause-Continue +} + +while ($true) { + Show-Title + Show-Intro + Show-Menu + $choice = Read-Host "请选择" + switch ($choice) { + "1" { + & powershell.exe -NoProfile -ExecutionPolicy Bypass -File $codexEntry + } + "2" { Invoke-StartHere } + "3" { Invoke-StartHere } + "4" { + & powershell.exe -NoProfile -ExecutionPolicy Bypass -File $startHere + Write-Host "" + Write-Host "接下来进入 Codex 工作台..." -ForegroundColor Green + Start-Sleep -Milliseconds 700 + & powershell.exe -NoProfile -ExecutionPolicy Bypass -File $codexEntry + } + "5" { + Show-Title + Show-Usage + } + "0" { break } + default { + Write-Host "无效选项。" -ForegroundColor Yellow + Start-Sleep -Milliseconds 700 + } + } +} + diff --git a/scripts/codex/python/open_reverselab_codex_manager.py b/scripts/codex/python/open_reverselab_codex_manager.py new file mode 100644 index 0000000..25100dc --- /dev/null +++ b/scripts/codex/python/open_reverselab_codex_manager.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +import open_reverselab_codex_support as support + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") +if hasattr(sys.stderr, "reconfigure"): + sys.stderr.reconfigure(encoding="utf-8") + + +def print_json(payload: dict) -> int: + print(json.dumps(payload, ensure_ascii=False, indent=2)) + if payload.get("overall") == "NEEDS_CONFIRMATION" or payload.get("requires_confirmation"): + return 2 + return 0 if payload.get("overall") not in {"FAIL"} and payload.get("error") is None else 1 + + +def resolve_project_path(value: str) -> Path: + return support.normalize_project_root(value) + + +def command_install(args: argparse.Namespace) -> int: + payload = support.install_or_repair_global_config( + replace_related=args.replace_related, + confirm_project_upgrades=args.confirm_project_upgrades, + ) + return print_json(payload) + + +def command_uninstall(args: argparse.Namespace) -> int: + payload = support.uninstall_global_config() + return print_json(payload) + + +def command_status(args: argparse.Namespace) -> int: + payload = support.codex_status_summary() + return print_json(payload) + + +def command_bind(args: argparse.Namespace) -> int: + payload = support.enable_project_mode(args.project_root, confirm=args.confirm_text) + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 0 if payload.get("ok") else 2 if payload.get("requires_confirmation") else 1 + + +def command_unbind(args: argparse.Namespace) -> int: + payload = support.disable_project_mode(args.project_root, confirm=args.confirm_text) + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 0 if payload.get("ok") else 2 if payload.get("requires_confirmation") else 1 + + +def command_project_status(args: argparse.Namespace) -> int: + payload = support.project_mode_status(args.project_root) + return print_json(payload) + + +def command_project_repair(args: argparse.Namespace) -> int: + payload = support.repair_project_mode(args.project_root, confirm=args.confirm_text) + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 0 if payload.get("ok") else 2 if payload.get("requires_confirmation") else 1 + + +def command_backup_list(args: argparse.Namespace) -> int: + payload = { + "schema": 1, + "generated_at": support.utc_now_iso(), + "backups": support.list_backups(), + } + return print_json(payload) + + +def command_backup_restore(args: argparse.Namespace) -> int: + payload = support.restore_backup(args.backup_id) + return print_json(payload) + + +def command_upgrade_batch_list(args: argparse.Namespace) -> int: + payload = { + "schema": 1, + "adapter_schema": support.ADAPTER_SCHEMA, + "adapter_version": support.ADAPTER_VERSION, + "generated_at": support.utc_now_iso(), + "batches": support.list_upgrade_batches(), + } + return print_json(payload) + + +def command_upgrade_batch_restore(args: argparse.Namespace) -> int: + payload = support.restore_upgrade_batch_project(args.batch_id, args.project_id) + return print_json(payload) + + +def command_doctor(args: argparse.Namespace) -> int: + payload = support.doctor() + return print_json(payload) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="open-reverselab Codex local manager") + sub = parser.add_subparsers(dest="command", required=True) + + install = sub.add_parser("install", help="Install or repair global Codex integration") + install.add_argument("--replace-related", action="store_true", help="Replace older related MCP entries that match this repo by path evidence") + install.add_argument("--confirm-project-upgrades", action="store_true", help="Confirm batch upgrade of registered public-version projects") + install.set_defaults(func=command_install) + + uninstall = sub.add_parser("uninstall", help="Remove global Codex integration") + uninstall.set_defaults(func=command_uninstall) + + status = sub.add_parser("status", help="Show global adapter status") + status.set_defaults(func=command_status) + + bind = sub.add_parser("bind-project", help="Bind another project to open-reverselab Codex mode") + bind.add_argument("project_root") + bind.add_argument("--confirm-text", default="") + bind.set_defaults(func=command_bind) + + unbind = sub.add_parser("unbind-project", help="Unbind a project from open-reverselab Codex mode") + unbind.add_argument("project_root") + unbind.add_argument("--confirm-text", default="") + unbind.set_defaults(func=command_unbind) + + project_status = sub.add_parser("project-status", help="Show project binding status") + project_status.add_argument("project_root") + project_status.set_defaults(func=command_project_status) + + repair = sub.add_parser("repair-project", help="Repair a bound project") + repair.add_argument("project_root") + repair.add_argument("--confirm-text", default="") + repair.set_defaults(func=command_project_repair) + + backup_list = sub.add_parser("backup-list", help="List local adapter backups") + backup_list.set_defaults(func=command_backup_list) + + backup_restore = sub.add_parser("backup-restore", help="Restore a local adapter backup") + backup_restore.add_argument("backup_id") + backup_restore.set_defaults(func=command_backup_restore) + + upgrade_batch_list = sub.add_parser("upgrade-batch-list", help="List recorded project upgrade batches") + upgrade_batch_list.set_defaults(func=command_upgrade_batch_list) + + upgrade_batch_restore = sub.add_parser("upgrade-batch-restore", help="Restore one project from an upgrade batch") + upgrade_batch_restore.add_argument("batch_id") + upgrade_batch_restore.add_argument("project_id") + upgrade_batch_restore.set_defaults(func=command_upgrade_batch_restore) + + doctor = sub.add_parser("doctor", help="Verify wrapper and adapter status") + doctor.set_defaults(func=command_doctor) + + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/codex/python/open_reverselab_codex_mcp.py b/scripts/codex/python/open_reverselab_codex_mcp.py new file mode 100644 index 0000000..58214a3 --- /dev/null +++ b/scripts/codex/python/open_reverselab_codex_mcp.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import argparse +import importlib +import inspect +import json +import os +import sys +from pathlib import Path +from typing import Any + +from mcp.server.fastmcp import Context, FastMCP + +import open_reverselab_codex_support as support + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") +if hasattr(sys.stderr, "reconfigure"): + sys.stderr.reconfigure(encoding="utf-8") + + +def _ensure_upstream_import_path() -> None: + module_dir = str(support.upstream_module_dir()) + if module_dir not in sys.path: + sys.path.insert(0, module_dir) + + +def _load_upstream_module(): + _ensure_upstream_import_path() + return importlib.import_module("reverse_lab_tools_mcp") + + +def _load_upstream_tools() -> dict[str, Any]: + upstream = _load_upstream_module() + return upstream.mcp._tool_manager._tools + + +async def _context_roots(ctx: Context | None) -> list[str]: + if ctx is None: + return [] + try: + result = await ctx.session.list_roots() + except Exception: + return [] + return [root.uri for root in getattr(result, "roots", [])] + + +def _selected_project_root(explicit_project_root: str, roots: list[str]) -> Path: + return support.select_project_root(explicit_project_root, roots) + + +def _wrapper_description(description: str) -> str: + description = description.strip() + extra = "Codex adapter: optional project_root routes generated artifacts to the current project." + if description: + return f"{description}\n\n{extra}" + return extra + + +def _make_upstream_wrapper(tool_name: str, tool_obj: Any): + async def wrapper(*args, ctx: Context, project_root: str = "", **kwargs): + roots = await _context_roots(ctx) + target_project = _selected_project_root(project_root, roots) + result = support.invoke_upstream_tool(tool_name, kwargs, target_project) + if isinstance(result, dict): + result.setdefault("project_root_used", str(target_project)) + return result + + wrapper.__name__ = tool_name + wrapper.__doc__ = _wrapper_description(tool_obj.description or "") + original_signature = inspect.signature(tool_obj.fn) + parameters = list(original_signature.parameters.values()) + parameters.append( + inspect.Parameter( + "project_root", + inspect.Parameter.KEYWORD_ONLY, + default="", + annotation=str, + ) + ) + parameters.append( + inspect.Parameter( + "ctx", + inspect.Parameter.KEYWORD_ONLY, + annotation=Context, + ) + ) + wrapper.__signature__ = original_signature.replace(parameters=parameters) + return wrapper + + +def _run_upstream_invocation(tool_name: str, arguments_json: str, project_root: str = "") -> int: + args = json.loads(arguments_json or "{}") + if not isinstance(args, dict): + raise ValueError("arguments_json must decode to an object") + + target_root = support.select_project_root(project_root or support.repo_root(), []) + env = support.wrapper_environment(target_root) + os.environ.update(env) + upstream = _load_upstream_module() + fn = getattr(upstream, tool_name, None) + if fn is None: + raise AttributeError(f"unknown upstream tool: {tool_name}") + result = fn(**args) + print(json.dumps(result, ensure_ascii=False)) + return 0 + + +def _list_upstream_tools() -> int: + tools = sorted(_load_upstream_tools().keys()) + print(json.dumps({"tools": tools}, ensure_ascii=False)) + return 0 + + +def _doctor() -> int: + payload = support.doctor() + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 0 + + +def build_server() -> FastMCP: + server = FastMCP(support.SERVER_NAME) + + @server.tool( + name="codex_enable_project_mode", + description="Enable open-reverselab Codex mode for the current or specified project. First activation previews changes and requires an exact confirmation text before writing files.", + ) + async def codex_enable_project_mode(project_root: str = "", confirm_text: str = "", ctx: Context | None = None) -> dict[str, Any]: + roots = await _context_roots(ctx) + target_root = _selected_project_root(project_root, roots) + return support.enable_project_mode(target_root, confirm=confirm_text) + + @server.tool( + name="codex_disable_project_mode", + description="Disable open-reverselab Codex mode for the current or specified project. This removes only the integration state and leaves analysis artifacts in place.", + ) + async def codex_disable_project_mode(project_root: str = "", confirm_text: str = "", ctx: Context | None = None) -> dict[str, Any]: + roots = await _context_roots(ctx) + target_root = _selected_project_root(project_root, roots) + return support.disable_project_mode(target_root, confirm=confirm_text) + + @server.tool( + name="codex_project_mode_status", + description="Show whether the current or specified project is bound to open-reverselab Codex mode and whether the binding is healthy.", + ) + async def codex_project_mode_status(project_root: str = "", ctx: Context | None = None) -> dict[str, Any]: + roots = await _context_roots(ctx) + target_root = _selected_project_root(project_root, roots) + return support.project_mode_status(target_root) + + @server.tool( + name="codex_repair_project_mode", + description="Repair a bound project by recreating the managed marker files, AGENTS block, and namespaced directories. Requires the same confirmation text as enable.", + ) + async def codex_repair_project_mode(project_root: str = "", confirm_text: str = "", ctx: Context | None = None) -> dict[str, Any]: + roots = await _context_roots(ctx) + target_root = _selected_project_root(project_root, roots) + return support.repair_project_mode(target_root, confirm=confirm_text) + + for tool_name, tool_obj in sorted(_load_upstream_tools().items()): + server.add_tool( + _make_upstream_wrapper(tool_name, tool_obj), + name=tool_name, + description=_wrapper_description(tool_obj.description or ""), + ) + + return server + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="open-reverselab Codex MCP wrapper") + parser.add_argument("--invoke-upstream", default="", help="Invoke one upstream tool and exit") + parser.add_argument("--arguments-json", default="{}", help="JSON object with upstream tool arguments") + parser.add_argument("--project-root", default="", help="Project root for routed output") + parser.add_argument("--list-upstream-tools", action="store_true", help="List upstream tool names and exit") + parser.add_argument("--doctor", action="store_true", help="Run local wrapper diagnostics and exit") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + if args.list_upstream_tools: + return _list_upstream_tools() + if args.doctor: + return _doctor() + if args.invoke_upstream: + return _run_upstream_invocation(args.invoke_upstream, args.arguments_json, args.project_root) + build_server().run() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/codex/python/open_reverselab_codex_support.py b/scripts/codex/python/open_reverselab_codex_support.py new file mode 100644 index 0000000..9348d74 --- /dev/null +++ b/scripts/codex/python/open_reverselab_codex_support.py @@ -0,0 +1,1817 @@ +from __future__ import annotations + +import json +import os +import re +import shutil +import subprocess +import sys +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from urllib.parse import urlparse, unquote +from urllib.request import url2pathname + + +SERVER_NAME = "open_reverselab_codex" +OUTPUT_NAMESPACE = "open-reverselab" +ADAPTER_NAME = "open-reverselab-codex" +ADAPTER_SCHEMA = 1 +ADAPTER_VERSION = "1.0.0" +LOCAL_RUNTIME_DIR = ".open-reverselab-local" +PROJECT_MARKER_DIR = ".open-reverselab-codex" +PROJECT_MARKER_FILE = "project.json" +PROJECT_README = "README.md" +PROJECT_QUICK_START = "QUICK_START.md" +PROJECT_TASK_TEMPLATE = "TASK_TEMPLATE.md" +PROJECT_CODEX_DIR = ".codex" +PROJECT_CODEX_CONFIG = "config.toml" +PROJECT_CODEX_MANIFEST = "open-reverselab.config.toml" +PROJECT_CODEX_CTF_CONFIG = "open-reverselab.ctf.config.toml" +PROJECT_CODEX_PROMPT = "open-reverselab.ctf_optimized.md" +GLOBAL_BLOCK_BEGIN = "[OPEN-REVERSELAB-CODEX MANAGED BLOCK BEGIN]" +GLOBAL_BLOCK_END = "[OPEN-REVERSELAB-CODEX MANAGED BLOCK END]" +PROJECT_BLOCK_BEGIN = "" +PROJECT_BLOCK_END = "" +PROJECT_CODEX_BLOCK_BEGIN = "# OPEN-REVERSELAB-CODEX PROJECT CONFIG BEGIN" +PROJECT_CODEX_BLOCK_END = "# OPEN-REVERSELAB-CODEX PROJECT CONFIG END" +MANAGED_IGNORE_HEADER = "# open-reverselab Codex managed ignore entries" +MANAGED_IGNORE_ENTRIES = [ + "/.open-reverselab-codex/", + "/.codex/open-reverselab.config.toml", + "/.codex/open-reverselab.ctf.config.toml", + "/.codex/open-reverselab.ctf_optimized.md", + "/notes/open-reverselab/", + "/reports/open-reverselab/", + "/exports/open-reverselab/", + "/patches/open-reverselab/", + "/projects/open-reverselab/", + "/samples/open-reverselab/", + "/cases/open-reverselab/", +] +ENABLE_CONFIRM_PREFIX = "确认启用" +DISABLE_CONFIRM_PREFIX = "确认停用" +PROJECT_CODEX_KEYS = ( + "model_instructions_file", + "approval_policy", + "sandbox_mode", +) +PROJECT_CODEX_PROMPT_TEXT = """# CTF Optimized Agent Prompt + +你当前在一个已经接入 open-reverselab Codex 模式的项目中工作。 +默认目标是授权 CTF/lab 自动化，并把产物写回当前项目。 + +## Working Mode + +- 当前项目的 open-reverselab 接入状态由 `.open-reverselab-codex/project.json` 标记。 +- 优先使用 open_reverselab_codex MCP 工具推进逆向、CTF 和样本分析。 +- 产物默认写入当前项目的 `notes/open-reverselab`、`reports/open-reverselab`、`exports/open-reverselab`。 +- 仅在需要时再写入 `patches/open-reverselab`、`projects/open-reverselab`、`samples/open-reverselab`、`cases/open-reverselab`。 + +## Unattended Loop Protocol + +- 不等待人工审批；运行器层面的 approval/sandbox 配置已允许自动执行。 +- 24h Web CTF 使用 `/loop + ctf-24h-round + ai_manifest.json` 模式。 +- 单轮任务必须有界执行，输出 `STATUS: CONTINUE|DONE|EXHAUSTED`。 +- `cases//ai_manifest.json` 是恢复点；中断后读取它继续，不从头开始。 +- 每发现信号先读 `kb/ctf-website/techniques/attack-network.md`，再用 `scripts/ctf-website/kb_router.py ""` 路由技术文档。 +- 非 Python allowlist 动作标记为 `agent_required`，由下一轮 Agent 自动处理，不交给人。 + +## Evidence + +- 原始请求/响应、截图、flag 和日志只写入当前项目内的 case/export/report 目录。 +- 不把真实目标、凭据、Cookie、token、flag 或个人路径提交到公开仓库。 +""" + + +@dataclass(frozen=True) +class AdapterPaths: + repo_root: Path + local_root: Path + python_root: Path + state_root: Path + backups_root: Path + batch_root: Path + reports_root: Path + docs_root: Path + global_codex_root: Path + global_codex_config: Path + workspace_codex_config: Path + registry_file: Path + runtime_state_file: Path + install_report: Path + verify_report: Path + backup_report: Path + restore_report: Path + start_report: Path + menu_report: Path + local_manifest: Path + + +def utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def utc_now_iso() -> str: + return utc_now().isoformat().replace("+00:00", "Z") + + +def version_key(value: str) -> tuple[int, ...]: + parts = [int(item) for item in re.findall(r"\d+", value or "0")] + return tuple(parts or [0]) + + +def adapter_identity(mode: str = "standard_active") -> dict[str, Any]: + return { + "name": ADAPTER_NAME, + "schema": ADAPTER_SCHEMA, + "version": ADAPTER_VERSION, + "mode": mode, + } + + +def repo_root() -> Path: + return Path(__file__).resolve().parents[3] + + +def adapter_paths() -> AdapterPaths: + root = repo_root() + local_root = root / LOCAL_RUNTIME_DIR + state_root = local_root / "state" + reports_root = local_root / "reports" + docs_root = root / "docs" / "codex" + python_root = root / "scripts" / "codex" / "python" + global_codex_root = Path.home() / ".codex" + return AdapterPaths( + repo_root=root, + local_root=local_root, + python_root=python_root, + state_root=state_root, + backups_root=local_root / "backups", + batch_root=local_root / "upgrade-batches", + reports_root=reports_root, + docs_root=docs_root, + global_codex_root=global_codex_root, + global_codex_config=global_codex_root / "config.toml", + workspace_codex_config=root / ".codex" / "config.toml", + registry_file=state_root / "project_registry.json", + runtime_state_file=state_root / "adapter_state.json", + install_report=reports_root / "codex-install-report.json", + verify_report=reports_root / "codex-verify-report.json", + backup_report=reports_root / "codex-backup-report.json", + restore_report=reports_root / "codex-restore-report.json", + start_report=reports_root / "codex-start-report.json", + menu_report=reports_root / "codex-menu-report.json", + local_manifest=state_root / "manifest.json", + ) + + +def ensure_layout(paths: AdapterPaths | None = None) -> AdapterPaths: + paths = paths or adapter_paths() + for target in ( + paths.local_root, + paths.state_root, + paths.backups_root, + paths.batch_root, + paths.reports_root, + paths.docs_root, + ): + target.mkdir(parents=True, exist_ok=True) + return paths + + +def slug(text: str) -> str: + value = "".join(ch if ch.isalnum() or ch in "._-" else "-" for ch in text.strip()) + value = re.sub(r"-{2,}", "-", value).strip(".-") + return value or "default" + + +def sanitize_component(path: Path) -> str: + raw = str(path).replace(":", "").replace("\\", "_").replace("/", "_") + return slug(raw) + + +def read_text(path: Path, default: str = "") -> str: + if not path.exists(): + return default + return path.read_text(encoding="utf-8") + + +def write_text(path: Path, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8", newline="\n") + + +def read_json(path: Path, default: Any) -> Any: + if not path.exists(): + return default + return json.loads(path.read_text(encoding="utf-8")) + + +def write_json(path: Path, payload: Any) -> None: + write_text(path, json.dumps(payload, ensure_ascii=False, indent=2) + "\n") + + +def detect_python() -> str | None: + for name in ("python", "py"): + found = shutil.which(name) + if found: + return found + return None + + +def detect_uv() -> str | None: + return shutil.which("uv") + + +def wrapper_script_path() -> Path: + return adapter_paths().python_root / "open_reverselab_codex_mcp.py" + + +def manager_script_path() -> Path: + return adapter_paths().python_root / "open_reverselab_codex_manager.py" + + +def upstream_project_dir(paths: AdapterPaths | None = None) -> Path: + paths = paths or adapter_paths() + return paths.repo_root / "tools" / "skills" / "mcp" / "ReverseLabToolsMCP" + + +def upstream_module_dir(paths: AdapterPaths | None = None) -> Path: + return upstream_project_dir(paths) + + +def desired_server_config(paths: AdapterPaths | None = None, uv_path: str | None = None) -> dict[str, Any]: + paths = paths or adapter_paths() + uv_path = uv_path or detect_uv() or "uv" + wrapper = wrapper_script_path() + project_dir = upstream_project_dir(paths) + return { + "name": SERVER_NAME, + "command": uv_path, + "args": [ + "run", + "--project", + str(project_dir), + "python", + str(wrapper), + ], + "env": { + "OPEN_REVERSELAB_CODEX_REPO_ROOT": str(paths.repo_root), + }, + } + + +def toml_basic_string(value: str) -> str: + escaped = value.replace("\\", "\\\\").replace('"', '\\"') + return f'"{escaped}"' + + +def desired_server_block(paths: AdapterPaths | None = None, uv_path: str | None = None) -> str: + server = desired_server_config(paths, uv_path) + lines = [ + f"[mcp_servers.{SERVER_NAME}]", + f"command = {toml_basic_string(server['command'])}", + "args = [", + ] + for arg in server["args"]: + lines.append(f" {toml_basic_string(arg)},") + lines.append("]") + lines.append("") + lines.append(f"[mcp_servers.{SERVER_NAME}.env]") + for key in sorted(server["env"]): + lines.append(f"{toml_basic_string(key)} = {toml_basic_string(server['env'][key])}") + return "\n".join(lines).rstrip() + "\n" + + +@dataclass +class McpSection: + dotted_name: str + start: int + end: int + text: str + + +def find_table_sections(text: str) -> list[McpSection]: + matches = list(re.finditer(r"(?m)^\[(?P[^\]]+)\]\s*$", text)) + sections: list[McpSection] = [] + for index, match in enumerate(matches): + start = match.start() + end = matches[index + 1].start() if index + 1 < len(matches) else len(text) + sections.append( + McpSection( + dotted_name=match.group("name").strip(), + start=start, + end=end, + text=text[start:end], + ) + ) + return sections + + +def remove_mcp_server_sections(text: str, server_name: str) -> str: + sections = find_table_sections(text) + spans: list[tuple[int, int]] = [] + for section in sections: + if section.dotted_name == f"mcp_servers.{server_name}" or section.dotted_name.startswith( + f"mcp_servers.{server_name}." + ): + spans.append((section.start, section.end)) + if not spans: + return text + parts: list[str] = [] + cursor = 0 + for start, end in spans: + parts.append(text[cursor:start]) + cursor = end + parts.append(text[cursor:]) + result = "".join(parts) + return result.strip() + "\n" + + +def find_related_mcp_sections(text: str, paths: AdapterPaths | None = None) -> list[dict[str, Any]]: + paths = paths or adapter_paths() + repo_marker = str(paths.repo_root).lower() + old_markers = [ + repo_marker, + "reverse_lab_tools_mcp.py", + "start_codex.bat", + "install_codex_mcp.bat", + str(wrapper_script_path()).lower(), + ] + matches: list[dict[str, Any]] = [] + for section in find_table_sections(text): + if not section.dotted_name.startswith("mcp_servers."): + continue + lowered = section.text.lower() + if any(marker in lowered for marker in old_markers): + matches.append( + { + "name": section.dotted_name.split(".", 1)[1], + "header": section.dotted_name, + "text": section.text.strip(), + } + ) + return matches + + +def find_assignment_span(text: str, key: str) -> tuple[int, int, str] | None: + pattern = re.compile( + rf'(?ms)^(?P\s*{re.escape(key)}\s*=\s*)(?P"""|\'\'\')(?P.*?)(?P=quote)\s*$' + ) + match = pattern.search(text) + if not match: + return None + return match.start(), match.end(), match.group("body") + + +def developer_instruction_block() -> str: + return "\n".join( + [ + GLOBAL_BLOCK_BEGIN, + "This block is idle by default.", + "Only react when the user says one of these exact phrases:", + "- 启用 open-reverselab Codex 模式", + "- enable open-reverselab codex mode", + "- 停用 open-reverselab Codex 模式", + "- disable open-reverselab codex mode", + "- 查看 open-reverselab Codex 状态", + "- show open-reverselab codex status", + "", + "When the enable phrase appears:", + "- Call the MCP tool codex_enable_project_mode from the open_reverselab_codex server.", + "- If the current workspace already contains .open-reverselab-codex/project.json, treat the project as already bound.", + "- If codex_enable_project_mode returns requires_confirmation=true, ask the user to type the exact confirmation text and then call the same tool again with confirm_text.", + "", + "When the disable phrase appears:", + "- Call the MCP tool codex_disable_project_mode from the open_reverselab_codex server.", + "- If codex_disable_project_mode returns requires_confirmation=true, ask the user to type the exact confirmation text and then call the same tool again with confirm_text.", + "", + "When the status phrase appears:", + "- Call the MCP tool codex_project_mode_status from the open_reverselab_codex server.", + "", + "When the current project is bound to open-reverselab Codex mode:", + "- Prefer tools from the open_reverselab_codex MCP server for reverse engineering, CTF, and analysis automation.", + "- Keep project outputs inside the current project, not inside the open-reverselab repo.", + "- Write notes under notes/open-reverselab, reports under reports/open-reverselab, and exports under exports/open-reverselab.", + "- Create patches/projects/samples/cases under their open-reverselab namespaced subdirectories only when needed.", + GLOBAL_BLOCK_END, + ] + ).strip() + + +def upsert_developer_instructions(text: str, block: str | None = None) -> str: + block = block or developer_instruction_block() + assignment = find_assignment_span(text, "developer_instructions") + if assignment is not None: + start, end, body = assignment + if GLOBAL_BLOCK_BEGIN in body and GLOBAL_BLOCK_END in body: + updated_body = re.sub( + rf"{re.escape(GLOBAL_BLOCK_BEGIN)}.*?{re.escape(GLOBAL_BLOCK_END)}", + block, + body, + flags=re.S, + ) + else: + updated_body = body.rstrip() + "\n\n" + block + "\n" + replacement = f'developer_instructions = """\n{updated_body.rstrip()}\n"""' + return text[:start] + replacement + text[end:] + + header_match = re.search(r"(?m)^\[", text) + insert_at = header_match.start() if header_match else len(text) + insertion = f'developer_instructions = """\n{block}\n"""\n\n' + return text[:insert_at] + insertion + text[insert_at:] + + +def remove_developer_instruction_block(text: str) -> str: + assignment = find_assignment_span(text, "developer_instructions") + if assignment is None: + return text + start, end, body = assignment + if GLOBAL_BLOCK_BEGIN not in body or GLOBAL_BLOCK_END not in body: + return text + updated_body = re.sub( + rf"\n?{re.escape(GLOBAL_BLOCK_BEGIN)}.*?{re.escape(GLOBAL_BLOCK_END)}\n?", + "\n", + body, + flags=re.S, + ).strip() + if updated_body: + replacement = f'developer_instructions = """\n{updated_body}\n"""' + return text[:start] + replacement + text[end:] + prefix = text[:start] + suffix = text[end:] + merged = prefix.rstrip() + "\n" + suffix.lstrip() + return merged + + +def ensure_global_server_and_instructions(text: str, paths: AdapterPaths | None = None, uv_path: str | None = None) -> str: + text = remove_mcp_server_sections(text, SERVER_NAME) + text = upsert_developer_instructions(text) + block = desired_server_block(paths, uv_path) + stripped = text.rstrip() + "\n\n" + block + return stripped.rstrip() + "\n" + + +def remove_global_server_and_instructions(text: str) -> str: + text = remove_mcp_server_sections(text, SERVER_NAME) + text = remove_developer_instruction_block(text) + return text.rstrip() + "\n" + + +def current_global_status(paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = paths or adapter_paths() + config_text = read_text(paths.global_codex_config) + related = find_related_mcp_sections(config_text, paths) + desired = desired_server_config(paths) + return { + "global_config": str(paths.global_codex_config), + "config_exists": paths.global_codex_config.exists(), + "workspace_codex_config_exists": paths.workspace_codex_config.exists(), + "related_mcp_sections": related, + "desired_server": desired, + "developer_block_present": GLOBAL_BLOCK_BEGIN in config_text and GLOBAL_BLOCK_END in config_text, + "server_block_present": f"[mcp_servers.{SERVER_NAME}]" in config_text, + } + + +def file_snapshot(target: Path) -> dict[str, Any]: + if not target.exists(): + return {"path": str(target), "exists": False, "is_dir": False} + return { + "path": str(target), + "exists": True, + "is_dir": target.is_dir(), + "size": target.stat().st_size if target.is_file() else None, + } + + +def copy_path_for_backup(source: Path, destination: Path) -> None: + if source.is_dir(): + if destination.exists(): + shutil.rmtree(destination) + shutil.copytree(source, destination) + else: + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + + +def trim_backups(paths: AdapterPaths | None = None, keep: int = 10) -> None: + paths = paths or adapter_paths() + backups = sorted((item for item in paths.backups_root.iterdir() if item.is_dir()), key=lambda item: item.name, reverse=True) + for stale in backups[keep:]: + shutil.rmtree(stale, ignore_errors=True) + + +def create_backup( + action: str, + paths: AdapterPaths | None = None, + target_project_root: Path | None = None, + extra_files: list[Path] | None = None, + include_global_state: bool = True, +) -> dict[str, Any]: + paths = ensure_layout(paths) + timestamp = utc_now().strftime("%Y%m%d-%H%M%S-%f") + backup_id = f"{timestamp}-{slug(action)}" + backup_root = paths.backups_root / backup_id + files_root = backup_root / "files" + manifest = { + "schema": 1, + "backup_id": backup_id, + "action": action, + "created_at": utc_now_iso(), + "repo_root": str(paths.repo_root), + "target_project_root": str(target_project_root) if target_project_root else "", + "entries": [], + } + tracked: list[Path] = [] + if include_global_state: + tracked.extend( + [ + paths.global_codex_config, + paths.registry_file, + paths.runtime_state_file, + paths.local_manifest, + ] + ) + if target_project_root: + project_codex = project_codex_paths(target_project_root) + tracked.extend( + [ + target_project_root / "AGENTS.md", + target_project_root / ".gitignore", + target_project_root / PROJECT_MARKER_DIR, + project_codex["config"], + project_codex["manifest"], + project_codex["ctf_config"], + project_codex["prompt"], + ] + ) + if extra_files: + tracked.extend(extra_files) + + seen: set[str] = set() + for item in tracked: + key = str(item.resolve()) if item.exists() else str(item) + if key in seen: + continue + seen.add(key) + entry = { + "path": str(item), + "exists": item.exists(), + "is_dir": item.is_dir() if item.exists() else False, + } + if item.exists(): + rel_name = sanitize_component(item) + destination = files_root / rel_name + copy_path_for_backup(item, destination) + entry["backup_path"] = str(destination) + manifest["entries"].append(entry) + + write_json(backup_root / "manifest.json", manifest) + trim_backups(paths) + return manifest + + +def list_backups(paths: AdapterPaths | None = None) -> list[dict[str, Any]]: + paths = ensure_layout(paths) + manifests: list[dict[str, Any]] = [] + for folder in sorted((item for item in paths.backups_root.iterdir() if item.is_dir()), key=lambda item: item.name, reverse=True): + manifest_path = folder / "manifest.json" + if manifest_path.exists(): + manifests.append(read_json(manifest_path, {})) + return manifests + + +def restore_backup(backup_id: str, paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + backup_root = paths.backups_root / backup_id + manifest_path = backup_root / "manifest.json" + if not manifest_path.exists(): + raise FileNotFoundError(f"backup not found: {backup_id}") + + manifest = read_json(manifest_path, {}) + restore_manifest = create_backup(f"pre-restore-{backup_id}", paths=paths) + restored: list[dict[str, Any]] = [] + for entry in manifest.get("entries", []): + target = Path(entry["path"]) + exists = entry.get("exists", False) + backup_path = entry.get("backup_path", "") + if exists and backup_path: + source = Path(backup_path) + if target.exists(): + if target.is_dir(): + shutil.rmtree(target, ignore_errors=True) + else: + target.unlink() + copy_path_for_backup(source, target) + restored.append({"path": str(target), "action": "restored"}) + elif not exists and target.exists(): + if target.is_dir(): + shutil.rmtree(target, ignore_errors=True) + else: + target.unlink() + restored.append({"path": str(target), "action": "removed"}) + + payload = { + "schema": 1, + "backup_id": backup_id, + "restored_at": utc_now_iso(), + "pre_restore_backup": restore_manifest["backup_id"], + "restored": restored, + } + write_json(paths.restore_report, payload) + return payload + + +def list_upgrade_batches(paths: AdapterPaths | None = None) -> list[dict[str, Any]]: + paths = ensure_layout(paths) + manifests: list[dict[str, Any]] = [] + for manifest_path in sorted(paths.batch_root.glob("*.json"), reverse=True): + manifests.append(read_json(manifest_path, {})) + return manifests + + +def write_upgrade_batch_manifest( + project_results: list[dict[str, Any]], + global_backup_id: str = "", + paths: AdapterPaths | None = None, +) -> dict[str, Any]: + paths = ensure_layout(paths) + batch_id = f"{utc_now().strftime('%Y%m%d-%H%M%S-%f')}-upgrade-batch" + summary = { + "upgraded": sum(1 for item in project_results if item.get("result") == "upgraded"), + "already_current": sum(1 for item in project_results if item.get("result") == "already_current"), + "offline_missing": sum(1 for item in project_results if item.get("result") == "offline_missing"), + "upgrade_failed": sum(1 for item in project_results if item.get("result") == "upgrade_failed"), + "marker_missing": sum(1 for item in project_results if item.get("result") == "marker_missing"), + } + payload = { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "batch_id": batch_id, + "created_at": utc_now_iso(), + "global_backup_id": global_backup_id, + "summary": summary, + "projects": project_results, + } + write_json(paths.batch_root / f"{batch_id}.json", payload) + return payload + + +def restore_upgrade_batch_project(batch_id: str, project_id: str, paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + batch = read_json(paths.batch_root / f"{batch_id}.json", {}) + if not batch: + raise FileNotFoundError(f"upgrade batch not found: {batch_id}") + for project in batch.get("projects", []): + if project.get("project_id") == project_id: + backup_id = project.get("backup_id", "") + if not backup_id: + raise FileNotFoundError(f"project backup not found in batch: {project_id}") + restored = restore_backup(backup_id, paths) + restored["batch_id"] = batch_id + restored["project_id"] = project_id + return restored + raise FileNotFoundError(f"project not found in upgrade batch: {project_id}") + + +def default_registry() -> dict[str, Any]: + return { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "updated_at": "", + "projects": {}, + } + + +def load_registry(paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + registry = read_json(paths.registry_file, default_registry()) + if not registry: + registry = default_registry() + registry.setdefault("schema", 1) + registry.setdefault("adapter_schema", ADAPTER_SCHEMA) + registry.setdefault("adapter_version", ADAPTER_VERSION) + registry.setdefault("projects", {}) + registry.setdefault("updated_at", "") + return registry + + +def save_registry(registry: dict[str, Any], paths: AdapterPaths | None = None) -> None: + paths = ensure_layout(paths) + registry["adapter_schema"] = ADAPTER_SCHEMA + registry["adapter_version"] = ADAPTER_VERSION + registry["updated_at"] = utc_now_iso() + write_json(paths.registry_file, registry) + + +def project_marker_root(project_root: Path) -> Path: + return project_root / PROJECT_MARKER_DIR + + +def project_marker_path(project_root: Path) -> Path: + return project_marker_root(project_root) / PROJECT_MARKER_FILE + + +def namespaced_data_roots(project_root: Path) -> dict[str, Path]: + return { + "marker_root": project_marker_root(project_root), + "notes_root": project_root / "notes" / OUTPUT_NAMESPACE, + "reports_root": project_root / "reports" / OUTPUT_NAMESPACE, + "exports_root": project_root / "exports" / OUTPUT_NAMESPACE, + "patches_root": project_root / "patches" / OUTPUT_NAMESPACE, + "projects_root": project_root / "projects" / OUTPUT_NAMESPACE, + "samples_root": project_root / "samples" / OUTPUT_NAMESPACE, + "cases_root": project_root / "cases" / OUTPUT_NAMESPACE, + } + + +def project_codex_paths(project_root: Path) -> dict[str, Path]: + codex_root = project_root / PROJECT_CODEX_DIR + return { + "root": codex_root, + "config": codex_root / PROJECT_CODEX_CONFIG, + "manifest": codex_root / PROJECT_CODEX_MANIFEST, + "ctf_config": codex_root / PROJECT_CODEX_CTF_CONFIG, + "prompt": codex_root / PROJECT_CODEX_PROMPT, + } + + +def project_codex_manifest_text(project_root: Path) -> str: + return "\n".join( + [ + "# open-reverselab Codex project manifest", + "#", + "# This file is generated for documentation and recovery.", + "# Codex does not currently support include/import/extends for project config.toml,", + "# so the active runtime keys are mirrored into .codex/config.toml as a managed block.", + f'project_root = {toml_basic_string(str(project_root))}', + f'output_namespace = {toml_basic_string(OUTPUT_NAMESPACE)}', + f'prompt_file = {toml_basic_string(PROJECT_CODEX_PROMPT)}', + f'ctf_config_file = {toml_basic_string(PROJECT_CODEX_CTF_CONFIG)}', + ] + ) + "\n" + + +def project_codex_ctf_config_text() -> str: + return "\n".join( + [ + "# open-reverselab Codex CTF runtime template", + f'model_instructions_file = {toml_basic_string(PROJECT_CODEX_PROMPT)}', + 'approval_policy = "never"', + 'sandbox_mode = "danger-full-access"', + "", + ] + ) + + +def project_codex_managed_block() -> str: + return "\n".join( + [ + PROJECT_CODEX_BLOCK_BEGIN, + "# Managed by open-reverselab CODEX.bat / LAUNCHER.bat", + "# Source reference: .codex/open-reverselab.ctf.config.toml", + f'model_instructions_file = {toml_basic_string(PROJECT_CODEX_PROMPT)}', + 'approval_policy = "never"', + 'sandbox_mode = "danger-full-access"', + PROJECT_CODEX_BLOCK_END, + ] + ).strip() + + +def strip_project_codex_block(text: str) -> str: + if PROJECT_CODEX_BLOCK_BEGIN not in text or PROJECT_CODEX_BLOCK_END not in text: + return text + updated = re.sub( + rf"\n?{re.escape(PROJECT_CODEX_BLOCK_BEGIN)}.*?{re.escape(PROJECT_CODEX_BLOCK_END)}\n?", + "\n", + text, + flags=re.S, + ) + return updated.strip() + ("\n" if updated.strip() else "") + + +def capture_project_codex_restore_state(text: str) -> dict[str, str]: + cleaned = strip_project_codex_block(text) + restored: dict[str, str] = {} + for key in PROJECT_CODEX_KEYS: + match = re.search(rf"(?m)^\s*{re.escape(key)}\s*=.*$", cleaned) + if match: + restored[key] = match.group(0).strip() + return restored + + +def remove_project_codex_key_lines(text: str) -> str: + cleaned = strip_project_codex_block(text) + for key in PROJECT_CODEX_KEYS: + cleaned = re.sub(rf"(?m)^\s*{re.escape(key)}\s*=.*\n?", "", cleaned) + cleaned = re.sub(r"\n{3,}", "\n\n", cleaned).strip() + return cleaned + ("\n" if cleaned else "") + + +def ensure_project_codex_runtime(project_root: Path, restore_state: dict[str, str] | None = None) -> dict[str, Any]: + codex = project_codex_paths(project_root) + codex["root"].mkdir(parents=True, exist_ok=True) + existing_config = read_text(codex["config"]) + preserved = dict(restore_state or capture_project_codex_restore_state(existing_config)) + base = remove_project_codex_key_lines(existing_config) + block = project_codex_managed_block() + updated = (base.rstrip() + "\n\n" if base.strip() else "") + block + "\n" + write_text(codex["config"], updated) + write_text(codex["manifest"], project_codex_manifest_text(project_root)) + write_text(codex["ctf_config"], project_codex_ctf_config_text()) + write_text(codex["prompt"], PROJECT_CODEX_PROMPT_TEXT) + return { + "updated": True, + "paths": {name: str(path) for name, path in codex.items()}, + "restore_state": preserved, + } + + +def disable_project_codex_runtime(project_root: Path, restore_state: dict[str, str] | None = None) -> dict[str, Any]: + codex = project_codex_paths(project_root) + codex["root"].mkdir(parents=True, exist_ok=True) + existing_config = read_text(codex["config"]) + base = remove_project_codex_key_lines(existing_config) + restored_lines = [restore_state.get(key, "").strip() for key in PROJECT_CODEX_KEYS] if restore_state else [] + restored_lines = [line for line in restored_lines if line] + updated = base.rstrip() + if restored_lines: + updated = (updated + "\n\n" if updated else "") + "\n".join(restored_lines) + if updated: + updated += "\n" + else: + updated = "# open-reverselab Codex runtime files are retained after disable.\n" + write_text(codex["config"], updated) + if not codex["manifest"].exists(): + write_text(codex["manifest"], project_codex_manifest_text(project_root)) + if not codex["ctf_config"].exists(): + write_text(codex["ctf_config"], project_codex_ctf_config_text()) + if not codex["prompt"].exists(): + write_text(codex["prompt"], PROJECT_CODEX_PROMPT_TEXT) + return { + "updated": True, + "paths": {name: str(path) for name, path in codex.items()}, + } + + +def confirm_text(project_root: Path, action: str) -> str: + name = project_root.name.strip() or "当前项目" + prefix = ENABLE_CONFIRM_PREFIX if action == "enable" else DISABLE_CONFIRM_PREFIX + return f"{prefix} {name} open-reverselab Codex 模式" + + +def project_doc_readme(project_root: Path) -> str: + return "\n".join( + [ + "# open-reverselab Codex 接入说明", + "", + "这个目录由 open-reverselab 的 Codex 适配层生成。", + "", + "用途：", + "- 标记当前项目已经接入 open-reverselab Codex 模式。", + "- 保存当前项目的接入元数据和快速说明。", + "- 帮助后续会话继续把分析产物写回当前项目，而不是写回 open-reverselab 仓库。", + "", + "重要说明：", + "- 这里不是原版 open-reverselab 的 upstream 文件。", + "- 删除这个目录会导致项目接入状态丢失，可以通过 CODEX.bat 重新修复。", + "- 项目级 Codex 运行层位于 .codex/ 目录，停用时默认保留这些文件。", + "- 分析产物默认写入 notes/open-reverselab、reports/open-reverselab、exports/open-reverselab。", + "", + ] + ) + "\n" + + +def project_doc_quick_start(project_root: Path) -> str: + return "\n".join( + [ + "# QUICK START", + "", + "当前项目已经接入 open-reverselab Codex 模式。", + "", + "常用会话短语：", + "- 启用 open-reverselab Codex 模式", + "- 查看 open-reverselab Codex 状态", + "- 停用 open-reverselab Codex 模式", + "", + "常用产物目录：", + "- notes/open-reverselab", + "- reports/open-reverselab", + "- exports/open-reverselab", + "", + "项目级运行层：", + "- .codex/config.toml", + "- .codex/open-reverselab.config.toml", + "- .codex/open-reverselab.ctf.config.toml", + "- .codex/open-reverselab.ctf_optimized.md", + "", + "如果 Codex 没识别到当前接入状态：", + "- 先说“查看 open-reverselab Codex 状态”", + "- 如果状态异常，再用 CODEX.bat 的项目修复功能", + "", + ] + ) + "\n" + + +def project_doc_task_template(project_root: Path) -> str: + return "\n".join( + [ + "# TASK TEMPLATE", + "", + "参考模板：", + "", + "```text", + "启用 open-reverselab Codex 模式", + "这是我的 CTF / 逆向 / 样本分析任务。", + "目标范围是授权的实验环境或竞赛沙箱。", + "请优先使用 open_reverselab_codex MCP 工具推进，并把产物写到当前项目的 namespaced 目录。", + "```", + "", + ] + ) + "\n" + + +def project_agents_block(project_root: Path, project_id: str) -> str: + return "\n".join( + [ + PROJECT_BLOCK_BEGIN, + "## open-reverselab Codex 接入", + "", + f"- project_id: `{project_id}`", + "- 这个受管块由 open-reverselab 的 Codex 适配层写入。", + "- 当前项目已接入 open_reverselab_codex MCP。", + "- 分析产物写入当前项目，而不是写回 open-reverselab 仓库。", + "- 项目级运行层写入 `.codex/`，并在 `.codex/config.toml` 中插入受管块。", + "- 默认产物目录：`notes/open-reverselab/`、`reports/open-reverselab/`、`exports/open-reverselab/`。", + "- 启用短语：`启用 open-reverselab Codex 模式` / `enable open-reverselab codex mode`。", + "- 停用短语：`停用 open-reverselab Codex 模式` / `disable open-reverselab codex mode`。", + "- 状态短语：`查看 open-reverselab Codex 状态` / `show open-reverselab codex status`。", + PROJECT_BLOCK_END, + ] + ).strip() + + +def ensure_gitignore_block(project_root: Path) -> dict[str, Any]: + if not (project_root / ".git").exists(): + return {"updated": False, "reason": "not-a-git-project", "path": str(project_root / ".gitignore")} + + gitignore = project_root / ".gitignore" + existing = read_text(gitignore) + block = MANAGED_IGNORE_HEADER + "\n" + "\n".join(MANAGED_IGNORE_ENTRIES) + if MANAGED_IGNORE_HEADER in existing: + updated = re.sub( + rf"{re.escape(MANAGED_IGNORE_HEADER)}.*?(?=\n\n|\Z)", + block, + existing, + flags=re.S, + ).rstrip() + "\n" + else: + updated = existing.rstrip() + if updated: + updated += "\n\n" + updated += block + "\n" + write_text(gitignore, updated) + return {"updated": True, "path": str(gitignore)} + + +def remove_gitignore_block(project_root: Path) -> dict[str, Any]: + gitignore = project_root / ".gitignore" + if not gitignore.exists(): + return {"updated": False, "path": str(gitignore)} + existing = read_text(gitignore) + if MANAGED_IGNORE_HEADER not in existing: + return {"updated": False, "path": str(gitignore)} + updated = re.sub( + rf"\n?{re.escape(MANAGED_IGNORE_HEADER)}.*?(?=\n\n|\Z)", + "", + existing, + flags=re.S, + ).strip() + if updated: + write_text(gitignore, updated + "\n") + else: + gitignore.unlink() + return {"updated": True, "path": str(gitignore)} + + +def ensure_agents_block(project_root: Path, project_id: str) -> dict[str, Any]: + agents = project_root / "AGENTS.md" + block = project_agents_block(project_root, project_id) + existing = read_text(agents) + if existing: + if PROJECT_BLOCK_BEGIN in existing and PROJECT_BLOCK_END in existing: + updated = re.sub( + rf"{re.escape(PROJECT_BLOCK_BEGIN)}.*?{re.escape(PROJECT_BLOCK_END)}", + block, + existing, + flags=re.S, + ) + else: + updated = existing.rstrip() + "\n\n" + block + "\n" + else: + updated = "# AGENTS\n\n本文件包含当前项目的本地协作说明。\n\n" + block + "\n" + write_text(agents, updated) + return {"updated": True, "path": str(agents)} + + +def remove_agents_block(project_root: Path) -> dict[str, Any]: + agents = project_root / "AGENTS.md" + if not agents.exists(): + return {"updated": False, "path": str(agents)} + existing = read_text(agents) + if PROJECT_BLOCK_BEGIN not in existing or PROJECT_BLOCK_END not in existing: + return {"updated": False, "path": str(agents)} + updated = re.sub( + rf"\n?{re.escape(PROJECT_BLOCK_BEGIN)}.*?{re.escape(PROJECT_BLOCK_END)}\n?", + "\n", + existing, + flags=re.S, + ).strip() + if updated: + write_text(agents, updated + "\n") + else: + agents.unlink() + return {"updated": True, "path": str(agents)} + + +def normalize_project_root(project_root: str | Path) -> Path: + root = Path(project_root).expanduser().resolve() + if not root.exists(): + raise FileNotFoundError(f"project path does not exist: {root}") + if not root.is_dir(): + raise NotADirectoryError(f"project path is not a directory: {root}") + return root + + +def project_adapter_metadata(marker_data: dict[str, Any]) -> dict[str, Any]: + adapter = marker_data.get("adapter") + if isinstance(adapter, dict): + return { + "name": adapter.get("name") or ADAPTER_NAME, + "schema": int(adapter.get("schema") or 0), + "version": str(adapter.get("version") or "0"), + "mode": adapter.get("mode") or "standard_active", + } + return { + "name": ADAPTER_NAME, + "schema": int(marker_data.get("adapter_schema") or 0), + "version": str(marker_data.get("adapter_version") or "0"), + "mode": str(marker_data.get("project_mode") or "standard_active"), + } + + +def project_requires_upgrade(marker_data: dict[str, Any]) -> bool: + adapter = project_adapter_metadata(marker_data) + return ( + adapter["schema"] < ADAPTER_SCHEMA + or version_key(adapter["version"]) < version_key(ADAPTER_VERSION) + ) + + +def project_state(project_root: Path, paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + registry = load_registry(paths) + marker = project_marker_path(project_root) + marker_data = read_json(marker, {}) + data_roots = namespaced_data_roots(project_root) + codex_paths = project_codex_paths(project_root) + codex_config_text = read_text(codex_paths["config"]) + project_id = marker_data.get("project_id", "") + adapter = project_adapter_metadata(marker_data) + registry_entry = registry["projects"].get(project_id, {}) if project_id else {} + healthy = bool(marker.exists()) + if marker.exists(): + for key in ("notes_root", "reports_root", "exports_root"): + if not data_roots[key].exists(): + healthy = False + for key in ("config", "manifest", "ctf_config", "prompt"): + if not codex_paths[key].exists(): + healthy = False + if (project_root / "AGENTS.md").exists(): + healthy = healthy and PROJECT_BLOCK_BEGIN in read_text(project_root / "AGENTS.md") + healthy = healthy and PROJECT_CODEX_BLOCK_BEGIN in codex_config_text and PROJECT_CODEX_BLOCK_END in codex_config_text + return { + "project_root": str(project_root), + "project_id": project_id, + "bound": marker.exists(), + "healthy": healthy, + "project_mode": adapter["mode"], + "adapter_name": adapter["name"], + "adapter_schema": adapter["schema"], + "adapter_version": adapter["version"], + "upgrade_needed": bool(marker.exists()) and project_requires_upgrade(marker_data), + "marker_path": str(marker), + "registry_entry": registry_entry, + "paths": {name: str(path) for name, path in data_roots.items()}, + "codex_paths": {name: str(path) for name, path in codex_paths.items()}, + "has_agents_block": PROJECT_BLOCK_BEGIN in read_text(project_root / "AGENTS.md"), + "has_gitignore_block": MANAGED_IGNORE_HEADER in read_text(project_root / ".gitignore"), + "has_project_codex_block": PROJECT_CODEX_BLOCK_BEGIN in codex_config_text and PROJECT_CODEX_BLOCK_END in codex_config_text, + } + + +def build_project_plan(project_root: Path, action: str, paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + state = project_state(project_root, paths) + data_roots = namespaced_data_roots(project_root) + codex_paths = project_codex_paths(project_root) + project_id = state["project_id"] or str(uuid.uuid4()) + actions: list[str] = [] + if action in {"enable", "repair"}: + if not project_marker_root(project_root).exists(): + actions.append(f"创建目录 {project_marker_root(project_root)}") + if not codex_paths["root"].exists(): + actions.append(f"创建目录 {codex_paths['root']}") + for key in ("notes_root", "reports_root", "exports_root"): + if not data_roots[key].exists(): + actions.append(f"创建目录 {data_roots[key]}") + if not state["has_agents_block"]: + actions.append(f"更新 {project_root / 'AGENTS.md'}") + if (project_root / ".git").exists() and not state["has_gitignore_block"]: + actions.append(f"更新 {project_root / '.gitignore'}") + actions.extend( + [ + f"写入 {project_marker_root(project_root) / PROJECT_MARKER_FILE}", + f"写入 {project_marker_root(project_root) / PROJECT_README}", + f"写入 {project_marker_root(project_root) / PROJECT_QUICK_START}", + f"写入 {project_marker_root(project_root) / PROJECT_TASK_TEMPLATE}", + f"写入 {codex_paths['manifest']}", + f"写入 {codex_paths['ctf_config']}", + f"写入 {codex_paths['prompt']}", + f"更新 {codex_paths['config']} 中的 open-reverselab 受管块", + f"更新本地注册表 {paths.registry_file}", + ] + ) + elif action == "disable": + actions.extend( + [ + f"删除目录 {project_marker_root(project_root)}", + f"移除 {project_root / 'AGENTS.md'} 中的受管块", + f"移除 {project_root / '.gitignore'} 中的受管忽略项", + f"移除 {codex_paths['config']} 中的 open-reverselab 受管块并保留 .codex 文件", + f"移除本地注册表 {paths.registry_file} 中的项目绑定", + ] + ) + return { + "action": action, + "project_root": str(project_root), + "project_id": project_id, + "confirmation_text": confirm_text(project_root, "enable" if action != "disable" else "disable"), + "changes": actions, + "state": state, + } + + +def apply_project_binding(project_root: Path, project_id: str, paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + data_roots = namespaced_data_roots(project_root) + codex_runtime = ensure_project_codex_runtime(project_root) + existing_marker = read_json(project_marker_path(project_root), {}) + project_marker_root(project_root).mkdir(parents=True, exist_ok=True) + for key in ("notes_root", "reports_root", "exports_root"): + data_roots[key].mkdir(parents=True, exist_ok=True) + marker_payload = { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "adapter": adapter_identity(), + "project_id": project_id, + "project_root": str(project_root), + "repo_root": str(paths.repo_root), + "output_namespace": OUTPUT_NAMESPACE, + "bound_at": existing_marker.get("bound_at") or utc_now_iso(), + "updated_at": utc_now_iso(), + "project_codex": { + "restore_state": codex_runtime["restore_state"], + "paths": codex_runtime["paths"], + }, + "trigger_phrases": { + "enable": [ + "启用 open-reverselab Codex 模式", + "enable open-reverselab codex mode", + ], + "disable": [ + "停用 open-reverselab Codex 模式", + "disable open-reverselab codex mode", + ], + "status": [ + "查看 open-reverselab Codex 状态", + "show open-reverselab codex status", + ], + }, + } + write_json(project_marker_path(project_root), marker_payload) + write_text(project_marker_root(project_root) / PROJECT_README, project_doc_readme(project_root)) + write_text(project_marker_root(project_root) / PROJECT_QUICK_START, project_doc_quick_start(project_root)) + write_text(project_marker_root(project_root) / PROJECT_TASK_TEMPLATE, project_doc_task_template(project_root)) + ensure_agents_block(project_root, project_id) + ensure_gitignore_block(project_root) + registry = load_registry(paths) + registry["projects"][project_id] = { + "project_id": project_id, + "project_root": str(project_root), + "repo_root": str(paths.repo_root), + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "project_mode": "standard_active", + "bound_at": marker_payload["bound_at"], + "updated_at": marker_payload["updated_at"], + } + save_registry(registry, paths) + return project_state(project_root, paths) + + +def remove_project_binding(project_root: Path, paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + marker_payload = read_json(project_marker_path(project_root), {}) + project_id = marker_payload.get("project_id", "") + restore_state = ((marker_payload.get("project_codex") or {}).get("restore_state") or {}) + remove_agents_block(project_root) + remove_gitignore_block(project_root) + disable_project_codex_runtime(project_root, restore_state) + marker_root = project_marker_root(project_root) + if marker_root.exists(): + shutil.rmtree(marker_root, ignore_errors=True) + registry = load_registry(paths) + if project_id and project_id in registry["projects"]: + registry["projects"].pop(project_id, None) + save_registry(registry, paths) + return { + "project_root": str(project_root), + "project_id": project_id, + "bound": False, + "removed": True, + } + + +def rebuild_registry_entry_from_marker(project_root: Path, marker_data: dict[str, Any], paths: AdapterPaths | None = None) -> None: + paths = ensure_layout(paths) + project_id = marker_data.get("project_id", "") + if not project_id: + return + registry = load_registry(paths) + adapter = project_adapter_metadata(marker_data) + registry["projects"][project_id] = { + "project_id": project_id, + "project_root": str(project_root), + "repo_root": str(paths.repo_root), + "adapter_schema": adapter["schema"] or ADAPTER_SCHEMA, + "adapter_version": adapter["version"] or ADAPTER_VERSION, + "project_mode": adapter["mode"] or "standard_active", + "bound_at": marker_data.get("bound_at") or utc_now_iso(), + "updated_at": utc_now_iso(), + } + save_registry(registry, paths) + + +def refresh_bound_project(project_root: Path, paths: AdapterPaths | None = None, reason: str = "status") -> dict[str, Any] | None: + paths = ensure_layout(paths) + marker = project_marker_path(project_root) + if not marker.exists(): + return None + marker_data = read_json(marker, {}) + rebuild_registry_entry_from_marker(project_root, marker_data, paths) + if not project_requires_upgrade(marker_data): + return None + project_id = marker_data.get("project_id", "") or str(uuid.uuid4()) + backup = create_backup( + f"upgrade-project-{reason}", + paths=paths, + target_project_root=project_root, + include_global_state=False, + ) + status = apply_project_binding(project_root, project_id, paths) + return { + "project_root": str(project_root), + "project_id": project_id, + "backup_id": backup["backup_id"], + "status": status, + "upgraded": True, + "from_version": project_adapter_metadata(marker_data)["version"], + "to_version": ADAPTER_VERSION, + } + + +def enable_project_mode(project_root: str | Path, confirm: str = "", paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + root = normalize_project_root(project_root) + plan = build_project_plan(root, "enable", paths) + state = plan["state"] + if state["bound"] and state["healthy"]: + marker_data = read_json(project_marker_path(root), {}) + refresh = refresh_bound_project(root, paths, reason="enable") + project_id = marker_data.get("project_id", "") + return { + "ok": True, + "already_bound": True, + "auto_upgraded": bool(refresh), + "upgrade_result": refresh, + "project_root": str(root), + "project_id": project_id, + "status": project_state(root, paths), + } + + expected = plan["confirmation_text"] + if confirm.strip() != expected: + return { + "ok": False, + "requires_confirmation": True, + "confirmation_text": expected, + "preview": plan, + } + + backup = create_backup("bind-project", paths=paths, target_project_root=root, include_global_state=False) + result = apply_project_binding(root, plan["project_id"], paths) + payload = { + "ok": True, + "action": "enable", + "backup_id": backup["backup_id"], + "project_root": str(root), + "project_id": plan["project_id"], + "status": result, + } + write_json(paths.start_report, payload) + return payload + + +def disable_project_mode(project_root: str | Path, confirm: str = "", paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + root = normalize_project_root(project_root) + state = project_state(root, paths) + expected = confirm_text(root, "disable") + if not state["bound"]: + return { + "ok": True, + "already_disabled": True, + "project_root": str(root), + "status": state, + } + if confirm.strip() != expected: + return { + "ok": False, + "requires_confirmation": True, + "confirmation_text": expected, + "preview": build_project_plan(root, "disable", paths), + } + backup = create_backup("unbind-project", paths=paths, target_project_root=root, include_global_state=False) + result = remove_project_binding(root, paths) + payload = { + "ok": True, + "action": "disable", + "backup_id": backup["backup_id"], + "project_root": str(root), + "status": result, + } + write_json(paths.start_report, payload) + return payload + + +def project_mode_status(project_root: str | Path, paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + root = normalize_project_root(project_root) + marker_data = read_json(project_marker_path(root), {}) + auto_upgrade = refresh_bound_project(root, paths, reason="status") if marker_data else None + return { + "ok": True, + "project_root": str(root), + "auto_upgraded": bool(auto_upgrade), + "upgrade_result": auto_upgrade, + "status": project_state(root, paths), + } + + +def repair_project_mode(project_root: str | Path, confirm: str = "", paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + root = normalize_project_root(project_root) + marker_data = read_json(project_marker_path(root), {}) + project_id = marker_data.get("project_id", "") or str(uuid.uuid4()) + expected = confirm_text(root, "enable") + plan = build_project_plan(root, "repair", paths) + if confirm.strip() != expected: + return { + "ok": False, + "requires_confirmation": True, + "confirmation_text": expected, + "preview": plan, + } + backup = create_backup("repair-project", paths=paths, target_project_root=root, include_global_state=False) + result = apply_project_binding(root, project_id, paths) + payload = { + "ok": True, + "action": "repair", + "backup_id": backup["backup_id"], + "project_root": str(root), + "project_id": project_id, + "status": result, + } + write_json(paths.start_report, payload) + return payload + + +def discover_registered_project_upgrades(paths: AdapterPaths | None = None) -> list[dict[str, Any]]: + paths = ensure_layout(paths) + registry = load_registry(paths) + candidates: list[dict[str, Any]] = [] + for project in sorted(registry.get("projects", {}).values(), key=lambda item: item.get("project_root", "")): + project_root = Path(project.get("project_root", "")) + item = { + "project_id": project.get("project_id", ""), + "project_root": str(project_root), + } + if not project_root.exists(): + item["result"] = "offline_missing" + candidates.append(item) + continue + marker_data = read_json(project_marker_path(project_root), {}) + if not marker_data: + item["result"] = "marker_missing" + candidates.append(item) + continue + adapter = project_adapter_metadata(marker_data) + item["current_adapter_version"] = adapter["version"] + item["current_adapter_schema"] = adapter["schema"] + item["result"] = "upgrade_needed" if project_requires_upgrade(marker_data) else "already_current" + candidates.append(item) + return candidates + + +def upgrade_registered_projects(global_backup_id: str = "", paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + results: list[dict[str, Any]] = [] + for item in discover_registered_project_upgrades(paths): + result = dict(item) + project_root = Path(item["project_root"]) + if item["result"] != "upgrade_needed": + results.append(result) + continue + try: + marker_data = read_json(project_marker_path(project_root), {}) + project_id = marker_data.get("project_id", "") or item["project_id"] or str(uuid.uuid4()) + backup = create_backup("upgrade-project", paths=paths, target_project_root=project_root, include_global_state=False) + state = apply_project_binding(project_root, project_id, paths) + result.update( + { + "result": "upgraded", + "backup_id": backup["backup_id"], + "from_version": project_adapter_metadata(marker_data)["version"], + "to_version": ADAPTER_VERSION, + "status": state, + } + ) + except Exception as exc: + result.update( + { + "result": "upgrade_failed", + "error": str(exc), + } + ) + results.append(result) + batch = write_upgrade_batch_manifest(results, global_backup_id=global_backup_id, paths=paths) + return { + "batch_id": batch["batch_id"], + "summary": batch["summary"], + "projects": results, + } + + +def install_or_repair_global_config( + replace_related: bool = False, + confirm_project_upgrades: bool = False, + paths: AdapterPaths | None = None, +) -> dict[str, Any]: + paths = ensure_layout(paths) + uv_path = detect_uv() + python_path = detect_python() + status = current_global_status(paths) + if not python_path: + payload = { + "schema": 1, + "generated_at": utc_now_iso(), + "overall": "FAIL", + "error": "python not found", + } + write_json(paths.install_report, payload) + return payload + if not uv_path: + payload = { + "schema": 1, + "generated_at": utc_now_iso(), + "overall": "FAIL", + "error": "uv not found", + } + write_json(paths.install_report, payload) + return payload + + related = [ + section + for section in status["related_mcp_sections"] + if section["name"] != SERVER_NAME + ] + if related and not replace_related: + payload = { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "generated_at": utc_now_iso(), + "overall": "NEEDS_CONFIRMATION", + "confirmation_kind": "replace_related", + "message": "Found old related Codex MCP entries that match this repo by command/path evidence.", + "related_entries": related, + "desired_server": status["desired_server"], + } + write_json(paths.install_report, payload) + return payload + + project_upgrade_scan = discover_registered_project_upgrades(paths) + pending_project_upgrades = [item for item in project_upgrade_scan if item.get("result") == "upgrade_needed"] + if pending_project_upgrades and not confirm_project_upgrades: + payload = { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "generated_at": utc_now_iso(), + "overall": "NEEDS_CONFIRMATION", + "confirmation_kind": "project_upgrades", + "message": "Registered public-version projects need adapter upgrades before they are reopened.", + "close_sessions_required": True, + "project_upgrade_summary": { + "upgrade_needed": len(pending_project_upgrades), + "offline_missing": sum(1 for item in project_upgrade_scan if item.get("result") == "offline_missing"), + "marker_missing": sum(1 for item in project_upgrade_scan if item.get("result") == "marker_missing"), + "already_current": sum(1 for item in project_upgrade_scan if item.get("result") == "already_current"), + }, + "project_upgrade_candidates": pending_project_upgrades, + } + write_json(paths.install_report, payload) + return payload + + sync_result = subprocess.run( + [uv_path, "sync", "--project", str(upstream_project_dir(paths))], + cwd=str(paths.repo_root), + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + ) + if sync_result.returncode != 0: + payload = { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "generated_at": utc_now_iso(), + "overall": "FAIL", + "error": "uv sync failed", + "stdout": sync_result.stdout[-4000:], + "stderr": sync_result.stderr[-4000:], + } + write_json(paths.install_report, payload) + return payload + + backup = create_backup("install-global-config", paths=paths, extra_files=[paths.workspace_codex_config]) + config_text = read_text(paths.global_codex_config) + if replace_related: + for section in related: + config_text = remove_mcp_server_sections(config_text, section["name"]) + updated_text = ensure_global_server_and_instructions(config_text, paths, uv_path) + write_text(paths.global_codex_config, updated_text) + upgrade_batch = upgrade_registered_projects(global_backup_id=backup["backup_id"], paths=paths) if project_upgrade_scan else None + + runtime_state = { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "updated_at": utc_now_iso(), + "repo_root": str(paths.repo_root), + "server_name": SERVER_NAME, + "workspace_codex_config_preserved": True, + "backup_id": backup["backup_id"], + "project_upgrade_batch_id": upgrade_batch["batch_id"] if upgrade_batch else "", + } + write_json(paths.runtime_state_file, runtime_state) + write_json( + paths.local_manifest, + { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "updated_at": utc_now_iso(), + "managed_root_bats": [ + "LAUNCHER.bat", + "CODEX.bat", + ], + "local_runtime_dir": LOCAL_RUNTIME_DIR, + "server_name": SERVER_NAME, + "backup_id": backup["backup_id"], + "project_upgrade_batch_id": upgrade_batch["batch_id"] if upgrade_batch else "", + }, + ) + payload = { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "generated_at": utc_now_iso(), + "overall": "PASS", + "backup_id": backup["backup_id"], + "uv_sync_project": str(upstream_project_dir(paths)), + "global_config": str(paths.global_codex_config), + "workspace_codex_config_exists": paths.workspace_codex_config.exists(), + "server_name": SERVER_NAME, + "related_entries_replaced": related, + "project_upgrade_batch": upgrade_batch, + } + write_json(paths.install_report, payload) + return payload + + +def uninstall_global_config(paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + backup = create_backup("remove-global-config", paths=paths) + config_text = read_text(paths.global_codex_config) + updated_text = remove_global_server_and_instructions(config_text) + write_text(paths.global_codex_config, updated_text) + payload = { + "schema": 1, + "generated_at": utc_now_iso(), + "overall": "PASS", + "backup_id": backup["backup_id"], + "global_config": str(paths.global_codex_config), + } + write_json(paths.install_report, payload) + return payload + + +def codex_status_summary(paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + registry = load_registry(paths) + global_status = current_global_status(paths) + wrapper_exists = wrapper_script_path().exists() + manager_exists = manager_script_path().exists() + global_install_ready = bool( + global_status["config_exists"] and global_status["developer_block_present"] and global_status["server_block_present"] + ) + return { + "schema": 1, + "adapter_schema": ADAPTER_SCHEMA, + "adapter_version": ADAPTER_VERSION, + "generated_at": utc_now_iso(), + "repo_root": str(paths.repo_root), + "local_runtime_dir": str(paths.local_root), + "global_codex_config": str(paths.global_codex_config), + "workspace_codex_config_exists": paths.workspace_codex_config.exists(), + "server_name": SERVER_NAME, + "global_install_ready": global_install_ready, + "developer_block_present": global_status["developer_block_present"], + "server_block_present": global_status["server_block_present"], + "wrapper_exists": wrapper_exists, + "manager_exists": manager_exists, + "related_mcp_sections": global_status["related_mcp_sections"], + "replaceable_related_mcp_sections": [ + section for section in global_status["related_mcp_sections"] if section.get("name") != SERVER_NAME + ], + "bound_project_count": len(registry["projects"]), + "bound_projects": sorted(registry["projects"].values(), key=lambda item: item.get("project_root", "")), + "available_backups": [item.get("backup_id", "") for item in list_backups(paths)[:10]], + "available_upgrade_batches": [item.get("batch_id", "") for item in list_upgrade_batches(paths)[:10]], + } + + +def write_report(path: Path, payload: dict[str, Any]) -> dict[str, Any]: + write_json(path, payload) + return payload + + +def file_uri_to_path(uri: str) -> Path: + parsed = urlparse(uri) + if parsed.scheme and parsed.scheme != "file": + raise ValueError(f"unsupported root uri: {uri}") + if parsed.scheme == "file": + return Path(url2pathname(unquote(parsed.path))).resolve() + return Path(uri).expanduser().resolve() + + +def select_project_root(explicit: str = "", roots: list[str] | None = None) -> Path: + if explicit: + return normalize_project_root(explicit) + if roots: + for root in roots: + try: + candidate = file_uri_to_path(root) + except Exception: + continue + if candidate.exists() and candidate.is_dir(): + return candidate + cwd = Path.cwd().resolve() + return cwd + + +def wrapper_environment(project_root: Path | None = None, paths: AdapterPaths | None = None) -> dict[str, str]: + paths = paths or adapter_paths() + target_root = project_root or paths.repo_root + env = os.environ.copy() + env["REVERSELAB_LAB_ROOT"] = str(paths.repo_root) + env["REVERSELAB_DATA_ROOT"] = str(target_root) + env["REVERSELAB_OUTPUT_NAMESPACE"] = OUTPUT_NAMESPACE + env["PYTHONIOENCODING"] = "utf-8" + return env + + +def invoke_upstream_tool(tool_name: str, arguments: dict[str, Any], project_root: Path | None = None, paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = paths or adapter_paths() + script = wrapper_script_path() + payload = json.dumps(arguments, ensure_ascii=False) + command = [ + sys.executable, + str(script), + "--invoke-upstream", + tool_name, + "--arguments-json", + payload, + ] + if project_root: + command.extend(["--project-root", str(project_root)]) + result = subprocess.run( + command, + cwd=str(paths.repo_root), + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + # The MCP wrapper itself runs over stdio. If the child process inherits the + # parent's stdin pipe, upstream tool calls can hang indefinitely while the + # client waits for a response. Detach stdin for the helper subprocess. + stdin=subprocess.DEVNULL, + env=wrapper_environment(project_root, paths), + ) + if result.returncode != 0: + return { + "error": "upstream tool invocation failed", + "tool": tool_name, + "exit_code": result.returncode, + "stdout": result.stdout[-4000:], + "stderr": result.stderr[-4000:], + } + try: + return json.loads(result.stdout) + except json.JSONDecodeError: + return { + "error": "invalid upstream tool output", + "tool": tool_name, + "stdout": result.stdout[-4000:], + "stderr": result.stderr[-4000:], + } + + +def discover_upstream_tool_names(paths: AdapterPaths | None = None) -> list[str]: + paths = paths or adapter_paths() + env = wrapper_environment(paths.repo_root, paths) + uv_path = detect_uv() + if uv_path: + command = [ + uv_path, + "run", + "--project", + str(upstream_project_dir(paths)), + "python", + str(wrapper_script_path()), + "--list-upstream-tools", + ] + else: + command = [ + sys.executable, + str(wrapper_script_path()), + "--list-upstream-tools", + ] + result = subprocess.run( + command, + cwd=str(paths.repo_root), + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + env=env, + ) + if result.returncode != 0: + raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "failed to list upstream tools") + payload = json.loads(result.stdout) + return payload["tools"] + + +def doctor(paths: AdapterPaths | None = None) -> dict[str, Any]: + paths = ensure_layout(paths) + payload = { + "schema": 1, + "generated_at": utc_now_iso(), + "repo_root": str(paths.repo_root), + "wrapper_script": str(wrapper_script_path()), + "manager_script": str(manager_script_path()), + "global_status": current_global_status(paths), + "upstream_tool_count": len(discover_upstream_tool_names(paths)), + } + write_json(paths.verify_report, payload) + return payload diff --git a/tests/codex/test_codex_support.py b/tests/codex/test_codex_support.py new file mode 100644 index 0000000..bb7c26c --- /dev/null +++ b/tests/codex/test_codex_support.py @@ -0,0 +1,251 @@ +from __future__ import annotations + +import importlib.util +import json +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + + +MODULE_PATH = Path(__file__).resolve().parents[2] / "scripts" / "codex" / "python" / "open_reverselab_codex_support.py" +SPEC = importlib.util.spec_from_file_location("open_reverselab_codex_support", MODULE_PATH) +support = importlib.util.module_from_spec(SPEC) +assert SPEC and SPEC.loader +sys.modules[SPEC.name] = support +SPEC.loader.exec_module(support) + + +def make_paths(tmp_path: Path) -> support.AdapterPaths: + repo_root = tmp_path / "repo" + local_root = repo_root / support.LOCAL_RUNTIME_DIR + global_codex_root = tmp_path / "home" / ".codex" + return support.AdapterPaths( + repo_root=repo_root, + local_root=local_root, + python_root=repo_root / "scripts" / "codex" / "python", + state_root=local_root / "state", + backups_root=local_root / "backups", + batch_root=local_root / "upgrade-batches", + reports_root=local_root / "reports", + docs_root=repo_root / "docs" / "codex", + global_codex_root=global_codex_root, + global_codex_config=global_codex_root / "config.toml", + workspace_codex_config=repo_root / ".codex" / "config.toml", + registry_file=local_root / "state" / "project_registry.json", + runtime_state_file=local_root / "state" / "adapter_state.json", + install_report=local_root / "reports" / "codex-install-report.json", + verify_report=local_root / "reports" / "codex-verify-report.json", + backup_report=local_root / "reports" / "codex-backup-report.json", + restore_report=local_root / "reports" / "codex-restore-report.json", + start_report=local_root / "reports" / "codex-start-report.json", + menu_report=local_root / "reports" / "codex-menu-report.json", + local_manifest=local_root / "state" / "manifest.json", + ) + + +def test_upsert_and_remove_global_instructions_and_server(tmp_path: Path) -> None: + paths = make_paths(tmp_path) + support.ensure_layout(paths) + original = """model = "gpt-5.4" +service_tier = "fast" + +[mcp_servers] + +[projects.'e:\\demo'] +trust_level = "trusted" +""" + + updated = support.ensure_global_server_and_instructions(original, paths, uv_path="C:\\Tools\\uv.exe") + assert "developer_instructions" in updated + assert support.GLOBAL_BLOCK_BEGIN in updated + assert f"[mcp_servers.{support.SERVER_NAME}]" in updated + assert "uv.exe" in updated + + restored = support.remove_global_server_and_instructions(updated) + assert support.GLOBAL_BLOCK_BEGIN not in restored + assert f"[mcp_servers.{support.SERVER_NAME}]" not in restored + assert "model = \"gpt-5.4\"" in restored + + +def test_bind_and_unbind_project_updates_marker_agents_and_registry(tmp_path: Path) -> None: + paths = make_paths(tmp_path) + support.ensure_layout(paths) + project_root = tmp_path / "target-project" + project_root.mkdir(parents=True) + (project_root / ".git").mkdir() + + preview = support.enable_project_mode(project_root, confirm="", paths=paths) + assert preview["requires_confirmation"] is True + confirmation = preview["confirmation_text"] + + bound = support.enable_project_mode(project_root, confirm=confirmation, paths=paths) + assert bound["ok"] is True + marker_path = project_root / support.PROJECT_MARKER_DIR / support.PROJECT_MARKER_FILE + assert marker_path.exists() + assert (project_root / "notes" / support.OUTPUT_NAMESPACE).exists() + assert (project_root / "reports" / support.OUTPUT_NAMESPACE).exists() + assert (project_root / "exports" / support.OUTPUT_NAMESPACE).exists() + assert (project_root / ".codex" / "config.toml").exists() + assert (project_root / ".codex" / support.PROJECT_CODEX_MANIFEST).exists() + assert (project_root / ".codex" / support.PROJECT_CODEX_CTF_CONFIG).exists() + assert (project_root / ".codex" / support.PROJECT_CODEX_PROMPT).exists() + codex_config_text = (project_root / ".codex" / "config.toml").read_text(encoding="utf-8") + assert support.PROJECT_CODEX_BLOCK_BEGIN in codex_config_text + assert support.PROJECT_BLOCK_BEGIN in (project_root / "AGENTS.md").read_text(encoding="utf-8") + assert support.MANAGED_IGNORE_HEADER in (project_root / ".gitignore").read_text(encoding="utf-8") + + registry = json.loads(paths.registry_file.read_text(encoding="utf-8")) + marker_payload = json.loads(marker_path.read_text(encoding="utf-8")) + project_id = marker_payload["project_id"] + assert project_id in registry["projects"] + assert marker_payload["adapter_version"] == support.ADAPTER_VERSION + assert registry["projects"][project_id]["adapter_version"] == support.ADAPTER_VERSION + + disable_preview = support.disable_project_mode(project_root, confirm="", paths=paths) + assert disable_preview["requires_confirmation"] is True + disabled = support.disable_project_mode(project_root, confirm=disable_preview["confirmation_text"], paths=paths) + assert disabled["ok"] is True + assert not (project_root / support.PROJECT_MARKER_DIR).exists() + agents_text = (project_root / "AGENTS.md").read_text(encoding="utf-8") + assert support.PROJECT_BLOCK_BEGIN not in agents_text + codex_config_text = (project_root / ".codex" / "config.toml").read_text(encoding="utf-8") + assert support.PROJECT_CODEX_BLOCK_BEGIN not in codex_config_text + assert (project_root / ".codex" / support.PROJECT_CODEX_MANIFEST).exists() + gitignore = project_root / ".gitignore" + if gitignore.exists(): + gitignore_text = gitignore.read_text(encoding="utf-8") + assert support.MANAGED_IGNORE_HEADER not in gitignore_text + + +def test_restore_backup_restores_only_integration_state(tmp_path: Path) -> None: + paths = make_paths(tmp_path) + support.ensure_layout(paths) + paths.global_codex_root.mkdir(parents=True, exist_ok=True) + paths.global_codex_config.write_text('model = "gpt-5.5"\n', encoding="utf-8") + project_root = tmp_path / "restore-project" + project_root.mkdir(parents=True) + (project_root / ".git").mkdir() + + preview = support.enable_project_mode(project_root, confirm="", paths=paths) + bound = support.enable_project_mode(project_root, confirm=preview["confirmation_text"], paths=paths) + backup_id = bound["backup_id"] + + marker_root = project_root / support.PROJECT_MARKER_DIR + marker_root.rename(project_root / ".open-reverselab-codex-missing") + codex_config = project_root / ".codex" / "config.toml" + codex_config.write_text("# broken\n", encoding="utf-8") + artifacts_root = project_root / "reports" / support.OUTPUT_NAMESPACE + artifact = artifacts_root / "result.txt" + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.write_text("keep me", encoding="utf-8") + paths.global_codex_config.write_text('model = "gpt-5.6"\n', encoding="utf-8") + + restored = support.restore_backup(backup_id, paths=paths) + assert restored["backup_id"] == backup_id + assert not marker_root.exists() + assert artifact.exists() + assert artifact.read_text(encoding="utf-8") == "keep me" + assert not codex_config.exists() + assert paths.global_codex_config.read_text(encoding="utf-8") == 'model = "gpt-5.6"\n' + + +def test_existing_project_codex_values_are_restored_on_disable(tmp_path: Path) -> None: + paths = make_paths(tmp_path) + support.ensure_layout(paths) + project_root = tmp_path / "existing-config-project" + project_root.mkdir(parents=True) + (project_root / ".git").mkdir() + codex_root = project_root / ".codex" + codex_root.mkdir() + original_text = ( + 'model = "gpt-5.5"\n' + 'approval_policy = "on-request"\n' + 'sandbox_mode = "workspace-write"\n' + 'model_instructions_file = "custom.md"\n' + ) + (codex_root / "config.toml").write_text(original_text, encoding="utf-8") + + preview = support.enable_project_mode(project_root, confirm="", paths=paths) + bound = support.enable_project_mode(project_root, confirm=preview["confirmation_text"], paths=paths) + assert bound["ok"] is True + enabled_text = (codex_root / "config.toml").read_text(encoding="utf-8") + assert support.PROJECT_CODEX_BLOCK_BEGIN in enabled_text + assert 'approval_policy = "on-request"' not in enabled_text + + disable_preview = support.disable_project_mode(project_root, confirm="", paths=paths) + disabled = support.disable_project_mode(project_root, confirm=disable_preview["confirmation_text"], paths=paths) + assert disabled["ok"] is True + restored_text = (codex_root / "config.toml").read_text(encoding="utf-8") + assert support.PROJECT_CODEX_BLOCK_BEGIN not in restored_text + assert 'approval_policy = "on-request"' in restored_text + assert 'sandbox_mode = "workspace-write"' in restored_text + assert 'model_instructions_file = "custom.md"' in restored_text + + +def test_project_status_rebuilds_registry_and_upgrades_outdated_marker(tmp_path: Path) -> None: + paths = make_paths(tmp_path) + support.ensure_layout(paths) + project_root = tmp_path / "status-project" + project_root.mkdir(parents=True) + (project_root / ".git").mkdir() + + preview = support.enable_project_mode(project_root, confirm="", paths=paths) + bound = support.enable_project_mode(project_root, confirm=preview["confirmation_text"], paths=paths) + assert bound["ok"] is True + + marker_path = project_root / support.PROJECT_MARKER_DIR / support.PROJECT_MARKER_FILE + marker_payload = json.loads(marker_path.read_text(encoding="utf-8")) + marker_payload["adapter_version"] = "0.9.0" + marker_payload["adapter"]["version"] = "0.9.0" + marker_path.write_text(json.dumps(marker_payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + paths.registry_file.unlink() + + status = support.project_mode_status(project_root, paths=paths) + assert status["ok"] is True + assert status["auto_upgraded"] is True + assert status["status"]["adapter_version"] == support.ADAPTER_VERSION + registry = json.loads(paths.registry_file.read_text(encoding="utf-8")) + assert marker_payload["project_id"] in registry["projects"] + + +def test_install_preserves_workspace_codex_config(tmp_path: Path, monkeypatch) -> None: + paths = make_paths(tmp_path) + support.ensure_layout(paths) + paths.global_codex_root.mkdir(parents=True, exist_ok=True) + paths.workspace_codex_config.parent.mkdir(parents=True, exist_ok=True) + paths.workspace_codex_config.write_text('approval_policy = "on-request"\n', encoding="utf-8") + + monkeypatch.setattr(support, "detect_python", lambda: "python") + monkeypatch.setattr(support, "detect_uv", lambda: "uv") + monkeypatch.setattr( + support.subprocess, + "run", + lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="", stderr=""), + ) + + payload = support.install_or_repair_global_config(paths=paths) + assert payload["overall"] == "PASS" + assert paths.workspace_codex_config.exists() + assert payload["workspace_codex_config_exists"] is True + + +def test_invoke_upstream_tool_detaches_stdin(tmp_path: Path, monkeypatch) -> None: + paths = make_paths(tmp_path) + support.ensure_layout(paths) + project_root = tmp_path / "target-project" + project_root.mkdir(parents=True) + + calls: list[dict[str, object]] = [] + + def fake_run(*args, **kwargs): + calls.append({"args": args, "kwargs": kwargs}) + return SimpleNamespace(returncode=0, stdout='{"ok": true}', stderr="") + + monkeypatch.setattr(support.subprocess, "run", fake_run) + + payload = support.invoke_upstream_tool("project_skills_status", {}, project_root=project_root, paths=paths) + + assert payload == {"ok": True} + assert len(calls) == 1 + assert calls[0]["kwargs"]["stdin"] is subprocess.DEVNULL diff --git a/tools/skills/mcp/ReverseLabToolsMCP/reverselab_mcp/config.py b/tools/skills/mcp/ReverseLabToolsMCP/reverselab_mcp/config.py index 228b443..b52db67 100644 --- a/tools/skills/mcp/ReverseLabToolsMCP/reverselab_mcp/config.py +++ b/tools/skills/mcp/ReverseLabToolsMCP/reverselab_mcp/config.py @@ -11,7 +11,7 @@ # ── Project root discovery ── PACKAGE_ROOT = Path(__file__).resolve().parents[1] -REVERSE_ROOT = next( +DISCOVERED_ROOT = next( ( parent for parent in [PACKAGE_ROOT, *PACKAGE_ROOT.parents] @@ -19,6 +19,15 @@ ), PACKAGE_ROOT.parents[4], ) +REVERSE_ROOT = Path(os.environ.get("REVERSELAB_LAB_ROOT", str(DISCOVERED_ROOT))).expanduser().resolve() +DATA_ROOT = Path(os.environ.get("REVERSELAB_DATA_ROOT", str(REVERSE_ROOT))).expanduser().resolve() +OUTPUT_NAMESPACE = os.environ.get("REVERSELAB_OUTPUT_NAMESPACE", "").strip() + + +def _data_dir(name: str) -> Path: + if OUTPUT_NAMESPACE: + return DATA_ROOT / name / OUTPUT_NAMESPACE + return DATA_ROOT / name # ── Directory shortcuts ── TOOLS_DIR = REVERSE_ROOT / "tools" @@ -26,27 +35,28 @@ TOOLS_WINDOWS_DIR = TOOLS_DIR / "windows" TOOLS_ANDROID_DIR = TOOLS_DIR / "android" -EXPORTS_DIR = REVERSE_ROOT / "exports" / "windows" / "triage" -EXPORTS_ROOT = REVERSE_ROOT / "exports" -AUDIT_DIR = REVERSE_ROOT / "exports" / "misc" / "audit" +EXPORTS_ROOT = _data_dir("exports") +EXPORTS_DIR = EXPORTS_ROOT / "windows" / "triage" +AUDIT_DIR = EXPORTS_ROOT / "misc" / "audit" AUDIT_LOG = AUDIT_DIR / "reverse_lab_tools_audit.jsonl" -GHIDRA_EXPORTS_DIR = REVERSE_ROOT / "exports" / "windows" / "ghidra" -GHIDRA_PROJECTS_DIR = REVERSE_ROOT / "projects" / "windows" / "ghidra-headless" +GHIDRA_EXPORTS_DIR = EXPORTS_ROOT / "windows" / "ghidra" +PROJECTS_DIR = _data_dir("projects") +GHIDRA_PROJECTS_DIR = PROJECTS_DIR / "windows" / "ghidra-headless" GHIDRA_SCRIPT_DIR = REVERSE_ROOT / "scripts" / "_shared" / "ghidra" -PATCHES_DIR = REVERSE_ROOT / "patches" -PROJECTS_DIR = REVERSE_ROOT / "projects" -REPORTS_DIR = REVERSE_ROOT / "reports" -SAMPLES_DIR = REVERSE_ROOT / "samples" +PATCHES_DIR = _data_dir("patches") +REPORTS_DIR = _data_dir("reports") +SAMPLES_DIR = _data_dir("samples") SAMPLE_QUARANTINE_DIR = SAMPLES_DIR / "_quarantine" -PROCMON_EXPORTS_DIR = REVERSE_ROOT / "exports" / "windows" / "procmon" -IOC_EXPORTS_DIR = REVERSE_ROOT / "exports" / "windows" / "iocs" -YARA_EXPORTS_DIR = REVERSE_ROOT / "exports" / "windows" / "yara" -SIGMA_EXPORTS_DIR = REVERSE_ROOT / "exports" / "windows" / "sigma" -ANDROID_EXPORTS_DIR = REVERSE_ROOT / "exports" / "android" +PROCMON_EXPORTS_DIR = EXPORTS_ROOT / "windows" / "procmon" +IOC_EXPORTS_DIR = EXPORTS_ROOT / "windows" / "iocs" +YARA_EXPORTS_DIR = EXPORTS_ROOT / "windows" / "yara" +SIGMA_EXPORTS_DIR = EXPORTS_ROOT / "windows" / "sigma" +ANDROID_EXPORTS_DIR = EXPORTS_ROOT / "android" DEBUG_SCRIPTS_DIR = REVERSE_ROOT / "scripts" / "windows" / "debug" PROCMON_FILTERS_DIR = REVERSE_ROOT / "scripts" / "windows" / "procmon" SCRIPTS_DIR = REVERSE_ROOT / "scripts" -NOTES_DIR = REVERSE_ROOT / "notes" +NOTES_DIR = _data_dir("notes") +CASES_DIR = _data_dir("cases") # ── Tool autodiscovery ── @@ -173,10 +183,11 @@ def _first_existing(*candidates: Path) -> Path: # ── Security: allow-listed roots ── ALLOWED_ROOTS = [ REVERSE_ROOT, + DATA_ROOT, ] GENERATED_ROOTS = [ - REVERSE_ROOT / "exports", + EXPORTS_ROOT, PATCHES_DIR, PROJECTS_DIR, REPORTS_DIR, diff --git a/tools/skills/mcp/ReverseLabToolsMCP/reverselab_mcp/tools/web_ctf.py b/tools/skills/mcp/ReverseLabToolsMCP/reverselab_mcp/tools/web_ctf.py index 180527c..57a9987 100644 --- a/tools/skills/mcp/ReverseLabToolsMCP/reverselab_mcp/tools/web_ctf.py +++ b/tools/skills/mcp/ReverseLabToolsMCP/reverselab_mcp/tools/web_ctf.py @@ -13,7 +13,7 @@ import sys from pathlib import Path -from ..config import REVERSE_ROOT, SCRIPTS_DIR, TOOLS_DIR +from ..config import CASES_DIR, DATA_ROOT, EXPORTS_ROOT, REVERSE_ROOT, SCRIPTS_DIR, TOOLS_DIR from ..paths import ensure_under # ── KB roots for all boards ── @@ -29,7 +29,7 @@ KB_ROUTER = SCRIPTS_DIR / "ctf-website" / "kb_router.py" CTF_AUTOPILOT = SCRIPTS_DIR / "ctf-website" / "ctf_autopilot.py" CTF_TOOLS_DIR = REVERSE_ROOT / "tools" / "ctf-website" -CTF_EXPORTS_DIR = REVERSE_ROOT / "exports" / "ctf-website" +CTF_EXPORTS_DIR = EXPORTS_ROOT / "ctf-website" BIN_DIR = REVERSE_ROOT / "tools" / "bin" BURP_DIR = CTF_TOOLS_DIR / "burp" @@ -246,7 +246,7 @@ def kb_catalog(board: str = "") -> dict: def ctf_new_challenge(name: str, url: str = "") -> dict: """Create a new CTF challenge case directory.""" - case_dir = REVERSE_ROOT / "cases" / name + case_dir = CASES_DIR / name template_dir = REVERSE_ROOT / "templates" / "cases" if case_dir.exists(): @@ -276,9 +276,9 @@ def ctf_new_challenge(name: str, url: str = "") -> dict: ) return { - "case": str(case_dir.relative_to(REVERSE_ROOT)), + "case": str(case_dir), "url": url, - "links": str(links.relative_to(REVERSE_ROOT)), + "links": str(links), } @@ -287,9 +287,9 @@ def _resolve_manifest_file(manifest_path: str) -> Path: raise ValueError("manifest_path is required") resolved = Path(manifest_path).expanduser() if not resolved.is_absolute(): - resolved = REVERSE_ROOT / resolved + resolved = DATA_ROOT / resolved resolved = resolved.resolve(strict=True) - ensure_under(resolved, [REVERSE_ROOT], "manifest path") + ensure_under(resolved, [DATA_ROOT], "manifest path") if not resolved.is_file(): raise ValueError(f"not a file: {resolved}") if resolved.name != "ai_manifest.json": @@ -516,9 +516,9 @@ def run_sqlmap_request(request_path: str, extra_args: str = "--batch", timeout: try: resolved = Path(request_path).expanduser() if not resolved.is_absolute(): - resolved = REVERSE_ROOT / resolved + resolved = DATA_ROOT / resolved resolved = resolved.resolve(strict=True) - ensure_under(resolved, [REVERSE_ROOT], "request path") + ensure_under(resolved, [DATA_ROOT], "request path") except Exception as e: return {"error": str(e)} args = f'-r "{resolved}" {extra_args}'.strip()