Pythonのロガーでログのマスキングを実装してみる

Pythonの標準ロギングライブラリでマスキング機能を実装する方法を紹介します。AWS Lambda環境での使用を想定し、外部ライブラリに頼らない実装にしました。
2026.10.09
こんばんは、情報システム室の夏目です。
先月TSのロギングライブラリlogtapeでログのマスキング試してみました。
そういえば、Pythonを書くときにはログのマスキングをしてなかったと思い、実装しようと思います。

(ライブラリの使用ではなく実装を選んだのは、AWS Lambdaの実装にPythonを使うことが多いので外部ライブラリを使いたくなかったからです)
 Python標準ライブラリのロガーPythonでは標準ライブラリに logging があるので基本的にこれを使う想定で進めようかと思います。
logging の仕組みを考えると、 Logger/Handler/Filter/Formatter/LogRecord の5つの要素があります。
Logger
役割: ログの受付・管理
ログを受け取り、ログレベルとFilterによる出力判定を行い、LogRecordにしてからHandlerに渡す
FilterとHandlerをそれぞれ0個以上設定できる

Handler
役割: ログの出力先・出力処理
コンソール、ファイル、ネットワークなどログ出力を担当する
受け取ったLogRecordに対して、ログレベルとFilterによる出力判定を行い、Formatterで加工してから出力を行う
Filterを0個以上、Formatterを0個もしくは1個設定できる

Filter
役割: ログのフィルタリング・加工
LogRecordの内容に応じて出力可否を判定したり、LogRecordを加工したり差し替えたりする

Formatter
役割: ログの整形
LogRecordを文字列に変換し、日時やログレベルなどを含む出力形式を決定する

LogRecord
役割: ログ情報の保持
ログメッセージ、ログレベル、日時、例外情報、追加情報などを保持するオブジェクト

これをわかりやすく整形すると次の二つになります。
以上を踏まえてマスキングの方法を考えた結果、Formatterにマスキングを仕込もうかと思います。
Filterはフィルタリングと加工を行うものなので、最初はFilterの方がいいかとも思いました。

しかし、Filterを通した後にFormatterが動くこと、Formatterが文字列化するのでオブジェクトを文字列にする過程で混入することがありそうだと考えたからです。
 Formatterにマスキングを仕込むFormatterにマスキングを仕込もうと思います。

JSON構造化されたログを多用するので、JSON構造化とマスキングを行います。
今回のマスキングでは以下の二つをできるようにしようと思います。
正規表現で一致したkeyがあれば除外する
正規表現で一致したvalueがあればマスキングをする
JSON構造化ログにするためのコードを含めて次のようになりました。
json_log_formatter.py
import json
from logging import Formatter, LogRecord
from re import Match, Pattern

def redact(
    obj: str | dict | list | int | float | bool | None,
    exclude_key_patterns: list[Pattern],
    masking_value_patterns: list[Pattern],
) -> str | dict | list | int | float | bool | None:
    if isinstance(obj, str):
        for pattern in masking_value_patterns:
            match: Match | None = pattern.search(obj)
            if match:
                return pattern.sub("[MASKED]", obj)
        return obj
    elif isinstance(obj, dict):
        data = {}
        for key, value in obj.items():
            flag_exclude_key = False
            for pattern in exclude_key_patterns:
                match: Match | None = pattern.search(key)
                if match:
                    flag_exclude_key = True
                    break
            if flag_exclude_key:
                continue

            data[key] = redact(value, exclude_key_patterns, masking_value_patterns)
        return data
    elif isinstance(obj, list):
        return [redact(x, exclude_key_patterns, masking_value_patterns) for x in obj]
    else:
        return obj

class JsonLogFormatter(Formatter):
    exclude_key_patterns: list[Pattern]
    masking_value_patterns: list[Pattern]

    def __init__(
        self,
        exclude_key_patterns: list[Pattern] | None = None,
        masking_value_patterns: list[Pattern] | None = None,
    ):
        super().__init__()
        if exclude_key_patterns:
            self.exclude_key_patterns = exclude_key_patterns
        else:
            self.exclude_key_patterns = []

        if masking_value_patterns:
            self.masking_value_patterns = masking_value_patterns
        else:
            self.masking_value_patterns = []

    def format(self, record: LogRecord) -> str:
        ret = {}

        for attr, value in record.__dict__.items():
            if attr == "asctime":
                value = self.formatTime(record)
            elif attr == "msg":
                try:
                    value = record.getMessage()
                except Exception:
                    pass

            ret[attr] = value

        raw = json.dumps(
            ret,
            ensure_ascii=False,
            default=lambda x: {"type": str(type(x)), "value": str(x)},
        )

        data = json.loads(raw)

        redacted = redact(data, self.exclude_key_patterns, self.masking_value_patterns)
        return json.dumps(redacted, ensure_ascii=False, indent=2)
