MySQLのAES_ENCRYPTを SQLAlchemy で使う方法

• 概要
• 環境
• サンプルコード
  • 解説
• SQL の場合には
• 最後に

概要

前回 Python 側で暗号化してデータを保存する方法を紹介しました
今回は MySQL 自体が提供する暗号化用の関数 aes_encrypt を sqlalchemy から使ってみました

環境

• 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

サンプルコード

• 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 sqlalchemy import cast
from sqlalchemy.dialects.mysql import CHAR


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 = "xxx"


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(db.String(50), nullable=False)

    debug_password = db.column_property(
        cast(
            db.func.aes_decrypt(
                db.func.unhex(password), db.func.unhex(db.func.sha2(key, 512))
            ),
            CHAR(charset="utf8"),
        )
    )

    def __init__(self, name, password):
        self.name = name
        self.password = db.func.hex(
            db.func.aes_encrypt(
                password, db.func.unhex(db.func.sha2(key, 512))
            )
        )


class EncryptedUserSchema(ma.Schema):

    class Meta:
        fields = ("id", "name", "password", "debug_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})

解説

MySQL 側の関数を直接使う場合には db.func を通してコールします

保存する場合に db.func.aes_encrypt を使い暗号化し取得する場合には db.func.aes_decrypt を使って復号化します
機能はどちらも SQLAlchemy の機能なので上記らのメソッドを扱う場合は marshmallow 側ではなく db.Model 側で扱う必要があります

今回はデバッグとして column_property を使って取得専用のフィールドを定義しています
aes_decrypt 後は Python で扱えるように文字列に変換する必要があります
convert 関数でもいいのですが db.func.convert が使えないの (using utf8 が使えない) ので今回は cast 関数を使っています
また cast 関数だけ直接 SQLAlchemy パッケージを参照していますがうまくやれば db.func からでも参照できるかもしれません

SQL の場合には

当然 SQL を直接コールすることもできます
select の場合には convert or cast 関数どちらかが選択できます

• insert

insert into encrypted_user values (null, 'hawk', hex(AES_ENCRYPT('fuga', UNHEX(SHA2('xxx',512)))));

• select

select convert(AES_DECRYPT(unhex(password), UNHEX(SHA2('xxx',512))) USING utf8) from encrypted_user;

or

select CAST(aes_decrypt(unhex(password), unhex(sha2('xxx',512))) AS CHAR CHARACTER SET utf8) from encrypted_user;

最後に

MySQL の暗号化機能を SQLAlchemy で扱う方法を紹介しました
データベース側の機能を使うことで Python 側で暗号化復号化を考える必要がないのとカラムの型も単純な文字列として定義できるのでコードはシンプルになります

レスポンススキーマ側ではすでに復号化されたデータを扱うことになるのでシンプルになります