数字身份与访问管理安全取证深度分析

aws iam list-policies --scope Local --query 'Policies[].{Name:PolicyName,Arn:Arn}' --output table

aws iam simulate-principal-policy \
  --policy-source-arn arn:aws:iam::123456789012:role/EC2-Role \
  --action-names iam:CreateUser iam:AttachUserPolicy iam:CreateAccessKey \
  --output json | jq '.EvaluationResults[] | {Action:EvalActionName, Effect:EvalPolicyEffect}'

aws iam list-roles --query 'Roles[].{RoleName:RoleName,AssumeRolePolicyDocument:AssumeRolePolicyDocument}' --output json | \
python3 -c "
import json, sys
for role in json.load(sys.stdin):
    doc = role['AssumeRolePolicyDocument']
    for stmt in doc.get('Statement', []):
        if stmt.get('Effect') == 'Allow':
            principal = stmt.get('Principal', {})
            if '*' in str(principal):
                print(f'[CRITICAL] Wildcard trust: {role[\"RoleName\"]}')
            elif 'AWS' in principal:
                aws_prin = principal['AWS']
                if isinstance(aws_prin, str):
                    aws_prin = [aws_prin]
                for p in aws_prin:
                    if 'root' in p or '*' in p:
                        print(f'[HIGH] Overly broad trust: {role[\"RoleName\"]} -> {p}')
"

az rest --method GET \
  --url "https://graph.microsoft.com/v1.0/directoryRoleAssignments" \
  --query "value[?contains(roleId, 'e8611ab8-c189-46e8-94e1-60193e7e4f99') || contains(roleId, 'f28a1ee5-3d28-4e93-88f5-53b03e48f73e')].{Principal:principalId,Role:roleId}" \
  --output table

az rest --method GET \
  --url "https://graph.microsoft.com/v1.0/auditLogs/directories" \
  --query "value[?contains(category, 'RoleManagement')].{Time:activityDateTime,Operation:operationName,Actor:initiatedBy.user.userPrincipalName,Target:targetResources[0].displayName}" \
  --output table

gcloud asset search-all-iam-policies \
  --scope=organizations/123456789 \
  --query="policy:roles/owner OR policy:roles/editor" \
  --format="table(policy.bindings[].members, policy.bindings[].role, resource)" \
  --flatten="policy.bindings[]"

import json
import re
import sys
from collections import defaultdict
from datetime import datetime, timedelta

def analyze_oauth_redirect_logs(log_file):
    redirects = defaultdict(list)
    anomalies = []
    
    with open(log_file, 'r') as f:
        for line in f:
            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                continue
            
            if entry.get('event_type') not in ('oauth_redirect', 'authorization_code_grant'):
                continue
            
            client_id = entry.get('client_id', 'unknown')
            redirect_uri = entry.get('redirect_uri', 'unknown')
            user_agent = entry.get('user_agent', 'unknown')
            source_ip = entry.get('source_ip', 'unknown')
            timestamp = entry.get('timestamp', '')
            state_param = entry.get('state', '')
            session_id = entry.get('session_id', '')
            
            key = f"{client_id}:{session_id}"
            redirects[key].append({
                'timestamp': timestamp,
                'redirect_uri': redirect_uri,
                'source_ip': source_ip,
                'user_agent': user_agent,
                'state': state_param
            })
    
    for session_key, events in redirects.items():
        if len(events) < 2:
            continue
        
        uris = set(e['redirect_uri'] for e in events)
        ips = set(e['source_ip'] for e in events)
        agents = set(e['user_agent'] for e in events)
        states = set(e['state'] for e in events)
        
        if len(uris) > 1:
            anomalies.append({
                'type': 'REDIRECT_URI_VARIANCE',
                'severity': 'HIGH',
                'session': session_key,
                'detail': f'Multiple redirect URIs: {uris}',
                'events': events
            })
        
        if len(states) > 1 and len(ips) > 1:
            anomalies.append({
                'type': 'STATE_PARAM_REUSE',
                'severity': 'CRITICAL',
                'session': session_key,
                'detail': f'State reuse across IPs: {ips}',
                'events': events
            })
        
        if len(agents) > 1:
            anomalies.append({
                'type': 'USER_AGENT_VARIANCE',
                'severity': 'MEDIUM',
                'session': session_key,
                'detail': f'UA change within session: {agents}',
                'events': events
            })
        
        timestamps = []
        for e in events:
            try:
                ts = datetime.fromisoformat(e['timestamp'].replace('Z', '+00:00'))
                timestamps.append(ts)
            except (ValueError, AttributeError):
                pass
        
        if len(timestamps) >= 2:
            timestamps.sort()
            time_span = (timestamps[-1] - timestamps[0]).total_seconds()
            if time_span < 10 and len(ips) > 1:
                anomalies.append({
                    'type': 'RAPID_IP_CHANGE',
                    'severity': 'CRITICAL',
                    'session': session_key,
                    'detail': f'IP changed within {time_span:.1f}s: {ips}',
                    'events': events
                })
    
    return anomalies

def print_report(anomalies):
    severity_order = {'CRITICAL': 0, 'HIGH': 1, 'MEDIUM': 2, 'LOW': 3}
    anomalies.sort(key=lambda x: severity_order.get(x['severity'], 99))
    
    print(f'OAuth Anomaly Report - {datetime.now().isoformat()}')
    print(f'Total anomalies found: {len(anomalies)}')
    print('=' * 70)
    
    for i, anomaly in enumerate(anomalies, 1):
        print(f'\n[{anomaly["severity"]}] #{i} - {anomaly["type"]}')
        print(f'Session: {anomaly["session"]}')
        print(f'Detail: {anomaly["detail"]}')
        print(f'Related events: {len(anomaly["events"])}')
        print('-' * 50)

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f'Usage: {sys.argv[0]} <oauth_log_file>')
        sys.exit(1)
    
    results = analyze_oauth_redirect_logs(sys.argv[1])
    print_report(results)

