OAuth Connect

Category: Security & Compliance

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

The OAuth2 authorization code flow with refresh, per user. This is what sits behind every "Connect your Google Calendar" button: the user grants access individually, you store their refresh token, and you mint fresh access tokens from then on.

This is not client-credentials. That flow authenticates your app to an API and involves no user at all. Here every user has their own connection and their own tokens.

Endpoints

Method	Route	Purpose
`GET`	`/oauth/start?user_id=alice`	Begin the flow — 302 to the provider's consent screen
`GET`	`/oauth/callback?code=…&state=…`	Provider redirects here; exchanges the code and stores tokens
`GET`	`/oauth/token?user_id=alice`	Returns a valid access token, refreshing if needed
`GET`	`/oauth/connections?limit=25`	Who is connected, until when — never returns token material

/oauth/token is the one the rest of your app calls. It hides the entire refresh dance:

{ "access_token": "ya29.a0Af…", "source": "cache" }
{ "access_token": "ya29.a0Af…", "source": "refreshed" }

Four things this gets right

1. CSRF on the callback. The state parameter is not decoration. Without it an attacker can hand your callback their authorization code and bind their account to your user's row. State here is bound to the user, single-use, and expires in 10 minutes — the callback consumes it with an UPDATE … WHERE used_at IS NULL AND expires_at > NOW().

2. Tokens are stored encrypted. A refresh token is a long-lived bearer credential: whoever holds it can mint access tokens until the user revokes consent. Both tokens are sealed with AES-256-GCM via encrypt_value under OAUTH_ENC_KEY, which lives in a managed variable and never in the database. Each seal needs its ciphertext, nonce and GCM tag to be reversible, which is why there are three columns per token.

A fresh 12-byte nonce is minted per encryption — never reuse a nonce under GCM. Note the nonces come from their own action: markers are resolved before an action's post_transforms run, so encrypt_value.nonce cannot reference a generate_bytes key written in the same transform list. It has to come from a prior action.

3. The provider sends refresh_token only ONCE. Google returns it on the first consent and omits it on every later one. The naive upsert overwrites the stored value with the absent one and silently breaks refresh forever.

The subtle part: you cannot guard this with NULLIF(value, '') after encrypting, because encrypt_value('') returns perfectly real ciphertext — the guard never fires and you destroy the token anyway. The refresh token is sealed in a separate action gated on the token actually being present, so when the provider omits it that action is skipped, every reference resolves to null, and the upsert's COALESCE keeps what is already stored.

4. Expiry is computed in Postgres. expires_in is a relative number of seconds and is useless stored as-is. It becomes an absolute timestamptz, and the refresh decision is expires_at <= NOW() + INTERVAL '60 seconds' — a SQL comparison, so no clock arithmetic happens in the config. The 60 seconds of slack covers the gap between minting a token here and the caller actually using it.

A pattern worth copying: zero rows fails an action

extract_value: "[0]" fails when a query returns no rows, which skips every dependent action — including the assert you wrote to return a clean 400/404. The caller gets a 500 instead. Both lookups here are written to always return exactly one row, with NULLs when nothing matched, so the asserts can do their job:

-- unknown / used / expired state -> one row, user_id NULL
WITH consumed AS (
  UPDATE oauth_states SET used_at = NOW()
   WHERE state = '…' AND used_at IS NULL AND expires_at > NOW()
  RETURNING user_id
)
SELECT (SELECT user_id FROM consumed) AS user_id;

-- unknown user -> one row of NULLs
SELECT c.user_id, c.access_value, …
  FROM (SELECT 1) AS dummy
  LEFT JOIN oauth_connections c ON c.user_id = '…';

URL-encoding

airpipe has no url_encode interpolation filter, so percent-encoding is done in SQL by ap_urlencode, created by seed.yml. It escapes % space + / : = & ? # (with % first, so the escapes are not themselves re-escaped) — the characters that actually occur in OAuth values: authorization codes (Google's contain /), client secrets, redirect URIs, and space-separated scopes. It is not a general-purpose RFC 3986 encoder; it covers this flow deterministically.

Required variables

Variable	What to set it to
`DATABASE_URL`	Postgres connection string
`OAUTH_AUTH_URL`	Consent endpoint — Google: `https://accounts.google.com/o/oauth2/v2/auth`
`OAUTH_TOKEN_URL`	Token endpoint — Google: `https://oauth2.googleapis.com/token`
`OAUTH_CLIENT_ID`	From the provider's console
`OAUTH_CLIENT_SECRET`	From the provider's console
`OAUTH_REDIRECT_URI`	This config's `/oauth/callback` URL, verbatim, as registered with the provider
`OAUTH_SCOPE`	Space-separated scopes
`OAUTH_ENC_KEY`	base64 of 32 random bytes — the AES-256 key sealing stored tokens

Generate the encryption key with:

openssl rand -base64 32

Losing OAUTH_ENC_KEY means every stored token becomes undecryptable and all users must reconnect. Rotating it requires re-sealing existing rows.

Provider setup (Google shown)

Google Cloud Console → APIs & Services → Credentials → OAuth client ID → Web application.
Add your deployed /oauth/callback URL as an Authorized redirect URI — it must match OAUTH_REDIRECT_URI character for character.
Copy the client ID and secret into the variables above.
Set OAUTH_SCOPE to the scopes you need, space-separated.

access_type=offline and prompt=consent are already in the consent URL — without them Google will not issue a refresh token at all.

Database

schema.sql contains the DDL. Two tables plus one function:

ap_urlencode(text) — the percent-encoder described above.
oauth_states — single-use CSRF states, with expires_at and used_at.
oauth_connections — one row per user: sealed access and refresh tokens (value/nonce/tag each), granted scope, and absolute expires_at.

POST /api/seed (in seed.yml) creates all three; it is idempotent.

Usage example

# 1. Send the user to consent (in a browser — this 302s to the provider)
open 'https://api.airpipe.io/<org>/production/oauth/start?user_id=alice'

# 2. The provider redirects back to /oauth/callback and the tokens are stored.

# 3. From then on, ask for a valid token whenever you need one
curl 'https://api.airpipe.io/<org>/production/oauth/token?user_id=alice'
# -> {"access_token":"ya29.…","source":"cache"}      (still valid)
# -> {"access_token":"ya29.…","source":"refreshed"}  (was expired, refreshed for you)

If the user revokes consent, /oauth/token returns 401 with "refresh failed — the user must reconnect via /oauth/start". That is deliberately not a 5xx: the fix is to send the user through consent again, not to retry.

Verified

Runtime-tested against a mock OAuth2 provider, 36/36: the 302 and its percent-encoding, state rejection/replay/expiry, the token request arriving genuinely urlencoded with a secret containing / + = round-tripping byte-for-byte, tokens absent from the database in plaintext (asserted against the raw rows), transparent refresh on expiry, re-encryption of the refreshed token, re-consent without a refresh_token leaving the stored one intact and still usable, a revoked refresh token surfacing as 401, unknown user as 404, and the list route leaking no token material.

Configuration

config.yml

name: OauthConnect

docs: true

