Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 97 Pull Requests 7 Actions Packages Projects Releases Wiki Activity

feat(arch01): atomic platform install + authority kernel (Closes #822)

Browse Source 
ARCH-01 Foundation Slice A — first implementation leaf of the #820/#821
program. Adds a new, disabled-by-default SQLite kernel:

- Connection-bound trusted-service actor context (cp_actor_principal/kind,
  cp_operation_mode, cp_service_session, cp_context_epoch) that SQL may read
  but never set; every mutating trigger runs the actor protocol and fails
  closed on a missing, stale, or epoch-shifted context.
- Immutable authority-dominance lattice with the exact seeded tuple set,
  validated individually by the install-state trigger.
- Principal-equivalence root: a class exists before its first principal and
  current_class_id is NOT NULL.
- Single atomic BEGIN IMMEDIATE install seeding the installer principal, the
  distinguished operator-key issuer, dominance, the initial NULL-grantor
  bootstrap grant, the active invariant, and the immutable installed marker
  last; second install returns ALREADY_INSTALLED with no mutation.
- Immutability triggers on marker/seed/dominance/issuer/installer
  registration/initial-grant identity; append-only audit_records; last-active
  grant floored by CHECK so concurrent revokes cannot drop it to zero.

Files: arch01_platform.py (new), tests/test_arch01_platform.py (new).

Tests: 25 tests + 21 subtests cover ACs 1-14 including negative, rollback,
raw-write-bypass, and concurrency. Focused suite green; full suite 4422
passed / 11 pre-existing baseline failures at 53c2c92, zero new.

Subsystem disabled by default until readiness checks pass. SQLite-first;
PostgreSQL parity is #827.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
jcwalker3
2026-07-22 22:02:01 -05:00
co-authored by Claude Opus 4.8
parent 53c2c92782
commit 6449594fd7
2 changed files with 1464 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_arch01_platform.py
+572
View File
@@ -0,0 +1,572 @@
"""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()