title: OAuth Token Replay from Unusual Location
id: 5a2d3e4f-6b7c-8d9e-0f1a-2b3c4d5e6f7a
status: experimental
description: Detects potential OAuth token replay attacks based on geographic impossibility
references:
  - https://attack.mitre.org/techniques/T1550/001/
author: BlueTeam Forensics
date: 2026/07/18
tags:
  - attack.credential_access
  - attack.t1550.001
  - attack.t1078
logsource:
  category: authentication
  service: oauth
detection:
  selection_event:
    event_id:
      - '53000'
      - '53001'
      - '53002'
  selection_token:
    token_type:
      - 'access_token'
      - 'refresh_token'
  condition: selection_event and selection_token
  timeframe: 1h
  fields:
    - source_ip
    - user_agent
    - token_issuer
    - client_id
    - geographic_location
  filters:
    geo_impossible:
      field: geographic_distance_km
      operator: '>'
      value: 800
    time_impossible:
      field: time_between_requests_sec
      operator: '<'
      value: 300
falsepositives:
  - VPN reconnection
  - CDN edge node routing
level: high

jwt_tool() {
    local token="$1"
    echo "$token" | cut -d'.' -f2 | base64 -d 2>/dev/null | python3 -c "
import json, sys
try:
    payload = json.load(sys.stdin)
    print(f'Header: {json.dumps(header, indent=2)}')
    print(f'Payload: {json.dumps(payload, indent=2)}')
    alg = payload.get('header', {}).get('alg', payload.get('alg', 'unknown'))
    if alg in ('none', 'HS256'):
        print('[WARNING] Weak algorithm detected')
    if 'pkce' not in str(payload).lower():
        print('[INFO] No PKCE claim in token')
except Exception as e:
    print(f'Parse error: {e}')
"
}

grep -E "(sms_receive|sms_forward|ss7_route|location_update)" /var/log/sms_gateway.log | \
python3 -c "
import sys
from collections import Counter
numbers = []
for line in sys.stdin:
    parts = line.strip().split('|')
    if len(parts) >= 4:
        action = parts[2]
        number = parts[3]
        numbers.append((action, number))
        
action_counts = Counter(a for a, n in numbers)
for action, count in action_counts.most_common(20):
    print(f'{action}: {count}')
"

title: MFA Fatigue Attack Detection
id: 7c8d9e0f-1a2b-3c4d-5e6f-7a8b9c0d1e2f
status: stable
description: Detects potential MFA fatigue attacks through repeated push notification approvals
references:
  - https://attack.mitre.org/techniques/T1621/
author: BlueTeam Forensics
date: 2026/07/18
tags:
  - attack.initial_access
  - attack.t1621
  - attack.credential_access
logsource:
  category: authentication
  service: mfa
detection:
  selection_push:
    auth_method:
      - 'push_notification'
      - 'mfa_push'
      - 'totp'
  selection_failed:
    auth_result:
      - 'denied'
      - 'timeout'
  selection_success:
    auth_result:
      - 'approved'
      - 'success'
  condition: selection_push and (selection_failed R10 selection_success)
  timeframe: 10m
  count_threshold: 5
  fields:
    - user_id
    - source_ip
    - device_id
    - user_agent
    - notification_timestamp
    - approval_timestamp
  extra:
    time_diff:
      field: approval_timestamp
      minus: notification_timestamp
      threshold_seconds: 60
falsepositives:
  - User legitimately declined and re-approved
  - Network latency causing delayed notifications
level: critical

import json
import sys
from collections import defaultdict
from datetime import datetime, timedelta

