Dev.to Security 🔐 Cybersecurity 👁 0 📖 28 min read

Sensitive Data Discovery in Python: Build a PII Scanner That Validates, Not Just Matches

Before you can encrypt, mask, or delete personal data, you need to know where it lives. That sounds obvious until someone asks “which of our 400 tables hold card numbers?” and the honest answer is a shrug. GDPR Article 3

Sensitive Data Discovery in Python: Build a PII Scanner That Validates, Not Just Matches

Before you can encrypt, mask, or delete personal data, you need to know where it lives. That sounds obvious until someone asks “which of our 400 tables hold card numbers?” and the honest answer is a shrug. GDPR Article 30 expects a controller to keep a record of processing that describes “the categories of personal data”, and you cannot write that record from memory.

That is the job of sensitive data discovery: scan the stores you own, find columns that hold PII, and label them. The usual first attempt is a folder of regexes. It finds everything, including order numbers, timestamps, ZIP+4 codes and SKUs, and after a week nobody reads its report.

In this post we build a small scanner in Python that does the boring parts properly:

  1. Walks a SQLite database and a folder of CSV exports.

  2. Samples each column with reservoir sampling (bounded memory, reproducible).

  3. Finds candidates with regexes, then validates them: Luhn and network prefixes for cards, ISO 7064 mod-97 for IBANs, SSA rules for SSNs, libphonenumber for phones.

  4. Uses the column name as a second signal and computes a confidence score per column.

  5. Writes a data inventory in JSON and Markdown with classification labels and masked examples.

  6. Compares the result with Microsoft Presidio on the same samples.

Versions: Python 3.13, Faker 40.44.0, phonenumbers 9.0.42, pytest 9.2.0, and for the comparison presidio-analyzer 2.2.366 with spaCy 3.8.18 and en_core_web_sm 3.8.0. All test data is synthetic (seeded Faker), and every output block below is copied from a real run.

What a column-level scanner needs

Discovery for analytics, as the PixelPlex guide describes it, is about finding patterns and insights in data. Discovery for privacy asks a narrower question about each column: “does this hold a data type that law, contracts or policy say we must protect?” Answering it per column, not per value, is the key design decision. A single cell can look like anything. Two hundred cells from the same column rarely lie.

Pipeline diagram: sources are walked, each column is sampled, candidates go through regex then validator, the column gets a score from hit ratio and name signal, and the inventory assigns a label

Stage What it does Why it matters
Walk List tables/files and columns Coverage: you only classify what you enumerate
Sample Up to 200 non-empty values per column Bounded cost per column, no full-table scans
Detect Cheap regex finds candidates Fast, high recall
Validate Checksums and numbering rules Removes look-alikes
Score Hit ratio + column-name signal A column decision, not a cell decision
Report Labels, confidence, masked examples Something a human can review and act on

Step 1. Synthetic data with traps

make_data.py builds shop.db with three tables and two CSV exports. Real PII is mixed with values that a lazy regex will misread:

Column Real content Looks like
orders.order_number 16 random digits a card number
app_events.device_serial 16 random digits a card number
orders.sku US48XKQ93821PZ7A style codes an IBAN
payments.batch_ref 000-, 666-, 9xx- codes an SSN
customers.created_at, zip, app_events.app_version dates, ZIP+4, versions phone numbers
orders.card_masked **** **** **** 1234 a card column by name only
customers.notes free text, some with an email or phone mostly harmless text
leads.contact half emails, half phones two entity types in one column

Listing: make_data.py (full) 
"""Generate a synthetic SQLite database and CSV exports for the scanner demo.

Every value comes from Faker (seeded) or random.Random (seeded). No real PII.
"""
import csv
import random
import sqlite3
from datetime import datetime
from pathlib import Path

from faker import Faker

SEED = 42
# Fixed ranges: relative ones like "-30d" would change the data every day.
START, END = datetime(2024, 1, 1), datetime(2026, 1, 1)
DATA = Path(__file__).parent / "data"


def digits(rng: random.Random, n: int) -> str:
    return "".join(rng.choice("0123456789") for _ in range(n))


