Skip to content

Commit

Permalink
add support for aes256-gcm and aes128-gcm
Browse files Browse the repository at this point in the history
  • Loading branch information
@cmason3 @bitprophet
cmason3 authored and bitprophet committed Sep 14, 2024
1 parent d2e7c0c commit 39871d2
Show file tree
Hide file tree
Showing 2 changed files with 161 additions and 21 deletions.
60 changes: 54 additions & 6 deletions paramiko/packet.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,12 @@ def __init__(self, socket):
self.__etm_out = False
self.__etm_in = False

# aead cipher use
self.__aead_out = False
self.__aead_in = False
self.__iv_out = None
self.__iv_in = None

# lock around outbound writes (packet computation)
self.__write_lock = threading.RLock()

Expand Down Expand Up @@ -152,6 +158,8 @@ def set_outbound_cipher(
mac_key,
sdctr=False,
etm=False,
aead=False,
iv_out=None,
):
"""
Switch outbound data cipher.
Expand All @@ -166,6 +174,8 @@ def set_outbound_cipher(
self.__sent_bytes = 0
self.__sent_packets = 0
self.__etm_out = etm
self.__aead_out = aead
self.__iv_out = iv_out
# wait until the reset happens in both directions before clearing
# rekey flag
self.__init_count |= 1
Expand All @@ -181,6 +191,8 @@ def set_inbound_cipher(
mac_size,
mac_key,
etm=False,
aead=False,
iv_in=None,
):
"""
Switch inbound data cipher.
Expand All @@ -196,6 +208,8 @@ def set_inbound_cipher(
self.__received_bytes_overflow = 0
self.__received_packets_overflow = 0
self.__etm_in = etm
self.__aead_in = aead
self.__iv_in = iv_in
# wait until the reset happens in both directions before clearing
# rekey flag
self.__init_count |= 2
Expand Down Expand Up @@ -386,6 +400,20 @@ def readline(self, timeout):
buf = buf[:-1]
return u(buf)

def _inc_iv_counter(self, iv):
# refer https://www.rfc-editor.org/rfc/rfc5647.html#section-7.1
iv_counter_b = iv[4:]
iv_counter = int.from_bytes(iv_counter_b, "big")
inc_iv_counter = iv_counter + 1
inc_iv_counter_b = inc_iv_counter.to_bytes(8, "big")
new_iv = iv[0:4] + inc_iv_counter_b
self._log(
DEBUG,
"old-iv_count[%s], new-iv_count[%s]"
% (iv_counter, inc_iv_counter),
)
return new_iv

def send_message(self, data):
"""
Write a block of data using the current cipher, as an SSH block.
Expand Down Expand Up @@ -415,12 +443,18 @@ def send_message(self, data):
out = packet[0:4] + self.__block_engine_out.update(
packet[4:]
)
elif self.__aead_out:
# packet length is used to associated_data
out = packet[0:4] + self.__block_engine_out.encrypt(
self.__iv_out, packet[4:], packet[0:4]
)
self.__iv_out = self._inc_iv_counter(self.__iv_out)
else:
out = self.__block_engine_out.update(packet)
else:
out = packet
# + mac
if self.__block_engine_out is not None:
# + mac, aead no need hmac
if self.__block_engine_out is not None and not self.__aead_out:
packed = struct.pack(">I", self.__sequence_number_out)
payload = packed + (out if self.__etm_out else packet)
out += compute_hmac(
Expand Down Expand Up @@ -460,7 +494,9 @@ def read_message(self):
:raises: `.SSHException` -- if the packet is mangled
:raises: `.NeedRekeyException` -- if the transport should rekey
"""
self._log(DEBUG, "read message from sock")
header = self.read_all(self.__block_size_in, check_rekey=True)
self._log(DEBUG, "raw data length[%s]" % len(header))
if self.__etm_in:
packet_size = struct.unpack(">I", header[:4])[0]
remaining = packet_size - self.__block_size_in + 4
Expand All @@ -477,14 +513,26 @@ def read_message(self):
raise SSHException("Mismatched MAC")
header = packet

if self.__block_engine_in is not None:
if self.__aead_in:
packet_size = struct.unpack(">I", header[:4])[0]
aad = header[:4]
remaining = (
packet_size - self.__block_size_in + 4 + self.__mac_size_in
)
packet = header[4:] + self.read_all(remaining, check_rekey=False)
self._log(DEBUG, "len(aad)=%s, aad->%s" % (len(aad), aad.hex()))
header = self.__block_engine_in.decrypt(self.__iv_in, packet, aad)

self.__iv_in = self._inc_iv_counter(self.__iv_in)

if self.__block_engine_in is not None and not self.__aead_in:
header = self.__block_engine_in.update(header)
if self.__dump_packets:
self._log(DEBUG, util.format_binary(header, "IN: "))

# When ETM is in play, we've already read the packet size & decrypted
# everything, so just set the packet back to the header we obtained.
if self.__etm_in:
if self.__etm_in or self.__aead_in:
packet = header
# Otherwise, use the older non-ETM logic
else:
Expand All @@ -508,7 +556,7 @@ def read_message(self):
if self.__dump_packets:
self._log(DEBUG, util.format_binary(packet, "IN: "))

if self.__mac_size_in > 0 and not self.__etm_in:
if self.__mac_size_in > 0 and not self.__etm_in and not self.__aead_in:
mac = post_packet[: self.__mac_size_in]
mac_payload = (
struct.pack(">II", self.__sequence_number_in, packet_size)
Expand Down Expand Up @@ -631,7 +679,7 @@ def _build_packet(self, payload):
bsize = self.__block_size_out
# do not include payload length in computations for padding in EtM mode
# (payload length won't be encrypted)
addlen = 4 if self.__etm_out else 8
addlen = 4 if self.__etm_out or self.__aead_out else 8
padding = 3 + bsize - ((len(payload) + addlen) % bsize)
packet = struct.pack(">IB", len(payload) + padding + 1, padding)
packet += payload
Expand Down
122 changes: 107 additions & 15 deletions paramiko/transport.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
from hashlib import md5, sha1, sha256, sha512

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import algorithms, Cipher, modes
from cryptography.hazmat.primitives.ciphers import algorithms, Cipher, modes, aead

import paramiko
from paramiko import util
Expand Down Expand Up @@ -179,6 +179,8 @@ class Transport(threading.Thread, ClosingContextManager):
"aes192-cbc",
"aes256-cbc",
"3des-cbc",
"[email protected]",
"[email protected]",
)
_preferred_macs = (
"hmac-sha2-256",
Expand Down Expand Up @@ -237,44 +239,66 @@ class Transport(threading.Thread, ClosingContextManager):
"class": algorithms.AES,
"mode": modes.CTR,
"block-size": 16,
"iv-size": 16,
"key-size": 16,
},
"aes192-ctr": {
"class": algorithms.AES,
"mode": modes.CTR,
"block-size": 16,
"iv-size": 16,
"key-size": 24,
},
"aes256-ctr": {
"class": algorithms.AES,
"mode": modes.CTR,
"block-size": 16,
"iv-size": 16,
"key-size": 32,
},
"aes128-cbc": {
"class": algorithms.AES,
"mode": modes.CBC,
"block-size": 16,
"iv-size": 16,
"key-size": 16,
},
"aes192-cbc": {
"class": algorithms.AES,
"mode": modes.CBC,
"block-size": 16,
"iv-size": 16,
"key-size": 24,
},
"aes256-cbc": {
"class": algorithms.AES,
"mode": modes.CBC,
"block-size": 16,
"iv-size": 16,
"key-size": 32,
},
"3des-cbc": {
"class": TripleDES,
"mode": modes.CBC,
"block-size": 8,
"iv-size": 8,
"key-size": 24,
},
# aead cipher
"[email protected]": {
"class": aead.AESGCM,
"block-size": 16,
"iv-size": 12,
"key-size": 16,
"is_aead": True,
},
"[email protected]": {
"class": aead.AESGCM,
"block-size": 16,
"iv-size": 12,
"key-size": 32,
"is_aead": True,
},
}

