# docker-compose.security.yml - 日志采集服务
version: '3.8'
services:
  loki:
    image: grafana/loki:3.0
    ports:
      - "3100:3100"
    volumes:
      - ./loki/config.yaml:/etc/loki/local-config.yaml
      - loki-data:/loki
    command: -config.file=/etc/loki/local-config.yaml

  promtail:
    image: grafana/promtail:3.0
    volumes:
      - /var/log:/var/log
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
      - ./promtail/config.yaml:/etc/promtail/config.yaml
    command: -config.file=/etc/promtail/config.yaml

  # AI 日志分析代理
  security-agent:
    build: ./security-agent
    environment:
      - LLM_ENDPOINT=http://ollama:11434
      - MODEL_NAME=mistral
      - LOGI_ENDPOINT=http://loki:3100
    volumes:
      - ./security-agent/rules:/rules
    depends_on:
      - loki
      - ollama

#!/bin/bash
# security-agent/scripts/config_snapshot.sh
# 采集 VPS 关键配置文件用于 AI 审计

SNAPSHOT_DIR="/var/security/snapshots/$(date +%Y%m%d_%H%M%S)"
mkdir -p "$SNAPSHOT_DIR"

echo "[*] 采集系统配置..."
# 用户与权限
cp /etc/passwd "$SNAPSHOT_DIR/"
cp /etc/shadow "$SNAPSHOT_DIR/" 2>/dev/null || true
cp /etc/sudoers "$SNAPSHOT_DIR/"
getent group sudo wheel docker admin > "$SNAPSHOT_DIR/groups.txt" 2>/dev/null

# SSH 配置
ssh-keygen -lf /etc/ssh/sshd_config 2>/dev/null
cp /etc/ssh/sshd_config "$SNAPSHOT_DIR/"
ls -la /root/.ssh/ > "$SNAPSHOT_DIR/root_ssh_keys.txt" 2>/dev/null

# 防火墙规则
iptables-save > "$SNAPSHOT_DIR/iptables_rules.txt"
ufw status verbose > "$SNAPSHOT_DIR/ufw_status.txt" 2>/dev/null

# Docker 安全配置
docker ps --format '{{.Names}}: {{.Image}} (ports: {{.Ports}})' \
  > "$SNAPSHOT_DIR/docker_containers.txt"

# 检查暴露的端口
ss -tlnp > "$SNAPSHOT_DIR/listening_ports.txt"

# 定时任务
crontab -l > "$SNAPSHOT_DIR/crontab.txt" 2>/dev/null || true

# 环境变量敏感信息检查
printenv > "$SNAPSHOT_DIR/environment_vars.txt" 2>/dev/null

echo "[+] 配置快照完成: $SNAPSHOT_DIR"

# security-agent/collector/process_monitor.py
"""实时监控进程和网络连接，检测异常行为"""
import psutil
import socket
from datetime import datetime

class ProcessMonitor:
    def __init__(self):
        self.baseline_procs = {}  # 基线进程快照
        self.alert_thresholds = {
            'cpu_percent': 90,
            'memory_percent': 85,
            'open_files': 500,
        }

    def collect_snapshot(self):
        """采集当前进程状态"""
        snapshot = []
        for proc in psutil.process_iter(['pid', 'name', 'cmdline', 
                                          'cpu_percent', 'memory_percent',
                                          'create_time']):
            info = proc.info
            # 跳过内核线程
            if info['create_time'] == 0:
                continue
            snapshot.append({
                'pid': info['pid'],
                'name': info['name'],
                'cmdline': ' '.join(info['cmdline'] or [])[:200],
                'cpu': info['cpu_percent'] or 0,
                'mem': info['memory_percent'] or 0,
            })
        return snapshot

    def detect_anomalies(self, current, baseline):
        """对比基线，检测异常进程"""
        anomalies = []
        
        # 未知进程
        known_pids = {p['pid'] for p in baseline}
        for proc in current:
            if proc['pid'] not in known_pids:
                anomalies.append({
                    'type': 'unknown_process',
                    'pid': proc['pid'],
                    'name': proc['name'],
                    'cmdline': proc['cmdline'],
                    'severity': 'high',
                })
        
        # 资源异常
        for proc in current:
            if proc['cpu'] > self.alert_thresholds['cpu_percent']:
                anomalies.append({
                    'type': 'high_cpu',
                    'pid': proc['pid'],
                    'name': proc['name'],
                    'cpu': proc['cpu'],
                    'severity': 'medium',
                })
        
        return anomalies

