# 安装 Trivy
curl -sfL https://raw.githubusercontent.com/aquasecurity/trivy/main/contrib/install.sh | sh -s -- -b /usr/local/bin

# 扫描运行中的容器
trivy image --severity HIGH,CRITICAL your-image:tag

# 扫描宿主机包
trivy fs --security-checks vuln /

#!/bin/bash
# 每日凌晨 2 点运行，扫描所有运行中容器
REPORT_DATE=$(date +%Y-%m-%d)
LOG_FILE="/var/log/security/container-scan-${REPORT_DATE}.log"

echo "=== 容器漏洞扫描开始: $(date) ===" >> "$LOG_FILE"

docker ps --format '{{.Names}}' | while read -r container; do
    echo "扫描容器: $container" >> "$LOG_FILE"
    trivy image \
        --severity HIGH,CRITICAL \
        --format json \
        "$(docker inspect --format='{{.Config.Image}}' "$container")" 2>/dev/null \
        | jq '.Results[]?.Vulnerabilities // [] | length' >> "$LOG_FILE"
done

echo "=== 扫描完成: $(date) ===" >> "$LOG_FILE"

0 2 * * * /opt/security/scan-containers.sh

#!/bin/bash
# /opt/security/scan-system.sh
# 每周日执行系统级漏洞扫描

apt list --upgradable 2>/dev/null | grep -v "^Listing" > /tmp/upgradable-packages.txt

# 使用 Trivy 扫描操作系统包
trivy fs --security-checks vuln \
    --scanners vuln \
    --severity CRITICAL,HIGH \
    / 2>/dev/null | jq -r '.Results[]?.Vulnerabilities[]? | 
    "\(.Severity) [\(.VulnerabilityID)] \(.Title)"' > /tmp/critical-vulns.txt

# 如果有严重漏洞，发送告警
if [ -s /tmp/critical-vulns.txt ]; then
    curl -X POST "https://api.telegram.org/bot${TG_BOT_TOKEN}/sendMessage" \
        -d "chat_id=${TG_CHAT_ID}" \
        -d "text=*⚠️ 严重漏洞发现!*$(cat /tmp/critical-vulns.txt | head -5)" \
        -d "parse_mode=Markdown"
fi

# 创建专用安全分析模型
cat > /etc/ollama/modelfiles/vuln-analyzer << 'EOF'
FROM qwen2.5:7b

SYSTEM """你是一个专业的安全分析师。你的职责是：
1. 分析 CVE 漏洞详情，评估实际风险等级
2. 判断漏洞是否影响当前系统环境
3. 提供具体的修复建议和优先级排序
4. 识别漏洞之间的关联性（如多个低危漏洞组合成高危场景）

输出格式为 JSON：
{
  "cve_id": "CVE-2024-XXXX",
  "actual_risk": "critical|high|medium|low",
  "impact_analysis": "该漏洞在当前环境中的实际影响",
  "fix_priority": 1,
  "fix_steps": ["具体修复步骤"],
  "workaround": "临时缓解措施（如有）",
  "related_vulns": ["相关漏洞列表"]
}"""
EOF

ollama create vuln-analyzer -f /etc/ollama/modelfiles/vuln-analyzer

#!/usr/bin/env python3
"""
AI 漏洞分析引擎
自动分析 Trivy 扫描结果，生成可操作的修复建议
"""

import json
import subprocess
import logging
from datetime import datetime
from pathlib import Path
from typing import List, Dict

try:
    import ollama
except ImportError:
    print("请安装 ollama Python 库: pip install ollama")
    exit(1)

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger("vuln-analyzer")


