SaaS OAuth令牌滥用与API密钥泄露取证深度分析

eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJodHRwczovL2lkcy5leGFtcGxlLmNvbSIs
InN1YiI6InVzZXJAMTIzIiwiYXVkIjoiaHR0cHM6Ly9hcGkuZXhhbXBsZS5jb20iLCJleHAiOjE3
MjM0NTY3ODksImlhdCI6MTcyMzQ1MzE4OSwic2NvcGUiOiJyZWFkIHdyaXRlIn0.signature

Get-MgAuditLogDirectoryAudit -Filter "activityDisplayName eq 'Consent to application'" `
  | Select-Object activityDateTime, initiatedBy, targetResources, result

gcloud workspace audit-logs list --filter="protopayload.methodName:AddToProject" \
  --format="table(timestamp, protoPayload.authenticationInfo.principalEmail, \
  protoPayload.servicedata_v1_1.AddToProjectRequest.event)"

grep -E "(state=|oauth_callback|code=)" /var/log/apache2/access.log \
  | awk '{print $1, $4, $7}' \
  | sort -k2 \
  | uniq -c \
  | sort -rn

log2timeline.py --storage-file session_analysis.plaso \
  /path/to/browser_evidence/Chrome/Default/Cookies \
  /path/to/browser_evidence/Chrome/Default/Local\ Storage/

psort.py -o l2tcsv session_analysis.plaso \
  --datetime-field "datetime" \
  -w session_timeline.csv

#!/bin/bash
LOG_DIR="/var/log/oauth"
THRESHOLD=50
WINDOW=3600

echo "=== OAuth Refresh Token 异常使用检测 ==="
echo "检测时间窗口: ${WINDOW}秒"
echo "阈值: ${THRESHOLD}次/窗口"
echo ""

find "$LOG_DIR" -name "*.json" -mtime -1 | while read log_file; do
    jq -r 'select(.grant_type == "refresh_token") | 
    "\(.timestamp) \(.client_ip) \(.user_id) \(.client_id)"' "$log_file" 2>/dev/null
done | awk -v threshold="$THRESHOLD" -v window="$WINDOW" '
{
    key = $2 "|" $3 "|" $4
    ts = mktime(gensub(/[-:]/, " ", "g", substr($1, 1, 19)))
    if (!(key in first_ts)) {
        first_ts[key] = ts
        count[key] = 1
    } else if (ts - first_ts[key] <= window) {
        count[key]++
    } else {
        first_ts[key] = ts
        count[key] = 1
    }
    if (count[key] >= threshold) {
        printf "[ALERT] %s 在 %d 秒内发起 %d 次 Refresh Token 请求 (IP: %s)\n", $3, window, count[key], $2
    }
}'

title = "API Key Detection Rules"

[[rules]]
description = "AWS Access Key ID"
regex = '''(?i)(aws[_\-]?access[_\-]?key[_\-]?id|AKIA[0-9A-Z]{16})'''
tags = ["key", "aws"]

[[rules]]
description = "GitHub Personal Access Token"
regex = '''ghp_[0-9a-zA-Z]{36}'''
tags = ["key", "github"]

[[rules]]
description = "Slack Bot Token"
regex = '''xoxb-[0-9]{10,13}-[0-9]{10,13}-[a-zA-Z0-9]{24}'''
tags = ["key", "slack"]

[[rules]]
description = "Generic API Key Pattern"
regex = '''(?i)(api[_\-]?key|apikey|secret[_\-]?key|auth[_\-]?token)[\"'\s]*[:=][\"'\s]*([0-9a-zA-Z\-_]{20,})'''
tags = ["key", "generic"]

trufflehog filesystem --directory /path/to/repo \
  --only-verified \
  --json \
  --output trufflehog_results.json

trufflehog git file:///path/to/repo \
  --since-commit HEAD~50 \
  --fail \
  --json

aws cloudtrail lookup-events \
  --lookup-attributes AttributeKey=AccessKeyId,AttributeValue=AKIAIOSFODNN7EXAMPLE \
  --start-time 2026-08-01T00:00:00Z \
  --end-time 2026-08-12T23:59:59Z \
  --query 'Events[*].{Time:EventTime,Event:EventName,Source:EventSource,User:Username}' \
  --output table

Connect-MgGraph -Scopes "Application.Read.All","Directory.Read.All"

Get-MgServicePrincipal -Filter "tags/any(t:t eq 'WindowsAzureActiveDirectoryIntegratedApp')" `
  | Select-Object DisplayName, AppId, CreatedDateTime, ServicePrincipalType `
  | Export-Csv -Path "entra_oauth_apps.csv" -NoTypeInformation

Get-MgServicePrincipalOAuth2PermissionGrant `
  | Where-Object { $_.Scope -match "ReadWrite|FullAccess|Directory" } `
  | Select-Object PrincipalDisplayName, ServicePrincipalDisplayName, Scope `
  | Export-Csv -Path "high_risk_scopes.csv" -NoTypeInformation

