Port SSL verification rules from tasks in `requests.py`

teklia_toolbox/requests.py

 import logging
+from pathlib import Path
+from urllib.parse import urlparse
 
 import requests
+import urllib3
 from apistar.exceptions import ErrorResponse
 from tenacity import (
     before_sleep_log,
@@ -16,15 +19,34 @@ from arkindex import ArkindexClient, options_from_env
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_CLIENT = ArkindexClient(**options_from_env())
 
-# Time to wait before retrying the IIIF image information fetching
-HTTP_GET_RETRY_BACKOFF = 10
+def should_verify_cert(url: str) -> bool:
+    """
+    Skip SSL certification validation when hitting a development instance
+    """
+    if not url:
+        return True
 
-DOWNLOAD_CHUNK_SIZE = 8192
+    host = urlparse(url).netloc
+    return not host.endswith("ark.localhost")
+
+
+def _get_arkindex_client() -> ArkindexClient:
+    options = options_from_env()
+
+    # Skip SSL verification in Arkindex API client for local development hosts
+    verify = should_verify_cert(options.get("base_url"))
+    if not verify:
+        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+        logger.warn("SSL certificate verification is disabled for Arkindex API calls")
+
+    return ArkindexClient(verify=verify, **options)
 
 
-def _is_500_error(exc):
+DEFAULT_CLIENT = _get_arkindex_client()
+
+
+def _is_500_error(exc: Exception) -> bool:
     """
     Check if an Arkindex API error is a 50x
     This is used to retry most API calls implemented here
@@ -55,17 +77,23 @@ def retried_request(*args, **kwargs):
     return DEFAULT_CLIENT.request(*args, **kwargs)
 
 
+# Time to wait before retrying the IIIF image information fetching
+HTTP_GET_RETRY_BACKOFF = 10
+
+DOWNLOAD_CHUNK_SIZE = 8192
+
+
 @retry(
     reraise=True,
     retry=retry_if_exception_type(requests.RequestException),
     stop=stop_after_attempt(3),
     wait=wait_fixed(HTTP_GET_RETRY_BACKOFF),
 )
-def download_file(url, path):
+def download_file(url: str, path: Path) -> None:
     """
     Download a URL into a local path, retrying if necessary
     """
-    with requests.get(url, stream=True) as r:
+    with requests.get(url, stream=True, verify=should_verify_cert(url)) as r:
         r.raise_for_status()
         with path.open("wb") as f:
             for chunk in r.iter_content(chunk_size=DOWNLOAD_CHUNK_SIZE):