# "Let my users connect their Google account" — the OAuth2 AUTHORIZATION CODE
# flow, with refresh, per user.
#
# This is the flow behind every "Connect your Google Calendar / HubSpot / Slack
# workspace" button. It is NOT client-credentials (that authenticates YOUR app
# to an API and needs no user). Here each end user grants access individually,
# you store their refresh token, and you mint fresh access tokens forever after.
#
# Four things this handles that a naive implementation gets wrong:
#
#   1. CSRF ON THE CALLBACK. The `state` parameter is not decoration. Without it
#      an attacker can feed your callback their own `code` and bind THEIR account
#      to YOUR user's row. State here is single-use and expires in 10 minutes.
#
#   2. REFRESH TOKENS ARE STORED ENCRYPTED. A refresh token is a long-lived
#      bearer credential — anyone holding it can mint access tokens until the
#      user revokes consent. It is sealed with AES-256-GCM (`encrypt_value`)
#      under a key that lives in a managed variable, not in the database.
#
#   3. THE PROVIDER ONLY SENDS refresh_token ONCE. Google returns it on the FIRST
#      consent and omits it on every later one. Overwriting the stored value with
#      an absent one silently breaks refresh, so the upsert below KEEPS the old
#      refresh token whenever the new response has none.
#
#   4. EXPIRY IS COMPUTED IN POSTGRES. `expires_in` is a relative number of
#      seconds; storing it as-is is useless. It becomes an absolute timestamptz,
#      and the refresh decision is `expires_at <= NOW() + 60s` — a SQL
#      comparison, so no clock maths happens in the config.
#
# NOTE ON URL-ENCODING: airpipe has no url_encode filter, so the percent-encoding
# is done in SQL (`ap_urlencode` below, created by seed.yml). That is deliberate
# and deterministic — see the README.
#
# Required managed variables:
#   DATABASE_URL         — Postgres connection string
#   OAUTH_AUTH_URL       — provider consent endpoint
#                          (Google: https://accounts.google.com/o/oauth2/v2/auth)
#   OAUTH_TOKEN_URL      — provider token endpoint
#                          (Google: https://oauth2.googleapis.com/token)
#   OAUTH_CLIENT_ID      — from the provider's console
#   OAUTH_CLIENT_SECRET  — from the provider's console
#   OAUTH_REDIRECT_URI   — this config's /oauth/callback route, verbatim, as
#                          registered with the provider
#   OAUTH_SCOPE          — space-separated scopes
#   OAUTH_ENC_KEY        — base64 32-byte AES-256 key sealing the stored tokens

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

