tests/test_users_and_apps_audit.py — Users and Apps Audit

#!/usr/bin/env python3
# By Zero Width Lab (zerowidthlab.com)
"""Synthetic CLI behavior tests."""
from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
from datetime import datetime, timezone
import importlib.util
from pathlib import Path

__author__ = "Zero Width Lab"

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "users_and_apps_audit.py"
GRAPH = "https://graph.microsoft.com/v1.0/"
TOKEN = "https://login.microsoftonline.com/example-tenant/oauth2/v2.0/token"
SPEC = importlib.util.spec_from_file_location("users_and_apps_audit", SCRIPT)
assert SPEC and SPEC.loader
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)


def fixture() -> dict[str, list[dict]]:
    return {
        "POST " + TOKEN: [{"body": {"access_token": "synthetic-token"}}],
        "GET " + GRAPH + "users?$select=id,displayName,userPrincipalName,accountEnabled,userType,createdDateTime,assignedLicenses": [
            {"body": {"value": [{"id": "one", "displayName": "Alice\nAdmin", "userPrincipalName": "alice-at-example", "accountEnabled": True, "userType": "Member", "assignedLicenses": [{"skuId": "sku-a"}]}], "@odata.nextLink": GRAPH + "users?page=2"}}],
        "GET " + GRAPH + "users?page=2": [{"body": {"value": [{"id": "two", "displayName": "Guest", "userPrincipalName": "guest-at-example", "accountEnabled": False, "userType": "Guest", "createdDateTime": "2025-01-01T00:00:00Z"}]}}],
        "GET " + GRAPH + "users?$select=id,displayName,userPrincipalName,signInActivity": [{"status": 403}],
        "GET " + GRAPH + "subscribedSkus": [{"body": {"value": [{"skuId": "sku-a", "skuPartNumber": "SYNTHETIC", "prepaidUnits": {"enabled": 3}, "consumedUnits": 1}]}}],
        "GET " + GRAPH + "applications?$select=displayName,passwordCredentials,keyCredentials": [{"body": {"value": [{"displayName": "Demo", "passwordCredentials": [{"endDateTime": "2026-09-30T00:00:00Z"}], "keyCredentials": []}]}}],
        "GET " + GRAPH + "servicePrincipals?$select=displayName,passwordCredentials,keyCredentials": [{"body": {"value": []}}],
        "GET " + GRAPH + "directoryRoles?$select=id,displayName": [{"body": {"value": [{"id": "role-a", "displayName": "Global Administrator"}]}}],
        "GET " + GRAPH + "directoryRoles/role-a/members?$select=displayName,userPrincipalName": [{"body": {"value": [{"displayName": "Alice\nAdmin", "userPrincipalName": "alice-at-example"}]}}],
    }


