148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
import sqlite3
|
|
|
|
from src.passkey_store import PasskeyStore
|
|
|
|
|
|
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",
|
|
)
|
|
|
|
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").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",
|
|
)
|
|
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",
|
|
)
|
|
|
|
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
|