#!/usr/bin/env python3 """Python Logging Audit Script A safe, vendor-neutral helper for reviewing logging hygiene in Python projects. What it checks: - Presence of structured log fields - Correlation identifiers in log records - Potential sensitive data exposure - Log level consistency - Basic rotation and retention settings in sample config - High-frequency or overly verbose logging patterns in text samples Usage examples: # Check a JSON Lines log file python logging_audit.py --input app.log --format jsonl # Check a plain-text log file python logging_audit.py --input app.log --format text # Validate a sample logging configuration file python logging_audit.py --config logging_config.json Notes: - This script does not modify files. - It does not connect to external systems. - It is designed to be adapted to your application and CI checks. """ from __future__ import annotations import argparse import json import re import sys from dataclasses import dataclass, field from pathlib import Path from typing import Dict, Iterable, List, Optional, Sequence, Tuple SENSITIVE_PATTERNS = { "password": re.compile(r"(?i)\bpassword\b\s*[:=]\s*[^\s,;]+"), "token": re.compile(r"(?i)\b(token|api[_-]?key|secret|bearer)\b\s*[:=]\s*[^\s,;]+"), "authorization": re.compile(r"(?i)authorization\s*[:=]\s*[^\s,;]+"), "cookie": re.compile(r"(?i)\bcookie\b\s*[:=]\s*[^\n]+"), "email": re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"), } COMMON_CORRELATION_FIELDS = ( "request_id", "trace_id", "span_id", "correlation_id", "job_id", "transaction_id", ) ALLOWED_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"} LEVEL_HINTS = re.compile(r"\b(DEBUG|INFO|WARNING|WARN|ERROR|CRITICAL|FATAL)\b") @dataclass class Finding: severity: str message: str evidence: Optional[str] = None @dataclass class Report: findings: List[Finding] = field(default_factory=list) stats: Dict[str, int] = field(default_factory=dict) def add(self, severity: str, message: str, evidence: Optional[str] = None) -> None: self.findings.append(Finding(severity=severity, message=message, evidence=evidence)) def count(self, key: str, amount: int = 1) -> None: self.stats[key] = self.stats.get(key, 0) + amount def has_errors(self) -> bool: return any(f.severity == "ERROR" for f in self.findings) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Audit Python logging output and sample config for production readiness.") parser.add_argument("--input", type=Path, help="Path to a log file to inspect.") parser.add_argument("--format", choices=("jsonl", "text"), default="jsonl", help="Log input format: JSON Lines or plain text.") parser.add_argument("--config", type=Path, help="Path to a JSON sample config to validate.") parser.add_argument("--max-lines", type=int, default=5000, help="Maximum number of lines to read from the input log file.") parser.add_argument("--require-fields", nargs="*", default=["level", "message"], help="Field names expected in structured log records.") parser.add_argument("--require-correlation", action="store_true", help="Warn if no correlation identifier fields are present.") parser.add_argument("--min-rotation-days", type=int, default=7, help="Minimum recommended retention/rotation days when checking config.") return parser.parse_args() def read_lines(path: Path, max_lines: int) -> 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}") lines: List[str] = [] with path.open("r", encoding="utf-8", errors="replace") as handle: for idx, line in enumerate(handle): if idx >= max_lines: break lines.append(line.rstrip("\n")) return lines def detect_sensitive_content(text: str) -> List[Tuple[str, str]]: hits: List[Tuple[str, str]] = [] for name, pattern in SENSITIVE_PATTERNS.items(): match = pattern.search(text) if match: snippet = match.group(0) hits.append((name, snippet[:120])) return hits def analyze_text_logs(lines: Sequence[str], require_correlation: bool) -> Report: report = Report() level_counts: Dict[str, int] = {lvl: 0 for lvl in ALLOWED_LEVELS} correlation_seen = False for line in lines: if not line.strip(): continue report.count("lines_scanned") for name, snippet in detect_sensitive_content(line): report.add("ERROR", f"Potential sensitive data detected ({name}).", snippet) level_match = LEVEL_HINTS.search(line) if level_match: level = level_match.group(1).replace("WARN", "WARNING") if level in ALLOWED_LEVELS: level_counts[level] += 1 else: report.add("WARNING", f"Unrecognized log level label: {level_match.group(1)}", line[:160]) if any(field in line for field in COMMON_CORRELATION_FIELDS): correlation_seen = True if len(line) > 4000: report.add("WARNING", "Very long log line detected; consider truncation or structure.", line[:160]) if require_correlation and not correlation_seen: report.add("WARNING", "No correlation identifier fields detected in text logs.") debug_count = level_counts.get("DEBUG", 0) info_count = level_counts.get("INFO", 0) error_count = level_counts.get("ERROR", 0) report.stats.update(level_counts) report.stats["level_hints_found"] = sum(level_counts.values()) if debug_count > max(50, len(lines) // 5): report.add("WARNING", "DEBUG appears very frequent; verify production verbosity is intentional.") if info_count == 0 and lines: report.add("WARNING", "No INFO-level activity detected; check whether lifecycle events are logged.") if error_count > 0: report.add("INFO", f"{error_count} error-level events detected; review if expected or actionable.") return report def analyze_jsonl_logs(lines: Sequence[str], require_fields: Sequence[str], require_correlation: bool) -> Report: report = Report() seen_fields: Dict[str, int] = {} correlation_seen = False for raw in lines: if not raw.strip(): continue report.count("lines_scanned") try: record = json.loads(raw) except json.JSONDecodeError: report.add("ERROR", "Invalid JSON line encountered.", raw[:200]) continue if not isinstance(record, dict): report.add("ERROR", "JSON log record is not an object.", raw[:200]) continue for field in require_fields: if field not in record: report.add("WARNING", f"Missing expected field: {field}") else: seen_fields[field] = seen_fields.get(field, 0) + 1 for key in COMMON_CORRELATION_FIELDS: if key in record and record[key]: correlation_seen = True serialized = json.dumps(record, ensure_ascii=False) for name, snippet in detect_sensitive_content(serialized): report.add("ERROR", f"Potential sensitive data detected ({name}).", snippet) level = str(record.get("level", record.get("severity", ""))).upper() if level and level not in ALLOWED_LEVELS: report.add("WARNING", f"Unrecognized level value: {level}", serialized[:200]) if len(serialized) > 4000: report.add("WARNING", "Large structured record detected; consider field trimming or truncation.") if require_correlation and not correlation_seen: report.add("WARNING", "No correlation identifier fields detected in structured logs.") report.stats.update({f"field_{k}": v for k, v in seen_fields.items()}) return report def validate_config(path: Path, min_rotation_days: int) -> Report: report = Report() if not path.exists(): raise FileNotFoundError(f"Config file not found: {path}") with path.open("r", encoding="utf-8", errors="replace") as handle: try: config = json.load(handle) except json.JSONDecodeError as exc: raise ValueError(f"Config file must be valid JSON: {exc}") from exc if not isinstance(config, dict): raise ValueError("Config file must contain a JSON object at the top level.") handlers = config.get("handlers", {}) if not isinstance(handlers, dict): report.add("ERROR", "Config 'handlers' should be a JSON object.") return report rotating_found = False for name, handler in handlers.items(): if not isinstance(handler, dict): report.add("WARNING", f"Handler {name!r} is not an object.") continue handler_type = str(handler.get("type", "")).lower() if any(token in handler_type for token in ("rotating", "timed", "watched")): rotating_found = True backup_count = handler.get("backupCount") if isinstance(backup_count, int) and backup_count < 1: report.add("WARNING", f"Handler {name!r} has backupCount < 1.") retention_days = handler.get("retention_days") if isinstance(retention_days, int) and retention_days < min_rotation_days: report.add("WARNING", f"Handler {name!r} retention_days is below recommended minimum of {min_rotation_days}.") if not rotating_found: report.add("WARNING", "No rotating/timed handler detected in sample config.") return report def print_report(report: Report) -> int: if report.stats: print("Stats:") for key in sorted(report.stats): print(f"- {key}: {report.stats[key]}") print() if not report.findings: print("No findings. The sample appears reasonably aligned with the checks performed.") return 0 print("Findings:") for finding in report.findings: print(f"- [{finding.severity}] {finding.message}") if finding.evidence: print(f" Evidence: {finding.evidence}") return 1 if report.has_errors() else 0 def main() -> int: args = parse_args() overall = Report() if args.input: try: lines = read_lines(args.input, args.max_lines) except (OSError, ValueError) as exc: print(f"Error reading input: {exc}", file=sys.stderr) return 2 if args.format == "jsonl": input_report = analyze_jsonl_logs(lines, args.require_fields, args.require_correlation) else: input_report = analyze_text_logs(lines, args.require_correlation) overall.findings.extend(input_report.findings) overall.stats.update(input_report.stats) if args.config: try: config_report = validate_config(args.config, args.min_rotation_days) except (OSError, ValueError) as exc: print(f"Error validating config: {exc}", file=sys.stderr) return 2 overall.findings.extend(config_report.findings) for key, value in config_report.stats.items(): overall.stats[key] = overall.stats.get(key, 0) + value if not args.input and not args.config: print("Provide --input, --config, or both.", file=sys.stderr) return 2 return print_report(overall) if __name__ == "__main__": raise SystemExit(main())