#!/usr/bin/env python3
# Ghostbuster for Entra ID: find ghost accounts, risky access and license waste in Microsoft Entra ID.
# Created by Jens Naterman at Adcyma. Latest version: https://adcyma.com/tools/ghostbuster-entra
#
# SPDX-License-Identifier: MIT
#
# MIT License
#
# Copyright (c) 2026 Jens Naterman, Adcyma
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""
Ghostbuster for Entra ID: find ghost accounts, risky access and license waste
in a Microsoft Entra ID (Azure AD) tenant.
It answers four questions about your tenant:
1. Who has an account, and which of them have admin rights?
2. Are they active? (sign-in activity, or Microsoft 365 usage)
3. Are they still with you? (with --mapping, a CSV from HR)
4. What does the leftover access cost and risk? (licenses, apps, secrets)
Read-only: it never changes anything in Entra ID. It signs you in through your
browser with read-only Microsoft Graph permissions, talks only to Microsoft,
and writes its results to a local folder. Suggested fixes are written as
commented-out Microsoft Graph PowerShell commands for you to review.
Needs Python 3.8+ and nothing else.
python ghostbuster-entra.py
python ghostbuster-entra.py --tenant contoso.com --days 90
python ghostbuster-entra.py --mapping people.csv
Created by Jens Naterman at Adcyma. Free to use, change and share under the MIT
license above, as long as the copyright notice stays in.
"""
from __future__ import annotations
import argparse
import base64
import csv
import datetime as dt
import hashlib
import html
import http.server
import io
import json
import os
import re
import secrets
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
from collections import Counter, defaultdict
VERSION = "0.1.0"
AUTHOR = "Jens Naterman, Adcyma"
HOME_URL = "https://adcyma.com/tools/ghostbuster-entra"
TOOL_NAME = "Ghostbuster for Entra ID"
GRAPH = "https://graph.microsoft.com/v1.0"
GRAPH_BETA = "https://graph.microsoft.com/beta"
LOGIN = "https://login.microsoftonline.com"
# Public client used by Microsoft Graph PowerShell (Connect-MgGraph). Admins can
# point the script at their own app registration instead with --client-id.
DEFAULT_CLIENT_ID = "14d82eec-204b-4c2f-b7e8-296a70dab67e"
SCOPES = [
"Directory.Read.All", # users, groups, roles, apps, licenses
"AuditLog.Read.All", # sign-in activity and MFA registration
"Reports.Read.All", # Microsoft 365 usage reports (fallback activity)
"Policy.Read.All", # security defaults and Conditional Access
"RoleManagement.Read.Directory", # role assignments, including PIM eligibility
]
SEVERITIES = ("critical", "high", "medium", "low", "info")
SEV_COLOR = {"critical": "1;31", "high": "31", "medium": "33", "low": "36", "info": "2"}
VERDICT_ORDER = {"departed": 0, "dormant": 1, "never": 2, "pending": 3, "leftovers": 4,
"unknown": 5, "service": 6, "disabled": 7, "active": 8}
LEFT_VALUES = {"left", "departed", "terminated", "offboarded", "inactive", "false", "no", "0", "slutat"}
ACTIVE_VALUES = {"active", "employed", "true", "yes", "1", "aktiv"}
SERVICE_VALUES = {"service", "emergency", "break-glass", "breakglass", "shared", "system", "tjänst"}
MICROSOFT_TENANT = "f8cdef31-a31e-4b4a-93e4-5f571e91255a"
GRAPH_APP_ID = "00000003-0000-0000-c000-000000000000"
EXCHANGE_APP_ID = "00000002-0000-0ff1-ce00-000000000000"
SHAREPOINT_APP_ID = "00000003-0000-0ff1-ce00-000000000000"
GLOBAL_ADMIN = "62e90394-69f5-4237-9190-012177145e10"
SUCCESS_TRACKED_FROM = dt.datetime(2023, 12, 1, tzinfo=dt.timezone.utc)
DIRECTORY_SYNC_ROLE = "d29b2b05-8046-44ba-8758-1e26182fcf32"
# Built-in roles that can take over the tenant, its users or its data.
PRIVILEGED_ROLES = {
GLOBAL_ADMIN,
"e8611ab8-c189-46e8-94e1-60213ab1f814", # Privileged Role Administrator
"7be44c8a-adaf-4e2a-84d6-ab2649e08a13", # Privileged Authentication Administrator
"194ae4cb-b126-40b2-bd5b-6091b380977d", # Security Administrator
"29232cdf-9323-42fd-ade2-1d097af3e4de", # Exchange Administrator
"f28a1f50-f6e7-4571-818b-6a12f2af6b6c", # SharePoint Administrator
"fe930be7-5e62-47db-91af-98c3a49a38b1", # User Administrator
"9b895d92-2cd3-44c7-9d02-a6ac2d5ea5c3", # Application Administrator
"158c047a-c907-4556-b7ef-446551a6b5f7", # Cloud Application Administrator
"b1be1c3e-b65d-4f19-8427-f6fa0d97feb9", # Conditional Access Administrator
"c4e39bd9-1100-46d3-8c65-fb160da0071f", # Authentication Administrator
"729827e3-9c14-49f7-bb1b-9608f156bbb8", # Helpdesk Administrator
"3a2c62db-5318-420d-8d74-23affee5d9d5", # Intune Administrator
"8ac3fc64-6eca-42ea-9e69-59f4c7b60eb2", # Hybrid Identity Administrator
}
# Application permissions that let an app take over the tenant (high) or read everyone's data (medium).
RISKY_APP_PERMISSIONS = {
GRAPH_APP_ID: {
"RoleManagement.ReadWrite.Directory": ("high", "can make any account a Global Administrator"),
"AppRoleAssignment.ReadWrite.All": ("high", "can grant itself any other permission"),
"Application.ReadWrite.All": ("high", "can add credentials to any app and act as it"),
"Directory.ReadWrite.All": ("high", "can change almost everything in the directory"),
"UserAuthenticationMethod.ReadWrite.All": ("high", "can change anyone's sign-in methods"),
"Policy.ReadWrite.ConditionalAccess": ("high", "can switch off Conditional Access"),
"User.ReadWrite.All": ("medium", "can change any user"),
"Group.ReadWrite.All": ("medium", "can change any group"),
"GroupMember.ReadWrite.All": ("medium", "can add anyone to any group"),
"Mail.ReadWrite": ("medium", "can read and change every mailbox"),
"Mail.Read": ("medium", "can read every mailbox"),
"Mail.Send": ("medium", "can send mail as anyone"),
"Files.ReadWrite.All": ("medium", "can read and change every file"),
"Files.Read.All": ("medium", "can read every file"),
"Sites.FullControl.All": ("medium", "has full control of every SharePoint site"),
"Sites.ReadWrite.All": ("medium", "can change every SharePoint site"),
"Sites.Read.All": ("medium", "can read every SharePoint site"),
"Chat.Read.All": ("medium", "can read every Teams chat"),
"ChannelMessage.Read.All": ("medium", "can read every Teams channel message"),
},
EXCHANGE_APP_ID: {
"full_access_as_app": ("medium", "can read and change every mailbox"),
"Mail.ReadWrite": ("medium", "can read and change every mailbox"),
"Mail.Read": ("medium", "can read every mailbox"),
"Mail.Send": ("medium", "can send mail as anyone"),
},
SHAREPOINT_APP_ID: {
"Sites.FullControl.All": ("medium", "has full control of every SharePoint site"),
"Sites.ReadWrite.All": ("medium", "can change every SharePoint site"),
"Sites.Read.All": ("medium", "can read every SharePoint site"),
},
}
# Friendly names for common license SKUs; anything else shows its part number.
SKU_NAMES = {
"SPB": "Microsoft 365 Business Premium",
"O365_BUSINESS_PREMIUM": "Microsoft 365 Business Standard",
"O365_BUSINESS_ESSENTIALS": "Microsoft 365 Business Basic",
"O365_BUSINESS": "Microsoft 365 Apps for business",
"SPE_E3": "Microsoft 365 E3",
"SPE_E5": "Microsoft 365 E5",
"SPE_F1": "Microsoft 365 F3",
"M365_F1": "Microsoft 365 F1",
"ENTERPRISEPACK": "Office 365 E3",
"ENTERPRISEPREMIUM": "Office 365 E5",
"STANDARDPACK": "Office 365 E1",
"DESKLESSPACK": "Office 365 F3",
"OFFICESUBSCRIPTION": "Microsoft 365 Apps for enterprise",
"EMS": "Enterprise Mobility + Security E3",
"EMSPREMIUM": "Enterprise Mobility + Security E5",
"AAD_PREMIUM": "Microsoft Entra ID P1",
"AAD_PREMIUM_P2": "Microsoft Entra ID P2",
"Microsoft_Entra_ID_Governance": "Microsoft Entra ID Governance",
"EXCHANGESTANDARD": "Exchange Online (Plan 1)",
"EXCHANGEENTERPRISE": "Exchange Online (Plan 2)",
"EXCHANGEARCHIVE_ADDON": "Exchange Online Archiving",
"POWER_BI_PRO": "Power BI Pro",
"PBI_PREMIUM_PER_USER": "Power BI Premium Per User",
"PROJECTPREMIUM": "Project Plan 5",
"PROJECTPROFESSIONAL": "Project Plan 3",
"VISIOCLIENT": "Visio Plan 2",
"VISIO_PLAN1_DEPT": "Visio Plan 1",
"MCOEV": "Teams Phone Standard",
"MCOMEETADV": "Microsoft 365 Audio Conferencing",
"Microsoft_Teams_Premium": "Teams Premium",
"Microsoft_365_Copilot": "Microsoft 365 Copilot",
"INTUNE_A": "Microsoft Intune Plan 1",
"WIN_DEF_ATP": "Microsoft Defender for Endpoint P2",
"DYN365_ENTERPRISE_SALES": "Dynamics 365 Sales Enterprise",
}
# Free, trial and self-service SKUs: holding one costs nothing, so they never count as waste.
FREE_SKUS = {
"FLOW_FREE", "POWER_BI_STANDARD", "TEAMS_EXPLORATORY", "WINDOWS_STORE", "STREAM",
"POWERAPPS_VIRAL", "POWERAPPS_DEV", "CCIBOTS_PRIVPREV_VIRAL", "MICROSOFT_BUSINESS_CENTER",
"RIGHTSMANAGEMENT_ADHOC", "PHONESYSTEM_VIRTUALUSER", "MCOPSTNC", "FORMS_PRO",
"TEAMS_FREE", "MEETING_ROOM_FREE", "DYN365_ENTERPRISE_P1_IW", "PROJECT_MADEIRA_PREVIEW_IW_SKU",
"POWERAPPS_PER_APP_IW", "Microsoft_Teams_Rooms_Basic", "VIVA", "Power_Pages_vTrial_for_Makers",
}
# ---------------------------------------------------------------- shared report kit
# Console helpers, dates, CSV safety and the Adcyma report styling are shared with
# Ghostbuster for GitHub, so every Ghostbuster report looks the same.
USE_COLOR = False
# ---------------------------------------------------------------- console
def _setup_console():
global USE_COLOR
for stream in (sys.stdout, sys.stderr):
try:
stream.reconfigure(errors="replace")
except (AttributeError, ValueError):
pass
if os.name == "nt":
os.system("") # switches older Windows consoles into ANSI mode
USE_COLOR = sys.stdout.isatty() and "NO_COLOR" not in os.environ
def paint(text, code):
return f"\033[{code}m{text}\033[0m" if USE_COLOR else text
def say(msg=""):
print(msg, flush=True)
def step(msg):
say(f"{paint('>', '36')} {msg}")
def note(msg):
say(f" {paint(msg, '2')}")
def warn(msg):
say(f"{paint('!', '33')} {msg}")
def die(msg):
print(f"{paint('x', '31')} {msg}", file=sys.stderr, flush=True)
sys.exit(1)
def ask_yes_no(question, default=False):
if not sys.stdin.isatty():
return False
suffix = " [Y/n] " if default else " [y/N] "
try:
answer = input(question + suffix).strip().lower()
except EOFError:
return False
if not answer:
return default
return answer in ("y", "yes", "j", "ja")
class Progress:
def __init__(self, label, total):
self.label, self.total, self.done = label, total, 0
self.lock = threading.Lock()
self.tty = sys.stdout.isatty()
def tick(self):
with self.lock:
self.done += 1
if self.tty:
print(f"\r {self.label}: {self.done}/{self.total}", end="", flush=True)
def finish(self):
if self.tty and self.total:
print(flush=True)
def utcnow():
return dt.datetime.now(dt.timezone.utc)
def parse_ts(value):
if value in (None, ""):
return None
if isinstance(value, (int, float)): # audit log uses epoch milliseconds
return dt.datetime.fromtimestamp(value / 1000, dt.timezone.utc)
text = str(value).replace("Z", "+00:00")
text = re.sub(r"\.(\d{1,6})\d*(?=[+-])", lambda m: "." + m.group(1).ljust(6, "0"), text)
try:
ts = dt.datetime.fromisoformat(text)
except ValueError:
return None
return ts if ts.tzinfo else ts.replace(tzinfo=dt.timezone.utc)
def later(a, b):
if a is None:
return b
if b is None:
return a
return max(a, b)
def fmt_date(ts):
return ts.strftime("%Y-%m-%d") if ts else ""
def iso(ts):
return ts.isoformat().replace("+00:00", "Z") if ts else None
def csv_safe(value):
"""Stops spreadsheet apps from running cell values (names come from GitHub users) as formulas."""
text = "" if value is None else str(value)
return "'" + text if text[:1] in ("=", "+", "-", "@", "\t", "\r") else text
def write_csv(path, rows, columns):
with open(path, "w", newline="", encoding="utf-8-sig") as fh:
writer = csv.writer(fh)
writer.writerow(columns)
for row in rows:
writer.writerow([csv_safe(row.get(c)) for c in columns])
LIGHT_TOKENS = ("--bg:#f5f6f3;--paper:#fff;--surface-2:#f1f3ef;--text:#1b1d1b;--muted:#5e625d;--border:#e1e4de;"
"--accent:#365b3f;--accent-soft:#e6efe8;--brand-light:#74ad82;--slate:#3d5a80;--slate-soft:#e6edf6;"
"--ok:#1b7745;--ok-soft:#e2f2e8;--warn:#985800;--warn-soft:#fbefd9;"
"--bad:#b42318;--bad-soft:#fbe5e2;--crit:#6b0f0b;--neutral-soft:#ecedea;--code:#f1f3ef")
DARK_TOKENS = ("--bg:#111311;--paper:#1a1c1a;--surface-2:#232622;--text:#eceeea;--muted:#a1a59f;--border:#323630;"
"--accent:#74ad82;--accent-soft:#1f3326;--brand-light:#74ad82;--slate:#9db8dc;--slate-soft:#1e2a3a;"
"--ok:#6fd39a;--ok-soft:#173325;--warn:#f0b85a;--warn-soft:#3a2c12;"
"--bad:#ff8a80;--bad-soft:#3d1c19;--crit:#ffb4ab;--neutral-soft:#292b28;--code:#292b28")
REPORT_CSS = """
:root{%LIGHT%}
@media (prefers-color-scheme:dark){:root:not([data-theme="light"]){%DARK%}}
:root[data-theme="dark"]{%DARK%}
*{box-sizing:border-box}
html{-webkit-text-size-adjust:100%}
body{margin:0;background:var(--bg);color:var(--text);font:15px/1.55 Inter,-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,"Helvetica Neue",Arial,sans-serif}
.page{max-width:1080px;margin:24px auto 64px;padding:0 16px}
.sheet{background:var(--paper);border:1px solid var(--border);border-radius:16px;padding:0 44px 40px;overflow:hidden}
.brandbar{height:5px;margin:0 -44px 28px;background:linear-gradient(90deg,var(--accent),var(--brand-light))}
.top{display:flex;justify-content:space-between;align-items:center;gap:16px;padding-bottom:18px;border-bottom:1px solid var(--border)}
.top a{display:block;color:var(--accent)}
.adcyma-logo{display:block;height:26px;width:auto;fill:currentColor}
.eyebrow{display:flex;align-items:center;gap:8px;margin-top:26px;color:var(--accent);font-weight:650;font-size:.82rem;letter-spacing:.08em;text-transform:uppercase}
.eyebrow svg{width:24px;height:24px}
.eyebrow .sep{color:var(--muted);font-weight:400}
.eyebrow .what{color:var(--muted);font-weight:600}
h1{font-size:clamp(1.7rem,4.4vw,2.4rem);margin:6px 0 0;letter-spacing:-.02em;line-height:1.15;overflow-wrap:anywhere}
.btn{font:inherit;font-weight:600;font-size:.88rem;padding:8px 16px;border-radius:10px;border:1px solid var(--accent);background:var(--accent);color:var(--paper);cursor:pointer;white-space:nowrap}
.meta{display:grid;grid-template-columns:repeat(4,minmax(0,1fr));gap:12px;margin:22px 0 0;padding:16px 0;border-top:1px solid var(--border);border-bottom:1px solid var(--border)}
.meta dt{font-size:.72rem;text-transform:uppercase;letter-spacing:.06em;color:var(--muted)}
.meta dd{margin:2px 0 0;font-weight:600}
nav.toc{display:flex;flex-wrap:wrap;gap:6px 16px;font-size:.86rem;margin:14px 0 0}
nav.toc a{color:var(--muted);text-decoration:none}
nav.toc a:hover{color:var(--accent);text-decoration:underline}
section{margin-top:40px}
.h{display:flex;align-items:baseline;gap:10px;border-bottom:2px solid var(--text);padding-bottom:6px;margin-bottom:12px}
.h .num{font-weight:700;color:var(--accent);font-variant-numeric:tabular-nums;white-space:nowrap}
h2{font-size:1.25rem;margin:0;line-height:1.3}
.h .count{margin-left:auto;color:var(--muted);font-size:.85rem;font-variant-numeric:tabular-nums}
h3{font-size:.98rem;margin:24px 0 8px}
p{margin:0 0 12px}
.intro{color:var(--muted);max-width:76ch}
.lead{font-size:1.08rem;max-width:78ch}
.tiles{display:grid;grid-template-columns:repeat(5,minmax(0,1fr));gap:10px;margin:18px 0 4px}
.tile{border:1px solid var(--border);border-radius:12px;padding:12px 14px}
.tile .n{font-size:1.7rem;font-weight:750;line-height:1.1;font-variant-numeric:tabular-nums}
.tile .l{font-size:.8rem;color:var(--muted)}
.tile.alert .n{color:var(--bad)}
.bar{display:flex;height:14px;border-radius:7px;overflow:hidden;background:var(--surface-2);margin:8px 0}
.seg{display:block;height:100%}
.seg+.seg{border-left:2px solid var(--paper)}
.legend{display:flex;flex-wrap:wrap;gap:6px 18px;font-size:.84rem;color:var(--muted)}
.legend i{display:inline-block;width:10px;height:10px;border-radius:3px;margin-right:6px;vertical-align:-1px}
.legend strong{color:var(--text);margin-left:4px}
.v-departed{background:var(--crit)}.v-dormant{background:var(--bad)}.v-over-privileged{background:var(--slate)}.v-unknown{background:var(--muted)}.v-active{background:var(--ok)}
ol.actions{list-style:none;padding:0;margin:8px 0 0;counter-reset:a;border:1px solid var(--border);border-radius:12px}
ol.actions li{counter-increment:a;display:grid;grid-template-columns:24px 92px 1fr auto;gap:10px;align-items:center;padding:10px 14px;border-bottom:1px solid var(--border)}
ol.actions li:last-child{border-bottom:none}
ol.actions .pill{justify-self:start}
ol.actions li::before{content:counter(a);font-weight:700;color:var(--muted);font-variant-numeric:tabular-nums}
ol.actions a{font-size:.82rem;color:var(--accent);text-decoration:none;white-space:nowrap}
.wrap{overflow-x:auto;border:1px solid var(--border);border-radius:12px}
table{border-collapse:collapse;width:100%;font-size:.86rem}
th,td{text-align:left;vertical-align:top;padding:8px 12px;border-bottom:1px solid var(--border)}
th{background:var(--surface-2);color:var(--muted);font-size:.7rem;text-transform:uppercase;letter-spacing:.05em;white-space:nowrap}
tbody tr:last-child td{border-bottom:none}
td strong{font-weight:650}
.sub{color:var(--muted);font-size:.78rem}
.sub-inline{color:var(--muted);font-size:.78rem}
.muted{color:var(--muted)}
.empty{color:var(--muted);font-style:italic}
.pill{display:inline-block;font-size:.7rem;font-weight:700;padding:2px 8px;border-radius:999px;white-space:nowrap;text-transform:uppercase;letter-spacing:.03em}
.critical,.departed{background:var(--bad);color:var(--paper)}
.high,.dormant,.off,.fix,.skipped{background:var(--bad-soft);color:var(--bad)}
.medium,.sms,.partial{background:var(--warn-soft);color:var(--warn)}
.low,.over-privileged{background:var(--slate-soft);color:var(--slate)}
.info,.unknown,.not-available,.not-requested{background:var(--neutral-soft);color:var(--muted)}
.active,.on,.good,.checked{background:var(--ok-soft);color:var(--ok)}
.callout{border:1px solid var(--border);background:var(--surface-2);border-radius:12px;padding:14px 18px}
ul.limits{margin:0;padding-left:20px}
ul.limits li{margin:4px 0}
details.fixes{margin-top:10px}
details.fixes summary{cursor:pointer;color:var(--accent);font-size:.84rem}
pre{font:12px/1.5 ui-monospace,SFMono-Regular,Consolas,monospace;background:var(--code);padding:10px 12px;border-radius:8px;overflow-x:auto;margin:8px 0}
code{font:12px/1.4 ui-monospace,SFMono-Regular,Consolas,monospace}
footer.about{margin-top:52px}
.cta{display:grid;grid-template-columns:1fr auto;gap:20px 32px;align-items:center;padding:26px 28px;border-radius:14px;background:var(--accent-soft);border:1px solid var(--border)}
.cta .adcyma-logo{height:24px;color:var(--accent);margin-bottom:14px}
.cta h2{font-size:1.35rem;margin:0 0 6px}
.cta p{margin:0;max-width:62ch;color:var(--muted)}
.cta-btn{display:inline-block;text-decoration:none;font-size:.95rem;padding:11px 20px}
.cta-btn span{margin-left:6px}
.fineprint{margin:14px 4px 0;color:var(--muted);font-size:.8rem}
.fineprint a{color:var(--accent)}
@media screen and (max-width:760px){
.sheet{padding:0 16px 22px;border-radius:12px}
.page{margin-top:12px}
.cta{grid-template-columns:1fr;padding:22px 20px}
.brandbar{margin:0 -16px 20px}
.meta{grid-template-columns:1fr 1fr}
.tiles{grid-template-columns:1fr 1fr}
ol.actions li{display:flex;flex-wrap:wrap;gap:6px 10px}
th,td{padding:7px 9px}
}
@page{size:A4;margin:14mm 12mm 16mm;
@bottom-left{content:"Ghostbuster by Adcyma · adcyma.com/tools";font:8pt -apple-system,"Segoe UI",Roboto,Arial,sans-serif;color:#777}
@bottom-right{content:"%ORG% · page " counter(page) " of " counter(pages);font:8pt -apple-system,"Segoe UI",Roboto,Arial,sans-serif;color:#777}}
@media print{
:root,:root:not([data-theme="light"]),:root[data-theme="dark"]{%LIGHT%}
*{-webkit-print-color-adjust:exact;print-color-adjust:exact}
body{background:#fff;font-size:10pt}
.page{max-width:none;margin:0;padding:0}
.sheet{border:none;border-radius:0;padding:0;overflow:visible}
.brandbar{margin:0 0 18px}
footer.about{break-inside:avoid}
.cta-btn span{display:none}
.no-print{display:none!important}
section{margin-top:24px}
.h,.intro,h3{break-after:avoid}
tr,.tile,ol.actions li,.callout,.meta{break-inside:avoid}
thead{display:table-header-group}
.wrap{overflow:visible}
table{font-size:8.5pt}
th,td{padding:5px 8px}
.appendix-start{break-before:page}
a{color:inherit;text-decoration:none}
}
"""
# Adcyma logo from frontend/public/images, coloured through currentColor.
ADCYMA_LOGO = (
'')
GHOST_SVG = """"""
# ---------------------------------------------------------------- sign-in
class AuthError(Exception):
pass
def _post_form(url, data):
body = urllib.parse.urlencode(data).encode()
req = urllib.request.Request(url, data=body, method="POST",
headers={"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": f"ghostbuster-entra/{VERSION}"})
try:
with urllib.request.urlopen(req, timeout=60) as resp:
return json.loads(resp.read() or b"{}")
except urllib.error.HTTPError as err:
try:
return json.loads(err.read() or b"{}")
except ValueError:
raise AuthError(f"HTTP {err.code} from Microsoft sign-in") from None
def _token_error(payload):
code = payload.get("error", "unknown_error")
text = (payload.get("error_description") or "").split("\r\n")[0]
hints = {
"AADSTS65001": "An administrator has to approve the read-only permissions once. Ask a Global Administrator, "
"Privileged Role Administrator or Cloud Application Administrator to run Ghostbuster first "
"and tick 'Consent on behalf of your organization'.",
"AADSTS90094": "An administrator has to approve the read-only permissions once. Ask a Global Administrator "
"to run Ghostbuster first and tick 'Consent on behalf of your organization'.",
"AADSTS53003": "A Conditional Access policy blocked this sign-in. Try the browser sign-in (drop --device-code) "
"or run it from a managed device.",
"AADSTS50105": "Your account isn't allowed to use this app in your tenant (user assignment is required).",
"AADSTS530035": "Security defaults block sign-in with a code. Drop --device-code to sign in through the browser.",
}
for key, hint in hints.items():
if key in text:
return f"{text}\n {hint}"
return f"{code}: {text}" if text else code
SIGNED_IN_PAGE = """Ghostbuster sign-in
%TITLE%
%TEXT%
"""
def browser_sign_in(tenant, client_id, scope, timeout=300):
verifier = secrets.token_urlsafe(64)
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
state = secrets.token_urlsafe(16)
result = {}
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
query = dict(urllib.parse.parse_qsl(urllib.parse.urlparse(self.path).query))
if "code" not in query and "error" not in query:
self.send_response(404)
self.end_headers()
return
ok = query.get("state") == state and "code" in query
if ok:
result["code"] = query["code"]
title, text = "You're signed in", "Ghostbuster is reading your tenant. You can close this tab and go back to the terminal."
else:
result["error"] = query.get("error_description") or query.get("error") or "state mismatch"
title, text = "Sign-in didn't work", html.escape(result["error"].split("\r\n")[0])
page = SIGNED_IN_PAGE.replace("%TITLE%", title).replace("%TEXT%", text).encode()
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(page)))
self.end_headers()
self.wfile.write(page)
def log_message(self, *args):
pass
server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
server.timeout = 1
redirect_uri = f"http://localhost:{server.server_address[1]}"
url = f"{LOGIN}/{tenant}/oauth2/v2.0/authorize?" + urllib.parse.urlencode({
"client_id": client_id, "response_type": "code", "redirect_uri": redirect_uri, "response_mode": "query",
"scope": scope, "state": state, "code_challenge": challenge, "code_challenge_method": "S256",
"prompt": "select_account",
})
note("A browser window opens for the Microsoft sign-in. If it doesn't, open this link:")
say(f" {url}")
webbrowser.open(url)
deadline = time.time() + timeout
try:
while not result and time.time() < deadline:
server.handle_request()
finally:
server.server_close()
if "error" in result:
raise AuthError(result["error"].split("\r\n")[0])
if "code" not in result:
raise AuthError("No sign-in within 5 minutes.")
payload = _post_form(f"{LOGIN}/{tenant}/oauth2/v2.0/token", {
"client_id": client_id, "grant_type": "authorization_code", "code": result["code"],
"redirect_uri": redirect_uri, "code_verifier": verifier, "scope": scope,
})
if "access_token" not in payload:
raise AuthError(_token_error(payload))
return payload
def device_code_sign_in(tenant, client_id, scope):
start = _post_form(f"{LOGIN}/{tenant}/oauth2/v2.0/devicecode", {"client_id": client_id, "scope": scope})
if "device_code" not in start:
raise AuthError(_token_error(start))
say(f" {start.get('message') or ('Go to ' + start['verification_uri'] + ' and enter ' + start['user_code'])}")
interval = int(start.get("interval", 5))
deadline = time.time() + int(start.get("expires_in", 900))
while time.time() < deadline:
time.sleep(interval)
payload = _post_form(f"{LOGIN}/{tenant}/oauth2/v2.0/token", {
"client_id": client_id, "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
"device_code": start["device_code"],
})
if "access_token" in payload:
return payload
error = payload.get("error")
if error == "authorization_pending":
continue
if error == "slow_down":
interval += 5
continue
raise AuthError(_token_error(payload))
raise AuthError("The sign-in code expired.")
def sign_in(args):
scope = " ".join(f"https://graph.microsoft.com/{s}" for s in SCOPES)
step("Signing in to Microsoft (read-only permissions)")
if args.device_code:
return device_code_sign_in(args.tenant, args.client_id, scope)
try:
return browser_sign_in(args.tenant, args.client_id, scope)
except OSError as err:
warn(f"The browser sign-in couldn't start ({err}). Using a sign-in code instead.")
return device_code_sign_in(args.tenant, args.client_id, scope)
def granted_scopes(token_payload):
raw = token_payload.get("scope") or ""
return {s.rsplit("/", 1)[-1] for s in raw.split()}
# ---------------------------------------------------------------- Microsoft Graph client
class GraphError(Exception):
def __init__(self, status, code, message, url=""):
super().__init__(f"HTTP {status} {code}: {message}")
self.status, self.code, self.message, self.url = status, code, message, url
def reason(self):
if self.code == "Authentication_RequestFromNonPremiumTenantOrB2CTenant":
return "needs Microsoft Entra ID P1 or P2"
if self.code == "AadPremiumLicenseRequired" or "P2" in (self.message or "") and self.status == 400:
return "needs Microsoft Entra ID P2 or ID Governance"
if self.status == 403:
return "no permission (needs a role such as Global Reader, and the read-only consent)"
if self.status == 404:
return "not available in this tenant"
if self.status == 0:
return f"network error: {self.message}"
return f"{self.code}: {self.message}"[:200]
class Graph:
def __init__(self, token):
self._token = token
self._lock = threading.Lock()
self.calls = 0
def _url(self, path, params=None, beta=False):
url = path if path.startswith("http") else (GRAPH_BETA if beta else GRAPH) + path
if params:
url += ("&" if "?" in url else "?") + urllib.parse.urlencode(params, safe="$,()'=")
return url
def request(self, url, headers=None, raw=False):
all_headers = {"Authorization": f"Bearer {self._token}", "Accept": "application/json",
"User-Agent": f"ghostbuster-entra/{VERSION}"}
all_headers.update(headers or {})
for attempt in range(8):
req = urllib.request.Request(url, headers=all_headers)
try:
with urllib.request.urlopen(req, timeout=120) as resp:
body = resp.read()
with self._lock:
self.calls += 1
if raw:
return body
return json.loads(body) if body else {}
except urllib.error.HTTPError as err:
body = err.read()
with self._lock:
self.calls += 1
if err.code in (429, 503, 504) and attempt < 7:
wait = err.headers.get("Retry-After") if err.headers else None
wait = int(wait) if wait and str(wait).isdigit() else min(60, 2 ** (attempt + 1))
if wait > 30:
say("")
warn(f"Microsoft Graph asked us to slow down; waiting {wait}s.")
time.sleep(min(wait, 300))
continue
try:
error = json.loads(body).get("error") or {}
except ValueError:
error = {}
raise GraphError(err.code, error.get("code", "Error"), error.get("message", str(err.reason)), url) from None
except (urllib.error.URLError, OSError) as err:
if attempt < 3:
time.sleep(2 ** attempt)
continue
raise GraphError(0, "NetworkError", str(getattr(err, "reason", err)), url) from None
raise GraphError(429, "TooManyRequests", "still throttled after several retries", url)
def get(self, path, params=None, beta=False, headers=None):
return self.request(self._url(path, params, beta), headers)
def paged(self, path, params=None, beta=False, headers=None):
url = self._url(path, params, beta)
while url:
data = self.request(url, headers)
for item in data.get("value") or []:
yield item
url = data.get("@odata.nextLink")
def list(self, path, params=None, beta=False, headers=None):
return list(self.paged(path, params, beta, headers))
def text(self, path, params=None):
"""Reports answer with a redirect to a pre-signed download link, which must not get the Graph token."""
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *args, **kwargs):
return None
opener = urllib.request.build_opener(NoRedirect)
req = urllib.request.Request(self._url(path, params), headers={
"Authorization": f"Bearer {self._token}", "User-Agent": f"ghostbuster-entra/{VERSION}"})
try:
with opener.open(req, timeout=120) as resp:
body = resp.read()
except urllib.error.HTTPError as err:
location = err.headers.get("Location") if err.headers else None
if err.code not in (301, 302, 303, 307, 308) or not location:
try:
error = json.loads(err.read()).get("error") or {}
except ValueError:
error = {}
raise GraphError(err.code, error.get("code", "Error"), error.get("message", str(err.reason)), req.full_url) from None
with urllib.request.urlopen(urllib.request.Request(location), timeout=120) as resp:
body = resp.read()
with self._lock:
self.calls += 1
return body.decode("utf-8-sig", errors="replace")
class Coverage:
def __init__(self):
self.rows = []
def add(self, check, status, detail=""):
self.rows.append({"check": check, "status": status, "detail": detail})
def guarded(cov, check, fn, describe=None):
try:
result = fn()
except GraphError as err:
cov.add(check, "skipped", err.reason())
return None
cov.add(check, "checked", describe(result) if describe else "")
return result
# ---------------------------------------------------------------- collection
USER_FIELDS = ("id,displayName,userPrincipalName,mail,accountEnabled,userType,createdDateTime,externalUserState,"
"externalUserStateChangeDateTime,assignedLicenses,onPremisesSyncEnabled,onPremisesSamAccountName,"
"department,jobTitle,employeeType,companyName,licenseAssignmentStates")
def collect(graph, args, now, scopes):
since = now - dt.timedelta(days=args.days)
cov = Coverage()
missing = [s for s in SCOPES if s not in scopes]
if scopes and missing:
warn(f"The sign-in didn't grant {', '.join(missing)}. Checks that need them are listed as skipped.")
step("Reading the tenant and its licenses")
org = (graph.get("/organization", {"$select": "id,displayName,verifiedDomains,onPremisesSyncEnabled,createdDateTime"})
.get("value") or [{}])[0]
me = graph.get("/me", {"$select": "displayName,userPrincipalName"})
skus = guarded(cov, "Licenses", lambda: graph.list("/subscribedSkus"), lambda r: f"{len(r)} subscriptions") or []
plans = {p.get("servicePlanName") for s in skus if s.get("capabilityStatus") == "Enabled"
for p in s.get("servicePlans") or []}
entra_plan = "P2" if "AAD_PREMIUM_P2" in plans else "P1" if "AAD_PREMIUM" in plans else "Free"
# Sign-in activity is only returned to tenants with Entra ID P1 or P2.
signin_available, signin_reason = False, ""
try:
probe = graph.get("/users", {"$top": "1", "$select": "id,signInActivity"})
signin_available = bool(probe.get("value")) and "signInActivity" in probe["value"][0]
except GraphError as err:
signin_reason = err.reason()
step("Reading users")
select = USER_FIELDS + (",signInActivity" if signin_available else "")
users = graph.list("/users", {"$select": select, "$top": str(args.page_size if signin_available else 999)})
cov.add("Users and guests", "checked", f"{len(users)} accounts")
if signin_available:
cov.add("Sign-in activity", "checked", "last successful, interactive and background sign-ins")
else:
cov.add("Sign-in activity", "skipped", signin_reason or "needs Microsoft Entra ID P1 or P2")
usage, usage_note = {}, ""
if not signin_available or args.usage_report:
step("Reading the Microsoft 365 usage report")
usage, usage_note = load_usage(graph)
cov.add("Microsoft 365 usage report", "checked" if usage else "skipped", usage_note)
step("Reading admin roles")
roles = load_roles(graph, cov)
step("Reading sign-in security")
security = load_security(graph, cov, entra_plan)
step("Reading MFA registration")
mfa = None
try:
mfa = {r["id"]: r for r in graph.paged("/reports/authenticationMethods/userRegistrationDetails",
{"$top": "999"})}
cov.add("MFA registration", "checked", f"{len(mfa)} accounts")
except GraphError as err:
cov.add("MFA registration", "skipped", err.reason())
step("Reading apps, permissions and secrets")
apps = guarded(cov, "App registrations and secrets",
lambda: [a for a in graph.paged("/applications", {"$select": "id,appId,displayName,createdDateTime,"
"passwordCredentials,keyCredentials", "$top": "999"})
if "agentIdentityBlueprint" not in (a.get("@odata.type") or "")],
lambda r: f"{len(r)} apps") or []
grants = load_app_permissions(graph, cov, org.get("id"))
return {
"org": org, "me": me, "now": now, "since": since, "days": args.days, "entra_plan": entra_plan,
"skus": skus, "users": users, "signin_available": signin_available, "usage": usage,
"roles": roles, "security": security, "mfa": mfa, "apps": apps, "app_grants": grants,
"coverage": cov.rows, "scopes": sorted(scopes),
}
def load_usage(graph):
"""Microsoft 365 usage: per-user last activity in Exchange, OneDrive, SharePoint and Teams, no P1 needed."""
try:
text = graph.text("/reports/getOffice365ActiveUserDetail(period='D180')")
except GraphError as err:
return {}, err.reason()
rows = list(csv.DictReader(io.StringIO(text)))
if not rows:
return {}, "the report was empty"
if not any("@" in (r.get("User Principal Name") or "") for r in rows):
return {}, ("user names are hidden. Either the tenant conceals them (a Global Administrator can change that "
"under Microsoft 365 admin center > Settings > Org settings > Reports), or your account only sees "
"totals: per-user usage needs the Reports Reader role")
usage = {}
for row in rows:
upn = (row.get("User Principal Name") or "").lower()
last = None
for key, value in row.items():
if key and key.endswith("Last Activity Date") and value:
last = later(last, parse_ts(value + "T00:00:00Z"))
if upn:
usage[upn] = last
return usage, f"{len(rows)} licensed users, last 180 days"
def load_roles(graph, cov):
definitions = {}
try:
for d in graph.paged("/roleManagement/directory/roleDefinitions", {"$select": "id,displayName,templateId"}):
definitions[d["id"]] = d
except GraphError as err:
cov.add("Admin roles", "skipped", err.reason())
return {"assignments": [], "definitions": {}, "eligible_checked": False}
def expand(kind, path, beta=False):
rows = []
for a in graph.paged(path, {"$expand": "principal"}, beta=beta):
d = definitions.get(a.get("roleDefinitionId"), {})
principal = a.get("principal") or {}
ptype = (principal.get("@odata.type") or "").rsplit(".", 1)[-1]
rows.append({"id": a.get("id"), "kind": kind, "role": d.get("displayName", a.get("roleDefinitionId")),
"templateId": d.get("templateId") or a.get("roleDefinitionId"),
"principalId": a.get("principalId"), "principalType": ptype,
"principalName": principal.get("displayName") or principal.get("userPrincipalName") or "",
"scope": a.get("directoryScopeId", "/")})
return rows
assignments = []
try:
assignments = expand("active", "/roleManagement/directory/roleAssignments")
cov.add("Admin roles", "checked", f"{len(assignments)} active assignments")
except GraphError as err:
try:
by_template = {d.get("templateId"): d for d in definitions.values()}
for role in graph.paged("/directoryRoles", {"$expand": "members"}):
template = role.get("roleTemplateId")
for member in role.get("members") or []:
assignments.append({
"id": None, "kind": "active", "role": role.get("displayName"), "templateId": template,
"principalId": member.get("id"),
"principalType": (member.get("@odata.type") or "").rsplit(".", 1)[-1],
"principalName": member.get("displayName") or member.get("userPrincipalName") or "",
"scope": "/", "directoryRoleId": role.get("id"),
"roleDefinitionId": (by_template.get(template) or {}).get("id")})
cov.add("Admin roles", "checked", f"{len(assignments)} active assignments (directory roles)")
except GraphError:
cov.add("Admin roles", "skipped", err.reason())
eligible_checked = False
try:
eligible = expand("eligible", "/roleManagement/directory/roleEligibilitySchedules")
assignments += eligible
eligible_checked = True
cov.add("Eligible admin roles (PIM)", "checked", f"{len(eligible)} eligible assignments")
except GraphError as err:
cov.add("Eligible admin roles (PIM)", "not available", err.reason())
# Role-assignable groups: their members hold the role too.
group_members = {}
for a in assignments:
if a["principalType"] == "group" and a["principalId"] not in group_members:
try:
group_members[a["principalId"]] = [m["id"] for m in graph.paged(
f"/groups/{a['principalId']}/transitiveMembers", {"$select": "id"})]
except GraphError:
group_members[a["principalId"]] = []
return {"assignments": assignments, "definitions": definitions, "group_members": group_members,
"eligible_checked": eligible_checked}
def load_security(graph, cov, entra_plan):
security = {"security_defaults": None, "ca_policies": None, "ca_mfa_all": False}
try:
security["security_defaults"] = bool(graph.get("/policies/identitySecurityDefaultsEnforcementPolicy").get("isEnabled"))
cov.add("Security defaults", "checked", "on" if security["security_defaults"] else "off")
except GraphError as err:
cov.add("Security defaults", "skipped", err.reason())
try:
policies = graph.list("/identity/conditionalAccess/policies")
security["ca_policies"] = policies
for p in policies:
if p.get("state") != "enabled":
continue
grant = p.get("grantControls") or {}
needs_mfa = "mfa" in (grant.get("builtInControls") or []) or bool(grant.get("authenticationStrength"))
users = ((p.get("conditions") or {}).get("users") or {})
if needs_mfa and "All" in (users.get("includeUsers") or []):
security["ca_mfa_all"] = True
cov.add("Conditional Access", "checked", f"{len(policies)} policies")
except GraphError as err:
cov.add("Conditional Access", "not available" if entra_plan == "Free" else "skipped", err.reason())
return security
def load_app_permissions(graph, cov, tenant_id):
"""Application permissions granted on Microsoft Graph, Exchange Online and SharePoint, to any app."""
found = []
try:
for resource_app in (GRAPH_APP_ID, EXCHANGE_APP_ID, SHAREPOINT_APP_ID):
sps = graph.list("/servicePrincipals", {"$filter": f"appId eq '{resource_app}'",
"$select": "id,appId,displayName,appRoles"})
if not sps:
continue
resource = sps[0]
role_names = {r["id"]: r.get("value") for r in resource.get("appRoles") or []}
for a in graph.paged(f"/servicePrincipals/{resource['id']}/appRoleAssignedTo", {"$top": "999"}):
if (a.get("principalType") or "ServicePrincipal") != "ServicePrincipal":
continue
name = role_names.get(a.get("appRoleId"))
risk = RISKY_APP_PERMISSIONS.get(resource_app, {}).get(name)
if risk:
found.append({"assignmentId": a.get("id"), "resourceSpId": resource["id"],
"resource": resource.get("displayName"), "permission": name,
"severity": risk[0], "meaning": risk[1],
"principalId": a.get("principalId"), "app": a.get("principalDisplayName"),
"grantedAt": a.get("createdDateTime")})
owners = {}
for principal in {g["principalId"] for g in found}:
try:
sp = graph.get(f"/servicePrincipals/{principal}",
{"$select": "id,appOwnerOrganizationId,publisherName,verifiedPublisher,accountEnabled"})
owners[principal] = sp
except GraphError:
owners[principal] = {}
for g in found:
sp = owners.get(g["principalId"], {})
g["ownApp"] = sp.get("appOwnerOrganizationId") == tenant_id
g["microsoft"] = sp.get("appOwnerOrganizationId") == MICROSOFT_TENANT
g["publisher"] = ((sp.get("verifiedPublisher") or {}).get("displayName") or sp.get("publisherName") or "")
g["disabled"] = sp.get("accountEnabled") is False
found = [g for g in found if not g["microsoft"]]
cov.add("App permissions", "checked", f"{len(found)} powerful permissions granted to apps")
except GraphError as err:
cov.add("App permissions", "skipped", err.reason())
return found
# ---------------------------------------------------------------- mapping file
def load_mapping(path):
try:
with open(path, newline="", encoding="utf-8-sig") as fh:
text = fh.read()
except OSError as err:
die(f"Can't read mapping file {path}: {err}")
first_line = text.splitlines()[0] if text else ""
delimiter = max(",;\t", key=first_line.count) if first_line else ","
reader = csv.DictReader(text.splitlines(), delimiter=delimiter)
columns = {name.strip().lower(): name for name in reader.fieldnames or []}
def pick(*names):
return next((columns[n] for n in names if n in columns), None)
upn_col = pick("user_principal_name", "userprincipalname", "upn", "email", "mail", "e-post")
status_col = pick("employment_status", "status", "anställningsstatus")
if not upn_col or not status_col:
die(f"{path} needs a user_principal_name (or email) column and a status column "
f"(found: {', '.join(columns) or 'nothing'}).")
mapping = {}
for row in reader:
key = (row.get(upn_col) or "").strip().lower()
raw = (row.get(status_col) or "").strip().lower()
if not key:
continue
status = ("left" if raw in LEFT_VALUES else "service" if raw in SERVICE_VALUES
else "active" if raw in ACTIVE_VALUES else "unknown")
mapping[key] = status
note(f"Loaded {len(mapping)} people from {path}")
return mapping
# ---------------------------------------------------------------- analysis
def sku_name(part):
return SKU_NAMES.get(part, part)
def ps_quote(value):
return "'" + str(value).replace("'", "''") + "'"
def finding(severity, category, subject, title, detail, recommendation, fixes=()):
return {"severity": severity, "category": category, "subject": subject, "title": title,
"detail": detail, "recommendation": recommendation, "fixes": list(fixes)}
def analyze(scan, mapping, prices):
now, since, days = scan["now"], scan["since"], scan["days"]
org = scan["org"]
tenant = org.get("displayName") or "your tenant"
findings = []
skus = {s["skuId"]: s for s in scan["skus"]}
sku_parts = {sku_id: s.get("skuPartNumber", sku_id) for sku_id, s in skus.items()}
# Roles per user (direct, or through a role-assignable group)
roles_by_user = defaultdict(list)
sp_roles = []
group_members = scan["roles"].get("group_members", {})
for a in scan["roles"]["assignments"]:
if a["principalType"] == "user":
roles_by_user[a["principalId"]].append(a)
elif a["principalType"] == "group":
for member in group_members.get(a["principalId"], []):
roles_by_user[member].append(dict(a, viaGroup=a["principalName"]))
elif a["principalType"] == "servicePrincipal":
sp_roles.append(a)
people = []
by_id = {}
for u in scan["users"]:
upn = u.get("userPrincipalName") or ""
kind = "guest" if (u.get("userType") or "").lower() == "guest" else "member"
enabled = u.get("accountEnabled") is not False
created = parse_ts(u.get("createdDateTime"))
activity = u.get("signInActivity") or {}
last_success = parse_ts(activity.get("lastSuccessfulSignInDateTime"))
attempts = [t for t in (parse_ts(activity.get("lastSignInDateTime")),
parse_ts(activity.get("lastNonInteractiveSignInDateTime"))) if t]
# The other two timestamps include failed attempts, so a password-spray target could look
# active. lastSuccessfulSignInDateTime exists from December 2023: before that, the attempt
# timestamps are the best we have; after it, an attempt without a success was a failure.
if last_success:
last_signin = last_success
else:
older = [t for t in attempts if t < SUCCESS_TRACKED_FROM]
last_signin = max(older) if older else None
failed_attempt = max((t for t in attempts if t >= SUCCESS_TRACKED_FROM and (not last_success or t > last_success)),
default=None)
usage_last = scan["usage"].get(upn.lower()) if scan["usage"] else None
last_seen = later(last_signin, usage_last)
visible = scan["signin_available"] or (bool(scan["usage"]) and upn.lower() in scan["usage"])
status = (mapping or {}).get(upn.lower()) or (mapping or {}).get((u.get("mail") or "").lower()) or "unknown"
user_roles = roles_by_user.get(u["id"], [])
role_names = sorted({r["role"] for r in user_roles})
privileged = any(r["templateId"] in PRIVILEGED_ROLES for r in user_roles)
is_sync_account = (any(r["templateId"] == DIRECTORY_SYNC_ROLE for r in user_roles)
or upn.lower().startswith("sync_"))
if is_sync_account:
status = "service"
licenses = [l.get("skuId") for l in u.get("assignedLicenses") or []]
paid = [s for s in licenses if sku_parts.get(s, s) not in FREE_SKUS]
# Licenses that come from a group can only be removed by taking the user out of that group.
group_assigned = {st.get("skuId") for st in u.get("licenseAssignmentStates") or [] if st.get("assignedByGroup")}
direct_paid = [s for s in paid if s not in group_assigned]
mfa_row = (scan["mfa"] or {}).get(u["id"])
mfa = None if mfa_row is None else bool(mfa_row.get("isMfaRegistered"))
old_enough = created is None or created < since
if status == "service":
verdict = "service"
elif status == "left" and enabled:
verdict = "departed"
elif not enabled:
verdict = "leftovers" if (paid or user_roles) else "disabled"
elif kind == "guest" and (u.get("externalUserState") or "") == "PendingAcceptance" and old_enough:
verdict = "pending"
elif not visible:
verdict = "unknown"
elif last_seen is None and old_enough:
verdict = "never"
elif last_seen is not None and last_seen < since and old_enough:
verdict = "dormant"
else:
verdict = "active"
person = {
"id": u["id"], "user_principal_name": upn, "display_name": u.get("displayName") or "",
"mail": u.get("mail") or "", "kind": kind, "enabled": enabled, "verdict": verdict,
"employment_status": status if status != "unknown" else "",
"created": iso(created), "last_seen": iso(last_seen), "last_sign_in": iso(last_signin),
"last_m365_activity": iso(usage_last),
"synced_from_ad": bool(u.get("onPremisesSyncEnabled")), "sam_account_name": u.get("onPremisesSamAccountName") or "",
"department": u.get("department") or "", "job_title": u.get("jobTitle") or "",
"roles": role_names, "privileged": privileged,
"licenses": [sku_name(sku_parts.get(s, s)) for s in paid], "license_ids": paid,
"direct_license_ids": direct_paid,
"mfa_registered": mfa, "guest_state": u.get("externalUserState") or "",
"failed_sign_in": iso(failed_attempt) if failed_attempt and failed_attempt >= since else None,
"group_licenses": sorted({sku_name(sku_parts.get(st.get("skuId"), st.get("skuId")))
for st in u.get("licenseAssignmentStates") or [] if st.get("assignedByGroup")}),
}
people.append(person)
by_id[u["id"]] = person
# Per-person findings
for p in people:
who = p["user_principal_name"]
disable = f"Update-MgUser -UserId {ps_quote(p['id'])} -AccountEnabled:$false"
if p["synced_from_ad"]:
disable = (f"Disable-ADAccount -Identity {ps_quote(p['sam_account_name'] or who)}"
" # synced from Active Directory: run this on a domain controller instead")
revoke = f"Revoke-MgUserSignInSession -UserId {ps_quote(p['id'])}"
unlicense = (f"Set-MgUserLicense -UserId {ps_quote(p['id'])} -AddLicenses @() -RemoveLicenses @("
+ ",".join(ps_quote(s) for s in p["direct_license_ids"]) + ")") if p["direct_license_ids"] else None
group_note = (f" {', '.join(p['group_licenses'])} {'is' if len(p['group_licenses']) == 1 else 'are'} assigned "
"through a group, so take the account out of that group to free it." if p["group_licenses"] else "")
admin_note = f" Admin roles: {', '.join(p['roles'])}." if p["roles"] else ""
lic_note = (f" Paid licenses: {', '.join(p['licenses'])}." if p["licenses"] else "") + group_note
seen = (f"Last seen {fmt_date(parse_ts(p['last_seen']))}." if p["last_seen"] else "Never signed in.")
if p["failed_sign_in"]:
seen += f" Someone tried to sign in on {fmt_date(parse_ts(p['failed_sign_in']))} and failed."
source = " Synced from Active Directory." if p["synced_from_ad"] else ""
if p["verdict"] == "departed":
findings.append(finding(
"critical", "departed", who, "Has left the company but the account is still enabled",
f"Marked as left in the mapping file.{admin_note}{lic_note} {seen}{source}",
"Disable the account and sign it out everywhere, then remove its licenses and roles.",
[disable, revoke] + ([unlicense] if unlicense else [])))
elif p["verdict"] in ("dormant", "never"):
guest = p["kind"] == "guest"
# An unused account that someone is trying to get into is a likely target.
sev = "high" if (p["roles"] or p["failed_sign_in"]) else "medium"
title = ("Never signed in" if p["verdict"] == "never" else f"No sign-in in {days} days")
if guest:
title = "Guest: " + title[0].lower() + title[1:]
if p["roles"]:
title += " (admin)"
findings.append(finding(
sev, "dormant", who, title,
f"{seen} Created {fmt_date(parse_ts(p['created'])) or 'unknown'}.{admin_note}{lic_note}{source}",
("Remove the guest if nobody needs it." if guest else
"Check with their manager. If nobody needs the account, disable it and remove its licenses."),
([f"Remove-MgUser -UserId {ps_quote(p['id'])} # deletes the guest (restorable for 30 days)"] if guest
else [disable] + ([unlicense] if unlicense else []))))
elif p["verdict"] == "pending":
findings.append(finding(
"low", "pending", who, "Guest invitation never accepted",
f"Invited {fmt_date(parse_ts(p['created'])) or 'a while ago'} and never accepted.{admin_note}",
"Delete the guest. You can invite them again if they still need access.",
[f"Remove-MgUser -UserId {ps_quote(p['id'])}"]))
elif p["verdict"] == "leftovers":
has_roles = bool(p["roles"])
findings.append(finding(
"medium" if has_roles else "low", "leftovers", who,
"Disabled account still holds " + ("admin roles" if has_roles else "paid licenses"),
f"The account is disabled.{admin_note}{lic_note} Shared mailboxes only need a license above "
"50 GB or with archiving.",
"Remove the licenses and roles, or delete the account if it's no longer needed.",
([unlicense] if unlicense else []) +
[(f"Remove-MgRoleManagementDirectoryRoleAssignment -UnifiedRoleAssignmentId {ps_quote(r['id'])}"
if r.get("id") else
f"Remove-MgDirectoryRoleMemberByRef -DirectoryRoleId {ps_quote(r.get('directoryRoleId'))} "
f"-DirectoryObjectId {ps_quote(p['id'])}")
for r in scan["roles"]["assignments"]
if r["principalId"] == p["id"] and r["kind"] == "active"]))
if p["roles"] and p["kind"] == "guest" and p["verdict"] not in ("dormant", "never", "pending"):
findings.append(finding(
"high", "admin", who, "Guest with an admin role",
f"Admin roles: {', '.join(p['roles'])}. A guest's account is controlled by another organization.",
"Give the person a member account in your tenant if they need admin rights, or remove the role."))
if p["privileged"] and p["synced_from_ad"] and p["verdict"] not in ("departed", "leftovers", "disabled"):
findings.append(finding(
"medium", "admin", who, "Admin role on an account synced from Active Directory",
f"Admin roles: {', '.join(p['roles'])}. If the on-premises account is taken over, so is the cloud admin.",
"Use a separate cloud-only account for admin work."))
if p["roles"] and p["mfa_registered"] is False and p["enabled"] and p["verdict"] != "service":
findings.append(finding(
"high", "mfa", who, "Admin without MFA registered",
f"Admin roles: {', '.join(p['roles'])}. A stolen password is enough to take over the tenant.",
"Register an authenticator app, passkey or security key for this account today."))
# Tenant-wide findings
security = scan["security"]
if security["security_defaults"] is False and not security["ca_mfa_all"]:
findings.append(finding(
"high", "baseline", tenant, "MFA isn't enforced for everyone",
"Security defaults are off and no Conditional Access policy requires MFA for all users.",
"Turn on security defaults, or create a Conditional Access policy that requires MFA for all users."))
no_mfa = [p for p in people if p["mfa_registered"] is False and p["enabled"] and p["kind"] == "member"
and p["verdict"] not in ("service",) and not p["roles"]]
if no_mfa:
findings.append(finding(
"medium", "mfa", tenant, f"{len(no_mfa)} accounts haven't registered MFA",
"These enabled accounts have no MFA method registered: "
+ ", ".join(p["user_principal_name"] for p in no_mfa[:12]) + (" and more." if len(no_mfa) > 12 else "."),
"Ask them to register an authenticator app. Security defaults or Conditional Access can require it."))
global_admins = sorted({by_id[a["principalId"]]["user_principal_name"]
for a in scan["roles"]["assignments"]
if a["templateId"] == GLOBAL_ADMIN and a["kind"] == "active" and a["principalId"] in by_id
and by_id[a["principalId"]]["enabled"]})
if len(global_admins) > 4:
findings.append(finding(
"medium", "admin", tenant, f"{len(global_admins)} permanent Global Administrators",
"Microsoft recommends fewer than five: " + ", ".join(global_admins) + ".",
"Move people to narrower roles such as User Administrator, or make the role eligible through PIM."))
elif len(global_admins) == 1:
findings.append(finding(
"low", "admin", tenant, "Only one Global Administrator",
f"Only {global_admins[0]} can fix everything. If that account is locked out, so are you.",
"Keep a second, cloud-only emergency access account with a long password and a security key."))
# Apps
for g in scan["app_grants"]:
if g.get("disabled"):
continue
owner = "your own app" if g["ownApp"] else (f"published by {g['publisher']}" if g["publisher"] else "a third-party app")
findings.append(finding(
g["severity"], "app", g["app"] or g["principalId"], f"App {g['meaning']}",
f"{g['app']} ({owner}) has the {g['resource']} application permission {g['permission']}, "
f"granted {fmt_date(parse_ts(g['grantedAt'])) or 'at an unknown date'}. It works without any user signed in.",
"Check that the app still needs it, and who looks after it. Remove the permission if nobody does.",
[f"Remove-MgServicePrincipalAppRoleAssignedTo -ServicePrincipalId {ps_quote(g['resourceSpId'])} "
f"-AppRoleAssignmentId {ps_quote(g['assignmentId'])}"]))
secrets_rows = []
for app in scan["apps"]:
for kind, creds in (("secret", app.get("passwordCredentials") or []), ("certificate", app.get("keyCredentials") or [])):
for c in creds:
end = parse_ts(c.get("endDateTime"))
if not end:
continue
days_left = (end - now).days
state = "expired" if end < now else "expiring" if days_left <= 30 else "ok"
secrets_rows.append({"app": app.get("displayName"), "appObjectId": app.get("id"), "kind": kind,
"name": c.get("displayName") or "", "keyId": c.get("keyId"),
"expires": iso(end), "state": state, "days_left": days_left})
if state == "expired":
findings.append(finding(
"low", "secret", app.get("displayName") or app.get("appId"), f"Expired {kind} still listed",
f"'{c.get('displayName') or c.get('keyId')}' expired {fmt_date(end)}.",
"Delete it so nobody mistakes it for a working one.",
[f"Remove-MgApplicationPassword -ApplicationId {ps_quote(app['id'])} -KeyId {ps_quote(c.get('keyId'))}"]
if kind == "secret" else []))
elif state == "expiring":
findings.append(finding(
"medium", "secret", app.get("displayName") or app.get("appId"),
f"{kind.capitalize()} expires in {days_left} days",
f"'{c.get('displayName') or c.get('keyId')}' expires {fmt_date(end)}. Whatever uses it stops working then.",
"Create a new one, update the system that uses it, then delete the old one."))
# Licenses
price_of = {}
for part, price in (prices or {}).items():
price_of[part.upper()] = price
license_rows = []
wasted_total, currency = 0.0, ""
ghost_verdicts = {"departed", "dormant", "never", "leftovers", "pending"}
for sku_id, s in skus.items():
part = s.get("skuPartNumber", sku_id)
if part in FREE_SKUS:
continue
holders = [p for p in people if sku_id in p["license_ids"]]
wasted = [p for p in holders if p["verdict"] in ghost_verdicts]
purchased = (s.get("prepaidUnits") or {}).get("enabled", 0)
price = price_of.get(part.upper())
cost = None
if price and wasted:
cost = price[0] * len(wasted) * 12
wasted_total += cost
currency = price[1]
license_rows.append({"sku": part, "name": sku_name(part), "skuId": sku_id, "purchased": purchased,
"assigned": s.get("consumedUnits", 0), "unused": max(0, purchased - s.get("consumedUnits", 0)),
"on_ghosts": len(wasted), "yearly_cost_on_ghosts": cost})
license_rows.sort(key=lambda r: (-r["on_ghosts"], -r["assigned"], r["name"]))
ghost_licenses = sum(r["on_ghosts"] for r in license_rows)
if ghost_licenses:
money = f" That is about {wasted_total:,.0f} {currency} a year at the prices you gave." if wasted_total else ""
findings.append(finding(
"low", "license", tenant, f"{ghost_licenses} paid licenses on inactive or disabled accounts",
"; ".join(f"{r['name']}: {r['on_ghosts']}" for r in license_rows if r["on_ghosts"]) + "." + money,
"Remove the licenses from accounts nobody uses; the fix for each account is listed with it."))
unused_seats = sum(r["unused"] for r in license_rows)
if not mapping:
findings.append(finding(
"info", "identity", tenant, "Accounts weren't checked against HR",
"Entra ID only knows whether an account is enabled, not whether the person still works for you.",
"Fill in employment_status (active, left or service) in people.csv and run again with --mapping people.csv."))
findings.sort(key=lambda f: (SEVERITIES.index(f["severity"]), f["category"], str(f["subject"]).lower()))
people.sort(key=lambda p: (VERDICT_ORDER[p["verdict"]], p["kind"] != "member", p["user_principal_name"].lower()))
counts = {s: sum(1 for f in findings if f["severity"] == s) for s in SEVERITIES}
members = [p for p in people if p["kind"] == "member" and p["enabled"] and p["verdict"] != "service"]
guests = [p for p in people if p["kind"] == "guest"]
admins = sorted({p["id"] for p in people if p["roles"]})
domains = [d.get("name") for d in org.get("verifiedDomains") or [] if d.get("isDefault")]
return {
"tool": {"name": "ghostbuster-entra", "version": VERSION, "schema": 1},
"tenant": tenant, "tenant_id": org.get("id"), "domain": domains[0] if domains else "",
"entra_plan": scan["entra_plan"], "scanned_by": (scan["me"] or {}).get("userPrincipalName", ""),
"scanned_at": iso(now), "window_days": days, "window_start": iso(since),
"activity_source": ("sign-in activity" if scan["signin_available"]
else "Microsoft 365 usage" if scan["usage"] else "none"),
"hybrid": bool(org.get("onPremisesSyncEnabled")),
"settings": {"security_defaults": security["security_defaults"],
"ca_policies": len(security["ca_policies"]) if security["ca_policies"] is not None else None,
"ca_mfa_all": security["ca_mfa_all"],
"mfa_checked": scan["mfa"] is not None, "pim_checked": scan["roles"].get("eligible_checked", False)},
"summary": {
"accounts": len(people), "members": len(members), "guests": len(guests), "admins": len(admins),
"global_admins": global_admins,
"possible_ghosts": sum(1 for p in people if p["verdict"] in ("departed", "dormant", "never")),
"ghost_licenses": ghost_licenses, "unused_seats": unused_seats,
"yearly_cost_on_ghosts": wasted_total or None, "currency": currency,
"findings": counts,
},
"people": people, "licenses": license_rows, "secrets": secrets_rows,
"app_grants": scan["app_grants"], "sp_roles": sp_roles,
"role_assignments": scan["roles"]["assignments"], "findings": findings, "coverage": scan["coverage"],
}
# ---------------------------------------------------------------- output
LIMITATIONS = [
"Sign-in activity needs Microsoft Entra ID P1 or P2. Without it, Ghostbuster falls back to the Microsoft 365 "
"usage report, which only covers licensed users and shows dates, not times.",
"Microsoft updates sign-in activity with a delay, and a sign-in only counts when it reached Entra ID. "
"Someone who only uses an app with its own login won't show up.",
"Entra ID can't tell who has left the company. Accounts are checked against HR only when you give a mapping file.",
"License counts come from Entra ID. Ghostbuster doesn't know your prices unless you pass --prices.",
"Only application permissions on Microsoft Graph, Exchange Online and SharePoint are checked, not delegated consents.",
]
def write_outputs(rep, out_dir):
os.makedirs(out_dir, exist_ok=True)
with open(os.path.join(out_dir, "report.json"), "w", encoding="utf-8") as fh:
json.dump(rep, fh, indent=2, ensure_ascii=False)
rows = []
for p in rep["people"]:
row = dict(p)
row["roles"] = "; ".join(p["roles"])
row["licenses"] = "; ".join(p["licenses"])
row["last_seen"] = (p["last_seen"] or "")[:10]
row["created"] = (p["created"] or "")[:10]
row["mfa_registered"] = "" if p["mfa_registered"] is None else ("yes" if p["mfa_registered"] else "no")
rows.append(row)
write_csv(os.path.join(out_dir, "people.csv"), rows,
["user_principal_name", "employment_status", "display_name", "kind", "verdict", "enabled", "last_seen",
"created", "roles", "licenses", "mfa_registered", "synced_from_ad", "department", "job_title", "mail", "id"])
write_csv(os.path.join(out_dir, "licenses.csv"), rep["licenses"],
["name", "sku", "purchased", "assigned", "unused", "on_ghosts", "yearly_cost_on_ghosts"])
write_csv(os.path.join(out_dir, "findings.csv"), [dict(f, fix_commands=" ; ".join(f["fixes"])) for f in rep["findings"]],
["severity", "category", "subject", "title", "detail", "recommendation", "fix_commands"])
with open(os.path.join(out_dir, "suggested-fixes.ps1"), "w", encoding="utf-8-sig", newline="\n") as fh:
fh.write(render_fixes(rep))
with open(os.path.join(out_dir, "report.html"), "w", encoding="utf-8") as fh:
fh.write(render_html(rep))
def render_fixes(rep):
lines = [
f"# {TOOL_NAME}: suggested fixes for {rep['tenant']}, generated {rep['scanned_at'][:10]}",
"#",
"# NOTHING IN THIS FILE RUNS AS-IS: every command is commented out.",
"# Review each one (ideally with the person's manager), remove the leading '# ' from",
"# the ones you agree with, then run them in PowerShell 7 with Microsoft Graph PowerShell:",
"#",
"# Install-Module Microsoft.Graph -Scope CurrentUser",
'# Connect-MgGraph -Scopes "User.ReadWrite.All","Directory.AccessAsUser.All","RoleManagement.ReadWrite.Directory","Application.ReadWrite.All"',
"#",
"# Accounts synced from Active Directory must be disabled in AD; those lines say so.",
"# Deleted users and guests can be restored from Deleted users for 30 days.",
"",
]
start = len(lines)
for item in rep["findings"]:
if not item["fixes"]:
continue
lines.append(f"# [{item['severity'].upper()}] {item['title']} - {item['subject']}")
lines += [f"# {command}" for command in item["fixes"]]
lines.append("")
if len(lines) == start:
lines.append("# No fixes to suggest.")
return "\n".join(lines) + "\n"
ENTRA_CSS = """
.never,.pending{background:var(--warn-soft);color:var(--warn)}
.leftovers{background:var(--slate-soft);color:var(--slate)}
.service,.disabled{background:var(--neutral-soft);color:var(--muted)}
.nowrap{white-space:nowrap}
.yes{background:var(--ok-soft);color:var(--ok)}.no{background:var(--bad-soft);color:var(--bad)}
.v-never{background:var(--warn)}.v-pending{background:#c9a35a}.v-leftovers{background:var(--slate)}
.v-service{background:#b9bdb6}.v-disabled{background:#d8dbd4}
"""
VERDICT_LABELS = (("departed", "Left the company"), ("dormant", "No recent sign-in"), ("never", "Never signed in"),
("pending", "Invite not accepted"), ("leftovers", "Disabled, not cleaned up"),
("unknown", "Activity not visible"), ("service", "Service account"), ("disabled", "Disabled"),
("active", "Active"))
VERDICT_PILL = {"departed": "Left", "dormant": "Inactive", "never": "Never signed in", "pending": "Not accepted",
"leftovers": "Leftovers", "unknown": "Not visible", "service": "Service", "disabled": "Disabled",
"active": "Active"}
REPORT_FILES = (
("report.html", "This report. Use Save as PDF to keep a copy."),
("people.csv", "One row per account. Fill in employment_status (active, left or service) and re-run with --mapping."),
("licenses.csv", "Each paid license: purchased, assigned, unused and held by inactive or disabled accounts."),
("findings.csv", "Every finding with its recommendation and fix command."),
("suggested-fixes.ps1", "Microsoft Graph PowerShell commands that fix the findings. All commented out: review, uncomment, run."),
("report.json", "Everything above, machine-readable."),
)
def _plural(n, one, many):
return f"{n} {one if n == 1 else many}"
def render_html(rep):
e = html.escape
s = rep["summary"]
tenant, days = rep["tenant"], rep["window_days"]
now = parse_ts(rep["scanned_at"])
date = rep["scanned_at"][:10]
people = rep["people"]
by_cat = defaultdict(list)
for item in rep["findings"]:
by_cat[item["category"]].append(item)
verdicts = Counter(p["verdict"] for p in people)
def pill(text, cls=None):
return f'{e(text)}'
def verdict_pill(p):
return pill(VERDICT_PILL[p["verdict"]], p["verdict"])
def seen(value, never="never"):
ts = parse_ts(value)
if not ts:
return f'{e(never)}'
return f'{e(fmt_date(ts))}
{(now - ts).days} days ago
'
def who(p):
name = f'
{e(p["user_principal_name"])}
'
return f"{e(p['display_name'] or p['user_principal_name'])}{name}"
def kind_text(p):
bits = [p["kind"]]
if p["synced_from_ad"]:
bits.append("synced from AD")
return e(", ".join(bits))
def listing(items, limit=6):
shown = " ".join(e(i) for i in items[:limit])
if len(items) > limit:
shown += f' and {len(items) - limit} more'
return shown or 'none'
def mfa_pill(p):
if p["mfa_registered"] is None:
return 'unknown'
return pill("yes" if p["mfa_registered"] else "no")
def table(head, rows, empty):
if not rows:
return f'
{e(empty)}
'
th = "".join(f"
{e(h)}
" for h in head)
return f'
{th}
{"".join(rows)}
'
def tr(*cells):
return "
" + "".join(f"
{c}
" for c in cells) + "
"
def fixes(*categories):
commands = [c for cat in categories for item in by_cat.get(cat, []) for c in item["fixes"]]
if not commands:
return ""
return ('Commands to fix this'
f"
{e(chr(10).join(commands))}
"
'
The same commands are in suggested-fixes.ps1. Review each one before you run it.
')
sections = []
def add(anchor, title, intro, body, count=None, appendix=False):
sections.append({"anchor": anchor, "title": title, "intro": intro, "body": body,
"count": count, "appendix": appendix})
# Possible ghosts (members)
ghosts = [p for p in people if p["kind"] == "member" and p["verdict"] in ("departed", "dormant", "never")]
rows = [tr(who(p), verdict_pill(p), seen(p["last_seen"]), f'{e((p["created"] or "")[:10])}',
listing(p["licenses"], 3), listing(p["roles"], 3), kind_text(p)) for p in ghosts]
intro = (f"Enabled accounts that belong to someone who has left, or that haven't signed in for {days} days. "
"Check with their manager before you disable anyone; some accounts are only used a few times a year.")
if rep["activity_source"] == "none":
intro = "Sign-in activity wasn't available, so only people marked as left in the mapping file are listed."
add("ghosts", "Possible ghosts", intro,
table(["Account", "Status", "Last seen", "Created", "Paid licenses", "Admin roles", "Type"], rows,
"No possible ghosts found.") + fixes("departed", "dormant"), len(ghosts))
# Guests
guests = [p for p in people if p["kind"] == "guest"]
flagged_guests = [p for p in guests if p["verdict"] in ("dormant", "never", "pending", "departed")]
rows = [tr(who(p), verdict_pill(p), e((p["created"] or "")[:10]), seen(p["last_seen"]), listing(p["roles"], 3))
for p in flagged_guests]
add("guests", "Guests",
f"People from other organizations with an account in your tenant: {len(guests)} in total. Listed here are "
f"the ones that never accepted their invitation or haven't signed in for {days} days.",
table(["Guest", "Status", "Invited", "Last seen", "Admin roles"], rows,
"Every guest has signed in recently.") + fixes("pending"), len(flagged_guests))
# Admin roles
admins = [p for p in people if p["roles"]]
admins.sort(key=lambda p: (not p["privileged"], p["user_principal_name"].lower()))
eligible = defaultdict(set)
for a in rep["role_assignments"]:
if a["kind"] == "eligible":
eligible[a["principalId"]].add(a["role"])
rows = []
for p in admins:
roles_html = " ".join(e(r) + (' eligible' if r in eligible.get(p["id"], set()) else "")
for r in p["roles"])
rows.append(tr(who(p), roles_html, kind_text(p), mfa_pill(p), seen(p["last_seen"]), verdict_pill(p)))
sp_rows = [tr(e(a["principalName"]), e(a["role"]), e(a["kind"])) for a in rep["sp_roles"]]
ga = s["global_admins"]
body = (f'
{_plural(len(ga), "permanent Global Administrator", "permanent Global Administrators")}'
+ (f": {e(', '.join(ga))}" if ga else "") + ".
" + table(["App", "Role", "Assignment"], sp_rows, "") if sp_rows else "")
+ fixes("admin", "mfa"))
add("admins", "Admin roles",
"Everyone with an Entra ID admin role. Admin accounts should be few, cloud-only, protected with MFA and in use.",
body, len(admins))
# Sign-in security
st = rep["settings"]
def state(good, bad):
return pill("good") if good else pill("fix") if bad else pill("unknown")
sd, ca = st["security_defaults"], st["ca_policies"]
mfa_known = [p for p in people if p["mfa_registered"] is not None and p["enabled"] and p["kind"] == "member"
and p["verdict"] != "service"]
registered = sum(1 for p in mfa_known if p["mfa_registered"])
unenforced = sd is False and not st["ca_mfa_all"]
settings_rows = [
tr("Security defaults", "On" if sd else "Off" if sd is False else "Unknown", state(sd, unenforced)),
tr("Conditional Access policies", "Not available" if ca is None else
f"{ca}, " + ("one requires MFA for all users" if st["ca_mfa_all"] else "none requires MFA for all users"),
state(st["ca_mfa_all"], unenforced) if ca is not None else pill("unknown")),
tr("MFA required for all users", "Yes" if (sd or st["ca_mfa_all"]) else "No",
state(sd or st["ca_mfa_all"], sd is False and not st["ca_mfa_all"])),
tr("Members with MFA registered", f"{registered} of {len(mfa_known)}" if mfa_known else "Not checked",
state(mfa_known and registered == len(mfa_known), mfa_known and registered < len(mfa_known))),
]
no_mfa = [p for p in mfa_known if not p["mfa_registered"]]
rows = [tr(who(p), e(", ".join(p["roles"]) or "none"), seen(p["last_seen"])) for p in no_mfa[:60]]
add("security", "Sign-in security", "Whether MFA protects every account, so that a stolen password isn't enough.",
table(["Setting", "Value", "Status"], settings_rows, "")
+ "
Accounts without MFA registered
"
+ table(["Account", "Admin roles", "Last seen"], rows,
"Everyone has MFA registered." if st["mfa_checked"] else "Not checked: needs Entra ID P1 or P2.")
+ (f'
Showing 60 of {len(no_mfa)}. The full list is in people.csv.
' if len(no_mfa) > 60 else "")
+ fixes("baseline"), len(no_mfa))
# Licenses
lic_rows = []
for r in rep["licenses"]:
cost = (f"{r['yearly_cost_on_ghosts']:,.0f} {e(s['currency'])}" if r["yearly_cost_on_ghosts"] else "")
lic_rows.append(tr(f"{e(r['name'])}", str(r["purchased"]), str(r["assigned"]),
str(r["unused"]), (f'{r["on_ghosts"]}' if r["on_ghosts"] else "0"),
cost or 'add --prices'))
leftovers = [p for p in people if p["verdict"] == "leftovers"]
left_rows = [tr(who(p), listing(p["licenses"], 3), listing(p["roles"], 3)) for p in leftovers]
cost_note = (f" That's about {s['yearly_cost_on_ghosts']:,.0f} {e(s['currency'])} a year."
if s["yearly_cost_on_ghosts"] else "")
add("licenses", "Licenses",
"Paid licenses, and how many of them sit on accounts nobody uses. Free and trial licenses are left out.",
f"
{_plural(s['ghost_licenses'], 'paid license is', 'paid licenses are')} assigned to inactive or disabled "
f"accounts, and {_plural(s['unused_seats'], 'purchased license is', 'purchased licenses are')} not assigned at all.{cost_note}
"
+ table(["License", "Purchased", "Assigned", "Not assigned", "On inactive or disabled accounts", "Yearly cost of those"],
lic_rows, "No paid licenses found.")
+ "
Disabled accounts that still hold licenses or roles
"
+ table(["Account", "Paid licenses", "Admin roles"], left_rows, "None.")
+ fixes("leftovers", "license"), s["ghost_licenses"])
# Apps and secrets
grant_rows = []
for g in rep["app_grants"]:
owner = "Your organization" if g["ownApp"] else (g["publisher"] or "Third party")
grant_rows.append(tr(f"{e(g['app'] or '')}" + (' disabled' if g.get("disabled") else ""),
e(owner), f"{e(g['permission'])}
{e(g['resource'])}
",
e(g["meaning"]), pill(g["severity"])))
secret_rows = [tr(e(r["app"] or ""), e(r["kind"]), e(r["name"]), e((r["expires"] or "")[:10]),
pill("expired", "fix") if r["state"] == "expired" else pill(f"{r['days_left']} days", "medium"))
for r in sorted(rep["secrets"], key=lambda r: r["days_left"]) if r["state"] != "ok"]
add("apps", "Apps, permissions and secrets",
"Apps that can act on the whole tenant without anyone signed in, and app secrets that have expired or soon will.",
"
Apps with powerful permissions
"
+ table(["App", "Owner", "Permission", "What it allows", "Risk"], grant_rows, "No app has powerful application permissions.")
+ "
Secrets and certificates
"
+ table(["App", "Type", "Name", "Expires", "Status"], secret_rows, "No secrets have expired or expire within 30 days.")
+ fixes("app", "secret"), len(grant_rows) + len(secret_rows))
# Appendices
shown = people if len(people) <= 400 else [p for p in people if p["verdict"] != "active"]
rows = [tr(who(p), kind_text(p), verdict_pill(p), seen(p["last_seen"]), listing(p["licenses"], 2),
listing(p["roles"], 2), mfa_pill(p)) for p in shown]
add("people", "Everyone", "Every account in the tenant." if shown is people else
f"The tenant has {len(people)} accounts, so only those that need attention are listed. people.csv has everyone.",
table(["Account", "Type", "Status", "Last seen", "Paid licenses", "Admin roles", "MFA"], rows, "Nobody found."),
len(shown), appendix=True)
cov_rows = [tr(e(c["check"]), pill(c["status"]), e(c["detail"])) for c in rep["coverage"]]
files = [tr(f"{e(name)}", e(what)) for name, what in REPORT_FILES]
add("method", "How this was checked",
f"Ghostbuster for Entra ID {VERSION} read this tenant through Microsoft Graph with read-only permissions. It changed nothing.",
table(["Check", "Status", "Details"], cov_rows, "")
+ "
What this scan can't see
" + "".join(f"
{e(i)}
" for i in LIMITATIONS) + "
"
+ "
Files next to this report
" + table(["File", "What's in it"], files, ""), appendix=True)
# Summary
sentences = [f"{e(tenant)} has {_plural(s['members'], 'enabled member account', 'enabled member accounts')} "
f"and {_plural(s['guests'], 'guest', 'guests')}."]
if verdicts["departed"]:
sentences.append(f"{_plural(verdicts['departed'], 'person has', 'people have')} left the company but can still sign in.")
member_dormant = sum(1 for p in ghosts if p["verdict"] in ("dormant", "never"))
if member_dormant:
sentences.append(f"{_plural(member_dormant, 'account has', 'accounts have')} not signed in for {days} days or never.")
if flagged_guests:
sentences.append(f"{_plural(len(flagged_guests), 'guest looks', 'guests look')} abandoned.")
if s["ghost_licenses"]:
sentences.append(f"{_plural(s['ghost_licenses'], 'paid license sits', 'paid licenses sit')} on accounts nobody uses.")
if st["security_defaults"] is False and not st["ca_mfa_all"]:
sentences.append("MFA isn't enforced for everyone.")
if len(sentences) == 1:
sentences.append("No ghosts or leftover access found.")
action_text = (
("departed", "ghosts", lambda n: f"Disable {_plural(n, 'account', 'accounts')} of people who left"),
("admin", "admins", None),
("mfa", "security", None),
("baseline", "security", None),
("dormant", "ghosts", lambda n: f"Review {_plural(n, 'inactive account', 'inactive accounts')}"),
("app", "apps", lambda n: f"Check {_plural(n, 'app permission', 'app permissions')} that reach the whole tenant"),
("secret", "apps", lambda n: f"Renew or delete {_plural(n, 'app secret', 'app secrets')}"),
("pending", "guests", lambda n: f"Delete {_plural(n, 'guest', 'guests')} who never accepted"),
("leftovers", "licenses", lambda n: f"Clean up {_plural(n, 'disabled account', 'disabled accounts')}"),
("license", "licenses", None),
("identity", "ghosts", None),
)
actions = []
for order, (category, anchor, text) in enumerate(action_text):
items = by_cat.get(category, [])
if not items:
continue
if text is None:
actions += [(SEVERITIES.index(i["severity"]), order, i["severity"], i["title"], anchor) for i in items[:3]]
else:
worst = min(SEVERITIES.index(i["severity"]) for i in items)
actions.append((worst, order, SEVERITIES[worst], text(len(items)), anchor))
actions.sort()
actions = actions[:12]
actions_html = ("" + "".join(
f'
'
for _, _, sev, text, anchor in actions) + "") if actions else '
Nothing to do.
'
total = max(1, len(people))
segments = "".join(f''
for v, _ in VERDICT_LABELS if verdicts[v])
legend = "".join(f'{e(label)}{verdicts[v]}'
for v, label in VERDICT_LABELS if verdicts[v])
serious = s["findings"]["critical"] + s["findings"]["high"]
tiles = [
(s["members"], "enabled members", False),
(s["possible_ghosts"], "possible ghosts", s["possible_ghosts"] > 0),
(s["guests"], "guests", False),
(s["admins"], "admins", len(s["global_admins"]) > 4),
(s["ghost_licenses"], "paid licenses on ghosts", s["ghost_licenses"] > 0),
(serious, "critical or high findings", serious > 0),
]
tiles_html = "".join(f'
{n}
{e(label)}
'
for n, label, alert in tiles)
body, number, letter = [], 0, 0
for sec in sections:
if sec["appendix"]:
label = f"Appendix {chr(ord('A') + letter)}"
css_class = ' class="appendix-start"' if letter == 0 else ""
letter += 1
else:
number += 1
label, css_class = str(number), ""
count = f'{sec["count"]}' if sec["count"] else ""
body.append(f'
"""
def print_summary(rep, out_dir):
s = rep["summary"]
counts = s["findings"]
say()
say(paint(rep["tenant"], "1") + paint(f" | {rep['entra_plan']} | last {rep['window_days']} days", "2"))
say(f" {s['members']} members | {s['guests']} guests | {s['admins']} admins | {s['possible_ghosts']} possible ghosts"
f" | {s['ghost_licenses']} paid licenses on ghosts")
say(" Findings: " + ", ".join(paint(f"{counts[sev]} {sev}", SEV_COLOR[sev]) for sev in SEVERITIES if counts[sev])
if any(counts.values()) else " No findings.")
top = [f for f in rep["findings"] if f["severity"] != "info"][:10]
if top:
say()
for item in top:
say(f" {paint(item['severity'].upper().ljust(8), SEV_COLOR[item['severity']])} {item['title']} - {item['subject']}")
hidden = len(rep["findings"]) - len(top)
if hidden > 0:
note(f"...and {hidden} more in the report")
say()
say(f" Report: {os.path.join(out_dir, 'report.html')}")
say(" Spreadsheets: people.csv, licenses.csv, findings.csv")
say(" Suggested fixes: suggested-fixes.ps1 (every command is commented out)")
note("The report lists names and access details. Share it with care.")
# ---------------------------------------------------------------- main
def load_prices(path):
prices = {}
with open(path, newline="", encoding="utf-8-sig") as fh:
text = fh.read()
first = text.splitlines()[0] if text else ""
reader = csv.DictReader(text.splitlines(), delimiter=max(",;\t", key=first.count) if first else ",")
for row in reader:
cols = {k.strip().lower(): (v or "").strip() for k, v in row.items() if k}
sku = cols.get("sku") or cols.get("sku_part_number") or cols.get("license")
amount = (cols.get("monthly_price") or cols.get("price") or "").replace(" ", "").replace(",", ".")
if sku and amount:
try:
prices[sku] = (float(amount), cols.get("currency") or "")
except ValueError:
pass
note(f"Loaded prices for {len(prices)} licenses from {path}")
return prices
def parse_args(argv):
parser = argparse.ArgumentParser(
prog="ghostbuster-entra",
description="Find ghost accounts, risky access and license waste in Microsoft Entra ID. Read-only.")
parser.add_argument("--tenant", default="organizations",
help="tenant to sign in to: domain (contoso.com) or tenant ID (default: your account's tenant)")
parser.add_argument("--days", type=int, default=90, help="activity window in days, 1-365 (default 90)")
parser.add_argument("--mapping", metavar="CSV",
help="CSV from HR: user_principal_name (or email) and employment_status (active, left or "
"service). people.csv from an earlier run works.")
parser.add_argument("--prices", metavar="CSV",
help="CSV with sku, monthly_price and currency, to put a price on unused licenses")
parser.add_argument("--usage-report", action="store_true",
help="also read the Microsoft 365 usage report, even when sign-in activity is available")
parser.add_argument("--device-code", action="store_true", help="sign in with a code instead of a browser window")
parser.add_argument("--client-id", default=DEFAULT_CLIENT_ID,
help="app (client) ID to sign in with, if you'd rather use your own app registration")
parser.add_argument("--out", metavar="DIR", help="output folder (default ./ghostbuster-entra--)")
parser.add_argument("--page-size", type=int, default=500, help=argparse.SUPPRESS)
parser.add_argument("--no-open", action="store_true", help="don't open the report in a browser when done")
parser.add_argument("--version", action="version",
version=f"ghostbuster-entra {VERSION}, created by Jens Naterman at Adcyma. MIT license. {HOME_URL}")
args = parser.parse_args(argv)
args.days = max(1, min(args.days, 365))
return args
def main(argv=None):
_setup_console()
if sys.version_info < (3, 8):
die("Ghostbuster needs Python 3.8 or newer.")
args = parse_args(argv)
say(paint(f"{TOOL_NAME} {VERSION}", "1") + paint(" | read-only access review | by Jens Naterman, Adcyma", "2"))
mapping = load_mapping(args.mapping) if args.mapping else None
prices = load_prices(args.prices) if args.prices else None
try:
token = sign_in(args)
except AuthError as err:
die(f"Sign-in failed: {err}")
graph = Graph(token["access_token"])
now = utcnow()
started = time.time()
scan = collect(graph, args, now, granted_scopes(token))
note(f"Signed in as {scan['me'].get('userPrincipalName', '?')} to {scan['org'].get('displayName', '?')}")
report = analyze(scan, mapping, prices)
slug = re.sub(r"[^a-z0-9]+", "-", (report["domain"] or report["tenant"]).lower()).strip("-") or "tenant"
out_dir = args.out or f"ghostbuster-entra-{slug}-{now:%Y%m%d}"
write_outputs(report, out_dir)
note(f"Done in {int(time.time() - started)}s using {graph.calls} Microsoft Graph calls.")
print_summary(report, out_dir)
report_path = os.path.abspath(os.path.join(out_dir, "report.html"))
if not args.no_open:
target = report_path if os.name == "nt" else "file://" + urllib.request.pathname2url(report_path)
if webbrowser.open(target):
note("Opened the report in your browser. Use its Save as PDF button to keep a copy.")
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nCancelled.", file=sys.stderr)
sys.exit(130)
except GraphError as error:
die(f"Microsoft Graph error: {error.reason()} ({error.url})")