Python 日志记录模块加密

Question

我有一个带有日志记录的 python 脚本。现在我想使用 pycrypto 使用 AES 加密日志。

import logging
import base64
from Crypto.Cipher import AES
aes = AES.new(cryptoKey)
logging.basicConfig(filename='example.log',level=logging.DEBUG) #  file name, not custom file
logging.info('text')

我想在将其写入日志之前使用 base64.b64encode(aes.encrypt('#logging text#')) 。最有效的方法是什么？

Answer 1

加密不仅仅是数据转发。我建议编写您自己的日志格式化程序并将其设置为根格式化程序 - 这样无论您从应用程序的哪个位置登录，即使是不受您的代码控制的部分，它也将始终经过一层加密。所以，像这样:

import base64
import logging
from Crypto.Cipher import AES
from Crypto.Hash import SHA256
from Crypto import Random

class EncryptedLogFormatter(logging.Formatter):

    # make sure that the `key` is a byte stream on Python 3.x
    def __init__(self, key, fmt=None, datefmt=None):
        self._key = SHA256.new(key).digest()  # use SHA-256 for a proper-sized AES key
        super(EncryptedLogFormatter, self).__init__(fmt=fmt, datefmt=datefmt)

    def format(self, record):
        message = record.msg  # log message to encrypt, if any
        if message:  # no sense to encrypt empty log messages
            # on Python 3.x encode first: message = message.encode("utf-8")
            iv = Random.new().read(AES.block_size)  # we'll be using CBC so generate an IV
            cipher = AES.new(self._key, AES.MODE_CBC, iv)
            # AES demands all blocks to be of `AES.block_size` so we have to pad the message
            # you can use any padding you prefer, I think PKCS#7 is the best option
            padding = AES.block_size - len(message) % AES.block_size
            # pad the message...
            message += chr(padding) * padding # Python 3.x: bytes([padding]) * padding
            message_enc = iv + cipher.encrypt(message)  # add iv and encrypt
            # finally, replace our plain-text message with base64 encoded encrypted one
            record.msg = base64.b64encode(message_enc).decode("latin-1")
        # you can do more here, even print out your own string but we'll just
        # pass it to the default formatter now that the message is encrypted
        # so that it can respect other formatting options.
        return super(EncryptedLogFormatter, self).format(record)

然后您可以在任何可以更改日志格式器的地方使用它，即:

import sys
import logging

# lets get the root logger
root = logging.getLogger()
root.handlers = []  # blank out the existing handlers

# create a new handler, file handler instead of stdout is perfectly fine
handler = logging.StreamHandler(stream=sys.stdout)
# now lets get to business
handler.setFormatter(EncryptedLogFormatter("Whatever key/pass you'd like to use",
                                           "[%(levelname)s] %(message)s"))
# lets add it to the root logger so it gets called by the rest of the app automatically
root.addHandler(handler)

# And lets see what happens:
logging.warn("Sensitive stuff, hide me!")
# [WARNING] NDKeIav5G5DtbaSPB4Y/DR3+GZ9IwmXKzVTua1tTuDZ7uMwxBAKTXgIi0lam2dOQ
# YMMV, the IV is random so every block will be different every time

您当然可以加密级别、时间戳以及 logging.LogRecord 中的几乎所有内容，并且您可以输出您喜欢的任何格式。当需要阅读您的日志时，您只需执行相反的操作 - 请参阅 this answer 中的示例.

更新:根据要求，这里是“反向”的方法(即解密加密日志)。首先，让我们创建一些用于测试的日志条目(继续上一个):

root.setLevel(logging.DEBUG)  # let's make sure we support all levels

logging.warn("Lorem ipsum dolor sit amet.")
logging.info("Consectetur adipiscing elit.")
logging.debug("Sed do eiusmod tempor.")

如果格式保持不变 ([%(levelname)s] %(message)s)，这将导致类似的日志(当然，由于以下原因，它总是不同的)随机IV):

[WARNING] LQMLkbx3YF7ra3e5ZLRj3p1mi2dwCOJe/GMfo2Xg8BBSZMDmZO75rrgoiy/6kqjf
[INFO] D+ehnsq1kWQi61AsLOBkqglXla7jgc2myPFaCGcfCRe6drk9ZmNl+M3UkKPWkDiU
[DEBUG] +rHEHkM2YHJCkIL+YwWI4FNqg6AOXfaBLRyhZpk8/fQxrXLWxcGoGxh9A2vO+7bq

To create a reader for such a log (file) we need to be aware of the format so we can differentiate encrypted from non-encrypted data. In this case, separating the parts is easy - each log entry is on a new line, the levels are not encrypted and the actual encrypted data is always separated by a whitespace from the actual log level. So, to put all that together we might construct something like:

import base64
from Crypto.Cipher import AES
from Crypto.Hash import SHA256

# make sure that the `key` is a byte stream on Python 3.x
def log_decryptor(key, stream):  # assume the stream can be iterated line-by-line
    key = SHA256.new(key).digest()  # same derivation as in the EncryptedLogFormatter
    for line in stream:
        if not line.strip():  # empty line...
            continue  # ignore it!
        level, stream = line.split(None, 1)  # split on log level and log data
        message_enc = base64.b64decode(stream.encode("latin-1"))  # decode the stream
        iv = message_enc[:AES.block_size]  # grab the IV from the beginning
        # decrypt the stream
        message = AES.new(key, AES.MODE_CBC, iv).decrypt(message_enc[AES.block_size:])
        padding = ord(message[-1])  # get the padding value; Python 3.x: message[-1]
        if message[-padding:] != chr(padding) * padding:  # verify the padding
            # on Python 3.x:     bytes([padding]) * padding
            raise ValueError("Invalid padding encountered.")
        # Python 3.x: decode the message: message[:-padding].decode("utf-8")
        yield "{} {}".format(level, message[:-padding])   # yield the decrypted value

然后您可以将它用作常规生成器来解密您的日志，例如:

logs = ["[WARNING] LQMLkbx3YF7ra3e5ZLRj3p1mi2dwCOJe/GMfo2Xg8BBSZMDmZO75rrgoiy/6kqjf",
        "[INFO] D+ehnsq1kWQi61AsLOBkqglXla7jgc2myPFaCGcfCRe6drk9ZmNl+M3UkKPWkDiU",
        "[DEBUG] +rHEHkM2YHJCkIL+YwWI4FNqg6AOXfaBLRyhZpk8/fQxrXLWxcGoGxh9A2vO+7bq"]

for line in log_decryptor("Whatever key/pass you'd like to use", logs):
    print(line)

# [WARNING] Lorem ipsum dolor sit amet.
# [INFO] Consectetur adipiscing elit.
# [DEBUG] Sed do eiusmod tempor.

或者，如果您已将日志设置为流式传输到文件，则可以直接解密此类文件:

with open("path/to/encrypted.log", "r") as f:
    for line in log_decryptor("Whatever key/pass you'd like to use", f):
        print(line)  # or write to a 'decrypted.log' for a more persistent solution