mirror of
https://github.com/yogeshojha/rengine.git
synced 2026-10-20 23:31:31 +02:00
Yogesh Ojha
69239333bc
- Added `clean_name` function to `validation.py` for name validation with max length. - Simplified docstrings in `dnsx` client, parser, and service files for clarity. - Enhanced error handling in `ViewDNSService` to prevent disabling keys on rate limits. - Improved WHOIS service to handle concurrent inserts gracefully with integrity checks. - Updated `Settings` class in worker config to inherit from `BaseAppSettings`. - Added new dependencies in `pyproject.toml` for `apprise` and `cryptography`.
122 lines
3.4 KiB
Python
122 lines
3.4 KiB
Python
from typing import Annotated
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlmodel import select
|
|
|
|
from app.core.database import get_session
|
|
from app.core.ratelimit import is_token_revoked
|
|
from app.core.security import TOKEN_TYPE_ACCESS, decode_token
|
|
from shared.models.user import User
|
|
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
BEARER_HEADERS = {"WWW-Authenticate": "Bearer"}
|
|
|
|
|
|
async def get_token_from_request(
|
|
request: Request,
|
|
credentials: Annotated[
|
|
HTTPAuthorizationCredentials | None, Depends(security)
|
|
] = None,
|
|
) -> str:
|
|
if credentials:
|
|
return credentials.credentials
|
|
|
|
from app.api.v1.auth import ACCESS_TOKEN_COOKIE # noqa: PLC0415
|
|
|
|
token = request.cookies.get(ACCESS_TOKEN_COOKIE)
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
headers=BEARER_HEADERS,
|
|
)
|
|
|
|
return token
|
|
|
|
|
|
async def get_current_user(
|
|
token: Annotated[str, Depends(get_token_from_request)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
) -> User:
|
|
payload = decode_token(token)
|
|
if payload is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired token",
|
|
headers=BEARER_HEADERS,
|
|
)
|
|
|
|
if payload.get("type") != TOKEN_TYPE_ACCESS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid token type",
|
|
headers=BEARER_HEADERS,
|
|
)
|
|
|
|
jti = payload.get("jti")
|
|
if jti and await is_token_revoked(jti):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Token has been revoked",
|
|
headers=BEARER_HEADERS,
|
|
)
|
|
|
|
user_id_str = payload.get("sub")
|
|
if not user_id_str:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid token payload",
|
|
headers=BEARER_HEADERS,
|
|
)
|
|
|
|
try:
|
|
user_id = UUID(user_id_str)
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid user ID in token",
|
|
headers=BEARER_HEADERS,
|
|
) from e
|
|
|
|
result = await session.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="User not found",
|
|
headers=BEARER_HEADERS,
|
|
)
|
|
|
|
return user
|
|
|
|
|
|
async def get_current_active_user(
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
) -> User:
|
|
if not current_user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Inactive user",
|
|
)
|
|
return current_user
|
|
|
|
|
|
async def get_current_superuser(
|
|
current_user: Annotated[User, Depends(get_current_active_user)],
|
|
) -> User:
|
|
if not current_user.is_superuser:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Not enough privileges. Admin access required.",
|
|
)
|
|
return current_user
|
|
|
|
|
|
CurrentUser = Annotated[User, Depends(get_current_active_user)]
|
|
CurrentSuperuser = Annotated[User, Depends(get_current_superuser)]
|