5G核心网与移动通信安全取证深度分析

tcpdump -i any -w signaling_capture.pcap -s 0 "sctp or diameter" -G 3600 -W 24

tshark -r signaling_capture.pcap -Y "diameter.cmd.code == 318" -T fields -e diameter.src.host -e diameter.dest.host -e diameter.experimental-avp.2662044

python3 -c "
import dpkt, struct
with open('ss7_capture.pcap', 'rb') as f:
    pcap = dpkt.pcap.Reader(f)
    for ts, buf in pcap:
        eth = dpkt.ethernet.Ethernet(buf)
        if isinstance(eth.data, dpkt.ip.IP):
            ip = eth.data
            if isinstance(ip.data, dpkt.sctp.SCTP):
                payload = ip.data.data
                print(f'Timestamp: {ts} M3UA payload length: {len(payload)}')
"

title: SS7 Abnormal ProvideSubscriberLocation Request
id: d4e8a2f1-7b3c-4d5e-9f1a-2c3b4e5d6f70
status: experimental
description: Detects abnormal SS7 MAP ProvideSubscriberLocation requests indicating location tracking attacks
references:
  - https://www.3gpp.org/ftp/Specs/archive/29_series/29.002/
  - https://www.cadosecurity.com/blog/telecom-fraud
author: Blue Team Forensics
date: 2026-07-15
modified: 2026-07-15
tags:
  - attack.defense-evasion
  - attack.discovery
  - ss7
  - telecom
  - forensics
logsource:
  category: telecom_signaling
  product: ss7_diameter
detection:
  selection_msg_type:
    message_type: "MAP_ProvideSubscriberLocation_Request"
  selection_condition:
    source_global_title|re: "^[0-9]{7,15}$"
  timeframe: 5m
  condition: selection_msg_type and selection_condition and count() > 5
level: high
falsepositives:
  - Legitimate network optimization procedures
  - Roaming partner queries
  - lawful interception with proper authorization
fields:
  - source_global_title
  - destination_global_title
  - imsi
  - location_estimate
  - timestamp

title: Diameter Abnormal Authentication Vector Request
id: a1b2c3d4-e5f6-7890-abcd-ef1234567890
status: experimental
description: Detects abnormal Diameter S6a Authentication-Request patterns indicating鉴权向量窃取攻击
references:
  - https://tools.ietf.org/html/rfc6733
  - https://www.enisa.europa.eu/publications/5g-security
author: Blue Team Forensics
date: 2026-07-15
modified: 2026-07-15
tags:
  - attack.collection
  - attack.credential-access
  - diameter
  - telecom
  - forensics
logsource:
  category: telecom_signaling
  product: ss7_diameter
detection:
  selection_cmd:
    command_code: 318
    application_id: 16777251
  selection_src:
    origin_host|re: "^.*(proxy|relay|grey).*"
  timeframe: 10m
  condition: selection_cmd and selection_src and count() > 20
level: critical
falsepositives:
  - Large-scale subscriber migration events
  - Network capacity testing
fields:
  - origin_host
  - destination_host
  - imsi
  - request_count
  - timeframe

find /var/log/open5gs/ -name "*.log" -mmin -60 -exec grep -l "oauth_token_error\|mtls_auth_fail\|nf_registration_reject" {} \;

cat /var/log/open5gs/amfd.log | jq 'select(.level == "error" or .level == "warn") | {timestamp: .timestamp, nf: .nf, msg: .message, src: .file}' 2>/dev/null | head -100

grep -E "(401|403|429|500)" /var/log/open5gs/*/http_access.log | awk '{print $1, $4, $7, $9}' | sort | uniq -c | sort -rn | head -30

journalctl -u open5gs-amfd --since "2026-07-15 08:00" --until "2026-07-15 10:00" | grep -iE "(auth_fail|token_invalid|cert_reject|nf_forbidden)" | tee /tmp/5g_auth_anomalies.log

ssh edge-node-01 "docker ps --format '{{.ID}} {{.Names}} {{.Image}}' && docker logs --since 1h $(docker ps -q | head -1) 2>&1 | tail -500" > /tmp/mec_edge取证.log

for node in edge-{01..10}; do echo "=== $node ===" && ssh $node "curl -s http://localhost:8080/mec_appstore/v1/applications | jq '.[].appId, .[].appName' 2>/dev/null" && echo; done > /tmp/mec_inventory.log

python3 -c "
import subprocess, re, json

result = subprocess.run(['iwlist', 'scan'], capture_output=True, text=True)
cells = re.findall(r'Cell \d+ - Address: ([\w:]+).*?Frequency:([\d.]+).*?Signal level:(-?\d+)', result.stdout, re.DOTALL)

alerts = []
for bssid, freq, signal in cells:
    if int(signal) > -30:
        alerts.append({'bssid': bssid, 'frequency': freq, 'signal': signal, 'severity': 'HIGH'})

with open('/tmp/rf_anomalies.json', 'w') as f:
    json.dump(alerts, f, indent=2)
"

#!/bin/bash
LOG_DIR="/var/log/5gc/upf"
ALERT_THRESHOLD=1000
WINDOW=300

find "$LOG_DIR" -name "*.log" -mmin -5 | while read logfile; do
  top_ips=$(awk -v threshold="$ALERT_THRESHOLD" '
    BEGIN {OFS=","}
    /data_plane_packet/ {
      split($0, a, "src=");
      split(a[2], b, " ");
      src_ip = b[1];
      count[src_ip]++;
    }
    END {
      for (ip in count) {
        if (count[ip] > threshold) print ip, count[ip]
      }
    }' "$logfile")
  if [ -n "$top_ips" ]; then
    echo "[ALERT] $(date +%Y-%m-%dT%H:%M:%S%z) High volume detected: $top_ips" >> /tmp/iot_anomaly_alerts.log
  fi
done

cat /var/log/5gc/upf/*.log | grep "data_plane_packet" | awk '{
  for(i=1;i<=NF;i++) {
    if($i ~ /src=/) {
      split($i, a, "=");
      src_ip = a[2];
      gsub(/[^0-9.]/, "", src_ip);
      count[src_ip]++;
    }
  }
} END {
  for (ip in count) {
    if (count[ip] > 500) printf "%s: %d packets\n", ip, count[ip]
  }
}' | sort -t: -k2 -rn | head -20

title: SS7 Abnormal Location Tracking Request Storm
id: b7f2e4a1-8c3d-4e5f-a6b7-c8d9e0f1a2b3
status: stable
description: Detects SS7 MAP AnyTimeInterrogation/ProvideSubscriberLocation request storms indicating mass location tracking attacks
references:
  - https://www.3gpp.org/DynaReport/29002.htm
  - https://media.defense.gov/2023/Apr/27/2003205891/-1/-1/0/CSI-DETECTING-AND-PREVENTING-SS7-ATTACKS.PDF
author: Blue Team Forensics
date: 2026-07-15
modified: 2026-07-15
tags:
  - attack.discovery
  - attack.collection
  - ss7
  - telecom
  - location_tracking
logsource:
  category: telecom_signaling
  product: ss7
detection:
  selection_ati:
    message_type: "MAP_AnyTimeInterrogation_Request"
  selection_psl:
    message_type: "MAP_ProvideSubscriberLocation_Request"
  selection_spfi:
    message_type: "MAP_SendParametersForSubscriberInfo_Request"
  filter_known_sources:
    source_global_title|startswith: "00101"
    source_global_title|endswith: "001"
  condition: (selection_ati or selection_psl or selection_spfi) and not filter_known_sources and count() > 10 within 5m
level: critical
falsepositives:
  - Authorized lawful interception
  - Roaming partner location query during inter-network testing
fields:
  - source_global_title
  - destination_global_title
  - imsi_prefix
  - request_count
  - unique_targets
  - time_window

title: IMSI Catcher RF Anomaly Detection
id: c9d1e3f5-a7b8-4c0d-e2f4-a6b8c0d2e4f6
status: experimental
description: Detects RF-level anomalies indicating possible IMSI catcher or false base station presence
references:
  - https://ieeexplore.ieee.org/document/8408509
  - https://arxiv.org/abs/2003.01521
author: Blue Team Forensics
date: 2026-07-15
modified: 2026-07-15
tags:
  - attack.defense-evasion
  - attack.discovery
  - imsi_catcher
  - false_base_station
  - rf_monitoring
logsource:
  category: rf_monitoring
  product: imsi_catcher_detection
detection:
  selection_signal_anomaly:
    signal_delta_db|gt: 20
    cell_reselection_count|gt: 5
  selection_frequency_anomaly:
    frequency_offset_khz|gt: 0.5
    timing_advance_jump|gt: 50
  selection_behavior_anomaly:
    encryption_disabled: true
    band_reselection_to_2g: true
  condition: selection_signal_anomaly or selection_frequency_anomaly or selection_behavior_anomaly
level: high
falsepositives:
  - Normal cell handover during high-speed movement
  - Network maintenance with temporary configuration changes
  - Environmental RF interference
fields:
  - serving_cell_id
  - signal_strength_dbm
  - frequency_offset_khz
  - timing_advance
  - band_indicator
  - encryption_status
  - timestamp
  - location

#!/bin/bash
LOG_BASE="/var/log/open5gs"
OUTPUT_DIR="/tmp/5g_forensics_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "=== 5G Core Network Log Forensics ===" > "$OUTPUT_DIR/analysis_report.txt"
echo "Analysis time: $(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$OUTPUT_DIR/analysis_report.txt"
echo "" >> "$OUTPUT_DIR/analysis_report.txt"

echo "--- Authentication Anomalies ---" >> "$OUTPUT_DIR/analysis_report.txt"
find "$LOG_BASE" -name "*.log" -mmin -60 | while read logf; do
  grep -c "authentication_failure\|auth_reject\|auth_timeout" "$logf" 2>/dev/null | while read cnt; do
    if [ "$cnt" -gt 10 ]; then
      echo "[WARN] $logf has $cnt auth failures" >> "$OUTPUT_DIR/analysis_report.txt"
      grep "authentication_failure\|auth_reject\|auth_timeout" "$logf" | head -20 >> "$OUTPUT_DIR/analysis_report.txt"
    fi
  done
done

echo "" >> "$OUTPUT_DIR/analysis_report.txt"
echo "--- NF Registration Anomalies ---" >> "$OUTPUT_DIR/analysis_report.txt"
find "$LOG_BASE" -name "*.log" -mmin -60 | xargs grep "nf_registration" 2>/dev/null | awk '{
  for(i=1;i<=NF;i++) {
    if($i ~ /nf_type=/) {
      nf[$i]++;
    }
  }
} END {
  for (k in nf) {
    if (nf[k] > 50) print "[HIGH] " k " count=" nf[k]
    else if (nf[k] > 20) print "[MED] " k " count=" nf[k]
  }
}' >> "$OUTPUT_DIR/analysis_report.txt"

echo "" >> "$OUTPUT_DIR/analysis_report.txt"
echo "--- API Error Rate Analysis ---" >> "$OUTPUT_DIR/analysis_report.txt"
find "$LOG_BASE" -name "*.log" -mmin -30 | xargs grep -E "status.*[45][0-9][0-9]" 2>/dev/null | awk -F'"' '{
  split($2, a, " ");
  code = a[1];
  codes[code]++;
} END {
  for (c in codes) {
    printf "HTTP %s: %d occurrences\n", c, codes[c]
  }
}' | sort -t: -k2 -rn >> "$OUTPUT_DIR/analysis_report.txt"

echo "" >> "$OUTPUT_DIR/analysis_report.txt"
echo "--- Session Management Anomalies ---" >> "$OUTPUT_DIR/analysis_report.txt"
find "$LOG_BASE" -name "*.log" -mmin -60 | xargs grep "pdu_session" 2>/dev/null | grep -iE "error|fail|reject|timeout" | awk '{
  for(i=1;i<=NF;i++) {
    if($i ~ /smf_id=/) smf[$i]++;
    if($i ~ /cause=/) cause[$i]++;
  }
} END {
  print "By SMF:"
  for (s in smf) printf "  %s: %d\n", s, smf[s]
  print "By Cause:"
  for (c in cause) printf "  %s: %d\n", c, cause[c]
}' >> "$OUTPUT_DIR/analysis_report.txt"

echo "" >> "$OUTPUT_DIR/analysis_report.txt"
echo "--- Slice Selection Anomalies ---" >> "$OUTPUT_DIR/analysis_report.txt"
find "$LOG_BASE" -name "*.log" -mmin -120 | xargs grep "nssf" 2>/dev/null | grep -iE "slice_selection|nssai" | awk '{
  for(i=1;i<=NF;i++) {
    if($i ~ /sst=/) sst[$i]++;
  }
} END {
  for (s in sst) printf "SST %s: %d selections\n", s, sst[s]
}' >> "$OUTPUT_DIR/analysis_report.txt"

echo "Report generated: $OUTPUT_DIR/analysis_report.txt"
wc -l "$OUTPUT_DIR/analysis_report.txt"

import sys
import struct
import json
import time
from collections import defaultdict, Counter
from datetime import datetime

class DiameterMessageParser:
    COMMAND_CODES = {
        274: "Abort-Session",
        275: "Accounting",
        280: "Capabilities-Exchange",
        282: "Re-Auth",
        285: "Query-User",
        291: "Update-Location",
        300: "Subscription-Information",
        303: "Push-Profile",
        316: "Server-Assignment",
        318: "Authentication-Information",
    }

    AVP_CODES = {
        1: "Session-Id",
        263: "Origin-Host",
        264: "Destination-Host",
        268: "Auth-Application-Id",
        277: "Destination-Realm",
        283: "User-Name",
        701: "Subscription-Data",
        1406: "Non-3GPP-Subscriber-Data",
        1425: "UICC-Access-Application-Data",
    }

    def __init__(self):
        self.messages = []
        self.anomalies = []
        self.stats = defaultdict(int)

    def parse_header(self, data):
        if len(data) < 20:
            return None
        version = data[0]
        msg_length = struct.unpack("!I", b'\x00' + data[1:4])[0]
        command_flags = data[4]
        command_code = struct.unpack("!I", data[5:9])[0]
        application_id = struct.unpack("!I", data[9:13])[0]
        hop_by_hop = struct.unpack("!I", data[13:17])[0]
        end_to_end = struct.unpack("!I", data[17:21])[0]
        return {
            "version": version,
            "length": msg_length,
            "flags": command_flags,
            "command_code": command_code,
            "application_id": application_id,
            "hop_by_hop_id": hop_by_hop,
            "end_to_end_id": end_to_end,
            "is_request": bool(command_flags & 0x80),
        }

    def parse_avps(self, data, offset):
        avps = {}
        while offset < len(data) - 4:
            avp_code = struct.unpack("!I", data[offset:offset+4])[0]
            avp_flags = data[offset+4]
            avp_length = struct.unpack("!I", b'\x00' + data[offset+5:offset+8])[0]
            if avp_length < 8 or offset + avp_length > len(data):
                break
            padded_length = (avp_length + 3) & ~3
            if avp_code in self.AVP_CODES and avp_length > 8:
                avp_value = data[offset+8:offset+avp_length]
                try:
                    avps[self.AVP_CODES[avp_code]] = avp_value.decode('utf-8', errors='replace')
                except Exception:
                    avps[self.AVP_CODES[avp_code]] = avp_value.hex()
            offset += padded_length
        return avps

    def analyze_message(self, header, avps, src_ip, dst_ip, timestamp):
        cmd_name = self.COMMAND_CODES.get(header["command_code"], f"Unknown({header['command_code']})")
        self.stats[cmd_name] += 1

        if header["command_code"] == 318 and header["is_request"]:
            origin = avps.get("Origin-Host", "unknown")
            target = avps.get("User-Name", "unknown")
            self.messages.append({
                "timestamp": timestamp,
                "type": "AIR",
                "command_code": 318,
                "origin_host": origin,
                "imsi": target,
                "src_ip": src_ip,
                "dst_ip": dst_ip,
            })

        if header["command_code"] == 318:
            key = f"{avps.get('Origin-Host', 'unknown')}:{avps.get('User-Name', 'unknown')}"
            self.stats[f"air_{key}"] += 1

    def detect_anomalies(self, window_seconds=300):
        now = time.time()
        recent_messages = [m for m in self.messages if now - m.get("ts_epoch", now) < window_seconds]

        origin_counts = Counter(m.get("origin_host", "unknown") for m in recent_messages)
        imsi_counts = Counter(m.get("imsi", "unknown") for m in recent_messages)

        for origin, count in origin_counts.items():
            if count > 20:
                self.anomalies.append({
                    "type": "HIGH_AUTH_REQUEST_RATE",
                    "severity": "HIGH",
                    "origin_host": origin,
                    "request_count": count,
                    "window_seconds": window_seconds,
                    "description": f"Excessive Authentication-Information-Request from {origin}",
                })

        if len(set(m.get("imsi", "") for m in recent_messages)) > 50:
            self.anomalies.append({
                "type": "MULTI_SUBSCRIBER_QUERY",
                "severity": "MEDIUM",
                "unique_imsi_count": len(set(m.get("imsi", "") for m in recent_messages)),
                "description": "Unusually large number of unique IMSIs queried in short window",
            })

        return self.anomalies


def parse_pcap_diameter(filepath, max_packets=10000):
    parser = DiameterMessageParser()
    packet_count = 0
    with open(filepath, 'rb') as f:
        magic = f.read(4)
        if magic != b'\xd4\xc3\xb2\xa1':
            print("Error: Not a valid pcap file (big-endian)")
            f.seek(0)
            magic = f.read(4)
            if magic != b'\xa1\xb2\xc3\xd4':
                print("Error: Invalid pcap format")
                return parser

        f.read(4)
        f.read(4)
        f.read(4)
        f.read(4)

        while packet_count < max_packets:
            ts_sec = struct.unpack("!I", f.read(4))[0]
            ts_usec = struct.unpack("!I", f.read(4))[0]
            incl_len = struct.unpack("!I", f.read(4))[0]
            orig_len = struct.unpack("!I", f.read(4))[0]
            if incl_len == 0 or incl_len > 65535:
                break
            packet_data = f.read(incl_len)
            if len(packet_data) < incl_len:
                break
            packet_count += 1

            timestamp = datetime.fromtimestamp(ts_sec + ts_usec / 1000000).isoformat()

            try:
                eth_header_len = 14
                ip_header_start = eth_header_len
                ip_header = packet_data[ip_header_start:ip_header_start+20]
                ip_proto = ip_header[9]
                src_ip = ".".join(str(b) for b in ip_header[12:16])
                dst_ip = ".".join(str(b) for b in ip_header[16:20])

                if ip_proto == 17:
                    udp_header_start = ip_header_start + 20
                    udp_header = packet_data[udp_header_start:udp_header_start+8]
                    src_port = struct.unpack("!H", udp_header[0:2])[0]
                    dst_port = struct.unpack("!H", udp_header[2:4])[0]

                    if src_port == 3868 or dst_port == 3868:
                        diameter_start = udp_header_start + 8
                        diameter_data = packet_data[diameter_start:]
                        if len(diameter_data) >= 20:
                            header = parser.parse_header(diameter_data)
                            if header:
                                avps = parser.parse_avps(diameter_data, 20)
                                header["ts_epoch"] = ts_sec
                                parser.analyze_message(header, avps, src_ip, dst_ip, timestamp)
            except Exception:
                continue

    anomalies = parser.detect_anomalies()
    return parser


def print_report(parser):
    print("=" * 70)
    print("5G DIAMETER SIGNALING FORENSICS REPORT")
    print("=" * 70)
    print(f"Generated: {datetime.now().isoformat()}")
    print(f"Total Diameter messages analyzed: {sum(v for k, v in parser.stats.items() if not k.startswith('air_'))}")
    print()

    print("--- Command Code Distribution ---")
    for cmd, count in sorted(parser.stats.items(), key=lambda x: -x[1]):
        if not cmd.startswith("air_"):
            print(f"  {cmd}: {count}")
    print()

    print("--- Authentication-Information-Request by Origin ---")
    air_stats = {k: v for k, v in parser.stats.items() if k.startswith("air_")}
    for key, count in sorted(air_stats.items(), key=lambda x: -x[1]):
        print(f"  {key}: {count}")
    print()

    if parser.anomalies:
        print("--- DETECTED ANOMALIES ---")
        for a in parser.anomalies:
            print(f"  [{a['severity']}] {a['type']}: {a['description']}")
            if "request_count" in a:
                print(f"    Request count: {a['request_count']}")
            if "unique_imsi_count" in a:
                print(f"    Unique IMSIs: {a['unique_imsi_count']}")
    else:
        print("--- No anomalies detected in current analysis window ---")
    print()


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <diameter_pcap_file>")
        sys.exit(1)

    pcap_file = sys.argv[1]
    print(f"Analyzing Diameter signaling from: {pcap_file}")
    parser = parse_pcap_diameter(pcap_file)
    print_report(parser)