stackchain-dashboard/tests/test_login_attempt_store.py

import sqlite3

from src.login_attempt_store import LoginAttemptStore, client_source


def test_forwarded_client_is_used_only_for_explicitly_trusted_proxies():
    trusted = "127.0.0.0/8, 10.0.0.0/8"

    assert client_source("198.51.100.4", "203.0.113.7", trusted) == "198.51.100.4"
    assert (
        client_source("127.0.0.1", "203.0.113.7, 10.1.2.3", trusted)
        == "203.0.113.7"
    )
    assert client_source("127.0.0.1", "not-an-ip", trusted) == "127.0.0.1"


def test_failure_budget_is_shared_across_store_instances_and_expires(tmp_path):
    now = [1_000.0]
    database = tmp_path / "login-attempts.sqlite3"
    first = LoginAttemptStore(
        database,
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
    )
    second = LoginAttemptStore(
        database,
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
    )

    assert first.retry_after("203.0.113.7") == 0
    first.record_failure("203.0.113.7")
    second.record_failure("203.0.113.7")
    first.record_failure("203.0.113.7")

    assert second.retry_after("203.0.113.7") == 60
    now[0] = 1_060.0
    assert first.retry_after("203.0.113.7") == 0


def test_recording_a_failure_prunes_expired_source_records(tmp_path):
    now = [1_000.0]
    store = LoginAttemptStore(
        tmp_path / "login-attempts.sqlite3",
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
    )
    store.record_failure("203.0.113.1")
    store.record_failure("203.0.113.2")

    now[0] = 1_060.0
    store.record_failure("203.0.113.3")

    with sqlite3.connect(store.path) as connection:
        assert connection.execute("SELECT COUNT(*) FROM login_attempts").fetchone() == (1,)


def test_failure_ledger_evicts_oldest_sources_at_its_size_limit(tmp_path):
    store = LoginAttemptStore(
        tmp_path / "login-attempts.sqlite3",
        clock=lambda: 1_000.0,
        max_failures=3,
        window_seconds=60,
        max_entries=2,
    )

    for address in ("203.0.113.1", "203.0.113.2", "203.0.113.3"):
        store.record_failure(address)

    with sqlite3.connect(store.path) as connection:
        assert connection.execute("SELECT COUNT(*) FROM login_attempts").fetchone() == (2,)


def test_failed_sign_ins_are_aggregated_across_workers_and_survive_throttle_clear(tmp_path):
    now = [1_000.0]
    database = tmp_path / "login-attempts.sqlite3"
    first = LoginAttemptStore(
        database,
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
        alert_bucket_seconds=300,
    )
    second = LoginAttemptStore(
        database,
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
        alert_bucket_seconds=300,
    )

    first.record_failure("203.0.113.7", method="token")
    now[0] = 1_010.0
    second.record_failure("198.51.100.4", method="token")
    first.clear("203.0.113.7")

    assert second.list_alerts() == [
        {
            "method": "token",
            "failed_count": 2,
            "blocked_count": 0,
            "first_at": 1_000,
            "last_at": 1_010,
        }
    ]


def test_rate_blocked_sign_ins_increment_a_distinct_privacy_safe_count(tmp_path):
    store = LoginAttemptStore(
        tmp_path / "login-attempts.sqlite3",
        clock=lambda: 1_000.0,
        max_failures=3,
        window_seconds=60,
    )

    store.record_blocked(method="token")

    assert store.list_alerts() == [
        {
            "method": "token",
            "failed_count": 0,
            "blocked_count": 1,
            "first_at": 1_000,
            "last_at": 1_000,
        }
    ]
    with sqlite3.connect(store.path) as connection:
        columns = [row[1] for row in connection.execute("PRAGMA table_info(login_alerts)")]
    assert "source_hash" not in columns


def test_expired_sign_in_alert_buckets_are_not_returned(tmp_path):
    now = [1_000.0]
    store = LoginAttemptStore(
        tmp_path / "login-attempts.sqlite3",
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
        alert_retention_seconds=300,
    )
    store.record_failure("203.0.113.7")

    now[0] = 1_301.0

    assert store.list_alerts() == []


def test_sign_in_alert_bucket_count_is_bounded_during_rotating_source_floods(tmp_path):
    now = [1_000.0]
    store = LoginAttemptStore(
        tmp_path / "login-attempts.sqlite3",
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
        alert_bucket_seconds=60,
        max_alert_buckets=2,
    )
    for index, timestamp in enumerate((1_000.0, 1_060.0, 1_120.0)):
        now[0] = timestamp
        store.record_failure(f"203.0.113.{index}")

    with sqlite3.connect(store.path) as connection:
        count = connection.execute("SELECT COUNT(*) FROM login_alerts").fetchone()[0]

    assert count == 2


def test_named_admission_budget_is_atomic_across_instances_and_source_scoped(tmp_path):
    now = [1_000.0]
    database = tmp_path / "login-attempts.sqlite3"
    first = LoginAttemptStore(
        database,
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
    )
    second = LoginAttemptStore(
        database,
        clock=lambda: now[0],
        max_failures=3,
        window_seconds=60,
    )

    assert first.admit("passkey_options", "203.0.113.7", limit=2) == 0
    assert second.admit("passkey_options", "203.0.113.7", limit=2) == 0
    assert first.admit("passkey_options", "203.0.113.7", limit=2) == 60
    assert first.admit("passkey_options", "203.0.113.8", limit=2) == 0

    now[0] = 1_060.0
    assert second.admit("passkey_options", "203.0.113.7", limit=2) == 0