class MFADetector:
    def __init__(self):
        self.events = defaultdict(list)
        self.alerts = []
    
    def load_events(self, log_file):
        with open(log_file, 'r') as f:
            for line in f:
                try:
                    entry = json.loads(line)
                except json.JSONDecodeError:
                    continue
                
                if entry.get('category') == 'mfa':
                    user = entry.get('user_id', 'unknown')
                    self.events[user].append({
                        'timestamp': entry.get('timestamp'),
                        'method': entry.get('auth_method'),
                        'result': entry.get('auth_result'),
                        'source_ip': entry.get('source_ip'),
                        'device_id': entry.get('device_id'),
                        'user_agent': entry.get('user_agent', '')
                    })
    
    def detect_fatigue_attack(self, window_minutes=10, threshold=5):
        for user, events in self.events.items():
            events.sort(key=lambda x: x.get('timestamp', ''))
            
            for i, event in enumerate(events):
                if event.get('result') not in ('approved', 'success'):
                    continue
                
                window_start = datetime.fromisoformat(
                    event['timestamp'].replace('Z', '+00:00')
                ) - timedelta(minutes=window_minutes)
                
                recent_denials = []
                for j in range(max(0, i - 20), i):
                    prev = events[j]
                    try:
                        prev_ts = datetime.fromisoformat(
                            prev['timestamp'].replace('Z', '+00:00')
                        )
                    except (ValueError, KeyError):
                        continue
                    
                    if prev_ts >= window_start and prev.get('result') in ('denied', 'timeout'):
                        recent_denials.append(prev)
                
                if len(recent_denials) >= threshold:
                    self.alerts.append({
                        'type': 'MFA_FATIGUE',
                        'severity': 'CRITICAL',
                        'user': user,
                        'denial_count': len(recent_denials),
                        'approval_event': event,
                        'window_start': window_start.isoformat(),
                        'mitre': 'T1621'
                    })
    
    def detect_impossible_travel(self, max_speed_kmh=900):
        for user, events in self.events.items():
            for i in range(1, len(events)):
                curr = events[i]
                prev = events[i - 1]
                
                try:
                    curr_ts = datetime.fromisoformat(
                        curr['timestamp'].replace('Z', '+00:00')
                    )
                    prev_ts = datetime.fromisoformat(
                        prev['timestamp'].replace('Z', '+00:00')
                    )
                except (ValueError, KeyError):
                    continue
                
                time_diff = (curr_ts - prev_ts).total_seconds()
                if time_diff <= 0 or time_diff > 86400:
                    continue
                
                if curr.get('source_ip') != prev.get('source_ip'):
                    self.alerts.append({
                        'type': 'MFA_GEO_ANOMALY',
                        'severity': 'HIGH',
                        'user': user,
                        'prev_ip': prev.get('source_ip'),
                        'curr_ip': curr.get('source_ip'),
                        'time_diff_seconds': time_diff,
                        'mitre': 'T1078'
                    })
    
    def detect_device_anomaly(self):
        user_devices = defaultdict(lambda: defaultdict(int))
        user_agents = defaultdict(lambda: defaultdict(int))
        
        for user, events in self.events.items():
            for event in events:
                if event.get('result') in ('approved', 'success'):
                    device = event.get('device_id', 'unknown')
                    ua = event.get('user_agent', 'unknown')
                    user_devices[user][device] += 1
                    user_agents[user][ua] += 1
        
        for user, devices in user_devices.items():
            if len(devices) > 3:
                self.alerts.append({
                    'type': 'MULTI_DEVICE_MFA',
                    'severity': 'MEDIUM',
                    'user': user,
                    'device_count': len(devices),
                    'devices': dict(devices),
                    'mitre': 'T1078'
                })
    
    def generate_report(self):
        severity_order = {'CRITICAL': 0, 'HIGH': 1, 'MEDIUM': 2, 'LOW': 3}
        self.alerts.sort(key=lambda x: severity_order.get(x['severity'], 99))
        
        print(f'MFA Security Analysis Report')
        print(f'Generated: {datetime.now().isoformat()}')
        print(f'Total users analyzed: {len(self.events)}')
        print(f'Total alerts: {len(self.alerts)}')
        print('=' * 70)
        
        for i, alert in enumerate(self.alerts, 1):
            print(f'\n[{alert["severity"]}] Alert #{i}: {alert["type"]}')
            print(f'  User: {alert["user"]}')
            print(f'  MITRE ATT&CK: {alert["mitre"]}')
            for k, v in alert.items():
                if k not in ('type', 'severity', 'user', 'mitre'):
                    print(f'  {k}: {v}')
            print('-' * 50)

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f'Usage: {sys.argv[0]} <mfa_log_file>')
        sys.exit(1)
    
    detector = MFADetector()
    detector.load_events(sys.argv[1])
    detector.detect_fatigue_attack()
    detector.detect_impossible_travel()
    detector.detect_device_anomaly()
    detector.generate_report()

#!/bin/bash
PAM_LOG="${1:-/var/log/pam-audit.log}"
REPORT_FILE="/tmp/pam_audit_report_$(date +%Y%m%d_%H%M%S).txt"

echo "PAM Security Audit Report" > "$REPORT_FILE"
echo "Generated: $(date)" >> "$REPORT_FILE"
echo "Log Source: $PAM_LOG" >> "$REPORT_FILE"
echo "========================================" >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "[1] Failed Authentication Attempts" >> "$REPORT_FILE"
grep -c "authentication_failure\|auth_fail\|login_failed" "$PAM_LOG" >> "$REPORT_FILE"
echo "" >> "$REPORT_FILE"
grep "authentication_failure\|auth_fail\|login_failed" "$PAM_LOG" | \
  awk '{print $1, $2, $3, $5}' | sort | uniq -c | sort -rn | head -20 >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "[2] Privilege Escalation Events" >> "$REPORT_FILE"
grep -i "sudo\|su_exec\|runas\|elevate" "$PAM_LOG" | \
  awk '{print $1, $2, $3, $5}' | sort | uniq -c | sort -rn | head -20 >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "[3] Credential Retrieval Operations" >> "$REPORT_FILE"
grep -i "retrieve\|checkout\|read_password\|get_credential" "$PAM_LOG" | \
  awk '{print $1, $2, $3, $5}' | sort | uniq -c | sort -rn | head -20 >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "[4] After-Hours Access (18:00-08:00)" >> "$REPORT_FILE"
grep -E "2[0-3]:|0[0-7]:" "$PAM_LOG" | grep -i "login\|access\|auth" | \
  awk '{print $1, $2, $3, $5}' | sort | uniq -c | sort -rn | head -20 >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "[5] Admin Account Usage" >> "$REPORT_FILE"
grep -iE "admin|root|superuser|privileged" "$PAM_LOG" | \
  awk '{print $1, $2, $3, $5}' | sort | uniq -c | sort -rn | head -20 >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "[6] Concurrent Session Anomalies" >> "$REPORT_FILE"
grep -i "session_start\|session_open" "$PAM_LOG" | \
  awk '{print $1, $2, $3, $5}' | sort | uniq -c | sort -rn | \
  awk '$1 > 3 {print "[ALERT] Multiple concurrent sessions:", $0}' >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "[7] Bulk Credential Access" >> "$REPORT_FILE"
