SOC 2 & HIPAA Compliance for Hosted AI Agents

# openclaw_config.py — SOC 2 & HIPAA encryption settings
import os
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

class ComplianceEncryption:
    """AES-256-GCM encryption for PHI and sensitive agent data."""

    def __init__(self):
        # Key from HSM or KMS — never hardcode
        self.key = os.environ["OPENCLAW_ENCRYPTION_KEY"]
        self.aesgcm = AESGCM(bytes.fromhex(self.key))

    def encrypt_phi(self, plaintext: bytes, context: str) -> bytes:
        """Encrypt PHI with authenticated context.

        The 'context' parameter (e.g., 'agent_memory',
        'tool_output') is bound to the ciphertext via AAD,
        preventing cross-context decryption attacks.
        """
        nonce = os.urandom(12)
        ciphertext = self.aesgcm.encrypt(
            nonce,
            plaintext,
            context.encode("utf-8"),  # AAD
        )
        return nonce + ciphertext

    def decrypt_phi(self, data: bytes, context: str) -> bytes:
        """Decrypt PHI, verifying authenticated context."""
        nonce, ciphertext = data[:12], data[12:]
        return self.aesgcm.decrypt(
            nonce,
            ciphertext,
            context.encode("utf-8"),
        )

# Usage in agent pipeline
enc = ComplianceEncryption()

# Encrypt conversation memory before persistence
encrypted_memory = enc.encrypt_phi(
    memory_bytes, context="agent_memory"
)

# Encrypt tool call output containing PHI
encrypted_output = enc.encrypt_phi(
    tool_result, context="tool_output"
)

# hermes-config.yaml — SOC 2 & HIPAA encryption
security:
  encryption:
    at_rest:
      algorithm: AES-256-GCM
      key_source: aws-kms          # or gcp-kms, azure-keyvault
      key_id: alias/hermes-phi-key
      rotate_days: 90
    in_transit:
      min_tls_version: "1.3"
      cipher_suites:
        - TLS_AES_256_GCM_SHA384
        - TLS_CHACHA20_POLY1305_SHA256
      mtls_enabled: true
      client_cert_path: /etc/hermes/certs/client.pem

  # Encrypt these data categories
  encrypt_categories:
    - conversation_history
    - agent_memory
    - tool_call_inputs
    - tool_call_outputs
    - audit_logs

  # PHI-specific settings
  phi:
    enabled: true
    auto_detect: true      # Flag fields matching PHI patterns
    redact_in_logs: true   # Redact PHI from non-audit logs
    retention_days: 2190   # 6 years per HIPAA

# rbac_policy.py — Role-based access for OpenClaw agents
from dataclasses import dataclass, field
from enum import Enum
from typing import Set

class Permission(Enum):
    READ_PHI = "read_phi"
    WRITE_PHI = "write_phi"
    TOOL_CALL = "tool_call"
    EXTERNAL_API = "external_api"
    FILE_ACCESS = "file_access"
    ADMIN = "admin"

@dataclass
class AgentRole:
    name: str
    permissions: Set[Permission]
    allowed_tools: Set[str] = field(default_factory=set)
    allowed_apis: Set[str] = field(default_factory=set)
    max_tokens_per_session: int = 10_000

# Least-privilege role definitions
ROLES = {
    "phi_reader": AgentRole(
        name="phi_reader",
        permissions={Permission.READ_PHI, Permission.TOOL_CALL},
        allowed_tools={"ehr_lookup", "drug_interaction_check"},
        allowed_apis={"ehr-api.internal"},
        max_tokens_per_session=50_000,
    ),
    "report_generator": AgentRole(
        name="report_generator",
        permissions={Permission.READ_PHI, Permission.FILE_ACCESS},
        allowed_tools={"generate_pdf", "send_secure_email"},
        allowed_apis={"smtp.internal"},
        max_tokens_per_session=100_000,
    ),
    "general_assistant": AgentRole(
        name="general_assistant",
        permissions={Permission.TOOL_CALL},
        allowed_tools={"web_search", "calculator", "calendar"},
        allowed_apis=set(),  # No external API access
        max_tokens_per_session=25_000,
    ),
}

def enforce_rbac(agent_role: str, tool_name: str) -> bool:
    """Check if agent role permits this tool call."""
    role = ROLES.get(agent_role)
    if not role:
        return False
    if Permission.TOOL_CALL not in role.permissions:
        return False
    return tool_name in role.allowed_tools

