diff --git a/app/middlewares/__init__.py b/app/middlewares/__init__.py index 0d581a4d2..58a373012 100644 --- a/app/middlewares/__init__.py +++ b/app/middlewares/__init__.py @@ -9,10 +9,13 @@ def setup_middleware(app: FastAPI): + allowed_origins = cors_settings.allowed_origins + # With a wildcard plus credentials, Starlette echoes every Origin back with Allow-Credentials. + allow_credentials = "*" not in allowed_origins app.add_middleware( CORSMiddleware, - allow_origins=cors_settings.allowed_origins, - allow_credentials=True, + allow_origins=allowed_origins, + allow_credentials=allow_credentials, allow_methods=["*"], allow_headers=["*"], ) diff --git a/app/utils/jwt.py b/app/utils/jwt.py index cbbfe6275..dfb8b70f6 100644 --- a/app/utils/jwt.py +++ b/app/utils/jwt.py @@ -188,7 +188,11 @@ async def get_subscription_payload(token: str) -> dict | None: :10 ] u_token_hex_resign = sha256((u_token + secret).encode("utf-8")).hexdigest()[:10] - if u_signature in (u_token_resign, u_token_hex_resign): + # Constant-time compare against both encodings; bytes, since str input must be ASCII. + signature = u_signature.encode("utf-8") + if hmac.compare_digest(signature, u_token_resign.encode("utf-8")) | hmac.compare_digest( + signature, u_token_hex_resign.encode("utf-8") + ): return _parse_subscription_data(u_token_dec_str) return except jwt.exceptions.PyJWTError: diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py new file mode 100644 index 000000000..d06c95cf1 --- /dev/null +++ b/tests/test_security_hardening.py @@ -0,0 +1,51 @@ +import asyncio +import time +from base64 import b64encode +from hashlib import sha256 +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app import middlewares +from app.utils import jwt as jwt_utils + + +def _legacy_v2_token(user_id: int, secret: str) -> str: + data_b64 = b64encode(f"v2,{user_id},{int(time.time())}".encode(), altchars=b"-_").decode().rstrip("=") + return data_b64 + b64encode(sha256((data_b64 + secret).encode()).digest(), altchars=b"-_").decode()[:10] + + +@pytest.fixture +def secret(monkeypatch): + async def fake_get_secret_key(): + return "test-secret" + + monkeypatch.setattr(jwt_utils, "get_secret_key", fake_get_secret_key) + return "test-secret" + + +def test_legacy_subscription_token_rejects_bad_signatures_without_errors(secret): + token = _legacy_v2_token(789, secret) + + assert asyncio.run(jwt_utils.get_subscription_payload(token))["user_id"] == 789 + for signature in ("A" * 10, "é" * 10): + assert asyncio.run(jwt_utils.get_subscription_payload(token[:-10] + signature)) is None + + +@pytest.mark.parametrize( + ("origins", "credentials"), + [("*", None), ("https://panel.example.com", "true")], +) +def test_cors_credentials_only_for_explicit_origins(monkeypatch, origins, credentials): + monkeypatch.setattr(middlewares, "cors_settings", SimpleNamespace(allowed_origins=[origins])) + app = FastAPI() + middlewares.setup_middleware(app) + app.get("/ping")(lambda: {"ok": True}) + + response = TestClient(app).get("/ping", headers={"Origin": "https://panel.example.com", "Cookie": "session=1"}) + + assert response.headers.get("access-control-allow-credentials") == credentials + expected_origin = "*" if origins == "*" else "https://panel.example.com" + assert response.headers.get("access-control-allow-origin") == expected_origin