# security-agent/auditor/compliance_checker.py
"""使用 LLM 进行 VPS 配置合规审计"""
import json
from pathlib import Path

class ComplianceChecker:
    def __init__(self, llm_client):
        self.llm = llm_client
        self.checks = self._load_check_rules()

    def _load_check_rules(self):
        """加载安全检查规则"""
        return {
            'ssh_hardening': {
                'name': 'SSH 安全加固',
                'priority': 'critical',
                'items': [
                    {'check': 'PermitRootLogin', 'expected': 'no', 'desc': '禁止 root 远程登录'},
                    {'check': 'PasswordAuthentication', 'expected': 'no', 'desc': '禁用密码认证'},
                    {'check': 'Port', 'expected': 'non_standard', 'desc': '使用非标准端口'},
                    {'check': 'MaxAuthTries', 'expected': '<4', 'desc': '最大认证尝试次数'},
                    {'check': 'X11Forwarding', 'expected': 'no', 'desc': '禁用 X11 转发'},
                ]
            },
            'docker_security': {
                'name': 'Docker 安全配置',
                'priority': 'high',
                'items': [
                    {'check': 'rootless_mode', 'expected': True, 'desc': '使用 Rootless Docker'},
                    {'check': 'no_latest_tag', 'expected': True, 'desc': '不使用 :latest 标签'},
                    {'check': 'healthcheck', 'expected': True, 'desc': '容器健康检查'},
                    {'check': 'readonly_rootfs', 'expected': True, 'desc': '只读根文件系统'},
                ]
            },
            'network_security': {
                'name': '网络安全配置',
                'priority': 'high',
                'items': [
                    {'check': 'firewall_enabled', 'expected': True, 'desc': '防火墙启用'},
                    {'check': 'unnecessary_ports', 'expected': 0, 'desc': '关闭不必要端口'},
                    {'check': 'fail2ban', 'expected': True, 'desc': 'Fail2ban 运行中'},
                ]
            },
        }

    async def audit_ssh_config(self, config_content: str) -> dict:
        """审计 SSH 配置文件"""
        prompt = f"""你是一个专业的 Linux 安全审计员。请分析以下 SSH 配置，找出所有安全问题和合规项。

配置文件内容：

# security-agent/detector/anomaly_detector.py
"""使用 LLM 进行日志异常检测和威胁分析"""
import re
from datetime import datetime, timedelta

class AnomalyDetector:
    def __init__(self, llm_client):
        self.llm = llm_client
        self.threat_patterns = self._load_threat_signatures()

    def _load_threat_signatures(self):
        """加载已知攻击模式签名"""
        return {
            'brute_force': {
                'pattern': r'Failed password for .* from (\d+\.\d+\.\d+\.\d+)',
                'threshold': 5,
                'window_minutes': 10,
                'severity': 'high',
            },
            'port_scan': {
                'pattern': r'Connection closed by .* port \d+.*preauth',
                'threshold': 20,
                'window_minutes': 5,
                'severity': 'medium',
            },
            'sql_injection': {
                'pattern': r"(?i)(union\s+select|or\s+1=1|drop\s+table|;\s*delete)",
                'threshold': 1,
                'window_minutes': 60,
                'severity': 'critical',
            },
            'directory_traversal': {
                'pattern': r'(\.\./){2,}',
                'threshold': 3,
                'window_minutes': 5,
                'severity': 'high',
            },
        }

    async def analyze_auth_logs(self, log_lines: list[str]) -> dict:
        """分析认证日志，检测暴力破解"""
        ip_counts = {}
        for line in log_lines:
            match = re.search(r'from\s+(\d+\.\d+\.\d+\.\d+)', line)
            if match:
                ip = match.group(1)
                ip_counts[ip] = ip_counts.get(ip, 0) + 1

        threats = []
        for ip, count in ip_counts.items():
            if count >= 5:
                threats.append({
                    'type': 'potential_brute_force',
                    'source_ip': ip,
                    'attempt_count': count,
                    'severity': 'high' if count > 20 else 'medium',
                    'recommendation': f'建议封禁 IP {ip} 或启用 fail2ban',
                })

        # 使用 LLM 进行深度分析
        if threats:
            llm_prompt = f"""分析以下安全事件，判断是否为真实攻击还是误报：

