Increase PEP8 compliance; Add input validation

Author: IngoKl

projects/cli_proxy_tester/.gitignore

@@ -0,0 +1,141 @@
+# Project
+proxies.csv
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/

projects/cli_proxy_tester/README.md

@@ -13,7 +13,8 @@ This script tests proxies by querying (GET request) a testing website that retur
 
 `python cli.py single http://1.1.1.1`
 
-This will test the HTTP proxy 1.1.1.1 against the default testing website [iptest.ingokleiber.de](http://iptest.ingokleiber.de). You can run your own testing service using the PHP script in `/ipinfo`.
+This will test the HTTP proxy 1.1.1.1 against the default testing website [iptest.ingokleiber.de](http://iptest.ingokleiber.de).
+You can run your own testing service using the PHP script in `/ipinfo`. This service should be offered both via HTTP and HTTPs.
 
 `python cli.py single http://1.1.1.1 --iptest iptest.yourdomain.com`
 

projects/cli_proxy_tester/cli.py

@@ -1,28 +1,51 @@
-import logging
+import re
+
 import click
 
-from proxytest import test_single_proxy, test_file, add_from_text_file
+from proxytest import add_from_text_file
+from proxytest import test_csv_file
+from proxytest import test_single_proxy
+
+
+def validate_proxy(ctx, param, value):
+    '''Validate proxy input. The RegEx crudely matches both IPv4 and URLs.'''
+    validator = re.compile(r'(https|http|socks4|socks5):\/\/'
+                           r'((?:[0-9]{1,3}\.){3}[0-9]{1,3}(:[0-9]{2,5})?'
+                           r'|([\da-z\.-]+)\.([a-z\.]{2,6})([\/\w \.-]*)*\/?)')
+
+    if not validator.match(value):
+        raise click.BadParameter('Please provide a proxy in the format'
+                                 'type://address (e.g., https://42.42.42.42)')
+    else:
+        return value
+
 
 @click.group()
 def cli():
     pass
 
+
 @cli.command()
-@click.argument('proxy')
-@click.option('--iptest', default='http://iptest.ingokleiber.de', help='iptest address')
+@click.argument('proxy', callback=validate_proxy)
+@click.option('--iptest', default='iptest.ingokleiber.de',
+              help='iptest address')
 @click.option('--csv', default='proxies.csv', help='CSV path')
 def single(proxy, iptest, csv):
     test_single_proxy(proxy, iptest, csv)
 
+
 @cli.command()
 @click.argument('csv')
-@click.option('--iptest', default='http://iptest.ingokleiber.de', help='iptest address')
+@click.option('--iptest', default='iptest.ingokleiber.de',
+              help='iptest address')
 def csv_file(iptest, csv):
     test_csv_file(iptest, csv)
 
+
 @cli.command()
 @click.argument('txt')
-@click.option('--iptest', default='http://iptest.ingokleiber.de', help='iptest address')
+@click.option('--iptest', default='iptest.ingokleiber.de',
+              help='iptest address')
 @click.option('--csv', default='proxies.csv', help='CSV path')
 def add_from_txt_file(iptest, txt, csv):
     add_from_text_file(iptest, txt, csv)

projects/cli_proxy_tester/proxytest.py

@@ -1,9 +1,10 @@
-from pathlib import Path
 import logging
+from json.decoder import JSONDecodeError
+from pathlib import Path
+
+import pandas as pd
 import requests
 from requests.exceptions import ProxyError
-import pandas as pd
-from json.decoder import JSONDecodeError
 
 logging.basicConfig(level=logging.INFO)
 
@@ -12,55 +13,66 @@ def add_proxies_to_file(csv_path: str, proxies: list):
     '''This function will add one or multiple proxies to the CSV file.'''
 
     if not csv_path.exists():
-        proxies_file: pd.DataFrame = pd.DataFrame(columns=['proxy_type', 'proxy_address', 'proxy_status'])
+        pr_file: pd.DataFrame = pd.DataFrame(
+            columns=['proxy_type', 'proxy_address', 'proxy_status'])
         logging.info('New CSV file will be created')
     else:
-        proxies_file: pd.DataFrame = pd.read_csv(csv_path)
+        pr_file: pd.DataFrame = pd.read_csv(csv_path)
         logging.info('Existing CSV file has been loaded')
 
     for proxy in proxies:
-        if len(proxies_file) == 0:
+        if len(pr_file) == 0:
             # First proxy in the file
-            proxies_file = proxies_file.append(proxy, ignore_index=True)
+            pr_file = pr_file.append(proxy, ignore_index=True)
         else:
-            if len(proxies_file.loc[ (proxies_file['proxy_type'] == proxy['proxy_type']) & (proxies_file['proxy_address'] == proxy['proxy_address'])]) > 0:
+            if len(pr_file.loc[(pr_file['proxy_type'] == proxy['proxy_type']) &
+                               (pr_file['proxy_address'] == proxy['proxy_address'])]) > 0:
                 # Proxy is already in the file
-                proxies_file.loc[ (proxies_file['proxy_type'] == proxy['proxy_type']) & (proxies_file['proxy_address'] == proxy['proxy_address']) , ['proxy_status']] = proxy['proxy_status']
+                pr_file.loc[(pr_file['proxy_type'] == proxy['proxy_type']) &
+                            (pr_file['proxy_address'] == proxy['proxy_address']),
+                            ['proxy_status']] = proxy['proxy_status']
             else:
                 # Proxy is not yet in the file
-                proxies_file = proxies_file.append(proxy, ignore_index=True)
+                pr_file = pr_file.append(proxy, ignore_index=True)
 
-        
-    proxies_file = proxies_file.drop_duplicates()
-    proxies_file.to_csv(csv_path, index=False)
+    pr_file = pr_file.drop_duplicates()
+    pr_file.to_csv(csv_path, index=False)
     logging.info('CSV file has been written')
 
 
 def test_proxy(proxy_type: str, proxy_address: str, iptest: str):
-    '''This function takes a proxy (type, address) and tests it against a given iptest adress.'''
+    '''This function takes a proxy (type, address)
+    and tests it against a given iptest adress.'''
 
     logging.info(f'Testing proxy: {proxy_address}')
 
     try:
         proxies = {proxy_type: proxy_address}
         proxy_status: str = ''
-        r = requests.get('http://iptest.ingokleiber.de', proxies=proxies)
+
+        if proxy_type == 'https':
+            r = requests.get(f'https://{iptest}', proxies=proxies)
+        else:
+            r = requests.get(f'http://{iptest}', proxies=proxies)
 
         try:
             json_response: dict = r.json()
-            
-            if json_response["ip"] == proxy_address:
+
+            if json_response["ip"] in proxy_address:
                 proxy_status = 'Proxy functional'
             else:
-                logging.warning(f'Proxy "{proxy_address}" returned {json_response}')
+                logging.warning(f'Proxy "{proxy_address}"'
+                                f'returned {json_response}')
                 proxy_status = 'Proxy not functional'
         except JSONDecodeError:
             proxy_status = 'Invalid response'
     except ProxyError:
         proxy_status = 'Proxy error'
 
     logging.info(f'Proxy {proxy_address}: {proxy_status}')
-    return {'proxy_type': proxy_type, 'proxy_address': proxy_address, 'proxy_status': proxy_status}
+    return {'proxy_type': proxy_type,
+            'proxy_address': proxy_address,
+            'proxy_status': proxy_status}
 
 
 def test_single_proxy(proxy: str, iptest: str, csv_path: str):
@@ -71,33 +83,37 @@ def test_single_proxy(proxy: str, iptest: str, csv_path: str):
     add_proxies_to_file(Path(csv_path), [result])
 
 
-def test_file(iptest: str, csv_path: str):
+def test_csv_file(iptest: str, csv_path: str):
     '''This function (re)tests every proxy in a given CSV file.'''
 
     csv_path: Path = Path(csv_path)
 
     if csv_path.exists():
-        proxies_file: pd.DataFrame = pd.read_csv(csv_path)
+        pr_file: pd.DataFrame = pd.read_csv(csv_path)
     else:
         raise FileNotFoundError
 
     proxies: list = []
 
-    for index, proxy in proxies_file.iterrows():
-        proxies.append(test_proxy(proxy['proxy_type'], proxy['proxy_address'], iptest))
+    for index, proxy in pr_file.iterrows():
+        proxies.append(test_proxy(proxy['proxy_type'],
+                                  proxy['proxy_address'],
+                                  iptest))
 
     add_proxies_to_file(csv_path, proxies)
 
 
 def add_from_text_file(iptest: str, text_path: str, csv_path: str):
-    ''' This function adds a list of proxies from a text file (line by line).'''
+    ''' This function adds a list of proxies
+    from a text file (line by line).'''
     text_path: Path = Path(text_path)
 
     if text_path.exists():
         proxies: list = text_path.read_text().splitlines()
 
         for proxy in proxies:
-            # We will treat each proxy as a single proxy and leverage the existing function
+            '''We will treat each proxy as a single proxy
+            and leverage the existing function'''
             test_single_proxy(proxy, iptest, csv_path)
     else:
         raise FileNotFoundError