compat_expanduser,
compat_shlex_split,
compat_str,
+ compat_struct_unpack,
compat_urllib_parse_unquote,
compat_urllib_parse_unquote_plus,
compat_urllib_parse_urlencode,
- struct_unpack,
)
self.assertTrue(isinstance(doc.find('foo/bar').text, compat_str))
def test_struct_unpack(self):
- self.assertEqual(struct_unpack('!B', b'\x00'), (0,))
+ self.assertEqual(compat_struct_unpack('!B', b'\x00'), (0,))
if __name__ == '__main__':
except TypeError:
# In Python 2.6 and 2.7.x < 2.7.7, struct requires a bytes argument
# See https://bugs.python.org/issue19099
- def struct_pack(spec, *args):
+ def compat_struct_pack(spec, *args):
if isinstance(spec, compat_str):
spec = spec.encode('ascii')
return struct.pack(spec, *args)
- def struct_unpack(spec, *args):
+ def compat_struct_unpack(spec, *args):
if isinstance(spec, compat_str):
spec = spec.encode('ascii')
return struct.unpack(spec, *args)
else:
- struct_pack = struct.pack
- struct_unpack = struct.unpack
+ compat_struct_pack = struct.pack
+ compat_struct_unpack = struct.unpack
__all__ = [
'compat_shlex_split',
'compat_socket_create_connection',
'compat_str',
+ 'compat_struct_pack',
+ 'compat_struct_unpack',
'compat_subprocess_get_DEVNULL',
'compat_tokenize_tokenize',
'compat_urllib_error',
'compat_xml_parse_error',
'compat_xpath',
'shlex_quote',
- 'struct_pack',
- 'struct_unpack',
'subprocess_check_output',
'workaround_optparse_bug9161',
]
compat_urlparse,
compat_urllib_error,
compat_urllib_parse_urlparse,
- struct_pack,
- struct_unpack,
+ compat_struct_pack,
+ compat_struct_unpack,
)
from ..utils import (
encodeFilename,
# Utility functions for reading numbers and strings
def read_unsigned_long_long(self):
- return struct_unpack('!Q', self.read(8))[0]
+ return compat_struct_unpack('!Q', self.read(8))[0]
def read_unsigned_int(self):
- return struct_unpack('!I', self.read(4))[0]
+ return compat_struct_unpack('!I', self.read(4))[0]
def read_unsigned_char(self):
- return struct_unpack('!B', self.read(1))[0]
+ return compat_struct_unpack('!B', self.read(1))[0]
def read_string(self):
res = b''
def write_unsigned_int(stream, val):
- stream.write(struct_pack('!I', val))
+ stream.write(compat_struct_pack('!I', val))
def write_unsigned_int_24(stream, val):
- stream.write(struct_pack('!I', val)[1:])
+ stream.write(compat_struct_pack('!I', val)[1:])
def write_flv_header(stream):
from .common import InfoExtractor
from ..compat import (
- struct_unpack,
+ compat_struct_unpack,
)
from ..utils import (
ExtractorError,
encrypted_data = base64.b64decode(png.encode('utf-8'))
text_index = encrypted_data.find(b'tEXt')
text_chunk = encrypted_data[text_index - 4:]
- length = struct_unpack('!I', text_chunk[:4])[0]
+ length = compat_struct_unpack('!I', text_chunk[:4])[0]
# Use bytearray to get integers when iterating in both python 2.x and 3.x
data = bytearray(text_chunk[8:8 + length])
data = [chr(b) for b in data if b != 0]
from .compat import (
compat_ord,
- struct_pack,
- struct_unpack,
+ compat_struct_pack,
+ compat_struct_unpack,
)
__author__ = 'Timo Schmid <coding@timoschmid.de>'
# if the client cannot resolve the destination host's domain name to find its
# IP address, it should set the first three bytes of DSTIP to NULL and the last
# byte to a non-zero value.
-SOCKS4_DEFAULT_DSTIP = struct_pack('!BBBB', 0, 0, 0, 0xFF)
+SOCKS4_DEFAULT_DSTIP = compat_struct_pack('!BBBB', 0, 0, 0, 0xFF)
SOCKS5_VERSION = 5
SOCKS5_USER_AUTH_VERSION = 0x01
def _recv_bytes(self, cnt):
data = self.recvall(cnt)
- return struct_unpack('!{0}B'.format(cnt), data)
+ return compat_struct_unpack('!{0}B'.format(cnt), data)
@staticmethod
def _len_and_data(data):
- return struct_pack('!B', len(data)) + data
+ return compat_struct_pack('!B', len(data)) + data
def _check_response_version(self, expected_version, got_version):
if got_version != expected_version:
ipaddr = self._resolve_address(destaddr, SOCKS4_DEFAULT_DSTIP, use_remote_dns=is_4a)
- packet = struct_pack('!BBH', SOCKS4_VERSION, Socks4Command.CMD_CONNECT, port) + ipaddr
+ packet = compat_struct_pack('!BBH', SOCKS4_VERSION, Socks4Command.CMD_CONNECT, port) + ipaddr
username = (self._proxy.username or '').encode('utf-8')
packet += username + b'\x00'
self.sendall(packet)
- version, resp_code, dstport, dsthost = struct_unpack('!BBHI', self.recvall(8))
+ version, resp_code, dstport, dsthost = compat_struct_unpack('!BBHI', self.recvall(8))
self._check_response_version(SOCKS4_REPLY_VERSION, version)
self._setup_socks4(address, is_4a=True)
def _socks5_auth(self):
- packet = struct_pack('!B', SOCKS5_VERSION)
+ packet = compat_struct_pack('!B', SOCKS5_VERSION)
auth_methods = [Socks5Auth.AUTH_NONE]
if self._proxy.username and self._proxy.password:
auth_methods.append(Socks5Auth.AUTH_USER_PASS)
- packet += struct_pack('!B', len(auth_methods))
- packet += struct_pack('!{0}B'.format(len(auth_methods)), *auth_methods)
+ packet += compat_struct_pack('!B', len(auth_methods))
+ packet += compat_struct_pack('!{0}B'.format(len(auth_methods)), *auth_methods)
self.sendall(packet)
if method == Socks5Auth.AUTH_USER_PASS:
username = self._proxy.username.encode('utf-8')
password = self._proxy.password.encode('utf-8')
- packet = struct_pack('!B', SOCKS5_USER_AUTH_VERSION)
+ packet = compat_struct_pack('!B', SOCKS5_USER_AUTH_VERSION)
packet += self._len_and_data(username) + self._len_and_data(password)
self.sendall(packet)
self._socks5_auth()
reserved = 0
- packet = struct_pack('!BBB', SOCKS5_VERSION, Socks5Command.CMD_CONNECT, reserved)
+ packet = compat_struct_pack('!BBB', SOCKS5_VERSION, Socks5Command.CMD_CONNECT, reserved)
if ipaddr is None:
destaddr = destaddr.encode('utf-8')
- packet += struct_pack('!B', Socks5AddressType.ATYP_DOMAINNAME)
+ packet += compat_struct_pack('!B', Socks5AddressType.ATYP_DOMAINNAME)
packet += self._len_and_data(destaddr)
else:
- packet += struct_pack('!B', Socks5AddressType.ATYP_IPV4) + ipaddr
- packet += struct_pack('!H', port)
+ packet += compat_struct_pack('!B', Socks5AddressType.ATYP_IPV4) + ipaddr
+ packet += compat_struct_pack('!H', port)
self.sendall(packet)
destaddr = self.recvall(alen)
elif atype == Socks5AddressType.ATYP_IPV6:
destaddr = self.recvall(16)
- destport = struct_unpack('!H', self.recvall(2))[0]
+ destport = compat_struct_unpack('!H', self.recvall(2))[0]
return (destaddr, destport)
from .compat import (
compat_str,
- struct_unpack,
+ compat_struct_unpack,
)
from .utils import (
ExtractorError,
file_contents[:1])
# Determine number of bits in framesize rectangle
- framesize_nbits = struct_unpack('!B', content[:1])[0] >> 3
+ framesize_nbits = compat_struct_unpack('!B', content[:1])[0] >> 3
framesize_len = (5 + 4 * framesize_nbits + 7) // 8
pos = framesize_len + 2 + 2
while pos < len(content):
- header16 = struct_unpack('<H', content[pos:pos + 2])[0]
+ header16 = compat_struct_unpack('<H', content[pos:pos + 2])[0]
pos += 2
tag_code = header16 >> 6
tag_len = header16 & 0x3f
if tag_len == 0x3f:
- tag_len = struct_unpack('<I', content[pos:pos + 4])[0]
+ tag_len = compat_struct_unpack('<I', content[pos:pos + 4])[0]
pos += 4
assert pos + tag_len <= len(content), \
('Tag %d ends at %d+%d - that\'s longer than the file (%d)'
for _ in range(5):
buf = reader.read(1)
assert len(buf) == 1
- b = struct_unpack('<B', buf)[0]
+ b = compat_struct_unpack('<B', buf)[0]
res = res | ((b & 0x7f) << shift)
if b & 0x80 == 0:
break
bs = reader.read(3)
assert len(bs) == 3
last_byte = b'\xff' if (ord(bs[2:3]) >= 0x80) else b'\x00'
- return struct_unpack('<i', bs + last_byte)[0]
+ return compat_struct_unpack('<i', bs + last_byte)[0]
def _read_string(reader):
def _read_byte(reader):
resb = _read_bytes(1, reader=reader)
- res = struct_unpack('<B', resb)[0]
+ res = compat_struct_unpack('<B', resb)[0]
return res
compat_parse_qs,
compat_socket_create_connection,
compat_str,
+ compat_struct_pack,
compat_urllib_error,
compat_urllib_parse,
compat_urllib_parse_urlencode,
compat_urlparse,
compat_xpath,
shlex_quote,
- struct_pack,
)
from .socks import (
def intlist_to_bytes(xs):
if not xs:
return b''
- return struct_pack('%dB' % len(xs), *xs)
+ return compat_struct_pack('%dB' % len(xs), *xs)
# Cross-platform file locking
projects / youtube-dl / commitdiff