今回は1度JSON化してから再度dictに変換し、マスキング用の関数 redact() に通しています。

これはJSON化すると型が str | dict | list | int | float | bool | None に限定されるためです。
ちゃんと作り込むときには方式を変更した方が良いかと思います。
マスキング用の関数 redact() は再帰的に処理させることで、dictやlistにも対応させています。

また、今回は表示内容をわかりやすくするために最後にJSON化する際に indent=2 としています。
 試してみる試してみます。
trial_formatter.py
from logging import DEBUG, StreamHandler, getLogger
from re import IGNORECASE, Pattern, compile

from json_log_formatter import JsonLogFormatter

exclude_key_patterns: list[Pattern] = [
    compile(r"^authorization$"),
    compile(r"^address$", flags=IGNORECASE),
    compile(r"^pathname$"),
]

masking_value_patterns: list[Pattern] = [compile(r"\d{3}-\d{4}-\d{4}")]

handler_with_masking = StreamHandler()
handler_with_masking.setLevel(DEBUG)
handler_with_masking.setFormatter(
    JsonLogFormatter(exclude_key_patterns, masking_value_patterns)
)

handler_without_masking = StreamHandler()
handler_without_masking.setLevel(DEBUG)
handler_without_masking.setFormatter(JsonLogFormatter())

logger_with_masking = getLogger("with_masking")
logger_with_masking.setLevel(DEBUG)
logger_with_masking.addHandler(handler_with_masking)

logger_without_masking = getLogger("without_masking")
logger_without_masking.setLevel(DEBUG)
logger_without_masking.addHandler(handler_without_masking)

logger_with_masking.info(
    "user info (191-9191-1919)",
    extra={
        "data": {
            "authorization": "Bearer aaaaa",
            "user_name": "test-user",
            "phone": "191-9191-1919",
            "address": {"country": "japan", "city": "munakata"},
        }
    },
)

logger_without_masking.info(
    "user info (191-9191-1919)",
    extra={
        "data": {
            "authorization": "Bearer aaaaa",
            "user_name": "test-user",
            "phone": "191-9191-1919",
            "address": {"country": "japan", "city": "munakata"},
        }
    },
)
同じログをマスキングを設定したロガーと設定していないロガー両方に渡します。
実行すると次のようになりました。
$ python trial_formatter.py
{
  "name": "with_masking",
  "msg": "user info ([MASKED])",
  "args": [],
  "levelname": "INFO",
  "levelno": 20,
  "filename": "trial_formatter.py",
  "module": "trial_formatter",
  "exc_info": null,
  "exc_text": null,
  "stack_info": null,
  "lineno": 33,
  "funcName": "<module>",
  "created": 1791539214.809972,
  "msecs": 809.0,
  "relativeCreated": 2.23,
  "thread": 8495112576,
  "threadName": "MainThread",
  "processName": "MainProcess",
  "process": 45403,
  "taskName": null,
  "data": {
    "user_name": "test-user",
    "phone": "[MASKED]"
  }
}
{
  "name": "without_masking",
  "msg": "user info (191-9191-1919)",
  "args": [],
  "levelname": "INFO",
  "levelno": 20,
  "pathname": "/Users/natsume.yuta/spaces/work/blog/025_python_redaction/logger_redaction/trial_formatter.py",
  "filename": "trial_formatter.py",
  "module": "trial_formatter",
  "exc_info": null,
  "exc_text": null,
  "stack_info": null,
  "lineno": 45,
  "funcName": "<module>",
  "created": 1791539214.8101811,
  "msecs": 810.0,
  "relativeCreated": 2.439,
  "thread": 8495112576,
  "threadName": "MainThread",
  "processName": "MainProcess",
  "process": 45403,
  "taskName": null,
  "data": {
    "authorization": "Bearer aaaaa",
    "user_name": "test-user",
    "phone": "191-9191-1919",
    "address": {
      "country": "japan",
      "city": "munakata"
    }
  }
}
マスキングが問題なく動いていることがわかります。
 AWS Lambda PowertoolsのLoggerにマスキングを仕込みたいLambdaの開発においては自分はAWS Lambda PowertoolsのLoggerを使っています。

こちらにもマスキングを仕込んでみようと思います。
AWS Lambda Powertoolsの LambdaPowertoolsFormatter を継承するFormatterを作成し、 format() に処理を追加することで対応します。
my_extended_formatter.py
import json
import logging
from logging import LogRecord
from re import Match, Pattern
from typing import Any, Callable

from aws_lambda_powertools.logging.formatter import LambdaPowertoolsFormatter

