From d83ac6f56b24864eb41deae03789ec0f864a4d55 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Fri, 30 Oct 2026 18:02:05 +0800 Subject: [PATCH] [arbiter-authentication/2.10.1] Local issuer trust and unique agent bindings Manage exact public owner-issuer trust through existing authenticated local controls, with bounded registration/export, paginated summaries and inactive revisioned tombstones. Upgrade authority schema atomically, preserve historical enrollment results and denial state, and maintain one unique agent binding in the existing fresh-v1 activation transaction and exact replay. This is corrective cut2.10.1; portable identity reuse and ordinary authenticated API enforcement remain subsequent work. Validation: isolated Python suite1832passed43platform/prerequisite skips; formatting/type checks; Go tests and vet; current general, actualDBB and fresh independent general review; focused abrupt-migration crash/recovery probe. --- docs/enrollment.md | 42 ++ .../news/+portable-issuer-controls.feature.md | 3 + server/src/arbiter_server/authority.py | 24 +- .../arbiter_server/enrollment_activation.py | 14 + server/src/arbiter_server/generations.py | 2 +- server/src/arbiter_server/issuer_admin.py | 62 +++ server/src/arbiter_server/local_control.py | 11 + server/src/arbiter_server/main.py | 29 ++ server/src/arbiter_server/owner_issuers.py | 197 +++++++++ server/tests/unit/test_authority.py | 4 +- .../tests/unit/test_enrollment_activation.py | 4 + server/tests/unit/test_owner_issuers.py | 386 ++++++++++++++++++ 12 files changed, 771 insertions(+), 7 deletions(-) create mode 100644 docs/enrollment.md create mode 100644 server/news/+portable-issuer-controls.feature.md create mode 100644 server/src/arbiter_server/issuer_admin.py create mode 100644 server/src/arbiter_server/owner_issuers.py create mode 100644 server/tests/unit/test_owner_issuers.py diff --git a/docs/enrollment.md b/docs/enrollment.md new file mode 100644 index 0000000..18957d7 --- /dev/null +++ b/docs/enrollment.md @@ -0,0 +1,42 @@ +# Enrollment and owner issuer trust + +Fresh client enrollment uses the existing `arbiter enrol` command and local +operator approval. Registering an owner issuer establishes public trust only; +it does not enroll a client or grant account access. Selecting an existing +portable identity for another server remains unavailable in this revision. + +Use `arbiter-server authority owner-issuer` through the same authenticated local +OS account as server administration. Commands work while the instance is stopped +or through its authenticated local control while running. Remote administration +uses verified SSH to run these commands on the server host. + +```sh +arbiter-server authority owner-issuer list --state-dir ROOT --json +arbiter-server authority owner-issuer export --state-dir ROOT --issuer-sha256 FINGERPRINT > owner.der +arbiter-server authority owner-issuer register --state-dir ROOT --certificate-file owner.der --json +arbiter-server authority owner-issuer remove --state-dir ROOT --issuer-sha256 FINGERPRINT --json +``` + +Obtain public owner DER from an independently trusted operator. Registration +requires the exact valid Arbiter owner certificate. Keep private issuer and +recovery keys private. The fingerprint is the lowercase SHA-256 of the exact +canonical DER. Distinct issuer keys can share an owner ID; each certificate +requires its own explicit registration. + +List returns at most 32 summaries and a `next_cursor`. Pass that value with +`list --cursor FINGERPRINT` to continue. Export returns one public DER certificate +as binary stdout. Register reads at most 4,096 bytes from the local CLI file; +it sends certificate bytes, not a file path for the server to open. + +Removing trust retains an inactive record. Repeating the same registration or +removal is idempotent; reactivating a removed certificate advances its trust +revision. Restart does not reactivate removed trust. Removal blocks new +activation under that issuer and does not revoke an already activated access +credential or change its historical enrollment result. + +The first upgraded startup performs an atomic, one-way authority schema upgrade. +It preserves existing credentials, exact enrollment results, denial and inactivity +state, and binds each existing agent to its enrollment. Invalid or duplicate +historical bindings stop startup and preserve the old database. Preserve the +state and resolve the reported error; do not delete authority files to retry. +New fresh enrollments maintain that binding in the same activation transaction. diff --git a/server/news/+portable-issuer-controls.feature.md b/server/news/+portable-issuer-controls.feature.md new file mode 100644 index 0000000..e63a190 --- /dev/null +++ b/server/news/+portable-issuer-controls.feature.md @@ -0,0 +1,3 @@ +Add authenticated local public owner-issuer registration, paginated listing, +export and removal. Upgrade existing authority state atomically and preserve +unique agent-to-enrollment bindings during fresh activation and exact replay. diff --git a/server/src/arbiter_server/authority.py b/server/src/arbiter_server/authority.py index 11f1734..318c0a2 100644 --- a/server/src/arbiter_server/authority.py +++ b/server/src/arbiter_server/authority.py @@ -16,7 +16,7 @@ DATABASE = "authority.sqlite3" _PENDING = ".authority.pending.sqlite3" _LOCK = ".authority.lock" -SCHEMA_VERSION = 3 +SCHEMA_VERSION = 4 # Windows Modules Installer is part of the trusted host OS. This exception is # for ancestors only; authority-root and file ACL allowlists remain unchanged. _TRUSTED_INSTALLER = "S-1-5-80-956008885-3418522649-1831038044-1853292631-2271478464" @@ -268,6 +268,9 @@ def _write_database(path: Path, record: AuthorityRecord) -> None: record.server_id, datetime.fromisoformat(record.created_at), ) + from .owner_issuers import create_tables + + create_tables(db, record) db.execute(f"PRAGMA user_version={SCHEMA_VERSION}") db.commit() with path.open("r+b") as handle: @@ -379,7 +382,13 @@ def _current_record(root: Path) -> AuthorityRecord | None: raise AuthorityError( "authority journal recovery failed; preserve state" ) from exc - return _read_record(path) if path is not None else None + if path is None: + return None + record = _read_record(path, allow_previous=True) + from .owner_issuers import upgrade + + upgrade(path, record) + return _read_record(path) @contextmanager @@ -390,7 +399,7 @@ def running_instance(state_dir: Path) -> Iterator[AuthorityRecord | None]: yield _current_record(root) -def _read_record(path: Path) -> AuthorityRecord: +def _read_record(path: Path, *, allow_previous: bool = False) -> AuthorityRecord: from .credentials import verify_setup_credentials from .recovery import verify_registration @@ -402,8 +411,15 @@ def _read_record(path: Path) -> AuthorityRecord: sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=0) ) as db: version = db.execute("PRAGMA user_version").fetchone()[0] - if version != SCHEMA_VERSION: + if version != SCHEMA_VERSION and not (allow_previous and version == 3): raise AuthorityError("unsupported authority database schema version") + if version == SCHEMA_VERSION: + db.execute( + "SELECT issuer_sha256, owner_id, certificate_der, active, revision FROM owner_issuers LIMIT 0" + ) + db.execute( + "SELECT agent_id, enrollment_id FROM agent_enrollments LIMIT 0" + ) rows = db.execute( "SELECT singleton, owner_id, server_id, created_at FROM authority" ).fetchall() diff --git a/server/src/arbiter_server/enrollment_activation.py b/server/src/arbiter_server/enrollment_activation.py index 530b527..d6a117b 100644 --- a/server/src/arbiter_server/enrollment_activation.py +++ b/server/src/arbiter_server/enrollment_activation.py @@ -4,6 +4,7 @@ from contextlib import closing from datetime import datetime import json +import hashlib from pathlib import Path import sqlite3 @@ -99,7 +100,20 @@ def activate( or existing[2] <= now.timestamp() ): raise ValueError("Enrollment activation unavailable") + if db.execute( + "SELECT enrollment_id FROM agent_enrollments WHERE agent_id=?", + (ids[2],), + ).fetchone() != (ids[3],): + raise ValueError("Enrollment binding unavailable") else: + if db.execute( + "SELECT active FROM owner_issuers WHERE issuer_sha256=? AND owner_id=? AND certificate_der=?", + (hashlib.sha256(fields[2]).hexdigest(), ids[1], fields[2]), + ).fetchone() != (1,): + raise ValueError("Enrollment issuer unavailable") + db.execute( + "INSERT INTO agent_enrollments VALUES (?, ?)", (ids[2], ids[3]) + ) db.execute( "INSERT INTO enrollments VALUES (?, ?, 0, ?, ?, ?)", ( diff --git a/server/src/arbiter_server/generations.py b/server/src/arbiter_server/generations.py index a5b271f..16ccbbe 100644 --- a/server/src/arbiter_server/generations.py +++ b/server/src/arbiter_server/generations.py @@ -181,7 +181,7 @@ def reconcile(root: Path) -> Path | None: raise a.AuthorityError("pending generation digest mismatch") # An already durable selector is authoritative. Its live DB can change # after activation; a stale intent must never roll those changes back. - a._read_record(database) + a._read_record(database, allow_previous=True) if source == staged: a._publish(staged, final) if active is None: diff --git a/server/src/arbiter_server/issuer_admin.py b/server/src/arbiter_server/issuer_admin.py new file mode 100644 index 0000000..377344f --- /dev/null +++ b/server/src/arbiter_server/issuer_admin.py @@ -0,0 +1,62 @@ +"""Bounded public-issuer commands over the existing authenticated local control.""" + +import base64 +import json +from pathlib import Path +import sqlite3 +import sys + +from . import authority as a +from .cli_errors import print_cli_error +from .local_control import _exchange +from .owner_issuers import control + + +def issuer_cli( + command: str, + state_dir: Path, + *, + certificate_file: Path | None = None, + issuer_sha256: str | None = None, + cursor: str | None = None, + as_json: bool = False, +) -> int: + try: + arguments: dict[str, object] = {} + if command == "register": + if certificate_file is None: + raise ValueError("select --certificate-file with public owner DER") + with certificate_file.open("rb") as handle: + der = handle.read(4097) + if not 1 <= len(der) <= 4096: + raise ValueError("owner DER must contain 1–4096 bytes") + arguments["certificate_der"] = base64.b64encode(der).decode() + elif command in {"export", "remove"}: + arguments["issuer_sha256"] = issuer_sha256 + elif cursor is not None: + arguments["cursor"] = cursor + root = a._root(state_dir, create=False) + try: + with a._locked(root): + result = control(root, command, arguments) + except a.AuthorityBusy: + result = _exchange( + root, + { + "root": str(root), + "action": "owner-issuer", + "command": command, + "arguments": arguments, + }, + ) + if command == "export": + encoded = result.get("certificate_der") + if not isinstance(encoded, str) or len(encoded) > 5464: + raise ValueError("invalid public issuer export") + sys.stdout.buffer.write(base64.b64decode(encoded, validate=True)) + else: + print(json.dumps(result, sort_keys=True, indent=None if as_json else 2)) + return 0 + except (OSError, ValueError, sqlite3.Error) as exc: + print_cli_error(str(exc), area="owner-issuer") + return 1 diff --git a/server/src/arbiter_server/local_control.py b/server/src/arbiter_server/local_control.py index 21a6a62..18f0c5a 100644 --- a/server/src/arbiter_server/local_control.py +++ b/server/src/arbiter_server/local_control.py @@ -210,6 +210,17 @@ def __init__( def dispatch(self, value: dict[str, Any]) -> dict[str, Any]: action = value.get("action") + if action == "owner-issuer": + from .owner_issuers import control + + if ( + set(value) != {"root", "action", "command", "arguments"} + or value["root"] != str(self.root) + or not isinstance(value["command"], str) + or not isinstance(value["arguments"], dict) + ): + raise a.AuthorityError("invalid owner-issuer control request") + return control(self.root, value["command"], value["arguments"]) if action == "enrollment": if ( set(value) != {"root", "action", "command", "arguments"} diff --git a/server/src/arbiter_server/main.py b/server/src/arbiter_server/main.py index 34711e4..99f72db 100644 --- a/server/src/arbiter_server/main.py +++ b/server/src/arbiter_server/main.py @@ -4574,6 +4574,24 @@ def _build_parser() -> argparse.ArgumentParser: help="reuse existing recovery keys and register this server's backup verifier", ) + issuer = authority_commands.add_parser( + "owner-issuer", help="manage independently registered public owner issuers" + ) + issuer_commands = issuer.add_subparsers(dest="owner_issuer_command", required=True) + for action in ("register", "list", "export", "remove"): + command = issuer_commands.add_parser(action) + command.add_argument("--state-dir", type=Path, required=True) + if action != "export": + command.add_argument("--json", action="store_true") + if action == "register": + command.add_argument("--certificate-file", type=Path, required=True) + elif action in {"export", "remove"}: + command.add_argument("--issuer-sha256", required=True) + else: + command.add_argument( + "--cursor", help="continue after this exact issuer fingerprint" + ) + enrollment = subcommands.add_parser( "enrollment", help="open or approve enrollment through authenticated local control", @@ -4813,6 +4831,17 @@ def main(argv: Sequence[str] | None = None) -> int: as_json=namespace.json, ) if namespace.command == "authority": + if namespace.authority_command == "owner-issuer": + from .issuer_admin import issuer_cli + + return issuer_cli( + namespace.owner_issuer_command, + namespace.state_dir, + certificate_file=getattr(namespace, "certificate_file", None), + issuer_sha256=getattr(namespace, "issuer_sha256", None), + cursor=getattr(namespace, "cursor", None), + as_json=getattr(namespace, "json", False), + ) from .authority import authority_cli return authority_cli( diff --git a/server/src/arbiter_server/owner_issuers.py b/server/src/arbiter_server/owner_issuers.py new file mode 100644 index 0000000..0886802 --- /dev/null +++ b/server/src/arbiter_server/owner_issuers.py @@ -0,0 +1,197 @@ +"""Exact public issuer trust and the authority's unique agent bindings.""" + +from contextlib import closing +from datetime import datetime, timezone +import base64 +import hashlib +from pathlib import Path +import sqlite3 + +from cryptography import x509 + +from . import authority as a +from .credentials import Identity, Role, parse_certificate, verify_certificate +from .enrollment_receipt import MAX_RESULT, result_fields +from .generations import ACTIVE, _control, _database + + +def create_tables(db: sqlite3.Connection, record: a.AuthorityRecord) -> None: + db.execute( + "CREATE TABLE owner_issuers (issuer_sha256 TEXT PRIMARY KEY " + "CHECK(length(issuer_sha256)=64 AND issuer_sha256 NOT GLOB '*[^0-9a-f]*'), " + "owner_id TEXT NOT NULL, certificate_der BLOB NOT NULL " + "CHECK(typeof(certificate_der)='blob' AND length(certificate_der) BETWEEN 1 AND 4096), " + "active INTEGER NOT NULL CHECK(active IN (0,1)), revision INTEGER NOT NULL " + "CHECK(typeof(revision)='integer' AND revision>0))" + ) + db.execute( + "CREATE TABLE agent_enrollments (agent_id TEXT PRIMARY KEY, " + "enrollment_id TEXT NOT NULL UNIQUE)" + ) + der = db.execute( + "SELECT certificate_der FROM credentials WHERE role=?", (Role.OWNER.value,) + ).fetchone()[0] + db.execute( + "INSERT INTO owner_issuers VALUES (?, ?, ?, 1, 1)", + (hashlib.sha256(der).hexdigest(), record.owner_id, der), + ) + + +def upgrade(path: Path, record: a.AuthorityRecord) -> None: + """Caller holds the instance owner, after publication/journal reconciliation.""" + with closing(sqlite3.connect(path, timeout=0, isolation_level=None)) as db: + version = db.execute("PRAGMA user_version").fetchone()[0] + if version == a.SCHEMA_VERSION: + return + if version != 3: + raise a.AuthorityError("unsupported authority database schema version") + db.execute("PRAGMA synchronous=FULL") + db.execute("BEGIN IMMEDIATE") + try: + create_tables(db, record) + if db.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='enrollments'" + ).fetchone(): + for enrollment, size, raw in db.execute( + "SELECT enrollment_id, length(prepared), substr(prepared,1,?) " + "FROM enrollments", + (MAX_RESULT + 1,), + ): + if not isinstance(raw, bytes) or size > MAX_RESULT: + raise a.AuthorityError("invalid historical enrollment result") + ids, _ = result_fields(raw) + if ( + ids[:2] != [record.server_id, record.owner_id] + or ids[3] != enrollment + ): + raise a.AuthorityError( + "historical enrollment identity mismatch" + ) + db.execute( + "INSERT INTO agent_enrollments VALUES (?, ?)", (ids[2], ids[3]) + ) + db.execute(f"PRAGMA user_version={a.SCHEMA_VERSION}") + db.commit() + except BaseException: + db.rollback() + raise + + +def fingerprint(value: object) -> str: + if ( + not isinstance(value, str) + or len(value) != 64 + or any(c not in "0123456789abcdef" for c in value) + ): + raise a.AuthorityError("select an exact lowercase issuer SHA-256 fingerprint") + return value + + +def control( + root: Path, command: str, arguments: dict[str, object] +) -> dict[str, object]: + """Existing OS-authenticated worker owns all I/O and certificate work.""" + expected = { + "register": {"certificate_der"}, + "remove": {"issuer_sha256"}, + "export": {"issuer_sha256"}, + } + if (command == "list" and set(arguments) not in (set(), {"cursor"})) or ( + command != "list" + and (command not in expected or set(arguments) != expected[command]) + ): + raise a.AuthorityError("invalid owner-issuer control arguments") + owner, der, selected = None, None, None + if command == "register": + encoded = arguments["certificate_der"] + if not isinstance(encoded, str) or len(encoded) > 5464: + raise a.AuthorityError( + "owner certificate must be bounded canonical DER/base64" + ) + der = base64.b64decode(encoded, validate=True) + if base64.b64encode(der).decode() != encoded: + raise a.AuthorityError("noncanonical owner certificate encoding") + cert = parse_certificate(der) + try: + uris = cert.extensions.get_extension_for_class( + x509.SubjectAlternativeName + ).value.get_values_for_type(x509.UniformResourceIdentifier) + except x509.ExtensionNotFound as exc: + raise a.AuthorityError("owner certificate has no identity") from exc + prefix = "urn:arbiter:owner:" + if len(uris) != 1 or not uris[0].startswith(prefix): + raise a.AuthorityError("owner certificate has the wrong identity") + owner = uris[0][len(prefix) :] + verify_certificate( + der, Identity(Role.OWNER, owner), now=datetime.now(timezone.utc) + ) + selected = hashlib.sha256(der).hexdigest() + elif command in {"remove", "export"}: + selected = fingerprint(arguments["issuer_sha256"]) + cursor = fingerprint(arguments["cursor"]) if "cursor" in arguments else "" + if a._current_record(root) is None: + raise a.AuthorityError("initialize server authority before managing issuers") + active = _control(root / ACTIVE) + assert active is not None + path = _database(root, active["active"]) + with closing( + sqlite3.connect( + path.as_uri() + "?mode=rw", uri=True, timeout=0, isolation_level=None + ) + ) as db: + db.execute("PRAGMA synchronous=FULL") + db.execute("BEGIN IMMEDIATE" if command in {"register", "remove"} else "BEGIN") + if command == "register": + row = db.execute( + "SELECT owner_id, certificate_der FROM owner_issuers WHERE issuer_sha256=?", + (selected,), + ).fetchone() + if row is not None and row != (owner, der): + raise a.AuthorityError( + "registered issuer identity mismatch; preserve state" + ) + db.execute( + "INSERT INTO owner_issuers VALUES (?, ?, ?, 1, 1) " + "ON CONFLICT(issuer_sha256) DO UPDATE SET active=1, revision=revision+1 WHERE active=0", + (selected, owner, der), + ) + elif command == "remove": + db.execute( + "UPDATE owner_issuers SET active=0, revision=revision+1 WHERE issuer_sha256=? AND active=1", + (selected,), + ) + if command == "list": + rows = db.execute( + "SELECT issuer_sha256, owner_id, active, revision FROM owner_issuers WHERE issuer_sha256>? ORDER BY issuer_sha256 LIMIT 33", + (cursor,), + ).fetchall() + result: dict[str, object] = { + "issuers": [ + dict(zip(("issuer_sha256", "owner_id", "active", "revision"), row)) + for row in rows[:32] + ], + "next_cursor": rows[31][0] if len(rows) > 32 else None, + } + else: + row = db.execute( + "SELECT owner_id, active, revision, substr(certificate_der,1,4097) FROM owner_issuers WHERE issuer_sha256=?", + (selected,), + ).fetchone() + if row is None: + raise a.AuthorityError("owner issuer is not registered") + result = { + "issuer_sha256": selected, + "owner_id": row[0], + "active": row[1], + "revision": row[2], + } + if command == "export": + if ( + not isinstance(row[3], bytes) + or len(row[3]) > 4096 + or hashlib.sha256(row[3]).hexdigest() != selected + ): + raise a.AuthorityError("registered issuer bytes mismatch") + result["certificate_der"] = base64.b64encode(row[3]).decode() + db.commit() + return {"initialized": True, **result} diff --git a/server/tests/unit/test_authority.py b/server/tests/unit/test_authority.py index 0175ee5..881709a 100644 --- a/server/tests/unit/test_authority.py +++ b/server/tests/unit/test_authority.py @@ -54,7 +54,7 @@ def test_initialize_reopen_and_independent_roots(tmp_path: Path) -> None: ) with closing(sqlite3.connect(dbpath(root))) as db: assert db.execute("PRAGMA integrity_check").fetchone() == ("ok",) - assert db.execute("PRAGMA user_version").fetchone() == (3,) + assert db.execute("PRAGMA user_version").fetchone() == (4,) if os.name != "nt": assert root.stat().st_mode & 0o777 == 0o700 assert (dbpath(root)).stat().st_mode & 0o777 == 0o600 @@ -119,7 +119,7 @@ def no_composition(*args, **kwargs): ) output = json.loads(capsys.readouterr().out) assert output["initialized"] is True - assert output["schema_version"] == 3 + assert output["schema_version"] == 4 assert main([*args, "status", *state]) == 0 assert json.loads(capsys.readouterr().out) == output assert ( diff --git a/server/tests/unit/test_enrollment_activation.py b/server/tests/unit/test_enrollment_activation.py index f738143..973425e 100644 --- a/server/tests/unit/test_enrollment_activation.py +++ b/server/tests/unit/test_enrollment_activation.py @@ -129,6 +129,7 @@ def commit(): commit() with closing(sqlite3.connect(path)) as db: assert enrollment_table(db) is None + assert db.execute("SELECT * FROM agent_enrollments").fetchall() == [] else: assert commit() == raw before = path.read_bytes() @@ -138,6 +139,9 @@ def commit(): with closing(sqlite3.connect(path)) as db: row = db.execute("SELECT * FROM enrollments").fetchall() assert len(row) == 1 and bytes(row[0][1]) == raw + assert db.execute("SELECT * FROM agent_enrollments").fetchall() == [ + (ids[2], ids[3]) + ] assert row[0][3] == int(now.timestamp()) + 180 * 86400 assert json.loads(fields[8]) == [["imap", "personal"]] db.execute("UPDATE enrollments SET denied=1") diff --git a/server/tests/unit/test_owner_issuers.py b/server/tests/unit/test_owner_issuers.py new file mode 100644 index 0000000..790f976 --- /dev/null +++ b/server/tests/unit/test_owner_issuers.py @@ -0,0 +1,386 @@ +"""Real authority migration, issuer administration and fresh-v1 integrity.""" + +import asyncio +import base64 +from contextlib import closing +from datetime import datetime, timedelta, timezone +import hashlib +import json +import sqlite3 +from uuid import UUID, uuid4 + +import pytest + +from arbiter_server import authority as a, credentials as c, generations as g +from arbiter_server import local_control, owner_issuers as issuers, prepared +from arbiter_server.enrollment_activation import activate +from arbiter_server.enrollment_receipt import result_fields +from arbiter_server.main import main +from server.tests.unit.test_pairing_proofs import proof_setup + +pytestmark = pytest.mark.usefixtures("private_windows_test_parent") + + +def setup(tmp_path): + root = tmp_path / "authority" + record = a.initialize_authority(root, recovery_file=tmp_path / "operator.json") + selected = g._control(root / g.ACTIVE) + assert selected is not None + return root, record, g._database(root, selected["active"]) + + +def certificate(owner=None, now=None): + return c.create_issuer( + c.Identity(c.Role.OWNER, owner or str(uuid4())), now=now + ).certificate_der + + +def register(root, der): + return issuers.control( + root, "register", {"certificate_der": base64.b64encode(der).decode()} + ) + + +def commit(root): + _, approval, _, _, _, now = proof_setup() + raw = prepared.prepare(root, approval, int(now.timestamp()) + 300) + activate(root, approval, raw, set, lambda: None, now) + return approval, raw, now + + +def downgrade(path): + with closing(sqlite3.connect(path)) as db: + db.execute("DROP TABLE owner_issuers") + db.execute("DROP TABLE agent_enrollments") + db.execute("PRAGMA user_version=3") + db.commit() + + +def test_exact_issuer_revision_tombstone_and_independent_same_owner(tmp_path): + root, record, path = setup(tmp_path) + der = certificate(record.owner_id) + fp = hashlib.sha256(der).hexdigest() + with a.running_instance(root): + first = register(root, der) + assert first == register(root, der) + assert ( + first["issuer_sha256"] == fp and first["revision"] == first["active"] == 1 + ) + rows = issuers.control(root, "list", {})["issuers"] + assert isinstance(rows, list) + assert len(rows) == 2 and {row["owner_id"] for row in rows} == {record.owner_id} + assert len({row["issuer_sha256"] for row in rows}) == 2 + exported = issuers.control(root, "export", {"issuer_sha256": fp}) + encoded = exported["certificate_der"] + assert isinstance(encoded, str) and base64.b64decode(encoded) == der + removed = issuers.control(root, "remove", {"issuer_sha256": fp}) + assert removed["active"] == 0 and removed["revision"] == 2 + assert issuers.control(root, "remove", {"issuer_sha256": fp}) == removed + before = path.read_bytes() + assert a.authority_status(root) == record + assert path.read_bytes() == before + with a.running_instance(root): + assert issuers.control(root, "export", {"issuer_sha256": fp})["active"] == 0 + revived = register(root, der) + assert revived["revision"] == 3 and revived["active"] == 1 + assert register(root, der) == revived + + +@pytest.mark.parametrize( + "invalid", + [ + "der", + "oversize", + "base64", + "noncanonical", + "wrong-role", + "expired", + "future", + "extra", + ], +) +def test_invalid_registration_preserves_database(tmp_path, invalid): + root, _, path = setup(tmp_path) + now = datetime.now(timezone.utc) + der = certificate( + now=( + now + timedelta(days=1) + if invalid == "future" + else now - timedelta(days=4000) if invalid == "expired" else now + ) + ) + if invalid == "wrong-role": + der = c.create_issuer( + c.Identity(c.Role.SERVER_AUTHORITY, str(uuid4())) + ).certificate_der + if invalid in {"der", "oversize"}: + der = b"x" * (4097 if invalid == "oversize" else 10) + encoded = base64.b64encode(der).decode() + args: dict[str, object] = { + "certificate_der": ( + "!" + if invalid == "base64" + else encoded + "\n" if invalid == "noncanonical" else encoded + ) + } + if invalid == "extra": + args["path"] = "/server/private-file" + with a.running_instance(root): + before = path.read_bytes() + with pytest.raises(ValueError): + issuers.control(root, "register", args) + assert path.read_bytes() == before + + +def test_bounded_pages_and_strict_local_arguments(tmp_path): + root, _, _ = setup(tmp_path) + with a.running_instance(root): + for _ in range(33): + register(root, certificate()) + page = issuers.control(root, "list", {}) + assert isinstance(page["issuers"], list) + assert len(page["issuers"]) == 32 and page["next_cursor"] + tail = issuers.control(root, "list", {"cursor": page["next_cursor"]}) + assert isinstance(tail["issuers"], list) + rows = page["issuers"] + tail["issuers"] + assert len(rows) == len({row["issuer_sha256"] for row in rows}) == 34 + assert tail["next_cursor"] is None + assert all( + set(row) == {"issuer_sha256", "owner_id", "active", "revision"} + for row in rows + ) + invalid_arguments: list[tuple[str, dict[str, object]]] = [ + ("list", {"cursor": "A" * 64}), + ("remove", {"issuer_sha256": True}), + ("register", {"certificate_file": "private"}), + ("replace", {}), + ] + for command, arguments in invalid_arguments: + with pytest.raises(ValueError): + issuers.control(root, command, arguments) + dispatch = local_control.AuthorityControl(root) + with pytest.raises(a.AuthorityError): + dispatch.dispatch( + { + "root": str(root / "other"), + "action": "owner-issuer", + "command": "list", + "arguments": {}, + } + ) + + +def test_upgrade_backfills_real_results_without_mutating_history(tmp_path): + root, record, path = setup(tmp_path) + results = [commit(root)[1] for _ in range(3)] + with closing(sqlite3.connect(path)) as db: + db.execute( + "UPDATE enrollments SET denied=1 WHERE enrollment_id=?", + (result_fields(results[1])[0][3],), + ) + db.execute( + "UPDATE enrollments SET inactive_after=1 WHERE enrollment_id=?", + (result_fields(results[2])[0][3],), + ) + db.commit() + before = db.execute( + "SELECT * FROM enrollments ORDER BY enrollment_id" + ).fetchall() + credentials = db.execute("SELECT * FROM credentials").fetchall() + downgrade(path) + with pytest.raises(a.AuthorityError, match="schema"): + a._read_record(path) + selected = g._control(root / g.ACTIVE) + assert selected is not None + g._write_control( + root, + g._INTENT, + {"version": 1, "active": selected["active"], "sha256": g._digest(path)}, + ) + assert a.authority_status(root) == record + assert not (root / g._INTENT).exists() + with closing(sqlite3.connect(path)) as db: + assert db.execute("PRAGMA user_version").fetchone() == (4,) + assert ( + db.execute("SELECT * FROM enrollments ORDER BY enrollment_id").fetchall() + == before + ) + assert db.execute("SELECT * FROM credentials").fetchall() == credentials + assert set(db.execute("SELECT * FROM agent_enrollments")) == { + (result_fields(raw)[0][2], result_fields(raw)[0][3]) for raw in results + } + unchanged = path.read_bytes() + assert a.authority_status(root) == record and path.read_bytes() == unchanged + _, raw, _ = commit(root) + with closing(sqlite3.connect(path)) as db: + assert db.execute("SELECT COUNT(*) FROM agent_enrollments").fetchone() == (4,) + assert db.execute( + "SELECT prepared FROM enrollments WHERE enrollment_id=?", + (result_fields(raw)[0][3],), + ).fetchone() == (raw,) + + +@pytest.mark.parametrize( + "damage", ["malformed", "server", "enrollment", "duplicate", "interrupted"] +) +def test_failed_upgrade_is_atomic_and_preserves_old_database( + tmp_path, monkeypatch, damage +): + root, record, path = setup(tmp_path) + first, second = commit(root)[1], commit(root)[1] + ids, _ = result_fields(second) + if damage != "interrupted": + replacement = ( + b"broken" + if damage == "malformed" + else ( + second.replace(ids[0].encode(), str(uuid4()).encode(), 1) + if damage == "server" + else ( + second.replace(ids[3].encode(), str(uuid4()).encode(), 1) + if damage == "enrollment" + else second.replace( + ids[2].encode(), result_fields(first)[0][2].encode(), 1 + ) + ) + ) + ) + with closing(sqlite3.connect(path)) as db: + db.execute( + "UPDATE enrollments SET prepared=? WHERE enrollment_id=?", + (replacement, ids[3]), + ) + db.commit() + downgrade(path) + original = issuers.create_tables + if damage == "interrupted": + + def fail(db, authority): + original(db, authority) + raise OSError("synthetic failure after schema writes") + + monkeypatch.setattr(issuers, "create_tables", fail) + before = path.read_bytes() + expected: tuple[type[Exception], ...] = (ValueError, sqlite3.Error, OSError) + with pytest.raises(expected): + a.authority_status(root) + assert path.read_bytes() == before + with closing(sqlite3.connect(path)) as db: + assert db.execute("PRAGMA user_version").fetchone() == (3,) + assert ( + db.execute( + "SELECT 1 FROM sqlite_master WHERE name='agent_enrollments'" + ).fetchone() + is None + ) + if damage == "interrupted": + monkeypatch.setattr(issuers, "create_tables", original) + assert a.authority_status(root) == record + + +def test_empty_upgrade_creates_no_synthetic_enrollment(tmp_path): + root, record, path = setup(tmp_path) + downgrade(path) + assert a.authority_status(root) == record + with closing(sqlite3.connect(path)) as db: + assert db.execute("SELECT * FROM agent_enrollments").fetchall() == [] + assert ( + db.execute( + "SELECT 1 FROM sqlite_master WHERE name='enrollments'" + ).fetchone() + is None + ) + + +@pytest.mark.parametrize("state", ["active", "denied", "inactive"]) +def test_fresh_v1_cannot_duplicate_existing_agent(tmp_path, monkeypatch, state): + root, _, path = setup(tmp_path) + _, raw, _ = commit(root) + ids, _ = result_fields(raw) + with closing(sqlite3.connect(path)) as db: + if state != "active": + db.execute( + "UPDATE enrollments SET " + + ("denied=1" if state == "denied" else "inactive_after=1") + ) + db.commit() + values = iter([UUID(ids[2]), uuid4()]) + monkeypatch.setattr(prepared, "uuid4", lambda: next(values)) + with pytest.raises(sqlite3.IntegrityError): + commit(root) + with closing(sqlite3.connect(path)) as db: + assert db.execute("SELECT * FROM agent_enrollments").fetchall() == [ + (ids[2], ids[3]) + ] + assert db.execute("SELECT COUNT(*) FROM enrollments").fetchone() == (1,) + + +def test_exact_replay_does_not_repair_missing_binding_or_reactivate_issuer(tmp_path): + root, record, path = setup(tmp_path) + approval, raw, now = commit(root) + with a.running_instance(root): + rows = issuers.control(root, "list", {})["issuers"] + assert isinstance(rows, list) + local = next(row for row in rows if row["owner_id"] == record.owner_id) + issuers.control(root, "remove", {"issuer_sha256": local["issuer_sha256"]}) + assert activate(root, approval, raw, set, lambda: None, now) == raw + with pytest.raises(ValueError, match="issuer unavailable"): + commit(root) + with closing(sqlite3.connect(path)) as db: + db.execute("DELETE FROM agent_enrollments") + db.commit() + before = path.read_bytes() + with pytest.raises(ValueError, match="binding unavailable"): + activate(root, approval, raw, set, lambda: None, now) + assert path.read_bytes() == before + + +def test_real_native_control_and_cli_public_export(tmp_path, capsysbinary): + root, _, _ = setup(tmp_path) + der = certificate() + path = tmp_path / "public.der" + path.write_bytes(der) + fp = hashlib.sha256(der).hexdigest() + + async def exercise(): + with a.running_instance(root): + async with local_control.running(root): + prefix = ["authority", "owner-issuer"] + + async def command(action, *args): + return await asyncio.to_thread( + main, prefix + [action, "--state-dir", str(root), *args] + ) + + assert ( + await command("register", "--certificate-file", str(path), "--json") + == 0 + ) + reply = json.loads(capsysbinary.readouterr().out) + assert reply["issuer_sha256"] == fp and reply["revision"] == 1 + assert await command("export", "--issuer-sha256", fp) == 0 + assert capsysbinary.readouterr().out == der + assert await command("remove", "--issuer-sha256", fp) == 0 + assert json.loads(capsysbinary.readouterr().out)["active"] == 0 + assert await command("list", "--json") == 0 + assert len(json.loads(capsysbinary.readouterr().out)["issuers"]) == 2 + + asyncio.run(exercise()) + + +def test_stopped_cli_and_oversize_input_preserve_state(tmp_path, capsys): + root, _, path = setup(tmp_path) + prefix = ["authority", "owner-issuer"] + assert main(prefix + ["list", "--state-dir", str(root), "--json"]) == 0 + assert len(json.loads(capsys.readouterr().out)["issuers"]) == 1 + public = tmp_path / "oversize.der" + public.write_bytes(b"x" * 4097) + before = path.read_bytes() + assert ( + main( + prefix + + ["register", "--state-dir", str(root), "--certificate-file", str(public)] + ) + == 1 + ) + assert path.read_bytes() == before