158 lines
6.6 KiB
Python
158 lines
6.6 KiB
Python
"""Login: AR_AUTH=auto turns on with the first user; identity feeds sign-off fields."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api import auth as auth_mod
|
|
from app.api.main import app
|
|
from app.db import models
|
|
from app.db.database import SessionLocal, init_db
|
|
|
|
|
|
@pytest.fixture()
|
|
def clean_users():
|
|
"""Users flip AR_AUTH=auto to 'required' for the WHOLE shared test DB — always remove
|
|
them again so the rest of the suite keeps running unauthenticated."""
|
|
init_db()
|
|
yield
|
|
db = SessionLocal()
|
|
try:
|
|
db.query(models.User).delete()
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
auth_mod.invalidate_users_cache()
|
|
|
|
|
|
def _add_user(username: str, display: str, password: str) -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
db.add(models.User(username=username, display_name=display,
|
|
password_hash=auth_mod.hash_password(password), is_active=True))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
auth_mod.invalidate_users_cache()
|
|
|
|
|
|
def test_password_hash_roundtrip():
|
|
h = auth_mod.hash_password("s3cret-pw!")
|
|
assert h.startswith("scrypt$")
|
|
assert auth_mod.verify_password("s3cret-pw!", h)
|
|
assert not auth_mod.verify_password("wrong", h)
|
|
assert not auth_mod.verify_password("s3cret-pw!", "garbage")
|
|
|
|
|
|
def test_token_roundtrip_and_tamper():
|
|
user = models.User(id=7, username="jane", display_name="Jane D",
|
|
password_hash="x", is_active=True)
|
|
tok = auth_mod.create_token(user)
|
|
data = auth_mod.parse_token(tok)
|
|
assert data and data["uid"] == 7 and data["dn"] == "Jane D"
|
|
payload, sig = tok.split(".")
|
|
assert auth_mod.parse_token(f"{payload}x.{sig}") is None # tampered payload
|
|
assert auth_mod.parse_token(f"{payload}.{sig[:-2]}aa") is None # tampered signature
|
|
|
|
|
|
def test_api_open_with_no_users(clean_users):
|
|
auth_mod.invalidate_users_cache()
|
|
with TestClient(app) as c:
|
|
assert c.get("/api/auth/status").json()["auth_required"] is False
|
|
assert c.get("/api/sessions").status_code == 200
|
|
assert c.get("/api/auth/me").json()["authenticated"] is False
|
|
|
|
|
|
def test_first_user_turns_auth_on_and_login_works(clean_users):
|
|
_add_user("talha", "Talha Ahmed", "correct-horse-9")
|
|
with TestClient(app) as c:
|
|
assert c.get("/api/auth/status").json()["auth_required"] is True
|
|
# Locked out without a token; health stays open for probes.
|
|
assert c.get("/api/sessions").status_code == 401
|
|
assert c.get("/api/health").status_code == 200
|
|
|
|
assert c.post("/api/auth/login", json={
|
|
"username": "talha", "password": "nope"}).status_code == 401
|
|
assert c.post("/api/auth/login", json={
|
|
"username": "ghost", "password": "correct-horse-9"}).status_code == 401
|
|
|
|
r = c.post("/api/auth/login", json={"username": "TALHA", # case-insensitive
|
|
"password": "correct-horse-9"})
|
|
assert r.status_code == 200
|
|
token = r.json()["token"]
|
|
assert r.json()["user"]["display_name"] == "Talha Ahmed"
|
|
|
|
hdr = {"Authorization": f"Bearer {token}"}
|
|
assert c.get("/api/sessions", headers=hdr).status_code == 200
|
|
me = c.get("/api/auth/me", headers=hdr).json()
|
|
assert me["authenticated"] and me["display_name"] == "Talha Ahmed"
|
|
|
|
|
|
def test_signed_in_identity_overrides_body_name(clean_users):
|
|
"""Accountability fields record the VERIFIED identity, not whatever the body claims."""
|
|
_add_user("ayesha", "Ayesha K", "another-pw-123")
|
|
with TestClient(app) as c:
|
|
token = c.post("/api/auth/login", json={
|
|
"username": "ayesha", "password": "another-pw-123"}).json()["token"]
|
|
hdr = {"Authorization": f"Bearer {token}"}
|
|
|
|
sid = c.post("/api/sessions", json={"name": "identity", "month_end_date": "2029-01-31",
|
|
"allow_duplicate": True}, headers=hdr).json()["id"]
|
|
r = c.post(f"/api/sessions/{sid}/reconciliation-control/verify",
|
|
json={"verified_by": "Somebody Else", "comment": "spoof attempt"},
|
|
headers=hdr)
|
|
assert r.status_code == 200
|
|
db = SessionLocal()
|
|
try:
|
|
fc = db.query(models.FinanceControl).filter_by(session_id=sid).first()
|
|
assert fc.verified_by == "Ayesha K" # not "Somebody Else"
|
|
finally:
|
|
db.close()
|
|
c.delete(f"/api/sessions/{sid}", headers=hdr)
|
|
|
|
|
|
def test_change_password_flow(clean_users):
|
|
_add_user("changer", "Change Person", "old-password-1")
|
|
with TestClient(app) as c:
|
|
token = c.post("/api/auth/login", json={
|
|
"username": "changer", "password": "old-password-1"}).json()["token"]
|
|
hdr = {"Authorization": f"Bearer {token}"}
|
|
|
|
# Wrong current password / too short / unchanged are all refused.
|
|
assert c.post("/api/auth/change-password", headers=hdr, json={
|
|
"current_password": "nope", "new_password": "new-password-2"}).status_code == 400
|
|
assert c.post("/api/auth/change-password", headers=hdr, json={
|
|
"current_password": "old-password-1", "new_password": "short"}).status_code == 400
|
|
assert c.post("/api/auth/change-password", headers=hdr, json={
|
|
"current_password": "old-password-1",
|
|
"new_password": "old-password-1"}).status_code == 400
|
|
# Not signed in -> refused.
|
|
assert c.post("/api/auth/change-password", json={
|
|
"current_password": "old-password-1",
|
|
"new_password": "new-password-2"}).status_code == 401
|
|
|
|
r = c.post("/api/auth/change-password", headers=hdr, json={
|
|
"current_password": "old-password-1", "new_password": "new-password-2"})
|
|
assert r.status_code == 200 and r.json()["changed"] is True
|
|
|
|
# Old password dead, new one works, existing token still valid until expiry.
|
|
assert c.post("/api/auth/login", json={
|
|
"username": "changer", "password": "old-password-1"}).status_code == 401
|
|
assert c.post("/api/auth/login", json={
|
|
"username": "changer", "password": "new-password-2"}).status_code == 200
|
|
assert c.get("/api/auth/me", headers=hdr).status_code == 200
|
|
|
|
|
|
def test_inactive_user_cannot_login(clean_users):
|
|
_add_user("gone", "Gone Person", "some-pw-12345")
|
|
db = SessionLocal()
|
|
try:
|
|
db.query(models.User).filter_by(username="gone").first().is_active = False
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
auth_mod.invalidate_users_cache()
|
|
with TestClient(app) as c:
|
|
assert c.post("/api/auth/login", json={
|
|
"username": "gone", "password": "some-pw-12345"}).status_code == 401
|