obsguard/
├── app.py                      # 主应用入口，模拟业务API与Web控制台
├── config.py                   # 配置文件
├── models.py                   # SQLAlchemy 数据模型定义
├── requirements.txt            # Python 依赖
├── obsguard.db                 # SQLite 数据库文件（运行后生成）
├── core/
│   ├── __init__.py
│   ├── monitor_agent.py        # 监控数据采集与异常检测
│   ├── baseline_engine.py      # 安全基线检查引擎
│   ├── alert_manager.py        # 告警规则管理与触发
│   └── attack_simulator.py     # 攻防模拟与验证
└── templates/
    └── index.html              # 简易Web控制台

pip install -r requirements.txt

flask>=2.0.0
    flask-sqlalchemy>=3.0.0
    sqlalchemy>=2.0.0
    requests>=2.25.0
    schedule>=1.2.0
    python-dotenv>=1.0.0

python app.py

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    """应用配置"""
    SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secure-key-change-in-prod')
    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///obsguard.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False

    # 监控代理配置
    MONITOR_INTERVAL = 10  # 指标采集间隔（秒）
    BASELINE_CHECK_INTERVAL = 60  # 基线检查间隔（秒）

    # 应用模拟配置
    SIMULATED_NORMAL_LATENCY_MS = (50, 80)  # 正常延迟范围
    SIMULATED_SPIKE_LATENCY_MS = (200, 500)  # 延迟突增范围
    SPIKE_PROBABILITY = 0.05  # 发生延迟突增的概率

    # 告警阈值
    ALERT_LATENCY_THRESHOLD_MS = 100
    ALERT_ERROR_RATE_THRESHOLD = 0.1  # 10%

    # 安全基线配置
    BASELINE_MIN_PASSWORD_LENGTH = 12
    BASELINE_REQUIRE_AUTH_FOR_METRICS = True
    BASELINE_LOG_LEVEL = 'WARNING'

from datetime import datetime
from app import db

class PerformanceMetric(db.Model):
    """性能指标数据模型"""
    id = db.Column(db.Integer, primary_key=True)
    timestamp = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    endpoint = db.Column(db.String(64), default='/order')
    latency_ms = db.Column(db.Float)
    cpu_usage = db.Column(db.Float)  # 模拟CPU使用率百分比
    memory_usage = db.Column(db.Float) # 模拟内存使用率百分比
    request_rate = db.Column(db.Float) # 请求/秒
    error_rate = db.Column(db.Float)   # 错误率

class AlertRule(db.Model):
    """告警规则模型"""
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), nullable=False)
    metric = db.Column(db.String(64), nullable=False)  # 如 'latency_ms'
    condition = db.Column(db.String(16), nullable=False)  # 如 'gt', 'lt'
    threshold = db.Column(db.Float, nullable=False)
    severity = db.Column(db.String(32), default='WARNING')  # WARNING, ERROR, CRITICAL
    is_active = db.Column(db.Boolean, default=True)

class AlertEvent(db.Model):
    """告警事件记录"""
    id = db.Column(db.Integer, primary_key=True)
    timestamp = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    rule_id = db.Column(db.Integer, db.ForeignKey('alert_rule.id'))
    rule = db.relationship('AlertRule')
    metric_value = db.Column(db.Float)
    description = db.Column(db.String(512))
    resolved = db.Column(db.Boolean, default=False)

class AttackLog(db.Model):
    """攻防验证日志"""
    id = db.Column(db.Integer, primary_key=True)
    timestamp = db.Column(db.DateTime, default=datetime.utcnow)
    attack_type = db.Column(db.String(64))  # 如 'metric_injection', 'brute_force'
    target = db.Column(db.String(128))
    payload = db.Column(db.Text, nullable=True)
    success = db.Column(db.Boolean)  # 攻击是否成功绕过防护
    detected_by = db.Column(db.String(128), nullable=True)  # 被哪个模块检测到
    details = db.Column(db.Text, nullable=True)

import time
import random
import threading
from datetime import datetime, timedelta
from app import db
from models import PerformanceMetric
from config import Config

