|
1
|
""" |
|
2
|
Sensitive content detection and redaction. |
|
3
|
|
|
4
|
Scans source text for credentials, API keys, private keys, connection strings, |
|
5
|
and other high-value secrets before they are persisted in the graph. |
|
6
|
""" |
|
7
|
|
|
8
|
import re |
|
9
|
from dataclasses import dataclass |
|
10
|
from pathlib import Path |
|
11
|
|
|
12
|
# --------------------------------------------------------------------------- |
|
13
|
# Data model |
|
14
|
# --------------------------------------------------------------------------- |
|
15
|
|
|
16
|
REDACTED = "[REDACTED]" |
|
17
|
|
|
18
|
|
|
19
|
@dataclass |
|
20
|
class SensitiveMatch: |
|
21
|
"""A single sensitive-content finding.""" |
|
22
|
|
|
23
|
pattern_name: str |
|
24
|
line_number: int |
|
25
|
match_text: str # the matched text — stored already-redacted |
|
26
|
severity: str # "high" or "medium" |
|
27
|
|
|
28
|
|
|
29
|
# --------------------------------------------------------------------------- |
|
30
|
# Pattern registry |
|
31
|
# --------------------------------------------------------------------------- |
|
32
|
# Each entry: (name, compiled-regex, severity) |
|
33
|
# The regex must capture the full sensitive token (group 0 is what's replaced). |
|
34
|
|
|
35
|
_PATTERNS: list[tuple[str, re.Pattern[str], str]] = [ |
|
36
|
# AWS access key IDs — 20-char uppercase alphanumeric starting with AKIA/ASIA/AROA |
|
37
|
( |
|
38
|
"aws_access_key", |
|
39
|
re.compile(r"(?<![A-Z0-9])(AKIA|ASIA|AROA)[A-Z0-9]{16}(?![A-Z0-9])"), |
|
40
|
"high", |
|
41
|
), |
|
42
|
# AWS secret access key — 40-char base64-ish value that typically follows "aws_secret" |
|
43
|
( |
|
44
|
"aws_secret_key", |
|
45
|
re.compile( |
|
46
|
r'(?i)aws[_\-\s]*secret[_\-\s]*(?:access[_\-\s]*)?key\s*[=:]\s*["\']?([A-Za-z0-9/+]{40})["\']?' |
|
47
|
), |
|
48
|
"high", |
|
49
|
), |
|
50
|
# GitHub personal access tokens (classic ghp_ and fine-grained github_pat_) |
|
51
|
( |
|
52
|
"github_token", |
|
53
|
re.compile(r"(ghp_[A-Za-z0-9]{36,}|github_pat_[A-Za-z0-9_]{80,})"), |
|
54
|
"high", |
|
55
|
), |
|
56
|
# Generic "sk-" prefixed keys (OpenAI, Anthropic, Stripe, etc.) |
|
57
|
( |
|
58
|
"api_key_sk", |
|
59
|
re.compile(r"\bsk-[A-Za-z0-9\-_]{20,}"), |
|
60
|
"high", |
|
61
|
), |
|
62
|
# Generic API key / token assignment pattern |
|
63
|
( |
|
64
|
"api_key_assignment", |
|
65
|
re.compile( |
|
66
|
r'(?i)(?:api[_\-]?key|api[_\-]?token|access[_\-]?token|auth[_\-]?token)\s*[=:]\s*["\']([A-Za-z0-9\-_\.]{16,})["\']' |
|
67
|
), |
|
68
|
"high", |
|
69
|
), |
|
70
|
# Password in assignment |
|
71
|
( |
|
72
|
"password_assignment", |
|
73
|
re.compile(r'(?i)(?:password|passwd|secret)\s*[=:]\s*["\']([^"\']{4,})["\']'), |
|
74
|
"high", |
|
75
|
), |
|
76
|
# PEM private key header |
|
77
|
( |
|
78
|
"private_key_pem", |
|
79
|
re.compile(r"-----BEGIN (?:RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----"), |
|
80
|
"high", |
|
81
|
), |
|
82
|
# PostgreSQL / MySQL / MongoDB / Redis connection strings with credentials |
|
83
|
( |
|
84
|
"connection_string", |
|
85
|
re.compile( |
|
86
|
r'(?i)(postgres(?:ql)?|mysql|mongodb(?:\+srv)?|redis|amqp)://[^:@\s"\']+:[^@\s"\'@]+@[^\s"\']{4,}' |
|
87
|
), |
|
88
|
"high", |
|
89
|
), |
|
90
|
# JWT — three base64url segments separated by dots (ey… header is a strong signal) |
|
91
|
( |
|
92
|
"jwt_token", |
|
93
|
re.compile(r"eyJ[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+"), |
|
94
|
"medium", |
|
95
|
), |
|
96
|
] |
|
97
|
|
|
98
|
|
|
99
|
# --------------------------------------------------------------------------- |
|
100
|
# Detector |
|
101
|
# --------------------------------------------------------------------------- |
|
102
|
|
|
103
|
|
|
104
|
class SensitiveContentDetector: |
|
105
|
""" |
|
106
|
Detect and redact sensitive content (credentials, tokens, keys) in source text. |
|
107
|
|
|
108
|
Usage:: |
|
109
|
|
|
110
|
detector = SensitiveContentDetector() |
|
111
|
matches = detector.scan_content(text) |
|
112
|
clean = detector.redact(text) |
|
113
|
""" |
|
114
|
|
|
115
|
def scan_content(self, text: str) -> list[SensitiveMatch]: |
|
116
|
""" |
|
117
|
Scan *text* for sensitive patterns. |
|
118
|
|
|
119
|
Returns a list of :class:`SensitiveMatch` objects, one per finding. |
|
120
|
The ``match_text`` field contains the matched string already rendered as |
|
121
|
``[REDACTED]`` so callers never need to touch the raw secret. |
|
122
|
""" |
|
123
|
findings: list[SensitiveMatch] = [] |
|
124
|
|
|
125
|
for pattern_name, regex, severity in _PATTERNS: |
|
126
|
for m in regex.finditer(text): |
|
127
|
# Determine line number (1-based) by counting newlines before match start |
|
128
|
line_number = text.count("\n", 0, m.start()) + 1 |
|
129
|
findings.append( |
|
130
|
SensitiveMatch( |
|
131
|
pattern_name=pattern_name, |
|
132
|
line_number=line_number, |
|
133
|
match_text=REDACTED, |
|
134
|
severity=severity, |
|
135
|
) |
|
136
|
) |
|
137
|
|
|
138
|
# Stable, deterministic order: by line then pattern name |
|
139
|
findings.sort(key=lambda f: (f.line_number, f.pattern_name)) |
|
140
|
return findings |
|
141
|
|
|
142
|
def redact(self, text: str) -> str: |
|
143
|
""" |
|
144
|
Return a copy of *text* with all sensitive patterns replaced by ``[REDACTED]``. |
|
145
|
""" |
|
146
|
result = text |
|
147
|
for _pattern_name, regex, _severity in _PATTERNS: |
|
148
|
result = regex.sub(REDACTED, result) |
|
149
|
return result |
|
150
|
|
|
151
|
def scan_file(self, path: Path) -> list[SensitiveMatch]: |
|
152
|
""" |
|
153
|
Convenience wrapper: read *path* and run :meth:`scan_content`. |
|
154
|
|
|
155
|
Returns an empty list if the file cannot be read. |
|
156
|
""" |
|
157
|
try: |
|
158
|
text = path.read_text(encoding="utf-8", errors="replace") |
|
159
|
except OSError: |
|
160
|
return [] |
|
161
|
return self.scan_content(text) |
|
162
|
|