Coverage for astrocyte/policy/barriers.py: 97%

1"""Barrier policies — PII scanning, content validation, metadata sanitization.

3All functions are sync (Rust migration candidates) except scan_async/apply_async.

4See docs/_design/policy-layer.md section 2 and docs/_design/data-governance.md.

5"""

7from __future__ import annotations

9import json

10import logging

11import re

12from typing import TYPE_CHECKING

14from astrocyte.errors import PiiRejected

15from astrocyte.types import Metadata, PiiMatch

17if TYPE_CHECKING:

18 from astrocyte.policy.llm_scanner import LlmPiiScanner

19 from astrocyte.policy.ner_scanner import NerPiiScanner

21logger = logging.getLogger("astrocyte.pii")

23# ---------------------------------------------------------------------------

24# PII regex patterns — global

25# ---------------------------------------------------------------------------

27_PII_PATTERNS: dict[str, tuple[re.Pattern[str], str]] = {

28 "email": (

29 re.compile(

30 r"\b[a-zA-Z0-9](?:[a-zA-Z0-9._%+\-]*[a-zA-Z0-9])?@[a-zA-Z0-9](?:[a-zA-Z0-9\-]*[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9\-]*[a-zA-Z0-9])?)*\.[a-zA-Z]{2,}\b"

31 ),

32 "[EMAIL_REDACTED]",

33 ),

34 "phone": (

35 re.compile(r"(?:\+?1[-.\s]?)?$?\d{3}$?[-.\s]?\d{3}[-.\s]?\d{4}"),

36 "[PHONE_REDACTED]",

37 ),

38 "ssn": (

39 re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),

40 "[SSN_REDACTED]",

41 ),

42 "credit_card": (

43 re.compile(r"\b(?:\d{4}[-\s]?){3}\d{4}\b"),

44 "[CC_REDACTED]",

45 ),

46 "ip_address": (

47 re.compile(

48 r"\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b"

49 r"|(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}"

50 ),

51 "[IP_REDACTED]",

52 ),

53 # Global: date of birth (with context words)

54 "date_of_birth": (

55 re.compile(

56 r"(?i)(?:born|dob|date\s+of\s+birth|birthday)[:\s]+(\d{4}[-/]\d{2}[-/]\d{2}|\d{1,2}[-/]\d{1,2}[-/]\d{2,4})"

57 ),

58 "[DOB_REDACTED]",

59 ),

60 # Global: IBAN

61 "iban": (

62 re.compile(r"\b[A-Z]{2}\d{2}\s?[A-Z0-9]{4}(?:\s?[A-Z0-9]{4}){2,7}(?:\s?[A-Z0-9]{1,4})?\b"),

63 "[IBAN_REDACTED]",

64 ),

65}

68# ---------------------------------------------------------------------------

69# Credentials — global

70#

71# Memory is replayed into future prompts, possibly for another agent and model

72# vendor than the one it was captured from: a key pasted into one session

73# would otherwise travel to every provider that later recalls it. Formats are

74# vendor prefixes or structural markers, chosen for precision — a false

75# positive costs a few characters of a memory, a miss leaks a credential.

76#

77# A pattern with a group named ``v`` redacts only that group, so

78# ``DB_PASSWORD=hunter2hunter2`` keeps its name and loses its value.

79# ---------------------------------------------------------------------------

81_SECRET = "[SECRET_REDACTED]"

83_SECRET_PATTERNS: dict[str, tuple[re.Pattern[str], str]] = {

84 "private_key": (

85 re.compile(

86 r"-----BEGIN (?:[A-Z0-9]+ )*PRIVATE KEY(?: BLOCK)?-----.*?(?:-----END (?:[A-Z0-9]+ )*PRIVATE KEY(?: BLOCK)?-----|\Z)",

87 re.DOTALL,

88 ),

89 _SECRET,

90 ),

91 # OpenAI (sk-, sk-proj-), Anthropic (sk-ant-), and other sk- style keys.

92 # A digit is required so hyphenated identifiers ("sk-learn-…") don't match.

93 "api_key": (re.compile(r"\bsk-(?=[A-Za-z0-9_\-]*\d)[A-Za-z0-9_\-]{20,}"), _SECRET),

94 "aws_access_key": (re.compile(r"\b(?:AKIA|ASIA|ABIA|ACCA)[0-9A-Z]{16}\b"), _SECRET),

95 "github_token": (re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9]{36,}|github_pat_[A-Za-z0-9_]{22,})\b"), _SECRET),

