Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 40 additions & 33 deletions src/redactable/detectors/base.py
Original file line number Diff line number Diff line change
@@ -1,53 +1,70 @@
"""
Core types and helpers for detectors.
"""Core types and helpers for detectors.

Contents:
- Finding: dataclass representing a detected entity.
- Detector: Protocol interface that all detectors must implement.
- Shared helper functions: digits_only, luhn_ok, guess_card_brand.
"""
This module defines the public API surface that detectors consume:

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]
* :class:`Match` – lightweight results used by the built-in registry.
* :class:`Finding` – richer results used by the legacy registry layer.
* :class:`Detector` – the protocol detectors must implement.
* Helper utilities shared by multiple detectors.
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Iterable, Protocol, TypedDict, Optional, Dict, Any, List
from typing import Any, Iterable, Optional, Protocol

@dataclass(slots=True)
class Match:
"""A lightweight finding produced by detectors registered in this module."""

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
meta: dict[str, Any] | None = None


class Detector(Protocol):
"""Protocol that built-in detectors adhere to.

Detectors expose a :pydata:`name`, a collection of :pydata:`labels`, and a
:py:meth:`detect` method that yields :class:`Match` instances. The optional
``context`` keyword argument allows callers to pass detector-specific
configuration without breaking the common interface.
"""

name: str
labels: tuple[str, ...]
def detect(self, text: str, *, context: Optional[dict[str, Any]] = None) -> Iterable[Match]: ...

def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
...


# Simple registry
_REGISTRY: Dict[str, Detector] = {}
_LABEL_TO_DETECTORS: Dict[str, List[str]] = {}
_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]


def detectors_for(label: str) -> list[Detector]:
return [ _REGISTRY[n] for n in _LABEL_TO_DETECTORS.get(label, []) ]
return [_REGISTRY[n] for n in _LABEL_TO_DETECTORS.get(label, [])]


def all_detectors() -> list[Detector]:
return list(_REGISTRY.values())
Expand All @@ -67,10 +84,10 @@ class Finding:
"""
kind: str
value: str
span: Span
span: tuple[int, int]
confidence: float
normalized: Optional[str] = None
extras: Extras | None = None
extras: dict[str, Any] | None = None

def __post_init__(self) -> None:
if not (0.0 <= self.confidence <= 1.0):
Expand All @@ -81,16 +98,6 @@ def __post_init__(self) -> None:
def __str__(self) -> str:
return f"<Finding {self.kind} value='{self.value}' conf={self.confidence:.2f}>"


class Detector(Protocol):
"""
Protocol that all detectors must follow.
Each detector must expose a `name` and a `detect` method.
"""
name: str

def detect(self, text: str) -> Iterable[Finding]: ...

# --------------------------------------------------------------------
# Shared helpers

Expand Down
9 changes: 8 additions & 1 deletion src/redactable/detectors/credit_card.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import re
from typing import Any, Iterable, Optional

from .base import Match, register
from .utils import luhn_check

Expand All @@ -20,7 +22,12 @@ class CreditCardDetector:
name = "credit_card"
labels = ("CREDIT_CARD",)

def detect(self, text: str, *, context=None):
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
for m in _PAN.finditer(text):
raw = m.group(1)
digits = ''.join(ch for ch in raw if ch.isdigit())
Expand Down
11 changes: 9 additions & 2 deletions src/redactable/detectors/email.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import re
from .base import Match, Detector, register
from typing import Any, Iterable, Optional

from .base import Match, register