事件列表：
{json.dumps(threats, indent=2)}

请提供：
1. 攻击类型判定
2. 风险等级评估
3. 推荐的响应措施
4. 是否需要上报威胁情报平台

返回结构化分析结果。"""
            analysis = await self.llm.generate(llm_prompt, model='mistral')
            threats[0]['llm_analysis'] = analysis

        return {'threats': threats, 'total_events': len(log_lines)}

    async def analyze_web_logs(self, access_log_path: str) -> dict:
        """分析 Web 访问日志，检测 Web 攻击"""
        with open(access_log_path) as f:
            lines = f.readlines()[-1000:]  # 最近 1000 行

        suspicious_requests = []
        for line in lines:
            for pattern_name, config in self.threat_patterns.items():
                if re.search(config['pattern'], line):
                    suspicious_requests.append({
                        'pattern': pattern_name,
                        'line': line.strip()[:200],
                    })

        if suspicious_requests:
            llm_prompt = f"""检测到以下可疑请求，请分析是否为真实攻击：

{suspicious_requests[:20]}

请分析：
1. 攻击类型和意图
2. 目标系统脆弱性
3. 紧急响应步骤
4. 长期防护建议"""
            analysis = await self.llm.generate(llm_prompt, model='mistral')
            return {'suspicious': suspicious_requests, 'analysis': analysis}

        return {'suspicious': [], 'status': 'clean'}

# security-agent/alerts/notifier.py
"""多渠道安全告警通知"""
import smtplib
import requests
from email.mime.text import MIMEText

class AlertNotifier:
    def __init__(self, config: dict):
        self.config = config
        self.channels = []

    def register_channel(self, channel_type: str, **kwargs):
        """注册告警渠道"""
        if channel_type == 'email':
            self.channels.append(('email', kwargs))
        elif channel_type == 'telegram':
            self.channels.append(('telegram', kwargs))
        elif channel_type == 'webhook':
            self.channels.append(('webhook', kwargs))

    async def send_alert(self, severity: str, title: str, message: str,
                         details: dict = None):
        """发送告警到所有已注册的渠道"""
        alert = {
            'severity': severity,
            'title': title,
            'message': message,
            'timestamp': datetime.now().isoformat(),
            'host': self.config.get('hostname', 'unknown'),
            'details': details,
        }

        for channel, kwargs in self.channels:
            try:
                if channel == 'email':
                    self._send_email(alert, kwargs)
                elif channel == 'telegram':
                    await self._send_telegram(alert, kwargs)
                elif channel == 'webhook':
                    await self._send_webhook(alert, kwargs)
            except Exception as e:
                print(f"[!] 告警发送失败 ({channel}): {e}")

    def _send_email(self, alert: dict, config: dict):
        msg = MIMEText(f"""
VPS 安全告警
{'=' * 40}
严重程度: {alert['severity'].upper()}
标题: {alert['title']}
时间: {alert['timestamp']}
主机: {alert['host']}

详情:
{alert['message']}
        """, 'plain', 'utf-8')
        msg['Subject'] = f"[{alert['severity'].upper()}] {alert['title']}"
        msg['From'] = config['from']
        msg['To'] = config['to']

        with smtplib.SMTP(config['smtp_server'], config['smtp_port']) as server:
            server.starttls()
            server.login(config['username'], config['password'])
            server.send_message(msg)

    async def _send_telegram(self, alert: dict, config: dict):
        severity_emoji = {
            'critical': '🔴',
            'high': '🟠',
            'medium': '🟡',
            'low': '🔵',
        }
        emoji = severity_emoji.get(alert['severity'], '⚪')
        text = f"{emoji} *[{alert['severity'].upper()}]*\n"
        text += f"{alert['title']}\n\n"
        text += f"{alert['message']}\n\n"
        text += f"🖥 {alert['host']} | 🕐 {alert['timestamp']}"

        url = f"https://api.telegram.org/bot{config['bot_token']}/sendMessage"
        await requests.post(url, json={
            'chat_id': config['chat_id'],
            'text': text,
            'parse_mode': 'Markdown',
        })

# security-agent/remediation/auto_fixer.py
"""基于 LLM 分析结果的自动安全修复"""
import subprocess
import shlex

class AutoFixer:
    def __init__(self, dry_run: bool = True):
        self.dry_run = dry_run
        self.fix_history = []

    def apply_ssh_fix(self, issue: dict) -> dict:
        """自动修复 SSH 安全问题"""
        fix_actions = []

        if issue.get('rule') == 'PermitRootLogin':
            fix_actions.append({
                'action': 'edit_sshd_config',
                'command': 'sed -i "s/^PermitRootLogin.*/PermitRootLogin no/" /etc/ssh/sshd_config',
                'requires_restart': True,
            })

        if issue.get('rule') == 'PasswordAuthentication':
            fix_actions.append({
                'action': 'edit_sshd_config',
                'command': 'sed -i "s/^PasswordAuthentication.*/PasswordAuthentication no/" /etc/ssh/sshd_config',
                'requires_restart': False,
            })

        if issue.get('rule') == 'MaxAuthTries':
            fix_actions.append({
                'action': 'edit_sshd_config',
                'command': 'grep -q "^MaxAuthTries" /etc/ssh/sshd_config && \\\n  sed -i "s/^MaxAuthTries.*/MaxAuthTries 3/" /etc/ssh/sshd_config || \\\n  echo "MaxAuthTries 3" >> /etc/ssh/sshd_config',
                'requires_restart': False,
            })

        for action in fix_actions:
            if self.dry_run:
                print(f"[DRY RUN] {action['command']}")
            else:
                result = subprocess.run(action['command'], shell=True,
                                       capture_output=True, text=True)
                if result.returncode == 0:
                    print(f"[✓] 修复成功: {action['action']}")
                else:
                    print(f"[✗] 修复失败: {result.stderr}")

        return {'actions_applied': len(fix_actions), 'dry_run': self.dry_run}

    def apply_docker_fix(self, issue: dict) -> dict:
        """自动修复 Docker 安全问题"""
        fixes = []

        if issue.get('issue_type') == 'root_socket_mount':
            fixes.append({
                'action': 'remove_docker_socket_mount',
                'description': '移除容器中的 Docker socket 挂载',
                'manual_step': '编辑 docker-compose.yml，删除 volumes 中的 /var/run/docker.sock',
            })

        if issue.get('issue_type') == 'missing_resource_limits':
            fixes.append({
                'action': 'add_resource_limits',
                'template': '''deploy:
  resources:
    limits:
      cpus: '1.0'
      memory: 512M
    reservations:
      cpus: '0.25'
      memory: 128M''',
                'description': '为容器添加资源限制',
            })

        return {'fixes': fixes}

    def generate_fix_report(self, issues: list, fixes: dict) -> str:
        """生成修复报告"""
        report = f"""# VPS 安全修复报告
