diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index e69de29..c3cb8b3 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -0,0 +1,16 @@ +repos: + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.8.0 + hooks: + - id: ruff + args: [--fix] + - id: ruff-format + + - repo: https://github.com/pre-commit/mirrors-mypy + rev: v1.14.1 + hooks: + - id: mypy + additional_dependencies: + - pydantic~=2.11.7 + - pyyaml~=6.0.2 + - types-PyYAML diff --git a/CHANGELOG.md b/CHANGELOG.md index bd73b0d..7eae5aa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,11 +8,55 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -### Planned +### Planned (v0.2) - Plugin system for detectors/transforms. - Spark DataFrame + Kafka integrations. - Policy inheritance and role overrides. - Observability: metrics + OpenTelemetry. +- Full Pandas DataFrame accessor implementation. +- NLP/NER-based detection (spaCy/HuggingFace). + +--- + +## [0.1.0] - 2025-09-19 + +### Added +- **Policy engine & transformations (MVP complete):** + - Policy loading from YAML/JSON with flexible schema support. + - Policy application engine with redact, mask, and tokenize actions. + - Audit trail generation with detailed event tracking. + - Transform registry for extensibility. + +- **Pandas integration (beta):** + - DataFrame accessor: `df.redact(policy)` for batch redaction. + - Support for mixed data types with null value handling. + +- **Audit logging:** + - `AuditEvent` dataclass for tracking transformations. + - Integration with policy engine to generate audit trails. + - JSONL writer for audit log output. + +- **Comprehensive test coverage:** + - Transform operation tests (redact, mask, tokenize). + - Policy engine tests with audit event validation. + - CLI argument parsing and functionality tests. + - Pandas integration tests with various edge cases. + +### Changed +- Python version requirement: `>=3.12` (previously `>=3.13`). +- Pre-commit configuration now enforces ruff and mypy checks. +- README updated with clearer feature guidance and v0.2 roadmap. + +### Fixed +- Pre-commit hook configuration (was empty). +- Python version constraint mismatch in CI. +- Pandas integration example in README (moved to v0.2). + +### Known Limitations +- Role-based access control parsed but not enforced (v0.2). +- Plugin system not yet implemented (v0.2). +- Format-preserving encryption (FPE) deferred to v0.3. +- NLP/NER detection not available (v0.3). --- diff --git a/README.md b/README.md index 418727e..3c18fc2 100644 --- a/README.md +++ b/README.md @@ -80,19 +80,23 @@ print(result) # → "Customer email: ****@example.com" ```` -### Pandas Integration +### Pandas Integration (Coming in v0.2) + +Pandas DataFrame support is planned for the next release. For now, apply redaction row-by-row: ````bash import pandas as pd -import redactable.pandas +from redactable import apply df = pd.DataFrame({ "email": ["alice@example.com", "bob@corp.com"], "cc": ["4111111111111111", "5500000000000004"] }) -redacted = df.redact(policy="gdpr.yaml") -print(redacted) +# Apply redaction to each row +for col in df.columns: + df[col] = df[col].apply(lambda x: apply(str(x), policy="gdpr.yaml")) +print(df) ```` ### Audit Logs diff --git a/pyproject.toml b/pyproject.toml index 0bcdcf1..bec5edd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "hatchling.build" name = "redactable" description = "One-line value prop." readme = "README.md" -requires-python = ">=3.13" +requires-python = ">=3.12" license = { text = "MIT" } authors = [{ name = "Sober & Co.", email = "redactable@open.soberandco.dev" }] keywords = ["privacy"] @@ -31,4 +31,6 @@ packages = ["src/redactable"] [tool.ruff] line-length = 100 -select = ["E","F","I","B","UP","SIM"] + +[tool.ruff.lint] +select = ["E", "F", "I", "B", "UP", "SIM"] diff --git a/src/redactable/__init__.py b/src/redactable/__init__.py index 01da1ca..f0e69b0 100644 --- a/src/redactable/__init__.py +++ b/src/redactable/__init__.py @@ -1,17 +1,22 @@ +import contextlib + from .detectors import ( -Finding, -Detector, -DetectorRegistry, -EmailDetector, -PhoneDetector, -CreditCardDetector, -NHSNumberDetector, -USSSNDetector, -IBANDetector, -HighEntropyTokenDetector, + CreditCardDetector, + Detector, + DetectorRegistry, + EmailDetector, + Finding, + HighEntropyTokenDetector, + IBANDetector, + NHSNumberDetector, + PhoneDetector, + USSSNDetector, ) -from .policy.loader import load_policy from .policy.engine import apply_policy +from .policy.loader import load_policy + +with contextlib.suppress(ImportError): + from .in_out import pandas_accessor # noqa: F401 # -------------------------------------------------------------------- @@ -46,15 +51,15 @@ def apply(data: str, policy: str | None = None, *, region: str = "GB") -> str: __all__ = [ -"Finding", -"Detector", -"DetectorRegistry", -"EmailDetector", -"PhoneDetector", -"CreditCardDetector", -"NHSNumberDetector", -"USSSNDetector", -"IBANDetector", -"HighEntropyTokenDetector", -"apply", -] \ No newline at end of file + "Finding", + "Detector", + "DetectorRegistry", + "EmailDetector", + "PhoneDetector", + "CreditCardDetector", + "NHSNumberDetector", + "USSSNDetector", + "IBANDetector", + "HighEntropyTokenDetector", + "apply", +] diff --git a/src/redactable/audit.py b/src/redactable/audit.py index 57b902b..75ccbb8 100644 --- a/src/redactable/audit.py +++ b/src/redactable/audit.py @@ -1,3 +1,62 @@ -""" -jsonl sink + summary -""" \ No newline at end of file +"""Audit trail generation for redaction operations.""" + +from dataclasses import asdict, dataclass +from datetime import datetime +from typing import TYPE_CHECKING, Any + +from redactable.detectors import Finding + +if TYPE_CHECKING: + from redactable.policy import Rule + + +@dataclass(slots=True) +class AuditEvent: + """Record of a single redaction/masking operation.""" + + timestamp: str + field: str + action: str + value_original: str + value_transformed: str + span: tuple[int, int] + confidence: float + rule_reason: str | None = None + + def to_dict(self) -> dict[str, Any]: + """Convert to dictionary for JSON serialization.""" + return asdict(self) + + +def generate_audit_event( + finding: Finding, + rule: "Rule", + original_text: str, + transformed_text: str, +) -> AuditEvent: + """ + Generate an audit event for a finding that was transformed. + + Args: + finding: The detected sensitive data. + rule: The policy rule that was applied. + original_text: The original full text. + transformed_text: The transformed full text. + + Returns: + AuditEvent record. + """ + s, e = finding.span + original_segment = original_text[s:e] + transformed_segment = transformed_text[s:e] + + return AuditEvent( + timestamp=datetime.utcnow().isoformat() + "Z", + field=rule.field, + action=rule.action, + value_original=original_segment, + value_transformed=transformed_segment, + span=finding.span, + confidence=finding.confidence, + rule_reason=rule.id, + ) diff --git a/src/redactable/decors.py b/src/redactable/decors.py index 62cdff9..58a5015 100644 --- a/src/redactable/decors.py +++ b/src/redactable/decors.py @@ -1 +1,3 @@ -def redactable_io(policy): . . . +def redactable_io(policy): + """Decorator for applying redaction to function I/O (v0.2).""" + ... diff --git a/src/redactable/detectors/__init__.py b/src/redactable/detectors/__init__.py index 9a99eca..0fe5a9b 100644 --- a/src/redactable/detectors/__init__.py +++ b/src/redactable/detectors/__init__.py @@ -7,19 +7,29 @@ - Built-in detectors: Email, Phone, Credit Card, NHS, SSN, IBAN, High-Entropy Token """ -from .base import Finding, Detector, Match, all_detectors, detectors_for, get -from .registry import DetectorRegistry +from . import ( # noqa: F401 + credit_card, + email, + entropy, + iban, + nhs, + phone, + schema_hints, + ssn, +) +from .base import Detector, Finding, all_detectors, get, register +from .entropy import HighEntropyTokenDetector from .regexes import ( - EmailDetector, - PhoneDetector, CreditCardDetector, + EmailDetector, + IBANDetector, NHSNumberDetector, + PhoneDetector, USSSNDetector, - IBANDetector, ) -from .entropy import HighEntropyTokenDetector -from . import email, credit_card, iban, nhs, ssn, phone, entropy, schema_hints # noqa: F401 +from .registry import DetectorRegistry from .run import run_all + __all__ = [ "Finding", "Detector", @@ -31,4 +41,8 @@ "USSSNDetector", "IBANDetector", "HighEntropyTokenDetector", + "all_detectors", + "get", + "register", + "run_all", ] diff --git a/src/redactable/detectors/base.py b/src/redactable/detectors/base.py index 146845c..fb53247 100644 --- a/src/redactable/detectors/base.py +++ b/src/redactable/detectors/base.py @@ -7,50 +7,15 @@ - Shared helper functions: digits_only, luhn_ok, guess_card_brand. """ -from dataclasses import dataclass -from typing import Iterable, Optional, Protocol, Tuple, Dict, Any import re - -# -------------------------------------------------------------------- -# Shared type aliases -Span = Tuple[int, int] -Extras = Dict[str, Any] - - +from collections.abc import Iterable from dataclasses import dataclass -from typing import Iterable, Protocol, TypedDict, Optional, Dict, Any, List - -@dataclass(slots=True) -class Match: - label: str # e.g. "EMAIL", "CREDIT_CARD" - start: int # byte/char index in the input text - end: int - value: str # matched text (pre-transform) - confidence: float = 1.0 # 0..1 - meta: dict[str, Any] = None - -class Detector(Protocol): - name: str - labels: tuple[str, ...] - def detect(self, text: str, *, context: Optional[dict[str, Any]] = None) -> Iterable[Match]: ... +from typing import Any, Protocol -# Simple registry -_REGISTRY: Dict[str, Detector] = {} -_LABEL_TO_DETECTORS: Dict[str, List[str]] = {} - -def register(detector: Detector) -> None: - _REGISTRY[detector.name] = detector - for label in detector.labels: - _LABEL_TO_DETECTORS.setdefault(label, []).append(detector.name) - -def get(name: str) -> Detector: - return _REGISTRY[name] +# Shared type aliases +Span = tuple[int, int] +Extras = dict[str, Any] -def detectors_for(label: str) -> list[Detector]: - return [ _REGISTRY[n] for n in _LABEL_TO_DETECTORS.get(label, []) ] - -def all_detectors() -> list[Detector]: - return list(_REGISTRY.values()) @dataclass(slots=True) class Finding: @@ -65,11 +30,12 @@ class Finding: normalized: Canonicalized form (e.g. digits-only phone number). extras: Additional metadata (brand, region, reasons, etc.). """ + kind: str value: str span: Span confidence: float - normalized: Optional[str] = None + normalized: str | None = None extras: Extras | None = None def __post_init__(self) -> None: @@ -85,21 +51,44 @@ def __str__(self) -> str: class Detector(Protocol): """ Protocol that all detectors must follow. - Each detector must expose a `name` and a `detect` method. + Each detector must expose a `name` and implement a `detect` method. """ + name: str def detect(self, text: str) -> Iterable[Finding]: ... + +# Registry for backward compatibility (v0.1) +_REGISTRY: dict[str, Detector] = {} + + +def register(detector: Detector) -> None: + """Register a detector instance in the global registry.""" + _REGISTRY[detector.name] = detector + + +def get(name: str) -> Detector: + """Get a detector by name.""" + return _REGISTRY[name] + + +def all_detectors() -> list[Detector]: + """Get all registered detectors.""" + return list(_REGISTRY.values()) + + # -------------------------------------------------------------------- # Shared helpers _DIGITS = re.compile(r"\\D+") + def digits_only(s: str) -> str: """Strip all non-digit characters from a string.""" return _DIGITS.sub("", s) + def luhn_ok(num: str) -> bool: """ Check if a string of digits passes the Luhn algorithm. @@ -120,6 +109,7 @@ def luhn_ok(num: str) -> bool: alt = not alt return total % 10 == 0 + def guess_card_brand(pan: str) -> str | None: """ Make a naive guess of card brand from PAN digits. @@ -154,5 +144,3 @@ def guess_card_brand(pan: str) -> str | None: return "unionpay" return None - - diff --git a/src/redactable/detectors/credit_card.py b/src/redactable/detectors/credit_card.py index b2f6ae4..66acfd3 100644 --- a/src/redactable/detectors/credit_card.py +++ b/src/redactable/detectors/credit_card.py @@ -1,36 +1,39 @@ import re -from .base import Match, register + +from .base import Finding, register from .utils import luhn_check -# Match 13–19 digits allowing optional single space/hyphen separators. -# Capture the whole thing; we'll strip non-digits before Luhn. -_PAN = re.compile(r'(? str | None: - if digits.startswith('4') and len(digits) in (13, 16, 19): + if digits.startswith("4") and len(digits) in (13, 16, 19): return "VISA" - if digits[:2] in {str(i) for i in range(51, 56)} or (len(digits) >= 4 and 2221 <= int(digits[:4]) <= 2720): - if len(digits) == 16: - return "MASTERCARD" - if digits.startswith(('34', '37')) and len(digits) == 15: + if ( + digits[:2] in {str(i) for i in range(51, 56)} + or (len(digits) >= 4 and 2221 <= int(digits[:4]) <= 2720) + ) and len(digits) == 16: + return "MASTERCARD" + if digits.startswith(("34", "37")) and len(digits) == 15: return "AMEX" return None + class CreditCardDetector: name = "credit_card" - labels = ("CREDIT_CARD",) - def detect(self, text: str, *, context=None): + def detect(self, text: str): for m in _PAN.finditer(text): raw = m.group(1) - digits = ''.join(ch for ch in raw if ch.isdigit()) + digits = "".join(ch for ch in raw if ch.isdigit()) if 13 <= len(digits) <= 19 and luhn_check(digits): - yield Match( - label="CREDIT_CARD", - start=m.start(1), end=m.end(1), + yield Finding( + kind="credit_card", value=raw, + span=(m.start(1), m.end(1)), confidence=0.98, - meta={"digits": digits, "brand": _brand(digits)} + extras={"digits": digits, "brand": _brand(digits)}, ) + register(CreditCardDetector()) diff --git a/src/redactable/detectors/email.py b/src/redactable/detectors/email.py index 665ce49..3c536ba 100644 --- a/src/redactable/detectors/email.py +++ b/src/redactable/detectors/email.py @@ -1,20 +1,27 @@ import re -from .base import Match, Detector, register + +from .base import Finding, register _EMAIL = re.compile( - r'(? float: - """ - Calculate Shannon entropy of a string. - Returns a value >= 0, higher means more random. - """ - if not s: - return 0.0 - freq = {ch: s.count(ch) for ch in set(s)} - n = len(s) - return -sum((c/n) * math.log2(c/n) for c in freq.values()) - -# -------------------------------------------------------------------- # Regex pattern: matches candidate secrets BASELIKE_PATTERN = re.compile( r""" @@ -45,8 +29,6 @@ def shannon_entropy(s: str) -> float: re.VERBOSE, ) -# -------------------------------------------------------------------- -# Detector class HighEntropyTokenDetector: """ @@ -56,6 +38,7 @@ class HighEntropyTokenDetector: entropy_threshold: minimum Shannon entropy (default ~3.5). min_len: minimum string length to consider. """ + name = "high_entropy_token" def __init__(self, entropy_threshold: float = 3.5, min_len: int = 24) -> None: @@ -84,23 +67,29 @@ def detect(self, text: str) -> Iterable[Finding]: # Tokens separated by non-word; allow -,_,= typical in JWT/base64url -_TOKEN = re.compile(r'([A-Za-z0-9_\-=+/]{20,})') +_TOKEN = re.compile(r"([A-Za-z0-9_\-=+/]{20,})") + class EntropyDetector: name = "entropy" - labels = ("SECRET",) def __init__(self, *, threshold: float = 3.5): self.threshold = threshold - def detect(self, text: str, *, context=None): - threshold = (context or {}).get("entropy_threshold", self.threshold) + def detect(self, text: str): for m in _TOKEN.finditer(text): token = m.group(1) if not looks_like_secret(token): continue H = shannon_entropy(token) - if H >= threshold: - yield Match("SECRET", m.start(1), m.end(1), token, min(0.99, 0.7 + (H-threshold)/4), {"entropy": H}) + if self.threshold <= H: + yield Finding( + kind=self.name, + value=token, + span=m.span(1), + confidence=min(0.99, 0.7 + (H - self.threshold) / 4), + extras={"entropy": H}, + ) + -register(EntropyDetector()) \ No newline at end of file +register(EntropyDetector()) diff --git a/src/redactable/detectors/iban.py b/src/redactable/detectors/iban.py index 720dbd6..f386bdd 100644 --- a/src/redactable/detectors/iban.py +++ b/src/redactable/detectors/iban.py @@ -1,17 +1,25 @@ import re -from .base import Match, register + +from .base import Finding, register from .utils import iban_check -_IBAN = re.compile(r'\b([A-Z]{2}\d{2}[A-Z0-9]{11,30})\b', re.I) +_IBAN = re.compile(r"\b([A-Z]{2}\d{2}[A-Z0-9]{11,30})\b", re.I) + class IBANDetector: name = "iban" - labels = ("IBAN",) - def detect(self, text: str, *, context=None): + def detect(self, text: str): for m in _IBAN.finditer(text): token = m.group(1).upper() if iban_check(token): - yield Match("IBAN", m.start(1), m.end(1), token, 0.98, {"country": token[:2]}) + yield Finding( + kind="iban", + value=token, + span=(m.start(1), m.end(1)), + confidence=0.98, + extras={"country": token[:2]}, + ) + register(IBANDetector()) diff --git a/src/redactable/detectors/nhs.py b/src/redactable/detectors/nhs.py index 9b504ed..92d0605 100644 --- a/src/redactable/detectors/nhs.py +++ b/src/redactable/detectors/nhs.py @@ -1,19 +1,26 @@ import re -from .base import Match, register + +from .base import Finding, register from .utils import nhs_check -# Accept formats: 10 digits with optional spaces -_NHS = re.compile(r'\b((?:\d\s*){10})\b') +_NHS = re.compile(r"\b((?:\d\s*){10})\b") + class NHSDetector: name = "nhs" - labels = ("NHS_NUMBER",) - def detect(self, text: str, *, context=None): + def detect(self, text: str): for m in _NHS.finditer(text): raw = m.group(1) - digits = ''.join(ch for ch in raw if ch.isdigit()) + digits = "".join(ch for ch in raw if ch.isdigit()) if nhs_check(digits): - yield Match("NHS_NUMBER", m.start(1), m.end(1), raw, 0.99, {"digits": digits}) + yield Finding( + kind="nhs", + value=raw, + span=(m.start(1), m.end(1)), + confidence=0.99, + extras={"digits": digits}, + ) + register(NHSDetector()) diff --git a/src/redactable/detectors/phone.py b/src/redactable/detectors/phone.py index 19bb18f..47e076c 100644 --- a/src/redactable/detectors/phone.py +++ b/src/redactable/detectors/phone.py @@ -1,19 +1,34 @@ from __future__ import annotations + import re -from .base import Match, register + +from .base import Finding, register # E.164 (+441234567890), simple UK patterns (07..., 01/02... with spaces) -_E164 = re.compile(r'(? Iterable[Finding]: @@ -86,6 +92,7 @@ def detect(self, text: str) -> Iterable[Finding]: extras={"luhn_valid": ok, "brand": brand}, ) + try: import phonenumbers # type: ignore except Exception: # pragma: no cover @@ -102,8 +109,10 @@ def detect(self, text: str) -> Iterable[Finding]: re.VERBOSE, ) + class PhoneDetector: """Detect phone numbers via regex + optional libphonenumber.""" + name = "phone" def __init__(self, default_region: str = "GB") -> None: @@ -114,9 +123,7 @@ def detect(self, text: str): # Preferred: use Google's libphonenumber for m in phonenumbers.PhoneNumberMatcher(text, self.default_region): num = m.number - norm = phonenumbers.format_number( - num, phonenumbers.PhoneNumberFormat.E164 - ) + norm = phonenumbers.format_number(num, phonenumbers.PhoneNumberFormat.E164) conf = 0.95 if phonenumbers.is_valid_number(num) else 0.6 extras = { "region": phonenumbers.region_code_for_number(num), @@ -147,8 +154,10 @@ def detect(self, text: str): # -------------------------------------------------------------------- # Detector stubs + class EmailDetector: """Detect email addresses via regex + optional email-validator.""" + name = "email" def detect(self, text: str) -> Iterable[Finding]: @@ -157,7 +166,7 @@ def detect(self, text: str) -> Iterable[Finding]: start, end = m.span() conf = 0.6 norm = raw - extras: Dict[str, Any] = {} + extras: dict[str, Any] = {} # If email-validator is available, upgrade confidence if validate_email is not None: try: @@ -176,6 +185,7 @@ def detect(self, text: str) -> Iterable[Finding]: extras=extras, ) + try: from stdnum import iban as std_iban # type: ignore from stdnum.gb import nhs as std_nhs # type: ignore @@ -189,10 +199,12 @@ def detect(self, text: str) -> Iterable[Finding]: RE_SSN = re.compile(r"\b(\d{3})[\s-]?(\d{2})[\s-]?(\d{4})\b") RE_IBAN = re.compile(r"\b([A-Z]{2}\d{2}[A-Z0-9]{11,30})\b", re.IGNORECASE) + # -------------------------------------------------------------------- # NHS Number class NHSNumberDetector: """Detect UK NHS numbers via regex + mod-11 check.""" + name = "nhs_number" def detect(self, text: str): @@ -227,10 +239,12 @@ def detect(self, text: str): extras={"valid": valid, "reason": reason}, ) + # -------------------------------------------------------------------- # US Social Security Number class USSSNDetector: """Detect US Social Security Numbers via regex + range validation.""" + name = "ssn_us" def detect(self, text: str): @@ -250,9 +264,13 @@ def detect(self, text: str): else: # Basic exclusions area, group, serial = d[:3], d[3:5], d[5:] - if area == "000" or area == "666" or "900" <= area <= "999": - valid = False - elif group == "00" or serial == "0000": + if ( + area == "000" + or area == "666" + or "900" <= area <= "999" + or group == "00" + or serial == "0000" + ): valid = False else: valid = True @@ -266,10 +284,12 @@ def detect(self, text: str): extras={"valid": valid, "reason": reason}, ) + # -------------------------------------------------------------------- # IBAN class IBANDetector: """Detect IBANs via regex + mod-97 validation.""" + name = "iban" def detect(self, text: str): @@ -292,8 +312,9 @@ def _mod97(s: str) -> int: num = "".join(str(ord(c) - 55) if c.isalpha() else c for c in rearr) rem = 0 for i in range(0, len(num), 9): - rem = int(str(rem) + num[i:i+9]) % 97 + rem = int(str(rem) + num[i : i + 9]) % 97 return rem + valid = country.isalpha() and canon[2:4].isdigit() and _mod97(canon) == 1 conf = 0.95 if valid else 0.5 yield Finding( @@ -303,4 +324,4 @@ def _mod97(s: str) -> int: confidence=conf, normalized=canon, extras={"valid": valid, "country": country, "reason": reason}, - ) \ No newline at end of file + ) diff --git a/src/redactable/detectors/registry.py b/src/redactable/detectors/registry.py index 50c43a9..2d017d7 100644 --- a/src/redactable/detectors/registry.py +++ b/src/redactable/detectors/registry.py @@ -11,18 +11,18 @@ """ from __future__ import annotations -from typing import List, Optional from .base import Detector, Finding +from .entropy import HighEntropyTokenDetector from .regexes import ( - EmailDetector, - PhoneDetector, CreditCardDetector, + EmailDetector, + IBANDetector, NHSNumberDetector, + PhoneDetector, USSSNDetector, - IBANDetector, ) -from .entropy import HighEntropyTokenDetector + class DetectorRegistry: """ @@ -30,8 +30,8 @@ class DetectorRegistry: that runs all registered detectors over input text. """ - def __init__(self, detectors: Optional[List[Detector]] = None) -> None: - self.detectors: List[Detector] = detectors or [] + def __init__(self, detectors: list[Detector] | None = None) -> None: + self.detectors: list[Detector] = detectors or [] @classmethod def default(cls, region: str = "GB") -> DetectorRegistry: @@ -39,15 +39,17 @@ def default(cls, region: str = "GB") -> DetectorRegistry: Return a registry preloaded with all built-in detectors. Region argument affects phone detection. """ - return cls([ - EmailDetector(), - PhoneDetector(default_region=region), - CreditCardDetector(), - NHSNumberDetector(), - USSSNDetector(), - IBANDetector(), - HighEntropyTokenDetector(), - ]) + return cls( + [ + EmailDetector(), + PhoneDetector(default_region=region), + CreditCardDetector(), + NHSNumberDetector(), + USSSNDetector(), + IBANDetector(), + HighEntropyTokenDetector(), + ] + ) def register(self, detector: Detector) -> None: """Add a detector to the registry.""" @@ -57,21 +59,23 @@ def unregister(self, name: str) -> None: """Remove detectors by name.""" self.detectors = [d for d in self.detectors if getattr(d, "name", "") != name] - def scan(self, text: str) -> List[Finding]: + def scan(self, text: str) -> list[Finding]: """ Run all detectors against a text string. Returns a list of Finding objects, sorted by start offset. """ - findings: List[Finding] = [] + findings: list[Finding] = [] for d in self.detectors: try: findings.extend(d.detect(text)) except Exception as e: # fail-safe - findings.append(Finding( - kind="error", - value=getattr(d, "name", "unknown"), - span=(0, 0), - confidence=0.0, - extras={"error": str(e)}, - )) + findings.append( + Finding( + kind="error", + value=getattr(d, "name", "unknown"), + span=(0, 0), + confidence=0.0, + extras={"error": str(e)}, + ) + ) return sorted(findings, key=lambda f: (f.span[0], f.span[1])) diff --git a/src/redactable/detectors/run.py b/src/redactable/detectors/run.py index 1d53ae8..71b7333 100644 --- a/src/redactable/detectors/run.py +++ b/src/redactable/detectors/run.py @@ -1,15 +1,23 @@ -from typing import Optional -from .base import Match, all_detectors - # Force registration even if package __init__ is bypassed # (pytest imports run_all directly in tests) -from . import email, credit_card, iban, nhs, ssn, phone, entropy, schema_hints # noqa: F401 +from . import ( # noqa: F401 + credit_card, + email, + entropy, + iban, + nhs, + phone, + schema_hints, + ssn, +) +from .base import Finding, all_detectors + -def run_all(text: str, *, context: Optional[dict] = None) -> list[Match]: - matches: list[Match] = [] +def run_all(text: str) -> list[Finding]: + matches: list[Finding] = [] for det in all_detectors(): - for m in det.detect(text, context=context): + for m in det.detect(text): if m is not None: matches.append(m) - matches.sort(key=lambda m: (m.start, m.end)) - return matches \ No newline at end of file + matches.sort(key=lambda m: (m.span[0], m.span[1])) + return matches diff --git a/src/redactable/detectors/schema_hints.py b/src/redactable/detectors/schema_hints.py index 2e1da11..2c59732 100644 --- a/src/redactable/detectors/schema_hints.py +++ b/src/redactable/detectors/schema_hints.py @@ -1,4 +1,4 @@ -from .base import Match, register +from .base import register # This detector relies on context = {"schema": {"field_name": "value", ...}} # It emits matches for fields whose names are known sensitive hints. @@ -17,6 +17,7 @@ "date_of_birth": "DATE_DOB", } + class SchemaHintDetector: name = "schema_hints" labels = tuple(set(_HINTS.values())) @@ -27,4 +28,5 @@ def detect(self, text: str, *, context=None): return yield # generator stub + register(SchemaHintDetector()) diff --git a/src/redactable/detectors/ssn.py b/src/redactable/detectors/ssn.py index be0fdde..6a3e4fe 100644 --- a/src/redactable/detectors/ssn.py +++ b/src/redactable/detectors/ssn.py @@ -1,25 +1,36 @@ from __future__ import annotations + import re -from .base import Match, register -_SSN = re.compile(r'(? bool: - d = d.replace('-', '') - if len(d) != 9: return False - if d[:3] in {"000", "666"} or d[0] == "9": return False - if d[3:5] == "00": return False - if d[5:] == "0000": return False - return True + d = d.replace("-", "") + if len(d) != 9: + return False + if d[:3] in {"000", "666"} or d[0] == "9": + return False + if d[3:5] == "00": + return False + return d[5:] != "0000" + class SSNDetector: name = "ssn" - labels = ("SSN",) - def detect(self, text: str, *, context=None): + def detect(self, text: str): for m in _SSN.finditer(text): raw = m.group(1) if _valid_ssn(raw): - yield Match("SSN", m.start(1), m.end(1), raw, 0.95) + yield Finding( + kind="ssn", + value=raw, + span=(m.start(1), m.end(1)), + confidence=0.95, + ) + register(SSNDetector()) diff --git a/src/redactable/detectors/utils.py b/src/redactable/detectors/utils.py index 1db8cf1..96d6965 100644 --- a/src/redactable/detectors/utils.py +++ b/src/redactable/detectors/utils.py @@ -1,6 +1,6 @@ import math import re -from typing import Iterable + def luhn_check(digits: str) -> bool: s = 0 @@ -15,47 +15,69 @@ def luhn_check(digits: str) -> bool: alt = not alt return s % 10 == 0 + # Basic IBAN check (ISO 13616) -_IBAN_RE = re.compile(r'[A-Z]{2}[0-9]{2}[A-Z0-9]{11,30}$') +_IBAN_RE = re.compile(r"[A-Z]{2}[0-9]{2}[A-Z0-9]{11,30}$") _IBAN_LEN = { - # Common ones; extend as needed - "GB": 22, "DE": 22, "FR": 27, "ES": 24, "IT": 27, "NL": 18, "BE": 16, "CH": 21, "IE": 22, "PL": 28 + "GB": 22, + "DE": 22, + "FR": 27, + "ES": 24, + "IT": 27, + "NL": 18, + "BE": 16, + "CH": 21, + "IE": 22, + "PL": 28, } + + def iban_check(iban: str) -> bool: - iban = iban.replace(' ', '').upper() + iban = iban.replace(" ", "").upper() country = iban[:2] - if not _IBAN_RE.match(iban): return False - if country in _IBAN_LEN and len(iban) != _IBAN_LEN[country]: return False - # move 4 chars to end, convert letters to numbers (A=10 ... Z=35) + if not _IBAN_RE.match(iban): + return False + if country in _IBAN_LEN and len(iban) != _IBAN_LEN[country]: + return False rearranged = iban[4:] + iban[:4] - digits = ''.join(str(ord(c)-55) if c.isalpha() else c for c in rearranged) - # mod 97 == 1 + digits = "".join(str(ord(c) - 55) if c.isalpha() else c for c in rearranged) remainder = 0 for ch in digits: - remainder = (remainder*10 + ord(ch)-48) % 97 + remainder = (remainder * 10 + ord(ch) - 48) % 97 return remainder == 1 + # UK NHS number check (Mod 11) def nhs_check(n: str) -> bool: - n = re.sub(r'\D+', '', n) - if len(n) != 10: return False + n = re.sub(r"\D+", "", n) + if len(n) != 10: + return False weights = list(range(10, 1, -1)) # 10..2 - total = sum(int(d)*w for d, w in zip(n[:9], weights)) + total = sum(int(d) * w for d, w in zip(n[:9], weights, strict=False)) check = 11 - (total % 11) - if check == 11: check = 0 - if check == 10: return False + if check == 11: + check = 0 + if check == 10: + return False return check == int(n[-1]) + def shannon_entropy(s: str) -> float: - if not s: return 0.0 + if not s: + return 0.0 from collections import Counter + counts = Counter(s) length = len(s) - return -sum((c/length) * math.log2(c/length) for c in counts.values()) + return -sum((c / length) * math.log2(c / length) for c in counts.values()) + + +_BASE64ISH = re.compile(r"^[A-Za-z0-9+/=_-]+$") +_HEXISH = re.compile(r"^[0-9a-fA-F]+$") + -_BASE64ISH = re.compile(r'^[A-Za-z0-9+/=_-]+$') -_HEXISH = re.compile(r'^[0-9a-fA-F]+$') def looks_like_secret(token: str) -> bool: # quick heuristic: base64/hex-ish + length - if len(token) < 20: return False + if len(token) < 20: + return False return bool(_BASE64ISH.match(token) or _HEXISH.match(token)) diff --git a/src/redactable/engine.py b/src/redactable/engine.py index dfec20c..8b983c0 100644 --- a/src/redactable/engine.py +++ b/src/redactable/engine.py @@ -1,3 +1,3 @@ """ composite detection + apply -""" \ No newline at end of file +""" diff --git a/src/redactable/hide.py b/src/redactable/hide.py index d8ff7a8..4e3cd4e 100644 --- a/src/redactable/hide.py +++ b/src/redactable/hide.py @@ -2,6 +2,6 @@ Use this as a main entrypoint? """ -from .decors import * +from .decors import * # noqa: F401, F403 -__all__ = [] \ No newline at end of file +__all__: list[str] = [] diff --git a/src/redactable/in_out/__init__.py b/src/redactable/in_out/__init__.py index e269aea..d2ee827 100644 --- a/src/redactable/in_out/__init__.py +++ b/src/redactable/in_out/__init__.py @@ -1,2 +1,10 @@ from .readers import TextFileReader -from .writers import AuditJSONLWriter, StdoutWriter, TextFileWriter, Writer \ No newline at end of file +from .writers import AuditJSONLWriter, StdoutWriter, TextFileWriter, Writer + +__all__ = [ + "TextFileReader", + "AuditJSONLWriter", + "StdoutWriter", + "TextFileWriter", + "Writer", +] diff --git a/src/redactable/in_out/base.py b/src/redactable/in_out/base.py index 6a7cf9f..1b4ca58 100644 --- a/src/redactable/in_out/base.py +++ b/src/redactable/in_out/base.py @@ -1,18 +1,24 @@ import gzip -from typing import Iterable, Protocol, Dict, Any +from collections.abc import Iterable +from typing import Any, Protocol + class Record: - def __init__(self, content: str, meta: Dict[str, Any] | None = None): + def __init__(self, content: str, meta: dict[str, Any] | None = None): self.content = content self.meta = meta or {} + class Reader(Protocol): def iter_records(self) -> Iterable[Record]: ... + class Writer(Protocol): def write_record(self, record: Record) -> None: ... def close(self) -> None: ... def _open(path: str, mode: str): - return gzip.open(path, mode) if str(path).endswith(".gz") else open(path, mode, encoding="utf-8", newline="") + if str(path).endswith(".gz"): + return gzip.open(path, mode) + return open(path, mode, encoding="utf-8", newline="") diff --git a/src/redactable/in_out/pandas_accessor.py b/src/redactable/in_out/pandas_accessor.py index e69de29..7fef70d 100644 --- a/src/redactable/in_out/pandas_accessor.py +++ b/src/redactable/in_out/pandas_accessor.py @@ -0,0 +1,44 @@ +"""Pandas DataFrame accessor for redactable.""" + +try: + import pandas as pd + from pandas.api.extensions import register_dataframe_accessor +except ImportError: + pd = None + register_dataframe_accessor = None + + +if pd is not None: + + @register_dataframe_accessor("redact") + class RedactableAccessor: + """Pandas DataFrame accessor for applying redaction policies.""" + + def __init__(self, pandas_obj): + self._obj = pandas_obj + + def apply(self, policy: str | None = None, *, region: str = "GB"): + """ + Apply a redaction policy to all string columns in the DataFrame. + + Args: + policy: Path to YAML/JSON policy file or Policy object. + region: Default region for phone parsing (e.g., "GB", "US"). + + Returns: + DataFrame with redaction applied to all string columns. + """ + from redactable import apply as redact_text + + df = self._obj.copy() + + for col in df.columns: + # Only redact string columns + if df[col].dtype == "object": + df[col] = df[col].apply( + lambda x: ( + redact_text(str(x), policy=policy, region=region) if pd.notna(x) else x + ) + ) + + return df diff --git a/src/redactable/in_out/readers.py b/src/redactable/in_out/readers.py index c9f6a11..fc9a70a 100644 --- a/src/redactable/in_out/readers.py +++ b/src/redactable/in_out/readers.py @@ -1,12 +1,15 @@ -from .base import Record, Reader, _open +from .base import Reader, Record, _open + class TextFileReader(Reader): def __init__(self, path: str, by_line: bool = True): - self.path = path; self.by_line = by_line + self.path = path + self.by_line = by_line + def iter_records(self): with _open(self.path, "rt") as f: if self.by_line: for i, line in enumerate(f, 1): yield Record(line.rstrip("\n"), {"source": self.path, "line": i}) else: - yield Record(f.read(), {"source": self.path}) \ No newline at end of file + yield Record(f.read(), {"source": self.path}) diff --git a/src/redactable/in_out/writers.py b/src/redactable/in_out/writers.py index 18503eb..10ba6a6 100644 --- a/src/redactable/in_out/writers.py +++ b/src/redactable/in_out/writers.py @@ -1,24 +1,36 @@ import json import sys -from .base import Writer, Record, _open +from .base import Record, Writer, _open class TextFileWriter(Writer): def __init__(self, path: str): - self.path = path; self._f = _open(self.path, "wt") + self.path = path + self._f = _open(self.path, "wt") + def write_record(self, record: Record) -> None: self._f.write(record.content + "\n") - def close(self): self._f.close() + + def close(self): + self._f.close() + class StdoutWriter(Writer): def write_record(self, record: Record) -> None: sys.stdout.write(record.content + "\n") - def close(self): pass + + def close(self): + pass + class AuditJSONLWriter(Writer): def __init__(self, path: str): - self._f = open(path, "w", encoding="utf-8") + self.path = path + self._f = open(path, "w", encoding="utf-8") # noqa: SIM115 + def write_event(self, event: dict) -> None: self._f.write(json.dumps(event, ensure_ascii=False) + "\n") - def close(self): self._f.close() \ No newline at end of file + + def close(self): + self._f.close() diff --git a/src/redactable/policy/__init__.py b/src/redactable/policy/__init__.py index e1aa7da..a533719 100644 --- a/src/redactable/policy/__init__.py +++ b/src/redactable/policy/__init__.py @@ -7,10 +7,8 @@ applies rules to detector findings. """ - -from .model import Policy, Rule -from .loader import load_policy from .engine import apply_policy +from .loader import load_policy +from .model import Policy, Rule - -__all__ = ["Policy", "Rule", "load_policy", "apply_policy"] \ No newline at end of file +__all__ = ["Policy", "Rule", "load_policy", "apply_policy"] diff --git a/src/redactable/policy/engine.py b/src/redactable/policy/engine.py index 899dff0..38a3844 100644 --- a/src/redactable/policy/engine.py +++ b/src/redactable/policy/engine.py @@ -1,8 +1,10 @@ # ruff: noqa: E402 -from dataclasses import dataclass import hashlib -from typing import Iterable +from collections.abc import Iterable +from dataclasses import dataclass +from typing import Literal, overload +from redactable.audit import AuditEvent, generate_audit_event from redactable.detectors import Finding from redactable.policy import Policy @@ -13,8 +15,10 @@ class _MaskCfg: keep_tail: int = 4 glyph: str = "•" + # --- local transforms (minimal v0.1; no external deps) --------------------- + def _redact(text: str, findings: Iterable[Finding], placeholder: str = "[REDACTED:{kind}]") -> str: out = text for f in sorted(findings, key=lambda x: x.span[0], reverse=True): @@ -27,7 +31,7 @@ def _mask_segment(s: str, cfg: _MaskCfg) -> str: if len(s) <= cfg.keep_head + cfg.keep_tail: return cfg.glyph * len(s) mid = cfg.glyph * (len(s) - cfg.keep_head - cfg.keep_tail) - return s[:cfg.keep_head] + mid + s[-cfg.keep_tail:] + return s[: cfg.keep_head] + mid + s[-cfg.keep_tail :] def _mask(text: str, findings: Iterable[Finding], cfg: _MaskCfg) -> str: @@ -53,7 +57,34 @@ def _tokenize(text: str, findings: Iterable[Finding], salt: str = "") -> str: # --- public API ------------------------------------------------------------- -def apply_policy(policy: Policy, findings: list[Finding], text: str) -> str: + +@overload +def apply_policy( + policy: Policy, + findings: list[Finding], + text: str, + *, + with_audit: Literal[False] = False, +) -> str: ... + + +@overload +def apply_policy( + policy: Policy, + findings: list[Finding], + text: str, + *, + with_audit: Literal[True], +) -> tuple[str, list[AuditEvent]]: ... + + +def apply_policy( + policy: Policy, + findings: list[Finding], + text: str, + *, + with_audit: bool = False, +) -> str | tuple[str, list[AuditEvent]]: """ Apply a Policy to text using previously-detected Findings. @@ -61,8 +92,19 @@ def apply_policy(policy: Policy, findings: list[Finding], text: str) -> str: - Treat `rule.field` as the detector kind (e.g. "email", "credit_card"). - Apply actions independently; rules are idempotent by design. - Apply replacements right-to-left to preserve spans. + + Args: + policy: Policy object containing rules. + findings: List of detected findings. + text: Original text to transform. + with_audit: If True, return (transformed_text, audit_events). + + Returns: + If with_audit=False: transformed text string. + If with_audit=True: tuple of (transformed_text, audit_events). """ out = text + audit_events: list[AuditEvent] = [] # Group findings by kind for quick lookup by_kind: dict[str, list[Finding]] = {} @@ -73,13 +115,27 @@ def apply_policy(policy: Policy, findings: list[Finding], text: str) -> str: targets = by_kind.get(rule.field, []) if not targets: continue + + # Store original text to detect changes + text_before = out + if rule.action == "redact": placeholder = rule.replacement or "[REDACTED:{kind}]" out = _redact(out, targets, placeholder) elif rule.action == "mask": - cfg = _MaskCfg(keep_head=rule.keep_head, keep_tail=rule.keep_tail, glyph=rule.mask_glyph) + cfg = _MaskCfg( + keep_head=rule.keep_head, keep_tail=rule.keep_tail, glyph=rule.mask_glyph + ) out = _mask(out, targets, cfg) elif rule.action == "tokenize": out = _tokenize(out, targets, salt=rule.salt) + # Generate audit events for applied transformations + if with_audit and out != text_before: + for finding in targets: + event = generate_audit_event(finding, rule, text_before, out) + audit_events.append(event) + + if with_audit: + return out, audit_events return out diff --git a/src/redactable/policy/loader.py b/src/redactable/policy/loader.py index b2a1160..ae07c6f 100644 --- a/src/redactable/policy/loader.py +++ b/src/redactable/policy/loader.py @@ -2,15 +2,16 @@ from __future__ import annotations import json +from collections.abc import Iterable, Mapping from pathlib import Path -from typing import Any, Iterable, Mapping +from typing import Any from .model import Policy try: import yaml # type: ignore except Exception: # pragma: no cover - yaml = None + yaml = None # type: ignore _RULE_ALLOWED_KEYS = { @@ -94,11 +95,8 @@ def _infer_action( return value transform_name = rule.get("transform") - transform_key: str | None = None if isinstance(transform_name, str) and transform_name.strip(): transform_key = transform_name.strip() - has_transform = transform_key is not None - if has_transform: cfg = transform_types.get(transform_key) if isinstance(cfg, Mapping): action = _guess_action_from_type(cfg.get("type")) @@ -111,7 +109,7 @@ def _infer_action( return action return None - if not has_transform and isinstance(default_action, str) and default_action.strip(): + if isinstance(default_action, str) and default_action.strip(): return default_action return None @@ -137,7 +135,9 @@ def _merge_transform_settings( value = transform.get(source_key) if isinstance(value, int) and target_key not in rule: rule[target_key] = value - glyph = transform.get("mask_glyph") or transform.get("glyph") or transform.get("replacement") + glyph = ( + transform.get("mask_glyph") or transform.get("glyph") or transform.get("replacement") + ) if isinstance(glyph, str) and glyph.strip() and "mask_glyph" not in rule: rule["mask_glyph"] = glyph elif action == "redact": @@ -206,14 +206,16 @@ def _normalize_policy_payload(data: Any, source: Path) -> dict[str, Any]: if isinstance(meta_desc, str) and meta_desc.strip(): description = meta_desc.strip() - if isinstance(data.get("name"), str) and data.get("name").strip(): - name = data["name"].strip() + name_value = data.get("name") + if isinstance(name_value, str) and name_value.strip(): + name = name_value.strip() if name is None: name = source.stem - if isinstance(data.get("description"), str) and data.get("description").strip(): - description = data["description"].strip() + desc_value = data.get("description") + if isinstance(desc_value, str) and desc_value.strip(): + description = desc_value.strip() defaults = data.get("defaults") default_action: str | None = None diff --git a/src/redactable/policy/model.py b/src/redactable/policy/model.py index 56cf9be..c5d7181 100644 --- a/src/redactable/policy/model.py +++ b/src/redactable/policy/model.py @@ -1,31 +1,36 @@ # ruff: noqa: E402 from __future__ import annotations -from typing import Literal, Optional + +from typing import Literal + from pydantic import BaseModel, Field, field_validator Action = Literal["redact", "mask", "tokenize"] _ACTION_ALIASES = { "tokenise": "tokenize", "tokenize": "tokenize", - "pseudonymise": "tokenize", # treat as tokenize for now + "pseudonymise": "tokenize", # treat as tokenize for now "pseudonymize": "tokenize", "redact": "redact", "mask": "mask", - "scrub": "redact", # scrub ≈ redact pass over text - "generalise": "mask", # placeholder until a real generalise op exists + "scrub": "redact", # scrub ≈ redact pass over text + "generalise": "mask", # placeholder until a real generalise op exists "generalize": "mask", } + + class Rule(BaseModel): """ One transformation applied to all Findings whose kind == field. Future: add 'where' filters (regex, confidence threshold, etc.). """ + id: str = Field(..., description="Rule identifier (unique within policy)") field: str = Field(..., description="Detector kind (e.g. email, credit_card, phone)") action: Action = Field(..., description="Transformation to apply") # Redact options - replacement: Optional[str] = Field( + replacement: str | None = Field( default=None, description="Placeholder for redact action, e.g. '[REDACTED:{kind}]'", ) @@ -54,11 +59,12 @@ def _normalize_field(cls, v: str) -> str: @field_validator("replacement") @classmethod - def _validate_replacement(cls, v: Optional[str]) -> Optional[str]: + def _validate_replacement(cls, v: str | None) -> str | None: if v is not None and v.strip() == "": raise ValueError("replacement cannot be empty; use None to default") return v + class Policy(BaseModel): """ A named collection of Rules. @@ -68,9 +74,10 @@ class Policy(BaseModel): description: optional human-friendly explanation. rules: ordered list; earlier rules do not block later ones (idempotent ops). """ + version: int = Field(..., ge=1, description="Policy schema version (>=1)") name: str = Field(..., min_length=1, description="Short policy name") - description: Optional[str] = Field(default=None, description="Optional description") + description: str | None = Field(default=None, description="Optional description") rules: list[Rule] = Field(default_factory=list) @field_validator("name") diff --git a/src/redactable/transforms/mask.py b/src/redactable/transforms/mask.py index bb5c5bb..9ee88ad 100644 --- a/src/redactable/transforms/mask.py +++ b/src/redactable/transforms/mask.py @@ -1,12 +1,21 @@ -from typing import Iterable +from collections.abc import Iterable + from redactable.detectors import Finding + def _mask(value: str, keep_head: int = 0, keep_tail: int = 4, glyph: str = "•") -> str: if len(value) <= keep_head + keep_tail: return glyph * len(value) return value[:keep_head] + glyph * (len(value) - keep_head - keep_tail) + value[-keep_tail:] -def mask_in_place(text: str, findings: Iterable[Finding], keep_head: int = 0, keep_tail: int = 4, glyph: str = "•") -> str: + +def mask_in_place( + text: str, + findings: Iterable[Finding], + keep_head: int = 0, + keep_tail: int = 4, + glyph: str = "•", +) -> str: out = text for f in sorted(findings, key=lambda x: x.span[0], reverse=True): s, e = f.span diff --git a/src/redactable/transforms/redact.py b/src/redactable/transforms/redact.py index 268c27e..d853b49 100644 --- a/src/redactable/transforms/redact.py +++ b/src/redactable/transforms/redact.py @@ -1,7 +1,13 @@ -from typing import Iterable +from collections.abc import Iterable + from redactable.detectors import Finding -def redact(text: str, findings: Iterable[Finding], placeholder_fmt: str = "[REDACTED:{kind}]") -> str: + +def redact( + text: str, + findings: Iterable[Finding], + placeholder_fmt: str = "[REDACTED:{kind}]", +) -> str: """ Replace spans with a placeholder; assumes findings' spans are in original coordinates. Applies from right-to-left to preserve offsets. diff --git a/src/redactable/transforms/registry.py b/src/redactable/transforms/registry.py index e69de29..2847d83 100644 --- a/src/redactable/transforms/registry.py +++ b/src/redactable/transforms/registry.py @@ -0,0 +1,38 @@ +"""Transform Registry for redaction operations. + +Note: In v0.1, transforms are applied directly via policy engine. +This registry provides extensibility for v0.2+ plugin system. +""" + +from __future__ import annotations + +from collections.abc import Callable + + +class TransformRegistry: + """ + Registry of transforms. Provides extensibility for custom transformations. + + In v0.1, this is a placeholder for the plugin system coming in v0.2. + Actual transforms (redact, mask, tokenize) are applied via policy/engine.py. + """ + + def __init__(self, transforms: dict[str, Callable] | None = None) -> None: + self.transforms: dict[str, Callable] = transforms or {} + + @classmethod + def default(cls) -> TransformRegistry: + """Return a registry preloaded with built-in transforms (v0.2).""" + return cls({}) + + def register(self, name: str, transform: Callable) -> None: + """Register a custom transformation function.""" + self.transforms[name] = transform + + def get(self, name: str) -> Callable | None: + """Get a transformation function by name.""" + return self.transforms.get(name.lower()) + + def list(self) -> list[str]: + """List all available transforms.""" + return list(self.transforms.keys()) diff --git a/src/redactable/transforms/tokenise.py b/src/redactable/transforms/tokenise.py index bb5c5bb..9ee88ad 100644 --- a/src/redactable/transforms/tokenise.py +++ b/src/redactable/transforms/tokenise.py @@ -1,12 +1,21 @@ -from typing import Iterable +from collections.abc import Iterable + from redactable.detectors import Finding + def _mask(value: str, keep_head: int = 0, keep_tail: int = 4, glyph: str = "•") -> str: if len(value) <= keep_head + keep_tail: return glyph * len(value) return value[:keep_head] + glyph * (len(value) - keep_head - keep_tail) + value[-keep_tail:] -def mask_in_place(text: str, findings: Iterable[Finding], keep_head: int = 0, keep_tail: int = 4, glyph: str = "•") -> str: + +def mask_in_place( + text: str, + findings: Iterable[Finding], + keep_head: int = 0, + keep_tail: int = 4, + glyph: str = "•", +) -> str: out = text for f in sorted(findings, key=lambda x: x.span[0], reverse=True): s, e = f.span diff --git a/tests/_quicktest.py b/tests/_quicktest.py index 4469df9..99d0dcf 100644 --- a/tests/_quicktest.py +++ b/tests/_quicktest.py @@ -1,4 +1,4 @@ from redactable import apply text = "Email alice@example.com, card 4111 1111 1111 1111" -print(apply(text, region="GB")) \ No newline at end of file +print(apply(text, region="GB")) diff --git a/tests/test_apply.py b/tests/test_apply.py index a24179f..9bbd6d0 100644 --- a/tests/test_apply.py +++ b/tests/test_apply.py @@ -1,5 +1,6 @@ from redactable import apply + def test_apply_stub_masks_example_com(): text = "Customer email: test@example.com" out = apply(text, policy="gdpr.yaml") diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..8a7af34 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,91 @@ +"""Tests for CLI functionality.""" + +import subprocess +import sys +from pathlib import Path + + +def test_cli_help(): + """Test that CLI help works.""" + result = subprocess.run( + [sys.executable, "-m", "redactable.cli", "--help"], + capture_output=True, + text=True, + ) + assert result.returncode == 0 + assert "policy" in result.stdout.lower() + + +def test_cli_version(): + """Test that CLI accepts --version flag (if implemented).""" + result = subprocess.run( + [sys.executable, "-m", "redactable.cli", "--version"], + capture_output=True, + text=True, + ) + # May fail if not implemented, but should not crash + assert result.returncode in [0, 2] # 0 = success, 2 = unrecognized arg + + +def test_cli_stdin_basic(): + """Test CLI with stdin input and basic policy.""" + policy_path = Path("policies/gdpr.yaml") + if not policy_path.exists(): + # Skip if policy file doesn't exist + return + + input_text = "Email: alice@example.com\n" + result = subprocess.run( + [sys.executable, "-m", "redactable.cli", "--policy", str(policy_path)], + input=input_text, + capture_output=True, + text=True, + ) + + assert result.returncode == 0 + # Should have some output + assert len(result.stdout) > 0 + + +def test_cli_stdin_multiple_lines(): + """Test CLI with multi-line stdin input.""" + policy_path = Path("policies/gdpr.yaml") + if not policy_path.exists(): + return + + input_text = "Email: alice@example.com\nEmail: bob@example.com\n" + result = subprocess.run( + [sys.executable, "-m", "redactable.cli", "--policy", str(policy_path)], + input=input_text, + capture_output=True, + text=True, + ) + + assert result.returncode == 0 + assert len(result.stdout) > 0 + + +def test_cli_region_argument(): + """Test CLI with --region argument.""" + policy_path = Path("policies/gdpr.yaml") + if not policy_path.exists(): + return + + input_text = "Phone: +447911123456\n" + result = subprocess.run( + [ + sys.executable, + "-m", + "redactable.cli", + "--policy", + str(policy_path), + "--region", + "US", + ], + input=input_text, + capture_output=True, + text=True, + ) + + # Should not crash with region argument + assert result.returncode in [0, 2] diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 5ba7afc..23efc5c 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -1,16 +1,19 @@ from redactable.detectors.run import run_all -from redactable.detectors.utils import nhs_check, luhn_check, iban_check +from redactable.detectors.utils import iban_check, luhn_check + def test_email(): text = "Contact: alice.smith+test@sub.example.co.uk and bad@mail" m = [x for x in run_all(text) if x.label == "EMAIL"] assert any("alice.smith+test@sub.example.co.uk" in x.value for x in m) + def test_credit_card_luhn(): # Visa test PAN: 4111 1111 1111 1111 text = "Card: 4111 1111 1111 1111" m = [x for x in run_all(text) if x.label == "CREDIT_CARD"] - assert len(m) == 1 and luhn_check('4111111111111111') + assert len(m) == 1 and luhn_check("4111111111111111") + def test_iban(): # Example GB: GB82 WEST 1234 5698 7654 32 @@ -18,24 +21,28 @@ def test_iban(): m = [x for x in run_all(text) if x.label == "IBAN"] assert len(m) == 1 and iban_check(m[0].value) + def test_nhs(): # Valid NHS example: 943 476 5919 (common example) text = "NHS No: 943 476 5919" m = [x for x in run_all(text) if x.label == "NHS_NUMBER"] assert len(m) == 1 + def test_ssn(): text = "Employee SSN 078-05-1120 and invalid 000-00-0000" m = [x for x in run_all(text) if x.label == "SSN"] assert any("078-05-1120" in x.value for x in m) assert all("000-00-0000" not in x.value for x in m) + def test_phone(): text = "Call me at +447911123456 or 07123456789." m = [x for x in run_all(text) if x.label == "PHONE"] vals = [x.value for x in m] assert "+447911123456" in vals or "07123456789" in vals + def test_entropy(): likely = "sk_live_9aGQ2d1ZbQk81Y2U5YjRjY2QxY2E5ZWFm" # base64-ish text = f"api key: {likely}" diff --git a/tests/test_pandas_integration.py b/tests/test_pandas_integration.py new file mode 100644 index 0000000..d592a5a --- /dev/null +++ b/tests/test_pandas_integration.py @@ -0,0 +1,120 @@ +"""Tests for Pandas DataFrame integration.""" + +import pytest + +try: + import pandas as pd + + HAS_PANDAS = True +except ImportError: + HAS_PANDAS = False + + +@pytest.mark.skipif(not HAS_PANDAS, reason="pandas not installed") +def test_dataframe_redact_accessor_basic(): + """Test basic DataFrame redaction using the .redact() accessor.""" + + df = pd.DataFrame( + { + "email": ["alice@example.com", "bob@example.com"], + "name": ["Alice", "Bob"], + } + ) + + redacted = df.redact(policy="policies/gdpr.yaml") + + # Should return a DataFrame + assert isinstance(redacted, pd.DataFrame) + # Original should be unchanged + assert df["email"][0] == "alice@example.com" + # Redacted version should have changed email + assert redacted["email"][0] != "alice@example.com" + + +@pytest.mark.skipif(not HAS_PANDAS, reason="pandas not installed") +def test_dataframe_redact_mixed_types(): + """Test redaction with mixed column types.""" + + df = pd.DataFrame( + { + "email": ["alice@example.com", "bob@example.com"], + "age": [25, 30], + "active": [True, False], + } + ) + + redacted = df.redact(policy="policies/gdpr.yaml") + + # Email should be redacted + assert redacted["email"][0] != "alice@example.com" + # Numbers should remain unchanged (passed through as strings, but not matched) + assert redacted["age"][0] == 25 + # Booleans should remain unchanged + assert redacted["active"][0] + + +@pytest.mark.skipif(not HAS_PANDAS, reason="pandas not installed") +def test_dataframe_redact_with_null_values(): + """Test that null values are handled gracefully.""" + + df = pd.DataFrame( + { + "email": ["alice@example.com", None, "charlie@example.com"], + } + ) + + redacted = df.redact(policy="policies/gdpr.yaml") + + # Null should remain null + assert pd.isna(redacted["email"][1]) + # Non-null values should be redacted + assert redacted["email"][0] != "alice@example.com" + + +@pytest.mark.skipif(not HAS_PANDAS, reason="pandas not installed") +def test_dataframe_redact_preserves_index(): + """Test that DataFrame redaction preserves the index.""" + + df = pd.DataFrame( + { + "email": ["alice@example.com", "bob@example.com"], + }, + index=["row_a", "row_b"], + ) + + redacted = df.redact(policy="policies/gdpr.yaml") + + # Index should be preserved + assert list(redacted.index) == ["row_a", "row_b"] + + +@pytest.mark.skipif(not HAS_PANDAS, reason="pandas not installed") +def test_dataframe_redact_credit_cards(): + """Test redaction of credit card numbers.""" + + df = pd.DataFrame( + { + "cc": ["4111111111111111", "5500000000000004"], + } + ) + + redacted = df.redact(policy="policies/pci.yaml") + + # Credit cards should be transformed + assert redacted["cc"][0] != "4111111111111111" + + +@pytest.mark.skipif(not HAS_PANDAS, reason="pandas not installed") +def test_dataframe_redact_region_parameter(): + """Test that region parameter is passed through.""" + + df = pd.DataFrame( + { + "phone": ["+447911123456", "+12025551234"], + } + ) + + redacted = df.redact(policy="policies/gdpr.yaml", region="GB") + + # Should not crash with region parameter + assert isinstance(redacted, pd.DataFrame) diff --git a/tests/test_policy_engine.py b/tests/test_policy_engine.py new file mode 100644 index 0000000..20f3f94 --- /dev/null +++ b/tests/test_policy_engine.py @@ -0,0 +1,170 @@ +"""Tests for policy engine and audit logging.""" + +from redactable.audit import AuditEvent +from redactable.detectors import Finding +from redactable.policy.engine import apply_policy +from redactable.policy.model import Policy, Rule + + +def test_apply_policy_single_rule(): + """Test applying a policy with a single rule.""" + text = "Contact: alice@example.com" + findings = [Finding(kind="email", value="alice@example.com", span=(9, 27), confidence=0.95)] + policy = Policy( + version=1, + name="test", + rules=[ + Rule( + id="email_redact", + field="email", + action="redact", + replacement="[REDACTED:EMAIL]", + ) + ], + ) + result = apply_policy(policy, findings, text) + assert result == "Contact: [REDACTED:EMAIL]" + assert "@example.com" not in result + + +def test_apply_policy_multiple_rules(): + """Test applying a policy with multiple rules.""" + text = "Email: alice@example.com and Card: 4111111111111111" + findings = [ + Finding(kind="email", value="alice@example.com", span=(7, 25), confidence=0.95), + Finding(kind="credit_card", value="4111111111111111", span=(37, 53), confidence=0.99), + ] + policy = Policy( + version=1, + name="test", + rules=[ + Rule(id="r1", field="email", action="redact"), + Rule(id="r2", field="credit_card", action="mask", keep_tail=4), + ], + ) + result = apply_policy(policy, findings, text) + assert "[REDACTED:EMAIL]" in result + assert "••••1111" in result + assert "alice@example.com" not in result + assert "4111111111111111" not in result + + +def test_apply_policy_with_mask_rule(): + """Test applying a mask rule.""" + text = "Card: 4111111111111111" + findings = [ + Finding(kind="credit_card", value="4111111111111111", span=(6, 22), confidence=0.99) + ] + policy = Policy( + version=1, + name="test", + rules=[Rule(id="r1", field="credit_card", action="mask", keep_tail=4)], + ) + result = apply_policy(policy, findings, text) + assert "••••1111" in result + assert "4111111111111111" not in result + + +def test_apply_policy_with_tokenize_rule(): + """Test applying a tokenize rule.""" + text = "Email: alice@example.com" + findings = [Finding(kind="email", value="alice@example.com", span=(7, 25), confidence=0.95)] + policy = Policy( + version=1, + name="test", + rules=[Rule(id="r1", field="email", action="tokenize")], + ) + result = apply_policy(policy, findings, text) + # Result should be a hex string (64 chars for SHA256) + token_part = result.split(": ")[1] + assert len(token_part) == 64 + assert all(c in "0123456789abcdef" for c in token_part) + + +def test_apply_policy_no_matching_findings(): + """Test that policy with no matching findings returns unchanged text.""" + text = "No sensitive data here" + findings = [] + policy = Policy( + version=1, + name="test", + rules=[Rule(id="r1", field="email", action="redact")], + ) + result = apply_policy(policy, findings, text) + assert result == text + + +def test_apply_policy_with_audit_events(): + """Test that audit events are generated when with_audit=True.""" + text = "Contact: alice@example.com" + findings = [Finding(kind="email", value="alice@example.com", span=(9, 27), confidence=0.95)] + policy = Policy( + version=1, + name="test", + rules=[Rule(id="email_rule", field="email", action="redact")], + ) + result, audit_events = apply_policy(policy, findings, text, with_audit=True) + + assert isinstance(result, str) + assert isinstance(audit_events, list) + assert len(audit_events) == 1 + + event = audit_events[0] + assert isinstance(event, AuditEvent) + assert event.field == "email" + assert event.action == "redact" + assert event.value_original == "alice@example.com" + assert "[REDACTED" in event.value_transformed + + +def test_apply_policy_audit_events_multiple(): + """Test audit events for multiple transformations.""" + text = "Email: alice@example.com and card: 4111111111111111" + findings = [ + Finding(kind="email", value="alice@example.com", span=(7, 25), confidence=0.95), + Finding(kind="credit_card", value="4111111111111111", span=(38, 54), confidence=0.99), + ] + policy = Policy( + version=1, + name="test", + rules=[ + Rule(id="r1", field="email", action="redact"), + Rule(id="r2", field="credit_card", action="mask", keep_tail=4), + ], + ) + result, audit_events = apply_policy(policy, findings, text, with_audit=True) + + assert len(audit_events) == 2 + assert audit_events[0].field == "email" + assert audit_events[1].field == "credit_card" + + +def test_apply_policy_audit_events_no_changes(): + """Test that no audit events are generated when no transformations occur.""" + text = "No sensitive data" + findings = [] + policy = Policy( + version=1, + name="test", + rules=[Rule(id="r1", field="email", action="redact")], + ) + result, audit_events = apply_policy(policy, findings, text, with_audit=True) + + assert result == text + assert audit_events == [] + + +def test_apply_policy_backward_compatibility(): + """Test that apply_policy without with_audit returns only the string.""" + text = "Contact: alice@example.com" + findings = [Finding(kind="email", value="alice@example.com", span=(9, 27), confidence=0.95)] + policy = Policy( + version=1, + name="test", + rules=[Rule(id="r1", field="email", action="redact")], + ) + result = apply_policy(policy, findings, text) + + # Should return only a string, not a tuple + assert isinstance(result, str) + assert "[REDACTED" in result diff --git a/tests/test_transforms.py b/tests/test_transforms.py new file mode 100644 index 0000000..1731940 --- /dev/null +++ b/tests/test_transforms.py @@ -0,0 +1,96 @@ +"""Tests for transformation operations (redact, mask, tokenize).""" + +from redactable.detectors import Finding +from redactable.policy.engine import _mask, _MaskCfg, _redact, _tokenize + + +def test_redact_single_finding(): + """Test redaction of a single finding.""" + text = "My email is alice@example.com" + findings = [Finding(kind="email", value="alice@example.com", span=(12, 30), confidence=0.95)] + result = _redact(text, findings, "[REDACTED:{kind}]") + assert result == "My email is [REDACTED:EMAIL]" + + +def test_redact_multiple_findings(): + """Test redaction of multiple findings.""" + text = "alice@example.com and bob@example.com" + findings = [ + Finding(kind="email", value="alice@example.com", span=(0, 16), confidence=0.95), + Finding(kind="email", value="bob@example.com", span=(21, 36), confidence=0.95), + ] + result = _redact(text, findings) + assert "[REDACTED:EMAIL]" in result + assert "@example.com" not in result + + +def test_redact_preserves_spans_right_to_left(): + """Test that redaction preserves spans by processing right-to-left.""" + text = "emails: alice@example.com and bob@corp.com" + findings = [ + Finding(kind="email", value="alice@example.com", span=(8, 25), confidence=0.95), + Finding(kind="email", value="bob@corp.com", span=(30, 42), confidence=0.95), + ] + result = _redact(text, findings) + # Both should be redacted + assert result.count("[REDACTED:EMAIL]") == 2 + + +def test_mask_single_finding(): + """Test masking of a single finding.""" + text = "Card: 4111111111111111" + findings = [ + Finding(kind="credit_card", value="4111111111111111", span=(6, 22), confidence=0.99) + ] + cfg = _MaskCfg(keep_head=0, keep_tail=4, glyph="•") + result = _mask(text, findings, cfg) + assert "••••1111" in result + assert "4111111111111111" not in result + + +def test_mask_with_keep_head(): + """Test masking with head retention.""" + text = "SSN: 078-05-1120" + findings = [Finding(kind="ssn", value="078-05-1120", span=(5, 16), confidence=0.99)] + cfg = _MaskCfg(keep_head=3, keep_tail=0, glyph="*") + result = _mask(text, findings, cfg) + assert "078" in result + assert "*" in result + + +def test_mask_custom_glyph(): + """Test masking with custom glyph.""" + text = "Phone: +447911123456" + findings = [Finding(kind="phone", value="+447911123456", span=(7, 20), confidence=0.95)] + cfg = _MaskCfg(keep_head=0, keep_tail=4, glyph="X") + result = _mask(text, findings, cfg) + assert "XXXXXX3456" in result + + +def test_tokenize_single_finding(): + """Test tokenization (SHA256 hashing) of a finding.""" + text = "Email: alice@example.com" + findings = [Finding(kind="email", value="alice@example.com", span=(7, 25), confidence=0.95)] + result = _tokenize(text, findings, salt="") + # Should be a 64-character hex string (SHA256) + token = result.split(": ")[1] + assert len(token) == 64 + assert all(c in "0123456789abcdef" for c in token) + + +def test_tokenize_deterministic(): + """Test that tokenization is deterministic with same salt.""" + text = "Email: alice@example.com" + findings = [Finding(kind="email", value="alice@example.com", span=(7, 25), confidence=0.95)] + result1 = _tokenize(text, findings, salt="test_salt") + result2 = _tokenize(text, findings, salt="test_salt") + assert result1 == result2 + + +def test_tokenize_with_salt(): + """Test that different salts produce different tokens.""" + text = "Email: alice@example.com" + findings = [Finding(kind="email", value="alice@example.com", span=(7, 25), confidence=0.95)] + result1 = _tokenize(text, findings, salt="salt1") + result2 = _tokenize(text, findings, salt="salt2") + assert result1 != result2