perf: support ed25519 SSH Key

fix: codacy ci
fix: password use bytes

apps/assets/models/base.py

@@ -6,23 +6,21 @@ import uuid
 from hashlib import md5
 
 import sshpubkeys
+from django.conf import settings
 from django.core.cache import cache
 from django.db import models
+from django.db.models import QuerySet
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
-from django.conf import settings
-from django.db.models import QuerySet
 
-from common.utils import random_string, signer
+from common.db import fields
+from common.utils import random_string
 from common.utils import (
     ssh_key_string_to_obj, ssh_key_gen, get_logger, lazyproperty
 )
-from common.utils.encode import ssh_pubkey_gen
+from common.utils.encode import parse_ssh_public_key_str
-from common.validators import alphanumeric
-from common.db import fields
 from orgs.mixins.models import OrgModelMixin
 
-
 logger = get_logger(__file__)
 
 
@@ -68,7 +66,7 @@ class AuthMixin:
             public_key = self.public_key
         elif self.private_key:
             try:
-                public_key = ssh_pubkey_gen(private_key=self.private_key, password=self.password)
+                public_key = parse_ssh_public_key_str(private_key=self.private_key, password=self.password)
             except IOError as e:
                 return str(e)
         else:
@@ -234,4 +232,3 @@ class BaseUser(OrgModelMixin, AuthMixin):
 
     class Meta:
         abstract = True
-

apps/assets/serializers/base.py

@@ -1,24 +1,24 @@
 # -*- coding: utf-8 -*-
 #
-from io import StringIO
 
 from django.utils.translation import ugettext_lazy as _
 from rest_framework import serializers
 
-from common.utils import ssh_pubkey_gen, ssh_private_key_gen, validate_ssh_private_key
-from common.drf.fields import EncryptedField
 from assets.models import Type
+from common.drf.fields import EncryptedField
+from common.utils import validate_ssh_private_key, parse_ssh_private_key_str, parse_ssh_public_key_str
 from .utils import validate_password_for_ansible
 
 
 class AuthSerializer(serializers.ModelSerializer):
     password = EncryptedField(required=False, allow_blank=True, allow_null=True, max_length=1024, label=_('Password'))
-    private_key = EncryptedField(required=False, allow_blank=True, allow_null=True, max_length=16384, label=_('Private key'))
+    private_key = EncryptedField(required=False, allow_blank=True, allow_null=True, max_length=16384,
+                                 label=_('Private key'))
 
     def gen_keys(self, private_key=None, password=None):
         if private_key is None:
             return None, None
-        public_key = ssh_pubkey_gen(private_key=private_key, password=password)
+        public_key = parse_ssh_public_key_str(text=private_key, password=password)
         return private_key, public_key
 
     def save(self, **kwargs):
@@ -57,10 +57,7 @@ class AuthSerializerMixin(serializers.ModelSerializer):
         if not valid:
             raise serializers.ValidationError(_("private key invalid or passphrase error"))
 
-        private_key = ssh_private_key_gen(private_key, password=passphrase)
+        private_key = parse_ssh_private_key_str(private_key, password=passphrase)
-        string_io = StringIO()
-        private_key.write_private_key(string_io)
-        private_key = string_io.getvalue()
         return private_key
 
     def validate_public_key(self, public_key):

apps/assets/serializers/utils.py

@@ -1,6 +1,8 @@
 from django.utils.translation import ugettext_lazy as _
 from rest_framework import serializers
 
+from common.utils import validate_ssh_private_key, parse_ssh_private_key_str
+
 
 def validate_password_for_ansible(password):
     """ 校验 Ansible 不支持的特殊字符 """
@@ -15,3 +17,9 @@ def validate_password_for_ansible(password):
     if '"' in password:
         raise serializers.ValidationError(_('Password can not contains `"` '))
 
+
+def validate_ssh_key(ssh_key, passphrase=None):
+    valid = validate_ssh_private_key(ssh_key, password=passphrase)
+    if not valid:
+        raise serializers.ValidationError(_("private key invalid or passphrase error"))
+    return parse_ssh_private_key_str(ssh_key, passphrase)

apps/common/utils/encode.py

@@ -1,24 +1,23 @@
 # -*- coding: utf-8 -*-
 #
-import re
-import json
-from six import string_types
 import base64
-import os
-import time
 import hashlib
+import json
+import os
+import re
+import time
 from io import StringIO
