import logging from hashlib import sha256 from asgiref.sync import async_to_sync from fastapi import APIRouter, Depends, HTTPException from fastapi.security import HTTPBasic, HTTPBasicCredentials from starlette import status from starlette.requests import Request from starlette.responses import HTMLResponse from werkzeug.security import check_password_hash from service.conf.settings import ADMIN_USER_CREDENTIALS from service.tasks.management.admin.pre_signup import ( ingest_pre_signup_instructions, page_pre_signup, ) from service.tasks.management.admin.users import ( ingest_manage_users_instructions, page_manage_users, ) from service.tasks.management.admin.utils import make_iso_date_now logger = logging.getLogger(__name__) basic_auth = HTTPBasic() router = APIRouter(tags=["Admin"]) NAUGHTY_SALT = "3bc85b79e3fa4f37c0db7247b338c1595bd684c0335464592a54c0180af4a49f" def generate_random_string() -> str: h = sha256(NAUGHTY_SALT.encode()) h.update(make_iso_date_now().encode()) return h.hexdigest() def login_required(credentials: HTTPBasicCredentials = Depends(basic_auth)): if credentials.username in ADMIN_USER_CREDENTIALS and check_password_hash( pwhash=ADMIN_USER_CREDENTIALS.get(credentials.username) or "", password=credentials.password, ): return credentials.username raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Basic"}, ) def do_admin(make_page, ingest_data, path, data, request: Request): message = "Hi, select stuff and press buttons." if data: naughty_bits = request.cookies.get(path) message = ingest_data(data, naughty_bits) naughty_bits = generate_random_string() response = HTMLResponse(make_page(message, data, naughty_bits), 200) response.set_cookie( path, naughty_bits, # type: ignore secure=True, httponly=True, samesite="lax", path=f"/{path}", ) return response @router.api_route( "/admin_pre-signup", methods=["GET", "POST"], dependencies=[Depends(login_required)], response_class=HTMLResponse, ) def admin_pre_signup(request: Request): return do_admin( page_pre_signup, ingest_pre_signup_instructions, "admin_pre-signup", async_to_sync(request.form)(), request, ) @router.api_route( "/admin_users", methods=["GET", "POST"], dependencies=[Depends(login_required)], response_class=HTMLResponse, ) def admin_manage_users(request: Request): return do_admin( page_manage_users, ingest_manage_users_instructions, "admin_users", async_to_sync(request.form)(), request, )