Appearance

External Processors & Connectors ​

Use external services in Data Motion pipelines for inspection and transformation. Processors are declared globally under configuration.integrations.processors and referenced from dataMotion rules via action: "process".

What You'll Learn

  • Processor types (builtin, http, grpc, script, queue)
  • Security, privacy, and sovereignty controls
  • Timeouts, concurrency, rate limits, and offline behavior
  • Authoring examples for common integrations (DLP, OCR, AV, redaction)

Registry schema ​

Declare processors under configuration.integrations.processors.

  • Common fields
    • id: stable identifier used by pipelines (e.g., dlp-purview)
    • type: builtin | http | grpc | script | queue
    • capabilities: [inspect|transform]
    • regionAllowlist: optional list of allowed regions/sovereign tags
    • dataPolicies: payload minimization and chunking options (sendRaw | hashOnly | maxBytes | redactPatterns | chunking)
    • limits: timeoutMs, maxConcurrent, rateLimit, offlinePolicy ("deny" | "allow" | "skipProcessor")
    • audit: redactRequest, redactResponse, includeFindingsInAudit

When a processor is used from a Data Motion pipeline with mode: "observe", the runtime treats the call as observe-only: the payload for the current operation is not modified and any results are recorded as semanticFindings for audit and analytics, not as a basis for denying the operation.

  • Type-specific fields
    • builtin: name (e.g., htmlSanitize, stripFormats, watermark, blurBackground, rateLimit)
    • http: endpoint, auth (apiKey | basic | oidc | bearer), security (mTLS + pin), headers
    • grpc: target, security (mTLS + pin)
    • script: command, args[], workingDirectory
    • queue: queue (name/ARN/url), optional endpoint/auth/security

See the Schema Reference for the authoritative definition.

Security and privacy ​

  • Credentials via {{CRED:*}} variables; never bake secrets in policy.
  • mTLS with certificate pinning where possible; use regionAllowlist to enforce sovereignty.
  • Minimize payloads: prefer hashOnly or feature extraction when full content is not required; set maxBytes and redactPatterns.
  • Strict audit with redaction: emit metadata (sizes, hashes, timings, provider id) without payloads.

Failure and offline policy ​

  • Per-processor: limits.offlinePolicy controls behavior when the processor is unreachable.
  • Per-pipeline: action.process.onProcessorFailure provides a default for all steps; each step can override via step.onFailure.
  • Recommended defaults for high assurance: offlinePolicy: "deny" and onProcessorFailure: "deny".

Performance and scalability ​

External processors participate in runtime-critical paths such as clipboard, drag-and-drop, screen capture, and AI interactions. Configure them with explicit performance and capacity limits so that slow or unavailable services do not degrade user experience.

Recommended patterns:

  • Set explicit latency bounds: Use limits.timeoutMs to cap how long the Launcher waits for a processor. Avoid unbounded waits on latency-sensitive channels.
  • Control concurrency and throughput: Use limits.maxConcurrent and limits.rateLimit to protect downstream services from overload and to prevent the client from saturating local resources.
  • Choose an offline policy per integration:
    • offlinePolicy: "deny" for mandatory security gates where inspection is required before allowing data to move.
    • offlinePolicy: "skipProcessor" for optional enrichments where it is acceptable to proceed without the external result.
  • Use queue processors for heavy workloads: For large file scans, batch OCR, or multi-step classification, prefer type: "queue" processors so that inspection can be completed off the critical user path.
  • Combine with Data Motion caching: When processors feed back normalized classifications or findings, use dataMotion rules with action.process.cacheTtlSec to avoid re-submitting identical content.
  • Prefer observe mode for AI telemetry: For AI-related channels (such as ai.promptInput, ai.modelOutput, and rag.contextRetrieval), prefer mode: "observe" with type: "queue" or other asynchronous processors so that telemetry and classification run off the critical user path. Combine offlinePolicy: "skipProcessor" on the processor with onProcessorFailure: "allow" in the pipeline when observation should never block the AI interaction.

Example (queue-backed AV scan with bounded latency and concurrency):

json
{
  "configuration": {
    "integrations": {
      "processors": [
        {
          "id": "av-gateway-queue",
          "type": "queue",
          "queue": "arn:aws:sqs:us-gov-west-1:123456789012:av-scan",
          "capabilities": ["inspect"],
          "limits": {
            "timeoutMs": 3000,
            "offlinePolicy": "deny",
            "maxConcurrent": 8
          }
        }
      ]
    }
  }
}

Paired with Data Motion rules that use classifications produced by the AV service, this allows the Launcher to enforce DLP decisions without performing full content scans synchronously on every transfer.

Examples ​

HTTP DLP inspection ​

json
{
  "configuration": {
    "integrations": {
      "processors": [
        {
          "id": "dlp-purview",
          "type": "http",
          "endpoint": "https://dlp.api.corp/inspect",
          "auth": { "type": "oidc", "clientId": "{{CRED:Purv:clientId}}", "clientSecret": "{{CRED:Purv:secret}}", "tokenUrl": "https://login.corp/token" },
          "security": { "mtls": true, "pinSha256": ["ABCD…"] },
          "capabilities": ["inspect"],
          "dataPolicies": { "sendRaw": true, "maxBytes": 1048576 },
          "limits": { "timeoutMs": 2000, "offlinePolicy": "deny" }
        }
      ]
    }
  }
}

gRPC face redaction (transform) ​

json
{
  "configuration": {
    "integrations": {
      "processors": [
        {
          "id": "grpc.faceRedact",
          "type": "grpc",
          "target": "facesvc.corp:8443",
          "capabilities": ["transform"],
          "limits": { "timeoutMs": 1500, "offlinePolicy": "skipProcessor" }
        }
      ]
    }
  }
}

Queue AV scan (inspect) ​

json
{
  "configuration": {
    "integrations": {
      "processors": [
        {
          "id": "av-gateway",
          "type": "queue",
          "queue": "arn:aws:sqs:us-gov-west-1:123456789012:av-scan",
          "capabilities": ["inspect"],
          "limits": { "timeoutMs": 5000, "offlinePolicy": "deny" }
        }
      ]
    }
  }
}

Script-based OCR (inspect) ​

json
{
  "configuration": {
    "integrations": {
      "processors": [
        {
          "id": "tesseract-ocr",
          "type": "script",
          "command": "C:/tools/tesseract.exe",
          "args": ["stdin", "--oem", "1", "--psm", "3"],
          "capabilities": ["inspect"],
          "limits": { "timeoutMs": 2000, "offlinePolicy": "skipProcessor" }
        }
      ]
    }
  }
}

Using processors in rules ​

json
{
  "configuration": {
    "launch": { "runtime": { "dataMotion": {
      "rules": [
        {
          "id": "clip-out-dlp-sanitize",
          "enabled": true,
          "priority": 950,
          "channel": "clipboard",
          "direction": "out",
          "action": {
            "type": "process",
            "pipeline": [
              { "processorRef": "dlp-purview", "mode": "inspect" },
              { "processorRef": "builtin.htmlSanitize", "mode": "transform", "with": { "allowElements": [], "allowAttributes": [], "maxDepth": 5 } }
            ],
            "onProcessorFailure": "deny"
          }
        }
      ]
    } } }
  }
}

Troubleshooting ​

  • Verify health: timeouts and offlinePolicy determine observed behavior during outages.
  • Check audit records: confirm which processor ran, timings, decision, and any step-level failures.
  • Validate payload limits: if maxBytes is exceeded, the step may be skipped or denied based on limits and onFailure.

See also ​

Turbo.net © 2025