46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from functools import wraps
|
|
|
|
import jwt
|
|
from flask import current_app, g, jsonify, request
|
|
|
|
from app.models import User, db
|
|
|
|
|
|
def generate_token(user):
|
|
now = datetime.now(timezone.utc)
|
|
payload = {
|
|
"sub": str(user.id),
|
|
"username": user.username,
|
|
"iat": now,
|
|
"exp": now + timedelta(minutes=current_app.config["JWT_EXP_MINUTES"]),
|
|
}
|
|
return jwt.encode(payload, current_app.config["JWT_SECRET"], algorithm="HS256")
|
|
|
|
|
|
def token_required(f):
|
|
@wraps(f)
|
|
def wrapper(*args, **kwargs):
|
|
header = request.headers.get("Authorization", "")
|
|
parts = header.split()
|
|
if len(parts) != 2 or parts[0].lower() != "bearer":
|
|
return jsonify({"error": "missing or malformed Authorization header"}), 401
|
|
|
|
try:
|
|
payload = jwt.decode(
|
|
parts[1], current_app.config["JWT_SECRET"], algorithms=["HS256"]
|
|
)
|
|
except jwt.ExpiredSignatureError:
|
|
return jsonify({"error": "token expired"}), 401
|
|
except jwt.InvalidTokenError:
|
|
return jsonify({"error": "invalid token"}), 401
|
|
|
|
user = db.session.get(User, int(payload["sub"]))
|
|
if user is None:
|
|
return jsonify({"error": "user not found"}), 401
|
|
|
|
g.current_user = user
|
|
return f(*args, **kwargs)
|
|
|
|
return wrapper
|