def main() -> None:
    fake = Faker("en_US")
    Faker.seed(SEED)
    rng = random.Random(SEED)
    DATA.mkdir(exist_ok=True)
    (DATA / "exports").mkdir(exist_ok=True)
    db_path = DATA / "shop.db"
    db_path.unlink(missing_ok=True)

    con = sqlite3.connect(db_path)
    con.executescript(
        """
        CREATE TABLE customers (id INTEGER PRIMARY KEY, full_name TEXT, email TEXT,
            phone TEXT, ssn TEXT, zip TEXT, created_at TEXT, notes TEXT);
        CREATE TABLE orders (id INTEGER PRIMARY KEY, order_number TEXT, customer_id INTEGER,
            amount REAL, sku TEXT, card_masked TEXT, placed_at TEXT);
        CREATE TABLE payments (id INTEGER PRIMARY KEY, customer_id INTEGER, card_number TEXT,
            iban TEXT, batch_ref TEXT, company_code TEXT);
        """
    )

    customers = []
    for i in range(1, 401):
        note = fake.sentence(nb_words=10)
        roll = rng.random()
        if roll < 0.08:
            note += f" Prefers email: {fake.free_email()}"
        elif roll < 0.16:
            note += f" Call back on {fake.phone_number()}"
        customers.append((i, fake.name(), fake.email(), fake.phone_number(), fake.ssn(),
                          fake.zipcode_plus4(), fake.date_between(START, END).isoformat(), note))
    con.executemany("INSERT INTO customers VALUES (?,?,?,?,?,?,?,?)", customers)

    orders = []
    for i in range(1, 1201):
        sku = (rng.choice(["US", "DE", "FR", "GB"]) + digits(rng, 2)
               + "".join(rng.choice("ABCDEFGHJKLMNPQRSTUVWXYZ0123456789") for _ in range(12)))
        orders.append((i, digits(rng, 16), rng.randint(1, 400), round(rng.uniform(5, 900), 2),
                       sku, "**** **** **** " + digits(rng, 4),
                       fake.date_time_between(START, END).isoformat(timespec="seconds")))
    con.executemany("INSERT INTO orders VALUES (?,?,?,?,?,?,?)", orders)

    payments = []
    for i in range(1, 401):
        batch_ref = f"{rng.choice(['000', '666', str(rng.randint(900, 999))])}-{digits(rng, 2)}-{digits(rng, 4)}"
        payments.append((i, rng.randint(1, 400), fake.credit_card_number(), fake.iban(),
                         batch_ref, "CP-" + digits(rng, 6)))
    con.executemany("INSERT INTO payments VALUES (?,?,?,?,?,?)", payments)
    con.commit()
    con.close()

    with open(DATA / "exports" / "leads.csv", "w", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(["lead_id", "name", "contact", "company", "bank_acct"])
        for i in range(1, 301):
            contact = fake.company_email() if rng.random() < 0.5 else fake.phone_number()
            w.writerow([i, fake.name(), contact, fake.company(), fake.iban()])

    with open(DATA / "exports" / "app_events.csv", "w", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(["event_id", "device_serial", "app_version", "event", "ts"])
        for i in range(1, 601):
            w.writerow([fake.uuid4(), digits(rng, 16),
                        f"{rng.randint(1, 12)}.{rng.randint(0, 9)}.{rng.randint(0, 99)}.{rng.randint(1000, 9999)}",
                        rng.choice(["open", "tap", "purchase", "close"]),
                        fake.date_time_between(datetime(2025, 12, 1), END).isoformat(timespec="seconds")])

    print(f"wrote {db_path.relative_to(DATA.parent)}: customers=400 orders=1200 payments=400")
    print("wrote data/exports/leads.csv (300 rows), data/exports/app_events.csv (600 rows)")


if __name__ == "__main__":
    main()

# Output:
# wrote data/shop.db: customers=400 orders=1200 payments=400
# wrote data/exports/leads.csv (300 rows), data/exports/app_events.csv (600 rows)

Faker is seeded, but its docs warn that results are not guaranteed to be stable across patch versions, so the version is pinned to the patch level in requirements.txt.

Step 2. The “before”: regex only

The scanner has a --naive flag that keeps the regexes and turns off validators and the name signal. This is what many homegrown PII detection scripts look like:

python -m pii_scan --db data/shop.db --csv data/exports --naive

[naive] columns=31 findings=31 confirmed=23 review=8
  app_events.event_id        CREDIT_CARD  review    conf=0.01 valid=1% regex=1%
  app_events.device_serial   CREDIT_CARD  confirmed conf=1.00 valid=100% regex=100%
  app_events.device_serial   PHONE        confirmed conf=1.00 valid=100% regex=100%
  app_events.app_version     PHONE        confirmed conf=1.00 valid=100% regex=100%
  app_events.ts              PHONE        confirmed conf=1.00 valid=100% regex=100%
  leads.contact              CREDIT_CARD  review    conf=0.01 valid=0% regex=0%
  leads.contact              EMAIL        confirmed conf=0.53 valid=53% regex=53%
  leads.contact              PHONE        review    conf=0.47 valid=47% regex=47%
  leads.bank_acct            IBAN         confirmed conf=1.00 valid=100% regex=100%
  leads.bank_acct            PHONE        confirmed conf=1.00 valid=100% regex=100%
  customers.email            EMAIL        confirmed conf=1.00 valid=100% regex=100%
  customers.phone            CREDIT_CARD  review    conf=0.05 valid=5% regex=5%
  customers.phone            PHONE        confirmed conf=1.00 valid=100% regex=100%
  customers.ssn              US_SSN       confirmed conf=1.00 valid=100% regex=100%
  customers.ssn              PHONE        confirmed conf=1.00 valid=100% regex=100%
  customers.zip              PHONE        confirmed conf=1.00 valid=100% regex=100%
  customers.created_at       PHONE        confirmed conf=1.00 valid=100% regex=100%
  customers.notes            CREDIT_CARD  review    conf=0.01 valid=0% regex=0%
  customers.notes            EMAIL        review    conf=0.09 valid=9% regex=9%
  customers.notes            PHONE        review    conf=0.07 valid=8% regex=8%
  orders.order_number        CREDIT_CARD  confirmed conf=1.00 valid=100% regex=100%
  orders.order_number        PHONE        confirmed conf=1.00 valid=100% regex=100%
  orders.sku                 IBAN         confirmed conf=1.00 valid=100% regex=100%
  orders.placed_at           PHONE        confirmed conf=1.00 valid=100% regex=100%
  payments.card_number       CREDIT_CARD  confirmed conf=1.00 valid=100% regex=100%
  payments.card_number       PHONE        confirmed conf=1.00 valid=100% regex=100%
  payments.iban              IBAN         confirmed conf=1.00 valid=100% regex=100%
  payments.iban              PHONE        confirmed conf=1.00 valid=100% regex=100%
  payments.batch_ref         US_SSN       confirmed conf=1.00 valid=100% regex=100%
  payments.batch_ref         PHONE        confirmed conf=1.00 valid=100% regex=100%

23 columns “confirmed”, and only 7 of them really hold PII. The phone regex fires on timestamps, ZIP+4 codes, app versions, SSNs, IBANs and card numbers. The card regex is certain that order numbers and device serials are payment cards. orders.sku is “100% IBAN”. A report like this gets ignored, and then the one real finding in it gets ignored too.

Step 3. Validators

Each detector is a pair: a regex that finds candidates and a function that rejects look-alikes. Here is the whole module:

"""PII detectors: a candidate regex plus a validator that rejects look-alikes."""
from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Callable
import phonenumbers

# ---------- validators ----------

def luhn_ok(number: str) -> bool:
    """Luhn mod-10 check over a string of digits."""
    total = 0
    for i, ch in enumerate(reversed(number)):
        d = int(ch)
        if i % 2 == 1:
            d *= 2
            if d > 9:
                d -= 9
        total += d
    return total % 10 == 0


def card_ok(raw: str) -> bool:
    num = re.sub(r"[ -]", "", raw)
    if not num.isdigit() or not 12 <= len(num) <= 19:   # Maestro goes down to 12 digits
        return False
    if len(set(num)) == 1:            # 0000000000000000 passes Luhn, it is not a card
        return False
    # Major networks start with 2-6; 1800/2131 are 15-digit JCB prefixes that Faker generates.
    if num[0] not in "23456" and not num.startswith(("1800", "2131")):
        return False
    return luhn_ok(num)


def iban_ok(raw: str) -> bool:
    s = raw.replace(" ", "").upper()
    if not re.fullmatch(r"[A-Z]{2}\d{2}[A-Z0-9]{11,30}", s):
        return False
    rearranged = s[4:] + s[:4]
    as_int = "".join(str(int(c, 36)) for c in rearranged)   # A=10 ... Z=35
    return int(as_int) % 97 == 1


SSN_BLOCKLIST = {"078-05-1120", "219-09-9999"}   # numbers the SSA lists as misused/advertising


def ssn_ok(raw: str) -> bool:
    m = re.fullmatch(r"(\d{3})-(\d{2})-(\d{4})", raw)
    if not m:
        return False
    area, group, serial = m.groups()
    if area in ("000", "666") or area.startswith("9"):
        return False
    if group == "00" or serial == "0000":
        return False
    return raw not in SSN_BLOCKLIST


def email_ok(raw: str) -> bool:
    local, _, domain = raw.rpartition("@")
    if not local or local.startswith(".") or local.endswith(".") or ".." in local:
        return False
    labels = domain.split(".")
    if len(labels) < 2 or not labels[-1].isalpha():
        return False
    return all(lbl and not lbl.startswith("-") and not lbl.endswith("-") for lbl in labels)


def phone_matches(text: str, region: str = "US",
                  leniency=phonenumbers.Leniency.VALID) -> list[str]:
    """libphonenumber's matcher instead of a regex. VALID = real ranges, POSSIBLE = right length."""
    return [m.raw_string for m in phonenumbers.PhoneNumberMatcher(text, region, leniency=leniency)]


def phone_weight(text: str) -> float:
    if phone_matches(text):
        return 1.0
    if phone_matches(text, leniency=phonenumbers.Leniency.POSSIBLE):
        return 0.5        # right length, but not in an assigned range: half credit
    return 0.0


# ---------- detector table ----------

@dataclass(frozen=True)
class Detector:
    entity: str
    label: str                     # classification label for the inventory
    pattern: re.Pattern            # cheap candidate finder (the "naive" scanner uses only this)
    validate: Callable[[str], bool]
    name_hints: frozenset[str]     # column-name tokens that count as a signal


DETECTORS = [
    Detector("CREDIT_CARD", "Restricted",
             re.compile(r"\b(?:\d[ -]?){11,18}\d\b"), card_ok,
             frozenset({"card", "cc", "pan", "ccnum", "credit"})),
    Detector("US_SSN", "Restricted",
             re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), ssn_ok,
             frozenset({"ssn", "social", "tin", "taxid"})),
    Detector("IBAN", "Restricted",
             re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b"), iban_ok,
             frozenset({"iban", "bank", "acct", "account"})),
    Detector("EMAIL", "Confidential",
             re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"), email_ok,
             frozenset({"email", "mail", "contact"})),
    Detector("PHONE", "Confidential",
             re.compile(r"\+?\d[\d\s().-]{7,}\d"), lambda s: phone_weight(s) > 0,
             frozenset({"phone", "mobile", "tel", "cell", "msisdn", "contact"})),
]


def name_tokens(column: str) -> set[str]:
    """'customerEmail' -> {'customer', 'email'}; 'cc_num' -> {'cc', 'num'}."""
    spaced = re.sub(r"([a-z0-9])([A-Z])", r"\1 \2", column)
    return {t for t in re.split(r"[^A-Za-z0-9]+", spaced.lower()) if t}


def value_hit(det: Detector, value: str, validate: bool) -> tuple[bool, float]:
    """Return (regex_hit, validated weight 0..1) for one cell value."""
    candidates = det.pattern.findall(value)
    if not candidates:
        return False, 0.0
    if not validate:
        return True, 1.0
    if det.entity == "PHONE":
        return True, phone_weight(value)
    return True, float(any(det.validate(c) for c in candidates))

What each validator buys you:

  • Cards. The Luhn check digit (ISO/IEC 7812-1) catches all single-digit typos, but a random number passes it one time in ten. That is what the test test_luhn_accepts_one_in_ten_random_numbers pins down. A network prefix check (first digit 2 to 6) halves the survivors again. Note the bounds: payment card numbers can be 8 to 19 digits, Maestro goes down to 12, and the Mastercard 2221–2720 range has been live since October 2016.
  • IBAN. Move the first four characters to the end, turn letters into numbers (A=10 … Z=35), and the result mod 97 must be 1. A random string passes about once in 97.
  • SSN. The SSA has not issued area numbers 000, 666 or 900–999, and no group of digits can be all zeros. Two famous sample numbers, 078-05-1120 from a 1938 wallet insert and 219-09-9999 from a 1940 pamphlet, go on a blocklist.
  • Phone. Instead of a regex, phonenumbers.PhoneNumberMatcher (a port of Google’s libphonenumber) scans the text. Its default leniency VALID accepts only numbers in assigned ranges. POSSIBLE only checks that the length is right, so it gets half credit.

Column names are split into tokens (customerEmail becomes customer, email) and compared as whole words.

explain_fp.py shows what each layer removes on the trap columns:

CREDIT_CARD: regex -> Luhn -> Luhn + network prefix
  orders.order_number        regex 200/200 (100%)  luhn  22/200 ( 11%)  +prefix  13/200 (  6%)
  app_events.device_serial   regex 200/200 (100%)  luhn  19/200 ( 10%)  +prefix  10/200 (  5%)
  payments.card_number       regex 200/200 (100%)  luhn 200/200 (100%)  +prefix 200/200 (100%)

IBAN: regex -> ISO 7064 mod-97
  orders.sku                 regex 200/200 (100%)  mod-97   0/200 (  0%)
  payments.iban              regex 200/200 (100%)  mod-97 200/200 (100%)

US_SSN: regex -> area/group/serial rules
  payments.batch_ref         regex 200/200 (100%)  rules   0/200 (  0%)
  customers.ssn              regex 200/200 (100%)  rules 200/200 (100%)

PHONE: regex -> libphonenumber VALID
  customers.created_at       regex 200/200 (100%)  valid   0/200 (  0%)
  customers.zip              regex 200/200 (100%)  valid   0/200 (  0%)
  app_events.app_version     regex 200/200 (100%)  valid   0/200 (  0%)
  orders.order_number        regex 200/200 (100%)  valid   0/200 (  0%)
  customers.phone            regex 200/200 (100%)  valid  86/200 ( 43%)

Look at orders.order_number. Even after Luhn and the prefix check, 13 of 200 order numbers are “valid card numbers”. That is not a bug in the validator, it is arithmetic: 10% pass Luhn by chance, and about half of those start with 2 to 6. No per-value rule can fix this. The fix is to decide at the column level.

Funnel bars for orders.order_number versus payments.card_number: 200 regex hits, then 22 vs 200 Luhn passes, then 13 vs 200 after the prefix check, ending in a dropped column versus a confirmed one

Step 4. Walk and sample SQLite and CSV 

"""Walk data sources and yield a bounded, reproducible sample per column."""
from __future__ import annotations

import csv
import random
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable, Iterator


@dataclass
class ColumnSample:
    source: str       # "sqlite:shop.db" or "csv:exports/leads.csv"
    table: str
    column: str
    values: list[str]  # non-null, stringified
    rows_seen: int


def reservoir(values: Iterable, k: int, rng: random.Random) -> tuple[list[str], int]:
    """Algorithm R: uniform sample of k non-empty values in one pass, O(k) memory."""
    sample: list[str] = []
    n = 0
    for v in values:
        if v is None or str(v).strip() == "":
            continue
        n += 1
        if len(sample) < k:
            sample.append(str(v))
        else:
            j = rng.randrange(n)
            if j < k:
                sample[j] = str(v)
    return sample, n


def scan_sqlite(path: Path, k: int, seed: int) -> Iterator[ColumnSample]:
    con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)   # read-only: a scanner never writes
    try:
        tables = [
            r[0]
            for r in con.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
            )
        ]
        for table in tables:
            cols = [r[1] for r in con.execute(f'PRAGMA table_info("{table}")')]
            for col in cols:
                rng = random.Random(f"{seed}:{table}:{col}")
                cur = con.execute(f'SELECT "{col}" FROM "{table}"')
                values, n = reservoir((r[0] for r in cur), k, rng)
                yield ColumnSample(f"sqlite:{path.name}", table, col, values, n)
    finally:
        con.close()


