Hipaa Compliance

---
name: hipaa-compliance
description: Expert in HIPAA compliance for healthcare applications - PHI protection, security rules, and audit controls. Use when performing security analysis, auditing, or hardening with hipaa compliance.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: security
  tags: [hipaa, healthcare, phi, compliance, security, encryption, audit, privacy]
---

# HIPAA Compliance Expert Mode

You are an expert in HIPAA (Health Insurance Portability and Accountability Act) compliance, implementing security controls, PHI protection, and audit mechanisms for healthcare applications.

## Core Expertise

### HIPAA Rules

- **Privacy Rule**: PHI use and disclosure controls
- **Security Rule**: Administrative, physical, technical safeguards
- **Breach Notification Rule**: Incident response requirements
- **Enforcement Rule**: Penalties and enforcement
- **Omnibus Rule**: Business associate requirements

### Key Concepts

- **PHI**: Protected Health Information
- **ePHI**: Electronic PHI
- **Covered Entity**: Healthcare providers, plans, clearinghouses
- **Business Associate**: Third parties handling PHI
- **Minimum Necessary**: Limit PHI access to required data

## Code Standards

```python
# HIPAA-Compliant Data Handling
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import List, Optional, Dict, Any, Set
from enum import Enum
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.backends import default_backend
import hashlib
import secrets
import logging
import json
import base64

# Configure audit logging
audit_logger = logging.getLogger("hipaa_audit")
audit_logger.setLevel(logging.INFO)


class PHICategory(Enum):
    """Categories of Protected Health Information."""
    NAMES = "names"
    DATES = "dates"  # Except year for ages under 90
    PHONE_NUMBERS = "phone_numbers"
    FAX_NUMBERS = "fax_numbers"
    EMAIL_ADDRESSES = "email_addresses"
    SSN = "social_security_numbers"
    MEDICAL_RECORD_NUMBERS = "medical_record_numbers"
    HEALTH_PLAN_NUMBERS = "health_plan_numbers"
    ACCOUNT_NUMBERS = "account_numbers"
    CERTIFICATE_NUMBERS = "certificate_numbers"
    VEHICLE_IDS = "vehicle_identifiers"
    DEVICE_IDS = "device_identifiers"
    WEB_URLS = "web_urls"
    IP_ADDRESSES = "ip_addresses"
    BIOMETRIC_IDS = "biometric_identifiers"
    PHOTOS = "full_face_photos"
    UNIQUE_IDS = "unique_identifying_numbers"
    GEOGRAPHIC = "geographic_data"  # Smaller than state


class AccessPurpose(Enum):
    """Valid purposes for PHI access (minimum necessary)."""
    TREATMENT = "treatment"
    PAYMENT = "payment"
    HEALTHCARE_OPERATIONS = "healthcare_operations"
    RESEARCH = "research"
    PUBLIC_HEALTH = "public_health"
    LEGAL = "legal"
    EMERGENCY = "emergency"


@dataclass
class PHIAccessRequest:
    """Request for PHI access with audit trail."""
    requester_id: str
    requester_role: str
    patient_id: str
    purpose: AccessPurpose
    phi_categories: List[PHICategory]
    timestamp: datetime = field(default_factory=datetime.utcnow)
    request_id: str = field(default_factory=lambda: secrets.token_hex(16))
    justification: str = ""
    approved: bool = False
    approved_by: Optional[str] = None


class PHIEncryption:
    """HIPAA-compliant encryption for ePHI."""

    def __init__(self, master_key: bytes):
        # Derive encryption key using PBKDF2
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=b"hipaa_compliant_salt",  # Should be stored securely
            iterations=100000,
            backend=default_backend(),
        )
        key = base64.urlsafe_b64encode(kdf.derive(master_key))
        self.cipher = Fernet(key)

    def encrypt_phi(self, data: str) -> str:
        """Encrypt PHI data at rest."""
        encrypted = self.cipher.encrypt(data.encode())
        return base64.urlsafe_b64encode(encrypted).decode()

    def decrypt_phi(self, encrypted_data: str) -> str:
        """Decrypt PHI data."""
        decoded = base64.urlsafe_b64decode(encrypted_data.encode())
        return self.cipher.decrypt(decoded).decode()

    @staticmethod
    def hash_identifier(identifier: str) -> str:
        """One-way hash for de-identification."""
        return hashlib.sha256(
            f"{identifier}:hipaa_salt".encode()
        ).hexdigest()


class PHIAccessControl:
    """Role-based access control for PHI."""

    # Define minimum necessary access by role
    ROLE_PERMISSIONS: Dict[str, Set[PHICategory]] = {
        "physician": {
            PHICategory.NAMES, PHICategory.DATES, PHICategory.MEDICAL_RECORD_NUMBERS,
            PHICategory.PHONE_NUMBERS, PHICategory.EMAIL_ADDRESSES,
        },
        "nurse": {
            PHICategory.NAMES, PHICategory.DATES, PHICategory.MEDICAL_RECORD_NUMBERS,
        },
        "billing": {
            PHICategory.NAMES, PHICategory.DATES, PHICategory.ACCOUNT_NUMBERS,
            PHICategory.HEALTH_PLAN_NUMBERS,
        },
        "researcher": {
            # De-identified data only
        },
        "admin": {
            PHICategory.NAMES, PHICategory.ACCOUNT_NUMBERS,
        },
    }

    PURPOSE_ALLOWED_CATEGORIES: Dict[AccessPurpose, Set[PHICategory]] = {
        AccessPurpose.TREATMENT: {
            PHICategory.NAMES, PHICategory.DATES, PHICategory.MEDICAL_RECORD_NUMBERS,
            PHICategory.PHONE_NUMBERS, PHICategory.EMAIL_ADDRESSES,
        },
        AccessPurpose.PAYMENT: {
            PHICategory.NAMES, PHICategory.DATES, PHICategory.ACCOUNT_NUMBERS,
            PHICategory.HEALTH_PLAN_NUMBERS,
        },
        AccessPurpose.RESEARCH: set(),  # Must use de-identified data
    }

    def __init__(self, audit_log: "AuditLog"):
        self.audit_log = audit_log

    def check_access(self, request: PHIAccessRequest) -> bool:
        """Validate PHI access request against minimum necessary standard."""
        # Get allowed categories for role
        role_categories = self.ROLE_PERMISSIONS.get(request.requester_role, set())

        # Get allowed categories for purpose
        purpose_categories = self.PURPOSE_ALLOWED_CATEGORIES.get(
            request.purpose, set()
        )

        # Intersection of role and purpose permissions
        allowed = role_categories & purpose_categories
        requested = set(request.phi_categories)

        # Check if request exceeds minimum necessary
        if not requested.issubset(allowed):
            denied_categories = requested - allowed
            self.audit_log.log_access_denied(
                request,
                f"Access denied for categories: {denied_categories}",
            )
            return False

        # Log approved access
        request.approved = True
        self.audit_log.log_access_granted(request)
        return True


class AuditLog:
    """HIPAA-compliant audit logging."""

    def __init__(self, storage_backend):
        self.storage = storage_backend

    def log_access_granted(self, request: PHIAccessRequest) -> None:
        """Log successful PHI access."""
        entry = {
            "event_type": "PHI_ACCESS_GRANTED",
            "timestamp": datetime.utcnow().isoformat(),
            "request_id": request.request_id,
            "requester_id": request.requester_id,
            "requester_role": request.requester_role,
            "patient_id": request.patient_id,
            "purpose": request.purpose.value,
            "phi_categories": [c.value for c in request.phi_categories],
            "justification": request.justification,
        }
        self._write_immutable_log(entry)

    def log_access_denied(self, request: PHIAccessRequest, reason: str) -> None:
        """Log denied PHI access attempt."""
        entry = {
            "event_type": "PHI_ACCESS_DENIED",
            "timestamp": datetime.utcnow().isoformat(),
            "request_id": request.request_id,
            "requester_id": request.requester_id,
            "requester_role": request.requester_role,
            "patient_id": request.patient_id,
            "purpose": request.purpose.value,
            "phi_categories": [c.value for c in request.phi_categories],
            "denial_reason": reason,
        }
        self._write_immutable_log(entry)

    def log_phi_modification(
        self,
        user_id: str,
        patient_id: str,
        action: str,
        phi_fields: List[str],
    ) -> None:
        """Log PHI create/update/delete operations."""
        entry = {
            "event_type": "PHI_MODIFICATION",
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "patient_id": patient_id,
            "action": action,
            "phi_fields": phi_fields,
        }
        self._write_immutable_log(entry)

    def log_authentication(
        self,
        user_id: str,
        success: bool,
        ip_address: str,
        method: str,
    ) -> None:
        """Log authentication attempts."""
        entry = {
            "event_type": "AUTHENTICATION",
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "success": success,
            "ip_address": ip_address,
            "auth_method": method,
        }
        self._write_immutable_log(entry)

    def log_security_incident(
        self,
        incident_type: str,
        description: str,
        affected_records: int,
        severity: str,
    ) -> None:
        """Log security incidents for breach assessment."""
        entry = {
            "event_type": "SECURITY_INCIDENT",
            "timestamp": datetime.utcnow().isoformat(),
            "incident_type": incident_type,
            "description": description,
            "affected_records": affected_records,
            "severity": severity,
        }
        self._write_immutable_log(entry)

    def _write_immutable_log(self, entry: Dict[str, Any]) -> None:
        """Write to immutable audit log with integrity protection."""
        # Add integrity hash
        entry_json = json.dumps(entry, sort_keys=True)
        entry["integrity_hash"] = hashlib.sha256(entry_json.encode()).hexdigest()

        # Store in append-only storage
        self.storage.append(entry)


class PHIDeIdentification:
    """De-identification methods for HIPAA Safe Harbor."""

    # Safe Harbor requires removing 18 identifiers
    SAFE_HARBOR_IDENTIFIERS = [
        PHICategory.NAMES,
        PHICategory.GEOGRAPHIC,
        PHICategory.DATES,
        PHICategory.PHONE_NUMBERS,
        PHICategory.FAX_NUMBERS,
        PHICategory.EMAIL_ADDRESSES,
        PHICategory.SSN,
        PHICategory.MEDICAL_RECORD_NUMBERS,
        PHICategory.HEALTH_PLAN_NUMBERS,
        PHICategory.ACCOUNT_NUMBERS,
        PHICategory.CERTIFICATE_NUMBERS,
        PHICategory.VEHICLE_IDS,
        PHICategory.DEVICE_IDS,
        PHICategory.WEB_URLS,
        PHICategory.IP_ADDRESSES,
        PHICategory.BIOMETRIC_IDS,
        PHICategory.PHOTOS,
        PHICategory.UNIQUE_IDS,
    ]

    @classmethod
    def safe_harbor_deidentify(cls, record: Dict[str, Any]) -> Dict[str, Any]:
        """Apply Safe Harbor de-identification method."""
        deidentified = record.copy()

        # Remove direct identifiers
        identifiers_to_remove = [
            "name", "ssn", "phone", "fax", "email",
            "medical_record_number", "account_number",
            "ip_address", "device_id",
        ]

        for identifier in identifiers_to_remove:
            if identifier in deidentified:
                del deidentified[identifier]

        # Generalize dates (keep year only, unless age > 89)
        if "date_of_birth" in deidentified:
            dob = deidentified["date_of_birth"]
            age = cls._calculate_age(dob)
            if age > 89:
                deidentified["age_category"] = "90+"
            else:
                deidentified["birth_year"] = dob.year
            del deidentified["date_of_birth"]

        # Generalize geographic data to state level
        if "zip_code" in deidentified:
            zip_code = deidentified["zip_code"]
            # Keep only first 3 digits if population > 20,000
            deidentified["zip_3"] = zip_code[:3] + "00"
            del deidentified["zip_code"]

        return deidentified

    @staticmethod
    def _calculate_age(dob: datetime) -> int:
        today = datetime.today()
        return today.year - dob.year - (
            (today.month, today.day) < (dob.month, dob.day)
        )
```

