This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new a46d4e047b4 Redact credentials from captured audit payloads (#19555)
a46d4e047b4 is described below
commit a46d4e047b4abe422fd71317bd428d9387049374
Author: Dylan Dreyer <[email protected]>
AuthorDate: Fri Oct 16 09:36:29 2026 -0700
Redact credentials from captured audit payloads (#19555)
---
.../apache/pinot/common/audit/AuditRedactor.java | 424 +++++++++++++++++++++
.../pinot/common/audit/AuditRequestProcessor.java | 70 +++-
.../pinot/common/audit/AuditRedactorTest.java | 377 ++++++++++++++++++
.../common/audit/AuditRequestProcessorTest.java | 264 +++++++++++--
4 files changed, 1076 insertions(+), 59 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditRedactor.java
b/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditRedactor.java
new file mode 100644
index 00000000000..6da3ef99687
--- /dev/null
+++
b/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditRedactor.java
@@ -0,0 +1,424 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.audit;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.node.TextNode;
+import java.io.IOException;
+import java.io.StringWriter;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Masks credential-bearing values out of captured audit payloads before they
are serialized into an
+/// [AuditEvent].
+///
+/// Audit records are written to stdout and to a rotating file, and are
usually shipped onward to a log
+/// pipeline whose operators are not entitled to the credentials a caller
happens to send. Anything that
+/// leaves this class must therefore be assumed to be readable by more people
than the request itself was.
+///
+/// The rules that follow from that are deliberate:
+///
+/// - **Redaction is unconditional.** There is no configuration that turns
it off, because every
+/// configuration that turns it off is a credential leak.
+/// - **Secrets are only located inside JSON that is understood.** A
complete body is parsed only if it is
+/// one JSON object or array with nothing after it (see
[#parseWholeValue(String)]); anything else,
+/// a bare scalar included, has no key to judge it by and is reported by
size only.
+/// - **A body cut at the capture cap keeps its redacted prefix.** It is
read token by token, so
+/// everything before the cut is masked exactly as in a complete body, and
only the value the cut
+/// falls inside is dropped (see [#redactTruncatedPrefix(String)]).
+///
+/// Key matching errs towards over-redaction: masking a harmless `tokenUri`
costs an audit reader
+/// nothing, while missing one `accessKey` is the bug this class exists to
prevent.
+///
+/// ## Known residual gaps
+///
+/// Values are inspected only for URL userinfo
(`scheme://user:password@host`), which is stripped. Any
+/// other credential embedded in the *value* of an innocuously named key is
not caught - a JDBC URL with a
+/// `password=` query parameter, a presigned URL whose signature sits inside
the value, or JSON serialized
+/// into a string, whose nested keys are not inspected. A body that is not
JSON - form-encoded, plain text,
+/// multipart - is recorded by size only, even where it could in principle be
redacted by key. Callers must
+/// not read this class as a guarantee that a captured payload is free of
credentials.
+///
+/// This class is stateless and thread-safe.
+final class AuditRedactor {
+
+ static final String MASKED_VALUE = "*****";
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AuditRedactor.class);
+
+ private static final JsonFactory JSON_FACTORY =
JsonUtils.DEFAULT_READER.getFactory();
+
+ /// Separator allowed between the words of a key name, so that `accessKey`,
`access_key`, `access-key`
+ /// and the Hadoop-style `access.key` are treated alike.
+ private static final String SEP = "[\\s._-]*";
+
+ /// Key patterns whose values are credential material, matched against the
whole lowercased key, so
+ /// dotted keys such as `input.fs.prop.accessKey` are matched by their
suffix. A match is final: no
+ /// exemption overrides it, so a benign suffix cannot hide a
credential-bearing prefix.
+ ///
+ /// This is a superset of [org.apache.pinot.spi.utils.Obfuscator]'s
defaults, which anchor on a suffix
+ /// and so miss the AWS `accessKey` / `accessKeyId` pair that motivated this
class. A key naming a key's
+ /// id, path, file or content (`sseKmsKeyId`, `gcpKeyPath`, `keyFile`,
`keyContentBase64`) is sensitive,
+ /// as is any key ending in `pass` (`dbPass`, `ssl.keypass`), an Azure
`connectionString` (which carries
+ /// `AccountKey=`), a session id, and a SAS `sig` parameter.
+ private static final Pattern SENSITIVE_KEY =
Pattern.compile(String.join("|", ".*secret.*", ".*password.*",
+ ".*passwd.*", ".*pwd", ".*pass", ".*passphrase.*", ".*key" + SEP +
"(id|path|file|content).*",
+ ".*access" + SEP + "key.*", ".*api" + SEP + "key.*", ".*private" + SEP +
"key.*",
+ ".*encryption" + SEP + "key.*", ".*signing" + SEP + "key.*",
".*keytab.*", ".*token.*", ".*jaas.*config.*",
+ ".*signature.*", ".*cookie.*", ".*keystore.*", ".*truststore.*",
".*user" + SEP + "info",
+ ".*connection" + SEP + "string.*", ".*session" + SEP + "id",
"(.*[\\s._-])?sig"));
+
+ /// Credential and authentication words, matched broadly - `auth` rather
than `authorization`, to cover
+ /// Confluent's `basic.auth.user.info`, a literal `user:password` pair
present in every
+ /// schema-registry-backed realtime table. Kept apart from [#SENSITIVE_KEY]
because these alone are
+ /// overridden by [#SAFE_KEY_SUFFIXES]: a key that names an authentication
mode mentions them without
+ /// holding a credential.
+ private static final Pattern CREDENTIAL_OR_AUTH =
Pattern.compile(".*(credential|creds|auth).*");
+
+ /// Any key ending in `key` is also sensitive (`fs.s3a.access.key`,
`gcpKey`, `jsonKey`,
+ /// `stream.confluent.key`), except the structural names in
[#SAFE_KEY_NAMES]. Kept apart from
+ /// [#SENSITIVE_KEY] so that those exemptions apply to this rule alone.
+ private static final Pattern ENDS_IN_KEY = Pattern.compile(".*key");
+
+ /// Names that end in `key` without being a credential. Compared,
lowercased, against both the full key
+ /// and its last dotted segment.
+ private static final Set<String> SAFE_KEY_NAMES =
+ Set.of("primarykey", "partitionkey", "sortkey", "joinkey", "groupbykey",
"dimensionkey", "metrickey",
+ "timecolumnkey", "schemaregistrykey");
+
+ /// Key suffixes that name an authentication *mode*, not a credential - a
provider class name, or
+ /// `URL` / `USER_INFO` for the schema registry. They tell an audit reader
how a caller authenticated,
+ /// so they survive [#CREDENTIAL_OR_AUTH], though not [#SENSITIVE_KEY]:
`my.secret.credentials.source`
+ /// is still masked.
+ private static final Set<String> SAFE_KEY_SUFFIXES =
+ Set.of(".credentials.source", ".credentials.provider",
".credentials.providers");
+
+ /// Fields that hold the *name* of a setting in a name/value pair such as
+ /// `{"name": "password", "value": "..."}`. When one names a sensitive key,
the pair's [#PAIR_VALUE_FIELD]
+ /// is masked, since the key-name rules alone would see only the harmless
word `value`.
+ private static final Set<String> PAIR_NAME_FIELDS = Set.of("name", "key");
+ private static final String PAIR_VALUE_FIELD = "value";
+
+ private AuditRedactor() {
+ }
+
+ /// Returns the captured request body with every credential-bearing value
masked. A complete body that is
+ /// not one parseable JSON object or array is reported by size only. A
truncated body keeps its redacted
+ /// JSON prefix, followed by [AuditRequestProcessor#TRUNCATION_MARKER], or
is reported by size if no JSON
+ /// object or array begins it.
+ ///
+ /// @param body the captured body, possibly null
+ /// @return the redacted body, or null if there was nothing to record
+ static String redactBody(AuditRequestProcessor.CapturedBody body) {
+ if (body == null || StringUtils.isBlank(body.getText())) {
+ return null;
+ }
+ try {
+ if (body.isTruncated()) {
+ String prefix = redactTruncatedPrefix(body.getText());
+ if (prefix != null) {
+ return prefix + AuditRequestProcessor.TRUNCATION_MARKER;
+ }
+ } else {
+ JsonNode parsed = parseWholeValue(body.getText());
+ if (parsed != null && parsed.isContainerNode()) {
+ return JsonUtils.objectToString(redactNode(parsed));
+ }
+ }
+ } catch (Exception e) {
+ // Never log the exception itself. Jackson embeds the offending token in
its message, so a body
+ // that is a bare credential would write that credential into the log
this class protects.
+ LOG.debug("Audit payload is not parseable JSON ({}); recording size
only", e.getClass().getSimpleName());
+ }
+ return body.isTruncated()
+ ? String.format("[redacted: payload truncated at %d captured bytes,
not parseable]", body.getCapturedBytes())
+ : String.format("[redacted: unparseable payload, %d bytes]",
body.getCapturedBytes());
+ }
+
+ /// Parses `body` only if it is a single complete JSON value with nothing
following it.
+ ///
+ /// [com.fasterxml.jackson.databind.ObjectReader#readTree(String)] stops at
the first value and
+ /// silently discards whatever follows, because `FAIL_ON_TRAILING_TOKENS` is
off by default. Relying on
+ /// it would record a concatenated payload as a well-formed body with the
remainder quietly dropped.
+ /// Reading through an explicit parser and then demanding the stream be
exhausted rejects that, so it
+ /// falls to the size-only path instead.
+ ///
+ /// @return the parsed value, or null if the body is empty or carries
trailing content
+ private static JsonNode parseWholeValue(String body)
+ throws IOException {
+ try (JsonParser parser = JSON_FACTORY.createParser(body)) {
+ JsonNode node = JsonUtils.DEFAULT_READER.readTree(parser);
+ if (node == null || node.isMissingNode() || parser.nextToken() != null) {
+ return null;
+ }
+ return node;
+ }
+ }
+
+ /// Returns the redacted JSON that precedes the cut in a truncated body,
with the objects and arrays the
+ /// cut left open closed again, or null if the body does not begin with a
JSON object or array.
+ ///
+ /// The prefix is copied token by token rather than parsed as a tree, which
a truncated body can never
+ /// be. A sensitive key is masked as soon as its name is complete, whether
or not its value made it in
+ /// before the cut. Any other name is written only together with its value,
and the value the cut falls
+ /// inside is dropped rather than recorded as if it were complete: a string
or literal missing its end
+ /// fails to parse, and a number that runs up to the cut may be missing
digits, so it is dropped too.
+ private static String redactTruncatedPrefix(String text)
+ throws IOException {
+ StringWriter out = new StringWriter();
+ // The generator is closed first, and closing it writes the end of every
object and array still open.
+ try (JsonParser parser = JSON_FACTORY.createParser(text);
+ JsonGenerator generator = JSON_FACTORY.createGenerator(out)) {
+ JsonToken first = parser.nextToken();
+ if (first != JsonToken.START_OBJECT && first != JsonToken.START_ARRAY) {
+ return null;
+ }
+ try {
+ copyRedacted(parser, generator, first, text.length());
+ } catch (IOException e) {
+ // Expected: the cut leaves the last token incomplete, and everything
before it is already written.
+ }
+ }
+ return out.toString();
+ }
+
+ /// Copies the token stream, masking as [#redactNode(JsonNode)] does. A
name/value pair is recognized only
+ /// when its name field comes before its value field, since nothing is
buffered.
+ private static void copyRedacted(JsonParser parser, JsonGenerator generator,
JsonToken first, int textLength)
+ throws IOException {
+ // Per open object: whether a name field in it has named a sensitive key.
+ Deque<Boolean> sensitivePairs = new ArrayDeque<>();
+ String pendingName = null;
+ JsonToken token = first;
+ while (token != null) {
+ switch (token) {
+ case FIELD_NAME:
+ String name = parser.currentName();
+ boolean pairValue = PAIR_VALUE_FIELD.equals(name) &&
Boolean.TRUE.equals(sensitivePairs.peek());
+ if (isSensitiveKey(name) || pairValue) {
+ generator.writeStringField(name, MASKED_VALUE);
+ JsonToken value = parser.nextToken();
+ if (value == JsonToken.VALUE_STRING &&
PAIR_NAME_FIELDS.contains(name)) {
+ markSensitivePair(sensitivePairs, parser.getText());
+ } else if (value != null && value.isStructStart()) {
+ parser.skipChildren();
+ }
+ } else {
+ pendingName = name;
+ }
+ token = parser.nextToken();
+ break;
+ case START_OBJECT:
+ case START_ARRAY:
+ if (pendingName != null) {
+ generator.writeFieldName(pendingName);
+ pendingName = null;
+ }
+ if (token == JsonToken.START_OBJECT) {
+ sensitivePairs.push(false);
+ }
+ generator.copyCurrentEvent(parser);
+ token = parser.nextToken();
+ break;
+ case END_OBJECT:
+ case END_ARRAY:
+ if (token == JsonToken.END_OBJECT) {
+ sensitivePairs.pop();
+ }
+ generator.copyCurrentEvent(parser);
+ token = parser.nextToken();
+ break;
+ default:
+ if (token.isNumeric() && parser.currentLocation().getCharOffset() >=
textLength) {
+ return;
+ }
+ String text = token == JsonToken.VALUE_STRING ?
redactUrlUserInfo(parser.getText()) : parser.getText();
+ if (token == JsonToken.VALUE_STRING && pendingName != null &&
PAIR_NAME_FIELDS.contains(pendingName)) {
+ markSensitivePair(sensitivePairs, parser.getText());
+ }
+ if (pendingName != null) {
+ generator.writeFieldName(pendingName);
+ pendingName = null;
+ }
+ writeScalar(generator, token, text);
+ token = parser.nextToken();
+ break;
+ }
+ }
+ }
+
+ private static void markSensitivePair(Deque<Boolean> sensitivePairs, String
settingName) {
+ if (isSensitiveKey(settingName) && !sensitivePairs.isEmpty()) {
+ sensitivePairs.pop();
+ sensitivePairs.push(true);
+ }
+ }
+
+ private static void writeScalar(JsonGenerator generator, JsonToken scalar,
String text)
+ throws IOException {
+ switch (scalar) {
+ case VALUE_NUMBER_INT:
+ case VALUE_NUMBER_FLOAT:
+ generator.writeNumber(text);
+ break;
+ case VALUE_TRUE:
+ case VALUE_FALSE:
+ generator.writeBoolean(scalar == JsonToken.VALUE_TRUE);
+ break;
+ case VALUE_NULL:
+ generator.writeNull();
+ break;
+ default:
+ generator.writeString(text);
+ break;
+ }
+ }
+
+ /// Returns a copy of `map` with the values of credential-bearing keys
masked. Multi-valued entries are
+ /// replaced wholesale rather than element-wise, so the number of values is
not disclosed either.
+ static Map<String, Object> redact(Map<String, Object> map) {
+ if (map == null || map.isEmpty()) {
+ return map;
+ }
+ Map<String, Object> redacted = new HashMap<>(map.size());
+ map.forEach((key, value) -> redacted.put(key, isSensitiveKey(key) ?
MASKED_VALUE : redactValue(value)));
+ return redacted;
+ }
+
+ static boolean isSensitiveKey(String key) {
+ if (key == null) {
+ return false;
+ }
+ String lowerKey = key.toLowerCase(Locale.ROOT);
+ if (SENSITIVE_KEY.matcher(lowerKey).matches()) {
+ return true;
+ }
+ if (CREDENTIAL_OR_AUTH.matcher(lowerKey).matches()) {
+ return SAFE_KEY_SUFFIXES.stream().noneMatch(lowerKey::endsWith);
+ }
+ String lastSegment = lowerKey.substring(lowerKey.lastIndexOf('.') + 1);
+ return ENDS_IN_KEY.matcher(lowerKey).matches() &&
!SAFE_KEY_NAMES.contains(lowerKey)
+ && !SAFE_KEY_NAMES.contains(lastSegment);
+ }
+
+ /// Returns `value` with the userinfo of every URL in it replaced by the
mask, keeping each scheme, host
+ /// and path. Returns the value unchanged when it carries no userinfo.
+ ///
+ /// Each `://` preceded by a scheme character starts a URL, so a scheme may
contain colons
+ /// (`jdbc:postgresql://`). Its userinfo runs to the last `@` before
whitespace or the next `://`, so a
+ /// password containing `@` or `/` is masked whole, and each URL of a list
is handled on its own. A URL
+ /// whose path contains `@` has its host masked as well - an over-redaction
accepted for the sake of the
+ /// password cases.
+ ///
+ /// This is a single forward pass rather than a regex, so its cost stays
linear in the length of the
+ /// value, which may be a query parameter or header of the caller's
choosing. A backtracking pattern
+ /// doing the same job took quadratic time on a long run of letters ending
in `@`.
+ static String redactUrlUserInfo(String value) {
+ if (value == null || value.indexOf('@') < 0) {
+ return value;
+ }
+ StringBuilder redacted = null;
+ int copied = 0;
+ int separator = value.indexOf("://");
+ while (separator >= 0) {
+ int start = separator + 3;
+ int next = value.indexOf("://", start);
+ int end = next < 0 ? value.length() : next;
+ for (int i = start; i < end; i++) {
+ if (Character.isWhitespace(value.charAt(i))) {
+ end = i;
+ break;
+ }
+ }
+ if (separator > 0 && isSchemeChar(value.charAt(separator - 1))) {
+ int at = end - 1;
+ while (at >= start && value.charAt(at) != '@') {
+ at--;
+ }
+ if (at >= start) {
+ if (redacted == null) {
+ redacted = new StringBuilder(value.length());
+ }
+ redacted.append(value, copied, start).append(MASKED_VALUE);
+ copied = at;
+ }
+ }
+ separator = next;
+ }
+ return redacted == null ? value : redacted.append(value, copied,
value.length()).toString();
+ }
+
+ private static boolean isSchemeChar(char c) {
+ return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c
<= '9') || c == '+' || c == '.'
+ || c == '-';
+ }
+
+ /// Masks, in place, every value under a sensitive key - wholesale, whether
it is a scalar, an array or
+ /// a subtree - and the value of every name/value pair whose name is a
sensitive key, and strips URL
+ /// userinfo from every remaining text value, array elements included.
+ private static JsonNode redactNode(JsonNode node) {
+ if (node.isObject()) {
+ ObjectNode object = (ObjectNode) node;
+ // Decided before the loop below masks the name field itself, as it does
a field called "key".
+ boolean sensitivePair = PAIR_NAME_FIELDS.stream().map(object::get)
+ .anyMatch(name -> name != null && name.isTextual() &&
isSensitiveKey(name.textValue()));
+ object.fieldNames().forEachRemaining(field -> object.set(field,
+ isSensitiveKey(field) || (sensitivePair &&
PAIR_VALUE_FIELD.equals(field)) ? TextNode.valueOf(MASKED_VALUE)
+ : redactNode(object.get(field))));
+ } else if (node.isArray()) {
+ ArrayNode array = (ArrayNode) node;
+ for (int i = 0; i < array.size(); i++) {
+ array.set(i, redactNode(array.get(i)));
+ }
+ } else if (node.isTextual()) {
+ return TextNode.valueOf(redactUrlUserInfo(node.textValue()));
+ }
+ return node;
+ }
+
+ private static Object redactValue(Object value) {
+ if (value instanceof String) {
+ return redactUrlUserInfo((String) value);
+ }
+ if (value instanceof List) {
+ return ((List<?>) value).stream()
+ .map(element -> element instanceof String ?
redactUrlUserInfo((String) element) : element)
+ .collect(Collectors.toList());
+ }
+ return value;
+ }
+}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditRequestProcessor.java
b/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditRequestProcessor.java
index 1aee5ffc68a..9b3dcce449f 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditRequestProcessor.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/audit/AuditRequestProcessor.java
@@ -25,6 +25,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -44,10 +45,17 @@ import org.slf4j.LoggerFactory;
/// Handles all the complex logic for IP address extraction, user
identification,
/// and request payload capture for audit logging purposes.
/// Uses dynamic configuration to control audit behavior.
+///
+/// Everything captured from the caller - query parameters, allow-listed
headers and the request
+/// body - passes through [AuditRedactor] before it reaches the [AuditEvent],
so that credentials
+/// sent to an audited endpoint are masked rather than written to the audit
log. That step is
+/// unconditional and has no corresponding configuration key. It matches on
key names, so it is a
+/// strong default rather than a guarantee - see the residual gap noted on
[AuditRedactor].
@Singleton
public class AuditRequestProcessor {
private static final Logger LOG =
LoggerFactory.getLogger(AuditRequestProcessor.class);
+ /// Appended to the redacted prefix of a body that was cut at the capture
cap.
static final String TRUNCATION_MARKER = "...[truncated]";
private final AuditConfigManager _configManager;
@@ -153,7 +161,7 @@ public class AuditRequestProcessor {
UriInfo uriInfo = requestContext.getUriInfo();
MultivaluedMap<String, String> queryParams =
uriInfo.getQueryParameters();
if (!queryParams.isEmpty()) {
- payload.setQueryParameters(toMap(queryParams));
+ payload.setQueryParameters(AuditRedactor.redact(toMap(queryParams)));
}
final AuditConfig config = _configManager.getCurrentConfig();
@@ -164,16 +172,13 @@ public class AuditRequestProcessor {
if (!allHeaders.isEmpty()) {
Map<String, Object> filteredHeaders = toMap(allHeaders,
allowedHeaders);
if (!filteredHeaders.isEmpty()) {
- payload.setHeaders(filteredHeaders);
+ payload.setHeaders(AuditRedactor.redact(filteredHeaders));
}
}
}
if (config.isCaptureRequestPayload() && requestContext.hasEntity()) {
- String requestBody = readRequestBody(requestContext,
config.getMaxPayloadSize());
- if (StringUtils.isNotBlank(requestBody)) {
- payload.setBody(requestBody);
- }
+
payload.setBody(AuditRedactor.redactBody(readRequestBody(requestContext,
config.getMaxPayloadSize())));
}
if (payload.getQueryParameters() == null && payload.getBody() == null &&
payload.getHeaders() == null) {
@@ -215,11 +220,12 @@ public class AuditRequestProcessor {
/// Limits the amount of data read based on configuration.
///
/// @param requestContext the request context
- /// @param maxPayloadSize maximum bytes to read from the request body
- /// @return the request body as string (potentially truncated)
+ /// @param maxPayloadSize maximum bytes to capture from the request body
+ /// @return the captured body, or null if there is none
@VisibleForTesting
- String readRequestBody(ContainerRequestContext requestContext, int
maxPayloadSize) {
- if (!requestContext.hasEntity()) {
+ CapturedBody readRequestBody(ContainerRequestContext requestContext, int
maxPayloadSize) {
+ // With no room to capture anything, the stream is left alone, and nothing
is reported as truncated.
+ if (!requestContext.hasEntity() || maxPayloadSize <= 0) {
return null;
}
@@ -234,8 +240,10 @@ public class AuditRequestProcessor {
requestContext.setEntityStream(bufferedStream);
bufferedStream.mark(maxPayloadSize + 1);
- final InputStream limitedStream = ByteStreams.limit(bufferedStream,
maxPayloadSize);
- final byte[] capturedBytes = ByteStreams.toByteArray(limitedStream);
+ // One byte beyond the cap is read only to learn whether the body
continues past it, so that a body of
+ // exactly maxPayloadSize bytes is not mistaken for a truncated one.
+ final InputStream limitedStream = ByteStreams.limit(bufferedStream,
maxPayloadSize + 1L);
+ final byte[] readBytes = ByteStreams.toByteArray(limitedStream);
try {
bufferedStream.reset();
@@ -244,17 +252,45 @@ public class AuditRequestProcessor {
LOG.error("Failed to reset stream for downstream consumers",
resetException);
}
- if (capturedBytes.length > 0) {
- String requestBody = new String(capturedBytes, StandardCharsets.UTF_8);
- if (capturedBytes.length >= maxPayloadSize) {
- requestBody += TRUNCATION_MARKER;
+ if (readBytes.length > 0) {
+ boolean truncated = readBytes.length > maxPayloadSize;
+ byte[] capturedBytes = truncated ? Arrays.copyOf(readBytes,
maxPayloadSize) : readBytes;
+ if (truncated) {
_auditMetrics.addMeteredGlobalValue(AuditMetrics.AuditMeter.AUDIT_REQUEST_PAYLOAD_TRUNCATED,
1L);
}
- return requestBody;
+ return new CapturedBody(new String(capturedBytes,
StandardCharsets.UTF_8), capturedBytes.length, truncated);
}
} catch (IOException e) {
LOG.warn("Failed to capture request body", e);
}
return null;
}
+
+ /// A request body as captured for the audit log. Truncation is carried as a
flag, and the size as the
+ /// number of bytes actually captured, so neither has to be re-derived from
the decoded text.
+ static final class CapturedBody {
+ private final String _text;
+ private final int _capturedBytes;
+ private final boolean _truncated;
+
+ CapturedBody(String text, int capturedBytes, boolean truncated) {
+ _text = text;
+ _capturedBytes = capturedBytes;
+ _truncated = truncated;
+ }
+
+ /// The captured bytes decoded as UTF-8. A cut through a multi-byte
character decodes to U+FFFD.
+ String getText() {
+ return _text;
+ }
+
+ int getCapturedBytes() {
+ return _capturedBytes;
+ }
+
+ /// True when the body continued past the capture cap.
+ boolean isTruncated() {
+ return _truncated;
+ }
+ }
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/audit/AuditRedactorTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/audit/AuditRedactorTest.java
new file mode 100644
index 00000000000..534abb21f44
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/common/audit/AuditRedactorTest.java
@@ -0,0 +1,377 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.audit;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/// Unit tests for [AuditRedactor].
+public class AuditRedactorTest {
+
+ private static final String SECRET = "AKIAIOSFODNN7EXAMPLE-SECRET";
+
+ @DataProvider(name = "sensitiveKeys")
+ public Object[][] sensitiveKeys() {
+ return new Object[][]{
+ // Batch-ingestion S3 credentials, as sent in a connection's params.
+ {"input.fs.prop.secretKey"}, {"input.fs.prop.accessKey"},
+ // AWS / S3.
+ {"accessKey"}, {"accessKeyId"}, {"secretKey"},
{"aws.secret.access.key"}, {"sessionToken"},
+ {"catalog.s3.auth.storage.secretKey"},
+ // Kafka / SASL.
+ {"sasl.jaas.config"}, {"ssl.keystore.password"},
{"ssl.truststore.password"},
+ // Kafka inline PEM material, which no *private*key* pattern reaches.
+ {"ssl.keystore.key"}, {"ssl.keystore.certificate.chain"},
+ // Confluent schema registry: a literal "user:password" pair. "auth",
not "authorization".
+ {"basic.auth.user.info"},
{"stream.kafka.decoder.prop.basic.auth.user.info"},
+ // Session material an operator may allow-list as a header.
+ {"Cookie"}, {"Set-Cookie"},
+ // Warehouses and catalogs.
+ {"snowflake.password"}, {"restOauthClientSecret"}, {"privateKey"},
{"passphrase"},
+ // Hadoop S3A, GCS, BigQuery, Delta Lake, Pulsar and tier-storage SSE
keys.
+ {"fs.s3a.access.key"}, {"fs.s3a.secret.key"},
{"fs.gs.auth.service.account.json.keyfile"},
+ {"gcpKey"}, {"gcpKeyPath"}, {"sql.bigquery.jsonKey"},
{"delta.ingestion.table.gcs.keyFile"},
+ {"delta.ingestion.table.gcs.keyContentBase64"},
{"stream.pulsar.credsFilePath"},
+ {"stream.pulsar.authenticationToken"}, {"sseKmsKeyId"},
{"private_key_id"},
+ // Confluent Cloud API key, and its schema registry user info under
any prefix.
+ {"stream.confluent.key"}, {"stream.confluent.secret"}, {"userInfo"},
+ // Iceberg REST catalog and Azure.
+ {"restAccessKeyId"}, {"storageSecretAccessKey"},
{"storageSessionToken"}, {"clientSecret"},
+ {"sasToken"}, {"proxyPassword"},
+ // Headers that carry a credential.
+ {"Proxy-Authorization"}, {"X-Auth-Token"}, {"X-Startree-Token"},
+ // Dotted words, as in Hadoop-style property names.
+ {"api.key.value"}, {"private.key.pem"}, {"encryption.key.material"},
{"signing.key"},
+ // A credential-bearing prefix is not exempted by a structural last
segment or an auth-mode suffix.
+ {"secret.sortKey"}, {"auth.partitionKey"},
{"stream.confluent.secret.joinKey"},
+ {"my.secret.credentials.source"},
{"kafka.sasl.password.credentials.provider"},
+ // Short password names, Azure connection strings, session ids and SAS
signatures.
+ {"dbPass"}, {"ssl.keypass"}, {"storepass"}, {"connectionString"},
{"azure.storage.connection.string"},
+ {"sessionId"}, {"JSESSIONID"}, {"sig"}, {"x-ms-sig"},
+ // Generic.
+ {"password"}, {"PASSWORD"}, {"myPassword"}, {"passwd"}, {"pwd"},
{"credentials"}, {"apiKey"},
+ {"api_key"}, {"keytab"}, {"token"}, {"authorization"}, {"signature"},
{"keyPath"}
+ };
+ }
+
+ @DataProvider(name = "benignKeys")
+ public Object[][] benignKeys() {
+ return new Object[][]{
+ {"tableName"}, {"primaryKeyColumns"}, {"partitionKey"},
{"sortedColumn"}, {"user"}, {"username"},
+ {"principal"}, {"endpoint"}, {"region"}, {"bucketName"}, {"roleArn"},
{"design"}, {"bypassCache"},
+ // Structural names that merely end in "key".
+ {"primaryKey"}, {"sortKey"}, {"joinKey"}, {"groupByKey"},
{"dimensionKey"}, {"metricKey"},
+ {"timeColumnKey"}, {"schemaRegistryKey"}, {"segment.partitionKey"},
+ // Authentication modes, which say how a caller authenticated without
being a credential.
+ {"stream.kafka.decoder.prop.basic.auth.credentials.source"},
{"fs.s3a.aws.credentials.provider"}
+ };
+ }
+
+ @Test(dataProvider = "sensitiveKeys")
+ public void testSensitiveKeysAreRedacted(String key) {
+ assertThat(AuditRedactor.isSensitiveKey(key)).as(key).isTrue();
+ assertThat(complete(String.format("{\"%s\":\"%s\"}", key,
SECRET))).doesNotContain(SECRET);
+ }
+
+ @Test(dataProvider = "benignKeys")
+ public void testBenignKeysAreKept(String key) {
+ assertThat(AuditRedactor.isSensitiveKey(key)).as(key).isFalse();
+ assertThat(complete(String.format("{\"%s\":\"visible\"}",
key))).contains("visible");
+ }
+
+ @Test
+ public void testRedactsNestedObject() {
+ String body =
"{\"connection\":{\"type\":\"CUSTOM_BATCH\",\"params\":{\"input.fs.prop.secretKey\":\""
+ SECRET
+ + "\",\"input.fs.prop.accessKey\":\"" + SECRET +
"\",\"input.fs.className\":\"S3PinotFS\"}}}";
+
+ String redacted = complete(body);
+
+ assertThat(redacted).doesNotContain(SECRET);
+ assertThat(redacted).contains(AuditRedactor.MASKED_VALUE);
+ // Non-credential context survives, which is the whole point of keeping
the body at all.
+ assertThat(redacted).contains("CUSTOM_BATCH", "S3PinotFS",
"input.fs.prop.secretKey");
+ }
+
+ @Test
+ public void testRedactsInsideArrays() {
+ // Kafka credentials live in streamConfigMaps, a list of maps.
+ String body =
"{\"ingestionConfig\":{\"streamIngestionConfig\":{\"streamConfigMaps\":"
+ + "[{\"streamType\":\"kafka\",\"sasl.jaas.config\":\"" + SECRET +
"\"}]}}}";
+
+ String redacted = complete(body);
+
+ assertThat(redacted).doesNotContain(SECRET);
+ assertThat(redacted).contains("kafka");
+ }
+
+ @Test
+ public void testRedactsTopLevelArray() {
+ String redacted = complete("[{\"password\":\"" + SECRET + "\"}]");
+ assertThat(redacted).doesNotContain(SECRET);
+ }
+
+ @Test
+ public void testMasksValueOfSensitiveNameValuePair() {
+ String redacted =
complete("{\"props\":[{\"key\":\"fs.s3a.secret.key\",\"value\":\"" + SECRET +
"\"},"
+ + "{\"value\":\"" + SECRET +
"\",\"name\":\"password\"},{\"name\":\"region\",\"value\":\"us-east-1\"}]}");
+
+ assertThat(redacted).doesNotContain(SECRET);
+ // The name of the setting stays readable, and so does the value of a pair
that is not sensitive.
+ assertThat(redacted).contains("\"name\":\"password\"",
"\"value\":\"us-east-1\"");
+ }
+
+ @Test
+ public void testTruncatedBodyMasksValueOfSensitiveNameValuePair() {
+ String redacted = truncated("[{\"name\":\"password\",\"value\":\"" +
SECRET + "\"},{\"key\":\"accessKey\","
+ + "\"value\":\"" + SECRET + "\"},{\"name\":\"region\",\"value\":\"us");
+
+
assertThat(redacted).isEqualTo("[{\"name\":\"password\",\"value\":\"*****\"},"
+ + "{\"key\":\"*****\",\"value\":\"*****\"},{\"name\":\"region\"}]" +
AuditRequestProcessor.TRUNCATION_MARKER);
+ }
+
+ @Test
+ public void testSensitiveKeyMasksWholeSubtree() {
+ String redacted = complete(
+ "{\"credentials\":{\"user\":\"svc\",\"value\":\"" + SECRET +
"\"},\"region\":\"us-east-1\"}");
+
+ assertThat(redacted).doesNotContain(SECRET).doesNotContain("svc");
+ assertThat(redacted).contains("us-east-1");
+ }
+
+ @Test
+ public void testStripsUrlUserInfoUnderBenignKey() {
+ String redacted = complete(
+ "{\"schema.registry.url\":\"https://svc:"; + SECRET +
"@registry.example.com:8081/path\"}");
+
+ assertThat(redacted).doesNotContain(SECRET).doesNotContain("svc:");
+ assertThat(redacted).contains("https://"; + AuditRedactor.MASKED_VALUE +
"@registry.example.com:8081/path");
+ }
+
+ @Test
+ public void testStripsUrlUserInfoInsideArrays() {
+ String redacted = complete("{\"urls\":[\"s3://AKIA:" + SECRET +
"@bucket/key\"]}");
+
+ assertThat(redacted).doesNotContain(SECRET);
+ assertThat(redacted).contains("@bucket/key");
+ }
+
+ @DataProvider(name = "urlsWithUserInfo")
+ public Object[][] urlsWithUserInfo() {
+ String mask = AuditRedactor.MASKED_VALUE;
+ return new Object[][]{
+ // A password containing '@' or '/' is masked whole.
+ {"https://svc:p@ss@registry:8081";, "https://"; + mask +
"@registry:8081"},
+ {"https://svc:pa/ss@host/path";, "https://"; + mask + "@host/path"},
+ // A scheme with a colon, as in JDBC URLs.
+ {"jdbc:postgresql://u:pw@db:5432/x", "jdbc:postgresql://" + mask +
"@db:5432/x"},
+ // Every URL of a list, and a URL that is not the whole value.
+ {"https://u1:p1@h1,https://u2:p2@h2";, "https://"; + mask +
"@h1,https://"; + mask + "@h2"},
+ {"--uri https://u:p@h --region x", "--uri https://"; + mask + "@h
--region x"}
+ };
+ }
+
+ @Test(dataProvider = "urlsWithUserInfo")
+ public void testStripsUrlUserInfo(String value, String expected) {
+ assertThat(AuditRedactor.redactUrlUserInfo(value)).isEqualTo(expected);
+ }
+
+ @Test(timeOut = 2000)
+ public void testUrlUserInfoScanIsLinear() {
+ // A backtracking pattern took seconds on these; each is a 64 KB value a
caller could send as a header.
+ String letters = "a".repeat(65536) + "@";
+ assertThat(AuditRedactor.redactUrlUserInfo(letters)).isEqualTo(letters);
+ String separators = "a://".repeat(16384) + "@";
+ assertThat(AuditRedactor.redactUrlUserInfo(separators)).endsWith("a://" +
AuditRedactor.MASKED_VALUE + "@");
+ String noUserInfo = "https://host/"; + "x@".repeat(32768);
+
assertThat(AuditRedactor.redactUrlUserInfo(noUserInfo)).startsWith("https://"; +
AuditRedactor.MASKED_VALUE + "@");
+ }
+
+ @Test
+ public void testValueWithoutUrlUserInfoIsKept() {
+
assertThat(AuditRedactor.redactUrlUserInfo("https://host:8081/a";)).isEqualTo("https://host:8081/a";);
+
assertThat(AuditRedactor.redactUrlUserInfo("[email protected]")).isEqualTo("[email protected]");
+ }
+
+ @Test
+ public void testUrlWithAtSignInPathMasksTheHostToo() {
+ // Deliberate over-redaction: the userinfo match runs to the last '@', so
that '@' and '/' in passwords
+ // are covered, and a path containing '@' is indistinguishable from that.
+ assertThat(AuditRedactor.redactUrlUserInfo("https://host:8081/a@b";))
+ .isEqualTo("https://"; + AuditRedactor.MASKED_VALUE + "@b");
+ }
+
+ @Test
+ public void testRootScalarRecordedBySizeOnly() {
+ // A bare value has no key to judge it by, so it could be a credential on
its own.
+ String body = "\"" + SECRET + "\"";
+
+ assertThat(complete(body)).isEqualTo("[redacted: unparseable payload, " +
body.length()
+ + " bytes]");
+ assertThat(complete("12345")).isEqualTo("[redacted: unparseable payload, 5
bytes]");
+ }
+
+ @Test
+ public void testUnparseableBodyRecordedBySizeOnly() {
+ String body = "accessKey=" + SECRET + "®ion=us-west-2";
+
+ String redacted = complete(body);
+
+ assertThat(redacted).doesNotContain(SECRET);
+ assertThat(redacted).doesNotContain("us-west-2");
+ assertThat(redacted).isEqualTo("[redacted: unparseable payload, " +
body.length() + " bytes]");
+ }
+
+ @Test
+ public void testTruncatedBodyKeepsRedactedPrefix() {
+ String body =
"{\"tableName\":\"t\",\"streamConfigs\":{\"sasl.jaas.config\":\"" + SECRET
+ + "\",\"url\":\"https://svc:"; + SECRET +
"@registry:8081\",\"stream.kafka.topic.name\":\"to";
+
+ // The cut falls inside the topic name, which is dropped; the objects it
left open are closed.
+
assertThat(truncated(body)).isEqualTo("{\"tableName\":\"t\",\"streamConfigs\":{\"sasl.jaas.config\":\"*****\","
+ + "\"url\":\"https://*****@registry:8081\"}}"; +
AuditRequestProcessor.TRUNCATION_MARKER);
+ }
+
+ @Test
+ public void testTruncatedInsideSensitiveValueKeepsTheMask() {
+ // The key is complete, so it is masked even though its value never
arrived in full.
+ assertThat(truncated("{\"a\":1,\"password\":\"" + SECRET.substring(0, 5)))
+ .isEqualTo("{\"a\":1,\"password\":\"*****\"}" +
AuditRequestProcessor.TRUNCATION_MARKER);
+ assertThat(truncated("{\"credentials\":{\"user\":\"svc\",\"value\":\"" +
SECRET.substring(0, 5)))
+ .isEqualTo("{\"credentials\":\"*****\"}" +
AuditRequestProcessor.TRUNCATION_MARKER);
+ }
+
+ @Test
+ public void testTruncatedValueIsDroppedNotShortened() {
+ // A number cut short would otherwise be recorded as a different, complete
number.
+ assertThat(truncated("{\"a\":1,\"b\":12")).isEqualTo("{\"a\":1}" +
AuditRequestProcessor.TRUNCATION_MARKER);
+ // A complete value just before the cut is kept.
+ assertThat(truncated("[\"x\",12,tru")).isEqualTo("[\"x\",12]" +
AuditRequestProcessor.TRUNCATION_MARKER);
+ // A key cut short is dropped together with its value.
+ assertThat(truncated("{\"a\":1,\"passw")).isEqualTo("{\"a\":1}" +
AuditRequestProcessor.TRUNCATION_MARKER);
+ }
+
+ @Test
+ public void testTruncatedAfterCompleteValueKeepsTheMarker() {
+ assertThat(truncated("{\"a\":1}")).isEqualTo("{\"a\":1}" +
AuditRequestProcessor.TRUNCATION_MARKER);
+ }
+
+ @Test
+ public void testTruncatedNonJsonRecordedBySizeOnly() {
+ String body = "accessKey=" + SECRET + "®ion=us";
+
+ assertThat(truncated(body)).isEqualTo(
+ "[redacted: payload truncated at " + body.length() + " captured bytes,
not parseable]");
+ // A bare scalar has no key to judge it by, truncated or not.
+ assertThat(truncated("\"" + SECRET)).startsWith("[redacted: payload
truncated at");
+ }
+
+ @Test
+ public void testConcatenatedJsonIsNotSilentlyTruncated() {
+ // Trailing content would otherwise be dropped, hiding the second document
and its secret.
+ String body = "{\"a\":1}{\"password\":\"" + SECRET + "\"}";
+
+ String redacted = complete(body);
+
+ assertThat(redacted).doesNotContain(SECRET);
+ assertThat(redacted).isEqualTo("[redacted: unparseable payload, " +
body.length() + " bytes]");
+ }
+
+ @Test
+ public void testSizeIsCountedInBytesNotCharacters() {
+ String body = "π=secret";
+ assertThat(complete(body)).isEqualTo("[redacted: unparseable payload, 11
bytes]");
+ }
+
+ @Test
+ public void testNullEmptyAndBlankBody() {
+ assertThat(AuditRedactor.redactBody(null)).isNull();
+ assertThat(complete("")).isNull();
+ // Whitespace parses to a MissingNode, which would otherwise serialize to
the literal string "null".
+ assertThat(complete(" ")).isNull();
+ }
+
+ @Test
+ public void testSizeIsTheCapturedByteCount() {
+ // The size comes from the capture, not from re-encoding a text whose last
character may have been split.
+ assertThat(AuditRedactor.redactBody(new
AuditRequestProcessor.CapturedBody("abc\uFFFD", 5, false)))
+ .isEqualTo("[redacted: unparseable payload, 5 bytes]");
+ }
+
+ @Test
+ public void testRedactMapMasksValuesAndPreservesOthers() {
+ Map<String, Object> params = new HashMap<>();
+ params.put("tableName", "myTable");
+ params.put("accessToken", SECRET);
+ params.put("tokens", Arrays.asList(SECRET, SECRET));
+
+ Map<String, Object> redacted = AuditRedactor.redact(params);
+
+ assertThat(redacted).containsEntry("tableName", "myTable");
+ assertThat(redacted).containsEntry("accessToken",
AuditRedactor.MASKED_VALUE);
+ // Multi-valued entries are replaced wholesale, so the value count is not
disclosed either.
+ assertThat(redacted).containsEntry("tokens", AuditRedactor.MASKED_VALUE);
+ }
+
+ @Test
+ public void testRedactMapStripsUrlUserInfo() {
+ Map<String, Object> params = new HashMap<>();
+ params.put("uri", "https://svc:"; + SECRET + "@host/path");
+ params.put("uris", Arrays.asList("https://svc:"; + SECRET + "@host/a",
"https://host/b";));
+
+ Map<String, Object> redacted = AuditRedactor.redact(params);
+
+ assertThat(redacted.toString()).doesNotContain(SECRET);
+ assertThat(redacted).containsEntry("uri", "https://"; +
AuditRedactor.MASKED_VALUE + "@host/path");
+ assertThat(redacted).containsEntry("uris",
+ Arrays.asList("https://"; + AuditRedactor.MASKED_VALUE + "@host/a",
"https://host/b";));
+ }
+
+ @Test
+ public void testRedactMapDoesNotMutateInput() {
+ Map<String, Object> params = new HashMap<>();
+ params.put("password", SECRET);
+
+ AuditRedactor.redact(params);
+
+ assertThat(params).containsEntry("password", SECRET);
+ }
+
+ @Test
+ public void testRedactNullAndEmptyMap() {
+ assertThat(AuditRedactor.redact(null)).isNull();
+ assertThat(AuditRedactor.redact(new HashMap<>())).isEmpty();
+ }
+
+ private static String complete(String body) {
+ return AuditRedactor.redactBody(body == null ? null
+ : new AuditRequestProcessor.CapturedBody(body,
body.getBytes(StandardCharsets.UTF_8).length, false));
+ }
+
+ private static String truncated(String body) {
+ return AuditRedactor.redactBody(
+ new AuditRequestProcessor.CapturedBody(body,
body.getBytes(StandardCharsets.UTF_8).length, true));
+ }
+}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/audit/AuditRequestProcessorTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/audit/AuditRequestProcessorTest.java
index 44070e6cfc9..1348b8fe5a8 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/audit/AuditRequestProcessorTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/audit/AuditRequestProcessorTest.java
@@ -143,8 +143,8 @@ public class AuditRequestProcessorTest {
_defaultConfig.setCaptureRequestHeaders("Content-Type,X-Custom-Header,Authorization,X-Password");
MultivaluedMap<String, String> headers =
createHeaders("Content-Type", "application/json", "X-Custom-Header",
"custom-value", "Authorization",
- "Bearer token123", // Should be filtered out
- "X-Password", "secret123" // Should be filtered out
+ "Bearer token123", // Captured, but redacted
+ "X-Password", "secret123" // Captured, but redacted
);
when(_requestContext.getUriInfo()).thenReturn(_uriInfo);
@@ -162,8 +162,9 @@ public class AuditRequestProcessorTest {
Map<String, Object> capturedHeaders = payload.getHeaders();
assertThat(capturedHeaders).containsEntry("Content-Type",
"application/json");
assertThat(capturedHeaders).containsEntry("X-Custom-Header",
"custom-value");
- assertThat(capturedHeaders).containsEntry("Authorization", "Bearer
token123");
- assertThat(capturedHeaders).containsEntry("X-Password", "secret123");
+ // Allow-listing a credential-bearing header records that it was sent, not
what it said.
+ assertThat(capturedHeaders).containsEntry("Authorization",
AuditRedactor.MASKED_VALUE);
+ assertThat(capturedHeaders).containsEntry("X-Password",
AuditRedactor.MASKED_VALUE);
}
@Test
@@ -193,8 +194,7 @@ public class AuditRequestProcessorTest {
MultivaluedMap<String, String> headers =
createHeaders("authorization", "Bearer token123", "x-auth-token",
"token456", "password-header", "pass123",
"api-secret", "secret789", "x-api-key", "key123",
- // Should be filtered (contains 'secret' logic might not catch
this)
- "content-type", "application/json" // Should be kept
+ "content-type", "application/json" // The only non-credential
header here
);
when(_requestContext.getUriInfo()).thenReturn(_uriInfo);
@@ -210,11 +210,11 @@ public class AuditRequestProcessorTest {
assertThat(payload).isNotNull();
Map<String, Object> capturedHeaders = payload.getHeaders();
assertThat(capturedHeaders).containsEntry("content-type",
"application/json");
- assertThat(capturedHeaders).containsEntry("authorization", "Bearer
token123");
- assertThat(capturedHeaders).containsEntry("x-auth-token", "token456");
- assertThat(capturedHeaders).containsEntry("password-header", "pass123");
- assertThat(capturedHeaders).containsEntry("api-secret", "secret789");
- assertThat(capturedHeaders).containsEntry("x-api-key", "key123");
+ assertThat(capturedHeaders).containsEntry("authorization",
AuditRedactor.MASKED_VALUE);
+ assertThat(capturedHeaders).containsEntry("x-auth-token",
AuditRedactor.MASKED_VALUE);
+ assertThat(capturedHeaders).containsEntry("password-header",
AuditRedactor.MASKED_VALUE);
+ assertThat(capturedHeaders).containsEntry("api-secret",
AuditRedactor.MASKED_VALUE);
+ assertThat(capturedHeaders).containsEntry("x-api-key",
AuditRedactor.MASKED_VALUE);
}
@Test
@@ -375,9 +375,10 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(originalStream);
- String result = _processor.readRequestBody(_requestContext, 100);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 100);
- assertThat(result).isEqualTo(testData);
+ assertThat(result.getText()).isEqualTo(testData);
+ assertThat(result.isTruncated()).isFalse();
ArgumentCaptor<InputStream> streamCaptor =
ArgumentCaptor.forClass(InputStream.class);
verify(_requestContext).setEntityStream(streamCaptor.capture());
@@ -399,9 +400,11 @@ public class AuditRequestProcessorTest {
when(_requestContext.getEntityStream()).thenReturn(originalStream);
int maxSize = 10;
- String result = _processor.readRequestBody(_requestContext, maxSize);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, maxSize);
- assertThat(result).isEqualTo("This is a " +
AuditRequestProcessor.TRUNCATION_MARKER);
+ assertThat(result.getText()).isEqualTo("This is a ");
+ assertThat(result.getCapturedBytes()).isEqualTo(maxSize);
+ assertThat(result.isTruncated()).isTrue();
// Verify stream is still set for downstream
verify(_requestContext).setEntityStream(any(InputStream.class));
@@ -416,9 +419,10 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(originalStream);
- String result = _processor.readRequestBody(_requestContext, 100);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 100);
- assertThat(result).isEqualTo(testData);
+ assertThat(result.getText()).isEqualTo(testData);
+ assertThat(result.isTruncated()).isFalse();
verify(_requestContext).setEntityStream(any(InputStream.class));
}
@@ -426,7 +430,7 @@ public class AuditRequestProcessorTest {
public void testReadRequestBodyNoEntity() {
when(_requestContext.hasEntity()).thenReturn(false);
- String result = _processor.readRequestBody(_requestContext, 100);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 100);
assertThat(result).isNull();
verify(_requestContext, times(0)).getEntityStream();
@@ -437,7 +441,7 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(null);
- String result = _processor.readRequestBody(_requestContext, 100);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 100);
assertThat(result).isNull();
verify(_requestContext, times(0)).setEntityStream(any());
@@ -450,7 +454,7 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(emptyStream);
- String result = _processor.readRequestBody(_requestContext, 100);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 100);
assertThat(result).isNull();
verify(_requestContext).setEntityStream(any(InputStream.class));
@@ -465,7 +469,7 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(failingStream);
- String result = _processor.readRequestBody(_requestContext, 100);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 100);
assertThat(result).isNull();
// BufferedInputStream wrapper should be set even if read fails
@@ -480,9 +484,11 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(stream);
- String result = _processor.readRequestBody(_requestContext, 10);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 10);
- assertThat(result).isEqualTo(exactData +
AuditRequestProcessor.TRUNCATION_MARKER);
+ assertThat(result.getText()).isEqualTo(exactData);
+ // A body that ends exactly at the cap was captured whole, so it is not
truncated.
+ assertThat(result.isTruncated()).isFalse();
verify(_requestContext).setEntityStream(any(InputStream.class));
}
@@ -494,9 +500,10 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(stream);
- String result = _processor.readRequestBody(_requestContext, 10);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 10);
- assertThat(result).isEqualTo(data); // No truncation marker
+ assertThat(result.getText()).isEqualTo(data);
+ assertThat(result.isTruncated()).isFalse();
verify(_requestContext).setEntityStream(any(InputStream.class));
}
@@ -508,9 +515,9 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(stream);
- String result = _processor.readRequestBody(_requestContext, 100);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 100);
- assertThat(result).isEqualTo(utf8Data);
+ assertThat(result.getText()).isEqualTo(utf8Data);
verify(_requestContext).setEntityStream(any(InputStream.class));
}
@@ -524,11 +531,13 @@ public class AuditRequestProcessorTest {
when(_requestContext.getEntityStream()).thenReturn(stream);
// Truncate at 8 bytes (will cut in middle of "δΈ")
- String result = _processor.readRequestBody(_requestContext, 8);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 8);
// The result should handle the truncation gracefully
- assertThat(result).isNotNull();
- assertThat(result).endsWith(AuditRequestProcessor.TRUNCATION_MARKER);
+ assertThat(result.getText()).startsWith("Hello ");
+ assertThat(result.isTruncated()).isTrue();
+ // The size is the bytes captured, not the re-encoded length of a text
whose last character was split.
+ assertThat(result.getCapturedBytes()).isEqualTo(8);
verify(_requestContext).setEntityStream(any(InputStream.class));
}
@@ -543,11 +552,12 @@ public class AuditRequestProcessorTest {
when(_requestContext.getEntityStream()).thenReturn(stream);
int maxSize = 1000;
- String result = _processor.readRequestBody(_requestContext, maxSize);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, maxSize);
- assertThat(result).hasSize(maxSize +
AuditRequestProcessor.TRUNCATION_MARKER.length());
- assertThat(result).startsWith("AAAAAAAAAA");
- assertThat(result).endsWith(AuditRequestProcessor.TRUNCATION_MARKER);
+ assertThat(result.getText()).hasSize(maxSize);
+ assertThat(result.getText()).startsWith("AAAAAAAAAA");
+ assertThat(result.getCapturedBytes()).isEqualTo(maxSize);
+ assertThat(result.isTruncated()).isTrue();
verify(_requestContext).setEntityStream(any(InputStream.class));
}
@@ -560,9 +570,9 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(stream);
- String result = _processor.readRequestBody(_requestContext, 1000);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 1000);
- assertThat(result).isEqualTo(jsonPayload);
+ assertThat(result.getText()).isEqualTo(jsonPayload);
// Verify the stream can still be read downstream
ArgumentCaptor<InputStream> streamCaptor =
ArgumentCaptor.forClass(InputStream.class);
@@ -592,11 +602,11 @@ public class AuditRequestProcessorTest {
when(_requestContext.getEntityStream()).thenReturn(stream);
int maxSize = 50;
- String result = _processor.readRequestBody(_requestContext, maxSize);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, maxSize);
- assertThat(result).hasSize(maxSize +
AuditRequestProcessor.TRUNCATION_MARKER.length());
- assertThat(result).startsWith("{\"users\":[{\"id\":0");
- assertThat(result).endsWith(AuditRequestProcessor.TRUNCATION_MARKER);
+ assertThat(result.getText()).hasSize(maxSize);
+ assertThat(result.getText()).startsWith("{\"users\":[{\"id\":0");
+ assertThat(result.isTruncated()).isTrue();
// Even though truncated, it should still be valid for downstream
verify(_requestContext).setEntityStream(any(InputStream.class));
@@ -612,9 +622,9 @@ public class AuditRequestProcessorTest {
when(_requestContext.hasEntity()).thenReturn(true);
when(_requestContext.getEntityStream()).thenReturn(stream);
- String result = _processor.readRequestBody(_requestContext, 1000);
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 1000);
- assertThat(result).isEqualTo(nestedJson);
+ assertThat(result.getText()).isEqualTo(nestedJson);
// Verify stream preservation
ArgumentCaptor<InputStream> streamCaptor =
ArgumentCaptor.forClass(InputStream.class);
@@ -624,4 +634,174 @@ public class AuditRequestProcessorTest {
byte[] readBytes = ByteStreams.toByteArray(capturedStream);
assertThat(new String(readBytes,
StandardCharsets.UTF_8)).isEqualTo(nestedJson);
}
+
+ @Test
+ public void testReadRequestBodyWithZeroCapCapturesNothing() {
+ when(_requestContext.hasEntity()).thenReturn(true);
+ when(_requestContext.getEntityStream()).thenReturn(
+ new ByteArrayInputStream("body".getBytes(StandardCharsets.UTF_8)));
+
+ assertThat(_processor.readRequestBody(_requestContext, 0)).isNull();
+ // The stream is left untouched, rather than read one byte past a cap of
zero and reported as truncated.
+ verify(_requestContext, times(0)).setEntityStream(any());
+ }
+
+ @Test
+ public void testReadRequestBodyOverCapLeavesWholeBodyForDownstream()
+ throws IOException {
+ // Reading one byte past the cap must not invalidate the mark, or the
resource method loses the body.
+ String body = "0123456789-the-rest-of-the-body";
+ when(_requestContext.hasEntity()).thenReturn(true);
+ when(_requestContext.getEntityStream()).thenReturn(new
ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)));
+
+ AuditRequestProcessor.CapturedBody result =
_processor.readRequestBody(_requestContext, 10);
+
+ assertThat(result.getText()).isEqualTo("0123456789");
+ assertThat(result.isTruncated()).isTrue();
+ ArgumentCaptor<InputStream> streamCaptor =
ArgumentCaptor.forClass(InputStream.class);
+ verify(_requestContext).setEntityStream(streamCaptor.capture());
+ assertThat(new String(ByteStreams.toByteArray(streamCaptor.getValue()),
StandardCharsets.UTF_8)).isEqualTo(body);
+ }
+
+ // Redaction tests: nothing captured from the caller may carry credential
material.
+
+ private static final String AWS_SECRET =
"wJalrXUtnFEMI-K7MDENG-bPxRfiCYEXAMPLEKEY";
+
+ @Test
+ public void testCapturedBodyRedactsCredentials() {
+ _defaultConfig.setCaptureRequestPayload(true);
+ String body =
"{\"connection\":{\"type\":\"CUSTOM_BATCH\",\"params\":{\"input.fs.prop.accessKey\":\""
+ + AWS_SECRET + "\",\"input.fs.prop.secretKey\":\"" + AWS_SECRET +
"\"}}}";
+
+ when(_requestContext.getUriInfo()).thenReturn(_uriInfo);
+ when(_uriInfo.getQueryParameters()).thenReturn(new MultivaluedHashMap<>());
+ when(_uriInfo.getPath()).thenReturn("/connections/browse");
+ when(_requestContext.getMethod()).thenReturn("POST");
+ when(_requestContext.getHeaders()).thenReturn(new MultivaluedHashMap<>());
+ when(_requestContext.hasEntity()).thenReturn(true);
+ when(_requestContext.getEntityStream()).thenReturn(
+ new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)));
+
+ AuditEvent result = _processor.processRequest(_requestContext, "10.0.0.1");
+
+ assertThat(result).isNotNull();
+ String capturedBody = result.getRequest().getBody();
+ assertThat(capturedBody).doesNotContain(AWS_SECRET);
+ assertThat(capturedBody).contains(AuditRedactor.MASKED_VALUE);
+ assertThat(capturedBody).contains("CUSTOM_BATCH");
+ }
+
+ @Test
+ public void testCapturedBodyPreservedForDownstreamAfterRedaction() {
+ _defaultConfig.setCaptureRequestPayload(true);
+ String body = "{\"password\":\"" + AWS_SECRET + "\"}";
+
+ when(_requestContext.getUriInfo()).thenReturn(_uriInfo);
+ when(_uriInfo.getQueryParameters()).thenReturn(new MultivaluedHashMap<>());
+ when(_uriInfo.getPath()).thenReturn("/tables");
+ when(_requestContext.getMethod()).thenReturn("POST");
+ when(_requestContext.getHeaders()).thenReturn(new MultivaluedHashMap<>());
+ when(_requestContext.hasEntity()).thenReturn(true);
+ when(_requestContext.getEntityStream()).thenReturn(
+ new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)));
+
+ _processor.processRequest(_requestContext, "10.0.0.1");
+
+ // Redaction must not disturb the entity stream handed to the resource
method.
+ ArgumentCaptor<InputStream> streamCaptor =
ArgumentCaptor.forClass(InputStream.class);
+ verify(_requestContext).setEntityStream(streamCaptor.capture());
+ assertThat(new String(readFully(streamCaptor.getValue()),
StandardCharsets.UTF_8)).isEqualTo(body);
+ }
+
+ @Test
+ public void testCapturedQueryParametersRedactCredentials() {
+ MultivaluedMap<String, String> queryParams =
+ createQueryParams("tableName", "myTable", "accessToken", AWS_SECRET);
+
+ when(_requestContext.getUriInfo()).thenReturn(_uriInfo);
+ when(_uriInfo.getQueryParameters()).thenReturn(queryParams);
+ when(_uriInfo.getPath()).thenReturn("/test");
+ when(_requestContext.getMethod()).thenReturn("GET");
+ when(_requestContext.getHeaders()).thenReturn(new MultivaluedHashMap<>());
+
+ AuditEvent result = _processor.processRequest(_requestContext, "10.0.0.1");
+
+ assertThat(result).isNotNull();
+ Map<String, Object> captured = result.getRequest().getQueryParameters();
+ assertThat(captured).containsEntry("tableName", "myTable");
+ assertThat(captured).containsEntry("accessToken",
AuditRedactor.MASKED_VALUE);
+ }
+
+ @Test
+ public void testCapturedHeadersRedactAuthorization() {
+ _defaultConfig.setCaptureRequestHeaders("content-type,authorization");
+ MultivaluedMap<String, String> headers =
+ createHeaders("Content-Type", "application/json", "Authorization",
"Bearer " + AWS_SECRET);
+
+ when(_requestContext.getUriInfo()).thenReturn(_uriInfo);
+ when(_uriInfo.getQueryParameters()).thenReturn(new MultivaluedHashMap<>());
+ when(_uriInfo.getPath()).thenReturn("/test");
+ when(_requestContext.getMethod()).thenReturn("GET");
+ when(_requestContext.getHeaders()).thenReturn(headers);
+
+ AuditEvent result = _processor.processRequest(_requestContext, "10.0.0.1");
+
+ assertThat(result).isNotNull();
+ Map<String, Object> captured = result.getRequest().getHeaders();
+ assertThat(captured).containsEntry("Content-Type", "application/json");
+ assertThat(captured).containsEntry("Authorization",
AuditRedactor.MASKED_VALUE);
+ }
+
+ @Test
+ public void testUnparseableBodyCapturedBySizeOnly() {
+ _defaultConfig.setCaptureRequestPayload(true);
+ String body = "accessKey=" + AWS_SECRET;
+
+ when(_requestContext.getUriInfo()).thenReturn(_uriInfo);
+ when(_uriInfo.getQueryParameters()).thenReturn(new MultivaluedHashMap<>());
+ when(_uriInfo.getPath()).thenReturn("/test");
+ when(_requestContext.getMethod()).thenReturn("POST");
+ when(_requestContext.getHeaders()).thenReturn(new MultivaluedHashMap<>());
+ when(_requestContext.hasEntity()).thenReturn(true);
+ when(_requestContext.getEntityStream()).thenReturn(
+ new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)));
+
+ AuditEvent result = _processor.processRequest(_requestContext, "10.0.0.1");
+
+ assertThat(result).isNotNull();
+ assertThat(result.getRequest().getBody()).isEqualTo(
+ "[redacted: unparseable payload, " + body.length() + " bytes]");
+ }
+
+ @Test
+ public void testTruncatedBodyKeepsRedactedPrefix() {
+ _defaultConfig.setCaptureRequestPayload(true);
+ _defaultConfig.setMaxPayloadSize(120);
+ String body =
"{\"tableName\":\"myTable\",\"tableType\":\"REALTIME\",\"streamConfigs\":{\"sasl.jaas.config\":\""
+ + AWS_SECRET + "\",\"stream.kafka.topic.name\":\"" + "x".repeat(200) +
"\"}}";
+
+ when(_requestContext.getUriInfo()).thenReturn(_uriInfo);
+ when(_uriInfo.getQueryParameters()).thenReturn(new MultivaluedHashMap<>());
+ when(_uriInfo.getPath()).thenReturn("/tables");
+ when(_requestContext.getMethod()).thenReturn("POST");
+ when(_requestContext.getHeaders()).thenReturn(new MultivaluedHashMap<>());
+ when(_requestContext.hasEntity()).thenReturn(true);
+ when(_requestContext.getEntityStream()).thenReturn(
+ new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)));
+
+ AuditEvent result = _processor.processRequest(_requestContext, "10.0.0.1");
+
+ // The cut falls inside the topic name, which is dropped; everything
before it is kept, redacted.
+
assertThat(result.getRequest().getBody()).isEqualTo("{\"tableName\":\"myTable\",\"tableType\":\"REALTIME\","
+ + "\"streamConfigs\":{\"sasl.jaas.config\":\"" +
AuditRedactor.MASKED_VALUE + "\"}}"
+ + AuditRequestProcessor.TRUNCATION_MARKER);
+ }
+
+ private static byte[] readFully(InputStream stream) {
+ try {
+ return ByteStreams.toByteArray(stream);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
