107 lines
3.2 KiB
Python
107 lines
3.2 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"),)])
|