token 발급

Author: msnodeve

Pipfile

@@ -18,6 +18,9 @@ flask-restplus = "*"
 flask-marshmallow = "*"
 marshmallow-jsonapi = "*"
 flask-script = "*"
+flask-basicauth = "*"
+pyjwt = "*"
+bcrypt = "*"
 
 [requires]
 python_version = "3.7"

Pipfile.lock

@@ -8,10 +8,11 @@
 from sqlalchemy.sql import text
 from flask_marshmallow import Marshmallow
 from app.api.database import DB
-from app.api import  REST_API
+from app.api import REST_API
 
 SQLALCHEMY_DATABASE_URI = \
     ("mysql+pymysql://{USER}:{PASSWORD}@{ADDR}:{PORT}/{NAME}?charset=utf8")
+    
 # 설명할 API에 대한 것
 MA = Marshmallow()
 def create_app() -> (Flask):

app/__init__.py

@@ -1,8 +1,10 @@
 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.add_namespace(users_api, '/users')
+REST_API = Api(authorizations={**ACCESS_TOKEN, **BASIC_AUTH})
+
+REST_API.add_namespace(users_api, '/user')
 REST_API.add_namespace(posts_api, '/posts')

app/api/__init__.py

@@ -0,0 +1,15 @@
+SECERET_KEY = "Hello"
+ACCESS_TOKEN = {
+    'Access Token': {
+        'type': 'apiKey',
+        'in': 'header',
+        'name': 'Authorization'
+    }
+}
+BASIC_AUTH = {
+	'Basic Auth': {
+        'type': 'basic',
+        'in': 'header',
+        'name': 'Authorization'
+    },
+}

app/api/auth_type.py

@@ -20,7 +20,6 @@
     'author_id': fields.Integer,
 })
 
-
 @API.route('')
 class Posts(Resource):
     parser = reqparse.RequestParser()

app/posts/views.py

@@ -0,0 +1,10 @@
+import pytest
+from app import create_app  # flask application factory
+
+@pytest.fixture(scope='session')
+def flask_app():
+    app = create_app()
+    app_context = app.app_context()
+    app_context.push()
+    yield app
+    app_context.pop()

app/tests/conftest.py

@@ -0,0 +1,35 @@
+from typing import Callable
+from functools import wraps
+from app import create_app
+from flask import current_app
+from flask import request
+from datetime import timezone, timedelta, datetime
+import jwt
+
+encoded_jwt = jwt.encode({'exp': datetime.utcnow()}, 'secret', algorithm='HS256')
+
+class UnAuthorizeError(Exception):
+    """ 인증 실패 """
+
+def requires_auth(f):
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        authorization = request.headers['Authorization']
+        if authorization != 'token': # db or jwt.decode
+            raise UnAuthorizeError("인증이 필요합니다")
+        return f(*args, **kwargs)
+    return decorated
+
+@requires_auth
+def something(*args, **kwargs):
+    # import pdb;pdb.set_trace()
+    return args, kwargs
+
+def test_decorator(flask_app):
+    with flask_app.test_request_context(headers={"Authorization":"adsf"}):
+        try:
+            result = something()
+            assert False
+        except UnAuthorizeError as err:
+            message = str(err)
+            print(message)

app/tests/test_auth_decorator.py

@@ -0,0 +1,7 @@
+from http import HTTPStatus
+from app import create_app
+
+def test_url(flask_app):
+    with flask_app.test_client() as client:
+        response = client.get('/users')
+        assert response.status_code == HTTPStatus.OK

app/tests/test_request.py

@@ -1,24 +1,34 @@
+
+import jwt
+import bcrypt
+import datetime
 from http import HTTPStatus
 from flask import jsonify
 from flask import make_response
+from flask import request
 from app.users.models import Users, UsersSchema
 from sqlalchemy.exc import SQLAlchemyError
 from flask_restplus import Api, Namespace, fields, reqparse, Resource
 from app.constants import STATUS_CODE
 from app.constants import GET, POST, PATCH, DELETE
 from app.api.database import DB
+from app.api.auth_type import BASIC_AUTH, ACCESS_TOKEN, SECERET_KEY
 
