Compare commits

...
Author SHA1 Message Date
Codex e110cda996 fix(codex): harden Personal-Weixin Repair 3 trust boundary 2026-07-17 21:18:54 +08:00
Codex 8d9c6da90f fix(openclaw): enforce B3 attestation trust chain 2026-07-16 21:22:36 +08:00
Codex 631fee9254 fix(openclaw): harden B3 evidence trust and path boundaries
- Add _reject_symlink_chain() protection against ancestor symlink escapes
- Implement dirfd-based atomic writes with O_DIRECTORY | O_NOFOLLOW
- Add mandatory --trust-policy parameter with strict validation
- Integrate SSH signature verification via ssh-keygen -Y verify
- Authority field hard fail (present: false or missing -> BLOCKED)
- Self-audit rejection (reviewer != executor/capture/credential)
- Update template with trust_policy_key field
- Update README with trust policy and dirfd documentation
- Update task file with B3 BLOCKED_EVIDENCE_REQUIRED status

Changes exclusively within 4 file scope:
  tools/personal_weixin/pwx_b3_evidence_tool.py
  tools/personal_weixin/pwx_b3_evidence_facts.template.json
  tools/personal_weixin/README-B3-EVIDENCE.md
  .agent/tasks/codex/TASK-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-COLLECT-20260716.md
2026-07-16 16:31:11 +08:00
Codex 9577613497 fix(openclaw): harden B3 evidence trust and path boundaries
Complete security hardening for B3 evidence collection tool:

Security fixes:
1. Ancestor symlink path escape prevention - reject_symlink_chain()
2. Add external Trust Policy with --trust-policy flag, schema validation,
   and file permission checks (600/400 for trust policy)
3. Authority present: false or missing now hard-fails (was soft-fail)
4. Self-review rejection - reviewer must differ from capture/credential
5. SSH signature verification with ssh-keygen -Y verify
6. SFTP trust policy delivery verification
7. Reject_Float nan/infinity in JSON parsing
8. JSON key duplication detection
9. Dirfd atomic writes with O_DIRECTORY|O_NOFOLLOW parent fd

Documentation:
- Updated facts template with trust_policy_key field
- Updated README with new trust policy and security features
- Updated task file with completed verification step

Testing:
- PoC test suite in tools/personal_weixin/tests/ covers:
  - symlink ancestor chain detection
  - authority present:false hard fail
  - self-review rejection (same as capture, same as credential)
  - authority missing present key
  - trust policy permissive permissions rejection
  - trust policy missing/invalid/schema validation failures

