Flask-SqlAlchemy、Bcrypt、Postgres 的编码问题

Posted 2023-02-23

【中文标题】Flask-SqlAlchemy、Bcrypt、Postgres 的编码问题

【英文标题】：Flask-SqlAlchemy, Bcrypt, Postgres issue with encoding

【发布时间】：2021-09-05 15:25:12

【问题描述】：

我正在从头开始编写我的第一个 API，并且有一个 /login 端点，在使用 bcrypt 验证用户密码时出错，但仅在使用 Postgres 作为我的数据库时，使用 SQLite3 时才能正常工作。

此外，任何以更好的方式构建我的模型或路线中的任何内容的帮助总是受欢迎的，这是我在 Flask / Python 中的第一个 API，所以我仍在学习。

提前致谢！

错误：

* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
[2021-06-22 12:06:14,415] ERROR in app: Exception on /api/v1/login [POST]
Traceback (most recent call last):
  File "C:\Users\x4c8\Projects\money_api\venv\lib\site-packages\flask\app.py", line 2070, in wsgi_app
response = self.full_dispatch_request()
File "C:\Users\x4c8\Projects\money_api\venv\lib\site-packages\flask\app.py", line 1515, in full_dispatch_request
rv = self.handle_user_exception(e)
File "C:\Users\x4c8\Projects\money_api\venv\lib\site-packages\flask\app.py", line 1513, in full_dispatch_request
rv = self.dispatch_request()
File "C:\Users\x4c8\Projects\money_api\venv\lib\site-packages\flask\app.py", line 1499, in dispatch_request
return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)
File "C:\Users\x4c8\Projects\money_api\routes.py", line 47, in token_get
check = user.verify_password(password)
File "C:\Users\x4c8\Projects\money_api\models.py", line 40, in verify_password
return bcrypt.checkpw(enc_pw, self.password_hash)
File "C:\Users\x4c8\Projects\money_api\venv\lib\site-packages\bcrypt\__init__.py", line 120, in checkpw
raise TypeError("Unicode-objects must be encoded before checking")
TypeError: Unicode-objects must be encoded before checking
127.0.0.1 - - [22/Jun/2021 12:06:14] "POST /api/v1/login HTTP/1.1" 500 -

Models.py 中的用户类：

class User(db.Model, Serializer):
__tablename__ = 'user'
id = db.Column(db.Integer, primary_key=True)
first_name = db.Column(db.String(15), unique=False, nullable=True)
last_name = db.Column(db.String(20), unique=False, nullable=True)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(255), unique=False, nullable=False)
country = db.Column(db.String(2), unique=False, nullable=True)
subscription_level = db.Column(db.Integer, default=0)
subscription_purchase_date = db.Column(db.DateTime(), unique=False, nullable=True)
last_login = db.Column(db.DateTime(), unique=False, default=datetime.utcnow)
modified_at = db.Column(db.DateTime(), unique=False, default=datetime.utcnow)
created_at = db.Column(db.DateTime(), unique=False, default=datetime.utcnow)

# relationships
portfolios = db.relationship('StockPortfolio', foreign_keys='StockPortfolio.fk_user', backref='user',
                             lazy='dynamic', cascade='all, delete-orphan')

@property
def password(self):
    raise AttributeError('password not readable')

@password.setter
def password(self, password):
    enc_pw = password.encode('utf-8')
    self.password_hash = bcrypt.hashpw(enc_pw, bcrypt.gensalt()).decode('utf-8')

def verify_password(self, password):
    enc_pw = password.encode('utf-8')
    return bcrypt.checkpw(enc_pw, self.password_hash)

def serialize(self):
    d = Serializer.serialize(self)
    del d['password_hash']
    del d['modified_at']
    del d['created_at']
    del d['last_login']
    return d

/从 routes.py 登录

# POST /login
@routes.route(api_v1 + 'login', methods=['POST'])
def token_get():
    if request.method == 'POST':
        body = request.get_json()

        # fail on missing params
        if body.get('email') is None:
            return jsonify(msg='email parameter is missing'), 422
        if body.get('password') is None:
            return jsonify(msg='password parameter is missing'), 422

        # fail on email not in use
        user = User.query.filter_by(email=body.get('email')).first()
        if user is None:
            return jsonify(msg='Email is not in use'), 404
        else:
            password = body.get('password')
            check = user.verify_password(password)

            if check:
                # record last login
                user.last_login = datetime.utcnow()

                # prep and return tokens
                access_token = create_access_token(identity=user.id)
                refresh_token = create_refresh_token(identity=user.id)
                return jsonify(msg='login successful', access_token=access_token, refresh_token=refresh_token), 200
            else:
                return jsonify(msg='incorrect email or password'), 409

【问题讨论】：

您可能应该设置密码哈希列db.Binary - 哈希是二进制对象，将它们存储为字符串只会让事情变得困难。我不久前研究了这个here，尽管与另一个错误有关。

db.BINARY 是所有可用的并导致：“sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedObject) 类型“二进制”不存在”。但是切换到 DB.LargeBinary 完全解决了这个问题 - 谢谢。如果您想将此作为答案，我会很乐意将其标记为关闭

这能回答你的问题吗？ Bcrypt Hash Returns TypeError("Unicode-objects must be encoded before hashing") and Invalid Salt

对建议的副本的答案比我有时间写的任何东西都要好 - 我建议接受它:-)

【参考方案1】：

您只需更改这部分代码即可将password_hash 转换为字节：

def verify_password(self, password):
    enc_pw = password.encode('utf-8')
    return bcrypt.checkpw(enc_pw, bytes(self.password_hash, 'utf-8'))