|
# retoor <retoor@molodetz.nl>
|
|
|
|
import logging
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, APIRouter, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from devplacepy.database import get_table
|
|
from devplacepy.utils import verify_password_async, get_current_user
|
|
from devplacepy.models import LoginForm
|
|
from devplacepy.dependencies import json_or_form
|
|
from devplacepy.services.access_tokens import issue_token
|
|
from devplacepy.services.audit import record as audit
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/token")
|
|
async def token(
|
|
request: Request,
|
|
data: Annotated[LoginForm, Depends(json_or_form(LoginForm))],
|
|
):
|
|
"""Issue a DevPlace access token.
|
|
|
|
Accepts ``email`` + ``password`` (JSON or form-encoded). Returns a JSON
|
|
object with ``access_token``, ``token_type``, and ``expires_in`` on success,
|
|
or a ``401`` error on bad credentials.
|
|
"""
|
|
identifier = data.email.strip().lower()
|
|
password = data.password
|
|
|
|
if not identifier or not password:
|
|
return JSONResponse(
|
|
{"error": "email and password are required"},
|
|
status_code=400,
|
|
)
|
|
|
|
users = get_table("users")
|
|
user = users.find_one(email=identifier) or users.find_one(
|
|
username=identifier
|
|
)
|
|
|
|
if not user or not await verify_password_async(
|
|
password, user.get("password_hash", "")
|
|
):
|
|
audit.record(
|
|
request,
|
|
"auth.token.failure",
|
|
user=None,
|
|
actor_kind="guest",
|
|
result="failure",
|
|
metadata={"identifier": identifier},
|
|
summary=f"failed token request for {identifier}",
|
|
)
|
|
return JSONResponse(
|
|
{"error": "Invalid credentials"},
|
|
status_code=401,
|
|
)
|
|
|
|
if not user.get("is_active", True):
|
|
audit.record(
|
|
request,
|
|
"auth.token.failure",
|
|
user=None,
|
|
actor_kind="guest",
|
|
result="failure",
|
|
metadata={"identifier": identifier, "reason": "inactive"},
|
|
summary=f"token request for inactive user {identifier}",
|
|
)
|
|
return JSONResponse(
|
|
{"error": "Account is deactivated"},
|
|
status_code=403,
|
|
)
|
|
|
|
token = issue_token(user)
|
|
|
|
logger.info("Token issued for %s", user["username"])
|
|
audit.record(
|
|
request,
|
|
"auth.token.issued",
|
|
user=user,
|
|
target_type="user",
|
|
target_uid=user["uid"],
|
|
target_label=user["username"],
|
|
metadata={"token_uid": token["uid"], "expires_in": token["expires_in"]},
|
|
summary=f"access token issued for {user['username']}",
|
|
links=[audit.target("user", user["uid"], user["username"])],
|
|
)
|
|
|
|
return JSONResponse(
|
|
{
|
|
"access_token": token["access_token"],
|
|
"token_type": token["token_type"],
|
|
"expires_in": token["expires_in"],
|
|
}
|
|
)
|