-API = Namespace('Users',description="User's REST API")
+API = Namespace('Users', description="User's REST API")
 
 USERS_SCHEMA = UsersSchema()
 
-@API.route('<int:user_id>')
+
+@API.route('/<int:user_id>')
 @API.param('user_id', 'The user identifier')
 class UserItem(Resource):
     parser = reqparse.RequestParser()
-    parser.add_argument('name', required=True, type=str, help="user's name", location='json')
-    parser.add_argument('email', required=True, type=str, help="user's email", location='json')
-    parser.add_argument('password', required=True, type=str, help="password", location='json')
+    parser.add_argument('name', required=True, type=str,
+                        help="user's name", location='json')
+    parser.add_argument('email', required=True, type=str,
+                        help="user's email", location='json')
+    parser.add_argument('password', required=True, type=str,
+                        help="password", location='json')
 
     user_field = API.model('Users', {
         'name': fields.String,
@@ -29,6 +39,7 @@ class UserItem(Resource):
     @API.doc(responses=GET)
     def get(self, user_id):
         user = Users.query.get_or_404(user_id)
+        Users.session.close()
         user = USERS_SCHEMA.dump(user).data
         return user
 
@@ -48,19 +59,22 @@ def delete(self, user_id):
         return response
 
 
-@API.route('')
+@API.route('s')
 class UsersList(Resource):
     parser = reqparse.RequestParser()
-    parser.add_argument('name', required=True, type=str, help="user's name", location='json')
-    parser.add_argument('email', required=True, type=str, help="user's email", location='json')
-    parser.add_argument('password', required=True, type=str, help="password", location='json')
+    parser.add_argument('name', required=True, type=str,
+                        help="user's name", location='json')
+    parser.add_argument('email', required=True, type=str,
+                        help="user's email", location='json')
+    parser.add_argument('password', required=True, type=str,
+                        help="password", location='json')
 
     user_field = API.model('Users', {
         'name': fields.String,
         'email': fields.String,
         'password': fields.String
     })
-
+    @API.doc(responses=GET, security=ACCESS_TOKEN)
     def get(self):
         users_query = Users.query.all()
         results = USERS_SCHEMA.dump(users_query, many=True).data
@@ -82,30 +96,40 @@ def post(self):
             code = HTTPStatus.INTERNAL_SERVER_ERROR
         return make_response(body, code.value)
 
+
 @API.route('/auth')
 class GetUser(Resource):
     parser = reqparse.RequestParser()
-    parser.add_argument('name', required=True, type=str, help="user's name", location='json')
-    parser.add_argument('password', required=True, type=str, help="user's password", location='json')
+    parser.add_argument('name', required=True, type=str,
+                        help="user's name", location='json')
+    parser.add_argument('password', required=True, type=str,
+                        help="user's password", location='json')
 
     user_field = API.model('Auth', {
         'name': fields.String,
         'password': fields.String
     })
 
+    @API.doc(responses=POST, security=ACCESS_TOKEN)
     @API.expect(user_field)
     def post(self):
         args = self.parser.parse_args()
         try:
-            user = Users.query.filter(Users.name == args['name'], Users.password == args['password']).first()
-            body = jsonify({"user" : USERS_SCHEMA.dump(user).data})
+            #, Users.password == args['password']
+            user = Users.query.filter(Users.name == args['name']).first()
+            if bcrypt.checkpw(args['password'].encode("UTF-8"), user.password.encode("UTF-8")):
+                #여기서 이제 토큰 발급해서 보내주기
+                payload = {
+                    "exp" : str(datetime.date.today())
+                }
+                token = jwt.encode(payload, SECERET_KEY, "HS256")
+                body = jsonify({"access_token": token.decode("UTF-8"), "user": USERS_SCHEMA.dump(user).data})
             if user:
                 code = HTTPStatus.OK
             else:
                 code = HTTPStatus.NOT_FOUND
         except SQLAlchemyError as err:
             message = str(err)
-            body = jsonify({"message" : message})
+            body = jsonify({"message": message})
             code = HTTPStatus.INTERNAL_SERVER_ERROR
         return make_response(body, code.value)
-