152 lines
4.5 KiB
Python
152 lines
4.5 KiB
Python
"""Authentication tests: registration, login, sessions, throttling, CSRF."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api import app
|
|
|
|
PASSWORD = "correct horse battery staple"
|
|
|
|
|
|
def new_client() -> TestClient:
|
|
return TestClient(app)
|
|
|
|
|
|
def register(client: TestClient, username: str = "alice", password: str = PASSWORD):
|
|
return client.post(
|
|
"/api/v1/auth/register", json={"username": username, "password": password}
|
|
)
|
|
|
|
|
|
def login(client: TestClient, username: str = "alice", password: str = PASSWORD):
|
|
return client.post(
|
|
"/api/v1/auth/login", json={"username": username, "password": password}
|
|
)
|
|
|
|
|
|
def test_register_returns_user_and_authenticates():
|
|
client = new_client()
|
|
resp = register(client)
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["user"]["username"] == "alice"
|
|
assert "password" not in body["user"]
|
|
assert client.cookies.get("sc_session")
|
|
|
|
me = client.get("/api/v1/auth/me")
|
|
assert me.status_code == 200
|
|
assert me.json()["user"]["username"] == "alice"
|
|
|
|
|
|
def test_register_duplicate_username_is_conflict_case_insensitive():
|
|
client = new_client()
|
|
assert register(client, "Alice").status_code == 201
|
|
assert register(new_client(), "alice").status_code == 409
|
|
|
|
|
|
def test_register_rejects_weak_password():
|
|
resp = register(new_client(), password="short")
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_register_rejects_invalid_username():
|
|
for username in ["ab", "has space", "bad!", "x" * 33]:
|
|
assert register(new_client(), username).status_code == 422
|
|
|
|
|
|
def test_me_requires_authentication():
|
|
resp = new_client().get("/api/v1/auth/me")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_login_is_case_insensitive_and_works_after_logout():
|
|
client = new_client()
|
|
assert register(client, "alice").status_code == 201
|
|
assert client.post("/api/v1/auth/logout").status_code == 204
|
|
assert client.get("/api/v1/auth/me").status_code == 401
|
|
|
|
resp = login(client, "ALICE")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["user"]["username"] == "alice"
|
|
assert client.get("/api/v1/auth/me").status_code == 200
|
|
|
|
|
|
def test_wrong_password_and_unknown_user_are_indistinguishable():
|
|
client = new_client()
|
|
assert register(client, "alice").status_code == 201
|
|
|
|
wrong = login(new_client(), "alice", "definitely wrong")
|
|
unknown = login(new_client(), "nobody", "definitely wrong")
|
|
|
|
assert wrong.status_code == unknown.status_code == 401
|
|
assert wrong.json()["detail"] == unknown.json()["detail"]
|
|
|
|
|
|
def test_logout_invalidates_session_token():
|
|
client = new_client()
|
|
assert register(client).status_code == 201
|
|
token = client.cookies.get("sc_session")
|
|
|
|
assert client.post("/api/v1/auth/logout").status_code == 204
|
|
|
|
# Replay the old cookie on a fresh client: it must be rejected.
|
|
replay = new_client()
|
|
replay.cookies.set("sc_session", token)
|
|
assert replay.get("/api/v1/auth/me").status_code == 401
|
|
|
|
|
|
def test_login_is_rate_limited():
|
|
client = new_client()
|
|
assert register(client, "alice").status_code == 201
|
|
|
|
for _ in range(5):
|
|
assert login(new_client(), "alice", "wrong password").status_code == 401
|
|
|
|
blocked = login(new_client(), "alice", PASSWORD)
|
|
assert blocked.status_code == 429
|
|
assert int(blocked.headers["Retry-After"]) > 0
|
|
|
|
|
|
def test_cross_origin_mutation_is_rejected():
|
|
client = new_client()
|
|
resp = client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "alice", "password": PASSWORD},
|
|
headers={"Origin": "http://evil.example"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_same_origin_mutation_is_allowed():
|
|
client = new_client()
|
|
resp = client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "alice", "password": PASSWORD},
|
|
headers={"Origin": "http://testserver"},
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
|
|
def test_same_origin_with_forwarded_port_is_allowed():
|
|
client = new_client()
|
|
resp = client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "alice", "password": PASSWORD},
|
|
headers={
|
|
"Origin": "http://testserver:8080",
|
|
"X-Forwarded-Host": "testserver:8080",
|
|
},
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
|
|
def test_forwarded_host_without_port_is_rejected():
|
|
client = new_client()
|
|
resp = client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "alice", "password": PASSWORD},
|
|
headers={
|
|
"Origin": "http://testserver:8080",
|
|
"X-Forwarded-Host": "testserver",
|
|
},
|
|
)
|
|
assert resp.status_code == 403
|