服务端：
#!/usr/bin/env python
import os
import time
from flask import Flask, abort, request, jsonify, g, url_for
from flask_sqlalchemy import SQLAlchemy
from flask_httpauth import HTTPBasicAuth
import jwt
from werkzeug.security import generate_password_hash, check_password_hash
# initialization
app = Flask(__name__)
app.config['SECRET_KEY'] = 'the quick brown fox jumps over the lazy dog'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db.sqlite'
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
# extensions
db = SQLAlchemy(app)
auth = HTTPBasicAuth()
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(32), index=True)
password_hash = db.Column(db.String(128))
def hash_password(self, password):
self.password_hash = generate_password_hash(password)
def verify_password(self, password):
return check_password_hash(self.password_hash, password)
def generate_auth_token(self, expires_in=600):
return jwt.encode(
{'id': self.id, 'exp': time.time() + expires_in},
app.config['SECRET_KEY'], algorithm='HS256')
@staticmethod
def verify_auth_token(token):
try:
data = jwt.decode(token, app.config['SECRET_KEY'],
algorithms=['HS256'])
except:
return
return User.query.get(data['id'])
@auth.verify_password
def verify_password(username_or_token, password):
# first try to authenticate by token
user = User.verify_auth_token(username_or_token)
if not user:
# try to authenticate with username/password
user = User.query.filter_by(username=username_or_token).first()
if not user or not user.verify_password(password):
return False
g.user = user
return True
@app.route('/api/users', methods=['POST'])
def new_user():
username = request.json.get('username')
password = request.json.get('password')
if username is None or password is None:
abort(400) # missing arguments
if User.query.filter_by(username=username).first() is not None:
abort(400) # existing user
user = User(username=username)
user.hash_password(password)
db.session.add(user)
db.session.commit()
return (jsonify({'username': user.username}), 201,
{'Location': url_for('get_user', id=user.id, _external=True)})
@app.route('/api/users/<int:id>')
def get_user(id):
user = User.query.get(id)
if not user:
abort(400)
return jsonify({'username': user.username})
@app.route('/api/token')
@auth.login_required
def get_auth_token():
token = g.user.generate_auth_token(600)
return jsonify({'token': token, 'duration': 600})
@app.route('/api/resource')
@auth.login_required
def get_resource():
return jsonify({'data': 'Hello, %s!' % g.user.username})
@app.route('/api/hello')
def sayhello():
return "hello,jcstone"
if __name__ == '__main__':
if not os.path.exists('db.sqlite'):
db.create_all()
app.run(debug=True)
客户端：
import requests, base64
import json
urltoken = "http://127.0.0.1:5000/api/token"
payload = {}
name = "jcstone"
pwd = "123456"
# auth = str(base64.b64encode(f'{name}:{pwd}'.encode('utf-8')), 'utf-8')
# headers = {
# 'Content-Type': 'application/json',
# 'Authorization': 'Basic %s' % auth
# }
# response = requests.request("GET", urltoken, headers=headers, data=payload)
# 获取token
headers = {'Content-Type': 'application/json'}
response = requests.get(urltoken, auth=(name, pwd),headers=headers, data=payload)
reptokens = json.loads(response.text.encode('utf8'))
print(reptokens["token"])
# 利用token验证访问restful api
urlapi = 'http://127.0.0.1:5000/api/resource'
token = reptokens["token"]
headers = {
# 将User-Agent改成 Chrome 浏览器
'User-Agent':
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.100 Safari/537.36',
}
response = requests.get(urlapi, auth=(token, ''), headers=headers)
print(response.text.encode('utf8'))
联系客服
