From 2ea6af5c9787c5febfb06e1f2c3241972bcf825f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 12 Nov 2026 18:41:58 -0800 Subject: [PATCH 1/4] fix(pii): keep encoded files away from Presidio and split texts too long for one NER pass --- .../platform/enterprise/data-retention.mdx | 2 + apps/pii/server.py | 35 ++- .../api/guardrails/mask-batch/route.test.ts | 17 ++ apps/sim/lib/guardrails/mask-client.test.ts | 82 ++++++ apps/sim/lib/guardrails/mask-client.ts | 58 +++-- apps/sim/lib/guardrails/pii-batching.ts | 105 +++++++- apps/sim/lib/guardrails/validate-pii.ts | 19 +- .../lib/logs/execution/pii-redaction.test.ts | 63 +++++ apps/sim/lib/logs/execution/pii-redaction.ts | 197 +++++++++++---- apps/sim/scripts/test-pii-redaction-e2e.ts | 235 ++++++++++++++++++ 10 files changed, 743 insertions(+), 70 deletions(-) create mode 100644 apps/sim/scripts/test-pii-redaction-e2e.ts diff --git a/apps/docs/content/docs/platform/enterprise/data-retention.mdx b/apps/docs/content/docs/platform/enterprise/data-retention.mdx index b12a72e6054..02a0e768eea 100644 --- a/apps/docs/content/docs/platform/enterprise/data-retention.mdx +++ b/apps/docs/content/docs/platform/enterprise/data-retention.mdx @@ -118,6 +118,8 @@ Redaction applies to text strings at the configured workflow stages. It does not If log redaction fails, affected strings are replaced with `[REDACTION_FAILED]` rather than stored unmasked. Failure at Workflow input or Block outputs stops the run instead of passing unredacted values downstream. +Base64-encoded file content, such as a `data:` URL or a long base64 string, can't be checked for PII as text. In stored logs it is replaced with `[BINARY_CONTENT_REDACTED]`. At Workflow input and Block outputs it passes through unchanged, so later blocks can still read the file. **Run from block** and re-running with a previous run's input read that run's log, so they see the marker in place of that content, the same way they see masked PII. + ## Self-hosted setup ### Enable cleanup and schedule it diff --git a/apps/pii/server.py b/apps/pii/server.py index 6934daf8355..970acc131df 100644 --- a/apps/pii/server.py +++ b/apps/pii/server.py @@ -184,6 +184,7 @@ def build_analyzer() -> AnalyzerEngine: anonymizer: AnonymizerEngine NER_ENTITIES: frozenset _BLANK_ARTIFACTS: dict +_NLP_MAX_LENGTH: dict # Every entity the spaCy NER recognizers can actually produce. A request touching # any of these must run spaCy; a request naming only non-NER (regex/checksum) @@ -227,6 +228,32 @@ def _regex_only(entities: list[str] | None, score_threshold: float | None) -> bo ) +def _require_nlp_length( + texts: list[str], + language: str, + entities: list[str] | None, + score_threshold: float | None, +) -> None: + """Refuse, as a 422, any text longer than the language model's spaCy + `max_length` when the request runs the NLP pass. spaCy raises ValueError E088 + for such a text, which FastAPI answers as an opaque 500 that callers retry as + a transient failure; the same text fails the same way every time, so callers + must split it rather than resend it. Regex-only requests skip spaCy and are + not limited.""" + if _regex_only(entities, score_threshold): + return + limit = _NLP_MAX_LENGTH.get(language) + if limit is None: + return + for index, text in enumerate(texts): + if len(text) > limit: + raise HTTPException( + status_code=422, + detail=f"Text {index} has {len(text)} characters; NER accepts at most " + f"{limit} per text. Split it before sending.", + ) + + def _analyze_one( text: str, language: str, @@ -235,6 +262,7 @@ def _analyze_one( return_decision_process: bool = False, ad_hoc_recognizers: list[PatternRecognizer] | None = None, ): + _require_nlp_length([text], language, entities, score_threshold) # Regex-only requests reuse a blank NlpArtifacts to skip the spaCy NLP pass; # otherwise analyze() computes artifacts (runs spaCy) as usual. Custom-pattern # recognizers are regex-based, so they run fine against the blank artifacts. @@ -265,6 +293,7 @@ def _analyze_many( ad_hoc_recognizers: list[PatternRecognizer] | None = None, ): """Analyze many texts, skipping the spaCy pass for regex-only requests.""" + _require_nlp_length(texts, language, entities, score_threshold) if _regex_only(entities, score_threshold): blank = _BLANK_ARTIFACTS.get(language) return [ @@ -292,7 +321,7 @@ def _analyze_many( @asynccontextmanager async def lifespan(_app: FastAPI) -> AsyncIterator[None]: - global analyzer, batch_analyzer, anonymizer, NER_ENTITIES, _BLANK_ARTIFACTS + global analyzer, batch_analyzer, anonymizer, NER_ENTITIES, _BLANK_ARTIFACTS, _NLP_MAX_LENGTH logger.info("building analyzer (spacy)") analyzer = build_analyzer() batch_analyzer = BatchAnalyzerEngine(analyzer_engine=analyzer) @@ -319,6 +348,10 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]: language: analyzer.nlp_engine.process_text("", language) for language in SUPPORTED_LANGUAGES } + _NLP_MAX_LENGTH = { + language: nlp.max_length + for language, nlp in getattr(analyzer.nlp_engine, "nlp", {}).items() + } yield diff --git a/apps/sim/app/api/guardrails/mask-batch/route.test.ts b/apps/sim/app/api/guardrails/mask-batch/route.test.ts index 757b94bcf55..eea962e905f 100644 --- a/apps/sim/app/api/guardrails/mask-batch/route.test.ts +++ b/apps/sim/app/api/guardrails/mask-batch/route.test.ts @@ -1,6 +1,7 @@ import { hybridAuthMockFns } from '@sim/testing/mocks/hybrid-auth.mock' import { createMockRequest } from '@sim/testing/mocks/request.mock' import { beforeEach, describe, expect, it, vi } from 'vitest' +import { PII_MAX_TEXT_CHARS } from '@/lib/guardrails/pii-batching' import { POST } from '@/app/api/guardrails/mask-batch/route' const mockCheckInternalAuth = hybridAuthMockFns.mockCheckInternalAuth @@ -51,4 +52,20 @@ describe('POST /api/guardrails/mask-batch', () => { expect(res.status).toBe(status) }) + + it('splits a text too long for one NER pass before calling Presidio, then rejoins it', async () => { + presidioFetch.mockImplementation(async (_url: string, init: { body: string }) => { + const { texts } = JSON.parse(init.body) as { texts: string[] } + return Response.json({ texts: texts.map((t) => t.replaceAll('Ada', '')) }) + }) + const text = 'Ada wrote this line.\n'.repeat(60_000) + + const res = await POST(createMockRequest('POST', { texts: [text], entityTypes: ['PERSON'] })) + + expect(res.status).toBe(200) + expect(await res.json()).toEqual({ masked: [text.replaceAll('Ada', '')] }) + const sent = presidioFetch.mock.calls.flatMap(([, init]) => JSON.parse(init.body).texts) + expect(sent.length).toBeGreaterThan(1) + for (const piece of sent) expect(piece.length).toBeLessThanOrEqual(PII_MAX_TEXT_CHARS) + }) }) diff --git a/apps/sim/lib/guardrails/mask-client.test.ts b/apps/sim/lib/guardrails/mask-client.test.ts index 48ed3feedde..6696f9f250e 100644 --- a/apps/sim/lib/guardrails/mask-client.test.ts +++ b/apps/sim/lib/guardrails/mask-client.test.ts @@ -16,6 +16,7 @@ vi.mock('@/lib/auth/internal', () => authInternalMock) vi.mock('@sim/utils/helpers', () => utilsHelpersMock) import { maskPIIBatchViaHttp } from '@/lib/guardrails/mask-client' +import { PII_MAX_TEXT_CHARS } from '@/lib/guardrails/pii-batching' import { POST as maskBatchRoute } from '@/app/api/guardrails/mask-batch/route' const mockSleep = utilsHelpersMockFns.mockSleep @@ -73,6 +74,87 @@ describe('maskPIIBatchViaHttp', () => { expect(mockSleep).toHaveBeenCalledTimes(7) }) + it('stops after three attempts on a persistent 500, which can be a deterministic input failure', async () => { + fetchMock.mockImplementation(async () => new Response('failed', { status: 500 })) + + await expect(maskPIIBatchViaHttp(['a'], [])).rejects.toThrow(/mask-batch request failed/) + expect(fetchMock).toHaveBeenCalledTimes(3) + expect(mockSleep).toHaveBeenCalledTimes(2) + }) + + describe('with a text longer than one Presidio pass allows', () => { + /** Each request masks `Ada` to ``, so pieces rejoin to the masked whole. */ + const maskingAda = () => + fetchMock.mockImplementation(async (_url: string, init: { body: string }) => { + const { texts } = JSON.parse(init.body) as { texts: string[] } + return Response.json({ masked: texts.map((t) => t.replaceAll('Ada', '')) }) + }) + const sentPieces = () => + fetchMock.mock.calls.flatMap(([, init]) => JSON.parse(init.body).texts as string[]) + + it('sends newline-aligned pieces within the limit and rejoins the masked result', async () => { + maskingAda() + const text = Array.from({ length: 60_000 }, (_, i) => `row ${i} owner Ada\n`).join('') + + const { masked } = await maskPIIBatchViaHttp([text, 'short Ada'], ['PERSON']) + + expect(masked).toEqual([text.replaceAll('Ada', ''), 'short ']) + const pieces = sentPieces().filter((p) => p !== 'short Ada') + expect(pieces.length).toBeGreaterThan(1) + expect(pieces.join('')).toBe(text) + for (const piece of pieces) expect(piece.length).toBeLessThanOrEqual(PII_MAX_TEXT_CHARS) + for (const piece of pieces.slice(0, -1)) expect(piece.endsWith('\n')).toBe(true) + }) + + it('cuts minified JSON at punctuation so no email is split across pieces', async () => { + maskingAda() + const record = '{"email":"ada.lovelace@example.com","id":1234}' + const text = `[${Array.from({ length: 25_000 }, () => record).join(',')}]` + + await maskPIIBatchViaHttp([text], ['EMAIL_ADDRESS']) + + const pieces = sentPieces() + expect(pieces.length).toBeGreaterThan(1) + const emailsSent = pieces.reduce( + (count, piece) => count + piece.split('ada.lovelace@example.com').length - 1, + 0 + ) + expect(emailsSent).toBe(25_000) + }) + + it('never splits a surrogate pair when the text has nothing to cut at', async () => { + // One leading unit puts a high surrogate exactly where a hard cut would land. + const text = `a${'\u{1F600}'.repeat(PII_MAX_TEXT_CHARS)}` + + const { masked } = await maskPIIBatchViaHttp([text], []) + + const pieces = sentPieces() + expect(pieces.length).toBeGreaterThan(1) + expect(pieces.join('')).toBe(text) + expect(masked).toEqual([`M(${pieces.join(')M(')})`]) + }) + + it('replaces the whole text when any one of its pieces fails', async () => { + const text = `${'safe line\n'.repeat(120_000)}poison` + fetchMock.mockImplementation(async (_url: string, init: { body: string }) => { + const { texts } = JSON.parse(init.body) as { texts: string[] } + if (texts.some((t) => t.endsWith('poison'))) return new Response('no', { status: 422 }) + return Response.json({ masked: texts }) + }) + + const { masked, scrubbedCount } = await maskPIIBatchViaHttp( + [text], + [], + undefined, + undefined, + { failedChunkPlaceholder: '[FAILED]' } + ) + + expect(masked).toEqual(['[FAILED]']) + expect(scrubbedCount).toBe(1) + }) + }) + describe('with a failed-chunk placeholder', () => { const PLACEHOLDER = '[FAILED]' const CHUNK_SIZE = 2000 diff --git a/apps/sim/lib/guardrails/mask-client.ts b/apps/sim/lib/guardrails/mask-client.ts index 07725cbe3d7..3437f7e4ac3 100644 --- a/apps/sim/lib/guardrails/mask-client.ts +++ b/apps/sim/lib/guardrails/mask-client.ts @@ -8,7 +8,7 @@ import { generateInternalToken } from '@/lib/auth/internal' import { env } from '@/lib/core/config/env' import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import { getInternalApiBaseUrl } from '@/lib/core/utils/urls' -import { chunkIndicesByBudget } from '@/lib/guardrails/pii-batching' +import { chunkIndicesByBudget, joinSegments, segmentTexts } from '@/lib/guardrails/pii-batching' import type { CustomPiiPattern } from '@/lib/guardrails/pii-entities' const logger = createLogger('PiiMaskClient') @@ -36,6 +36,15 @@ const CHUNK_CONCURRENCY = env.PII_MASK_CHUNK_CONCURRENCY ?? 64 */ const MAX_CHUNK_ATTEMPTS = 8 +/** + * Attempts for a chunk the route answers with a 500. The route keeps 503 for a + * Presidio outage and 422 for input Presidio refused, so a 500 is an unclassified + * failure that may be deterministic for this input: two retries cover a one-off + * crash without spending the full outage budget resending input that fails the + * same way every time. + */ +const MAX_ATTEMPTS_AFTER_SERVER_ERROR = 3 + const RETRYABLE_STATUSES = new Set([408, 429, 500, 502, 503, 504]) class MaskChunkHttpError extends Error { @@ -97,9 +106,9 @@ function isOutageError(error: unknown): boolean { interface MaskPIIBatchOptions { /** - * When set, a chunk that cannot be masked has each of its strings replaced with - * this value instead of rejecting the call, so one bad chunk costs only its own - * strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's + * When set, a chunk that cannot be masked has every string with a piece in it + * replaced with this value instead of rejecting the call, so one bad chunk costs + * only its own strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's * retries, chunks not yet sent are replaced the same way. */ failedChunkPlaceholder?: string @@ -118,7 +127,9 @@ interface MaskPIIBatchResult { * Only the app task reaches the Presidio service (it holds `PII_URL`), but the * log-redaction persist path also runs inside the trigger.dev runtime — so * redaction always routes through HTTP, the same way the guardrails tool does. - * Strings are grouped into byte/count-budgeted chunks (keeping each request far + * A string too long for one NER pass is cut into pieces first (see + * {@link segmentTexts}), so no request carries a text Presidio would refuse. + * Pieces are grouped into byte/count-budgeted chunks (keeping each request far * under the 10MB Next body limit) and the chunks are sent with bounded * concurrency, so a large payload fans out rather than serializing; order is * preserved, so the returned `masked` array matches `texts` length. @@ -141,38 +152,47 @@ export async function maskPIIBatchViaHttp( const url = `${getInternalApiBaseUrl()}/api/guardrails/mask-batch` const placeholder = options.failedChunkPlaceholder - const masked = new Array(texts.length) + const { segments, owners } = segmentTexts(texts) + const maskedSegments = new Array(segments.length) + const scrubbedTexts = new Set() let serviceUnavailable = false - let scrubbedCount = 0 let firstFailure: unknown - const scrubChunk = (indices: number[], value: string) => { - for (const i of indices) masked[i] = value - scrubbedCount += indices.length + const scrubChunk = (indices: number[]) => { + for (const k of indices) scrubbedTexts.add(owners[k]) } - await mapWithConcurrency(chunkIndicesByBudget(texts), CHUNK_CONCURRENCY, async (indices) => { + await mapWithConcurrency(chunkIndicesByBudget(segments), CHUNK_CONCURRENCY, async (indices) => { if (serviceUnavailable && placeholder !== undefined) { - scrubChunk(indices, placeholder) + scrubChunk(indices) return } try { - const chunk = indices.map((i) => texts[i]) + const chunk = indices.map((k) => segments[k]) const out = await postChunk(url, chunk, entityTypes, language, customPatterns) if (out.length !== chunk.length) { throw new Error('PII mask-batch returned an unexpected result') } - indices.forEach((originalIndex, k) => { - masked[originalIndex] = out[k] + indices.forEach((segmentIndex, k) => { + maskedSegments[segmentIndex] = out[k] }) } catch (error) { if (placeholder === undefined) throw error if (isOutageError(error)) serviceUnavailable = true firstFailure ??= error - scrubChunk(indices, placeholder) + scrubChunk(indices) } }) + const joined = joinSegments(maskedSegments, owners, texts.length) + // A text with any unmasked piece is replaced whole: its other pieces are masked, + // but keeping them would persist a fragment around the hole. + const masked = + placeholder === undefined + ? joined + : joined.map((text, i) => (scrubbedTexts.has(i) ? placeholder : text)) + const scrubbedCount = scrubbedTexts.size + if (scrubbedCount > 0) { logger.error('PII masking failed for some chunks; replaced their strings', { error: getErrorMessage(firstFailure), @@ -196,7 +216,11 @@ async function postChunk( try { return await postChunkOnce(url, texts, entityTypes, language, customPatterns) } catch (error) { - if (attempt >= MAX_CHUNK_ATTEMPTS || !isRetryableChunkError(error)) { + const maxAttempts = + error instanceof MaskChunkHttpError && error.status === 500 + ? MAX_ATTEMPTS_AFTER_SERVER_ERROR + : MAX_CHUNK_ATTEMPTS + if (attempt >= maxAttempts || !isRetryableChunkError(error)) { throw error } const retryAfterMs = error instanceof MaskChunkHttpError ? error.retryAfterMs : null diff --git a/apps/sim/lib/guardrails/pii-batching.ts b/apps/sim/lib/guardrails/pii-batching.ts index 2dd419d45cc..8c5470f84b7 100644 --- a/apps/sim/lib/guardrails/pii-batching.ts +++ b/apps/sim/lib/guardrails/pii-batching.ts @@ -1,8 +1,8 @@ /** * Per-request bounds shared by both Presidio hops: the app→route HTTP call * (`mask-client`) and the route→service call (`validate-pii`). Keeping a single - * source of truth ensures every request stays far under the 10MB Next body limit - * and small enough for one short spaCy NER pass per Presidio request. + * source of truth ensures every request stays far under the 10MB Next body limit, + * every text fits one spaCy NER pass, and each request is one short pass. */ /** Max UTF-8 bytes of text per Presidio request. ~40× under the 10MB Next limit. */ @@ -10,6 +10,107 @@ export const PII_REQUEST_MAX_BYTES = 256 * 1024 /** Max strings per request; caps per-item overhead and stays well under the contract's 100k-entry cap. */ export const PII_REQUEST_MAX_COUNT = 2_000 +/** + * Max characters of one text in one Presidio NER pass: spaCy's default + * `max_length`, past which it refuses the text outright (the Presidio service + * answers 422). Every text Presidio accepts is still sent whole, so no entity can + * straddle a cut in it; only a longer text, which could never be masked whole, is + * split by {@link segmentTexts}, following spaCy's guidance to segment long + * documents before NER. A JavaScript length counts UTF-16 units, never fewer than + * the code points spaCy counts, so a text within it is always within spaCy's. + */ +export const PII_MAX_TEXT_CHARS = 1_000_000 + +/** + * The shortest piece a cut may leave, so a boundary is never chosen so early that + * a long text splinters into many tiny requests. + */ +const MIN_SEGMENT_CHARS = PII_MAX_TEXT_CHARS / 2 + +/** Letters, combining marks, digits, and `_`: characters an entity can run through. */ +const WORD_CHARACTER = /[\p{L}\p{M}\p{N}_]/u + +/** + * Structural delimiters (JSON, CSV, markup) that no detected entity contains, unlike + * the `.`, `@`, `-`, `+`, `:`, and `/` inside emails, URLs, phone numbers, and ids. + */ +const STRUCTURAL_DELIMITER = /[,;"`()[\]{}<>|]/ + +function isSurrogate(code: number): boolean { + return code >= 0xd800 && code <= 0xdfff +} + +function isHighSurrogate(code: number): boolean { + return code >= 0xd800 && code <= 0xdbff +} + +/** + * Where to end the piece of `text` that starts at `start`, at most + * {@link PII_MAX_TEXT_CHARS} later and no earlier than {@link MIN_SEGMENT_CHARS}. + * Prefers the end of a line, then whitespace, then a structural delimiter, then any + * other punctuation, since NER and Presidio's pattern recognizers match within + * those boundaries and an entity that straddles a cut would be missed. Only a run + * with no boundary at all is cut mid-run, and never between the two halves of a + * surrogate pair. + */ +function cutPoint(text: string, start: number): number { + const limit = start + PII_MAX_TEXT_CHARS + const floor = start + MIN_SEGMENT_CHARS + + const newline = text.slice(floor, limit).lastIndexOf('\n') + if (newline !== -1) return floor + newline + 1 + + for (let i = limit - 1; i >= floor; i--) { + if (/\s/.test(text[i])) return i + 1 + } + for (let i = limit - 1; i >= floor; i--) { + if (STRUCTURAL_DELIMITER.test(text[i])) return i + 1 + } + for (let i = limit - 1; i >= floor; i--) { + if (!isSurrogate(text.charCodeAt(i)) && !WORD_CHARACTER.test(text[i])) return i + 1 + } + return isHighSurrogate(text.charCodeAt(limit - 1)) ? limit - 1 : limit +} + +/** Every text's pieces, in order, each within {@link PII_MAX_TEXT_CHARS}. */ +export interface SegmentedTexts { + segments: string[] + /** `owners[k]` is the index of the input text `segments[k]` was cut from. */ + owners: number[] +} + +/** + * Split each text longer than {@link PII_MAX_TEXT_CHARS} into consecutive pieces + * (see {@link cutPoint}); shorter texts pass through as one piece. The pieces of a + * text concatenate back to it exactly, so masking them one by one and joining them + * with {@link joinSegments} masks the whole text. + */ +export function segmentTexts(texts: string[]): SegmentedTexts { + const segments: string[] = [] + const owners: number[] = [] + texts.forEach((text, owner) => { + let start = 0 + while (text.length - start > PII_MAX_TEXT_CHARS) { + const end = cutPoint(text, start) + segments.push(text.slice(start, end)) + owners.push(owner) + start = end + } + segments.push(text.slice(start)) + owners.push(owner) + }) + return { segments, owners } +} + +/** Concatenate masked pieces back into one string per original text, in order. */ +export function joinSegments(pieces: string[], owners: number[], textCount: number): string[] { + const joined = new Array(textCount).fill('') + pieces.forEach((piece, k) => { + joined[owners[k]] += piece + }) + return joined +} + /** * Group `texts` into chunks of original indices, flushing a chunk when adding the * next string would exceed {@link PII_REQUEST_MAX_BYTES} or {@link PII_REQUEST_MAX_COUNT}. diff --git a/apps/sim/lib/guardrails/validate-pii.ts b/apps/sim/lib/guardrails/validate-pii.ts index 962870c61e6..a12cc72a4a3 100644 --- a/apps/sim/lib/guardrails/validate-pii.ts +++ b/apps/sim/lib/guardrails/validate-pii.ts @@ -8,7 +8,7 @@ import { readResponseJsonWithLimit, readResponseTextWithLimit, } from '@/lib/core/utils/stream-limits' -import { chunkIndicesByBudget } from '@/lib/guardrails/pii-batching' +import { chunkIndicesByBudget, joinSegments, segmentTexts } from '@/lib/guardrails/pii-batching' import type { CustomPiiPattern } from '@/lib/guardrails/pii-entities' import { MAX_PII_VALIDATION_DETECTED_ENTITIES, @@ -501,9 +501,11 @@ export async function validatePII(input: PIIValidationInput): Promise { if (texts.length === 0) return [] - const result = new Array(texts.length) + const { segments, owners } = segmentTexts(texts) + const result = new Array(segments.length) - await mapWithConcurrency(chunkIndicesByBudget(texts), CHUNK_CONCURRENCY, async (indices) => { - const chunkTexts = indices.map((i) => texts[i]) + await mapWithConcurrency(chunkIndicesByBudget(segments), CHUNK_CONCURRENCY, async (indices) => { + const chunkTexts = indices.map((i) => segments[i]) if (combinedRedactAvailable) { const masked = await redactBatch(chunkTexts, entityTypes, language, customPatterns) @@ -571,7 +574,7 @@ export async function maskPIIBatch( }) }) - return result + return joinSegments(result, owners, texts.length) } export { type PIIEntityType, SUPPORTED_PII_ENTITIES } from '@/lib/guardrails/pii-entities' diff --git a/apps/sim/lib/logs/execution/pii-redaction.test.ts b/apps/sim/lib/logs/execution/pii-redaction.test.ts index b32e4c5c85e..3cf599657c6 100644 --- a/apps/sim/lib/logs/execution/pii-redaction.test.ts +++ b/apps/sim/lib/logs/execution/pii-redaction.test.ts @@ -5,6 +5,7 @@ import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' vi.mock('@/lib/auth/internal', () => authInternalMock) import { + BINARY_CONTENT_REDACTED_MARKER, PiiRedactionError, REDACTION_FAILED_MARKER, redactObjectStrings, @@ -129,6 +130,68 @@ describe('redactPIIFromExecution', () => { }) }) +/** Base64 of 48KB of non-text bytes, the shape a file body takes in a payload. */ +const encodedFile = Buffer.from( + Array.from({ length: 48 * 1024 }, (_, i) => (i * 7919 + 13) % 256) +).toString('base64') + +describe('encoded binary content', () => { + it('replaces encoded file bodies in a persisted log without sending them to Presidio', async () => { + const payload = { + traceSpans: [{ blockId: 'write', input: { content: encodedFile, name: 'Ada Lovelace' } }], + workflowInput: { file: `data:application/pdf;base64,${encodedFile}` }, + } + + const result = await redactPIIFromExecution(payload, { entityTypes: ['PERSON'] }) + + const [span] = result.traceSpans as { input: { content: string; name: string } }[] + expect(span.input.content).toBe(BINARY_CONTENT_REDACTED_MARKER) + expect((result.workflowInput as { file: string }).file).toBe(BINARY_CONTENT_REDACTED_MARKER) + expect(span.input.name).toBe('MASKED(Ada Lovelace)') + expect(sentTexts()).toEqual(['Ada Lovelace']) + }) + + it('passes encoded file bodies through live data unchanged and unsent', async () => { + const value = { content: encodedFile, url: `data:image/png;base64,${encodedFile}`, note: 'hi' } + + const result = await redactObjectStrings(value, { entityTypes: [], onFailure: 'throw' }) + + expect(result.content).toBe(encodedFile) + expect(result.url).toBe(`data:image/png;base64,${encodedFile}`) + expect(result.note).toBe('MASKED(hi)') + expect(sentTexts()).toEqual(['hi']) + }) + + /** `encodedFile` wrapped at `width` columns, as MIME (76) and PEM (64) encoders emit it. */ + const wrapped = (width: number, newline: string) => + (encodedFile.match(new RegExp(`.{1,${width}}`, 'g')) ?? []).join(newline) + + it.each([ + ['with a trailing newline', `${encodedFile}\n`], + ['wrapped at 76 columns', wrapped(76, '\n')], + ['wrapped at 64 columns with CRLF', `${wrapped(64, '\r\n')}\r\n`], + ['as a wrapped data URL', `data:application/pdf;base64,${wrapped(76, '\n')}`], + ])('replaces an encoded file body %s in a persisted log', async (_label, body) => { + const result = await redactPIIFromExecution({ finalOutput: body }, { entityTypes: [] }) + + expect(result.finalOutput).toBe(BINARY_CONTENT_REDACTED_MARKER) + expect(sentTexts()).toEqual([]) + }) + + it.each([ + ['a newline-separated list of usernames', 'AdaLovelace1815\n'.repeat(100)], + ['a long run of digits', '4'.repeat(4096)], + ['lowercase hex', 'deadbeef0123'.repeat(400)], + ['a short base64 token', 'QWRhIExvdmVsYWNl'], + ['a long space-separated sentence', 'Ada Lovelace wrote code. '.repeat(200)], + ])('still masks %s, which is not an encoded file', async (_label, text) => { + const result = await redactObjectStrings({ text }, { entityTypes: [], onFailure: 'throw' }) + + expect(result.text).not.toBe(text) + expect(sentTexts().join('')).toBe(text) + }) +}) + describe('redactObjectStrings', () => { it('throws PiiRedactionError on masking failure when onFailure is throw', async () => { maskRoute.mockResolvedValueOnce(rejected()) diff --git a/apps/sim/lib/logs/execution/pii-redaction.ts b/apps/sim/lib/logs/execution/pii-redaction.ts index 34d791e0491..04c298d05d4 100644 --- a/apps/sim/lib/logs/execution/pii-redaction.ts +++ b/apps/sim/lib/logs/execution/pii-redaction.ts @@ -10,13 +10,89 @@ const logger = createLogger('PiiRedaction') /** Replaces text we could not safely mask, so PII is never persisted on failure. */ export const REDACTION_FAILED_MARKER = '[REDACTION_FAILED]' +/** Replaces an encoded file body in a persisted record (see {@link isEncodedBinary}). */ +export const BINARY_CONTENT_REDACTED_MARKER = '[BINARY_CONTENT_REDACTED]' + +/** + * Shortest bare base64 string treated as an encoded file body. Far longer than any + * word, identifier, hash, or token, so only encoded payloads reach it. + */ +const MIN_ENCODED_BINARY_CHARS = 1024 + +const STANDARD_BASE64 = /^[A-Za-z0-9+/]+={0,2}$/ +const URL_SAFE_BASE64 = /^[A-Za-z0-9_-]+={0,2}$/ +const BASE64_DATA_URL = /^data:(?:[\w.+-]+\/[\w.+-]+)?(?:;[\w.+-]+=[^;,]*)*;base64,/i + +/** Narrowest and widest lines encoders wrap base64 at: 60 (common), 64 (PEM), 76 (MIME). */ +const MIN_WRAP_WIDTH = 60 +const MAX_WRAP_WIDTH = 76 + +function isBase64(value: string): boolean { + return STANDARD_BASE64.test(value) || URL_SAFE_BASE64.test(value) +} + +/** + * `value` with trailing whitespace dropped and, when it is wrapped the way base64 + * encoders wrap (every line but the last one fixed width, a multiple of 4 between + * {@link MIN_WRAP_WIDTH} and {@link MAX_WRAP_WIDTH}), its line breaks removed. + * Returns `null` for multi-line text without that shape, which is never base64. + */ +function unwrapBase64(value: string): string | null { + const body = value.trimEnd() + const firstBreak = body.indexOf('\n') + if (firstBreak === -1) return body + const width = body[firstBreak - 1] === '\r' ? firstBreak - 1 : firstBreak + if (width < MIN_WRAP_WIDTH || width > MAX_WRAP_WIDTH || width % 4 !== 0) return null + const lines = body.split(/\r?\n/) + const last = lines[lines.length - 1] + const fixedWidth = lines.slice(0, -1).every((line) => line.length === width) + return fixedWidth && last.length > 0 && last.length <= width ? lines.join('') : null +} + +/** + * Whether `value` is an encoded file body: a `data:` URL declaring base64, or a + * bare base64 string at least {@link MIN_ENCODED_BINARY_CHARS} long, either one + * optionally wrapped the way encoders wrap it (see {@link unwrapBase64}). A bare + * string must also mix upper case, lower case, and digits, as encoded bytes always + * do, so a run of digits, lowercase hex, or a list of codes is still masked as text. + * + * Presidio cannot redact such a value: spaCy reads base64 as one token, and the + * pattern recognizers only ever match it by coincidence at a `+` or `/`, where a + * replacement would corrupt the bytes rather than remove anything readable. Any + * PII is inside the decoded file, which text masking never sees. + */ +function isEncodedBinary(value: string): boolean { + if (value.startsWith('data:')) { + const prefix = BASE64_DATA_URL.exec(value.slice(0, 512)) + if (prefix === null) return false + const payload = unwrapBase64(value.slice(prefix[0].length)) + return payload !== null && isBase64(payload) + } + if (value.length < MIN_ENCODED_BINARY_CHARS) return false + const body = unwrapBase64(value) + return ( + body !== null && + body.length >= MIN_ENCODED_BINARY_CHARS && + isBase64(body) && + /[A-Z]/.test(body) && + /[a-z]/.test(body) && + /\d/.test(body) + ) +} + /** - * How to handle a masking failure (Presidio error or over-ceiling payload): - * - `'scrub'` (default): replace eligible strings with {@link REDACTION_FAILED_MARKER}. - * Safe for the log stage — execution already succeeded. - * - `'throw'`: throw {@link PiiRedactionError}. Used for the execution-altering - * stages (input/block outputs) where a marker would corrupt computed data and - * fail-open would leak — so the run aborts instead. + * Whether a redaction pass is producing the execution record or data a later step + * reads, which decides both how a masking failure is handled and what becomes of + * an encoded file body Presidio cannot mask: + * - `'scrub'` (default), the execution log: a failure replaces the affected strings + * with {@link REDACTION_FAILED_MARKER}, and an encoded file body is replaced with + * {@link BINARY_CONTENT_REDACTED_MARKER}, since the log must not keep content + * that cannot be checked for PII. Safe because execution already succeeded. + * - `'throw'`, data a later step reads: workflow input, block outputs, and agent + * memory stored for later turns. A failure throws {@link PiiRedactionError} and + * aborts the run, since a marker would corrupt that data and fail-open would + * leak. An encoded file body passes through unchanged, since the step that + * reads it needs its bytes and masking could only corrupt them. */ export type PiiRedactionFailureMode = 'scrub' | 'throw' @@ -79,14 +155,24 @@ function isEligibleString(value: string): boolean { } /** - * Rebuild `value` replacing every eligible string leaf with `handle(leaf)`. - * Used for both collection (handle records and returns the input) and - * substitution (handle returns the masked value), so traversal order and - * eligibility are guaranteed identical across the two passes. + * What one pass does with each string leaf: `text` for a string Presidio masks, + * `encodedBinary` for an encoded file body it never sees (see {@link isEncodedBinary}). + */ +interface StringVisitor { + text: (s: string) => string + encodedBinary: (s: string) => string +} + +/** + * Rebuild `value` replacing every eligible string leaf with the visitor's result. + * Used for both collection (the visitor records and returns the input) and + * substitution (it returns the masked value), so traversal order, eligibility, and + * the text/binary split are guaranteed identical across the two passes. */ -function transformStrings(value: unknown, handle: (s: string) => string): unknown { +function transformStrings(value: unknown, visit: StringVisitor): unknown { if (typeof value === 'string') { - return isEligibleString(value) ? handle(value) : value + if (!isEligibleString(value)) return value + return isEncodedBinary(value) ? visit.encodedBinary(value) : visit.text(value) } // Treat offloaded large-value refs as opaque: masking their internal `key`/`id` // strings would corrupt the reference. Their content is handled separately @@ -95,12 +181,12 @@ function transformStrings(value: unknown, handle: (s: string) => string): unknow return value } if (Array.isArray(value)) { - return value.map((item) => transformStrings(item, handle)) + return value.map((item) => transformStrings(item, visit)) } if (value !== null && typeof value === 'object') { const out: Record = {} for (const [key, v] of Object.entries(value)) { - out[key] = transformStrings(v, handle) + out[key] = transformStrings(v, visit) } return out } @@ -112,17 +198,17 @@ function transformStrings(value: unknown, handle: (s: string) => string): unknow * nested `children` are walked, leaving structural metadata (blockId, name, * status, timing) untouched so log correlation/display is preserved. */ -function transformSpan(span: unknown, handle: (s: string) => string): unknown { +function transformSpan(span: unknown, visit: StringVisitor): unknown { if (span === null || typeof span !== 'object' || Array.isArray(span)) { - return transformStrings(span, handle) + return transformStrings(span, visit) } const source = span as Record const out: Record = { ...source } for (const field of SPAN_CONTENT_FIELDS) { - if (field in out) out[field] = transformStrings(out[field], handle) + if (field in out) out[field] = transformStrings(out[field], visit) } if (Array.isArray(source.children)) { - out.children = source.children.map((child) => transformSpan(child, handle)) + out.children = source.children.map((child) => transformSpan(child, visit)) } return out } @@ -130,12 +216,34 @@ function transformSpan(span: unknown, handle: (s: string) => string): unknown { function transformUnit( key: keyof RedactablePayload, value: unknown, - handle: (s: string) => string + visit: StringVisitor ): unknown { if (key === 'traceSpans' && Array.isArray(value)) { - return value.map((span) => transformSpan(span, handle)) + return value.map((span) => transformSpan(span, visit)) + } + return transformStrings(value, visit) +} + +/** What an encoded file body becomes under `onFailure` (see {@link PiiRedactionFailureMode}). */ +function encodedBinaryReplacement(onFailure: PiiRedactionFailureMode): (s: string) => string { + return onFailure === 'scrub' ? () => BINARY_CONTENT_REDACTED_MARKER : (s) => s +} + +/** A collection pass: records every string to mask and counts encoded file bodies. */ +function collectStrings() { + const collected: string[] = [] + let encodedBinaryCount = 0 + const visit: StringVisitor = { + text: (s) => { + collected.push(s) + return s + }, + encodedBinary: (s) => { + encodedBinaryCount++ + return s + }, } - return transformStrings(value, handle) + return { collected, visit, encodedBinaryCount: () => encodedBinaryCount } } /** @@ -194,17 +302,19 @@ async function maskCollected( * block-output stages, so it defaults callers toward `onFailure: 'throw'`. */ export async function redactObjectStrings(value: T, options: PiiRedactionOptions): Promise { - const collected: string[] = [] - transformStrings(value, (s) => { - collected.push(s) - return s - }) + const collection = collectStrings() + transformStrings(value, collection.visit) + const { collected } = collection - if (collected.length === 0) return value + if (collected.length === 0 && collection.encodedBinaryCount() === 0) return value - const { masked } = await maskCollected(collected, options) + const { masked } = + collected.length === 0 ? { masked: [] } : await maskCollected(collected, options) let index = 0 - return transformStrings(value, () => masked[index++]) as T + return transformStrings(value, { + text: () => masked[index++], + encodedBinary: encodedBinaryReplacement(options.onFailure ?? 'scrub'), + }) as T } /** @@ -230,29 +340,32 @@ export async function redactPIIFromExecution( value: payload[key], })) - const collected: string[] = [] - let totalBytes = 0 - for (const unit of units) { - transformUnit(unit.key, unit.value, (s) => { - collected.push(s) - totalBytes += Buffer.byteLength(s, 'utf8') - return s - }) - } + const collection = collectStrings() + for (const unit of units) transformUnit(unit.key, unit.value, collection.visit) + const { collected } = collection + const encodedBinaryCount = collection.encodedBinaryCount() - if (collected.length === 0) return payload + if (collected.length === 0 && encodedBinaryCount === 0) return payload - const { masked, scrubbedCount } = await maskCollected(collected, options) + const { masked, scrubbedCount } = + collected.length === 0 + ? { masked: [], scrubbedCount: 0 } + : await maskCollected(collected, options) let index = 0 + const visit: StringVisitor = { + text: () => masked[index++], + encodedBinary: encodedBinaryReplacement(options.onFailure ?? 'scrub'), + } const result: RedactablePayload = { ...payload } for (const unit of units) { - result[unit.key] = transformUnit(unit.key, unit.value, () => masked[index++]) + result[unit.key] = transformUnit(unit.key, unit.value, visit) } logger.info('PII redaction completed', { stringCount: collected.length, - totalBytes, + totalBytes: collected.reduce((sum, s) => sum + Buffer.byteLength(s, 'utf8'), 0), + encodedBinaryCount, durationMs: Math.round(performance.now() - startedAt), scrubbedCount, }) diff --git a/apps/sim/scripts/test-pii-redaction-e2e.ts b/apps/sim/scripts/test-pii-redaction-e2e.ts new file mode 100644 index 00000000000..7b5d2400c34 --- /dev/null +++ b/apps/sim/scripts/test-pii-redaction-e2e.ts @@ -0,0 +1,235 @@ +import assert from 'node:assert/strict' +import { randomBytes } from 'node:crypto' +import { mkdir, writeFile } from 'node:fs/promises' +import { dirname } from 'node:path' +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' + +/** + * Proves PII redaction of oversized and encoded content against a running local + * app and a real Presidio container (`docker/pii.Dockerfile`, the app's `PII_URL`). + * The redaction library runs in this process the way the log persist path runs in + * a worker: every mask request crosses the app's `/api/guardrails/mask-batch` route + * over HTTP, and the route calls Presidio, whose spaCy NER refuses texts over + * 1,000,000 characters. + * + * Checks, each recorded in the report: + * - Presidio answers an over-length NER text with a 422, not a 500, and still + * serves the same text when the request needs only pattern recognizers. + * - A persisted log holding a 5M-character base64 PDF and a 3M-character text that + * names a person at the end: the PDF becomes the binary marker without reaching + * Presidio, the name is masked by real NER, and no request fails. + * - The same content as live block output: the PDF passes through byte-identical + * and the name is masked. + * + * Exits non-zero when any check fails, after writing the report. + */ +const logger = createLogger('PiiRedactionE2E') + +function requiredEnvironment(name: string): string { + const value = process.env[name] + assert(value, `${name} must be explicitly provided`) + return value +} + +const LOOPBACK_HOSTS = new Set(['localhost', '127.0.0.1', '[::1]']) +const baseUrl = new URL(requiredEnvironment('PII_E2E_BASE_URL')) +const presidioUrl = new URL(requiredEnvironment('PII_E2E_PRESIDIO_URL')) +const reportPath = requiredEnvironment('PII_E2E_REPORT_PATH') +assert(LOOPBACK_HOSTS.has(baseUrl.hostname), 'The app must use a loopback host') +assert(LOOPBACK_HOSTS.has(presidioUrl.hostname), 'Presidio must use a loopback host') +assert( + requiredEnvironment('INTERNAL_API_SECRET').length >= 32, + 'INTERNAL_API_SECRET must match the running app' +) +process.env.INTERNAL_API_BASE_URL = baseUrl.origin + +/** What a persisted log shows in place of an encoded file body. */ +const BINARY_MARKER = '[BINARY_CONTENT_REDACTED]' +const FAILED_MARKER = '[REDACTION_FAILED]' +/** spaCy's default `max_length`, the per-text ceiling of one NER pass. */ +const SPACY_MAX_LENGTH = 1_000_000 +/** The longest text one mask request may carry once texts are split. */ +const MAX_TEXT_CHARS = 1_000_000 + +/** 3.75MB of bytes behind a PDF header: 5,000,000 characters of base64. */ +const pdfBase64 = Buffer.concat([Buffer.from('%PDF-1.7\n'), randomBytes(3_750_000 - 9)]).toString( + 'base64' +) +const FILLER = 'The quarterly report covers revenue, costs, and the hiring plan.\n' +const NAME_SENTENCE = 'Please send questions to Ada Lovelace before Friday.' +const longText = `${FILLER.repeat(Math.ceil(3_000_000 / FILLER.length))}${NAME_SENTENCE}` + +interface MaskRequest { + status: number + texts: number + longestText: number + totalChars: number +} + +/** Every mask-batch request this process sends, observed at `fetch`, unmodified. */ +const maskRequests: MaskRequest[] = [] +const realFetch = globalThis.fetch +globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const response = await realFetch(input, init) + const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url + if (url.endsWith('/api/guardrails/mask-batch') && typeof init?.body === 'string') { + const { texts } = JSON.parse(init.body) as { texts: string[] } + maskRequests.push({ + status: response.status, + texts: texts.length, + longestText: Math.max(0, ...texts.map((text) => text.length)), + totalChars: texts.reduce((sum, text) => sum + text.length, 0), + }) + } + return response +}) as typeof fetch + +const checks: { + name: string + status: 'passed' | 'failed' + durationMs: number + details?: Record + error?: string +}[] = [] + +async function check( + name: string, + run: () => Promise | undefined> +): Promise { + const startedAt = performance.now() + const requestsBefore = maskRequests.length + try { + const details = await run() + checks.push({ + name, + status: 'passed', + durationMs: Math.round(performance.now() - startedAt), + details: { ...details, maskRequests: maskRequests.slice(requestsBefore) }, + }) + } catch (error) { + checks.push({ + name, + status: 'failed', + durationMs: Math.round(performance.now() - startedAt), + details: { maskRequests: maskRequests.slice(requestsBefore) }, + error: getErrorMessage(error), + }) + } +} + +function assertMaskRequestsSucceeded(since: number, mustNotCarry: number): void { + const requests = maskRequests.slice(since) + assert(requests.length > 0, 'Expected mask-batch requests') + for (const request of requests) { + assert.equal(request.status, 200, `A mask-batch request answered ${request.status}`) + assert( + request.longestText <= MAX_TEXT_CHARS, + `A mask-batch request carried a ${request.longestText}-character text` + ) + } + const sent = requests.reduce((sum, request) => sum + request.totalChars, 0) + assert(sent < mustNotCarry, `Requests carried ${sent} characters, so the encoded file was sent`) +} + +function assertNameMasked(masked: unknown): void { + assert.equal(typeof masked, 'string', 'Expected the long text back as a string') + const text = masked as string + assert(!text.includes(FAILED_MARKER), 'The long text was scrubbed instead of masked') + assert(!text.includes('Ada Lovelace'), 'The name at the end of the long text was not masked') + assert(text.endsWith('before Friday.'), 'The end of the long text was lost') + assert(text.includes(''), 'Expected a token where the name was') +} + +async function main(): Promise { + const { redactObjectStrings, redactPIIFromExecution } = await import( + '@/lib/logs/execution/pii-redaction' + ) + + await check('Presidio refuses an over-length NER text with a 422', async () => { + const overLength = 'a '.repeat(SPACY_MAX_LENGTH / 2 + 1) + const ner = await realFetch(new URL('/redact_batch', presidioUrl), { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ texts: [overLength], language: 'en', entities: ['PERSON'] }), + }) + const detail = await ner.text() + assert.equal(ner.status, 422, `NER pass answered ${ner.status}: ${detail.slice(0, 200)}`) + + const patternsOnly = await realFetch(new URL('/redact_batch', presidioUrl), { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ texts: [overLength], language: 'en', entities: ['EMAIL_ADDRESS'] }), + }) + assert.equal(patternsOnly.status, 200, 'A pattern-only pass must not be length-limited') + return { nerStatus: ner.status, detail: detail.slice(0, 200), patternOnlyStatus: 200 } + }) + + await check('A persisted log replaces the PDF and masks the long text', async () => { + const since = maskRequests.length + const result = await redactPIIFromExecution( + { + traceSpans: [ + { + blockId: 'write-file', + input: { content: pdfBase64, name: 'datasheet.pdf' }, + output: { text: longText }, + }, + ], + workflowInput: { file: `data:application/pdf;base64,${pdfBase64}` }, + }, + { entityTypes: ['PERSON'] } + ) + const [span] = result.traceSpans as { + input: { content: string } + output: { text: string } + }[] + assert.equal(span.input.content, BINARY_MARKER) + assert.equal((result.workflowInput as { file: string }).file, BINARY_MARKER) + assertNameMasked(span.output.text) + assertMaskRequestsSucceeded(since, pdfBase64.length) + return { maskedTail: span.output.text.slice(-60) } + }) + + await check('Live block output keeps the PDF bytes and masks the long text', async () => { + const since = maskRequests.length + const result = await redactObjectStrings( + { content: pdfBase64, text: longText }, + { entityTypes: ['PERSON'], onFailure: 'throw' } + ) + assert.equal(result.content, pdfBase64, 'The encoded file changed') + assertNameMasked(result.text) + assertMaskRequestsSucceeded(since, pdfBase64.length) + return { maskedTail: result.text.slice(-60) } + }) +} + +const startedAt = new Date().toISOString() +try { + await main() +} catch (error) { + checks.push({ name: 'setup', status: 'failed', durationMs: 0, error: getErrorMessage(error) }) +} +const failed = checks.filter((entry) => entry.status === 'failed') +const report = { + suite: 'pii-redaction-e2e', + startedAt, + finishedAt: new Date().toISOString(), + baseUrl: baseUrl.origin, + presidioUrl: presidioUrl.origin, + fixtures: { pdfBase64Chars: pdfBase64.length, longTextChars: longText.length }, + passed: checks.length - failed.length, + failed: failed.length, + checks, +} +await mkdir(dirname(reportPath), { recursive: true }) +await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`) +for (const entry of checks) { + logger.info('PII redaction check', { + name: entry.name, + status: entry.status, + durationMs: entry.durationMs, + error: entry.error, + }) +} +process.exit(failed.length === 0 ? 0 : 1) From 994e5400e8d2f83a54e4755e229a17c407297cb4 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 12 Nov 2026 21:16:39 -0800 Subject: [PATCH 2/4] fix(pii): mask long texts inside Presidio and limit encoded-file replacement to the log - Presidio runs pattern and custom recognizers over the whole text and spaCy NER in overlapping windows when a text exceeds spaCy's max_length, so no match can straddle a cut; the app no longer splits texts - Encoded file bodies are replaced only in the execution log (explicit replaceEncodedBinary option set by the log entry points); live data keeps every string on the masking path - Data URL headers are parsed through the first comma, and log-mode detection no longer requires mixed character classes - e2e: setup failures are recorded in the report --- .../platform/enterprise/data-retention.mdx | 2 +- apps/pii/server.py | 144 +++++++++---- .../api/guardrails/mask-batch/route.test.ts | 17 -- apps/sim/lib/guardrails/mask-client.test.ts | 74 ------- apps/sim/lib/guardrails/mask-client.ts | 43 ++-- apps/sim/lib/guardrails/pii-batching.ts | 105 +--------- apps/sim/lib/guardrails/validate-pii.ts | 19 +- .../logs/execution/pii-large-values.test.ts | 31 +++ .../lib/logs/execution/pii-large-values.ts | 9 +- .../lib/logs/execution/pii-redaction.test.ts | 29 +-- apps/sim/lib/logs/execution/pii-redaction.ts | 182 +++++++++-------- apps/sim/scripts/test-pii-redaction-e2e.ts | 193 ++++++++++-------- 12 files changed, 387 insertions(+), 461 deletions(-) diff --git a/apps/docs/content/docs/platform/enterprise/data-retention.mdx b/apps/docs/content/docs/platform/enterprise/data-retention.mdx index 02a0e768eea..c3252f45023 100644 --- a/apps/docs/content/docs/platform/enterprise/data-retention.mdx +++ b/apps/docs/content/docs/platform/enterprise/data-retention.mdx @@ -118,7 +118,7 @@ Redaction applies to text strings at the configured workflow stages. It does not If log redaction fails, affected strings are replaced with `[REDACTION_FAILED]` rather than stored unmasked. Failure at Workflow input or Block outputs stops the run instead of passing unredacted values downstream. -Base64-encoded file content, such as a `data:` URL or a long base64 string, can't be checked for PII as text. In stored logs it is replaced with `[BINARY_CONTENT_REDACTED]`. At Workflow input and Block outputs it passes through unchanged, so later blocks can still read the file. **Run from block** and re-running with a previous run's input read that run's log, so they see the marker in place of that content, the same way they see masked PII. +Base64-encoded file content, such as a `data:` URL or a long base64 string, can't be checked for PII as text, because any PII is inside the encoded file. In stored logs it is replaced with `[BINARY_CONTENT_REDACTED]`. At Workflow input and Block outputs it is masked like any other string. **Run from block** and re-running with a previous run's input read that run's log, so they see the marker in place of that content, the same way they see masked PII. ## Self-hosted setup diff --git a/apps/pii/server.py b/apps/pii/server.py index 970acc131df..81c91d4cb4e 100644 --- a/apps/pii/server.py +++ b/apps/pii/server.py @@ -228,30 +228,93 @@ def _regex_only(entities: list[str] | None, score_threshold: float | None) -> bo ) -def _require_nlp_length( - texts: list[str], +# spaCy refuses (error E088) any text longer than its model's `max_length`, so a +# longer text runs NER in windows of this many characters, following spaCy's +# guidance to segment long documents (it sizes NER memory at roughly 1GB per +# 100,000 characters of one text). +NER_WINDOW_CHARS = 100_000 +# How much of the previous window each window repeats. Far longer than any entity +# spaCy tags, so an entity cut by one window's edge lies whole inside the next. +NER_WINDOW_OVERLAP = 1_000 + + +def _exceeds_nlp_length(text: str, language: str) -> bool: + limit = _NLP_MAX_LENGTH.get(language) + return limit is not None and len(text) > limit + + +def _ner_window_entities(entities: list[str] | None, language: str) -> list[str]: + """The requested entities only spaCy NER produces and this language serves, or + all of them when `entities` is empty (detect-all).""" + ner = NER_ENTITIES & set(analyzer.get_supported_entities(language)) + return sorted(ner if not entities else ner.intersection(entities)) + + +def _analyze_long( + text: str, language: str, entities: list[str] | None, score_threshold: float | None, -) -> None: - """Refuse, as a 422, any text longer than the language model's spaCy - `max_length` when the request runs the NLP pass. spaCy raises ValueError E088 - for such a text, which FastAPI answers as an opaque 500 that callers retry as - a transient failure; the same text fails the same way every time, so callers - must split it rather than resend it. Regex-only requests skip spaCy and are - not limited.""" - if _regex_only(entities, score_threshold): - return - limit = _NLP_MAX_LENGTH.get(language) - if limit is None: - return - for index, text in enumerate(texts): - if len(text) > limit: - raise HTTPException( - status_code=422, - detail=f"Text {index} has {len(text)} characters; NER accepts at most " - f"{limit} per text. Split it before sending.", + return_decision_process: bool = False, + ad_hoc_recognizers: list[PatternRecognizer] | None = None, +) -> list[RecognizerResult]: + """Analyze a text longer than spaCy's `max_length`. + + Pattern, checksum, and custom recognizers have no length limit, so they run + once over the whole text (against blank NLP artifacts, as the regex-only path + does) and no match can straddle a cut. Only NER runs in overlapping windows + (see NER_WINDOW_CHARS), asked for just its own entities. A span touching a + window's far edge may be cut short, so it is left to the next window, which + sees it whole; a span ending inside the overlap before that edge was already + found by the previous window and is skipped. The rest are shifted to + whole-text offsets, and overlapping spans are left to the anonymizer, which + merges them. Context words cannot raise a pattern match's score here, which + matters only when a score_threshold is set.""" + entities = entities or None + results = analyzer.analyze( + text=text, + language=language, + entities=entities, + score_threshold=score_threshold, + return_decision_process=return_decision_process, + nlp_artifacts=_BLANK_ARTIFACTS.get(language), + ad_hoc_recognizers=ad_hoc_recognizers or None, + ) + window_entities = _ner_window_entities(entities, language) + if not window_entities: + return results + step = NER_WINDOW_CHARS - NER_WINDOW_OVERLAP + for start in range(0, len(text), step): + window = text[start : start + NER_WINDOW_CHARS] + is_last = start + NER_WINDOW_CHARS >= len(text) + seen_through = NER_WINDOW_OVERLAP if start else 0 + for result in analyzer.analyze( + text=window, + language=language, + entities=window_entities, + score_threshold=score_threshold, + return_decision_process=return_decision_process, + ): + if result.end < seen_through or (not is_last and result.end >= len(window)): + continue + results.append( + RecognizerResult( + result.entity_type, + result.start + start, + result.end + start, + result.score, + result.analysis_explanation, + result.recognition_metadata, + ) ) + if is_last: + break + unique: dict[tuple[str, int, int], RecognizerResult] = {} + for result in results: + key = (result.entity_type, result.start, result.end) + if key not in unique or result.score > unique[key].score: + unique[key] = result + return list(unique.values()) def _analyze_one( @@ -262,13 +325,15 @@ def _analyze_one( return_decision_process: bool = False, ad_hoc_recognizers: list[PatternRecognizer] | None = None, ): - _require_nlp_length([text], language, entities, score_threshold) + regex_only = _regex_only(entities, score_threshold) + if not regex_only and _exceeds_nlp_length(text, language): + return _analyze_long( + text, language, entities, score_threshold, return_decision_process, ad_hoc_recognizers + ) # Regex-only requests reuse a blank NlpArtifacts to skip the spaCy NLP pass; # otherwise analyze() computes artifacts (runs spaCy) as usual. Custom-pattern # recognizers are regex-based, so they run fine against the blank artifacts. - nlp_artifacts = ( - _BLANK_ARTIFACTS.get(language) if _regex_only(entities, score_threshold) else None - ) + nlp_artifacts = _BLANK_ARTIFACTS.get(language) if regex_only else None return analyzer.analyze( text=text, language=language, @@ -292,8 +357,9 @@ def _analyze_many( score_threshold: float | None, ad_hoc_recognizers: list[PatternRecognizer] | None = None, ): - """Analyze many texts, skipping the spaCy pass for regex-only requests.""" - _require_nlp_length(texts, language, entities, score_threshold) + """Analyze many texts, skipping the spaCy pass for regex-only requests and + analyzing any text longer than spaCy's `max_length` on its own (see + `_analyze_long`).""" if _regex_only(entities, score_threshold): blank = _BLANK_ARTIFACTS.get(language) return [ @@ -307,16 +373,24 @@ def _analyze_many( ) for text in texts ] - return list( - batch_analyzer.analyze_iterator( - texts=texts, - language=language, - batch_size=NLP_BATCH_SIZE, - entities=entities or None, - score_threshold=score_threshold, - ad_hoc_recognizers=ad_hoc_recognizers or None, - ) + long_indices = {i for i, text in enumerate(texts) if _exceeds_nlp_length(text, language)} + short_indices = [i for i in range(len(texts)) if i not in long_indices] + results: list[list[RecognizerResult]] = [[] for _ in texts] + short_results = batch_analyzer.analyze_iterator( + texts=[texts[i] for i in short_indices], + language=language, + batch_size=NLP_BATCH_SIZE, + entities=entities or None, + score_threshold=score_threshold, + ad_hoc_recognizers=ad_hoc_recognizers or None, ) + for i, per_text in zip(short_indices, short_results): + results[i] = list(per_text) + for i in sorted(long_indices): + results[i] = _analyze_long( + texts[i], language, entities, score_threshold, ad_hoc_recognizers=ad_hoc_recognizers + ) + return results @asynccontextmanager diff --git a/apps/sim/app/api/guardrails/mask-batch/route.test.ts b/apps/sim/app/api/guardrails/mask-batch/route.test.ts index eea962e905f..757b94bcf55 100644 --- a/apps/sim/app/api/guardrails/mask-batch/route.test.ts +++ b/apps/sim/app/api/guardrails/mask-batch/route.test.ts @@ -1,7 +1,6 @@ import { hybridAuthMockFns } from '@sim/testing/mocks/hybrid-auth.mock' import { createMockRequest } from '@sim/testing/mocks/request.mock' import { beforeEach, describe, expect, it, vi } from 'vitest' -import { PII_MAX_TEXT_CHARS } from '@/lib/guardrails/pii-batching' import { POST } from '@/app/api/guardrails/mask-batch/route' const mockCheckInternalAuth = hybridAuthMockFns.mockCheckInternalAuth @@ -52,20 +51,4 @@ describe('POST /api/guardrails/mask-batch', () => { expect(res.status).toBe(status) }) - - it('splits a text too long for one NER pass before calling Presidio, then rejoins it', async () => { - presidioFetch.mockImplementation(async (_url: string, init: { body: string }) => { - const { texts } = JSON.parse(init.body) as { texts: string[] } - return Response.json({ texts: texts.map((t) => t.replaceAll('Ada', '')) }) - }) - const text = 'Ada wrote this line.\n'.repeat(60_000) - - const res = await POST(createMockRequest('POST', { texts: [text], entityTypes: ['PERSON'] })) - - expect(res.status).toBe(200) - expect(await res.json()).toEqual({ masked: [text.replaceAll('Ada', '')] }) - const sent = presidioFetch.mock.calls.flatMap(([, init]) => JSON.parse(init.body).texts) - expect(sent.length).toBeGreaterThan(1) - for (const piece of sent) expect(piece.length).toBeLessThanOrEqual(PII_MAX_TEXT_CHARS) - }) }) diff --git a/apps/sim/lib/guardrails/mask-client.test.ts b/apps/sim/lib/guardrails/mask-client.test.ts index 6696f9f250e..9da95d11d06 100644 --- a/apps/sim/lib/guardrails/mask-client.test.ts +++ b/apps/sim/lib/guardrails/mask-client.test.ts @@ -16,7 +16,6 @@ vi.mock('@/lib/auth/internal', () => authInternalMock) vi.mock('@sim/utils/helpers', () => utilsHelpersMock) import { maskPIIBatchViaHttp } from '@/lib/guardrails/mask-client' -import { PII_MAX_TEXT_CHARS } from '@/lib/guardrails/pii-batching' import { POST as maskBatchRoute } from '@/app/api/guardrails/mask-batch/route' const mockSleep = utilsHelpersMockFns.mockSleep @@ -82,79 +81,6 @@ describe('maskPIIBatchViaHttp', () => { expect(mockSleep).toHaveBeenCalledTimes(2) }) - describe('with a text longer than one Presidio pass allows', () => { - /** Each request masks `Ada` to ``, so pieces rejoin to the masked whole. */ - const maskingAda = () => - fetchMock.mockImplementation(async (_url: string, init: { body: string }) => { - const { texts } = JSON.parse(init.body) as { texts: string[] } - return Response.json({ masked: texts.map((t) => t.replaceAll('Ada', '')) }) - }) - const sentPieces = () => - fetchMock.mock.calls.flatMap(([, init]) => JSON.parse(init.body).texts as string[]) - - it('sends newline-aligned pieces within the limit and rejoins the masked result', async () => { - maskingAda() - const text = Array.from({ length: 60_000 }, (_, i) => `row ${i} owner Ada\n`).join('') - - const { masked } = await maskPIIBatchViaHttp([text, 'short Ada'], ['PERSON']) - - expect(masked).toEqual([text.replaceAll('Ada', ''), 'short ']) - const pieces = sentPieces().filter((p) => p !== 'short Ada') - expect(pieces.length).toBeGreaterThan(1) - expect(pieces.join('')).toBe(text) - for (const piece of pieces) expect(piece.length).toBeLessThanOrEqual(PII_MAX_TEXT_CHARS) - for (const piece of pieces.slice(0, -1)) expect(piece.endsWith('\n')).toBe(true) - }) - - it('cuts minified JSON at punctuation so no email is split across pieces', async () => { - maskingAda() - const record = '{"email":"ada.lovelace@example.com","id":1234}' - const text = `[${Array.from({ length: 25_000 }, () => record).join(',')}]` - - await maskPIIBatchViaHttp([text], ['EMAIL_ADDRESS']) - - const pieces = sentPieces() - expect(pieces.length).toBeGreaterThan(1) - const emailsSent = pieces.reduce( - (count, piece) => count + piece.split('ada.lovelace@example.com').length - 1, - 0 - ) - expect(emailsSent).toBe(25_000) - }) - - it('never splits a surrogate pair when the text has nothing to cut at', async () => { - // One leading unit puts a high surrogate exactly where a hard cut would land. - const text = `a${'\u{1F600}'.repeat(PII_MAX_TEXT_CHARS)}` - - const { masked } = await maskPIIBatchViaHttp([text], []) - - const pieces = sentPieces() - expect(pieces.length).toBeGreaterThan(1) - expect(pieces.join('')).toBe(text) - expect(masked).toEqual([`M(${pieces.join(')M(')})`]) - }) - - it('replaces the whole text when any one of its pieces fails', async () => { - const text = `${'safe line\n'.repeat(120_000)}poison` - fetchMock.mockImplementation(async (_url: string, init: { body: string }) => { - const { texts } = JSON.parse(init.body) as { texts: string[] } - if (texts.some((t) => t.endsWith('poison'))) return new Response('no', { status: 422 }) - return Response.json({ masked: texts }) - }) - - const { masked, scrubbedCount } = await maskPIIBatchViaHttp( - [text], - [], - undefined, - undefined, - { failedChunkPlaceholder: '[FAILED]' } - ) - - expect(masked).toEqual(['[FAILED]']) - expect(scrubbedCount).toBe(1) - }) - }) - describe('with a failed-chunk placeholder', () => { const PLACEHOLDER = '[FAILED]' const CHUNK_SIZE = 2000 diff --git a/apps/sim/lib/guardrails/mask-client.ts b/apps/sim/lib/guardrails/mask-client.ts index 3437f7e4ac3..ddd9199b08d 100644 --- a/apps/sim/lib/guardrails/mask-client.ts +++ b/apps/sim/lib/guardrails/mask-client.ts @@ -8,7 +8,7 @@ import { generateInternalToken } from '@/lib/auth/internal' import { env } from '@/lib/core/config/env' import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import { getInternalApiBaseUrl } from '@/lib/core/utils/urls' -import { chunkIndicesByBudget, joinSegments, segmentTexts } from '@/lib/guardrails/pii-batching' +import { chunkIndicesByBudget } from '@/lib/guardrails/pii-batching' import type { CustomPiiPattern } from '@/lib/guardrails/pii-entities' const logger = createLogger('PiiMaskClient') @@ -106,9 +106,9 @@ function isOutageError(error: unknown): boolean { interface MaskPIIBatchOptions { /** - * When set, a chunk that cannot be masked has every string with a piece in it - * replaced with this value instead of rejecting the call, so one bad chunk costs - * only its own strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's + * When set, a chunk that cannot be masked has each of its strings replaced with + * this value instead of rejecting the call, so one bad chunk costs only its own + * strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's * retries, chunks not yet sent are replaced the same way. */ failedChunkPlaceholder?: string @@ -127,9 +127,7 @@ interface MaskPIIBatchResult { * Only the app task reaches the Presidio service (it holds `PII_URL`), but the * log-redaction persist path also runs inside the trigger.dev runtime — so * redaction always routes through HTTP, the same way the guardrails tool does. - * A string too long for one NER pass is cut into pieces first (see - * {@link segmentTexts}), so no request carries a text Presidio would refuse. - * Pieces are grouped into byte/count-budgeted chunks (keeping each request far + * Strings are grouped into byte/count-budgeted chunks (keeping each request far * under the 10MB Next body limit) and the chunks are sent with bounded * concurrency, so a large payload fans out rather than serializing; order is * preserved, so the returned `masked` array matches `texts` length. @@ -152,47 +150,38 @@ export async function maskPIIBatchViaHttp( const url = `${getInternalApiBaseUrl()}/api/guardrails/mask-batch` const placeholder = options.failedChunkPlaceholder - const { segments, owners } = segmentTexts(texts) - const maskedSegments = new Array(segments.length) - const scrubbedTexts = new Set() + const masked = new Array(texts.length) let serviceUnavailable = false + let scrubbedCount = 0 let firstFailure: unknown - const scrubChunk = (indices: number[]) => { - for (const k of indices) scrubbedTexts.add(owners[k]) + const scrubChunk = (indices: number[], value: string) => { + for (const i of indices) masked[i] = value + scrubbedCount += indices.length } - await mapWithConcurrency(chunkIndicesByBudget(segments), CHUNK_CONCURRENCY, async (indices) => { + await mapWithConcurrency(chunkIndicesByBudget(texts), CHUNK_CONCURRENCY, async (indices) => { if (serviceUnavailable && placeholder !== undefined) { - scrubChunk(indices) + scrubChunk(indices, placeholder) return } try { - const chunk = indices.map((k) => segments[k]) + const chunk = indices.map((i) => texts[i]) const out = await postChunk(url, chunk, entityTypes, language, customPatterns) if (out.length !== chunk.length) { throw new Error('PII mask-batch returned an unexpected result') } - indices.forEach((segmentIndex, k) => { - maskedSegments[segmentIndex] = out[k] + indices.forEach((originalIndex, k) => { + masked[originalIndex] = out[k] }) } catch (error) { if (placeholder === undefined) throw error if (isOutageError(error)) serviceUnavailable = true firstFailure ??= error - scrubChunk(indices) + scrubChunk(indices, placeholder) } }) - const joined = joinSegments(maskedSegments, owners, texts.length) - // A text with any unmasked piece is replaced whole: its other pieces are masked, - // but keeping them would persist a fragment around the hole. - const masked = - placeholder === undefined - ? joined - : joined.map((text, i) => (scrubbedTexts.has(i) ? placeholder : text)) - const scrubbedCount = scrubbedTexts.size - if (scrubbedCount > 0) { logger.error('PII masking failed for some chunks; replaced their strings', { error: getErrorMessage(firstFailure), diff --git a/apps/sim/lib/guardrails/pii-batching.ts b/apps/sim/lib/guardrails/pii-batching.ts index 8c5470f84b7..2dd419d45cc 100644 --- a/apps/sim/lib/guardrails/pii-batching.ts +++ b/apps/sim/lib/guardrails/pii-batching.ts @@ -1,8 +1,8 @@ /** * Per-request bounds shared by both Presidio hops: the app→route HTTP call * (`mask-client`) and the route→service call (`validate-pii`). Keeping a single - * source of truth ensures every request stays far under the 10MB Next body limit, - * every text fits one spaCy NER pass, and each request is one short pass. + * source of truth ensures every request stays far under the 10MB Next body limit + * and small enough for one short spaCy NER pass per Presidio request. */ /** Max UTF-8 bytes of text per Presidio request. ~40× under the 10MB Next limit. */ @@ -10,107 +10,6 @@ export const PII_REQUEST_MAX_BYTES = 256 * 1024 /** Max strings per request; caps per-item overhead and stays well under the contract's 100k-entry cap. */ export const PII_REQUEST_MAX_COUNT = 2_000 -/** - * Max characters of one text in one Presidio NER pass: spaCy's default - * `max_length`, past which it refuses the text outright (the Presidio service - * answers 422). Every text Presidio accepts is still sent whole, so no entity can - * straddle a cut in it; only a longer text, which could never be masked whole, is - * split by {@link segmentTexts}, following spaCy's guidance to segment long - * documents before NER. A JavaScript length counts UTF-16 units, never fewer than - * the code points spaCy counts, so a text within it is always within spaCy's. - */ -export const PII_MAX_TEXT_CHARS = 1_000_000 - -/** - * The shortest piece a cut may leave, so a boundary is never chosen so early that - * a long text splinters into many tiny requests. - */ -const MIN_SEGMENT_CHARS = PII_MAX_TEXT_CHARS / 2 - -/** Letters, combining marks, digits, and `_`: characters an entity can run through. */ -const WORD_CHARACTER = /[\p{L}\p{M}\p{N}_]/u - -/** - * Structural delimiters (JSON, CSV, markup) that no detected entity contains, unlike - * the `.`, `@`, `-`, `+`, `:`, and `/` inside emails, URLs, phone numbers, and ids. - */ -const STRUCTURAL_DELIMITER = /[,;"`()[\]{}<>|]/ - -function isSurrogate(code: number): boolean { - return code >= 0xd800 && code <= 0xdfff -} - -function isHighSurrogate(code: number): boolean { - return code >= 0xd800 && code <= 0xdbff -} - -/** - * Where to end the piece of `text` that starts at `start`, at most - * {@link PII_MAX_TEXT_CHARS} later and no earlier than {@link MIN_SEGMENT_CHARS}. - * Prefers the end of a line, then whitespace, then a structural delimiter, then any - * other punctuation, since NER and Presidio's pattern recognizers match within - * those boundaries and an entity that straddles a cut would be missed. Only a run - * with no boundary at all is cut mid-run, and never between the two halves of a - * surrogate pair. - */ -function cutPoint(text: string, start: number): number { - const limit = start + PII_MAX_TEXT_CHARS - const floor = start + MIN_SEGMENT_CHARS - - const newline = text.slice(floor, limit).lastIndexOf('\n') - if (newline !== -1) return floor + newline + 1 - - for (let i = limit - 1; i >= floor; i--) { - if (/\s/.test(text[i])) return i + 1 - } - for (let i = limit - 1; i >= floor; i--) { - if (STRUCTURAL_DELIMITER.test(text[i])) return i + 1 - } - for (let i = limit - 1; i >= floor; i--) { - if (!isSurrogate(text.charCodeAt(i)) && !WORD_CHARACTER.test(text[i])) return i + 1 - } - return isHighSurrogate(text.charCodeAt(limit - 1)) ? limit - 1 : limit -} - -/** Every text's pieces, in order, each within {@link PII_MAX_TEXT_CHARS}. */ -export interface SegmentedTexts { - segments: string[] - /** `owners[k]` is the index of the input text `segments[k]` was cut from. */ - owners: number[] -} - -/** - * Split each text longer than {@link PII_MAX_TEXT_CHARS} into consecutive pieces - * (see {@link cutPoint}); shorter texts pass through as one piece. The pieces of a - * text concatenate back to it exactly, so masking them one by one and joining them - * with {@link joinSegments} masks the whole text. - */ -export function segmentTexts(texts: string[]): SegmentedTexts { - const segments: string[] = [] - const owners: number[] = [] - texts.forEach((text, owner) => { - let start = 0 - while (text.length - start > PII_MAX_TEXT_CHARS) { - const end = cutPoint(text, start) - segments.push(text.slice(start, end)) - owners.push(owner) - start = end - } - segments.push(text.slice(start)) - owners.push(owner) - }) - return { segments, owners } -} - -/** Concatenate masked pieces back into one string per original text, in order. */ -export function joinSegments(pieces: string[], owners: number[], textCount: number): string[] { - const joined = new Array(textCount).fill('') - pieces.forEach((piece, k) => { - joined[owners[k]] += piece - }) - return joined -} - /** * Group `texts` into chunks of original indices, flushing a chunk when adding the * next string would exceed {@link PII_REQUEST_MAX_BYTES} or {@link PII_REQUEST_MAX_COUNT}. diff --git a/apps/sim/lib/guardrails/validate-pii.ts b/apps/sim/lib/guardrails/validate-pii.ts index a12cc72a4a3..962870c61e6 100644 --- a/apps/sim/lib/guardrails/validate-pii.ts +++ b/apps/sim/lib/guardrails/validate-pii.ts @@ -8,7 +8,7 @@ import { readResponseJsonWithLimit, readResponseTextWithLimit, } from '@/lib/core/utils/stream-limits' -import { chunkIndicesByBudget, joinSegments, segmentTexts } from '@/lib/guardrails/pii-batching' +import { chunkIndicesByBudget } from '@/lib/guardrails/pii-batching' import type { CustomPiiPattern } from '@/lib/guardrails/pii-entities' import { MAX_PII_VALIDATION_DETECTED_ENTITIES, @@ -501,11 +501,9 @@ export async function validatePII(input: PIIValidationInput): Promise { if (texts.length === 0) return [] - const { segments, owners } = segmentTexts(texts) - const result = new Array(segments.length) + const result = new Array(texts.length) - await mapWithConcurrency(chunkIndicesByBudget(segments), CHUNK_CONCURRENCY, async (indices) => { - const chunkTexts = indices.map((i) => segments[i]) + await mapWithConcurrency(chunkIndicesByBudget(texts), CHUNK_CONCURRENCY, async (indices) => { + const chunkTexts = indices.map((i) => texts[i]) if (combinedRedactAvailable) { const masked = await redactBatch(chunkTexts, entityTypes, language, customPatterns) @@ -574,7 +571,7 @@ export async function maskPIIBatch( }) }) - return joinSegments(result, owners, texts.length) + return result } export { type PIIEntityType, SUPPORTED_PII_ENTITIES } from '@/lib/guardrails/pii-entities' diff --git a/apps/sim/lib/logs/execution/pii-large-values.test.ts b/apps/sim/lib/logs/execution/pii-large-values.test.ts index 90f7f9ba227..ad917046339 100644 --- a/apps/sim/lib/logs/execution/pii-large-values.test.ts +++ b/apps/sim/lib/logs/execution/pii-large-values.test.ts @@ -289,6 +289,37 @@ describe('redactManifest — chunk-wise', () => { }) }) +describe('encoded file bodies inside large values', () => { + const encodedFile = Buffer.from( + Array.from({ length: 48 * 1024 }, (_, i) => (i * 7919 + 13) % 256) + ).toString('base64') + + it('replaces them in the execution log', async () => { + chunkData.set(REF.id, { file: encodedFile, note: 'contact bob' }) + + const result = await redactLargeValueRefs( + { finalOutput: REF }, + { entityTypes: ['PERSON'], language: 'en', store: STORE } + ) + + expect(result.finalOutput).toEqual({ + file: '[BINARY_CONTENT_REDACTED]', + note: 'MASKED(contact bob)', + }) + }) + + it('masks them as text in live block state', async () => { + chunkData.set(REF.id, { file: encodedFile }) + + const result = await redactLargeValueRefsInValue( + { 'block-1': { output: REF } }, + { entityTypes: ['PERSON'], language: 'en', store: STORE, onFailure: 'throw' } + ) + + expect(result).toEqual({ 'block-1': { output: { file: `MASKED(${encodedFile})` } } }) + }) +}) + describe('redactLargeValueRefsInValue (arbitrary blockStates)', () => { it('hydrates + re-stores a ref nested in a non-RedactablePayload shape', async () => { chunkData.set(REF.id, { note: 'contact bob' }) diff --git a/apps/sim/lib/logs/execution/pii-large-values.ts b/apps/sim/lib/logs/execution/pii-large-values.ts index bcefb5381ca..98f241a17c8 100644 --- a/apps/sim/lib/logs/execution/pii-large-values.ts +++ b/apps/sim/lib/logs/execution/pii-large-values.ts @@ -43,6 +43,12 @@ export interface RedactLargeValueRefsOptions { * than feeding a marker (or leaking raw bytes) downstream. */ onFailure?: PiiRedactionFailureMode + /** + * @internal Set by {@link redactLargeValueRefs}, the execution log's entry point, + * so encoded file bodies inside its large values are replaced like inline ones + * (see `PiiRedactionOptions.replaceEncodedBinary`). Callers never set it. + */ + replaceEncodedBinary?: boolean /** * @internal Shared across every nested invocation spawned from one entry-point * call, so oversized hydrations serialize globally — not just within their own @@ -100,7 +106,7 @@ export async function redactLargeValueRefs( payload: RedactablePayload, options: RedactLargeValueRefsOptions ): Promise { - const gatedOptions = withOversizedGate(options) + const gatedOptions = withOversizedGate({ ...options, replaceEncodedBinary: true }) // Collect refs across the WHOLE payload first (shared `seen`), so every ref is // hydrated+masked+re-stored in one bounded-concurrency pass instead of one // sequential pass per key. A ref shared across keys is walked/masked once. @@ -269,6 +275,7 @@ async function maskMaterializedValue( language: options.language, customPatterns: options.customPatterns, onFailure: options.onFailure ?? 'scrub', + replaceEncodedBinary: options.replaceEncodedBinary, }) return redactValueRefs(masked, options) } diff --git a/apps/sim/lib/logs/execution/pii-redaction.test.ts b/apps/sim/lib/logs/execution/pii-redaction.test.ts index 3cf599657c6..0ca7e19b5ae 100644 --- a/apps/sim/lib/logs/execution/pii-redaction.test.ts +++ b/apps/sim/lib/logs/execution/pii-redaction.test.ts @@ -93,7 +93,7 @@ describe('redactPIIFromExecution', () => { }) it('masks large strings too (never left unredacted)', async () => { - const big = 'x'.repeat(200 * 1024) + const big = 'lorem ipsum '.repeat(17_000) const payload = { finalOutput: { big, small: 'pii' } } const result = await redactPIIFromExecution(payload, { entityTypes: [] }) @@ -151,15 +151,14 @@ describe('encoded binary content', () => { expect(sentTexts()).toEqual(['Ada Lovelace']) }) - it('passes encoded file bodies through live data unchanged and unsent', async () => { - const value = { content: encodedFile, url: `data:image/png;base64,${encodedFile}`, note: 'hi' } + it('keeps live data on the masking path, so no string skips redaction by its shape', async () => { + const shapedLikeAFile = `${'Ab1_'.repeat(256)}123-45-6789` + const value = { content: encodedFile, id: shapedLikeAFile } const result = await redactObjectStrings(value, { entityTypes: [], onFailure: 'throw' }) - expect(result.content).toBe(encodedFile) - expect(result.url).toBe(`data:image/png;base64,${encodedFile}`) - expect(result.note).toBe('MASKED(hi)') - expect(sentTexts()).toEqual(['hi']) + expect(result).toEqual({ content: `MASKED(${encodedFile})`, id: `MASKED(${shapedLikeAFile})` }) + expect(sentTexts()).toEqual([encodedFile, shapedLikeAFile]) }) /** `encodedFile` wrapped at `width` columns, as MIME (76) and PEM (64) encoders emit it. */ @@ -171,6 +170,10 @@ describe('encoded binary content', () => { ['wrapped at 76 columns', wrapped(76, '\n')], ['wrapped at 64 columns with CRLF', `${wrapped(64, '\r\n')}\r\n`], ['as a wrapped data URL', `data:application/pdf;base64,${wrapped(76, '\n')}`], + [ + 'as a data URL with long metadata', + `data:application/pdf;name=${'quarterly-report-'.repeat(40)}.pdf;base64,${encodedFile}`, + ], ])('replaces an encoded file body %s in a persisted log', async (_label, body) => { const result = await redactPIIFromExecution({ finalOutput: body }, { entityTypes: [] }) @@ -180,15 +183,13 @@ describe('encoded binary content', () => { it.each([ ['a newline-separated list of usernames', 'AdaLovelace1815\n'.repeat(100)], - ['a long run of digits', '4'.repeat(4096)], - ['lowercase hex', 'deadbeef0123'.repeat(400)], ['a short base64 token', 'QWRhIExvdmVsYWNl'], - ['a long space-separated sentence', 'Ada Lovelace wrote code. '.repeat(200)], - ])('still masks %s, which is not an encoded file', async (_label, text) => { - const result = await redactObjectStrings({ text }, { entityTypes: [], onFailure: 'throw' }) + ['a long space-separated sentence', 'Ada Lovelace wrote code in 1843 '.repeat(150)], + ])('still masks %s in a persisted log, which is not an encoded file', async (_label, text) => { + const result = await redactPIIFromExecution({ finalOutput: text }, { entityTypes: [] }) - expect(result.text).not.toBe(text) - expect(sentTexts().join('')).toBe(text) + expect(result.finalOutput).toBe(`MASKED(${text})`) + expect(sentTexts()).toEqual([text]) }) }) diff --git a/apps/sim/lib/logs/execution/pii-redaction.ts b/apps/sim/lib/logs/execution/pii-redaction.ts index 04c298d05d4..433458910fd 100644 --- a/apps/sim/lib/logs/execution/pii-redaction.ts +++ b/apps/sim/lib/logs/execution/pii-redaction.ts @@ -10,7 +10,7 @@ const logger = createLogger('PiiRedaction') /** Replaces text we could not safely mask, so PII is never persisted on failure. */ export const REDACTION_FAILED_MARKER = '[REDACTION_FAILED]' -/** Replaces an encoded file body in a persisted record (see {@link isEncodedBinary}). */ +/** Replaces an encoded file body in the execution log (see {@link isEncodedBinary}). */ export const BINARY_CONTENT_REDACTED_MARKER = '[BINARY_CONTENT_REDACTED]' /** @@ -21,7 +21,8 @@ const MIN_ENCODED_BINARY_CHARS = 1024 const STANDARD_BASE64 = /^[A-Za-z0-9+/]+={0,2}$/ const URL_SAFE_BASE64 = /^[A-Za-z0-9_-]+={0,2}$/ -const BASE64_DATA_URL = /^data:(?:[\w.+-]+\/[\w.+-]+)?(?:;[\w.+-]+=[^;,]*)*;base64,/i +/** A `data:` URL's header through its comma, declaring base64 (RFC 2397). */ +const BASE64_DATA_URL_HEADER = /^data:(?:[\w.+-]+\/[\w.+-]+)?(?:;[\w.+-]+=[^;,]*)*;base64,$/i /** Narrowest and widest lines encoders wrap base64 at: 60 (common), 64 (PEM), 76 (MIME). */ const MIN_WRAP_WIDTH = 60 @@ -32,67 +33,53 @@ function isBase64(value: string): boolean { } /** - * `value` with trailing whitespace dropped and, when it is wrapped the way base64 - * encoders wrap (every line but the last one fixed width, a multiple of 4 between - * {@link MIN_WRAP_WIDTH} and {@link MAX_WRAP_WIDTH}), its line breaks removed. - * Returns `null` for multi-line text without that shape, which is never base64. + * The base64 in `value` with trailing whitespace dropped and, when it is wrapped + * the way encoders wrap it (every line but the last a fixed width, a multiple of 4 + * between {@link MIN_WRAP_WIDTH} and {@link MAX_WRAP_WIDTH}), its line breaks + * removed. Returns `null` when `value` is not base64. */ -function unwrapBase64(value: string): string | null { +function base64Body(value: string): string | null { const body = value.trimEnd() const firstBreak = body.indexOf('\n') - if (firstBreak === -1) return body + if (firstBreak === -1) return isBase64(body) ? body : null const width = body[firstBreak - 1] === '\r' ? firstBreak - 1 : firstBreak if (width < MIN_WRAP_WIDTH || width > MAX_WRAP_WIDTH || width % 4 !== 0) return null + if (!isBase64(body.slice(0, width))) return null const lines = body.split(/\r?\n/) const last = lines[lines.length - 1] - const fixedWidth = lines.slice(0, -1).every((line) => line.length === width) - return fixedWidth && last.length > 0 && last.length <= width ? lines.join('') : null + const fixedWidth = lines.every((line, i) => i === lines.length - 1 || line.length === width) + if (!fixedWidth || last.length === 0 || last.length > width) return null + const joined = lines.join('') + return isBase64(joined) ? joined : null } /** * Whether `value` is an encoded file body: a `data:` URL declaring base64, or a * bare base64 string at least {@link MIN_ENCODED_BINARY_CHARS} long, either one - * optionally wrapped the way encoders wrap it (see {@link unwrapBase64}). A bare - * string must also mix upper case, lower case, and digits, as encoded bytes always - * do, so a run of digits, lowercase hex, or a list of codes is still masked as text. - * - * Presidio cannot redact such a value: spaCy reads base64 as one token, and the - * pattern recognizers only ever match it by coincidence at a `+` or `/`, where a - * replacement would corrupt the bytes rather than remove anything readable. Any - * PII is inside the decoded file, which text masking never sees. + * optionally wrapped the way encoders wrap it (see {@link base64Body}). Text + * masking cannot check such a value for PII, which lives inside the decoded file. */ function isEncodedBinary(value: string): boolean { if (value.startsWith('data:')) { - const prefix = BASE64_DATA_URL.exec(value.slice(0, 512)) - if (prefix === null) return false - const payload = unwrapBase64(value.slice(prefix[0].length)) - return payload !== null && isBase64(payload) + const comma = value.indexOf(',') + return ( + comma !== -1 && + BASE64_DATA_URL_HEADER.test(value.slice(0, comma + 1)) && + base64Body(value.slice(comma + 1)) !== null + ) } if (value.length < MIN_ENCODED_BINARY_CHARS) return false - const body = unwrapBase64(value) - return ( - body !== null && - body.length >= MIN_ENCODED_BINARY_CHARS && - isBase64(body) && - /[A-Z]/.test(body) && - /[a-z]/.test(body) && - /\d/.test(body) - ) + const body = base64Body(value) + return body !== null && body.length >= MIN_ENCODED_BINARY_CHARS } /** - * Whether a redaction pass is producing the execution record or data a later step - * reads, which decides both how a masking failure is handled and what becomes of - * an encoded file body Presidio cannot mask: - * - `'scrub'` (default), the execution log: a failure replaces the affected strings - * with {@link REDACTION_FAILED_MARKER}, and an encoded file body is replaced with - * {@link BINARY_CONTENT_REDACTED_MARKER}, since the log must not keep content - * that cannot be checked for PII. Safe because execution already succeeded. - * - `'throw'`, data a later step reads: workflow input, block outputs, and agent - * memory stored for later turns. A failure throws {@link PiiRedactionError} and - * aborts the run, since a marker would corrupt that data and fail-open would - * leak. An encoded file body passes through unchanged, since the step that - * reads it needs its bytes and masking could only corrupt them. + * How to handle a masking failure (Presidio error or over-ceiling payload): + * - `'scrub'` (default): replace eligible strings with {@link REDACTION_FAILED_MARKER}. + * Safe for the log stage — execution already succeeded. + * - `'throw'`: throw {@link PiiRedactionError}. Used for the execution-altering + * stages (input/block outputs) where a marker would corrupt computed data and + * fail-open would leak — so the run aborts instead. */ export type PiiRedactionFailureMode = 'scrub' | 'throw' @@ -104,6 +91,16 @@ export interface PiiRedactionOptions { customPatterns?: CustomPiiPattern[] /** Failure handling. Defaults to `'scrub'`. */ onFailure?: PiiRedactionFailureMode + /** + * Replace encoded file bodies (see {@link isEncodedBinary}) with + * {@link BINARY_CONTENT_REDACTED_MARKER} instead of masking them as text. Only + * for the execution log: masking cannot see PII inside a file, so the log drops + * the file rather than keep content it never checked. Replacing can only remove + * content, so a string misread as a file costs log detail, never PII. Live data + * leaves this off, keeping every string on the masking path so none can skip + * redaction by its shape, and keeping file bytes the next block reads. + */ + replaceEncodedBinary?: boolean } /** Thrown when in-flight redaction (`onFailure: 'throw'`) cannot mask safely. */ @@ -156,11 +153,12 @@ function isEligibleString(value: string): boolean { /** * What one pass does with each string leaf: `text` for a string Presidio masks, - * `encodedBinary` for an encoded file body it never sees (see {@link isEncodedBinary}). + * and, when set, `encodedBinary` for an encoded file body it never sees (see + * {@link isEncodedBinary}). Without `encodedBinary` every string is text. */ interface StringVisitor { text: (s: string) => string - encodedBinary: (s: string) => string + encodedBinary?: (s: string) => string } /** @@ -172,7 +170,9 @@ interface StringVisitor { function transformStrings(value: unknown, visit: StringVisitor): unknown { if (typeof value === 'string') { if (!isEligibleString(value)) return value - return isEncodedBinary(value) ? visit.encodedBinary(value) : visit.text(value) + return visit.encodedBinary && isEncodedBinary(value) + ? visit.encodedBinary(value) + : visit.text(value) } // Treat offloaded large-value refs as opaque: masking their internal `key`/`id` // strings would corrupt the reference. Their content is handled separately @@ -224,26 +224,48 @@ function transformUnit( return transformStrings(value, visit) } -/** What an encoded file body becomes under `onFailure` (see {@link PiiRedactionFailureMode}). */ -function encodedBinaryReplacement(onFailure: PiiRedactionFailureMode): (s: string) => string { - return onFailure === 'scrub' ? () => BINARY_CONTENT_REDACTED_MARKER : (s) => s +/** What a collection pass found: the strings to mask and how many encoded file bodies. */ +interface CollectedStrings { + collected: string[] + encodedBinaryCount: number } -/** A collection pass: records every string to mask and counts encoded file bodies. */ -function collectStrings() { +function collectStrings( + walk: (visit: StringVisitor) => void, + replaceEncodedBinary: boolean +): CollectedStrings { const collected: string[] = [] let encodedBinaryCount = 0 - const visit: StringVisitor = { + walk({ text: (s) => { collected.push(s) return s }, - encodedBinary: (s) => { - encodedBinaryCount++ - return s + encodedBinary: replaceEncodedBinary + ? (s) => { + encodedBinaryCount++ + return s + } + : undefined, + }) + return { collected, encodedBinaryCount } +} + +/** Mask the collected strings and return the substitution pass's visitor. */ +async function substitutionVisitor( + collected: string[], + options: PiiRedactionOptions, + replaceEncodedBinary: boolean +): Promise<{ visit: StringVisitor; scrubbedCount: number }> { + const { masked, scrubbedCount } = await maskCollected(collected, options) + let index = 0 + return { + visit: { + text: () => masked[index++], + encodedBinary: replaceEncodedBinary ? () => BINARY_CONTENT_REDACTED_MARKER : undefined, }, + scrubbedCount, } - return { collected, visit, encodedBinaryCount: () => encodedBinaryCount } } /** @@ -302,19 +324,12 @@ async function maskCollected( * block-output stages, so it defaults callers toward `onFailure: 'throw'`. */ export async function redactObjectStrings(value: T, options: PiiRedactionOptions): Promise { - const collection = collectStrings() - transformStrings(value, collection.visit) - const { collected } = collection - - if (collected.length === 0 && collection.encodedBinaryCount() === 0) return value + const replaceEncodedBinary = options.replaceEncodedBinary ?? false + const found = collectStrings((visit) => transformStrings(value, visit), replaceEncodedBinary) + if (found.collected.length === 0 && found.encodedBinaryCount === 0) return value - const { masked } = - collected.length === 0 ? { masked: [] } : await maskCollected(collected, options) - let index = 0 - return transformStrings(value, { - text: () => masked[index++], - encodedBinary: encodedBinaryReplacement(options.onFailure ?? 'scrub'), - }) as T + const { visit } = await substitutionVisitor(found.collected, options, replaceEncodedBinary) + return transformStrings(value, visit) as T } /** @@ -327,7 +342,9 @@ export async function redactObjectStrings(value: T, options: PiiRedactionOpti * envelope). On a hard masking failure, the eligible strings of every request * chunk that failed are replaced with {@link REDACTION_FAILED_MARKER} (the default * `onFailure: 'scrub'`) rather than left unredacted — PII is never persisted on - * the failure path. + * the failure path. This payload is the execution log, so encoded file bodies are + * always replaced with {@link BINARY_CONTENT_REDACTED_MARKER} (see + * {@link PiiRedactionOptions.replaceEncodedBinary}). */ export async function redactPIIFromExecution( payload: RedactablePayload, @@ -340,32 +357,21 @@ export async function redactPIIFromExecution( value: payload[key], })) - const collection = collectStrings() - for (const unit of units) transformUnit(unit.key, unit.value, collection.visit) - const { collected } = collection - const encodedBinaryCount = collection.encodedBinaryCount() - - if (collected.length === 0 && encodedBinaryCount === 0) return payload + const found = collectStrings((visit) => { + for (const unit of units) transformUnit(unit.key, unit.value, visit) + }, true) + if (found.collected.length === 0 && found.encodedBinaryCount === 0) return payload - const { masked, scrubbedCount } = - collected.length === 0 - ? { masked: [], scrubbedCount: 0 } - : await maskCollected(collected, options) - - let index = 0 - const visit: StringVisitor = { - text: () => masked[index++], - encodedBinary: encodedBinaryReplacement(options.onFailure ?? 'scrub'), - } + const { visit, scrubbedCount } = await substitutionVisitor(found.collected, options, true) const result: RedactablePayload = { ...payload } for (const unit of units) { result[unit.key] = transformUnit(unit.key, unit.value, visit) } logger.info('PII redaction completed', { - stringCount: collected.length, - totalBytes: collected.reduce((sum, s) => sum + Buffer.byteLength(s, 'utf8'), 0), - encodedBinaryCount, + stringCount: found.collected.length, + totalBytes: found.collected.reduce((sum, s) => sum + Buffer.byteLength(s, 'utf8'), 0), + encodedBinaryCount: found.encodedBinaryCount, durationMs: Math.round(performance.now() - startedAt), scrubbedCount, }) diff --git a/apps/sim/scripts/test-pii-redaction-e2e.ts b/apps/sim/scripts/test-pii-redaction-e2e.ts index 7b5d2400c34..97956436fb1 100644 --- a/apps/sim/scripts/test-pii-redaction-e2e.ts +++ b/apps/sim/scripts/test-pii-redaction-e2e.ts @@ -1,5 +1,4 @@ import assert from 'node:assert/strict' -import { randomBytes } from 'node:crypto' import { mkdir, writeFile } from 'node:fs/promises' import { dirname } from 'node:path' import { createLogger } from '@sim/logger' @@ -10,80 +9,41 @@ import { getErrorMessage } from '@sim/utils/errors' * app and a real Presidio container (`docker/pii.Dockerfile`, the app's `PII_URL`). * The redaction library runs in this process the way the log persist path runs in * a worker: every mask request crosses the app's `/api/guardrails/mask-batch` route - * over HTTP, and the route calls Presidio, whose spaCy NER refuses texts over - * 1,000,000 characters. + * over HTTP, and the route calls Presidio, whose spaCy NER refuses any single text + * over 1,000,000 characters. * * Checks, each recorded in the report: - * - Presidio answers an over-length NER text with a 422, not a 500, and still - * serves the same text when the request needs only pattern recognizers. + * - Presidio masks a text over spaCy's limit: a name straddling an NER window edge, + * a name at the very end, and a custom pattern that spans whitespace. * - A persisted log holding a 5M-character base64 PDF and a 3M-character text that * names a person at the end: the PDF becomes the binary marker without reaching * Presidio, the name is masked by real NER, and no request fails. - * - The same content as live block output: the PDF passes through byte-identical - * and the name is masked. + * - The same content as live block output: nothing is exempt from masking, the + * run is not aborted, and the name is masked. * * Exits non-zero when any check fails, after writing the report. */ const logger = createLogger('PiiRedactionE2E') -function requiredEnvironment(name: string): string { - const value = process.env[name] - assert(value, `${name} must be explicitly provided`) - return value -} +const reportPath = process.env.PII_E2E_REPORT_PATH +if (!reportPath) throw new Error('PII_E2E_REPORT_PATH must be explicitly provided') -const LOOPBACK_HOSTS = new Set(['localhost', '127.0.0.1', '[::1]']) -const baseUrl = new URL(requiredEnvironment('PII_E2E_BASE_URL')) -const presidioUrl = new URL(requiredEnvironment('PII_E2E_PRESIDIO_URL')) -const reportPath = requiredEnvironment('PII_E2E_REPORT_PATH') -assert(LOOPBACK_HOSTS.has(baseUrl.hostname), 'The app must use a loopback host') -assert(LOOPBACK_HOSTS.has(presidioUrl.hostname), 'Presidio must use a loopback host') -assert( - requiredEnvironment('INTERNAL_API_SECRET').length >= 32, - 'INTERNAL_API_SECRET must match the running app' -) -process.env.INTERNAL_API_BASE_URL = baseUrl.origin - -/** What a persisted log shows in place of an encoded file body. */ -const BINARY_MARKER = '[BINARY_CONTENT_REDACTED]' -const FAILED_MARKER = '[REDACTION_FAILED]' /** spaCy's default `max_length`, the per-text ceiling of one NER pass. */ const SPACY_MAX_LENGTH = 1_000_000 -/** The longest text one mask request may carry once texts are split. */ -const MAX_TEXT_CHARS = 1_000_000 +/** Where Presidio's first NER window ends (`NER_WINDOW_CHARS` in apps/pii/server.py). */ +const FIRST_NER_WINDOW_END = 100_000 -/** 3.75MB of bytes behind a PDF header: 5,000,000 characters of base64. */ -const pdfBase64 = Buffer.concat([Buffer.from('%PDF-1.7\n'), randomBytes(3_750_000 - 9)]).toString( - 'base64' -) const FILLER = 'The quarterly report covers revenue, costs, and the hiring plan.\n' const NAME_SENTENCE = 'Please send questions to Ada Lovelace before Friday.' -const longText = `${FILLER.repeat(Math.ceil(3_000_000 / FILLER.length))}${NAME_SENTENCE}` interface MaskRequest { status: number texts: number - longestText: number totalChars: number } /** Every mask-batch request this process sends, observed at `fetch`, unmodified. */ const maskRequests: MaskRequest[] = [] -const realFetch = globalThis.fetch -globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { - const response = await realFetch(input, init) - const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url - if (url.endsWith('/api/guardrails/mask-batch') && typeof init?.body === 'string') { - const { texts } = JSON.parse(init.body) as { texts: string[] } - maskRequests.push({ - status: response.status, - texts: texts.length, - longestText: Math.max(0, ...texts.map((text) => text.length)), - totalChars: texts.reduce((sum, text) => sum + text.length, 0), - }) - } - return response -}) as typeof fetch const checks: { name: string @@ -118,51 +78,99 @@ async function check( } } -function assertMaskRequestsSucceeded(since: number, mustNotCarry: number): void { +function assertMaskRequestsSucceeded(since: number): number { const requests = maskRequests.slice(since) assert(requests.length > 0, 'Expected mask-batch requests') for (const request of requests) { assert.equal(request.status, 200, `A mask-batch request answered ${request.status}`) - assert( - request.longestText <= MAX_TEXT_CHARS, - `A mask-batch request carried a ${request.longestText}-character text` - ) } - const sent = requests.reduce((sum, request) => sum + request.totalChars, 0) - assert(sent < mustNotCarry, `Requests carried ${sent} characters, so the encoded file was sent`) + return requests.reduce((sum, request) => sum + request.totalChars, 0) } -function assertNameMasked(masked: unknown): void { +function assertNameMasked(masked: unknown, failedMarker: string): void { assert.equal(typeof masked, 'string', 'Expected the long text back as a string') const text = masked as string - assert(!text.includes(FAILED_MARKER), 'The long text was scrubbed instead of masked') + assert(!text.includes(failedMarker), 'The long text was scrubbed instead of masked') assert(!text.includes('Ada Lovelace'), 'The name at the end of the long text was not masked') assert(text.endsWith('before Friday.'), 'The end of the long text was lost') assert(text.includes(''), 'Expected a token where the name was') } async function main(): Promise { - const { redactObjectStrings, redactPIIFromExecution } = await import( - '@/lib/logs/execution/pii-redaction' + const loopbackHosts = new Set(['localhost', '127.0.0.1', '[::1]']) + const baseUrl = new URL(process.env.PII_E2E_BASE_URL ?? '') + const presidioUrl = new URL(process.env.PII_E2E_PRESIDIO_URL ?? '') + assert(loopbackHosts.has(baseUrl.hostname), 'The app must use a loopback host') + assert(loopbackHosts.has(presidioUrl.hostname), 'Presidio must use a loopback host') + assert( + (process.env.INTERNAL_API_SECRET ?? '').length >= 32, + 'INTERNAL_API_SECRET must match the running app' ) - - await check('Presidio refuses an over-length NER text with a 422', async () => { - const overLength = 'a '.repeat(SPACY_MAX_LENGTH / 2 + 1) - const ner = await realFetch(new URL('/redact_batch', presidioUrl), { + process.env.INTERNAL_API_BASE_URL = baseUrl.origin + + const realFetch = globalThis.fetch + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const response = await realFetch(input, init) + const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url + if (url.endsWith('/api/guardrails/mask-batch') && typeof init?.body === 'string') { + const { texts } = JSON.parse(init.body) as { texts: string[] } + maskRequests.push({ + status: response.status, + texts: texts.length, + totalChars: texts.reduce((sum, text) => sum + text.length, 0), + }) + } + return response + }) as typeof fetch + + const { + BINARY_CONTENT_REDACTED_MARKER, + REDACTION_FAILED_MARKER, + redactObjectStrings, + redactPIIFromExecution, + } = await import('@/lib/logs/execution/pii-redaction') + + /** 3.75MB of fixed non-text bytes behind a PDF header: 5,000,000 characters of base64. */ + const pdfHeader = Buffer.from('%PDF-1.7\n') + const pdfBytes = Buffer.alloc(3_750_000) + pdfHeader.copy(pdfBytes) + for (let i = pdfHeader.length; i < pdfBytes.length; i++) pdfBytes[i] = (i * 7919 + 13) % 256 + const pdfBase64 = pdfBytes.toString('base64') + const longText = `${FILLER.repeat(Math.ceil(3_000_000 / FILLER.length))}${NAME_SENTENCE}` + + await check('Presidio masks a text longer than one NER pass allows', async () => { + const prose = (length: number) => + FILLER.repeat(Math.ceil(length / FILLER.length)).slice(0, length) + const lead = 'Please send questions to ' + const text = + prose(FIRST_NER_WINDOW_END - lead.length - 6) + + `${lead}Grace Hopper before Friday.\n` + + prose(400_000) + + 'badge EMP 123456 issued\n' + + prose(SPACY_MAX_LENGTH) + + NAME_SENTENCE + const response = await realFetch(new URL('/redact_batch', presidioUrl), { method: 'POST', headers: { 'content-type': 'application/json' }, - body: JSON.stringify({ texts: [overLength], language: 'en', entities: ['PERSON'] }), + body: JSON.stringify({ + texts: [text], + language: 'en', + entities: ['PERSON'], + patterns: [{ regex: 'EMP\\s+\\d{6}', replacement: 'EMPLOYEE_ID', name: 'Employee ID' }], + }), }) - const detail = await ner.text() - assert.equal(ner.status, 422, `NER pass answered ${ner.status}: ${detail.slice(0, 200)}`) - - const patternsOnly = await realFetch(new URL('/redact_batch', presidioUrl), { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify({ texts: [overLength], language: 'en', entities: ['EMAIL_ADDRESS'] }), - }) - assert.equal(patternsOnly.status, 200, 'A pattern-only pass must not be length-limited') - return { nerStatus: ner.status, detail: detail.slice(0, 200), patternOnlyStatus: 200 } + const body = await response.text() + assert.equal( + response.status, + 200, + `Presidio answered ${response.status}: ${body.slice(0, 200)}` + ) + const [masked] = (JSON.parse(body) as { texts: string[] }).texts + assert(!masked.includes('Grace Hopper'), 'The name across the first NER window edge survived') + assert(!masked.includes('Ada Lovelace'), 'The name at the end survived') + assert(!masked.includes('EMP 123456'), 'The custom pattern spanning whitespace survived') + assert(masked.includes(''), 'Expected the custom replacement') + return { textChars: text.length } }) await check('A persisted log replaces the PDF and masks the long text', async () => { @@ -172,7 +180,7 @@ async function main(): Promise { traceSpans: [ { blockId: 'write-file', - input: { content: pdfBase64, name: 'datasheet.pdf' }, + input: { content: pdfBase64, name: 'report.pdf' }, output: { text: longText }, }, ], @@ -184,23 +192,31 @@ async function main(): Promise { input: { content: string } output: { text: string } }[] - assert.equal(span.input.content, BINARY_MARKER) - assert.equal((result.workflowInput as { file: string }).file, BINARY_MARKER) - assertNameMasked(span.output.text) - assertMaskRequestsSucceeded(since, pdfBase64.length) - return { maskedTail: span.output.text.slice(-60) } + assert.equal(span.input.content, BINARY_CONTENT_REDACTED_MARKER) + assert.equal((result.workflowInput as { file: string }).file, BINARY_CONTENT_REDACTED_MARKER) + assertNameMasked(span.output.text, REDACTION_FAILED_MARKER) + const sentChars = assertMaskRequestsSucceeded(since) + assert( + sentChars < pdfBase64.length, + `Requests carried ${sentChars} characters: the PDF was sent` + ) + return { sentChars, maskedTail: span.output.text.slice(-60) } }) - await check('Live block output keeps the PDF bytes and masks the long text', async () => { + await check('Live block output is masked in full without aborting the run', async () => { const since = maskRequests.length const result = await redactObjectStrings( { content: pdfBase64, text: longText }, { entityTypes: ['PERSON'], onFailure: 'throw' } ) - assert.equal(result.content, pdfBase64, 'The encoded file changed') - assertNameMasked(result.text) - assertMaskRequestsSucceeded(since, pdfBase64.length) - return { maskedTail: result.text.slice(-60) } + assertNameMasked(result.text, REDACTION_FAILED_MARKER) + const sentChars = assertMaskRequestsSucceeded(since) + assert(sentChars >= pdfBase64.length + longText.length, 'Live data skipped part of its masking') + return { + sentChars, + pdfUnchanged: result.content === pdfBase64, + maskedTail: result.text.slice(-60), + } }) } @@ -215,9 +231,6 @@ const report = { suite: 'pii-redaction-e2e', startedAt, finishedAt: new Date().toISOString(), - baseUrl: baseUrl.origin, - presidioUrl: presidioUrl.origin, - fixtures: { pdfBase64Chars: pdfBase64.length, longTextChars: longText.length }, passed: checks.length - failed.length, failed: failed.length, checks, From aefa16a368ea4d5a8fb1a33393eb115a48bb8fe1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 12 Nov 2026 21:26:30 -0800 Subject: [PATCH 3/4] fix(pii): keep context scoring for thresholded long texts and match data: case-insensitively - Long-text NER windows also run the pattern entities when a score_threshold is set, so context-boosted matches clear it as they do on short texts - The data: URI scheme is matched case-insensitively - e2e asserts each name is masked as one span and is registered as test:pii-redaction:e2e --- apps/pii/server.py | 28 ++++++++++++------- .../lib/logs/execution/pii-redaction.test.ts | 1 + apps/sim/lib/logs/execution/pii-redaction.ts | 2 +- apps/sim/package.json | 1 + apps/sim/scripts/test-pii-redaction-e2e.ts | 14 ++++++---- 5 files changed, 30 insertions(+), 16 deletions(-) diff --git a/apps/pii/server.py b/apps/pii/server.py index 81c91d4cb4e..27e9aa9c846 100644 --- a/apps/pii/server.py +++ b/apps/pii/server.py @@ -243,11 +243,18 @@ def _exceeds_nlp_length(text: str, language: str) -> bool: return limit is not None and len(text) > limit -def _ner_window_entities(entities: list[str] | None, language: str) -> list[str]: - """The requested entities only spaCy NER produces and this language serves, or - all of them when `entities` is empty (detect-all).""" - ner = NER_ENTITIES & set(analyzer.get_supported_entities(language)) - return sorted(ner if not entities else ner.intersection(entities)) +def _window_entities( + entities: list[str] | None, language: str, score_threshold: float | None +) -> list[str]: + """The requested built-in entities this language serves that windows must find, + or all of them when `entities` is empty (detect-all). Without a score_threshold + that is only what spaCy NER produces; with one it also includes the pattern + entities, whose context-boosted score can decide whether a match clears the + threshold and needs the real NLP artifacts only a window has.""" + served = set(analyzer.get_supported_entities(language)) + if score_threshold is None or score_threshold <= 0: + served &= NER_ENTITIES + return sorted(served if not entities else served.intersection(entities)) def _analyze_long( @@ -263,13 +270,14 @@ def _analyze_long( Pattern, checksum, and custom recognizers have no length limit, so they run once over the whole text (against blank NLP artifacts, as the regex-only path does) and no match can straddle a cut. Only NER runs in overlapping windows - (see NER_WINDOW_CHARS), asked for just its own entities. A span touching a + (see NER_WINDOW_CHARS), asked for just its own entities, plus the pattern + entities when a score_threshold makes context scoring matter (see + `_window_entities`). A span touching a window's far edge may be cut short, so it is left to the next window, which sees it whole; a span ending inside the overlap before that edge was already found by the previous window and is skipped. The rest are shifted to - whole-text offsets, and overlapping spans are left to the anonymizer, which - merges them. Context words cannot raise a pattern match's score here, which - matters only when a score_threshold is set.""" + whole-text offsets; a span found by both passes keeps its higher score, and + overlapping spans are left to the anonymizer, which merges them.""" entities = entities or None results = analyzer.analyze( text=text, @@ -280,7 +288,7 @@ def _analyze_long( nlp_artifacts=_BLANK_ARTIFACTS.get(language), ad_hoc_recognizers=ad_hoc_recognizers or None, ) - window_entities = _ner_window_entities(entities, language) + window_entities = _window_entities(entities, language, score_threshold) if not window_entities: return results step = NER_WINDOW_CHARS - NER_WINDOW_OVERLAP diff --git a/apps/sim/lib/logs/execution/pii-redaction.test.ts b/apps/sim/lib/logs/execution/pii-redaction.test.ts index 0ca7e19b5ae..4c314539b7e 100644 --- a/apps/sim/lib/logs/execution/pii-redaction.test.ts +++ b/apps/sim/lib/logs/execution/pii-redaction.test.ts @@ -170,6 +170,7 @@ describe('encoded binary content', () => { ['wrapped at 76 columns', wrapped(76, '\n')], ['wrapped at 64 columns with CRLF', `${wrapped(64, '\r\n')}\r\n`], ['as a wrapped data URL', `data:application/pdf;base64,${wrapped(76, '\n')}`], + ['as a data URL with an uppercase scheme', `DATA:application/pdf;base64,${encodedFile}`], [ 'as a data URL with long metadata', `data:application/pdf;name=${'quarterly-report-'.repeat(40)}.pdf;base64,${encodedFile}`, diff --git a/apps/sim/lib/logs/execution/pii-redaction.ts b/apps/sim/lib/logs/execution/pii-redaction.ts index 433458910fd..c72e6790590 100644 --- a/apps/sim/lib/logs/execution/pii-redaction.ts +++ b/apps/sim/lib/logs/execution/pii-redaction.ts @@ -60,7 +60,7 @@ function base64Body(value: string): string | null { * masking cannot check such a value for PII, which lives inside the decoded file. */ function isEncodedBinary(value: string): boolean { - if (value.startsWith('data:')) { + if (/^data:/i.test(value)) { const comma = value.indexOf(',') return ( comma !== -1 && diff --git a/apps/sim/package.json b/apps/sim/package.json index 3f69e82872e..09cebb10844 100644 --- a/apps/sim/package.json +++ b/apps/sim/package.json @@ -25,6 +25,7 @@ "test:scim:e2e": "bun run scripts/test-scim-e2e.ts", "test:sim-links:e2e": "bun run scripts/test-sim-links-e2e.ts", "test:people:e2e": "bun run scripts/test-people-e2e.ts", + "test:pii-redaction:e2e": "bun --no-env-file scripts/test-pii-redaction-e2e.ts", "test:browser-extension:e2e": "bun --no-env-file scripts/test-browser-extension-e2e.ts", "test:isolated-vm-remote:e2e": "bun run scripts/test-isolated-vm-remote-e2e.ts", "test:mobile:e2e": "bun --no-env-file scripts/test-mobile-e2e.ts", diff --git a/apps/sim/scripts/test-pii-redaction-e2e.ts b/apps/sim/scripts/test-pii-redaction-e2e.ts index 97956436fb1..dd7cf8c6efa 100644 --- a/apps/sim/scripts/test-pii-redaction-e2e.ts +++ b/apps/sim/scripts/test-pii-redaction-e2e.ts @@ -91,9 +91,10 @@ function assertNameMasked(masked: unknown, failedMarker: string): void { assert.equal(typeof masked, 'string', 'Expected the long text back as a string') const text = masked as string assert(!text.includes(failedMarker), 'The long text was scrubbed instead of masked') - assert(!text.includes('Ada Lovelace'), 'The name at the end of the long text was not masked') - assert(text.endsWith('before Friday.'), 'The end of the long text was lost') - assert(text.includes(''), 'Expected a token where the name was') + assert( + text.endsWith('Please send questions to before Friday.'), + 'The name at the end of the long text was not masked as one span' + ) } async function main(): Promise { @@ -166,8 +167,11 @@ async function main(): Promise { `Presidio answered ${response.status}: ${body.slice(0, 200)}` ) const [masked] = (JSON.parse(body) as { texts: string[] }).texts - assert(!masked.includes('Grace Hopper'), 'The name across the first NER window edge survived') - assert(!masked.includes('Ada Lovelace'), 'The name at the end survived') + assert( + masked.includes(`${lead} before Friday.`), + 'The name across the first NER window edge was not masked as one span' + ) + assert(masked.endsWith('to before Friday.'), 'The name at the end survived') assert(!masked.includes('EMP 123456'), 'The custom pattern spanning whitespace survived') assert(masked.includes(''), 'Expected the custom replacement') return { textChars: text.length } From 3ada462100c79c4e34afcef77c306b8a0de0b348 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 12 Nov 2026 21:33:42 -0800 Subject: [PATCH 4/4] fix(pii): keep every NER window's spans instead of assuming a neighbour found them NER reads context, so a window can find a name the previous one missed or cut short; spans are now only deduplicated by exact position, and overlapping spans are left to the anonymizer to merge --- apps/pii/server.py | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-) diff --git a/apps/pii/server.py b/apps/pii/server.py index 27e9aa9c846..405d96e13bc 100644 --- a/apps/pii/server.py +++ b/apps/pii/server.py @@ -272,12 +272,10 @@ def _analyze_long( does) and no match can straddle a cut. Only NER runs in overlapping windows (see NER_WINDOW_CHARS), asked for just its own entities, plus the pattern entities when a score_threshold makes context scoring matter (see - `_window_entities`). A span touching a - window's far edge may be cut short, so it is left to the next window, which - sees it whole; a span ending inside the overlap before that edge was already - found by the previous window and is skipped. The rest are shifted to - whole-text offsets; a span found by both passes keeps its higher score, and - overlapping spans are left to the anonymizer, which merges them.""" + `_window_entities`). Every window's spans are kept, shifted to whole-text + offsets: NER reads context, so a window can find a name its neighbour missed + or cut short. A span found more than once keeps its highest score, and + overlapping spans of one type are merged by the anonymizer.""" entities = entities or None results = analyzer.analyze( text=text, @@ -293,18 +291,13 @@ def _analyze_long( return results step = NER_WINDOW_CHARS - NER_WINDOW_OVERLAP for start in range(0, len(text), step): - window = text[start : start + NER_WINDOW_CHARS] - is_last = start + NER_WINDOW_CHARS >= len(text) - seen_through = NER_WINDOW_OVERLAP if start else 0 for result in analyzer.analyze( - text=window, + text=text[start : start + NER_WINDOW_CHARS], language=language, entities=window_entities, score_threshold=score_threshold, return_decision_process=return_decision_process, ): - if result.end < seen_through or (not is_last and result.end >= len(window)): - continue results.append( RecognizerResult( result.entity_type, @@ -315,7 +308,7 @@ def _analyze_long( result.recognition_metadata, ) ) - if is_last: + if start + NER_WINDOW_CHARS >= len(text): break unique: dict[tuple[str, int, int], RecognizerResult] = {} for result in results: