社会工程学攻击取证深度分析

MITRE_ATT&CK_Social_Engineering_Mapping:
  Initial_Access:
    - T1566.001: Spearphishing Attachment（鱼叉式钓鱼附件）
    - T1566.002: Spearphishing Link（鱼叉式钓鱼链接）
    - T1566.003: Spearphishing via Service（通过第三方服务钓鱼）
    - T1566.004: Spearphishing Voice Call（鱼叉式语音通话）
  Credential_Access:
    - T1598.001: Phishing for Credential - Spearphishing Service
    - T1598.002: Spearphishing Attachment (Credential Harvesting)
    - T1598.003: Spearphishing Link (Credential Harvesting)
  Collection:
    - T1534: Internal Spearphishing（内部鱼叉式钓鱼）
  Reconnaissance:
    - T1592: Gather Victim Host Information
    - T1593: Search Open Websites/Databases
    - T1594: Search Victim-Owned Websites
    - T1595: Active Scanning
    - T1596: Search Open Technical Databases
    - T1597: Search Closed Sources
  Resource_Development:
    - T1583.006: Acquire Infrastructure - Web Services
    - T1586: Compromise Accounts
    - T1586.002: Email Accounts
    - T1650: Acquire Access

FORENSIC_FRAMEWORK = {
    "Phase_1_Timeline_Reconstruction": {
        "goal": "重建完整攻击时间线",
        "steps": [
            "收集所有相关日志源（邮件头/服务器日志/终端日志/网络流量）",
            "统一时间戳为UTC并标注时区偏移",
            "按时间排序构建事件序列",
            "识别时间窗口内的异常行为节点"
        ]
    },
    "Phase_2_Infrastructure_Correlation": {
        "goal": "关联攻击基础设施组件",
        "steps": [
            "提取所有可疑域名/IP/URL",
            "WHOIS + DNS + SSL证书交叉查询",
            "被动DNS数据库验证历史解析关系",
            "构建基础设施拓扑图"
        ]
    },
    "Phase_3_Psychological_Profiling": {
        "goal": "构建攻击者心理画像",
        "steps": [
            "分析话术模板语言特征",
            "识别文化背景暗示（节日/时区/语言习惯）",
            "评估攻击定制化程度",
            "推断攻击者组织能力层级"
        ]
    },
    "Phase_4_Evidence_Gradation": {
        "goal": "证据强度分层评估",
        "levels": [
            "确认恶意（Confirmed Malicious）: 可直接证明恶意意图的证据",
            "高度可疑（Highly Suspicious）: 强烈暗示恶意但需佐证的证据",
            "需要关注（Requires Attention）: 存在异常但无法直接归因的指标"
        ]
    }
}

完整邮件头示例分析：

Return-Path: <no-reply@micros0ft-security.com>
Delivered-To: target@company.com
Received: from mail.company.com (10.0.1.25) by exchange.company.com
    with SMTP; Mon, 15 Jan 2026 08:23:41 +0800
Received: from mx-relay.external.net (203.0.113.50) by mail.company.com
    with ESMTPS; Mon, 15 Jan 2026 08:23:39 +0800
Received: from [198.51.100.23] (helo=sender.campaign-server.io)
    by mx-relay.external.net with SMTP; Mon, 15 Jan 2026 00:23:35 +0000
From: "Microsoft Security Team" <no-reply@micros0ft-security.com>
Reply-To: security-alerts@ms-support-verify.com
Subject: Urgent: Your Account Requires Immediate Verification
X-Mailer: PHPMailer v6.8.0
X-Originating-IP: [198.51.100.23]

import email
from datetime import datetime
import re

def parse_email_headers(raw_headers):
    msg = email.message_from_string(raw_headers)
    received_headers = msg.get_all('Received')
    
    hops = []
    for header in reversed(received_headers):
        timestamp_match = re.search(r';\s*(.+)$', header)
        if timestamp_match:
            time_str = timestamp_match.group(1).strip()
            try:
                parsed_time = email.utils.parsedate_to_datetime(time_str)
                hops.append({
                    'hop_raw': header.split(';')[0].strip(),
                    'timestamp': parsed_time,
                    'timezone': parsed_time.tzinfo
                })
            except Exception:
                continue
    
    analysis = []
    for i in range(1, len(hops)):
        delay = (hops[i]['timestamp'] - hops[i-1]['timestamp']).total_seconds()
        analysis.append({
            'from_hop': i-1,
            'to_hop': i,
            'delay_seconds': delay,
            'anomaly': True if abs(delay) > 30 or delay < 0 else False
        })
    
    return analysis

使用多种工具交叉验证域名注册信息：

whois micros0ft-security.com
Domain Name: MICROS0FT-SECURITY.COM
Registry Domain ID: 2845671234_DOMAIN_COM-vrsn
Registrar WHOIS Server: whois.namecheap.com
Registrar: NameCheap, Inc.
Creation Date: 2025-12-28T09:15:00Z
Updated Date: 2025-12-28T09:15:00Z
Expiry Date: 2027-12-28T09:15:00Z

Registrant Organization: PrivacyProtect.org
Registrant State/Province: MA
Registrant Country: US

dig micros0ft-security.com ANY +noall +answer

micros0ft-security.com. 300 IN A     198.51.100.23
micros0ft-security.com. 300 IN MX 10 mail.micros0ft-security.com.
micros0ft-security.com. 300 IN TXT  "v=spf1 include:_spf.mailgun.org ~all"
mail._domainkey.micros0ft-security.com. 300 IN TXT "v=DKIM1; k=rsa; p=MIGfMA0GCSq..."

dig micros0ft-security.com NS +short
ns1.suspended-domain.com.
ns2.suspended-domain.com.

dig _dmarc.micros0ft-security.com TXT +short
"v=DMARC1; p=none; rua=mailto:dmarc@micros0ft-security.com"

import dns.resolver
from itertools import permutations

def generate_typosquat_variants(domain):
    name, tld = domain.rsplit('.', 1)
    variants = set()
    
    for i in range(len(name)):
        variants.add(name[:i] + name[i+1:])
    
    char_substitutions = {
        'o': '0', '0': 'o', 'l': '1', '1': 'l',
        'e': '3', 'a': '@', 's': 'z', 'rn': 'm',
        'cl': 'd', 'vv': 'w'
    }
    for original, replacement in char_substitutions.items():
        if original in name:
            variants.add(name.replace(original, replacement))
    
    for i in range(len(name) - 1):
        variants.add(name[:i] + name[i+1:i+2] + name[i:i+1] + name[i+2:])
    
    for sep in ['-', '.']:
        for i in range(1, len(name)):
            variants.add(name[:i] + sep + name[i:])
    
    homoglyph_map = {
        'a': 'а', 'c': 'ϲ', 'e': 'е', 'o': 'ο',
        'p': 'р', 's': 'ѕ', 'x': 'х'
    }
    for orig, hglyph in homoglyph_map.items():
        if orig in name:
            variants.add(name.replace(orig, hglyph))
    
    return [v + '.' + tld for v in variants]

def check_domain_resolved(domains):
    resolved = {}
    resolver = dns.resolver.Resolver()
    resolver.nameservers = ['8.8.8.8']
    resolver.lifetime = 3
    
    for domain in domains:
        try:
            answers = resolver.resolve(domain, 'A')
            ips = [rdata.address for rdata in answers]
            resolved[domain] = ips
        except Exception:
            resolved[domain] = None
    return resolved

from bs4 import BeautifulSoup
import hashlib

def extract_template_fingerprint(html_content):
    soup = BeautifulSoup(html_content, 'html.parser')
    
    structure_hash = hashlib.sha256()
    for tag in soup.descendants:
        if tag.name:
            structure_hash.update(tag.name.encode())
    fingerprint = structure_hash.hexdigest()[:16]
    
    styles = soup.find_all(['style', 'link'], rel='stylesheet')
    style_content = ''.join(s.get_text() for s in soup.find_all('style'))
    style_hash = hashlib.md5(style_content.encode()).hexdigest()[:12]
    
    images = [img.get('src', '') for img in soup.find_all('img')]
    tracking_pixels = [src for src in images if 'track' in src.lower() or 'pixel' in src.lower()]
    
    links = [a.get('href', '') for a in soup.find_all('a')]
    suspicious_links = [l for l in links if l.startswith('http') and 'target' not in str(soup.find('a', href=l))]
    
    return {
        'structure_fingerprint': fingerprint,
        'style_fingerprint': style_hash,
        'tracking_pixels': tracking_pixels,
        'suspicious_links': suspicious_links,
        'total_images': len(images),
        'total_links': len(links),
        'form_count': len(soup.find_all('form')),
        'iframe_count': len(soup.find_all('iframe'))
    }