gcloud alpha iap oauth-clients list \
  --format="table(name,secret,expirationTokenUri)" \
  > gcp_oauth_clients.json

gcloud services consumer-org-policies list \
  --organization=123456789 \
  --filter="constraint=constraints/iam.allowedPolicyMemberDomains"

curl -s -H "Authorization: SSWS ${OKTA_API_TOKEN}" \
  "https://${OKTA_DOMAIN}/api/v1/apps?type=BASIC_AUTH&limit=200" \
  | jq '.[] | {label: .label, status: .status, created: .created, 
  signOnMode: .signOnMode, credentials: .credentials}'

TOKEN=$(curl -X PUT "http://169.254.169.254/latest/api/token" \
  -H "X-aws-ec2-metadata-token-ttl-seconds: 21600" \
  -H "X-aws-ec2-metadata-token: " 2>/dev/null)

curl -H "X-aws-ec2-metadata-token: $TOKEN" \
  http://169.254.169.254/latest/meta-data/iam/security-credentials/

aws ec2 flow-logs describe-flow-logs \
  --filter "Name=resource-id,Values=i-0123456789abcdef0" \
  --query 'FlowLogs[*].FlowLogId'

aws logs filter-log-events \
  --log-group-name /aws/vpc/flowlogs \
  --filter-pattern "169.254.169.254" \
  --start-time 1691000000000 \
  --output json | jq '.events[] | {src: .message}'

import boto3
import json
from datetime import datetime, timedelta

def detect_imds_access():
    client = boto3.client('logs')
    end_time = int(datetime.utcnow().timestamp() * 1000)
    start_time = int((datetime.utcnow() - timedelta(hours=24)).timestamp() * 1000)
    
    results = []
    log_groups = client.describe_log_groups(logGroupNamePrefix='/aws/lambda/')
    
    for lg in log_groups['logGroups']:
        events = client.filter_log_events(
            logGroupName=lg['logGroupName'],
            startTime=start_time,
            endTime=end_time,
            filterPattern='169.254.169.254'
        )
        if events['events']:
            results.append({
                'function': lg['logGroupName'],
                'access_count': len(events['events']),
                'timestamps': [e['timestamp'] for e in events['events'][:10]]
            })
    
    return results

if __name__ == '__main__':
    results = detect_imds_access()
    for r in results:
        print(f"[ALERT] Lambda {r['function']} accessed IMDS {r['access_count']} times")

title: Suspicious OAuth Application Consent - Consent Phishing Detection
id: f7a8b9c0-d1e2-4f3a-5b6c-7d8e9f0a1b2c
status: experimental
description: Detects suspicious third-party OAuth application consent events in Microsoft Entra ID that may indicate Consent Phishing attacks
references:
  - https://www.microsoft.com/en-us/security/blog/2023/11/16/protecting-against-vendor-in-the-middle-attacks/
author: Blue Team Forensics
date: 2026-08-12
tags:
  - attack.t1566.002
  - attack.t1199
  - oauth.security
logsource:
  product: azure
  service: auditlogs
