npx skills add mukul975/Anthropic-Cybersecurity-SkillsMITRE ATT&CK
NIST CSF 2.0
MITRE D3FEND
Overview
Business Email Compromise (BEC) is a sophisticated fraud scheme where attackers impersonate executives, vendors, or trusted partners to trick employees into transferring funds, sharing sensitive data, or changing payment details. Unlike traditional phishing, BEC often contains no malicious links or attachments, relying purely on social engineering. This skill covers detection techniques using email gateway rules, behavioral analytics, and financial process controls.
When to Use
- When investigating security incidents that require detecting business email 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
- Email security gateway with BEC detection capabilities
- Understanding of organizational financial processes and approval chains
- Access to email logs and SIEM platform
- Knowledge of social engineering tactics
Key Concepts
BEC Attack Types (FBI IC3 Classification)
- CEO Fraud: Attacker impersonates CEO, requests urgent wire transfer
- Account Compromise: Employee email compromised, used to request payments from vendors
- False Invoice Scheme: Fake invoices from "vendor" with changed bank details
- Attorney Impersonation: Impersonates legal counsel for urgent confidential transfers
- Data Theft: Requests W-2, tax forms, or PII from HR
Detection Indicators
- Urgency and secrecy language ("confidential", "do not discuss with others")
- New or changed payment instructions
- Executive communication outside normal patterns
- Display name matches executive but email domain differs
- Reply-to address differs from From address
- First-time communication pattern between sender and recipient
- Request for gift cards or cryptocurrency
Workflow
Step 1: Configure BEC-Specific Email Rules
- Flag emails with VIP display names from external domains
- Detect financial keywords combined with urgency language
- Alert on first-time sender to finance/accounting staff
- Check for Reply-To domain mismatch
Step 2: Deploy Behavioral Analytics
- Baseline normal communication patterns per user
- Detect anomalous requests (unusual recipient, unusual time, unusual request type)
- Monitor for email forwarding rule changes (T1114.003)
Step 3: Implement Financial Controls
- Dual-authorization for wire transfers above threshold
- Out-of-band verification for payment detail changes (phone callback)
- Vendor payment change verification process
- Finance team training on BEC red flags
Step 4: Monitor for Account Compromise
- Detect impossible travel in email login locations
- Alert on email forwarding rule creation
- Monitor for mailbox delegation changes
- Check for inbox rules hiding BEC-related emails
Tools & Resources
- Microsoft Defender for O365 Anti-BEC: Built-in BEC detection
- Proofpoint Email Fraud Defense: BEC-specific solution
- Abnormal Security: AI-driven BEC detection
- FBI IC3 BEC Advisory: https://www.ic3.gov/
- FinCEN BEC Advisory: Financial institution guidance
Validation
- BEC detection rules trigger on test scenarios
- Financial controls prevent unauthorized transfers in drills
- Account compromise detection catches simulated attacks
- Reduced BEC susceptibility in awareness assessments
References and resources
Everything below is rendered for inspection. Script files are read-only and never run.
References 3
api-reference.md1.9 KB
API Reference: Detecting Business Email Compromise
Python email Library
import email
from email import policy
# Parse .eml file
with open("message.eml") as f:
msg = email.message_from_file(f, policy=policy.default)
msg.get("From") # sender header
msg.get("Reply-To") # reply-to header
msg.get("Authentication-Results") # SPF/DKIM/DMARC results
body = msg.get_body(preferencelist=("plain", "html"))
body.get_content() # decoded body textAuthentication Header Patterns
| Result | Meaning |
|---|---|
spf=pass |
Sender IP authorized by domain SPF record |
spf=fail |
Sender IP NOT in SPF record |
dkim=pass |
DKIM signature valid |
dkim=fail |
DKIM signature invalid or missing |
dmarc=pass |
SPF or DKIM aligned with From domain |
dmarc=fail |
Neither SPF nor DKIM aligned |
BEC Attack Types (FBI IC3)
| Type | Description |
|---|---|
| CEO Fraud | Impersonates executive requesting wire transfer |
| Invoice Fraud | Fake invoice with changed bank details |
| Account Compromise | Compromised email used for payment requests |
| Attorney Impersonation | Urgent legal matter requiring funds |
| Data Theft | Requests for W-2 / PII from HR |
BEC Indicator Regex Patterns
# Financial urgency
r"\b(wire transfer|bank transfer|routing number)\b"
# Secrecy pressure
r"\b(confidential|do not share|keep this between us)\b"
# Gift card fraud
r"\b(gift card|bitcoin|crypto|western union)\b"
# Account change
r"\b(change.*(bank|account|payment))\b"Microsoft Graph API - Mail Security
GET https://graph.microsoft.com/v1.0/me/messages?$filter=internetMessageHeaders/any(h: h/name eq 'Authentication-Results')
Authorization: Bearer {token}CLI Usage
python agent.py --email-file suspicious.eml --vip-names "John Smith" "Jane CEO"
python agent.py --scan-dir /var/mail/quarantine/ --vip-names "CFO Name"standards.md1.4 KB
Standards & References: Detecting Business Email Compromise
FBI IC3 BEC Classification
- Type 1: CEO Fraud / Executive Impersonation
- Type 2: Account Compromise (compromised employee email)
- Type 3: False Invoice / Vendor Email Compromise
- Type 4: Attorney Impersonation
- Type 5: Data Theft (W-2/PII requests)
MITRE ATT&CK References
- T1566.001/002: Phishing (initial access for BEC)
- T1534: Internal Spearphishing
- T1114.003: Email Collection: Email Forwarding Rule
- T1098.002: Account Manipulation: Additional Email Delegate Access
- T1586.002: Compromise Accounts: Email Accounts
NIST / Regulatory
- NIST SP 800-177 Rev.1: Trustworthy Email
- FinCEN Advisory FIN-2019-A005: Advisory on BEC targeting businesses
- FBI IC3 Annual Report: BEC statistics and trends ($2.9B losses in 2023)
Detection Rule Categories
| Rule | Description | Priority |
|---|---|---|
| VIP impersonation | External email with internal VIP display name | Critical |
| Payment language | Wire transfer/payment keywords + urgency | High |
| Reply-to mismatch | Reply-to domain differs from From | High |
| First-time sender | No prior communication with recipient | Medium |
| Forwarding rule | New auto-forward rule to external address | Critical |
| Gift card request | Request for gift card purchase | High |
| Vendor change | Payment detail change notification | High |
workflows.md2.3 KB
Workflows: Detecting Business Email Compromise
Workflow 1: BEC Detection Pipeline
Inbound email arrives
|
v
[Check: Does display name match VIP list?]
+-- YES + external domain --> HIGH ALERT: Possible CEO fraud
+-- NO --> Continue standard checks
|
v
[Check: Financial keywords present?]
+-- "wire transfer", "payment", "invoice", "bank details" detected
+-- Combined with urgency: "urgent", "confidential", "today"
+-- YES --> ELEVATED: Flag for finance team review
|
v
[Check: Reply-To mismatch?]
+-- Reply-To domain differs from From domain
+-- YES --> HIGH: Likely BEC attempt
|
v
[Check: Communication pattern anomaly?]
+-- First-time sender to finance/HR staff
+-- Unusual time of day for this sender
+-- YES --> MEDIUM: Requires verification
|
v
[Decision]
+-- BLOCK: High-confidence BEC
+-- QUARANTINE: Moderate confidence
+-- TAG: Warning banner for recipient
+-- DELIVER: Low riskWorkflow 2: BEC Incident Response
BEC attempt detected or reported
|
v
[Immediate actions (first 30 minutes)]
+-- Quarantine the email
+-- Search for similar messages to other recipients
+-- Alert affected users not to comply with request
|
v
[Investigation (next 2 hours)]
+-- Analyze email headers for true origin
+-- Check if any user already complied (check sent folders)
+-- If payment was made: Initiate bank recall immediately
+-- Search for compromised accounts (forwarding rules, login anomalies)
|
v
[Containment]
+-- Block sender domain/IP
+-- If account compromised: Force password reset, revoke sessions
+-- Remove malicious forwarding rules
+-- Notify finance to halt pending payments
|
v
[Recovery]
+-- Work with bank for fund recovery
+-- File FBI IC3 report (ic3.gov)
+-- Notify affected parties
+-- Update detection rules
+-- Targeted training for affected employeesWorkflow 3: Vendor Payment Change Verification
Vendor requests payment detail change
|
v
[Do NOT use contact info from the email]
|
v
[Look up vendor contact from existing records]
|
v
[Call vendor using known phone number]
+-- Verify the payment change request is legitimate
|
+-- CONFIRMED --> Process change with dual authorization
+-- NOT CONFIRMED --> Report as BEC attempt, block sender
+-- UNABLE TO REACH --> Hold payment, escalateScripts 2
agent.py5.5 KB
#!/usr/bin/env python3
"""BEC detection agent - analyzes email headers and content for Business Email Compromise indicators.
Parses email headers for spoofing signals, checks DMARC/SPF/DKIM alignment,
detects urgency language patterns, and flags financial request anomalies.
"""
import argparse
import email
import json
import re
from email import policy
from pathlib import Path
BEC_URGENCY_PATTERNS = [
r"\b(urgent|immediately|asap|right away|time.?sensitive)\b",
r"\b(confidential|do not share|keep this between us|don't tell)\b",
r"\b(wire transfer|bank transfer|payment|invoice|routing number)\b",
r"\b(gift card|bitcoin|crypto|western union|moneygram)\b",
r"\b(ceo|cfo|president|director) (asked|requested|needs|wants)\b",
r"\b(change.*(bank|account|payment)|new.*(bank|account|routing))\b",
r"\b(act now|deadline today|end of day|before close)\b",
]
EXECUTIVE_TITLES = ["ceo", "cfo", "coo", "cto", "president", "chairman",
"managing director", "vice president", "vp", "director"]
def parse_email_file(filepath):
with open(filepath, "r", encoding="utf-8", errors="replace") as f:
return email.message_from_file(f, policy=policy.default)
def check_spf_dkim_dmarc(msg):
results = {"spf": "none", "dkim": "none", "dmarc": "none"}
auth_results = msg.get("Authentication-Results", "")
if "spf=pass" in auth_results.lower():
results["spf"] = "pass"
elif "spf=fail" in auth_results.lower():
results["spf"] = "fail"
if "dkim=pass" in auth_results.lower():
results["dkim"] = "pass"
elif "dkim=fail" in auth_results.lower():
results["dkim"] = "fail"
if "dmarc=pass" in auth_results.lower():
results["dmarc"] = "pass"
elif "dmarc=fail" in auth_results.lower():
results["dmarc"] = "fail"
return results
def check_display_name_spoofing(msg, vip_names):
from_header = msg.get("From", "")
match = re.match(r'"?([^"<]+)"?\s*<([^>]+)>', from_header)
if not match:
return None
display_name = match.group(1).strip().lower()
email_addr = match.group(2).strip().lower()
for vip in vip_names:
if vip.lower() in display_name:
domain = email_addr.split("@")[-1] if "@" in email_addr else ""
return {"display_name": display_name, "email": email_addr,
"matched_vip": vip, "domain": domain,
"indicator": "Display name matches VIP but email may be external"}
return None
def check_reply_to_mismatch(msg):
from_addr = msg.get("From", "")
reply_to = msg.get("Reply-To", "")
if not reply_to:
return None
from_match = re.search(r'<([^>]+)>', from_addr) or re.search(r'(\S+@\S+)', from_addr)
reply_match = re.search(r'<([^>]+)>', reply_to) or re.search(r'(\S+@\S+)', reply_to)
if from_match and reply_match:
from_email = from_match.group(1).lower()
reply_email = reply_match.group(1).lower()
from_domain = from_email.split("@")[-1]
reply_domain = reply_email.split("@")[-1]
if from_domain != reply_domain:
return {"from": from_email, "reply_to": reply_email,
"indicator": "Reply-To domain differs from From domain"}
return None
def detect_urgency_language(body):
matches = []
for pattern in BEC_URGENCY_PATTERNS:
found = re.findall(pattern, body, re.IGNORECASE)
if found:
matches.extend(found)
return matches
def calculate_bec_score(auth, spoofing, reply_mismatch, urgency_matches):
score = 0
if auth.get("spf") == "fail":
score += 25
if auth.get("dkim") == "fail":
score += 20
if auth.get("dmarc") == "fail":
score += 30
if spoofing:
score += 35
if reply_mismatch:
score += 25
score += min(len(urgency_matches) * 10, 40)
return min(score, 100)
def analyze_email(filepath, vip_names):
msg = parse_email_file(filepath)
body = msg.get_body(preferencelist=("plain", "html"))
body_text = body.get_content() if body else ""
auth = check_spf_dkim_dmarc(msg)
spoofing = check_display_name_spoofing(msg, vip_names)
reply_mismatch = check_reply_to_mismatch(msg)
urgency = detect_urgency_language(body_text)
score = calculate_bec_score(auth, spoofing, reply_mismatch, urgency)
risk = "CRITICAL" if score >= 70 else "HIGH" if score >= 50 else "MEDIUM" if score >= 30 else "LOW"
return {
"file": str(filepath),
"from": msg.get("From", ""),
"to": msg.get("To", ""),
"subject": msg.get("Subject", ""),
"date": msg.get("Date", ""),
"authentication": auth,
"display_name_spoofing": spoofing,
"reply_to_mismatch": reply_mismatch,
"urgency_indicators": urgency,
"bec_score": score,
"risk_level": risk,
}
def main():
parser = argparse.ArgumentParser(description="BEC Email Analyzer")
parser.add_argument("--email-file", required=True, help="Path to .eml file")
parser.add_argument("--vip-names", nargs="+", default=[], help="VIP display names to check")
parser.add_argument("--scan-dir", help="Scan all .eml files in directory")
args = parser.parse_args()
results = []
if args.scan_dir:
for eml in Path(args.scan_dir).glob("*.eml"):
results.append(analyze_email(str(eml), args.vip_names))
else:
results.append(analyze_email(args.email_file, args.vip_names))
print(json.dumps(results, indent=2))
if __name__ == "__main__":
main()
process.py11.7 KB
#!/usr/bin/env python3
"""
Business Email Compromise (BEC) Detection Engine
Analyzes emails for BEC indicators including executive impersonation,
financial urgency, payment change requests, and communication anomalies.
Usage:
python process.py detect --email-json email.json
python process.py analyze-log --log-file email_log.json
python process.py vip-list --add "John CEO" --email "john@company.com"
"""
import argparse
import json
import re
import sys
from datetime import datetime, timezone
from dataclasses import dataclass, field, asdict
from collections import defaultdict
@dataclass
class BECIndicator:
"""A BEC detection indicator."""
category: str = ""
description: str = ""
severity: str = "medium"
confidence: float = 0.0
bec_type: str = ""
@dataclass
class BECAnalysis:
"""Complete BEC analysis result."""
from_address: str = ""
from_display_name: str = ""
to_address: str = ""
subject: str = ""
indicators: list = field(default_factory=list)
bec_score: float = 0.0
bec_type: str = "unknown"
is_bec: bool = False
recommended_action: str = ""
# Financial keywords
FINANCIAL_KEYWORDS = [
r'\bwire\s+transfer\b', r'\bbank\s+transfer\b', r'\bpayment\b',
r'\binvoice\b', r'\bpurchase\s+order\b', r'\baccount\s+number\b',
r'\brouting\s+number\b', r'\biban\b', r'\bswift\b', r'\back\b',
r'\bgift\s+card\b', r'\bbitcoin\b', r'\bcrypto\b', r'\bvenmo\b',
r'\bzelle\b', r'\bpaypal\b', r'\bw-2\b', r'\btax\s+form\b',
]
# Urgency keywords
URGENCY_KEYWORDS = [
r'\burgent\b', r'\bimmediately\b', r'\basap\b', r'\btoday\b',
r'\bright\s+now\b', r'\btime\s+sensitive\b', r'\bdo\s+not\s+(share|tell|discuss)\b',
r'\bconfidential\b', r'\bkeep\s+this\s+between\b', r'\bquietly\b',
r'\bbefore\s+end\s+of\s+day\b', r'\bcritical\b', r'\boverdue\b',
]
# Authority/impersonation keywords
AUTHORITY_KEYWORDS = [
r'\bi\s+need\s+you\s+to\b', r'\bplease\s+handle\b',
r'\bi\'m\s+in\s+a\s+meeting\b', r'\bi\'m\s+traveling\b',
r'\bdon\'t\s+call\s+me\b', r'\bemail\s+me\s+back\b',
r'\bcan\s+you\s+take\s+care\s+of\b', r'\bapproved\s+by\b',
]
def detect_bec(headers: dict, body: str = "", vip_list: list = None,
internal_domains: list = None) -> BECAnalysis:
"""Analyze email for BEC indicators."""
analysis = BECAnalysis()
analysis.from_address = headers.get("from", "")
analysis.from_display_name = headers.get("from_display_name", "")
analysis.to_address = headers.get("to", "")
analysis.subject = headers.get("subject", "")
from_domain = ""
match = re.search(r'@([\w.-]+)', analysis.from_address)
if match:
from_domain = match.group(1).lower()
if internal_domains is None:
internal_domains = []
score = 0.0
full_text = f"{analysis.subject} {body}".lower()
# Check 1: VIP display name impersonation
if vip_list and analysis.from_display_name:
name_lower = analysis.from_display_name.lower()
for vip in vip_list:
vip_name = vip.get("name", "").lower()
vip_domain = vip.get("domain", "").lower()
if vip_name and vip_name in name_lower:
if from_domain and vip_domain and from_domain != vip_domain:
analysis.indicators.append(BECIndicator(
category="vip_impersonation",
description=f"Display name '{analysis.from_display_name}' matches VIP "
f"'{vip.get('name')}' but email is from external domain '{from_domain}'",
severity="critical",
confidence=0.9,
bec_type="ceo_fraud"
))
score += 35
# Check 2: Financial keywords
financial_matches = []
for pattern in FINANCIAL_KEYWORDS:
if re.search(pattern, full_text, re.IGNORECASE):
financial_matches.append(pattern)
if financial_matches:
analysis.indicators.append(BECIndicator(
category="financial_language",
description=f"Found {len(financial_matches)} financial keyword(s)",
severity="medium",
confidence=min(len(financial_matches) * 0.2, 0.8),
bec_type="payment_fraud"
))
score += min(len(financial_matches) * 5, 20)
# Check 3: Urgency keywords
urgency_matches = []
for pattern in URGENCY_KEYWORDS:
if re.search(pattern, full_text, re.IGNORECASE):
urgency_matches.append(pattern)
if urgency_matches:
analysis.indicators.append(BECIndicator(
category="urgency_language",
description=f"Found {len(urgency_matches)} urgency/secrecy keyword(s)",
severity="medium",
confidence=min(len(urgency_matches) * 0.2, 0.8),
bec_type="social_engineering"
))
score += min(len(urgency_matches) * 5, 15)
# Check 4: Combined financial + urgency = higher risk
if financial_matches and urgency_matches:
analysis.indicators.append(BECIndicator(
category="combined_financial_urgency",
description="Financial request combined with urgency/secrecy language - strong BEC signal",
severity="high",
confidence=0.8,
bec_type="ceo_fraud"
))
score += 20
# Check 5: Authority language
authority_matches = []
for pattern in AUTHORITY_KEYWORDS:
if re.search(pattern, full_text, re.IGNORECASE):
authority_matches.append(pattern)
if authority_matches and (financial_matches or urgency_matches):
analysis.indicators.append(BECIndicator(
category="authority_language",
description="Authority/directive language combined with financial or urgency content",
severity="high",
confidence=0.7,
bec_type="ceo_fraud"
))
score += 15
# Check 6: Reply-to mismatch
reply_to = headers.get("reply_to", "")
if reply_to:
reply_domain = ""
match = re.search(r'@([\w.-]+)', reply_to)
if match:
reply_domain = match.group(1).lower()
if reply_domain and from_domain and reply_domain != from_domain:
analysis.indicators.append(BECIndicator(
category="reply_to_mismatch",
description=f"Reply-To ({reply_domain}) differs from From ({from_domain})",
severity="high",
confidence=0.85,
bec_type="account_compromise"
))
score += 20
# Check 7: External sender to finance/HR (if role info available)
to_role = headers.get("to_role", "").lower()
if from_domain and internal_domains and from_domain not in internal_domains:
if any(r in to_role for r in ["finance", "accounting", "payroll", "hr", "human resources"]):
analysis.indicators.append(BECIndicator(
category="external_to_finance",
description=f"External sender to {to_role} staff",
severity="medium",
confidence=0.5,
bec_type="vendor_fraud"
))
score += 10
# Calculate final verdict
analysis.bec_score = min(score, 100)
if analysis.bec_score >= 60:
analysis.is_bec = True
analysis.recommended_action = "BLOCK and alert SOC"
elif analysis.bec_score >= 40:
analysis.is_bec = True
analysis.recommended_action = "QUARANTINE for manual review"
elif analysis.bec_score >= 20:
analysis.recommended_action = "TAG with warning banner"
else:
analysis.recommended_action = "DELIVER normally"
# Determine most likely BEC type
type_scores = defaultdict(float)
for ind in analysis.indicators:
type_scores[ind.bec_type] += ind.confidence * 10
if type_scores:
analysis.bec_type = max(type_scores, key=type_scores.get)
return analysis
def format_bec_report(analysis: BECAnalysis) -> str:
"""Format BEC analysis as text report."""
lines = []
lines.append("=" * 60)
lines.append(" BUSINESS EMAIL COMPROMISE DETECTION REPORT")
lines.append("=" * 60)
lines.append(f" BEC Score: {analysis.bec_score:.0f}/100")
lines.append(f" Verdict: {'BEC DETECTED' if analysis.is_bec else 'NOT DETECTED'}")
lines.append(f" BEC Type: {analysis.bec_type}")
lines.append(f" Action: {analysis.recommended_action}")
lines.append("")
lines.append(f" From: {analysis.from_display_name} <{analysis.from_address}>")
lines.append(f" To: {analysis.to_address}")
lines.append(f" Subject: {analysis.subject}")
lines.append("")
if analysis.indicators:
lines.append(f"[INDICATORS] ({len(analysis.indicators)})")
for i, ind in enumerate(analysis.indicators, 1):
lines.append(f" {i}. [{ind.severity.upper()}] {ind.description}")
lines.append(f" Category: {ind.category} | Confidence: {ind.confidence:.0%}")
lines.append("=" * 60)
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="BEC Detection Engine")
subparsers = parser.add_subparsers(dest="command")
detect_parser = subparsers.add_parser("detect", help="Detect BEC in email")
detect_parser.add_argument("--email-json", help="Email JSON file")
detect_parser.add_argument("--from", dest="from_addr")
detect_parser.add_argument("--from-name", default="")
detect_parser.add_argument("--to", dest="to_addr", default="")
detect_parser.add_argument("--subject", default="")
detect_parser.add_argument("--body", default="")
detect_parser.add_argument("--vip-file", help="VIP list JSON file")
detect_parser.add_argument("--internal-domains", nargs="+", default=[])
log_parser = subparsers.add_parser("analyze-log", help="Analyze email log for BEC")
log_parser.add_argument("--log-file", required=True)
log_parser.add_argument("--vip-file")
log_parser.add_argument("--internal-domains", nargs="+", default=[])
parser.add_argument("--json", action="store_true")
args = parser.parse_args()
vip_list = []
vip_file = getattr(args, "vip_file", None)
if vip_file:
with open(vip_file) as f:
vip_list = json.load(f)
if args.command == "detect":
if args.email_json:
with open(args.email_json) as f:
email_data = json.load(f)
headers = email_data.get("headers", email_data)
body = email_data.get("body", "")
else:
headers = {
"from": args.from_addr or "",
"from_display_name": args.from_name,
"to": args.to_addr,
"subject": args.subject,
}
body = args.body
analysis = detect_bec(headers, body, vip_list,
getattr(args, "internal_domains", []))
if args.json:
print(json.dumps(asdict(analysis), indent=2, default=str))
else:
print(format_bec_report(analysis))
elif args.command == "analyze-log":
with open(args.log_file) as f:
log = json.load(f)
bec_count = 0
for entry in log:
analysis = detect_bec(
entry.get("headers", entry),
entry.get("body", ""),
vip_list,
getattr(args, "internal_domains", [])
)
if analysis.is_bec:
bec_count += 1
if args.json:
print(json.dumps(asdict(analysis), indent=2, default=str))
else:
print(format_bec_report(analysis))
print()
print(f"\nTotal analyzed: {len(log)}, BEC detected: {bec_count}")
else:
parser.print_help()
if __name__ == "__main__":
main()