958 lines
37 KiB
Markdown
958 lines
37 KiB
Markdown
# 03-IMPLEMENTATION.md
|
|
|
|
# Implementation Guide
|
|
|
|
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.
|
|
|
|
## File Structure
|
|
|
|
```
|
|
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
|
|
```
|
|
|
|
## Building the Detection Engine
|
|
|
|
### Detection Rules
|
|
|
|
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:
|
|
|
|
```python
|
|
@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`:
|
|
|
|
```python
|
|
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.
|
|
|
|
### Checksum Validation
|
|
|
|
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`):
|
|
|
|
```python
|
|
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 == 0
|
|
```
|
|
|
|
The 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):
|
|
|
|
```python
|
|
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 == 1
|
|
```
|
|
|
|
Move 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:
|
|
|
|
```python
|
|
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`):
|
|
|
|
```python
|
|
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 == 0
|
|
```
|
|
|
|
NPI (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.
|
|
|
|
### Pattern Detection
|
|
|
|
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:
|
|
|
|
```python
|
|
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 matches
|
|
```
|
|
|
|
When 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.
|
|
|
|
### Context Keyword Scoring
|
|
|
|
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:
|
|
|
|
```python
|
|
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.
|
|
|
|
### Co-occurrence Boost
|
|
|
|
After context boosting, `_apply_cooccurrence_boost` checks whether multiple different PII types appear near each other:
|
|
|
|
```python
|
|
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.
|
|
|
|
### Shannon Entropy Detection
|
|
|
|
The `EntropyDetector` in `detectors/entropy.py` finds high-entropy regions that may contain secrets, encrypted data, or base64-encoded credentials:
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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 += step
|
|
```
|
|
|
|
When 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.
|
|
|
|
## File Extraction Pipeline
|
|
|
|
### The Extractor Protocol
|
|
|
|
All extractors implement a two-method protocol:
|
|
|
|
```python
|
|
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`.
|
|
|
|
### Plaintext Extraction
|
|
|
|
The `PlaintextExtractor` reads files in 500-line chunks to keep memory bounded:
|
|
|
|
```python
|
|
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 chunks
|
|
```
|
|
|
|
Each `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.
|
|
|
|
### Extension Map Construction
|
|
|
|
The `_build_extension_map` function in `file_scanner.py` constructs the mapping from extensions to extractors:
|
|
|
|
```python
|
|
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_map
|
|
```
|
|
|
|
Adding 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.
|
|
|
|
### File Scanner Walk Logic
|
|
|
|
The `FileScanner._scan_directory` method applies a chain of filters before dispatching to an extractor:
|
|
|
|
```python
|
|
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 += 1
|
|
```
|
|
|
|
The `_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.
|
|
|
|
## Network Analysis
|
|
|
|
### Scanner Orchestration
|
|
|
|
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:
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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.
|
|
|
|
### PCAP Parsing
|
|
|
|
The `read_pcap` function in `network/pcap.py` reads packets using dpkt and yields `PacketInfo` structs:
|
|
|
|
```python
|
|
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 += 1
|
|
```
|
|
|
|
The 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.
|
|
|
|
### TCP Flow Reassembly
|
|
|
|
The `FlowTracker` in `network/flow_tracker.py` groups packets into flows and reassembles TCP streams:
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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.
|
|
|
|
### Protocol Identification
|
|
|
|
The `identify_protocol` function in `network/protocols.py` performs Deep Packet Inspection using byte prefix matching:
|
|
|
|
```python
|
|
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.
|
|
|
|
### DNS Exfiltration Detection
|
|
|
|
The `DnsExfilDetector` in `network/exfiltration.py` analyzes DNS queries for patterns that suggest data tunneling:
|
|
|
|
```python
|
|
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.
|
|
|
|
## Compliance and Severity Classification
|
|
|
|
### Severity Mapping
|
|
|
|
The `score_to_severity` function in `compliance.py` maps confidence scores to severity levels using a threshold table:
|
|
|
|
```python
|
|
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
|
|
|
|
### Framework Mapping
|
|
|
|
The `RULE_FRAMEWORK_MAP` in `compliance.py` is a static lookup table:
|
|
|
|
```python
|
|
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.
|
|
|
|
## Shared Scoring Module
|
|
|
|
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:
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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.
|
|
|
|
## Redaction
|
|
|
|
The `redact` function in `redaction.py` builds a snippet with masked content:
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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.
|
|
|
|
## CLI Integration
|
|
|
|
### Global Option Propagation
|
|
|
|
The Typer callback stores global options in Click's context dict:
|
|
|
|
```python
|
|
@app.callback()
|
|
def main(ctx: typer.Context, config: ..., verbose: ..., version: ...):
|
|
ctx.ensure_object(dict)
|
|
ctx.obj["config_path"] = config
|
|
ctx.obj["verbose"] = verbose
|
|
```
|
|
|
|
Subcommands retrieve these via `ctx.ensure_object(dict)`:
|
|
|
|
```python
|
|
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.
|
|
|
|
### Logging Strategy
|
|
|
|
The logging level adapts to the output format:
|
|
|
|
```python
|
|
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.
|
|
|
|
### Report Conversion
|
|
|
|
The `report convert` command reads a JSON scan result and regenerates it in another format:
|
|
|
|
```python
|
|
@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.
|
|
|
|
## Testing Strategy
|
|
|
|
### Property-Based Testing
|
|
|
|
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.
|
|
|
|
### Running Tests
|
|
|
|
```bash
|
|
uv run pytest -m unit # fast unit tests
|
|
uv run pytest -m integration # tests with file I/O
|
|
uv run pytest --cov=src # coverage report
|
|
```
|
|
|
|
The 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.
|
|
|
|
## Dependencies
|
|
|
|
- **typer**: CLI framework with type-hint argument declaration. The `Annotated` style avoids decorators stacking up.
|
|
- **rich**: Terminal tables with colors. Used by `ConsoleReporter` for 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.
|
|
|
|
## Next Steps
|
|
|
|
You have seen how the code works. Now:
|
|
1. Try the challenges in [04-CHALLENGES.md](./04-CHALLENGES.md) for extension ideas
|
|
2. Modify a detection rule and run the tests to see how the scoring changes
|
|
3. Scan your own files with `dlp-scan file ./your-directory` and inspect the output
|