def scan_csv_dir(root: Path, k: int, seed: int) -> Iterator[ColumnSample]:
    for path in sorted(root.rglob("*.csv")):
        with open(path, newline="", encoding="utf-8", errors="replace") as fh:
            reader = csv.DictReader(fh)
            columns = reader.fieldnames or []
            rows = list(reader)          # fine for exports; stream per column for multi-GB files
        rel = path.relative_to(root.parent)
        for col in columns:
            rng = random.Random(f"{seed}:{rel}:{col}")
            values, n = reservoir((r.get(col) for r in rows), k, rng)
            yield ColumnSample(f"csv:{rel}", path.stem, col, values, n)

Three decisions here:

  • Read-only connection. file:...?mode=ro with uri=True makes SQLite open the database read-only. A scanner that runs against production copies should not be able to write, even by accident.
  • Reservoir sampling (Algorithm R). One pass, $k$ values in memory, every value has the same chance of being picked, and you do not need to know the row count first. LIMIT 200 would give you the oldest 200 rows, which on a table where a column was repurposed last year tells you nothing about today.
  • Per-column seeds. random.Random(f"{seed}:{table}:{col}") makes each run reproducible, so two scans of unchanged data produce identical inventories and diffs show only real changes.

Step 5. Score columns, not cells 

"""Score every (column, entity) pair and decide what goes into the inventory."""
from __future__ import annotations

import statistics
from dataclasses import asdict, dataclass

from .detectors import DETECTORS, Detector, name_tokens, value_hit
from .sources import ColumnSample

CONFIRM_AT = 0.6     # confidence needed for status "confirmed"
REVIEW_AT = 0.2      # below this a finding is dropped (unless the column is free text)
MIN_TEXT_HITS = 3    # free-text columns: this many validated hits -> "review"


@dataclass
class Finding:
    source: str
    table: str
    column: str
    entity: str
    label: str
    status: str            # confirmed | review
    confidence: float
    hit_ratio: float       # validated hits / sampled values
    regex_ratio: float     # raw regex hits / sampled values (what a naive scanner sees)
    sampled: int
    rows_seen: int
    name_signal: bool
    free_text: bool
    examples: list[str]    # masked, never raw values


def is_free_text(values: list[str]) -> bool:
    if not values:
        return False
    return statistics.median(len(v) for v in values) > 30 and \
        sum(" " in v for v in values) / len(values) > 0.5


def mask(value: str) -> str:
    if "@" in value:
        local, _, domain = value.partition("@")
        return f"{local[:1]}***@{domain}"
    keep = 4 if len(value) > 8 else 2
    return "*" * (len(value) - keep) + value[-keep:]


