Реализация NetCat на Python

15.07.2026 19:01

Приветствую всех пользователей форума LIMB.cc
Сделано эксклюзивно для форума LIMB.cc

В данной статье я хотел бы продемонстрировать самописный netcat, его функции и задачи. Задумкой послужило давно прочтенная мной тема в Black Hat Python, 2nd Edition, в ней описывалась концепция "слушания" сервера.

Что такое NetCat

Netcat — утилита для установки соединения между клиентом и сервером, использующая TCP/UDP протоколы. Netcat позволяет запускать команды или оболочку удалённо, сканировать порты, передавать файлы и т.д. Сама концепция в том, чтобы слушать определённый порт — то есть ожидать входящее соединение или инициировать его.

Я буду использовать Oracle VM Kali как сервер, Windows 11 как хост.

Язык написания скрипта: Python

Немного теории.

Основные задачи это:
1. Удаленно запускать команды
2. Пересылать файлы

То есть мы как бы получаем доступ к сессии, и передаем на сервер команду которую мы хотим там выполнить, после чего команда обрабатывается в оболочке шелла как обычно.

Не удалось прогрузить медиа

Подключаемся мы только к одному сокету

Практика.

Сначала настраиваем тип подключения в ВМ

Не удалось прогрузить медиа

Выбираем сетевой мост

После чего заходим, запускаем VS Code, и пишем скрипт для сервера(мой скрипт не точная копия из книжки Black Hat Python, он немного переделан для удобства и актуальной поддержки, урезана: без передачи файлов и т.д)

Подключаем либы:

Python:

import argparse     # Для обработки аргументов командной строки (IP, порт, флаги)
import socket                                    # Работа с TCP-соединением (создание сокета)
import shlex                                       # Безопасное разбиение командной строки
import subprocess                           # Запуск внешних команд (например, whoami, ls и т.д.)
import sys                                          # Для выхода из программы при ошибке
import threading                              # Для обработки нескольких клиентов одновременно

Создаем класс с конструктором и методами:

Python:

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)

args - аргументы из командной строки
buffer - используется для стартового сообщения
socket() - создает TCP-сокет
SO_REUSEADDR - повторное использование порта без ожидания

Запускаем слушание:

Python:

def run(self):
    self.listen()

Пишем сам метод:

Python:

def listen(self):
        try:
            self.socket.bind((self.args.target, self.args.port))
            self.socket.listen(5)
            print(f"[*] Listening on {self.args.target}:{self.args.port}...")

            while True:
                client_socket, client_address = self.socket.accept()
                print(f"[+] Connection established from {client_address[0]}:{client_address[1]}")
                client_thread = threading.Thread(target=self.handle, args=(client_socket,))
                client_thread.start()
        except KeyboardInterrupt:
            print("[!] Server terminated by user.")
        except Exception as e:
            print(f"[-] Error in listen: {e}")
        finally:
            self.socket.close()

Привязываем ip и порт(bind)
Далее начинаем слушать - self.socket.listen(5)
При подключении клиента запускает новый поток (threading.Thread), чтобы обслуживать каждого клиента независимо

Метод handle():

Python:

def handle(self, client_socket):
        try:
            while True:
                try:
                    client_socket.send(b'BHP: #> ')
                    cmd_buffer = b''
                    while b'\n' not in cmd_buffer:
                        cmd_buffer += client_socket.recv(64)
                    cmd = cmd_buffer.decode().strip()
                    if not cmd:
                        continue
                    response = execute(cmd)
                    if response:
                        client_socket.send(response.encode())
                except Exception as e:
                    print(f"[-] Server killed: {e}")
                    break
        finally:
            client_socket.close()

Оправляем само приглашение - client_socket.send(b'BHP: #> ')
Далее простая реализация ожидания и возврата команды клиента

Функция execute(cmd):

Python:

def execute(cmd):
    cmd = cmd.strip()
    if not cmd:
        return ''
    try:
        output = subprocess.check_output(shlex.split(cmd), stderr=subprocess.STDOUT)
        return output.decode()
    except subprocess.CalledProcessError as e:
        return f"Command failed with error: {e.output.decode()}"
    except Exception as e:
        return f"Failed to execute command: {str(e)}"

