feat(auth): implement login, jwt guard, and change password (#4)

* feat(auth): add schemas, jwt guard, and change password endpoint

* feat(api): register auth routes and protected test route

* test(auth): cover 401 matrix, login timing, and change password

app/deps.py

@@ -0,0 +1,45 @@
+from uuid import UUID
+import jwt
+from fastapi import Depends
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy.orm import Session
+from app.config import settings
+from app.db import get_db
+from app.models import User
+from app.errors import AppError
+
+bearer_scheme = HTTPBearer(auto_error=False)
+
+def get_current_user(
+    creds: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
+    db: Session = Depends(get_db)
+) -> User:
+    if creds is None or creds.scheme.lower() != "bearer":
+        raise AppError(
+            401, "AUTH_MISSING", 
+            "Authorization header with a Bearer token is required.",
+            headers={"WWW-Authenticate": "Bearer"}
+        )
+
+    try:
+        payload = jwt.decode(
+            creds.credentials, 
+            settings.JWT_SECRET_KEY,
+            algorithms=["HS256"], 
+            options={"require": ["exp", "sub"]}
+        )
+    except jwt.ExpiredSignatureError:
+        raise AppError(401, "TOKEN_EXPIRED", "Access token has expired. Log in again.")
+    except jwt.PyJWTError:
+        raise AppError(401, "TOKEN_INVALID", "Access token is malformed or has an invalid signature.")
+
+    try:
+        user_id = UUID(payload["sub"])
+    except (ValueError, KeyError):
+        raise AppError(401, "TOKEN_INVALID", "Token subject is not a valid user id.")
+
+    user = db.get(User, user_id)
+    if user is None or not user.is_active:
+        raise AppError(401, "TOKEN_INVALID", "Token does not map to an active user.")
+
+    return user

app/main.py

@@ -1,10 +1,12 @@
 import uuid
 import logging
-from fastapi import FastAPI, Request
+from fastapi import FastAPI, Request, Depends
 from fastapi.exceptions import RequestValidationError
 from starlette.exceptions import HTTPException as StarletteHTTPException
 from app.errors import AppError, error_response
-from app.routers import auth  # <-- IMPORT THE ROUTER
+from app.routers import auth
+from app.deps import get_current_user
+from app.models import User
 
 logger = logging.getLogger("uvicorn.error")
 
@@ -16,7 +18,6 @@ def create_app() -> FastAPI:
         redoc_url=None,
     )
 
-    # Register the router
     app.include_router(auth.router, prefix="/api/v1")
 
     @app.middleware("http")
@@ -62,6 +63,10 @@ def create_app() -> FastAPI:
     @app.get("/healthz", tags=["system"])
     def healthz() -> dict[str, str]:
         return {"status": "ok"}
+        
+    @app.get("/api/v1/protected", tags=["system"])
+    def protected_route(current_user: User = Depends(get_current_user)):
+        return {"message": "You are authenticated!", "user_id": str(current_user.id)}
 
     return app
 

app/routers/auth.py

@@ -3,15 +3,15 @@ from sqlalchemy.orm import Session
 from sqlalchemy.exc import IntegrityError
 from app.db import get_db
 from app.models import User
-from app.schemas import UserCreate, UserResponse
-from app.security import hash_password
+from app.schemas import UserCreate, UserResponse, UserLogin, TokenResponse, ChangePasswordRequest
+from app.security import hash_password, verify_password, create_access_token
 from app.errors import AppError
