nopii

 77@dataclass
 78class Policy:
 79    """
 80    Main policy configuration for PII detection and transformation.
 81
 82    Attributes:
 83        name: Policy name for identification
 84        version: Policy schema version
 85        locale_packs: List of locale packs to enable
 86        default_action: Default transformation action
 87        thresholds: Detection and coverage thresholds
 88        reporting: Reporting configuration
 89        secrets: Secret management configuration
 90        rules: List of transformation rules
 91        exceptions: List of policy exceptions
 92        policy_hash: SHA-256 hash of policy content for audit trails
 93    """
 94
 95    name: str = "default_policy"
 96    version: str = "1"
 97    description: Optional[str] = None
 98    locale_packs: List[str] = field(default_factory=lambda: ["generic"])
 99    default_action: str = "mask"
100    thresholds: Dict[str, Any] = field(default_factory=dict)
101    reporting: Dict[str, Any] = field(default_factory=dict)
102    secrets: Dict[str, str] = field(default_factory=dict)
103    rules: List[Rule] = field(default_factory=list)
104    exceptions: List[PolicyException] = field(default_factory=list)
105    policy_hash: Optional[str] = None
106
107    def __post_init__(self) -> None:
108        """Set default values and compute policy hash."""
109        # Set default thresholds
110        default_thresholds = {
111            "min_confidence": 0.65,
112            "fail_on_untransform": False,
113            "coverage_target": 0.85,
114        }
115        for k1, v1 in default_thresholds.items():
116            self.thresholds.setdefault(k1, v1)
117
118        # Set default reporting config
119        default_reporting = {
120            "formats": ["json"],
121            "output_dir": "reports",
122            "store_samples": 3,
123            "include_trends": False,
124        }
125        for k2, v2 in default_reporting.items():
126            self.reporting.setdefault(k2, v2)
127
128        # Set default secrets config
129        default_secrets = {
130            "tokenization_key_env": "REDACT_PII_KEY",
131            "namespace_env": "REDACT_PII_NS",
132        }
133        for k3, v3 in default_secrets.items():
134            self.secrets.setdefault(k3, v3)
135
136        # Compute policy hash if not provided
137        if self.policy_hash is None:
138            self.policy_hash = self._compute_hash()
139
140    def _compute_hash(self) -> str:
141        """Compute SHA-256 hash of policy content for audit trails."""
142        # Create a deterministic string representation
143        content = f"{self.version}|{sorted(self.locale_packs)}|{self.default_action}"
144        content += f"|{sorted(self.thresholds.items())}"
145        content += f"|{len(self.rules)}|{len(self.exceptions)}"
146
147        return hashlib.sha256(content.encode()).hexdigest()[:16]
148
149    def get_rule_for_column(self, column: str) -> Optional[Rule]:
150        """Get the most specific rule for a column."""
151        # First check for column-specific rules
152        for rule in self.rules:
153            if rule.columns and column in rule.columns:
154                return rule
155        return None
156
157    def get_rule_for_type(self, pii_type: str) -> Optional[Rule]:
158        """Get the rule for a specific PII type."""
159        for rule in self.rules:
160            if rule.match == pii_type:
161                return rule
162        return None
163
164    def is_allowed(self, dataset: str, pii_type: str) -> bool:
165        """Check if a PII type is allowed in a specific dataset."""
166        for exception in self.exceptions:
167            if exception.dataset == dataset and pii_type in exception.allow_types:
168                return True
169        return False
170
171    def model_dump(self, exclude: Optional[set] = None) -> Dict[str, Any]:
172        """Convert policy to dictionary for serialization."""
173        exclude = exclude or set()
174
175        result = {}
176        for field_name in [
177            "name",
178            "version",
179            "description",
180            "locale_packs",
181            "default_action",
182            "thresholds",
183            "reporting",
184            "secrets",
185            "rules",
186            "exceptions",
187            "policy_hash",
188        ]:
189            if field_name not in exclude:
190                value = getattr(self, field_name)
191                if field_name == "rules":
192                    result[field_name] = [self._rule_to_dict(rule) for rule in value]
193                elif field_name == "exceptions":
194                    result[field_name] = [self._exception_to_dict(exc) for exc in value]
195                else:
196                    result[field_name] = value
197
198        return result
199
200    def _rule_to_dict(self, rule: Rule) -> Dict[str, Any]:
201        """Convert Rule to dictionary."""
202        out: Dict[str, Any] = {
203            "action": rule.action,
204            "options": rule.options,
205            "override_confidence": rule.override_confidence,
206        }
207
208        # Only include non-None values for match and columns
209        if rule.match is not None:
210            out["match"] = rule.match
211        if rule.columns is not None:
212            out["columns"] = list(rule.columns)
213
214        return out
215
216    def _exception_to_dict(self, exception: PolicyException) -> Dict[str, Any]:
217        """Convert Exception to dictionary."""
218        return {
219            "dataset": exception.dataset,
220            "allow_types": exception.allow_types,
221            "conditions": exception.conditions,
222        }

