stackchain-dashboard/tests/test_passkey_store.py

import sqlite3

from src.passkey_store import PasskeyStore


def test_credentials_are_bound_to_one_upstream_principal_and_legacy_rows_fail_closed(
    tmp_path,
):
    database = tmp_path / "passkeys.sqlite3"
    with sqlite3.connect(database) as connection:
        connection.execute(
            """
            CREATE TABLE passkey_credentials (
                credential_id BLOB PRIMARY KEY,
                public_key BLOB NOT NULL,
                sign_count INTEGER NOT NULL,
                device_label TEXT NOT NULL,
                management_id TEXT NOT NULL UNIQUE,
                created_at INTEGER NOT NULL
            )
            """
        )
        connection.execute(
            "INSERT INTO passkey_credentials VALUES (?, ?, ?, ?, ?, ?)",
            (b"legacy", b"legacy-key", 0, "Old phone", "legacy-management", 900),
        )

    store = PasskeyStore(database, clock=lambda: 1_000.0)
    store.register(
        credential_id=b"principal-101",
        public_key=b"public-key",
        sign_count=0,
        device_label="Phone",
        management_id="phone-management",
        principal_id=101,
    )

    assert [item.credential_id for item in store.all(principal_id=101)] == [b"principal-101"]
    assert store.all(principal_id=202) == []
    assert store.get(b"principal-101", principal_id=101).principal_id == 101
    assert store.get(b"principal-101", principal_id=202) is None
    assert store.get(b"legacy", principal_id=101) is None
    with sqlite3.connect(database) as connection:
        assert connection.execute(
            "SELECT principal_id FROM passkey_credentials WHERE credential_id = ?",
            (b"legacy",),
        ).fetchone() == (None,)


def test_active_challenges_are_bounded_per_source_and_globally_across_instances(tmp_path):
    now = [1_000.0]
    database = tmp_path / "passkeys.sqlite3"
    first = PasskeyStore(
        database,
        clock=lambda: now[0],
        max_challenges=3,
        max_challenges_per_source=2,
    )
    second = PasskeyStore(
        database,
        clock=lambda: now[0],
        max_challenges=3,
        max_challenges_per_source=2,
    )

    for challenge in (b"first", b"second", b"third"):
        first.issue_challenge(
            challenge,
            session_id=None,
            purpose="authentication",
            action="sign_in",
            target="dashboard",
            source="203.0.113.7",
        )
    second.issue_challenge(
        b"other-one",
        session_id=None,
        purpose="authentication",
        action="sign_in",
        target="dashboard",
        source="203.0.113.8",
    )
    second.issue_challenge(
        b"other-two",
        session_id=None,
        purpose="authentication",
        action="sign_in",
        target="dashboard",
        source="203.0.113.8",
    )

    with sqlite3.connect(database) as connection:
        assert connection.execute("SELECT COUNT(*) FROM passkey_challenges").fetchone() == (3,)
        counts = connection.execute(
            "SELECT source_hash, COUNT(*) FROM passkey_challenges GROUP BY source_hash"
        ).fetchall()
    assert sorted(count for _source, count in counts) == [1, 2]

    now[0] = 1_121.0
    first.issue_challenge(
        b"after-expiry",
        session_id=None,
        purpose="authentication",
        action="sign_in",
        target="dashboard",
        source="203.0.113.9",
    )
    with sqlite3.connect(database) as connection:
        assert connection.execute("SELECT COUNT(*) FROM passkey_challenges").fetchone() == (1,)


def test_existing_challenge_registry_is_migrated_without_losing_live_challenges(tmp_path):
    database = tmp_path / "passkeys.sqlite3"
    with sqlite3.connect(database) as connection:
        connection.execute(
            """
            CREATE TABLE passkey_challenges (
                challenge_hash TEXT PRIMARY KEY,
                session_hash TEXT,
                purpose TEXT NOT NULL,
                action TEXT NOT NULL,
                target TEXT NOT NULL,
                expires_at INTEGER NOT NULL
            )
            """
        )
        connection.execute(
            "INSERT INTO passkey_challenges VALUES ('existing', NULL, "
            "'authentication', 'sign_in', 'dashboard', 1120)"
        )
    store = PasskeyStore(database, clock=lambda: 1_000.0)

    store.issue_challenge(
        b"new",
        session_id=None,
        purpose="authentication",
        action="sign_in",
        target="dashboard",
        source="203.0.113.7",
    )

    with sqlite3.connect(database) as connection:
        columns = {
            row[1] for row in connection.execute("PRAGMA table_info(passkey_challenges)")
        }
        rows = connection.execute(
            "SELECT challenge_hash FROM passkey_challenges ORDER BY challenge_hash"
        ).fetchall()
    assert "source_hash" in columns
    assert sorted(rows) == sorted([("existing",), (store._digest(b"new"),)])


def test_passkey_counter_advancement_is_atomic_across_store_instances(tmp_path):
    database = tmp_path / "passkeys.sqlite3"
    first = PasskeyStore(database, clock=lambda: 1_000.0)
    second = PasskeyStore(database, clock=lambda: 1_000.0)
    first.register(
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=4,
        device_label="Phone",
        management_id="phone-management-id",
        principal_id=42,
    )

    assert first.advance_counter(b"phone-credential", expected=4, new=5) is True
    assert second.advance_counter(b"phone-credential", expected=4, new=5) is False
    assert first.get(b"phone-credential", principal_id=42).sign_count == 5


def test_passkey_counter_rejects_non_advancing_values_but_supports_counterless_devices(
    tmp_path,
):
    store = PasskeyStore(tmp_path / "passkeys.sqlite3", clock=lambda: 1_000.0)
    store.register(
        credential_id=b"phone-credential",
        public_key=b"phone-public-key",
        sign_count=4,
        device_label="Phone",
        management_id="phone-management-id",
        principal_id=42,
    )
    store.register(
        credential_id=b"counterless-credential",
        public_key=b"counterless-public-key",
        sign_count=0,
        device_label="Security key",
        management_id="counterless-management-id",
        principal_id=42,
    )

    assert store.advance_counter(b"phone-credential", expected=4, new=4) is False
    assert store.advance_counter(b"phone-credential", expected=4, new=3) is False
    assert store.advance_counter(b"counterless-credential", expected=0, new=0) is True