From 210248af6a74739163dfd912e1ca248f3fe8f72d Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Mon, 15 Feb 2021 14:44:00 -0400 Subject: [PATCH 01/11] First pass at fleshed out README --- clients/python/README.md | 145 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 140 insertions(+), 5 deletions(-) diff --git a/clients/python/README.md b/clients/python/README.md index 332c15735c..663aa53737 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,6 +43,121 @@ TRACKER_PASS=YOURPASSWORDHERE EOF ``` +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 +>>> from tracker_client import client +>>> client = create_client(auth_token=get_auth_token()) +>>> 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 and a `client` object with a valid token exists). + +#### Get a DMARC summary for a domain + +```python +>>> 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 +>>> 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 +>>> get_domain_status(client, "foo.bar") +Getting domain status for cse-cst.gc.ca... +{ + "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,12 @@ 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. +When additions or significant changes are made, check test coverage with: + +```shell +pipenv run pytest --cov=tracker_client +``` + #### Note about IPython/Jupyter Because of a limitation in gql, the client is not currently compatible with IPython or Jupyter. From 033390a20d72e27a05f9c2a9c2166dc15a5c49a3 Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Mon, 15 Feb 2021 14:47:11 -0400 Subject: [PATCH 02/11] Remove duplicate IPython notice in README --- clients/python/README.md | 3 --- 1 file changed, 3 deletions(-) diff --git a/clients/python/README.md b/clients/python/README.md index 663aa53737..268edb78ef 100644 --- a/clients/python/README.md +++ b/clients/python/README.md @@ -176,6 +176,3 @@ When additions or significant changes are made, check test coverage with: pipenv run pytest --cov=tracker_client ``` -#### Note about IPython/Jupyter - -Because of a limitation in gql, the client is not currently compatible with IPython or Jupyter. From ff60297754763a65f4c76110cececfc6c63374ad Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Mon, 15 Feb 2021 14:49:44 -0400 Subject: [PATCH 03/11] Improve consistency of headers --- clients/python/README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/clients/python/README.md b/clients/python/README.md index 268edb78ef..98cb74fc0a 100644 --- a/clients/python/README.md +++ b/clients/python/README.md @@ -156,9 +156,9 @@ Getting domain status for cse-cst.gc.ca... > **NOTE**: Because of gql limitations, the client is not currently compatible with IPython or Jupyter. -### Development +## Development -#### Testing +### Testing Pytest is used for testing. To run tests, run the following in the project root (the folder containing this README.md): From d2d0328d5012594b4002229ff17cc13d7dc85009 Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Mon, 15 Feb 2021 15:06:50 -0400 Subject: [PATCH 04/11] Fix examples match output shown --- clients/python/README.md | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/clients/python/README.md b/clients/python/README.md index 98cb74fc0a..ca0d8d3362 100644 --- a/clients/python/README.md +++ b/clients/python/README.md @@ -52,12 +52,13 @@ You will generally start by creating a client with `create_client(auth_token=get ### Examples #### Get all domains in my organizations + Supposing I belong to two organizations with the acronyms "FOO" and "BAR": ```python ->>> from tracker_client import client ->>> client = create_client(auth_token=get_auth_token()) ->>> get_all_domains(client) +>>> 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", @@ -71,12 +72,12 @@ Supposing I belong to two organizations with the acronyms "FOO" and "BAR": } ``` -The following examples continue the previous one (assume the package has been imported and a `client` object with a valid token exists). +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 ->>> get_dmarc_summary(client, "foo.bar", "september", 2020) +>>> print(tracker_client.get_dmarc_summary(client, "foo.bar", "september", 2020)) { "foo.bar": { "month": "SEPTEMBER", @@ -95,7 +96,7 @@ The following examples continue the previous one (assume the package has been im #### Get summary metrics for an organization ```python ->>> get_summary_by_acronym(client, "foo") +>>> print(tracker_client.get_summary_by_acronym(client, "foo")) { "FOO": { "domainCount": 10, @@ -138,8 +139,7 @@ The following examples continue the previous one (assume the package has been im #### Get the status of a domain ```python ->>> get_domain_status(client, "foo.bar") -Getting domain status for cse-cst.gc.ca... +>>> print(tracker_client.get_domain_status(client, "foo.bar")) { "foo.bar": { "lastRan": "2021-01-23 22:33:26.921529", From bc10696e9d43397167ff97e08ac934f2facadcb3 Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Mon, 15 Feb 2021 17:25:40 -0400 Subject: [PATCH 05/11] Convert existing docstrings to rst style --- clients/python/tracker_client/client.py | 180 +++++++++++++++++------- 1 file changed, 131 insertions(+), 49 deletions(-) diff --git a/clients/python/tracker_client/client.py b/clients/python/tracker_client/client.py index f7ff5cc110..ac4a407074 100644 --- a/clients/python/tracker_client/client.py +++ b/clients/python/tracker_client/client.py @@ -30,9 +30,12 @@ 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 + :param str url: the Tracker GraphQL endpoint url + :param str auth_token: JWT auth token string, omit when initially obtaining the token (default is none) + :raise: ValueError if auth_token is not a valid JWT + :raise: TypeError if auth_token is not a string + :return: A gql transport + :rtype: AIOHTTPTransport """ if auth_token is None: transport = AIOHTTPTransport(url=url) @@ -58,9 +61,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 string, 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 +74,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 + :rtype: str + """ client = create_client(url) username = os.environ.get("TRACKER_UNAME") @@ -88,7 +99,17 @@ 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. + + :param Client client: a gql client + :param DocumentNode query: a gql query string that has been parsed with gql() + :param dict params: variables to pass along with query + :raise: TransportProtocolError if server response is not GraphQL + :raise: TransportServerError if there is a server error + :raise: Exception if any unhandled exception is raised within function + :return: Results of executing query on API + :rtype: dict + """ try: result = client.execute(query, variable_values=params) @@ -116,8 +137,9 @@ 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 + :param Client client: a GQL Client object + :return: formatted JSON data with all organizations and their domains as string + :rtype: str """ result = execute_query(client, ALL_DOMAINS_QUERY) # If there is an error the result contains the key "error" @@ -128,7 +150,12 @@ def get_all_domains(client): def format_all_domains(result): - """ Formats the dict obtained by ALL_DOMAINS_QUERY """ + """ 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 @@ -149,9 +176,10 @@ def format_all_domains(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 + :param Client client: a GQL Client object + :param str acronym: string containing an acronym belonging to an organization + :return: formatted JSON data as string + :rtype: str """ # API doesn't allow query by acronym so we filter the get_all_domains result result = execute_query(client, ALL_DOMAINS_QUERY) @@ -173,18 +201,26 @@ def get_domains_by_acronym(client, acronym): def format_acronym_domains(result, acronym): - """ Formats the dict obtained by ALL_DOMAINS_QUERY to show only one org""" + """ Formats the result dict in get_domains_by_acronym + + :param dict result: unformatted dict with results of ALL_DOMAINS_QUERY + :param str acronym: string containing an acronym belonging to an organization + :raise: KeyError if user does not have membership in an org with a matching acronym + :return: formatted results, filtered to only the org identified by acronym + :rtype: dict + """ 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 + """Return the domains belonging to the organization identified by name - Arguments: - name -- string containing the name of an organization - client -- a GQL Client object + :param Client client: a GQL Client object + :param str name: string containing the full name of an organization + :return: formatted JSON data as string + :rtype: str """ slugified_name = slugify(name) # API expects a slugified string for name params = {"orgSlug": slugified_name} @@ -198,7 +234,12 @@ def get_domains_by_name(client, name): def format_name_domains(result): - """Formats the dict obtained by DOMAINS_BY_SLUG""" + """ 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"]] @@ -210,11 +251,12 @@ def format_name_domains(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 + :param Client client: a GQL Client object + :param str domain: domain name string + :param str month: string containing the full name of a month + :param int year: positive integer representing a year + :return: formatted JSON data as string + :rtype: str """ params = {"domain": domain, "month": month.upper(), "year": str(year)} @@ -227,7 +269,12 @@ def get_dmarc_summary(client, domain, month, year): def format_dmarc_monthly(result): - """Formats the dict obtained by DMARC_SUMMARY""" + """ 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 @@ -236,9 +283,10 @@ def format_dmarc_monthly(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 + :param Client client: a GQL Client object + :param str domain: domain name string + :return: formatted JSON data as string + :rtype: str """ params = {"domain": domain} @@ -251,7 +299,12 @@ def get_yearly_dmarc_summaries(client, domain): def format_dmarc_yearly(result): - """Formats the dict obtained by DMARC_YEARLY_SUMMARIES""" + """ 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 @@ -260,8 +313,9 @@ def format_dmarc_yearly(result): def get_all_summaries(client): """Returns summary metrics for all organizations you are a member of. - Arguments: - client -- a GQL Client object + :param Client client: a GQL Client object + :return: formatted JSON data with all organizations and their metrics as string + :rtype: str """ result = execute_query(client, ALL_ORG_SUMMARIES) @@ -272,7 +326,12 @@ def get_all_summaries(client): def format_all_summaries(result): - """Formats the dict obtained by ALL_ORG_SUMMARIES""" + """ 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 @@ -281,9 +340,10 @@ def format_all_summaries(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: string containing an acronym belonging to an organization + :return: formatted JSON data as string + :rtype: str """ # API doesn't allow query by acronym so we filter the get_all_summaries result result = execute_query(client, ALL_ORG_SUMMARIES) @@ -303,7 +363,14 @@ def get_summary_by_acronym(client, acronym): def format_acronym_summary(result, acronym): - """Formats the dict obtained by ALL_ORG_SUMMARIES to show only one org""" + """ Formats the result dict in get_summary_by_acronym + + :param dict result: unformatted dict with results of ALL_ORG_SUMMARIES + :param str acronym: string containing an acronym belonging to an organization + :raise: KeyError if user does not have membership in an org with a matching acronym + :return: formatted results, filtered to only the org identified by acronym + :rtype: dict + """ result = format_all_summaries(result) # dict in assignment is to keep the org identified in the return value result = {acronym.upper(): result[acronym.upper()]} @@ -313,9 +380,10 @@ def format_acronym_summary(result, acronym): 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 + :param Client client: a GQL Client object + :param str name: string containing the full name of an organization + :return: formatted JSON data as string + :rtype: str """ slugified_name = slugify(name) # API expects a slugified string for name params = {"orgSlug": slugified_name} @@ -329,7 +397,11 @@ def get_summary_by_name(client, name): def format_name_summary(result): - """Formats the dict obtained by SUMMARY_BY_SLUG""" + """ 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" @@ -341,9 +413,10 @@ def format_name_summary(result): def get_domain_results(client, domain): """Return scan results for a domain - Arguments: - domain -- domain name string - client -- a GQL Client object + :param Client client: a GQL Client object + :param str domain: domain name string + :return: formatted JSON data as string + :rtype: str """ params = {"domain": domain} @@ -356,7 +429,11 @@ def get_domain_results(client, domain): def format_domain_results(result): - """Format the dict obtained by DOMAIN_RESULTS""" + """ 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"] = { @@ -405,9 +482,10 @@ def format_domain_results(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 name string + :return: formatted JSON data as string + :rtype: str """ params = {"domain": domain} @@ -420,7 +498,11 @@ def get_domain_status(client, domain): def format_domain_status(result): - """Formats the dict obtained by DOMAIN_STATUS""" + """ 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 From d4ee4b802be3bcd037bad79b5e0663a263e57c08 Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Tue, 16 Feb 2021 10:43:16 -0400 Subject: [PATCH 06/11] revise function docstrings in client --- clients/python/tracker_client/client.py | 138 ++++++++++++------------ 1 file changed, 69 insertions(+), 69 deletions(-) diff --git a/clients/python/tracker_client/client.py b/clients/python/tracker_client/client.py index ac4a407074..8ce04d8091 100644 --- a/clients/python/tracker_client/client.py +++ b/clients/python/tracker_client/client.py @@ -31,11 +31,11 @@ def create_transport(url, auth_token=None): """Create and return a gql transport object :param str url: the Tracker GraphQL endpoint url - :param str auth_token: JWT auth token string, omit when initially obtaining the token (default is none) - :raise: ValueError if auth_token is not a valid JWT - :raise: TypeError if auth_token is not a string - :return: A gql transport + :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) @@ -62,7 +62,7 @@ def create_client(url="https://tracker.alpha.canada.ca/graphql", auth_token=None """Create and return a gql client object :param str url: the Tracker GraphQL endpoint url (default is "https://tracker.alpha.canada.ca/graphql") - :param str auth_token: JWT auth token string, omit when initially obtaining the token (default is none) + :param str auth_token: JWT auth token, omit when initially obtaining the token (default is None) :return: A gql client with AIOHTTPTransport :rtype: Client """ @@ -74,12 +74,12 @@ def create_client(url="https://tracker.alpha.canada.ca/graphql", auth_token=None def get_auth_token(url="https://tracker.alpha.canada.ca/graphql"): - """Get a token to use for authentication + """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 + :return: JWT auth token to allow access to Tracker :rtype: str """ client = create_client(url) @@ -100,16 +100,16 @@ 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. - - :param Client client: a gql client + + :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 - :raise: TransportProtocolError if server response is not GraphQL - :raise: TransportServerError if there is a server error - :raise: Exception if any unhandled exception is raised within function :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) @@ -135,10 +135,10 @@ 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 + """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 as string + :param Client client: a gql Client object + :return: formatted JSON data with all organizations and their domains :rtype: str """ result = execute_query(client, ALL_DOMAINS_QUERY) @@ -150,12 +150,12 @@ def get_all_domains(client): def format_all_domains(result): - """ Formats the result dict in get_all_domains + """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 @@ -174,11 +174,11 @@ def format_all_domains(result): def get_domains_by_acronym(client, acronym): - """Return the domains belonging to the organization identified by acronym + """Get the domains belonging to the organization identified by acronym - :param Client client: a GQL Client object - :param str acronym: string containing an acronym belonging to an organization - :return: formatted JSON data as string + :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 """ # API doesn't allow query by acronym so we filter the get_all_domains result @@ -201,25 +201,25 @@ def get_domains_by_acronym(client, acronym): def format_acronym_domains(result, acronym): - """ Formats the result dict in get_domains_by_acronym + """Formats the result dict in get_domains_by_acronym :param dict result: unformatted dict with results of ALL_DOMAINS_QUERY - :param str acronym: string containing an acronym belonging to an organization - :raise: KeyError if user does not have membership in an org with a matching acronym + :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 get_domains_by_name(client, name): - """Return the domains belonging to the organization identified by name + """Get the domains belonging to the organization identified by name - :param Client client: a GQL Client object - :param str name: string containing the full name of an organization - :return: formatted JSON data as string + :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 """ slugified_name = slugify(name) # API expects a slugified string for name @@ -234,12 +234,12 @@ def get_domains_by_name(client, name): def format_name_domains(result): - """ Formats the result dict in get_domains_by_name + """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"]] @@ -249,13 +249,13 @@ def format_name_domains(result): def get_dmarc_summary(client, domain, month, year): - """Return the DMARC summary for the specified domain and month + """Get the DMARC summary for the specified domain and month - :param Client client: a GQL Client object - :param str domain: domain name string - :param str month: string containing the full name of a 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 as string + :return: formatted JSON data with a DMARC summary :rtype: str """ params = {"domain": domain, "month": month.upper(), "year": str(year)} @@ -269,7 +269,7 @@ def get_dmarc_summary(client, domain, month, year): def format_dmarc_monthly(result): - """ Formats the result dict in get_dmarc_summary + """Formats the result dict in get_dmarc_summary :param dict result: unformatted dict with results of DMARC_SUMMARY :return: formatted results @@ -281,11 +281,11 @@ def format_dmarc_monthly(result): def get_yearly_dmarc_summaries(client, domain): - """Return yearly DMARC summaries for a domain + """Get yearly DMARC summaries for a domain - :param Client client: a GQL Client object - :param str domain: domain name string - :return: formatted JSON data as string + :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 """ params = {"domain": domain} @@ -299,7 +299,7 @@ def get_yearly_dmarc_summaries(client, domain): def format_dmarc_yearly(result): - """ Formats the result dict in get_yearly_dmarc_summaries + """Formats the result dict in get_yearly_dmarc_summaries :param dict result: unformatted dict with results of YEARLY_DMARC_SUMMARIES :return: formatted results @@ -311,10 +311,10 @@ def format_dmarc_yearly(result): def get_all_summaries(client): - """Returns summary metrics for all organizations you are a member of. + """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 as string + :param Client client: a gql Client object + :return: formatted JSON data with all organizations and their metrics :rtype: str """ result = execute_query(client, ALL_ORG_SUMMARIES) @@ -326,7 +326,7 @@ def get_all_summaries(client): def format_all_summaries(result): - """ Formats the result dict in get_all_summaries + """Formats the result dict in get_all_summaries :param dict result: unformatted dict with results of ALL_ORG_SUMMARIES :return: formatted results @@ -341,8 +341,8 @@ def get_summary_by_acronym(client, acronym): """Returns summary metrics for the organization identified by acronym :param Client client: a GQL Client object - :param str acronym: string containing an acronym belonging to an organization - :return: formatted JSON data as string + :param str acronym: an acronym referring to an organization + :return: formatted JSON with summary metrics for an organization :rtype: str """ # API doesn't allow query by acronym so we filter the get_all_summaries result @@ -363,14 +363,14 @@ def get_summary_by_acronym(client, acronym): def format_acronym_summary(result, acronym): - """ Formats the result dict in get_summary_by_acronym + """Formats the result dict in get_summary_by_acronym :param dict result: unformatted dict with results of ALL_ORG_SUMMARIES - :param str acronym: string containing an acronym belonging to an organization - :raise: KeyError if user does not have membership in an org with a matching acronym + :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()]} @@ -378,11 +378,11 @@ def format_acronym_summary(result, acronym): def get_summary_by_name(client, name): - """Return summary metrics for the organization identified by name + """Get summary metrics for the organization identified by name - :param Client client: a GQL Client object - :param str name: string containing the full name of an organization - :return: formatted JSON data as string + :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 """ slugified_name = slugify(name) # API expects a slugified string for name @@ -397,7 +397,7 @@ def get_summary_by_name(client, name): def format_name_summary(result): - """ Formats the result dict in get_summary_by_name + """Formats the result dict in get_summary_by_name :param dict result: unformatted dict with results of SUMMARY_BY_SLUG :return: formatted results @@ -411,11 +411,11 @@ def format_name_summary(result): def get_domain_results(client, domain): - """Return scan results for a domain + """Get scan results for a domain - :param Client client: a GQL Client object - :param str domain: domain name string - :return: formatted JSON data as string + :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 """ params = {"domain": domain} @@ -429,7 +429,7 @@ def get_domain_results(client, domain): def format_domain_results(result): - """ Formats the result dict in get_domain_results + """Formats the result dict in get_domain_results :param dict result: unformatted dict with results of DOMAIN_RESULTS :return: formatted results @@ -482,9 +482,9 @@ def format_domain_results(result): def get_domain_status(client, domain): """Return pass/fail status information for a domain - :param Client client: a GQL Client object - :param str domain: domain name string - :return: formatted JSON data as string + :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 """ params = {"domain": domain} @@ -498,7 +498,7 @@ def get_domain_status(client, domain): def format_domain_status(result): - """ Formats the result dict in get_domain_status + """Formats the result dict in get_domain_status :param dict result: unformatted dict with results of DOMAIN_STATUS :return: formatted results @@ -509,7 +509,7 @@ def format_domain_status(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" From 66c547d2fc70770f27ae0dadc21befe0aa6006be Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Tue, 16 Feb 2021 12:34:36 -0400 Subject: [PATCH 07/11] Module docstrings, document queries --- clients/python/tracker_client/client.py | 3 ++ clients/python/tracker_client/queries.py | 46 +++++++++++++++++++++++- 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/clients/python/tracker_client/client.py b/clients/python/tracker_client/client.py index 8ce04d8091..46195fb4f6 100644 --- a/clients/python/tracker_client/client.py +++ b/clients/python/tracker_client/client.py @@ -1,3 +1,5 @@ +"""Provides functions that get and format JSON data from Tracker (https://github.com/canada-ca/tracker)""" + import json import os import re @@ -25,6 +27,7 @@ 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): diff --git a/clients/python/tracker_client/queries.py b/clients/python/tracker_client/queries.py index bbb390c60a..ab9343fa4a 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 { @@ -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!) { From ed569ad521942a46b962032dd8f3f0eaaa5443cd Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Tue, 16 Feb 2021 12:59:48 -0400 Subject: [PATCH 08/11] More test docstrings --- clients/python/tests/test_formatting.py | 15 +++++++++++++++ clients/python/tests/test_get_dmarc.py | 1 + clients/python/tests/test_get_domains.py | 1 + clients/python/tests/test_get_results.py | 1 + clients/python/tests/test_get_summaries.py | 1 + clients/python/tests/test_gql.py | 1 + 6 files changed, 20 insertions(+) diff --git a/clients/python/tests/test_formatting.py b/clients/python/tests/test_formatting.py index d65903b0b9..eff79892cd 100644 --- a/clients/python/tests/test_formatting.py +++ b/clients/python/tests/test_formatting.py @@ -1,3 +1,7 @@ +"""Tests for query response formatting functions. + +The queries named in ALL_CAPS can be found in tracker_client/queries.py +""" from tracker_client.client import ( format_all_domains, format_acronym_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 From 9a178413f0d14b28b52856b22eb1b3e39dfaa5f5 Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Tue, 16 Feb 2021 14:08:41 -0400 Subject: [PATCH 09/11] Refactor formatting functions to own module This change is necessary to reduce the length of client.py in preparation for adding examples to docstrings. Otherwise it would become unmanageably long after examples are added. --- clients/python/tests/test_formatting.py | 2 +- clients/python/tracker_client/client.py | 194 ++------------------ clients/python/tracker_client/formatting.py | 180 ++++++++++++++++++ clients/python/tracker_client/queries.py | 4 +- 4 files changed, 197 insertions(+), 183 deletions(-) create mode 100644 clients/python/tracker_client/formatting.py diff --git a/clients/python/tests/test_formatting.py b/clients/python/tests/test_formatting.py index eff79892cd..24d18dd170 100644 --- a/clients/python/tests/test_formatting.py +++ b/clients/python/tests/test_formatting.py @@ -2,7 +2,7 @@ The queries named in ALL_CAPS can be found in tracker_client/queries.py """ -from tracker_client.client import ( +from tracker_client.formatting import ( format_all_domains, format_acronym_domains, format_name_domains, diff --git a/clients/python/tracker_client/client.py b/clients/python/tracker_client/client.py index 46195fb4f6..5c4fe0e74f 100644 --- a/clients/python/tracker_client/client.py +++ b/clients/python/tracker_client/client.py @@ -1,4 +1,4 @@ -"""Provides functions that get and format JSON data from Tracker (https://github.com/canada-ca/tracker)""" +"""Provides functions that get JSON data from Tracker (https://github.com/canada-ca/tracker)""" import json import os @@ -13,6 +13,7 @@ TransportProtocolError, ) +# TODO: decide if we should just import whole modules instead from queries import ( ALL_DOMAINS_QUERY, DOMAINS_BY_SLUG, @@ -24,6 +25,18 @@ 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-_.+/=]*$" @@ -152,30 +165,6 @@ def get_all_domains(client): return json.dumps(result, indent=4) -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 get_domains_by_acronym(client, acronym): """Get the domains belonging to the organization identified by acronym @@ -203,20 +192,6 @@ def get_domains_by_acronym(client, acronym): return json.dumps(result, indent=4) -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 get_domains_by_name(client, name): """Get the domains belonging to the organization identified by name @@ -236,21 +211,6 @@ def get_domains_by_name(client, name): return json.dumps(result, indent=4) -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 get_dmarc_summary(client, domain, month, year): """Get the DMARC summary for the specified domain and month @@ -271,18 +231,6 @@ def get_dmarc_summary(client, domain, month, year): return json.dumps(result, indent=4) -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 get_yearly_dmarc_summaries(client, domain): """Get yearly DMARC summaries for a domain @@ -301,18 +249,6 @@ def get_yearly_dmarc_summaries(client, domain): return json.dumps(result, indent=4) -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 get_all_summaries(client): """Get summary metrics for all organizations you are a member of. @@ -328,18 +264,6 @@ def get_all_summaries(client): return json.dumps(result, indent=4) -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 get_summary_by_acronym(client, acronym): """Returns summary metrics for the organization identified by acronym @@ -365,21 +289,6 @@ def get_summary_by_acronym(client, acronym): return json.dumps(result, indent=4) -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 get_summary_by_name(client, name): """Get summary metrics for the organization identified by name @@ -399,20 +308,6 @@ def get_summary_by_name(client, name): return json.dumps(result, indent=4) -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 get_domain_results(client, domain): """Get scan results for a domain @@ -431,57 +326,6 @@ def get_domain_results(client, domain): return json.dumps(result, indent=4) -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 get_domain_status(client, domain): """Return pass/fail status information for a domain @@ -500,16 +344,6 @@ def get_domain_status(client, domain): return json.dumps(result, indent=4) -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 - - def main(): # pragma: no cover """main() currently tries all implemented functions and prints results for diagnostic purposes and to demo available features. To be removed in future. 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 ab9343fa4a..53dbb62ae1 100644 --- a/clients/python/tracker_client/queries.py +++ b/clients/python/tracker_client/queries.py @@ -3,7 +3,7 @@ :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 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 @@ -47,7 +47,7 @@ domain } } - } + } } } } From 6ecabaa0e546c2bfdd5c74f8a0bce6f879acc668 Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Tue, 16 Feb 2021 15:16:13 -0400 Subject: [PATCH 10/11] Add examples for all get functions in client.py --- clients/python/tracker_client/client.py | 285 ++++++++++++++++++++++++ 1 file changed, 285 insertions(+) diff --git a/clients/python/tracker_client/client.py b/clients/python/tracker_client/client.py index 5c4fe0e74f..6f16fa7505 100644 --- a/clients/python/tracker_client/client.py +++ b/clients/python/tracker_client/client.py @@ -46,6 +46,8 @@ def create_transport(url, auth_token=None): """Create and return a gql transport object + 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 @@ -117,6 +119,9 @@ def get_auth_token(url="https://tracker.alpha.canada.ca/graphql"): def execute_query(client, query, params=None): """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 @@ -156,6 +161,23 @@ def get_all_domains(client): :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" @@ -172,6 +194,18 @@ def get_domains_by_acronym(client, acronym): :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) @@ -199,6 +233,18 @@ def get_domains_by_name(client, name): :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} @@ -220,6 +266,25 @@ def get_dmarc_summary(client, domain, month, year): :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)} @@ -238,6 +303,40 @@ def get_yearly_dmarc_summaries(client, domain): :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} @@ -255,6 +354,84 @@ def get_all_summaries(client): :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,6 +448,49 @@ def get_summary_by_acronym(client, acronym): :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) @@ -296,6 +516,49 @@ def get_summary_by_name(client, name): :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} @@ -315,6 +578,10 @@ def get_domain_results(client, domain): :param str domain: domain to get results for :return: formatted JSON data with scan results for the domain :rtype: str + + :Example: + + Coming soon, function likely to change """ params = {"domain": domain} @@ -333,6 +600,24 @@ def get_domain_status(client, domain): :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} From 76f24262d8272a24d6f58eb18149012f51f416e7 Mon Sep 17 00:00:00 2001 From: Thomas Nickerson <> Date: Tue, 16 Feb 2021 15:17:27 -0400 Subject: [PATCH 11/11] Run black on last commit --- clients/python/tracker_client/client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clients/python/tracker_client/client.py b/clients/python/tracker_client/client.py index 6f16fa7505..a3827a864b 100644 --- a/clients/python/tracker_client/client.py +++ b/clients/python/tracker_client/client.py @@ -336,7 +336,7 @@ def get_yearly_dmarc_summaries(client, domain): } }, ] - } + } """ params = {"domain": domain}