Merge pull request #6 from PitterPatterPython/dev

Initial build and prototype pattern refinement

ppp_connectors/__init__.py

@@ -0,0 +1,5 @@
+__all__ = [
+    'check_required_env_vars',
+    'make_request',
+    'spycloud_sip_cookie_domains'
+]

ppp_connectors/broker.py

@@ -0,0 +1,73 @@
+import sys
+from typing import Callable, Dict, Any, List
+from dotenv import dotenv_values, find_dotenv
+import niquests
+from .helpers import check_required_env_vars
+
+
+env_config: Dict = dotenv_values(find_dotenv())
+
+if not env_config:
+    print('[!] Error: The .env file doesn\'t exist or is empty. Did you copy the'
+          '.env.sample file to .env and set your values?', file=sys.stderr)
+    sys.exit(1)
+
+
+def make_request(
+    method: str,
+    url: str,
+    headers: Dict[str, str] = None,
+    params: Dict[str, Any] = None,
+    data: Dict[str, Any] = None,
+    json: Dict[str, Any] = None
+) -> niquests.Response:
+    """Perform an HTTP request on behalf of a calling function
+
+    Args:
+        method (str): the HTTP method to use
+        url (str): the API URL to call
+        headers (Dict[str, str], optional): the HTTP headers to use in the request. Defaults to None.
+        params (Dict[str, Any], optional): the query parameters to use in the request. Defaults to None.
+        data (Dict[str, Any], optional): the data to use in the request. Defaults to None.
+        json (Dict[str, Any], optional): the json data to use in the request. Defaults to None.
+
+    Raises:
+        ValueError: this will raise if an invalid HTTP method is passed
+
+    Returns:
+        niquests.Response: the HTTP response from the request
+    """
+
+    # Define required environment variables
+    required_vars: List[str] = [
+        'HTTP_PROXY',
+        'HTTPS_PROXY',
+        'VERIFY_SSL'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    proxies: Dict = {
+        'http_proxy': env_config['HTTP_PROXY'],
+        'https_proxy': env_config['HTTPS_PROXY']
+    }
+
+    verify: str = False if env_config['VERIFY_SSL'].lower() == "false" else True
+    if verify is False:
+        import urllib3
+        urllib3.disable_warnings()
+
+    method_map: Dict[str, Callable] = {
+        'GET': niquests.get,
+        'POST': niquests.post,
+        'PUT': niquests.put,
+        'DELETE': niquests.delete,
+        'PATCH': niquests.patch
+    }
+
+    request_func = method_map.get(method.upper())
+    if not request_func:
+        raise ValueError(f'Unsupported HTTP method: {method}')
+
+    return request_func(url, headers=headers, params=params, data=data, json=json, proxies=proxies, verify=verify)

ppp_connectors/helpers.py

@@ -0,0 +1,16 @@
+import sys
+from typing import Dict, Set, List
+
+def check_required_env_vars(config: Dict[str, str], required_vars: List[str]) -> None:
+    """Ensure that the env variables required for a function are present in the .env file.
+
+    Args:
+        config (Dict[str, str]): the env_config variable that contains values from the .env file
+        required_vars (List[str]): the env variables required for a function to successfully function
+    """
+
+    missing_vars: Set[str] = set(required_vars) - set(config.keys())
+    if missing_vars:
+        print(f"[!] Error: missing required environment variables: {', '.join(missing_vars)}. "
+              "Please ensure these are present in your .env file.", file=sys.stderr)
+        sys.exit(1)

ppp_connectors/spycloud.py

@@ -0,0 +1,44 @@
+import sys
+from typing import Dict, Any, List
+from dotenv import dotenv_values, find_dotenv
+from niquests import Response
+from .broker import make_request
+from .helpers import check_required_env_vars
+
+# Load environment variables
+env_config: Dict = dotenv_values(find_dotenv())
+
+if not env_config:
+    print('[!] Error: The .env file doesn\'t exist or is empty. Did you copy the'
+          '.env.sample file to .env and set your values?', file=sys.stderr)
+    sys.exit(1)
+
+def spycloud_sip_cookie_domains(cookie_domains: str, **kwargs: Dict[str, Any]) -> Response:
+    """Return botnet sourced cookie data for your domain and its subdomains
+
+    Args:
+        cookie_domains (str): the domain and subdomains to search against
+
+    Returns:
+        Response: niquests.Response json response from the request
+    """
+
+    # Define required environment variables
+    required_vars: List[str] = [
+        'SPYCLOUD_API_SIP_KEY'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    method: str = 'get'
+    url: str = f'https://api.spycloud.io/sip-v1/breach/data/cookie-domains/{cookie_domains}'
+    headers: Dict = {
+        'accept': 'application/json',
+        'x-api-key': env_config['SPYCLOUD_API_SIP_KEY']
+    }
+    params: Dict = dict(kwargs)
+
+    result: Response = make_request(method=method, url=url, headers=headers, params=params)
+
+    return result

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "ppp-connectors"
-packages = [{ include = "connectors" }]
-version = "0.1.0"
+packages = [{ include = "ppp_connectors" }]
+version = "0.1.1"
 description = "A simple, lightweight set of connectors and functions to various APIs, controlled by a central broker."
 authors = [
     "Rob D'Aveta <[email protected]>",