Статья Змеиный кот. Создаем аналог NetCat на Python

Введение
Что тебе приходит в голову, когда ты слышишь слово Python? Многофункциональность или кроссплатформенность? Может быстрая скорость работы, ведь у каждого этот язык ассоциируется с различными задачами и преимуществами. Но сегодня давай уделим время Python в сфере информационной безопасности. Здесь он идеально подходит под написания скриптов и различных эксплоитов. Его эффективность оставляет конкурентов позади. Поэтому давай попробуем написать на нем полезную утилиту для быстрой работы.

План работы
Когда я первый раз задумался о способностях питона мне почему то на ум сразу пришла его скорость работы. Сам процесс обработки информации и ее отправки в другую точку. Поэтому в этой статье с целью эксперимента я хочу показать насколько хорошо получится у этого языка справиться с сетью. Проще говоря я покажу тебе, как написать свой аналог NetCat, но на более современно языке. Оригинальная утилита использует ресурсы семейства C и поэтому тягаться с ней будет весьма проблематично, хотя не будем забывать о том, что один из главных компонентов Python является C++. Приступим к работе!

Немного теории
Здесь я оставил пару подсказок от себя, чтобы не нагружать статью лишней теорией о которой я говорил ранее. Также добавим к списку термин NetCat, для тех, кто только познает мир информационной безопасности.

NetCat — утилита Unix, позволяющая устанавливать TCP/UDP соединения, принимать оттуда данные и передавать их. Несмотря на свою полезность и простоту, данная утилита не входит ни в какой стандарт. Поэтому продвинутые системные администраторы стараются по быстрому удалить эту вещь со своих устройств.

Из за своих "стандартов" этой утилиты и может не быть на нужном тебе устройстве. И в этом случаи как раз приходят прямые руки и наличие IDE под рукой. Также если ты совсем новичок и не знаешь как устроена сеть, то рекомендую прочесть пару статей с форума и журнала. В остальном начнем писать наш код.

1. Знакомство с ARP и понятием проктола;
2. Основы TLS и работы в Python;
3. Ссылка скрыта от гостей
   .

Пишем собственный NetCat
Для работы я использовал PyCharm и Python версии 3.10. Скачать все это ты можешь с официального сайта разработчиков. Также тебе потребуется любой дистрибутив Linux, так как программа будет иметь флаги и запускаться в формате питон файла. Ну а чтобы наш код работал давай импортируем все необходимые модули.

import argparse
import socket
import shlex
import subprocess
import sys
import textwrap
import threading

Так как NetCat это утилита удаленного доступа, то для ее работы потребуется правильная настройка пакетов, также работа с системой и процессами. Все эти модули помогут нам создать качественное и беспрерывное соединение с устройством. После этого создадим ниже функцию, которая будет обрабатывать и возвращать команды в нужном нам виде. В этом нам поможет subprocess и прямые руки. Сам код выглядит следующим образом:

def execute(cmd):
    cmd = cmd.strip()    # Проверка строки
    if not cmd:        # Условие, если строка пустая
        return
    output = subprocess.check_output(shlex.split(cmd), stderr=subprocess.STDOUT)    # Чтение и возрват (1)
    return output.decode()

