零信任架构安全取证深度分析

cat azure_ad_signin.log | jq 'select(.status.errorCode == "500121" or .status.errorCode == "0") | {time: .createdDateTime, user: .userPrincipalName, ip: .ipAddress, status: .status, mfaDetail: .mfaDetail}'

from geopy.distance import geodesic
from datetime import datetime

def detect_impossible_travel(signin_events):
    alerts = []
    for i in range(1, len(signin_events)):
        prev = signin_events[i - 1]
        curr = signin_events[i]
        if prev.get("user") != curr.get("user"):
            continue
        prev_loc = (prev["latitude"], prev["longitude"])
        curr_loc = (curr["latitude"], curr["longitude"])
        distance_km = geodesic(prev_loc, curr_loc).kilometers
        time_diff_hours = (curr["timestamp"] - prev["timestamp"]).total_seconds() / 3600
        if time_diff_hours > 0:
            speed_kmh = distance_km / time_diff_hours
            if speed_kmh > 900:
                alerts.append({
                    "user": curr["user"],
                    "prev_location": prev["location_name"],
                    "curr_location": curr["location_name"],
                    "distance_km": round(distance_km, 2),
                    "time_diff_hours": round(time_diff_hours, 2),
                    "implied_speed_kmh": round(speed_kmh, 2),
                    "risk_level": "CRITICAL"
                })
    return alerts

import jwt
import json
from datetime import datetime

def analyze_jwt_security(token):
    findings = []
    try:
        header = jwt.get_unverified_header(token)
        payload = jwt.decode(token, options={"verify_signature": False})

        if header.get("alg") == "none":
            findings.append({"issue": "None Algorithm", "severity": "CRITICAL"})

        if header.get("alg") == "HS256" and "kid" not in header:
            findings.append({"issue": "HS256 without kid - possible key confusion", "severity": "HIGH"})

        if "exp" not in payload:
            findings.append({"issue": "Missing expiration claim", "severity": "HIGH"})
        else:
            exp_time = datetime.fromtimestamp(payload["exp"])
            iat_time = datetime.fromtimestamp(payload.get("iat", 0))
            validity_hours = (exp_time - iat_time).total_seconds() / 3600
            if validity_hours > 24:
                findings.append({"issue": f"Excessive token validity: {validity_hours:.0f}h", "severity": "MEDIUM"})

        sensitive_claims = ["password", "ssn", "credit_card", "secret"]
        for claim in payload:
            if claim.lower() in sensitive_claims:
                findings.append({"issue": f"Sensitive data in claim: {claim}", "severity": "HIGH"})

    except jwt.InvalidTokenError as e:
        findings.append({"issue": f"Invalid token: {str(e)}", "severity": "CRITICAL"})

    return findings

{
  "policyName": "Require MFA for Admin Roles",
  "conditions": {
    "users": {"includeGroups": ["Global Admins"]},
    "applications": {"includeApplications": ["Office 365"]},
    "locations": {"includeLocations": ["All"]}
  },
  "grantControls": {
    "operator": "OR",
    "builtInControls": ["mfa", "compliantDevice"]
  },
  "sessionControls": {
    "signInFrequency": {"value": 4, "type": "hours"}
  }
}

title: 零信任环境 - MFA疲劳攻击检测
id: 7a3b1c4d-5e6f-7890-abcd-ef1234567890
status: experimental
description: 检测短时间内多次MFA推送后突然成功的模式，可能指示MFA疲劳攻击
references:
  - https://learn.microsoft.com/en-us/entra/identity/monitoring-health/concept-sign-in-log-activity-details
author: x7peeps
date: 2026/07/11
tags:
  - attack.credential_access
  - attack.t1621
  - zero_trust
logsource:
  product: azure
  service: signinlogs
detection:
  selection:
    status.errorCode:
      - 500121
      - 500125
  timeframe: 10m
  condition: selection | count by userPrincipalName > 5
