ytstream/tests/test_auth.py

"""Password hashing, session cookies, CSRF tokens and login throttling."""

import time

import pytest

from ytstream.web import auth


class TestPasswords:
    def test_round_trip(self):
        stored = auth.hash_password("correct horse battery staple")
        assert auth.verify_password(stored, "correct horse battery staple")

    def test_wrong_password_rejected(self):
        stored = auth.hash_password("secret")
        assert not auth.verify_password(stored, "Secret")
        assert not auth.verify_password(stored, "")

    def test_salt_makes_hashes_unique(self):
        assert auth.hash_password("same") != auth.hash_password("same")

    def test_hash_is_not_the_plaintext(self):
        assert "secret" not in auth.hash_password("secret")

    def test_empty_stored_hash_rejects_everything(self):
        assert not auth.verify_password("", "anything")

    def test_malformed_stored_hash_does_not_raise(self):
        for junk in ("nonsense", "scrypt$bad", "a$b$c$d$e$f", "scrypt$x$y$z$q$r"):
            assert auth.verify_password(junk, "anything") is False

    def test_unicode_password(self):
        stored = auth.hash_password("pässwörd🎬")
        assert auth.verify_password(stored, "pässwörd🎬")


class TestSessions:
    def test_issue_and_verify(self):
        secret = auth.new_secret()
        token = auth.issue_session(secret)
        assert auth.verify_session(secret, token)

    def test_a_different_secret_rejects(self):
        token = auth.issue_session(auth.new_secret())
        assert not auth.verify_session(auth.new_secret(), token)

    def test_tampered_payload_rejected(self):
        secret = auth.new_secret()
        token = auth.issue_session(secret)
        payload, signature = token.split(".", 1)
        assert not auth.verify_session(secret, f"{payload}x.{signature}")

    def test_tampered_signature_rejected(self):
        secret = auth.new_secret()
        payload, _ = auth.issue_session(secret).split(".", 1)
        assert not auth.verify_session(secret, f"{payload}.deadbeef")

    def test_garbage_rejected(self):
        secret = auth.new_secret()
        for junk in ("", "no-dot", "a.b.c", "...."):
            assert auth.verify_session(secret, junk) is False

    def test_expires_after_a_year(self):
        secret = auth.new_secret()
        issued = time.time() - auth.SESSION_MAX_AGE - 10
        token = auth.issue_session(secret, issued_at=issued)
        assert not auth.verify_session(secret, token)

    def test_still_valid_just_inside_a_year(self):
        secret = auth.new_secret()
        issued = time.time() - auth.SESSION_MAX_AGE + 60
        token = auth.issue_session(secret, issued_at=issued)
        assert auth.verify_session(secret, token)

    def test_a_token_from_the_future_is_rejected(self):
        secret = auth.new_secret()
        token = auth.issue_session(secret, issued_at=time.time() + 3600)
        assert not auth.verify_session(secret, token)


class TestCookie:
    def test_carries_the_hardening_flags(self):
        header = auth.cookie_header("abc")
        for flag in ("HttpOnly", "Secure", "SameSite=Lax", "Path=/"):
            assert flag in header
        assert f"Max-Age={auth.SESSION_MAX_AGE}" in header

    def test_secure_can_be_omitted_for_local_http_testing(self):
        assert "Secure" not in auth.cookie_header("abc", secure=False)

    def test_clear_cookie_expires_immediately(self):
        assert "Max-Age=0" in auth.clear_cookie_header()


