441 lines
17 KiB
Python
441 lines
17 KiB
Python
# Copyright (C) 2017-2024 RhodeCode GmbH
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License, version 3
|
|
# (only), as published by the Free Software Foundation.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
# This program is dual-licensed. If you wish to learn more about the
|
|
# RhodeCode Enterprise Edition, including its added features, Support services,
|
|
# and proprietary license terms, please see https://rhodecode.com/licenses/
|
|
|
|
import base64
|
|
import binascii
|
|
import hashlib
|
|
import re
|
|
import struct
|
|
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.hazmat.primitives.asymmetric.dsa import DSAParameterNumbers, DSAPublicNumbers
|
|
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicNumbers
|
|
|
|
|
|
class InvalidKeyException(Exception):
|
|
"""Invalid key - something is wrong with the key, and it should not be accepted, as OpenSSH will not work with it."""
|
|
|
|
|
|
class InvalidKeyError(InvalidKeyException):
|
|
"""Invalid key - something is wrong with the key, and it should not be accepted, as OpenSSH will not work with it."""
|
|
|
|
|
|
class InvalidKeyLengthError(InvalidKeyError):
|
|
"""Invalid key length - either too short or too long.
|
|
|
|
See also TooShortKeyError and TooLongKeyError."""
|
|
|
|
|
|
class TooShortKeyError(InvalidKeyLengthError):
|
|
"""Key is shorter than what the specification allows."""
|
|
|
|
|
|
class TooLongKeyError(InvalidKeyLengthError):
|
|
"""Key is longer than what the specification allows."""
|
|
|
|
|
|
class InvalidTypeError(InvalidKeyException):
|
|
"""Key type is invalid or unrecognized."""
|
|
|
|
|
|
class MalformedDataError(InvalidKeyError):
|
|
"""The key is invalid - unable to parse the data. The data may be corrupted, truncated, or includes extra content that is not allowed."""
|
|
|
|
|
|
class InvalidOptionsError(MalformedDataError):
|
|
"""Options string is invalid: it contains invalid characters, unrecognized options, or is otherwise malformed."""
|
|
|
|
|
|
class InvalidOptionNameError(MalformedDataError):
|
|
"""Invalid option name (contains disallowed characters, or is unrecognized.)."""
|
|
|
|
|
|
class MissingMandatoryOptionValueError(InvalidOptionNameError):
|
|
"""Mandatory option value is missing."""
|
|
|
|
|
|
class SSHKey: # pylint:disable=too-many-instance-attributes
|
|
"""Represents a single SSH public key.
|
|
|
|
Usage:
|
|
ssh_key = SSHKey()
|
|
ssh_key.parse(key_data)
|
|
"""
|
|
|
|
# Key length constraints
|
|
RSA_MIN_LENGTH_STRICT = 1024
|
|
RSA_MAX_LENGTH_STRICT = 16384
|
|
|
|
DSA_MIN_LENGTH_STRICT = 1024
|
|
DSA_MAX_LENGTH_STRICT = 1024
|
|
DSA_N_LENGTH = 160
|
|
|
|
# Supported ECDSA curves
|
|
ECDSA_CURVE_DATA = {
|
|
b"nistp256": ec.SECP256R1(),
|
|
b"nistp192": ec.SECP192R1(),
|
|
b"nistp224": ec.SECP224R1(),
|
|
b"nistp384": ec.SECP384R1(),
|
|
b"nistp521": ec.SECP521R1(),
|
|
}
|
|
|
|
# SSH options specification (OpenSSH 8.3)
|
|
# Format: (option_name, value_is_mandatory)
|
|
# Option names are case-insensitive but must be lowercase here
|
|
OPTIONS_SPEC = [
|
|
("agent-forwarding", False),
|
|
("cert-authority", False),
|
|
("command", True),
|
|
("environment", True),
|
|
("expiry-time", True),
|
|
("from", True),
|
|
("no-agent-forwarding", False),
|
|
("no-port-forwarding", False),
|
|
("no-pty", False),
|
|
("no-touch-required", False),
|
|
("no-user-rc", False),
|
|
("no-x11-forwarding", False),
|
|
("permitlisten", True),
|
|
("permitopen", True),
|
|
("port-forwarding", False),
|
|
("principals", True),
|
|
("pty", False),
|
|
("restrict", False),
|
|
("tunnel", True),
|
|
("user-rc", False),
|
|
("x11-forwarding", False),
|
|
]
|
|
OPTION_NAME_RE = re.compile(r"^[A-Za-z0-9-]+$")
|
|
|
|
# Binary format constants
|
|
INT_LEN = 4
|
|
FIELDS = ["rsa", "dsa", "ecdsa", "bits", "comment", "options", "options_raw", "key_type"]
|
|
|
|
def __init__(self):
|
|
self._decoded_key = None
|
|
self.rsa = None
|
|
self.dsa = None
|
|
self.ecdsa = None
|
|
self.bits = None
|
|
self.comment = None
|
|
self.options = None
|
|
self.options_raw = None
|
|
self.key_type = None
|
|
self._key_data = None
|
|
|
|
def __str__(self):
|
|
return f"Key type: {self.key_type.decode()}, bits: {self.bits}, options: {self.options}"
|
|
|
|
@property
|
|
def key(self):
|
|
"""Base64 encoded key"""
|
|
return base64.b64encode(self._decoded_key)
|
|
|
|
@property
|
|
def keydata(self):
|
|
return self._key_data
|
|
|
|
def hash_md5(self):
|
|
"""Calculate md5 fingerprint."""
|
|
fp_plain = hashlib.md5(self._decoded_key).hexdigest()
|
|
return "MD5:" + ":".join(a + b for a, b in zip(fp_plain[::2], fp_plain[1::2]))
|
|
|
|
def hash_sha256(self) -> str:
|
|
"""Calculate sha256 fingerprint."""
|
|
fp_plain = hashlib.sha256(self._decoded_key).digest()
|
|
return (b"SHA256:" + base64.b64encode(fp_plain).replace(b"=", b"")).decode("utf-8")
|
|
|
|
def hash_sha512(self) -> str:
|
|
"""Calculates sha512 fingerprint."""
|
|
fp_plain = hashlib.sha512(self._decoded_key).digest()
|
|
return (b"SHA512:" + base64.b64encode(fp_plain).replace(b"=", b"")).decode("utf-8")
|
|
|
|
@classmethod
|
|
def _parse_long(cls, data):
|
|
"""Calculate two's complement."""
|
|
ret = 0
|
|
for byte in data:
|
|
ret = (ret << 8) + byte
|
|
return ret
|
|
|
|
@classmethod
|
|
def _bits_in_number(cls, number):
|
|
return len(format(number, "b"))
|
|
|
|
def _process_ssh_rsa(self, data):
|
|
"""Parses ssh-rsa public keys."""
|
|
current_position, raw_e = self._unpack_by_int(data, 0)
|
|
current_position, raw_n = self._unpack_by_int(data, current_position)
|
|
|
|
unpacked_e = self._parse_long(raw_e)
|
|
unpacked_n = self._parse_long(raw_n)
|
|
|
|
self.rsa = RSAPublicNumbers(unpacked_e, unpacked_n).public_key(default_backend())
|
|
self.bits = self.rsa.key_size
|
|
|
|
min_length = self.RSA_MIN_LENGTH_STRICT
|
|
max_length = self.RSA_MAX_LENGTH_STRICT
|
|
|
|
if self.bits < min_length:
|
|
raise TooShortKeyError(
|
|
f"{self.key_type.decode()} key data can not be shorter than {min_length} bits (was {self.bits})"
|
|
)
|
|
if self.bits > max_length:
|
|
raise TooLongKeyError(
|
|
f"{self.key_type.decode()} key data can not be longer than {max_length} bits (was {self.bits})"
|
|
)
|
|
return current_position
|
|
|
|
def _process_ssh_dss(self, data):
|
|
"""Parses ssh-dsa public keys."""
|
|
data_fields = {}
|
|
current_position = 0
|
|
for item in ("p", "q", "g", "y"):
|
|
current_position, value = self._unpack_by_int(data, current_position)
|
|
data_fields[item] = self._parse_long(value)
|
|
|
|
q_bits = self._bits_in_number(data_fields["q"])
|
|
p_bits = self._bits_in_number(data_fields["p"])
|
|
if q_bits != self.DSA_N_LENGTH:
|
|
raise InvalidKeyError(f"Incorrect DSA key parameters: bits(p)={self.bits}, q={q_bits}")
|
|
|
|
min_length = self.DSA_MIN_LENGTH_STRICT
|
|
max_length = self.DSA_MAX_LENGTH_STRICT
|
|
|
|
if p_bits < min_length:
|
|
raise TooShortKeyError(
|
|
f"{self.key_type.decode()} key can not be shorter than {min_length} bits (was {p_bits})"
|
|
)
|
|
if p_bits > max_length:
|
|
raise TooLongKeyError(
|
|
f"{self.key_type.decode()} key data can not be longer than {max_length} bits (was {p_bits})"
|
|
)
|
|
|
|
dsa_parameters = DSAParameterNumbers(data_fields["p"], data_fields["q"], data_fields["g"])
|
|
self.dsa = DSAPublicNumbers(data_fields["y"], dsa_parameters).public_key(default_backend())
|
|
self.bits = self.dsa.key_size
|
|
|
|
return current_position
|
|
|
|
def _process_ecdsa_sha(self, data):
|
|
"""Parses ecdsa-sha public keys."""
|
|
current_position, curve_information = self._unpack_by_int(data, 0)
|
|
if curve_information not in self.ECDSA_CURVE_DATA:
|
|
raise NotImplementedError(f"Invalid curve type: {curve_information}")
|
|
curve = self.ECDSA_CURVE_DATA[curve_information]
|
|
|
|
current_position, key_data = self._unpack_by_int(data, current_position)
|
|
try:
|
|
self.ecdsa = ec.EllipticCurvePublicKey.from_encoded_point(curve, key_data)
|
|
except ValueError as ex:
|
|
raise InvalidKeyError("Invalid ecdsa key") from ex
|
|
self.bits = curve.key_size
|
|
return current_position
|
|
|
|
def _process_ed25519(self, data):
|
|
"""Parses ed25519 keys.
|
|
|
|
There is no (clear) way to validate ed25519 keys. This only
|
|
checks data length 256 bits, but does not try to validate
|
|
the key in any way."""
|
|
|
|
current_position, verifying_key = self._unpack_by_int(data, 0)
|
|
verifying_key_length = len(verifying_key) * 8
|
|
verifying_key = self._parse_long(verifying_key)
|
|
|
|
if verifying_key < 0:
|
|
raise InvalidKeyError("ed25519 verifying key must be >0.")
|
|
|
|
self.bits = verifying_key_length
|
|
if self.bits != 256:
|
|
raise InvalidKeyLengthError(f"ed25519 keys must be 256 bits (was {self.bits} bits)")
|
|
return current_position
|
|
|
|
def _process_sk_ecdsa_sha(self, data):
|
|
"""Parses sk_ecdsa-sha public keys."""
|
|
current_position = self._process_ecdsa_sha(data)
|
|
current_position, application = self._unpack_by_int(data, current_position)
|
|
return current_position
|
|
|
|
def _process_sk_ed25519(self, data):
|
|
"""Parses sk_ed25519 public keys."""
|
|
current_position = self._process_ed25519(data)
|
|
current_position, application = self._unpack_by_int(data, current_position)
|
|
return current_position
|
|
|
|
def _process_key(self, data):
|
|
if self.key_type == b"ssh-rsa":
|
|
return self._process_ssh_rsa(data)
|
|
if self.key_type == b"ssh-dss":
|
|
return self._process_ssh_dss(data)
|
|
if self.key_type.strip().startswith(b"ecdsa-sha"):
|
|
return self._process_ecdsa_sha(data)
|
|
if self.key_type == b"ssh-ed25519":
|
|
return self._process_ed25519(data)
|
|
if self.key_type.strip().startswith(b"sk-ecdsa-sha"):
|
|
return self._process_sk_ecdsa_sha(data)
|
|
if self.key_type.strip().startswith(b"sk-ssh-ed25519"):
|
|
return self._process_sk_ed25519(data)
|
|
raise NotImplementedError(f"Invalid key type: {self.key_type.decode()}")
|
|
|
|
@classmethod
|
|
def decode_key(cls, pubkey_content):
|
|
"""Decode base64 coded part of the key."""
|
|
try:
|
|
decoded_key = base64.b64decode(pubkey_content.encode("ascii"))
|
|
except (TypeError, binascii.Error) as ex:
|
|
raise MalformedDataError("Unable to decode the key") from ex
|
|
return decoded_key
|
|
|
|
def _split_key(self, data):
|
|
options_raw = None
|
|
# Terribly inefficient way to remove options, but hey, it works.
|
|
if not data.startswith("ssh-") and not data.startswith("ecdsa-") and not data.startswith("sk-"):
|
|
quote_open = False
|
|
for i, character in enumerate(data):
|
|
if character == '"': # only double quotes are allowed, no need to care about single quotes
|
|
quote_open = not quote_open
|
|
if quote_open:
|
|
continue
|
|
if character == " ":
|
|
# Data begins after the first space
|
|
options_raw = data[:i]
|
|
data = data[i + 1 :]
|
|
break
|
|
else:
|
|
raise MalformedDataError("Couldn't find beginning of the key data")
|
|
key_parts = data.strip().split(None, 2)
|
|
if len(key_parts) < 2: # Key type and content are mandatory fields.
|
|
raise InvalidKeyError("Unexpected key format: at least type and base64 encoded value is required")
|
|
if len(key_parts) == 3:
|
|
self.comment = key_parts[2]
|
|
key_parts = key_parts[0:2]
|
|
if options_raw:
|
|
# Populate and parse options field.
|
|
self.options_raw = options_raw
|
|
self.options = self.parse_options(self.options_raw)
|
|
else:
|
|
# Set empty defaults for fields
|
|
self.options_raw = None
|
|
self.options = {}
|
|
return key_parts
|
|
|
|
def _unpack_by_int(self, data, current_position):
|
|
"""Returns a tuple with (location of next data field, contents of requested data field)."""
|
|
# Unpack length of data field
|
|
try:
|
|
requested_data_length = struct.unpack(">I", data[current_position : current_position + self.INT_LEN])[0]
|
|
except struct.error as ex:
|
|
raise MalformedDataError(f"Unable to unpack {self.INT_LEN} bytes from the data") from ex
|
|
|
|
# Move pointer to the beginning of the data field
|
|
current_position += self.INT_LEN
|
|
remaining_data_length = len(data[current_position:])
|
|
|
|
if remaining_data_length < requested_data_length:
|
|
raise MalformedDataError(
|
|
f"Requested {requested_data_length} bytes, but only {remaining_data_length} bytes available."
|
|
)
|
|
|
|
next_data = data[current_position : current_position + requested_data_length]
|
|
# Move the pointer to the end of the data field
|
|
current_position += requested_data_length
|
|
return current_position, next_data
|
|
|
|
def parse_options(self, options):
|
|
"""Parses ssh options string."""
|
|
quote_open = False
|
|
parsed_options = {}
|
|
|
|
def parse_add_single_option(opt):
|
|
"""Parses and validates a single option, and adds it to parsed_options field."""
|
|
if "=" in opt:
|
|
opt_name, opt_value = opt.split("=", 1)
|
|
opt_value = opt_value.replace('"', "")
|
|
else:
|
|
opt_name = opt
|
|
opt_value = True
|
|
if " " in opt_name or not self.OPTION_NAME_RE.match(opt_name):
|
|
raise InvalidOptionNameError(f"{opt_name} is not a valid option name.")
|
|
|
|
for valid_opt_name, value_required in self.OPTIONS_SPEC:
|
|
if opt_name.lower() == valid_opt_name:
|
|
if value_required and opt_value is True:
|
|
raise MissingMandatoryOptionValueError(f"{opt_name} is missing a mandatory value.")
|
|
break
|
|
|
|
if opt_name not in parsed_options:
|
|
parsed_options[opt_name] = []
|
|
parsed_options[opt_name].append(opt_value)
|
|
|
|
start_of_current_opt = 0
|
|
for i, character in enumerate(options):
|
|
if character == '"': # only double quotes are allowed, no need to care about single quotes
|
|
quote_open = not quote_open
|
|
if quote_open:
|
|
continue
|
|
if character == ",":
|
|
opt = options[start_of_current_opt:i]
|
|
parse_add_single_option(opt)
|
|
start_of_current_opt = i + 1
|
|
|
|
# Parse the remaining option after the last comma (or the only option if no commas)
|
|
if start_of_current_opt < len(options):
|
|
parse_add_single_option(options[start_of_current_opt:])
|
|
if quote_open:
|
|
raise InvalidOptionsError("Unbalanced quotes.")
|
|
return parsed_options
|
|
|
|
def parse(self, key_data=None):
|
|
"""Validates the SSH public key.
|
|
|
|
Throws exception for invalid keys. Otherwise returns None.
|
|
|
|
Populates key_type, bits and bits fields.
|
|
|
|
For rsa keys, see the field "rsa" for raw public key data.
|
|
For dsa keys, see the field "dsa".
|
|
For ecdsa keys, see the field "ecdsa"."""
|
|
if key_data is None:
|
|
raise ValueError("Key data must be supplied either in constructor or to parse()")
|
|
self._key_data = key_data
|
|
|
|
if key_data.startswith("---- BEGIN SSH2 PUBLIC KEY ----"):
|
|
# SSH2 key format
|
|
key_type = None # There is no redundant key-type field - skip comparing plain-text and encoded data.
|
|
pubkey_content = "".join([line for line in key_data.split("\n") if ":" not in line and "----" not in line])
|
|
else:
|
|
key_parts = self._split_key(key_data)
|
|
key_type = key_parts[0]
|
|
pubkey_content = key_parts[1]
|
|
|
|
self._decoded_key = self.decode_key(pubkey_content)
|
|
|
|
# Check key type
|
|
current_position, unpacked_key_type = self._unpack_by_int(self._decoded_key, 0)
|
|
if key_type is not None and key_type != unpacked_key_type.decode():
|
|
raise InvalidTypeError(f"Key type mismatch: {key_type} != {unpacked_key_type.decode()}")
|
|
|
|
self.key_type = unpacked_key_type
|
|
|
|
key_data_length = self._process_key(self._decoded_key[current_position:])
|
|
current_position = current_position + key_data_length
|
|
|
|
if current_position != len(self._decoded_key):
|
|
raise MalformedDataError(f"Leftover data: {len(self._decoded_key) - current_position} bytes")
|