Security Skills Atlas
incident response

Detecting Email Account Compromise

Detect compromised O365 and Google Workspace email accounts by analyzing inbox rule creation, suspicious sign-in locations, mail forwarding rules, and unusual API access patterns via Microsoft Graph and audit logs.

account-takeoverbecemail-compromiseinbox-rulesmicrosoft-graphoffice365sign-in-analysis
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1070 on the official MITRE ATT&CK siteT1078 on the official MITRE ATT&CK siteT1486 on the official MITRE ATT&CK siteT1490 on the official MITRE ATT&CK siteT1566 on the official MITRE ATT&CK site

NIST CSF 2.0

RC.RP-01 on the official NIST CSF 2.0 siteRS.AN-03 on the official NIST CSF 2.0 siteRS.MA-01 on the official NIST CSF 2.0 siteRS.MA-02 on the official NIST CSF 2.0 site

Overview

Email account compromise (EAC) is a prevalent attack vector where adversaries gain unauthorized access to mailboxes to exfiltrate sensitive data, conduct business email compromise (BEC), or establish persistence through inbox rule manipulation. Attackers commonly create forwarding rules to siphon emails, delete rules to hide evidence, or use OAuth tokens for persistent access. Detection relies on analyzing Microsoft 365 Unified Audit Logs, Azure AD sign-in logs for impossible travel or suspicious locations, inbox rule creation events (Set-InboxRule, New-InboxRule), and Microsoft Graph API access patterns. Key indicators include forwarding rules to external addresses, rules that delete or move messages matching keywords like "invoice" or "payment", and sign-ins from unusual user agents such as python-requests.

When to Use

  • When investigating security incidents that require detecting email account compromise
  • When building detection rules or threat hunting queries for this domain
  • When SOC analysts need structured procedures for this analysis type
  • When validating security monitoring coverage for related attack techniques

Prerequisites

  • Microsoft 365 with Unified Audit Logging enabled
  • Azure AD P1/P2 for risk detection APIs
  • Python 3.9+ with requests, msal libraries
  • Microsoft Graph API application registration with Mail.Read, AuditLog.Read.All permissions
  • Understanding of OAuth2 client credential flows

Steps

  1. Export audit logs or connect to Microsoft Graph API using MSAL authentication
  2. Query inbox rules for all monitored mailboxes via /users/{id}/mailFolders/inbox/messageRules
  3. Analyze rules for external forwarding (ForwardTo, RedirectTo external addresses)
  4. Detect suspicious rule patterns: deletion rules, keyword-matching rules targeting financial terms
  5. Query sign-in logs via /auditLogs/signIns for unusual locations and impossible travel
  6. Check for suspicious user agent strings (python-requests, PowerShell, curl)
  7. Identify OAuth application consent grants for suspicious third-party apps
  8. Correlate findings across users to detect campaign-level compromise
  9. Generate compromise indicators report with severity scores

Expected Output

A JSON report listing compromised or suspicious accounts, malicious inbox rules detected, impossible travel events, suspicious OAuth grants, and recommended containment actions with severity ratings.

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.4 KB

Email Account Compromise Detection API Reference

Microsoft Graph API Endpoints

List Inbox Rules

GET https://graph.microsoft.com/v1.0/users/{userId}/mailFolders/inbox/messageRules
Authorization: Bearer {token}

Get Sign-In Logs

GET https://graph.microsoft.com/v1.0/auditLogs/signIns
  ?$filter=createdDateTime ge {startDate}
  &$top=100
Authorization: Bearer {token}

Risk Detections (Azure AD P2)

GET https://graph.microsoft.com/v1.0/identityProtection/riskDetections
  ?$filter=riskLevel eq 'high'
Authorization: Bearer {token}

OAuth2 Permission Grants

GET https://graph.microsoft.com/v1.0/oauth2PermissionGrants
Authorization: Bearer {token}

Authentication with MSAL

from msal import ConfidentialClientApplication
 
app = ConfidentialClientApplication(
    client_id="<app-id>",
    client_credential="<secret>",
    authority="https://login.microsoftonline.com/<tenant-id>"
)
token = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
headers = {"Authorization": f"Bearer {token['access_token']}"}

Inbox Rule Compromise Indicators

Indicator Field Description
External forwarding actions.forwardTo Rule forwards to external domain
External redirect actions.redirectTo Rule redirects to external address
Auto-delete actions.delete Rule auto-deletes matching messages
Financial keywords conditions.subjectContains Targets "invoice", "payment", "wire"

Sign-In Risk Indicators

Signal Detection Method
Impossible travel Haversine distance / time > 900 km/h
Suspicious UA python-requests, curl, PowerShell in userAgent
Unfamiliar location New country/region for user
Token replay Same token from different IPs

CLI Usage

python agent.py --input audit_data.json --output report.json

Required API Permissions

  • Mail.Read - Read inbox rules
  • AuditLog.Read.All - Read sign-in and audit logs
  • IdentityRiskEvent.Read.All - Read risk detections
  • Directory.Read.All - Read OAuth permission grants

References

Scripts 1