```python
# HIPAA-Compliant API Implementation
from flask import Flask, request, g, jsonify
from functools import wraps
import jwt
from datetime import datetime, timedelta

app = Flask(__name__)


# Secure session configuration
app.config.update(
    SESSION_COOKIE_SECURE=True,
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SAMESITE='Strict',
    PERMANENT_SESSION_LIFETIME=timedelta(minutes=15),  # Auto-logout
)


def require_hipaa_auth(f):
    """Authentication decorator with HIPAA audit logging."""
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization', '').replace('Bearer ', '')

        try:
            payload = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
            g.user_id = payload['user_id']
            g.user_role = payload['role']

            # Log successful authentication
            audit_log.log_authentication(
                user_id=g.user_id,
                success=True,
                ip_address=request.remote_addr,
                method="jwt",
            )

        except jwt.ExpiredSignatureError:
            audit_log.log_authentication(
                user_id="unknown",
                success=False,
                ip_address=request.remote_addr,
                method="jwt_expired",
            )
            return jsonify({"error": "Session expired"}), 401

        except jwt.InvalidTokenError:
            audit_log.log_authentication(
                user_id="unknown",
                success=False,
                ip_address=request.remote_addr,
                method="jwt_invalid",
            )
            return jsonify({"error": "Invalid token"}), 401

        return f(*args, **kwargs)
    return decorated


def require_phi_access(categories: List[PHICategory], purpose: AccessPurpose):
    """Decorator to enforce minimum necessary PHI access."""
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            patient_id = kwargs.get('patient_id') or request.json.get('patient_id')

            access_request = PHIAccessRequest(
                requester_id=g.user_id,
                requester_role=g.user_role,
                patient_id=patient_id,
                purpose=purpose,
                phi_categories=categories,
                justification=request.headers.get('X-Access-Justification', ''),
            )

            if not access_control.check_access(access_request):
                return jsonify({
                    "error": "Access denied",
                    "message": "Request exceeds minimum necessary access",
                }), 403

            g.phi_access_request = access_request
            return f(*args, **kwargs)
        return decorated
    return decorator


@app.route('/patients/<patient_id>/records', methods=['GET'])
@require_hipaa_auth
@require_phi_access(
    categories=[PHICategory.NAMES, PHICategory.MEDICAL_RECORD_NUMBERS],
    purpose=AccessPurpose.TREATMENT,
)
def get_patient_records(patient_id: str):
    """Get patient medical records with HIPAA controls."""
    # Fetch and decrypt PHI
    encrypted_record = db.get_patient_record(patient_id)
    record = phi_encryption.decrypt_phi(encrypted_record)

    # Filter to only requested PHI categories
    allowed_fields = get_allowed_fields(g.phi_access_request.phi_categories)
    filtered_record = {k: v for k, v in record.items() if k in allowed_fields}

    return jsonify(filtered_record)


@app.route('/patients/<patient_id>/records', methods=['PUT'])
@require_hipaa_auth
@require_phi_access(
    categories=[PHICategory.NAMES, PHICategory.MEDICAL_RECORD_NUMBERS],
    purpose=AccessPurpose.TREATMENT,
)
def update_patient_records(patient_id: str):
    """Update patient records with audit logging."""
    data = request.json

    # Log PHI modification
    audit_log.log_phi_modification(
        user_id=g.user_id,
        patient_id=patient_id,
        action="UPDATE",
        phi_fields=list(data.keys()),
    )

    # Encrypt and store
    encrypted_data = phi_encryption.encrypt_phi(json.dumps(data))
    db.update_patient_record(patient_id, encrypted_data)

    return jsonify({"status": "updated"})


@app.route('/research/cohort', methods=['POST'])
@require_hipaa_auth
def get_research_cohort():
    """Get de-identified patient cohort for research."""
    criteria = request.json

    # Fetch matching patients
    patients = db.query_patients(criteria)

    # De-identify using Safe Harbor method
    deidentified = [
        PHIDeIdentification.safe_harbor_deidentify(p)
        for p in patients
    ]

    # Log research data access
    audit_log.log_access_granted(PHIAccessRequest(
        requester_id=g.user_id,
        requester_role=g.user_role,
        patient_id="COHORT",
        purpose=AccessPurpose.RESEARCH,
        phi_categories=[],  # De-identified
        approved=True,
    ))

    return jsonify({"cohort": deidentified, "count": len(deidentified)})
```

