def set_alpn_protos(self, protos):
"""
Specify the clients ALPN protocol list.
These protocols are offered to the server during protocol negotiation.
:param protos: A list of the protocols to be offered to the server.
This list should be a Python list of bytestrings representing the
protocols to offer, e.g. ``[b'http/1.1', b'spdy/2']``.
"""
# Take the list of protocols and join them together, prefixing them
# with their lengths.
protostr = b''.join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
# Build a C string from the list. We don't need to save this off
# because OpenSSL immediately copies the data out.
input_str = _ffi.new("unsigned char[]", protostr)
input_str_len = _ffi.cast("unsigned", len(protostr))
_lib.SSL_CTX_set_alpn_protos(self._context, input_str, input_str_len)
python类int2byte()的实例源码
def set_alpn_protos(self, protos):
"""
Specify the client's ALPN protocol list.
These protocols are offered to the server during protocol negotiation.
:param protos: A list of the protocols to be offered to the server.
This list should be a Python list of bytestrings representing the
protocols to offer, e.g. ``[b'http/1.1', b'spdy/2']``.
"""
# Take the list of protocols and join them together, prefixing them
# with their lengths.
protostr = b''.join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
# Build a C string from the list. We don't need to save this off
# because OpenSSL immediately copies the data out.
input_str = _ffi.new("unsigned char[]", protostr)
input_str_len = _ffi.cast("unsigned", len(protostr))
_lib.SSL_set_alpn_protos(self._ssl, input_str, input_str_len)
def _truncate_digest(digest, order_bits):
digest_len = len(digest)
if 8 * digest_len > order_bits:
digest_len = (order_bits + 7) // 8
digest = digest[:digest_len]
if 8 * digest_len > order_bits:
rshift = 8 - (order_bits & 0x7)
assert 0 < rshift < 8
mask = 0xFF >> rshift << rshift
# Set the bottom rshift bits to 0
digest = digest[:-1] + six.int2byte(six.indexbytes(digest, -1) & mask)
return digest
def serialize(self):
# fixup
byte_length = (self.length + 7) // 8
bin_addr = self._to_bin(self.addr)
if (self.length % 8) == 0:
bin_addr = bin_addr[:byte_length]
else:
# clear trailing bits in the last octet.
# rfc doesn't require this.
mask = 0xff00 >> (self.length % 8)
last_byte = six.int2byte(
six.indexbytes(bin_addr, byte_length - 1) & mask)
bin_addr = bin_addr[:byte_length - 1] + last_byte
self.addr = self._from_bin(bin_addr)
buf = bytearray()
msg_pack_into(self._PACK_STR, buf, 0, self.length)
return buf + bytes(bin_addr)
def encode(self, b64=False, always_bytes=True):
"""Encode the packet for transmission."""
if self.binary and not b64:
encoded_packet = six.int2byte(self.packet_type)
else:
encoded_packet = six.text_type(self.packet_type)
if self.binary and b64:
encoded_packet = 'b' + encoded_packet
if self.binary:
if b64:
encoded_packet += base64.b64encode(self.data).decode('utf-8')
else:
encoded_packet += self.data
elif isinstance(self.data, six.string_types):
encoded_packet += self.data
elif isinstance(self.data, dict) or isinstance(self.data, list):
encoded_packet += self.json.dumps(self.data,
separators=(',', ':'))
elif self.data is not None:
encoded_packet += str(self.data)
if always_bytes and not isinstance(encoded_packet, six.binary_type):
encoded_packet = encoded_packet.encode('utf-8')
return encoded_packet
def encode(self, b64=False):
"""Encode the payload for transmission."""
encoded_payload = b''
for pkt in self.packets:
encoded_packet = pkt.encode(b64=b64)
packet_len = len(encoded_packet)
if b64:
encoded_payload += str(packet_len).encode('utf-8') + b':' + \
encoded_packet
else:
binary_len = b''
while packet_len != 0:
binary_len = six.int2byte(packet_len % 10) + binary_len
packet_len = int(packet_len / 10)
if not pkt.binary:
encoded_payload += b'\0'
else:
encoded_payload += b'\1'
encoded_payload += binary_len + b'\xff' + encoded_packet
return encoded_payload
def test_read_rain_collector_type(self, mock_read_config_setting):
mock_read_config_setting.return_value = six.int2byte(0b10101110)
collector_type = RainCollectorTypeSerial(self.communicator.read_rain_collector_type())
self.assertEqual(RainCollectorTypeSerial.millimeters_0_1, collector_type)
mock_read_config_setting.assert_called_once_with('2B', '01')
mock_read_config_setting.reset_mock()
mock_read_config_setting.return_value = six.int2byte(0b10011110)
collector_type = RainCollectorTypeSerial(self.communicator.read_rain_collector_type())
self.assertEqual(RainCollectorTypeSerial.millimeters_0_2, collector_type)
mock_read_config_setting.assert_called_once_with('2B', '01')
mock_read_config_setting.reset_mock()
mock_read_config_setting.return_value = six.int2byte(0b10001110)
collector_type = RainCollectorTypeSerial(self.communicator.read_rain_collector_type())
self.assertEqual(RainCollectorTypeSerial.inches_0_01, collector_type)
mock_read_config_setting.assert_called_once_with('2B', '01')
def set_alpn_protos(self, protos):
"""
Specify the clients ALPN protocol list.
These protocols are offered to the server during protocol negotiation.
:param protos: A list of the protocols to be offered to the server.
This list should be a Python list of bytestrings representing the
protocols to offer, e.g. ``[b'http/1.1', b'spdy/2']``.
"""
# Take the list of protocols and join them together, prefixing them
# with their lengths.
protostr = b''.join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
# Build a C string from the list. We don't need to save this off
# because OpenSSL immediately copies the data out.
input_str = _ffi.new("unsigned char[]", protostr)
input_str_len = _ffi.cast("unsigned", len(protostr))
_lib.SSL_CTX_set_alpn_protos(self._context, input_str, input_str_len)
def set_alpn_protos(self, protos):
"""
Specify the client's ALPN protocol list.
These protocols are offered to the server during protocol negotiation.
:param protos: A list of the protocols to be offered to the server.
This list should be a Python list of bytestrings representing the
protocols to offer, e.g. ``[b'http/1.1', b'spdy/2']``.
"""
# Take the list of protocols and join them together, prefixing them
# with their lengths.
protostr = b''.join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
# Build a C string from the list. We don't need to save this off
# because OpenSSL immediately copies the data out.
input_str = _ffi.new("unsigned char[]", protostr)
input_str_len = _ffi.cast("unsigned", len(protostr))
_lib.SSL_set_alpn_protos(self._ssl, input_str, input_str_len)
def _truncate_digest(digest, order_bits):
digest_len = len(digest)
if 8 * digest_len > order_bits:
digest_len = (order_bits + 7) // 8
digest = digest[:digest_len]
if 8 * digest_len > order_bits:
rshift = 8 - (order_bits & 0x7)
assert 0 < rshift < 8
mask = 0xFF >> rshift << rshift
# Set the bottom rshift bits to 0
digest = digest[:-1] + six.int2byte(six.indexbytes(digest, -1) & mask)
return digest
def _truncate_digest(digest, order_bits):
digest_len = len(digest)
if 8 * digest_len > order_bits:
digest_len = (order_bits + 7) // 8
digest = digest[:digest_len]
if 8 * digest_len > order_bits:
rshift = 8 - (order_bits & 0x7)
assert 0 < rshift < 8
mask = 0xFF >> rshift << rshift
# Set the bottom rshift bits to 0
digest = digest[:-1] + six.int2byte(six.indexbytes(digest, -1) & mask)
return digest
def decode(value):
"""Decodes bytes from a base65536 string."""
stream = io.BytesIO()
done = False
for ch in value:
code_point = ord(ch)
b1 = code_point & ((1 << 8) - 1)
try:
b2 = B2[code_point - b1]
except KeyError:
raise ValueError('Invalid base65536 code '
'point: %d' % code_point)
b = int2byte(b1) if b2 == -1 else int2byte(b1) + int2byte(b2)
if len(b) == 1:
if done:
raise ValueError('base65536 sequence '
'continued after final byte')
done = True
stream.write(b)
return stream.getvalue()
def _initialize_telnet(connection):
logger.info('Initializing telnet connection')
# Iac Do Linemode
connection.send(IAC + DO + LINEMODE)
# Suppress Go Ahead. (This seems important for Putty to do correct echoing.)
# This will allow bi-directional operation.
connection.send(IAC + WILL + SUPPRESS_GO_AHEAD)
# Iac sb
connection.send(IAC + SB + LINEMODE + MODE + int2byte(0) + IAC + SE)
# IAC Will Echo
connection.send(IAC + WILL + ECHO)
# Negotiate window size
connection.send(IAC + DO + NAWS)
def encode(self, b64=False, always_bytes=True):
"""Encode the packet for transmission."""
if self.binary and not b64:
encoded_packet = six.int2byte(self.packet_type)
else:
encoded_packet = six.text_type(self.packet_type)
if self.binary and b64:
encoded_packet = 'b' + encoded_packet
if self.binary:
if b64:
encoded_packet += base64.b64encode(self.data).decode('utf-8')
else:
encoded_packet += self.data
elif isinstance(self.data, six.string_types):
encoded_packet += self.data
elif isinstance(self.data, dict) or isinstance(self.data, list):
encoded_packet += self.json.dumps(self.data,
separators=(',', ':'))
elif self.data is not None:
encoded_packet += str(self.data)
if always_bytes and not isinstance(encoded_packet, six.binary_type):
encoded_packet = encoded_packet.encode('utf-8')
return encoded_packet
def encode(self, b64=False):
"""Encode the payload for transmission."""
encoded_payload = b''
for pkt in self.packets:
encoded_packet = pkt.encode(b64=b64)
packet_len = len(encoded_packet)
if b64:
encoded_payload += str(packet_len).encode('utf-8') + b':' + \
encoded_packet
else:
binary_len = b''
while packet_len != 0:
binary_len = six.int2byte(packet_len % 10) + binary_len
packet_len = int(packet_len / 10)
if not pkt.binary:
encoded_payload += b'\0'
else:
encoded_payload += b'\1'
encoded_payload += binary_len + b'\xff' + encoded_packet
return encoded_payload
def set_alpn_protos(self, protos):
"""
Specify the clients ALPN protocol list.
These protocols are offered to the server during protocol negotiation.
:param protos: A list of the protocols to be offered to the server.
This list should be a Python list of bytestrings representing the
protocols to offer, e.g. ``[b'http/1.1', b'spdy/2']``.
"""
# Take the list of protocols and join them together, prefixing them
# with their lengths.
protostr = b''.join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
# Build a C string from the list. We don't need to save this off
# because OpenSSL immediately copies the data out.
input_str = _ffi.new("unsigned char[]", protostr)
input_str_len = _ffi.cast("unsigned", len(protostr))
_lib.SSL_CTX_set_alpn_protos(self._context, input_str, input_str_len)
def set_alpn_protos(self, protos):
"""
Specify the client's ALPN protocol list.
These protocols are offered to the server during protocol negotiation.
:param protos: A list of the protocols to be offered to the server.
This list should be a Python list of bytestrings representing the
protocols to offer, e.g. ``[b'http/1.1', b'spdy/2']``.
"""
# Take the list of protocols and join them together, prefixing them
# with their lengths.
protostr = b''.join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
# Build a C string from the list. We don't need to save this off
# because OpenSSL immediately copies the data out.
input_str = _ffi.new("unsigned char[]", protostr)
input_str_len = _ffi.cast("unsigned", len(protostr))
_lib.SSL_set_alpn_protos(self._ssl, input_str, input_str_len)
def _truncate_digest(digest, order_bits):
digest_len = len(digest)
if 8 * digest_len > order_bits:
digest_len = (order_bits + 7) // 8
digest = digest[:digest_len]
if 8 * digest_len > order_bits:
rshift = 8 - (order_bits & 0x7)
assert 0 < rshift < 8
mask = 0xFF >> rshift << rshift
# Set the bottom rshift bits to 0
digest = digest[:-1] + six.int2byte(six.indexbytes(digest, -1) & mask)
return digest
def encode(self, b64=False, always_bytes=True):
"""Encode the packet for transmission."""
if self.binary and not b64:
encoded_packet = six.int2byte(self.packet_type)
else:
encoded_packet = six.text_type(self.packet_type)
if self.binary and b64:
encoded_packet = 'b' + encoded_packet
if self.binary:
if b64:
encoded_packet += base64.b64encode(self.data).decode('utf-8')
else:
encoded_packet += self.data
elif isinstance(self.data, six.string_types):
encoded_packet += self.data
elif isinstance(self.data, dict) or isinstance(self.data, list):
encoded_packet += self.json.dumps(self.data,
separators=(',', ':'))
elif self.data is not None:
encoded_packet += str(self.data)
if always_bytes and not isinstance(encoded_packet, six.binary_type):
encoded_packet = encoded_packet.encode('utf-8')
return encoded_packet
def encode(self, b64=False):
"""Encode the payload for transmission."""
encoded_payload = b''
for pkt in self.packets:
encoded_packet = pkt.encode(b64=b64)
packet_len = len(encoded_packet)
if b64:
encoded_payload += str(packet_len).encode('utf-8') + b':' + \
encoded_packet
else:
binary_len = b''
while packet_len != 0:
binary_len = six.int2byte(packet_len % 10) + binary_len
packet_len = int(packet_len / 10)
encoded_payload += b'\0' + binary_len + b'\xff' + \
encoded_packet
return encoded_payload