grep -i "retrieve\|checkout\|read_password" "$PAM_LOG" | \
  awk '{print $5}' | sort | uniq -c | sort -rn | \
  awk '$1 > 10 {print "[ALERT] Excessive credential access by:", $2, "- Count:", $1}' >> "$REPORT_FILE"

echo "" >> "$REPORT_FILE"
echo "Report saved to: $REPORT_FILE"
cat "$REPORT_FILE"

title: JIT Privilege Escalation Abuse Detection
id: 3f4a5b6c-7d8e-9f0a-1b2c-3d4e5f6a7b8c
status: experimental
description: Detects potential abuse of Just-In-Time privilege elevation mechanisms
references:
  - https://attack.mitre.org/techniques/T1078/004/
author: BlueTeam Forensics
date: 2026/07/18
tags:
  - attack.privilege_escalation
  - attack.t1078.004
logsource:
  category: pam
  service: jit
detection:
  selection_jit:
    event_type:
      - 'jit_access_request'
      - 'jit_access_granted'
      - 'jit_elevation'
  selection_anomaly:
    request_outside_business_hours: true
    no_approval: true
    duration_exceeds_normal: true
  condition: selection_jit and selection_anomaly
  fields:
    - user_id
    - requested_role
    - approval_status
    - duration_minutes
    - request_timestamp
    - activation_timestamp
    - source_ip
    - justification
  timeframe: 24h
  aggregation:
    field: user_id
    function: count
    operator: '>'
    value: 3
falsepositives:
  - Emergency on-call access
  - Planned maintenance window
level: high

import base64
import json
import sys
import xml.etree.ElementTree as ET
from datetime import datetime
from urllib.parse import unquote

def decode_saml_response(encoded_response):
    decoded = base64.b64decode(unquote(encoded_response))
    return decoded.decode('utf-8', errors='replace')

def parse_saml_assertion(xml_content):
    findings = []
    
    try:
        root = ET.fromstring(xml_content)
    except ET.ParseError as e:
        findings.append({
            'type': 'PARSE_ERROR',
            'severity': 'HIGH',
            'detail': f'SAML XML parse failure: {str(e)}'
        })
        return findings
    
    ns = {
        'saml': 'urn:oasis:names:tc:SAML:2.0:assertion',
        'samlp': 'urn:oasis:names:tc:SAML:2.0:protocol',
        'ds': 'http://www.w3.org/2000/09/xmldsig#'
    }
    
    assertions = root.findall('.//saml:Assertion', ns)
    for assertion in assertions:
        issue_instant = assertion.get('IssueInstant', '')
        if issue_instant:
            try:
                issue_dt = datetime.fromisoformat(
                    issue_instant.replace('Z', '+00:00')
                )
                now = datetime.now(issue_dt.tzinfo)
                age = (now - issue_dt).total_seconds()
                if age > 300:
                    findings.append({
                        'type': 'STALE_ASSERTION',
                        'severity': 'MEDIUM',
                        'detail': f'Assertion age: {age:.0f}s (threshold: 300s)',
                        'issue_instant': issue_instant
                    })
            except ValueError:
                pass
        
        subject = assertion.find('.//saml:Subject', ns)
        if subject is not None:
            name_id = subject.find('.//saml:NameID', ns)
            if name_id is not None:
                name_value = name_id.text or ''
                if 'admin' in name_value.lower() or 'root' in name_value.lower():
                    findings.append({
                        'type': 'ADMIN_ASSERTION',
                        'severity': 'INFO',
                        'detail': f'Admin subject: {name_value}',
                        'subject': name_value
                    })
        
        conditions = assertion.find('.//saml:Conditions', ns)
        if conditions is not None:
            not_before = conditions.get('NotBefore', '')
            not_on_or_after = conditions.get('NotOnOrAfter', '')
            
            if not_before and not_on_or_after:
                try:
                    nb = datetime.fromisoformat(not_before.replace('Z', '+00:00'))
                    noa = datetime.fromisoformat(not_on_or_after.replace('Z', '+00:00'))
                    validity = (noa - nb).total_seconds()
                    if validity > 3600:
                        findings.append({
                            'type': 'OVERLY_LONG_VALIDITY',
                            'severity': 'MEDIUM',
                            'detail': f'Assertion validity window: {validity:.0f}s (recommended: < 300s)'
                        })
                except ValueError:
                    pass
        
        authn_statement = assertion.find('.//saml:AuthnStatement', ns)
        if authn_statement is not None:
            authn_context = authn_statement.find('.//saml:AuthnContext', ns)
            if authn_context is not None:
                authn_class = authn_context.find('.//saml:AuthnContextClassRef', ns)
                if authn_class is not None and authn_class.text:
                    ctx = authn_class.text
                    if 'unspecified' in ctx or 'none' in ctx.lower():
                        findings.append({
                            'type': 'WEAK_AUTHN_CONTEXT',
                            'severity': 'HIGH',
                            'detail': f'Weak authentication context: {ctx}'
                        })
        
        signature = assertion.find('.//ds:Signature', ns)
        if signature is None:
            findings.append({
                'type': 'UNSIGNED_ASSERTION',
                'severity': 'CRITICAL',
                'detail': 'Assertion is not signed'
            })
        
        attributes = assertion.find('.//saml:AttributeStatement', ns)
        if attributes is not None:
            for attr in attributes.findall('saml:Attribute', ns):
                attr_name = attr.get('Name', '')
                for val in attr.findall('saml:AttributeValue', ns):
                    value = val.text or ''
                    if 'role' in attr_name.lower() and 'admin' in value.lower():
                        findings.append({
                            'type': 'ADMIN_ROLE_ATTRIBUTE',
                            'severity': 'HIGH',
                            'detail': f'Admin role in attribute: {attr_name}={value}'
                        })
    
    return findings

