Authentik loadtest using locust

AUTHENTIK_LOAD_TEST.md

This is a simple load test for a full Oauth2 authorize flow on Authentik.

Dependencies

• locust: Install with pip install locust (or however you install Python packages)

Seeding the database with test users

There's a script (python create_loadtest_users.py) that creates 1000 test users in a Authentik instance. AUTHENTIK_URL and AUTHENTIK_TOKEN environment variables need to be set.

Running the load test

Start locust with the included test: locust -f authentik_load_test.py. Open it's web UI and start the test. AUTHENTIK_URL and CLIENT_ID environment variables need to be set.

Disclaimer

All code was created with the assistance of AI. This is not meant as production-ready software

authentik_load_test.py

# OAuth2 Authorization Code Flow with PKCE against authentik
# using the default-authentication-flow (identification → password → redirect)
#
# Observed browser requests:
# 1. GET  /application/o/authorize/  (→ redirects to flow)
# 2. GET  /api/v3/flows/executor/default-authentication-flow/  (identification challenge)
# 3. POST /api/v3/flows/executor/default-authentication-flow/  {uid_field: "…"}
# 4. GET  /api/v3/flows/executor/default-authentication-flow/  (password challenge)
# 5. POST /api/v3/flows/executor/default-authentication-flow/  {password: "…"}
# 6. GET  /api/v3/flows/executor/default-authentication-flow/  (redirect challenge)
# 7. GET  /application/o/authorize/  (→ 302 to redirect_uri with ?code=…)
# 8. POST /application/o/token/  (exchange code for tokens)
#
# Usage:
#   locust -f authentik_load_test.py
#
# Load-test users:
#   loadtest_000@on.com … loadtest_999@on.com, password 12345678

import base64
import hashlib
import logging
import os
import random
import secrets
from urllib.parse import urlparse, parse_qs

from locust import HttpUser, task, between

logger = logging.getLogger(__name__)

# ── Constants ─────────────────────────────────────────────────────────────────

AUTHENTIK_URL = os.environ.get("AUTHENTIK_URL")
print(f"AUTHENTIK_URL={AUTHENTIK_URL}")
if not AUTHENTIK_URL:
    raise SystemExit("Provide an AUTHENTIK_URL as environment variable.")

CLIENT_ID = os.environ.get("CLIENT_ID")
if not CLIENT_ID:
    raise SystemExit(
        "Provide a CLIENT_ID as environment variable. "
        "The CLIENT_ID should belong to an Oauth provider that uses the "
        "default-authentication-flow as authentication flow"
    )
REDIRECT_URI = "http://localhost/callback"
SCOPES = "openid email profile"

FLOW_SLUG = "default-authentication-flow"
FLOW_URL = f"/api/v3/flows/executor/{FLOW_SLUG}/"

EMAIL_PATTERN = "loadtest_{:03d}@on.com"
PASSWORD = "12345678"

DEFAULT_HEADERS = { "User-Agent": "Locust, like Safari" }
API_HEADERS = { **DEFAULT_HEADERS, "Accept": "application/json" }


# ── Helpers ─────────────────────────────────────────────────────────────────


def random_user():
    """Return a (email, password) tuple for a random load-test user."""
    return EMAIL_PATTERN.format(random.randint(0, 999)), PASSWORD


def generate_pkce_pair():
    """Return (code_verifier, code_challenge) using S256."""
    code_verifier = (
        base64.urlsafe_b64encode(os.urandom(32)).rstrip(b"=").decode("ascii")
    )
    code_challenge = (
        base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode("ascii")).digest())
        .rstrip(b"=")
        .decode("ascii")
    )
    return code_verifier, code_challenge


def csrf_token(client):
    """Read the CSRF token that authentik sets as a cookie."""
    return client.cookies.get("authentik_csrf") or client.cookies.get("csrftoken") or ""