生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
修复模式: {'只读预览' if self.dry_run else '实际执行'}

## 修复概览
- 发现问题: {len(issues)} 个
- 自动修复: {sum(len(v) for v in fixes.values())} 个
- 需手动处理: {len(issues) - sum(len(v) for v in fixes.values())} 个

## 详细修复记录
"""
        for issue, fix_list in zip(issues, fixes.values()):
            report += f"\n### {issue.get('rule', 'Unknown')}\n"
            for fix in fix_list:
                report += f"- {fix.get('description', fix.get('action'))}\n"

        return report

# 更新系统
sudo apt update && sudo apt upgrade -y

# 安装必要依赖
sudo apt install -y python3 python3-pip docker.io \
  fail2ban ufw jq

# 安装 Ollama（本地 LLM 运行时）
curl -fsSL https://ollama.com/install.sh | sh

# 拉取适合安全审计的小模型
ollama pull mistral:7b

# 克隆安全代理项目
git clone https://github.com/selfvps/ai-security-agent.git
cd ai-security-agent

# 创建配置文件
cat > config.yaml << 'EOF'
llm:
  endpoint: http://localhost:11434
  model: mistral:7b

audit:
  schedule: "0 */6 * * *"  # 每6小时审计一次
  ssh_config: /etc/ssh/sshd_config
  docker_compose_dir: /opt/services

alerts:
  email:
    enabled: true
    smtp_server: smtp.gmail.com
    smtp_port: 587
    from: alert@yourdomain.com
    to: admin@yourdomain.com
  telegram:
    enabled: true
    bot_token: YOUR_BOT_TOKEN
    chat_id: YOUR_CHAT_ID

remediation:
  dry_run: true  # 先开启干跑模式！
  allowed_fixes:
    - ssh_hardening
    - firewall_rules
    - docker_security
EOF

# 添加到 crontab
crontab -e

# 每日凌晨2点执行完整安全审计
0 2 * * * /opt/ai-security-agent/run_audit.sh --report
# 每6小时检查日志异常
0 */6 * * * /opt/ai-security-agent/check_anomalies.sh
# 每周日生成周报
0 9 * * 0 /opt/ai-security-agent/generate_weekly_report.sh

# /etc/fail2ban/jail.local
[DEFAULT]
bantime = 3600
findtime = 600
maxretry = 3
backend = auto

[sshd-ai]
enabled = true
port = ssh
filter = sshd
logpath = /var/log/auth.log
maxretry = 3
bantime = 86400

# AI 增强的自定义过滤规则
[nginx-ai-bot]
enabled = true
port = http,https
filter = nginx-ai-bot
logpath = /var/log/nginx/access.log
maxretry = 10
bantime = 43200

# 查询当前安全评分
$ echo "我的 VPS 安全状况如何？" | ./security-agent/chat

📊 你的 VPS 当前安全评分: 78/100
整体状态: 🟡 需要注意

主要风险:
1. SSH 允许密码认证（中等风险）
2. 2 个 Docker 容器未设置资源限制（低风险）
3. UFW 防火墙部分规则缺失（低风险）

建议优先处理:
→ 禁用 SSH 密码认证，改用密钥登录
→ 为所有容器添加 CPU/内存限制

# 查询最近的威胁事件
$ echo "最近有什么安全威胁？" | ./security-agent/chat

🔒 过去 24 小时安全事件:

[🔴 严重] 检测到来自 185.220.xx.xx 的 SSH 暴力破解
  - 尝试次数: 342 次
  - 源 IP 已加入黑名单
  - 建议: 确认 fail2ban 规则正常

[🟡 警告] Nginx 访问日志中发现 SQL 注入尝试
  - 来源: 45.142.xx.xx
  - 攻击路径: /api/search?q=1' OR 1=1--
  - 已自动拦截

[🔵 信息] 系统证书将在 15 天后过期
  - 证书: *.yourdomain.com
  - 建议: 运行 certbot renew 提前续期