多租户SaaS平台安全取证深度分析

SELECT * FROM customers WHERE tenant_id = 'T-2024-001' AND status = 'active';

SELECT * FROM customers WHERE status = 'active';

ALTER SYSTEM SET pgaudit.log = 'write, ddl';
ALTER SYSTEM SET pgaudit.log_parameter = on;
SELECT pg_reload_conf();

SET search_path TO tenant_target_schema;
SELECT * FROM sensitive_table;

curl -X GET "https://api.saas-platform.com/v2/documents/TARGET-TENANT-DOC-001" \
  -H "Authorization: Bearer eyJhbGciOiJSUzI1NiIs..." \
  -H "X-Tenant-ID: ATTACKER-TENANT-999"

cat api_access.log | jq 'select(
  .request_uri | test("/v2/(documents|files|records)/") and
  .tenant_id_from_token != .tenant_id_from_resource
) | {
  timestamp: .timestamp,
  user: .user_id,
  ip: .client_ip,
  method: .http_method,
  resource: .request_uri,
  token_tenant: .tenant_id_from_token,
  resource_tenant: .tenant_id_from_resource,
  response_code: .response_code
}'

apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: tenant-isolation-policy
  namespace: tenant-alpha
spec:
  podSelector: {}
  policyTypes:
    - Ingress
    - Egress
  ingress:
    - from:
        - namespaceSelector:
            matchLabels:
              tenant: alpha
  egress:
    - to:
        - namespaceSelector:
            matchLabels:
              tenant: alpha

kubectl logs -n kube-apiserver -l component=kube-apiserver --tail=50000 | \
  jq 'select(
    .verb == "create" or .verb == "update" or .verb == "patch"
  ) | select(
    .objectRef.resource == "networkpolicies" or
    .objectRef.resource == "pods"
  ) | {
    timestamp: .requestReceivedTimestamp,
    user: .user.username,
    verb: .verb,
    resource: .objectRef.resource,
    namespace: .objectRef.namespace,
    name: .objectRef.name,
    userAgent: .userAgent,
    sourceIPs: .sourceIPs
}'

import json
from collections import defaultdict

def detect_horizontal_privilege_escape(api_logs):
    user_resource_map = defaultdict(lambda: defaultdict(set))
    alerts = []

    for log in api_logs:
        user = log.get("user_id")
        tenant = log.get("tenant_id")
        resource_prefix = log.get("resource_prefix")
        resource_id = log.get("resource_id")

        if user and tenant and resource_prefix:
            user_resource_map[user][(tenant, resource_prefix)].add(resource_id)

    for user, tenant_resources in user_resource_map.items():
        tenants_accessed = set(t for t, _ in tenant_resources.keys())
        if len(tenants_accessed) > 1:
            alerts.append({
                "user": user,
                "tenants_accessed": list(tenants_accessed),
                "severity": "HIGH",
                "attack_type": "Horizontal Privilege Escalation",
                "mitre_attack": "T1530"
            })

    return alerts

echo "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c2VyMTIzIiwidGVuYW50X2lkIjoiVC0yMDI0LTAwMSIsInNjb3BlIjoiZG9jdW1lbnRzOnJlYWQgZG9jdW1lbnRzOndyaXRlIiwicm9sZSI6InVzZXIiLCJpYXQiOjE3MjAwMDAwMDAsImV4cCI6MTcyMDAwMzYwMCwiYXVkIjoiaHR0cHM6Ly9hcGkuc2Fhcy1wbGF0Zm9ybS5jb20iLCJpc3MiOiJodHRwczovL2lkcy5zYWFzLXBsYXRmb3JtLmNvbSIsImp0aSI6IjEyMzQ1Njc4OTAifQ" | \
  cut -d'.' -f2 | python3 -m base64 -d 2>/dev/null | python3 -m json.tool

{
  "sub": "user123",
  "tenant_id": "T-2024-001",
  "scope": "documents:read documents:write",
  "role": "user",
  "iat": 1720000000,
  "exp": 1720003600,
  "aud": "https://api.saas-platform.com",
  "iss": "https://ids.saas-platform.com",
  "jti": "1234567890"
}