class VulnAnalyzer:
    """漏洞分析器"""
    
    def __init__(self):
        self.model = "vuln-analyzer"
    
    def scan_trivy(self, target: str = "/") -> List[Dict]:
        """运行 Trivy 扫描并返回结构化结果"""
        try:
            result = subprocess.run(
                ["trivy", "fs", "--security-checks", "vuln",
                 "--severity", "HIGH,CRITICAL",
                 "--format", "json", target],
                capture_output=True, text=True, timeout=300
            )
            data = json.loads(result.stdout)
            
            vulnerabilities = []
            for item in data.get("Results", []):
                for vuln in item.get("Vulnerabilities", []):
                    vulnerabilities.append({
                        "vulnerability_id": vuln["VulnerabilityID"],
                        "package": vuln.get("PkgName", ""),
                        "installed_version": vuln.get("InstalledVersion", ""),
                        "fixed_version": vuln.get("FixedVersion", "N/A"),
                        "severity": vuln["Severity"],
                        "title": vuln["Title"],
                        "description": vuln.get("Description", ""),
                        "cvss_score": vuln.get("CVSS", {}).get("nvd", {}).get("v3", {}).get("vectorString", ""),
                        "references": vuln.get("References", []),
                    })
            
            return vulnerabilities
        
        except Exception as e:
            logger.error(f"扫描失败: {e}")
            return []
    
    def analyze_vuln(self, vuln: Dict) -> Dict:
        """使用 LLM 分析单个漏洞"""
        prompt = f"""请分析以下安全漏洞：

CVE ID: {vuln['vulnerability_id']}
标题: {vuln['title']}
严重程度: {vuln['severity']}
已安装版本: {vuln['installed_version']}
修复版本: {vuln['fixed_version']}
描述: {vuln['description']}
CVSS: {vuln['cvss_score']}

请分析该漏洞在当前 VPS 环境中的实际影响，并提供修复建议。"""
        
        try:
            response = ollama.generate(
                model=self.model,
                prompt=prompt,
                options={"temperature": 0.2, "num_predict": 500}
            )
            
            output = response.get("response", "")
            # 提取 JSON 部分
            json_start = output.find("{")
            json_end = output.rfind("}") + 1
            if json_start >= 0 and json_end > json_start:
                return json.loads(output[json_start:json_end])
            else:
                return {"error": "Failed to parse LLM response"}
        
        except Exception as e:
            logger.error(f"分析失败: {e}")
            return {"error": str(e)}
    
    def prioritize_fixes(self, analyses: List[Dict]) -> List[Dict]:
        """对漏洞修复进行优先级排序"""
        priority_map = {"CRITICAL": 1, "HIGH": 2, "MEDIUM": 3, "LOW": 4}
        
        sorted_analyses = sorted(
            analyses,
            key=lambda x: priority_map.get(x.get("severity", "LOW"), 4)
        )
        
        for i, analysis in enumerate(sorted_analyses):
            analysis["fix_priority"] = i + 1
        
        return sorted_analyses
    
    def generate_report(self, vulnerabilities: List[Dict]) -> str:
        """生成安全报告"""
        report = f"""
# 🔒 VPS 安全分析报告
**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M')}

## 概览
- 发现漏洞总数: {len(vulnerabilities)}
- 严重(CRITICAL): {sum(1 for v in vulnerabilities if v['severity'] == 'CRITICAL')}
- 高危(HIGH): {sum(1 for v in vulnerabilities if v['severity'] == 'HIGH')}
- 中危(MEDIUM): {sum(1 for v in vulnerabilities if v['severity'] == 'MEDIUM')}
- 低危(LOW): {sum(1 for v in vulnerabilities if v['severity'] == 'LOW')}

## 详细分析
"""
        for vuln in vulnerabilities[:10]:
            report += f"""
### [{vuln['severity']}] {vuln['vulnerability_id']}
- **包**: {vuln['package']}
- **当前版本**: {vuln['installed_version']}
- **修复版本**: {vuln['fixed_version']}
- **描述**: {vuln['title']}
"""
        
        report += """
## 建议操作
1. 立即修复所有 CRITICAL 级别漏洞
2. 24 小时内修复 HIGH 级别漏洞
3. 本周内处理 MEDIUM 级别漏洞
4. 制定 LOW 级别漏洞的修复计划
"""
        
        return report


if __name__ == "__main__":
    analyzer = VulnAnalyzer()
    
    # 扫描
    vulns = analyzer.scan_trivy("/")
    logger.info(f"发现 {len(vulns)} 个漏洞")
    
    # 分析前5个严重漏洞
    critical_vulns = [v for v in vulns if v["severity"] in ["CRITICAL", "HIGH"]]
    for vuln in critical_vulns[:5]:
        analysis = analyzer.analyze_vuln(vuln)
        logger.info(f"分析 {vuln['vulnerability_id']}: {analysis}")
    
    # 生成报告
    report = analyzer.generate_report(critical_vulns)
    print(report)

