Skip to content

Commit ee18853

Browse files
committed
fix(pii): keep encoded files away from Presidio and split texts too long for one NER pass
1 parent 5b4ce4a commit ee18853

10 files changed

Lines changed: 743 additions & 70 deletions

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,8 @@ 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.
122+
121123
## Self-hosted setup
122124

123125
### Enable cleanup and schedule it

‎apps/pii/server.py‎

Lines changed: 34 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -184,6 +184,7 @@ def build_analyzer() -> AnalyzerEngine:
184184
anonymizer: AnonymizerEngine
185185
NER_ENTITIES: frozenset
186186
_BLANK_ARTIFACTS: dict
187+
_NLP_MAX_LENGTH: dict
187188

188189
# Every entity the spaCy NER recognizers can actually produce. A request touching
189190
# 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
227228
)
228229

229230

231+
def _require_nlp_length(
232+
texts: list[str],
233+
language: str,
234+
entities: list[str] | None,
235+
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.",
254+
)
255+
256+
230257
def _analyze_one(
231258
text: str,
232259
language: str,
@@ -235,6 +262,7 @@ def _analyze_one(
235262
return_decision_process: bool = False,
236263
ad_hoc_recognizers: list[PatternRecognizer] | None = None,
237264
):
265+
_require_nlp_length([text], language, entities, score_threshold)
238266
# Regex-only requests reuse a blank NlpArtifacts to skip the spaCy NLP pass;
239267
# otherwise analyze() computes artifacts (runs spaCy) as usual. Custom-pattern
240268
# recognizers are regex-based, so they run fine against the blank artifacts.
@@ -265,6 +293,7 @@ def _analyze_many(
265293
ad_hoc_recognizers: list[PatternRecognizer] | None = None,
266294
):
267295
"""Analyze many texts, skipping the spaCy pass for regex-only requests."""
296+
_require_nlp_length(texts, language, entities, score_threshold)
268297
if _regex_only(entities, score_threshold):
269298
blank = _BLANK_ARTIFACTS.get(language)
270299
return [
@@ -292,7 +321,7 @@ def _analyze_many(
292321

293322
@asynccontextmanager
294323
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
295-
global analyzer, batch_analyzer, anonymizer, NER_ENTITIES, _BLANK_ARTIFACTS
324+
global analyzer, batch_analyzer, anonymizer, NER_ENTITIES, _BLANK_ARTIFACTS, _NLP_MAX_LENGTH
296325
logger.info("building analyzer (spacy)")
297326
analyzer = build_analyzer()
298327
batch_analyzer = BatchAnalyzerEngine(analyzer_engine=analyzer)
@@ -319,6 +348,10 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
319348
language: analyzer.nlp_engine.process_text("", language)
320349
for language in SUPPORTED_LANGUAGES
321350
}
351+
_NLP_MAX_LENGTH = {
352+
language: nlp.max_length
353+
for language, nlp in getattr(analyzer.nlp_engine, "nlp", {}).items()
354+
}
322355
yield
323356

324357

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

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
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'
45
import { POST } from '@/app/api/guardrails/mask-batch/route'
56

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

5253
expect(res.status).toBe(status)
5354
})
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+
})
5471
})

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

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ 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'
1920
import { POST as maskBatchRoute } from '@/app/api/guardrails/mask-batch/route'
2021

2122
const mockSleep = utilsHelpersMockFns.mockSleep
@@ -73,6 +74,87 @@ describe('maskPIIBatchViaHttp', () => {
7374
expect(mockSleep).toHaveBeenCalledTimes(7)
7475
})
7576

