#!/usr/bin/env python3 """Secure C# logging validation helper. This script helps review exported log files for common secure-logging concerns: - presence of correlation IDs - structured fields versus plain text only - possible sensitive data exposure - basic consistency checks across events It is intentionally vendor-neutral and avoids any destructive actions. Typical usage: python secure_logging_audit.py --input logs.jsonl python secure_logging_audit.py --input logs.txt --format text python secure_logging_audit.py --input logs.jsonl --require-field CorrelationId Notes: - This is a lightweight validation helper, not a full SIEM or DLP product. - Review findings before using them in production workflows. """ from __future__ import annotations import argparse import json import re import sys from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple SENSITIVE_PATTERNS = { "password": re.compile(r"(?i)\bpassword\b\s*[:=]"), "token": re.compile(r"(?i)\b(?:access[_-]?token|refresh[_-]?token|token)\b\s*[:=]"), "bearer": re.compile(r"(?i)\bBearer\s+[A-Za-z0-9\-\._~\+/]+=*"), "api_key": re.compile(r"(?i)\b(?:api[_-]?key|apikey|secret)\b\s*[:=]"), "cookie": re.compile(r"(?i)\b(?:set-cookie|cookie)\b\s*[:=]"), "jwt": re.compile(r"(?i)\beyJ[A-Za-z0-9\-_=]+\.[A-Za-z0-9\-_=]+\.[A-Za-z0-9\-_.+/=]*\b"), } DEFAULT_REQUIRED_FIELDS = ["CorrelationId"] @dataclass class Finding: severity: str message: str line_no: Optional[int] = None def render(self) -> str: prefix = f"line {self.line_no}: " if self.line_no is not None else "" return f"[{self.severity}] {prefix}{self.message}" @dataclass class AuditReport: total_events: int = 0 structured_events: int = 0 correlation_ids: Dict[str, int] = field(default_factory=dict) findings: List[Finding] = field(default_factory=list) def add_finding(self, severity: str, message: str, line_no: Optional[int] = None) -> None: self.findings.append(Finding(severity=severity, message=message, line_no=line_no)) def summary(self) -> str: return ( f"events={self.total_events}, structured={self.structured_events}, " f"unique_correlation_ids={len(self.correlation_ids)}, findings={len(self.findings)}" ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Validate logs for structured fields, correlation IDs, and potential sensitive data exposure." ) parser.add_argument("--input", required=True, help="Path to the log file to inspect.") parser.add_argument( "--format", choices=["jsonl", "text"], default="jsonl", help="Input format: JSON Lines (default) or plain text.", ) parser.add_argument( "--require-field", action="append", default=list(DEFAULT_REQUIRED_FIELDS), help="Structured field required in each JSON event. May be provided multiple times.", ) parser.add_argument( "--show-events", action="store_true", help="Print parsed event summaries while auditing.", ) return parser.parse_args() def read_lines(path: Path) -> List[str]: if not path.exists(): raise FileNotFoundError(f"Input file not found: {path}") if not path.is_file(): raise ValueError(f"Input path is not a file: {path}") return path.read_text(encoding="utf-8", errors="replace").splitlines() def looks_structured(obj: Any) -> bool: return isinstance(obj, dict) def extract_correlation_id(event: Dict[str, Any], required_fields: Sequence[str]) -> Optional[str]: for field_name in required_fields: value = event.get(field_name) if value is not None and str(value).strip(): return str(value).strip() return None def flatten_values(value: Any) -> Iterable[str]: if value is None: return if isinstance(value, dict): for nested in value.values(): yield from flatten_values(nested) elif isinstance(value, list): for item in value: yield from flatten_values(item) else: yield str(value) def detect_sensitive_content(text: str) -> List[str]: hits: List[str] = [] for name, pattern in SENSITIVE_PATTERNS.items(): if pattern.search(text): hits.append(name) return hits def audit_jsonl(lines: Sequence[str], required_fields: Sequence[str], show_events: bool) -> AuditReport: report = AuditReport() for idx, raw in enumerate(lines, start=1): line = raw.strip() if not line: continue report.total_events += 1 try: event = json.loads(line) except json.JSONDecodeError as exc: report.add_finding("high", f"invalid JSON event: {exc.msg}", idx) continue if not looks_structured(event): report.add_finding("medium", "event is not a JSON object", idx) continue report.structured_events += 1 correlation_id = extract_correlation_id(event, required_fields) if correlation_id: report.correlation_ids[correlation_id] = report.correlation_ids.get(correlation_id, 0) + 1 else: report.add_finding("medium", f"missing required correlation field(s): {', '.join(required_fields)}", idx) serialized = json.dumps(event, ensure_ascii=False) sensitive_hits = detect_sensitive_content(serialized) if sensitive_hits: report.add_finding( "high", f"possible sensitive data exposure detected: {', '.join(sorted(set(sensitive_hits)))}", idx, ) if show_events: fields = ", ".join(sorted(event.keys())) print(f"[event {idx}] fields={fields}") return report def audit_text(lines: Sequence[str], show_events: bool) -> AuditReport: report = AuditReport() for idx, raw in enumerate(lines, start=1): line = raw.rstrip("\n") if not line.strip(): continue report.total_events += 1 hits = detect_sensitive_content(line) if hits: report.add_finding( "high", f"possible sensitive data exposure detected in plain text: {', '.join(sorted(set(hits)))}", idx, ) if "correlationid" in line.lower() or "requestid" in line.lower(): match = re.search(r"(?i)\b(?:correlationid|requestid)\b\s*[:=]\s*([A-Za-z0-9\-_.]+)", line) if match: cid = match.group(1) report.correlation_ids[cid] = report.correlation_ids.get(cid, 0) + 1 else: report.add_finding("low", "plain text log line may be harder to query consistently", idx) if show_events: print(f"[event {idx}] text={line[:120]}") return report def print_report(report: AuditReport) -> None: print("Audit summary") print("-------------") print(report.summary()) if report.correlation_ids: print("\nCorrelation IDs observed:") for cid, count in sorted(report.correlation_ids.items(), key=lambda item: (-item[1], item[0])): print(f"- {cid} ({count} event(s))") if report.findings: print("\nFindings:") for finding in report.findings: print(finding.render()) else: print("\nNo findings detected.") def main() -> int: args = parse_args() path = Path(args.input) try: lines = read_lines(path) if args.format == "jsonl": report = audit_jsonl(lines, args.require_field, args.show_events) else: report = audit_text(lines, args.show_events) print_report(report) return 0 if not any(f.severity in {"high", "medium"} for f in report.findings) else 2 except (OSError, ValueError) as exc: print(f"Error: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())