MdeMoLii/mediaflow-proxy

1import logging2from importlib import resources3 4from fastapi import FastAPI, Depends, Security, HTTPException5from fastapi.security import APIKeyQuery, APIKeyHeader6from starlette.middleware.cors import CORSMiddleware7from starlette.responses import RedirectResponse8from starlette.staticfiles import StaticFiles9 10from mediaflow_proxy.configs import settings11from mediaflow_proxy.routes import proxy_router12from mediaflow_proxy.schemas import GenerateUrlRequest13from mediaflow_proxy.utils.crypto_utils import EncryptionHandler, EncryptionMiddleware14from mediaflow_proxy.utils.http_utils import encode_mediaflow_proxy_url15 16logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")17app = FastAPI()18api_password_query = APIKeyQuery(name="api_password", auto_error=False)19api_password_header = APIKeyHeader(name="api_password", auto_error=False)20app.add_middleware(21    CORSMiddleware,22    allow_origins=["*"],23    allow_credentials=True,24    allow_methods=["*"],25    allow_headers=["*"],26)27app.add_middleware(EncryptionMiddleware)28 29 30async def verify_api_key(api_key: str = Security(api_password_query), api_key_alt: str = Security(api_password_header)):31    """32    Verifies the API key for the request.33 34    Args:35        api_key (str): The API key to validate.36        api_key_alt (str): The alternative API key to validate.37 38    Raises:39        HTTPException: If the API key is invalid.40    """41    if api_key == settings.api_password or api_key_alt == settings.api_password:42        return43 44    raise HTTPException(status_code=403, detail="Could not validate credentials")45 46 47@app.get("/health")48async def health_check():49    return {"status": "healthy"}50 51 52@app.get("/favicon.ico")53async def get_favicon():54    return RedirectResponse(url="/logo.png")55 56 57@app.post("/generate_encrypted_or_encoded_url")58async def generate_encrypted_or_encoded_url(request: GenerateUrlRequest):59    if "api_password" not in request.query_params:60        request.query_params["api_password"] = request.api_password61 62    encoded_url = encode_mediaflow_proxy_url(63        request.mediaflow_proxy_url,64        request.endpoint,65        request.destination_url,66        request.query_params,67        request.request_headers,68        request.response_headers,69        EncryptionHandler(request.api_password) if request.api_password else None,70        request.expiration,71        str(request.ip) if request.ip else None,72    )73    return {"encoded_url": encoded_url}74 75 76app.include_router(proxy_router, prefix="/proxy", tags=["proxy"], dependencies=[Depends(verify_api_key)])77 78static_path = resources.files("mediaflow_proxy").joinpath("static")79app.mount("/", StaticFiles(directory=str(static_path), html=True), name="static")80 81 82def run():83    import uvicorn84 85    uvicorn.run(app, host="0.0.0.0", port=8888, log_level="info", workers=3)86 87 88if __name__ == "__main__":89    run()90