96 "gitlab_token": (re.compile(r"\bglpat-[A-Za-z0-9_\-]{20,}"), _SECRET),

97 "slack_token": (re.compile(r"\bxox[abposr]-[A-Za-z0-9\-]{10,}"), _SECRET),

98 "stripe_key": (re.compile(r"\b(?:sk|rk|pk)_(?:live|test)_[A-Za-z0-9]{16,}"), _SECRET),

99 "google_api_key": (re.compile(r"\bAIza[0-9A-Za-z_\-]{35}\b"), _SECRET),

100 "huggingface_token": (re.compile(r"\bhf_[A-Za-z0-9]{30,}\b"), _SECRET),

101 "npm_token": (re.compile(r"\bnpm_[A-Za-z0-9]{36}\b"), _SECRET),

102 "doppler_token": (re.compile(r"\bdp\.(?:st|pt|ct|sa|scim|audit)\.[A-Za-z0-9_\-.]{20,}"), _SECRET),

103 "jwt": (re.compile(r"\beyJ[A-Za-z0-9_\-]{8,}\.eyJ[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}"), _SECRET),

104 # scheme://user:PASSWORD@host

105 "url_password": (re.compile(r"\b[a-zA-Z][a-zA-Z0-9+.\-]*://[^\s:/@]+:(?P<v>[^\s@/]{3,})@"), _SECRET),

106 # NAME_KEY=value / "password": "value" — the value must look like a

107 # secret (12+ token characters mixing letters and digits), not a

108 # variable reference, a placeholder, a count or prose.

109 "credential_assignment": (

110 re.compile(

112 r"[A-Za-z0-9_]*[\"']?\s*[:=]\s*[\"']?"

113 r"(?P<v>(?=[A-Za-z0-9_\-+/=.]*\d)(?=[A-Za-z0-9_\-+/=.]*[A-Za-z])[A-Za-z0-9_\-+/=.]{12,})"

114 ),

115 _SECRET,

116 ),

117}

118

119# ---------------------------------------------------------------------------

120# Country-specific patterns

121# ---------------------------------------------------------------------------

122