falsepositives:
  - 用户设备问题导致的重复MFA失败
level: high

title: 零信任环境 - 不可能旅行检测
id: 8b4c2d5e-6f70-8901-bcde-f23456789012
status: experimental
description: 检测同一用户在极短时间内从相距甚远的地理位置登录
author: x7peeps
date: 2026/07/11
tags:
  - attack.initial_access
  - attack.t1078
  - zero_trust
logsource:
  product: azure
  service: signinlogs
detection:
  selection_success:
    status.errorCode: 0
  timeframe: 1h
  condition: selection_success | dcount(locationDetails.city) > 1
    by userPrincipalName
falsepositives:
  - 使用VPN导致的IP地理位置不准
  - 用户确实在快速旅行
level: high

curl -X GET "https://graph.microsoft.com/v1.0/deviceManagement/managedDevices" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" | jq '.value[] | {
    deviceName: .deviceName,
    complianceState: .complianceState,
    lastSyncDateTime: .lastSyncDateTime,
    osVersion: .osVersion,
    jailBroken: .jailBroken,
    encryptionDiscovered: .encryptionDiscovered,
    threatProtectionEnabled: .threatProtectionEnabled
  }'

openssl s_client -connect device-enrollment.company.com:443 -showcerts < /dev/null 2>/dev/null | openssl x509 -noout -subject -issuer -dates -serial

jq 'select(.eventSource == "DeviceCompliance" and .complianceState == "Noncompliant") | {
  timestamp: .timestamp,
  deviceName: .deviceName,
  user: .userPrincipalName,
  previousState: .previousComplianceState,
  newState: .complianceState,
  reason: .complianceReason
}' device_compliance_events.json

#!/bin/bash
TENANT_ID="${AZURE_TENANT_ID}"
CLIENT_ID="${AZURE_CLIENT_ID}"
CLIENT_SECRET="${AZURE_CLIENT_SECRET}"
TOKEN=$(curl -s -X POST "https://login.microsoftonline.com/${TENANT_ID}/oauth2/v2.0/token" \
  -d "client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}&scope=https://graph.microsoft.com/.default" \
  | jq -r '.access_token')

echo "=== Device Compliance Summary ==="
curl -s -X GET "https://graph.microsoft.com/v1.0/deviceManagement/managedDevices?\$top=999" \
  -H "Authorization: Bearer ${TOKEN}" | jq '{
    total: (.value | length),
    compliant: [.value[] | select(.complianceState == "compliant")] | length,
    noncompliant: [.value[] | select(.complianceState == "noncompliant")] | length,
    conflict: [.value[] | select(.complianceState == "conflict")] | length,
    error: [.value[] | select(.complianceState == "error")] | length
  }'

echo ""
echo "=== High Risk: Non-compliant Devices ==="
curl -s -X GET "https://graph.microsoft.com/v1.0/deviceManagement/managedDevices?\$filter=complianceState eq 'noncompliant'&\$top=50" \
  -H "Authorization: Bearer ${TOKEN}" | jq '.value[] | {
    device: .deviceName,
    os: .operatingSystem,
    version: .osVersion,
    user: .userPrincipalName,
    lastSync: .lastSyncDateTime
  }'

echo ""
echo "=== Jailbroken/Rooted Devices ==="
curl -s -X GET "https://graph.microsoft.com/v1.0/deviceManagement/managedDevices?\$filter=jailBroken eq 'True'" \
  -H "Authorization: Bearer ${TOKEN}" | jq '.value[] | {
    device: .deviceName,
    platform: .managedDeviceOwnerType,
    user: .userPrincipalName
  }'

import json
import csv
from collections import defaultdict
from datetime import datetime

