数据库自动化运维2-Restful API JWT and Redis authorization

Posted 2023-02-26 renfengjun

上一篇文章提供了一个简单的Restful API的Demo，今天来完善一下它的验证机制。

这里使用redis存放关于用户的密码以及roles等信息，并且根据roles以及客户端请求的endpoint来完成权限验证。

redis里面存放的信息：

[root@adirfj ~]# redis-cli
127.0.0.1:6379> AUTH "password"
OK
127.0.0.1:6379> SMEMBERS roles:userrfj
 1) "REFRESHPDB"
 2) "ALL"
 3) "INSTANCELIST"
 4) "PDBSTATUS"
 5) "GETPDB"
 6) "UPDATEPDB"
 7) "ACLSLIST"
 8) "TEMPLATELIST,PDBLIST"
 9) "ACLLIST"
10) "PDBLIST"
11) "RFJAPIS"
12) "LISTACL"
13) "PDBCREATE"
14) "DELETERESTOREPOINT"
15) "CREATERESTOREPOINT"
16) "CREATEACL"
17) "DROPPDB"
18) "CREATEPDB"
19) "SNAPCLONE"
20) "DBA"
21) "GETACLS"
22) "SYSINFO"
23) "HEALTH"
24) "LISTACLS"
25) "DELETEACL"
26) "MIGRATEPDB"
27) "GETRESTOREPOINTS"
28) "GETCDB"

flask启动文件jwttoken.py代码

jwttoken.py
from flask import Flask,request, jsonify, json
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from itsdangerous import SignatureExpired, BadSignature, BadData
import time
import redis
import os


import cx_Oracle
from orcltest import util
from orcltest import myora
app = Flask(__name__)
#These things should be setup on os envirement variables before flask run
#like
#export SECRET_KEY='RFJNIUBI'
#How to get
#app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY')
app.config['SECRET_KEY'] = 'RFJBUILDTHISHEISCOOL'
app.config['AUTH_SALT'] = 'RFJBUILDTHISHEISCOO'
app.config['ORAPWD'] = 'oracle123'
app.config['ORAUSR'] = 'c##dex'
app.config['db'] = '123456'
app.config['REDISPWD'] = 'password'
@app.route('/')
def index():
    return '<h1> Hello world </h1>'
@app.route('/index')
def hello():
    return '<h2> abc orz </h2>'
@app.route('/aut', methods=['POST'])
def aut():
    data = request.get_json()
    a = data['userId']
    b = data['passwd']
    s = Serializer(app.config['SECRET_KEY'], expires_in = 180)
    s = Serializer(
            secret_key=app.config['SECRET_KEY'],
            salt=app.config['AUTH_SALT'],
            expires_in=180)
    timestamp = time.time()
    b = getrole()
    b = list(b)
    print b
    sf = s.dumps(
            'user_id': a,
             'user_role': b,
             'iat': timestamp).decode('utf-8')
    return jsonify( 'token':sf)
@app.route('/pdb/<string:pdbname>')
def pdbstatus(pdbname):
    usr=app.config['ORAUSR']
    passwd=app.config['ORAPWD']
    try:
        new_ora = myora.myOra(usr=usr, passwd=passwd, db=pdbname, isdba=True)
    except Exception as e :
        app.logger.warning(str(e).strip())
        return quick(404,pdbname+' does not exists')
    rows = util.pdb_status(pdbname, new_ora)[0][0]
    return quick(200,'Status:'+rows)
def quick(http_code , message):
    response = jsonify('message': message)
    response.status_code =http_code
    return response
@app.before_request
def before_request():
    endp = request.endpoint
    if endp == 'aut' :
#        get env print os.environ.get('FLASK_APP')
        return aut()
    auth_fail_msg = 'Auth failed'
    try:
        token = request.headers.get('Authentication').decode('utf-8')
        token = token.replace('Bearer ', '')
        print token
    except Exception as e :
        return quick(401, auth_fail_msg)
    s = Serializer(
          secret_key=app.config['SECRET_KEY'],
          salt=app.config['AUTH_SALT'])
    try:
        data = s.loads(token)
            # token decoding faild
            # if it happend a plenty of times, there might be someone
            # trying to attact your server, so it should be a warning.
    except SignatureExpired:
        msg = 'token expired'
        app.logger.warning(msg)
        return quick(401,auth_fail_msg)
    except BadSignature, e:
        encoded_payload = e.payload
        if encoded_payload is not None:
            try:
                s.load_payload(encoded_payload)
            except BadData:
               # the token is tampered.
                msg = 'token tampered'
                app.logger.warning(msg)
                return quick(401,auth_fail_msg)
        msg = 'badSignature of token'
        app.logger.warning(msg)
        return quick(401,auth_fail_msg)
    except:
        msg = 'wrong token with unknown reason'
        app.logger.warning(msg)
        return quick(401,msg)
    if ('user_id' not in data) or ('user_role' not in data):
        msg = 'illegal payload inside'
        app.logger.warning(msg)
        return quick(401,auth_fail_msg)
#    msg = 'user(' + data['user_id'] + ') '+ data['user_role']+'logged in by token.'
#        app.logger.info(msg)
    userId = data['user_id']
    roleId = data['user_role']
    endp = request.endpoint.upper()
    if not endp in roleId :
         msg = 'you don''t have access to this API'
         app.logger.warning(msg)
         return quick(401, auth_fail_msg)
    #return quick(200,','.join(roleId))

def getrole():
    r=redis.Redis(host='localhost', port=6379, db=0, password=app.config['REDISPWD'] )
    s=r.smembers('roles:userrfj')
    return s

使用效果：

[root@adirfj flaskspace]# token=$(curl -s -X POST -H "Content-Type: application/json" http://127.0.0.1:5000/aut -d "$APICREDS" |jq -r '.token')
[root@adirfj flaskspace]# curl -s -X GET -H "Content-Type: application/json" -H "Authentication: $token" http://127.0.0.1:5000/pdb/DB000111

  "message": "Status:READ WRITE"

这里调用的endpoint是PDBSTATUS，所以如果我们删除相关用户的roles，则调用情况为：

127.0.0.1:6379> SREM  roles:gb-adi-api "PDBSTATUS"
(integer) 1
[root@adirfj flaskspace]# token=$(curl -s -X POST -H "Content-Type: application/json" http://127.0.0.1:5000/aut -d "$APICREDS" |jq -r '.token')
[root@adirfj flaskspace]# curl -s -X GET -H "Content-Type: application/json" -H "Authentication: $token" http://127.0.0.1:5000/pdb/DB000111

  "message": "Auth failed"

后台日志：
[2019-06-20 09:11:19,833] WARNING in jwttoken: you dont have access to this API
127.0.0.1 - - [20/Jun/2019 09:11:19] "GET /pdb/DB000111 HTTP/1.1" 401 -

好了，今天先到这里，后面将更新对于postgresql 的db helper。