77+
it('stops after three attempts on a persistent 500, which can be a deterministic input failure', async () => {
78+
fetchMock.mockImplementation(async () => new Response('failed', { status: 500 }))
79+
80+
await expect(maskPIIBatchViaHttp(['a'], [])).rejects.toThrow(/mask-batch request failed/)
81+
expect(fetchMock).toHaveBeenCalledTimes(3)
82+
expect(mockSleep).toHaveBeenCalledTimes(2)
83+
})
84+
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+
76158
describe('with a failed-chunk placeholder', () => {
77159
const PLACEHOLDER = '[FAILED]'
78160
const CHUNK_SIZE = 2000

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

Lines changed: 41 additions & 17 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 } from '@/lib/guardrails/pii-batching'
11+
import { chunkIndicesByBudget, joinSegments, segmentTexts } from '@/lib/guardrails/pii-batching'
1212
import type { CustomPiiPattern } from '@/lib/guardrails/pii-entities'
1313

1414
const logger = createLogger('PiiMaskClient')
@@ -36,6 +36,15 @@ const CHUNK_CONCURRENCY = env.PII_MASK_CHUNK_CONCURRENCY ?? 64
3636
*/
3737
const MAX_CHUNK_ATTEMPTS = 8
3838

39+
/**
40+
* Attempts for a chunk the route answers with a 500. The route keeps 503 for a
41+
* Presidio outage and 422 for input Presidio refused, so a 500 is an unclassified
42+
* failure that may be deterministic for this input: two retries cover a one-off
43+
* crash without spending the full outage budget resending input that fails the
44+
* same way every time.
45+
*/
46+
const MAX_ATTEMPTS_AFTER_SERVER_ERROR = 3
47+
3948
const RETRYABLE_STATUSES = new Set([408, 429, 500, 502, 503, 504])
4049

4150
class MaskChunkHttpError extends Error {
@@ -97,9 +106,9 @@ function isOutageError(error: unknown): boolean {
97106

98107
interface MaskPIIBatchOptions {
99108
/**
100-
* When set, a chunk that cannot be masked has each of its strings replaced with
101-
* this value instead of rejecting the call, so one bad chunk costs only its own
102-
* strings. After an outage (see {@link OUTAGE_STATUSES}) exhausts a chunk's
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
103112
* retries, chunks not yet sent are replaced the same way.
104113
*/
105114
failedChunkPlaceholder?: string
@@ -118,7 +127,9 @@ interface MaskPIIBatchResult {
118127
* Only the app task reaches the Presidio service (it holds `PII_URL`), but the
119128
* log-redaction persist path also runs inside the trigger.dev runtime — so
120129
* redaction always routes through HTTP, the same way the guardrails tool does.
121-
* Strings are grouped into byte/count-budgeted chunks (keeping each request far
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
122133
* under the 10MB Next body limit) and the chunks are sent with bounded
123134
* concurrency, so a large payload fans out rather than serializing; order is
124135
* preserved, so the returned `masked` array matches `texts` length.
@@ -141,38 +152,47 @@ export async function maskPIIBatchViaHttp(
141152

142153
const url = `${getInternalApiBaseUrl()}/api/guardrails/mask-batch`
143154
const placeholder = options.failedChunkPlaceholder
144-
const masked = new Array<string>(texts.length)
155+
const { segments, owners } = segmentTexts(texts)
156+
const maskedSegments = new Array<string>(segments.length)
157+
const scrubbedTexts = new Set<number>()
145158
let serviceUnavailable = false
146-
let scrubbedCount = 0
147159
let firstFailure: unknown
148160

149-
const scrubChunk = (indices: number[], value: string) => {
150-
for (const i of indices) masked[i] = value
151-
scrubbedCount += indices.length
161+
const scrubChunk = (indices: number[]) => {
162+
for (const k of indices) scrubbedTexts.add(owners[k])
152163
}
153164

154-
await mapWithConcurrency(chunkIndicesByBudget(texts), CHUNK_CONCURRENCY, async (indices) => {
165+
await mapWithConcurrency(chunkIndicesByBudget(segments), CHUNK_CONCURRENCY, async (indices) => {
155166
if (serviceUnavailable && placeholder !== undefined) {
156-
scrubChunk(indices, placeholder)
167+
scrubChunk(indices)
157168
return
158169
}
159170
try {
160-
const chunk = indices.map((i) => texts[i])
171+
const chunk = indices.map((k) => segments[k])
161172
const out = await postChunk(url, chunk, entityTypes, language, customPatterns)
162173
if (out.length !== chunk.length) {
163174
throw new Error('PII mask-batch returned an unexpected result')
164175
}
165-
indices.forEach((originalIndex, k) => {
166-
masked[originalIndex] = out[k]
176+
indices.forEach((segmentIndex, k) => {
177+
maskedSegments[segmentIndex] = out[k]
167178
})
168179
} catch (error) {
169180
if (placeholder === undefined) throw error
170181
if (isOutageError(error)) serviceUnavailable = true
171182
firstFailure ??= error
172-
scrubChunk(indices, placeholder)
183+
scrubChunk(indices)
173184
}
174185
})
175186

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+
176196
if (scrubbedCount > 0) {
177197
logger.error('PII masking failed for some chunks; replaced their strings', {
178198
error: getErrorMessage(firstFailure),
@@ -196,7 +216,11 @@ async function postChunk(
196216
try {
197217
return await postChunkOnce(url, texts, entityTypes, language, customPatterns)
198218
} catch (error) {
199-
if (attempt >= MAX_CHUNK_ATTEMPTS || !isRetryableChunkError(error)) {
219+
const maxAttempts =
220+
error instanceof MaskChunkHttpError && error.status === 500
221+
? MAX_ATTEMPTS_AFTER_SERVER_ERROR
222+
: MAX_CHUNK_ATTEMPTS
223+
if (attempt >= maxAttempts || !isRetryableChunkError(error)) {
200224
throw error
201225
}
202226
const retryAfterMs = error instanceof MaskChunkHttpError ? error.retryAfterMs : null

0 commit comments

Comments
 (0)