Suricata with Raspberry Piで遊ぼう！【通知アプリ開発編】

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

from config.notice import *

# メール送信
class EMailSend():
    # メール送信
    def sendEmail(self, message: str):
        msg: MIMEMultipart = MIMEMultipart()
        msg["From"] = SENDER_EMAIL
        msg["To"] = RECEIVER_EMAIL
        msg["Subject"] = "[SuricataAlertNotice] 不審なパケットを検知しました！"
        msg.attach(MIMEText(message, "plain"))

        # GmailのSMTPサーバーへの接続とメール送信
        try:
            with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
                server.starttls() # TLS暗号化の開始
                server.login(SMTP_USER, SMTP_PASS) # SMTPサーバーへのログイン

                server.send_message(msg)
        except Exception as e:
            print(e)

import requests

from config.notice import LINE_NOTIFY_TOKEN

# Notify送信
class LineNotifySend():
    # Line Notify API セットアップ
    def __init__(self):
        self.url: str = "https://notify-api.line.me/api/notify"
        self.line_headers: dict[str, str] = {'Authorization': 'Bearer ' + LINE_NOTIFY_TOKEN}

    # メッセージ送信
    def sendLineMessage(self, message: str):
        try:
            payload: dict[str, str] = {'message': message}
            # Line送信
            requests.post(self.url, headers=self.line_headers, params=payload)
        except requests.exceptions.RequestException as e:
            print(e)

import json
import os
import time
from datetime import datetime, timezone, timedelta

from send_mail import EMailSend
from send_line import LineNotifySend

from config.general import *

class Notice():
    def __init__(self):
        self.email_notice: EMailSend = EMailSend()
        self.line_notify: LineNotifySend = LineNotifySend()

        self.notice_target_eve: list[dict] = []

    # 通知
    def notice(self, notice_target_eve: list[dict]):
        self.notice_target_eve = notice_target_eve

        message: str = "不審な通信を検知しました！\n"
        message += f"件数: {len(self.notice_target_eve)} 件\n"
        if not (NOTICE_TYPE is NoticeType.Nothing):
            if NOTICE_TYPE is NoticeType.LineNotify or NOTICE_TYPE is NoticeType.Both: # LINE通知
                self.send_line(message)
            if NOTICE_TYPE is NoticeType.Email or NOTICE_TYPE is NoticeType.Both: # Email送信
                self.send_email(message)

    # メッセージ作成
    def create_message(self, eve: dict) -> str:
        try:
            message: str = "\n** 情報 **\n"
            message += f"timestamp: {eve['timestamp'].strftime('%Y/%m/%d %H:%M:%S')}\n"
            message += f"src: {eve['src_ip']}:{eve['src_port']} -> dest: {eve['dest_ip']}:{eve['dest_port']}\n"
            message += f"protocol: {eve['proto']}\n"
            message += f"\n{eve['alert']['signature']}\n"
            message += f"signature ID: {eve['alert']['signature_id']}\n"
            if "http" in eve:
                eve_http: dict = eve["http"]
                if eve_http != {}:
                    message += f"\n** HTTP情報 **\n"
                    message += f"hostname: {eve_http['hostname']}\n"
                    message += f"url: {eve_http['url']}\n"
                    message += f"{eve_http['http_content_type']} {eve_http['http_method']} {eve_http['status']}\n"
        except KeyError:
            pass
        return message

    # Email送信
    def send_email(self, message: str):
        for eve in self.notice_target_eve:
            message += self.create_message(eve)
            message += "\n===========================================================\n"
        self.email_notice.sendEmail(message)
    
    # LINE送信
    def send_line(self, message: str):
        self.line_notify.sendLineMessage(message)
        for eve in self.notice_target_eve:
            message = self.create_message(eve)
            self.line_notify.sendLineMessage(message)

def main():
    notice: Notice = Notice()

    prev_timestamp: datetime = datetime.now(timezone(timedelta(hours=9)))
    last_modified: float = 0.0
    current_modified: float = os.path.getmtime(EVE_JSONL_PATH)
    lower_sig_id, upper_sig_id = SIGNATURE_ID_RANGE

    while True:
        notice_target_eve: list[dict] = []

        if current_modified != last_modified:
            # eve.json取得
            with open(EVE_JSONL_PATH, "r") as eve_jsonl:
                for jsonl in eve_jsonl:
                    eve: dict = json.loads(jsonl)

                    cur_timestamp: datetime = datetime.strptime(eve["timestamp"], "%Y-%m-%dT%H:%M:%S.%f%z")
                    eve["timestamp"] = cur_timestamp
                    if cur_timestamp > prev_timestamp and "alert" in eve:
                        sig_id: int = eve["alert"]["signature_id"] 
                        if lower_sig_id <= sig_id and sig_id <= upper_sig_id:
                            notice_target_eve.append(eve)
                        prev_timestamp = cur_timestamp
            # 通知
            if len(notice_target_eve) > 0:
                notice.notice(notice_target_eve)
            last_modified = current_modified
        current_modified = os.path.getmtime(EVE_JSONL_PATH)
        time.sleep(1)

if __name__ == '__main__':
    main()

from enum import Enum

class NoticeType(Enum):
    Nothing = 0
    Email = 1
    LineNotify = 2
    Both = 3

EVE_JSONL_PATH: str = "/usr/local/var/log/suricata/eve.json"
NOTICE_TYPE = NoticeType.Email
# SIGNATURE_ID_RANGE内のシグネチャのみ通知
SIGNATURE_ID_RANGE: tuple[int, int] = (2000000, 2099999) # ET Open Rulesets

SENDER_EMAIL: str = "your_email@example.com"
RECEIVER_EMAIL: str = "your_email@gmail.com"
SMTP_SERVER: str = "smtp.gmail.com"
SMTP_PORT: int = 587
SMTP_USER: str = "your_email@gmail.com"
SMTP_PASS: str = "**********************"

LINE_NOTIFY_TOKEN: str = "***************************" # Line Notify API Access Token