404 lines
16 KiB
Python
404 lines
16 KiB
Python
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from src import session_store
|
|
from src.session_store import SessionStore, SessionStoreError
|
|
|
|
|
|
def test_revocation_is_durable_and_scoped_to_one_session(tmp_path):
|
|
now = [1_000.0]
|
|
database = tmp_path / "sessions.sqlite3"
|
|
first = SessionStore(database, clock=lambda: now[0])
|
|
first.activate("first-session-secret", 2_000)
|
|
first.activate("second-session-secret", 2_000)
|
|
|
|
reconstructed = SessionStore(database, clock=lambda: now[0])
|
|
reconstructed.revoke("first-session-secret")
|
|
|
|
assert reconstructed.is_active("first-session-secret", 2_000) is False
|
|
assert reconstructed.is_active("second-session-secret", 2_000) is True
|
|
assert b"first-session-secret" not in database.read_bytes()
|
|
assert b"second-session-secret" not in database.read_bytes()
|
|
|
|
|
|
def test_revoke_all_is_durable_and_does_not_block_future_sessions(tmp_path):
|
|
now = [1_000.0]
|
|
database = tmp_path / "sessions.sqlite3"
|
|
first = SessionStore(database, clock=lambda: now[0])
|
|
first.activate("phone-session", 2_000)
|
|
first.activate("laptop-session", 2_000)
|
|
|
|
first.revoke_all()
|
|
|
|
reconstructed = SessionStore(database, clock=lambda: now[0])
|
|
assert reconstructed.is_active("phone-session", 2_000) is False
|
|
assert reconstructed.is_active("laptop-session", 2_000) is False
|
|
reconstructed.activate("new-session", 2_000)
|
|
assert reconstructed.is_active("new-session", 2_000) is True
|
|
|
|
|
|
def test_validation_does_not_create_a_missing_registry(tmp_path):
|
|
database = tmp_path / "sessions.sqlite3"
|
|
store = SessionStore(database, clock=lambda: 1_000.0)
|
|
|
|
with pytest.raises(SessionStoreError):
|
|
store.is_active("unknown-session", 2_000)
|
|
|
|
assert database.exists() is False
|
|
|
|
|
|
def test_validation_executes_only_a_read_query(tmp_path, monkeypatch):
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
|
store.activate("active-session", 2_000)
|
|
statements = []
|
|
connect = sqlite3.connect
|
|
|
|
def traced_connect(*args, **kwargs):
|
|
connection = connect(*args, **kwargs)
|
|
connection.set_trace_callback(statements.append)
|
|
return connection
|
|
|
|
monkeypatch.setattr(session_store.sqlite3, "connect", traced_connect)
|
|
|
|
assert store.is_active("active-session", 2_000) is True
|
|
assert [statement.split()[0].upper() for statement in statements] == ["SELECT"]
|
|
|
|
|
|
def test_expired_sessions_are_rejected_without_writing_during_validation(tmp_path):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("expiring-session", 1_001)
|
|
|
|
now[0] = 1_001.0
|
|
|
|
assert store.is_active("expiring-session", 1_001) is False
|
|
with sqlite3.connect(store.path) as connection:
|
|
assert connection.execute("SELECT COUNT(*) FROM active_sessions").fetchone() == (1,)
|
|
|
|
|
|
def test_session_status_reports_idle_without_background_validation_extending_activity(tmp_path):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("phone-session", 3_000)
|
|
|
|
now[0] = 1_899.0
|
|
assert store.status("phone-session", 3_000, idle_timeout_seconds=900) == "active"
|
|
now[0] = 1_900.0
|
|
assert store.status("phone-session", 3_000, idle_timeout_seconds=900) == "idle"
|
|
|
|
|
|
def test_active_session_status_carries_bound_upstream_identity(tmp_path):
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
|
|
|
store.activate(
|
|
"phone-session",
|
|
3_000,
|
|
principal_id=42,
|
|
principal_login="timmy",
|
|
)
|
|
|
|
status = store.status("phone-session", 3_000, idle_timeout_seconds=900)
|
|
|
|
assert status == "active"
|
|
assert status.principal_id == 42
|
|
assert status.principal_login == "timmy"
|
|
|
|
|
|
def test_managed_session_status_enforces_absolute_and_idle_expiry(tmp_path):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("phone-session", 3_000, management_id="phone-device")
|
|
store.activate("short-session", 1_200, management_id="short-device")
|
|
|
|
assert store.managed_status("phone-device", idle_timeout_seconds=900) == "active"
|
|
now[0] = 1_200.0
|
|
assert store.managed_status("short-device", idle_timeout_seconds=900) == "revoked"
|
|
now[0] = 1_900.0
|
|
assert store.managed_status("phone-device", idle_timeout_seconds=900) == "idle"
|
|
assert store.managed_status("missing-device", idle_timeout_seconds=900) == "revoked"
|
|
|
|
|
|
def test_managed_session_statuses_classify_a_batch_with_one_read(tmp_path, monkeypatch):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("active-session", 3_000, management_id="active-device")
|
|
store.activate("idle-session", 3_000, management_id="idle-device")
|
|
store.activate("expired-session", 1_500, management_id="expired-device")
|
|
now[0] = 2_000.0
|
|
store.touch("active-session", 3_000, idle_timeout_seconds=1_500)
|
|
statements = []
|
|
connect = sqlite3.connect
|
|
|
|
def traced_connect(*args, **kwargs):
|
|
connection = connect(*args, **kwargs)
|
|
connection.set_trace_callback(statements.append)
|
|
return connection
|
|
|
|
monkeypatch.setattr(session_store.sqlite3, "connect", traced_connect)
|
|
|
|
assert store.managed_statuses(
|
|
["active-device", "idle-device", "expired-device", "missing-device", "active-device"],
|
|
idle_timeout_seconds=900,
|
|
) == {
|
|
"active-device": "active",
|
|
"idle-device": "idle",
|
|
"expired-device": "revoked",
|
|
"missing-device": "revoked",
|
|
}
|
|
assert [statement for statement in statements if statement.lstrip().upper().startswith("SELECT")] == [
|
|
next(statement for statement in statements if "FROM active_sessions" in statement)
|
|
]
|
|
|
|
|
|
def test_managed_session_statuses_reject_other_and_legacy_principals_in_one_read(
|
|
tmp_path, monkeypatch
|
|
):
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
|
store.activate(
|
|
"matching-session",
|
|
3_000,
|
|
management_id="matching-device",
|
|
principal_id=42,
|
|
principal_login="timmy",
|
|
)
|
|
store.activate(
|
|
"other-session",
|
|
3_000,
|
|
management_id="other-device",
|
|
principal_id=84,
|
|
principal_login="other",
|
|
)
|
|
store.activate("legacy-session", 3_000, management_id="legacy-device")
|
|
statements = []
|
|
connect = sqlite3.connect
|
|
|
|
def traced_connect(*args, **kwargs):
|
|
connection = connect(*args, **kwargs)
|
|
connection.set_trace_callback(statements.append)
|
|
return connection
|
|
|
|
monkeypatch.setattr(session_store.sqlite3, "connect", traced_connect)
|
|
|
|
assert store.managed_statuses(
|
|
["matching-device", "other-device", "legacy-device"],
|
|
idle_timeout_seconds=900,
|
|
expected_principal_id=42,
|
|
) == {
|
|
"matching-device": "active",
|
|
"other-device": "principal_mismatch",
|
|
"legacy-device": "principal_mismatch",
|
|
}
|
|
assert [statement for statement in statements if statement.lstrip().upper().startswith("SELECT")] == [
|
|
next(statement for statement in statements if "FROM active_sessions" in statement)
|
|
]
|
|
|
|
|
|
def test_managed_session_statuses_skips_database_for_an_empty_batch(tmp_path, monkeypatch):
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
|
monkeypatch.setattr(store, "_connect", lambda *args, **kwargs: pytest.fail("database opened"))
|
|
|
|
assert store.managed_statuses([], idle_timeout_seconds=900) == {}
|
|
|
|
|
|
def test_touch_extends_only_the_matching_live_session(tmp_path):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("phone-session", 3_000)
|
|
store.activate("laptop-session", 3_000)
|
|
|
|
now[0] = 1_800.0
|
|
assert store.touch("phone-session", 3_000, idle_timeout_seconds=900) is True
|
|
now[0] = 2_000.0
|
|
|
|
assert store.status("phone-session", 3_000, idle_timeout_seconds=900) == "active"
|
|
assert store.status("laptop-session", 3_000, idle_timeout_seconds=900) == "idle"
|
|
assert store.touch("missing-session", 3_000, idle_timeout_seconds=900) is False
|
|
|
|
|
|
def test_touch_cannot_revive_a_session_at_the_idle_boundary(tmp_path):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("phone-session", 3_000)
|
|
|
|
now[0] = 1_900.0
|
|
|
|
assert store.touch("phone-session", 3_000, idle_timeout_seconds=900) is False
|
|
assert store.status("phone-session", 3_000, idle_timeout_seconds=900) == "idle"
|
|
|
|
|
|
def test_active_devices_are_listed_without_exposing_session_secrets(tmp_path):
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
|
store.activate("phone-session-secret", 2_000, device_label="Pixel 9")
|
|
store.activate("laptop-session-secret", 3_000, device_label="Work laptop")
|
|
|
|
devices = store.list_active("phone-session-secret")
|
|
|
|
assert [device.device_label for device in devices] == ["Work laptop", "Pixel 9"]
|
|
assert [device.current for device in devices] == [False, True]
|
|
assert all(device.management_id for device in devices)
|
|
assert all(device.created_at == 1_000 for device in devices)
|
|
assert "phone-session-secret" not in repr(devices)
|
|
|
|
|
|
def test_revoke_managed_device_removes_only_the_selected_session(tmp_path):
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
|
store.activate("phone", 2_000, device_label="Phone")
|
|
store.activate("laptop", 2_000, device_label="Laptop")
|
|
phone = next(device for device in store.list_active("laptop") if device.device_label == "Phone")
|
|
|
|
assert store.revoke_managed(phone.management_id) is True
|
|
assert store.is_active("phone", 2_000) is False
|
|
assert store.is_active("laptop", 2_000) is True
|
|
assert store.revoke_managed(phone.management_id) is False
|
|
|
|
|
|
def test_existing_session_registry_migrates_without_invalidating_sessions(tmp_path):
|
|
database = tmp_path / "sessions.sqlite3"
|
|
digest = SessionStore._digest("existing-session")
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"CREATE TABLE active_sessions (session_hash TEXT PRIMARY KEY, expires_at INTEGER NOT NULL)"
|
|
)
|
|
connection.execute("INSERT INTO active_sessions VALUES (?, ?)", (digest, 2_000))
|
|
|
|
store = SessionStore(database, clock=lambda: 1_000.0)
|
|
store.activate("new-session", 3_000, device_label="New phone")
|
|
|
|
assert store.is_active("existing-session", 2_000) is True
|
|
devices = store.list_active("existing-session")
|
|
assert len(devices) == 2
|
|
assert next(device for device in devices if device.current).device_label == "Existing device"
|
|
|
|
|
|
def test_idle_status_migrates_existing_registry_and_starts_legacy_idle_clock_now(tmp_path):
|
|
database = tmp_path / "sessions.sqlite3"
|
|
digest = SessionStore._digest("existing-session")
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"CREATE TABLE active_sessions (session_hash TEXT PRIMARY KEY, expires_at INTEGER NOT NULL)"
|
|
)
|
|
connection.execute("INSERT INTO active_sessions VALUES (?, ?)", (digest, 2_000))
|
|
|
|
store = SessionStore(database, clock=lambda: 1_000.0)
|
|
|
|
assert store.status("existing-session", 2_000, idle_timeout_seconds=900) == "active"
|
|
with sqlite3.connect(database) as connection:
|
|
last_active_at = connection.execute(
|
|
"SELECT last_active_at FROM active_sessions WHERE session_hash = ?", (digest,)
|
|
).fetchone()[0]
|
|
assert last_active_at == 1_000
|
|
|
|
|
|
def test_identity_status_migrates_registry_that_already_has_idle_tracking(tmp_path):
|
|
database = tmp_path / "sessions.sqlite3"
|
|
digest = SessionStore._digest("existing-session")
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"CREATE TABLE active_sessions ("
|
|
"session_hash TEXT PRIMARY KEY, expires_at INTEGER NOT NULL, "
|
|
"management_id TEXT, device_label TEXT, created_at INTEGER, last_active_at INTEGER"
|
|
")"
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO active_sessions VALUES (?, ?, ?, ?, ?, ?)",
|
|
(digest, 2_000, "legacy-device", "Existing device", 900, 1_000),
|
|
)
|
|
|
|
status = SessionStore(database, clock=lambda: 1_000.0).status(
|
|
"existing-session", 2_000, idle_timeout_seconds=900
|
|
)
|
|
|
|
assert status == "active"
|
|
assert status.principal_id is None
|
|
assert status.principal_login is None
|
|
|
|
|
|
def test_existing_session_can_mint_first_step_up_grant_during_schema_upgrade(tmp_path):
|
|
database = tmp_path / "sessions.sqlite3"
|
|
digest = SessionStore._digest("existing-session")
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"CREATE TABLE active_sessions (session_hash TEXT PRIMARY KEY, expires_at INTEGER NOT NULL)"
|
|
)
|
|
connection.execute("INSERT INTO active_sessions VALUES (?, ?)", (digest, 2_000))
|
|
|
|
store = SessionStore(database, clock=lambda: 1_000.0)
|
|
grant = store.mint_step_up(
|
|
"existing-session", action="revoke_all_sessions", target="all", ttl_seconds=90
|
|
)
|
|
|
|
assert store.consume_step_up(
|
|
grant,
|
|
"existing-session",
|
|
action="revoke_all_sessions",
|
|
target="all",
|
|
) is True
|
|
|
|
|
|
def test_step_up_grant_is_single_use_and_bound_to_session_action_and_target(tmp_path):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("phone-session", 2_000)
|
|
|
|
grant = store.mint_step_up(
|
|
"phone-session", action="merge_pull", target="stackchain/api#7", ttl_seconds=90
|
|
)
|
|
|
|
assert store.consume_step_up(
|
|
grant, "phone-session", action="merge_pull", target="stackchain/api#8"
|
|
) is False
|
|
assert store.consume_step_up(
|
|
grant, "other-session", action="merge_pull", target="stackchain/api#7"
|
|
) is False
|
|
assert store.consume_step_up(
|
|
grant, "phone-session", action="close_issue", target="stackchain/api#7"
|
|
) is False
|
|
assert store.consume_step_up(
|
|
grant, "phone-session", action="merge_pull", target="stackchain/api#7"
|
|
) is True
|
|
assert store.consume_step_up(
|
|
grant, "phone-session", action="merge_pull", target="stackchain/api#7"
|
|
) is False
|
|
assert grant.encode() not in store.path.read_bytes()
|
|
|
|
|
|
def test_step_up_grants_expire_and_are_removed_with_parent_session(tmp_path):
|
|
now = [1_000.0]
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
|
store.activate("phone-session", 2_000)
|
|
expired = store.mint_step_up(
|
|
"phone-session", action="close_issue", target="stackchain/api#7", ttl_seconds=90
|
|
)
|
|
revoked = store.mint_step_up(
|
|
"phone-session", action="revoke_all_sessions", target="all", ttl_seconds=90
|
|
)
|
|
|
|
now[0] = 1_091.0
|
|
assert store.consume_step_up(
|
|
expired, "phone-session", action="close_issue", target="stackchain/api#7"
|
|
) is False
|
|
|
|
now[0] = 1_010.0
|
|
store.revoke("phone-session")
|
|
assert store.consume_step_up(
|
|
revoked, "phone-session", action="revoke_all_sessions", target="all"
|
|
) is False
|
|
|
|
|
|
def test_managed_session_revocation_invalidates_its_outstanding_grants(tmp_path):
|
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
|
|
store.activate("phone-session", 2_000, device_label="Phone")
|
|
grant = store.mint_step_up(
|
|
"phone-session", action="merge_pull", target="stackchain/api#7", ttl_seconds=90
|
|
)
|
|
phone = store.list_active("phone-session")[0]
|
|
|
|
assert store.revoke_managed(phone.management_id) is True
|
|
assert store.consume_step_up(
|
|
grant,
|
|
"phone-session",
|
|
action="merge_pull",
|
|
target="stackchain/api#7",
|
|
) is False
|