def track_device_trust_changes(events_file):
    events = json.load(open(events_file))
    device_history = defaultdict(list)
    alerts = []

    for event in events:
        device_id = event.get("deviceId")
        device_history[device_id].append({
            "timestamp": datetime.fromisoformat(event["timestamp"]),
            "complianceState": event.get("complianceState"),
            "trustLevel": event.get("trustLevel"),
            "event": event.get("eventType")
        })

    for device_id, history in device_history.items():
        history.sort(key=lambda x: x["timestamp"])
        for i in range(1, len(history)):
            prev = history[i - 1]
            curr = history[i]
            if prev["complianceState"] == "compliant" and curr["complianceState"] == "noncompliant":
                alerts.append({
                    "device": device_id,
                    "event": "Trust Downgrade",
                    "from": prev["complianceState"],
                    "to": curr["complianceState"],
                    "timestamp": curr["timestamp"].isoformat(),
                    "severity": "HIGH"
                })
            if prev["trustLevel"] == "high" and curr["trustLevel"] in ("low", "none"):
                alerts.append({
                    "device": device_id,
                    "event": "Trust Level Revoked",
                    "from": prev["trustLevel"],
                    "to": curr["trustLevel"],
                    "timestamp": curr["timestamp"].isoformat(),
                    "severity": "CRITICAL"
                })
            time_diff = (curr["timestamp"] - prev["timestamp"]).total_seconds()
            if time_diff < 60 and curr["complianceState"] == "noncompliant":
                alerts.append({
                    "device": device_id,
                    "event": "Rapid Compliance Change",
                    "details": f"Two state changes within {time_diff:.0f}s",
                    "severity": "HIGH"
                })

    return alerts

apiVersion: projectcalico.org/v3
kind: GlobalNetworkPolicy
metadata:
  name: deny-cross-namespace
spec:
  selector: all()
  order: 100
  ingress:
    - action: Allow
      source:
        namespaceSelector: kubernetes.io/metadata.name == "frontend"
      destination:
        namespaceSelector: kubernetes.io/metadata.name == "backend"
      protocol: TCP
      ports:
        - 8443
  egress:
    - action: Allow
      destination:
        namespaceSelector: kubernetes.io/metadata.name == "kube-system"
      protocol: UDP
      ports:
        - 53

# Calico flow log 分析
cat calico_flow_logs.json | jq 'select(.action == "deny") | {
  timestamp: .timestamp,
  src_pod: .meta.src_pod_name,
  src_ns: .meta.src_namespace,
  dst_pod: .meta.dst_pod_name,
  dst_ns: .meta.dst_namespace,
  proto: .meta.protocol,
  dst_port: .meta.dst_port,
  bytes: .bytes
}' | jq -s 'group_by(.src_ns, .dst_ns) | map({
  source_ns: .[0].src_ns,
  target_ns: .[0].dst_ns,
  denied_count: length,
  total_bytes: (map(.bytes) | add),
  unique_src_pods: (map(.src_pod) | unique | length),
  unique_dst_pods: (map(.dst_pod) | unique | length)
})'

# 检测微隔离环境中的端口扫描行为
cat calico_flow_logs.json | jq 'select(.action == "deny")' | \
  jq -s 'group_by(.meta.src_pod_name) | map({
    src_pod: .[0].meta.src_pod_name,
    unique_dst_pods: (map(.meta.dst_pod_name) | unique | length),
    unique_dst_ports: (map(.meta.dst_port) | unique | length),
    total_denied: length
  }) | map(select(.unique_dst_pods > 10 or .unique_dst_ports > 20)) | sort_by(.total_denied) | reverse'

title: 零信任环境 - 微隔离Deny事件异常激增
id: 9c5d3e4f-6071-8901-cdef-345678901234
status: experimental
description: 检测微隔离策略Deny事件在短时间内异常激增，可能指示横向移动扫描
author: x7peeps
date: 2026/07/11
tags:
  - attack.lateral_movement
  - attack.t1046
  - zero_trust
  - micro_segmentation
logsource:
  product: calico
  service: flow_log
