Merge pull request #19 from PitterPatterPython/add-twilio

Add twilio functions for lookups and usage

.env.sample

@@ -5,13 +5,6 @@ PPP_HTTP_PROXY=
 PPP_HTTPS_PROXY=
 VERIFY_SSL=
 
-############
-# SPYCLOUD #
-############
-SPYCLOUD_API_ATO_KEY=
-SPYCLOUD_API_INV_KEY=
-SPYCLOUD_API_SIP_KEY=
-
 ##############
 # FLASHPOINT #
 ##############
@@ -22,6 +15,20 @@ FLASHPOINT_API_KEY=
 ##################
 IPQS_API_KEY=
 
+############
+# SPYCLOUD #
+############
+SPYCLOUD_API_ATO_KEY=
+SPYCLOUD_API_INV_KEY=
+SPYCLOUD_API_SIP_KEY=
+
+###########
+# TWILIO #
+###########
+TWILIO_ACCOUNT_SID=
+TWILIO_API_SID=
+TWILIO_API_SECRET=
+
 ###########
 # URLSCAN #
 ###########

ppp_connectors/broker.py

@@ -1,6 +1,7 @@
 import sys
 from typing import Callable, Dict, Any, List
 import requests
+from requests.auth import HTTPBasicAuth
 from .helpers import check_required_env_vars, combine_env_configs
 
 
@@ -11,6 +12,7 @@ def make_request(
     method: str,
     url: str,
     headers: Dict[str, str] = None,
+    auth: HTTPBasicAuth = None,
     params: Dict[str, Any] = None,
     data: Dict[str, Any] = None,
     json: Dict[str, Any] = None
@@ -60,4 +62,11 @@ def make_request(
     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)
+    return request_func(url,
+                        headers=headers,
+                        auth=auth,
+                        params=params,
+                        data=data,
+                        json=json,
+                        proxies=proxies,
+                        verify=verify)

ppp_connectors/helpers.py

@@ -1,4 +1,4 @@
-
+from datetime import date, datetime
 from dotenv import dotenv_values, find_dotenv
 import os
 import sys
@@ -37,4 +37,19 @@ def combine_env_configs() -> Dict[str, Any]:
 
     combined_config: Dict[str, Any] = {**env_config, **dict(os.environ)}
 
-    return combined_config
+    return combined_config
+
+def validate_date_string(date_str: str) -> bool:
+    """Validates that a date string is, well, a valid date string
+
+    Args:
+        date_str (str): a string in "YYYY-MM-DD" format
+
+    Returns:
+        bool: True or False as valid or not
+    """
+    try:
+        datetime.strptime(date_str, "%Y-%m-%d")
+        return True
+    except ValueError:
+        return False

ppp_connectors/twilio.py

@@ -0,0 +1,108 @@
+from datetime import date, datetime
+from typing import Dict, Any, List, Set, Union, Optional
+from requests import Response
+from requests.auth import HTTPBasicAuth
+import sys
+from .broker import make_request
+from .helpers import check_required_env_vars, combine_env_configs, validate_date_string
+
+
+env_config: Dict[str, Any] = combine_env_configs()
+
+def twilio_lookup(phone_number: str, data_packages: list=[], **kwargs: Dict[str, Any]) -> Response:
+    """query information on a phone number so that you can make a trusted interaction with your user.
+        With this endpoint, you can format and validate phone numbers with the free Basic Lookup request
+        and add on data packages to get even more in-depth carrier and caller information.
+
+    Args:
+        phone_number (str): The phone number to look up
+        data_packages (list): A Python list of fields to return. Possible values are validation,
+            caller_name, sim_swap, call_forwarding, line_status, line_type_intelligence, identity_match,
+            reassigned_number, sms_pumping_risk, phone_number_quality_score, pre_fill.
+
+    Returns:
+        Response: requests.Response json response from the request
+    """
+
+    # Define required environment variables
+    required_vars: List[str] = [
+        'TWILIO_API_SID',
+        'TWILIO_API_SECRET'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    # Valid set of data packages for Twilio. Compare the ones that the user passed in
+    # to ensure that they've passed valid ones. Exit immediately if they didn't.
+    valid_data_packages: Set = {'validation', 'caller_name', 'sim_swap',
+                           'call_forwarding', 'line_status', 'line_type_intelligence',
+                           'identity_match', 'reassigned_number', 'sms_pumping_risk',
+                           'phone_number_quality_score', 'pre_fill'}
+    data_packages_set: Set = set(data_packages)
+    invalid_packages =  data_packages_set - valid_data_packages
+    if len(invalid_packages) != 0:
+        print(f'[!] Error: "{", ".join(invalid_packages)}" are not valid data packages. Valid '
+              f'packages include {", ".join(valid_data_packages)}', file=sys.stderr)
+        sys.exit(1)
+
+    method: str = 'get'
+    url: str = f'https://lookups.twilio.com/v2/PhoneNumbers/{phone_number}'
+
+    auth = HTTPBasicAuth(env_config['TWILIO_API_SID'], env_config['TWILIO_API_SECRET'])
+
+    params: Dict = {
+        'Fields': ','.join(data_packages),
+        **kwargs
+    }
+
+    result: Response = make_request(method=method, url=url, auth=auth, params=params)
+
+    return result
+
+def twilio_usage_report(start_date: Union[str, date],
+                        end_date: Optional[Union[str, date]]=None) -> Response:
+    """Return a usage report for all activities between the start_date and end_date.
+
+    Args:
+        start_date (Union[str, date]): Only include usage that has occurred on or after this
+            date. Specify the date in GMT and format as YYYY-MM-DD
+        end_date (Optional[Union[str, date]], optional): Only include usage that occurred on
+            or before this date. Specify the date in GMT and format as YYYY-MM-DD. Defaults to None.
+
+    Returns:
+        Response: requests.Response json response from the request
+    """
+
+    # Define required environment variables
+    required_vars: List[str] = [
+        'TWILIO_ACCOUNT_SID',
+        'TWILIO_API_SID',
+        'TWILIO_API_SECRET'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    if end_date is None:
+        end_date: str = datetime.now().strftime("%Y-%m-%d")
+
+    if not validate_date_string(start_date) or not validate_date_string(end_date):
+        print(f'[!] Error: One of your start date {start_date} or end date {end_date} '
+              'does not match the format YYYY-MM-DD')
+        sys.exit()
+
+
+    method: str = 'get'
+    url: str = f'https://api.twilio.com/2010-04-01/Accounts/{env_config["TWILIO_ACCOUNT_SID"]}/Usage/Records.json'
+
+    auth = HTTPBasicAuth(env_config['TWILIO_API_SID'], env_config['TWILIO_API_SECRET'])
+
+    params: Dict = {
+        'StartDate': start_date,
+        'EndDate': end_date
+    }
+
+    result: Response = make_request(method=method, url=url, auth=auth, params=params)
+
+    return result

pyproject.toml

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