На пункте 1 ты можешь заметить пример эксплуатации нашего модуля. В этом случаи используется команда check_out, которая выполняет команду в локальной операционной системе. Также не будем забывать про создания мэйна нашего кода, который будет отвечать за аргументы и другие функции создания консольного интерфейса. Здесь все максимально просто. При использовании аргумента help наш код будет выводить всю информацию о каждом флаге и его применении при работе. Для реализации этого дела стоит воспользоваться парсером аргументов. Весь код я оставил ниже.

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Python Cat Network', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=textwrap.dedent('''
Example:
netcat.py -t 127.0.0.1 -p 4444 -l -c
// командная оболочка
netcat.py -t 127.0.0.1 -p 4444 -l -u=mytest.txt
// загружаем в файл
netcat.py -t 127.0.0.1 -p 4444 -l -e=\"cat /etc/passwd\"
// выполняем команду
echo 'ABC' | ./netcat.py -t 127.0.0.1 -p 1234
// шлем текст на порт сервера 1234
netcat.py -t 127.0.0.1 -p 4444
// соединяемся с сервером
'''))
    parser.add_argument('-c', '--command', action='store_true',
                        help='command shell')
    parser.add_argument('-e', '--execute', help='execute specified command')
    parser.add_argument('-l', '--listen', action='store_true', help='listen')
    parser.add_argument('-p', '--port', type=int, default=5555,
                        help='specified port')
    parser.add_argument('-t', '--target', default='192.168.1.203',
                        help='specified IP')
    parser.add_argument('-u', '--upload', help='upload file')
    args = parser.parse_args()
    if args.listen:
        buffer = ''
    else:
        buffer = sys.stdin.read()
    nc = NetCat(args, buffer.encode())
    nc.run()

Как ты видишь, код максимально простой. Чаще всего мы используем аргумент добавления из модуля Parser, чтобы добавить весь список команд помощи к нашему коду. Также если утилита используется в качестве слушателя, то мы вызываем объект с пустым буфером, иначе происходит сохранения содержимого из stdin. После этого происходит запуск главного класса с аргументами и расшифровкой буфера. Но пока что вызывать нам нечего, поэтому предлагаю исправить ситуацию. Для этого создаем класс NetCat и начнем собирать его как конструктор, поочередно добавляя аргументы.

class NetCat:
    def __init__(self, args, buffer=None):
        self.args = args    # Наши будующие аргументы
        self.buffer = buffer    # Буфер с данными
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)        # Подключение к серверу
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)    # Обработка запроса и протоколов

    def run(self):
        if self.args.listen:
            self.listen()
        else:
            self.send()

Этот класс служит главным звеном в нашей программе. Он выполняет функцию пульта управления и вызывает определенные методы для работы утилиты. Первым делом мы инициализируем все объекты, а после переходим к запуску. Метод run() является точкой входа и управляет выполнением двух других методов listen() и send(). Почему именно они? Давай я немного объясню. В функции init, класса NetCat мы создаем сокет, который изначально существует в двух состояниях: слушатель, отправитель. Поэтому если он ничего не отправляет, значит должен слушать. По такому принципу мы реализуем это действие дальше. Также в этой ситуации при помощи функции run() мы переходим к проверки на прослушивание и если оно неактивно, то пробуем отправить информацию на наш сервер. После небольших объяснений предлагаю перейти к написанию функций прослушивания и отправки содержимого буфера.

Перед тем как отправлять данные нужно предусмотреть функцию аварийной остановки передачи. Для этого подойдем нестандартным путем и вызовем исключение KeyboardInterrupt. Суть в том, что такого типа ошибки возникают при нажатии клавиш Ctrl + C, поэтому, чтобы не придумывать велосипед снова просто перехватим ее и создадим собственный вывод. На коде все это дело выглядит примерно так:

def send(self):
    self.socket.connect((self.args.target, self.args.port))        # Установка соединения с сервером
    if self.buffer:
        self.socket.send(self.buffer)    # Проверка буфера на наличие данных
    try:
        while True:
            recv_len = 1    # Задаем длину запроса
            response = ''
            while recv_len:
                data = self.socket.recv(4096)    # Размер буфера в битах
                recv_len = len(data)    # Проверяем длину
                response += data.decode()    # Декодируем запрос
            if recv_len > 4096:        # Сверяем данные
                print("[/] Message is too long.")
                break
        if response:
            print(response)        # Выводим его на экран
            buffer = input('> ')    # Задаем приглашение для ввода
            buffer += '\n'    # Добавляем новый абзац для корректного чтения
            self.socket.send(buffer.encode())    # Отправляем информацию
    except KeyboardInterrupt:
        print('[!] Operation aborted')
        self.socket.close()
        sys.exit()

Здесь мы производим обмен между клиентом и сервером. Первым делом читаем наш буфер и отправляем его содержимое, далее перехватываем исключение, чтобы иметь ручное управления соединением и в случаи чего прервать связь с хостом. Далее производим зацикливание, тебе же не хочется каждый раз запускать программу, чтобы отправлять одно предложение? После этого задаем размер буфера для приема и сверяем всю информацию при помощи условных циклов. Эти данные ты можешь спокойно подстроить под себя. После этого получаем и выводим наш ответ на экран, попутно предлагая пользователю ответить на сообщение при помощи приглашения в переменной buffer. Не забываем добавлять к этому декодирования и новую строчку, чтобы все выглядело красиво и привлекательно. Попутно закрываем наш try/except и завершаем сессию при нажатие клавиш Ctrl+C.

Теперь давай добавим к этому всему режим прослушивания, чтобы программа ожидала нового ответа, а не закрывала соединение. Для этого в классе NetCat мы указали нашу функцию listen(), ее как раз и следует реализовать. Для этого ниже send() оставляем место и пишем следующий код:

def listen(self):
    self.socket.bind((self.args.target, self.args.port))    # Биндим соединение с хостом
    self.socket.listen(5)    # Устанавливаем прослушку и кол-во подключений
    while True:
        client_socket, _ = self.socket.accept()        # Принимаем соединение с клиентом
        client_thread = threading.Thread(target=self.handle, args=(client_socket,))
        client_thread.start()

Здесь я думаю объяснения излишни. Взгляни на комментарии к коду и все сразу встанет на свои места. Ничего нового или необычного у нас нету. Переменная client_thread обращается к следующей ступени нашего кода и забирает от туда данные машины вместе с аргументами. Ну а далее у нас самое интересное. Чтобы программа не казалась такой простой усложним ее путем создания передатчика файлов. Это все реализовано в несколько ступеней и имеет своеобразный формат. Чтобы это все понять рекомендую взглянуть на код ниже:

def handle(self, client_socket):
    if self.args.execute:
        output = execute(self.args.execute)        # Обращаемся к командной строке
        client_socket.send(output.encode())
    elif self.args.upload:
        file_buffer = b''    # Задаем буфер обмена
        while True:
            data = client_socket.recv(4096)        # Размер буфера в битах
            if data:
                file_buffer += data        # Помещаем файл в наш запрос
            else:
                break
        with open(self.args.upload, 'wb') as f:
            f.write(file_buffer)    # Открываем и читаем файл в бинарном виде
        message = f'Saved file {self.args.upload}'    # Выгружаем и отправляем на сервер
        client_socket.send(message.encode())

Теперь давай все поп порядку. Первым делом мы обращаемся к командной строке, за нее отвечает у нас функция execute. После чего производим соединение с сервером и задаем размер буфера обмена в месте его содержимым. Заметь, все файлы отправляются в бинарном виде и собираются на другом устройстве в нормальный вид. Так мы экономим время и делаем передачу более безопасной. После чего открываем файл для чтения в бинарном виде и отправляем всю информацию на другое устройство. Все данные читаются по строкам и это может занять какое-то время. Также давай здесь же дополним нашу функцию управлением командной строкой. Проще говоря отправлять команды мы можем, но их обработчика у нас нету. Поэтому в нашем условном цикле создаем дополнительное условие и помещаем туда такие строки:

    elif self.args.command:
        cmd_buffer = b''    # Снова задаем буфер
        while True:
            try:
                client_socket.send(b'Unknown: #> ')        # Приглашение для ввода команды
                while '\n' not in cmd_buffer.decode():
                    cmd_buffer += client_socket.recv(64)
                response = execute(cmd_buffer.decode())        # Декодирование команды в читаемый для пк вид
                if response:
                    client_socket.send(response.encode())    # Отправка ответа
                cmd_buffer = b''    # Очистка буфера
            except Exception as e:    # В случаи ошибки говорим что сервер умер от потери питания
                print(f'Server died from power loss {e}')
                self.socket.close()
                sys.exit()

Теперь у нас есть обработчик консольных команд и ты спокойно можешь отправить на сервер форк бомбу или прописать rm -rf / от имени суперпользователя, чтобы уничтожить систему. Также давай теперь посчитаем сколько строчек занимает наша программа. Со всеми пробелами и отступами выходит 134 строки кода. При этом мы имеем в нашем арсенале отправку данных на другое устройство, режим прослушивания и скачивание информации. Также выполнение команд консоли, что не мало важно. То есть мы подошли максимально близко к созданию подобия NetCat. Конечно у этой утилиты размер флагов и функций гораздо больше, но на основе этого кода ты имеешь полное право добавить новые флаги и функции для более многофункциональной работы с сетью.


Подводим итоги
Итак, про результатам эксперимента нам удалось воссоздать компактную версию известной утилиты NetCat, но с другим языком программирования и более сложной установкой. Для каких целей пригодится данная программа? Как минимум ты сможешь оставлять бэкдоры в системе добавив к программе автозапуск и скрытие в диспетчере задач. Найти такой тайник будет весьма проблематично. Но об этом я смогу рассказать в последующих статьях.

netcat.py

import argparse
import socket
import shlex
import subprocess
import sys
import textwrap
import threading


def execute(cmd):
    cmd = cmd.strip()
    if not cmd:
        return
    output = subprocess.check_output(shlex.split(cmd), stderr=subprocess.STDOUT)
    return output.decode()


class NetCat:
    def __init__(self, args, buffer=None):
        self.args = args
        self.buffer = buffer
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    def run(self):
        if self.args.listen:
            self.listen()
        else:
            self.send()


def send(self):
    self.socket.connect((self.args.target, self.args.port))
    if self.buffer:
        self.socket.send(self.buffer)
    try:
        while True:
            recv_len = 1
            response = ''
            while recv_len:
                data = self.socket.recv(4096)
                recv_len = len(data)
                response += data.decode()
            if recv_len > 4096:
                print("[/] Message is too long.")
                break
        if response:
            print(response)
            buffer = input('> ')
            buffer += '\n'
            self.socket.send(buffer.encode())
    except KeyboardInterrupt:
        print('[!] Operation aborted')
        self.socket.close()
        sys.exit()


def listen(self):
    self.socket.bind((self.args.target, self.args.port))
    self.socket.listen(5)
    while True:
        client_socket, _ = self.socket.accept()
        client_thread = threading.Thread(target=self.handle, args=(client_socket,))
        client_thread.start()


def handle(self, client_socket):
    if self.args.execute:
        output = execute(self.args.execute)
        client_socket.send(output.encode())
    elif self.args.upload:
        file_buffer = b''
        while True:
            data = client_socket.recv(4096)
            if data:
                file_buffer += data
            else:
                break
        with open(self.args.upload, 'wb') as f:
            f.write(file_buffer)
        message = f'Saved file {self.args.upload}'
        client_socket.send(message.encode())
    elif self.args.command:
        cmd_buffer = b''
        while True:
            try:
                client_socket.send(b'Unknown: #> ')
                while '\n' not in cmd_buffer.decode():
                    cmd_buffer += client_socket.recv(64)
                response = execute(cmd_buffer.decode())
                if response:
                    client_socket.send(response.encode())
                cmd_buffer = b''
            except Exception as e:
                print(f'Server died from power loss {e}')
                self.socket.close()
                sys.exit()


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='PyCat: the quieter you go, the further you will get',
                                     formatter_class=argparse.RawDescriptionHelpFormatter, epilog=textwrap.dedent('''
