Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| HIPAA Compliance Configuration for Clinical Deployment. | |
| This module provides security configurations and utilities to help | |
| ensure HIPAA compliance for the DCE Intent Classifier service. | |
| IMPORTANT: This is a helper module. Full HIPAA compliance requires | |
| organizational policies, training, and infrastructure beyond this code. | |
| """ | |
| import hashlib | |
| import logging | |
| import os | |
| import re | |
| from datetime import datetime, timezone | |
| from typing import Any, Dict, List, Optional, Tuple | |
| # ============================================================================= | |
| # HIPAA SECURITY CONFIGURATION | |
| # ============================================================================= | |
| # PHI (Protected Health Information) patterns to detect and redact in logs | |
| PHI_PATTERNS = { | |
| "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), | |
| "phone": re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b"), | |
| "email": re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"), | |
| "mrn": re.compile(r"\b(MRN|mrn|Medical Record)[\s:#]*\d{6,12}\b", re.IGNORECASE), | |
| "dob": re.compile(r"\b(DOB|dob|birth)[\s:#]*\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b", re.IGNORECASE), | |
| "credit_card": re.compile(r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b"), | |
| } | |
| # Session timeout in seconds (HIPAA recommends automatic logoff) | |
| SESSION_TIMEOUT_SECONDS = int(os.getenv("SESSION_TIMEOUT", "900")) # 15 minutes default | |
| # Audit log retention days (HIPAA requires 6 years minimum) | |
| AUDIT_LOG_RETENTION_DAYS = int(os.getenv("AUDIT_RETENTION_DAYS", "2190")) # 6 years | |
| # Maximum failed login attempts before lockout | |
| MAX_FAILED_ATTEMPTS = int(os.getenv("MAX_FAILED_ATTEMPTS", "5")) | |
| LOCKOUT_DURATION_SECONDS = int(os.getenv("LOCKOUT_DURATION", "900")) # 15 minutes | |
| # Encryption requirements | |
| REQUIRE_TLS = os.getenv("REQUIRE_TLS", "true").lower() == "true" | |
| MIN_TLS_VERSION = os.getenv("MIN_TLS_VERSION", "1.2") | |
| # Data handling | |
| LOG_PHI = os.getenv("LOG_PHI", "false").lower() == "true" # Never log PHI by default | |
| REDACT_PHI_IN_LOGS = os.getenv("REDACT_PHI", "true").lower() == "true" | |
| # ============================================================================= | |
| # PHI DETECTION AND REDACTION | |
| # ============================================================================= | |
| def detect_phi(text: str) -> List[Tuple[str, str]]: | |
| """ | |
| Detect potential PHI in text. | |
| Returns: | |
| List of tuples (phi_type, matched_text) | |
| """ | |
| findings = [] | |
| for phi_type, pattern in PHI_PATTERNS.items(): | |
| matches = pattern.findall(text) | |
| for match in matches: | |
| findings.append((phi_type, match)) | |
| return findings | |
| def redact_phi(text: str) -> str: | |
| """ | |
| Redact potential PHI from text for safe logging. | |
| Returns: | |
| Text with PHI replaced by [REDACTED-{type}] | |
| """ | |
| redacted = text | |
| for phi_type, pattern in PHI_PATTERNS.items(): | |
| redacted = pattern.sub(f"[REDACTED-{phi_type.upper()}]", redacted) | |
| return redacted | |
| def hash_identifier(identifier: str) -> str: | |
| """ | |
| Create a one-way hash of an identifier for audit logging. | |
| This allows tracking without storing actual PHI. | |
| WARNING: In production, set PHI_HASH_SALT to a unique secret value. | |
| """ | |
| salt = os.getenv("PHI_HASH_SALT", "") | |
| environment = os.getenv("ENVIRONMENT", "development").lower() | |
| if not salt: | |
| if environment == "production": | |
| raise RuntimeError( | |
| "CRITICAL: PHI_HASH_SALT environment variable must be set in production! " | |
| "This is required for HIPAA-compliant identifier hashing." | |
| ) | |
| import logging | |
| logging.getLogger("hipaa_audit").warning( | |
| "PHI_HASH_SALT not configured - using default. Set this to a unique secret in production!" | |
| ) | |
| salt = "dce-intent-classifier-default" | |
| return hashlib.sha256(f"{salt}:{identifier}".encode()).hexdigest()[:16] | |
| # ============================================================================= | |
| # AUDIT LOGGING | |
| # ============================================================================= | |
| class HIPAAAuditLogger: | |
| """ | |
| HIPAA-compliant audit logger. | |
| Tracks all access to the system with required audit fields. | |
| """ | |
| def __init__(self, logger_name: str = "hipaa_audit"): | |
| self.logger = logging.getLogger(logger_name) | |
| self.logger.setLevel(logging.INFO) | |
| def log_access( | |
| self, | |
| user_id: Optional[str], | |
| action: str, | |
| resource: str, | |
| client_ip: str, | |
| request_id: str, | |
| success: bool, | |
| details: Optional[str] = None, | |
| ) -> None: | |
| """ | |
| Log an access event with HIPAA-required fields. | |
| Required fields per HIPAA: | |
| - User identification (hashed if applicable) | |
| - Date and time of access | |
| - Action performed | |
| - Resource accessed | |
| - Success/failure status | |
| """ | |
| timestamp = datetime.now(timezone.utc).isoformat() | |
| # Hash user_id if present to avoid logging actual identifiers | |
| user_hash = hash_identifier(user_id) if user_id else "anonymous" | |
| # Redact any PHI in details | |
| safe_details = redact_phi(details) if details else None | |
| log_entry = { | |
| "timestamp": timestamp, | |
| "user_hash": user_hash, | |
| "action": action, | |
| "resource": resource, | |
| "client_ip": client_ip, | |
| "request_id": request_id, | |
| "success": success, | |
| "details": safe_details, | |
| } | |
| self.logger.info(f"HIPAA_AUDIT: {log_entry}") | |
| def log_escalation( | |
| self, | |
| request_id: str, | |
| client_ip: str, | |
| escalation_type: str, | |
| confidence: float, | |
| ) -> None: | |
| """ | |
| Log a clinical escalation event. | |
| This is critical for tracking urgent patient safety events. | |
| """ | |
| timestamp = datetime.now(timezone.utc).isoformat() | |
| log_entry = { | |
| "timestamp": timestamp, | |
| "event_type": "CLINICAL_ESCALATION", | |
| "escalation_type": escalation_type, | |
| "confidence": round(confidence, 4), | |
| "request_id": request_id, | |
| "client_ip": client_ip, | |
| } | |
| self.logger.warning(f"ESCALATION_ALERT: {log_entry}") | |
| def log_security_event( | |
| self, | |
| event_type: str, | |
| client_ip: str, | |
| request_id: str, | |
| details: str, | |
| ) -> None: | |
| """ | |
| Log a security-related event (failed auth, rate limit, etc). | |
| """ | |
| timestamp = datetime.now(timezone.utc).isoformat() | |
| log_entry = { | |
| "timestamp": timestamp, | |
| "event_type": f"SECURITY_{event_type}", | |
| "client_ip": client_ip, | |
| "request_id": request_id, | |
| "details": redact_phi(details), | |
| } | |
| self.logger.warning(f"SECURITY_EVENT: {log_entry}") | |
| # ============================================================================= | |
| # SECURITY HEADERS | |
| # ============================================================================= | |
| SECURITY_HEADERS = { | |
| "X-Content-Type-Options": "nosniff", | |
| "X-Frame-Options": "DENY", | |
| "X-XSS-Protection": "1; mode=block", | |
| "Strict-Transport-Security": "max-age=31536000; includeSubDomains", | |
| "Cache-Control": "no-store, no-cache, must-revalidate, private", | |
| "Pragma": "no-cache", | |
| "Referrer-Policy": "strict-origin-when-cross-origin", | |
| "Content-Security-Policy": "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; font-src 'self'; connect-src 'self'; frame-ancestors 'none'", | |
| } | |
| # Relaxed headers for docs endpoints (allows Swagger UI to load) | |
| DOCS_SECURITY_HEADERS = { | |
| "X-Content-Type-Options": "nosniff", | |
| "X-Frame-Options": "SAMEORIGIN", | |
| "X-XSS-Protection": "1; mode=block", | |
| "Referrer-Policy": "strict-origin-when-cross-origin", | |
| } | |
| def get_security_headers(is_docs: bool = False) -> Dict[str, str]: | |
| """Get security headers for responses.""" | |
| if is_docs: | |
| return DOCS_SECURITY_HEADERS.copy() | |
| return SECURITY_HEADERS.copy() | |
| # ============================================================================= | |
| # INPUT VALIDATION FOR CLINICAL SAFETY | |
| # ============================================================================= | |
| # Words/phrases that should trigger extra caution in clinical context | |
| CLINICAL_SAFETY_KEYWORDS = { | |
| "high_urgency": [ | |
| "suicide", "suicidal", "kill myself", "end my life", "want to die", | |
| "hurt myself", "self harm", "overdose", "od'd", "not breathing", | |
| "unconscious", "unresponsive", "chest pain", "heart attack", "stroke", | |
| "seizure", "bleeding heavily", "cant breathe", "allergic reaction", | |
| ], | |
| "mental_health": [ | |
| "depressed", "depression", "anxiety", "panic attack", "voices", | |
| "hallucinating", "psychotic", "manic", "bipolar", "schizophrenia", | |
| "ptsd", "trauma", "abuse", "assault", "rape", | |
| ], | |
| "substance": [ | |
| "overdose", "withdrawal", "detox", "drunk", "high", "using", | |
| "relapse", "addiction", "alcoholic", "drugs", | |
| ], | |
| } | |
| def _has_temporal_resolution(text_lower: str, keyword: str) -> bool: | |
| """ | |
| Detect if a safety keyword appears in a resolved/historical context. | |
| Returns True if the keyword is described as past/resolved (should NOT escalate). | |
| Returns False if the keyword appears acute/current (SHOULD escalate). | |
| Safety bias: if uncertain, returns False (i.e. escalate). | |
| """ | |
| # Find keyword position for proximity-based checks | |
| kw_pos = text_lower.find(keyword) | |
| if kw_pos == -1: | |
| return False | |
| # Extract a window around the keyword for context analysis | |
| window_start = max(0, kw_pos - 80) | |
| window_end = min(len(text_lower), kw_pos + len(keyword) + 80) | |
| window = text_lower[window_start:window_end] | |
| # Past-tense / historical markers that precede the keyword | |
| past_prefixes = [ | |
| r"\bi\s+had\b", r"\bi\s+used\s+to\s+have\b", r"\bi\s+was\s+having\b", | |
| r"\bi\s+experienced\b", r"\bi\s+went\s+through\b", r"\bi\s+suffered\b", | |
| r"\bi\s+got\b", r"\bpreviously\s+had\b", r"\bhistory\s+of\b", | |
| r"\bpast\s+episode\b", r"\bformer\b", r"\bprevious\b", | |
| r"\blast\s+(week|month|year|time|night|tuesday|wednesday|thursday|friday|saturday|sunday|monday)\b", | |
| r"\byesterday\b", r"\b\d+\s+(days?|weeks?|months?|years?)\s+ago\b", | |
| r"\bback\s+in\b", r"\bwhen\s+i\s+was\b", r"\ba\s+while\s+ago\b", | |
| r"\bin\s+the\s+past\b", r"\ba\s+few\s+(days|weeks|months)\s+ago\b", | |
| r"\bearlier\s+(this|today|last)\b", r"\bused\s+to\b", | |
| r"\bhad\s+a\b", r"\bhad\s+some\b", r"\bhad\s+an\b", | |
| ] | |
| # Resolution / improvement markers anywhere in the text | |
| resolution_markers = [ | |
| r"\bfeeling\s+(much\s+)?better\b", r"\bi'?m\s+(feeling\s+)?(much\s+)?better\b", | |
| r"\bimproved\b", r"\bresolved\b", r"\bwent\s+away\b", r"\bgone\s+now\b", | |
| r"\bno\s+longer\b", r"\bstopped\b", r"\bcleared\s+up\b", | |
| r"\bfine\s+now\b", r"\bokay\s+now\b", r"\bok\s+now\b", | |
| r"\brecovered\b", r"\brecovering\b", r"\bheal(ed|ing)\b", | |
| r"\bwent\s+to\s+(the\s+)?(doctor|er|hospital|urgent\s+care|clinic)\b", | |
| r"\bsaw\s+(the\s+)?(doctor|my\s+doctor|a\s+doctor|specialist)\b", | |
| r"\bbeen\s+treated\b", r"\bgot\s+(it\s+)?checked\b", | |
| r"\bthey\s+said\s+(it'?s|i'?m)\s+(fine|ok|normal|nothing)\b", | |
| r"\bnot\s+anymore\b", r"\bdoesn'?t\s+hurt\s+(anymore|now)\b", | |
| r"\bfeeling\s+good\b", r"\bmuch\s+improved\b", | |
| r"\bunder\s+control\b", r"\bmanaged\b", r"\bstable\b", | |
| ] | |
| # "Still active" markers that OVERRIDE past tense (should still escalate) | |
| still_active_markers = [ | |
| r"\bstill\s+(have|having|feel|feeling|hurts?|experiencing)\b", | |
| r"\bgetting\s+worse\b", r"\bnot\s+getting\s+better\b", | |
| r"\bcame\s+back\b", r"\bback\s+again\b", r"\breturned\b", | |
| r"\bwon'?t\s+(stop|go\s+away)\b", r"\bkeeps?\s+coming\b", | |
| r"\bright\s+now\b", r"\bat\s+this\s+moment\b", | |
| r"\bcurrently\b", r"\bpresently\b", | |
| r"\bi'?m\s+having\b", r"\bi\s+have\b", r"\bi\s+feel\b", | |
| r"\bworried\s+(it'?s|about)\b", r"\bscared\b", | |
| r"\bworse\s+(than|today|now)\b", | |
| # BUT-clause reversals: "felt better but now the pain is back" | |
| r"\bbut\s+now\b", r"\bbut\s+today\b", r"\bbut\s+currently\b", | |
| r"\bbut\s+(it'?s?|the|i'?m|i\s+am|i\s+have|they)\b", | |
| r"\bhowever\s+(now|today|it'?s?|the|i'?m)\b", | |
| r"\bexcept\s+(now|today|it'?s?|the)\b", | |
| # Recurrence patterns | |
| r"\b(it'?s?|pain|symptoms?)\s+(is\s+)?back\b", | |
| r"\bflaring\s+up\b", r"\bacting\s+up\b", | |
| r"\bstarted\s+again\b", r"\bhappening\s+again\b", | |
| ] | |
| # Check for "still active" signals first — if present, do NOT suppress | |
| for pattern in still_active_markers: | |
| if re.search(pattern, text_lower): | |
| return False | |
| # Check for past-tense prefix near the keyword | |
| has_past = any(re.search(p, window) for p in past_prefixes) | |
| # Check for resolution markers anywhere in full text | |
| has_resolution = any(re.search(p, text_lower) for p in resolution_markers) | |
| # Need BOTH past tense AND resolution to suppress escalation | |
| # This is the conservative approach — just past tense alone is not enough | |
| return has_past and has_resolution | |
| def check_clinical_safety_keywords(text: str) -> Dict[str, List[str]]: | |
| """ | |
| Check text for clinical safety keywords with temporal context awareness. | |
| Returns dict of category -> matched keywords. | |
| Suppresses matches when keywords appear in clearly resolved/historical context. | |
| """ | |
| text_lower = text.lower() | |
| matches = {} | |
| for category, keywords in CLINICAL_SAFETY_KEYWORDS.items(): | |
| found = [] | |
| for kw in keywords: | |
| if kw in text_lower: | |
| # Check if keyword is in a resolved/historical context | |
| if _has_temporal_resolution(text_lower, kw): | |
| continue # Suppress — past/resolved | |
| found.append(kw) | |
| if found: | |
| matches[category] = found | |
| return matches | |
| # ============================================================================= | |
| # COMPLIANCE CHECKLIST | |
| # ============================================================================= | |
| def get_compliance_status() -> Dict[str, Any]: | |
| """ | |
| Get current HIPAA compliance configuration status. | |
| Returns a dict indicating which controls are enabled. | |
| """ | |
| return { | |
| "authentication_required": os.getenv("REQUIRE_AUTH", "true").lower() == "true", | |
| "tls_required": REQUIRE_TLS, | |
| "phi_logging_disabled": not LOG_PHI, | |
| "phi_redaction_enabled": REDACT_PHI_IN_LOGS, | |
| "session_timeout_configured": SESSION_TIMEOUT_SECONDS > 0, | |
| "audit_logging_enabled": True, | |
| "rate_limiting_enabled": True, | |
| "input_sanitization_enabled": True, | |
| "security_headers_configured": True, | |
| "min_tls_version": MIN_TLS_VERSION, | |
| "audit_retention_days": AUDIT_LOG_RETENTION_DAYS, | |
| } | |
| # Global audit logger instance | |
| hipaa_audit = HIPAAAuditLogger() | |