Как использовать Raw Socket в Python?
Я пишу приложение для тестирования сетевого драйвера для обработки поврежденных данных. И я подумал о отправке этих данных с помощью raw-сокета, поэтому он не будет исправлен стеком TCP-IP отправляющей машины.
Я пишу это приложение исключительно на Linux. У меня есть примеры кода для использования сырых сокетов в системных вызовах, но мне бы очень хотелось, чтобы мой тест был как можно более динамичным, и пишите, если не все, в Python.
Я немного искал интернет для объяснений и примеров использования сырых сокетов в python, но не нашел ничего действительно просветительного. Просто очень старый пример кода, который демонстрирует эту идею, но ни в коем случае не работает.
Из того, что я собрал, использование Raw Socket в Python почти идентично семантике для сырого сокета UNIX, но без struct
, которые определяют структуру пакетов.
Мне было интересно, не было бы лучше не писать сырую сокетную часть теста в Python, а в C с системными вызовами и вызывать его из основного кода Python?
Ответы
Ответ 1
Вы делаете это так:
Сначала отключите автоматическую проверку контрольной суммы вашей сетевой карты:
sudo ethtool -K eth1 tx off
А затем отправьте свой хитрый кадр из Python 2 (вам придется конвертировать в Python 3 самостоятельно):
#!/usr/bin/env python
from socket import socket, AF_PACKET, SOCK_RAW
s = socket(AF_PACKET, SOCK_RAW)
s.bind(("eth1", 0))
# We're putting together an ethernet frame here,
# but you could have anything you want instead
# Have a look at the 'struct' module for more
# flexible packing/unpacking of binary data
# and 'binascii' for 32 bit CRC
src_addr = "\x01\x02\x03\x04\x05\x06"
dst_addr = "\x01\x02\x03\x04\x05\x06"
payload = ("["*30)+"PAYLOAD"+("]"*30)
checksum = "\x1a\x2b\x3c\x4d"
ethertype = "\x08\x01"
s.send(dst_addr+src_addr+ethertype+payload+checksum)
Готово.
Ответ 2
Системные вызовы сокетов (или Winsocks, в Windows) уже завернуты в стандартный модуль socket
: intro, ссылка.
Я никогда не использовал сырые сокеты, но похоже, что они могут использоваться с этим модулем:
В последнем примере показано, как написать очень простой сетевой сниффер с сырым сокетов в Windows. Пример требует прав администратора изменить интерфейс:
import socket
# the public network interface
HOST = socket.gethostbyname(socket.gethostname())
# create a raw socket and bind it to the public interface
s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP)
s.bind((HOST, 0))
# Include IP headers
s.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
# receive all packages
s.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
# receive a package
print s.recvfrom(65565)
# disabled promiscuous mode
s.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)
Ответ 3
Является этот старый код, который вы упомянули о поиске? Мне это кажется разумным, но я сам его не тестировал (или использовал сырые сокеты). Этот пример из документации показывает, как использовать сырые сокеты для обнюхивания пакетов, и он выглядит достаточно похожим.
Ответ 4
В конечном итоге лучшим решением для этого случая было написать всю вещь в C, потому что это не большое приложение, поэтому для написания такой мелочи было бы больше штрафа на более чем 1 языке.
После долгих игр с C и python RAW-сокетами я в конечном итоге предпочел сокеты C RAW. Сокеты RAW требуют модификации на уровне бит менее 8 бит для записи заголовков пакетов. Иногда записывается только 4 бит или меньше. python не помогает в этом, тогда как Linux C имеет полный API для этого.
Но я определенно полагаю, что если бы только эта небольшая инициализация заголовка была удобна в python, я бы никогда не использовал C здесь.
Ответ 5
s = socket(AF_PACKET, SOCK_RAW)
s = socket(PF_PACKET, SOCK_RAW)
результат:
[[email protected] python]# tcpdump -i eth0
capture size 96 bytes
11:01:46.850438
01:02:03:04:05:06 (oui Unknown) > 01:02:03:04:05:06 (oui Unknown), ethertype Unknown (0x0801), length 85:
0x0000: 5b5b 5b5b 5b5b 5b5b 5b5b 5b5b 5b5b 5b5b [[[[[[[[[[[[[[[[
0x0010: 5b5b 5b5b 5b5b 5b5b 5b5b 5b5b 5b5b 5041 [[[[[[[[[[[[[[PA
0x0020: 594c 4f41 445d 5d5d 5d5d 5d5d 5d5d 5d5d YLOAD]]]]]]]]]]]
0x0030: 5d5d 5d5d 5d5d 5d5d 5d5d 5d5d 5d5d 5d5d ]]]]]]]]]]]]]]]]
0x0040: 5d5d 5d00 0000 00 ]]]....
Ответ 6
Класс сокета должен помочь. Если нет, вам нужно либо написать модуль Python на C, либо просто использовать C. См. http://mail.python.org/pipermail/python-list/2001-April/077454.html.
Основной поиск Google нашел, что.
Я действительно попробовал пример кода, который указал "unwind". AF_PACKET действительно работал у меня в python 2.7.4
Ответ 7
FTR, если вам нужен доступ 2-го уровня (Ethernet, RadioTap...), это будет невозможно изначально в Windows (на сегодняшний день).
Если вы хотите получить доступ к тем с помощью кроссплатформенного метода, вы можете выбрать libpcap
и его привязки к Python (так как он будет использовать Npcap/WinPcap
для работы в Windows).
У вас есть множество привязок Python, доступных для libpcap, на всех уровнях (очень высокий или очень низкий).
Мой совет - использовать Sockets of scapy
(даже если вы не используете его для анализа), который реализует как вызовы Native, так и Libpcap (и пусть вы выбираете против них с conf.use_pcap = True
).
from scapy.all import conf
# conf.use_pcap = True (will be automatic if required)
socket = conf.L2socket(iface="eth0")
# On any platforms, you have 'get_if_list()' in 'scapy.all' available, to see the ifaces available. You could also ignore it to use the default one
Ответ 8
Вы можете использовать эту библиотеку Python: rawsocketpy, она позволяет использовать необработанные сокеты на уровне 2 => Нет заголовков IP/TCP/UDP.
#!/usr/bin/env python
from rawsocketpy import RawSocket
sock = RawSocket("wlp2s0", 0xEEFA)
sock.send("some data")
sock.send("personal data", dest="\xAA\xBB\xCC\xDD\xEE\xFF")
или форма сервера:
#!/usr/bin/env python
from rawsocketpy import RawRequestHandler, RawAsyncServerCallback
import time
def callback(handler, server):
print("Testing")
handler.setup()
handler.handle()
handler.finish()
class LongTaskTest(RawRequestHandler):
def handle(self):
time.sleep(1)
print(self.packet)
def finish(self):
print("End")
def setup(self):
print("Begin")
def main():
rs = RawAsyncServerCallback("wlp2s0", 0xEEFA, LongTaskTest, callback)
rs.spin()
if __name__ == '__main__':
main()