cesnimda
107 lines
5.5 KiB
Python
107 lines
5.5 KiB
Python
"""Read only the explicitly named synthetic owner's committed SQLite rows."""
|
|
import json
|
|
from contextlib import closing
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
CHILDREN = ("CareerExperiences", "CareerEducations", "CareerSkills", "CareerProjects",
|
|
"CareerCertifications", "CareerLanguages")
|
|
|
|
|
|
def connect_readonly(filename):
|
|
path = Path(filename).resolve(strict=True)
|
|
connection = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)
|
|
connection.row_factory = sqlite3.Row
|
|
return connection
|
|
|
|
|
|
def snapshot(connection, owner, job_id=None):
|
|
def rows(sql, parameters):
|
|
return [dict(row) for row in connection.execute(sql, parameters)]
|
|
|
|
identities = rows('SELECT "Id", "ProfileCvText", "ProfileCvStructureJson", '
|
|
'"CurrentCvProfileVersion" FROM "AspNetUsers" WHERE "Id" = ?', (owner,))
|
|
if len(identities) != 1:
|
|
raise ValueError("Expected exactly one explicitly owned synthetic Identity row")
|
|
careers = rows('SELECT * FROM "CareerProfiles" WHERE "OwnerUserId" = ?', (owner,))
|
|
if len(careers) > 1:
|
|
raise ValueError("More than one canonical Career Profile for owner")
|
|
result = {
|
|
"identity": identities[0],
|
|
"career": careers[0] if careers else None,
|
|
"careerHistory": rows('SELECT * FROM "CareerProfileVersions" WHERE "OwnerUserId" = ? ORDER BY "Version", "Id"', (owner,)),
|
|
"children": {table: rows(f'SELECT * FROM "{table}" WHERE "OwnerUserId" = ? ORDER BY "SortOrder", "Id"', (owner,)) for table in CHILDREN},
|
|
"job": None, "opportunity": None, "events": [],
|
|
}
|
|
if job_id is not None:
|
|
jobs = rows('SELECT * FROM "JobApplications" WHERE "OwnerUserId" = ? AND "Id" = ?', (owner, job_id))
|
|
if len(jobs) != 1:
|
|
raise ValueError("Expected exactly one explicitly owned synthetic job")
|
|
result["job"] = jobs[0]
|
|
opportunities = rows('SELECT * FROM "Jobs" WHERE "OwnerUserId" = ? AND "Id" = ?', (owner, jobs[0]["JobId"]))
|
|
if len(opportunities) != 1:
|
|
raise ValueError("Expected the job's explicitly owned opportunity")
|
|
result["opportunity"] = opportunities[0]
|
|
result["events"] = rows('SELECT e.* FROM "JobEvents" e JOIN "JobApplications" j ON e."JobApplicationId" = j."Id" '
|
|
'WHERE j."OwnerUserId" = ? AND j."Id" = ? ORDER BY e."Id"', (owner, job_id))
|
|
return result
|
|
|
|
|
|
def self_test():
|
|
with tempfile.TemporaryDirectory(prefix="rd21-readonly-proof-") as directory:
|
|
filename = Path(directory) / "jobtracker.db"
|
|
with closing(sqlite3.connect(filename)) as writer, writer:
|
|
writer.executescript('''
|
|
CREATE TABLE AspNetUsers (Id TEXT, ProfileCvText TEXT, ProfileCvStructureJson TEXT, CurrentCvProfileVersion INTEGER);
|
|
INSERT INTO AspNetUsers VALUES ('owner', 'raw', '{}', 1), ('other', 'private', '{}', 1);
|
|
CREATE TABLE CareerProfiles (Id INTEGER, OwnerUserId TEXT, Version INTEGER, ProfileJson TEXT);
|
|
INSERT INTO CareerProfiles VALUES (1, 'owner', 1, '{}'), (2, 'other', 1, '{}');
|
|
CREATE TABLE CareerProfileVersions (Id INTEGER, OwnerUserId TEXT, Version INTEGER);
|
|
INSERT INTO CareerProfileVersions VALUES (1, 'owner', 1), (2, 'other', 1);
|
|
CREATE TABLE JobApplications (Id INTEGER, OwnerUserId TEXT, JobId INTEGER, Version INTEGER);
|
|
INSERT INTO JobApplications VALUES (1, 'owner', 10, 2), (2, 'other', 20, 3);
|
|
CREATE TABLE Jobs (Id INTEGER, OwnerUserId TEXT);
|
|
INSERT INTO Jobs VALUES (10, 'owner'), (20, 'other');
|
|
CREATE TABLE JobEvents (Id INTEGER, JobApplicationId INTEGER);
|
|
INSERT INTO JobEvents VALUES (1, 1), (2, 2);
|
|
''')
|
|
for table in CHILDREN:
|
|
writer.execute(f'CREATE TABLE "{table}" (Id INTEGER, OwnerUserId TEXT, SortOrder INTEGER)')
|
|
writer.executemany(f'INSERT INTO "{table}" VALUES (?, ?, ?)', [(1, "owner", 0), (2, "other", 0)])
|
|
with closing(connect_readonly(filename)) as reader:
|
|
before = snapshot(reader, "owner", 1)
|
|
assert before["career"]["Version"] == 1
|
|
assert before["job"]["Version"] == 2
|
|
assert before["opportunity"]["Id"] == 10
|
|
assert before["events"] == [{"Id": 1, "JobApplicationId": 1}]
|
|
assert all(len(rows) == 1 for rows in before["children"].values())
|
|
assert "private" not in json.dumps(before)
|
|
try:
|
|
reader.execute('UPDATE CareerProfiles SET Version = 99')
|
|
except sqlite3.OperationalError as error:
|
|
assert "readonly" in str(error).lower()
|
|
else:
|
|
raise AssertionError("Read-only URI allowed mutation")
|
|
assert snapshot(reader, "owner", 1) == before
|
|
try:
|
|
snapshot(reader, "owner", 2)
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("Cross-owner job was returned")
|
|
print("readonly-db-proof self-test passed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if sys.argv[1:] == ["--self-test"]:
|
|
self_test()
|
|
else:
|
|
if len(sys.argv) not in (3, 4):
|
|
raise SystemExit("Usage: readonly-db-proof.py DATABASE OWNER_ID [JOB_ID] | --self-test")
|
|
with closing(connect_readonly(sys.argv[1])) as connection:
|
|
connection.execute("BEGIN")
|
|
print(json.dumps(snapshot(connection, sys.argv[2], int(sys.argv[3]) if len(sys.argv) == 4 else None)))
|