interfaces:

  # GET /oauth/start?user_id=alice
  # Send the user here. Responds 302 to the provider's consent screen.
  oauth/start:
    output: http
    method: GET
    summary: Begin the OAuth flow — redirects the user to the provider's consent screen
    description: >
      Mints a single-use CSRF state bound to the user, builds the consent URL,
      and returns a 302 redirect.
    tags: [oauth, auth, redirect]

    actions:
      - name: ReadParams
        input: a|params|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "user_id is required"
          tests:
            - value: user_id
              is_not_null: true
              is_not_empty: true

      # State row + the consent URL in one statement. ap_urlencode escapes the
      # values that actually need it (redirect_uri has `:` and `/`, scope has
      # spaces and `:`/`/`). `access_type=offline` + `prompt=consent` is what
      # makes Google return a refresh_token at all.
      - name: CreateState
        run_when_succeeded: [ReadParams]
        database: main
        query: |-
          WITH s AS (
            INSERT INTO oauth_states (state, user_id, expires_at)
            VALUES (
              a|uuid->single_quote|,
              a|ReadParams::user_id->single_quote|,
              NOW() + INTERVAL '10 minutes'
            )
            RETURNING state
          )
          SELECT s.state,
                 $1
                   || '?response_type=code'
                   || '&client_id='     || ap_urlencode($2)
                   || '&redirect_uri='  || ap_urlencode($3)
                   || '&scope='         || ap_urlencode($4)
                   || '&access_type=offline&include_granted_scopes=true&prompt=consent'
                   || '&state='         || s.state
                 AS consent_url
            FROM s
        params:
          - a|ap_var::OAUTH_AUTH_URL|
          - a|ap_var::OAUTH_CLIENT_ID|
          - a|ap_var::OAUTH_REDIRECT_URI|
          - a|ap_var::OAUTH_SCOPE|
        post_transforms:
          - extract_value: "[0]"

      # The redirect itself. response_on_success ends the interface here, so it
      # must be a LATER action than the one it reads.
      - name: Redirect
        run_when_succeeded: [CreateState]
        input: a|CreateState|
        hide_data_on_success: true
        response_on_success:
          http_code: 302
          headers:
            location: a|CreateState::consent_url|
            cache-control: "no-store"
          body: ""

  # GET /oauth/callback?code=...&state=...
  # The provider redirects the user back here. Register this exact URL as the
  # redirect URI in the provider console.
  oauth/callback:
    output: http
    method: GET
    summary: Exchange the authorization code for tokens and store them
    description: >
      Validates the single-use state, exchanges the code at the token endpoint,
      encrypts the tokens and upserts the connection.
    tags: [oauth, auth, postgres, security]

    actions:
      - name: ReadParams
        input: a|params|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "code and state are required"
          tests:
            - value: code
              is_not_null: true
              is_not_empty: true
            - value: state
              is_not_null: true
              is_not_empty: true

      # Single-use: the UPDATE consumes the state, so a replayed callback finds
      # nothing. Expiry is enforced in the same statement.
      #
      # The scalar subquery matters. A bare UPDATE…RETURNING yields ZERO rows for
      # an unknown, used or expired state — and `extract_value: "[0]"` then FAILS
      # the action, so RequireState below never runs and the caller gets a 500
      # instead of the intended 400. Wrapping it makes the statement always
      # return exactly one row, with a NULL user_id when nothing matched, which
      # is what the assert is written to catch.
      - name: ConsumeState
        run_when_succeeded: [ReadParams]
        database: main
        query: |-
          WITH consumed AS (
            UPDATE oauth_states
               SET used_at = NOW()
             WHERE state = a|ReadParams::state->single_quote|
               AND used_at IS NULL
               AND expires_at > NOW()
            RETURNING user_id
          )
          SELECT (SELECT user_id FROM consumed) AS user_id
        post_transforms:
          - extract_value: "[0]"

      - name: RequireState
        run_when_succeeded: [ConsumeState]
        input: a|ConsumeState|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "state is unknown, already used, or expired"
          tests:
            - value: user_id
              is_not_null: true
              is_not_empty: true

      # Token endpoints take application/x-www-form-urlencoded, so the body is
      # built as a string. A string body is sent verbatim (verified), which is
      # why the percent-encoding must already be correct here.
      - name: BuildTokenForm
        run_when_succeeded: [RequireState]
        database: main
        query: |-
          SELECT 'grant_type=authorization_code'
                 || '&code='          || ap_urlencode($1)
                 || '&client_id='     || ap_urlencode($2)
                 || '&client_secret=' || ap_urlencode($3)
                 || '&redirect_uri='  || ap_urlencode($4)
                 AS form
        params:
          - a|ReadParams::code|
          - a|ap_var::OAUTH_CLIENT_ID|
          - a|ap_var::OAUTH_CLIENT_SECRET|
          - a|ap_var::OAUTH_REDIRECT_URI|
        post_transforms:
          - extract_value: "[0]"

      - name: ExchangeCode
        run_when_succeeded: [BuildTokenForm]
        hide_data_on_error: true
        http:
          url: a|ap_var::OAUTH_TOKEN_URL|
          method: POST
          headers:
            content-type: application/x-www-form-urlencoded
            accept: application/json
          body: a|BuildTokenForm::form|

      # OAuth errors arrive as {"error": "invalid_grant"} — often with a non-2xx,
      # but not always. Assert on the payload rather than trusting the status.
      - name: CheckExchange
        run_when_succeeded: [ExchangeCode]
        input: a|ExchangeCode|
        hide_data_on_success: true
        assert:
          http_code_on_error: 502
          error_message: "token exchange failed"
          tests:
            - value: body.access_token
              is_not_null: true
              is_not_empty: true

      # A fresh 12-byte nonce per encryption — never reuse one under AES-GCM.
      # generate_bytes emits base64, which is the form encrypt_value wants.
      #
      # The nonces are minted in their OWN action: markers are resolved BEFORE an
      # action's post_transforms run, so `encrypt_value.nonce` cannot reference a
      # key written by a generate_bytes sitting in the same transform list. It has
      # to come from a PRIOR action's output.
      - name: MakeNonces
        run_when_succeeded: [CheckExchange]
        json_output: |-
          { "minted": true }
        hide_data_on_success: true
        post_transforms:
          - generate_bytes:
              key: nonce_access
              length: 12
          - generate_bytes:
              key: nonce_refresh
              length: 12

      - name: SealTokens
        run_when_succeeded: [MakeNonces]
        json_output: |-
          { "sealed": true }
        hide_data_on_success: true
        post_transforms:
          - encrypt_value:
              key: access_sealed
              data: a|ExchangeCode::body.access_token|
              secret_key: a|ap_var::OAUTH_ENC_KEY|
              nonce: a|MakeNonces::nonce_access|

      # Sealed SEPARATELY, and only when the provider actually sent one.
      #
      # Encrypting an absent refresh token unconditionally is a silent
      # data-loss bug: encrypt_value('') returns real ciphertext, so a
      # `NULLIF(value, '')` guard downstream never fires and the upsert
      # overwrites the user's ONLY refresh token with the sealed empty string.
      # Skipping the action instead leaves every reference to it null, which is
      # what makes the COALESCE in SaveConnection keep the stored value.
      - name: SealRefresh
        run_when_succeeded: [MakeNonces]
        run_on_assertion:
          tests:
            - custom_value: "a|ExchangeCode::body.refresh_token->default('')|"
              is_not_empty: true
        json_output: |-
          { "sealed": true }
        hide_data_on_success: true
        post_transforms:
          - encrypt_value:
              key: refresh_sealed
              data: a|ExchangeCode::body.refresh_token|
              secret_key: a|ap_var::OAUTH_ENC_KEY|
              nonce: a|MakeNonces::nonce_refresh|

      # THE refresh_token TRAP. The provider sends it on the FIRST consent only;
      # on re-consent the response carries none, SealRefresh is skipped, and
      # these three params resolve to NULL — so COALESCE below keeps whatever is
      # already stored instead of destroying it.
      - name: SaveConnection
        run_when_succeeded: [SealTokens]
        database: main
        query: |-
          INSERT INTO oauth_connections (
            user_id, access_value, access_nonce, access_tag,
            refresh_value, refresh_nonce, refresh_tag,
            scope, expires_at, updated_at
          )
          VALUES (
            $1, $2, $3, $4,
            $5, $6, $7,
            $8, NOW() + ($9 || ' seconds')::INTERVAL, NOW()
          )
          ON CONFLICT (user_id) DO UPDATE SET
            access_value  = EXCLUDED.access_value,
            access_nonce  = EXCLUDED.access_nonce,
            access_tag    = EXCLUDED.access_tag,
            refresh_value = COALESCE(EXCLUDED.refresh_value, oauth_connections.refresh_value),
            refresh_nonce = COALESCE(EXCLUDED.refresh_nonce, oauth_connections.refresh_nonce),
            refresh_tag   = COALESCE(EXCLUDED.refresh_tag,   oauth_connections.refresh_tag),
            scope         = EXCLUDED.scope,
            expires_at    = EXCLUDED.expires_at,
            updated_at    = NOW()
          RETURNING user_id, scope, expires_at
        params:
          - a|ConsumeState::user_id|
          - a|SealTokens::access_sealed.value|
          - a|SealTokens::access_sealed.nonce|
          - a|SealTokens::access_sealed.tag|
          - a|SealRefresh::refresh_sealed.value->default(null)|
          - a|SealRefresh::refresh_sealed.nonce->default(null)|
          - a|SealRefresh::refresh_sealed.tag->default(null)|
          - a|ExchangeCode::body.scope->default('')|
          - a|ExchangeCode::body.expires_in->default(3600)|
        post_transforms:
          - extract_value: "[0]"

      - name: Respond
        run_when_succeeded: [SaveConnection]
        json_output: |-
          {
            "connected": true,
            "user_id": "a|SaveConnection::user_id|",
            "scope": "a|SaveConnection::scope|"
          }

  # GET /oauth/token?user_id=alice
  # Returns a VALID access token, refreshing transparently when it has expired.
  # This is what the rest of your app calls.
  oauth/token:
    output: http
    method: GET
    summary: Get a valid access token for a user, refreshing if it has expired
    description: >
      Reads the stored connection, decrypts it, and refreshes via the stored
      refresh token when the access token is within 60 seconds of expiry.
    tags: [oauth, auth, refresh, postgres]

    actions:
      - name: ReadParams
        input: a|params|
        hide_data_on_success: true
        assert:
          http_code_on_error: 400
          error_message: "user_id is required"
          tests:
            - value: user_id
              is_not_null: true
              is_not_empty: true

      # `needs_refresh` is decided in SQL — a timestamptz comparison, so no
      # clock arithmetic happens in the config. 60s of slack covers the time
      # between minting the token here and the caller actually using it.
      #
      # LEFT JOIN off a one-row dummy for the same reason as ConsumeState: an
      # unknown user must yield one row of NULLs, not zero rows. Zero rows would
      # fail extract_value and turn the intended 404 into a 500.
      - name: LoadConnection
        run_when_succeeded: [ReadParams]
        database: main
        query: |-
          SELECT c.user_id, c.access_value, c.access_nonce, c.access_tag,
                 c.refresh_value, c.refresh_nonce, c.refresh_tag,
                 c.expires_at,
                 (c.expires_at <= NOW() + INTERVAL '60 seconds') AS needs_refresh,
                 (c.refresh_value IS NOT NULL)                   AS can_refresh
            FROM (SELECT 1) AS dummy
            LEFT JOIN oauth_connections c
                   ON c.user_id = a|ReadParams::user_id->single_quote|
        post_transforms:
          - extract_value: "[0]"

      - name: RequireConnection
        run_when_succeeded: [LoadConnection]
        input: a|LoadConnection|
        hide_data_on_success: true
        assert:
          http_code_on_error: 404
          error_message: "no connection for that user — send them to /oauth/start"
          tests:
            - value: user_id
              is_not_null: true
              is_not_empty: true

      # --- still valid: decrypt and return -------------------------------
      - name: DecryptCurrent
        run_when_succeeded: [RequireConnection]
        run_on_assertion:
          tests:
            - action: LoadConnection
              value: needs_refresh
              is_equal_to: false
        json_output: |-
          { "source": "cache" }
        post_transforms:
          - decrypt_value:
              key: access_token
              value: a|LoadConnection::access_value|
              nonce: a|LoadConnection::access_nonce|
              tag: a|LoadConnection::access_tag|
              secret_key: a|ap_var::OAUTH_ENC_KEY|

      # --- expired: refresh ----------------------------------------------
      - name: DecryptRefresh
        run_when_succeeded: [RequireConnection]
        run_on_assertion:
          tests:
            - action: LoadConnection
              value: needs_refresh
              is_equal_to: true
            - action: LoadConnection
              value: can_refresh
              is_equal_to: true
        json_output: |-
          { "refreshing": true }
        post_transforms:
          - decrypt_value:
              key: refresh_token
              value: a|LoadConnection::refresh_value|
              nonce: a|LoadConnection::refresh_nonce|
              tag: a|LoadConnection::refresh_tag|
              secret_key: a|ap_var::OAUTH_ENC_KEY|

      - name: BuildRefreshForm
        run_when_succeeded: [DecryptRefresh]
        database: main
        query: |-
          SELECT 'grant_type=refresh_token'
                 || '&refresh_token=' || ap_urlencode($1)
                 || '&client_id='     || ap_urlencode($2)
                 || '&client_secret=' || ap_urlencode($3)
                 AS form
        params:
          - a|DecryptRefresh::refresh_token|
          - a|ap_var::OAUTH_CLIENT_ID|
          - a|ap_var::OAUTH_CLIENT_SECRET|
        post_transforms:
          - extract_value: "[0]"

      - name: RefreshToken
        run_when_succeeded: [BuildRefreshForm]
        hide_data_on_error: true
        http:
          url: a|ap_var::OAUTH_TOKEN_URL|
          method: POST
          headers:
            content-type: application/x-www-form-urlencoded
            accept: application/json
          body: a|BuildRefreshForm::form|

      # A refresh token can be revoked by the user at any time; the provider
      # answers invalid_grant. That is a 401, not a 502 — the caller has to send
      # the user back through consent, not retry.
      - name: CheckRefresh
        run_when_succeeded: [RefreshToken]
        input: a|RefreshToken|
        hide_data_on_success: true
        assert:
          http_code_on_error: 401
          error_message: "refresh failed — the user must reconnect via /oauth/start"
          tests:
            - value: body.access_token
              is_not_null: true
              is_not_empty: true

      # Same two-step as the callback: nonce first, in its own action.
      - name: MakeRefreshNonce
        run_when_succeeded: [CheckRefresh]
        json_output: |-
          { "minted": true }
        hide_data_on_success: true
        post_transforms:
          - generate_bytes:
              key: nonce_access
              length: 12

      - name: SealRefreshed
        run_when_succeeded: [MakeRefreshNonce]
        json_output: |-
          { "sealed": true }
        hide_data_on_success: true
        post_transforms:
          - encrypt_value:
              key: access_sealed
              data: a|RefreshToken::body.access_token|
              secret_key: a|ap_var::OAUTH_ENC_KEY|
              nonce: a|MakeRefreshNonce::nonce_access|

      - name: StoreRefreshed
        run_when_succeeded: [SealRefreshed]
        database: main
        query: |-
          UPDATE oauth_connections
             SET access_value = $2,
                 access_nonce = $3,
                 access_tag   = $4,
                 expires_at   = NOW() + ($5 || ' seconds')::INTERVAL,
                 updated_at   = NOW()
           WHERE user_id = $1
          RETURNING user_id, expires_at
        params:
          - a|LoadConnection::user_id|
          - a|SealRefreshed::access_sealed.value|
          - a|SealRefreshed::access_sealed.nonce|
          - a|SealRefreshed::access_sealed.tag|
          - a|RefreshToken::body.expires_in->default(3600)|
        post_transforms:
          - extract_value: "[0]"

      # One terminal action for both branches. Exactly one of the two token
      # sources ran, so the other resolves to null and coalesce picks the winner.
      - name: Respond
        run_when_succeeded:
          at_least: 1
          actions: [DecryptCurrent, StoreRefreshed]
        database: main
        query: |-
          SELECT COALESCE($1, $2) AS access_token,
                 CASE WHEN $1 IS NOT NULL THEN 'cache' ELSE 'refreshed' END AS source
        params:
          - a|DecryptCurrent::access_token->default(null)|
          - a|RefreshToken::body.access_token->default(null)|
        post_transforms:
          - extract_value: "[0]"

  # GET /oauth/connections?limit=25
  # Deliberately returns NO token material — only who is connected and until when.
  oauth/connections:
    output: http
    method: GET
    summary: List connected users (never returns token material)
    tags: [oauth, postgres]

    actions:
      - name: ReadParams
        input: a|params|
        hide_data_on_success: true

      - name: ListConnections
        run_when_succeeded: [ReadParams]
        database: main
        query: |-
          SELECT user_id, scope, expires_at, updated_at,
                 (refresh_value IS NOT NULL) AS can_refresh,
                 (expires_at <= NOW())       AS expired
            FROM oauth_connections
           ORDER BY updated_at DESC
           LIMIT a|ReadParams::limit->default(25)|

