Статья Honeypot-fakeSSH TG Bot (paramiko, aiogram3)

[rand]

Код не мой, моего знакомого. Его тут нет. Есть конечно вопросы по коду (проблески говнокода), но бот работает, потом нормально переделаю как настроение будет. Для себя вижу пользу для сбора логинов и паролей которые используются в бруте SSH при атаке ботами и т.д.

Яп: Python.
Библиотеки: paramiko, aiogram3, sqllite, aiosqlite
Описание:
Сетап проиходит установкой зависимостей, генерацией ключа, заполнению json файла chatid, токеном бота и временем задержки ответа(а также максимальным количеством попыток аутентификации; порта(по дефолту 22)). Поднимается на указанном порту и имитирует ssh сервак. Все запросы передает в указанный чат телеграма (botapi), с указанием страны и провайдера, а также кладет в базу данных. В чате с ботом также можно запросить статистику по ip адресам, странам.

Пример отработки:

Код:

check_config.py:

import sqlite3

def load_config():
    conn = sqlite3.connect("config.db")
    cursor = conn.cursor()
    cursor.execute("SELECT tg_bot_token, chat_id, cb, ma, delay, port FROM config LIMIT 1")
    config = cursor.fetchone()
    conn.close()
    if not config:
        raise ValueError("No configuration")
    return {
        "tg_bot_token": config[0],
        "chat_id": config[1],
        "cb": config[2],
        "ma": config[3],
        "delay": config[4],
        "port": config[5],
    }

if __name__ == "__main__":
    try:
        config = load_config()
        print("Configuration:")
        for key, value in config.items():
            print(f"{key}: {value}")
    except ValueError as e:
        print(e)

main.py:

import asyncio
import paramiko
import socket
import aiosqlite
from aiosqlite import connect
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.types import FSInputFile
import threading
import time
import matplotlib.pyplot as plt
import os


async def load_config():
    """async def that loads config from a database
    """
    async with aiosqlite.connect("config.db") as conn:
        async with conn.execute("SELECT tg_bot_token, chat_id, cb, ma, delay, port FROM config LIMIT 1") as cursor:
            config = await cursor.fetchone()
            if not config:
                raise ValueError("No configuration")
            return config


async def send_msg(token, chat_id, message):
    """Fynction that let alerts about a login attempts in telegram chat exist

    Args:
        token (int): telegram bot token
        chat_id (int): telegram chat id with alerts in botapi standart
        message (int): message, that sending to chat when someone is trying to brute-force this honeypot
    """
    bot = Bot(token=token)
    await bot.send_message(chat_id=chat_id, text=message)
    await bot.session.close()


async def log_db(username, password, client_ip):
    """Async fynction wich logging all login attempts in database

    Args:
        username (str): username entered by attackers
        password (str): password entered by attackers
        client_ip (str): attackers ip
    """
    async with aiosqlite.connect("honeypot_logs.db") as conn:
        await conn.execute('''CREATE TABLE IF NOT EXISTS logs (
                                id INTEGER PRIMARY KEY AUTOINCREMENT,
                                username TEXT,
                                password TEXT,
                                ip TEXT,
                                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
                            )''')
        await conn.execute('INSERT INTO logs (username, password, ip) VALUES (?, ?, ?)',
                           (username, password, client_ip))
        await conn.commit()


class FakeSSHServer(paramiko.ServerInterface):
    """Class wich raises ssh server using paramiko
    """
    def __init__(self, client_ip, send_message_func, log_func, ma, delay):
        super().__init__()
        self.client_ip = client_ip
        self.send_message_func = send_message_func
        self.log_func = log_func
        self.auth_attempts = {}
        self.ma = ma
        self.delay = delay
        self.completion_event = threading.Event()

    def check_channel_request(self, kind, chanid):
        if kind == "session":
            return paramiko.OPEN_SUCCEEDED
        return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED

    def check_auth_password(self, username, password):
        asyncio.run(self.log_func(username, password, self.client_ip))
        asyncio.run(self.send_message_func(
            f"Login attempt from {self.client_ip}\nUsername: {username}\nPassword: {password}"))

        if username not in self.auth_attempts:
            self.auth_attempts[username] = 0

        self.auth_attempts[username] += 1
        if self.auth_attempts[username] >= self.ma:
            asyncio.run(self.send_message_func(
                f"Failed login attempts exceeded for {self.client_ip}"
            ))
        time.sleep(self.delay / 1000.0)
        return paramiko.AUTH_FAILED