# hermes-rbac.yaml — Role-based access control
roles:
  phi_reader:
    description: "Read-only access to PHI via EHR tools"
    permissions:
      - read_phi
      - tool_call
    allowed_tools:
      - ehr_lookup
      - drug_interaction_check
    allowed_endpoints:
      - "ehr-api.internal:443"
    session_limits:
      max_tokens: 50000
      max_tool_calls: 100
      timeout_minutes: 30

  report_generator:
    description: "Generate and deliver clinical reports"
    permissions:
      - read_phi
      - file_access
    allowed_tools:
      - generate_pdf
      - send_secure_email
    data_access:
      phi_categories:
        - demographics
        - diagnoses
        - medications
      excluded:
        - ssn
        - financial

  general_assistant:
    description: "No PHI access, general tools only"
    permissions:
      - tool_call
    allowed_tools:
      - web_search
      - calculator
      - calendar
    phi_access: false  # Explicit deny

# audit_logger.py — HIPAA-compliant audit logging
import hashlib
import json
import time
from datetime import datetime, timezone
from typing import Any, Optional

class ComplianceAuditLogger:
    """Immutable, append-only audit logger for SOC 2 & HIPAA."""

    def __init__(self, storage_backend: str = "s3"):
        self.storage = storage_backend
        self.prev_hash = "GENESIS"

    def log_event(
        self,
        event_type: str,
        agent_id: str,
        user_id: str,
        resource: str,
        action: str,
        phi_accessed: bool = False,
        details: Optional[dict[str, Any]] = None,
    ) -> dict:
        """Log a compliance audit event with chain hash."""
        event = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "event_type": event_type,
            "agent_id": agent_id,
            "user_id": user_id,
            "resource": resource,
            "action": action,
            "phi_accessed": phi_accessed,
            "details": details or {},
            "prev_hash": self.prev_hash,
        }

        # Chain hash for tamper detection
        event_bytes = json.dumps(
            event, sort_keys=True
        ).encode()
        event["hash"] = hashlib.sha256(
            event_bytes
        ).hexdigest()
        self.prev_hash = event["hash"]

        # Write to immutable storage
        self._persist(event)
        return event

    def _persist(self, event: dict) -> None:
        """Write to S3 with Object Lock (WORM)."""
        # Implementation: boto3 put_object with
        # ObjectLockMode='COMPLIANCE',
        # ObjectLockRetainUntilDate=6_years_from_now
        ...

# Usage examples
logger = ComplianceAuditLogger()

# Log tool call
logger.log_event(
    event_type="tool_call",
    agent_id="agent-healthcare-01",
    user_id="dr-smith",
    resource="ehr_lookup",
    action="read",
    phi_accessed=True,
    details={
        "patient_id_hash": "sha256:abc...",
        "fields_accessed": ["diagnosis", "meds"],
    },
)

# Log model invocation
logger.log_event(
    event_type="model_invocation",
    agent_id="agent-healthcare-01",
    user_id="dr-smith",
    resource="claude-sonnet-4",
    action="inference",
    phi_accessed=True,
    details={
        "input_tokens": 2400,
        "output_tokens": 800,
        "phi_in_prompt": True,
    },
)

# hermes-audit.yaml — Compliance audit logging
audit:
  enabled: true
  level: comprehensive    # log everything

  # What to log
  events:
    - tool_calls          # every tool invocation
    - model_invocations   # every LLM call
    - memory_access       # reads/writes to agent memory
    - permission_changes  # RBAC modifications
    - data_access         # any PHI or sensitive data
    - auth_events         # login, logout, token refresh
    - config_changes      # agent config modifications

  # Tamper protection
  integrity:
    chain_hashing: true   # SHA-256 hash chain
    signature: hmac-sha256
    signing_key_source: aws-kms

  # Storage
  storage:
    backend: s3
    bucket: hermes-audit-logs-prod
    object_lock: true          # WORM compliance
    retention_mode: COMPLIANCE
    retention_years: 6         # HIPAA minimum

  # PHI handling in logs
  phi_redaction:
    enabled: true
    strategy: hash             # hash PHI identifiers
    preserve_fields:
      - patient_id_hash
      - access_timestamp
      - accessing_agent
      - action_type

# data_residency.py — Region enforcement for OpenClaw
from dataclasses import dataclass

@dataclass
class ResidencyPolicy:
    allowed_regions: list[str]
    model_endpoints: dict[str, str]
    storage_regions: dict[str, str]
    enforce_strict: bool = True

# US-only policy for HIPAA workloads
HIPAA_US_POLICY = ResidencyPolicy(
    allowed_regions=["us-east-1", "us-west-2"],
    model_endpoints={
        "anthropic": "https://api.us.anthropic.com",
        "openai": "https://api.openai.com",  # US-based
    },
    storage_regions={
        "agent_memory": "us-east-1",
        "audit_logs": "us-east-1",
        "conversation_history": "us-east-1",
        "tool_outputs": "us-west-2",  # DR region
    },
    enforce_strict=True,
)

