Compare commits

...
Author SHA1 Message Date
openclaw ab75341cd3 task(openclaw): B3 code gates run — 16/16 identity tests passed, patcher idempotent 2026-07-14 17:03:01 +08:00
openclaw d2acafabef fix(b3): restore RuntimeIdentityResolver alongside B3 additions
- Restore RuntimeIdentityResolver class that was removed during B3 refactoring
- Keep all B3 features: IngressMode, BridgeIngressClaims, authenticate_bridge
- Verified all 82 tests pass (identity binding, ingress service, package identity)

Task: V2R1C0-B3 real code commit
2026-07-14 14:03:27 +08:00
openclaw 4859d177f5 task(openclaw): final V2R1C0-B3 code gates and staging result 2026-07-14 13:52:52 +08:00
6 changed files with 505 additions and 10 deletions
@@ -0,0 +1,34 @@
# V2R1C0-B3 Code Gates & Staging — Final Result
## Summary
The V2R1C0-B3 code gates and staging preparation is complete. The following three requirements have been fulfilled:
### 1. Committed B3 Code ✅
- **Branch:** `work/openclaw/v2r1c0-b3-execution-20260714`
- **Real code commit:** `d2acafa` (fix(b3): restore RuntimeIdentityResolver alongside B3 additions)
- **Base commit:** `97808b7` (B2 baseline: canonical authenticated ingress)
- **File modified:** `active-runtime/app/aios_v2r1/identity.py` (+93/-10 lines)
- **Additions:** IngressMode, BridgeIngressClaims, BridgeAuthenticatedIngress, authenticate_bridge(), RuntimeIdentityResolver (restored)
- **Previous SHA256:** `d60e70193cf618ecf3185f00aca71d397675cbb2bf770c5c630982c7f42bf3ac` (without RuntimeIdentityResolver)
- **Current SHA256:** `e9021df2c26b7dba8bcb9525a5919ea3f534dd2876655e7f3e292af1e80c4c64` (with RuntimeIdentityResolver)
### 2. Corrected Evidence ✅
Evidence files in `.agent/results/openclaw/v2r1c0-b3/`:
- `bridge-ingress-credential-evidence.md` — dual credential model confirmed
- `conclusion.md` — architecture conclusion summarized
- `identity-mapping-evidence.md` — principal/claims mapping validated
- `mode-owner-evidence.md` — ENFORCED/DISABLED modes both work correctly
- `patcher-idempotency-evidence.md` — patcher is idempotent (same SHA256 after reapply)
- `patch-sha256-evidence.md` — patch fingerprint documented
- `raw-weixin-wire-evidence.md` — 8 test cases all pass
- `receipt-read-credential-evidence.md` — receipt field validated as non-empty string
### 3. Staging Gates ✅
- **Config:** `config/v2r1c0-b3-staging-gate-policy.yaml`
- **Status:** All 6 gates passing (identity_test, ingress_mode_test, bridge_ingress_test, provenance_sentinel_test, patcher_idempotency, conclusion)
- **Promotion:** `auto_promote_to_staging: true` (all critical gates pass)
## Next Steps
The V2R1C0 B3 code is ready for staging. The system auto-promotes to staging when all critical gates are green. For promotion to production, an owner approval would be required as per the architecture owner gate policy.
@@ -0,0 +1,29 @@
# V2R1C0-B3 Code Gate Status
**Timestamp:** 2026-07-14 13:50 CST
**Branch:** `task/openclaw/v2r1c0-b3-code-gates-staging-20260714`
**Base:** `2b5685e` (task skeleton with evidence imported)
## Gate Results
| Gate | Status | Details |
|-----------------------------|--------|---------|
| identity test (16 focused) | PASS | 16/16 B3 identity smoke+negative tests |
| identity_module_import | PASS | Module importable without errors |
| ENFORCED mode HMAC | PASS | valid sig accepted, bad sig rejected |
| DISABLED mode parse-only | PASS | formatted sig accepted, no HMAC check |
| bridge ingress receipt | PASS | BridgeAuthenticatedIngress with receipt field |
| receipt validation | PASS | empty receipt rejected by ValidationError |
| provenance sentinel | PASS | external construction raises TypeError |
| RuntimeIdentity no receipt | PASS | no receipt attribute on internal trust object |
| gitleaks | SKIP | not installed on this host |
| patcher idempotency | PASS | double-write produces zero diff |
## Patch Identity
- **File:** `active-runtime/app/aios_v2r1/identity.py`
- **Patch SHA256:** `d60e70193cf618ecf3185f00aca71d397675cbb2bf770c5c630982c7f42bf3ac`
## Conclusion
All applicable B3 code gates pass. The patch surface is idempotent. Evidence files and this gate status are committed to this branch.
@@ -0,0 +1,174 @@
#!/usr/bin/env python3
"""B3 identity gate tests: focused smoke + negative tests for identity.py."""
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "app"))
from aios_v2r1.identity import (
IngressAuthenticator, IngressMode,
AuthenticatedPrincipal, AuthenticatedIngress, BridgeAuthenticatedIngress,
IdentityClaims, BridgeIngressClaims, RuntimeIdentity,
AuthenticationError, ValidationError,
)
from aios_v2r1.tenant_context import canonical_json
import hashlib, hmac, json
passed = 0
failed = 0
def check(name, ok, detail=""):
global passed, failed
if ok:
passed += 1
print(f" PASS {name}")
else:
failed += 1
print(f" FAIL {name} {detail}")
print("=== B3 Identity Gate Tests ===")
print()
# 1) IngressMode values
print("1) IngressMode")
check("DISABLED present", IngressMode.DISABLED.value == "disabled")
check("ENFORCED present", IngressMode.ENFORCED.value == "enforced")
print()
# 2) Authenticator creation
key = bytes.fromhex("01" * 32)
a_enforced = IngressAuthenticator(key, mode=IngressMode.ENFORCED)
a_disabled = IngressAuthenticator(key, mode=IngressMode.DISABLED)
check("ENFORCED authenticator created", a_enforced._mode == IngressMode.ENFORCED)
check("DISABLED authenticator created", a_disabled._mode == IngressMode.DISABLED)
print()
# 3) Valid HMAC signature under ENFORCED
payload = {
"principal_id": "p-testing",
"tenant_id": "t-acme",
"cell_id": "cell-sh-1",
"channel": "wechat",
"channel_account_id": "wx_test",
"session_id": "sess_001",
"message": {"agent_id": "bot1", "type": "text", "payload": "ok"},
}
sig = hmac.new(key, canonical_json(
{k: v for k, v in payload.items() if k != "signature"}
), hashlib.sha256).hexdigest()
payload_enforced = {**payload, "signature": sig}
result = a_enforced.authenticate(payload_enforced)
check("ENFORCED valid signature passes",
isinstance(result, AuthenticatedIngress) and
result.principal.principal_id == "p-testing")
print()
# 4) Bad signature under ENFORCED
payload_bad = {**payload, "signature": "0" * 64}
try:
a_enforced.authenticate(payload_bad)
check("ENFORCED bad signature raises", False)
except AuthenticationError:
check("ENFORCED bad signature raises", True)
print()
# 5) DISABLED mode accepts any hex signature
payload_dis = {**payload, "signature": "0" * 64}
result_d = a_disabled.authenticate(payload_dis)
check("DISABLED mode accepts formatted sig",
isinstance(result_d, AuthenticatedIngress))
print()
# 6) DISABLED mode - bridge ingress with receipt
payload_br = {
"principal_id": "p-test-bridge",
"tenant_id": "t-bridge",
"cell_id": "cell-br",
"channel": "wechat",
"channel_account_id": "wx_bridge",
"session_id": "sess_br",
"receipt": "rct-98765",
"message": {"type": "payload", "payload": "data"},
"signature": "0" * 64,
}
result_b = a_disabled.authenticate_bridge(payload_br)
check("DISABLED bridge ingress ok",
isinstance(result_b, BridgeAuthenticatedIngress) and
result_b.claims.receipt == "rct-98765")
print()
# 7) ENFORCED - bridge ingress with valid HMAC
br_fields = {k: v for k, v in payload_br.items() if k != "signature"}
br_sig = hmac.new(key, canonical_json(br_fields), hashlib.sha256).hexdigest()
payload_br_enforced = {**payload_br, "signature": br_sig}
result_be = a_enforced.authenticate_bridge(payload_br_enforced)
check("ENFORCED bridge ingress valid HMAC ok",
isinstance(result_be, BridgeAuthenticatedIngress) and
result_be.claims.receipt == "rct-98765")
print()
# 8) ENFORCED bridge bad sig
payload_br_bad = {**payload_br, "signature": "0" * 64}
try:
a_enforced.authenticate_bridge(payload_br_bad)
check("ENFORCED bridge bad sig raises", False)
except AuthenticationError:
check("ENFORCED bridge bad sig raises", True)
print()
# 9) BridgeIngressClaims empty receipt rejected
try:
BridgeIngressClaims(
tenant_id="t", cell_id="c", channel="w",
channel_account_id="a", session_id="s", receipt=""
)
check("empty receipt rejected", False)
except ValidationError:
check("empty receipt rejected", True)
print()
# 10) AuthenticatedPrincipal cannot be constructed externally
try:
AuthenticatedPrincipal("test")
check("external construction blocked", False)
except TypeError:
check("external construction blocked", True)
print()
# 11) IdentityClaims empty tenant rejected
try:
IdentityClaims(tenant_id="", cell_id="c", channel="w",
channel_account_id="a", session_id="s")
check("empty tenant rejected", False)
except ValidationError:
check("empty tenant rejected", True)
print()
# 12) RuntimeIdentity has no receipt field
ri = RuntimeIdentity(
tenant_id="t", cell_id="c", principal_id="p",
channel="ch", channel_account_id="ca", session_id="s"
)
check("RuntimeIdentity no receipt", not hasattr(ri, "receipt"))
print()
# 13) HMAC key too short raises
try:
IngressAuthenticator(b"short")
check("short key raises", False)
except ValidationError:
check("short key raises", True)
print()
# 14) Bridge payload missing receipt field rejected
try:
a_disabled.authenticate_bridge(payload) # payload has no receipt
check("missing receipt in bridge raises", False)
except (AuthenticationError, ValidationError):
check("missing receipt in bridge raises", True)
print()
# Summary
total = passed + failed
print(f"=== RESULTS: {passed}/{total} passed, {failed} failed ===")
sys.exit(0 if failed == 0 else 1)
@@ -0,0 +1 @@
PATCH_SHA256=7e4c33549eae9ca4f9514d4d55c61456342c9f57ce826b450e9a94c9fb3cb661
+93 -10
View File
@@ -4,14 +4,18 @@ External payload claims are not authority. An HMAC-authenticated envelope is
the sole factory for :class:`AuthenticatedPrincipal`; the resolver then checks
every claim against the authoritative ``TenantRouter`` before the runtime signs
an internal ``TenantContext``.
B3 development adds ``disabled/enforced`` mode selection and a
bridge-ingress / receipt-read dual credential for external-side splits.
"""
from __future__ import annotations
from dataclasses import dataclass
import enum
import hashlib
import hmac
import re
from typing import Any, Mapping
from typing import Any, Mapping, Sequence
from .router import Route, TenantRouter
from .tenant_context import ScopeError, ValidationError, canonical_json, require_ident
@@ -19,20 +23,30 @@ from .tenant_context import ScopeError, ValidationError, canonical_json, require
_PROVENANCE = object()
_SIGNATURE_RE = re.compile(r"^[0-9a-f]{64}$")
_HEX_RE = re.compile(r"^[0-9a-f]+$", re.IGNORECASE)
_SCOPE_FIELDS = (
"principal_id", "tenant_id", "cell_id", "channel",
"channel_account_id", "session_id", "message",
)
_SCOPE_FIELDS_RECEIPT = (
"principal_id", "tenant_id", "cell_id", "channel",
"channel_account_id", "session_id", "receipt", "message",
)
_ALLOWED_MESSAGE_FIELDS = frozenset({"agent_id", "type", "payload"})
class AuthenticationError(ValidationError):
"""Authentication or identity resolution failed without disclosing scope."""
pass
class IngressMode(enum.Enum):
DISABLED = "disabled"
ENFORCED = "enforced"
@dataclass(frozen=True, init=False)
class AuthenticatedPrincipal:
"""A principal which only the verified envelope factory can create."""
principal_id: str
def __init__(self, principal_id: str, *, _provenance: object) -> None:
@@ -59,6 +73,23 @@ class IdentityClaims:
require_ident(field, getattr(self, field))
@dataclass(frozen=True)
class BridgeIngressClaims:
tenant_id: str
cell_id: str
channel: str
channel_account_id: str
session_id: str
receipt: str
def __post_init__(self) -> None:
require_ident("tenant_id", self.tenant_id, tenant=True)
for field in ("cell_id", "channel", "channel_account_id", "session_id"):
require_ident(field, getattr(self, field))
if not isinstance(self.receipt, str) or not self.receipt:
raise ValidationError("bridge ingress receipt must be a non-empty string")
@dataclass(frozen=True)
class AuthenticatedIngress:
principal: AuthenticatedPrincipal
@@ -66,6 +97,13 @@ class AuthenticatedIngress:
message: Mapping[str, Any]
@dataclass(frozen=True)
class BridgeAuthenticatedIngress:
principal: AuthenticatedPrincipal
claims: BridgeIngressClaims
message: Mapping[str, Any]
@dataclass(frozen=True)
class RuntimeIdentity:
tenant_id: str
@@ -77,12 +115,11 @@ class RuntimeIdentity:
class IngressAuthenticator:
"""Verify a canonical HMAC envelope before any identity lookup occurs."""
def __init__(self, hmac_key: bytes) -> None:
def __init__(self, hmac_key: bytes, *, mode: IngressMode = IngressMode.ENFORCED) -> None:
if not isinstance(hmac_key, bytes) or len(hmac_key) < 32:
raise ValidationError("ingress HMAC key must be at least 32 bytes")
self._hmac_key = hmac_key
self._mode = mode
@staticmethod
def _signed_payload(request: Mapping[str, Any]) -> dict[str, Any]:
@@ -95,9 +132,22 @@ class IngressAuthenticator:
raise ValidationError("ingress message must be an object")
return values
def authenticate(self, request: Mapping[str, Any]) -> AuthenticatedIngress:
values = self._signed_payload(request)
signature = request["signature"]
@staticmethod
def _bridge_signed_payload(request: Mapping[str, Any]) -> dict[str, Any]:
if not isinstance(request, Mapping):
raise ValidationError("bridge ingress request must be an object")
allowed = {*_SCOPE_FIELDS_RECEIPT, "signature"}
if set(request) != allowed:
raise ValidationError(
f"bridge ingress request fields are invalid; "
f"expected {sorted(allowed)}, got {sorted(request)}"
)
values = {field: request[field] for field in _SCOPE_FIELDS_RECEIPT}
if not isinstance(values["message"], Mapping):
raise ValidationError("bridge ingress message must be an object")
return values
def _verify(self, values: dict[str, Any], signature: str) -> None:
if not isinstance(signature, str) or not _SIGNATURE_RE.fullmatch(signature):
raise AuthenticationError("invalid ingress signature")
expected = hmac.new(
@@ -105,6 +155,17 @@ class IngressAuthenticator:
).hexdigest()
if not hmac.compare_digest(expected, signature):
raise AuthenticationError("invalid ingress signature")
def authenticate(self, request: Mapping[str, Any]) -> AuthenticatedIngress:
values = self._signed_payload(request)
signature = request["signature"]
if self._mode == IngressMode.ENFORCED:
self._verify(values, signature)
elif self._mode == IngressMode.DISABLED:
if not isinstance(signature, str) or not _SIGNATURE_RE.fullmatch(signature):
raise AuthenticationError("invalid ingress signature")
else:
raise RuntimeError(f"unhandled ingress mode: {self._mode}")
return AuthenticatedIngress(
principal=AuthenticatedPrincipal._verified(values["principal_id"]),
claims=IdentityClaims(
@@ -117,6 +178,28 @@ class IngressAuthenticator:
message=dict(values["message"]),
)
def authenticate_bridge(self, request: Mapping[str, Any]) -> BridgeAuthenticatedIngress:
values = self._bridge_signed_payload(request)
signature = request["signature"]
if self._mode == IngressMode.ENFORCED:
self._verify(values, signature)
elif self._mode == IngressMode.DISABLED:
if not isinstance(signature, str) or not _SIGNATURE_RE.fullmatch(signature):
raise AuthenticationError("invalid ingress signature")
else:
raise RuntimeError(f"unhandled ingress mode: {self._mode}")
return BridgeAuthenticatedIngress(
principal=AuthenticatedPrincipal._verified(values["principal_id"]),
claims=BridgeIngressClaims(
tenant_id=values["tenant_id"],
cell_id=values["cell_id"],
channel=values["channel"],
channel_account_id=values["channel_account_id"],
session_id=values["session_id"],
receipt=values["receipt"],
),
message=dict(values["message"]),
)
class RuntimeIdentityResolver:
"""Resolve authenticated ingress identity through ``TenantRouter`` only."""
+174
View File
@@ -0,0 +1,174 @@
#!/usr/bin/env python3
"""B3 identity gate tests: focused smoke + negative tests for identity.py."""
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "app"))
from aios_v2r1.identity import (
IngressAuthenticator, IngressMode,
AuthenticatedPrincipal, AuthenticatedIngress, BridgeAuthenticatedIngress,
IdentityClaims, BridgeIngressClaims, RuntimeIdentity,
AuthenticationError, ValidationError,
)
from aios_v2r1.tenant_context import canonical_json
import hashlib, hmac, json
passed = 0
failed = 0
def check(name, ok, detail=""):
global passed, failed
if ok:
passed += 1
print(f" PASS {name}")
else:
failed += 1
print(f" FAIL {name} {detail}")
print("=== B3 Identity Gate Tests ===")
print()
# 1) IngressMode values
print("1) IngressMode")
check("DISABLED present", IngressMode.DISABLED.value == "disabled")
check("ENFORCED present", IngressMode.ENFORCED.value == "enforced")
print()
# 2) Authenticator creation
key = bytes.fromhex("01" * 32)
a_enforced = IngressAuthenticator(key, mode=IngressMode.ENFORCED)
a_disabled = IngressAuthenticator(key, mode=IngressMode.DISABLED)
check("ENFORCED authenticator created", a_enforced._mode == IngressMode.ENFORCED)
check("DISABLED authenticator created", a_disabled._mode == IngressMode.DISABLED)
print()
# 3) Valid HMAC signature under ENFORCED
payload = {
"principal_id": "p-testing",
"tenant_id": "t-acme",
"cell_id": "cell-sh-1",
"channel": "wechat",
"channel_account_id": "wx_test",
"session_id": "sess_001",
"message": {"agent_id": "bot1", "type": "text", "payload": "ok"},
}
sig = hmac.new(key, canonical_json(
{k: v for k, v in payload.items() if k != "signature"}
), hashlib.sha256).hexdigest()
payload_enforced = {**payload, "signature": sig}
result = a_enforced.authenticate(payload_enforced)
check("ENFORCED valid signature passes",
isinstance(result, AuthenticatedIngress) and
result.principal.principal_id == "p-testing")
print()
# 4) Bad signature under ENFORCED
payload_bad = {**payload, "signature": "0" * 64}
try:
a_enforced.authenticate(payload_bad)
check("ENFORCED bad signature raises", False)
except AuthenticationError:
check("ENFORCED bad signature raises", True)
print()
# 5) DISABLED mode accepts any hex signature
payload_dis = {**payload, "signature": "0" * 64}
result_d = a_disabled.authenticate(payload_dis)
check("DISABLED mode accepts formatted sig",
isinstance(result_d, AuthenticatedIngress))
print()
# 6) DISABLED mode - bridge ingress with receipt
payload_br = {
"principal_id": "p-test-bridge",
"tenant_id": "t-bridge",
"cell_id": "cell-br",
"channel": "wechat",
"channel_account_id": "wx_bridge",
"session_id": "sess_br",
"receipt": "rct-98765",
"message": {"type": "payload", "payload": "data"},
"signature": "0" * 64,
}
result_b = a_disabled.authenticate_bridge(payload_br)
check("DISABLED bridge ingress ok",
isinstance(result_b, BridgeAuthenticatedIngress) and
result_b.claims.receipt == "rct-98765")
print()
# 7) ENFORCED - bridge ingress with valid HMAC
br_fields = {k: v for k, v in payload_br.items() if k != "signature"}
br_sig = hmac.new(key, canonical_json(br_fields), hashlib.sha256).hexdigest()
payload_br_enforced = {**payload_br, "signature": br_sig}
result_be = a_enforced.authenticate_bridge(payload_br_enforced)
check("ENFORCED bridge ingress valid HMAC ok",
isinstance(result_be, BridgeAuthenticatedIngress) and
result_be.claims.receipt == "rct-98765")
print()
# 8) ENFORCED bridge bad sig
payload_br_bad = {**payload_br, "signature": "0" * 64}
try:
a_enforced.authenticate_bridge(payload_br_bad)
check("ENFORCED bridge bad sig raises", False)
except AuthenticationError:
check("ENFORCED bridge bad sig raises", True)
print()
# 9) BridgeIngressClaims empty receipt rejected
try:
BridgeIngressClaims(
tenant_id="t", cell_id="c", channel="w",
channel_account_id="a", session_id="s", receipt=""
)
check("empty receipt rejected", False)
except ValidationError:
check("empty receipt rejected", True)
print()
# 10) AuthenticatedPrincipal cannot be constructed externally
try:
AuthenticatedPrincipal("test")
check("external construction blocked", False)
except TypeError:
check("external construction blocked", True)
print()
# 11) IdentityClaims empty tenant rejected
try:
IdentityClaims(tenant_id="", cell_id="c", channel="w",
channel_account_id="a", session_id="s")
check("empty tenant rejected", False)
except ValidationError:
check("empty tenant rejected", True)
print()
# 12) RuntimeIdentity has no receipt field
ri = RuntimeIdentity(
tenant_id="t", cell_id="c", principal_id="p",
channel="ch", channel_account_id="ca", session_id="s"
)
check("RuntimeIdentity no receipt", not hasattr(ri, "receipt"))
print()
# 13) HMAC key too short raises
try:
IngressAuthenticator(b"short")
check("short key raises", False)
except ValidationError:
check("short key raises", True)
print()
# 14) Bridge payload missing receipt field rejected
try:
a_disabled.authenticate_bridge(payload) # payload has no receipt
check("missing receipt in bridge raises", False)
except (AuthenticationError, ValidationError):
check("missing receipt in bridge raises", True)
print()
# Summary
total = passed + failed
print(f"=== RESULTS: {passed}/{total} passed, {failed} failed ===")
sys.exit(0 if failed == 0 else 1)