helixordevelopers
Tutorials

Guard an existing API

You add one ASGI middleware class in front of an existing Python web API. It evaluates every JSON request body against a pack before your handlers see it: a fatal decision is rejected with 422, a remedied body is forwarded in its place, and each response carries the decision's receipt in a header. With the built-in example pack (data protection), fatal data is rejected and contact details are redacted in place.

Tier 040 minutesAdvancedPython 3.9+

About the example

This tutorial uses the built-in data-protection pack so you can run everything without writing a pack first. The same calls work for any decision pack: eligibility, limits, routing and so on. See Core concepts.

What you'll build#

  • helixor_guard.py: a HelixorGuardMiddleware class that works with any ASGI application.
  • app.py: a plain ASGI echo app standing in for your API.
  • test_middleware.py: tests that drive the middleware in memory, with no server.
Request bodyHandler seesClient gets
CleanThe body unchangedYour response + X-Helixor-Receipt, X-Helixor-Action
Contact detailsThe body with those values redactedYour response + both headers
Fatal dataNothing; the handler is not called422 with action, rule IDs and receipt
Not JSON, or a GETThe request unchangedYour response, no headers

Prerequisites#

  • Make your first decision.
  • An ASGI server (uvicorn is installed with the runtime) and, for the tests, pytest and httpx.
  • Familiarity with the ASGI interface: scope, receive and send.

Steps#

  1. Choose what to screen

    The middleware screens POST, PUT and PATCH requests with a JSON content type, which is where user-supplied data enters an API. Everything else passes straight through. It makes one decision per request by evaluating the whole body serialized as JSON, so every request gets exactly one receipt and no field can be missed.

    Why not pass the dict to evaluate()?

    evaluate(dict) reads only the text, message or payload key and ignores the rest. An API body can carry what the pack must see in any field, so the middleware serializes the whole body instead.

  2. Write the middleware 
    """ASGI middleware that screens JSON request bodies with the Decision Runtime."""
import json

from helixor_runtime import HelixorEngine

GUARDED_METHODS = {"POST", "PUT", "PATCH"}
MAX_BODY_BYTES = 1_048_576


class HelixorGuardMiddleware:
    def __init__(self, app, engine=None, max_body_bytes=MAX_BODY_BYTES):
        self.app = app
        self.engine = engine or HelixorEngine()
        self.max_body_bytes = max_body_bytes

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http" or scope["method"] not in GUARDED_METHODS:
            return await self.app(scope, receive, send)
        headers = dict(scope["headers"])
        if not headers.get(b"content-type", b"").startswith(b"application/json"):
            return await self.app(scope, receive, send)

        # 1. Read the whole body (bounded).
        body = b""
        more = True
        while more:
            message = await receive()
            body += message.get("body", b"")
            more = message.get("more_body", False)
            if len(body) > self.max_body_bytes:
                return await _json(send, 413, {"error": "body_too_large"})

        try:
            payload = json.loads(body or b"null")
        except ValueError:
            return await _json(send, 400, {"error": "invalid_json"})

        # 2. One decision for the whole request.
        decision = self.engine.evaluate(json.dumps(payload, ensure_ascii=False))
        receipt = decision.receipt_hash
        rule_ids = [t.rule_id for t in decision.triggers]

        if decision.action.startswith("block"):
            return await _json(send, 422, {
                "error": "blocked_by_policy",
                "action": decision.action,
                "rule_ids": rule_ids,
                "receipt_hash": receipt,
            }, receipt)

        # 3. Repair each string value when the decision says redact.
        if not decision.invariants_passed:
            payload = self._redact(payload)
            body = json.dumps(payload, ensure_ascii=False).encode()

        # 4. Hand the (possibly repaired) body to the app, tag the response.
        scope = dict(scope)
        scope["headers"] = [(k, v) for k, v in scope["headers"] if k != b"content-length"]
        scope["headers"].append((b"content-length", str(len(body)).encode()))

        sent = False

        async def replay():
            nonlocal sent
            if sent:
                return {"type": "http.disconnect"}
            sent = True
            return {"type": "http.request", "body": body, "more_body": False}

        async def send_with_receipt(message):
            if message["type"] == "http.response.start":
                message = dict(message)
                message["headers"] = list(message.get("headers", [])) + [
                    (b"x-helixor-receipt", receipt.encode()),
                    (b"x-helixor-action", decision.action.encode()),
                ]
            await send(message)

        await self.app(scope, replay, send_with_receipt)

    def _redact(self, value):
        if isinstance(value, str):
            d = self.engine.evaluate(value)
            if d.action.startswith("block"):
                # The whole-body decision said redact; a block here means the
                # two disagree. Fail closed.
                raise RuntimeError(f"field-level block {d.action} after body-level redact")
            return d.remedy.clean_text
        if isinstance(value, list):
            return [self._redact(v) for v in value]
        if isinstance(value, dict):
            return {k: self._redact(v) for k, v in value.items()}
        return value


async def _json(send, status, body, receipt=None):
    data = json.dumps(body).encode()
    headers = [(b"content-type", b"application/json"),
               (b"content-length", str(len(data)).encode())]
    if receipt:
        headers.append((b"x-helixor-receipt", receipt.encode()))
    await send({"type": "http.response.start", "status": status, "headers": headers})
    await send({"type": "http.response.body", "body": data})

    Three details matter:

    • The body is bounded. A request over max_body_bytes gets 413 before it is evaluated.
    • The remedy is applied per string value. Replacing text inside the serialized body could break the JSON, so the middleware walks the parsed body and replaces each string with its own clean_text. Keys and non-string values are left alone.
    • It fails closed. Invalid JSON is rejected with 400. If a single field were to block after the body as a whole was judged redactable, the middleware raises, and your server returns 500 instead of forwarding the value.
  3. Wrap your application

    ASGI middleware is an application that wraps another application, so wrapping is one line. echo_app stands in for your API and returns the JSON it received.

    """A plain ASGI app that echoes the JSON it receives. Stands in for your API."""
import json

from helixor_guard import HelixorGuardMiddleware


async def echo_app(scope, receive, send):
    body = b""
    while True:
        message = await receive()
        body += message.get("body", b"")
        if not message.get("more_body"):
            break
    data = json.dumps({"received": json.loads(body or b"null")}).encode()
    await send({"type": "http.response.start", "status": 200,
                "headers": [(b"content-type", b"application/json")]})
    await send({"type": "http.response.body", "body": data})


app = HelixorGuardMiddleware(echo_app)

    If your framework registers middleware classes with add_middleware(), pass the class instead: app.add_middleware(HelixorGuardMiddleware). The framework supplies the wrapped app as the first argument.

  4. Test it in memory

    httpx.ASGITransport sends requests straight into the ASGI app, so the tests need no server and no port.

    import httpx
import pytest

from app import app


@pytest.fixture
def client():
    transport = httpx.ASGITransport(app=app)
    return httpx.AsyncClient(transport=transport, base_url="http://testserver")


@pytest.mark.anyio
async def test_clean_body_passes_through(client):
    r = await client.post("/tickets", json={"subject": "Printer jam", "priority": 2})
    assert r.status_code == 200
    assert r.json() == {"received": {"subject": "Printer jam", "priority": 2}}
    assert r.headers["x-helixor-action"] == "permit_clean_payload"
    assert r.headers["x-helixor-receipt"].startswith("hx_proof_")


@pytest.mark.anyio
async def test_contact_details_are_redacted(client):
    r = await client.post("/tickets", json={
        "subject": "Call back",
        "contact": {"email": "dana.reyes@example.com", "phone": "(415) 555-0100"},
    })
    assert r.status_code == 200
    assert r.json()["received"]["contact"] == {
        "email": "[REDACTED_EMAIL]", "phone": "[REDACTED_PHONE]"}
    assert r.headers["x-helixor-action"] == "redact_and_permit_contact_pii"