detection:
  selection_event:
    operationName: 'Consent to application'
  selection_high_risk:
    properties|contains:
      - 'Mail.ReadWrite'
      - 'Files.ReadWrite.All'
      - 'Directory.ReadWrite.All'
      - 'User.ReadWrite.All'
      - 'full_access_as_user'
  selection_unverified:
    properties|contains:
      - 'unverified'
  condition: selection_event and (selection_high_risk or selection_unverified)
level: high
falsepositives:
  - Legitimate enterprise application integration
  - IT department approved applications

title: Anomalous OAuth Refresh Token Usage Pattern
id: a1b2c3d4-e5f6-7890-abcd-ef1234567890
status: experimental
description: Detects anomalous patterns in OAuth Refresh Token usage that may indicate token abuse or compromise
references:
  - https://datatracker.ietf.org/doc/html/rfc6749#section-10.4
author: Blue Team Forensics
date: 2026-08-12
tags:
  - attack.t1528
  - attack.t1550.004
  - oauth.security
logsource:
  product: oauth
  service: authorization_server
detection:
  selection:
    grant_type: 'refresh_token'
  timeframe: 1h
  condition: selection | count() by user_id > 30
level: medium
falsepositives:
  - Legitimate high-frequency API clients
  - Automated service accounts with proper justification

title: API Key Hardcoded in Source Code Repository
id: b2c3d4e5-f6a7-8901-bcde-f12345678901
status: experimental
description: Detects potential API key or secret hardcoded in source code based on GitHub Secret Scanning alerts
references:
  - https://docs.github.com/en/code-security/secret-scanning/about-secret-scanning
author: Blue Team Forensics
date: 2026-08-12
tags:
  - attack.t1552.001
  - api.key.security
logsource:
  product: github
  service: secret_scanning
detection:
  selection:
    alert_type:
      - 'aws_access_key'
      - 'github_personal_access_token'
      - 'slack_bot_token'
      - 'stripe_api_key'
      - 'generic_api_key'
  condition: selection
level: high
falsepositives:
  - False positive from secret scanning
  - Test keys with no production access

#!/bin/bash
TARGET_DIR="${1:-.}"
OUTPUT_DIR="/tmp/oauth_hunt_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "=== SaaS OAuth 令牌滥用狩猎 ==="
echo "目标目录: $TARGET_DIR"
echo "输出目录: $OUTPUT_DIR"
echo ""

echo "[1/5] 扫描 Git 历史中的 OAuth 令牌和 API 密钥..."
find "$TARGET_DIR" -name "*.git" -prune -o -type f -print | \
  xargs grep -rlE "(ghp_[0-9a-zA-Z]{36}|xoxb-[0-9]{10}|AKIA[0-9A-Z]{16}|sk-[0-9a-zA-Z]{32,})" \
  2>/dev/null > "$OUTPUT_DIR/credential_matches.txt" || true

echo "[2/5] 分析 OAuth 授权日志..."
find "$TARGET_DIR" -name "*.log" -o -name "*.json" | \
  xargs grep -E "(grant_type|authorization_code|refresh_token|client_credentials)" \
  2>/dev/null | \
  awk -F: '{print $1}' | sort | uniq -c | sort -rn > "$OUTPUT_DIR/oauth_grant_analysis.txt" || true

echo "[3/5] 检测异常 API 调用模式..."
find "$TARGET_DIR" -name "access.log" -o -name "*.jsonl" | while read log_file; do
    awk '/oauth|token|api_key|bearer/ {
        ip = $1
        count[ip]++
    }
    END {
        for (ip in count)
            if (count[ip] > 100)
                printf "%s: %d 次请求\n", ip, count[ip]
    }' "$log_file" 2>/dev/null
done > "$OUTPUT_DIR/high_frequency_ips.txt" || true

echo "[4/5] 检查浏览器 Cookie 和令牌存储..."
find "$TARGET_DIR" -name "Cookies" -path "*/Default/*" 2>/dev/null | \
  while read cookie_file; do
    sqlite3 "$cookie_file" \
      "SELECT name, host_key, path, expires_utc FROM cookies WHERE name LIKE '%session%' OR name LIKE '%token%' OR name LIKE '%oauth%';" \
      2>/dev/null
done > "$OUTPUT_DIR/browser_tokens.txt" || true

