scripts/users_and_apps_audit.py — Users and Apps Audit

#!/usr/bin/env python3
# By Zero Width Lab (zerowidthlab.com)
"""Read-only Microsoft Graph users and applications audit."""
from __future__ import annotations

import argparse
import errno
import hashlib
import json
import os
import sys
import time
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

__author__ = "Zero Width Lab"

SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
HOSTS = {"graph.microsoft.com", "login.microsoftonline.com"}
FAKE_FILE = "GRAPH_AUDIT_FAKE_RESPONSES"
FAKE_ACK = "GRAPH_AUDIT_FAKE_TRANSPORT_TEST_ACK"
MAX_INPUT_BYTES = 8 * 1024 * 1024
SECTIONS = ("users", "licenses", "apps", "roles", "guests")
PRIVILEGED_ROLES = {
    "Global Administrator", "Privileged Role Administrator", "User Administrator",
    "Application Administrator", "Cloud Application Administrator",
    "Security Administrator", "Exchange Administrator", "SharePoint Administrator",
    "Authentication Administrator", "Privileged Authentication Administrator",
}


class AuditError(Exception):
    def __init__(self, message: str, status: int | None = None) -> None:
        super().__init__(message)
        self.status = status


def bounded_json(stream: Any) -> Any:
    data = stream.read(MAX_INPUT_BYTES + 1)
    if len(data) > MAX_INPUT_BYTES:
        raise AuditError("JSON input exceeds 8 MiB limit")
    try:
        return json.loads(data)
    except (ValueError, UnicodeError, RecursionError) as error:
        raise AuditError("invalid JSON response or fixture") from error


def validate_remote_path(path: str) -> None:
    for segment in path.split("/"):
        decoded = segment
        for _ in range(8):
            newer = urllib.parse.unquote(decoded)
            if newer == decoded:
                break
            decoded = newer
        if ("%" in decoded or "/" in decoded or "\\" in decoded or
                decoded in (".", "..") or
                any(unicodedata.category(character) == "Cc" for character in decoded)):
            raise AuditError("remote path traversal, control characters or encoded separators are forbidden")
    if "//" in path:
        raise AuditError("remote path traversal is forbidden")


def validate_record(record: dict[str, Any]) -> None:
    for field in ("assignedLicenses", "passwordCredentials", "keyCredentials"):
        value = record.get(field)
        if value is not None and (not isinstance(value, list) or any(not isinstance(v, dict) for v in value)):
            raise AuditError("invalid Graph field: " + field)
    for field in ("signInActivity", "prepaidUnits"):
        if record.get(field) is not None and not isinstance(record[field], dict):
            raise AuditError("invalid Graph field: " + field)
    for field in ("id", "displayName", "userPrincipalName", "userType", "createdDateTime", "skuId", "skuPartNumber"):
        if record.get(field) is not None and not isinstance(record[field], str):
            raise AuditError("invalid Graph field: " + field)
    for field in ("assignedLicenses", "passwordCredentials", "keyCredentials"):
        for nested in record.get(field) or []:
            validate_record(nested)
    for field in ("signInActivity", "prepaidUnits"):
        if record.get(field) is not None:
            validate_record(record[field])
    for field in ("lastSignInDateTime", "endDateTime"):
        if record.get(field) is not None and not isinstance(record[field], str):
            raise AuditError("invalid Graph timestamp: " + field)
    if record.get("accountEnabled") is not None and not isinstance(record["accountEnabled"], bool):
        raise AuditError("invalid Graph field: accountEnabled")
    for field in ("consumedUnits", "enabled"):
        if record.get(field) is not None and (not isinstance(record[field], int) or isinstance(record[field], bool)):
            raise AuditError("invalid Graph field: " + field)


def display_text(value: Any) -> str:
    parts = []
    for character in str(value):
        if unicodedata.category(character).startswith("C") or character in "\u2028\u2029":
            point = ord(character)
            parts.append(f"\\u{point:04x}" if point <= 0xffff else f"\\U{point:08x}")
        else:
            parts.append(character)
    return "".join(parts)


