RubyでICMPを使ったネットワークプログラミング

今回はRubyでルートスキャナーを作ります。

Rubyでのプログラミングには慣れていませんが、前回ポートスキャナーを作った時の記憶が残っているのでそれを頼りに実装していきたいと思います。

ルートスキャナーについては以前に記事を書きました。

ICMPを使ったネットワークプログラム

1 必要なこと
- 1.1 ソケット
- 1.2 パケット
2 ソースコード全体
3 感想

必要なこと

tarcerouteを実装する場合いろいろ方法はありますが、
前回記事にした時と同様にicmpエコー要求を使う方向で実装したいと思います。

そのために必要なことをまとめます。

ソケット

ポートスキャンと違いルートをトレースするためにipパケットのttlフィールドをこちらで設定します。
ですのでカーネルに「ipパケットもこちらでつくるよ」ということを伝える必要があります。

    @soc = Socket::open(Socket::AF_INET,Socket::SOCK_RAW,Socket::IPPROTO_ICMP)
    @soc.setsockopt(Socket::IPPROTO_IP,Socket::IP_HDRINCL,1)

1 2	@soc = Socket::open(Socket::AF_INET,Socket::SOCK_RAW,Socket::IPPROTO_ICMP) @soc.setsockopt(Socket::IPPROTO_IP,Socket::IP_HDRINCL,1)

icmpのソケットを作成し、setsockoptでipヘッダーも作成するよう設定します。

パケット

今回はipパケットとicmpパケットを作成します。

ipパケットを作成する関数です。
パケット送信時にttlを指定します。

  # ipパケット作成
  def make_ip(ver: 4,hl: 5,tos: 0,len: 28,id: 1,off: 0,ttl: 64,p: Socket::IPPROTO_ICMP,src: "127.0.0.1",dst: "127.0.0.1")
    ver_hl = (ver << 4) + 5
    saddr,daddr = IPAddr.new(src).to_i,IPAddr.new(dst).to_i

    cksum = 0

    data = [ver_hl,tos,len,id,off,ttl,p,cksum,saddr,daddr]
    ip_packet = data.pack('C2n3C2n1N2')
    cksum = checksum(ip_packet)
    data = [ver_hl,tos,len,id,off,ttl,p,cksum,saddr,daddr]
    data.pack('C2n3C2n1N2')
  end

# ipパケット作成

def make_ip(ver: 4,hl: 5,tos: 0,len: 28,id: 1,off: 0,ttl: 64,p: Socket::IPPROTO_ICMP,src: "127.0.0.1",dst: "127.0.0.1")

ver_hl = (ver << 4) + 5

saddr,daddr = IPAddr.new(src).to_i,IPAddr.new(dst).to_i

cksum = 0

data = [ver_hl,tos,len,id,off,ttl,p,cksum,saddr,daddr]

ip_packet = data.pack('C2n3C2n1N2')

cksum = checksum(ip_packet)

data = [ver_hl,tos,len,id,off,ttl,p,cksum,saddr,daddr]

data.pack('C2n3C2n1N2')

end

icmpパケットを作成する関数です。
icmpエコー要求パケットを送信しますので、タイプに8を指定します。

  # icmpパケット作成
  def make_icmp_echo()
    type = 8
    code = 0
    seq = 1
    id = 1
    cksum = 0

    data = [type, code, cksum, id, seq]
    cksum = checksum(data.pack('C2n3'))
    data = [type, code, cksum, id, seq]
    data.pack('C2n3')
  end

# icmpパケット作成

def make_icmp_echo()

type = 8

code = 0

seq = 1

id = 1

cksum = 0

data = [type, code, cksum, id, seq]

cksum = checksum(data.pack('C2n3'))

data = [type, code, cksum, id, seq]

data.pack('C2n3')

end

以上が今回作るツールの核になる部分です。

ソースコード全体

RouteScannerというクラスを作成しました。

ttlに設定する値として1からスタートし255までインクリメントしていきます。

readyの待ち時間を過ぎた場合は初回の込みで3回まで再送信します。