echo "[5/5] 生成狩猎报告..."
cat > "$OUTPUT_DIR/hunt_report.md" << EOF
# OAuth 令牌滥用狩猎报告
- 扫描时间: $(date '+%Y-%m-%d %H:%M:%S')
- 目标目录: $TARGET_DIR

## 发现汇总
- 硬编码凭据匹配: $(wc -l < "$OUTPUT_DIR/credential_matches.txt") 处
- 高频 API 调用 IP: $(wc -l < "$OUTPUT_DIR/high_frequency_ips.txt") 个
- 浏览器令牌存储: $(wc -l < "$OUTPUT_DIR/browser_tokens.txt") 条
EOF

echo ""
echo "狩猎完成。报告已保存至: $OUTPUT_DIR/hunt_report.md"

import json
import sys
import base64
from datetime import datetime
from collections import Counter, defaultdict

def decode_jwt_payload(token):
    try:
        parts = token.split('.')
        if len(parts) < 2:
            return None
        payload = parts[1]
        padding = 4 - len(payload) % 4
        payload += '=' * padding
        decoded = base64.urlsafe_b64decode(payload)
        return json.loads(decoded)
    except Exception:
        return None

def analyze_token_claims(claims):
    findings = []
    
    if 'exp' in claims and 'iat' in claims:
        lifetime = claims['exp'] - claims['iat']
        if lifetime > 86400:
            findings.append({
                'severity': 'MEDIUM',
                'message': f'Token lifetime exceeds 24h: {lifetime/3600:.1f} hours',
                'claim': 'exp/iat'
            })
    
    scope = claims.get('scope', claims.get('scp', ''))
    if isinstance(scope, str):
        scopes = scope.split()
    elif isinstance(scope, list):
        scopes = scope
    else:
        scopes = []
    
    dangerous_scopes = [
        'write', 'admin', 'full_access', 'manage',
        'Mail.ReadWrite', 'Files.ReadWrite.All', 'Directory.ReadWrite.All'
    ]
    for s in scopes:
        if s in dangerous_scopes:
            findings.append({
                'severity': 'HIGH',
                'message': f'Dangerous scope granted: {s}',
                'claim': 'scope'
            })
    
    roles = claims.get('roles', [])
    admin_roles = [r for r in roles if 'admin' in str(r).lower()]
    if admin_roles:
        findings.append({
            'severity': 'HIGH',
            'message': f'Admin roles detected: {admin_roles}',
            'claim': 'roles'
        })
    
    return findings

def analyze_token_dataset(log_file):
    token_claims = []
    issuer_counter = Counter()
    scope_counter = Counter()
    anomaly_tokens = []
    
    with open(log_file, 'r') as f:
        for line in f:
            try:
                entry = json.loads(line.strip())
                token = entry.get('token', entry.get('access_token', ''))
                if token:
                    claims = decode_jwt_payload(token)
                    if claims:
                        token_claims.append(claims)
                        issuer_counter[claims.get('iss', 'unknown')] += 1
                        scope_val = claims.get('scope', claims.get('scp', ''))
                        if isinstance(scope_val, str):
                            for s in scope_val.split():
                                scope_counter[s] += 1
                        
                        findings = analyze_token_claims(claims)
                        if any(f['severity'] == 'HIGH' for f in findings):
                            anomaly_tokens.append({
                                'claims': claims,
                                'findings': findings
                            })
            except json.JSONDecodeError:
                continue
    
    report = {
        'total_tokens_analyzed': len(token_claims),
        'unique_issuers': dict(issuer_counter),
        'top_scopes': dict(scope_counter.most_common(20)),
        'anomalous_tokens': len(anomaly_tokens),
        'high_risk_tokens': anomaly_tokens[:10]
    }
    
    return report

