diff --git a/client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/client/transport/jsonrpc/JSONRPCTransport.java b/client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/client/transport/jsonrpc/JSONRPCTransport.java index d92956714..0d099ecc7 100644 --- a/client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/client/transport/jsonrpc/JSONRPCTransport.java +++ b/client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/client/transport/jsonrpc/JSONRPCTransport.java @@ -34,6 +34,7 @@ import org.a2aproject.sdk.grpc.utils.JSONRPCUtils; import org.a2aproject.sdk.grpc.utils.ProtoUtils; import org.a2aproject.sdk.jsonrpc.common.json.JsonProcessingException; +import org.a2aproject.sdk.jsonrpc.common.logging.JsonPayloadLoggingHelper; import org.a2aproject.sdk.jsonrpc.common.wrappers.A2AResponse; import org.a2aproject.sdk.jsonrpc.common.wrappers.CancelTaskResponse; import org.a2aproject.sdk.jsonrpc.common.wrappers.DeleteTaskPushNotificationConfigResponse; @@ -94,13 +95,16 @@ public JSONRPCTransport(@Nullable A2AHttpClient httpClient, @Nullable AgentCard @Override public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext context) throws A2AClientException { checkNotNullParam("request", request); + JsonPayloadLoggingHelper.logTaskRequest(SEND_MESSAGE_METHOD, request, request.message().taskId(), request.message().contextId()); PayloadAndHeaders payloadAndHeaders = applyInterceptors(SEND_MESSAGE_METHOD, ProtoUtils.ToProto.sendMessageRequest(request), agentCard, context); try { String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, SEND_MESSAGE_METHOD); SendMessageResponse response = unmarshalResponse(httpResponseBody, SEND_MESSAGE_METHOD); - return response.getResult(); + EventKind result = response.getResult(); + JsonPayloadLoggingHelper.logTaskResult(result, result != null ? extractTaskIdFromEventKind(result) : null); + return result; } catch (A2AClientException e) { throw e; } catch (IOException | InterruptedException | JsonProcessingException e) { @@ -113,11 +117,22 @@ public void sendMessageStreaming(MessageSendParams request, Consumer errorConsumer, @Nullable ClientCallContext context) throws A2AClientException { checkNotNullParam("request", request); checkNotNullParam("eventConsumer", eventConsumer); + JsonPayloadLoggingHelper.logTaskRequest(SEND_STREAMING_MESSAGE_METHOD, request, request.message().taskId(), request.message().contextId()); PayloadAndHeaders payloadAndHeaders = applyInterceptors(SEND_STREAMING_MESSAGE_METHOD, ProtoUtils.ToProto.sendMessageRequest(request), agentCard, context); final AtomicReference> ref = new AtomicReference<>(); - SSEEventListener sseEventListener = new SSEEventListener(eventConsumer, errorConsumer); + Consumer loggingEventConsumer = event -> { + String taskId = event instanceof Task t ? t.id() : event instanceof org.a2aproject.sdk.spec.Message m ? m.taskId() + : event instanceof org.a2aproject.sdk.spec.TaskStatusUpdateEvent tue ? tue.taskId() + : event instanceof org.a2aproject.sdk.spec.TaskArtifactUpdateEvent taue ? taue.taskId() : null; + JsonPayloadLoggingHelper.logStreamingEvent(event, taskId, event != null ? event.getClass().getSimpleName() : null); + eventConsumer.accept(event); + }; + Consumer loggingErrorConsumer = error -> { + if (errorConsumer != null) errorConsumer.accept(error); + }; + SSEEventListener sseEventListener = new SSEEventListener(loggingEventConsumer, loggingErrorConsumer); try { A2AHttpClient.PostBuilder builder = createPostBuilder(agentInterface.url(), payloadAndHeaders, SEND_STREAMING_MESSAGE_METHOD); @@ -140,13 +155,16 @@ public void sendMessageStreaming(MessageSendParams request, Consumer e checkNotNullParam("request", request); checkNotNullParam("eventConsumer", eventConsumer); checkNotNullParam("errorConsumer", errorConsumer); + JsonPayloadLoggingHelper.logPushSubscription(SUBSCRIBE_TO_TASK_METHOD, request, request.id(), null); PayloadAndHeaders payloadAndHeaders = applyInterceptors(SUBSCRIBE_TO_TASK_METHOD, ProtoUtils.ToProto.subscribeToTaskRequest(request), agentCard, context); AtomicReference> ref = new AtomicReference<>(); - SSEEventListener sseEventListener = new SSEEventListener(eventConsumer, errorConsumer); + Consumer loggingEventConsumer = event -> { + String taskId = event instanceof Task t ? t.id() : event instanceof org.a2aproject.sdk.spec.Message m ? m.taskId() + : event instanceof org.a2aproject.sdk.spec.TaskStatusUpdateEvent tue ? tue.taskId() + : event instanceof org.a2aproject.sdk.spec.TaskArtifactUpdateEvent taue ? taue.taskId() : null; + JsonPayloadLoggingHelper.logStreamingEvent(event, taskId, event != null ? event.getClass().getSimpleName() : null); + eventConsumer.accept(event); + }; + Consumer loggingErrorConsumer = error -> errorConsumer.accept(error); + SSEEventListener sseEventListener = new SSEEventListener(loggingEventConsumer, loggingErrorConsumer); try { A2AHttpClient.PostBuilder builder = createPostBuilder(agentInterface.url(), payloadAndHeaders, SUBSCRIBE_TO_TASK_METHOD); @@ -290,13 +334,16 @@ public void subscribeToTask(TaskIdParams request, Consumer e @Override public AgentCard getExtendedAgentCard(GetExtendedAgentCardParams params, @Nullable ClientCallContext context) throws A2AClientException { try { + JsonPayloadLoggingHelper.logNegotiation(GET_EXTENDED_AGENT_CARD_METHOD, params); PayloadAndHeaders payloadAndHeaders = applyInterceptors(GET_EXTENDED_AGENT_CARD_METHOD, ProtoUtils.ToProto.extendedAgentCard(params), agentCard, context); try { String httpResponseBody = sendPostRequest(agentInterface.url(), payloadAndHeaders, GET_EXTENDED_AGENT_CARD_METHOD); GetExtendedAgentCardResponse response = unmarshalResponse(httpResponseBody, GET_EXTENDED_AGENT_CARD_METHOD); - return response.getResult(); + AgentCard result = response.getResult(); + JsonPayloadLoggingHelper.logNegotiation(GET_EXTENDED_AGENT_CARD_METHOD + "/result", result); + return result; } catch (IOException | InterruptedException | JsonProcessingException e) { throw new A2AClientException("Failed to get authenticated extended agent card: " + e, e); } @@ -378,4 +425,13 @@ private > T unmarshalResponse(String response, String m private @Nullable Map getHttpHeaders(@Nullable ClientCallContext context) { return context != null ? context.getHeaders() : null; } + + private @Nullable String extractTaskIdFromEventKind(EventKind eventKind) { + if (eventKind instanceof Task task) { + return task.id(); + } else if (eventKind instanceof org.a2aproject.sdk.spec.Message msg) { + return msg.taskId(); + } + return null; + } } \ No newline at end of file diff --git a/common/pom.xml b/common/pom.xml index c1b3b4851..0dfa96b50 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -21,11 +21,34 @@ org.slf4j slf4j-api + + org.jspecify + jspecify + + + com.google.code.gson + gson + org.junit.jupiter junit-jupiter-api test + + org.junit.jupiter + junit-jupiter-params + test + + + ch.qos.logback + logback-classic + test + + + org.mockito + mockito-core + test + \ No newline at end of file diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/A2AEventType.java b/common/src/main/java/org/a2aproject/sdk/common/logging/A2AEventType.java new file mode 100644 index 000000000..e4db00f2c --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/A2AEventType.java @@ -0,0 +1,75 @@ +package org.a2aproject.sdk.common.logging; + +/** + * Enumeration of A2A protocol payload event types for structured logging. + *

+ * Each event type corresponds to a category of protocol message that can be + * logged with structured fields for observability and debugging. + *

+ */ +public enum A2AEventType { + /** + * Task request: message/send, message/stream, getTask, listTasks, cancelTask, etc. + */ + TASK_REQUEST("task.request"), + + /** + * Task state change: TaskStatusUpdateEvent and status transitions. + */ + STATE_CHANGE("task.state_change"), + + /** + * Result artifact: TaskArtifactUpdateEvent and artifact payloads. + */ + ARTIFACT("task.artifact"), + + /** + * Task result: final Task snapshot returned to client. + */ + TASK_RESULT("task.result"), + + /** + * Negotiation payload: extension negotiation, version negotiation, AgentCard capabilities. + */ + NEGOTIATION("negotiation"), + + /** + * Authorization decision: allow/deny from TaskAuthorizationProvider. + */ + AUTHORIZATION("authorization"), + + /** + * Notification subscription: push config create/get/list/delete, subscribeToTask. + */ + NOTIFICATION_SUBSCRIPTION("notification.subscription"), + + /** + * Notification delivery: push webhook dispatch to external URL. + */ + NOTIFICATION_DELIVERY("notification.delivery"), + + /** + * Streaming event: per-event SSE streaming payload. + */ + STREAMING_EVENT("streaming.event"), + + /** + * Generic payload for any other structured log. + */ + GENERIC("generic"); + + private final String value; + + A2AEventType(String value) { + this.value = value; + } + + public String value() { + return value; + } + + @Override + public String toString() { + return value; + } +} diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/A2APayloadLoggingConfig.java b/common/src/main/java/org/a2aproject/sdk/common/logging/A2APayloadLoggingConfig.java new file mode 100644 index 000000000..d21e69d98 --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/A2APayloadLoggingConfig.java @@ -0,0 +1,174 @@ +package org.a2aproject.sdk.common.logging; + +import java.util.Set; + +import org.jspecify.annotations.Nullable; + +/** + * Configuration for structured payload logging. + *

+ * Supports both system properties and programmatic configuration via builder. + *

+ *

System Properties

+ *
    + *
  • {@code a2a.logging.payload.enabled} (boolean, default false) — enable payload logging
    • + *
  • {@code a2a.logging.payload.max-size} (int, default 4096) — max payload size before truncation
    • + *
  • {@code a2a.logging.payload.masking.enabled} (boolean, default true) — enable sensitive masking
    • + *
  • {@code a2a.logging.payload.masking.strategy} (class name) — custom masking strategy
    • + *
+ */ +public final class A2APayloadLoggingConfig { + + public static final String ENABLED_PROPERTY = "a2a.logging.payload.enabled"; + public static final String MAX_SIZE_PROPERTY = "a2a.logging.payload.max-size"; + public static final String MASKING_ENABLED_PROPERTY = "a2a.logging.payload.masking.enabled"; + public static final String MASKING_STRATEGY_PROPERTY = "a2a.logging.payload.masking.strategy"; + public static final String MASKING_SENSITIVE_KEYS_PROPERTY = "a2a.logging.payload.masking.sensitive-keys"; + public static final int DEFAULT_MAX_SIZE = 4096; + + private final boolean enabled; + private final int maxSize; + private final boolean maskingEnabled; + private final @Nullable PayloadMaskingStrategy maskingStrategy; + private final @Nullable Set sensitiveKeys; + + private A2APayloadLoggingConfig(Builder builder) { + this.enabled = builder.enabled; + this.maxSize = builder.maxSize; + this.maskingEnabled = builder.maskingEnabled; + this.maskingStrategy = builder.maskingStrategy; + this.sensitiveKeys = builder.sensitiveKeys; + } + + public boolean isEnabled() { + return enabled; + } + + public int getMaxSize() { + return maxSize; + } + + public boolean isMaskingEnabled() { + return maskingEnabled; + } + + public @Nullable PayloadMaskingStrategy getMaskingStrategy() { + return maskingStrategy; + } + + public @Nullable Set getSensitiveKeys() { + return sensitiveKeys; + } + + /** + * Creates config from system properties. + */ + public static A2APayloadLoggingConfig fromSystemProperties() { + Builder builder = new Builder(); + String enabledProp = System.getProperty(ENABLED_PROPERTY); + if (enabledProp != null) { + builder.enabled(Boolean.parseBoolean(enabledProp)); + } + String maxSizeProp = System.getProperty(MAX_SIZE_PROPERTY); + if (maxSizeProp != null) { + try { + builder.maxSize(Integer.parseInt(maxSizeProp.trim())); + } catch (NumberFormatException ignored) { + // use default + } + } + String maskingEnabledProp = System.getProperty(MASKING_ENABLED_PROPERTY); + if (maskingEnabledProp != null) { + builder.maskingEnabled(Boolean.parseBoolean(maskingEnabledProp)); + } + // Strategy loaded separately by logger via ServiceLoader / reflection; + // we don't instantiate here to avoid class loading issues in fromSystemProperties + String strategyClass = System.getProperty(MASKING_STRATEGY_PROPERTY); + if (strategyClass != null && !strategyClass.isBlank()) { + try { + Class clazz = Class.forName(strategyClass.trim()); + Object instance = clazz.getDeclaredConstructor().newInstance(); + if (instance instanceof PayloadMaskingStrategy pms) { + builder.maskingStrategy(pms); + } + } catch (Exception ignored) { + } + } + String sensitiveKeysProp = System.getProperty(MASKING_SENSITIVE_KEYS_PROPERTY); + if (sensitiveKeysProp != null && !sensitiveKeysProp.isBlank()) { + Set keys = Set.of(sensitiveKeysProp.split(",")); + // Trim each key + Set trimmed = new java.util.HashSet<>(); + for (String k : keys) { + if (k != null && !k.isBlank()) { + trimmed.add(k.trim()); + } + } + if (!trimmed.isEmpty()) { + builder.sensitiveKeys(trimmed); + } + } + return builder.build(); + } + + public Builder toBuilder() { + Builder b = new Builder(); + b.enabled = this.enabled; + b.maxSize = this.maxSize; + b.maskingEnabled = this.maskingEnabled; + b.maskingStrategy = this.maskingStrategy; + b.sensitiveKeys = this.sensitiveKeys; + return b; + } + + public static Builder builder() { + return new Builder(); + } + + public static final class Builder { + private boolean enabled = false; + private int maxSize = DEFAULT_MAX_SIZE; + private boolean maskingEnabled = true; + private @Nullable PayloadMaskingStrategy maskingStrategy = null; + private @Nullable Set sensitiveKeys = null; + + public Builder enabled(boolean enabled) { + this.enabled = enabled; + return this; + } + + public Builder maxSize(int maxSize) { + this.maxSize = maxSize; + return this; + } + + public Builder maskingEnabled(boolean maskingEnabled) { + this.maskingEnabled = maskingEnabled; + return this; + } + + public Builder maskingStrategy(@Nullable PayloadMaskingStrategy maskingStrategy) { + this.maskingStrategy = maskingStrategy; + return this; + } + + public Builder sensitiveKeys(@Nullable Set sensitiveKeys) { + this.sensitiveKeys = sensitiveKeys; + return this; + } + + public A2APayloadLoggingConfig build() { + return new A2APayloadLoggingConfig(this); + } + } + + @Override + public String toString() { + return "A2APayloadLoggingConfig{" + + "enabled=" + enabled + + ", maxSize=" + maxSize + + ", maskingEnabled=" + maskingEnabled + + ", maskingStrategy=" + (maskingStrategy != null ? maskingStrategy.getClass().getName() : "default") + + '}'; + } +} diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/DefaultPayloadMaskingStrategy.java b/common/src/main/java/org/a2aproject/sdk/common/logging/DefaultPayloadMaskingStrategy.java new file mode 100644 index 000000000..4433e70fe --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/DefaultPayloadMaskingStrategy.java @@ -0,0 +1,204 @@ +package org.a2aproject.sdk.common.logging; + +import java.util.HashSet; +import java.util.Set; + +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import com.google.gson.JsonSyntaxException; + +/** + * Default masking strategy that redacts sensitive fields by key name. + *

+ * The strategy parses the payload as JSON and recursively traverses the structure, + * masking values whose keys contain sensitive substrings (case-insensitive). + * If parsing fails, it falls back to regex-based masking for robustness. + *

+ *

+ * Default sensitive keys (case-insensitive substring match): + *

    + *
  • token, credentials, password, secret, apikey, api_key
    • + *
  • authorization, authentication, privatekey, private_key
    • + *
  • accesstoken, access_token, refreshtoken, refresh_token
    • + *
  • clientsecret, client_secret
    • + *
  • url (push notification URLs are sensitive; can be disabled via builder)
    • + *
+ *

+ */ +public class DefaultPayloadMaskingStrategy implements PayloadMaskingStrategy { + + private static final String MASKED_VALUE = "***"; + + /** + * Default sensitive key substrings (lowercase). + */ + public static final Set DEFAULT_SENSITIVE_KEYS = Set.of( + "token", + "credentials", + "password", + "secret", + "apikey", + "api_key", + "privatekey", + "private_key", + "authorization", + "authentication", + "accesstoken", + "access_token", + "refreshtoken", + "refresh_token", + "clientsecret", + "client_secret", + "url" + ); + + private final Set sensitiveKeys; + + /** + * Creates with default sensitive keys. + */ + public DefaultPayloadMaskingStrategy() { + this.sensitiveKeys = normalizeKeys(DEFAULT_SENSITIVE_KEYS); + } + + /** + * Creates with custom sensitive keys (case-insensitive substring match). + * + * @param sensitiveKeys set of sensitive key substrings (case-insensitive) + */ + public DefaultPayloadMaskingStrategy(Set sensitiveKeys) { + if (sensitiveKeys == null || sensitiveKeys.isEmpty()) { + this.sensitiveKeys = normalizeKeys(DEFAULT_SENSITIVE_KEYS); + } else { + this.sensitiveKeys = normalizeKeys(sensitiveKeys); + } + } + + private static Set normalizeKeys(Set keys) { + Set normalized = new HashSet<>(); + for (String k : keys) { + if (k != null) { + normalized.add(k.toLowerCase()); + } + } + return normalized; + } + + /** + * Returns the configured sensitive keys (lowercase). + */ + public Set getSensitiveKeys() { + return Set.copyOf(sensitiveKeys); + } + + @Override + public String mask(String jsonPayload) { + if (jsonPayload == null || jsonPayload.isBlank()) { + return jsonPayload; + } + try { + JsonElement element = JsonParser.parseString(jsonPayload); + boolean masked = maskJsonElement(element, null); + // Return original if nothing was masked to avoid unnecessary serialization cost? + // But we still need to return serialized form if masked; to keep behavior consistent, + // we serialize only if masking occurred or to normalize. For simplicity, serialize only if masked. + if (masked) { + return element.toString(); + } + return jsonPayload; + } catch (JsonSyntaxException e) { + // Fallback to regex masking for non-JSON or malformed payloads + return maskWithRegex(jsonPayload); + } catch (Exception e) { + // On any unexpected error, return original to avoid log loss + return jsonPayload; + } + } + + /** + * Recursively masks JsonElement if its key is sensitive. + * + * @param element current element + * @param parentKey the key of this element in its parent object, may be null for root + * @return true if any masking occurred in this subtree + */ + private boolean maskJsonElement(JsonElement element, String parentKey) { + if (element == null || element.isJsonNull()) { + return false; + } + if (parentKey != null && isSensitiveKey(parentKey)) { + // Mask the entire value regardless of type + // For objects/arrays, we replace with masked string at parent level (handled below) + // But if this is called recursively for child, parent already masked + return false; // shouldn't happen because parent masks before recursing + } + + boolean masked = false; + if (element.isJsonObject()) { + JsonObject obj = element.getAsJsonObject(); + // Collect keys to avoid concurrent modification + Set keys = new HashSet<>(obj.keySet()); + for (String key : keys) { + JsonElement child = obj.get(key); + if (isSensitiveKey(key)) { + // Replace value with masked string + obj.addProperty(key, MASKED_VALUE); + masked = true; + } else if (child != null && (child.isJsonObject() || child.isJsonArray())) { + boolean childMasked = maskJsonElement(child, key); + if (childMasked) { + masked = true; + } + } + } + } else if (element.isJsonArray()) { + JsonArray array = element.getAsJsonArray(); + for (int i = 0; i < array.size(); i++) { + JsonElement child = array.get(i); + if (child != null && (child.isJsonObject() || child.isJsonArray())) { + boolean childMasked = maskJsonElement(child, null); + if (childMasked) { + masked = true; + } + } + } + } + return masked; + } + + private boolean isSensitiveKey(String key) { + if (key == null) { + return false; + } + String lower = key.toLowerCase(); + for (String sensitive : sensitiveKeys) { + if (lower.contains(sensitive)) { + return true; + } + } + return false; + } + + /** + * Fallback regex masking for non-JSON payloads. + * Masks patterns like "key": "value" where key is sensitive. + */ + private String maskWithRegex(String payload) { + String result = payload; + for (String sensitive : sensitiveKeys) { + // Regex: "sensitiveKey" optional spaces : optional spaces "value" or 'value' or bare token + // We use case-insensitive matching for key + // This is best-effort; avoid catastrophic backtracking + String regex = "(\"(?i)" + escapeRegex(sensitive) + "[^\"]*\"\\s*:\\s*\")(\"[^\"]*\"|'[^']*'|[^,\\}\\]]+)"; + // Replace value part with "***" preserving quotes if present + result = result.replaceAll(regex, "$1\"***\""); + } + return result; + } + + private String escapeRegex(String input) { + return input.replaceAll("([\\\\.*+?\\[^\\]$(){}=!<>|:\\-])", "\\\\$1"); + } +} diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/NoOpPayloadMaskingStrategy.java b/common/src/main/java/org/a2aproject/sdk/common/logging/NoOpPayloadMaskingStrategy.java new file mode 100644 index 000000000..6c90dd701 --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/NoOpPayloadMaskingStrategy.java @@ -0,0 +1,16 @@ +package org.a2aproject.sdk.common.logging; + +/** + * No-op masking strategy that returns the payload unchanged. + *

+ * Use this when sensitive data masking is disabled or when you want to preserve + * original payloads for debugging in secure environments. + *

+ */ +public class NoOpPayloadMaskingStrategy implements PayloadMaskingStrategy { + + @Override + public String mask(String jsonPayload) { + return jsonPayload; + } +} diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/PayloadMaskingStrategy.java b/common/src/main/java/org/a2aproject/sdk/common/logging/PayloadMaskingStrategy.java new file mode 100644 index 000000000..cd5d17013 --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/PayloadMaskingStrategy.java @@ -0,0 +1,25 @@ +package org.a2aproject.sdk.common.logging; + +/** + * Strategy for masking sensitive data in payload logs. + *

+ * Implementations can provide custom logic to redact or replace sensitive fields + * (e.g., tokens, credentials, URLs) before payloads are written to logs. + *

+ *

+ * Users can configure a custom strategy via the system property + * {@code a2a.logging.payload.masking.strategy} (fully qualified class name) + * or programmatically via {@link A2APayloadLoggingConfig.Builder}. + *

+ */ +public interface PayloadMaskingStrategy { + + /** + * Masks sensitive data in the given JSON payload string. + * + * @param jsonPayload the original JSON payload, may be null + * @return the masked payload, or original if masking not applicable. Must not return null + * if input is non-null; return original string on any error to avoid log loss. + */ + String mask(String jsonPayload); +} diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/PayloadTruncator.java b/common/src/main/java/org/a2aproject/sdk/common/logging/PayloadTruncator.java new file mode 100644 index 000000000..61d8d4272 --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/PayloadTruncator.java @@ -0,0 +1,47 @@ +package org.a2aproject.sdk.common.logging; + +/** + * Utility for truncating payloads to a configurable size with a marker. + *

+ * Truncation is based on character length (not bytes) for simplicity and + * consistency with typical string handling. The truncated marker + * {@value #TRUNCATED_MARKER} is appended when truncation occurs. + *

+ */ +public final class PayloadTruncator { + + /** + * Marker appended when payload is truncated. + */ + public static final String TRUNCATED_MARKER = " [truncated]"; + + private PayloadTruncator() { + } + + /** + * Truncates the payload if it exceeds maxSize. + * + * @param payload the payload string, may be null + * @param maxSize maximum allowed size in characters; if <=0, no truncation + * @return truncated payload with marker, or original if within limit; null if input null + */ + public static String truncate(String payload, int maxSize) { + if (payload == null) { + return null; + } + if (maxSize <= 0 || payload.length() <= maxSize) { + return payload; + } + return payload.substring(0, maxSize) + TRUNCATED_MARKER; + } + + /** + * Checks if payload was truncated. + * + * @param payload the payload to check + * @return true if payload ends with truncated marker + */ + public static boolean isTruncated(String payload) { + return payload != null && payload.endsWith(TRUNCATED_MARKER); + } +} diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/StructuredPayloadLogger.java b/common/src/main/java/org/a2aproject/sdk/common/logging/StructuredPayloadLogger.java new file mode 100644 index 000000000..dc7c677c6 --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/StructuredPayloadLogger.java @@ -0,0 +1,454 @@ +package org.a2aproject.sdk.common.logging; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; + +import org.jspecify.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Central logger for structured A2A payload logging. + *

+ * Provides DEBUG-level structured log events for key protocol messages: + * task requests, state changes, result artifacts, negotiation payloads, + * authorization decisions, notification subscriptions and deliveries. + *

+ *

+ * Features: + *

    + *
  • Configurable payload size truncation (default 4KB) with {@code [truncated]} marker
    • + *
  • Sensitive data masking with customizable strategies
    • + *
  • Trace correlation via {@code trace_id} / {@code span_id} from OpenTelemetry or MDC
    • + *
  • Structured fields via SLF4J fluent API (keyValue) when available
    • + *
+ *

+ *

+ * Logs are emitted only when: + *

    + *
  • Payload logging is enabled via {@link A2APayloadLoggingConfig} (system property {@code a2a.logging.payload.enabled=true} or programmatic config)
    • + *
  • Underlying SLF4J logger is DEBUG enabled
    • + *
+ *

+ */ +public final class StructuredPayloadLogger { + + private static final Logger LOGGER = LoggerFactory.getLogger("org.a2aproject.sdk.payload"); + + /** + * System property to enable payload logging via logger level without config. + * For backward compatibility, also checks {@code org.a2aproject.sdk.payload.enabled}. + */ + private static final String LEGACY_ENABLED_PROP = "org.a2aproject.sdk.payload.enabled"; + + private static final AtomicReference CONFIG_REF = + new AtomicReference<>(A2APayloadLoggingConfig.fromSystemProperties()); + + private static volatile PayloadMaskingStrategy maskingStrategy = createDefaultMaskingStrategy(); + + private StructuredPayloadLogger() { + } + + /** + * Returns current config (volatile read). + */ + public static A2APayloadLoggingConfig getConfig() { + return CONFIG_REF.get(); + } + + /** + * Sets config programmatically. Useful for tests or runtime reconfiguration. + * + * @param config new config + */ + public static void setConfig(A2APayloadLoggingConfig config) { + if (config == null) { + throw new IllegalArgumentException("config must not be null"); + } + CONFIG_REF.set(config); + // Refresh masking strategy if sensitive keys changed + if (config.getMaskingStrategy() != null) { + maskingStrategy = config.getMaskingStrategy(); + } else if (config.getSensitiveKeys() != null) { + maskingStrategy = new DefaultPayloadMaskingStrategy(config.getSensitiveKeys()); + } else { + maskingStrategy = createDefaultMaskingStrategy(); + } + } + + /** + * Resets to system properties (useful for tests). + */ + public static void resetToSystemProperties() { + CONFIG_REF.set(A2APayloadLoggingConfig.fromSystemProperties()); + maskingStrategy = createDefaultMaskingStrategy(); + // Try to load strategy from system property if present + String strategyClass = System.getProperty(A2APayloadLoggingConfig.MASKING_STRATEGY_PROPERTY); + if (strategyClass != null && !strategyClass.isBlank()) { + try { + Class clazz = Class.forName(strategyClass.trim()); + Object instance = clazz.getDeclaredConstructor().newInstance(); + if (instance instanceof PayloadMaskingStrategy pms) { + maskingStrategy = pms; + } + } catch (Exception ignored) { + } + } + } + + /** + * Sets custom masking strategy. + */ + public static void setMaskingStrategy(PayloadMaskingStrategy strategy) { + if (strategy == null) { + throw new IllegalArgumentException("strategy must not be null"); + } + maskingStrategy = strategy; + } + + public static PayloadMaskingStrategy getMaskingStrategy() { + return maskingStrategy; + } + + /** + * Returns underlying SLF4J logger for payload events. + */ + public static Logger getLogger() { + return LOGGER; + } + + /** + * Checks if payload logging should be emitted. + * True when config enabled (or legacy property) and DEBUG enabled. + */ + public static boolean isEnabled() { + A2APayloadLoggingConfig cfg = CONFIG_REF.get(); + boolean enabled = cfg.isEnabled() || Boolean.getBoolean(LEGACY_ENABLED_PROP); + return enabled && LOGGER.isDebugEnabled(); + } + + /** + * Checks if logging would be enabled ignoring logger level (for testing). + */ + public static boolean isConfigEnabled() { + A2APayloadLoggingConfig cfg = CONFIG_REF.get(); + return cfg.isEnabled() || Boolean.getBoolean(LEGACY_ENABLED_PROP); + } + + private static PayloadMaskingStrategy createDefaultMaskingStrategy() { + A2APayloadLoggingConfig cfg = CONFIG_REF.get(); + Set sensitiveKeys = cfg.getSensitiveKeys(); + if (sensitiveKeys != null) { + return new DefaultPayloadMaskingStrategy(sensitiveKeys); + } + // Check system property for sensitive keys override + String keysProp = System.getProperty(A2APayloadLoggingConfig.MASKING_SENSITIVE_KEYS_PROPERTY); + if (keysProp != null && !keysProp.isBlank()) { + Set keys = Set.of(keysProp.split(",")); + java.util.Set trimmed = new java.util.HashSet<>(); + for (String k : keys) { + if (k != null && !k.isBlank()) { + trimmed.add(k.trim()); + } + } + if (!trimmed.isEmpty()) { + return new DefaultPayloadMaskingStrategy(trimmed); + } + } + return new DefaultPayloadMaskingStrategy(); + } + + // ==================== Core logging methods ==================== + + /** + * Logs a structured payload event. + * + * @param eventType the event type + * @param payload the payload as JSON string (or any string); may be null + */ + public static void log(A2AEventType eventType, @Nullable String payload) { + log(eventType, payload, null); + } + + /** + * Logs a structured payload event with additional context fields. + * + * @param eventType the event type + * @param payload the payload as JSON string; may be null + * @param context additional context key-values (e.g., taskId, operation), may be null + */ + public static void log(A2AEventType eventType, @Nullable String payload, @Nullable Map context) { + if (!isEnabled()) { + return; + } + A2APayloadLoggingConfig cfg = CONFIG_REF.get(); + String masked = payload; + if (cfg.isMaskingEnabled() && payload != null) { + try { + masked = maskingStrategy.mask(payload); + } catch (Exception e) { + // Fallback to original payload on masking failure to avoid log loss + masked = payload; + } + } + String truncated = PayloadTruncator.truncate(masked, cfg.getMaxSize()); + String traceId = TraceCorrelation.getTraceId(); + String spanId = TraceCorrelation.getSpanId(); + + // Build structured message + StringBuilder msg = new StringBuilder(); + msg.append("a2a.payload event_type=").append(eventType.value()); + if (truncated != null) { + msg.append(" payload=").append(truncated); + } + if (context != null && !context.isEmpty()) { + for (Map.Entry e : context.entrySet()) { + if (e.getKey() != null && e.getValue() != null) { + msg.append(" ").append(e.getKey()).append("=").append(e.getValue()); + } + } + } + if (traceId != null) { + msg.append(" trace_id=").append(traceId); + } + if (spanId != null) { + msg.append(" span_id=").append(spanId); + } + + // Attempt fluent API for structured logging (SLF4J 2.0+) + try { + var builder = LOGGER.atDebug(); + // addKeyValue returns LoggingEventBuilder; we can chain + builder = addKeyValueSafe(builder, "a2a.event.type", eventType.value()); + if (truncated != null) { + builder = addKeyValueSafe(builder, "a2a.payload", truncated); + builder = addKeyValueSafe(builder, "a2a.payload.size", String.valueOf(payload != null ? payload.length() : 0)); + if (PayloadTruncator.isTruncated(truncated)) { + builder = addKeyValueSafe(builder, "a2a.payload.truncated", "true"); + } + } + if (traceId != null) { + builder = addKeyValueSafe(builder, "trace_id", traceId); + } + if (spanId != null) { + builder = addKeyValueSafe(builder, "span_id", spanId); + } + if (context != null) { + for (Map.Entry e : context.entrySet()) { + if (e.getKey() != null && e.getValue() != null) { + builder = addKeyValueSafe(builder, e.getKey(), e.getValue()); + } + } + } + builder.log(msg.toString()); + return; + } catch (NoSuchMethodError | NoClassDefFoundError | Exception ignored) { + // Fallback to classic debug + } + // Fallback + LOGGER.debug(msg.toString()); + } + + /** + * Logs with object payload; attempts JSON serialization via toString() fallback. + * Callers that have access to JsonUtil should serialize first and call string overload. + * + * @param eventType event type + * @param payloadObject payload object; will be converted via toString() or JSON if possible + * @param context extra context + */ + public static void logObject(A2AEventType eventType, @Nullable Object payloadObject, @Nullable Map context) { + if (!isEnabled()) { + return; + } + String payloadStr = null; + if (payloadObject != null) { + // Try to use JsonUtil if available via reflection to get proper JSON + payloadStr = trySerializeToJson(payloadObject); + if (payloadStr == null) { + payloadStr = payloadObject.toString(); + } + } + log(eventType, payloadStr, context); + } + + /** + * Convenience: logs without context. + */ + public static void logObject(A2AEventType eventType, @Nullable Object payloadObject) { + logObject(eventType, payloadObject, null); + } + + /** + * Attempts to serialize object to JSON using JsonUtil via reflection if available. + */ + private static @Nullable String trySerializeToJson(Object payloadObject) { + try { + Class jsonUtilClass = Class.forName("org.a2aproject.sdk.jsonrpc.common.json.JsonUtil"); + // Try to find toJson method + var method = jsonUtilClass.getMethod("toJson", Object.class); + Object result = method.invoke(null, payloadObject); + if (result instanceof String s) { + return s; + } + } catch (ClassNotFoundException ignored) { + // JsonUtil not on classpath, fallback to toString + } catch (Exception ignored) { + } + // Try Gson directly if available + try { + Class gsonClass = Class.forName("com.google.gson.Gson"); + Object gson = gsonClass.getDeclaredConstructor().newInstance(); + var toJsonMethod = gsonClass.getMethod("toJson", Object.class); + Object result = toJsonMethod.invoke(gson, payloadObject); + if (result instanceof String s) { + return s; + } + } catch (Exception ignored) { + } + return null; + } + + @SuppressWarnings("unchecked") + private static org.slf4j.spi.LoggingEventBuilder addKeyValueSafe(org.slf4j.spi.LoggingEventBuilder builder, String key, String value) { + try { + return builder.addKeyValue(key, value); + } catch (Exception e) { + return builder; + } + } + + // ==================== Convenience methods for specific event types ==================== + + /** + * Logs a task request payload. + * + * @param operation operation name (e.g., "message/send") + * @param payload JSON payload string + * @param taskId task ID if available + * @param contextId context ID if available + */ + public static void logTaskRequest(String operation, @Nullable String payload, @Nullable String taskId, @Nullable String contextId) { + Map ctx = new java.util.HashMap<>(); + ctx.put("a2a.operation", operation); + if (taskId != null) ctx.put("a2a.task_id", taskId); + if (contextId != null) ctx.put("a2a.context_id", contextId); + log(A2AEventType.TASK_REQUEST, payload, ctx); + } + + /** + * Logs a task request with object payload. + */ + public static void logTaskRequest(String operation, @Nullable Object payloadObject, @Nullable String taskId, @Nullable String contextId) { + Map ctx = new java.util.HashMap<>(); + ctx.put("a2a.operation", operation); + if (taskId != null) ctx.put("a2a.task_id", taskId); + if (contextId != null) ctx.put("a2a.context_id", contextId); + logObject(A2AEventType.TASK_REQUEST, payloadObject, ctx); + } + + /** + * Logs a state change event. + * + * @param payload JSON payload + * @param taskId task ID + * @param newState new state name + */ + public static void logStateChange(@Nullable String payload, @Nullable String taskId, @Nullable String newState) { + Map ctx = new java.util.HashMap<>(); + if (taskId != null) ctx.put("a2a.task_id", taskId); + if (newState != null) ctx.put("a2a.task_state", newState); + log(A2AEventType.STATE_CHANGE, payload, ctx); + } + + /** + * Logs an artifact event. + * + * @param payload JSON payload + * @param taskId task ID + * @param artifactId artifact ID + */ + public static void logArtifact(@Nullable String payload, @Nullable String taskId, @Nullable String artifactId) { + Map ctx = new java.util.HashMap<>(); + if (taskId != null) ctx.put("a2a.task_id", taskId); + if (artifactId != null) ctx.put("a2a.artifact_id", artifactId); + log(A2AEventType.ARTIFACT, payload, ctx); + } + + /** + * Logs an authorization decision. + * + * @param operation operation name + * @param taskId task ID + * @param allowed true if allowed + * @param reason reason or user info + */ + public static void logAuthorization(String operation, @Nullable String taskId, boolean allowed, @Nullable String reason) { + Map ctx = new java.util.HashMap<>(); + ctx.put("a2a.operation", operation); + if (taskId != null) ctx.put("a2a.task_id", taskId); + ctx.put("a2a.auth.allowed", String.valueOf(allowed)); + if (reason != null) ctx.put("a2a.auth.reason", reason); + // Authorization payload is the decision itself; we log with minimal payload to avoid leaking + String payload = "{\"operation\":\"" + operation + "\",\"taskId\":\"" + taskId + "\",\"allowed\":" + allowed + "}"; + log(A2AEventType.AUTHORIZATION, payload, ctx); + } + + /** + * Logs a push subscription event (create/get/list/delete/subscribe). + * + * @param operation operation name + * @param payload JSON payload + * @param taskId task ID + * @param configId config ID + */ + public static void logPushSubscription(String operation, @Nullable String payload, @Nullable String taskId, @Nullable String configId) { + Map ctx = new java.util.HashMap<>(); + ctx.put("a2a.operation", operation); + if (taskId != null) ctx.put("a2a.task_id", taskId); + if (configId != null) ctx.put("a2a.config_id", configId); + log(A2AEventType.NOTIFICATION_SUBSCRIPTION, payload, ctx); + } + + /** + * Logs a push delivery event. + * + * @param taskId task ID + * @param url push URL (will be masked) + * @param payload event payload + * @param success delivery success + */ + public static void logPushDelivery(@Nullable String taskId, @Nullable String url, @Nullable String payload, boolean success) { + Map ctx = new java.util.HashMap<>(); + if (taskId != null) ctx.put("a2a.task_id", taskId); + if (url != null) ctx.put("a2a.push.url", url); + ctx.put("a2a.push.success", String.valueOf(success)); + log(A2AEventType.NOTIFICATION_DELIVERY, payload, ctx); + } + + /** + * Logs a negotiation payload (extensions, capabilities). + * + * @param operation operation name + * @param payload JSON payload + */ + public static void logNegotiation(String operation, @Nullable String payload) { + Map ctx = Map.of("a2a.operation", operation); + log(A2AEventType.NEGOTIATION, payload, ctx); + } + + /** + * Logs a streaming event. + * + * @param payload JSON payload + * @param taskId task ID + * @param eventKind event kind (task, statusUpdate, artifactUpdate, message) + */ + public static void logStreamingEvent(@Nullable String payload, @Nullable String taskId, @Nullable String eventKind) { + Map ctx = new java.util.HashMap<>(); + if (taskId != null) ctx.put("a2a.task_id", taskId); + if (eventKind != null) ctx.put("a2a.streaming.kind", eventKind); + log(A2AEventType.STREAMING_EVENT, payload, ctx); + } +} diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/TraceCorrelation.java b/common/src/main/java/org/a2aproject/sdk/common/logging/TraceCorrelation.java new file mode 100644 index 000000000..564be97f0 --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/TraceCorrelation.java @@ -0,0 +1,169 @@ +package org.a2aproject.sdk.common.logging; + +import java.lang.reflect.Method; + +import org.jspecify.annotations.Nullable; +import org.slf4j.MDC; + +/** + * Utility to correlate log entries with traces via trace_id and span_id. + *

+ * Tries to retrieve trace context from: + *

    + *
  1. OpenTelemetry API via reflection (if on classpath): {@code io.opentelemetry.api.trace.Span.current().getSpanContext()}
    2. + *
  2. SLF4J MDC keys: {@code traceId}, {@code trace_id}, {@code traceID}, {@code X-B3-TraceId}
    3. + *
  3. Similarly for span: {@code spanId}, {@code span_id}
    4. + *
+ *

+ *

+ * Reflection is used to avoid a hard dependency on OpenTelemetry. If OTel is not + * on the classpath, the lookup gracefully falls back to MDC. + *

+ */ +public final class TraceCorrelation { + + private static final String INVALID_TRACE_ID = "00000000000000000000000000000000"; + private static final String INVALID_SPAN_ID = "0000000000000000"; + + private TraceCorrelation() { + } + + /** + * Retrieves current trace ID from OpenTelemetry or MDC. + * + * @return trace ID or null if not available + */ + public static @Nullable String getTraceId() { + String traceId = getTraceIdFromOtel(); + if (traceId != null && !traceId.isBlank() && !INVALID_TRACE_ID.equals(traceId)) { + return traceId; + } + return getFromMdc("traceId", "trace_id", "traceID", "X-B3-TraceId", "X-B3-TraceID", "trace-id"); + } + + /** + * Retrieves current span ID from OpenTelemetry or MDC. + * + * @return span ID or null if not available + */ + public static @Nullable String getSpanId() { + String spanId = getSpanIdFromOtel(); + if (spanId != null && !spanId.isBlank() && !INVALID_SPAN_ID.equals(spanId)) { + return spanId; + } + return getFromMdc("spanId", "span_id", "spanID", "X-B3-SpanId", "X-B3-SpanID", "span-id"); + } + + private static @Nullable String getFromMdc(String... keys) { + for (String key : keys) { + String value = MDC.get(key); + if (value != null && !value.isBlank()) { + return value; + } + } + return null; + } + + private static @Nullable String getTraceIdFromOtel() { + try { + Class spanClass = Class.forName("io.opentelemetry.api.trace.Span"); + Method currentMethod = spanClass.getMethod("current"); + Object span = currentMethod.invoke(null); + if (span == null) { + return null; + } + Method getSpanContextMethod = spanClass.getMethod("getSpanContext"); + Object spanContext = getSpanContextMethod.invoke(span); + if (spanContext == null) { + return null; + } + Class spanContextClass = spanContext.getClass(); + Method isValidMethod = spanContextClass.getMethod("isValid"); + Boolean isValid = (Boolean) isValidMethod.invoke(spanContext); + if (isValid == null || !isValid) { + return null; + } + Method getTraceIdMethod = spanContextClass.getMethod("getTraceId"); + Object traceId = getTraceIdMethod.invoke(spanContext); + if (traceId instanceof String str) { + return str.isBlank() ? null : str; + } + return null; + } catch (ClassNotFoundException e) { + // OTel not on classpath, ignore + return null; + } catch (Exception e) { + return null; + } + } + + private static @Nullable String getSpanIdFromOtel() { + try { + Class spanClass = Class.forName("io.opentelemetry.api.trace.Span"); + Method currentMethod = spanClass.getMethod("current"); + Object span = currentMethod.invoke(null); + if (span == null) { + return null; + } + Method getSpanContextMethod = spanClass.getMethod("getSpanContext"); + Object spanContext = getSpanContextMethod.invoke(span); + if (spanContext == null) { + return null; + } + Class spanContextClass = spanContext.getClass(); + Method isValidMethod = spanContextClass.getMethod("isValid"); + Boolean isValid = (Boolean) isValidMethod.invoke(spanContext); + if (isValid == null || !isValid) { + return null; + } + Method getSpanIdMethod = spanContextClass.getMethod("getSpanId"); + Object spanId = getSpanIdMethod.invoke(spanContext); + if (spanId instanceof String str) { + return str.isBlank() ? null : str; + } + return null; + } catch (ClassNotFoundException e) { + return null; + } catch (Exception e) { + return null; + } + } + + /** + * Injects trace correlation IDs into MDC for log correlation. + * Useful for manual propagation when OTel context is not automatically in MDC. + * + * @param traceId trace ID to set, may be null to remove + * @param spanId span ID to set, may be null to remove + */ + public static void setMdc(@Nullable String traceId, @Nullable String spanId) { + if (traceId != null) { + MDC.put("trace_id", traceId); + MDC.put("traceId", traceId); + } else { + MDC.remove("trace_id"); + MDC.remove("traceId"); + } + if (spanId != null) { + MDC.put("span_id", spanId); + MDC.put("spanId", spanId); + } else { + MDC.remove("span_id"); + MDC.remove("spanId"); + } + } + + /** + * Clears trace correlation from MDC. + */ + public static void clearMdc() { + MDC.remove("trace_id"); + MDC.remove("traceId"); + MDC.remove("span_id"); + MDC.remove("spanId"); + MDC.remove("traceID"); + MDC.remove("spanID"); + MDC.remove("X-B3-TraceId"); + MDC.remove("X-B3-SpanId"); + } +} diff --git a/common/src/main/java/org/a2aproject/sdk/common/logging/package-info.java b/common/src/main/java/org/a2aproject/sdk/common/logging/package-info.java new file mode 100644 index 000000000..7310a98c4 --- /dev/null +++ b/common/src/main/java/org/a2aproject/sdk/common/logging/package-info.java @@ -0,0 +1,9 @@ +/** + * Structured payload logging for A2A protocol messages. + *

+ * Provides DEBUG-level structured log events for task requests, state changes, + * artifacts, negotiation, authorization, and notification events with + * configurable truncation and sensitive data masking. + *

+ */ +package org.a2aproject.sdk.common.logging; diff --git a/common/src/test/java/org/a2aproject/sdk/common/logging/A2APayloadLoggingConfigTest.java b/common/src/test/java/org/a2aproject/sdk/common/logging/A2APayloadLoggingConfigTest.java new file mode 100644 index 000000000..77daa6910 --- /dev/null +++ b/common/src/test/java/org/a2aproject/sdk/common/logging/A2APayloadLoggingConfigTest.java @@ -0,0 +1,75 @@ +package org.a2aproject.sdk.common.logging; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +public class A2APayloadLoggingConfigTest { + + @AfterEach + void tearDown() { + System.clearProperty(A2APayloadLoggingConfig.ENABLED_PROPERTY); + System.clearProperty(A2APayloadLoggingConfig.MAX_SIZE_PROPERTY); + System.clearProperty(A2APayloadLoggingConfig.MASKING_ENABLED_PROPERTY); + } + + @Test + void testDefaultConfig() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder().build(); + assertFalse(config.isEnabled()); + assertEquals(4096, config.getMaxSize()); + assertTrue(config.isMaskingEnabled()); + } + + @Test + void testBuilder() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder() + .enabled(true) + .maxSize(8192) + .maskingEnabled(false) + .build(); + assertTrue(config.isEnabled()); + assertEquals(8192, config.getMaxSize()); + assertFalse(config.isMaskingEnabled()); + } + + @Test + void testFromSystemPropertiesEnabled() { + System.setProperty(A2APayloadLoggingConfig.ENABLED_PROPERTY, "true"); + A2APayloadLoggingConfig cfg = A2APayloadLoggingConfig.fromSystemProperties(); + assertTrue(cfg.isEnabled()); + } + + @Test + void testFromSystemPropertiesMaxSize() { + System.setProperty(A2APayloadLoggingConfig.MAX_SIZE_PROPERTY, "2048"); + A2APayloadLoggingConfig cfg = A2APayloadLoggingConfig.fromSystemProperties(); + assertEquals(2048, cfg.getMaxSize()); + } + + @Test + void testFromSystemPropertiesMaskingDisabled() { + System.setProperty(A2APayloadLoggingConfig.MASKING_ENABLED_PROPERTY, "false"); + A2APayloadLoggingConfig cfg = A2APayloadLoggingConfig.fromSystemProperties(); + assertFalse(cfg.isMaskingEnabled()); + } + + @Test + void testFromSystemPropertiesInvalidNumberUsesDefault() { + System.setProperty(A2APayloadLoggingConfig.MAX_SIZE_PROPERTY, "invalid"); + A2APayloadLoggingConfig cfg = A2APayloadLoggingConfig.fromSystemProperties(); + assertEquals(4096, cfg.getMaxSize()); + } + + @Test + void testToBuilder() { + A2APayloadLoggingConfig original = A2APayloadLoggingConfig.builder().enabled(true).maxSize(1234).build(); + A2APayloadLoggingConfig copy = original.toBuilder().maxSize(5678).build(); + assertTrue(copy.isEnabled()); + assertEquals(5678, copy.getMaxSize()); + assertEquals(1234, original.getMaxSize()); + } +} diff --git a/common/src/test/java/org/a2aproject/sdk/common/logging/DefaultPayloadMaskingStrategyTest.java b/common/src/test/java/org/a2aproject/sdk/common/logging/DefaultPayloadMaskingStrategyTest.java new file mode 100644 index 000000000..329893993 --- /dev/null +++ b/common/src/test/java/org/a2aproject/sdk/common/logging/DefaultPayloadMaskingStrategyTest.java @@ -0,0 +1,114 @@ +package org.a2aproject.sdk.common.logging; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Set; + +import org.junit.jupiter.api.Test; + +public class DefaultPayloadMaskingStrategyTest { + + @Test + public void testMaskSensitiveFields() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + String json = "{\"token\":\"secret123\",\"user\":\"john\",\"credentials\":\"mycreds\"}"; + String masked = strategy.mask(json); + assertTrue(masked.contains("\"token\":\"***\"")); + assertTrue(masked.contains("\"credentials\":\"***\"")); + assertFalse(masked.contains("secret123")); + assertFalse(masked.contains("mycreds")); + assertTrue(masked.contains("\"user\":\"john\"")); + } + + @Test + public void testMaskNestedObject() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + String json = "{\"config\":{\"url\":\"https://example.com\",\"token\":\"abc\"},\"data\":\"value\"}"; + String masked = strategy.mask(json); + assertTrue(masked.contains("\"token\":\"***\"")); + assertTrue(masked.contains("\"url\":\"***\"")); + assertFalse(masked.contains("abc")); + assertFalse(masked.contains("https://example.com")); + assertTrue(masked.contains("\"data\":\"value\"")); + } + + @Test + public void testMaskAuthenticationObject() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + String json = "{\"authentication\":{\"scheme\":\"bearer\",\"credentials\":\"supersecret\"}}"; + String masked = strategy.mask(json); + // authentication key is sensitive, so whole object should be masked + assertTrue(masked.contains("\"authentication\":\"***\"")); + assertFalse(masked.contains("supersecret")); + } + + @Test + public void testMaskArray() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + String json = "{\"items\":[{\"token\":\"t1\"},{\"token\":\"t2\"}]}"; + String masked = strategy.mask(json); + assertTrue(masked.contains("\"token\":\"***\"")); + assertFalse(masked.contains("t1")); + assertFalse(masked.contains("t2")); + } + + @Test + public void testCustomSensitiveKeys() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(Set.of("myCustomField")); + String json = "{\"myCustomField\":\"sensitive\",\"other\":\"value\"}"; + String masked = strategy.mask(json); + assertTrue(masked.contains("\"myCustomField\":\"***\"")); + assertTrue(masked.contains("\"other\":\"value\"")); + } + + @Test + public void testNoSensitiveData() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + String json = "{\"name\":\"test\",\"value\":123}"; + String masked = strategy.mask(json); + assertEquals(json, masked); + } + + @Test + public void testNullInput() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + assertEquals(null, strategy.mask(null)); + } + + @Test + public void testBlankInput() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + String blank = " "; + assertEquals(blank, strategy.mask(blank)); + } + + @Test + public void testNoOpStrategy() { + NoOpPayloadMaskingStrategy strategy = new NoOpPayloadMaskingStrategy(); + String json = "{\"token\":\"secret\"}"; + assertEquals(json, strategy.mask(json)); + } + + @Test + public void testPushNotificationConfigMasking() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + String json = "{\"taskId\":\"123\",\"url\":\"https://webhook.example.com\",\"token\":\"push-token-xyz\",\"authentication\":{\"credentials\":\"auth-creds\"}}"; + String masked = strategy.mask(json); + assertTrue(masked.contains("\"url\":\"***\"")); + assertTrue(masked.contains("\"token\":\"***\"")); + // authentication object is sensitive, should be masked wholly + assertTrue(masked.contains("\"authentication\":\"***\"")); + assertFalse(masked.contains("push-token-xyz")); + } + + @Test + public void testUrlFieldMasking() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(Set.of("url")); + String json = "{\"pushUrl\":\"https://example.com/hook\"}"; + String masked = strategy.mask(json); + // contains 'url' substring, should be masked + assertTrue(masked.contains("\"pushUrl\":\"***\"")); + } +} diff --git a/common/src/test/java/org/a2aproject/sdk/common/logging/PayloadTruncatorTest.java b/common/src/test/java/org/a2aproject/sdk/common/logging/PayloadTruncatorTest.java new file mode 100644 index 000000000..1afae6634 --- /dev/null +++ b/common/src/test/java/org/a2aproject/sdk/common/logging/PayloadTruncatorTest.java @@ -0,0 +1,58 @@ +package org.a2aproject.sdk.common.logging; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +public class PayloadTruncatorTest { + + @Test + public void testTruncateWithinLimit() { + String payload = "short payload"; + assertEquals(payload, PayloadTruncator.truncate(payload, 4096)); + assertFalse(PayloadTruncator.isTruncated(PayloadTruncator.truncate(payload, 4096))); + } + + @Test + public void testTruncateExceedsLimit() { + String payload = "a".repeat(5000); + String truncated = PayloadTruncator.truncate(payload, 4096); + assertTrue(truncated.endsWith(PayloadTruncator.TRUNCATED_MARKER)); + assertTrue(truncated.length() == 4096 + PayloadTruncator.TRUNCATED_MARKER.length()); + assertTrue(PayloadTruncator.isTruncated(truncated)); + } + + @Test + public void testTruncateExactLimit() { + String payload = "a".repeat(4096); + String result = PayloadTruncator.truncate(payload, 4096); + assertEquals(payload, result); + assertFalse(PayloadTruncator.isTruncated(result)); + } + + @Test + public void testTruncateNull() { + assertNull(PayloadTruncator.truncate(null, 4096)); + } + + @Test + public void testTruncateZeroMaxSize() { + String payload = "test"; + assertEquals(payload, PayloadTruncator.truncate(payload, 0)); + } + + @Test + public void testIsTruncatedNull() { + assertFalse(PayloadTruncator.isTruncated(null)); + } + + @Test + public void testDefaultMaxSize() { + String payload = "a".repeat(4097); + String truncated = PayloadTruncator.truncate(payload, A2APayloadLoggingConfig.DEFAULT_MAX_SIZE); + assertTrue(PayloadTruncator.isTruncated(truncated)); + } +} diff --git a/common/src/test/java/org/a2aproject/sdk/common/logging/StructuredPayloadLoggerTest.java b/common/src/test/java/org/a2aproject/sdk/common/logging/StructuredPayloadLoggerTest.java new file mode 100644 index 000000000..48e3ccab3 --- /dev/null +++ b/common/src/test/java/org/a2aproject/sdk/common/logging/StructuredPayloadLoggerTest.java @@ -0,0 +1,216 @@ +package org.a2aproject.sdk.common.logging; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.MDC; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import org.slf4j.LoggerFactory; + +public class StructuredPayloadLoggerTest { + + private ListAppender appender; + private Logger payloadLogger; + private A2APayloadLoggingConfig originalConfig; + + @BeforeEach + void setUp() { + originalConfig = StructuredPayloadLogger.getConfig(); + payloadLogger = (Logger) LoggerFactory.getLogger("org.a2aproject.sdk.payload"); + payloadLogger.setLevel(Level.DEBUG); + appender = new ListAppender<>(); + appender.start(); + payloadLogger.addAppender(appender); + } + + @AfterEach + void tearDown() { + payloadLogger.detachAppender(appender); + StructuredPayloadLogger.setConfig(originalConfig); + MDC.clear(); + TraceCorrelation.clearMdc(); + System.clearProperty(A2APayloadLoggingConfig.ENABLED_PROPERTY); + System.clearProperty(A2APayloadLoggingConfig.MAX_SIZE_PROPERTY); + System.clearProperty(A2APayloadLoggingConfig.MASKING_ENABLED_PROPERTY); + } + + @Test + void testLogWithTruncation() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder() + .enabled(true) + .maxSize(10) + .maskingEnabled(false) + .build(); + StructuredPayloadLogger.setConfig(config); + + String largePayload = "a".repeat(20); + StructuredPayloadLogger.log(A2AEventType.TASK_REQUEST, largePayload); + + assertEquals(1, appender.list.size()); + ILoggingEvent event = appender.list.get(0); + // Check message contains truncated marker + assertTrue(event.getFormattedMessage().contains(PayloadTruncator.TRUNCATED_MARKER)); + // Check structured keyValues + boolean hasTruncatedFlag = event.getKeyValuePairs().stream() + .anyMatch(kvp -> "a2a.payload.truncated".equals(kvp.key) && "true".equals(kvp.value)); + assertTrue(hasTruncatedFlag); + } + + @Test + void testLogWithMasking() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder() + .enabled(true) + .maxSize(4096) + .maskingEnabled(true) + .build(); + StructuredPayloadLogger.setConfig(config); + + String payload = "{\"token\":\"secret\",\"data\":\"value\"}"; + StructuredPayloadLogger.log(A2AEventType.TASK_REQUEST, payload); + + assertEquals(1, appender.list.size()); + ILoggingEvent event = appender.list.get(0); + // Masked payload should not contain secret + assertFalse(event.getFormattedMessage().contains("secret")); + assertTrue(event.getFormattedMessage().contains("\"token\":\"***\"")); + } + + @Test + void testLogWithoutMasking() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder() + .enabled(true) + .maxSize(4096) + .maskingEnabled(false) + .build(); + StructuredPayloadLogger.setConfig(config); + + String payload = "{\"token\":\"secret\",\"data\":\"value\"}"; + StructuredPayloadLogger.log(A2AEventType.TASK_REQUEST, payload); + + assertEquals(1, appender.list.size()); + ILoggingEvent event = appender.list.get(0); + assertTrue(event.getFormattedMessage().contains("secret")); + } + + @Test + void testLogDisabledWhenConfigDisabled() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder() + .enabled(false) + .build(); + StructuredPayloadLogger.setConfig(config); + StructuredPayloadLogger.log(A2AEventType.TASK_REQUEST, "{\"data\":\"value\"}"); + assertEquals(0, appender.list.size()); + } + + @Test + void testLogWithTraceCorrelationViaMDC() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder() + .enabled(true) + .build(); + StructuredPayloadLogger.setConfig(config); + MDC.put("trace_id", "trace-123"); + MDC.put("span_id", "span-456"); + + StructuredPayloadLogger.log(A2AEventType.TASK_REQUEST, "{\"data\":\"value\"}"); + + assertEquals(1, appender.list.size()); + ILoggingEvent event = appender.list.get(0); + String msg = event.getFormattedMessage(); + assertTrue(msg.contains("trace_id=trace-123")); + assertTrue(msg.contains("span_id=span-456")); + // Check keyValues + boolean hasTrace = event.getKeyValuePairs().stream().anyMatch(kvp -> "trace_id".equals(kvp.key) && "trace-123".equals(kvp.value)); + assertTrue(hasTrace); + } + + @Test + void testCustomMaskingStrategy() { + PayloadMaskingStrategy custom = payload -> payload.replace("secret", "***CUSTOM***"); + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder() + .enabled(true) + .maskingEnabled(true) + .maskingStrategy(custom) + .build(); + StructuredPayloadLogger.setConfig(config); + // Also set via setter to override default creation + StructuredPayloadLogger.setMaskingStrategy(custom); + + StructuredPayloadLogger.log(A2AEventType.TASK_REQUEST, "{\"data\":\"secret\"}"); + + assertEquals(1, appender.list.size()); + assertTrue(appender.list.get(0).getFormattedMessage().contains("***CUSTOM***")); + assertFalse(appender.list.get(0).getFormattedMessage().contains("secret")); + } + + @Test + void testConfigFromSystemProperties() { + System.setProperty(A2APayloadLoggingConfig.ENABLED_PROPERTY, "true"); + System.setProperty(A2APayloadLoggingConfig.MAX_SIZE_PROPERTY, "100"); + A2APayloadLoggingConfig cfg = A2APayloadLoggingConfig.fromSystemProperties(); + assertTrue(cfg.isEnabled()); + assertEquals(100, cfg.getMaxSize()); + } + + @Test + void testDefaultMaskingKeysCoverPushNotification() { + DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(); + String payload = "{\"url\":\"https://hook.example.com\",\"token\":\"mytoken\",\"taskId\":\"123\"}"; + String masked = strategy.mask(payload); + assertTrue(masked.contains("\"url\":\"***\"")); + assertTrue(masked.contains("\"token\":\"***\"")); + assertFalse(masked.contains("mytoken")); + assertFalse(masked.contains("https://hook.example.com")); + // taskId should not be masked + assertTrue(masked.contains("123")); + } + + @Test + void testLogTaskRequestConvenience() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder().enabled(true).build(); + StructuredPayloadLogger.setConfig(config); + + StructuredPayloadLogger.logTaskRequest("message/send", "{\"msg\":\"hello\"}", "task-123", "ctx-456"); + assertEquals(1, appender.list.size()); + String msg = appender.list.get(0).getFormattedMessage(); + assertTrue(msg.contains("a2a.task_id=task-123")); + assertTrue(msg.contains("a2a.context_id=ctx-456")); + assertTrue(msg.contains("a2a.operation=message/send")); + } + + @Test + void testLogAuthorization() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder().enabled(true).build(); + StructuredPayloadLogger.setConfig(config); + + StructuredPayloadLogger.logAuthorization("GET_TASK", "task-123", true, "allowed by owner"); + assertEquals(1, appender.list.size()); + String msg = appender.list.get(0).getFormattedMessage(); + assertTrue(msg.contains("a2a.auth.allowed=true")); + assertTrue(msg.contains("a2a.task_id=task-123")); + } + + @Test + void testLogPushDelivery() { + A2APayloadLoggingConfig config = A2APayloadLoggingConfig.builder().enabled(true).build(); + StructuredPayloadLogger.setConfig(config); + + StructuredPayloadLogger.logPushDelivery("task-1", "https://example.com/hook", "{\"event\":\"data\"}", true); + assertEquals(1, appender.list.size()); + String msg = appender.list.get(0).getFormattedMessage(); + assertTrue(msg.contains("a2a.task_id=task-1")); + assertTrue(msg.contains("a2a.push.success=true")); + } +} diff --git a/common/src/test/java/org/a2aproject/sdk/common/logging/TraceCorrelationTest.java b/common/src/test/java/org/a2aproject/sdk/common/logging/TraceCorrelationTest.java new file mode 100644 index 000000000..c34836580 --- /dev/null +++ b/common/src/test/java/org/a2aproject/sdk/common/logging/TraceCorrelationTest.java @@ -0,0 +1,59 @@ +package org.a2aproject.sdk.common.logging; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.slf4j.MDC; + +public class TraceCorrelationTest { + + @AfterEach + void tearDown() { + MDC.clear(); + TraceCorrelation.clearMdc(); + } + + @Test + void testGetTraceIdFromMDC() { + MDC.put("trace_id", "trace-abc-123"); + assertEquals("trace-abc-123", TraceCorrelation.getTraceId()); + } + + @Test + void testGetSpanIdFromMDC() { + MDC.put("span_id", "span-xyz-789"); + assertEquals("span-xyz-789", TraceCorrelation.getSpanId()); + } + + @Test + void testGetTraceIdAlternativeKeys() { + MDC.put("traceId", "trace-alt"); + assertEquals("trace-alt", TraceCorrelation.getTraceId()); + } + + @Test + void testNoTraceReturnsNull() { + assertNull(TraceCorrelation.getTraceId()); + assertNull(TraceCorrelation.getSpanId()); + } + + @Test + void testSetAndClearMdc() { + TraceCorrelation.setMdc("t-123", "s-456"); + assertEquals("t-123", MDC.get("trace_id")); + assertEquals("s-456", MDC.get("span_id")); + TraceCorrelation.clearMdc(); + assertNull(MDC.get("trace_id")); + assertNull(MDC.get("span_id")); + } + + @Test + void testSetMdcNullRemoves() { + MDC.put("trace_id", "existing"); + TraceCorrelation.setMdc(null, null); + assertNull(MDC.get("trace_id")); + assertNull(MDC.get("traceId")); + } +} diff --git a/docs/content/dev/extra/payload-logging.md b/docs/content/dev/extra/payload-logging.md new file mode 100644 index 000000000..98539e5cd --- /dev/null +++ b/docs/content/dev/extra/payload-logging.md @@ -0,0 +1,237 @@ +--- +title: Structured Payload Logging +description: Structured payload logging for A2A protocol messages with truncation, masking, and trace correlation. +layout: page +--- + +# Structured Payload Logging + +Provides structured, DEBUG-level log events for A2A protocol message payloads with configurable truncation, sensitive-data masking, and trace correlation. + +This is the **proper logging mechanism** for payload inspection — unlike the previous opt-in `toString()` on span attributes, it gives you structured fields, safe masking, and trace-linked logs suitable for production. + +## Why Not Span Attributes? + +The previous mechanism (`-Dorg.a2aproject.sdk.server.extract.request=true` / `...response=true`) puts raw `toString()` into span attributes: + +- No truncation → risk of huge spans +- No masking → sensitive data leaks into tracing backend +- Not structured → cannot query by fields +- Not a log → no log-level control, no audit trail + +Payload logging fixes all of these: DEBUG-structured logs with masking, truncation, and `trace_id`/`span_id` correlation. + +## What Is Logged + +When enabled, the SDK emits structured logs (logger `org.a2aproject.sdk.payload` at DEBUG) for: + +| Category | Event Type | Logged Entities | +|----------|------------|-----------------| +| **Task requests** | `task.request` | `message/send`, `message/stream`, `tasks/get`, `tasks/list`, `tasks/cancel` | +| **State changes** | `task.state_change` | `TaskStatusUpdateEvent` (e.g., `submitted → working → completed`) | +| **Result artifacts** | `task.artifact` | `TaskArtifactUpdateEvent` with `Artifact` parts | +| **Task results** | `task.result` | Final `Task` snapshots | +| **Negotiation** | `negotiation` | Extension lists, `GetExtendedAgentCard` | +| **Authorization** | `authorization` | Allow/deny decisions from `TaskAuthorizationProvider` | +| **Push subscriptions** | `notification.subscription` | Create/get/list/delete push configs, `subscribeToTask` | +| **Push deliveries** | `notification.delivery` | Webhook dispatch attempts (url, success) | +| **Streaming events** | `streaming.event` | Per-event SSE payloads | + +Each log line includes: + +- `a2a.event.type` — one of the values above +- `a2a.payload` — the JSON payload (masked, truncated) +- `a2a.payload.size` — original size +- `a2a.payload.truncated` — `true` if truncated +- `a2a.task_id`, `a2a.context_id`, `a2a.operation`, etc. — as applicable +- `trace_id` / `span_id` — from OpenTelemetry `Span.current()` or MDC (`trace_id`/`traceId`, `span_id`/`spanId`) + +Example (Logback/Text): + +``` +10:15:23.123 [executor-thread-1] DEBUG org.a2aproject.sdk.payload - a2a.payload event_type=task.request payload={"message":{"taskId":"t-123","parts":[{"text":"hello"}]}} a2a.task_id=t-123 a2a.operation=message/send trace_id=abc123 span_id=def456 +``` + +With a JSON encoder (e.g., logstash-logback), the `keyValue` pairs become JSON fields. + +## Enabling + +### System Properties (quickest) + +```bash +# Enable payload logging +-Da2a.logging.payload.enabled=true + +# Optional tuning (defaults shown) +-Da2a.logging.payload.max-size=4096 +-Da2a.logging.payload.masking.enabled=true +``` + +And set the logger to DEBUG (Logback `logback.xml` or `application.properties`): + +```xml + +``` + +```properties +quarkus.log.category."org.a2aproject.sdk.payload".level=DEBUG +``` + +### Programmatic (Quarkus, tests) + +```java +import org.a2aproject.sdk.common.logging.*; + +A2APayloadLoggingConfig cfg = A2APayloadLoggingConfig.builder() + .enabled(true) + .maxSize(4096) + .maskingEnabled(true) + .build(); +StructuredPayloadLogger.setConfig(cfg); + +// Optional: custom masking strategy +StructuredPayloadLogger.setMaskingStrategy(new MyMaskingStrategy()); +``` + +### In `a2a-defaults.properties` + +These defaults ship with `server-common`: + +```properties +a2a.logging.payload.enabled=false +a2a.logging.payload.max-size=4096 +a2a.logging.payload.masking.enabled=true +``` + +Override via MicroProfile Config, system properties, or environment variables per your runtime. + +## Truncation + +- Default limit is **4096 characters** (system property `a2a.logging.payload.max-size`). +- If the payload exceeds the limit, it is cut and ` [truncated]` is appended. +- Log field `a2a.payload.truncated=true` signals truncation; `a2a.payload.size` keeps original length. + +```json +{"a2a.payload":"{\"large\": \"...\"} [truncated]","a2a.payload.truncated":"true","a2a.payload.size":"8192"} +``` + +## Sensitive Data Masking + +By default, fields whose keys contain these substrings (case-insensitive) are masked to `"***"`: + +`token`, `credentials`, `password`, `secret`, `apikey`/`api_key`, `privatekey`/`private_key`, `authorization`, `authentication`, `accesstoken`/`access_token`, `refreshtoken`/`refresh_token`, `clientsecret`/`client_secret`, `url` + +This covers push notification `url`/`token`, `authentication.credentials`, security schemes, etc. + +- Masking is **recursive**: nested objects and arrays are traversed. +- Non-JSON payloads fall back to regex masking. +- Malformed payloads return the original string (never drop a log due to masking). + +### Customizing Keys + +```bash +-Da2a.logging.payload.masking.sensitive-keys=token,mySecret,internalUrl +``` + +Or programmatically: + +```java +DefaultPayloadMaskingStrategy strategy = new DefaultPayloadMaskingStrategy(Set.of("token","mySecret")); +StructuredPayloadLogger.setMaskingStrategy(strategy); +``` + +### Disabling Masking + +```bash +-Da2a.logging.payload.masking.enabled=false +``` + +Or `NoOpPayloadMaskingStrategy`: + +```java +StructuredPayloadLogger.setMaskingStrategy(new NoOpPayloadMaskingStrategy()); +``` + +### Custom Strategy + +Implement `PayloadMaskingStrategy`: + +```java +public class MyMasking implements PayloadMaskingStrategy { + public String mask(String json) { + // your logic, e.g., encrypt, hash, or different placeholder + return ...; + } +} +``` + +Register via system property: + +```bash +-Da2a.logging.payload.masking.strategy=com.example.MyMasking +``` + +Or via `A2APayloadLoggingConfig.builder().maskingStrategy(new MyMasking()).build()`. + +## Trace Correlation + +Logs auto-include `trace_id`/`span_id` if available: + +1. **OpenTelemetry**: reflection on `io.opentelemetry.api.trace.Span.current().getSpanContext()` (no hard dependency — if OTel not on classpath, skipped). +2. **MDC**: keys `trace_id`/`traceId`/`X-B3-TraceId` and `span_id`/`spanId`/`X-B3-SpanId`. + +If you use OpenTelemetry, spans and logs share the same IDs — link them in your backend (Jaeger, Grafana, etc.) for root-cause analysis. + +You can also set MDC manually: + +```java +TraceCorrelation.setMdc(traceId, spanId); +// ... log will pick it up +TraceCorrelation.clearMdc(); +``` + +## Architecture + +``` +MessageSendParams → DefaultRequestHandler → StructuredPayloadLogger → SLF4J (DEBUG) + ├─ JsonUtil.toJson (serialize) + ├─ PayloadMaskingStrategy (mask) + ├─ PayloadTruncator (4KB → [truncated]) + └─ TraceCorrelation (trace_id/span_id) + +AgentEmitter.updateStatus / addArtifact → ServerPayloadLoggingHelper → StructuredPayloadLogger +BasePushNotificationSender.dispatch → ServerPayloadLoggingHelper.logPushDelivery +DefaultRequestHandler.enforce* → ServerPayloadLoggingHelper.logAuthDecision +JSONRPCTransport / OpenTelemetry decorators → JsonPayloadLoggingHelper → StructuredPayloadLogger +``` + +All call sites check `StructuredPayloadLogger.isEnabled()` first (config enabled + logger DEBUG), so overhead is near-zero when disabled. + +## Security Best Practices + +- **Keep masking ON** in production (default). +- **Don't enable `extract.request/response` span attributes** in production — they have no masking/truncation. +- **Use a JSON log encoder** (e.g., logstash-logback-encoder) to get true structured fields; the keyValues are then queryable in Loki/Elasticsearch. +- **Sampling**: payload logging can be verbose for high-throughput agents; enable selectively or with a short retention for diagnostics. + +## Troubleshooting + +**No logs appear?** + +- Ensure `a2a.logging.payload.enabled=true` and logger level DEBUG. +- Check `A2APayloadLoggingConfig.fromSystemProperties()` is re-read after you set system properties (or call `StructuredPayloadLogger.resetToSystemProperties()` in tests). + +**Payload shows `***` unexpectedly?** + +- A field name matched a sensitive key (e.g., `url` masks `pushUrl`). Customize `sensitive-keys` or provide your own strategy. + +**Trace IDs missing?** + +- Ensure OTel is initialized and the current `Span` is valid (`isValid()`), or populate MDC (`trace_id`, `span_id`). +- If using Quarkus, the `quarkus-opentelemetry` extension propagates context automatically; check your collector config. + +## See Also + +- [OpenTelemetry](../extra/opentelemetry) — span-based tracing (payload logging adds log/trace linking) +- [Configuration](../../configuration) — `a2a.*` properties +- [Server](../../server) — request handler and task lifecycle diff --git a/docs/content/dev/extras.md b/docs/content/dev/extras.md index c8daf3470..6083162c9 100644 --- a/docs/content/dev/extras.md +++ b/docs/content/dev/extras.md @@ -47,3 +47,7 @@ Serve multiple tenants from a single A2A server with per-tenant `AgentExecutor` ## [OpenTelemetry](../extra/opentelemetry) Distributed tracing, metrics, and context propagation for A2A servers and clients using OpenTelemetry. The server module (`a2a-java-sdk-opentelemetry-server`) adds automatic span creation for all protocol methods with context propagation across async boundaries. Client modules (`a2a-java-sdk-opentelemetry-client`, `a2a-java-sdk-opentelemetry-client-propagation`) instrument A2A client operations. + +## [Structured Payload Logging](../extra/payload-logging) + +DEBUG-level structured logs for every A2A payload (task requests, state changes, artifacts, auth decisions, push notifications) with 4KB truncation, sensitive-field masking, and `trace_id`/`span_id` correlation. Enable via `a2a.logging.payload.enabled=true` and `org.a2aproject.sdk.payload` logger at DEBUG. diff --git a/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransport.java b/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransport.java index 574253978..18467554a 100644 --- a/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransport.java +++ b/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransport.java @@ -15,6 +15,7 @@ import org.a2aproject.sdk.client.transport.spi.ClientTransport; import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; +import org.a2aproject.sdk.jsonrpc.common.logging.JsonPayloadLoggingHelper; import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.A2AClientException; import org.a2aproject.sdk.spec.A2AMethods; @@ -92,12 +93,14 @@ public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallCont } Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { + JsonPayloadLoggingHelper.logTaskRequest(A2AMethods.SEND_MESSAGE_METHOD, request, request.message() != null ? request.message().taskId() : null, request.message() != null ? request.message().contextId() : null); EventKind result = delegate.sendMessage(request, clientContext); if (result != null && extractResponse()) { span.setAttribute(GENAI_RESPONSE, result.toString()); } if (result != null) { span.setStatus(StatusCode.OK); + JsonPayloadLoggingHelper.logTaskResult(result, result instanceof Task t ? t.id() : result instanceof org.a2aproject.sdk.spec.Message m ? m.taskId() : null); } return result; } catch (Exception ex) { @@ -450,7 +453,11 @@ public void accept(StreamingEventKind t) { spanBuilder.setAttribute("gen_ai.agent.a2a.streaming-event", t.toString()); spanBuilder.addLink(context); Span span = spanBuilder.startSpan(); - try { + try (io.opentelemetry.context.Scope scope = span.makeCurrent()) { + String taskId = t instanceof Task task ? task.id() : t instanceof org.a2aproject.sdk.spec.Message m ? m.taskId() + : t instanceof org.a2aproject.sdk.spec.TaskStatusUpdateEvent tue ? tue.taskId() + : t instanceof org.a2aproject.sdk.spec.TaskArtifactUpdateEvent taue ? taue.taskId() : null; + JsonPayloadLoggingHelper.logStreamingEvent(t, taskId, t != null ? t.getClass().getSimpleName() : null); delegate.accept(t); span.setStatus(StatusCode.OK); } finally { diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecorator.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecorator.java index 2166b06b9..03732c79e 100644 --- a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecorator.java +++ b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecorator.java @@ -33,6 +33,7 @@ import org.a2aproject.sdk.spec.TaskIdParams; import org.a2aproject.sdk.spec.TaskPushNotificationConfig; import org.a2aproject.sdk.spec.TaskQueryParams; +import org.a2aproject.sdk.server.logging.ServerPayloadLoggingHelper; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.api.trace.StatusCode; @@ -111,6 +112,9 @@ public Task onGetTask(TaskQueryParams params, ServerCallContext context) throws spanBuilder.setAttribute(GENAI_REQUEST, params.toString()); } + // Structured payload logging with trace correlation (DEBUG level, respects config) + ServerPayloadLoggingHelper.logTaskRequest(A2AMethods.GET_TASK_METHOD, params, params.id(), null); + Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { @@ -119,6 +123,9 @@ public Task onGetTask(TaskQueryParams params, ServerCallContext context) throws if (result != null && extractResponse()) { span.setAttribute(GENAI_RESPONSE, result.toString()); } + if (result != null) { + ServerPayloadLoggingHelper.logTaskResult(result, result.id()); + } span.setStatus(StatusCode.OK); return result; diff --git a/jsonrpc-common/pom.xml b/jsonrpc-common/pom.xml index 33b9222db..9b9a93021 100644 --- a/jsonrpc-common/pom.xml +++ b/jsonrpc-common/pom.xml @@ -25,6 +25,14 @@ com.google.code.gson gson + + org.jspecify + jspecify + + + org.slf4j + slf4j-api + org.junit.jupiter diff --git a/jsonrpc-common/src/main/java/org/a2aproject/sdk/jsonrpc/common/logging/JsonPayloadLoggingHelper.java b/jsonrpc-common/src/main/java/org/a2aproject/sdk/jsonrpc/common/logging/JsonPayloadLoggingHelper.java new file mode 100644 index 000000000..11a294f58 --- /dev/null +++ b/jsonrpc-common/src/main/java/org/a2aproject/sdk/jsonrpc/common/logging/JsonPayloadLoggingHelper.java @@ -0,0 +1,112 @@ +package org.a2aproject.sdk.jsonrpc.common.logging; + +import org.a2aproject.sdk.common.logging.A2AEventType; +import org.a2aproject.sdk.common.logging.StructuredPayloadLogger; +import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil; +import org.jspecify.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Helper for payload logging with JSON serialization via JsonUtil. + *

+ * Provides convenience methods to log A2A protocol payloads as JSON strings, + * handling truncation, masking, and trace correlation via {@link StructuredPayloadLogger}. + *

+ */ +public final class JsonPayloadLoggingHelper { + + private static final Logger LOGGER = LoggerFactory.getLogger(JsonPayloadLoggingHelper.class); + + private JsonPayloadLoggingHelper() { + } + + /** + * Serializes object to JSON safely, fallback to toString. + */ + public static @Nullable String toJsonSafe(@Nullable Object obj) { + if (obj == null) { + return null; + } + if (obj instanceof String s) { + return s; + } + try { + return JsonUtil.toJson(obj); + } catch (Exception e) { + LOGGER.debug("Failed to serialize payload for logging: {}", e.getMessage()); + try { + return obj.toString(); + } catch (Exception ex) { + return null; + } + } + } + + public static void logTaskRequest(String operation, @Nullable Object payload, @Nullable String taskId, @Nullable String contextId) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(payload); + StructuredPayloadLogger.logTaskRequest(operation, json, taskId, contextId); + } + + public static void logTaskResult(@Nullable Object result, @Nullable String taskId) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(result); + java.util.Map ctx = new java.util.HashMap<>(); + if (taskId != null) ctx.put("a2a.task_id", taskId); + ctx.put("a2a.operation", "task.result"); + StructuredPayloadLogger.log(A2AEventType.TASK_RESULT, json, ctx); + } + + public static void logStateChange(@Nullable Object event, @Nullable String taskId, @Nullable String newState) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(event); + StructuredPayloadLogger.logStateChange(json, taskId, newState); + } + + public static void logArtifact(@Nullable Object event, @Nullable String taskId, @Nullable String artifactId) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(event); + StructuredPayloadLogger.logArtifact(json, taskId, artifactId); + } + + public static void logStreamingEvent(@Nullable Object event, @Nullable String taskId, @Nullable String kind) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(event); + StructuredPayloadLogger.logStreamingEvent(json, taskId, kind); + } + + public static void logPushSubscription(String operation, @Nullable Object payload, @Nullable String taskId, @Nullable String configId) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(payload); + StructuredPayloadLogger.logPushSubscription(operation, json, taskId, configId); + } + + public static void logPushDelivery(@Nullable String taskId, @Nullable String url, @Nullable Object event, boolean success) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(event); + StructuredPayloadLogger.logPushDelivery(taskId, url, json, success); + } + + public static void logNegotiation(String operation, @Nullable Object payload) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(payload); + StructuredPayloadLogger.logNegotiation(operation, json); + } +} diff --git a/jsonrpc-common/src/main/java/org/a2aproject/sdk/jsonrpc/common/logging/package-info.java b/jsonrpc-common/src/main/java/org/a2aproject/sdk/jsonrpc/common/logging/package-info.java new file mode 100644 index 000000000..a64f1ac2e --- /dev/null +++ b/jsonrpc-common/src/main/java/org/a2aproject/sdk/jsonrpc/common/logging/package-info.java @@ -0,0 +1,4 @@ +/** + * JSON payload logging helpers delegating to structured logger. + */ +package org.a2aproject.sdk.jsonrpc.common.logging; diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/logging/ServerPayloadLoggingHelper.java b/server-common/src/main/java/org/a2aproject/sdk/server/logging/ServerPayloadLoggingHelper.java new file mode 100644 index 000000000..5d8281e06 --- /dev/null +++ b/server-common/src/main/java/org/a2aproject/sdk/server/logging/ServerPayloadLoggingHelper.java @@ -0,0 +1,121 @@ +package org.a2aproject.sdk.server.logging; + +import org.a2aproject.sdk.common.logging.A2AEventType; +import org.a2aproject.sdk.common.logging.StructuredPayloadLogger; +import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil; +import org.jspecify.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Helper for server-side structured payload logging. + *

+ * Serializes A2A protocol objects to JSON via {@link JsonUtil} and delegates + * to {@link StructuredPayloadLogger} with appropriate event types and context. + *

+ */ +public final class ServerPayloadLoggingHelper { + + private static final Logger LOGGER = LoggerFactory.getLogger(ServerPayloadLoggingHelper.class); + + private ServerPayloadLoggingHelper() { + } + + /** + * Safely serializes object to JSON string via JsonUtil. + */ + public static @Nullable String toJsonSafe(@Nullable Object obj) { + if (obj == null) { + return null; + } + // Fast path for already-string + if (obj instanceof String s) { + return s; + } + try { + return JsonUtil.toJson(obj); + } catch (Exception e) { + // Fallback to toString on serialization failure (should not happen for valid spec types) + LOGGER.debug("Failed to serialize payload for logging: {}", e.getMessage()); + try { + return obj.toString(); + } catch (Exception ex) { + return null; + } + } + } + + public static void logTaskRequest(String operation, @Nullable Object payload, @Nullable String taskId, @Nullable String contextId) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(payload); + StructuredPayloadLogger.logTaskRequest(operation, json, taskId, contextId); + } + + public static void logTaskResult(@Nullable Object result, @Nullable String taskId) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(result); + java.util.Map ctx = new java.util.HashMap<>(); + if (taskId != null) ctx.put("a2a.task_id", taskId); + ctx.put("a2a.operation", "task.result"); + StructuredPayloadLogger.log(A2AEventType.TASK_RESULT, json, ctx); + } + + public static void logStateChange(@Nullable Object event, @Nullable String taskId, @Nullable String newState) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(event); + StructuredPayloadLogger.logStateChange(json, taskId, newState); + } + + public static void logArtifact(@Nullable Object event, @Nullable String taskId, @Nullable String artifactId) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(event); + StructuredPayloadLogger.logArtifact(json, taskId, artifactId); + } + + public static void logStreamingEvent(@Nullable Object event, @Nullable String taskId, @Nullable String kind) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(event); + StructuredPayloadLogger.logStreamingEvent(json, taskId, kind); + } + + public static void logAuthDecision(String operation, @Nullable String taskId, boolean allowed, @Nullable String reason) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + StructuredPayloadLogger.logAuthorization(operation, taskId, allowed, reason); + } + + public static void logPushSubscription(String operation, @Nullable Object payload, @Nullable String taskId, @Nullable String configId) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(payload); + StructuredPayloadLogger.logPushSubscription(operation, json, taskId, configId); + } + + public static void logPushDelivery(@Nullable String taskId, @Nullable String url, @Nullable Object event, boolean success) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(event); + StructuredPayloadLogger.logPushDelivery(taskId, url, json, success); + } + + public static void logNegotiation(String operation, @Nullable Object payload) { + if (!StructuredPayloadLogger.isEnabled()) { + return; + } + String json = toJsonSafe(payload); + StructuredPayloadLogger.logNegotiation(operation, json); + } +} diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/logging/package-info.java b/server-common/src/main/java/org/a2aproject/sdk/server/logging/package-info.java new file mode 100644 index 000000000..e41349dbc --- /dev/null +++ b/server-common/src/main/java/org/a2aproject/sdk/server/logging/package-info.java @@ -0,0 +1,4 @@ +/** + * Server-side structured payload logging helpers. + */ +package org.a2aproject.sdk.server.logging; diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 3a669ae7a..814af4f9f 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -75,6 +75,7 @@ import org.a2aproject.sdk.spec.TaskState; import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; import org.a2aproject.sdk.spec.UnsupportedOperationError; +import org.a2aproject.sdk.server.logging.ServerPayloadLoggingHelper; import org.a2aproject.sdk.util.Assert; import org.jspecify.annotations.NonNull; import org.jspecify.annotations.Nullable; @@ -504,6 +505,7 @@ public DefaultRequestHandler build() { @Override public Task onGetTask(TaskQueryParams params, ServerCallContext context) throws A2AError { LOGGER.debug("onGetTask {}", params.id()); + ServerPayloadLoggingHelper.logTaskRequest("tasks/get", params, params.id(), null); enforceRead(context, params.id(), TaskOperation.GET_TASK); Task task = taskStore.get(params.id()); if (task == null) { @@ -512,6 +514,7 @@ public Task onGetTask(TaskQueryParams params, ServerCallContext context) throws } task = limitTaskHistory(task, params.historyLength()); LOGGER.debug("Task found {}", task); + ServerPayloadLoggingHelper.logTaskResult(task, task.id()); return task; } @@ -542,11 +545,15 @@ private static Task limitTaskHistory(Task task, @Nullable Integer historyLength) private void enforceRead(ServerCallContext context, String taskId, TaskOperation operation) throws A2AError { if (authorizationProvider == null) { if (authorizationRequired) { + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), taskId, false, "no provider, authorization required"); throw new TaskNotFoundError(); } + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), taskId, true, "no provider, authorization not required"); return; } - if (!authorizationProvider.checkRead(context, taskId, operation)) { + boolean allowed = authorizationProvider.checkRead(context, taskId, operation); + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), taskId, allowed, allowed ? "checkRead allowed" : "checkRead denied"); + if (!allowed) { throw new TaskNotFoundError(); } } @@ -554,11 +561,15 @@ private void enforceRead(ServerCallContext context, String taskId, TaskOperation private void enforceWrite(ServerCallContext context, String taskId, TaskOperation operation) throws A2AError { if (authorizationProvider == null) { if (authorizationRequired) { + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), taskId, false, "no provider, authorization required"); throw new TaskNotFoundError(); } + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), taskId, true, "no provider, authorization not required"); return; } - if (!authorizationProvider.checkWrite(context, taskId, operation)) { + boolean allowed = authorizationProvider.checkWrite(context, taskId, operation); + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), taskId, allowed, allowed ? "checkWrite allowed" : "checkWrite denied"); + if (!allowed) { throw new TaskNotFoundError(); } } @@ -566,11 +577,15 @@ private void enforceWrite(ServerCallContext context, String taskId, TaskOperatio private void enforceCreate(ServerCallContext context, TaskOperation operation) throws A2AError { if (authorizationProvider == null) { if (authorizationRequired) { + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), null, false, "no provider, authorization required"); throw new TaskNotFoundError(); } + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), null, true, "no provider, authorization not required"); return; } - if (!authorizationProvider.checkCreate(context, operation)) { + boolean allowed = authorizationProvider.checkCreate(context, operation); + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), null, allowed, allowed ? "checkCreate allowed" : "checkCreate denied"); + if (!allowed) { throw new TaskNotFoundError(); } } @@ -578,11 +593,15 @@ private void enforceCreate(ServerCallContext context, TaskOperation operation) t private void enforceListRead(ServerCallContext context) throws A2AError { if (authorizationProvider == null) { if (authorizationRequired) { + ServerPayloadLoggingHelper.logAuthDecision(TaskOperation.LIST_TASKS.name(), LIST_TASKS_SCOPE_ID, false, "no provider, authorization required"); throw new TaskNotFoundError(); } + ServerPayloadLoggingHelper.logAuthDecision(TaskOperation.LIST_TASKS.name(), LIST_TASKS_SCOPE_ID, true, "no provider, authorization not required"); return; } - if (!authorizationProvider.checkRead(context, LIST_TASKS_SCOPE_ID, TaskOperation.LIST_TASKS)) { + boolean allowed = authorizationProvider.checkRead(context, LIST_TASKS_SCOPE_ID, TaskOperation.LIST_TASKS); + ServerPayloadLoggingHelper.logAuthDecision(TaskOperation.LIST_TASKS.name(), LIST_TASKS_SCOPE_ID, allowed, allowed ? "checkRead allowed" : "checkRead denied"); + if (!allowed) { throw new TaskNotFoundError(); } } @@ -609,6 +628,7 @@ private void recordOwnershipIfNeeded(ServerCallContext context, @Nullable String @Override public ListTasksResult onListTasks(ListTasksParams params, ServerCallContext context) throws A2AError { + ServerPayloadLoggingHelper.logTaskRequest("tasks/list", params, null, params.contextId()); enforceListRead(context); LOGGER.debug("onListTasks with contextId={}, status={}, pageSize={}, pageToken={}, statusTimestampAfter={}", params.contextId(), params.status(), params.pageSize(), params.pageToken(), params.statusTimestampAfter()); @@ -647,11 +667,13 @@ public ListTasksResult onListTasks(ListTasksParams params, ServerCallContext con } } LOGGER.debug("Found {} tasks (total: {})", result.pageSize(), result.totalSize()); + ServerPayloadLoggingHelper.logTaskResult(result, null); return result; } @Override public Task onCancelTask(CancelTaskParams params, ServerCallContext context) throws A2AError { + ServerPayloadLoggingHelper.logTaskRequest("tasks/cancel", params, params.id(), null); enforceWrite(context, params.id(), TaskOperation.CANCEL_TASK); // Serialize check-then-act per task so two concurrent cancels (or a cancel racing // a concurrent completion) cannot both act on the pre-transition state. @@ -753,6 +775,11 @@ private Task doCancelTask(CancelTaskParams params, ServerCallContext context) th public EventKind onMessageSend(MessageSendParams rawParams, ServerCallContext context) throws A2AError { MessageSendParams params = normalizeBlankIds(rawParams); LOGGER.debug("onMessageSend - task: {}; context {}", params.message().taskId(), params.message().contextId()); + ServerPayloadLoggingHelper.logTaskRequest("message/send", params, params.message().taskId(), params.message().contextId()); + // Log negotiation payload if extensions present + if (params.message().extensions() != null && !params.message().extensions().isEmpty()) { + ServerPayloadLoggingHelper.logNegotiation("message/send extensions", params.message().extensions()); + } String msgTaskId = params.message().taskId(); if (msgTaskId != null) { enforceWrite(context, msgTaskId, TaskOperation.MESSAGE_SEND); @@ -957,6 +984,7 @@ public EventKind onMessageSend(MessageSendParams rawParams, ServerCallContext co recordOwnershipIfNeeded(context, extractTaskId(kind), TaskOperation.MESSAGE_SEND); LOGGER.debug("Returning: {}", kind); + ServerPayloadLoggingHelper.logTaskResult(kind, extractTaskId(kind)); return kind; } @@ -967,6 +995,10 @@ public Flow.Publisher onMessageSendStream( MessageSendParams params = normalizeBlankIds(rawParams); LOGGER.debug("onMessageSendStream START - task: {}; context: {}; runningAgents: {}", params.message().taskId(), params.message().contextId(), runningAgents.size()); + ServerPayloadLoggingHelper.logTaskRequest("message/stream", params, params.message().taskId(), params.message().contextId()); + if (params.message().extensions() != null && !params.message().extensions().isEmpty()) { + ServerPayloadLoggingHelper.logNegotiation("message/stream extensions", params.message().extensions()); + } String msgTaskId = params.message().taskId(); if (msgTaskId != null) { enforceWrite(context, msgTaskId, TaskOperation.MESSAGE_SEND_STREAM); @@ -1074,6 +1106,9 @@ public void cancel() { @Override public void onNext(StreamingEventKind item) { LOGGER.debug("onNext: {} for task {}", item.getClass().getSimpleName(), taskId.get()); + String taskIdForLog = extractTaskId(item); + String kind = item != null ? item.getClass().getSimpleName() : "unknown"; + ServerPayloadLoggingHelper.logStreamingEvent(item, taskIdForLog, kind); if (item instanceof Task task) { Integer historyLength = params.configuration() != null ? params.configuration().historyLength() : null; item = limitTaskHistory(task, historyLength); @@ -1129,6 +1164,7 @@ public void onComplete() { @Override public TaskPushNotificationConfig onCreateTaskPushNotificationConfig( TaskPushNotificationConfig params, ServerCallContext context) throws A2AError { + ServerPayloadLoggingHelper.logPushSubscription("tasks/pushNotificationConfig/create", params, params.taskId(), params.id()); if (pushConfigStore == null || !pushNotificationsEnabled) { throw new UnsupportedOperationError(); } @@ -1143,12 +1179,14 @@ public TaskPushNotificationConfig onCreateTaskPushNotificationConfig( String version = context != null ? context.getRequestedProtocolVersion() : null; TaskPushNotificationConfig result = pushConfigStore.setInfo(params, version); + ServerPayloadLoggingHelper.logPushSubscription("tasks/pushNotificationConfig/create/result", result, result.taskId(), result.id()); return result; } @Override public TaskPushNotificationConfig onGetTaskPushNotificationConfig( GetTaskPushNotificationConfigParams params, ServerCallContext context) throws A2AError { + ServerPayloadLoggingHelper.logPushSubscription("tasks/pushNotificationConfig/get", params, params.taskId(), params.id()); if (pushConfigStore == null) { throw new UnsupportedOperationError(); } @@ -1165,7 +1203,9 @@ public TaskPushNotificationConfig onGetTaskPushNotificationConfig( String requestedConfigId = params.id(); String configId = requestedConfigId == null || requestedConfigId.isEmpty() ? params.taskId() : requestedConfigId; - return getTaskPushNotificationConfig(listTaskPushNotificationConfigsResult, configId); + TaskPushNotificationConfig result = getTaskPushNotificationConfig(listTaskPushNotificationConfigsResult, configId); + ServerPayloadLoggingHelper.logPushSubscription("tasks/pushNotificationConfig/get/result", result, result.taskId(), result.id()); + return result; } private TaskPushNotificationConfig getTaskPushNotificationConfig(ListTaskPushNotificationConfigsResult notificationConfigList, @@ -1181,6 +1221,7 @@ private TaskPushNotificationConfig getTaskPushNotificationConfig(ListTaskPushNot @Override public Flow.Publisher onSubscribeToTask(TaskIdParams params, ServerCallContext context) throws A2AError { LOGGER.debug("onSubscribeToTask - taskId: {}", params.id()); + ServerPayloadLoggingHelper.logPushSubscription("tasks/subscribe", params, params.id(), null); enforceRead(context, params.id(), TaskOperation.SUBSCRIBE_TO_TASK); Task task = taskStore.get(params.id()); if (task == null) { @@ -1228,6 +1269,7 @@ public Flow.Publisher onSubscribeToTask(TaskIdParams params, @Override public ListTaskPushNotificationConfigsResult onListTaskPushNotificationConfigs( ListTaskPushNotificationConfigsParams params, ServerCallContext context) throws A2AError { + ServerPayloadLoggingHelper.logPushSubscription("tasks/pushNotificationConfig/list", params, params.id(), null); if (pushConfigStore == null) { throw new UnsupportedOperationError(); } @@ -1236,12 +1278,15 @@ public ListTaskPushNotificationConfigsResult onListTaskPushNotificationConfigs( if (task == null) { throw new TaskNotFoundError(); } - return pushConfigStore.getInfo(params); + ListTaskPushNotificationConfigsResult result = pushConfigStore.getInfo(params); + ServerPayloadLoggingHelper.logPushSubscription("tasks/pushNotificationConfig/list/result", result, params.id(), null); + return result; } @Override public void onDeleteTaskPushNotificationConfig( DeleteTaskPushNotificationConfigParams params, ServerCallContext context) throws A2AError { + ServerPayloadLoggingHelper.logPushSubscription("tasks/pushNotificationConfig/delete", params, params.taskId(), params.id()); if (pushConfigStore == null) { throw new UnsupportedOperationError(); } @@ -1253,6 +1298,7 @@ public void onDeleteTaskPushNotificationConfig( } pushConfigStore.deleteInfo(params.taskId(), params.id()); + ServerPayloadLoggingHelper.logPushSubscription("tasks/pushNotificationConfig/delete/result", null, params.taskId(), params.id()); } private boolean shouldAddPushInfo(MessageSendParams params) { @@ -1523,8 +1569,10 @@ AgentExecutor resolveAgentExecutor(@Nullable String tenant) { @Override public void authorizeTaskAccess(@Nullable String requestedTaskId, ServerCallContext context, TaskOperation operation) throws A2AError { if (requestedTaskId == null || requestedTaskId.isEmpty()) { + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), requestedTaskId, true, "no taskId, skipping auth"); return; } + ServerPayloadLoggingHelper.logAuthDecision(operation.name(), requestedTaskId, true, "authorizeTaskAccess check"); enforceRead(context, requestedTaskId, operation); Task task = taskStore.get(requestedTaskId); if (task == null) { diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java index 2f5dcad8a..6c44daf71 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java @@ -7,6 +7,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.a2aproject.sdk.server.agentexecution.RequestContext; +import org.a2aproject.sdk.server.logging.ServerPayloadLoggingHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.a2aproject.sdk.server.events.EventQueue; @@ -139,6 +140,7 @@ public void updateStatus(TaskState taskState, @Nullable Message message) { .contextId(contextId) .status(new TaskStatus(taskState, message, null)) .build(); + ServerPayloadLoggingHelper.logStateChange(event, taskId, taskState.name()); if (isFinal) { enqueueTerminalEvent(event); } else { @@ -232,6 +234,7 @@ public void addArtifact(List> parts, @Nullable String artifactId, @Nulla .append(append) .lastChunk(lastChunk) .build(); + ServerPayloadLoggingHelper.logArtifact(event, taskId, artifactId); eventQueue.enqueueEvent(event); } @@ -316,6 +319,7 @@ public boolean tryFail(A2AError error) { if (!terminalStateReached.compareAndSet(false, true)) { return false; } + ServerPayloadLoggingHelper.logTaskResult(error, taskId); enqueueTerminalEvent(error); // Status transition happens automatically in MainEventBusProcessor return true; @@ -333,6 +337,7 @@ public boolean tryFailWithTimeout(A2AError error, long timeout, TimeUnit unit) { if (!terminalStateReached.compareAndSet(false, true)) { return false; } + ServerPayloadLoggingHelper.logTaskResult(error, taskId); enqueueTerminalEvent(error, timeout, unit); return true; } @@ -532,6 +537,7 @@ public void sendMessage(List> parts) { */ public void sendMessage(List> parts, @Nullable Map metadata) { Message message = newAgentMessage(parts, metadata); + ServerPayloadLoggingHelper.logTaskRequest("message", message, taskId, contextId); eventQueue.enqueueEvent(message); } @@ -566,6 +572,7 @@ public void sendMessage(Message message) { LOGGER.error("Message contextId mismatch: expected={}, actual={}", contextId, message.contextId()); throw new IllegalArgumentException("Message contextId does not match the emitter's contextId"); } + ServerPayloadLoggingHelper.logTaskRequest("message", message, taskId, contextId); eventQueue.enqueueEvent(message); } @@ -596,6 +603,7 @@ public void sendMessage(Message message) { * @since 1.0.0 */ public void addTask(Task task) { + ServerPayloadLoggingHelper.logTaskResult(task, task.id()); eventQueue.enqueueEvent(task); } @@ -624,6 +632,16 @@ public void addTask(Task task) { * @since 1.0.0 */ public void emitEvent(Event event) { + // Log based on event type + if (event instanceof TaskStatusUpdateEvent tue) { + ServerPayloadLoggingHelper.logStateChange(event, tue.taskId(), tue.status() != null ? tue.status().state().name() : null); + } else if (event instanceof TaskArtifactUpdateEvent taue) { + ServerPayloadLoggingHelper.logArtifact(event, taue.taskId(), taue.artifact() != null ? taue.artifact().artifactId() : null); + } else if (event instanceof Task t) { + ServerPayloadLoggingHelper.logTaskResult(event, t.id()); + } else if (event instanceof Message m) { + ServerPayloadLoggingHelper.logTaskRequest("message", event, m.taskId(), m.contextId()); + } eventQueue.enqueueEvent(event); } diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java index 9dff9da60..a00d0c3a3 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java @@ -19,6 +19,7 @@ import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.A2AHttpClientFactory; import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil; +import org.a2aproject.sdk.server.logging.ServerPayloadLoggingHelper; import org.a2aproject.sdk.spec.AuthenticationInfo; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult; @@ -179,11 +180,15 @@ private boolean dispatchNotification(StreamingEventKind event, TaskPushNotificationConfig pushInfo, Map versionsByConfigId) { String url = pushInfo.url(); + String taskId = extractTaskId(event); + // Log subscription delivery attempt with payload (will be masked/truncated) + ServerPayloadLoggingHelper.logPushDelivery(taskId, url, event, true); try { urlValidator.validate(url); } catch (IllegalArgumentException e) { LOGGER.warn("Rejecting push notification to {}: {}", url, e.getMessage()); + ServerPayloadLoggingHelper.logPushDelivery(taskId, url, event, false); return false; } @@ -242,17 +247,22 @@ private boolean dispatchNotification(StreamingEventKind event, } } + boolean success; try { postBuilder .url(url) .addHeader(CONTENT_TYPE, APPLICATION_JSON) .body(body) .post(); + success = true; } catch (IOException | InterruptedException e) { LOGGER.debug("Error pushing data to " + url + ": {}", e.getMessage(), e); + ServerPayloadLoggingHelper.logPushDelivery(taskId, url, event, false); return false; } - return true; + // Log successful delivery with body (already masked via helper) + ServerPayloadLoggingHelper.logPushDelivery(taskId, url, event, success); + return success; } /** diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java index 775975727..d2892e8d5 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java @@ -22,6 +22,7 @@ import org.a2aproject.sdk.spec.TaskState; import org.a2aproject.sdk.spec.TaskStatus; import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; +import org.a2aproject.sdk.server.logging.ServerPayloadLoggingHelper; import org.jspecify.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -77,6 +78,7 @@ boolean saveTaskEvent(Task task, boolean isReplicated, @Nullable AtomicReference && task.status() != null && task.status().state() != null) { validateStateTransition(current.status().state(), task.status().state(), task.id()); } + ServerPayloadLoggingHelper.logTaskResult(task, task.id()); Task savedTask = saveTask(task, isReplicated); if (taskSnapshot != null) { taskSnapshot.set(savedTask); @@ -117,6 +119,7 @@ boolean saveTaskEvent(TaskStatusUpdateEvent event, boolean isReplicated, @Nullab } task = builder.build(); + ServerPayloadLoggingHelper.logStateChange(event, event.taskId(), newState != null ? newState.name() : null); Task savedTask = saveTask(task, isReplicated); if (taskSnapshot != null) { taskSnapshot.set(savedTask); @@ -138,6 +141,7 @@ boolean saveTaskEvent(TaskArtifactUpdateEvent event, boolean isReplicated, @Null throw new IllegalStateException("taskId should not be null after checkIdsAndUpdateIfNecessary"); } task = appendArtifactToTask(task, event, nonNullTaskId); + ServerPayloadLoggingHelper.logArtifact(event, event.taskId(), event.artifact() != null ? event.artifact().artifactId() : null); Task savedTask = saveTask(task, isReplicated); if (taskSnapshot != null) { taskSnapshot.set(savedTask); diff --git a/server-common/src/main/resources/META-INF/a2a-defaults.properties b/server-common/src/main/resources/META-INF/a2a-defaults.properties index 5c2f7cdeb..871a4a91b 100644 --- a/server-common/src/main/resources/META-INF/a2a-defaults.properties +++ b/server-common/src/main/resources/META-INF/a2a-defaults.properties @@ -59,6 +59,19 @@ a2a.push-notification.url.allowed-schemes=https # addresses (loopback, link-local, site-local, cloud metadata). Defaults to # false to prevent SSRF. Set to true ONLY in trusted development environments. a2a.push-notification.url.allow-private-network-targets=false +# Payload Logging - Structured payload logging +# When true, structured log events are emitted at DEBUG level for task requests, +# state changes, artifacts, authorization decisions, and notification events. +# Payloads are masked (sensitive fields replaced with ***) and truncated (default 4KB). +a2a.logging.payload.enabled=false +# Maximum payload size in characters before truncation (with [truncated] marker) +a2a.logging.payload.max-size=4096 +# When true, sensitive fields (token, credentials, etc.) are masked before logging +a2a.logging.payload.masking.enabled=true +# Comma-separated list of sensitive key substrings (case-insensitive) to mask +# Defaults to token, credentials, password, secret, apikey, authorization, etc. +# a2a.logging.payload.masking.sensitive-keys=token,credentials,password,secret + # TTL in seconds for DNS resolution caching during URL validation. # Avoids a blocking DNS lookup on every push notification. Set to 0 to disable. a2a.push-notification.url.dns-cache-ttl-seconds=30