Rooba
/
scapy
mirror of https://github.com/secdev/scapy.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Bytes conversion fixes, NTP fixes

This commit is contained in:
gpotter2 2017-09-15 23:22:13 +02:00
parent 6e0d5ba2d2
commit 5e76be46bd
13 changed files with 66 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
scapy/arch/pcapdnet.py
View File

@ -105,7 +105,7 @@ if conf.use_winpcapy:
if a.contents.addr.contents.sa_family == socket.AF_INET:
ap = a.contents.addr
val = cast(ap, POINTER(sockaddr_in))
ret = b"".join(chr(x) for x in val.contents.sin_addr[:4])
ret = b"".join(chb(x) for x in val.contents.sin_addr[:4])
a = a.contents.next
break
p = p.contents.next

3
scapy/asn1/asn1.py
View File

@ -16,6 +16,7 @@ from scapy.config import conf
from scapy.error import Scapy_Exception, warning
from scapy.volatile import RandField, RandIP
from scapy.utils import Enum_metaclass, EnumElement, binrepr
from scapy.compat import plain_str
import scapy.modules.six as six
from scapy.modules.six.moves import range
@ -333,7 +334,7 @@ class ASN1_NULL(ASN1_Object):
class ASN1_OID(ASN1_Object):
tag = ASN1_Class_UNIVERSAL.OID
def __init__(self, val):
val = conf.mib._oid(val)
val = conf.mib._oid(plain_str(val))
ASN1_Object.__init__(self, val)
self.oidname = conf.mib._oidname(val)
def __repr__(self):

21
scapy/asn1/ber.py
View File

@ -71,7 +71,7 @@ def BER_len_enc(l, size=0):
raise BER_Exception("BER_len_enc: Length too long (%i) to be encoded [%r]" % (len(s),s))
return chb(len(s)|0x80)+s
def BER_len_dec(s):
l = ord(s[0])
l = orb(s[0])
if not l & 0x80:
return l,s[1:]
l &= 0x7f
@ -80,7 +80,7 @@ def BER_len_dec(s):
ll = 0
for c in s[1:l+1]:
ll <<= 8
ll |= ord(c)
ll |= orb(c)
return ll,s[l+1:]
def BER_num_enc(l, size=1):
@ -97,7 +97,7 @@ def BER_num_dec(s, cls_id=0):
raise BER_Decoding_Error("BER_num_dec: got empty string", remaining=s)
x = cls_id
for i, c in enumerate(s):
c = ord(c)
c = orb(c)
x <<= 7
x |= c&0x7f
if not c&0x80:
@ -123,7 +123,7 @@ def BER_id_dec(s):
# encoded in scapy's tag in order to reuse it for packet building.
# Note that tags thus may have to be hard-coded with their extended
# information, e.g. a SEQUENCE from asn1.py has a direct tag 0x20|16.
x = ord(s[0])
x = orb(s[0])
if x & 0x1f != 0x1f:
# low-tag-number
return x,s[1:]
@ -137,7 +137,7 @@ def BER_id_enc(n):
else:
# high-tag-number
s = BER_num_enc(n)
tag = ord(s[0]) # first byte, as an int
tag = orb(s[0]) # first byte, as an int
tag &= 0x07 # reset every bit from 8 to 4
tag <<= 5 # move back the info bits on top
tag |= 0x1f # pad with 1s every bit from 5 to 1
@ -294,11 +294,11 @@ class BERcodec_INTEGER(BERcodec_Object):
l,s,t = cls.check_type_check_len(s)
x = 0
if s:
if ord(s[0])&0x80: # negative int
if orb(s[0])&0x80: # negative int
x = -1
for c in s:
x <<= 8
x |= ord(c)
x |= orb(c)
return cls.asn1_object(x),t
class BERcodec_BOOLEAN(BERcodec_INTEGER):
@ -311,10 +311,10 @@ class BERcodec_BIT_STRING(BERcodec_Object):
# /!\ the unused_bits information is lost after this decoding
l,s,t = cls.check_type_check_len(s)
if len(s) > 0:
unused_bits = ord(s[0])
unused_bits = orb(s[0])
if safe and unused_bits > 7:
raise BER_Decoding_Error("BERcodec_BIT_STRING: too many unused_bits advertised", remaining=s)
s = "".join(binrepr(ord(x)).zfill(8) for x in s[1:])
s = "".join(binrepr(orb(x)).zfill(8) for x in s[1:])
if unused_bits > 0:
s = s[:-unused_bits]
return cls.tag.asn1_object(s),t
@ -323,12 +323,13 @@ class BERcodec_BIT_STRING(BERcodec_Object):
@classmethod
def enc(cls,s):
# /!\ this is DER encoding (bit strings are only zero-bit padded)
s = raw(s)
if len(s) % 8 == 0:
unused_bits = 0
else:
unused_bits = 8 - len(s)%8
s += b"0"*unused_bits
s = b"".join(chb(int(b"".join(x),2)) for x in zip(*[iter(s)]*8))
s = b"".join(chb(int(b"".join(chb(y) for y in x),2)) for x in zip(*[iter(s)]*8))
s = chb(unused_bits) + s
return chb(hash(cls.tag))+BER_len_enc(len(s))+s