def log_response(resp, message=None, level=logging.DEBUG):
    """Log status, relevant headers, and body.

    Run locust with ``--loglevel DEBUG`` to see output.
    """
    headers_of_interest = {}
    for h in ("Location", "Content-Type", "Set-Cookie", "X-Authentik-Id"):
        val = resp.headers.get(h)
        if val:
            headers_of_interest[h] = val
    body = resp.text[:2000] if resp.text else "<empty>"
    msg_part = f" {message} |" if message else ""
    logger.log(
        level,
        "%s %s %s | status=%s | headers=%s | body=%s",
        msg_part,
        resp.request.method,
        resp.url,
        resp.status_code,
        headers_of_interest,
        body,
    )


def fail(resp, name, detail):
    """Log an error and mark the locust response as failed."""
    msg = f"{name} - {detail}"
    log_response(resp, msg, level=logging.ERROR)
    resp.failure(msg)


# ── Load test ─────────────────────────────────────────────────────────────────


class AuthentikAuthorizeWithSolidusAuthenticationFlow(HttpUser):
    host = AUTHENTIK_URL
    wait_time = between(1, 3)

    @task
    def full_authorize_flow(self):
        # Start with a clean cookie jar so every iteration exercises the
        # full authentication flow instead of reusing an existing session.
        self.client.cookies.clear()

        code_verifier, code_challenge = generate_pkce_pair()
        state = secrets.token_hex(32)

        authorize_params = {
            "client_id": CLIENT_ID,
            "redirect_uri": REDIRECT_URI,
            "response_type": "code",
            "scope": SCOPES,
            "state": state,
            "code_challenge": code_challenge,
            "code_challenge_method": "S256",
        }

        # ── 1. Initiate authorize (follows redirects → flow executor UI) ──
        name = "01 GET /application/o/authorize/ (start)"
        with self.client.get(
            "/application/o/authorize/",
            params=authorize_params,
            headers=DEFAULT_HEADERS,
            name=name,
            allow_redirects=True,
            catch_response=True,
        ) as resp:
            if resp.status_code != 200:
                fail(resp, name, f"expected 200, got {resp.status_code}")
                return
            content_type = resp.headers.get("Content-Type", "")
            if "text/html" not in content_type:
                fail(resp, name, f"expected text/html, got {content_type}")
                return
            resp.success()
            # The redirect chain lands on the flow-executor UI whose query
            # string encodes all OAuth + flow params.  The JS client passes
            # this verbatim as ?query=… to the REST API.
            flow_query = urlparse(resp.url).query

        email, password = random_user()

        # ── GET identification challenge ──────────────────────────────────────
        name = "02 GET flow-executor (identification)"
        with self.client.get(
            FLOW_URL,
            params={"query": flow_query},
            headers=API_HEADERS,
            name=name,
            catch_response=True,
        ) as resp:
            if resp.status_code != 200:
                fail(resp, name, f"expected 200, got {resp.status_code}")
                return
            body = resp.json()
            if body.get("component") != "ak-stage-identification":
                fail(
                    resp,
                    name,
                    f"expected ak-stage-identification, got {body.get('component')}",
                )
                return
            resp.success()

        # ── POST email / uid_field ────────────────────────────────────────────
        name = "03 POST flow-executor (email)"
        with self.client.post(
            FLOW_URL,
            params={"query": flow_query},
            headers={**API_HEADERS, "X-authentik-CSRF": csrf_token(self.client)},
            json={"uid_field": email},
            name=name,
            allow_redirects=False,
            catch_response=True,
        ) as resp:
            if resp.status_code != 302:
                fail(resp, name, f"expected 302, got {resp.status_code}")
                return
            location = resp.headers.get("Location", "")
            if FLOW_URL not in location:
                fail(resp, name, f"unexpected redirect to {location}")
                return
            resp.success()

        # ── GET password challenge ────────────────────────────────────────────
        name = "04 GET flow-executor (password stage)"
        with self.client.get(
            FLOW_URL,
            params={"query": flow_query},
            headers=API_HEADERS,
            name=name,
            catch_response=True,
        ) as resp:
            if resp.status_code != 200:
                fail(resp, name, f"expected 200, got {resp.status_code}")
                return
            body = resp.json()
            if body.get("component") not in ("ak-stage-prompt", "ak-stage-password"):
                fail(
                    resp,
                    name,
                    f"expected ak-stage-prompt or ak-stage-password, got {body.get('component')}",
                )
                return
            if body.get("component") == "ak-stage-prompt":
                fields = body.get("fields", [])
                if not any(f.get("field_key") == "password" for f in fields):
                    fail(resp, name, "no password field in prompt stage")
                    return
            resp.success()

        # ── POST password ─────────────────────────────────────────────────────
        name = "05 POST flow-executor (password)"
        with self.client.post(
            FLOW_URL,
            params={"query": flow_query},
            headers={**API_HEADERS, "X-authentik-CSRF": csrf_token(self.client)},
            json={"password": password},
            name=name,
            allow_redirects=False,
            catch_response=True,
        ) as resp:
            if resp.status_code != 302:
                fail(resp, name, f"expected 302, got {resp.status_code}")
                return
            location = resp.headers.get("Location", "")
            if FLOW_URL not in location:
                fail(resp, name, f"unexpected redirect to {location}")
                return
            resp.success()

        # ── GET redirect / consent challenge ──────────────────────────────────
        name = "06 GET flow-executor (redirect)"
        with self.client.get(
            FLOW_URL,
            params={"query": flow_query},
            headers=API_HEADERS,
            name=name,
            catch_response=True,
        ) as resp:
            if resp.status_code != 200:
                fail(resp, name, f"expected 200, got {resp.status_code}")
                return
            body = resp.json()
            if body.get("component") != "xak-flow-redirect":
                fail(resp, name, f"expected xak-flow-redirect, got {body.get('component')}")
                return
            resp.success()

        # ── 7. Authorize again (authenticated) → 302 with ?code= ─────────
        name = "07 GET /application/o/authorize/ (get code)"
        with self.client.get(
            "/application/o/authorize/",
            params=authorize_params,
            headers=DEFAULT_HEADERS,
            name=name,
            allow_redirects=False,
            catch_response=True,
        ) as resp:
            if resp.status_code not in (302, 303):
                fail(resp, name, f"expected redirect, got {resp.status_code}")
                return

            location = resp.headers.get("Location", "")
            qs = parse_qs(urlparse(location).query)

            if "code" not in qs:
                fail(resp, name, f"no code in Location: {location}")
                return

            auth_code = qs["code"][0]
            resp.success()

        # ── 8. Exchange authorization code for tokens ─────────────────────
        name = "08 POST /application/o/token/ (exchange)"
        with self.client.post(
            "/application/o/token/",
            name=name,
            headers=DEFAULT_HEADERS,
            data={
                "grant_type": "authorization_code",
                "client_id": CLIENT_ID,
                "code": auth_code,
                "redirect_uri": REDIRECT_URI,
                "code_verifier": code_verifier,
            },
            catch_response=True,
        ) as resp:
            if resp.status_code != 200:
                fail(resp, name, f"expected 200, got {resp.status_code}")
                return
            body = resp.json()
            missing = [
                k for k in ("access_token", "token_type", "id_token") if k not in body
            ]
            if missing:
                fail(resp, name, f"missing OIDC token fields: {missing}")
                return
            resp.success()

