Same skip paths, Bearer/query token, Clerk RS256 leftover, and local HS256 contract. routers/auth.py, user records, and AUTH_JWT_SECRET are untouched.
101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
"""JWT request verification lives under periscope/src (same HTTP contract)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from starlette.requests import Request
|
|
|
|
from backend.middleware import auth as auth_mod
|
|
from backend.services.local_jwt import issue_token
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SRC_AUTH = ROOT / "periscope" / "src" / "backend" / "middleware" / "auth.py"
|
|
SRC_HOOK = ROOT / "periscope" / "src" / "frontend" / "src" / "hooks" / "use-auth-api.ts"
|
|
|
|
|
|
def test_auth_modules_are_src():
|
|
path = Path(auth_mod.__file__).resolve()
|
|
assert path == SRC_AUTH
|
|
assert "Native Periscope overlay" not in path.read_text(encoding="utf-8")[:400]
|
|
text = SRC_HOOK.read_text(encoding="utf-8")
|
|
assert "export function useAuthApi" in text
|
|
assert "setTokenGetter" in text
|
|
assert "Native Periscope overlay" not in text[:400]
|
|
|
|
|
|
def test_skip_paths_keep_login_contract():
|
|
assert "/api/auth/login" in auth_mod._SKIP_PATHS
|
|
assert "/api/auth/register" in auth_mod._SKIP_PATHS
|
|
assert "/api/auth/mode" in auth_mod._SKIP_PATHS
|
|
assert "/api/contact" in auth_mod._SKIP_PATHS
|
|
assert "/health" in auth_mod._SKIP_PATHS
|
|
|
|
|
|
def _request(path: str, token: str | None = None, query_token: str | None = None) -> Request:
|
|
headers: list[tuple[bytes, bytes]] = []
|
|
if token:
|
|
headers.append((b"authorization", f"Bearer {token}".encode()))
|
|
raw_qs = f"token={query_token}".encode() if query_token else b""
|
|
scope = {
|
|
"type": "http",
|
|
"asgi": {"version": "3.0"},
|
|
"http_version": "1.1",
|
|
"method": "GET",
|
|
"scheme": "http",
|
|
"path": path,
|
|
"raw_path": path.encode(),
|
|
"query_string": raw_qs,
|
|
"headers": headers,
|
|
"client": ("127.0.0.1", 0),
|
|
"server": ("test", 80),
|
|
}
|
|
return Request(scope)
|
|
|
|
|
|
@pytest.fixture()
|
|
def local_auth_settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
from backend.config import settings
|
|
|
|
monkeypatch.setattr(settings, "auth_jwt_secret", "test-secret-not-for-prod")
|
|
monkeypatch.setattr(settings, "clerk_secret_key", "")
|
|
monkeypatch.setattr(settings, "clerk_jwks_url", "")
|
|
monkeypatch.setattr(settings, "data_dir", tmp_path)
|
|
return settings
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_local_jwt_accepts_bearer(local_auth_settings):
|
|
token = issue_token("user-1", "a@example.com")
|
|
user = await auth_mod.verify_local_token(_request("/api/projects", token=token))
|
|
assert user == "user-1"
|
|
dispatched = await auth_mod.verify_request_user(_request("/api/projects", token=token))
|
|
assert dispatched == "user-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_local_jwt_query_token_and_skips(local_auth_settings):
|
|
token = issue_token("user-2", "b@example.com")
|
|
user = await auth_mod.verify_local_token(
|
|
_request("/api/pipeline/x/events", query_token=token)
|
|
)
|
|
assert user == "user-2"
|
|
skipped = await auth_mod.verify_local_token(_request("/api/auth/login"))
|
|
assert skipped == "anonymous"
|
|
missing = await auth_mod.verify_local_token(_request("/api/projects"))
|
|
assert missing is None
|
|
bad = await auth_mod.verify_local_token(_request("/api/projects", token="not-a-jwt"))
|
|
assert bad is None
|
|
|
|
|
|
def test_routers_auth_module_unchanged_contract():
|
|
from backend.routers import auth as routers_auth
|
|
|
|
src = inspect.getsource(routers_auth.login)
|
|
assert "local_users.authenticate" in src
|
|
assert "local_jwt.issue_token" in src
|