def print_report(report):
    print("=" * 60)
    print("JWT 令牌滥用分析报告")
    print("=" * 60)
    print(f"分析令牌总数: {report['total_tokens_analyzed']}")
    print(f"异常令牌数量: {report['anomalous_tokens']}")
    print()
    print("--- 签发者分布 ---")
    for issuer, count in report['unique_issuers'].items():
        print(f"  {issuer}: {count}")
    print()
    print("--- 高频 Scope ---")
    for scope, count in report['top_scopes'].items():
        print(f"  {scope}: {count}")
    print()
    if report['high_risk_tokens']:
        print("--- 高风险令牌 ---")
        for item in report['high_risk_tokens']:
            sub = item['claims'].get('sub', 'unknown')
            for finding in item['findings']:
                print(f"  [{finding['severity']}] {sub}: {finding['message']}")

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <token_log_file.json>")
        sys.exit(1)
    report = analyze_token_dataset(sys.argv[1])
    print_report(report)
    with open('token_analysis_report.json', 'w') as f:
        json.dump(report, f, indent=2, default=str)

import os
import re
import json
import hashlib
from pathlib import Path

PATTERNS = {
    'AWS Access Key': r'AKIA[0-9A-Z]{16}',
    'GitHub Token': r'ghp_[0-9a-zA-Z]{36}|github_pat_[0-9a-zA-Z]{22}_[0-9a-zA-Z]{59}',
    'Slack Token': r'xoxb-[0-9]{10,13}-[0-9]{10,13}-[a-zA-Z0-9]{24}',
    'Slack Webhook': r'https://hooks\.slack\.com/services/T[A-Z0-9]+/B[A-Z0-9]+/[a-zA-Z0-9]+',
    'Stripe Key': r'sk_live_[0-9a-zA-Z]{24,}|pk_live_[0-9a-zA-Z]{24,}',
    'Generic API Key': r'(?i)(?:api[_-]?key|apikey|secret[_-]?key|auth[_-]?token|access[_-]?token)["\'\s:=]+["\']?([0-9a-zA-Z\-_]{20,})',
    'Private Key': r'-----BEGIN (?:RSA |EC )?PRIVATE KEY-----',
    'JWT Token': r'eyJ[a-zA-Z0-9_-]+\.eyJ[a-zA-Z0-9_-]+\.[a-zA-Z0-9_-]+',
}

IGNORE_DIRS = {'.git', 'node_modules', 'venv', '__pycache__', '.venv', 'vendor', 'dist', 'build'}

def scan_file(filepath):
    findings = []
    try:
        with open(filepath, 'r', errors='ignore') as f:
            content = f.read()
            for name, pattern in PATTERNS.items():
                matches = re.finditer(pattern, content)
                for match in matches:
                    line_num = content[:match.start()].count('\n') + 1
                    findings.append({
                        'file': str(filepath),
                        'line': line_num,
                        'type': name,
                        'match': match.group()[:20] + '...',
                        'hash': hashlib.sha256(match.group().encode()).hexdigest()[:16]
                    })
    except (IOError, UnicodeDecodeError):
        pass
    return findings

def scan_directory(root_dir):
    all_findings = []
    files_scanned = 0
    for root, dirs, files in os.walk(root_dir):
        dirs[:] = [d for d in dirs if d not in IGNORE_DIRS]
        for filename in files:
            if filename.endswith(('.py', '.js', '.ts', '.yml', '.yaml', '.json', '.env', '.cfg', '.conf', '.sh', '.bash')):
                filepath = Path(root) / filename
                findings = scan_file(filepath)
                all_findings.extend(findings)
                files_scanned += 1
    return all_findings, files_scanned

def generate_report(findings, files_scanned):
    type_counts = Counter(f['type'] for f in findings)
    
    report = {
        'scan_summary': {
            'files_scanned': files_scanned,
            'total_findings': len(findings),
            'by_type': dict(type_counts)
        },
        'findings': findings
    }
    return report

from collections import Counter

if __name__ == '__main__':
    target = sys.argv[1] if len(sys.argv) > 1 else '.'
    print(f"扫描目标: {target}")
    findings, count = scan_directory(target)
    report = generate_report(findings, count)
    
    with open('secret_scan_report.json', 'w') as f:
        json.dump(report, f, indent=2)
    
    print(f"扫描完成: {count} 个文件, {len(findings)} 个发现")
    for ftype, cnt in report['scan_summary']['by_type'].items():
        print(f"  {ftype}: {cnt}")