Address potential path injection issues in OFRAK server

Author: whyitfor

ofrak_core/CHANGELOG.md

@@ -12,6 +12,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ### Fixed
 - `build_image.py` uses `OFRAK_DIR` from `extra_build_args` to identify `pytest_ofrak` location for develop builds ([#657](https://github.com/redballoonsecurity/ofrak/pull/657/))
 
+### Security
+- Adding path validation and sanitization using `werkzeug.secure_filename` and path normalization to OFRAK Server ([#664](https://github.com/redballoonsecurity/ofrak/pull/664))
+
 ## [3.3.0](https://github.com/redballoonsecurity/ofrak/compare/ofrak-v3.2.0...ofrak-v3.3.0) - 2025-10-03
 
 ### Added

ofrak_core/requirements.txt

@@ -22,4 +22,5 @@ synthol~=0.1.1
 tempfile312~=1.0.1
 typeguard~=2.13.3
 ubi-reader==0.8.12
+werkzeug==3.1.3
 xattr==0.10.1;platform_system!="Windows"

ofrak_core/setup.py

@@ -96,6 +96,7 @@ class CTypesExtension(setuptools.Extension):
         "tempfile312~=1.0.1",
         "typeguard~=2.13.3",
         "ubi-reader>=0.8.12",
+        "werkzeug>=3.0.0",
         "xattr>=0.10.1;platform_system!='Windows'",
     ],
     author="Red Balloon Security",

ofrak_core/src/ofrak/gui/server.py

@@ -17,6 +17,7 @@
 import os
 import webbrowser
 from collections import defaultdict
+from werkzeug.utils import secure_filename
 from typing import (
     Iterable,
     Optional,
@@ -1245,9 +1246,9 @@ def recurse_path_collisions(path: str, count: int) -> str:
 
         body = await request.json()
         url = body.get("url")
-        path = recurse_path_collisions(
-            os.path.join(self.projects_dir, url.split(":")[-1].split("/")[-1]), 0
-        )
+        sanitized_name = sanitize_repo_name(url)
+        safe_path = validate_safe_directory_path(self.projects_dir, sanitized_name)
+        path = recurse_path_collisions(safe_path, 0)
         project = OfrakProject.clone_from_git(url, path)
         self.projects.add(project)
         return json_response({"id": project.session_id.hex()})
@@ -1319,9 +1320,10 @@ async def get_projects_path(self, request: Request) -> Response:
     async def set_projects_path(self, request: Request) -> Response:
         body = await request.json()
         new_path = body["path"]
-        if not os.path.exists(new_path):
-            os.mkdir(new_path)
-        self.projects_dir = new_path
+        validated_path = os.path.normpath(os.path.abspath(new_path))
+        if not os.path.exists(validated_path):
+            os.mkdir(validated_path)
+        self.projects_dir = validated_path
         self.projects = self._slurp_projects_from_dir()
         return json_response(self.projects_dir)
 
@@ -1386,11 +1388,13 @@ async def get_project_by_resource_id(self, request: Request) -> Response:
 
     def _slurp_projects_from_dir(self) -> Set:
         projects = set()
-        if not os.path.exists(self.projects_dir):
-            os.makedirs(self.projects_dir)
-        for dir in os.listdir(self.projects_dir):
+        validated_projects_dir = os.path.normpath(os.path.abspath(self.projects_dir))
+        if not os.path.exists(validated_projects_dir):
+            os.makedirs(validated_projects_dir)
+        for dir in os.listdir(validated_projects_dir):
             try:
-                project = OfrakProject.init_from_path(os.path.join(self.projects_dir, dir))
+                safe_dir = validate_safe_directory_path(validated_projects_dir, dir)
+                project = OfrakProject.init_from_path(safe_dir)
                 projects.add(project)
             except Exception as e:
                 logging.warning(f"{dir} is in the projects directory but is not a valid project")
@@ -1758,3 +1762,48 @@ def _format_component_docstring(component: ComponentInterface) -> str:
         )
 
     return docstring
+
+
+def validate_safe_directory_path(base_dir: str, user_path: str) -> str:
+    """
+    Validate that a user-provided directory path is within the allowed base directory
+
+    :param base_dir: the base directory that paths must be within
+    :param user_path: the user-provided path to validate
+
+    :return: the safely resolved absolute path within the base directory
+
+    :raises ValueError: if the resolved path escapes the base directory or is invalid
+
+    """
+    base_dir_resolved = os.path.realpath(os.path.abspath(base_dir))
+
+    if os.path.isabs(user_path):
+        target_path = os.path.realpath(user_path)
+    else:
+        target_path = os.path.realpath(os.path.join(base_dir_resolved, user_path))
+
+    if not target_path.startswith(base_dir_resolved + os.sep) and target_path != base_dir_resolved:
+        raise ValueError(f"Path traversal attempt detected: {user_path}")
+
+    return target_path
+
+
+def sanitize_repo_name(git_url: str) -> str:
+    """
+    Extract and sanitize repository name from a git URL
+
+    :param git_url: the git repository URL
+
+    :return: sanitized repository name safe for use as a directory name
+
+    :raises ValueError: if the URL is invalid or produces an empty name
+    """
+    repo_name = git_url.split(":")[-1].split("/")[-1]
+    repo_name = repo_name.replace(".git", "")
+    sanitized = secure_filename(repo_name)
+
+    if not sanitized:
+        raise ValueError(f"Invalid repository URL: {git_url}")
+
+    return sanitized

ofrak_core/tests/unit/test_ofrak_server.py

@@ -17,7 +17,12 @@
 from ofrak import Analyzer, Unpacker, Modifier, Packer
 from ofrak.core import File
 from ofrak.core.entropy import DataSummaryAnalyzer
-from ofrak.gui.server import AiohttpOFRAKServer, start_server
+from ofrak.gui.server import (
+    AiohttpOFRAKServer,
+    start_server,
+    validate_safe_directory_path,
+    sanitize_repo_name,
+)
 from ofrak.model.component_filters import ComponentOrMetaFilter, ComponentTypeFilter
 from ofrak.service.serialization.pjson import (
     PJSONSerializationService,
@@ -2169,3 +2174,192 @@ async def test_get_project_by_resource_id(ofrak_client: TestClient, test_project
     project_resp = await ofrak_client.get(f"/{resource_id}/get_project_by_resource_id")
     project = await project_resp.json()
     assert project["session_id"] == id
+
+
+def test_validate_safe_directory_path_valid_relative(tmpdir):
+    """
+    Test that valid relative paths are properly resolved within the base directory
+
+    This test verifies that:
+    - Relative paths are converted to absolute paths
+    - The resolved path stays within the base directory
+    - Path normalization works correctly
+    """
+    base_dir = str(tmpdir)
+    result = validate_safe_directory_path(base_dir, "myproject")
+    expected = os.path.join(os.path.realpath(base_dir), "myproject")
+    assert result == expected
+
+
+def test_validate_safe_directory_path_valid_nested(tmpdir):
+    """
+    Test that nested relative paths are allowed within the base directory
+
+    This test verifies that:
+    - Multi-level relative paths are properly resolved
+    - Nested directories remain within the base directory boundary
+    """
+    base_dir = str(tmpdir)
+    result = validate_safe_directory_path(base_dir, "foo/bar/baz")
+    expected = os.path.join(os.path.realpath(base_dir), "foo/bar/baz")
+    assert result == expected
+
+
+def test_validate_safe_directory_path_traversal_attack(tmpdir):
+    """
+    Test that directory traversal attacks are blocked
+
+    This test verifies that:
+    - Path traversal attempts using ../ are detected
+    - A ValueError is raised with appropriate error message
+    - Attackers cannot escape the base directory
+    """
+    base_dir = str(tmpdir)
+    with pytest.raises(ValueError, match="Path traversal attempt detected"):
+        validate_safe_directory_path(base_dir, "../../../etc/passwd")
+
+
+def test_validate_safe_directory_path_absolute_escape(tmpdir):
+    """
+    Test that absolute paths outside the base directory are rejected
+
+    This test verifies that:
+    - Absolute paths pointing outside base directory are blocked
+    - A ValueError is raised for such attempts
+    """
+    base_dir = str(tmpdir)
+    with pytest.raises(ValueError, match="Path traversal attempt detected"):
+        validate_safe_directory_path(base_dir, "/etc/passwd")
+
+
+def test_validate_safe_directory_path_symlink_escape(tmpdir):
+    """
+    Test that symlink-based directory escapes are caught
+
+    This test verifies that:
+    - Symlinks pointing outside the base directory are detected
+    - Path resolution follows symlinks to detect escapes
+    """
+    base_dir = str(tmpdir)
+    symlink_path = tmpdir.join("evil_link")
+    os.symlink("/etc", str(symlink_path))
+
+    with pytest.raises(ValueError, match="Path traversal attempt detected"):
+        validate_safe_directory_path(base_dir, "evil_link")
+
+
+def test_validate_safe_directory_path_same_as_base(tmpdir):
+    """
+    Test that current directory reference returns the base directory
+
+    This test verifies that:
+    - Using "." as the path returns the base directory itself
+    - This is treated as a valid case (staying within bounds)
+    """
+    base_dir = str(tmpdir)
+    result = validate_safe_directory_path(base_dir, ".")
+    expected = os.path.realpath(base_dir)
+    assert result == expected
+
+
+def test_validate_safe_directory_path_double_dots_in_middle(tmpdir):
+    """
+    Test that complex path traversal attempts are handled correctly
+
+    This test verifies that:
+    - Paths with .. in the middle are normalized
+    - The final normalized path must still be within base directory
+    """
+    base_dir = str(tmpdir)
+    # This should fail because it attempts to escape
+    with pytest.raises(ValueError, match="Path traversal attempt detected"):
+        validate_safe_directory_path(base_dir, "foo/../../bar")
+
+
+def test_sanitize_repo_name_https_url():
+    """
+    Test that HTTPS git URLs are properly parsed to extract repo name
+
+    This test verifies that:
+    - HTTPS URLs are parsed correctly
+    - The .git extension is removed
+    - Only the repository name is returned
+    """
+    result = sanitize_repo_name("https://github.com/redballoonsecurity/ofrak.git")
+    assert result == "ofrak"
+
+
+def test_sanitize_repo_name_ssh_url():
+    """
+    Test that SSH git URLs are properly parsed to extract repo name
+
+    This test verifies that:
+    - SSH-style URLs (git@host:path) are parsed correctly
+    - The repository name is extracted from the path
+    """
+    result = sanitize_repo_name("git@github.com:redballoonsecurity/ofrak.git")
+    assert result == "ofrak"
+
+
+def test_sanitize_repo_name_no_git_extension():
+    """
+    Test that URLs without .git extension work correctly
+
+    This test verifies that:
+    - URLs without the .git suffix are handled
+    - The repository name is still extracted correctly
+    """
+    result = sanitize_repo_name("https://github.com/redballoonsecurity/ofrak")
+    assert result == "ofrak"
+
+
+def test_sanitize_repo_name_special_characters():
+    """
+    Test that special characters are removed by secure_filename
+
+    This test verifies that:
+    - Path traversal characters in repo names are sanitized
+    - The result is safe for use as a directory name
+    """
+    result = sanitize_repo_name("https://github.com/user/../malicious/repo.git")
+    # secure_filename should remove or replace dangerous characters
+    assert ".." not in result
+    assert "/" not in result
+
+
+def test_sanitize_repo_name_empty_result():
+    """
+    Test that invalid URLs that produce empty names are rejected
+
+    This test verifies that:
+    - URLs that result in empty repo names raise ValueError
+    - The error message indicates an invalid repository URL
+    """
+    with pytest.raises(ValueError, match="Invalid repository URL"):
+        sanitize_repo_name("https://github.com/user/")
+
+
+def test_sanitize_repo_name_only_special_chars():
+    """
+    Test that URLs that sanitize to empty strings are rejected
+
+    This test verifies that:
+    - URLs containing only special characters are detected
+    - A ValueError is raised with appropriate message
+    """
+    with pytest.raises(ValueError, match="Invalid repository URL"):
+        sanitize_repo_name("https://github.com/user/../../../.git")
+
+
+def test_sanitize_repo_name_unicode():
+    """
+    Test that unicode characters are handled by secure_filename
+
+    This test verifies that:
+    - Unicode characters in repository names are converted to ASCII
+    - The result is safe for filesystem use
+    """
+    result = sanitize_repo_name("https://github.com/user/repö.git")
+    # secure_filename should convert to ASCII-safe characters
+    assert result.isascii()
+    assert len(result) > 0

ofrak_core/version.py

@@ -1 +1 @@
-VERSION = "3.4.0rc1"
+VERSION = "3.4.0rc2"