# 安装 CrowdSec
curl -s https://raw.githubusercontent.com/crowdsecurity/crowdsec/main/scripts/install_crowdsec.sh | bash

# 安装 Nginx 解析器
cscli parsers install crowdsecurity/nginx

# 安装 Linux 解析器
cscli parsers install crowdsecurity/linux

# 安装封禁动作（iptables）
cscli actions install crowdsecurity/whitelists
cscli actions install crowdsecurity/iptables

# 启动并启用
systemctl enable --now crowdsec

#!/usr/bin/env python3
"""
AI 增强的威胁情报分析
分析 CrowdSec 拦截的 IP，判断是否为误报
"""

import json
import requests
import logging
from datetime import datetime

try:
    import ollama
except ImportError:
    print("请安装 ollama: pip install ollama")
    exit(1)

logger = logging.getLogger("threat-analyzer")


class ThreatAnalyzer:
    """威胁情报分析器"""
    
    def __init__(self):
        self.crowdsec_api = "http://localhost:8080"
    
    def get_bans(self, limit: int = 50) -> list:
        """获取最近的封禁记录"""
        try:
            response = requests.get(
                f"{self.crowdsec_api}/v1/decisions/signal",
                headers={"X-Api-Key": "your-api-key"},
                params={"limit": limit}
            )
            return response.json()
        except Exception as e:
            logger.error(f"获取封禁记录失败: {e}")
            return []
    
    def analyze_ip_threat(self, ban_entry: dict) -> dict:
        """分析单个封禁事件的威胁等级"""
        scenario = ban_entry.get("Scenario", "")
        source_ip = ban_entry.get("Value", "")
        country = ban_entry.get("Country", "unknown")
        
        prompt = f"""分析以下安全事件：

- 来源IP: {source_ip}
- 国家: {country}
- 触发场景: {scenario}
- 封禁时间: {datetime.fromtimestamp(ban_entry.get("Timestamp", 0)).isoformat()}

请判断：
1. 这是真实攻击还是误报？
2. 威胁等级（critical/high/medium/low）
3. 建议的封禁时长
4. 是否需要手动审查"""
        
        try:
            response = ollama.generate(
                model="vuln-analyzer",
                prompt=prompt,
                options={"temperature": 0.1}
            )
            
            output = response.get("response", "")
            json_start = output.find("{")
            json_end = output.rfind("}") + 1
            if json_start >= 0 and json_end > json_start:
                return json.loads(output[json_start:json_end])
            
            return {"assessment": "manual_review_needed"}
        
        except Exception as e:
            return {"error": str(e)}
    
    def auto_unban_false_positive(self, ip: str):
        """自动解除误报 IP 的封禁"""
        try:
            # 使用 CrowdSec API 解封
            subprocess.run([
                "cscli", "bouncer", "delete", ip
            ], check=True)
            logger.info(f"已自动解封误报 IP: {ip}")
        except Exception as e:
            logger.error(f"解封失败: {e}")


# 每日运行威胁分析
if __name__ == "__main__":
    analyzer = ThreatAnalyzer()
    bans = analyzer.get_bans(limit=100)
    
    for ban in bans[:20]:
        analysis = analyzer.analyze_ip_threat(ban)
        logger.info(f"分析结果: {analysis}")

name: LocalCustomDetection
description: "自定义 VPS 安全检测规则"

filters:
  - Alert.Type == "ban" && Alert.GetSource().Country == "CN"

whens:
  - event.After("10m")

process:
  - name: ai_threat_intel
    enabled: true
    priority: 100
    args:
      model: "vuln-analyzer"
      action: "analyze"

#!/usr/bin/env python3
"""
智能补丁管理器
评估补丁风险并生成安全的更新计划
"""

import subprocess
import json
import logging
from datetime import datetime

try:
    import ollama
except ImportError:
    exit(1)

logger = logging.getLogger("patch-manager")


