JWT 토큰을 사용해서, 현재 사용자의 인증을 체크 해 주자.

JSON Web Token (JWT) 은 웹표준 (RFC 7519) 으로서 두 개체에서 JSON 객체를 사용하여 가볍고 자가수용적인 (self-contained) 방식으로 정보를 안전성 있게 전달해줍니다.

• reference from - https://velopert.com/2389

front-back의 소스는 여기 에 올려놓았다

JWT Auth Token

Flask서버에서 Auth Token 발행하기

structure

backend
├── app.py
├── back.dev.Dockerfile
├── my_model
│   ├── __init__.py
│   ├── tweet_model.py
│   └── user_model.py
├── my_provider
│   ├── __init__.py
│   ├── baseball_scrapper.py
│   └── movie_scrapper.py
├── my_router
│   ├── __init__.py
│   └── api
│       ├── __init__.py
│       ├── auth.py
│       └── tweet.py
├── my_util
│   ├── __init__.py
│   ├── auth_util.py
│   └── my_logger.py
└── requirements.txt

구조를 보면 알겠지만, backend의 구조가 많이 변경되었다.

기존의 app.py에서 작성 되어있던 라우터들을 jwt토큰을 사용 한 버전 2의 api를 사용하기 위해서 my_router로 빼내었다.

이번에는 'tweet'라는 포스트를 추가하는 동작을 해보려고 한다.

먼저 필요한 모듈을 다운받고, 데이터 저장에 필요한 db를 생성 해 주도록 하겠다.

Install pyjwt

(venv)$ pip3 install PyJWT

create table

create table tweet(
	id int(5) not null auto_increment,
	title varchar(100) not null,
	words text not null,
	creator varchar(20) not null,
	created_at datetime default current_timestamp,
	updated_at datetime default current_timestamp on update current_timestamp,
	PRIMARY KEY (id,creator) 
)

create flask model(backend/my_model/tweet_model.py)

# coding: utf-8
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text
from sqlalchemy.schema import FetchedValue
from sqlalchemy.orm import relationship
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Tweet(db.Model):
    __tablename__ = 'tweet'

    id = db.Column(db.Integer, primary_key=True, nullable=False, autoincrement=True)
    title = db.Column(db.String(100), nullable=False)
    words = db.Column(db.Text, nullable=False)
    creator = db.Column(db.String(100), nullable=False, primary_key=True)
    created_at = db.Column(db.DateTime, server_default=db.FetchedValue())
    updated_at = db.Column(db.DateTime, server_default=db.FetchedValue())

    def to_dict(self):
        return dict(id=self.id,
                    title=self.title,
                    words=self.words,
                    creator=self.creator,
                    created_at=self.created_at,
                    updated_at=self.updated_at)

이제 데이터를 생성/조회 할 수 있도록 세팅이 되었다.

이제 권한과 관련된 작업을 해 주도록 하겠다.

backend/my_util/auth_util.py

import jwt
from flask import request, jsonify, current_app
from my_model.user_model import User
from my_util.my_logger import my_logger
from functools import wraps

def token_required(f):
    @wraps(f)
    def _verify(*args, **kwargs):
        auth_headers = request.headers.get("Authorization", '').split()

        invalid_msg = {
            'message': "Invalid Token. Registeration / authentication required",
            'authenticated': False
        }

        expired_msg = {
            'message': "expired Token. Reauthentication required",
            'authenticated': False
        }

        if len(auth_headers) != 2:
            return jsonify(invalid_msg), 401

        try:
            token = auth_headers[1]
            data = jwt.decode(token, 'qwersdaiofjhoqwihlzxcjvjl')
            parse_name = data['sub']
            user = User.query.filter_by(username=parse_name).first()
            if not user:
                raise RuntimeError('User not found')
            return f(user, *args, **kwargs)
        except jwt.ExpiredSignatureError:
            return jsonify(expired_msg), 401  # 401 is Unauthorized HTTP status code
        except (jwt.InvalidTokenError, Exception) as e:
            my_logger.error(e)
            return jsonify(invalid_msg), 401

    return _verify

토큰을 생성하고, 확인하는 작업까지 완료되었다.

이제 어떤 라우터에서든, 위의 모듈을 import 하고 데코레이터 형태로 사용하면 인증이 필요한 절차가 된다.

backend/my_router/api/auth.py

from flask import Blueprint, request, jsonify, current_app
from my_util.my_logger import my_logger
from my_model import user_model

import jwt
import datetime

auth_route = Blueprint('auth_route', __name__)

@auth_route.route('/logout', methods=['POST'])
def logout():
    return jsonify("hi!")

@auth_route.route('/get_user', methods=['GET'])
def get_login_user():
    return jsonify("hi!")

@auth_route.route('/signup', methods=['POST'])
def user_register():
    # 로그
    my_logger.info("user register")
    data = request.get_json()
    db = user_model.db

    user_data = user_model.User.query.filter_by(username=data.get('username')).first()
    if user_data is not None:
        my_logger.error("Username is Already exist")
        return {"success": "username is already exist"}

    user = user_model.User(**data)
    user.has_password()
    db.session.add(user)
    db.session.commit()

    my_logger.info("user Save Success")

    return jsonify(user.to_dict()), 200

@auth_route.route('/login', methods=['POST'])
def login():
    my_logger.info("User Login")

    data = request.get_json()
    user_data = user_model.User.query.filter_by(username=data.get('username')).first()

    if user_data is not None:
        auth = user_data.check_password(data.get("userpwd"))

        if not auth:
            my_logger.error("Authentication valid error")
            return jsonify({'message': 'Authentication valid error', "authenticated": False}), 401

        my_logger.info("User Authentication token setting")
        token = jwt.encode({
            'sub': user_data.username,
            'iat': datetime.datetime.utcnow(),
            'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=30)
        }, 'qwersdaiofjhoqwihlzxcjvjl')

        my_logger.info("Set up Success")
        my_logger.debug(token.decode('UTF-8'))
        return jsonify({"token": token.decode('UTF-8')})
    else:
        my_logger.error("User Does Not Exist")
        return jsonify({'message': 'User Does Not Exist', "authenticated": False}), 401

backend/my_router/api/tweet.py

from flask import Blueprint, request, jsonify, current_app
from my_util.my_logger import my_logger
from my_model import user_model
from my_model import tweet_model
from my_util.auth_util import token_required

import jwt
import datetime

tweet_route = Blueprint('tweet_route', __name__)

@tweet_route.route('/tweet', methods=["GET"])
@token_required
def tweet_list(current_user):
    my_logger.info("get Tweet List")
    tweet_list = tweet_model.Tweet.query.all()
    return jsonify([t.to_dict() for t in tweet_list])

@tweet_route.route('/tweet', methods=["POST"])
@token_required
def create_tweet(current_user):
    my_logger.info("Tweet Post!")
    try:
        data = request.get_json()
        title = data.get("title")
        words = data.get("words")
        created_at = datetime.datetime.utcnow()
        updated_at = datetime.datetime.utcnow()

        db = tweet_model.db
        tweet = tweet_model.Tweet(
            title=title,
            words=words,
            creator=current_user.username,
            created_at=created_at,
            updated_at=updated_at
        )
        db.session.add(tweet)
        db.session.commit()
        return jsonify(tweet.to_dict()), 200
    except Exception as e:
        my_logger.error(e)
        return jsonify({'message': 'fail to save tweet'}), 200

backend는 완성했으니, 이제 Vue를 작업 해 보도록 하자...! 너무 어렵다..