from __future__ import annotations import argparse import json import math import multiprocessing import os import re import sys import warnings from collections import Counter from dataclasses import dataclass from pathlib import Path from typing import Literal from ._shim import LIVE_PREFIX_RE, PLACEHOLDER_RE, REFERENCE_RE, SLOT_RE, add_common_args, is_inert_value, nfc, read_jsonl, seeded_rng, stable_id, write_jsonl, write_report MARKER_RE = re.compile(r"\[REDACTED:([A-Z_]+)\]") KV_RE = re.compile(r"(?im)(?P[A-Za-zА-Яа-яЁё][\w.\-/ ]{0,64}?)[\"']?\s*(?P[:=])\s*(?:(?P[\"'])(?P[^\n\"']*)(?P=quote)|[«“](?P[^\n»”]*)[»”]|(?P[^\s,;&\]\}\n]+))") EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b") PHONE_RE = re.compile(r"(? float: frequencies = Counter(value) return -sum((count / len(value)) * math.log2(count / len(value)) for count in frequencies.values()) if value else 0.0 def _key_parts(key: str) -> tuple[str, list[str]]: camel = re.sub(r"(?<=[a-zа-яё])(?=[A-ZА-ЯЁ])", "_", key) camel = re.sub(r"(?<=[A-ZА-ЯЁ])(?=[A-ZА-ЯЁ][a-zа-яё])", "_", camel) normalized = re.sub(r"[\s.\-/]+", "_", camel.lower()).strip("_") return normalized, [part for part in normalized.split("_") if part] _FILE_EXT_RE = re.compile(r"\.(?:md|mdx|go|js|mjs|cjs|ts|tsx|jsx|py|sh|rb|rs|java|kt|cs|cpp|cc|c|h|hpp|yml|yaml|json|toml|txt|cfg|conf|ini|env|html|css|scss|sql|log|xml|lock)(?:$|[-:]\d)", re.I) def classify_key(key: str) -> str | None: if "/" in key or _FILE_EXT_RE.search(key.strip()): return None normalized, parts = _key_parts(key) if normalized in set(_ONTOLOGY["stop"]) or any(part in set(_ONTOLOGY["stop"]) for part in parts): return None for label, names in _ONTOLOGY["exact"].items(): if normalized in names: return label for label, segments in _ONTOLOGY["strong"].items(): if any(part in segments for part in parts): return label for label, weak in _ONTOLOGY["weak"].items(): for segment, qualifiers in weak.items(): if segment not in parts: continue if not qualifiers or any(part in qualifiers for part in parts): return label return None def _is_live_opaque(value: str) -> bool: compact = value.strip("\"'«»“”") return len(compact) >= 16 and any(character.isdigit() for character in compact) and re.fullmatch(r"[A-Za-z0-9+/=_-]+", compact) is not None and _entropy(compact) >= 3.5 def _format_preserving(value: str, *, seed: int, record_key: str, offset: int) -> str: rng = seeded_rng(seed, record_key, offset) if value.isdigit(): return "".join(rng.choice("0123456789") for _ in value) if re.fullmatch(r"[0-9A-Fa-f]+", value): alphabet = "0123456789ABCDEF" if value.upper() == value else "0123456789abcdef" return "".join(rng.choice(alphabet) for _ in value) replacement = [] for character in value: if character.isdigit(): replacement.append(rng.choice("0123456789")) elif character.isupper(): replacement.append(rng.choice("ABCDEFGHIJKLMNOPQRSTUVWXYZ")) elif character.islower(): replacement.append(rng.choice("abcdefghijklmnopqrstuvwxyz")) else: replacement.append(character) return "".join(replacement) _LITERAL_ESCAPE_RE = re.compile(r"\\+[nrt]") _STRIP_CHARS = "\"'«»“”\\()" def _kv_value(match: re.Match[str]) -> tuple[int, int, str, bool]: for group, quoted in (("quoted_value", True), ("smart_value", True), ("bare_value", False)): if match.group(group) is not None: start, end = match.span(group) value = match.group(group) if not quoted: cut = _LITERAL_ESCAPE_RE.search(value) if cut and cut.start() > 0: value = value[:cut.start()]; end = start + len(value) text = match.string opened = value.count("{") - value.count("}") + value.count("[") - value.count("]") while opened > 0 and end < len(text) and text[end] in "}]": value += text[end]; end += 1; opened -= 1 return start, end, value, quoted raise AssertionError("K/V regex produced no value") def _kv_core(start: int, end: int, value: str) -> tuple[int, int, str, bool]: core = value.strip(_STRIP_CHARS) lead = value.find(core) if core else 0 opening, closing = value[:lead], value[lead + len(core):] quotes = "\"'«»“”" wrapped = any(character in quotes for character in opening) and any(character in quotes for character in closing) return start + lead, start + lead + len(core), core, wrapped def _kv_matches(text: str): tokens = [match.span() for match in SLOT_RE.finditer(text)] position = 0 while True: match = KV_RE.search(text, position) if match is None: return key_start = match.start("key") enclosing = next((span for span in tokens if span[0] <= key_start < span[1]), None) if enclosing: position = enclosing[1] continue yield match position = match.end("key") def _overlaps(start: int, end: int, replacements: list[tuple[int, int, str]]) -> bool: return any(not (end <= left or start >= right) for left, right, _ in replacements) def _replace(text: str, replacements: list[tuple[int, int, str]]) -> str: for start, end, replacement in sorted(replacements, reverse=True): text = text[:start] + replacement + text[end:] return text def _merge_ranges(ranges: list[list[int]]) -> list[list[int]]: merged: list[list[int]] = [] for start, end in sorted(ranges): if merged and start <= merged[-1][1]: merged[-1][1] = max(merged[-1][1], end) else: merged.append([start, end]) return merged def sanitize_synthetic_text(text: str, covered_ranges: list[list[int]], *, seed: int, record_key: str, rules: list[GitleaksRule], max_rounds: int = 64) -> tuple[str, list[list[int]], int, int]: covered = _merge_ranges([list(item) for item in covered_ranges]); added: list[list[int]] = []; replaced_count = 0; rounds = 0 for _iteration in range(max_rounds): pending: list[tuple[int, int, str]] = [] def add_uncovered(start: int, end: int) -> None: pieces = [(start, end)] for covered_start, covered_end in covered: remainder = [] for piece_start, piece_end in pieces: if piece_end <= covered_start or piece_start >= covered_end: remainder.append((piece_start, piece_end)) else: if piece_start < covered_start: remainder.append((piece_start, covered_start)) if covered_end < piece_end: remainder.append((covered_end, piece_end)) pieces = remainder for piece_start, piece_end in pieces: if piece_start >= piece_end or SLOT_RE.search(text[piece_start:piece_end]) or _overlaps(piece_start, piece_end, pending): continue pending.append((piece_start, piece_end, _format_preserving(text[piece_start:piece_end], seed=seed, record_key=record_key, offset=piece_start))) for site in credential_sites(text, rules): if site.disposition != "quarantine": add_uncovered(site.start, site.end) if not pending: break text = _replace(text, pending) fresh = [[start, end] for start, end, _replacement in pending] added.extend(fresh); covered = _merge_ranges([*covered, *fresh]) replaced_count += len(pending); rounds += 1 return text, _merge_ranges(added), replaced_count, rounds _CREDENTIAL_ENTITIES = frozenset({"PASSWORD", "LOGIN", "JWT", "BEARER_TOKEN", "API_KEY", "PRIVATE_KEY", "DB_URL"}) def _pii_recognizers() -> list: global _PROJECT_PII_RECOGNIZERS if _PROJECT_PII_RECOGNIZERS is None: try: candidates = (Path(__file__).resolve().parents[2] / "presidio", Path("/app")) for directory in candidates: if (directory / "recognizers").is_dir() and str(directory) not in sys.path: sys.path.insert(0, str(directory)) from recognizers import ALL_RECOGNIZERS selected = [] for factory in ALL_RECOGNIZERS: recognizer = factory() supported = set(getattr(recognizer, "supported_entities", []) or []) if supported and supported.isdisjoint(_CREDENTIAL_ENTITIES): selected.append(recognizer) _PROJECT_PII_RECOGNIZERS = selected except Exception: _PROJECT_PII_RECOGNIZERS = [] return _PROJECT_PII_RECOGNIZERS def pii_recognizer_names() -> list[str]: return sorted(type(recognizer).__name__ for recognizer in _pii_recognizers()) def _project_pii_matches(text: str) -> list[tuple[int, int]]: matches = [] for recognizer in _pii_recognizers(): try: matches.extend((result.start, result.end) for result in recognizer.analyze(text, entities=None, nlp_artifacts=None)) except Exception: continue return matches def load_gitleaks_rules(path: Path | None) -> tuple[list[GitleaksRule], int]: if path is None or not path.exists(): return [], 0 try: import tomllib raw_rules = tomllib.loads(path.read_text(encoding="utf-8")).get("rules", []) except ImportError: raw_rules = [] for block in path.read_text(encoding="utf-8").split("[[rules]]")[1:]: identifier = re.search(r'^id\s*=\s*"([^"]+)"', block, re.M) regex = re.search(r"^regex\s*=\s*'''(.*?)'''", block, re.M | re.S) if identifier and regex: keywords = re.search(r"^keywords\s*=\s*\[(.*?)\]", block, re.M | re.S) secret_group = re.search(r"^secretGroup\s*=\s*(\d+)", block, re.M) entropy = re.search(r"^entropy\s*=\s*([\d.]+)", block, re.M) raw_rules.append({"id": identifier.group(1), "regex": regex.group(1), "keywords": re.findall(r'"([^"]+)"', keywords.group(1)) if keywords else [], "secretGroup": int(secret_group.group(1)) if secret_group else None, "entropy": float(entropy.group(1)) if entropy else None}) rules = []; skipped = 0 for rule in raw_rules: identifier = str(rule.get("id", "")).lower(); regex = rule.get("regex") if not isinstance(regex, str): continue label = "AUTH_TOKEN" if "token" in identifier else "SECRET_KEY" if "key" in identifier or "secret" in identifier else None if label is None: continue try: keywords = tuple(str(item).lower() for item in rule.get("keywords", []) if isinstance(item, str)) with warnings.catch_warnings(): warnings.simplefilter("error", DeprecationWarning) warnings.simplefilter("error", FutureWarning) pattern = re.compile(regex) group = int(rule.get("secretGroup") or (1 if pattern.groups else 0)) rules.append((pattern, label, keywords, min(group, pattern.groups), float(rule.get("entropy") or 0.0))) except (re.error, DeprecationWarning, FutureWarning): skipped += 1 return rules, skipped def _value_allowed(label: str, value: str, quoted: bool) -> bool: value = value.strip() if value.isdigit() and label != "PASSWORD": return False if label == "LOGIN": return bool(re.fullmatch(r"[\w.@][\w.@\-]{2,63}", value)) return 4 <= len(value) <= 256 and (quoted or not any(character.isspace() for character in value)) def detector_core(text: str, start: int, end: int) -> tuple[int, int]: value = text[start:end] if "=" in value: prefix, suffix = value.split("=", 1) if suffix and re.fullmatch(r"[?&]?[A-Za-zА-Яа-яЁё_][\w.-]{0,64}", prefix): start += len(prefix) + 1 while start < end and text[start] in "/?&": start += 1 while end > start and text[end - 1] in ".,;)]}": end -= 1 return start, end def _kv_site(match: re.Match[str], text: str) -> tuple[int, int, str, str | None, str | None]: start, end, value, quoted = _kv_value(match) start, end, value, wrapped = _kv_core(start, end, value); quoted = quoted or wrapped label = classify_key(match.group("key")) if not label or value.startswith("[PII_REDACTED"): return start, end, value, None, None if PLACEHOLDER_RE.match(value): return start, end, value, None, "placeholder" if is_inert_value(value, quoted): return start, end, value, None, "code_reference" if not _value_allowed(label, value, quoted) or _looks_like_user_sentence(match, text, end, quoted): return start, end, value, None, None start, end = detector_core(text, start, end) value = text[start:end] if not value: return start, end, value, None, "empty_core" return start, end, value, label, None def kv_credential_sites(text: str) -> list[tuple[int, int, str, str]]: sites = [] for match in _kv_matches(text): start, end, value, label, _reason = _kv_site(match, text) if label: sites.append((start, end, value, label)) return sites _CLI_CREDENTIAL_FLAGS = { "-u": CliCredentialFlag(frozenset({"curl"}), "userinfo", True), "--user": CliCredentialFlag(frozenset({"curl"}), "userinfo"), "-U": CliCredentialFlag(frozenset({"curl"}), "userinfo", True), "--proxy-user": CliCredentialFlag(frozenset({"curl"}), "userinfo"), "--oauth2-bearer": CliCredentialFlag(frozenset({"curl"}), "token"), "--password": CliCredentialFlag(None, "password"), "--passwd": CliCredentialFlag(None, "password"), "--pass": CliCredentialFlag(None, "password"), "--http-password": CliCredentialFlag(frozenset({"wget"}), "password"), "--proxy-password": CliCredentialFlag(frozenset({"wget"}), "password"), "--ftp-password": CliCredentialFlag(frozenset({"wget"}), "password"), "--auth": CliCredentialFlag(frozenset({"http", "https"}), "userinfo"), "-a": CliCredentialFlag(frozenset({"http", "https"}), "userinfo", True), "-p": CliCredentialFlag(frozenset({"mysql"}), "password", True), } _CLI_FLAG_RE = re.compile(r"(? tuple[int, int, str] | None: if position >= len(text): return None if text[position] in "\"'": closing = text.find(text[position], position + 1) if closing == -1: return None return position + 1, closing, text[position + 1:closing] match = _CLI_VALUE_RE.match(text, position) if not match: return None start, end = match.start(), match.end() while end > start: char = text[end - 1] if char in "\"',;": end -= 1 elif char in ")]}" and {")": "(", "]": "[", "}": "{"}[char] not in text[start:end - 1]: end -= 1 else: break return (start, end, text[start:end]) if end > start else None def _cli_part_kind(part: str) -> str: part = part.strip().strip("\"'") if not part: return "empty" if re.fullmatch(r"\$\{[A-Za-z_]\w*:-\}", part): return "empty" if re.fullmatch(r"\$\{[A-Za-z_]\w*:-.+\}", part): return "dynamic_with_literal_fallback" if _CLI_REFERENCE_RE.match(part) or PLACEHOLDER_RE.match(part) or is_inert_value(part, quoted=True): return "reference" return "literal" def _cli_read_shell_value(text: str, position: int) -> tuple[int, int, str, str] | None: if position >= len(text) or text[position].isspace(): return None start = position; quote = None; escaped = backtick = saw_escape = False; parens = braces = 0 while position < len(text): char = text[position] if escaped: escaped = False; position += 1; continue if char == "\\": escaped = saw_escape = True; position += 1; continue if quote: if char == quote: quote = None position += 1; continue if backtick: if char == "`": backtick = False position += 1; continue if char in "\"'": quote = char; position += 1; continue if char == "`": backtick = True; position += 1; continue if text.startswith("$(", position): parens += 1; position += 2; continue if text.startswith("${", position): braces += 1; position += 2; continue if char == ")" and parens: parens -= 1; position += 1; continue if char == "}" and braces: braces -= 1; position += 1; continue if char.isspace() and not parens and not braces: break position += 1 end = position malformed = any((quote, escaped, backtick, parens, braces, saw_escape)) wrapped = end - start >= 2 and text[start] in "\"'" and text[end - 1] == text[start] if wrapped: start += 1; end -= 1 elif not malformed: while end > start and text[end - 1] in "\"',;": end -= 1 value = text[start:end] if not value: return None kind = "ambiguous" if malformed else _cli_part_kind(value) if not malformed and (value.startswith("$(") or value.startswith("${") or value.startswith("`")): kind = "reference" return start, end, value, kind def _cli_command(text: str, position: int) -> str | None: segment = re.split(r"(?:;|&&|\|\| |\|)", text[:position])[-1] embedded = re.findall(r"(?:^|[\s\"'])((?:curl|wget|mysql|http|https))(?:\.exe)?(?=$|[\s\"'])", segment, re.I) if embedded: return embedded[-1].lower() for word in re.findall(r"(?:[^\s'\"]+|'[^']*'|\"[^\"]*\")+", segment): token = re.sub(r"\.exe$", "", word.strip("'\"").rsplit("/", 1)[-1].lower()) if token in _CLI_WRAPPERS or token.startswith("-") or re.fullmatch(r"[A-Za-z_]\w*=.*", token): continue return token return None def _cli_credential_parts(text: str): for flag_match in _CLI_FLAG_RE.finditer(text): flag = flag_match.group(1) config = _CLI_CREDENTIAL_FLAGS.get(flag) if config is None: continue command = _cli_command(text, flag_match.start()) if config.commands is not None and command not in config.commands: continue after = flag_match.end() if after < len(text) and text[after] == "=": read = _cli_read_shell_value(text, after + 1) elif config.attached and after < len(text) and not text[after].isspace(): read = _cli_read_shell_value(text, after) else: cursor = after while cursor < len(text) and text[cursor] in " \t": cursor += 1 read = _cli_read_shell_value(text, cursor) if cursor > after or (after < len(text) and text[after].isspace()) else None if read is None: continue value_start, value_end, value, value_kind = read part = {"password": None, "username": None} if config.mode == "userinfo": colon = value.find(":") if colon == -1: continue username = value[:colon] part["username"] = (value_start, value_start + colon, username, "LOGIN", _cli_part_kind(username)) password_kind = value_kind if value_kind in {"ambiguous", "dynamic_with_literal_fallback"} else _cli_part_kind(value[colon + 1:]) part["password"] = (value_start + colon + 1, value_end, value[colon + 1:], "PASSWORD", password_kind) else: part["password"] = (value_start, value_end, value, "AUTH_TOKEN" if config.mode == "token" else "PASSWORD", value_kind) yield part def cli_credential_sites(text: str) -> list[tuple[int, int, str, str, str]]: sites: list[tuple[int, int, str, str, str]] = [] for part in _cli_credential_parts(text): password = part["password"] if password and password[4] != "empty": sites.append(password) return sites def cli_credential_slots(text: str) -> list[tuple[int, int, str, str]]: slots: list[tuple[int, int, str, str]] = [] for part in _cli_credential_parts(text): password = part["password"] if not password or password[4] != "literal": continue username = part["username"] if username and username[4] == "literal": slots.append((username[0], username[1], "LOGIN", "cli")) slots.append((password[0], password[1], password[3], "cli")) return slots def credential_sites(text: str, rules: list[GitleaksRule]) -> list[CredentialSite]: sites = [CredentialSite(start, end, label, "kv", "slot") for start, end, _value, label in kv_credential_sites(text)] sites += [CredentialSite(start, end, label, "cli", "slot" if kind == "literal" else "quarantine") for start, end, _value, label, kind in cli_credential_sites(text) if kind in {"literal", "ambiguous", "dynamic_with_literal_fallback"}] sites += [CredentialSite(start, end, label or "SECRET", "opaque", "replace") for start, end, label, _match_start, _match_end in _find_live(text, rules)] return sorted(set(sites), key=lambda item: (item.start, item.end, item.label, item.detector)) def _looks_like_user_sentence(match: re.Match[str], text: str, value_end: int, quoted: bool) -> bool: key = match.group("key") _, parts = _key_parts(key) if quoted: return False remainder = text[value_end:text.find("\n", value_end) if text.find("\n", value_end) != -1 else len(text)] continues_as_sentence = bool(re.match(r"\s+[\wА-Яа-яЁё]{2,}", remainder)) return continues_as_sentence and ( "user" in parts or (match.group("sep") == ":" and " " in key and len(parts) >= 3) ) def _find_live(text: str, rules: list[GitleaksRule]) -> list[tuple[int, int, str | None, int, int]]: found = [] for match in OPAQUE_RE.finditer(text): if not _is_live_opaque(match.group()): continue start, end, value, label = match.start(), match.end(), match.group(), None if "=" in value: prefix, suffix = value.split("=", 1) classified = classify_key(prefix) if classified and suffix: start += len(prefix) + 1; value = suffix; label = classified found.append((start, end, label, match.start(), match.end())) lowered = text.lower() for pattern, label, keywords, group, entropy in rules: if keywords and not any(keyword in lowered for keyword in keywords): continue for match in pattern.finditer(text): start, end = match.span(group) if start < 0 or end <= start: start, end = match.span() if entropy and _entropy(text[start:end]) < entropy: continue found.append((start, end, label, match.start(), match.end())) normalized = [] for start, end, label, match_start, match_end in found: start, end = detector_core(text, start, end) if start < end: normalized.append((start, end, label, match_start, match_end)) return sorted(set(normalized), key=lambda item: (item[0], item[1], item[2] or "")) def scrub_row(raw: dict, slot_counter: int, *, seed: int = 42, gitleaks_rules: list[GitleaksRule] | None = None) -> tuple[dict | None, list[dict], dict[str, int], int]: text = nfc(str(raw.get("content", raw.get("text", "")))) record_key = str(raw.get("group_id") or raw.get("conversation_id") or raw.get("session_id") or stable_id(text)) counts: Counter[str] = Counter(); slots: list[dict] = []; flags = list(raw.get("flags", [])); negative_reason = raw.get("negative_reason") pii_replacements: list[tuple[int, int, str]] = [] for pattern in (EMAIL_RE, PHONE_RE, SNILS_RE, CARD_RE, PASSPORT_RE, INN_RE): for match in pattern.finditer(text): start, end = match.span(1) if pattern in {INN_RE, PASSPORT_RE} else match.span() if not _overlaps(start, end, pii_replacements): pii_replacements.append((start, end, "[PII_REDACTED]")); counts["pii_replaced"] += 1 for start, end in _project_pii_matches(text): if not _overlaps(start, end, pii_replacements): pii_replacements.append((start, end, "[PII_REDACTED]")); counts["pii_replaced"] += 1 text = _replace(text, pii_replacements) if pii_replacements: flags.append("pii_replaced") replacements: list[tuple[int, int, str]] = []; entropy_replacements: list[tuple[int, int, str]] = [] def add_slot(start: int, end: int, label: str, slot_type: str) -> None: nonlocal slot_counter if _overlaps(start, end, replacements): return token = f"⟦SLOT:{label}:{slot_counter}⟧"; slot_counter += 1 replacements.append((start, end, token)) slots.append({"slot_id": token, "record_key": record_key, "klass": label, "injection_slot_type": slot_type}) counts["slots"] += 1 for marker in MARKER_RE.finditer(text): kind = marker.group(1); label = MARKER_LABELS.get(kind) or classify_key(kind) if kind == "ENV": context = text[max(0, marker.start() - 120):marker.end() + 120] key_match = re.search(r"([\w.-]+)\s*[:=]\s*\[REDACTED:ENV\]", context, re.I) label = classify_key(key_match.group(1)) if key_match else None if label: add_slot(marker.start(), marker.end(), label, "redacted_marker") cli_quarantine_ranges = [(site.start, site.end) for site in credential_sites(text, gitleaks_rules or []) if site.detector == "cli" and site.disposition == "quarantine"] for match in _kv_matches(text): start, end, _value, label, reason = _kv_site(match, text) if any(not (end <= left or start >= right) for left, right in cli_quarantine_ranges): continue if reason: negative_reason = negative_reason or reason; counts[reason] += 1; continue if label: add_slot(start, end, label, "dsn" if "://" in text[max(0, match.start() - 50):match.end() + 50] else "kv_pair") for match in CONTRACT_RE.finditer(text): if not _overlaps(match.start(1), match.end(1), replacements): add_slot(match.start(1), match.end(1), "CONTRACT_NUMBER", "kv_pair") for start, end, label, slot_type in cli_credential_slots(text): add_slot(start, end, label, slot_type) for start, end, _label, _match_start, _match_end in _find_live(text, gitleaks_rules or []): if _overlaps(start, end, replacements): overlapping = sorted((left, right) for left, right, _ in replacements if left < end and right > start) leftovers = []; cursor = start for left, right in overlapping: if left > cursor: leftovers.append((cursor, left)) cursor = max(cursor, right) if cursor < end: leftovers.append((cursor, end)) if any(_is_live_opaque(text[piece_start:piece_end]) for piece_start, piece_end in leftovers): counts["quarantined_conflict"] += 1 return None, [], dict(counts), slot_counter continue replacement = _format_preserving(text[start:end], seed=seed, record_key=record_key, offset=start) item = (start, end, replacement); replacements.append(item); entropy_replacements.append(item); counts["entropy_replaced_values"] += 1 safe_text = _replace(text, replacements) final_ranges = []; delta = 0 entropy_starts = {(start, end, replacement) for start, end, replacement in entropy_replacements} for start, end, replacement in sorted(replacements): final_start = start + delta; final_end = final_start + len(replacement) if (start, end, replacement) in entropy_starts: final_ranges.append([final_start, final_end]) delta += len(replacement) - (end - start) final_ranges = _merge_ranges(final_ranges) if entropy_replacements: flags.append("entropy_replaced") for _iteration in range(64): pending: list[tuple[int, int, str]] = [] def add_post_replacement(start: int, end: int) -> None: pieces = [(start, end)] for covered_start, covered_end in final_ranges: remainder = [] for piece_start, piece_end in pieces: if piece_end <= covered_start or piece_start >= covered_end: remainder.append((piece_start, piece_end)) else: if piece_start < covered_start: remainder.append((piece_start, covered_start)) if covered_end < piece_end: remainder.append((covered_end, piece_end)) pieces = remainder for piece_start, piece_end in pieces: if piece_start >= piece_end or SLOT_RE.search(safe_text[piece_start:piece_end]) or _overlaps(piece_start, piece_end, pending): continue pending.append((piece_start, piece_end, _format_preserving(safe_text[piece_start:piece_end], seed=seed, record_key=record_key, offset=piece_start))) for site in credential_sites(safe_text, gitleaks_rules or []): if site.disposition != "quarantine": add_post_replacement(site.start, site.end) if not pending: break safe_text = _replace(safe_text, pending) final_ranges = _merge_ranges([*final_ranges, *[[start, end] for start, end, _replacement in pending]]) counts["postscan_replaced_values"] += len(pending) counts["postscan_rounds"] += 1 if "entropy_replaced" not in flags: flags.append("entropy_replaced") slot_spans = [list(match.span()) for match in SLOT_RE.finditer(safe_text)] covered = _merge_ranges([*final_ranges, *slot_spans]) for site in credential_sites(safe_text, gitleaks_rules or []): if site.detector == "cli" and site.disposition == "quarantine" and not any(left <= site.start and site.end <= right for left, right in covered): counts["cli_credential_quarantine"] += 1 return None, [], dict(counts), slot_counter row = dict(raw); row.pop("content", None); row["text"] = safe_text; row["flags"] = sorted(set(flags)); row["negative_reason"] = negative_reason; row["entropy_replaced_ranges"] = final_ranges return row, slots, dict(counts), slot_counter def unresolved_live_values(row: dict, rules: list[GitleaksRule]) -> int: ranges = _merge_ranges([list(item) for item in row.get("entropy_replaced_ranges", [])]) slots = [match.span() for match in SLOT_RE.finditer(row["text"])] def covered(start: int, end: int) -> bool: return any(left <= start and end <= right for left, right in [*ranges, *slots]) return sum(not covered(site.start, site.end) for site in credential_sites(row["text"], rules)) _SLOT_BASE_STRIDE = 65536 _WORKER_RULES: list[GitleaksRule] = [] def _init_worker(config: str) -> None: global _WORKER_RULES _WORKER_RULES = load_gitleaks_rules(Path(config))[0] if config else [] def _scrub_chunk(payload: tuple[int, int, list[str]]) -> tuple[list[str], list[dict], dict[str, int]]: seed, start_index, lines = payload out_rows: list[str] = []; out_slots: list[dict] = []; counts: Counter[str] = Counter() for offset, line in enumerate(lines): raw = json.loads(line); counts["input"] += 1 safe, found, row_counts, _ = scrub_row(raw, (start_index + offset) * _SLOT_BASE_STRIDE, seed=seed, gitleaks_rules=_WORKER_RULES) counts.update(row_counts) if safe is None: counts["quarantined"] += 1 else: counts["safe"] += 1; out_rows.append(json.dumps(safe, ensure_ascii=False, sort_keys=True, separators=(",", ":"))); out_slots.extend(found) return out_rows, out_slots, dict(counts) def _unresolved_chunk(lines: list[str]) -> int: return sum(unresolved_live_values(json.loads(line), _WORKER_RULES) for line in lines) def _line_chunks(paths: list[Path], size: int): index = 0; buffer: list[str] = [] for path in paths: with path.open(encoding="utf-8") as stream: for line in stream: if not line.strip(): continue buffer.append(line) if len(buffer) >= size: yield index, buffer; index += len(buffer); buffer = [] if buffer: yield index, buffer def main() -> None: parser = argparse.ArgumentParser(description=__doc__); parser.add_argument("--input", type=Path, action="append", help="turns JSONL; defaults to both extract outputs"); parser.add_argument("--gitleaks-config", type=Path, default=Path("../dataset/gitleaks/config/gitleaks.toml")); parser.add_argument("--workers", type=int, default=max(1, (os.cpu_count() or 2) - 2)); parser.add_argument("--verify-only", action="store_true", help="re-scan the existing scrubbed_turns.jsonl without rewriting it"); parser.add_argument("--allow-missing-pii-recognizers", action="store_true", help="proceed without personal-data recognizers; the output is then NOT safe to export"); add_common_args(parser); args = parser.parse_args() inputs = args.input or [args.staging / "codechat_turns.jsonl", args.staging / "swechat_turns.jsonl"] recognizers = pii_recognizer_names() if not recognizers and not args.allow_missing_pii_recognizers: raise SystemExit("PII recognizers unavailable (is presidio_analyzer installed?); " "re-run inside the pipeline image or pass --allow-missing-pii-recognizers") rules, skipped_rules = load_gitleaks_rules(args.gitleaks_config); slots: list[dict] = []; counts: Counter[str] = Counter() config = str(args.gitleaks_config) if args.gitleaks_config and Path(args.gitleaks_config).exists() else "" pool = multiprocessing.get_context("spawn").Pool(args.workers, initializer=_init_worker, initargs=(config,)) scrubbed_path = args.staging / "scrubbed_turns.jsonl"; scrubbed_path.parent.mkdir(parents=True, exist_ok=True) if args.verify_only: report_path = args.staging / "quarantine_report.json" report = json.loads(report_path.read_text(encoding="utf-8")) if report_path.is_file() else {} if report.get("scrubber_policy_version") != SCRUBBER_POLICY_VERSION: raise SystemExit("staging was not produced by the current scrubber policy") with pool: unresolved = sum(pool.imap_unordered(_unresolved_chunk, (lines for _, lines in _line_chunks([scrubbed_path], 2000)))) print(json.dumps({"unresolved_live_values": unresolved, "workers": args.workers}, sort_keys=True)) raise SystemExit(1 if unresolved else 0) with pool, scrubbed_path.open("w", encoding="utf-8", newline="\n") as out: payloads = ((args.seed, index, lines) for index, lines in _line_chunks(inputs, 2000)) for out_rows, out_slots, chunk_counts in pool.imap(_scrub_chunk, payloads): counts.update(chunk_counts); slots.extend(out_slots) for line in out_rows: out.write(line + "\n") write_jsonl(args.staging / "slots.jsonl", slots) out.flush() unresolved = sum(pool.imap_unordered(_unresolved_chunk, (lines for _, lines in _line_chunks([scrubbed_path], 2000)))) report = {"scrubber_policy_version": SCRUBBER_POLICY_VERSION, "input": counts["input"], "safe": counts["safe"], "quarantined": counts["quarantined"], "unresolved_live_values": unresolved, "gitleaks_rules": len(rules), "gitleaks_rules_skipped": skipped_rules, "workers": args.workers, "pii_recognizers": recognizers, "by_type": dict(sorted(counts.items()))} (args.staging / "quarantine_report.json").write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8"); write_report(args.staging, "scrubber", **report); print(json.dumps(report, ensure_ascii=False, sort_keys=True)) if unresolved: raise SystemExit(1) if __name__ == "__main__": main()