Skip to main content
INS // Insights

Healthcare Data Pipeline Development: HIPAA-Compliant at

Updated June 2026 · 7 min read

Healthcare data pipelines operate under compliance constraints that most software development ignores. HIPAA's Technical Safeguard requirements aren't guidelines — they're legal obligations with significant breach penalties. Building a healthcare data pipeline correctly means designing compliance in from the architecture stage, not bolting it on before an audit.

This post covers the technical implementation of HIPAA-compliant data pipelines for healthcare technology companies, health systems, and any organization that handles Protected Health Information (PHI).

HIPAA Technical Safeguard Requirements

The HIPAA Security Rule's Technical Safeguards (45 CFR § 164.312) require:

  • Access controls: Unique user identification, emergency access procedures, automatic logoff, and encryption/decryption mechanisms
  • Audit controls: Hardware, software, and procedural mechanisms to record and examine activity in information systems containing PHI
  • Integrity controls: Electronic mechanisms to corroborate that PHI has not been altered or destroyed in an unauthorized manner
  • Transmission security: Technical security measures to guard against unauthorized access to PHI transmitted over electronic communications networks

These requirements translate directly into specific architectural and implementation decisions.

Encryption: At Rest and In Transit

At rest: All PHI must be encrypted at rest. In AWS:

# S3 bucket with SSE-KMS for PHI storage
resource "aws_s3_bucket" "phi_storage" {
  bucket = "healthcare-phi-data-${var.environment}"
}

resource "aws_s3_bucket_server_side_encryption_configuration" "phi_storage" {
  bucket = aws_s3_bucket.phi_storage.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = "aws:kms"
      kms_master_key_id = aws_kms_key.phi_key.arn  # Customer-managed key
    }
    bucket_key_enabled = true  # Reduce KMS API call costs
  }
}

# Customer-managed KMS key for PHI encryption
resource "aws_kms_key" "phi_key" {
  description             = "PHI encryption key"
  deletion_window_in_days = 30
  enable_key_rotation     = true  # Annual automatic rotation

  policy = jsonencode({
    Statement = [
      # Restrict key use to specific IAM roles
      {
        Effect    = "Allow"
        Principal = { AWS = [aws_iam_role.pipeline_role.arn] }
        Action    = ["kms:Decrypt", "kms:GenerateDataKey"]
        Resource  = "*"
      }
    ]
  })
}

# RDS encryption for PHI database
resource "aws_db_instance" "phi_database" {
  storage_encrypted   = true
  kms_key_id          = aws_kms_key.phi_key.arn
  # ... other config
}

In transit: All PHI must be encrypted in transit. TLS 1.2 minimum (TLS 1.3 preferred). No unencrypted connections to any system that contains PHI.

# Enforce TLS for all database connections
import ssl
import psycopg2

ssl_context = ssl.create_default_context()
ssl_context.verify_mode = ssl.CERT_REQUIRED

conn = psycopg2.connect(
    host=DB_HOST,
    database=DB_NAME,
    user=DB_USER,
    password=DB_PASSWORD,
    sslmode="verify-full",  # Not "require" — "verify-full" validates the certificate
    sslrootcert="/etc/ssl/certs/rds-ca-bundle.pem",
)

PHI Data Classification and Handling

Not all health data is PHI. PHI is health information linked to an individual. De-identified data (per HIPAA Safe Harbor or Expert Determination standards) is not PHI and doesn't require the same controls.

For pipeline design, classify data explicitly:

from enum import Enum
from dataclasses import dataclass, field
from typing import Any

class DataClassification(str, Enum):
    PHI = "phi"                    # Protected Health Information
    DE_IDENTIFIED = "de_identified"  # Safe Harbor de-identified
    AGGREGATE = "aggregate"         # Aggregate statistics, no individual data
    PUBLIC = "public"               # No restrictions

@dataclass
class DataRecord:
    record_id: str
    classification: DataClassification
    data: dict

    def assert_phi_handling_controls_active(self):
        """Raise if PHI record is about to be processed without verified controls."""
        if self.classification == DataClassification.PHI:
            if not encryption_active():
                raise ComplianceError("PHI processing requires encryption to be active")
            if not audit_logging_active():
                raise ComplianceError("PHI processing requires audit logging to be active")

def de_identify_record(record: DataRecord) -> DataRecord:
    """Apply HIPAA Safe Harbor de-identification to a PHI record."""
    phi_fields_to_remove = [
        "name", "address", "city", "zip_code", "phone",
        "email", "ssn", "mrn", "dob", "admission_date",
        "discharge_date", "ip_address", "account_number",
    ]

    de_identified_data = {
        k: v for k, v in record.data.items()
        if k not in phi_fields_to_remove
    }

    # Generalize dates to year only (Safe Harbor requires year only for dates)
    if "year_of_birth" in de_identified_data:
        # Remove birth year if patient is 90+ (Safe Harbor requirement)
        if de_identified_data.get("age_years", 0) >= 90:
            del de_identified_data["year_of_birth"]

    return DataRecord(
        record_id=record.record_id,
        classification=DataClassification.DE_IDENTIFIED,
        data=de_identified_data,
    )

Audit Logging Requirements

HIPAA requires audit logs for all access to PHI. These logs must be: - Tamper-evident (users who can access PHI cannot delete their own audit records) - Retained for 6 years from creation or last effective date - Reviewed regularly for anomalous access patterns

import boto3
import json
from datetime import datetime

cloudwatch_logs = boto3.client("logs", region_name="us-east-1")

