ICMP (Internet Control Message Protocol) 프로그래밍

ICMP

- IP프로토콜은 호스트와의 관리 질의를 위한 메커니즘이 없다

- 호스트는 간혹 라우터나 다른 호스트가 동작하고 있는지 알 필요가 있다

- 그리고 네트워크 관리자는 다른 호스트나 라우터로부터 정보를 획득할 필요가 있다

- ICMP 메시지는 오류보고 메시지와 질의 메시지로 나누어진다

- 오류보고 메시지 : 라우터나 호스트가 IP패킷을 처리하는 도중에 탐지하는 문제를 보고

- 질의 메시지 : 호스트나 네트워크 관리자가 라우터나 다른 호스트로부터 특정 정보를 획득하기 위해 사용

Type(1byte)

Code(1byte)

Checksum(2byte)

Identifier(2byte)

Sequence (2byte)

Payload(data)

[ 질의 메시지 구조]

[ 실습 ]

에코요청 메시지( type 8 ) 와 에코응답 메시지( type 0 ) 의 쌍을 이용해 다른 호스트나 라우터가 활성화 되어있는지 확인할 수 있다

#vi ip.py

import struct

class Ip: 

  def __init__( self, raw=None ):

    if raw != None:

      self._ver_and_len = raw[:1]

      self._service = raw[1:2]

      self._total = raw[2:4]

      self._id = raw[4:6]

      self._flag_and_offset = raw[6:8]

      self._ttl = raw[8:9]

      self._type = raw[9:10]

      self._chksum = raw[10:12]

      self._src = raw[12:16]

      self._dst = raw[16:20]

    else:

      self._flag_and_offset = b'\x00\x00'

      self._ver_and_len = b'\x00'

  @property

  def header( self ):

    return self._ver_and_len + self._service + self._total + self._id + \ 

           self._flag_and_offset + self._ttl + self._type + self._chksum + \ 

           self._src + self._dst

  @property

  def ver( self ):

    (ver,) = struct.unpack('!B', self._ver_and_len )

    ver = ver >> 4

    return ver 

  @ver.setter        

  def ver( self, ver ):

    (len,) = struct.unpack('!B', self._ver_and_len )

    len = len & 0x0F

    ver = ver << 4

    tmp = ver + len

    self._ver_and_len = struct.pack('!B', tmp )

  @property

  def length( self ):

    (len,) = struct.unpack('!B', self._ver_and_len )

    len = ( len & 0x0F ) << 2

    return len

  @length.setter

  def length( self, len ):

    (ver,) = struct.unpack('!B', self._ver_and_len )

    ver = ver & 0xF0

    len = len >> 2

    tmp = ver + len

    self._ver_and_len = struct.pack('!B', tmp )

  @property

  def service( self ):

    (service,) = struct.unpack('!B', self._service )

    return service

  @service.setter

  def service( self, service ):

    self._service = struct.pack('!B', service )

  @property

  def total( self ):

    (total,) = struct.unpack('!H', self._total )

  @total.setter

  def total( self, total ):

    self._total = struct.pack('!H', total )

  @property

  def id( self ):

    (id,) = struct.unpack('!H', self._id )

    return id

  @id.setter

  def id( self, id ):

    self._id = struct.pack('!H', id )

  @property

  def flag( self ):

    (flag,) = struct.unpack('!H', self._flag_and_offset )

    flag = flag >> 13

    return flag

  @flag.setter    #flag는 3bit만을 사용한다

  def flag( self, flag ):

    (offset,) = struct.unpack('!H', self._flag_and_offset)

    offset = offset & 0x1FFF

    flag = flag << 13

    tmp = flag + offset

    self._flag_and_offset = struct.pack('!H', tmp )

  @property

  def offset( self ):

    (offset,) = struct.unpack('!H', self._flag_and_offset )

    offset = (offset & 0x1FFF) << 2

    return offset

  @offset.setter

  def offset( self, offset ):

    (flag,) = struct.unpack('!H', self._flag_and_offset )

    flag = flag & 0xE000

    offset = offset >> 2

    tmp = flag + offset

    self._flag_and_offset = struct.pack('!H', tmp)

  @property

  def ttl( self ):

    (ttl,) = struct.unpack('!B', self._ttl )

    return ttl

  @ttl.setter

  def ttl( self, ttl ):

    self._ttl = struct.pack('!B', ttl )

  @property

  def type( self ):

    (type,) = struct.unpack('!B', self._type )

    return type

  @type.setter

  def type( self, type ):

    self._type = struct.pack('!B', type )

  @property

  def chksum( self ):

  def chksum( self ):

    (chksum,) = struct.unpack('!H', self._chksum )

    return chksum

  @chksum.setter

  def chksum( self, chksum ):

    self._chksum = struct.pack('!H', chksum )

  @property

  def src( self ):

    src = struct.unpack('!4B', self._src )

    src = '%d.%d.%d.%d' % src

    return src

  @src.setter

  def src( self, ip ):

    ip = ip.split('.')

    for i in range( len(ip) ):

      ip[i] = int( ip[i] )

    self._src = b''

    for i in range( len(ip) ):

      self._src += struct.pack('!B', ip[i] )

  @property

  def dst( self ):

    dst = struct.unpack('!4B', self._dst )

    dst = '%d.%d.%d.%d' % dst

    return dst

  @dst.setter

  def dst( self, ip ):

    ip = ip.split('.')

    for i in range( len(ip) ):

      ip[i] = int( ip[i] )

    self._dst = b''

    for i in range( len(ip) ):

      self._dst += struct.pack('!B', ip[i] )

