91 lines
4.0 KiB
Python
91 lines
4.0 KiB
Python
 
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from fakedir import FakeDirectory
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
from pshelpers import ORIGIN_HEADERS, make_settings
|
||
|
|
|
||
|
|
from packetstation.api.origin import allowed
|
||
|
|
from packetstation.main import create_app
|
||
|
|
|
||
|
|
|
||
|
|
def scope(method: str = "POST", path: str = "/api/auth/lock", scheme: str = "https", **headers: str) -> dict[str, Any]:
|
||
|
|
return {
|
||
|
|
"type": "http",
|
||
|
|
"method": method,
|
||
|
|
"path": path,
|
||
|
|
"scheme": scheme,
|
||
|
|
"headers": [(k.replace("_", "-").encode(), v.encode()) for k, v in headers.items()],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_allowed_rules() -> None:
|
||
|
|
good = {"host": "ps.pesco.com", "origin": "https://ps.pesco.com", "x_requested_with": "PacketStation"}
|
||
|
|
assert allowed(scope(**good))
|
||
|
|
assert allowed(scope(method="GET", host="ps.pesco.com")) # safe methods pass
|
||
|
|
assert allowed(scope(path="/healthz", host="x")) # outside /api
|
||
|
|
assert not allowed(scope(**{**good, "origin": "https://evil.example"}))
|
||
|
|
assert not allowed(scope(**{**good, "origin": "http://ps.pesco.com"})) # scheme must match
|
||
|
|
assert not allowed(scope(host="ps.pesco.com", origin="https://ps.pesco.com")) # marker missing
|
||
|
|
assert not allowed(scope(**{**good, "x_requested_with": "XMLHttpRequest"}))
|
||
|
|
no_origin = {"host": "ps.pesco.com", "x_requested_with": "PacketStation"}
|
||
|
|
assert not allowed(scope(**no_origin))
|
||
|
|
assert allowed(scope(**no_origin, referer="https://ps.pesco.com/account"))
|
||
|
|
assert not allowed(scope(**no_origin, referer="https://ps.pesco.com.evil.example/"))
|
||
|
|
for method in ("PUT", "PATCH", "DELETE"):
|
||
|
|
assert not allowed(scope(method=method, host="ps.pesco.com"))
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def client(tmp_path: Path, dev_visual: None) -> TestClient:
|
||
|
|
return TestClient(create_app(make_settings(tmp_path), directory=FakeDirectory()))
|
||
|
|
|
||
|
|
|
||
|
|
def test_wrong_origin_refused_on_signin(client: TestClient) -> None:
|
||
|
|
body = {"login": "clerk.clara", "password": "x"}
|
||
|
|
r = client.post("/api/auth/signin/password", json=body, headers={**ORIGIN_HEADERS, "Origin": "http://evil.example"})
|
||
|
|
assert r.status_code == 403
|
||
|
|
assert r.json() == {"detail": "Cross-origin request refused", "code": "bad_origin"}
|
||
|
|
assert r.headers["x-frame-options"] == "DENY"
|
||
|
|
r = client.post("/api/auth/signin/password", json=body, headers={"Origin": "http://testserver"})
|
||
|
|
assert r.status_code == 403 and r.json()["code"] == "bad_origin"
|
||
|
|
|
||
|
|
|
||
|
|
def test_right_origin_passes_the_check(client: TestClient) -> None:
|
||
|
|
# No database here, so the request gets past the Origin check and stops at the database
|
||
|
|
r = client.post("/api/auth/signin/password", json={"login": "a", "password": "b"}, headers=ORIGIN_HEADERS)
|
||
|
|
assert r.status_code == 503
|
||
|
|
assert r.json()["code"] == "database_unavailable"
|
||
|
|
|
||
|
|
|
||
|
|
def test_me_without_cookie_is_no_session(client: TestClient) -> None:
|
||
|
|
r = client.get("/api/auth/me")
|
||
|
|
assert r.status_code == 401
|
||
|
|
assert r.json()["code"] == "no_session"
|
||
|
|
assert client.get("/api/events").json()["code"] == "no_session"
|
||
|
|
assert client.get("/api/account").status_code == 401
|
||
|
|
|
||
|
|
|
||
|
|
def test_context_without_device(client: TestClient) -> None:
|
||
|
|
r = client.get("/api/auth/context")
|
||
|
|
assert r.status_code == 200
|
||
|
|
assert r.json() == {"device": None, "stack": "ci", "insecure_ldap": False, "pin_digits": {"min": 6, "max": 12}}
|
||
|
|
|
||
|
|
|
||
|
|
def test_validation_error_carries_a_code(client: TestClient) -> None:
|
||
|
|
r = client.post("/api/auth/signin/password", json={"login": "x"}, headers=ORIGIN_HEADERS)
|
||
|
|
assert r.status_code == 422
|
||
|
|
assert r.json()["code"] == "invalid_request"
|
||
|
|
|
||
|
|
|
||
|
|
def test_missing_session_key_fails_cleanly(tmp_path: Path, dev_visual: None) -> None:
|
||
|
|
s = make_settings(tmp_path, session_key_file=tmp_path / "missing")
|
||
|
|
c = TestClient(create_app(s, directory=FakeDirectory()))
|
||
|
|
r = c.post("/api/auth/signin/password", json={"login": "a", "password": "b"}, headers=ORIGIN_HEADERS)
|
||
|
|
assert r.status_code == 503
|
||
|
|
assert r.json()["code"] == "server_misconfigured"
|
||
|
|
assert str(tmp_path) not in r.json()["detail"]
|