94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.audit.service import write_event
|
||
from app.auth.schemas import Credentials, TokenResponse, UserResponse
|
||
from app.core.deps import get_current_user
|
||
from app.core.jwt import create_access_token
|
||
from app.core.passwords import (
|
||
WeakPasswordError,
|
||
hash_password,
|
||
validate_policy,
|
||
verify_password,
|
||
)
|
||
from app.db import get_db
|
||
from app.models import User
|
||
|
||
router = APIRouter(prefix="/auth", tags=["auth"])
|
||
|
||
|
||
def _client_info(request: Request) -> tuple[str | None, str | None]:
|
||
ip = request.client.host if request.client else None
|
||
return ip, request.headers.get("user-agent")
|
||
|
||
|
||
# POST /auth/register – принимает JSON `{email, password}`
|
||
@router.post(
|
||
"/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED
|
||
)
|
||
def register(
|
||
payload: Credentials, request: Request, db: Session = Depends(get_db)
|
||
) -> TokenResponse:
|
||
# проверить, что email ещё не занят;
|
||
if db.scalar(select(User).where(User.email == payload.email)) is not None:
|
||
raise HTTPException(status_code=409, detail="Email already registered")
|
||
|
||
# валидировать пароль через validate_policy
|
||
try:
|
||
validate_policy(payload.password, user_inputs=[payload.email])
|
||
except WeakPasswordError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
|
||
# создать пользователя с hashed_password
|
||
user = User(email=payload.email, hashed_password=hash_password(payload.password))
|
||
db.add(user)
|
||
db.commit()
|
||
db.refresh(user)
|
||
|
||
ip, user_agent = _client_info(request)
|
||
# записать событие аудита `user.created`;
|
||
write_event(db, "user.created", actor_id=user.id, ip=ip, user_agent=user_agent)
|
||
return TokenResponse(access_token=create_access_token(str(user.id)))
|
||
|
||
|
||
# POST /auth/login – принимает {email, password}
|
||
@router.post("/login", response_model=TokenResponse)
|
||
def login(
|
||
payload: Credentials, request: Request, db: Session = Depends(get_db)
|
||
) -> TokenResponse:
|
||
ip, user_agent = _client_info(request)
|
||
|
||
# найти пользователя по email;
|
||
user = db.scalar(select(User).where(User.email == payload.email))
|
||
|
||
# проверить пароль через verify_password;
|
||
if user is None or not verify_password(payload.password, user.hashed_password):
|
||
# при неудаче:
|
||
# записать user.login.failure (с email в metadata, но без пароля!)
|
||
# и вернуть 401 с нейтральным сообщением «Invalid credentials»
|
||
# (не раскрывать, что именно неверно – email или пароль);
|
||
write_event(
|
||
db,
|
||
"user.login.failure",
|
||
actor_id=user.id if user else None,
|
||
ip=ip,
|
||
user_agent=user_agent,
|
||
metadata={"email": payload.email},
|
||
)
|
||
raise HTTPException(status_code=401, detail="Invalid credentials")
|
||
|
||
# при успехе: записать user.login.success, вернуть токен.
|
||
write_event(
|
||
db, "user.login.success", actor_id=user.id, ip=ip, user_agent=user_agent
|
||
)
|
||
return TokenResponse(access_token=create_access_token(str(user.id)))
|
||
|
||
|
||
# GET /auth/me – защищённый:
|
||
# принимает зависимость get_current_user;
|
||
# возвращает {id, email, is_admin}.
|
||
@router.get("/me", response_model=UserResponse)
|
||
def me(user: User = Depends(get_current_user)) -> UserResponse:
|
||
return UserResponse(id=user.id, email=user.email, is_admin=user.is_admin)
|