def analyze_saml_request(request_data):
    findings = []
    
    if 'SAMLRequest' in request_data:
        encoded = request_data.split('SAMLRequest=')[1].split('&')[0]
        try:
            decoded = base64.b64decode(unquote(encoded))
            root = ET.fromstring(decoded)
            assertion_consumer_url = root.get('AssertionConsumerServiceURL', '')
            if assertion_consumer_url:
                findings.append({
                    'type': 'ACS_URL_INFO',
                    'severity': 'INFO',
                    'detail': f'ACS URL: {assertion_consumer_url}'
                })
        except Exception as e:
            findings.append({
                'type': 'REQUEST_PARSE_ERROR',
                'severity': 'MEDIUM',
                'detail': f'SAML Request parse issue: {str(e)}'
            })
    
    if 'SAMLResponse' in request_data:
        encoded = request_data.split('SAMLResponse=')[1].split('&')[0]
        xml_content = decode_saml_response(encoded)
        findings.extend(parse_saml_assertion(xml_content))
    
    return findings

def print_saml_report(findings):
    severity_order = {'CRITICAL': 0, 'HIGH': 1, 'MEDIUM': 2, 'LOW': 3, 'INFO': 4}
    findings.sort(key=lambda x: severity_order.get(x['severity'], 99))
    
    print('SAML Security Analysis Report')
    print('=' * 60)
    
    for i, f in enumerate(findings, 1):
        print(f'[{f["severity"]}] #{i} - {f["type"]}')
        print(f'  Detail: {f["detail"]}')
        print()

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f'Usage: {sys.argv[0]} <saml_post_data_file>')
        sys.exit(1)
    
    with open(sys.argv[1], 'r') as f:
        data = f.read().strip()
    
    if data.startswith('{'):
        request_data = json.loads(data)
    else:
        request_data = {}
        for pair in data.split('&'):
            if '=' in pair:
                key, val = pair.split('=', 1)
                request_data[key] = unquote(val)
    
    results = analyze_saml_request(request_data)
    print_saml_report(results)

title: Kerberos Golden Ticket Detection
id: 9a0b1c2d-3e4f-5a6b-7c8d-9e0f1a2b3c4d
status: stable
description: Detects potential Golden Ticket attacks through abnormal Kerberos TGT patterns
references:
  - https://attack.mitre.org/techniques/T1558/001/
author: BlueTeam Forensics
date: 2026/07/18
tags:
  - attack.credential_access
  - attack.t1558.001
logsource:
  category: authentication
  service: kerberos
detection:
  selection_tgt:
    event_id:
      - '4768'
      - '4769'
    ticket_type:
      - 'TGT'
      - 'TGS'
  selection_anomaly:
    encryption_type:
      - '0x17'
      - '0x18'
    pre_auth_result: '0x0'
    service_name: 'krbtgt'
  condition: selection_tgt and selection_anomaly
  fields:
    - account_name
    - source_ip
    - ticket_encryption_type
    - ticket_options
    - target_service
    - logon_id
  timeframe: 1h
  aggregation:
    field: account_name
    function: count
    operator: '>'
    value: 100
falsepositives:
  - Normal TGT renewal cycle
  - Large user base concurrent authentication
level: critical

az rest --method GET \
  --url "https://login.microsoftonline.com/{tenant_id}/federationSettings/DomainDns" \
  --query "value[?authenticationProtocol=='saml'].{Domain:name,Protocol:authenticationProtocol,SigningCert:signingCertificate}" \
  --output table

python3 << 'EOF'
import json
import sys
from datetime import datetime, timedelta

def correlate_identity_events(cloudtrail_log, ad_log, vpn_log, oauth_log):
    events = []
    
    with open(cloudtrail_log, 'r') as f:
        for line in f:
            try:
                entry = json.loads(line)
                events.append({
                    'source': 'CloudTrail',
                    'timestamp': entry.get('eventTime', ''),
                    'user': entry.get('userIdentity', {}).get('arn', ''),
                    'action': entry.get('eventName', ''),
                    'source_ip': entry.get('sourceIPAddress', ''),
                    'event_id': entry.get('eventID', ''),
                    'detail': json.dumps(entry.get('requestParameters', {}))
                })
            except json.JSONDecodeError:
                continue
    
    with open(ad_log, 'r') as f:
        for line in f:
            try:
                entry = json.loads(line)
                events.append({
                    'source': 'AD',
                    'timestamp': entry.get('timestamp', ''),
                    'user': entry.get('subjectUserName', ''),
                    'action': entry.get('eventID', ''),
                    'source_ip': entry.get('ipAddress', ''),
                    'event_id': str(entry.get('eventRecordId', '')),
                    'detail': entry.get('description', '')
                })
            except json.JSONDecodeError:
                continue
    
    with open(vpn_log, 'r') as f:
        for line in f:
            try:
                entry = json.loads(line)
                events.append({
                    'source': 'VPN',
                    'timestamp': entry.get('timestamp', ''),
                    'user': entry.get('username', ''),
                    'action': entry.get('event_type', ''),
                    'source_ip': entry.get('public_ip', ''),
                    'event_id': entry.get('session_id', ''),
                    'detail': entry.get('message', '')
                })
            except json.JSONDecodeError:
                continue
    
    events.sort(key=lambda x: x.get('timestamp', ''))
    
    correlations = []
    for i, event in enumerate(events):
        if not event.get('source_ip'):
            continue
        
        window_start = datetime.fromisoformat(
            event['timestamp'].replace('Z', '+00:00')
        ) - timedelta(minutes=5)
        window_end = datetime.fromisoformat(
            event['timestamp'].replace('Z', '+00:00')
        ) + timedelta(minutes=5)
        
        related = []
        for j, other in enumerate(events):
            if i == j:
                continue
            try:
                other_ts = datetime.fromisoformat(
                    other['timestamp'].replace('Z', '+00:00')
                )
                if window_start <= other_ts <= window_end:
                    if (other['user'] == event['user'] or 
                        other['source_ip'] == event['source_ip']):
                        related.append({
                            'source': other['source'],
                            'timestamp': other['timestamp'],
                            'action': other['action'],
                            'match_reason': 'user' if other['user'] == event['user'] else 'ip'
                        })
            except (ValueError, KeyError):
                continue
        
        if len(related) >= 2:
            sources = set(r['source'] for r in related) | {event['source']}
            if len(sources) >= 2:
                correlations.append({
                    'primary_event': event,
                    'correlated_events': related,
                    'correlation_strength': len(sources),
                    'cross_source': list(sources)
                })
    
    return correlations

