SyncAPI/modules/app.py

55 lines
2.1 KiB
Python
Raw Normal View History

2023-01-18 15:25:22 +02:00
from typing import Union
2023-01-16 14:23:18 +02:00
from fastapi import FastAPI, Security, HTTPException
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
from fastapi.security import APIKeyQuery, APIKeyHeader, APIKeyCookie
2023-01-18 15:25:22 +02:00
from fastapi.openapi.models import APIKey
2023-01-16 14:23:18 +02:00
from fastapi.openapi.docs import get_swagger_ui_html, get_redoc_html
from starlette.status import HTTP_401_UNAUTHORIZED
2023-01-18 15:31:22 +02:00
from modules.utils import configGet
2023-01-18 15:25:22 +02:00
from modules.security import passEncode
from modules.database import col_apikeys, col_expired
2023-01-16 14:23:18 +02:00
app = FastAPI(title="Stardew Sync", docs_url=None, redoc_url=None, version="0.1")
api_key_query = APIKeyQuery(name="apikey", auto_error=False)
api_key_header = APIKeyHeader(name="apikey", auto_error=False)
async def get_api_key(
api_key_query: str = Security(api_key_query),
api_key_header: str = Security(api_key_header),
) -> str:
def is_valid(key):
2023-01-18 15:25:22 +02:00
return True if col_apikeys.find_one({"hash": passEncode(key)}) is not None else False
2023-01-16 14:23:18 +02:00
if is_valid(api_key_query):
return api_key_query
elif is_valid(api_key_header):
return api_key_header
else:
2023-01-20 17:02:46 +02:00
if (col_expired.find_one({"hash": passEncode(api_key_query)}) is not None) or (col_expired.find_one({"hash": passEncode(api_key_header)}) is not None):
2023-01-16 14:23:18 +02:00
raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail=configGet("key_expired", "messages"))
else:
raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail=configGet("key_invalid", "messages"))
2023-01-18 15:25:22 +02:00
def user_by_key(apikey: Union[str, APIKey]) -> Union[str, None]:
db_key = col_apikeys.find_one({"hash": passEncode(apikey)})
return db_key["user"] if db_key is not None else None
2023-01-18 15:31:22 +02:00
2023-01-16 14:23:18 +02:00
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
return get_swagger_ui_html(
openapi_url=app.openapi_url, # type: ignore
title=app.title + " - Documentation",
swagger_favicon_url="/favicon.ico"
)
@app.get("/redoc", include_in_schema=False)
async def custom_redoc_html():
return get_redoc_html(
openapi_url=app.openapi_url, # type: ignore
title=app.title + " - Documentation",
redoc_favicon_url="/favicon.ico"
)