Transform Middlewares

TransformMiddleware lets you run logic before and after every transform execution. Typical use cases include logging, request shaping, per-user concurrency limiting, shared execution context, audit trails, and OAuth token handling.

Lifecycle

Every transform execution goes through the same middleware lifecycle:

1. before_transform(...) runs before the transform starts.
2. The transform executes.
3. after_transform(...) runs with the output entities, final execution state, and any collected exceptions.

By default after_transform(...) also runs on failures because TransformMiddleware.call_on_failure defaults to True. Set call_on_failure = False on your middleware if you only want the after-hook for successful runs.

Timeouts

Both middleware hooks use MaltegoServerSettings.middleware_execution_timeout. This budget applies to each middleware call individually, not to the whole chain -- if one middleware's before_transform exceeds it, the SDK raises a transform exception naming that middleware and aborts the execution; other middlewares each get their own fresh budget. The transform function itself has a separate budget, transform_execution_timeout.

Where this fits in the request chain

TransformMiddleware is not the first thing that sees a request. For a transform-execution call, in order:

1. Your reverse proxy/gateway (TLS, WAF, rate limiting) and the ASGI/uvicorn layer reject malformed requests and unmatched routes before any SDK code runs. See Deploying Behind a Reverse Proxy.
2. If AuthSettings.enabled is set, the SDK's authentication dependency validates the JWT/OIDC/SAML credential and populates context.identity / context.auth_claims. This runs before the transform is even scheduled -- in strict mode, a rejected credential means before_transform never runs at all. See Authentication.
3. Only then does before_transform run for each configured middleware, in this order: built-ins first (UserConcurrencyLimitMiddleware if configured, then VerifyMetadataMiddleware, then OAuthMiddleware if an OAuth authenticator is registered), then your transform_middlewares in the order you passed them to run_server.
4. The transform function.
5. after_transform for each middleware, same order as step 3.

Because authentication has already run by step 3, context.identity and context.auth_claims are available and trustworthy inside before_transform/after_transform -- this is what makes TransformMiddleware the right place for authorization decisions (is this already-authenticated caller allowed to run this transform), as opposed to authentication (is this credential valid at all), which belongs in AuthSettings instead. See "Example: decoupling authorization, auditing, and logging" below for a worked pattern.

Creating a middleware

from typing import Any, Dict, List, Optional, Sequence, Union
from maltego.middlewares.middlewares import TransformMiddleware
from maltego.model.context import MaltegoContext
from maltego.model.entity import MaltegoEntity
from maltego.model.graph import MaltegoGraph
from maltego.model.transform import MaltegoTransform
from maltego.model.types import ExecutionState, MaltegoSettingTypes


class LoggingMiddleware(TransformMiddleware):
    async def before_transform(
        self,
        transform: MaltegoTransform,
        transform_input: Union[MaltegoEntity, List[MaltegoEntity], MaltegoGraph[Any]],
        properties: Dict[str, MaltegoSettingTypes],
        context: MaltegoContext,
        soft_limit: int,
        hard_limit: int,
    ) -> None:
        context.log.inform(f"Starting {transform.display_name}")

    async def after_transform(
        self,
        transform: MaltegoTransform,
        transform_input: Union[MaltegoEntity, List[MaltegoEntity], MaltegoGraph[Any]],
        output_entities: List[MaltegoEntity],
        context: MaltegoContext,
        state: ExecutionState,
        exceptions: Optional[Sequence[Exception]] = None,
    ) -> None:
        context.log.inform(
            f"Finished {transform.display_name} with state {state} and "
            f"{len(output_entities)} output entities"
        )

Registering middlewares

Pass middleware instances to run_server:

run_server(
    settings=settings,
    transform_middlewares=[LoggingMiddleware()],
)

run_server accepts a transform_middlewares parameter (a list of TransformMiddleware instances) alongside its other server-startup options. Configure HTTP, HTTPS, and certificate paths on ServerHTTPSettings inside MaltegoServerSettings so server startup uses a single HTTP configuration source. Middleware instances passed here are used to add additional logic right before or right after each transform runs.

