Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions app/middlewares/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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=["*"],
)
Expand Down
6 changes: 5 additions & 1 deletion app/utils/jwt.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
51 changes: 51 additions & 0 deletions tests/test_security_hardening.py
Original file line number Diff line number Diff line change
@@ -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
Loading