cat saas_audit.log | jq 'select(
  .event_type == "token_refresh" and
  .client_ip != null
) | {
  timestamp: .timestamp,
  user: .user_id,
  refresh_token_id: .refresh_token_jti,
  client_ip: .client_ip,
  user_agent: .user_agent,
  device_id: .device_id
}' | python3 -c "
import sys, json
from collections import defaultdict
logs = [json.loads(l) for l in sys.stdin]
token_ips = defaultdict(list)
for log in logs:
    token_ips[log['refresh_token_id']].append({
        'ip': log['client_ip'],
        'timestamp': log['timestamp'],
        'user_agent': log['user_agent']
    })
for token_id, accesses in token_ips.items():
    unique_ips = set(a['ip'] for a in accesses)
    if len(unique_ips) > 2:
        print(f'[ALERT] Token {token_id} used from {len(unique_ips)} different IPs')
        for a in accesses:
            print(f'  {a[\"timestamp\"]} - {a[\"ip\"]} ({a[\"user_agent\"]})')
"

python3 -c "
import base64, json
header = base64.urlsafe_b64encode(json.dumps({'alg':'none','typ':'JWT'}).encode()).rstrip(b'=')
payload = base64.urlsafe_b64encode(json.dumps({'sub':'admin','tenant_id\":\"T-999-ADMIN\",\"role\":\"super_admin\"}).encode()).rstrip(b'=')
print(f'{header.decode()}.{payload.decode()}.')
"

cat auth_service.log | jq 'select(
  .event == "jwt_verification_failed" or
  .event == "token_validation_error"
) | {
  timestamp: .timestamp,
  token_header_alg: .jwt_header_alg,
  token_issuer: .jwt_iss,
  error: .error_message,
  client_ip: .client_ip,
  user_agent: .user_agent
}' | sort_by(.timestamp)

cat web_server.log | grep "session_id" | \
  awk '{
    split($0, a, " ");
    timestamp = a[1];
    session_id = a[5];
    event = a[6];
    if (event == "login_attempt") {
      pre_login_session[session_id] = timestamp;
    }
    if (event == "login_success" && session_id in pre_login_session) {
      print "[ALERT] Session not regenerated after login: " session_id;
      print "  Pre-login: " pre_login_session[session_id];
      print "  Post-login: " timestamp;
    }
  }'

cat scim_audit.log | jq 'select(
  .event_type == "scim_operation" and
  (.operation == "POST" or .operation == "PATCH" or .operation == "DELETE")
) | {
  timestamp: .timestamp,
  operation: .operation,
  resource_type: .resource_type,
  target_user: .target_user_id,
  changes: .patch_operations,
  initiated_by: .initiated_by,
  source_ip: .source_ip
}' | python3 -c "
import sys, json
from datetime import datetime, timedelta
logs = [json.loads(l) for l in sys.stdin]
window = timedelta(minutes=10)
for i, log in enumerate(logs):
    recent = [l for l in logs[:i+1]
              if datetime.fromisoformat(l['timestamp']) >= datetime.fromisoformat(log['timestamp']) - window
              and l['operation'] == 'POST']
    if len(recent) > 5:
        print(f'[ALERT] Mass user creation detected: {len(recent)} users in {window}')
        print(f'  Initiated by: {log[\"initiated_by\"]}')
        print(f'  Source IP: {log[\"source_ip\"]}')
        break
"

cat saml_debug.log | grep -A 100 "SAMLResponse" | \
  python3 -c "
import sys, xml.etree.ElementTree as ET
import base64, re

content = sys.stdin.read()
match = re.search(r'SAMLResponse=([A-Za-z0-9+/=]+)', content)
if match:
    decoded = base64.b64decode(match.group(1)).decode('utf-8')
    root = ET.fromstring(decoded)
    ns = {'saml': 'urn:oasis:names:tc:SAML:2.0:assertion'}
    for attr in root.findall('.//saml:AttributeStatement/saml:Attribute', ns):
        name = attr.get('Name')
        values = [v.text for v in attr.findall('saml:AttributeValue', ns)]
        print(f'{name}: {values}')
    issuer = root.find('.//saml:Issuer', ns)
    subject = root.find('.//saml:Subject/saml:NameID', ns)
    conditions = root.find('.//saml:Conditions', ns)
    print(f'Issuer: {issuer.text if issuer is not None else \"N/A\"}')
    print(f'Subject: {subject.text if subject is not None else \"N/A\"}')
    print(f'NotBefore: {conditions.get(\"NotBefore\") if conditions is not None else \"N/A\"}')
    print(f'NotOnOrAfter: {conditions.get(\"NotOnOrAfter\") if conditions is not None else \"N/A\"}')