123_COUNTRY_PATTERNS: dict[str, dict[str, tuple[re.Pattern[str], str]]] = {

124 # ── Singapore ──

125 "SG": {

126 "nric": (

127 re.compile(r"\b[STFGM]\d{7}[A-Z]\b"),

128 "[NRIC_REDACTED]",

129 ),

130 "sg_phone": (

131 re.compile(r"\+65\s?\d{4}\s?\d{4}\b"),

132 "[PHONE_REDACTED]",

133 ),

134 },

135 # ── India ──

136 "IN": {

137 "aadhaar": (

138 # Aadhaar: 12 digits, first digit 2-9 (never starts with 0 or 1)

139 re.compile(r"\b[2-9]\d{3}\s?\d{4}\s?\d{4}\b"),

140 "[AADHAAR_REDACTED]",

141 ),

142 "pan": (

143 re.compile(r"\b[A-Z]{5}\d{4}[A-Z]\b"),

144 "[PAN_REDACTED]",

145 ),

146 "in_phone": (

147 re.compile(r"\+91\s?\d{5}\s?\d{5}\b"),

148 "[PHONE_REDACTED]",

149 ),

150 },

151 # ── United States ──

152 "US": {

153 "us_passport": (

154 # US passport: letter prefix (optional since 2021) + 8-9 digits

155 re.compile(r"\b[A-Z]?\d{8,9}\b"),

156 "[PASSPORT_REDACTED]",

157 ),

158 },

159 # ── United Kingdom ──

160 "UK": {

161 "uk_nino": (

162 re.compile(r"\b[A-Z]{2}\s?\d{2}\s?\d{2}\s?\d{2}\s?[A-D]\b"),

163 "[NINO_REDACTED]",

164 ),

165 "uk_nhs": (

166 re.compile(r"\b\d{3}\s?\d{3}\s?\d{4}\b"),

167 "[NHS_REDACTED]",

168 ),

169 "uk_phone": (

170 re.compile(r"(?:\+44\s?\d{4}\s?\d{6}|0\d{4}\s?\d{6})\b"),

171 "[PHONE_REDACTED]",

172 ),

173 },

174 # ── EU: Germany ──

175 "DE": {

176 "de_personalausweis": (

177 # German ID: letter + digit + 8 alphanum + check digit (structured, not generic 10-char)

178 re.compile(r"\b[CFGHJKLMNPRTVWXYZ]\d[A-Z0-9]{6}\d\b"),

179 "[DE_ID_REDACTED]",

180 ),

181 },

182 # ── EU: France ──

183 "FR": {

184 "fr_insee": (

185 re.compile(r"\b[12]\s?\d{2}\s?\d{2}\s?\d{2}\s?\d{3}\s?\d{3}\s?\d{2}\b"),

186 "[FR_SSN_REDACTED]",

187 ),

188 },

189 # ── EU: Italy ──

190 "IT": {

191 "it_codice_fiscale": (

192 re.compile(r"\b[A-Z]{6}\d{2}[A-Z]\d{2}[A-Z]\d{3}[A-Z]\b"),

193 "[IT_CF_REDACTED]",

194 ),

195 },

196 # ── EU: Spain ──

197 "ES": {

198 "es_dni": (

199 re.compile(r"\b\d{8}[A-Z]\b"),

200 "[ES_DNI_REDACTED]",

201 ),

202 "es_nie": (

203 re.compile(r"\b[XYZ]\d{7}[A-Z]\b"),

204 "[ES_NIE_REDACTED]",

205 ),

206 },

207 # ── Australia ──

208 "AU": {

209 "au_tfn": (

210 re.compile(r"\b\d{3}\s?\d{3}\s?\d{2,3}\b"),

211 "[TFN_REDACTED]",

212 ),

213 "au_medicare": (

214 re.compile(r"\b\d{4}\s?\d{5}\s?\d{1,2}\b"),

215 "[MEDICARE_REDACTED]",

216 ),

217 "au_phone": (

218 re.compile(r"\+61\s?\d\s?\d{4}\s?\d{4}\b"),

219 "[PHONE_REDACTED]",

220 ),

221 },

222 # ── Canada ──

223 "CA": {

224 "ca_sin": (

225 re.compile(r"\b\d{3}\s?\d{3}\s?\d{3}\b"),

226 "[SIN_REDACTED]",

227 ),

228 },

229 # ── Japan ──

230 "JP": {

231 "jp_my_number": (

232 # My Number: 12 digits, context-aware to avoid Aadhaar overlap

233 re.compile(r"(?i)(?:my\s*number|マイナンバー)[:\s]+(\d{4}\s?\d{4}\s?\d{4})\b"),

234 "[MY_NUMBER_REDACTED]",

235 ),

236 "jp_phone": (

237 re.compile(r"\+81\s?\d{1,4}\s?\d{1,4}\s?\d{4}\b"),

238 "[PHONE_REDACTED]",

239 ),

240 },

241 # ── China ──

242 "CN": {

243 "cn_resident_id": (

244 re.compile(r"\b\d{6}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]\b"),

245 "[CN_ID_REDACTED]",

246 ),

247 "cn_phone": (

248 re.compile(r"\+86\s?1\d{2}\s?\d{4}\s?\d{4}\b"),

249 "[PHONE_REDACTED]",

250 ),

251 },

252}

