JWT Verification Starter

Category: Security & Compliance

Get this pack →

This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.

Verify JSON Web Tokens from any source and protect a Postgres-backed API — no auth microservice to build. Validate RS256/ES256 tokens from Auth0, Clerk, Cognito, Firebase, or Okta against their JWKS (key rotation handled for you), or verify a static public key or an HS256 secret you issue yourself. Includes a one-call Postgres migration that shows off multi-statement batching.

Requires Air Pipe engine ≥ 0.196.0 for the RS256/ES256/JWKS modes and multi: true batching.

---

What's included

File	Purpose
schema.sql	The notes table (per-user, keyed by the JWT sub)
migrate.yml	POST /migrate and /migrate/reset — schema setup in one multi: true batch
verify.yml	POST /verify/{jwks,pem,hs256} — one reference endpoint per verification mode
notes.yml	A JWT-protected, owner-scoped notes API (list + create)

---

Endpoints

Method	Path	Auth	Description
POST	/migrate	—	Create the schema in one batch
POST	/migrate/reset	—	Drop, recreate, and seed (destructive)
POST	/verify/jwks	JWKS token	Return the token's claims (provider tokens)
POST	/verify/pem	PEM token	Return the token's claims (static public key)
POST	/verify/hs256	HS256 token	Return the token's claims (shared secret)
GET	/notes	JWKS token	List the caller's own notes
POST	/notes/create	JWKS token	Create a note owned by the caller

The JWT is sent in the airpipe-jwt header on every protected call.

---

The three verification modes

Pick the one that matches where your tokens come from — it's a single is_valid_jwt block.

1. Provider JWKS (Auth0 / Clerk / Cognito / Firebase / Okta) — the common case. Air Pipe fetches and caches the provider's public keys, selects the signer by the token's kid, and enforces iss/aud/exp. Provider key rotation just works.

is_valid_jwt:
  jwks_url: a|ap_var::OIDC_JWKS_URL|
  alg: RS256
  iss: a|ap_var::OIDC_ISSUER|
  aud: a|ap_var::OIDC_AUDIENCE|

2. Static public key — verify RS256/ES256/EdDSA offline against a PEM you paste in.

is_valid_jwt:
  public_key: a|ap_var::JWT_PUBLIC_KEY|
  alg: RS256

3. HS256 shared secret — for tokens you issue yourself (the original bare-string form).

is_valid_jwt: a|ap_var::JWT_SHARED_SECRET|

On success the decoded claims are available to later actions (a|ValidateJwt::sub|, a|ValidateJwt::email|, …). See notes.yml for a query scoped to the caller by sub.

---

Multi-statement batching (multi: true)

migrate.yml runs its whole DDL block in a single action:

- name: Migrate
  database: main
  multi: true          # run all the ; -separated statements as one batch
  query: |
    CREATE EXTENSION IF NOT EXISTS pgcrypto;
    CREATE TABLE IF NOT EXISTS notes ( ... );
    CREATE INDEX IF NOT EXISTS idx_notes_owner ON notes (owner, created_at DESC);

Without multi: true, the native Postgres driver prepares the statement and Postgres rejects more than one command (SQLSTATE 42601), so you'd otherwise split every statement into its own action. Use it for migrations, schema setup, and seed scripts. It's for no-parameter batches; keep parameterized writes as normal single-statement actions.

---

Setup

1. Managed variables

In the Air Pipe dashboard, add the variables for the mode(s) you use:

Name	Used by	Example
DATABASE_URL	migrate, notes	postgresql://user:pass@host:5432/db
OIDC_JWKS_URL	JWKS mode	https://YOUR.us.auth0.com/.well-known/jwks.json
OIDC_ISSUER	JWKS mode	https://YOUR.us.auth0.com/
OIDC_AUDIENCE	JWKS mode	https://your-api
JWT_PUBLIC_KEY	PEM mode	-----BEGIN PUBLIC KEY-----\n…
JWT_SHARED_SECRET	HS256 mode	a 32+ char secret

2. Create the schema

curl -X POST https://your-airpipe-host/migrate