class TestCookieParsing:
    """Regression tests for a real failure: http.cookies.SimpleCookie silently
    drops everything after a value it dislikes, which made valid sessions
    invisible and bounced users back to the login page with no error."""

    def test_finds_our_cookie_alone(self):
        assert auth.cookie_value("yta_session=abc") == "abc"

    def test_finds_it_after_a_neighbour(self):
        assert auth.cookie_value("sessionid=xyz; yta_session=abc") == "abc"

    def test_finds_it_before_a_neighbour(self):
        assert auth.cookie_value("yta_session=abc; sessionid=xyz") == "abc"

    @pytest.mark.parametrize(
        "neighbour",
        [
            'prefs={"a":1}',          # JSON value — what actually broke it
            "junk=[1,2,3]",
            "weird=a b c",
            "empty=",
            "novalue",
            "quoted=\"has spaces\"",
            "path=/a/b/c",
            "colons=a:b:c",
            "comma=a,b",
        ],
    )
    def test_survives_hostile_neighbours(self, neighbour):
        assert auth.cookie_value(f"{neighbour}; yta_session=abc") == "abc"
        assert auth.cookie_value(f"yta_session=abc; {neighbour}") == "abc"

    def test_strips_surrounding_quotes(self):
        assert auth.cookie_value('yta_session="abc"') == "abc"

    def test_tolerates_whitespace(self):
        assert auth.cookie_value("  yta_session = abc  ") == "abc"

    def test_absent_cookie_returns_empty(self):
        assert auth.cookie_value("sessionid=xyz") == ""

    def test_empty_header_returns_empty(self):
        assert auth.cookie_value("") == ""
        assert auth.cookie_value(None) == ""

    def test_does_not_match_a_name_that_merely_contains_ours(self):
        assert auth.cookie_value("not_yta_session=nope") == ""

    def test_real_token_round_trips_through_the_header(self):
        secret = auth.new_secret()
        token = auth.issue_session(secret)
        header = f'prefs={{"theme":"dark"}}; yta_session={token}; other=1'
        assert auth.verify_session(secret, auth.cookie_value(header))


class TestCsrf:
    def test_token_verifies(self):
        secret, session = auth.new_secret(), auth.issue_session(auth.new_secret())
        token = auth.csrf_token(secret, session)
        assert auth.verify_csrf(secret, session, token)

    def test_token_is_bound_to_the_session(self):
        secret = auth.new_secret()
        one = auth.issue_session(secret, issued_at=1000)
        two = auth.issue_session(secret, issued_at=2000)
        assert not auth.verify_csrf(secret, two, auth.csrf_token(secret, one))

    def test_empty_token_rejected(self):
        secret, session = auth.new_secret(), "sess"
        assert not auth.verify_csrf(secret, session, "")

    def test_wrong_token_rejected(self):
        secret, session = auth.new_secret(), "sess"
        assert not auth.verify_csrf(secret, session, "deadbeef")


class TestThrottle:
    def test_allows_up_to_the_limit(self):
        throttle = auth.LoginThrottle(max_failures=3, lockout=60)
        for _ in range(2):
            throttle.record_failure("1.2.3.4", now=1000)
        assert not throttle.locked("1.2.3.4", now=1000)

    def test_locks_after_the_limit(self):
        throttle = auth.LoginThrottle(max_failures=3, lockout=60)
        for _ in range(3):
            throttle.record_failure("1.2.3.4", now=1000)
        assert throttle.locked("1.2.3.4", now=1000)

    def test_lock_expires(self):
        throttle = auth.LoginThrottle(max_failures=3, lockout=60)
        for _ in range(3):
            throttle.record_failure("1.2.3.4", now=1000)
        assert not throttle.locked("1.2.3.4", now=1061)

    def test_success_clears_the_counter(self):
        throttle = auth.LoginThrottle(max_failures=3, lockout=60)
        for _ in range(2):
            throttle.record_failure("1.2.3.4", now=1000)
        throttle.record_success("1.2.3.4")
        assert not throttle.locked("1.2.3.4", now=1000)

    def test_addresses_are_tracked_separately(self):
        throttle = auth.LoginThrottle(max_failures=2, lockout=60)
        for _ in range(2):
            throttle.record_failure("1.1.1.1", now=1000)
        assert throttle.locked("1.1.1.1", now=1000)
        assert not throttle.locked("2.2.2.2", now=1000)