[Security] Mitigate the risk of path traversal attacks through a request validator

api/tests/exceptions/test_exception_handlers.py

@@ -47,11 +47,11 @@ def test_unauthenticated_error_handler(client, app, monkeypatch):
 def test_validation_error_handler(client, app, monkeypatch):
     with app.test_request_context('/'):
         app.preprocess_request()
-        response, status_code = validation_error_handler(ValidationError('Input validation failed for /manager/ec2_action', data={'field': ['validation-error']}))
+        response, status_code = validation_error_handler(ValidationError('Input validation failed for requested resource /manager/ec2_action', data={'field': ['validation-error']}))
 
     assert status_code == 400
     assert response.json == {
-        'code': 400, 'message': 'Input validation failed for /manager/ec2_action',
+        'code': 400, 'message': 'Input validation failed for requested resource /manager/ec2_action',
         'validation_errors': {'field': ['validation-error']}
     }
 

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/__init__.py

@@ -30,7 +30,7 @@ def wrapper(func):
         def decorated(*pargs, **kwargs):
             errors = __validate_request(request, body_schema=body, params_schema=params, cookies_schema=cookies, raise_on_missing_body=raise_on_missing_body)
             if errors:
-                raise ValidationError(f'Input validation failed for {request.path}', data=errors)
+                raise ValidationError(f'Input validation failed for requested resource {request.path}', data=errors)
             return func(*pargs, **kwargs)
 
         return decorated

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,27 @@ 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:
+        - "/v3/clusters" (safe)
+        - "v3/clusters" (safe)
+         "/v3/../../stage/v3/clusters" (safe)
+        - "../stage/v3/clusters" (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)