Apache Tika Document Security and Ingestion Gate

id: tika-document-content-security-and-ingestion-gate
namespace: company.analytics

inputs:
  - id: document_url
    type: STRING
    displayName: Document URL
    description: Public or internal storage URL of the untrusted document to inspect
      and sanitize.
    defaults: "https://huggingface.co/datasets/kestra/datasets/resolve/main/pdf/app\
      _store.pdf"

  - id: declared_mime_type
    type: SELECT
    displayName: Declared MIME type
    description: The MIME type declared by the client or file extension. Tika
      verifies true content magic bytes against this value.
    values:
      - "application/pdf"
      - "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
      - "text/plain"
    defaults: "application/pdf"

  - id: min_characters_threshold
    type: INT
    displayName: Minimum extractable character threshold
    description: Minimum number of extracted text characters required for multi-page
      documents to pass without OCR escalation.
    defaults: 100

  - id: strict_mime_validation
    type: BOOL
    displayName: Enforce strict MIME matching
    description: When true, any mismatch between declared MIME type and Tika
      auto-detected MIME type triggers an immediate quarantine.
    defaults: true

  - id: quarantine_on_failure
    type: BOOL
    displayName: Quarantine on failure
    description: When true, documents that fail security or quality checks fail the
      execution gate and are barred from downstream ingestion.
    defaults: true

  - id: notify_slack
    type: BOOL
    displayName: Send Slack alert
    description: Post security quarantine alerts to Slack. Requires the
      SLACK_WEBHOOK_URL secret.
    defaults: false

triggers:
  - id: on_document_upload
    type: io.kestra.plugin.core.trigger.Webhook
    description: Real-time trigger invoked by upstream S3 bucket notification or
      upload API when a new document lands.
    key: "{{ secret('TIKA_INGESTION_WEBHOOK_KEY') }}"

  - id: scheduled_batch_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled batch sweep to inspect newly staged documents.
    cron: "0 4 * * *"
    disabled: true

tasks:
  - id: download_document
    type: io.kestra.plugin.core.http.Download
    description: Streams the untrusted document into Kestra internal storage.
    uri: "{{ inputs.document_url }}"

  - id: parse_content_and_metadata
    type: io.kestra.plugin.tika.Parse
    description: Extracts true MIME types, cryptographic encryption metadata, page
      metrics, and body text using Apache Tika.
    from: "{{ outputs.download_document.uri }}"
    contentType: TEXT
    store: false

  - id: evaluate_content_security
    type: io.kestra.plugin.scripts.python.Script
    description: Inspects Tika metadata for MIME spoofing, password encryption
      locks, ghost image scans, and metadata exposure.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install kestra
    env:
      TIKA_METADATA: "{{ outputs.parse_content_and_metadata.result.metadata | toJson }}"
      DECLARED_MIME: "{{ inputs.declared_mime_type }}"
      MIN_CHARS: "{{ inputs.min_characters_threshold }}"
      STRICT_MIME: "{{ inputs.strict_mime_validation }}"
      CONTENT_LENGTH: "{{ outputs.parse_content_and_metadata.result.content ?
        outputs.parse_content_and_metadata.result.content | length : 0 }}"
    script: |
      import os
      import json
      from kestra import Kestra

      metadata_raw = os.environ.get("TIKA_METADATA", "{}")
      try:
          metadata = json.loads(metadata_raw)
      except Exception:
          metadata = {}

      declared_mime = os.environ.get("DECLARED_MIME", "application/pdf").strip().lower()
      min_chars = int(os.environ.get("MIN_CHARS", "100"))
      strict_mime = os.environ.get("STRICT_MIME", "true").lower() == "true"
      content_len = int(os.environ.get("CONTENT_LENGTH", "0"))

      detected_mime = str(metadata.get("Content-Type", metadata.get("dc:format", "application/octet-stream"))).split(";")[0].strip().lower()

      is_encrypted = False
      for enc_key in ["pdf:encrypted", "isEncrypted", "hasOwnerPassword"]:
          val = str(metadata.get(enc_key, "")).lower()
          if val in ["true", "yes"]:
              is_encrypted = True
              break

      page_count = 1
      for page_key in ["xmpTPg:NPages", "Page-Count", "meta:page-count"]:
          if page_key in metadata:
              try:
                  page_count = int(metadata[page_key])
                  break
              except (ValueError, TypeError):
                  pass

      dangerous_mimes = [
          "application/x-dosexec", "application/x-executable", "application/x-sharedlib",
          "application/x-sh", "application/x-msdos-program", "application/x-msdownload",
          "application/x-bat", "application/x-csh", "application/javascript"
      ]

      verdict = "PASS_CLEAN"
      details = []

      if is_encrypted:
          verdict = "QUARANTINE_ENCRYPTED"
          details.append("Document is password protected or encrypted; automated RAG extraction blocked.")
      elif detected_mime in dangerous_mimes:
          verdict = "QUARANTINE_MIME_SPOOFED"
          details.append(f"Security Alert: Executable or script payload ({detected_mime}) disguised as {declared_mime}.")
      elif strict_mime and declared_mime and detected_mime != declared_mime:
          verdict = "QUARANTINE_MIME_SPOOFED"
          details.append(f"MIME Mismatch: Declared {declared_mime} but Tika auto-detected {detected_mime}.")
      elif content_len < min_chars and page_count >= 1:
          verdict = "ROUTE_OCR_REQUIRED"
          details.append(f"Ghost Scanned Document: {page_count} page(s) detected but only {content_len} characters extracted. Secondary OCR required.")
      else:
          details.append(f"Clean document: {content_len} characters extracted across {page_count} page(s), verified MIME ({detected_mime}).")

      print(f"Verdict: {verdict} - {'; '.join(details)}")

      Kestra.outputs({
          "verdict": verdict,
          "detected_mime": detected_mime,
          "page_count": page_count,
          "content_length": content_len,
          "is_encrypted": is_encrypted,
          "summary": " ".join(details)
      })

  - id: route_security_decision
    type: io.kestra.plugin.core.flow.Switch
    description: Routes the document payload, quarantine records, or OCR dispatch
      work orders based on the evaluated security verdict.
    value: "{{ outputs.evaluate_content_security.vars.verdict }}"
    cases:
      PASS_CLEAN:
        - id: stage_clean_payload
          type: io.kestra.plugin.core.storage.Write
          description: Persists clean extracted text payload to internal storage for
            downstream RAG vector database ingestion.
          extension: .txt
          content: "{{ outputs.parse_content_and_metadata.result.content }}"

        - id: log_clean_admission
          type: io.kestra.plugin.core.log.Log
          description: Logs admission verification metrics for the clean document.
          message: "Document passed content security gate. Staged {{
            outputs.evaluate_content_security.vars.content_length }} clean
            characters across {{
            outputs.evaluate_content_security.vars.page_count }} page(s) (MIME:
            {{ outputs.evaluate_content_security.vars.detected_mime }})."

      QUARANTINE_MIME_SPOOFED:
        - id: log_mime_violation
          type: io.kestra.plugin.core.log.Log
          description: Logs high-severity security alert for MIME masquerading or script
            payload injection.
          message: "SECURITY ALERT: Quarantining document {{ inputs.document_url }}. True
            MIME {{ outputs.evaluate_content_security.vars.detected_mime }}
            violates declared {{ inputs.declared_mime_type }}."

        - id: stage_spoofed_quarantine_manifest
          type: io.kestra.plugin.core.storage.Write
          description: Persists security quarantine manifest containing provenance, client
            headers, and detected payload markers.
          extension: .json
          content: |
            {
              "verdict": "QUARANTINE_MIME_SPOOFED",
              "executionId": "{{ execution.id }}",
              "sourceUrl": "{{ inputs.document_url }}",
              "declaredMime": "{{ inputs.declared_mime_type }}",
              "detectedMime": "{{ outputs.evaluate_content_security.vars.detected_mime }}",
              "summary": "{{ outputs.evaluate_content_security.vars.summary }}"
            }

        - id: check_slack_on_spoof
          type: io.kestra.plugin.core.flow.If
          description: Dispatches Slack security alert if Slack notification is enabled.
          condition: "{{ inputs.notify_slack }}"
          then:
            - id: notify_security_violation
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              description: Posts security alert to SecOps Slack channel.
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "text": "🚨 *Document Content Security Violation* | Execution `{{ execution.id }}`\n• *File:* `{{ inputs.document_url }}`\n• *Declared MIME:* `{{ inputs.declared_mime_type }}`\n• *Detected MIME:* `{{ outputs.evaluate_content_security.vars.detected_mime }}`\n• *Action:* Quarantined"
                }

      QUARANTINE_ENCRYPTED:
        - id: log_encryption_hold
          type: io.kestra.plugin.core.log.Log
          description: Logs encryption hold requiring document password decryption.
          message: "DOCUMENT HOLD: Document {{ inputs.document_url }} is encrypted or
            password-protected. Quarantining to prevent RAG extraction failure."

        - id: stage_encrypted_quarantine_manifest
          type: io.kestra.plugin.core.storage.Write
          description: Persists password retrieval manifest for client re-submission.
          extension: .json
          content: |
            {
              "verdict": "QUARANTINE_ENCRYPTED",
              "executionId": "{{ execution.id }}",
              "sourceUrl": "{{ inputs.document_url }}",
              "isEncrypted": true,
              "summary": "{{ outputs.evaluate_content_security.vars.summary }}"
            }

      ROUTE_OCR_REQUIRED:
        - id: log_ocr_escalation
          type: io.kestra.plugin.core.log.Log
          description: Logs escalation to secondary OCR processing pool for image-only
            scanned PDF.
          message: "OCR ESCALATION: Document {{ inputs.document_url }} contains {{
            outputs.evaluate_content_security.vars.page_count }} page(s) but
            insufficient extracted text ({{
            outputs.evaluate_content_security.vars.content_length }} chars).
            Dispatching to OCR queue."

        - id: stage_ocr_dispatch_work_order
          type: io.kestra.plugin.core.storage.Write
          description: Generates OCR processing work order manifest with storage pointer
            and page counts.
          extension: .json
          content: |
            {
              "verdict": "ROUTE_OCR_REQUIRED",
              "executionId": "{{ execution.id }}",
              "sourceUrl": "{{ inputs.document_url }}",
              "pageCount": {{ outputs.evaluate_content_security.vars.page_count }},
              "extractedChars": {{ outputs.evaluate_content_security.vars.content_length }},
              "internalStorageUri": "{{ outputs.download_document.uri }}"
            }

  - id: enforce_ingestion_gate
    type: io.kestra.plugin.core.flow.If
    description: Enforces pipeline gate, failing the execution if
      quarantine_on_failure is enabled and the document was not clean.
    condition: "{{ inputs.quarantine_on_failure and
      outputs.evaluate_content_security.vars.verdict != 'PASS_CLEAN' }}"
    then:
      - id: log_gate_rejection
        type: io.kestra.plugin.core.log.Log
        description: Logs final policy rejection decision before stopping execution.
        message: "Ingestion gate rejected document. Status: {{
          outputs.evaluate_content_security.vars.verdict }}. Details: {{
          outputs.evaluate_content_security.vars.summary }}"

      - id: fail_execution
        type: io.kestra.plugin.core.execution.Fail
        description: Fails the execution so upstream orchestrators and dead-letter
          monitoring catch the rejection.
        errorMessage: "Document ingestion blocked by content security gate: {{
          outputs.evaluate_content_security.vars.summary }}"

Automates pre-ingestion content security, MIME magic-byte verification, and quality triage for untrusted document uploads using Apache Tika. When users or external partners submit documents to enterprise RAG pipelines, search clusters, or compliance archives, downstream embedding models and vector databases face silent corruption or security hazards if files are password-locked, disguise malicious executables, or contain flat scanned bitmaps with zero extractable text. This blueprint evaluates every incoming file with Apache Tika, extracts genuine MIME types and cryptographic flags, classifies the payload into four operational states, and routes clean text to storage while isolating quarantined files with full audit manifests.

How it works

An event-driven Webhook trigger captures incoming document upload events from cloud storage notifications (e.g. S3 or MinIO bucket notifications), while a disabled Schedule provides periodic batch processing capabilities.
download_document (io.kestra.plugin.core.http.Download) streams the incoming file from the source URL or object storage directly into Kestra internal storage (kestra://...).
parse_content_and_metadata (io.kestra.plugin.tika.Parse) auto-detects true document MIME types, cryptographic encryption attributes, page counts, and raw body text using Apache Tika.
evaluate_content_security (io.kestra.plugin.scripts.python.Script) interrogates Tika metadata against operational policies:
- Checks for MIME masquerading by comparing declared_mime_type against the detected Content-Type magic bytes.
- Identifies password encryption locks (pdf:encrypted, isEncrypted, hasOwnerPassword) that cause downstream LLM chunking or text extraction workers to crash.
- Validates text density against min_characters_threshold and page count to isolate ghost scanned images lacking an OCR text layer.
- Emits typed Kestra metrics and an operational verdict (PASS_CLEAN, QUARANTINE_MIME_SPOOFED, QUARANTINE_ENCRYPTED, or ROUTE_OCR_REQUIRED).
route_security_decision (io.kestra.plugin.core.flow.Switch) executes branch-specific actions based on the classified verdict:
- PASS_CLEAN: Persists clean extracted text payload to internal storage via io.kestra.plugin.core.storage.Write, ready for zero-copy vectorization.
- QUARANTINE_MIME_SPOOFED: Writes a security quarantine manifest, logs a high-severity alert, and conditionally dispatches a Slack webhook notification.
- QUARANTINE_ENCRYPTED: Creates a password hold manifest so client applications can re-submit with credentials.
- ROUTE_OCR_REQUIRED: Emits an OCR dispatch work order with storage URIs for secondary OCR workers.
enforce_ingestion_gate (io.kestra.plugin.core.flow.If) evaluates quarantine_on_failure. If enabled and the document was not clean, it logs a policy rejection and triggers io.kestra.plugin.core.execution.Fail to alert monitoring systems.

What you get

Automated MIME magic-byte verification protecting downstream embedding models and parsers from disguised executables, scripts, or polyglot files.
Early detection of password-protected documents, preventing silent null-vector indexing and worker thread freezes.
Intelligent ghost-scan triage that diverts bitmap-only PDFs to OCR pipelines instead of poisoning vector stores with empty embeddings.
Structured JSON audit manifests and Kestra metrics logging character count, page count, and routing decisions.
Clean text extraction stored directly in Kestra internal storage for zero-copy handoff to vector database loaders.

Who it's for

AI and ML Platform Engineers building RAG pipelines who need to protect vector databases from corrupted or malicious document uploads.
Data Platform Engineers handling multi-tenant document ingestion from untrusted external sources.
Security and Compliance Operations teams requiring document sanitation, metadata scrubbing, and quarantine audit trails.

Why orchestrate this with Kestra

Seamless chaining of Apache Tika's native Java parser with containerized security verification and multi-path flow control in a single declarative YAML workflow.
Ephemeral internal storage management (kestra://...) that avoids saving sensitive or untrusted documents to persistent disk across pipeline steps.
Built-in auditability, execution history, and stateful triggers (Webhook and Schedule) that replace brittle custom ingest microservices.

Prerequisites

Docker daemon or container runner available to Kestra for executing the python evaluation task.
Network connectivity to download document payloads or access internal object storage endpoints.

Secrets

TIKA_INGESTION_WEBHOOK_KEY: Secret authentication key for securing the incoming upload Webhook trigger.
SLACK_WEBHOOK_URL: Optional Slack incoming webhook URL for dispatching quarantine alerts (required only if notify_slack is enabled).

Quick start

Import this blueprint into your Kestra instance.
Click Execute using the default public sample PDF (app_store.pdf) to verify clean parsing and pass-through routing.
Test failure branches by overriding declared_mime_type (e.g. setting declared_mime_type to application/vnd.openxmlformats-officedocument.wordprocessingml.document with a PDF file) or adjusting min_characters_threshold to observe automated quarantine and routing behavior.
Activate the on_document_upload webhook trigger to connect your upstream upload service.

How to extend

Connect downstream vector database upsert tasks (e.g. io.kestra.plugin.qdrant.Upsert, io.kestra.plugin.weaviate.BatchCreate, or Pinecone) in the PASS_CLEAN branch using the staged text URI.
Add Tesseract OCR execution via Kestra shell tasks in the ROUTE_OCR_REQUIRED branch to extract text from scanned images on-the-fly.
Wire PagerDuty or incident management workflows into the QUARANTINE_MIME_SPOOFED branch to alert SecOps on intentional file masquerading.

Apache Tika Untrusted Document Content Security and Ingestion Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Apache Tika Untrusted Document Content Security and Ingestion Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

Apache ActiveMQ Artemis Dead Letter Address (DLA) Message Accumulation Sentinel

Automate Customer Feedback Triage, Sentiment Analysis, and Escalation with Google Gemini and Slack

AES-256 Argon2id Cryptographic Landing and Compliance Gate

Actian Vector Table Fragmentation and Storage Sentinel

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

New to Kestra?