Robots/bkmk_rrequests.py

   1 """Robot based on requests
   2
   3 This file is a part of Bookmarks database and Internet robot.
   4
   5 """
   6
   7 __author__ = "Oleg Broytman <phd@phdru.name>"
   8 __copyright__ = "Copyright (C) 2024, 2025 PhiloSoft Design"
   9 __license__ = "GNU GPL"
  10
  11 __all__ = ['robot_requests']
  12
  13
  14 import warnings
  15
  16 from requests.adapters import HTTPAdapter
  17 from requests.packages.urllib3.util.ssl_ import create_urllib3_context
  18 import requests
  19 import urllib3
  20
  21 from Robots.base import robot_base
  22 from Robots.util import get_ftp
  23
  24
  25 class robot_requests(robot_base):
  26     def version_str(self):
  27         return 'python-requests urllib3/%s' % urllib3.__version__
  28
  29     async def get(self, url, req_headers, use_proxy=False):
  30         if url.startswith('ftp://'):
  31             error, welcome, body = get_ftp(url, self.timeout)
  32             if error is not None:
  33                 return error, None, None, None
  34             self.welcome = welcome
  35             return None, None, None, body
  36
  37         if use_proxy:
  38             proxies = {'http': self.proxy, 'https': self.proxy}
  39         else:
  40             proxies = None
  41
  42         s = requests.Session()
  43         s.mount('https://', AllCiphersAdapter())
  44
  45         error = r = None
  46         try:
  47             r = s.get(url, headers=req_headers, timeout=self.timeout,
  48                       allow_redirects=False, proxies=proxies,
  49                       verify=False)
  50         except requests.RequestException as e:
  51             error = str(e)
  52             return error, None, None, None
  53
  54         return None, r.status_code, r.headers, r.content
  55
  56     def get_ftp_welcome(self):
  57         welcome = self.welcome
  58         self.welcome = ''
  59         return welcome
  60
  61
  62 # See https://lukasa.co.uk/2017/02/Configuring_TLS_With_Requests/
  63
  64 class AllCiphersAdapter(HTTPAdapter):
  65     """
  66     A TransportAdapter that re-enables 3DES support in Requests.
  67     """
  68     def init_poolmanager(self, *args, **kwargs):
  69         context = create_urllib3_context(cert_reqs=0,
  70                                          ciphers='ALL:@SECLEVEL=1')
  71         kwargs['ssl_context'] = context
  72         return super(AllCiphersAdapter, self).init_poolmanager(*args, **kwargs)
  73
  74     def proxy_manager_for(self, *args, **kwargs):
  75         context = create_urllib3_context(cert_reqs=0,
  76                                          ciphers='ALL:@SECLEVEL=1')
  77         kwargs['ssl_context'] = context
  78         return super(AllCiphersAdapter, self).proxy_manager_for(
  79             *args, **kwargs)
  80
  81
  82 warnings.filterwarnings('ignore', 'Unverified HTTPS request is being made')