26@dataclass
27class Rule:
28    """
29    A policy rule that defines how to handle specific PII types or columns.
30
31    Attributes:
32        match: PII type to match (e.g., 'email', 'phone')
33        columns: Specific column names to apply this rule to
34        action: Transformation action ('mask', 'hash', 'tokenize', 'redact', 'nullify')
35        options: Action-specific configuration options
36        override_confidence: Override the detector's confidence score
37    """
38
39    action: str = "mask"
40    match: Optional[str] = None
41    columns: Optional[List[str]] = None
42    options: Dict[str, Any] = field(default_factory=dict)
43    override_confidence: Optional[float] = None
44
45    def __post_init__(self) -> None:
46        """Validate rule configuration."""
47        if not self.match and not self.columns:
48            raise ValueError("Rule must specify either 'match' or 'columns'")
49
50        valid_actions = {"mask", "hash", "tokenize", "redact", "nullify"}
51        if self.action not in valid_actions:
52            raise ValueError(
53                f"Invalid action '{self.action}'. Must be one of {valid_actions}"
54            )
55
56        if self.override_confidence is not None:
57            if not 0.0 <= self.override_confidence <= 1.0:
58                raise ValueError("override_confidence must be between 0.0 and 1.0")

225@dataclass
226class Finding:
227    """
228    A detected PII instance with metadata.
229
230    Attributes:
231        type: Type of PII detected (e.g., 'email', 'phone')
232        value: The detected PII value
233        span: Start and end positions in text (for string data)
234        column: Column name where PII was found
235        row_index: Row index where PII was found
236        confidence: Confidence score (0.0 to 1.0)
237        evidence: Explanation of why this was detected as PII
238        transformed_value: Value after transformation (if applicable)
239        action_taken: Transformation action applied
240    """
241
242    type: str
243    value: str
244    span: Tuple[int, int]
245    column: str
246    row_index: int
247    confidence: float
248    evidence: str
249    transformed_value: Optional[str] = None
250    action_taken: Optional[str] = None
251
252    def __post_init__(self) -> None:
253        """Validate finding data."""
254        if not 0.0 <= self.confidence <= 1.0:
255            raise ValueError("Confidence must be between 0.0 and 1.0")
256
257        if self.span[0] > self.span[1]:
258            raise ValueError("Invalid span: start must be <= end")