class ReportTests(unittest.TestCase):
    def run_cli(self, responses: dict, *args: str, fixture_exists: bool = True) -> subprocess.CompletedProcess[str]:
        with tempfile.TemporaryDirectory() as temp:
            path = Path(temp) / "responses.json"
            if fixture_exists:
                path.write_text(json.dumps(responses), encoding="utf-8")
            env = {key: value for key, value in os.environ.items() if key not in
                   {"GRAPH_AUDIT_FAKE_RESPONSES", "GRAPH_AUDIT_FAKE_TRANSPORT_TEST_ACK"}}
            env.update({"GRAPH_TENANT_ID": "example-tenant", "GRAPH_CLIENT_ID": "synthetic-client",
                        "GRAPH_CLIENT_SECRET": "synthetic-secret", "GRAPH_AUDIT_FAKE_RESPONSES": str(path),
                        "GRAPH_AUDIT_FAKE_TRANSPORT_TEST_ACK": "1"})
            return subprocess.run([sys.executable, str(SCRIPT), "report", *args],
                                  env=env, text=True, capture_output=True, check=False)

    def test_pagination_sections_and_optional_403(self) -> None:
        completed = self.run_cli(fixture(), "--json")
        self.assertEqual(completed.returncode, 0, completed.stderr)
        report = json.loads(completed.stdout)
        self.assertEqual(report["status"], "complete")
        self.assertEqual(report["sections"]["users"]["enabled"], 1)
        self.assertEqual(report["sections"]["users"]["guests"], 1)
        self.assertIn("permission missing", report["sections"]["users"]["no_sign_in_for_days"])
        self.assertEqual(report["sections"]["licenses"][0]["users"], 1)
        self.assertEqual(report["sections"]["guests"][0]["created"], "2025-01-01T00:00:00Z")
        self.assertEqual(len(report["sections"]["roles"]), 1)

    def test_expiry_math(self) -> None:
        now = datetime(2026, 9, 30, 12, tzinfo=timezone.utc)
        self.assertEqual(MODULE.days_until("2026-09-30T00:00:00Z", now), 0)
        self.assertEqual(MODULE.days_until("2026-10-30T23:00:00Z", now), 30)
        self.assertEqual(MODULE.days_until("2026-09-29T00:00:00Z", now), -1)

    def test_apps_include_expired_and_day_30_but_exclude_day_31(self) -> None:
        class SyntheticClient:
            def pages(self, path: str) -> list[dict]:
                if path.startswith("applications?"):
                    return [{"displayName": "Registration", "passwordCredentials": [
                        {"endDateTime": "2026-09-29T00:00:00Z"},
                        {"endDateTime": "2026-10-30T23:00:00Z"},
                        {"endDateTime": "2026-10-31T00:00:00Z"}],
                        "keyCredentials": [{"endDateTime": "2026-10-30T00:00:00Z"}]}]
                if path.startswith("servicePrincipals?"):
                    return [{"displayName": "Principal", "passwordCredentials": [
                        {"endDateTime": "2026-10-30T00:00:00Z"}],
                        "keyCredentials": [{"endDateTime": "2026-10-31T00:00:00Z"}]}]
                raise AssertionError("unexpected page")

        report = MODULE.audit(SyntheticClient(), ["apps"], 30,
                              datetime(2026, 9, 30, 12, tzinfo=timezone.utc))
        self.assertEqual(report["status"], "complete")
        self.assertEqual(report["sections"]["apps"], [
            {"kind": "application", "display_name": "Registration", "credential_type": "passwordCredentials",
             "expires": "2026-09-29T00:00:00Z", "days_until_expiry": -1},
            {"kind": "application", "display_name": "Registration", "credential_type": "passwordCredentials",
             "expires": "2026-10-30T23:00:00Z", "days_until_expiry": 30},
            {"kind": "application", "display_name": "Registration", "credential_type": "keyCredentials",
             "expires": "2026-10-30T00:00:00Z", "days_until_expiry": 30},
            {"kind": "service_principal", "display_name": "Principal", "credential_type": "passwordCredentials",
             "expires": "2026-10-30T00:00:00Z", "days_until_expiry": 30},
        ])

    def test_redirect_checks_allowed_and_refused_targets(self) -> None:
        handler = MODULE.GuardedRedirect()
        request = MODULE.urllib.request.Request(GRAPH + "users")
        allowed = handler.redirect_request(request, None, 302, "Found", {}, GRAPH + "users?page=2")
        self.assertEqual(allowed.full_url, GRAPH + "users?page=2")
        for target in ("https://example.invalid/users", "http://graph.microsoft.com/v1.0/users",
                       "https://login.microsoftonline.com/oauth2/token"):
            with self.subTest(target=target), self.assertRaises(MODULE.AuditError):
                handler.redirect_request(request, None, 302, "Found", {}, target)

    def test_guid_and_domain_tenant_identifiers(self) -> None:
        synthetic_guid = "12345678-1234-" + "1234-1234-123456789abc"
        for tenant in (synthetic_guid, "example.test-tenant"):
            with self.subTest(tenant=tenant), patch.dict(os.environ, {
                    "GRAPH_TENANT_ID": tenant, "GRAPH_CLIENT_ID": "synthetic-client",
                    "GRAPH_CLIENT_SECRET": "synthetic-secret"}):
                client = MODULE.GraphClient()
                with patch.object(client, "request", return_value={"access_token": "synthetic-token"}) as request:
                    client.authenticate()
                self.assertEqual(request.call_args.args[0],
                                 f"https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token")
        for tenant in ("example/test", "example?test", "example#test", "example test", "example_test"):
            with self.subTest(tenant=tenant), patch.dict(os.environ, {
                    "GRAPH_TENANT_ID": tenant, "GRAPH_CLIENT_ID": "synthetic-client",
                    "GRAPH_CLIENT_SECRET": "synthetic-secret"}):
                client = MODULE.GraphClient()
                with patch.object(client, "request") as request:
                    with self.assertRaisesRegex(MODULE.AuditError, "invalid tenant identifier"):
                        client.authenticate()
                    request.assert_not_called()

    def test_required_403_is_incomplete(self) -> None:
        responses = fixture()
        responses["GET " + GRAPH + "subscribedSkus"] = [{"status": 403}]
        completed = self.run_cli(responses, "--sections", "licenses", "--json")
        self.assertEqual(completed.returncode, 2)
        report = json.loads(completed.stdout)
        self.assertEqual(report["status"], "incomplete")
        self.assertTrue(any("licenses" in item for item in report["skipped"]))
        allowed = self.run_cli(responses, "--sections", "licenses", "--json", "--allow-partial")
        self.assertEqual(allowed.returncode, 0)
        self.assertEqual(json.loads(allowed.stdout)["status"], "incomplete")

    def test_text_escaping(self) -> None:
        completed = self.run_cli(fixture(), "--sections", "roles")
        self.assertEqual(completed.returncode, 0, completed.stderr)
        self.assertIn("Alice\\u000aAdmin", completed.stdout)
        self.assertNotIn("Alice\nAdmin", completed.stdout)

    def test_read_throttle_retries(self) -> None:
        responses = fixture()
        key = "GET " + GRAPH + "directoryRoles?$select=id,displayName"
        responses[key].insert(0, {"status": 429, "retry_after": 0})
        completed = self.run_cli(responses, "--sections", "roles", "--json")
        self.assertEqual(completed.returncode, 0, completed.stderr)
        self.assertEqual(len(json.loads(completed.stdout)["sections"]["roles"]), 1)

    def test_token_post_is_not_retried(self) -> None:
        responses = fixture()
        responses["POST " + TOKEN] = [{"status": 429}, {"body": {"access_token": "unexpected"}}]
        completed = self.run_cli(responses, "--sections", "roles")
        self.assertEqual(completed.returncode, 2)
        self.assertIn("HTTP 429", completed.stderr)

    def test_missing_fixture_fails_closed(self) -> None:
        completed = self.run_cli({}, "--sections", "roles", fixture_exists=False)
        self.assertEqual(completed.returncode, 2)
        self.assertIn("invalid or missing fake response fixture", completed.stderr)

    def test_pagination_host_guard(self) -> None:
        responses = fixture()
        key = "GET " + GRAPH + "users?$select=id,displayName,userPrincipalName,accountEnabled,userType,createdDateTime,assignedLicenses"
        responses[key][0]["body"]["@odata.nextLink"] = "https://example.invalid/users"
        completed = self.run_cli(responses, "--sections", "users", "--json")
        self.assertEqual(completed.returncode, 2)
        self.assertEqual(json.loads(completed.stdout)["status"], "incomplete")

    def test_out_requires_matching_digest_and_is_create_only(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            output = str(Path(temp) / "report.txt")
            preview = self.run_cli(fixture(), "--sections", "roles", "--out", output, "--dry-run")
            self.assertEqual(preview.returncode, 0, preview.stderr)
            digest = preview.stdout.split("Approval digest: ")[-1].strip()
            denied = self.run_cli(fixture(), "--sections", "roles", "--out", output)
            self.assertEqual(denied.returncode, 2)
            self.assertFalse(Path(output).exists())
            created = self.run_cli(fixture(), "--sections", "roles", "--out", output, "--approve", digest)
            self.assertEqual(created.returncode, 0, created.stderr)
            self.assertIn("Global Administrator", Path(output).read_text(encoding="utf-8"))
            if os.name == "posix":
                self.assertEqual(Path(output).stat().st_mode & 0o777, 0o600)
            refused = self.run_cli(fixture(), "--sections", "roles", "--out", output, "--approve", digest)
            self.assertEqual(refused.returncode, 2)

    def test_swapped_symlink_and_directory_identity_refused(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp).resolve()
            first, second = root / "first", root / "second"
            first.mkdir(); second.mkdir()
            link = root / "selected"
            link.symlink_to(first, target_is_directory=True)
            plan = MODULE.output_plan(str(link / "report"))
            link.unlink(); link.symlink_to(second, target_is_directory=True)
            with self.assertRaises(MODULE.AuditError):
                MODULE.create_output(plan, "synthetic")
            self.assertFalse((first / "report").exists())
            self.assertFalse((second / "report").exists())
            plan = MODULE.output_plan(str(first / "report"))
            first.rename(root / "old"); first.mkdir()
            with self.assertRaises(MODULE.AuditError):
                MODULE.create_output(plan, "synthetic")
            self.assertFalse((first / "report").exists())

    def test_resolved_parent_swapped_to_symlink_is_clear_error(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp).resolve()
            parent, alternate = root / "parent", root / "alternate"
            parent.mkdir(); alternate.mkdir()
            plan = MODULE.output_plan(str(parent / "report"))
            parent.rename(root / "original")
            parent.symlink_to(alternate, target_is_directory=True)
            with self.assertRaisesRegex(MODULE.AuditError, "symlink.*new preview"):
                MODULE.create_output(plan, "synthetic")
            self.assertFalse((alternate / "report").exists())
            self.assertFalse((root / "original" / "report").exists())

    def test_nextlink_encoded_controls_and_segment_separators_refused(self) -> None:
        for encoded in ("%00", "%1f", "%7f", "%2f", "%5c", "%2500", "%252f", "%255c"):
            with self.subTest(encoded=encoded):
                responses = fixture()
                key = "GET " + GRAPH + "users?$select=id,displayName,userPrincipalName,accountEnabled,userType,createdDateTime,assignedLicenses"
                responses[key][0]["body"]["@odata.nextLink"] = GRAPH + "users/a" + encoded + "b"
                with patch.dict(os.environ, {MODULE.FAKE_FILE: "", MODULE.FAKE_ACK: ""}):
                    client = MODULE.GraphClient()
                client.fake = responses
                with self.assertRaisesRegex(MODULE.AuditError, "remote path"):
                    client.pages(key.removeprefix("GET " + GRAPH))

    def test_missing_output_ancestors_never_created(self) -> None:
        with tempfile.TemporaryDirectory() as temp:
            output = Path(temp) / "missing" / "nested" / "report"
            completed = self.run_cli(fixture(), "--out", str(output), "--dry-run")
            self.assertEqual(completed.returncode, 2)
            self.assertFalse(output.parent.parent.exists())

    def test_unsupported_platform_before_io(self) -> None:
        with patch.object(MODULE.os, "name", "nt"), patch.object(MODULE, "GraphClient") as client, patch.object(MODULE, "output_plan") as plan:
            self.assertEqual(MODULE.main(["report", "--out", "new-report"]), 2)
            client.assert_not_called(); plan.assert_not_called()

    def test_remote_traversal_refused_before_request(self) -> None:
        client = MODULE.GraphClient()
        for path in ("../users", "%2e%2e/users", "%252e%252e/users", "roles\\..\\users", "/users"):
            with self.subTest(path=path), patch.object(client.opener, "open") as opened:
                with self.assertRaises(MODULE.AuditError):
                    client.pages(path)
                opened.assert_not_called()
        for tenant in (".", ".."):
            with patch.dict(os.environ, {"GRAPH_TENANT_ID": tenant, "GRAPH_CLIENT_ID": "synthetic", "GRAPH_CLIENT_SECRET": "synthetic"}), patch.object(client, "request") as request:
                with self.assertRaises(MODULE.AuditError):
                    client.authenticate()
                request.assert_not_called()

    def test_input_bound_and_malformed_records(self) -> None:
        import io
        with patch.object(MODULE, "MAX_INPUT_BYTES", 8):
            with self.assertRaisesRegex(MODULE.AuditError, "limit"):
                MODULE.bounded_json(io.BytesIO(b" " * 9))
        with self.assertRaisesRegex(MODULE.AuditError, "invalid JSON"):
            MODULE.bounded_json(io.BytesIO(b"{"))
        for record in ({"assignedLicenses": [4]}, {"signInActivity": []}, {"displayName": {}}, {"prepaidUnits": "wrong"}):
            with self.subTest(record=record), self.assertRaises(MODULE.AuditError):
                MODULE.validate_record(record)
        with self.assertRaises(MODULE.AuditError):
            MODULE.days_until("bad-date", datetime.now(timezone.utc))
        responses = fixture()
        responses["GET " + GRAPH + "users?$select=id,displayName,userPrincipalName,signInActivity"] = [{"body": {"value": [{"signInActivity": {"lastSignInDateTime": "bad-date"}}]}}]
        completed = self.run_cli(responses, "--sections", "users", "--json")
        self.assertEqual(completed.returncode, 2)
        self.assertIn("invalid sign-in timestamp", completed.stdout)


    def test_fetched_role_identifier_is_validated(self) -> None:
        for identifier in ("..", "/users", "%2e%2e", "role\\name", None):
            responses = fixture()
            responses["GET " + GRAPH + "directoryRoles?$select=id,displayName"][0]["body"]["value"][0]["id"] = identifier
            completed = self.run_cli(responses, "--sections", "roles", "--json")
            self.assertEqual(completed.returncode, 2)
            self.assertIn("invalid role identifier", completed.stdout)

    def test_malformed_urls_and_response_status(self) -> None:
        with self.assertRaises(MODULE.AuditError):
            MODULE.guarded_url("https://[bad/users")
        responses = fixture()
        responses["POST " + TOKEN][0]["status"] = "not-numeric"
        completed = self.run_cli(responses, "--sections", "roles")
        self.assertEqual(completed.returncode, 2)
        self.assertIn("invalid response status", completed.stderr)


if __name__ == "__main__":
    unittest.main()

License: MIT (read it).

Back to the listing