219 lines
9.0 KiB
Python
219 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
from arr_web.app import PortalApplication, RuntimeHealth
|
|
from arr_web.auth import LoginAttemptLedger, LoginCredentials
|
|
|
|
|
|
USERNAME = "finance-operator"
|
|
PASSWORD = "correct-horse-battery-staple"
|
|
CREDENTIALS = LoginCredentials(USERNAME, PASSWORD)
|
|
STATIC_ROOT = Path(__file__).resolve().parents[1] / "arr_web" / "static"
|
|
|
|
|
|
def decoded(response: Any) -> Dict[str, Any]:
|
|
return json.loads(response.body.decode("utf-8"))
|
|
|
|
|
|
def login(
|
|
app: PortalApplication,
|
|
*,
|
|
username: object = USERNAME,
|
|
password: object = PASSWORD,
|
|
client_id: str = "test-client",
|
|
) -> Any:
|
|
return app.handle(
|
|
"POST",
|
|
"/api/login",
|
|
{"Content-Type": "application/json"},
|
|
json.dumps({"username": username, "password": password}).encode("utf-8"),
|
|
client_id=client_id,
|
|
)
|
|
|
|
|
|
def auth_headers(response: Any) -> Dict[str, str]:
|
|
payload = decoded(response)
|
|
return {
|
|
"Cookie": response.headers["Set-Cookie"].split(";", 1)[0],
|
|
"X-ARR-CSRF": payload["data"]["csrf_token"],
|
|
}
|
|
|
|
|
|
class LoginCredentialTests(unittest.TestCase):
|
|
def test_environment_configuration_is_required_and_validated(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "requires ARR_WEB_USERNAME"):
|
|
LoginCredentials.from_environment({})
|
|
with self.assertRaisesRegex(ValueError, "12"):
|
|
LoginCredentials.from_environment(
|
|
{"ARR_WEB_USERNAME": USERNAME, "ARR_WEB_PASSWORD": "short"}
|
|
)
|
|
configured = LoginCredentials.from_environment(
|
|
{"ARR_WEB_USERNAME": USERNAME, "ARR_WEB_PASSWORD": PASSWORD}
|
|
)
|
|
self.assertTrue(configured.verify(USERNAME, PASSWORD))
|
|
self.assertFalse(configured.verify(USERNAME, "wrong-password-value"))
|
|
self.assertFalse(configured.verify("other-operator", PASSWORD))
|
|
|
|
|
|
class PortalAuthenticationTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = PortalApplication(
|
|
credentials=CREDENTIALS,
|
|
health=RuntimeHealth(True, True, False),
|
|
)
|
|
|
|
def test_only_login_assets_and_minimal_readiness_are_public(self) -> None:
|
|
desktop = self.app.handle("GET", "/", {})
|
|
mobile = self.app.handle("GET", "/h5", {})
|
|
protected_api = self.app.handle("GET", "/api/health", {})
|
|
protected_asset = self.app.handle("GET", "/assets/app.js", {})
|
|
login_page = self.app.handle("GET", "/login", {})
|
|
login_css = self.app.handle("GET", "/assets/login.css", {})
|
|
readiness = self.app.handle("GET", "/healthz", {})
|
|
|
|
self.assertEqual((desktop.status, desktop.headers["Location"]), (303, "/login?next=%2F"))
|
|
self.assertEqual((mobile.status, mobile.headers["Location"]), (303, "/login?next=%2Fh5"))
|
|
self.assertEqual(protected_api.status, 401)
|
|
self.assertEqual(decoded(protected_api)["error"]["code"], "AUTH_REQUIRED")
|
|
self.assertEqual(protected_asset.status, 303)
|
|
self.assertEqual(login_page.status, 200)
|
|
self.assertIn('<h1 id="context-title">ARR Report</h1>', login_page.body.decode("utf-8"))
|
|
self.assertEqual(login_css.status, 200)
|
|
self.assertEqual((readiness.status, readiness.body), (200, b"ready\n"))
|
|
|
|
unavailable = PortalApplication(
|
|
credentials=CREDENTIALS,
|
|
health=RuntimeHealth(True, False, False),
|
|
).handle("GET", "/healthz", {})
|
|
self.assertEqual((unavailable.status, unavailable.body), (503, b"unavailable\n"))
|
|
|
|
def test_login_is_strict_generic_and_issues_an_authenticated_session(self) -> None:
|
|
wrong_media = self.app.handle(
|
|
"POST",
|
|
"/api/login",
|
|
{"Content-Type": "application/x-www-form-urlencoded"},
|
|
b"username=x&password=y",
|
|
)
|
|
wrong = login(self.app, password="incorrect-password-value")
|
|
success = login(self.app)
|
|
headers = auth_headers(success)
|
|
session = self.app.handle("GET", "/api/session", headers)
|
|
|
|
self.assertEqual(wrong_media.status, 415)
|
|
self.assertEqual(wrong.status, 401)
|
|
self.assertEqual(decoded(wrong)["error"]["code"], "LOGIN_FAILED")
|
|
self.assertNotIn("incorrect-password-value", wrong.body.decode("utf-8"))
|
|
self.assertNotIn("Set-Cookie", wrong.headers)
|
|
self.assertEqual(success.status, 200)
|
|
self.assertIn("HttpOnly", success.headers["Set-Cookie"])
|
|
self.assertIn("SameSite=Strict", success.headers["Set-Cookie"])
|
|
self.assertEqual(decoded(session)["data"]["username"], USERNAME)
|
|
self.assertEqual(
|
|
decoded(session)["data"]["csrf_token"],
|
|
headers["X-ARR-CSRF"],
|
|
)
|
|
|
|
def test_authenticated_login_page_redirect_is_allowlisted(self) -> None:
|
|
headers = auth_headers(login(self.app))
|
|
mobile = self.app.handle("GET", "/login?next=/h5", headers)
|
|
external = self.app.handle(
|
|
"GET",
|
|
"/login?next=https%3A%2F%2Fevil.example",
|
|
headers,
|
|
)
|
|
duplicate = self.app.handle("GET", "/login?next=/h5&next=/", headers)
|
|
|
|
self.assertEqual(mobile.headers["Location"], "/h5")
|
|
self.assertEqual(external.headers["Location"], "/")
|
|
self.assertEqual(duplicate.headers["Location"], "/")
|
|
|
|
def test_logout_requires_csrf_revokes_session_and_expires_cookie(self) -> None:
|
|
headers = auth_headers(login(self.app))
|
|
no_csrf = self.app.handle(
|
|
"POST",
|
|
"/api/logout",
|
|
{"Cookie": headers["Cookie"]},
|
|
)
|
|
logged_out = self.app.handle("POST", "/api/logout", headers)
|
|
after = self.app.handle("GET", "/api/session", headers)
|
|
|
|
self.assertEqual(no_csrf.status, 403)
|
|
self.assertEqual(logged_out.status, 200)
|
|
self.assertIn("Max-Age=0", logged_out.headers["Set-Cookie"])
|
|
self.assertEqual(after.status, 401)
|
|
|
|
def test_login_attempts_are_temporarily_rate_limited_per_client(self) -> None:
|
|
now = [100.0]
|
|
attempts = LoginAttemptLedger(
|
|
max_failures=2,
|
|
window_seconds=60,
|
|
block_seconds=30,
|
|
clock=lambda: now[0],
|
|
)
|
|
app = PortalApplication(credentials=CREDENTIALS, login_attempts=attempts)
|
|
|
|
first = login(app, password="incorrect-password-value", client_id="client-a")
|
|
second = login(app, password="incorrect-password-value", client_id="client-a")
|
|
blocked_correct = login(app, client_id="client-a")
|
|
other_client = login(app, client_id="client-b")
|
|
now[0] += 31
|
|
recovered = login(app, client_id="client-a")
|
|
|
|
self.assertEqual(first.status, 401)
|
|
self.assertEqual(second.status, 429)
|
|
self.assertEqual(second.headers["Retry-After"], "30")
|
|
self.assertEqual(blocked_correct.status, 429)
|
|
self.assertEqual(other_client.status, 200)
|
|
self.assertEqual(recovered.status, 200)
|
|
|
|
|
|
class StaticLoginContractTests(unittest.TestCase):
|
|
def test_login_and_logout_assets_keep_accessibility_and_session_contracts(self) -> None:
|
|
login_html = (STATIC_ROOT / "login.html").read_text(encoding="utf-8")
|
|
login_css = (STATIC_ROOT / "login.css").read_text(encoding="utf-8")
|
|
login_js = (STATIC_ROOT / "login.js").read_text(encoding="utf-8")
|
|
desktop_html = (STATIC_ROOT / "index.html").read_text(encoding="utf-8")
|
|
desktop_js = (STATIC_ROOT / "app.js").read_text(encoding="utf-8")
|
|
mobile_html = (STATIC_ROOT / "h5.html").read_text(encoding="utf-8")
|
|
mobile_js = (STATIC_ROOT / "h5.js").read_text(encoding="utf-8")
|
|
|
|
for expected in (
|
|
'<label for="username">username</label>',
|
|
'autocomplete="username"',
|
|
'<label for="password">password</label>',
|
|
'autocomplete="current-password"',
|
|
'id="password-toggle"',
|
|
'aria-live="assertive"',
|
|
):
|
|
self.assertIn(expected, login_html)
|
|
for removed in (
|
|
"OPERA ARR WORKFLOW",
|
|
"让每一次到店数据",
|
|
"上传 XML",
|
|
"AUTHORIZED ACCESS",
|
|
"欢迎回来",
|
|
"无法登录？",
|
|
"仅限获得授权的 Finance 操作人员使用",
|
|
):
|
|
self.assertNotIn(removed, login_html)
|
|
self.assertIn("@media (prefers-reduced-motion: reduce)", login_css)
|
|
self.assertIn('candidate === "/h5" ? "/h5" : "/"', login_js)
|
|
self.assertIn('id="logout-button"', desktop_html)
|
|
self.assertIn('id="h5-logout"', mobile_html)
|
|
self.assertIn('id="logout-button" type="button" disabled hidden', desktop_html)
|
|
self.assertIn('id="h5-logout" type="button" disabled hidden', mobile_html)
|
|
self.assertIn("if (!session.username) return", desktop_js)
|
|
self.assertIn("if (session.username)", mobile_js)
|
|
self.assertIn("response.status === 401", desktop_js)
|
|
self.assertIn('api("/api/logout", { method: "POST" })', desktop_js)
|
|
self.assertIn("response.status === 401", mobile_js)
|
|
self.assertIn('api("/api/logout", { method: "POST" })', mobile_js)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|