diff --git a/src/ypms/auth/router.py b/src/ypms/auth/router.py index 3c10f57..9a4d9d4 100644 --- a/src/ypms/auth/router.py +++ b/src/ypms/auth/router.py @@ -62,6 +62,20 @@ def _issue_access_token(user: User) -> str: ) +def _audit_login_failure(db: Session, user: User | None, ip: str, reason: str) -> None: + """NIST AU-2 / source RFP C.5.7.39: record unsuccessful logon attempts. + + The attempted username is recorded only when it matches a real account + (as the audit actor). Unknown-user attempts store no username: people + sometimes type a password into the username field. + """ + audit.record( + db, actor=user, action="login_failed", entity="User", + entity_id=user.id if user else None, after={"reason": reason}, ip=ip, + ) + db.commit() + + @router.post("/login", response_model=LoginResponse) def login( request: Request, @@ -71,6 +85,7 @@ def login( ip = request.client.host if request.client else "unknown" if not login_limiter.allow(ip): # NIST AC-7: cap unsuccessful logon attempts. + _audit_login_failure(db, None, ip, "throttled") raise HTTPException( status.HTTP_429_TOO_MANY_REQUESTS, "too many login attempts; try again shortly", @@ -78,10 +93,12 @@ def login( ) user = db.query(User).filter(User.username == form.username).one_or_none() if not user or not verify_password(form.password, user.password_hash): + _audit_login_failure(db, user, ip, "bad_credentials" if user else "unknown_user") # Deliberately identical error surface so we don't leak whether the # username exists (IA-6 authenticator feedback discipline). raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials") if not user.is_active: + _audit_login_failure(db, user, ip, "inactive") raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user inactive") audit.record( diff --git a/tests/auth/test_router.py b/tests/auth/test_router.py index 60530f5..8cfe87f 100644 --- a/tests/auth/test_router.py +++ b/tests/auth/test_router.py @@ -251,3 +251,50 @@ def test_login_lockout_after_five_failed_attempts( ) assert r.status_code == 429 assert r.headers["Retry-After"] == "900" + + +# ---------------- failed-login audit (NIST AU-2, C.5.7.39) ---------------- + + +def _failures(db: Session) -> list[tuple[str | None, int | None, dict[str, str] | None]]: + from sqlalchemy import select + + from ypms.audit.models import AuditLog + + rows = db.execute(select(AuditLog).where(AuditLog.action == "login_failed")).scalars() + return [(r.actor_username, r.entity_id, r.after) for r in rows] + + +def test_bad_password_is_audited_against_the_account( + app_client: TestClient, db: Session, staff_user: User +) -> None: + app_client.post("/auth/login", data={"username": staff_user.username, "password": "nope"}) + assert _failures(db) == [(staff_user.username, staff_user.id, {"reason": "bad_credentials"})] + + +def test_unknown_user_is_audited_without_the_typed_username( + app_client: TestClient, db: Session +) -> None: + app_client.post("/auth/login", data={"username": "my-actual-password!", "password": "x"}) + assert _failures(db) == [(None, None, {"reason": "unknown_user"})] + + +def test_inactive_user_login_is_audited( + app_client: TestClient, db: Session, staff_user: User +) -> None: + staff_user.is_active = False + db.flush() + r = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + assert r.status_code == 401 + assert _failures(db) == [(staff_user.username, staff_user.id, {"reason": "inactive"})] + + +def test_throttled_login_is_audited(app_client: TestClient, db: Session) -> None: + for _ in range(6): + r = app_client.post("/auth/login", data={"username": "ghost", "password": "x"}) + assert r.status_code == 429 + reasons = [after["reason"] for _, _, after in _failures(db) if after] + assert reasons == ["unknown_user"] * 5 + ["throttled"]