def guarded_url(url: str, host: str | None = None) -> None:
    try:
        parsed = urllib.parse.urlsplit(url)
        validate_remote_path(parsed.path)
        port = parsed.port
    except ValueError as error:
        raise AuditError("invalid destination URL") from error
    if (parsed.scheme != "https" or parsed.hostname not in HOSTS or
            (host is not None and parsed.hostname != host) or
            port not in (None, 443) or parsed.username or parsed.password or
            parsed.fragment):
        raise AuditError("destination is outside the declared HTTPS hosts")


class GuardedRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req: Any, fp: Any, code: int, msg: str,
                         headers: Any, newurl: str) -> Any:
        previous = urllib.parse.urlsplit(req.full_url).hostname
        guarded_url(newurl, previous)
        return super().redirect_request(req, fp, code, msg, headers, newurl)


class GraphClient:
    def __init__(self) -> None:
        fixture_path = os.environ.get(FAKE_FILE)
        self.fake: dict[str, list[dict[str, Any]]] | None = None
        if fixture_path:
            if os.environ.get(FAKE_ACK) != "1":
                raise AuditError(f"{FAKE_FILE} requires {FAKE_ACK}=1")
            print(f"warning: {FAKE_FILE} is set; all requests use a synthetic fixture; no network requests are made", file=sys.stderr)
            try:
                with open(fixture_path, "rb") as stream:
                    fixture = bounded_json(stream)
                if not isinstance(fixture, dict) or not all(isinstance(k, str) and isinstance(v, list) and all(isinstance(r, dict) for r in v) for k, v in fixture.items()):
                    raise ValueError("fixture must map URLs to response lists")
                self.fake = fixture
            except (OSError, ValueError, TypeError) as error:
                raise AuditError("invalid or missing fake response fixture") from error
        self.opener = urllib.request.build_opener(GuardedRedirect())
        self.token: str | None = None

    def request(self, url: str, *, method: str = "GET", data: bytes | None = None,
                host: str | None = None) -> dict[str, Any]:
        guarded_url(url, host)
        for attempt in range(4):
            if self.fake is not None:
                queue = self.fake.get(f"{method} {url}")
                if not queue:
                    raise AuditError("fixture response missing for request")
                response = queue.pop(0)
                status = response.get("status", 200)
                body = response.get("body", {})
                retry_after = response.get("retry_after", 0)
            else:
                headers = {"Accept": "application/json"}
                if self.token:
                    headers["Authorization"] = f"Bearer {self.token}"
                if data is not None:
                    headers["Content-Type"] = "application/x-www-form-urlencoded"
                request = urllib.request.Request(url, data=data, headers=headers, method=method)
                try:
                    with self.opener.open(request, timeout=30) as opened:
                        status = opened.status
                        body = bounded_json(opened)
                        retry_after = opened.headers.get("Retry-After", "0")
                except urllib.error.HTTPError as error:
                    status = error.code
                    retry_after = error.headers.get("Retry-After", "0")
                    body = {}
                except (urllib.error.URLError, TimeoutError, ValueError) as error:
                    raise AuditError("Graph request failed") from error
            if not isinstance(status, int) or isinstance(status, bool):
                raise AuditError("invalid response status")
            if method == "GET" and status in (429, 500, 502, 503, 504) and attempt < 3:
                if self.fake is None:
                    try:
                        delay = min(max(float(retry_after), 0), 5)
                    except (ValueError, TypeError):
                        delay = 1
                    time.sleep(delay or attempt + 1)
                continue
            if status < 200 or status >= 300:
                raise AuditError(f"Graph returned HTTP {status}", status)
            if not isinstance(body, dict):
                raise AuditError("Graph response is not an object")
            return body
        raise AuditError("Graph request retries exhausted")

    def authenticate(self) -> None:
        tenant = os.environ.get("GRAPH_TENANT_ID", "")
        client = os.environ.get("GRAPH_CLIENT_ID", "")
        secret = os.environ.get("GRAPH_CLIENT_SECRET", "")
        if not tenant or not client or not secret:
            raise AuditError("GRAPH_TENANT_ID, GRAPH_CLIENT_ID and GRAPH_CLIENT_SECRET are required")
        if not tenant.isascii() or not all(c.isalnum() or c in "-." for c in tenant) or tenant in (".", ".."):
            raise AuditError("invalid tenant identifier")
        url = f"https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token"
        data = urllib.parse.urlencode({"client_id": client, "client_secret": secret,
            "scope": "https://graph.microsoft.com/.default", "grant_type": "client_credentials"}).encode()
        result = self.request(url, method="POST", data=data, host="login.microsoftonline.com")
        token = result.get("access_token")
        if not isinstance(token, str) or not token:
            raise AuditError("token response is missing an access token")
        self.token = token

    def pages(self, path: str) -> list[dict[str, Any]]:
        url = "https://graph.microsoft.com/v1.0/" + path
        seen: set[str] = set()
        items: list[dict[str, Any]] = []
        while url:
            guarded_url(url, "graph.microsoft.com")
            if url in seen or len(seen) >= 1000:
                raise AuditError("pagination loop or limit reached")
            seen.add(url)
            result = self.request(url, host="graph.microsoft.com")
            values = result.get("value")
            if not isinstance(values, list) or any(not isinstance(v, dict) for v in values):
                raise AuditError("Graph page is missing a valid value list")
            for value in values:
                validate_record(value)
            items.extend(values)
            next_url = result.get("@odata.nextLink")
            if next_url is not None and not isinstance(next_url, str):
                raise AuditError("invalid pagination URL")
            url = next_url or ""
        return items