2
scapy/compat.py
View File

@ -106,7 +106,7 @@ else:
else:
if hasattr(x, "__int__") and not isinstance(x, int):
return bytes(chr(int(x)), encoding="utf8")
return bytes(chr(x), encoding="utf8")
return bytes([x])
def bytes_codec(x, codec, force_str=False):
"""Hexify a str or a bytes object"""

6
scapy/layers/eap.py
View File

@ -19,7 +19,7 @@ PacketListField, ConditionalField, PadField
from scapy.packet import Packet, bind_layers
from scapy.layers.l2 import SourceMACField, Ether, CookedLinux, GRE, SNAP
from scapy.config import conf
from scapy.compat import orb
#
# EAPOL
@ -231,9 +231,9 @@ class EAP(Packet):
@classmethod
def dispatch_hook(cls, _pkt=None, *args, **kargs):
if _pkt:
c = ord(_pkt[0])
c = orb(_pkt[0])
if c in [1, 2] and len(_pkt) >= 5:
t = ord(_pkt[4])
t = orb(_pkt[4])
return cls.registered_methods.get(t, cls)
return cls

15
scapy/layers/inet6.py
View File

@ -2175,18 +2175,15 @@ def names2dnsrepr(x):
if isinstance(x, str):
if x and x[-1] == '\x00': # stupid heuristic
return x.encode("utf8")
x = [x.encode("utf8")]
elif type(x) is bytes:
if x and x[-1] == 0:
return x
x = [x]
res = []
for n in x:
termin = b"\x00"
if n.count(b'.') == 0: # single-component gets one more
termin += b'\x00'
n = b"".join(chb(len(y)) + y for y in n.split(b'.')) + termin
res.append(n)
termin = "\x00"
if n.count('.') == 0: # single-component gets one more
termin += '\x00'
n = "".join(chr(len(y)) + y for y in n.split('.')) + termin
res.append(n.encode("utf8"))
return b"".join(res)

14
scapy/layers/ntp.py
View File

@ -199,7 +199,7 @@ def _ntp_dispatcher(payload):
else:
length = len(payload)
if length >= _NTP_PACKET_MIN_SIZE:
first_byte = struct.unpack("!B", raw(payload[0]))[0]
first_byte = orb(payload[0])
# Extract NTP mode
mode_mask = 0x07
@ -307,7 +307,7 @@ class NTPAuthenticator(Packet):
]
def extract_padding(self, s):
return "", s
return b"", s
class NTPExtension(Packet):
@ -660,7 +660,7 @@ class NTPStatusPacket(Packet):
fields_desc = [ShortField("status", 0)]
def extract_padding(self, s):
return "", s
return b"", s
class NTPSystemStatusPacket(Packet):
@ -678,7 +678,7 @@ class NTPSystemStatusPacket(Packet):
]
def extract_padding(self, s):
return "", s
return b"", s
class NTPPeerStatusPacket(Packet):
@ -699,7 +699,7 @@ class NTPPeerStatusPacket(Packet):
]
def extract_padding(self, s):
return "", s
return b"", s
class NTPClockStatusPacket(Packet):
@ -714,7 +714,7 @@ class NTPClockStatusPacket(Packet):
]
def extract_padding(self, s):
return "", s
return b"", s
class NTPErrorStatusPacket(Packet):
@ -729,7 +729,7 @@ class NTPErrorStatusPacket(Packet):
]
def extract_padding(self, s):
return "", s
return b"", s
class NTPControlStatusField(PacketField):

