159 lines
6.0 KiB
Python
159 lines
6.0 KiB
Python
import hashlib
|
|
import os
|
|
|
|
from app import app
|
|
from auth import hash_password
|
|
from tests.support import ApiTestCase, db_execute
|
|
|
|
|
|
class AuthTests(ApiTestCase):
|
|
def session_token(self, client=None) -> str:
|
|
return (
|
|
(client or self.client)
|
|
.get_cookie("sid", domain="localhost", path="/api")
|
|
.value
|
|
)
|
|
|
|
def test_register_logs_in_and_me_returns_user(self):
|
|
response = self.register("Alice")
|
|
self.assertEqual(response.get_json(), {"id": 1, "username": "Alice"})
|
|
me = self.call("GET", "/api/auth/me")
|
|
self.assertEqual(me.status_code, 200)
|
|
self.assertEqual(me.get_json(), {"id": 1, "username": "Alice"})
|
|
|
|
def test_session_cookie_flags(self):
|
|
cookie = self.register().headers["Set-Cookie"]
|
|
for flag in ("HttpOnly", "Secure", "SameSite=Lax", "Path=/api"):
|
|
self.assertIn(flag, cookie)
|
|
|
|
def test_session_token_is_stored_hashed(self):
|
|
self.register()
|
|
digest = hashlib.sha256(self.session_token().encode()).digest()
|
|
self.assertEqual(
|
|
db_execute(
|
|
"SELECT count(*) FROM sessions WHERE token_hash = %s", (digest,)
|
|
)[0][0],
|
|
1,
|
|
)
|
|
|
|
def test_duplicate_username_is_case_insensitive(self):
|
|
self.register("Alice")
|
|
response = self.call(
|
|
"POST",
|
|
"/api/auth/register",
|
|
{"username": "alice", "password": "another long password"},
|
|
)
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertEqual(response.get_json()["field"], "username")
|
|
|
|
def test_register_validation(self):
|
|
cases = [
|
|
({"username": "al", "password": "long enough password"}, "username"),
|
|
({"username": "al ice", "password": "long enough password"}, "username"),
|
|
({"username": "alice", "password": "short"}, "password"),
|
|
({"username": "alice", "password": "x" * 1025}, "password"),
|
|
({"username": "alice", "password": "\ud800" * 12}, "password"),
|
|
(
|
|
{
|
|
"username": "alice",
|
|
"password": "long enough password",
|
|
"is_admin": True,
|
|
},
|
|
"is_admin",
|
|
),
|
|
]
|
|
for payload, field in cases:
|
|
with self.subTest(payload=payload):
|
|
response = self.call("POST", "/api/auth/register", payload)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertEqual(response.get_json()["field"], field)
|
|
|
|
def test_login_is_case_insensitive_on_username(self):
|
|
self.register("Alice")
|
|
self.client = app.test_client()
|
|
response = self.call(
|
|
"POST",
|
|
"/api/auth/login",
|
|
{"username": "ALICE", "password": "correct horse battery"},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.get_json(), {"id": 1, "username": "Alice"})
|
|
|
|
def test_wrong_password_and_unknown_user_are_indistinguishable(self):
|
|
self.register()
|
|
wrong = self.call(
|
|
"POST",
|
|
"/api/auth/login",
|
|
{"username": "alice", "password": "not the password"},
|
|
)
|
|
unknown = self.call(
|
|
"POST",
|
|
"/api/auth/login",
|
|
{"username": "nobody", "password": "not the password"},
|
|
)
|
|
self.assertEqual(wrong.status_code, 401)
|
|
self.assertEqual(unknown.status_code, 401)
|
|
self.assertEqual(wrong.get_json(), unknown.get_json())
|
|
self.assertEqual(wrong.get_json(), {"error": "Invalid username or password"})
|
|
|
|
def test_login_with_malformed_username_is_401_not_500(self):
|
|
response = self.call(
|
|
"POST",
|
|
"/api/auth/login",
|
|
{"username": "a\x00b", "password": "whatever password"},
|
|
)
|
|
self.assertEqual(response.status_code, 401)
|
|
|
|
def test_me_without_session_is_401(self):
|
|
response = self.call("GET", "/api/auth/me")
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertEqual(
|
|
response.get_json(),
|
|
{"error": "Not signed in or session expired; log in again"},
|
|
)
|
|
|
|
def test_logout_revokes_token_server_side(self):
|
|
self.register()
|
|
token = self.session_token()
|
|
self.assertEqual(self.call("POST", "/api/auth/logout").status_code, 204)
|
|
self.client.set_cookie("sid", token, domain="localhost", path="/api")
|
|
self.assertEqual(self.call("GET", "/api/auth/me").status_code, 401)
|
|
|
|
def test_idle_session_expires(self):
|
|
self.register()
|
|
db_execute("UPDATE sessions SET expires_at = now() - interval '1 second'")
|
|
self.assertEqual(self.call("GET", "/api/auth/me").status_code, 401)
|
|
|
|
def test_absolute_cap_applies_even_inside_idle_window(self):
|
|
self.register()
|
|
db_execute(
|
|
"UPDATE sessions SET created_at = now() - interval '12 hours 1 second'"
|
|
)
|
|
self.assertEqual(self.call("GET", "/api/auth/me").status_code, 401)
|
|
|
|
def test_activity_slides_idle_expiry(self):
|
|
self.register()
|
|
db_execute("UPDATE sessions SET expires_at = now() + interval '1 minute'")
|
|
self.assertEqual(self.call("GET", "/api/auth/me").status_code, 200)
|
|
remaining = db_execute(
|
|
"SELECT expires_at - now() > interval '29 minutes' FROM sessions"
|
|
)[0][0]
|
|
self.assertTrue(remaining)
|
|
|
|
def test_old_hash_is_upgraded_on_login(self):
|
|
self.register()
|
|
pepper = os.environ["PASSWORD_PEPPER"].encode()
|
|
old = hash_password("correct horse battery", pepper, iterations=1000)
|
|
db_execute("UPDATE users SET password_hash = %s", (old,))
|
|
response = self.call(
|
|
"POST",
|
|
"/api/auth/login",
|
|
{"username": "alice", "password": "correct horse battery"},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertTrue(
|
|
db_execute("SELECT password_hash FROM users")[0][0].startswith(
|
|
"pbkdf2_sha256$600000$"
|
|
)
|
|
)
|