enforcing-nophi-logging

Enforcing No-PHI Logging

Logs are a top breach vector: a clinical string lands in a log line, gets shipped to a centralized log store and an error tracker, and is now PHI sitting outside the de-id boundary. OpenMed's local-first stance says no raw PHI in logs, caches, or error reports — this skill enforces it with a redaction guard that runs before any record is emitted.

When to use this skill

• An OpenMed service logs request text, model output, or exception messages.
• You ship logs/traces to a centralized store or error tracker (Sentry, ELK).
• You need a logging.Filter (or OTel processor) that redacts PHI pre-emit.
• You want structured, no-PHI log fields (offsets, hashes, counts) for debugging.

Quick start — a redacting logging.Filter

import logging
import re
import openmed

# Cheap regex pre-filter for the highest-risk structured identifiers. This runs
# on every record, so keep it fast; the model is the fallback for free-text PHI.
_FAST_PATTERNS = [
    (re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), "[SSN]"),
    (re.compile(r"\b\d{16}\b"), "[CARD]"),
    (re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b"), "[EMAIL]"),
    (re.compile(r"\b(?:\+?\d[\d().\-\s]{7,}\d)\b"), "[PHONE]"),
]

class NoPHIFilter(logging.Filter):
    """Redact PHI from a log record before it is emitted. Fail closed."""

    def __init__(self, model_name: str | None = None, use_model: bool = True):
        super().__init__()
        self.model_name = model_name
        self.use_model = use_model

    def filter(self, record: logging.LogRecord) -> bool:
        try:
            message = record.getMessage()
            record.msg = self._scrub(message)
            record.args = ()                 # message already rendered & scrubbed
        except Exception:
            # Never let the logger leak on error — drop the message, keep the level.
            record.msg = "[REDACTED: scrub error]"
            record.args = ()
        return True                          # keep the (now-clean) record

    def _scrub(self, text: str) -> str:
        for pattern, tag in _FAST_PATTERNS:
            text = pattern.sub(tag, text)
        if not self.use_model:
            return text
        # Model fallback for free-text PHI (names, locations, dates). Replace by
        # offset, right-to-left, so earlier offsets stay valid.
        spans = openmed.extract_pii(text, model_name=self.model_name) \
            if self.model_name else openmed.extract_pii(text)
        for e in sorted(spans.entities, key=lambda s: s.start, reverse=True):
            text = text[:e.start] + f"[{e.label}]" + text[e.end:]
        return text

# Attach to every handler that might emit clinical text.
handler = logging.StreamHandler()
handler.addFilter(NoPHIFilter(model_name="OpenMed/Privacy-PII-Detection"))
logging.getLogger("openmed.service").addHandler(handler)