JSON Web トークンを使用して Flask REST API を保護する方法?

JSON Web トークンを使用して REST API を保護し、ユーザーやサードパーティのアプリケーションによる REST API の悪用を防ぐ方法を学びましょう。

SQLite を使用してデータベース サービスを構築し、ユーザーが POST や PUT などの HTTP メソッドを使用して REST API 経由でデータベース サービスにアクセスできるようにします。

さらに、REST API を保護する方法として、ダイジェスト認証や基本認証ではなく JSON Web トークンが適している理由を理解します。先に進む前に、JSON Web トークン、REST API、および Flask フレームワークという用語を理解しましょう。

JSON Webトークン

JSON Web トークン ( JWT とも呼ばれます) は、2 つの当事者またはエンティティ間でランダム トークンを転送する安全な方法です。 JSON は通常、次の 3 つの部分で構成されます。

• ペイロード
• ヘッダ
• サイン

JSON は、2 者間でデータまたは情報を転送するときに 2 種類の構造形式を使用します。

• 連載中
• デシリアライズされた

シリアル化された形式は、各リクエストと応答を通じてネットワークにデータを転送するときに使用され、逆シリアル化された形式は、Web トークンにデータを読み書きするときに使用されます。

シリアル化された形式には 3 つのコンポーネントがあります。

• ヘッダ
• ペイロード
• サイン

ヘッダー コンポーネントは、トークンに関する暗号化情報を定義します。例えば：

• JWT は署名付きですか、それとも署名なしですか?
• アルゴリズム手法を定義する

逆シリアル化された形式には、シリアル化された形式とは異なり、2 つのコンポーネントが含まれています。

• ペイロード
• ヘッダ

REST API

API (アプリケーション プログラミング インターフェイス) を使用すると、2 つのアプリケーション間で通信してデータを取得または送信できます。 API には、Web API とシステム API という 2 つの一般的なタイプがあります。

この記事では、Web API についてのみ説明します。 Web APIには2種類あります。

• リクエスト – レスポンス API: Rest、GraphQL、リモート プロシージャ コール (RPC)
• イベント駆動型 API: WebHook、Web Sockets、HTTP ストリーミング

REST API はリクエスト/レスポンスのカテゴリに分類されます。 GET、POST、PUT などの HTTP メソッドを使用して API 操作を実行します。

典型的な例は、ユーザーが GET メソッドを Web サービスに送信して、特定のリソースまたはリソースのコレクションを要求または取得する場合です。その後、サーバーは特定のリソースまたはリソースのコレクションを、それを要求したユーザーに送り返します。

フラスコフレームワーク

Flask は Python をベースにしたフレームワークです。これは、Python 開発者が REST API を構築するために使用するマイクロフレームワークです。これは、開発者が設定に基づいてカスタム認証やその他のバックエンド システムを追加できるため、マイクロ フレームワークと呼ばれます。

実装から始めましょう。私のシステム設定は次のとおりです。

• OSとしてのUbuntu
• Python 2.7以降
• 郵便屋さん

virtualenv を使用して仮想環境をセットアップする

一部のパッケージがシステム パッケージと競合しないように、仮想環境をセットアップする必要があります。 virtualenv 使用して新しい仮想環境をセットアップしましょう。

システムで pip コマンドが使用できると仮定して、 pip 経由で次のコマンドを実行してインストールします。

 pip install virtualenv

マシンに pip がない場合は、この ドキュメント に従ってシステムに pip をインストールしてください。

次に、仮想環境を保存または保持するディレクトリを作成しましょう。以下に示す mkdir コマンドを使用してディレクトリを作成します

mkdir flaskproject

次のコマンドを使用して、 flaskproject ディレクトリに移動します。

 cd flaskproject

flaskproject ディレクトリ内で、 virtualenv ツールを使用して、以下に示すように仮想環境を作成します。

 virtualenv flaskapi

virtualenv ツールを使用して仮想環境を作成した後、 cd コマンドを実行して仮想環境として flaskapi ディレクトリに移動し、以下のコマンドを使用してアクティブ化します。

 source bin/activate