パケットを受信したら中身を確認します。
icmpのタイプがecho_replyだった場合はターゲットに到達したと判断してループを終了します。
time_exceededだった場合は再送信ループを抜けてttlをインクリメントして送信処理を続行します。

以上がこのクラスがやってくれる処理の大まかな流れです。

require 'socket'
require 'ipaddr'

class RouteScanner
  def initialize()
    @soc = Socket::open(Socket::AF_INET,Socket::SOCK_RAW,Socket::IPPROTO_ICMP)
    @soc.setsockopt(Socket::IPPROTO_IP,Socket::IP_HDRINCL,1)
  end

  # チェックサム計算
  def checksum(data)
    len = data.size / 2 * 2
    sum = 0

    0.step(len-1,2) do |i|
     sum += (data[i+1].ord << 8) + data[i].ord
    end

    if data.size % 2 != 0
      sum += data[-1].ord
    end

    while (sum >> 16) != 0
      sum = (sum >> 16) + (sum & 0xffff)
    end

    sum = sum >> 8 | (sum << 8 & 0xff00)
    return ~sum & 0xffff
  end

  # ipパケット作成
  def make_ip(ver: 4,hl: 5,tos: 0,len: 28,id: 1,off: 0,ttl: 64,p: Socket::IPPROTO_ICMP,src: "127.0.0.1",dst: "127.0.0.1")
    ver_hl = (ver << 4) + 5
    saddr,daddr = IPAddr.new(src).to_i,IPAddr.new(dst).to_i

    cksum = 0

    data = [ver_hl,tos,len,id,off,ttl,p,cksum,saddr,daddr]
    ip_packet = data.pack('C2n3C2n1N2')
    cksum = checksum(ip_packet)
    data = [ver_hl,tos,len,id,off,ttl,p,cksum,saddr,daddr]
    data.pack('C2n3C2n1N2')
  end

  # icmpパケット作成
  def make_icmp_echo()
    type = 8
    code = 0
    seq = 1
    id = 1
    cksum = 0

    data = [type, code, cksum, id, seq]
    cksum = checksum(data.pack('C2n3'))
    data = [type, code, cksum, id, seq]
    data.pack('C2n3')
  end

  # 受信パケットからipヘッダ取得
  def extract_ip(packet)
    packet.unpack('C2n3C2n1N2')
  end

  # 一応受信したパケットを確認
  def analyze(packet)
    # エコー応答
    icmp_echo_reply = 0
    # 時間超過
    icmp_timex = 11

    icmp = packet.unpack('C2n3')
    type = icmp[0]

    if type == icmp_echo_reply
      return 1
    elsif type == icmp_timex
      return 0
    end
    2
  end

  # スキャン開始
  def scan(target_ip)
    puts "Scanning to #{target_ip} ..."

    # 到達したかどうか
    reach = 0

    # 1~255までttlをインクリメントする
    for i in 1..256
      packet = make_ip(src: "0.0.0.0", dst: target_ip, ttl: i)+make_icmp_echo()
      addr = Socket.pack_sockaddr_in(0, target_ip)

      # 3回繰り返す
      for _ in 0..3

        @soc.send(packet, 0, addr)
        sel = IO.select([@soc], nil, nil, 10)
        if sel != nil && sel_soc = sel[0]
          packet, sockaddr = sel_soc[0].recvfrom(1024)
          ip = extract_ip(packet)
          # protocol
          if ip[6] == Socket::IPPROTO_ICMP
            sockaddr = Socket.unpack_sockaddr_in(sockaddr)
            start = (ip[0] & 0x0f) * 4
            res = analyze(packet[start...start+8])

            # 1なら到達 0なら時間超過
            if res == 1
              puts "- Reach time to live #{i} from #{sockaddr[1]}"
              reach = 1
              break
            elsif res == 0
              puts " - time to live #{i} from #{sockaddr[1]}"
              break
            end
          end
        end
      end
      # 到達したらループ終了
      if reach == 1
        break
      end
    end
  end
end

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

require 'socket'

