Repository: salesforce/policy_sentry (python)

Issue:
How does one list all the services?
## Problem

I was having issues finding the service name for amazon health events (`health`) and privateca (`acm-pca`) and it would be nice to do query a service-table that could match on the service name, description, etc

## Proposal

```bash
✗ policy_sentry query service-table | grep -i health
health | AWS Health APIs and Notifications
```

## Workaround

My workaround which works if I can guess a portion of the name like `health`. 

```bash
✗ policy_sentry initialize --fetch
```

```bash
✗ jq 'keys' ~/.policy_sentry/iam-definition.json | grep health
  "health",
  "healthlake",
```

This is harder with `privateca` where my guesses would be difficult unless I searched for `ca` unless there was some kind of description associated with each service.

```bash
✗ jq 'keys' ~/.policy_sentry/iam-definition.json | grep ca
  "acm-pca",
  "application-autoscaling",
  "application-cost-profiler",
  "application-transformation",
  "applicationinsights",
  "autoscaling",
  "autoscaling-plans",
  "cases",
  "cassandra",
  "codecatalyst",
  "codestar-notifications",
  "comprehendmedical",
  "connect-campaigns",
  "controlcatalog",
  "customer-verification",
  "elasticache",
  "elemental-support-cases",
  "forecast",
  "inspector-scan",
  "mechanicalturk",
  "medical-imaging",
  "notifications",
  "notifications-contacts",
  "pca-connector-ad",
  "servicecatalog",
```

Some extra parsing and we can do it using the `.service_name`

```bash
✗ jq 'del(.policy_sentry_schema_version) | to_entries | .[].value | {name: .service_name, prefix: .prefix}' ~/.policy_sentry/iam-definition.json | grep -i private -B1 -A1
{
  "name": "AWS Certificate Manager Private Certificate Authority",
  "prefix": "acm-pca"
--
{
  "name": "AWS Private CA Connector for Active Directory",
  "prefix": "pca-connector-ad"
--
{
  "name": "AWS re:Post Private",
  "prefix": "repostspace"
--
{
  "name": "AWS service providing managed private networks",
  "prefix": "private-networks"
}
```

Thank you

Pull request:
add query command for service data
## What does this PR do?

- this adds a new `query` sub-command `service-table`
- also added for this specific output a `csv` format variant

Fixes #475

ex.
```
$ policy_sentry query service-table --fmt csv

prefix,service_name
a4b,Alexa for Business
execute-api,Amazon API Gateway
apigateway,Amazon API Gateway Management
...
```

## What gif best describes this PR or how it makes you feel?