3. Deploy the configs and start calling protected routes.

---

Quick start: curl walkthrough

Responses are wrapped in Air Pipe's standard {"data":{ "<Action>": {"data": …} }} trace; extract with jq.

BASE=https://your-airpipe-host
TOKEN='<a JWT from your provider>'

# Migrate (one multi:true batch), then seed demo rows
curl -sX POST $BASE/migrate | jq '.data.Status.data'
curl -sX POST $BASE/migrate/reset | jq '.data.Status.data'    # => { status: reset, notes: 3 }

# Inspect what your provider issued
curl -sX POST $BASE/verify/jwks -H "airpipe-jwt: $TOKEN" | jq '.data.ValidateJwt.data'

# Use it: list and create notes, scoped to your `sub`
curl -s      $BASE/notes         -H "airpipe-jwt: $TOKEN"
curl -sX POST $BASE/notes/create -H "airpipe-jwt: $TOKEN" \
  -H 'content-type: application/json' -d '{"title":"Hello","body":"My first note"}'

---

Getting a test token

• Auth0/Clerk/Cognito: grab an access token from your app's login, set OIDC_JWKS_URL/OIDC_ISSUER/OIDC_AUDIENCE to your tenant's values, and call /verify/jwks.
• HS256 (self-issued): mint one at jwt.io (HS256, secret = JWT_SHARED_SECRET, payload e.g. {"sub":"user-123","exp":9999999999}) and call /verify/hs256.

---

Notes

• Algorithms. HS256/384/512, RS256/384/512, PS256/384/512, ES256/384, and EdDSA. iss/aud are enforced only when you set them; exp is always checked (with the standard 60s leeway).
• Key rotation. JWKS documents are cached briefly and re-fetched, and the signer is chosen per-token by kid, so rotating provider keys need no redeploy.
• Token transport. Send the JWT in the airpipe-jwt header. (Air Pipe also forwards a bearer credential into this slot for MCP tools.)
• Scoping. notes is filtered by owner = a|ValidateJwt::sub| and bound as a parameter, so a user only ever sees their own rows and the claim is never string-interpolated into SQL.

Configuration

migrate.yml

name: JwtStarterMigrate
description: One-call schema setup using multi:true — runs a whole multi-statement DDL batch in a single action instead of splitting every CREATE into its own step.

docs: true

# multi:true runs a multi-statement SQL batch in ONE action. Without it the
# native Postgres driver prepares the query, and a prepared statement can hold
# only one command (SQLSTATE 42601). This is the clean way to ship migrations,
# schema setup, and seed scripts. Requires engine >= 0.196.0.
#
# Required managed variable:
#   DATABASE_URL — Postgres connection string (pgcrypto enabled)

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # POST /migrate
  # Idempotently create the schema in one batch, then report the row count.
  migrate:
    output: http
    method: POST
    summary: Run the schema migration
    description: Creates the extension, table, and index in a single multi:true batch.
    tags: [setup]
    response_example:
      status: ok
      notes: 0

    actions:
      - name: Migrate
        database: main
        multi: true
        hide_data_on_success: true
        query: |
          CREATE EXTENSION IF NOT EXISTS pgcrypto;

          CREATE TABLE IF NOT EXISTS notes (
            id         UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
            owner      TEXT        NOT NULL,
            title      TEXT        NOT NULL,
            body       TEXT        NOT NULL DEFAULT '',
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
          );

          CREATE INDEX IF NOT EXISTS idx_notes_owner ON notes (owner, created_at DESC);

      - name: Status
        run_when_succeeded: [Migrate]
        database: main
        query: SELECT 'ok' AS status, count(*) AS notes FROM notes;
        post_transforms:
          - extract_value: "[0]"

  # POST /migrate/reset
  # Drop and recreate the table, then load sample rows for two owners — a fuller
  # batch (DROP + CREATE + INDEX + multi-row INSERT) in a single action.
  migrate/reset:
    output: http
    method: POST
    summary: Reset and seed
    description: Drops, recreates, and seeds the notes table in one multi:true batch. Destroys existing data.
    tags: [setup]
    response_example:
      status: reset
      notes: 3

    actions:
      - name: Reset
        database: main
        multi: true
        hide_data_on_success: true
        query: |
          DROP TABLE IF EXISTS notes;

          CREATE TABLE notes (
            id         UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
            owner      TEXT        NOT NULL,
            title      TEXT        NOT NULL,
            body       TEXT        NOT NULL DEFAULT '',
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
          );

          CREATE INDEX idx_notes_owner ON notes (owner, created_at DESC);

          INSERT INTO notes (owner, title, body) VALUES
            ('user-123', 'Welcome',      'This note belongs to user-123.'),
            ('user-123', 'Shopping list','Milk, eggs, bread.'),
            ('user-456', 'Someone else', 'This note belongs to a different user.');

      - name: Status
        run_when_succeeded: [Reset]
        database: main
        query: SELECT 'reset' AS status, count(*) AS notes FROM notes;
        post_transforms:
          - extract_value: "[0]"

