APP-2226 - bypass monitor mode

README.md

@@ -28,6 +28,7 @@ Table of Contents
     * [Custom Request Handler](#custom_request_handler)
     * [Additional Activity Handler](#additional_activity_handler)
     * [Px Disable Request](#px_disable_request)
+    * [Test Block Flow on Monitoring Mode](#bypass_monitor_header)
 
 ## <a name="installation"></a> Installation
 
@@ -254,3 +255,21 @@ environ['px_disable_request'] = True #The enforcer shall be disabled for that re
 
 ```
 
+#### <a name=“bypassMonitorHeader”></a> Test Block Flow on Monitoring Mode
+
+Allows you to test an enforcer’s blocking flow while you are still in Monitor Mode.
+
+When the header name is set(eg. `x-px-block`) and the value is set to `1`, when there is a block response (for example from using a User-Agent header with the value of `PhantomJS/1.0`) the Monitor Mode is bypassed and full block mode is applied. If one of the conditions is missing you will stay in Monitor Mode. This is done per request.
+To stay in Monitor Mode, set the header value to `0`.
+
+The Header Name is configurable using the `bypass_monitor_header` property.
+
+**Default:** Empty
+
+```python
+config = {
+  ...
+  bypass_monitor_header: 'x-px-block',
+  ...
+}
+```

perimeterx/px_config.py

@@ -37,6 +37,7 @@ def __init__(self, config_dict):
         self._ip_headers = config_dict.get('ip_headers', [])
         self._proxy_url = config_dict.get('proxy_url', None)
         self._max_buffer_len = config_dict.get('max_buffer_len', 30)
+        self._bypass_monitor_header = config_dict.get('bypass_monitor_header','')
 
         sensitive_routes = config_dict.get('sensitive_routes', [])
         if not isinstance(sensitive_routes, list):
@@ -191,6 +192,10 @@ def enrich_custom_parameters(self):
     def testing_mode(self):
         return self._testing_mode
 
+    @property
+    def bypass_monitor_header(self):
+        return self._bypass_monitor_header
+
     def __instantiate_user_defined_handlers(self, config_dict):
         self._custom_request_handler = self.__set_handler('custom_request_handler', config_dict)
         self._get_user_ip = self.__set_handler('get_user_ip', config_dict)

perimeterx/px_request_verifier.py

@@ -50,9 +50,10 @@ def handle_verification(self, ctx, request):
         else:
             logger.debug('Risk score is higher or equal than blocking score')
             self.report_block_traffic(ctx)
+            should_bypass_monitor = config.bypass_monitor_header and ctx.headers.get(config.bypass_monitor_header) == '1';
             if config.additional_activity_handler:
                 config.additional_activity_handler(ctx, config)
-            if config.module_mode == px_constants.MODULE_MODE_BLOCKING:
+            if config.module_mode == px_constants.MODULE_MODE_BLOCKING or should_bypass_monitor:
                 data, headers, status = self.px_blocker.handle_blocking(ctx=ctx, config=config)
                 response_function = generate_blocking_response(data, headers, status)
             else:

test/test_px_request_validator.py

@@ -71,3 +71,92 @@ def test_handle_verification_failed(self):
         context.score = 100
         response = self.request_handler.handle_verification(context, request)
         self.assertEqual(response.status, '403 Forbidden')
+
+    def test_handle_monitor(self):
+        config = PxConfig({'app_id': 'PXfake_app_id',
+                               'auth_token': '',
+                               'module_mode': px_constants.MODULE_MODE_MONITORING
+                               });
+        request_handler = PxRequestVerifier(config)
+        builder = EnvironBuilder(headers=self.headers, path='/')
+        env = builder.get_environ()
+        request = Request(env)
+        context = PxContext(request, request_handler.config)
+        context.score = 100
+        response = request_handler.handle_verification(context, request)
+        self.assertEqual(response, True)
+
+    def test_bypass_monitor_header_enabled(self):
+        config = PxConfig({'app_id': 'PXfake_app_id',
+                               'auth_token': '',
+                               'module_mode': px_constants.MODULE_MODE_MONITORING,
+                               'bypass_monitor_header': 'x-px-block'
+                               });
+        headers = {'X-FORWARDED-FOR': '127.0.0.1',
+                       'remote-addr': '127.0.0.1',
+                       'x-px-block': '1',
+                       'content_length': '100'}
+        request_handler = PxRequestVerifier(config)
+        builder = EnvironBuilder(headers=headers, path='/')
+        env = builder.get_environ()
+        request = Request(env)
+        context = PxContext(request, request_handler.config)
+        context.score = 100
+        response = request_handler.handle_verification(context, request)
+        self.assertEqual(response.status, '403 Forbidden')
+
+    def test_bypass_monitor_header_disabled(self):
+        config = PxConfig({'app_id': 'PXfake_app_id',
+                               'auth_token': '',
+                               'module_mode': px_constants.MODULE_MODE_MONITORING,
+                               'bypass_monitor_header': 'x-px-block'
+                               });
+        headers = {'X-FORWARDED-FOR': '127.0.0.1',
+                       'remote-addr': '127.0.0.1',
+                       'x-px-block': '0',
+                       'content_length': '100'}
+        request_handler = PxRequestVerifier(config)
+        builder = EnvironBuilder(headers=headers, path='/')
+        env = builder.get_environ()
+        request = Request(env)
+        context = PxContext(request, request_handler.config)
+        context.score = 100
+        response = request_handler.handle_verification(context, request)
+        self.assertEqual(response, True)
+
+    def test_bypass_monitor_header_configured_but_missing(self):
+        config = PxConfig({'app_id': 'PXfake_app_id',
+                               'auth_token': '',
+                               'module_mode': px_constants.MODULE_MODE_MONITORING,
+                               'bypass_monitor_header': 'x-px-block'
+                               });
+        headers = {'X-FORWARDED-FOR': '127.0.0.1',
+                       'remote-addr': '127.0.0.1',
+                       'content_length': '100'}
+        request_handler = PxRequestVerifier(config)
+        builder = EnvironBuilder(headers=headers, path='/')
+        env = builder.get_environ()
+        request = Request(env)
+        context = PxContext(request, request_handler.config)
+        context.score = 100
+        response = request_handler.handle_verification(context, request)
+        self.assertEqual(response, True)
+
+    def test_bypass_monitor_header_on_valid_request(self):
+        config = PxConfig({'app_id': 'PXfake_app_id',
+                               'auth_token': '',
+                               'module_mode': px_constants.MODULE_MODE_MONITORING,
+                               'bypass_monitor_header': 'x-px-block'
+                               });
+        headers = {'X-FORWARDED-FOR': '127.0.0.1',
+                       'remote-addr': '127.0.0.1',
+                       'x-px-block': '1',
+                       'content_length': '100'}
+        request_handler = PxRequestVerifier(config)
+        builder = EnvironBuilder(headers=headers, path='/')
+        env = builder.get_environ()
+        request = Request(env)
+        context = PxContext(request, request_handler.config)
+        context.score = 0
+        response = request_handler.handle_verification(context, request)
+        self.assertEqual(response, True)