Developing a Data Motion Processor

This guide describes how to implement external processors (HTTP/gRPC/script/queue) that integrate with Data Motion pipelines for inspection and transformation.

What You'll Learn

• The request/response contract for inspect and transform
• Security, authentication, and reliability expectations
• Payload minimization and redaction
• Testing and troubleshooting

Processor types at a glance

• HTTP: Simple JSON over HTTPS; ideal for DLP, classification, redaction services
• gRPC: Low-latency binary RPC; ideal for image/video transforms (e.g., face redaction)
• Script: Local command execution in a controlled context; ideal for adapters like Tesseract OCR
• Queue: Asynchronous workflows (submit + poll/consume decision); ideal for AV gateways

See the registry schema: External Processors & Connectors

Common contract

Pipelines pass a payload with metadata to processors. Two modes:

• inspect: Analyze payload; emit findings (entities, severities, categories, normalized labels). No mutation required.
• transform: Return a new payload (e.g., sanitized HTML, blurred image) and optional findings.

HTTP request (example)

POST {endpoint} Headers include auth as configured (OIDC/Bearer/API key) and content-type.

Request body

{
  "mode": "inspect",                 // inspect | transform
  "channel": "clipboard",            // clipboard | dragDrop | screenCapture | screenShare | custom:*
  "direction": "out",                // in | out
  "contentType": "text/html",        // MIME type when available
  "metadata": {
    "bytes": 12345,
    "user": { "upn": "user@corp" },
    "classification": ["CUI"],
    "correlationId": "9f5c..."
  },
  "payload": "<base64>"              // optional when hashOnly=true or size > maxBytes
}

HTTP response (example)

{
  "status": "ok",                    // ok | deny | error
  "message": "blocked: ITAR detected",
  "findings": {
    "dlp": [ { "provider": "purview", "category": "ITAR", "severity": "high" } ],
    "entities": [ { "type": "Email", "value": "jane@corp", "confidence": 0.92 } ],
    "labels": ["CUI"]
  },
  "transform": {
    "contentType": "text/plain",
    "payload": "<base64>"
  }
}

Semantics

• status=deny: The step advises deny. The engine applies step.onFailure/onProcessorFailure per policy.
• For inspect: omit transform when no change is needed.
• For transform: include transform.payload; if omitted the engine treats it as no-op.

gRPC

Define a unary endpoint with a similar schema in protobuf (bytes payload, metadata, mode). Use TLS; prefer ALTS/mTLS in trusted environments.

Script

The engine invokes the command with args and passes the payload via stdin (or temp file). Return JSON on stdout using the same response schema as HTTP.

Queue

Submit a job message and block/poll within timeoutMs. On timeout, obey limits.offlinePolicy.

Security expectations

• Always use TLS for HTTP/gRPC; enable mTLS and pinning where possible.
• Credentials must be provided via {{CRED:*}} variables resolved on-device; never embed raw secrets.
• Enforce region/sovereignty via regionAllowlist when processors may egress data.
• Minimize payloads: prefer hashes or feature vectors; enforce maxBytes.

Reliability & limits

• Respect timeoutMs and rate limits; return within budget.
• Emit deterministic decisions: idempotent processing with correlationId.
• Handle chunked processing if configured (chunking.maxChunkBytes, overlapBytes).

Testing

1. Author a minimal processor config with low timeouts and clear audit.
2. Use a dedicated rule with a high priority and action.process pipeline.
3. Exercise payloads locally; inspect audit events (decision, timings, step failures).

Troubleshooting checklist

• Timeouts/offline: confirm limits.* settings and onProcessorFailure behavior.
• Auth failures: verify {{CRED:*}} variables are available on the host; check tokenUrl and scopes.
• Payload too large: adjust dataPolicies.maxBytes or implement chunking.
• Sovereign mismatch: ensure regionAllowlist matches the runtime placement/enterprise policy.

Examples

• See: External Processors & Connectors for complete examples (Purview DLP, gRPC face redaction, SQS AV scan, Tesseract OCR).