modscan.py

#!/usr/bin/env python


"""

File: modscan.py
Desc: Modbus TCP Scanner
Version: 0.1

Copyright (c) 2008 Mark Bristow

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation version either version 3 of the License,
or (at your option) any later version.


This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.

ModScan is a new tool designed to map a SCADA MODBUS TCP based network. The tool is written in python for portability and can be used on virtually any system with few required libraries.

"""

import socket
import array
import optparse
from IPy import IP
import sys

def main():

  p = optparse.OptionParser(  description=' Finds modbus devices in IP range and determines slave id.\nOutputs in ip:port <tab> sid format.',
                prog='modscan',
                version='modscan 0.1',
                usage = "usage: %prog [options] IPRange")
  p.add_option('--port', '-p', type='int', dest="port", default=502, help='modbus port DEFAULT:502')
  p.add_option('--timeout', '-t', type='int', dest="timeout", default=500, help='socket timeout (mills) DEFAULT:500')
  p.add_option('--aggressive', '-a', action ='store_true', help='continues checking past first found SID')
  p.add_option('--function', '-f', type='int', dest="function", default=17, help='MODBUS Function Code DEFAULT:17')
  p.add_option('--data', type='string', dest="fdata", help='MODBUS Function Data.  Unicode escaped "\x00\x01"')
  p.add_option('-v', '--verbose', action ='store_true', help='returns verbose output')
  p.add_option('-d', '--debug', action ='store_true', help='returns extremely verbose output')

  options, arguments = p.parse_args()

  #make sure we have at least 1 argument (IP Addresses)
  if len(arguments) == 1:

    #build basic packet for this test

    """
    Modbus Packet Structure
    \x00\x00  \x00\x00  \x00\x00  \x11    \x00    <=================>
    Trans ID  ProtoID(0)  Length    UnitID    FunctCode  Data len(0-253byte)
    """

    #this must be stored in a unsigned byte aray so we can make the assignment later... no string[] in python :(
    rsid = array.array('B')
    rsid.fromstring("\x00\x00\x00\x00\x00\x02\x01\x01")

    #set function
    rsid[7]=options.function

    #add function data
    if (options.fdata):
      aFData = array.array('B')

      #we must decode the escaped unicode before calling fromstring otherwise the literal \xXX will be interpreted
      aFData.fromstring(options.fdata.decode('unicode-escape') )
      rsid += aFData

      #update length
      rsid[5]=len(aFData)+2

    #assign IP range
    iprange=IP(arguments[0])

    #print friendly user message
    print "Starting Scan..."

    #primary loop over IP addresses
    for ip in iprange:

      #print str(ip)+" made it"
      #loop over possible sid values (1-247)
      for sid in range (1, 247):

        #error messaging
        fError=0
        msg = str(ip)+":"+str(options.port)+"\t"+str(sid)

        #print "msg="+msg

        #Wrap connect in a try box
        try:
          #socket object instantiation
          s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

          #set socket timeout, value from cmd is in mills
          s.settimeout(float(options.timeout) / float(1000))

          #connect requires ip addresses in string format so it must be cast
          s.connect((str(ip), options.port))

        except socket.error:
          #clean up
          fError=1
          msg += "\tFAILED TO CONNECT"
          s.close()
          break
        #end try


        #send query to device
        try:
          #set slave id
          rsid[6]=sid

          #send data to device
          s.send(rsid)

        except socket.error:
          #failed send close socket
          fError=1
          msg += "\tFAILED TO SEND"
          s.close()
          break
        #end try

        try:

          #recieve data
          data = s.recv(1024)

        except socket.timeout:
          fError=1
          msg += "\tFAILED TO RECV"
          break
        #end try

        #examine response
        if data:
          #parse response
          resp = array.array('B')
          resp.fromstring(data)

          if (options.debug):
            print "Recieved: "+str(resp)
            print (int(resp[7]) == int(options.function))

          #if the function matches the one sent we are all good
          if (int(resp[7]) == int(options.function)):
            print msg

            #in aggressive mode we keep going
            if (not options.aggressive):
              break

          #If the function matches the one sent + 0x80 a positive response error code is detected
          elif int(resp[7]) == (int(options.function)+128):
            #if debug output message
            msg += "\tPositive Error Response"
            if (options.debug):
              print msg
          else:
            #if debug output message
            if (options.debug):
              print msg
        else:
          fError=1
          msg += "\tFAILED TO RECIEVE"
          s.close()
          break

      #end SID for


      #report based on verbosity
      if (options.verbose and fError):
        print msg
      elif (options.debug):
        print msg
    #end IP for

    #close socket, no longer needed
    #s.shutdown(socket.SHUT_RDWR)
    s.close()

    print "Scan Complete."

  #bad number of arguments.  print help
  else:
    p.print_help()


if __name__ == '__main__':
  try : main()
  except KeyboardInterrupt:
    print "Scan canceled by user."
    print "Thank you for using ModScan"
  except :
    sys.exit()