def redact(
    obj: str | dict | list | int | float | bool | None,
    exclude_key_patterns: list[Pattern],
    masking_value_patterns: list[Pattern],
) -> str | dict | list | int | float | bool | None:
    if isinstance(obj, str):
        for pattern in masking_value_patterns:
            match: Match | None = pattern.search(obj)
            if match:
                return pattern.sub("[MASKED]", obj)
        return obj
    elif isinstance(obj, dict):
        data = {}
        for key, value in obj.items():
            flag_exclude_key = False
            for pattern in exclude_key_patterns:
                match: Match | None = pattern.search(key)
                if match:
                    flag_exclude_key = True
                    break
            if flag_exclude_key:
                continue

            data[key] = redact(value, exclude_key_patterns, masking_value_patterns)
        return data
    elif isinstance(obj, list):
        return [redact(x, exclude_key_patterns, masking_value_patterns) for x in obj]
    else:
        return obj

class MyExtendedFormatter(LambdaPowertoolsFormatter):
    exclude_key_patterns: list[Pattern]
    masking_value_patterns: list[Pattern]

    def __init__(
        self,
        exclude_key_patterns: list[Pattern] | None = None,
        masking_value_patterns: list[Pattern] | None = None,
        json_serializer: Callable[[LogRecord], str] | None = None,
        json_deserializer: Callable[[dict | str | bool | int | float], str]
        | None = None,
        json_default: Callable[[Any], Any] | None = None,
        datefmt: str | None = None,
        use_datetime_directive: bool = False,
        log_record_order: list[str] | None = None,
        utc: bool = False,
        use_rfc3339: bool = False,
        serialize_stacktrace: bool = True,
        **kwargs,
    ) -> None:
        super().__init__(
            json_serializer=json_serializer,
            json_deserializer=json_deserializer,
            json_default=json_default,
            datefmt=datefmt,
            use_datetime_directive=use_datetime_directive,
            log_record_order=log_record_order,
            utc=utc,
            use_rfc3339=use_rfc3339,
            serialize_stacktrace=serialize_stacktrace,
            **kwargs,
        )

        if exclude_key_patterns:
            self.exclude_key_patterns = exclude_key_patterns
        else:
            self.exclude_key_patterns = []

        if masking_value_patterns:
            self.masking_value_patterns = masking_value_patterns
        else:
            self.masking_value_patterns = []

    def format(self, record: logging.LogRecord) -> str:
        raw = super().format(record)
        data = json.loads(raw)
        redacted = redact(data, self.exclude_key_patterns, self.masking_value_patterns)
        return json.dumps(redacted, ensure_ascii=False, indent=2)
trial_my_extended_formatter.py
from re import IGNORECASE, Pattern, compile

from aws_lambda_powertools import Logger

from my_extended_formatter import MyExtendedFormatter

exclude_key_patterns: list[Pattern] = [
    compile(r"^authorization$"),
    compile(r"^address$", flags=IGNORECASE),
    compile(r"^pathname$"),
]

masking_value_patterns: list[Pattern] = [compile(r"\d{3}-\d{4}-\d{4}")]

logger_with_masking = Logger(
    service="with_masking",
    logger_formatter=MyExtendedFormatter(
        exclude_key_patterns=exclude_key_patterns,
        masking_value_patterns=masking_value_patterns,
    ),
)

logger_without_masking = Logger("without_masking")

logger_with_masking.info(
    "user info (191-9191-1919)",
    extra={
        "data": {
            "authorization": "Bearer aaaaa",
            "user_name": "test-user",
            "phone": "191-9191-1919",
            "address": {"country": "japan", "city": "munakata"},
        }
    },
)

logger_without_masking.info(
    "user info (191-9191-1919)",
    extra={
        "data": {
            "authorization": "Bearer aaaaa",
            "user_name": "test-user",
            "phone": "191-9191-1919",
            "address": {"country": "japan", "city": "munakata"},
        }
    },
)
$ uv init --python 3.14 --bare
$ uv add aws-lambda-powertools
$ uv run trial_my_extended_formatter.py
{
  "level": "INFO",
  "location": "<module>:25",
  "message": "user info ([MASKED])",
  "timestamp": "2026-10-09 19:07:55,708+0900",
  "service": "with_masking",
  "data": {
    "user_name": "test-user",
    "phone": "[MASKED]"
  }
}
{"level":"INFO","location":"<module>:37","message":"user info (191-9191-1919)","timestamp":"2026-10-09 19:07:55,708+0900","service":"without_masking","data":{"authorization":"Bearer aaaaa","user_name":"test-user","phone":"191-9191-1919","address":{"country":"japan","city":"munakata"}}}
こちらも問題なく動きました。
 まとめ以上、Pythonのロガーでマスキングを実装する話でした。
何かのお役に立てたら幸いです。