261@dataclass
262class ScanResult:
263    """
264    Results from scanning data for PII.
265
266    Attributes:
267        findings: List of detected PII instances
268        coverage_score: Percentage of PII properly handled (0.0 to 1.0)
269        scan_metadata: Additional metadata about the scan
270        policy_hash: Hash of the policy used for scanning
271        timestamp: When the scan was performed
272        dataset_name: Name of the scanned dataset
273        total_rows: Total number of rows scanned
274        total_columns: Total number of columns scanned
275    """
276
277    findings: List[Finding]
278    coverage_score: float
279    scan_metadata: Dict[str, Any]
280    policy_hash: str
281    timestamp: datetime
282    dataset_name: str = "unknown"
283    total_rows: int = 0
284    total_columns: int = 0
285
286    def __post_init__(self) -> None:
287        """Validate scan result data."""
288        if not 0.0 <= self.coverage_score <= 1.0:
289            raise ValueError("Coverage score must be between 0.0 and 1.0")
290
291    def get_findings_by_type(self) -> Dict[str, List[Finding]]:
292        """Group findings by PII type."""
293        by_type: Dict[str, List[Finding]] = {}
294        for finding in self.findings:
295            by_type.setdefault(finding.type, []).append(finding)
296        return by_type
297
298    def get_findings_by_column(self) -> Dict[str, List[Finding]]:
299        """Group findings by column."""
300        by_column: Dict[str, List[Finding]] = {}
301        for finding in self.findings:
302            by_column.setdefault(finding.column, []).append(finding)
303        return by_column
304
305    def get_summary_stats(self) -> Dict[str, int]:
306        """Get summary statistics about the findings."""
307        by_type = self.get_findings_by_type()
308        return {
309            "total_findings": len(self.findings),
310            "unique_types": len(by_type),
311            "affected_columns": len(self.get_findings_by_column()),
312            "high_confidence": len([f for f in self.findings if f.confidence >= 0.8]),
313            "medium_confidence": len(
314                [f for f in self.findings if 0.5 <= f.confidence < 0.8]
315            ),
316            "low_confidence": len([f for f in self.findings if f.confidence < 0.5]),
317        }