Built-in middleware behavior

The SDK already wires in a few middleware behaviors for you:

• VerifyMetadataMiddleware is added by default.
• OAuthMiddleware is added automatically when the first OAuth authenticator is registered.
• UserConcurrencyLimitMiddleware is added when max_concurrent_transforms_per_user is configured.

See the Execution Runner article for how middleware execution fits into the wider transform execution lifecycle.

Example: decoupling authorization, auditing, and logging

A common request is to keep authorization, auditing, and logging out of transform business logic, and to be able to swap the systems behind them (an RBAC service, Kafka, Splunk, ...) without touching every transform.

The pattern: keep each middleware thin, and have it delegate to a small adapter class built once at startup. Swapping systems later means changing the adapter, not the middleware or the transforms:

import logging
from maltego.middlewares.middlewares import TransformMiddleware
from maltego.model.exception import MaltegoException
from maltego.server import AuthSettings, run_server

logger = logging.getLogger(__name__)

# Adapters -- the only thing that changes when you swap systems.
class PolicyChecker:                 # back this with OPA, your RBAC service, etc.
    async def allowed(self, *, identity, claims) -> bool: ...

class AuditWriter:                   # today Kafka, tomorrow Splunk, same interface
    async def record(self, *, identity, outcome): ...

# Thin middlewares: they read the context and delegate to an adapter.
class AuthorizationMiddleware(TransformMiddleware):
    def __init__(self, policy):
        self.policy = policy

    async def before_transform(self, transform, transform_input, properties,
                                context, soft_limit, hard_limit):
        # context.identity/context.auth_claims are already populated here --
        # authentication ran before this middleware, not inside it.
        if not await self.policy.allowed(identity=context.identity,
                                          claims=context.auth_claims):
            raise MaltegoException("Not authorised")

class AuditMiddleware(TransformMiddleware):
    call_on_failure = True           # explicit here for clarity; it's the default

    def __init__(self, audit):
        self.audit = audit

    async def after_transform(self, transform, transform_input, output_entities,
                               context, state, exceptions=None):
        logger.info("transform finished, outcome=%s", state)
        await self.audit.record(identity=context.identity, outcome=state)

# Wire it once at server startup. Swap the adapter, not the middleware.
# `settings` here is the MaltegoServerSettings from earlier in this article.
run_server(
    settings=settings,
    auth_settings=AuthSettings(enabled=True, token_origin="sso", provider_type="oidc",
                                provider_url="https://id.example.com/..."),
    transform_middlewares=[
        AuthorizationMiddleware(PolicyChecker()),
        AuditMiddleware(AuditWriter()),
    ],
)

A few things worth calling out:

• AuthorizationMiddleware and AuditMiddleware never call an identity provider or parse a token themselves -- that already happened in the auth layer (see "Where this fits in the request chain" above). They only make policy/audit decisions from an already-resolved identity.
• AuditMiddleware.call_on_failure = True means the audit record and log line are written even when the transform raises, since after_transform still runs on failure by default.
• If an adapter talks to an HTTP API, the Integration Client article's client gives you retries, rate limiting, and concurrency control for it. For queues, caches, or databases, use your usual client libraries.
• Network-level concerns -- rate limiting, WAF rules, mTLS, protocol-level rejection -- belong at your reverse proxy/gateway, not in a TransformMiddleware. See Deploying Behind a Reverse Proxy.

A runnable version of this pattern, including a demo transform that reads context.identity, ships in the generated starter project as transforms/middleware_example.py -- see Installing and Setting Up the SDK.

API reference

TransformMiddleware (maltego.middlewares.middlewares.TransformMiddleware) is an abstract base class. To create a middleware, subclass it and implement both abstract methods below.

For the full class reference (including MaltegoTransform, MaltegoContext, MaltegoEntity, and MaltegoGraph), see the SDK API Reference article.