Export Pipeline Architecture¶

import hashlib
import json
from decimal import Decimal
from typing import Any


class CanonicalHasher:
    """Produces deterministic hashes from workbook cell data.

    Cells are serialized in row-major order (sheet -> row -> column)
    with type-tagged values to prevent hash collisions between
    different types that share string representations.
    """

    @staticmethod
    def serialize_cell(value: Any, cell_type: str) -> str:
        """Serialize a single cell value deterministically.

        Type tagging prevents collisions: the string "42" and
        the number 42 produce different serializations.
        """
        if value is None:
            return "null:"
        if isinstance(value, Decimal):
            # Normalize Decimal to remove trailing zeros
            value = value.normalize()
        return f"{cell_type}:{value}"

    @classmethod
    def compute_sheet_hash(cls, sheet_name: str, rows: list[dict]) -> str:
        """Compute SHA-256 hash for a single sheet."""
        hasher = hashlib.sha256()
        hasher.update(f"sheet:{sheet_name}\n".encode("utf-8"))

        for row_idx, row in enumerate(rows):
            for col_key in sorted(row.keys()):
                cell = row[col_key]
                serialized = cls.serialize_cell(
                    cell.get("value"), cell.get("type", "string")
                )
                entry = f"{row_idx}|{col_key}|{serialized}\n"
                hasher.update(entry.encode("utf-8"))

        return hasher.hexdigest()

    @classmethod
    def compute_workbook_hash(
        cls, sheets: dict[str, list[dict]]
    ) -> str:
        """Compute canonical hash across all exported sheets."""
        hasher = hashlib.sha256()
        for sheet_name in sorted(sheets.keys()):
            sheet_hash = cls.compute_sheet_hash(
                sheet_name, sheets[sheet_name]
            )
            hasher.update(f"{sheet_name}:{sheet_hash}\n".encode("utf-8"))

        return hasher.hexdigest()

import hashlib
import hmac
import time
from dataclasses import dataclass


@dataclass
class SignatureResult:
    signature: str
    algorithm: str
    timestamp: int
    key_id: str


class HMACSigner:
    """Signs canonical hashes with tenant-specific HMAC keys.

    Keys are stored encrypted in the database and retrieved
    via the EncryptionService at signing time.
    """

    ALGORITHM = "hmac-sha256"

    def __init__(self, encryption_service):
        self._encryption_service = encryption_service

    def sign(
        self,
        canonical_hash: str,
        tenant_id: str,
        key_version: int = 1,
    ) -> SignatureResult:
        """Produce an HMAC-SHA256 signature over the canonical hash."""
        key_id = f"export-signing-{tenant_id}-v{key_version}"
        signing_key = self._encryption_service.get_hmac_key(
            tenant_id, key_version
        )

        timestamp = int(time.time())
        message = f"{canonical_hash}|{tenant_id}|{timestamp}".encode("utf-8")

        signature = hmac.new(
            signing_key, message, hashlib.sha256
        ).hexdigest()

        return SignatureResult(
            signature=signature,
            algorithm=self.ALGORITHM,
            timestamp=timestamp,
            key_id=key_id,
        )

from dataclasses import dataclass, field


@dataclass
class ExportManifest:
    """Complete provenance manifest for an export."""

    version: str = "1.0"
    export_id: str = ""
    workbook_id: str = ""
    tenant_id: str = ""
    exported_by: str = ""
    exported_at: str = ""
    format: str = ""
    canonical_hash: str = ""
    signature: str = ""
    algorithm: str = ""
    key_id: str = ""
    sheets_included: list[str] = field(default_factory=list)
    row_count: int = 0
    redaction_applied: bool = False
    redacted_fields: list[str] = field(default_factory=list)
    calculation_snapshot_id: str = ""

    def to_dict(self) -> dict:
        """Serialize manifest to dictionary for embedding."""
        return {
            "provenance": {
                "version": self.version,
                "export_id": self.export_id,
                "workbook_id": self.workbook_id,
                "tenant_id": self.tenant_id,
                "exported_by": self.exported_by,
                "exported_at": self.exported_at,
                "format": self.format,
            },
            "integrity": {
                "canonical_hash": self.canonical_hash,
                "signature": self.signature,
                "algorithm": self.algorithm,
                "key_id": self.key_id,
            },
            "content": {
                "sheets": self.sheets_included,
                "row_count": self.row_count,
            },
            "security": {
                "redaction_applied": self.redaction_applied,
                "redacted_fields": self.redacted_fields,
            },
            "traceability": {
                "calculation_snapshot_id": self.calculation_snapshot_id,
            },
        }

