通过python读取authing IAM中的admin、user审计日志

方式:

通过python-logstash库，将读取的日志传给logstash的udp input地址。

 logstash config：

input {
  udp {
    port => 5959
    codec => json
  }
}

filter {
  json {
    source => "message"
    target => "parsed_data"
  }


  mutate {

        rename => {
              "[parsed_data][clientIp]" => "clientIp"
              "[parsed_data][operationType]" => "operationType"
              "[parsed_data][resourceType]" => "resourceType"
              "[parsed_data][requestId]" => "requestId"
              "[parsed_data][geoip]" => "geoip"
              "[parsed_data][timestamp]" => "timestamp"
              "[parsed_data][originValue]" => "originValue"
              "[parsed_data][targetValue]" => "targetValue"
              "[parsed_data][operationParam]" => "operationParam"
              "[parsed_data][userAgent]" => "userAgent"
            }

        remove_field => ["path", "@version", "logger_name", "@timestamp", "message", "parsed_data", "level", "host"]  
  }
}

output {
  stdout {
    codec => rubydebug
  }
}

python读authing审计日志：

import requests
import sys
import logging
import re
import json
import time
from datetime import datetime,timedelta
import logstash

accessKeyId = 'f3'
accessKeySecret = 'c6'

host = "https://console.authing.cn/"
userpool_id = '638'

logstashHost = "127.0.0.1"
logstashPort = 5959

def GetToken(ak=accessKeyId, sk=accessKeySecret, host=host):
    host = host
    header = {"Accept": "application/json, text/plain, */*",
              "Content-Type": "application/json",
              "authorization": "ssss"
              }
    data = {
        "accessKeyId": ak,
        "accessKeySecret": sk
    }
    target = "{u}api/v3/get-management-token".format(u=host)

    try:
        r1 = requests.post(url=target, headers=header, data=json.dumps(data), verify=True, timeout=15)

        if r1.status_code == 200:
            data = json.loads(r1.text)
            access_token = data['data']['access_token']
            return (access_token)
        else:
            print("失败:" + str(r1.status_code) + r1.text + '\n')
    except Exception as e:
        print(e)

def GetTime():
    currentTime = datetime.now()
    #查询前5分钟的日志
    startTime = currentTime - timedelta(seconds=5555510)
    endTime = currentTime - timedelta(seconds=10)

    startTimeStr = startTime.strftime("%Y-%m-%d %H:%M:%S.%f")
    endTimeStr = endTime.strftime("%Y-%m-%d %H:%M:%S.%f")
    # 先转换为时间数组
    startTimeArray = datetime.strptime(startTimeStr, "%Y-%m-%d %H:%M:%S.%f")
    endTimeArray = datetime.strptime(endTimeStr, "%Y-%m-%d %H:%M:%S.%f")

    # 转换为时间戳
    startTimeStamp = int(time.mktime(startTimeArray.timetuple()) * 1000.0 + startTimeArray.microsecond / 1000.0)
    endTimeStamp = int(time.mktime(endTimeArray.timetuple()) * 1000.0 + endTimeArray.microsecond / 1000.0)

    print("startTime:",startTimeStamp)
    print("endTime:",endTimeStamp)
    return(startTimeStamp,endTimeStamp)

def GetAdminPages(access_token, start, end, ak=accessKeyId, sk=accessKeySecret, host=host, pool_id=userpool_id):

    host = host
    header = {
        "Accept": "application/json, text/plain, */*",
        "Content-Type": "application/json",
        "x-authing-userpool-id": pool_id,
        "authorization": access_token
    }

    data = {
        "start": start,
        "end": end
    }
    target = "{u}api/v3/get-admin-audit-logs".format(u=host)

    try:
        r2 = requests.post(url=target, headers=header, data=json.dumps(data), verify=True, timeout=15)
        if r2.status_code == 200:
            data = json.loads(r2.text)
            totalCount = data['data']['totalCount']
            pages = totalCount / 10 + 1
            return (int(pages))
        else:
            print("失败1" + str(r1.status_code) + r1.text + '\n')
    except Exception as e:
        print(e)

def GetAdminLog(access_token, pages, ak=accessKeyId, sk=accessKeySecret, host=host, pool_id=userpool_id):
    #通过python logstash向logstash发送收到的日志
    logstashLogger = logging.getLogger('python-logstash-logger')
    logstashLogger.addHandler(logstash.LogstashHandler(logstashHost, logstashPort, version=1))
    logstashLogger.setLevel(logging.INFO)

    host = host
    header = {
        "Accept": "application/json, text/plain, */*",
        "Content-Type": "application/json",
        "x-authing-userpool-id": pool_id,
        "authorization": access_token
    }
    target = "{u}api/v3/get-admin-audit-logs".format(u=host)

    page = 1
    print("total_page:", str(pages))
    if pages == 1:
        try:
            r2 = requests.post(url=target, headers=header, verify=True, timeout=15)
            print("current_page:", str(page))
            if r2.status_code == 200:
                data = json.loads(r2.text)
                #print("page1 data:",data['data']['list'])
                if len(data['data']['list']) == 0:
                    return 0
                else:
                    for item in data['data']['list']:
                        result = json.dumps(item, ensure_ascii=False)
                        adminLog = result.encode('utf-8')
                        logstashLogger.info(adminLog.decode('utf-8'))
                        print(adminLog.decode('utf-8'))
            else:
                print("失败1" + str(r1.status_code) + r1.text + '\n')
        except Exception as e:
            print(e)

    else:
        for page in range(1,pages):
            data = {
                "pagination": {
                    "page": page,
                    "limit": 10
                }
            }
            try:
                r2 = requests.post(url=target, headers=header, data=json.dumps(data), verify=True, timeout=15)
                print("current_page:", str(page))
                if r2.status_code == 200:
                    data = json.loads(r2.text)
                    for item in data['data']['list']:
                        result = json.dumps(item, ensure_ascii=False)
                        adminLog = result.encode('utf-8')
                        logstashLogger.info(adminLog.decode('utf-8'))
                        print(adminLog.decode('utf-8'))
                else:
                    print("失败1" + str(r1.status_code) + r1.text + '\n')
            except Exception as e:
                print(e)

if __name__ == '__main__':
    token = GetToken()
    start,end = GetTime()
    pages = GetAdminPages(token,start,end)
    GetAdminLog(token, pages)