diff --git a/backend/tests/main_test.py b/backend/tests/main_test.py index c2bbd5c2e..6d5fe0b8c 100644 --- a/backend/tests/main_test.py +++ b/backend/tests/main_test.py @@ -3,9 +3,9 @@ import shutil import tempfile import unittest -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler -from unittest.mock import MagicMock, mock_open, patch +from unittest.mock import MagicMock, call, mock_open, patch from wb.homeui_backend.cert import CertificateState from wb.homeui_backend.gates import CUSTOM_MENU_DIR as GATES_CUSTOM_MENU_DIR @@ -21,6 +21,7 @@ ) from wb.homeui_backend.main import ( CUSTOM_MENU_DIRS, + MAX_ID_COOKIE_CANDIDATES, RequestHandler, WebRequestHandler, WebRequestHandlerContext, @@ -29,7 +30,9 @@ custom_menu_handler, delete_user_handler, device_info_handler, + get_id_cookie_values, get_required_user_type, + get_session, get_users_handler, make_certificate_usable_change_handler, security_check_handler, @@ -42,6 +45,132 @@ from wb.homeui_backend.users_storage import User, UsersStorage, UserType +class GetIdCookieValuesTest(unittest.TestCase): + def test_header_parsing(self): + """Foreign cookies planted by apps on other ports of the same host (JSON-valued, + named like reserved attributes, duplicates of "id") must not hide our session + cookie: every distinct "id" value is collected in header order, the name is + matched case-sensitively, a value keeps everything after the first "=" (base64 + padding survives), and parts without "=" are skipped.""" + for header, expected in ( + ("id=abc", ["abc"]), + ('foo={"a": 1}; id=abc', ["abc"]), + ("path=/; expires=Sat, 01 Jan 2028 00:00:00 GMT; id=abc", ["abc"]), + ("id=stale; id=abc", ["stale", "abc"]), + ("id=abc; id=abc; id=other", ["abc", "other"]), + ("other=1; theme=dark", []), + ("", []), + ("ID=abc", []), + ("id=dG9rZW4=; other=1", ["dG9rZW4="]), + ("garbage; id=abc", ["abc"]), + ): + with self.subTest(header=header): + self.assertEqual(get_id_cookie_values(header), expected) + + def test_candidate_cap(self): + """More "id" values than MAX_ID_COOKIE_CANDIDATES (an attacker-shaped header — + real browsers send a couple at most) are cut to the first ones; duplicates are + dropped before the cap so repeats cannot consume it.""" + header = "; ".join(f"id=c{i}" for i in range(MAX_ID_COOKIE_CANDIDATES + 2)) + self.assertEqual( + get_id_cookie_values("id=c0; " + header), + [f"c{i}" for i in range(MAX_ID_COOKIE_CANDIDATES)], + ) + + +class GetSessionTest(unittest.TestCase): + def setUp(self): + self.request = MagicMock(spec=BaseHTTPRequestHandler) + self.users_storage_mock = MagicMock(spec=UsersStorage) + self.sessions_storage_mock = MagicMock(spec=SessionsStorage) + self.session = Session( + "valid", User("1", "user1", "password1", UserType.USER, False), datetime.now(timezone.utc) + ) + + def _get_session(self): + return get_session(self.request, self.users_storage_mock, self.sessions_storage_mock) + + def test_id_after_json_foreign_cookie_resolves_session(self): + """A JSON-valued foreign cookie before "id" (the SimpleCookie parse-abort case) + must not prevent the session from being resolved.""" + self.request.headers = {"Cookie": 'foo={"a": 1}; id=valid'} + self.sessions_storage_mock.get_session_by_id.return_value = self.session + self.assertIs(self._get_session(), self.session) + self.sessions_storage_mock.get_session_by_id.assert_called_once_with("valid", self.users_storage_mock) + + def test_duplicate_id_stale_first_tries_candidates_in_order(self): + """With two "id" cookies where a stale/foreign value comes first, both candidates + are looked up in header order and the second (resolving) one wins.""" + self.request.headers = {"Cookie": "id=stale; id=valid"} + self.sessions_storage_mock.get_session_by_id.side_effect = [None, self.session] + self.assertIs(self._get_session(), self.session) + self.assertEqual( + self.sessions_storage_mock.get_session_by_id.call_args_list, + [call("stale", self.users_storage_mock), call("valid", self.users_storage_mock)], + ) + + def test_duplicate_id_valid_first_wins_without_trying_the_rest(self): + """With the resolving "id" first, the lookup stops there and the garbage duplicate + is never tried.""" + self.request.headers = {"Cookie": "id=valid; id=garbage"} + self.sessions_storage_mock.get_session_by_id.side_effect = [self.session] + self.assertIs(self._get_session(), self.session) + self.sessions_storage_mock.get_session_by_id.assert_called_once_with("valid", self.users_storage_mock) + + def test_no_cookie_header_returns_none(self): + self.request.headers = {} + self.assertIsNone(self._get_session()) + self.request.log_error.assert_called_once_with("Cookie not found") + self.sessions_storage_mock.get_session_by_id.assert_not_called() + + def test_candidates_but_no_session_returns_none(self): + self.request.headers = {"Cookie": "id=unknown"} + self.sessions_storage_mock.get_session_by_id.return_value = None + self.assertIsNone(self._get_session()) + self.request.log_error.assert_called_once_with("Session not found") + + def test_expired_admin_session_is_rejected(self): + """An ADMIN session idle for more than 14 days resolves from the storage but is + rejected ("Cookie expired").""" + admin_session = Session( + "valid", + User("1", "admin", "password1", UserType.ADMIN, False), + datetime.now(timezone.utc) - timedelta(days=15), + ) + self.request.headers = {"Cookie": "id=valid"} + self.sessions_storage_mock.get_session_by_id.return_value = admin_session + self.assertIsNone(self._get_session()) + self.request.log_error.assert_called_once_with("Cookie expired") + + def test_fresh_admin_session_is_accepted(self): + """An ADMIN session idle for less (13 days) than the 14-day ADMIN_COOKIE_LIFETIME + passes the expiry check and is returned.""" + admin_session = Session( + "valid", + User("1", "admin", "password1", UserType.ADMIN, False), + datetime.now(timezone.utc) - timedelta(days=13), + ) + self.request.headers = {"Cookie": "id=valid"} + self.sessions_storage_mock.get_session_by_id.return_value = admin_session + self.assertIs(self._get_session(), admin_session) + + def test_expired_admin_first_candidate_is_not_skipped(self): + """Contract pin: the loop stops at the first candidate that resolves to a stored + session, and the 14-day check then rejects it without falling back to later + candidates — a second id that also resolves against this store is not a state a + real browser produces, so the simpler rule stands.""" + admin_session = Session( + "stale", + User("1", "admin", "password1", UserType.ADMIN, False), + datetime.now(timezone.utc) - timedelta(days=15), + ) + self.request.headers = {"Cookie": "id=stale; id=valid"} + self.sessions_storage_mock.get_session_by_id.return_value = admin_session + self.assertIsNone(self._get_session()) + self.sessions_storage_mock.get_session_by_id.assert_called_once_with("stale", self.users_storage_mock) + self.request.log_error.assert_called_once_with("Cookie expired") + + class DeleteUserHandlerTest(unittest.TestCase): def setUp(self): self.request = MagicMock() diff --git a/backend/tests/users_storage_test.py b/backend/tests/users_storage_test.py new file mode 100644 index 000000000..4e4d01be6 --- /dev/null +++ b/backend/tests/users_storage_test.py @@ -0,0 +1,36 @@ +import sqlite3 +import unittest + +from wb.homeui_backend.db import create_tables +from wb.homeui_backend.users_storage import User, UsersStorage, UserType + + +class GetAutologinUserTest(unittest.TestCase): + def setUp(self): + self.connection = sqlite3.connect(":memory:") + create_tables(self.connection) + self.storage = UsersStorage(self.connection) + + def tearDown(self): + self.connection.close() + + def test_returns_the_autologin_user(self): + self.storage.add_user(User("", "kiosk", "hash", UserType.USER, True)) + self.assertEqual(self.storage.get_autologin_user().login, "kiosk") + + def test_no_autologin_user(self): + self.storage.add_user(User("", "admin", "hash", UserType.ADMIN, False)) + self.assertIsNone(self.storage.get_autologin_user()) + + def test_stale_autologin_row_of_a_non_user_account_is_returned_with_the_flag_off(self): + """A row written before the User.type-setter fix can still hold autologin=1 for an + operator/admin account. The row is still returned, only the model normalises the + flag to off.""" + self.connection.execute( + "INSERT INTO users (user_id, login, pwd_hash, type, autologin) VALUES (?, ?, ?, ?, ?)", + ("1", "op", "hash", UserType.OPERATOR.value, 1), + ) + self.connection.commit() + user = self.storage.get_autologin_user() + self.assertEqual(user.login, "op") + self.assertFalse(user.autologin) diff --git a/backend/wb/homeui_backend/main.py b/backend/wb/homeui_backend/main.py index 009cec3a3..9255c5a2e 100644 --- a/backend/wb/homeui_backend/main.py +++ b/backend/wb/homeui_backend/main.py @@ -91,17 +91,38 @@ def make_set_cookie_header(cookie: cookies.SimpleCookie) -> list[str]: return ["Set-Cookie", cookie.output(header="")] +MAX_ID_COOKIE_CANDIDATES = 8 + + +def get_id_cookie_values(cookie_header: str) -> list[str]: + """Unique values of the cookies named exactly "id", in header order. + + Parsed manually because http.cookies.SimpleCookie silently stops parsing at the first + foreign cookie it cannot match — JSON-valued cookies or cookies named like reserved + attributes ("path", "expires", ...) — and apps on other ports of the same host share + the browser's cookie jar, so such cookies do reach us and must not hide our "id". + """ + values = [] + for part in cookie_header.split(";"): + name, sep, value = part.partition("=") + if sep and name.strip() == "id": + values.append(value.strip()) + return list(dict.fromkeys(values))[:MAX_ID_COOKIE_CANDIDATES] + + def get_session( request: BaseHTTPRequestHandler, users_storage: UsersStorage, sessions_storage: SessionsStorage ) -> Optional[Session]: try: - request_cookie = cookies.SimpleCookie() - request_cookie.load(request.headers.get("Cookie", "")) - cookie_id = request_cookie.get("id") - if cookie_id is None: + candidates = get_id_cookie_values(request.headers.get("Cookie", "")) + if not candidates: request.log_error("Cookie not found") return None - session = sessions_storage.get_session_by_id(cookie_id.value, users_storage) + session = None + for candidate in candidates: + session = sessions_storage.get_session_by_id(candidate, users_storage) + if session is not None: + break if session is None: request.log_error("Session not found") return None diff --git a/backend/wb/homeui_backend/users_storage.py b/backend/wb/homeui_backend/users_storage.py index 5507f960f..871798de0 100644 --- a/backend/wb/homeui_backend/users_storage.py +++ b/backend/wb/homeui_backend/users_storage.py @@ -43,7 +43,7 @@ def type(self) -> UserType: def type(self, value: UserType): self._type = value if not self.supports_autologin(): - self.autologin = False + self._autologin = False @property def autologin(self) -> bool: diff --git a/debian/changelog b/debian/changelog index e9a885380..a05deb5af 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,10 @@ +wb-mqtt-homeui (2.255.1) stable; urgency=medium + + * Fix session cookie parsing when foreign cookies are present + * Clear autologin when a user's type changes + + -- Petr Krasnoshchekov Thu, 10 Sep 2026 12:10:00 +0500 + wb-mqtt-homeui (2.255.0) stable; urgency=medium * Add slider editor for numeric config parameters with range format