#!/usr/bin/env python3
"""
直接通过 SQL 插入测试数据
用法：
# 生成常规测试数据
python backend/scripts/generate_test_data_sql.py
python backend/scripts/generate_test_data_sql.py --clear # 清除后重新生成
# 生成百万级测试数据(用于测试 Dashboard 卡片溢出)
python backend/scripts/generate_test_data_sql.py --million
python backend/scripts/generate_test_data_sql.py --million --clear # 清除后生成百万级数据
百万级数据说明：
- 目标: 1,000
- 子域名: 200,000
- 网站: 200,000
- 端点: 200,000
- IP (host_port_mapping): 200,000
- 漏洞: 200,000 (critical: 50k, high: 50k, medium: 50k, low: 30k, info: 20k)
- 总资产: ~660,000
"""
import argparse
import random
import json
import os
from datetime import datetime, timedelta
from decimal import Decimal
from pathlib import Path
import psycopg2
from psycopg2.extras import execute_values
def generate_fixed_length_url(target_name: str, length: int = 245, path_hint: str = '') -> str:
"""
生成固定长度的 URL
Args:
target_name: 目标域名
length: 目标URL长度，默认245
path_hint: 可选的路径提示，用于区分不同类型的URL
Returns:
固定长度的URL字符串
"""
base = f'https://{target_name}'
# 基础路径
paths = [
'/api/v3/enterprise/security-assessment/vulnerability-management',
'/admin/dashboard/system-configuration/advanced-settings',
'/portal/user-authentication/multi-factor/verification',
'/services/cloud-infrastructure/monitoring/metrics',
'/internal/system-administration/audit-logging/events',
]
path = random.choice(paths) if not path_hint else f'/{path_hint}'
url = f'{base}{path}'
# 添加查询参数
param_idx = 0
while len(url) < length - 20:
param_idx += 1
param = f'p{param_idx}={random.randint(10000000, 99999999)}'
separator = '?' if '?' not in url else '&'
url = f'{url}{separator}{param}'
# 精确调整到目标长度
if len(url) < length:
# 添加填充参数
padding_needed = length - len(url) - 1 # -1 for '&' or '?'
if padding_needed > 0:
separator = '?' if '?' not in url else '&'
# 创建精确长度的填充
padding = 'x' * padding_needed
url = f'{url}{separator}{padding}'
# 截断到精确长度
if len(url) > length:
url = url[:length]
return url
def generate_fixed_length_text(length: int = 300, text_type: str = 'description') -> str:
"""
生成固定长度的文本内容
Args:
length: 目标文本长度，默认300
text_type: 文本类型，用于选择不同的内容模板
Returns:
固定长度的文本字符串
"""
# 基础文本模板
templates = {
'description': [
'A critical security vulnerability was discovered in the application authentication module. This vulnerability allows attackers to bypass security controls and gain unauthorized access to sensitive system resources. The issue stems from improper input validation and insufficient access control mechanisms. Exploitation could lead to complete system compromise, data exfiltration, and service disruption. Immediate remediation is strongly recommended including implementing proper input sanitization, strengthening authentication mechanisms, and deploying additional security monitoring. The vulnerability affects multiple components including user authentication, session management, API endpoints, and data processing pipelines. Risk assessment indicates high severity with potential for significant business impact.',
'Server-side request forgery (SSRF) vulnerability detected in the API gateway service. An attacker can manipulate server-side requests to access internal network resources, potentially exposing sensitive configuration data, internal services, and cloud metadata endpoints. The vulnerability exists due to insufficient URL validation in the proxy functionality. This could allow attackers to scan internal networks, access cloud instance metadata, retrieve sensitive credentials, and pivot to other internal systems. Recommended mitigations include implementing strict URL allowlisting, blocking requests to internal IP ranges, and adding network segmentation controls. The vulnerability has been assigned a high severity rating due to potential for lateral movement.',
'Remote code execution vulnerability identified in the file upload processing module. Insufficient file type validation allows attackers to upload malicious executable files that can be triggered to execute arbitrary code on the server. The vulnerability bypasses existing security controls through specially crafted file headers and extension manipulation. Successful exploitation grants attackers full control over the affected server, enabling data theft, malware deployment, and establishment of persistent backdoor access. Critical remediation steps include implementing strict file type validation, sandboxed file processing, content inspection, and removal of execution permissions from upload directories. This vulnerability requires immediate attention.',
'Cross-site scripting (XSS) vulnerability found in the user profile management interface. User-supplied input is rendered without proper encoding, allowing injection of malicious JavaScript code. Attackers can exploit this to steal session tokens, perform actions on behalf of authenticated users, redirect victims to phishing sites, and exfiltrate sensitive personal information. The vulnerability affects multiple input fields including display name, bio, and custom URL parameters. Remediation requires implementing context-aware output encoding, Content Security Policy headers, and input validation. Additionally, consider implementing HTTP-only and Secure flags on session cookies to limit the impact of successful XSS attacks.',
'SQL injection vulnerability discovered in the advanced search functionality. The application constructs database queries using unsanitized user input, enabling attackers to manipulate query logic, extract sensitive data, modify database contents, or execute administrative operations. The vulnerability affects the product search, user lookup, and reporting modules. Exploitation could result in complete database compromise, unauthorized data access, data manipulation, and potential privilege escalation. Immediate remediation includes implementing parameterized queries, stored procedures, input validation, and principle of least privilege for database accounts. Consider deploying a web application firewall as an additional defense layer.',
],
'organization': [
'A leading global technology corporation specializing in enterprise software solutions, cloud computing infrastructure, cybersecurity services, and digital transformation consulting. The organization operates across multiple continents with regional headquarters in North America, Europe, and Asia-Pacific. Core business units include enterprise resource planning systems, customer relationship management platforms, supply chain optimization tools, and advanced analytics solutions. The company maintains strategic partnerships with major cloud providers and technology vendors. Annual revenue exceeds several billion dollars with consistent year-over-year growth. The organization employs thousands of professionals including software engineers, security researchers, and business consultants.',
'An innovative financial technology company providing comprehensive digital banking services, payment processing solutions, and investment management platforms. The organization serves millions of customers globally through mobile applications, web portals, and API integrations. Key offerings include real-time payment processing, cryptocurrency trading, automated investment advisory, and small business lending. The company maintains regulatory compliance across multiple jurisdictions and holds various financial services licenses. Security infrastructure includes advanced fraud detection, multi-factor authentication, and end-to-end encryption. The organization has received multiple industry awards for innovation and customer satisfaction.',
'A healthcare technology enterprise focused on electronic health records, telemedicine platforms, medical device integration, and healthcare analytics. The organization partners with hospitals, clinics, and healthcare systems worldwide to improve patient outcomes and operational efficiency. Core products include comprehensive EHR systems, patient engagement portals, clinical decision support tools, and population health management platforms. The company maintains strict compliance with healthcare regulations including HIPAA, GDPR, and regional data protection requirements. Research and development investments focus on artificial intelligence applications in diagnostics, treatment optimization, and predictive health analytics.',
],
'title': [
'Enterprise Resource Planning System - Comprehensive Business Management Dashboard with Real-time Analytics, Workflow Automation, and Multi-department Integration Capabilities for Global Operations Management and Strategic Decision Support',
'Advanced Security Operations Center - Unified Threat Detection and Response Platform featuring Machine Learning-powered Anomaly Detection, Automated Incident Response, and Comprehensive Security Posture Management',
'Customer Experience Management Platform - Omnichannel Engagement Solution with AI-driven Personalization, Journey Orchestration, Sentiment Analysis, and Predictive Customer Behavior Modeling Capabilities',
'Cloud Infrastructure Management Console - Multi-cloud Orchestration Platform supporting AWS, Azure, and GCP with Automated Provisioning, Cost Optimization, Compliance Monitoring, and Performance Analytics',
'Data Analytics and Business Intelligence Suite - Self-service Analytics Platform with Advanced Visualization, Predictive Modeling, Natural Language Query Processing, and Automated Report Generation',
],
}
# 选择模板
template_list = templates.get(text_type, templates['description'])
base_text = random.choice(template_list)
# 调整到目标长度
if len(base_text) < length:
# 需要扩展文本
padding_words = [
'Additionally', 'Furthermore', 'Moreover', 'Consequently', 'Subsequently',
'comprehensive', 'implementation', 'infrastructure', 'configuration', 'authentication',
'vulnerability', 'exploitation', 'remediation', 'mitigation', 'assessment',
]
while len(base_text) < length - 20:
base_text += f' {random.choice(padding_words)}'
# 精确填充
if len(base_text) < length:
padding_needed = length - len(base_text)
base_text += ' ' + 'x' * (padding_needed - 1)
# 截断到精确长度
if len(base_text) > length:
base_text = base_text[:length]
return base_text
def load_env_file(env_path: str) -> dict:
"""从 .env 文件加载环境变量"""
env_vars = {}
if os.path.exists(env_path):
with open(env_path, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#') and '=' in line:
key, value = line.split('=', 1)
env_vars[key.strip()] = value.strip()
return env_vars
def get_db_config() -> dict:
"""从 docker/.env 读取数据库配置"""
# 获取项目根目录
script_dir = Path(__file__).resolve().parent
project_root = script_dir.parent.parent
env_path = project_root / 'docker' / '.env'
env_vars = load_env_file(str(env_path))
# 获取数据库配置，docker/.env 中 DB_HOST=postgres 是容器内地址，本地运行需要用 localhost
db_host = env_vars.get('DB_HOST', 'postgres')
if db_host == 'postgres':
db_host = 'localhost' # 本地运行脚本时使用 localhost
return {
'host': db_host,
'port': int(env_vars.get('DB_PORT', 5432)),
'dbname': env_vars.get('DB_NAME', 'xingrin'),
'user': env_vars.get('DB_USER', 'postgres'),
'password': env_vars.get('DB_PASSWORD', ''),
}
def generate_raw_response_headers(headers_dict: dict) -> str:
"""
将响应头字典转换为原始 HTTP 响应头字符串格式
Args:
headers_dict: 响应头字典
Returns:
原始 HTTP 响应头字符串，格式如：
HTTP/1.1 200 OK
Server: nginx
Content-Type: text/html
...
"""
lines = ['HTTP/1.1 200 OK']
for key, value in headers_dict.items():
# 将下划线转换为连字符，并首字母大写
header_name = key.replace('_', '-').title()
lines.append(f'{header_name}: {value}')
return '\r\n'.join(lines)
DB_CONFIG = get_db_config()
class TestDataGenerator:
def __init__(self, clear: bool = False):
self.conn = psycopg2.connect(**DB_CONFIG)
self.conn.autocommit = False
self.clear = clear
def run(self):
try:
if self.clear:
print("🗑️ 清除现有数据...")
self.clear_data()
print("🚀 开始生成测试数据...\n")
engine_ids = self.create_engines()
worker_ids = self.create_workers()
org_ids = self.create_organizations()
target_ids = self.create_targets(org_ids)
scan_ids = self.create_scans(target_ids, engine_ids, worker_ids)
self.create_scheduled_scans(org_ids, target_ids, engine_ids)
self.create_subdomains(target_ids)
website_ids = self.create_websites(target_ids)
self.create_endpoints(target_ids)
self.create_directories(target_ids, website_ids)
self.create_host_port_mappings(target_ids)
self.create_vulnerabilities(target_ids)
# 生成快照数据(扫描历史详细页面使用)
self.create_subdomain_snapshots(scan_ids)
self.create_website_snapshots(scan_ids)
self.create_endpoint_snapshots(scan_ids)
self.create_directory_snapshots(scan_ids)
self.create_host_port_mapping_snapshots(scan_ids)
self.create_vulnerability_snapshots(scan_ids)
# 生成指纹数据
self.create_ehole_fingerprints()
self.create_goby_fingerprints()
self.create_wappalyzer_fingerprints()
self.create_fingers_fingerprints()
self.create_fingerprinthub_fingerprints()
self.create_arl_fingerprints()
self.conn.commit()
print("\n✅ 测试数据生成完成！")
except Exception as e:
self.conn.rollback()
print(f"\n❌ 生成失败: {e}")
raise
finally:
self.conn.close()
def clear_data(self):
"""清除所有测试数据"""
cur = self.conn.cursor()
# 先删除 IMMV（避免 pg_ivm 的 anyarray bug）
print(" 删除 IMMV...")
cur.execute("DROP TABLE IF EXISTS asset_search_view CASCADE")
self.conn.commit()
tables = [
# 指纹表
'ehole_fingerprint', 'goby_fingerprint', 'wappalyzer_fingerprint',
'fingers_fingerprint', 'fingerprinthub_fingerprint', 'arl_fingerprint',
# 快照表(先删除，因为有外键依赖 scan)
'vulnerability_snapshot', 'host_port_mapping_snapshot', 'directory_snapshot',
'endpoint_snapshot', 'website_snapshot', 'subdomain_snapshot',
# 资产表
'vulnerability', 'host_port_mapping', 'directory', 'endpoint',
'website', 'subdomain', 'scheduled_scan', 'scan',
'organization_targets', 'target', 'organization',
'nuclei_template_repo', 'wordlist', 'scan_engine', 'worker_node'
]
for table in tables:
cur.execute(f"DELETE FROM {table}")
self.conn.commit()
# 重建 IMMV
print(" 重建 IMMV...")
cur.execute("""
SELECT pgivm.create_immv('asset_search_view', $$
SELECT
w.id,
w.url,
w.host,
w.title,
w.tech,
w.status_code,
w.response_headers,
w.response_body,
w.created_at,
w.target_id
FROM website w
$$)
""")
self.conn.commit()
print(" ✓ 数据清除完成\n")
def create_workers(self) -> list:
"""创建 Worker 节点"""
print("👷 创建 Worker 节点...")
cur = self.conn.cursor()
# 生成随机后缀确保唯一性
suffix = random.randint(1000, 9999)
regions = ['asia-singapore-1', 'asia-singapore-2', 'asia-tokyo-1', 'asia-tokyo-2', 'asia-hongkong-1',
'asia-mumbai-1', 'asia-seoul-1', 'asia-sydney-1', 'asia-jakarta-1', 'asia-osaka-1',
'europe-frankfurt-1', 'europe-frankfurt-2', 'europe-london-1', 'europe-london-2',
'europe-paris-1', 'europe-ireland-1', 'europe-stockholm-1', 'europe-milan-1',
'us-east-virginia-1', 'us-east-virginia-2', 'us-east-ohio-1', 'us-west-oregon-1',
'us-west-oregon-2', 'us-west-california-1', 'us-central-iowa-1',
'australia-sydney-1', 'australia-melbourne-1', 'brazil-saopaulo-1',
'canada-montreal-1', 'southafrica-capetown-1', 'middleeast-bahrain-1']
statuses = ['online', 'offline', 'pending', 'deploying', 'maintenance', 'error', 'upgrading']
workers = [
(f'local-worker-primary-high-performance-{suffix}', '127.0.0.1', True, 'online'),
(f'local-worker-secondary-backup-{suffix}', '127.0.0.2', True, 'online'),
]
# 随机生成 30-50 个远程 worker
num_remote = random.randint(30, 50)
selected_regions = random.sample(regions, min(num_remote, len(regions)))
for i, region in enumerate(selected_regions):
ip = f'192.168.{random.randint(1, 254)}.{random.randint(1, 254)}'
status = random.choice(statuses)
workers.append((f'remote-worker-{region}-{suffix}-{i:02d}', ip, False, status))
ids = []
for name, ip, is_local, status in workers:
cur.execute("""
INSERT INTO worker_node (name, ip_address, ssh_port, username, password, is_local, status, created_at, updated_at)
VALUES (%s, %s, 22, 'root', '', %s, %s, NOW(), NOW())
ON CONFLICT (name) DO UPDATE SET updated_at = NOW()
RETURNING id
""", (name, ip, is_local, status))
row = cur.fetchone()
if row:
ids.append(row[0])
print(f" ✓ 创建了 {len(ids)} 个 Worker 节点\n")
return ids
def create_engines(self) -> list:
"""创建扫描引擎"""
print("⚙️ 创建扫描引擎...")
cur = self.conn.cursor()
suffix = random.randint(1000, 9999)
engine_templates = [
('Full-Comprehensive-Security-Assessment-Enterprise-Grade-Vulnerability-Detection-System', 'subdomain_discovery:\n enabled: true\n tools: [subfinder, amass, findomain, assetfinder, chaos]\n timeout: {timeout}\n resolvers: [8.8.8.8, 1.1.1.1, 9.9.9.9]\nvulnerability_scanning:\n enabled: true\n nuclei:\n severity: critical,high,medium,low,info\n rate_limit: {rate}\n concurrency: {conc}\n templates: [cves, vulnerabilities, exposures, misconfigurations, default-logins]'),
('Quick-Reconnaissance-Fast-Discovery-Lightweight-Asset-Enumeration', 'subdomain_discovery:\n enabled: true\n tools: [subfinder, assetfinder]\n timeout: {timeout}\n passive_only: true\nport_scanning:\n enabled: true\n top_ports: {ports}\n rate: {rate}'),
('Deep-Vulnerability-Assessment-Extended-Security-Analysis-Framework', 'vulnerability_scanning:\n enabled: true\n nuclei:\n severity: critical,high,medium,low,info\n templates: [cves, vulnerabilities, exposures, misconfigurations, default-logins, takeovers]\n rate_limit: {rate}\n concurrency: {conc}\n dalfox:\n enabled: true\n blind_xss: true\n sqlmap:\n enabled: true\n level: 3\n risk: 2'),
('Passive-Information-Gathering-OSINT-Intelligence-Collection-Platform', 'subdomain_discovery:\n enabled: true\n passive_only: true\n sources: [crtsh, hackertarget, threatcrowd, virustotal, securitytrails, shodan, censys, binaryedge]\n timeout: {timeout}\n dns_bruteforce: false'),
('Web-Application-Security-Scanner-OWASP-Compliance-Testing-Suite', 'web_discovery:\n enabled: true\n httpx:\n threads: {conc}\n follow_redirects: true\n screenshot: true\nvulnerability_scanning:\n enabled: true\n dalfox:\n enabled: true\n blind_xss: true\n nuclei:\n templates: [cves, vulnerabilities, exposures]'),
('API-Endpoint-Security-Audit-RESTful-GraphQL-Assessment-Tool', 'endpoint_discovery:\n enabled: true\n katana:\n depth: {depth}\n concurrency: {conc}\n js_crawl: true\n automatic_form_fill: true\nvulnerability_scanning:\n enabled: true\n nuclei:\n templates: [exposures, misconfigurations]'),
('Infrastructure-Port-Scanner-Network-Service-Detection-Engine', 'port_scanning:\n enabled: true\n naabu:\n top_ports: {ports}\n rate: {rate}\n scan_all_ips: true\n service_detection: true\n version_detection: true\n os_detection: true'),
('Directory-Bruteforce-Engine-Content-Discovery-Fuzzing-Platform', 'directory_bruteforce:\n enabled: true\n ffuf:\n threads: {conc}\n wordlist: [common.txt, raft-large-directories.txt, raft-large-files.txt]\n recursion_depth: {depth}\n extensions: [php, asp, aspx, jsp, html, js, json, xml]'),
('Cloud-Infrastructure-Security-Assessment-AWS-Azure-GCP-Scanner', 'cloud_scanning:\n enabled: true\n providers: [aws, azure, gcp]\n services: [s3, ec2, rds, lambda, storage, compute, sql]\n misconfigurations: true\n public_exposure: true'),
('Container-Security-Scanner-Kubernetes-Docker-Vulnerability-Detector', 'container_scanning:\n enabled: true\n kubernetes:\n enabled: true\n rbac_audit: true\n network_policies: true\n docker:\n enabled: true\n image_scanning: true\n dockerfile_lint: true'),
('Mobile-Application-Security-Testing-iOS-Android-Assessment-Framework', 'mobile_scanning:\n enabled: true\n platforms: [ios, android]\n static_analysis: true\n dynamic_analysis: true\n api_testing: true\n ssl_pinning_bypass: true'),
('Compliance-Audit-Scanner-PCI-DSS-HIPAA-SOC2-Assessment-Tool', 'compliance_scanning:\n enabled: true\n frameworks: [pci-dss, hipaa, soc2, gdpr, iso27001]\n automated_reporting: true\n evidence_collection: true'),
]
# 随机选择 8-12 个引擎模板
num_engines = random.randint(8, 12)
selected = random.sample(engine_templates, min(num_engines, len(engine_templates)))
ids = []
for name_base, config_template in selected:
name = f'{name_base}-{suffix}'
config = config_template.format(
rate=random.choice([100, 150, 200, 300]),
conc=random.choice([10, 20, 50, 100]),
timeout=random.choice([300, 600, 900, 1200]),
ports=random.choice([100, 1000, 'full']),
depth=random.choice([2, 3, 4, 5])
)
cur.execute("""
INSERT INTO scan_engine (name, configuration, created_at, updated_at)
VALUES (%s, %s, NOW(), NOW())
ON CONFLICT (name) DO UPDATE SET configuration = EXCLUDED.configuration, updated_at = NOW()
RETURNING id
""", (name, config))
row = cur.fetchone()
if row:
ids.append(row[0])
print(f" ✓ 创建了 {len(ids)} 个扫描引擎\n")
return ids
def create_organizations(self) -> list:
"""创建组织"""
print("🏢 创建组织...")
cur = self.conn.cursor()
suffix = random.randint(1000, 9999)
org_templates = [
('Acme Corporation', '全球领先的技术解决方案提供商，专注于企业级软件开发、云计算服务和网络安全解决方案。公司成立于1995年，总部位于硅谷，在全球50多个国家设有分支机构，员工超过10万人，年营收超过500亿美元。'),
('TechStart Innovation Labs', '专注于人工智能、机器学习和区块链技术研发的创新实验室。拥有超过200名博士级研究人员，与全球顶尖大学建立了深度合作关系，已获得超过500项技术专利。'),
('Global Financial Services', '提供全方位数字银行服务的金融科技公司，包括移动支付、在线贷款、投资理财等服务。服务覆盖全球180个国家和地区，注册用户超过5亿，日均交易额超过100亿美元。'),
('HealthCare Plus Medical', '医疗信息化解决方案提供商，专注于电子病历系统、医院信息管理系统和远程医疗平台开发。产品已部署在全球3000多家医疗机构，服务超过1亿患者。'),
('E-Commerce Mega Platform', '亚太地区最大的电子商务平台之一，提供 B2B、B2C 和 C2C 多种交易模式。平台入驻商家超过500万，SKU数量超过10亿，日均订单量超过5000万单。'),
('Smart City Infrastructure', '智慧城市基础设施解决方案提供商，专注于物联网传感器网络、智能交通系统、城市大脑平台开发。已在全球100多个城市部署智慧城市解决方案，管理超过1000万个IoT设备。'),
('Educational Technology', '在线教育技术联盟，提供 K-12 和高等教育在线学习平台。平台拥有超过10万门课程，注册学员超过1亿人，与全球500多所知名大学建立了合作关系。'),
('Green Energy Solutions', '可再生能源管理系统提供商，专注于太阳能、风能发电站的监控、调度和优化管理。管理的清洁能源装机容量超过100GW，每年减少碳排放超过5000万吨。'),
('CyberSec Defense Corp', '网络安全防御公司，提供渗透测试、漏洞评估和安全咨询服务。拥有超过1000名认证安全专家，服务全球500强企业中的300多家，年处理安全事件超过100万起。'),
('CloudNative Systems', '云原生系统开发商，专注于 Kubernetes、微服务架构和 DevOps 工具链。产品被全球超过10万家企业采用，管理的容器实例超过1亿个，是CNCF的核心贡献者。'),
('DataFlow Analytics', '大数据分析平台，提供实时数据处理、商业智能和预测分析服务。平台日处理数据量超过100PB，支持超过1000种数据源接入，服务全球5000多家企业客户。'),
('MobileFirst Technologies', '移动优先技术公司，专注于 iOS/Android 应用开发和跨平台解决方案。已开发超过5000款移动应用，累计下载量超过50亿次，月活跃用户超过10亿。'),
('Quantum Computing Research', '量子计算研究机构，致力于量子算法、量子纠错和量子网络的前沿研究。拥有全球最先进的量子计算机之一，已实现1000+量子比特的稳定运算。'),
('Autonomous Vehicles Corp', '自动驾驶技术公司，专注于L4/L5级别自动驾驶系统研发。测试车队已累计行驶超过1亿公里，在全球20个城市开展商业化运营。'),
('Biotech Innovations', '生物技术创新企业，专注于基因编辑、细胞治疗和精准医疗。拥有超过100项生物技术专利，多款创新药物已进入临床试验阶段。'),
('Space Technology Systems', '航天技术系统公司，提供卫星通信、遥感数据和太空探索服务。已成功发射超过500颗卫星，建立了覆盖全球的低轨卫星互联网星座。'),
]
divisions = ['Global Division', 'Asia Pacific', 'EMEA Region', 'Americas', 'R&D Center', 'Digital Platform',
'Cloud Services', 'Security Team', 'Innovation Lab', 'Enterprise Solutions', 'Consumer Products',
'Infrastructure Services', 'Data Analytics', 'AI Research', 'Mobile Development', 'DevOps Platform']
# 随机选择 15-20 个组织
num_orgs = random.randint(15, 20)
selected = random.sample(org_templates, min(num_orgs, len(org_templates)))
ids = []
for name_base, _ in selected:
division = random.choice(divisions)
name = f'{name_base} - {division} ({suffix})'
# 生成固定 300 长度的描述
desc = generate_fixed_length_text(length=300, text_type='organization')
cur.execute("""
INSERT INTO organization (name, description, created_at, deleted_at)
VALUES (%s, %s, NOW() - INTERVAL '%s days', NULL)
ON CONFLICT DO NOTHING
RETURNING id
""", (name, desc, random.randint(0, 365)))
row = cur.fetchone()
if row:
ids.append(row[0])
print(f" ✓ 创建了 {len(ids)} 个组织\n")
return ids
def create_targets(self, org_ids: list) -> list:
"""创建扫描目标"""
print("🎯 创建扫描目标...")
cur = self.conn.cursor()
suffix = random.randint(1000, 9999)
# 超长域名生成，目标 200 字符左右
# 格式: {env}-{region}-{service}-{version}.{subdomain}.{company}-{project}-{team}-{suffix}.{domain}{tld}
envs = ['production', 'staging', 'development', 'testing', 'integration', 'performance', 'security-audit']
regions = ['us-east-1', 'us-west-2', 'eu-central-1', 'ap-southeast-1', 'ap-northeast-1', 'sa-east-1', 'eu-west-3']
services = ['api-gateway', 'authentication-service', 'user-management', 'payment-processing', 'notification-center', 'analytics-engine', 'content-delivery', 'search-indexer']
versions = ['v1', 'v2', 'v3', 'v2-beta', 'v3-alpha', 'v1-legacy', 'v2-stable']
subdomains = ['internal-services', 'external-facing', 'partner-integration', 'customer-portal', 'admin-dashboard', 'developer-tools', 'monitoring-system']
companies = ['acme-corporation-international', 'techstart-innovation-labs', 'globalfinance-services-group', 'healthcare-plus-medical-systems', 'ecommerce-platform-solutions', 'smartcity-infrastructure-development', 'cybersecurity-defense-corporation', 'cloudnative-enterprise-systems']
projects = ['digital-transformation-initiative', 'cloud-migration-project', 'security-enhancement-program', 'customer-experience-platform', 'data-analytics-modernization', 'infrastructure-automation-suite']
teams = ['engineering-team-alpha', 'devops-squad-bravo', 'security-team-charlie', 'platform-team-delta', 'infrastructure-team-echo']
domains = ['enterprise', 'platform', 'services', 'solutions', 'systems']
tlds = ['.com', '.io', '.net', '.org', '.dev', '.app', '.cloud', '.tech', '.systems']
ids = []
# 随机生成 100-150 个域名目标
num_domains = random.randint(100, 150)
used_domains = set()
for i in range(num_domains):
env = random.choice(envs)
region = random.choice(regions)
service = random.choice(services)
version = random.choice(versions)
subdomain = random.choice(subdomains)
company = random.choice(companies)
project = random.choice(projects)
team = random.choice(teams)
domain_name = random.choice(domains)
tld = random.choice(tlds)
# 生成超长域名，约 150-200 字符
domain = f'{env}-{region}-{service}-{version}.{subdomain}.{company}-{project}-{team}-{suffix}.{domain_name}{tld}'
if domain in used_domains:
continue
used_domains.add(domain)
cur.execute("""
INSERT INTO target (name, type, created_at, last_scanned_at, deleted_at)
VALUES (%s, 'domain', NOW() - INTERVAL '%s days', NOW() - INTERVAL '%s days', NULL)
ON CONFLICT DO NOTHING
RETURNING id
""", (domain, random.randint(30, 365), random.randint(0, 30)))
row = cur.fetchone()
if row:
ids.append(row[0])
# 随机关联到组织
if org_ids:
# 20% 概率关联多个组织(3-5个)，50% 概率关联1个组织，30% 不关联
rand_val = random.random()
if rand_val < 0.2:
# 关联多个组织 (3-5个)
num_orgs = min(random.randint(3, 5), len(org_ids))
selected_orgs = random.sample(org_ids, num_orgs)
for org_id in selected_orgs:
cur.execute("""
INSERT INTO organization_targets (organization_id, target_id)
VALUES (%s, %s)
ON CONFLICT DO NOTHING
""", (org_id, row[0]))
elif rand_val < 0.7:
# 关联1个组织
org_id = random.choice(org_ids)
cur.execute("""
INSERT INTO organization_targets (organization_id, target_id)
VALUES (%s, %s)
ON CONFLICT DO NOTHING
""", (org_id, row[0]))
# 随机生成 50-80 个 IP 目标
num_ips = random.randint(50, 80)
for _ in range(num_ips):
# 使用文档保留的 IP 范围
ip_ranges = [
(203, 0, 113), # TEST-NET-3
(198, 51, 100), # TEST-NET-2
(192, 0, 2), # TEST-NET-1
]
base = random.choice(ip_ranges)
ip = f'{base[0]}.{base[1]}.{base[2]}.{random.randint(1, 254)}'
cur.execute("""
INSERT INTO target (name, type, created_at, last_scanned_at, deleted_at)
VALUES (%s, 'ip', NOW() - INTERVAL '%s days', NOW() - INTERVAL '%s days', NULL)
ON CONFLICT DO NOTHING
RETURNING id
""", (ip, random.randint(30, 365), random.randint(0, 30)))
row = cur.fetchone()
if row:
ids.append(row[0])
# 随机生成 30-50 个 CIDR 目标
num_cidrs = random.randint(30, 50)
cidr_bases = ['10.0', '172.16', '172.17', '172.18', '192.168']
for _ in range(num_cidrs):
base = random.choice(cidr_bases)
third_octet = random.randint(0, 255)
mask = random.choice([24, 25, 26, 27, 28])
cidr = f'{base}.{third_octet}.0/{mask}'
cur.execute("""
INSERT INTO target (name, type, created_at, last_scanned_at, deleted_at)
VALUES (%s, 'cidr', NOW() - INTERVAL '%s days', NOW() - INTERVAL '%s days', NULL)
ON CONFLICT DO NOTHING
RETURNING id
""", (cidr, random.randint(30, 365), random.randint(0, 30)))
row = cur.fetchone()
if row:
ids.append(row[0])
print(f" ✓ 创建了 {len(ids)} 个扫描目标\n")
return ids
def create_scans(self, target_ids: list, engine_ids: list, worker_ids: list) -> list:
"""创建扫描任务"""
print("🔍 创建扫描任务...")
cur = self.conn.cursor()
if not target_ids or not engine_ids:
print(" ⚠ 缺少目标或引擎，跳过\n")
return []
statuses = ['cancelled', 'completed', 'failed', 'initiated', 'running']
status_weights = [0.05, 0.6, 0.1, 0.1, 0.15] # completed 占比最高
stages = ['subdomain_discovery', 'port_scanning', 'web_discovery', 'vulnerability_scanning', 'directory_bruteforce', 'endpoint_discovery']
error_messages = [
'Connection timeout while scanning target. Please check network connectivity.',
'DNS resolution failed for target domain.',
'Rate limit exceeded. Scan paused and will resume automatically.',
'Worker node disconnected during scan execution.',
'Insufficient disk space on worker node.',
'Target returned too many errors, scan aborted.',
'Authentication failed for protected resources.',
]
# 获取引擎名称映射
cur.execute("SELECT id, name FROM scan_engine WHERE id = ANY(%s)", (engine_ids,))
engine_name_map = {row[0]: row[1] for row in cur.fetchall()}
ids = []
# 随机选择目标数量 - 增加到 80-120 个
num_targets = min(random.randint(80, 120), len(target_ids))
selected_targets = random.sample(target_ids, num_targets)
for target_id in selected_targets:
# 每个目标随机 3-15 个扫描任务
num_scans = random.randint(3, 15)
for _ in range(num_scans):
status = random.choices(statuses, weights=status_weights)[0]
# 随机选择 1-3 个引擎
num_engines = random.randint(1, min(3, len(engine_ids)))
selected_engine_ids = random.sample(engine_ids, num_engines)
selected_engine_names = [engine_name_map.get(eid, f'Engine-{eid}') for eid in selected_engine_ids]
worker_id = random.choice(worker_ids) if worker_ids else None
progress = random.randint(10, 95) if status == 'running' else (100 if status == 'completed' else random.randint(0, 50))
stage = random.choice(stages) if status == 'running' else ''
error_msg = random.choice(error_messages) if status == 'failed' else ''
# 随机生成更真实的统计数据
subdomains = random.randint(50, 2000)
websites = random.randint(10, 500)
endpoints = random.randint(100, 5000)
ips = random.randint(20, 300)
directories = random.randint(200, 8000)
vulns_critical = random.randint(0, 20)
vulns_high = random.randint(0, 50)
vulns_medium = random.randint(0, 100)
vulns_low = random.randint(0, 150)
vulns_total = vulns_critical + vulns_high + vulns_medium + vulns_low + random.randint(0, 100) # info
days_ago = random.randint(0, 90)
cur.execute("""
INSERT INTO scan (
target_id, engine_ids, engine_names, yaml_configuration, status, worker_id, progress, current_stage,
results_dir, error_message, container_ids, stage_progress,
cached_subdomains_count, cached_websites_count, cached_endpoints_count,
cached_ips_count, cached_directories_count, cached_vulns_total,
cached_vulns_critical, cached_vulns_high, cached_vulns_medium, cached_vulns_low,
created_at, stopped_at, deleted_at
) VALUES (
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
NOW() - INTERVAL '%s days', %s, NULL
)
RETURNING id
""", (
target_id, selected_engine_ids, json.dumps(selected_engine_names), '', status, worker_id, progress, stage,
f'/app/results/scan_{target_id}_{random.randint(1000, 9999)}', error_msg, '{}', '{}',
subdomains, websites, endpoints, ips, directories, vulns_total,
vulns_critical, vulns_high, vulns_medium, vulns_low,
days_ago,
datetime.now() - timedelta(days=days_ago, hours=random.randint(0, 23)) if status in ['completed', 'failed', 'cancelled'] else None
))
row = cur.fetchone()
if row:
ids.append(row[0])
print(f" ✓ 创建了 {len(ids)} 个扫描任务\n")
return ids
def create_scheduled_scans(self, org_ids: list, target_ids: list, engine_ids: list):
"""创建定时扫描任务"""
print("⏰ 创建定时扫描任务...")
cur = self.conn.cursor()
if not engine_ids:
print(" ⚠ 缺少引擎，跳过\n")
return
suffix = random.randint(1000, 9999)
schedule_templates = [
('Daily-Full-Security-Assessment-Enterprise-Wide-Comprehensive-Vulnerability-Detection', '0 {hour} * * *'),
('Weekly-Vulnerability-Scan-Critical-Infrastructure-Protection-Program', '0 {hour} * * {dow}'),
('Monthly-Penetration-Testing-External-Attack-Surface-Management', '0 {hour} {dom} * *'),
('Hourly-Quick-Reconnaissance-Real-Time-Threat-Intelligence-Gathering', '{min} * * * *'),
('Bi-Weekly-Compliance-Check-Regulatory-Standards-Verification-Audit', '0 {hour} 1,15 * *'),
('Quarterly-Infrastructure-Audit-Network-Security-Posture-Assessment', '0 {hour} 1 1,4,7,10 *'),
('Daily-API-Security-Scan-RESTful-GraphQL-Endpoint-Protection', '{min} {hour} * * *'),
('Weekly-Web-Application-Scan-OWASP-Top-10-Vulnerability-Detection', '0 {hour} * * {dow}'),
('Nightly-Asset-Discovery-Shadow-IT-Detection-Inventory-Management', '0 {hour} * * *'),
('Weekend-Deep-Scan-Intensive-Security-Analysis-Full-Coverage', '0 {hour} * * 0,6'),
('Business-Hours-Monitor-Real-Time-Security-Event-Detection-Response', '0 9-17 * * 1-5'),
('Off-Hours-Intensive-Scan-Low-Impact-Comprehensive-Assessment', '0 {hour} * * *'),
('Continuous-Monitoring-Zero-Day-Vulnerability-Detection-System', '{min} * * * *'),
('Cloud-Infrastructure-Security-Assessment-AWS-Azure-GCP-Multi-Cloud', '0 {hour} * * *'),
('Container-Security-Scan-Kubernetes-Docker-Image-Vulnerability-Check', '0 {hour} * * {dow}'),
('Database-Security-Audit-SQL-Injection-Data-Exposure-Prevention', '0 {hour} {dom} * *'),
('Network-Perimeter-Scan-Firewall-Configuration-Compliance-Check', '0 {hour} * * *'),
('SSL-TLS-Certificate-Monitoring-Expiration-Vulnerability-Detection', '0 {hour} * * *'),
('DNS-Security-Assessment-Zone-Transfer-Subdomain-Takeover-Check', '0 {hour} * * {dow}'),
('Email-Security-Scan-SPF-DKIM-DMARC-Configuration-Verification', '0 {hour} {dom} * *'),
('Mobile-Application-Security-Testing-iOS-Android-API-Assessment', '0 {hour} * * *'),
('IoT-Device-Security-Scan-Firmware-Vulnerability-Network-Exposure', '0 {hour} * * {dow}'),
('Third-Party-Risk-Assessment-Vendor-Security-Posture-Evaluation', '0 {hour} 1 * *'),
('Incident-Response-Readiness-Security-Control-Effectiveness-Test', '0 {hour} 15 * *'),
('Ransomware-Prevention-Scan-Backup-Integrity-Recovery-Verification', '0 {hour} * * *'),
]
# 随机选择 40-50 个定时任务
num_schedules = random.randint(40, 50)
selected = random.sample(schedule_templates, min(num_schedules, len(schedule_templates)))
# 获取引擎名称映射
cur.execute("SELECT id, name FROM scan_engine WHERE id = ANY(%s)", (engine_ids,))
engine_name_map = {row[0]: row[1] for row in cur.fetchall()}
count = 0
for name_base, cron_template in selected:
name = f'{name_base}-{suffix}-{count:02d}'
cron = cron_template.format(
hour=random.randint(0, 23),
min=random.randint(0, 59),
dow=random.randint(0, 6),
dom=random.randint(1, 28)
)
enabled = random.random() > 0.3 # 70% 启用
# 随机选择 1-3 个引擎
num_engines = random.randint(1, min(3, len(engine_ids)))
selected_engine_ids = random.sample(engine_ids, num_engines)
selected_engine_names = [engine_name_map.get(eid, f'Engine-{eid}') for eid in selected_engine_ids]
# 随机决定关联组织还是目标
if org_ids and target_ids:
if random.random() > 0.5:
org_id = random.choice(org_ids)
target_id = None
else:
org_id = None
target_id = random.choice(target_ids)
elif org_ids:
org_id = random.choice(org_ids)
target_id = None
elif target_ids:
org_id = None
target_id = random.choice(target_ids)
else:
org_id = None
target_id = None
run_count = random.randint(0, 200)
has_run = random.random() > 0.2 # 80% 已运行过
cur.execute("""
INSERT INTO scheduled_scan (
name, engine_ids, engine_names, yaml_configuration, organization_id, target_id, cron_expression, is_enabled,
run_count, last_run_time, next_run_time, created_at, updated_at
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW() - INTERVAL '%s days', NOW())
ON CONFLICT DO NOTHING
""", (
name, selected_engine_ids, json.dumps(selected_engine_names), '', org_id, target_id, cron, enabled,
run_count if has_run else 0,
datetime.now() - timedelta(days=random.randint(0, 14), hours=random.randint(0, 23)) if has_run else None,
datetime.now() + timedelta(hours=random.randint(1, 336)) # 最多 2 周后
, random.randint(30, 180)))
count += 1
print(f" ✓ 创建了 {count} 个定时扫描任务\n")
def create_subdomains(self, target_ids: list):
"""创建子域名"""
print("🌐 创建子域名...")
cur = self.conn.cursor()
prefixes = [
# 基础服务
'api', 'admin', 'portal', 'dashboard', 'app', 'mobile', 'staging', 'dev',
'test', 'qa', 'uat', 'beta', 'alpha', 'demo', 'sandbox', 'internal',
'secure', 'auth', 'login', 'sso', 'oauth', 'identity', 'accounts',
'mail', 'smtp', 'imap', 'webmail', 'ftp', 'sftp', 'files', 'storage',
'cdn', 'static', 'assets', 'media', 'db', 'database', 'mysql', 'postgres',
'redis', 'mongo', 'elastic', 'vpn', 'remote', 'gateway', 'proxy',
'monitoring', 'metrics', 'grafana', 'prometheus', 'kibana', 'logs',
'jenkins', 'ci', 'cd', 'gitlab', 'jira', 'confluence', 'kubernetes', 'k8s',
'www', 'www2', 'www3', 'ns1', 'ns2', 'mx', 'mx1', 'mx2', 'autodiscover',
'webdisk', 'cpanel', 'whm', 'webmail2', 'email', 'smtp2', 'pop', 'pop3',
'imap2', 'calendar', 'contacts', 'drive', 'docs', 'sheets', 'slides',
'meet', 'chat', 'teams', 'slack', 'discord', 'zoom', 'video', 'stream',
'blog', 'news', 'press', 'media2', 'images', 'img', 'photos', 'video2',
'shop', 'store', 'cart', 'checkout', 'pay', 'payment', 'billing', 'invoice',
'support', 'help', 'helpdesk', 'ticket', 'tickets', 'status', 'health',
'api-v1', 'api-v2', 'api-v3', 'graphql', 'rest', 'soap', 'rpc', 'grpc',
# 扩展服务
'analytics', 'reporting', 'bi', 'data', 'warehouse', 'etl', 'pipeline',
'ml', 'ai', 'inference', 'training', 'model', 'prediction', 'recommendation',
'search', 'solr', 'elasticsearch', 'opensearch', 'algolia', 'typesense',
'cache', 'memcached', 'varnish', 'haproxy', 'loadbalancer', 'nginx-lb',
'queue', 'rabbitmq', 'kafka', 'pulsar', 'nats', 'activemq', 'sqs',
'workflow', 'airflow', 'prefect', 'dagster', 'temporal', 'conductor',
'registry', 'harbor', 'nexus', 'artifactory', 'pypi', 'npm-registry',
'vault', 'secrets', 'keycloak', 'okta', 'auth0', 'cognito', 'firebase-auth',
'notification', 'push', 'websocket', 'socket', 'realtime', 'pubsub',
'backup', 'archive', 'snapshot', 'restore', 'disaster-recovery', 'dr',
'audit', 'compliance', 'security', 'waf', 'firewall', 'ids', 'ips',
'tracing', 'jaeger', 'zipkin', 'tempo', 'honeycomb', 'lightstep',
'config', 'consul', 'etcd', 'zookeeper', 'nacos', 'apollo-config',
'service-mesh', 'istio', 'linkerd', 'envoy', 'traefik', 'kong',
]
# 二级前缀，用于生成更复杂的子域名
secondary_prefixes = ['', 'prod-', 'dev-', 'staging-', 'test-', 'int-', 'ext-', 'us-', 'eu-', 'ap-',
'us-east-', 'us-west-', 'eu-central-', 'ap-southeast-', 'ap-northeast-',
'primary-', 'secondary-', 'backup-', 'dr-', 'canary-', 'blue-', 'green-']
# 获取域名目标
cur.execute("SELECT id, name FROM target WHERE type = 'domain' AND deleted_at IS NULL")
domain_targets = cur.fetchall()
count = 0
batch_data = []
for target_id, target_name in domain_targets:
# 每个目标随机 80-150 个子域名
num = random.randint(80, 150)
selected = random.sample(prefixes, min(num, len(prefixes)))
for prefix in selected:
# 随机添加二级前缀
sec_prefix = random.choice(secondary_prefixes) if random.random() > 0.7 else ''
subdomain_name = f'{sec_prefix}{prefix}.{target_name}'
days_ago = random.randint(0, 90)
batch_data.append((subdomain_name, target_id, days_ago))
count += 1
# 批量插入
if batch_data:
execute_values(cur, """
INSERT INTO subdomain (name, target_id, created_at)
VALUES %s
ON CONFLICT DO NOTHING
""", batch_data, template="(%s, %s, NOW() - INTERVAL '%s days')")
print(f" ✓ 创建了 {count} 个子域名\n")
def create_websites(self, target_ids: list) -> list:
"""创建网站"""
print("🌍 创建网站...")
cur = self.conn.cursor()
titles = [
'Enterprise Resource Planning System - Comprehensive Dashboard | Acme Corporation International Global Operations Management Portal v3.2.1 - Integrated Business Process Automation and Real-time Analytics Platform for Enterprise-wide Resource Optimization',
'Customer Relationship Management Platform - Secure Login Portal | Multi-Factor Authentication Enabled - Advanced Customer Data Analytics and Sales Pipeline Management System with AI-Powered Insights and Predictive Modeling',
'Human Resources Information System - Employee Self Service Portal v3.2.1 | Comprehensive Payroll Benefits Time-Off Management - Performance Review Talent Acquisition Onboarding Workflow Automation Platform',
'Supply Chain Management - Global Logistics Tracking Dashboard | Real-time Updates - Worldwide Distribution Network Monitor with Predictive Analytics Inventory Optimization and Supplier Relationship Management',
'Business Intelligence Analytics - Executive Summary Report Generator | Advanced Data Visualization Decision Support System - Machine Learning Powered Predictive Analytics and Custom Dashboard Builder',
'Content Management System - Admin Panel | Headless CMS API Gateway - Multi-tenant Enterprise Publishing Platform with Workflow Automation Digital Asset Management and Multi-language Support',
'Project Management Collaboration Tools - Team Workspace | Agile Board - Sprint Planning Resource Allocation Time Tracking Budget Management Gantt Charts Kanban Boards and Team Communication Hub',
'E-Commerce Platform - Product Catalog Management | Inventory Control - Order Processing Fulfillment System with Multi-channel Sales Integration Payment Gateway and Customer Analytics Dashboard',
'Financial Trading Platform - Real-time Market Data Dashboard | Portfolio Management Risk Analysis System - Algorithmic Trading Support Technical Analysis Tools and Regulatory Compliance Reporting',
'Healthcare Patient Management System - Electronic Health Records | HIPAA Compliant Medical Information Portal - Appointment Scheduling Prescription Management Lab Results Integration and Telemedicine Support',
]
webservers = ['nginx/1.24.0', 'nginx/1.25.3', 'nginx/1.26.0', 'Apache/2.4.57', 'Apache/2.4.58', 'Apache/2.4.59',
'Microsoft-IIS/10.0', 'Microsoft-IIS/8.5', 'Microsoft-IIS/7.5', 'cloudflare',
'gunicorn/21.2.0', 'gunicorn/22.0.0', 'gunicorn/23.0.0', 'uvicorn/0.24.0', 'uvicorn/0.25.0',
'Caddy/2.7.5', 'Caddy/2.8.0', 'LiteSpeed', 'LiteSpeed/6.1', 'OpenResty/1.21.4', 'OpenResty/1.25.3',
'Tomcat/10.1.15', 'Tomcat/9.0.83', 'Jetty/11.0.18', 'Jetty/12.0.5', 'WildFly/30.0.0',
'Kestrel', 'Puma/6.4.0', 'Unicorn/6.1.0', 'Passenger/6.0.18', 'Waitress/2.1.2',
'Hypercorn/0.16.0', 'Daphne/4.0.0', 'Twisted/23.10.0', 'CherryPy/18.9.0']
tech_stacks = [
['React 18.2.0', 'React Router 6.21', 'Redux Toolkit 2.0', 'RTK Query', 'Node.js 20.10 LTS', 'Express 4.18.2', 'MongoDB 7.0.4', 'Mongoose 8.0', 'Redis 7.2.3', 'Bull Queue 4.12', 'Nginx 1.25.3', 'Docker 24.0', 'Kubernetes 1.28.4', 'Helm 3.13', 'Prometheus 2.48', 'Grafana 10.2'],
['Vue.js 3.4.5', 'Vuex 4.1', 'Vue Router 4.2', 'Pinia 2.1', 'Nuxt 3.9.0', 'Django 5.0.1', 'Django REST Framework 3.14', 'PostgreSQL 16.1', 'Celery 5.3.6', 'RabbitMQ 3.12.10', 'Gunicorn 21.2', 'Nginx 1.25', 'Docker Compose', 'Prometheus', 'Grafana', 'Sentry'],
['Angular 17.1.0', 'NgRx 17.0', 'RxJS 7.8', 'Angular Material 17', 'Spring Boot 3.2.1', 'Spring Security 6.2', 'Spring Data JPA', 'MySQL 8.2.0', 'Elasticsearch 8.11.3', 'Apache Kafka 3.6.1', 'Grafana 10.2', 'Jenkins 2.426', 'SonarQube 10.3', 'JUnit 5.10', 'Mockito 5.8'],
['Next.js 14.0.4', 'React 18.2', 'TypeScript 5.3', 'Tailwind CSS 3.4', 'FastAPI 0.109.0', 'Pydantic 2.5', 'SQLAlchemy 2.0', 'Redis 7.2', 'PostgreSQL 16', 'Docker 24.0', 'Kubernetes 1.28', 'Istio 1.20', 'ArgoCD 2.9', 'Prometheus', 'Grafana', 'Jaeger'],
['Svelte 4.2.8', 'SvelteKit 2.0.6', 'TypeScript 5.3', 'Tailwind CSS 3.4', 'Go 1.21.5', 'Gin 1.9', 'GORM 1.25', 'CockroachDB 23.2', 'NATS 2.10.7', 'Traefik 3.0', 'Consul 1.17', 'Vault 1.15', 'Terraform 1.6', 'Prometheus', 'Grafana', 'Loki'],
['React 18.2.0', 'NestJS 10.3.0', 'TypeORM 0.3.17', 'GraphQL 16.8', 'Apollo Server 4.10', 'PostgreSQL 16.1', 'Bull 4.12', 'Redis 7.2.3', 'Swagger 7.1', 'Jest 29.7', 'Supertest 6.3', 'Docker', 'Kubernetes', 'Helm', 'ArgoCD', 'Datadog'],
['Vue.js 3.4.5', 'Inertia.js 1.0', 'Laravel 10.40', 'PHP 8.3', 'MySQL 8.2', 'Redis 7.2', 'Laravel Horizon 5.21', 'Laravel Telescope', 'Nginx 1.25', 'Vite 5.0', 'PHPUnit 10.5', 'Pest 2.28', 'Docker', 'GitHub Actions', 'Sentry', 'New Relic'],
['Angular 17.1', 'NgRx 17.0', '.NET 8.0', 'Entity Framework Core 8.0', 'ASP.NET Core 8.0', 'SQL Server 2022', 'Azure Service Bus', 'Azure Functions', 'IIS 10', 'SignalR 8.0', 'xUnit 2.6', 'Moq 4.20', 'Azure DevOps', 'Application Insights', 'Azure Monitor'],
]
# 真实的 body preview 内容
response_bodies = [
'Login - Enterprise Portal',
'Dashboard',
'{"status":"ok","version":"2.4.1","environment":"production","timestamp":"2024-12-22T10:30:00Z","services":{"database":"healthy","cache":"healthy","queue":"healthy"},"uptime":864000}',
'403 Forbidden
403 Forbidden
You don\'t have permission to access this resource. Please contact the administrator if you believe this is an error.
nginx/1.24.0',
'系统维护中
系统正在维护中
预计恢复时间：2024-12-23 08:00
',
'{"error":"Unauthorized","message":"Invalid or expired authentication token. Please login again.","code":"AUTH_001","timestamp":"2024-12-22T15:45:30.123Z","path":"/api/v1/users/profile"}',
'Welcome to nginx!
Welcome to nginx!
If you see this page, the nginx web server is successfully installed and working.
',
'500Internal Server ErrorAn unexpected error occurred while processing your request. Please try again later or contact support.req_abc123xyz789',
'Redirecting...
Redirecting to login page...
Click here if not redirected',
'{"data":{"user":{"id":12345,"username":"admin","email":"admin@example.com","role":"administrator","lastLogin":"2024-12-21T18:30:00Z","permissions":["read","write","delete","admin"]},"token":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."}}',
'API Documentation - Swagger UI',
'{"openapi":"3.0.3","info":{"title":"Enterprise API","description":"RESTful API for enterprise resource management","version":"1.0.0","contact":{"email":"api-support@example.com"}},"servers":[{"url":"https://api.example.com/v1"}]}',
'404 Not Found
404
Page not found
',
'PING OK - Packet loss = 0%, RTA = 0.45 ms|rta=0.450000ms;100.000000;500.000000;0.000000 pl=0%;20;60;0',
'{"metrics":{"requests_total":1234567,"requests_per_second":450.5,"avg_response_time_ms":23.4,"error_rate":0.02,"active_connections":1250,"memory_usage_mb":2048,"cpu_usage_percent":45.6}}',
'Under Construction