Migrate to Flask 2

app/models.py

@@ -1,12 +1,12 @@
-from app import db, login, mail, socketio
+from app import db, hashids, login, mail, socketio
 from app.converters.vrt import normalize_vrt_file
 from app.email import create_message
+from authlib.jose import jwt, JoseError
 from datetime import datetime, timedelta
 from enum import Enum, IntEnum
 from flask import current_app, url_for
 from flask_hashids import HashidMixin
 from flask_login import UserMixin
-from itsdangerous import BadSignature, TimedJSONWebSignatureSerializer
 from time import sleep
 from tqdm import tqdm
 from werkzeug.security import generate_password_hash, check_password_hash
@@ -15,6 +15,7 @@ import json
 import os
 import requests
 import shutil
+import time
 import xml.etree.ElementTree as ET
 import yaml
 
@@ -292,10 +293,14 @@ class User(HashidMixin, UserMixin, db.Model):
         return self.role.has_permission(permission)
 
     def confirm(self, token):
-        s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
+        # s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
+        # try:
+        #     data = s.loads(token.encode('utf-8'))
+        # except BadSignature:
+        #     return False
         try:
-            data = s.loads(token.encode('utf-8'))
-        except BadSignature:
+            data = jwt.decode(token, current_app.config['SECRET_KEY'])
+        except JoseError:
             return False
         if data.get('confirm') != self.hashid:
             return False
@@ -308,14 +313,21 @@ class User(HashidMixin, UserMixin, db.Model):
         db.session.delete(self)
 
     def generate_confirmation_token(self, expiration=3600):
-        s = TimedJSONWebSignatureSerializer(
-            current_app.config['SECRET_KEY'], expiration)
-        return s.dumps({'confirm': self.hashid}).decode('utf-8')
+        # s = TimedJSONWebSignatureSerializer(
+        #     current_app.config['SECRET_KEY'], expiration)
+        # return s.dumps({'confirm': self.hashid}).decode('utf-8')
+        header = {'alg': 'HS256', 'exp': int(time.time()) + expiration}
+        payload = {'confirm': self.hashid}
+        return jwt.encode(header, payload, current_app.config['SECRET_KEY'])
+
 
     def generate_reset_token(self, expiration=3600):
-        s = TimedJSONWebSignatureSerializer(
-            current_app.config['SECRET_KEY'], expiration)
-        return s.dumps({'reset': self.hashid}).decode('utf-8')
+        # s = TimedJSONWebSignatureSerializer(
+        #     current_app.config['SECRET_KEY'], expiration)
+        # return s.dumps({'reset': self.hashid}).decode('utf-8')
+        header = {'alg': 'HS256', 'exp': int(time.time()) + expiration}
+        payload = {'reset': self.hashid}
+        return jwt.encode(header, payload, current_app.config['SECRET_KEY'])
 
     def get_token(self, expires_in=3600):
         now = datetime.utcnow()
@@ -398,12 +410,26 @@ class User(HashidMixin, UserMixin, db.Model):
 
     @staticmethod
     def reset_password(token, new_password):
-        s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
+        # s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
+        # try:
+        #     data = s.loads(token.encode('utf-8'))
+        # except BadSignature:
+        #     return False
+        # user = User.query.get(data.get('reset'))
+        # if user is None:
+        #     return False
+        # user.password = new_password
+        # db.session.add(user)
+        # return True
         try:
-            data = s.loads(token.encode('utf-8'))
-        except BadSignature:
+            data = jwt.decode(token, current_app.config['SECRET_KEY'])
+        except JoseError:
             return False
-        user = User.query.get(data.get('reset'))
+        user_hashid = data.get('reset')
+        if user_hashid is None:
+            return False
+        user_id = hashids.decode(user_hashid)
+        user = User.query.get(user_id)
         if user is None:
             return False
         user.password = new_password