import logging
import time
from datetime import datetime
from collections import defaultdict
import pandas as pd
import matplotlib.pyplot as plt
 
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
# 模擬數(shù)據(jù)庫日志
class DatabaseLogger:
    def __init__(self):
        self.logs = []
 
    def log_query(self, user, query, timestamp=None):
        if not timestamp:
            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_entry = {"user": user, "query": query, "timestamp": timestamp}
        self.logs.append(log_entry)
        logging.info(f"Logged query: {log_entry}")
 
    def get_logs(self):
        return self.logs
 
# 規(guī)則引擎
class RuleEngine:
    def __init__(self):
        self.rules = [
            {"name": "SQL Injection", "pattern": ["' OR '1'='1", ";--", "UNION SELECT"]},
            {"name": "Sensitive Data Access", "pattern": ["SELECT * FROM users", "SELECT * FROM credit_cards"]},
            {"name": "Brute Force", "threshold": 5}  # 5 queries within 10 seconds
        ]
 
    def analyze_logs(self, logs):
        risks = []
        user_query_count = defaultdict(int)
 
        for log in logs:
            user = log["user"]
            query = log["query"]
            timestamp = log["timestamp"]
 
            # 規(guī)則1: SQL注入檢測
            for rule in self.rules:
                if "pattern" in rule:
                    for pattern in rule["pattern"]:
                        if pattern in query:
                            risks.append({
                                "user": user,
                                "query": query,
                                "timestamp": timestamp,
                                "risk": rule["name"],
                                "severity": "High"
                            })

            # 規(guī)則2: 破解檢測
            if "threshold" in rule:
                user_query_count[user] += 1
                if user_query_count[user] > rule["threshold"]:
                    risks.append({
                        "user": user,
                        "query": query,
                        "timestamp": timestamp,
                        "risk": rule["name"],
                        "severity": "Medium"
                    })
 
        return risks
 
# 風(fēng)險評估
class RiskAssessor:
    @staticmethod
    def assess_risks(risks):
        risk_summary = defaultdict(int)
        for risk in risks:
            risk_summary[risk["risk"]] += 1
        return risk_summary
 
# 報警系統(tǒng)
class AlertSystem:
    @staticmethod
    def send_alert(risk):
        logging.warning(f"ALERT: Risk detected - {risk}")
 
# 可視化模塊
class Visualizer:
    @staticmethod
    def plot_risks(risk_summary):
        risks = list(risk_summary.keys())
        counts = list(risk_summary.values())
 
        plt.bar(risks, counts, color='red')
        plt.xlabel('Risk Type')
        plt.ylabel('Count')
        plt.title('Database Risk Summary')
        plt.show()
 
# 主系統(tǒng)
class DatabaseRiskDiscoverySystem:
    def __init__(self):
        self.logger = DatabaseLogger()
        self.rule_engine = RuleEngine()
        self.risk_assessor = RiskAssessor()
        self.alert_system = AlertSystem()
        self.visualizer = Visualizer()
 
    def run(self):
        # 模擬日志數(shù)據(jù)
        self.logger.log_query("admin", "SELECT * FROM users WHERE id = 1")
        self.logger.log_query("hacker", "SELECT * FROM users WHERE id = 1 OR '1'='1'")
        self.logger.log_query("hacker", "SELECT * FROM credit_cards")
        self.logger.log_query("hacker", "SELECT * FROM users;--")
        self.logger.log_query("hacker", "SELECT * FROM users")
        self.logger.log_query("hacker", "SELECT * FROM users")
        self.logger.log_query("hacker", "SELECT * FROM users")
        self.logger.log_query("hacker", "SELECT * FROM users")
 
        # 獲取日志并分析風(fēng)險
        logs = self.logger.get_logs()
        risks = self.rule_engine.analyze_logs(logs)
 
        # 評估風(fēng)險
        risk_summary = self.risk_assessor.assess_risks(risks)
 
        # 發(fā)送報警
        for risk in risks:
            self.alert_system.send_alert(risk)
 
        # 可視化風(fēng)險
        self.visualizer.plot_risks(risk_summary)
 
# 運行系統(tǒng)
if __name__ == "__main__":
    system = DatabaseRiskDiscoverySystem()
    system.run()