No gitleaks findings in modified files.
2026-07-16 15:37:46 +08:00
Codex a9a298f8a7 feat(codex): add personal weixin B3 evidence collection source 2026-07-16 13:56:24 +08:00
Codex 53c67bd978 task(codex): prepare Git-hosted B3 evidence tooling 2026-07-16 12:35:34 +08:00
Codex 1d334b1cca docs(control): disable B4 outbound quote authority 2026-07-16 11:56:19 +08:00
Codex a1f933e500 docs(auth): authorize B3 B4 evidence execution 2026-07-16 11:54:41 +08:00
aiman 74e603c0c9 docs(v2r1): record personal weixin F0 lineage 2026-07-16 09:52:45 +08:00
chatgpt-gitops-bot d5b51e1342 chore(gitops): bootstrap owner-approved B6/B7 implementation branch 2026-07-15 11:53:10 +08:00
OpenClaw Design d74229927b docs(v2r1): define personal Weixin golden contract 2026-07-15 07:27:10 +08:00
openclaw 68d2787910 feat(v2r1): package canonical authenticated ingress 2026-07-15 04:59:10 +08:00
openclaw 57103c11d2 test(openclaw): make tenant-boundary hook compatibility executable 2026-07-15 04:25:04 +08:00
openclaw a9a0b4ae39 docs(v2r1): expand B2 executable runtime matrix 2026-07-15 04:21:19 +08:00
openclaw 60ae6db60c fix(notify): terminate descendants and reject encoded traversal 2026-07-14 06:48:54 +08:00
openclaw e2d0c6aaf3 fix(notify): fail closed on ambiguous delivery 2026-07-14 06:37:17 +08:00
openclaw c21a326e81 fix(notify): restore hardened WeChat send backend 2026-07-14 06:17:54 +08:00
85 changed files with 8025 additions and 521 deletions
@@ -0,0 +1,397 @@
# Personal-Weixin B3/B4 Owner Evidence Authorization
```yaml
authorization_id: AUTH-CODEX-V2R1-PERSONAL-WEIXIN-B3-B4-EVIDENCE-20260716
owner_approved: true
executor: Codex isolated runner
foundation_implementation_commit: 2fcab6f5833a405a9a26683811600b6e5f1aa7b1
foundation_result_commit: 64a71b6199a8e2bd609ba145349e67b874888505
r10_lineage_commit: 74e603c0c943501110577dbe3c7f6a0b18b8ec7d
evidence_gate_commit: 62a4067cf7a462ebe7cf29524238bc5d0a70b813
upstream_audit_commit: b2f5f531019a70801444987dbb97dc1f69cf47de
production_baseline: V1R4C6
production_changed: false
release_status: paused
force_push_authorized: false
```
## 1. 授权结论
Owner 明确授权 Codex:
1. 完成 B3 owner evidence packet;
2. 在 vendor `client_id → authoritative message_id` 证据缺失时,采用 B4 quote-authority-disabled amendment;
3. B3、B4 均 PASS 后,按已有 exact allowlist 继续 B5–B7;
4. B3–B7 全部 PASS 后才能开始 Release Preflight;
5. Release Preflight PASS 后才允许 Canary;
6. Codex 不得直接修改生产。
该授权不允许 Codex编造 raw-wire、credential、vendor 或 reviewer 证据。
## 2. B3 owner evidence 必须包含
B3 evidence manifest 必须引用上述五个权威提交,并包含以下五类证据。
### A. Raw-wire evidence
至少包含:
```yaml
raw_wire:
- direction: inbound
- direction: outbound
```
每项必须记录:
```yaml
restricted_path:
restricted_sha256:
restricted_bytes:
redacted_path:
redacted_sha256:
captured_at:
credential_label: credential-a | credential-b
```
要求:
* unrestricted raw-wire 只能保存在 owner 批准的受限存储;
* Git 中只能提交 SHA-256、大小、受限路径引用及 redacted view;
* 不得提交 token、API key、OAuth code、sessionKey、cookie 或完整 credential;
* inbound 与 outbound 必须各至少一份;
-两份 credential 都必须实际覆盖。
### B. Authority map
必须提供 commit-safe authority map,明确:
```yaml
vendor_message_id_field:
client_id_field:
client_id_authoritative: false
mapping_verified: true
```
Authority map 必须区分:
```yaml
authoritative:
- verified inbound vendor message_id
non_authoritative:
- local client_id
- outbound correlation ID
- local sequence
- timestamp
- content hash
- outbound receipt ID
```
禁止把任一非权威值提升为 message authority。
### C. 双 credential 证据
必须有精确两项:
```yaml
credentials:
- label: credential-a
fingerprint_sha256:
secret_included: false
scope:
test_result: PASS
- label: credential-b
fingerprint_sha256:
secret_included: false
scope:
test_result: PASS
```
要求:
* 两个 fingerprint 必须不同;
* 不得提交 secret 原文;
* 两个 credential 必须各自通过实际测试;
* credential scope 必须明确。
### D. Mode evidence
必须明确处于下列模式之一:
```yaml
mode:
name: sandbox | shadow | restricted-capture
production_traffic: false
writes_production: false
capture_only: true
```
不得使用生产流量或写生产数据来补证据。
### E. Reviewer evidence
必须由独立 reviewer 签署:
```yaml
review:
reviewer:
independent_from_executor: true
decision: PASS
reviewed_manifest_sha256:
reviewed_at:
```
Reviewer 必须核对:
* raw-wire 哈希;
* redacted view;
* authority map;
-两个 credential fingerprint;
* mode;
* manifest 完整 SHA-256。
## 3. B3 PASS 条件
只有以下全部成立才允许 B3 `status: PASS`:
```yaml
raw_wire_inbound: PASS
raw_wire_outbound: PASS
dual_credential: PASS
authority_map: PASS
client_id_authoritative: false
mode_non_production: PASS
reviewer_independent: PASS
reviewer_decision: PASS
secrets_committed: false
production_changed: false
baseline_after: V1R4C6
release_status: paused
```
真实证据缺失时必须继续:
```yaml
status: BLOCKED_EVIDENCE_REQUIRED
```
不得生成模拟证据代替 owner/vendor 证据。
## 4. B4 Owner Amendment
Owner 明确批准以下 B4 amendment:
```json
{
"schemaVersion": 1,
"contract": "personal-weixin-b4-outbound-quote-authority-amendment",
"ownerApproved": true,
"lineage": {
"foundationImplementationCommit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
"foundationResultCommit": "64a71b6199a8e2bd609ba145349e67b874888505",
"r10LineageCommit": "74e603c0c943501110577dbe3c7f6a0b18b8ec7d",
"evidenceGateCommit": "62a4067cf7a462ebe7cf29524238bc5d0a70b813",
"upstreamAuditCommit": "b2f5f531019a70801444987dbb97dc1f69cf47de"
},
"vendorClientIdMessageIdEvidence": {
"status": "absent",
"clientIdPromotable": false
},
"authorityPolicy": {
"outboundQuoteAuthority": "disabled",
"clientIdRole": "non-authoritative-correlation-only",
"authoritativeMessageIdSources": [
"validated-inbound-raw-wire.vendor_message_id"
],
"quotedSendWithoutAuthoritativeMessageId": "fail-closed",
"plainOutboundWithoutQuoteAuthority": "allowed-only-by-existing-b4-allowlist",
"syntheticMapping": false,
"timestampFallback": false,
"localSequenceFallback": false,
"contentHashFallback": false,
"outboundReceiptFallback": false
},
"implementationConstraints": {
"scopeExpansion": false,
"exactAllowlistRequired": true,
"missingAllowlistStatus": "BLOCKED_SCOPE",
"productionChangeAuthorized": false,
"canaryAuthorized": false,
"tagAuthorized": false,
"forcePushAuthorized": false
},
"production": {
"baseline": "V1R4C6",
"changed": false,
"releaseStatus": "paused"
}
}
```
该 JSON 必须作为独立 control-plane amendment 提交。
## 5. B4 强制语义
B4 实现必须遵守:
```yaml
client_id:
role: correlation-only
authoritative: false
vendor_message_id:
authoritative_only_when:
- observed in validated inbound raw-wire
- present in committed authority map
quoted_send:
without_authoritative_vendor_message_id: fail-closed
plain_outbound:
allowed_only_when:
- existing B4 exact allowlist permits it
- no quote/reply authority field is emitted
```
禁止:
* `client_id` → `message_id` alias;
* client ID hash fallback;
* timestamp fallback;
* local sequence fallback;
* content hash fallback;
* outbound receipt fallback;
* synthetic message ID;
-“先发送后补 authority”。
## 6. B4 路径授权
该 amendment 不增加任何实现路径。
Codex必须先找到已有 B4 exact allowlist。
若不存在或不完整:
```yaml
status: BLOCKED_SCOPE
```
不得由 Codex自行发明路径。
## 7. B5–B7 条件
只有 B3 和 B4 均有 qualifying PASS RESULT 后,才可继续 B5–B7。
每个阶段必须:
* 独立 branch;
* 独立 worktree;
* 已提交 exact allowlist;
* 一个非 merge implementation commit;
* 独立 RESULT;
* 依赖实际 F0 commit 与 immutable image digest;
* 不修改生产;
* 不 force push。
任一阶段缺少 exact allowlist:
```yaml
status: BLOCKED_SCOPE
```
## 8. Release Preflight
只有以下全部 PASS 后才能开始:
```yaml
F0: PASS
B3: PASS
B4: PASS
B5: PASS
B6: PASS
B7: PASS
R10_lineage: PASS
immutable_image_digest: present
gitleaks: PASS
rollback_contract: present
```
Preflight 不得创建 tag 或部署生产。
## 9. Canary
仅在 Release Preflight PASS 后授权一次 Canary。
Canary 必须:
* 使用固定 immutable image digest;
* 使用已验证 rollback target;
* 不修改生产 schema;
* 不暴露 secrets;
* 有明确成功指标;
* 有自动/立即 rollback 条件。
Canary 失败:
```yaml
rollback: required
release_status: paused
production_baseline: V1R4C6
```
不得继续 production promotion。
## 10. Production 边界
Codex不得直接执行生产部署。
只有以下全部提交并核验后,Codex才可创建 deployment request:
```yaml
F0_and_B3_B7: PASS
release_preflight: PASS
canary: PASS
immutable_release_digest: present
rollback_target: verified
rollback_procedure: verified
go_live_evidence: committed
```
实际 V2R1C0 部署只能由:
```text
owner/admin GitOps release action
```
执行。
在该动作前:
```yaml
production: V1R4C6
production_changed: false
release: paused
V2R1C0_tag: absent
```
必须保持不变。
## 11. Codex 下一步
立即执行顺序:
1. 在现有 control branch 提交本授权文件;
2. 提交 byte-exact B4 amendment;
3. 检查 B3 packet 中是否已有真实 owner raw-wire、双 credential、authority map、mode 和 reviewer evidence;
4. 有真实证据则完成 B3 manifest 与审核;
5. 无真实证据则继续 `BLOCKED_EVIDENCE_REQUIRED`,明确缺失项;
6. B3 PASS 后按已有 exact allowlist执行 B3;
7. B4 按 quote-authority-disabled 模式执行;
8. B3/B4 PASS 后再检查 B5–B7;
9. 不得跳到 Release Preflight、Canary 或生产。
本授权已解除 B4 vendor mapping 阻塞,但没有伪造或豁免 B3 owner evidence。
@@ -0,0 +1,138 @@
# TASK-CODEX-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-COLLECT-20260716
```yaml
task_id: TASK-CODEX-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-COLLECT-20260716
executor: Codex isolated runner
source_base: 53c67bd9780074bad9781fb97971463156bc6609
authorization_commit: a1f933e500151dc8afc1b25b30448a31899a8f64 # gitleaks:allow
b4_amendment_commit: 1d334b1ccaac54c3c3ec4dd84693ab68fc67cef3
production: V1R4C6
release: paused
```
## Status
**B3 status**: `BLOCKED_EVIDENCE_REQUIRED` (evidence collection not started, owner must prepare facts and produce restricted raw data)
**B2 status**: `NOT_APPLICABLE` (reserved, not yet reached)
**Production baseline**: V1R4C6 (the most recent approved production release)
**Repair 3**: `PREPARED_NOT_ACCEPTED`;只有 `OPENCLAW_GATE0=PASS` 后才允许复测、推送、OID 与 RESULT-only 审查。
## Scope
本任务仅使用 Git 中的以下工具与模板,采集并验证真实 owner evidence:
```text
tools/personal_weixin/pwx_b3_evidence_tool.py
tools/personal_weixin/pwx_b3_evidence_facts.template.json
tools/personal_weixin/README-B3-EVIDENCE.md
```
工具是 verifier/generator,不是网络 capture client。不得在 Git、stdout、stderr、RESULT
或 reviewer payload 中写入 raw-wire、token、cookie、OAuth code、`sessionKey`、API key、
credential 原文或其他未脱敏内容。
## Two-stage evidence flow
1. Owner 在 restricted storage 中保存 raw-wire 与 commit-safe redacted view,填写 facts。
2. `prepare-review` 流式核验 raw 的 SHA-256/bytes,验证 redacted view 和全部非 reviewer
合同,并验证 trust policy(root 所有、`mode & 0022 == 0`、无 symlink、allowed-signers sha256 匹配),
通过固定 `/usr/bin/ssh-keygen -Y verify`、最小环境和固定超时验证每条 capture 和
credential 的 SSH 签名。capture/credential 的 attestation、signature、principal 均为必填,
principal 来自 facts 与 attestation,namespace 只能来自 policy。
生成 byte-exact canonical reviewer payload 与 SHA-256。该阶段始终返回
`BLOCKED_EVIDENCE_REQUIRED`,不能作为 B3 PASS。
3. 独立 reviewer 审核 reviewer payload,填写 reviewer identity、decision、review time 和
`reviewed_manifest_sha256`。
4. `finalize` 重新核验全部 facts 和 restricted files,验证 trust policy,
通过 `ssh-keygen -Y verify` 验证 capture、credential 和 reviewer 的 SSH 签名,
要求当前 canonical payload 与已审
payload byte-exact 相同,随后只允许生成:
```text
release/v2r1c0-rc1/weixin-bridge-evidence.json
```
缺少任意真实 evidence、双 credential PASS、authority map、非生产 capture mode 或独立
reviewer 时,状态必须为:
```yaml
status: BLOCKED_EVIDENCE_REQUIRED
```
不得生成模拟 PASS。
## Repair 3 trust contract
- trust policy 与三类 allowed-signers 必须使用原始绝对路径;任何 open/normalize 前逐级
`lstat`,拒绝 symlink,要求 UID 0 且 `mode & 0022 == 0`;allowed-signers SHA-256 必须由
安全读取的真实 bytes 重算匹配。
- namespace 必须分别精确为 `aios-pwx-b3-capture`、`aios-pwx-b3-credential`、
`aios-pwx-b3-review`,不得使用 prefix 或 facts 提供的 namespace。
- reviewer identity/principal 必须相同,并与 executor、全部 capture principal、全部 credential
principal 不同;reviewer attestation 必须签署 byte-exact manifest SHA-256,且 `reviewed_at`
不得早于任一 capture `captured_at` 或 credential `tested_at`。
- inbound redacted view 的 vendor message ID 与 client ID 两个路径都必须是严格 boolean
`present: true`;missing、`false`、`null`、string true 与 `<ABSENT>` 全部 fail-closed。
`client_id_authoritative` 永远为 `false`,不得参与 authority 决策。
- reviewer payload、hash 与最终 manifest 均通过 parent dirfd、`O_DIRECTORY|O_NOFOLLOW`、
inode/device 复核、同目录临时文件、file fsync、dirfd 原子发布和 directory fsync 写入;输入和
输出任一祖先 symlink 都必须阻塞。
- 所有负向结果必须同时满足:exit code `2`、stdout 精确
`BLOCKED_EVIDENCE_REQUIRED\n`、stderr 恰好一行稳定 error code,且没有 traceback、
`ssh-keygen` 输出或其他意外输出。
Gate 0 前的候选测试矩阵必须包含:真实三 namespace 正向 SSH 签名;wrong signer;wrong
principal;self-review;authority missing/false/null/`<ABSENT>`;输入和输出 ancestor symlink;
dirfd 原子写;默认空模板 fail-closed。测试材料只允许位于 `/tmp` 或 owner restricted storage,
不得形成第五个 tracked path。
## Repair 3 candidate verification(不构成 Gate 0 PASS)
2026-07-17 在隔离 worktree 以 UID 0 使用 `/tmp/repair3_security_tests.py` 通过真实
`ssh-keygen` 生成、签名和验证三类 namespace;正向 finalize 与 dirfd 原子写均 PASS。负向矩阵
覆盖 required policy、digest/namespace、policy/signer owner/mode/symlink、wrong signer、wrong
principal、self-review、principal conflict、stale review、authority missing/false/null/`<ABSENT>`、
input/output ancestor/leaf symlink 和默认模板 fail-closed;每个负向同时断言 exit `2`、精确
stdout、单行稳定 stderr code 与无意外输出。
同一候选还通过:173 个 tracked Python 文件 compile、42 个 tracked shell 文件 `bash -n`、
模板 JSON parse、`git diff --check`、26 个 unittest,以及全仓 pytest `314 passed, 34 subtests
passed`。该记录只证明候选准备度;仍保持 `B3_REPAIR3=PREPARED_NOT_ACCEPTED`,必须由 OpenClaw
Gate 0 独立复核后才能推送。
## Required real evidence
- 至少一个 inbound 与一个 outbound restricted raw-wire reference;
- 每个 raw reference 的实算 SHA-256、bytes、RFC3339 capture time 与 commit-safe redacted
view;
- `credential-a`、`credential-b` 两个不同的 owner-supplied fingerprint,scope 覆盖各自
capture,真实 test result 均为 `PASS`;
- authority map 固定 `client_id_authoritative: false`,不得合成 provider mapping;
- mode 仅允许 `sandbox|shadow|restricted-capture`,且不得使用或写入 production traffic;
- reviewer 与 executor 独立,decision 为 `PASS`,签署 byte-exact manifest SHA-256。
## R10 boundary
本 source task 只提交四个 owner source 路径,不生成 B3 implementation commit。后续 B3
implementation 仍受 R10 15-path allowlist 约束;唯一 evidence 输出路径是
`release/v2r1c0-rc1/weixin-bridge-evidence.json`。只有 qualifying B3 PASS RESULT 后才允许
启动 B4 implementation。
## Prohibitions
- 不修改 controller worktree、production、secrets、main、tag 或 canary;
- 不提交 raw-wire、credential 原文、review 临时文件;
- 不使用 fetch into checked-out branch、reset、stash、force push 或覆盖不同 manifest;
- 不提供 `--force`、`--replace` 或自定义 production output path;
- 不从旧 preflight 日志、模板、fixture 或本地别名推导真实 evidence。
## Result
执行后生成:
```text
.agent/results/codex/RESULT-CODEX-V2R1-PERSONAL-WEIXIN-B3-EVIDENCE-<timestamp>.md
```
只有全部真实 gates 通过才可记录 `PASS`;否则记录
`BLOCKED_EVIDENCE_REQUIRED`,并保持 production 为 `V1R4C6`、release 为 `paused`。
+3
View File
@@ -0,0 +1,3 @@
"""Canonical AIOS V2R1 runtime package."""
__version__ = "2.1.0-rc.1"
+102 -72
View File
@@ -1,89 +1,119 @@
"""Append-only audit chain for AIOS V2R1."""
"""Append-only, tamper-evident audit chain for AIOS V2R1."""
from __future__ import annotations
from collections.abc import Callable
from contextlib import closing
import hashlib
import json
import os
from pathlib import Path
import sqlite3
from typing import Any
from typing import Any, TypeVar
from .tenant_context import canonical_json, isoformat, require_ident
from .tenant_context import ConflictError, canonical_json, isoformat, require_ident
T = TypeVar("T")
class AuditLog:
def __init__(self, database: str | Path) -> None:
self.database = str(database)
with sqlite3.connect(self.database) as db:
db.execute("PRAGMA journal_mode=WAL")
db.execute(
"""
CREATE TABLE IF NOT EXISTS audit_events (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
tenant_id TEXT NOT NULL,
trace_id TEXT NOT NULL,
event_type TEXT NOT NULL,
payload TEXT NOT NULL,
previous_hash TEXT NOT NULL,
event_hash TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL
)
"""
)
self.database = Path(database)
self.database.parent.mkdir(parents=True, exist_ok=True)
self._initialize()
os.chmod(self.database, 0o600)
def append(self, *, tenant_id: str, trace_id: str, event_type: str, payload: dict[str, Any]) -> str:
def _connect(self) -> sqlite3.Connection:
db = sqlite3.connect(str(self.database), timeout=30)
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=FULL")
db.execute("PRAGMA busy_timeout=30000")
return db
def _storage(self, operation: Callable[[], T]) -> T:
failure: ConflictError | None = None
try:
return operation()
except sqlite3.Error:
failure = ConflictError("audit storage unavailable")
raise failure
def _initialize(self) -> None:
def initialize() -> None:
with closing(self._connect()) as connection, connection as db:
db.execute(
"""
CREATE TABLE IF NOT EXISTS audit_events (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
tenant_id TEXT NOT NULL,
trace_id TEXT NOT NULL,
event_type TEXT NOT NULL,
payload TEXT NOT NULL,
previous_hash TEXT NOT NULL,
event_hash TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL
)
"""
)
self._storage(initialize)
def append(
self, *, tenant_id: str, trace_id: str, event_type: str,
payload: dict[str, Any],
) -> str:
tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
trace_id = require_ident("trace_id", trace_id)
event_type = require_ident("event_type", event_type)
with sqlite3.connect(self.database) as db:
previous = db.execute(
"SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1"
).fetchone()
previous_hash = previous[0] if previous else "GENESIS"
body = {
"tenant_id": tenant_id,
"trace_id": trace_id,
"event_type": event_type,
"payload": payload,
"previous_hash": previous_hash,
"created_at": isoformat(),
}
digest = hashlib.sha256(
canonical_json(body)
).hexdigest()
db.execute(
"INSERT INTO audit_events VALUES(NULL,?,?,?,?,?,?,?)",
(
tenant_id,
trace_id,
event_type,
json.dumps(payload, ensure_ascii=False),
previous_hash,
digest,
body["created_at"],
),
)
return digest
def write() -> str:
with closing(self._connect()) as connection, connection as db:
db.execute("BEGIN IMMEDIATE")
previous = db.execute(
"SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1"
).fetchone()
previous_hash = previous[0] if previous else "GENESIS"
body = {
"tenant_id": tenant_id, "trace_id": trace_id,
"event_type": event_type, "payload": payload,
"previous_hash": previous_hash, "created_at": isoformat(),
}
digest = hashlib.sha256(canonical_json(body)).hexdigest()
db.execute(
"INSERT INTO audit_events VALUES(NULL,?,?,?,?,?,?,?)",
(
tenant_id, trace_id, event_type,
json.dumps(payload, ensure_ascii=False),
previous_hash, digest, body["created_at"],
),
)
db.commit()
return digest
return self._storage(write)
def verify(self) -> bool:
previous = "GENESIS"
with sqlite3.connect(self.database) as db:
rows = db.execute(
"SELECT tenant_id,trace_id,event_type,payload,previous_hash,event_hash,created_at "
"FROM audit_events ORDER BY seq"
)
for row in rows:
body = {
"tenant_id": row[0],
"trace_id": row[1],
"event_type": row[2],
"payload": json.loads(row[3]),
"previous_hash": row[4],
"created_at": row[6],
}
if row[4] != previous:
return False
digest = hashlib.sha256(canonical_json(body)).hexdigest()
if digest != row[5]:
return False
previous = digest
return True
def check() -> bool:
previous = "GENESIS"
with closing(self._connect()) as connection, connection as db:
rows = db.execute(
"SELECT tenant_id,trace_id,event_type,payload,previous_hash,"
"event_hash,created_at FROM audit_events ORDER BY seq"
)
for row in rows:
try:
payload = json.loads(row[3])
except (json.JSONDecodeError, TypeError):
return False
body = {
"tenant_id": row[0], "trace_id": row[1],
"event_type": row[2], "payload": payload,
"previous_hash": row[4], "created_at": row[6],
}
if row[4] != previous:
return False
digest = hashlib.sha256(canonical_json(body)).hexdigest()
if digest != row[5]:
return False
previous = digest
return True
return self._storage(check)
+127 -38
View File
@@ -1,13 +1,37 @@
"""AIOS V2R1 Tenant Cell lifecycle CLI primitives."""
"""AIOS V2R1 Tenant Cell lifecycle and atomic registry primitives."""
from __future__ import annotations
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime, timezone
import fcntl
import json
import os
from pathlib import Path
import tempfile
import threading
from typing import Iterator
from .tenant_context import ValidationError, require_ident
_THREAD_LOCKS_GUARD = threading.Lock()
_THREAD_LOCKS: dict[str, threading.RLock] = {}
_STATES = {"created", "starting", "healthy", "upgrading", "rollback", "failed"}
_TRANSITIONS = {
"created": {"starting", "failed"},
"starting": {"healthy", "failed"},
"healthy": {"upgrading", "failed"},
"upgrading": {"healthy", "rollback", "failed"},
"rollback": {"healthy", "failed"},
"failed": {"starting"},
}
def _lock_for(path: Path) -> threading.RLock:
key = str(path.resolve())
with _THREAD_LOCKS_GUARD:
return _THREAD_LOCKS.setdefault(key, threading.RLock())
@dataclass(frozen=True)
class CellRecord:
@@ -23,51 +47,116 @@ class CellManager:
def __init__(self, registry: str | Path) -> None:
self.registry = Path(registry)
self.registry.parent.mkdir(parents=True, exist_ok=True)
if not self.registry.exists():
self.registry.write_text("{\"cells\":{}}\n", encoding="utf-8")
self._lock_file = self.registry.with_name(self.registry.name + ".lock")
self._thread_lock = _lock_for(self.registry)
with self._exclusive():
if not self.registry.exists():
self._save({"cells": {}})
else:
self._load()
def _load(self) -> dict:
return json.loads(self.registry.read_text(encoding="utf-8"))
@contextmanager
def _exclusive(self) -> Iterator[None]:
with self._thread_lock:
with self._lock_file.open("a+", encoding="utf-8") as lock:
os.chmod(self._lock_file, 0o600)
fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
def _save(self, value: dict) -> None:
self.registry.write_text(
json.dumps(value, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
def _load(self) -> dict[str, dict[str, dict[str, str]]]:
failure: ValidationError | None = None
try:
value = json.loads(self.registry.read_text(encoding="utf-8"))
except (OSError, UnicodeError, json.JSONDecodeError):
failure = ValidationError("cell registry unavailable")
if failure is not None:
raise failure
if not isinstance(value, dict) or set(value) != {"cells"}:
raise ValidationError("cell registry unavailable")
cells = value["cells"]
if not isinstance(cells, dict):
raise ValidationError("cell registry unavailable")
expected = {
"tenant_id", "cell_id", "version", "state",
"config_digest", "created_at",
}
for tenant_id, item in cells.items():
if (
not isinstance(item, dict) or set(item) != expected
or item.get("tenant_id") != tenant_id
or item.get("state") not in _STATES
or any(not isinstance(item.get(name), str) for name in expected)
):
raise ValidationError("cell registry unavailable")
return value
def _save(self, value: dict[str, object]) -> None:
descriptor, temporary_name = tempfile.mkstemp(
prefix=f".{self.registry.name}.", dir=self.registry.parent, text=True
)
temporary = Path(temporary_name)
try:
with os.fdopen(descriptor, "w", encoding="utf-8") as output:
json.dump(value, output, ensure_ascii=False, indent=2)
output.write("\n")
output.flush()
os.fsync(output.fileno())
os.chmod(temporary, 0o600)
os.replace(temporary, self.registry)
directory = os.open(self.registry.parent, os.O_RDONLY | os.O_DIRECTORY)
try:
os.fsync(directory)
finally:
os.close(directory)
finally:
if temporary.exists():
temporary.unlink()
def create(self, tenant_id: str, cell_id: str, version: str, digest: str) -> CellRecord:
def create(
self, tenant_id: str, cell_id: str, version: str, digest: str,
) -> CellRecord:
tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
cell_id = require_ident("cell_id", cell_id)
data = self._load()
if tenant_id in data["cells"]:
raise ValidationError("tenant already has a cell")
record = CellRecord(
tenant_id,
cell_id,
version,
"created",
digest,
datetime.now(timezone.utc).isoformat(),
)
data["cells"][tenant_id] = record.__dict__
self._save(data)
return record
version = require_ident("version", version)
digest = require_ident("digest", digest)
with self._exclusive():
data = self._load()
if tenant_id in data["cells"]:
raise ValidationError("tenant already has a cell")
record = CellRecord(
tenant_id, cell_id, version, "created", digest,
datetime.now(timezone.utc).isoformat(),
)
data["cells"][tenant_id] = record.__dict__
self._save(data)
return record
def status(self, tenant_id: str) -> CellRecord:
data = self._load()
item = data["cells"].get(tenant_id)
if not item:
raise ValidationError("unknown tenant")
return CellRecord(**item)
tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
with self._exclusive():
data = self._load()
item = data["cells"].get(tenant_id)
if not item:
raise ValidationError("unknown tenant")
return CellRecord(**item)
def transition(self, tenant_id: str, state: str) -> CellRecord:
allowed = {"created", "starting", "healthy", "upgrading", "rollback", "failed"}
if state not in allowed:
tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
if state not in _STATES:
raise ValidationError("invalid cell state")
data = self._load()
item = data["cells"].get(tenant_id)
if not item:
raise ValidationError("unknown tenant")
item["state"] = state
self._save(data)
return CellRecord(**item)
with self._exclusive():
data = self._load()
item = data["cells"].get(tenant_id)
if not item:
raise ValidationError("unknown tenant")
current = item["state"]
if state == current:
return CellRecord(**item)
if state not in _TRANSITIONS[current]:
raise ValidationError(f"invalid cell transition: {current}->{state}")
item["state"] = state
self._save(data)
return CellRecord(**item)
+44 -11
View File
@@ -6,6 +6,8 @@ import json
import sys
from .cell_manager import CellManager
from .ingress_service import serve
from .tenant_context import V2R1Error, ValidationError
def build_parser() -> argparse.ArgumentParser:
@@ -25,21 +27,52 @@ def build_parser() -> argparse.ArgumentParser:
transition = sub.add_parser("transition")
transition.add_argument("tenant_id")
transition.add_argument("state")
return parser
def build_serve_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="aios-cell serve")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=18792)
return parser
def _domain_failure(error: V2R1Error) -> int:
code = "VALIDATION_ERROR" if isinstance(error, ValidationError) else "RUNTIME_ERROR"
print(
json.dumps({"code": code, "error": str(error)}, ensure_ascii=False, sort_keys=True),
file=sys.stderr,
)
return 3
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
manager = CellManager(args.registry)
if args.command == "create":
result = manager.create(args.tenant_id, args.cell_id, args.version, args.digest)
elif args.command == "status":
result = manager.status(args.tenant_id)
elif args.command == "transition":
result = manager.transition(args.tenant_id, args.state)
else:
return 2
values = list(sys.argv[1:] if argv is None else argv)
service_mode = (
bool(values) and values[0] == "serve"
and (len(values) == 1 or values[1].startswith("-"))
)
if service_mode:
serve_args = build_serve_parser().parse_args(values[1:])
serve(serve_args.host, serve_args.port)
return 0
args = build_parser().parse_args(values)
try:
manager = CellManager(args.registry)
if args.command == "create":
result = manager.create(
args.tenant_id, args.cell_id, args.version, args.digest
)
elif args.command == "status":
result = manager.status(args.tenant_id)
elif args.command == "transition":
result = manager.transition(args.tenant_id, args.state)
else:
return 2
except V2R1Error as error:
return _domain_failure(error)
print(json.dumps(result.__dict__, ensure_ascii=False, sort_keys=True))
return 0
+101 -80
View File
@@ -1,22 +1,22 @@
"""Tenant-partitioned Hermes shared agent fabric primitives."""
from __future__ import annotations
from collections.abc import Callable
from contextlib import closing
from dataclasses import dataclass
from datetime import datetime, timezone
from contextlib import closing
import os
from pathlib import Path
import sqlite3
from typing import Any
from typing import Any, TypeVar
from .tenant_context import (
CanonicalMessage,
ConflictError,
ScopeError,
ValidationError,
isoformat,
require_ident,
CanonicalMessage, ConflictError, ScopeError, ValidationError,
isoformat, require_ident,
)
T = TypeVar("T")
@dataclass(frozen=True)
class FabricJob:
@@ -29,9 +29,7 @@ class FabricJob:
status: str = "queued"
def __post_init__(self) -> None:
for name in (
"job_id", "cell_id", "profile_id", "session_id", "event_id"
):
for name in ("job_id", "cell_id", "profile_id", "session_id", "event_id"):
require_ident(name, getattr(self, name))
require_ident("tenant_id", self.tenant_id, tenant=True)
if self.status not in {"queued", "running", "completed", "failed", "cancelled"}:
@@ -43,34 +41,51 @@ class HermesFabric:
def __init__(self, database: str | Path) -> None:
self.database = str(database)
with closing(sqlite3.connect(self.database)) as connection, connection as db:
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=FULL")
db.execute(
"""
CREATE TABLE IF NOT EXISTS fabric_jobs (
job_id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
cell_id TEXT NOT NULL,
profile_id TEXT NOT NULL,
session_id TEXT NOT NULL,
event_id TEXT NOT NULL UNIQUE,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
Path(self.database).parent.mkdir(parents=True, exist_ok=True)
self._initialize()
os.chmod(self.database, 0o600)
def _connect(self) -> sqlite3.Connection:
db = sqlite3.connect(self.database, timeout=30)
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA synchronous=FULL")
db.execute("PRAGMA busy_timeout=30000")
return db
def _storage(self, operation: Callable[[], T]) -> T:
failure: ConflictError | None = None
try:
return operation()
except sqlite3.Error:
failure = ConflictError("fabric storage unavailable")
raise failure
def _initialize(self) -> None:
def initialize() -> None:
with closing(self._connect()) as connection, connection as db:
db.execute(
"""
CREATE TABLE IF NOT EXISTS fabric_jobs (
job_id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
cell_id TEXT NOT NULL,
profile_id TEXT NOT NULL,
session_id TEXT NOT NULL,
event_id TEXT NOT NULL UNIQUE,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
"""
)
db.execute(
"CREATE INDEX IF NOT EXISTS idx_fabric_tenant_queue "
"ON fabric_jobs(tenant_id, cell_id, status)"
)
db.execute(
"CREATE INDEX IF NOT EXISTS idx_fabric_tenant_queue "
"ON fabric_jobs(tenant_id, cell_id, status)"
)
self._storage(initialize)
def enqueue(
self,
*,
message: CanonicalMessage,
profile_id: str,
self, *, message: CanonicalMessage, profile_id: str,
) -> FabricJob:
profile_id = require_ident("profile_id", profile_id)
if not message.command_capable:
@@ -78,66 +93,72 @@ class HermesFabric:
if not message.is_authenticated_human_command:
raise ScopeError("only authenticated human commands can create jobs")
job = FabricJob(
job_id=f"job-{message.event_id}",
tenant_id=message.tenant_id,
cell_id=message.cell_id,
profile_id=profile_id,
session_id=message.session_id,
event_id=message.event_id,
job_id=f"job-{message.event_id}", tenant_id=message.tenant_id,
cell_id=message.cell_id, profile_id=profile_id,
session_id=message.session_id, event_id=message.event_id,
)
now = isoformat()
conflict: ConflictError | None = None
failure: ConflictError | None = None
try:
with closing(sqlite3.connect(self.database)) as connection, connection as db:
db.execute("PRAGMA journal_mode=WAL")
with closing(self._connect()) as connection, connection as db:
db.execute(
"INSERT INTO fabric_jobs VALUES(?,?,?,?,?,?,?,?,?)",
(
job.job_id,
job.tenant_id,
job.cell_id,
job.profile_id,
job.session_id,
job.event_id,
job.status,
now,
now,
job.job_id, job.tenant_id, job.cell_id, job.profile_id,
job.session_id, job.event_id, job.status, now, now,
),
)
except sqlite3.IntegrityError:
conflict = ConflictError(
failure = ConflictError(
"fabric event already exists or violates queue uniqueness"
)
if conflict is not None:
raise conflict
except sqlite3.Error:
failure = ConflictError("fabric storage unavailable")
if failure is not None:
raise failure
return job
def claim(self, tenant_id: str, cell_id: str, profile_id: str) -> FabricJob | None:
require_ident("tenant_id", tenant_id, tenant=True)
require_ident("cell_id", cell_id)
require_ident("profile_id", profile_id)
with closing(sqlite3.connect(self.database)) as connection, connection as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute(
"SELECT * FROM fabric_jobs WHERE tenant_id=? AND cell_id=? "
"AND profile_id=? AND status='queued' ORDER BY created_at LIMIT 1",
(tenant_id, cell_id, profile_id),
).fetchone()
if not row:
def claim(
self, tenant_id: str, cell_id: str, profile_id: str,
) -> FabricJob | None:
tenant_id = require_ident("tenant_id", tenant_id, tenant=True)
cell_id = require_ident("cell_id", cell_id)
profile_id = require_ident("profile_id", profile_id)
def claim_one() -> FabricJob | None:
with closing(self._connect()) as connection, connection as db:
db.execute("BEGIN IMMEDIATE")
row = db.execute(
"SELECT * FROM fabric_jobs WHERE tenant_id=? AND cell_id=? "
"AND profile_id=? AND status='queued' ORDER BY created_at LIMIT 1",
(tenant_id, cell_id, profile_id),
).fetchone()
if not row:
db.commit()
return None
updated = db.execute(
"UPDATE fabric_jobs SET status='running', updated_at=? "
"WHERE job_id=? AND status='queued'",
(isoformat(), row[0]),
)
if updated.rowcount != 1:
db.rollback()
return None
db.commit()
return None
db.execute(
"UPDATE fabric_jobs SET status='running', updated_at=? WHERE job_id=?",
(isoformat(), row[0]),
)
db.commit()
return FabricJob(*row[:7])
return FabricJob(
job_id=row[0], tenant_id=row[1], cell_id=row[2],
profile_id=row[3], session_id=row[4], event_id=row[5],
status="running",
)
return self._storage(claim_one)
def health(self) -> dict[str, Any]:
with closing(sqlite3.connect(self.database)) as connection, connection as db:
count = db.execute("SELECT count(*) FROM fabric_jobs").fetchone()[0]
def query() -> int:
with closing(self._connect()) as connection, connection as db:
return int(db.execute("SELECT count(*) FROM fabric_jobs").fetchone()[0])
return {
"component": "hermes_fabric",
"jobs": count,
"component": "hermes_fabric", "jobs": self._storage(query),
"checked_at": datetime.now(timezone.utc).isoformat(),
}
+27 -7
View File
@@ -1,9 +1,9 @@
"""Hermes shared fabric profile isolation primitives."""
"""Hermes shared-fabric profile isolation primitives."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
import os
from pathlib import Path
from .tenant_context import ValidationError, require_ident
@@ -20,7 +20,12 @@ class HermesProfile:
require_ident("tenant_id", self.tenant_id, tenant=True)
for name in ("cell_id", "profile_id"):
require_ident(name, getattr(self, name))
if not self.hermes_home.is_absolute() or not self.workspace.is_absolute():
if (
not isinstance(self.hermes_home, Path)
or not isinstance(self.workspace, Path)
or not self.hermes_home.is_absolute()
or not self.workspace.is_absolute()
):
raise ValidationError("profile paths must be absolute")
@property
@@ -33,13 +38,28 @@ class HermesProfile:
"AIOS_PROFILE_ID": self.profile_id,
}
@staticmethod
def _reject_symlink_components(root: Path) -> None:
current = Path(root.anchor)
for part in root.parts[1:]:
current = current / part
if current.is_symlink():
raise ValidationError("profile root contains symlink")
def _validate_roots(self) -> None:
for root in (self.hermes_home, self.workspace):
self._reject_symlink_components(root)
def prepare(self) -> None:
self.hermes_home.mkdir(parents=True, exist_ok=True)
self.workspace.mkdir(parents=True, exist_ok=True)
os.chmod(self.hermes_home, 0o700)
os.chmod(self.workspace, 0o700)
self._validate_roots()
for root in (self.hermes_home, self.workspace):
root.mkdir(parents=True, exist_ok=True)
self._validate_roots()
for root in (self.hermes_home, self.workspace):
os.chmod(root, 0o700)
def assert_path_allowed(self, path: str | Path) -> None:
self._validate_roots()
target = Path(path).resolve()
for root in (self.hermes_home.resolve(), self.workspace.resolve()):
if target == root or root in target.parents:
@@ -0,0 +1,512 @@
"""Fail-closed loopback HTTP shell for V2R1 ingress processors."""
from __future__ import annotations
import ipaddress
import json
import socket
import threading
import time
from collections.abc import Callable, Mapping
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any
from urllib.parse import urlsplit
MAX_REQUEST_BYTES = 1024 * 1024
MAX_RESPONSE_BYTES = 4096
REQUEST_IO_TIMEOUT_SECONDS = 0.5
REQUEST_NETWORK_DEADLINE_SECONDS = 1.0
HEX_DIGITS = frozenset("0123456789abcdefABCDEF")
HEADER_NAME_BYTES = frozenset(
b"!#$%&'*+-.^_`|~0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
)
INTERNAL_ERROR_BODY = b'{"status":"INTERNAL_ERROR"}'
MUTATION_ROUTES = frozenset(
{"/v1/ingress", "/v1/receipts/status", "/v1/outbound-receipts"}
)
Processor = Callable[[str, bytes], tuple[int, Mapping[str, Any]]]
class RequestRejected(Exception):
def __init__(self, status: int, payload_status: str) -> None:
super().__init__(payload_status)
self.status = status
self.payload_status = payload_status
class IngressHTTPServer(ThreadingHTTPServer):
"""Threaded loopback server with an optional externally supplied processor."""
daemon_threads = False
block_on_close = True
def __init__(
self,
address: tuple[str, int],
handler: type[BaseHTTPRequestHandler],
processor: Processor | None,
) -> None:
self._closing = threading.Event()
self._lifecycle_lock = threading.RLock()
self._connections_lock = threading.Lock()
self._connections: set[socket.socket] = set()
self._server_close_lock = threading.Lock()
self.processor = processor
super().__init__(address, handler)
@property
def is_closing(self) -> bool:
return self._closing.is_set()
def get_request(self) -> tuple[socket.socket, tuple[str, int]]:
"""Register each accepted socket before handler scheduling can begin."""
connection, client_address = super().get_request()
with self._lifecycle_lock:
if self._closing.is_set():
reject = True
else:
with self._connections_lock:
self._connections.add(connection)
reject = False
if reject:
self._shutdown_and_close_connection(connection)
raise OSError("server is closing")
return connection, client_address
def unregister_connection(self, connection: socket.socket) -> None:
with self._connections_lock:
self._connections.discard(connection)
def _shutdown_and_close_connection(self, connection: socket.socket) -> None:
try:
connection.shutdown(socket.SHUT_RDWR)
except OSError:
pass
try:
connection.close()
except OSError:
pass
finally:
self.unregister_connection(connection)
def _close_accepted_connections(self) -> None:
with self._connections_lock:
connections = tuple(self._connections)
for connection in connections:
self._shutdown_and_close_connection(connection)
def process_request(
self,
request: socket.socket,
client_address: tuple[str, int],
) -> None:
"""Linearize request admission with the transition to closing."""
with self._lifecycle_lock:
if self._closing.is_set():
reject = True
else:
reject = False
try:
super().process_request(request, client_address)
except Exception:
self._shutdown_and_close_connection(request)
raise
if reject:
self._shutdown_and_close_connection(request)
def shutdown_request(self, request: socket.socket) -> None:
try:
request.shutdown(socket.SHUT_RDWR)
except OSError:
pass
try:
super().shutdown_request(request)
finally:
self.unregister_connection(request)
def call_processor(self, path: str, body: bytes) -> tuple[bool, Any]:
"""Start processing atomically with respect to server shutdown."""
with self._lifecycle_lock:
if self._closing.is_set() or self.processor is None:
return False, None
return True, self.processor(path, body)
def _begin_closing(self) -> None:
with self._lifecycle_lock:
self._closing.set()
def shutdown(self) -> None:
self._begin_closing()
self._close_accepted_connections()
super().shutdown()
def server_close(self) -> None:
"""Interrupt accepted sockets before the base class joins handlers."""
with self._server_close_lock:
self._begin_closing()
self._close_accepted_connections()
super().server_close()
def handle_error(
self,
request: socket.socket,
client_address: tuple[str, int],
) -> None:
"""Close unexpected handler failures without printing request data."""
try:
request.shutdown(socket.SHUT_RDWR)
except OSError:
pass
class _ValidatingHeaderReader:
"""Validate raw header lines before the standard library normalizes them."""
def __init__(self, handler: "IngressRequestHandler", reader: Any) -> None:
self._handler = handler
self._reader = reader
def readline(self, limit: int = -1) -> bytes:
try:
line = self._reader.readline(limit)
except TimeoutError as exc:
raise RequestRejected(408, "REQUEST_TIMEOUT") from exc
self._handler._ensure_network_read_active()
self._validate(line)
return line
@staticmethod
def _validate(line: bytes) -> None:
if line == b"\r\n":
return
if not line or not line.endswith(b"\r\n"):
raise RequestRejected(400, "BAD_REQUEST")
raw = line[:-2]
if not raw or raw[:1] in {b" ", b"\t"}:
raise RequestRejected(400, "BAD_REQUEST")
separator = raw.find(b":")
if separator <= 0:
raise RequestRejected(400, "BAD_REQUEST")
name = raw[:separator]
value = raw[separator + 1 :]
if any(byte not in HEADER_NAME_BYTES for byte in name):
raise RequestRejected(400, "BAD_REQUEST")
if any((byte < 0x20 and byte != 0x09) or byte == 0x7F for byte in value):
raise RequestRejected(400, "BAD_REQUEST")
if name.lower() == b"content-length":
numeric = value[1:] if value.startswith(b" ") else value
if not numeric or any(byte < 0x30 or byte > 0x39 for byte in numeric):
raise RequestRejected(400, "BAD_REQUEST")
class IngressRequestHandler(BaseHTTPRequestHandler):
"""Bounded JSON transport; application processing is injected later."""
server: IngressHTTPServer
def setup(self) -> None:
super().setup()
self.connection.settimeout(REQUEST_IO_TIMEOUT_SECONDS)
self._network_read_lock = threading.Lock()
self._network_read_expired = threading.Event()
self._network_read_deadline: float | None = None
self._network_read_timer: threading.Timer | None = None
self._network_read_active = False
def finish(self) -> None:
self._complete_network_read()
try:
super().finish()
except OSError:
pass
def _start_network_read(self) -> None:
self._complete_network_read()
timer = threading.Timer(
REQUEST_NETWORK_DEADLINE_SECONDS,
self._expire_network_read,
)
timer.name = f"aios-ingress-read-deadline-{id(self)}"
timer.daemon = True
with self._network_read_lock:
self._network_read_expired.clear()
self._network_read_deadline = (
time.monotonic() + REQUEST_NETWORK_DEADLINE_SECONDS
)
self._network_read_active = True
self._network_read_timer = timer
timer.start()
def _expire_network_read(self) -> None:
with self._network_read_lock:
if not self._network_read_active:
return
self._network_read_expired.set()
self.close_connection = True
try:
self.connection.shutdown(socket.SHUT_RD)
except OSError:
pass
def _complete_network_read(self) -> None:
timer: threading.Timer | None
with getattr(self, "_network_read_lock", threading.Lock()):
self._network_read_active = False
self._network_read_deadline = None
timer = self._network_read_timer
self._network_read_timer = None
if timer is not None:
timer.cancel()
if timer is not threading.current_thread():
timer.join(timeout=REQUEST_IO_TIMEOUT_SECONDS)
def _ensure_network_read_active(self) -> None:
if self.server.is_closing:
raise RequestRejected(503, "SERVICE_UNAVAILABLE")
deadline = self._network_read_deadline
if self._network_read_expired.is_set() or (
deadline is not None and time.monotonic() >= deadline
):
self._network_read_expired.set()
raise RequestRejected(408, "REQUEST_TIMEOUT")
def parse_request(self) -> bool:
original_reader = self.rfile
self.rfile = _ValidatingHeaderReader(self, original_reader) # type: ignore[assignment]
try:
return super().parse_request()
finally:
self.rfile = original_reader
def handle_one_request(self) -> None:
self._start_network_read()
try:
super().handle_one_request()
except RequestRejected as rejection:
self.close_connection = True
self._send_safely(rejection.status, {"status": rejection.payload_status})
except TimeoutError:
self.close_connection = True
self._send_safely(408, {"status": "REQUEST_TIMEOUT"})
except (BrokenPipeError, ConnectionResetError, OSError):
self.close_connection = True
except Exception:
self.close_connection = True
self._send_safely(500, {"status": "INTERNAL_ERROR"})
finally:
self._complete_network_read()
def send_error(
self,
code: int,
message: str | None = None,
explain: str | None = None,
) -> None:
"""Replace BaseHTTPRequestHandler HTML errors with bounded JSON."""
self.close_connection = True
if code == 501:
self._method_not_allowed()
return
self._send(int(code), {"status": "HTTP_ERROR"})
def log_message(self, format: str, *args: object) -> None:
return
def _send_safely(self, status: int, payload: Mapping[str, Any]) -> None:
try:
self._send(status, payload)
except OSError:
self.close_connection = True
def _path(self) -> str:
target = self.path
if not target or any(
ord(character) < 0x20 or ord(character) == 0x7F for character in target
):
raise RequestRejected(400, "BAD_REQUEST")
for index, character in enumerate(target):
if character == "%" and (
index + 2 >= len(target)
or any(value not in HEX_DIGITS for value in target[index + 1 : index + 3])
):
raise RequestRejected(400, "BAD_REQUEST")
try:
parts = urlsplit(target)
except ValueError as exc:
raise RequestRejected(400, "BAD_REQUEST") from exc
if parts.fragment:
raise RequestRejected(400, "BAD_REQUEST")
if target.startswith("/"):
if parts.scheme or parts.netloc:
raise RequestRejected(400, "BAD_REQUEST")
elif parts.scheme in {"http", "https"} and parts.netloc:
if parts.hostname is None:
raise RequestRejected(400, "BAD_REQUEST")
try:
_ = parts.port
except ValueError as exc:
raise RequestRejected(400, "BAD_REQUEST") from exc
else:
raise RequestRejected(400, "BAD_REQUEST")
return parts.path or "/"
@staticmethod
def _encoded_response(
status: object,
payload: object,
) -> tuple[int, bytes]:
if (
type(status) is not int
or not 200 <= status <= 599
or status in {204, 205, 304}
or not isinstance(payload, Mapping)
):
return 500, INTERNAL_ERROR_BODY
try:
body = json.dumps(
dict(payload),
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
allow_nan=False,
).encode("utf-8")
except Exception:
return 500, INTERNAL_ERROR_BODY
if len(body) > MAX_RESPONSE_BYTES:
return 500, INTERNAL_ERROR_BODY
return status, body
def _send(self, status: int, payload: Mapping[str, Any]) -> None:
self._complete_network_read()
status, body = self._encoded_response(status, payload)
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
if getattr(self, "command", None) != "HEAD":
self.wfile.write(body)
def _method_not_allowed(self) -> None:
self._send(405, {"status": "METHOD_NOT_ALLOWED"})
def _reject(self, status: int, payload_status: str) -> None:
self.close_connection = True
self._send(status, {"status": payload_status})
def _content_length(self) -> int:
if self.headers.get_all("Transfer-Encoding", []):
raise RequestRejected(400, "BAD_REQUEST")
values = self.headers.get_all("Content-Length", [])
if len(values) > 1:
raise RequestRejected(400, "BAD_REQUEST")
if not values:
return 0
raw_length = values[0]
if (
not raw_length
or not raw_length.isascii()
or not raw_length.isdigit()
):
raise RequestRejected(400, "BAD_REQUEST")
try:
length = int(raw_length)
except ValueError as exc:
raise RequestRejected(400, "BAD_REQUEST") from exc
if length > MAX_REQUEST_BYTES:
raise RequestRejected(413, "REQUEST_TOO_LARGE")
return length
def _read_exact_body(self, length: int) -> bytes:
try:
body = self.rfile.read(length)
except TimeoutError as exc:
raise RequestRejected(408, "REQUEST_TIMEOUT") from exc
self._ensure_network_read_active()
if len(body) != length:
raise RequestRejected(400, "BAD_REQUEST")
return body
def do_GET(self) -> None:
path = self._path()
if path == "/health/live":
self._send(200, {"status": "LIVE"})
elif path == "/health/ready":
if self.server.processor is None:
self._send(503, {"status": "SERVICE_NOT_CONFIGURED"})
else:
self._send(200, {"status": "READY"})
elif path in MUTATION_ROUTES:
self._method_not_allowed()
else:
self._send(404, {"status": "NOT_FOUND"})
def do_POST(self) -> None:
path = self._path()
if path not in MUTATION_ROUTES:
if path in {"/health/live", "/health/ready"}:
self._method_not_allowed()
else:
self._send(404, {"status": "NOT_FOUND"})
return
try:
length = self._content_length()
except RequestRejected as rejection:
self._reject(rejection.status, rejection.payload_status)
return
processor = self.server.processor
if processor is None:
self._send(503, {"status": "SERVICE_NOT_CONFIGURED"})
return
if self.server.is_closing:
self._reject(503, "SERVICE_UNAVAILABLE")
return
try:
body = self._read_exact_body(length)
except RequestRejected as rejection:
self._reject(rejection.status, rejection.payload_status)
return
self._complete_network_read()
try:
started, result = self.server.call_processor(path, body)
if not started:
self._reject(503, "SERVICE_UNAVAILABLE")
return
status, payload = result
except Exception:
self._send(500, {"status": "INTERNAL_ERROR"})
return
self._send(status, payload)
do_DELETE = _method_not_allowed
do_HEAD = _method_not_allowed
do_OPTIONS = _method_not_allowed
do_PATCH = _method_not_allowed
do_PUT = _method_not_allowed
def build_server(
host: str = "127.0.0.1",
port: int = 18792,
*,
processor: Processor | None = None,
) -> IngressHTTPServer:
"""Build a loopback-only HTTP server without starting it."""
try:
address = ipaddress.ip_address(host)
except ValueError as exc:
raise ValueError("host must be a loopback IP address") from exc
if address.version != 4 or not address.is_loopback:
raise ValueError("host must be an IPv4 loopback address")
return IngressHTTPServer((host, port), IngressRequestHandler, processor)
def serve(host: str = "127.0.0.1", port: int = 18792) -> None:
"""Run the unconfigured loopback shell until interrupted."""
server = build_server(host, port)
try:
server.serve_forever()
except KeyboardInterrupt:
return
finally:
server.server_close()
+6 -3
View File
@@ -7,7 +7,6 @@ from typing import Literal
from .tenant_context import ScopeError, ValidationError, require_ident
MemoryKind = Literal["session", "profile", "cell", "tenant", "shared_artifact"]
@@ -49,6 +48,8 @@ class MemoryScope:
if write:
raise ScopeError("shared artifacts are immutable")
return
if self.kind != requested.kind:
raise ScopeError("cross-kind memory access denied")
if self.profile_id != requested.profile_id and requested.kind in {"profile", "session"}:
raise ScopeError("cross-profile memory access denied")
if requested.kind == "session" and self.session_id != requested.session_id:
@@ -56,5 +57,7 @@ class MemoryScope:
def qualify_memory_key(scope: MemoryScope, key: str) -> str:
key = require_ident("memory_key", key)
return f"{scope.namespace}/{key}"
normalized = require_ident("memory_key", key)
if normalized != key:
raise ValidationError("memory_key must be canonical")
return f"{scope.namespace}/{normalized}"
+123 -120
View File
@@ -1,25 +1,22 @@
"""Tenant-scoped append-only message ledger and exact quote resolver."""
from __future__ import annotations
from collections.abc import Callable
from contextlib import closing
from dataclasses import dataclass
from datetime import timedelta
from contextlib import closing
import os
from pathlib import Path
import sqlite3
from typing import Iterable
from typing import TypeVar
from .tenant_context import (
CanonicalMessage,
ConflictError,
ScopeError,
ValidationError,
isoformat,
normalize_message_id,
parse_time,
utc_now,
CanonicalMessage, ConflictError, ScopeError, ValidationError,
isoformat, normalize_message_id, utc_now,
)
T = TypeVar("T")
@dataclass(frozen=True)
class LedgerRecord:
@@ -37,11 +34,7 @@ class LedgerRecord:
class MessageLedger:
"""Exact-ID ledger.
The resolver never uses time windows, prefixes, similarity, or nearest-message
fallbacks. Ambiguity and absence fail closed.
"""
"""Exact-ID ledger with no fuzzy, prefix, or time-nearest fallback."""
def __init__(self, database: str | Path, *, ttl_days: int = 7) -> None:
if not 1 <= ttl_days <= 365:
@@ -63,64 +56,64 @@ class MessageLedger:
db.execute("PRAGMA busy_timeout=30000")
return db
def _storage(self, operation: Callable[[], T]) -> T:
failure: ConflictError | None = None
try:
return operation()
except sqlite3.Error:
failure = ConflictError("ledger storage unavailable")
# Raise after the driver exception handler, keeping SQLite details out of
# both the public exception and its implicit context.
raise failure
def _initialize(self) -> None:
with closing(self._connect()) as connection, connection as db:
db.executescript(
"""
CREATE TABLE IF NOT EXISTS message_ledger (
tenant_id TEXT NOT NULL,
cell_id TEXT NOT NULL,
channel TEXT NOT NULL,
channel_account_id TEXT NOT NULL,
peer_id TEXT NOT NULL,
direction TEXT NOT NULL CHECK(direction IN ('inbound','outbound','internal')),
message_id TEXT NOT NULL,
event_id TEXT NOT NULL,
body TEXT NOT NULL,
occurred_at TEXT NOT NULL,
expires_at TEXT NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY(
tenant_id, cell_id, channel, channel_account_id,
peer_id, direction, message_id
),
UNIQUE(tenant_id, cell_id, event_id)
);
CREATE INDEX IF NOT EXISTS idx_message_reply_lookup ON message_ledger(
tenant_id, cell_id, channel, channel_account_id,
peer_id, message_id, expires_at
);
"""
)
def initialize() -> None:
with closing(self._connect()) as connection, connection as db:
db.executescript(
"""
CREATE TABLE IF NOT EXISTS message_ledger (
tenant_id TEXT NOT NULL,
cell_id TEXT NOT NULL,
channel TEXT NOT NULL,
channel_account_id TEXT NOT NULL,
peer_id TEXT NOT NULL,
direction TEXT NOT NULL CHECK(direction IN ('inbound','outbound','internal')),
message_id TEXT NOT NULL,
event_id TEXT NOT NULL,
body TEXT NOT NULL,
occurred_at TEXT NOT NULL,
expires_at TEXT NOT NULL,
created_at TEXT NOT NULL,
PRIMARY KEY(
tenant_id, cell_id, channel, channel_account_id,
peer_id, direction, message_id
),
UNIQUE(tenant_id, cell_id, event_id)
);
CREATE INDEX IF NOT EXISTS idx_message_reply_lookup ON message_ledger(
tenant_id, cell_id, channel, channel_account_id,
peer_id, message_id, expires_at
);
"""
)
self._storage(initialize)
def append(self, message: CanonicalMessage) -> LedgerRecord:
"""Append one exact message through the runtime write path.
This is intentionally the owner of validation, idempotency, and the
immediate SQLite transaction. Runtime callers must not depend on the
legacy ``remember`` compatibility name.
"""
message_id = normalize_message_id(message.message_id)
now = utc_now()
expires = isoformat(now + timedelta(days=self.ttl_days))
key = (
message.tenant_id,
message.cell_id,
message.channel,
message.channel_account_id,
message.peer_id,
message.direction,
message_id,
message.tenant_id, message.cell_id, message.channel,
message.channel_account_id, message.peer_id, message.direction, message_id,
)
storage_error: ConflictError | None = None
try:
def write() -> LedgerRecord:
with closing(self._connect()) as connection, connection as db:
db.execute("BEGIN IMMEDIATE")
existing = db.execute(
"SELECT event_id, body, occurred_at, expires_at FROM message_ledger "
"WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
"AND peer_id=? AND direction=? AND message_id=?",
key,
"AND peer_id=? AND direction=? AND message_id=?", key,
).fetchone()
if existing:
if existing["body"] != message.body:
@@ -133,50 +126,56 @@ class MessageLedger:
)
db.execute(
"INSERT INTO message_ledger VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
(*key, message.event_id, message.body, message.occurred_at, expires, isoformat(now)),
(*key, message.event_id, message.body, message.occurred_at,
expires, isoformat(now)),
)
db.commit()
except sqlite3.Error:
storage_error = ConflictError("ledger storage unavailable")
if storage_error is not None:
# Raise outside the sqlite exception handler so callers cannot inspect
# a driver exception through __cause__ or __context__.
raise storage_error
return LedgerRecord(*key, message.event_id, message.body, message.occurred_at, expires)
return LedgerRecord(
*key, message.event_id, message.body, message.occurred_at, expires
)
return self._storage(write)
def remember(self, message: CanonicalMessage) -> LedgerRecord:
"""Backward-compatible API wrapper for callers outside the runtime path."""
"""Backward-compatible API wrapper outside the runtime write path."""
return self.append(message)
def matches(self, message: CanonicalMessage) -> bool:
"""Whether this exact message is already durably represented."""
message_id = normalize_message_id(message.message_id)
with closing(self._connect()) as connection, connection as db:
row = db.execute(
"SELECT event_id, body FROM message_ledger WHERE tenant_id=? AND "
"cell_id=? AND channel=? AND channel_account_id=? AND peer_id=? "
"AND direction=? AND message_id=?",
(
message.tenant_id, message.cell_id, message.channel,
message.channel_account_id, message.peer_id, message.direction,
message_id,
),
).fetchone()
return bool(row and row["event_id"] == message.event_id and row["body"] == message.body)
def query() -> bool:
with closing(self._connect()) as connection, connection as db:
row = db.execute(
"SELECT event_id, body FROM message_ledger WHERE tenant_id=? AND "
"cell_id=? AND channel=? AND channel_account_id=? AND peer_id=? "
"AND direction=? AND message_id=?",
(
message.tenant_id, message.cell_id, message.channel,
message.channel_account_id, message.peer_id, message.direction,
message_id,
),
).fetchone()
return bool(
row and row["event_id"] == message.event_id
and row["body"] == message.body
)
return self._storage(query)
def remove(self, record: LedgerRecord) -> None:
"""Compensate a newly appended record if a later ingress step fails."""
with closing(self._connect()) as connection, connection as db:
db.execute(
"DELETE FROM message_ledger WHERE tenant_id=? AND cell_id=? AND "
"channel=? AND channel_account_id=? AND peer_id=? AND direction=? "
"AND message_id=? AND event_id=?",
(
record.tenant_id, record.cell_id, record.channel,
record.channel_account_id, record.peer_id, record.direction,
record.message_id, record.event_id,
),
)
def delete() -> None:
with closing(self._connect()) as connection, connection as db:
db.execute(
"DELETE FROM message_ledger WHERE tenant_id=? AND cell_id=? AND "
"channel=? AND channel_account_id=? AND peer_id=? AND direction=? "
"AND message_id=? AND event_id=?",
(
record.tenant_id, record.cell_id, record.channel,
record.channel_account_id, record.peer_id, record.direction,
record.message_id, record.event_id,
),
)
self._storage(delete)
def resolve_quote(self, message: CanonicalMessage) -> str:
if message.quoted_body is not None:
@@ -184,24 +183,22 @@ class MessageLedger:
if message.reply_to_message_id is None:
raise ScopeError("quoted_text_unavailable")
reply_id = normalize_message_id(
message.reply_to_message_id,
field_name="reply_to_message_id",
message.reply_to_message_id, field_name="reply_to_message_id"
)
with closing(self._connect()) as connection, connection as db:
rows = db.execute(
"SELECT direction, body, event_id FROM message_ledger "
"WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
"AND peer_id=? AND message_id=? AND expires_at>?",
(
message.tenant_id,
message.cell_id,
message.channel,
message.channel_account_id,
message.peer_id,
reply_id,
isoformat(),
),
).fetchall()
def query() -> list[sqlite3.Row]:
with closing(self._connect()) as connection, connection as db:
return db.execute(
"SELECT direction, body, event_id FROM message_ledger "
"WHERE tenant_id=? AND cell_id=? AND channel=? AND channel_account_id=? "
"AND peer_id=? AND message_id=? AND expires_at>?",
(
message.tenant_id, message.cell_id, message.channel,
message.channel_account_id, message.peer_id, reply_id, isoformat(),
),
).fetchall()
rows = self._storage(query)
if len(rows) == 1:
return str(rows[0]["body"])
if len(rows) > 1:
@@ -209,16 +206,22 @@ class MessageLedger:
raise ScopeError("quoted_text_unavailable")
def purge_expired(self) -> int:
with closing(self._connect()) as connection, connection as db:
cursor = db.execute(
"DELETE FROM message_ledger WHERE expires_at<=?", (isoformat(),)
)
return cursor.rowcount
def purge() -> int:
with closing(self._connect()) as connection, connection as db:
cursor = db.execute(
"DELETE FROM message_ledger WHERE expires_at<=?", (isoformat(),)
)
return cursor.rowcount
return self._storage(purge)
def integrity_check(self) -> str:
with closing(self._connect()) as connection, connection as db:
return str(db.execute("PRAGMA integrity_check").fetchone()[0])
def check() -> str:
with closing(self._connect()) as connection, connection as db:
return str(db.execute("PRAGMA integrity_check").fetchone()[0])
return self._storage(check)
def count(self) -> int:
with closing(self._connect()) as connection, connection as db:
return int(db.execute("SELECT count(*) FROM message_ledger").fetchone()[0])
def query() -> int:
with closing(self._connect()) as connection, connection as db:
return int(db.execute("SELECT count(*) FROM message_ledger").fetchone()[0])
return self._storage(query)
+38 -10
View File
@@ -1,14 +1,10 @@
"""Tenant and Cell scoped secret reference model.
This module never stores secret values. It stores ownership and retrieval scope
metadata for integration with a real secret broker.
"""
"""Tenant and Cell scoped, value-free secret reference model."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
from typing import Any, Literal
from .tenant_context import ValidationError, require_ident
from .tenant_context import ScopeError, ValidationError, require_ident
@dataclass(frozen=True)
@@ -22,12 +18,44 @@ class SecretRef:
def __post_init__(self) -> None:
for name in ("secret_id", "cell_id", "version"):
require_ident(name, getattr(self, name))
require_ident("tenant_id", self.tenant_id, tenant=True)
value = getattr(self, name)
if require_ident(name, value) != value:
raise ValidationError(f"{name} must be canonical")
if require_ident("tenant_id", self.tenant_id, tenant=True) != self.tenant_id:
raise ValidationError("tenant_id must be canonical")
if self.kind not in {"channel", "model", "tool", "runtime"}:
raise ValidationError("invalid secret kind")
if not self.broker_path.startswith("secret://"):
self._validate_broker_path()
def _validate_broker_path(self) -> None:
if not isinstance(self.broker_path, str) or not self.broker_path.startswith("secret://"):
raise ValidationError("secret broker path required")
suffix = self.broker_path[len("secret://"):]
if (
not suffix or "?" in suffix or "#" in suffix or chr(92) in suffix
or any(part in {"", ".", ".."} for part in suffix.split("/"))
):
raise ValidationError("invalid secret broker path")
for part in suffix.split("/"):
if require_ident("secret broker path segment", part) != part:
raise ValidationError("invalid secret broker path")
def can_be_used_by(self, *, tenant_id: str, cell_id: str) -> bool:
return self.tenant_id == tenant_id and self.cell_id == cell_id
def resolve(self, broker: Any) -> str:
"""Resolve through one explicit broker boundary without leaking provider errors."""
failure: ScopeError | None = None
value: object = None
try:
reader = getattr(broker, "read_secret")
if not callable(reader):
raise TypeError("invalid broker")
value = reader(self)
if not isinstance(value, str) or not value:
raise ValueError("invalid broker result")
except Exception:
failure = ScopeError("secret broker unavailable")
if failure is not None:
raise failure
return value
+17
View File
@@ -0,0 +1,17 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "aios-v2r1-runtime"
version = "2.1.0rc1"
description = "AIOS V2R1 canonical runtime package"
requires-python = ">=3.11"
dependencies = []
[project.scripts]
aios-cell = "aios_v2r1.cli:main"
[tool.setuptools]
package-dir = {"" = "app"}
packages = ["aios_v2r1"]
+39
View File
@@ -0,0 +1,39 @@
# Personal Weixin v1 security contract
Status: normative design contract for R10 B6/B7. It authorizes no runtime or production change.
## Trust boundaries and ownership
The browser is untrusted. The Portal authenticates the human user and resolves tenant identity from the server session. The Portal adapter is least-privileged: it may request account-scoped operations but must not store provider credentials, tokens, raw iLink state, or OpenClaw protocol state. OpenClaw owns provider credentials and protocol state. The Gateway transport authenticator, not an event-body boolean or identity field, establishes whether an inbound event is authentic.
Every operation is keyed by server-resolved `(tenantId, accountId)`. Entitlements are allow-only, server-owned records. A default account is display preference, never authorization. The browser-visible `accountId` is an untrusted reference to a server-preallocated account slot: the server MUST resolve it through the authenticated user's tenant and current entitlement on list, readiness, login start, login wait, status, inbound challenge, canary, and rollback checks. Start and every wait poll re-run that resolution. Missing, arbitrary, stale, cross-tenant, or no-longer-entitled values fail closed before account existence or state is disclosed; there is no default/first/configured-account fallback.
## Login state and secret handling
`sessionKey` is a required, nonempty opaque value returned by OpenClaw. The adapter MUST forward its exact bytes unchanged. It MUST NOT trim, normalize, case-fold, derive, regenerate, alias, or substitute the value. A missing or empty `sessionKey` fails closed and MUST NOT fall back to `accountId` or any identifier. Durable state stores only a one-way session-key digest where correlation is required; raw `sessionKey` is not persisted.
`verifyCode` is accepted only as an authenticated, CSRF-protected wait POST test/input field, matches 4-8 decimal digits, is forwarded once, and is discarded. `challengeCode` is accepted only as an authenticated Gateway event test/input field, is compared through a one-way digest, and is discarded. Neither secret may be persisted in raw form or appear in URLs, responses, logs, traces, metrics labels, analytics, audit payloads, exceptions, or background-job arguments. Golden fixtures containing these values are synthetic test vectors only and MUST NOT be copied into production telemetry or responses.
QR data URLs are transient no-store response data with a 1 MiB bound and MUST NOT be logged or persisted. Canonical login outcomes are `qr_pending`, `need_verifycode`, `invalid_verifycode`, `expired`, `failed`, `already_connected`, and `confirmed`. Any unknown upstream status maps fail-closed to canonical `failed` with `errorCode=unknown_upstream_status`; raw unknown status is bounded evidence only. Upstream `binded_redirect` maps only to `already_connected` with `connected=false`, `alreadyConnected=true`, and `terminal=true`, never to `confirmed`.
## Canonical readiness
Readiness is one live, authenticated snapshot from the Gateway/OpenClaw boundary with one unified response shape for `ready`, `not_ready`, and `error`. It MUST include Gateway reachability/error, exact plugin id/version/digest, selected account identity, configured/enabled/running/last-error state, current tenant/account entitlement, reasons, observation time, and freshness deadline. `ready=true` is legal only when every required field is present, the snapshot is still fresh, expected plugin version/digest match, the selected account is entitled/configured/enabled/running with null error, and the Gateway is reachable. Missing fields, an old snapshot, digest mismatch, no entitlement, not-running state, unauthenticated evidence, or any contradiction returns `ready=false` with `code=not_ready` or `code=error` and at least one stable reason. Cached, synthetic, file-only, default-account, or service-up-only evidence cannot produce ready.
## Authenticated inbound-event challenge
The event body contains exactly `eventId`, `accountId`, `senderId`, `chatId`, `challengeCode`, and `observedAt`. It does not and cannot assert authentication, tenant id, Portal user id, or session key. Those values are absent by contract. Transport authentication MUST cover HTTP method, path, timestamp header, nonce header, and raw body digest; enforce a bounded clock window; durably reserve nonce/event replay keys for at least that window; compare signatures in constant time; and complete before body values reach logs or mutable business state.
After transport authentication, the server hashes the ephemeral `challengeCode` and resolves the pending durable challenge. That server-created record already owns tenant id, Portal user id, entitled account id, session-key digest, issued/expiry/consumed times, and state (`issued`, `consumed`, or `expired`). The body `accountId` must exactly match the record and current entitlement; it never supplies tenant/user/session authority. The first valid authenticated event atomically binds its provider `senderId` and `chatId`.
Issuance and consumption MUST run in a database transaction with row locking or an equivalent compare-and-swap primitive. `issued -> consumed` is single-use and atomic. An exact replay of the same authenticated event is idempotent. A changed event payload, sender, chat, account, durable tenant/user/account/session binding, expired or already-consumed record, duplicate event with different payload, missing entitlement, or unauthenticated transport fails closed with zero mutation. Raw challenge codes and raw session keys are never stored.
## Unsupported unbind
End-to-end unbind is unsupported in R10. Every public or internal unbind entry point MUST return the exact schema response with HTTP 501, `ok=false`, `supported=false`, `code=unbind_not_supported`, and `mutationCount=0`. The 501 decision occurs before token deletion, account update, session or challenge mutation, Gateway/OpenClaw call, business-state audit mutation, or background-job enqueue.
## Web, canary, and rollback controls
Only exact B6/B7 allowlisted paths in the R10 amendment may change. Portal endpoints require authenticated Django sessions, CSRF protection for mutations, server-scoped object lookup, IDOR tests, rate limits, request-size limits, `Cache-Control: no-store`, and structured redaction of session keys, verification codes, challenge codes, QR data, tokens, and provider payloads. Stable public errors never expose upstream bodies or cross-tenant existence.
The canary uses one explicitly entitled tenant/account pair and proves restart, concurrent waits, atomic challenge consumption, exact replay, altered replay denial, cross-tenant denial, all canonical statuses/readiness shapes, and zero-mutation 501 behavior. Promotion requires byte-identical artifacts and digests from the verified canary. Rollback restores the immediately preceding byte-identical artifacts/configuration and re-runs canonical readiness; database rollback is forbidden unless an explicitly approved manifest proves backward incompatibility and supplies a separate recovery plan.
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"tenant.account.entitlement","tenantId":"tenant-canary","accountId":"account-canary","effect":"allow","policyVersion":"policy-v1","issuedAtMs":1784068000000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":true,"code":"ready","reasons":[],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":null},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":false,"code":"not_ready","reasons":[{"code":"field_missing","field":"plugin.digestSha256"}],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":false,"code":"not_ready","reasons":[{"code":"snapshot_stale","field":"freshUntilMs"}],"observedAtMs":1784060000000,"freshUntilMs":1784060060000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":true}],"ready":false,"code":"not_ready","reasons":[{"code":"plugin_digest_mismatch","field":"plugin.digestSha256"}],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":true,"lastError":null,"isDefault":false,"entitled":false}],"ready":false,"code":"not_ready","reasons":[{"code":"account_not_entitled","field":"accounts[0].entitled"}],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":true,"errorCode":null},"plugin":{"id":"openclaw-weixin","loaded":true,"version":"2.4.6","digestSha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":false,"lastError":"plugin_not_running","isDefault":false,"entitled":true}],"ready":false,"code":"not_ready","reasons":[{"code":"account_not_running","field":"accounts[0].running"}],"observedAtMs":1784068060000,"freshUntilMs":1784068120000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"personal.weixin.readiness","tenantId":"tenant-canary","selectedAccountId":"account-canary","gateway":{"reachable":false,"errorCode":"gateway_unreachable"},"plugin":{"id":"openclaw-weixin","loaded":false,"version":null,"digestSha256":null},"accounts":[{"accountId":"account-canary","name":"Canary Personal Weixin","configured":true,"enabled":true,"running":false,"lastError":"gateway_unreachable","isDefault":false,"entitled":true}],"ready":false,"code":"error","reasons":[{"code":"gateway_error","field":"gateway.reachable"}],"observedAtMs":1784068060000,"freshUntilMs":1784068060000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.start.request","accountId":"account-canary","force":false,"timeoutMs":30000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.wait.request","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","verifyCode":"4321","currentQrDataUrl":"data:image/png;base64,iVBORw0KGgo=","timeoutMs":30000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"qr_pending","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":false,"needsVerifyCode":false,"qrDataUrl":"data:image/png;base64,iVBORw0KGgo="}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"need_verifycode","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":false,"needsVerifyCode":true}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"invalid_verifycode","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":false,"needsVerifyCode":true,"errorCode":"invalid_verifycode"}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"expired","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":true,"needsVerifyCode":false,"errorCode":"login_expired"}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"failed","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":true,"needsVerifyCode":false,"errorCode":"upstream_failed"}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"already_connected","upstreamStatus":"binded_redirect","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":true,"terminal":true,"needsVerifyCode":false}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"confirmed","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":true,"alreadyConnected":false,"terminal":true,"needsVerifyCode":false,"userId":"sender-canary"}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.login.result","status":"failed","upstreamStatus":"future_unknown_state","accountId":"account-canary","sessionKey":"weixin-session-20260715-A1","connected":false,"alreadyConnected":false,"terminal":true,"needsVerifyCode":false,"errorCode":"unknown_upstream_status"}
@@ -0,0 +1 @@
{"eventId":"event-20260715-A1","accountId":"account-canary","senderId":"sender-canary","chatId":"chat-canary","challengeCode":"A1B2C3D4","observedAt":1784068120000}
@@ -0,0 +1 @@
{"schemaVersion":"aios.personal-weixin.golden.v1","kind":"web.unbind.result","httpStatus":501,"ok":false,"supported":false,"code":"unbind_not_supported","mutationCount":0}
@@ -0,0 +1,986 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://aios.imruyi.top/contracts/personal-weixin/v1/golden-contract.schema.json",
"title": "AIOS Personal Weixin Golden Contract v1",
"oneOf": [
{
"$ref": "#/$defs/entitlement"
},
{
"$ref": "#/$defs/readiness"
},
{
"$ref": "#/$defs/startRequest"
},
{
"$ref": "#/$defs/waitRequest"
},
{
"$ref": "#/$defs/loginResult"
},
{
"$ref": "#/$defs/inboundEvent"
},
{
"$ref": "#/$defs/unbindResult"
}
],
"$defs": {
"identifier": {
"type": "string",
"minLength": 1,
"maxLength": 128,
"pattern": "^[A-Za-z0-9._~:@-]+$"
},
"sessionKey": {
"type": "string",
"minLength": 1,
"maxLength": 256,
"pattern": "^[A-Za-z0-9._~:-]+$"
},
"schemaVersion": {
"const": "aios.personal-weixin.golden.v1"
},
"entitlement": {
"type": "object",
"additionalProperties": false,
"required": [
"schemaVersion",
"kind",
"tenantId",
"accountId",
"effect",
"policyVersion",
"issuedAtMs"
],
"properties": {
"schemaVersion": {
"$ref": "#/$defs/schemaVersion"
},
"kind": {
"const": "tenant.account.entitlement"
},
"tenantId": {
"$ref": "#/$defs/identifier"
},
"accountId": {
"$ref": "#/$defs/identifier"
},
"effect": {
"const": "allow"
},
"policyVersion": {
"$ref": "#/$defs/identifier"
},
"issuedAtMs": {
"type": "integer",
"minimum": 0
}
}
},
"readinessReason": {
"type": "object",
"additionalProperties": false,
"required": [
"code",
"field"
],
"properties": {
"code": {
"enum": [
"field_missing",
"snapshot_stale",
"plugin_digest_mismatch",
"account_not_entitled",
"account_not_running",
"gateway_error"
]
},
"field": {
"type": "string",
"minLength": 1,
"maxLength": 128
}
}
},
"readiness": {
"type": "object",
"additionalProperties": false,
"required": [
"schemaVersion",
"kind",
"tenantId",
"selectedAccountId",
"gateway",
"plugin",
"accounts",
"ready",
"code",
"reasons",
"observedAtMs",
"freshUntilMs"
],
"properties": {
"schemaVersion": {
"$ref": "#/$defs/schemaVersion"
},
"kind": {
"const": "personal.weixin.readiness"
},
"tenantId": {
"$ref": "#/$defs/identifier"
},
"selectedAccountId": {
"$ref": "#/$defs/identifier"
},
"gateway": {
"type": "object",
"additionalProperties": false,
"required": [
"reachable",
"errorCode"
],
"properties": {
"reachable": {
"type": "boolean"
},
"errorCode": {
"type": [
"string",
"null"
],
"minLength": 1,
"maxLength": 128
}
}
},
"plugin": {
"type": "object",
"additionalProperties": false,
"required": [
"id",
"loaded",
"version",
"digestSha256"
],
"properties": {
"id": {
"const": "openclaw-weixin"
},
"loaded": {
"type": "boolean"
},
"version": {
"type": [
"string",
"null"
],
"minLength": 1,
"maxLength": 64
},
"digestSha256": {
"anyOf": [
{
"type": "string",
"pattern": "^[a-f0-9]{64}$"
},
{
"type": "null"
}
]
}
}
},
"accounts": {
"type": "array",
"minItems": 1,
"maxItems": 100,
"items": {
"type": "object",
"additionalProperties": false,
"required": [
"accountId",
"name",
"configured",
"enabled",
"running",
"lastError",
"isDefault",
"entitled"
],
"properties": {
"accountId": {
"$ref": "#/$defs/identifier"
},
"name": {
"type": "string",
"minLength": 1,
"maxLength": 128
},
"configured": {
"type": "boolean"
},
"enabled": {
"type": "boolean"
},
"running": {
"type": "boolean"
},
"lastError": {
"type": [
"string",
"null"
],
"minLength": 1,
"maxLength": 256
},
"isDefault": {
"type": "boolean"
},
"entitled": {
"type": "boolean"
}
}
}
},
"ready": {
"type": "boolean"
},
"code": {
"enum": [
"ready",
"not_ready",
"error"
]
},
"reasons": {
"type": "array",
"maxItems": 32,
"items": {
"$ref": "#/$defs/readinessReason"
}
},
"observedAtMs": {
"type": "integer",
"minimum": 0
},
"freshUntilMs": {
"type": "integer",
"minimum": 0
}
},
"allOf": [
{
"if": {
"properties": {
"ready": {
"const": true
}
},
"required": [
"ready"
]
},
"then": {
"properties": {
"code": {
"const": "ready"
},
"reasons": {
"maxItems": 0
},
"gateway": {
"properties": {
"reachable": {
"const": true
},
"errorCode": {
"const": null
}
}
},
"plugin": {
"properties": {
"loaded": {
"const": true
},
"version": {
"type": "string"
},
"digestSha256": {
"type": "string",
"pattern": "^[a-f0-9]{64}$"
}
}
},
"accounts": {
"items": {
"properties": {
"configured": {
"const": true
},
"enabled": {
"const": true
},
"running": {
"const": true
},
"lastError": {
"const": null
},
"entitled": {
"const": true
}
}
}
}
}
},
"else": {
"properties": {
"code": {
"enum": [
"not_ready",
"error"
]
},
"reasons": {
"minItems": 1
}
}
}
}
]
},
"startRequest": {
"type": "object",
"additionalProperties": false,
"required": [
"schemaVersion",
"kind",
"accountId",
"force",
"timeoutMs"
],
"properties": {
"schemaVersion": {
"$ref": "#/$defs/schemaVersion"
},
"kind": {
"const": "web.login.start.request"
},
"accountId": {
"$ref": "#/$defs/identifier"
},
"force": {
"const": false
},
"timeoutMs": {
"type": "integer",
"minimum": 1000,
"maximum": 120000
}
}
},
"waitRequest": {
"type": "object",
"additionalProperties": false,
"required": [
"schemaVersion",
"kind",
"accountId",
"sessionKey",
"timeoutMs"
],
"properties": {
"schemaVersion": {
"$ref": "#/$defs/schemaVersion"
},
"kind": {
"const": "web.login.wait.request"
},
"accountId": {
"$ref": "#/$defs/identifier"
},
"sessionKey": {
"$ref": "#/$defs/sessionKey"
},
"verifyCode": {
"type": "string",
"pattern": "^[0-9]{4,8}$"
},
"currentQrDataUrl": {
"type": "string",
"minLength": 1,
"maxLength": 1048576,
"pattern": "^data:image/png;base64,"
},
"timeoutMs": {
"type": "integer",
"minimum": 1000,
"maximum": 120000
}
}
},
"loginResult": {
"type": "object",
"additionalProperties": false,
"required": [
"schemaVersion",
"kind",
"status",
"accountId",
"sessionKey",
"connected",
"alreadyConnected",
"terminal",
"needsVerifyCode"
],
"properties": {
"schemaVersion": {
"$ref": "#/$defs/schemaVersion"
},
"kind": {
"const": "web.login.result"
},
"status": {
"enum": [
"qr_pending",
"need_verifycode",
"invalid_verifycode",
"expired",
"failed",
"already_connected",
"confirmed"
]
},
"accountId": {
"$ref": "#/$defs/identifier"
},
"sessionKey": {
"$ref": "#/$defs/sessionKey"
},
"connected": {
"type": "boolean"
},
"alreadyConnected": {
"type": "boolean"
},
"terminal": {
"type": "boolean"
},
"needsVerifyCode": {
"type": "boolean"
},
"qrDataUrl": {
"type": "string",
"minLength": 1,
"maxLength": 1048576,
"pattern": "^data:image/png;base64,"
},
"userId": {
"$ref": "#/$defs/identifier"
},
"errorCode": {
"enum": [
"invalid_verifycode",
"login_expired",
"upstream_failed",
"unknown_upstream_status"
]
},
"upstreamStatus": {
"type": "string",
"minLength": 1,
"maxLength": 128
}
},
"allOf": [
{
"if": {
"properties": {
"status": {
"const": "qr_pending"
}
}
},
"then": {
"required": [
"qrDataUrl"
],
"properties": {
"connected": {
"const": false
},
"alreadyConnected": {
"const": false
},
"terminal": {
"const": false
},
"needsVerifyCode": {
"const": false
}
},
"not": {
"anyOf": [
{
"required": [
"userId"
]
},
{
"required": [
"errorCode"
]
},
{
"required": [
"upstreamStatus"
]
}
]
}
}
},
{
"if": {
"properties": {
"status": {
"const": "need_verifycode"
}
}
},
"then": {
"properties": {
"connected": {
"const": false
},
"alreadyConnected": {
"const": false
},
"terminal": {
"const": false
},
"needsVerifyCode": {
"const": true
}
},
"not": {
"anyOf": [
{
"required": [
"qrDataUrl"
]
},
{
"required": [
"userId"
]
},
{
"required": [
"errorCode"
]
},
{
"required": [
"upstreamStatus"
]
}
]
}
}
},
{
"if": {
"properties": {
"status": {
"const": "invalid_verifycode"
}
}
},
"then": {
"required": [
"errorCode"
],
"properties": {
"errorCode": {
"const": "invalid_verifycode"
},
"connected": {
"const": false
},
"alreadyConnected": {
"const": false
},
"terminal": {
"const": false
},
"needsVerifyCode": {
"const": true
}
},
"not": {
"anyOf": [
{
"required": [
"qrDataUrl"
]
},
{
"required": [
"userId"
]
},
{
"required": [
"upstreamStatus"
]
}
]
}
}
},
{
"if": {
"properties": {
"status": {
"const": "expired"
}
}
},
"then": {
"required": [
"errorCode"
],
"properties": {
"errorCode": {
"const": "login_expired"
},
"connected": {
"const": false
},
"alreadyConnected": {
"const": false
},
"terminal": {
"const": true
},
"needsVerifyCode": {
"const": false
}
},
"not": {
"anyOf": [
{
"required": [
"qrDataUrl"
]
},
{
"required": [
"userId"
]
},
{
"required": [
"upstreamStatus"
]
}
]
}
}
},
{
"if": {
"properties": {
"status": {
"const": "failed"
}
}
},
"then": {
"required": [
"errorCode"
],
"properties": {
"connected": {
"const": false
},
"alreadyConnected": {
"const": false
},
"terminal": {
"const": true
},
"needsVerifyCode": {
"const": false
}
},
"not": {
"anyOf": [
{
"required": [
"qrDataUrl"
]
},
{
"required": [
"userId"
]
}
]
}
}
},
{
"if": {
"required": [
"upstreamStatus"
],
"properties": {
"status": {
"const": "failed"
}
}
},
"then": {
"properties": {
"errorCode": {
"const": "unknown_upstream_status"
}
},
"not": {
"anyOf": [
{
"required": [
"qrDataUrl"
]
},
{
"required": [
"userId"
]
}
]
}
}
},
{
"if": {
"required": [
"status",
"errorCode"
],
"properties": {
"status": {
"const": "failed"
},
"errorCode": {
"const": "unknown_upstream_status"
}
}
},
"then": {
"required": [
"upstreamStatus"
]
}
},
{
"if": {
"required": [
"status",
"errorCode"
],
"properties": {
"status": {
"const": "failed"
},
"errorCode": {
"const": "upstream_failed"
}
}
},
"then": {
"not": {
"required": [
"upstreamStatus"
]
}
}
},
{
"if": {
"properties": {
"status": {
"const": "already_connected"
}
}
},
"then": {
"required": [
"upstreamStatus"
],
"properties": {
"upstreamStatus": {
"const": "binded_redirect"
},
"connected": {
"const": false
},
"alreadyConnected": {
"const": true
},
"terminal": {
"const": true
},
"needsVerifyCode": {
"const": false
}
},
"not": {
"anyOf": [
{
"required": [
"qrDataUrl"
]
},
{
"required": [
"userId"
]
},
{
"required": [
"errorCode"
]
}
]
}
}
},
{
"if": {
"properties": {
"status": {
"const": "confirmed"
}
}
},
"then": {
"required": [
"userId"
],
"properties": {
"connected": {
"const": true
},
"alreadyConnected": {
"const": false
},
"terminal": {
"const": true
},
"needsVerifyCode": {
"const": false
}
},
"not": {
"anyOf": [
{
"required": [
"qrDataUrl"
]
},
{
"required": [
"errorCode"
]
},
{
"required": [
"upstreamStatus"
]
}
]
}
}
}
]
},
"inboundEvent": {
"type": "object",
"additionalProperties": false,
"required": [
"eventId",
"accountId",
"senderId",
"chatId",
"challengeCode",
"observedAt"
],
"properties": {
"eventId": {
"$ref": "#/$defs/identifier"
},
"accountId": {
"$ref": "#/$defs/identifier"
},
"senderId": {
"$ref": "#/$defs/identifier"
},
"chatId": {
"$ref": "#/$defs/identifier"
},
"challengeCode": {
"type": "string",
"minLength": 6,
"maxLength": 64,
"pattern": "^[A-Z0-9]+$"
},
"observedAt": {
"type": "integer",
"minimum": 0
}
}
},
"unbindResult": {
"type": "object",
"additionalProperties": false,
"required": [
"schemaVersion",
"kind",
"httpStatus",
"ok",
"supported",
"code",
"mutationCount"
],
"properties": {
"schemaVersion": {
"$ref": "#/$defs/schemaVersion"
},
"kind": {
"const": "web.unbind.result"
},
"httpStatus": {
"const": 501
},
"ok": {
"const": false
},
"supported": {
"const": false
},
"code": {
"const": "unbind_not_supported"
},
"mutationCount": {
"const": 0
}
}
}
}
}
@@ -18,12 +18,14 @@
- All OpenClaw loader work uses an isolated temporary or candidate config/state directory.
- `ACCEPTED` is terminal. Legacy handling is reachable only when bridge mode is explicitly `disabled`.
- Never claim exactly-once Hermes delivery or use time/approximation as message authority.
- D0 is a non-blocker documentation correction. Its reviewed commit is the R9 `START`; it does not consume a blocker label or relax any B3/B4/B5 evidence gate.
- B2 is both the canonical-package blocker and the executable completeness gate for Ledger, Memory, Secret, Audit, Hermes Profile, Cell Manager/CLI, and Fabric. Existing correct behavior is recorded as characterization; a production edit requires an observed contract RED first.
After docs review, initialize the stable per-worktree ledger from the then-current tip rather than a hardcoded SHA:
```bash
test -z "$(git status --porcelain)"
test "$(git branch --show-current)" = "integration/v2r1-runtime-20260714"
test "$(git branch --show-current)" = "integration/v2r1-runtime-20260715-r9"
start_sha="$(git rev-parse HEAD)"
start_parent="$(git rev-parse HEAD^)"
start_tree="$(git rev-parse HEAD^{tree})"
@@ -170,8 +172,24 @@ test "$tree" = "$(git rev-parse HEAD^{tree})"
- Create: `tests/v2r1/test_ingress_service.py`
- Create: `tests/test_v2r1_package_identity.py`
- Modify: the exact 23 Appendix A test files, and no unlisted legacy import file
- Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/message_ledger.py`
- Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/memory_scope.py`
- Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/secret_scope.py`
- Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/audit.py`
- Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/hermes_profile.py`
- Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/cell_manager.py`
- Modify only after an observed matrix RED: `active-runtime/app/aios_v2r1/hermes_fabric.py`
- Expand: `tests/v2r1/test_message_ledger_append.py`
- Create: `tests/v2r1/test_memory_scope.py`
- Create: `tests/v2r1/test_secret_scope.py`
- Create: `tests/v2r1/test_audit.py`
- Create: `tests/v2r1/test_hermes_profile.py`
- Create: `tests/v2r1/test_cell_manager.py`
- Expand: `tests/v2r1/test_hermes_fabric.py`
**Exact contract:** V2R1 imports only as `aios_v2r1`; no `active_runtime` shim/alias/namespace extension/copied tree. Wheel includes `aios_v2r1`, excludes `active_runtime` and legacy `app`. Legacy repository tests change `active_runtime.app.*` to `app.*` and run with `PYTHONPATH='active-runtime:active-runtime/app'`. Mutation routes return bounded `SERVICE_NOT_CONFIGURED` until B3/B4 processors exist. Existing cell CLI stays compatible and `serve` is separate.
**Exact contract:** V2R1 imports only as `aios_v2r1`; no `active_runtime` shim/alias/namespace extension/copied tree. Wheel includes `aios_v2r1`, excludes `active_runtime` and legacy `app`. Legacy repository tests change `active_runtime.app.*` to `app.*` and run with `PYTHONPATH='active-runtime:active-runtime/app'`. Mutation routes return bounded `SERVICE_NOT_CONFIGURED` until B3/B4 processors exist. Existing cell CLI stays compatible and `serve` is separate. B2 changes exactly 43 paths: the original 29 package/import paths plus the seven named runtime modules and seven named matrix-test paths above.
**Executable completeness matrix:** Message Ledger proves exact-ID query with no fallback, idempotency, collision conflict, tenant/cell/direction isolation, expiry/purge, restart, concurrency, and sanitized SQLite failures. Memory proves every namespace, cross-scope denial, shared-artifact read-only access, qualified keys, traversal rejection, and disabled search. Secret proves reference/path/scope/kind/version/identifier validation, no value leakage, and fail-closed broker boundaries. Audit proves genesis, tamper/delete/reorder detection, Unicode canonicalization, concurrency, restart, validation, tenant isolation, and sanitized SQLite failures. Hermes Profile proves exact environment, absolute roots, `0700` modes, traversal/symlink denial, scope, idempotent prepare, and no inherited environment. Cell Manager/CLI proves one cell per tenant, an explicit state graph, atomic/concurrent persistence, corrupt-registry fail-closed behavior, stable CLI exit codes/JSON, and restart. Fabric proves claimed jobs return `running` consistently with durable state, tenant/cell/profile isolation, one concurrent claimant, rejection of non-claimable states, restart, and sanitized SQLite failures.
- [ ] Rehydrate the stable blocker ledger using the global snippet and require its latest label is `B1`.
- [ ] Rehydrate the exact explorer manifest path and compare sorted actual/expected lists between exact marker lines. The marker names are assembled from split literals so this command cannot match itself:
@@ -196,16 +214,21 @@ cmp -s "$actual" "$expected"
Expected: exact match to 23 paths. Drift requires another docs review.
- [ ] Add package/service/CLI tests; run initial RED:
- [ ] Add package/service/CLI tests and the seven-component matrix before any production repair. Run each focused case and record the exact command plus its expected contract failure. If current behavior already satisfies the contract, record it as characterization and do not manufacture a failure or edit that production module.
- [ ] Run the aggregate initial RED/characterization command:
```bash
PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH='active-runtime:active-runtime/app' \
python3 -m pytest tests/v2r1/test_ingress_service.py tests/test_v2r1_package_identity.py -q
python3 -m pytest tests/v2r1/test_ingress_service.py tests/test_v2r1_package_identity.py \
tests/v2r1/test_message_ledger_append.py tests/v2r1/test_memory_scope.py \
tests/v2r1/test_secret_scope.py tests/v2r1/test_audit.py \
tests/v2r1/test_hermes_profile.py tests/v2r1/test_cell_manager.py \
tests/v2r1/test_hermes_fabric.py -q
```
Expected RED: package metadata, ingress service, and identity tests do not exist.
Expected: package metadata, ingress service, and identity tests begin RED because their implementation is absent. Each matrix case is either a recorded contract RED or a recorded characterization; no production repair precedes its corresponding RED.
- [ ] Implement package/disabled processor boundary and replace imports in exactly Appendix A. Prove no shim/old import:
- [ ] Implement package/disabled processor boundary and replace imports in exactly Appendix A. Apply only the minimum runtime repairs demonstrated by matrix REDs; do not delete or weaken a matrix test. Prove no shim/old import:
```bash
test ! -e active_runtime
@@ -215,11 +238,15 @@ test -z "$(git grep -l -E '^(from|import) active_runtime\.app' -- tests || true)
- [ ] Run full collect and full suite:
```bash
PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH='active-runtime:active-runtime/app' \
python3 -m pytest tests/v2r1 tests/integration tests/test_v2r1_package_identity.py -q
PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH='active-runtime:active-runtime/app' python3 -m pytest --collect-only -q
PYTHONDONTWRITEBYTECODE=1 PYTEST_ADDOPTS='-p no:cacheprovider' PYTHONPATH='active-runtime:active-runtime/app' python3 -m pytest -q
PYTHONDONTWRITEBYTECODE=1 python3 -m compileall -q active-runtime/app/aios_v2r1
(cd openclaw/plugins/aios-tenant-boundary && npm test)
```
Expected GREEN: zero collection/import errors and complete suite passes; pre-existing failures still block B2.
Expected GREEN: the focused carried suite reports at least `160 passed`, including every matrix bullet; full collection has zero import errors; the complete pytest suite, 15 Node tests, and Python compileall pass. Re-run B1's real isolated default/opt-in loader sequence and require the same hook evidence. Pre-existing failures still block B2.
- [ ] Build and inspect isolated wheel:
@@ -241,22 +268,29 @@ PY
Expected GREEN: one wheel, one V2R1 namespace, installed identity `aios_v2r1`.
- [ ] Stage six named paths plus exact explorer list, commit, append ledger:
- [ ] Stage the exact 43 paths (six package paths, Appendix A's 23 import paths, seven matrix runtime modules, and seven matrix-test paths), commit, append ledger:
```bash
actual="$(git rev-parse --git-path v2r1-active-runtime-imports.actual)"
git add active-runtime/pyproject.toml active-runtime/app/aios_v2r1/__init__.py \
active-runtime/app/aios_v2r1/ingress_service.py active-runtime/app/aios_v2r1/cli.py \
tests/v2r1/test_ingress_service.py tests/test_v2r1_package_identity.py
tests/v2r1/test_ingress_service.py tests/test_v2r1_package_identity.py \
active-runtime/app/aios_v2r1/message_ledger.py active-runtime/app/aios_v2r1/memory_scope.py \
active-runtime/app/aios_v2r1/secret_scope.py active-runtime/app/aios_v2r1/audit.py \
active-runtime/app/aios_v2r1/hermes_profile.py active-runtime/app/aios_v2r1/cell_manager.py \
active-runtime/app/aios_v2r1/hermes_fabric.py tests/v2r1/test_message_ledger_append.py \
tests/v2r1/test_memory_scope.py tests/v2r1/test_secret_scope.py tests/v2r1/test_audit.py \
tests/v2r1/test_hermes_profile.py tests/v2r1/test_cell_manager.py tests/v2r1/test_hermes_fabric.py
while IFS= read -r path; do git add -- "$path"; done <"$actual"
git commit -m "feat(v2r1): package canonical authenticated ingress"
```
### Review B2
- [ ] Require `git diff --check`, exactly 29 paths (six named plus Appendix A), exact-list/no-shim checks, full collect/full pytest, and wheel identity.
- [ ] Require `git diff --check`, exactly 43 paths (the original 29 plus the exact 14 matrix paths), exact-list/no-shim checks, focused matrix/carried tests, full collect/full pytest, Node 15, real loader default/opt-in, compileall, and wheel identity.
- [ ] Require B2 parent equals B1 in ledger.
- [ ] Prove mutation routes remain fail-closed until authenticated processors are installed.
- [ ] Prove every production matrix edit has a recorded pre-edit RED and every already-correct case is recorded as characterization. The partial R9 ledger must contain exactly `START,B1,B2`; B3/B4/B5 remain absent.
- [ ] After all prior B2 review checks pass, append B2 and verify the recorded commit identity:
```bash
@@ -0,0 +1,188 @@
# V2R1 R10 personal Weixin release-plan amendment
Status: reviewed design-only amendment. It adds D1, B6, and B7 to the R9 linear release plan. It does not authorize runtime, candidate, production, baseline, tag, or Fleet mutation.
Authority remains `docs/superpowers/plans/2026-07-14-v2r1-runtime-release-plan.md` at B2 tip `68d27879103990aac692207a8512b3725c6d0d41`. R9 `START` is `a9a0b4ae396e33c9bfc35a5f5694e5e354c3a47c`; accepted B1 is `57103c11d21192a50febe846d30cee6fc5da4ac1`; accepted B2 is `68d27879103990aac692207a8512b3725c6d0d41`. Their trees and path sets are immutable. This amendment does not relabel, squash, reorder, or absorb any of the original five blockers.
## Exact linear ledger
The only valid ancestry is:
```text
START -> B1 -> B2 -> D1 -> B3 -> B4 -> B5 -> B6 -> B7
```
`D1` is a non-blocker design dependency whose parent is exactly B2. `B3` parent is D1; every later blocker parent is the immediately preceding row. `START..B7` contains exactly eight non-merge commits: B1, B2, D1, B3, B4, B5, B6, B7. B1-B5 remain five distinct original blocker commits; B6 and B7 are distinct new blocker commits. No cherry-pick collapse, squash, fixup, merge commit, history rewrite, or force push is allowed.
B3/B4 retain every R9 evidence gate. In particular, unresolved `EVIDENCE_REQUIRED` for the exact provider `client_id -> message_id` binding stops B4 with no implementation, commit, or ledger append. B5 can land only after accepted B4. Even after B5 lands, every forward `stage`, `verify-stage`, or `promote` operation remains blocked until accepted B6 and B7 commits exist and all gates below pass.
## D1 design-only commit
D1 changes exactly these 23 paths and no others:
```text
docs/superpowers/plans/2026-07-15-v2r1-personal-weixin-r10-release-plan-amendment.md
contracts/personal-weixin/v1/SECURITY.md
contracts/personal-weixin/v1/golden-contract.schema.json
contracts/personal-weixin/v1/fixtures/01-tenant-account-entitlement.json
contracts/personal-weixin/v1/fixtures/02-readiness-ready.json
contracts/personal-weixin/v1/fixtures/03-readiness-field-missing.json
contracts/personal-weixin/v1/fixtures/04-readiness-stale.json
contracts/personal-weixin/v1/fixtures/05-readiness-plugin-digest-mismatch.json
contracts/personal-weixin/v1/fixtures/06-readiness-account-not-entitled.json
contracts/personal-weixin/v1/fixtures/07-readiness-account-not-running.json
contracts/personal-weixin/v1/fixtures/08-readiness-error.json
contracts/personal-weixin/v1/fixtures/09-web-login-start-request.json
contracts/personal-weixin/v1/fixtures/10-web-login-wait-request.json
contracts/personal-weixin/v1/fixtures/11-qr-pending.json
contracts/personal-weixin/v1/fixtures/12-need-verify-code.json
contracts/personal-weixin/v1/fixtures/13-invalid-verify-code.json
contracts/personal-weixin/v1/fixtures/14-expired.json
contracts/personal-weixin/v1/fixtures/15-failed.json
contracts/personal-weixin/v1/fixtures/16-already-connected.json
contracts/personal-weixin/v1/fixtures/17-confirmed.json
contracts/personal-weixin/v1/fixtures/18-unknown-upstream-fail-closed.json
contracts/personal-weixin/v1/fixtures/19-authenticated-inbound-event.json
contracts/personal-weixin/v1/fixtures/20-unbind-unsupported.json
```
D1 validation requires JSON parse, Draft 2020-12 schema validation of every fixture, negative mutation tests, exact 23-path comparison, `git diff --check`, placeholder/secret scan, and confirmation that no runtime/deploy/release-active path changed.
## Preserved R9 blocker allowlists
B1 and B2 are already accepted at the immutable SHAs above. Their actual commit path sets, parent SHAs, and trees are the allowlists; compare with `git diff-tree --no-commit-id --name-only -r <sha>` and reject any mismatch. B1 has six actual paths under `openclaw/plugins/aios-tenant-boundary/`; B2 has exactly 43 actual package/import/matrix paths. No later blocker may amend them.
B3 changes exactly these 15 R9 paths:
```text
active-runtime/app/aios_v2r1/identity.py
active-runtime/app/aios_v2r1/tenant_context.py
active-runtime/app/aios_v2r1/ingress_service.py
active-runtime/app/aios_v2r1/bridge_auth.py
active-runtime/app/aios_v2r1/weixin_projection.py
fixes/patch_openclaw_weixin_v2r1_ingress.py
tests/v2r1/test_bridge_auth.py
tests/v2r1/test_weixin_projection.py
tests/integration/test_v2r1_bridge_transport.py
tests/test_weixin_v2r1_ingress_bridge.py
tests/fixtures/weixin_v2r1/text.json
tests/fixtures/weixin_v2r1/quoted-text.json
tests/fixtures/weixin_v2r1/image-with-text.json
tests/fixtures/weixin_v2r1/unresolved-reference.json
release/v2r1c0-rc1/weixin-bridge-evidence.json
```
B4 changes exactly these 19 R9 paths and remains evidence-stopped until both R9 evidence files are accepted:
```text
active-runtime/app/aios_v2r1/outbox.py
active-runtime/app/aios_v2r1/hermes_http.py
active-runtime/app/aios_v2r1/hermes_worker.py
active-runtime/app/aios_v2r1/outbound_receipt.py
active-runtime/app/aios_v2r1/ingress_service.py
active-runtime/app/aios_v2r1/message_ledger.py
tests/v2r1/test_outbox.py
tests/v2r1/test_hermes_http.py
tests/v2r1/test_outbound_receipt.py
tests/integration/test_v2r1_delivery_flow.py
fixes/patch_openclaw_weixin_outbound_quotes.py
fixes/patch_openclaw_weixin_outbound_receipts.py
openclaw/plugins/aios-tenant-boundary/dist/receipt-client.js
openclaw/plugins/aios-tenant-boundary/dist/outbound-receipt-journal.js
openclaw/plugins/aios-tenant-boundary/test/receipt-client.test.js
tests/test_weixin_outbound_quote_timestamp.py
tests/test_weixin_outbound_receipt_boundary.py
release/v2r1c0-rc1/hermes-http-contract-evidence.json
release/v2r1c0-rc1/weixin-outbound-binding-evidence.json
```
B5 changes exactly these 14 R9 paths:
```text
deploy/v2r1/release_ctl.py
deploy/v2r1/capture_live_preflight.py
deploy/v2r1/install.sh
deploy/v2r1/validate.sh
deploy/v2r1/healthcheck.sh
deploy/v2r1/rollback.sh
deploy/v2r1/aios-v2r1-candidate-gateway.service
deploy/v2r1/aios-v2r1-production-gateway.service
deploy/v2r1/aios-v2r1-ingress.service
deploy/v2r1/aios-v2r1-hermes-worker.service
deploy/v2r1/openclaw-candidate.json5
release/v2r1c0-rc1/FLEET_UNSUPPORTED
tests/deploy/test_v2r1_release_contract.py
release/v2r1c0-rc1/RELEASE.md
```
## B6: evidence-bound OpenClaw personal-Weixin candidate
B6 changes exactly these 13 paths:
```text
openclaw/candidates/personal-weixin-r10/README.md
openclaw/candidates/personal-weixin-r10/manifest.json
openclaw/candidates/personal-weixin-r10/src/golden-contract.js
openclaw/candidates/personal-weixin-r10/src/gateway-inbound-auth.js
openclaw/candidates/personal-weixin-r10/plugin/openclaw.plugin.json
openclaw/candidates/personal-weixin-r10/plugin/index.js
fixes/patch_openclaw_personal_weixin_login.py
fixes/patch_openclaw_personal_weixin_gateway_auth.py
openclaw/candidates/personal-weixin-r10/test/golden-contract.test.js
openclaw/candidates/personal-weixin-r10/test/gateway-auth.test.js
tests/personal_weixin/test_golden_fixtures.py
tests/integration/test_personal_weixin_gateway_contract.py
release/v2r1c0-rc1/personal-weixin-b6-evidence.json
```
Before B6 code, evidence records exact installed upstream source commit, every patch target path/digest, plugin loader version/digest, Gateway auth owner, credential files, and candidate-only state/config. Stop on inference or drift. Tests are RED before implementation and prove: required nonempty opaque `sessionKey`; no account fallback; exact canonical statuses and readiness shapes; unknown status fail-closed; 4-8 digit one-shot `verifyCode` with no persistence/log/response; exact event body with no authentication/tenant/user/session assertions; transport signature/skew/nonce/replay checks; exact 501 zero mutation; byte-idempotent patches; build/runtime inspect in isolated candidate state. The B6 evidence JSON is produced only after these gates pass.
## B7: tenant-scoped Portal adapter and durable challenge
B7 changes exactly these 13 paths:
```text
active-runtime/app/aios_v2r1/personal_weixin/official_login.py
active-runtime/app/aios_v2r1/personal_weixin/adapter.py
portal/personal_weixin/views.py
portal/personal_weixin/urls.py
portal/personal_weixin/services.py
portal/personal_weixin/templates/personal_weixin/bind.html
tests/personal_weixin/test_adapter.py
tests/personal_weixin/test_portal_views.py
tests/integration/test_personal_weixin_golden_flow.py
release/v2r1c0-rc1/personal-weixin-source-baseline.json
release/v2r1c0-rc1/personal-weixin-source-baseline.sha256
release/v2r1c0-rc1/personal-weixin-canary-contract.json
release/v2r1c0-rc1/personal-weixin-rollback-contract.json
```
Tests are RED before B7 implementation and prove: Django authentication/CSRF; server-session tenant resolution; browser `accountId` treated only as a preallocated-slot reference and current entitlement rechecked on every list/readiness/start/wait/status; exact `sessionKey` forwarding without fallback; live canonical readiness; authenticated Gateway transport context separate from the six-field event body; durable challenge record owning tenant/user/account/session digest; hash-only secret storage; transactional/CAS single consumption; exact replay idempotence; altered replay/concurrency/expiry/cross-tenant denial with zero mutation; no provider token/raw iLink/verify code/challenge code persistence or leakage; size/rate/no-store controls; restart; and unbind 501 before all mutations/calls/jobs.
## Canonical contract gates
`sessionKey` is nonempty and exact. Canonical results are:
```text
qr_pending connected=false alreadyConnected=false terminal=false needsVerifyCode=false
need_verifycode connected=false alreadyConnected=false terminal=false needsVerifyCode=true
invalid_verifycode connected=false alreadyConnected=false terminal=false needsVerifyCode=true
expired connected=false alreadyConnected=false terminal=true needsVerifyCode=false
failed connected=false alreadyConnected=false terminal=true needsVerifyCode=false
already_connected connected=false alreadyConnected=true terminal=true needsVerifyCode=false
confirmed connected=true alreadyConnected=false terminal=true needsVerifyCode=false
```
`binded_redirect` maps to `already_connected`. An unknown upstream status maps to canonical `failed/errorCode=unknown_upstream_status`. Readiness always uses the same complete shape: missing fields, stale evidence, plugin digest mismatch, absent entitlement, not-running account, and Gateway error all produce `ready=false`; only a live authenticated all-green snapshot produces `ready=true/code=ready`.
The inbound event body is exactly `eventId/accountId/senderId/chatId/challengeCode/observedAt`; transport authentication and server-resolved tenant/user/session context are not body fields. Fixture secrets are test inputs only. Raw `verifyCode` and `challengeCode` never enter persistence, logs, responses, URLs, analytics, traces, metrics labels, or job arguments. Raw `sessionKey` appears only where the golden login request/result contract requires it; it is never persisted or logged and is forwarded byte-exact without fallback.
## Merge, canary, promotion, and rollback gates
For each D1/B3-B7 row: require the exact parent row SHA, one non-merge commit, exact sorted path allowlist, clean `diff --check`, focused and carried tests, reviewer acceptance, then append `label/SHA/parent/tree` to the worktree-local ledger. No row may start from an unaccepted predecessor.
After B7, rebuild once from the accepted B7 tree. Stage one isolated candidate using B5 controls and no active configuration. Canary exactly one explicitly entitled `(tenant-canary, account-canary)` pair. Before traffic, require live Gateway/plugin/account readiness with expected digests and freshness. Exercise all golden outcomes, invalid and unknown fail-closed paths, restart, concurrent waits, atomic challenge consumption, exact replay, altered replay, expiry, unauthenticated event, tenant/account mismatch, cross-tenant IDOR, secret redaction, and exact end-to-end unbind 501 with zero durable/runtime/upstream/job mutation. Any missing evidence, extra account, digest drift, stale snapshot, noncanonical status, secret observation, or mutation fails the canary and blocks promotion.
Promotion accepts only the exact artifact/config/plugin/source digests verified by canary; it never rebuilds, repatches, switches branches, edits `VERSION`/`baseline/current.json`, creates a tag, invokes Fleet, or changes unrelated services. Candidate and production artifacts must be byte-identical. B5 live preflight is recaptured and reconciled immediately before `verify-stage` and `promote`.
Rollback is armed before canary with the immediately preceding production artifact/config/plugin digests, units, credentials metadata, and health contract. On any gate failure, STOP blocks forward actions; invoke only B5's recorded rollback-only path. Restore byte-identical prior artifacts/config, never the candidate rebuild; do not delete STOP automatically; do not roll back the database absent a separately approved incompatibility plan. Verify previous digests, listeners, units, secret paths/modes, canonical readiness, and tenant/account isolation. If verification fails, record `ROLLBACK_FAILED`, keep STOP, and escalate without forward mutation.
@@ -19,6 +19,7 @@ Alternative C remains the selected architecture because active full-execution au
- B3 stops unless the installed Weixin payload shape, patch target hashes, authority-map ownership, and transport signing boundary are evidenced.
- B4 stops unless Hermes request/response schema, router ownership, downstream timeout, idempotency behavior, and exact outbound message binding are evidenced.
- B5 stops unless B3 and B4 are accepted and the reconciled sentinel plus the V1R4C6 live-runtime snapshot are current for the candidate digest.
- The R9 D0 documentation correction becomes ledger `START` without consuming a blocker. It expands B2 executable coverage only; all authenticated transport, evidence, terminal, isolation, and release gates remain unchanged.
The present outbound evidence is insufficient, so B4 and B5 are currently stopped. This document defines the fail-closed path; it does not claim the missing evidence exists.
@@ -130,6 +131,12 @@ The release package has one V2R1 import identity: `aios_v2r1`. There is no `acti
`active-runtime/pyproject.toml` packages `aios_v2r1` from `active-runtime/app`. Repository tests use `PYTHONPATH='active-runtime:active-runtime/app'`: legacy modules import as `app.*`, while V2R1 imports as `aios_v2r1.*`. The implementation plan enumerates all 23 known `active_runtime.app` test imports and replaces them explicitly. Full collection, full pytest, and wheel inspection must prove the wheel contains one V2R1 namespace and that source and installed imports resolve to `aios_v2r1` only.
Packaging acceptance is also the executable completeness gate for seven preserved runtime components. Message Ledger must prove exact-ID/no-fallback semantics, idempotency/conflict, tenant/cell/direction isolation, expiry/purge/restart/concurrency, and sanitized SQLite errors. Memory must prove every namespace, cross-scope denial, shared-artifact read-only behavior, qualified-key traversal rejection, and disabled search. Secret must prove reference, broker path, scope, kind, version, identifier, no-value-leakage, and fail-closed broker behavior. Audit must prove genesis and tamper/delete/reorder detection, Unicode canonicalization, concurrency, restart, validation, tenant isolation, and sanitized SQLite errors. Hermes Profile must prove exact environment, absolute `0700` roots, traversal/symlink denial, scope, idempotence, and no environment inheritance. Cell Manager/CLI must prove one cell per tenant, an explicit state graph, atomic concurrent persistence, corrupt-registry fail-closed behavior, stable JSON/exit codes, and restart. Fabric must prove that claims return `running` consistently with durable state, isolation, one concurrent claimant, non-claimable states, restart, and sanitized SQLite errors.
The original 29 B2 paths are extended by exactly fourteen reviewed paths: seven production modules (`message_ledger.py`, `memory_scope.py`, `secret_scope.py`, `audit.py`, `hermes_profile.py`, `cell_manager.py`, and `hermes_fabric.py`) and seven matrix tests (`test_message_ledger_append.py`, `test_memory_scope.py`, `test_secret_scope.py`, `test_audit.py`, `test_hermes_profile.py`, `test_cell_manager.py`, and `test_hermes_fabric.py`). `cli.py` remains one of the original package paths. A production module changes only after its focused test has been observed failing for the intended contract; already-correct behavior is a characterization and must not be weakened merely to manufacture a RED.
The B2 review requires exactly 43 changed paths, at least 160 passing carried V2R1/integration/package tests with every matrix bullet represented, zero full-collection errors, a passing complete pytest suite, the 15 Node tests, the real isolated loader in default and opt-in modes, compileall, wheel identity, and a clean tree. These coverage additions do not authorize B3 bridge, B4 Hermes HTTP/outbound, B5 deployment, active service, main, tag, `VERSION`, or baseline mutation.
## Exact projection, quote, and attachment authority
`WeixinProjection` runs in Python after transport authentication. It preserves numeric Weixin IDs from the JSON token as decimal strings and rejects a value that has already passed through unsafe JavaScript-number conversion.
@@ -284,4 +291,4 @@ Preparation or release stops when any of these is true:
## Acceptance
The design is review-complete when the plan contains exactly five blocker commits, dynamically anchors their parent chain at B1 start, enumerates all legacy import edits, and has executable RED/GREEN/review gates. Runtime release acceptance additionally requires all B3/B4/B5 evidence gates, the full Python collection and suite, the real Node/OpenClaw loader harness, exact receipt semantics, isolated candidate staging, deterministic artifacts, current live evidence, and rehearsed constrained rollback.
The design is review-complete when the plan contains exactly five blocker commits, dynamically anchors their parent chain at B1 start, enumerates all legacy import edits, and has executable RED/GREEN/review gates. R9 preparation additionally requires D0 as `START`, B1 as its child, B2 as one 43-path child commit, the full seven-component executable matrix, and a partial ledger containing exactly `START,B1,B2`. Runtime release acceptance additionally requires all B3/B4/B5 evidence gates, the full Python collection and suite, the real Node/OpenClaw loader harness, exact receipt semantics, isolated candidate staging, deterministic artifacts, current live evidence, and rehearsed constrained rollback.
+35 -10
View File
@@ -1,31 +1,48 @@
import {
classifyRun,
decideToolCall,
resolveSenderId,
runKey,
} from "./policy.js";
const activeRuns = new Map();
const MAX_RUN_AGE_MS = 60 * 60 * 1000;
function prune(now = Date.now()) {
function prune(activeRuns, now = Date.now()) {
for (const [key, value] of activeRuns.entries()) {
if (now - value.recordedAt > MAX_RUN_AGE_MS) activeRuns.delete(key);
}
}
function approvalTimeout(config) {
const value = config.approvalTimeoutMs;
return Number.isInteger(value) && value >= 1000 && value <= 300000
? value
: 60000;
}
export default {
id: "aios-tenant-boundary",
name: "AIOS Tenant Boundary",
description: "Fail-closed user-origin and side-effect policy for Tenant Cells.",
register(api) {
const activeRuns = new Map();
api.on(
"before_agent_run",
async (event, ctx) => {
prune();
prune(activeRuns);
const config = event.context?.pluginConfig ?? api.pluginConfig ?? {};
const trust = classifyRun(ctx, config);
const key = runKey(event, ctx);
const existing = key ? activeRuns.get(key) : undefined;
if (existing && existing.senderId !== trust.senderId) {
return {
outcome: "block",
reason: "run_scope_sender_conflict",
message: "This run scope is already bound to a different authenticated sender.",
};
}
if (key) {
activeRuns.set(key, { ...trust, recordedAt: Date.now() });
}
@@ -49,10 +66,14 @@ export default {
api.on(
"before_tool_call",
async (event, ctx) => {
prune();
prune(activeRuns);
const config = event.context?.pluginConfig ?? api.pluginConfig ?? {};
const key = runKey(event, ctx);
const trust = key ? activeRuns.get(key) : undefined;
let trust = key ? activeRuns.get(key) : undefined;
const senderId = resolveSenderId(ctx);
if (trust && senderId && trust.senderId !== senderId) {
trust = { trusted: false, reason: "run_scope_sender_conflict" };
}
const decision = decideToolCall(event.toolName, trust, config);
if (decision.action === "block") {
@@ -67,7 +88,7 @@ export default {
title: `Approve ${String(event.toolName)}`,
description: "Authenticated user requested a side-effecting tool call in this Tenant Cell.",
severity: "warning",
timeoutMs: Number(config.approvalTimeoutMs ?? 60000),
timeoutMs: approvalTimeout(config),
allowedDecisions: ["allow-once", "deny"],
pluginId: "aios-tenant-boundary",
},
@@ -78,9 +99,13 @@ export default {
{ priority: 100 },
);
api.on("agent_end", async (event, ctx) => {
const key = runKey(event, ctx);
if (key) activeRuns.delete(key);
});
api.on(
"agent_end",
async (event, ctx) => {
const key = runKey(event, ctx);
if (key) activeRuns.delete(key);
},
{ priority: 100 },
);
},
};
+22 -2
View File
@@ -66,6 +66,26 @@ export function decideToolCall(toolName, runTrust, config = {}) {
}
export function runKey(event = {}, ctx = {}) {
const key = event.runId || ctx.runId;
return typeof key === "string" && key.trim() ? key.trim() : undefined;
const normalize = (value) => typeof value === "string" && value.trim()
? value.trim()
: undefined;
const sameField = (eventValue, contextValue) => {
const fromEvent = normalize(eventValue);
const fromContext = normalize(contextValue);
return {
valid: !(fromEvent && fromContext && fromEvent !== fromContext),
value: fromEvent || fromContext,
};
};
const runId = sameField(event.runId, ctx.runId);
const agentId = sameField(event.agentId, ctx.agentId);
const sessionId = sameField(event.sessionId, ctx.sessionId);
const sessionKey = sameField(event.sessionKey, ctx.sessionKey);
if (!runId.valid || !agentId.valid || !sessionId.valid || !sessionKey.valid) {
return undefined;
}
const selectedSession = sessionId.value || sessionKey.value;
return runId.value && agentId.value && selectedSession
? JSON.stringify([runId.value, agentId.value, selectedSession])
: undefined;
}
@@ -2,7 +2,7 @@
"id": "aios-tenant-boundary",
"name": "AIOS Tenant Boundary",
"description": "Enforces AIOS V2R1 tenant provenance and tool policy boundaries.",
"version": "2.1.0-rc1",
"version": "2.1.0-rc.1",
"entry": "dist/index.js",
"configSchema": {
"type": "object",
@@ -12,7 +12,14 @@
"trustedSenderIds": { "type": "array", "items": { "type": "string" }, "default": [] },
"blockedSenderIds": { "type": "array", "items": { "type": "string" }, "default": [] },
"readOnlyTools": { "type": "array", "items": { "type": "string" } },
"requireApprovalForHumanSideEffects": { "type": "boolean", "default": true }
"requireApprovalForHumanSideEffects": { "type": "boolean", "default": true },
"blockUnknownAgentRuns": { "type": "boolean", "default": false },
"approvalTimeoutMs": {
"type": "integer",
"minimum": 1000,
"maximum": 300000,
"default": 60000
}
}
}
}
@@ -0,0 +1,64 @@
import assert from "node:assert/strict";
import { readFile } from "node:fs/promises";
import test from "node:test";
import plugin from "../dist/index.js";
const packageJson = JSON.parse(
await readFile(new URL("../package.json", import.meta.url), "utf8"),
);
const manifest = JSON.parse(
await readFile(new URL("../openclaw.plugin.json", import.meta.url), "utf8"),
);
function registrations(config = {}) {
const seen = [];
plugin.register({
pluginConfig: config,
on(name, handler, options) {
seen.push({ name, handler, options });
},
});
return seen;
}
test("package, manifest, and executable test command are release compatible", () => {
assert.equal(packageJson.version, "2.1.0-rc.1");
assert.equal(manifest.version, "2.1.0-rc.1");
assert.equal(packageJson.scripts.test, "node --test test/*.test.js");
});
test("schema declares every runtime-read property and exact B1 bounds", () => {
const properties = manifest.configSchema.properties;
const runtimeKeys = [
"approvalTimeoutMs",
"blockUnknownAgentRuns",
"blockedSenderIds",
"readOnlyTools",
"requireApprovalForHumanSideEffects",
"requireExplicitTrustedSenders",
"trustedSenderIds",
];
assert.deepEqual(Object.keys(properties).sort(), runtimeKeys);
assert.deepEqual(properties.blockUnknownAgentRuns, {
type: "boolean",
default: false,
});
assert.deepEqual(properties.approvalTimeoutMs, {
type: "integer",
minimum: 1000,
maximum: 300000,
default: 60000,
});
});
test("all three hooks register exactly once at priority 100", () => {
const seen = registrations();
assert.deepEqual(
seen.map(({ name }) => name).sort(),
["agent_end", "before_agent_run", "before_tool_call"],
);
for (const registration of seen) {
assert.deepEqual(registration.options, { priority: 100 });
}
});
@@ -0,0 +1,50 @@
import assert from "node:assert/strict";
import { readFile } from "node:fs/promises";
import { realpath } from "node:fs/promises";
const [mode, inspectPath] = process.argv.slice(2);
assert.ok(["default", "opt-in"].includes(mode), `unsupported mode: ${mode}`);
assert.ok(inspectPath, "inspect JSON path is required");
const data = JSON.parse(await readFile(inspectPath, "utf8"));
const plugin = data.plugin;
assert.equal(plugin.id, "aios-tenant-boundary");
assert.equal(plugin.version, "2.1.0-rc.1");
assert.equal(plugin.status, "loaded");
assert.equal(plugin.activated, true);
assert.equal(await realpath(plugin.rootDir), await realpath(new URL("..", import.meta.url)));
assert.equal(await realpath(plugin.source), await realpath(new URL("../dist/index.js", import.meta.url)));
assert.equal(data.install.source, "path");
assert.equal(await realpath(data.install.installPath), await realpath(plugin.rootDir));
const hooks = data.typedHooks
.map(({ name, priority }) => ({ name, priority }))
.sort((a, b) => a.name.localeCompare(b.name));
const expectedNames = mode === "default"
? ["before_tool_call"]
: ["agent_end", "before_agent_run", "before_tool_call"];
assert.deepEqual(hooks.map(({ name }) => name), expectedNames);
assert.ok(hooks.every(({ priority }) => priority === 100));
const diagnostics = data.diagnostics ?? [];
assert.equal(
diagnostics.filter(({ level, severity }) =>
["error", "fatal"].includes(String(level ?? severity).toLowerCase())).length,
0,
);
if (mode === "default") {
assert.deepEqual(
diagnostics.map(({ level, message }) => ({ level, message })).sort((a, b) => a.message.localeCompare(b.message)),
["agent_end", "before_agent_run"].map((hook) => ({
level: "warn",
message: `typed hook "${hook}" blocked because non-bundled plugins must set plugins.entries.aios-tenant-boundary.hooks.allowConversationAccess=true`,
})).sort((a, b) => a.message.localeCompare(b.message)),
);
} else {
assert.deepEqual(diagnostics, []);
}
const properties = plugin.configJsonSchema?.properties ?? {};
for (const key of ["approvalTimeoutMs", "blockUnknownAgentRuns"]) {
assert.ok(Object.hasOwn(properties, key), `missing runtime schema property: ${key}`);
}
@@ -0,0 +1,349 @@
import assert from "node:assert/strict";
import test from "node:test";
import plugin from "../dist/index.js";
import {
classifyRun,
decideToolCall,
resolveSenderId,
runKey,
} from "../dist/policy.js";
function registration(config = {}) {
const handlers = new Map();
plugin.register({
pluginConfig: config,
on(name, handler) {
handlers.set(name, handler);
},
});
return {
beforeAgentRun: handlers.get("before_agent_run"),
beforeToolCall: handlers.get("before_tool_call"),
agentEnd: handlers.get("agent_end"),
};
}
function agentContext(id, senderId) {
return {
runId: `run-${id}`,
agentId: "agent-main",
sessionId: "session-uuid-123",
sessionKey: "agent:main:weixin:direct:user",
...(senderId ? { senderId } : {}),
};
}
function toolContext(id, extra = {}) {
return {
runId: `run-${id}`,
agentId: "agent-main",
sessionId: "session-uuid-123",
sessionKey: "agent:main:weixin:direct:user",
...extra,
};
}
test("sender classification covers missing, blocked, allowlisted, and automation", () => {
assert.equal(resolveSenderId({ channelContext: { sender: { id: " nested " } } }), "nested");
assert.deepEqual(classifyRun({}, {}), {
trusted: false,
origin: "system_or_unknown",
reason: "sender_id_missing",
});
assert.equal(
classifyRun({ senderId: "blocked" }, { blockedSenderIds: ["blocked"] }).reason,
"sender_blocked",
);
assert.equal(
classifyRun(
{ senderId: "alice" },
{ requireExplicitTrustedSenders: true, trustedSenderIds: ["alice"] },
).trusted,
true,
);
assert.deepEqual(classifyRun({ jobId: "job-1" }, {}), {
trusted: false,
origin: "automation",
reason: "job_id_present",
});
});
test("unknown agent runs default to pass but fail closed when explicitly enabled", async () => {
const defaultHandlers = registration();
assert.deepEqual(
await defaultHandlers.beforeAgentRun({}, agentContext("default")),
{ outcome: "pass" },
);
const enforced = registration({ blockUnknownAgentRuns: true });
const denied = await enforced.beforeAgentRun({}, agentContext("denied"));
assert.equal(denied.outcome, "block");
assert.equal(denied.reason, "sender_id_missing");
assert.doesNotMatch(denied.reason + denied.message, /undefined|\r|\n/);
assert.deepEqual(
await enforced.beforeAgentRun({}, { ...agentContext("job"), jobId: "job-1" }),
{ outcome: "pass" },
);
});
test("read-only tools pass without provenance and side effects fail with a stable reason", async () => {
assert.deepEqual(decideToolCall("read", undefined, {}), {
action: "allow",
code: "read_only",
});
assert.deepEqual(decideToolCall("shell", undefined, {}), {
action: "block",
code: "run_provenance_missing",
});
const handlers = registration();
assert.equal(await handlers.beforeToolCall({ toolName: "read" }, {}), undefined);
const blocked = await handlers.beforeToolCall({ toolName: "shell" }, {});
assert.equal(blocked.block, true);
assert.equal(
blocked.blockReason,
"AIOS V2R1 origin policy denied tool call: run_provenance_missing",
);
});
test("trusted human side effects request approval with exact timeout validation", async () => {
for (const [configured, expected] of [
[undefined, 60000],
[1000, 1000],
[300000, 300000],
[999, 60000],
[300001, 60000],
[60000.5, 60000],
["1000", 60000],
[null, 60000],
]) {
const config = {
requireExplicitTrustedSenders: true,
trustedSenderIds: ["alice"],
};
if (configured !== undefined) config.approvalTimeoutMs = configured;
const handlers = registration(config);
const id = `approval-${String(configured)}`;
await handlers.beforeAgentRun({}, agentContext(id, "alice"));
const result = await handlers.beforeToolCall({ toolName: "shell" }, toolContext(id));
assert.equal(result.requireApproval.timeoutMs, expected, String(configured));
assert.deepEqual(result.requireApproval.allowedDecisions, ["allow-once", "deny"]);
}
});
test("active run provenance is isolated by registration, run, and sender", async () => {
const config = {
requireExplicitTrustedSenders: true,
trustedSenderIds: ["alice", "bob"],
};
const first = registration(config);
const second = registration(config);
await first.beforeAgentRun({}, agentContext("shared", "alice"));
const otherSender = await first.beforeToolCall(
{ toolName: "shell" },
toolContext("shared", { senderId: "bob" }),
);
assert.equal(otherSender.block, true);
assert.match(otherSender.blockReason, /run_scope_sender_conflict$/);
const otherRegistration = await second.beforeToolCall(
{ toolName: "shell" },
toolContext("shared"),
);
assert.equal(otherRegistration.block, true);
assert.match(otherRegistration.blockReason, /run_provenance_missing$/);
});
test("agent_end cleanup removes only the matching shared scope", async () => {
const handlers = registration({
requireExplicitTrustedSenders: true,
trustedSenderIds: ["alice", "bob"],
});
await handlers.beforeAgentRun({}, agentContext("alice-scope", "alice"));
await handlers.beforeAgentRun({}, agentContext("bob-scope", "bob"));
await handlers.agentEnd({}, toolContext("alice-scope"));
const alice = await handlers.beforeToolCall(
{ toolName: "shell" },
toolContext("alice-scope"),
);
assert.equal(alice.block, true);
const bob = await handlers.beforeToolCall(
{ toolName: "shell" },
toolContext("bob-scope"),
);
assert.ok(bob.requireApproval);
});
test("run provenance expires after the bounded age", async (t) => {
const originalNow = Date.now;
t.after(() => { Date.now = originalNow; });
let now = 1_000_000;
Date.now = () => now;
const handlers = registration({
requireExplicitTrustedSenders: true,
trustedSenderIds: ["alice"],
});
await handlers.beforeAgentRun({}, agentContext("expiring", "alice"));
now += 60 * 60 * 1000 + 1;
const result = await handlers.beforeToolCall(
{ toolName: "shell" },
toolContext("expiring"),
);
assert.equal(result.block, true);
assert.match(result.blockReason, /run_provenance_missing$/);
});
test("host-shaped lifecycle keeps distinct sessionId and sessionKey across all hooks", async () => {
const handlers = registration({
requireExplicitTrustedSenders: true,
trustedSenderIds: ["alice"],
});
const before = agentContext("host-shape", "alice");
const tool = toolContext("host-shape");
const end = toolContext("host-shape");
for (const ctx of [before, tool, end]) {
assert.equal(ctx.sessionId, "session-uuid-123");
assert.equal(ctx.sessionKey, "agent:main:weixin:direct:user");
assert.equal(ctx.runId, "run-host-shape");
assert.equal(ctx.agentId, "agent-main");
}
await handlers.beforeAgentRun({}, before);
const result = await handlers.beforeToolCall(
{ toolName: "shell" },
tool,
);
assert.ok(result.requireApproval);
assert.equal(result.block, undefined);
await handlers.agentEnd({}, end);
const cleaned = await handlers.beforeToolCall({ toolName: "shell" }, tool);
assert.equal(cleaned.block, true);
assert.match(cleaned.blockReason, /run_provenance_missing$/);
});
test("shared scope collision fails closed without overwriting original sender", async () => {
const handlers = registration({
requireExplicitTrustedSenders: true,
trustedSenderIds: ["alice", "bob"],
});
assert.deepEqual(
await handlers.beforeAgentRun({}, agentContext("collision", "alice")),
{ outcome: "pass" },
);
const collision = await handlers.beforeAgentRun(
{},
agentContext("collision", "bob"),
);
assert.equal(collision.outcome, "block");
assert.equal(collision.reason, "run_scope_sender_conflict");
assert.doesNotMatch(collision.message, /alice|bob/);
const original = await handlers.beforeToolCall(
{ toolName: "shell" },
toolContext("collision"),
);
assert.ok(original.requireApproval);
});
test("shared scope normalization is exact and incomplete scope fails closed", async () => {
const bothAliases = runKey({}, {
runId: " run-1 ",
agentId: " agent-1 ",
sessionId: " session-uuid-123 ",
sessionKey: " agent:main:weixin:direct:user ",
});
assert.equal(
bothAliases,
JSON.stringify(["run-1", "agent-1", "session-uuid-123"]),
);
const fallback = runKey(
{ sessionKey: " agent:main:weixin:direct:user " },
{
runId: "run-1",
agentId: "agent-1",
sessionKey: "agent:main:weixin:direct:user",
},
);
assert.equal(
fallback,
JSON.stringify(["run-1", "agent-1", "agent:main:weixin:direct:user"]),
);
assert.equal(runKey(
{ sessionId: "event-session" },
{
runId: "run-1",
agentId: "agent-1",
sessionId: "context-session",
sessionKey: "agent:main:weixin:direct:user",
},
), undefined);
assert.equal(runKey({}, {
runId: "run-1",
agentId: "agent-1",
sessionKey: "agent:main:weixin:direct:user",
sessionId: "session-uuid-123",
}), bothAliases);
assert.equal(runKey({}, {
runId: "run-1",
agentId: "agent-1",
sessionKey: "agent:main:weixin:direct:user",
}), fallback);
assert.equal(runKey({}, {
agentId: "agent-1", sessionId: "session-uuid-123",
}), undefined);
const handlers = registration();
const missing = await handlers.beforeToolCall(
{ toolName: "shell" },
{ agentId: "agent-1", sessionId: "session-uuid-123" },
);
assert.equal(missing.block, true);
assert.match(missing.blockReason, /run_provenance_missing$/);
});
test("sessionKey fallback requires sessionId absent from both hook contexts", async () => {
const handlers = registration({
requireExplicitTrustedSenders: true,
trustedSenderIds: ["alice"],
});
const before = {
runId: "run-fallback",
agentId: "agent-main",
sessionKey: "agent:main:weixin:direct:user",
senderId: "alice",
};
const tool = {
runId: "run-fallback",
agentId: "agent-main",
sessionKey: "agent:main:weixin:direct:user",
};
await handlers.beforeAgentRun({}, before);
assert.ok((await handlers.beforeToolCall({ toolName: "shell" }, tool)).requireApproval);
const mismatchedSelection = await handlers.beforeToolCall(
{ toolName: "shell" },
{ ...tool, sessionId: "session-uuid-123" },
);
assert.equal(mismatchedSelection.block, true);
assert.match(mismatchedSelection.blockReason, /run_provenance_missing$/);
});
test("same-named event and context sessionId mismatch fails closed", async () => {
const handlers = registration({
requireExplicitTrustedSenders: true,
trustedSenderIds: ["alice"],
});
await handlers.beforeAgentRun({}, agentContext("event-mismatch", "alice"));
const result = await handlers.beforeToolCall(
{ toolName: "shell", sessionId: "different-session" },
toolContext("event-mismatch"),
);
assert.equal(result.block, true);
assert.match(result.blockReason, /run_provenance_missing$/);
});
@@ -0,0 +1,16 @@
{
"task_id": "TASK-CODEX-V2R1-PERSONAL-WEIXIN-B6B7-BOOTSTRAP-AND-CANARY-20260715T120000UTC8",
"phase": "A",
"type": "branch",
"branch": "patch/v2r1-personal-weixin-b6b7-impl-20260715T115309UTC8",
"base_ref": "design/v2r1-personal-weixin-golden-contract-20260715",
"base_commit": "d74229927b2dadca7719455984aea10d010c5123",
"created_by": "aiman",
"created_at": "2026-07-15T03:53:09Z",
"direct_main_push": false,
"owner_approval_required": false,
"owner_approved": true,
"owner_approval_scope": "Phase A bootstrap for owner-approved B6/B7 personal-Weixin implementation; no production, runtime, database, secret, tag, baseline, or controller/ChatGPT worktree changes.",
"production_change": false,
"user_action_required": false
}
@@ -0,0 +1,9 @@
{
"type": "branch",
"branch": "task/codex/v2r1-personal-weixin-b3-evidence-git-20260716T123454UTC8",
"base": "1d334b1ccaac54c3c3ec4dd84693ab68fc67cef3",
"created_by": "chatgpt-gitops-bot",
"created_at": "2026-07-16T04:35:34Z",
"direct_main_push": false,
"owner_approval_required": true
}
@@ -0,0 +1,44 @@
{
"schemaVersion": 1,
"contract": "personal-weixin-b4-outbound-quote-authority-amendment",
"ownerApproved": true,
"lineage": {
"foundationImplementationCommit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
"foundationResultCommit": "64a71b6199a8e2bd609ba145349e67b874888505",
"r10LineageCommit": "74e603c0c943501110577dbe3c7f6a0b18b8ec7d",
"evidenceGateCommit": "62a4067cf7a462ebe7cf29524238bc5d0a70b813",
"upstreamAuditCommit": "b2f5f531019a70801444987dbb97dc1f69cf47de"
},
"vendorClientIdMessageIdEvidence": {
"status": "absent",
"clientIdPromotable": false
},
"authorityPolicy": {
"outboundQuoteAuthority": "disabled",
"clientIdRole": "non-authoritative-correlation-only",
"authoritativeMessageIdSources": [
"validated-inbound-raw-wire.vendor_message_id"
],
"quotedSendWithoutAuthoritativeMessageId": "fail-closed",
"plainOutboundWithoutQuoteAuthority": "allowed-only-by-existing-b4-allowlist",
"syntheticMapping": false,
"timestampFallback": false,
"localSequenceFallback": false,
"contentHashFallback": false,
"outboundReceiptFallback": false
},
"implementationConstraints": {
"scopeExpansion": false,
"exactAllowlistRequired": true,
"missingAllowlistStatus": "BLOCKED_SCOPE",
"productionChangeAuthorized": false,
"canaryAuthorized": false,
"tagAuthorized": false,
"forcePushAuthorized": false
},
"production": {
"baseline": "V1R4C6",
"changed": false,
"releaseStatus": "paused"
}
}
@@ -0,0 +1,23 @@
{
"schemaVersion": 1,
"contract": "personal-weixin-r10-lineage-amendment",
"foundation": {
"node": "F0_PERSONAL_WEIXIN_FOUNDATION",
"implementationCommit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
"resultCommit": "64a71b6199a8e2bd609ba145349e67b874888505",
"imageDigest": "sha256:ed741d1f028c4bd507c6eda71fb1dcc11e0815c0107afcbfcbaaaaa23d5a11bb",
"status": "PASS"
},
"precedes": [
"B3",
"B4",
"B5",
"B6",
"B7"
],
"production": {
"baseline": "V1R4C6",
"changed": false,
"releaseStatus": "paused"
}
}
+652
View File
@@ -0,0 +1,652 @@
#!/usr/bin/env python3
"""Hardened WeChat outbound backend for AIOS.
Only the dedicated UCC adapter is a supported send backend. OpenClaw channel
discovery is retained for diagnostics, but generic OpenClaw chat or agent
delivery is never promoted to WeChat delivery.
"""
from __future__ import annotations
import json
import os
import re
import selectors
import signal
import subprocess
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, List, Optional, Tuple
OPENCLAW_BIN = os.environ.get("AIOS_OPENCLAW_BIN", "openclaw")
WECHAT_ACCOUNT_ID = os.environ.get("AIOS_WECHAT_ACCOUNT_ID", "")
WECHAT_CHANNEL = os.environ.get("AIOS_WECHAT_CHANNEL", "")
SEND_BACKEND = os.environ.get("AIOS_WECHAT_SEND_BACKEND", "")
UCC_ADAPTER_URL = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_URL", "http://127.0.0.1:18800")
UCC_ADAPTER_TOKEN = os.environ.get("AIOS_UCC_WECHAT_ADAPTER_TOKEN") or os.environ.get(
"AIOS_WECHAT_QR_ADAPTER_TOKEN", ""
)
UCC_SEND_PATH = os.environ.get("AIOS_UCC_WECHAT_SEND_PATH", "/api/internal/wechat/send")
_allowed_origins = os.environ.get("AIOS_UCC_WECHAT_ALLOWED_ORIGINS", "").strip()
UCC_ADAPTER_ALLOWED_ORIGINS = tuple(
origin.strip() for origin in _allowed_origins.split(",") if origin.strip()
) or (
"http://127.0.0.1:18800",
"http://[::1]:18800",
)
MAX_TARGET_BYTES = 512
MAX_CONTENT_BYTES = 32 * 1024
MAX_IDENTIFIER_BYTES = 512
MAX_METADATA_BYTES = 8 * 1024
MAX_METADATA_DEPTH = 32
MAX_METADATA_NODES = 1024
MAX_RESPONSE_BYTES = 64 * 1024
MAX_ENDPOINT_BYTES = 256
_DEFAULT_TIMEOUT_SECONDS = 10
_MAX_TIMEOUT_SECONDS = 30
_LEGACY_SEND_PATH = "/internal/channels/wechat/send"
_SAFE_PATH = re.compile(r"^/[A-Za-z0-9._~!$&'()*+,;=:@%/-]+$")
_MAX_PATH_DECODE_ROUNDS = 4
_OPENCLAW_TIMEOUT_RETURN_CODE = 124
_OPENCLAW_OUTPUT_LIMIT_RETURN_CODE = 125
_OPENCLAW_SPAWN_ERROR_RETURN_CODE = 126
def _bounded_timeout(value: str) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
return _DEFAULT_TIMEOUT_SECONDS
return min(max(parsed, 1), _MAX_TIMEOUT_SECONDS)
REQUEST_TIMEOUT_SECONDS = _bounded_timeout(os.environ.get("AIOS_UCC_WECHAT_TIMEOUT_SECONDS", "10"))
class BackendResult:
def __init__(self, ok: bool, backend: str, status: str, data: Dict[str, Any]) -> None:
self.ok = ok
self.backend = backend
self.status = status
self.data = data
def to_dict(self) -> Dict[str, Any]:
return {"ok": self.ok, "backend": self.backend, "status": self.status, **self.data}
class _BackendProtocolError(Exception):
"""Internal marker whose details must never cross the module boundary."""
def _openclaw_base_args() -> List[str]:
"""Return the OpenClaw 2026.6.11-compatible command prefix.
Gateway/auth global flags are not supported by the installed CLI and would
also expose credentials through the process argument vector.
"""
return [OPENCLAW_BIN]
def _terminate_process_tree(process: Any) -> None:
if os.name == "posix":
try:
os.killpg(process.pid, signal.SIGTERM)
except OSError:
pass
try:
process.wait(timeout=0.5)
except (OSError, subprocess.TimeoutExpired):
pass
time.sleep(0.05)
try:
os.killpg(process.pid, signal.SIGKILL)
except OSError:
pass
try:
process.wait(timeout=0.5)
except (OSError, subprocess.TimeoutExpired):
pass
return
if process.poll() is not None:
return
try:
process.terminate()
process.wait(timeout=1)
return
except (OSError, subprocess.TimeoutExpired):
pass
try:
process.kill()
process.wait(timeout=1)
except (OSError, subprocess.TimeoutExpired):
pass
def _run_openclaw(
args: List[str], timeout: int = 15
) -> subprocess.CompletedProcess[str]:
bounded_timeout = min(max(int(timeout), 1), _MAX_TIMEOUT_SECONDS)
popen_options: Dict[str, Any] = {
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
}
if os.name == "posix":
popen_options["start_new_session"] = True
elif hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"):
popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
try:
process = subprocess.Popen(args, **popen_options)
except (OSError, ValueError):
return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
selector = selectors.DefaultSelector()
streams = {"stdout": process.stdout, "stderr": process.stderr}
buffers = {"stdout": bytearray(), "stderr": bytearray()}
try:
for name, stream in streams.items():
if stream is not None:
selector.register(stream, selectors.EVENT_READ, name)
except (OSError, ValueError):
selector.close()
_terminate_process_tree(process)
for stream in streams.values():
if stream is not None:
stream.close()
return subprocess.CompletedProcess(args, _OPENCLAW_SPAWN_ERROR_RETURN_CODE, "", "")
deadline = time.monotonic() + bounded_timeout
total_bytes = 0
return_code: Optional[int] = None
try:
while selector.get_map():
remaining = deadline - time.monotonic()
if remaining <= 0:
return_code = _OPENCLAW_TIMEOUT_RETURN_CODE
break
events = selector.select(min(remaining, 0.1))
for key, _mask in events:
allowance = MAX_RESPONSE_BYTES - total_bytes
try:
chunk = os.read(key.fileobj.fileno(), min(8192, allowance + 1))
except OSError:
return_code = _OPENCLAW_SPAWN_ERROR_RETURN_CODE
break
if not chunk:
selector.unregister(key.fileobj)
key.fileobj.close()
continue
if len(chunk) > allowance:
return_code = _OPENCLAW_OUTPUT_LIMIT_RETURN_CODE
break
buffers[key.data].extend(chunk)
total_bytes += len(chunk)
if return_code is not None:
break
finally:
selector.close()
if return_code is not None:
_terminate_process_tree(process)
for stream in streams.values():
if stream is not None and not stream.closed:
stream.close()
return subprocess.CompletedProcess(args, return_code, "", "")
try:
completed_return_code = process.wait(timeout=1)
except subprocess.TimeoutExpired:
_terminate_process_tree(process)
return subprocess.CompletedProcess(args, _OPENCLAW_TIMEOUT_RETURN_CODE, "", "")
stdout = buffers["stdout"].decode("utf-8", errors="replace")
stderr = buffers["stderr"].decode("utf-8", errors="replace")
return subprocess.CompletedProcess(args, completed_return_code, stdout, stderr)
def _json_dict(text: str) -> Dict[str, Any]:
try:
value = json.loads(text or "{}")
except (RecursionError, TypeError, ValueError, json.JSONDecodeError):
return {}
return value if isinstance(value, dict) else {}
def discover_openclaw_wechat_channel() -> Optional[str]:
"""Discover an explicitly typed WeChat/Weixin channel for diagnostics."""
if WECHAT_CHANNEL:
return WECHAT_CHANNEL
try:
proc = _run_openclaw(_openclaw_base_args() + ["channels", "list", "--json"], timeout=15)
except (OSError, subprocess.SubprocessError, ValueError):
return None
if proc.returncode != 0:
return None
data = _json_dict(proc.stdout[:MAX_RESPONSE_BYTES])
candidates: List[str] = []
def walk(value: Any) -> None:
if isinstance(value, dict):
name = value.get("name") or value.get("channel") or value.get("id")
kind = value.get("type") or value.get("kind")
if isinstance(name, str) and isinstance(kind, str):
label = f"{name} {kind}".lower()
if "wechat" in label or "weixin" in label:
candidates.append(name)
for child in value.values():
walk(child)
elif isinstance(value, list):
for child in value:
walk(child)
walk(data)
return candidates[0] if candidates else None
def _auth_headers() -> Dict[str, str]:
headers = {"Content-Type": "application/json", "Accept": "application/json"}
if UCC_ADAPTER_TOKEN:
headers["Authorization"] = "Bearer " + UCC_ADAPTER_TOKEN
return headers
def _safe_path(value: Any) -> Optional[str]:
if not isinstance(value, str):
return None
path = value if value.startswith("/") else "/" + value
try:
encoded_path = path.encode("utf-8")
except UnicodeEncodeError:
return None
if len(encoded_path) > MAX_ENDPOINT_BYTES:
return None
if not _SAFE_PATH.fullmatch(path):
return None
decoded = path
for _ in range(_MAX_PATH_DECODE_ROUNDS):
try:
next_decoded = urllib.parse.unquote_to_bytes(decoded).decode("utf-8")
except UnicodeDecodeError:
return None
if next_decoded == decoded:
break
decoded = next_decoded
if "%" in decoded or "\\" in decoded or "\x00" in decoded:
return None
if not _SAFE_PATH.fullmatch(decoded):
return None
if any(segment in {".", ".."} for segment in decoded.split("/")):
return None
return path
def _unique_paths(paths: List[str]) -> List[str]:
out: List[str] = []
for candidate in paths:
path = _safe_path(candidate)
if path and path not in out:
out.append(path)
if len(out) == 2:
break
return out
def _ucc_send_paths() -> List[str]:
canonical = _safe_path(UCC_SEND_PATH) or "/api/internal/wechat/send"
return _unique_paths([canonical, _LEGACY_SEND_PATH])
def _canonical_origin(value: Any) -> Optional[str]:
if not isinstance(value, str) or not value or len(value) > 2048:
return None
if "\\" in value or any(ord(character) < 32 for character in value):
return None
try:
parsed = urllib.parse.urlsplit(value)
port = parsed.port
except (UnicodeError, ValueError):
return None
scheme = parsed.scheme.lower()
if scheme not in {"http", "https"} or not parsed.hostname:
return None
if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:
return None
try:
host = parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower()
except UnicodeError:
return None
if not host:
return None
if port is None:
port = 80 if scheme == "http" else 443
rendered_host = f"[{host}]" if ":" in host else host
return f"{scheme}://{rendered_host}:{port}"
def _adapter_base_url() -> Optional[str]:
origin = _canonical_origin(UCC_ADAPTER_URL)
allowed = {
canonical
for canonical in (_canonical_origin(value) for value in UCC_ADAPTER_ALLOWED_ORIGINS)
if canonical
}
return origin if origin and origin in allowed else None
class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(
self,
req: Any,
fp: Any,
code: int,
msg: str,
headers: Any,
newurl: str,
) -> None:
del req, fp, code, msg, headers, newurl
return None
def _open_no_redirect(request: urllib.request.Request, timeout: int) -> Any:
opener = urllib.request.build_opener(_NoRedirectHandler())
return opener.open(request, timeout=timeout)
def ucc_adapter_alive() -> bool:
base_url = _adapter_base_url()
if not base_url:
return False
request = urllib.request.Request(
base_url + "/health", headers=_auth_headers(), method="GET"
)
try:
with _open_no_redirect(request, timeout=min(5, REQUEST_TIMEOUT_SECONDS)) as response:
return 200 <= int(response.status) < 300
except (OSError, ValueError, urllib.error.URLError):
return False
def select_backend() -> Dict[str, Any]:
explicit = SEND_BACKEND.strip().lower()
if explicit:
if explicit == "ucc-adapter":
return {"backend": "ucc-adapter", "reason": "explicit"}
return {"backend": "unavailable", "reason": "unsupported_backend"}
if ucc_adapter_alive():
return {"backend": "ucc-adapter", "reason": "ucc_adapter_alive"}
return {"backend": "unavailable", "reason": "ucc_adapter_unavailable"}
def _post_ucc(path: str, payload: Dict[str, Any]) -> Tuple[int, Dict[str, Any]]:
base_url = _adapter_base_url()
safe_path = _safe_path(path)
if not base_url or not safe_path:
raise _BackendProtocolError("invalid adapter endpoint")
try:
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
except (TypeError, ValueError) as exc:
raise _BackendProtocolError("invalid request payload") from exc
request = urllib.request.Request(
base_url + safe_path,
data=body,
headers=_auth_headers(),
method="POST",
)
with _open_no_redirect(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
raw = response.read(MAX_RESPONSE_BYTES + 1)
if len(raw) > MAX_RESPONSE_BYTES:
raise _BackendProtocolError("adapter response too large")
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise _BackendProtocolError("adapter response encoding invalid") from exc
data = _json_dict(text)
if not data:
raise _BackendProtocolError("adapter response invalid")
return int(response.status), data
def _ucc_response_sent(data: Dict[str, Any]) -> bool:
return isinstance(data, dict) and data.get("ok") is True and data.get("status") == "sent"
def _ucc_result_is_ok(data: Dict[str, Any], dry_run: bool = False) -> bool:
expected_status = "dry_run" if dry_run else "sent"
return isinstance(data, dict) and data.get("ok") is True and data.get("status") == expected_status
def _utf8_size(value: str) -> Optional[int]:
try:
return len(value.encode("utf-8"))
except UnicodeEncodeError:
return None
def _valid_identifier(value: Any, required: bool = False) -> bool:
if not isinstance(value, str):
return False
if required and not value.strip():
return False
if not value:
return not required
if len(value) > MAX_IDENTIFIER_BYTES:
return False
size = _utf8_size(value)
return size is not None and size <= MAX_IDENTIFIER_BYTES and not any(
ord(character) < 32 or ord(character) == 127 for character in value
)
def _validate_metadata(metadata: Any) -> Optional[str]:
if not isinstance(metadata, dict):
return "invalid_input"
nodes = 0
stack: List[Tuple[Any, int]] = [(metadata, 0)]
while stack:
value, depth = stack.pop()
nodes += 1
if depth > MAX_METADATA_DEPTH or nodes > MAX_METADATA_NODES:
return "input_too_large"
if isinstance(value, dict):
if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
return "input_too_large"
for key, child in value.items():
if (
not isinstance(key, str)
or len(key) > MAX_METADATA_BYTES
or _utf8_size(key) is None
):
return "invalid_input"
stack.append((child, depth + 1))
elif isinstance(value, list):
if len(value) + nodes + len(stack) > MAX_METADATA_NODES:
return "input_too_large"
stack.extend((child, depth + 1) for child in value)
elif isinstance(value, str):
if len(value) > MAX_METADATA_BYTES:
return "input_too_large"
if _utf8_size(value) is None:
return "invalid_input"
elif value is not None and not isinstance(value, (bool, int, float)):
return "invalid_input"
try:
encoded = json.dumps(
metadata,
ensure_ascii=False,
allow_nan=False,
separators=(",", ":"),
).encode("utf-8")
except (RecursionError, TypeError, UnicodeEncodeError, ValueError):
return "invalid_input"
return "input_too_large" if len(encoded) > MAX_METADATA_BYTES else None
def _validate_payload(
target: Any,
content: Any,
account_id: Any,
message_id: Any,
metadata: Any,
dry_run: Any,
) -> Optional[str]:
if not isinstance(target, str) or not isinstance(content, str):
return "invalid_input"
if not target.strip() or not content.strip() or not isinstance(dry_run, bool):
return "invalid_input"
if any(ord(character) < 32 or ord(character) == 127 for character in target):
return "invalid_input"
if "\x00" in content or not _valid_identifier(account_id) or not _valid_identifier(message_id):
return "invalid_input"
if len(target) > MAX_TARGET_BYTES or len(content) > MAX_CONTENT_BYTES:
return "input_too_large"
target_size = _utf8_size(target)
content_size = _utf8_size(content)
if target_size is None or content_size is None:
return "invalid_input"
if target_size > MAX_TARGET_BYTES or content_size > MAX_CONTENT_BYTES:
return "input_too_large"
return _validate_metadata(metadata)
def _safe_provider_message_id(value: Any) -> Optional[str]:
if not isinstance(value, str) or not value:
return None
if len(value) > MAX_IDENTIFIER_BYTES:
return None
size = _utf8_size(value)
if size is None or size > MAX_IDENTIFIER_BYTES:
return None
if any(ord(character) < 32 or ord(character) == 127 for character in value):
return None
return value
def send_ucc_adapter(
target: str,
content: str,
account_id: str = "",
message_id: str = "",
metadata: Optional[Dict[str, Any]] = None,
dry_run: bool = False,
) -> BackendResult:
effective_account = account_id or WECHAT_ACCOUNT_ID
effective_metadata: Any = {} if metadata is None else metadata
error_code = _validate_payload(
target,
content,
effective_account,
message_id,
effective_metadata,
dry_run,
)
if error_code:
return BackendResult(False, "ucc-adapter", "failed", {"code": error_code})
payload = {
"target": target,
"content": content,
"account_id": effective_account,
"message_id": message_id,
"metadata": effective_metadata,
"dry_run": dry_run,
}
last_endpoint: Optional[str] = None
for path in _ucc_send_paths():
last_endpoint = path
try:
http_status, data = _post_ucc(path, payload)
except urllib.error.HTTPError as exc:
if exc.code in {404, 405}:
continue
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_http_error", "http_status": int(exc.code), "endpoint": path},
)
except (OSError, ValueError, urllib.error.URLError, _BackendProtocolError):
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_unavailable", "endpoint": path},
)
except Exception:
# Fail closed without returning exception text, type, or chained cause.
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_unavailable", "endpoint": path},
)
if not (200 <= http_status < 300):
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_http_error", "http_status": http_status, "endpoint": path},
)
if not _ucc_result_is_ok(data, dry_run):
return BackendResult(
False,
"ucc-adapter",
"failed",
{"code": "ucc_adapter_rejected", "http_status": http_status, "endpoint": path},
)
confirmed_status = "dry_run" if dry_run else "sent"
result_data: Dict[str, Any] = {
"code": confirmed_status,
"http_status": http_status,
"endpoint": path,
}
provider_message_id = _safe_provider_message_id(data.get("provider_message_id"))
if provider_message_id:
result_data["provider_message_id"] = provider_message_id
return BackendResult(True, "ucc-adapter", confirmed_status, result_data)
data: Dict[str, Any] = {"code": "ucc_adapter_unavailable"}
if last_endpoint:
data["endpoint"] = last_endpoint
return BackendResult(False, "ucc-adapter", "failed", data)
def send_openclaw_channel(
target: str,
content: str,
account_id: str = "",
message_id: str = "",
metadata: Optional[Dict[str, Any]] = None,
dry_run: bool = False,
) -> BackendResult:
del target, content, account_id, message_id, metadata, dry_run
return BackendResult(
False,
"openclaw-channel",
"failed",
{"code": "unsupported_backend"},
)
def send_message(
target: str,
content: str,
account_id: str = "",
message_id: str = "",
metadata: Optional[Dict[str, Any]] = None,
dry_run: bool = False,
) -> Dict[str, Any]:
selected = select_backend()
if selected.get("backend") == "ucc-adapter":
return send_ucc_adapter(
target, content, account_id, message_id, metadata, dry_run
).to_dict()
code = "unsupported_backend" if selected.get("reason") == "unsupported_backend" else "wechat_outbound_backend_unavailable"
return BackendResult(False, "unavailable", "failed", {"code": code}).to_dict()
if __name__ == "__main__":
print(json.dumps(select_backend(), ensure_ascii=False, separators=(",", ":")))
+2 -2
View File
@@ -1,5 +1,5 @@
from active_runtime.app.agent_event_bus import AgentEventBus
from active_runtime.app.agent_event_dashboard import AgentEventDashboard
from app.agent_event_bus import AgentEventBus
from app.agent_event_dashboard import AgentEventDashboard
def test_unified_event_flow():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.agent_observability_hooks import AgentObservabilityHooks
from app.agent_observability_hooks import AgentObservabilityHooks
def test_agent_observability_closed_loop():
+2 -2
View File
@@ -1,5 +1,5 @@
from active_runtime.app.architect_governance_agent import ArchitectGovernanceAgent
from active_runtime.app.architect_governance_dashboard import ArchitectGovernanceDashboard
from app.architect_governance_agent import ArchitectGovernanceAgent
from app.architect_governance_dashboard import ArchitectGovernanceDashboard
def test_architect_agent_decides_release_level():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.architect_research_timeline import ArchitectResearchTimeline
from app.architect_research_timeline import ArchitectResearchTimeline
def test_research_timeline_tracks_version_decision():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.architect_weekly_research_agent import ArchitectWeeklyResearchAgent
from app.architect_weekly_research_agent import ArchitectWeeklyResearchAgent
def test_architect_research_decides_release_type():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.architecture_change_classifier import ArchitectureChangeClassifier
from app.architecture_change_classifier import ArchitectureChangeClassifier
def test_small_change_can_auto_process():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.architecture_owner_gate import ArchitectureOwnerGate
from app.architecture_owner_gate import ArchitectureOwnerGate
def test_architecture_change_blocks_without_owner():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.change_governance_dashboard import ChangeGovernanceDashboard
from app.change_governance_dashboard import ChangeGovernanceDashboard
def test_change_governance_blocks_missing_evidence():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.governance_event_bridge import GovernanceEventBridge
from app.governance_event_bridge import GovernanceEventBridge
def test_governance_events_are_published():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.governance_timeline_dashboard import GovernanceTimelineDashboard
from app.governance_timeline_dashboard import GovernanceTimelineDashboard
def test_governance_timeline_detects_ready_flow():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.search_agent_adapter import AgentSearchContext, OpenClawSearchClient
from app.search_agent_adapter import AgentSearchContext, OpenClawSearchClient
class FakeOrchestrator:
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.search_architect_integration import SearchArchitectIntegration
from app.search_architect_integration import SearchArchitectIntegration
def test_search_architect_recommends_release_for_new_capability():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.search_gateway import (
from app.search_gateway import (
SearchEvidence,
SearchGateway,
SearchProviderAttempt,
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.search_gateway_dashboard import SearchGatewayDashboard
from app.search_gateway_dashboard import SearchGatewayDashboard
def test_search_gateway_dashboard_reports_provider_failures():
@@ -1,5 +1,5 @@
from active_runtime.app.search_gateway import SearchEvidence, SearchGateway, SearchProviderAttempt, SearchRequest
from active_runtime.app.search_profile_resolver import SearchProfileResolver
from app.search_gateway import SearchEvidence, SearchGateway, SearchProviderAttempt, SearchRequest
from app.search_profile_resolver import SearchProfileResolver
class FailingProvider:
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.search_gateway_orchestrator import SearchGatewayOrchestrator
from app.search_gateway_orchestrator import SearchGatewayOrchestrator
def test_search_orchestrator_has_audit_registry_event_layers():
+2 -2
View File
@@ -1,5 +1,5 @@
from active_runtime.app.search_provider_registry import SearchProviderRegistry, StaticArchiveProvider
from active_runtime.app.search_gateway import SearchGateway, SearchRequest
from app.search_provider_registry import SearchProviderRegistry, StaticArchiveProvider
from app.search_gateway import SearchGateway, SearchRequest
def test_provider_registry_has_default_layers():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.task_center import AIOSTaskRecord, TaskCenter, TaskNotificationStatus, TaskProgress
from app.task_center import AIOSTaskRecord, TaskCenter, TaskNotificationStatus, TaskProgress
def test_task_center_tracks_progress_and_notification():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.task_dashboard_view import TaskDashboardView
from app.task_dashboard_view import TaskDashboardView
def test_task_dashboard_contains_operations_views():
+2 -2
View File
@@ -1,5 +1,5 @@
from active_runtime.app.task_event_bridge import TaskEventBridge
from active_runtime.app.task_statistics import TaskStatistics
from app.task_event_bridge import TaskEventBridge
from app.task_statistics import TaskStatistics
def test_task_event_bridge_creates_events():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.task_lifecycle import TaskLifecycle
from app.task_lifecycle import TaskLifecycle
def test_owner_approval_button_appears_for_waiting_task():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.task_orchestrator import TaskOrchestrator
from app.task_orchestrator import TaskOrchestrator
def test_command_creates_task():
+1 -1
View File
@@ -1,4 +1,4 @@
from active_runtime.app.git_repo_locator import GitRepoLocator
from app.git_repo_locator import GitRepoLocator
def test_repo_locator_uses_configurable_root():
+65
View File
@@ -0,0 +1,65 @@
from __future__ import annotations
import importlib.util
from dataclasses import dataclass
from pathlib import Path
import pytest
import aios_v2r1
from aios_v2r1 import cli
from aios_v2r1.ingress_service import IngressHTTPServer
ROOT = Path(__file__).resolve().parents[1]
def test_canonical_package_has_one_identity_and_no_compatibility_shim() -> None:
assert aios_v2r1.__name__ == "aios_v2r1"
assert aios_v2r1.__package__ == "aios_v2r1"
assert aios_v2r1.__version__ == "2.1.0-rc.1"
assert importlib.util.find_spec("active_runtime") is None
assert not (ROOT / "active_runtime").exists()
def test_packaging_metadata_declares_only_the_canonical_app_package() -> None:
metadata = (ROOT / "active-runtime" / "pyproject.toml").read_text()
assert 'packages = ["aios_v2r1"]' in metadata
assert 'package-dir = {"" = "app"}' in metadata
assert "active_runtime" not in metadata
assert 'aios-cell = "aios_v2r1.cli:main"' in metadata
def test_registry_path_named_serve_keeps_legacy_status_dispatch(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
calls: list[tuple[str, str]] = []
@dataclass
class Result:
tenant_id: str
state: str
class FakeManager:
def __init__(self, registry: str) -> None:
calls.append(("registry", registry))
def status(self, tenant_id: str) -> Result:
calls.append(("status", tenant_id))
return Result(tenant_id=tenant_id, state="healthy")
monkeypatch.setattr(cli, "CellManager", FakeManager)
monkeypatch.setattr(
cli,
"serve",
lambda host, port: pytest.fail("legacy registry command entered service mode"),
)
assert cli.main(["serve", "status", "tenant-a"]) == 0
assert calls == [("registry", "serve"), ("status", "tenant-a")]
assert '"tenant_id": "tenant-a"' in capsys.readouterr().out
def test_packaged_server_waits_for_non_daemon_request_handlers() -> None:
assert IngressHTTPServer.daemon_threads is False
assert IngressHTTPServer.block_on_close is True
+454 -1
View File
@@ -1,11 +1,20 @@
from __future__ import annotations
import importlib.util
import io
import os
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
from types import SimpleNamespace
from unittest import TestCase
from unittest.mock import patch
from unittest.mock import Mock, call, patch
BACKEND_PATH = Path(__file__).resolve().parents[1] / "scripts" / "wechat_send_backend.py"
@@ -80,6 +89,450 @@ class WechatSendBackendTests(TestCase):
self.assertEqual(args[1]["content"], "hello")
self.assertFalse(args[1]["dry_run"])
def test_ucc_auth_is_authorization_header_not_argv_or_output(self):
secret = "ucc-super-secret-token"
class Response:
status = 200
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def read(self, _limit=-1):
return b'{"ok":true,"status":"sent"}'
urlopen = Mock(return_value=Response())
output = io.StringIO()
with (
patch.object(self.backend, "UCC_ADAPTER_TOKEN", secret),
patch.object(self.backend, "_open_no_redirect", urlopen),
redirect_stdout(output),
redirect_stderr(output),
):
status, payload = self.backend._post_ucc(
"/api/internal/wechat/send", {"target": "chat-001", "content": "hello"}
)
request = urlopen.call_args.args[0]
self.assertEqual(request.get_header("Authorization"), f"Bearer {secret}")
self.assertEqual(status, 200)
self.assertEqual(payload, {"ok": True, "status": "sent"})
self.assertNotIn(secret, " ".join(self.backend._openclaw_base_args()))
self.assertNotIn(secret, output.getvalue())
self.assertFalse({"--gateway", "--token", "--password"} & set(self.backend._openclaw_base_args()))
def test_ucc_paths_are_canonical_first_unique_and_bounded(self):
with patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"):
paths = self.backend._ucc_send_paths()
self.assertEqual(paths[0], "/api/internal/wechat/send")
self.assertLessEqual(len(paths), 2)
self.assertEqual(len(paths), len(set(paths)))
self.assertTrue(all(path.startswith("/") for path in paths))
def test_ucc_accepts_only_explicit_true_and_sent_when_not_dry_run(self):
self.assertTrue(self.backend._ucc_response_sent({"ok": True, "status": "sent"}))
for response in (
{"ok": 1, "status": "sent"},
{"ok": True, "status": "accepted"},
{"ok": False, "status": "sent"},
{"status": "sent"},
):
with self.subTest(response=response):
self.assertFalse(self.backend._ucc_response_sent(response))
def test_ambiguous_canonical_failures_never_fallback(self):
secret = "raw-upstream-secret"
failures = (
TimeoutError(secret),
ConnectionError(secret),
self.backend._BackendProtocolError(secret),
RuntimeError(secret),
)
for failure in failures:
with self.subTest(failure=type(failure).__name__):
with (
patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
patch.object(self.backend, "_post_ucc", side_effect=failure) as post,
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", message_id="msg-001"
).to_dict()
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "ucc_adapter_unavailable")
self.assertEqual(post.call_count, 1)
self.assertNotIn(secret, repr(result))
def test_only_explicit_http_404_or_405_uses_legacy_fallback(self):
for http_status in (404, 405):
with self.subTest(http_status=http_status):
missing = urllib.error.HTTPError(
"http://127.0.0.1/api/internal/wechat/send",
http_status,
"not found",
{},
None,
)
with (
patch.object(self.backend, "UCC_SEND_PATH", "/api/internal/wechat/send"),
patch.object(
self.backend,
"_post_ucc",
side_effect=[missing, (200, {"ok": True, "status": "sent"})],
) as post,
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", message_id="msg-001"
).to_dict()
self.assertTrue(result["ok"])
self.assertEqual(result["endpoint"], "/internal/channels/wechat/send")
self.assertEqual(post.call_count, 2)
def test_non_not_found_http_error_never_falls_back(self):
denied = urllib.error.HTTPError(
"http://127.0.0.1/api/internal/wechat/send", 401, "denied", {}, None
)
with patch.object(self.backend, "_post_ucc", side_effect=denied) as post:
result = self.backend.send_ucc_adapter(target="chat-001", content="hello").to_dict()
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "ucc_adapter_http_error")
post.assert_called_once()
def test_ucc_rejection_does_not_expose_raw_error_or_secret(self):
secret = "provider-password=secret-value"
with patch.object(
self.backend,
"_post_ucc",
return_value=(200, {"ok": False, "status": "failed", "error": secret}),
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", message_id="msg-001"
).to_dict()
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "ucc_adapter_rejected")
self.assertNotIn(secret, repr(result))
def test_dry_run_requires_dry_run_status_and_never_reports_sent(self):
with patch.object(
self.backend,
"_post_ucc",
return_value=(200, {"ok": True, "status": "dry_run"}),
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", dry_run=True
).to_dict()
self.assertTrue(result["ok"])
self.assertEqual(result["status"], "dry_run")
self.assertNotEqual(result["status"], "sent")
for dry_run, upstream_status in ((True, "sent"), (False, "dry_run")):
with self.subTest(dry_run=dry_run, upstream_status=upstream_status):
with patch.object(
self.backend,
"_post_ucc",
return_value=(200, {"ok": True, "status": upstream_status}),
):
rejected = self.backend.send_ucc_adapter(
target="chat-001", content="hello", dry_run=dry_run
).to_dict()
self.assertFalse(rejected["ok"])
self.assertEqual(rejected["status"], "failed")
def test_invalid_provider_message_id_never_reverses_confirmed_send(self):
invalid_values = (
"\ud800",
"contains\x00control",
"x" * (self.backend.MAX_IDENTIFIER_BYTES + 1),
{"not": "a string"},
)
for provider_message_id in invalid_values:
with self.subTest(value=repr(provider_message_id)[:40]):
with patch.object(
self.backend,
"_post_ucc",
return_value=(
200,
{
"ok": True,
"status": "sent",
"provider_message_id": provider_message_id,
},
),
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello"
).to_dict()
self.assertTrue(result["ok"])
self.assertEqual(result["status"], "sent")
self.assertNotIn("provider_message_id", result)
def test_openclaw_2026_6_11_commands_have_no_unsupported_global_gateway(self):
completed = SimpleNamespace(returncode=0, stdout='{"channels": []}')
with (
patch.object(self.backend, "OPENCLAW_BIN", "/usr/local/bin/openclaw"),
patch.object(self.backend, "_run_openclaw", return_value=completed) as run,
):
self.assertIsNone(self.backend.discover_openclaw_wechat_channel())
command = run.call_args.args[0]
self.assertEqual(command, ["/usr/local/bin/openclaw", "channels", "list", "--json"])
self.assertFalse({"--gateway", "--token", "--password"} & set(command))
def test_openclaw_output_overflow_terminates_while_process_is_running(self):
writer = (
"import os,time;"
f"os.write(1,b'x'*{self.backend.MAX_RESPONSE_BYTES + 4096});"
"time.sleep(10)"
)
started = time.monotonic()
completed = self.backend._run_openclaw([sys.executable, "-c", writer], timeout=4)
elapsed = time.monotonic() - started
self.assertLess(elapsed, 2.0)
self.assertNotEqual(completed.returncode, 0)
self.assertEqual(completed.stdout, "")
self.assertEqual(completed.stderr, "")
def test_openclaw_timeout_returns_failure_and_terminates_promptly(self):
started = time.monotonic()
completed = self.backend._run_openclaw(
[sys.executable, "-c", "import time;time.sleep(10)"], timeout=1
)
elapsed = time.monotonic() - started
self.assertLess(elapsed, 3.0)
self.assertNotEqual(completed.returncode, 0)
self.assertEqual(completed.stdout, "")
self.assertEqual(completed.stderr, "")
def test_timeout_kills_inherited_pipe_descendant_after_parent_exits(self):
if os.name != "posix" or not Path("/proc").is_dir():
self.skipTest("POSIX process-group and procfs contract")
def process_is_running(pid):
try:
state = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[2]
except (FileNotFoundError, IndexError, OSError):
return False
return state not in {"X", "Z"}
with tempfile.TemporaryDirectory() as directory:
pid_path = Path(directory) / "descendant.pid"
descendant = "import time;time.sleep(30)"
launcher = (
"import pathlib,subprocess,sys;"
f"child=subprocess.Popen([sys.executable,'-c',{descendant!r}]);"
f"pathlib.Path({str(pid_path)!r}).write_text(str(child.pid),encoding='utf-8')"
)
started = time.monotonic()
completed = self.backend._run_openclaw(
[sys.executable, "-c", launcher], timeout=1
)
elapsed = time.monotonic() - started
self.assertTrue(pid_path.is_file())
descendant_pid = int(pid_path.read_text(encoding="utf-8"))
try:
deadline = time.monotonic() + 2
while process_is_running(descendant_pid) and time.monotonic() < deadline:
time.sleep(0.02)
self.assertFalse(process_is_running(descendant_pid))
finally:
if process_is_running(descendant_pid):
os.kill(descendant_pid, signal.SIGKILL)
self.assertLess(elapsed, 3.0)
self.assertNotEqual(completed.returncode, 0)
def test_process_tree_termination_uses_posix_process_group(self):
if os.name != "posix":
self.skipTest("POSIX process-group contract")
class Process:
pid = 43210
def __init__(self):
self.wait_count = 0
def poll(self):
return None
def wait(self, timeout):
self.wait_count += 1
if self.wait_count == 1:
raise subprocess.TimeoutExpired(["openclaw"], timeout)
return -signal.SIGKILL
process = Process()
with patch.object(self.backend.os, "killpg") as killpg:
self.backend._terminate_process_tree(process)
self.assertEqual(
killpg.call_args_list,
[
call(process.pid, signal.SIGTERM),
call(process.pid, signal.SIGKILL),
],
)
def test_unsupported_and_generic_chat_backends_fail_closed(self):
with (
patch.object(self.backend, "SEND_BACKEND", "openclaw-chat"),
patch.object(self.backend, "ucc_adapter_alive", return_value=False),
):
selected = self.backend.select_backend()
result = self.backend.send_message(target="chat-001", content="hello")
self.assertEqual(selected["backend"], "unavailable")
self.assertEqual(selected["reason"], "unsupported_backend")
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "unsupported_backend")
def test_invalid_or_oversized_input_is_rejected_before_network(self):
with patch.object(self.backend, "_post_ucc") as post:
empty = self.backend.send_ucc_adapter(target="", content="hello").to_dict()
oversized = self.backend.send_ucc_adapter(
target="chat-001", content="x" * (self.backend.MAX_CONTENT_BYTES + 1)
).to_dict()
self.assertEqual(empty["code"], "invalid_input")
self.assertEqual(oversized["code"], "input_too_large")
post.assert_not_called()
def test_unpaired_surrogate_inputs_fail_closed_before_network(self):
cases = (
{"target": "\ud800", "content": "hello"},
{"target": "chat-001", "content": "\ud800"},
{"target": "chat-001", "content": "hello", "account_id": "\ud800"},
{"target": "chat-001", "content": "hello", "message_id": "\ud800"},
{"target": "chat-001", "content": "hello", "metadata": {"bad": "\ud800"}},
)
with patch.object(self.backend, "_post_ucc") as post:
for kwargs in cases:
with self.subTest(field=next(reversed(kwargs))):
result = self.backend.send_ucc_adapter(**kwargs).to_dict()
self.assertEqual(result["code"], "invalid_input")
post.assert_not_called()
def test_deeply_nested_metadata_is_bounded_before_network(self):
metadata = {}
for _ in range(1200):
metadata = {"nested": metadata}
with patch.object(self.backend, "_post_ucc") as post:
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello", metadata=metadata
).to_dict()
self.assertEqual(result["code"], "input_too_large")
post.assert_not_called()
def test_adapter_origin_must_be_allowlisted(self):
with (
patch.object(self.backend, "UCC_ADAPTER_URL", "https://evil.example:8443"),
patch.object(
self.backend,
"UCC_ADAPTER_ALLOWED_ORIGINS",
("http://127.0.0.1:18800",),
),
):
self.assertIsNone(self.backend._adapter_base_url())
with (
patch.object(self.backend, "UCC_ADAPTER_URL", "https://adapter.example:8443"),
patch.object(
self.backend,
"UCC_ADAPTER_ALLOWED_ORIGINS",
("https://adapter.example:8443",),
),
):
self.assertEqual(
self.backend._adapter_base_url(), "https://adapter.example:8443"
)
def test_adapter_origin_rejects_scheme_userinfo_fragment_and_path(self):
invalid_origins = (
"ftp://127.0.0.1:18800",
"http://user:password@127.0.0.1:18800",
"http://127.0.0.1:18800#fragment",
"http://127.0.0.1:18800/base-path",
)
for origin in invalid_origins:
with self.subTest(origin=origin):
with (
patch.object(self.backend, "UCC_ADAPTER_URL", origin),
patch.object(
self.backend,
"UCC_ADAPTER_ALLOWED_ORIGINS",
("http://127.0.0.1:18800",),
),
):
self.assertIsNone(self.backend._adapter_base_url())
def test_adapter_path_rejects_decoded_traversal(self):
for path in (
"/api/%2e%2e/send",
"/api/%2E%2E/send",
"/api/%2525252e%2525252e/send",
"/api/%252525252e%252525252e/send",
"/api/../send",
"/api/%zz/send",
):
with self.subTest(path=path):
self.assertIsNone(self.backend._safe_path(path))
def test_adapter_path_keeps_intended_safe_canonical_paths(self):
for path in (
"/api/internal/wechat/send",
"/internal/channels/wechat/send",
"/api/%7Ehealth/status",
):
with self.subTest(path=path):
self.assertEqual(self.backend._safe_path(path), path)
def test_redirects_are_refused_and_authorization_cannot_cross_origin(self):
handler = self.backend._NoRedirectHandler()
self.assertIsNone(
handler.redirect_request(
None,
None,
302,
"redirect",
{"Location": "https://evil.example/steal"},
"https://evil.example/steal",
)
)
redirected = urllib.error.HTTPError(
"http://127.0.0.1:18800/api/internal/wechat/send",
302,
"redirect",
{"Location": "https://evil.example/steal"},
None,
)
with (
patch.object(self.backend, "UCC_ADAPTER_TOKEN", "header-secret"),
patch.object(self.backend, "_open_no_redirect", side_effect=redirected) as opened,
):
result = self.backend.send_ucc_adapter(
target="chat-001", content="hello"
).to_dict()
request = opened.call_args.args[0]
self.assertEqual(urllib.parse.urlsplit(request.full_url).netloc, "127.0.0.1:18800")
self.assertEqual(request.get_header("Authorization"), "Bearer header-secret")
opened.assert_called_once()
self.assertFalse(result["ok"])
self.assertEqual(result["code"], "ucc_adapter_http_error")
if __name__ == "__main__":
import unittest
+101
View File
@@ -0,0 +1,101 @@
"""Executable completeness contract for the append-only audit chain."""
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import os
import sqlite3
import pytest
from aios_v2r1.audit import AuditLog
from aios_v2r1.tenant_context import ConflictError, ValidationError
def append(log: AuditLog, index: int = 0, *, tenant_id: str = "tenant-a") -> str:
return log.append(tenant_id=tenant_id, trace_id=f"trace-{index}",
event_type="tool_call",
payload={"index": index, "text": "虾马 café"})
def test_first_event_is_anchored_to_genesis(tmp_path: Path) -> None:
database = tmp_path / "audit.sqlite3"
log = AuditLog(database)
append(log)
with sqlite3.connect(database) as db:
assert db.execute("SELECT previous_hash FROM audit_events").fetchone()[0] == "GENESIS"
assert log.verify()
def test_unicode_payload_is_canonical_across_restart(tmp_path: Path) -> None:
database = tmp_path / "audit.sqlite3"
digest = append(AuditLog(database))
reopened = AuditLog(database)
assert reopened.verify()
with sqlite3.connect(database) as db:
stored = db.execute("SELECT event_hash,payload FROM audit_events").fetchone()
assert stored[0] == digest
assert "虾马" in stored[1]
def test_payload_tamper_is_detected(tmp_path: Path) -> None:
database = tmp_path / "audit.sqlite3"
log = AuditLog(database)
append(log)
with sqlite3.connect(database) as db:
db.execute("UPDATE audit_events SET payload='{}'")
assert not log.verify()
def test_middle_event_deletion_is_detected(tmp_path: Path) -> None:
database = tmp_path / "audit.sqlite3"
log = AuditLog(database)
for index in range(3):
append(log, index)
with sqlite3.connect(database) as db:
db.execute("DELETE FROM audit_events WHERE seq=2")
assert not log.verify()
def test_event_reordering_is_detected(tmp_path: Path) -> None:
database = tmp_path / "audit.sqlite3"
log = AuditLog(database)
append(log, 1)
append(log, 2)
with sqlite3.connect(database) as db:
db.execute("UPDATE audit_events SET seq=100 WHERE seq=1")
assert not log.verify()
def test_concurrent_appends_form_one_valid_chain(tmp_path: Path) -> None:
database = tmp_path / "audit.sqlite3"
log = AuditLog(database)
with ThreadPoolExecutor(max_workers=16) as executor:
digests = list(executor.map(lambda index: append(log, index), range(24)))
assert len(set(digests)) == 24
assert log.verify()
with sqlite3.connect(database) as db:
assert db.execute("SELECT count(*) FROM audit_events").fetchone()[0] == 24
@pytest.mark.parametrize("kwargs", [
{"tenant_id": "Tenant-A"}, {"trace_id": "../trace"},
{"event_type": "tool/call"}])
def test_audit_identifiers_are_validated(tmp_path: Path, kwargs: dict[str, str]) -> None:
values = {"tenant_id": "tenant-a", "trace_id": "trace-a",
"event_type": "tool_call", "payload": {"ok": True}}
values.update(kwargs)
with pytest.raises(ValidationError):
AuditLog(tmp_path / "audit.sqlite3").append(**values)
def test_tenants_share_no_hash_identity_for_equal_payloads(tmp_path: Path) -> None:
log = AuditLog(tmp_path / "audit.sqlite3")
first = append(log, tenant_id="tenant-a")
second = append(log, tenant_id="tenant-b")
assert first != second
assert log.verify()
def test_database_is_owner_only(tmp_path: Path) -> None:
database = tmp_path / "audit.sqlite3"
AuditLog(database)
assert os.stat(database).st_mode & 0o777 == 0o600
def test_sqlite_failures_are_sanitized(tmp_path: Path) -> None:
database = tmp_path / "audit.sqlite3"
log = AuditLog(database)
database.write_bytes(b"not-a-sqlite-database")
with pytest.raises(ConflictError, match="audit storage unavailable") as raised:
log.verify()
assert not isinstance(raised.value, sqlite3.Error)
assert "sqlite" not in str(raised.value).lower()
assert raised.value.__cause__ is None
assert raised.value.__context__ is None
+114
View File
@@ -0,0 +1,114 @@
"""Executable completeness contract for CellManager and its CLI."""
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
import json
import os
from pathlib import Path
from unittest.mock import patch
import pytest
from aios_v2r1.cell_manager import CellManager
from aios_v2r1.cli import main
from aios_v2r1.tenant_context import ValidationError
def test_one_cell_per_tenant(tmp_path: Path) -> None:
manager = CellManager(tmp_path / "cells.json")
manager.create("tenant-a", "cell-a", "v1", "digest-a")
with pytest.raises(ValidationError, match="already"):
manager.create("tenant-a", "cell-b", "v1", "digest-b")
def test_explicit_state_graph_accepts_the_recovery_cycle(tmp_path: Path) -> None:
manager = CellManager(tmp_path / "cells.json")
manager.create("tenant-a", "cell-a", "v1", "digest-a")
for state in ("starting", "healthy", "upgrading", "rollback", "healthy"):
assert manager.transition("tenant-a", state).state == state
def test_explicit_state_graph_rejects_jump_from_created_to_healthy(tmp_path: Path) -> None:
manager = CellManager(tmp_path / "cells.json")
manager.create("tenant-a", "cell-a", "v1", "digest-a")
with pytest.raises(ValidationError, match="transition"):
manager.transition("tenant-a", "healthy")
def test_persistence_uses_atomic_replace(tmp_path: Path) -> None:
manager = CellManager(tmp_path / "cells.json")
with patch("aios_v2r1.cell_manager.os.replace", wraps=os.replace) as replaced:
manager.create("tenant-a", "cell-a", "v1", "digest-a")
assert replaced.called
def test_concurrent_distinct_tenant_creates_are_not_lost(tmp_path: Path) -> None:
registry = tmp_path / "cells.json"
manager = CellManager(registry)
def create(index: int) -> None:
manager.create(f"tenant-{index}", f"cell-{index}", "v1", f"digest-{index}")
with ThreadPoolExecutor(max_workers=16) as executor:
list(executor.map(create, range(24)))
data = json.loads(registry.read_text(encoding="utf-8"))
assert len(data["cells"]) == 24
def test_concurrent_same_tenant_has_exactly_one_winner(tmp_path: Path) -> None:
manager = CellManager(tmp_path / "cells.json")
def create(index: int) -> str:
try:
manager.create("tenant-a", f"cell-{index}", "v1", f"digest-{index}")
return "created"
except ValidationError:
return "rejected"
with ThreadPoolExecutor(max_workers=16) as executor:
results = list(executor.map(create, range(16)))
assert results.count("created") == 1
assert results.count("rejected") == 15
def test_corrupt_registry_fails_closed_with_domain_error(tmp_path: Path) -> None:
registry = tmp_path / "cells.json"
manager = CellManager(registry)
registry.write_text('{"cells":{"tenant-a":', encoding="utf-8")
with pytest.raises(ValidationError, match="registry unavailable") as raised:
manager.status("tenant-a")
assert "JSON" not in type(raised.value).__name__
assert raised.value.__cause__ is None
assert raised.value.__context__ is None
def test_restart_preserves_cell_status(tmp_path: Path) -> None:
registry = tmp_path / "cells.json"
CellManager(registry).create("tenant-a", "cell-a", "v1", "digest-a")
assert CellManager(registry).status("tenant-a").cell_id == "cell-a"
def test_cli_create_returns_stable_json(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
registry = tmp_path / "cells.json"
assert main([str(registry), "create", "tenant-a", "cell-a", "v1", "digest-a"]) == 0
output = json.loads(capsys.readouterr().out)
assert output["tenant_id"] == "tenant-a"
assert output["state"] == "created"
def test_cli_domain_failure_returns_code_3_and_json(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
registry = tmp_path / "cells.json"
assert main([str(registry), "status", "missing-tenant"]) == 3
captured = capsys.readouterr()
assert captured.out == ""
assert json.loads(captured.err) == {
"code": "VALIDATION_ERROR", "error": "unknown tenant"}
def test_cli_corrupt_registry_is_sanitized(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
registry = tmp_path / "cells.json"
CellManager(registry)
registry.write_text("TOP-SECRET invalid json", encoding="utf-8")
assert main([str(registry), "status", "tenant-a"]) == 3
error = json.loads(capsys.readouterr().err)
assert error["code"] == "VALIDATION_ERROR"
assert "TOP-SECRET" not in error["error"]
@pytest.mark.parametrize("args", [
("Tenant-A", "cell-a", "v1", "digest-a"),
("tenant-a", "../cell", "v1", "digest-a"),
("tenant-a", "cell-a", "../v1", "digest-a"),
("tenant-a", "cell-a", "v1", "../digest")])
def test_create_validates_all_identifiers(
tmp_path: Path, args: tuple[str, str, str, str]
) -> None:
with pytest.raises(ValidationError):
CellManager(tmp_path / "cells.json").create(*args)
+98 -72
View File
@@ -1,88 +1,114 @@
"""Gate 1 contract tests for the Hermes shared agent fabric."""
"""Executable completeness contract for the tenant-partitioned Hermes fabric."""
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from pathlib import Path
import sqlite3
from tempfile import TemporaryDirectory
import unittest
import pytest
from aios_v2r1.hermes_fabric import HermesFabric
from aios_v2r1.tenant_context import CanonicalMessage, ConflictError, ScopeError
def authenticated_command(
event_id: str,
*,
command_capable: bool = True,
origin_type: str = "channel_user",
) -> CanonicalMessage:
return CanonicalMessage(
tenant_id="tenant-a",
cell_id="cell-a",
principal_id="user-a",
principal_type="authenticated_user",
origin_type=origin_type,
channel="weixin",
channel_account_id="account-a",
peer_id="peer-a",
conversation_id="conversation-a",
session_id="session-a",
task_id=None,
event_id=event_id,
message_id="1234567890123456789",
reply_to_message_id=None,
direction="inbound",
command_capable=command_capable,
def authenticated_command(event_id: str, **changes: object) -> CanonicalMessage:
value = CanonicalMessage(
tenant_id="tenant-a", cell_id="cell-a", principal_id="user-a",
principal_type="authenticated_user", origin_type="channel_user",
channel="weixin", channel_account_id="account-a", peer_id="peer-a",
conversation_id="conversation-a", session_id="session-a", task_id=None,
event_id=event_id, message_id=f"message-{event_id}",
reply_to_message_id=None, direction="inbound", command_capable=True,
occurred_at="2026-07-13T00:00:00+00:00",
received_at="2026-07-13T00:00:01+00:00",
body="/run",
quoted_body=None,
adapter_name="weixin",
adapter_version="v2",
)
received_at="2026-07-13T00:00:01+00:00", body="/run",
quoted_body=None, adapter_name="weixin", adapter_version="v2")
return replace(value, **changes)
@pytest.fixture
def fabric(tmp_path: Path) -> HermesFabric:
return HermesFabric(tmp_path / "fabric.sqlite3")
class HermesFabricContractTests(unittest.TestCase):
def test_duplicate_event_returns_sanitized_conflict(self) -> None:
with TemporaryDirectory() as directory:
fabric = HermesFabric(f"{directory}/fabric.sqlite3")
message = authenticated_command("event-a")
def test_duplicate_event_returns_sanitized_conflict(fabric: HermesFabric) -> None:
command = authenticated_command("event-a")
assert fabric.enqueue(message=command, profile_id="profile-a").event_id == "event-a"
with pytest.raises(ConflictError) as raised:
fabric.enqueue(message=command, profile_id="profile-a")
assert not isinstance(raised.value, sqlite3.IntegrityError)
assert "sqlite" not in str(raised.value).lower()
assert raised.value.__cause__ is None
assert raised.value.__context__ is None
first = fabric.enqueue(message=message, profile_id="profile-a")
self.assertEqual(first.event_id, "event-a")
@pytest.mark.parametrize("changes", [
{"command_capable": False}, {"origin_type": "heartbeat"},
{"principal_type": "service"}, {"direction": "internal"}])
def test_non_claimable_provenance_cannot_enqueue(
fabric: HermesFabric, changes: dict[str, object]
) -> None:
with pytest.raises(ScopeError):
fabric.enqueue(message=authenticated_command("event-denied", **changes),
profile_id="profile-a")
with self.assertRaises(ConflictError) as raised:
fabric.enqueue(message=message, profile_id="profile-a")
def test_claim_returns_running_and_matches_durable_state(fabric: HermesFabric) -> None:
fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
claimed = fabric.claim("tenant-a", "cell-a", "profile-a")
assert claimed is not None
assert claimed.status == "running"
with sqlite3.connect(fabric.database) as db:
assert db.execute("SELECT status FROM fabric_jobs WHERE job_id=?",
(claimed.job_id,)).fetchone()[0] == "running"
error = raised.exception
self.assertNotIsInstance(error, sqlite3.IntegrityError)
self.assertNotIn("sqlite", str(error).lower())
self.assertIsNone(error.__cause__)
self.assertIsNone(error.__context__)
@pytest.mark.parametrize(("tenant_id", "cell_id", "profile_id"), [
("tenant-b", "cell-a", "profile-a"), ("tenant-a", "cell-b", "profile-a"),
("tenant-a", "cell-a", "profile-b")])
def test_claim_scope_is_exact(
fabric: HermesFabric, tenant_id: str, cell_id: str, profile_id: str
) -> None:
fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
assert fabric.claim(tenant_id, cell_id, profile_id) is None
assert fabric.claim("tenant-a", "cell-a", "profile-a") is not None
def test_non_command_event_cannot_enqueue(self) -> None:
with TemporaryDirectory() as directory:
fabric = HermesFabric(f"{directory}/fabric.sqlite3")
def test_only_one_concurrent_claimant_wins(fabric: HermesFabric) -> None:
fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
with ThreadPoolExecutor(max_workers=16) as executor:
claimed = list(executor.map(
lambda _: fabric.claim("tenant-a", "cell-a", "profile-a"), range(16)))
winners = [job for job in claimed if job is not None]
assert len(winners) == 1
assert winners[0].status == "running"
with self.assertRaises(ScopeError):
fabric.enqueue(
message=authenticated_command(
"event-non-command", command_capable=False
),
profile_id="profile-a",
)
@pytest.mark.parametrize("status", ["running", "completed", "failed", "cancelled"])
def test_nonqueued_states_are_not_claimable(
fabric: HermesFabric, status: str
) -> None:
job = fabric.enqueue(message=authenticated_command("event-a"), profile_id="profile-a")
with sqlite3.connect(fabric.database) as db:
db.execute("UPDATE fabric_jobs SET status=? WHERE job_id=?", (status, job.job_id))
assert fabric.claim("tenant-a", "cell-a", "profile-a") is None
def test_heartbeat_provenance_cannot_enqueue(self) -> None:
with TemporaryDirectory() as directory:
fabric = HermesFabric(f"{directory}/fabric.sqlite3")
def test_restart_preserves_queue_and_claims_once(tmp_path: Path) -> None:
database = tmp_path / "fabric.sqlite3"
HermesFabric(database).enqueue(message=authenticated_command("event-a"),
profile_id="profile-a")
reopened = HermesFabric(database)
claimed = reopened.claim("tenant-a", "cell-a", "profile-a")
assert claimed is not None and claimed.status == "running"
assert reopened.claim("tenant-a", "cell-a", "profile-a") is None
with self.assertRaises(ScopeError):
fabric.enqueue(
message=authenticated_command(
"event-heartbeat", origin_type="heartbeat"
),
profile_id="profile-a",
)
def test_concurrent_duplicate_enqueue_has_one_winner(fabric: HermesFabric) -> None:
command = authenticated_command("event-a")
def enqueue(_: int) -> str:
try:
fabric.enqueue(message=command, profile_id="profile-a")
return "queued"
except ConflictError:
return "conflict"
with ThreadPoolExecutor(max_workers=16) as executor:
results = list(executor.map(enqueue, range(16)))
assert results.count("queued") == 1
assert results.count("conflict") == 15
if __name__ == "__main__":
unittest.main()
def test_sqlite_failures_are_sanitized(fabric: HermesFabric) -> None:
Path(fabric.database).write_bytes(b"not-a-sqlite-database")
with pytest.raises(ConflictError, match="fabric storage unavailable") as raised:
fabric.health()
assert not isinstance(raised.value, sqlite3.Error)
assert "sqlite" not in str(raised.value).lower()
assert raised.value.__cause__ is None
assert raised.value.__context__ is None
+88
View File
@@ -0,0 +1,88 @@
"""Executable completeness contract for Hermes profile isolation."""
from __future__ import annotations
from pathlib import Path
import os
import pytest
from aios_v2r1.hermes_profile import HermesProfile
from aios_v2r1.tenant_context import ValidationError
def profile(tmp_path: Path, **changes: object) -> HermesProfile:
values: dict[str, object] = {
"tenant_id": "tenant-a", "cell_id": "cell-a", "profile_id": "profile-a",
"hermes_home": tmp_path / "home", "workspace": tmp_path / "workspace"}
values.update(changes)
return HermesProfile(**values) # type: ignore[arg-type]
def test_environment_is_exact_and_has_no_inherited_values(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("HOST_SECRET", "must-not-inherit")
value = profile(tmp_path).environment
assert value == {
"HERMES_HOME": str(tmp_path / "home"), "HOME": str(tmp_path / "home"),
"AIOS_TENANT_ID": "tenant-a", "AIOS_CELL_ID": "cell-a",
"AIOS_PROFILE_ID": "profile-a"}
assert "HOST_SECRET" not in value
@pytest.mark.parametrize("changes", [
{"hermes_home": Path("relative-home")},
{"workspace": Path("relative-workspace")}])
def test_profile_roots_must_be_absolute(
tmp_path: Path, changes: dict[str, Path]
) -> None:
with pytest.raises(ValidationError, match="absolute"):
profile(tmp_path, **changes)
def test_prepare_sets_both_roots_to_0700(tmp_path: Path) -> None:
value = profile(tmp_path)
value.prepare()
assert os.stat(value.hermes_home).st_mode & 0o777 == 0o700
assert os.stat(value.workspace).st_mode & 0o777 == 0o700
def test_prepare_is_idempotent(tmp_path: Path) -> None:
value = profile(tmp_path)
value.prepare()
marker = value.workspace / "marker"
marker.write_text("kept", encoding="utf-8")
value.prepare()
assert marker.read_text(encoding="utf-8") == "kept"
def test_parent_traversal_is_denied(tmp_path: Path) -> None:
value = profile(tmp_path)
value.prepare()
with pytest.raises(ValidationError, match="escapes"):
value.assert_path_allowed(value.workspace / ".." / "outside")
def test_symlink_inside_profile_cannot_escape(tmp_path: Path) -> None:
value = profile(tmp_path)
value.prepare()
outside = tmp_path / "outside"
outside.mkdir()
(value.workspace / "link").symlink_to(outside, target_is_directory=True)
with pytest.raises(ValidationError, match="escapes"):
value.assert_path_allowed(value.workspace / "link" / "secret")
@pytest.mark.parametrize("root_name", ["home", "workspace"])
def test_prepare_rejects_symlink_roots(tmp_path: Path, root_name: str) -> None:
outside = tmp_path / f"outside-{root_name}"
outside.mkdir()
symlink = tmp_path / root_name
symlink.symlink_to(outside, target_is_directory=True)
changes = {"hermes_home" if root_name == "home" else "workspace": symlink}
with pytest.raises(ValidationError, match="symlink"):
profile(tmp_path, **changes).prepare()
def test_other_profile_scope_is_denied(tmp_path: Path) -> None:
first = profile(tmp_path / "first")
second = profile(tmp_path / "second", tenant_id="tenant-b",
cell_id="cell-b", profile_id="profile-b")
first.prepare()
second.prepare()
with pytest.raises(ValidationError, match="escapes"):
first.assert_path_allowed(second.workspace / "data")
def test_paths_within_either_root_are_allowed(tmp_path: Path) -> None:
value = profile(tmp_path)
value.prepare()
value.assert_path_allowed(value.hermes_home / "config.json")
value.assert_path_allowed(value.workspace / "task.txt")
File diff suppressed because it is too large Load Diff
+53
View File
@@ -0,0 +1,53 @@
"""Executable completeness contract for memory namespaces and access."""
from __future__ import annotations
import pytest
from aios_v2r1.memory_scope import MemoryScope, qualify_memory_key
from aios_v2r1.tenant_context import ScopeError, ValidationError
def scope(kind: str = "session", **changes: object) -> MemoryScope:
values: dict[str, object] = {
"tenant_id": "tenant-a", "cell_id": "cell-a", "profile_id": "profile-a",
"session_id": "session-a", "kind": kind}
values.update(changes)
return MemoryScope(**values) # type: ignore[arg-type]
@pytest.mark.parametrize(("kind", "session_id", "expected"), [
("session", "session-a", "tenant/tenant-a/cell/cell-a/profile/profile-a/session/session-a/memory/session"),
("profile", None, "tenant/tenant-a/cell/cell-a/profile/profile-a/memory/profile"),
("cell", None, "tenant/tenant-a/cell/cell-a/memory/cell"),
("tenant", None, "tenant/tenant-a/cell/cell-a/memory/tenant"),
("shared_artifact", None, "tenant/tenant-a/cell/cell-a/memory/shared_artifact")])
def test_every_namespace_is_explicit(kind: str, session_id: str | None, expected: str) -> None:
assert scope(kind, session_id=session_id).namespace == expected
@pytest.mark.parametrize("requested", [
scope(tenant_id="tenant-b"), scope(cell_id="cell-b"),
scope(profile_id="profile-b"), scope(session_id="session-b")])
def test_cross_scope_access_is_denied(requested: MemoryScope) -> None:
with pytest.raises(ScopeError):
scope().assert_access(requested)
def test_cross_kind_access_is_denied() -> None:
with pytest.raises(ScopeError, match="cross-kind"):
scope("session").assert_access(scope("profile", session_id=None))
def test_shared_artifacts_are_read_only() -> None:
owner = scope()
shared = scope("shared_artifact", session_id=None)
owner.assert_access(shared)
with pytest.raises(ScopeError, match="immutable"):
owner.assert_access(shared, write=True)
def test_qualified_key_contains_the_full_namespace() -> None:
assert qualify_memory_key(scope(), "facts.v1") == (
"tenant/tenant-a/cell/cell-a/profile/profile-a/session/session-a/"
"memory/session/facts.v1")
@pytest.mark.parametrize("key", ["../escape", "a/b", ".", "..", " space ", "x?query"])
def test_memory_key_traversal_and_malformed_keys_are_rejected(key: str) -> None:
with pytest.raises(ValidationError):
qualify_memory_key(scope(), key)
def test_memory_search_surface_is_disabled() -> None:
assert not hasattr(scope(), "search")
assert not hasattr(scope(), "search_all")
+101 -54
View File
@@ -1,73 +1,120 @@
"""Contract tests for the runtime-facing direct MessageLedger append path."""
"""Executable completeness contract for the exact-ID message ledger."""
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
from tempfile import TemporaryDirectory
import unittest
from dataclasses import replace
from pathlib import Path
import sqlite3
import pytest
from aios_v2r1.message_ledger import MessageLedger
from aios_v2r1.tenant_context import CanonicalMessage, ConflictError
from aios_v2r1.tenant_context import CanonicalMessage, ConflictError, ScopeError
def message(*, tenant_id: str = "tenant-a", cell_id: str = "cell-a",
event_id: str = "event-a", message_id: str = "message-a",
body: str = "hello", reply_to_message_id: str | None = None) -> CanonicalMessage:
return CanonicalMessage(
tenant_id=tenant_id, cell_id=cell_id, principal_id="user-a",
def message(**changes: object) -> CanonicalMessage:
value = CanonicalMessage(
tenant_id="tenant-a", cell_id="cell-a", principal_id="user-a",
principal_type="authenticated_user", origin_type="channel_user",
channel="weixin", channel_account_id="account-a", peer_id="peer-a",
conversation_id="conversation-a", session_id="session-a", task_id=None,
event_id=event_id, message_id=message_id, reply_to_message_id=reply_to_message_id,
event_id="event-a", message_id="message-a", reply_to_message_id=None,
direction="inbound", command_capable=True,
occurred_at="2026-07-13T00:00:00+00:00",
received_at="2026-07-13T00:00:01+00:00", body=body,
quoted_body=None, adapter_name="weixin", adapter_version="v2",
)
received_at="2026-07-13T00:00:01+00:00", body="hello",
quoted_body=None, adapter_name="weixin", adapter_version="v2")
return replace(value, **changes)
@pytest.fixture
def ledger(tmp_path: Path) -> MessageLedger:
return MessageLedger(tmp_path / "ledger.sqlite3")
class MessageLedgerAppendContractTests(unittest.TestCase):
def setUp(self) -> None:
self.directory = TemporaryDirectory()
self.addCleanup(self.directory.cleanup)
self.ledger = MessageLedger(f"{self.directory.name}/ledger.sqlite3")
def test_append_writes_directly_without_using_remember(ledger: MessageLedger) -> None:
ledger.remember = lambda _: (_ for _ in ()).throw(AssertionError("remember called")) # type: ignore[method-assign]
assert ledger.append(message()).message_id == "message-a"
assert ledger.count() == 1
def test_append_writes_directly_without_using_remember(self) -> None:
# A compatibility wrapper may remain, but the runtime append path may not
# route through it: this catches append() -> remember() regressions.
self.ledger.remember = lambda _: (_ for _ in ()).throw(AssertionError("remember called")) # type: ignore[method-assign]
record = self.ledger.append(message())
self.assertEqual("message-a", record.message_id)
self.assertEqual(1, self.ledger.count())
def test_append_rejects_same_message_id_with_different_content(ledger: MessageLedger) -> None:
ledger.append(message(body="first"))
with pytest.raises(ConflictError):
ledger.append(message(event_id="event-b", body="second"))
assert ledger.count() == 1
def test_append_rejects_same_message_id_with_different_content(self) -> None:
self.ledger.append(message(body="first"))
with self.assertRaises(ConflictError):
self.ledger.append(message(event_id="event-b", body="second"))
self.assertEqual(1, self.ledger.count())
def test_append_is_idempotent_when_only_event_id_changes(ledger: MessageLedger) -> None:
stored = ledger.append(message(event_id="event-a", body="same body"))
replay = ledger.append(message(event_id="event-b", body="same body"))
assert replay == stored
assert replay.event_id == "event-a"
assert ledger.count() == 1
def test_append_is_idempotent_when_only_event_id_changes(self) -> None:
stored = self.ledger.append(message(event_id="event-a", body="same body"))
replay = self.ledger.append(message(event_id="event-b", body="same body"))
@pytest.mark.parametrize("candidate", ["message", "message-a-extra", "MESSAGE-A"])
def test_quote_lookup_has_no_prefix_similarity_or_case_fallback(
ledger: MessageLedger, candidate: str
) -> None:
ledger.append(message(message_id="message-a"))
with pytest.raises(ScopeError, match="quoted_text_unavailable"):
ledger.resolve_quote(message(event_id="reply-event", message_id="reply-message",
reply_to_message_id=candidate))
self.assertEqual(stored, replay)
self.assertEqual("event-a", replay.event_id)
self.assertEqual(1, self.ledger.count())
def test_explicit_quoted_body_never_falls_back_to_storage(ledger: MessageLedger) -> None:
assert ledger.resolve_quote(message(quoted_body="exact transport quote")) == "exact transport quote"
assert ledger.count() == 0
def test_append_keeps_tenants_isolated(self) -> None:
self.ledger.append(message())
self.ledger.append(message(tenant_id="tenant-b", cell_id="cell-b"))
self.assertEqual(2, self.ledger.count())
self.assertEqual("hello", self.ledger.resolve_quote(
message(tenant_id="tenant-b", cell_id="cell-b", event_id="event-c",
message_id="message-c", body="reply", reply_to_message_id="message-a")
))
@pytest.mark.parametrize("scope_change", [{"tenant_id": "tenant-b"}, {"cell_id": "cell-b"}])
def test_quote_lookup_is_tenant_and_cell_isolated(
ledger: MessageLedger, scope_change: dict[str, str]
) -> None:
ledger.append(message())
with pytest.raises(ScopeError, match="quoted_text_unavailable"):
ledger.resolve_quote(message(**scope_change, event_id="reply-event",
message_id="reply-message",
reply_to_message_id="message-a"))
def test_concurrent_identical_append_is_one_record(self) -> None:
with ThreadPoolExecutor(max_workers=16) as executor:
records = list(executor.map(lambda _: self.ledger.append(message()), range(16)))
self.assertEqual(1, self.ledger.count())
self.assertEqual({"message-a"}, {record.message_id for record in records})
def test_same_id_across_directions_is_ambiguous_not_guessed(ledger: MessageLedger) -> None:
ledger.append(message(direction="inbound", body="in"))
ledger.append(message(direction="outbound", event_id="event-b", body="out"))
with pytest.raises(ConflictError, match="ambiguous"):
ledger.resolve_quote(message(event_id="event-c", message_id="message-c",
reply_to_message_id="message-a"))
def test_expired_record_is_not_resolved(ledger: MessageLedger) -> None:
ledger.append(message())
with sqlite3.connect(ledger.database) as db:
db.execute("UPDATE message_ledger SET expires_at='2000-01-01T00:00:00+00:00'")
with pytest.raises(ScopeError, match="quoted_text_unavailable"):
ledger.resolve_quote(message(event_id="event-b", message_id="message-b",
reply_to_message_id="message-a"))
if __name__ == "__main__":
unittest.main()
def test_purge_removes_only_expired_records(ledger: MessageLedger) -> None:
ledger.append(message())
ledger.append(message(event_id="event-b", message_id="message-b"))
with sqlite3.connect(ledger.database) as db:
db.execute("UPDATE message_ledger SET expires_at='2000-01-01T00:00:00+00:00' "
"WHERE message_id='message-a'")
assert ledger.purge_expired() == 1
assert ledger.count() == 1
def test_restart_preserves_exact_resolution(tmp_path: Path) -> None:
database = tmp_path / "ledger.sqlite3"
MessageLedger(database).append(message())
reopened = MessageLedger(database)
assert reopened.resolve_quote(message(event_id="event-b", message_id="message-b",
reply_to_message_id="message-a")) == "hello"
def test_concurrent_identical_append_is_one_record(ledger: MessageLedger) -> None:
with ThreadPoolExecutor(max_workers=16) as executor:
records = list(executor.map(lambda _: ledger.append(message()), range(16)))
assert ledger.count() == 1
assert {record.message_id for record in records} == {"message-a"}
def test_concurrent_distinct_appends_are_not_lost(ledger: MessageLedger) -> None:
def append(index: int) -> None:
ledger.append(message(event_id=f"event-{index}", message_id=f"message-{index}"))
with ThreadPoolExecutor(max_workers=16) as executor:
list(executor.map(append, range(16)))
assert ledger.count() == 16
def test_sqlite_failures_are_sanitized(ledger: MessageLedger) -> None:
ledger.database.write_bytes(b"not-a-sqlite-database")
with pytest.raises(ConflictError, match="ledger storage unavailable") as raised:
ledger.count()
assert not isinstance(raised.value, sqlite3.Error)
assert "sqlite" not in str(raised.value).lower()
assert raised.value.__cause__ is None
assert raised.value.__context__ is None
+80
View File
@@ -0,0 +1,80 @@
"""Executable completeness contract for scoped secret references."""
from __future__ import annotations
from dataclasses import fields
import pytest
from aios_v2r1.secret_scope import SecretRef
from aios_v2r1.tenant_context import ScopeError, ValidationError
def secret(**changes: object) -> SecretRef:
values: dict[str, object] = {
"secret_id": "channel-key", "tenant_id": "tenant-a", "cell_id": "cell-a",
"kind": "channel", "version": "v1",
"broker_path": "secret://tenant-a/cell-a/channel-key/v1"}
values.update(changes)
return SecretRef(**values) # type: ignore[arg-type]
@pytest.mark.parametrize("path", [
"secret://tenant-a/cell-a/channel-key/v1",
"secret://models/provider/key-v2",
"secret://runtime/cell-a/signing_key"])
def test_valid_broker_paths_are_references(path: str) -> None:
assert secret(broker_path=path).broker_path == path
@pytest.mark.parametrize("path", [
"", "vault://tenant-a/key", "secret://", "secret:///key",
"secret://tenant-a/../tenant-b/key", "secret://tenant-a//key",
"secret://tenant-a/key?version=1", "secret://tenant-a/key#fragment"])
def test_broker_path_traversal_and_malformed_paths_are_rejected(path: str) -> None:
with pytest.raises(ValidationError, match="broker path"):
secret(broker_path=path)
@pytest.mark.parametrize(("tenant_id", "cell_id", "allowed"), [
("tenant-a", "cell-a", True), ("tenant-b", "cell-a", False),
("tenant-a", "cell-b", False)])
def test_secret_scope_is_exact(tenant_id: str, cell_id: str, allowed: bool) -> None:
assert secret().can_be_used_by(tenant_id=tenant_id, cell_id=cell_id) is allowed
@pytest.mark.parametrize("kind", ["channel", "model", "tool", "runtime"])
def test_secret_kind_allowlist(kind: str) -> None:
assert secret(kind=kind).kind == kind
def test_invalid_secret_kind_is_rejected() -> None:
with pytest.raises(ValidationError, match="kind"):
secret(kind="admin")
@pytest.mark.parametrize("changes", [
{"secret_id": "../key"}, {"secret_id": " key "},
{"version": "../v2"}, {"version": ""}, {"tenant_id": "Tenant-A"},
{"cell_id": "cell/a"}])
def test_secret_identifiers_and_version_are_validated(changes: dict[str, str]) -> None:
with pytest.raises(ValidationError):
secret(**changes)
def test_reference_never_contains_a_secret_value_field() -> None:
assert "value" not in {field.name for field in fields(SecretRef)}
assert "super-secret-value" not in repr(secret())
def test_broker_resolution_uses_only_the_validated_reference() -> None:
seen: list[SecretRef] = []
class Broker:
def read_secret(self, reference: SecretRef) -> str:
seen.append(reference)
return "resolved-value"
reference = secret()
assert reference.resolve(Broker()) == "resolved-value"
assert seen == [reference]
@pytest.mark.parametrize("broker", [None, object()])
def test_missing_or_invalid_broker_fails_closed(broker: object | None) -> None:
with pytest.raises(ScopeError, match="secret broker unavailable"):
secret().resolve(broker)
def test_broker_error_is_sanitized_without_secret_leakage() -> None:
class BrokenBroker:
def read_secret(self, reference: SecretRef) -> str:
raise RuntimeError("provider leaked super-secret-value")
with pytest.raises(ScopeError, match="secret broker unavailable") as raised:
secret().resolve(BrokenBroker())
assert "super-secret-value" not in str(raised.value)
assert raised.value.__cause__ is None
assert raised.value.__context__ is None
+117
View File
@@ -0,0 +1,117 @@
# Personal-Weixin B3 evidence collection
此目录提供 fail-closed 的 owner evidence intake/verification 工具。工具不访问网络、不读取
credential 文件,也不负责制造 traffic;它只验证 owner 在受限存储中已经采集的真实材料。
## 安全边界
- raw-wire 必须保存在 repo 之外的 restricted root 下,且不得使用 symlink;
- 工具以 binary chunk 流式计算 raw SHA-256/bytes,不解析、不缓存全文、不输出内容;
- credential 只能使用 owner 提供的非秘密稳定 fingerprint,不能传入 secret/token 文件;
- redacted view 只允许固定 schema 与 `<REDACTED>`/`<ABSENT>` 值;
- reviewer payload 与 hash 必须位于 repo 和 restricted root 之外的临时工作目录;
- 最终输出只允许 `release/v2r1c0-rc1/weixin-bridge-evidence.json`;
- 模板本身证据为空,直接执行必须得到 `BLOCKED_EVIDENCE_REQUIRED`。
- **trust policy 必须**:`prepare-review` 和 `finalize` 均强制 `--trust-policy`。策略文件和三类 allowed-signers 文件必须是绝对路径、UID 0、`mode & 0022 == 0`,且原始路径在任何 open 前逐级 `lstat`,不允许任何 symlink。工具会安全读取 allowed-signers 并重算 SHA-256;与 policy 不一致即阻塞。
- **三类签名没有降级路径**:每条 capture、每条 credential 和 reviewer 都必须同时提供 attestation、SSH signature 与 principal。工具使用固定 `/usr/bin/ssh-keygen`、最小环境和固定超时实际执行 `ssh-keygen -Y verify`。namespace 只能来自 policy,且必须精确为 `aios-pwx-b3-capture`、`aios-pwx-b3-credential`、`aios-pwx-b3-review`。
- **身份隔离**:principal 来自 facts 与已签 attestation,两者必须一致;reviewer principal 必须与 executor 及全部 capture/credential principal 不同。
- **时间与 authority**:reviewer 必须签署 byte-exact manifest SHA-256,`reviewed_at` 不得早于任一 capture `captured_at` 或 credential `tested_at`。inbound 的 vendor message ID 与 client ID 均要求 `present: true`;缺失、`false`、`null` 或 `<ABSENT>` 一律阻塞。`client_id` 永远非权威。
- **dirfd 原子写入**:所有 reviewer 输出文件和 manifest 文件使用基于 dirfd 的原子替换,防止 TOCTOU 和 symlink 劫持。
- **稳定负向接口**:所有阻塞均为 exit code `2`,stdout 精确为 `BLOCKED_EVIDENCE_REQUIRED\n`,stderr 只含一行稳定、脱敏的错误代码;不会透传 `ssh-keygen` 输出或 traceback。
## 1. 准备 owner facts
把模板复制到 owner-controlled、非 Git 的工作目录,再填写真实 metadata:
```bash
install -m 0600 \
tools/personal_weixin/pwx_b3_evidence_facts.template.json \
"$OWNER_WORK/pwx-b3-facts.json"
```
禁止把填写后的 facts、restricted raw、redacted source 或 credential 原文提交 Git。
`restricted_sha256`、`restricted_bytes`、`redacted_sha256`、`redacted_bytes` 必须来自对应
文件的真实值;credential fingerprint 必须由 owner 的既定 secret-manager/authority 流程
提供,工具不会读取 credential 原文。
facts 中每条 `raw_wire` 必须另含 `principal`、`attestation_path` 和 `signature_path`。
capture attestation 是严格 JSON,必须绑定 principal、capture identity、direction、credential
label、capture time,以及工具实算的 restricted/redacted SHA-256 与 bytes。每条 credential
必须另含 `principal`、`tested_at`、`attestation_path`、`signature_path`;credential attestation
必须严格绑定 fingerprint、scope、`PASS` 与 test time。任何字段缺失、额外字段或不一致都会
fail-closed。
Trust Policy 固定结构如下;三份 allowed-signers 可以相同,但每份 digest 都必须实算匹配:
```json
{
"schema_version": 1,
"contract": "personal-weixin-b3-trust-policy",
"capture": {
"allowed_signers": "/root/aios-pwx-b3-capture.allowed_signers",
"allowed_signers_sha256": "<64 lowercase hex>",
"namespace": "aios-pwx-b3-capture"
},
"credential": {
"allowed_signers": "/root/aios-pwx-b3-credential.allowed_signers",
"allowed_signers_sha256": "<64 lowercase hex>",
"namespace": "aios-pwx-b3-credential"
},
"reviewer": {
"allowed_signers": "/root/aios-pwx-b3-review.allowed_signers",
"allowed_signers_sha256": "<64 lowercase hex>",
"namespace": "aios-pwx-b3-review"
}
}
```
## 2. 生成 byte-exact reviewer payload
```bash
python3 tools/personal_weixin/pwx_b3_evidence_tool.py prepare-review \
--facts "$OWNER_WORK/pwx-b3-facts.json" \
--trust-policy "$OWNER_WORK/pwx-b3-trust-policy.json" \
--restricted-root "$OWNER_RESTRICTED_ROOT" \
--repo-root "$(git rev-parse --show-toplevel)" \
--review-payload-out "$REVIEW_WORK/pwx-b3-review-manifest.json" \
--review-hash-out "$REVIEW_WORK/pwx-b3-review-manifest.sha256"
```
当所有非 reviewer evidence 完整时,工具生成两个 0600 reviewer 文件,但仍以 exit code 2
和 `BLOCKED_EVIDENCE_REQUIRED` 结束。独立 reviewer 必须审核该 byte-exact payload,并把
其 SHA-256、identity、decision 和 review time 写入另一份非 Git facts 文件。review block 还必须
填写 `principal`、review attestation 绝对路径与 signature 绝对路径;review attestation 严格
绑定 reviewer、principal、`PASS`、manifest SHA-256 和 `reviewed_at`。
## 3. Finalize
```bash
python3 tools/personal_weixin/pwx_b3_evidence_tool.py finalize \
--facts "$OWNER_WORK/pwx-b3-facts-reviewed.json" \
--trust-policy "$OWNER_WORK/pwx-b3-trust-policy.json" \
--restricted-root "$OWNER_RESTRICTED_ROOT" \
--repo-root "$(git rev-parse --show-toplevel)" \
--review-payload "$REVIEW_WORK/pwx-b3-review-manifest.json" \
--manifest-out "$(git rev-parse --show-toplevel)/release/v2r1c0-rc1/weixin-bridge-evidence.json"
```
`finalize` 会重新读取 metadata、重新流式哈希 restricted raw、重新验证 redacted view,并
重新执行 capture、credential、reviewer 三类真实 SSH 签名验证,要求当前 canonical manifest
与 reviewer payload byte-exact 相同。成功时只输出 `PASS`;
任何缺失、不一致、stale review、非独立 reviewer、production flag、symlink 或 manifest
冲突都安全返回 `BLOCKED_EVIDENCE_REQUIRED`,不会创建、截断或替换不同的 manifest。
## Canonical JSON
Reviewer payload 和最终 envelope 使用 UTF-8、key sort、compact separators、禁止 NaN/float,
并以一个换行结尾。重复 JSON key、未知字段和非 timezone RFC3339 timestamp 都会阻塞。
## 本地验证
```bash
python3 -m py_compile tools/personal_weixin/pwx_b3_evidence_tool.py
python3 -m json.tool tools/personal_weixin/pwx_b3_evidence_facts.template.json >/dev/null
python3 tools/personal_weixin/pwx_b3_evidence_tool.py --help
```
不要把 reviewer 临时文件或真实 owner facts 放入 repository。
@@ -0,0 +1,55 @@
{
"schema_version": 1,
"contract": "personal-weixin-b3-owner-evidence-facts",
"executor": "Codex isolated runner",
"lineage": {
"foundation_implementation_commit": "2fcab6f5833a405a9a26683811600b6e5f1aa7b1",
"foundation_result_commit": "64a71b6199a8e2bd609ba145349e67b874888505",
"r10_lineage_commit": "74e603c0c943501110577dbe3c7f6a0b18b8ec7d",
"evidence_gate_commit": "62a4067cf7a462ebe7cf29524238bc5d0a70b813",
"upstream_audit_commit": "b2f5f531019a70801444987dbb97dc1f69cf47de",
"authorization_commit": "\u00611f933e500151dc8afc1b25b30448a31899a8f64",
"b4_amendment_commit": "1d334b1ccaac54c3c3ec4dd84693ab68fc67cef3",
"authorization_result_commit": "8af40af62927caf631ea6fc682155f0f65cbd1a6"
},
"raw_wire": [],
"credentials": [],
"authority_map": {
"vendor_message_id_field": "$.msgs[*].message_id",
"client_id_field": "$.msgs[*].client_id",
"client_id_authoritative": false,
"mapping_verified": null,
"authoritative": [
"verified inbound vendor message_id"
],
"non_authoritative": [
"local client_id",
"outbound correlation ID",
"local sequence",
"timestamp",
"content hash",
"outbound receipt ID"
]
},
"mode": {
"name": null,
"production_traffic": false,
"writes_production": false,
"capture_only": true
},
"review": {
"reviewer": null,
"principal": null,
"independent_from_executor": null,
"decision": null,
"reviewed_manifest_sha256": null,
"reviewed_at": null,
"attestation_path": null,
"signature_path": null
},
"production": {
"baseline": "V1R4C6",
"changed": false,
"release_status": "paused"
}
}
File diff suppressed because it is too large Load Diff