1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
import logging
import time
from collections.abc import Awaitable, Callable
from fastapi import FastAPI, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from src.routes import top_level_router
from src.settings import CONFIG
log = logging.getLogger(__name__)
fastapi_app = FastAPI(debug=CONFIG.debug, root_path=CONFIG.app_prefix)
fastapi_app.include_router(top_level_router)
@fastapi_app.get("/heartbeat")
def health_check() -> JSONResponse:
"""Return basic response, for use as a health check."""
return JSONResponse({"detail": "I am alive!"})
@fastapi_app.exception_handler(RequestValidationError)
def pydantic_validation_error(request: Request, error: RequestValidationError) -> JSONResponse:
"""Raise a warning for pydantic validation errors, before returning."""
log.warning("Error from %s: %s", request.url, error)
return JSONResponse({"error": str(error)}, status_code=422)
@fastapi_app.middleware("http")
async def add_process_time_and_security_headers(
request: Request,
call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
"""Add process time and some security headers before sending the response."""
start_time = time.perf_counter()
response = await call_next(request)
response.headers["X-Process-Time"] = str(time.perf_counter() - start_time)
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Strict-Transport-Security"] = "max-age=31536000"
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["Content-Security-Policy"] = "default-src 'self'"
response.headers["Referrer-Policy"] = "no-referrer"
response.headers["Permissions-Policy"] = (
"camera=(), display-capture(), fullscreen(), geolocation=(), microphone(), screen-wake-lock(), web-share()"
)
return response
|