def print_correlation_report(correlations):
    print(f'Identity Event Correlation Report')
    print(f'Total correlated chains: {len(correlations)}')
    print('=' * 70)
    
    for i, corr in enumerate(correlations, 1):
        primary = corr['primary_event']
        print(f'\nCorrelation Chain #{i} (Strength: {corr["correlation_strength"]})')
        print(f'Sources: {", ".join(corr["cross_source"])}')
        print(f'Primary: [{primary["source"]}] {primary["action"]} by {primary["user"]}')
        print(f'  Time: {primary["timestamp"]}')
        print(f'  IP: {primary["source_ip"]}')
        print(f'  Related events:')
        for evt in corr['correlated_events']:
            print(f'    [{evt["source"]}] {evt["action"]} ({evt["match_reason"]} match)')
            print(f'      Time: {evt["timestamp"]}')
        print('-' * 50)

if __name__ == '__main__':
    if len(sys.argv) < 5:
        print(f'Usage: {sys.argv[0]} <cloudtrail> <ad_log> <vpn_log> <oauth_log>')
        sys.exit(1)
    
    results = correlate_identity_events(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4])
    print_correlation_report(results)
EOF

title: Cloud IAM Policy Tampering Detection
id: b1c2d3e4-f5a6-b7c8-d9e0-f1a2b3c4d5e6
status: stable
description: Detects potential IAM policy tampering activities across cloud platforms
references:
  - https://attack.mitre.org/techniques/T1098/003/
  - https://attack.mitre.org/techniques/T1098/001/
author: BlueTeam Forensics
date: 2026/07/18
tags:
  - attack.persistence
  - attack.t1098.003
  - attack.t1098.001
  - attack.privilege_escalation
logsource:
  category: cloud_trail
  service: iam
detection:
  selection_aws:
    event_source: 'iam.amazonaws.com'
    event_name:
      - 'CreatePolicyVersion'
      - 'SetDefaultPolicyVersion'
      - 'AttachUserPolicy'
      - 'AttachRolePolicy'
      - 'AttachGroupPolicy'
      - 'PutUserPolicy'
      - 'PutRolePolicy'
      - 'PutGroupPolicy'
      - 'CreateAccessKey'
      - 'UpdateAssumeRolePolicy'
  selection_azure:
    category: 'RoleManagement'
    operationName:
      - 'Add member to role'
      - 'Add eligible member to role'
      - 'Update role assignment'
  selection_gcp:
    methodName:
      - 'SetIamPolicy'
      - 'TestIamPermissions'
      - 'AddAllowedPolicyBinding'
  condition: selection_aws or selection_azure or selection_gcp
  fields:
    - event_time
    - user_identity_arn
    - event_name
    - request_parameters
    - source_ip_address
    - user_agent
  timeframe: 1h
  filters:
    privileged_actions:
      event_name:
        - 'CreatePolicyVersion'
        - 'AttachUserPolicy'
        - 'AttachRolePolicy'
        - 'UpdateAssumeRolePolicy'
    privileged_roles:
      role_name:
        - 'AdministratorAccess'
        - 'PowerUserAccess'
        - 'OrganizationAccountAccessRole'
falsepositives:
  - Legitimate IAM policy updates during maintenance
  - Infrastructure-as-Code deployments
  - Automated CI/CD pipeline role management
level: critical

title: Abnormal Authentication Pattern Detection
id: c2d3e4f5-a6b7-c8d9-e0f1-a2b3c4d5e6f7
status: experimental
description: Detects abnormal authentication patterns indicative of credential compromise
references:
  - https://attack.mitre.org/techniques/T1078/
  - https://attack.mitre.org/techniques/T1110/
author: BlueTeam Forensics
date: 2026/07/18
tags:
  - attack.credential_access
  - attack.t1078
  - attack.t1110
logsource:
  category: authentication
  service: all
detection:
  selection_brute_force:
    auth_result:
      - 'failure'
      - 'failed'
    event_id:
      - '4625'
      - '4771'
      - '4776'
  selection_password_spray:
    auth_result:
      - 'failure'
      - 'failed'
  selection_impossible_travel:
    auth_result:
      - 'success'
      - 'approved'
    fields_required:
      - source_ip
      - geographic_location
      - user_agent
  condition: selection_brute_force | count by source_ip > 10 within 5m
    or selection_password_spray | count(distinct account_name) by source_ip > 5 within 15m
  timeframe: 15m
  fields:
    - account_name
    - source_ip
    - auth_method
    - auth_result
    - user_agent
    - geographic_location
    - failure_reason