require 'ipaddr'

class RouteScanner

def initialize()

@soc = Socket::open(Socket::AF_INET,Socket::SOCK_RAW,Socket::IPPROTO_ICMP)

@soc.setsockopt(Socket::IPPROTO_IP,Socket::IP_HDRINCL,1)

end

# チェックサム計算

def checksum(data)

len = data.size / 2 * 2

sum = 0

0.step(len-1,2) do |i|

sum += (data[i+1].ord << 8) + data[i].ord

end

if data.size % 2 != 0

sum += data[-1].ord

end

while (sum >> 16) != 0

sum = (sum >> 16) + (sum & 0xffff)

end

sum = sum >> 8 | (sum << 8 & 0xff00)

return ~sum & 0xffff

end

# ipパケット作成

def make_ip(ver: 4,hl: 5,tos: 0,len: 28,id: 1,off: 0,ttl: 64,p: Socket::IPPROTO_ICMP,src: "127.0.0.1",dst: "127.0.0.1")

ver_hl = (ver << 4) + 5

saddr,daddr = IPAddr.new(src).to_i,IPAddr.new(dst).to_i

cksum = 0

data = [ver_hl,tos,len,id,off,ttl,p,cksum,saddr,daddr]

ip_packet = data.pack('C2n3C2n1N2')

cksum = checksum(ip_packet)

data = [ver_hl,tos,len,id,off,ttl,p,cksum,saddr,daddr]

data.pack('C2n3C2n1N2')

end

# icmpパケット作成

def make_icmp_echo()

type = 8

code = 0

seq = 1

id = 1

cksum = 0

data = [type, code, cksum, id, seq]

cksum = checksum(data.pack('C2n3'))

data = [type, code, cksum, id, seq]

data.pack('C2n3')

end

# 受信パケットからipヘッダ取得

def extract_ip(packet)

packet.unpack('C2n3C2n1N2')

end

# 一応受信したパケットを確認

def analyze(packet)

# エコー応答

icmp_echo_reply = 0

# 時間超過

icmp_timex = 11

icmp = packet.unpack('C2n3')

type = icmp[0]

if type == icmp_echo_reply

return 1

elsif type == icmp_timex

return 0

end

# スキャン開始

def scan(target_ip)

puts "Scanning to #{target_ip} ..."

# 到達したかどうか

reach = 0

# 1~255までttlをインクリメントする

for i in 1..256

packet = make_ip(src: "0.0.0.0", dst: target_ip, ttl: i)+make_icmp_echo()

addr = Socket.pack_sockaddr_in(0, target_ip)

# 3回繰り返す

for _ in 0..3

@soc.send(packet, 0, addr)

sel = IO.select([@soc], nil, nil, 10)

if sel != nil && sel_soc = sel[0]

packet, sockaddr = sel_soc[0].recvfrom(1024)

ip = extract_ip(packet)

# protocol

if ip[6] == Socket::IPPROTO_ICMP

sockaddr = Socket.unpack_sockaddr_in(sockaddr)

start = (ip[0] & 0x0f) * 4

res = analyze(packet[start...start+8])

# 1なら到達 0なら時間超過

if res == 1

puts "- Reach time to live #{i} from #{sockaddr[1]}"

reach = 1

break

elsif res == 0

puts " - time to live #{i} from #{sockaddr[1]}"

break

end

# 到達したらループ終了

if reach == 1

break

end

このクラスは以下のように使います。

target_ip = {ターゲットのipアドレス}
scanner = RouteScanner.new()
scanner.scan(target_ip)

target_ip = {ターゲットのipアドレス}

scanner = RouteScanner.new()

scanner.scan(target_ip)

パケットを受信したルーターによっては時間超過を知らせるパケットを送ってくれないこともあるようです。

感想

以前Rustとc言語でも作りましたが、やはりスクリプト言語なのか結構軽めにかけた気がします。
この調子でがむしゃらにいろいろ作っていきたいと思います。

ポートスキャンの時もそうでしたがPythonでも作ることになりそうな予感がしてます。