Refactor path helper module; enforce stricter file serving policy

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):

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(

src/pyrobusta/protocol/http.py

@@ -9,12 +9,13 @@
 
 from ..utils.config import (
     get_config,
+    is_protected_file,
     CONF_HTTP_MULTIPART,
     CONF_HTTP_FILES_API,
     CONF_HTTP_SERVED_PATHS,
     CONF_HTTP_AUTH,
 )
-from ..utils import logging, helpers
+from ..utils import logging, lexpath
 from ..stream.buffer import BufferFullError
 
 
@@ -299,7 +300,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
@@ -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 = helpers.normalize_path(target_path)
-        is_path_served = helpers.is_norm_path_served(
-            norm_path, get_config(CONF_HTTP_SERVED_PATHS)
-        )
+        norm_path = lexpath.normalize_path(target_path)
 
         try:
-            if not is_path_served:
+            if not self.is_norm_path_served(norm_path):
                 stat(norm_path)
                 self.terminate(403)
                 return

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,

src/pyrobusta/protocol/http_file_server.py

@@ -8,19 +8,13 @@
 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,
     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"

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
 
 

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

src/pyrobusta/utils/config.py

@@ -12,10 +12,10 @@ 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"
+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.

src/pyrobusta/utils/helpers.py → 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
 

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")

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/<passwd-file>
+        # 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/<roles-file>
+        # 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()
 

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)