ripple20-treck-scan.py

ripple20-treck-scan.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
#author：swing
from scapy.all import *

ICMP_MS_SYNC_REQ_TYPE = 0xa5
ICMP_MS_SYNC_RSP_TYPE = 0xa6


def keep_icmp_handler(func):
    def wrapper(*args, **kwargs):
        backup = ICMP.answers
        backup_err = ICMPerror.answers
        ICMP.answers = _icmp_answers
        ICMPerror.answers = _icmperror_answers
        res = func(*args, **kwargs)
        ICMP.answers = backup
        ICMPerror.answers = backup_err
        return res
    return wrapper

class Tester():
    name = "ICMP_MS_SYNC"

    def __init__(self ,timeout):
        self.timeout = timeout

    @keep_icmp_handler
    def run(self,address):
    # must use supperuser run 
        p = IP(dst=address)/ICMP(type=ICMP_MS_SYNC_REQ_TYPE)
        ans ,unans = sr(p, timeout=self.timeout)

        if not ans:
            print("NO RESPONSE")
        for req, resp in ans:
            if ICMP in resp and resp[ICMP].type == ICMP_MS_SYNC_RSP_TYPE:
                print("Yes,it's treck tcp")

def _icmp_answers(self, other):
    if not isinstance(other,ICMP):
        return 0
    if (self[ICMP].type == ICMP_MS_SYNC_RSP_TYPE and other[ICMP].type == ICMP_MS_SYNC_REQ_TYPE): # allow also destination unreachable + invalid protocol
        return 1
    return 0

def _icmperror_answers(self, other):
    if not isinstance(other, ICMP):
        return 0
    if bytes(self)[0] == 0xa5: # our special code
        return 1
    if not ((self.type == other.type) and
            (self.code == other.code)):
        return 0
    if self.code in [0, 8, 13, 14, 17, 18]:
        if (self.id == other.id and
                self.seq == other.seq):
            return 1
        else:
            return 0
    else:
        return 1

if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser(description = "check treck tcp")
    parser.add_argument('host',action='store',help='target ip address')
    parser.add_argument('-t', '--timeout', type=int, default=0.2, help="packet sniffing timeout (for the response)")
    args = parser.parse_args()
    target_host = args.host
    
    Tester(args.timeout).run(target_host)