class PatchManager:
    """补丁管理器"""
    
    def __init__(self):
        self.model = "vuln-analyzer"
    
    def get_upgradable_packages(self) -> list:
        """获取可更新的包列表"""
        result = subprocess.run(
            ["apt", "list", "--upgradable"],
            capture_output=True, text=True
        )
        packages = []
        for line in result.stdout.splitlines():
            if "upgradable" in line:
                pkg_name = line.split("/")[0].split(":")[-1]
                current_ver = line.split(" ")[-1]
                packages.append({"name": pkg_name, "current_version": current_ver})
        return packages
    
    def assess_patch_risk(self, package: dict) -> dict:
        """评估补丁风险"""
        prompt = f"""评估以下软件包升级的风险：

包名: {package['name']}
当前版本: {package['current_version']}
服务器环境: Ubuntu/Debian VPS, Docker 容器化应用

请分析：
1. 升级此包是否可能导致服务中断？
2. 是否有已知的兼容性问题？
3. 推荐的升级策略（立即/计划/跳过）
4. 升级前的备份建议"""
        
        try:
            response = ollama.generate(
                model=self.model,
                prompt=prompt,
                options={"temperature": 0.1, "num_predict": 300}
            )
            
            output = response.get("response", "")
            json_start = output.find("{")
            json_end = output.rfind("}") + 1
            if json_start >= 0 and json_end > json_start:
                return json.loads(output[json_start:json_end])
            
            return {"risk_level": "unknown"}
        
        except Exception as e:
            return {"error": str(e)}
    
    def safe_update(self, packages: list, dry_run: bool = True):
        """执行安全更新"""
        if dry_run:
            logger.info("干跑模式 - 不实际执行更新")
            cmd = ["apt", "upgrade", "-s"]
        else:
            logger.info("执行实际更新")
            cmd = ["sudo", "apt", "upgrade", "-y"]
        
        result = subprocess.run(cmd, capture_output=True, text=True)
        logger.info(result.stdout)
        if result.stderr:
            logger.warning(result.stderr)
        
        return result.returncode == 0


if __name__ == "__main__":
    manager = PatchManager()
    
    # 获取可更新包
    packages = manager.get_upgradable_packages()
    logger.info(f"发现 {len(packages)} 个可更新包")
    
    # 评估每个包的风险
    risk_assessments = []
    for pkg in packages[:10]:  # 只评估前10个
        assessment = manager.assess_patch_risk(pkg)
        risk_assessments.append({"package": pkg, "assessment": assessment})
    
    # 生成更新计划
    critical_updates = [a for a in risk_assessments 
                       if a["assessment"].get("recommended_action") == "immediate"]
    
    if critical_updates:
        logger.info(f"需要立即更新: {len(critical_updates)} 个包")
        manager.safe_update([p["package"]["name"] for p in critical_updates], dry_run=False)
    else:
        logger.info("没有需要紧急更新的包")

#!/bin/bash
# /opt/security/auto-update.sh
# 每周三凌晨 3 点自动执行安全更新

set -euo pipefail

LOG_FILE="/var/log/security/auto-update-$(date +%Y%m%d).log"
BACKUP_DIR="/opt/security/backups/$(date +%Y%m%d)"

echo "=== 安全更新开始: $(date) ===" >> "$LOG_FILE"

