youtube-automate/tests/test_auth.py

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

import time

from youtube_automate.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 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)