1.環境與認證 · 使用flask從零構建自動化運維平臺

[TOC] # 使用flask從零構建自動化運維平臺 ## 安裝事先準備好python環境 ``` pip install flask ``` ## 開發ide 推薦使用pycharm ![](https://box.kancloud.cn/51f6398a74ef18cd6397610d78011b2f_1680x1050.png) ## 開發思路擺在面前的兩條路，Django那種MVC（Model View Controller），另一種就是比較流行的動靜分離，也就是前端和后端分開開發。我個人傾向于前后分離，前后分離又有一個選擇是采用RPC（Remote Procedure Call）or RestFul（(Representation State Transfer) 兩種方式各有優點。我個人傾向于接口集中化，就是一個地址來處理所有和前后端交互。總合幾方面的考慮最后還是使用了flask框架 ## 使用到的flask拓展 | 拓展名 | 地址 | 描述 | | ------------ | ------------ | ------------ | | flask-jsonrpc | [git](http://github.com/cenobites/flask-jsonrpc)| jsonrpc-flask拓展 | | flask_sqlalchemy| | flask ORM| |flask-marshmallow | |將sqlalchemy查詢的結果轉成json| |flask_script | |flask的拓展命令行插件| |flask_migrate | |數據庫同步插件| ## 設計一個最常用的helloworld接口 ```python from flask import Flask from flask_jsonrpc import JSONRPC app = Flask(__name__) # 啟用一個web界面的api調試 jsonrpc = JSONRPC(app, '/api', enable_web_browsable_api=True) # 增加一個方法，前端可以調用 @jsonrpc.method('App.index') def index(): return u'Hello World!' if __name__ == '__main__': app.run(port=2001, debug=True) ``` 前端調用 ``` POST /api { "date": "Wed, 21 Mar 2018 02:23:22 GMT", "server": "Werkzeug/0.14.1 Python/3.6.2", "content-length": "101", "content-type": "application/json", "data": { "jsonrpc": "2.0", "method": "App.index", "params": [], "id": "3f46e50f-b46b-49e3-b16b-56af7ab21867" } } ``` 后端返回 ``` HTTP 200 { "id": "3f46e50f-b46b-49e3-b16b-56af7ab21867", "jsonrpc": "2.0", "result": "Hello World!" } ``` ## 添加驗證就是身份認證通過后才可以請求到資源，這樣權限也可以得到控制,這里使用的是目前比較熱門的token認證方式。 token本身是經過加密后的字符串，攜帶了一部分加密的內容和有效期，在這個有效期，使用這串字符串就能訪問到資源。過了之后就不能訪問了。需要導入的包和一些配置 ```python from flask import Flask from flask_jsonrpc import JSONRPC from flask_sqlalchemy import SQLAlchemy from itsdangerous import TimedJSONWebSignatureSerializer as Serializer import werkzeug app = Flask(__name__) jsonrpc = JSONRPC(app, '/api', enable_web_browsable_api=True) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True app.config['SECRET_KEY'] = '1q2w3e' db = SQLAlchemy(app) ``` ### 1. 使用ORM創建一個用戶表 [flask-sqlalchemy教程](http://docs.jinkan.org/docs/flask-sqlalchemy/quickstart.html "flask-sqlalchemy教程") 定義數據模型 ```python class User(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True) password_hash = db.Column(db.String(164)) def password(self, password): """ 設置密碼hash值 """ self.password_hash = werkzeug.security.generate_password_hash(password) def verify_password(self, password): """ 將用戶輸入的密碼明文與數據庫比對 """ return werkzeug.security.check_password_hash(password) def __init__(self, username): self.username = username def __repr__(self): return '<User %r>' % self.username ``` 使用模型創建表結構 ``` from SmartOps import db db.create_all() ``` ### 2.設計用戶登錄接口前端： ``` POST /api { "date": "Wed, 21 Mar 2018 06:24:12 GMT", "server": "Werkzeug/0.14.1 Python/3.6.2", "content-length": "158", "content-type": "application/json", "data": { "jsonrpc": "2.0", "method": "user.register", "params": { "username": "test", "password": "123456" }, "id": "cde5749c-de2f-46ea-b9c5-824cf1f3fc92" } } ``` 后臺： ``` HTTP 200 { "id": "cde5749c-de2f-46ea-b9c5-824cf1f3fc92", "jsonrpc": "2.0", "result": { "message": "用戶已存在", "status": 1 } } ``` > 0:代表注冊成功 1:代表注冊失敗后臺代碼： ```python @jsonrpc.method('user.register(username=str,password=str)') def user_register(username, password): if not User.query.filter_by(username=username).first(): user = User(username=username) user.password(password) db.session.add(user) db.session.commit() return {'status': 0, 'message': u'注冊成功'} else: return {'status': 1, 'message': u'用戶已存在'} ``` ### 3. 用戶注冊完成需要登錄前端： ``` POST /api { "date": "Wed, 21 Mar 2018 06:27:48 GMT", "server": "Werkzeug/0.14.1 Python/3.6.2", "content-length": "1148", "content-type": "application/json", "data": { "jsonrpc": "2.0", "method": "user.verify", "params": { "username": "test", "password": "123456" }, "id": "36d8d630-4092-4add-abfc-55a257dae7d9" } } ``` 后臺： ``` HTTP 200 { "id": "945307c4-8009-49ed-ab43-96d29ae37294", "jsonrpc": "2.0", "result": { "message": "歡迎test2", "status": 0, "token": "eyJhbGciOiJIUzI1NiIsImlhdCI6MTUyMTYxNTYyNywiZXhwIjoxNTIxNjE2MjI3fQ.eyJpZCI6M30.fw8tSBcDfITEgG5mHeMpyFio821jlmVQgmlXlZxDadI" } } ``` 后臺代碼： ```python @jsonrpc.method('user.verify(username=str,password=str)') def user_verify(username, password): user = User.query.filter_by(username=username).first() if not user: return {'status': 1, 'message': u'用戶名不存在'} if user.verify_password(password): token = user.generate_auth_token() return {'status': 0, 'message': u'歡迎%s' % username, 'token': token} return {'status': 1, 'message': u'密碼錯誤'} ``` ### 4. token的生成與驗證生成函數 ```python class User(db.Model): ... def generate_auth_token(self, expiration=600): s = Serializer(app.config['SECRET_KEY'], expires_in=expiration) return bytes.decode(s.dumps({'id': self.id})) @staticmethod def verify_auth_token(token): s = Serializer(app.config['SECRET_KEY']) try: data = s.loads(token) except SignatureExpired: return None # valid token, but expired except BadSignature: return None # invalid token user = User.query.get(data['id']) return user ... ``` 這里需要修改一下JSONRPC的源碼，默認的是通過認證用戶名和密碼，用戶名和密碼一直經過互聯網傳輸很不安全，給修改成通過token驗證修改method方法 ```python def method(self, name, authenticated=False, safe=False, validate=False, **options): def decorator(f): arg_names = getargspec(f)[0] X = {'name': name, 'arg_names': arg_names} if authenticated: # TODO: this is an assumption # X['arg_names'] = ['username', 'password'] + X['arg_names'] X['arg_names'] = ['token'] + X['arg_names'] X['name'] = _inject_args(X['name'], ('String', 'String')) _f = self.auth_backend(f, authenticated) else: _f = f method, arg_types, return_type = _parse_sig(X['name'], X['arg_names'], validate) _f.json_args = X['arg_names'] _f.json_arg_types = arg_types _f.json_return_type = return_type _f.json_method = method _f.json_safe = safe _f.json_sig = X['name'] _f.json_validate = validate self.site.register(method, _f) return _f return decorator ``` 然后再新寫一下認證接口 ```python def authenticate(f, f_check_auth): @wraps(f) def _f(*args, **kwargs): is_auth = False try: creds = args[:2] is_auth = f_check_auth(creds[0], creds[1]) if is_auth: args = args[2:] except IndexError: print(kwargs) if 'token' in kwargs: is_auth = f_check_auth(kwargs['token']) if is_auth: kwargs.pop('token') else: raise InvalidParamsError('Authenticated methods require at least ' '[token] or {token: } arguments') if not is_auth: raise InvalidCredentialsError() return f(*args, **kwargs) return _f ``` 修改一下初始化JSONRPC讓他能識別到咱們自己寫的認證接口 ```python jsonrpc = JSONRPC(app, '/api', enable_web_browsable_api=True, auth_backend=authenticate) ``` ### 5.認證這里算是完成了。前端使用token認證就能請求到資源了前端： ``` POST /api { "date": "Wed, 21 Mar 2018 07:29:49 GMT", "server": "Werkzeug/0.14.1 Python/3.6.2", "content-length": "100", "content-type": "application/json", "data": { "jsonrpc": "2.0", "method": "App.hello", "params": { "token": "eyJhbGciOiJIUzI1NiIsImlhdCI6MTUyMTYxNzMzMywiZXhwIjoxNTIxNjE3OTMzfQ.eyJpZCI6M30.pswj1Sbny8EM2u8T01s7gizS02LQ-RS2PSvme2jQQLs", "name": "jack" }, "id": "61a746df-21c5-485c-9aa5-b7c9b3a938e6" } } ``` 后端： ``` HTTP 200 { "id": "61a746df-21c5-485c-9aa5-b7c9b3a938e6", "jsonrpc": "2.0", "result": "Hello jack!" } ``` 后臺測試代碼： ```python @jsonrpc.method('App.hello(name=str)', authenticated=check_auth) def index(name): return u'Hello %s!' % name ``` ## 附上完整代碼 ```python from flask import Flask from flask_jsonrpc import JSONRPC from flask_sqlalchemy import SQLAlchemy from itsdangerous import TimedJSONWebSignatureSerializer as Serializer import werkzeug app = Flask(__name__) def authenticate(f, f_check_auth): @wraps(f) def _f(*args, **kwargs): is_auth = False try: creds = args[:2] is_auth = f_check_auth(creds[0], creds[1]) if is_auth: args = args[2:] except IndexError: if 'token' in kwargs: is_auth = f_check_auth(kwargs['token']) if is_auth: kwargs.pop('token') else: raise InvalidParamsError('Authenticated methods require at least ' '[token] or {token: } arguments') if not is_auth: raise InvalidCredentialsError() return f(*args, **kwargs) return _f jsonrpc = JSONRPC(app, '/api', enable_web_browsable_api=True, auth_backend=authenticate) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True app.config['SECRET_KEY'] = '1q2w3e' app.debug = True db = SQLAlchemy(app) from functools import wraps class User(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True) password_hash = db.Column(db.String(164)) def password(self, password): """ 設置密碼hash值 """ self.password_hash = werkzeug.security.generate_password_hash(password) def verify_password(self, password): """ 將用戶輸入的密碼明文與數據庫比對 """ print(self.username) if self.password_hash: return werkzeug.security.check_password_hash(self.password_hash, password) return None def generate_auth_token(self, expiration=600): s = Serializer(app.config['SECRET_KEY'], expires_in=expiration) return bytes.decode(s.dumps({'id': self.id})) @staticmethod def verify_auth_token(token): s = Serializer(app.config['SECRET_KEY']) try: data = s.loads(token) except SignatureExpired: return None # valid token, but expired except BadSignature: return None # invalid token user = User.query.get(data['id']) return user def __init__(self, username): self.username = username def __repr__(self): return '<User %r>' % self.username @jsonrpc.method('user.register(username=str,password=str)') def user_register(username, password): if not User.query.filter_by(username=username).first(): user = User(username=username) user.password(password) db.session.add(user) db.session.commit() return {'status': 0, 'message': u'注冊成功'} else: return {'status': 1, 'message': u'用戶已存在'} @jsonrpc.method('user.verify(username=str,password=str)') def user_verify(username, password): user = User.query.filter_by(username=username).first() if not user: return {'status': 1, 'message': u'用戶名不存在'} if user.verify_password(password): token = user.generate_auth_token() return {'status': 0, 'message': u'歡迎%s' % username, 'token': token} return {'status': 1, 'message': u'密碼錯誤'} def check_auth(token): #啟用debug模式不需要進行token認證 if app.debug: return True user = User.verify_auth_token(token) if user: return True return False @jsonrpc.method('App.hello(name=str)', authenticated=check_auth) def index(name): return u'Hello %s!' % name if __name__ == '__main__': app.run(port=2001) ```