text: #!/usr/bin/env python3
  
  import argparse
  import threading
  
  from scapy.all import *
  from requests import get
  
  
  PACKET_COUNTER = 1
  EXTERNAL_IP = get('https://checkip.amazonaws.com/').text.rstrip("\n")
  LOCAL_IP = ''
  SOURCE_IP = '.'
  RESULT_FILE = open('Results.txt', 'w')
  LOCK_FILE = threading.Lock()
  LOCK_COUNTER = threading.Lock()
  
  
  def get_packet(pkt):
      global PACKET_COUNTER
      global LOCAL_IP
      global SOURCE_IP
      global RESULT_FILE
      global LOCK_FILE
      global LOCK_COUNTER
  
      if IP in pkt:
          ip_src = pkt[IP].src
          ip_dst = pkt[IP].dst
          if SOURCE_IP in ip_src:
              if TCP in pkt and (str(pkt[IP].flags) == 'DF' or pkt[IP].ttl > 220) and str(pkt[TCP].flags) == 'S':
                  tcp_dport = pkt[TCP].dport
                  if str(ip_dst) == EXTERNAL_IP or str(ip_dst) == LOCAL_IP:
                      string = "{0} {1} [TCP]\n".format(str(ip_src), str(tcp_dport))
                      with LOCK_FILE:
                          RESULT_FILE.write(string)
                          RESULT_FILE.flush()
                      with LOCK_COUNTER:
                          sys.stdout.write('\x1b[1A')
                          sys.stdout.write('\x1b[2K')
                          print('PorTest - SUCCESS - Count Packets: {0}.'.format(PACKET_COUNTER))
                          PACKET_COUNTER = PACKET_COUNTER + 1
  
              elif UDP in pkt:
                  tcp_dport = pkt[UDP].dport
                  if raw(pkt.getlayer(UDP).getlayer(1)) == b"PortTest":
                      if str(ip_dst) == EXTERNAL_IP or str(ip_dst) == LOCAL_IP:
                          string = "{0} {1} [UDP]\n".format(str(ip_src), str(tcp_dport))
                          with LOCK_FILE:
                              RESULT_FILE.write(string)
                              RESULT_FILE.flush()
                          with LOCK_COUNTER:
                              sys.stdout.write('\x1b[1A')
                              sys.stdout.write('\x1b[2K')
                              print('PorTest - SUCCESS - Count Packets: {0}.'.format(PACKET_COUNTER))
                              PACKET_COUNTER = PACKET_COUNTER + 1
  
  
  def sniff_function(timeout=None):
      conf.sniff_promisc=True
      print("PorTest - INFO - Start sniffing for {0} sec...".format(timeout))
      print(" ")
      sniff(filter="ip", prn=get_packet, timeout=timeout, store=False)
      print("PorTest - INFO - Sniffing Done!.")
  
  def parse_argumets():
      parser = argparse.ArgumentParser(description='Port Test Server listen to the world!')
      parser.add_argument('--timeout', '-t', default=86400, type=int, help='Timout in second. Default: 86400 (24H).')
      parser.add_argument('--source', '-s', default='.', type=str, help='Source ip if known. Default all sources.')
      parser.add_argument('--interface', '-i', default=None, type=str, help='Local IP to sniff packets.')
      return parser.parse_args()
  
  def main():
      global LOCAL_IP
      global SOURCE_IP
  
      args = parse_argumets()
      LOCAL_IP = args.interface
      SOURCE_IP = args.source
  
      sniff_function(timeout=args.timeout)
      RESULT_FILE.close()
  
  if __name__ == '__main__':
      main()
  
  
  
  
  