| ## 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 |
| |
| """ |
| Global variables and functions for handling external data sets. |
| """ |
| |
| |
| import os |
| import re |
| import sys |
| import time |
| |
| |
| from scapy.dadict import DADict |
| from scapy.consts import DARWIN, FREEBSD, NETBSD, OPENBSD |
| from scapy.error import log_loading |
| from scapy.compat import * |
| |
| |
| ############ |
| ## Consts ## |
| ############ |
| |
| ETHER_ANY = b"\x00"*6 |
| ETHER_BROADCAST = b"\xff"*6 |
| |
| ETH_P_ALL = 3 |
| ETH_P_IP = 0x800 |
| ETH_P_ARP = 0x806 |
| ETH_P_IPV6 = 0x86dd |
| ETH_P_MACSEC = 0x88e5 |
| |
| # From net/if_arp.h |
| ARPHDR_ETHER = 1 |
| ARPHDR_METRICOM = 23 |
| ARPHDR_PPP = 512 |
| ARPHDR_LOOPBACK = 772 |
| ARPHDR_TUN = 65534 |
| |
| # From pcap/dlt.h |
| DLT_NULL = 0 |
| DLT_EN10MB = 1 |
| DLT_EN3MB = 2 |
| DLT_AX25 = 3 |
| DLT_PRONET = 4 |
| DLT_CHAOS = 5 |
| DLT_IEEE802 = 6 |
| DLT_ARCNET = 7 |
| DLT_SLIP = 8 |
| DLT_PPP = 9 |
| DLT_FDDI = 10 |
| if OPENBSD: |
| DLT_RAW = 14 |
| else: |
| DLT_RAW = 12 |
| DLT_RAW_ALT = 101 # At least in Argus |
| if FREEBSD or NETBSD: |
| DLT_SLIP_BSDOS = 13 |
| DLT_PPP_BSDOS = 14 |
| else: |
| DLT_SLIP_BSDOS = 15 |
| DLT_PPP_BSDOS = 16 |
| if FREEBSD: |
| DLT_PFSYNC = 121 |
| else: |
| DLT_PFSYNC = 18 |
| DLT_HHDLC = 121 |
| DLT_ATM_CLIP = 19 |
| DLT_PPP_SERIAL = 50 |
| DLT_PPP_ETHER = 51 |
| DLT_SYMANTEC_FIREWALL = 99 |
| DLT_C_HDLC = 104 |
| DLT_IEEE802_11 = 105 |
| if OPENBSD: |
| DLT_LOOP = 12 |
| DLT_ENC = 13 |
| else: |
| DLT_LOOP = 108 |
| DLT_ENC = 109 |
| DLT_LINUX_SLL = 113 |
| DLT_PFLOG = 117 |
| DLT_PRISM_HEADER = 119 |
| DLT_AIRONET_HEADER = 120 |
| DLT_IEEE802_11_RADIO = 127 |
| DLT_LINUX_IRDA = 144 |
| DLT_BLUETOOTH_HCI_H4 = 187 |
| DLT_PPI = 192 |
| DLT_CAN_SOCKETCAN = 227 |
| DLT_IPV4 = 228 |
| DLT_IPV6 = 229 |
| |
| # From net/ipv6.h on Linux (+ Additions) |
| IPV6_ADDR_UNICAST = 0x01 |
| IPV6_ADDR_MULTICAST = 0x02 |
| IPV6_ADDR_CAST_MASK = 0x0F |
| IPV6_ADDR_LOOPBACK = 0x10 |
| IPV6_ADDR_GLOBAL = 0x00 |
| IPV6_ADDR_LINKLOCAL = 0x20 |
| IPV6_ADDR_SITELOCAL = 0x40 # deprecated since Sept. 2004 by RFC 3879 |
| IPV6_ADDR_SCOPE_MASK = 0xF0 |
| #IPV6_ADDR_COMPATv4 = 0x80 # deprecated; i.e. ::/96 |
| #IPV6_ADDR_MAPPED = 0x1000 # i.e.; ::ffff:0.0.0.0/96 |
| IPV6_ADDR_6TO4 = 0x0100 # Added to have more specific info (should be 0x0101 ?) |
| IPV6_ADDR_UNSPECIFIED = 0x10000 |
| |
| |
| # On windows, epoch is 01/02/1970 at 00:00 |
| EPOCH = time.mktime((1970, 1, 2, 0, 0, 0, 3, 1, 0))-86400 |
| |
| MTU = 0xffff # a.k.a give me all you have |
| |
| WINDOWS=sys.platform.startswith("win") |
| |
| |
| # file parsing to get some values : |
| |
| def load_protocols(filename): |
| spaces = re.compile(b"[ \t]+|\n") |
| dct = DADict(_name=filename) |
| try: |
| for l in open(filename, "rb"): |
| try: |
| shrp = l.find(b"#") |
| if shrp >= 0: |
| l = l[:shrp] |
| l = l.strip() |
| if not l: |
| continue |
| lt = tuple(re.split(spaces, l)) |
| if len(lt) < 2 or not lt[0]: |
| continue |
| dct[lt[0]] = int(lt[1]) |
| except Exception as e: |
| log_loading.info("Couldn't parse file [%s]: line [%r] (%s)", filename, l, e) |
| except IOError: |
| log_loading.info("Can't open %s file", filename) |
| return dct |
| |
| def load_ethertypes(filename): |
| spaces = re.compile(b"[ \t]+|\n") |
| dct = DADict(_name=filename) |
| try: |
| f=open(filename, "rb") |
| for l in f: |
| try: |
| shrp = l.find(b"#") |
| if shrp >= 0: |
| l = l[:shrp] |
| l = l.strip() |
| if not l: |
| continue |
| lt = tuple(re.split(spaces, l)) |
| if len(lt) < 2 or not lt[0]: |
| continue |
| dct[lt[0]] = int(lt[1], 16) |
| except Exception as e: |
| log_loading.info("Couldn't parse file [%s]: line [%r] (%s)", filename, l, e) |
| f.close() |
| except IOError as msg: |
| pass |
| return dct |
| |
| def load_services(filename): |
| spaces = re.compile(b"[ \t]+|\n") |
| tdct=DADict(_name="%s-tcp"%filename) |
| udct=DADict(_name="%s-udp"%filename) |
| try: |
| f=open(filename, "rb") |
| for l in f: |
| try: |
| shrp = l.find(b"#") |
| if shrp >= 0: |
| l = l[:shrp] |
| l = l.strip() |
| if not l: |
| continue |
| lt = tuple(re.split(spaces, l)) |
| if len(lt) < 2 or not lt[0]: |
| continue |
| if lt[1].endswith(b"/tcp"): |
| tdct[lt[0]] = int(lt[1].split(b'/')[0]) |
| elif lt[1].endswith(b"/udp"): |
| udct[lt[0]] = int(lt[1].split(b'/')[0]) |
| except Exception as e: |
| log_loading.warning("Couldn't parse file [%s]: line [%r] (%s)", filename, l, e) |
| f.close() |
| except IOError: |
| log_loading.info("Can't open /etc/services file") |
| return tdct,udct |
| |
| |
| class ManufDA(DADict): |
| def fixname(self, val): |
| return plain_str(val) |
| def _get_manuf_couple(self, mac): |
| oui = ":".join(mac.split(":")[:3]).upper() |
| return self.__dict__.get(oui,(mac,mac)) |
| def _get_manuf(self, mac): |
| return self._get_manuf_couple(mac)[1] |
| def _get_short_manuf(self, mac): |
| return self._get_manuf_couple(mac)[0] |
| def _resolve_MAC(self, mac): |
| oui = ":".join(mac.split(":")[:3]).upper() |
| if oui in self: |
| return ":".join([self[oui][0]]+ mac.split(":")[3:]) |
| return mac |
| def __repr__(self): |
| return "\n".join("<%s %s, %s>" % (i[0], i[1][0], i[1][1]) for i in self.__dict__.items()) |
| |
| |
| |
| def load_manuf(filename): |
| manufdb=ManufDA(_name=filename) |
| with open(filename, "rb") as fdesc: |
| for l in fdesc: |
| try: |
| l = l.strip() |
| if not l or l.startswith(b"#"): |
| continue |
| oui,shrt=l.split()[:2] |
| i = l.find(b"#") |
| if i < 0: |
| lng=shrt |
| else: |
| lng = l[i+2:] |
| manufdb[oui] = plain_str(shrt), plain_str(lng) |
| except Exception: |
| log_loading.warning("Couldn't parse one line from [%s] [%r]", |
| filename, l, exc_info=True) |
| return manufdb |
| |
| |
| if WINDOWS: |
| ETHER_TYPES=load_ethertypes("ethertypes") |
| IP_PROTOS=load_protocols(os.environ["SystemRoot"]+"\system32\drivers\etc\protocol") |
| TCP_SERVICES,UDP_SERVICES=load_services(os.environ["SystemRoot"] + "\system32\drivers\etc\services") |
| # Default value, will be updated by arch.windows |
| try: |
| MANUFDB = load_manuf(os.environ["ProgramFiles"] + "\\wireshark\\manuf") |
| except IOError: |
| MANUFDB = None |
| else: |
| IP_PROTOS=load_protocols("/etc/protocols") |
| ETHER_TYPES=load_ethertypes("/etc/ethertypes") |
| TCP_SERVICES,UDP_SERVICES=load_services("/etc/services") |
| MANUFDB = None |
| for prefix in ['/usr', '/usr/local', '/opt', '/opt/wireshark']: |
| try: |
| MANUFDB = load_manuf(os.path.join(prefix, "share", "wireshark", |
| "manuf")) |
| if MANUFDB: |
| break |
| except IOError: |
| pass |
| if not MANUFDB: |
| log_loading.warning("Cannot read wireshark manuf database") |
| |
| |
| ##################### |
| ## knowledge bases ## |
| ##################### |
| |
| class KnowledgeBase: |
| def __init__(self, filename): |
| self.filename = filename |
| self.base = None |
| |
| def lazy_init(self): |
| self.base = "" |
| |
| def reload(self, filename = None): |
| if filename is not None: |
| self.filename = filename |
| oldbase = self.base |
| self.base = None |
| self.lazy_init() |
| if self.base is None: |
| self.base = oldbase |
| |
| def get_base(self): |
| if self.base is None: |
| self.lazy_init() |
| return self.base |
| |
| |