_EMAIL = re.compile(
r'(?<![A-Za-z0-9._%+-])' # left boundary
Expand All @@ -13,7 +15,12 @@ class EmailDetector:
name = "email"
labels = ("EMAIL",)

def detect(self, text: str, *, context=None):
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
for m in _EMAIL.finditer(text):
yield Match(label="EMAIL", start=m.start(1), end=m.end(1), value=m.group(1), confidence=0.95)

Expand Down
19 changes: 15 additions & 4 deletions src/redactable/detectors/entropy.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,10 @@
"""

import re
from .base import Match, register, Finding, Detector
from typing import Any, Iterable, Optional

from .base import Match, register, Finding
from .utils import shannon_entropy, looks_like_secret
from typing import Iterable
# --------------------------------------------------------------------
# Regex pattern: matches candidate secrets
BASELIKE_PATTERN = re.compile(
Expand Down Expand Up @@ -45,7 +46,12 @@ def __init__(self, entropy_threshold: float = 3.5, min_len: int = 24) -> None:
self.entropy_threshold = entropy_threshold
self.min_len = min_len

def detect(self, text: str) -> Iterable[Finding]:
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Finding]:
"""
Scan text for high-entropy sequences.
Yields Finding objects for each match.
Expand Down Expand Up @@ -76,7 +82,12 @@ class EntropyDetector:
def __init__(self, *, threshold: float = 3.5):
self.threshold = threshold

def detect(self, text: str, *, context=None):
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
threshold = (context or {}).get("entropy_threshold", self.threshold)
for m in _TOKEN.finditer(text):
token = m.group(1)
Expand Down
9 changes: 8 additions & 1 deletion src/redactable/detectors/iban.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import re
from typing import Any, Iterable, Optional

from .base import Match, register
from .utils import iban_check

Expand All @@ -8,7 +10,12 @@ class IBANDetector:
name = "iban"
labels = ("IBAN",)

def detect(self, text: str, *, context=None):
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
for m in _IBAN.finditer(text):
token = m.group(1).upper()
if iban_check(token):
Expand Down
9 changes: 8 additions & 1 deletion src/redactable/detectors/nhs.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import re
from typing import Any, Iterable, Optional

from .base import Match, register
from .utils import nhs_check

Expand All @@ -9,7 +11,12 @@ class NHSDetector:
name = "nhs"
labels = ("NHS_NUMBER",)

def detect(self, text: str, *, context=None):
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
for m in _NHS.finditer(text):
raw = m.group(1)
digits = ''.join(ch for ch in raw if ch.isdigit())
Expand Down
10 changes: 9 additions & 1 deletion src/redactable/detectors/phone.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
from __future__ import annotations

import re
from typing import Any, Iterable, Optional

from .base import Match, register

# E.164 (+441234567890), simple UK patterns (07..., 01/02... with spaces)
Expand All @@ -10,7 +13,12 @@ class PhoneDetector:
name = "phone"
labels = ("PHONE",)

def detect(self, text: str, *, context=None):
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
for m in _E164.finditer(text):
yield Match("PHONE", m.start(1), m.end(1), m.group(1), 0.9, {"format": "E164"})
for m in _UK.finditer(text):
Expand Down
7 changes: 4 additions & 3 deletions src/redactable/detectors/run.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,16 @@
from typing import Optional
from typing import Any, 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

def run_all(text: str, *, context: Optional[dict] = None) -> list[Match]:
def run_all(text: str, *, context: Optional[dict[str, Any]] = None) -> list[Match]:
matches: list[Match] = []
for det in all_detectors():
for m in det.detect(text, context=context):
if m is not None:
matches.append(m)
matches.sort(key=lambda m: (m.start, m.end))
return matches
return matches
8 changes: 7 additions & 1 deletion src/redactable/detectors/schema_hints.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import re
from collections.abc import Iterable, Mapping
from typing import Any, Optional

from .base import Match, register

Expand All @@ -27,7 +28,12 @@ class SchemaHintDetector:
labels = tuple(set(_HINTS.values()))
confidence = 0.6

def detect(self, text: str, *, context=None):
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
schema = (context or {}).get("schema")
if not schema:
return []
Expand Down
10 changes: 9 additions & 1 deletion src/redactable/detectors/ssn.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
from __future__ import annotations

import re
from typing import Any, Iterable, Optional

from .base import Match, register

_SSN = re.compile(r'(?<!\d)(\d{3}-?\d{2}-?\d{4})(?!\d)')
Expand All @@ -16,7 +19,12 @@ class SSNDetector:
name = "ssn"
labels = ("SSN",)

def detect(self, text: str, *, context=None):
def detect(
self,
text: str,
*,
context: Optional[dict[str, Any]] = None,
) -> Iterable[Match]:
for m in _SSN.finditer(text):
raw = m.group(1)
if _valid_ssn(raw):
Expand Down
Loading