В настоящее время я использую базовый API для получения журналов от поставщика облачной безопасности, но это не очень чистый процесс. Я хотел бы перейти на службу потоковой передачи событий (AMQPS), которую они предоставляют.
Я уже настроил очередь и имею соответствующую информацию на стороне облака, но проблема, с которой я сталкиваюсь, - это выяснить, как лучше всего принять этот поток и выгрузить их в системный журнал. Я использую syslog-ng прямо сейчас, но похоже, что он может отправлять только RabbitMQ и не принимать его.
В настоящее время я смотрю на Pika, чтобы узнать, сработает ли это, чтобы принять AMQPS, а затем направить его в файл журнала, но именно здесь у меня возникают некоторые проблемы. Любая помощь будет принята с благодарностью.
Спасибо
Я нашел страницу на github, сделанную walbit (https://github.com/walbit/AMP_API), который, похоже, у меня работает правильно. Я использую "consumer_event_stream_by_name.py", и он выгружает весь ввод в стандартный формат. Я сделал небольшую модификацию, чтобы отправить его в системный журнал, и теперь syslog-ng принимает его.
Моя единственная проблема на данный момент с этим кодом заключается в том, что время от времени кажется, что время от времени истекает или он умирает в целом, если соединения не поступают. Поэтому мне нужно выяснить, что с этим происходит.
#!/usr/bin/env python
import argparse
import pika
import pprint
import requests
import sys
import logging
import logging.handlers
# YOU NEED TO CREATE AN auth.py FILE WITH CLIENT_ID AND API_KEY STRINGS
from auth import CLIENT_ID, API_KEY
parser = argparse.ArgumentParser()
parser.add_argument('event_stream_name', metavar='event_stream_name',
nargs=1, help='event stream name')
parser.parse_args()
event_stream_name = parser.parse_args().event_stream_name[0]
api_endpoint = 'https://api.amp.cisco.com/v1/event_streams'
session = requests.Session()
session.auth = (CLIENT_ID, API_KEY)
event_streams = session.get(api_endpoint).json()['data']
event_stream = {}
for e in event_streams:
if e['name'] is event_stream_name:
event_stream = e
amqp_url = 'amqps://{user_name}:{password}@{host}:{port}'.format(
**e['amqp_credentials'])
queue = e['amqp_credentials']['queue_name']
parameters = pika.URLParameters(amqp_url)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
parameters = pika.URLParameters(amqp_url)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
my_logger = logging.getLogger('MyLogger')
my_logger.setLevel(logging.INFO)
handler = logging.handlers.SysLogHandler(address = '/dev/log')
my_logger.addHandler(handler)
def callback(ch, method, properties, body):
# print(" [x] Received meth:\t%r" % method)
# print(" [x] Received prop:\t%r" % properties)
# print(" [x] Received body:\t%r" % body)
# print(body)
my_logger.info('Fireamp: ' + body)
channel.basic_consume(callback, queue, no_ack=True)
print(" [*] Connecting to:\t%r" % amqp_url)
print(" [*] Waiting for messages. To exit press CTRL+C")
channel.start_consuming()