create_loadtest_users.py

#!/usr/bin/env python3
"""
Create loadtest users in Authentik via the REST API.

Idempotent: users that already exist are skipped (password is re-applied).

Required environment variables:
  AUTHENTIK_TOKEN   - API token with user-management permissions
  AUTHENTIK_URL     - Base URL of the Authentik instance, e.g. https://auth.example.com

Usage:
  python create_loadtest_users.py

Users created:
  username / email: loadtest_000@on.com ... loadtest_999@on.com
  password:         12345678
"""

import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

AUTHENTIK_TOKEN = os.environ.get("AUTHENTIK_TOKEN")
AUTHENTIK_URL = os.environ.get("AUTHENTIK_URL", "").rstrip("/")

USER_COUNT = 1000
USERNAME_PATTERN = "loadtest_{:03d}@on.com"
PASSWORD = "12345678"
MAX_WORKERS = 10  # concurrent API calls


def die(msg: str) -> None:
    print(f"ERROR: {msg}", file=sys.stderr)
    sys.exit(1)


if not AUTHENTIK_TOKEN:
    die("AUTHENTIK_TOKEN is not set.")
if not AUTHENTIK_URL:
    die("AUTHENTIK_URL is not set.")

API_BASE = f"{AUTHENTIK_URL}/api/v3"
HEADERS = {
    "Authorization": f"Bearer {AUTHENTIK_TOKEN}",
    "Content-Type": "application/json",
}


