From 94abe79d170cc2e5b0a92fcfca83e552083e847b Mon Sep 17 00:00:00 2001 From: szeka9 Date: Sun, 26 Jul 2026 22:06:06 +0200 Subject: [PATCH 1/2] Rename helper module for path normalization Rename pyrobusta.utils.helpers to "lexpath" as the module mostly contains helper functions related to lexical path normalization and validation. --- docs/application_development/request.md | 4 +-- example/mip_repo/app.py | 4 +-- src/pyrobusta/protocol/http.py | 8 +++--- src/pyrobusta/protocol/http_basic_auth.py | 2 +- src/pyrobusta/protocol/http_file_server.py | 2 +- src/pyrobusta/server/http_server.py | 2 +- src/pyrobusta/utils/assets.py | 2 +- src/pyrobusta/utils/config.py | 2 +- .../utils/{helpers.py => lexpath.py} | 0 tests/functional/test_http.py | 2 +- tests/unit/http_base.py | 4 +-- tests/unit/test_helpers.py | 28 +++++++++---------- 12 files changed, 30 insertions(+), 30 deletions(-) rename src/pyrobusta/utils/{helpers.py => lexpath.py} (100%) diff --git a/docs/application_development/request.md b/docs/application_development/request.md index 9552840..ff8119d 100644 --- a/docs/application_development/request.md +++ b/docs/application_development/request.md @@ -118,7 +118,7 @@ incrementally rather than assuming the full payload is available at once. ``` import pyrobusta.server.http_server as http_server from pyrobusta.protocol.http import HttpEngine -from pyrobusta.utils.helpers import normalize_path +from pyrobusta.utils.lexpath import normalize_path @HttpEngine.route("/app/chunks", "POST") def upload_chunks(http_ctx, payload: bytes): @@ -174,7 +174,7 @@ from os import listdir, remove, rename, mkdir import pyrobusta.server.http_server as http_server from pyrobusta.protocol.http import HttpEngine -from pyrobusta.utils.helpers import normalize_path +from pyrobusta.utils.lexpath import normalize_path @HttpEngine.route("/app/parts", "POST") def handle_parts(http_ctx, payload: tuple): diff --git a/example/mip_repo/app.py b/example/mip_repo/app.py index 8332230..7539b3a 100644 --- a/example/mip_repo/app.py +++ b/example/mip_repo/app.py @@ -2,14 +2,14 @@ import pyrobusta.server.http_server as http_server from pyrobusta.protocol.http import HttpEngine -from pyrobusta.utils import logging, config, assets, helpers +from pyrobusta.utils import logging, config, assets, lexpath def append_package_files(dir, package_files, host_name, protocol): """ Construct package file list recursively. """ - dir = helpers.normalize_path(dir) + dir = lexpath.normalize_path(dir) for asset in assets.iterate_fs(dir): package_files["urls"].append( diff --git a/src/pyrobusta/protocol/http.py b/src/pyrobusta/protocol/http.py index 50bc356..0e310ed 100644 --- a/src/pyrobusta/protocol/http.py +++ b/src/pyrobusta/protocol/http.py @@ -14,7 +14,7 @@ CONF_HTTP_SERVED_PATHS, CONF_HTTP_AUTH, ) -from ..utils import logging, helpers +from ..utils import logging, lexpath from ..stream.buffer import BufferFullError @@ -299,7 +299,7 @@ def get_cookie(self, name, default=None): """ cookie_header = self.headers.get("cookie", "") name = name.lower() - for part in helpers.iterate_segments(cookie_header, ";"): + for part in lexpath.iterate_segments(cookie_header, ";"): cookie_sep = part.find("=") if cookie_sep == -1: continue @@ -846,8 +846,8 @@ def _fs_retrieve_st(self, _): else: target_path = "/www" + self.url.decode("ascii") - norm_path = helpers.normalize_path(target_path) - is_path_served = helpers.is_norm_path_served( + norm_path = lexpath.normalize_path(target_path) + is_path_served = lexpath.is_norm_path_served( norm_path, get_config(CONF_HTTP_SERVED_PATHS) ) diff --git a/src/pyrobusta/protocol/http_basic_auth.py b/src/pyrobusta/protocol/http_basic_auth.py index 587bd8d..7b9f54a 100644 --- a/src/pyrobusta/protocol/http_basic_auth.py +++ b/src/pyrobusta/protocol/http_basic_auth.py @@ -15,7 +15,7 @@ from pyrobusta.protocol import http from pyrobusta.utils.patch import add_method -from pyrobusta.utils.helpers import iterate_segments +from pyrobusta.utils.lexpath import iterate_segments from pyrobusta.utils.crypto import ( constant_time_equal, create_signed_token, diff --git a/src/pyrobusta/protocol/http_file_server.py b/src/pyrobusta/protocol/http_file_server.py index 93281b8..99dfb5d 100644 --- a/src/pyrobusta/protocol/http_file_server.py +++ b/src/pyrobusta/protocol/http_file_server.py @@ -8,7 +8,7 @@ from json import dumps from pyrobusta.protocol import http -from pyrobusta.utils.helpers import ( +from pyrobusta.utils.lexpath import ( normalize_path, is_norm_path_served, is_file_path_valid, diff --git a/src/pyrobusta/server/http_server.py b/src/pyrobusta/server/http_server.py index c8e838e..4d12b82 100644 --- a/src/pyrobusta/server/http_server.py +++ b/src/pyrobusta/server/http_server.py @@ -17,7 +17,7 @@ CONF_TLS, CONF_SOCKET_MAX_CON, ) -from ..utils.helpers import normalize_path +from ..utils.lexpath import normalize_path from ..utils import logging diff --git a/src/pyrobusta/utils/assets.py b/src/pyrobusta/utils/assets.py index c5f5a12..78d2f2d 100644 --- a/src/pyrobusta/utils/assets.py +++ b/src/pyrobusta/utils/assets.py @@ -4,7 +4,7 @@ from os import mkdir, listdir, stat -from .helpers import normalize_path +from .lexpath import normalize_path FS_ITER_ABS = 0 FS_ITER_REL = 1 diff --git a/src/pyrobusta/utils/config.py b/src/pyrobusta/utils/config.py index d11045a..362b16c 100644 --- a/src/pyrobusta/utils/config.py +++ b/src/pyrobusta/utils/config.py @@ -12,7 +12,7 @@ def const(n): # pylint: disable=C0116 return n -from .helpers import normalize_path +from .lexpath import normalize_path PYROBUSTA_VERSION = "v0.8.0" CONFIG_LOCATION = "pyrobusta.env" diff --git a/src/pyrobusta/utils/helpers.py b/src/pyrobusta/utils/lexpath.py similarity index 100% rename from src/pyrobusta/utils/helpers.py rename to src/pyrobusta/utils/lexpath.py diff --git a/tests/functional/test_http.py b/tests/functional/test_http.py index 344dd45..83ac320 100644 --- a/tests/functional/test_http.py +++ b/tests/functional/test_http.py @@ -14,7 +14,7 @@ from pyrobusta.server import http_server from pyrobusta.protocol.http import HttpEngine -from pyrobusta.utils.helpers import normalize_path +from pyrobusta.utils.lexpath import normalize_path @HttpEngine.route("/test/simple", "GET") diff --git a/tests/unit/http_base.py b/tests/unit/http_base.py index 4cd4645..7be8641 100644 --- a/tests/unit/http_base.py +++ b/tests/unit/http_base.py @@ -33,9 +33,9 @@ def setUp(self): # ------------------------------- # Patch current working directory # ------------------------------- - self.helpers_module = load_module("pyrobusta/utils/helpers.py") + self.lexpath_module = load_module("pyrobusta/utils/lexpath.py") self.cwd_patcher = patch.object( - self.helpers_module, "getcwd", return_value=self.cwd + self.lexpath_module, "getcwd", return_value=self.cwd ) self.cwd_patcher.start() self.addCleanup(self.cwd_patcher.stop) diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py index fc0834f..a8bcb71 100644 --- a/tests/unit/test_helpers.py +++ b/tests/unit/test_helpers.py @@ -15,7 +15,7 @@ def setUpClass(cls): cls.config = {} def setUp(self): - self.helpers_module = load_module("pyrobusta/utils/helpers.py") + self.lexpath_module = load_module("pyrobusta/utils/lexpath.py") def test_path_normalization_virtual_root(self): """ @@ -35,9 +35,9 @@ def test_path_normalization_virtual_root(self): ("/path/../../resource", f"{cwd}/resource"), ("/path/../../resource/..", f"{cwd}"), ): - self.assertEqual(self.helpers_module.normalize_path(case[0]), case[1]) + self.assertEqual(self.lexpath_module.normalize_path(case[0]), case[1]) - @patch("pyrobusta.utils.helpers.getcwd", return_value="/") + @patch("pyrobusta.utils.lexpath.getcwd", return_value="/") def test_path_normalization_host_root(self, _): """ Test lexical path normalization assuming the working directory @@ -55,9 +55,9 @@ def test_path_normalization_host_root(self, _): ("/path/../../resource", "/resource"), ("/path/../../resource/..", "/"), ): - self.assertEqual(self.helpers_module.normalize_path(case[0]), case[1]) + self.assertEqual(self.lexpath_module.normalize_path(case[0]), case[1]) - @patch("pyrobusta.utils.helpers.getcwd", return_value="/") + @patch("pyrobusta.utils.lexpath.getcwd", return_value="/") def test_path_serving_list(self, _): served_paths = ["/path/to/dir1", "/path/to/dir2"] @@ -73,10 +73,10 @@ def test_path_serving_list(self, _): ("/path/to", False), ): self.assertEqual( - self.helpers_module.is_norm_path_served(case[0], served_paths), case[1] + self.lexpath_module.is_norm_path_served(case[0], served_paths), case[1] ) - @patch("pyrobusta.utils.helpers.getcwd", return_value="/") + @patch("pyrobusta.utils.lexpath.getcwd", return_value="/") def test_path_serving_root(self, _): served_paths = ["/"] @@ -86,10 +86,10 @@ def test_path_serving_root(self, _): ("/path/to/served", True), ): self.assertEqual( - self.helpers_module.is_norm_path_served(case[0], served_paths), case[1] + self.lexpath_module.is_norm_path_served(case[0], served_paths), case[1] ) - @patch("pyrobusta.utils.helpers.getcwd", return_value="/") + @patch("pyrobusta.utils.lexpath.getcwd", return_value="/") def test_path_serving_none(self, _): served_paths = [] @@ -99,7 +99,7 @@ def test_path_serving_none(self, _): ("/path/to/served", False), ): self.assertEqual( - self.helpers_module.is_norm_path_served(case[0], served_paths), case[1] + self.lexpath_module.is_norm_path_served(case[0], served_paths), case[1] ) def test_path_segment_validation(self): @@ -114,10 +114,10 @@ def test_path_segment_validation(self): ] for segment in valid_segments: - self.assertTrue(self.helpers_module.is_path_segment_valid(segment)) + self.assertTrue(self.lexpath_module.is_path_segment_valid(segment)) for segment in invalid_segments: - self.assertFalse(self.helpers_module.is_path_segment_valid(segment)) + self.assertFalse(self.lexpath_module.is_path_segment_valid(segment)) def test_file_path_validation(self): valid_paths = ["/file", "/dir1/file", "/dir-2/file", "/dir_3/file"] @@ -131,7 +131,7 @@ def test_file_path_validation(self): ] for path in valid_paths: - self.assertTrue(self.helpers_module.is_file_path_valid(path)) + self.assertTrue(self.lexpath_module.is_file_path_valid(path)) for path in invalid_paths: - self.assertFalse(self.helpers_module.is_file_path_valid(path)) + self.assertFalse(self.lexpath_module.is_file_path_valid(path)) From 5b0007bb1c22afc3d5ee6276c2dd37188597cddb Mon Sep 17 00:00:00 2001 From: szeka9 Date: Sun, 26 Jul 2026 22:09:09 +0200 Subject: [PATCH 2/2] Prevent server configuration files to be served over HTTP Add a new helper function to determine if a file is used for configuration. Prevent such files to be served over HTTP. Although the default configuration prevented the aforementioned scenario, misconfiguration still allows config files to be accidentally served. --- src/pyrobusta/protocol/http.py | 16 ++-- src/pyrobusta/protocol/http_file_server.py | 9 +- src/pyrobusta/utils/config.py | 14 +++- src/pyrobusta/utils/lexpath.py | 8 +- tests/functional/test_http_file_server.py | 98 +++++++++++++++++++++- tests/unit/test_helpers.py | 6 +- 6 files changed, 129 insertions(+), 22 deletions(-) diff --git a/src/pyrobusta/protocol/http.py b/src/pyrobusta/protocol/http.py index 0e310ed..edb8216 100644 --- a/src/pyrobusta/protocol/http.py +++ b/src/pyrobusta/protocol/http.py @@ -9,6 +9,7 @@ from ..utils.config import ( get_config, + is_protected_file, CONF_HTTP_MULTIPART, CONF_HTTP_FILES_API, CONF_HTTP_SERVED_PATHS, @@ -836,6 +837,15 @@ def _handle_route_st(self, rx): self._handle_route_response(handler_response) + @staticmethod + def is_norm_path_served(norm_path: str): + """ + Returns true if a directory is configured to be served. + """ + return lexpath.is_child_path_of( + norm_path, get_config(CONF_HTTP_SERVED_PATHS) + ) and not is_protected_file(norm_path) + def _fs_retrieve_st(self, _): """ State for retrieving a file under /www. @@ -845,14 +855,10 @@ def _fs_retrieve_st(self, _): target_path = "/www/index.html" else: target_path = "/www" + self.url.decode("ascii") - norm_path = lexpath.normalize_path(target_path) - is_path_served = lexpath.is_norm_path_served( - norm_path, get_config(CONF_HTTP_SERVED_PATHS) - ) try: - if not is_path_served: + if not self.is_norm_path_served(norm_path): stat(norm_path) self.terminate(403) return diff --git a/src/pyrobusta/protocol/http_file_server.py b/src/pyrobusta/protocol/http_file_server.py index 99dfb5d..90de7f2 100644 --- a/src/pyrobusta/protocol/http_file_server.py +++ b/src/pyrobusta/protocol/http_file_server.py @@ -10,17 +10,11 @@ from pyrobusta.protocol import http from pyrobusta.utils.lexpath import ( normalize_path, - is_norm_path_served, is_file_path_valid, is_path_segment_valid, ) from pyrobusta.utils.assets import iterate_fs, FS_ITER_FILE -from ..utils.config import ( - get_config, - CONF_HTTP_SERVED_PATHS, -) - _UPLOAD_ROOT = normalize_path("/www/user_data") _TMP_DIR = normalize_path("/tmp") @@ -38,10 +32,9 @@ def fs_retrieve(http_ctx, _): """ target_path = http_ctx.url[len(b"/files") :].decode("ascii") norm_path = normalize_path(target_path) - is_path_served = is_norm_path_served(norm_path, get_config(CONF_HTTP_SERVED_PATHS)) try: - if not is_path_served: + if not http_ctx.is_norm_path_served(norm_path): stat(norm_path) http_ctx.terminate(403) return "text/plain", "Forbidden" diff --git a/src/pyrobusta/utils/config.py b/src/pyrobusta/utils/config.py index 362b16c..1d9a80b 100644 --- a/src/pyrobusta/utils/config.py +++ b/src/pyrobusta/utils/config.py @@ -15,7 +15,7 @@ def const(n): # pylint: disable=C0116 from .lexpath import normalize_path PYROBUSTA_VERSION = "v0.8.0" -CONFIG_LOCATION = "pyrobusta.env" +CONFIG_LOCATION = normalize_path("/pyrobusta.env") # ------------------------------------------- # Global runtime configuration keys. @@ -138,6 +138,18 @@ def read_config(config=CONFIG_LOCATION): pass +def is_protected_file(norm_path: str): + """ + Determines if a file path is required for core configuration + and is not meant to be served or handled by the server. + """ + return norm_path in ( + CONFIG_LOCATION, + get_config(CONF_PASSWD_FILE), + get_config(CONF_ROLES_FILE), + ) + + def get_config(key): """ Read configuration by key. diff --git a/src/pyrobusta/utils/lexpath.py b/src/pyrobusta/utils/lexpath.py index 7c6432d..55f45ba 100644 --- a/src/pyrobusta/utils/lexpath.py +++ b/src/pyrobusta/utils/lexpath.py @@ -29,18 +29,18 @@ def normalize_path(path: str): return cwd -def is_norm_path_served(path: str, served_paths: list): +def is_child_path_of(path: str, parent_paths: list): """ - Returns true if a normalized path is configured to be served. + Returns true if a normalized path is the child of a parent path. :param path: path to check - :param served_paths: list of paths configured to be served + :param parent_paths: parent paths to check against """ parts = path.split("/") for i, _ in enumerate(parts): current_path = "/".join(parts[: i + 1]) if not current_path: current_path = "/" - if current_path in served_paths: + if current_path in parent_paths: return True return False diff --git a/tests/functional/test_http_file_server.py b/tests/functional/test_http_file_server.py index 3439c67..07fd93c 100644 --- a/tests/functional/test_http_file_server.py +++ b/tests/functional/test_http_file_server.py @@ -13,7 +13,13 @@ delete_path, ) -from pyrobusta.utils.config import normalize_path +from pyrobusta.utils.config import ( + normalize_path, + get_config, + CONF_PASSWD_FILE, + CONF_ROLES_FILE, + CONFIG_LOCATION, +) @garbage_collect @@ -134,6 +140,95 @@ async def test_fs_access_control(): await server.terminate() +@garbage_collect +async def test_misconfigured_path(): + setup_config(files_api_enabled=True, served_paths="/") + server = await start_server() + + with open(normalize_path("/allowed"), "w") as roles: + roles.write( + "This file is located under root and is served due to configuration." + ) + + with open(CONFIG_LOCATION, "w") as passwd: + passwd.write("config_key=config_value") + + with open(get_config(CONF_PASSWD_FILE), "w") as passwd: + passwd.write( + "user1:0b14d501a594442a01c6859541bcb3e8164d183d32937b851835442f69d5c94e:role1" + ) + + with open(get_config(CONF_ROLES_FILE), "w") as roles: + roles.write("/*\n*:*") + + try: + # Case #1: /test/allowed + # Served due to configuration + response = await send_request( + b"GET /files/allowed HTTP/1.1\r\n" + b"Content-Length: 0\r\n" + b"Connection: close\r\n" + b"Host: localhost\r\n\r\n" + ) + + response_body = response.split(b"\r\n\r\n")[1] + test_assert( + f"FS access misconfiguration - test file downloaded from root", + response_body, + b"This file is located under root and is served due to configuration.", + ) + + # Case #2: /test/pyrobusta.env + # Must be rejected by policy + config_file = CONFIG_LOCATION.split("/")[-1] + response = await send_request( + b"GET /files/" + config_file.encode() + b" HTTP/1.1\r\n" + b"Content-Length: 0\r\n" + b"Connection: close\r\n" + b"Host: localhost\r\n\r\n" + ) + + test_assert( + f"FS access misconfiguration - config_file file restricted", + response.startswith(b"HTTP/1.1 403 Forbidden"), + True, + ) + + # Case #3: /test/ + # Must be rejected by policy + passwd_file = get_config(CONF_PASSWD_FILE).split("/")[-1] + response = await send_request( + b"GET /files/" + passwd_file.encode() + b" HTTP/1.1\r\n" + b"Content-Length: 0\r\n" + b"Connection: close\r\n" + b"Host: localhost\r\n\r\n" + ) + + test_assert( + f"FS access misconfiguration - passwd file restricted", + response.startswith(b"HTTP/1.1 403 Forbidden"), + True, + ) + + # Case #4: /test/ + # Must be rejected by policy + roles_file = get_config(CONF_ROLES_FILE).split("/")[-1] + response = await send_request( + b"GET /files/" + roles_file.encode() + b" HTTP/1.1\r\n" + b"Content-Length: 0\r\n" + b"Connection: close\r\n" + b"Host: localhost\r\n\r\n" + ) + + test_assert( + f"FS access misconfiguration - roles file restricted", + response.startswith(b"HTTP/1.1 403 Forbidden"), + True, + ) + finally: + await server.terminate() + + @garbage_collect async def test_bulk_file_upload(): setup_config(files_api_enabled=True, http_multipart_enabled=True) @@ -239,6 +334,7 @@ async def test_chunked_file_upload(): async def test_main(): await test_fs_path_traversal() await test_fs_access_control() + await test_misconfigured_path() await test_bulk_file_upload() await test_chunked_file_upload() diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py index a8bcb71..8cb5b4b 100644 --- a/tests/unit/test_helpers.py +++ b/tests/unit/test_helpers.py @@ -73,7 +73,7 @@ def test_path_serving_list(self, _): ("/path/to", False), ): self.assertEqual( - self.lexpath_module.is_norm_path_served(case[0], served_paths), case[1] + self.lexpath_module.is_child_path_of(case[0], served_paths), case[1] ) @patch("pyrobusta.utils.lexpath.getcwd", return_value="/") @@ -86,7 +86,7 @@ def test_path_serving_root(self, _): ("/path/to/served", True), ): self.assertEqual( - self.lexpath_module.is_norm_path_served(case[0], served_paths), case[1] + self.lexpath_module.is_child_path_of(case[0], served_paths), case[1] ) @patch("pyrobusta.utils.lexpath.getcwd", return_value="/") @@ -99,7 +99,7 @@ def test_path_serving_none(self, _): ("/path/to/served", False), ): self.assertEqual( - self.lexpath_module.is_norm_path_served(case[0], served_paths), case[1] + self.lexpath_module.is_child_path_of(case[0], served_paths), case[1] ) def test_path_segment_validation(self):