|
2 | 2 | Audit group membership changes — track who was added or removed from |
3 | 3 | Microsoft 365 and security groups. |
4 | 4 |
|
5 | | -Uses directory audit logs to detect Add member to group and Remove |
6 | | -member from group events. Useful for compliance monitoring and detecting |
7 | | -unauthorized group membership modifications. |
8 | | -
|
9 | | -Inspired by Report-GroupMembershipChanges.PS1 from the community. |
| 5 | +Uses directory audit logs to detect group membership modifications. |
| 6 | +Useful for compliance monitoring and detecting unauthorized changes. |
10 | 7 |
|
11 | 8 | Required delegated permissions: |
12 | 9 | AuditLog.Read.All Read directory audit logs |
|
21 | 18 | from office365.graph_client import GraphClient |
22 | 19 | from tests import test_client_id, test_client_secret, test_tenant |
23 | 20 |
|
| 21 | +TARGET_ACTIVITIES = { |
| 22 | + "Add member to group", |
| 23 | + "Remove member from group", |
| 24 | + "Add member to role", |
| 25 | + "Remove member from role", |
| 26 | +} |
24 | 27 |
|
25 | | -def query_group_membership_changes(days_back: int = 30) -> list[dict]: |
26 | | - """Query directory audit logs for group membership changes. |
27 | | -
|
28 | | - Args: |
29 | | - days_back: How many days to look back. |
30 | 28 |
|
31 | | - Returns: |
32 | | - List of change event dicts. |
33 | | - """ |
| 29 | +def main(): |
34 | 30 | client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret) |
35 | 31 |
|
36 | 32 | end = datetime.now(timezone.utc) |
37 | | - start = end - timedelta(days=days_back) |
| 33 | + start = end - timedelta(days=30) |
38 | 34 |
|
39 | | - changes = [] |
40 | | - |
41 | | - try: |
42 | | - audit_logs = ( |
43 | | - client.audit_logs.directory_audits.filter( |
44 | | - f"activityDateTime ge {start.strftime('%Y-%m-%dT%H:%M:%SZ')} and " |
45 | | - f"activityDateTime le {end.strftime('%Y-%m-%dT%H:%M:%SZ')}" |
46 | | - ) |
47 | | - .get() |
48 | | - .execute_query() |
| 35 | + audit_logs = ( |
| 36 | + client.audit_logs.directory_audits.filter( |
| 37 | + f"activityDateTime ge {start.strftime('%Y-%m-%dT%H:%M:%SZ')} and " |
| 38 | + f"activityDateTime le {end.strftime('%Y-%m-%dT%H:%M:%SZ')}" |
49 | 39 | ) |
| 40 | + .get() |
| 41 | + .execute_query() |
| 42 | + ) |
50 | 43 |
|
51 | | - target_activities = { |
52 | | - "Add member to group", |
53 | | - "Remove member from group", |
54 | | - "Add member to role", |
55 | | - "Remove member from role", |
56 | | - } |
57 | | - |
58 | | - for log in audit_logs: |
59 | | - activity = getattr(log, "activity_display_name", "") |
60 | | - if activity not in target_activities: |
61 | | - continue |
62 | | - |
63 | | - # Parse target resource (the group) |
64 | | - targets = getattr(log, "target_resources", []) |
65 | | - group_name = "Unknown" |
66 | | - member_name = "Unknown" |
67 | | - member_upn = "" |
68 | | - |
69 | | - for t in targets or []: |
70 | | - ttype = getattr(t, "type", "").lower() |
71 | | - if ttype == "group": |
72 | | - group_name = getattr(t, "display_name", getattr(t, "id", "Unknown")) |
73 | | - elif ttype == "user": |
74 | | - member_name = getattr(t, "display_name", "Unknown") |
75 | | - member_upn = getattr(t, "user_principal_name", "") |
76 | | - |
77 | | - changes.append( |
78 | | - { |
79 | | - "timestamp": getattr(log, "activity_date_time", None), |
80 | | - "activity": activity, |
81 | | - "group": group_name, |
82 | | - "member": member_name, |
83 | | - "member_upn": member_upn, |
84 | | - "initiated_by": getattr(log, "initiated_by", {}) |
85 | | - .get("user", {}) |
86 | | - .get("user_principal_name", "Unknown"), |
87 | | - } |
88 | | - ) |
89 | | - except Exception as e: |
90 | | - print(f" Warning: could not query audit logs: {e}") |
91 | | - |
92 | | - changes.sort(key=lambda x: x["timestamp"] or datetime.min, reverse=True) |
93 | | - return changes |
94 | | - |
95 | | - |
96 | | -def main(): |
97 | | - print("Group membership change audit (last 30 days)\n") |
98 | | - changes = query_group_membership_changes(days_back=30) |
| 44 | + changes = [] |
| 45 | + for log in audit_logs: |
| 46 | + if log.activity_display_name not in TARGET_ACTIVITIES: |
| 47 | + continue |
| 48 | + |
| 49 | + changes.append( |
| 50 | + { |
| 51 | + "timestamp": log.activity_datetime, |
| 52 | + "activity": log.activity_display_name, |
| 53 | + "group": str(next((t for t in log.target_resources if (t.type or "").lower() == "group"), "")), |
| 54 | + "member": str(next((t for t in log.target_resources if (t.type or "").lower() == "user"), "")), |
| 55 | + "initiated_by": str(log.initiated_by), |
| 56 | + } |
| 57 | + ) |
99 | 58 |
|
100 | 59 | if not changes: |
101 | | - print("No group membership changes found in directory audit logs.") |
| 60 | + print("No group membership changes found in the last 30 days.") |
102 | 61 | return |
103 | 62 |
|
104 | | - print(f"Found {len(changes)} changes:\n") |
| 63 | + changes.sort(key=lambda x: x["timestamp"] or datetime.min, reverse=True) |
| 64 | + |
| 65 | + print(f"Group membership changes (last 30 days): {len(changes)} found\n") |
105 | 66 | for c in changes: |
106 | | - ts = c["timestamp"].strftime("%Y-%m-%d %H:%M") if c["timestamp"] else "Unknown" |
107 | | - action = "➕" if "Add" in c["activity"] else "➖" |
108 | | - member = c["member_upn"] or c["member"] |
109 | | - print(f" {action} {c['activity']:28s} {member:40s} → {c['group']:30s} by {c['initiated_by']} ({ts})") |
| 67 | + ts = c["timestamp"].strftime("%Y-%m-%d %H:%M") if c["timestamp"] else "?" |
| 68 | + action = "[+]" if "Add" in c["activity"] else "[-]" |
| 69 | + print(f" {action} {c['activity']:28s} {c['member']:40s} -> {c['group']:30s} by {c['initiated_by']} ({ts})") |
110 | 70 |
|
111 | 71 |
|
112 | 72 | if __name__ == "__main__": |
|
0 commit comments