detection:
  deny_events:
    action: deny
  timeframe: 5m
  condition: deny_events | count() by meta.src_pod_name > 100
falsepositives:
  - 新部署的服务在收敛过程中的正常Deny事件
level: high

title: 零信任环境 - 跨分段拒绝访问
id: 0d6e4f5a-7182-9012-defa-456789012345
status: experimental
description: 检测跨安全分段的高频拒绝访问事件
author: x7peeps
date: 2026/07/11
tags:
  - attack.lateral_movement
  - attack.t1021
  - zero_trust
logsource:
  product: nsx
  service: distributed_firewall
detection:
  selection:
    action: deny
    direction: inbound
  timeframe: 10m
  condition: selection | count() by source_ip > 50
falsepositives:
  - 应用部署期间的临时通信尝试
level: medium

import json
from collections import defaultdict

def analyze_ztna_sessions(log_file):
    sessions = json.load(open(log_file))
    user_sessions = defaultdict(list)
    anomalies = []

    for session in sessions:
        user = session.get("user")
        user_sessions[user].append(session)

    for user, sessions_list in user_sessions.items():
        sessions_list.sort(key=lambda x: x["startTime"])

        for i in range(1, len(sessions_list)):
            prev = sessions_list[i - 1]
            curr = sessions_list[i]

            prev_end = prev.get("endTime", curr["startTime"])
            if isinstance(prev_end, str):
                from datetime import datetime
                prev_end = datetime.fromisoformat(prev_end)
            curr_start = datetime.fromisoformat(curr["startTime"]) if isinstance(curr["startTime"], str) else curr["startTime"]

            if prev["accessedApp"] == curr["accessedApp"]:
                time_gap = (curr_start - prev_end).total_seconds() if isinstance(prev_end, type(curr_start)) else 0
                if time_gap < 1 and time_gap >= 0:
                    anomalies.append({
                        "user": user,
                        "type": "Rapid Reconnection",
                        "app": curr["accessedApp"],
                        "gap_seconds": time_gap
                    })

            if curr.get("deviceTrustScore", 100) < 50:
                anomalies.append({
                    "user": user,
                    "type": "Low Device Trust",
                    "device": curr.get("deviceName"),
                    "trustScore": curr.get("deviceTrustScore"),
                    "app": curr.get("accessedApp"),
                    "severity": "HIGH"
                })

            if curr.get("geoLocation") != prev.get("geoLocation"):
                anomalies.append({
                    "user": user,
                    "type": "Geographic Change",
                    "from": prev.get("geoLocation"),
                    "to": curr.get("geoLocation"),
                    "severity": "MEDIUM"
                })

    return anomalies

IF (用户 IN 高风险组 OR 位置 IN 未知位置)
  AND (应用 IN [高敏感应用])
  AND (设备.合规状态 == false OR 设备.信任等级 < 中)
THEN
  REJECT 或 REQUIRE_STEP_UP_AUTH
ELSE
  ALLOW

# PAM 日志分析：检测异常特权会话
cat pam_session_logs.json | jq 'select(
  .sessionType == "interactive" and
  (.targetSystem == "database" or .targetSystem == "domain_controller")
) | {
  timestamp: .startTime,
  user: .requestor,
  target: .targetSystem,
  target_account: .targetAccount,
  duration_seconds: (.durationSeconds),
  commands_executed: (.commandCount),
  data_transferred: (.bytesTransferred)
}' | jq -s 'sort_by(.timestamp) | reverse | .[:20]'

title: 零信任环境 - 特权角色批量授予检测
id: 1e7f6a5b-8293-0123-efab-567890123456
status: experimental
description: 检测用户在短时间内被授予多个特权角色，可能指示权限提升攻击
author: x7peeps
date: 2026/07/11
tags:
  - attack.privilege_escalation
  - attack.t1098
  - zero_trust
  - iam
logsource:
  product: azure
  service: auditlogs