320@dataclass
321class AuditReport:
322    """
323    Comprehensive audit report for compliance and governance.
324
325    Attributes:
326        job_name: Name of the job/process that generated this report
327        timestamp: When the report was generated
328        policy_hash: Hash of the policy used
329        coverage_score: Overall PII coverage score
330        residual_risk: Calculated residual risk score
331        summary_stats: Summary statistics
332        findings_by_type: Findings grouped by PII type
333        performance_metrics: Performance and timing metrics
334        samples: Sample transformed values for review
335        scan_result: The underlying scan result
336    """
337
338    job_name: str
339    timestamp: datetime
340    policy_hash: str
341    coverage_score: float
342    residual_risk: float
343    summary_stats: Dict[str, int]
344    findings_by_type: Dict[str, List[Finding]]
345    performance_metrics: Dict[str, float]
346    samples: Dict[str, List[str]]
347    scan_result: ScanResult
348    metadata: Dict[str, Any] = field(default_factory=dict)
349
350    def __post_init__(self) -> None:
351        """Validate audit report data."""
352        if not 0.0 <= self.coverage_score <= 1.0:
353            raise ValueError("Coverage score must be between 0.0 and 1.0")
354
355        if not 0.0 <= self.residual_risk <= 1.0:
356            raise ValueError("Residual risk must be between 0.0 and 1.0")
357
358    def get_coverage_by_type(self) -> Dict[str, float]:
359        """Calculate coverage score by PII type."""
360        coverage_by_type = {}
361        for pii_type, findings in self.findings_by_type.items():
362            protected = len([f for f in findings if f.action_taken])
363            total = len(findings)
364            coverage_by_type[pii_type] = protected / total if total > 0 else 1.0
365        return coverage_by_type
366
367    def get_risk_factors(self) -> Dict[str, float]:
368        """Calculate risk factors contributing to residual risk."""
369        total_findings = len(self.scan_result.findings)
370        if total_findings == 0:
371            return {"no_pii_detected": 0.0}
372
373        unprotected = len([f for f in self.scan_result.findings if not f.action_taken])
374        low_confidence = len(
375            [f for f in self.scan_result.findings if f.confidence < 0.5]
376        )
377
378        return {
379            "unprotected_ratio": unprotected / total_findings,
380            "low_confidence_ratio": low_confidence / total_findings,
381            "coverage_gap": 1.0 - self.coverage_score,
382        }
383
384    def passes_threshold(self, threshold: float) -> bool:
385        """Check if coverage score meets the specified threshold."""
386        return self.coverage_score >= threshold
387
388    def model_dump(self) -> Dict[str, Any]:
389        """Return a serializable dictionary representation of the report."""
390        return asdict(self)
391
392    @property
393    def findings(self) -> List[Finding]:
394        """Compatibility property to access findings directly from the report."""
395        return self.scan_result.findings

 23class Scanner:
 24    """
 25    Core PII scanner that detects sensitive information in various data formats.
 26
 27    The scanner uses registered detectors to find PII and applies policy rules
 28    to determine confidence thresholds and detection strategies.
 29    """
 30
 31    def __init__(self, policy: Policy) -> None:
 32        """
 33        Initialize scanner with a policy.
 34
 35        Args:
 36            policy: Policy configuration for detection rules and thresholds
 37        """
 38        self.policy = policy
 39        self.detector_registry = DetectorRegistry()
 40        self._load_locale_packs()
 41
 42    def _load_locale_packs(self) -> None:
 43        """Load detectors for enabled locale packs."""
 44        for locale_pack in self.policy.locale_packs:
 45            self.detector_registry.load_locale_pack(locale_pack)
 46
 47    def scan_dataframe(
 48        self,
 49        df: Any,
 50        dataset_name: str = "unknown",
 51        confidence_threshold: Optional[float] = None,
 52    ) -> ScanResult:
 53        """
 54        Scan a pandas DataFrame for PII.
 55
 56        Args:
 57            df: DataFrame to scan
 58            dataset_name: Name of the dataset for reporting
 59            confidence_threshold: Override policy confidence threshold
 60
 61        Returns:
 62            ScanResult with detected PII and metadata
 63        """
 64        start_time = time.time()
 65
 66        threshold = self._resolve_threshold(confidence_threshold)
 67        findings: List[Finding] = []
 68
 69        if pd is None:
 70            raise ImportError(
 71                "pandas is required for scan_dataframe; install pandas to use this feature"
 72            )
 73
 74        # Scan each column
 75        for column in df.columns:
 76            column_findings = self._scan_column(df, column, threshold)
 77            findings.extend(column_findings)
 78
 79        # Calculate coverage score
 80        coverage_score = self._calculate_coverage_score(findings, dataset_name)
 81
 82        # Create scan metadata
 83        scan_metadata = {
 84            "scan_duration": time.time() - start_time,
 85            "confidence_threshold": threshold,
 86            "detectors_used": list(self.detector_registry.get_detector_names()),
 87            "locale_packs": self.policy.locale_packs,
 88        }
 89
 90        return ScanResult(
 91            findings=findings,
 92            coverage_score=coverage_score,
 93            scan_metadata=scan_metadata,
 94            policy_hash=self.policy.policy_hash or "",
 95            timestamp=datetime.now(),
 96            dataset_name=dataset_name,
 97            total_rows=len(df),
 98            total_columns=len(df.columns),
 99        )
