Spaces:
Sleeping
Sleeping
File size: 3,997 Bytes
9e00302 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """
utils/audit_log.py
SecureLens — Audit Trail System
Logs all encrypted inferences with timestamps.
"""
import os, json, hashlib, logging
from datetime import datetime
LOGS_DIR = os.path.join(
os.path.dirname(__file__), "..", "logs")
os.makedirs(LOGS_DIR, exist_ok=True)
LOG_FILE = os.path.join(LOGS_DIR, "audit_trail.jsonl")
APP_LOG = os.path.join(LOGS_DIR, "app.log")
# Configure Python logger
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler(APP_LOG),
logging.StreamHandler(),
]
)
logger = logging.getLogger("securelens")
class AuditLogger:
"""
Logs every encrypted inference request.
Stores: timestamp, image hash, prediction,
confidence, encryption params, latency.
Never stores: raw image data, patient identifiers.
"""
def __init__(self, log_file=LOG_FILE):
self.log_file = log_file
os.makedirs(os.path.dirname(log_file), exist_ok=True)
def log_inference(
self,
image_bytes: bytes,
prediction: str,
confidence: float,
latency_ms: float,
encryption_params: dict,
endpoint: str = "/api/predict",
):
"""
Logs one encrypted inference event.
Image is hashed — never stored raw.
"""
image_hash = hashlib.sha256(image_bytes).hexdigest()[:16]
entry = {
"timestamp" : datetime.utcnow().isoformat() + "Z",
"endpoint" : endpoint,
"image_hash" : image_hash,
"prediction" : prediction,
"confidence_pct" : round(confidence, 2),
"latency_ms" : round(latency_ms, 1),
"encryption_scheme": encryption_params.get(
"scheme", "CKKS"),
"security_bits" : encryption_params.get(
"security_bits", 128),
"ciphertext_kb" : encryption_params.get(
"ciphertext_size_kb", 0),
"data_exposed" : "none",
}
with open(self.log_file, "a", encoding="utf-8") as f:
f.write(json.dumps(entry) + "\n")
logger.info(
f"Inference logged | hash={image_hash} "
f"pred={prediction} conf={confidence:.1f}% "
f"latency={latency_ms:.1f}ms")
return entry
def get_recent_logs(self, n=20):
"""Returns last n log entries."""
if not os.path.exists(self.log_file):
return []
with open(self.log_file, encoding="utf-8") as f:
lines = f.readlines()
entries = []
for line in lines[-n:]:
try:
entries.append(json.loads(line.strip()))
except Exception:
continue
return list(reversed(entries))
def get_stats(self):
"""Returns summary statistics of all logged inferences."""
if not os.path.exists(self.log_file):
return {"total": 0}
entries = []
with open(self.log_file, encoding="utf-8") as f:
for line in f:
try:
entries.append(json.loads(line.strip()))
except Exception:
continue
if not entries:
return {"total": 0}
preds = [e["prediction"] for e in entries]
latencies = [e["latency_ms"] for e in entries]
return {
"total" : len(entries),
"normal_count" : preds.count("Normal"),
"pneumonia_count": preds.count("Pneumonia"),
"avg_latency_ms" : round(
sum(latencies)/len(latencies), 1),
"max_latency_ms" : round(max(latencies), 1),
"min_latency_ms" : round(min(latencies), 1),
"first_log" : entries[-1]["timestamp"],
"last_log" : entries[0]["timestamp"],
}
# Global instance
audit_logger = AuditLogger() |