From 513e95121966038c0f950ad4ea78d066e0f7ad46 Mon Sep 17 00:00:00 2001 From: RhodeCode Admin Date: Sun, 11 Jan 2026 09:35:27 +0100 Subject: [PATCH] deps: remove sshpubkeys dependency as this project is deprecated --- requirements.txt | 6 - rhodecode/apps/admin/views/users.py | 3 +- .../my_account/views/my_account_ssh_keys.py | 3 +- .../apps/repository/views/repo_commits.py | 15 +- rhodecode/lib/sshpubkeys.py | 441 ++++++++++++++++++ rhodecode/model/ssh_key.py | 7 +- 6 files changed, 453 insertions(+), 22 deletions(-) create mode 100644 rhodecode/lib/sshpubkeys.py diff --git a/requirements.txt b/requirements.txt index 5f0873e6..59792712 100644 --- a/requirements.txt +++ b/requirements.txt @@ -279,12 +279,6 @@ s3fs==2025.10.0 multidict==6.7.0 propcache==0.4.1 simplejson==3.19.2 -sshpubkeys==3.3.1 - cryptography==44.0.1 - cffi==2.0.0 - pycparser==2.23 - ecdsa==0.19.1 - six==1.17.0 sqlalchemy==1.4.54 greenlet==3.3.0 typing_extensions==4.15.0 diff --git a/rhodecode/apps/admin/views/users.py b/rhodecode/apps/admin/views/users.py index f9c8f6b5..16c328f4 100644 --- a/rhodecode/apps/admin/views/users.py +++ b/rhodecode/apps/admin/views/users.py @@ -855,8 +855,9 @@ class UsersView(UserAppView): key = SshKeyModel().parse_key(key_data.strip()) fingerprint = key.hash_md5() + keydata = key.keydata - ssh_key = SshKeyModel().create(c.user.user_id, fingerprint, key.keydata, description) + ssh_key = SshKeyModel().create(c.user.user_id, fingerprint, keydata, description) ssh_key_data = ssh_key.get_api_data() audit_logger.store_web( diff --git a/rhodecode/apps/my_account/views/my_account_ssh_keys.py b/rhodecode/apps/my_account/views/my_account_ssh_keys.py index 60e07793..838b02e8 100644 --- a/rhodecode/apps/my_account/views/my_account_ssh_keys.py +++ b/rhodecode/apps/my_account/views/my_account_ssh_keys.py @@ -82,8 +82,9 @@ class MyAccountSshKeysView(BaseAppView, DataGridAppView): key = SshKeyModel().parse_key(key_data.strip()) fingerprint = key.hash_md5() + keydata = key.keydata - ssh_key = SshKeyModel().create(c.user.user_id, fingerprint, key.keydata, description) + ssh_key = SshKeyModel().create(c.user.user_id, fingerprint, keydata, description) ssh_key_data = ssh_key.get_api_data() audit_logger.store_web( diff --git a/rhodecode/apps/repository/views/repo_commits.py b/rhodecode/apps/repository/views/repo_commits.py index ce8e2fc1..105c8135 100644 --- a/rhodecode/apps/repository/views/repo_commits.py +++ b/rhodecode/apps/repository/views/repo_commits.py @@ -714,22 +714,17 @@ class RepoCommitsView(RepoAppView): text = self.request.POST.get("text") version = self.request.POST.get("version") if text == comment.text: - log.warning( - "Comment(repo): Trying to create new version with the same comment body {}".format( - comment_id, - ) - ) + log.warning("Comment(repo): Trying to create new version with the same comment body %s", comment_id) raise HTTPNotFound() if version.isdigit(): version = int(version) else: log.warning( - "Comment(repo): Wrong version type {} {} for comment {}".format( - version, - type(version), - comment_id, - ) + "Comment(repo): Wrong version type %s %s for comment %s", + version, + type(version), + comment_id, ) raise HTTPNotFound() diff --git a/rhodecode/lib/sshpubkeys.py b/rhodecode/lib/sshpubkeys.py new file mode 100644 index 00000000..23f57a2b --- /dev/null +++ b/rhodecode/lib/sshpubkeys.py @@ -0,0 +1,441 @@ +# 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 . +# +# 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") diff --git a/rhodecode/model/ssh_key.py b/rhodecode/model/ssh_key.py index 9a1ed24f..b0a4179d 100644 --- a/rhodecode/model/ssh_key.py +++ b/rhodecode/model/ssh_key.py @@ -19,8 +19,7 @@ import logging import traceback -import sshpubkeys -import sshpubkeys.exceptions +from rhodecode.lib.sshpubkeys import SSHKey, InvalidKeyException from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives import serialization as crypto_serialization @@ -52,11 +51,11 @@ class SshKeyModel(BaseModel): :param key_data: :return: """ - ssh = sshpubkeys.SSHKey(strict_mode=True) + ssh = SSHKey() try: ssh.parse(key_data) return ssh - except sshpubkeys.exceptions.InvalidKeyException as err: + except InvalidKeyException as err: log.error("Invalid key: %s", err) raise except NotImplementedError as err: