まろ茶ネットブログ

bipset =  list(map(lambda x: ipv4adrset(x.decode('UTF-8').rstrip('\n')), proc.stdout))
　↓
bipset =  [ipv4adrset(x.decode('UTF-8').rstrip('\n')) for x in proc.stdout]

from blocklib import ipv4adrset
import os, glob, gzip
logger = getLogger(__name__)
import re, json

# djangoのアクセスログ(json形式)を読み込に認証履歴のレポートを返す
# 認証履歴のログはdjango認証バックエンドを継承して認証時のログ出力機能を追加したもの
# [引数]djlog : 解析するアクセスログ(json形式)をフルパスで与える（gzipもOK）
#              Noneの場合は、デフォルト django_log_json を使用（logorotateされたものも含み全てを読み込む）
#       st : レポート対象の開始日時（logrotateされたファイルはこの日時より新しいものを読み込み対象とする）
#       ed : レポート対象の終了日時
#       address : レポート対象のIPアドレスオブジェクト（Noneの場合は全て）サブネット検索対応
def get_django_loginfails(djlog, st, ed, address=None):
    if djlog is not None:
        djlogs = [djlog]
    else:
        djlogs = [x for x in glob.glob(django_log_json + '*') if re.match('.*(log|gz|[0-9])',x) and os.path.getmtime(x) > st.timestamp()]
        djlogs.sort(key=lambda x: os.path.getmtime(x))
    retlist = []
    for jlog in djlogs:
        logger.debug(jlog)
        with open(jlog, 'r') if not re.match('.*\.gz$', jlog) else gzip.open(jlog, 'rt') as jf:
            #jl = json.load(jf)
            decoder = json.JSONDecoder()
            jlines = map(lambda x: decoder.raw_decode(x), jf)
                #aa = decoder.raw_decode(line)
            for jl in map(lambda x: x[0], jlines):
                # fromisoformatで読み込むために末尾のZが邪魔なので削除
                # UTCのタイムゾーンを設定してJST時刻に変換する
                ts = datetime.fromisoformat(jl['timestamp'].rstrip('Z')).replace(tzinfo=timezone.utc).astimezone(tz_jst)
                ts = ts.replace(microsecond=0)
                if ts > st and ts < ed:
                    (dat, event) = (jl['ip'], jl['event'])
                    res0 = re.match('user logged in : (.+)$', event)
                    res1 = re.match('login faild.*: (.+)$', event)
                    event = False if res1 else True if res0 else None
                    if event is None:
                        continue
                    name = res1[1] if res1 else res0[1] if res0 else ''
                    if address is not None and address != ipv4adrset(dat):
                        continue
                    logger.debug('%s : %s - %s(%s)' % (str(ts), dat, name, event))
                    retlist.append(
                        {
                            'name':name,
                            'datetime':ts,
                            'address':ipv4adrset(dat),
                            'status':event,
                            'source':'django',
                        }
                    )
                    #print('%s : %s : %s' % (str(ts), dat, event))
    return retlist

www-data        ALL=(ALL)       NOPASSWD:/usr/local/sbin/getblocklist.py

import subprocess

get_blocklist_proccess_cmd = ['sudo','/usr/local/sbin/getblocklist.py']

# root権限でブロックリスト取得コマンドを走らせて標準入力を得る
def get_blocklist():
    proc = subprocess.Popen(get_blocklist_proccess_cmd, stdout=subprocess.PIPE)
    bipset =  list(map(lambda x: ipv4adrset(x.decode('UTF-8').rstrip('\n')), proc.stdout))
    return bipset