stackchain-dashboard/tests/test_access_revocation.py

import sqlite3

import pytest

from src.passkey_store import PasskeyStore
from src.session_store import SessionStore, SessionStoreError


def test_global_access_revocation_preserves_other_principal_credentials_and_sessions(
    tmp_path,
):
    database = tmp_path / "sessions.sqlite3"
    sessions = SessionStore(database, clock=lambda: 1_000.0)
    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
    for principal_id in (42, 84):
        sessions.activate(
            f"session-{principal_id}",
            2_000,
            management_id=f"management-{principal_id}",
            principal_id=principal_id,
            principal_login=f"operator-{principal_id}",
        )
        passkeys.register(
            credential_id=f"credential-{principal_id}".encode(),
            public_key=f"public-key-{principal_id}".encode(),
            sign_count=0,
            device_label=f"Device {principal_id}",
            management_id=f"management-{principal_id}",
            principal_id=principal_id,
        )

    passkeys.revoke_all_access(principal_id=42)

    assert passkeys.all(principal_id=42) == []
    assert [item.principal_id for item in passkeys.all(principal_id=84)] == [84]
    with sqlite3.connect(database) as connection:
        assert connection.execute(
            "SELECT principal_id FROM active_sessions ORDER BY principal_id"
        ).fetchall() == [(84,)]


def test_global_access_revocation_rolls_back_every_credential_and_session_change(tmp_path):
    database = tmp_path / "sessions.sqlite3"
    sessions = SessionStore(database, clock=lambda: 1_000.0)
    sessions.activate(
        "phone-session",
        2_000,
        management_id="phone-management-id",
        device_label="Phone",
        principal_id=42,
        principal_login="timmy",
    )
    sessions.mint_step_up(
        "phone-session",
        action="revoke_all_sessions",
        target="all",
        ttl_seconds=90,
    )
    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
    passkeys.register(
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=0,
        device_label="Phone",
        management_id="phone-management-id",
        principal_id=42,
    )
    passkeys.issue_challenge(
        b"pending-challenge",
        session_id="phone-session",
        purpose="authentication",
        action="sign_in",
        target="dashboard",
    )
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TRIGGER block_session_delete BEFORE DELETE ON active_sessions "
            "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
        )

    with pytest.raises(SessionStoreError):
        passkeys.revoke_all_access(principal_id=42)

    with sqlite3.connect(database) as connection:
        assert connection.execute("SELECT COUNT(*) FROM passkey_credentials").fetchone() == (1,)
        assert connection.execute("SELECT COUNT(*) FROM passkey_challenges").fetchone() == (1,)
        assert connection.execute("SELECT COUNT(*) FROM step_up_grants").fetchone() == (1,)
        assert connection.execute("SELECT COUNT(*) FROM active_sessions").fetchone() == (1,)


def test_device_access_revocation_rolls_back_target_and_preserves_other_devices(tmp_path):
    database = tmp_path / "sessions.sqlite3"
    sessions = SessionStore(database, clock=lambda: 1_000.0)
    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
    for label in ("Phone", "Laptop"):
        slug = label.lower()
        sessions.activate(
            f"{slug}-session",
            2_000,
            management_id=f"{slug}-management-id",
            device_label=label,
            principal_id=42,
            principal_login="timmy",
        )
        sessions.mint_step_up(
            f"{slug}-session",
            action="close_issue",
            target="stackchain/dashboard#1",
            ttl_seconds=90,
        )
        passkeys.register(
            credential_id=f"{slug}-credential".encode(),
            public_key=f"{slug}-public-key".encode(),
            sign_count=0,
            device_label=label,
            management_id=f"{slug}-management-id",
            principal_id=42,
        )
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TRIGGER block_phone_delete BEFORE DELETE ON active_sessions "
            "WHEN OLD.management_id = 'phone-management-id' "
            "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
        )

    with pytest.raises(SessionStoreError):
        passkeys.revoke_device_access("phone-management-id", principal_id=42)

    with sqlite3.connect(database) as connection:
        assert connection.execute(
            "SELECT management_id FROM active_sessions ORDER BY management_id"
        ).fetchall() == [("laptop-management-id",), ("phone-management-id",)]
        assert connection.execute(
            "SELECT management_id FROM passkey_credentials ORDER BY management_id"
        ).fetchall() == [("laptop-management-id",), ("phone-management-id",)]
        assert connection.execute("SELECT COUNT(*) FROM step_up_grants").fetchone() == (2,)


def test_device_access_revocation_cannot_cross_principal_boundary(tmp_path):
    database = tmp_path / "sessions.sqlite3"
    sessions = SessionStore(database, clock=lambda: 1_000.0)
    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
    for principal_id in (42, 84):
        sessions.activate(
            f"session-{principal_id}",
            2_000,
            management_id=f"management-{principal_id}",
            principal_id=principal_id,
            principal_login=f"operator-{principal_id}",
        )
        sessions.mint_step_up(
            f"session-{principal_id}",
            action="close_issue",
            target="stackchain/dashboard#1",
            ttl_seconds=90,
        )
        passkeys.register(
            credential_id=f"credential-{principal_id}".encode(),
            public_key=f"public-key-{principal_id}".encode(),
            sign_count=0,
            device_label=f"Device {principal_id}",
            management_id=f"management-{principal_id}",
            principal_id=principal_id,
        )

    assert passkeys.revoke_device_access("management-84", principal_id=42) is False
    assert passkeys.revoke_device_access("management-42", principal_id=42) is True

    with sqlite3.connect(database) as connection:
        assert connection.execute(
            "SELECT principal_id FROM active_sessions ORDER BY principal_id"
        ).fetchall() == [(84,)]
        assert connection.execute(
            "SELECT principal_id FROM passkey_credentials ORDER BY principal_id"
        ).fetchall() == [(84,)]
        assert connection.execute("SELECT COUNT(*) FROM step_up_grants").fetchone() == (1,)


def test_global_access_revocation_returns_only_affected_management_ids(tmp_path):
    database = tmp_path / "sessions.sqlite3"
    sessions = SessionStore(database, clock=lambda: 1_000.0)
    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
    for principal_id in (42, 84):
        for suffix in ("phone", "laptop"):
            sessions.activate(
                f"session-{principal_id}-{suffix}",
                2_000,
                management_id=f"management-{principal_id}-{suffix}",
                principal_id=principal_id,
                principal_login=f"operator-{principal_id}",
            )

    affected = passkeys.revoke_all_access(principal_id=42)

    assert affected == ["management-42-laptop", "management-42-phone"]
    with sqlite3.connect(database) as connection:
        assert connection.execute(
            "SELECT management_id FROM active_sessions ORDER BY management_id"
        ).fetchall() == [
            ("management-84-laptop",),
            ("management-84-phone",),
        ]