def handle_client(client_socket, client_address, tg_bot_token, chat_id, cb, ma, delay):
    """Fuction wich handles ssh clients

    Args:
        client_address (str): client ip adress
        tg_bot_token (str): telegram bot token
        chat_id (str):  telegram chat id
        cb (str): banner, wich paramiko sends to clients
        ma (int): max attempts from a client
        delay (int): delay in answer
    """
    transport = paramiko.Transport(client_socket)
    try:
        server_key = paramiko.RSAKey(filename="server_key.pem")
        transport.add_server_key(server_key)
        server = FakeSSHServer(
            client_ip=client_address[0],
            send_message_func=lambda msg: send_msg(tg_bot_token, chat_id, msg),
            log_func=log_db,
            ma=ma,
            delay=delay
        )
        transport.local_version = cb

        transport.start_server(server=server)

        channel = transport.accept(20)
        if channel:
            channel.close()
    finally:
        transport.close()
        client_socket.close()


async def start_server(tg_bot_token, chat_id, cb, ma, delay, port):
    """Starts a paramiko ssh server

    Args:
        tg_bot_token (str): telegram bot token
        chat_id (str):  telegram chat id
        cb (str): banner, wich paramiko sends to clients
        ma (int): max attempts from a client
        delay (int): delay in answer to client
        port (ште): port on which the ssh server is raised
    """
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(("", port))
    server_socket.listen(100)

    print(f"Listening for connections on port {port}...")

    server_socket.setblocking(False)
    loop = asyncio.get_event_loop()
    while True:
        client_socket, client_address = await loop.sock_accept(server_socket)

        threading.Thread(target=handle_client, args=(
            client_socket, client_address, tg_bot_token, chat_id, cb, ma, delay)).start()


async def tg_bot(tg_bot_token, chat_id):
    """Async function with a telegram bot

    Args:
        tg_bot_token (str): variable with a telegram bot token
        chat_id (int): variable with a telegram chat id
    """
    bot = Bot(token=tg_bot_token)
    dp = Dispatcher()

    @dp.message(Command('start'))
    async def cmd_start(message: types.Message):
        if str(message.chat.id) == chat_id:
            await message.answer("90")

    @dp.message(Command("stat_ip"))
    async def statip(message: types.Message):
     if str(message.chat.id) == chat_id:
         async with connect("honeypot_logs.db") as conn:
             async with conn.execute("SELECT ip, COUNT(*) AS attempts FROM logs GROUP BY ip ORDER BY attempts DESC") as cursor:
                 rows = await cursor.fetchall()
         stats = "\n".join([f"{ip}: {count}" for ip, count in rows])
         await message.answer(f"IP Stats:\n{stats}")

    @dp.message(Command("stat_ip_chart"))
    async def statip_chart(message: types.Message):
     if str(message.chat.id) == chat_id:
        async with connect("honeypot_logs.db") as conn:
            async with conn.execute("SELECT ip, COUNT(*) AS attempts FROM logs GROUP BY ip ORDER BY attempts DESC") as cursor:
                rows = await cursor.fetchall()

        if not rows:
            await message.answer("No data")
            return
        labels = [row[0] for row in rows]
        sizes = [row[1] for row in rows]

        plt.figure(figsize=(8, 8))
        plt.pie(
            sizes,
            labels=None,
            autopct='%1.1f%%',
            colors=plt.cm.tab20.colors[:len(sizes)]
        )
        plt.legend(labels=[f"{ip}: {count}" for ip, count in rows],
                   loc="lower left", bbox_to_anchor=(1, 0))
        plt.title("Shares of IP addresses from the total array")

        plt.savefig("temp.png", bbox_inches="tight")
        plt.close()

        png_path = os.path.join(os.path.dirname(__file__), 'temp.png')

        await message.answer(f'{png_path}')
        png = FSInputFile(png_path)
        await message.answer_photo(photo=png, caption='Ip statistic graph')

        os.remove(png_path)

    await dp.start_polling(bot)


