refactor: final-review cleanup
- delete dead magic-link helpers (resolve_token, ensure_project_client, mint_link_token, provision_preview_session) + now-unused datetime import - key brute-force lockout on link_token alone (IP term only enabled a source-IP-rotation bypass; behind the proxy all clients share one IP) - drop unused PORTAL_BASE_URL from the retired CLI - add WebSocket ownership tests (unauth + cross-project both close 1008)
This commit is contained in:
@@ -1,5 +1,8 @@
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
import pytest
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from starlette.testclient import WebSocketDisconnect
|
||||
from tests.conftest import make_project
|
||||
from backend import portal_auth as pa
|
||||
from backend.auth_passwords import hash_password
|
||||
@@ -41,3 +44,38 @@ def test_session_can_open_its_own_location(client, db_session):
|
||||
assert r.status_code == 303
|
||||
r2 = client.get(f"/portal/location/{a_loc.id}")
|
||||
assert r2.status_code == 200
|
||||
|
||||
|
||||
def test_ws_stream_rejects_unauthenticated(client, db_session):
|
||||
# The live-feed WebSocket must refuse a connection with no session cookie (1008).
|
||||
a = make_project(db_session, portal_enabled=True, portal_link_token="tw1",
|
||||
portal_password_hash=hash_password("pw"))
|
||||
a_loc = _sound_location(db_session, a)
|
||||
with pytest.raises(WebSocketDisconnect) as exc:
|
||||
with client.websocket_connect(f"/portal/api/location/{a_loc.id}/stream") as ws:
|
||||
ws.receive_text()
|
||||
assert exc.value.code == 1008
|
||||
|
||||
|
||||
def test_ws_stream_rejects_cross_project(client, db_session, monkeypatch):
|
||||
# The WebSocket enforces the SAME per-project ownership as the HTTP routes: a
|
||||
# B-session opening A's stream is closed 1008 (ownership) before any device feed.
|
||||
# The handler uses SessionLocal() directly (not the get_db override), so point it
|
||||
# at the test DB engine so this genuinely exercises the ownership check (not a
|
||||
# vacuous "client not found").
|
||||
import backend.routers.portal as portal_router
|
||||
monkeypatch.setattr(portal_router, "SessionLocal",
|
||||
sessionmaker(bind=db_session.get_bind()))
|
||||
|
||||
a = make_project(db_session, portal_enabled=True, portal_link_token="tw2",
|
||||
portal_password_hash=hash_password("pw"))
|
||||
a_loc = _sound_location(db_session, a)
|
||||
make_project(db_session, portal_enabled=True, portal_link_token="tw3",
|
||||
portal_password_hash=hash_password("pw"))
|
||||
# Log in as project B, then aim the stream at project A's location.
|
||||
assert client.post("/portal/p/tw3", data={"password": "pw"},
|
||||
follow_redirects=False).status_code == 303
|
||||
with pytest.raises(WebSocketDisconnect) as exc:
|
||||
with client.websocket_connect(f"/portal/api/location/{a_loc.id}/stream") as ws:
|
||||
ws.receive_text()
|
||||
assert exc.value.code == 1008
|
||||
|
||||
Reference in New Issue
Block a user