"

Get-AuditLogSearch -StartDate (Get-Date).AddDays(-7) -EndDate (Get-Date) -RecordType 12 | \
  Where-Object {
    $_.Status.ErrorCode -ne 0 -or
    $_.LocationInfo.Country -notin @("CN", "US") -or
    $_.AppliedConditionalAccessPolicies -contains "Block"
  } | Select-Object CreatedDateTime, UserPrincipalName, IPAddress, \
    Status.ErrorCode, Status.FailureReason, LocationInfo, \
    DeviceDetail.OperatingSystem, ConditionalAccessStatus | \
  Export-Csv -Path "suspicious_signins.csv" -NoTypeInformation

cat proxy_access.log | awk '{
  split($0, a, " ");
  host = a[7];
  if (host !~ /approved-saas-list/) {
    print a[1], a[2], host, a[4], a[8]
  }
}' | sort -k3 | uniq -c | sort -rn | head -50

#!/bin/bash
CONFIG_ENDPOINT="https://admin.saas-platform.com/api/v1/security-settings"
API_TOKEN=$(cat /secure/path/to/saas_api_token)
BASELINE_FILE="/secure/path/to/config_baseline.json"

CURRENT_CONFIG=$(curl -s -H "Authorization: Bearer $API_TOKEN" "$CONFIG_ENDPOINT")

BASELINE_HASH=$(sha256sum "$BASELINE_FILE" | awk '{print $1}')
CURRENT_HASH=$(echo "$CURRENT_CONFIG" | sha256sum | awk '{print $1}')

if [ "$BASELINE_HASH" != "$CURRENT_HASH" ]; then
    diff_result=$(diff <(python3 -m json.tool "$BASELINE_FILE") \
                       <(echo "$CURRENT_CONFIG" | python3 -m json.tool))
    echo "CRITICAL: SaaS configuration drift detected"
    echo "Baseline: $BASELINE_HASH"
    echo "Current: $CURRENT_HASH"
    echo "Differences:"
    echo "$diff_result"

    curl -s -X POST "https://siem.internal/api/v1/alerts" \
      -H "Content-Type: application/json" \
      -H "X-API-Key: $SIEM_API_KEY" \
      -d "{
        \"severity\": \"critical\",
        \"type\": \"saas_config_drift\",
        \"details\": {
          \"baseline_hash\": \"$BASELINE_HASH\",
          \"current_hash\": \"$CURRENT_HASH\",
          \"diff\": $(echo "$diff_result" | python3 -c 'import sys,json; print(json.dumps(sys.stdin.read()))')
        }
      }"
fi

cat api_gateway.log | jq 'select(
  .response_code >= 200 and .response_code < 300 and
  .response_size_bytes > 1048576
) | {
  timestamp: .timestamp,
  method: .http_method,
  endpoint: .request_path,
  client_ip: .client_ip,
  user: .authenticated_user,
  tenant: .tenant_id,
  response_size: .response_size_bytes,
  duration_ms: .request_duration_ms
}' | sort_by(.response_size) | reverse | head -20

cat graphql_api.log | jq 'select(
  .query_depth > 5 or
  .query_complexity > 1000 or
  .resolver_errors > 0
) | {
  timestamp: .timestamp,
  query_hash: .query_hash,
  query_depth: .query_depth,
  query_complexity: .query_complexity,
  fields_accessed: .fields_resolved,
  user: .user_id,
  tenant: .tenant_id,
  client_ip: .client_ip
}' | sort_by(.query_complexity) | reverse | head -20

import json
from collections import defaultdict