253

254

255def _luhn_check(number: str) -> bool:

256 """Validate credit card or SIN number with Luhn algorithm."""

257 digits = [int(d) for d in number if d.isdigit()]

258 if len(digits) < 9 or len(digits) > 19:

259 return False

260 checksum = 0

261 for i, d in enumerate(reversed(digits)):

262 if i % 2 == 1:

263 d *= 2

264 if d > 9:

265 d -= 9

266 checksum += d

267 return checksum % 10 == 0

268

269

270# ---------------------------------------------------------------------------

271# PII Scanner

272# ---------------------------------------------------------------------------

273

274

275class PiiScanner:

276 """PII detection with multiple modes: regex, NER, LLM, rules_then_llm.

277

278 Sync scan() works for regex and NER modes.

279 Async scan_async()/apply_async() required for LLM and rules_then_llm modes.

280 """

281

282 def __init__(

283 self,

284 mode: str = "regex",

285 action: str = "redact",

286 custom_patterns: dict[str, tuple[str, str]] | None = None,

287 countries: list[str] | None = None,

288 type_overrides: dict[str, dict[str, str]] | None = None,

289 llm_provider: object | None = None,

290 ner_model: str = "en_core_web_sm",

291 ) -> None:

292 self.mode = mode

293 self.action = action # "redact" | "reject" | "warn"

294 self._type_overrides = type_overrides or {}

295

296 # Build pattern dict: credentials + global + country-specific.

297 self._patterns = {**_SECRET_PATTERNS, **_PII_PATTERNS}

298 if countries:

299 for country in countries:

300 country_upper = country.upper()

301 if country_upper in _COUNTRY_PATTERNS:

302 self._patterns.update(_COUNTRY_PATTERNS[country_upper])

303 if custom_patterns:

304 for name, (pattern, replacement) in custom_patterns.items():

305 self._patterns[name] = (re.compile(pattern), replacement)

306

307 # NER scanner (lazy init)

308 self._ner_scanner: NerPiiScanner | None = None

309 if mode in ("ner", "rules_then_llm"):

310 self._init_ner(ner_model)

311

312 # LLM scanner (lazy init)

313 self._llm_scanner: LlmPiiScanner | None = None

314 if mode in ("llm", "rules_then_llm") and llm_provider:

315 from astrocyte.policy.llm_scanner import LlmPiiScanner as _LlmScanner

316

317 self._llm_scanner = _LlmScanner(llm_provider)

318

319 def _init_ner(self, model: str) -> None:

320 """Initialize NER scanner. Fails gracefully if spaCy not installed."""

321 try:

322 from astrocyte.policy.ner_scanner import NerPiiScanner as _NerScanner

323

324 self._ner_scanner = _NerScanner(model)

325 except ImportError:

326 logger.warning("NER mode requested but spaCy not installed. Install with: pip install astrocyte[ner]")

327

328 # ── Sync scanning (regex + NER) ──

329

330 def scan(self, text: str) -> list[PiiMatch]:

331 """Scan text for PII. Returns list of matches.

332

333 Works for regex, ner, and rules_then_llm (regex+NER portion) modes.

334 For llm mode, use scan_async().

335 """

336 if self.mode == "disabled":

337 return []

338

339 matches = self._scan_regex(text)

340

341 # Include NER for ner mode and rules_then_llm (NER is part of "rules")

342 if self.mode in ("ner", "rules_then_llm") and self._ner_scanner:

343 ner_matches = self._ner_scanner.scan(text)

344 matches = self._merge_matches(matches, ner_matches)

345

346 return matches

347

348 def _scan_regex(self, text: str) -> list[PiiMatch]:

349 """Regex-only scan. Sync, pure."""

350 matches: list[PiiMatch] = []

351 for pii_type, (pattern, replacement) in self._patterns.items():

