claudetools/clients/cascades-tucson/reports/user-detail-batches/batch-4-collect.py

#!/usr/bin/env python3
"""Collect M365 user detail for batch 4 of Cascades Tucson."""
import csv
import json
import sys
import time
import urllib.request
import urllib.error

TOKEN = open(r"C:\claudetools\clients\cascades-tucson\reports\user-detail-batches\.token").read().strip()
OUT = r"C:\claudetools\clients\cascades-tucson\reports\user-detail-batches\batch-4.csv"

USERS = [
    "jodi.ramstack@cascadestucson.com",
    "john.trozzi@cascadestucson.com",
    "karen.rossini@cascadestucson.com",
    "lauren.hasselman@cascadestucson.com",
    "lois.lane@cascadestucson.com",
]

HEADERS = {"Authorization": f"Bearer {TOKEN}"}

MFA_TYPE_MAP = {
    "#microsoft.graph.microsoftAuthenticatorAuthenticationMethod": "Authenticator",
    "#microsoft.graph.phoneAuthenticationMethod": "SMS",
    "#microsoft.graph.fido2AuthenticationMethod": "FIDO2",
    "#microsoft.graph.temporaryAccessPassAuthenticationMethod": "TAP",
    "#microsoft.graph.softwareOathAuthenticationMethod": "OATH",
    "#microsoft.graph.passwordAuthenticationMethod": "Password",
    "#microsoft.graph.emailAuthenticationMethod": "Email",
    "#microsoft.graph.windowsHelloForBusinessAuthenticationMethod": "WindowsHello",
    "#microsoft.graph.passwordlessMicrosoftAuthenticatorAuthenticationMethod": "PasswordlessAuthenticator",
}