_mac_info = {
Expand Down Expand Up @@ -2039,6 +2063,10 @@ def _get_cipher(self, name, key, iv, operation):
else:
return cipher.decryptor()

def _get_aead_cipher(self, name, key):
aead_cipher = self._cipher_info[name]["class"](key)
return aead_cipher

def _set_forward_agent_handler(self, handler):
if handler is None:

Expand Down Expand Up @@ -2707,18 +2735,32 @@ def _activate_inbound(self):
inbound traffic"""
block_size = self._cipher_info[self.remote_cipher]["block-size"]
if self.server_mode:
IV_in = self._compute_key("A", block_size)
IV_in = self._compute_key(
"A", self._cipher_info[self.remote_cipher]["iv-size"]
)
key_in = self._compute_key(
"C", self._cipher_info[self.remote_cipher]["key-size"]
)
else:
IV_in = self._compute_key("B", block_size)
IV_in = self._compute_key(
"B", self._cipher_info[self.remote_cipher]["iv-size"]
)
key_in = self._compute_key(
"D", self._cipher_info[self.remote_cipher]["key-size"]
)
engine = self._get_cipher(
self.remote_cipher, key_in, IV_in, self._DECRYPT

is_aead = (
True
if self._cipher_info[self.remote_cipher].get("is_aead")
else False
)

if is_aead:
engine = self._get_aead_cipher(self.remote_cipher, key_in)
else:
engine = self._get_cipher(
self.remote_cipher, key_in, IV_in, self._DECRYPT
)
etm = "[email protected]" in self.remote_mac
mac_size = self._mac_info[self.remote_mac]["size"]
mac_engine = self._mac_info[self.remote_mac]["class"]
Expand All @@ -2728,9 +2770,24 @@ def _activate_inbound(self):
mac_key = self._compute_key("E", mac_engine().digest_size)
else:
mac_key = self._compute_key("F", mac_engine().digest_size)
self.packetizer.set_inbound_cipher(
engine, block_size, mac_engine, mac_size, mac_key, etm=etm
)

if is_aead:
self._log(DEBUG, "use aead-cipher, so set mac to None")
self.packetizer.set_inbound_cipher(
engine,
block_size,
None,
16,
bytes(),
etm=False,
aead=is_aead,
iv_in=IV_in,
)
else:
self.packetizer.set_inbound_cipher(
engine, block_size, mac_engine, mac_size, mac_key, etm=etm
)

compress_in = self._compression_info[self.remote_compression][1]
if compress_in is not None and (
self.remote_compression != "[email protected]" or self.authenticated
Expand Down Expand Up @@ -2760,18 +2817,32 @@ def _activate_outbound(self):
self.packetizer.reset_seqno_out()
block_size = self._cipher_info[self.local_cipher]["block-size"]
if self.server_mode:
IV_out = self._compute_key("B", block_size)
IV_out = self._compute_key(
"B", self._cipher_info[self.local_cipher]["iv-size"]
)
key_out = self._compute_key(
"D", self._cipher_info[self.local_cipher]["key-size"]
)
else:
IV_out = self._compute_key("A", block_size)
IV_out = self._compute_key(
"A", self._cipher_info[self.local_cipher]["iv-size"]
)
key_out = self._compute_key(
"C", self._cipher_info[self.local_cipher]["key-size"]
)
engine = self._get_cipher(
self.local_cipher, key_out, IV_out, self._ENCRYPT

is_aead = (
True
if self._cipher_info[self.local_cipher].get("is_aead")
else False
)

if is_aead:
engine = self._get_aead_cipher(self.local_cipher, key_out)
else:
engine = self._get_cipher(
self.local_cipher, key_out, IV_out, self._ENCRYPT
)
etm = "[email protected]" in self.local_mac
mac_size = self._mac_info[self.local_mac]["size"]
mac_engine = self._mac_info[self.local_mac]["class"]
Expand All @@ -2782,9 +2853,30 @@ def _activate_outbound(self):
else:
mac_key = self._compute_key("E", mac_engine().digest_size)
sdctr = self.local_cipher.endswith("-ctr")
self.packetizer.set_outbound_cipher(
engine, block_size, mac_engine, mac_size, mac_key, sdctr, etm=etm
)

if is_aead:
self.packetizer.set_outbound_cipher(
engine,
block_size,
None,
16,
bytes(),
sdctr,
etm=False,
aead=is_aead,
iv_out=IV_out,
)
else:
self.packetizer.set_outbound_cipher(
engine,
block_size,
mac_engine,
mac_size,
mac_key,
sdctr,
etm=etm,
)

compress_out = self._compression_info[self.local_compression][0]
if compress_out is not None and (
self.local_compression != "[email protected]" or self.authenticated
Expand Down

0 comments on commit 39871d2

Please sign in to comment.