agent.py8.5 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Email Account Compromise Detection agent - analyzes inbox rules, sign-in logs, and OAuth grants to detect O365/Google Workspace account compromise"""

import argparse
import json
import math
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path

SUSPICIOUS_UA_PATTERNS = [
    "python-requests", "python-urllib", "curl", "wget", "powershell",
    "go-http-client", "httpie", "postman", "insomnia",
]

FINANCIAL_KEYWORDS = [
    "invoice", "payment", "wire", "transfer", "bank", "account",
    "payroll", "salary", "remittance", "ach", "swift",
]


def load_data(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))


def haversine_km(lat1, lon1, lat2, lon2):
    R = 6371.0
    rlat1, rlat2 = math.radians(lat1), math.radians(lat2)
    dlat = math.radians(lat2 - lat1)
    dlon = math.radians(lon2 - lon1)
    a = math.sin(dlat / 2) ** 2 + math.cos(rlat1) * math.cos(rlat2) * math.sin(dlon / 2) ** 2
    return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))


def analyze_inbox_rules(rules):
    """Detect malicious inbox rules: external forwarding, deletion rules, keyword-based filters."""
    findings = []
    for rule in rules:
        user = rule.get("user", rule.get("mailbox", ""))
        rule_name = rule.get("displayName", rule.get("name", ""))
        actions = rule.get("actions", {})
        forward_to = actions.get("forwardTo", []) or actions.get("forward_to", [])
        redirect_to = actions.get("redirectTo", []) or actions.get("redirect_to", [])
        delete_msg = actions.get("delete", False) or actions.get("moveToDeletedItems", False)
        move_to = actions.get("moveToFolder", "") or ""
        conditions = rule.get("conditions", {})
        subject_contains = conditions.get("subjectContains", []) or []
        body_contains = conditions.get("bodyContains", []) or []
        for dest in forward_to + redirect_to:
            addr = dest.get("emailAddress", {}).get("address", dest) if isinstance(dest, dict) else str(dest)
            domain = addr.split("@")[-1] if "@" in str(addr) else ""
            user_domain = user.split("@")[-1] if "@" in user else ""
            if domain and domain != user_domain:
                findings.append({
                    "type": "external_forwarding_rule",
                    "severity": "critical",
                    "resource": user,
                    "detail": f"Rule '{rule_name}' forwards to external address: {addr}",
                })
        if delete_msg:
            findings.append({
                "type": "deletion_rule",
                "severity": "high",
                "resource": user,
                "detail": f"Rule '{rule_name}' auto-deletes messages",
            })
        keyword_hits = [kw for kw in FINANCIAL_KEYWORDS
                        if any(kw in s.lower() for s in subject_contains + body_contains)]
        if keyword_hits:
            findings.append({
                "type": "financial_keyword_filter",
                "severity": "high",
                "resource": user,
                "detail": f"Rule '{rule_name}' targets financial keywords: {', '.join(keyword_hits)}",
            })
    return findings


def analyze_sign_ins(sign_ins):
    """Detect impossible travel, suspicious user agents, and risky sign-in patterns."""
    findings = []
    user_logins = defaultdict(list)
    for si in sign_ins:
        user = si.get("userPrincipalName", si.get("user", ""))
        ua = si.get("userAgent", si.get("user_agent", ""))
        ip = si.get("ipAddress", si.get("ip", ""))
        ts = si.get("createdDateTime", si.get("timestamp", ""))
        lat = si.get("location", {}).get("geoCoordinates", {}).get("latitude", 0)
        lon = si.get("location", {}).get("geoCoordinates", {}).get("longitude", 0)
        country = si.get("location", {}).get("countryOrRegion", "")
        risk = si.get("riskLevelAggregated", si.get("risk_level", "none"))
        for pattern in SUSPICIOUS_UA_PATTERNS:
            if pattern.lower() in (ua or "").lower():
                findings.append({
                    "type": "suspicious_user_agent",
                    "severity": "high",
                    "resource": user,
                    "detail": f"Sign-in from suspicious UA '{ua[:60]}' at IP {ip}",
                })
                break
        if risk in ("high", "medium"):
            findings.append({
                "type": "risky_sign_in",
                "severity": "high" if risk == "high" else "medium",
                "resource": user,
                "detail": f"Azure AD risk level '{risk}' from {country or ip}",
            })
        if lat and lon and ts:
            user_logins[user].append({"ts": ts, "lat": lat, "lon": lon, "ip": ip})
    for user, logins in user_logins.items():
        try:
            logins.sort(key=lambda x: x["ts"])
        except TypeError:
            continue
        for i in range(1, len(logins)):
            try:
                t1 = datetime.fromisoformat(logins[i - 1]["ts"].replace("Z", "+00:00"))
                t2 = datetime.fromisoformat(logins[i]["ts"].replace("Z", "+00:00"))
                hours = abs((t2 - t1).total_seconds()) / 3600.0
                if hours < 0.01:
                    continue
                dist = haversine_km(logins[i - 1]["lat"], logins[i - 1]["lon"], logins[i]["lat"], logins[i]["lon"])
                speed = dist / hours
                if speed > 900:
                    findings.append({
                        "type": "impossible_travel",
                        "severity": "critical",
                        "resource": user,
                        "detail": f"Impossible travel: {dist:.0f} km in {hours:.1f}h ({speed:.0f} km/h) between IPs {logins[i-1]['ip']} and {logins[i]['ip']}",
                    })
            except (ValueError, TypeError):
                continue
    return findings


def analyze_oauth_grants(grants):
    """Detect suspicious OAuth application consent grants."""
    findings = []
    for grant in grants:
        user = grant.get("user", grant.get("principalDisplayName", ""))
        app = grant.get("appDisplayName", grant.get("app_name", ""))
        scopes = grant.get("scope", grant.get("scopes", ""))
        consent_type = grant.get("consentType", "")
        risky_scopes = ["Mail.ReadWrite", "Mail.Send", "MailboxSettings.ReadWrite", "Files.ReadWrite.All"]
        granted_risky = [s for s in risky_scopes if s.lower() in (scopes or "").lower()]
        if granted_risky:
            findings.append({
                "type": "risky_oauth_grant",
                "severity": "high",
                "resource": user,
                "detail": f"App '{app}' granted risky scopes: {', '.join(granted_risky)}",
            })
        if consent_type == "AllPrincipals":
            findings.append({
                "type": "admin_consent_grant",
                "severity": "critical",
                "resource": user,
                "detail": f"App '{app}' has admin consent (AllPrincipals) with scopes: {scopes[:80]}",
            })
    return findings


def analyze(data):
    findings = []
    if isinstance(data, list):
        findings.extend(analyze_inbox_rules(data))
        return findings
    findings.extend(analyze_inbox_rules(data.get("inbox_rules", data.get("rules", []))))
    findings.extend(analyze_sign_ins(data.get("sign_ins", data.get("logins", []))))
    findings.extend(analyze_oauth_grants(data.get("oauth_grants", data.get("app_consents", []))))
    return findings


def generate_report(input_path):
    data = load_data(input_path)
    findings = analyze(data)
    sev = Counter(f["severity"] for f in findings)
    compromised = set(f["resource"] for f in findings if f["severity"] in ("critical", "high"))
    return {
        "report": "email_account_compromise_detection",
        "generated_at": datetime.utcnow().isoformat() + "Z",
        "total_findings": len(findings),
        "severity_summary": dict(sev),
        "potentially_compromised_accounts": sorted(compromised),
        "findings": findings,
    }


def main():
    ap = argparse.ArgumentParser(description="Email Account Compromise Detection Agent")
    ap.add_argument("--input", required=True, help="Input JSON with inbox rules/sign-in/OAuth data")
    ap.add_argument("--output", help="Output JSON report path")
    args = ap.parse_args()
    report = generate_report(args.input)
    out = json.dumps(report, indent=2)
    if args.output:
        Path(args.output).write_text(out, encoding="utf-8")
        print(f"Report written to {args.output}")
    else:
        print(out)


if __name__ == "__main__":
    main()
Keep exploring
Analyzing Linux Audit Logs for IntrusionUses the Linux Audit framework (auditd) with ausearch and aureport utilities to detect intrusion attempts, unauthorized access, privilege escalation, and suspicious system activity. Covers audit rule configuration, log querying, timeline reconstruction, and integration with SIEM platforms. Activates for requests involving auditd analysis, Linux audit log investigation, ausearch queries, aureport summaries, or host-based intrusion detection on Linux.Analyzing Network Traffic for IncidentsAnalyzes network traffic captures and flow data to identify adversary activity during security incidents, including command-and-control communications, lateral movement, data exfiltration, and exploitation attempts. Uses Wireshark, Zeek, and NetFlow analysis techniques. Activates for requests involving network traffic analysis, packet capture investigation, PCAP analysis, network forensics, C2 traffic detection, or exfiltration detection.Analyzing Security Logs with SplunkLeverages Splunk Enterprise Security and SPL (Search Processing Language) to investigate security incidents through log correlation, timeline reconstruction, and anomaly detection. Covers Windows event logs, firewall logs, proxy logs, and authentication data analysis. Activates for requests involving Splunk investigation, SPL queries, SIEM log analysis, security event correlation, or log-based incident investigation.Building Incident Response PlaybooksDesigns and documents structured incident response playbooks that define step-by-step procedures for specific incident types aligned with NIST SP 800-61r3 and SANS PICERL frameworks. Covers playbook structure, decision trees, escalation criteria, RACI matrices, and integration with SOAR platforms. Activates for requests involving IR playbook creation, incident response procedure documentation, response runbook development, or SOAR playbook design.Building Incident Timeline with TimesketchBuild collaborative forensic incident timelines using Timesketch to ingest, normalize, and analyze multi-source event data for attack chain reconstruction and investigation documentation.Building Malware Incident Communication TemplateBuild structured communication templates for malware incidents including stakeholder notifications, executive briefings, technical advisories, and regulatory disclosures with severity-based escalation procedures.