## 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
|
|
"""
|
General utility functions.
|
"""
|
|
from __future__ import absolute_import
|
from __future__ import print_function
|
import os, sys, socket, types
|
import random, time
|
import gzip, zlib
|
import re, struct, array
|
import subprocess
|
import tempfile
|
|
import warnings
|
import scapy.modules.six as six
|
from scapy.modules.six.moves import range
|
warnings.filterwarnings("ignore","tempnam",RuntimeWarning, __name__)
|
|
from scapy.config import conf
|
from scapy.consts import DARWIN, WINDOWS
|
from scapy.data import MTU
|
from scapy.compat import *
|
from scapy.error import log_runtime, log_loading, log_interactive, Scapy_Exception, warning
|
from scapy.base_classes import BasePacketList
|
|
###########
|
## Tools ##
|
###########
|
|
def get_temp_file(keep=False, autoext=""):
|
"""Create a temporary file and return its name. When keep is False,
|
the file is deleted when scapy exits.
|
|
"""
|
fname = tempfile.NamedTemporaryFile(prefix="scapy", suffix=autoext,
|
delete=False).name
|
if not keep:
|
conf.temp_files.append(fname)
|
return fname
|
|
def sane_color(x):
|
r=""
|
for i in x:
|
j = orb(i)
|
if (j < 32) or (j >= 127):
|
r=r+conf.color_theme.not_printable(".")
|
else:
|
r=r+chr(j)
|
return r
|
|
def sane(x):
|
r=""
|
for i in x:
|
j = orb(i)
|
if (j < 32) or (j >= 127):
|
r=r+"."
|
else:
|
r=r+chr(j)
|
return r
|
|
@conf.commands.register
|
def restart():
|
"""Restarts scapy"""
|
if not conf.interactive or not os.path.isfile(sys.argv[0]):
|
raise OSError("Scapy was not started from console")
|
if WINDOWS:
|
os._exit(subprocess.call([sys.executable] + sys.argv))
|
os.execv(sys.executable, [sys.executable] + sys.argv)
|
|
def lhex(x):
|
if type(x) in six.integer_types:
|
return hex(x)
|
elif isinstance(x, tuple):
|
return "(%s)" % ", ".join(map(lhex, x))
|
elif isinstance(x, list):
|
return "[%s]" % ", ".join(map(lhex, x))
|
else:
|
return x
|
|
@conf.commands.register
|
def hexdump(x, dump=False):
|
""" Build a tcpdump like hexadecimal view
|
|
:param x: a Packet
|
:param dump: define if the result must be printed or returned in a variable
|
:returns: a String only when dump=True
|
"""
|
s = ""
|
x = raw(x)
|
l = len(x)
|
i = 0
|
while i < l:
|
s += "%04x " % i
|
for j in range(16):
|
if i+j < l:
|
s += "%02X" % orb(x[i+j])
|
else:
|
s += " "
|
if j%16 == 7:
|
s += ""
|
s += " "
|
s += sane_color(x[i:i+16])
|
i += 16
|
s += "\n"
|
# remove trailing \n
|
if s.endswith("\n"):
|
s = s[:-1]
|
if dump:
|
return s
|
else:
|
print(s)
|
|
|
@conf.commands.register
|
def linehexdump(x, onlyasc=0, onlyhex=0, dump=False):
|
""" Build an equivalent view of hexdump() on a single line
|
|
Note that setting both onlyasc and onlyhex to 1 results in a empty output
|
|
:param x: a Packet
|
:param onlyasc: 1 to display only the ascii view
|
:param onlyhex: 1 to display only the hexadecimal view
|
:param dump: print the view if False
|
:returns: a String only when dump=True
|
"""
|
s = ""
|
x = raw(x)
|
l = len(x)
|
if not onlyasc:
|
for i in range(l):
|
s += "%02X" % orb(x[i])
|
if not onlyhex: # separate asc & hex if both are displayed
|
s += " "
|
if not onlyhex:
|
s += sane_color(x)
|
if dump:
|
return s
|
else:
|
print(s)
|
|
@conf.commands.register
|
def chexdump(x, dump=False):
|
""" Build a per byte hexadecimal representation
|
|
Example:
|
>>> chexdump(IP())
|
0x45, 0x00, 0x00, 0x14, 0x00, 0x01, 0x00, 0x00, 0x40, 0x00, 0x7c, 0xe7, 0x7f, 0x00, 0x00, 0x01, 0x7f, 0x00, 0x00, 0x01
|
|
:param x: a Packet
|
:param dump: print the view if False
|
:returns: a String only if dump=True
|
"""
|
x = raw(x)
|
s = ", ".join("%#04x" % orb(x) for x in x)
|
if dump:
|
return s
|
else:
|
print(s)
|
|
@conf.commands.register
|
def hexstr(x, onlyasc=0, onlyhex=0):
|
s = []
|
if not onlyasc:
|
s.append(" ".join("%02x" % orb(b) for b in x))
|
if not onlyhex:
|
s.append(sane(x))
|
return " ".join(s)
|
|
def repr_hex(s):
|
""" Convert provided bitstring to a simple string of hex digits """
|
return "".join("%02x" % orb(x) for x in s)
|
|
@conf.commands.register
|
def hexdiff(x,y):
|
"""Show differences between 2 binary strings"""
|
x=raw(x)[::-1]
|
y=raw(y)[::-1]
|
SUBST=1
|
INSERT=1
|
d = {(-1, -1): (0, (-1, -1))}
|
for j in range(len(y)):
|
d[-1,j] = d[-1,j-1][0]+INSERT, (-1,j-1)
|
for i in range(len(x)):
|
d[i,-1] = d[i-1,-1][0]+INSERT, (i-1,-1)
|
|
for j in range(len(y)):
|
for i in range(len(x)):
|
d[i,j] = min( ( d[i-1,j-1][0]+SUBST*(x[i] != y[j]), (i-1,j-1) ),
|
( d[i-1,j][0]+INSERT, (i-1,j) ),
|
( d[i,j-1][0]+INSERT, (i,j-1) ) )
|
|
|
backtrackx = []
|
backtracky = []
|
i=len(x)-1
|
j=len(y)-1
|
while not (i == j == -1):
|
i2,j2 = d[i,j][1]
|
backtrackx.append(x[i2+1:i+1])
|
backtracky.append(y[j2+1:j+1])
|
i,j = i2,j2
|
|
|
|
x = y = i = 0
|
colorize = { 0: lambda x:x,
|
-1: conf.color_theme.left,
|
1: conf.color_theme.right }
|
|
dox=1
|
doy=0
|
l = len(backtrackx)
|
while i < l:
|
separate=0
|
linex = backtrackx[i:i+16]
|
liney = backtracky[i:i+16]
|
xx = sum(len(k) for k in linex)
|
yy = sum(len(k) for k in liney)
|
if dox and not xx:
|
dox = 0
|
doy = 1
|
if dox and linex == liney:
|
doy=1
|
|
if dox:
|
xd = y
|
j = 0
|
while not linex[j]:
|
j += 1
|
xd -= 1
|
print(colorize[doy-dox]("%04x" % xd), end=' ')
|
x += xx
|
line=linex
|
else:
|
print(" ", end=' ')
|
if doy:
|
yd = y
|
j = 0
|
while not liney[j]:
|
j += 1
|
yd -= 1
|
print(colorize[doy-dox]("%04x" % yd), end=' ')
|
y += yy
|
line=liney
|
else:
|
print(" ", end=' ')
|
|
print(" ", end=' ')
|
|
cl = ""
|
for j in range(16):
|
if i+j < l:
|
if line[j]:
|
col = colorize[(linex[j]!=liney[j])*(doy-dox)]
|
print(col("%02X" % orb(line[j])), end=' ')
|
if linex[j]==liney[j]:
|
cl += sane_color(line[j])
|
else:
|
cl += col(sane(line[j]))
|
else:
|
print(" ", end=' ')
|
cl += " "
|
else:
|
print(" ", end=' ')
|
if j == 7:
|
print("", end=' ')
|
|
|
print(" ",cl)
|
|
if doy or not yy:
|
doy=0
|
dox=1
|
i += 16
|
else:
|
if yy:
|
dox=0
|
doy=1
|
else:
|
i += 16
|
|
if struct.pack("H",1) == b"\x00\x01": # big endian
|
def checksum(pkt):
|
if len(pkt) % 2 == 1:
|
pkt += b"\0"
|
s = sum(array.array("H", pkt))
|
s = (s >> 16) + (s & 0xffff)
|
s += s >> 16
|
s = ~s
|
return s & 0xffff
|
else:
|
def checksum(pkt):
|
if len(pkt) % 2 == 1:
|
pkt += b"\0"
|
s = sum(array.array("H", pkt))
|
s = (s >> 16) + (s & 0xffff)
|
s += s >> 16
|
s = ~s
|
return (((s>>8)&0xff)|s<<8) & 0xffff
|
|
|
def _fletcher16(charbuf):
|
# This is based on the GPLed C implementation in Zebra <http://www.zebra.org/>
|
c0 = c1 = 0
|
for char in charbuf:
|
c0 += orb(char)
|
c1 += c0
|
|
c0 %= 255
|
c1 %= 255
|
return (c0,c1)
|
|
@conf.commands.register
|
def fletcher16_checksum(binbuf):
|
""" Calculates Fletcher-16 checksum of the given buffer.
|
|
Note:
|
If the buffer contains the two checkbytes derived from the Fletcher-16 checksum
|
the result of this function has to be 0. Otherwise the buffer has been corrupted.
|
"""
|
(c0,c1)= _fletcher16(binbuf)
|
return (c1 << 8) | c0
|
|
|
@conf.commands.register
|
def fletcher16_checkbytes(binbuf, offset):
|
""" Calculates the Fletcher-16 checkbytes returned as 2 byte binary-string.
|
|
Including the bytes into the buffer (at the position marked by offset) the
|
global Fletcher-16 checksum of the buffer will be 0. Thus it is easy to verify
|
the integrity of the buffer on the receiver side.
|
|
For details on the algorithm, see RFC 2328 chapter 12.1.7 and RFC 905 Annex B.
|
"""
|
|
# This is based on the GPLed C implementation in Zebra <http://www.zebra.org/>
|
if len(binbuf) < offset:
|
raise Exception("Packet too short for checkbytes %d" % len(binbuf))
|
|
binbuf = binbuf[:offset] + b"\x00\x00" + binbuf[offset + 2:]
|
(c0,c1)= _fletcher16(binbuf)
|
|
x = ((len(binbuf) - offset - 1) * c0 - c1) % 255
|
|
if (x <= 0):
|
x += 255
|
|
y = 510 - c0 - x
|
|
if (y > 255):
|
y -= 255
|
return chb(x) + chb(y)
|
|
|
def mac2str(mac):
|
return b"".join(chb(int(x, 16)) for x in mac.split(':'))
|
|
def str2mac(s):
|
if isinstance(s, str):
|
return ("%02x:"*6)[:-1] % tuple(map(ord, s))
|
return ("%02x:"*6)[:-1] % tuple(s)
|
|
def randstring(l):
|
"""
|
Returns a random string of length l (l >= 0)
|
"""
|
return b"".join(struct.pack('B', random.randint(0, 255)) for _ in range(l))
|
|
def zerofree_randstring(l):
|
"""
|
Returns a random string of length l (l >= 0) without zero in it.
|
"""
|
return b"".join(struct.pack('B', random.randint(1, 255)) for _ in range(l))
|
|
def strxor(s1, s2):
|
"""
|
Returns the binary XOR of the 2 provided strings s1 and s2. s1 and s2
|
must be of same length.
|
"""
|
return b"".join(map(lambda x,y:chb(orb(x)^orb(y)), s1, s2))
|
|
def strand(s1, s2):
|
"""
|
Returns the binary AND of the 2 provided strings s1 and s2. s1 and s2
|
must be of same length.
|
"""
|
return b"".join(map(lambda x,y:chb(orb(x)&orb(y)), s1, s2))
|
|
|
# Workaround bug 643005 : https://sourceforge.net/tracker/?func=detail&atid=105470&aid=643005&group_id=5470
|
try:
|
socket.inet_aton("255.255.255.255")
|
except socket.error:
|
def inet_aton(x):
|
if x == "255.255.255.255":
|
return b"\xff"*4
|
else:
|
return socket.inet_aton(x)
|
else:
|
inet_aton = socket.inet_aton
|
|
inet_ntoa = socket.inet_ntoa
|
from scapy.pton_ntop import *
|
|
|
def atol(x):
|
try:
|
ip = inet_aton(x)
|
except socket.error:
|
ip = inet_aton(socket.gethostbyname(x))
|
return struct.unpack("!I", ip)[0]
|
def ltoa(x):
|
return inet_ntoa(struct.pack("!I", x&0xffffffff))
|
|
def itom(x):
|
return (0xffffffff00000000>>x)&0xffffffff
|
|
class ContextManagerSubprocess(object):
|
"""
|
Context manager that eases checking for unknown command.
|
|
Example:
|
>>> with ContextManagerSubprocess("my custom message"):
|
>>> subprocess.Popen(["unknown_command"])
|
|
"""
|
def __init__(self, name, prog):
|
self.name = name
|
self.prog = prog
|
|
def __enter__(self):
|
pass
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
if isinstance(exc_value, (OSError, TypeError)):
|
msg = "%s: executing %r failed" % (self.name, self.prog) if self.prog else "Could not execute %s, is it installed ?" % self.name
|
if not conf.interactive:
|
raise OSError(msg)
|
else:
|
log_runtime.error(msg, exc_info=True)
|
return True # Suppress the exception
|
|
class ContextManagerCaptureOutput(object):
|
"""
|
Context manager that intercept the console's output.
|
|
Example:
|
>>> with ContextManagerCaptureOutput() as cmco:
|
... print("hey")
|
... assert cmco.get_output() == "hey"
|
"""
|
def __init__(self):
|
self.result_export_object = ""
|
try:
|
import mock
|
except:
|
raise ImportError("The mock module needs to be installed !")
|
def __enter__(self):
|
import mock
|
def write(s, decorator=self):
|
decorator.result_export_object += s
|
mock_stdout = mock.Mock()
|
mock_stdout.write = write
|
self.bck_stdout = sys.stdout
|
sys.stdout = mock_stdout
|
return self
|
def __exit__(self, *exc):
|
sys.stdout = self.bck_stdout
|
return False
|
def get_output(self, eval_bytes=False):
|
if self.result_export_object.startswith("b'") and eval_bytes:
|
return plain_str(eval(self.result_export_object))
|
return self.result_export_object
|
|
def do_graph(graph,prog=None,format=None,target=None,type=None,string=None,options=None):
|
"""do_graph(graph, prog=conf.prog.dot, format="svg",
|
target="| conf.prog.display", options=None, [string=1]):
|
string: if not None, simply return the graph string
|
graph: GraphViz graph description
|
format: output type (svg, ps, gif, jpg, etc.), passed to dot's "-T" option
|
target: filename or redirect. Defaults pipe to Imagemagick's display program
|
prog: which graphviz program to use
|
options: options to be passed to prog"""
|
|
if format is None:
|
if WINDOWS:
|
format = "png" # use common format to make sure a viewer is installed
|
else:
|
format = "svg"
|
if string:
|
return graph
|
if type is not None:
|
format=type
|
if prog is None:
|
prog = conf.prog.dot
|
start_viewer=False
|
if target is None:
|
if WINDOWS:
|
target = get_temp_file(autoext="."+format)
|
start_viewer = True
|
else:
|
with ContextManagerSubprocess("do_graph()", conf.prog.display):
|
target = subprocess.Popen([conf.prog.display],
|
stdin=subprocess.PIPE).stdin
|
if format is not None:
|
format = "-T%s" % format
|
if isinstance(target, str):
|
if target.startswith('|'):
|
target = subprocess.Popen(target[1:].lstrip(), shell=True,
|
stdin=subprocess.PIPE).stdin
|
elif target.startswith('>'):
|
target = open(target[1:].lstrip(), "wb")
|
else:
|
target = open(os.path.abspath(target), "wb")
|
proc = subprocess.Popen("\"%s\" %s %s" % (prog, options or "", format or ""),
|
shell=True, stdin=subprocess.PIPE, stdout=target)
|
proc.stdin.write(raw(graph))
|
try:
|
target.close()
|
except:
|
pass
|
if start_viewer:
|
# Workaround for file not found error: We wait until tempfile is written.
|
waiting_start = time.time()
|
while not os.path.exists(target.name):
|
time.sleep(0.1)
|
if time.time() - waiting_start > 3:
|
warning("Temporary file '%s' could not be written. Graphic will not be displayed.", tempfile)
|
break
|
else:
|
if conf.prog.display == conf.prog._default:
|
os.startfile(target.name)
|
else:
|
with ContextManagerSubprocess("do_graph()", conf.prog.display):
|
subprocess.Popen([conf.prog.display, target.name])
|
|
_TEX_TR = {
|
"{":"{\\tt\\char123}",
|
"}":"{\\tt\\char125}",
|
"\\":"{\\tt\\char92}",
|
"^":"\\^{}",
|
"$":"\\$",
|
"#":"\\#",
|
"~":"\\~",
|
"_":"\\_",
|
"&":"\\&",
|
"%":"\\%",
|
"|":"{\\tt\\char124}",
|
"~":"{\\tt\\char126}",
|
"<":"{\\tt\\char60}",
|
">":"{\\tt\\char62}",
|
}
|
|
def tex_escape(x):
|
s = ""
|
for c in x:
|
s += _TEX_TR.get(c,c)
|
return s
|
|
def colgen(*lstcol,**kargs):
|
"""Returns a generator that mixes provided quantities forever
|
trans: a function to convert the three arguments into a color. lambda x,y,z:(x,y,z) by default"""
|
if len(lstcol) < 2:
|
lstcol *= 2
|
trans = kargs.get("trans", lambda x,y,z: (x,y,z))
|
while True:
|
for i in range(len(lstcol)):
|
for j in range(len(lstcol)):
|
for k in range(len(lstcol)):
|
if i != j or j != k or k != i:
|
yield trans(lstcol[(i+j)%len(lstcol)],lstcol[(j+k)%len(lstcol)],lstcol[(k+i)%len(lstcol)])
|
|
def incremental_label(label="tag%05i", start=0):
|
while True:
|
yield label % start
|
start += 1
|
|
def binrepr(val):
|
return bin(val)[2:]
|
|
def long_converter(s):
|
return int(s.replace('\n', '').replace(' ', ''), 16)
|
|
#########################
|
#### Enum management ####
|
#########################
|
|
class EnumElement:
|
_value=None
|
def __init__(self, key, value):
|
self._key = key
|
self._value = value
|
def __repr__(self):
|
return "<%s %s[%r]>" % (self.__dict__.get("_name", self.__class__.__name__), self._key, self._value)
|
def __getattr__(self, attr):
|
return getattr(self._value, attr)
|
def __str__(self):
|
return self._key
|
def __bytes__(self):
|
return raw(self.__str__())
|
def __hash__(self):
|
return self._value
|
def __int__(self):
|
return int(self._value)
|
def __eq__(self, other):
|
return self._value == int(other)
|
def __neq__(self, other):
|
return not self.__eq__(other)
|
|
|
class Enum_metaclass(type):
|
element_class = EnumElement
|
def __new__(cls, name, bases, dct):
|
rdict={}
|
for k,v in six.iteritems(dct):
|
if isinstance(v, int):
|
v = cls.element_class(k,v)
|
dct[k] = v
|
rdict[v] = k
|
dct["__rdict__"] = rdict
|
return super(Enum_metaclass, cls).__new__(cls, name, bases, dct)
|
def __getitem__(self, attr):
|
return self.__rdict__[attr]
|
def __contains__(self, val):
|
return val in self.__rdict__
|
def get(self, attr, val=None):
|
return self.__rdict__.get(attr, val)
|
def __repr__(self):
|
return "<%s>" % self.__dict__.get("name", self.__name__)
|
|
|
|
###################
|
## Object saving ##
|
###################
|
|
|
def export_object(obj):
|
print(bytes_base64(gzip.zlib.compress(six.moves.cPickle.dumps(obj, 2), 9)))
|
|
def import_object(obj=None):
|
if obj is None:
|
obj = sys.stdin.read()
|
return six.moves.cPickle.loads(gzip.zlib.decompress(base64_bytes(obj.strip())))
|
|
|
def save_object(fname, obj):
|
"""Pickle a Python object"""
|
|
fd = gzip.open(fname, "wb")
|
six.moves.cPickle.dump(obj, fd)
|
fd.close()
|
|
def load_object(fname):
|
"""unpickle a Python object"""
|
return six.moves.cPickle.load(gzip.open(fname,"rb"))
|
|
@conf.commands.register
|
def corrupt_bytes(s, p=0.01, n=None):
|
"""Corrupt a given percentage or number of bytes from a string"""
|
s = array.array("B",raw(s))
|
l = len(s)
|
if n is None:
|
n = max(1,int(l*p))
|
for i in random.sample(range(l), n):
|
s[i] = (s[i]+random.randint(1,255))%256
|
return s.tostring()
|
|
@conf.commands.register
|
def corrupt_bits(s, p=0.01, n=None):
|
"""Flip a given percentage or number of bits from a string"""
|
s = array.array("B",raw(s))
|
l = len(s)*8
|
if n is None:
|
n = max(1,int(l*p))
|
for i in random.sample(range(l), n):
|
s[i // 8] ^= 1 << (i % 8)
|
return s.tostring()
|
|
|
|
|
#############################
|
## pcap capture file stuff ##
|
#############################
|
|
@conf.commands.register
|
def wrpcap(filename, pkt, *args, **kargs):
|
"""Write a list of packets to a pcap file
|
|
filename: the name of the file to write packets to, or an open,
|
writable file-like object. The file descriptor will be
|
closed at the end of the call, so do not use an object you
|
do not want to close (e.g., running wrpcap(sys.stdout, [])
|
in interactive mode will crash Scapy).
|
gz: set to 1 to save a gzipped capture
|
linktype: force linktype value
|
endianness: "<" or ">", force endianness
|
sync: do not bufferize writes to the capture file
|
|
"""
|
with PcapWriter(filename, *args, **kargs) as fdesc:
|
fdesc.write(pkt)
|
|
@conf.commands.register
|
def rdpcap(filename, count=-1):
|
"""Read a pcap or pcapng file and return a packet list
|
|
count: read only <count> packets
|
|
"""
|
with PcapReader(filename) as fdesc:
|
return fdesc.read_all(count=count)
|
|
|
class PcapReader_metaclass(type):
|
"""Metaclass for (Raw)Pcap(Ng)Readers"""
|
|
def __new__(cls, name, bases, dct):
|
"""The `alternative` class attribute is declared in the PcapNg
|
variant, and set here to the Pcap variant.
|
|
"""
|
newcls = super(PcapReader_metaclass, cls).__new__(cls, name, bases, dct)
|
if 'alternative' in dct:
|
dct['alternative'].alternative = newcls
|
return newcls
|
|
def __call__(cls, filename):
|
"""Creates a cls instance, use the `alternative` if that
|
fails.
|
|
"""
|
i = cls.__new__(cls, cls.__name__, cls.__bases__, cls.__dict__)
|
filename, fdesc, magic = cls.open(filename)
|
try:
|
i.__init__(filename, fdesc, magic)
|
except Scapy_Exception:
|
if "alternative" in cls.__dict__:
|
cls = cls.__dict__["alternative"]
|
i = cls.__new__(cls, cls.__name__, cls.__bases__, cls.__dict__)
|
try:
|
i.__init__(filename, fdesc, magic)
|
except Scapy_Exception:
|
raise
|
try:
|
i.f.seek(-4, 1)
|
except:
|
pass
|
raise Scapy_Exception("Not a supported capture file")
|
|
return i
|
|
@staticmethod
|
def open(filename):
|
"""Open (if necessary) filename, and read the magic."""
|
if isinstance(filename, six.string_types):
|
try:
|
fdesc = gzip.open(filename,"rb")
|
magic = fdesc.read(4)
|
except IOError:
|
fdesc = open(filename, "rb")
|
magic = fdesc.read(4)
|
else:
|
fdesc = filename
|
filename = (fdesc.name
|
if hasattr(fdesc, "name") else
|
"No name")
|
magic = fdesc.read(4)
|
return filename, fdesc, magic
|
|
|
class RawPcapReader(six.with_metaclass(PcapReader_metaclass)):
|
"""A stateful pcap reader. Each packet is returned as a string"""
|
def __init__(self, filename, fdesc, magic):
|
self.filename = filename
|
self.f = fdesc
|
if magic == b"\xa1\xb2\xc3\xd4": # big endian
|
self.endian = ">"
|
self.nano = False
|
elif magic == b"\xd4\xc3\xb2\xa1": # little endian
|
self.endian = "<"
|
self.nano = False
|
elif magic == b"\xa1\xb2\x3c\x4d": # big endian, nanosecond-precision
|
self.endian = ">"
|
self.nano = True
|
elif magic == b"\x4d\x3c\xb2\xa1": # little endian, nanosecond-precision
|
self.endian = "<"
|
self.nano = True
|
else:
|
raise Scapy_Exception(
|
"Not a pcap capture file (bad magic: %r)" % magic
|
)
|
hdr = self.f.read(20)
|
if len(hdr)<20:
|
raise Scapy_Exception("Invalid pcap file (too short)")
|
vermaj, vermin, tz, sig, snaplen, linktype = struct.unpack(
|
self.endian + "HHIIII", hdr
|
)
|
self.linktype = linktype
|
|
def __iter__(self):
|
return self
|
|
def next(self):
|
"""implement the iterator protocol on a set of packets in a pcap file"""
|
pkt = self.read_packet()
|
if pkt == None:
|
raise StopIteration
|
return pkt
|
__next__ = next
|
|
|
def read_packet(self, size=MTU):
|
"""return a single packet read from the file
|
|
returns None when no more packets are available
|
"""
|
hdr = self.f.read(16)
|
if len(hdr) < 16:
|
return None
|
sec,usec,caplen,wirelen = struct.unpack(self.endian+"IIII", hdr)
|
s = self.f.read(caplen)[:size]
|
return s,(sec,usec,wirelen) # caplen = len(s)
|
|
|
def dispatch(self, callback):
|
"""call the specified callback routine for each packet read
|
|
This is just a convenience function for the main loop
|
that allows for easy launching of packet processing in a
|
thread.
|
"""
|
for p in self:
|
callback(p)
|
|
def read_all(self,count=-1):
|
"""return a list of all packets in the pcap file
|
"""
|
res=[]
|
while count != 0:
|
count -= 1
|
p = self.read_packet()
|
if p is None:
|
break
|
res.append(p)
|
return res
|
|
def recv(self, size=MTU):
|
""" Emulate a socket
|
"""
|
return self.read_packet(size=size)[0]
|
|
def fileno(self):
|
return self.f.fileno()
|
|
def close(self):
|
return self.f.close()
|
|
def __enter__(self):
|
return self
|
|
def __exit__(self, exc_type, exc_value, tracback):
|
self.close()
|
|
|
class PcapReader(RawPcapReader):
|
def __init__(self, filename, fdesc, magic):
|
RawPcapReader.__init__(self, filename, fdesc, magic)
|
try:
|
self.LLcls = conf.l2types[self.linktype]
|
except KeyError:
|
warning("PcapReader: unknown LL type [%i]/[%#x]. Using Raw packets" % (self.linktype,self.linktype))
|
self.LLcls = conf.raw_layer
|
def read_packet(self, size=MTU):
|
rp = RawPcapReader.read_packet(self, size=size)
|
if rp is None:
|
return None
|
s,(sec,usec,wirelen) = rp
|
|
try:
|
p = self.LLcls(s)
|
except KeyboardInterrupt:
|
raise
|
except:
|
if conf.debug_dissector:
|
raise
|
p = conf.raw_layer(s)
|
p.time = sec + (0.000000001 if self.nano else 0.000001) * usec
|
return p
|
def read_all(self,count=-1):
|
res = RawPcapReader.read_all(self, count)
|
from scapy import plist
|
return plist.PacketList(res,name = os.path.basename(self.filename))
|
def recv(self, size=MTU):
|
return self.read_packet(size=size)
|
|
|
class RawPcapNgReader(RawPcapReader):
|
"""A stateful pcapng reader. Each packet is returned as a
|
string.
|
|
"""
|
|
alternative = RawPcapReader
|
|
def __init__(self, filename, fdesc, magic):
|
self.filename = filename
|
self.f = fdesc
|
# A list of (linktype, snaplen, tsresol); will be populated by IDBs.
|
self.interfaces = []
|
self.blocktypes = {
|
1: self.read_block_idb,
|
2: self.read_block_pkt,
|
3: self.read_block_spb,
|
6: self.read_block_epb,
|
}
|
if magic != b"\x0a\x0d\x0d\x0a": # PcapNg:
|
raise Scapy_Exception(
|
"Not a pcapng capture file (bad magic: %r)" % magic
|
)
|
# see https://github.com/pcapng/pcapng
|
blocklen, magic = self.f.read(4), self.f.read(4)
|
if magic == b"\x1a\x2b\x3c\x4d":
|
self.endian = ">"
|
elif magic == b"\x4d\x3c\x2b\x1a":
|
self.endian = "<"
|
else:
|
raise Scapy_Exception("Not a pcapng capture file (bad magic)")
|
try:
|
self.f.seek(0)
|
except:
|
pass
|
|
def read_packet(self, size=MTU):
|
"""Read blocks until it reaches either EOF or a packet, and
|
returns None or (packet, (linktype, sec, usec, wirelen)),
|
where packet is a string.
|
|
"""
|
while True:
|
try:
|
blocktype, blocklen = struct.unpack(self.endian + "2I",
|
self.f.read(8))
|
except struct.error:
|
return None
|
block = self.f.read(blocklen - 12)
|
if blocklen % 4:
|
pad = self.f.read(4 - (blocklen % 4))
|
warning("PcapNg: bad blocklen %d (MUST be a multiple of 4. "
|
"Ignored padding %r" % (blocklen, pad))
|
try:
|
if (blocklen,) != struct.unpack(self.endian + 'I',
|
self.f.read(4)):
|
warning("PcapNg: Invalid pcapng block (bad blocklen)")
|
except struct.error:
|
return None
|
res = self.blocktypes.get(blocktype,
|
lambda block, size: None)(block, size)
|
if res is not None:
|
return res
|
|
def read_block_idb(self, block, _):
|
"""Interface Description Block"""
|
options = block[16:]
|
tsresol = 1000000
|
while len(options) >= 4:
|
code, length = struct.unpack(self.endian + "HH", options[:4])
|
# PCAP Next Generation (pcapng) Capture File Format
|
# 4.2. - Interface Description Block
|
# http://xml2rfc.tools.ietf.org/cgi-bin/xml2rfc.cgi?url=https://raw.githubusercontent.com/pcapng/pcapng/master/draft-tuexen-opsawg-pcapng.xml&modeAsFormat=html/ascii&type=ascii#rfc.section.4.2
|
if code == 9 and length == 1 and len(options) >= 5:
|
tsresol = orb(options[4])
|
tsresol = (2 if tsresol & 128 else 10) ** (tsresol & 127)
|
if code == 0:
|
if length != 0:
|
warning("PcapNg: invalid option length %d for end-of-option" % length)
|
break
|
if length % 4:
|
length += (4 - (length % 4))
|
options = options[4 + length:]
|
self.interfaces.append(struct.unpack(self.endian + "HxxI", block[:8])
|
+ (tsresol,))
|
|
def read_block_epb(self, block, size):
|
"""Enhanced Packet Block"""
|
intid, tshigh, tslow, caplen, wirelen = struct.unpack(
|
self.endian + "5I",
|
block[:20],
|
)
|
return (block[20:20 + caplen][:size],
|
(self.interfaces[intid][0], self.interfaces[intid][2],
|
tshigh, tslow, wirelen))
|
|
def read_block_spb(self, block, size):
|
"""Simple Packet Block"""
|
# "it MUST be assumed that all the Simple Packet Blocks have
|
# been captured on the interface previously specified in the
|
# first Interface Description Block."
|
intid = 0
|
wirelen, = struct.unpack(self.endian + "I", block[:4])
|
caplen = min(wirelen, self.interfaces[intid][1])
|
return (block[4:4 + caplen][:size],
|
(self.interfaces[intid][0], self.interfaces[intid][2],
|
None, None, wirelen))
|
|
def read_block_pkt(self, block, size):
|
"""(Obsolete) Packet Block"""
|
intid, drops, tshigh, tslow, caplen, wirelen = struct.unpack(
|
self.endian + "HH4I",
|
block[:20],
|
)
|
return (block[20:20 + caplen][:size],
|
(self.interfaces[intid][0], self.interfaces[intid][2],
|
tshigh, tslow, wirelen))
|
|
|
class PcapNgReader(RawPcapNgReader):
|
|
alternative = PcapReader
|
|
def __init__(self, filename, fdesc, magic):
|
RawPcapNgReader.__init__(self, filename, fdesc, magic)
|
|
def read_packet(self, size=MTU):
|
rp = RawPcapNgReader.read_packet(self, size=size)
|
if rp is None:
|
return None
|
s, (linktype, tsresol, tshigh, tslow, wirelen) = rp
|
try:
|
p = conf.l2types[linktype](s)
|
except KeyboardInterrupt:
|
raise
|
except:
|
if conf.debug_dissector:
|
raise
|
p = conf.raw_layer(s)
|
if tshigh is not None:
|
p.time = float((tshigh << 32) + tslow) / tsresol
|
return p
|
def read_all(self,count=-1):
|
res = RawPcapNgReader.read_all(self, count)
|
from scapy import plist
|
return plist.PacketList(res, name=os.path.basename(self.filename))
|
def recv(self, size=MTU):
|
return self.read_packet()
|
|
|
class RawPcapWriter:
|
"""A stream PCAP writer with more control than wrpcap()"""
|
def __init__(self, filename, linktype=None, gz=False, endianness="",
|
append=False, sync=False, nano=False):
|
"""
|
filename: the name of the file to write packets to, or an open,
|
writable file-like object.
|
linktype: force linktype to a given value. If None, linktype is taken
|
from the first writer packet
|
gz: compress the capture on the fly
|
endianness: force an endianness (little:"<", big:">"). Default is native
|
append: append packets to the capture file instead of truncating it
|
sync: do not bufferize writes to the capture file
|
nano: use nanosecond-precision (requires libpcap >= 1.5.0)
|
|
"""
|
|
self.linktype = linktype
|
self.header_present = 0
|
self.append = append
|
self.gz = gz
|
self.endian = endianness
|
self.sync = sync
|
self.nano = nano
|
bufsz=4096
|
if sync:
|
bufsz = 0
|
|
if isinstance(filename, six.string_types):
|
self.filename = filename
|
self.f = [open,gzip.open][gz](filename,append and "ab" or "wb", gz and 9 or bufsz)
|
else:
|
self.f = filename
|
self.filename = (filename.name
|
if hasattr(filename, "name") else
|
"No name")
|
|
def fileno(self):
|
return self.f.fileno()
|
|
def _write_header(self, pkt):
|
self.header_present=1
|
|
if self.append:
|
# Even if prone to race conditions, this seems to be
|
# safest way to tell whether the header is already present
|
# because we have to handle compressed streams that
|
# are not as flexible as basic files
|
g = [open,gzip.open][self.gz](self.filename,"rb")
|
if g.read(16):
|
return
|
|
self.f.write(struct.pack(self.endian+"IHHIIII", 0xa1b23c4d if self.nano else 0xa1b2c3d4,
|
2, 4, 0, 0, MTU, self.linktype))
|
self.f.flush()
|
|
|
def write(self, pkt):
|
"""accepts either a single packet or a list of packets to be
|
written to the dumpfile
|
|
"""
|
if isinstance(pkt, str):
|
if not self.header_present:
|
self._write_header(pkt)
|
self._write_packet(pkt)
|
else:
|
pkt = pkt.__iter__()
|
if not self.header_present:
|
try:
|
p = next(pkt)
|
except StopIteration:
|
self._write_header(b"")
|
return
|
self._write_header(p)
|
self._write_packet(p)
|
for p in pkt:
|
self._write_packet(p)
|
|
def _write_packet(self, packet, sec=None, usec=None, caplen=None, wirelen=None):
|
"""writes a single packet to the pcap file
|
"""
|
if isinstance(packet, tuple):
|
for pkt in packet:
|
self._write_packet(pkt, sec=sec, usec=usec, caplen=caplen,
|
wirelen=wirelen)
|
return
|
if caplen is None:
|
caplen = len(packet)
|
if wirelen is None:
|
wirelen = caplen
|
if sec is None or usec is None:
|
t=time.time()
|
it = int(t)
|
if sec is None:
|
sec = it
|
if usec is None:
|
usec = int(round((t - it) * (1000000000 if self.nano else 1000000)))
|
self.f.write(struct.pack(self.endian+"IIII", sec, usec, caplen, wirelen))
|
self.f.write(packet)
|
if self.sync:
|
self.f.flush()
|
|
def flush(self):
|
return self.f.flush()
|
|
def close(self):
|
return self.f.close()
|
|
def __enter__(self):
|
return self
|
def __exit__(self, exc_type, exc_value, tracback):
|
self.flush()
|
self.close()
|
|
|
class PcapWriter(RawPcapWriter):
|
"""A stream PCAP writer with more control than wrpcap()"""
|
def _write_header(self, pkt):
|
if isinstance(pkt, tuple) and pkt:
|
pkt = pkt[0]
|
if self.linktype == None:
|
try:
|
self.linktype = conf.l2types[pkt.__class__]
|
except KeyError:
|
warning("PcapWriter: unknown LL type for %s. Using type 1 (Ethernet)", pkt.__class__.__name__)
|
self.linktype = 1
|
RawPcapWriter._write_header(self, pkt)
|
|
def _write_packet(self, packet):
|
if isinstance(packet, tuple):
|
for pkt in packet:
|
self._write_packet(pkt)
|
return
|
sec = int(packet.time)
|
usec = int(round((packet.time - sec) * (1000000000 if self.nano else 1000000)))
|
s = raw(packet)
|
caplen = len(s)
|
RawPcapWriter._write_packet(self, s, sec, usec, caplen, caplen)
|
|
|
re_extract_hexcap = re.compile("^((0x)?[0-9a-fA-F]{2,}[ :\t]{,3}|) *(([0-9a-fA-F]{2} {,2}){,16})")
|
|
@conf.commands.register
|
def import_hexcap():
|
p = ""
|
try:
|
while True:
|
l = input().strip()
|
try:
|
p += re_extract_hexcap.match(l).groups()[2]
|
except:
|
warning("Parsing error during hexcap")
|
continue
|
except EOFError:
|
pass
|
|
p = p.replace(" ","")
|
return p.decode("hex")
|
|
|
|
@conf.commands.register
|
def wireshark(pktlist):
|
"""Run wireshark on a list of packets"""
|
f = get_temp_file()
|
wrpcap(f, pktlist)
|
with ContextManagerSubprocess("wireshark()", conf.prog.wireshark):
|
subprocess.Popen([conf.prog.wireshark, "-r", f])
|
|
@conf.commands.register
|
def tcpdump(pktlist, dump=False, getfd=False, args=None,
|
prog=None, getproc=False, quiet=False):
|
"""Run tcpdump or tshark on a list of packets
|
|
pktlist: a Packet instance, a PacketList instance or a list of Packet
|
instances. Can also be a filename (as a string) or an open
|
file-like object that must be a file format readable by
|
tshark (Pcap, PcapNg, etc.)
|
|
dump: when set to True, returns a string instead of displaying it.
|
getfd: when set to True, returns a file-like object to read data
|
from tcpdump or tshark from.
|
getproc: when set to True, the subprocess.Popen object is returned
|
args: arguments (as a list) to pass to tshark (example for tshark:
|
args=["-T", "json"]). Defaults to ["-n"].
|
prog: program to use (defaults to tcpdump, will work with tshark)
|
quiet: when set to True, the process stderr is discarded
|
|
Examples:
|
|
>>> tcpdump([IP()/TCP(), IP()/UDP()])
|
reading from file -, link-type RAW (Raw IP)
|
16:46:00.474515 IP 127.0.0.1.20 > 127.0.0.1.80: Flags [S], seq 0, win 8192, length 0
|
16:46:00.475019 IP 127.0.0.1.53 > 127.0.0.1.53: [|domain]
|
|
>>> tcpdump([IP()/TCP(), IP()/UDP()], prog=conf.prog.tshark)
|
1 0.000000 127.0.0.1 -> 127.0.0.1 TCP 40 20->80 [SYN] Seq=0 Win=8192 Len=0
|
2 0.000459 127.0.0.1 -> 127.0.0.1 UDP 28 53->53 Len=0
|
|
To get a JSON representation of a tshark-parsed PacketList(), one can:
|
>>> import json, pprint
|
>>> json_data = json.load(tcpdump(IP(src="217.25.178.5", dst="45.33.32.156"),
|
... prog=conf.prog.tshark, args=["-T", "json"],
|
... getfd=True))
|
>>> pprint.pprint(json_data)
|
[{u'_index': u'packets-2016-12-23',
|
u'_score': None,
|
u'_source': {u'layers': {u'frame': {u'frame.cap_len': u'20',
|
u'frame.encap_type': u'7',
|
[...]
|
u'frame.time_relative': u'0.000000000'},
|
u'ip': {u'ip.addr': u'45.33.32.156',
|
u'ip.checksum': u'0x0000a20d',
|
[...]
|
u'ip.ttl': u'64',
|
u'ip.version': u'4'},
|
u'raw': u'Raw packet data'}},
|
u'_type': u'pcap_file'}]
|
>>> json_data[0]['_source']['layers']['ip']['ip.ttl']
|
u'64'
|
|
"""
|
getfd = getfd or getproc
|
if prog is None:
|
prog = [conf.prog.tcpdump]
|
elif isinstance(prog, six.string_types):
|
prog = [prog]
|
_prog_name = "windump()" if WINDOWS else "tcpdump()"
|
if pktlist is None:
|
with ContextManagerSubprocess(_prog_name, prog[0]):
|
proc = subprocess.Popen(
|
prog + (args if args is not None else []),
|
stdout=subprocess.PIPE if dump or getfd else None,
|
stderr=open(os.devnull) if quiet else None,
|
)
|
elif isinstance(pktlist, six.string_types):
|
with ContextManagerSubprocess(_prog_name, prog[0]):
|
proc = subprocess.Popen(
|
prog + ["-r", pktlist] + (args if args is not None else []),
|
stdout=subprocess.PIPE if dump or getfd else None,
|
stderr=open(os.devnull) if quiet else None,
|
)
|
elif DARWIN:
|
# Tcpdump cannot read from stdin, see
|
# <http://apple.stackexchange.com/questions/152682/>
|
tmpfile = tempfile.NamedTemporaryFile(delete=False)
|
try:
|
tmpfile.writelines(iter(lambda: pktlist.read(1048576), b""))
|
except AttributeError:
|
wrpcap(tmpfile, pktlist)
|
else:
|
tmpfile.close()
|
with ContextManagerSubprocess(_prog_name, prog[0]):
|
proc = subprocess.Popen(
|
prog + ["-r", tmpfile.name] + (args if args is not None else []),
|
stdout=subprocess.PIPE if dump or getfd else None,
|
stderr=open(os.devnull) if quiet else None,
|
)
|
conf.temp_files.append(tmpfile.name)
|
else:
|
with ContextManagerSubprocess(_prog_name, prog[0]):
|
proc = subprocess.Popen(
|
prog + ["-r", "-"] + (args if args is not None else []),
|
stdin=subprocess.PIPE,
|
stdout=subprocess.PIPE if dump or getfd else None,
|
stderr=open(os.devnull) if quiet else None,
|
)
|
try:
|
proc.stdin.writelines(iter(lambda: pktlist.read(1048576), b""))
|
except AttributeError:
|
wrpcap(proc.stdin, pktlist)
|
else:
|
proc.stdin.close()
|
if dump:
|
return b"".join(iter(lambda: proc.stdout.read(1048576), b""))
|
if getproc:
|
return proc
|
if getfd:
|
return proc.stdout
|
proc.wait()
|
|
@conf.commands.register
|
def hexedit(x):
|
x = str(x)
|
f = get_temp_file()
|
open(f,"wb").write(x)
|
with ContextManagerSubprocess("hexedit()", conf.prog.hexedit):
|
subprocess.call([conf.prog.hexedit, f])
|
x = open(f).read()
|
os.unlink(f)
|
return x
|
|
def get_terminal_width():
|
"""Get terminal width if in a window"""
|
if WINDOWS:
|
from ctypes import windll, create_string_buffer
|
# http://code.activestate.com/recipes/440694-determine-size-of-console-window-on-windows/
|
h = windll.kernel32.GetStdHandle(-12)
|
csbi = create_string_buffer(22)
|
res = windll.kernel32.GetConsoleScreenBufferInfo(h, csbi)
|
if res:
|
import struct
|
(bufx, bufy, curx, cury, wattr,
|
left, top, right, bottom, maxx, maxy) = struct.unpack("hhhhHhhhhhh", csbi.raw)
|
sizex = right - left + 1
|
#sizey = bottom - top + 1
|
return sizex
|
else:
|
return None
|
else:
|
sizex = 0
|
try:
|
import struct, fcntl, termios
|
s = struct.pack('HHHH', 0, 0, 0, 0)
|
x = fcntl.ioctl(1, termios.TIOCGWINSZ, s)
|
sizex = struct.unpack('HHHH', x)[1]
|
except IOError:
|
pass
|
if not sizex:
|
try:
|
sizex = int(os.environ['COLUMNS'])
|
except:
|
pass
|
if sizex:
|
return sizex
|
else:
|
return None
|
|
def pretty_list(rtlst, header, sortBy=0):
|
"""Pretty list to fit the terminal, and add header"""
|
_l_header = len(header[0])
|
_space = " "
|
# Sort correctly
|
rtlst.sort(key=lambda x: x[sortBy])
|
# Append tag
|
rtlst = header + rtlst
|
# Detect column's width
|
colwidth = [max([len(y) for y in x]) for x in zip(*rtlst)]
|
# Make text fit in box (if exist)
|
# TODO: find a better and more precise way of doing this. That's currently working but very complicated
|
width = get_terminal_width()
|
if width:
|
if sum(colwidth) > width:
|
# Needs to be cropped
|
_med = (width // _l_header) - (1 if WINDOWS else 0) # Windows has a fat window border
|
# Crop biggest until size is correct
|
for i in range(1, len(colwidth)): # Should use while, but this is safer
|
if (sum(colwidth)+6) <= width:
|
break
|
_max = max(colwidth)
|
colwidth = [_med if x == _max else x for x in colwidth]
|
def _crop(x, width):
|
_r = x[:width]
|
if _r != x:
|
_r = x[:width-3]
|
return _r + "..."
|
return _r
|
rtlst = [tuple([_crop(rtlst[j][i], colwidth[i]) for i in range(0, len(rtlst[j]))]) for j in range(0, len(rtlst))]
|
# Recalculate column's width
|
colwidth = [max([len(y) for y in x]) for x in zip(*rtlst)]
|
fmt = _space.join(["%%-%ds"%x for x in colwidth])
|
rt = "\n".join([fmt % x for x in rtlst])
|
return rt
|
|
def __make_table(yfmtfunc, fmtfunc, endline, data, fxyz, sortx=None, sorty=None, seplinefunc=None):
|
vx = {}
|
vy = {}
|
vz = {}
|
vxf = {}
|
vyf = {}
|
l = 0
|
for e in data:
|
xx, yy, zz = [str(s) for s in fxyz(e)]
|
l = max(len(yy),l)
|
vx[xx] = max(vx.get(xx,0), len(xx), len(zz))
|
vy[yy] = None
|
vz[(xx,yy)] = zz
|
|
vxk = list(vx)
|
vyk = list(vy)
|
if sortx:
|
vxk.sort(key=sortx)
|
else:
|
try:
|
vxk.sort(key=int)
|
except:
|
try:
|
vxk.sort(key=atol)
|
except:
|
vxk.sort()
|
if sorty:
|
vyk.sort(key=sorty)
|
else:
|
try:
|
vyk.sort(key=int)
|
except:
|
try:
|
vyk.sort(key=atol)
|
except:
|
vyk.sort()
|
|
|
if seplinefunc:
|
sepline = seplinefunc(l, [vx[x] for x in vxk])
|
print(sepline)
|
|
fmt = yfmtfunc(l)
|
print(fmt % "", end=' ')
|
for x in vxk:
|
vxf[x] = fmtfunc(vx[x])
|
print(vxf[x] % x, end=' ')
|
print(endline)
|
if seplinefunc:
|
print(sepline)
|
for y in vyk:
|
print(fmt % y, end=' ')
|
for x in vxk:
|
print(vxf[x] % vz.get((x,y), "-"), end=' ')
|
print(endline)
|
if seplinefunc:
|
print(sepline)
|
|
def make_table(*args, **kargs):
|
__make_table(lambda l:"%%-%is" % l, lambda l:"%%-%is" % l, "", *args, **kargs)
|
|
def make_lined_table(*args, **kargs):
|
__make_table(lambda l:"%%-%is |" % l, lambda l:"%%-%is |" % l, "",
|
seplinefunc=lambda a,x:"+".join('-'*(y+2) for y in [a-1]+x+[-2]),
|
*args, **kargs)
|
|
def make_tex_table(*args, **kargs):
|
__make_table(lambda l: "%s", lambda l: "& %s", "\\\\", seplinefunc=lambda a,x:"\\hline", *args, **kargs)
|
|
###############################################
|
### WHOIS CLIENT (not available on windows) ###
|
###############################################
|
|
def whois(ip_address):
|
"""Whois client for Python"""
|
whois_ip = str(ip_address)
|
try:
|
query = socket.gethostbyname(whois_ip)
|
except:
|
query = whois_ip
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
s.connect(("whois.ripe.net", 43))
|
s.send(query.encode("utf8") + b"\r\n")
|
answer = b""
|
while True:
|
d = s.recv(4096)
|
answer += d
|
if not d:
|
break
|
s.close()
|
ignore_tag = b"remarks:"
|
# ignore all lines starting with the ignore_tag
|
lines = [ line for line in answer.split(b"\n") if not line or (line and not line.startswith(ignore_tag))]
|
# remove empty lines at the bottom
|
for i in range(1, len(lines)):
|
if not lines[-i].strip():
|
del lines[-i]
|
else:
|
break
|
return b"\n".join(lines[3:])
|