@pytest.mark.anyio
async def test_fatal_body_is_rejected(client):
    r = await client.post("/tickets", json={"notes": ["verified", "SSN 123-45-6789"]})
    assert r.status_code == 422
    body = r.json()
    assert body["action"] == "block_glba_ssn_leakage"
    assert body["rule_ids"] == ["RULE-GLBA-SSN-BLOCK"]
    assert body["receipt_hash"] == r.headers["x-helixor-receipt"]
    assert "123-45-6789" not in r.text


@pytest.mark.anyio
async def test_get_requests_are_not_screened(client):
    r = await client.get("/tickets")
    assert "x-helixor-receipt" not in r.headers


@pytest.fixture
def anyio_backend():
    return "asyncio"

    pytest -v test_middleware.py

    collecting ... collected 4 items

test_middleware.py::test_clean_body_passes_through PASSED                [ 25%]
test_middleware.py::test_contact_details_are_redacted PASSED             [ 50%]
test_middleware.py::test_fatal_body_is_rejected PASSED                   [ 75%]
test_middleware.py::test_get_requests_are_not_screened PASSED            [100%]

============================== 4 passed in 0.15s ===============================
  5. Run it for real 
    uvicorn app:app --host 127.0.0.1 --port 8000

    In a second terminal, send a body with an email address:

    curl -si -X POST http://127.0.0.1:8000/tickets \
  -H 'Content-Type: application/json' \
  -d '{"subject": "Call back", "contact": {"email": "dana.reyes@example.com"}}'

    HTTP/1.1 200 OK
server: uvicorn
content-type: application/json
x-helixor-receipt: hx_proof_4ce864e459d3032ae12d17ed
x-helixor-action: redact_and_permit_contact_pii
transfer-encoding: chunked

{"received": {"subject": "Call back", "contact": {"email": "[REDACTED_EMAIL]"}}}

    And one with an SSN:

    curl -si -X POST http://127.0.0.1:8000/tickets \
  -H 'Content-Type: application/json' \
  -d '{"notes": "SSN 123-45-6789"}'

    HTTP/1.1 422 Unprocessable Entity
server: uvicorn
content-type: application/json
content-length: 156
x-helixor-receipt: hx_proof_09a42743fc43b8091f4c0409

{"error": "blocked_by_policy", "action": "block_glba_ssn_leakage", "rule_ids": ["RULE-GLBA-SSN-BLOCK"], "receipt_hash": "hx_proof_09a42743fc43b8091f4c0409"}

    The rejection names the rule and gives the receipt, but never echoes the value. The date header is omitted from both outputs.

How it works#

An ASGI request body arrives as a series of http.request messages. The middleware drains them, decides, and then calls your app with a replacement receive that returns the (possibly repaired) body in one message, with content-length updated to match. It wraps send so it can add the two headers to http.response.start. Your handlers never see the original body when a rule fired.

The receipt in X-Helixor-Receipt fingerprints the pack, the action, the rules and a SHA-256 of the serialized body. Log it with your request ID, and return it to clients so a support request can quote it. See Receipts.

The middleware evaluates in the request path, on the event loop. Evaluation is in-process and short, but it is synchronous; if your bodies are large, measure with your own traffic and consider moving evaluation to a thread pool. See Performance.

Variations#

  • Log instead of reject while you roll out: forward the original body, but log the action, rule IDs and receipt. Switch to rejecting once the log shows no false positives.
  • Screen responses too: buffer http.response.body messages in send_with_receipt and apply the same decision before sending.
  • Non-Python services: run the decision as a sidecar and call it from your gateway. See Serve decisions to other languages.

Next steps#

Tutorial

Testing policies

Golden tests for the policy behind the middleware.

 Architecture

Deployment patterns

Embedded, sidecar and gateway placements.

 Reference

Errors

Every error the runtime raises.