If only there were a windows variant

2007-06-25T01:36:00.000-07:00

syscalltrack tracks and allows for 'hijacking' of practically all kernel syscalls.

TCP Replay

2007-06-24T01:52:00.000-07:00

I put together a python script based on Scapy which allows for easy replay of Layer3 network traffic, more specifically, TCP traffic.

The script uses a pcap file as a 'script' and simulates one of the parties in the file, listening for live packets and responding with the 'answers' that match from the pcap file.

The script:



#!/usr/bin/env python

##
## replay by c1de0x 
## ----------------------------------
## v0.1
##
## Scapy extension to allow pcap-captured network traffic to be used as
## a 'script' or 'template' for responses to live requests.
##


import scapy
from util.matches import get_match
from socket import gethostname,gethostbyname
import sys
import getopt


class TCPReplayer(scapy.AnsweringMachine):
   """
   Replays TCP traffic by matching incoming live packets to packets in a given
   list, and then transmitting the answering packets from the list on the wire.
   """
   filter="tcp"
   function_name="replay_tcp"
   send_function = staticmethod(scapy.sendp)

   def parse_options(self,list,resequence=0,port=80):
       self.list = list
       self.serverip = gethostbyname(gethostname())
       self.resequence = resequence
       self.port = port
       if self.resequence: self.seq = 1L
  
   def get_answers(self, packet):
       answers = []
       for other in self.list:
           if other.answers(packet):
              answers.append(other)
  
       return answers
  
   def fixup_response(self,resp,req):
       if resp.haslayer(scapy.Ether):
           resp[scapy.Ether].dst = req[scapy.Ether].src
           resp[scapy.Ether].src = req[scapy.Ether].dst
       if resp.haslayer(scapy.IP):
           resp[scapy.IP].src = self.serverip
           del resp[scapy.IP].chksum
       if resp.haslayer(scapy.TCP):
           resp[scapy.TCP].dport = req[scapy.TCP].sport
          
          
           if not isinstance(req[scapy.TCP].payload,scapy.NoPayload) \
               and not isinstance(req[scapy.TCP].payload,scapy.Padding): #have real payload
                   resp[scapy.TCP].ack = req[scapy.TCP].seq + len(req[scapy.TCP].payload)
           else:
               resp[scapy.TCP].ack = req[scapy.TCP].seq + 1L

           if self.resequence:
               resp[scapy.TCP].seq = self.seq
               if not isinstance(resp[scapy.TCP].payload,scapy.NoPayload) \
                   and not isinstance(resp[scapy.TCP].payload,scapy.Padding): #have real payload
                   #load bearing packets increment seq for following packets
                   self.seq += len(req[scapy.TCP].payload)
          
           del resp[scapy.TCP].chksum
       return resp
          
   def is_request(self, req):
       if not len(self.list):
           print "Playback list is empty - exiting."
           sys.exit(0)
       if not req.haslayer(scapy.TCP):
           return 0
       tcp = req.getlayer(scapy.TCP)
       if req.getlayer(scapy.IP).dst != self.serverip or tcp.dport != self.port:
           return 0
      
       if scapy.TCPflags2str(tcp.flags) == 'A': #packet is an ack, do nothing.
            if verbose: print "a-> ", repr(req)
            match = get_match(self.list,req)
            if match:
                self.list.remove(match)
            else:
                if verbose: print "Couldn't match ack"
            return 0

       self.currentmatch = get_match(self.list,req)
       if not self.currentmatch:
           if verbose: print "Couldn't find match for: ", repr(req)
           return 0
       self.list.remove(self.currentmatch)

       answers = self.get_answers(self.currentmatch)
       if not len(answers):
           if verbose: print "Couldn't find answer for: ", repr(req)
           return 0
      
       resp = []
       for answer in answers:
           self.list.remove(answer)
           answer = self.fixup_response(answer,req)
           resp.append(answer)

       if self.resequence: self.seq += 1L # Increment sequence every 'response set'
       self.currentanswer = resp
       return 1

   def print_reply(self,req,reply):
       if verbose:
           print "i-> ", repr(req)
           print "o-> ", reply
  
   def make_reply(self, req):
       if not self.currentmatch or not self.currentanswer:
           raise Error("ASSERT: should never get here!")
       resp = self.currentanswer
       return resp


def Usage():
       print """Usage: replay --recording=file.pcap [--port=80] [--resequence] [-v]
       --recording     The pcap file to replay.
       --port          The port to 'listen' on. If no port is specified, 80 is assumed.
       --resequence    If specified, new sequence numbers will be assigned to outgoing packets,
                       otherwise the original sequence values will be used.
       -v, --verbose   Turn on verbose logging.
       -h, --help      This help message.
       """
       sys.exit(1)

port = 80
verbose = recording = resequence = 0
try:
   opts=getopt.getopt(sys.argv[1:], "hv",["help","verbose","recording=","resequence","port="])
   for opt, parm in opts[0]:
       if opt in ["-h","--help"]:
           Usage()
       elif opt == "--recording":
           recording = parm
       elif opt == "--resequence":
           resequence = 1
       elif opt == "--port":
           port = int(parm)
       elif opt in ["-v","--verbose"]:
           verbose = 1
          