# ---------------------------------------------------------------------------
# API helpers
# ---------------------------------------------------------------------------


def get_user_pk(session: requests.Session, username: str) -> int | None:
    """Return the pk of an existing user, or None if they don't exist."""
    resp = session.get(
        f"{API_BASE}/core/users/",
        params={"username": username},
    )
    resp.raise_for_status()
    results = resp.json().get("results", [])
    for user in results:
        if user["username"] == username:
            return user["pk"]
    return None


def create_user(session: requests.Session, username: str, email: str) -> int:
    """Create a user and return their pk."""
    resp = session.post(
        f"{API_BASE}/core/users/",
        json={
            "username": username,
            "name": username,
            "email": email,
            "is_active": True,
        },
    )
    resp.raise_for_status()
    return resp.json()["pk"]


def set_password(session: requests.Session, pk: int, password: str) -> None:
    """Set the password for a user identified by pk."""
    resp = session.post(
        f"{API_BASE}/core/users/{pk}/set_password/",
        json={"password": password},
    )
    resp.raise_for_status()


# ---------------------------------------------------------------------------
# Per-user task
# ---------------------------------------------------------------------------


def ensure_user(index: int) -> tuple[int, str]:
    """
    Ensure a single loadtest user exists with the correct password.
    Returns (index, status) where status is 'created' or 'exists'.
    """
    username = email = USERNAME_PATTERN.format(index)

    # Each thread gets its own session for connection pooling.
    session = requests.Session()
    session.headers.update(HEADERS)

    pk = get_user_pk(session, username)
    created = False

    if pk is None:
        pk = create_user(session, username, email)
        created = True

    set_password(session, pk, PASSWORD)

    return index, "created" if created else "exists"


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------


def main() -> None:
    print(f"Target: {AUTHENTIK_URL}")
    print(
        f"Users:  {USER_COUNT} ({USERNAME_PATTERN.format(0)} ... {USERNAME_PATTERN.format(USER_COUNT - 1)})"
    )
    print(f"Workers:{MAX_WORKERS}")
    print()

    created = 0
    skipped = 0
    failed = 0

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        futures = {pool.submit(ensure_user, i): i for i in range(USER_COUNT)}

        completed = 0
        for future in as_completed(futures):
            completed += 1
            try:
                index, status = future.result()
                if status == "created":
                    created += 1
                else:
                    skipped += 1
            except Exception as exc:
                failed += 1
                i = futures[future]
                print(f"  FAIL  {USERNAME_PATTERN.format(i)}: {exc}")

            # Progress line (overwrite in place)
            print(
                f"\r  Progress: {completed}/{USER_COUNT}  "
                f"created={created}  skipped={skipped}  failed={failed}",
                end="",
                flush=True,
            )

    print()  # newline after progress
    print()
    print("Done.")
    print(f"  Created: {created}")
    print(f"  Skipped: {skipped} (already existed)")
    print(f"  Failed:  {failed}")

    if failed:
        sys.exit(1)


if __name__ == "__main__":
    main()