"""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