2023年3月21日火曜日

SQLAlchemyのEncriptedTypeを使ってみた

概要
環境
サンプルコード
- ちょっと解説
マイグレーションスクリプトの修正
マイグレーション
動作確認
最後に
おまけ: 固定の鍵を生成する方法
参考サイト

概要

sqlalchemy-utils という拡張に EncryptedType があったので使ってみました
データを暗号化して保存したい場合に便利です

今回は Flask 環境下で使用しています

環境

macOS 11.7.4
Python 3.10.2
Flask 2.2.3
Flask-SQLAlchemy 3.0.3
Flask-marshmallow 0.14.0
Flask-Migrate 4.0.4
SQLAlchemy 2.0.7
SQLAlchemy-Utils 0.40.0
cryptography 39.0.2

サンプルコード

vim app.py

from flask import (Flask,
                   request,
                   jsonify)
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from flask_migrate import Migrate
from cryptography.fernet import Fernet
from sqlalchemy_utils import EncryptedType

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+mysqldb://{}:{}@{}/{}?charset=utf8".format("root", "", "localhost", "test")
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)
migrate = Migrate(app, db)
ma = Marshmallow(app)
# 暗号化キーはアプリ起動ごとに変わる
key = Fernet.generate_key()


class EncryptedUser(db.Model):

    id = db.Column(db.Integer, primary_key=True, autoincrement="auto")
    name = db.Column(db.String(50), nullable=False)
    password = db.Column(EncryptedType(db.String, key), nullable=True)

    def __init__(self,
                 name: str,
                 password: str):
        cipher_suite = Fernet(key)
        self.name = name
        # 文字列のpasswordをバイトに変換しencryptで暗号化、戻っていた暗号化データのバイトを文字列に変換してデータベースに保存
        self.password = cipher_suite.encrypt(bytes(password, "utf-8")).decode("utf-8")


class EncryptedUserSchema(ma.Schema):

    class Meta:
        fields = ("id", "name", "password")

    def decode_password(self, obj) -> str:
        cipher_suite = Fernet(key)
        # 暗号化された文字列データをバイトに変換しdecryptで復号化、戻ってきたバイトデータを文字列にしてレスポンスで返却
        return cipher_suite.decrypt(bytes(obj.password, "utf-8")).decode("utf-8")
        # 暗号化したデータのままレスポンスに含めたい場合は以下の通り
        # return bytes(obj.password, "utf-8").decode("utf-8")

    password = ma.Method("decode_password")


@app.route('/insert')
def insert():
    name = request.args.get('name', 'default_user')
    password = request.args.get('password', 'default_password')
    user = EncryptedUser(name=name, password=password)
    db.session.add(user)
    db.session.commit()
    return jsonify({'status': 'ok'})


@app.route('/get')
def get():
    users = EncryptedUserSchema(many=True).dump(EncryptedUser.query.all())
    return jsonify({'users': users})

ちょっと解説

アプリのコードやマイグレーションのコード、レスポンススキーマのコードがすべて含まれています

EncryptedType カラムとして定義した場合にはコンストラクタを定義して該当のカラムの暗号化方法を定義します
今回は単純な共通鍵での暗号化 (AES) を行います
暗号化する場合は cipher_suite.encrypt を使います
今回は Flask アプリで扱うので最終的な入力や出力はすべて文字列ですが暗号化/復号化で扱うデータはすべてバイトになるので変換してあげます
復号化する場合は cipher_suite.decrypt を使います
MySQL から取得した暗号化データは文字列になっているのでバイトに変換し復号化したあとで再度レスポンス用に文字列に変換してあげましょう

MySQL 上での実態は blob 型として保存されています
当然データは鍵で暗号化されているので文字列に変換してもどんなデータ化はわからないようになっています

% mysql -u root test -e "show create table encrypted_user \G"                     
*************************** 1. row ***************************
       Table: encrypted_user
Create Table: CREATE TABLE `encrypted_user` (
  `id` int NOT NULL AUTO_INCREMENT,
  `name` varchar(50) NOT NULL,
  `password` blob,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=17 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

% mysql -u root test -e "select convert(password using utf8) from encrypted_user \G"
*************************** 1. row ***************************
convert(password using utf8): hKC9sQXHgSYEw7NSLbMVFJkIwhlCBP6P0K+EvByrmkgGuXzR6oGV52WoqFv4MOb+wzaZPxTMTCW8shqbo6F7kO3aAvJqR8a45zhu6fQkws7WAH0YLQhDCUbePcE2AowwE+5LuCEIOeVbCaKvqCxhFA==

鍵は Fernet.generate_key() で生成しています
今回はアプリ起動ごとに鍵が変わるような仕組みになっているので実際は一度鍵を生成したらその鍵を使い回すような仕組みにしましょう

マイグレーションスクリプトの修正

Flask-Migrate が EncryptedType に完全に対応していないようなのでマイグレーションスクリプトを手動で修正します
先頭に import 文を 1 行追加してあげるだけになります

FLASK_APP=app pipenv run flask db init
FLASK_APP=app pipenv run flask db migrate -m "Initial migration."
vim migrations/versions/147b8d1c4486_initial_migration.py

import sqlalchemy_utils

マイグレーション

実際にマイグレーションしてテーブルが作成されることを確認しましょう

FLASK_APP=app pipenv run flask db upgrade

動作確認

アプリを起動しデータを登録して確認します
動作確認は復号化されたデータで取得できるのでわかりづらいですがデータベースではしっかりデータが暗号化されていることが確認できると思います

pipenv run flask run
curl 'localhost:5000/insert?password=hoge'
curl 'localhost:5000/get'

最後に

SQLAlchemy-Utils の EncryptedType を使ってみました
内部的には blob で扱うタイプの暗号化ツールになります
今回は Python 側で暗号化してデータベースに登録するような流れですがデータベース自体に暗号化するような機能もある場合があるので用途に合わえて使い分けられるといいかなと思います

おまけ: 固定の鍵を生成する方法

import base64
keystr = 'JYZdCbdf7WSCRHgq7LjUMMYV18EbVIq0'
key = base64.urlsafe_b64encode(keystr.encode())

hawksnowlog