Skip to content

Commit 5ff04d2

Browse files
committed
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
1 parent ee18853 commit 5ff04d2

12 files changed

Lines changed: 387 additions & 461 deletions

File tree

‎apps/docs/content/docs/platform/enterprise/data-retention.mdx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,7 @@ Redaction applies to text strings at the configured workflow stages. It does not
118118

119119
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.
120120

121-
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.
121+
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.
122122

123123
## Self-hosted setup
124124

‎apps/pii/server.py‎

Lines changed: 109 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -228,30 +228,93 @@ def _regex_only(entities: list[str] | None, score_threshold: float | None) -> bo
228228
)
229229

230230

231-
def _require_nlp_length(
232-
texts: list[str],
231+
# spaCy refuses (error E088) any text longer than its model's `max_length`, so a
232+
# longer text runs NER in windows of this many characters, following spaCy's
233+
# guidance to segment long documents (it sizes NER memory at roughly 1GB per
234+
# 100,000 characters of one text).
235+
NER_WINDOW_CHARS = 100_000
236+
# How much of the previous window each window repeats. Far longer than any entity
237+
# spaCy tags, so an entity cut by one window's edge lies whole inside the next.
238+
NER_WINDOW_OVERLAP = 1_000
239+
240+
241+
def _exceeds_nlp_length(text: str, language: str) -> bool:
242+
limit = _NLP_MAX_LENGTH.get(language)
243+
return limit is not None and len(text) > limit
244+
245+
246+
def _ner_window_entities(entities: list[str] | None, language: str) -> list[str]:
247+
"""The requested entities only spaCy NER produces and this language serves, or
248+
all of them when `entities` is empty (detect-all)."""
249+
ner = NER_ENTITIES & set(analyzer.get_supported_entities(language))
250+
return sorted(ner if not entities else ner.intersection(entities))
251+
252+
253+
def _analyze_long(
254+
text: str,
233255
language: str,
234256
entities: list[str] | None,
235257
score_threshold: float | None,
236-
) -> None:
237-
"""Refuse, as a 422, any text longer than the language model's spaCy
238-
`max_length` when the request runs the NLP pass. spaCy raises ValueError E088
239-
for such a text, which FastAPI answers as an opaque 500 that callers retry as
240-
a transient failure; the same text fails the same way every time, so callers
241-
must split it rather than resend it. Regex-only requests skip spaCy and are
242-
not limited."""
243-
if _regex_only(entities, score_threshold):
244-
return
245-
limit = _NLP_MAX_LENGTH.get(language)
246-
if limit is None:
247-
return
248-
for index, text in enumerate(texts):
249-
if len(text) > limit:
250-
raise HTTPException(
251-
status_code=422,
252-
detail=f"Text {index} has {len(text)} characters; NER accepts at most "
253-
f"{limit} per text. Split it before sending.",
258+
return_decision_process: bool = False,
259+
ad_hoc_recognizers: list[PatternRecognizer] | None = None,
260+
) -> list[RecognizerResult]:
261+
"""Analyze a text longer than spaCy's `max_length`.
262+
263+
Pattern, checksum, and custom recognizers have no length limit, so they run
264+
once over the whole text (against blank NLP artifacts, as the regex-only path
265+
does) and no match can straddle a cut. Only NER runs in overlapping windows
266+
(see NER_WINDOW_CHARS), asked for just its own entities. A span touching a
267+
window's far edge may be cut short, so it is left to the next window, which
268+
sees it whole; a span ending inside the overlap before that edge was already
269+
found by the previous window and is skipped. The rest are shifted to
270+
whole-text offsets, and overlapping spans are left to the anonymizer, which
271+
merges them. Context words cannot raise a pattern match's score here, which
272+
matters only when a score_threshold is set."""
273+
entities = entities or None
274+
results = analyzer.analyze(
275+
text=text,
276+
language=language,
277+
entities=entities,
278+
score_threshold=score_threshold,
279+
return_decision_process=return_decision_process,
280+
nlp_artifacts=_BLANK_ARTIFACTS.get(language),
281+
ad_hoc_recognizers=ad_hoc_recognizers or None,
282+
)
283+
window_entities = _ner_window_entities(entities, language)
284+
if not window_entities:
285+
return results
286+
step = NER_WINDOW_CHARS - NER_WINDOW_OVERLAP
287+
for start in range(0, len(text), step):
288+
window = text[start : start + NER_WINDOW_CHARS]
289+
is_last = start + NER_WINDOW_CHARS >= len(text)
290+
seen_through = NER_WINDOW_OVERLAP if start else 0
291+
for result in analyzer.analyze(
292+
text=window,
293+
language=language,
294+
entities=window_entities,
295+
score_threshold=score_threshold,
296+
return_decision_process=return_decision_process,
297+
):
298+
if result.end < seen_through or (not is_last and result.end >= len(window)):
299+
continue
300+
results.append(
301+
RecognizerResult(
302+
result.entity_type,
303+
result.start + start,
304+
result.end + start,
305+
result.score,
306+
result.analysis_explanation,
307+
result.recognition_metadata,
308+
)
254309
)
310+
if is_last:
311+
break
312+
unique: dict[tuple[str, int, int], RecognizerResult] = {}
313+
for result in results:
314+
key = (result.entity_type, result.start, result.end)
315+
if key not in unique or result.score > unique[key].score:
316+
unique[key] = result
317+
return list(unique.values())
255318

256319

257320
def _analyze_one(
@@ -262,13 +325,15 @@ def _analyze_one(
262325
return_decision_process: bool = False,
263326
ad_hoc_recognizers: list[PatternRecognizer] | None = None,
264327
):
265-
_require_nlp_length([text], language, entities, score_threshold)
328+
regex_only = _regex_only(entities, score_threshold)
329+
if not regex_only and _exceeds_nlp_length(text, language):
330+
return _analyze_long(
331+
text, language, entities, score_threshold, return_decision_process, ad_hoc_recognizers
332+
)
266333
# Regex-only requests reuse a blank NlpArtifacts to skip the spaCy NLP pass;
267334
# otherwise analyze() computes artifacts (runs spaCy) as usual. Custom-pattern
268335
# recognizers are regex-based, so they run fine against the blank artifacts.
269-
nlp_artifacts = (
270-
_BLANK_ARTIFACTS.get(language) if _regex_only(entities, score_threshold) else None
271-
)
336+
nlp_artifacts = _BLANK_ARTIFACTS.get(language) if regex_only else None
272337
return analyzer.analyze(
273338
text=text,
274339
language=language,
@@ -292,8 +357,9 @@ def _analyze_many(
292357
score_threshold: float | None,
293358
ad_hoc_recognizers: list[PatternRecognizer] | None = None,
294359
):
295-
"""Analyze many texts, skipping the spaCy pass for regex-only requests."""
296-
_require_nlp_length(texts, language, entities, score_threshold)
360+
"""Analyze many texts, skipping the spaCy pass for regex-only requests and
361+
analyzing any text longer than spaCy's `max_length` on its own (see
362+
`_analyze_long`)."""
297363
if _regex_only(entities, score_threshold):
298364
blank = _BLANK_ARTIFACTS.get(language)
299365
return [
@@ -307,16 +373,24 @@ def _analyze_many(
307373
)
308374
for text in texts
309375
]
310-
return list(
311-
batch_analyzer.analyze_iterator(
312-
texts=texts,
313-
language=language,
314-
batch_size=NLP_BATCH_SIZE,
315-
entities=entities or None,
316-
score_threshold=score_threshold,
317-
ad_hoc_recognizers=ad_hoc_recognizers or None,
318-
)
376+
long_indices = {i for i, text in enumerate(texts) if _exceeds_nlp_length(text, language)}
377+
short_indices = [i for i in range(len(texts)) if i not in long_indices]
378+
results: list[list[RecognizerResult]] = [[] for _ in texts]
379+
short_results = batch_analyzer.analyze_iterator(
380+
texts=[texts[i] for i in short_indices],
381+
language=language,
382+
batch_size=NLP_BATCH_SIZE,
383+
entities=entities or None,
384+
score_threshold=score_threshold,
385+
ad_hoc_recognizers=ad_hoc_recognizers or None,
319386
)
387+
for i, per_text in zip(short_indices, short_results):
388+
results[i] = list(per_text)
389+
for i in sorted(long_indices):
390+
results[i] = _analyze_long(
391+
texts[i], language, entities, score_threshold, ad_hoc_recognizers=ad_hoc_recognizers
392+
)
393+
return results
320394

321395

322396
@asynccontextmanager

‎apps/sim/app/api/guardrails/mask-batch/route.test.ts‎

Lines changed: 0 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import { hybridAuthMockFns } from '@sim/testing/mocks/hybrid-auth.mock'
22
import { createMockRequest } from '@sim/testing/mocks/request.mock'
33
import { beforeEach, describe, expect, it, vi } from 'vitest'
4-
import { PII_MAX_TEXT_CHARS } from '@/lib/guardrails/pii-batching'
54
import { POST } from '@/app/api/guardrails/mask-batch/route'
65

76
const mockCheckInternalAuth = hybridAuthMockFns.mockCheckInternalAuth
@@ -52,20 +51,4 @@ describe('POST /api/guardrails/mask-batch', () => {
5251

5352
expect(res.status).toBe(status)
5453
})
55-
56-
it('splits a text too long for one NER pass before calling Presidio, then rejoins it', async () => {
57-
presidioFetch.mockImplementation(async (_url: string, init: { body: string }) => {
58-
const { texts } = JSON.parse(init.body) as { texts: string[] }
59-
return Response.json({ texts: texts.map((t) => t.replaceAll('Ada', '<PERSON>')) })
60-
})
61-
const text = 'Ada wrote this line.\n'.repeat(60_000)
62-
63-
const res = await POST(createMockRequest('POST', { texts: [text], entityTypes: ['PERSON'] }))
64-
65-
expect(res.status).toBe(200)
66-
expect(await res.json()).toEqual({ masked: [text.replaceAll('Ada', '<PERSON>')] })
67-
const sent = presidioFetch.mock.calls.flatMap(([, init]) => JSON.parse(init.body).texts)
68-
expect(sent.length).toBeGreaterThan(1)
69-
for (const piece of sent) expect(piece.length).toBeLessThanOrEqual(PII_MAX_TEXT_CHARS)
70-
})
7154
})

‎apps/sim/lib/guardrails/mask-client.test.ts‎

Lines changed: 0 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@ vi.mock('@/lib/auth/internal', () => authInternalMock)
1616
vi.mock('@sim/utils/helpers', () => utilsHelpersMock)
1717

1818
import { maskPIIBatchViaHttp } from '@/lib/guardrails/mask-client'
19-
import { PII_MAX_TEXT_CHARS } from '@/lib/guardrails/pii-batching'
2019
import { POST as maskBatchRoute } from '@/app/api/guardrails/mask-batch/route'
2120

2221
const mockSleep = utilsHelpersMockFns.mockSleep
@@ -82,79 +81,6 @@ describe('maskPIIBatchViaHttp', () => {
8281
expect(mockSleep).toHaveBeenCalledTimes(2)
8382
})
8483

85-
describe('with a text longer than one Presidio pass allows', () => {
86-
/** Each request masks `Ada` to `<PERSON>`, so pieces rejoin to the masked whole. */
87-
const maskingAda = () =>
88-
fetchMock.mockImplementation(async (_url: string, init: { body: string }) => {
89-
const { texts } = JSON.parse(init.body) as { texts: string[] }
90-
return Response.json({ masked: texts.map((t) => t.replaceAll('Ada', '<PERSON>')) })
91-
})
92-
const sentPieces = () =>
93-
fetchMock.mock.calls.flatMap(([, init]) => JSON.parse(init.body).texts as string[])
94-
95-
it('sends newline-aligned pieces within the limit and rejoins the masked result', async () => {
96-
maskingAda()
97-
const text = Array.from({ length: 60_000 }, (_, i) => `row ${i} owner Ada\n`).join('')
98-
99-
const { masked } = await maskPIIBatchViaHttp([text, 'short Ada'], ['PERSON'])
100-
101-
expect(masked).toEqual([text.replaceAll('Ada', '<PERSON>'), 'short <PERSON>'])
102-
const pieces = sentPieces().filter((p) => p !== 'short Ada')
103-
expect(pieces.length).toBeGreaterThan(1)
104-
expect(pieces.join('')).toBe(text)
105-
for (const piece of pieces) expect(piece.length).toBeLessThanOrEqual(PII_MAX_TEXT_CHARS)
106-
for (const piece of pieces.slice(0, -1)) expect(piece.endsWith('\n')).toBe(true)
107-
})
108-
109-
it('cuts minified JSON at punctuation so no email is split across pieces', async () => {
110-
maskingAda()
111-
const record = '{"email":"ada.lovelace@example.com","id":1234}'
112-
const text = `[${Array.from({ length: 25_000 }, () => record).join(',')}]`
113-
114-
await maskPIIBatchViaHttp([text], ['EMAIL_ADDRESS'])
115-
116-
const pieces = sentPieces()
117-
expect(pieces.length).toBeGreaterThan(1)
118-
const emailsSent = pieces.reduce(
119-
(count, piece) => count + piece.split('ada.lovelace@example.com').length - 1,
120-
0
121-
)
122-
expect(emailsSent).toBe(25_000)
123-
})
124-
125-
it('never splits a surrogate pair when the text has nothing to cut at', async () => {
126-
// One leading unit puts a high surrogate exactly where a hard cut would land.
127-
const text = `a${'\u{1F600}'.repeat(PII_MAX_TEXT_CHARS)}`
128-
129-
const { masked } = await maskPIIBatchViaHttp([text], [])
130-
131-
const pieces = sentPieces()
132-
expect(pieces.length).toBeGreaterThan(1)
133-
expect(pieces.join('')).toBe(text)
134-
expect(masked).toEqual([`M(${pieces.join(')M(')})`])
135-
})
136-
137-
it('replaces the whole text when any one of its pieces fails', async () => {
138-
const text = `${'safe line\n'.repeat(120_000)}poison`
139-
fetchMock.mockImplementation(async (_url: string, init: { body: string }) => {
140-
const { texts } = JSON.parse(init.body) as { texts: string[] }
141-
if (texts.some((t) => t.endsWith('poison'))) return new Response('no', { status: 422 })
142-
return Response.json({ masked: texts })
143-
})
144-
145-
const { masked, scrubbedCount } = await maskPIIBatchViaHttp(
146-
[text],
147-
[],
148-
undefined,
149-
undefined,
150-
{ failedChunkPlaceholder: '[FAILED]' }
151-
)
152-
153-
expect(masked).toEqual(['[FAILED]'])
154-
expect(scrubbedCount).toBe(1)
155-
})
156-
})
157-
15884
describe('with a failed-chunk placeholder', () => {
15985
const PLACEHOLDER = '[FAILED]'
16086
const CHUNK_SIZE = 2000

‎apps/sim/lib/guardrails/mask-client.ts‎

Lines changed: 16 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import { generateInternalToken } from '@/lib/auth/internal'
88
import { env } from '@/lib/core/config/env'
99
import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
1010
import { getInternalApiBaseUrl } from '@/lib/core/utils/urls'
11-
import { chunkIndicesByBudget, joinSegments, segmentTexts } from '@/lib/guardrails/pii-batching'
11+
import { chunkIndicesByBudget } from '@/lib/guardrails/pii-batching'
1212
import type { CustomPiiPattern } from '@/lib/guardrails/pii-entities'
1313

1414
const logger = createLogger('PiiMaskClient')
@@ -106,9 +106,9 @@ function isOutageError(error: unknown): boolean {
106106

107107
interface MaskPIIBatchOptions {
108108
/**
109-
* When set, a chunk that cannot be masked has every string with a piece in it
110-
* replaced with this value instead of rejecting the call, so one bad chunk costs
111-
* only its own strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's
109+
* When set, a chunk that cannot be masked has each of its strings replaced with
110+
* this value instead of rejecting the call, so one bad chunk costs only its own
111+
* strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's
112112
* retries, chunks not yet sent are replaced the same way.
113113
*/
114114
failedChunkPlaceholder?: string
@@ -127,9 +127,7 @@ interface MaskPIIBatchResult {
127127
* Only the app task reaches the Presidio service (it holds `PII_URL`), but the
128128
* log-redaction persist path also runs inside the trigger.dev runtime — so
129129
* redaction always routes through HTTP, the same way the guardrails tool does.
130-
* A string too long for one NER pass is cut into pieces first (see
131-
* {@link segmentTexts}), so no request carries a text Presidio would refuse.
132-
* Pieces are grouped into byte/count-budgeted chunks (keeping each request far
130+
* Strings are grouped into byte/count-budgeted chunks (keeping each request far
133131
* under the 10MB Next body limit) and the chunks are sent with bounded
134132
* concurrency, so a large payload fans out rather than serializing; order is
135133
* preserved, so the returned `masked` array matches `texts` length.
@@ -152,47 +150,38 @@ export async function maskPIIBatchViaHttp(
152150

153151
const url = `${getInternalApiBaseUrl()}/api/guardrails/mask-batch`
154152
const placeholder = options.failedChunkPlaceholder
155-
const { segments, owners } = segmentTexts(texts)
156-
const maskedSegments = new Array<string>(segments.length)
157-
const scrubbedTexts = new Set<number>()
153+
const masked = new Array<string>(texts.length)
158154
let serviceUnavailable = false
155+
let scrubbedCount = 0
159156
let firstFailure: unknown
160157

161-
const scrubChunk = (indices: number[]) => {
162-
for (const k of indices) scrubbedTexts.add(owners[k])
158+
const scrubChunk = (indices: number[], value: string) => {
159+
for (const i of indices) masked[i] = value
160+
scrubbedCount += indices.length
163161
}
164162

165-
await mapWithConcurrency(chunkIndicesByBudget(segments), CHUNK_CONCURRENCY, async (indices) => {
163+
await mapWithConcurrency(chunkIndicesByBudget(texts), CHUNK_CONCURRENCY, async (indices) => {
166164
if (serviceUnavailable && placeholder !== undefined) {
167-
scrubChunk(indices)
165+
scrubChunk(indices, placeholder)
168166
return
169167
}
170168
try {
171-
const chunk = indices.map((k) => segments[k])
169+
const chunk = indices.map((i) => texts[i])
172170
const out = await postChunk(url, chunk, entityTypes, language, customPatterns)
173171
if (out.length !== chunk.length) {
174172
throw new Error('PII mask-batch returned an unexpected result')
175173
}
176-
indices.forEach((segmentIndex, k) => {
177-
maskedSegments[segmentIndex] = out[k]
174+
indices.forEach((originalIndex, k) => {
175+
masked[originalIndex] = out[k]
178176
})
179177
} catch (error) {
180178
if (placeholder === undefined) throw error
181179
if (isOutageError(error)) serviceUnavailable = true
182180
firstFailure ??= error
183-
scrubChunk(indices)
181+
scrubChunk(indices, placeholder)
184182
}
185183
})
186184

187-
const joined = joinSegments(maskedSegments, owners, texts.length)
188-
// A text with any unmasked piece is replaced whole: its other pieces are masked,
189-
// but keeping them would persist a fragment around the hole.
190-
const masked =
191-
placeholder === undefined
192-
? joined
193-
: joined.map((text, i) => (scrubbedTexts.has(i) ? placeholder : text))
194-
const scrubbedCount = scrubbedTexts.size
195-
196185
if (scrubbedCount > 0) {
197186
logger.error('PII masking failed for some chunks; replaced their strings', {
198187
error: getErrorMessage(firstFailure),

0 commit comments

Comments
 (0)