5.8.3

- Use timeout values for HTTP client timeouts

Author: seanthegeek

CHANGELOG.md

@@ -1,6 +1,11 @@
 Changelog
 =========
 
+5.8.3
+-----
+
+- Use timeout values for HTTP client timeouts
+
 5.8.2
 -----
 

checkdmarc/_constants.py

@@ -18,9 +18,10 @@
 See the License for the specific language governing permissions and
 limitations under the License."""
 
-__version__ = "5.8.2"
+__version__ = "5.8.3"
 
 OS = platform.system()
 OS_RELEASE = platform.release()
 USER_AGENT = f"Mozilla/5.0 (({OS} {OS_RELEASE})) checkdmarc/{__version__}"
 SYNTAX_ERROR_MARKER = "➞"
+DEFAULT_HTTP_TIMEOUT = 2.0

checkdmarc/bimi.py

@@ -35,7 +35,7 @@
 )
 
 import checkdmarc.resources
-from checkdmarc._constants import SYNTAX_ERROR_MARKER, USER_AGENT
+from checkdmarc._constants import SYNTAX_ERROR_MARKER, USER_AGENT, DEFAULT_HTTP_TIMEOUT
 from checkdmarc.utils import WSP_REGEX, HTTPS_REGEX, query_dns, get_base_domain
 
 """Copyright 2019-2023 Sean Whalen
@@ -485,6 +485,7 @@ def parse_bimi_record(
     domain: str = None,
     include_tag_descriptions: bool = False,
     syntax_error_marker: str = SYNTAX_ERROR_MARKER,
+    http_timeout: float = DEFAULT_HTTP_TIMEOUT,
 ) -> OrderedDict:
     """
     Parses a BIMI record
@@ -494,6 +495,7 @@ def parse_bimi_record(
         domain (str): The domain where the BIMI record was located
         include_tag_descriptions (bool): Include descriptions in parsed results
         syntax_error_marker (str): The maker for pointing out syntax errors
+        http_timeout (float): HTTP timeout in seconds
 
     Returns:
         OrderedDict: An ``OrderedDict`` with the following keys:
@@ -572,7 +574,7 @@ def parse_bimi_record(
         if tag == "l" and tag_value != "":
             raw_xml = None
             try:
-                response = session.get(tag_value)
+                response = session.get(tag_value, timeout=http_timeout)
                 response.raise_for_status()
                 raw_xml = response.content
             except Exception as e:
@@ -592,7 +594,7 @@ def parse_bimi_record(
         elif tag == "a" and tag_value != "":
             cert_metadata = None
             try:
-                response = session.get(tag_value)
+                response = session.get(tag_value, timeout=http_timeout)
                 response.raise_for_status()
                 pem_bytes = response.content
                 cert_metadata = get_certificate_metadata(pem_bytes, domain=domain)
@@ -677,6 +679,7 @@ def check_bimi(
             bimi_results["record"],
             include_tag_descriptions=include_tag_descriptions,
             domain=domain,
+            http_timeout=timeout,
         )
         bimi_results["tags"] = parsed_bimi["tags"]
         if "image" in parsed_bimi.keys():

checkdmarc/mta_sts.py

@@ -17,7 +17,7 @@
 )
 
 from checkdmarc.utils import query_dns, WSP_REGEX
-from checkdmarc._constants import SYNTAX_ERROR_MARKER, USER_AGENT
+from checkdmarc._constants import SYNTAX_ERROR_MARKER, USER_AGENT, DEFAULT_HTTP_TIMEOUT
 
 """Copyright 2019-2023 Sean Whalen
 
@@ -308,12 +308,15 @@ def parse_mta_sts_record(
     return OrderedDict(tags=tags, warnings=warnings)
 
 
-def download_mta_sts_policy(domain: str) -> OrderedDict:
+def download_mta_sts_policy(
+    domain: str, http_timeout: float = DEFAULT_HTTP_TIMEOUT
+) -> OrderedDict:
     """
     Downloads a domains MTA-HTS policy
 
     Args:
         domain (str): A domain name
+        http_timeout (float): HTTP timeout in seconds
 
     Returns:
         OrderedDict: An ``OrderedDict`` with the following keys:
@@ -331,7 +334,7 @@ def download_mta_sts_policy(domain: str) -> OrderedDict:
     url = f"https://mta-sts.{domain}/.well-known/mta-sts.txt"
     logging.debug(f"Attempting to download HTA-MTS policy from {url}")
     try:
-        response = session.get(url)
+        response = session.get(url, timeout=http_timeout)
         response.raise_for_status()
         if "Content-Type" in response.headers:
             content_type = response.headers["Content-Type"].split(";")[0]
@@ -467,7 +470,7 @@ def check_mta_sts(
         warnings = mta_sts_record["warnings"]
         mta_sts_record = parse_mta_sts_record(mta_sts_record["record"])
         mta_sts_results["id"] = mta_sts_record["tags"]["id"]["value"]
-        policy = download_mta_sts_policy(domain)
+        policy = download_mta_sts_policy(domain, timeout=timeout)
         warnings += policy["warnings"]
         policy = parse_mta_sts_policy(policy["policy"])
         warnings += policy["warnings"]