async def main():
    """loads all other functions
    """
    tg_bot_token, chat_id, cb, ma, delay, port = await load_config()
    chat_id = str(chat_id)
    await asyncio.gather(
        start_server(tg_bot_token, chat_id, cb, ma, delay, port),
        tg_bot(tg_bot_token, chat_id)
    )

if __name__ == "__main__":
    asyncio.run(main())

setup.py:

import sqlite3
import os
import paramiko
from typing import Callable


def create_database():
    conn = sqlite3.connect("config.db")
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS config (
            tg_bot_token TEXT NOT NULL,
            chat_id TEXT NOT NULL,
            cb TEXT NOT NULL,
            ma INTEGER NOT NULL,
            delay INTEGER NOT NULL,
            port INTEGER NOT NULL
        )
    """)
    conn.commit()
    conn.close()


kf = "server_key.pem"


def generate_rsa_key(key_filename=kf, key_size=2048):
    key = paramiko.RSAKey.generate(key_size)
    key.write_private_key_file(key_filename)
    print(f"Key generated: {key_filename}")
    print("Goodbye!")


def delete_existing_key():
    delete_input = input(
        f"Key {kf} exists. Delete it? (y/n): ").strip().lower()
    if delete_input == 'y':
        os.remove(kf)
        print(f"Key {kf} deleted.")
        generate_rsa_key()
    else:
        print('Okay, goodbye!')


def prompt_for_key_generation():
    if os.path.exists(kf):
        delete_existing_key()
    else:
        generate_rsa_key()


def handle_user_input():
    user_input = input("Enter 1 to generate the key or 0 to exit: ").strip()
    if user_input == "1":
        prompt_for_key_generation()
    elif user_input == "0":
        print("Bye")
    else:
        print("Invalid input. Please try again.")


def main1():
    handle_user_input()


def save_config(tg_bot_token, chat_id, cb, ma, delay, port):
    conn = sqlite3.connect("config.db")
    cursor = conn.cursor()
    cursor.execute("DELETE FROM config")
    cursor.execute("""
        INSERT INTO config (tg_bot_token, chat_id, cb, ma, delay, port)
        VALUES (?, ?, ?, ?, ?, ?)
    """, (tg_bot_token, chat_id, cb, ma, delay, port))
    conn.commit()
    conn.close()


def check(prompt: str, condition: Callable[[int], bool], default=None) -> int:
    while True:
        try:
            raw = input(prompt + ": ").strip()

            if not raw and default is not None:
                return default

            value = int(raw)
            if condition(value):
                return value
            print("Incorrect value, please try again..")
        except ValueError:
            print("You must enter an integer value.")


def main():
    print("Configure honeypot for yourself:")

    while True:
        tg_bot_token = input("Telegram Bot Token: ").strip()
        if not tg_bot_token:
            print("Token can't be blank")
        else:
            break

    chat_id = check("Telegram Chat ID (botapi) ", lambda x: x < 0)
    cb = "SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.10"
    ma = check("Max Attempts (leave blank for default value of 3)",
               lambda x: x > 0, default=3)
    delay = check("Delay (in ms(leave blank for default value of 929)) ",
                  lambda x: x >= 0, default=929)
    port = check("Port (leave blank for default value of 22)",
                 lambda x: 0 < x < 65536, default=22)

    save_config(tg_bot_token, chat_id, cb, ma, delay, port)
    print("The configuration is saved, now you can run honeypot")


if __name__ == "__main__":
    create_database()
    main()
    main1()