def days_until(value: Any, now: datetime) -> int | None:
    if not isinstance(value, str):
        raise AuditError("invalid credential expiry timestamp")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
        if parsed.tzinfo is None:
            raise AuditError("credential expiry timestamp has no timezone")
        return (parsed.astimezone(timezone.utc).date() - now.date()).days
    except ValueError as error:
        raise AuditError("invalid credential expiry timestamp") from error


def audit(client: GraphClient, selected: list[str], window: int, now: datetime) -> dict[str, Any]:
    result: dict[str, Any] = {"status": "complete", "expiring_days": window,
        "generated_at": now.date().isoformat(), "sections": {}, "skipped": []}
    users: list[dict[str, Any]] | None = None
    if any(s in selected for s in ("users", "licenses", "guests")):
        try:
            users = client.pages("users?$select=id,displayName,userPrincipalName,accountEnabled,userType,createdDateTime,assignedLicenses")
        except AuditError as error:
            for section in ("users", "licenses", "guests"):
                if section in selected:
                    result["skipped"].append(f"{section}: {error}")
            result["status"] = "incomplete"
    if "users" in selected and users is not None:
        summary: dict[str, Any] = {"enabled": sum(u.get("accountEnabled") is True for u in users),
            "disabled": sum(u.get("accountEnabled") is False for u in users),
            "guests": sum(u.get("userType") == "Guest" for u in users)}
        try:
            activity = client.pages("users?$select=id,displayName,userPrincipalName,signInActivity")
            cutoff = now - timedelta(days=window)
            stale = []
            for user in activity:
                last = (user.get("signInActivity") or {}).get("lastSignInDateTime")
                if last is None:
                    continue
                if not isinstance(last, str) or not last:
                    raise AuditError("invalid sign-in timestamp")
                try:
                    stamp = datetime.fromisoformat(last.replace("Z", "+00:00"))
                    if stamp.tzinfo is None:
                        raise AuditError("sign-in timestamp has no timezone")
                    if stamp < cutoff:
                        stale.append({"display_name": user.get("displayName"), "upn": user.get("userPrincipalName"), "last_sign_in": last})
                except (ValueError, AttributeError) as error:
                    raise AuditError("invalid sign-in timestamp") from error
            summary["no_sign_in_for_days"] = stale
        except (AuditError, AttributeError, TypeError) as error:
            if getattr(error, "status", None) == 403:
                summary["no_sign_in_for_days"] = "skipped (permission missing: AuditLog.Read.All)"
            else:
                result["status"] = "incomplete"
                result["skipped"].append(f"users sign-in activity: {error}")
        result["sections"]["users"] = summary
    if "licenses" in selected and users is not None:
        try:
            skus = client.pages("subscribedSkus")
            result["sections"]["licenses"] = [{"sku": s.get("skuPartNumber"),
                "prepaid": (s.get("prepaidUnits") or {}).get("enabled"),
                "consumed": s.get("consumedUnits"),
                "users": sum(s.get("skuId") == license.get("skuId") for u in users
                             for license in (u.get("assignedLicenses") or []))} for s in skus]
        except (AuditError, AttributeError, TypeError) as error:
            result["status"] = "incomplete"
            result["skipped"].append(f"licenses: {error}")
    if "guests" in selected and users is not None:
        result["sections"]["guests"] = [{"display_name": u.get("displayName"),
            "upn": u.get("userPrincipalName"), "created": u.get("createdDateTime")}
            for u in users if u.get("userType") == "Guest"]
    if "apps" in selected:
        try:
            entries = []
            for kind, path in (("application", "applications?$select=displayName,passwordCredentials,keyCredentials"),
                               ("service_principal", "servicePrincipals?$select=displayName,passwordCredentials,keyCredentials")):
                for app in client.pages(path):
                    for credential_type in ("passwordCredentials", "keyCredentials"):
                        for credential in app.get(credential_type) or []:
                            remaining = days_until(credential.get("endDateTime"), now)
                            if remaining is not None and remaining <= window:
                                entries.append({"kind": kind, "display_name": app.get("displayName"),
                                    "credential_type": credential_type, "expires": credential.get("endDateTime"),
                                    "days_until_expiry": remaining})
            result["sections"]["apps"] = entries
        except (AuditError, AttributeError, TypeError) as error:
            result["status"] = "incomplete"
            result["skipped"].append(f"apps: {error}")
    if "roles" in selected:
        try:
            memberships = []
            for role in client.pages("directoryRoles?$select=id,displayName"):
                if role.get("displayName") in PRIVILEGED_ROLES:
                    identifier = role.get("id")
                    if not isinstance(identifier, str) or not identifier or not all(c.isascii() and (c.isalnum() or c == "-") for c in identifier):
                        raise AuditError("invalid role identifier")
                    role_id = identifier
                    for member in client.pages(f"directoryRoles/{role_id}/members?$select=displayName,userPrincipalName"):
                        memberships.append({"role": role.get("displayName"), "display_name": member.get("displayName"),
                            "upn": member.get("userPrincipalName")})
            result["sections"]["roles"] = memberships
        except AuditError as error:
            result["status"] = "incomplete"
            result["skipped"].append(f"roles: {error}")
    return result


