#!/usr/bin/env python3
"""Audit ContraForce enterprise application permissions in Microsoft Entra ID.
Queries Microsoft Graph for all documented ContraForce enterprise applications,
resolves their delegated and application permissions to human-readable names,
and outputs a structured JSON file suitable for diffing against documentation.
Requires:
- Python 3.10+ (older versions will produce a clear version error).
See https://devguide.python.org/versions/ for supported Python versions.
- ``azure-identity>=1.19`` and ``httpx>=0.28``::
pip install azure-identity httpx
- Azure CLI (``az login``) session — used by ``AzureCliCredential`` for token acquisition
- The target tenant must have the ContraForce enterprise applications consented
Usage:
python audit_enterprise_apps.py
python audit_enterprise_apps.py -o audit-2026-02-12.json
python audit_enterprise_apps.py -c AzureUSGovernment
python audit_enterprise_apps.py -c AzureUSGovernment -o gov-audit.json
"""
import argparse
import dataclasses
import importlib.util
import json
import os
import stat
import sys
import time
from datetime import datetime, timezone
from typing import Any
# ── Requires: Python 3.10+, azure-identity>=1.19, httpx>=0.28 ────────────────
# Mirrors PowerShell's #Requires — validate before any third-party import.
if sys.version_info < (3, 10): # noqa: UP036
print(
f"Python 3.10+ is required (running {sys.version.split()[0]})",
file=sys.stderr,
)
raise SystemExit(1)
_REQUIRED_PACKAGES = [
("httpx", "httpx>=0.28"),
("azure.identity", "azure-identity>=1.19"),
]
_missing = [spec for mod, spec in _REQUIRED_PACKAGES if importlib.util.find_spec(mod) is None]
if _missing:
print(
f"Missing required packages: {', '.join(_missing)}\n"
"Install via: pip install azure-identity httpx",
file=sys.stderr,
)
raise SystemExit(1)
del _REQUIRED_PACKAGES, _missing
import httpx # noqa: E402
from azure.core.exceptions import ClientAuthenticationError # noqa: E402
from azure.identity import AzureCliCredential, CredentialUnavailableError # noqa: E402
TOOL_VERSION = "2.1.0"
# ── Cloud environment endpoints ───────────────────────────────────────────────
# Microsoft Graph endpoints differ by cloud environment. Commercial (including
# GCC) uses graph.microsoft.com; GCC High and DoD use graph.microsoft.us.
# See: https://learn.microsoft.com/en-us/graph/deployments
CLOUD_ENVIRONMENTS: dict[str, dict[str, str]] = {
"AzureCloud": {
"graph_base": "https://graph.microsoft.com/v1.0",
"graph_scope": "https://graph.microsoft.com/.default",
"name": "Commercial",
},
"AzureUSGovernment": {
"graph_base": "https://graph.microsoft.us/v1.0",
"graph_scope": "https://graph.microsoft.us/.default",
"name": "US Government (GCC High / DoD)",
},
}
# ContraForce enterprise applications to audit (Commercial / AzureCloud only).
# App IDs differ by cloud environment. Government cloud app IDs are provided
# by ContraForce upon request and passed via --apps-file.
COMMERCIAL_APPS = [
{"name": "ContraForce API", "app_id": "24d97bc0-8f2b-45d5-8e0b-7fe286732ef2"},
{"name": "ContraForce Portal", "app_id": "8b7cb435-9526-47ee-b79a-34433f0daad2"},
{"name": "ContraForce for MDE", "app_id": "6efccc6a-f0d3-49e5-92d0-17d4afa9ba52"},
{"name": "ContraForce Gamebooks for MDE", "app_id": "ad7b0e79-3c37-4408-bf8f-eb89522cc920"},
{
"name": "ContraForce Gamebooks for Identity",
"app_id": "36b0d51c-4c0f-4810-9cc4-bfbd40c7dd4a",
},
{
"name": "ContraForce Gamebooks for Email",
"app_id": "44dbf6fe-45e3-48a3-bac3-f8d4cf1dba6d",
},
{
"name": "ContraForce Sentinel Hunting",
"app_id": "6bf1c74d-7ade-4671-a507-166936f89a1f",
},
]
# Well-known first-party resource APIs that ContraForce integrates with.
RESOURCE_APIS = [
"Microsoft Graph",
"Windows Azure Service Management API",
"WindowsDefenderATP",
"Microsoft Threat Protection",
"Log Analytics API",
]
# Friendly display names for resource APIs.
FRIENDLY_NAMES: dict[str, str] = {
"Windows Azure Service Management API": "Azure Service Management",
}
@dataclasses.dataclass
class PermissionRegistry:
"""Lookup tables for resolving permission IDs to human-readable names."""
app_roles: dict[str, dict[str, dict[str, str | None]]] = dataclasses.field(
default_factory=dict,
)
delegated_scopes: dict[str, dict[str, dict[str, str | None]]] = dataclasses.field(
default_factory=dict,
)
resource_names: dict[str, str] = dataclasses.field(default_factory=dict)
_scope_desc_by_name: dict[str, dict[str, str | None]] = dataclasses.field(
default_factory=dict, repr=False,
)
def index_service_principal(self, sp: dict, *, friendly_name: str | None = None) -> None:
"""Index a service principal's roles and scopes for later resolution."""
sp_id = sp["id"]
self.resource_names[sp_id] = friendly_name or sp["displayName"]
self.app_roles[sp_id] = {
role["id"]: {"name": role["value"], "description": role.get("description")}
for role in sp.get("appRoles") or []
}
scopes_by_id: dict[str, dict[str, str | None]] = {}
desc_by_name: dict[str, str | None] = {}
for scope in sp.get("oauth2PermissionScopes") or []:
desc = scope.get("adminConsentDescription")
scopes_by_id[scope["id"]] = {"name": scope["value"], "description": desc}
desc_by_name[scope["value"]] = desc
self.delegated_scopes[sp_id] = scopes_by_id
self._scope_desc_by_name[sp_id] = desc_by_name
def resolve_scope_description(self, resource_id: str, scope_name: str) -> str | None:
"""Look up a delegated scope description by name."""
return self._scope_desc_by_name.get(resource_id, {}).get(scope_name)
class AuditAuthError(Exception):
"""Raised when Azure CLI authentication fails."""
class GraphClient:
"""Lightweight Microsoft Graph client backed by httpx and AzureCliCredential."""
def __init__(self, *, cloud: str = "AzureCloud") -> None:
env = CLOUD_ENVIRONMENTS[cloud]
self._graph_base = env["graph_base"]
self._graph_scope = env["graph_scope"]
try:
self._credential = AzureCliCredential()
self._token = self._credential.get_token(self._graph_scope)
except (CredentialUnavailableError, ClientAuthenticationError) as e:
raise AuditAuthError(
"Azure CLI is not authenticated. Run 'az login' first.\n"
"For government cloud: az cloud set --name AzureUSGovernment && az login"
) from e
self._http = httpx.Client(
timeout=httpx.Timeout(30.0, connect=10.0),
)
self._refresh_auth()
# Resolve identity from Microsoft Graph API rather than decoding the
# access token JWT on the client side. Access tokens are intended for
# the resource server, not the client — decoding them without signature
# verification is architecturally incorrect and raises red flags in
# security reviews. Using /me and /organization is both correct and
# eliminates the need for a JWT verification library.
self._resolve_identity()
def _resolve_identity(self) -> None:
"""Resolve authenticated user and tenant identity from Microsoft Graph."""
me_resp = self._request_with_retry(
f"{self._graph_base}/me",
params={"$select": "id,userPrincipalName"},
)
if me_resp is not None:
me = me_resp.json()
self.user_upn: str = me.get("userPrincipalName", "")
self.user_oid: str = me.get("id", "")
else:
self.user_upn = ""
self.user_oid = ""
print(
" WARNING: Could not resolve user identity from /me.",
file=sys.stderr,
)
org_resp = self._request_with_retry(
f"{self._graph_base}/organization",
params={"$select": "id"},
)
if org_resp is not None:
orgs = org_resp.json().get("value", [])
self.tenant_id: str = orgs[0]["id"] if orgs else ""
else:
self.tenant_id = ""
print(
" WARNING: Could not resolve tenant from /organization.",
file=sys.stderr,
)
# NOTE: Remove quotes when minimum version is Python 3.14+ (PEP 649).
def __enter__(self) -> "GraphClient": # quoted: class name isn't bound yet
return self
def __exit__(self, *exc: object) -> None:
self._http.close()
def _refresh_auth(self) -> None:
"""Refresh the bearer token if it is within 5 minutes of expiry."""
if self._token.expires_on - time.time() < 300:
self._token = self._credential.get_token(self._graph_scope)
self._http.headers["Authorization"] = f"Bearer {self._token.token}"
def _request_with_retry(
self, url: str, params: dict[str, str] | None = None, *, max_retries: int = 3,
) -> httpx.Response | None:
"""GET with retry and 429/Retry-After handling."""
for attempt in range(max_retries + 1):
self._refresh_auth()
try:
resp = self._http.get(url, params=params)
if resp.status_code == 429:
try:
retry_after = int(resp.headers.get("Retry-After", 2**attempt))
except ValueError:
retry_after = 2**attempt
print(f" Throttled, retrying in {retry_after}s...", file=sys.stderr)
time.sleep(retry_after)
continue
resp.raise_for_status()
return resp
except httpx.HTTPError as e:
if attempt < max_retries:
time.sleep(2**attempt)
continue
print(
f" WARNING: Graph call failed after {max_retries + 1} attempts: {url}\n"
f" {e}",
file=sys.stderr,
)
return None
return None
def paginated_get(self, url: str, params: dict[str, str] | None = None) -> list[dict]:
"""GET with automatic @odata.nextLink pagination."""
all_values: list[dict] = []
current_url: str | None = url
current_params = params
while current_url:
resp = self._request_with_retry(current_url, current_params)
if resp is None:
print(
f" WARNING: Pagination interrupted — returning {len(all_values)} "
f"partial result(s) for {url}",
file=sys.stderr,
)
break
data = resp.json()
all_values.extend(data.get("value", []))
current_url = data.get("@odata.nextLink")
current_params = None # nextLink includes query params
return all_values
def list_service_principals(
self, *, odata_filter: str, select: list[str],
) -> list[dict]:
"""Query /servicePrincipals with an OData filter."""
return self.paginated_get(
f"{self._graph_base}/servicePrincipals",
params={"$filter": odata_filter, "$select": ",".join(select)},
)
def get_oauth2_permission_grants(self, sp_id: str) -> list[dict]:
"""Get delegated permission grants for a service principal."""
return self.paginated_get(
f"{self._graph_base}/servicePrincipals/{sp_id}/oauth2PermissionGrants",
)
def get_app_role_assignments(self, sp_id: str) -> list[dict]:
"""Get application permission assignments for a service principal."""
return self.paginated_get(
f"{self._graph_base}/servicePrincipals/{sp_id}/appRoleAssignments",
)
def resolve_resource_apis(graph: GraphClient, registry: PermissionRegistry) -> None:
"""Resolve resource API service principals and populate the registry."""
print("\033[36mResolving resource API service principals...\033[0m")
select = ["id", "displayName", "appRoles", "oauth2PermissionScopes"]
for api_name in RESOURCE_APIS:
results = graph.list_service_principals(
odata_filter=f"displayName eq '{api_name}'",
select=select,
)
if not results:
print(f" WARNING: Resource API not found in tenant: {api_name}", file=sys.stderr)
continue
sp = results[0]
sp_id = sp["id"]
friendly_name = FRIENDLY_NAMES.get(sp["displayName"], sp["displayName"])
registry.index_service_principal(sp, friendly_name=friendly_name)
print(
f" Resolved: {friendly_name} ({sp_id})"
f" — {len(registry.app_roles[sp_id])} app roles,"
f" {len(registry.delegated_scopes[sp_id])} delegated scopes"
)
def index_contraforce_apps(graph: GraphClient, registry: PermissionRegistry) -> None:
"""Index ContraForce apps for internal cross-app scope resolution.
Some ContraForce applications delegate to each other via custom OAuth2
scopes (e.g., the Portal delegates to the API). This queries for ALL
service principals whose displayName starts with ``ContraForce`` — not
just the seven audited apps — so that cross-app scopes resolve to
human-readable names instead of raw GUIDs in the output.
This broader query does NOT grant any additional access; it only reads
public service principal metadata visible to any authenticated directory
reader. Permissions returned from this query are tagged with
``"internal": true`` in the output to distinguish them from permissions
that grant access to tenant data.
"""
cf_sps = graph.list_service_principals(
odata_filter="startswith(displayName, 'ContraForce')",
select=["id", "displayName", "oauth2PermissionScopes"],
)
for cf_sp in cf_sps:
if cf_sp["id"] not in registry.resource_names:
registry.index_service_principal(cf_sp)
def audit_delegated_permissions(
graph: GraphClient,
sp_id: str,
registry: PermissionRegistry,
) -> list[dict[str, Any]]:
"""Query and resolve delegated permissions (oauth2PermissionGrants)."""
grants = graph.get_oauth2_permission_grants(sp_id)
delegated: list[dict[str, Any]] = []
for grant in grants:
resource_id = grant["resourceId"]
resource_name = registry.resource_names.get(resource_id, resource_id)
is_internal = resource_name.startswith("ContraForce ")
scope_names = sorted(s for s in grant.get("scope", "").split() if s)
for scope in scope_names:
delegated.append({
"permission": scope,
"api": resource_name,
"type": "Delegated",
"description": registry.resolve_scope_description(resource_id, scope),
"internal": is_internal,
})
return sorted(delegated, key=lambda p: (p["api"], p["permission"]))
def audit_application_permissions(
graph: GraphClient,
sp_id: str,
registry: PermissionRegistry,
) -> list[dict[str, Any]]:
"""Query and resolve application permissions (appRoleAssignments)."""
assignments = graph.get_app_role_assignments(sp_id)
app_perms: list[dict[str, Any]] = []
for assignment in assignments:
resource_name = assignment.get("resourceDisplayName", "")
resource_id = assignment["resourceId"]
role_id = assignment["appRoleId"]
# Resolve the role ID to a permission name and description
role = registry.app_roles.get(resource_id, {}).get(role_id)
perm_name = role["name"] if role else role_id
desc = role["description"] if role else None
app_perms.append({
"permission": perm_name,
"api": registry.resource_names.get(resource_id, resource_name),
"type": "Application",
"description": desc,
})
return sorted(app_perms, key=lambda p: (p["api"], p["permission"]))
def audit_app(
graph: GraphClient,
app: dict[str, str],
registry: PermissionRegistry,
) -> dict[str, Any]:
"""Audit a single enterprise application."""
print(f"\n\033[36mAuditing: {app['name']} ({app['app_id']})...\033[0m")
results = graph.list_service_principals(
odata_filter=f"appId eq '{app['app_id']}'",
select=["id", "displayName", "appId"],
)
if not results:
print(" WARNING: NOT FOUND in tenant — skipping", file=sys.stderr)
return {
"applicationName": app["name"],
"appId": app["app_id"],
"status": "NOT_FOUND",
"delegatedPermissions": [],
"applicationPermissions": [],
}
sp = results[0]
sp_id = sp["id"]
delegated = audit_delegated_permissions(graph, sp_id, registry)
app_perms = audit_application_permissions(graph, sp_id, registry)
print(f" Delegated: {len(delegated)} | Application: {len(app_perms)}")
return {
"applicationName": sp["displayName"],
"appId": sp["appId"],
"status": "OK",
"delegatedPermissions": delegated,
"applicationPermissions": app_perms,
}
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
parser = argparse.ArgumentParser(
description="Audit ContraForce enterprise application permissions in Microsoft Entra ID.",
)
parser.add_argument(
"-o", "--output",
default="enterprise-apps-audit.json",
help="Path for the output JSON file (default: enterprise-apps-audit.json)",
)
parser.add_argument(
"-c", "--cloud",
choices=list(CLOUD_ENVIRONMENTS),
default="AzureCloud",
help=(
"Cloud environment to audit: AzureCloud (default) or"
" AzureUSGovernment (GCC High / DoD). Must match the environment"
" set with 'az cloud set --name <value>'."
),
)
parser.add_argument(
"-a", "--apps-file",
help=(
"Path to a JSON file listing the applications to audit. Each entry"
" must have 'name' and 'app_id' fields. Required for government"
" cloud environments where app IDs differ from commercial."
" Contact support@contraforce.com for your environment's app IDs."
),
)
parser.add_argument(
"--redact-upn",
action="store_true",
help=(
"Record the operator's Entra object ID instead of UPN in output"
" metadata. Automatically enabled for government environments."
),
)
return parser.parse_args()
def main() -> int:
"""Run the enterprise application audit and return exit code."""
args = parse_args()
cloud = args.cloud
# Resolve the application list: built-in for commercial, file-based for gov
if args.apps_file:
with open(args.apps_file, encoding="utf-8") as f:
apps_to_audit = json.load(f)
elif cloud == "AzureCloud":
apps_to_audit = COMMERCIAL_APPS
else:
print(
"ERROR: Government cloud environments require --apps-file.\n"
"App IDs differ by cloud environment. Contact support@contraforce.com\n"
"to obtain the app IDs for your environment.",
file=sys.stderr,
)
return 1
try:
graph = GraphClient(cloud=cloud)
except AuditAuthError as e:
print(f"ERROR: {e}", file=sys.stderr)
return 1
with graph:
# Auto-enable UPN redaction for government cloud environments
redact_upn = args.redact_upn or cloud != "AzureCloud"
generated_by = graph.user_oid if redact_upn else graph.user_upn
tenant_id = graph.tenant_id
env_name = CLOUD_ENVIRONMENTS[cloud]["name"]
print(
f"\033[32mAuthenticated as: {generated_by}"
f" (Tenant: {tenant_id}, Environment: {env_name})\033[0m"
)
registry = PermissionRegistry()
resolve_resource_apis(graph, registry)
index_contraforce_apps(graph, registry)
results = [audit_app(graph, app, registry) for app in apps_to_audit]
# Build output document
output = {
"metadata": {
"generatedAt": datetime.now(timezone.utc).isoformat(), # noqa: UP017
"tenantId": tenant_id,
"generatedBy": generated_by,
"toolVersion": TOOL_VERSION,
"environment": cloud,
"description": (
"ContraForce enterprise application permissions snapshot"
" for documentation auditing."
),
},
"applications": results,
}
output_json = json.dumps(output, indent=2, ensure_ascii=False) + "\n"
with open(args.output, "w", encoding="utf-8", newline="\n") as f:
f.write(output_json)
# Restrict file permissions for government cloud environments
if cloud != "AzureCloud":
try:
os.chmod(args.output, stat.S_IRUSR | stat.S_IWUSR)
except OSError:
print(
f" WARNING: Could not restrict file permissions on {args.output}.\n"
' On Windows, run: icacls <file> /inheritance:r /grant:r "%USERNAME%":F',
file=sys.stderr,
)
print(f"\n\033[32mAudit complete. Output written to: {args.output}\033[0m")
print(f"Applications audited: {len(results)}")
print(f"Tenant: {tenant_id}")
print(f"Environment: {env_name}")
if redact_upn:
print("Operator identity: redacted (object ID used)")
# Exit with non-zero code if any apps were not found
failures = [r for r in results if r["status"] == "NOT_FOUND"]
if failures:
print(
f"\nWARNING: {len(failures)} application(s) were not found in the tenant.",
file=sys.stderr,
)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