-from itertools import chain
 
 import paramiko
 import sshpubkeys
+from cryptography.hazmat.primitives import serialization
+from django.conf import settings
+from django.core.serializers.json import DjangoJSONEncoder
 from itsdangerous import (
     TimedJSONWebSignatureSerializer, JSONWebSignatureSerializer,
     BadSignature, SignatureExpired
 )
-from django.conf import settings
+from six import string_types
-from django.core.serializers.json import DjangoJSONEncoder
-from django.db.models.fields.files import FileField
 
 from .http import http_date
 
@@ -69,22 +68,20 @@ class Signer(metaclass=Singleton):
             return None
 
 
+_supported_paramiko_ssh_key_types = (paramiko.RSAKey, paramiko.DSSKey, paramiko.Ed25519Key)
+
+
 def ssh_key_string_to_obj(text, password=None):
     key = None
-    try:
+    for ssh_key_type in _supported_paramiko_ssh_key_types:
-        key = paramiko.RSAKey.from_private_key(StringIO(text), password=password)
+        if not isinstance(ssh_key_type, paramiko.PKey):
-    except paramiko.SSHException:
+            continue
-        pass
+        try:
-    else:
+            key = ssh_key_type.from_private_key(StringIO(text), password=password)
-        return key
+        except paramiko.SSHException:
-
+            pass
-    try:
+        else:
-        key = paramiko.DSSKey.from_private_key(StringIO(text), password=password)
+            return key
-    except paramiko.SSHException:
-        pass
-    else:
-        return key
-
     return key
 
 
@@ -98,7 +95,7 @@ def ssh_private_key_gen(private_key, password=None):
 
 def ssh_pubkey_gen(private_key=None, username='jumpserver', hostname='localhost', password=None):
     private_key = ssh_private_key_gen(private_key, password=password)
-    if not isinstance(private_key, (paramiko.RSAKey, paramiko.DSSKey)):
+    if not isinstance(private_key, _supported_paramiko_ssh_key_types):
         raise IOError('Invalid private key')
 
     public_key = "%(key_type)s %(key_content)s %(username)s@%(hostname)s" % {
@@ -137,17 +134,63 @@ def ssh_key_gen(length=2048, type='rsa', password=None, username='jumpserver', h
 
 
 def validate_ssh_private_key(text, password=None):
-    if isinstance(text, bytes):
+    if isinstance(text, str):
         try:
-            text = text.decode("utf-8")
+            text = text.encode("utf-8")
         except UnicodeDecodeError:
             return False
 
-    key = ssh_key_string_to_obj(text, password=password)
+    key = parse_ssh_private_key_str(text, password=password)
-    if key is None:
+    return bool(key)
-        return False
+
-    else:
+
-        return True
+def parse_ssh_private_key_str(text: bytes, password=None) -> str:
+    private_key = _parse_ssh_private_key(text, password=password)
+    if private_key is None:
+        return ""
+    private_key_bytes = private_key.private_bytes(serialization.Encoding.PEM,
+                                                  serialization.PrivateFormat.OpenSSH,
+                                                  serialization.NoEncryption())
+    return private_key_bytes.decode('utf-8')
+
+
+def parse_ssh_public_key_str(text: bytes = "", password=None) -> str:
+    private_key = _parse_ssh_private_key(text, password=password)
+    if private_key is None:
+        return ""
+    public_key_bytes = private_key.public_key().public_bytes(serialization.Encoding.OpenSSH,
+                                                             serialization.PublicFormat.OpenSSH)
+    return public_key_bytes.decode('utf-8')
+
+
+def _parse_ssh_private_key(text, password=None):
+    """
+    text: bytes
+    password: str
+    return:private key types:
+                ec.EllipticCurvePrivateKey,
+                rsa.RSAPrivateKey,
+                dsa.DSAPrivateKey,
+                ed25519.Ed25519PrivateKey,
+    """
+    if isinstance(text, str):
+        try:
+            text = text.encode("utf-8")
+        except UnicodeDecodeError:
+            return None
+    if password is not None:
+        if isinstance(password, str):
+            try:
+                password = password.encode("utf-8")
+            except UnicodeDecodeError:
+                return None
+
+    try:
+        private_key = serialization.load_ssh_private_key(text, password=password)
+        return private_key
+    except (ValueError, TypeError):
+        pass
+    return None
 
 
 def validate_ssh_public_key(text):