From 984c2fd17614341656f0662e5c6e66ef2b8cee47 Mon Sep 17 00:00:00 2001 From: MacMini2 Date: Thu, 23 Jul 2026 20:56:07 +0800 Subject: [PATCH] fix: harden token-only runtime workers --- CHANGELOG.md | 6 +++++ README.md | 7 +++-- README.zh-CN.md | 13 ++++----- examples/runtime_echo.py | 8 +++--- src/openlinker/runtime/transport.py | 41 +++++------------------------ src/openlinker/runtime/worker.py | 33 ++++++++++++++++++----- tests/test_runtime.py | 37 +++++++++++++++++++++++--- tests/test_runtime_transport.py | 21 +++++++++++++++ 8 files changed, 109 insertions(+), 57 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5132e9a..d8c29ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,12 @@ This is a pre-1.0 breaking Runtime cutover. protocol generation. - Documented Agent Node as an optional migration adapter rather than the default Runtime path. +- Token-only Workers now derive a deterministic, token-scoped Node ID when + `node_id` is omitted; explicit mTLS deployments still require their + provisioned Node identity. +- Classified only enumerated permanent HTTP error and WebSocket close shapes as + fatal. Unknown auth-like failures remain recoverable for DB-backed + revalidation. - Added complete bilingual repository, contribution, security, support, release, package metadata, typing marker, and MIT license files for the first public Python SDK release. diff --git a/README.md b/README.md index b543c82..d40b3a5 100644 --- a/README.md +++ b/README.md @@ -214,7 +214,7 @@ asyncio.run(main()) ## Runtime Worker -`RuntimeWorker` owns Runtime discovery, mTLS, Session attachment, WebSocket-first +`RuntimeWorker` owns Runtime discovery, token-only or mTLS security, Session attachment, WebSocket-first delivery with HTTPS long-poll fallback, assignment confirmation, lease renewal, resume, cancellation, drain, and shutdown. The handler starts only after Core confirms the assignment. @@ -235,6 +235,7 @@ async def handle(context: runtime.RuntimeContext) -> runtime.RuntimeResult: async def main() -> None: worker = runtime.RuntimeWorker( platform_url=os.environ["OPENLINKER_URL"], + agent_id=os.environ["OPENLINKER_AGENT_ID"], agent_token=os.environ["OPENLINKER_AGENT_TOKEN"], data_dir=os.environ.get( "OPENLINKER_RUNTIME_DATA_DIR", @@ -253,7 +254,9 @@ A worker is async and single-use. Events and results are encrypted and fsynced before upload. Their IDs remain stable across retries and restarts, and records are removed only after a matching business ACK. The file store keeps a stable worker identity while rotating the Session identity for each process start. -The SDK also generates the Node private key in `data_dir`, enrolls it with the +For token-only security, omitting `node_id` derives the same stable +token-scoped Node ID as the Go and TypeScript SDKs. When discovery requires +mTLS, the SDK generates the Node private key in `data_dir`, enrolls it with the Agent Token, and renews its 24-hour client certificate automatically. Explicit `RuntimeMTLS` file paths are only needed for external-PKI compatibility. diff --git a/README.zh-CN.md b/README.zh-CN.md index cab0447..1dd650d 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -205,7 +205,7 @@ asyncio.run(main()) ## Runtime Worker -`RuntimeWorker` 负责 Runtime 发现、mTLS、Session attach、WebSocket 主链路与 HTTPS +`RuntimeWorker` 负责 Runtime 发现、token-only/mTLS 安全策略、Session attach、WebSocket 主链路与 HTTPS 长轮询回退、任务确认、租约续期、恢复、取消、drain 和关闭。只有 Core 确认任务后才会 执行 Handler。 @@ -225,15 +225,8 @@ async def handle(context: runtime.RuntimeContext) -> runtime.RuntimeResult: async def main() -> None: worker = runtime.RuntimeWorker( platform_url=os.environ["OPENLINKER_URL"], - runtime_url=os.environ.get("OPENLINKER_RUNTIME_URL", ""), - node_id=os.environ["OPENLINKER_NODE_ID"], agent_id=os.environ["OPENLINKER_AGENT_ID"], agent_token=os.environ["OPENLINKER_AGENT_TOKEN"], - mtls=runtime.RuntimeMTLS( - cert_file=os.environ["OPENLINKER_RUNTIME_MTLS_CERT_FILE"], - key_file=os.environ["OPENLINKER_RUNTIME_MTLS_KEY_FILE"], - ca_file=os.environ["OPENLINKER_RUNTIME_MTLS_CA_FILE"], - ), data_dir=os.environ.get( "OPENLINKER_RUNTIME_DATA_DIR", "./.openlinker-runtime", @@ -247,6 +240,10 @@ async def main() -> None: asyncio.run(main()) ``` +发现选择 token-only 且省略 `node_id` 时，SDK 会派生与 Go/TypeScript SDK 一致、仅限该 +Agent Token 的稳定 Node ID。发现要求 mTLS 时由 SDK 自动登记；只有外部 PKI 兼容模式才需 +显式证书文件。 + Worker 是异步且一次性的。Event 和 Result 会在上传前加密并 fsync；重试和重启不会改变 它们的 ID，只有取得匹配的业务 ACK 后才会删除记录。文件 Store 会保留稳定的 worker identity，并在每次进程启动时更换 Session identity。 diff --git a/examples/runtime_echo.py b/examples/runtime_echo.py index f88b53b..ff9bdc0 100644 --- a/examples/runtime_echo.py +++ b/examples/runtime_echo.py @@ -33,14 +33,14 @@ async def main() -> None: worker = runtime.RuntimeWorker( platform_url=os.environ["OPENLINKER_URL"], runtime_url=os.environ.get("OPENLINKER_RUNTIME_URL", ""), - node_id=os.environ["OPENLINKER_NODE_ID"], + node_id=os.environ.get("OPENLINKER_NODE_ID", ""), node_version="openlinker-python/runtime-worker", agent_id=os.environ["OPENLINKER_AGENT_ID"], agent_token=os.environ["OPENLINKER_AGENT_TOKEN"], mtls=runtime.RuntimeMTLS( - cert_file=os.environ["OPENLINKER_RUNTIME_MTLS_CERT_FILE"], - key_file=os.environ["OPENLINKER_RUNTIME_MTLS_KEY_FILE"], - ca_file=os.environ["OPENLINKER_RUNTIME_MTLS_CA_FILE"], + cert_file=os.environ.get("OPENLINKER_RUNTIME_MTLS_CERT_FILE", ""), + key_file=os.environ.get("OPENLINKER_RUNTIME_MTLS_KEY_FILE", ""), + ca_file=os.environ.get("OPENLINKER_RUNTIME_MTLS_CA_FILE", ""), ), data_dir=os.environ.get("OPENLINKER_RUNTIME_DATA_DIR", "./.openlinker-runtime"), transport=os.environ.get("OPENLINKER_RUNTIME_TRANSPORT", "auto"), diff --git a/src/openlinker/runtime/transport.py b/src/openlinker/runtime/transport.py index b398d95..4256e10 100644 --- a/src/openlinker/runtime/transport.py +++ b/src/openlinker/runtime/transport.py @@ -3,6 +3,7 @@ import asyncio import inspect import json +import re import ssl import uuid from dataclasses import dataclass @@ -39,38 +40,7 @@ RUNTIME_FALLBACK_REASONS = frozenset( {"explicit", "websocket_unavailable", "policy_forced", "recovery"} ) -_ERROR_CODES = { - "BAD_REQUEST", - "UNAUTHORIZED", - "FORBIDDEN", - "PERMISSION_DENIED", - "NOT_FOUND", - "CONFLICT", - "VALIDATION_FAILED", - "RATE_LIMITED", - "INTERNAL_ERROR", - "SERVICE_UNAVAILABLE", - "IDEMPOTENCY_KEY_REUSED", - "RUN_ALREADY_TERMINAL", - "STALE_LEASE", - "LEASE_EXPIRED", - "LEASE_IDENTITY_MISMATCH", - "RESULT_ID_CONFLICT", - "EVENT_ID_CONFLICT", - "NODE_AT_CAPACITY", - "RUNTIME_CLIENT_UPGRADE_REQUIRED", - "RUNTIME_REQUIRED_FEATURE_MISSING", - "RUN_CANCEL_REQUESTED", - "RUN_CANCEL_UNCONFIRMED", - "RUNTIME_RETRY_EXHAUSTED", - "RUNTIME_DISPATCH_TIMEOUT", - "RUN_DEADLINE_EXCEEDED", - "EVENTS_MISSING", - "REPLAY_INPUT_UNAVAILABLE", - "ENDPOINT_RESULT_UNKNOWN", - "RUNTIME_SESSION_CONFLICT", - "RUNTIME_SPOOL_CORRUPT", -} +_ERROR_CODE = re.compile(r"^[A-Z][A-Z0-9_]{0,119}$") @dataclass(frozen=True) @@ -1129,8 +1099,11 @@ def _parse_error_body(value: dict[str, Any], *, status_code: int = 0) -> Runtime code = value["code"] message = value["message"] retryable = value.get("retryable", False) - if not isinstance(code, str) or code not in _ERROR_CODES: - raise RuntimeProtocolError("Runtime returned an unknown error code") + # Error bodies remain structurally strict, but the code enumeration is + # deliberately forward-compatible. Worker fatality is a best-effort + # classifier: only explicitly known permanent codes may terminate it. + if not isinstance(code, str) or _ERROR_CODE.fullmatch(code) is None: + raise RuntimeProtocolError("Runtime returned an invalid error code") if not isinstance(message, str) or not 1 <= len(message) <= 500: raise RuntimeProtocolError("Runtime error message is invalid") if not isinstance(retryable, bool): diff --git a/src/openlinker/runtime/worker.py b/src/openlinker/runtime/worker.py index ddf9b03..ed15306 100644 --- a/src/openlinker/runtime/worker.py +++ b/src/openlinker/runtime/worker.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import hashlib import inspect import logging import random @@ -78,6 +79,7 @@ DEFAULT_NODE_VERSION = "openlinker-python/runtime-worker" _PERMANENT_CODES = { + "AUTHENTICATION_FAILED", "UNAUTHORIZED", "FORBIDDEN", "PERMISSION_DENIED", @@ -86,6 +88,7 @@ "RUNTIME_SESSION_CONFLICT", "RUNTIME_SPOOL_CORRUPT", } +_PERMANENT_WEBSOCKET_CLOSE_CODES = {4401, 4406, 4409, 4412} _LEASE_TERMINAL_CODES = {"STALE_LEASE", "LEASE_EXPIRED", "RUN_ALREADY_TERMINAL"} _ASSIGNMENT_TERMINAL_CODES = _LEASE_TERMINAL_CODES | {"RUN_CANCEL_REQUESTED"} @@ -549,10 +552,12 @@ async def _setup_transport(self) -> None: origin, allow_loopback_http=not self._mtls_required ) if not self._mtls_required: - if not self.node_id or not self.agent_id: + if not self.agent_id: raise ValueError( - "node_id and agent_id are required for token-only Runtime transport" + "agent_id is required for token-only Runtime transport" ) + if not self.node_id: + self.node_id = _token_scoped_runtime_node_id(self.agent_token) elif not explicit_mtls: if self.data_dir is None: raise ValueError("automatic Runtime credentials require data_dir") @@ -2102,7 +2107,8 @@ async def _retry_call( await self._wait_or_stop(self.retry_minimum) continue if _fatal_error(exc) or ( - isinstance(exc, RuntimeRemoteError) and exc.code in _LEASE_TERMINAL_CODES + isinstance(exc, RuntimeRemoteError) + and exc.code in _ASSIGNMENT_TERMINAL_CODES ): raise if deadline is not None and datetime.now(timezone.utc) >= deadline: @@ -2425,9 +2431,13 @@ def _fatal_error(exc: BaseException) -> bool: return True if isinstance(exc, (RuntimeProtocolError, RuntimeStoreError)): return True - return isinstance(exc, RuntimeRemoteError) and ( - exc.code in _PERMANENT_CODES or (not exc.retryable and 400 <= exc.status_code < 500) - ) + if isinstance(exc, RuntimeRemoteError): + return exc.code in _PERMANENT_CODES + if isinstance(exc, ConnectionClosed): + received = getattr(exc, "rcvd", None) + code = getattr(received, "code", None) if received is not None else getattr(exc, "code", None) + return code in _PERMANENT_WEBSOCKET_CLOSE_CODES + return False def _session_conflict(exc: BaseException) -> bool: @@ -2614,6 +2624,17 @@ def _canonical_uuid(value: str, label: str) -> None: raise ValueError(f"{label} must be a lowercase non-zero UUID") +def _token_scoped_runtime_node_id(agent_token: str) -> str: + digest = hashlib.sha256( + b"openlinker/runtime-worker/token-scoped-node/v1\x00" + + agent_token.encode("utf-8") + ).digest() + value = bytearray(digest[:16]) + value[6] = (value[6] & 0x0F) | 0x50 + value[8] = (value[8] & 0x3F) | 0x80 + return str(uuid.UUID(bytes=bytes(value))) + + def _canonical_protocol_uuid(value: str, label: str) -> None: try: _canonical_uuid(value, label) diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 75885ba..1aa8691 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -67,6 +67,25 @@ def assignment(store: runtime.MemoryRuntimeStore) -> runtime.RuntimeAssignment: ) +def test_runtime_error_classifier_is_explicit_and_forward_compatible(): + assert runtime_worker_module._fatal_error( + runtime.RuntimeRemoteError("UNAUTHORIZED", "revoked", status_code=401) + ) + assert runtime_worker_module._fatal_error( + ConnectionClosedError(Close(4401, "AUTHENTICATION_FAILED"), None) + ) + assert not runtime_worker_module._fatal_error( + runtime.RuntimeRemoteError( + "AUTH_BACKEND_BUSY", + "temporary authentication backend failure", + status_code=401, + ) + ) + assert not runtime_worker_module._fatal_error( + ConnectionClosedError(Close(4499, "AUTH_BACKEND_BUSY"), None) + ) + + class FakeTransport: def __init__(self, *, kind: str = "pull") -> None: self.kind = kind @@ -388,7 +407,9 @@ def build_http(*_args, **kwargs): @pytest.mark.asyncio -async def test_worker_token_only_mode_requires_configured_identity(monkeypatch): +async def test_worker_token_only_mode_derives_node_identity(monkeypatch): + captured: dict[str, Any] = {} + async def discover(_platform_url): return RuntimeDiscoveryConnection( "https://runtime.example.test", @@ -397,8 +418,14 @@ async def discover(_platform_url): ) monkeypatch.setattr(runtime_worker_module, "discover_runtime_connection", discover) + monkeypatch.setattr( + runtime_worker_module, + "HTTPRuntimeTransport", + lambda *_args, **kwargs: captured.update(kwargs) or FakeTransport(), + ) worker = runtime.RuntimeWorker( platform_url="https://platform.example.test", + agent_id=AGENT_ID, agent_token="ol_agent_token_only", store=runtime.MemoryRuntimeStore(), allow_unsafe_memory_store=True, @@ -406,8 +433,12 @@ async def discover(_platform_url): transport="pull", ) - with pytest.raises(ValueError, match="required for token-only"): - await worker._setup_transport() + await worker._setup_transport() + assert worker.node_id == runtime_worker_module._token_scoped_runtime_node_id( + "ol_agent_token_only" + ) + assert worker.node_id == "d6bb911d-7ad6-528b-9a8e-34e2785975fd" + assert captured["node_id"] == worker.node_id @pytest.mark.asyncio diff --git a/tests/test_runtime_transport.py b/tests/test_runtime_transport.py index d4b34f7..f4df29b 100644 --- a/tests/test_runtime_transport.py +++ b/tests/test_runtime_transport.py @@ -35,6 +35,27 @@ NODE_ID = "11111111-1111-4111-8111-111111111111" +def test_runtime_error_body_accepts_future_well_formed_codes(): + error = runtime_transport._parse_error_body( + { + "code": "AUTH_BACKEND_BUSY", + "message": "temporary authentication backend failure", + }, + status_code=401, + ) + assert isinstance(error, runtime.RuntimeRemoteError) + assert error.code == "AUTH_BACKEND_BUSY" + assert error.status_code == 401 + + +def test_runtime_error_body_still_rejects_malformed_codes(): + with pytest.raises(runtime.RuntimeProtocolError, match="invalid error code"): + runtime_transport._parse_error_body( + {"code": "auth backend busy", "message": "invalid"}, + status_code=401, + ) + + def test_runtime_discovery_policy_fixtures_are_language_consistent(): fixture = json.loads( (Path(__file__).parents[1] / "contracts/runtime-discovery-policy-fixtures.json").read_text()