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 PhiloSoft Design"
   9 __license__ = "GNU GPL"
  10
  11 __all__ = ['robot_requests']
  12
  13
  14 from urllib.parse import urlsplit
  15 import ftplib
  16 import socket
  17 import warnings
  18
  19 from requests.adapters import HTTPAdapter
  20 from requests.packages.urllib3.util.ssl_ import create_urllib3_context
  21 import requests
  22 import urllib3
  23
  24 from Robots.base import robot_base
  25
  26
  27 class robot_requests(robot_base):
  28     def version_str(self):
  29         return 'python-requests urllib3/%s' % urllib3.__version__
  30
  31     async def get(self, url, req_headers, use_proxy=False):
  32         if url.startswith('ftp://'):
  33             error, welcome, body = _get_ftp(url, self.timeout)
  34             if error is not None:
  35                 return error, None, None, None, None
  36             self.welcome = welcome
  37             return None, None, None, body
  38
  39         if use_proxy:
  40             proxies = {'http': self.proxy, 'https': self.proxy}
  41         else:
  42             proxies = None
  43
  44         s = requests.Session()
  45         s.mount('https://', AllCiphersAdapter())
  46
  47         error = r = None
  48         try:
  49             r = s.get(url, headers=req_headers, timeout=self.timeout,
  50                       allow_redirects=False, proxies=proxies,
  51                       verify=False)
  52         except requests.RequestException as e:
  53             error = str(e)
  54             return error, None, None, None
  55
  56         return None, r.status_code, r.headers, r.content
  57
  58     def get_ftp_welcome(self):
  59         welcome = self.welcome
  60         self.welcome = ''
  61         return welcome
  62
  63
  64 def _get_ftp(url, timeout):
  65     split_results = urlsplit(url)
  66     path = split_results.path or '/'
  67     user = split_results.username or 'anonymous'
  68     password = split_results.password or 'ftp'
  69     host = split_results.hostname
  70     port = split_results.port or 21
  71
  72     ftp = ftplib.FTP()
  73     try:
  74         ftp.connect(host, port, timeout)
  75         ftp.login(user, password)
  76         if path != '/':
  77             ftp.cwd(path)
  78     except (socket.error, ftplib.Error) as e:
  79         return "Error: %s" % e, None, None
  80
  81     files = []
  82     try:
  83         files = ftp.nlst()
  84     except ftplib.error_perm as e:
  85         if str(e).startswith("550 No files found"):
  86             return None, ftp.welcome, ''
  87         return "Error: %s" % e, None, None
  88     except ftplib.Error as e:
  89         return "Error: %s" % e, None, None
  90
  91     return None, ftp.welcome, '\n'.join(files)
  92
  93
  94 # See https://lukasa.co.uk/2017/02/Configuring_TLS_With_Requests/
  95
  96 class AllCiphersAdapter(HTTPAdapter):
  97     """
  98     A TransportAdapter that re-enables 3DES support in Requests.
  99     """
 100     def init_poolmanager(self, *args, **kwargs):
 101         context = create_urllib3_context(cert_reqs=0,
 102                                          ciphers='ALL:@SECLEVEL=1')
 103         kwargs['ssl_context'] = context
 104         return super(AllCiphersAdapter, self).init_poolmanager(*args, **kwargs)
 105
 106     def proxy_manager_for(self, *args, **kwargs):
 107         context = create_urllib3_context(cert_reqs=0,
 108                                          ciphers='ALL:@SECLEVEL=1')
 109         kwargs['ssl_context'] = context
 110         return super(AllCiphersAdapter, self).proxy_manager_for(
 111             *args, **kwargs)
 112
 113
 114 warnings.filterwarnings('ignore', 'Unverified HTTPS request is being made')