100
101    def _scan_column(self, df: Any, column: str, threshold: float) -> List[Finding]:
102        """Scan a single column for PII."""
103        findings: List[Finding] = []
104
105        # Check if column has a specific rule
106        column_rule = self.policy.get_rule_for_column(column)
107        if column_rule and column_rule.override_confidence is not None:
108            threshold = column_rule.override_confidence
109
110        # Convert column to string and scan each value
111        series = df[column].astype(str)
112
113        for row_idx, value in enumerate(series):
114            if pd.isna(value) or value == "nan" or value.strip() == "":
115                continue
116
117            findings.extend(
118                self._scan_value(
119                    value, threshold, column, row_idx, {"column_name": column}
120                )
121            )
122
123        return findings
124
125    def _resolve_threshold(self, confidence_threshold: Optional[float]) -> float:
126        """Use the policy default only when no override was supplied."""
127        return (
128            self.policy.thresholds["min_confidence"]
129            if confidence_threshold is None
130            else confidence_threshold
131        )
132
133    def _scan_value(
134        self,
135        value: str,
136        threshold: float,
137        column: str,
138        row_index: int,
139        context: Optional[Dict[str, Any]] = None,
140    ) -> List[Finding]:
141        """Run detectors and normalize their matches for every input format."""
142        findings: List[Finding] = []
143        for detector in self.detector_registry.get_detectors():
144            matches = (
145                detector.find(value)
146                if context is None
147                else detector.find(value, context)
148            )
149            for match in matches:
150                finding = self._finding_from_match(
151                    match, detector.pii_type, detector.name, value, column, row_index
152                )
153                if finding is not None and finding.confidence >= threshold:
154                    findings.append(finding)
155        return findings
156
157    @staticmethod
158    def _finding_from_match(
159        match: Any,
160        pii_type: str,
161        detector_name: str,
162        value: str,
163        column: str,
164        row_index: int,
165    ) -> Optional[Finding]:
166        """Support tuple matches and legacy objects at the detector boundary."""
167        evidence = f"Detected by {detector_name} detector"
168        if isinstance(match, tuple) and len(match) == 3:
169            start, end, confidence = match
170            detected_value = value[start:end]
171            span = (start, end)
172        elif hasattr(match, "confidence"):
173            confidence = match.confidence
174            detected_value = match.value
175            span = match.span
176            evidence = getattr(match, "evidence", evidence)
177        else:
178            return None
179
180        return Finding(
181            type=pii_type,
182            value=detected_value,
183            span=span,
184            column=column,
185            row_index=row_index,
186            confidence=confidence,
187            evidence=evidence,
188        )
189
190    def _calculate_coverage_score(
191        self, findings: List[Finding], dataset_name: str
192    ) -> float:
193        """
194        Calculate PII coverage score based on policy rules.
195
196        Coverage = (Protected PII Items) / (Detected PII Items + Policy-Declared PII Fields)
197        """
198        if not findings:
199            return 1.0  # No PII detected = perfect coverage
200
201        protected_count = sum(
202            self._would_be_protected(finding, dataset_name) for finding in findings
203        )
204        declared_fields = sum(len(rule.columns or []) for rule in self.policy.rules)
205        return (protected_count + declared_fields) / (len(findings) + declared_fields)
206
207    def _would_be_protected(self, finding: Finding, dataset_name: str) -> bool:
208        """Check if a finding would be protected by policy rules."""
209        # Explicitly allowed by exception counts as handled by policy intent
210        if self.policy.is_allowed(dataset_name, finding.type):
211            return True
212
213        # Check if there's a specific rule for this type or column
214        type_rule = self.policy.get_rule_for_type(finding.type)
215        column_rule = self.policy.get_rule_for_column(finding.column)
216
217        return bool(type_rule or column_rule)
218
219    def scan_text(
220        self, text: str, confidence_threshold: Optional[float] = None
221    ) -> List[Finding]:
222        """
223        Scan a text string for PII.
224
225        Args:
226            text: Text to scan
227            confidence_threshold: Override policy confidence threshold
228
229        Returns:
230            List of findings
231        """
232        threshold = self._resolve_threshold(confidence_threshold)
233        return self._scan_value(text, threshold, "text", 0)
234
235    def scan_text_result(
236        self,
237        text: str,
238        dataset_name: str = "text",
239        confidence_threshold: Optional[float] = None,
240    ) -> ScanResult:
241        """Scan text and return a ScanResult object for reporting."""
242        threshold = self._resolve_threshold(confidence_threshold)
243        findings = self.scan_text(text, threshold)
244
245        coverage_score = self._calculate_coverage_score(findings, dataset_name)
246
247        scan_metadata = {
248            "scan_duration": 0.0,
249            "confidence_threshold": threshold,
250            "detectors_used": list(self.detector_registry.get_detector_names()),
251            "locale_packs": self.policy.locale_packs,
252        }
253
254        return ScanResult(
255            findings=findings,
256            coverage_score=coverage_score,
257            scan_metadata=scan_metadata,
258            policy_hash=self.policy.policy_hash or "",
259            timestamp=datetime.now(),
260            dataset_name=dataset_name,
261            total_rows=1,
262            total_columns=1,
263        )
264
265    def scan_file(
266        self,
267        file_path: Union[str, Path],
268        confidence_threshold: Optional[float] = None,
269    ) -> ScanResult:
270        """
271        Stream-scan a text or CSV file without loading it fully into memory.
272
273        Supports:
274        - .txt/.md: line-by-line
275        - .csv: row-by-row via csv module
276
277        For JSON/Parquet and other tabular formats, use SDKScanner (pandas) or
278        CLI load helpers.
279        """
280        start_time = time.time()
281        path = Path(file_path)
282        if not path.exists():
283            raise FileNotFoundError(f"File not found: {path}")
284
285        threshold = self._resolve_threshold(confidence_threshold)
286        findings: List[Finding] = []
287        total_rows = 0
288        total_columns = 1
289        dataset_name = path.stem
290
291        suffix = path.suffix.lower()
292        if suffix in [".txt", ".md"]:
293            with open(path, "r", encoding="utf-8", errors="ignore") as f:
294                for line_idx, line in enumerate(f):
295                    line = line.rstrip("\n")
296                    findings.extend(self._scan_value(line, threshold, "line", line_idx))
297                    total_rows += 1
298            total_columns = 1
299        elif suffix == ".csv":
300            with open(path, "r", encoding="utf-8", newline="") as f:
301                reader = csv.reader(f)
302                headers = next(reader, [])
303                total_columns = len(headers)
304                for row_idx, row in enumerate(reader):
305                    # Track rows processed
306                    total_rows += 1
307                    for col_idx, cell in enumerate(row):
308                        column_name = (
309                            headers[col_idx]
310                            if col_idx < len(headers)
311                            else f"col_{col_idx}"
312                        )
313                        findings.extend(
314                            self._scan_value(
315                                cell,
316                                threshold,
317                                column_name,
318                                row_idx,
319                                {"column_name": column_name},
320                            )
321                        )
322        else:
323            raise ValueError(
324                f"Unsupported file format for streaming scan: {suffix}. "
325                "Use SDKScanner (pandas) for JSON/Parquet."
326            )
327
328        coverage_score = self._calculate_coverage_score(findings, dataset_name)
329
330        scan_metadata = {
331            "scan_duration": time.time() - start_time,
332            "confidence_threshold": threshold,
333            "detectors_used": list(self.detector_registry.get_detector_names()),
334            "locale_packs": self.policy.locale_packs,
335            "file_path": str(path),
336            "streaming": True,
337        }
338
339        return ScanResult(
340            findings=findings,
341            coverage_score=coverage_score,
342            scan_metadata=scan_metadata,
343            policy_hash=self.policy.policy_hash or "",
344            timestamp=datetime.now(),
345            dataset_name=dataset_name,
346            total_rows=total_rows,
347            total_columns=total_columns,
348        )
349
350    def scan_dict(
351        self, data: Dict[str, Any], confidence_threshold: Optional[float] = None
352    ) -> List[Finding]:
353        """
354        Scan a dictionary for PII.
355
356        Args:
357            data: Dictionary to scan
358            confidence_threshold: Override policy confidence threshold
359
360        Returns:
361            List of findings
362        """
363        threshold = self._resolve_threshold(confidence_threshold)
364        findings: List[Finding] = []
365
366        for key, value in data.items():
367            if isinstance(value, str):
368                text_findings = self.scan_text(value, threshold)
369                # Update column name for each finding
370                for finding in text_findings:
371                    finding.column = key
372                    findings.append(finding)
373
374        return findings