detection:
  selection:
    operationName: "Add member to role"
    properties.targetResources[*].type: "Group"
    properties.targetResources[*].displayName|contains:
      - "Global Admin"
      - "Exchange Admin"
      - "SharePoint Admin"
      - "User Admin"
  timeframe: 30m
  condition: selection | count() by initiatedBy.userPrincipalName > 2
falsepositives:
  - 正常的管理员角色分配流程
level: critical

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

class ZeroTrustLogCorrelator:
    def __init__(self):
        self.events = []
        self.correlations = []

    def load_idp_logs(self, filepath):
        with open(filepath) as f:
            for line in f:
                event = json.loads(line)
                self.events.append({
                    "source": "idp",
                    "timestamp": datetime.fromisoformat(event["timestamp"]),
                    "user": event["userPrincipalName"],
                    "ip": event["ipAddress"],
                    "event_type": event["status"].get("errorCode"),
                    "mfa_result": event.get("mfaDetail", {}).get("authResult"),
                    "device_id": event.get("deviceId"),
                    "raw": event
                })

    def load_edr_logs(self, filepath):
        with open(filepath) as f:
            for line in f:
                event = json.loads(line)
                self.events.append({
                    "source": "edr",
                    "timestamp": datetime.fromisoformat(event["timestamp"]),
                    "user": event.get("processUser"),
                    "device_id": event.get("deviceId"),
                    "event_type": event["eventType"],
                    "process_name": event.get("processName"),
                    "process_path": event.get("processPath"),
                    "raw": event
                })

    def load_network_logs(self, filepath):
        with open(filepath) as f:
            for line in f:
                event = json.loads(line)
                self.events.append({
                    "source": "network",
                    "timestamp": datetime.fromisoformat(event["timestamp"]),
                    "src_ip": event["src_ip"],
                    "dst_ip": event["dst_ip"],
                    "dst_port": event["dst_port"],
                    "action": event["action"],
                    "bytes": event.get("bytes", 0),
                    "raw": event
                })

    def correlate_by_user(self, time_window_minutes=30):
        user_events = defaultdict(list)
        for event in self.events:
            if "user" in event and event["user"]:
                user_events[event["user"]].append(event)

        for user, events in user_events.items():
            events.sort(key=lambda x: x["timestamp"])
            for i in range(len(events)):
                window_events = [
                    e for e in events
                    if e["timestamp"] >= events[i]["timestamp"] and
                    e["timestamp"] <= events[i]["timestamp"] + timedelta(minutes=time_window_minutes)
                ]
                sources = set(e["source"] for e in window_events)
                if len(sources) >= 2:
                    self.correlations.append({
                        "user": user,
                        "trigger_event": events[i],
                        "correlated_events": window_events,
                        "sources_involved": list(sources),
                        "time_span_minutes": time_window_minutes
                    })

        return self.correlations

    def detect_anomalous_patterns(self):
        anomalies = []
        user_event_counts = defaultdict(lambda: defaultdict(int))

        for event in self.events:
            if "user" in event and event["user"]:
                key = f"{event['source']}:{event.get('event_type', 'unknown')}"
                user_event_counts[event["user"]][key] += 1

        for user, counts in user_event_counts.items():
            for event_key, count in counts.items():
                if "idp:500126" in event_key and count > 10:
                    anomalies.append({
                        "user": user,
                        "anomaly": "Excessive failed authentications",
                        "count": count,
                        "severity": "HIGH"
                    })
                if "idp:0" in event_key and count > 50:
                    anomalies.append({
                        "user": user,
                        "anomaly": "Excessive successful authentications",
                        "count": count,
                        "severity": "MEDIUM"
                    })

        return anomalies

    def build_timeline(self):
        self.events.sort(key=lambda x: x["timestamp"])
        timeline = []
        for event in self.events:
            timeline.append({
                "time": event["timestamp"].isoformat(),
                "source": event["source"],
                "user": event.get("user", "N/A"),
                "event_type": event.get("event_type", event.get("action", "N/A")),
                "details": {k: v for k, v in event.items()
                           if k not in ("source", "timestamp", "user", "event_type", "raw")}
            })
        return timeline

    def export_timeline(self, output_file):
        timeline = self.build_timeline()
        with open(output_file, "w") as f:
            json.dump(timeline, f, indent=2, default=str)
        print(f"Timeline exported to {output_file} ({len(timeline)} events)")