falsepositives:
  - Service account automation
  - Credential rotation scripts
  - Legitimate pentesting activities
level: high

#!/bin/bash
echo "============================================"
echo "  Multi-Cloud IAM Security Audit Script"
echo "  Generated: $(date)"
echo "============================================"

check_aws_iam() {
    echo ""
    echo "[AWS IAM AUDIT]"
    echo "-------------------------"
    
    echo "[1] IAM Users without MFA:"
    aws iam list-users --query 'Users[?MFADevices==`[]`].{UserName:UserName,CreateDate:CreateDate}' --output table 2>/dev/null
    
    echo ""
    echo "[2] IAM Users with Console Access:"
    aws iam generate-credential-report 2>/dev/null
    sleep 2
    aws iam get-credential-report --query 'Content' --output text 2>/dev/null | \
      base64 -d 2>/dev/null | \
      awk -F',' 'NR==1 || ($4=="true" && $8=="true") {print $1, $4, $8, $9}'
    
    echo ""
    echo "[3] Overly Permissive Policies:"
    for policy_arn in $(aws iam list-policies --scope Local --query 'Policies[].[Arn]' --output text 2>/dev/null); do
        default_ver=$(aws iam get-policy --policy-arn "$policy_arn" --query 'Policy.DefaultVersionId' --output text 2>/dev/null)
        has_wildcard=$(aws iam get-policy-version --policy-arn "$policy_arn" --version-id "$default_ver" --query 'PolicyVersion.Document.Statement[?Action==`*`]' --output text 2>/dev/null)
        if [ -n "$has_wildcard" ]; then
            echo "[CRITICAL] Wildcard policy: $policy_arn"
        fi
    done
    
    echo ""
    echo "[4] IAM Roles with Cross-Account Trust:"
    aws iam list-roles --query 'Roles[].{RoleName:RoleName,Arn:Arn}' --output json 2>/dev/null | \
      python3 -c "
import json, sys
roles = json.load(sys.stdin)
for role in roles:
    arn = role['Arn']
    trust = aws iam get-role --role-name $(echo $arn | cut -d'/' -f2) 2>/dev/null
" 2>/dev/null || true
}

check_azure_ad() {
    echo ""
    echo "[Azure AD AUDIT]"
    echo "-------------------------"
    
    if command -v az &> /dev/null; then
        echo "[1] Global Admins:"
        az ad directory-role list --query "[?displayName=='Global Administrator'].assignments[].{User:principalDisplayName}" --output table 2>/dev/null
        
        echo ""
        echo "[2] Users with No MFA:"
        az rest --method GET \
          --url "https://graph.microsoft.com/v1.0/users?\$select=displayName,userPrincipalName,authenticationMethods" \
          --query 'value[].{Name:displayName,UPN:userPrincipalName}' \
          --output table 2>/dev/null
    else
        echo "[SKIP] Azure CLI not installed"
    fi
}

check_gcp_iam() {
    echo ""
    echo "[GCP IAM AUDIT]"
    echo "-------------------------"
    
    if command -v gcloud &> /dev/null; then
        echo "[1] Project IAM Bindings:"
        gcloud asset search-all-iam-policies \
          --scope=organizations/$(gcloud organizations list --format="value(name)" 2>/dev/null | head -1 | cut -d'/' -f2) \
          --query="policy.bindings.role:" \
          --flatten="policy.bindings[]" \
          --format="table(policy.bindings.role, policy.bindings.members, resource)" 2>/dev/null | \
          grep -i "roles/owner\|roles/editor\|roles/iam.securityAdmin"
        
        echo ""
        echo "[2] Service Accounts with Keys:"
        gcloud iam service-accounts list --format="table(email,disabled)" 2>/dev/null | \
          while read -r sa_email _; do
            key_count=$(gcloud iam service-accounts keys list --iam-account="$sa_email" --format="value(name)" 2>/dev/null | wc -l)
            if [ "$key_count" -gt 0 ]; then
                echo "[WARNING] SA $sa_email has $key_count keys"
            fi
        done
    else
        echo "[SKIP] gcloud CLI not installed"
    fi
}

check_aws_iam
check_azure_ad
check_gcp_iam

echo ""
echo "============================================"
echo "  Audit Complete"
echo "============================================"

import json
import sys
from collections import defaultdict
from datetime import datetime, timedelta
from statistics import mean, stdev

