Check for HSTS on all redirects on same endpoint

pshtt/__init__.py

@@ -1 +1 @@
-__version__ = '0.6.6'
+__version__ = '0.6.7'

pshtt/models.py

@@ -34,6 +34,8 @@ def __init__(self, protocol, host, base_domain):
 
         # all HTTP/HTTPS endpoints have these
         self.headers = {}  # will be replaced with a requests.structures.CaseInsensitiveDict
+        self.ultimate_req = None  # used to track redirects
+        self.redirect_chain = []
         self.status = None
         self.live = None
         self.ip = None
@@ -72,6 +74,7 @@ def __init__(self, protocol, host, base_domain):
         self.https_missing_intermediate_cert = None
         self.hsts = None
         self.hsts_header = None
+        self.hsts_url = None
         self.hsts_max_age = None
         self.hsts_all_subdomains = None
         self.hsts_preload = None
@@ -105,6 +108,7 @@ def to_object(self):
             'redirect_eventually_to_http': self.redirect_eventually_to_http,
             'redirect_eventually_to_external': self.redirect_eventually_to_external,
             'redirect_eventually_to_subdomain': self.redirect_eventually_to_subdomain,
+            'redirect_chain': self.redirect_chain,
             'server_header': self.server_header,
             'server_version': self.server_version,
             'notes': self.notes,
@@ -125,6 +129,7 @@ def to_object(self):
             obj['https_missing_intermediate_cert'] = self.https_missing_intermediate_cert
             obj['hsts'] = self.hsts
             obj['hsts_header'] = self.hsts_header
+            obj['hsts_url'] = self.hsts_url
             obj['hsts_max_age'] = self.hsts_max_age
             obj['hsts_all_subdomains'] = self.hsts_all_subdomains
             obj['hsts_preload'] = self.hsts_preload

pshtt/pshtt.py

@@ -490,6 +490,10 @@ def basic_check(endpoint):
                     (subdomain_original != subdomain_eventual)
                 )
 
+                # Store the redirect responses to check for HSTS later
+                endpoint.ultimate_req = ultimate_req
+                check_redirect_chain(endpoint, ultimate_req)
+
             # If we were able to make the first redirect, but not the ultimate redirect,
             # and if the immediate redirect is external, then it's accurate enough to
             # say that the eventual redirect is the immediate redirect, since you're capturing
@@ -508,6 +512,40 @@ def basic_check(endpoint):
             utils.debug("{}: {}".format(endpoint.url, err))
 
 
+def check_redirect_chain(endpoint, ultimate_req):
+    try:
+        downgrade = False
+        https = False
+        if "https://" in endpoint.url:
+            https = True
+        redirects = []
+        redirect_chain = []
+        if endpoint.ultimate_req:
+            if endpoint.ultimate_req.history:
+                redirects.extend(endpoint.ultimate_req.history)
+            redirects.append(endpoint.ultimate_req)
+        for redirect_entry in redirects:
+            entry_downgrade = ""
+            entry_https = "HTTP"
+            entry_hsts = ""
+            if "https://" in redirect_entry.url:
+                https = True
+                entry_https = "HTTPS"
+                if redirect_entry.headers.get("Strict-Transport-Security"):
+                    entry_hsts = "+HSTS"
+            if https and "http://" in redirect_entry.url:
+                downgrade = True
+                entry_downgrade = "-Downgrade"
+                logging.warning("{}: Downgrade in redirect to {}.".format(endpoint.url, redirect_entry.url))
+            redirect_chain.append("{} ({}{}{})".format(redirect_entry.url, entry_https, entry_hsts, entry_downgrade))
+        if downgrade:
+            logging.warning("{}: Downgrade found in redirect chain {}.".format(endpoint.url, redirect_chain))
+        endpoint.redirect_chain = redirect_chain
+    except Exception as err:
+        logging.warning("{}: Unexpected exception when checking for downgrades in redirects.".format(endpoint.url))
+        utils.debug("{}: {}".format(endpoint.url, err))
+
+
 def hsts_check(endpoint):
     """
     Given an endpoint and its detected headers, extract and parse
@@ -521,9 +559,24 @@ def hsts_check(endpoint):
             return
 
         header = endpoint.headers.get("Strict-Transport-Security")
+        endpoint.hsts_url = endpoint.url
+
+        # If there is no HSTS header, check the eventual response from redirects
+        if header is None and endpoint.ultimate_req:
+            redirects = []
+            if endpoint.ultimate_req.history:
+                redirects.extend(endpoint.ultimate_req.history)
+            redirects.append(endpoint.ultimate_req)
+            for redirect_entry in redirects:
+                if header is None and endpoint.url in redirect_entry.url:
+                    header = redirect_entry.headers.get("Strict-Transport-Security")
+                    if header:
+                        endpoint.hsts_url = redirect_entry.url
+                        logging.warning("{}: Found HSTS in redirected response from {}.".format(endpoint.url, redirect_entry.url))
 
         if header is None:
             endpoint.hsts = False
+            endpoint.hsts_url = None
             return
 
         endpoint.hsts = True