mirror of
https://code.lenaisten.de/Lenaisten/advent22.git
synced 2024-12-24 13:33:00 +00:00
module routers._security for user/admin stuff
This commit is contained in:
parent
29f02d2545
commit
973f7546b6
3 changed files with 66 additions and 43 deletions
47
api/advent22_api/routers/_security.py
Normal file
47
api/advent22_api/routers/_security.py
Normal file
|
@ -0,0 +1,47 @@
|
|||
import secrets
|
||||
from datetime import date
|
||||
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||||
|
||||
from ..core.config import Config
|
||||
|
||||
security = HTTPBasic()
|
||||
|
||||
|
||||
async def user_is_admin(
|
||||
credentials: HTTPBasicCredentials = Depends(security),
|
||||
config: Config = Depends(Config.get_config),
|
||||
) -> bool:
|
||||
username_correct = secrets.compare_digest(credentials.username, config.admin.name)
|
||||
|
||||
password_correct = secrets.compare_digest(
|
||||
credentials.password, config.admin.password
|
||||
)
|
||||
|
||||
return username_correct and password_correct
|
||||
|
||||
|
||||
async def require_admin(
|
||||
is_admin: bool = Depends(user_is_admin),
|
||||
) -> None:
|
||||
if not is_admin:
|
||||
raise HTTPException(status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
|
||||
async def user_visible_days() -> int:
|
||||
today = date.today()
|
||||
|
||||
if today.month == 12:
|
||||
return today.day
|
||||
|
||||
if today.month in (1, 2, 3):
|
||||
return 24
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
async def user_can_view_day(
|
||||
day: int,
|
||||
) -> bool:
|
||||
return day < await user_visible_days()
|
|
@ -7,7 +7,7 @@ from PIL import Image
|
|||
from ..core.config import Config
|
||||
from ..core.depends import get_image, get_part, shuffle_solution
|
||||
from ..core.image_helpers import api_return_image
|
||||
from .user import user_is_admin
|
||||
from ._security import user_can_view_day, user_is_admin, user_visible_days
|
||||
|
||||
router = APIRouter(prefix="/days", tags=["days"])
|
||||
|
||||
|
@ -23,30 +23,30 @@ async def startup() -> None:
|
|||
|
||||
@router.get("/date")
|
||||
async def get_date() -> str:
|
||||
"""
|
||||
Aktuelles Server-Datum
|
||||
"""
|
||||
|
||||
return date.today().isoformat()
|
||||
|
||||
|
||||
@router.get("/visible_days")
|
||||
async def get_visible_days() -> int:
|
||||
today = date.today()
|
||||
"""
|
||||
Sichtbare Türchen
|
||||
"""
|
||||
|
||||
if today.month == 12:
|
||||
return today.day
|
||||
|
||||
if today.month in (1, 2, 3):
|
||||
return 24
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
async def user_can_view(
|
||||
day: int,
|
||||
) -> bool:
|
||||
return day < await get_visible_days()
|
||||
return await user_visible_days()
|
||||
|
||||
|
||||
@router.get("/part/{day}")
|
||||
async def get_part_for_day(part: str = Depends(get_part)) -> str:
|
||||
async def get_part_for_day(
|
||||
part: str = Depends(get_part),
|
||||
) -> str:
|
||||
"""
|
||||
Heutiger Lösungsteil
|
||||
"""
|
||||
|
||||
return part
|
||||
|
||||
|
||||
|
@ -56,7 +56,7 @@ async def get_part_for_day(part: str = Depends(get_part)) -> str:
|
|||
)
|
||||
async def get_image_for_day(
|
||||
image: Image.Image = Depends(get_image),
|
||||
can_view: bool = Depends(user_can_view),
|
||||
can_view: bool = Depends(user_can_view_day),
|
||||
is_admin: bool = Depends(user_is_admin),
|
||||
) -> StreamingResponse:
|
||||
"""
|
||||
|
|
|
@ -1,32 +1,8 @@
|
|||
import secrets
|
||||
from fastapi import APIRouter, Depends
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||||
|
||||
from ..core.config import Config
|
||||
from ._security import require_admin
|
||||
|
||||
router = APIRouter(prefix="/user", tags=["user"])
|
||||
security = HTTPBasic()
|
||||
|
||||
|
||||
async def user_is_admin(
|
||||
credentials: HTTPBasicCredentials = Depends(security),
|
||||
config: Config = Depends(Config.get_config),
|
||||
) -> bool:
|
||||
username_correct = secrets.compare_digest(credentials.username, config.admin.name)
|
||||
|
||||
password_correct = secrets.compare_digest(
|
||||
credentials.password, config.admin.password
|
||||
)
|
||||
|
||||
return username_correct and password_correct
|
||||
|
||||
|
||||
async def require_admin(
|
||||
is_admin: bool = Depends(user_is_admin),
|
||||
) -> None:
|
||||
if not is_admin:
|
||||
raise HTTPException(status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
|
||||
@router.get("/admin")
|
||||
|
|
Loading…
Reference in a new issue