except getopt.GetoptError:
   Usage()
      
if not recording:
   print "--recording must be specified."
   Usage()

file = scapy.rdpcap(recording)

from scapy import replay_tcp
print "Replayer active, waiting for incoming traffic...\n"
replay_tcp(list=file,port=port,resequence=resequence)

It also makes use of the following utility script:


import scapy

def matches(packet,other):
   if not isinstance(other,packet.__class__):
       return 0

   if isinstance(packet,scapy.IP):
       #print packet.src, other.src
       if packet.src != other.src:
           return 0
   elif isinstance(packet,scapy.TCP):
       #print packet.dport, other.dport
       #print packet.flags, other.flags
       if packet.dport != other.dport or packet.flags != other.flags:
           return 0
   elif isinstance(packet,scapy.Raw) or isinstance(packet,scapy.Padding):
       #print repr(packet),repr(other)
       if packet.load != other.load:
           return 0

   if not isinstance(packet.payload,scapy.NoPayload) and \
           not isinstance(other.payload,scapy.NoPayload): # if both packets have got payloads, recurse
       return matches(packet.payload,other.payload)
   elif (isinstance(packet.payload,scapy.NoPayload) and isinstance(other.payload,scapy.Padding)) \
       or (isinstance(packet.payload,scapy.Padding) and isinstance(other.payload,scapy.NoPayload)):
           # special case, if one has padding and the other has no payload, still consider it a match.
           return 1
   elif packet.payload.__class__ != other.payload.__class__:
       print packet.payload.__class__ , other.payload.__class__
       return 0
   else:
       return 1

def get_match(list, packet):
   for other in list:
       #print "|||", repr(other)
       if matches(packet,other):
           return other
   return 0



#=============================== Tests =========================================
class AssertionError(Exception):
   pass

def test_MatchSelf():
   print "test_matchSelf... \t\t\t",
   packet = scapy.Ether()/scapy.IP(src="10.0.1.57",dst="10.0.1.60")/scapy.TCP(dport=80,sport=11111,flags='S')/scapy.Raw(load="test")
   if not matches(packet,packet):
       raise AssertionError('packet should match itself')
   print "pass"

def test_MatchEtherNoPayload():
   print "test_MatchEtherNoPayload... \t\t",
   packet = scapy.Ether()
   if not matches(packet,packet):
       raise AssertionError('packet should match')
   print "pass"

def test_MatchIPNoPayload():
   print "test_MatchIPNoPayload... \t\t",
   packet = scapy.Ether()/scapy.IP(src="10.0.1.57",dst="10.0.1.60")
   if not matches(packet,packet):
       raise AssertionError('packet should match')
   print "pass"

def test_MatchIPMissingSrc():
   print "test_MatchIPMissingSrc... \t\t",
   packet = scapy.Ether()/scapy.IP()
   if not matches(packet,packet):
       raise AssertionError('packet should match')
   print "pass"

def test_NoMatchIP():
   print "test_NoMatchIP... \t\t\t",
   packet = scapy.Ether()/scapy.IP(src="10.0.1.57")
   packet2 = scapy.Ether()/scapy.IP(src="10.0.1.67")
   if matches(packet,packet2):
       raise AssertionError('packet should not match')
   print "pass"

def test_NoMatchTCP():
   print "test_NoMatchTCP... \t\t\t",
   packet = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8118)
   packet2 = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8117)
   if matches(packet,packet2):
       raise AssertionError('packet should not match')
   print "pass"

def test_NoMatchTCP2():
   print "test_NoMatchTCP2... \t\t\t",
   packet = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8118)
   packet2 = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP()
   if matches(packet,packet2):
       raise AssertionError('packet should not match')
   print "pass"

def test_PartialMatchTCP():
   print "test_PartialMatchTCP... \t\t",
   packet = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8118, flags='SA')
   packet2 = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8118, flags='A')
   if matches(packet,packet2):
       raise AssertionError('packet should not match')
   print "pass"

def test_PartialMatchTCP2():
   print "test_PartialMatchTCP2... \t\t",
   packet = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8118, flags='SA')
   packet2 = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8118)
   if matches(packet,packet2):
       raise AssertionError('packet should not match')
   print "pass"

def test_NoMatchRaw():
   print "test_NoMatchRaw... \t\t\t",
   packet = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8118)/"Test"
   packet2 = scapy.Ether()/scapy.IP(src="10.0.1.57")/scapy.TCP(dport=8118)/"Test2"
   if matches(packet,packet2):
       raise AssertionError('packet should not match')
   print "pass"

def run_tests():
   test_MatchSelf()
   test_MatchEtherNoPayload()
   test_MatchIPNoPayload()
   test_MatchIPMissingSrc()
   test_NoMatchIP()
   test_NoMatchTCP()
   test_NoMatchTCP2()
   test_NoMatchRaw()
   test_PartialMatchTCP()
   test_PartialMatchTCP2()