From a13b649bec2ab6e87bc106846d2052aadf265790 Mon Sep 17 00:00:00 2001 From: szeka9 Date: Tue, 21 Jul 2026 23:03:36 +0200 Subject: [PATCH 1/5] Support HTTP Basic Authentication Add authentication and authorization module as an extension of the HTTP state machine. Add authorization capability based on stored user data and role definitions. Role definitions allow glob-based pattern matching with additive role-based acccess control (RBAC). Introduce additional configuration parameter to select the authentication method. --- .gitignore | 2 + Makefile | 9 +- src/pyrobusta/protocol/http.py | 14 + src/pyrobusta/protocol/http_basic_auth.py | 339 ++++++++++++++++++++++ src/pyrobusta/utils/config.py | 11 +- src/pyrobusta/utils/helpers.py | 19 ++ 6 files changed, 387 insertions(+), 7 deletions(-) create mode 100644 src/pyrobusta/protocol/http_basic_auth.py diff --git a/.gitignore b/.gitignore index 7edd53f..384ec00 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,8 @@ venv/ __pycache__ pyrobusta.env +pyrobusta.passwd +pyrobusta.roles build/ runtime/ runtime-test/ diff --git a/Makefile b/Makefile index 168d012..8203ae7 100644 --- a/Makefile +++ b/Makefile @@ -100,7 +100,10 @@ deploy-config: @echo "Uploading pyrobusta.env" @mpremote $(DEVICE) soft-reset @if [ -f pyrobusta.env ]; then mpremote $(DEVICE) cp pyrobusta.env :pyrobusta.env; fi + @if [ -f pyrobusta.roles ]; then mpremote $(DEVICE) cp pyrobusta.roles :pyrobusta.roles; fi + @if [ -f pyrobusta.passwd ]; then mpremote $(DEVICE) cp pyrobusta.passwd :pyrobusta.passwd; fi @mpremote $(DEVICE) reset + @sleep 5 # ----------------------------- @@ -191,6 +194,8 @@ stage-app: @cp $(TLS_DIR)/key.der $(RUNTIME_DIR)/ @if [ -f pyrobusta.env ]; then cp pyrobusta.env $(RUNTIME_DIR)/; fi + @if [ -f pyrobusta.roles ]; then cp pyrobusta.roles $(RUNTIME_DIR)/; fi + @if [ -f pyrobusta.passwd ]; then cp pyrobusta.passwd $(RUNTIME_DIR)/; fi @echo "http_port=8080" >> $(RUNTIME_DIR)/pyrobusta.env @echo "https_port=4443" >> $(RUNTIME_DIR)/pyrobusta.env @@ -212,9 +217,7 @@ deploy-app: mpremote $(DEVICE) cp $(APP_DIR)/boot.py :boot.py mpremote $(DEVICE) cp $(APP_DIR)/app.py :app.py - @echo "Uploading pyrobusta.env" - @if [ -f pyrobusta.env ]; then mpremote $(DEVICE) cp pyrobusta.env :pyrobusta.env; fi - @mpremote $(DEVICE) reset + $(MAKE) deploy-config @echo "\e[32m$(APP_DIR) app is successfully deployed, \n"\ "run 'make DEVICE=$(DEVICE) run-device' to restart the device and check the output.\e[0m" diff --git a/src/pyrobusta/protocol/http.py b/src/pyrobusta/protocol/http.py index f58a1ad..225550f 100644 --- a/src/pyrobusta/protocol/http.py +++ b/src/pyrobusta/protocol/http.py @@ -12,6 +12,7 @@ CONF_HTTP_MULTIPART, CONF_HTTP_FILES_API, CONF_HTTP_SERVED_PATHS, + CONF_HTTP_AUTH, ) from ..utils import logging, helpers from ..stream.buffer import BufferFullError @@ -49,6 +50,7 @@ class HttpEngine: - http_files_api: serve files at the /files API, with support for uploads, removal and directory listing - http_multipart: support for multipart requests/responses + - http_auth: authenticate and authorize users """ __slots__ = ( @@ -79,6 +81,8 @@ class HttpEngine: b"204 No Content", 400, b"400 Bad Request", + 401, + b"401 Unauthorized", 403, b"403 Forbidden", 404, @@ -692,6 +696,11 @@ def _parse_headers_st(self, rx): if self.version == b"HTTP/1.1" and "host" not in self.headers: raise InvalidHeaders() rx.consume(blank_idx + 4) + self.state = self._handle_auth_st + + def _handle_auth_st(self, _): + # This a placeholder for authentication & authorization + # purposes to be overridden by extensions. self.state = self._route_request_st def _route_request_st(self, _): @@ -904,3 +913,8 @@ def enable_optional_features(): from pyrobusta.protocol import http_file_server http_file_server.apply_patches() + + if get_config(CONF_HTTP_AUTH) == "basic": + from pyrobusta.protocol import http_basic_auth + + http_basic_auth.apply_patches() diff --git a/src/pyrobusta/protocol/http_basic_auth.py b/src/pyrobusta/protocol/http_basic_auth.py new file mode 100644 index 0000000..327dc8d --- /dev/null +++ b/src/pyrobusta/protocol/http_basic_auth.py @@ -0,0 +1,339 @@ +""" +Module for HTTP Basic Authentication. + +This module overrides the auth placeholder HttpEngine._handle_auth_st(), +and applies the basic authentication scheme, with hash-based password storage, +and role-based authorization. +""" + +# pylint: disable=W0212,R0401 + +import hashlib +import gc +import binascii + +from pyrobusta.protocol import http +from pyrobusta.utils.patch import add_method +from pyrobusta.utils.helpers import iterate_segments + +_PASSWD_LOCATION = "pyrobusta.passwd" +_ROLES_LOCATION = "pyrobusta.roles" + +_MAX_ROLES = 32 +_NO_POLICY = 2**_MAX_ROLES +_ALL_ROLES = 2**_MAX_ROLES - 1 + +_ROLE_INDEX = {} +_USERS = {} # {"user": [b"", b""]} +_ATTR_TREE = None # root node: / + +_DUMMY_HASH = hashlib.sha256(b"invalid-user").digest() + + +class AttributeNode: + """ + Tree-based data structure for authorization. + The tree represents predefined authorization masks defined + for different segments of URL paths. + """ + + __slots__ = ("name", "children", "attributes") + + def __init__(self, name: str): + self.name = name + self.children = [] + self.attributes = None + + def add_child(self, node): + """ + Add a child node to the current node. + """ + self.children.append(node) + + def iter_tree(self, path: str, glob: bool = False): + """ + Generator for iterating over the nodes of the tree + on a predefined path. + """ + current_node = self + for segment in iterate_segments(path, "/"): + segment = segment.lower() + if not segment: + continue + child_node = None + while child_node is None: + for child in current_node.children: + if child.name == segment: + child_node = child + break + if glob and child_node is None: + for child in current_node.children: + if child.name == "*": + child_node = child + break + if child_node is None: + # Yield (None, segment) until the caller inserts the + # missing child, or handles this condition in another way. + # Once the child exists, resume traversal automatically. + yield None, segment + yield child_node, segment + current_node = child_node + + def insert_path(self, path: str, attributes: dict): + """ + Introduce missing nodes in the tree based on + a specific path with attributes defined for the + last node corresponding to the trailing path segment. + """ + if not path.startswith("/") or path.find("/**/") != -1: + raise ValueError() + current_node = self + for iter_node, segment in self.iter_tree(path, glob=False): + if not iter_node: + n = AttributeNode(segment) + current_node.add_child(n) + else: + current_node = iter_node + current_node.attributes = attributes.copy() + + def get_attributes(self, path: str): + """ + Retrieve the attributes of a node, corresponding + to the last segment of a path. Use glob-based rules + if nodes of trailing path segments are missing. + - when a single path segment is missing: resolve attributes from '*' node + - when multiple trailing path segments are missing: resolve attributes from '**' node + """ + current_node = self + + # Calculate number of segments for globbing + num_segments = 0 + prev = "/" + for c in path: + if prev == "/" and c != "/": + num_segments += 1 + prev = c + + # Iterate nodes + parent_glob = ( + None # Track parent attributes corresponding to recursive globs (**) + ) + num_nodes = 0 + + for iter_node, _ in self.iter_tree(path, glob=True): + num_nodes += 1 + glob = None + + for child in current_node.children: + if num_nodes == num_segments and child.name == "*": + glob = child + if child.name == "**": + parent_glob = child + + current_node = iter_node + if not iter_node: + break + + if not current_node: + if glob: + return glob.attributes + if parent_glob: + return parent_glob.attributes + return + return current_node.attributes + + +def index_roles(roles: str): + """ + Create a common index for roles defined in user + definitions and role definitions, and assign a binary + mask to each role, used for authorization. + """ + role_mask = 0b0 + for role in iterate_segments(roles, ","): + role = role.strip().lower() + if not role: + continue + if role not in _ROLE_INDEX and role != "*": + if len(_ROLE_INDEX) == _MAX_ROLES: + raise ValueError() + _ROLE_INDEX[role] = len(_ROLE_INDEX) + if role == "*": + role_mask = _NO_POLICY + else: + role_mask |= 1 << _ROLE_INDEX[role] + return role_mask + + +def _load_users(config=_PASSWD_LOCATION): + with open(config, encoding="utf-8") as users: + for line in users: + comment_idx = line.find("#") + line = line[:comment_idx] if comment_idx != -1 else line + if not line: + continue + if not line.count(":") == 2: + raise ValueError() + user_sep = line.find(":") + password_sep = line.find(":", user_sep + 1) + user = line[:user_sep].strip().lower() + password_hash = line[user_sep + 1 : password_sep].strip() + roles = line[password_sep + 1 :] + role_mask = index_roles(roles) + if not user or not password_hash: + raise ValueError() + if user in _USERS: + raise ValueError() + _USERS[user] = [bytes.fromhex(password_hash), role_mask] + + +def _load_roles(config=_ROLES_LOCATION): + with open(config, encoding="utf-8") as roles: + paths = [] + attributes = {} + for line in roles: + comment_idx = line.find("#") + line = line[:comment_idx] if comment_idx != -1 else line + + # Parse URL path + if line.startswith("/"): + if paths and attributes: + for path in paths: + _ATTR_TREE.insert_path(path, attributes) + paths = [line] + attributes = {} + else: + paths.append(line) + + # Parse attributes + elif line.strip(): + if not paths: + raise ValueError() + sep = line.find(":") + if sep in (0, -1): + raise ValueError() + role_mask = index_roles(line[sep + 1 :]) + for attr in iterate_segments(line[0:sep].strip(), ","): + if attr in attributes: + raise ValueError() + if attr: + attributes[attr] = role_mask + for path in paths: + _ATTR_TREE.insert_path(path, attributes) + + +def _handle_auth_st(self, _): + # Determine security policy + is_public = False + method = self.method.decode("ascii") + url = self.url.decode("ascii") + + policy = _ATTR_TREE.get_attributes(url) + if not policy: + self.state = self._handle_auth_header_st + return + + if method not in policy: + if policy.get("*") == _NO_POLICY: + is_public = True + elif policy[method] == _NO_POLICY: + is_public = True + + if is_public: + self.state = self._route_request_st + else: + self.state = self._handle_auth_header_st + + +def compare_digest(a, b): + """ + Constant time comparison of hash digests to prevent timing attacks. + """ + if len(a) != len(b): + return False + diff = 0 + for x, y in zip(a, b): + diff |= x ^ y + return diff == 0 + + +def _handle_auth_header_st(self, _): + method = self.method.decode("ascii") + url = self.url.decode("ascii") + auth_header = self.headers.get("authorization", "").strip() + www_auth_header = b"WWW-Authenticate" + www_auth_method = b'Basic realm="Device"' + + # Protocol validation + if not auth_header or auth_header.strip()[:5].lower() != "basic": + self.set_response_header(www_auth_header, www_auth_method) + self.terminate(401) + return + + # Decoding + user_data = auth_header[5:].strip() + try: + user_data = binascii.a2b_base64(user_data).decode() + except binascii.Error: + self.set_response_header(www_auth_header, www_auth_method) + self.terminate(401) + return + + # Authentication + user_sep = user_data.find(":") + if user_sep < 0: + self.set_response_header(www_auth_header, www_auth_method) + self.terminate(401) + return + + username = user_data[:user_sep].lower() + user_info = _USERS.get(username) + stored_hash = user_info[0] if user_info else _DUMMY_HASH + password_hash = hashlib.sha256( + user_data[user_sep + 1 :].strip().encode("ascii") + ).digest() + + hash_ok = compare_digest(password_hash, stored_hash) + user_ok = user_info is not None + + if not (user_ok and hash_ok): + self.set_response_header(www_auth_header, www_auth_method) + self.terminate(401) + return + + # Authorization + policy = _ATTR_TREE.get_attributes(url) + + if not policy: + allowed_roles = 0 + elif method not in policy: + allowed_roles = policy.get("*", 0) + else: + allowed_roles = policy[method] + + if (allowed_roles & user_info[1]) == 0: + self.terminate(403) + return + + self.state = self._route_request_st + + +def apply_patches(): + """ + Apply patches to class attributes for HTTP basic authentication. + """ + global _ATTR_TREE # pylint: disable=W0603 + try: + add_method(http.HttpEngine, _handle_auth_st) + add_method(http.HttpEngine, _handle_auth_header_st) + + _ATTR_TREE = AttributeNode("") + if _USERS: + _USERS.clear() + + _load_users() + _load_roles() + finally: + # Clean up temporary data structures + _ROLE_INDEX.clear() + gc.collect() diff --git a/src/pyrobusta/utils/config.py b/src/pyrobusta/utils/config.py index 1acb070..ecea291 100644 --- a/src/pyrobusta/utils/config.py +++ b/src/pyrobusta/utils/config.py @@ -29,9 +29,10 @@ def const(n): # pylint: disable=C0116 CONF_HTTP_MEM_CAP = const(5) CONF_HTTP_SERVED_PATHS = const(6) CONF_HTTP_FILES_API = const(7) -CONF_SOCKET_MAX_CON = const(8) -CONF_TLS = const(9) -CONF_LOG_LEVEL = const(10) +CONF_HTTP_AUTH = const(8) +CONF_SOCKET_MAX_CON = const(9) +CONF_TLS = const(10) +CONF_LOG_LEVEL = const(11) # ------------------- # Configuration state @@ -54,6 +55,8 @@ def const(n): # pylint: disable=C0116 ["/www", "/lib/pyrobusta"], CONF_HTTP_FILES_API, False, + CONF_HTTP_AUTH, + None, CONF_SOCKET_MAX_CON, 2, CONF_TLS, @@ -78,7 +81,7 @@ def parse_config(key, value): return float(value) if key == CONF_HTTP_SERVED_PATHS: return [normalize_path(p) for p in value.split()] - if key not in (CONF_WIFI_SSID, CONF_WIFI_PASSWORD): + if key not in (CONF_WIFI_SSID, CONF_WIFI_PASSWORD, CONF_HTTP_AUTH): return value.lower() return value diff --git a/src/pyrobusta/utils/helpers.py b/src/pyrobusta/utils/helpers.py index 11fcf90..7c6432d 100644 --- a/src/pyrobusta/utils/helpers.py +++ b/src/pyrobusta/utils/helpers.py @@ -77,3 +77,22 @@ def is_path_segment_valid(filename: str): ): return False return True + + +def iterate_segments(data: str, delimiter: str): + """ + Generator for iterating over each segment of a string, + separated by an arbitrary delimiter. The generator + does not ignore empty segments (including leading and + trailing segments). + """ + start = 0 + end = 0 + while start < len(data): + end = data.find(delimiter, start) + if end == -1: + end = len(data) + yield data[start:end].strip() + start = end + 1 + if data.endswith(delimiter): + yield "" From 0003afd8e47509e8fda84bd6708133bf06d42b7a Mon Sep 17 00:00:00 2001 From: szeka9 Date: Thu, 23 Jul 2026 01:13:24 +0200 Subject: [PATCH 2/5] Add unit test cases for HTTP basic auth & authorization Test resource policy selection, protocol correctness and prefix-tree abstraction for authorization. --- src/pyrobusta/protocol/http_basic_auth.py | 4 +- tests/.pylintrc | 9 +- tests/unit/http_base.py | 1 + tests/unit/test_http.py | 2 +- tests/unit/test_http_basic_auth.py | 611 ++++++++++++++++++++++ 5 files changed, 622 insertions(+), 5 deletions(-) create mode 100644 tests/unit/test_http_basic_auth.py diff --git a/src/pyrobusta/protocol/http_basic_auth.py b/src/pyrobusta/protocol/http_basic_auth.py index 327dc8d..f2a6d34 100644 --- a/src/pyrobusta/protocol/http_basic_auth.py +++ b/src/pyrobusta/protocol/http_basic_auth.py @@ -193,7 +193,7 @@ def _load_roles(config=_ROLES_LOCATION): attributes = {} for line in roles: comment_idx = line.find("#") - line = line[:comment_idx] if comment_idx != -1 else line + line = line[:comment_idx].strip() if comment_idx != -1 else line.strip() # Parse URL path if line.startswith("/"): @@ -206,7 +206,7 @@ def _load_roles(config=_ROLES_LOCATION): paths.append(line) # Parse attributes - elif line.strip(): + elif line: if not paths: raise ValueError() sep = line.find(":") diff --git a/tests/.pylintrc b/tests/.pylintrc index 533feb6..c8dbc83 100644 --- a/tests/.pylintrc +++ b/tests/.pylintrc @@ -1,5 +1,6 @@ [MAIN] -disable=W0212, +disable=W0201, + W0212, C0114, C0115, C0116, @@ -7,4 +8,8 @@ disable=W0212, R0902, R0801 -init-hook='import os, sys; tests = os.path.join(os.getcwd(), "tests"); [sys.path.append(d) for d, _, _ in os.walk(tests)];' \ No newline at end of file +init-hook='import os, sys; tests = os.path.join(os.getcwd(), "tests"); [sys.path.append(d) for d, _, _ in os.walk(tests)];' + +[DESIGN] +max-args=7 +max-positional-arguments=7 diff --git a/tests/unit/http_base.py b/tests/unit/http_base.py index 3f20c37..35cb683 100644 --- a/tests/unit/http_base.py +++ b/tests/unit/http_base.py @@ -57,6 +57,7 @@ def open_side_effect(*args, **kwargs): self.http_module = load_module("pyrobusta/protocol/http.py") self.fs_module = load_module("pyrobusta/protocol/http_file_server.py") self.multipart_module = load_module("pyrobusta/protocol/http_multipart.py") + self.basic_auth_module = load_module("pyrobusta/protocol/http_basic_auth.py") self.fs_patcher = patch.object(self.fs_module, "setup_directories") self.fs_patcher.start() diff --git a/tests/unit/test_http.py b/tests/unit/test_http.py index 34b49db..8ab2193 100644 --- a/tests/unit/test_http.py +++ b/tests/unit/test_http.py @@ -172,7 +172,7 @@ def test_header_parsing_valid(self): self.engine.headers, ) self.assertEqual(self.rx.peek(), b"") - self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertNotEqual(self.engine.state, self.engine._parse_headers_st) def test_header_parsing_incomplete_header(self): request = b"GET /index.html HTTP/1.1\r\nContent-Type\r\n\r\n" diff --git a/tests/unit/test_http_basic_auth.py b/tests/unit/test_http_basic_auth.py new file mode 100644 index 0000000..6ca9976 --- /dev/null +++ b/tests/unit/test_http_basic_auth.py @@ -0,0 +1,611 @@ +import os +import unittest +import hashlib +import base64 +import tempfile + +from http_base import TestHttpBase + + +class TestBasicAuthPolicyStateMachine(TestHttpBase): + """ + Tests for HTTP basic auth policy. + """ + + @classmethod + def setUpClass(cls): + cls.base_config = {"http_auth": "basic"} + cls.cwd = os.getcwd() + + def test_basic_auth_public_resource(self): + self.basic_auth_module._ATTR_TREE.insert_path( + "/app/public", {"GET": self.basic_auth_module._NO_POLICY} + ) + self.engine.state = self.engine._handle_auth_st + self.engine.url = b"/app/public" + self.engine.method = b"GET" + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + + def test_basic_auth_private_resource_exact_rule(self): + self.basic_auth_module._ATTR_TREE.insert_path("/app/private", {"GET": 0b001}) + self.engine.state = self.engine._handle_auth_st + self.engine.url = b"/app/private" + self.engine.method = b"GET" + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._handle_auth_header_st) + + def test_basic_auth_private_resource_no_rule(self): + self.engine.state = self.engine._handle_auth_st + self.engine.url = b"/app/public" + self.engine.method = b"GET" + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._handle_auth_header_st) + + +class TestBasicAuthStateMachine(TestHttpBase): + """ + Tests for HTTP authentication & authorization. + """ + + @classmethod + def setUpClass(cls): + cls.base_config = {"http_auth": "basic"} + cls.cwd = os.getcwd() + + def prepare_auth(self, user, password, auth_header, user_roles, route_roles): + if user is not None and password is not None: + password_hash = hashlib.sha256(password.encode()).digest() + self.basic_auth_module._USERS[user] = [password_hash, user_roles] + + self.basic_auth_module._ATTR_TREE.insert_path( + "/app/private", {"GET": route_roles} + ) + self.engine.state = self.engine._handle_auth_header_st + self.engine.url = b"/app/private" + self.engine.method = b"GET" + if auth_header: + self.engine.headers["authorization"] = auth_header + + def test_basic_auth_successful(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertNotEqual(self.engine.status_code, 401) + + def test_basic_auth_unknown_user(self): + self.prepare_auth( + user=None, + password=None, + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_user_nok(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "invalid:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_user_empty(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode(":".encode("ascii")).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_password_nok(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode("dummy-user:invalid".encode("ascii")).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_password_empty(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode("dummy-user:".encode("ascii")).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_role_nok(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b010, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 403) + + def test_basic_auth_header_missing(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="", + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_header_incomplete(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic ", + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_invalid_encoding(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic invalid-base64", + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_missing_colon(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Basic " + + base64.b64encode( + "dummy-user".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + def test_basic_auth_multiple_colon(self): + self.prepare_auth( + user="dummy-user", + password="dummy:password", + auth_header="Basic " + + base64.b64encode("dummy-user:dummy:password".encode("ascii")).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertNotEqual(self.engine.status_code, 401) + + def test_basic_auth_scheme_case_sensitivity(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="bAsIc " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._route_request_st) + self.assertNotEqual(self.engine.status_code, 401) + + def test_basic_auth_invalid_scheme(self): + self.prepare_auth( + user="dummy-user", + password="", + auth_header="Bearer " + + base64.b64encode( + "dummy-user:".encode("ascii") + ).decode(), + user_roles=0b001, + route_roles=0b001, + ) + + self.engine.state(self.rx) + + self.assertEqual(self.engine.state, self.engine._terminal_st) + self.assertEqual(self.engine.status_code, 401) + + +class TestBasicAuthPrefixTree(TestHttpBase): + """ + Tests for authorization based on prefix tree. + """ + + @classmethod + def setUpClass(cls): + cls.base_config = {"http_auth": "basic"} + cls.cwd = os.getcwd() + + def setup_roles(self, roles: dict): + self.attr_tree = self.basic_auth_module.AttributeNode("") + for path, attributes in roles.items(): + self.attr_tree.insert_path(path, attributes) + + def test_attribute_retrieval_exact_match(self): + self.setup_roles({"/app/resource": {"GET": 0b001}}) + + attributes = self.attr_tree.get_attributes("/app/resource") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/") + self.assertEqual(attributes, {"GET": 0b001}) + + def test_attribute_retrieval_trailing_glob(self): + self.setup_roles({"/app/resource/*": {"GET": 0b001}}) + + attributes = self.attr_tree.get_attributes("/app/resource") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details/") + self.assertEqual(attributes, None) + + def test_attribute_retrieval_intermediate_glob(self): + self.setup_roles({"/app/resource/*/details": {"GET": 0b001}}) + + attributes = self.attr_tree.get_attributes("/app/resource") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name/") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details/") + self.assertEqual(attributes, {"GET": 0b001}) + + def test_attribute_retrieval_trailing_recursive_glob(self): + self.setup_roles( + { + "/app/resource/**": {"GET": 0b001}, + } + ) + + attributes = self.attr_tree.get_attributes("/app/resource") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/") + self.assertEqual(attributes, None) + + attributes = self.attr_tree.get_attributes("/app/resource/name") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/") + self.assertEqual(attributes, {"GET": 0b001}) + + attributes = self.attr_tree.get_attributes("/app/resource/name/details") + self.assertEqual(attributes, {"GET": 0b001}) + + def test_attribute_retrieval_glob_precedence(self): + self.setup_roles( + { + "/**": {"GET": 0}, + "/app": {"GET": 1}, + "/app/*": {"GET": 2}, + "/app/**": {"GET": 3}, + "/app/*/logs": {"GET": 4}, + "/app/faults/logs": {"GET": 5}, + "/app/endpoint": {"GET": 6}, + "/app/endpoint/*": {"GET": 7}, + "/app/endpoint/**": {"GET": 8}, + "/app/endpoint/*/logs": {"GET": 9}, + "/app/endpoint/runtime/logs": {"GET": 10}, + } + ) + attributes = self.attr_tree.get_attributes("/system/runtime/logs") + self.assertEqual(attributes, {"GET": 0}) + + attributes = self.attr_tree.get_attributes("/app") + self.assertEqual(attributes, {"GET": 1}) + + attributes = self.attr_tree.get_attributes("/app/api") + self.assertEqual(attributes, {"GET": 2}) + + attributes = self.attr_tree.get_attributes("/app/admin/status") + self.assertEqual(attributes, {"GET": 3}) + + attributes = self.attr_tree.get_attributes("/app/admin/logs") + self.assertEqual(attributes, {"GET": 4}) + + attributes = self.attr_tree.get_attributes("/app/faults/logs") + self.assertEqual(attributes, {"GET": 5}) + + attributes = self.attr_tree.get_attributes("/app/endpoint") + self.assertEqual(attributes, {"GET": 6}) + + attributes = self.attr_tree.get_attributes("/app/endpoint/status") + self.assertEqual(attributes, {"GET": 7}) + + attributes = self.attr_tree.get_attributes("/app/endpoint/resource/details") + self.assertEqual(attributes, {"GET": 8}) + + attributes = self.attr_tree.get_attributes("/app/endpoint/resource/logs") + self.assertEqual(attributes, {"GET": 9}) + + attributes = self.attr_tree.get_attributes("/app/endpoint/runtime/logs") + self.assertEqual(attributes, {"GET": 10}) + + +class TestBasicAuthUserConfigReader(TestHttpBase): + """ + Tests for user configuration reader. + """ + + @classmethod + def setUpClass(cls): + cls.base_config = {"http_auth": "basic"} + cls.cwd = os.getcwd() + + def test_user_reader_valid_config(self): + file_content = ( + "# User configuration\n" + "user-1:70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145:\n" + "user-2:718a8ca4dd4d30b8dc0756ad9d7de727079869b215b03782279e372ab6911ecf:role1\n" + "user-3:280b73d2ac8375035501e5c06acb4b770e74ec95f479e6e2643227d7f57ce7ad:role1,role2\n" + ) + + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.cwd + ) as file: + self.basic_auth_module._USERS.clear() + file.write(file_content) + file.flush() + self.basic_auth_module._load_users(file.name) + + self.assertDictEqual( + self.basic_auth_module._USERS, + { + "user-1": [ + bytes.fromhex( + "70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145" + ), + 0b00, + ], + "user-2": [ + bytes.fromhex( + "718a8ca4dd4d30b8dc0756ad9d7de727079869b215b03782279e372ab6911ecf" + ), + 0b01, + ], + "user-3": [ + bytes.fromhex( + "280b73d2ac8375035501e5c06acb4b770e74ec95f479e6e2643227d7f57ce7ad" + ), + 0b11, + ], + }, + ) + + def test_user_reader_empty_user(self): + file_content = ( + ":70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145:\n" + ) + + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.cwd + ) as file: + file.write(file_content) + file.flush() + with self.assertRaises(ValueError): + self.basic_auth_module._load_users(file.name) + + def test_user_reader_duplicate_user(self): + file_content = ( + "user-1:70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145:\n" + "user-1:718a8ca4dd4d30b8dc0756ad9d7de727079869b215b03782279e372ab6911ecf:\n" + ) + + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.cwd + ) as file: + file.write(file_content) + file.flush() + with self.assertRaises(ValueError): + self.basic_auth_module._load_users(file.name) + + def test_user_reader_empty_password(self): + file_content = "user-1::\n" + + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.cwd + ) as file: + file.write(file_content) + file.flush() + with self.assertRaises(ValueError): + self.basic_auth_module._load_users(file.name) + + +class TestBasicAuthRoleConfigReader(TestHttpBase): + """ + Tests for role configuration reader. + """ + + @classmethod + def setUpClass(cls): + cls.base_config = {"http_auth": "basic"} + cls.cwd = os.getcwd() + + def test_role_reader_valid_config(self): + file_content = ( + "/*\n" + " *:*\n" + "# Comment\n" + "/app/resource\n" + "/app/resource/* # inline comment\n" + " GET: role_1\n" + " POST,PUT:role_2,role_3\n" + " OPTIONS: *# inline comment\n" + " DELETE:\n" + " \n" + " /app/api\n\n" + "GET: role_4 \n\n" + "POST , PUT : role_1 , role_2 " + ) + + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.cwd + ) as file: + self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") + file.write(file_content) + file.flush() + self.basic_auth_module._load_roles(file.name) + + attributes = self.basic_auth_module._ATTR_TREE.get_attributes("/index.html") + self.assertEqual(attributes, {"*": self.basic_auth_module._NO_POLICY}) + + attributes = self.basic_auth_module._ATTR_TREE.get_attributes("/app/resource") + self.assertEqual( + attributes, + { + "GET": 0b001, + "POST": 0b110, + "PUT": 0b110, + "DELETE": 0b000, + "OPTIONS": self.basic_auth_module._NO_POLICY, + }, + ) + + attributes = self.basic_auth_module._ATTR_TREE.get_attributes("/app/api") + self.assertEqual( + attributes, + { + "GET": 0b1000, + "POST": 0b0011, + "PUT": 0b0011, + }, + ) + + def test_role_reader_missing_path(self): + file_content = " *:*\n" + + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.cwd + ) as file: + self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") + file.write(file_content) + file.flush() + with self.assertRaises(ValueError): + self.basic_auth_module._load_roles(file.name) + + def test_role_reader_duplicate_attribute(self): + file_content = "/app/resource\n" + " GET: role_1\n" + " GET: role_2\n" + + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=self.cwd + ) as file: + self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") + file.write(file_content) + file.flush() + with self.assertRaises(ValueError): + self.basic_auth_module._load_roles(file.name) + + +if __name__ == "__main__": + unittest.main(verbosity=2) From 2eeecd6737649cd11481b0ea79ac697149ab7a33 Mon Sep 17 00:00:00 2001 From: szeka9 Date: Thu, 23 Jul 2026 01:20:52 +0200 Subject: [PATCH 3/5] Add documentation about HTTP basic auth and RBAC --- README.md | 1 + docs/application_development/configuration.md | 1 + docs/application_development/index.md | 1 + docs/application_development/request.md | 3 +- docs/application_development/security.md | 176 +++++++++++++++++- 5 files changed, 177 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index eb36c1a..112a146 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ PyRobusta is a memory-conscious HTTP/1.1 server library built for embedded devic - Built-in API for uploading, downloading, and deleting files stored on the server - Persistent connection handling via the `Connection: keep-alive` header - HTTP/1.0 and HTTP/1.1 support +- Basic authentication and RBAC authorization - TLS support ## Design Principles diff --git a/docs/application_development/configuration.md b/docs/application_development/configuration.md index 6eb8582..4967c71 100644 --- a/docs/application_development/configuration.md +++ b/docs/application_development/configuration.md @@ -51,6 +51,7 @@ $ mpremote a0 cp pyrobusta.env :/pyrobusta.env | `http_mem_cap` | Fraction of available heap memory reserved for stream buffers. Valid range: (0, 1]. | 0.1 | | `http_served_paths` | Space-separated list of filesystem paths that may be served over HTTP. | `/www /lib/pyrobusta` | | `http_files_api` | Enables or disables the file management API endpoint (`/files`), allowing upload, download, and listing of files. | False | +| `http_auth` | Selects the type of authentication method enforced by the server. Currently, basic authentication (`basic`) is supported. | None | | `socket_max_con` | Maximum number of simultaneous socket connections. | 2 | | `tls` | Enables or disables TLS. When enabled, `cert.der` and `key.der` must be installed at the server root. | False | | `log_level` | Logging level. Can be one of: `warning`, `info`, `debug`. | info | diff --git a/docs/application_development/index.md b/docs/application_development/index.md index f97dcd5..2784d8e 100644 --- a/docs/application_development/index.md +++ b/docs/application_development/index.md @@ -17,6 +17,7 @@ Then continue with the remaining guides in the order listed below. * [Response Processing](response.md) * [Static Content](static_content.md) * [File Server API](file_server.md) +* [Authentication & Security](security.md) * [Source Code](https://github.com/szeka9/PyRobusta) --- diff --git a/docs/application_development/request.md b/docs/application_development/request.md index 11a986d..9552840 100644 --- a/docs/application_development/request.md +++ b/docs/application_development/request.md @@ -155,7 +155,8 @@ Unlike regular request bodies, multipart parts are parsed by the server before b to the application. Preprocessed parts consist of headers (dictionary) and the raw part body (bytes), passed as a tuple. -**Multipart state tracking** +### Multipart State Tracking + The HTTP context exposes the boolean attributes `mp_is_first` and `mp_is_last` diff --git a/docs/application_development/security.md b/docs/application_development/security.md index 5f7038d..70e7e26 100644 --- a/docs/application_development/security.md +++ b/docs/application_development/security.md @@ -2,26 +2,194 @@ [← Back](index.md) -This page is work in progress. Refer to the [main page](index.md) for available documentation. +This page documents the security features provided by PyRobusta, +including HTTP Basic Authentication, role-based authorization, and TLS. + +Note: Authentication verifies the client's identity. Authorization determines +whether the authenticated user is permitted to access a resource. --- ## Table of Contents * [Authentication & Security](#authentication-security) - + [API Key Authentication](#api-key-authentication) + [Basic Authentication](#basic-authentication) + + [Authorization](#authorization) + [HTTPS / TLS](#https-tls) + [Certificate Installation](#certificate-installation) --- -## API Key Authentication - ## Basic Authentication +PyRobusta supports HTTP Basic Authentication with per-user credentials +stored on the device. Passwords are stored as SHA-256 hashes. +Basic authentication is disabled by default. It can be enabled by setting +`http_auth=basic` in [pyrobusta.env](configuration.md). + +During initialization, PyRobusta reads `pyrobusta.passwd` +from the server's working directory. Each entry specifies a username, +a password hash, and one or more role names. + +PyRobusta defines a single realm (`realm="Device"`) to authenticate against. +The server response includes the `WWW-Authenticate: Basic realm="Device"` header +when a request does not contain valid credentials. + +### pyrobusta.passwd + +`pyrobusta.passwd` utilizes a passwd-like format, specifying one user per line, +with each element of the user data separated by `:`. The expected format +of a user entry is `::`. + +``` +# Example +# File: /pyrobusta.passwd +szeka9:9b085649ac73a164314f00ec70010ec37d6c663cf71e9672a0600d6609b3ae12:api_admin +user-1:c0418ce1f43b9013c6871d4ac4941de6cd83a3fac29677785b250b86c88320a0:api_user,app_viewer +[...] +user-n:a8105c3c8b75556a9099b8dcab9cc13362d5a6f9fa5888ce39efc57961deb519:api_user,app_maintainer +``` + +When adding new users, follow the below steps to calculate the password hash: + +```python3 +import hashlib + +password_raw = "" +password_hash = hashlib.sha256(password_raw.encode()).digest().hex() +print(password_hash) # paste the output into pyrobusta.passwd + +b9fa35baa8069f3dabe214aed3525da0824efefdbce5ad83dfe4e7f48f8ce15f +``` + +## Authorization + +When HTTP Basic Authentication is enabled, PyRobusta provides role-based access control (RBAC). +Resource permissions are determined by the authenticated user's assigned roles. +Each user can be assigned one or more roles, configured in `pyrobusta.passwd`. + +While `pyrobusta.passwd` assigns roles to users, `pyrobusta.roles` maps those roles to +HTTP methods and server resources. Authorization follows a least-privilege model. +Users have no permissions unless explicitly granted by one or more authorization rules. +Each matching rule grants additional permissions. Requests that do not match an authorization +rule receive HTTP 403 Forbidden. + +### pyrobusta.roles + +`pyrobusta.roles` consists of one or more authorization blocks. Each block defines one or +more path patterns, followed by one or more authorization rules. Every authorization rule +applies to every path pattern declared within the same block. + +``` + + +... + : ,,... + : ,,... + ... + + + ... +``` + +Authorization rules are selected based on pattern specificity rather than their +order in `pyrobusta.roles`. Path patterns support the following forms, ordered +from least to most specific. Each request is matched against the most specific +pattern, and only the roles associated with the selected pattern are considered +during authorization. + +``` +/** # Match any path +/* # Match direct children of / +/app # Exact match +/app/** # Match any descendant of /app +/app/* # Match direct children of /app +/app/*/resources # Match a single wildcard segment +/app/endpoint/resources # Exact match +``` + +**Note**: recursive globs (`**`) are only supported as the final path segment. +Allowing `**` in intermediate segments would introduce ambiguous matches +and is therefore not supported. + +``` +# File: /pyrobusta.roles + +# --------------------------------------------------------- +# Example: public resources (/, /index.html, /styles.css, ...) +/ +/* + GET,HEAD: * + +# --------------------------------------------------------- +# Example: apply rule to parent and child resources +/app/resource +/app/resource/* + GET: resource_viewer + POST,PUT,DELETE: resource_maintainer + +# --------------------------------------------------------- +# Example: apply rule to paths of any length with ** +/app/api/** + GET: api_viewer + POST,PUT,DELETE: api_maintainer + +# --------------------------------------------------------- +# Example: require api_admin access for all HTTP methods +/app/api/management + *: api_admin + +# --------------------------------------------------------- +# Example: restrict all access to a resource +/app/secret + *: + +# --------------------------------------------------------- +# Example: assign multiple roles +/app/logs + GET: api_admin,api_maintainer +``` + +**Note**: The special role `*` grants public access regardless of the authenticated user's roles. +Rules that specify `*` do not require authentication. + +## Authentication & Authorization Flow + +```mermaid +flowchart TD + A[Request] --> B[Resolve requested resource] + B --> C[Determine security policy for resource] + + C --> D[Public] + C --> E[Protected] + + D --> F[Ignore auth requirement] + F --> G[Process request] + + E --> H{Is Authorization header present?} + + H -->|No| I[401 +
WWW-Authenticate] + H -->|Yes| J[Authenticate] + + J --> K{Valid credentials?} + + K -->|No| L[401 +
WWW-Authenticate] + K -->|Yes| M{Authorized?} + + M -->|No| N[403] + M -->|Yes| G +``` + +| Condition | Response | +| --- | --- | +| Missing credentials | 401 Unauthorized | +| Invalid credentials | 401 Unauthorized | +| Authenticated but insufficient permissions | 403 Forbidden | + ## HTTPS / TLS + + ## Certificate Installation --- From 9263472213edb2a777c3c35c201ff44b74d5d8eb Mon Sep 17 00:00:00 2001 From: szeka9 Date: Fri, 24 Jul 2026 01:11:41 +0200 Subject: [PATCH 4/5] Add functional tests for HTTP basic auth; refactor common helpers Add simple test cases for HTTP basic auth for positive cases. Refactor duplicate helper functions and move them to a common module, imported by test modules. Add missing content-length value to HTTP requests sent during the tests. --- Makefile | 1 + tests/functional/env_utils.py | 119 +++++++++++++++ tests/functional/test_http.py | 143 +++--------------- tests/functional/test_http_basic_auth.py | 127 ++++++++++++++++ tests/functional/test_http_file_server.py | 170 +++------------------- tests/functional/test_http_multipart.py | 153 +++---------------- 6 files changed, 309 insertions(+), 404 deletions(-) create mode 100644 tests/functional/env_utils.py create mode 100644 tests/functional/test_http_basic_auth.py diff --git a/Makefile b/Makefile index 8203ae7..1981023 100644 --- a/Makefile +++ b/Makefile @@ -302,6 +302,7 @@ test-unix: stage-test tls-cert .PHONY: test-device test-device: stage-test #clean-device upload @mpremote $(DEVICE) soft-reset + @mpremote $(DEVICE) cp $(TEST_RUNTIME)/env_utils.py :/env_utils.py @cd $(TEST_RUNTIME); \ for test in test_*.py; do \ echo "\n==================================="; \ diff --git a/tests/functional/env_utils.py b/tests/functional/env_utils.py new file mode 100644 index 0000000..b7aebdb --- /dev/null +++ b/tests/functional/env_utils.py @@ -0,0 +1,119 @@ +import asyncio +import ssl +import gc + +from os import mkdir, listdir, remove, rmdir + +from pyrobusta.server import http_server +from pyrobusta.protocol.http import enable_optional_features +from pyrobusta.utils.config import ( + CONF_TLS, + CONF_LOG_LEVEL, + CONF_HTTP_MULTIPART, + CONF_HTTP_FILES_API, + CONF_HTTP_SERVED_PATHS, + CONF_HTTP_AUTH, + _CONFIG_CACHE, + parse_config, +) + + +def garbage_collect(coroutine): + async def decorated(*args, **kwargs): + gc.collect() + await coroutine(*args, **kwargs) + gc.collect() + + return decorated + + +def test_assert(name, actual, expected): + print(f"Test {name}: ", end="") + if actual == expected: + print("OK") + else: + print("Fail") + raise AssertionError(f"{actual} != {expected}") + + +async def start_server(): + """ + Start an HTTP server as a background task. + """ + server = http_server.HttpServer() + await server.start_socket_server() + await asyncio.sleep_ms(100) + return server + + +async def send_request(request, tls=False): + port = ( + http_server.HttpServer.LISTEN_PORT_HTTPS + if tls + else http_server.HttpServer.LISTEN_PORT_HTTP + ) + + ctx = None + if tls: + # Disable certificate verification due to self-signed cert + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.verify_mode = ssl.CERT_NONE + + reader, writer = await asyncio.open_connection("127.0.0.1", port, ssl=ctx) + writer.write(request) + await writer.drain() + + to_read = True + response = b"" + while to_read: + response_part = await reader.read(1024) + response += response_part + to_read = len(response_part) + writer.close() + return response + + +def fmkdir(path: str): + try: + mkdir(path) + except OSError: + pass + + +def delete_path(path): + for name in listdir(path): + if path == "/": + full = "/" + name + else: + full = path + "/" + name + + try: + remove(full) + except OSError: + delete_path(full) + try: + rmdir(full) + except OSError: + pass + + +def setup_config( + tls_enabled=False, + files_api_enabled=False, + http_multipart_enabled=False, + served_paths="", + http_auth="", +): + http_server.HttpServer.LISTEN_PORT_HTTP = 8080 + http_server.HttpServer.LISTEN_PORT_HTTPS = 4443 + + _CONFIG_CACHE[2 * CONF_LOG_LEVEL + 1] = "warning" + _CONFIG_CACHE[2 * CONF_TLS + 1] = tls_enabled + _CONFIG_CACHE[2 * CONF_HTTP_SERVED_PATHS + 1] = parse_config( + CONF_HTTP_SERVED_PATHS, served_paths + ) + _CONFIG_CACHE[2 * CONF_HTTP_MULTIPART + 1] = http_multipart_enabled + _CONFIG_CACHE[2 * CONF_HTTP_FILES_API + 1] = files_api_enabled + _CONFIG_CACHE[2 * CONF_HTTP_AUTH + 1] = http_auth + + enable_optional_features() diff --git a/tests/functional/test_http.py b/tests/functional/test_http.py index f459dad..92905b8 100644 --- a/tests/functional/test_http.py +++ b/tests/functional/test_http.py @@ -1,99 +1,20 @@ import asyncio -import ssl import json -import gc -from os import mkdir, remove, rmdir, listdir +from env_utils import ( + garbage_collect, + test_assert, + send_request, + setup_config, + start_server, + fmkdir, + delete_path, +) from pyrobusta.server import http_server from pyrobusta.protocol.http import HttpEngine -from pyrobusta.utils.config import ( - CONF_HTTP_SERVED_PATHS, - CONF_HTTP_FILES_API, - CONF_TLS, - CONF_LOG_LEVEL, - _CONFIG_CACHE, - parse_config, -) -from pyrobusta.utils.helpers import normalize_path - -################################################# -# Test helpers -################################################# - - -def garbage_collect(coroutine): - async def decorated(*args, **kwargs): - gc.collect() - await coroutine(*args, **kwargs) - gc.collect() - - return decorated - - -def test_assert(name, actual, expected): - print(f"Test {name}: ", end="") - if actual == expected: - print("OK") - else: - print("Fail") - raise AssertionError(f"{actual} != {expected}") - - -async def send_request(request, tls=False): - port = ( - http_server.HttpServer.LISTEN_PORT_HTTPS - if tls - else http_server.HttpServer.LISTEN_PORT_HTTP - ) - - ctx = None - if tls: - # Disable certificate verification due to self-signed cert - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.verify_mode = ssl.CERT_NONE - - reader, writer = await asyncio.open_connection("127.0.0.1", port, ssl=ctx) - writer.write(request) - await writer.drain() - - to_read = True - response = b"" - while to_read: - response_part = await reader.read(1024) - response += response_part - to_read = len(response_part) - writer.close() - return response - -def fmkdir(path: str): - try: - mkdir(path) - except OSError: - pass - - -def delete_path(path): - for name in listdir(path): - if path == "/": - full = "/" + name - else: - full = path + "/" + name - - try: - remove(full) - except OSError: - delete_path(full) - try: - rmdir(full) - except OSError: - pass - - -################################################# -# Test driver -################################################# +from pyrobusta.utils.helpers import normalize_path @HttpEngine.route("/test/simple", "GET") @@ -121,16 +42,6 @@ def chunked_handler(http_ctx, chunk): recv_chunks.append(chunk.decode("utf8")) -async def start_server(): - """ - Start an HTTP server as a background task - """ - server = http_server.HttpServer() - await server.start_socket_server() - await asyncio.sleep_ms(100) - return server - - @garbage_collect async def test_simple_response(tls_enabled): setup_config(tls_enabled=tls_enabled) @@ -366,23 +277,6 @@ async def test_keepalive(): await server.terminate() -################################################# -# Test methods -################################################# - - -def setup_config(tls_enabled=False, served_paths="", files_api_enabled=False): - http_server.HttpServer.LISTEN_PORT_HTTP = 8080 - http_server.HttpServer.LISTEN_PORT_HTTPS = 4443 - - _CONFIG_CACHE[2 * CONF_LOG_LEVEL + 1] = "warning" - _CONFIG_CACHE[2 * CONF_TLS + 1] = tls_enabled - _CONFIG_CACHE[2 * CONF_HTTP_SERVED_PATHS + 1] = parse_config( - CONF_HTTP_SERVED_PATHS, served_paths - ) - _CONFIG_CACHE[2 * CONF_HTTP_FILES_API + 1] = files_api_enabled - - def test_registration(): test_assert( "simple route registration", @@ -397,15 +291,14 @@ def test_registration(): ) -def test_main(): +async def test_main(): test_registration() - asyncio.run(test_simple_response(tls_enabled=False)) - asyncio.run(test_simple_response(tls_enabled=True)) - - asyncio.run(test_server_busy()) - asyncio.run(test_chunked_transfer_encoding()) - asyncio.run(test_fs_access_control()) - asyncio.run(test_keepalive()) + await test_simple_response(tls_enabled=False) + await test_simple_response(tls_enabled=True) + await test_server_busy() + await test_chunked_transfer_encoding() + await test_fs_access_control() + await test_keepalive() -test_main() +asyncio.run(test_main()) diff --git a/tests/functional/test_http_basic_auth.py b/tests/functional/test_http_basic_auth.py new file mode 100644 index 0000000..7cba210 --- /dev/null +++ b/tests/functional/test_http_basic_auth.py @@ -0,0 +1,127 @@ +import asyncio + +from pyrobusta.protocol import http_basic_auth +from pyrobusta.protocol.http import HttpEngine +from env_utils import ( + garbage_collect, + test_assert, + send_request, + setup_config, + start_server, +) + + +@HttpEngine.route("/test/auth", "GET") +def auth_handler(http_ctx, _): + return "text/plain", "OK" + + +@garbage_collect +async def test_missing_auth_header(): + setup_config(http_auth="basic") + server = await start_server() + + # Test: unauthenticated & unauthorized + plain_response = await send_request( + b"GET /test/auth HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Length: 0\r\n" + b"Connection: close\r\n\r\n" + ) + + test_assert( + f"request rejected with 401 Unauthorized", + b"401 Unauthorized" in plain_response, + True, + ) + + await server.terminate() + + +@garbage_collect +async def test_missing_role(): + setup_config(http_auth="basic") + server = await start_server() + + # Test: authenticated & unauthorized + plain_response = await send_request( + b"GET /test/auth HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Length: 0\r\n" + b"Authorization: Basic dXNlcjE6cGFzc3dvcmQx\r\n" + b"Connection: close\r\n\r\n" + ) + + test_assert( + f"request rejected with 403 Forbidden", + b"403 Forbidden" in plain_response, + True, + ) + + await server.terminate() + + +@garbage_collect +async def test_user_authorized(): + setup_config(http_auth="basic") + server = await start_server() + + # Test: authenticated & authorized + plain_response = await send_request( + b"GET /test/auth HTTP/1.1\r\n" + b"Host: localhost\r\n" + b"Content-Length: 0\r\n" + b"Authorization: Basic dXNlcjI6cGFzc3dvcmQy\r\n" + b"Connection: close\r\n\r\n" + ) + + test_assert( + f"request accepted with 200 OK", + b"200 OK" in plain_response, + True, + ) + + await server.terminate() + + +def setup_auth(): + with open("pyrobusta.passwd", "w") as users: + users.write( + "user1:0b14d501a594442a01c6859541bcb3e8164d183d32937b851835442f69d5c94e:role1\n" + ) + users.write( + "user2:6cf615d5bcaac778352a8f1f3360d23f02f34ec182e259897fd6ce485d7870d4:role2\n" + ) + + with open("pyrobusta.roles", "w") as users: + users.write("/test/auth\n*:role2\n") + + +def test_registration(): + test_assert( + "auth route registration", + auth_handler, + HttpEngine._get_handler(b"/test/auth", b"GET"), + ) + + +def test_auth_patches(): + setup_auth() + setup_config(http_auth="basic") + test_assert( + "auth state machine patches", + http_basic_auth._handle_auth_st, + HttpEngine._handle_auth_st, + ) + + +async def test_main(): + test_registration() + test_auth_patches() + + await test_missing_auth_header() + await test_missing_role() + await test_user_authorized() + + +asyncio.run(test_main()) diff --git a/tests/functional/test_http_file_server.py b/tests/functional/test_http_file_server.py index 618d860..3439c67 100644 --- a/tests/functional/test_http_file_server.py +++ b/tests/functional/test_http_file_server.py @@ -1,132 +1,24 @@ import asyncio import json -import json -import ssl -import gc -from os import mkdir, listdir, remove, rmdir, stat +from os import stat -from pyrobusta.server import http_server -from pyrobusta.protocol.http import ( - CONF_HTTP_SERVED_PATHS, - enable_optional_features, - stat, -) -from pyrobusta.utils.config import ( - CONF_TLS, - CONF_LOG_LEVEL, - CONF_HTTP_MULTIPART, - CONF_HTTP_FILES_API, - _CONFIG_CACHE, - normalize_path, - parse_config, +from env_utils import ( + garbage_collect, + test_assert, + send_request, + setup_config, + start_server, + fmkdir, + delete_path, ) -################################################# -# Test helpers -################################################# - - -def garbage_collect(coroutine): - async def decorated(*args, **kwargs): - gc.collect() - await coroutine(*args, **kwargs) - gc.collect() - - return decorated - - -def test_assert(name, actual, expected): - print(f"Test {name}: ", end="") - if actual == expected: - print("OK") - else: - print("Fail") - raise AssertionError(f"{actual} != {expected}") - - -async def send_request(request, tls=False): - port = ( - http_server.HttpServer.LISTEN_PORT_HTTPS - if tls - else http_server.HttpServer.LISTEN_PORT_HTTP - ) - - ctx = None - if tls: - # Disable certificate verification due to self-signed cert - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.verify_mode = ssl.CERT_NONE - - reader, writer = await asyncio.open_connection("127.0.0.1", port, ssl=ctx) - writer.write(request) - await writer.drain() - - to_read = True - response = b"" - while to_read: - response_part = await reader.read(1024) - response += response_part - to_read = len(response_part) - writer.close() - return response - - -def multipart_response(num_responses): - i = 0 - - def response_generator(): - nonlocal i - i += 1 - if i > num_responses: - return None - return "text/plain", b"Response %s" % i - - return response_generator - - -def fmkdir(path: str): - try: - mkdir(path) - except OSError: - pass - - -def delete_path(path): - for name in listdir(path): - if path == "/": - full = "/" + name - else: - full = path + "/" + name - - try: - remove(full) - except OSError: - delete_path(full) - try: - rmdir(full) - except OSError: - pass - - -################################################# -# Test driver -################################################# - - -async def start_server(): - """ - Start an HTTP server as a background task. - """ - server = http_server.HttpServer() - await server.start_socket_server() - await asyncio.sleep_ms(100) - return server +from pyrobusta.utils.config import normalize_path @garbage_collect async def test_fs_path_traversal(): - setup_config(served_paths="/test") + setup_config(files_api_enabled=True, served_paths="/test") server = await start_server() test_root = normalize_path("/test") styles_dir = normalize_path("/test/style") @@ -145,6 +37,7 @@ async def test_fs_path_traversal(): # Test case response = await send_request( b"GET /files/test HTTP/1.1\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" b"Host: localhost\r\n\r\n" ) @@ -187,7 +80,7 @@ async def test_fs_path_traversal(): @garbage_collect async def test_fs_access_control(): - setup_config(served_paths="/test/allowed") + setup_config(files_api_enabled=True, served_paths="/test/allowed") server = await start_server() test_root = normalize_path("/test") @@ -211,6 +104,7 @@ async def test_fs_access_control(): # Case #1: /test/allowed/index.html response = await send_request( b"GET /files/test/allowed/index.html HTTP/1.1\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" b"Host: localhost\r\n\r\n" ) @@ -225,6 +119,7 @@ async def test_fs_access_control(): # Case #2: /test/rejected/index.html response = await send_request( b"GET /files/test/rejected/index.html HTTP/1.1\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" b"Host: localhost\r\n\r\n" ) @@ -241,7 +136,7 @@ async def test_fs_access_control(): @garbage_collect async def test_bulk_file_upload(): - setup_config(http_multipart_enabled=True) + setup_config(files_api_enabled=True, http_multipart_enabled=True) server = await start_server() user_data = normalize_path("/www/user_data") @@ -298,7 +193,7 @@ async def test_bulk_file_upload(): @garbage_collect async def test_chunked_file_upload(): - setup_config() + setup_config(files_api_enabled=True) server = await start_server() user_data = normalize_path("/www/user_data") @@ -341,30 +236,11 @@ async def test_chunked_file_upload(): await server.terminate() -################################################# -# Test methods -################################################# - - -def setup_config(tls_enabled=False, http_multipart_enabled=False, served_paths=""): - http_server.HttpServer.LISTEN_PORT_HTTP = 8080 - http_server.HttpServer.LISTEN_PORT_HTTPS = 4443 - - _CONFIG_CACHE[2 * CONF_LOG_LEVEL + 1] = "warning" - _CONFIG_CACHE[2 * CONF_TLS + 1] = tls_enabled - _CONFIG_CACHE[2 * CONF_HTTP_SERVED_PATHS + 1] = parse_config( - CONF_HTTP_SERVED_PATHS, served_paths - ) - _CONFIG_CACHE[2 * CONF_HTTP_MULTIPART + 1] = http_multipart_enabled - _CONFIG_CACHE[2 * CONF_HTTP_FILES_API + 1] = True - enable_optional_features() - - -def test_main(): - asyncio.run(test_fs_path_traversal()) - asyncio.run(test_fs_access_control()) - asyncio.run(test_bulk_file_upload()) - asyncio.run(test_chunked_file_upload()) +async def test_main(): + await test_fs_path_traversal() + await test_fs_access_control() + await test_bulk_file_upload() + await test_chunked_file_upload() -test_main() +asyncio.run(test_main()) diff --git a/tests/functional/test_http_multipart.py b/tests/functional/test_http_multipart.py index 6fde540..692a980 100644 --- a/tests/functional/test_http_multipart.py +++ b/tests/functional/test_http_multipart.py @@ -1,71 +1,15 @@ import asyncio -import ssl -import gc -from os import mkdir, listdir, remove, rmdir - -from pyrobusta.server import http_server -from pyrobusta.protocol import http_multipart -from pyrobusta.protocol.http import ( - HttpEngine, - enable_optional_features, +from env_utils import ( + garbage_collect, + test_assert, + send_request, + setup_config, + start_server, ) -from pyrobusta.utils.config import ( - CONF_TLS, - CONF_LOG_LEVEL, - CONF_HTTP_MULTIPART, - CONF_HTTP_FILES_API, - _CONFIG_CACHE, -) - -################################################# -# Test helpers -################################################# - - -def garbage_collect(coroutine): - async def decorated(*args, **kwargs): - gc.collect() - await coroutine(*args, **kwargs) - gc.collect() - - return decorated - -def test_assert(name, actual, expected): - print(f"Test {name}: ", end="") - if actual == expected: - print("OK") - else: - print("Fail") - raise AssertionError(f"{actual} != {expected}") - - -async def send_request(request, tls=False): - port = ( - http_server.HttpServer.LISTEN_PORT_HTTPS - if tls - else http_server.HttpServer.LISTEN_PORT_HTTP - ) - - ctx = None - if tls: - # Disable certificate verification due to self-signed cert - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.verify_mode = ssl.CERT_NONE - - reader, writer = await asyncio.open_connection("127.0.0.1", port, ssl=ctx) - writer.write(request) - await writer.drain() - - to_read = True - response = b"" - while to_read: - response_part = await reader.read(1024) - response += response_part - to_read = len(response_part) - writer.close() - return response +from pyrobusta.protocol import http_multipart +from pyrobusta.protocol.http import HttpEngine def multipart_response(num_responses): @@ -81,67 +25,28 @@ def response_generator(): return response_generator -def fmkdir(path: str): - try: - mkdir(path) - except OSError: - pass - - -def delete_path(path): - for name in listdir(path): - if path == "/": - full = "/" + name - else: - full = path + "/" + name - - try: - remove(full) - except OSError: - delete_path(full) - try: - rmdir(full) - except OSError: - pass - - -################################################# -# Test driver -################################################# - - @HttpEngine.route("/test/multipart", "GET") def multipart_handler(http_ctx, _): part_count = int(http_ctx.headers["x-part-count"]) return "multipart/form-data", multipart_response(part_count) -async def start_server(): - """ - Start an HTTP server as a background task. - """ - server = http_server.HttpServer() - await server.start_socket_server() - await asyncio.sleep_ms(100) - return server - - @garbage_collect -async def test_multipart_response(tls_enabled): - setup_config(tls_enabled=tls_enabled) +async def test_multipart_response(): + setup_config(http_multipart_enabled=True) server = await start_server() # Test: 1 part plain_response = await send_request( b"GET /test/multipart HTTP/1.1\r\n" b"Host: localhost\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" - b"X-Part-Count: 1\r\n\r\n", - tls_enabled, + b"X-Part-Count: 1\r\n\r\n" ) test_assert( - f"http{"s" if tls_enabled else ""} response contains 1 part", + f"multipart response contains 1 part", b"Response 1" in plain_response, True, ) @@ -150,12 +55,13 @@ async def test_multipart_response(tls_enabled): plain_response = await send_request( b"GET /test/multipart HTTP/1.1\r\n" b"Host: localhost\r\n" + b"Content-Length: 0\r\n" b"Connection: close\r\n" - b"X-Part-Count: 10\r\n\r\n", - tls_enabled, + b"X-Part-Count: 10\r\n\r\n" ) + test_assert( - f"http{"s" if tls_enabled else ""} response contains 10 parts", + f"multipart response contains 10 parts", [b"Response %s" % i in plain_response for i in range(1, 11)], [True] * 10, ) @@ -163,22 +69,6 @@ async def test_multipart_response(tls_enabled): await server.terminate() -################################################# -# Test methods -################################################# - - -def setup_config(tls_enabled=False, files_api_enabled=False): - http_server.HttpServer.LISTEN_PORT_HTTP = 8080 - http_server.HttpServer.LISTEN_PORT_HTTPS = 4443 - - _CONFIG_CACHE[2 * CONF_LOG_LEVEL + 1] = "warning" - _CONFIG_CACHE[2 * CONF_TLS + 1] = tls_enabled - _CONFIG_CACHE[2 * CONF_HTTP_MULTIPART + 1] = True - _CONFIG_CACHE[2 * CONF_HTTP_FILES_API + 1] = files_api_enabled - enable_optional_features() - - def test_registration(): test_assert( "multipart route registration", @@ -188,7 +78,7 @@ def test_registration(): def test_multipart_patches(): - setup_config() + setup_config(http_multipart_enabled=True) test_assert( "multipart state machine patches", http_multipart._start_multipart_parser_st, @@ -196,11 +86,10 @@ def test_multipart_patches(): ) -def test_main(): +async def test_main(): test_registration() test_multipart_patches() - asyncio.run(test_multipart_response(tls_enabled=False)) - asyncio.run(test_multipart_response(tls_enabled=True)) + await test_multipart_response() -test_main() +asyncio.run(test_main()) From 53ec4463c283678b770315cbf1a0ff99c389716e Mon Sep 17 00:00:00 2001 From: szeka9 Date: Sat, 25 Jul 2026 14:35:16 +0200 Subject: [PATCH 5/5] Configurable passwd and roles file; fix bugs in config loading and normalization Make passwd and roles files configurable to improve usability and to simply unit testing. Additionally, this change fixes bugs in the configuration module: - normalize default file path values - only read configuration from file if the configuration is not loaded; do not re-read configuration if a value is None --- docs/application_development/configuration.md | 4 +- docs/application_development/security.md | 9 +- src/pyrobusta/protocol/http_basic_auth.py | 117 ++++++++++-------- src/pyrobusta/utils/config.py | 19 ++- tests/unit/http_base.py | 52 ++++++-- tests/unit/test_http_basic_auth.py | 114 +++++++++-------- 6 files changed, 187 insertions(+), 128 deletions(-) diff --git a/docs/application_development/configuration.md b/docs/application_development/configuration.md index 4967c71..86a0e4c 100644 --- a/docs/application_development/configuration.md +++ b/docs/application_development/configuration.md @@ -54,7 +54,9 @@ $ mpremote a0 cp pyrobusta.env :/pyrobusta.env | `http_auth` | Selects the type of authentication method enforced by the server. Currently, basic authentication (`basic`) is supported. | None | | `socket_max_con` | Maximum number of simultaneous socket connections. | 2 | | `tls` | Enables or disables TLS. When enabled, `cert.der` and `key.der` must be installed at the server root. | False | -| `log_level` | Logging level. Can be one of: `warning`, `info`, `debug`. | info | +| `passwd_file` | Path to the file containing user credentials used for authentication. | `/pyrobusta.passwd` | +| `roles_file` | Path to the file containing RBAC role definitions used for authorization. | `/pyrobusta.roles` | +| `log_level` | Logging level. Can be one of: `warning`, `info`, `debug`. | `info` | ## Configuration API diff --git a/docs/application_development/security.md b/docs/application_development/security.md index 70e7e26..be512c9 100644 --- a/docs/application_development/security.md +++ b/docs/application_development/security.md @@ -27,9 +27,11 @@ stored on the device. Passwords are stored as SHA-256 hashes. Basic authentication is disabled by default. It can be enabled by setting `http_auth=basic` in [pyrobusta.env](configuration.md). -During initialization, PyRobusta reads `pyrobusta.passwd` +During initialization, PyRobusta reads `pyrobusta.passwd` by default from the server's working directory. Each entry specifies a username, -a password hash, and one or more role names. +a password hash, and one or more role names. Set `passwd_file` in the configuration +to specify an alternative user credentials file to load. For more information, +see [Server Configuration](./configuration.md). PyRobusta defines a single realm (`realm="Device"`) to authenticate against. The server response includes the `WWW-Authenticate: Basic realm="Device"` header @@ -74,6 +76,9 @@ Users have no permissions unless explicitly granted by one or more authorization Each matching rule grants additional permissions. Requests that do not match an authorization rule receive HTTP 403 Forbidden. +Set `roles_file` in the configuration to specify an alternative role configuration file to load. +For more information, see [Server Configuration](./configuration.md). + ### pyrobusta.roles `pyrobusta.roles` consists of one or more authorization blocks. Each block defines one or diff --git a/src/pyrobusta/protocol/http_basic_auth.py b/src/pyrobusta/protocol/http_basic_auth.py index f2a6d34..21460fb 100644 --- a/src/pyrobusta/protocol/http_basic_auth.py +++ b/src/pyrobusta/protocol/http_basic_auth.py @@ -15,9 +15,8 @@ from pyrobusta.protocol import http from pyrobusta.utils.patch import add_method from pyrobusta.utils.helpers import iterate_segments - -_PASSWD_LOCATION = "pyrobusta.passwd" -_ROLES_LOCATION = "pyrobusta.roles" +from pyrobusta.utils.config import get_config, CONF_PASSWD_FILE, CONF_ROLES_FILE +from pyrobusta.utils.logging import warning _MAX_ROLES = 32 _NO_POLICY = 2**_MAX_ROLES @@ -165,61 +164,69 @@ def index_roles(roles: str): return role_mask -def _load_users(config=_PASSWD_LOCATION): - with open(config, encoding="utf-8") as users: - for line in users: - comment_idx = line.find("#") - line = line[:comment_idx] if comment_idx != -1 else line - if not line: - continue - if not line.count(":") == 2: - raise ValueError() - user_sep = line.find(":") - password_sep = line.find(":", user_sep + 1) - user = line[:user_sep].strip().lower() - password_hash = line[user_sep + 1 : password_sep].strip() - roles = line[password_sep + 1 :] - role_mask = index_roles(roles) - if not user or not password_hash: - raise ValueError() - if user in _USERS: - raise ValueError() - _USERS[user] = [bytes.fromhex(password_hash), role_mask] - - -def _load_roles(config=_ROLES_LOCATION): - with open(config, encoding="utf-8") as roles: - paths = [] - attributes = {} - for line in roles: - comment_idx = line.find("#") - line = line[:comment_idx].strip() if comment_idx != -1 else line.strip() - - # Parse URL path - if line.startswith("/"): - if paths and attributes: - for path in paths: - _ATTR_TREE.insert_path(path, attributes) - paths = [line] - attributes = {} - else: - paths.append(line) - - # Parse attributes - elif line: - if not paths: +def _load_users(): + passwd_file = get_config(CONF_PASSWD_FILE) + try: + with open(passwd_file, encoding="utf-8") as users: + for line in users: + comment_idx = line.find("#") + line = line[:comment_idx] if comment_idx != -1 else line + if not line: + continue + if not line.count(":") == 2: raise ValueError() - sep = line.find(":") - if sep in (0, -1): + user_sep = line.find(":") + password_sep = line.find(":", user_sep + 1) + user = line[:user_sep].strip().lower() + password_hash = line[user_sep + 1 : password_sep].strip() + roles = line[password_sep + 1 :] + role_mask = index_roles(roles) + if not user or not password_hash: raise ValueError() - role_mask = index_roles(line[sep + 1 :]) - for attr in iterate_segments(line[0:sep].strip(), ","): - if attr in attributes: + if user in _USERS: + raise ValueError() + _USERS[user] = [bytes.fromhex(password_hash), role_mask] + except OSError: + warning("Unable to open: " + passwd_file) + + +def _load_roles(): + roles_file = get_config(CONF_ROLES_FILE) + try: + with open(roles_file, encoding="utf-8") as roles: + paths = [] + attributes = {} + for line in roles: + comment_idx = line.find("#") + line = line[:comment_idx].strip() if comment_idx != -1 else line.strip() + + # Parse URL path + if line.startswith("/"): + if paths and attributes: + for path in paths: + _ATTR_TREE.insert_path(path, attributes) + paths = [line] + attributes = {} + else: + paths.append(line) + + # Parse attributes + elif line: + if not paths: + raise ValueError() + sep = line.find(":") + if sep in (0, -1): raise ValueError() - if attr: - attributes[attr] = role_mask - for path in paths: - _ATTR_TREE.insert_path(path, attributes) + role_mask = index_roles(line[sep + 1 :]) + for attr in iterate_segments(line[0:sep].strip(), ","): + if attr in attributes: + raise ValueError() + if attr: + attributes[attr] = role_mask + for path in paths: + _ATTR_TREE.insert_path(path, attributes) + except OSError: + warning("Unable to open: " + roles_file) def _handle_auth_st(self, _): diff --git a/src/pyrobusta/utils/config.py b/src/pyrobusta/utils/config.py index ecea291..be194aa 100644 --- a/src/pyrobusta/utils/config.py +++ b/src/pyrobusta/utils/config.py @@ -33,6 +33,8 @@ def const(n): # pylint: disable=C0116 CONF_SOCKET_MAX_CON = const(9) CONF_TLS = const(10) CONF_LOG_LEVEL = const(11) +CONF_PASSWD_FILE = const(12) +CONF_ROLES_FILE = const(13) # ------------------- # Configuration state @@ -52,7 +54,7 @@ def const(n): # pylint: disable=C0116 CONF_HTTP_MEM_CAP, 0.1, CONF_HTTP_SERVED_PATHS, - ["/www", "/lib/pyrobusta"], + [normalize_path("/www"), normalize_path("/lib/pyrobusta")], CONF_HTTP_FILES_API, False, CONF_HTTP_AUTH, @@ -63,12 +65,17 @@ def const(n): # pylint: disable=C0116 False, CONF_LOG_LEVEL, "info", + CONF_PASSWD_FILE, + normalize_path("/pyrobusta.passwd"), + CONF_ROLES_FILE, + normalize_path("/pyrobusta.roles"), ] # -------------- # Public helpers # -------------- +# pylint: disable=R0911 def parse_config(key, value): """ Normalize a configuration value depending on the key. @@ -81,9 +88,11 @@ def parse_config(key, value): return float(value) if key == CONF_HTTP_SERVED_PATHS: return [normalize_path(p) for p in value.split()] - if key not in (CONF_WIFI_SSID, CONF_WIFI_PASSWORD, CONF_HTTP_AUTH): - return value.lower() - return value + if key in (CONF_PASSWD_FILE, CONF_ROLES_FILE): + return normalize_path(value) + if key in (CONF_WIFI_SSID, CONF_WIFI_PASSWORD): + return value + return value.lower() def read_config(config=CONFIG_LOCATION): @@ -125,7 +134,7 @@ def get_config(key): or the value is set to None. """ global _CONFIG_LOADED # pylint: disable=W0603 - if _CONFIG_CACHE[2 * key + 1] is None or not _CONFIG_LOADED: + if not _CONFIG_LOADED: read_config() _CONFIG_LOADED = True return _CONFIG_CACHE[2 * key + 1] diff --git a/tests/unit/http_base.py b/tests/unit/http_base.py index 35cb683..7756d50 100644 --- a/tests/unit/http_base.py +++ b/tests/unit/http_base.py @@ -11,6 +11,19 @@ class TestHttpBase(unittest.TestCase): Base class for HTTP file server module. """ + def patch_config_loader(self, config, config_module): + def open_side_effect(*args, **kwargs): + data = "\n".join(f"{k}={v}" for k, v in config.items()) + return mock_open(read_data=data)(*args, **kwargs) + + self.open_patcher = patch.object( + config_module, + "open", + side_effect=open_side_effect, + ) + self.open_patcher.start() + self.addCleanup(self.open_patcher.stop) + @classmethod def setUpClass(cls): cls.base_config = {} @@ -27,30 +40,36 @@ def setUp(self): self.cwd_patcher.start() self.addCleanup(self.cwd_patcher.stop) + # -------------------------------- + # Workspace, temporary directories + # -------------------------------- + if "tmp" not in os.listdir(self.cwd): + os.mkdir(self.cwd + "/tmp") + + if "passwd_file" in self.base_config: + self.passwd_file = self.cwd + self.base_config["passwd_file"] + with open(self.passwd_file, "w", encoding="utf-8"): + pass + if "roles_file" in self.base_config: + self.roles_file = self.cwd + self.base_config["roles_file"] + with open(self.roles_file, "w", encoding="utf-8"): + pass + # ------------------- # Patch config module # ------------------- self.config = dict(self.base_config) self.config_module = load_module("pyrobusta/utils/config.py") + self.patch_config_loader(self.config, self.config_module) + self.module_patcher = patch.dict( sys.modules, {"pyrobusta.utils.config": self.config_module}, ) + self.module_patcher.start() self.addCleanup(self.module_patcher.stop) - def open_side_effect(*args, **kwargs): - data = "\n".join(f"{k}={v}" for k, v in self.config.items()) - return mock_open(read_data=data)(*args, **kwargs) - - self.open_patcher = patch.object( - self.config_module, - "open", - side_effect=open_side_effect, - ) - self.open_patcher.start() - self.addCleanup(self.open_patcher.stop) - # ------------------------------------------------ # Load remaining modules, enable optional features # ------------------------------------------------ @@ -72,3 +91,12 @@ def open_side_effect(*args, **kwargs): buffer_module = load_module("pyrobusta/stream/buffer.py") self.rx = buffer_module.SlidingBuffer(bytearray(1024)) self.tx = buffer_module.SlidingBuffer(bytearray(1024)) + + def tearDown(self): + try: + if "passwd_file" in self.config: + os.remove(self.passwd_file) + if "roles_file" in self.config: + os.remove(self.roles_file) + finally: + super().tearDown() diff --git a/tests/unit/test_http_basic_auth.py b/tests/unit/test_http_basic_auth.py index 6ca9976..9778068 100644 --- a/tests/unit/test_http_basic_auth.py +++ b/tests/unit/test_http_basic_auth.py @@ -2,7 +2,6 @@ import unittest import hashlib import base64 -import tempfile from http_base import TestHttpBase @@ -14,8 +13,12 @@ class TestBasicAuthPolicyStateMachine(TestHttpBase): @classmethod def setUpClass(cls): - cls.base_config = {"http_auth": "basic"} cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } def test_basic_auth_public_resource(self): self.basic_auth_module._ATTR_TREE.insert_path( @@ -56,8 +59,12 @@ class TestBasicAuthStateMachine(TestHttpBase): @classmethod def setUpClass(cls): - cls.base_config = {"http_auth": "basic"} cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } def prepare_auth(self, user, password, auth_header, user_roles, route_roles): if user is not None and password is not None: @@ -302,8 +309,12 @@ class TestBasicAuthPrefixTree(TestHttpBase): @classmethod def setUpClass(cls): - cls.base_config = {"http_auth": "basic"} cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } def setup_roles(self, roles: dict): self.attr_tree = self.basic_auth_module.AttributeNode("") @@ -440,8 +451,12 @@ class TestBasicAuthUserConfigReader(TestHttpBase): @classmethod def setUpClass(cls): - cls.base_config = {"http_auth": "basic"} cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } def test_user_reader_valid_config(self): file_content = ( @@ -451,13 +466,11 @@ def test_user_reader_valid_config(self): "user-3:280b73d2ac8375035501e5c06acb4b770e74ec95f479e6e2643227d7f57ce7ad:role1,role2\n" ) - with tempfile.NamedTemporaryFile( - mode="w", encoding="utf-8", dir=self.cwd - ) as file: + with open(self.passwd_file, "w", encoding="utf-8") as passwd: self.basic_auth_module._USERS.clear() - file.write(file_content) - file.flush() - self.basic_auth_module._load_users(file.name) + passwd.write(file_content) + + self.basic_auth_module._load_users() self.assertDictEqual( self.basic_auth_module._USERS, @@ -488,13 +501,12 @@ def test_user_reader_empty_user(self): ":70b0577b18482fd0e42b92ba9a1ce90ac3b976648aa68fb8484098f76f816145:\n" ) - with tempfile.NamedTemporaryFile( - mode="w", encoding="utf-8", dir=self.cwd - ) as file: - file.write(file_content) - file.flush() - with self.assertRaises(ValueError): - self.basic_auth_module._load_users(file.name) + with open(self.passwd_file, "w", encoding="utf-8") as passwd: + self.basic_auth_module._USERS.clear() + passwd.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_users() def test_user_reader_duplicate_user(self): file_content = ( @@ -502,24 +514,22 @@ def test_user_reader_duplicate_user(self): "user-1:718a8ca4dd4d30b8dc0756ad9d7de727079869b215b03782279e372ab6911ecf:\n" ) - with tempfile.NamedTemporaryFile( - mode="w", encoding="utf-8", dir=self.cwd - ) as file: - file.write(file_content) - file.flush() - with self.assertRaises(ValueError): - self.basic_auth_module._load_users(file.name) + with open(self.passwd_file, "w", encoding="utf-8") as passwd: + self.basic_auth_module._USERS.clear() + passwd.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_users() def test_user_reader_empty_password(self): file_content = "user-1::\n" - with tempfile.NamedTemporaryFile( - mode="w", encoding="utf-8", dir=self.cwd - ) as file: - file.write(file_content) - file.flush() - with self.assertRaises(ValueError): - self.basic_auth_module._load_users(file.name) + with open(self.passwd_file, "w", encoding="utf-8") as passwd: + self.basic_auth_module._USERS.clear() + passwd.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_users() class TestBasicAuthRoleConfigReader(TestHttpBase): @@ -529,8 +539,12 @@ class TestBasicAuthRoleConfigReader(TestHttpBase): @classmethod def setUpClass(cls): - cls.base_config = {"http_auth": "basic"} cls.cwd = os.getcwd() + cls.base_config = { + "http_auth": "basic", + "passwd_file": "/tmp/pyrobusta.passwd", + "roles_file": "/tmp/pyrobusta.roles", + } def test_role_reader_valid_config(self): file_content = ( @@ -549,13 +563,11 @@ def test_role_reader_valid_config(self): "POST , PUT : role_1 , role_2 " ) - with tempfile.NamedTemporaryFile( - mode="w", encoding="utf-8", dir=self.cwd - ) as file: + with open(self.roles_file, "w", encoding="utf-8") as roles: self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") - file.write(file_content) - file.flush() - self.basic_auth_module._load_roles(file.name) + roles.write(file_content) + + self.basic_auth_module._load_roles() attributes = self.basic_auth_module._ATTR_TREE.get_attributes("/index.html") self.assertEqual(attributes, {"*": self.basic_auth_module._NO_POLICY}) @@ -585,26 +597,22 @@ def test_role_reader_valid_config(self): def test_role_reader_missing_path(self): file_content = " *:*\n" - with tempfile.NamedTemporaryFile( - mode="w", encoding="utf-8", dir=self.cwd - ) as file: + with open(self.roles_file, "w", encoding="utf-8") as roles: self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") - file.write(file_content) - file.flush() - with self.assertRaises(ValueError): - self.basic_auth_module._load_roles(file.name) + roles.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_roles() def test_role_reader_duplicate_attribute(self): file_content = "/app/resource\n" + " GET: role_1\n" + " GET: role_2\n" - with tempfile.NamedTemporaryFile( - mode="w", encoding="utf-8", dir=self.cwd - ) as file: + with open(self.roles_file, "w", encoding="utf-8") as roles: self.basic_auth_module._ATTR_TREE = self.basic_auth_module.AttributeNode("") - file.write(file_content) - file.flush() - with self.assertRaises(ValueError): - self.basic_auth_module._load_roles(file.name) + roles.write(file_content) + + with self.assertRaises(ValueError): + self.basic_auth_module._load_roles() if __name__ == "__main__":