このプロジェクトに関連するすべてのタスクを仮想環境内で実行します。

pip を使用してパッケージをインストールする

次に、Flask フレームワークや PyJWT などのパッケージをインストールします。これらは、API プロジェクトの残りの API やその他の必要なパッケージを構築するために使用します。

次のパッケージを使用して、 requirements.txt ファイルを作成します。

 Flask  
datetime 
uuid
Flask-SQLAlchemy
PyJWT

pip でインストールします。

 pip install -r requirements.txt

データベースをセットアップする

SQLiteをインストールしましょう。

 apt-get install sqlite3

library という名前のデータベースを作成します。このデータベース内に、 Users と Authors テーブルという 2 つのテーブルを作成します。

Users テーブルには登録済みのユーザーが含まれます。登録ユーザーのみが Authors テーブルにアクセスできます。

著者テーブルには、登録ユーザーによって送信された著者の名前、出身国などの著者情報または詳細が保存されます。

次のコマンドを使用してデータベースを作成します。

 sqlite3 library.db

以下のコマンドを使用して、データベースが正常に作成されたかどうかを確認できます。

 .databases

新しいターミナルを開き、先ほど作成した仮想環境で以下を実行します。

 touch app.py

app.py という名前のファイル内に次のコードを貼り付けます。

 from flask import Flask, request, jsonify, make_response
from flask_sqlalchemy import SQLAlchemy
from werkzeug.security import generate_password_hash, check_password_hash
import uuid
import jwt
import datetime
from functools import wraps

上記のコードの最初の行は、 request や jsonify などのパッケージをインポートします。 request を使用してリクエスト中にリクエストレベルのデータを追跡し、 jsonify を使用してレスポンスを JSON 形式で出力します。

次の行では、SQLAlchemy の機能をフラスコに統合するために、 flask_sqlalchemy から SQLAlchemy をインポートしました。

werkzeug.security から、ユーザーのパスワード ハッシュを生成するための generate_password_hash と、ユーザーが送信したパスワードとデータベースに保存されているユーザーのパスワードを比較するときにユーザーのパスワードをチェックするための check_password_hash をインポートしました。

最後に、ユニバーサル一意識別子としても知られる uuid インポートして、ユーザーのランダムな ID 番号を生成しました。

さらに、 app.py ファイル内で、以下のコードを使用してライブラリ API の構成設定を実装します。

次のコードを import ステートメントの下に配置します。

 app = Flask(__name__)

app.config['SECRET_KEY']='Th1s1ss3cr3t'
app.config['SQLALCHEMY_DATABASE_URI']='sqlite://///home/michael/geekdemos/geekapp/library.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True

db = SQLAlchemy(app)

次に、以下に示すように、Users および Authors テーブルの 2 つのモデルを作成します。 app.py ファイル内のコードをコピーして貼り付けます。

このデータベース設定 db = SQLAlchemy(app) の直下に以下のコードを配置します。

 class Users(db.Model):
     id = db.Column(db.Integer, primary_key=True)
     public_id = db.Column(db.Integer)
     name = db.Column(db.String(50))
     password = db.Column(db.String(50))
     admin = db.Column(db.Boolean)

 class Authors(db.Model):
     id = db.Column(db.Integer, primary_key=True)
     name = db.Column(db.String(50), unique=True, nullable=False))
     book = db.Column(db.String(20), unique=True, nullable=False))
     country = db.Column(db.String(50), nullable=False))
     booker_prize = db.Column(db.Boolean)

ユーザーと作成者のテーブルを生成する

ターミナル上で、仮想環境内に次のコードを入力して、次に示すように Users テーブルと Authors テーブルの両方のテーブルを生成または作成します。

 from app import db
db.create_all()

その後、仮想環境内で app.py ファイルを開き、別の関数を作成します。

この関数は、登録ユーザーのみが Authors テーブルにアクセスし、一連の API 操作を実行できるようにするためにトークンを生成します。