352 for m in pattern.finditer(text):

353 start, end = m.span("v") if "v" in pattern.groupindex else m.span()

354 matched_text = text[start:end]

355

356 # Credit card: validate with Luhn

357 if pii_type == "credit_card" and not _luhn_check(matched_text):

358 continue

359

360 # CA SIN: validate with Luhn

361 if pii_type == "ca_sin" and not _luhn_check(matched_text):

362 continue

363

364 matches.append(

365 PiiMatch(

366 pii_type=pii_type,

367 start=start,

368 end=end,

369 matched_text=matched_text,

370 replacement=replacement,

371 )

372 )

373

374 # Overlapping spans would be spliced with stale offsets when redacted

375 # back to front, corrupting the text; keep the earliest, longest.

376 return self._merge_matches(matches, [])

377

378 @staticmethod

379 def _merge_matches(a: list[PiiMatch], b: list[PiiMatch]) -> list[PiiMatch]:

380 """Merge two match lists, removing overlaps (prefer earlier/longer)."""

381 combined = sorted(a + b, key=lambda m: (m.start, -(m.end - m.start)))

382 merged: list[PiiMatch] = []

383 last_end = -1

384 for match in combined:

385 if match.start >= last_end:

386 merged.append(match)

387 last_end = match.end

388 return merged

389

390 # ── Async scanning (LLM + rules_then_llm) ──

391

392 async def scan_async(self, text: str) -> list[PiiMatch]:

393 """Async scan — supports all modes including LLM."""

394 if self.mode in ("disabled", "regex", "ner"):

395 return self.scan(text)

396

397 if self.mode == "llm" and self._llm_scanner:

398 return await self._llm_scanner.scan(text)

399

400 if self.mode == "rules_then_llm":

401 # Try regex + NER first

402 matches = self.scan(text)

403 if matches:

404 return matches

405 # Fall back to LLM

406 if self._llm_scanner:

407 return await self._llm_scanner.scan(text)

408

409 return self.scan(text)

410

411 # ── Apply actions ──

412

413 def apply(self, text: str) -> tuple[str, list[PiiMatch]]:

414 """Scan and apply action. Returns (processed_text, matches).

415

416 Raises PiiRejected if action is 'reject' and PII is found.

417 """

418 matches = self.scan(text)

419 return self._apply_matches(text, matches)

420

421 async def apply_async(self, text: str) -> tuple[str, list[PiiMatch]]:

422 """Async scan and apply — supports LLM modes."""

423 matches = await self.scan_async(text)

424 return self._apply_matches(text, matches)

425

426 def _apply_matches(self, text: str, matches: list[PiiMatch]) -> tuple[str, list[PiiMatch]]:

427 """Apply action to detected matches."""

428 if not matches:

429 return text, []

430

431 # Check per-type overrides for reject

432 reject_types = []

433 for match in matches:

434 override = self._type_overrides.get(match.pii_type)

435 if override and override.get("action") == "reject":

436 reject_types.append(match.pii_type)

437 elif not override and self.action == "reject":

438 reject_types.append(match.pii_type)

439

440 if reject_types:

441 raise PiiRejected(reject_types)

442

443 # Apply per-type actions

444 if self.action == "redact" or any(

445 self._type_overrides.get(m.pii_type, {}).get("action") == "redact" for m in matches

446 ):

447 result = text

448 for match in sorted(matches, key=lambda m: m.start, reverse=True):

449 override = self._type_overrides.get(match.pii_type)

450 action = override.get("action", self.action) if override else self.action

451 if action == "redact":

452 replacement = override.get("replacement", match.replacement) if override else match.replacement

453 result = result[: match.start] + (replacement or "[REDACTED]") + result[match.end :]

454 # warn: leave in place

455 return result, matches

456

457 # action == "warn": return original text with matches for logging

458 return text, matches

459

460

461# ---------------------------------------------------------------------------

462# Content validator

463# ---------------------------------------------------------------------------

464

465