def build_ownership_baseline(audit_logs):
    ownership_map = defaultdict(set)
    for log in audit_logs:
        user = log.get("user_id")
        resource_id = log.get("resource_id")
        tenant = log.get("tenant_id")
        if user and resource_id:
            ownership_map[resource_id].add((user, tenant))
    return ownership_map

def detect_bola(ownership_map, access_logs):
    alerts = []
    for log in access_logs:
        resource_id = log.get("resource_id")
        accessing_user = log.get("user_id")
        accessing_tenant = log.get("tenant_id")

        if resource_id in ownership_map:
            owners = ownership_map[resource_id]
            if (accessing_user, accessing_tenant) not in owners:
                alerts.append({
                    "severity": "CRITICAL",
                    "attack_type": "BOLA",
                    "mitre_attack": "T1530",
                    "user": accessing_user,
                    "tenant": accessing_tenant,
                    "resource": resource_id,
                    "legitimate_owners": [
                        {"user": u, "tenant": t} for u, t in owners
                    ],
                    "timestamp": log.get("timestamp"),
                    "http_method": log.get("http_method"),
                    "endpoint": log.get("request_path")
                })
    return alerts

gitleaks detect --source /path/to/repo --report-format json --report-path gitleaks_report.json --verbose

cat gitleaks_report.json | jq '.[] | {
  rule: .RuleID,
  file: .File,
  start_line: .StartLine,
  end_line: .EndLine,
  secret: (.Secret | split("") | last_n(4) | join("")),
  entropy: .Entropy,
  author: .Author,
  email: .Email,
  commit: .Commit,
  date: .Date
}'

cat api_rate_limit.log | jq 'select(
  .event == "rate_limit_bypass_suspected" or
  (.request_count > 100 and .unique_source_ips > 50)
) | {
  timestamp: .window_start,
  api_key_prefix: .api_key_prefix,
  endpoint: .endpoint,
  request_count: .request_count,
  unique_source_ips: .unique_source_ips,
  unique_user_agents: .unique_user_agents,
  status: .rate_limit_status
}'

def build_saas_attack_chain(event_logs):
    attack_chain = {
        "reconnaissance": [],
        "initial_access": [],
        "privilege_escalation": [],
        "lateral_movement": [],
        "data_exfiltration": [],
        "anti_forensics": []
    }

    for log in event_logs:
        event_type = log.get("event_type")
        confidence = log.get("evidence_strength")

        if event_type == "api_enumeration" and confidence == "HIGH":
            attack_chain["reconnaissance"].append(log)
        elif event_type in ("credential_theft", "token_abuse") and confidence in ("HIGH", "MEDIUM"):
            attack_chain["initial_access"].append(log)
        elif event_type in ("role_change", "permission_grant") and confidence == "HIGH":
            attack_chain["privilege_escalation"].append(log)
        elif event_type in ("cross_tenant_access", "bola_detected") and confidence == "HIGH":
            attack_chain["lateral_movement"].append(log)
        elif event_type in ("bulk_export", "api_mass_download") and confidence in ("HIGH", "MEDIUM"):
            attack_chain["data_exfiltration"].append(log)
        elif event_type in ("log_tampering", "config_drift") and confidence == "MEDIUM":
            attack_chain["anti_forensics"].append(log)

    chain_completeness = sum(1 for phase in attack_chain.values() if phase) / len(attack_chain)
    return attack_chain, chain_completeness

title: SaaS平台跨租户BOLA攻击检测
id: d7f8e9a0-1234-5678-9abc-def012345678
status: experimental
description: 检测多租户SaaS平台中可能的BOLA（Broken Object Level Authorization）跨租户访问攻击
references:
  - https://owasp.org/API-Security/editions/2023/en/0xa1-broken-object-level-authorisation/
author: x7peeps蓝队
date: 2026/07/13
tags:
  - attack.t1530
  - attack.lateral_movement
  - saas_security
logsource:
  category: api_access
  product: saas_platform
detection:
  selection_bola:
    event_type: api_request
    response_code:
      - 200
      - 201
  selection_tenant_mismatch:
    tenant_id_from_token|field_is_not: tenant_id_from_resource
  filter_legitimate:
    endpoint|contains:
      - '/public/'
      - '/healthcheck'
  condition: selection_bola and selection_tenant_mismatch and not filter_legitimate