# 检测设备信任降级后紧接着的身份认证事件
cat device_trust_events.json | jq 'select(.event == "trust_downgrade")' | \
  while read -r event; do
    device_id=$(echo "$event" | jq -r '.deviceId')
    downgrade_time=$(echo "$event" | jq -r '.timestamp')
    echo "Device $device_id trust downgraded at $downgrade_time"
    cat azure_ad_signin.log | jq --arg did "$device_id" --arg dt "$downgrade_time" \
      'select(.deviceId == $did and .timestamp > $dt and .status.errorCode == 0) |
       {user: .userPrincipalName, app: .appDisplayName, time: .timestamp}'
  done

初始入侵 → 身份认证日志（首次异常登录）
    ↓
凭据窃取 → EDR 遥测（Mimikatz/LSASS 访问）
    ↓
设备信任降级 → UEM 日志（安全软件被禁用）
    ↓
横向移动尝试 → 微隔离日志（Deny 事件）
    ↓
横向移动成功 → ZTNA 日志（新应用访问）
    ↓
数据访问 → 应用日志（敏感数据查询）

title: 零信任环境 - 条件访问策略批量变更
id: 2f8a7b6c-9304-1234-fabc-678901234567
status: experimental
description: 检测条件访问策略在短时间内被批量修改，可能指示攻击者试图绕过访问控制
author: x7peeps
date: 2026/07/11
tags:
  - attack.defense_evasion
  - attack.t1562
  - zero_trust
  - conditional_access
logsource:
  product: azure
  service: auditlogs
detection:
  selection:
    operationName:
      - "Update conditional access policy"
      - "Create conditional access policy"
      - "Delete conditional access policy"
  timeframe: 15m
  condition: selection | count() by initiatedBy.userPrincipalName > 3
falsepositives:
  - 合法的策略批量部署
level: critical

title: 零信任环境 - 设备合规状态批量异常变化
id: 3a9b8c7d-0415-2345-abcd-789012345678
status: experimental
description: 检测大量设备在短时间内从合规变为不合规，可能指示攻击者批量破坏设备安全状态
author: x7peeps
date: 2026/07/11
tags:
  - attack.impact
  - attack.t1562.001
  - zero_trust
  - device_compliance
logsource:
  product: intune
  service: device_compliance
detection:
  selection:
    complianceState: "noncompliant"
  timeframe: 1h
  condition: selection | count() > 20
falsepositives:
  - 企业统一推送安全更新导致设备暂时不合规
level: high

title: 零信任环境 - ZTNA异常高并发会话检测
id: 4b0c9d8e-1526-3456-bcde-890123456789
status: experimental
description: 检测同一用户在ZTNA中创建异常高数量的并发会话，可能指示Token盗用或自动化攻击
author: x7peeps
date: 2026/07/11
tags:
  - attack.credential_access
  - attack.t1550.004
  - zero_trust
  - ztna
logsource:
  product: zscaler
  service: ztna_session
detection:
  selection:
    event_type: "session_create"
  timeframe: 5m
  condition: selection | count() by user > 15
falsepositives:
  - 自动化脚本或监控探针的正常轮询
level: high

#!/bin/bash
REPORT_DIR="/tmp/ztna_forensics_$(date +%Y%m%d_%H%M%S)"
mkdir -p "${REPORT_DIR}/identity" "${REPORT_DIR}/device" "${REPORT_DIR}/network"

