573 lines
26 KiB
Python
573 lines
26 KiB
Python
"""Executable acceptance tests for ARCH-01 Slice A (#822).
|
|
|
|
Each acceptance criterion (#822 §12) and named test (#822 §13) is exercised
|
|
against a real SQLite database. The migration runs on a fresh DB in ``setUp``;
|
|
the test-run output is the durable evidence the issue requires (§14).
|
|
|
|
Enforcement being proven:
|
|
|
|
* ``[TRUSTED-SERVICE]`` — the ``cp_*`` actor functions exist only on the
|
|
trusted kernel connection; a raw connection cannot satisfy the triggers.
|
|
* ``[SCHEMA]`` — fail-closed aborts, exact dominance set, NOT-NULL class,
|
|
immutability, and the last-active-grant floor are enforced by
|
|
CHECK/FK/trigger, verified here including raw-write bypass and concurrency.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
import arch01_platform as ap
|
|
from arch01_platform import (
|
|
ALREADY_INSTALLED,
|
|
AUTHORIZATION_DENIED,
|
|
CONCURRENT_INSTALLATION_LOST,
|
|
DISTINGUISHED_ISSUER_ID,
|
|
DOMINANCE_SET_MISMATCH,
|
|
DOMINANCE_TUPLES,
|
|
INSTALLED,
|
|
INVALID_ACTOR_CONTEXT,
|
|
INVALID_BOOTSTRAP_STATE,
|
|
PlatformKernel,
|
|
)
|
|
|
|
INSTALLER = "platform.installer"
|
|
|
|
_BOOTSTRAP_TABLES = (
|
|
"principal_equivalence_classes",
|
|
"principals",
|
|
"authoritative_issuers",
|
|
"authority_dominance",
|
|
"platform_bootstrap_seed",
|
|
"platform_bootstrap_grants",
|
|
"platform_active_invariant",
|
|
"install_state",
|
|
)
|
|
|
|
|
|
def _count(kernel: PlatformKernel, table: str) -> int:
|
|
return kernel._conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
|
|
|
|
|
|
def _count_where(kernel: PlatformKernel, table: str, where: str) -> int:
|
|
return kernel._conn.execute(f"SELECT COUNT(*) FROM {table} WHERE {where}").fetchone()[0]
|
|
|
|
|
|
def _all_bootstrap_empty(kernel: PlatformKernel) -> bool:
|
|
return all(_count(kernel, t) == 0 for t in _BOOTSTRAP_TABLES)
|
|
|
|
|
|
class Arch01MemoryTest(unittest.TestCase):
|
|
"""Single-connection behavior on an in-memory database."""
|
|
|
|
def setUp(self) -> None:
|
|
self.kernel = PlatformKernel(":memory:")
|
|
|
|
def tearDown(self) -> None:
|
|
self.kernel.close()
|
|
|
|
# -- AC1 -------------------------------------------------------------- #
|
|
def test_install_clean(self) -> None: # t_install_clean(+)
|
|
res = self.kernel.install_platform(INSTALLER)
|
|
self.assertEqual(res.code, INSTALLED)
|
|
self.assertTrue(self.kernel.is_installed())
|
|
self.assertEqual(_count(self.kernel, "install_state"), 1)
|
|
self.assertEqual(self.kernel.active_grant_count(), 1)
|
|
self.assertIn(ap.EVT_PLATFORM_INSTALLED, self.kernel.audit_events())
|
|
rows = set(
|
|
self.kernel._conn.execute(
|
|
"SELECT dominant, subordinate FROM authority_dominance"
|
|
).fetchall()
|
|
)
|
|
self.assertEqual(rows, set(DOMINANCE_TUPLES))
|
|
issuer_ref = self.kernel._conn.execute(
|
|
"SELECT i.issuer_ref FROM principals p JOIN authoritative_issuers i "
|
|
"ON p.issuer_id = i.issuer_id WHERE p.principal_id = ?",
|
|
(INSTALLER,),
|
|
).fetchone()
|
|
self.assertEqual(issuer_ref[0], DISTINGUISHED_ISSUER_ID)
|
|
|
|
# -- AC2 -------------------------------------------------------------- #
|
|
def test_install_twice(self) -> None: # t_install_twice(-)
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
res2 = self.kernel.install_platform(INSTALLER)
|
|
self.assertEqual(res2.code, ALREADY_INSTALLED)
|
|
self.assertEqual(_count(self.kernel, "principals"), 1)
|
|
self.assertEqual(_count(self.kernel, "platform_bootstrap_grants"), 1)
|
|
self.assertEqual(_count(self.kernel, "install_state"), 1)
|
|
|
|
# -- AC3 / AC5 -------------------------------------------------------- #
|
|
def test_install_stage_rollback(self) -> None: # t_install_stage_rollback
|
|
for stop in range(1, 9):
|
|
with self.subTest(stages=stop):
|
|
k = PlatformKernel(":memory:")
|
|
try:
|
|
self._partial_bootstrap_then_rollback(k, stop)
|
|
self.assertTrue(
|
|
_all_bootstrap_empty(k),
|
|
f"partial rows survived rollback at stage {stop}",
|
|
)
|
|
self.assertFalse(k.is_installed())
|
|
finally:
|
|
k.close()
|
|
|
|
def test_no_partial_after_rollback(self) -> None: # t_no_partial_after_rollback
|
|
k = PlatformKernel(":memory:")
|
|
try:
|
|
code = self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1])
|
|
self.assertEqual(code, DOMINANCE_SET_MISMATCH)
|
|
self.assertTrue(_all_bootstrap_empty(k))
|
|
self.assertFalse(k.is_installed())
|
|
finally:
|
|
k.close()
|
|
|
|
# -- AC4 -------------------------------------------------------------- #
|
|
def test_dominance_missing(self) -> None: # t_dominance_missing(-)
|
|
k = PlatformKernel(":memory:")
|
|
try:
|
|
self.assertEqual(
|
|
self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1]),
|
|
DOMINANCE_SET_MISMATCH,
|
|
)
|
|
self.assertFalse(k.is_installed())
|
|
finally:
|
|
k.close()
|
|
|
|
def test_dominance_extra(self) -> None: # t_dominance_extra(-)
|
|
k = PlatformKernel(":memory:")
|
|
try:
|
|
extra = DOMINANCE_TUPLES + (("platform.bootstrap", "rogue.extra"),)
|
|
self.assertEqual(
|
|
self._seed_bootstrap_and_mark(k, dominance=extra),
|
|
DOMINANCE_SET_MISMATCH,
|
|
)
|
|
self.assertFalse(k.is_installed())
|
|
finally:
|
|
k.close()
|
|
|
|
def test_dominance_malformed(self) -> None: # t_dominance_malformed(-)
|
|
k = PlatformKernel(":memory:")
|
|
try:
|
|
malformed = DOMINANCE_TUPLES[:-1] + (("supervisor.root", "WRONG.subordinate"),)
|
|
self.assertEqual(
|
|
self._seed_bootstrap_and_mark(k, dominance=malformed),
|
|
DOMINANCE_SET_MISMATCH,
|
|
)
|
|
self.assertFalse(k.is_installed())
|
|
finally:
|
|
k.close()
|
|
|
|
# -- AC6 -------------------------------------------------------------- #
|
|
def test_principal_no_class(self) -> None: # t_principal_no_class(-)
|
|
with self.kernel.actor_context("op", "operator", "install"):
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.kernel._conn.execute(
|
|
"INSERT INTO principals"
|
|
"(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
|
|
"VALUES ('x', 'operator', NULL, NULL, NULL, '2026-01-01T00:00:00Z')"
|
|
)
|
|
|
|
# -- AC7 -------------------------------------------------------------- #
|
|
def test_noninstaller_null_issuer(self) -> None: # t_nonobstaller_null_issuer(-)
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
with self.kernel.actor_context("op", "operator", "normal"):
|
|
cur = self.kernel._conn.execute(
|
|
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
|
|
)
|
|
class_id = cur.lastrowid
|
|
with self.assertRaises(sqlite3.IntegrityError) as ctx:
|
|
self.kernel._conn.execute(
|
|
"INSERT INTO principals"
|
|
"(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
|
|
"VALUES ('rogue', 'operator', ?, NULL, NULL, '2026-01-01T00:00:00Z')",
|
|
(class_id,),
|
|
)
|
|
self.assertIn("INVALID_BOOTSTRAP_STATE", str(ctx.exception))
|
|
|
|
def test_installer_null_issuer_only_during_install(self) -> None:
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
with self.kernel.actor_context("i2", "installer", "install"):
|
|
cur = self.kernel._conn.execute(
|
|
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
|
|
)
|
|
class_id = cur.lastrowid
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.kernel._conn.execute(
|
|
"INSERT INTO principals"
|
|
"(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
|
|
"VALUES ('i2', 'installer', ?, NULL, NULL, '2026-01-01T00:00:00Z')",
|
|
(class_id,),
|
|
)
|
|
|
|
# -- AC8 -------------------------------------------------------------- #
|
|
def test_context_missing(self) -> None: # t_context_missing(-)
|
|
self.assertIsNone(self.kernel._ctx)
|
|
with self.assertRaises(sqlite3.IntegrityError) as ctx:
|
|
self.kernel._conn.execute(
|
|
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
|
|
)
|
|
self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
|
|
|
|
def test_context_stale(self) -> None: # t_context_stale(-)
|
|
with self.kernel.actor_context("op", "operator", "normal"):
|
|
self.kernel._ctx.expired = True
|
|
with self.assertRaises(sqlite3.IntegrityError) as ctx:
|
|
self.kernel._conn.execute(
|
|
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
|
|
)
|
|
self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
|
|
|
|
def test_context_epoch_shift(self) -> None: # t_context_epoch_shift(-)
|
|
with self.kernel.actor_context("op", "operator", "normal"):
|
|
self.kernel._ctx.live_epoch = self.kernel._ctx.bound_epoch + 99
|
|
with self.assertRaises(sqlite3.IntegrityError) as ctx:
|
|
self.kernel._conn.execute(
|
|
"INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
|
|
)
|
|
self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))
|
|
|
|
def test_bad_actor_kind_or_mode_rejected(self) -> None:
|
|
for kind, mode in (("intruder", "normal"), ("operator", "sabotage")):
|
|
with self.subTest(kind=kind, mode=mode):
|
|
with self.kernel.actor_context("op", kind, mode):
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
self.kernel._conn.execute(
|
|
"INSERT INTO principal_equivalence_classes(created_at) "
|
|
"VALUES ('2026-01-01T00:00:00Z')"
|
|
)
|
|
|
|
# -- AC9 -------------------------------------------------------------- #
|
|
def test_bootstrap_immutable_update(self) -> None: # t_bootstrap_immutable_{update}
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
cases = [
|
|
("UPDATE install_state SET installed_at = 'x' WHERE id = 1", "IMMUTABLE_INSTALL_STATE"),
|
|
("UPDATE platform_bootstrap_seed SET created_at = 'x' WHERE seed_id = 1", "IMMUTABLE_SEED"),
|
|
("UPDATE authority_dominance SET subordinate = 'x' WHERE dominant = 'supervisor.root'", "IMMUTABLE_DOMINANCE"),
|
|
(f"UPDATE authoritative_issuers SET issuer_ref = 'x' WHERE issuer_ref = '{DISTINGUISHED_ISSUER_ID}'", "IMMUTABLE_ISSUER"),
|
|
(f"UPDATE principals SET actor_kind = 'operator' WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"),
|
|
]
|
|
for sql, tag in cases:
|
|
with self.subTest(sql=sql):
|
|
with self.kernel.actor_context("op", "operator", "normal"):
|
|
with self.assertRaises(sqlite3.IntegrityError) as ctx:
|
|
self.kernel._conn.execute(sql)
|
|
self.assertIn(tag, str(ctx.exception))
|
|
|
|
def test_bootstrap_immutable_delete(self) -> None: # t_bootstrap_immutable_{delete}
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
cases = [
|
|
("DELETE FROM install_state WHERE id = 1", "IMMUTABLE_INSTALL_STATE"),
|
|
("DELETE FROM platform_bootstrap_seed WHERE seed_id = 1", "IMMUTABLE_SEED"),
|
|
("DELETE FROM authority_dominance", "IMMUTABLE_DOMINANCE"),
|
|
("DELETE FROM authoritative_issuers", "IMMUTABLE_ISSUER"),
|
|
(f"DELETE FROM principals WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"),
|
|
("DELETE FROM platform_bootstrap_grants", "IMMUTABLE_GRANT"),
|
|
]
|
|
for sql, tag in cases:
|
|
with self.subTest(sql=sql):
|
|
with self.kernel.actor_context("op", "operator", "normal"):
|
|
with self.assertRaises(sqlite3.IntegrityError) as ctx:
|
|
self.kernel._conn.execute(sql)
|
|
self.assertIn(tag, str(ctx.exception))
|
|
|
|
def test_grant_reactivation_rejected(self) -> None:
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
self.kernel.register_principal(
|
|
"op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
|
|
)
|
|
self.assertEqual(
|
|
self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED
|
|
)
|
|
gid = self.kernel._conn.execute(
|
|
"SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'"
|
|
).fetchone()[0]
|
|
self.assertEqual(
|
|
self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code,
|
|
INSTALLED,
|
|
)
|
|
with self.kernel.actor_context("op", "operator", "normal"):
|
|
with self.assertRaises(sqlite3.IntegrityError) as ctx:
|
|
self.kernel._conn.execute(
|
|
"UPDATE platform_bootstrap_grants SET active = 1 WHERE grant_id = ?",
|
|
(gid,),
|
|
)
|
|
self.assertIn("IMMUTABLE_GRANT", str(ctx.exception))
|
|
|
|
# -- AC12 ------------------------------------------------------------- #
|
|
def test_raw_write_bypass(self) -> None: # t_raw_write_bypass(raw-bypass)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = os.path.join(tmp, "p.sqlite3")
|
|
k = PlatformKernel(path)
|
|
self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED)
|
|
k.close()
|
|
raw = sqlite3.connect(path)
|
|
raw.execute("PRAGMA foreign_keys = ON")
|
|
try:
|
|
with self.assertRaises(sqlite3.Error):
|
|
raw.execute(
|
|
"INSERT INTO audit_records(event, created_at) "
|
|
"VALUES ('forged', '2026-01-01T00:00:00Z')"
|
|
)
|
|
raw.commit()
|
|
with self.assertRaises(sqlite3.Error):
|
|
raw.execute("UPDATE install_state SET installed_at = 'x' WHERE id = 1")
|
|
raw.commit()
|
|
with self.assertRaises(sqlite3.Error):
|
|
raw.execute("DELETE FROM platform_bootstrap_grants")
|
|
raw.commit()
|
|
finally:
|
|
raw.close()
|
|
|
|
# -- AC13 ------------------------------------------------------------- #
|
|
def test_audit_created(self) -> None: # t_audit_created(+)
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
self.kernel.register_principal(
|
|
"op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
|
|
)
|
|
self.assertEqual(
|
|
self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED
|
|
)
|
|
gid = self.kernel._conn.execute(
|
|
"SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'"
|
|
).fetchone()[0]
|
|
self.assertEqual(
|
|
self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code,
|
|
INSTALLED,
|
|
)
|
|
events = self.kernel.audit_events()
|
|
for evt in (
|
|
ap.EVT_PLATFORM_INSTALLED,
|
|
ap.EVT_GRANT_CREATED,
|
|
ap.EVT_GRANT_REVOKED,
|
|
ap.EVT_PRINCIPAL_REGISTERED,
|
|
):
|
|
self.assertIn(evt, events)
|
|
|
|
# -- AC14 ------------------------------------------------------------- #
|
|
def test_audit_immutable(self) -> None: # t_audit_immutable(raw-bypass)
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
with self.kernel.actor_context("op", "operator", "normal"):
|
|
with self.assertRaises(sqlite3.IntegrityError) as up:
|
|
self.kernel._conn.execute("UPDATE audit_records SET event = 'x' WHERE audit_id = 1")
|
|
self.assertIn("IMMUTABLE_AUDIT", str(up.exception))
|
|
with self.assertRaises(sqlite3.IntegrityError) as dl:
|
|
self.kernel._conn.execute("DELETE FROM audit_records WHERE audit_id = 1")
|
|
self.assertIn("IMMUTABLE_AUDIT", str(dl.exception))
|
|
|
|
# -- meta ------------------------------------------------------------- #
|
|
def test_schema_meta(self) -> None:
|
|
rows = dict(self.kernel._conn.execute("SELECT key, value FROM arch01_meta").fetchall())
|
|
self.assertEqual(rows["schema_version"], str(ap.SCHEMA_VERSION))
|
|
self.assertIn("disabled by default", rows["architecture"])
|
|
|
|
def test_register_principal_creates_class_first(self) -> None:
|
|
self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
|
|
res = self.kernel.register_principal(
|
|
"svc1", "service", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
|
|
)
|
|
self.assertEqual(res.code, INSTALLED)
|
|
row = self.kernel._conn.execute(
|
|
"SELECT current_class_id FROM principals WHERE principal_id = 'svc1'"
|
|
).fetchone()
|
|
self.assertIsNotNone(row[0])
|
|
|
|
# -- helpers ---------------------------------------------------------- #
|
|
def _partial_bootstrap_then_rollback(self, k: PlatformKernel, stop: int) -> None:
|
|
"""Execute the first ``stop`` bootstrap statements, then ROLLBACK."""
|
|
now = "2026-01-01T00:00:00Z"
|
|
k._conn.execute("BEGIN IMMEDIATE")
|
|
class_id = None
|
|
issuer_id = None
|
|
try:
|
|
with k.actor_context(INSTALLER, "installer", "install"):
|
|
c = k._conn
|
|
if stop >= 1:
|
|
class_id = c.execute(
|
|
"INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,)
|
|
).lastrowid
|
|
if stop >= 2:
|
|
c.execute(
|
|
"INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
|
|
"VALUES (?, 'installer', ?, NULL, ?, ?)",
|
|
(INSTALLER, class_id, INSTALLER, now),
|
|
)
|
|
if stop >= 3:
|
|
issuer_id = c.execute(
|
|
"INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)",
|
|
(DISTINGUISHED_ISSUER_ID, now),
|
|
).lastrowid
|
|
if stop >= 4:
|
|
c.execute(
|
|
"UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
|
|
(issuer_id, INSTALLER),
|
|
)
|
|
if stop >= 5:
|
|
c.executemany(
|
|
"INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
|
|
DOMINANCE_TUPLES,
|
|
)
|
|
if stop >= 6:
|
|
c.execute(
|
|
"INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)",
|
|
(INSTALLER, now),
|
|
)
|
|
if stop >= 7:
|
|
c.execute(
|
|
"INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)",
|
|
(INSTALLER, now),
|
|
)
|
|
if stop >= 8:
|
|
c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)")
|
|
finally:
|
|
k._conn.execute("ROLLBACK")
|
|
|
|
def _seed_bootstrap_and_mark(self, k: PlatformKernel, dominance) -> str:
|
|
"""Seed a full bootstrap with a caller-supplied dominance set, then
|
|
attempt the marker insert. Returns the classified failure code (or
|
|
INSTALLED). Rolls back on failure so no partial rows remain."""
|
|
now = "2026-01-01T00:00:00Z"
|
|
k._conn.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
with k.actor_context(INSTALLER, "installer", "install"):
|
|
c = k._conn
|
|
class_id = c.execute(
|
|
"INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,)
|
|
).lastrowid
|
|
c.execute(
|
|
"INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
|
|
"VALUES (?, 'installer', ?, NULL, ?, ?)",
|
|
(INSTALLER, class_id, INSTALLER, now),
|
|
)
|
|
issuer_id = c.execute(
|
|
"INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)",
|
|
(DISTINGUISHED_ISSUER_ID, now),
|
|
).lastrowid
|
|
c.execute(
|
|
"UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
|
|
(issuer_id, INSTALLER),
|
|
)
|
|
c.executemany(
|
|
"INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
|
|
dominance,
|
|
)
|
|
c.execute(
|
|
"INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)",
|
|
(INSTALLER, now),
|
|
)
|
|
c.execute(
|
|
"INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)",
|
|
(INSTALLER, now),
|
|
)
|
|
c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)")
|
|
c.execute(
|
|
"INSERT INTO install_state(id, marker, installed_at) VALUES (1, 'installed', ?)",
|
|
(now,),
|
|
)
|
|
k._conn.execute("COMMIT")
|
|
return INSTALLED
|
|
except sqlite3.Error as exc:
|
|
k._safe_rollback()
|
|
return PlatformKernel._classify(exc)
|
|
|
|
|
|
class Arch01ConcurrencyTest(unittest.TestCase):
|
|
"""Concurrency invariants require file-backed DBs and independent connections."""
|
|
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.path = os.path.join(self._tmp.name, "p.sqlite3")
|
|
|
|
def tearDown(self) -> None:
|
|
self._tmp.cleanup()
|
|
|
|
# -- AC10 ------------------------------------------------------------- #
|
|
def test_concurrent_install(self) -> None: # t_concurrent_install(concurrency)
|
|
k1 = PlatformKernel(self.path, busy_timeout_ms=0)
|
|
k2 = PlatformKernel(self.path, busy_timeout_ms=0)
|
|
barrier = threading.Barrier(2)
|
|
results = {}
|
|
|
|
def _install(name, kernel):
|
|
barrier.wait()
|
|
results[name] = kernel.install_platform(INSTALLER).code
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=2) as ex:
|
|
f1 = ex.submit(_install, "a", k1)
|
|
f2 = ex.submit(_install, "b", k2)
|
|
f1.result()
|
|
f2.result()
|
|
codes = sorted(results.values())
|
|
self.assertEqual(codes.count(INSTALLED), 1, f"exactly one install expected: {results}")
|
|
other = [c for c in results.values() if c != INSTALLED][0]
|
|
self.assertIn(other, (ALREADY_INSTALLED, CONCURRENT_INSTALLATION_LOST))
|
|
self.assertTrue(k1.is_installed())
|
|
self.assertEqual(_count(k1, "install_state"), 1)
|
|
self.assertEqual(_count(k1, "principals"), 1)
|
|
finally:
|
|
k1.close()
|
|
k2.close()
|
|
|
|
# -- AC11 ------------------------------------------------------------- #
|
|
def test_concurrent_last_grant_revoke(self) -> None: # t_concurrent_last_grant_revoke
|
|
setup = PlatformKernel(self.path)
|
|
self.assertEqual(setup.install_platform(INSTALLER).code, INSTALLED)
|
|
setup.register_principal("op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER)
|
|
self.assertEqual(setup.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED)
|
|
self.assertEqual(setup.active_grant_count(), 2)
|
|
gids = [
|
|
r[0]
|
|
for r in setup._conn.execute(
|
|
"SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1 ORDER BY grant_id"
|
|
).fetchall()
|
|
]
|
|
setup.close()
|
|
self.assertEqual(len(gids), 2)
|
|
|
|
k1 = PlatformKernel(self.path, busy_timeout_ms=3000)
|
|
k2 = PlatformKernel(self.path, busy_timeout_ms=3000)
|
|
barrier = threading.Barrier(2)
|
|
results = {}
|
|
|
|
def _revoke(name, kernel, gid):
|
|
barrier.wait()
|
|
results[name] = kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=2) as ex:
|
|
f1 = ex.submit(_revoke, "a", k1, gids[0])
|
|
f2 = ex.submit(_revoke, "b", k2, gids[1])
|
|
f1.result()
|
|
f2.result()
|
|
codes = list(results.values())
|
|
self.assertEqual(codes.count(INSTALLED), 1, f"exactly one revoke should win: {results}")
|
|
self.assertEqual(codes.count(AUTHORIZATION_DENIED), 1, f"one revoke must be denied: {results}")
|
|
self.assertEqual(k1.active_grant_count(), 1)
|
|
self.assertEqual(_count_where(k1, "platform_bootstrap_grants", "active = 1"), 1)
|
|
finally:
|
|
k1.close()
|
|
k2.close()
|
|
|
|
def test_revoke_final_grant_denied(self) -> None:
|
|
k = PlatformKernel(self.path)
|
|
try:
|
|
self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED)
|
|
gid = k._conn.execute(
|
|
"SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1"
|
|
).fetchone()[0]
|
|
res = k.revoke_platform_bootstrap(gid, actor_principal=INSTALLER)
|
|
self.assertEqual(res.code, AUTHORIZATION_DENIED)
|
|
self.assertEqual(k.active_grant_count(), 1)
|
|
self.assertEqual(_count_where(k, "platform_bootstrap_grants", "active = 1"), 1)
|
|
finally:
|
|
k.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|