class ProvenanceEmbedder:
    """Embeds provenance manifests into export files."""

    @staticmethod
    def embed_xlsx(workbook, manifest: ExportManifest):
        """Embed manifest as a Custom XML Part in XLSX."""
        xml_content = manifest_to_xml(manifest.to_dict())
        workbook.custom_xml_parts.append(xml_content)

    @staticmethod
    def embed_csv(output_stream, manifest: ExportManifest):
        """Prepend provenance as comment headers in CSV."""
        manifest_json = json.dumps(manifest.to_dict(), indent=None)
        output_stream.write(f"# PROVENANCE: {manifest_json}\n")
        output_stream.write(
            f"# SIGNATURE: {manifest.signature}\n"
        )
        output_stream.write(
            f"# HASH: {manifest.canonical_hash}\n"
        )

    @staticmethod
    def embed_pdf(pdf_document, manifest: ExportManifest):
        """Embed manifest in PDF document properties."""
        pdf_document.set_metadata(
            "provenance", json.dumps(manifest.to_dict())
        )
        pdf_document.set_metadata("signature", manifest.signature)
        # Add visible watermark with export ID
        add_provenance_watermark(
            pdf_document, manifest.export_id
        )

    @staticmethod
    def embed_json(data: dict, manifest: ExportManifest) -> dict:
        """Add provenance as top-level key in JSON output."""
        data["_provenance"] = manifest.to_dict()
        return data

class ExportVerificationService:
    """Verifies the integrity and authenticity of exported files."""

    def verify(self, file_path: str, format: str) -> VerificationResult:
        """Full verification pipeline.

        1. Extract embedded manifest
        2. Re-compute canonical hash from file data
        3. Verify HMAC signature against stored key
        4. Check timestamp is within acceptable window
        """
        manifest = self._extract_manifest(file_path, format)
        recomputed_hash = self._recompute_hash(file_path, format)

        hash_valid = manifest.canonical_hash == recomputed_hash
        signature_valid = self._verify_signature(manifest)
        timestamp_valid = self._check_timestamp(
            manifest.timestamp, max_age_days=365
        )

        return VerificationResult(
            is_valid=hash_valid and signature_valid and timestamp_valid,
            hash_match=hash_valid,
            signature_match=signature_valid,
            timestamp_valid=timestamp_valid,
            manifest=manifest,
        )

class PIIRedactionService:
    """Applies field-level PII redaction before export rendering.

    Redaction rules are configured per tenant and stored in the
    database. Rules can target specific columns by name or pattern.
    """

    REDACTION_STRATEGIES = {
        "mask": lambda v: "***REDACTED***",
        "partial_mask": lambda v: v[:2] + "*" * (len(str(v)) - 2) if v else "",
        "hash": lambda v: hashlib.sha256(str(v).encode()).hexdigest()[:12],
        "truncate": lambda v: str(v)[:4] + "...",
        "remove": lambda v: None,
    }

    def redact_dataframe(
        self, df: pd.DataFrame, rules: list[RedactionRule]
    ) -> tuple[pd.DataFrame, list[str]]:
        """Apply redaction rules to a DataFrame.

        Returns the redacted DataFrame and a list of affected field names.
        """
        redacted_fields = []
        df = df.copy()

        for rule in rules:
            matching_cols = self._match_columns(df.columns, rule.pattern)
            strategy = self.REDACTION_STRATEGIES[rule.strategy]

            for col in matching_cols:
                df[col] = df[col].apply(strategy)
                redacted_fields.append(col)

        return df, redacted_fields

class AntivirusScanner:
    """ClamAV integration for scanning generated export files."""

    def __init__(self, clamd_socket: str = "/var/run/clamav/clamd.ctl"):
        self._socket = clamd_socket

    def scan_file(self, file_path: str) -> ScanResult:
        """Scan a file for malware signatures.

        Returns ScanResult with status (clean/infected/error)
        and optional threat name.
        """
        cd = pyclamd.ClamdUnixSocket(self._socket)
        result = cd.scan_file(file_path)

        if result is None:
            return ScanResult(status="clean")

        status, threat = result[file_path]
        return ScanResult(
            status="infected" if status == "FOUND" else "error",
            threat_name=threat,
        )

