"""Insert one inert queued task into the existing disposable native-test database. Stdin accepts only dataRoot, ownerId and ownerEmail. No notification, worker, provider or product endpoint is created. --self-test uses its own temporary DBs. """ from __future__ import annotations import json import re import sqlite3 import sys import tempfile import uuid from contextlib import closing, contextmanager from datetime import datetime, timezone from pathlib import Path class SeedDenied(ValueError): pass SCHEMA = { "Id": ("TEXT", 1, 1), "OwnerUserId": ("TEXT", 1, 0), "TaskType": ("TEXT", 1, 0), "IdempotencyKey": ("TEXT", 1, 0), "Status": ("TEXT", 1, 0), "Priority": ("INTEGER", 1, 0), "EntitlementDecision": ("TEXT", 1, 0), "PrivacyPolicy": ("TEXT", 1, 0), "SubjectType": ("TEXT", 0, 0), "SubjectId": ("TEXT", 0, 0), "Provider": ("TEXT", 0, 0), "Model": ("TEXT", 0, 0), "AttemptCount": ("INTEGER", 1, 0), "MaxAttempts": ("INTEGER", 1, 0), "CreatedAtUtc": ("TEXT", 1, 0), "AvailableAtUtc": ("TEXT", 1, 0), "StartedAtUtc": ("TEXT", 0, 0), "CompletedAtUtc": ("TEXT", 0, 0), "DeadlineAtUtc": ("TEXT", 0, 0), "CancellationRequestedAtUtc": ("TEXT", 0, 0), "LeaseToken": ("TEXT", 0, 0), "LeaseExpiresAtUtc": ("TEXT", 0, 0), "LastHeartbeatAtUtc": ("TEXT", 0, 0), "ProgressStage": ("TEXT", 0, 0), "ProgressPercent": ("INTEGER", 0, 0), "FailureCategory": ("TEXT", 0, 0), "FailureMessage": ("TEXT", 0, 0), "ResultReference": ("TEXT", 0, 0), } def insert_only_authorizer(): inserts = 0 def authorize(action, target, column, database, trigger): nonlocal inserts if action == sqlite3.SQLITE_INSERT: inserts += 1 return sqlite3.SQLITE_OK if target == "UserOperations" and inserts == 1 and trigger is None else sqlite3.SQLITE_DENY if action in {sqlite3.SQLITE_READ, sqlite3.SQLITE_SELECT, sqlite3.SQLITE_TRANSACTION, sqlite3.SQLITE_FUNCTION}: return sqlite3.SQLITE_OK if action == sqlite3.SQLITE_PRAGMA and target == "table_info": return sqlite3.SQLITE_OK return sqlite3.SQLITE_DENY return authorize def seed_operation(payload: object) -> str: if not isinstance(payload, dict) or set(payload) != {"dataRoot", "ownerId", "ownerEmail"}: raise SeedDenied("Exactly dataRoot, ownerId and ownerEmail are required") if any(not isinstance(value, str) or not value.strip() for value in payload.values()): raise SeedDenied("All inputs must be nonempty strings") requested = Path(payload["dataRoot"]) if not requested.is_absolute(): raise SeedDenied("dataRoot must be absolute") try: root, temporary = requested.resolve(strict=True), Path(tempfile.gettempdir()).resolve(strict=True) except OSError as error: raise SeedDenied("The isolated dataRoot must already exist") from error if (root != requested or root.parent != temporary or not root.is_dir() or requested.is_symlink() or requested.is_junction() or not re.fullmatch(r"jobtracker-e2e-[A-Za-z0-9][A-Za-z0-9_-]*", root.name)): raise SeedDenied("Only an unlinked jobtracker-e2e-* direct child of system temp is allowed") database = root / "jobtracker.db" if (not database.is_file() or database.is_symlink() or database.is_junction() or database.resolve(strict=True) != database or database.stat().st_nlink != 1): raise SeedDenied("The unlinked existing direct jobtracker.db file is required") # EF's SQLite Guid converter writes uppercase canonical text. operation_id = str(uuid.uuid4()).upper() timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f") with closing(sqlite3.connect(database.as_uri() + "?mode=rw", uri=True, timeout=5)) as connection, connection: connection.execute("PRAGMA foreign_keys=ON") connection.set_authorizer(insert_only_authorizer()) connection.execute("BEGIN IMMEDIATE") actual = {row[1]: (row[2].upper(), row[3], row[5]) for row in connection.execute('PRAGMA table_info("UserOperations")')} if actual != SCHEMA: raise SeedDenied("UserOperations must match its existing 28-column SQLite migration") users = {row[1]: row[2].upper() for row in connection.execute('PRAGMA table_info("AspNetUsers")')} if users.get("Id") != "TEXT" or users.get("Email") != "TEXT": raise SeedDenied("Existing AspNetUsers Id/Email schema is required") owner = connection.execute('SELECT "Id", "Email" FROM "AspNetUsers" WHERE "Id" = ?', (payload["ownerId"],)).fetchone() if owner != (payload["ownerId"], payload["ownerEmail"]): raise SeedDenied("Owner ID/email must exactly match the existing real fixture account") if connection.execute("SELECT name FROM sqlite_master WHERE type='trigger' AND tbl_name='UserOperations' LIMIT 1").fetchone(): raise SeedDenied("UserOperations triggers are outside this insert-only fixture contract") connection.execute( 'INSERT INTO "UserOperations" ("Id", "OwnerUserId", "TaskType", "IdempotencyKey", "Status", "Priority", "EntitlementDecision", "PrivacyPolicy", "AttemptCount", "MaxAttempts", "CreatedAtUtc", "AvailableAtUtc") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', (operation_id, payload["ownerId"], "native-offline-fixture", uuid.uuid4().hex, "queued", 0, "native_fixture", "local_only", 0, 3, timestamp, timestamp), ) return operation_id.lower() @contextmanager def temporary_fixture(): with tempfile.TemporaryDirectory(prefix="jobtracker-e2e-operation-selftest-") as directory: root = Path(directory).resolve() try: yield root finally: assert root.parent == Path(tempfile.gettempdir()).resolve() assert root.name.startswith("jobtracker-e2e-operation-selftest-") and not root.is_symlink() and not root.is_junction() def self_test() -> dict: passed = [] owner, email = "owner' OR 1=1 --", "quote'@example.test" def fixture(root, variant=None): database = root / "jobtracker.db" columns = [] for name, (kind, required, primary) in SCHEMA.items(): if variant == "missing_column" and name == "ResultReference": continue if variant == "wrong_type" and name == "Status": kind = "INTEGER" if variant == "wrong_nullability" and name == "Status": required = 0 columns.append(f'"{name}" {kind}' + (" NOT NULL" if required else "") + (" PRIMARY KEY" if primary else "")) if variant == "extra_column": columns.append('"Unexpected" TEXT') with closing(sqlite3.connect(database)) as connection, connection: connection.execute('CREATE TABLE "UserOperations" (' + ",".join(columns) + ")") connection.execute('CREATE TABLE "AspNetUsers" ("Id" TEXT PRIMARY KEY, "Email" TEXT, "Canonical" TEXT)') connection.execute('INSERT INTO "AspNetUsers" VALUES (?, ?, ?)', (owner, email, "Canonical source — æøå")) connection.execute('CREATE TABLE "UserNotifications" ("Id" TEXT PRIMARY KEY, "Title" TEXT)') connection.execute('INSERT INTO "UserNotifications" VALUES (?, ?)', ("existing-notice", "Preserve notice")) return database def snapshot(database): with closing(sqlite3.connect(database)) as connection: return {table: connection.execute(f'SELECT * FROM "{table}" ORDER BY rowid').fetchall() for table in ("AspNetUsers", "UserOperations", "UserNotifications", "sqlite_master")} def payload(root, **changes): return {"dataRoot": str(root), "ownerId": owner, "ownerEmail": email, **changes} def deny(name, value): try: seed_operation(value) except SeedDenied: passed.append(name) else: raise AssertionError(name + " was accepted") with temporary_fixture() as root: deny("missing_database_does_not_create", payload(root)) assert not (root / "jobtracker.db").exists() database = fixture(root) before = snapshot(database) for name, value in (("unknown_key", payload(root, taskType="live")), ("missing_key", {"dataRoot": str(root)}), ("relative_root", payload(root, dataRoot=".")), ("wrong_owner", payload(root, ownerId="other")), ("wrong_email", payload(root, ownerEmail="other@example.test")), ("exact_email_case", payload(root, ownerEmail=email.upper())), ("empty_owner", payload(root, ownerId="")), ("non_object", [])): deny(name, value) assert snapshot(database) == before passed.append("denied_inputs_preserve_rows_and_schema") identifier = seed_operation(payload(root)) after = snapshot(database) assert len(after["UserOperations"]) == 1 assert all(after[key] == value for key, value in before.items() if key != "UserOperations") with closing(sqlite3.connect(database)) as connection: connection.row_factory = sqlite3.Row row = dict(connection.execute('SELECT * FROM "UserOperations"').fetchone()) assert row["Id"] == identifier.upper() and str(uuid.UUID(row["Id"])).upper() == row["Id"] assert row["OwnerUserId"] == owner and row["TaskType"] == "native-offline-fixture" assert row["Status"] == "queued" and row["EntitlementDecision"] == "native_fixture" and row["PrivacyPolicy"] == "local_only" assert row["AttemptCount"] == 0 and row["MaxAttempts"] == 3 and row["CreatedAtUtc"] == row["AvailableAtUtc"] assert all(row[key] is None for key, (_, required, _) in SCHEMA.items() if not required) passed.extend(("parameterized_exact_insert", "ef_uppercase_guid", "no_subject_provider_model_lease_result", "existing_rows_and_schema_unchanged")) for statement in ('UPDATE "AspNetUsers" SET "Canonical" = \'lost\'', 'DELETE FROM "UserOperations"', 'INSERT INTO "UserNotifications" VALUES (\'new\', \'wrong\')', 'CREATE TABLE "Forbidden" ("Id" INTEGER)', "ATTACH DATABASE ':memory:' AS forbidden", "PRAGMA writable_schema=ON"): with closing(sqlite3.connect(database)) as connection: connection.set_authorizer(insert_only_authorizer()) try: connection.execute(statement) except sqlite3.DatabaseError: passed.append("authorizer_denies_" + statement.split()[0].lower()) else: raise AssertionError("Forbidden SQL was accepted: " + statement) assert snapshot(database) == after passed.append("denied_sql_preserves_all_rows_and_schema") (root / "fixture-hardlink.db").hardlink_to(database) deny("hardlinked_database", payload(root)) assert snapshot(database) == after for variant in ("missing_column", "extra_column", "wrong_type", "wrong_nullability"): with temporary_fixture() as root: database = fixture(root, variant) before = snapshot(database) deny(variant, payload(root)) assert snapshot(database) == before with temporary_fixture() as root: database = fixture(root) with closing(sqlite3.connect(database)) as connection, connection: connection.execute('CREATE TRIGGER "unexpected_write" AFTER INSERT ON "UserOperations" BEGIN UPDATE "AspNetUsers" SET "Canonical" = \'lost\'; END') before = snapshot(database) deny("trigger_denied_before_insert", payload(root)) assert snapshot(database) == before with temporary_fixture() as root: nested = root / "jobtracker-e2e-nested" nested.mkdir() database = fixture(nested) before = snapshot(database) deny("nested_root", payload(nested)) deny("wrong_prefix", payload(root, dataRoot=str(root.parent))) assert snapshot(database) == before return {"selfTest": "PASS", "checks": len(passed), "passed": passed, "databaseScope": "own direct temporary SQLite fixtures only"} def main(): try: if sys.argv[1:] == ["--self-test"]: print(json.dumps(self_test())) elif sys.argv[1:]: raise SeedDenied("Only stdin JSON or --self-test is supported") else: print(json.dumps({"operationId": seed_operation(json.load(sys.stdin))})) return 0 except (SeedDenied, json.JSONDecodeError, sqlite3.Error, OSError) as error: print(json.dumps({"error": str(error)}), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())