cesnimda
238 lines
12 KiB
Python
238 lines
12 KiB
Python
"""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())
|