#vi icmp.py

import struct

class Icmp:

  

  def __init__( self, raw=None ):

    if raw != None:

      self._type = raw[:1]

      self._code = raw[1:2]

      self._chksum = raw[2:4]

  @property

  def header( self ):

    return self._type + self._code + self._chksum

  @property

  def type( self ):

    (type,) = struct.unpack('!B', self._type )

    return type

  @type.setter

  def type( self, type ):

    self._type = struct.pack('!B', type )

  @property

  def code( self ):

    (code,) = struct.unpack('!B', self._code )

    return code

  @code.setter

  def code( self, code ):

    self._code = struct.pack('!B', code )

  @property

  def chksum( self ):

    (chksum,) = struct.unpack('!H', self._chksum )

    return chksum

  @chksum.setter

  def chksum( self, chksum ):

    self._chksum = struct.pack('!H', chksum )

      

class Echo( Icmp ):    # ICMP클래스 상속

  def __init__( self, raw=None ):

    if raw != None:

      self._id = raw[:2]

      self._seq = raw[2:4]

      self._payload = raw[4:]

  

  @property 

  def header( self ):

    return self._type + self._code + self._chksum + self._id + self._seq + self._payload

  

  @property

  def id( self ):

    (id,) = struct.unpack('!H', self._id )

    return id

    

  @id.setter

  def id( self, id ):

    self._id = struct.pack('!H', id )

    

  @property

  def seq( self ):

    (seq,) = struct.unpack('!H', self._seq )

    return seq

  @seq.setter

  def seq( self, seq ):

    self._seq = struct.pack('!H', seq )

  @property

  def payload( self ):

    return self._payload.decode( errors=ignore )

  @payload.setter

  def payload( self, payload ):

    self._payload = payload.encode()

#vi packet.py

from header.eth import *

from header.ip import *

from header.udp import *

from header.arp import *

from header.icmp import *

class Packet:

  def __init__( self, raw ):

    self._eth = Eth( raw[:14] )

    if self._eth.type == 0x0800:        # IPv4통신이라면?

      self.analyze_ip( raw[14:] )         # ip클래스객체 생성함수 실행

    elif self._eth.type == 0x0806:      # arp

      self.analyze_arp( raw[14:] )        # arp클래스객체 생성함수 실행

  def analyze_ip( self, raw ):            # ip클래스객체 생성함수

    self._ip = Ip( raw )                    # ip클래스 객체 생성

    if self._ip.type == 17:                # ip의 protocol type == 17 : UDP통신

      self.analyze_udp( raw[20:] )

    elif self._ip.type == 1:                # ip의 protocol type == 1 : icmp통신

      self.analyze_icmp( raw[20:] )

  def analyze_icmp( self, raw ):

    self._icmp = Icmp( raw )

  def analyze_arp( self, raw ):

    self._arp = Arp( raw )

  def analyze_udp( self, raw ):

    self._udp = Udp( raw )

  @property

  def icmp( self ):

    return self._icmp

  @property

  def eth( self ):

    return self._eth

  @property

  def ip( self ):

    return self._ip

  @property

  def udp( self ):

    return self._udp

  @property

  def arp( self ):

    return self._arp