seed.yml

name: OauthConnectSeed
description: Creates ap_urlencode + the oauth_states / oauth_connections tables (idempotent).

docs: true

# Required managed variables:
#   DATABASE_URL — Postgres connection string

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

interfaces:

  # POST /api/seed
  api/seed:
    output: http
    method: POST

    actions:
      # The percent-encoder the config depends on. airpipe has no url_encode
      # filter, so this is where OAuth values get escaped. '%' first, or the
      # escapes get re-escaped.
      - name: CreateUrlEncode
        database: main
        hide_data_on_success: true
        query: |
          CREATE OR REPLACE FUNCTION ap_urlencode(input TEXT) RETURNS TEXT AS $$
            SELECT replace(replace(replace(replace(replace(replace(replace(replace(replace(
                     coalesce(input, ''),
                     '%', '%25'),
                     ' ', '%20'),
                     '+', '%2B'),
                     '/', '%2F'),
                     ':', '%3A'),
                     '=', '%3D'),
                     '&', '%26'),
                     '?', '%3F'),
                     '#', '%23');
          $$ LANGUAGE SQL IMMUTABLE;

      - name: CreateStates
        database: main
        run_when_succeeded: [CreateUrlEncode]
        hide_data_on_success: true
        query: |
          CREATE TABLE IF NOT EXISTS oauth_states (
            state       TEXT         PRIMARY KEY,
            user_id     TEXT         NOT NULL,
            created_at  TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
            expires_at  TIMESTAMPTZ  NOT NULL,
            used_at     TIMESTAMPTZ
          );

      - name: CreateStatesIndex
        database: main
        run_when_succeeded: [CreateStates]
        hide_data_on_success: true
        query: |
          CREATE INDEX IF NOT EXISTS idx_oauth_states_expires_at
            ON oauth_states (expires_at);

      - name: CreateConnections
        database: main
        run_when_succeeded: [CreateStatesIndex]
        hide_data_on_success: true
        query: |
          CREATE TABLE IF NOT EXISTS oauth_connections (
            user_id        TEXT         PRIMARY KEY,
            access_value   TEXT         NOT NULL,
            access_nonce   TEXT         NOT NULL,
            access_tag     TEXT         NOT NULL,
            refresh_value  TEXT,
            refresh_nonce  TEXT,
            refresh_tag    TEXT,
            scope          TEXT         NOT NULL DEFAULT '',
            expires_at     TIMESTAMPTZ  NOT NULL,
            created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
            updated_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW()
          );

      - name: CreateConnectionsIndex
        database: main
        run_when_succeeded: [CreateConnections]
        hide_data_on_success: true
        query: |
          CREATE INDEX IF NOT EXISTS idx_oauth_connections_expires_at
            ON oauth_connections (expires_at);

      - name: Respond
        run_when_succeeded: [CreateConnectionsIndex]
        json_output: |-
          { "status": "seeded" }

