"""Password hashing, session cookies, CSRF tokens and login throttling.""" import time import pytest 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 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)