procurement-ingest/lambdas/api/pagination.py

"""Opaque cursor pagination over DynamoDB ``LastEvaluatedKey``.

The cursor is base64url(JSON(LastEvaluatedKey)). It is untrusted client input:
``decode_cursor`` validates shape strictly (a flat dict whose keys are exactly
a subset of the table's key attributes and whose values are non-empty strings)
and raises ``BadCursor`` -- mapped to HTTP 400 by the handler -- on anything
else, so a malformed or tampered cursor can never reach DynamoDB as an
arbitrary ``ExclusiveStartKey`` or surface as a 500.
"""

import base64
import binascii
import json

DEFAULT_LIMIT = 100
MAX_LIMIT = 500
_MAX_CURSOR_CHARS = 2048


class BadCursor(ValueError):
    pass


def clamp_limit(raw) -> int:
    if raw is None or raw == "":
        return DEFAULT_LIMIT
    try:
        value = int(raw)
    except (TypeError, ValueError):
        raise BadCursor("limit must be an integer") from None
    return max(1, min(MAX_LIMIT, value))


def encode_cursor(last_evaluated_key: dict) -> str:
    raw = json.dumps(last_evaluated_key, default=str, sort_keys=True)
    return base64.urlsafe_b64encode(raw.encode()).decode()


def decode_cursor(cursor: str, key_attrs: frozenset) -> dict:
    """Decode a cursor and require it to be EXACTLY the table's key attributes.

    ``key_attrs`` is the full key schema of the operation the cursor feeds
    (e.g. ``{work_order_id, comment_id}`` for the comments Query). An exact
    match -- not a subset -- is required: a partial composite key or a cursor
    minted for a different endpoint would otherwise pass a subset check, reach
    DynamoDB as an incomplete/inconsistent ``ExclusiveStartKey``, and raise a
    ValidationException that surfaces as a 500 (and pages the 5xx alarm). Here
    it fails closed as a 400 instead. Callers additionally pin the
    partition-key value to the request path (see ``wo_repo.list_comments``).
    """
    if len(cursor) > _MAX_CURSOR_CHARS:
        raise BadCursor("cursor too long")
    try:
        decoded = json.loads(base64.urlsafe_b64decode(cursor.encode()))
    except (binascii.Error, UnicodeDecodeError, json.JSONDecodeError, ValueError):
        raise BadCursor("cursor is not valid") from None
    if not isinstance(decoded, dict) or set(decoded) != key_attrs:
        raise BadCursor("cursor is not valid")
    for value in decoded.values():
        if not isinstance(value, str) or not value:
            raise BadCursor("cursor is not valid")
    return decoded