# 1. 创建系统快照（如果有 LVM）
if lvdisplay /dev/mapper/*-root >/dev/null 2>&1; then
    echo "创建 LVM 快照..." >> "$LOG_FILE"
    lvcreate --snapshot --name update_backup \
        --size 2G /dev/mapper/*-root >> "$LOG_FILE" 2>&1
fi

# 2. 备份关键配置
mkdir -p "$BACKUP_DIR"
cp -r /etc/nginx /etc/ssh /etc/docker "$BACKUP_DIR/" 2>/dev/null
echo "配置文件已备份到 $BACKUP_DIR" >> "$LOG_FILE"

# 3. 运行 AI 风险评估
python3 /opt/security/patch-manager.py >> "$LOG_FILE" 2>&1

# 4. 执行安全更新
apt-get update -qq >> "$LOG_FILE" 2>&1
apt-get upgrade -y --only-upgrade security >> "$LOG_FILE" 2>&1 || true

# 5. 重启必要的服务
systemctl restart docker >> "$LOG_FILE" 2>&1 || true

echo "=== 安全更新完成: $(date) ===" >> "$LOG_FILE"

# 6. 发送更新报告
TELEGRAM_MSG="✅ VPS 安全更新完成\n更新时间: $(date '+%Y-%m-%d %H:%M')\n日志: /var/log/security/auto-update-$(date +%Y%m%d).log"
curl -s -X POST "https://api.telegram.org/bot${TG_BOT_TOKEN}/sendMessage" \
    -d "chat_id=${TG_CHAT_ID}" \
    -d "text=${TELEGRAM_MSG}" >/dev/null 2>&1

#!/usr/bin/env python3
"""
安全事件自动响应框架
根据威胁类型执行对应的响应剧本
"""

import subprocess
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List

logger = logging.getLogger("incident-responder")


class IncidentResponder:
    """事件响应器"""
    
    def __init__(self):
        self.playbooks = {
            "brute_force": self._handle_brute_force,
            "malware_detected": self._handle_malware,
            "port_scan": self._handle_port_scan,
            "data_exfiltration": self._handle_data_exfiltration,
            "privilege_escalation": self._handle_privilege_escalation,
        }
    
    def detect_incident(self, log_entries: List[str]) -> Dict:
        """检测安全事件"""
        # 暴力破解检测
        failed_logins = sum(1 for log in log_entries 
                          if "Failed password" in log or "authentication failure" in log)
        
        if failed_logins > 10:
            return {
                "type": "brute_force",
                "severity": "high",
                "details": f"检测到 {failed_logins} 次失败登录",
                "affected_services": ["sshd"],
            }
        
        # 端口扫描检测
        port_connections = sum(1 for log in log_entries 
                             if "SYN_RECV" in log or "connection refused" in log.lower())
        
        if port_connections > 50:
            return {
                "type": "port_scan",
                "severity": "medium",
                "details": f"检测到 {port_connections} 次可疑连接",
                "affected_services": ["network"],
            }
        
        return None
    
    def execute_playbook(self, incident: Dict) -> bool:
        """执行响应剧本"""
        playbook_func = self.playbooks.get(incident["type"])
        if playbook_func:
            return playbook_func(incident)
        return False
    
    def _handle_brute_force(self, incident: Dict) -> bool:
        """处理暴力破解"""
        logger.info(f"执行暴力破解响应: {incident}")
        
        # 1. 获取攻击 IP
        result = subprocess.run(
            ["journalctl", "-u", "sshd", "--since", "1 hour ago"],
            capture_output=True, text=True
        )
        
        attack_ips = set()
        for line in result.stdout.splitlines():
            if "Failed password" in line:
                parts = line.split()
                for i, part in enumerate(parts):
                    if part == "from" and i + 1 < len(parts):
                        ip = parts[i + 1]
                        attack_ips.add(ip)
        
        # 2. 封禁 IP
        for ip in attack_ips:
            subprocess.run(["iptables", "-A", "INPUT", "-s", ip, "-j", "DROP"],
                         capture_output=True)
            logger.info(f"已封禁 IP: {ip}")
        
        # 3. 通知
        self._notify(f"🛡️ 暴力破解攻击已阻断\n封禁 {len(attack_ips)} 个 IP")
        
        return True
    
    def _handle_malware(self, incident: Dict) -> bool:
        """处理恶意软件"""
        logger.info("执行恶意软件响应")
        
        # 1. 隔离受影响的服务
        subprocess.run(["systemctl", "stop", "docker"], capture_output=True)
        
        # 2. 扫描系统
        subprocess.run(["clamscan", "-r", "/"], capture_output=True)
        
        # 3. 通知管理员
        self._notify("🚨 检测到恶意软件！系统已隔离")
        
        return True
    
    def _handle_port_scan(self, incident: Dict) -> bool:
        """处理端口扫描"""
        logger.info("执行端口扫描响应")
        
        # 1. 检查开放端口
        result = subprocess.run(["ss", "-tlnp"], capture_output=True, text=True)
        open_ports = result.stdout.strip().split("\n")[1:]
        
        # 2. 确认不必要的端口并关闭
        allowed_ports = {"22", "80", "443", "8080"}
        for port_line in open_ports:
            port = port_line.split(":")[-1].strip()
            if port not in allowed_ports:
                logger.warning(f"发现非标准开放端口: {port}")
        
        self._notify(f"🔍 端口扫描检测完成\n发现 {len(open_ports)} 个开放端口")
        return True
    
    def _handle_data_exfiltration(self, incident: Dict) -> bool:
        """处理数据外泄"""
        logger.info("🚨 执行数据外泄响应")
        
        # 1. 检查异常网络流量
        result = subprocess.run(
            ["ss", "-tunap"],
            capture_output=True, text=True
        )
        
        # 2. 检查异常文件访问
        subprocess.run(["find", "/", "-mtime", "-1", "-type", "f", "-size", "+100M"],
                      capture_output=True)
        
        # 3. 紧急通知
        self._notify("🚨🚨 疑似数据外泄！请立即检查服务器状态")
        return True
    
    def _handle_privilege_escalation(self, incident: Dict) -> bool:
        """处理权限提升"""
        logger.info("🚨 执行权限提升响应")
        
        # 1. 检查 sudo 使用记录
        subprocess.run(["last", "-f", "/var/log/wtmp"], capture_output=True)
        
        # 2. 检查异常 root 会话
        subprocess.run(["w", "-h"], capture_output=True)
        
        # 3. 紧急通知
        self._notify("🚨🚨 检测到异常权限提升活动！")
        return True
    
    def _notify(self, message: str):
        """发送 Telegram 通知"""
        try:
            subprocess.run([
                "curl", "-s", "-X", "POST",
                f"https://api.telegram.org/bot{os.environ.get('TG_BOT_TOKEN', '')}/sendMessage",
                "-d", f"chat_id={os.environ.get('TG_CHAT_ID', '')}",
                "-d", f"text={message}",
                "-d", "parse_mode=Markdown",
            ], capture_output=True)
        except Exception:
            pass


if __name__ == "__main__":
    import os
    
    responder = IncidentResponder()
    
    # 读取最近的系统日志
    result = subprocess.run(
        ["journalctl", "-u", "sshd", "-u", "nginx", "--since", "1 hour ago"],
        capture_output=True, text=True
    )
    
    log_entries = result.stdout.splitlines()
    
    # 检测事件
    incident = responder.detect_incident(log_entries)
    
    if incident:
        logger.info(f"检测到安全事件: {incident['type']}")
        responder.execute_playbook(incident)
    else:
        logger.info("未检测到安全事件")

#!/usr/bin/env python3
"""
生成 AI 安全周报
汇总一周的安全事件、漏洞修复和威胁情报
"""

import subprocess
import json
import logging
from datetime import datetime, timedelta

try:
    import ollama
except ImportError:
    exit(1)

logger = logging.getLogger("weekly-report")


def generate_weekly_report():
    """生成周报"""
    
    # 收集一周的安全数据
    week_start = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
    
    # 1. 漏洞扫描结果
    scan_result = subprocess.run(
        ["ls", "/var/log/security/container-scan-*"],
        capture_output=True, text=True
    )
    scan_files = scan_result.stdout.strip().split("\n")
    
    # 2. CrowdSec 统计
    cs_stats = subprocess.run(
        ["cscli", "metrics"],
        capture_output=True, text=True
    )
    
    # 3. 封禁记录
    bans = subprocess.run(
        ["cscli", "stories", "list", "--limit", "50"],
        capture_output=True, text=True
    )
    
    # 构建报告数据
    report_data = {
        "week_start": week_start,
        "scan_files": len(scan_files),
        "crowdsec_stats": cs_stats.stdout,
        "bans": bans.stdout[:2000],
    }
    
    # 使用 LLM 生成总结
    prompt = f"""根据以下 VPS 安全数据，生成一份简洁的安全周报：

{json.dumps(report_data, indent=2)}

请用中文输出，包含：
1. 本周安全概览（一句话总结）
2. 主要威胁分析
3. 漏洞修复进度
4. 下周安全建议
5. 安全评分（0-100）"""
    
    response = ollama.generate(
        model="vuln-analyzer",
        prompt=prompt,
        options={"temperature": 0.3}
    )
    
    report = response.get("response", "")
    
    # 发送到 Telegram
    subprocess.run([
        "curl", "-s", "-X", "POST",
        f"https://api.telegram.org/bot{os.environ.get('TG_BOT_TOKEN')}/sendMessage",
        "-d", f"chat_id={os.environ.get('TG_CHAT_ID')}",
        "-d", f"text=*📊 本周安全周报*\n\n{report}",
        "-d", "parse_mode=Markdown",
    ])
    
    return report


if __name__ == "__main__":
    import os
    report = generate_weekly_report()
    print(report)