Это само так скажем ядро, сначала делаем команды корректными, добавляем проверки
Метод реализует саму интерпретацию команд в терминал, эмуляцию того как будто мы сами ее вводим

Ну и обработка аргументов:

Python:

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='BHP Net Tool')
    parser.add_argument('-t', '--target', default='0.0.0.0', help='specified IP')
    parser.add_argument('-p', '--port', type=int, default=5555, help='specified port')
    parser.add_argument('-c', '--command', action='store_true', help='command shell')
    args = parser.parse_args()
   
    buffer = b''
    nc = NetCat(args, buffer)
    nc.run()

Все, сам скрипт готов, можем запускать

Не удалось прогрузить медиа

После чего заходим в сам Windows(клиент) и пишем скрипт теперь для него:

Либы:

Python:

import argparse
import socket    
import sys

Класс NetCat и инициализация:

Python:

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)

self.args - аргументы, переданные из командной строки
self.buffer - байтовый буфер, который можно сразу отправить после подключения
Создаётся TCP-сокет(AF_INET, SOCK_STREAM)

Далее запуск, метод def run(self):

Python:

def run(self):
    self.send()

Сам метод send(основа кода):

Python:

def send(self):
    try:
        self.socket.connect((self.args.target, self.args.port))
        print(f"[+] Connected to {self.args.target}:{self.args.port}")
        if self.buffer:
            self.socket.send(self.buffer)

        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:
                    break
           
            if response:
                print(response, end="")
                buffer = input('> ')
                buffer += '\n'
                self.socket.send(buffer.encode())

Осуществляем подключение к IP и порту
Проверка на изначальный буфер - if self.buffer:
После чего идет цикл для обмена данными
• Получаем данные из сокета - 4096 байт
• Складываем в строку response
• if recv_len < 4096 - проверка на кол-во байт, если меньше пакет закончился, выходим с ВНУТРЕННЕГО цикла
• Если ответ есть печатаем его в терминале без доп. переноса строки - end=""
• Ввод пользователем данных - input('> ')
• Добавляем перевод строки и отправляем обратно на сервер

Обработка исключений:

Python:

except ConnectionRefusedError:
        print(f"[-] Connection to {self.args.target}:{self.args.port} refused.")
    except Exception as e:
        print(f"[-] Error: {e}")
    except KeyboardInterrupt:
        print('[!] User terminated the connection.')
        self.socket.close()
        sys.exit()

Дефолт вывод сообщений об информации о подключении
Соединения нет - увед
Ошибка - увед и т.д

Точка входа:

Python:

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='BHP Net Tool')
    parser.add_argument('-t', '--target', default='192.168.0.80', help='specified IP')
    parser.add_argument('-p', '--port', type=int, default=5555, help='specified port')
    args = parser.parse_args()
   
    buffer = b''
    nc = NetCat(args, buffer)
    nc.run()

Парс аргументов
Таргет по умолчанию - 192.168.0.80:5555

Подключение.
После запуска скрипта на Linux, давайте запустим на Windows, сначала узнаем адрес Kali
Заходим в кали и пишем ip a

Не удалось прогрузить медиа

нам нужен 192.168.0.80/24

Открываем терминал(windows), указываем путь к скрипту и запускаем

Не удалось прогрузить медиа

Успешное подключение, теперь давайте затестим пару команд:

Не удалось прогрузить медиа

Как мы видим все успешно работает.

Заключение.

В этой теме я привел простейшую работающую реализацию NetCat на Python, это самая урезанная версия чисто для наглядности.

Ссылки:
Код для сервера: https://pastecode.dev/s/yxlyjrlw
Код для клиента: https://pastecode.dev/s/ip3jyz3c
Книга которая послужила идее: https://github.com/fedcal/libriInformatica/blob/master/Black%20Hat%20Python%2C%202nd%20Edition%20-%20Justin%20Seitz.pdf

Список библиотек:
import argparse
import socket
import shlex
import subprocess
import sys
import threading

Не удалось прогрузить медиа

by DESTR0Y
Сделано эксклюзивно для форума LIMB.cc

👍43 🔥4 😁9 🤔8 😢8