Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 131 additions & 2 deletions backend/tests/main_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -21,6 +21,7 @@
)
from wb.homeui_backend.main import (
CUSTOM_MENU_DIRS,
MAX_ID_COOKIE_CANDIDATES,
RequestHandler,
WebRequestHandler,
WebRequestHandlerContext,
Expand All @@ -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,
Expand All @@ -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()
Expand Down
36 changes: 36 additions & 0 deletions backend/tests/users_storage_test.py
Original file line number Diff line number Diff line change
@@ -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)
31 changes: 26 additions & 5 deletions backend/wb/homeui_backend/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion backend/wb/homeui_backend/users_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
7 changes: 7 additions & 0 deletions debian/changelog
Original file line number Diff line number Diff line change
@@ -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 <petr.krasnoshchekov@wirenboard.com> 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
Expand Down