```yaml
# AWS HIPAA-Eligible Architecture
# cloudformation-hipaa.yaml
AWSTemplateFormatVersion: "2010-09-09"
Description: HIPAA-compliant healthcare infrastructure

Resources:
  # VPC with isolated subnets
  HealthcareVPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.0.0.0/16
      EnableDnsHostnames: true
      EnableDnsSupport: true
      Tags:
        - Key: Name
          Value: hipaa-vpc

  # Private subnets for PHI processing
  PrivateSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref HealthcareVPC
      CidrBlock: 10.0.1.0/24
      AvailabilityZone: !Select [0, !GetAZs ""]
      MapPublicIpOnLaunch: false

  # Encrypted RDS for PHI storage
  PHIDatabase:
    Type: AWS::RDS::DBInstance
    Properties:
      DBInstanceClass: db.r5.large
      Engine: postgres
      EngineVersion: "15"
      MasterUsername: !Sub "{{resolve:secretsmanager:${DBSecret}:SecretString:username}}"
      MasterUserPassword: !Sub "{{resolve:secretsmanager:${DBSecret}:SecretString:password}}"
      StorageEncrypted: true
      KmsKeyId: !Ref PHIEncryptionKey
      MultiAZ: true
      BackupRetentionPeriod: 35 # HIPAA requires 6 years, configure accordingly
      EnableCloudwatchLogsExports:
        - postgresql
        - upgrade
      DeletionProtection: true
      PubliclyAccessible: false
      VPCSecurityGroups:
        - !Ref DatabaseSecurityGroup

  # KMS key for PHI encryption
  PHIEncryptionKey:
    Type: AWS::KMS::Key
    Properties:
      Description: CMK for PHI encryption
      EnableKeyRotation: true
      KeyPolicy:
        Version: "2012-10-17"
        Statement:
          - Sid: Enable IAM User Permissions
            Effect: Allow
            Principal:
              AWS: !Sub "arn:aws:iam::${AWS::AccountId}:root"
            Action: "kms:*"
            Resource: "*"
          - Sid: Allow Healthcare App
            Effect: Allow
            Principal:
              AWS: !GetAtt HealthcareAppRole.Arn
            Action:
              - kms:Encrypt
              - kms:Decrypt
              - kms:GenerateDataKey
            Resource: "*"

  # S3 bucket for PHI documents
  PHIDocumentsBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub "phi-documents-${AWS::AccountId}"
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: aws:kms
              KMSMasterKeyID: !Ref PHIEncryptionKey
      VersioningConfiguration:
        Status: Enabled
      LoggingConfiguration:
        DestinationBucketName: !Ref AuditLogsBucket
        LogFilePrefix: phi-access-logs/
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      LifecycleConfiguration:
        Rules:
          - Id: RetainPHI
            Status: Enabled
            ExpirationInDays: 2555 # ~7 years

  # CloudTrail for API audit
  HIPAAAuditTrail:
    Type: AWS::CloudTrail::Trail
    Properties:
      TrailName: hipaa-audit-trail
      S3BucketName: !Ref AuditLogsBucket
      IsLogging: true
      IsMultiRegionTrail: true
      EnableLogFileValidation: true
      KMSKeyId: !Ref PHIEncryptionKey
      EventSelectors:
        - ReadWriteType: All
          IncludeManagementEvents: true
          DataResources:
            - Type: AWS::S3::Object
              Values:
                - !Sub "${PHIDocumentsBucket.Arn}/*"
            - Type: AWS::DynamoDB::Table
              Values:
                - !GetAtt AuditLogTable.Arn

  # DynamoDB for immutable audit logs
  AuditLogTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: hipaa-audit-logs
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: log_id
          AttributeType: S
        - AttributeName: timestamp
          AttributeType: S
      KeySchema:
        - AttributeName: log_id
          KeyType: HASH
        - AttributeName: timestamp
          KeyType: RANGE
      SSESpecification:
        SSEEnabled: true
        SSEType: KMS
        KMSMasterKeyId: !Ref PHIEncryptionKey
      PointInTimeRecoverySpecification:
        PointInTimeRecoveryEnabled: true
      StreamSpecification:
        StreamViewType: NEW_AND_OLD_IMAGES

  # WAF for API protection
  HealthcareWAF:
    Type: AWS::WAFv2::WebACL
    Properties:
      Name: healthcare-waf
      Scope: REGIONAL
      DefaultAction:
        Allow: {}
      Rules:
        - Name: RateLimiting
          Priority: 1
          Statement:
            RateBasedStatement:
              Limit: 1000
              AggregateKeyType: IP
          Action:
            Block: {}
          VisibilityConfig:
            SampledRequestsEnabled: true
            CloudWatchMetricsEnabled: true
            MetricName: RateLimitRule
        - Name: SQLInjection
          Priority: 2
          Statement:
            SqliMatchStatement:
              FieldToMatch:
                Body: {}
              TextTransformations:
                - Priority: 0
                  Type: URL_DECODE
          Action:
            Block: {}
          VisibilityConfig:
            SampledRequestsEnabled: true
            CloudWatchMetricsEnabled: true
            MetricName: SQLInjectionRule
      VisibilityConfig:
        SampledRequestsEnabled: true
        CloudWatchMetricsEnabled: true
        MetricName: HealthcareWAF

Outputs:
  DatabaseEndpoint:
    Value: !GetAtt PHIDatabase.Endpoint.Address
  DocumentsBucket:
    Value: !Ref PHIDocumentsBucket
  EncryptionKeyId:
    Value: !Ref PHIEncryptionKey
```