class ZipBombDetector:
    """Detects potential zip bombs by analyzing compression ratios."""

    MAX_COMPRESSION_RATIO = 100  # 100:1 ratio threshold
    MAX_UNCOMPRESSED_SIZE = 500 * 1024 * 1024  # 500 MB
    MAX_NESTED_DEPTH = 3

    def check(self, file_path: str) -> SafetyResult:
        """Analyze a ZIP/XLSX file for zip bomb indicators."""
        with zipfile.ZipFile(file_path, "r") as zf:
            total_compressed = sum(i.compress_size for i in zf.infolist())
            total_uncompressed = sum(i.file_size for i in zf.infolist())

            if total_compressed == 0:
                return SafetyResult(safe=False, reason="Empty archive")

            ratio = total_uncompressed / total_compressed

            if ratio > self.MAX_COMPRESSION_RATIO:
                return SafetyResult(
                    safe=False,
                    reason=f"Compression ratio {ratio:.1f}:1 exceeds limit",
                )

            if total_uncompressed > self.MAX_UNCOMPRESSED_SIZE:
                return SafetyResult(
                    safe=False,
                    reason=f"Uncompressed size {total_uncompressed} exceeds limit",
                )

        return SafetyResult(safe=True)

from celery import shared_task


@shared_task(
    bind=True,
    queue="export",
    max_retries=2,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    soft_time_limit=300,
    time_limit=600,
    acks_late=True,
)
def execute_export(
    self,
    tenant_id: str,
    export_id: str,
    workbook_id: str,
    format: str,
    options: dict,
):
    """Execute a workbook export with provenance signing.

    This task runs in the 'export' queue, separate from
    calculation and parsing queues to prevent resource contention.

    Args:
        tenant_id: Tenant owning the workbook.
        export_id: Unique export job identifier.
        workbook_id: Source workbook to export.
        format: Target format (xlsx, csv, pdf, json).
        options: Export options (sheets, redaction, etc.).
    """
    update_export_status(export_id, "processing")

    try:
        # 1. Load workbook data with tenant context
        data = load_workbook_data(tenant_id, workbook_id, options)

        # 2. Apply PII redaction
        update_export_status(export_id, "redacting")
        redacted_data, redacted_fields = apply_redaction(
            tenant_id, data, options.get("redaction_profile")
        )

        # 3. Render to target format
        output_path = render_export(format, redacted_data, options)

        # 4. Sign with provenance
        update_export_status(export_id, "signing")
        signed_path = sign_export(
            output_path, format, tenant_id, export_id,
            workbook_id, redacted_fields
        )

        # 5. Antivirus scan
        update_export_status(export_id, "scanning")
        scan_result = scan_file(signed_path)

        if scan_result.status != "clean":
            update_export_status(export_id, "quarantined")
            raise SecurityError(f"AV threat: {scan_result.threat_name}")

        # 6. Store and mark complete
        store_export(export_id, signed_path)
        update_export_status(export_id, "complete")

    except SecurityError:
        raise  # Do not retry security failures
    except Exception as exc:
        update_export_status(export_id, "failed", error=str(exc))
        raise self.retry(exc=exc)

class StreamingCSVRenderer:
    """Streams CSV output in chunks to avoid loading
    entire datasets into memory.
    """

    CHUNK_SIZE = 10_000  # rows per chunk

    def render_streaming(
        self,
        data_iterator,
        output_path: str,
        columns: list[str],
    ):
        """Write CSV in chunks from a database cursor iterator."""
        with open(output_path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=columns)
            writer.writeheader()

            for chunk in self._chunked(data_iterator, self.CHUNK_SIZE):
                writer.writerows(chunk)

    @staticmethod
    def _chunked(iterator, size):
        """Yield successive chunks from an iterator."""
        chunk = []
        for item in iterator:
            chunk.append(item)
            if len(chunk) >= size:
                yield chunk
                chunk = []
        if chunk:
            yield chunk