[codex] Add Aliyun API permission audit

src/iac_code/a2a/events.py

@@ -25,11 +25,19 @@
 from iac_code.a2a.exposure import A2AExposureType, normalize_a2a_exposure_types
 from iac_code.a2a.metadata_redaction import A2AMetadataEchoRedactor
 from iac_code.i18n import _
+from iac_code.services.permissions.audit import (
+    build_input_summary,
+    build_redacted_tool_input,
+    emit_auto_permission_audit,
+    emit_permission_boundary_audit,
+    is_aliyun_api_non_read_only_permission_event,
+)
 from iac_code.types.stream_events import (
     ErrorEvent,
     MCPProgressEvent,
     MessageEndEvent,
     PermissionRequestEvent,
+    SubPipelineStreamEvent,
     TextDeltaEvent,
     ThinkingDeltaEvent,
     ToolInputDeltaEvent,
@@ -77,6 +85,22 @@ def _sanitize_trace_input(value: Any) -> Any:
     return _METADATA_REDACTOR.redact(_truncate(value))
 
 
+def _public_tool_input_metadata(tool_name: str, tool_input: dict[str, Any]) -> dict[str, Any]:
+    return {"inputSummary": build_input_summary(tool_name, tool_input)}
+
+
+def _public_permission_input_metadata(tool_name: str, tool_input: dict[str, Any]) -> dict[str, Any]:
+    if tool_name == "aliyun_api":
+        return {"inputSummary": build_input_summary(tool_name, tool_input)}
+    return {"toolInput": build_redacted_tool_input(tool_input)}
+
+
+def _permission_request_event(event: Any) -> PermissionRequestEvent | None:
+    while isinstance(event, SubPipelineStreamEvent):
+        event = event.inner
+    return event if isinstance(event, PermissionRequestEvent) else None
+
+
 def make_text_part(text: str) -> Part:
     return Part(text=text)
 
@@ -285,7 +309,7 @@ async def publish_stream_event(
                     "tool": {
                         "status": "input_delta",
                         "toolUseId": event.tool_use_id,
-                        "partialJson": _truncate(event.partial_json),
+                        "partialJsonLength": len(event.partial_json),
                     }
                 }
             },
@@ -307,7 +331,7 @@ async def publish_stream_event(
                         "status": "input_complete",
                         "toolUseId": event.tool_use_id,
                         "name": event.name,
-                        "input": _sanitize_trace_input(event.input),
+                        **_public_tool_input_metadata(event.name, event.input),
                     }
                 }
             },
@@ -369,13 +393,23 @@ async def publish_stream_event(
         )
         return None
 
-    if isinstance(event, PermissionRequestEvent):
+    permission_event = _permission_request_event(event)
+    if permission_event is not None:
         approved = auto_approve_permissions
         if permission_resolver is not None:
-            decision = permission_resolver(event)
+            decision = permission_resolver(permission_event)
             approved = bool(await decision) if inspect.isawaitable(decision) else bool(decision)
