#AtlanticCouncil #CyberSpace $ETH

201.78 UAH ❗🕸️💰 import scapy.all as scapy import pandas as pd import re import smtplib from email.mime.text import MIMEText from sklearn.ensemble import IsolationForest import os import logging import requests from datetime import datetime # Налаштування логування logging.basicConfig(filename='cybersecurity.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Список підозрілих IP-адрес (приклад, можна оновлювати з бази даних) SUSPICIOUS_IPS = [ '185.125.190.0/24', # Приклад діапазону IP, пов'язаних з Росією '91.217.20.0/24', # Приклад діапазону IP, пов'язаних з Іраном '103.224.182.0/24' # Приклад діапазону IP, пов'язаних з Китаєм ] # 1. Модуль моніторингу мережевого трафіку (виявлення підозрілих IP) def monitor_network(interface="eth0"): """Моніторинг мережевого трафіку для виявлення підозрілих IP.""" logging.info("Початок моніторингу мережі на інтерфейсі %s", interface) def packet_callback(packet): if packet.haslayer(scapy.IP): src_ip = packet[scapy.IP].src dst_ip = packet[scapy.IP].dst if any(src_ip.startswith(ip.split('/')[0]) for ip in SUSPICIOUS_IPS): logging.warning(f"Підозрілий трафік з IP: {src_ip} до {dst_ip}") send_alert(f"Виявлено підозрілий трафік з {src_ip} до {dst_ip}") try: scapy.sniff(iface=interface, prn=packet_callback, store=0) except Exception as e: logging.error(f"Помилка моніторингу мережі: {e}") # 2. Модуль аналізу логів для виявлення аномалій def analyze_logs(log_file="access.log"): """Аналіз лог-файлів для виявлення аномалій з використанням Isolation Forest.""" logging.info(f"Аналіз лог-файлу: {log_file}") try: # Читання лог-файлу with open(log_file, 'r') as f: logs = f.readlines() # Парсинг логів (приклад для Apache/Nginx) log_data = [] for line in logs: match = re.match(r'(\S+) - - \[(.*?)\] "(\S+) (\S+) \S+" (\d+) (\d+)', line) if match: ip, timestamp, method, url, status, size = match.groups() log_data.append([ip, timestamp, method, url, int(status), int(size)]) # Створення DataFrame df = pd.DataFrame(log_data, columns=['IP', 'Timestamp', 'Method', 'URL', 'Status', 'Size']) # Використання Isolation Forest для виявлення аномалій iso_forest = IsolationForest(contamination=0.1, random_state=42) df['Anomaly'] = iso_forest.fit_predict(df[['Status', 'Size']]) # Виявлення аномалій anomalies = df[df['Anomaly'] == -1] if not anomalies.empty: logging.warning(f"Виявлено аномалії в логах:\n{anomalies}") send_alert(f"Виявлено аномалії в логах:\n{anomalies.to_string()}") except Exception as e: logging.error(f"Помилка аналізу логів: {e}") # 3. Модуль захисту від фішингу def check_phishing_emails(email_content): """Перевірка вмісту електронної пошти на ознаки фішингу.""" phishing_keywords = ['urgent', 'login', 'password', 'account suspended', 'verify'] suspicious_urls = re.findall(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', email_content) for url in suspicious_urls: try: response = requests.get(url, timeout=5) if response.status_code == 200: logging.warning(f"Підозрілий URL у листі: {url}") send_alert(f"Виявлено підозрілий URL у листі: {url}") except requests.RequestException: logging.error(f"Помилка перевірки URL: {url}") if any(keyword in email_content.lower() for keyword in phishing_keywords): logging.warning("Виявлено потенційний фішинговий лист") send_alert("Виявлено потенційний фішинговий лист") # 4. Модуль відправки сповіщень def send_alert(message): """Відправка сповіщень про інциденти на електронну пошту.""" try: msg = MIMEText(message) msg['Subject'] = 'Cybersecurity Alert' msg['From'] = '[email protected]' msg['To'] = '[email protected]' with smtplib.SMTP('smtp.yourdomain.com', 587) as server: server.starttls() server.login('[email protected]', 'your_password') server.sendmail(msg['From'], msg['To'], msg.as_string()) logging.info("Сповіщення відправлено: %s", message) except Exception as e: logging.error(f"Помилка відправки сповіщення: {e}") # 5. Модуль аналізу людського фактору def analyze_user_behavior(user_data_file="user_activity.csv"): """Аналіз поведінки користувачів для виявлення підозрілих дій.""" try: df = pd.read_csv(user_data_file) # Приклад: перевірка частоти входів із різних IP user_ip_counts = df.groupby('user_id')['ip_address'].nunique() suspicious_users = user_ip_counts[user_ip_counts > 3] # Поріг: більше 3 різних IP if not suspicious_users.empty: logging.warning(f"Підозріла поведінка користувачів:\n{suspicious_users}") send_alert(f"Підозріла поведінка користувачів:\n{suspicious_users.to_string()}") except Exception as e: logging.error(f"Помилка аналізу поведінки користувачів: {e}") # Основна функція def main(): logging.info("Запуск системи кіберзахисту") # Запуск моніторингу мережі import threading network_thread = threading.Thread(target=monitor_network, args=("eth0",)) network_thread.start() # Аналіз логів analyze_logs("access.log") # Приклад перевірки email на фішинг sample_email = "Urgent! Your account is suspended. Please verify at http://fake-site.com" check_phishing_emails(sample_email) # Аналіз поведінки користувачів analyze_user_behavior("user_activity.csv") if __name__ == "__main__": main() 👊 р с о у о е е