#!/usr/bin/env python3 """Original fictional continuation exercise; Python 3.11+, standard library. Actual SQLite reads and committed grant changes, in deleted temporary files. No network, credentials, external data or production writes. The fixed local continuation handle is not a credential or an implementation of API tokens. Unsafe cases deliberately separate query binding from current authority. Two page observations do not establish completion, production isolation or a race-free authorization design. No retained read transaction spans the pages. """ import argparse from contextlib import ExitStack import json from pathlib import Path import sqlite3 import sys from tempfile import TemporaryDirectory CASES = ("same-query", "filter-unbound", "filter-bound", "tenant-unbound", "tenant-bound", "authority-cached", "authority-current") SUBJECT = "fictional-receiver" PAGE_SQL = ("SELECT tenant, item_id, state FROM items " "WHERE tenant = ? AND state = ? AND item_id > ? " "ORDER BY item_id LIMIT ?") GRANT_SQL = "SELECT allowed FROM grants WHERE subject = ? AND tenant = ?" def rows(connection, parameters): return [dict(row) for row in connection.execute(PAGE_SQL, parameters).fetchall()] def keys(records): return [[row["tenant"], row["item_id"]] for row in records] def run_case(case, page_size): with TemporaryDirectory(prefix="fde-pagination-scope-") as scratch, ExitStack() as stack: database = Path(scratch) / "fictional.sqlite3" def connect(): connection = sqlite3.connect(database, isolation_level=None) connection.row_factory = sqlite3.Row stack.callback(connection.close) return connection writer = connect() writer.executescript(""" CREATE TABLE items (tenant TEXT, item_id INTEGER, state TEXT, PRIMARY KEY (tenant, item_id)); CREATE TABLE grants (subject TEXT, tenant TEXT, allowed INTEGER, PRIMARY KEY (subject, tenant)); CREATE TABLE continuations (handle TEXT PRIMARY KEY, subject TEXT, tenant TEXT, state TEXT, last_id INTEGER); """) writer.executemany("INSERT INTO items VALUES (?, ?, ?)", [(tenant, number, "open") for tenant in ("east", "west") for number in (10, 20, 30, 40)] + [("east", 15, "closed"), ("east", 25, "closed")]) writer.executemany("INSERT INTO grants VALUES (?, ?, 1)", [(SUBJECT, "east"), (SUBJECT, "west")]) # Independently retained initial reference, before any request changes. original = [dict(row) for row in writer.execute( "SELECT tenant, item_id, state FROM items " "WHERE tenant = 'east' AND state = 'open' ORDER BY item_id").fetchall()] expected_window = original[page_size:page_size * 2] first_reader = connect() first_grant = bool(first_reader.execute(GRANT_SQL, (SUBJECT, "east")).fetchone()[0]) if not first_grant: raise RuntimeError("The fictional first request must be authorized") first_parameters = ["east", "open", 0, page_size] first = rows(first_reader, first_parameters) writer.execute("INSERT INTO continuations VALUES (?, ?, ?, ?, ?)", ("local-teaching-handle", SUBJECT, "east", "open", first[-1]["item_id"])) revocation = None if case.startswith("authority-"): statement = "UPDATE grants SET allowed = 0 WHERE subject = ? AND tenant = ?" writer.execute("BEGIN") changed = writer.execute(statement, (SUBJECT, "east")).rowcount writer.execute("COMMIT") revocation = {"sql": statement, "parameters": [SUBJECT, "east"], "changed_rows": changed, "committed": not writer.in_transaction} second_reader = connect() cursor = dict(second_reader.execute( "SELECT subject, tenant, state, last_id FROM continuations WHERE handle = ?", ("local-teaching-handle",)).fetchone()) requested_tenant = "west" if case.startswith("tenant-") else "east" requested_state = "closed" if case.startswith("filter-") else "open" request = {"subject": SUBJECT, "tenant": requested_tenant, "state": requested_state} same_scope = all(cursor[field] == request[field] for field in request) # Observer always reads current authority. The unsafe cached case # deliberately ignores this observation when deciding to release rows. current_grant = bool(second_reader.execute( GRANT_SQL, (SUBJECT, requested_tenant)).fetchone()[0]) use_cached_grant = case == "authority-cached" decision_grant = first_grant if use_cached_grant else current_grant bind_scope = not case.endswith("-unbound") if bind_scope and not same_scope: decision = "reject-scope-change" elif not decision_grant: decision = "reject-current-authority" else: decision = "read-second-page" second, second_query = [], None if decision == "read-second-page": parameters = [requested_tenant, requested_state, cursor["last_id"], page_size] second = rows(second_reader, parameters) second_query = {"sql": PAGE_SQL, "parameters": parameters} original_keys = keys(original) combined = first + second return { "case": case, "page_size": page_size, "initial_population": original, "expected_original_second_window": expected_window, "first_page": {"sql": PAGE_SQL, "parameters": first_parameters, "rows": first}, "saved_continuation": cursor, "second_request": request, "same_query_scope": same_scope, "scope_binding_enforced": bind_scope, "first_request_allowed": first_grant, "current_grant_observation": {"sql": GRANT_SQL, "parameters": [SUBJECT, requested_tenant], "allowed": current_grant}, "authority_used_for_decision": "cached-first-request" if use_cached_grant else "current-read", "revocation": revocation, "decision": decision, "second_page_executed": second_query is not None, "second_page_query": second_query, "second_page_rows": second, "same_as_original_second_window": second == expected_window, "combined_numeric_ids": [row["item_id"] for row in combined], "combined_scoped_keys": keys(combined), "outside_initial_population_keys": [key for key in keys(combined) if key not in original_keys], "released_without_current_grant_keys": keys(second) if not current_grant else [], "separate_request_connections": first_reader is not second_reader, "retained_read_transaction": first_reader.in_transaction or second_reader.in_transaction, } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("case", choices=("all",) + CASES) parser.add_argument("--page-size", type=int, choices=(1, 2, 3), default=2) args = parser.parse_args() selected = CASES if args.case == "all" else (args.case,) try: result = {"exercise": "fictional continuation receiving", "cases": [run_case(case, args.page_size) for case in selected]} except (OSError, sqlite3.Error, RuntimeError) as error: print(f"Lab setup failed: {error}", file=sys.stderr) return 2 print(json.dumps(result, indent=2)) return 0 # Completed observations include deliberate refusals and unsafe cases. if __name__ == "__main__": raise SystemExit(main())