[Security] Mitigate the risk of path traversal attacks by adding a request validator that rejects requests containing a 'path' argument matching the unsafe path traversal pattern.

Author: gmarciani

api/PclusterApiHandler.py

@@ -21,7 +21,7 @@
 from flask import abort, redirect, request, Blueprint
 from jose import jwt
 
-from api.exception.exceptions import RefreshTokenError
+from api.exception.exceptions import RefreshTokenError, ForbiddenPCAPIResourceError
 from api.pcm_globals import set_auth_cookies_in_context, logger, auth_cookies
 from api.security.csrf.constants import CSRF_COOKIE_NAME
 from api.security.csrf.csrf import csrf_needed
@@ -738,6 +738,13 @@ def _get_params(_request):
 
 pc = Blueprint('pc', __name__)
 
+def validate_path(path):
+    logger.info(f"validate_path: {path}")
+    # Block path traversal attack
+    if not path.startswith('/v'):
+        logger.info(f"validate_path: EXCEPTION DRYRUN")
+        # raise ForbiddenPCAPIResourceError()
+
 @pc.get('/', strict_slashes=False)
 @authenticated({'admin'})
 @validated(params=PCProxyArgs)

api/exception/exceptions.py

@@ -22,4 +22,14 @@ def __init__(self, description=None):
             self.description = self.ERROR_FMT.format(description=description)
 
     def __str__(self):
-        return self.description
+        return self.description
+
+class ForbiddenPCAPIResourceError(Forbidden):
+    """Raise if the client requests access to a ParallelCluster API resource it is not authorized for.
+    Generates a 403 Forbidden response with the failure reason by default.
+    """
+
+    DEFAULT_DESCRIPTION = "Forbidden access to ParallelCluster API resource."
+
+    def __init__(self, description=DEFAULT_DESCRIPTION):
+        self.description = description

api/tests/validation/test_api_custom_validators.py

@@ -1,7 +1,7 @@
 import pytest
 from marshmallow import ValidationError
 
-from api.validation.validators import size_not_exceeding
+from api.validation.validators import size_not_exceeding, is_safe_path
 
 
 def test_size_not_exceeding():
@@ -15,4 +15,41 @@ def test_size_not_exceeding_failing():
     test_str_not_exceeding = 'a' * max_size # will produce "aaa...", max_size + 2
 
     with pytest.raises(ValidationError):
-        size_not_exceeding(test_str_not_exceeding, max_size)
+        size_not_exceeding(test_str_not_exceeding, max_size)
+
+
+@pytest.mark.parametrize(
+    "path, expected_result", [
+        pytest.param(
+            "/whatever_api_version/whatever_api_resource",
+            True,
+            id="safe api path absolute"
+        ),
+        pytest.param(
+            "whatever_api_version/whatever_api_resource",
+            True,
+            id="safe api path relative"
+        ),
+        pytest.param(
+            "/../whatever",
+            False,
+            id="unsafe path traversal 1"
+        ),
+        pytest.param(
+            "./../whatever",
+            False,
+            id="unsafe path traversal 2"
+        ),
+        pytest.param(
+            "/whatever/../whatever",
+            False,
+            id="unsafe path traversal 3"
+        ),
+        pytest.param(
+            "whatever/../whatever",
+            False,
+            id="unsafe path traversal 4"
+        ),
+    ])
+def test_is_safe_path(path: str, expected_result: bool):
+    assert is_safe_path(path) == expected_result

api/validation/schemas.py

@@ -1,7 +1,7 @@
 from marshmallow import Schema, fields, validate, INCLUDE, validates_schema
 
 from api.validation.validators import aws_region_validator, is_alphanumeric_with_hyphen, \
-    valid_api_log_levels_predicate, size_not_exceeding
+    valid_api_log_levels_predicate, size_not_exceeding, is_safe_path
 
 
 class EC2ActionSchema(Schema):
@@ -119,7 +119,7 @@ class PriceEstimateSchema(Schema):
 PriceEstimate = PriceEstimateSchema(unknown=INCLUDE)
 
 class PCProxyArgsSchema(Schema):
-    path = fields.String(required=True, validate=validate.Length(max=512))
+    path = fields.String(required=True, validate=validate.And(is_safe_path, validate.Length(max=512)))
 
 PCProxyArgs = PCProxyArgsSchema(unknown=INCLUDE)
 

api/validation/validators.py

@@ -31,4 +31,26 @@ def size_not_exceeding(data, size):
     bytes_ = bytes(json.dumps(data), 'utf-8')
     byte_size = len(bytes_)
     if byte_size > size:
-        raise ValidationError(f'Request body exceeded max size of {size} bytes')
+        raise ValidationError(f'Request body exceeded max size of {size} bytes')
+
+def is_safe_path(arg: str):
+    """
+    Validates if a given path is safe from path traversal attacks.
+
+    This function checks for the presence of directory traversal patterns
+    (../ or ..\) in the provided path string. These patterns are commonly
+    used in path traversal attacks to access files outside the intended
+    directory.
+
+    Args:
+    arg (str): The path string to validate.
+        Examples:
+        - "folder/file.txt" (safe)
+        - "../etc/passwd" (unsafe)
+        - "folder/../config" (unsafe)
+
+    Returns:
+    bool: True if the path is safe (no traversal patterns)
+          False if the path contains traversal patterns
+    """
+    return not re.search(r'\.\.[\\/]', arg)