INTER-2511: [api-v3] Reject doth parameters for request and visitor IDs

.changeset/reject-dot-segment-path-parameters.md

@@ -0,0 +1,5 @@
+---
+'fingerprint-pro-server-api-python-sdk': patch
+---
+
+Reject `.` and `..` as request and visitor IDs. `get_event`, `update_event`, `get_visits`, and `delete_visitor_data` now raise the new `InvalidArgumentError` without sending a request.

docs/FingerprintApi.md

@@ -72,6 +72,11 @@ Name | Type | Description  | Notes
 ------------- | ------------- | ------------- | -------------
  **visitor_id** | **str**| The [visitor ID](https://dev.fingerprint.com/reference/get-function#visitorid) you want to delete. | 
 
+> [!WARNING]
+> #### Invalid values
+>
+> `visitor_id` must not be `.` or `..`. When you use them, the call raises `InvalidArgumentError` without sending a request.
+
 ### Return type
 
 void (empty response body)
@@ -131,6 +136,11 @@ Name | Type | Description  | Notes
 ------------- | ------------- | ------------- | -------------
  **request_id** | **str**| The unique [identifier](https://dev.fingerprint.com/reference/get-function#requestid) of each identification request. | 
 
+> [!WARNING]
+> #### Invalid values
+>
+> `request_id` must not be `.` or `..`. When you use them, the call raises `InvalidArgumentError` without sending a request.
+
 ### Return type
 
 [**EventsGetResponse**](EventsGetResponse.md)
@@ -264,6 +274,11 @@ Name | Type | Description  | Notes
  **pagination_key** | **str**| Deprecated pagination parameter retained for backward compatibility.  `GET /visitors/{visitor_id}` currently returns at most one visit, so pagination is not expected. Use `GET /events/search` for paginated results.  | [optional] 
  **before** | **int**| ⚠️ Deprecated pagination method, please use `paginationKey` instead. Timestamp (in milliseconds since epoch) used to paginate results. `GET /visitors/{visitor_id}` currently returns at most one visit, so pagination is not expected.  | [optional] 
 
+> [!WARNING]
+> #### Invalid values
+>
+> `visitor_id` must not be `.` or `..`. When you use them, the call raises `InvalidArgumentError` without sending a request.
+
 ### Return type
 
 [**VisitorsGetResponse**](VisitorsGetResponse.md)
@@ -453,6 +468,11 @@ Name | Type | Description  | Notes
  **body** | [**EventsUpdateRequest**](EventsUpdateRequest.md)|  | 
  **request_id** | **str**| The unique event [identifier](https://dev.fingerprint.com/reference/get-function#requestid). | 
 
+> [!WARNING]
+> #### Invalid values
+>
+> `request_id` must not be `.` or `..`. When you use them, the call raises `InvalidArgumentError` without sending a request.
+
 ### Return type
 
 void (empty response body)

fingerprint_pro_server_api_sdk/__init__.py

@@ -25,6 +25,7 @@
 from fingerprint_pro_server_api_sdk.api.fingerprint_api import FingerprintApi
 # import ApiClient
 from fingerprint_pro_server_api_sdk.api_client import ApiClient
+from fingerprint_pro_server_api_sdk.rest import InvalidArgumentError
 from fingerprint_pro_server_api_sdk.configuration import Configuration
 # import BaseModel
 from fingerprint_pro_server_api_sdk.base_model import BaseModel

fingerprint_pro_server_api_sdk/api_client.py

@@ -22,7 +22,7 @@
 from fingerprint_pro_server_api_sdk.configuration import Configuration
 import fingerprint_pro_server_api_sdk.models
 from fingerprint_pro_server_api_sdk import rest
-from fingerprint_pro_server_api_sdk.rest import ApiException, RESTResponse
+from fingerprint_pro_server_api_sdk.rest import ApiException, InvalidArgumentError, RESTResponse
 from fingerprint_pro_server_api_sdk.base_model import BaseModel
 
 PRIMITIVE_TYPES = (float, bool, bytes, str, int)
@@ -125,10 +125,10 @@ def __call_api(
                                                     collection_formats)
             for k, v in path_params:
                 # specified safe chars, encode everything
-                resource_path = resource_path.replace(
-                    '{%s}' % k,
-                    quote(str(v), safe=config.safe_chars_for_path_param)
-                )
+                encoded = quote(str(v), safe=config.safe_chars_for_path_param)
+                if encoded in ('.', '..'):
+                    raise InvalidArgumentError(k, str(v))
+                resource_path = resource_path.replace('{%s}' % k, encoded)
 
         # query parameters
         if query_params:

fingerprint_pro_server_api_sdk/rest.py

@@ -260,3 +260,14 @@ def __init__(self, original_exception: ApiException, structured_error: Any):
         self.headers = original_exception.headers
         self.body = original_exception.body
         self.structured_error = structured_error
+
+
+class InvalidArgumentError(ValueError):
+    """Exception when ``.`` or ``..`` used as resource identifier."""
+
+    def __init__(self, argument: str, value: str):
+        self.argument = argument
+        self.value = value
+        super(InvalidArgumentError, self).__init__(
+            "invalid value {0!r} for {1}: not a valid identifier".format(value, argument)
+        )

template/__init__package.mustache

@@ -18,6 +18,7 @@ warnings.warn(
 {{/apis}}{{/apiInfo}}
 # import ApiClient
 from {{packageName}}.api_client import ApiClient
+from {{packageName}}.rest import InvalidArgumentError
 from {{packageName}}.configuration import Configuration
 # import BaseModel
 from {{packageName}}.base_model import BaseModel

template/api_client.mustache

@@ -14,7 +14,7 @@ from datetime import date, datetime
 from {{packageName}}.configuration import Configuration
 import {{modelPackage}}
 from {{packageName}} import rest
-from {{packageName}}.rest import ApiException, RESTResponse
+from {{packageName}}.rest import ApiException, InvalidArgumentError, RESTResponse
 from {{packageName}}.base_model import BaseModel
 
 PRIMITIVE_TYPES = (float, bool, bytes, str, int)
@@ -117,10 +117,10 @@ class ApiClient:
                                                     collection_formats)
             for k, v in path_params:
                 # specified safe chars, encode everything
-                resource_path = resource_path.replace(
-                    '{%s}' % k,
-                    quote(str(v), safe=config.safe_chars_for_path_param)
-                )
+                encoded = quote(str(v), safe=config.safe_chars_for_path_param)
+                if encoded in ('.', '..'):
+                    raise InvalidArgumentError(k, str(v))
+                resource_path = resource_path.replace('{%s}' % k, encoded)
 
         # query parameters
         if query_params:

template/api_doc.mustache

@@ -47,6 +47,13 @@ Name | Type | Description  | Notes
 {{#allParams}} **{{paramName}}** | {{#isFile}}**{{dataType}}**{{/isFile}}{{^isFile}}{{#isPrimitiveType}}**{{dataType}}**{{/isPrimitiveType}}{{^isPrimitiveType}}[**{{dataType}}**]({{baseType}}.md){{/isPrimitiveType}}{{/isFile}}| {{{description}}} | {{^required}}[optional] {{/required}}{{#defaultValue}}[default to {{defaultValue}}]{{/defaultValue}}
 {{/allParams}}
 
+{{#pathParams}}
+> [!WARNING]
+> #### Invalid values
+>
+> `{{paramName}}` must not be `.` or `..`. When you use them, the call raises `InvalidArgumentError` without sending a request.
+
+{{/pathParams}}
 ### Return type
 
 {{#returnType}}{{#returnTypeIsPrimitive}}**{{{returnType}}}**{{/returnTypeIsPrimitive}}{{^returnTypeIsPrimitive}}[**{{{returnType}}}**]({{returnBaseType}}.md){{/returnTypeIsPrimitive}}{{/returnType}}{{^returnType}}void (empty response body){{/returnType}}

template/rest.mustache

@@ -252,3 +252,14 @@ class KnownApiException(ApiException):
         self.headers = original_exception.headers
         self.body = original_exception.body
         self.structured_error = structured_error
+
+
+class InvalidArgumentError(ValueError):
+    """Exception when ``.`` or ``..`` used as resource identifier."""
+
+    def __init__(self, argument: str, value: str):
+        self.argument = argument
+        self.value = value
+        super(InvalidArgumentError, self).__init__(
+            "invalid value {0!r} for {1}: not a valid identifier".format(value, argument)
+        )

test/test_path_params.py

@@ -0,0 +1,194 @@
+# coding: utf-8
+
+"""Tests for how path parameter values reach the Server API."""
+
+import threading
+import unittest
+
+from collections import namedtuple
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from typing import Optional
+
+from fingerprint_pro_server_api_sdk import (Configuration, EventsUpdateRequest,
+                                            InvalidArgumentError)
+from fingerprint_pro_server_api_sdk.api.fingerprint_api import FingerprintApi  # noqa: E501
+
+API_KEY = 'private_key'
+
+EMPTY_BODY = b'{}'
+EVENT_BODY = b'{"products": {}}'
+VISITS_BODY = b'{"visitorId": "visitor_id", "visits": []}'
+
+PathParamOperation = namedtuple(
+    'PathParamOperation', ['name', 'argument', 'prefix', 'call', 'response_body']
+)
+
+OPERATIONS = (
+    PathParamOperation(
+        name='get_event',
+        argument='request_id',
+        prefix='/events/',
+        call=lambda api, request_id: api.get_event(request_id),
+        response_body=EVENT_BODY,
+    ),
+    PathParamOperation(
+        name='update_event',
+        argument='request_id',
+        prefix='/events/',
+        call=lambda api, request_id: api.update_event(
+            EventsUpdateRequest(linked_id='linked_id'), request_id
+        ),
+        response_body=EMPTY_BODY,
+    ),
+    PathParamOperation(
+        name='get_visits',
+        argument='visitor_id',
+        prefix='/visitors/',
+        call=lambda api, visitor_id: api.get_visits(visitor_id),
+        response_body=VISITS_BODY,
+    ),
+    PathParamOperation(
+        name='delete_visitor_data',
+        argument='visitor_id',
+        prefix='/visitors/',
+        call=lambda api, visitor_id: api.delete_visitor_data(visitor_id),
+        response_body=EMPTY_BODY,
+    ),
+)
+
+
+class StubServer:
+    """A local HTTP server recording the request target it was last asked for."""
+
+    def __init__(self):
+        self.request_target = None  # type: Optional[str]
+        self.response_body = EMPTY_BODY
+        server = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def _handle(self):
+                server.request_target = self.path.split('?', 1)[0]
+
+                self.send_response(200)
+                self.send_header('Content-Type', 'application/json')
+                self.send_header('Content-Length', str(len(server.response_body)))
+                self.end_headers()
+                self.wfile.write(server.response_body)
+
+            do_GET = _handle
+            do_PUT = _handle
+            do_DELETE = _handle
+
+            def log_message(self, *args):
+                """Silence the default logging."""
+
+        self._server = HTTPServer(('127.0.0.1', 0), Handler)
+        self._thread = threading.Thread(target=self._server.serve_forever)
+        self._thread.daemon = True
+
+    def start(self):
+        self._thread.start()
+
+    def stop(self):
+        self._server.shutdown()
+        self._server.server_close()
+        self._thread.join()
+
+    def reset(self, response_body=EMPTY_BODY):
+        self.request_target = None
+        self.response_body = response_body
+
+    @property
+    def base_path(self):
+        host, port = self._server.server_address[:2]
+        return 'http://%s:%s/base' % (host, port)
+
+
+class TestPathParams(unittest.TestCase):
+    """Test path parameter handling for every operation that takes an ID in the path."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.server = StubServer()
+        cls.server.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.server.stop()
+
+    def setUp(self):
+        self.server.reset()
+        configuration = Configuration(api_key=API_KEY)
+        configuration.host = self.server.base_path
+        self.api = FingerprintApi(configuration)
+
+    def call(self, operation, value):
+        """Run operation with value and return the request target the server received."""
+        self.server.reset(operation.response_body)
+        operation.call(self.api, value)
+        return self.server.request_target
+
+    def test_value_is_encoded_into_a_single_path_segment(self):
+        cases = (
+            ('path', '../events', '..%2Fevents'),
+            ('nested path', '../../base/events', '..%2F..%2Fbase%2Fevents'),
+            ('leading slash', '/events/123', '%2Fevents%2F123'),
+            ('absolute url', 'https://test.com/events', 'https%3A%2F%2Ftest.com%2Fevents'),
+            ('protocol relative url', '//test.com', '%2F%2Ftest.com'),
+            ('query injection', '123?limit=1', '123%3Flimit%3D1'),
+            ('fragment injection', '123#fragment', '123%23fragment'),
+            ('whitespace', 'hello world', 'hello%20world'),
+            ('non ascii', u'\xe9', '%C3%A9'),
+            ('pre-encoded', '..%2Fevents', '..%252Fevents'),
+        )
+
+        for operation in OPERATIONS:
+            for name, value, encoded in cases:
+                with self.subTest(operation=operation.name, case=name):
+                    request_target = self.call(operation, value)
+
+                    self.assertEqual('/base' + operation.prefix + encoded, request_target)
+
+    def test_dot_in_value_is_not_touched(self):
+        """Test dots in the value stay literal."""
+        cases = (
+            ('request id', '1708102555327.NLOjmg'),
+            ('three dots', '...'),
+            ('leading dot', '.leading'),
+            ('trailing dot', 'trailing.'),
+        )
+
+        for operation in OPERATIONS:
+            for name, value in cases:
+                with self.subTest(operation=operation.name, case=name):
+                    request_target = self.call(operation, value)
+
+                    self.assertEqual('/base' + operation.prefix + value, request_target)
+
+    def test_dot_segment_is_rejected_without_sending_a_request(self):
+        """`.` and `..` are refused"""
+        for operation in OPERATIONS:
+            for value in ('.', '..'):
+                with self.subTest(operation=operation.name, value=value):
+                    self.server.reset(operation.response_body)
+
+                    with self.assertRaises(InvalidArgumentError) as context:
+                        operation.call(self.api, value)
+
+                    self.assertIsNone(self.server.request_target)
+                    self.assertEqual(operation.argument, context.exception.argument)
+                    self.assertEqual(value, context.exception.value)
+                    self.assertIn(operation.argument, str(context.exception))
+
+    def test_empty_value_does_not_address_the_collection(self):
+        """An empty ID leaves a trailing slash"""
+        for operation in OPERATIONS:
+            with self.subTest(operation=operation.name):
+                request_target = self.call(operation, '')
+
+                self.assertEqual('/base' + operation.prefix, request_target)
+                self.assertNotEqual('/base' + operation.prefix.rstrip('/'), request_target)
+
+
+if __name__ == '__main__':
+    unittest.main()