From bdaa7f9e8fc50cf91c8a8a18e2a0bec4bf21fb3b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B4=AA=E6=B3=BD=E9=91=AB?= Date: Tue, 18 Aug 2026 15:30:48 +0800 Subject: [PATCH] feat(gui): configure official Codex with managed access --- README.md | 12 +++++- corp_codex_pool/cli.py | 56 ++++++++++++++++++++++++- corp_codex_pool/codex_config.py | 72 +++++++++++++++++++++++++++++---- pyproject.toml | 2 +- tests/test_cli.py | 53 ++++++++++++++++++++++++ tests/test_codex_config.py | 26 ++++++++++++ 6 files changed, 210 insertions(+), 11 deletions(-) create mode 100644 tests/test_cli.py diff --git a/README.md b/README.md index 288c1da..1704c09 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ 把多个 Codex 订阅变成一个内部号池，让 [Multica](https://github.com/multica-ai/multica) 任务按人、按任务使用和审计公司额度。 -**CHEK 正式环境不向员工分发永久 Key，只允许 Multica 任务中的 `mcodex` 申请短时、任务绑定凭证；你自己的 `codex` 不受影响。** +**CHEK 正式环境提供两条受管路径：Multica 任务使用运行期绑定凭证；员工也可为官方 Codex GUI 申请无固定到期日、但可即时撤销且会话自动同步的个人凭证。** ``` codex → provider: openai → 你的个人订阅 @@ -86,7 +86,7 @@ flowchart LR G -.per-request 计量.-> DB[(用量表)] ``` -CHEK 正式环境只覆盖一种运行场景：**Multica daemon 拉起**。daemon 把 `CODEX_HOME` 和 `mat_` 任务票据注入进程，`mcodex` 用任务票向生产桥接层换取两小时有效的 `mcx_` 凭证，再 exec 真 Codex。直接在终端运行 `mcodex` 因缺少任务上下文会失败。 +Multica daemon 拉起的任务仍走 `mcodex`：daemon 注入 `CODEX_HOME` 与任务上下文，网关在每次请求时确认 Run 仍处于活动状态，Run 结束后凭证立即失效。官方 Codex GUI 则使用员工在 Multica 自助生成的 `mck_` 凭证，每次请求回查工作空间成员与密钥状态，并把会话同步回 Multica。 ### 依赖的三个上游机制 @@ -179,6 +179,14 @@ poolctl enroll ✓ daemon 使用的 codex：/…/mcodex（走号池） ``` +### 官方 Codex GUI / CC Switch + +1. 飞连登录 Multica，进入 `设置 → Tokens → 公司 Codex 访问`。 +2. 点击“创建访问”。推荐直接点“在 CC Switch 中打开”，确认导入后重启官方 Codex GUI。 +3. 不使用 CC Switch 时，安装本仓库后运行 `poolctl gui configure`，按隐藏提示粘贴凭证，再重启官方 Codex GUI。 + +该命令只更新官方 Codex 的 `config.toml` 与 `auth.json`，保留已有官方登录字段，不安装启动器。凭证没有固定过期时间；轮换、主动撤销或被移出工作空间后会立即失效。GUI 中的提问、回复、模型、Token 与时间会同步到 Multica，禁止用于非公司工作。 + ### 验证 ```bash diff --git a/corp_codex_pool/cli.py b/corp_codex_pool/cli.py index 4c0e8ce..253d9c2 100644 --- a/corp_codex_pool/cli.py +++ b/corp_codex_pool/cli.py @@ -2,7 +2,7 @@ 设计原则： - 所有会改变状态的命令都支持 --dry-run，且默认打印将要发生的变更。 -- 正式员工入口不分发永久密钥；旧管理员命令输出的密钥其余场合一律打码。 +- 正式员工只拿到网关包装凭证，不接触底层号池密钥；旧管理员命令输出的密钥其余场合一律打码。 - 每一步失败都给出可执行的下一步，而不是只报错。 """ @@ -19,9 +19,11 @@ from .codex_config import ( ConfigInjectionError, ProviderSpec, + default_auth_path, default_config_path, inject, remove, + write_openai_auth_key, ) from .config import Settings from .gateway import ( @@ -118,6 +120,58 @@ def _print_block_diff(before: str, after: str) -> None: click.secho(line.rstrip("\n"), fg=color) +# ---------------------------------------------------------------- official GUI + +@main.group() +def gui(): + """配置官方 Codex GUI 使用公司受管号池。""" + + +@gui.command("configure") +@click.option("--key-stdin", is_flag=True, help="从 stdin 读取凭证") +@click.option("--config", "config_path", type=click.Path(path_type=Path)) +@click.option("--auth", "auth_path", type=click.Path(path_type=Path)) +@click.option("--dry-run", is_flag=True, help="仅校验，不写文件") +@click.pass_context +def gui_configure(ctx, key_stdin, config_path, auth_path, dry_run): + """一次性配置官方 Codex GUI；不安装或替换 Codex 启动器。""" + settings = _settings(ctx) + if key_stdin: + credential = sys.stdin.read().strip() + else: + credential = click.prompt("粘贴 Multica 里生成的 mck_ 凭证", hide_input=True).strip() + if not credential.startswith("mck_"): + _fail("凭证格式不正确", "请在 Multica 设置 → Tokens → 公司 Codex 访问中重新生成") + + spec = ProviderSpec( + provider_id="chek", + name="CHEK Company Codex", + base_url=settings.pool_base_url, + requires_openai_auth=True, + env_http_headers={}, + ) + target_config = config_path or default_config_path() + target_auth = auth_path or default_auth_path() + try: + config_result = inject(spec, path=target_config, dry_run=dry_run) + auth_backup = write_openai_auth_key(credential, path=target_auth, dry_run=dry_run) + except (ConfigInjectionError, OSError) as exc: + _fail(str(exc)) + return + + if dry_run: + click.secho("[dry-run] 配置与凭证校验通过，未写入文件", fg="yellow") + return + click.secho("✓ 官方 Codex GUI 已切换到 CHEK 公司号池", fg="green") + click.echo(f" 配置：{config_result.path}") + click.echo(f" 凭证：{target_auth}（600，仅更新 OPENAI_API_KEY）") + if config_result.backup: + click.echo(f" 配置备份：{config_result.backup}") + if auth_backup: + click.echo(f" 凭证备份：{auth_backup}") + click.echo("\n请完全退出并重新打开官方 Codex GUI。所有公司号池对话会同步到 Multica。") + + @main.command("unsetup") @click.option("--config", "config_path", type=click.Path(path_type=Path)) @click.option("--dry-run", is_flag=True) diff --git a/corp_codex_pool/codex_config.py b/corp_codex_pool/codex_config.py index 43f29be..a734a43 100644 --- a/corp_codex_pool/codex_config.py +++ b/corp_codex_pool/codex_config.py @@ -19,6 +19,7 @@ from __future__ import annotations +import json import os import re import shutil @@ -51,6 +52,9 @@ class ProviderSpec: base_url: str = "https://codex.chekkk.com/v1" env_key: str = "GW_API_KEY" env_key_instructions: str = "由号池控制台下发，请勿手工设置" + # Official Codex GUI does not inherit a shell environment. In this mode + # Codex reads OPENAI_API_KEY from auth.json instead of env_key. + requires_openai_auth: bool = False # wire_api 只接受 "responses"："chat" 已从 codex 移除 # (codex-rs/model-provider-info/src/lib.rs:49, :71-81) wire_api: str = "responses" @@ -77,7 +81,7 @@ def validate(self) -> None: raise ConfigInjectionError( f"provider_id 只允许字母数字下划线连字符，收到 {self.provider_id!r}" ) - if not re.fullmatch(r"[A-Z][A-Z0-9_]*", self.env_key): + if not self.requires_openai_auth and not re.fullmatch(r"[A-Z][A-Z0-9_]*", self.env_key): raise ConfigInjectionError( f"env_key 应为大写环境变量名，收到 {self.env_key!r}" ) @@ -112,9 +116,12 @@ def render_block(spec: ProviderSpec) -> str: lines.append(f"name = {_toml_str(spec.name)}") lines.append(f"base_url = {_toml_str(spec.base_url)}") lines.append(f"wire_api = {_toml_str(spec.wire_api)}") - lines.append(f"env_key = {_toml_str(spec.env_key)}") - if spec.env_key_instructions: - lines.append(f"env_key_instructions = {_toml_str(spec.env_key_instructions)}") + if spec.requires_openai_auth: + lines.append("requires_openai_auth = true") + else: + lines.append(f"env_key = {_toml_str(spec.env_key)}") + if spec.env_key_instructions: + lines.append(f"env_key_instructions = {_toml_str(spec.env_key_instructions)}") if spec.env_http_headers: lines.append("") @@ -177,11 +184,15 @@ def verify(text: str, spec: ProviderSpec) -> dict: ) provider = providers[spec.provider_id] - for key, expected in ( + expected_fields = [ ("base_url", spec.base_url), ("wire_api", spec.wire_api), - ("env_key", spec.env_key), - ): + ] + if spec.requires_openai_auth: + expected_fields.append(("requires_openai_auth", True)) + else: + expected_fields.append(("env_key", spec.env_key)) + for key, expected in expected_fields: if provider.get(key) != expected: raise ConfigInjectionError( f"model_providers.{spec.provider_id}.{key} 期望 {expected!r}，" @@ -212,6 +223,53 @@ def default_config_path() -> Path: return base / "config.toml" +def default_auth_path() -> Path: + """Official Codex auth.json next to config.toml.""" + return default_config_path().with_name("auth.json") + + +def write_openai_auth_key( + key: str, + path: Path | None = None, + *, + dry_run: bool = False, +) -> Path | None: + """Upsert OPENAI_API_KEY without removing an existing official login.""" + key = key.strip() + if not key: + raise ConfigInjectionError("API 凭证不能为空") + path = path or default_auth_path() + original = path.read_text(encoding="utf-8") if path.exists() else "{}" + try: + data = json.loads(original) + except json.JSONDecodeError as exc: + raise ConfigInjectionError(f"现有 auth.json 不是合法 JSON：{exc}") from exc + if not isinstance(data, dict): + raise ConfigInjectionError("现有 auth.json 必须是 JSON 对象") + data["OPENAI_API_KEY"] = key + rendered = json.dumps(data, ensure_ascii=False, indent=2) + "\n" + if dry_run: + return None + + path.parent.mkdir(parents=True, exist_ok=True) + backup_path = None + if path.exists(): + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + backup_path = path.with_suffix(f".json.bak-{stamp}") + shutil.copy2(path, backup_path) + tmp = path.with_suffix(".json.tmp") + try: + tmp.write_text(rendered, encoding="utf-8") + os.chmod(tmp, 0o600) + os.replace(tmp, path) + except Exception: + tmp.unlink(missing_ok=True) + if backup_path and backup_path.exists(): + shutil.copy2(backup_path, path) + raise + return backup_path + + @dataclass class InjectResult: path: Path diff --git a/pyproject.toml b/pyproject.toml index 6a00fc4..ccc6f19 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "corp-codex-pool" -version = "0.2.1" +version = "0.3.0" description = "Codex 号池与 multica 运行时的集成层：provider 注入、密钥下发、用量对账" requires-python = ">=3.11" dependencies = [ diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..1382972 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,53 @@ +import json +import tomllib + +from click.testing import CliRunner + +from corp_codex_pool.cli import main + + +def test_gui_configure_writes_official_codex_files(tmp_path): + config = tmp_path / "config.toml" + auth = tmp_path / "auth.json" + auth.write_text(json.dumps({"tokens": {"access_token": "official"}}), encoding="utf-8") + + result = CliRunner().invoke( + main, + [ + "gui", + "configure", + "--key-stdin", + "--config", + str(config), + "--auth", + str(auth), + ], + input="mck_employee\n", + ) + + assert result.exit_code == 0, result.output + parsed = tomllib.loads(config.read_text(encoding="utf-8")) + assert parsed["model_provider"] == "chek" + assert parsed["model_providers"]["chek"]["requires_openai_auth"] is True + stored_auth = json.loads(auth.read_text(encoding="utf-8")) + assert stored_auth["OPENAI_API_KEY"] == "mck_employee" + assert stored_auth["tokens"]["access_token"] == "official" + + +def test_gui_configure_rejects_unmanaged_key(tmp_path): + result = CliRunner().invoke( + main, + [ + "gui", + "configure", + "--key-stdin", + "--config", + str(tmp_path / "config.toml"), + "--auth", + str(tmp_path / "auth.json"), + ], + input="sk-not-managed\n", + ) + + assert result.exit_code != 0 + assert "凭证格式不正确" in result.output diff --git a/tests/test_codex_config.py b/tests/test_codex_config.py index bc818ff..73b86af 100644 --- a/tests/test_codex_config.py +++ b/tests/test_codex_config.py @@ -1,3 +1,4 @@ +import json import tomllib import pytest @@ -13,6 +14,7 @@ render_block, strip_block, verify, + write_openai_auth_key, ) # 取自真实宿主配置的形状：顶层键在前，[projects.*] 在后 @@ -108,6 +110,12 @@ def test_quotes_in_values_escaped(self): data = tomllib.loads(build_new_text(REAL_CONFIG, s)) assert data["model_providers"]["gw"]["name"] == 'Acme "Pool" Inc' + def test_official_gui_uses_auth_json_instead_of_environment(self): + gui = spec(requires_openai_auth=True, env_http_headers={}) + provider = tomllib.loads(build_new_text(REAL_CONFIG, gui))["model_providers"]["gw"] + assert provider["requires_openai_auth"] is True + assert "env_key" not in provider + class TestIdempotence: def test_second_run_is_noop(self): @@ -233,3 +241,21 @@ def test_multica_managed_block_untouched(self, tmp_path): assert tomllib.loads(out)["model_provider"] == "gw" # 我们的块不能插进 multica 的块里 assert out.index("# END multica-managed") < out.index(BEGIN) + + +class TestOfficialGuiAuth: + def test_preserves_official_login_fields(self, tmp_path): + auth = tmp_path / "auth.json" + auth.write_text(json.dumps({"tokens": {"access_token": "official"}}), encoding="utf-8") + + backup = write_openai_auth_key("mck_test", auth) + + stored = json.loads(auth.read_text(encoding="utf-8")) + assert stored["tokens"]["access_token"] == "official" + assert stored["OPENAI_API_KEY"] == "mck_test" + assert backup is not None and backup.exists() + assert auth.stat().st_mode & 0o777 == 0o600 + + def test_rejects_empty_key(self, tmp_path): + with pytest.raises(ConfigInjectionError, match="不能为空"): + write_openai_auth_key("", tmp_path / "auth.json")