Authorization, Password 암호화

Author: msnodeve

Pipfile

@@ -17,6 +17,7 @@ flask-marshmallow = "*"
 flask-migrate = "*"
 pymysql = "*"
 flask-restplus = "*"
+pyjwt = "*"
 
 [requires]
 python_version = "3.7"

Pipfile.lock

@@ -5,8 +5,9 @@
 from flask_restplus import Api
 from app.users.views import API as users_api
 from app.posts.views import API as posts_api
+from app.api.auth_type import ACCESS_TOKEN, BASIC_AUTH
 
-REST_API = Api()
+REST_API = Api(authorizations={**ACCESS_TOKEN, **BASIC_AUTH})
 
 REST_API.add_namespace(users_api, '/user')
 REST_API.add_namespace(posts_api, '/post')

app/api/__init__.py

@@ -0,0 +1,38 @@
+import jwt
+from flask import request, Response
+from functools import wraps
+
+SECERET_KEY = "Secret Hellow"
+ACCESS_TOKEN = {
+    'Access Token': {
+        'type': 'apiKey',
+        'in': 'header',
+        'name': 'Authorization'
+    }
+}
+BASIC_AUTH = {
+	'Basic Auth': {
+        'type': 'basic',
+        'in': 'header',
+        'name': 'Authorization'
+    },
+}
+
+def confirm_token(f):
+    @wraps(f)
+    def decorated_function(*args, **kwargs):
+        access_token = request.headers['Authorization']
+        if access_token is not None:
+            try:
+                payload = jwt.decode(access_token, SECERET_KEY, "HS256")
+            except jwt.InvalidTokenError:
+                payload = None
+            if payload is None:
+                return Response(status=401)
+            user_id = payload["user_id"]
+            # 원하는 작업
+        else:
+            return Response(status=401)
+
+        return f(*args, **kwargs)
+    return decorated_function

app/api/auth_type.py

@@ -9,6 +9,7 @@
 from app.posts.models import Posts, PostsSchema
 from app.users.models import Users, UsersSchema
 from app.api.database import DB
+from app.api.auth_type import confirm_token, ACCESS_TOKEN, BASIC_AUTH
 
 API = Namespace('Posts', description="Post's REST API")
 POSTS_SCHEMA = PostsSchema()
@@ -41,14 +42,17 @@ def get(self):
         return make_response(body, code.value)
 
     @API.expect(post_field)
-    @API.doc('post')
+    @confirm_token
+    @API.doc('post', security=ACCESS_TOKEN)
     def post(self):
         args_ = self.parser.parse_args()
         post = Posts(author_id=args_['author_id'], title=args_['title'], body=args_['body'])
         return post.add(post, POSTS_SCHEMA)
 
 @API.route('/<int:reqno>')
 class PostItem(Resource):
+    @confirm_token
+    @API.doc('get', security=ACCESS_TOKEN)
     def get(self, reqno):
         try:
             post = DB.session.query(Posts).outerjoin(

app/posts/views.py

@@ -0,0 +1,16 @@
+import pytest
+from http import HTTPStatus
+from app.users.models import Users
+from app.posts.models import Posts
+from app.users.views import USERS_SCHEMA
+from app.posts.views import POSTS_SCHEMA
+from app.api.database import CRUD
+
+def test_add():
+    crud = CRUD()
+    try:
+        users = Users('test', 'test','test')
+        result = crud.add(users, USERS_SCHEMA)
+    except Exception as err:
+        print(err)
+    assert True

app/tests/test_crud.py

@@ -18,3 +18,5 @@ def test_practice():
         except ValueError as err:
             message = str(err)
             print(message)
+    result = worker.work()
+    assert result == "work"

app/tests/test_practice.py

@@ -1,59 +1,77 @@
 """
     User views file
 """
+import jwt
+import bcrypt
 from http import HTTPStatus
 from flask import jsonify, make_response
 from sqlalchemy.exc import SQLAlchemyError
 from flask_restplus import Namespace, Resource, reqparse, fields
 from app.users.models import Users, UsersSchema
 from app.api.database import DB
+from app.api.auth_type import SECERET_KEY
 
 API = Namespace('Users', description="User's RESTPlus - API")
 USERS_SCHEMA = UsersSchema()
 
+
 @API.route('s')
 class UsersAuth(Resource):
     parser = reqparse.RequestParser()
-    parser.add_argument('user_id', required=True, type=str, help="User's ID", location='json')
-    parser.add_argument('user_password', required=True, type=str, help="User's PW", location='json')
-    parser.add_argument('user_email', required=True, type=str, help="User's Email", location='json')
+    parser.add_argument('user_id', required=True, type=str,
+                        help="User's ID", location='json')
+    parser.add_argument('user_password', required=True,
+                        type=str, help="User's PW", location='json')
+    parser.add_argument('user_email', required=True, type=str,
+                        help="User's Email", location='json')
 
     users_field = API.model('userRegister', {
-        'user_id' : fields.String,
-        'user_password' : fields.String,
-        'user_email' : fields.String
+        'user_id': fields.String,
+        'user_password': fields.String,
+        'user_email': fields.String
     })
 
     @API.doc('post')
     @API.expect(users_field)
     def post(self):
         args_ = self.parser.parse_args()
-        user = Users(args_['user_id'], args_['user_password'], args_['user_email'])
+        password = args_['user_password']
+        hash_pw = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())
+        user = Users(args_['user_id'], hash_pw, args_['user_email'])
         return user.add(user, USERS_SCHEMA)
 
+
 @API.route('/auth')
 class UserAuth(Resource):
     parser = reqparse.RequestParser()
-    parser.add_argument('user_id', required=True, type=str, help="User's ID", location='json')
-    parser.add_argument('user_password', required=True, type=str, help="User's PW", location='json')
+    parser.add_argument('user_id', required=True, type=str,
+                        help="User's ID", location='json')
+    parser.add_argument('user_password', required=True,
+                        type=str, help="User's PW", location='json')
 
     user_login_field = API.model('userLogin', {
-        'user_id' : fields.String,
-        'user_password' : fields.String
+        'user_id': fields.String,
+        'user_password': fields.String
     })
 
     @API.doc('post')
     @API.expect(user_login_field)
     def post(self):
         args_ = self.parser.parse_args()
         try:
-            user = Users.query.filter(Users.user_id == args_['user_id'], Users.user_password  == args_['user_password']).first()
-            body = jsonify({'user_id' : user.user_id})
+            user = Users.query.filter(Users.user_id == args_['user_id']).first()
+            if bcrypt.checkpw(args_['user_password'].encode('utf-8'), user.user_password.encode('utf-8')):
+                # token 발급
+                payload = {
+                    'user_id' : user.user_id
+                }
+                token = jwt.encode(payload, SECERET_KEY, "HS256")
+                body = jsonify({'access_token': token.decode('utf-8'),'user': user.id})
             if user:
                 code = HTTPStatus.OK
             else:
                 code = HTTPStatus.NOT_FOUND
         except SQLAlchemyError as err:
-            body = jsonify({'message' : str(err)})
+            body = jsonify({'message': str(err)})
             code = HTTPStatus.INTERNAL_SERVER_ERROR
         return make_response(body, code.value)