75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
|
from os import path
|
||
|
|
from fastapi import FastAPI, Security, HTTPException
|
||
|
|
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
|
||
|
|
from fastapi.security import APIKeyQuery, APIKeyHeader, APIKeyCookie
|
||
|
|
from fastapi.openapi.docs import get_swagger_ui_html, get_redoc_html
|
||
|
|
from starlette.status import HTTP_401_UNAUTHORIZED
|
||
|
|
|
||
|
|
from modules.utils import configGet, jsonLoad
|
||
|
|
|
||
|
|
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)
|
||
|
|
api_key_cookie = APIKeyCookie(name="apikey", auto_error=False)
|
||
|
|
|
||
|
|
|
||
|
|
def get_all_api_keys() -> list:
|
||
|
|
return jsonLoad(path.join(configGet("data", "locations"), "api_keys.json"))
|
||
|
|
|
||
|
|
def get_all_expired_keys() -> list:
|
||
|
|
return jsonLoad(path.join(configGet("data", "locations"), "expired_keys.json"))
|
||
|
|
|
||
|
|
# def check_project_key(project: str, apikey: APIKey) -> bool:
|
||
|
|
# keys = jsonLoad(path.join(configGet("data", "locations"), "api_keys.json"))
|
||
|
|
# if apikey in keys:
|
||
|
|
# if keys[apikey] != []:
|
||
|
|
# if project in keys[apikey]:
|
||
|
|
# return True
|
||
|
|
# else:
|
||
|
|
# return False
|
||
|
|
# else:
|
||
|
|
# return False
|
||
|
|
# else:
|
||
|
|
# return False
|
||
|
|
|
||
|
|
|
||
|
|
async def get_api_key(
|
||
|
|
api_key_query: str = Security(api_key_query),
|
||
|
|
api_key_header: str = Security(api_key_header),
|
||
|
|
api_key_cookie: str = Security(api_key_cookie),
|
||
|
|
) -> str:
|
||
|
|
|
||
|
|
keys = get_all_api_keys()
|
||
|
|
expired = get_all_expired_keys()
|
||
|
|
|
||
|
|
def is_valid(key):
|
||
|
|
return True if key in keys else False
|
||
|
|
|
||
|
|
if is_valid(api_key_query):
|
||
|
|
return api_key_query
|
||
|
|
elif is_valid(api_key_header):
|
||
|
|
return api_key_header
|
||
|
|
elif is_valid(api_key_cookie):
|
||
|
|
return api_key_cookie
|
||
|
|
else:
|
||
|
|
if (api_key_query in expired) or (api_key_header in expired) or (api_key_cookie in expired):
|
||
|
|
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"))
|
||
|
|
|
||
|
|
@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"
|
||
|
|
)
|