![](https://i.giphy.com/media/v1.Y2lkPTc5MGI3NjExYjZpem13MWUwNmVycHQ0N2tvaXZrcWVpZXRibDM5emg5N3Y2dXY1bCZlcD12MV9pbnRlcm5hbF9naWZfYnlfaWQmY3Q9Zw/qYytk8KxBPA1q/giphy.gif)

## Completion checklist

- [x] Additions and changes have unit tests
- [x] [Unit tests, Pylint, security testing, and Integration tests are passing.](https://github.com/salesforce/policy_sentry/actions). GitHub actions does this automatically
- [x] The pull request has been appropriately labeled using the provided PR labels


Patch:
diff --git a/policy_sentry/command/query.py b/policy_sentry/command/query.py
index e2735e0d..76a7f092 100644
--- a/policy_sentry/command/query.py
+++ b/policy_sentry/command/query.py
@@ -11,6 +11,8 @@ from typing import Any
 
 import click
 import yaml
+
+from policy_sentry.querying.services import get_services_data
 from policy_sentry.util.access_levels import transform_access_level_text
 from policy_sentry.querying.all import get_all_service_prefixes
 from policy_sentry.querying.arns import (
@@ -47,8 +49,17 @@ def print_list(output: Any, fmt: str = "json") -> None:
 
 
 def print_dict(output: Any, fmt: str = "json") -> None:
-    """Common method on how to print a dict, depending on whether the user requests JSON or YAML output"""
-    print(yaml.dump(output)) if fmt == "yaml" else [print(json.dumps(output, indent=4))]
+    """Common method on how to print a dict, depending on whether the user requests JSON, YAML or CSV output"""
+    if fmt == "csv":
+        if not output:
+            return None
+        print(",".join(output[0].keys()))
+        for entry in output:
+            print(",".join(entry.values()))
+    elif fmt == "json":
+        print(json.dumps(output, indent=4))
+    elif fmt == "yaml":
+        print(yaml.dump(output))
 
 
 @click.group()
@@ -97,7 +108,7 @@ def query() -> None:
     type=click.Choice(["yaml", "json"]),
     default="json",
     required=False,
-    help='Format output as YAML or JSON. Defaults to "yaml"',
+    help='Format output as YAML or JSON. Defaults to "json"',
 )
 @click.option(
     "--verbose",
@@ -229,7 +240,7 @@ def query_action_table(
     type=click.Choice(["yaml", "json"]),
     default="json",
     required=False,
-    help='Format output as YAML or JSON. Defaults to "yaml"',
+    help='Format output as YAML or JSON. Defaults to "json"',
 )
 @click.option(
     "--verbose",
@@ -296,7 +307,7 @@ def query_arn_table(
     type=click.Choice(["yaml", "json"]),
     default="json",
     required=False,
-    help='Format output as YAML or JSON. Defaults to "yaml"',
+    help='Format output as YAML or JSON. Defaults to "json"',
 )
 @click.option(
     "--verbose",
@@ -334,3 +345,42 @@ def query_condition_table(
         output = get_condition_key_details(service, name)
         print_dict(output=output, fmt=fmt)
     return output
+
+
+@query.command(short_help="Query the service table.")
+@click.option(
+    "--fmt",
+    type=click.Choice(["yaml", "json", "csv"]),
+    default="json",
+    required=False,
+    help='Format output as YAML, JSON or CSV. Defaults to "json"',
+)
+@click.option(
+    "--verbose",
+    "-v",
+    type=click.Choice(
+        ["critical", "error", "warning", "info", "debug"], case_sensitive=False
+    ),
+)
+def service_table(fmt: str, verbose: str | None) -> None:
+    """Query the service table from the Policy Sentry database"""
+    if verbose:
+        log_level = getattr(logging, verbose.upper())
+        set_stream_logger(level=log_level)
+    query_service_table(fmt)
+
+
+def query_service_table(fmt: str = "json") -> list[dict[str, str]]:
+    """Query the service table from the Policy Sentry database.
+    Use this one when leveraging Policy Sentry as a library."""
+    if os.path.exists(LOCAL_DATASTORE_FILE_PATH):
+        logger.info(
+            f"Using the Local IAM definition: {LOCAL_DATASTORE_FILE_PATH}. To leverage the bundled definition instead, remove the folder $HOME/.policy_sentry/"
+        )
+    else:
+        # Otherwise, leverage the datastore inside the python package
+        logger.debug("Leveraging the bundled IAM Definition.")
+
+    output = get_services_data()
+    print_dict(output=output, fmt=fmt)
+    return output
diff --git a/policy_sentry/querying/services.py b/policy_sentry/querying/services.py
new file mode 100644
index 00000000..a78235f6
--- /dev/null
+++ b/policy_sentry/querying/services.py
@@ -0,0 +1,19 @@
+"""
+Methods that execute specific queries against the SQLite database for the SERIVCES table.
+This supports the policy_sentry query functionality
+"""
+
+from __future__ import annotations
+
+from policy_sentry.shared.iam_data import iam_definition
+
+
+def get_services_data() -> list[dict[str, str]]:
+    return [
+        {
+            "prefix": service_prefix,
+            "service_name": data["service_name"],
+        }
+        for service_prefix, data in iam_definition.items()
+        if isinstance(data, dict)
+    ]


Test patch:
diff --git a/test/querying/test_query_services.py b/test/querying/test_query_services.py
new file mode 100644
index 00000000..8bcadf6a
--- /dev/null
+++ b/test/querying/test_query_services.py
@@ -0,0 +1,16 @@
+import unittest
+
+from policy_sentry.querying.services import get_services_data
+
+
+class QueryServicesTestCase(unittest.TestCase):
+    def test_get_services_data(self):
+        # when
+        results = get_services_data()
+
+        # then
+        self.assertGreater(len(results), 400)  # in 07/24 it was 405
+
+        # both should be a non-empty string
+        self.assertTrue(results[0]["prefix"])
+        self.assertTrue(results[0]["service_name"])