Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

把多个 Codex 订阅变成一个内部号池,让 [Multica](https://github.com/multica-ai/multica) 任务按人、按任务使用和审计公司额度。

**CHEK 正式环境不向员工分发永久 Key,只允许 Multica 任务中的 `mcodex` 申请短时、任务绑定凭证;你自己的 `codex` 不受影响。**
**CHEK 正式环境提供两条受管路径:Multica 任务使用运行期绑定凭证;员工也可为官方 Codex GUI 申请无固定到期日、但可即时撤销且会话自动同步的个人凭证。**

```
codex → provider: openai → 你的个人订阅
Expand Down Expand Up @@ -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

### 依赖的三个上游机制

Expand Down Expand Up @@ -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
Expand Down
56 changes: 55 additions & 1 deletion corp_codex_pool/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

设计原则:
- 所有会改变状态的命令都支持 --dry-run,且默认打印将要发生的变更。
- 正式员工入口不分发永久密钥;旧管理员命令输出的密钥其余场合一律打码。
- 正式员工只拿到网关包装凭证,不接触底层号池密钥;旧管理员命令输出的密钥其余场合一律打码。
- 每一步失败都给出可执行的下一步,而不是只报错。
"""

Expand All @@ -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 (
Expand Down Expand Up @@ -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)
Expand Down
72 changes: 65 additions & 7 deletions corp_codex_pool/codex_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

from __future__ import annotations

import json
import os
import re
import shutil
Expand Down Expand Up @@ -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"
Expand All @@ -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}"
)
Expand Down Expand Up @@ -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("")
Expand Down Expand Up @@ -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},"
Expand Down Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -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 = [
Expand Down
53 changes: 53 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
@@ -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
26 changes: 26 additions & 0 deletions tests/test_codex_config.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import json
import tomllib

import pytest
Expand All @@ -13,6 +14,7 @@
render_block,
strip_block,
verify,
write_openai_auth_key,
)

# 取自真实宿主配置的形状:顶层键在前,[projects.*] 在后
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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")
Loading