schema.sql

-- OAuth2 authorization-code flow: single-use CSRF states + per-user connections.
--
-- ap_urlencode: airpipe has no url_encode interpolation filter, so percent-
-- encoding is done here. It escapes the RFC 3986 characters that actually occur
-- in OAuth values — authorization codes (Google's contain '/'), client secrets,
-- redirect URIs (':' and '/'), and space-separated scopes. '%' is escaped FIRST
-- so the escapes themselves are not re-escaped.
CREATE OR REPLACE FUNCTION ap_urlencode(input TEXT) RETURNS TEXT AS $$
  SELECT replace(replace(replace(replace(replace(replace(replace(replace(replace(
           coalesce(input, ''),
           '%', '%25'),
           ' ', '%20'),
           '+', '%2B'),
           '/', '%2F'),
           ':', '%3A'),
           '=', '%3D'),
           '&', '%26'),
           '?', '%3F'),
           '#', '%23');
$$ LANGUAGE SQL IMMUTABLE;

-- Single-use CSRF state. Consumed by the callback (used_at set) and expiring in
-- 10 minutes, so a replayed or stale callback cannot bind an account.
CREATE TABLE IF NOT EXISTS oauth_states (
  state       TEXT         PRIMARY KEY,
  user_id     TEXT         NOT NULL,
  created_at  TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
  expires_at  TIMESTAMPTZ  NOT NULL,
  used_at     TIMESTAMPTZ
);

CREATE INDEX IF NOT EXISTS idx_oauth_states_expires_at ON oauth_states (expires_at);

-- One row per connected user. Tokens are AES-256-GCM sealed by encrypt_value:
-- each needs its ciphertext, nonce and GCM tag to be decryptable, so all three
-- are stored. The encryption KEY is never here — it lives in OAUTH_ENC_KEY.
--
-- refresh_* is nullable on purpose: the provider returns a refresh token on the
-- FIRST consent only, and the upsert keeps the existing one when a later
-- response omits it.
CREATE TABLE IF NOT EXISTS oauth_connections (
  user_id        TEXT         PRIMARY KEY,
  access_value   TEXT         NOT NULL,
  access_nonce   TEXT         NOT NULL,
  access_tag     TEXT         NOT NULL,
  refresh_value  TEXT,
  refresh_nonce  TEXT,
  refresh_tag    TEXT,
  scope          TEXT         NOT NULL DEFAULT '',
  expires_at     TIMESTAMPTZ  NOT NULL,
  created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
  updated_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS idx_oauth_connections_expires_at ON oauth_connections (expires_at);

OAuth Connect

Endpoints​

Four things this gets right​

A pattern worth copying: zero rows fails an action​

URL-encoding​

Required variables​

Provider setup (Google shown)​

Database​

Usage example​

Verified​

Configuration​

config.yml​

seed.yml​

schema.sql​