Adds new licensing table (#1072)

* Added license table

* fixing linter

* Removes hardcoding

* Added prerequisites api

* Updated documentation

* Updates based on PR comments

* Updated failure logic and progress bar

* fixing linter

* fix smoke tests

README.md

@@ -32,6 +32,7 @@ We use a three-step process:
 ### Prerequisites
 
 - [Permissions](https://github.com/cisagov/ScubaGoggles/blob/main/docs/prerequisites/Prerequisites.md#permissions)
+- [Google Cloud APIs](https://github.com/cisagov/ScubaGoggles/blob/main/docs/prerequisites/Prerequisites.md#google-cloud-apis)
 - [Create a Project](https://github.com/cisagov/ScubaGoggles/blob/main/docs/prerequisites/Prerequisites.md#create-a-project)
 
 ### Authentication

docs/authentication/OAuth.md

@@ -37,7 +37,8 @@ Only complete this section if not authenticating via [Service Account](ServiceAc
 24. Search for and enable the **Admin SDK API**
 25. Search for and enable the **Groups Settings API**
 26. Search for and enable the **Cloud Identity** API
-27. During the first run of this tool your default web browser will open up a page to consent to the API scopes needed to run this tool. Sign in
+27. Search for and enable the **Enterprise License Manager API**
+28. During the first run of this tool your default web browser will open up a page to consent to the API scopes needed to run this tool. Sign in
     with an account with the necessary privileges and click allow.
 
 ## Add the Oauth App to the allowlist

docs/authentication/ServiceAccount.md

@@ -26,6 +26,7 @@ Only complete this section if not authenticating via [OAuth](OAuth.md). See [Aut
 1. On the toolbar, click **+ Enable APIs & Services**
 1. Search for and enable the **Admin SDK API**
 1. Search for and enable the **Groups Settings API**
+1. Search for and enable the **Enterprise License Manager API**
 1. Search for and enable the **Cloud Identity** API
 1. Finally, run ScubaGoggles with the `--subjectemail` option set to the email of a Workspace admin user with necessary permissions to run ScubaGoggles.  Do **not** use the service account email you created in a previous step.
 

docs/prerequisites/Prerequisites.md

@@ -16,11 +16,24 @@ https://www.googleapis.com/auth/admin.directory.rolemanagement.readonly,
 https://www.googleapis.com/auth/admin.directory.customer.readonly,
 https://www.googleapis.com/auth/cloud-identity.policies.readonly,
 https://www.googleapis.com/auth/cloud-identity.inboundsso.readonly,
+https://www.googleapis.com/auth/apps.licensing
 https://www.googleapis.com/auth/apps.groups.settings
 ```
 
 When running ScubaGoggles for the first time you will be prompted to consent to
-these API scopes. 
+these API scopes.
+
+## Google Cloud APIs
+
+In addition to consenting to the scopes above, the following APIs must be
+enabled in your Google Cloud project (see [Using OAuth](../authentication/OAuth.md)
+or [Using a Service Account](../authentication/ServiceAccount.md) for steps):
+
+- **Admin SDK API**
+- **Groups Settings API**
+- **Cloud Identity API**
+- **Enterprise License Manager API** (required for license and subscription data
+  in the Common Controls report)
 
 Additionally, because the Groups Settings API does not natively support a read-only API scope, we use the Groups Reader role with Delegated Admin Service Account (DASA) authorization to achieve least privilege and address the risks associated with the update operations available through the Groups Settings API.
 

docs/troubleshooting/Troubleshooting.md

@@ -10,7 +10,7 @@ RuntimeWarning: An exception was thrown trying to get the tenant data:
 ```
 Ensure that you consented to the following API scopes as a user with the proper
 [permissions to consent](../prerequisites/Prerequisites.md#permissions) and have
-enabled the required [APIs and Services](../authentication/OAuth.md).
+enabled the required [APIs and Services](../prerequisites/Prerequisites.md#google-cloud-apis).
 
 ## macOS: Certificate Verification Error
 

scubagoggles/Testing/Functional/SmokeTests/smoke_test_utils.py

@@ -246,6 +246,11 @@ def run_selenium(browser, customerdomain):
             # Verify indicators and legend are present
             verify_indicators_and_legend(browser)
 
+            if product == 'Common Controls':
+                verify_tenant_licensing_table(browser)
+            else:
+                verify_tenant_licensing_table_absent(browser)
+
             policy_tables = browser.find_elements(By.CSS_SELECTOR, "table:not(.dns-logs table)")
             for table in policy_tables[1:]:
 
@@ -467,3 +472,58 @@ def verify_indicators_and_legend(browser):
     except Exception as e:
         raise AssertionError(
             f'Indicator verification in requirements failed: {e}') from e
+
+
+def verify_tenant_licensing_table(browser):
+    """
+    Verify the Common Controls report includes the tenant licensing section.
+
+    The section should contain either a populated license table or a message
+    indicating that no licenses were found or that collection failed.
+
+    Args:
+        browser: A Selenium WebDriver instance
+    """
+    heading = browser.find_element(By.ID, 'subscriptions')
+    assert heading.text == 'Tenant Licensing Information'
+
+    main = browser.find_element(By.TAG_NAME, 'main')
+    license_table_found = False
+
+    for table in main.find_elements(By.TAG_NAME, 'table'):
+        theads = table.find_elements(By.TAG_NAME, 'thead')
+        if not theads:
+            continue
+        headers = theads[0].find_elements(By.TAG_NAME, 'th')
+        header_text = [header.text.strip() for header in headers]
+        if header_text == ['Product Name', 'Status', 'Assigned Licenses']:
+            license_table_found = True
+            tbody = table.find_element(By.TAG_NAME, 'tbody')
+            rows = tbody.find_elements(By.TAG_NAME, 'tr')
+            assert len(rows) > 0, 'License table should contain at least one row'
+            break
+
+    status_messages = main.find_elements(
+        By.XPATH,
+        (
+            "//p[contains(., 'No licenses found.') or "
+            "contains(., 'An error occurred when collecting license information.')]"
+        ),
+    )
+
+    assert license_table_found or status_messages, (
+        'Tenant licensing section should include a license table or status message'
+    )
+
+
+def verify_tenant_licensing_table_absent(browser):
+    """
+    Verify non-Common Controls reports do not include the tenant licensing section.
+
+    Args:
+        browser: A Selenium WebDriver instance
+    """
+    headings = browser.find_elements(By.ID, 'subscriptions')
+    assert not headings, (
+        'Tenant licensing section should only appear on the Common Controls report'
+    )

scubagoggles/Testing/Unit/Python/provider/license_cases.py

@@ -0,0 +1,188 @@
+"""
+Parametrized test case data for license data Provider methods.
+"""
+
+GET_LICENSE_DATA_CASES = [
+    # Aggregates SKU assignment counts across a single product response.
+    {
+        "product_ids": ["Google-Apps"],
+        "domains": [{"domainName": "example.com", "isPrimary": True}],
+        "product_responses": {
+            "Google-Apps": [
+                {
+                    "items": [
+                        {
+                            "skuId": "sku-1",
+                            "skuName": "Google Workspace Business Starter",
+                        },
+                        {
+                            "skuId": "sku-1",
+                            "skuName": "Google Workspace Business Starter",
+                        },
+                        {
+                            "skuId": "sku-2",
+                            "skuName": "Google Workspace Enterprise Plus",
+                        },
+                    ]
+                }
+            ]
+        },
+        "expected": {
+            "license_data": [
+                {
+                    "product_name": "Google Workspace Business Starter",
+                    "sku_id": "sku-1",
+                    "product_id": "Google-Apps",
+                    "status": "Active",
+                    "assigned": 2,
+                },
+                {
+                    "product_name": "Google Workspace Enterprise Plus",
+                    "sku_id": "sku-2",
+                    "product_id": "Google-Apps",
+                    "status": "Active",
+                    "assigned": 1,
+                },
+            ]
+        },
+        "expected_customer_id": "example.com",
+        "expect_success_call": True,
+        "expect_warning": False,
+    },
+    # Handles paginated license assignment responses.
+    {
+        "product_ids": ["Google-Apps"],
+        "domains": [{"domainName": "example.com", "isPrimary": True}],
+        "product_responses": {
+            "Google-Apps": [
+                {
+                    "items": [
+                        {
+                            "skuId": "sku-1",
+                            "skuName": "Google Workspace Business Starter",
+                        }
+                    ],
+                    "nextPageToken": "page-2",
+                },
+                {
+                    "items": [
+                        {
+                            "skuId": "sku-1",
+                            "skuName": "Google Workspace Business Starter",
+                        }
+                    ]
+                },
+            ]
+        },
+        "expected": {
+            "license_data": [
+                {
+                    "product_name": "Google Workspace Business Starter",
+                    "sku_id": "sku-1",
+                    "product_id": "Google-Apps",
+                    "status": "Active",
+                    "assigned": 2,
+                }
+            ]
+        },
+        "expected_customer_id": "example.com",
+        "expect_success_call": True,
+        "expect_warning": False,
+    },
+    # Successful API call with no assigned licenses.
+    {
+        "product_ids": ["Google-Apps"],
+        "domains": [{"domainName": "example.com", "isPrimary": True}],
+        "product_responses": {
+            "Google-Apps": [{"items": []}]
+        },
+        "expected": {"license_data": []},
+        "expected_customer_id": "example.com",
+        "expect_success_call": True,
+        "expect_warning": False,
+    },
+    # Falls back to the customer id when no primary domain is present.
+    {
+        "product_ids": ["Google-Apps"],
+        "domains": [{"domainName": "alias.example.com", "isPrimary": False}],
+        "product_responses": {
+            "Google-Apps": [{"items": []}]
+        },
+        "expected": {"license_data": []},
+        "expected_customer_id": "test_customer",
+        "expect_success_call": True,
+        "expect_warning": False,
+    },
+    # One product succeeds while another fails.
+    {
+        "product_ids": ["Google-Apps", "101047"],
+        "domains": [{"domainName": "example.com", "isPrimary": True}],
+        "product_responses": {
+            "Google-Apps": "raises",
+            "101047": [
+                {
+                    "items": [
+                        {
+                            "skuId": "gemini-sku",
+                            "skuName": "Gemini Enterprise",
+                        }
+                    ]
+                }
+            ],
+        },
+        "expected": {
+            "license_data": [
+                {
+                    "product_name": "Gemini Enterprise",
+                    "sku_id": "gemini-sku",
+                    "product_id": "101047",
+                    "status": "Active",
+                    "assigned": 1,
+                }
+            ]
+        },
+        "expected_customer_id": "example.com",
+        "expect_success_call": True,
+        "expect_warning": False,
+        "expect_product_warning": True,
+    },
+    # All product calls fail.
+    {
+        "product_ids": ["Google-Apps", "101047"],
+        "domains": [{"domainName": "example.com", "isPrimary": True}],
+        "product_responses": {
+            "Google-Apps": "raises",
+            "101047": "raises",
+        },
+        "expected": {"license_data": []},
+        "expected_customer_id": "example.com",
+        "expect_success_call": False,
+        "expect_warning": False,
+        "expect_product_warning": True,
+    },
+    # Licensing service build fails.
+    {
+        "product_ids": ["Google-Apps"],
+        "domains": [{"domainName": "example.com", "isPrimary": True}],
+        "product_responses": {},
+        "build_raises": Exception("licensing unavailable"),
+        "expected": {"license_data": []},
+        "expected_customer_id": "example.com",
+        "expect_success_call": False,
+        "expect_warning": True,
+    },
+    # Global API failure stops after the first request.
+    {
+        "product_ids": ["Google-Apps", "101047", "101001"],
+        "domains": [{"domainName": "example.com", "isPrimary": True}],
+        "product_responses": {
+            "Google-Apps": "raises_global",
+        },
+        "expected": {"license_data": []},
+        "expected_customer_id": "example.com",
+        "expect_success_call": False,
+        "expect_warning": False,
+        "expect_product_warning": True,
+        "expected_api_calls": 1,
+    },
+]

scubagoggles/Testing/Unit/Python/provider/test_provider.py

@@ -1,8 +1,10 @@
+# pylint: disable=too-many-lines
 """
 test_provider tests the Provider class.
 """
 import pytest
 from google.auth.exceptions import RefreshError
+from googleapiclient.errors import HttpError
 from scubagoggles.provider import Provider
 from scubagoggles.scuba_constants import ApiReference
 
@@ -31,6 +33,9 @@ from scubagoggles.Testing.Unit.Python.provider.group_cases import (
 from scubagoggles.Testing.Unit.Python.provider.get_list_cases import (
     GET_LIST_CASES,
 )
+from scubagoggles.Testing.Unit.Python.provider.license_cases import (
+    GET_LICENSE_DATA_CASES,
+)
 
 # Disable "protected-access" because we test some internal methods
 # pylint: disable=protected-access
@@ -752,6 +757,148 @@ class TestProvider:
             customer="test_customer"
         )
 
+    @staticmethod
+    def _setup_license_data_mock(mocker, provider, cases):
+        """
+        Configure licensing API mocks for get_license_data() tests.
+
+        :param mocker: pytest-mock fixture used to create mocks/patch functions.
+        :param provider: Provider instance under test.
+        :param cases: Parametrized test case metadata.
+        """
+        mocker.patch.object(provider, "list_domains", return_value=cases["domains"])
+        mocker.patch(
+            "scubagoggles.provider.KNOWN_PRODUCT_IDS",
+            cases["product_ids"],
+        )
+
+        license_assignments = mocker.Mock(name="license_assignments")
+        licensing_service = mocker.Mock(name="licensing_service")
+        licensing_service.licenseAssignments.return_value = license_assignments
+
+        product_pages = cases.get("product_responses", {})
+
+        def list_for_product(**kwargs):
+            product_id = kwargs["productId"]
+            pages = product_pages[product_id]
+            if pages == "raises":
+                raise Exception(f"{product_id} unavailable")
+            if pages == "raises_global":
+                response = mocker.Mock()
+                response.status = 403
+                response.reason = (
+                    'Enterprise License Manager API has not been used in project '
+                    '123 before or it is disabled.'
+                )
+                raise HttpError(response, b'accessNotConfigured')
+
+            if kwargs.get("pageToken"):
+                page = pages[1]
+            else:
+                page = pages[0]
+
+            request = mocker.Mock(name=f"license_request_{product_id}")
+            request.execute.return_value = page
+            return request
+
+        license_assignments.listForProduct.side_effect = list_for_product
+
+        if cases.get("build_raises") is not None:
+            mocker.patch(
+                "scubagoggles.provider.build",
+                side_effect=cases["build_raises"],
+            )
+        else:
+            def build_side_effect(service_name, version, **_kwargs):
+                if service_name == "licensing" and version == "v1":
+                    return licensing_service
+                return mocker.Mock(name=f"{service_name}_{version}")
+
+            mocker.patch("scubagoggles.provider.build", side_effect=build_side_effect)
+
+        return license_assignments
+
+    @pytest.mark.parametrize(
+        "cases",
+        GET_LICENSE_DATA_CASES
+    )
+    @pytest.mark.usefixtures("mock_build")
+    def test_get_license_data(self, mocker, cases):
+        """
+        Verifies Provider.get_license_data() aggregates license assignments
+        per SKU, handles pagination and partial product failures, and records
+        successful/unsuccessful ApiReference calls.
+
+        :param mocker: pytest-mock fixture used to create mocks/patch functions.
+        :param cases: Parametrized test cases containing license API metadata.
+        """
+        provider = self._provider()
+        provider._successful_calls.clear()
+        provider._unsuccessful_calls.clear()
+
+        license_assignments = self._setup_license_data_mock(mocker, provider, cases)
+
+        if cases.get("expect_warning"):
+            with pytest.warns(
+                RuntimeWarning,
+                match="Exception thrown while getting license data; "
+                      "subscription table will be omitted"
+            ):
+                result = provider.get_license_data(quiet=True)
+        elif cases.get("expect_product_warning"):
+            with pytest.warns(
+                RuntimeWarning,
+                match="Exception thrown while getting license data for "
+            ):
+                result = provider.get_license_data(quiet=True)
+        else:
+            result = provider.get_license_data(quiet=True)
+
+        assert result == cases["expected"]
+
+        api_call = ApiReference.LIST_LICENSE_ASSIGNMENTS.value
+        if cases["expect_success_call"]:
+            assert api_call in provider._successful_calls
+            assert api_call not in provider._unsuccessful_calls
+        else:
+            assert api_call not in provider._successful_calls
+            assert api_call in provider._unsuccessful_calls
+
+        if cases.get("build_raises") is None and cases["product_responses"]:
+            license_assignments.listForProduct.assert_any_call(
+                productId=cases["product_ids"][0],
+                customerId=cases["expected_customer_id"],
+                maxResults=1000,
+            )
+            if "expected_api_calls" in cases:
+                assert (
+                    license_assignments.listForProduct.call_count
+                    == cases["expected_api_calls"]
+                )
+
+    @pytest.mark.usefixtures("mock_build")
+    def test_is_global_license_failure(self, mocker):
+        """Verifies systemic license API failures are detected for fail-fast."""
+        response = mocker.Mock()
+        response.status = 403
+        response.reason = 'Forbidden'
+        content = (
+            b'{"error":{"errors":[{"reason":"accessNotConfigured"}],'
+            b'"message":"Enable licensing.googleapis.com"}}'
+        )
+        assert Provider._is_global_license_failure(HttpError(response, content))
+
+        response.reason = (
+            'Enterprise License Manager API has not been used in project '
+            '123 before or it is disabled.'
+        )
+        assert Provider._is_global_license_failure(HttpError(response, b''))
+
+        response.status = 404
+        assert not Provider._is_global_license_failure(
+            HttpError(response, b'Not Found')
+        )
+
     @pytest.mark.parametrize(
         "cases",
         GET_LIST_CASES

scubagoggles/orchestrator.py

@@ -1,3 +1,4 @@
+# pylint: disable=too-many-lines
 """
 orchestrator.py is the main module that starts and handles the output of the
 provider, rego, and report modules of the SCuBA tool
@@ -691,6 +692,11 @@ class Orchestrator:
         tenant_name = tenant_info['topLevelOU']
         successful_calls = set(settings_data['successful_calls'])
         unsuccessful_calls = set(settings_data['unsuccessful_calls'])
+        cc_license_data = (
+            settings_data.get('license_data', [])
+            if 'commoncontrols' in baselines
+            else None
+        )
         missing_policies = set(settings_data['missing_policies'])
         report_uuid = settings_data['report_uuid']
 
@@ -771,7 +777,12 @@ class Orchestrator:
                                 dns_logs,
                                 omissions,
                                 annotations,
-                                products_bar)
+                                products_bar,
+                                license_data=(
+                                    cc_license_data
+                                    if product == 'commoncontrols'
+                                    else None
+                                ))
             stats_and_data[product] = \
                 reporter.rego_json_to_ind_reports(test_results_data,
                                                   outputpath,

scubagoggles/provider.py

@@ -5,11 +5,13 @@ provider.py is where the GWS api calls are made.
 
 import logging
 import warnings
+import json
 from typing import Callable, ContextManager, Mapping, Optional, Protocol
 from pathlib import Path
 from tqdm import tqdm
 
 from googleapiclient.discovery import build
+from googleapiclient.errors import HttpError
 from google.auth.exceptions import RefreshError
 import google.auth.transport.requests as auth_requests
 from scubagoggles.auth import GwsAuth
@@ -111,6 +113,23 @@ EVENTS = {
 
 SELECTORS = ['google', 'selector1', 'selector2']
 
+# Top-level Google Workspace product IDs queried by get_license_data().
+# The API returns skuId and skuName per user, so individual SKUs never need
+# to be listed here.  New SKUs within these products are discovered automatically.
+# Source: https://developers.google.com/workspace/admin/licensing/v1/how-tos/products
+KNOWN_PRODUCT_IDS = [
+    'Google-Apps',   # Core Workspace editions (Business, Enterprise, Frontline…)
+    '101047',        # Gemini add-ons
+    '101001',        # Cloud Identity Free
+    '101005',        # Cloud Identity Premium
+    '101039',        # Assured Controls / Assured Controls Plus
+    'Google-Vault',  # Google Vault
+    '101033',        # Google Voice
+    '101040',        # Chrome Enterprise Premium
+    '101038',        # AppSheet
+    '101035',        # Cloud Search
+]
+
 # Privilege names (the values of role.rolePrivileges[*].privilegeName
 # returned by directory/v1/roles/list) that indicate a "highly privileged"
 # admin role for the purposes of GWS.COMMONCONTROLS.6.  Any custom role
@@ -203,6 +222,7 @@ class Provider:
         self._successful_calls = set()
         self._unsuccessful_calls = set()
         self._missing_policies = set()
+        self._deferred_license_warning = None
         self._dns_client = RobustDNSClient(dns_resolvers, doh_servers, skip_doh)
         self._domains = []
         self._alias_domains = []
@@ -912,6 +932,198 @@ class Provider:
             self._unsuccessful_calls.add(ApiReference.GET_GROUP.value)
             return {'group_settings': []}
 
+    @staticmethod
+    def _get_license_error_cause(exc: Exception) -> str:
+        if isinstance(exc, HttpError):
+            return f'HTTP {exc.status_code}: {exc.reason}'
+        return str(exc)
+
+    @staticmethod
+    def _is_global_license_failure(exc: Exception) -> bool:
+        """Return True when a license API error will affect every product query."""
+        is_global_failure = False
+
+        if isinstance(exc, HttpError):
+            if exc.status_code == 401:
+                is_global_failure = True
+            elif exc.status_code == 403:
+                error_text = f'{exc.reason} {exc.content}'.lower()
+                is_global_failure = (
+                    'accessnotconfigured' in error_text
+                    or (
+                        'has not been used in project' in error_text
+                        and 'disabled' in error_text
+                    )
+                    or 'licensing.googleapis.com' in error_text
+                )
+
+                if not is_global_failure:
+                    try:
+                        content = (
+                            exc.content.decode()
+                            if isinstance(exc.content, bytes)
+                            else str(exc.content)
+                        )
+                        payload = json.loads(content)
+                        is_global_failure = any(
+                            error.get('reason', '').lower() == 'accessnotconfigured'
+                            for error in payload.get('error', {}).get('errors', [])
+                        )
+                    except (ValueError, AttributeError, TypeError):
+                        pass
+
+        return is_global_failure
+
+    def get_license_data(self, quiet: bool = False, status_bar=None) -> dict:  # pylint: disable=too-many-branches
+        """
+        Gets license assignment counts per SKU using the Enterprise License
+        Manager API.
+
+        Calls listForProduct once per product ID in KNOWN_PRODUCT_IDS.  Each
+        response item already contains skuId and skuName, so no hardcoded SKU
+        mapping is needed — new SKUs are discovered automatically.
+
+        Scope required: https://www.googleapis.com/auth/apps.licensing
+
+        Returns a dict with key 'license_data': a list of subscription records,
+        one per active SKU, each with: product_name, sku_id, product_id,
+        status, assigned.
+        """
+        api_call = ApiReference.LIST_LICENSE_ASSIGNMENTS.value
+        subscriptions = []
+        self._deferred_license_warning = None
+
+        primary_domain = None
+        for domain in self.list_domains():
+            if domain.get('isPrimary'):
+                primary_domain = domain['domainName']
+                break
+        if not primary_domain:
+            primary_domain = self._customer_id
+
+        # googleapiclient logs every non-2xx response as a WARNING via the
+        # standard logging module regardless of how we handle the error, so
+        # silence it here since we already report failures via our own
+        # consolidated warning below.
+        http_logger = logging.getLogger('googleapiclient.http')
+        previous_log_level = http_logger.level
+        http_logger.setLevel(logging.ERROR)
+        try:
+            licensing_service = build('licensing', 'v1',
+                                      cache_discovery=False,
+                                      credentials=self._credentials,
+                                      num_retries=0)
+
+            any_success = False
+            failed_products: dict[str, list[str]] = {}
+            use_inner_bar = status_bar is None
+            products_bar = tqdm(
+                list(enumerate(KNOWN_PRODUCT_IDS)),
+                total=len(KNOWN_PRODUCT_IDS),
+                leave=False,
+                disable=quiet or not use_inner_bar,
+            )
+            for index, product_id in products_bar:
+                description = (
+                    'Running Provider: Exporting license data for '
+                    f'{product_id} (commoncontrols)...'
+                )
+                if status_bar is not None:
+                    status_bar.set_description(description)
+                else:
+                    products_bar.set_description(description)
+                try:
+                    sku_counts: dict[str, dict] = {}
+                    page_token = None
+                    while True:
+                        kwargs = {
+                            'productId': product_id,
+                            'customerId': primary_domain,
+                            'maxResults': 1000,
+                        }
+                        if page_token:
+                            kwargs['pageToken'] = page_token
+                        response = (
+                            licensing_service  # pylint: disable=no-member
+                            .licenseAssignments()
+                            .listForProduct(**kwargs)
+                            .execute()
+                        )
+                        for item in response.get('items', []):
+                            sku_id = item.get('skuId', '')
+                            if sku_id not in sku_counts:
+                                sku_counts[sku_id] = {
+                                    'name': item.get('skuName', sku_id),
+                                    'count': 0,
+                                }
+                            sku_counts[sku_id]['count'] += 1
+                        page_token = response.get('nextPageToken')
+                        if not page_token:
+                            break
+
+                    any_success = True
+                    for sku_id, info in sku_counts.items():
+                        if info['count'] > 0:
+                            subscriptions.append({
+                                'product_name': info['name'],
+                                'sku_id': sku_id,
+                                'product_id': product_id,
+                                'status': 'Active',
+                                'assigned': info['count'],
+                            })
+                except Exception as exc:
+                    cause = self._get_license_error_cause(exc)
+                    failed_products.setdefault(cause, []).append(product_id)
+                    if self._is_global_license_failure(exc):
+                        failed_products[cause].extend(
+                            KNOWN_PRODUCT_IDS[index + 1:]
+                        )
+                        skip_description = (
+                            'Running Provider: License API unavailable; '
+                            'skipping remaining products (commoncontrols)...'
+                        )
+                        if status_bar is not None:
+                            status_bar.set_description(skip_description)
+                        else:
+                            products_bar.set_description(skip_description)
+                        break
+
+            if failed_products:
+                causes = '; '.join(
+                    f'{cause} (products: {", ".join(products)})'
+                    for cause, products in failed_products.items()
+                )
+                message = (
+                    'Exception thrown while getting license data for '
+                    f'{sum(len(products) for products in failed_products.values())} '
+                    'product(s); these will be omitted from the '
+                    f'subscription table: {causes}'
+                )
+                if status_bar is None:
+                    warnings.warn(message, RuntimeWarning)
+                else:
+                    self._deferred_license_warning = message
+
+            if any_success:
+                self._successful_calls.add(api_call)
+            else:
+                self._unsuccessful_calls.add(api_call)
+
+        except Exception as exc:
+            message = (
+                f'Exception thrown while getting license data; '
+                f'subscription table will be omitted: {exc}'
+            )
+            if status_bar is None:
+                warnings.warn(message, RuntimeWarning)
+            else:
+                self._deferred_license_warning = message
+            self._unsuccessful_calls.add(api_call)
+        finally:
+            http_logger.setLevel(previous_log_level)
+
+        return {'license_data': subscriptions}
+
     def call_gws_providers(self, products: list, quiet) -> dict:
         """
         Calls the relevant GWS APIs to get the data we need for the baselines.
@@ -990,12 +1202,32 @@ class Provider:
                 product_to_items.update(self.get_dnsinfo())
 
             if 'commoncontrols' in product_to_logs:
-                # add list of super admins if CC is being run
-                product_to_items.update(self.get_super_admins())
-                # add list of highly privileged users (CC 6.1)
-                product_to_items.update(self.get_privileged_users())
-                # add effective SSO assignment state (CC 6.1 API-based check)
-                product_to_items.update(self.get_inbound_sso_assignments())
+                commoncontrols_steps = (
+                    ('super admin users', self.get_super_admins),
+                    ('privileged users', self.get_privileged_users),
+                    ('inbound SSO assignments', self.get_inbound_sso_assignments),
+                    ('license data', None),
+                )
+                cc_bar = tqdm(total=len(commoncontrols_steps),
+                              leave=False,
+                              disable=quiet)
+                for step_name, step_func in commoncontrols_steps:
+                    cc_bar.set_description(
+                        'Running Provider: Exporting '
+                        f'{step_name} for commoncontrols...'
+                    )
+                    if step_name == 'license data':
+                        product_to_items.update(
+                            self.get_license_data(quiet=quiet, status_bar=cc_bar)
+                        )
+                    else:
+                        product_to_items.update(step_func())
+                    cc_bar.update(1)
+
+                cc_bar.close()
+                if self._deferred_license_warning is not None:
+                    warnings.warn(self._deferred_license_warning, RuntimeWarning)
+                    self._deferred_license_warning = None
 
             product_to_items.update(self.get_group_settings())
 

scubagoggles/reporter/IndividualReport/IndividualReportTemplate.html

@@ -23,6 +23,7 @@
             {{WARNING_NOTIFICATION}}
             {{METADATA}}
             {{TABLES}}
+            {{LICENSES}}
             {{RULES}}
             {{DNS_LOGS}}
         </main>

scubagoggles/reporter/reporter.py

@@ -49,7 +49,8 @@ class Reporter:
                  omissions: dict,
                  annotations: dict,
                  omit_ou: dict | None = None,
-                 progress_bar=None):
+                 progress_bar=None,
+                 license_data: list | None = None):
 
         """Reporter class initialization
         """
@@ -83,6 +84,7 @@ class Reporter:
         self.progress_bar = progress_bar
         self.rules_table = None
         self.annotated_failed_policies = {}
+        self._license_data = license_data if license_data is not None else []
 
     @staticmethod
     def _get_test_result(test: dict) -> str:
@@ -782,6 +784,8 @@ class Reporter:
                                                 product=self._product)
         if legend_html:
             fragments = [legend_html] + fragments
+        license_api = ApiReference.LIST_LICENSE_ASSIGNMENTS.value
+        include_licenses = self._product == 'commoncontrols'
         html, rules_table = rh.build_individual_report_html(
             fragments=fragments,
             rules_data=rules_data,
@@ -793,6 +797,13 @@ class Reporter:
             tenant_name=self._tenant_name,
             tenant_domain=self._tenant_domain,
             tenant_id=self._tenant_id,
+            include_licenses=include_licenses,
+            license_data=self._license_data,
+            license_collection_failed=(
+                license_api in self._unsuccessful_calls
+                if include_licenses
+                else False
+            ),
         )
         self.rules_table = rules_table
         return html

scubagoggles/reporter/reporter_html.py

@@ -235,7 +235,11 @@ def build_individual_report_html(*,
                                  main_report_name: str,
                                  tenant_name: str,
                                  tenant_domain: str,
-                                 tenant_id: str) -> tuple[str, list | None]:
+                                 tenant_id: str,
+                                 include_licenses: bool = False,
+                                 license_data: list | None = None,
+                                 license_collection_failed: bool = False,
+                                 ) -> tuple[str, list | None]:
 
     """Build an individual baseline report HTML page and optional rules
     table data.
@@ -263,6 +267,15 @@ def build_individual_report_html(*,
     html = html.replace('{{METADATA}}', meta)
     html = html.replace('{{TABLES}}', ''.join(fragments))
 
+    if include_licenses:
+        license_table = _build_license_table(
+            license_data or [],
+            license_collection_failed=license_collection_failed,
+        )
+    else:
+        license_table = ''
+    html = html.replace('{{LICENSES}}', license_table)
+
     rules_table = _build_rules_table(rules_data)
     html = html.replace('{{RULES}}', rules_table)
 
@@ -303,6 +316,48 @@ def build_individual_report_html(*,
     return html, rules_table
 
 
+def _build_license_table(license_data: list, *,
+                         license_collection_failed: bool = False) -> str:
+
+    """Build an HTML subscriptions table from the license data collected by
+    the Enterprise License Manager API.
+
+    :param list license_data: list of subscription dicts, each containing
+        product_name, status, and assigned.
+    :param bool license_collection_failed: True when the license API call did
+        not complete successfully.
+
+    :return: HTML fragment with the subscriptions table.
+    :rtype: str
+    """
+
+    if not license_data:
+        if license_collection_failed:
+            message = 'An error occurred when collecting license information.'
+        else:
+            message = 'No licenses found.'
+        return (
+            '\n<hr>\n'
+            '<h2 id="subscriptions">Tenant Licensing Information</h2>\n'
+            f'<p>{message}</p>\n'
+        )
+
+    rows = [
+        {
+            'Product Name': sub['product_name'],
+            'Status': sub['status'],
+            'Assigned Licenses': sub['assigned'],
+        }
+        for sub in license_data
+    ]
+
+    html_table = create_html_table(rows)
+    return (
+        '\n<hr>\n'
+        '<h2 id="subscriptions">Tenant Licensing Information</h2>\n'
+        + html_table
+        + '\n'
+    )
 def _create_meta_table(name: str,
                        domain: str,
                        ident: str,

scubagoggles/scuba_constants.py

@@ -25,6 +25,7 @@ class ApiReference(Enum):
     LIST_CUSTOMERS = 'directory/v1/customer/get'
     LIST_ACTIVITIES = 'reports/v1/activities/list'
     GET_GROUP = 'groups-settings/v1/groups/get'
+    LIST_LICENSE_ASSIGNMENTS = 'licensing/v1/product/{productId}/sku/{skuId}/users/list'
 
 
 class ApiUrl(Enum):
@@ -43,6 +44,10 @@ class ApiUrl(Enum):
     LIST_CUSTOMERS = f'{BASE_URL_ADMINSDK}/directory/v1/customer/get'
     LIST_ACTIVITIES = f'{BASE_URL_ADMINSDK}/reports/reference/rest/v1/activities/list'
     GET_GROUP = f'{BASE_URL_ADMINSDK}/groups-settings/v1/reference/groups/get'
+    LIST_LICENSE_ASSIGNMENTS = (
+        'https://developers.google.com/workspace/admin/licensing/reference/rest/v1/'
+        'licenseAssignments/listForProductAndSku'
+    )
 
 
 # Dictionary mapping short-hand reference to <a> tags linking to the documentation
@@ -84,6 +89,7 @@ API_SCOPES = {
     f'{BASE_AUTH_URL}/cloud-identity.policies.readonly':        AuthFlow.OAUTH | AuthFlow.DWD,
     f'{BASE_AUTH_URL}/cloud-identity.inboundsso.readonly':      AuthFlow.OAUTH | AuthFlow.DWD,
     f'{BASE_AUTH_URL}/apps.groups.settings':                    AuthFlow.OAUTH | AuthFlow.DASA,
+    f'{BASE_AUTH_URL}/apps.licensing':                          AuthFlow.OAUTH | AuthFlow.DASA,
 }
 
 def scopes_for(flow: AuthFlow) -> tuple: