使用 paramiko 和 sqlalchemy 在 python SFTP 服务器/postgresql 数据库中键入错误

Question

我正在尝试使用 paramiko 库在 python 中设置 SFTP 服务器。 我正在使用 PostgreSQL 来通过 SQLAlchemy 库存储用户凭据。

当我尝试使用 FileZilla 或 WinSCP 登录服务器时，此功能出现错误：

def authenticate_user(username: str, password:str = None, public_key:str = None) -> bool:
engine = provide_engine()
if not engine:
    return False

Session = sessionmaker(bind=engine)
session = Session()

try:
    stmt = select(User).where(User.username == username)
    user = session.execute(stmt).scalars().first()

    if not user:
        return False

    if password:
        print("Before checking.")
        if user.check_password(password):
            print("After checking.")
            return True
    elif public_key and user.ssh_keys and len(user.ssh_keys) > 0:
        for key in user.ssh_keys:
            if key.public_key == public_key:
                return True
            
    return False
finally:
    session.close()

错误是：

ERROR:paramiko.transport:Unknown exception: string argument without an encoding
ERROR:paramiko.transport:Traceback (most recent call last):
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\paramiko\transport.py", line 2262, in run
ERROR:paramiko.transport:    handler(m)
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\paramiko\auth_handler.py", line 609, in _parse_userauth_request
ERROR:paramiko.transport:    result = self.transport.server_object.check_auth_password(
ERROR:paramiko.transport:             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "d:\Files\*******\Python\cr-beta\database_sftp\SFTPServer.py", line 20, in check_auth_password
ERROR:paramiko.transport:    if authenticate_user(username, password=password):
ERROR:paramiko.transport:       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "d:\Files\*******\Python\cr-beta\database_sftp\SFTPServer.py", line 98, in authenticate_user
ERROR:paramiko.transport:    user = session.execute(stmt).scalars().first()
ERROR:paramiko.transport:           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\result.py", line 1786, in first  
ERROR:paramiko.transport:    return self._only_one_row(
ERROR:paramiko.transport:           ^^^^^^^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\result.py", line 749, in _only_one_row
ERROR:paramiko.transport:    row: Optional[_InterimRowType[Any]] = onerow(hard_close=True)
ERROR:paramiko.transport:                                          ^^^^^^^^^^^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\result.py", line 1673, in _fetchone_impl
ERROR:paramiko.transport:    return self._real_result._fetchone_impl(hard_close=hard_close)
ERROR:paramiko.transport:           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\result.py", line 2259, in _fetchone_impl
ERROR:paramiko.transport:    row = next(self.iterator, _NO_ROW)
ERROR:paramiko.transport:          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\orm\loading.py", line 219, in chunks    
ERROR:paramiko.transport:    fetch = cursor._raw_all_rows()
ERROR:paramiko.transport:            ^^^^^^^^^^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\result.py", line 541, in _raw_all_rows
ERROR:paramiko.transport:    return [make_row(row) for row in rows]
ERROR:paramiko.transport:            ^^^^^^^^^^^^^
ERROR:paramiko.transport:  File "lib\\sqlalchemy\\cyextension\\resultproxy.pyx", line 22, in sqlalchemy.cyextension.resultproxy.BaseRow.__init__
ERROR:paramiko.transport:  File "lib\\sqlalchemy\\cyextension\\resultproxy.pyx", line 79, in sqlalchemy.cyextension.resultproxy._apply_processors
ERROR:paramiko.transport:  File "C:\Users\*******\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\sql\sqltypes.py", line 913, in process  
ERROR:paramiko.transport:    value = bytes(value)
ERROR:paramiko.transport:            ^^^^^^^^^^^^
ERROR:paramiko.transport:TypeError: string argument without an encoding

我不明白问题出在哪里，还因为，用户名作为字符串存储在数据库中，而密码我认为处理得很好。 在User表的声明下：

class User(Base):
__tablename__ = USERS_TABLE
id = Column(Integer, primary_key=True)
username = Column(String, unique=True, nullable=False)
password_hash = Column(LargeBinary, nullable=False)
created_at = Column(DateTime, default=datetime.now(timezone.utc))

ssh_keys = relationship("SSHKey", back_populates="user")

def set_hashed_password(self, hashed_password: bytes):
    self.password_hash = hashed_password

def set_password(self, password: str):
    self.password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt())

def check_password(self, password: str) -> bool:
    return bcrypt.checkpw(password.encode(), self.password_hash)

我想指定我正在尝试使用用户名和密码登录，而不是使用 ssh 密钥登录。

更新： 我尝试使用以下方法检查真实的列类型：

from sqlalchemy import inspect
insp = inspect(engine)
print(list(x["type"] for x in insp.get_columns(USERS_TABLE) if x["name"] == "password_hash"))

并在控制台中打印 [VARCHAR()]。

我还检查了 pgAdmin4，列的类型是字符变化的。

为什么在 User 类中我将列设置为 LargeBinary 它是一个可变字符？这是问题吗？

Answer 1

PostgreSQL 表已使用错误的类型创建。 password_hashed 列类型是字符变化的，因为类型后来发生了更改。