class IdentityBehaviorBaseline:
    def __init__(self):
        self.user_events = defaultdict(list)
        self.baselines = {}
        self.anomalies = []
    
    def load_cloudtrail(self, log_file):
        with open(log_file, 'r') as f:
            for line in f:
                try:
                    event = json.loads(line)
                except json.JSONDecodeError:
                    continue
                
                user = event.get('userIdentity', {}).get('arn', 'unknown')
                ts = event.get('eventTime', '')
                action = event.get('eventName', '')
                source_ip = event.get('sourceIPAddress', '')
                ua = event.get('userAgent', '')
                
                try:
                    dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
                except (ValueError, AttributeError):
                    continue
                
                self.user_events[user].append({
                    'timestamp': dt,
                    'action': action,
                    'source_ip': source_ip,
                    'user_agent': ua,
                    'hour': dt.hour,
                    'weekday': dt.weekday()
                })
    
    def build_baselines(self):
        for user, events in self.user_events.items():
            if len(events) < 10:
                continue
            
            hours = [e['hour'] for e in events]
            action_types = defaultdict(int)
            ip_set = defaultdict(int)
            ua_set = defaultdict(int)
            
            for e in events:
                action_types[e['action']] += 1
                ip_set[e['source_ip']] += 1
                ua_set[e['user_agent']] += 1
            
            most_common_actions = sorted(action_types.items(), key=lambda x: x[1], reverse=True)[:10]
            most_common_ips = sorted(ip_set.items(), key=lambda x: x[1], reverse=True)[:5]
            
            self.baselines[user] = {
                'total_events': len(events),
                'mean_hour': mean(hours) if hours else 0,
                'hour_std': stdev(hours) if len(hours) > 1 else 0,
                'common_actions': set(a for a, _ in most_common_actions),
                'common_ips': set(ip for ip, _ in most_common_ips),
                'unique_actions': len(action_types),
                'unique_ips': len(ip_set),
                'activity_days': len(set(e['timestamp'].date() for e in events)),
                'events_per_day': len(events) / max(1, len(set(e['timestamp'].date() for e in events)))
            }
    
    def detect_anomalies(self, current_events_file):
        self.build_baselines()
        
        with open(current_events_file, 'r') as f:
            for line in f:
                try:
                    event = json.loads(line)
                except json.JSONDecodeError:
                    continue
                
                user = event.get('userIdentity', {}).get('arn', 'unknown')
                action = event.get('eventName', '')
                source_ip = event.get('sourceIPAddress', '')
                ts = event.get('eventTime', '')
                
                if user not in self.baselines:
                    continue
                
                baseline = self.baselines[user]
                
                try:
                    dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
                    hour = dt.hour
                except (ValueError, AttributeError):
                    continue
                
                if action not in baseline['common_actions']:
                    self.anomalies.append({
                        'type': 'NOVEL_ACTION',
                        'severity': 'HIGH',
                        'user': user,
                        'action': action,
                        'source_ip': source_ip,
                        'timestamp': ts,
                        'mitre': 'T1078'
                    })
                
                if source_ip not in baseline['common_ips']:
                    self.anomalies.append({
                        'type': 'NOVEL_IP',
                        'severity': 'MEDIUM',
                        'user': user,
                        'source_ip': source_ip,
                        'timestamp': ts,
                        'mitre': 'T1078'
                    })
                
                if abs(hour - baseline['mean_hour']) > 4:
                    self.anomalies.append({
                        'type': 'OFF_HOURS_ACTIVITY',
                        'severity': 'MEDIUM',
                        'user': user,
                        'action': action,
                        'hour': hour,
                        'mean_hour': baseline['mean_hour'],
                        'timestamp': ts,
                        'mitre': 'T1078'
                    })
    
    def generate_hunting_report(self):
        severity_order = {'CRITICAL': 0, 'HIGH': 1, 'MEDIUM': 2, 'LOW': 3}
        self.anomalies.sort(key=lambda x: severity_order.get(x['severity'], 99))
        
        print(f'Identity Behavior Anomaly Report')
        print(f'Total anomalies: {len(self.anomalies)}')
        print('=' * 70)
        
        for i, a in enumerate(self.anomalies, 1):
            print(f'[{a["severity"]}] #{i} - {a["type"]}')
            print(f'  User: {a["user"]}')
            print(f'  Time: {a.get("timestamp", "N/A")}')
            print(f'  MITRE: {a.get("mitre", "N/A")}')
            print('-' * 50)

if __name__ == '__main__':
    if len(sys.argv) < 3:
        print(f'Usage: {sys.argv[0]} <baseline_cloudtrail> <current_cloudtrail>')
        sys.exit(1)
    
    analyzer = IdentityBehaviorBaseline()
    analyzer.load_cloudtrail(sys.argv[1])
    analyzer.detect_anomalies(sys.argv[2])
    analyzer.generate_hunting_report()

SELECT
    useridentity.arn AS actor_arn,
    eventname AS action,
    sourceipaddress AS source_ip,
    eventtime AS event_time,
    requestparameters
FROM cloudtrail_logs
WHERE eventname = 'AssumeRole'
    AND useridentity.type = 'AssumedRole'
    AND datehour >= date_format(date_add('day', -7, current_date), '%Y%m%d%H')
ORDER BY eventtime DESC
LIMIT 500;

SELECT
    useridentity.arn AS actor,
    eventname AS action,
    eventtime AS time,
    requestparameters,
    sourceipaddress AS ip
FROM cloudtrail_logs
WHERE eventname IN (
    'CreatePolicyVersion', 'SetDefaultPolicyVersion',
    'AttachUserPolicy', 'AttachRolePolicy',
    'PutUserPolicy', 'PutRolePolicy',
    'UpdateAssumeRolePolicy'
)
    AND useridentity.type != 'AWSService'
ORDER BY eventtime DESC;

SELECT
    useridentity.arn AS user_arn,
    COUNT(DISTINCT eventname) AS unique_actions,
    COUNT(DISTINCT sourceipaddress) AS unique_ips,
    MIN(eventtime) AS first_seen,
    MAX(eventtime) AS last_seen,
    COUNT(*) AS total_events
FROM cloudtrail_logs
WHERE datehour >= date_format(date_add('day', -1, current_date), '%Y%m%d%H')
GROUP BY useridentity.arn
HAVING COUNT(DISTINCT sourceipaddress) > 5
    OR COUNT(DISTINCT eventname) > 50
ORDER BY unique_ips DESC, unique_actions DESC;