このコードを Authors テーブルのデータベース モデルの下に配置します。

 def token_required(f):
   @wraps(f)
   def decorator(*args, **kwargs):

      token = None

      if 'x-access-tokens' in request.headers:
         token = request.headers['x-access-tokens']

      if not token:
         return jsonify({'message': 'a valid token is missing'})

      try:
         data = jwt.decode(token, app.config[SECRET_KEY])
         current_user = Users.query.filter_by(public_id=data['public_id']).first()
      except:
         return jsonify({'message': 'token is invalid'})

        return f(current_user, *args, **kwargs)
   return decorator

ユーザーテーブルのルートを作成する

次に、以下に示すように、ユーザーがユーザー名とパスワードを介して Authors API に登録できるようにするルートを作成しましょう。

再び仮想環境内で app.py ファイルを開き、関数 token_required(f) の下に次のコードを貼り付けます。

 @app.route('/register', methods=['GET', 'POST'])
def signup_user():  
 data = request.get_json()  

 hashed_password = generate_password_hash(data['password'], method='sha256')
 
 new_user = Users(public_id=str(uuid.uuid4()), name=data['name'], password=hashed_password, admin=False) 
 db.session.add(new_user)  
 db.session.commit()    

 return jsonify({'message': 'registered successfully'})

仮想環境内で、 app.py ファイルに別のルートを作成し、登録ユーザーがログインできるようにします。

ユーザーがログインすると、ユーザーがライブラリ API にアクセスするためのランダム トークンが生成されます。

以前に作成したルートの下に以下のコードを貼り付けます。

 @app.route('/login', methods=['GET', 'POST'])  
def login_user(): 
 
  auth = request.authorization   

  if not auth or not auth.username or not auth.password:  
     return make_response('could not verify', 401, {'WWW.Authentication': 'Basic realm: "login required"'})    

  user = Users.query.filter_by(name=auth.username).first()   
     
  if check_password_hash(user.password, auth.password):  
     token = jwt.encode({'public_id': user.public_id, 'exp' : datetime.datetime.utcnow() + datetime.timedelta(minutes=30)}, app.config['SECRET_KEY'])  
     return jsonify({'token' : token.decode('UTF-8')}) 

  return make_response('could not verify',  401, {'WWW.Authentication': 'Basic realm: "login required"'})

ただし、仮想環境内で、 app.py ファイルに別のルートを作成して、登録されているすべてのユーザーを取得または取得します。

このコードは、Users テーブル内のすべての登録ユーザーをチェックし、最終結果を JSON 形式で返します。

以下のコードをログイン ルートの下に貼り付けます。

 @app.route('/users', methods=['GET'])
def get_all_users():  
   
   users = Users.query.all() 

   result = []   

   for user in users:   
       user_data = {}   
       user_data['public_id'] = user.public_id  
       user_data['name'] = user.name 
       user_data['password'] = user.password
       user_data['admin'] = user.admin 
       
       result.append(user_data)   

   return jsonify({'users': result})

authors テーブルのルートを作成する

ユーザーがデータベース内のすべての著者を取得したり、著者を削除したりできるように、Authors テーブルのルートを作成しましょう。

有効なトークンを持つユーザーのみがこれらの API 操作を実行できます。

app.py ファイル内に、登録ユーザーが新しい作成者を作成するためのルートを作成します。

このコードをルートの下に貼り付けると、ユーザーはすべての登録ユーザーを取得できるようになります。

 @app.route('/author', methods=['POST', 'GET'])
@token_required
def create_author(current_user):
   
   data = request.get_json() 

   new_authors = Authors(name=data['name'], country=data['country'], book=data['book'], booker_prize=True, user_id=current_user.id)  
   db.session.add(new_authors)   
   db.session.commit()   

   return jsonify({'message' : 'new author created'})

次に、以下に示すように、有効なトークンを持つ登録ユーザーが Authors テーブル内のすべての著者を取得できるようにする別のルートを作成します。

このコードをルートの下に貼り付けると、ユーザーが新しい作成者を作成できるようになります。

 @app.route('/authors', methods=['POST', 'GET'])
