feat: support ed25519 key

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 django.core.serializers.json import DjangoJSONEncoder
-from django.db.models.fields.files import FileField
+from six import string_types
 
 from .http import http_date
 
@@ -69,22 +68,19 @@ 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:
-        key = paramiko.RSAKey.from_private_key(StringIO(text), password=password)
-    except paramiko.SSHException:
-        pass
-    else:
-        return key
-
-    try:
-        key = paramiko.DSSKey.from_private_key(StringIO(text), password=password)
-    except paramiko.SSHException:
-        pass
-    else:
-        return key
-
+    for ssh_key_type in _supported_paramiko_ssh_key_types:
+        if not isinstance(ssh_key_type, paramiko.PKey):
+            continue
+        try:
+            key = ssh_key_type.from_private_key(StringIO(text), password=password)
+            return key
+        except paramiko.SSHException:
+            pass
     return key
 
 
@@ -137,17 +133,68 @@ 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
+    if isinstance(password, str):
+        try:
+            password = password.encode("utf-8")
         except UnicodeDecodeError:
             return False
 
-    key = ssh_key_string_to_obj(text, password=password)
-    if key is None:
-        return False
-    else:
-        return True
+    key = parse_ssh_private_key_str(text, password=password)
+    return bool(key)
+
+
+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):