engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

deps: remove sshpubkeys dependency as this project is deprecated

Browse source
This commit is contained in:
RhodeCode Admin 2026-01-11 09:35:27 +01:00
parent 511c54b606
commit 513e951219
6 changed files with 453 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

6
requirements.txt
View file

@ -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

3
rhodecode/apps/admin/views/users.py
View file

@ -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(

3
rhodecode/apps/my_account/views/my_account_ssh_keys.py
View file

@ -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(

15
rhodecode/apps/repository/views/repo_commits.py
View file

@ -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()

441
rhodecode/lib/sshpubkeys.py Normal file
View file

@ -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 <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")

7
rhodecode/model/ssh_key.py
View file

@ -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: