automatically detect and bypass cloudflare challenge pages

TODO: cache and re-apply cfclearance cookies

gallery_dl/cloudflare.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 
-# Copyright 2015-2018 Mike Fährmann
+# Copyright 2015-2019 Mike Fährmann
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License version 2 as
@@ -13,46 +13,52 @@ import time
 import operator
 import urllib.parse
 from . import text
-from .cache import cache
 
 
-def request_func(self, *args, **kwargs):
-    cookies = _cookiecache(self.root)
-    if cookies:
-        self.session.cookies.update(cookies)
-    response = self.session.get(*args, **kwargs)
-    if response.status_code == 503:
-        _cookiecache.invalidate(self.root)
-        self.log.info("Solving Cloudflare challenge")
-        response = solve_challenge(self.session, response)
-        _cookiecache(self.root, self.session.cookies)
-    return response
+def is_challenge(response):
+    return (response.status_code == 503 and
+            response.headers.get("Server", "").startswith("cloudflare") and
+            b"jschl-answer" in response.content)
 
 
-def solve_challenge(session, response):
+def solve_challenge(session, response, kwargs):
+    """Solve Cloudflare challenge and get cfclearance cookie"""
+    parsed = urllib.parse.urlsplit(response.url)
+    root = parsed.scheme + "://" + parsed.netloc
+
+    cf_kwargs = kwargs.copy()
+    headers = cf_kwargs["headers"] = (
+        kwargs["headers"].copy() if "headers" in kwargs else {})
+    params = cf_kwargs["params"] = (
+        kwargs["params"].copy() if "params" in kwargs else {})
 
-    session.headers["Referer"] = response.url
     page = response.text
-    params = text.extract_all(page, (
-        ('jschl_vc', 'name="jschl_vc" value="', '"'),
-        ('pass'    , 'name="pass" value="', '"'),
-    ))[0]
-    params["jschl_answer"] = solve_jschl(response.url, page)
+    params["pass"] = text.extract(page, 'name="pass" value="', '"')[0]
+    params["jschl_vc"] = text.extract(page, 'name="jschl_vc" value="', '"')[0]
+    params["jschl_answer"] = solve_js_challenge(page, parsed.netloc)
+    headers["Referer"] = response.url
 
     time.sleep(4)
-    url = text.urljoin(response.url, "/cdn-cgi/l/chk_jschl")
-    return session.get(url, params=params)
+
+    url = root + "/cdn-cgi/l/chk_jschl"
+    cf_kwargs["allow_redirects"] = False
+    cf_response = session.request(response.request.method, url, **cf_kwargs)
+
+    location = cf_response.headers["Location"]
+    if location[0] == "/":
+        location = root + location
+    return location
 
 
-def solve_jschl(url, page):
-    """Solve challenge to get 'jschl_answer' value"""
+def solve_js_challenge(page, netloc):
+    """Evaluate JS challenge in 'page' to get 'jschl_answer' value"""
 
     # build variable name
     # e.g. '...f, wqnVscP={"DERKbJk":+(...' --> wqnVscP.DERKbJk
     data, pos = text.extract_all(page, (
         ('var' , ',f, ', '='),
-        ('key' , '"', '"'),
-        ('expr', ':', '}'),
+        ('key' , '"'   , '"'),
+        ('expr', ':'   , '}'),
     ))
     variable = "{}.{}".format(data["var"], data["key"])
     vlength = len(variable)
@@ -67,19 +73,19 @@ def solve_jschl(url, page):
     for expr in expressions.split(";")[1:]:
 
         if expr.startswith(variable):
-            # select arithmetc function based on operator (+, -, *)
-            func = operator_functions[expr[vlength]]
+            # select arithmetc function based on operator (+/-/*)
+            func = OPERATORS[expr[vlength]]
             # evaluate the rest of the expression
             value = evaluate_expression(expr[vlength+2:])
-            # combine the expression value with our current solution
+            # combine expression value with our current solution
             solution = func(solution, value)
 
         elif expr.startswith("a.value"):
-            # add length of the hostname, i.e. add 11 for 'example.org'
-            solution += len(urllib.parse.urlsplit(url).netloc)
+            # add length of hostname
+            solution += len(netloc)
 
             if ".toFixed(" in expr:
-                # trim the solution to 10 decimal places
+                # trim solution to 10 decimal places
                 # and strip trailing zeros
                 solution = "{:.10f}".format(solution).rstrip("0")
 
@@ -87,7 +93,7 @@ def solve_jschl(url, page):
 
 
 def evaluate_expression(expr, split_re=re.compile(r"\(+([^)]*)\)")):
-    """Evaluate a Javascript expression for the challenge"""
+    """Evaluate a single Javascript expression for the challenge"""
 
     if "/" in expr:
         # split the expression in numerator and denominator subexpressions,
@@ -102,26 +108,21 @@ def evaluate_expression(expr, split_re=re.compile(r"\(+([^)]*)\)")):
     result = ""
     for subexpr in split_re.findall(expr):
         result += str(sum(
-            expression_values[part]
+            VALUES[part]
             for part in subexpr.split("[]")
         ))
     return int(result)
 
 
-operator_functions = {
+OPERATORS = {
     "+": operator.add,
     "-": operator.sub,
     "*": operator.mul,
 }
 
-expression_values = {
+VALUES = {
     "": 0,
     "+": 0,
     "!+": 1,
     "+!!": 1,
 }
-
-
-@cache(maxage=365*24*60*60, keyarg=0)
-def _cookiecache(key, item=None):
-    return item