387 lines
12 KiB
Python
387 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import threading
|
|
from collections.abc import Callable
|
|
from http.cookies import SimpleCookie
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
from starlette.requests import Request
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parents[1]
|
|
if str(ROOT_DIR) not in sys.path:
|
|
sys.path.insert(0, str(ROOT_DIR))
|
|
|
|
from dashboard import authentication
|
|
from dashboard.auth_server import create_auth_routes, login
|
|
|
|
|
|
ADMIN_ID = "admin"
|
|
ADMIN_PASSWORD = "correct horse battery staple"
|
|
SIGNING_SECRET = "s" * 64
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _configured_auth(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv(authentication.ADMIN_ID_ENV, ADMIN_ID)
|
|
monkeypatch.setenv(authentication.ADMIN_PASSWORD_ENV, ADMIN_PASSWORD)
|
|
monkeypatch.setenv(authentication.SIGNING_SECRET_ENV, SIGNING_SECRET)
|
|
monkeypatch.setattr(authentication.time, "sleep", lambda _seconds: None)
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> TestClient:
|
|
from starlette.applications import Starlette
|
|
|
|
return TestClient(
|
|
Starlette(routes=create_auth_routes()), base_url="https://testserver"
|
|
)
|
|
|
|
|
|
def _cookie_from(response) -> SimpleCookie[str]:
|
|
cookie: SimpleCookie[str] = SimpleCookie()
|
|
cookie.load(response.headers["set-cookie"])
|
|
return cookie
|
|
|
|
|
|
def test_successful_login_sets_hardened_24_hour_cookie(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
|
|
headers={"origin": "https://testserver"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/"
|
|
header = response.headers["set-cookie"]
|
|
assert f"{authentication.AUTH_COOKIE_NAME}=" in header
|
|
assert "HttpOnly" in header
|
|
assert "Secure" in header
|
|
assert "SameSite=lax" in header
|
|
assert "Path=/" in header
|
|
assert f"Max-Age={authentication.DEFAULT_TOKEN_TTL_SECONDS}" in header
|
|
assert ADMIN_PASSWORD not in header
|
|
assert SIGNING_SECRET not in header
|
|
|
|
|
|
def test_remember_me_sets_30_day_cookie(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={
|
|
"admin_id": ADMIN_ID,
|
|
"admin_password": ADMIN_PASSWORD,
|
|
"remember_me": "on",
|
|
},
|
|
headers={"origin": "https://testserver"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert (
|
|
f"Max-Age={authentication.REMEMBER_ME_TOKEN_TTL_SECONDS}"
|
|
in response.headers["set-cookie"]
|
|
)
|
|
|
|
|
|
def test_unicode_credentials_can_login_through_adapter(client: TestClient, monkeypatch) -> None:
|
|
unicode_id = "管理员"
|
|
unicode_password = "päss🔐"
|
|
monkeypatch.setenv(authentication.ADMIN_ID_ENV, unicode_id)
|
|
monkeypatch.setenv(authentication.ADMIN_PASSWORD_ENV, unicode_password)
|
|
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": unicode_id, "admin_password": unicode_password},
|
|
headers={"origin": "https://testserver"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/"
|
|
assert f"{authentication.AUTH_COOKIE_NAME}=" in response.headers["set-cookie"]
|
|
|
|
|
|
def test_unicode_credential_mismatch_is_generic_and_delayed(
|
|
client: TestClient, monkeypatch
|
|
) -> None:
|
|
import time as real_time
|
|
|
|
unicode_id = "管理员"
|
|
unicode_password = "päss🔐"
|
|
monkeypatch.setenv(authentication.ADMIN_ID_ENV, unicode_id)
|
|
monkeypatch.setenv(authentication.ADMIN_PASSWORD_ENV, unicode_password)
|
|
monkeypatch.setattr(authentication.time, "sleep", real_time.sleep)
|
|
|
|
started = real_time.monotonic()
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": unicode_id, "admin_password": "неверный🔒"},
|
|
headers={"origin": "https://testserver"},
|
|
follow_redirects=False,
|
|
)
|
|
elapsed = real_time.monotonic() - started
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/?auth_error=invalid"
|
|
assert "set-cookie" not in response.headers
|
|
assert elapsed >= authentication.MINIMUM_FAILURE_DELAY_SECONDS * 0.9
|
|
|
|
|
|
def test_invalid_credentials_do_not_set_cookie_and_use_general_error(
|
|
client: TestClient,
|
|
) -> None:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": ADMIN_ID, "admin_password": "wrong"},
|
|
headers={"origin": "https://testserver"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/?auth_error=invalid"
|
|
assert "set-cookie" not in response.headers
|
|
assert "password" not in response.headers["location"]
|
|
assert ADMIN_ID not in response.headers["location"]
|
|
|
|
|
|
def test_login_rejects_malformed_percent_encoded_utf8_without_cookie(
|
|
client: TestClient,
|
|
) -> None:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
content=b"admin_id=%FF&admin_password=wrong",
|
|
headers={
|
|
"origin": "https://testserver",
|
|
"content-type": "application/x-www-form-urlencoded",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "set-cookie" not in response.headers
|
|
|
|
|
|
def test_missing_configuration_fails_closed(client: TestClient, monkeypatch) -> None:
|
|
monkeypatch.delenv(authentication.SIGNING_SECRET_ENV)
|
|
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
|
|
headers={"origin": "https://testserver"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/?auth_error=unavailable"
|
|
assert "set-cookie" not in response.headers
|
|
|
|
|
|
def test_cross_site_login_is_rejected(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
|
|
headers={"origin": "https://attacker.example"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "set-cookie" not in response.headers
|
|
|
|
|
|
def test_different_port_is_not_same_origin(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
|
|
headers={"origin": "https://testserver:444"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "set-cookie" not in response.headers
|
|
|
|
|
|
def test_plain_http_is_rejected_even_with_spoofed_https_origin() -> None:
|
|
from starlette.applications import Starlette
|
|
|
|
with TestClient(
|
|
Starlette(routes=create_auth_routes()), base_url="http://testserver"
|
|
) as http_client:
|
|
response = http_client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
|
|
headers={
|
|
"origin": "https://testserver",
|
|
"x-forwarded-proto": "https",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "set-cookie" not in response.headers
|
|
|
|
|
|
def test_logout_clears_cookie_with_matching_scope(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/auth/logout",
|
|
headers={"origin": "https://testserver"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
header = response.headers["set-cookie"]
|
|
assert f"{authentication.AUTH_COOKIE_NAME}=" in header
|
|
assert "Max-Age=0" in header
|
|
assert "HttpOnly" in header
|
|
assert "Secure" in header
|
|
assert "SameSite=lax" in header
|
|
assert "Path=/" in header
|
|
|
|
|
|
def test_cleanup_route_clears_cookie_and_redirects(client: TestClient) -> None:
|
|
response = client.get(
|
|
"/api/auth/cleanup",
|
|
headers={"referer": "https://testserver/"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/"
|
|
header = response.headers["set-cookie"]
|
|
assert f"{authentication.AUTH_COOKIE_NAME}=" in header
|
|
assert "Max-Age=0" in header
|
|
assert "HttpOnly" in header
|
|
assert "Secure" in header
|
|
assert "SameSite=lax" in header
|
|
assert "Path=/" in header
|
|
|
|
|
|
def test_cleanup_route_rejects_cross_origin_navigation(client: TestClient) -> None:
|
|
response = client.get(
|
|
"/api/auth/cleanup",
|
|
headers={"referer": "https://attacker.example/"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "set-cookie" not in response.headers
|
|
|
|
|
|
def test_cookie_token_round_trip(client: TestClient) -> None:
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
data={"admin_id": ADMIN_ID, "admin_password": ADMIN_PASSWORD},
|
|
headers={"origin": "https://testserver"},
|
|
follow_redirects=False,
|
|
)
|
|
cookie = _cookie_from(response)
|
|
token = cookie[authentication.AUTH_COOKIE_NAME].value
|
|
|
|
assert authentication.verify_persistent_token(
|
|
authentication.load_auth_settings(), token
|
|
) is not None
|
|
|
|
|
|
def _streaming_login_request(
|
|
chunks: list[bytes],
|
|
) -> tuple[Request, Callable[[], int]]:
|
|
messages = [
|
|
{
|
|
"type": "http.request",
|
|
"body": chunk,
|
|
"more_body": index < len(chunks) - 1,
|
|
}
|
|
for index, chunk in enumerate(chunks)
|
|
]
|
|
calls = 0
|
|
|
|
async def receive():
|
|
nonlocal calls
|
|
calls += 1
|
|
return messages.pop(0)
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"method": "POST",
|
|
"scheme": "https",
|
|
"path": "/api/auth/login",
|
|
"raw_path": b"/api/auth/login",
|
|
"query_string": b"",
|
|
"headers": [
|
|
(b"host", b"testserver"),
|
|
(b"origin", b"https://testserver"),
|
|
(b"content-type", b"application/x-www-form-urlencoded"),
|
|
],
|
|
"client": ("testclient", 50000),
|
|
"server": ("testserver", 443),
|
|
}
|
|
return Request(scope, receive), lambda: calls
|
|
|
|
|
|
def test_login_offloads_failed_login_delay_from_event_loop(monkeypatch) -> None:
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
probe_ran = threading.Event()
|
|
probe_before_release = threading.Event()
|
|
|
|
def blocking_delay(*, started_at):
|
|
entered.set()
|
|
if probe_ran.wait(timeout=1):
|
|
probe_before_release.set()
|
|
release.set()
|
|
|
|
monkeypatch.setattr(authentication, "enforce_failed_login_delay", blocking_delay)
|
|
|
|
async def exercise() -> None:
|
|
request, _ = _streaming_login_request([b"admin_id=admin&admin_password=wrong"])
|
|
task = asyncio.create_task(login(request))
|
|
|
|
async def probe() -> None:
|
|
probe_ran.set()
|
|
|
|
asyncio.create_task(probe())
|
|
assert await asyncio.to_thread(entered.wait, 1)
|
|
release.set()
|
|
response = await task
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/?auth_error=invalid"
|
|
assert "set-cookie" not in response.headers
|
|
|
|
asyncio.run(exercise())
|
|
assert probe_before_release.is_set()
|
|
|
|
|
|
def test_login_rejects_exactly_one_byte_over_streaming_limit(monkeypatch) -> None:
|
|
def unexpected_verification(*_args, **_kwargs):
|
|
raise AssertionError("oversized body reached credential verification")
|
|
|
|
monkeypatch.setattr(authentication, "verify_credentials", unexpected_verification)
|
|
request, calls = _streaming_login_request([b"x" * 4097])
|
|
response = asyncio.run(login(request))
|
|
|
|
assert response.status_code == 400
|
|
assert "set-cookie" not in response.headers
|
|
assert calls() == 1
|
|
|
|
|
|
def test_login_accepts_body_at_streaming_limit() -> None:
|
|
prefix = b"admin_id=admin&admin_password="
|
|
body = prefix + (b"x" * (4096 - len(prefix)))
|
|
assert len(body) == 4096
|
|
request, calls = _streaming_login_request([body])
|
|
response = asyncio.run(login(request))
|
|
|
|
assert response.status_code == 303
|
|
assert calls() == 1
|
|
|
|
|
|
def test_login_stops_reading_stream_after_limit_is_exceeded() -> None:
|
|
request, calls = _streaming_login_request([b"x" * 4096, b"y", b"z"])
|
|
response = asyncio.run(login(request))
|
|
|
|
assert response.status_code == 400
|
|
assert calls() == 2
|