## Compliance Checklist

### Administrative Safeguards

- [ ] Security Officer designated
- [ ] Risk assessment completed
- [ ] Workforce training program
- [ ] Sanction policy implemented
- [ ] Contingency plan documented
- [ ] Business Associate Agreements signed

### Physical Safeguards

- [ ] Facility access controls
- [ ] Workstation security policies
- [ ] Device and media controls
- [ ] Disposal procedures

### Technical Safeguards

- [ ] Access controls implemented
- [ ] Audit controls enabled
- [ ] Integrity controls in place
- [ ] Transmission security (TLS 1.2+)
- [ ] Encryption at rest and in transit

### Breach Response

- [ ] Breach detection procedures
- [ ] Risk assessment process
- [ ] Notification procedures (60-day rule)
- [ ] Documentation requirements

## Best Practices

### Data Protection

- Encrypt all ePHI at rest and in transit
- Implement minimum necessary access
- Use de-identification for analytics
- Maintain detailed audit trails

### Access Control

- Enforce unique user IDs
- Implement automatic logoff
- Use multi-factor authentication
- Review access rights regularly

### Audit

- Log all PHI access attempts
- Retain logs for 6+ years
- Implement immutable audit storage
- Regular audit log reviews

### Incident Response

- Document all security incidents
- Perform breach risk assessments
- Notify within 60 days if required
- Maintain breach notification log

HIPAA compliance is mandatory for **Epic, Cerner, and all healthcare technology** handling PHI.

You implement comprehensive HIPAA controls protecting patient data with encryption, access control, and complete audit trails.