opuser
This commit is contained in:
parent
252cc5b5d9
commit
0908b836d3
@ -4,7 +4,7 @@ import math
|
||||
import json
|
||||
|
||||
import jwt
|
||||
from flask import Blueprint, request, jsonify, make_response
|
||||
from flask import Blueprint, request, jsonify, make_response, g
|
||||
from bcrypt import hashpw, gensalt, checkpw
|
||||
from functools import wraps
|
||||
|
||||
@ -22,7 +22,7 @@ def role_required(permreq: list):
|
||||
def decorator(f):
|
||||
@wraps(f)
|
||||
def decorated_function(*args, **kwargs):
|
||||
# get data
|
||||
# get data 嘗試解碼jwt
|
||||
key = os.getenv("JWT_KEY", None)
|
||||
jwtsession = request.cookies.get("token", None)
|
||||
if jwtsession == None: return error("You do not have permission to view this page."), 401
|
||||
@ -32,13 +32,13 @@ def role_required(permreq: list):
|
||||
except jwt.exceptions.DecodeError: return error("Invalid token!"), 401
|
||||
if "id" not in jwtdata or "user" not in jwtdata: return error("Invalid token!"), 401
|
||||
|
||||
# db
|
||||
# db 驗證帳號是否正確
|
||||
table = pgclass.SQLuser
|
||||
with db.getsession() as session:
|
||||
res = session.query(table).filter(table.user == jwtdata["user"], table.id == jwtdata["id"]).first()
|
||||
if res is None: return error("You do not have permission to view this page."), 401
|
||||
|
||||
# permission check
|
||||
# permission check 確保用戶有此路徑要求的權限 並且權限名稱皆合法
|
||||
permissionList = list(set(res.permission))
|
||||
for p in permissionList: # 檢查用戶JWT是否有不合法的權限名稱
|
||||
if p not in PLIST_ROOT: return error("The user has invalid permission."), 402
|
||||
@ -46,19 +46,10 @@ def role_required(permreq: list):
|
||||
if p not in permissionList: return error("You do not have permission to view this page."), 402
|
||||
|
||||
# return
|
||||
g.opuser = res
|
||||
return f(*args, **kwargs)
|
||||
return decorated_function
|
||||
return decorator
|
||||
# get operator
|
||||
def getopuser(session, cookie):
|
||||
table = pgclass.SQLuser
|
||||
jwtsession = str(cookie)
|
||||
try: opuser = jwt.decode(jwt = jwtsession, key = os.getenv("JWT_KEY"), algorithms = ["HS256"])
|
||||
except jwt.exceptions.ExpiredSignatureError: return error("Token expired!"), 401
|
||||
except jwt.exceptions.DecodeError: return error("Invalid token!"), 401
|
||||
if "id" not in opuser or "user" not in opuser: return error("Invalid token!"), 401
|
||||
opuser = session.query(table).filter(table.user==opuser["user"],table.id==opuser["id"]).first()
|
||||
return opuser, None
|
||||
|
||||
# login
|
||||
@admin.route("/login", methods=["POST"])
|
||||
@ -96,9 +87,7 @@ def login():
|
||||
@admin.route("me", methods=["GET"])
|
||||
@role_required([])
|
||||
def user_me():
|
||||
with db.getsession() as session:
|
||||
opuser, err = getopuser(session, request.cookies.get("token"))
|
||||
if err is not None: return opuser, err
|
||||
opuser = g.opuser
|
||||
return jsonify({"id":opuser.id, "user":opuser.user, "permission":opuser.permission}), 200
|
||||
|
||||
####################
|
||||
@ -128,9 +117,7 @@ def user_del(id:int):
|
||||
table = pgclass.SQLuser
|
||||
|
||||
with db.getsession() as session:
|
||||
# user who requested
|
||||
opuser, err = getopuser(session, request.cookies.get("token"))
|
||||
if err is not None: return opuser, err
|
||||
opuser = g.opuser # user who requested
|
||||
|
||||
# check root
|
||||
tguser = session.query(table).filter(table.id==int(id)).first()
|
||||
@ -151,8 +138,7 @@ def user_add():
|
||||
table = pgclass.SQLuser
|
||||
with db.getsession() as session:
|
||||
# user who requested
|
||||
opuser, err = getopuser(session, request.cookies.get("token"))
|
||||
if err is not None: return opuser, err
|
||||
opuser = g.opuser
|
||||
|
||||
# payload
|
||||
if "username" not in request.json or "password" not in request.json or \
|
||||
@ -206,9 +192,7 @@ def article_read(id:int):
|
||||
@admin.route("/article/<int:id>", methods=["DELETE"])
|
||||
@role_required(["article.del"])
|
||||
def article_del(id:int):
|
||||
with db.getsession() as session:
|
||||
opuser, err = getopuser(session, request.cookies.get("token"))
|
||||
if err is not None: return opuser, err
|
||||
opuser = g.opuser
|
||||
|
||||
result, code = solo_article_remover("admin", id=id)
|
||||
if "error" in result: return jsonify(result), code
|
||||
@ -251,10 +235,7 @@ def setting_get():
|
||||
@admin.route("/setting", methods=["POST"])
|
||||
@role_required(["setting.edit"])
|
||||
def setting_edit():
|
||||
with db.getsession() as session:
|
||||
opuser, err = getopuser(session, request.cookies.get("token"))
|
||||
if err is not None: return opuser, err
|
||||
opuser = opuser.user
|
||||
opuser = g.opuser
|
||||
|
||||
req = request.json
|
||||
d = None
|
||||
|
@ -2,7 +2,7 @@ import time
|
||||
import hashlib
|
||||
|
||||
import magic
|
||||
from flask import Blueprint, current_app, request, jsonify
|
||||
from flask import Blueprint, request, jsonify
|
||||
from google.protobuf.message import DecodeError
|
||||
|
||||
from utils import logger, pgclass, setting_loader
|
||||
|
Loading…
Reference in New Issue
Block a user