def render_text(report: dict[str, Any]) -> str:
    lines = ["Users and Apps Audit", f"Status: {report['status']}",
             f"Generated: {report['generated_at']}", f"Expiry window: {report['expiring_days']} days"]
    for name, value in report["sections"].items():
        lines.append(f"{name.upper()}:")
        if isinstance(value, list):
            for row in value:
                lines.append("  " + ", ".join(f"{display_text(k)}={display_text(v)}" for k, v in row.items()))
        else:
            for key, entry in value.items():
                if isinstance(entry, list):
                    lines.append(f"  {key}:")
                    for row in entry:
                        lines.append("    " + ", ".join(f"{display_text(k)}={display_text(v)}" for k, v in row.items()))
                else:
                    lines.append(f"  {key}: {display_text(entry)}")
    for skipped in report["skipped"]:
        lines.append("Skipped: " + display_text(skipped))
    return "\n".join(lines) + "\n"


def require_platform() -> None:
    if (os.name != "posix" or sys.platform not in ("darwin", "linux") or not hasattr(os, "O_DIRECTORY") or
            not hasattr(os, "O_NOFOLLOW") or os.open not in os.supports_dir_fd):
        raise AuditError("supported platforms: macOS and Linux with POSIX directory descriptors and no-follow APIs")