TENANT_ID="${AZURE_TENANT_ID}"
CLIENT_ID="${AZURE_CLIENT_ID}"
CLIENT_SECRET="${AZURE_CLIENT_SECRET}"
TOKEN=$(curl -s -X POST "https://login.microsoftonline.com/${TENANT_ID}/oauth2/v2.0/token" \
  -d "client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}&scope=https://graph.microsoft.com/.default" \
  | jq -r '.access_token')

echo "[+] Collecting sign-in logs for the last 24 hours..."
SINCE=$(date -u -d '24 hours ago' +%Y-%m-%dT%H:%M:%SZ)
curl -s -X GET "https://graph.microsoft.com/v1.0/auditLogs/signIns?\$filter=createdDateTime ge ${SINCE}&\$top=999" \
  -H "Authorization: Bearer ${TOKEN}" > "${REPORT_DIR}/identity/signin_logs.json"

echo "[+] Analyzing MFA failure patterns..."
jq -s 'group_by(.userPrincipalName) | map({
  user: .[0].userPrincipalName,
  mfa_failures: [.[] | select(.status.errorCode == 500121 or .status.errorCode == 500125)] | length,
  mfa_successes: [.[] | select(.status.errorCode == 0)] | length,
  unique_ips: ([.[].ipAddress] | unique | length),
  unique_cities: ([.[].locationDetails.city] | unique | length)
}) | map(select(.mfa_failures > 5)) | sort_by(.mfa_failures) | reverse' \
  "${REPORT_DIR}/identity/signin_logs.json" > "${REPORT_DIR}/identity/mfa_analysis.json"

echo "[+] Collecting device compliance status..."
curl -s -X GET "https://graph.microsoft.com/v1.0/deviceManagement/managedDevices?\$top=999" \
  -H "Authorization: Bearer ${TOKEN}" > "${REPORT_DIR}/device/managed_devices.json"

jq '{
  total: (.value | length),
  compliant: [.value[] | select(.complianceState == "compliant")] | length,
  noncompliant: [.value[] | select(.complianceState == "noncompliant")] | length,
  not_checked_in_30d: [.value[] | select(
    (.lastSyncDateTime | fromdateiso8601) < (now - 2592000)
  )] | length
}' "${REPORT_DIR}/device/managed_devices.json" > "${REPORT_DIR}/device/compliance_summary.json"

echo "[+] Analyzing conditional access policy changes..."
curl -s -X GET "https://graph.microsoft.com/v1.0/auditLogs/directoryAudits?\$filter=activityDisplayName eq 'Update conditional access policy' or activityDisplayName eq 'Create conditional access policy' or activityDisplayName eq 'Delete conditional access policy'&\$top=100" \
  -H "Authorization: Bearer ${TOKEN}" > "${REPORT_DIR}/identity/ca_policy_changes.json"

echo "[+] Generating forensic report..."
cat > "${REPORT_DIR}/report_summary.json" <<EOF
{
  "report_time": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
  "analysis_period": "Last 24 hours",
  "mfa_failures_flagged": $(jq 'length' "${REPORT_DIR}/identity/mfa_analysis.json"),
  "noncompliant_devices": $(jq '.noncompliant' "${REPORT_DIR}/device/compliance_summary.json"),
  "ca_policy_changes": $(jq 'if .value then .value | length else 0 end' "${REPORT_DIR}/identity/ca_policy_changes.json")
}
EOF

echo "[+] Report generated at: ${REPORT_DIR}"
cat "${REPORT_DIR}/report_summary.json"

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