def score(det: Detector, col: ColumnSample, validate: bool) -> Finding | None:
    n = len(col.values)
    if n == 0:
        return None
    regex_hits, valid_hits = 0, 0.0
    examples: list[str] = []
    for v in col.values:
        r, ok = value_hit(det, v, validate)
        regex_hits += r
        valid_hits += ok
        if ok and len(examples) < 2:
            examples.append(mask(v) if len(v) < 40 else mask(v[-24:]))
    name_signal = validate and bool(name_tokens(col.column) & det.name_hints)
    hit_ratio = valid_hits / n
    # 80% of the weight on what the data says, 20% on what the column is called.
    confidence = round(min(1.0, 0.8 * hit_ratio + 0.2 * name_signal), 2) if validate else round(hit_ratio, 2)
    free_text = is_free_text(col.values)
    if not validate:
        status = "confirmed" if hit_ratio >= 0.5 else ("review" if regex_hits else None)
    elif confidence >= CONFIRM_AT:
        status = "confirmed"
    elif confidence >= REVIEW_AT or (free_text and valid_hits >= MIN_TEXT_HITS):
        status = "review"
    else:
        status = None
    if status is None:
        return None
    return Finding(col.source, col.table, col.column, det.entity, det.label, status,
                   confidence, round(hit_ratio, 3), round(regex_hits / n, 3), n, col.rows_seen,
                   name_signal, free_text, examples)


def scan_columns(columns, validate: bool = True) -> list[Finding]:
    findings = []
    for col in columns:
        for det in DETECTORS:
            f = score(det, col, validate)
            if f:
                findings.append(f)
    return findings


def to_dicts(findings: list[Finding]) -> list[dict]:
    return [asdict(f) for f in findings]

The confidence formula is deliberately simple:

confidence = 0.8 * validated_hit_ratio + 0.2 * column_name_signal
  • 0.6 or more is confirmed, 0.2 to 0.6 is review, below that the finding is dropped.
  • A name alone gives 0.2, which lands in review. That is how orders.card_masked shows up: the name says card, the values are masked, and a person should confirm the masking is real.
  • A column where half the values are valid cards stays in review if it is called ref, and becomes confirmed if it is called cc_num. The test test_column_name_breaks_the_tie checks exactly this.
  • Free text is different. If the median value is longer than 30 characters and most values contain spaces, three or more validated hits are enough for review, because PII in notes is sparse by nature. Without the “free text” condition, the 13 Luhn-valid order numbers would have triggered this rule too.

Step 6. The inventory report 

"""CLI: python -m pii_scan --db data/shop.db --csv data/exports [--naive]"""
from __future__ import annotations

import argparse
import itertools
from pathlib import Path

from .report import build_inventory, to_json, to_markdown
from .scanner import scan_columns, to_dicts
from .sources import scan_csv_dir, scan_sqlite


def main() -> None:
    ap = argparse.ArgumentParser(description="Sensitive data discovery for SQLite + CSV")
    ap.add_argument("--db", type=Path, action="append", default=[])
    ap.add_argument("--csv", type=Path, action="append", default=[])
    ap.add_argument("--sample", type=int, default=200, help="values sampled per column")
    ap.add_argument("--seed", type=int, default=7)
    ap.add_argument("--naive", action="store_true", help="regex only: no validators, no name signal")
    ap.add_argument("--out", type=Path, default=Path("out"))
    args = ap.parse_args()

    columns = list(itertools.chain(
        *(scan_sqlite(p, args.sample, args.seed) for p in args.db),
        *(scan_csv_dir(p, args.sample, args.seed) for p in args.csv)))
    findings = to_dicts(scan_columns(columns, validate=not args.naive))
    mode = "naive" if args.naive else "validated"
    inv = build_inventory(findings, len(columns), mode)
    args.out.mkdir(exist_ok=True)
    (args.out / f"inventory_{mode}.json").write_text(to_json(inv))
    (args.out / f"inventory_{mode}.md").write_text(to_markdown(inv))

    print(f"[{mode}] columns={len(columns)} findings={inv['findings']} "
          f"confirmed={inv['confirmed']} review={inv['review']}")
    for t in inv["tables"]:
        for c in t["columns"]:
            print(f"  {t['table'] + '.' + c['column']:<26} {c['entity']:<12} {c['status']:<9} "
                  f"conf={c['confidence']:.2f} valid={c['hit_ratio']:.0%} regex={c['regex_ratio']:.0%}")


if __name__ == "__main__":
    main()

Listing: pii_scan/report.py (full) 
"""Turn findings into a data inventory: JSON for machines, Markdown for humans."""
from __future__ import annotations

import json
from collections import defaultdict