+from app.deps import get_current_user
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
 @router.post("/signup", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
 def signup(user_data: UserCreate, db: Session = Depends(get_db)):
-    # 1. Check if email exists (using lower() for case-insensitivity)
     existing_user = db.query(User).filter(User.email.ilike(user_data.email)).first()
     if existing_user:
         raise AppError(
@@ -20,7 +20,6 @@ def signup(user_data: UserCreate, db: Session = Depends(get_db)):
             message="An account with this email already exists.",
         )
 
-    # 2. Hash password and create user
     new_user = User(
         email=user_data.email,
         password_hash=hash_password(user_data.password)
@@ -31,7 +30,6 @@ def signup(user_data: UserCreate, db: Session = Depends(get_db)):
         db.commit()
         db.refresh(new_user)
     except IntegrityError:
-        # Catch race conditions where two requests hit at the exact same time
         db.rollback()
         raise AppError(
             status_code=409,
@@ -39,4 +37,35 @@ def signup(user_data: UserCreate, db: Session = Depends(get_db)):
             message="An account with this email already exists.",
         )
 
-    return new_user
+    return new_user
+
+@router.post("/login", response_model=TokenResponse)
+def login(credentials: UserLogin, db: Session = Depends(get_db)):
+    user = db.query(User).filter(User.email.ilike(credentials.email)).first()
+    
+    if not verify_password(credentials.password, user.password_hash if user else None):
+        raise AppError(401, "INVALID_CREDENTIALS", "Incorrect email or password.")
+        
+    if not user.is_active:
+        raise AppError(401, "INVALID_CREDENTIALS", "Incorrect email or password.")
+
+    token = create_access_token(user.id, user.role)
+    return {"access_token": token, "token_type": "bearer", "expires_in": 3600}
+
+@router.put("/change-password", status_code=status.HTTP_200_OK)
+def change_password(
+    data: ChangePasswordRequest,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    if not verify_password(data.current_password, current_user.password_hash):
+        raise AppError(
+            401,
+            "INVALID_CREDENTIALS",
+            "The current password provided is incorrect."
+        )
+    
+    current_user.password_hash = hash_password(data.new_password)
+    db.commit()
+    
+    return {"message": "Password updated successfully."}

app/schemas.py

@@ -16,9 +16,32 @@ class UserCreate(BaseModel):
             raise ValueError("must be at most 72 bytes when encoded as UTF-8")
         return v
 
+class UserLogin(BaseModel):
+    email: EmailStr
+    password: str
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    expires_in: int
+
 class UserResponse(BaseModel):
     model_config = ConfigDict(from_attributes=True)
     id: UUID
     email: EmailStr
     role: str
-    created_at: datetime
+    created_at: datetime
+
+class ChangePasswordRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    current_password: str
+    new_password: str = Field(min_length=8, max_length=128)
+
+    @field_validator("new_password")
+    @classmethod
+    def password_strength(cls, v: str) -> str:
+        if not any(c.isdigit() for c in v) or not any(c.isalpha() for c in v):
+            raise ValueError("must contain at least one letter and one digit")
+        if len(v.encode("utf-8")) > 72:
+            raise ValueError("must be at most 72 bytes when encoded as UTF-8")
+        return v

tests/test_auth.py

@@ -1,61 +1,132 @@
+from datetime import datetime, timedelta, timezone
+import jwt
+from app.config import settings
+
 def test_signup_valid(client):
-    response = client.post("/api/v1/auth/signup", json={
-        "email": "test@example.com",
-        "password": "ValidPass123"
-    })
+    response = client.post("/api/v1/auth/signup", json={"email": "test@example.com", "password": "ValidPass123"})
     assert response.status_code == 201
     data = response.json()
     assert data["email"] == "test@example.com"
     assert data["role"] == "owner"
     assert "password_hash" not in data
-    assert "id" in data
 
 def test_signup_duplicate_email_case_insensitive(client):
-    # First signup
-    client.post("/api/v1/auth/signup", json={
-        "email": "Test@Example.com",
-        "password": "ValidPass123"
-    })
-    # Second signup with different casing
-    response = client.post("/api/v1/auth/signup", json={
-        "email": "test@example.com",
-        "password": "ValidPass123"
-    })
+    client.post("/api/v1/auth/signup", json={"email": "Test@Example.com", "password": "ValidPass123"})
+    response = client.post("/api/v1/auth/signup", json={"email": "test@example.com", "password": "ValidPass123"})
     assert response.status_code == 409
     assert response.json()["error"]["code"] == "EMAIL_ALREADY_REGISTERED"
 
 def test_signup_weak_password(client):
-    response = client.post("/api/v1/auth/signup", json={
-        "email": "weak@example.com",
-        "password": "abc"
-    })
+    response = client.post("/api/v1/auth/signup", json={"email": "weak@example.com", "password": "abc"})
     assert response.status_code == 422
-    data = response.json()
-    assert data["error"]["code"] == "VALIDATION_ERROR"
-    # Check that the field is mapped correctly
-    fields = [d["field"] for d in data["error"]["details"]]
+    fields = [d["field"] for d in response.json()["error"]["details"]]
     assert "body.password" in fields
 
 def test_signup_password_over_72_bytes(client):
-    # 40 emojis * 4 bytes = 160 bytes
-    emoji_password = "😀" * 40
-    response = client.post("/api/v1/auth/signup", json={
-        "email": "emoji@example.com",
-        "password": emoji_password
-    })
+    response = client.post("/api/v1/auth/signup", json={"email": "emoji@example.com", "password": "😀" * 40})
     assert response.status_code == 422
-    data = response.json()
     fields = [d["field"] for d in response.json()["error"]["details"]]
     assert "body.password" in fields
 
 def test_signup_extra_field_forbidden(client):
-    response = client.post("/api/v1/auth/signup", json={
-        "email": "hacker@example.com",
-        "password": "ValidPass123",
-        "role": "admin"  # This should be forbidden!
-    })
+    response = client.post("/api/v1/auth/signup", json={"email": "hacker@example.com", "password": "ValidPass123", "role": "admin"})
     assert response.status_code == 422
-    data = response.json()
-    assert data["error"]["code"] == "VALIDATION_ERROR"
-    fields = [d["field"] for d in data["error"]["details"]]
-    assert "body.role" in fields
+    fields = [d["field"] for d in response.json()["error"]["details"]]
+    assert "body.role" in fields
+
+def test_login_valid(client):
+    client.post("/api/v1/auth/signup", json={"email": "login@example.com", "password": "ValidPass123"})
+    response = client.post("/api/v1/auth/login", json={"email": "login@example.com", "password": "ValidPass123"})
+    assert response.status_code == 200
+    assert "access_token" in response.json()
+
+def test_login_wrong_password_and_unknown_email_look_identical(client):
+    client.post("/api/v1/auth/signup", json={"email": "real@example.com", "password": "ValidPass123"})
+    r1 = client.post("/api/v1/auth/login", json={"email": "real@example.com", "password": "WrongPass123"})
+    r2 = client.post("/api/v1/auth/login", json={"email": "unknown@example.com", "password": "ValidPass123"})
+    
+    assert r1.status_code == 401
+    assert r2.status_code == 401
+    
+    # Compare the error fields, ignoring the request_id (which is unique per request)
+    err1 = r1.json()["error"]
+    err2 = r2.json()["error"]
+    
+    assert err1["code"] == err2["code"]
+    assert err1["message"] == err2["message"]
+    assert err1["status"] == err2["status"]
+    assert err1["details"] == err2["details"]
+def test_protected_route_no_header(client):
+    response = client.get("/api/v1/protected")
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "AUTH_MISSING"
+
+def test_protected_route_garbage_token(client):
+    response = client.get("/api/v1/protected", headers={"Authorization": "Bearer not.a.jwt"})
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "TOKEN_INVALID"
+
+def test_protected_route_expired_token(client):
+    client.post("/api/v1/auth/signup", json={"email": "expired@example.com", "password": "ValidPass123"})
+    login = client.post("/api/v1/auth/login", json={"email": "expired@example.com", "password": "ValidPass123"})
+    user_id = jwt.decode(login.json()["access_token"], options={"verify_signature": False})["sub"]
+    
+    now = datetime.now(timezone.utc)
+    expired_token = jwt.encode(
+        {"sub": user_id, "role": "owner", "iat": now - timedelta(hours=2), "exp": now - timedelta(hours=1)},
+        settings.JWT_SECRET_KEY, algorithm="HS256"
+    )
+    response = client.get("/api/v1/protected", headers={"Authorization": f"Bearer {expired_token}"})
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "TOKEN_EXPIRED"
+
+def test_protected_route_valid_token(client):
+    client.post("/api/v1/auth/signup", json={"email": "valid@example.com", "password": "ValidPass123"})
+    login = client.post("/api/v1/auth/login", json={"email": "valid@example.com", "password": "ValidPass123"})
+    response = client.get("/api/v1/protected", headers={"Authorization": f"Bearer {login.json()['access_token']}"})
+    assert response.status_code == 200
+
+def test_change_password_success_and_login(client):
+    client.post("/api/v1/auth/signup", json={"email": "change@example.com", "password": "OldPass123"})
+    login = client.post("/api/v1/auth/login", json={"email": "change@example.com", "password": "OldPass123"})
+    token = login.json()["access_token"]
+
+    response = client.put(
+        "/api/v1/auth/change-password",
+        headers={"Authorization": f"Bearer {token}"},
+        json={"current_password": "OldPass123", "new_password": "NewPass456"}
+    )
+    assert response.status_code == 200
+
+    old_login = client.post("/api/v1/auth/login", json={"email": "change@example.com", "password": "OldPass123"})
+    assert old_login.status_code == 401
+
+    new_login = client.post("/api/v1/auth/login", json={"email": "change@example.com", "password": "NewPass456"})
+    assert new_login.status_code == 200
+
+def test_change_password_wrong_current_password(client):
+    client.post("/api/v1/auth/signup", json={"email": "wrongcur@example.com", "password": "OldPass123"})
+    login = client.post("/api/v1/auth/login", json={"email": "wrongcur@example.com", "password": "OldPass123"})
+    response = client.put(
+        "/api/v1/auth/change-password",
+        headers={"Authorization": f"Bearer {login.json()['access_token']}"},
+        json={"current_password": "WrongOldPass", "new_password": "NewPass456"}
+    )
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "INVALID_CREDENTIALS"
+
+def test_change_password_requires_auth(client):
+    response = client.put("/api/v1/auth/change-password", json={"current_password": "OldPass123", "new_password": "NewPass456"})
+    assert response.status_code == 401
+    assert response.json()["error"]["code"] == "AUTH_MISSING"
+
+def test_change_password_weak_new_password(client):
+    client.post("/api/v1/auth/signup", json={"email": "weaknew@example.com", "password": "OldPass123"})
+    login = client.post("/api/v1/auth/login", json={"email": "weaknew@example.com", "password": "OldPass123"})
+    response = client.put(
+        "/api/v1/auth/change-password",
+        headers={"Authorization": f"Bearer {login.json()['access_token']}"},
+        json={"current_password": "OldPass123", "new_password": "abc"}
+    )
+    assert response.status_code == 422
+    assert "body.new_password" in [d["field"] for d in response.json()["error"]["details"]]