notes.yml

name: JwtStarterNotes
description: A JWT-protected, per-user notes API over Postgres. Verifies the caller's token via JWKS and scopes every query to the token's `sub` claim, so users only ever see their own rows.

docs: true

# A real, protected resource. Every endpoint:
#   1. verifies the caller's JWT (JWKS / provider tokens by default), and
#   2. scopes its query to the token's `sub` claim (a|ValidateJwt::sub|),
# so a user can only ever read or write their own notes. Swap the ValidateJwt
# block for verify/pem's or verify/hs256's if you issue your own tokens.
#
# Send the token in the `airpipe-jwt` header.
#
# Required managed variables:
#   DATABASE_URL, OIDC_JWKS_URL, OIDC_ISSUER, OIDC_AUDIENCE

global:
  databases:
    main:
      driver: postgres
      conn_string: "a|ap_var::DATABASE_URL|"

interfaces:

  # GET /notes  — list the caller's own notes.
  notes:
    output: http
    method: GET
    summary: List my notes
    description: Returns the authenticated caller's notes, newest first.
    tags: [notes]
    response_example:
      - id: 3f9a...
        title: Welcome
        body: This note belongs to you.
        created_at: "2026-07-03T12:00:00Z"

    actions:
      - name: ValidateJwt
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt:
                jwks_url: a|ap_var::OIDC_JWKS_URL|
                alg: RS256
                iss: a|ap_var::OIDC_ISSUER|
                aud: a|ap_var::OIDC_AUDIENCE|
        post_transforms:
          - extract_value: jwt_claims

      - name: ListNotes
        run_when_succeeded:
          actions: [ValidateJwt]
          http_code_on_error: 401
        database: main
        query: |
          SELECT id, title, body, created_at
          FROM notes
          WHERE owner = $1
          ORDER BY created_at DESC
          LIMIT 200;
        params:
          - a|ValidateJwt::sub|

  # POST /notes/create  — create a note owned by the caller.
  notes/create:
    output: http
    method: POST
    summary: Create a note
    description: Creates a note owned by the authenticated caller (owner = token sub).
    tags: [notes]
    request_example:
      title: Shopping list
      body: Milk, eggs, bread.
    response_example:
      id: 7c1e...
      title: Shopping list
      body: Milk, eggs, bread.

    actions:
      - name: ValidateJwt
        input: a|headers|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt:
                jwks_url: a|ap_var::OIDC_JWKS_URL|
                alg: RS256
                iss: a|ap_var::OIDC_ISSUER|
                aud: a|ap_var::OIDC_AUDIENCE|
        post_transforms:
          - extract_value: jwt_claims

      - name: CheckBody
        run_when_succeeded:
          actions: [ValidateJwt]
          http_code_on_error: 400
        input: a|body|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "title is required"
          tests:
            - value: title
              is_not_null: true
              is_not_empty: true

      - name: CreateNote
        run_when_succeeded: [CheckBody]
        database: main
        query: |
          INSERT INTO notes (owner, title, body)
          VALUES ($1, $2, COALESCE($3, ''))
          RETURNING id, title, body, created_at;
        params:
          - a|ValidateJwt::sub|
          - a|CheckBody::title|
          - a|body::body->default(null)|
        post_transforms:
          - extract_value: "[0]"

