#!/usr/bin/env python3 """ Fetch Intune + Entra (Microsoft Graph) counts, Intune managed-device rows (for CSV export), a sample list of disabled Entra users (for the dashboard Alerts list; optional per-user M365 license assignments for cleanup alerts), and write data/analytics.json for the dashboard. Setup: 1. Register an app in Entra ID → Certificates & secrets → client secret. 2. API permissions (Application) → admin consent: DeviceManagementManagedDevices.Read.All DeviceManagementApps.Read.All (Intune app count + appList in analytics.json) User.Read.All Organization.Read.All (M365 / Azure AD subscribed SKUs → licenses table) Optional (same app) for security.microsoft.com–class data via Graph → defenderGraph: SecurityEvents.Read.All (Secure Score) SecurityIncident.Read.All (incident counts + recent incidents list) IdentityRiskyUser.Read.All (at-risk identities count) 3. cp env.example .env # then edit .env (INTUNE_* or AZURE_* variable names; optional INTUNE_APPS_LIST_MAX for intune.appList size) 4. python3 -m venv .venv && source .venv/bin/activate # Windows: .venv\\Scripts\\activate 5. pip3 install -r requirements.txt # or: python3 -m pip install -r requirements.txt 6. ./scripts/run_dashboard_job.sh fetch_m365_analytics.py .env is read automatically from the project root (no extra package required). Optional: pip3 install python-dotenv for advanced .env features. """ from __future__ import annotations import json import os import sys from typing import Any, cast from datetime import datetime, timedelta, timezone from pathlib import Path import msal import requests from store import json_snapshot_exists, read_json_snapshot, write_json_snapshot from env_config import env_csv, env_frozenset_upper, env_str, license_sku_excluded from paths import DATA_DIR, REPO_ROOT ROOT = REPO_ROOT OUT = Path(os.environ.get("ANALYTICS_OUT", str(DATA_DIR / "analytics.json"))) GOOGLE_WORKSPACE_JSON = DATA_DIR / "google-workspace.json" GRAPH = "https://graph.microsoft.com/v1.0" # Friendly names for common skuPartNumber values (Graph); unknown SKUs use part number with spaces. SKU_DISPLAY_NAMES: dict[str, str] = { "AAD_PREMIUM": "Microsoft Entra ID P1", "AAD_PREMIUM_P2": "Microsoft Entra ID P2", "EMS": "Enterprise Mobility + Security E3", "EMSPREMIUM": "Enterprise Mobility + Security E5", "IDENTITY_THREAT_PROTECTION": "Microsoft 365 E5 Security", "M365EDU": "Microsoft 365 Education", "O365_BUSINESS": "Microsoft 365 Apps for business", "O365_BUSINESS_ESSENTIALS": "Microsoft 365 Business Basic", "O365_BUSINESS_PREMIUM": "Microsoft 365 Business Standard", "SPE_E3": "Microsoft 365 E3", "SPE_E5": "Microsoft 365 E5", "ENTERPRISEPACK": "Office 365 E3", "ENTERPRISEPREMIUM": "Office 365 E5", "DEFENDER_ENDPOINT_P1": "Microsoft Defender for Endpoint P1", "DEFENDER_ENDPOINT_P2": "Microsoft Defender for Endpoint P2", "WIN_DEF_ATP": "Microsoft Defender for Endpoint P2", "FLOW_FREE": "Microsoft Power Automate Free", "POWER_BI_STANDARD": "Power BI (free)", "POWERAPPS_DEV": "Microsoft Power Apps for Developer", "POWERAPPS_VIRAL": "Microsoft Power Apps Plan 2 Trial", "STREAM": "Microsoft Stream", "STREAM_TRIAL": "Microsoft Stream Trial", "TEAMS1": "Microsoft Teams", "TEAMS_EXPLORATORY": "Microsoft Teams Exploratory", "TEAMS_ESSENTIALS_AAD": "Microsoft Teams Essentials", "MCOMEETADV": "Microsoft Teams Audio Conferencing", "WINDOWS_STORE": "Windows Store for Business", "WINDOWS365_BUSINESS": "Windows 365 Business", "WINDOWS365_ENTERPRISE": "Windows 365 Enterprise", "DYN365_BUSINESS_ESSENTIAL": "Dynamics 365 Business Central Essentials", "DYN365_FINANCIALS": "Dynamics 365 Business Central", "GUIDANCE": "Microsoft Entra ID Governance", "AAD_PREMIUM_P2_AAD_GOVERNANCE": "Microsoft Entra ID Governance", "MICROSOFT_BUSINESS_CENTER": "Dynamics 365 Business Central", "M365_F1": "Microsoft 365 F1", "SPE_F1": "Microsoft 365 F3", "VISIOCLIENT": "Visio Plan 2", "PROJECTPROFESSIONAL": "Project Plan 3", "EXCHANGEENTERPRISE": "Exchange Online (Plan 2)", "SHAREPOINTSTANDARD": "SharePoint Online (Plan 1)", "SHAREPOINTENTERPRISE": "SharePoint Online (Plan 2)", "MCOPSTN2": "Domestic and International Calling Plan", "FABRIC_FREE": "Microsoft Fabric (Free)", "MICROSOFT_COPILOT_STUDIO_VIRAL_TRIAL": "Microsoft Copilot Studio Viral Trial", "POWER_BI_PRO": "Power BI Pro", } # Subscribed SKUs to omit from the dashboard licenses table (configure in .env). def _license_sku_excluded(part: str) -> bool: return license_sku_excluded( part, prefixes=env_csv("LICENSE_SKU_EXCLUDE_PREFIXES"), exact=env_frozenset_upper("LICENSE_SKU_EXCLUDE_EXACT"), ) def _license_sku_id_excluded(sku_id: str) -> bool: exclude_ids = {s.lower() for s in env_csv("LICENSE_SKU_EXCLUDE_IDS")} return (sku_id or "").strip().lower() in exclude_ids def load_env_file() -> None: """Load KEY=value from .env without requiring python-dotenv.""" path = REPO_ROOT / ".env" if not path.is_file(): return for raw in path.read_text(encoding="utf-8-sig").splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[7:].strip() if "=" not in line: continue key, _, val = line.partition("=") key = key.strip() val = val.strip().strip('"').strip("'") if not key: continue prev = os.environ.get(key) if prev is None or not str(prev).strip(): os.environ[key] = val def load_dotenv() -> None: try: from dotenv import load_dotenv # type: ignore load_dotenv(REPO_ROOT / ".env") except ImportError: pass def build_msal_app() -> msal.ConfidentialClientApplication: tenant = ( os.environ.get("INTUNE_TENANT_ID", "").strip() or os.environ.get("AZURE_TENANT_ID", "").strip() ) client_id = ( os.environ.get("INTUNE_CLIENT_ID", "").strip() or os.environ.get("AZURE_CLIENT_ID", "").strip() ) secret = ( os.environ.get("INTUNE_CLIENT_SECRET", "").strip() or os.environ.get("AZURE_CLIENT_SECRET", "").strip() ) if not tenant or not client_id or not secret: bits = [] if not tenant: bits.append("tenant id (INTUNE_TENANT_ID or AZURE_TENANT_ID)") if not client_id: bits.append("client id (INTUNE_CLIENT_ID or AZURE_CLIENT_ID)") if not secret: bits.append( "client secret (INTUNE_CLIENT_SECRET or AZURE_CLIENT_SECRET — paste value in .env)" ) print("Missing: " + ", ".join(bits), file=sys.stderr) print(f"Expected file: {REPO_ROOT / '.env'}", file=sys.stderr) sys.exit(1) return msal.ConfidentialClientApplication( client_id, authority=f"https://login.microsoftonline.com/{tenant}", client_credential=secret, ) def graph_token(app: msal.ConfidentialClientApplication) -> str: result = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"]) if "access_token" not in result: print(result.get("error"), result.get("error_description"), file=sys.stderr) sys.exit(1) return str(result["access_token"]) def headers(tok: str) -> dict[str, str]: return { "Authorization": f"Bearer {tok}", "ConsistencyLevel": "eventual", } # Fields stored under analytics.json → intune.devices for dashboard “Generate report” CSV. _MANAGED_DEVICE_SELECT = ( "id,deviceName,managedDeviceName,operatingSystem,osVersion,complianceState," "lastSyncDateTime,userPrincipalName,userDisplayName,serialNumber,model,manufacturer," "enrollmentProfileName,azureADDeviceId,deviceCategoryDisplayName" ) def _managed_device_export_row(d: dict[str, Any]) -> dict[str, str]: name = str(d.get("deviceName") or d.get("managedDeviceName") or "").strip() return { "id": str(d.get("id") or ""), "deviceName": name, "operatingSystem": str(d.get("operatingSystem") or ""), "osVersion": str(d.get("osVersion") or ""), "complianceState": str(d.get("complianceState") or ""), "lastSyncDateTime": str(d.get("lastSyncDateTime") or ""), "userPrincipalName": str(d.get("userPrincipalName") or ""), "userDisplayName": str(d.get("userDisplayName") or ""), "serialNumber": str(d.get("serialNumber") or ""), "model": str(d.get("model") or ""), "manufacturer": str(d.get("manufacturer") or ""), "enrollmentProfileName": str(d.get("enrollmentProfileName") or ""), "azureADDeviceId": str(d.get("azureADDeviceId") or ""), "category": str(d.get("deviceCategoryDisplayName") or ""), } def fetch_managed_devices_compliance_and_list( tok: str, ) -> tuple[list[dict[str, str]], int, int, list[dict[str, str]]]: """ Paginate managedDevices once: compliance tallies + export rows (same Graph permission). Stops storing device rows after INTUNE_DEVICE_LIST_MAX (default 10000) but keeps paginating so compliant / noncompliant counts reflect the full tenant. Non-compliant devices for dashboard Alerts are stored separately up to INTUNE_NONCOMPLIANT_ALERT_MAX (default 50). """ compliant = 0 noncompliant = 0 rows: list[dict[str, str]] = [] alert_rows: list[dict[str, str]] = [] max_rows = int(os.environ.get("INTUNE_DEVICE_LIST_MAX", "10000") or "10000") try: alert_max = int(os.environ.get("INTUNE_NONCOMPLIANT_ALERT_MAX", "50") or "50") except ValueError: alert_max = 50 alert_max = max(1, min(alert_max, 500)) url: str | None = ( f"{GRAPH}/deviceManagement/managedDevices" f"?$select={_MANAGED_DEVICE_SELECT}&$top=999" ) while url: r = requests.get(url, headers=headers(tok), timeout=120) if not r.ok: raise RuntimeError(f"managedDevices list: {r.status_code} {r.text[:500]}") data = r.json() for d in data.get("value", []): if not isinstance(d, dict): continue cs = (d.get("complianceState") or "").strip().lower() if cs == "compliant": compliant += 1 elif cs == "noncompliant": noncompliant += 1 if len(alert_rows) < alert_max: alert_rows.append(_managed_device_export_row(d)) if len(rows) < max_rows: rows.append(_managed_device_export_row(d)) url = data.get("@odata.nextLink") if len(rows) >= max_rows: print( f"[info] intune.devices capped at {max_rows} rows (INTUNE_DEVICE_LIST_MAX). " "Compliance counts still include all devices.", file=sys.stderr, ) if noncompliant > len(alert_rows): print( f"[info] intune.nonCompliantAlertDetail capped at {alert_max} rows " f"({noncompliant} non-compliant). Raise INTUNE_NONCOMPLIANT_ALERT_MAX to list more.", file=sys.stderr, ) return rows, compliant, noncompliant, alert_rows def odata_collection_count(tok: str, segment: str, filter_expr: str | None = None) -> int | None: """ Use $top=1&$count=true so Graph returns @odata.count without a large payload. """ url = f"{GRAPH}/{segment.lstrip('/')}" params: dict[str, str] = {"$top": "1", "$count": "true"} if filter_expr: params["$filter"] = filter_expr r = requests.get(url, headers=headers(tok), params=params, timeout=120) if not r.ok: raise RuntimeError(f"{segment}: {r.status_code} {r.text[:500]}") data = r.json() n = data.get("@odata.count") return int(n) if n is not None else None def _mobile_app_row(d: dict[str, Any]) -> dict[str, str]: odt_raw = str(d.get("@odata.type") or "") short = odt_raw if "microsoft.graph." in odt_raw: short = odt_raw.split("microsoft.graph.", 1)[-1] short = short.lstrip("#") return { "id": str(d.get("id") or ""), "displayName": str(d.get("displayName") or "").strip() or "—", "appType": short, } def count_intune_mobile_apps_ids_only(tok: str) -> int: """ Count-only pagination (id). Used as fallback if fetch_intune_mobile_apps fails. """ total = 0 url: str | None = f"{GRAPH}/deviceAppManagement/mobileApps?$select=id&$top=999" while url: r = requests.get(url, headers=headers(tok), timeout=120) if not r.ok: raise RuntimeError(f"mobileApps list: {r.status_code} {r.text[:500]}") data = r.json() vals = data.get("value") if not isinstance(vals, list): break total += len(vals) nxt = data.get("@odata.nextLink") url = str(nxt).strip() if nxt else None return total def fetch_intune_mobile_apps( tok: str, max_stored: int ) -> tuple[int, list[dict[str, str]], bool]: """ Paginate all mobileApps; total count matches Intune Apps → All apps. Stores up to max_stored rows (displayName + Graph type) for the dashboard / CSV. """ total = 0 stored: list[dict[str, str]] = [] url: str | None = ( f"{GRAPH}/deviceAppManagement/mobileApps" f"?$select=id,displayName&$top=999" ) while url: r = requests.get(url, headers=headers(tok), timeout=120) if not r.ok: raise RuntimeError(f"mobileApps list: {r.status_code} {r.text[:500]}") data = r.json() vals = data.get("value") if not isinstance(vals, list): break for item in vals: if not isinstance(item, dict): continue total += 1 if len(stored) < max_stored: stored.append(_mobile_app_row(item)) nxt = data.get("@odata.nextLink") url = str(nxt).strip() if nxt else None truncated = total > len(stored) return total, stored, truncated def safe_int(name: str, fn) -> int | None: try: return fn() except Exception as ex: # noqa: BLE001 print(f"[warn] {name}: {ex}", file=sys.stderr) return None def _entra_disabled_list_max() -> int: try: raw = int(os.environ.get("ENTRA_DISABLED_USERS_LIST_MAX", "100") or "100") except ValueError: raw = 100 return max(1, min(raw, 500)) def _entra_guest_list_max() -> int: try: raw = int(os.environ.get("ENTRA_GUEST_USERS_LIST_MAX", "500") or "500") except ValueError: raw = 500 return max(1, min(raw, 2000)) def fetch_guest_user_rows(tok: str, max_rows: int) -> list[dict[str, Any]]: """ Guest users (userType eq Guest). License assignment rows are Members only, so the dashboard guest tile list must come from this directory query, not licenseUserRows. """ cap = max(1, min(int(max_rows), 2000)) r = requests.get( f"{GRAPH}/users", headers=headers(tok), params={ "$filter": "userType eq 'Guest'", "$select": "id,displayName,userPrincipalName,mail,userType,accountEnabled", "$top": str(cap), }, timeout=120, ) if not r.ok: raise RuntimeError(f"users guest list: {r.status_code} {r.text[:500]}") out: list[dict[str, Any]] = [] for u in r.json().get("value") or []: if not isinstance(u, dict): continue out.append( { "id": str(u.get("id") or "").strip(), "displayName": str(u.get("displayName") or "").strip(), "userPrincipalName": str(u.get("userPrincipalName") or "").strip(), "mail": str(u.get("mail") or "").strip(), "userType": "Guest", "accountEnabled": bool(u.get("accountEnabled")), } ) out.sort(key=lambda row: str(row.get("displayName") or "").lower()) return out def _entra_disabled_include_sign_in() -> bool: """signInActivity needs AuditLog.Read.All; default off so User.Read.All can list disabled users.""" raw = (os.environ.get("ENTRA_DISABLED_INCLUDE_SIGNIN") or "").strip().lower() return raw in {"1", "true", "yes", "on"} def _entra_disabled_include_audit() -> bool: """directoryAudits disable timestamps need AuditLog.Read.All; on unless ENTRA_DISABLED_INCLUDE_AUDIT=0.""" raw = (os.environ.get("ENTRA_DISABLED_INCLUDE_AUDIT") or "1").strip().lower() return raw not in {"0", "false", "no", "off"} def _graph_permission_denied(resp: requests.Response) -> bool: if resp.status_code not in (401, 403): return False body = resp.text return ( "Authorization_RequestDenied" in body or "Authentication_MSGraphPermissionMissing" in body or "Insufficient privileges" in body ) def _disable_audit_max_pages() -> int: try: return max(1, min(int(os.environ.get("ENTRA_DISABLE_AUDIT_MAX_PAGES", "20") or "20"), 50)) except ValueError: return 20 def _latest_license_removed_at(plans: Any) -> str: """Latest assignedPlans.assignedDateTime where capabilityStatus is Deleted (offboarding signal).""" best = "" if not isinstance(plans, list): return best for plan in plans: if not isinstance(plan, dict): continue if str(plan.get("capabilityStatus") or "").strip().lower() != "deleted": continue when = str(plan.get("assignedDateTime") or "").strip() if when and (not best or when > best): best = when return best def _best_deactivation_iso( audit_at: str, sessions_at: str, license_removed_at: str = "" ) -> tuple[str, str]: """Pick newest deactivation proxy (audit, session revoke, or license removal).""" candidates: list[tuple[str, str]] = [] audit_at = audit_at.strip() sessions_at = sessions_at.strip() license_removed_at = license_removed_at.strip() if audit_at: candidates.append((audit_at, "audit")) if sessions_at: candidates.append((sessions_at, "sessionsRevoked")) if license_removed_at: candidates.append((license_removed_at, "licenseRemoved")) if not candidates: return "", "" return max(candidates, key=lambda item: item[0]) def _disabled_account_activity_iso(row: dict[str, Any]) -> str: """Newest deactivation time stored on the row (see enrich_disabled_deactivation_times).""" for key in ("deactivatedAt", "disabledAt", "signInSessionsValidFromDateTime"): val = str(row.get(key) or "").strip() if val: return val return "" def sort_disabled_accounts(rows: list[dict[str, Any]]) -> None: rows.sort( key=lambda row: ( _disabled_account_activity_iso(row), str(row.get("displayName") or "").lower(), ), reverse=True, ) def fetch_disable_audit_timestamps(tok: str) -> tuple[dict[str, str], bool]: """ Map Entra user id → latest Disable account audit time. Returns (map, permission_ok). permission_ok is False on 403 (AuditLog.Read.All missing). """ out: dict[str, str] = {} url: str | None = f"{GRAPH}/auditLogs/directoryAudits" params: dict[str, str] | None = { "$filter": "category eq 'UserManagement' and activityDisplayName eq 'Disable account'", "$top": "999", } max_pages = _disable_audit_max_pages() pages = 0 while url and pages < max_pages: r = requests.get( url, headers=headers(tok), params=params if pages == 0 else None, timeout=120, ) if not r.ok: if _graph_permission_denied(r): return {}, False raise RuntimeError(f"directoryAudits: {r.status_code} {r.text[:500]}") data = r.json() for event in data.get("value") or []: if not isinstance(event, dict): continue when = str(event.get("activityDateTime") or "").strip() if not when: continue for target in event.get("targetResources") or []: if not isinstance(target, dict): continue uid = str(target.get("id") or "").strip() if uid and (uid not in out or when > out[uid]): out[uid] = when nxt = data.get("@odata.nextLink") url = str(nxt).strip() if nxt else None params = None pages += 1 return out, True def fetch_latest_disable_audit_for_user(tok: str, user_id: str) -> str: """Latest Disable account audit for one user (needs AuditLog.Read.All).""" uid = str(user_id or "").strip() if not uid: return "" filt = ( "category eq 'UserManagement' and activityDisplayName eq 'Disable account' " f"and targetResources/any(t:t/id eq '{uid}')" ) r = requests.get( f"{GRAPH}/auditLogs/directoryAudits", headers=headers(tok), params={"$filter": filt, "$top": "50"}, timeout=60, ) if not r.ok: if _graph_permission_denied(r): return "" return "" best = "" for event in r.json().get("value") or []: if not isinstance(event, dict): continue when = str(event.get("activityDateTime") or "").strip() if when and (not best or when > best): best = when return best def enrich_disabled_deactivation_times( tok: str, rows: list[dict[str, Any]] ) -> bool: """ Set deactivatedAt + deactivatedSource on each disabled user row. Returns True when AuditLog.Read.All is available. """ audit_map: dict[str, str] = {} permission_ok = False try: audit_map, permission_ok = fetch_disable_audit_timestamps(tok) except Exception: # noqa: BLE001 audit_map, permission_ok = {}, False for row in rows: uid = str(row.get("id") or "").strip() sessions_at = str(row.get("signInSessionsValidFromDateTime") or "").strip() audit_at = audit_map.get(uid) or "" if permission_ok and not audit_at and uid: audit_at = fetch_latest_disable_audit_for_user(tok, uid) if audit_at: row["disabledAt"] = audit_at license_removed_at = str(row.get("licenseRemovedAt") or "").strip() best, source = _best_deactivation_iso(audit_at, sessions_at, license_removed_at) if best: row["deactivatedAt"] = best row["deactivatedSource"] = source return permission_ok def enrich_disabled_deactivation_from_sessions(rows: list[dict[str, Any]]) -> None: """No audit API — use signInSessionsValidFromDateTime + license removal dates.""" for row in rows: sessions_at = str(row.get("signInSessionsValidFromDateTime") or "").strip() license_removed_at = str(row.get("licenseRemovedAt") or "").strip() best, source = _best_deactivation_iso("", sessions_at, license_removed_at) if best: row["deactivatedAt"] = best row["deactivatedSource"] = source def fetch_disabled_user_rows(tok: str, max_rows: int) -> list[dict[str, Any]]: """ Users with accountEnabled eq false (User.Read.All). Used for dashboard Alerts (one per account). signInSessionsValidFromDateTime is a disable proxy (session revoke). lastSignInDateTime only when ENTRA_DISABLED_INCLUDE_SIGNIN=1 and AuditLog.Read.All is granted. """ cap = max(1, min(int(max_rows), 500)) select = ( "id,displayName,userPrincipalName,mail,signInSessionsValidFromDateTime,assignedPlans" ) if _entra_disabled_include_sign_in(): select += ",signInActivity" r = requests.get( f"{GRAPH}/users", headers=headers(tok), params={ "$filter": "accountEnabled eq false", "$select": select, # Graph does not support $orderby with this filter (Request_UnsupportedQuery). "$top": str(cap), }, timeout=120, ) if not r.ok: raise RuntimeError(f"users disabled list: {r.status_code} {r.text[:500]}") out: list[dict[str, Any]] = [] for u in r.json().get("value") or []: if not isinstance(u, dict): continue row: dict[str, Any] = { "id": str(u.get("id") or "").strip(), "displayName": str(u.get("displayName") or "").strip(), "userPrincipalName": str(u.get("userPrincipalName") or "").strip(), "mail": str(u.get("mail") or "").strip(), } sessions_from = str(u.get("signInSessionsValidFromDateTime") or "").strip() if sessions_from: row["signInSessionsValidFromDateTime"] = sessions_from license_removed = _latest_license_removed_at(u.get("assignedPlans")) if license_removed: row["licenseRemovedAt"] = license_removed sign_in = u.get("signInActivity") if isinstance(sign_in, dict): last_sign_in = str(sign_in.get("lastSignInDateTime") or "").strip() if last_sign_in: row["lastSignInDateTime"] = last_sign_in out.append(row) sort_disabled_accounts(out) return out def parse_subscribed_skus_payload( data: dict[str, Any], ) -> tuple[list[dict[str, int | str]], dict[str, str], dict[str, str]]: """ Build dashboard license rows + skuId (GUID) → display name / skuPartNumber for user assignments. """ out: list[dict[str, int | str]] = [] sku_id_to_name: dict[str, str] = {} sku_id_to_part: dict[str, str] = {} for sku in data.get("value", []) or []: if not isinstance(sku, dict): continue cap = (sku.get("capabilityStatus") or "").strip().lower() if cap == "deleted": continue pu = sku.get("prepaidUnits") or {} total = int(pu.get("enabled") or 0) assigned = int(sku.get("consumedUnits") or 0) warning = int(pu.get("warning") or 0) part = (sku.get("skuPartNumber") or "").strip() or "UNKNOWN" display = SKU_DISPLAY_NAMES.get(part, part.replace("_", " ")) sid = str(sku.get("skuId") or "").strip() if sid: sid_key = sid.lower() sku_id_to_name[sid_key] = display sku_id_to_part[sid_key] = part if _license_sku_excluded(part): continue out.append( { "skuId": sid, "skuPartNumber": part, "name": display, "total": total, "assigned": assigned, "available": max(0, total - assigned), "expiringSoon": warning, } ) out.sort(key=lambda row: str(row["name"]).lower()) return out, sku_id_to_name, sku_id_to_part def fetch_subscribed_license_rows_and_sku_map( tok: str, ) -> tuple[list[dict[str, int | str]], dict[str, str], dict[str, str]]: """ Microsoft Graph subscribedSkus — same totals as M365/Azure license blades (approx.). Requires Organization.Read.All (or Directory.Read.All) application + admin consent. """ r = requests.get(f"{GRAPH}/subscribedSkus", headers=headers(tok), timeout=120) if not r.ok: raise RuntimeError(f"subscribedSkus: {r.status_code} {r.text[:500]}") return parse_subscribed_skus_payload(r.json()) def _entra_disabled_license_check_enabled() -> bool: raw = (os.environ.get("ENTRA_DISABLED_LICENSE_CHECK") or "1").strip().lower() return raw not in ("0", "false", "no", "off") def _license_user_assignments_enabled() -> bool: raw = (os.environ.get("LICENSE_USER_ASSIGNMENTS") or "1").strip().lower() return raw not in ("0", "false", "no", "off") def fetch_license_user_rows( tok: str, sku_id_to_name: dict[str, str], sku_id_to_part: dict[str, str], ) -> list[dict[str, Any]]: """ One row per user × assigned SKU (Graph users.assignedLicenses). Requires User.Read.All (same as directory counts). """ out: list[dict[str, Any]] = [] url: str | None = ( f"{GRAPH}/users?$select=id,displayName,userPrincipalName,mail,userType," "accountEnabled,assignedLicenses&$orderby=displayName&$top=999" ) hdrs = headers(tok) while url: r = requests.get(url, headers=hdrs, timeout=180) if not r.ok: raise RuntimeError( f"users assignedLicenses: {r.status_code} {r.text[:500]}" ) data = r.json() for u in data.get("value") or []: if not isinstance(u, dict): continue base = { "userId": str(u.get("id") or "").strip(), "displayName": str(u.get("displayName") or "").strip(), "userPrincipalName": str(u.get("userPrincipalName") or "").strip(), "mail": str(u.get("mail") or "").strip(), "userType": str(u.get("userType") or "").strip(), "accountEnabled": bool(u.get("accountEnabled")), } for al in u.get("assignedLicenses") or []: if not isinstance(al, dict): continue sid = str(al.get("skuId") or "").strip().lower() if not sid: continue if sid and _license_sku_id_excluded(sid): continue part = sku_id_to_part.get(sid) or "" if part and _license_sku_excluded(part): continue name = sku_id_to_name.get(sid) or part.replace("_", " ") or sid[:8] + "…" out.append( { **base, "skuId": sid, "skuPartNumber": part, "name": name, } ) url = data.get("@odata.nextLink") out.sort( key=lambda row: ( str(row.get("name") or "").lower(), str(row.get("displayName") or "").lower(), ) ) return out def enrich_disabled_accounts_assigned_licenses( tok: str, accounts: list[dict[str, Any]], sku_id_to_name: dict[str, str], ) -> None: """ For each disabled account, set assignedMicrosoftLicenseCount and assignedMicrosoftLicenseNames via Graph $batch (User.Read.All reads assignedLicenses on users). """ if not accounts: return batch_size = 20 for start in range(0, len(accounts), batch_size): chunk = accounts[start : start + batch_size] reqs: list[dict[str, Any]] = [] order: list[dict[str, Any]] = [] for i, acct in enumerate(chunk): uid = str(acct.get("id") or "").strip() if not uid: continue rid = str(i) reqs.append( { "id": rid, "method": "GET", "url": f"/users/{uid}?$select=id,assignedLicenses", } ) order.append(acct) if not reqs: continue br = requests.post( f"{GRAPH}/$batch", headers=headers(tok), json={"requests": reqs}, timeout=180, ) if not br.ok: print( f"[warn] disabled user license batch: HTTP {br.status_code} {br.text[:400]}", file=sys.stderr, ) continue try: bundle = br.json() except ValueError: continue by_id: dict[str, Any] = {} for resp in bundle.get("responses") or []: if isinstance(resp, dict) and resp.get("id") is not None: by_id[str(resp["id"])] = resp for i, acct in enumerate(order): resp = by_id.get(str(i)) if not isinstance(resp, dict): continue if int(resp.get("status") or 0) != 200: continue body = resp.get("body") if not isinstance(body, dict): continue seen: set[str] = set() names: list[str] = [] n = 0 for al in body.get("assignedLicenses") or []: if not isinstance(al, dict): continue sid = str(al.get("skuId") or "").strip().lower() if not sid: continue n += 1 label = sku_id_to_name.get(sid) or sid[:8] + "…" if label not in seen: seen.add(label) if len(names) < 8: names.append(label) acct["assignedMicrosoftLicenseCount"] = n if names: acct["assignedMicrosoftLicenseNames"] = names def fetch_subscribed_license_rows(tok: str) -> list[dict[str, int | str]]: rows, _sku, _part = fetch_subscribed_license_rows_and_sku_map(tok) return rows def _odata_count(payload: dict[str, Any]) -> int | None: """Graph sometimes returns @odata.count as int or string.""" raw = payload.get("@odata.count") if raw is None: return None try: return int(raw) except (TypeError, ValueError): try: return int(float(str(raw).strip())) except (TypeError, ValueError): return None # Incidents still "open" in Defender / Graph (not only status eq active). _INCIDENT_OPEN_FILTER = ( "(status eq 'active') or (status eq 'inProgress') or (status eq 'awaitingAction')" ) _INCIDENT_ALERTABLE_SEVERITY_FILTER = ( "(severity eq 'medium' or severity eq 'high' or severity eq 'critical')" ) _INCIDENT_ALERTABLE_FILTER = ( f"({_INCIDENT_OPEN_FILTER}) and {_INCIDENT_ALERTABLE_SEVERITY_FILTER}" ) _INCIDENT_ACTIVE_ONLY_FILTER = "status eq 'active'" _INCIDENT_SELECT = ( "id,displayName,status,severity,createdDateTime,lastUpdateDateTime,summary" ) _ALERTS_V2_INCIDENT_CHUNK = 15 _IMPACTED_ASSETS_MAX_LEN = 220 # Graph security/incidents rejects $top > 50 (see OData "limit of '50' for Top query"). _GRAPH_INCIDENTS_PAGE_SIZE = 50 def _graph_incidents_page_top(requested: int) -> str: return str(min(max(1, requested), _GRAPH_INCIDENTS_PAGE_SIZE)) def _defender_incidents_months() -> int: try: return max( 1, min(int(os.environ.get("DEFENDER_INCIDENTS_MONTHS", "6") or "6"), 36), ) except ValueError: return 6 def _defender_incidents_since_iso(months: int | None = None) -> str: """UTC ISO timestamp for Graph createdDateTime ge (approx. N×30 days).""" m = months if months is not None else _defender_incidents_months() cutoff = datetime.now(timezone.utc) - timedelta(days=m * 30) return cutoff.strftime("%Y-%m-%dT%H:%M:%SZ") def _incident_created_since_clause(since_iso: str) -> str: return f"createdDateTime ge {since_iso}" def _incident_open_in_window_filter(since_iso: str) -> str: return f"{_incident_created_since_clause(since_iso)} and ({_INCIDENT_OPEN_FILTER})" def _incident_active_in_window_filter(since_iso: str) -> str: return f"{_incident_created_since_clause(since_iso)} and ({_INCIDENT_ACTIVE_ONLY_FILTER})" def _incident_alertable_in_window_filter(since_iso: str) -> str: return ( f"{_incident_created_since_clause(since_iso)} and ({_INCIDENT_ALERTABLE_FILTER})" ) def _incident_in_created_window(inc: dict[str, Any], since_iso: str) -> bool: created = str(inc.get("createdDateTime") or "").strip() if not created: return True return created >= since_iso def _defender_open_incidents_max() -> int: try: return max(1, min(int(os.environ.get("DEFENDER_OPEN_INCIDENTS_MAX", "50") or "50"), 200)) except ValueError: return 50 def _defender_incident_alert_max() -> int: try: return max(1, min(int(os.environ.get("DEFENDER_INCIDENT_ALERT_MAX", "50") or "50"), 200)) except ValueError: return 50 def _incident_export_row( inc: dict[str, Any], impacted_assets: str | None = None ) -> dict[str, str | None]: row: dict[str, str | None] = { "id": str(inc.get("id") or ""), "displayName": ((inc.get("displayName") or "")[:220] or "—"), "status": inc.get("status"), "severity": inc.get("severity"), "createdDateTime": inc.get("createdDateTime"), "lastUpdateDateTime": inc.get("lastUpdateDateTime"), "impactedAssets": impacted_assets, } return row def _odata_string_literal(value: str) -> str: return "'" + str(value).replace("'", "''") + "'" def _evidence_impacted_asset_label(ev: dict[str, Any]) -> str | None: otype = str(ev.get("@odata.type") or "") if "deviceEvidence" in otype: label = str(ev.get("deviceDnsName") or ev.get("deviceName") or "").strip() return label or None if "userEvidence" in otype or "mailboxEvidence" in otype: ua = ev.get("userAccount") if not isinstance(ua, dict): return None label = str( ua.get("userPrincipalName") or ua.get("displayName") or "" ).strip() return label or None if "cloudApplicationEvidence" in otype: label = str(ev.get("displayName") or "").strip() return label or None return None def _impacted_assets_from_alerts(alerts: list[dict[str, Any]]) -> str | None: labels: list[str] = [] seen: set[str] = set() for alert in alerts: if not isinstance(alert, dict): continue for ev in alert.get("evidence") or []: if not isinstance(ev, dict): continue label = _evidence_impacted_asset_label(ev) if not label or label in seen: continue seen.add(label) labels.append(label) if not labels: return None out = ", ".join(labels) if len(out) > _IMPACTED_ASSETS_MAX_LEN: out = out[: _IMPACTED_ASSETS_MAX_LEN - 1].rstrip(", ") + "…" return out def _fetch_impacted_assets_by_incident_ids( tok: str, incident_ids: list[str] ) -> dict[str, str]: """Map incident id → comma-separated users/devices/apps from alerts_v2 evidence.""" out: dict[str, str] = {} ids = [str(i).strip() for i in incident_ids if str(i).strip()] if not ids: return out chunk = max(1, _ALERTS_V2_INCIDENT_CHUNK) for start in range(0, len(ids), chunk): part = ids[start : start + chunk] flt = " or ".join( f"incidentId eq {_odata_string_literal(iid)}" for iid in part ) url: str | None = f"{GRAPH}/security/alerts_v2" params: dict[str, str] | None = { "$filter": flt, "$top": "999", } by_incident: dict[str, list[dict[str, Any]]] = {iid: [] for iid in part} while url: r = requests.get( url, headers=headers(tok), params=params if params else None, timeout=120, ) if not r.ok: if r.status_code == 403: return out break data = r.json() for alert in data.get("value") or []: if not isinstance(alert, dict): continue iid = str(alert.get("incidentId") or "").strip() if iid in by_incident: by_incident[iid].append(alert) nxt = data.get("@odata.nextLink") url = str(nxt).strip() if nxt else None params = None for iid, alerts in by_incident.items(): assets = _impacted_assets_from_alerts(alerts) if assets: out[iid] = assets return out def _attach_impacted_assets(tok: str, rows: list[dict[str, str | None]]) -> None: ids = [str(r.get("id") or "").strip() for r in rows if r.get("id")] if not ids: return try: assets_map = _fetch_impacted_assets_by_incident_ids(tok, ids) except Exception: # noqa: BLE001 return for row in rows: iid = str(row.get("id") or "").strip() if iid in assets_map: row["impactedAssets"] = assets_map[iid] def _fetch_open_incident_rows( tok: str, max_rows: int, since_iso: str ) -> list[dict[str, str | None]]: rows: list[dict[str, str | None]] = [] url: str | None = f"{GRAPH}/security/incidents" params: dict[str, str] | None = { "$filter": _incident_open_in_window_filter(since_iso), "$orderby": "lastUpdateDateTime desc", "$select": _INCIDENT_SELECT, "$top": _graph_incidents_page_top(max_rows), } while url and len(rows) < max_rows: r = requests.get( url, headers=headers(tok), params=params if params else None, timeout=120, ) if not r.ok: if r.status_code == 403: return rows if r.status_code == 400 and params and "createdDateTime ge" in params.get( "$filter", "" ): params = { "$filter": _INCIDENT_OPEN_FILTER, "$orderby": "lastUpdateDateTime desc", "$select": _INCIDENT_SELECT, "$top": _graph_incidents_page_top(max_rows), } url = f"{GRAPH}/security/incidents" continue raise RuntimeError(f"open incidents list: {r.status_code} {r.text[:500]}") data = r.json() for inc in data.get("value") or []: if not isinstance(inc, dict): continue if not _incident_in_created_window(inc, since_iso): continue rows.append(_incident_export_row(inc)) if len(rows) >= max_rows: break nxt = data.get("@odata.nextLink") url = str(nxt).strip() if nxt else None params = None _attach_impacted_assets(tok, rows) return rows def _risky_user_filter() -> str: return env_str("DEFENDER_RISKY_USER_FILTER") _RISKY_USER_SELECT = ( "id,userDisplayName,userPrincipalName,riskLevel,riskState,riskDetail,riskLastUpdatedDateTime" ) def _defender_risky_users_max() -> int: try: return max( 1, min(int(os.environ.get("DEFENDER_RISKY_USERS_MAX", "100") or "100"), 500), ) except ValueError: return 100 def _risky_user_export_row(u: dict[str, Any]) -> dict[str, str | None]: detail = str(u.get("riskDetail") or "").strip() return { "id": str(u.get("id") or ""), "userDisplayName": ((u.get("userDisplayName") or "")[:120] or None), "userPrincipalName": u.get("userPrincipalName"), "riskLevel": u.get("riskLevel"), "riskState": u.get("riskState"), "riskDetail": (detail[:200] or None) if detail else None, "riskLastUpdatedDateTime": u.get("riskLastUpdatedDateTime"), } def _fetch_risky_user_rows(tok: str, max_rows: int) -> list[dict[str, str | None]]: rows: list[dict[str, str | None]] = [] url: str | None = f"{GRAPH}/identityProtection/riskyUsers" params: dict[str, str] | None = { "$filter": _risky_user_filter(), "$orderby": "riskLastUpdatedDateTime desc", "$select": _RISKY_USER_SELECT, "$top": str(min(max_rows, 999)), } while url and len(rows) < max_rows: r = requests.get( url, headers=headers(tok), params=params if params else None, timeout=120, ) if not r.ok: if r.status_code == 403: return rows if r.status_code == 400 and params and "confirmedCompromised" in params.get( "$filter", "" ): params = { "$filter": "riskState eq 'atRisk'", "$orderby": "riskLastUpdatedDateTime desc", "$select": _RISKY_USER_SELECT, "$top": str(min(max_rows, 999)), } url = f"{GRAPH}/identityProtection/riskyUsers" continue raise RuntimeError(f"riskyUsers list: {r.status_code} {r.text[:500]}") data = r.json() for user in data.get("value") or []: if not isinstance(user, dict): continue rows.append(_risky_user_export_row(user)) if len(rows) >= max_rows: break nxt = data.get("@odata.nextLink") url = str(nxt).strip() if nxt else None params = None return rows def _incident_is_alertable(inc: dict[str, Any]) -> bool: status = str(inc.get("status") or "").strip().lower().replace(" ", "") if status not in {"active", "inprogress", "awaitingaction", "new"}: return False sev = str(inc.get("severity") or "").strip().lower() if sev not in {"medium", "high", "critical"}: return False return True def _count_alertable_open_incidents(tok: str, since_iso: str) -> int | None: """Paginate open pipeline incidents in window; count medium+ severity.""" try: total = 0 url: str | None = f"{GRAPH}/security/incidents" params: dict[str, str] | None = { "$filter": _incident_open_in_window_filter(since_iso), "$select": _INCIDENT_SELECT, "$top": _graph_incidents_page_top(_GRAPH_INCIDENTS_PAGE_SIZE), } while url: r = requests.get( url, headers=headers(tok), params=params if params else None, timeout=120, ) if not r.ok: if r.status_code == 400 and params and "createdDateTime ge" in params.get( "$filter", "" ): params = { "$filter": _INCIDENT_OPEN_FILTER, "$select": _INCIDENT_SELECT, "$top": _graph_incidents_page_top(_GRAPH_INCIDENTS_PAGE_SIZE), } url = f"{GRAPH}/security/incidents" continue return None data = r.json() for inc in data.get("value") or []: if not isinstance(inc, dict): continue if not _incident_in_created_window(inc, since_iso): continue if _incident_is_alertable(inc): total += 1 nxt = data.get("@odata.nextLink") url = str(nxt).strip() if nxt else None params = None return total except Exception: # noqa: BLE001 return None def fetch_defender_graph_summary(tok: str) -> dict[str, Any]: """ Microsoft Graph data that backs much of security.microsoft.com (Secure Score, incidents, ID risk). Each sub-call is best-effort; missing permissions yield nulls + defenderGraphError hints. """ window_months = _defender_incidents_months() incidents_since = _defender_incidents_since_iso(window_months) out: dict[str, Any] = { "secureScore": None, "secureScoreMax": None, "secureScoreAt": None, "activeIncidents": None, "openIncidentsActive": None, "alertableOpenIncidents": None, "incidentsTotal": None, "atRiskIdentities": None, "riskyUsers": [], "openIncidents": [], "recentIncidents": [], "incidentsWindowMonths": window_months, "incidentsSince": incidents_since, "defenderGraphError": None, } errs: list[str] = [] open_window_filter = _incident_open_in_window_filter(incidents_since) active_window_filter = _incident_active_in_window_filter(incidents_since) alertable_window_filter = _incident_alertable_in_window_filter(incidents_since) def perm_hint(kind: str, perms: str) -> None: errs.append( f"{kind}: add Graph application permission {perms}, grant admin consent, re-run." ) # Secure Score (same source as Secure Score in Microsoft 365 Defender / security center) try: r = requests.get( f"{GRAPH}/security/secureScores", headers=headers(tok), params={"$top": "1", "$orderby": "createdDateTime desc"}, timeout=120, ) if r.status_code == 403: perm_hint("Secure Score", "SecurityEvents.Read.All") elif r.ok: vals = r.json().get("value") or [] if vals: s0 = vals[0] cs = s0.get("currentScore") mx = s0.get("maxScore") if cs is not None: out["secureScore"] = float(cs) if mx is not None: out["secureScoreMax"] = float(mx) out["secureScoreAt"] = s0.get("createdDateTime") else: errs.append(f"secureScores: HTTP {r.status_code}") except Exception as ex: # noqa: BLE001 errs.append(f"secureScores: {ex}") # Open incidents in window (@odata.count) — active + inProgress + awaitingAction. try: r = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={"$filter": open_window_filter, "$count": "true", "$top": "1"}, timeout=120, ) if r.status_code == 403: perm_hint("Incidents", "SecurityIncident.Read.All") elif r.ok: out["activeIncidents"] = _odata_count(r.json()) elif r.status_code == 400: try: rfb = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={ "$filter": active_window_filter, "$count": "true", "$top": "1", }, timeout=120, ) if rfb.ok: out["activeIncidents"] = _odata_count(rfb.json()) else: rfb2 = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={ "$filter": _INCIDENT_ACTIVE_ONLY_FILTER, "$count": "true", "$top": "1", }, timeout=120, ) if rfb2.ok: out["activeIncidents"] = _odata_count(rfb2.json()) else: errs.append( f"incidents open count: HTTP {r.status_code}; " f"fallback HTTP {rfb.status_code}/{rfb2.status_code}" ) except Exception as ex2: # noqa: BLE001 errs.append(f"incidents open count: HTTP {r.status_code}; fallback {ex2}") else: errs.append(f"incidents open count: HTTP {r.status_code}") except Exception as ex: # noqa: BLE001 errs.append(f"incidents open count: {ex}") # Active-only open count in window (status eq active). try: r_act = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={ "$filter": active_window_filter, "$count": "true", "$top": "1", }, timeout=120, ) if r_act.status_code == 403: perm_hint("Incidents active count", "SecurityIncident.Read.All") elif r_act.ok: out["openIncidentsActive"] = _odata_count(r_act.json()) elif r_act.status_code == 400: r_act2 = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={ "$filter": _INCIDENT_ACTIVE_ONLY_FILTER, "$count": "true", "$top": "1", }, timeout=120, ) if r_act2.ok: out["openIncidentsActive"] = _odata_count(r_act2.json()) else: errs.append(f"incidents active count: HTTP {r_act.status_code}") except Exception as ex: # noqa: BLE001 errs.append(f"incidents active count: {ex}") # Actionable open incidents in window (open pipeline + medium/high/critical) — tile + alerts. alertable_count = _count_alertable_open_incidents(tok, incidents_since) if alertable_count is not None: out["alertableOpenIncidents"] = alertable_count else: try: r_alert = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={ "$filter": alertable_window_filter, "$count": "true", "$top": "1", }, timeout=120, ) if r_alert.status_code == 403: perm_hint("Incidents alertable count", "SecurityIncident.Read.All") elif r_alert.ok: out["alertableOpenIncidents"] = _odata_count(r_alert.json()) elif r_alert.status_code != 400: errs.append(f"incidents alertable count: HTTP {r_alert.status_code}") except Exception as ex: # noqa: BLE001 errs.append(f"incidents alertable count: {ex}") # Incidents created in window (any status). try: r_all = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={ "$filter": _incident_created_since_clause(incidents_since), "$count": "true", "$top": "1", }, timeout=120, ) if r_all.ok: out["incidentsTotal"] = _odata_count(r_all.json()) elif r_all.status_code == 403: pass elif r_all.status_code == 400: r_all2 = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={"$count": "true", "$top": "1"}, timeout=120, ) if r_all2.ok: out["incidentsTotal"] = _odata_count(r_all2.json()) else: errs.append(f"incidents window total count: HTTP {r_all.status_code}") except Exception as ex: # noqa: BLE001 errs.append(f"incidents window total count: {ex}") # Open pipeline incident list (newest activity first) for dashboard detail + alerts. open_max = _defender_open_incidents_max() try: open_rows = _fetch_open_incident_rows(tok, open_max, incidents_since) out["openIncidents"] = open_rows out["recentIncidents"] = open_rows if out.get("alertableOpenIncidents") is None: out["alertableOpenIncidents"] = sum( 1 for row in open_rows if _incident_is_alertable(cast(dict[str, Any], row)) ) pipeline_total = out.get("activeIncidents") if ( isinstance(pipeline_total, int) and pipeline_total > len(open_rows) ): out["openIncidentsTruncated"] = True except Exception as ex: # noqa: BLE001 errs.append(f"open incidents list: {ex}") # Legacy fallback: if open list failed, keep a small createdDateTime sample. if not out["openIncidents"]: try: r = requests.get( f"{GRAPH}/security/incidents", headers=headers(tok), params={ "$top": "10", "$orderby": "createdDateTime desc", "$select": _INCIDENT_SELECT, }, timeout=120, ) if r.status_code == 403: perm_hint("Incident list", "SecurityIncident.Read.All") elif r.ok: rows: list[dict[str, str | None]] = [] for inc in r.json().get("value", []): if not isinstance(inc, dict): continue if not _incident_in_created_window(inc, incidents_since): continue row = _incident_export_row(inc) status = str(row.get("status") or "").strip().lower().replace(" ", "") if status not in {"active", "inprogress", "awaitingaction", "new"}: continue rows.append(row) _attach_impacted_assets(tok, rows) out["recentIncidents"] = rows out["openIncidents"] = rows else: errs.append(f"incidents list: HTTP {r.status_code}") except Exception as ex: # noqa: BLE001 errs.append(f"incidents list: {ex}") # Entra ID Protection — at-risk + confirmed compromised (IdentityRiskyUser.Read.All; Entra ID P2) try: r = requests.get( f"{GRAPH}/identityProtection/riskyUsers", headers=headers(tok), params={ "$filter": _risky_user_filter(), "$count": "true", "$top": "1", }, timeout=120, ) if r.status_code == 403: perm_hint("Risky users", "IdentityRiskyUser.Read.All") elif r.ok: out["atRiskIdentities"] = _odata_count(r.json()) elif r.status_code == 400: try: rfb = requests.get( f"{GRAPH}/identityProtection/riskyUsers", headers=headers(tok), params={ "$filter": "riskState eq 'atRisk'", "$count": "true", "$top": "1", }, timeout=120, ) if rfb.ok: out["atRiskIdentities"] = _odata_count(rfb.json()) else: errs.append( f"riskyUsers: HTTP {r.status_code}; " f"fallback atRisk-only HTTP {rfb.status_code}" ) except Exception as ex2: # noqa: BLE001 errs.append(f"riskyUsers: HTTP {r.status_code}; fallback {ex2}") else: errs.append(f"riskyUsers: HTTP {r.status_code}") except Exception as ex: # noqa: BLE001 errs.append(f"riskyUsers: {ex}") risky_max = _defender_risky_users_max() try: risky_rows = _fetch_risky_user_rows(tok, risky_max) out["riskyUsers"] = risky_rows total_risky = out.get("atRiskIdentities") if ( isinstance(total_risky, int) and total_risky > len(risky_rows) ): out["riskyUsersTruncated"] = True if out.get("atRiskIdentities") is None and risky_rows: out["atRiskIdentities"] = len(risky_rows) except Exception as ex: # noqa: BLE001 errs.append(f"riskyUsers list: {ex}") has_any = ( out["secureScore"] is not None or out["activeIncidents"] is not None or out["openIncidentsActive"] is not None or out["alertableOpenIncidents"] is not None or out["incidentsTotal"] is not None or out["atRiskIdentities"] is not None or len(cast(list[Any], out.get("riskyUsers") or [])) > 0 or len(cast(list[Any], out["openIncidents"])) > 0 or len(cast(list[Any], out["recentIncidents"])) > 0 ) if errs: joined = " ".join(list(dict.fromkeys(errs))) msg = (joined[:900] + "…") if len(joined) > 900 else joined out["defenderGraphError"] = ("Partial: " + msg) if has_any else msg return out def _merge_google_workspace_licenses_into_payload(payload: dict[str, Any]) -> None: """Copy Google Workspace license rows from google-workspace.json into analytics.json.""" if not json_snapshot_exists(GOOGLE_WORKSPACE_JSON): return gw = read_json_snapshot(GOOGLE_WORKSPACE_JSON) if not isinstance(gw, dict): return licenses = gw.get("licenses") if isinstance(licenses, list) and licenses: payload["googleWorkspaceLicenses"] = licenses err = gw.get("googleLicensesError") if err: payload["googleLicensesError"] = err note = gw.get("googleLicensesNote") if note: payload["googleLicensesNote"] = note updated = gw.get("updatedAt") if updated: payload["googleLicensesUpdatedAt"] = updated def main() -> None: load_env_file() load_dotenv() app = build_msal_app() tok = graph_token(app) comp_ok: int | None = None comp_bad: int | None = None intune_devices: list[dict[str, str]] = [] intune_devices_ok = False try: intune_devices, comp_ok, comp_bad, noncomp_alert = ( fetch_managed_devices_compliance_and_list(tok) ) intune_devices_ok = True except Exception as ex: # noqa: BLE001 print(f"[warn] managedDevices list + compliance: {ex}", file=sys.stderr) try: raw_apps_max = int(os.environ.get("INTUNE_APPS_LIST_MAX", "5000") or "5000") except ValueError: raw_apps_max = 5000 apps_list_max = max(1, min(raw_apps_max, 50_000)) apps_n: int | None = None apps_rows: list[dict[str, str]] = [] apps_truncated = False try: apps_n, apps_rows, apps_truncated = fetch_intune_mobile_apps(tok, apps_list_max) except Exception as ex: # noqa: BLE001 print(f"[warn] mobileApps: {ex}", file=sys.stderr) apps_n = safe_int( "apps", lambda: count_intune_mobile_apps_ids_only(tok), ) intune: dict[str, Any] = { "managedDevices": safe_int( "managedDevices", lambda: odata_collection_count( tok, "deviceManagement/managedDevices", None ), ), "compliant": comp_ok, "nonCompliant": comp_bad, "apps": apps_n, } if apps_rows: apps_rows.sort(key=lambda r: str(r.get("displayName") or "").lower()) intune["appList"] = apps_rows if apps_truncated: intune["appListTruncated"] = True if intune_devices_ok: intune["devices"] = intune_devices if noncomp_alert: intune["nonCompliantAlertDetail"] = noncomp_alert if comp_bad is not None and comp_bad > len(noncomp_alert): intune["nonCompliantAlertTruncated"] = True entra = { "users": safe_int( "users", lambda: odata_collection_count(tok, "users", None) ), "guestUsers": safe_int( "guestUsers", lambda: odata_collection_count(tok, "users", "userType eq 'Guest'"), ), "activeUsers": safe_int( "activeUsers", lambda: odata_collection_count(tok, "users", "accountEnabled eq true"), ), "disabledUsers": safe_int( "disabledUsers", lambda: odata_collection_count(tok, "users", "accountEnabled eq false"), ), } guest_accounts: list[dict[str, Any]] = [] try: guest_accounts = fetch_guest_user_rows(tok, _entra_guest_list_max()) except Exception as ex: # noqa: BLE001 print(f"[warn] guest users list: {ex}", file=sys.stderr) if guest_accounts: entra["guestAccounts"] = guest_accounts _gu = entra.get("guestUsers") if guest_accounts and isinstance(_gu, int) and _gu > len(guest_accounts): entra["guestAccountsTruncated"] = True disabled_accounts: list[dict[str, Any]] = [] try: disabled_accounts = fetch_disabled_user_rows(tok, _entra_disabled_list_max()) except Exception as ex: # noqa: BLE001 print(f"[warn] disabled users list: {ex}", file=sys.stderr) if disabled_accounts: entra["disabledAccounts"] = disabled_accounts _du = entra.get("disabledUsers") if ( disabled_accounts and isinstance(_du, int) and _du > len(disabled_accounts) ): entra["disabledAccountsTruncated"] = True licenses: list[dict[str, int | str]] = [] sku_id_map: dict[str, str] = {} sku_id_to_part: dict[str, str] = {} licenses_error: str | None = None try: licenses, sku_id_map, sku_id_to_part = fetch_subscribed_license_rows_and_sku_map(tok) except Exception as ex: # noqa: BLE001 print(f"[warn] licenses: {ex}", file=sys.stderr) raw = str(ex) if "403" in raw or "Authorization_RequestDenied" in raw: licenses_error = ( "Could not read licenses (403). Add Application permission " "Organization.Read.All on Microsoft Graph for this app, then grant admin consent." ) else: licenses_error = raw[:280] + ("…" if len(raw) > 280 else "") if disabled_accounts and _entra_disabled_license_check_enabled(): try: enrich_disabled_accounts_assigned_licenses(tok, disabled_accounts, sku_id_map) except Exception as ex: # noqa: BLE001 print(f"[warn] disabled user assigned licenses: {ex}", file=sys.stderr) if disabled_accounts: disable_audit_ok = False if _entra_disabled_include_audit(): try: disable_audit_ok = enrich_disabled_deactivation_times(tok, disabled_accounts) except Exception as ex: # noqa: BLE001 print(f"[warn] disable audit timestamps: {ex}", file=sys.stderr) enrich_disabled_deactivation_from_sessions(disabled_accounts) if not disable_audit_ok: entra["disabledDeactivationNote"] = ( "Disabled dates use the newest of license-removal and session-revoke " "times. Disabling without revoking sessions can leave an old session date " "(e.g. Ian Bonnici). Add AuditLog.Read.All for exact Entra Activity times." ) else: enrich_disabled_deactivation_from_sessions(disabled_accounts) sort_disabled_accounts(disabled_accounts) license_user_rows: list[dict[str, Any]] = [] license_assignments_error: str | None = None if _license_user_assignments_enabled() and sku_id_map: try: license_user_rows = fetch_license_user_rows(tok, sku_id_map, sku_id_to_part) except Exception as ex: # noqa: BLE001 print(f"[warn] license user assignments: {ex}", file=sys.stderr) raw = str(ex) if "403" in raw or "Authorization_RequestDenied" in raw: license_assignments_error = ( "Could not read user license assignments (403). " "Add User.Read.All on Microsoft Graph for this app, then grant admin consent." ) else: license_assignments_error = raw[:280] + ("…" if len(raw) > 280 else "") defender_graph: dict[str, Any] = {} try: defender_graph = fetch_defender_graph_summary(tok) except Exception as ex: # noqa: BLE001 print(f"[warn] defenderGraph: {ex}", file=sys.stderr) defender_graph = { "secureScore": None, "secureScoreMax": None, "secureScoreAt": None, "activeIncidents": None, "openIncidentsActive": None, "alertableOpenIncidents": None, "incidentsTotal": None, "atRiskIdentities": None, "riskyUsers": [], "openIncidents": [], "recentIncidents": [], "incidentsWindowMonths": _defender_incidents_months(), "incidentsSince": _defender_incidents_since_iso(), "defenderGraphError": str(ex)[:500], } payload = { "updatedAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "intune": intune, "entra": entra, "defenderGraph": defender_graph, "licenses": licenses, } if licenses_error: payload["licensesError"] = licenses_error if license_user_rows: payload["licenseUserRows"] = license_user_rows if license_assignments_error: payload["licenseAssignmentsError"] = license_assignments_error def _defender_graph_is_empty(dg: dict[str, Any]) -> bool: return ( dg.get("secureScore") is None and dg.get("activeIncidents") is None and dg.get("openIncidentsActive") is None and dg.get("alertableOpenIncidents") is None and dg.get("incidentsTotal") is None and dg.get("atRiskIdentities") is None and not dg.get("openIncidents") and not dg.get("recentIncidents") and not (dg.get("defenderGraphError") or "").strip() ) if _defender_graph_is_empty(defender_graph): print( "[info] defenderGraph has no counts yet. On the app registration add Microsoft Graph " "application permissions SecurityEvents.Read.All, SecurityIncident.Read.All, " "IdentityRiskyUser.Read.All — admin consent — then re-run this script.", file=sys.stderr, ) _merge_google_workspace_licenses_into_payload(payload) out_path = write_json_snapshot(OUT, payload) print(f"Wrote {out_path}") if __name__ == "__main__": main()