466class ContentValidator:

467 """Validates retain content against policy rules.

468

469 Sync, stateless — Rust migration candidate.

470 """

471

472 def __init__(

473 self,

474 max_content_length: int = 50000,

475 reject_empty: bool = True,

476 allowed_content_types: list[str] | None = None,

477 ) -> None:

478 self.max_content_length = max_content_length

479 self.reject_empty = reject_empty

480 self.allowed_content_types = allowed_content_types or ["text", "conversation", "document"]

481

482 def validate(self, content: str, content_type: str = "text") -> list[str]:

483 """Validate content. Returns list of error messages (empty = valid)."""

484 errors: list[str] = []

485

486 if self.reject_empty and not content.strip():

487 errors.append("Content is empty")

488

489 if len(content) > self.max_content_length:

490 errors.append(f"Content too long: {len(content)} > {self.max_content_length}")

491

492 if content_type not in self.allowed_content_types:

493 errors.append(f"Content type '{content_type}' not allowed. Allowed: {self.allowed_content_types}")

494

495 return errors

496

497

498# ---------------------------------------------------------------------------

499# Metadata sanitizer

500# ---------------------------------------------------------------------------

501

502

503class MetadataSanitizer:

504 """Strips blocked keys and enforces size limits on metadata.

505

506 Sync, stateless — Rust migration candidate.

507 """

508

509 def __init__(

510 self,

511 blocked_keys: list[str] | None = None,

512 max_size_bytes: int = 4096,

513 ) -> None:

514 self.blocked_keys = set(blocked_keys or ["api_key", "password", "token", "secret"])

515 self.max_size_bytes = max_size_bytes

516

517 def sanitize(self, metadata: Metadata | None) -> tuple[Metadata | None, list[str]]:

518 """Sanitize metadata. Returns (cleaned_metadata, warnings)."""

519 if metadata is None:

520 return None, []

521

522 warnings: list[str] = []

523 cleaned: Metadata = {}

524

525 for key, value in metadata.items():

526 # Match exact key or key as a standalone word boundary segment

527 key_lower = key.lower()

528 if key_lower in self.blocked_keys or any(

529 re.search(rf"(?:^|[_.\-]){re.escape(bk)}(?:$|[_.\-])", key_lower) for bk in self.blocked_keys

530 ):

531 warnings.append(f"Blocked metadata key: '{key}'")

532 continue

533 cleaned[key] = value

534

535 # Size check — remove keys in reverse alphabetical order (deterministic)

536 serialized = json.dumps(cleaned, default=str)

537 if len(serialized.encode("utf-8")) > self.max_size_bytes:

538 warnings.append(f"Metadata exceeds {self.max_size_bytes} bytes, truncated")

539 keys_by_priority = sorted(cleaned.keys(), reverse=True) # z→a: drop least likely important first

540 for drop_key in keys_by_priority:

541 if len(json.dumps(cleaned, default=str).encode("utf-8")) <= self.max_size_bytes:

542 break

543 cleaned.pop(drop_key)

544

545 return cleaned if cleaned else None, warnings

546

547

548def redact_secrets(text: str) -> str:

549 """Replace credentials (API keys, tokens, private keys, passwords) with

550 ``[SECRET_REDACTED]``, leaving all other text untouched.

551

552 For text that is written somewhere before the retain barrier sees it,

553 such as the automatic-memory capture spool.

554 """

555 spans: list[tuple[int, int]] = []

556 for pattern, _ in _SECRET_PATTERNS.values():

557 for m in pattern.finditer(text):

558 spans.append(m.span("v") if "v" in pattern.groupindex else m.span())

559 kept: list[tuple[int, int]] = []

560 for start, end in sorted(spans, key=lambda se: (se[0], -se[1])): # earliest, then longest

561 if not kept or start >= kept[-1][1]:

562 kept.append((start, end))

563 for start, end in reversed(kept):

564 text = text[:start] + _SECRET + text[end:]

565 return text