Example:
netcat.py -t 127.0.0.1 -p 4444 -l -c
// командная оболочка
netcat.py -t 127.0.0.1 -p 4444 -l -u=mytest.txt
// загружаем в файл
netcat.py -t 127.0.0.1 -p 4444 -l -e=\"cat /etc/passwd\"
// выполняем команду
echo 'ABC' | ./netcat.py -t 127.0.0.1 -p 1234
// шлем текст на порт сервера 1234
netcat.py -t 127.0.0.1 -p 4444
// соединяемся с сервером
'''))
    parser.add_argument('-c', '--command', action='store_true',
                        help='command shell')
    parser.add_argument('-e', '--execute', help='execute specified command')
    parser.add_argument('-l', '--listen', action='store_true', help='listen')
    parser.add_argument('-p', '--port', type=int, default=5555,
                        help='specified port')
    parser.add_argument('-t', '--target', default='192.168.1.203',
                        help='specified IP')
    parser.add_argument('-u', '--upload', help='upload file')
    args = parser.parse_args()
    if args.listen:
        buffer = ''
    else:
        buffer = sys.stdin.read()
    nc = NetCat(args, buffer.encode())
    nc.run()

 

[WithoutIrony]

with open(self.args.upload, 'wb') as f: f.write(file_buffer) # Открывем и читаем файл в бинароном виде

* Открываем и пишем
Ну и вообще, прогоните проверку орфографии

self.socket.listen(5) # Устанавливаем прослушку и время дисконнекта (в секундах)

Это лимит на количество подключений, а не время дисконнекта -

Ссылка скрыта от гостей

if recv_len < 4096: # Сверяем данные print("[/] Message is too long.")

Там, случаем, не recv_len > 4096 должно было быть?

 

[Mark Klintov]

* Открываем и пишем
Ну и вообще, прогоните проверку орфографии


Это лимит на количество подключений, а не время дисконнекта -

Ссылка скрыта от гостей

Там, случаем, не recv_len > 4096 должно было быть?

Спасибо, что заметили ошибки! Все недочеты из вашего комментария были исправлены в статье

 

[WithoutIrony]

data = self.socket.recv(4096) # Размер буфера в битах

Размер в байтах, не битах

self.socket.bind((self.args.target, self.args.port)) # Биндим соединение с хостом

С каким ещё хостом? Это привязка сокета к адресу.

Не подумайте, что я придалбываюсь, но либо не пиши комментарии вообще, либо сначала разберитесь, что делает конкретный фрагмент кода.

 

[x0001]

класс