Captcha V2

perimeterx/middleware.py

@@ -3,9 +3,8 @@
 import px_activities_client
 import px_cookie_validator
 import px_httpc
-import px_captcha
+import px_blocker
 import px_api
-import px_template
 import Cookie
 
 
@@ -27,7 +26,8 @@ def __init__(self, app, config=None):
             'api_timeout': 1,
             'custom_logo': None,
             'css_ref': None,
-            'js_ref': None
+            'js_ref': None,
+            'is_mobile': False
         }
 
         self.config = dict(self.config.items() + config.items())
@@ -45,7 +45,7 @@ def __init__(self, app, config=None):
         if not config['cookie_key']:
             logger.error('PX Cookie Key is missing')
             raise ValueError('PX Cookie Key is missing')
-
+        self.PXBlocker = px_blocker.PXBlocker()
         px_httpc.init(self.config)
 
     def __call__(self, environ, start_response):
@@ -59,7 +59,7 @@ def custom_start_response(status, headers, exc_info=None):
                 self.config['logger'].debug('Cleared Cookie');
             return start_response(status, headers, exc_info)
 
-        return self._verify(environ, custom_start_response)
+        return self._verify(environ, start_response)
 
     def _verify(self, environ, start_response):
         logger = self.config['logger']
@@ -71,11 +71,6 @@ def _verify(self, environ, start_response):
                 return self.app(environ, start_response)
 
             cookies = Cookie.SimpleCookie(environ.get('HTTP_COOKIE'))
-            if self.config.get('captcha_enabled') and cookies.get('_pxCaptcha') and cookies.get('_pxCaptcha').value:
-                pxCaptcha = cookies.get('_pxCaptcha').value
-                if px_captcha.verify(ctx, self.config, pxCaptcha):
-                    logger.debug('User passed captcha verification. user ip: ' + ctx.get('socket_ip'))
-                    return self.app(environ, start_response)
 
             # PX Cookie verification
             if not px_cookie_validator.verify(ctx, self.config) and self.config.get('server_calls_enabled', True):
@@ -98,17 +93,7 @@ def handle_verification(self, ctx, config, environ, start_response):
             px_activities_client.send_block_activity(ctx, config)
             return config['custom_block_handler'](ctx, start_response)
         elif config.get('module_mode', 'active_monitoring') == 'active_blocking':
-            vid = ctx.get('vid', '')
-            uuid = ctx.get('uuid', '')
-            template = 'block'
-            if config.get('captcha_enabled', False):
-                template = 'captcha'
-
-            body = px_template.get_template(template, self.config, uuid, vid)
-
-            px_activities_client.send_block_activity(ctx, config)
-            start_response("403 Forbidden", [('Content-Type', 'text/html')])
-            return [str(body)]
+            return self.PXBlocker.handle_blocking(ctx=ctx, config=config, start_response=start_response)
         else:
             return self.pass_traffic(environ, start_response, ctx)
 

perimeterx/px_api.py

@@ -9,7 +9,7 @@ def send_risk_request(ctx, config):
 
 def verify(ctx, config):
     logger = config['logger']
-    logger.debug("PxAPI[verify]")
+    logger.debug("PXVerify")
     try:
         response = send_risk_request(ctx, config)
         if response:
@@ -18,8 +18,17 @@ def verify(ctx, config):
             ctx['uuid'] = response['uuid']
             ctx['block_action'] = response['action']
             if score >= config['blocking_score']:
-                logger.debug("PxAPI[verify] block score threshold reached")
+                logger.debug("PXVerify block score threshold reached, will initiate blocking")
                 ctx['block_reason'] = 's2s_high_score'
+            elif response['action'] is 'j' and response.get('action_data') is not None and response.get('action_data').get('body') is not None:
+                logger.debug("PXVerify received javascript challenge action")
+                ctx['block_action_data'] = response.get('action_data').get('body')
+                ctx['block_reason'] = 'challenge'
+            elif response['action'] is 'r':
+                logger.debug("PXVerify received javascript ratelimit action")
+                ctx['block_reason'] = 'exceeded_rate_limit'
+            else:
+                ctx['pass_reason'] = 's2s'
 
             logger.debug("PxAPI[verify] S2S completed")
             return True

perimeterx/px_blocker.py

