From 7416d03207e98c80ecd4555281f1a7886e597e57 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=A1=82=E9=A9=AC?= Date: Mon, 31 Aug 2026 19:04:53 +0800 Subject: [PATCH] fix(a2a): stabilize permission checkpoint identity --- .../permission_wait_fixture_server.py | 18 ++ src/iac_code/a2a/executor.py | 98 +++++++- src/iac_code/a2a/input_required.py | 68 ++++-- src/iac_code/a2a/runtime_overrides.py | 2 + src/iac_code/agent/agent_loop.py | 26 ++- src/iac_code/services/permission_wait.py | 138 +++++++++++- .../services/providers/aliyun_identity.py | 165 ++++++++++++++ tests/a2a/test_events.py | 210 +++++++++++++++++- tests/a2a/test_executor.py | 62 +++++- tests/a2a/test_input_required.py | 9 + tests/services/test_aliyun_identity.py | 109 +++++++++ tests/services/test_permission_wait.py | 92 ++++++++ 12 files changed, 934 insertions(+), 63 deletions(-) create mode 100644 src/iac_code/services/providers/aliyun_identity.py create mode 100644 tests/services/test_aliyun_identity.py diff --git a/scripts/a2a/e2e/permission_wait/permission_wait_fixture_server.py b/scripts/a2a/e2e/permission_wait/permission_wait_fixture_server.py index 0169e0bf..b4ebeecc 100644 --- a/scripts/a2a/e2e/permission_wait/permission_wait_fixture_server.py +++ b/scripts/a2a/e2e/permission_wait/permission_wait_fixture_server.py @@ -491,12 +491,30 @@ def main() -> int: os.environ["IAC_CODE_CONFIG_DIR"] = str(config_dir) os.environ["IAC_CODE_MODE"] = args.mode os.environ["IACCODE_A2A_ALLOWED_CWDS"] = str(workspace) + # Cloud-shaped permission fixtures must exercise the durable identity + # contract without making a real STS request. Keep one stable caller + # across both fixture-server generations while using explicitly fake + # credentials for the request-local credential plumbing. + os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"] = "permission-wait-fixture-ak" + os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"] = "permission-wait-fixture-secret" + os.environ["ALIBABA_CLOUD_REGION_ID"] = "cn-hangzhou" import uvicorn from iac_code.a2a import executor as executor_module from iac_code.a2a import pipeline_executor as pipeline_executor_module from iac_code.a2a.app import create_app + from iac_code.services.providers.aliyun_identity import AliyunCallerIdentity, AliyunCallerIdentityResolver + + async def resolve_fixture_identity( + self: AliyunCallerIdentityResolver, + credential: Any, + region_id: str, + ) -> AliyunCallerIdentity: + del self, credential, region_id + return AliyunCallerIdentity(kind="ram_role", account_id="1000000000000001", subject_id="fixture-role-id") + + AliyunCallerIdentityResolver.resolve = resolve_fixture_identity executor_module.create_agent_runtime = lambda options: _create_fixture_runtime( options, diff --git a/src/iac_code/a2a/executor.py b/src/iac_code/a2a/executor.py index 27c61242..04e64119 100644 --- a/src/iac_code/a2a/executor.py +++ b/src/iac_code/a2a/executor.py @@ -31,6 +31,7 @@ ) from iac_code.a2a.exposure import normalize_a2a_exposure_types from iac_code.a2a.input_required import ( + PermissionIdentityValidationError, PermissionInputRegistry, PermissionResponse, backup_permission_wait_checkpoint, @@ -110,11 +111,12 @@ PermissionWaitCheckpointStore, RecoveredPermissionAuditBoundary, canonical_digest, - permission_execution_identity, recover_permission_audit_boundary, + resolve_permission_execution_identity, ) from iac_code.services.permissions.audit import emit_permission_boundary_audit from iac_code.services.providers.aliyun import DEFAULT_REGION, AliyunCredential, AliyunCredentials +from iac_code.services.providers.aliyun_identity import AliyunCallerIdentityUnavailableError from iac_code.services.session_backup import ( BackupReason, SessionBackupBlocked, @@ -1245,10 +1247,23 @@ async def _execute(self, context: RequestContext, event_queue: EventQueue, *, co task_id = requested_task_id or "task-" + uuid.uuid4().hex[:12] permission_response = parse_permission_response(getattr(context, "message", None)) if permission_response is not None: + response_metadata = getattr(context, "metadata", None) or getattr( + getattr(context, "message", None), "metadata", None + ) + response_credential = self._resolve_aliyun_credential(response_metadata) pending = None try: pending = await self._permission_input_registry.pending_for_response(permission_response) - approved = await self._permission_input_registry.answer(permission_response) + with a2a_request_context(aliyun_credential=response_credential): + approved = await self._permission_input_registry.answer(permission_response) + except PermissionIdentityValidationError as exc: + await self._publish_permission_identity_error( + event_queue, + response=permission_response, + code=exc.code, + retryable=exc.retryable, + ) + return except InvalidParamsError: if pending is not None: await self._permission_input_registry.complete(pending) @@ -2251,15 +2266,46 @@ def make_pipeline_executor() -> IacCodeA2APipelineExecutor: except (OSError, RuntimeError, TypeError, ValueError) as exc: raise InvalidParamsError(f"permission_resume_invalid: {exc}") from exc - permission_audit = getattr(audit_event.permission_result, "audit", None) - with a2a_request_context(aliyun_credential=aliyun_credential): - principal_ref, region = permission_execution_identity( - tool_name=audit_event.tool_name, - tool_input=audit_event.tool_input, - permission_audit=permission_audit, - ) - if principal_ref != record.get("principalRef") or region != record.get("region"): - raise InvalidParamsError("permission_resume_invalid: cloud execution identity changed.") + if expected_value == "allow_once": + if record.get("identitySchemaVersion") != 1: + await self._publish_permission_identity_error( + event_queue, + response=response, + code="legacy_permission_identity", + retryable=False, + session_id=context_record.session_id, + ) + return True + permission_audit = getattr(audit_event.permission_result, "audit", None) + try: + with a2a_request_context(aliyun_credential=aliyun_credential): + identity = await resolve_permission_execution_identity( + tool_name=audit_event.tool_name, + tool_input=audit_event.tool_input, + permission_audit=permission_audit, + ) + except AliyunCallerIdentityUnavailableError as exc: + await self._publish_permission_identity_error( + event_queue, + response=response, + code=exc.reason, + retryable=exc.retryable, + session_id=context_record.session_id, + ) + return True + if ( + identity.principal_ref != record.get("principalRef") + or identity.principal_kind != record.get("principalKind") + or identity.region != record.get("region") + ): + await self._publish_permission_identity_error( + event_queue, + response=response, + code="cloud_execution_identity_changed", + retryable=False, + session_id=context_record.session_id, + ) + return True record = store.reconcile_deadline( boundary_id, @@ -2595,6 +2641,34 @@ async def _publish_permission_recovery_ack( session_id=session_id, ) + async def _publish_permission_identity_error( + self, + event_queue: EventQueue, + *, + response: PermissionResponse, + code: str, + retryable: bool, + session_id: str | None = None, + ) -> None: + await self._publish_status( + event_queue, + task_id=response.task_id, + context_id=response.context_id, + state=TaskState.TASK_STATE_INPUT_REQUIRED, + text=_("A temporary error occurred. Please retry."), + metadata={ + "iac_code": { + "permissionIdentityError": { + "code": code, + "retryable": retryable, + "inputId": response.input_id, + "toolUseId": response.tool_use_id, + } + } + }, + session_id=session_id, + ) + async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None: task_id = context.task_id context_id = context.context_id or "unknown" @@ -3354,6 +3428,8 @@ def _is_normal_handoff(handoff: dict[str, Any]) -> bool: def _is_retryable_executor_error(exc: Exception) -> bool: + if isinstance(exc, AliyunCallerIdentityUnavailableError): + return exc.retryable return isinstance( exc, ( diff --git a/src/iac_code/a2a/input_required.py b/src/iac_code/a2a/input_required.py index e2cbfaad..613e3704 100644 --- a/src/iac_code/a2a/input_required.py +++ b/src/iac_code/a2a/input_required.py @@ -21,7 +21,7 @@ PermissionWaitPolicy, build_permission_checkpoint, canonicalize_permission_continuation_frame, - permission_execution_identity, + resolve_permission_execution_identity, ) from iac_code.services.permissions.audit import ( build_display_tool_input, @@ -29,6 +29,7 @@ emit_permission_boundary_audit, sanitize_prompt_text, ) +from iac_code.services.providers.aliyun_identity import AliyunCallerIdentityUnavailableError from iac_code.types.stream_events import PermissionRequestEvent PERMISSION_SCHEMA_VERSION = 1 @@ -37,6 +38,16 @@ _SAFE_SUMMARY_MAX_CHARS = 1200 _DISPLAY_FIELD_MAX_CHARS = 500 + +class PermissionIdentityValidationError(InvalidParamsError): + """A permission decision could not safely use the current cloud identity.""" + + def __init__(self, code: str, *, retryable: bool) -> None: + super().__init__(f"permission_resume_invalid: {code}") + self.code = code + self.retryable = retryable + + _ROS_TEMPLATE_API_ACTIONS = { "ros_validate_template": "ValidateTemplate", "ros_get_template_parameter_constraints": "GetTemplateParameterConstraints", @@ -878,8 +889,12 @@ def fenced_backup(_cwd: str, _session_id: str, **kwargs: Any) -> Any: ) except ValueError as exc: raise SessionBackupBlocked("Permission checkpoint changed during critical backup.") from exc - if getattr(result, "enabled", False) and not getattr(result, "shared_committed", False): - raise SessionBackupBlocked("Critical permission backup did not reach the shared target.") + if ( + getattr(result, "enabled", False) + and not getattr(result, "shared_committed", False) + and not getattr(result, "staged_committed", False) + ): + raise SessionBackupBlocked("Critical permission backup did not reach a durable target.") return result @@ -932,8 +947,17 @@ async def open_durable_boundary( frame = canonicalize_permission_continuation_frame(source_frame, audit_context=audit_context) except ValueError as exc: raise RuntimeError(f"permission_resume_invalid: {exc}") from exc - principal_ref = audit_context.get("principal_ref") - region = audit_context.get("region") + permission_audit = getattr(pending.request.permission_result, "audit", None) + execution_identity = await resolve_permission_execution_identity( + tool_name=pending.request.tool_name, + tool_input=pending.request.tool_input, + permission_audit=permission_audit, + ) + audit_context.update( + principal_ref=execution_identity.principal_ref, + principal_kind=execution_identity.principal_kind, + region=execution_identity.region, + ) record = build_permission_checkpoint( session_id=session_id, task_id=pending.task_id, @@ -945,8 +969,9 @@ async def open_durable_boundary( permission_class="pipeline" if permission_class == "pipeline" else "normal", continuation_frame=frame, policy=coordinator.policy, - principal_ref=principal_ref if isinstance(principal_ref, str) else None, - region=region if isinstance(region, str) else None, + principal_ref=execution_identity.principal_ref, + principal_kind=execution_identity.principal_kind, + region=execution_identity.region, pipeline_coordinates=pipeline_coordinates, ) previous_boundary_id = frame.get("previousBoundaryId") @@ -1112,7 +1137,8 @@ async def answer(self, response: PermissionResponse) -> bool: coordinator = self._permission_wait_coordinator if coordinator is not None and pending.boundary_id is not None: - self._validate_live_execution_identity(pending) + if response.decision == "allow_once": + await self._validate_live_execution_identity(pending) def audit_new_claim(value: str) -> bool: return emit_permission_boundary_audit( @@ -1160,7 +1186,7 @@ def audit_new_claim(value: str) -> bool: return approved @staticmethod - def _validate_live_execution_identity(pending: PendingPermission) -> None: + async def _validate_live_execution_identity(pending: PendingPermission) -> None: store = pending.checkpoint_store boundary_id = pending.boundary_id if store is None or boundary_id is None: @@ -1169,13 +1195,22 @@ def _validate_live_execution_identity(pending: PendingPermission) -> None: if record is None: raise InvalidParamsError("permission_resume_invalid: permission checkpoint is unavailable.") permission_audit = getattr(pending.request.permission_result, "audit", None) - principal_ref, region = permission_execution_identity( - tool_name=pending.request.tool_name, - tool_input=pending.request.tool_input, - permission_audit=permission_audit, - ) - if principal_ref != record.get("principalRef") or region != record.get("region"): - raise InvalidParamsError("permission_resume_invalid: cloud execution identity changed.") + if record.get("identitySchemaVersion") != 1: + raise PermissionIdentityValidationError("legacy_permission_identity", retryable=False) + try: + identity = await resolve_permission_execution_identity( + tool_name=pending.request.tool_name, + tool_input=pending.request.tool_input, + permission_audit=permission_audit, + ) + except AliyunCallerIdentityUnavailableError as exc: + raise PermissionIdentityValidationError(exc.reason, retryable=exc.retryable) from exc + if ( + identity.principal_ref != record.get("principalRef") + or identity.principal_kind != record.get("principalKind") + or identity.region != record.get("region") + ): + raise PermissionIdentityValidationError("cloud_execution_identity_changed", retryable=False) async def _backup_claim_before_delivery(self, pending: PendingPermission, record: dict[str, Any]) -> None: if pending.before_claim_backup is not None: @@ -1413,6 +1448,7 @@ def _permission_input_id() -> str: __all__ = [ "PERMISSION_QUERY_PREFIX", "PendingPermission", + "PermissionIdentityValidationError", "PermissionInputRegistry", "PermissionResolutionOwner", "PermissionResponse", diff --git a/src/iac_code/a2a/runtime_overrides.py b/src/iac_code/a2a/runtime_overrides.py index e748da67..3fbe3849 100644 --- a/src/iac_code/a2a/runtime_overrides.py +++ b/src/iac_code/a2a/runtime_overrides.py @@ -9,6 +9,7 @@ from iac_code.i18n import SUPPORTED_LANGUAGES, use_request_language from iac_code.providers.request_policy import ProviderRequestPolicy +from iac_code.services.permission_wait import permission_execution_identity_cache_scope from iac_code.services.providers.aliyun import AliyunCredential, use_aliyun_credential from iac_code.services.telemetry import use_session_id, use_telemetry_channel, use_user_id @@ -52,6 +53,7 @@ def a2a_request_context( telemetry_channel: str | None = None, ) -> Iterator[None]: with contextlib.ExitStack() as stack: + stack.enter_context(permission_execution_identity_cache_scope()) if telemetry_channel: stack.enter_context(use_telemetry_channel(telemetry_channel)) if preferred_language: diff --git a/src/iac_code/agent/agent_loop.py b/src/iac_code/agent/agent_loop.py index 557bb40b..94678be2 100644 --- a/src/iac_code/agent/agent_loop.py +++ b/src/iac_code/agent/agent_loop.py @@ -1257,15 +1257,18 @@ async def resume_permission_boundary( if request_index == current_index: if permission is None: raise ValueError("permission_resume_invalid: current tool is unavailable") - principal_ref, region = permission_execution_identity( - tool_name=request.name, - tool_input=request.input, - permission_audit=getattr(permission, "audit", None), - ) - if principal_ref != checkpoint.get("principalRef") or region != checkpoint.get("region"): - raise ValueError("permission_resume_invalid: cloud execution identity changed") state = "allow" if decision["value"] == "allow_once" else "deny" source = "user" + principal_ref = None + region = None + if state == "allow": + principal_ref, region = permission_execution_identity( + tool_name=request.name, + tool_input=request.input, + permission_audit=getattr(permission, "audit", None), + ) + if principal_ref != checkpoint.get("principalRef") or region != checkpoint.get("region"): + raise ValueError("permission_resume_invalid: cloud execution identity changed") if state == "deny": recorded["deniedResult"] = _user_denied_tool_result() if permission.behavior != "deny": @@ -2198,6 +2201,15 @@ async def _run_streaming_inner( approved = False denial_source = "audit_failure" if approved: + # A2A durable permission publication may have resolved a + # stable STS caller while this generator was suspended. + # Re-read the request-local identity so continuation + # decisions never retain a rotating temporary AccessKeyId. + principal_ref, region = permission_execution_identity( + tool_name=request.name, + tool_input=request.input, + permission_audit=permission.audit, + ) allowed_requests.append(request) continuation_decisions[request_index].update( state="allow", diff --git a/src/iac_code/services/permission_wait.py b/src/iac_code/services/permission_wait.py index 6b119b6c..e1485bba 100644 --- a/src/iac_code/services/permission_wait.py +++ b/src/iac_code/services/permission_wait.py @@ -3,18 +3,26 @@ from __future__ import annotations import asyncio +import contextvars import hashlib import json import logging import math import re import uuid -from collections.abc import Awaitable, Callable, Mapping +from collections.abc import Awaitable, Callable, Iterator, Mapping +from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Literal, cast +from iac_code.services.providers.aliyun_identity import ( + AliyunCallerIdentity, + AliyunCallerIdentityResolver, + AliyunCallerIdentityUnavailableError, + CallerIdentityKind, +) from iac_code.services.session_layout import SessionPaths, ensure_session_owned_dir from iac_code.services.session_storage import SessionStorage from iac_code.types.stream_events import PermissionWaitOutcome @@ -22,6 +30,7 @@ from iac_code.utils.state_io import atomic_write_json, cross_process_file_lock PermissionClass = Literal["normal", "pipeline"] +PermissionPrincipalKind = CallerIdentityKind PermissionPhase = Literal[ "WAITING", "TIMEOUT_GRACE", @@ -40,6 +49,17 @@ logger = logging.getLogger(__name__) +@dataclass +class _PermissionIdentityCache: + identity: AliyunCallerIdentity | None = None + + +_permission_identity_cache: contextvars.ContextVar[_PermissionIdentityCache | None] = contextvars.ContextVar( + "iac_code_permission_identity_cache", + default=None, +) + + def _parse_timeout(value: object, *, name: str, allow_zero: bool) -> float | None: if value is None: return None @@ -129,6 +149,27 @@ def canonical_digest(value: object) -> str: return hashlib.sha256(encoded.encode("utf-8")).hexdigest() +@contextmanager +def permission_execution_identity_cache_scope() -> Iterator[None]: + """Install one request-local caller-identity cache, preserving nested scopes.""" + + if _permission_identity_cache.get() is not None: + yield + return + token = _permission_identity_cache.set(_PermissionIdentityCache()) + try: + yield + finally: + _permission_identity_cache.reset(token) + + +@dataclass(frozen=True) +class PermissionExecutionIdentity: + principal_ref: str | None + principal_kind: PermissionPrincipalKind | None + region: str | None + + @dataclass(frozen=True) class RecoveredPermissionAuditBoundary: """Canonical tool/audit data reconstructed from a persisted transcript.""" @@ -257,25 +298,17 @@ def recover_permission_audit_boundary( return None -def permission_execution_identity( +def _permission_operation_scope( *, tool_name: str, tool_input: Mapping[str, Any], permission_audit: object | None = None, -) -> tuple[str | None, str | None]: - """Return a non-secret Alibaba Cloud principal fingerprint and effective Region. - - Local permissions are deliberately not coupled to Alibaba Cloud credentials. - For a cloud operation, an unavailable stable credential anchor remains - ``None`` so durable recovery can fail closed instead of treating an - unknown principal as an approval for the current process identity. - """ - +) -> tuple[bool, str | None]: operation = getattr(permission_audit, "operation", None) operation = operation if isinstance(operation, Mapping) else {} cloud_operation = bool(operation.get("product")) or tool_name == "aliyun_api" or tool_name.startswith("ros_") if not cloud_operation: - return None, None + return False, None from iac_code.services.providers.aliyun import AliyunCredentials @@ -290,6 +323,75 @@ def permission_execution_identity( if (not isinstance(region, str) or not region) and credential is not None: region = credential.region_id effective_region = region if isinstance(region, str) and region else None + return True, effective_region + + +def _cached_caller_identity() -> AliyunCallerIdentity | None: + cache = _permission_identity_cache.get() + return cache.identity if cache is not None else None + + +def _principal_ref(identity: AliyunCallerIdentity) -> str: + return "aliyun:" + canonical_digest(identity.canonical_value()) + + +async def resolve_permission_execution_identity( + *, + tool_name: str, + tool_input: Mapping[str, Any], + permission_audit: object | None = None, + resolver: AliyunCallerIdentityResolver | None = None, +) -> PermissionExecutionIdentity: + """Resolve and request-cache a stable caller through STS GetCallerIdentity.""" + + cloud_operation, effective_region = _permission_operation_scope( + tool_name=tool_name, + tool_input=tool_input, + permission_audit=permission_audit, + ) + if not cloud_operation: + return PermissionExecutionIdentity(None, None, None) + + from iac_code.services.providers.aliyun import AliyunCredentials + + credential = AliyunCredentials.load() + if credential is None: + raise AliyunCallerIdentityUnavailableError("cloud_credentials_unavailable", retryable=False) + identity = _cached_caller_identity() + if identity is None: + identity = await (resolver or AliyunCallerIdentityResolver()).resolve( + credential, + effective_region or credential.region_id, + ) + cache = _permission_identity_cache.get() + if cache is not None: + cache.identity = identity + return PermissionExecutionIdentity(_principal_ref(identity), identity.kind, effective_region) + + +def permission_execution_identity( + *, + tool_name: str, + tool_input: Mapping[str, Any], + permission_audit: object | None = None, +) -> tuple[str | None, str | None]: + """Return the cached stable caller, with a legacy fallback outside A2A waits.""" + + cloud_operation, effective_region = _permission_operation_scope( + tool_name=tool_name, + tool_input=tool_input, + permission_audit=permission_audit, + ) + if not cloud_operation: + return None, None + + from iac_code.services.providers.aliyun import AliyunCredentials + + credential = AliyunCredentials.load() + if credential is not None: + identity = _cached_caller_identity() + if identity is not None: + return _principal_ref(identity), effective_region if credential is None: return None, effective_region @@ -740,6 +842,12 @@ def _validate_record(self, record: Mapping[str, Any]) -> str: raise ValueError("invalid permission checkpoint generation") if not isinstance(record.get("payloadDigest"), str) or not _SHA256.fullmatch(record["payloadDigest"]): raise ValueError("invalid permission payload digest") + identity_schema = record.get("identitySchemaVersion") + if identity_schema not in {None, 1}: + raise ValueError("invalid permission identity schema") + principal_kind = record.get("principalKind") + if principal_kind not in {None, "account", "ram_user", "ram_role"}: + raise ValueError("invalid permission principal kind") if record.get("phase") in _ACTIVE_PHASES: permission_class = record.get("permissionClass") mode = record.get("mode") @@ -1224,6 +1332,7 @@ def build_permission_checkpoint( continuation_frame: Mapping[str, Any], policy: PermissionWaitPolicy, principal_ref: str | None = None, + principal_kind: PermissionPrincipalKind | None = None, region: str | None = None, pipeline_coordinates: Mapping[str, Any] | None = None, now: datetime | None = None, @@ -1246,7 +1355,9 @@ def build_permission_checkpoint( "taskId": task_id, "contextId": context_id, "sessionId": session_id, + "identitySchemaVersion": 1, "principalRef": principal_ref, + "principalKind": principal_kind, "region": region, "mode": "normal" if permission_class == "normal" else "pipeline", "permissionClass": permission_class, @@ -1266,6 +1377,7 @@ def build_permission_checkpoint( __all__ = [ + "PermissionExecutionIdentity", "PermissionWaitCheckpointStore", "PermissionWaitCoordinator", "PermissionWaitPolicy", @@ -1276,5 +1388,7 @@ def build_permission_checkpoint( "format_utc", "parse_utc", "permission_execution_identity", + "permission_execution_identity_cache_scope", "recover_permission_audit_boundary", + "resolve_permission_execution_identity", ] diff --git a/src/iac_code/services/providers/aliyun_identity.py b/src/iac_code/services/providers/aliyun_identity.py new file mode 100644 index 00000000..bdf10ca5 --- /dev/null +++ b/src/iac_code/services/providers/aliyun_identity.py @@ -0,0 +1,165 @@ +"""Stable Alibaba Cloud caller identity resolution for permission checkpoints.""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from typing import Any, Literal + +from iac_code.services.providers.aliyun import AliyunCredential + +logger = logging.getLogger(__name__) + +CallerIdentityKind = Literal["account", "ram_user", "ram_role"] +IdentityRequest = Callable[[AliyunCredential, str], Awaitable[Mapping[str, Any]]] +Sleep = Callable[[float], Awaitable[None]] + +_TRANSIENT_CODES = { + "InternalError", + "ServiceUnavailable", + "SystemBusy", + "Throttling", + "Throttling.User", + "Throttling.Api", +} + + +class AliyunCallerIdentityUnavailableError(RuntimeError): + """The current cloud caller could not be verified safely.""" + + def __init__(self, reason: str, *, retryable: bool) -> None: + super().__init__(reason) + self.reason = reason + self.retryable = retryable + + +@dataclass(frozen=True) +class AliyunCallerIdentity: + """Stable, non-secret fields returned by STS GetCallerIdentity.""" + + kind: CallerIdentityKind + account_id: str + subject_id: str + + def canonical_value(self) -> dict[str, str]: + return { + "identityType": self.kind, + "accountId": self.account_id, + "subjectId": self.subject_id, + } + + +class AliyunCallerIdentityResolver: + """Resolve a stable caller, retrying only transient STS failures.""" + + def __init__( + self, + *, + request: IdentityRequest | None = None, + sleep: Sleep = asyncio.sleep, + retry_delays: tuple[float, ...] = (0.1, 0.3), + ) -> None: + self._request = request or _call_get_caller_identity + self._sleep = sleep + self._retry_delays = retry_delays + + async def resolve(self, credential: AliyunCredential, region_id: str) -> AliyunCallerIdentity: + attempts = len(self._retry_delays) + 1 + for attempt in range(attempts): + try: + response = await self._request(credential, region_id) + return _parse_caller_identity(response) + except AliyunCallerIdentityUnavailableError as exc: + error = exc + except Exception as exc: # SDK errors intentionally stay behind the service boundary. + error = _normalize_identity_error(exc) + if not error.retryable or attempt >= len(self._retry_delays): + raise error + delay = self._retry_delays[attempt] + logger.warning( + "GetCallerIdentity transient failure attempt=%s max_attempts=%s retry_delay_seconds=%s reason=%s", + attempt + 1, + attempts, + delay, + error.reason, + ) + await self._sleep(delay) + raise AliyunCallerIdentityUnavailableError("caller_identity_unavailable", retryable=False) + + +def _parse_caller_identity(response: Mapping[str, Any]) -> AliyunCallerIdentity: + identity_type = response.get("IdentityType") + account_id = response.get("AccountId") + if not isinstance(identity_type, str) or not isinstance(account_id, str) or not account_id: + raise AliyunCallerIdentityUnavailableError("caller_identity_response_invalid", retryable=False) + if identity_type == "Account": + return AliyunCallerIdentity(kind="account", account_id=account_id, subject_id=account_id) + if identity_type == "RAMUser": + user_id = response.get("UserId") + if isinstance(user_id, str) and user_id: + return AliyunCallerIdentity(kind="ram_user", account_id=account_id, subject_id=user_id) + elif identity_type == "AssumedRoleUser": + role_id = response.get("RoleId") + if isinstance(role_id, str) and role_id: + return AliyunCallerIdentity(kind="ram_role", account_id=account_id, subject_id=role_id) + raise AliyunCallerIdentityUnavailableError("caller_identity_response_invalid", retryable=False) + + +def _normalize_identity_error(exc: Exception) -> AliyunCallerIdentityUnavailableError: + status = getattr(exc, "status_code", None) + code = getattr(exc, "code", None) or getattr(exc, "error_code", None) + if not isinstance(code, str): + data = getattr(exc, "data", None) + code = data.get("Code") if isinstance(data, Mapping) else None + retryable = isinstance(exc, (TimeoutError, ConnectionError, OSError)) + retryable = retryable or status == 429 or (isinstance(status, int) and status >= 500) + retryable = retryable or (isinstance(code, str) and (code in _TRANSIENT_CODES or code.startswith("Throttling"))) + reason = code if isinstance(code, str) and code else type(exc).__name__ + return AliyunCallerIdentityUnavailableError(reason, retryable=retryable) + + +async def _call_get_caller_identity(credential: AliyunCredential, region_id: str) -> Mapping[str, Any]: + from alibabacloud_tea_openapi import models as open_api_models + from alibabacloud_tea_openapi.client import Client as OpenApiClient + from darabonba.runtime import RuntimeOptions + + from iac_code.services.providers.aliyun_credentials_runtime import aliyun_credential_runtime + + config_values: dict[str, Any] = { + "endpoint": "sts.aliyuncs.com", + "region_id": region_id, + } + dynamic_client = aliyun_credential_runtime().sdk_client(credential) + if dynamic_client is not None: + config_values["credential"] = dynamic_client + else: + config_values["access_key_id"] = credential.access_key_id + config_values["access_key_secret"] = credential.access_key_secret + if credential.mode in {"StsToken", "OAuth"}: + config_values["security_token"] = credential.sts_token + client = OpenApiClient(open_api_models.Config(**config_values)) + params = open_api_models.Params( + action="GetCallerIdentity", + version="2015-04-01", + protocol="HTTPS", + pathname="/", + method="POST", + auth_type="AK", + style="RPC", + body_type="json", + req_body_type="json", + ) + runtime = RuntimeOptions(autoretry=False, max_attempts=1) + result = await client.call_api_async(params, open_api_models.OpenApiRequest(), runtime) + body = result.get("body", result) + return body if isinstance(body, Mapping) else {} + + +__all__ = [ + "AliyunCallerIdentity", + "AliyunCallerIdentityResolver", + "AliyunCallerIdentityUnavailableError", + "CallerIdentityKind", +] diff --git a/tests/a2a/test_events.py b/tests/a2a/test_events.py index 6fdf10c0..c1eb0137 100644 --- a/tests/a2a/test_events.py +++ b/tests/a2a/test_events.py @@ -3,7 +3,6 @@ import pytest from a2a.types import TaskArtifactUpdateEvent -from a2a.utils.errors import InvalidParamsError from google.protobuf.json_format import MessageToDict from iac_code.a2a.events import ( @@ -15,16 +14,21 @@ publish_stream_event, ) from iac_code.a2a.exposure import A2AExposureType -from iac_code.a2a.input_required import PermissionInputRegistry, PermissionResponse +from iac_code.a2a.input_required import ( + PermissionIdentityValidationError, + PermissionInputRegistry, + PermissionResponse, +) from iac_code.a2a.projection import project_a2a_data from iac_code.services.permission_wait import ( + PermissionExecutionIdentity, PermissionWaitCheckpointStore, PermissionWaitCoordinator, PermissionWaitPolicy, - permission_execution_identity, ) from iac_code.services.permissions.audit import fingerprint_text from iac_code.services.providers.aliyun import AliyunCredential, AliyunCredentials +from iac_code.services.providers.aliyun_identity import AliyunCallerIdentityUnavailableError from iac_code.services.session_backup import BackupReason, SessionBackupBlocked from iac_code.services.session_storage import SessionStorage from iac_code.tools.cloud.aliyun.result_contract import ALIYUN_HTTP_METADATA_KEY @@ -54,11 +58,13 @@ def __init__( *, fail: bool = False, shared_committed: bool = True, + staged_committed: bool = False, ) -> None: self.queue = queue self.store = store self.fail = fail self.shared_committed = shared_committed + self.staged_committed = staged_committed self.calls: list[tuple[BackupReason, bool]] = [] def backup_session(self, _cwd, _session_id, *, reason, critical): @@ -75,6 +81,7 @@ def backup_session(self, _cwd, _session_id, *, reason, critical): succeeded=True, retry_count=0, shared_committed=self.shared_committed, + staged_committed=self.staged_committed, ) @@ -84,6 +91,11 @@ def _durable_permission_fixture(tmp_path, monkeypatch): workspace.mkdir() monkeypatch.setenv("IAC_CODE_CONFIG_DIR", str(config_dir)) monkeypatch.setattr(AliyunCredentials, "load", staticmethod(lambda: None)) + + async def stable_identity(**_kwargs): + return PermissionExecutionIdentity("aliyun:stable-principal", "ram_role", "cn-hangzhou") + + monkeypatch.setattr("iac_code.a2a.input_required.resolve_permission_execution_identity", stable_identity) session_id = "session-1" SessionStorage().ensure_v2_session_dir_for_new_session(str(workspace), session_id) store = PermissionWaitCheckpointStore(str(workspace), session_id) @@ -441,8 +453,9 @@ async def test_external_permission_is_backed_up_before_input_required_is_visible checkpoint = store.list_active() assert len(checkpoint) == 1 assert checkpoint[0]["phase"] == "WAITING" - assert checkpoint[0]["principalRef"] == "aliyun:principal-fingerprint" - assert checkpoint[0]["region"] == "cn-shanghai" + assert checkpoint[0]["principalRef"] == "aliyun:stable-principal" + assert checkpoint[0]["principalKind"] == "ram_role" + assert checkpoint[0]["region"] == "cn-hangzhou" await registry.complete(pending) future.cancel() @@ -600,7 +613,7 @@ async def test_failed_decision_backup_keeps_claim_retriable_without_delivering_f @pytest.mark.asyncio -async def test_live_cloud_permission_rejects_changed_principal_before_claim(tmp_path, monkeypatch) -> None: +async def test_live_cloud_permission_accepts_rotated_sts_for_same_stable_principal(tmp_path, monkeypatch) -> None: workspace, session_id, store, registry = _durable_permission_fixture(tmp_path, monkeypatch) queue = FakeEventQueue() backup = _ObservedBoundaryBackup(queue, store) @@ -613,7 +626,6 @@ async def test_live_cloud_permission_rejects_changed_principal_before_claim(tmp_ ) monkeypatch.setattr(AliyunCredentials, "load", staticmethod(lambda: original)) tool_input = {"product": "ros", "action": "CreateStack", "region_id": "cn-hangzhou"} - principal_ref, region = permission_execution_identity(tool_name="aliyun_api", tool_input=tool_input) future = pending_future() event = PermissionRequestEvent( tool_name="aliyun_api", @@ -621,7 +633,6 @@ async def test_live_cloud_permission_rejects_changed_principal_before_claim(tmp_ tool_use_id="tool-write", response_future=future, continuation_frame=_permission_frame("tool-write"), - audit_context={"principal_ref": principal_ref, "region": region}, ) pending = await publish_interactive_permission_boundary( queue, @@ -643,7 +654,58 @@ async def test_live_cloud_permission_rejects_changed_principal_before_claim(tmp_ ) monkeypatch.setattr(AliyunCredentials, "load", staticmethod(lambda: changed)) - with pytest.raises(InvalidParamsError, match="cloud execution identity changed"): + assert await registry.answer( + PermissionResponse( + task_id="task-1", + context_id="ctx-1", + request_task_id="task-1", + input_id=pending.input_id, + tool_use_id="tool-write", + decision="allow_once", + ) + ) + + assert future.result() is True + assert store.load(str(pending.boundary_id))["decision"]["status"] == "applied" + await registry.complete(pending) + + +@pytest.mark.asyncio +async def test_live_cloud_permission_rejects_different_stable_principal_before_claim(tmp_path, monkeypatch) -> None: + workspace, session_id, store, registry = _durable_permission_fixture(tmp_path, monkeypatch) + queue = FakeEventQueue() + backup = _ObservedBoundaryBackup(queue, store) + identities = iter( + [ + PermissionExecutionIdentity("aliyun:role-a", "ram_role", "cn-hangzhou"), + PermissionExecutionIdentity("aliyun:role-b", "ram_role", "cn-hangzhou"), + ] + ) + + async def changing_identity(**_kwargs): + return next(identities) + + monkeypatch.setattr("iac_code.a2a.input_required.resolve_permission_execution_identity", changing_identity) + future = pending_future() + pending = await publish_interactive_permission_boundary( + queue, + permission_event=PermissionRequestEvent( + tool_name="aliyun_api", + tool_input={"product": "ros", "action": "CreateStack", "region_id": "cn-hangzhou"}, + tool_use_id="tool-write", + response_future=future, + continuation_frame=_permission_frame("tool-write"), + ), + permission_input_registry=registry, + task_id="task-1", + context_id="ctx-1", + iac_code_session_id=session_id, + permission_wait_cwd=str(workspace), + permission_wait_backup_service=backup, + wait_for_response=False, + ) + + with pytest.raises(PermissionIdentityValidationError, match="cloud_execution_identity_changed"): await registry.answer( PermissionResponse( task_id="task-1", @@ -661,6 +723,103 @@ async def test_live_cloud_permission_rejects_changed_principal_before_claim(tmp_ future.cancel() +@pytest.mark.asyncio +async def test_live_identity_failure_keeps_permission_unclaimed_for_retry(tmp_path, monkeypatch) -> None: + workspace, session_id, store, registry = _durable_permission_fixture(tmp_path, monkeypatch) + queue = FakeEventQueue() + backup = _ObservedBoundaryBackup(queue, store) + calls = 0 + + async def identity(**_kwargs): + nonlocal calls + calls += 1 + if calls == 1: + return PermissionExecutionIdentity("aliyun:role-a", "ram_role", "cn-hangzhou") + raise AliyunCallerIdentityUnavailableError("InternalError", retryable=True) + + monkeypatch.setattr("iac_code.a2a.input_required.resolve_permission_execution_identity", identity) + future = pending_future() + pending = await publish_interactive_permission_boundary( + queue, + permission_event=PermissionRequestEvent( + tool_name="aliyun_api", + tool_input={"product": "ros", "action": "CreateStack", "region_id": "cn-hangzhou"}, + tool_use_id="tool-write", + response_future=future, + continuation_frame=_permission_frame("tool-write"), + ), + permission_input_registry=registry, + task_id="task-1", + context_id="ctx-1", + iac_code_session_id=session_id, + permission_wait_cwd=str(workspace), + permission_wait_backup_service=backup, + wait_for_response=False, + ) + + with pytest.raises(PermissionIdentityValidationError) as raised: + await registry.answer( + PermissionResponse( + task_id="task-1", + context_id="ctx-1", + request_task_id="task-1", + input_id=pending.input_id, + tool_use_id="tool-write", + decision="allow_once", + ) + ) + + assert raised.value.retryable is True + assert future.done() is False + assert store.load(str(pending.boundary_id))["decision"]["status"] == "none" + await registry.complete(pending) + future.cancel() + + +@pytest.mark.asyncio +async def test_live_deny_does_not_revalidate_cloud_identity(tmp_path, monkeypatch) -> None: + workspace, session_id, store, registry = _durable_permission_fixture(tmp_path, monkeypatch) + queue = FakeEventQueue() + backup = _ObservedBoundaryBackup(queue, store) + future = pending_future() + pending = await publish_interactive_permission_boundary( + queue, + permission_event=PermissionRequestEvent( + tool_name="aliyun_api", + tool_input={"product": "ros", "action": "CreateStack"}, + tool_use_id="tool-write", + response_future=future, + continuation_frame=_permission_frame("tool-write"), + ), + permission_input_registry=registry, + task_id="task-1", + context_id="ctx-1", + iac_code_session_id=session_id, + permission_wait_cwd=str(workspace), + permission_wait_backup_service=backup, + wait_for_response=False, + ) + + async def unexpected_identity_call(**_kwargs): + raise AssertionError("deny must not revalidate cloud identity") + + monkeypatch.setattr("iac_code.a2a.input_required.resolve_permission_execution_identity", unexpected_identity_call) + approved = await registry.answer( + PermissionResponse( + task_id="task-1", + context_id="ctx-1", + request_task_id="task-1", + input_id=pending.input_id, + tool_use_id="tool-write", + decision="deny", + ) + ) + + assert approved is False + assert future.result() is False + await registry.complete(pending) + + @pytest.mark.asyncio async def test_failed_critical_permission_backup_is_not_visible_or_recoverable(tmp_path, monkeypatch) -> None: workspace, session_id, store, registry = _durable_permission_fixture(tmp_path, monkeypatch) @@ -707,7 +866,7 @@ async def test_uncommitted_shared_permission_backup_is_not_visible_or_recoverabl continuation_frame=_permission_frame("tool-write"), ) - with pytest.raises(SessionBackupBlocked, match="did not reach the shared target"): + with pytest.raises(SessionBackupBlocked, match="did not reach a durable target"): await publish_interactive_permission_boundary( queue, permission_event=event, @@ -725,6 +884,37 @@ async def test_uncommitted_shared_permission_backup_is_not_visible_or_recoverabl assert future.result() is False +@pytest.mark.asyncio +async def test_staged_permission_backup_is_visible_without_shared_commit(tmp_path, monkeypatch) -> None: + workspace, session_id, store, registry = _durable_permission_fixture(tmp_path, monkeypatch) + queue = FakeEventQueue() + backup = _ObservedBoundaryBackup(queue, store, shared_committed=False, staged_committed=True) + future = pending_future() + + pending = await publish_interactive_permission_boundary( + queue, + permission_event=PermissionRequestEvent( + tool_name="aliyun_api", + tool_input={"product": "ros", "action": "CreateStack"}, + tool_use_id="tool-write", + response_future=future, + continuation_frame=_permission_frame("tool-write"), + ), + permission_input_registry=registry, + task_id="task-1", + context_id="ctx-1", + iac_code_session_id=session_id, + permission_wait_cwd=str(workspace), + permission_wait_backup_service=backup, + wait_for_response=False, + ) + + assert len(queue.events) == 1 + assert store.load(str(pending.boundary_id))["phase"] == "WAITING" + await registry.complete(pending) + future.cancel() + + @pytest.mark.parametrize("resolution", ["auto_approve", "resolver_allow", "resolver_deny"]) @pytest.mark.asyncio async def test_a2a_internal_permission_resolution_creates_no_checkpoint_or_critical_backup( diff --git a/tests/a2a/test_executor.py b/tests/a2a/test_executor.py index c2949f69..283c1f3a 100644 --- a/tests/a2a/test_executor.py +++ b/tests/a2a/test_executor.py @@ -7,14 +7,14 @@ from types import SimpleNamespace import pytest -from a2a.types import Task, TaskStatusUpdateEvent +from a2a.types import Task, TaskState, TaskStatusUpdateEvent from a2a.utils.errors import InvalidParamsError from google.protobuf.json_format import MessageToDict from iac_code.a2a.backup import backup_session_async from iac_code.a2a.executor import IacCodeA2AExecutor, _normal_handoff_has_backup_ack from iac_code.a2a.exposure import A2AExposureType -from iac_code.a2a.input_required import PermissionResponse +from iac_code.a2a.input_required import PermissionIdentityValidationError, PermissionResponse from iac_code.a2a.metrics import NoOpA2AMetrics from iac_code.a2a.persistence import A2AContextSnapshot, A2APersistenceStore, A2ATaskSnapshot from iac_code.a2a.pipeline_executor import recoverable_task_id_from_sidecar @@ -34,7 +34,7 @@ ScopedMCPServerConfig, ) from iac_code.pipeline.engine.user_input import PipelineUserInput -from iac_code.services.permission_wait import RecoveredPermissionAuditBoundary +from iac_code.services.permission_wait import PermissionExecutionIdentity, RecoveredPermissionAuditBoundary from iac_code.services.session_backup import ( BACKUP_STATE_FILENAME, BackupReason, @@ -4690,6 +4690,52 @@ async def resume(_context, _queue, *, response): assert calls == ["lookup", "answer", "complete", "recover"] +@pytest.mark.asyncio +async def test_identity_lookup_failure_keeps_live_permission_pending(monkeypatch: pytest.MonkeyPatch) -> None: + store = A2ATaskStore(metrics=NoOpA2AMetrics()) + executor = IacCodeA2AExecutor(task_store=store, model="qwen3.6-plus") + response = PermissionResponse( + task_id="task-1", + context_id="ctx-1", + request_task_id="task-1", + input_id="input-1", + tool_use_id="tool-1", + decision="allow_once", + ) + pending = SimpleNamespace() + published: list[dict] = [] + + async def pending_for_response(_response): + return pending + + async def answer(_response): + raise PermissionIdentityValidationError("InternalError", retryable=True) + + async def publish(_queue, **kwargs): + published.append(kwargs) + + monkeypatch.setattr("iac_code.a2a.executor.parse_permission_response", lambda _message: response) + monkeypatch.setattr(executor._permission_input_registry, "pending_for_response", pending_for_response) + monkeypatch.setattr(executor._permission_input_registry, "answer", answer) + monkeypatch.setattr(executor, "_publish_status", publish) + + await executor._execute( + FakeRequestContext(task_id="task-1", context_id="ctx-1"), + FakeEventQueue(), + context_id="ctx-1", + ) + + assert len(published) == 1 + assert published[0]["state"] == TaskState.TASK_STATE_INPUT_REQUIRED + error = published[0]["metadata"]["iac_code"]["permissionIdentityError"] + assert error == { + "code": "InternalError", + "retryable": True, + "inputId": "input-1", + "toolUseId": "tool-1", + } + + @pytest.mark.asyncio async def test_normal_persisted_permission_recovery_publishes_final_and_terminal_state( monkeypatch: pytest.MonkeyPatch, @@ -4902,7 +4948,9 @@ async def get_context_record(_context_id): "phase": "SUSPENDED", "permissionClass": "normal", "decision": {"status": "none", "value": None}, + "identitySchemaVersion": 1, "principalRef": "client-principal", + "principalKind": "ram_role", "region": "cn-beijing", } @@ -4936,15 +4984,15 @@ async def rebuild(**_kwargs): seen_access_key_ids: list[str | None] = [] - def identity(**_kwargs): + async def identity(**_kwargs): from iac_code.services.providers.aliyun import AliyunCredentials credential = AliyunCredentials.load() seen_access_key_ids.append(credential.access_key_id if credential is not None else None) - return "client-principal", "cn-beijing" + return PermissionExecutionIdentity("client-principal", "ram_role", "cn-beijing") monkeypatch.setattr(executor, "_rebuild_normal_permission_audit_event", rebuild) - monkeypatch.setattr("iac_code.a2a.executor.permission_execution_identity", identity) + monkeypatch.setattr("iac_code.a2a.executor.resolve_permission_execution_identity", identity) monkeypatch.setattr( "iac_code.services.providers.aliyun.AliyunCredentials._load_from_iac_code_config", lambda: None, @@ -4955,7 +5003,7 @@ def identity(**_kwargs): request_task_id="task-1", input_id="input-1", tool_use_id="tool-1", - decision="deny", + decision="allow_once", ) context = FakeRequestContext( task_id="task-1", diff --git a/tests/a2a/test_input_required.py b/tests/a2a/test_input_required.py index 98261ca3..06286e39 100644 --- a/tests/a2a/test_input_required.py +++ b/tests/a2a/test_input_required.py @@ -27,6 +27,7 @@ from iac_code.a2a.runtime_overrides import a2a_request_context from iac_code.a2a.task_store import A2ATaskStore from iac_code.services.permission_wait import ( + PermissionExecutionIdentity, PermissionWaitCheckpointStore, PermissionWaitCoordinator, PermissionWaitPolicy, @@ -39,6 +40,14 @@ from .fakes import FakeEventQueue, pending_future +@pytest.fixture(autouse=True) +def _stable_permission_identity(monkeypatch): + async def resolve(**_kwargs): + return PermissionExecutionIdentity(None, None, None) + + monkeypatch.setattr("iac_code.a2a.input_required.resolve_permission_execution_identity", resolve) + + def _permission_message( *, decision: str = "allow_once", diff --git a/tests/services/test_aliyun_identity.py b/tests/services/test_aliyun_identity.py new file mode 100644 index 00000000..9e90d76c --- /dev/null +++ b/tests/services/test_aliyun_identity.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import pytest + +from iac_code.services.providers.aliyun import AliyunCredential +from iac_code.services.providers.aliyun_identity import ( + AliyunCallerIdentityResolver, + AliyunCallerIdentityUnavailableError, +) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("response", "kind", "subject_id"), + [ + ({"IdentityType": "Account", "AccountId": "1001"}, "account", "1001"), + ( + {"IdentityType": "RAMUser", "AccountId": "1001", "UserId": "2002"}, + "ram_user", + "2002", + ), + ( + { + "IdentityType": "AssumedRoleUser", + "AccountId": "1001", + "RoleId": "3003", + "PrincipalId": "3003:rotating-session-name", + }, + "ram_role", + "3003", + ), + ], +) +async def test_resolver_uses_stable_caller_fields(response, kind, subject_id) -> None: + async def request(_credential, _region_id): + return response + + identity = await AliyunCallerIdentityResolver(request=request).resolve(AliyunCredential(), "cn-hangzhou") + + assert identity.kind == kind + assert identity.account_id == "1001" + assert identity.subject_id == subject_id + assert "rotating-session-name" not in identity.canonical_value().values() + + +@pytest.mark.asyncio +async def test_resolver_retries_transient_failure_twice() -> None: + attempts = 0 + delays: list[float] = [] + + class InternalError(Exception): + status_code = 500 + code = "InternalError" + + async def retrying_request(_credential, _region_id): + nonlocal attempts + attempts += 1 + if attempts < 3: + raise InternalError() + return {"IdentityType": "RAMUser", "AccountId": "1001", "UserId": "2002"} + + async def sleep(delay): + delays.append(delay) + + attempts = 0 + identity = await AliyunCallerIdentityResolver(request=retrying_request, sleep=sleep).resolve( + AliyunCredential(), + "cn-hangzhou", + ) + + assert identity.subject_id == "2002" + assert attempts == 3 + assert delays == [0.1, 0.3] + + +@pytest.mark.asyncio +async def test_resolver_does_not_retry_expired_token() -> None: + attempts = 0 + + class ExpiredTokenError(Exception): + status_code = 400 + code = "InvalidSecurityToken.Expired" + + async def request(_credential, _region_id): + nonlocal attempts + attempts += 1 + raise ExpiredTokenError() + + with pytest.raises(AliyunCallerIdentityUnavailableError) as raised: + await AliyunCallerIdentityResolver(request=request).resolve(AliyunCredential(), "cn-hangzhou") + + assert raised.value.reason == "InvalidSecurityToken.Expired" + assert raised.value.retryable is False + assert attempts == 1 + + +@pytest.mark.asyncio +async def test_resolver_rejects_incomplete_identity_without_retry() -> None: + attempts = 0 + + async def request(_credential, _region_id): + nonlocal attempts + attempts += 1 + return {"IdentityType": "AssumedRoleUser", "AccountId": "1001"} + + with pytest.raises(AliyunCallerIdentityUnavailableError, match="caller_identity_response_invalid"): + await AliyunCallerIdentityResolver(request=request).resolve(AliyunCredential(), "cn-hangzhou") + + assert attempts == 1 diff --git a/tests/services/test_permission_wait.py b/tests/services/test_permission_wait.py index adb4fb66..26e28633 100644 --- a/tests/services/test_permission_wait.py +++ b/tests/services/test_permission_wait.py @@ -18,10 +18,13 @@ canonicalize_permission_continuation_frame, format_utc, permission_execution_identity, + permission_execution_identity_cache_scope, recover_permission_audit_boundary, + resolve_permission_execution_identity, utc_now, ) from iac_code.services.providers.aliyun import AliyunCredential, AliyunCredentials +from iac_code.services.providers.aliyun_identity import AliyunCallerIdentityResolver from iac_code.services.session_layout import SessionPaths from iac_code.services.session_storage import SessionStorage from iac_code.types.stream_events import PermissionWaitOutcome @@ -482,6 +485,95 @@ def test_local_execution_identity_does_not_depend_on_cloud_credentials(monkeypat assert permission_execution_identity(tool_name="bash", tool_input={"cmd": "pwd"}) == (None, None) +@pytest.mark.asyncio +async def test_stable_execution_identity_ignores_rotating_sts_session_fields(monkeypatch) -> None: + current = AliyunCredential( + mode="StsToken", + access_key_id="sts-first", + access_key_secret="secret-first", + sts_token="token-first", + region_id="cn-hangzhou", + ) + monkeypatch.setattr(AliyunCredentials, "load", staticmethod(lambda: current)) + calls = 0 + + async def request(_credential, _region_id): + nonlocal calls + calls += 1 + return { + "IdentityType": "AssumedRoleUser", + "AccountId": "1001", + "RoleId": "3003", + "PrincipalId": "3003:session-{}".format(calls), + } + + resolver = AliyunCallerIdentityResolver(request=request) + first = await resolve_permission_execution_identity( + tool_name="aliyun_api", + tool_input={"region_id": "cn-hangzhou"}, + resolver=resolver, + ) + current = AliyunCredential( + mode="StsToken", + access_key_id="sts-second", + access_key_secret="secret-second", + sts_token="token-second", + region_id="cn-hangzhou", + ) + second = await resolve_permission_execution_identity( + tool_name="aliyun_api", + tool_input={"region_id": "cn-hangzhou"}, + resolver=resolver, + ) + + assert first == second + assert first.principal_kind == "ram_role" + assert "session-" not in str(first) + assert calls == 2 + + +@pytest.mark.asyncio +async def test_execution_identity_is_cached_once_per_a2a_request(monkeypatch) -> None: + credential_loads = 0 + + def load_credential(): + nonlocal credential_loads + credential_loads += 1 + return AliyunCredential( + mode="StsToken", + access_key_id="sts-{}".format(credential_loads), + sts_token="token-{}".format(credential_loads), + ) + + monkeypatch.setattr(AliyunCredentials, "load", staticmethod(load_credential)) + calls = 0 + + async def request(_credential, _region_id): + nonlocal calls + calls += 1 + return {"IdentityType": "RAMUser", "AccountId": "1001", "UserId": "2002"} + + resolver = AliyunCallerIdentityResolver(request=request) + with permission_execution_identity_cache_scope(): + first = await resolve_permission_execution_identity( + tool_name="aliyun_api", + tool_input={"region_id": "cn-hangzhou"}, + resolver=resolver, + ) + second = await resolve_permission_execution_identity( + tool_name="ros_stack", + tool_input={"region_id": "cn-hangzhou"}, + resolver=resolver, + ) + assert permission_execution_identity( + tool_name="aliyun_api", + tool_input={"region_id": "cn-hangzhou"}, + ) == (first.principal_ref, first.region) + + assert first == second + assert calls == 1 + + def test_checkpoint_normalizes_orphan_and_compacts_receipt(tmp_path) -> None: store = _store(tmp_path) record = _record(store, PermissionWaitPolicy())