-        if event.response_future is not None and not event.response_future.done():
-            event.response_future.set_result(approved)
+        elif is_aliyun_api_non_read_only_permission_event(permission_event):
+            approved = False
+        if permission_event.response_future is not None and not permission_event.response_future.done():
+            audit_ok = True
+            if permission_resolver is not None:
+                audit_ok = _emit_resolver_permission_audit(permission_event, approved)
+            else:
+                audit_ok = _emit_auto_permission_audit(permission_event, approved)
+            if approved and not audit_ok:
+                approved = False
+            permission_event.response_future.set_result(approved)
         if A2AExposureType.TOOL_TRACE not in enabled_exposure_types:
             return None
         await _enqueue_status(
@@ -387,9 +421,9 @@ async def publish_stream_event(
                 "iac_code": {
                     "permission": {
                         "autoApproved": approved,
-                        "toolName": event.tool_name,
-                        "toolUseId": event.tool_use_id,
-                        "toolInput": _sanitize_trace_input(event.tool_input),
+                        "toolName": permission_event.tool_name,
+                        "toolUseId": permission_event.tool_use_id,
+                        **_public_permission_input_metadata(permission_event.tool_name, permission_event.tool_input),
                     }
                 }
             },
@@ -440,3 +474,50 @@ async def publish_stream_event(
 
     logger.debug("Skipping unmapped A2A stream event: %s", type(event).__name__)
     return None
+
+
+def _emit_auto_permission_audit(
+    request: PermissionRequestEvent,
+    approved: bool,
+    *,
+    persistence_failure: bool = False,
+) -> bool:
+    if persistence_failure:
+        return emit_permission_boundary_audit(
+            request,
+            decision="deny",
+            scope="auto_deny",
+            source="a2a_auto_persistence_failure",
+            reason_type="persistence_failure",
+            reason_detail="permission metadata persistence failed",
+        )
+    source = "a2a_auto_approve" if approved else "a2a_auto_deny"
+    return emit_auto_permission_audit(
+        request,
+        decision="allow" if approved else "deny",
+        scope="auto_approve" if approved else "auto_deny",
+        source=source,
+    )
+
+
+def _emit_resolver_permission_audit(
+    request: PermissionRequestEvent,
+    approved: bool,
+    *,
+    persistence_failure: bool = False,
+) -> bool:
+    source = "a2a_resolver"
+    reason_type = "a2a_resolver"
+    reason_detail = "allow" if approved else "deny"
+    if persistence_failure:
+        source = "a2a_resolver_persistence_failure"
+        reason_type = "persistence_failure"
+        reason_detail = "permission metadata persistence failed"
+    return emit_permission_boundary_audit(
+        request,
+        decision="allow" if approved else "deny",
+        scope="a2a_resolver",
+        source=source,
+        reason_type=reason_type,
+        reason_detail=reason_detail,
+    )

src/iac_code/a2a/pipeline_events.py

@@ -11,6 +11,7 @@
 from iac_code.a2a.artifacts import UnsafeArtifactNameError, artifact_filename_from_path, sanitize_public_artifact_text
 from iac_code.a2a.events import _tool_result_metadata, _truncate
 from iac_code.pipeline.engine.events import PipelineEvent, PipelineEventType
+from iac_code.services.permissions.audit import build_input_summary, build_redacted_tool_input, fingerprint_text
 from iac_code.types.stream_events import (
     AskUserQuestionEvent,
     CandidateDetailEvent,
@@ -75,10 +76,27 @@
 _NESTED_DATA_KEY_ALIASES = {
     "error_id": "errorId",
 }
-_PERMISSION_METADATA_MAX_CHARS = 4000
-_PERMISSION_METADATA_MAX_DEPTH = 32
 _PERMISSION_SUMMARY_MAX_FIELDS = 20
 _PERMISSION_SUMMARY_MAX_CHARS = 256
+_SAFE_PERMISSION_SUMMARY_FIELDS = {
+    "action",
+    "body",
+    "cmd",
+    "command",
+    "content",
+    "cwd",
+    "input",
+    "method",
+    "params",
+    "path",
+    "pathname",
+    "product",
+    "query",
+    "region",
+    "region_id",
+    "timeout",
+    "url",
+}
 _SECRET_KEY_FRAGMENTS = (
     "api_key",
     "apikey",
@@ -96,6 +114,7 @@
     "credential",
     "authorization",
     "private_key",
+    "signature",
 )
 _STACK_TOOL_ACTIONS = {"CreateStack", "UpdateStack", "ContinueCreateStack", "DeleteStack"}
 _STACK_CLEAR_ACTIONS = {"DeleteStack"}
@@ -501,8 +520,9 @@ def _completion_artifact_events(self, completed: dict[str, Any]) -> list[dict[st
         return events
 
     def _translate_sub_pipeline_stream_event(self, event: SubPipelineStreamEvent) -> list[dict[str, Any]]:
-        state = self._candidate_from_stream_event(event)
-        inner = event.inner
+        stream_event = _innermost_sub_pipeline_stream_event(event)
+        state = self._candidate_from_stream_event(stream_event)
+        inner = stream_event.inner
         if isinstance(inner, TextDeltaEvent):
             event_type = "text_delta"
             data = {"text": inner.text}
@@ -1181,19 +1201,28 @@ def _permission_request_metadata(event: PermissionRequestEvent) -> dict[str, Any
     return safe_permission_metadata(event)
 
 
+def _innermost_sub_pipeline_stream_event(event: SubPipelineStreamEvent) -> SubPipelineStreamEvent:
+    while isinstance(event.inner, SubPipelineStreamEvent):
+        event = event.inner
+    return event
+
+
 def safe_permission_metadata(
     event: PermissionRequestEvent,
     *,
     include_tool_input: bool = True,
 ) -> dict[str, Any]:
-    metadata = {
+    metadata: dict[str, Any] = {
         "permissionId": f"perm-{event.tool_use_id}",
         "toolName": event.tool_name,
         "toolUseId": event.tool_use_id,
         "safeSummary": _permission_safe_summary(event),
     }
     if include_tool_input:
-        metadata["toolInput"] = _redact_permission_value(event.tool_input)
+        if event.tool_name == "aliyun_api":
+            metadata["inputSummary"] = build_input_summary(event.tool_name, event.tool_input)
+        else:
+            metadata["toolInput"] = build_redacted_tool_input(event.tool_input)
     return metadata
 
 
@@ -1219,29 +1248,17 @@ def _permission_safe_summary(event: PermissionRequestEvent) -> str:
     return summary[: _PERMISSION_SUMMARY_MAX_CHARS - 3] + "..."
 
 
-def _redact_permission_value(value: Any, *, _depth: int = 0) -> Any:
-    if _depth >= _PERMISSION_METADATA_MAX_DEPTH:
-        return "[truncated-depth]"
-    if isinstance(value, str):
-        return sanitize_public_artifact_text(value)[:_PERMISSION_METADATA_MAX_CHARS]
-    if isinstance(value, dict):
-        return {
-            str(key): "[redacted]" if _is_secret_key(key) else _redact_permission_value(item, _depth=_depth + 1)
-            for key, item in value.items()
-        }
-    if isinstance(value, list):
-        return [_redact_permission_value(item, _depth=_depth + 1) for item in value]
-    return _truncate(value, _depth=_depth)
-
-
 def _is_secret_key(key: Any) -> bool:
     normalized = str(key).lower().replace("-", "_")
     compact = normalized.replace("_", "")
     return any(fragment in normalized or fragment.replace("_", "") in compact for fragment in _SECRET_KEY_FRAGMENTS)
 
 
 def _safe_permission_field_name(key: Any) -> str:
-    return "[redacted]" if _is_secret_key(key) else str(key)
+    if _is_secret_key(key):
+        return "[redacted]"
+    text = str(key)
+    return text if text in _SAFE_PERMISSION_SUMMARY_FIELDS else fingerprint_text(text)
 
 
 def _tool_result_data(event: ToolResultEvent) -> dict[str, Any]: