203 lines
7.7 KiB
Python
203 lines
7.7 KiB
Python
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",),
|
|
]
|