Security Boundary Configuration: Implementation Guide for eDiscovery Processing Pipelines

import hashlib
import logging
from dataclasses import dataclass
from enum import Enum
from pathlib import Path

import magic

logger = logging.getLogger(__name__)

class BoundaryAction(Enum):
    ALLOW = "allow"
    QUARANTINE = "quarantine"
    REJECT = "reject"

@dataclass(frozen=True)
class IngestionBoundaryConfig:
    max_file_size_bytes: int = 2_147_483_648  # 2GB
    allowed_mimes: frozenset[str] = frozenset([
        "application/pdf", "application/msword", 
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        "text/plain", "text/html", "message/rfc822",
        "application/zip", "application/x-tar"
    ])
    max_archive_depth: int = 3
    quarantine_path: Path = Path("/secure/quarantine")
    reject_executables: bool = True

class IngestionBoundaryValidator:
    def __init__(self, config: IngestionBoundaryConfig):
        self.config = config
        self.mime_detector = magic.Magic(mime=True)

    def evaluate(self, file_path: Path) -> tuple[BoundaryAction, str]:
        if not file_path.exists():
            return BoundaryAction.REJECT, "File not found"
        
        stat = file_path.stat()
        if stat.st_size > self.config.max_file_size_bytes:
            return BoundaryAction.QUARANTINE, "Exceeds size boundary"

        detected_mime = self.mime_detector.from_file(str(file_path))
        if detected_mime not in self.config.allowed_mimes:
            return BoundaryAction.REJECT, f"MIME mismatch: {detected_mime}"

        if self.config.reject_executables and detected_mime.startswith("application/x-executable"):
            return BoundaryAction.REJECT, "Executable payload blocked"

        # Cryptographic fingerprint for downstream chain-of-custody
        file_hash = hashlib.sha256(file_path.read_bytes()).hexdigest()
        logger.info(f"Boundary passed: {file_path.name} | SHA256: {file_hash}")
        return BoundaryAction.ALLOW, file_hash

import logging
from dataclasses import dataclass
from typing import Dict, List, Set

logger = logging.getLogger(__name__)

@dataclass(frozen=True)
class PrivilegeBoundaryRule:
    rule_id: str
    condition_field: str
    expected_values: Set[str]
    isolation_action: str  # "seal", "quarantine", "mask_metadata"

class PrivilegeBoundaryEnforcer:
    def __init__(self, rules: List[PrivilegeBoundaryRule]):
        self.rules = rules
        self.violation_log: List[Dict[str, str]] = []

    def evaluate_document(self, doc_metadata: Dict[str, str], doc_id: str) -> str:
        """Return the enforced isolation action for the document.

        When a rule matches, the document carries privileged content, so the
        configured isolation action ("seal", "quarantine", or "mask_metadata")
        is returned and the boundary trigger is logged. If no rule matches, the
        document clears the privilege boundary and "allow" is returned.
        """
        for rule in self.rules:
            actual_value = doc_metadata.get(rule.condition_field, "")
            if actual_value in rule.expected_values:
                self._log_violation(doc_id, rule, actual_value)
                return rule.isolation_action

        return "allow"

    def _log_violation(self, doc_id: str, rule: PrivilegeBoundaryRule, value: str):
        self.violation_log.append({
            "doc_id": doc_id,
            "rule_id": rule.rule_id,
            "triggered_value": value,
            "action": rule.isolation_action
        })
        logger.warning(f"Privilege boundary triggered for {doc_id} under rule {rule.rule_id}")

# Kubernetes NetworkPolicy: Explicit Pipeline Segmentation
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: ediscovery-pipeline-boundary
  namespace: ediscovery-prod
spec:
  podSelector:
    matchLabels:
      app: ediscovery-worker
  policyTypes:
  - Ingress
  - Egress
  ingress:
  - from:
    - podSelector:
        matchLabels:
          app: ediscovery-normalizer
    ports:
    - protocol: TCP
      port: 8443
  egress:
  - to:
    - podSelector:
        matchLabels:
          app: ediscovery-privilege-router
    ports:
    - protocol: TCP
      port: 8443
  - to:
    - namespaceSelector:
        matchLabels:
          kubernetes.io/metadata.name: audit-log-system
    ports:
    - protocol: TCP
      port: 514

import hmac
import hashlib
import json
from dataclasses import dataclass, asdict
from typing import List, Optional

@dataclass
class AuditEvent:
    timestamp: float
    pipeline_stage: str
    boundary_action: str
    artifact_id: str
    hash_chain_prev: Optional[str] = None
    seal: Optional[str] = None

class CryptographicAuditChain:
    def __init__(self, hmac_key: bytes):
        self.hmac_key = hmac_key
        self.last_hash: Optional[str] = None

    def _compute_seal(self, event: AuditEvent) -> str:
        # The seal binds every field except the seal itself, including the
        # previous event's hash, so the HMAC covers the full chain position.
        payload = asdict(event)
        payload.pop("seal", None)
        encoded = json.dumps(payload, sort_keys=True).encode()
        return hmac.new(self.hmac_key, encoded, hashlib.sha256).hexdigest()

    def commit_event(self, event: AuditEvent) -> AuditEvent:
        event.hash_chain_prev = self.last_hash
        event.seal = self._compute_seal(event)
        self.last_hash = event.seal
        return event

    def verify_chain(self, events: List[AuditEvent]) -> bool:
        prev_hash: Optional[str] = None
        for event in events:
            expected_seal = self._compute_seal(event)
            if event.seal != expected_seal or event.hash_chain_prev != prev_hash:
                return False
            prev_hash = event.seal
        return True