| ## This file is part of Scapy |
| ## See http://www.secdev.org/projects/scapy for more informations |
| ## Copyright (C) Philippe Biondi <phil@secdev.org> |
| ## This program is published under a GPLv2 license |
| |
| """ |
| Answering machines. |
| """ |
| |
| ######################## |
| ## Answering machines ## |
| ######################## |
| |
| from __future__ import absolute_import |
| from __future__ import print_function |
| from scapy.sendrecv import send,sendp,sniff |
| from scapy.config import conf |
| from scapy.error import log_interactive |
| import scapy.modules.six as six |
| |
| class ReferenceAM(type): |
| def __new__(cls, name, bases, dct): |
| o = super(ReferenceAM, cls).__new__(cls, name, bases, dct) |
| if o.function_name: |
| globals()[o.function_name] = lambda o=o,*args,**kargs: o(*args,**kargs)() |
| return o |
| |
| |
| class AnsweringMachine(six.with_metaclass(ReferenceAM, object)): |
| function_name = "" |
| filter = None |
| sniff_options = { "store":0 } |
| sniff_options_list = [ "store", "iface", "count", "promisc", "filter", "type", "prn", "stop_filter" ] |
| send_options = { "verbose":0 } |
| send_options_list = ["iface", "inter", "loop", "verbose"] |
| send_function = staticmethod(send) |
| |
| |
| def __init__(self, **kargs): |
| self.mode = 0 |
| if self.filter: |
| kargs.setdefault("filter",self.filter) |
| kargs.setdefault("prn", self.reply) |
| self.optam1 = {} |
| self.optam2 = {} |
| self.optam0 = {} |
| doptsend,doptsniff = self.parse_all_options(1, kargs) |
| self.defoptsend = self.send_options.copy() |
| self.defoptsend.update(doptsend) |
| self.defoptsniff = self.sniff_options.copy() |
| self.defoptsniff.update(doptsniff) |
| self.optsend,self.optsniff = [{},{}] |
| |
| def __getattr__(self, attr): |
| for d in [self.optam2, self.optam1]: |
| if attr in d: |
| return d[attr] |
| raise AttributeError(attr) |
| |
| def __setattr__(self, attr, val): |
| mode = self.__dict__.get("mode",0) |
| if mode == 0: |
| self.__dict__[attr] = val |
| else: |
| [self.optam1, self.optam2][mode-1][attr] = val |
| |
| def parse_options(self): |
| pass |
| |
| def parse_all_options(self, mode, kargs): |
| sniffopt = {} |
| sendopt = {} |
| for k in list(kargs): # use list(): kargs is modified in the loop |
| if k in self.sniff_options_list: |
| sniffopt[k] = kargs[k] |
| if k in self.send_options_list: |
| sendopt[k] = kargs[k] |
| if k in self.sniff_options_list+self.send_options_list: |
| del kargs[k] |
| if mode != 2 or kargs: |
| if mode == 1: |
| self.optam0 = kargs |
| elif mode == 2 and kargs: |
| k = self.optam0.copy() |
| k.update(kargs) |
| self.parse_options(**k) |
| kargs = k |
| omode = self.__dict__.get("mode",0) |
| self.__dict__["mode"] = mode |
| self.parse_options(**kargs) |
| self.__dict__["mode"] = omode |
| return sendopt,sniffopt |
| |
| def is_request(self, req): |
| return 1 |
| |
| def make_reply(self, req): |
| return req |
| |
| def send_reply(self, reply): |
| self.send_function(reply, **self.optsend) |
| |
| def print_reply(self, req, reply): |
| print("%s ==> %s" % (req.summary(),reply.summary())) |
| |
| def reply(self, pkt): |
| if not self.is_request(pkt): |
| return |
| reply = self.make_reply(pkt) |
| self.send_reply(reply) |
| if conf.verb >= 0: |
| self.print_reply(pkt, reply) |
| |
| def run(self, *args, **kargs): |
| log_interactive.warning("run() method deprecated. The instance is now callable") |
| self(*args,**kargs) |
| |
| def __call__(self, *args, **kargs): |
| optsend,optsniff = self.parse_all_options(2,kargs) |
| self.optsend=self.defoptsend.copy() |
| self.optsend.update(optsend) |
| self.optsniff=self.defoptsniff.copy() |
| self.optsniff.update(optsniff) |
| |
| try: |
| self.sniff() |
| except KeyboardInterrupt: |
| print("Interrupted by user") |
| |
| def sniff(self): |
| sniff(**self.optsniff) |
| |
