```python #!/usr/bin/env python3 """secure_ml_monitor.py A safe, vendor-neutral starter script for monitoring ML model telemetry. What it does: - Loads baseline and live telemetry records from JSONL files - Validates schema and redacts sensitive fields - Computes lightweight health checks: - missingness rate - numeric summary drift - prediction distribution shift - latency regression - Flags suspicious usage patterns: - excessive retries - unusual request source concentration - oversized payloads - malformed or replayed events - Emits a JSON report suitable for dashboards, alerts, or incident triage Assumptions: - Input files are JSON Lines (one JSON object per line) - Telemetry is already minimized and sanitized at the source - No raw payloads, secrets, or PII are required for this script Usage examples: python secure_ml_monitor.py --baseline baseline.jsonl --live live.jsonl --output report.json python secure_ml_monitor.py --baseline baseline.jsonl --live live.jsonl --max-latency-regression 0.30 The script is intentionally conservative: records that fail validation are quarantined in the output report rather than silently ignored. """ from __future__ import annotations import argparse import json import math import statistics from collections import Counter, defaultdict from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple REQUIRED_FIELDS = { "timestamp": str, "request_id": str, "model_version": str, "feature_schema_version": str, "source": str, } OPTIONAL_NUMERIC_FIELDS = {"latency_ms", "prediction", "retry_count", "payload_size_bytes"} SENSITIVE_FIELD_NAMES = { "password", "token", "secret", "api_key", "authorization", "cookie", } DEFAULT_MAX_EVENT_SIZE_BYTES = 32_768 DEFAULT_MAX_RETRY_COUNT = 5 DEFAULT_SUSPICIOUS_SOURCE_SHARE = 0.80 DEFAULT_DRIFT_RATIO_THRESHOLD = 0.35 DEFAULT_LATENCY_REGRESSION = 0.25 @dataclass class ValidationResult: valid: bool redacted_record: Optional[Dict[str, Any]] = None errors: Optional[List[str]] = None def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Validate ML telemetry and compute secure monitoring signals." ) parser.add_argument("--baseline", required=True, help="Path to baseline JSONL telemetry file.") parser.add_argument("--live", required=True, help="Path to live JSONL telemetry file.") parser.add_argument("--output", required=True, help="Path to write JSON monitoring report.") parser.add_argument( "--max-event-size-bytes", type=int, default=DEFAULT_MAX_EVENT_SIZE_BYTES, help="Maximum allowed size for a telemetry record in bytes.", ) parser.add_argument( "--max-retry-count", type=int, default=DEFAULT_MAX_RETRY_COUNT, help="Retry count above which a request is flagged.", ) parser.add_argument( "--suspicious-source-share", type=float, default=DEFAULT_SUSPICIOUS_SOURCE_SHARE, help="Source concentration threshold used to flag unusual access patterns.", ) parser.add_argument( "--drift-ratio-threshold", type=float, default=DEFAULT_DRIFT_RATIO_THRESHOLD, help="Relative change threshold for numeric drift checks.", ) parser.add_argument( "--max-latency-regression", type=float, default=DEFAULT_LATENCY_REGRESSION, help="Relative increase in p95 latency tolerated before flagging regression.", ) return parser.parse_args() def load_jsonl(path: str) -> List[Dict[str, Any]]: records: List[Dict[str, Any]] = [] with open(path, "r", encoding="utf-8") as f: for line_no, line in enumerate(f, start=1): line = line.strip() if not line: continue try: obj = json.loads(line) if not isinstance(obj, dict): raise ValueError("record is not an object") records.append(obj) except Exception as exc: records.append({"__parse_error__": str(exc), "__line__": line_no}) return records def is_sensitive_key(key: str) -> bool: key_lower = key.lower() return any(token in key_lower for token in SENSITIVE_FIELD_NAMES) def redact_record(record: Dict[str, Any]) -> Dict[str, Any]: redacted: Dict[str, Any] = {} for key, value in record.items(): if is_sensitive_key(key): redacted[key] = "[REDACTED]" continue if key == "feature_stats" and isinstance(value, dict): redacted[key] = redact_feature_stats(value) continue redacted[key] = value return redacted def redact_feature_stats(feature_stats: Dict[str, Any]) -> Dict[str, Any]: out: Dict[str, Any] = {} for feature_name, stats in feature_stats.items(): if is_sensitive_key(feature_name): out[feature_name] = "[REDACTED]" continue if isinstance(stats, dict): safe_stats = {k: v for k, v in stats.items() if k in {"missing", "value_bucket", "present"}} out[feature_name] = safe_stats else: out[feature_name] = stats return out def validate_record(record: Dict[str, Any], max_event_size_bytes: int) -> ValidationResult: errors: List[str] = [] if "__parse_error__" in record: return ValidationResult(valid=False, errors=[f"parse_error: {record['__parse_error__']}"]) serialized_size = len(json.dumps(record, separators=(",", ":")).encode("utf-8")) if serialized_size > max_event_size_bytes: errors.append(f"event too large: {serialized_size} bytes") for field, expected_type in REQUIRED_FIELDS.items(): if field not in record: errors.append(f"missing required field: {field}") elif not isinstance(record[field], expected_type): errors.append(f"invalid type for {field}: expected {expected_type.__name__}") if "request_id" in record and isinstance(record.get("request_id"), str): if len(record["request_id"].strip()) == 0: errors.append("request_id is empty") if "feature_stats" in record and not isinstance(record["feature_stats"], dict): errors.append("feature_stats must be an object when present") if "retry_count" in record: if not isinstance(record["retry_count"], int) or record["retry_count"] < 0: errors.append("retry_count must be a non-negative integer") if "payload_size_bytes" in record: if not isinstance(record["payload_size_bytes"], int) or record["payload_size_bytes"] < 0: errors.append("payload_size_bytes must be a non-negative integer") if any(is_sensitive_key(k) for k in record.keys()): errors.append("sensitive field names detected in event") return ValidationResult(valid=not errors, redacted_record=redact_record(record), errors=errors) def numeric_values(records: Sequence[Dict[str, Any]], field: str) -> List[float]: values: List[float] = [] for r in records: v = r.get(field) if isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(float(v)): values.append(float(v)) return values def summarize(values: Sequence[float]) -> Dict[str, Optional[float]]: if not values: return {"count": 0, "mean": None, "p95": None, "min": None, "max": None} sorted_vals = sorted(values) p95_index = max(0, min(len(sorted_vals) - 1, int(math.ceil(0.95 * len(sorted_vals))) - 1)) return { "count": float(len(values)), "mean": statistics.fmean(values), "p95": sorted_vals[p95_index], "min": min(values), "max": max(values), } def relative_change(baseline: Optional[float], live: Optional[float]) -> Optional[float]: if baseline is None or live is None: return None if baseline == 0: return None return abs(live - baseline) / abs(baseline) def compute_source_concentration(records: Sequence[Dict[str, Any]]) -> Tuple[str, float]: sources = [str(r.get("source")) for r in records if isinstance(r.get("source"), str)] if not sources: return ("unknown", 0.0) counts = Counter(sources) source, count = counts.most_common(1)[0] return source, count / len(sources) def analyze(baseline: List[Dict[str, Any]], live: List[Dict[str, Any]], args: argparse.Namespace) -> Dict[str, Any]: baseline_valid: List[Dict[str, Any]] = [] live_valid: List[Dict[str, Any]] = [] quarantined: List[Dict[str, Any]] = [] for record in baseline: result = validate_record(record, args.max_event_size_bytes) if result.valid and result.redacted_record is not None: baseline_valid.append(result.redacted_record) else: quarantined.append({"stream": "baseline", "errors": result.errors or [], "record": redact_record(record)}) for record in live: result = validate_record(record, args.max_event_size_bytes) if result.valid and result.redacted_record is not None: live_valid.append(result.redacted_record) else: quarantined.append({"stream": "live", "errors": result.errors or [], "record": redact_record(record)}) report: Dict[str, Any] = { "summary": { "baseline_records": len(baseline_valid), "live_records": len(live_valid), "quarantined_records": len(quarantined), }, "health": {}, "security": {}, "quarantined": quarantined, } baseline_latency = summarize(numeric_values(baseline_valid, "latency_ms")) live_latency = summarize(numeric_values(live_valid, "latency_ms")) latency_change = relative_change(baseline_latency["p95"], live_latency["p95"]) baseline_predictions = numeric_values(baseline_valid, "prediction") live_predictions = numeric_values(live_valid, "prediction") baseline_missing = _missingness_rate(baseline_valid) live_missing = _missingness_rate(live_valid) baseline_source, baseline_source_share = compute_source_concentration(baseline_valid) live_source, live_source_share = compute_source_concentration(live_valid) report["health"] = { "latency": { "baseline": baseline_latency, "live": live_latency, "p95_relative_change": latency_change, "alert": bool(latency_change is not None and latency_change > args.max_latency_regression), }, "prediction_mean": { "baseline": statistics.fmean(baseline_predictions) if baseline_predictions else None, "live": statistics.fmean(live_predictions) if live_predictions else None, }, "missingness": { "baseline": baseline_missing, "live": live_missing, }, "drift": _drift_report(baseline_valid, live_valid, args.drift_ratio_threshold), } report["security"] = { "source_concentration": { "baseline_top_source": baseline_source, "baseline_share": baseline_source_share, "live_top_source": live_source, "live_share": live_source_share, "alert": bool(live_source_share >= args.suspicious_source_share), }, "retry_anomalies": _retry_anomalies(live_valid, args.max_retry_count), "payload_size_outliers": _payload_outliers(live_valid), } return report def _missingness_rate(records: Sequence[Dict[str, Any]]) -> Dict[str, float]: if not records: return {} feature_counts: Dict[str, int] = defaultdict(int) total = len(records) for r in records: feature_stats = r.get("feature_stats") if isinstance(feature_stats, dict): for feature, stats in feature_stats.items(): if isinstance(stats, dict) and stats.get("missing") is True: feature_counts[feature] += 1 return {feature: count / total for feature, count in feature_counts.items()} def _drift_report(baseline: Sequence[Dict[str, Any]], live: Sequence[Dict[str, Any]], threshold: float) -> Dict[str, Any]: features = set() for records in (baseline, live): for r in records: fs = r.get("feature_stats") if isinstance(fs, dict): features.update(fs.keys()) drifted: Dict[str, Any] = {} for feature in sorted(features): b_vals = _bucketized_feature_counts(baseline, feature) l_vals = _bucketized_feature_counts(live, feature) all_keys = set(b_vals) | set(l_vals) total_b = sum(b_vals.values()) or 1 total_l = sum(l_vals.values()) or 1 max_change = 0.0 for key in all_keys: b_share = b_vals.get(key, 0) / total_b l_share = l_vals.get(key, 0) / total_l max_change = max(max_change, abs(l_share - b_share)) if max_change > threshold: drifted[feature] = {"max_share_change": max_change, "alert": True} return {"features": drifted, "alert": bool(drifted)} def _bucketized_feature_counts(records: Sequence[Dict[str, Any]], feature: str) -> Counter: c: Counter = Counter() for r in records: fs = r.get("feature_stats") if not isinstance(fs, dict): continue stats = fs.get(feature) if isinstance(stats, dict): bucket = stats.get("value_bucket") if bucket is not None: c[str(bucket)] += 1 elif stats.get("missing") is True: c["[MISSING]"] += 1 else: c["[OTHER]"] += 1 return c def _retry_anomalies(records: Sequence[Dict[str, Any]], max_retry_count: int) -> Dict[str, Any]: offenders = [r.get("request_id") for r in records if isinstance(r.get("retry_count"), int) and r["retry_count"] > max_retry_count] return {"count": len(offenders), "request_ids": offenders[:50], "alert": bool(offenders)} def _payload_outliers(records: Sequence[Dict[str, Any]]) -> Dict[str, Any]: sizes = numeric_values(records, "payload_size_bytes") if not sizes: return {"count": 0, "alert": False} p95 = summarize(sizes)["p95"] outliers = [r.get("request_id") for r in records if isinstance(r.get("payload_size_bytes"), (int, float)) and p95 is not None and r["payload_size_bytes"] > p95] return {"count": len(outliers), "request_ids": outliers[:50], "alert": bool(outliers)} def main() -> int: args = parse_args() baseline_path = Path(args.baseline) live_path = Path(args.live) if not baseline_path.exists(): raise SystemExit(f"baseline file not found: {baseline_path}") if not live_path.exists(): raise SystemExit(f"live file not found: {live_path}") if args.max_event_size_bytes <= 0: raise SystemExit("--max-event-size-bytes must be positive") if not (0.0 < args.suspicious_source_share <= 1.0): raise SystemExit("--suspicious-source-share must be between 0 and 1") if not (0.0 < args.drift_ratio_threshold <= 1.0): raise SystemExit("--drift-ratio-threshold must be between 0 and 1") if not (0.0 < args.max_latency_regression <= 10.0): raise SystemExit("--max-latency-regression must be positive") baseline = load_jsonl(str(baseline_path)) live = load_jsonl(str(live_path)) report = analyze(baseline, live, args) output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) with open(output_path, "w", encoding="utf-8") as f: json.dump(report, f, indent=2, sort_keys=True) f.write("\n") print(f"Wrote monitoring report to {output_path}") return 0 if __name__ == "__main__": raise SystemExit(main()) ```