@@ -16,6 +16,8 @@ class Settings(BaseSettings):
|
||||
|
||||
# App
|
||||
debug: bool = Field(default=True, alias="DEBUG")
|
||||
swagger_auth_login: str = Field(default="", alias="SWAGGER_AUTH_LOGIN")
|
||||
swagger_auth_password: str = Field(default="", alias="SWAGGER_AUTH_PASSWORD")
|
||||
cors_allowed_origins: list[str] = Field(
|
||||
default_factory=lambda: [
|
||||
"http://localhost:3000",
|
||||
|
||||
+55
-5
@@ -1,18 +1,26 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from secrets import compare_digest
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends, FastAPI, HTTPException, status
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.openapi.docs import get_swagger_ui_html
|
||||
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||||
|
||||
from cpv3.infrastructure.settings import get_settings
|
||||
from cpv3.api.v1.router import api_router
|
||||
from cpv3.infrastructure.settings import get_settings
|
||||
|
||||
ERROR_SWAGGER_AUTH_INVALID = "Invalid Swagger credentials"
|
||||
|
||||
settings = get_settings()
|
||||
swagger_security = HTTPBasic()
|
||||
|
||||
app = FastAPI(
|
||||
title="Coffee Project Backend API",
|
||||
version="0.0.0",
|
||||
openapi_url="/api/schema/",
|
||||
docs_url="/api/schema/swagger/",
|
||||
openapi_url=None,
|
||||
docs_url=None,
|
||||
redoc_url=None,
|
||||
)
|
||||
|
||||
@@ -24,5 +32,47 @@ app.add_middleware(
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
|
||||
def verify_swagger_auth(
|
||||
credentials: Annotated[HTTPBasicCredentials, Depends(swagger_security)],
|
||||
) -> None:
|
||||
login_matches = compare_digest(
|
||||
credentials.username,
|
||||
settings.swagger_auth_login,
|
||||
)
|
||||
password_matches = compare_digest(
|
||||
credentials.password,
|
||||
settings.swagger_auth_password,
|
||||
)
|
||||
|
||||
if not settings.swagger_auth_login or not settings.swagger_auth_password:
|
||||
login_matches = False
|
||||
password_matches = False
|
||||
|
||||
if not login_matches or not password_matches:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail=ERROR_SWAGGER_AUTH_INVALID,
|
||||
headers={"WWW-Authenticate": "Basic"},
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/schema/", include_in_schema=False)
|
||||
async def openapi_schema(
|
||||
_: Annotated[None, Depends(verify_swagger_auth)],
|
||||
):
|
||||
return app.openapi()
|
||||
|
||||
|
||||
@app.get("/api/schema/swagger/", include_in_schema=False)
|
||||
async def swagger_ui(
|
||||
_: Annotated[None, Depends(verify_swagger_auth)],
|
||||
):
|
||||
return get_swagger_ui_html(
|
||||
openapi_url="/api/schema/",
|
||||
title="Coffee Project Backend API - Swagger UI",
|
||||
)
|
||||
|
||||
|
||||
# Include the versioned API router
|
||||
app.include_router(api_router)
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient, BasicAuth
|
||||
|
||||
from cpv3.infrastructure.settings import get_settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def swagger_app(monkeypatch):
|
||||
monkeypatch.setenv("SWAGGER_AUTH_LOGIN", "docs-user")
|
||||
monkeypatch.setenv("SWAGGER_AUTH_PASSWORD", "docs-password")
|
||||
get_settings.cache_clear()
|
||||
|
||||
import cpv3.main
|
||||
|
||||
module = importlib.reload(cpv3.main)
|
||||
yield module.app
|
||||
|
||||
get_settings.cache_clear()
|
||||
importlib.reload(cpv3.main)
|
||||
|
||||
|
||||
async def test_swagger_ui_requires_basic_auth(swagger_app):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=swagger_app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.get("/api/schema/swagger/")
|
||||
|
||||
assert response.status_code == 401
|
||||
assert response.headers["www-authenticate"] == "Basic"
|
||||
|
||||
|
||||
async def test_openapi_schema_requires_basic_auth(swagger_app):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=swagger_app),
|
||||
base_url="http://test",
|
||||
) as client:
|
||||
response = await client.get("/api/schema/")
|
||||
|
||||
assert response.status_code == 401
|
||||
assert response.headers["www-authenticate"] == "Basic"
|
||||
|
||||
|
||||
async def test_swagger_credentials_come_from_environment(swagger_app):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=swagger_app),
|
||||
base_url="http://test",
|
||||
auth=BasicAuth("docs-user", "docs-password"),
|
||||
) as client:
|
||||
response = await client.get("/api/schema/swagger/")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Swagger UI" in response.text
|
||||
Reference in New Issue
Block a user