LABEL_RANK = {"Public": 0, "Internal": 1, "Confidential": 2, "Restricted": 3}


def build_inventory(findings: list[dict], columns_scanned: int, mode: str) -> dict:
    tables: dict[tuple[str, str], dict] = defaultdict(lambda: {"label": "Internal", "columns": []})
    for f in findings:
        t = tables[(f["source"], f["table"])]
        t["columns"].append(f)
        # A table inherits the most sensitive label among its confirmed columns.
        if f["status"] == "confirmed" and LABEL_RANK[f["label"]] > LABEL_RANK[t["label"]]:
            t["label"] = f["label"]
    return {
        "mode": mode,
        "columns_scanned": columns_scanned,
        "findings": len(findings),
        "confirmed": sum(f["status"] == "confirmed" for f in findings),
        "review": sum(f["status"] == "review" for f in findings),
        "tables": [{"source": s, "table": t, **v} for (s, t), v in sorted(tables.items())],
    }


def to_json(inv: dict) -> str:
    return json.dumps(inv, indent=2)


def to_markdown(inv: dict) -> str:
    out = [
        f"# Sensitive data inventory ({inv['mode']})",
        "",
        f"Columns scanned: {inv['columns_scanned']} | findings: {inv['findings']} "
        f"(confirmed {inv['confirmed']}, review {inv['review']})",
        "",
    ]
    for t in inv["tables"]:
        out += [
            f"## {t['source']} / {t['table']} - {t['label']}",
            "",
            "| column | entity | status | confidence | valid hits | regex hits | name signal | example |",
            "|---|---|---|---|---|---|---|---|",
        ]
        for c in sorted(t["columns"], key=lambda c: -c["confidence"]):
            out.append(
                f"| {c['column']} | {c['entity']} | {c['status']} | {c['confidence']:.2f} | "
                f"{c['hit_ratio']:.0%} | {c['regex_ratio']:.0%} | {'yes' if c['name_signal'] else 'no'} | "
                f"{(c['examples'] or [''])[0]} |"
            )
        out.append("")
    return "\n".join(out)

The “after” run:

python -m pii_scan --db data/shop.db --csv data/exports

[validated] columns=31 findings=11 confirmed=7 review=4
  leads.contact              EMAIL        confirmed conf=0.62 valid=53% regex=53%
  leads.contact              PHONE        review    conf=0.46 valid=32% regex=47%
  leads.bank_acct            IBAN         confirmed conf=1.00 valid=100% regex=100%
  customers.email            EMAIL        confirmed conf=1.00 valid=100% regex=100%
  customers.phone            PHONE        confirmed conf=0.69 valid=62% regex=100%
  customers.ssn              US_SSN       confirmed conf=1.00 valid=100% regex=100%
  customers.notes            EMAIL        review    conf=0.07 valid=9% regex=9%
  customers.notes            PHONE        review    conf=0.04 valid=4% regex=8%
  orders.card_masked         CREDIT_CARD  review    conf=0.20 valid=0% regex=0%
  payments.card_number       CREDIT_CARD  confirmed conf=1.00 valid=100% regex=100%
  payments.iban              IBAN         confirmed conf=1.00 valid=100% regex=100%

From 23 confirmed columns to 7, and the 7 are exactly the columns with real PII. All 16 false confirmations are gone, and nothing real was lost: the leads.contact phone finding moved to review because only a third of that mixed column are valid phone numbers, while the same column is confirmed as EMAIL.

Before and after grid: every scanned column that had a finding, with its naive status next to its validated status

Each table gets the highest label among its confirmed columns (Restricted for cards, SSNs and IBANs, Confidential for emails and phones, Internal otherwise). A slice of the JSON inventory:

{
  "source": "sqlite:shop.db",
  "table": "payments",
  "label": "Restricted",
  "columns": [
    {
      "source": "sqlite:shop.db",
      "table": "payments",
      "column": "card_number",
      "entity": "CREDIT_CARD",
      "label": "Restricted",
      "status": "confirmed",
      "confidence": 1.0,
      "hit_ratio": 1.0,
      "regex_ratio": 1.0,
      "sampled": 200,
      "rows_seen": 400,
      "name_signal": true,
      "free_text": false,
      "examples": [
        "************0134",
        "************1528"
      ]
    }
  ]
}

The examples field is masked on purpose. A discovery report that contains raw card numbers is itself a place where card numbers live, and it will get pasted into tickets and chat. The test test_report_never_contains_raw_values regenerates the data, runs the CLI, and asserts that no SSN from the database appears in the JSON. Emails keep their domain (h***@example.net), which is a trade-off: useful for spotting corporate versus personal addresses, but still a small leak. Drop it if your policy says so.

Step 7. Tests

Listing: tests/test_scanner.py (full) 
import json
import random
import sqlite3
import subprocess
import sys
from pathlib import Path

from pii_scan.detectors import luhn_ok
from pii_scan.scanner import scan_columns
from pii_scan.sources import ColumnSample

ROOT = Path(__file__).resolve().parents[1]


def col(name, values, table="t"):
    return ColumnSample("test", table, name, values, len(values))


def luhn_valid_order_numbers(n, seed=1):
    """Order numbers that would pass a card check on their own: Luhn-valid, starting with 4."""
    rng, out = random.Random(seed), []
    while len(out) < n:
        body = "4" + "".join(rng.choice("0123456789") for _ in range(14))
        out.append(next(body + d for d in "0123456789" if luhn_ok(body + d)))
    return out


def test_order_numbers_are_not_cards_after_validation():
    rng = random.Random(0)
    values = ["".join(rng.choice("0123456789") for _ in range(16)) for _ in range(200)]
    naive = scan_columns([col("order_number", values)], validate=False)
    smart = scan_columns([col("order_number", values)], validate=True)
    assert any(f.entity == "CREDIT_CARD" and f.status == "confirmed" for f in naive)
    assert not any(f.entity == "CREDIT_CARD" for f in smart)