class ZTNAForensicsAnalyzer:
    def __init__(self):
        self.findings = []
        self.timeline = []

    def analyze_signin_patterns(self, signin_file):
        events = json.load(open(signin_file))
        user_stats = defaultdict(lambda: {
            "total": 0, "success": 0, "failure": 0,
            "ips": set(), "cities": set(), "devices": set(),
            "mfa_failures": 0, "timestamps": []
        })

        for event in events:
            user = event.get("userPrincipalName", "unknown")
            stats = user_stats[user]
            stats["total"] += 1
            stats["timestamps"].append(event.get("createdDateTime"))
            stats["ips"].add(event.get("ipAddress", ""))
            city = event.get("locationDetails", {}).get("city", "")
            if city:
                stats["cities"].add(city)
            device = event.get("deviceId", "")
            if device:
                stats["devices"].add(device)

            error_code = event.get("status", {}).get("errorCode")
            if error_code == 0:
                stats["success"] += 1
            else:
                stats["failure"] += 1
            if error_code in (500121, 500125):
                stats["mfa_failures"] += 1

        for user, stats in user_stats.items():
            if stats["mfa_failures"] > 5:
                self.findings.append({
                    "type": "MFA_FATIGUE_RISK",
                    "severity": "HIGH",
                    "user": user,
                    "mfa_failures": stats["mfa_failures"],
                    "unique_ips": len(stats["ips"]),
                    "unique_cities": len(stats["cities"])
                })
            if len(stats["ips"]) > 10:
                self.findings.append({
                    "type": "IP_DIVERSITY_ANOMALY",
                    "severity": "MEDIUM",
                    "user": user,
                    "unique_ips": len(stats["ips"]),
                    "unique_cities": len(stats["cities"])
                })
            if len(stats["cities"]) > 3:
                self.findings.append({
                    "type": "IMPOSSIBLE_TRAVEL_SUSPECT",
                    "severity": "HIGH",
                    "user": user,
                    "unique_cities": list(stats["cities"])
                })

    def analyze_device_compliance(self, device_file):
        devices = json.load(open(device_file))
        noncompliant = []
        stale_devices = []

        for device in devices.get("value", []):
            if device.get("complianceState") == "noncompliant":
                noncompliant.append({
                    "device": device.get("deviceName"),
                    "user": device.get("userPrincipalName"),
                    "os": device.get("operatingSystem"),
                    "lastSync": device.get("lastSyncDateTime")
                })
            if device.get("lastSyncDateTime"):
                last_sync = datetime.fromisoformat(device["lastSyncDateTime"].replace("Z", "+00:00"))
                if datetime.now(last_sync.tzinfo) - last_sync > timedelta(days=30):
                    stale_devices.append({
                        "device": device.get("deviceName"),
                        "user": device.get("userPrincipalName"),
                        "days_stale": (datetime.now(last_sync.tzinfo) - last_sync).days
                    })

        if len(noncompliant) > 20:
            self.findings.append({
                "type": "BULK_NONCOMPLIANT_DEVICES",
                "severity": "HIGH",
                "count": len(noncompliant),
                "devices": noncompliant[:10]
            })

        if stale_devices:
            self.findings.append({
                "type": "STALE_DEVICES",
                "severity": "MEDIUM",
                "count": len(stale_devices),
                "oldest_days": max(d["days_stale"] for d in stale_devices)
            })

    def generate_report(self):
        report = {
            "analysis_time": datetime.utcnow().isoformat(),
            "total_findings": len(self.findings),
            "findings_by_severity": Counter(f["severity"] for f in self.findings),
            "findings": sorted(self.findings, key=lambda x: {
                "CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3
            }.get(x["severity"], 4))
        }
        return report

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("Usage: python ztna_forensics.py <signin_file> <device_file> [output_file]")
        sys.exit(1)

    analyzer = ZTNAForensicsAnalyzer()
    analyzer.analyze_signin_patterns(sys.argv[1])
    analyzer.analyze_device_compliance(sys.argv[2])
    report = analyzer.generate_report()

    output = json.dumps(report, indent=2, default=str)
    if len(sys.argv) > 3:
        with open(sys.argv[3], "w") as f:
            f.write(output)
        print(f"Report written to {sys.argv[3]}")
    else:
        print(output)