Skip to content

Commit b850932

Browse files
author
Vadim
committed
Add Entra ID examples: risky users, access reviews, domains, auth methods
1 parent fd3b26a commit b850932

5 files changed

Lines changed: 368 additions & 0 deletions

File tree

‎examples/entraid/README.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,10 @@ directory roles, policies, and audit logs.
2222
| `Policy.Read.All` | Read tenant policies | [Policy permissions](https://learn.microsoft.com/en-us/graph/permissions-reference#policy-permissions) |
2323
| `IdentityProvider.Read.All` | Read identity providers (SAML, social) | [IdentityProvider permissions](https://learn.microsoft.com/en-us/graph/permissions-reference#identity-provider-permissions) |
2424
| `AuditLog.Read.All` | Read sign-in logs and audit logs | [AuditLog permissions](https://learn.microsoft.com/en-us/graph/permissions-reference#audit-log-permissions) |
25+
| `IdentityRiskyUser.Read.All` | Read risky users and risk detections | [Identity Protection permissions](https://learn.microsoft.com/en-us/graph/permissions-reference#identityprotection-permissions) |
26+
| `AccessReview.Read.All` | Read access reviews history | [Access review permissions](https://learn.microsoft.com/en-us/graph/permissions-reference#access-review-permissions) |
27+
| `Domain.ReadWrite.All` | Add and verify custom domains | [Domain permissions](https://learn.microsoft.com/en-us/graph/permissions-reference#domains-permissions) |
28+
| `UserAuthenticationMethod.Read.All` | Read registered MFA,FIDO2,passwordless methods | [Authentication methods permissions](https://learn.microsoft.com/en-us/graph/permissions-reference#authentication-methods-permissions) |
2529

2630
Admin consent is required for most permissions above.
2731

@@ -143,6 +147,10 @@ tenant-level policies, and audit logs.
143147
| Operation | File | Required role | API reference |
144148
|---|---|---|---|
145149
| List user sign-in logs | [`audit/list_signins.py`](./audit/list_signins.py) | `AuditLog.Read.All` | [list signins](https://learn.microsoft.com/en-us/graph/api/signin-list) |
150+
| **Risky users** — Identity Protection review, risk detections, dismiss / confirm-compromise | [`protection/risky_users.py`](./protection/risky_users.py) | `IdentityRiskyUser.Read.All`, `IdentityRiskDetection.Read.All` | [risky users](https://learn.microsoft.com/en-us/graph/api/resources/riskyuser) |
151+
| **Access reviews** — history definitions, instances, decisions, compliance reporting | [`governance/access_reviews.py`](./governance/access_reviews.py) | `AccessReview.Read.All` | [access reviews](https://learn.microsoft.com/en-us/graph/api/resources/accessreviewsv2-overview) |
152+
| **Domain management** — add, verify DNS, check service records | [`domains/manage.py`](./domains/manage.py) | `Domain.ReadWrite.All` | [domains API](https://learn.microsoft.com/en-us/graph/api/resources/domain) |
153+
| **Auth methods** — tenant MFA/passwordless readiness, per-user methods (FIDO2, phone, Authenticator) | [`users/auth_methods.py`](./users/auth_methods.py) | `UserAuthenticationMethod.Read.All`, `AuditLog.Read.All` | [auth methods overview](https://learn.microsoft.com/en-us/graph/api/resources/authenticationmethods-overview) |
146154

147155
---
148156

‎examples/entraid/domains/manage.py‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
"""
2+
Domain management — add, verify, check DNS records, and configure
3+
supported services.
4+
5+
Managing custom domains is one of the first steps when onboarding a
6+
new tenant. This example covers the full lifecycle:
7+
- Add a custom domain
8+
- Read the verification DNS records to paste into your registrar
9+
- Verify domain ownership
10+
- Show service configuration DNS records
11+
- Set supported services (Email, OfficeCommunicationsOnline, etc.)
12+
13+
Requires delegated permission ``Domain.ReadWrite.All``.
14+
15+
https://learn.microsoft.com/en-us/graph/api/resources/domain
16+
"""
17+
18+
import sys
19+
20+
from office365.graph_client import GraphClient
21+
from tests import test_client_id, test_client_secret, test_tenant
22+
23+
CUSTOM_DOMAIN = "contoso-marketing.com" # ← change to your domain
24+
25+
26+
def main():
27+
client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
28+
29+
# -- Step 1: list existing domains --
30+
domains = client.domains.get().execute_query()
31+
print("Domain(s):")
32+
for d in domains:
33+
state = "✔ verified" if d.state and d.state.status == "Verified" else "✗ unverified"
34+
svc = ", ".join(d.supported_services) if d.supported_services else "none"
35+
print(f" {d.id:30s} {state:14s} services: {svc}")
36+
print()
37+
38+
# -- Step 2: add a custom domain (if it doesn't exist already) --
39+
domain = next((d for d in domains if d.id == CUSTOM_DOMAIN), None)
40+
if domain is None:
41+
print(f"Adding domain '{CUSTOM_DOMAIN}'...")
42+
domain = client.domains.add(id=CUSTOM_DOMAIN).execute_query()
43+
print(f" Added: {domain.id}")
44+
else:
45+
print(f"Domain '{CUSTOM_DOMAIN}' already exists.")
46+
47+
# -- Step 3: show verification DNS records --
48+
# These need to be added to your DNS registrar's zone file.
49+
print(f"\nVerification DNS records for '{CUSTOM_DOMAIN}':")
50+
v_records = domain.verification_dns_records.get().execute_query()
51+
for r in v_records:
52+
rtype = r.properties.get("recordType", "?")
53+
label = r.properties.get("label", "?")
54+
value = r.properties.get("value", r.properties.get("text", "?"))
55+
ttl = r.properties.get("ttl", "?")
56+
print(f" {rtype:6s} {label:35s} TTL={ttl:>5} → {value}")
57+
58+
if not v_records:
59+
print(" (no verification records — domain may already be verified)")
60+
61+
# -- Step 4: verify the domain --
62+
# Do this only after you've added the DNS records at your registrar.
63+
response = input(f"\nAdd the DNS records above at your registrar, then type 'verify' to confirm: ")
64+
if response.strip().lower() == "verify":
65+
domain = domain.verify().execute_query()
66+
is_verified = domain.state and domain.state.status == "Verified"
67+
print(f" Domain verification: {'✔ SUCCESS' if is_verified else '✗ FAILED'}")
68+
else:
69+
print(" Skipped verification.")
70+
71+
# -- Step 5: show service configuration records --
72+
print(f"\nService configuration DNS records:")
73+
s_records = domain.service_configuration_records.get().execute_query()
74+
for r in s_records:
75+
rtype = r.properties.get("recordType", "?")
76+
label = r.properties.get("label", "?")
77+
value = r.properties.get("value", r.properties.get("text", "?"))
78+
ttl = r.properties.get("ttl", "?")
79+
print(f" {rtype:6s} {label:35s} TTL={ttl:>5} → {value}")
80+
81+
# -- Step 6: set supported services (optional) --
82+
# domain.set_property("supportedServices", ["Email", "OfficeCommunicationsOnline"])
83+
# domain.update().execute_query()
84+
# print(f"\nSupported services updated.")
85+
86+
87+
if __name__ == "__main__":
88+
main()

‎examples/entraid/governance/access_reviews.py‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
"""
2+
Access reviews — list review history definitions, instances, and
3+
decisions.
4+
5+
Access reviews are a core identity governance feature for certifying
6+
that users still need access to groups, apps, and roles.
7+
8+
This example focuses on what the SDK exposes via the
9+
``access_reviews`` navigation path:
10+
- List history definitions (one-time or recurring export jobs)
11+
- Show instances and their status
12+
- Show decisions per instance
13+
14+
For creating ad-hoc access review schedule definitions, use the Graph
15+
API endpoint ``POST /identityGovernance/accessReviews/definitions``.
16+
17+
Requires delegated permission ``AccessReview.Read.All`` or
18+
``AccessReview.ReadWrite.All``.
19+
20+
https://learn.microsoft.com/en-us/graph/api/resources/accessreviewsv2-overview
21+
"""
22+
23+
from office365.graph_client import GraphClient
24+
from tests import test_client_id, test_client_secret, test_tenant
25+
26+
27+
def main():
28+
client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
29+
30+
ar = client.identity_governance.access_reviews
31+
32+
# -- Step 1: access review history definitions (compliance exports) --
33+
hist_defs = ar.history_definitions.get().execute_query()
34+
print(f"History definitions: {len(hist_defs)}\n")
35+
36+
for hd in hist_defs:
37+
created_by = getattr(hd.created_by, "user_principal_name", "?") if hd.created_by else "?"
38+
print(f" {hd.display_name:35s} status={hd.status.name if hd.status else '?':12s} "
39+
f"created_by={created_by:30s} "
40+
f"period={str(hd.review_history_period_start_date_time)[:10] or '?'} → {str(hd.review_history_period_end_date_time)[:10] or '?'}")
41+
42+
# Enumerate instances
43+
instances = hd.instances.get().execute_query()
44+
for inst in instances:
45+
dt_str = inst.created_date_time.strftime("%Y-%m-%d %H:%M") if inst.created_date_time else "?"
46+
print(f" ↳ instance: created={dt_str} status={inst.status if hasattr(inst, 'status') else '?'}")
47+
48+
# -- Step 2: if no history definitions exist, show how to create one --
49+
if not hist_defs:
50+
print("(No history definitions found.)")
51+
print()
52+
print("To generate an access review history report:")
53+
print(""" from office365.directory.identitygovernance.accessreview.history.definition import AccessReviewHistoryDefinition
54+
from office365.directory.identitygovernance.accessreview.scope import AccessReviewScope
55+
from office365.directory.identitygovernance.accessreview.history.decisionfilter import AccessReviewHistoryDecisionFilter
56+
57+
definition = AccessReviewHistoryDefinition(client.context)
58+
definition.set_property("displayName", "Q1 Access Review History")
59+
definition.set_property("scopes", [AccessReviewScope()])
60+
definition.set_property("decisions", [AccessReviewHistoryDecisionFilter.approve])
61+
ar.history_definitions.add_child(definition)
62+
client.execute_query()
63+
""")
64+
65+
# -- Step 3: enumerate access reviews directly via the definitions endpoint --
66+
# The SDK doesn't expose definitions as a navigation property yet.
67+
# Use the Graph API directly via the SDK's underlying client:
68+
print("\nDirect API call to list schedule definitions:")
69+
try:
70+
response = client.execute_request_get(
71+
"/identityGovernance/accessReviews/definitions"
72+
"?$select=id,displayName,status,createdDateTime"
73+
"&$top=10"
74+
)
75+
definitions = response.json().get("value", [])
76+
print(f" Schedule definitions: {len(definitions)}")
77+
for d in definitions:
78+
print(f" {d.get('displayName','?'):40s} status={d.get('status','?'):15s} "
79+
f"created={str(d.get('createdDateTime',''))[:10]}")
80+
except Exception as e:
81+
print(f" (not available: {e})")
82+
83+
84+
if __name__ == "__main__":
85+
main()

‎examples/entraid/protection/risky_users.py‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
"""
2+
Risky users and risk detections — Identity Protection reporting and
3+
remediation.
4+
5+
Lists users flagged by Azure AD Identity Protection, reviews their
6+
risk history, shows raw risk detections with IP/location details,
7+
and demonstrates dismiss / confirm-compromise actions.
8+
9+
Security teams use this daily for incident response.
10+
11+
Requires delegated permission ``IdentityRiskyUser.Read.All``,
12+
``IdentityRiskDetection.Read.All`` to read, and
13+
``IdentityRiskyUser.ReadWrite.All`` to dismiss / confirm.
14+
15+
https://learn.microsoft.com/en-us/graph/api/resources/identityprotection-root
16+
"""
17+
18+
from datetime import datetime, timedelta, timezone
19+
20+
from office365.graph_client import GraphClient
21+
from tests import test_client_id, test_client_secret, test_tenant
22+
23+
24+
def main():
25+
client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
26+
27+
idp = client.identity_protection
28+
29+
# -- Step 1: list risky users with risk level and state --
30+
risky_users = idp.risky_users.get().execute_query()
31+
print(f"Risky users: {len(risky_users)}\n")
32+
33+
for u in risky_users:
34+
level = u.risk_level.name if u.risk_level else "?"
35+
state = u.risk_state.name if u.risk_state else "?"
36+
last = u.risk_last_updated_date_time
37+
last_str = last.strftime("%Y-%m-%d") if last else "?"
38+
print(f" {u.user_principal_name:35s} level={level:8s} state={state:12s} last={last_str}")
39+
# Details
40+
print(f" display={u.user_display_name} deleted={u.is_deleted} processing={u.is_processing}")
41+
42+
# -- Step 2: show recent risk history for the first risky user --
43+
if risky_users:
44+
print()
45+
user = risky_users[0]
46+
history = user.history.get().execute_query()
47+
print(f"Risk history for {user.user_principal_name} ({len(history)} events):")
48+
for h in history:
49+
dt = h.properties.get("activityDateTime", h.properties.get("detectedDateTime", ""))
50+
detail = h.properties.get("riskDetail", "?")
51+
event_type = h.properties.get("riskEventType", "?")
52+
print(f" {str(dt)[:19]} detail={detail:25s} type={event_type}")
53+
54+
# -- Step 3: list recent risk detections --
55+
since = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat()
56+
detections = idp.risk_detections.filter(f"detectedDateTime ge {since}").top(20).get().execute_query()
57+
print(f"\nRisk detections (last 7 days): {len(detections)}")
58+
59+
for d in detections:
60+
dt = d.detected_date_time.strftime("%m-%d %H:%M") if d.detected_date_time else "?"
61+
ip = d.ip_address or "?"
62+
loc = d.location
63+
city = loc.city if loc and hasattr(loc, "city") and loc.city else ""
64+
country = loc.country_or_region if loc and hasattr(loc, "country_or_region") and loc.country_or_region else ""
65+
location_str = f"{city}, {country}" if city or country else ""
66+
print(f" {dt} user={d.user_principal_name:30s} risk={d.risk_level.name:10s} "
67+
f"activity={d.activity.name or '?':12s} "
68+
f"ip={ip:15s} {location_str}")
69+
70+
# -- Step 4: dismiss and confirm-compromise (commented out by default) --
71+
72+
# Dismiss: marks user risk level as "none" (false positive)
73+
# idp.risky_users.dismiss(user_ids=["<user-id>"]).execute_query()
74+
75+
# Confirm compromised: marks as "high" (confirmed breach)
76+
# idp.risky_users.confirm_compromised(user_ids=["<user-id>"]).execute_query()
77+
78+
#
79+
# Example output:
80+
# To dismiss a user in the CLI:
81+
# client.identity_protection.risky_users.dismiss(
82+
# user_ids=["user-id-here"]
83+
# ).execute_query()
84+
85+
# -- Summary --
86+
print(f"\n--- Summary ---")
87+
print(f"Risky users: {len(risky_users)}")
88+
high_risk = [u for u in risky_users if u.risk_level and u.risk_level.name == "high"]
89+
print(f"High risk: {len(high_risk)}")
90+
print(f"Risk detections (7d): {len(detections)}")
91+
92+
93+
if __name__ == "__main__":
94+
main()

‎examples/entraid/users/auth_methods.py‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
"""
2+
Authentication methods — audit registered methods per user and check
3+
tenant-wide passwordless / MFA / SSPR readiness.
4+
5+
Shows three perspectives:
6+
1. Tenant-wide summary: users registered by method and by feature.
7+
2. Per-user detail: which methods each user has registered and what
8+
they're capable of (MFA, passwordless, SSPR).
9+
3. Per-user authentication methods: FIDO2 keys, Microsoft
10+
Authenticator, phone, password.
11+
12+
Useful for MFA roll-out tracking, passwordless adoption reporting,
13+
and security compliance audits.
14+
15+
Requires delegated permission ``UserAuthenticationMethod.Read.All``
16+
and ``AuditLog.Read.All``.
17+
18+
https://learn.microsoft.com/en-us/graph/api/resources/authenticationmethods-overview
19+
"""
20+
21+
from office365.graph_client import GraphClient
22+
from tests import test_client_id, test_client_secret, test_tenant
23+
24+
25+
def main():
26+
client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
27+
28+
am = client.reports.authentication_methods
29+
30+
# -- Step 1: tenant-wide summary by feature (MFA, SSPR, passwordless) --
31+
feature_summary = am.users_registered_by_feature().execute_query()
32+
if feature_summary and feature_summary.value:
33+
fs = feature_summary.value
34+
print("Users by authentication feature:")
35+
print(f" Total users : {fs.total_user_count}")
36+
print(f" MFA registered : {fs.mfa_registered_count or '?'}")
37+
print(f" SSPR capable : {fs.sspr_capable_count or '?'}")
38+
print(f" SSPR enabled : {fs.sspr_enabled_count or '?'}")
39+
print(f" Passwordless capable: {fs.passwordless_capable_count or '?'}")
40+
print()
41+
42+
# -- Step 2: tenant-wide summary by method --
43+
method_summary = am.users_registered_by_method().execute_query()
44+
if method_summary and method_summary.value:
45+
ms = method_summary.value
46+
print("Users by authentication method:")
47+
if hasattr(ms, "users_registered_by_method") and ms.users_registered_by_method:
48+
for m in ms.users_registered_by_method:
49+
print(f" {m.method_type or '?':30s} {m.user_count or 0} users")
50+
elif hasattr(ms, "registration_count") and ms.registration_count:
51+
for rc in ms.registration_count:
52+
print(f" {rc.method or '?':30s} {rc.count or 0} users")
53+
print()
54+
55+
# -- Step 3: per-user registration details --
56+
details = am.user_registration_details.top(20).get().execute_query()
57+
print(f"User registration details (showing {len(details)} users):\n")
58+
print(f"{'UPN':35s} {'MFA':6s} {'PW-less':9s} {'SSPR-cap':9s} {'SSPR-en':8s} {'Admin':6s} {'Preferred method'}")
59+
print("-" * 100)
60+
61+
for d in details:
62+
pref = d.user_preferred_method_for_secondary_authentication or "-"
63+
print(f"{d.user_principal_name[:33]:35s} "
64+
f"{'✔' if d.is_mfa_registered else '✗':6s} "
65+
f"{'✔' if d.is_passwordless_capable else '✗':9s} "
66+
f"{'✔' if d.is_sspr_capable else '✗':9s} "
67+
f"{'✔' if d.is_sspr_enabled else '✗':8s} "
68+
f"{'✔' if d.is_admin else '':6s} "
69+
f"{pref}")
70+
71+
# -- Step 4: per-user registered methods for a specific user --
72+
print()
73+
sample_upn = details[0].user_principal_name if details else "admin@contoso.onmicrosoft.com"
74+
users = client.users.filter(f"userPrincipalName eq '{sample_upn}'").get().execute_query()
75+
if users:
76+
user = users[0]
77+
auth = user.authentication
78+
all_methods = auth.methods.get().execute_query()
79+
print(f"Authentication methods for {user.user_principal_name}: {len(all_methods)}")
80+
for m in all_methods:
81+
mtype = m.properties.get("@odata.type", m.entity_type_name)
82+
extra = ""
83+
if "phone" in mtype.lower():
84+
extra = m.properties.get("phoneNumber", "") or m.properties.get("phoneType", "")
85+
elif "fido" in mtype.lower():
86+
extra = m.properties.get("displayName", "")
87+
elif "microsoftAuthenticator" in mtype.lower():
88+
extra = m.properties.get("displayName", "")
89+
print(f" {mtype:55s} {extra}")
90+
91+
92+
if __name__ == "__main__":
93+
main()

0 commit comments

Comments
 (0)