class MonitorAgent:
    """监控代理：负责采集性能指标并进行初步异常检测"""

    def __init__(self):
        self._stop_event = threading.Event()
        self._request_history = []  # 用于计算短期请求率

    def collect_metrics(self):
        """模拟采集应用的性能指标"""
        # 模拟实时请求率（最近10秒）
        current_time = time.time()
        ten_sec_ago = current_time - 10
        self._request_history = [t for t in self._request_history if t > ten_sec_ago]
        req_rate = len(self._request_history) / 10.0

        # 模拟延迟：大部分情况正常，小概率突增（模拟性能瓶颈）
        if random.random() < Config.SPIKE_PROBABILITY:
            latency = random.uniform(*Config.SIMULATED_SPIKE_LATENCY_MS)
            spike_detected = True
        else:
            latency = random.uniform(*Config.SIMULATED_NORMAL_LATENCY_MS)
            spike_detected = False

        # 模拟系统资源使用率，与延迟有一定关联
        cpu = random.uniform(30.0, 80.0) + (latency / 10)
        memory = random.uniform(40.0, 70.0)
        # 模拟错误率，高延迟时错误率可能升高
        error_rate = 0.01 if not spike_detected else random.uniform(0.1, 0.3)

        metric = PerformanceMetric(
            latency_ms=latency,
            cpu_usage=min(cpu, 100.0),  # 确保不超过100%
            memory_usage=memory,
            request_rate=req_rate,
            error_rate=error_rate
        )
        db.session.add(metric)
        db.session.commit()

        # 返回本次采集的数据，用于可能的实时告警
        return {
            'latency_ms': latency,
            'cpu_usage': cpu,
            'error_rate': error_rate,
            'is_spike': spike_detected
        }

    def record_request(self):
        """记录一次请求发生（由外部调用，如app.py处理订单时）"""
        self._request_history.append(time.time())

    def start(self):
        """启动后台采集线程"""
        def run():
            import schedule
            import time
            schedule.every(Config.MONITOR_INTERVAL).seconds.do(self.collect_metrics)
            while not self._stop_event.is_set():
                schedule.run_pending()
                time.sleep(1)

        thread = threading.Thread(target=run, daemon=True)
        thread.start()
        print(f"[MonitorAgent] 已启动，每 {Config.MONITOR_INTERVAL} 秒采集一次指标。")

    def stop(self):
        self._stop_event.set()

from datetime import datetime
from app import db, alert_manager
from models import AlertEvent, AlertRule
from config import Config

class BaselineEngine:
    """安全基线检查引擎：定期验证监控系统配置是否符合安全标准"""

    BASELINE_RULES = [
        {
            'id': 'BL001',
            'name': '指标端点认证检查',
            'check': lambda: Config.BASELINE_REQUIRE_AUTH_FOR_METRICS,
            'desc_success': '指标端点访问要求认证已启用。',
            'desc_fail': '警告：指标端点未配置认证，可能导致数据泄露。'
        },
        {
            'id': 'BL002',
            'name': '告警规则有效性检查',
            'check': self._check_alert_rules,
            'desc_success': '所有活跃告警规则均包含合理的阈值。',
            'desc_fail': '发现无效告警规则（如阈值为空或逻辑错误）。'
        },
        {
            'id': 'BL003',
            'name': '日志敏感信息过滤',
            'check': lambda: Config.BASELINE_LOG_LEVEL in ['WARNING', 'ERROR'],
            'desc_success': f'当前日志级别({Config.BASELINE_LOG_LEVEL})有助于避免信息过载。',
            'desc_fail': f'当前日志级别({Config.BASELINE_LOG_LEVEL})过低，可能记录过多敏感或冗余信息。'
        }
    ]

    def _check_alert_rules(self):
        """检查告警规则是否配置合理（示例：检查是否有阈值异常大的规则）"""
        bizarre_rules = AlertRule.query.filter(
            AlertRule.is_active == True,
            AlertRule.metric == 'latency_ms',
            AlertRule.threshold > 10000  # 不合理的高阈值
        ).all()
        return len(bizarre_rules) == 0

    def run_checks(self):
        """执行所有基线检查"""
        print(f"[BaselineEngine] {datetime.utcnow()}: 开始安全基线检查...")
        for rule in self.BASELINE_RULES:
            try:
                # 动态绑定 self 到类方法
                if rule['id'] == 'BL002':
                    check_result = self._check_alert_rules()
                else:
                    check_result = rule['check']()
            except Exception as e:
                check_result = False
                failure_desc = f"检查执行失败: {e}"

            if check_result:
                print(f"  [PASS] {rule['name']}: {rule['desc_success']}")
            else:
                print(f"  [FAIL] {rule['name']}: {rule['desc_fail']}")
                # 基线检查失败，触发一个告警事件
                alert_desc = f"安全基线违规 - {rule['name']}: {rule.get('desc_fail', failure_desc)}"
                alert_manager.trigger_alert(
                    rule_name=f"基线-{rule['id']}",
                    metric_name='baseline_violation',
                    metric_value=1.0,
                    description=alert_desc,
                    severity='ERROR'
                )

    def start(self):
        """启动定期基线检查（集成到主循环中，由app.py调用）"""
        def run():
            import schedule
            import time
            schedule.every(Config.BASELINE_CHECK_INTERVAL).seconds.do(self.run_checks)
            while True:
                schedule.run_pending()
                time.sleep(1)
        import threading
        thread = threading.Thread(target=run, daemon=True)
        thread.start()
        print(f"[BaselineEngine] 已启动，每 {Config.BASELINE_CHECK_INTERVAL} 秒检查一次安全基线。")

from datetime import datetime, timedelta
from app import db
from models import AlertRule, AlertEvent
from config import Config

class AlertManager:
    """告警管理器：评估指标并触发告警"""

    def __init__(self):
        # 初始化一些默认告警规则
        self._init_default_rules()

    def _init_default_rules(self):
        """创建默认告警规则（如果不存在）"""
        default_rules = [
            ('高延迟告警', 'latency_ms', 'gt', Config.ALERT_LATENCY_THRESHOLD_MS, 'WARNING'),
            ('高错误率告警', 'error_rate', 'gt', Config.ALERT_ERROR_RATE_THRESHOLD, 'ERROR'),
        ]
        for name, metric, cond, thresh, sev in default_rules:
            if not AlertRule.query.filter_by(name=name).first():
                rule = AlertRule(
                    name=name,
                    metric=metric,
                    condition=cond,
                    threshold=thresh,
                    severity=sev
                )
                db.session.add(rule)
        db.session.commit()

    def evaluate_metric(self, metric_name, metric_value):
        """根据规则评估单个指标值"""
        active_rules = AlertRule.query.filter_by(
            metric=metric_name,
            is_active=True
        ).all()

        triggered = []
        for rule in active_rules:
            condition_met = False
            if rule.condition == 'gt':
                condition_met = metric_value > rule.threshold
            elif rule.condition == 'lt':
                condition_met = metric_value < rule.threshold
            elif rule.condition == 'eq':
                condition_met = abs(metric_value - rule.threshold) < 0.001

            if condition_met:
                self.trigger_alert(
                    rule_name=rule.name,
                    metric_name=metric_name,
                    metric_value=metric_value,
                    description=f'{metric_name}={metric_value} 触发规则 "{rule.name}" ({rule.condition} {rule.threshold})',
                    severity=rule.severity,
                    rule_id=rule.id
                )
                triggered.append(rule)
        return triggered

    def trigger_alert(self, rule_name, metric_name, metric_value, description, severity='WARNING', rule_id=None):
        """触发并记录一个告警事件"""
        # 简单防抖动：检查过去1分钟内是否有相同描述的未解决告警
        one_min_ago = datetime.utcnow() - timedelta(minutes=1)
        existing = AlertEvent.query.filter(
            AlertEvent.description.like(f"%{rule_name}%"),
            AlertEvent.timestamp > one_min_ago,
            AlertEvent.resolved == False
        ).first()
        if existing:
            print(f"[AlertManager] 告警防抖动：类似告警 '{rule_name}' 已在近期触发。")
            return

        alert = AlertEvent(
            rule_id=rule_id,
            metric_value=metric_value,
            description=description,
            severity=severity
        )
        db.session.add(alert)
        db.session.commit()
        print(f"[AlertManager] 触发告警 [{severity}]: {description}")

    def get_recent_alerts(self, limit=20, unresolved_only=False):
        """获取最近的告警"""
        query = AlertEvent.query.order_by(AlertEvent.timestamp.desc())
        if unresolved_only:
            query = query.filter_by(resolved=False)
        return query.limit(limit).all()

    def mark_resolved(self, alert_id):
        """标记告警为已解决"""
        alert = AlertEvent.query.get(alert_id)
        if alert:
            alert.resolved = True
            db.session.commit()
            return True
        return False

import random
import string
import time
from datetime import datetime
from app import db, alert_manager
from models import AttackLog

class AttackSimulator:
    """
    攻防验证模块：
    模拟针对监控系统或应用的攻击，并验证安全防护（如基线检查、告警）是否生效。
    """

    def simulate_metric_injection(self):
        """模拟攻击：尝试注入伪造的指标数据（绕过正常采集路径）"""
        attack_type = 'metric_injection'
        target = '/api/fake_metrics'
        fake_payload = {
            'latency_ms': 1,  # 伪造极低的延迟
            'cpu_usage': 5,
            'source': 'malicious_pod'
        }
        success = False
        detected_by = None
        details = ""

        # 验证1：尝试直接写入数据库（模拟未授权写入）
        try:
            from models import PerformanceMetric
            fake_metric = PerformanceMetric(
                latency_ms=fake_payload['latency_ms'],
                cpu_usage=fake_payload['cpu_usage'],
                memory_usage=30.0,
                request_rate=100.0,
                error_rate=0.0,
                endpoint='INJECTED'
            )
            db.session.add(fake_metric)
            db.session.commit()
            success = True
            details = "伪造指标成功直接写入数据库。"
            # 触发告警：检测到来源异常的指标
            alert_manager.trigger_alert(
                rule_name='异常指标来源检测',
                metric_name='metric_source',
                metric_value=1,
                description=f'检测到来自非常规端点(INJECTED)的指标注入。',
                severity='CRITICAL'
            )
            detected_by = 'AlertManager (异常来源规则)'
            success = False  # 攻击虽执行，但被检测到，故整体视为失败
        except Exception as e:
            details = f"写入数据库失败（可能已有防护）: {e}"
            detected_by = 'Database Constraint/Auth'

        self._log_attack(attack_type, target, str(fake_payload), success, detected_by, details)
        return not success  # 返回防护是否成功

    def simulate_brute_force_alert_api(self):
        """模拟攻击：对告警API进行暴力破解尝试"""
        attack_type = 'brute_force'
        target = '/api/alerts'
        success = False
        detected_by = None
        details = ""

        # 模拟多次快速失败请求
        failure_count = 0
        for i in range(10):
            # 模拟一个错误的API密钥或令牌
            fake_token = ''.join(random.choices(string.ascii_letters, k=10))
            # 在实际中，这里会是一个HTTP请求。我们模拟请求被拒绝。
            time.sleep(0.1)  # 模拟网络延迟
            failure_count += 1

        if failure_count >= 5:
            # 触发告警：检测到暴力破解行为
            alert_manager.trigger_alert(
                rule_name='暴力破解检测',
                metric_name='auth_failures',
                metric_value=failure_count,
                description=f'检测到针对 {target} 的多次失败认证尝试({failure_count}次)。',
                severity='ERROR'
            )
            detected_by = 'AlertManager (速率限制规则)'
            details = f"攻击被检测到，触发暴力破解告警。失败次数: {failure_count}"
        else:
            success = True  # 模拟攻击未被有效检测（但实际上并未成功访问）
            details = "攻击未被现有规则检测到（模拟）。"

        self._log_attack(attack_type, target, f"attempts={failure_count}", success, detected_by, details)
        return detected_by is not None  # 返回是否被检测到

    def simulate_dos_on_monitor_endpoint(self):
        """模拟攻击：对监控指标端点发起高频率请求（DoS）"""
        attack_type = 'dos_attack'
        target = '/metrics'
        success = False
        detected_by = None
        details = ""

        # 模拟短时间内大量请求导致请求率激增
        # 此攻击会通过监控代理记录的高请求率，被标准的"高请求率告警"检测到（如果配置了该规则）。
        # 我们这里假设触发了基线检查中关于"资源过载"的检查。
        alert_manager.trigger_alert(
            rule_name='资源过载风险',
            metric_name='request_rate',
            metric_value=999.0,  # 模拟极高的值
            description=f'检测到对监控端点 {target} 的潜在DoS攻击，请求率异常高。',
            severity='CRITICAL'
        )
        detected_by = 'BaselineEngine & AlertManager'
        details = "DoS攻击模式被异常请求率检测规则识别。"
        # 攻击本身可能造成影响（success=True），但被检测到了。
        self._log_attack(attack_type, target, "high_rate_requests", True, detected_by, details)
        return detected_by is not None

    def run_all_simulations(self):
        """运行所有攻防模拟并返回结果摘要"""
        print(f"\n[AttackSimulator] {datetime.utcnow()}: 开始执行攻防验证...")
        results = []
        results.append(('指标注入', self.simulate_metric_injection()))
        results.append(('API暴力破解', self.simulate_brute_force_alert_api()))
        results.append(('监控端点DoS', self.simulate_dos_on_monitor_endpoint()))

        summary = {name: '防护成功' if detected else '防护存在缺口' for name, detected in results}
        print(f"[AttackSimulator] 验证完成。摘要: {summary}")
        return summary

    def _log_attack(self, attack_type, target, payload, success, detected_by, details):
        """记录攻击日志"""
        log = AttackLog(
            attack_type=attack_type,
            target=target,
            payload=payload,
            success=success,  # 攻击是否成功（绕过防护）
            detected_by=detected_by,
            details=details
        )
        db.session.add(log)
        db.session.commit()
        status = "成功(绕过)" if success else "被阻断/检测"
        print(f"  [AttackLog] {attack_type} -> {target}: {status} (检测方: {detected_by})")

from flask import Flask, render_template, jsonify, request, make_response
from flask_sqlalchemy import SQLAlchemy
import random
import time
import threading
from datetime import datetime, timedelta
import schedule

from config import Config
from models import *
from core.monitor_agent import MonitorAgent
from core.baseline_engine import BaselineEngine
from core.alert_manager import AlertManager
from core.attack_simulator import AttackSimulator

app = Flask(__name__)
app.config.from_object(Config)

db = SQLAlchemy(app)

# 初始化核心组件
monitor_agent = MonitorAgent()
baseline_engine = BaselineEngine()
alert_manager = AlertManager()
attack_simulator = AttackSimulator()

# 存储最近的指标用于图表（简单内存存储）
recent_metrics_cache = {'latency': [], 'cpu': [], 'timestamp': []}
MAX_CACHE_POINTS = 30

@app.before_first_request
def initialize():
    """初始化数据库和启动后台任务"""
    db.create_all()
    monitor_agent.start()
    baseline_engine.start()
    _start_simulated_load()
    print("[App] 系统初始化完成，后台任务已启动。")

def _start_simulated_load():
    """启动模拟请求生成器（独立线程）"""
    def generate_load():
        import time
        while True:
            # 随机间隔生成请求
            time.sleep(random.uniform(0.1, 0.5))
            monitor_agent.record_request()  # 通知监控代理
    threading.Thread(target=generate_load, daemon=True).start()

# ---------- 业务与监控 API ----------
@app.route('/order', methods=['POST'])
def create_order():
    """模拟订单处理API，是性能监控的主要目标"""
    start_time = time.time()
    # 模拟处理逻辑
    processing_time = random.uniform(0.05, 0.2)  # 基础处理时间
    # 小概率引入额外延迟（模拟瓶颈）
    if random.random() < 0.03:
        processing_time += random.uniform(0.5, 2.0)
        bottleneck = "库存数据库锁"
    elif random.random() < 0.02:
        processing_time += random.uniform(0.1, 0.5)
        bottleneck = "支付网关延迟"
    else:
        bottleneck = None
    time.sleep(processing_time)

    latency_ms = (time.time() - start_time) * 1000
    resp = {
        'order_id': random.randint(10000, 99999),
        'status': 'created',
        'processing_time_ms': round(latency_ms, 2)
    }
    if bottleneck:
        resp['bottleneck'] = bottleneck
    return jsonify(resp)

@app.route('/metrics')
def get_metrics():
    """获取最近性能指标（JSON格式）"""
    # 从数据库查询最近N条记录
    recent = PerformanceMetric.query.order_by(PerformanceMetric.timestamp.desc()).limit(MAX_CACHE_POINTS).all()
    recent = list(reversed(recent))  # 变成时间升序
    data = {
        'timestamps': [m.timestamp.strftime('%H:%M:%S') for m in recent],
        'latency': [m.latency_ms for m in recent],
        'cpu': [m.cpu_usage for m in recent],
        'error_rate': [m.error_rate for m in recent]
    }
    return jsonify(data)

@app.route('/alerts')
def get_alerts():
    """获取告警事件"""
    unresolved = request.args.get('unresolved', 'false').lower() == 'true'
    alerts = alert_manager.get_recent_alerts(unresolved_only=unresolved)
    result = [{
        'id': a.id,
        'time': a.timestamp.strftime('%Y-%m-%d %H:%M:%S'),
        'rule': a.rule.name if a.rule else '基线检查',
        'severity': a.severity,
        'description': a.description,
        'resolved': a.resolved
    } for a in alerts]
    return jsonify(result)

@app.route('/alerts/<int:alert_id>/resolve', methods=['POST'])
def resolve_alert(alert_id):
    """标记告警为已解决"""
    if alert_manager.mark_resolved(alert_id):
        return jsonify({'status': 'success'})
    return jsonify({'status': 'error', 'message': 'Alert not found'}), 404

# ---------- 安全基线与攻防验证 API ----------
@app.route('/api/baseline/run', methods=['POST'])
def run_baseline_check():
    """手动触发安全基线检查"""
    baseline_engine.run_checks()
    return jsonify({'status': '基线检查已执行'})

@app.route('/api/attack/run', methods=['POST'])
def run_attack_simulation():
    """手动触发攻防验证"""
    summary = attack_simulator.run_all_simulations()
    return jsonify({'status': '攻防验证完成', 'summary': summary})

@app.route('/api/attack/logs')
def get_attack_logs():
    """获取攻防验证日志"""
    logs = AttackLog.query.order_by(AttackLog.timestamp.desc()).limit(50).all()
    result = [{
        'time': l.timestamp.strftime('%H:%M:%S'),
        'type': l.attack_type,
        'target': l.target,
        'success': l.success,
        'detected_by': l.detected_by,
        'details': l.details
    } for l in logs]
    return jsonify(result)

# ---------- Web 控制台 ----------
@app.route('/')
def index():
    """主控制台页面"""
    # 获取一些初始数据用于渲染
    recent_alerts = alert_manager.get_recent_alerts(limit=5, unresolved_only=True)
    attack_logs = AttackLog.query.order_by(AttackLog.timestamp.desc()).limit(5).all()
    return render_template('index.html',
                         recent_alerts=recent_alerts,
                         attack_logs=attack_logs,
                         alert_threshold_latency=Config.ALERT_LATENCY_THRESHOLD_MS)

# ---------- 后台任务主循环 ----------
def background_scheduler():
    """运行后台定时任务（独立于Flask请求线程）"""
    while True:
        schedule.run_pending()
        time.sleep(1)

if __name__ == '__main__':
    # 启动后台调度器线程
    scheduler_thread = threading.Thread(target=background_scheduler, daemon=True)
    scheduler_thread.start()
    # 启动Flask应用
    app.run(debug=True, use_reloader=False)  # 禁用重载器以避免重复初始化

curl -X POST http://127.0.0.1:5000/order

curl -X POST http://127.0.0.1:5000/api/baseline/run

curl -X POST http://127.0.0.1:5000/api/attack/run