#vi iping.py : setter를 이용한 각 클래스의 멤버변수 초기화

import struct

import socket

from header.eth import *

from header.ip import *

from header.icmp import *

def make_chksum( header ):        #체크섬 계산 함수

  size = len( header )

  if size % 2:

    header = header + b'\x00'

    size = len( header )

  size = size // 2

  header = struct.unpack('!' + str(size) + 'H', header )

  chksum = sum( header )

  carry = chksum & 0xFF0000

  carry = carry >> 16

  while carry != 0:

    chksum = chksum & 0xFFFF

    chksum = chksum + carry

    carry = chksum & 0xFF0000

    carry = carry >> 16

  chksum = chksum ^ 0xFFFF

  return chksum  

eth = Eth()

ip = Ip()

echo = Echo()

echo.type = 8 

echo.code = 0 

echo.chksum = 0

echo.id = 0xabcd

echo.seq = 1

echo.payload = 'qazwsxedcrfvtgbyhnujmikopABCDCEFERTUACJI3456789'

echo.chksum = make_chksum( echo.header )

ip.ver = 4

ip.length = 20

ip.service = 0

ip.total = 20 + len( echo.header )

ip.id = 0x1234

ip.flag = 0

ip.offset = 0

ip.ttl = 64

ip.type = 1

ip.chksum = 0

ip.src = '192.168.6.123'

ip.dst = '192.168.6.200'py

ip.chksum = make_chksum( ip.header )

eth.dst = '00:50:56:31:a8:43'

eth.src = '0:0C:29:F0:62:73'

eth.type = 0x0800

sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW )

sock.bind( ('eth0', socket.SOCK_RAW) )

sock.send( eth.header + ip.header + echo.header )

#vi sniffer.py

import socket

import struct

import time

from header.packet import *

raw = socket.socket( socket.PF_PACKET, socket.SOCK_RAW )

raw.bind( ('eth0', socket.SOCK_RAW) )

while True:

  data, addr = raw.recvfrom( 65535 )

  packet = Packet( data )

  if packet.eth.type == 0x0800 and  (packet.ip.src == '192.168.6.123' or packet.ip.dst == '192.168.6.123') \

and  packet.ip.type == 1:       

    print( "data:", data)

    print( packet.ip.src + ' -> ' + packet.ip.dst )

    print( "type:", packet.icmp.type, "code:", packet.icmp.code )

    print()

[ 실행결과 ]

[ 그림1 ]

setter를 이용해 각 멤버변수들을 초기화하고 socket으로 데이터를 전송

eth0으로 들어오고 나가는 패킷들을 sniffer에서 출력하고 있다

에코요청메시지 request 에 대한 응답으로 에코응답메시지 response 가 들어온다

# 이더넷 헤더와 아이피 헤더의 역할 #

이더넷 헤더는 같은 네트워크 상에서 MAC주소를 이용한 통신

아이피 헤더는 다른 네트워크 상에 있는 호스트와 통신하기 위해 IP주소를 사용한다

호스트 => " 라우터 " => 다른 네트워크 상의 호스트 : 패킷이 라우터를 통해서 외부로 빠져나간다

외부와의 ICMP통신을 해보자 !

1> 이더넷 헤더에서 맥주소는 라우터의 맥주소를 입력한다

2> 아이피 헤더에서 아이피주소는 외부 네트워크의 아이피주소를 입력한다

=> 처음 호스트에서 라우터로 패킷을 보낸다. 

=> 패킷의 아이피 헤더에 써있는 dst( 외부 IP주소 )를 읽고 외부의 해당 네트워크로 패킷이 이동된다( ex)request메시지 보냈다 )

=> 외부에서 해당 패킷을 읽고 라우터의 맥주소를 보고 라우터로 패킷( ex)response메시지 )을 보낸다

=> 라우터는 외부에서 온 아이피 헤더에 써있는 dst( 호스트IP )를 읽어서 처음 호스트에게 패킷을 보내준다

=> 내부와 외부에서의 통신이 성공했다

[ 실행결과 ]