@@ -0,0 +1,81 @@
+import pystache
+import px_template
+import px_constants
+import json
+
+
+class PXBlocker(object):
+    def __init__(self):
+        self.mustache_renderer = pystache.Renderer()
+        self.ratelimit_rendered_page = self.mustache_renderer.render(
+            px_template.get_template(px_constants.RATELIMIT_TEMPLATE), {})
+
+    def handle_blocking(self, ctx, config, start_response):
+        action = ctx.get('block_action')
+        status = '403 Forbidden'
+
+        is_json_response = self.is_json_response(ctx)
+        if is_json_response:
+            content_type = 'application/json'
+        else:
+            content_type = 'text/html'
+        headers = [('Content-Type', content_type)]
+
+        if action is 'j':
+            blocking_props = ctx['block_action_data']
+            blocking_response = blocking_props
+        elif action is 'r':
+            blocking_response = self.ratelimit_rendered_page
+            status = '429 Too Many Requests'
+        else:
+            blocking_props = self.prepare_properties(ctx, config)
+            blocking_response = self.mustache_renderer.render(px_template.get_template(px_constants.BLOCK_TEMPLATE),
+                                                              blocking_props)
+        start_response(status, headers)
+        if is_json_response:
+            blocking_response = json.dumps(blocking_props)
+        return str(blocking_response)
+
+    def prepare_properties(self, ctx, config):
+        app_id = config.get('app_id').lower()
+        vid = ctx.get('vid') if ctx.get('vid') is not None else ''
+        uuid = ctx.get('uuid')
+        custom_logo = config.get('CUSTOM_LOGO') if config.get('CUSTOM_LOGO') is not None else ''
+        is_mobile_num = 1 if ctx.get('is_mobile') else 0
+        captcha_uri = 'captcha.js?a={}&u={}&v={}&m={}'.format(ctx.get('block_action'), uuid, vid, is_mobile_num)
+
+        if config.get('first_party') and not ctx.get('is_mobile'):
+            prefix = app_id[2:]
+            js_client_src = '/{}/{}'.format(prefix, px_constants.CLIENT_FP_PATH)
+            captcha_src = '/{}/{}/{}'.format(prefix, px_constants.CAPTCHA_FP_PATH, captcha_uri)
+            host_url = '/{}/{}'.format(prefix, px_constants.XHR_FP_PATH)
+        else:
+            js_client_src = '//{}/{}/main.min.js'.format(px_constants.CLIENT_HOST, app_id)
+            captcha_src = '//{}/{}/{}'.format(px_constants.CAPTCHA_HOST, app_id, captcha_uri)
+            host_url = px_constants.COLLECTOR_URL.format(app_id.lower())
+
+        return {
+            'refId': uuid,
+            'appId': app_id,
+            'vid': vid,
+            'uuid': uuid,
+            'customLogo': custom_logo,
+            'cssRef': config.get('css_ref'),
+            'jsRef': config.get('js_ref'),
+            'logoVisibility': 'visible' if custom_logo is not None else 'hidden',
+            'hostUrl': host_url,
+            'jsClientSrc': js_client_src,
+            'firstPartyEnabled': config.get('first_party'),
+            'blockScript': captcha_src
+        }
+
+    def is_json_response(self, ctx):
+        headers = ctx.get('headers')
+        if ctx.get('block_action') is not 'r':
+            for item in headers.keys():
+                if (item.lower() == 'accept' or item.lower() == 'content-type'):
+                    item_arr = headers[item].split(',')
+                    for header_item in item_arr:
+                        if header_item.strip() == 'application/json':
+                            return True
+        return False

perimeterx/px_constants.py

@@ -2,4 +2,18 @@
 PREFIX_PX_COOKIE_V3 = '_px3'
 
 TRANS_5C = b"".join(chr(x ^ 0x5C) for x in range(256))
-TRANS_36 = b"".join(chr(x ^ 0x36) for x in range(256))
+TRANS_36 = b"".join(chr(x ^ 0x36) for x in range(256))
+
+BLOCK_TEMPLATE = 'block_template.mustache'
+RATELIMIT_TEMPLATE = 'ratelimit.mustache'
+CAPTCHA_ACTION_CAPTCHA = 'c'
+BLOCK_ACTION_CAPTCHA = 'b'
+BLOCK_ACTION_CHALLENGE = 'j'
+BLOCK_ACTION_RATE = 'r'
+CLIENT_HOST = 'client.perimeterx.net'
+CAPTCHA_HOST = 'captcha.px-cdn.net'
+COLLECTOR_URL = 'https://collector-{}.perimeterx.net'
+CLIENT_FP_PATH = 'init.js'
+CAPTCHA_FP_PATH = 'captcha'
+XHR_FP_PATH = 'xhr'
+

perimeterx/px_template.py

@@ -1,29 +1,16 @@
-import pystache
 import os
 
-def get_template(template, config, uuid, vid):
-    template_content = get_content(template)
-    props = get_props(config, uuid, vid)
-    generatedHtml = pystache.render(template_content, props)
-    return generatedHtml
 
 def get_path():
     return os.path.dirname(os.path.abspath(__file__))
 
 def get_content(template):
-    templatePath = "%s/templates/%s.mustache" % (get_path(),template)
+    templatePath = "%s/templates/%s" % (get_path(), template)
     file = open(templatePath, "r")
     content = file.read()
     return content
 
-def get_props(config, uuid, vid):
-    return {
-        'refId': uuid,
-        'appId': config.get('app_id'),
-        'vid': vid,
-        'uuid': uuid,
-        'customLogo': config.get('custom_logo'),
-        'cssRef': config.get('css_ref'),
-        'jsRef': config.get('js_ref'),
-        'logoVisibility': 'visible' if config['custom_logo'] else 'hidden'
-    }
+
+
+def get_template(template_name):
+    return get_content(template_name)