1
scapy/layers/tls/cert.py
View File

@ -513,6 +513,7 @@ class PrivKeyECDSA(PrivKey):
@crypto_validator
def import_from_asn1pkt(self, privkey):
print(privkey)
self.key = serialization.load_der_private_key(raw(privkey), None,
backend=default_backend())
self.pubkey = self.key.public_key()

32
scapy/layers/x509.py
View File

@ -703,9 +703,9 @@ class ASN1F_X509_SubjectPublicKeyInfo(ASN1F_SEQUENCE):
def m2i(self, pkt, x):
c,s = ASN1F_SEQUENCE.m2i(self, pkt, x)
keytype = pkt.fields["signatureAlgorithm"].algorithm.oidname
if b"rsa" in keytype.lower():
if "rsa" in keytype.lower():
return ASN1F_X509_SubjectPublicKeyInfoRSA().m2i(pkt, x)
elif keytype == b"ecPublicKey":
elif keytype == "ecPublicKey":
return ASN1F_X509_SubjectPublicKeyInfoECDSA().m2i(pkt, x)
else:
raise Exception("could not parse subjectPublicKeyInfo")
@ -717,10 +717,10 @@ class ASN1F_X509_SubjectPublicKeyInfo(ASN1F_SEQUENCE):
ktype = pkt.fields['signatureAlgorithm'].algorithm.oidname
else:
ktype = pkt.default_fields["signatureAlgorithm"].algorithm.oidname
if b"rsa" in ktype.lower():
if "rsa" in ktype.lower():
pkt.default_fields["subjectPublicKey"] = RSAPublicKey()
return ASN1F_X509_SubjectPublicKeyInfoRSA().build(pkt)
elif ktype == b"ecPublicKey":
elif ktype == "ecPublicKey":
pkt.default_fields["subjectPublicKey"] = ECDSAPublicKey()
return ASN1F_X509_SubjectPublicKeyInfoECDSA().build(pkt)
else:
@ -924,9 +924,9 @@ class ASN1F_X509_Cert(ASN1F_SEQUENCE):
def m2i(self, pkt, x):
c,s = ASN1F_SEQUENCE.m2i(self, pkt, x)
sigtype = pkt.fields["signatureAlgorithm"].algorithm.oidname
if b"rsa" in sigtype.lower():
if "rsa" in sigtype.lower():
return c,s
elif b"ecdsa" in sigtype.lower():
elif "ecdsa" in sigtype.lower():
return ASN1F_X509_CertECDSA().m2i(pkt, x)
else:
raise Exception("could not parse certificate")
@ -938,9 +938,9 @@ class ASN1F_X509_Cert(ASN1F_SEQUENCE):
sigtype = pkt.fields['signatureAlgorithm'].algorithm.oidname
else:
sigtype = pkt.default_fields["signatureAlgorithm"].algorithm.oidname
if b"rsa" in sigtype.lower():
if "rsa" in sigtype.lower():
return ASN1F_SEQUENCE.build(self, pkt)
elif b"ecdsa" in sigtype.lower():
elif "ecdsa" in sigtype.lower():
pkt.default_fields["signatureValue"] = ECDSASignature()
return ASN1F_X509_CertECDSA().build(pkt)
else:
@ -1032,9 +1032,9 @@ class ASN1F_X509_CRL(ASN1F_SEQUENCE):
def m2i(self, pkt, x):
c,s = ASN1F_SEQUENCE.m2i(self, pkt, x)
sigtype = pkt.fields["signatureAlgorithm"].algorithm.oidname
if b"rsa" in sigtype.lower():
if "rsa" in sigtype.lower():
return c,s
elif b"ecdsa" in sigtype.lower():
elif "ecdsa" in sigtype.lower():
return ASN1F_X509_CRLECDSA().m2i(pkt, x)
else:
raise Exception("could not parse certificate")
@ -1046,9 +1046,9 @@ class ASN1F_X509_CRL(ASN1F_SEQUENCE):
sigtype = pkt.fields['signatureAlgorithm'].algorithm.oidname
else:
sigtype = pkt.default_fields["signatureAlgorithm"].algorithm.oidname
if b"rsa" in sigtype.lower():
if "rsa" in sigtype.lower():
return ASN1F_SEQUENCE.build(self, pkt)
elif b"ecdsa" in sigtype.lower():
elif "ecdsa" in sigtype.lower():
pkt.default_fields["signatureValue"] = ECDSASignature()
return ASN1F_X509_CRLECDSA().build(pkt)
else:
@ -1181,9 +1181,9 @@ class ASN1F_OCSP_BasicResponse(ASN1F_SEQUENCE):
def m2i(self, pkt, x):
c,s = ASN1F_SEQUENCE.m2i(self, pkt, x)
sigtype = pkt.fields["signatureAlgorithm"].algorithm.oidname
if b"rsa" in sigtype.lower():
if "rsa" in sigtype.lower():
return c,s
elif b"ecdsa" in sigtype.lower():
elif "ecdsa" in sigtype.lower():
return ASN1F_OCSP_BasicResponseECDSA().m2i(pkt, x)
else:
raise Exception("could not parse OCSP basic response")
@ -1195,9 +1195,9 @@ class ASN1F_OCSP_BasicResponse(ASN1F_SEQUENCE):
sigtype = pkt.fields['signatureAlgorithm'].algorithm.oidname
else:
sigtype = pkt.default_fields["signatureAlgorithm"].algorithm.oidname
if b"rsa" in sigtype.lower():
if "rsa" in sigtype.lower():
return ASN1F_SEQUENCE.build(self, pkt)
elif b"ecdsa" in sigtype.lower():
elif "ecdsa" in sigtype.lower():
pkt.default_fields["signatureValue"] = ECDSASignature()
return ASN1F_OCSP_BasicResponseECDSA().build(pkt)
else:

20
scapy/utils.py
View File

@ -46,7 +46,7 @@ def sane_color(x):
if (j < 32) or (j >= 127):
r=r+conf.color_theme.not_printable(".")
else:
r=r+chb(i)
r=r+i
return r
def sane(x):
@ -56,7 +56,7 @@ def sane(x):
if (j < 32) or (j >= 127):
r=r+"."
else:
r=r+chb(i)
r=r+i
return r
def lhex(x):
@ -89,7 +89,7 @@ def hexdump(x, dump=False):
s += "%04x " % i
for j in range(16):
if i+j < l:
s += "%02X" % ord(x[i+j])
s += "%02X" % orb(x[i+j])
else:
s += " "
if j%16 == 7:
@ -124,7 +124,7 @@ def linehexdump(x, onlyasc=0, onlyhex=0, dump=False):
l = len(x)
if not onlyasc:
for i in range(l):
s += "%02X" % ord(x[i])
s += "%02X" % orb(x[i])
if not onlyhex: # separate asc & hex if both are displayed
s += " "
if not onlyhex:
@ -147,7 +147,7 @@ def chexdump(x, dump=False):
:returns: a String only if dump=True
"""
x = raw(x)
s = ", ".join("%#04x" % ord(x) for x in x)
s = ", ".join("%#04x" % orb(x) for x in x)
if dump:
return s
else:
@ -157,14 +157,14 @@ def chexdump(x, dump=False):
def hexstr(x, onlyasc=0, onlyhex=0):
s = []
if not onlyasc:
s.append(" ".join("%02x" % ord(b) for b in x))
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" % ord(x) for x in s)
return "".join("%02x" % orb(x) for x in s)
@conf.commands.register
def hexdiff(x,y):
@ -248,7 +248,7 @@ def hexdiff(x,y):
if i+j < l:
if line[j]:
col = colorize[(linex[j]!=liney[j])*(doy-dox)]
print(col("%02X" % ord(line[j])), end=' ')
print(col("%02X" % orb(line[j])), end=' ')
if linex[j]==liney[j]:
cl += sane_color(line[j])
else:
@ -299,7 +299,7 @@ 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 += ord(char)
c0 += orb(char)
c1 += c0
c0 %= 255
@ -1380,7 +1380,7 @@ def pretty_routes(rtlst, header, sortBy=0):
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 apply(zip, rtlst)]
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

12
scapy/utils6.py
View File

@ -167,11 +167,11 @@ def in6_getAddrType(addr):
addrType = 0
# _Assignable_ Global Unicast Address space
# is defined in RFC 3513 as those in 2000::/3
if ((struct.unpack("B", raw(naddr[0]))[0] & 0xE0) == 0x20):
if ((orb(naddr[0]) & 0xE0) == 0x20):
addrType = (IPV6_ADDR_UNICAST | IPV6_ADDR_GLOBAL)
if naddr[:2] == b' \x02': # Mark 6to4 @
addrType |= IPV6_ADDR_6TO4
elif ord(naddr[0]) == 0xff: # multicast
elif orb(naddr[0]) == 0xff: # multicast
addrScope = paddr[3]
if addrScope == '2':
addrType = (IPV6_ADDR_LINKLOCAL | IPV6_ADDR_MULTICAST)
@ -179,7 +179,7 @@ def in6_getAddrType(addr):
addrType = (IPV6_ADDR_GLOBAL | IPV6_ADDR_MULTICAST)
else:
addrType = (IPV6_ADDR_GLOBAL | IPV6_ADDR_MULTICAST)
elif ((ord(naddr[0]) == 0xfe) and ((int(paddr[2], 16) & 0xC) == 0x8)):
elif ((orb(naddr[0]) == 0xfe) and ((int(paddr[2], 16) & 0xC) == 0x8)):
addrType = (IPV6_ADDR_UNICAST | IPV6_ADDR_LINKLOCAL)
elif paddr == "::1":
addrType = IPV6_ADDR_LOOPBACK
@ -384,10 +384,8 @@ def in6_getRandomizedIfaceId(ifaceid, previous=None):
a "printable" format as depicted below.
ex:
>>> in6_getRandomizedIfaceId('20b:93ff:feeb:2d3')
('4c61:76ff:f46a:a5f3', 'd006:d540:db11:b092')
>>> in6_getRandomizedIfaceId('20b:93ff:feeb:2d3',
previous='d006:d540:db11:b092')
('fe97:46fe:9871:bd38', 'eeed:d79c:2e3f:62e')
@ -397,13 +395,13 @@ def in6_getRandomizedIfaceId(ifaceid, previous=None):
if previous is None:
d = b"".join(chb(x) for x in range(256))
for _ in range(8):
s += random.choice(d)
s += chb(random.choice(d))
previous = s
s = inet_pton(socket.AF_INET6, "::"+ifaceid)[8:] + previous
import hashlib
s = hashlib.md5(s).digest()
s1,s2 = s[:8],s[8:]
s1 = raw(ord(s1[0]) | 0x04) + s1[1:]
s1 = chb(orb(s1[0]) | 0x04) + s1[1:]
s1 = inet_ntop(socket.AF_INET6, b"\xff"*8 + s1)[20:]
s2 = inet_ntop(socket.AF_INET6, b"\xff"*8 + s2)[20:]
return (s1, s2)

1
test/pipetool.uts
View File

@ -263,6 +263,7 @@ os.unlink("t2.pcap")
= Test InjectSink and Inject3Sink
~ needs_root
import mock
import mock

4
test/regression.uts
View File

@ -180,10 +180,10 @@ os.remove("scapySession1.dat")
tmpfile = get_temp_file(autoext=".ut")
if WINDOWS:
assert("scapy" in tmpfile and tmpfile.startswith('C:\\Users\\appveyor\\AppData\\Local\\Temp'))
assert("scapy" in tmpfile and tmpfile.lower().startswith('c:\\users\\appveyor\\appdata\\local\\temp'))
else:
import platform
IS_PYPY = platform.python_implementation() == "PyPy"
IS_PYPY = platform.python_implementation().lower() == "pypy"
assert("scapy" in tmpfile and (IS_PYPY == True or "/tmp/" in tmpfile))
assert(conf.temp_files[0].endswith(".ut"))