网站速度慢？安全防护弱？EdgeOne免费套餐一次性解决两大痛点

# EdgeOne WAF规则配置示例
waf_rules:
  # SQL注入防护
  - name: "SQL Injection Protection"
    conditions:
      - field: "uri"
        operator: "contains"
        values: ["union select", "drop table", "' or '1'='1"]
      - field: "args"
        operator: "regex"
        values: ["(union.*select)|(select.*from)|(insert.*into)"]
    action: "block"
    log: true
    
  # XSS攻击防护
  - name: "XSS Protection"
    conditions:
      - field: "args"
        operator: "contains"
        values: ["<script>", "javascript:", "onload="]
      - field: "headers"
        operator: "regex"
        values: ["<.*script.*>|javascript:.*"]
    action: "block"
    log: true
    
  # 恶意爬虫防护
  - name: "Bot Protection"
    conditions:
      - field: "user_agent"
        operator: "regex"
        values: ["bot|crawler|spider|scraper"]
      - field: "request_rate"
        operator: "gt"
        values: [100] # 每分钟超过100次请求
    action: "challenge" # 验证码挑战
    
  # 地理位置限制
  - name: "Geo Blocking"
    conditions:
      - field: "country_code"
        operator: "in"
        values: ["XX", "YY"] # 限制特定国家
    action: "block"
    custom_page: "/blocked.html"

import asyncio
import aiohttp
import time
import json
from datetime import datetime

class EdgeOneMonitor:
    def __init__(self, domains, check_interval=300):
        """
        EdgeOne性能监控器
        
        Args:
            domains: 需要监控的域名列表
            check_interval: 检查间隔（秒）
        """
        self.domains = domains
        self.check_interval = check_interval
        self.results = []
        
    async def check_single_domain(self, session, domain):
        """检查单个域名的性能"""
        start_time = time.time()
        
        try:
            async with session.get(f'https://{domain}', 
                                 timeout=aiohttp.ClientTimeout(total=30)) as response:
                await response.text()  # 读取响应内容
                
                return {
                    'domain': domain,
                    'status_code': response.status,
                    'response_time': round((time.time() - start_time) * 1000, 2),
                    'timestamp': datetime.now().isoformat(),
                    'headers': dict(response.headers),
                    'success': True
                }
                
        except Exception as e:
            return {
                'domain': domain,
                'error': str(e),
                'response_time': round((time.time() - start_time) * 1000, 2),
                'timestamp': datetime.now().isoformat(),
                'success': False
            }
    
    async def check_all_domains(self):
        """并发检查所有域名"""
        async with aiohttp.ClientSession() as session:
            tasks = [self.check_single_domain(session, domain) 
                    for domain in self.domains]
            
            results = await asyncio.gather(*tasks)
            return results
    
    def analyze_performance(self, results):
        """分析性能数据"""
        successful_requests = [r for r in results if r['success']]
        
        if not successful_requests:
            return {'error': 'No successful requests'}
        
        response_times = [r['response_time'] for r in successful_requests]
        
        analysis = {
            'total_domains': len(self.domains),
            'successful_requests': len(successful_requests),
            'success_rate': len(successful_requests) / len(self.domains),
            'avg_response_time': sum(response_times) / len(response_times),
            'min_response_time': min(response_times),
            'max_response_time': max(response_times),
            'timestamp': datetime.now().isoformat()
        }
        
        # 检查是否有EdgeOne标识
        for result in successful_requests:
            if 'headers' in result:
                headers = result['headers']
                if 'Server' in headers and 'EdgeOne' in headers['Server']:
                    result['using_edgeone'] = True
                else:
                    result['using_edgeone'] = False
        
        return analysis
    
    async def start_monitoring(self):
        """开始监控循环"""
        print(f"开始监控 {len(self.domains)} 个域名...")
        
        while True:
            try:
                results = await self.check_all_domains()
                analysis = self.analyze_performance(results)
                
                # 保存结果
                self.results.append({
                    'analysis': analysis,
                    'details': results
                })
                
                print(f"[{analysis['timestamp']}] "
                      f"成功率: {analysis['success_rate']:.2%}, "
                      f"平均响应时间: {analysis['avg_response_time']:.2f}ms")
                
                # 检测异常
                if analysis['success_rate'] < 0.95:
                    print("⚠️ 警告：成功率低于95%")
                
                if analysis['avg_response_time'] > 1000:
                    print("⚠️ 警告：平均响应时间超过1秒")
                
                await asyncio.sleep(self.check_interval)
                
            except Exception as e:
                print(f"监控过程中出现错误：{e}")
                await asyncio.sleep(60)  # 出错后等待1分钟再重试

# 使用示例
if __name__ == "__main__":
    domains = [
        "example.com",
        "api.example.com", 
        "cdn.example.com"
    ]
    
    monitor = EdgeOneMonitor(domains, check_interval=300)  # 5分钟检查一次
    asyncio.run(monitor.start_monitoring())