Audit failed logins (NIST AU-2, source RFP C.5.7.39)
Only successful logins wrote an audit row. Every unsuccessful attempt now writes `login_failed` with reason bad_credentials, unknown_user, inactive, or throttled, plus the client IP. The attempted username is recorded only when it matches a real account (as the actor); unknown-user attempts store no username, since people sometimes type a password into that field. Found while preparing Phase 0 control evidence (milestone 0.9). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
0c359c926a
commit
99a50dc86c
2 files changed
+64
No files matched your search
@@ -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(
|
||||
|
||||
@@ -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"]
|
||||
Reference in new issue
Block a user