exiftool phishing_logo.png

ExifTool Version Number         : 12.70
File Name                       : phishing_logo.png
File Size                       : 45 KB
File Modification Date/Time     : 2025:12:20 14:32:18+08:00
File Type                       : PNG
Image Width                     : 200
Image Height                    : 56
Software                        : Adobe Photoshop 25.3
Creator Tool                    : Canva
Color Space                     : sRGB

import requests
from bs4 import BeautifulSoup
import re

def analyze_landing_page(url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    response = requests.get(url, headers=headers, allow_redirects=False, timeout=15)
    
    redirect_chain = [url]
    while response.is_redirect or response.status_code in [301, 302, 303, 307, 308]:
        next_url = response.headers.get('Location')
        redirect_chain.append(next_url)
        response = requests.get(next_url, headers=headers, allow_redirects=False, timeout=15)
    
    final_html = response.text
    soup = BeautifulSoup(final_html, 'html.parser')
    
    forms = []
    for form in soup.find_all('form'):
        form_data = {
            'action': form.get('action', ''),
            'method': form.get('method', 'GET').upper(),
            'fields': [],
            'javascript_submit': bool(re.search(r'\.submit\(\)|\.click\(\)', final_html))
        }
        for inp in form.find_all(['input', 'textarea', 'select']):
            form_data['fields'].append({
                'name': inp.get('name', ''),
                'type': inp.get('type', 'text'),
                'id': inp.get('id', ''),
                'placeholder': inp.get('placeholder', ''),
                'autocomplete': inp.get('autocomplete', '')
            })
        forms.append(form_data)
    
    scripts = []
    for script in soup.find_all('script'):
        src = script.get('src', '')
        if src:
            scripts.append({'external': src})
        elif script.string:
            scripts.append({'inline': script.string[:500]})
    
    meta_tags = {}
    for meta in soup.find_all('meta'):
        name = meta.get('name', meta.get('property', ''))
        content = meta.get('content', '')
        if name:
            meta_tags[name] = content
    
    return {
        'final_url': response.url,
        'redirect_chain': redirect_chain,
        'status_code': response.status_code,
        'server_header': response.headers.get('Server', ''),
        'forms': forms,
        'scripts': scripts,
        'meta_tags': meta_tags,
        'tls_info': {
            'issuer': str(response.raw.connection.sock.getpeercert().get('issuer', '')),
            'subject': str(response.raw.connection.sock.getpeercert().get('subject', ''))
        } if hasattr(response.raw, 'connection') and response.raw.connection else None
    }

curl -s "https://archive.org/wayback/available?url=suspicious-site.com" | python3 -m json.tool

{
    "archived_snapshots": {
        "closest": {
            "available": true,
            "url": "https://web.archive.org/web/20260110/https://suspicious-site.com",
            "timestamp": "20260110",
            "status": "200"
        }
    }
}

检查页面是否曾经存在合法内容后被替换为钓鱼页面：
curl -s "https://web.archive.org/web/*/suspicious-site.com/*" | grep -oP 'datetime="[^"]*"'

#!/usr/bin/env python3
import email
import re
import json
import hashlib
from datetime import datetime, timezone
import dns.resolver
import requests

class PhishingEmailAnalyzer:
    def __init__(self):
        self.resolver = dns.resolver.Resolver()
        self.resolver.nameservers = ['8.8.8.8', '1.1.1.1']
        self.resolver.lifetime = 5
        self.extracted_iocs = {
            'domains': set(),
            'ips': set(),
            'urls': set(),
            'emails': set(),
            'hashes': set()
        }
    
    def parse_full_headers(self, raw_message):
        msg = email.message_from_string(raw_message)
        
        auth_results = {}
        for header_name in ['Authentication-Results', 'X-Microsoft-Antispam', 
                            'X-SES-Outgoing', 'X-Original-Sender']:
            value = msg.get(header_name)
            if value:
                auth_results[header_name] = value
        
        received_chain = []
        for header in msg.get_all('Received', []):
            received_chain.append(header)
        
        hop_details = []
        for i, hop in enumerate(reversed(received_chain)):
            ip_match = re.findall(r'\[(\d+\.\d+\.\d+\.\d+)\]', hop)
            host_match = re.search(r'from\s+(\S+)', hop)
            time_match = re.search(r';\s*(.+)$', hop)
            
            hop_info = {
                'sequence': i + 1,
                'raw': hop.strip(),
                'ips': ip_match,
                'hostname': host_match.group(1) if host_match else '',
                'timestamp': time_match.group(1).strip() if time_match else ''
            }
            hop_details.append(hop_info)
            for ip in ip_match:
                self.extracted_iocs['ips'].add(ip)
        
        sender_analysis = {
            'from': msg.get('From', ''),
            'reply_to': msg.get('Reply-To', ''),
            'return_path': msg.get('Return-Path', ''),
            'envelope_from_mismatch': msg.get('Reply-To', '') != msg.get('From', '')
        }
        
        url_pattern = re.compile(r'https?://[^\s<>"\'`)]+')
        email_pattern = re.compile(r'[\w.+-]+@[\w-]+\.[\w.-]+')
        
        body = ''
        if msg.is_multipart():
            for part in msg.walk():
                ct = part.get_content_type()
                if ct in ['text/html', 'text/plain']:
                    body += part.get_payload(decode=True).decode('utf-8', errors='ignore')
        else:
            body = msg.get_payload(decode=True).decode('utf-8', errors='ignore')
        
        for match in url_pattern.finditer(body):
            self.extracted_iocs['urls'].add(match.group(0))
        for match in email_pattern.finditer(body):
            self.extracted_iocs['emails'].add(match.group(0))
        
        display_urls = []
        anchor_pattern = re.compile(r'<a[^>]*href=["\']([^"\']+)["\'][^>]*>(.*?)</a>', re.DOTALL)
        for match in anchor_pattern.finditer(body):
            real_url = match.group(1)
            display_text = re.sub(r'<[^>]+>', '', match.group(2)).strip()
            if display_text and 'http' in display_text:
                display_urls.append({
                    'display_text': display_text,
                    'real_url': real_url,
                    'mismatch': display_text != real_url
                })
        
        return {
            'authentication': auth_results,
            'hop_chain': hop_details,
            'sender': sender_analysis,
            'display_url_mismatches': display_urls,
            'iocs': {k: list(v) for k, v in self.extracted_iocs.items()}
        }
    
    def enrich_domains(self):
        enrichment = {}
        for domain in list(self.extracted_iocs['domains']):
            info = {'domain': domain}
            
            try:
                whois_resp = requests.get(f'https://www.whoisxmlapi.com/whoisserver/WhoisService?domain={domain}&outputFormat=json', timeout=10)
                info['whois_available'] = whois_resp.status_code == 200
            except Exception:
                info['whois_available'] = False
            
            try:
                ns_records = self.resolver.resolve(domain, 'NS')
                info['nameservers'] = [str(r) for r in ns_records]
            except Exception:
                info['nameservers'] = []
            
            try:
                mx_records = self.resolver.resolve(domain, 'MX')
                info['mx_records'] = [(str(r.exchange), r.preference) for r in mx_records]
            except Exception:
                info['mx_records'] = []
            
            try:
                spf_record = self.resolver.resolve(domain, 'TXT')
                spf_texts = [str(r).strip('"') for r in spf_record]
                info['spf'] = [t for t in spf_texts if 'v=spf1' in t]
            except Exception:
                info['spf'] = []
            
            enrichment[domain] = info
        return enrichment
    
    def generate_report(self, header_analysis, domain_enrichment):
        report = {
            'analysis_timestamp': datetime.now(timezone.utc).isoformat(),
            'email_authentication': header_analysis['authentication'],
            'routing_chain': header_analysis['hop_chain'],
            'sender_analysis': header_analysis['sender'],
            'display_url_deception': header_analysis['display_url_mismatches'],
            'infrastructure_enrichment': domain_enrichment,
            'extracted_iocs': header_analysis['iocs'],
            'risk_indicators': [],
            'attribution_hints': []
        }
        
        if header_analysis['sender']['envelope_from_mismatch']:
            report['risk_indicators'].append('Reply-To 与 From 地址不一致')
        
        for mismatch in header_analysis['display_url_mismatches']:
            if mismatch['mismatch']:
                report['risk_indicators'].append(f'URL显示欺骗: 显示"{mismatch["display_text"]}" → 实际"{mismatch["real_url"]}"')
        
        for domain, info in domain_enrichment.items():
            if info.get('nameservers'):
                for ns in info['nameservers']:
                    if any(suspect in ns.lower() for suspect in ['suspended', 'parked', 'expired']):
                        report['risk_indicators'].append(f'{domain} 使用可疑NS: {ns}')
        
        return report

PowerShell 检查 O365 邮箱异常规则：

Get-InboxRule -Mailbox victim@company.com | Select-Object Name, Enabled, Description, Priority | Format-List

异常规则特征：
Name        : "Process Incoming Mail"（伪装正常名称）
Enabled     : True
Description : "将来自特定发件人的邮件标记为已读并移至RSS Feeds文件夹"
              "同时将副本转发至 external@gmail.com"
Priority    : -1（最高优先级，确保最先执行）

查找隐藏转发规则的命令：
Get-InboxRule -Mailbox victim@company.com | Where-Object {$_.ForwardTo -ne $null -or $_.RedirectTo -ne $null} | Format-List *

Azure AD Sign-in Logs 异常查询：

SigninLogs
| where UserPrincipalName == "victim@company.com"
| project TimeGenerated, IPAddress, Location, DeviceDetail, ClientAppUsed, ConditionalAccessStatus, ResultDescription
| order by TimeGenerated desc

异常检测维度：
- Impossible Travel: 短时间内从两个地理上不可达的位置登录
- 新 IP/国家首次出现
- 异常设备指纹（之前从未出现的浏览器/OS组合）
- 非工作时间登录模式突变
- ConditionalAccessStatus 变化（绕过MFA尝试）

import msal
import requests
from datetime import datetime

class O365TokenAuditor:
    def __init__(self, tenant_id, client_id, client_secret):
        self.authority = f"https://login.microsoftonline.com/{tenant_id}"
        self.app = msal.ConfidentialClientApplication(
            client_id, authority=self.authority, client_credential=client_secret
        )
        self.graph_base = "https://graph.microsoft.com/v1.0"
    
    def audit_oauth_consent(self):
        token_result = self.app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
        access_token = token_result['access_token']
        
        headers = {'Authorization': f'Bearer {access_token}'}
        
        apps_response = requests.get(
            f"{self.graph_base}/servicePrincipals",
            headers=headers,
            params={"$filter": "tags/any(t:t eq 'WindowsAzureActiveDirectoryIntegratedApp')"}
        ).json()
        
        suspicious_apps = []
        for app in apps_response.get('value', []):
            oauth_perms = app.get('oauth2PermissionScopes', [])
            delegated_perms = app.get('oauth2PermissionGrants', [])
            
            high_risk_scopes = ['Mail.Read', 'Mail.ReadWrite', 'Mail.Send',
                               'Files.Read.All', 'MailboxSettings.ReadWrite']
            
            for perm in delegated_perms:
                scope_list = perm.get('scope', '').split()
                risky = [s for s in scope_list if s in high_risk_scopes]
                if risky:
                    suspicious_apps.append({
                        'app_id': app.get('appId'),
                        'app_display_name': app.get('displayName'),
                        'publisher': app.get('publisherName'),
                        'risky_scopes': risky,
                        'consent_time': perm.get('startTime', 'unknown')
                    })
        
        return suspicious_apps
    
    def check_mailbox_delegation(self, user_upn):
        token_result = self.app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
        headers = {'Authorization': f'Bearer {token_result["access_token"]}'}
        
        delegates = requests.get(
            f"{self.graph_base}/users/{user_upn}/mailboxSettings",
            headers=headers
        ).json()
        
        return delegates

def trace_payment_instruction_change(internal_emails, bank_statements):
    timeline = []
    
    for email_record in internal_emails:
        subject = email_record['subject']
        keywords = ['bank', 'account', 'wire', 'payment', 'invoice', 'vendor',
                     'change', 'update', 'new account', 'replacement']
        
        if any(kw in subject.lower() for kw in keywords):
            timeline.append({
                'timestamp': email_record['received_time'],
                'event_type': 'potential_bank_change_request',
                'sender': email_record['sender'],
                'recipient': email_record['recipient'],
                'subject': subject,
                'has_attachment': bool(email_record.get('attachments')),
                'urgency_indicators': count_urgency_words(email_record['body']),
                'secrecy_requests': count_secrecy_words(email_record['body'])
            })
    
    for txn in bank_statements:
        matching_events = [e for e in timeline 
                         if abs((e['timestamp'] - txn['process_time']).total_seconds()) < 86400]
        if matching_events:
            txn['correlated_email_event'] = matching_events
    
    return sorted(timeline, key=lambda x: x['timestamp'])

def count_urgency_words(text):
    urgency_terms = ['urgent', 'immediately', 'asap', 'critical', 'confidential',
                     'do not share', 'secret', 'time-sensitive', 'rush', 'priority']
    text_lower = text.lower()
    return sum(1 for term in urgency_terms if term in text_lower)

def count_secrecy_words(text):
    secrecy_terms = ['do not tell', 'keep quiet', 'between us', 'not announce',
                     'merger preparation', 'board approval pending', 'internal only']
    text_lower = text.lower()
    return sum(1 for term in secrecy_terms if term in text_lower)

def detect_bec_communication_anomalies(mailbox_logs, baseline_period_days=30):
    import numpy as np
    
    normal_patterns = {}
    for user, logs in mailbox_logs.items():
        recent = logs[-baseline_period_days:]
        baseline = logs[:-baseline_period_days]
        
        baseline_external_ratio = np.mean([log['is_external'] for log in baseline])
        recent_external_ratio = np.mean([log['is_external'] for log in recent])
        
        baseline_outbound_hour = {}
        for log in baseline:
            hour = log['sent_time'].hour
            baseline_outbound_hour.setdefault(hour, 0)
            baseline_outbound_hour[hour] += 1
        
        recent_outbound_hour = {}
        for log in recent:
            hour = log['sent_time'].hour
            recent_outbound_hour.setdefault(hour, 0)
            recent_outbound_hour[hour] += 1
        
        thread_injection_candidates = []
        for log in recent:
            if log['in_reply_to'] and log['thread_position'] == 0:
                original_thread_starter = find_original_sender(log['in_reply_to'], logs)
                if original_thread_starter != log['sender'] and log['is_external']:
                    thread_injection_candidates.append(log)
        
        anomalies = {
            'external_ratio_spike': recent_external_ratio - baseline_external_ratio,
            'off_hours_spike': sum(recent_outbound_hour.get(h, 0) for h in range(0, 7)),
            'thread_injections': thread_injection_candidates
        }
        normal_patterns[user] = anomalies
    
    return normal_patterns

def analyze_urgency_patterns(text):
    urgency_categories = {
        'time_constraint': [
            r'(?:立即|马上|即刻)[^，。.]*?(?:处理|操作|回复|确认)',
            r'(?:within|in)\s+\d+\s+(?:minutes|hours|days)',
            r'by\s+(?:EOD|COB|end of day|tomorrow)',
            r' ASAP\b', r' urgent(?:ly)?\b', r' immediately\b'
        ],
        'threat_of_loss': [
            r'(?:账户|账号|account).*?(?:将被|will be).*?(?:锁定|关闭|suspend)',
            r'(?:penalty|罚款|处罚)',
            r'(?:legal action|法律诉讼)',
            r'(?:permanent|永久).*?(?:damage|损害|loss)'
        ],
        'authority_invocation': [
            r'(?:CEO|CTO|CFO|总裁|总经理).*?(?:要求|指示|命令|request|order)',
            r'(?:compliance|合规|regulatory|监管).*?(?:require|必须|需要)',
            r'(?:法务|legal|lawyer|律师).*?(?:advice|建议|opinion|意见)'
        ],
        'secrecy_demand': [
            r'(?:保密|不要告诉|confidential|do not share)',
            r'(?:between us|你我之间)',
            r'(?:do not (?:disclose|announce|inform))',
            r'(?:仅限|only for|restricted)'
        ]
    }
    
    results = {}
    for category, patterns in urgency_categories.items():
        matches = []
        for pattern in patterns:
            found = re.findall(pattern, text, re.IGNORECASE)
            matches.extend(found)
        results[category] = {
            'count': len(matches),
            'examples': matches[:5],
            'severity': 'high' if len(matches) >= 3 else 'medium' if len(matches) >= 1 else 'low'
        }
    
    overall_score = sum(r['count'] for r in results.values())
    results['overall_manipulation_score'] = min(overall_score / 10, 1.0)
    
    return results

def detect_translation_artifacts(text):
    indicators = {
        'mixed_scripts': bool(re.search(r'[\u4e00-\u9fff].*?[a-zA-Z]{3,}|[a-zA-Z]{3,}.*?[\u4e00-\u9fff]', text)),
        'inconsistent_honorifics': bool(re.search(r'您.*你|你.*您', text)),
        'english_punctuation': bool(re.search(r'[\u4e00-\u9fff]\s*[,;!?]\s*[\u4e00-\u9fff]', text)),
        'calque_phrases': bool(re.search(r'根据我们的记录|per our records|按照我们的信息', text, re.IGNORECASE)),
        'unnecessary_formality': bool(re.search(r'我们诚挚地通知您|we hereby inform you', text, re.IGNORECASE))
    }
    
    detected = sum(1 for v in indicators.values() if v)
    return {
        'artifacts_detected': detected,
        'translation_likelihood': 'very_high' if detected >= 3 else 'high' if detected >= 2 else 'possible' if detected >= 1 else 'unlikely',
        'details': indicators
    }

ATTACKER_PROFILING_MODEL = {
    "linguistic_background": {
        "native_language": "基于翻译痕迹判断",
        "education_level": "基于词汇复杂度和语法规范性",
        "cultural_context": "基于节日引用、时间表达习惯、称谓方式"
    },
    "organizational_capability": {
        "reconnaissance_depth": "基于对个人信息的掌握程度",
        "template_quality": "基于视觉设计和文案质量",
        "infrastructure_sophistication": "基于域名选择和服务器配置水平"
    },
    "psychological_techniques": {
        "primary_leverage": "主要利用的心理原则",
        "emotional_targets": ["恐惧", "好奇", "贪婪", "服从", "同情"],
        "escalation_pattern": "从小要求到大要求的递进模式"
    },
    "operational_security": {
        "anonymity_level": "身份信息保护措施",
        "reusability": "基础设施复用程度",
        "counter_forensics": "反取证意识强度"
    }
}

import speech_recognition as sr
from pydub import AudioSegment

def analyze_vishing_recording(audio_file_path):
    audio = AudioSegment.from_file(audio_file_path)
    
    duration = len(audio) / 1000
    sample_rate = audio.frame_rate
    channels = audio.channels
    
    caller_profile = {
        'duration_seconds': duration,
        'accent_detection': analyze_accent(audio),
        'background_noise': detect_background_environment(audio),
        'voice_gender': detect_voice_gender(audio),
        'estimated_age_range': estimate_age_from_voice(audio),
        'language_dialect': detect_dialect(audio),
        'voip_indicators': detect_voip_artifacts(audio),
        'caller_id_spoof': None
    }
    
    recognizer = sr.Recognizer()
    with sr.AudioFile(audio_file_path) as source:
        audio_data = recognizer.record(source)
        try:
            transcript = recognizer.recognize_google(audio_data, language='zh-CN')
            caller_profile['transcript'] = transcript
            caller_profile['manipulation_analysis'] = analyze_urgency_patterns(transcript)
        except sr.UnknownValueError:
            caller_profile['transcript'] = 'Unable to transcribe'
    
    return caller_profile

def detect_voip_artifacts(audio):
    spectral_centroid = compute_spectral_centroid(audio)
    codec_signatures = {
        'g711': check_g711_codec(audio),
        'g729': check_g729_codec(audio),
        'opus': check_opus_codec(audio),
        'amr': check_amr_codec(audio)
    }
    active_codecs = [k for k, v in codec_signatures.items() if v]
    is_voip = len(active_codecs) > 0
    
    jitter_level = measure_jitter(audio)
    packet_loss_estimate = estimate_packet_loss(audio)
    
    return {
        'is_voip': is_voip,
        'codec': active_codecs,
        'jitter_ms': jitter_level,
        'packet_loss_pct': packet_loss_estimate
    }

shodan search "http.html:'Microsoft Office' country:CN port:443" --fields ip_str,port,org,hostnames

shodan host 198.51.100.23
218.xxx.xxx.xxx
Hostnames: sender.campaign-server.io
Organization: XYZ Hosting LLC
OS: Linux
Ports: 22, 80, 443, 8080

shodan search "http.title:'Document Sharing Portal'" --fields ip_str,port,http.title

import requests
import json

def query_ct_logs(domain):
    url = f"https://crt.sh/?q=%25.{domain}&output=json"
    response = requests.get(url, timeout=30)
    certificates = response.json()
    
    cert_analysis = {}
    for cert in certificates:
        issuer = cert.get('issuer_name', '')
        not_before = cert.get('not_before', '')
        common_name = cert.get('common_name', '')
        name_value = cert.get('name_value', '')
        
        subdomains = name_value.split('\n')
        for sd in subdomains:
            sd_clean = sd.strip().lower()
            if sd_clean not in cert_analysis:
                cert_analysis[sd_clean] = []
            cert_analysis[sd_clean].append({
                'issuer': issuer,
                'issued_date': not_before,
                'cn': common_name
            })
    
    historical_subdomains = set()
    for sd, certs in cert_analysis.items():
        historical_subdomains.add(sd)
        if len(certs) > 1:
            print(f"[!] {sd} 有多张证书记录 → 可能频繁续期/重新申请")
    
    suspicious_tlds = ['.xyz', '.top', '.tk', '.pw', '.cc', '.su']
    for sd in historical_subdomains:
        if any(sd.endswith(tld) for tld in suspicious_tlds):
            print(f"[!] 可疑TLD子域名: {sd}")
    
    return cert_analysis

使用 PassiveTotal / SecurityTrails / CirclPDNS 查询历史解析：

curl -s "https://api.passivetotal.org/v2/dns/passive" \
  -u "$PT_USERNAME:$PT_API_KEY" \
  -d '{"query": "198.51.100.23"}' | python3 -m json.tool

输出示例：
{
    "results": [
        {"resolve": "campaign-server.io", "firstSeen": "2025-12-01", "lastSeen": "2026-01-15", "recordType": "A"},
        {"resolve": "micros0ft-security.com", "firstSeen": "2025-12-28", "lastSeen": "2026-01-20", "recordType": "A"},
        {"resolve": "secure-login-verify.net", "firstSeen": "2026-01-05", "lastSeen": "2026-01-22", "recordType": "A"},
        {"resolve": "office365-update.org", "firstSeen": "2026-01-10", "lastSeen": "2026-01-25", "recordType": "A"}
    ]
}

同一IP承载多个仿冒域名 → 确认为集中化钓鱼基础设施

theHarvester -d company.com -b all -f output/company_recon.html

模块覆盖范围：
- baidu/bing/duckduckgo/yahoo: 搜索引擎枚举
- crtsh/certspotter: 证书透明度
- github-codesearch: GitHub代码搜索
- hackertarget/rapidDNS: 子域名暴力枚举
- linkedin/linkedin_links: LinkedIn人员收集
- twitter: Twitter账号发现
- urlscan: URL扫描历史
- vulners: 漏洞信息关联

docker run -p 5001:5001 -v spiderfoot_data:/root/.spiderfoot spiderfoot/spiderfoot:latest

SpiderFoot 内置模块类别：
- DNS: 子域名枚举/反向DNS/区域传输
- Email: 邮箱泄露检查/PwnedDB查询
- Social: 社交媒体账号发现/用户名枚举
- Darknet: Tor市场搜索/暗网泄露监控
- Threat Intel: VirusTotal/Shodan/AbuseIPDB集成
- Geo: IP地理定位/GPS坐标解析

#!/usr/bin/env python3
import requests
import json
import dns.resolver
import socket
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from collections import defaultdict

class AutomatedOSINT:
    def __init__(self, target_domain, api_keys=None):
        self.target = target_domain
        self.api_keys = api_keys or {}
        self.results = defaultdict(dict)
        self.ioc_set = set()
    
    def passive_dns_lookup(self):
        resolvers = [
            ('securitytrails', f"https://api.securitytrails.com/v1/history/{self.target}/dns/a"),
            ('circl_pdns', f"https://www.circl.lu/pdns/query/{self.target}")
        ]
        
        for provider, url in resolvers:
            try:
                headers = {}
                if provider == 'securitytrails':
                    headers['APIKEY'] = self.api_keys.get('securitytrails', '')
                
                resp = requests.get(url, headers=headers, timeout=15)
                if resp.status_code == 200:
                    self.results['passive_dns'][provider] = resp.json()
            except Exception as e:
                self.results['errors'].append(f'{provider}: {str(e)}')
    
    def certificate_transparency_scan(self):
        url = f"https://crt.sh/?q=%25.{self.target}&output=json"
        try:
            resp = requests.get(url, timeout=30)
            certs = resp.json()
            subdomains = set()
            issuers = defaultdict(int)
            
            for cert in certs:
                names = cert.get('name_value', '').split('\n')
                for name in names:
                    cleaned = name.strip().lower().lstrip('*.')
                    if cleaned and '.' in cleaned:
                        subdomains.add(cleaned)
                
                issuer_cn = cert.get('issuer_ca_id', 'unknown')
                issuers[str(issuer_cn)] += 1
            
            self.results['cert_transparency'] = {
                'subdomains': sorted(subdomains),
                'certificate_count': len(certs),
                'unique_issuers': dict(issuers),
                'oldest_cert': min(c.get('not_before', '9999') for c in certs) if certs else None,
                'newest_cert': max(c.get('not_before', '0000') for c in certs) if certs else None
            }
        except Exception as e:
            self.results['errors'].append(f'crt.sh: {str(e)}')
    
    def shodan_host_query(self, ip_address):
        if 'shodan' not in self.api_keys:
            return
        url = f"https://api.shodan.io/shodan/host/{ip_address}?key={self.api_keys['shodan']}"
        try:
            resp = requests.get(url, timeout=15)
            if resp.status_code == 200:
                data = resp.json()
                self.results['shodan'] = {
                    'ip': data.get('ip_str'),
                    'hostnames': data.get('hostnames', []),
                    'org': data.get('org', ''),
                    'os': data.get('os', ''),
                    'ports': data.get('ports', []),
                    'vulns': data.get('vulns', []),
                    'country': data.get('country_name', ''),
                    'city': data.get('city', '')
                }
                for hostname in data.get('hostnames', []):
                    self.ioc_set.add(hostname)
        except Exception as e:
            self.results['errors'].append(f'shodan: {str(e)}')
    
    def subdomain_enum(self):
        wordlist = ['mail', 'ftp', 'vpn', 'remote', 'webmail', 'admin', 'portal',
                    'api', 'dev', 'staging', 'test', 'owa', 'exchange', 'autodiscover',
                    'login', 'sso', 'auth', 'cloud', 'cdn', 'static', 'assets']
        
        found = []
        def resolve_subdomain(sub):
            fqdn = f"{sub}.{self.target}"
            try:
                ip = socket.gethostbyname(fqdn)
                return fqdn, ip
            except socket.gaierror:
                return fqdn, None
        
        with ThreadPoolExecutor(max_workers=20) as executor:
            futures = {executor.submit(resolve_subdomain, sub): sub for sub in wordlist}
            for future in as_completed(futures):
                fqdn, ip = future.result()
                if ip:
                    found.append({'fqdn': fqdn, 'ip': ip})
                    self.ioc_set.add(ip)
        
        self.results['subdomains'] = found
    
    def generate_osint_report(self):
        tasks = [
            self.passive_dns_lookup,
            self.certificate_transparency_scan,
            self.subdomain_enum
        ]
        
        threads = []
        for task in tasks:
            t = threading.Thread(target=task)
            threads.append(t)
            t.start()
        
        for t in threads:
            t.join(timeout=60)
        
        return {
            'target': self.target,
            'timestamp': __import__('datetime').datetime.utcnow().isoformat(),
            'findings': dict(self.results),
            'ioc_summary': {
                'domains': [ioc for ioc in self.ioc_set if not ioc.replace('.','').isdigit()],
                'ips': [ioc for ioc in self.ioc_set if ioc.replace('.','').isdigit()]
            }
        }

使用 snscrape 或 twint 提取公开推文及元数据：

snscrape --jsonl twitter-user @suspect_account > tweets.jsonl

Python 分析脚本：
import json

tweets = []
with open('tweets.jsonl', 'r') as f:
    for line in f:
        tweets.append(json.loads(line))

metadata_extraction = {
    'account_created': tweets[0]['date'] if tweets else None,
    'total_tweets': len(tweets),
    'posting_schedule': analyze_posting_frequency(tweets),
    'geotagged_posts': [t for t in tweets if t.get('coordinates')],
    'mentioned_accounts': extract_mentions(tweets),
    'hashtags_used': extract_hashtags(tweets),
    'url_shorteners': extract_shortened_urls(tweets),
    'image_metadata': extract_image_urls(tweets)
}

from PIL import Image
from PIL.ExifTags import TAGS, GPSTAGS
import requests

def extract_gps_from_image(image_path):
    image = Image.open(image_path)
    exif_data = image._getexif()
    
    if not exif_data:
        return None
    
    gps_info = {}
    for tag_id, value in exif_data.items():
        tag_name = TAGS.get(tag_id, tag_id)
        if tag_name == 'GPSInfo':
            for gps_tag_id, gps_value in value.items():
                gps_tag_name = GPSTAGS.get(gps_tag_id, gps_tag_id)
                gps_info[gps_tag_name] = gps_value
    
    def convert_gps(value):
        d = float(value[0])
        m = float(value[1])
        s = float(value[2])
        return d + m/60.0 + s/3600.0
    
    lat = convert_gps(gps_info['GPSLatitude'])
    lon = convert_gps(gps_info['GPSLongitude'])
    if gps_info['GPSLatitudeRef'] == 'S':
        lat = -lat
    if gps_info['GPSLongitudeRef'] == 'W':
        lon = -lon
    
    return {'latitude': lat, 'longitude': lon, 'altitude': gps_info.get('GPSAltitude')}

def reverse_geocode(lat, lon):
    url = f"https://nominatim.openstreetmap.org/reverse?format=json&lat={lat}&lon={lon}"
    headers = {'User-Agent': 'ForensicGeoLoc/1.0'}
    resp = requests.get(url, headers=headers, timeout=10)
    return resp.json().get('address', {})

def analyze_image_background(image_path):
    from PIL import Image
    img = Image.open(image_path)
    width, height = img.size
    
    regions = {
        'sky': img.crop((0, 0, width, height//4)),
        'mid_ground': img.crop((0, height//4, width, height//2)),
        'foreground': img.crop((0, height//2, width, height))
    }
    
    dominant_colors = {}
    for region_name, region_img in regions.items():
        small = region_img.resize((50, 50))
        pixels = list(small.getdata())
        avg_color = tuple(sum(c)//len(pixels) for c in zip(*pixels))
        dominant_colors[region_name] = avg_color
    
    return dominant_colors

import networkx as nx
import matplotlib.pyplot as plt

def build_social_graph(accounts_data):
    G = nx.DiGraph()
    
    for account in accounts_data:
        username = account['username']
        G.add_node(username, 
                   created=account['created_at'],
                   followers=account['followers_count'],
                   verified=account.get('verified', False))
        
        for mention in account.get('mentions', []):
            G.add_edge(username, mention['username'],
                      interaction_type=mention['type'],
                      frequency=mention['count'])
        
        for follower in account.get('follows', []):
            if not G.has_edge(username, follower):
                G.add_edge(username, follower, interaction_type='follow')
    
    centrality = nx.degree_centrality(G)
    betweenness = nx.betweenness_centrality(G)
    
    hubs = sorted(centrality.items(), key=lambda x: x[1], reverse=True)[:10]
    bridges = sorted(betweenness.items(), key=lambda x: x[1], reverse=True)[:10]
    
    return {
        'graph': G,
        'node_count': G.number_of_nodes(),
        'edge_count': G.number_of_edges(),
        'key_hubs': hubs,
        'bridge_accounts': bridges,
        'communities': nx.community.louvain_communities(G.to_undirected())
    }

import telethon
from telethon.sync import TelegramClient

async def telegram_channel_forensics(api_id, api_hash, channel_username):
    async with TelegramClient('forensic_session', api_id, api_hash) as client:
        entity = await client.get_entity(channel_username)
        
        channel_info = {
            'id': entity.id,
            'title': entity.title,
            'username': entity.username,
            'date_created': entity.date,
            'members_count': getattr(entity, 'participants_count', 'hidden'),
            'description': getattr(entity, 'about', ''),
            'linked_chat': getattr(entity, 'linked_chat_id', None)
        }
        
        messages = []
        async for message in client.iter_messages(entity, limit=1000):
            msg_data = {
                'id': message.id,
                'date': message.date.isoformat(),
                'text': message.text,
                'views': message.views,
                'forwards': message.forwards,
                'media': message.media is not None,
                'edited': message.edit_date.isoformat() if message.edit_date else None,
                'via_bot': message.via_bot.username if message.via_bot else None,
                'fwd_from': message.fwd_from.from_id if message.fwd_from else None
            }
            messages.append(msg_data)
        
        admins = []
        try:
            async for admin in client.iter_participants(entity, filter=telethon.tl.types.ChannelParticipantsAdmins):
                admins.append({
                    'id': admin.id,
                    'username': admin.username,
                    'role': admin.participant_rank
                })
        except Exception:
            pass
        
        return {
            'channel_info': channel_info,
            'messages': messages,
            'admins': admins,
            'message_frequency': analyze_message_frequency(messages)
        }

import discord
import asyncio
from datetime import datetime

class DiscordForensicBot(discord.Client):
    def __init__(self, guild_id):
        intents = discord.Intents.all()
        super().__init__(intents=intents)
        self.guild_id = guild_id
        self.forensic_data = {
            'guild_info': None,
            'channels': {},
            'members': [],
            'roles': [],
            'audit_log': []
        }
    
    async def collect_guild_forensics(self):
        guild = self.get_guild(self.guild_id)
        
        self.forensic_data['guild_info'] = {
            'name': guild.name,
            'id': guild.id,
            'owner_id': guild.owner_id,
            'created_at': guild.created_at.isoformat(),
            'member_count': guild.member_count,
            'boost_level': guild.premium_tier,
            'verification_level': str(guild.verification_level),
            'features': guild.features
        }
        
        for role in guild.roles:
            self.forensic_data['roles'].append({
                'name': role.name,
                'id': role.id,
                'permissions': list(role.permissions),
                'member_count': len(role.members),
                'color': str(role.color),
                'hoisted': role.hoist,
                'mentionable': role.mentionable,
                'created_at': role.created_at.isoformat()
            })
        
        async for entry in guild.audit_logs(limit=500):
            self.forensic_data['audit_log'].append({
                'action': str(entry.action),
                'user': str(entry.user),
                'target': str(entry.target),
                'reason': entry.reason,
                'created_at': entry.created_at.isoformat(),
                'changes': str(entry.changes.before) + ' -> ' + str(entry.changes.after)
            })
        
        await self.close()

import cv2
import numpy as np
from PIL import Image

class DeepfakeDetector:
    def error_level_analysis(self, image_path, quality=95):
        img = cv2.imread(image_path)
        resized = cv2.resize(img, (512, 512))
        
        encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), quality]
        _, compressed = cv2.imencode('.jpg', resized, encode_param)
        decompressed = cv2.imdecode(compressed, cv2.IMREAD_UNCHANGED)
        
        ela = cv2.absdiff(resized, decompressed)
        ela_normalized = cv2.normalize(ela, None, 0, 255, cv2.NORM_MINMAX)
        
        mean_error = np.mean(ela)
        std_error = np.std(ela)
        
        regions = self.segment_into_grid(ela_normalized, grid_size=(4, 4))
        anomaly_regions = []
        global_mean = np.mean(ela_normalized)
        global_std = np.std(ela_normalized)
        
        for idx, region in enumerate(regions):
            region_mean = np.mean(region)
            z_score = (region_mean - global_mean) / (global_std + 1e-6)
            if abs(z_score) > 2.0:
                anomaly_regions.append({
                    'region_index': idx,
                    'mean_intensity': float(region_mean),
                    'z_score': float(z_score),
                    'anomaly_type': 'likely_manipulated' if z_score > 0 else 'likely_original'
                })
        
        return {
            'mean_ela_error': float(mean_error),
            'std_ela_error': float(std_error),
            'anomaly_regions': anomaly_regions,
            'manipulation_probability': min(len(anomaly_regions) / 16.0, 1.0)
        }
    
    def frequency_domain_analysis(self, image_path):
        img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)
        img_float = np.float32(img)
        
        dft = cv2.dft(img_float)
        magnitude_spectrum = cv2.magnitude(dft[:, :, 0], dft[:, :, 1])
        magnitude_log = np.log1p(magnitude_spectrum)
        
        h, w = magnitude_log.shape
        center_y, center_x = h // 2, w // 2
        
        low_freq_mask = np.zeros_like(magnitude_log)
        cv2.circle(low_freq_mask, (center_x, center_y), min(h, w) // 8, 1, -1)
        high_freq_energy = np.sum(magnitude_log * (1 - low_freq_mask))
        total_energy = np.sum(magnitude_log)
        
        high_freq_ratio = high_freq_energy / (total_energy + 1e-6)
        
        artifacts = self.detect_periodic_artifacts(magnitude_log)
        
        return {
            'high_frequency_ratio': float(high_freq_ratio),
            'periodic_artifacts_detected': artifacts['detected'],
            'artifact_frequencies': artifacts.get('frequencies', []),
            'synthetic_likelihood': 'high' if high_freq_ratio > 0.6 or artifacts['detected'] else 'low'
        }
    
    def facial_consistency_check(self, image_path):
        face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml')
        img = cv2.imread(image_path)
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        faces = face_cascade.detectMultiScale(gray, 1.1, 4)
        
        consistency_results = {}
        
        if len(faces) > 0:
            x, y, w, h = faces[0]
            face_roi = gray[y:y+h, x:x+w]
            
            left_eye_region = face_roi[0:h//3, 0:w//2]
            right_eye_region = face_roi[0:h//3, w//2:]
            
            left_eye_hist = cv2.calcHist([left_eye_region], [0], None, [256], [0, 256])
            right_eye_hist = cv2.calcHist([right_eye_region], [0], None, [256], [0, 256])
            
            eye_symmetry = cv2.compareHist(left_eye_hist, right_eye_hist, cv2.HISTCMP_CORREL)
            
            consistency_results['eye_symmetry'] = float(eye_symmetry)
            consistency_results['face_aspect_ratio'] = w / h if h > 0 else 0
            consistency_results['face_count'] = len(faces)
            
            skin_tone_regions = [
                face_roi[h//3:2*h//3, w//4:3*w//4],
                face_roi[2*h//3:, w//4:3*w//4]
            ]
            skin_colors = [np.mean(r) for r in skin_tone_regions if r.size > 0]
            if len(skin_colors) >= 2:
                consistency_results['skin_tone_variance'] = float(np.std(skin_colors))
        
        return consistency_results

import librosa
import numpy as np
from scipy import signal

class AudioDeepfakeDetector:
    def __init__(self, audio_path):
        self.audio_path = audio_path
        self.y, self.sr = librosa.load(audio_path, sr=22050)
    
    def spectral_analysis(self):
        spectral_centroid = librosa.feature.spectral_centroid(y=self.y, sr=self.sr)
        spectral_rolloff = librosa.feature.spectral_rolloff(y=self.y, sr=self.sr)
        spectral_bandwidth = librosa.feature.spectral_bandwidth(y=self.y, sr=self.sr)
        zero_crossing_rate = librosa.feature.zero_crossing_rate(self.y)
        
        centroid_mean = float(np.mean(spectral_centroid))
        rolloff_mean = float(np.mean(spectral_rolloff))
        bandwidth_mean = float(np.mean(spectral_bandwidth))
        
        synthetic_indicators = {
            'abnormally_smooth_spectrum': float(np.std(spectral_centroid)) < 100,
            'missing_high_frequency_detail': rolloff_mean < 4000,
            'unnatural_bandwidth': bandwidth_mean < 500,
            'zero_crossing_anomaly': float(np.std(zero_crossing_rate)) < 0.01
        }
        
        return {
            'spectral_centroid_mean': centroid_mean,
            'spectral_rolloff_mean': rolloff_mean,
            'spectral_bandwidth_mean': bandwidth_mean,
            'synthetic_likelihood': sum(1 for v in synthetic_indicators.values() if v),
            'indicators': synthetic_indicators
        }
    
    def breathing_pattern_analysis(self):
        hop_length = 512
        energy = librosa.feature.rms(y=self.y, hop_length=hop_length)[0]
        
        silent_frames = energy < np.percentile(energy, 10)
        breathing_candidates = []
        
        in_breath = False
        breath_start = 0
        for i, is_silent in enumerate(silent_frames):
            if is_silent and not in_breath:
                breath_start = i
                in_breath = True
            elif not is_silent and in_breath:
                breath_duration_frames = i - breath_start
                breath_duration_sec = breath_duration_frames * hop_length / self.sr
                if 0.1 < breath_duration_sec < 3.0:
                    breathing_candidates.append({
                        'frame': breath_start,
                        'duration_seconds': breath_duration_sec,
                        'time_seconds': breath_start * hop_length / self.sr
                    })
                in_breath = False
        
        regular_intervals = []
        for i in range(1, len(breathing_candidates)):
            interval = breathing_candidates[i]['time_seconds'] - breathing_candidates[i-1]['time_seconds']
            regular_intervals.append(interval)
        
        interval_regularity = np.std(regular_intervals) if regular_intervals else float('inf')
        
        return {
            'breathing_events_found': len(breathing_candidates),
            'average_interval': float(np.mean(regular_intervals)) if regular_intervals else None,
            'interval_regularity_std': float(interval_regularity),
            'too_regular': interval_regularity < 0.3 if regular_intervals else False
        }
    
    def environmental_noise_consistency(self):
        noise_frames = librosa.effects.split(self.y, top_db=30)
        
        if len(noise_frames) < 2:
            return {'consistent': True, 'note': 'insufficient_noise_segments'}
        
        noise_spectra = []
        for start, end in noise_frames:
            segment = self.y[start:end]
            if len(segment) > 0:
                spectrum = np.abs(np.fft.rfft(segment))
                noise_spectra.append(spectrum)
        
        correlations = []
        for i in range(len(noise_spectra)):
            for j in range(i+1, len(noise_spectra)):
                min_len = min(len(noise_spectra[i]), len(noise_spectra[j]))
                corr = np.corrcoef(noise_spectra[i][:min_len], noise_spectra[j][:min_len])[0, 1]
                correlations.append(corr)
        
        avg_correlation = float(np.mean(correlations)) if correlations else 0
        
        return {
            'noise_segments': len(noise_frames),
            'average_noise_correlation': avg_correlation,
            'consistent_environment': avg_correlation > 0.8,
            'possible_splicing': avg_correlation < 0.3 and len(noise_frames) > 3
        }

import cv2
import os
from datetime import datetime

class PhysicalSEForensics:
    def __init__(self, cctv_directory, access_log_path):
        self.cctv_dir = cctv_directory
        self.access_logs = self.parse_access_logs(access_log_path)
    
    def parse_access_logs(self, log_path):
        logs = []
        with open(log_path, 'r') as f:
            for line in f:
                parts = line.strip().split(',')
                if len(parts) >= 4:
                    logs.append({
                        'timestamp': datetime.strptime(parts[0], '%Y-%m-%d %H:%M:%S'),
                        'badge_id': parts[1],
                        'door_id': parts[2],
                        'access_result': parts[3],
                        'photo_ref': parts[4] if len(parts) > 4 else None
                    })
        return logs
    
    def detect_tailgating(self, video_path, fps=30):
        cap = cv2.VideoCapture(video_path)
        frame_count = 0
        person_detections = []
        
        person_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_fullbody.xml')
        
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            
            frame_count += 1
            if frame_count % fps == 0:
                persons = person_cascade.detectMultiScale(frame, 1.1, 3)
                
                if len(persons) > 1:
                    badge_time = self.find_closest_access_event(frame_count / fps)
                    if badge_time:
                        person_detections.append({
                            'timestamp_seconds': frame_count / fps,
                            'person_count': len(persons),
                            'badge_swipe_count': 1,
                            'tailgating_suspected': len(persons) > 1,
                            'closest_badge_event': badge_time
                        })
        
        cap.release()
        return person_detections
    
    def find_closest_access_event(self, video_timestamp):
        video_dt = self.base_time + __import__('datetime').timedelta(seconds=video_timestamp)
        closest = None
        min_diff = float('inf')
        
        for log in self.access_logs:
            diff = abs((log['timestamp'] - video_dt).total_seconds())
            if diff < min_diff:
                min_diff = diff
                closest = log
        
        return closest if min_diff < 5 else None
    
    def correlate_physical_digital(self, suspicious_events):
        correlations = []
        
        for event in suspicious_events:
            time_window = __import__('datetime').timedelta(minutes=5)
            event_time = event.get('timestamp')
            
            matching_digital = []
            for log in self.access_logs:
                if abs((log['timestamp'] - event_time).total_seconds()) <= time_window.total_seconds():
                    matching_digital.append(log)
            
            correlations.append({
                'physical_event': event,
                'matching_access_logs': matching_digital,
                'badge_used': event.get('badge_id'),
                'anomaly': len(matching_digital) > 1 or event.get('tailgating_suspected')
            })
        
        return correlations

USB 设备取证检查流程：

1. 物理检查：
   - 记录设备外观、品牌、型号、序列号
   - 拍照固定证据
   - 检查是否有物理损坏或改装痕迹

2. 写保护后连接取证工作站：
   dmesg | tail -20
   lsusb -v
   fdisk -l /dev/sdX

3. 创建磁盘镜像：
   dd if=/dev/sdX of=/evidence/usb_image.dd bs=4M status=progress
   sha256sum /evidence/usb_image.dd > /evidence/usb_image.dd.sha256

4. 挂载分析：
   mount -o ro /evidence/usb_image.dd /mnt/usb_analysis
   find /mnt/usb_analysis -type f -exec file {} \;
   find /mnt/usb_analysis -type f -exec md5sum {} \;

5. 检查恶意配置：
   - autorun.inf / autorun.sh 自动执行文件
   - 伪装的固件更新文件
   - BadUSB HID 注入脚本
   - 隐藏的分区或数据区

import usb.core
import usb.util

def detect_malicious_charging_station():
    devices = usb.core.find(find_all=True)
    
    suspicious_devices = []
    for dev in devices:
        device_info = {
            'vendor_id': hex(dev.idVendor),
            'product_id': hex(dev.idProduct),
            'manufacturer': usb.util.get_string(dev, dev.iManufacturer) if dev.iManufacturer else 'unknown',
            'product': usb.util.get_string(dev, dev.iProduct) if dev.iProduct else 'unknown',
            'serial': usb.util.get_string(dev, dev.iSerialNumber) if dev.iSerialNumber else 'unknown',
            'device_class': dev.bDeviceClass,
            'interfaces': []
        }
        
        for cfg in dev:
            for intf in cfg:
                device_info['interfaces'].append({
                    'class': intf.bInterfaceClass,
                    'subclass': intf.bInterfaceSubClass,
                    'protocol': intf.bInterfaceProtocol
                })
        
        is_data_device = any(
            intf['class'] in [0x08, 0x03, 0x02]
            for intf in device_info['interfaces']
        )
        is_known_charger = device_info['device_class'] == 0x00 and len(device_info['interfaces']) == 0
        
        if is_data_device:
            device_info['risk'] = 'HIGH - USB charging port with data capabilities detected'
            suspicious_devices.append(device_info)
        elif not is_known_charger:
            device_info['risk'] = 'MEDIUM - Unusual device characteristics'
            suspicious_devices.append(device_info)
    
    return suspicious_devices

import numpy as np
from datetime import datetime, timedelta

class InsiderThreatDetector:
    def __init__(self, user_activity_logs, baseline_days=90):
        self.logs = user_activity_logs
        self.baseline_days = baseline_days
    
    def detect_access_anomalies(self, user_id):
        user_logs = [l for l in self.logs if l['user_id'] == user_id]
        
        baseline_logs = [l for l in user_logs 
                        if (datetime.now() - l['timestamp']).days > self.baseline_days]
        recent_logs = [l for l in user_logs 
                      if (datetime.now() - l['timestamp']).days <= self.baseline_days]
        
        baseline_resources = set(l['resource_accessed'] for l in baseline_logs)
        recent_resources = set(l['resource_accessed'] for l in recent_logs)
        
        new_resources = recent_resources - baseline_resources
        sensitive_keywords = ['password', 'credential', 'secret', 'key', 'token',
                            'financial', 'hr', 'salary', 'customer_data', 'pii']
        
        sensitive_new_access = []
        for resource in new_resources:
            if any(kw in resource.lower() for kw in sensitive_keywords):
                sensitive_new_access.append(resource)
        
        baseline_hours = {}
        for log in baseline_logs:
            hour = log['timestamp'].hour
            baseline_hours.setdefault(hour, 0)
            baseline_hours[hour] += 1
        
        recent_hours = {}
        for log in recent_logs:
            hour = log['timestamp'].hour
            recent_hours.setdefault(hour, 0)
            recent_hours[hour] += 1
        
        off_hours_baseline = sum(baseline_hours.get(h, 0) for h in [0,1,2,3,4,5,23])
        off_hours_recent = sum(recent_hours.get(h, 0) for h in [0,1,2,3,4,5,23])
        
        return {
            'new_sensitive_resources': sensitive_new_access,
            'off_hours_baseline': off_hours_baseline,
            'off_hours_recent': off_hours_recent,
            'off_hours_spike': off_hours_recent > off_hours_baseline * 2,
            'total_new_resources': len(new_resources)
        }
    
    def detect_data_exfiltration(self, user_id):
        user_logs = [l for l in self.logs if l['user_id'] == user_id]
        
        large_transfers = [l for l in user_logs 
                         if l.get('transfer_size_mb', 0) > 100]
        
        external_uploads = [l for l in user_logs 
                          if l.get('destination_type') == 'external' 
                          and l.get('action') in ['upload', 'send', 'share']]
        
        archive_creations = [l for l in user_logs 
                           if l.get('action') in ['compress', 'zip', 'encrypt']
                           and l.get('file_count', 0) > 10]
        
        usb_copies = [l for l in user_logs 
                     if l.get('destination_type') == 'usb' 
                     and l.get('action') == 'copy']
        
        printing_events = [l for l in user_logs 
                         if l.get('action') == 'print'
                         and l.get('page_count', 0) > 50]
        
        exfil_indicators = {
            'large_transfers': len(large_transfers),
            'external_uploads': len(external_uploads),
            'archive_creations': len(archive_creations),
            'usb_copies': len(usb_copies),
            'bulk_printing': len(printing_events),
            'total_indicators': len(large_transfers) + len(external_uploads) + 
                              len(archive_creations) + len(usb_copies) + len(printing_events)
        }
        
        return exfil_indicators
    
    def detect_social_engineering_susceptibility(self, user_id):
        user_logs = [l for l in self.logs if l['user_id'] == user_id]
        
        phishing_clicks = [l for l in user_logs 
                         if l.get('event_type') == 'phishing_link_clicked']
        password_resets = [l for l in user_logs 
                         if l.get('event_type') == 'password_reset' 
                         and l.get('trigger') == 'user_request']
        unusual_grants = [l for l in user_logs 
                        if l.get('event_type') == 'permission_grant'
                        and l.get('authorized_by') != 'admin']
        
        return {
            'phishing_click_count': len(phishing_clicks),
            'self_service_resets': len(password_resets),
            'unauthorized_grants': len(unusual_grants),
            'susceptibility_score': min((len(phishing_clicks) * 3 + 
                                        len(password_resets) * 2 + 
                                        len(unusual_grants) * 5) / 20, 1.0)
        }

title: Suspicious Off-Hours Access to Sensitive Resources
id: a1b2c3d4-5678-90ab-cdef-111111111111
status: experimental
description: 检测非工作时间对敏感资源的异常访问行为
author: SOC Team
date: 2026/07/02
logsource:
    category: process_creation
    product: windows
detection:
    selection:
        ParentImage|endswith:
            - '\powershell.exe'
            - '\cmd.exe'
            - '\wscript.exe'
            - '\cscript.exe'
        CommandLine|contains:
            - 'password'
            - 'credential'
            - 'export'
            - 'download'
            - 'compress'
    timeframe:
        field: Timestamp
        condition:
            lt: '06:00:00'
            gt: '22:00:00'
    condition: selection and timeframe
level: high
tags:
    - attack.exfiltration
    - attack.credential_access
    - insider_threat

title: Unusual Volume of File Access by Single User
id: a1b2c3d4-5678-90ab-cdef-222222222222
status: experimental
description: 检测单用户在短时间内访问大量文件的行为
author: SOC Team
date: 2026/07/02
logsource:
    category: file_event
    product: windows
detection:
    selection:
        TargetFilename|contains:
            - '\\Finance\\'
            - '\\HR\\'
            - '\\Legal\\'
            - '\\Confidential\\'
    condition: selection | count(TargetFilename) by User > 100
    timeframe: 1h
level: medium
tags:
    - attack.collection
    - insider_threat

def assess_departing_employee_risk(employee_id, departure_type, access_history):
    risk_factors = {
        'departure_type': {
            'voluntary': 3,
            'involuntary': 7,
            'retirement': 1,
            'termination_for_cause': 9
        },
        'access_level': {
            'standard': 2,
            'elevated': 5,
            'admin': 8,
            'domain_admin': 10
        },
        'data_sensitivity': {
            'general': 1,
            'confidential': 5,
            'restricted': 8,
            'top_secret': 10
        },
        'recent_behavior': {
            'normal': 0,
            'policy_violations': 4,
            'access_anomalies': 6,
            'data_exfil_indicators': 9
        }
    }
    
    score = 0
    score += risk_factors['departure_type'].get(departure_type, 5)
    
    max_access = max(access_history, key=lambda x: x.get('access_level_score', 0))
    score += risk_factors['access_level'].get(max_access.get('access_level', 'standard'), 2)
    
    if access_history[-30:]:
        recent_anomalies = [a for a in access_history[-30:] if a.get('anomaly')]
        score += min(len(recent_anomalies) * 2, 10)
    
    total_score = min(score / 30, 1.0)
    
    return {
        'employee_id': employee_id,
        'risk_score': total_score,
        'risk_level': 'critical' if total_score > 0.8 else 'high' if total_score > 0.6 else 'medium' if total_score > 0.3 else 'low',
        'recommended_actions': get_recommended_actions(total_score)
    }

def get_recommended_actions(risk_score):
    if risk_score > 0.8:
        return ['immediate_access_revocation', 'escort_from_premises', 'device_confiscation', 'full_audit_log_review']
    elif risk_score > 0.6:
        return ['access_revocation_within_24h', 'manager_notification', 'data_access_audit']
    elif risk_score > 0.3:
        return ['standard_offboarding', 'knowledge_transfer_session', 'monitoring_during_notice_period']
    else:
        return ['standard_offboarding_procedure']

from collections import defaultdict
from datetime import datetime

class EvidenceCorrelator:
    def __init__(self):
        self.evidence_items = []
        self.correlation_graph = defaultdict(set)
    
    def add_evidence(self, evidence_type, source, indicator, timestamp, confidence_level):
        item = {
            'id': len(self.evidence_items),
            'type': evidence_type,
            'source': source,
            'indicator': indicator,
            'timestamp': timestamp,
            'confidence': confidence_level,
            'correlated_with': set()
        }
        self.evidence_items.append(item)
        return item['id']
    
    def correlate_by_indicator(self):
        indicator_map = defaultdict(list)
        for item in self.evidence_items:
            indicator_map[item['indicator']].append(item['id'])
        
        for indicator, ids in indicator_map.items():
            if len(ids) > 1:
                for i in range(len(ids)):
                    for j in range(i+1, len(ids)):
                        self.evidence_items[ids[i]]['correlated_with'].add(ids[j])
                        self.evidence_items[ids[j]]['correlated_with'].add(ids[i])
                        self.correlation_graph[ids[i]].add(ids[j])
                        self.correlation_graph[ids[j]].add(ids[i])
    
    def correlate_by_time_window(self, window_minutes=60):
        sorted_items = sorted(self.evidence_items, key=lambda x: x['timestamp'])
        
        for i in range(len(sorted_items)):
            for j in range(i+1, len(sorted_items)):
                time_diff = (sorted_items[j]['timestamp'] - sorted_items[i]['timestamp']).total_seconds()
                if time_diff > window_minutes * 60:
                    break
                if sorted_items[i]['source'] != sorted_items[j]['source']:
                    self.evidence_items[sorted_items[i]['id']]['correlated_with'].add(sorted_items[j]['id'])
                    self.evidence_items[sorted_items[j]['id']]['correlated_with'].add(sorted_items[i]['id'])
    
    def generate_attribution_assessment(self):
        clusters = self._find_connected_components()
        
        assessments = []
        for cluster in clusters:
            cluster_items = [self.evidence_items[i] for i in cluster]
            
            l1_count = sum(1 for i in cluster_items if i['confidence'] == 'L1')
            l2_count = sum(1 for i in cluster_items if i['confidence'] == 'L2')
            
            if l1_count >= 2:
                attribution_confidence = 'HIGH'
            elif l1_count >= 1 and l2_count >= 3:
                attribution_confidence = 'MEDIUM-HIGH'
            elif l2_count >= 3:
                attribution_confidence = 'MEDIUM'
            else:
                attribution_confidence = 'LOW'
            
            assessments.append({
                'cluster_size': len(cluster),
                'evidence_types': list(set(i['type'] for i in cluster_items)),
                'l1_evidence_count': l1_count,
                'l2_evidence_count': l2_count,
                'attribution_confidence': attribution_confidence,
                'indicators': list(set(i['indicator'] for i in cluster_items))
            })
        
        return assessments
    
    def _find_connected_components(self):
        visited = set()
        components = []
        
        def dfs(node, component):
            visited.add(node)
            component.append(node)
            for neighbor in self.correlation_graph.get(node, set()):
                if neighbor not in visited:
                    dfs(neighbor, component)
        
        for i in range(len(self.evidence_items)):
            if i not in visited:
                component = []
                dfs(i, component)
                if len(component) > 1:
                    components.append(component)
        
        return components