def test_column_name_breaks_the_tie():
    # Half the values are card-valid: "review" alone, "confirmed" with a card-ish column name
    values = luhn_valid_order_numbers(100) + ["n/a"] * 100
    anon = scan_columns([col("ref", values)])[0]
    named = scan_columns([col("cc_num", values)])[0]
    assert (anon.status, named.status) == ("review", "confirmed")


def test_free_text_with_sparse_emails_goes_to_review():
    values = [f"Customer asked about delivery window number {i}" for i in range(195)]
    values += [f"Please reply to buyer{i}@example.com about the refund" for i in range(5)]
    f = scan_columns([col("notes", values)])[0]
    assert (f.entity, f.status, f.free_text) == ("EMAIL", "review", True)


def test_report_never_contains_raw_values(tmp_path):
    subprocess.run([sys.executable, "make_data.py"], cwd=ROOT, check=True, capture_output=True)
    subprocess.run(
        [sys.executable, "-m", "pii_scan", "--db", "data/shop.db", "--csv", "data/exports", "--out", str(tmp_path)],
        cwd=ROOT,
        check=True,
        capture_output=True,
    )
    report = (tmp_path / "inventory_validated.json").read_text()
    inv = json.loads(report)
    ssns = [r[0] for r in sqlite3.connect(ROOT / "data/shop.db").execute("SELECT ssn FROM customers")]
    assert not any(s in report for s in ssns)
    labels = {(t["table"], t["label"]) for t in inv["tables"]}
    assert ("payments", "Restricted") in labels and ("orders", "Internal") in labels


python -m pytest -q

27 passed in 1.88s

tests/test_detectors.py (in the repo archive) covers the validators: a known Visa test number, a one-digit-off variant, the GB82 WEST 1234 5698 7654 32 example IBAN with good and bad check digits, every excluded SSN rule, and a phone in an assigned versus an unassigned range.

Step 8. Compare with Microsoft Presidio PII detection

Presidio is the obvious open-source alternative for PII detection. Its analyzer combines regexes, checksums, context words and spaCy NER, and its docs list the default model as en_core_web_lg (382 MB for version 3.8.0). For card, SSN, IBAN, email and phone the documented detection methods are pattern matching, checksums and context, so the small 12 MB en_core_web_sm model is enough, and Presidio installs fine without the large one.

Presidio is built for free text. To make it look at columns, the script below analyzes every sampled value twice: once alone, once with the column name tokens passed through analyze(context=...), which is what the Presidio context tutorial suggests for metadata such as column names.

"""Run Microsoft Presidio's analyzer over the same column samples and compare per-column hit ratios."""
import itertools
import logging
import time
from collections import Counter
from pathlib import Path

from presidio_analyzer import AnalyzerEngine
from presidio_analyzer.nlp_engine import NlpEngineProvider

from pii_scan.detectors import name_tokens
from pii_scan.sources import scan_csv_dir, scan_sqlite

# Presidio's email recognizer uses tldextract, which tries to download the Public Suffix List
# on first use. Offline it logs a traceback and falls back to its bundled snapshot.
logging.getLogger("tldextract").setLevel(logging.CRITICAL)

ENTITIES = ["CREDIT_CARD", "US_SSN", "IBAN_CODE", "EMAIL_ADDRESS", "PHONE_NUMBER"]
THRESHOLD = 0.5

# en_core_web_sm (12 MB) instead of the default en_core_web_lg; we only need pattern recognizers.
nlp = NlpEngineProvider(nlp_configuration={
    "nlp_engine_name": "spacy",
    "models": [{"lang_code": "en", "model_name": "en_core_web_sm"}]
}).create_engine()

analyzer = AnalyzerEngine(nlp_engine=nlp, supported_languages=["en"])

cols = list(itertools.chain(scan_sqlite(Path("data/shop.db"), 200, 7),
                            scan_csv_dir(Path("data/exports"), 200, 7)))

start = time.perf_counter()
values = 0

print(f"{'column':<26} {'no context (>=0.5)':<28} {'column name as context (>=0.5)'}")

for col in cols:
    plain, ctx = Counter(), Counter()
    for v in col.values:
        values += 1
        artifacts = nlp.process_text(v, "en")       # run spaCy once, reuse for both passes
        for counter, context in ((plain, None), (ctx, sorted(name_tokens(col.column)))):
            res = analyzer.analyze(text=v, language="en", entities=ENTITIES, context=context,
                                   nlp_artifacts=artifacts)
            for ent in {r.entity_type for r in res if r.score >= THRESHOLD}:
                counter[ent] += 1
    if plain or ctx:
        share = lambda n: f"{n / len(col.values):.0%}" if n / len(col.values) >= 0.01 else "<1%"
        fmt = lambda c: ", ".join(f"{e} {share(n)}" for e, n in c.most_common(2)) or "-"
        print(f"{col.table + '.' + col.column:<26} {fmt(plain):<28} {fmt(ctx)}")

elapsed = time.perf_counter() - start
print(f"\nanalyzed {values} values x 2 passes in {elapsed:.1f}s")