@token_required
def get_authors(current_user):

    authors = Authors.query.filter_by(user_id=current_user.id).all()

    output = []
    for author in authors:

           author_data = {}
           author_data['name'] = author.name
           author_data['book'] = author.book
           author_data['country'] = author.country
           author_data['booker_prize'] = author.booker_prize
           output.append(author_data)

     return jsonify({'list_of_authors' : output})

最後に、 app.py ファイル内に、以下に示すように、指定した作成者を削除するルートを作成します。

このコードをルートの下に貼り付けると、ユーザーは著者のリストを取得できるようになります。

 @app.route('/authors/<author_id>', methods=['DELETE'])
@token_required
def delete_author(current_user, author_id):  
    author = Author.query.filter_by(id=author_id, user_id=current_user.id).first()   
    if not author:   
       return jsonify({'message': 'author does not exist'})   


    db.session.delete(author)  
    db.session.commit()   

    return jsonify({'message': 'Author deleted'})


if  __name__ == '__main__':  
     app.run(debug=True)

その後、仮想環境内の app.py ファイルを保存して閉じます。

Postman を使用したライブラリ API のテスト

このセクションでは、postman ツールを使用してデータベース サービスにリクエストを送信します。お使いのマシンに postman がインストールされていない場合は、 こちらで ダウンロードしてインストールする方法を確認できます。

郵便配達員とは別に、 Curl などの他のツールを利用してサーバーにリクエストを送信できます。

新しいターミナルを開き、次のように入力します。

 postman

コマンド postman 実行すると、Web ブラウザに次のページが表示されます。

サインアップして無料アカウントを作成することもできますが、ここではスキップして、以下に示すようにライブラリ API をテストするためにアプリに直接アクセスします。

このセクションでは、以下の手順に従って、POST メソッドを使用して JSON 形式でユーザー名と一意のパスワードを指定することで、ユーザーがライブラリ API に登録できるようにします。

• 「本文」というタブをクリックします。
• 次に、raw ボタンを選択し、JSON 形式を選択します
• スクリーンショットに示されているように、ユーザー名とパスワードを入力して登録します
• 送信ボタンの横に、次の URL を挿入します http://127.0.0.1/register
• 最後にメソッドをPOSTに変更して送信ボタンを押します。

以下に示すように、次の出力が表示されます。

これでユーザーの登録が完了しました。次の手順を使用して、一時的なランダム トークンを生成するために登録したばかりのユーザーがログインして Authors テーブルにアクセスできるようにしましょう。

• 「認証」タブをクリックします。
• タイプセクションで、基本認証を選択します。
• 次に、ユーザー名とパスワードのフォームに、以前に登録したユーザー名とパスワードを入力します。
• 最後に、送信ボタンを押してログインし、ランダムなトークンを生成します。

ユーザーがログインに成功すると、スクリーンショットに示すように、ユーザーに対してランダムなトークンが生成されます。

生成されたランダム トークンを使用して、Authors テーブルにアクセスします。

このセクションでは、次の手順を使用して、POST メソッドを介して著者の情報を Authors テーブルに追加します。

• ヘッダータブをクリックします
• スクリーンショットに示されている次の HTTP ヘッダーを含めます。

• 次に、本文タブをクリックして、新しい著者の詳細を入力します。
• 次に、送信ボタンを押して、著者の詳細を著者テーブルに追加します。

次の方法で、Authors テーブルの著者の情報を取得することもできます。

• 生成されたトークンがヘッダー セクションにあることを確認してください。そこにない場合は、トークンを入力する必要があります。
• 送信ボタンの横に、URL http://127.0.0.1/authors を入力します。
• 次に、HTTP メソッドを GET に変更し、送信ボタンを押して作成者の詳細を取得します。

最後に、次の手順を使用して、 DELETE メソッドを使用して Authors テーブル内の著者を削除できます。

• トークンがまだヘッダー セクションにあることを確認してください。ヘッダー タブをチェックして、必要な情報が適切に配置されていることを確認できます。
• 送信ボタンの横に、URL http://127.0.0.1/sam を入力します。
• 次に、送信ボタンを押して、指定したユーザーを削除します。

完全なソース コードは Github で見つけることができます。 クローンを作成して、マシン上でチェックアウトできます。