Bidirectional pagination and opaque cursors

Capsule overview

Your cursor pagination already works, it's performant, and it's tested. But it's missing two features that separate a toy implementation from a production one:

1. Bidirectional navigation. Up to here you only navigate "forward" (next_cursor). Real APIs also need "backward" — the "Previous" button in infinite scroll, or the client who wants to go back to the previous page without reloading everything.

2. A cursor signed with HMAC. Up to here the cursor is visually opaque (base64), but any client can decode it and manipulate it. This capsule shows you what happens if someone changes a field in the cursor (e.g. to read another tenant's data) and how you defend against it with 10 lines of code.

You'll come out with a cursor the client can't forge and an endpoint that supports direction=next and direction=prev with the same elegance.

Why signing the cursor matters

Imagine your API has multi-tenancy. The cursor includes tenant_id to filter in the WHERE:

{
    "v": 1,
    "t": "2026-04-15T10:23:45Z",
    "i": 12345,
    "tenant_id": 7,         # ← the current request's tenant
    "d": "next"
}

Encoded in base64: eyJ0ZW5hbnRfaWQiOjcsLi4ufQ.

A curious client decodes the cursor:

echo "eyJ0ZW5hbnRfaWQiOjcsLi4ufQ" | base64 -d
# {"tenant_id":7,"t":"2026-04-15T10:23:45Z","i":12345,"d":"next","v":1}

They change tenant_id to 8:

echo '{"tenant_id":8,"t":"2026-04-15T10:23:45Z","i":12345,"d":"next","v":1}' \
  | base64 | tr -d '='
# eyJ0ZW5hbnRfaWQiOjgsLi4ufQ

They send it to your API: GET /tasks?cursor=eyJ0ZW5hbnRfaWQiOjgsLi4ufQ.

Your API decodes the cursor, sees tenant_id=8, and returns tasks from tenant 8 — data the tenant 7 client should never see.

This is a real security incident — a data leak between tenants caused by tamperable cursors. And it has happened in production at serious companies (search the internet for "cursor manipulation pagination vulnerability").

The solution: HMAC

HMAC (Hash-based Message Authentication Code) is an algorithm that produces a cryptographic "seal" of a message using a secret key. Only someone who knows the key can generate or validate the seal.

Original cursor:    {"t": "...", "i": 12345, "tenant_id": 7}
Signed cursor:      base64({"t": "...", "i": 12345, "tenant_id": 7}) + "." + hmac_signature

Client tampers to:  base64({"t": "...", "i": 12345, "tenant_id": 8}) + "." + hmac_signature_OLD

Your API validates the HMAC:  the OLD signature doesn't match the new payload → you reject with HTTP 400

The result: the client can read the cursor (it's just base64), can try to manipulate it, but cannot regenerate a valid HMAC without the secret key. Your API rejects any cursor with an invalid signature.

HMAC in Python: the minimum code

Python has hmac and hashlib in the stdlib. Ten lines and you have a signed cursor.

import hmac
import hashlib
import base64
import json
from secrets import compare_digest


# The secret key must come from env vars in production
# Generate it with: python -c "import secrets; print(secrets.token_urlsafe(32))"
SECRET_KEY = b"your-secret-key-of-at-least-32-bytes-in-bytes"


def _sign(payload_b64: str) -> str:
    """Generates an HMAC-SHA256 of the payload, encoded in URL-safe base64."""
    sig = hmac.new(SECRET_KEY, payload_b64.encode("ascii"), hashlib.sha256).digest()
    return base64.urlsafe_b64encode(sig).rstrip(b"=").decode("ascii")


def encode_signed_cursor(payload: dict) -> str:
    """Encodes the payload as a signed cursor: 'base64(payload).hmac_signature'."""
    raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
    payload_b64 = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
    signature = _sign(payload_b64)
    return f"{payload_b64}.{signature}"


def decode_signed_cursor(cursor: str) -> dict:
    """Decodes and validates a signed cursor. Raises ValueError if the signature is invalid."""
    if "." not in cursor:
        raise ValueError("Cursor without a signature")

    payload_b64, signature = cursor.rsplit(".", 1)
    expected_sig = _sign(payload_b64)

    # compare_digest is resistant to timing attacks
    if not compare_digest(signature, expected_sig):
        raise ValueError("Invalid cursor signature")

    padding = "=" * (-len(payload_b64) % 4)
    raw = base64.urlsafe_b64decode(payload_b64 + padding)
    return json.loads(raw)

Usage:

# Generate a cursor
payload = {"v": 1, "t": "2026-04-15T10:23:45Z", "i": 12345, "tenant_id": 7}
signed = encode_signed_cursor(payload)
print(signed)
# eyJ0ZW5hbnRfaWQiOjcsInQiOiIyMDI2LTA0LTE1VDEwOjIzOjQ1WiIsImkiOjEyMzQ1LCJ2IjoxfQ.k3jH9...

# Validate and decode
decoded = decode_signed_cursor(signed)
print(decoded)
# {'v': 1, 't': '2026-04-15T10:23:45Z', 'i': 12345, 'tenant_id': 7}

# Tampering attempt
manipulated_payload = "eyJ0ZW5hbnRfaWQiOjgsLi4ufQ"  # client changed it to tenant 8
manipulated_cursor = f"{manipulated_payload}.old_signature_of_the_original_payload"

decode_signed_cursor(manipulated_cursor)
# ValueError: Invalid cursor signature

Three important details

1. secrets.compare_digest, not ==. Comparing strings with == is vulnerable to timing attacks: the comparison stops at the first differing character, and the attacker can measure the time to deduce the correct bytes. compare_digest always takes the same time regardless of where the strings differ.

2. The secret key has to be genuinely secret.

• Generate it with python -c "import secrets; print(secrets.token_urlsafe(32))".
• Store it in an environment variable (CURSOR_SECRET_KEY), not hardcoded.
• Rotate it periodically (this capsule doesn't cover it, but the idea: include a key version in the cursor to support two active keys during rotation).

3. HMAC is not encryption. The client can read the cursor's payload (it's base64). It cannot modify it without invalidating the signature. If you want to hide the contents completely, add encryption on top (e.g. Fernet from cryptography), but typically HMAC alone is enough — the cursor's content isn't secret, what matters is that it can't be tampered with.

Bidirectional pagination: the model

Up to here your cursor only supports direction=next:

WHERE (created_at, id) < ($1, $2)
ORDER BY created_at DESC, id DESC LIMIT 50;

For direction=prev, you want items that come before the cursor (more recent). There are two approaches:

Approach 1: invert the comparison + invert the order + invert the result

-- direction=prev: items with (created_at, id) > cursor
SELECT ... FROM tasks
WHERE (created_at, id) > ($1, $2)
ORDER BY created_at ASC, id ASC  -- Inverted!
LIMIT 50;

-- Then: reverse the result's order in code (to keep the DESC order visible to the client)
items = items[::-1]

This approach keeps the ORDER BY correct at the SQL level (it can use the index in either direction) and reverses in memory.

Approach 2: two symmetric queries

-- direction=next
WHERE (created_at, id) < ($1, $2) ORDER BY created_at DESC, id DESC LIMIT 50;

-- direction=prev
WHERE (created_at, id) > ($1, $2) ORDER BY created_at DESC, id DESC LIMIT 50;

This is conceptually simpler, but it returns the oldest items first when you do prev — not what the client expects.

Recommendation: use approach 1. Subtler but more correct.

The cursor model with direction

{
    "v": 1,
    "t": "2026-04-15T10:23:45Z",
    "i": 12345,
    "d": "next"  # or "prev"
}

When you return a response, you generate two cursors:

• next_cursor: with d="next", pointing at the last returned item.
• previous_cursor: with d="prev", pointing at the first returned item.

{
  "items": [
    {"id": 12345, ...},  ← previous_cursor points here
    {"id": 12344, ...},
    {"id": 12343, ...}   ← next_cursor points here
  ],
  "next_cursor": "eyJ0Ijoi...",
  "previous_cursor": "eyJ0Ijoi...",
  "has_more": true
}

End-to-end implementation

app/pagination.py updated

"""Cursor pagination with HMAC and bidirectional support."""
import base64
import hashlib
import hmac
import json
import os
from datetime import datetime
from secrets import compare_digest
from typing import Any, Literal

CURSOR_VERSION = 1

# Secret key — in production, via an env var
SECRET_KEY = os.environ.get(
    "CURSOR_SECRET_KEY",
    "dev-only-key-NOT-for-prod-min-32-bytes-please"
).encode("utf-8")


class CursorError(ValueError):
    """Malformed, incompatible, or invalidly signed cursor."""


# --- HMAC helpers ---

def _sign(payload_b64: str) -> str:
    sig = hmac.new(SECRET_KEY, payload_b64.encode("ascii"), hashlib.sha256).digest()
    return base64.urlsafe_b64encode(sig).rstrip(b"=").decode("ascii")


def _verify(payload_b64: str, signature: str) -> bool:
    expected = _sign(payload_b64)
    return compare_digest(signature, expected)


# --- Signed encode / decode ---

def encode_signed_cursor(payload: dict[str, Any]) -> str:
    raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
    payload_b64 = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
    signature = _sign(payload_b64)
    return f"{payload_b64}.{signature}"


def decode_signed_cursor(cursor: str) -> dict[str, Any]:
    if "." not in cursor:
        raise CursorError("Cursor without a signature")

    payload_b64, signature = cursor.rsplit(".", 1)

    if not _verify(payload_b64, signature):
        raise CursorError("Invalid cursor signature")

    try:
        padding = "=" * (-len(payload_b64) % 4)
        raw = base64.urlsafe_b64decode(payload_b64 + padding)
        decoded = json.loads(raw)
    except (ValueError, json.JSONDecodeError) as e:
        raise CursorError("Malformed cursor") from e

    if decoded.get("v") != CURSOR_VERSION:
        raise CursorError(f"Cursor version {decoded.get('v')} not supported")

    return decoded


# --- Task-specific helpers ---

Direction = Literal["next", "prev"]


def make_task_cursor(
    created_at: datetime,
    last_id: int,
    direction: Direction,
) -> str:
    return encode_signed_cursor({
        "v": CURSOR_VERSION,
        "t": created_at.isoformat().replace("+00:00", "Z"),
        "i": last_id,
        "d": direction,
    })


def parse_task_cursor(cursor: str) -> tuple[datetime, int, Direction]:
    decoded = decode_signed_cursor(cursor)
    try:
        ts = datetime.fromisoformat(decoded["t"].replace("Z", "+00:00"))
        last_id = int(decoded["i"])
        direction = decoded.get("d", "next")
        if direction not in ("next", "prev"):
            raise CursorError(f"Invalid direction: {direction}")
    except (KeyError, ValueError, TypeError) as e:
        raise CursorError("Cursor with invalid fields") from e
    return ts, last_id, direction

app/repositories/tasks.py updated

"""Pagination with a signed, bidirectional cursor."""
from sqlalchemy import select, tuple_
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import Task
from app.pagination import (
    CursorError,
    make_task_cursor,
    parse_task_cursor,
)
from app.schemas import Page, TaskOut


async def list_tasks_paginated(
    session: AsyncSession,
    cursor: str | None = None,
    limit: int = 50,
) -> Page[TaskOut]:
    direction = "next"  # default when there's no cursor
    cursor_t = None
    cursor_i = None

    if cursor is not None:
        cursor_t, cursor_i, direction = parse_task_cursor(cursor)

    if direction == "next":
        # Items AFTER the cursor (older ones in a DESC sort)
        stmt = select(Task).order_by(Task.created_at.desc(), Task.id.desc())
        if cursor_t is not None:
            stmt = stmt.where(
                tuple_(Task.created_at, Task.id) < tuple_(cursor_t, cursor_i)
            )
    else:
        # direction == "prev"
        # Items BEFORE the cursor (more recent ones in a DESC sort)
        # Trick: invert the order to ASC to use the index efficiently,
        # then reverse the result in memory.
        stmt = select(Task).order_by(Task.created_at.asc(), Task.id.asc())
        if cursor_t is not None:
            stmt = stmt.where(
                tuple_(Task.created_at, Task.id) > tuple_(cursor_t, cursor_i)
            )

    stmt = stmt.limit(limit + 1)
    result = await session.execute(stmt)
    rows = list(result.scalars().all())

    has_more = len(rows) > limit
    page_items = rows[:limit]

    # If it was prev, reverse the order to keep DESC visible to the client
    if direction == "prev":
        page_items.reverse()

    # Generate cursors: next points at the last one, previous points at the first
    next_cursor = None
    previous_cursor = None

    if page_items:
        first = page_items[0]
        last = page_items[-1]

        # next_cursor only if there are more items forward
        # In direction=next, has_more tells us
        # In direction=prev, there's always a next one (the original cursor)
        if direction == "next" and has_more:
            next_cursor = make_task_cursor(last.created_at, last.id, "next")
        elif direction == "prev":
            next_cursor = make_task_cursor(last.created_at, last.id, "next")

        # previous_cursor only if there are items backward
        # In direction=next with a cursor: there's always a previous
        # In direction=prev: has_more tells us
        if direction == "next" and cursor_t is not None:
            previous_cursor = make_task_cursor(first.created_at, first.id, "prev")
        elif direction == "prev" and has_more:
            previous_cursor = make_task_cursor(first.created_at, first.id, "prev")

    return Page[TaskOut](
        items=[TaskOut.model_validate(t) for t in page_items],
        next_cursor=next_cursor,
        previous_cursor=previous_cursor,
        has_more=has_more,
    )

app/schemas.py with previous_cursor

class Page(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: str | None = None
    previous_cursor: str | None = None
    has_more: bool

Worked example: complete bidirectional navigation

# round_trip_bidi.py
"""
Demo of the bidirectional flow: navigate 5 pages forward, 2 back, 1 forward.
"""
import asyncio
import httpx


BASE = "http://localhost:8000"


async def main():
    async with httpx.AsyncClient() as client:
        # Page 1 (no cursor)
        r = await client.get(f"{BASE}/tasks?limit=2")
        p1 = r.json()
        print(f"Page 1: items={[t['id'] for t in p1['items']]}, "
              f"next={p1['next_cursor'][:20] if p1['next_cursor'] else None}..., "
              f"prev={p1['previous_cursor']}")

        # Page 2 (with page 1's next_cursor)
        r = await client.get(f"{BASE}/tasks?limit=2&cursor={p1['next_cursor']}")
        p2 = r.json()
        print(f"Page 2: items={[t['id'] for t in p2['items']]}, "
              f"prev={p2['previous_cursor'][:20]}...")

        # Page 3
        r = await client.get(f"{BASE}/tasks?limit=2&cursor={p2['next_cursor']}")
        p3 = r.json()
        print(f"Page 3: items={[t['id'] for t in p3['items']]}")

        # Go back to page 2 (with page 3's previous_cursor)
        r = await client.get(f"{BASE}/tasks?limit=2&cursor={p3['previous_cursor']}")
        back_to_p2 = r.json()
        print(f"Back to Page 2: items={[t['id'] for t in back_to_p2['items']]}")

        # Verification: the IDs must match
        assert [t['id'] for t in p2['items']] == [t['id'] for t in back_to_p2['items']]
        print("✅ Bidirectional is consistent: page 2 → 3 → back to 2 returns the same items")


if __name__ == "__main__":
    asyncio.run(main())

Expected output:

Page 1: items=[9847, 234], next=eyJkIjoibmV4dCIsImkiO..., prev=None
Page 2: items=[5621, 1842], prev=eyJkIjoicHJldiIsImkiO...
Page 3: items=[3417, 8902]
Back to Page 2: items=[5621, 1842]
✅ Bidirectional is consistent: page 2 → 3 → back to 2 returns the same items

Reading it:

• Page 1 has no previous_cursor (it's the first one).
• Intermediate pages have both cursors.
• Navigating forward and backward is consistent: the "back" returns exactly the same items.

Why does this matter in real work?

1. Defense against cursor manipulation. Multi-tenancy with a cursor and no HMAC is a security anti-pattern. If your cursor includes tenant_id, user_id, or any isolation field, HMAC is mandatory, not optional. Without a signature, the client can manipulate those fields.

Important note: in TaskFlow (module 8) you're going to use Row-Level Security (RLS) for multi-tenancy at the DB level. RLS protects you even if the cursor is tampered with, because PostgreSQL filters rows by tenant_id regardless of the query's WHERE. But don't relax the cursor signature because of that — defense in depth means you have both layers.

2. Pagination UX in mobile apps. Mobile apps typically have "pull to refresh" (going backward) and "infinite scroll" (going forward). Without bidirectional support, "pull to refresh" requires reloading everything from the start — poor UX. With bidirectional support, you only reload the current page.

3. Admin tool APIs. Internal tools with "next / previous / jump" to navigate records (e.g. a calendar event editor). Bidirectional is expected by the UX.

4. Senior differentiation. Any dev can implement next_cursor. Few can articulate bidirectional with correct handling of both cases. And very few know what HMAC is and why it's used in cursors. This is what stands out in code reviews and interviews.

Traps and common mistakes

Mistake 1 (conceptual): thinking HMAC is encryption

Symptom: "I signed the cursor, so the contents are secret."

Why it's wrong: HMAC signs, it doesn't encrypt. The client can read the payload (it's base64), but can't modify it without invalidating the signature. If the cursor's contents are genuinely secret (e.g. it includes user_email), you need additional encryption (Fernet, AES-GCM).

How to tell: ask yourself "can the client read the cursor's contents without causing harm?". If the answer is "yes" (the typical case), HMAC alone is enough. If the answer is "no" (rare), add encryption.

Mistake 2 (practical): comparing signatures with == instead of compare_digest

Symptom: the test works, the code passes code review, and in production nobody notices anything.

Why it's wrong: comparing strings with == cuts the comparison off at the first differing character. An attacker with access to precise timings (e.g. behind a cooperative load balancer) can deduce the signature's bytes one by one.

How to fix it: always use secrets.compare_digest(actual, expected). It takes constant time regardless of where the strings differ. It's ONE line of code that closes an entire class of attacks.

Mistake 3 (conceptual): hardcoding the secret key

Symptom: SECRET_KEY = b"your-default-secret-key" in the code.

Why it's wrong:

• The key stays in git history forever.
• Anyone with repo access (developers, CI/CD, scanning tools) can see it.
• If you rotate it, you have to change the code.

How to fix it:

SECRET_KEY = os.environ["CURSOR_SECRET_KEY"].encode("utf-8")
# No default — fail fast if it isn't configured

In production, the key comes from a secret manager (AWS Secrets Manager, HashiCorp Vault, a K8s Secret).

Mistake 4 (practical): not handling the "old-version cursor after key rotation" case

Symptom: you rotate the secret key. The cursors clients saved with the previous key start getting rejected with HTTP 400.

Why it happens: HMAC with the new key doesn't validate signatures made with the old key. That's what you want (the old key is revoked), but without a transition strategy, every saved cursor breaks at once.

How to handle it:

# Option A: two active keys for a period (key rotation)
KEYS = {
    "v1": os.environ["CURSOR_KEY_V1"],  # new
    "v0": os.environ["CURSOR_KEY_V0"],  # old, still accepted
}

def encode_signed_cursor(payload):
    payload["k"] = "v1"  # sign with the new one
    ...

def decode_signed_cursor(cursor):
    payload_b64, sig = cursor.rsplit(".", 1)
    decoded_preliminary = json.loads(base64.urlsafe_b64decode(payload_b64 + "=="))
    key_id = decoded_preliminary.get("k", "v0")  # default to the old one
    if key_id not in KEYS:
        raise CursorError("Unknown cursor key")
    expected_sig = hmac.new(KEYS[key_id], ..., ...).hexdigest()
    ...

After a period (e.g. 90 days), you retire KEYS["v0"].

Option B (simpler): you accept that old cursors break. Appropriate for ephemeral sessions. Not appropriate if clients persist cursors for days.

Mistake 5 (edge case): bidirectional isn't perfectly reversible when there are insertions

Symptom: you're on page 5, someone inserts new items, you hit "previous" to go back to page 4, but you see items that weren't there before.

Why it happens: cursor pagination is stable under insertions for the next flow. For prev, items inserted with a created_at greater than the cursor do show up in the "previous page" — they're new items.

How to tell: this is expected behavior, not a bug. A cursor with direction=prev means "items newer than the cursor," which includes the freshly inserted ones.

How to handle it (if the UX requires it):

• Document it: "previous may show new items that were inserted after your current page."
• If the client wants "exactly the same page 4 I saw before," add an as_of_timestamp to the cursor that filters WHERE created_at <= as_of_timestamp — it pins the "view" of the dataset.

Exercises

Exercise 1: test resistance to tampering

Generate a cursor with encode_signed_cursor. Manually manipulate a field in the payload (change tenant_id). Prove that decode_signed_cursor rejects the tampered cursor.

# test_hmac_resistance.py
import json
import base64
import pytest
from app.pagination import (
    encode_signed_cursor,
    decode_signed_cursor,
    CursorError,
)


def test_legitimate_cursor_decodes():
    payload = {"v": 1, "t": "2026-04-15T10:23:45Z", "i": 12345, "tenant_id": 7}
    signed = encode_signed_cursor(payload)
    decoded = decode_signed_cursor(signed)
    assert decoded == payload


def test_tampered_cursor_is_rejected():
    # Generate a legitimate cursor
    payload = {"v": 1, "t": "2026-04-15T10:23:45Z", "i": 12345, "tenant_id": 7}
    signed = encode_signed_cursor(payload)
    payload_b64, signature = signed.rsplit(".", 1)

    # Tamper with the payload (change tenant_id from 7 to 8)
    bad_payload = {"v": 1, "t": "2026-04-15T10:23:45Z", "i": 12345, "tenant_id": 8}
    bad_payload_b64 = base64.urlsafe_b64encode(
        json.dumps(bad_payload, separators=(",", ":"), sort_keys=True).encode()
    ).rstrip(b"=").decode()

    # The attacker tries to use the old signature with the new payload
    manipulated = f"{bad_payload_b64}.{signature}"

    with pytest.raises(CursorError, match="Invalid cursor signature"):
        decode_signed_cursor(manipulated)


def test_tampered_signature_is_rejected():
    payload = {"v": 1, "t": "2026-04-15T10:23:45Z", "i": 12345}
    signed = encode_signed_cursor(payload)
    payload_b64, _ = signed.rsplit(".", 1)
    bad_signature = "anysignature1234"
    manipulated = f"{payload_b64}.{bad_signature}"

    with pytest.raises(CursorError, match="Invalid cursor signature"):
        decode_signed_cursor(manipulated)


def test_unsigned_cursor_is_rejected():
    # A cursor without the "." separator
    bad = "eyJ0Ijoi..."
    with pytest.raises(CursorError, match="Cursor without a signature"):
        decode_signed_cursor(bad)

pytest test_hmac_resistance.py -v
# 4 passed in 0.05s

Exercise 2: implement key rotation (two active keys)

Implement support for two simultaneously active keys. New signatures use the new key (v1); signatures with the old key (v0) keep being accepted for a period.

# app/pagination_rotation.py
import hmac
import hashlib
import base64
import json
import os
from secrets import compare_digest

# Two active keys
KEYS = {
    "v1": os.environ.get("CURSOR_KEY_V1", "new-secret-key").encode("utf-8"),
    "v0": os.environ.get("CURSOR_KEY_V0", "old-secret-key").encode("utf-8"),
}
ACTIVE_KEY = "v1"


def _sign_with_key(payload_b64: str, key: bytes) -> str:
    sig = hmac.new(key, payload_b64.encode("ascii"), hashlib.sha256).digest()
    return base64.urlsafe_b64encode(sig).rstrip(b"=").decode("ascii")


def encode_signed_cursor(payload: dict) -> str:
    """Signs with the active key, including the key_id in the payload."""
    payload = {**payload, "k": ACTIVE_KEY}  # mark which key was used
    raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
    payload_b64 = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
    signature = _sign_with_key(payload_b64, KEYS[ACTIVE_KEY])
    return f"{payload_b64}.{signature}"


def decode_signed_cursor(cursor: str) -> dict:
    if "." not in cursor:
        raise ValueError("Cursor without a signature")

    payload_b64, signature = cursor.rsplit(".", 1)

    # Decode preliminarily to read the key_id
    padding = "=" * (-len(payload_b64) % 4)
    raw = base64.urlsafe_b64decode(payload_b64 + padding)
    decoded = json.loads(raw)

    key_id = decoded.get("k", "v0")  # default to v0 for legacy cursors without "k"
    if key_id not in KEYS:
        raise ValueError(f"Unknown cursor key: {key_id}")

    expected = _sign_with_key(payload_b64, KEYS[key_id])
    if not compare_digest(signature, expected):
        raise ValueError("Invalid signature")

    return decoded


# Test
def test_accepts_both_keys():
    # Create a cursor with the active key (v1)
    payload = {"v": 1, "t": "2026-04-15T10:23:45Z", "i": 12345}
    cursor_v1 = encode_signed_cursor(payload)
    decoded = decode_signed_cursor(cursor_v1)
    assert decoded.get("k") == "v1"

    # Create a cursor MANUALLY signed with the old key (simulating a legacy cursor)
    legacy_payload = {**payload, "k": "v0"}
    raw = json.dumps(legacy_payload, separators=(",", ":"), sort_keys=True).encode()
    legacy_b64 = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
    legacy_sig = _sign_with_key(legacy_b64, KEYS["v0"])
    cursor_v0 = f"{legacy_b64}.{legacy_sig}"

    # Decoding v0 still works
    decoded_v0 = decode_signed_cursor(cursor_v0)
    assert decoded_v0.get("k") == "v0"

Exercise 3: tests of the complete bidirectional flow

Write tests that validate: navigate forward 3 pages, navigate back 2 pages, verify the items are the same.

# tests/test_bidirectional.py
import pytest
from datetime import datetime, timedelta, timezone

from app.models import Task

pytestmark = pytest.mark.asyncio


async def _seed_5_tasks(session):
    now = datetime.now(timezone.utc).replace(microsecond=0)
    tasks = [
        Task(title=f"task_{i}", created_at=now - timedelta(minutes=i))
        for i in range(5)
    ]
    session.add_all(tasks)
    await session.commit()
    for t in tasks:
        await session.refresh(t)
    return tasks


async def test_navigating_forward_and_back_is_consistent(client, session):
    await _seed_5_tasks(session)

    # Page 1: items 0, 1
    r = await client.get("/tasks?limit=2")
    p1 = r.json()
    p1_ids = [t["id"] for t in p1["items"]]
    assert len(p1_ids) == 2
    assert p1["previous_cursor"] is None  # there's nothing before page 1
    assert p1["next_cursor"] is not None

    # Page 2: items 2, 3
    r = await client.get(f"/tasks?limit=2&cursor={p1['next_cursor']}")
    p2 = r.json()
    p2_ids = [t["id"] for t in p2["items"]]
    assert len(p2_ids) == 2
    assert p2["previous_cursor"] is not None  # now there IS a previous

    # Page 3: item 4 (only 1, it's the last)
    r = await client.get(f"/tasks?limit=2&cursor={p2['next_cursor']}")
    p3 = r.json()
    p3_ids = [t["id"] for t in p3["items"]]
    assert len(p3_ids) == 1
    assert p3["has_more"] is False
    assert p3["next_cursor"] is None
    assert p3["previous_cursor"] is not None

    # Go backward: previous from p3 → should be p2
    r = await client.get(f"/tasks?limit=2&cursor={p3['previous_cursor']}")
    back_to_p2 = r.json()
    assert [t["id"] for t in back_to_p2["items"]] == p2_ids

    # Go backward again: previous from back_to_p2 → should be p1
    r = await client.get(f"/tasks?limit=2&cursor={back_to_p2['previous_cursor']}")
    back_to_p1 = r.json()
    assert [t["id"] for t in back_to_p1["items"]] == p1_ids


async def test_previous_on_first_page_is_none(client, session):
    await _seed_5_tasks(session)

    # Page 1 without a cursor: previous_cursor must be None
    r = await client.get("/tasks?limit=2")
    assert r.json()["previous_cursor"] is None


async def test_items_cannot_repeat_while_navigating(client, session):
    await _seed_5_tasks(session)

    seen = set()
    cursor = None
    while True:
        url = "/tasks?limit=2"
        if cursor:
            url += f"&cursor={cursor}"
        r = await client.get(url)
        body = r.json()
        for item in body["items"]:
            assert item["id"] not in seen, "Duplicate item!"
            seen.add(item["id"])
        if not body["has_more"]:
            break
        cursor = body["next_cursor"]

    assert len(seen) == 5  # saw them all without duplicating

Exercise 4: add tenant_id to the cursor and validate HMAC in multi-tenancy

Assume your API is multi-tenant and the cursor includes tenant_id. Add HMAC and a test that demonstrates that changing tenant_id in the cursor fails.

# app/pagination_multitenant.py
def make_task_cursor_mt(
    tenant_id: int,
    created_at: datetime,
    last_id: int,
    direction: Direction,
) -> str:
    return encode_signed_cursor({
        "v": CURSOR_VERSION,
        "ti": tenant_id,            # tenant_id included
        "t": created_at.isoformat().replace("+00:00", "Z"),
        "i": last_id,
        "d": direction,
    })


def parse_task_cursor_mt(cursor: str) -> tuple[int, datetime, int, Direction]:
    decoded = decode_signed_cursor(cursor)
    try:
        tenant_id = int(decoded["ti"])
        ts = datetime.fromisoformat(decoded["t"].replace("Z", "+00:00"))
        last_id = int(decoded["i"])
        direction = decoded.get("d", "next")
    except (KeyError, ValueError, TypeError) as e:
        raise CursorError("Cursor with invalid fields") from e
    return tenant_id, ts, last_id, direction


# In the endpoint:
async def get_tasks_mt(
    cursor: str | None = None,
    limit: int = 50,
    current_tenant: int = Depends(get_current_tenant),
):
    if cursor:
        cursor_tenant, cursor_t, cursor_i, direction = parse_task_cursor_mt(cursor)
        # CRITICAL validation: the cursor's tenant_id must match the current tenant
        if cursor_tenant != current_tenant:
            raise HTTPException(
                status_code=403,
                detail="Cursor from another tenant"
            )
        # Continue with the query, filtering by current_tenant
        ...

async def test_cannot_use_another_tenants_cursor(client, session):
    # Create tenant 1's tasks
    ...

    # Generate a LEGITIMATE cursor for tenant 1
    cursor = make_task_cursor_mt(tenant_id=1, ..., direction="next")

    # Try to use the cursor in a tenant 2 request
    r = await client.get(
        f"/tasks?cursor={cursor}",
        headers={"X-Tenant-Id": "2"},  # tenant 2's request
    )
    assert r.status_code == 403


async def test_cursor_tampered_to_change_tenant_fails(client, session):
    # Generate a legitimate cursor for tenant 1
    cursor = make_task_cursor_mt(tenant_id=1, ..., direction="next")
    payload_b64, sig = cursor.rsplit(".", 1)

    # Try to tamper with the payload to switch to tenant 2
    bad_payload = {"v": 1, "ti": 2, "t": "...", "i": ..., "d": "next"}
    bad_b64 = base64.urlsafe_b64encode(
        json.dumps(bad_payload, separators=(",", ":"), sort_keys=True).encode()
    ).rstrip(b"=").decode()
    manipulated_cursor = f"{bad_b64}.{sig}"  # old signature

    # The API rejects it on the invalid HMAC (it never even reaches the tenant validation)
    r = await client.get(
        f"/tasks?cursor={manipulated_cursor}",
        headers={"X-Tenant-Id": "2"},
    )
    assert r.status_code == 400
    assert "signature" in r.json()["detail"]

Summary and next step

In this capsule you added:

• HMAC with SHA256 signs the cursor so the client can't tamper with it. Ten lines of code that close an entire class of security vulnerabilities.
• secrets.compare_digest for comparing signatures in a way that's resistant to timing attacks. Never use == to compare hashes/signatures.
• SECRET_KEY from env vars, never hardcoded. A rotation strategy with two active keys to avoid breaking saved cursors.
• Bidirectional pagination with direction=next | prev encoded in the cursor. The approach: invert the WHERE and reverse the result in memory to keep the sort visible to the client.
• previous_cursor points at the page's first item; next_cursor points at the last. They're only generated when there are items in that direction.
• An opaque cursor with HMAC = the de facto standard for serious public APIs (Stripe, GitHub).

Before moving on you should be able to:

• Implement encode_signed_cursor / decode_signed_cursor from memory
• Explain why compare_digest matters more than == for signatures
• Describe how the bidirectional flow works without looking at the code
• Justify HMAC to a colleague who asks "isn't base64 enough?"

Next capsule — Pagination with filters and dynamic ordering. You already have cursor pagination that's performant, secure, and bidirectional. The reality is that APIs aren't just GET /tasks — they're GET /tasks?status=active&priority=high&sort=updated_at. You're going to learn to combine a cursor with filters without breaking opacity, with parameterizable ordering that doesn't allow SQL injection, and with the right balance between flexibility and simplicity.

Resources

1. Python hmac module — the official reference, includes compare_digest.
2. Python secrets module — token_urlsafe for generating keys, compare_digest for secure comparison.
3. OWASP — "Secure Pseudo-Random Number Generators" — why to use secrets and not random for anything security-related.
4. Cloudflare — "Why HMAC matters for API authentication" — a clear explanation of HMAC in the context of APIs.
5. JWT.io — "Introduction" — a conceptual reference: JWT is the classic case of an HMAC-signed payload. Cursor pagination is the same pattern applied to pagination.
6. Stripe — "Pagination" (cursor signed) — the public API's documentation. Note that their cursors are opaque but acceptable for public use because they're signed internally.
7. GitHub Engineering — "Modeling deletes in cursor pagination" — a real case of how GitHub handles cursor pagination with deletes and key rotation.
8. Aaron Stannard — "Why HTTPS for everything?" — a reminder that HMAC doesn't protect you if the cursor travels over plain HTTP. Always HTTPS in production.

Module 1 — SQL Patterns for Production APIs Guide

Next capsule: Pagination with filters and dynamic ordering — the real case with ?status=active&sort=updated_at.