+39
-14
@@ -121,27 +121,52 @@ def validate_subscription_period(days: Union[str, int]) -> Optional[int]:
|
||||
|
||||
|
||||
def sanitize_html(text: str) -> str:
|
||||
"""
|
||||
Безопасно санитизирует HTML-текст, заменяя HTML-сущности на соответствующие теги,
|
||||
при этом предотвращая XSS-уязвимости за счет безопасной обработки атрибутов.
|
||||
|
||||
Args:
|
||||
text (str): Текст с HTML-сущностями (например, <b> жирный </b>)
|
||||
|
||||
Returns:
|
||||
str: Санитизированный HTML-текст (например, <b> жирный </b>)
|
||||
"""
|
||||
if not text:
|
||||
return text
|
||||
|
||||
text = html.escape(text)
|
||||
# Для безопасности нужно обработать разрешенные теги, заменяя их сущности на теги
|
||||
# Но при этом безопасно обрабатывая атрибуты, чтобы избежать XSS
|
||||
|
||||
allowed_tags = ALLOWED_HTML_TAGS.union(SELF_CLOSING_TAGS)
|
||||
|
||||
# Обработка всех разрешенных тегов
|
||||
for tag in allowed_tags:
|
||||
text = re.sub(
|
||||
f'<(/?{tag}\\b[^>]*)>',
|
||||
lambda m: "<"
|
||||
+ (
|
||||
m.group(1)
|
||||
.replace(""", "\"")
|
||||
.replace("'", "'")
|
||||
.replace("&", "&")
|
||||
)
|
||||
+ ">",
|
||||
text,
|
||||
flags=re.IGNORECASE
|
||||
)
|
||||
# Паттерн: захватываем <tag>, </tag>, или <tag атрибуты>
|
||||
# Используем более сложный паттерн, чтобы захватить атрибуты до закрывающего >
|
||||
# (?s) - позволяет . захватывать новую строку
|
||||
# [^>]*? - ленивый захват до >
|
||||
pattern = rf'(<)(/?{tag}\b)([^>]*?)(>)'
|
||||
|
||||
def replace_tag(match):
|
||||
opening = match.group(1) # <
|
||||
full_tag_content = match.group(2) # /?tagname
|
||||
attrs_part = match.group(3) # атрибуты (без >)
|
||||
closing = match.group(4) # >
|
||||
|
||||
# Убираем начальный пробел, если есть
|
||||
if attrs_part.startswith(' '):
|
||||
attrs_part = attrs_part[1:]
|
||||
|
||||
# Формируем результат
|
||||
if attrs_part:
|
||||
# Безопасно обрабатываем атрибуты, заменяя только безопасные сущности
|
||||
# Не разворачиваем < и > внутри атрибутов, чтобы избежать XSS
|
||||
processed_attrs = attrs_part.replace('"', '"').replace(''', "'")
|
||||
return f'<{full_tag_content} {processed_attrs}>'
|
||||
else:
|
||||
return f'<{full_tag_content}>'
|
||||
|
||||
text = re.sub(pattern, replace_tag, text, flags=re.IGNORECASE)
|
||||
|
||||
return text
|
||||
|
||||
|
||||
+9
-1
@@ -4,6 +4,7 @@ from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.config import settings
|
||||
from app.webapi.docs import add_redoc_endpoint
|
||||
|
||||
from .middleware import RequestLoggingMiddleware
|
||||
from .routes import (
|
||||
@@ -144,11 +145,18 @@ def create_web_api_app() -> FastAPI:
|
||||
title=settings.WEB_API_TITLE,
|
||||
version=settings.WEB_API_VERSION,
|
||||
docs_url=docs_config.get("docs_url"),
|
||||
redoc_url=docs_config.get("redoc_url"),
|
||||
redoc_url=None,
|
||||
openapi_url=docs_config.get("openapi_url"),
|
||||
swagger_ui_parameters={"persistAuthorization": True},
|
||||
)
|
||||
|
||||
add_redoc_endpoint(
|
||||
app,
|
||||
redoc_url=docs_config.get("redoc_url"),
|
||||
openapi_url=docs_config.get("openapi_url"),
|
||||
title=settings.WEB_API_TITLE,
|
||||
)
|
||||
|
||||
allowed_origins = settings.get_web_api_allowed_origins()
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
from fastapi import FastAPI
|
||||
from fastapi.openapi.docs import get_redoc_html
|
||||
|
||||
|
||||
def add_redoc_endpoint(
|
||||
app: FastAPI,
|
||||
*,
|
||||
redoc_url: str | None,
|
||||
openapi_url: str | None,
|
||||
title: str | None,
|
||||
) -> None:
|
||||
"""Attach a ReDoc endpoint if docs are enabled.
|
||||
|
||||
The default FastAPI ReDoc handler sometimes renders a blank page when the
|
||||
CDN bundle fails to load. By explicitly registering the handler and
|
||||
pinning the bundle version, we ensure the endpoint always returns a fully
|
||||
rendered page.
|
||||
"""
|
||||
|
||||
if not redoc_url or not openapi_url:
|
||||
return
|
||||
|
||||
for route in app.router.routes:
|
||||
if getattr(route, "path", None) == redoc_url:
|
||||
return
|
||||
|
||||
@app.get(redoc_url, include_in_schema=False)
|
||||
async def redoc_html(): # pragma: no cover - template rendering
|
||||
return get_redoc_html(
|
||||
openapi_url=openapi_url,
|
||||
title=f"{title or app.title} - ReDoc",
|
||||
redoc_js_url="https://cdn.jsdelivr.net/npm/redoc@2.1.5/bundles/redoc.standalone.js",
|
||||
)
|
||||
@@ -13,6 +13,7 @@ from aiogram import Dispatcher
|
||||
from app.config import settings
|
||||
from app.services.payment_service import PaymentService
|
||||
from app.webapi.app import create_web_api_app
|
||||
from app.webapi.docs import add_redoc_endpoint
|
||||
|
||||
from . import payments
|
||||
from . import telegram
|
||||
@@ -47,12 +48,19 @@ def _create_base_app() -> FastAPI:
|
||||
app = create_web_api_app()
|
||||
else:
|
||||
app = FastAPI(
|
||||
title="Bedolaga Unified Server",
|
||||
version=settings.WEB_API_VERSION,
|
||||
docs_url=docs_config.get("docs_url"),
|
||||
redoc_url=docs_config.get("redoc_url"),
|
||||
openapi_url=docs_config.get("openapi_url"),
|
||||
)
|
||||
title="Bedolaga Unified Server",
|
||||
version=settings.WEB_API_VERSION,
|
||||
docs_url=docs_config.get("docs_url"),
|
||||
redoc_url=None,
|
||||
openapi_url=docs_config.get("openapi_url"),
|
||||
)
|
||||
|
||||
add_redoc_endpoint(
|
||||
app,
|
||||
redoc_url=docs_config.get("redoc_url"),
|
||||
openapi_url=docs_config.get("openapi_url"),
|
||||
title="Bedolaga Unified Server",
|
||||
)
|
||||
|
||||
_attach_docs_alias(app, app.docs_url)
|
||||
return app
|
||||
|
||||
@@ -133,7 +133,6 @@ async def test_unified_app_docs_enabled_with_alias(monkeypatch: pytest.MonkeyPat
|
||||
app = _build_unified_app(monkeypatch, docs_enabled=True)
|
||||
|
||||
assert app.docs_url == "/docs"
|
||||
assert app.redoc_url == "/redoc"
|
||||
assert app.openapi_url == "/openapi.json"
|
||||
|
||||
alias_route = next(
|
||||
@@ -143,6 +142,16 @@ async def test_unified_app_docs_enabled_with_alias(monkeypatch: pytest.MonkeyPat
|
||||
assert alias_route is not None
|
||||
assert getattr(alias_route, "include_in_schema", True) is False
|
||||
|
||||
redoc_route = next(
|
||||
(route for route in app.routes if getattr(route, "path", None) == "/redoc"),
|
||||
None,
|
||||
)
|
||||
assert redoc_route is not None
|
||||
assert getattr(redoc_route, "include_in_schema", True) is False
|
||||
|
||||
response = await alias_route.endpoint() # type: ignore[func-returns-value]
|
||||
assert response.status_code == status.HTTP_307_TEMPORARY_REDIRECT
|
||||
assert response.headers["location"] == "/docs"
|
||||
|
||||
redoc_response = await redoc_route.endpoint() # type: ignore[func-returns-value]
|
||||
assert b"ReDoc" in redoc_response.body # type: ignore[attr-defined]
|
||||
|
||||
Reference in New Issue
Block a user