fields:
  - timestamp
  - user_id
  - client_ip
  - http_method
  - request_path
  - tenant_id_from_token
  - tenant_id_from_resource
  - response_code
  - response_size_bytes
falsepositives:
  - 合法的跨租户API调用（如管理平台的租户切换功能）
level: critical

title: SaaS平台SCIM批量用户创建检测
id: a1b2c3d4-5678-9abc-def0-123456789012
status: experimental
description: 检测SCIM Provisioning接口的批量用户创建行为，可能为后门账号植入
author: x7peeps蓝队
date: 2026/07/13
tags:
  - attack.t1136
  - attack.initial_access
  - saas_security
logsource:
  category: scim_audit
  product: saas_platform
detection:
  selection_scim:
    event_type: scim_operation
    operation: POST
    resource_type: User
  timeframe: 10m
  condition: selection_scim | count() by initiated_by, source_ip > 5
fields:
  - timestamp
  - initiated_by
  - source_ip
  - target_user_id
  - operation
falsepositives:
  - 批量员工入职场景
level: high

title: SaaS平台Refresh Token异常地理位置使用
id: e5f6a7b8-9012-3456-7890-abcdef123456
status: experimental
description: 检测Refresh Token在物理距离极远的地理位置之间快速切换使用
author: x7peeps蓝队
date: 2026/07/13
tags:
  - attack.t1550.001
  - attack.credential_access
  - saas_security
logsource:
  category: token_management
  product: saas_platform
detection:
  selection_refresh:
    event_type: token_refresh
  selection_geo_anomaly:
    geo_distance_km|gte: 500
    time_delta_minutes|lte: 60
  condition: selection_refresh and selection_geo_anomaly
fields:
  - timestamp
  - user_id
  - refresh_token_jti
  - previous_ip
  - current_ip
  - previous_country
  - current_country
  - geo_distance_km
  - time_delta_minutes
falsepositives:
  - 用户使用VPN切换出口
level: critical

#!/bin/bash
SaaS_API_ENDPOINT="https://admin.saas-platform.com/api/v1"
SIEM_ENDPOINT="https://siem.internal/api/v1/events"
API_TOKEN=$(cat /secure/saas_admin_token)
REPORT_DIR="/var/lib/saas-forensics/reports"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)

mkdir -p "$REPORT_DIR"

curl -s -H "Authorization: Bearer $API_TOKEN" \
  "$SaaS_API_ENDPOINT/audit-logs?since=24h&event_types=login,token_refresh,api_access,role_change,scim_operation" | \
  python3 -c "
import sys, json
from collections import defaultdict

logs = json.loads(sys.stdin.read())

anomalies = {
    'cross_tenant_access': [],
    'token_abuse': [],
    'privilege_escalation': [],
    'scim_anomaly': [],
    'impossible_travel': []
}

for log in logs:
    event = log.get('event_type', '')

    if event == 'api_access':
        token_tenant = log.get('tenant_id_from_token', '')
        resource_tenant = log.get('tenant_id_from_resource', '')
        if token_tenant and resource_tenant and token_tenant != resource_tenant:
            anomalies['cross_tenant_access'].append(log)

    if event == 'token_refresh':
        user = log.get('user_id', '')
        ips = log.get('unique_source_ips', 0)
        if ips > 3:
            anomalies['token_abuse'].append(log)

    if event == 'role_change':
        if log.get('new_role') in ('super_admin', 'tenant_admin') and \
           log.get('old_role') in ('user', 'viewer'):
            anomalies['privilege_escalation'].append(log)

    if event == 'scim_operation':
        if log.get('operation') == 'POST' and log.get('resource_type') == 'User':
            anomalies['scim_anomaly'].append(log)

    if event == 'login':
        if log.get('impossible_travel_flag'):
            anomalies['impossible_travel'].append(log)

report = {
    'scan_timestamp': '$TIMESTAMP',
    'total_logs_analyzed': len(logs),
    'anomalies': {k: len(v) for k, v in anomalies.items()},
    'details': anomalies
}

