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_h5_shell_is_public_while_desktop_and_detail_routes_stay_private(self) -> None: desktop = self.app.handle("GET", "/", {}) mobile = self.app.handle("GET", "/h5", {}) mobile_css = self.app.handle("GET", "/assets/h5.css", {}) mobile_js = self.app.handle("GET", "/assets/h5.js", {}) 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, 200) self.assertEqual(mobile_css.status, 200) self.assertEqual(mobile_js.status, 200) 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('

ARR Report

', 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 ( '', 'autocomplete="username"', '', '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('