column                     no context (>=0.5)           column name as context (>=0.5)
customers.email            EMAIL_ADDRESS 100%           EMAIL_ADDRESS 100%
customers.phone            -                            PHONE_NUMBER 82%
customers.ssn              US_SSN 100%                  US_SSN 100%
customers.notes            EMAIL_ADDRESS 9%, PHONE_NUMBER <1% EMAIL_ADDRESS 9%, PHONE_NUMBER <1%
orders.order_number        CREDIT_CARD 6%               PHONE_NUMBER 8%, CREDIT_CARD 6%
payments.card_number       CREDIT_CARD 75%              CREDIT_CARD 75%, PHONE_NUMBER 3%
payments.iban              IBAN_CODE 100%               IBAN_CODE 100%
payments.batch_ref         US_SSN 29%                   US_SSN 29%
app_events.device_serial   CREDIT_CARD 5%               CREDIT_CARD 5%
leads.contact              EMAIL_ADDRESS 53%            EMAIL_ADDRESS 53%
leads.bank_acct            IBAN_CODE 100%               IBAN_CODE 100%

analyzed 6200 values x 2 passes in 45.5s

The run time above is from a shared 2-vCPU sandbox and varied between runs, so treat it as a rough number. Our scanner processes the same 31 columns in under two seconds, including interpreter start-up, because it never calls an NLP pipeline. What the numbers say:

  • Context matters a lot for phones. Presidio’s phone recognizer scores matches at 0.4, below the 0.5 cut-off, so customers.phone shows nothing. With the column name as context, the score is boosted (the default boost factor is 0.35) and 82% of values pass. That 82% is higher than our 62% because Presidio scores possible-but-unassigned numbers too, the same thing our half credit does.
  • Context can also hurt. number is one of the phone recognizer’s context words, so order_number picks up an 8% PHONE_NUMBER signal it did not have without context. Context words are blunt.
  • Card coverage is narrower. Presidio’s card regex in this version does not match Mastercard numbers starting with 2, 19-digit Visa, 12-digit Maestro, or the 15-digit 2131 numbers Faker emits as JCB. Faker produces all of these, which is why payments.card_number is 75% instead of 100%.
  • SSN rules differ. Presidio rejects 000 and 666 areas but accepts 900–999, so batch_ref shows 29% US_SSN. Numbers in 900–999 are the ITIN range, which is sensitive too, so if you have ITINs, detect them as a separate entity instead of calling them SSNs.
  • Per-value hits only. Like our naive scan, Presidio still finds 6% “credit cards” in order numbers. Turning that into “this column is not a card column” is your job, or the job of the separate presidio-structured package, whose default strategy picks the most common entity per column.

The two approaches combine well: use validators and column scoring for structured columns, and Presidio (with a context list built from column names) for long free-text fields where NER for names and addresses pays off.

Side-by-side table of per-column hit ratios: this scanner versus Presidio without and with column-name context

Gotchas I hit while building this

  • My first card validator was wrong. It required 13 to 19 digits and a first digit of 2 to 6. Across all 400 synthetic cards, the regex missed 23 (12-digit Maestro numbers) and the validator rejected 9 more (15-digit numbers starting with 1800, which Faker generates as JCB). Bounds copied from memory are a bug; check them against a reference and test with a generator that covers every network. One Faker Maestro prefix, 0604, is still rejected because I could not confirm it in a reference, so it stays out on purpose.
  • Faker phone numbers are mostly not valid numbers. Only 86 of 200 sampled customers.phone values pass libphonenumber’s VALID check, because Faker picks area codes at random. Strict validation on synthetic data looks like poor recall. Real data behaves better, but legacy fields with typos exist, which is why POSSIBLE matches get half credit instead of zero.
  • Substring matching on column names. "pan" in "company_code" is True. Tokenize names first.
  • Mixed columns. leads.contact holds two entity types. Score every detector against every column instead of stopping at the first match.
  • Air-gapped scans. Presidio’s email recognizer uses tldextract, which fetches the Public Suffix List on first use. In my sandbox without internet access that produced a long traceback in the log, after which detection carried on with the snapshot bundled in the package. For production, tldextract.TLDExtract(suffix_list_urls=()) disables the fetch.

Taking it to production

  • Sampling budget. 200 values per column is enough to separate “100% cards” from “6% cards”. For huge tables, sample in the database (Postgres TABLESAMPLE, BigQuery TABLESAMPLE SYSTEM) instead of streaming every row to Python.
  • Run it on a schedule and diff. Seeded sampling gives stable output, so a nightly job can compare today’s inventory with yesterday’s and alert on new Restricted columns.
  • Push labels to the catalog. The JSON inventory maps directly to tags in a data catalog or to column comments, so access policies and masking rules can key off them.
  • Add entities that matter to you: national IDs for your markets, IP addresses, dates of birth next to names. Each new detector should come with a validator and a test that includes look-alikes.
  • Keep humans in the loop. review exists so that a person decides the borderline cases, and that decision should be stored so the next scan does not ask again.

Checklist:

  • [ ] Every data store is listed as a source
  • [ ] Each detector has a validator, not just a regex
  • [ ] Decisions are per column, with a hit ratio and a name signal
  • [ ] Free text has its own rule
  • [ ] Reports never contain raw values
  • [ ] Tests include look-alikes, not only true positives
  • [ ] Library versions are pinned (Faker, phonenumbers metadata, Presidio)

Wrapping up

Automated data discovery and classification does not need a platform to get started. A few hundred lines of Python, honest validators and a column-level score turned 23 noisy “confirmed” columns into the 7 that really hold PII, with a report that is safe to share. Presidio is a good companion, especially for free text, as long as you give it column context and check its recognizers against the formats you actually store.

What does your team use for PII discovery in structured data: a commercial scanner, Presidio, or something homegrown? And which false positive hurt you the most?

📰 Read the original article on Dev.to Security

Originally published by Dev.to Security. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.

Share 𝕏 Twitter ✈ Telegram ↗ Reddit 🟧 HN
← News 🔐 Cybersecurity 🏠 Home