'netaddr' 태그의 글 목록

netaddr

ICMP Decoding with Python 2016.03.16

ICMP Decoding with Python

2016. 3. 16. 17:44

Socket Class from Python

https://docs.python.org/2/library/socket.html

Address Family

AF_UNIX address family : a single string
AF_INET address family : (host, port)
AF_INET6 address family : (host, port, flowinfo, scopeid)
AF_NETLINK socket : (pid, groups)
AF_TIPC address family : (addr_type, v1, v2, v3 [, scope])

non-IP based networked protocol designed for use in clustered computer environments

Socket Type

socket.SOCK_STREAM :
socket.SOCK_DGRAM :
socket.SOCK_RAW :
socket.SOCK_RDM :
socket.SOCK_SEQPACKET :

IP header & ICMP message

Scanning Example from 'Black Hat Python'

1. Create socket

2. Bind

3. ioctl

import socket

import os
import struct
from ctypes import *
import threading
import time
from netaddr import IPNetwork, IPAddress

host = "172.20.10.4"
subnet = "172.20.10.0/24"

magic_message = "PYTHONRULES!"

def udp_sender(subnet, magic_message):
    time.sleep(3)
    print "UDP sender thread"
    sender = socket.socket (socket.AF_INET, socket.SOCK_DGRAM)

for ip in IPNetwork(subnet):
        try :
            sender.sendto(magic_message, ("%s" % ip, 65212))
        except :
            pass

def parseICMPtype(in_type):
    if in_type == 0 :
        retStr = "Echo Reply"
    elif in_type == 3 :
        retStr = "Destination unreachable"
    elif in_type == 4 :
        retStr = "Source quench"
    elif in_type == 11 :
        retStr = "Time exceeded"
    elif in_type == 31 :
        retStr = "Conversation error"

return retStr

class IP(Structure) :
    _fields_ = [
        ("ihl",        c_ubyte, 4),
        ("version",     c_ubyte, 4),
        ("tos",        c_ubyte),
        ("len",     c_ushort),
        ("id",         c_ushort),
        ("offset",     c_ushort),
        ("ttl",     c_ubyte),
        ("protocol_num", c_ubyte),
        ("sum",     c_ushort),
        ("src",     c_ulong),
        ("dst",     c_ulong)
    ]

def __new__(self, socket_buffer=None) :
        return self.from_buffer_copy(socket_buffer)

def __init__(self, socket_buffer=None) :
        self.protocol_map = {1:"ICMP", 6:"TCP", 17:"UDP"}
        self.src_address = socket.inet_ntoa(struct.pack("<L",self.src))
        self.dst_address = socket.inet_ntoa(struct.pack("<L",self.dst))

try :
            self.protocol = self.protocol_map[self.protocol_num]
        except :
            self.protocol = str(self.protocol_num)

class ICMP(Structure):
    _fields_ = [
        ("type",         c_ubyte),
        ("code",        c_ubyte),
        ("checksum",        c_ushort),
        ("unused",        c_ushort),
        ("next_hop_mtu",    c_ushort)
    ]

def __new__ ( self, socket_buffer ):
        return self.from_buffer_copy(socket_buffer)

def __init__( self, socket_buffer ):
        pass

if os.name == "nt":
    socket_protocol = socket.IPPROTO_IP
else :
    socket_protocol = socket.IPPROTO_ICMP

# create a raw socket and bind it to the public interface
sniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol)
sniffer.bind( (host, 0) )

# include IP headers
sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)

if os.name == "nt":
    # receive all packages
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)

t = threading.Thread(target=udp_sender, args=(subnet,magic_message))
t.start()
print "Thread has started"

try :
    while True:
        # receive a package
        raw_buffer = sniffer.recvfrom(65565)[0]

ip_header = IP(raw_buffer[0:20])

print "Protocol: %s %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)
        
        if ip_header.protocol == "ICMP" :
            offset = ip_header.ihl * 4
            buf = raw_buffer[offset:offset + sizeof(ICMP)]
            icmp_header = ICMP(buf)
            strType = parseICMPtype(icmp_header.type)
            print "ICMP -> Type : %s Code : %d" % ( strType , icmp_header.code)

if icmp_header.code == 3 and icmp_header.type == 3 :
            if IPAddress(ip_header.src_address) in IPNetwork(subnet):
                if raw_buffer[len(raw_buffer)-len(magic_message):] == magic_message :
                    print "Host Up: %s" % ip_header.src_address

except KeyboardInterrupt :
    if os.name == "nt":
        # disabled promiscuous mode
        sniffer.ioctl(socket.SIO_RCVALL, socket,RCVALL_OFF)
    print "sniff end"
</SPAN>

pip install netaddr

python scanner.py
Traceback (most recent call last):
File "scanner.py", line 7, in <module>
from netaddr import IPNetwork, IPAddress
ImportError: No module named netaddr

References :

http://www.kalitutorials.net/2014/04/denial-of-service-methods-icmp-syn.html

저작자표시 비영리

'Security&Encryption > Black Hat' 카테고리의 다른 글

Windows Debugger with PyDev (0)	2016.06.29
CPU registers and dis-assemble of Python (0)	2016.06.29
Pydev on Eclipse (0)	2016.06.28
Trojan using Git hub (0)	2016.03.30

PREV 1 NEXT

DailyBread

netaddr

ICMP Decoding with Python

'Security&Encryption > Black Hat' 카테고리의 다른 글

+ Recent posts

티스토리툴바