def validate_residency(
    data_type: str,
    target_region: str,
    policy: ResidencyPolicy,
) -> bool:
    """Block any operation targeting a non-compliant region."""
    if not policy.enforce_strict:
        return True
    if target_region not in policy.allowed_regions:
        raise ResidencyViolation(
            f"{data_type} cannot be stored in "
            f"{target_region}. Allowed: "
            f"{policy.allowed_regions}"
        )
    return True

# hermes-residency.yaml — Data residency enforcement
data_residency:
  mode: strict            # block non-compliant regions
  default_region: us-east-1

  # Per-category region rules
  categories:
    agent_memory:
      primary: us-east-1
      replica: us-west-2
    audit_logs:
      primary: us-east-1
      replica: us-west-2
      cross_region_replication: true
    conversation_history:
      primary: us-east-1
    tool_outputs:
      primary: us-east-1

  # Model inference routing
  inference:
    prefer_region: us-east-1
    allowed_providers:
      - provider: anthropic
        endpoint: "https://api.us.anthropic.com"
        region: us-east-1
      - provider: openai
        endpoint: "https://api.openai.com"
        region: us-east-2

  # Enforcement
  violations:
    action: block_and_alert
    alert_channel: "#compliance-alerts"
    log_level: critical

# compliance_middleware.py — SOC 2 + HIPAA middleware
from functools import wraps
from typing import Callable, Any

class ComplianceMiddleware:
    """Wraps OpenClaw agent pipeline with compliance
    controls for SOC 2 Type II and HIPAA."""

    def __init__(
        self,
        encryption: ComplianceEncryption,
        audit: ComplianceAuditLogger,
        residency: ResidencyPolicy,
        roles: dict,
    ):
        self.enc = encryption
        self.audit = audit
        self.residency = residency
        self.roles = roles

    def compliant_tool_call(
        self, agent_role: str
    ) -> Callable:
        """Decorator for compliant tool execution."""
        def decorator(func: Callable) -> Callable:
            @wraps(func)
            def wrapper(
                agent_id: str,
                user_id: str,
                *args: Any,
                **kwargs: Any,
            ) -> Any:
                tool = func.__name__

                # 1. RBAC check
                if not enforce_rbac(agent_role, tool):
                    self.audit.log_event(
                        "access_denied",
                        agent_id, user_id,
                        tool, "blocked",
                    )
                    raise PermissionError(
                        f"Role {agent_role} cannot "
                        f"use tool {tool}"
                    )

                # 2. Data residency check
                validate_residency(
                    "tool_output",
                    self.residency.storage_regions
                        .get("tool_outputs", ""),
                    self.residency,
                )

                # 3. Execute tool
                result = func(
                    agent_id, user_id,
                    *args, **kwargs,
                )

                # 4. Encrypt output if PHI
                role = self.roles[agent_role]
                has_phi = "read_phi" in {
                    p.value for p in role.permissions
                }
                if has_phi:
                    result = self.enc.encrypt_phi(
                        result.encode(),
                        context="tool_output",
                    )

                # 5. Audit log
                self.audit.log_event(
                    "tool_call",
                    agent_id, user_id,
                    tool, "executed",
                    phi_accessed=has_phi,
                )

                return result
            return wrapper
        return decorator

# hermes-compliance.yaml — Full SOC 2 + HIPAA config
compliance:
  frameworks:
    - soc2_type_ii
    - hipaa

  # Middleware pipeline — applied to every agent action
  middleware:
    - name: rbac_check
      order: 1
      fail_action: block
    - name: residency_check
      order: 2
      fail_action: block
    - name: encryption
      order: 3
      scope: [input, output, memory]
    - name: audit_log
      order: 4
      scope: [tool_calls, model_calls, data_access]
    - name: phi_detection
      order: 5
      action: flag_and_encrypt

  # Automated compliance checks
  continuous_monitoring:
    enabled: true
    checks:
      - name: encryption_at_rest
        frequency: daily
        alert_on_failure: true
      - name: access_review
        frequency: weekly
        notify: compliance-team@company.com
      - name: log_integrity
        frequency: hourly
        verify: chain_hash
      - name: key_rotation
        frequency: daily
        max_key_age_days: 90

  # Incident response
  breach_detection:
    enabled: true
    triggers:
      - unauthorized_phi_access
      - encryption_failure
      - residency_violation
      - audit_log_gap
    notification:
      channels: [email, slack, pagerduty]
      escalation_minutes: 15
      hipaa_notification_days: 60