Merge pull request #17 from PitterPatterPython/urlscan-and-ipqualitys…

…core-connectors

Urlscan and ipqualityscore connectors

.env.sample

@@ -15,4 +15,14 @@ SPYCLOUD_API_SIP_KEY=
 ##############
 # FLASHPOINT #
 ##############
-FLASHPOINT_API_KEY=
+FLASHPOINT_API_KEY=
+
+##################
+# IPQUALITYSCORE #
+##################
+IPQS_API_KEY=
+
+###########
+# URLSCAN #
+###########
+URLSCAN_API_KEY=

README.md

@@ -3,4 +3,13 @@ A simple, lightweight set of connectors and functions to various APIs, controlle
 
 ## How to install
 1. Install via pip to your environment: `pip install ppp-connectors`
-2. Load the required environment variables into your environment. You can find these in `env.sample`. This library is intelligent enough to look for both a `.env` file _and_ within your system's environment variables, so you can do either option.
+2. Load the required environment variables into your environment. You can find these in `env.sample`. This library is intelligent enough to look for both a `.env` file _and_ within your system's environment variables, so you can do either option.
+
+## Passing additional parameters to a function
+All functions will accept `**kwargs` as additional parameters. For example, the URLScan.io `/search` endpoint accepts a `size` parameter. You can include additional parameters like this:
+```python
+from ppp_connectors import urlscan
+r = urlscan.urlscan_search('domain:google.com', **{'size': 200})
+print(r.json())
+```
+Every individual API is different, so apply additional parameters after consulting the appropriate vendor's API documentation

ppp_connectors/ipqs.py

@@ -0,0 +1,43 @@
+from typing import Dict, Any, List
+from requests import Response
+from urllib.parse import quote
+from .broker import make_request
+from .helpers import check_required_env_vars, combine_env_configs
+
+env_config: Dict[str, Any] = combine_env_configs()
+
+def ipqs_malicious_url(query: str, **kwargs: Dict[str, Any]) -> Response:
+    """IPQualityScore's Malicious URL Scanner API scans links in real-time
+        to detect suspicious URLs. Accurately identify phishing links, malware
+        URLs and viruses, parked domains, and suspicious URLs with real-time risk
+        scores. Industry leading phishing detection and domain reputation provide
+        better signals for more accurate decision making.
+
+    Args:
+        query (str): The URL to scan
+
+    Returns:
+        Response: requests.Response json response from the request
+    """
+
+    # Define required environment variables
+    required_vars: List[str] = [
+        'IPQS_API_KEY'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    method: str = 'post'
+    url: str = f'https://ipqualityscore.com/api/json/url'
+    headers: Dict = {'accept': 'application/json'}
+    encoded_query: str = quote(query)
+    params: Dict = {
+        'key': env_config['IPQS_API_KEY'],
+        'url': encoded_query,
+        **kwargs
+    }
+
+    result: Response = make_request(method=method, url=url, headers=headers, json=params)
+
+    return result

ppp_connectors/urlscan.py

@@ -0,0 +1,102 @@
+from typing import Dict, Any, List
+from requests import Response
+from .broker import make_request
+from .helpers import check_required_env_vars, combine_env_configs
+
+env_config: Dict[str, Any] = combine_env_configs()
+
+def urlscan_search(query: str, **kwargs: Dict[str, Any]) -> Response:
+    """Find archived scans of URLs on urlscan.io. Search query syntax can
+        be found at https://urlscan.io/docs/search/
+
+    Args:
+        query (str): The query term (ElasticSearch Query String Query). Default: "*"
+
+    Returns:
+        Response: requests.Response json response from the request
+    """
+
+    # Define required environment variables
+    required_vars: List[str] = [
+        'URLSCAN_API_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://urlscan.io/api/v1/search/'
+    headers: Dict = {
+        'accept': 'application/json',
+        'API-Key': env_config['URLSCAN_API_KEY']
+    }
+    params: Dict = {
+        'q': query,
+        **kwargs
+    }
+
+    result: Response = make_request(method=method, url=url, headers=headers, params=params)
+
+    return result
+
+def urlscan_scan(query: str, **kwargs: Dict[str, Any]) -> Response:
+    """Submit a URL to be scanned
+
+    Args:
+        query (str): the URL to be scanned
+
+    Returns:
+        Response: requests.Response json response from the request
+    """
+
+    required_vars: List[str] = [
+        'URLSCAN_API_KEY'
+    ]
+
+    # Check and ensure that required variables are present, exits if not
+    check_required_env_vars(env_config, required_vars)
+
+    method: str = 'post'
+    url: str = 'https://urlscan.io/api/v1/scan'
+    headers: Dict = {
+        'accept': 'application/json',
+        'API-Key': env_config['URLSCAN_API_KEY']
+    }
+    payload: Dict = {
+        'url': query,
+        **kwargs
+    }
+
+    result: Response = make_request(method=method, url=url, headers=headers, json=payload)
+
+    return result
+
+def urlscan_results(uuid: str, **kwargs: Dict[str, Any]) -> Response:
+    """Retrieve results of a URLScan scan
+
+    Args:
+        uuid (str): the UUID of the submitted URL scan
+
+    Returns:
+        Response: requests.Response json response from the request
+    """
+
+    # Define required environment variables
+    required_vars: List[str] = [
+        'URLSCAN_API_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://urlscan.io/api/v1/result/{uuid}'
+    headers: Dict = {
+        'accept': 'application/json',
+        'API-Key': env_config['URLSCAN_API_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 = "ppp_connectors" }]
-version = "0.2.2"
+version = "0.3.0"
 description = "A simple, lightweight set of connectors and functions to various APIs, controlled by a central broker."
 authors = [
     "Rob D'Aveta <[email protected]>",