def identity(path: Path) -> list[int]:
    stat = path.stat()
    if not path.is_dir():
        raise AuditError("output directory must already exist")
    return [stat.st_dev, stat.st_ino]


def output_plan(path: str) -> dict[str, Any]:
    absolute = Path(os.path.abspath(path))
    resolved = absolute.resolve(strict=False)
    if os.path.lexists(absolute):
        raise AuditError("output path already exists")
    return {"selected": str(absolute), "resolved": str(resolved),
            "directory": identity(resolved.parent), "parent": identity(resolved.parent.parent)}


def create_output(plan: dict[str, Any], content: str) -> None:
    selected = Path(plan["selected"])
    descriptor = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        for component in Path(plan["resolved"]).parent.parts[1:]:
            next_fd = os.open(component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor)
            os.close(descriptor)
            descriptor = next_fd
        stat = os.fstat(descriptor)
        parent_fd = os.open("..", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor)
        try:
            parent_stat = os.fstat(parent_fd)
        finally:
            os.close(parent_fd)
        if ([stat.st_dev, stat.st_ino] != plan["directory"] or
                [parent_stat.st_dev, parent_stat.st_ino] != plan["parent"] or
                str(selected.resolve(strict=False)) != plan["resolved"]):
            raise AuditError("output directory identity changed; run a new preview")
        file_fd = os.open(selected.name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
                          0o600, dir_fd=descriptor)
        with os.fdopen(file_fd, "w", encoding="utf-8") as stream:
            stream.write(content)
    except OSError as error:
        if error.errno in (errno.ENOTDIR, errno.ELOOP):
            raise AuditError("output path component is a symlink or is no longer a directory; run a new preview") from error
        raise
    finally:
        os.close(descriptor)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Read-only Microsoft Graph users and apps report", epilog=SIGNATURE)
    sub = parser.add_subparsers(dest="command", required=True)
    report_parser = sub.add_parser("report", epilog=SIGNATURE)
    report_parser.add_argument("--expiring-days", type=int, default=30)
    report_parser.add_argument("--json", action="store_true")
    report_parser.add_argument("--out")
    report_parser.add_argument("--sections", default=",".join(SECTIONS))
    report_parser.add_argument("--allow-partial", action="store_true")
    report_parser.add_argument("--dry-run", action="store_true", help="preview output and approval digest without writing")
    report_parser.add_argument("--approve", help="digest printed by --dry-run for --out")
    args = parser.parse_args(argv)
    try:
        require_platform()
        plan = output_plan(args.out) if args.out else None
        if args.expiring_days < 0:
            raise AuditError("--expiring-days must be nonnegative")
        selected = args.sections.split(",")
        if not selected or any(s not in SECTIONS for s in selected) or len(set(selected)) != len(selected):
            raise AuditError("--sections must be a comma-separated subset without duplicates")
        if args.approve and not args.out:
            raise AuditError("--approve requires --out")
        if args.dry_run and args.approve:
            raise AuditError("--dry-run cannot be combined with --approve")
        if args.out and os.path.lexists(args.out):
            raise AuditError("output path already exists")
        client = GraphClient()
        client.authenticate()
        result = audit(client, selected, args.expiring_days, datetime.now(timezone.utc))
        output = json.dumps(result, indent=2, ensure_ascii=False) + "\n" if args.json else render_text(result)
        if args.out:
            digest = hashlib.sha256((json.dumps(plan, sort_keys=True) + "\n" + output).encode("utf-8")).hexdigest()
            if args.dry_run:
                sys.stdout.write(output)
                print("Approval digest: " + digest)
                return 2 if result["status"] == "incomplete" and not args.allow_partial else 0
            if args.approve != digest:
                raise AuditError("output approval digest missing or different; run --dry-run with the same options first")
            create_output(plan, output)
        else:
            sys.stdout.write(output)
        return 2 if result["status"] == "incomplete" and not args.allow_partial else 0
    except (AuditError, OSError) as error:
        print("error: " + display_text(error), file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

License: MIT (read it).

Back to the listing