diff --git a/clients/python/README.md b/clients/python/README.md index 332c15735c..ca0d8d3362 100644 --- a/clients/python/README.md +++ b/clients/python/README.md @@ -1,12 +1,21 @@ # Tracker Python API Client -The Tracker Python API Client will provide a Python wrapper for key features of Tracker. +The Tracker Python API Client provides a simple Python interface for the [Tracker GraphQL API](https://github.com/canada-ca/tracker/blob/master/api-js/README.md), with the aim of allowing users to easily integrate data from Tracker into existing workflows and platforms. It allows access to the JSON data served by the API without requiring specific knowledge of [GraphQL](https://graphql.org/) or the Tracker API. This is done by providing functions that execute canned queries against the API using [gql](https://github.com/graphql-python/gql). Responses are formatted to remove pagination related structures, and to ensure useful keys are always present. -It makes use of [gql](https://github.com/graphql-python/gql), a Python GraphQL client, to query the Tracker API. -#### Installing Dependencies +## Installation -Install [pipenv](https://pypi.org/project/pipenv/) if you don't already have it. +### For Users + +The client will soon be available to install as a package via pip or pipenv. Until then, follow the instructions for developers below. + +### For Developers + +Install [pipenv](https://pypi.org/project/pipenv/) if you don't already have it. The following instructions assume you are using pipenv. + +#### Installing Dependencies + +Make sure you have pulled the most recent version from the repo, then run: ```shell pipenv install --dev @@ -18,7 +27,12 @@ If you run into issues, ensure pipenv has installed the most recent GQL version. pipenv install -e git+https://github.com/graphql-python/gql.git#egg=gql ``` -#### Authentication +## Usage + + +### Authentication + +You must have a Tracker account that is a member of one or more organizations to make use of the Python client. You can manage your account in the [Tracker web interface](https://tracker.alpha.canada.ca/). The client will attempt to draw credentials from its environment in order to obtain an authentication token. Pipenv makes this easy to set up by importing environment variables from a `.env` file present in this directory whenever `pipenv run` or `pipenv shell` are used. The `.env` file can be created like so: @@ -29,7 +43,122 @@ TRACKER_PASS=YOURPASSWORDHERE EOF ``` -#### Testing +You should be mindful that setting these variables manually can result in credentials being stored in your shell command history. + +### Basic Usage + +You will generally start by creating a client with `create_client(auth_token=get_auth_token())` and storing the result. All functions that make queries expect such a client to be passed as the first argument. + +### Examples + +#### Get all domains in my organizations + +Supposing I belong to two organizations with the acronyms "FOO" and "BAR": + +```python +>>> import tracker_client.client as tracker_client +>>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) +>>> print(tracker_client.get_all_domains(client)) +{ + "FOO": [ + "foo.bar", + "foo.bar.baz" + ], + "BAR": [ + "fizz.buzz", + "buzz.bang", + "ab.cd.ef", + ] +} +``` + +The following examples continue the previous one (assume the package has been imported as above and a `client` object with a valid token exists). + +#### Get a DMARC summary for a domain + +```python +>>> print(tracker_client.get_dmarc_summary(client, "foo.bar", "september", 2020)) +{ + "foo.bar": { + "month": "SEPTEMBER", + "year": "2020", + "categoryPercentages": { + "fullPassPercentage": 87, + "passSpfOnlyPercentage": 0, + "passDkimOnlyPercentage": 6, + "failPercentage": 8, + "totalMessages": 10534 + } + } +} +``` + +#### Get summary metrics for an organization + +```python +>>> print(tracker_client.get_summary_by_acronym(client, "foo")) +{ + "FOO": { + "domainCount": 10, + "summaries": { + "web": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 1, + "percentage": 10 + }, + { + "name": "fail", + "count": 9, + "percentage": 90 + } + ] + }, + "mail": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 5, + "percentage": 50 + }, + { + "name": "fail", + "count": 5, + "percentage": 50 + } + ] + } + } + } +} +``` + +#### Get the status of a domain + +```python +>>> print(tracker_client.get_domain_status(client, "foo.bar")) +{ + "foo.bar": { + "lastRan": "2021-01-23 22:33:26.921529", + "status": { + "https": "FAIL", + "ssl": "FAIL", + "dmarc": "PASS", + "dkim": "PASS", + "spf": "PASS" + } + } +} +``` + +> **NOTE**: Because of gql limitations, the client is not currently compatible with IPython or Jupyter. + +## Development + +### Testing Pytest is used for testing. To run tests, run the following in the project root (the folder containing this README.md): @@ -41,6 +170,9 @@ Alternatively, if you are already in a pipenv shell, just run `pytest`. If tests are failing with ModuleNotFoundError, make sure tracker_client/ is on your PYTHONPATH. The .env file used to store your credentials is a good way to set this. -#### Note about IPython/Jupyter +When additions or significant changes are made, check test coverage with: + +```shell +pipenv run pytest --cov=tracker_client +``` -Because of a limitation in gql, the client is not currently compatible with IPython or Jupyter. diff --git a/clients/python/tests/test_formatting.py b/clients/python/tests/test_formatting.py index d65903b0b9..24d18dd170 100644 --- a/clients/python/tests/test_formatting.py +++ b/clients/python/tests/test_formatting.py @@ -1,4 +1,8 @@ -from tracker_client.client import ( +"""Tests for query response formatting functions. + +The queries named in ALL_CAPS can be found in tracker_client/queries.py +""" +from tracker_client.formatting import ( format_all_domains, format_acronym_domains, format_name_domains, @@ -11,41 +15,52 @@ format_domain_status, ) + def test_format_all_domains(all_domains_input, all_domains_output): + """Test formatting of ALL_DOMAINS_QUERY results""" assert format_all_domains(all_domains_input) == all_domains_output def test_format_acronym_domains(all_domains_input, org_domains_output): + """Test formatting + filtering by acronym of ALL_DOMAINS_QUERY results""" assert format_acronym_domains(all_domains_input, "def") == org_domains_output def test_format_name_domains(name_domain_input, org_domains_output): + """Test formatting of DOMAINS_BY_SLUG results""" assert format_name_domains(name_domain_input) == org_domains_output def test_format_dmarc_monthly(monthly_dmarc_input, monthly_dmarc_output): + """Test formatting of DMARC_SUMMARY results""" assert format_dmarc_monthly(monthly_dmarc_input) == monthly_dmarc_output def test_format_dmarc_yearly(yearly_dmarc_input, yearly_dmarc_output): + """Test formatting of YEARLY_DMARC_SUMMARIES results""" assert format_dmarc_yearly(yearly_dmarc_input) == yearly_dmarc_output def test_format_all_summaries(all_summaries_input, all_summaries_output): + """Test formatting of ALL_ORGS_SUMMARIES results""" assert format_all_summaries(all_summaries_input) == all_summaries_output def test_format_acronym_summary(all_summaries_input, org_summary_output): + """Test formatting + filtering by acronym of ALL_ORGS_SUMMARIES results""" assert format_acronym_summary(all_summaries_input, "def") == org_summary_output def test_format_name_summary(name_summary_input, org_summary_output): + """Test formatting of SUMMARY_BY_SLUG results""" assert format_name_summary(name_summary_input) == org_summary_output def test_format_domain_results(scan_results_input, scan_results_output): + """Test formatting of DOMAIN_RESULTS results""" assert format_domain_results(scan_results_input) == scan_results_output def test_format_domain_status(domain_status_input, domain_status_output): + """Test formatting of DOMAIN_STATUS results""" assert format_domain_status(domain_status_input) == domain_status_output diff --git a/clients/python/tests/test_get_dmarc.py b/clients/python/tests/test_get_dmarc.py index 0ce83774e7..21329af098 100644 --- a/clients/python/tests/test_get_dmarc.py +++ b/clients/python/tests/test_get_dmarc.py @@ -1,3 +1,4 @@ +"""Tests for functions that get DMARC summaries for a domain""" import json from gql import Client diff --git a/clients/python/tests/test_get_domains.py b/clients/python/tests/test_get_domains.py index 0722cfcb7d..93da1a259f 100644 --- a/clients/python/tests/test_get_domains.py +++ b/clients/python/tests/test_get_domains.py @@ -1,3 +1,4 @@ +"""Tests for functions that get domain lists""" import json from gql import Client diff --git a/clients/python/tests/test_get_results.py b/clients/python/tests/test_get_results.py index 82f368574b..487d998281 100644 --- a/clients/python/tests/test_get_results.py +++ b/clients/python/tests/test_get_results.py @@ -1,3 +1,4 @@ +"""Tests for functions that get scan results for a domain""" import json from gql import Client diff --git a/clients/python/tests/test_get_summaries.py b/clients/python/tests/test_get_summaries.py index 4d83c3cba7..03706a4c4e 100644 --- a/clients/python/tests/test_get_summaries.py +++ b/clients/python/tests/test_get_summaries.py @@ -1,3 +1,4 @@ +"""Tests for functions that get summary metrics""" import json from gql import Client diff --git a/clients/python/tests/test_gql.py b/clients/python/tests/test_gql.py index fa4bbbeb20..41f37eb3d9 100644 --- a/clients/python/tests/test_gql.py +++ b/clients/python/tests/test_gql.py @@ -1,3 +1,4 @@ +"""Tests for gql related utility functions""" import re import pytest diff --git a/clients/python/tracker_client/client.py b/clients/python/tracker_client/client.py index f7ff5cc110..a3827a864b 100644 --- a/clients/python/tracker_client/client.py +++ b/clients/python/tracker_client/client.py @@ -1,3 +1,5 @@ +"""Provides functions that get JSON data from Tracker (https://github.com/canada-ca/tracker)""" + import json import os import re @@ -11,6 +13,7 @@ TransportProtocolError, ) +# TODO: decide if we should just import whole modules instead from queries import ( ALL_DOMAINS_QUERY, DOMAINS_BY_SLUG, @@ -22,17 +25,35 @@ DOMAIN_RESULTS, DOMAIN_STATUS, ) +from formatting import ( + format_all_domains, + format_acronym_domains, + format_name_domains, + format_dmarc_monthly, + format_dmarc_yearly, + format_all_summaries, + format_acronym_summary, + format_name_summary, + format_domain_results, + format_domain_status, +) JWT_RE = r"^[A-Za-z0-9-_=]+\.[A-Za-z0-9-_=]+\.?[A-Za-z0-9-_.+/=]*$" +"""Regex to validate a JWT""" def create_transport(url, auth_token=None): """Create and return a gql transport object - Arguments: - url -- the Tracker GraphQL endpoint url - auth_token -- JWT auth token string, omit when initially obtaining the token + Users should rarely, if ever, need to call this + + :param str url: the Tracker GraphQL endpoint url + :param str auth_token: JWT auth token, omit when initially obtaining the token (default is none) + :return: A gql transport for given url + :rtype: AIOHTTPTransport + :raises ValueError: if auth_token is not a valid JWT + :raises TypeError: if auth_token is not a string """ if auth_token is None: transport = AIOHTTPTransport(url=url) @@ -58,9 +79,10 @@ def create_transport(url, auth_token=None): def create_client(url="https://tracker.alpha.canada.ca/graphql", auth_token=None): """Create and return a gql client object - Arguments: - url -- the Tracker GraphQL endpoint url - auth_token -- JWT auth token string, omit when initially obtaining the token + :param str url: the Tracker GraphQL endpoint url (default is "https://tracker.alpha.canada.ca/graphql") + :param str auth_token: JWT auth token, omit when initially obtaining the token (default is None) + :return: A gql client with AIOHTTPTransport + :rtype: Client """ client = Client( transport=create_transport(url=url, auth_token=auth_token), @@ -70,7 +92,14 @@ def create_client(url="https://tracker.alpha.canada.ca/graphql", auth_token=None def get_auth_token(url="https://tracker.alpha.canada.ca/graphql"): - """Takes in environment variables "TRACKER_UNAME" and "TRACKER_PASS", returns an auth token""" + """Get a token to use for authentication. + + Takes in environment variables "TRACKER_UNAME" and "TRACKER_PASS" to get credentials + + :param str url: the Tracker GraphQL endpoint url (default is "https://tracker.alpha.canada.ca/graphql") + :return: JWT auth token to allow access to Tracker + :rtype: str + """ client = create_client(url) username = os.environ.get("TRACKER_UNAME") @@ -88,7 +117,20 @@ def get_auth_token(url="https://tracker.alpha.canada.ca/graphql"): # TODO: Make error messages better def execute_query(client, query, params=None): - """Executes a query on given client, with given parameters. """ + """Executes a query on given client, with given parameters. + + Intended for internal use, but if for some reason you need an unformatted + response from the API you could call this. + + :param Client client: a gql client to execute the query on + :param DocumentNode query: a gql query string that has been parsed with gql() + :param dict params: variables to pass along with query + :return: Results of executing query on API + :rtype: dict + :raises TransportProtocolError: if server response is not GraphQL + :raises TransportServerError: if there is a server error + :raises Exception: if any unhandled exception is raised within function + """ try: result = client.execute(query, variable_values=params) @@ -114,10 +156,28 @@ def execute_query(client, query, params=None): def get_all_domains(client): - """Returns lists of all domains you have ownership of, with org as key - - Arguments: - client -- a GQL Client object + """Get lists of all domains you have ownership of, with org as key + + :param Client client: a gql Client object + :return: formatted JSON data with all organizations and their domains + :rtype: str + + :Example: + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_all_domains(client)) + { + "FOO": [ + "foo.bar", + "foo.bar.baz" + ], + "BAR": [ + "fizz.buzz", + "buzz.bang", + "ab.cd.ef", + ] + } """ result = execute_query(client, ALL_DOMAINS_QUERY) # If there is an error the result contains the key "error" @@ -127,31 +187,25 @@ def get_all_domains(client): return json.dumps(result, indent=4) -def format_all_domains(result): - """ Formats the dict obtained by ALL_DOMAINS_QUERY """ - # Extract the list of nodes from the resulting dict - result = result["findMyOrganizations"]["edges"] - # Move the dict value of "node" up a level - result = [n["node"] for n in result] - - # For each dict element of the list, change the value of "domains" - # to the list of domains contained in the nodes of its edges - for x in result: - x["domains"] = x["domains"]["edges"] - x["domains"] = [n["node"] for n in x["domains"]] - x["domains"] = [n["domain"] for n in x["domains"]] - - # Create a new dict in the desired format to return - result = {x["acronym"]: x["domains"] for x in result} - return result - - def get_domains_by_acronym(client, acronym): - """Return the domains belonging to the organization identified by acronym - - Arguments: - acronym -- string containing an acronym belonging to an organization - client -- a GQL Client object + """Get the domains belonging to the organization identified by acronym + + :param Client client: a gql Client object + :param str acronym: an acronym referring to an organization + :return: formatted JSON data with an organization's domains + :rtype: str + + :Example: + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_domains_by_acronym(client, "foo")) + { + "FOO": [ + "foo.bar", + "foo.bar.baz" + ] + } """ # API doesn't allow query by acronym so we filter the get_all_domains result result = execute_query(client, ALL_DOMAINS_QUERY) @@ -172,19 +226,25 @@ def get_domains_by_acronym(client, acronym): return json.dumps(result, indent=4) -def format_acronym_domains(result, acronym): - """ Formats the dict obtained by ALL_DOMAINS_QUERY to show only one org""" - result = format_all_domains(result) - result = {acronym.upper(): result[acronym.upper()]} - return result - - def get_domains_by_name(client, name): - """Return the domains belonging to the organization identified by full name - - Arguments: - name -- string containing the name of an organization - client -- a GQL Client object + """Get the domains belonging to the organization identified by name + + :param Client client: a gql Client object + :param str name: the full name of an organization + :return: formatted JSON data with an organization's domains + :rtype: str + + :Example: + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_domains_by_name(client, "foo bar")) + { + "FOO": [ + "foo.bar", + "foo.bar.baz" + ] + } """ slugified_name = slugify(name) # API expects a slugified string for name params = {"orgSlug": slugified_name} @@ -197,24 +257,34 @@ def get_domains_by_name(client, name): return json.dumps(result, indent=4) -def format_name_domains(result): - """Formats the dict obtained by DOMAINS_BY_SLUG""" - result = result["findOrganizationBySlug"] - result["domains"] = result["domains"]["edges"] - result["domains"] = [n["node"] for n in result["domains"]] - result["domains"] = [n["domain"] for n in result["domains"]] - result = {result["acronym"]: result["domains"]} - return result - - def get_dmarc_summary(client, domain, month, year): - """Return the DMARC summary for the specified domain and month - - Arguments: - domain -- domain name string - month -- string containing the full name of a month - year -- positive integer representing a year - client -- a GQL Client object + """Get the DMARC summary for the specified domain and month + + :param Client client: a gql Client object + :param str domain: the domain to get a DMARC summary for + :param str month: the full name of a month + :param int year: positive integer representing a year + :return: formatted JSON data with a DMARC summary + :rtype: str + + :Example: + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_dmarc_summary(client, "foo.bar", "september", 2020)) + { + "foo.bar": { + "month": "SEPTEMBER", + "year": "2020", + "categoryPercentages": { + "fullPassPercentage": 87, + "passSpfOnlyPercentage": 0, + "passDkimOnlyPercentage": 6, + "failPercentage": 8, + "totalMessages": 10534 + } + } + } """ params = {"domain": domain, "month": month.upper(), "year": str(year)} @@ -226,19 +296,47 @@ def get_dmarc_summary(client, domain, month, year): return json.dumps(result, indent=4) -def format_dmarc_monthly(result): - """Formats the dict obtained by DMARC_SUMMARY""" - result = result["findDomainByDomain"] - result[result.pop("domain")] = result.pop("dmarcSummaryByPeriod") - return result - - def get_yearly_dmarc_summaries(client, domain): - """Return yearly DMARC summaries for a domain - - Arguments: - domain -- domain name string - client -- a GQL Client object + """Get yearly DMARC summaries for a domain + + :param Client client: a gql Client object + :param str domain: domain to get DMARC summaries for + :return: formatted JSON data with yearly DMARC summaries + :rtype: str + + :Example: + + Output is truncated, you should expect more than 2 reports in the list + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_yearly_dmarc_summaries(client, "foo.bar")) + { + "foo.bar": [ + { + "month": "AUGUST", + "year": "2020", + "categoryPercentages": { + "fullPassPercentage": 90, + "passSpfOnlyPercentage": 0, + "passDkimOnlyPercentage": 5, + "failPercentage": 5, + "totalMessages": 7045 + } + }, + { + "month": "JULY", + "year": "2020", + "categoryPercentages": { + "fullPassPercentage": 82, + "passSpfOnlyPercentage": 0, + "passDkimOnlyPercentage": 11, + "failPercentage": 8, + "totalMessages": 6647 + } + }, + ] + } """ params = {"domain": domain} @@ -250,18 +348,90 @@ def get_yearly_dmarc_summaries(client, domain): return json.dumps(result, indent=4) -def format_dmarc_yearly(result): - """Formats the dict obtained by DMARC_YEARLY_SUMMARIES""" - result = result["findDomainByDomain"] - result[result.pop("domain")] = result.pop("yearlyDmarcSummaries") - return result - - def get_all_summaries(client): - """Returns summary metrics for all organizations you are a member of. - - Arguments: - client -- a GQL Client object + """Get summary metrics for all organizations you are a member of. + + :param Client client: a gql Client object + :return: formatted JSON data with all organizations and their metrics + :rtype: str + + :Example: + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_all_summaries(client)) + { + "FOO": { + "domainCount": 10, + "summaries": { + "web": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 1, + "percentage": 10 + }, + { + "name": "fail", + "count": 9, + "percentage": 90 + } + ] + }, + "mail": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 5, + "percentage": 50 + }, + { + "name": "fail", + "count": 5, + "percentage": 50 + } + ] + } + } + }, + "BAR": { + "domainCount": 10, + "summaries": { + "web": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 5, + "percentage": 50 + }, + { + "name": "fail", + "count": 5, + "percentage": 50 + } + ] + }, + "mail": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 1, + "percentage": 10 + }, + { + "name": "fail", + "count": 9, + "percentage": 90 + } + ] + } + } + } + } """ result = execute_query(client, ALL_ORG_SUMMARIES) @@ -271,19 +441,56 @@ def get_all_summaries(client): return json.dumps(result, indent=4) -def format_all_summaries(result): - """Formats the dict obtained by ALL_ORG_SUMMARIES""" - result = result["findMyOrganizations"]["edges"] - result = {x["node"].pop("acronym"): x["node"] for x in result} - return result - - def get_summary_by_acronym(client, acronym): """Returns summary metrics for the organization identified by acronym - Arguments: - acronym -- string containing an acronym belonging to an organization - client -- a GQL Client object + :param Client client: a GQL Client object + :param str acronym: an acronym referring to an organization + :return: formatted JSON with summary metrics for an organization + :rtype: str + + :Example: + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_summary_by_acronym(client, "foo")) + { + "FOO": { + "domainCount": 10, + "summaries": { + "web": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 1, + "percentage": 10 + }, + { + "name": "fail", + "count": 9, + "percentage": 90 + } + ] + }, + "mail": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 5, + "percentage": 50 + }, + { + "name": "fail", + "count": 5, + "percentage": 50 + } + ] + } + } + } + } """ # API doesn't allow query by acronym so we filter the get_all_summaries result result = execute_query(client, ALL_ORG_SUMMARIES) @@ -302,20 +509,56 @@ def get_summary_by_acronym(client, acronym): return json.dumps(result, indent=4) -def format_acronym_summary(result, acronym): - """Formats the dict obtained by ALL_ORG_SUMMARIES to show only one org""" - result = format_all_summaries(result) - # dict in assignment is to keep the org identified in the return value - result = {acronym.upper(): result[acronym.upper()]} - return result - - def get_summary_by_name(client, name): - """Return summary metrics for the organization identified by name - - Arguments: - name -- string containing the name of an organization - client -- a GQL Client object + """Get summary metrics for the organization identified by name + + :param Client client: a gql Client object + :param str name: the full name of an organization + :return: formatted JSON data with summary metrics for an organization + :rtype: str + + :Example: + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_summary_by_name(client, "foo bar")) + { + "FOO": { + "domainCount": 10, + "summaries": { + "web": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 1, + "percentage": 10 + }, + { + "name": "fail", + "count": 9, + "percentage": 90 + } + ] + }, + "mail": { + "total": 10, + "categories": [ + { + "name": "pass", + "count": 5, + "percentage": 50 + }, + { + "name": "fail", + "count": 5, + "percentage": 50 + } + ] + } + } + } + } """ slugified_name = slugify(name) # API expects a slugified string for name params = {"orgSlug": slugified_name} @@ -328,22 +571,17 @@ def get_summary_by_name(client, name): return json.dumps(result, indent=4) -def format_name_summary(result): - """Formats the dict obtained by SUMMARY_BY_SLUG""" - result = { - result["findOrganizationBySlug"].pop("acronym"): result[ - "findOrganizationBySlug" - ] - } - return result +def get_domain_results(client, domain): + """Get scan results for a domain + :param Client client: a gql Client object + :param str domain: domain to get results for + :return: formatted JSON data with scan results for the domain + :rtype: str -def get_domain_results(client, domain): - """Return scan results for a domain + :Example: - Arguments: - domain -- domain name string - client -- a GQL Client object + Coming soon, function likely to change """ params = {"domain": domain} @@ -355,59 +593,31 @@ def get_domain_results(client, domain): return json.dumps(result, indent=4) -def format_domain_results(result): - """Format the dict obtained by DOMAIN_RESULTS""" - - # Extract the contents of the list of nodes holding web results - result["findDomainByDomain"]["web"] = { - k: v["edges"][0]["node"] - for (k, v) in result["findDomainByDomain"]["web"].items() - } - - # Extract the contents of the list of nodes holding email results - result["findDomainByDomain"]["email"] = { - k: v["edges"][0]["node"] - for (k, v) in result["findDomainByDomain"]["email"].items() - } - - # Extract the contents of the list of edges for guidance tags - for x in result["findDomainByDomain"]["web"].keys(): - - # Remove edges by making the value of guidanceTags the list of nodes - result["findDomainByDomain"]["web"][x]["guidanceTags"] = result[ - "findDomainByDomain" - ]["web"][x]["guidanceTags"]["edges"] - - # Replace the list of nodes with a dict with tagIds as the keys - result["findDomainByDomain"]["web"][x]["guidanceTags"] = { - x["node"].pop("tagId"): x["node"] - for x in result["findDomainByDomain"]["web"][x]["guidanceTags"] - } - - # Do the same with email guidance tags - for x in result["findDomainByDomain"]["email"].keys(): - - # dkim results have different structure so exclude them - if x != "dkim": - result["findDomainByDomain"]["email"][x]["guidanceTags"] = result[ - "findDomainByDomain" - ]["email"][x]["guidanceTags"]["edges"] - - result["findDomainByDomain"]["email"][x]["guidanceTags"] = { - x["node"].pop("tagId"): x["node"] - for x in result["findDomainByDomain"]["email"][x]["guidanceTags"] - } - - result = {result["findDomainByDomain"].pop("domain"): result["findDomainByDomain"]} - return result - - def get_domain_status(client, domain): """Return pass/fail status information for a domain - Arguments: - domain -- domain name string - client -- a GQL Client object + :param Client client: a gql Client object + :param str domain: domain to get the status of + :return: formatted JSON data with the domain's status + :rtype: str + + :Example: + + >>> import tracker_client.client as tracker_client + >>> client = tracker_client.create_client(auth_token=tracker_client.get_auth_token()) + >>> print(tracker_client.get_domain_status(client, "foo.bar")) + { + "foo.bar": { + "lastRan": "2021-01-23 22:33:26.921529", + "status": { + "https": "FAIL", + "ssl": "FAIL", + "dmarc": "PASS", + "dkim": "PASS", + "spf": "PASS" + } + } + } """ params = {"domain": domain} @@ -419,15 +629,9 @@ def get_domain_status(client, domain): return json.dumps(result, indent=4) -def format_domain_status(result): - """Formats the dict obtained by DOMAIN_STATUS""" - result = {result["findDomainByDomain"].pop("domain"): result["findDomainByDomain"]} - return result - - def main(): # pragma: no cover """main() currently tries all implemented functions and prints results - for diagnostic purposes and to demo available features. + for diagnostic purposes and to demo available features. To be removed in future. """ acronym = "cse" name = "Communications Security Establishment Canada" diff --git a/clients/python/tracker_client/formatting.py b/clients/python/tracker_client/formatting.py new file mode 100644 index 0000000000..ea5e5e1623 --- /dev/null +++ b/clients/python/tracker_client/formatting.py @@ -0,0 +1,180 @@ +"""Provides results formatting functions used in client.py""" + + +def format_all_domains(result): + """Formats the result dict in get_all_domains + + :param dict result: unformatted dict with results of ALL_DOMAINS_QUERY + :return: formatted results + :rtype: dict + """ + # Extract the list of nodes from the resulting dict + result = result["findMyOrganizations"]["edges"] + # Move the dict value of "node" up a level + result = [n["node"] for n in result] + + # For each dict element of the list, change the value of "domains" + # to the list of domains contained in the nodes of its edges + for x in result: + x["domains"] = x["domains"]["edges"] + x["domains"] = [n["node"] for n in x["domains"]] + x["domains"] = [n["domain"] for n in x["domains"]] + + # Create a new dict in the desired format to return + result = {x["acronym"]: x["domains"] for x in result} + return result + + +def format_acronym_domains(result, acronym): + """Formats the result dict in get_domains_by_acronym + + :param dict result: unformatted dict with results of ALL_DOMAINS_QUERY + :param str acronym: an acronym referring to an organization + :return: formatted results, filtered to only the org identified by acronym + :rtype: dict + :raises KeyError: if user does not have membership in an org with a matching acronym + """ + result = format_all_domains(result) + result = {acronym.upper(): result[acronym.upper()]} + return result + + +def format_name_domains(result): + """Formats the result dict in get_domains_by_name + + :param dict result: unformatted dict with results of DOMAINS_BY_SLUG + :return: formatted results + :rtype: dict + """ + result = result["findOrganizationBySlug"] + result["domains"] = result["domains"]["edges"] + result["domains"] = [n["node"] for n in result["domains"]] + result["domains"] = [n["domain"] for n in result["domains"]] + result = {result["acronym"]: result["domains"]} + return result + + +def format_dmarc_monthly(result): + """Formats the result dict in get_dmarc_summary + + :param dict result: unformatted dict with results of DMARC_SUMMARY + :return: formatted results + :rtype: dict + """ + result = result["findDomainByDomain"] + result[result.pop("domain")] = result.pop("dmarcSummaryByPeriod") + return result + + +def format_dmarc_yearly(result): + """Formats the result dict in get_yearly_dmarc_summaries + + :param dict result: unformatted dict with results of YEARLY_DMARC_SUMMARIES + :return: formatted results + :rtype: dict + """ + result = result["findDomainByDomain"] + result[result.pop("domain")] = result.pop("yearlyDmarcSummaries") + return result + + +def format_all_summaries(result): + """Formats the result dict in get_all_summaries + + :param dict result: unformatted dict with results of ALL_ORG_SUMMARIES + :return: formatted results + :rtype: dict + """ + result = result["findMyOrganizations"]["edges"] + result = {x["node"].pop("acronym"): x["node"] for x in result} + return result + + +def format_acronym_summary(result, acronym): + """Formats the result dict in get_summary_by_acronym + + :param dict result: unformatted dict with results of ALL_ORG_SUMMARIES + :param str acronym: an acronym referring to an organization + :return: formatted results, filtered to only the org identified by acronym + :rtype: dict + :raises KeyError: if user does not have membership in an org with a matching acronym + """ + result = format_all_summaries(result) + # dict in assignment is to keep the org identified in the return value + result = {acronym.upper(): result[acronym.upper()]} + return result + + +def format_name_summary(result): + """Formats the result dict in get_summary_by_name + + :param dict result: unformatted dict with results of SUMMARY_BY_SLUG + :return: formatted results + :rtype: dict""" + result = { + result["findOrganizationBySlug"].pop("acronym"): result[ + "findOrganizationBySlug" + ] + } + return result + + +def format_domain_results(result): + """Formats the result dict in get_domain_results + + :param dict result: unformatted dict with results of DOMAIN_RESULTS + :return: formatted results + :rtype: dict""" + + # Extract the contents of the list of nodes holding web results + result["findDomainByDomain"]["web"] = { + k: v["edges"][0]["node"] + for (k, v) in result["findDomainByDomain"]["web"].items() + } + + # Extract the contents of the list of nodes holding email results + result["findDomainByDomain"]["email"] = { + k: v["edges"][0]["node"] + for (k, v) in result["findDomainByDomain"]["email"].items() + } + + # Extract the contents of the list of edges for guidance tags + for x in result["findDomainByDomain"]["web"].keys(): + + # Remove edges by making the value of guidanceTags the list of nodes + result["findDomainByDomain"]["web"][x]["guidanceTags"] = result[ + "findDomainByDomain" + ]["web"][x]["guidanceTags"]["edges"] + + # Replace the list of nodes with a dict with tagIds as the keys + result["findDomainByDomain"]["web"][x]["guidanceTags"] = { + x["node"].pop("tagId"): x["node"] + for x in result["findDomainByDomain"]["web"][x]["guidanceTags"] + } + + # Do the same with email guidance tags + for x in result["findDomainByDomain"]["email"].keys(): + + # dkim results have different structure so exclude them + if x != "dkim": + result["findDomainByDomain"]["email"][x]["guidanceTags"] = result[ + "findDomainByDomain" + ]["email"][x]["guidanceTags"]["edges"] + + result["findDomainByDomain"]["email"][x]["guidanceTags"] = { + x["node"].pop("tagId"): x["node"] + for x in result["findDomainByDomain"]["email"][x]["guidanceTags"] + } + + result = {result["findDomainByDomain"].pop("domain"): result["findDomainByDomain"]} + return result + + +def format_domain_status(result): + """Formats the result dict in get_domain_status + + :param dict result: unformatted dict with results of DOMAIN_STATUS + :return: formatted results + :rtype: dict""" + result = {result["findDomainByDomain"].pop("domain"): result["findDomainByDomain"]} + return result diff --git a/clients/python/tracker_client/queries.py b/clients/python/tracker_client/queries.py index bbb390c60a..53dbb62ae1 100644 --- a/clients/python/tracker_client/queries.py +++ b/clients/python/tracker_client/queries.py @@ -1,6 +1,21 @@ -from gql import gql +""" This module contains the gql documents used by client.py to query the Tracker API + +:var DocumentNode SIGN_IN_MUTATION: sign in and get authentication token +:var DocumentNode ALL_DOMAINS_QUERY: get all organizations and their domains +:var DocumentNode DOMAINS_BY_SLUG: get all domains for one organization +:var DocumentNode DMARC_SUMMARY: get a domain's DMARC summary for one month +:var DocumentNode YEARLY_DMARC_SUMMARIES: get a domain's yearly DMARC summaries +:var DocumentNode ALL_ORG_SUMMARIES: get summary metrics for all organizations +:var DocumentNode SUMMARY_BY_SLUG: get summary metrics for one organization +:var DocumentNode DOMAIN_RESULTS: get scan results for a domain +:var DocumentNode DOMAIN_STATUS: get pass/fail compliance statuses for a domain +""" +from gql import gql +# Sign in to Tracker and obtain an authentication token +# :param dict creds: a dict with a username and password +# Mutation variables should look like {"creds":{"userName": ${username}, "password": ${password}}} SIGNIN_MUTATION = gql( """ mutation signIn($creds: SignInInput!) { @@ -17,6 +32,8 @@ """ ) +# Get all organizations the user belongs to and all domains those organizations own +# Pagination details (edges and nodes) are stripped during formatting of response ALL_DOMAINS_QUERY = gql( """ query getAllDomains { @@ -30,7 +47,7 @@ domain } } - } + } } } } @@ -38,6 +55,11 @@ """ ) +# Get an organization by its slugified name and all domains that organization owns +# slugified-looks-like-this (all lowercase, alphanumeric, spaces replaced with hyphens) +# :param str orgSlug: a slugified organization name +# Query variables should look like {"orgSlug": ${slugified-str} } +# Pagination details (edges and nodes) are stripped during formatting of response DOMAINS_BY_SLUG = gql( """ query orgBySlug($orgSlug: Slug!){ @@ -55,6 +77,11 @@ """ ) +# Get the DMARC summary for one month for one domain +# :param str domain: url to get DMARC summary for +# :param str month: full name of a month in ALL CAPS to get summary for +# :param str year: year to get summary for +# Query variables should look like: {"domain": ${domain_url}, "month": ${MONTH_IN_ALL_CAPS}, "year": ${numeric_year_as_str}} DMARC_SUMMARY = gql( """ query domainDMARCSummary( @@ -80,6 +107,9 @@ """ ) +# Get yearly DMARC summaries for a domain +# :param str domain: url to get DMARC summary for +# Query variables should look like {"domain": ${domain_url} } DMARC_YEARLY_SUMMARIES = gql( """ query domainAllDMARCSummaries($domain: DomainScalar!) { @@ -101,6 +131,8 @@ """ ) +# Get summary metrics for all organizations user is a member of +# Pagination details (edges and nodes) are stripped during formatting of response ALL_ORG_SUMMARIES = gql( """ query getAllSummaries { @@ -135,6 +167,10 @@ """ ) +# Get an organization by its slugified name and get summary metrics +# slugified-looks-like-this (all lowercase, alphanumeric, spaces replaced with hyphens) +# :param str orgSlug: a slugified organization name +# Query variables should look like {"orgSlug": ${slugified-str} } SUMMARY_BY_SLUG = gql( """ query getSummaryBySlug($orgSlug: Slug!) { @@ -164,6 +200,11 @@ """ ) +# Get scan results for a domain +# :param str domain: url to get scan results +# Query variables should look like {"domain": ${domain_url} } +# Returns many fields, guidance tags are likely to be of most interest +# Pagination details (edges and nodes) are stripped during formatting of response DOMAIN_RESULTS = gql( """ query GetScanResults($domain: DomainScalar!) { @@ -317,6 +358,9 @@ """ ) +# Get pass/fail status indicators for a domain's compliance +# :param str domain: url to get scan results +# Query variables should look like {"domain": ${domain_url} } DOMAIN_STATUS = gql( """ query GetDomainStatus($domain: DomainScalar!) {