print(json.dumps(report, indent=2, default=str))
" > "$REPORT_DIR/saas_hunt_${TIMESTAMP}.json"

ALERT_COUNT=$(cat "$REPORT_DIR/saas_hunt_${TIMESTAMP}.json" | python3 -c "
import sys, json
report = json.loads(sys.stdin.read())
total = sum(report['anomalies'].values())
print(total)
")

if [ "$ALERT_COUNT" -gt 0 ]; then
    echo "[CRITICAL] Detected $ALERT_COUNT SaaS security anomalies"
    curl -s -X POST "$SIEM_ENDPOINT" \
      -H "Content-Type: application/json" \
      -H "X-API-Key: $SIEM_API_KEY" \
      -d @"$REPORT_DIR/saas_hunt_${TIMESTAMP}.json"
else
    echo "[INFO] No SaaS security anomalies detected"
fi

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

def load_siem_logs(log_path):
    with open(log_path, 'r') as f:
        return json.loads(f.read())

def analyze_saas_security_events(logs):
    results = {
        "bola_candidates": [],
        "token_abuse_candidates": [],
        "scim_anomalies": [],
        "privilege_escalation_events": [],
        "data_exfiltration_indicators": [],
        "summary": {}
    }

    user_api_patterns = defaultdict(lambda: defaultdict(int))
    token_usage_tracker = defaultdict(list)
    resource_access_map = defaultdict(set)
    user_role_history = defaultdict(list)

    for log in logs:
        event_type = log.get("event_type")
        user = log.get("user_id")
        tenant = log.get("tenant_id")
        timestamp = log.get("timestamp")
        client_ip = log.get("client_ip")

        if event_type == "api_access":
            resource_id = log.get("resource_id")
            resource_tenant = log.get("resource_tenant")
            endpoint = log.get("endpoint")
            response_code = log.get("response_code")

            user_api_patterns[user][endpoint] += 1

            if resource_tenant and resource_tenant != tenant:
                results["bola_candidates"].append({
                    "user": user,
                    "requesting_tenant": tenant,
                    "resource_tenant": resource_tenant,
                    "endpoint": endpoint,
                    "resource_id": resource_id,
                    "response_code": response_code,
                    "timestamp": timestamp,
                    "client_ip": client_ip,
                    "severity": "CRITICAL",
                    "mitre_attack": "T1530"
                })

            resource_access_map[user].add(resource_tenant or tenant)

        elif event_type == "token_refresh":
            token_usage_tracker[user].append({
                "timestamp": timestamp,
                "ip": client_ip,
                "device_id": log.get("device_id")
            })

        elif event_type == "role_change":
            user_role_history[user].append({
                "timestamp": timestamp,
                "old_role": log.get("old_role"),
                "new_role": log.get("new_role"),
                "changed_by": log.get("changed_by")
            })

            if log.get("new_role") in ("super_admin", "tenant_admin"):
                results["privilege_escalation_events"].append({
                    "user": user,
                    "new_role": log.get("new_role"),
                    "changed_by": log.get("changed_by"),
                    "timestamp": timestamp,
                    "severity": "HIGH",
                    "mitre_attack": "T1098"
                })

        elif event_type == "scim_operation":
            if log.get("operation") == "POST" and log.get("resource_type") == "User":
                results["scim_anomalies"].append({
                    "user_created": log.get("target_user_id"),
                    "created_by": log.get("initiated_by"),
                    "timestamp": timestamp,
                    "source_ip": log.get("source_ip")
                })

    for user, accesses in token_usage_tracker.items():
        if len(accesses) >= 2:
            for i in range(1, len(accesses)):
                ip_changes = accesses[i]["ip"] != accesses[i-1]["ip"]
                if ip_changes:
                    results["token_abuse_candidates"].append({
                        "user": user,
                        "previous_ip": accesses[i-1]["ip"],
                        "current_ip": accesses[i]["ip"],
                        "previous_time": accesses[i-1]["timestamp"],
                        "current_time": accesses[i]["timestamp"],
                        "severity": "MEDIUM",
                        "mitre_attack": "T1550.001"
                    })

    total_anomalies = (
        len(results["bola_candidates"]) +
        len(results["token_abuse_candidates"]) +
        len(results["scim_anomalies"]) +
        len(results["privilege_escalation_events"])
    )

    results["summary"] = {
        "total_logs_analyzed": len(logs),
        "total_anomalies": total_anomalies,
        "bola_candidates": len(results["bola_candidates"]),
        "token_abuse_candidates": len(results["token_abuse_candidates"]),
        "scim_anomalies": len(results["scim_anomalies"]),
        "privilege_escalation_events": len(results["privilege_escalation_events"]),
        "analysis_timestamp": datetime.utcnow().isoformat(),
        "risk_level": "CRITICAL" if total_anomalies > 10 else "HIGH" if total_anomalies > 0 else "LOW"
    }

    return results

def generate_hunting_queries(results):
    queries = []

    for bola in results["bola_candidates"]:
        queries.append({
            "query_type": "BOLA Investigation",
            "splunk_query": f'index=api_access user="{bola["user"]}" | stats values(resource_id) as accessed_resources values(resource_tenant) as tenants by endpoint',
            "purpose": f"Investigate cross-tenant access by user {bola['user']}"
        })

    for escalation in results["privilege_escalation_events"]:
        queries.append({
            "query_type": "Privilege Escalation Investigation",
            "splunk_query": f'index=audit_logs user="{escalation["user"]}" | table timestamp old_role new_role changed_by client_ip',
            "purpose": f"Investigate privilege escalation for user {escalation['user']}"
        })

    return queries

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python3 saas_analyzer.py <log_file.json>")
        sys.exit(1)

    logs = load_siem_logs(sys.argv[1])
    results = analyze_saas_security_events(logs)

    print(json.dumps(results, indent=2, default=str))

    hunting_queries = generate_hunting_queries(results)
    if hunting_queries:
        print("\n--- Recommended Hunting Queries ---")
        for q in hunting_queries:
            print(f"\n[{q['query_type']}] {q['purpose']}")
            print(f"  {q['splunk_query']}")

# 网络IOC
IP: 185.100.87[.]204（攻击者使用的VPN出口IP）
IP: 23.227.175[.]16（Lapsus$成员关联IP）

# 用户代理
UA: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/96.0.4664.110 Safari/537.36"

# Okta系统日志事件类型
EventType: user.session.impersonation.started
EventType: user.authentication.auth_via_mfa
EventType: user.account.reset_password
EventType: system.email.send

# 关联域名
lapsus0x[.]pro（Lapsus$暗网站）
lapsusgroup[.]telegram（Telegram频道）

# DNS IOC
CNAME: *.sfdc Salesforce Experience Cloud CNAME模式
检测工具: subfinder -d target.com -silent | grep salesforce

# 域名接管检测
工具: nuclei -t takeover/takeover-templates/ -u target.com

# 网络IOC
User-Agent特征: Salesforce内置页面特定UA字符串
钓鱼页面特征: 伪造Salesforce登录页面的HTML特征

# Salesforce API日志特征
EventType: /services/data/v*/sobjects/ 异常批量查询
EventType: /services/oauth2/token 异常Token交换

curl -s -H "Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGc..." \
  "https://graph.microsoft.com/v1.0/messages?\$top=500&\$select=subject,from,body,receivedDateTime" | \
  python3 -c "import sys,json; [print(f'{m[\"receivedDateTime\"]} | {m[\"from\"][\"emailAddress\"][\"address\"]} | {m[\"subject\"]}') for m in json.loads(sys.stdin.read()).get('value',[])]"

# Azure AD审计日志事件类型
EventType: Add application
EventType: Add OAuth2PermissionGrant
EventType: Consensus to application

# Microsoft Graph API异常调用模式
Endpoint: /v1.0/messages?$top=500
Endpoint: /v1.0/users/*/messages
Endpoint: /v1.0/drives/root/children
Endpoint: /v1.0/me/mailFolders/Inbox/messages

# Azure AD应用属性异常
RequiredResourceAccess: 高权限scope组合
SignInAudience: AzureADMultipleOrgs