160 lines
6.2 KiB
Python
160 lines
6.2 KiB
Python
"""Security contracts for authenticated OpenClaw runtime identity binding."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
from datetime import datetime, timezone
|
|
from tempfile import TemporaryDirectory
|
|
import unittest
|
|
|
|
from aios_v2r1.runtime import RuntimeWiring
|
|
from aios_v2r1.tenant_context import canonical_json
|
|
|
|
|
|
class MutableClock:
|
|
def __init__(self, now: datetime) -> None:
|
|
self.now = now
|
|
|
|
def __call__(self) -> datetime:
|
|
return self.now
|
|
|
|
|
|
class RuntimeIdentityBindingTests(unittest.TestCase):
|
|
internal_key = b"internal-context-signing-key-is-32-bytes"
|
|
ingress_key = b"external-ingress-hmac-key-is-32-bytes"
|
|
|
|
def setUp(self) -> None:
|
|
self.directory = TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.clock = MutableClock(datetime(2026, 7, 13, tzinfo=timezone.utc))
|
|
self.runtime = RuntimeWiring.create(
|
|
self.directory.name,
|
|
signing_key=self.internal_key,
|
|
ingress_hmac_key=self.ingress_key,
|
|
clock=self.clock,
|
|
)
|
|
self.runtime.router.register_tenant("tenant-a", "cell-a")
|
|
self.runtime.router.register_tenant("tenant-b", "cell-b")
|
|
self.runtime.router.bind_channel(
|
|
tenant_id="tenant-a", channel="weixin", channel_account_id="account-a"
|
|
)
|
|
self.runtime.router.bind_channel(
|
|
tenant_id="tenant-b", channel="weixin", channel_account_id="account-b"
|
|
)
|
|
self.runtime.router.bind_principal(
|
|
tenant_id="tenant-a", channel="weixin", channel_account_id="account-a",
|
|
principal_id="user-a",
|
|
)
|
|
self.runtime.router.bind_session(
|
|
tenant_id="tenant-a", session_id="session-a", cell_id="cell-a",
|
|
principal_id="user-a", ttl_seconds=60,
|
|
)
|
|
|
|
def message(self, **changes: object) -> dict[str, object]:
|
|
values: dict[str, object] = {
|
|
"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-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",
|
|
"content": {"body": "hello", "quoted_body": None},
|
|
"adapter": {"name": "weixin", "version": "v2"},
|
|
}
|
|
values.update(changes)
|
|
return values
|
|
|
|
def authenticated_principal_type(self):
|
|
try:
|
|
from aios_v2r1.identity import AuthenticatedPrincipal
|
|
except ImportError as exc:
|
|
self.fail(f"Identity resolver contract is missing: {exc}")
|
|
return AuthenticatedPrincipal
|
|
|
|
def signed_request(self, **changes: object) -> dict[str, object]:
|
|
values: dict[str, object] = {
|
|
"principal_id": "user-a",
|
|
"tenant_id": "tenant-a",
|
|
"cell_id": "cell-a",
|
|
"channel": "weixin",
|
|
"channel_account_id": "account-a",
|
|
"session_id": "session-a",
|
|
"message": self.message(),
|
|
}
|
|
values.update(changes)
|
|
signed = canonical_json(values)
|
|
values["signature"] = hmac.new(
|
|
self.ingress_key, signed, hashlib.sha256
|
|
).hexdigest()
|
|
return values
|
|
|
|
def test_raw_payload_cannot_construct_authenticated_principal(self) -> None:
|
|
AuthenticatedPrincipal = self.authenticated_principal_type()
|
|
with self.assertRaises(TypeError):
|
|
AuthenticatedPrincipal("user-a") # type: ignore[call-arg]
|
|
|
|
def test_forged_tenant_claim_is_rejected_after_valid_hmac(self) -> None:
|
|
request = self.signed_request(
|
|
tenant_id="tenant-b", cell_id="cell-b",
|
|
message=self.message(tenant_id="tenant-b", cell_id="cell-b"),
|
|
)
|
|
|
|
result = self.runtime.ingest(request=request)
|
|
|
|
self.assertEqual("AUTH_FAILED", result.status)
|
|
self.assertIsNone(result.result)
|
|
self.assertEqual(0, self.runtime.ledger.count())
|
|
|
|
def test_invalid_principal_is_rejected_after_valid_hmac(self) -> None:
|
|
request = self.signed_request(
|
|
principal_id="unknown-user",
|
|
message=self.message(principal_id="unknown-user"),
|
|
)
|
|
|
|
result = self.runtime.ingest(request=request)
|
|
|
|
self.assertEqual("AUTH_FAILED", result.status)
|
|
self.assertIsNone(result.result)
|
|
self.assertEqual(0, self.runtime.ledger.count())
|
|
|
|
def test_registered_principal_is_resolved_before_internal_context_is_issued(self) -> None:
|
|
request = self.signed_request()
|
|
result = self.runtime.ingest(request=request)
|
|
|
|
self.assertEqual("ACCEPTED", result.status)
|
|
self.assertIsNotNone(result.result)
|
|
assert result.result is not None
|
|
self.assertEqual("tenant-a", result.result.route.tenant_id)
|
|
self.assertEqual("message-a", result.result.record.message_id)
|
|
self.assertEqual(1, self.runtime.ledger.count())
|
|
self.assertEqual(1, self.runtime.session_pool.active_count())
|
|
|
|
def test_cross_tenant_channel_claim_cannot_retarget_authenticated_principal(self) -> None:
|
|
request = self.signed_request(
|
|
tenant_id="tenant-b", cell_id="cell-b", channel_account_id="account-b",
|
|
message=self.message(
|
|
tenant_id="tenant-b", cell_id="cell-b", channel_account_id="account-b"
|
|
),
|
|
)
|
|
|
|
result = self.runtime.ingest(request=request)
|
|
|
|
self.assertEqual("AUTH_FAILED", result.status)
|
|
self.assertEqual(0, self.runtime.ledger.count())
|
|
self.assertEqual(0, self.runtime.session_pool.active_count())
|
|
|
|
def test_hmac_binds_all_scope_claims(self) -> None:
|
|
request = self.signed_request()
|
|
request["tenant_id"] = "tenant-b"
|
|
|
|
result = self.runtime.ingest(request=request)
|
|
|
|
self.assertEqual("AUTH_FAILED", result.status)
|
|
self.assertEqual(0, self.runtime.ledger.count())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|