This document walks through how the code works. We cover the detection engine, file extraction, network analysis, and CLI integration, with code snippets from the actual project.
src/dlp_scanner/
├── __init__.py
├── cli.py # Typer entry point
├── engine.py # Scan orchestration
├── config.py # Pydantic config models
├── constants.py # Thresholds, types, defaults
├── models.py # Finding, Location, ScanResult
├── compliance.py # Rule-to-framework mapping
├── redaction.py # Snippet masking
├── log.py # structlog configuration
├── scoring.py # Shared match-to-finding conversion
├── commands/
│ ├── scan.py # file, db, network commands
│ └── report.py # convert, summary commands
├── detectors/
│ ├── base.py # DetectionRule, DetectorMatch
│ ├── pattern.py # Regex + checksum detection
│ ├── context.py # Keyword proximity scoring
│ ├── entropy.py # Shannon entropy detection
│ ├── registry.py # Central detector registry
│ └── rules/
│ ├── pii.py # SSN, email, phone, passport
│ ├── financial.py # Credit cards, IBAN, NHS
│ ├── credentials.py # AWS, GitHub, JWT, Stripe
│ └── health.py # Medical records, DEA, NPI
├── extractors/
│ ├── base.py # Extractor protocol
│ ├── plaintext.py # .txt, .log, .cfg, source code
│ ├── pdf.py # .pdf via PyMuPDF
│ ├── office.py # .docx, .xlsx, .xls
│ ├── structured.py # .csv, .json, .xml, .yaml, .parquet, .avro
│ ├── archive.py # .zip, .tar.gz, .tar.bz2
│ └── email.py # .eml, .msg
├── network/
│ ├── pcap.py # PCAP/PCAPNG packet reader
│ ├── flow_tracker.py # TCP flow reassembly
│ ├── protocols.py # DPI protocol identification
│ └── exfiltration.py # DNS exfil detection
├── reporters/
│ ├── base.py # Reporter protocol
│ ├── console.py # Rich terminal output
│ ├── json_report.py # Structured JSON
│ ├── sarif.py # SARIF 2.1.0
│ └── csv_report.py # Flat CSV
└── scanners/
├── base.py # Scanner protocol
├── file_scanner.py # Directory walking + extraction
├── db_scanner.py # DB schema introspection
└── network_scanner.py # PCAP payload scanning
Every detection rule is a data structure, not a class hierarchy. The DetectionRule dataclass holds the regex pattern, base confidence score, optional validator function, context keywords, and compliance framework tags:
@dataclass(frozen=True, slots=True)
class DetectionRule:
rule_id: str
rule_name: str
pattern: re.Pattern[str]
base_score: float
context_keywords: list[str] = field(default_factory=list)
validator: Callable[[str], bool] | None = None
compliance_frameworks: list[str] = field(
default_factory=list
)Rule modules export plain lists of these structs. Here is the SSN rule from detectors/rules/pii.py:
SSN_PATTERN = re.compile(
r"\b(?!000|666|9\d{2})\d{3}"
r"[-\s]?"
r"(?!00)\d{2}"
r"[-\s]?"
r"(?!0000)\d{4}\b"
)
PII_RULES: list[DetectionRule] = [
DetectionRule(
rule_id="PII_SSN",
rule_name="US Social Security Number",
pattern=SSN_PATTERN,
base_score=0.45,
context_keywords=SSN_CONTEXT,
validator=_validate_ssn,
compliance_frameworks=[
"HIPAA", "CCPA", "GLBA", "GDPR",
],
),
...
]The regex uses negative lookaheads ((?!000|666|9\d{2})) to reject SSN area numbers the Social Security Administration has never assigned. This is a first-pass structural filter. The real validation happens in _validate_ssn, which the PatternDetector calls for every regex match.
Why base_score is 0.45, not higher: A 9-digit number matching the SSN format appears in serial numbers, zip+4 codes, phone fragments, and test data constantly. The string 456-78-9012 matches the SSN pattern and passes area/group/serial validation, but without context it could be anything. A base of 0.45 keeps it in the "medium" severity tier until context boosts push it higher.
The three checksum validators demonstrate different mathematical approaches to the same problem: distinguishing real identifiers from random digit sequences.
Luhn algorithm for credit cards (in detectors/rules/financial.py):
def luhn_check(number: str) -> bool:
digits = [int(d) for d in number if d.isdigit()]
if len(digits) < 13:
return False
odd_digits = digits[-1::-2]
even_digits = digits[-2::-2]
total = sum(odd_digits)
for d in even_digits:
total += sum(divmod(d * 2, 10))
return total % 10 == 0The algorithm works right-to-left: take every other digit starting from the rightmost, sum them. For the remaining digits, double each, and if the result exceeds 9, subtract 9 (which is what sum(divmod(d * 2, 10)) does). If the grand total is divisible by 10, the number is valid. A random 16-digit number has about a 10% chance of passing Luhn, so it reduces false positives by roughly 90%.
Mod-97 for IBANs (ISO 7064):
def iban_check(value: str) -> bool:
cleaned = value.replace(" ", "").upper()
if len(cleaned) < 15 or len(cleaned) > 34:
return False
rearranged = cleaned[4:] + cleaned[:4]
numeric = ""
for char in rearranged:
if char.isalpha():
numeric += str(ord(char) - ord("A") + 10)
else:
numeric += char
return int(numeric) % 97 == 1Move the country code and check digits (first 4 chars) to the end, convert letters to two-digit numbers (A=10, B=11, etc.), then check that the entire number mod 97 equals 1. The false positive rate is approximately 1 in 97.
Mod-11 for NHS numbers:
def nhs_check(value: str) -> bool:
digits = value.replace("-", "").replace(" ", "")
if len(digits) != 10 or not digits.isdigit():
return False
weights = range(10, 1, -1)
total = sum(
int(d) * w
for d, w in zip(digits[:9], weights, strict=False)
)
remainder = 11 - (total % 11)
if remainder == 11:
remainder = 0
if remainder == 10:
return False
return remainder == int(digits[9])Multiply the first 9 digits by descending weights (10, 9, 8, ..., 2), sum them, compute 11 - (sum mod 11), and compare to the check digit. If the result is 10, the number is invalid (NHS never issues these). If the result is 11, the check digit is 0.
Luhn-80840 for NPIs (in detectors/rules/health.py):
def _validate_npi(value: str) -> bool:
digits = value.replace("-", "").replace(" ", "")
if len(digits) != 10 or not digits.isdigit():
return False
prefixed = "80840" + digits
total = 0
for i, d in enumerate(reversed(prefixed)):
n = int(d)
if i % 2 == 1:
n *= 2
if n > 9:
n -= 9
total += n
return total % 10 == 0NPI (National Provider Identifier) validation is a Luhn variant. The trick is prepending 80840 (the healthcare industry prefix assigned by ANSI) before running the standard Luhn algorithm. This prefix is not part of the NPI itself, but the ISO standard requires it for check digit computation. A random 10-digit number has about a 10% chance of passing, making this check useful but not definitive. The base score of 0.10 reflects that NPI patterns match many unrelated 10-digit numbers, and context keywords like "provider" or "npi" are needed to push the score into actionable territory.
The PatternDetector in detectors/pattern.py iterates over all active rules, runs each regex against the input text, filters through the allowlist, and applies checksum validation:
class PatternDetector:
def detect(self, text: str) -> list[DetectorMatch]:
matches: list[DetectorMatch] = []
for rule in self._rules:
for m in rule.pattern.finditer(text):
matched_text = m.group()
if self._is_allowlisted(matched_text):
continue
score = rule.base_score
if rule.validator is not None:
if rule.validator(matched_text):
score = min(1.0, score + CHECKSUM_BOOST)
else:
continue
matches.append(
DetectorMatch(
rule_id=rule.rule_id,
...
score=score,
)
)
return matchesWhen a rule has a validator and the match fails validation, the match is discarded entirely (continue). A Visa pattern that matches 4532015112830366 but fails Luhn is not a credit card. When validation passes, the score gets a +0.30 boost (CHECKSUM_BOOST). This is aggressive because checksum-passing matches are overwhelmingly real: the Luhn+Visa prefix combination has a false positive rate under 1%.
The allowlist uses a frozen set lookup, defaulting to KNOWN_TEST_VALUES (common test card numbers, example SSNs like 123-45-6789). This prevents DLP tools from flagging their own test data, which is a common complaint in production deployments.
After pattern detection, apply_context_boost in detectors/context.py scans the surrounding text for keywords that indicate the matched value is actually sensitive data:
def apply_context_boost(
text: str,
matches: list[DetectorMatch],
window_tokens: int = DEFAULT_CONTEXT_WINDOW_TOKENS,
) -> list[DetectorMatch]:
tokens = text.lower().split()
boosted: list[DetectorMatch] = []
for match in matches:
if not match.context_keywords:
boosted.append(match)
continue
char_to_token = _char_offset_to_token_index(
text, match.start
)
window_start = max(
0, char_to_token - window_tokens
)
window_end = min(
len(tokens), char_to_token + window_tokens
)
window_text = " ".join(
tokens[window_start:window_end]
)
boost = _compute_keyword_boost(
window_text,
match.context_keywords,
window_tokens,
)
new_score = min(1.0, match.score + boost)
...The window is bidirectional: 10 tokens in each direction from the match. The boost is distance-weighted: a keyword right next to the match contributes up to CONTEXT_BOOST_MAX (0.35), while one at the edge of the window contributes almost nothing. This reflects a real observation: "SSN: 456-78-9012" is almost certainly an SSN, while "SSN" appearing 50 words away from "456-78-9012" is weaker signal.
The _compute_keyword_boost function finds the best keyword match in the window and computes CONTEXT_BOOST_MAX * proximity_factor, where proximity is 1.0 - (distance / max_distance). Only the highest-scoring keyword matters, not the sum of all keywords. This prevents keyword stuffing from inflating scores.
After context boosting, _apply_cooccurrence_boost checks whether multiple different PII types appear near each other:
def _apply_cooccurrence_boost(
matches: list[DetectorMatch],
) -> list[DetectorMatch]:
if len(matches) < 2:
return matches
proximity_threshold = 500
for i, match in enumerate(matches):
has_neighbor = False
for j, other in enumerate(matches):
if i == j:
continue
if other.rule_id == match.rule_id:
continue
distance = abs(match.start - other.start)
if distance < proximity_threshold:
has_neighbor = True
break
if has_neighbor:
new_score = min(
1.0, match.score + COOCCURRENCE_BOOST
)
...An SSN near a credit card number is stronger evidence than either alone. The boost is +0.15 (COOCCURRENCE_BOOST), and it requires different rule_id values (two SSNs next to each other do not trigger it). The 500-character threshold roughly corresponds to a short paragraph or a few database columns.
This heuristic matters in practice. The Capital One breach data contained CSV exports where SSNs, credit card numbers, and addresses appeared in adjacent columns. Co-occurrence detection would have flagged these files as critical priority.
The EntropyDetector in detectors/entropy.py finds high-entropy regions that may contain secrets, encrypted data, or base64-encoded credentials:
def shannon_entropy(data: bytes) -> float:
if not data:
return 0.0
counts = Counter(data)
total = len(data)
return -sum(
(c / total) * math.log2(c / total)
for c in counts.values()
)Shannon entropy measures the average information content per byte. English text sits around 3.5-4.5 bits. Base64-encoded data is 5.5-6.0. Truly random bytes approach 8.0 (log2(256)). The detector uses a sliding window of 256 bytes with a 128-byte step:
def detect_high_entropy_regions(
data: bytes,
threshold: float = DEFAULT_ENTROPY_THRESHOLD,
window_size: int = WINDOW_SIZE,
step: int = WINDOW_STEP,
) -> list[tuple[int, int, float]]:
...
while i + window_size <= len(data):
window = data[i:i + window_size]
h = shannon_entropy(window)
if h >= threshold:
end = i + window_size
while end + step <= len(data):
next_window = data[
end - window_size + step:end + step
]
next_h = shannon_entropy(next_window)
if next_h < threshold:
break
h = max(h, next_h)
end += step
regions.append((i, end, h))
i = end
else:
i += stepWhen the entropy exceeds the threshold (default 7.2), the detector extends the region forward until entropy drops below the threshold. This merges adjacent high-entropy windows into a single region rather than reporting dozens of overlapping detections.
The default threshold of 7.2 is intentionally high. Network payloads containing binary protocol data or compressed content often hit 6.0-7.0, which would generate massive false positive volume. At 7.2, the detector primarily catches encrypted blobs, base64-encoded secrets, and random key material.
All extractors implement a two-method protocol:
class Extractor(Protocol):
def extract(self, path: str) -> list[TextChunk]: ...
@property
def supported_extensions(self) -> frozenset[str]: ...The FileScanner builds an extension-to-extractor map at initialization by iterating over all extractor instances and indexing by their supported extensions. When scanning a file, it looks up the extractor by the file's extension and calls extract.
The PlaintextExtractor reads files in 500-line chunks to keep memory bounded:
class PlaintextExtractor:
def extract(self, path: str) -> list[TextChunk]:
chunks: list[TextChunk] = []
with open(
path, encoding="utf-8", errors="replace",
) as f:
lines: list[str] = []
line_number = 1
chunk_start = 1
for line in f:
lines.append(line)
if len(lines) >= CHUNK_MAX_LINES:
chunks.append(
TextChunk(
text="".join(lines),
location=Location(
source_type="file",
uri=path,
line=chunk_start,
),
)
)
chunk_start = line_number + 1
lines = []
line_number += 1
if lines:
chunks.append(...)
return chunksEach TextChunk carries the starting line number in its Location, so findings can report where in the file the match occurred. The errors="replace" parameter means binary-contaminated text files (common in log files with embedded binary data) will not crash the extractor.
The _build_extension_map function in file_scanner.py constructs the mapping from extensions to extractors:
def _build_extension_map() -> dict[str, Extractor]:
extractors: list[Extractor] = [
PlaintextExtractor(),
PDFExtractor(),
DocxExtractor(),
XlsxExtractor(),
XlsExtractor(),
CsvExtractor(),
JsonExtractor(),
XmlExtractor(),
YamlExtractor(),
ParquetExtractor(),
AvroExtractor(),
ArchiveExtractor(),
EmlExtractor(),
MsgExtractor(),
]
ext_map: dict[str, Extractor] = {}
for extractor in extractors:
for ext in extractor.supported_extensions:
ext_map[ext] = extractor
return ext_mapAdding a new format means creating an extractor class with extract and supported_extensions, then adding it to this list. The scanner does not need to know anything about the format.
The FileScanner._scan_directory method applies a chain of filters before dispatching to an extractor:
def _scan_directory(self, directory, result):
iterator = (
directory.rglob("*")
if self._file_config.recursive
else directory.glob("*")
)
for path in iterator:
if not path.is_file():
continue
if self._is_excluded(path, directory):
continue
suffix = _get_full_suffix(path)
if suffix not in self._allowed_extensions:
continue
file_size = path.stat().st_size
if file_size > max_bytes:
continue
if file_size == 0:
continue
self._scan_file(path, result)
result.targets_scanned += 1The _get_full_suffix function handles compound extensions like .tar.gz and .tar.bz2 by checking the filename suffix before falling back to path.suffix.lower(). The exclusion check matches against the relative path, the filename, and individual path components, so a pattern like __pycache__ matches regardless of depth.
The NetworkScanner ties together the network modules into a multi-pass pipeline. The old implementation decoded raw packets as UTF-8 and ran detection directly. The rewrite is protocol-aware:
def _scan_pcap(self, path, result):
tracker = FlowTracker()
dns_detector = DnsExfilDetector(
entropy_threshold=(
self._net_config.dns_label_entropy_threshold
),
)
packet_count = 0
for packet in read_pcap(
path,
max_packets=self._net_config.max_packets,
):
packet_count += 1
tracker.add_packet(packet)
if (
packet.protocol == "udp"
and (
packet.src_port == DNS_PORT
or packet.dst_port == DNS_PORT
)
):
self._process_dns_packet(
packet.payload, packet.src_ip,
packet.dst_ip, path, packet_count,
dns_detector, result,
)
if packet.payload:
exfil_indicators = detect_base64_payload(
packet.payload,
src_ip=packet.src_ip,
dst_ip=packet.dst_ip,
)
for indicator in exfil_indicators:
finding = _indicator_to_finding(
indicator, str(path), packet_count,
)
result.findings.append(finding)
txt_indicators = dns_detector.check_txt_volume()
for indicator in txt_indicators:
...
self._scan_reassembled_flows(tracker, path, result)Three things happen during the packet loop: every packet goes into the FlowTracker for later TCP reassembly, UDP packets on port 53 are parsed as DNS and fed to the DnsExfilDetector, and every payload is checked for base64/hex-encoded data by detect_base64_payload. After the loop, TXT query volume ratios are checked and TCP flows are reassembled for content scanning.
The reassembled flow scanning uses protocol-aware text extraction:
def _extract_scannable_text(self, stream, protocol):
if protocol == "http":
return self._extract_http_text(stream)
if protocol in ("tls", "ssh"):
return ""
try:
return stream.decode("utf-8", errors="replace")
except Exception:
return ""HTTP flows get parsed by parse_http, which extracts URIs, sensitive headers (cookie, authorization, set-cookie), and bodies. TLS and SSH flows are skipped entirely since the content is encrypted and cannot be scanned. Everything else falls through to a UTF-8 decode attempt.
DNS exfiltration indicators and encoded payload detections are converted to Finding objects through _indicator_to_finding, which maps indicator types to rule IDs via the EXFIL_RULE_MAP lookup table. Regex-based detections from reassembled flows go through match_to_finding like the other scanners.
The read_pcap function in network/pcap.py reads packets using dpkt and yields PacketInfo structs:
def read_pcap(path, max_packets=0):
with open(path, "rb") as f:
try:
pcap = dpkt.pcap.Reader(f)
except ValueError:
f.seek(0)
pcap = dpkt.pcapng.Reader(f)
count = 0
for timestamp, buf in pcap:
if max_packets > 0 and count >= max_packets:
break
packet = _parse_ethernet(timestamp, buf)
if packet is not None:
yield packet
count += 1The try/except fallback handles both PCAP (libpcap) and PCAPNG (Wireshark's newer format). dpkt is used instead of Scapy because it is roughly 100x faster for bulk packet parsing. Scapy constructs rich protocol objects with dissection layers; dpkt does minimal parsing and gives you raw bytes.
The FlowTracker in network/flow_tracker.py groups packets into flows and reassembles TCP streams:
def make_flow_key(packet):
forward = (
packet.src_ip, packet.dst_ip,
packet.src_port, packet.dst_port,
)
reverse = (
packet.dst_ip, packet.src_ip,
packet.dst_port, packet.src_port,
)
return min(forward, reverse)The bidirectional key is the lexicographically smaller of the forward and reverse 4-tuples. This means (A->B) and (B->A) packets land in the same flow. The reassemble_stream method sorts segments by TCP sequence number and deduplicates retransmissions:
def reassemble_stream(self, key):
flow = self._flows.get(key)
if flow is None:
return b""
sorted_segments = sorted(
flow.segments, key=lambda s: s[0]
)
seen_offsets: set[int] = set()
parts: list[bytes] = []
for seq, data in sorted_segments:
if seq not in seen_offsets:
seen_offsets.add(seq)
parts.append(data)
return b"".join(parts)TCP retransmissions reuse the same sequence number, so deduplication by sequence number prevents duplicate data in the reassembled stream. This is a simplified reassembly that does not handle overlapping segments (where retransmissions contain different data), but it covers the common case.
The identify_protocol function in network/protocols.py performs Deep Packet Inspection using byte prefix matching:
def identify_protocol(payload: bytes) -> str:
if not payload:
return "unknown"
if _is_http_request(payload):
return "http"
if payload.startswith(HTTP_RESPONSE_PREFIX):
return "http"
if (len(payload) > 2
and payload[:2] == TLS_RECORD_PREFIX):
return "tls"
if payload.startswith(SSH_PREFIX):
return "ssh"
if payload.startswith(SMTP_BANNER_PREFIX):
return "smtp"
return "unknown"HTTP requests are identified by checking if the first word before a space is a known HTTP method (GET, POST, PUT, etc.). TLS records start with \x16\x03 (ContentType=Handshake + major version 3). SSH banners start with SSH-. SMTP server greetings start with 220.
This matters for DLP because the same sensitive data requires different handling depending on the transport protocol. An SSN in an HTTP body can be read and flagged with high confidence. The same SSN in a TLS-encrypted stream cannot be read, but you can flag the flow as "encrypted traffic containing unknown data" and correlate with other signals.
The DnsExfilDetector in network/exfiltration.py analyzes DNS queries for patterns that suggest data tunneling:
def _check_subdomain_entropy(self, name, src_ip, dst_ip):
parts = name.split(".")
if len(parts) < 3:
return None
subdomain = ".".join(parts[:-2])
if not subdomain:
return None
entropy = shannon_entropy_str(subdomain)
if entropy > self._entropy_threshold:
return ExfilIndicator(
indicator_type="dns_high_entropy",
description=(
f"High subdomain entropy ({entropy:.2f}) "
f"suggesting DNS tunneling"
),
confidence=min(
0.95,
0.50 + (entropy - 3.0) * 0.15,
),
source_ip=src_ip,
dst_ip=dst_ip,
evidence=name,
)Legitimate subdomains (www, mail, api, cdn) have very low entropy. A query like aGVsbG8gd29ybGQ.evil.com has subdomain entropy above 4.0 because the base64-encoded data uses most of the alphanumeric character space. The detector extracts everything before the last two domain labels (the registerable domain), computes Shannon entropy, and flags queries above the threshold.
The confidence score scales linearly from 0.50 (at entropy 3.0) to 0.95 (at entropy 6.0). This captures the observation that higher entropy means more confident detection: entropy 4.1 might be a CDN hash, but entropy 5.5 is almost certainly encoded data.
The score_to_severity function in compliance.py maps confidence scores to severity levels using a threshold table:
SEVERITY_SCORE_THRESHOLDS = [
(0.85, "critical"),
(0.65, "high"),
(0.40, "medium"),
(0.20, "low"),
]
def score_to_severity(score: float) -> Severity:
for threshold, severity in SEVERITY_SCORE_THRESHOLDS:
if score >= threshold:
return severity
return "low"The thresholds are tuned so that:
- Critical (0.85+): checksum-validated matches with context keywords (e.g., SSN near "social security")
- High (0.65+): checksum-validated matches or strong context without validation
- Medium (0.40+): pattern matches without strong validation or context
- Low (0.20+): weak matches that might be false positives
The RULE_FRAMEWORK_MAP in compliance.py is a static lookup table:
RULE_FRAMEWORK_MAP = {
"PII_SSN": ["HIPAA", "CCPA", "GLBA", "GDPR"],
"PII_DRIVERS_LICENSE_FL": ["CCPA", "HIPAA"],
"FIN_CREDIT_CARD_VISA": ["PCI_DSS", "GLBA"],
"FIN_CREDIT_CARD_MC": ["PCI_DSS", "GLBA"],
"FIN_IBAN": ["GDPR", "GLBA"],
"HEALTH_NPI": ["HIPAA"],
"NET_DNS_EXFIL_HIGH_ENTROPY": [],
...
}Rule IDs match actual detection rules rather than using generic categories. Credit card rules are split by brand (FIN_CREDIT_CARD_VISA, FIN_CREDIT_CARD_MC, FIN_CREDIT_CARD_AMEX, FIN_CREDIT_CARD_DISC), each triggering PCI-DSS and GLBA. State-specific driver's license rules (PII_DRIVERS_LICENSE_FL, PII_DRIVERS_LICENSE_IL) map to CCPA and HIPAA alongside the generic CA pattern. Network exfiltration indicators (NET_DNS_EXFIL_*, NET_ENCODED_*) carry empty framework lists since DNS tunneling is a detection concern, not a regulatory data type.
SSNs trigger four frameworks because they are considered protected health information (HIPAA), personal information (CCPA), financial identifiers (GLBA), and personal data (GDPR). Every rule also has a corresponding entry in RULE_REMEDIATION_MAP with specific guidance text. Unknown rules fall back to a generic default.
The mapping is intentionally conservative. An SSN could trigger SOX if it appears in financial reporting data, but without business context the scanner cannot determine that. The listed frameworks are the ones where the mere presence of the data type creates a compliance obligation.
The match_to_finding function in scoring.py centralizes the conversion from DetectorMatch to Finding. All three scanners import from this single location instead of duplicating the severity/compliance/redaction logic:
def match_to_finding(
match: DetectorMatch,
text: str,
location: Location,
redaction_style: RedactionStyle,
) -> Finding:
severity = score_to_severity(match.score)
frameworks = get_frameworks_for_rule(match.rule_id)
if match.compliance_frameworks:
combined = (
set(frameworks) | set(match.compliance_frameworks)
)
frameworks = sorted(combined)
remediation = get_remediation_for_rule(match.rule_id)
snippet = redact(
text, match.start, match.end,
style=redaction_style,
)
return Finding(
rule_id=match.rule_id,
rule_name=match.rule_name,
severity=severity,
confidence=match.score,
location=location,
redacted_snippet=snippet,
compliance_frameworks=frameworks,
remediation=remediation,
)The function chains severity classification, compliance framework lookup, remediation guidance, and redaction in one call. The framework merging logic handles the case where a detection rule carries its own compliance_frameworks list: those are merged with the frameworks from the compliance module, deduplicated, and sorted for deterministic output.
Each scanner calls this in its match loop:
for match in matches:
if match.score < min_confidence:
continue
finding = match_to_finding(
match, chunk.text, chunk.location,
self._redaction_style,
)
result.findings.append(finding)Adding a new compliance framework or changing severity thresholds affects all three scanners uniformly without touching scanner code.
The redact function in redaction.py builds a snippet with masked content:
def redact(text, start, end, style="partial"):
matched = text[start:end]
if style == "none":
return _build_snippet(text, start, end, matched)
if style == "full":
return _build_snippet(
text, start, end, REDACTED_LABEL
)
redacted = _partial_redact(matched)
return _build_snippet(text, start, end, redacted)The _partial_redact function applies format-aware masking:
def _partial_redact(value):
stripped = value.replace("-", "").replace(" ", "")
if len(stripped) >= 9 and stripped.isdigit():
return MASK_CHAR * (len(value) - 4) + value[-4:]
if "@" in value:
local, domain = value.rsplit("@", maxsplit=1)
masked_local = (
local[0] + MASK_CHAR * (len(local) - 1)
)
return f"{masked_local}@{domain}"
if len(value) > 8:
visible = max(4, len(value) // 4)
return (
MASK_CHAR * (len(value) - visible)
+ value[-visible:]
)
return MASK_CHAR * len(value)For digit sequences (SSNs, credit cards), it preserves the last 4 digits: ***-**-6789. For emails, it keeps the first character and domain: j****@example.com. For other strings (API keys, tokens), it shows the last 25%. Short values under 8 characters are fully masked.
The _build_snippet function adds 20 characters of context on each side and prepends/appends ... when the context is truncated. This gives analysts enough surrounding text to understand what the data was near without exposing full document contents.
The Typer callback stores global options in Click's context dict:
@app.callback()
def main(ctx: typer.Context, config: ..., verbose: ..., version: ...):
ctx.ensure_object(dict)
ctx.obj["config_path"] = config
ctx.obj["verbose"] = verboseSubcommands retrieve these via ctx.ensure_object(dict):
def _run_scan(ctx, scan_type, target, output_format, output_file):
obj: dict[str, Any] = ctx.ensure_object(dict)
config_path = obj.get("config_path", "")
verbose = obj.get("verbose", False)This pattern lets dlp-scan -v -c custom.yml file ./data propagate the verbose flag and config path to the file scan command without duplicating those options on every subcommand.
The logging level adapts to the output format:
if verbose:
configure_logging(level="DEBUG")
elif output_format == "console":
configure_logging(level="INFO")
else:
configure_logging(level="WARNING")When output is machine-readable (JSON, SARIF, CSV), logging is set to WARNING so that structlog messages written to stderr do not contaminate stdout. This prevents dlp-scan file ./data -f json | jq from breaking because log lines mixed into the JSON output. For console output, INFO-level logging provides progress feedback. Verbose mode enables DEBUG for troubleshooting.
The report convert command reads a JSON scan result and regenerates it in another format:
@report_app.command("convert")
def convert(input_file, output_format="sarif", output_file=""):
raw = path.read_bytes()
data = orjson.loads(raw)
result = _rebuild_result(data)
config = ScanConfig()
engine = ScanEngine(config)
output = engine.generate_report(result, fmt)
...The _rebuild_result function deserializes the JSON structure back into ScanResult, Finding, and Location objects. It reads from the scan_metadata section for scan-level fields and iterates findings to reconstruct each Finding with its Location. This is necessary because orjson.loads produces plain dicts, but the reporters expect typed dataclass instances.
The project uses Hypothesis for property-based testing of detection rules. Instead of testing a few known inputs, Hypothesis generates random strings constrained by rule formats and verifies that the detection pipeline handles them correctly.
For validators: Hypothesis generates random digit sequences and verifies that luhn_check, iban_check, and nhs_check only return True for inputs that satisfy the mathematical properties (divisibility by 10, mod 97 = 1, mod 11 check digit match).
For the context boost: Hypothesis generates random text with embedded keywords at varying distances and verifies that the boost is always between 0 and CONTEXT_BOOST_MAX, and that closer keywords produce higher boosts.
uv run pytest -m unit # fast unit tests
uv run pytest -m integration # tests with file I/O
uv run pytest --cov=src # coverage reportThe test suite uses markers (unit, integration, slow) to separate fast tests from those requiring real filesystem access. The conftest.py provides shared fixtures for temporary directories, sample configs, and test data files.
- typer: CLI framework with type-hint argument declaration. The
Annotatedstyle avoids decorators stacking up. - rich: Terminal tables with colors. Used by
ConsoleReporterfor severity-colored output. - structlog: Structured logging with stdlib integration. JSON or console rendering based on config.
- pydantic: Config validation. Catches invalid YAML values before the scan starts.
- orjson: Fast JSON serialization. 3-10x faster than stdlib json for large finding lists.
- ruamel.yaml: YAML parser that handles 1.2 spec and preserves comments.
- dpkt: PCAP parsing. ~100x faster than Scapy for bulk packet iteration.
- pymupdf: PDF text extraction with layout preservation.
- python-docx/openpyxl/xlrd: Office format extraction.
- defusedxml/lxml: Safe XML parsing (defusedxml blocks XXE attacks).
- pyarrow/fastavro: Columnar format extraction (Parquet, Avro).
- asyncpg/aiomysql/pymongo/aiosqlite: Async database drivers.
You have seen how the code works. Now:
- Try the challenges in 04-CHALLENGES.md for extension ideas
- Modify a detection rule and run the tests to see how the scoring changes
- Scan your own files with
dlp-scan file ./your-directoryand inspect the output