def _do(url, extra_headers=None):
    hdrs = dict(HEADERS)
    if extra_headers:
        hdrs.update(extra_headers)
    backoff = 2
    for attempt in range(6):
        req = urllib.request.Request(url, headers=hdrs)
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                return r.status, json.loads(r.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            if e.code == 429 or e.code >= 500:
                retry_after = e.headers.get("Retry-After")
                wait = int(retry_after) if retry_after and retry_after.isdigit() else backoff
                print(f"  [retry {attempt+1}] HTTP {e.code}, waiting {wait}s...", file=sys.stderr)
                time.sleep(wait)
                backoff = min(backoff * 2, 60)
                continue
            try:
                body = json.loads(e.read().decode("utf-8"))
            except Exception:
                body = {"error": str(e)}
            return e.code, body
        except Exception as e:
            print(f"  [retry {attempt+1}] transport err: {e}", file=sys.stderr)
            time.sleep(backoff)
            backoff = min(backoff * 2, 60)
            continue
    return 0, {"error": "max_retries_exceeded"}


def gget(url):
    return _do(url)


def gget_consistency(url):
    return _do(url, {"ConsistencyLevel": "eventual"})


def collect_user(upn):
    row = {
        "UPN": upn, "DisplayName": "", "AccountEnabled": "",
        "MailboxType": "User", "Licenses": "", "MFARegistered": "",
        "MFAMethods": "", "DefaultMFAMethod": "", "LastSignIn": "",
        "LastInteractiveSignIn": "", "AdminRoles": "", "JobTitle": "",
        "Department": "", "GroupCount": "", "Notes": "",
    }
    notes = []

    # 1a. Basic profile by UPN (signInActivity requires GUID addressing, fetched below)
    status, body = gget(
        f"https://graph.microsoft.com/v1.0/users/{upn}"
        "?$select=id,userPrincipalName,displayName,accountEnabled,jobTitle,department"
    )
    if status == 404:
        row["Notes"] = "not_found"
        return row
    if status != 200:
        row["Notes"] = f"profile_error:{status}"
        return row
    user_id = body.get("id", "")
    row["DisplayName"] = body.get("displayName", "") or ""
    row["AccountEnabled"] = str(body.get("accountEnabled", "")).lower()
    row["JobTitle"] = body.get("jobTitle", "") or ""
    row["Department"] = body.get("department", "") or ""

    # 1b. signInActivity - requires GUID addressing
    if user_id:
        status, body = gget(
            f"https://graph.microsoft.com/v1.0/users/{user_id}?$select=signInActivity"
        )
        if status == 200:
            sia = body.get("signInActivity") or {}
            # lastSignInDateTime = last interactive sign-in per MS docs
            row["LastInteractiveSignIn"] = sia.get("lastSignInDateTime", "") or ""
            # LastSignIn = most recent of either (interactive or non-interactive)
            li = sia.get("lastSignInDateTime", "") or ""
            lni = sia.get("lastNonInteractiveSignInDateTime", "") or ""
            row["LastSignIn"] = max(li, lni) if (li or lni) else ""
        elif status == 403:
            row["LastSignIn"] = "scope_unavailable"
            row["LastInteractiveSignIn"] = "scope_unavailable"
        else:
            notes.append(f"signInActivity:{status}")

    # 2. Licenses
    status, body = gget(f"https://graph.microsoft.com/v1.0/users/{upn}/licenseDetails")
    if status == 200:
        skus = [x.get("skuPartNumber", "") for x in body.get("value", []) if x.get("skuPartNumber")]
        row["Licenses"] = ";".join(skus)
    elif status == 403:
        row["Licenses"] = "scope_unavailable"
    else:
        notes.append(f"licenses:{status}")

    # 3. MFA methods
    status, body = gget(f"https://graph.microsoft.com/v1.0/users/{upn}/authentication/methods")
    methods = []
    if status == 200:
        for m in body.get("value", []):
            t = m.get("@odata.type", "")
            methods.append(MFA_TYPE_MAP.get(t, t.replace("#microsoft.graph.", "").replace("AuthenticationMethod", "")))
        row["MFAMethods"] = ";".join(methods)
        non_password = [m for m in methods if m != "Password"]
        row["MFARegistered"] = "true" if non_password else "false"
    elif status == 403:
        row["MFAMethods"] = "scope_unavailable"
        row["MFARegistered"] = "scope_unavailable"
    else:
        notes.append(f"methods:{status}")

    # 4. Default MFA method
    status, body = gget(f"https://graph.microsoft.com/beta/users/{upn}/authentication/signInPreferences")
    if status == 200:
        row["DefaultMFAMethod"] = body.get("userPreferredMethodForSecondaryAuthentication", "") or ""
    elif status == 403:
        row["DefaultMFAMethod"] = "scope_unavailable"
    else:
        notes.append(f"signInPrefs:{status}")

    # 5. Directory roles
    status, body = gget(
        f"https://graph.microsoft.com/v1.0/users/{upn}/transitiveMemberOf/microsoft.graph.directoryRole"
    )
    if status == 200:
        roles = [r.get("displayName", "") for r in body.get("value", []) if r.get("displayName")]
        row["AdminRoles"] = ";".join(roles)
    elif status == 403:
        row["AdminRoles"] = "scope_unavailable"
    else:
        notes.append(f"roles:{status}")

    # 6. Group count
    status, body = gget_consistency(
        f"https://graph.microsoft.com/v1.0/users/{upn}/memberOf?$count=true&$top=1"
    )
    if status == 200:
        row["GroupCount"] = str(body.get("@odata.count", ""))
    elif status == 403:
        row["GroupCount"] = "scope_unavailable"
    else:
        notes.append(f"groups:{status}")

    # Special note for jodi.ramstack
    if upn.lower().startswith("jodi.ramstack"):
        jodi_state = []
        jodi_state.append(f"flagged_for_delete_2026-04-13:enabled={row['AccountEnabled']}")
        jodi_state.append(f"licensed={'yes' if row['Licenses'] and row['Licenses'] != 'scope_unavailable' else 'no'}")
        notes.extend(jodi_state)

    if notes:
        row["Notes"] = ";".join(notes)

    return row


def main():
    rows = []
    for upn in USERS:
        print(f"Processing {upn}...", file=sys.stderr)
        row = collect_user(upn)
        rows.append(row)
        time.sleep(0.5)

    cols = ["UPN", "DisplayName", "AccountEnabled", "MailboxType", "Licenses",
            "MFARegistered", "MFAMethods", "DefaultMFAMethod", "LastSignIn",
            "LastInteractiveSignIn", "AdminRoles", "JobTitle", "Department",
            "GroupCount", "Notes"]
    with open(OUT, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=cols, quoting=csv.QUOTE_MINIMAL)
        w.writeheader()
        for r in rows:
            w.writerow(r)
    print(f"Wrote {len(rows)} rows to {OUT}", file=sys.stderr)
    # Print to stdout for visibility
    for r in rows:
        print(json.dumps(r))


if __name__ == "__main__":
    main()