PHI_AUDIT_LOG_GROUP = "/healthcare/phi-audit"

def log_phi_access(
    user_id: str,
    user_role: str,
    action: str,  # "read", "write", "delete", "export"
    record_ids: list[str],
    pipeline_name: str,
    justification: str = None,
) -> None:
    """Log every PHI access event to tamper-evident audit log."""
    audit_event = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "user_id": user_id,
        "user_role": user_role,
        "action": action,
        "record_count": len(record_ids),
        "record_ids": record_ids[:100],  # Sample for large operations
        "pipeline": pipeline_name,
        "justification": justification,
        "source_ip": get_current_source_ip(),
    }

    cloudwatch_logs.put_log_events(
        logGroupName=PHI_AUDIT_LOG_GROUP,
        logStreamName=f"{pipeline_name}/{datetime.utcnow().strftime('%Y/%m/%d')}",
        logEvents=[{
            "timestamp": int(datetime.utcnow().timestamp() * 1000),
            "message": json.dumps(audit_event),
        }]
    )

CloudWatch Logs with a log data protection policy that restricts who can read audit logs (audit readers should not include the people being audited) satisfies the tamper-evident requirement.

Access Control Architecture

HIPAA requires role-based access control with minimum necessary access:

from enum import Enum

class PHIAccessLevel(str, Enum):
    NONE = "none"
    AGGREGATE_ONLY = "aggregate_only"
    DE_IDENTIFIED = "de_identified"
    FULL_PHI = "full_phi"

ROLE_ACCESS_MAP = {
    "data_scientist": PHIAccessLevel.DE_IDENTIFIED,
    "clinical_analyst": PHIAccessLevel.FULL_PHI,
    "billing_analyst": PHIAccessLevel.FULL_PHI,
    "marketing": PHIAccessLevel.AGGREGATE_ONLY,
    "engineering": PHIAccessLevel.NONE,  # Engineers access by exception only
}

def get_authorized_data(user_role: str, records: list[DataRecord]) -> list[DataRecord]:
    """Return records appropriate for the user's role."""
    access_level = ROLE_ACCESS_MAP.get(user_role, PHIAccessLevel.NONE)

    if access_level == PHIAccessLevel.NONE:
        raise PermissionError(f"Role {user_role} has no PHI access")
    elif access_level == PHIAccessLevel.AGGREGATE_ONLY:
        return []  # Return only aggregate queries, not individual records
    elif access_level == PHIAccessLevel.DE_IDENTIFIED:
        return [de_identify_record(r) for r in records]
    else:  # FULL_PHI
        log_phi_access(current_user_id(), user_role, "read", [r.record_id for r in records], "pipeline")
        return records

Business Associate Agreements

Any third-party service that processes PHI on your behalf requires a Business Associate Agreement (BAA). Before sending PHI to any AWS service, cloud provider, or SaaS tool: - AWS: sign the AWS BAA through the AWS Artifact console (covers most AWS services) - Other vendors: require their BAA before sending any PHI - Document all BA relationships in your HIPAA compliance documentation

AWS services covered under the AWS BAA include: S3, RDS, Lambda, SQS, CloudWatch Logs, Glue, Athena, and others. Services NOT covered include some third-party marketplace products and some emerging services — verify before use.

Rutagon Builds Healthcare Data Infrastructure

If you're building data infrastructure that handles PHI and need compliant architecture from the start, Rutagon designs and implements HIPAA-ready pipelines. Contact us at rutagon.com/contact.

See also: custom software engineering capabilities and AWS infrastructure security practices.

FAQ

Is AWS HIPAA-eligible the same as HIPAA-compliant?

No. AWS HIPAA-eligible means the service can be used to process PHI under a signed BAA — AWS provides the infrastructure controls. HIPAA compliance is the responsibility of the covered entity (your organization). You must implement the application-level controls: access controls, audit logging, encryption key management, and data handling procedures. AWS provides the foundation; you build compliance on top of it.

Do engineers who build PHI pipelines need to access real PHI?

Generally no. Well-architected PHI systems allow engineers to develop and test with synthetic or de-identified test data. Production access to PHI by engineers should be by exception only, with documented justification, additional approval, and full audit logging. If your engineering workflow requires frequent engineer access to production PHI, that's an architecture problem worth fixing.

How do you handle PHI in log files?

PHI should never appear in log files. Log identifiers (record IDs, correlation IDs) not PHI values. If existing code logs PHI, it requires remediation before deployment to production. CloudWatch Logs Data Protection policies can mask or deny access to PHI patterns in logs — useful as a defense-in-depth measure, but not a substitute for not logging PHI in the first place.

What's the minimum viable audit trail for a small healthcare startup?

CloudWatch Logs for application audit events (PHI access), CloudTrail for AWS API audit (who called what AWS API), and S3 access logging for storage-level access. Store all three in a dedicated audit account or at minimum a separate S3 bucket with Object Lock (preventing deletion). Retain for 6 years. Review for anomalous patterns monthly. This is defensible for early-stage companies; expand as the compliance program matures.

How often should PHI access logs be reviewed?

HIPAA requires regular review but doesn't specify frequency. Industry standard for covered entities: monthly for standard access patterns, immediately for anomaly alerts (access by unusual users, bulk exports, access outside business hours). Build automated anomaly detection (unexpected access volume, off-hours access, new user accessing large record sets) rather than purely manual log review.

Ready to discuss your project?

We deliver production-grade software for government, defense, and commercial clients. Let's talk about what you need.

Initiate Contact