verify.yml

name: JwtStarterVerify
description: Reference endpoints for every is_valid_jwt verification mode — a provider JWKS (Auth0/Clerk/Cognito/Firebase), a static PEM public key, and an HS256 shared secret. Each returns the token's decoded claims.

docs: true

# Three ways to verify a JWT. Send the token in the `airpipe-jwt` header:
#   curl -X POST https://host/verify/jwks -H "airpipe-jwt: <token>"
#
# Each route validates the token and returns its claims, so you can see exactly
# what your provider issues. Copy the ValidateJwt block that matches your setup
# into your own endpoints (see notes.yml for a scoped, real example).
#
# Requires engine >= 0.196.0 for the RS256/ES256/JWKS modes.
#
# Required managed variables (only for the mode(s) you use):
#   OIDC_JWKS_URL    — your provider's JWKS, e.g. https://YOUR.us.auth0.com/.well-known/jwks.json
#   OIDC_ISSUER      — expected `iss`
#   OIDC_AUDIENCE    — expected `aud`
#   JWT_PUBLIC_KEY   — a PEM public key (for the static-key mode)
#   JWT_SHARED_SECRET— an HS256 secret (for the shared-secret mode)

interfaces:

  # POST /verify/jwks — RS256 against a provider JWKS (key selected by `kid`).
  # This is what you use for Auth0, Clerk, Cognito, Firebase, Okta, etc.
  verify/jwks:
    output: http
    method: POST
    summary: Verify via JWKS
    description: Verify an RS256 token against a provider's JWKS and return its claims.
    tags: [verify]
    response_example:
      sub: user-123
      iss: https://your.auth0.com/
      aud: https://your-api

    actions:
      - name: ValidateJwt
        input: a|headers|
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt:
                jwks_url: a|ap_var::OIDC_JWKS_URL|
                alg: RS256
                iss: a|ap_var::OIDC_ISSUER|
                aud: a|ap_var::OIDC_AUDIENCE|
        post_transforms:
          - extract_value: jwt_claims

  # POST /verify/pem — RS256 against a static PEM public key (no network calls).
  verify/pem:
    output: http
    method: POST
    summary: Verify via public key
    description: Verify an RS256 token against a static PEM public key and return its claims.
    tags: [verify]
    response_example:
      sub: user-123

    actions:
      - name: ValidateJwt
        input: a|headers|
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt:
                public_key: a|ap_var::JWT_PUBLIC_KEY|
                alg: RS256
        post_transforms:
          - extract_value: jwt_claims

  # POST /verify/hs256 — HS256 with a shared secret (tokens you issue yourself).
  verify/hs256:
    output: http
    method: POST
    summary: Verify via shared secret
    description: Verify an HS256 token against a shared secret and return its claims.
    tags: [verify]
    response_example:
      sub: user-123

    actions:
      - name: ValidateJwt
        input: a|headers|
        assert:
          http_code_on_error: 401
          error_message: "Invalid or missing token"
          tests:
            - value: airpipe-jwt
              is_not_null: true
              is_valid_jwt: a|ap_var::JWT_SHARED_SECRET|
        post_transforms:
          - extract_value: jwt_claims

schema.sql

-- JWT Verification Starter schema (Postgres)
--
-- Run once with psql, or deploy migrate.yml and call POST /migrate (which runs
-- the whole block in one `multi: true` action).
CREATE EXTENSION IF NOT EXISTS pgcrypto;

-- A per-user resource. `owner` holds the JWT `sub` claim, so every row belongs
-- to exactly one authenticated caller and queries scope to a|ValidateJwt::sub|.
CREATE TABLE IF NOT EXISTS notes (
  id         UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
  owner      TEXT        NOT NULL,           -- the verified JWT `sub`
  title      TEXT        NOT NULL,
  body       TEXT        NOT NULL DEFAULT '',
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS idx_notes_owner ON notes (owner, created_at DESC);