Gordon Sim wrote:
On 10/07/2011 05:50 PM, Fraser Adams wrote:
What I'd quite like to be able to do is to log, but not deny if a queue
is created that's not one of a named set. I'm suspecting that I can't do
that with acl and I might have to write a QMF client to do that.
I think you are right, QMF may be the simplest way to go. You can
quite easily get events sent out for queue creation and could then
check the name against an expected set and log any deviation.
So I gave this a go. This is my very first python program :-)
A bit of "copy and paste" reuse from qpid-printevents and some help from
Mr Google. It seems to work pretty well - I added some exchangeName
whitelisting to avoid triggering on the QMF exchanges.
As it's based on qpid-printevents it's QMF1 based which I'm less
familiar with than QMF2, but as qpid-printevents had all the broker
reconnection logic in place I thought I'd stick with that for now.
It wasn't quite as easy as just triggering on a couple of events as I
wanted to log the IP address of the connection and to whitelist against
exchangeNames, so I needed to trigger on quite a few object properties.
What do you reckon - useful?
As a slight aside... Gordon are there any plans to update qpid-config on
an official release with your patch that displays the binding.arguments
if they exist and so make it useful for headers exchange bindings. I've
got my own patched version, but I'd rather use a vanilla version from a
release. Similarly qpid-route had a test in place (at the start of the
addLink() method) to prevent one linking a broker to itself, things work
fine if I comment out the test. I can't see a good reason for the test -
or log a warning rather than throw an exception.
Frase
#!/usr/bin/env python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
# This application audits connections to one or more Qpid message brokers.
# Exchange and Queue names are checked against a whitelist and if no match is
found an alert is generated
import os
import os.path
import optparse
from optparse import IndentedHelpFormatter
import sys
import socket
from time import time, strftime, gmtime, localtime, sleep
from qmf.console import Console, Session
from xml.etree.ElementTree import ElementTree
class AuditConsole(Console):
# AuditConsole Constructor
def __init__(self, whitelistFile):
self.whitelistFile = whitelistFile
self.whitelistLastModified = None
# Declare dictionary to map queue name to queue objects
self.queueNameMap = {}
# Declare dictionary to map queue name to subscription objects
self.queueNameSubscriptionMap = {}
# Declare dictionaries to map object-ids to objects
self.queueMap = {}
self.exchangeMap = {}
self.bindingMap = {}
self.connectionMap = {}
self.sessionMap = {}
# Declare the sets to be used as the whitelists
self.exchangeWhitelist = set()
self.queueWhitelist = set()
# Overridden method from Console
# Handle property updates. This method registers or deletes the relevant
objects in the relevant Map
# These objects are used by the validateSubscription() in order to find out
information about the
# bindings and connections associated with a given queue in order to work
out if it needs to be tested
# against the whitelist. For example we don't validate queues bound to the
management exchanges as
# this would generate spurious results as most of these queues are
temporary queues.
def objectProps(self, broker, record):
classKey = record.getClassKey()
oid = record.getObjectId()
deleted = record.getTimestamps()[2] > 0
# Note that queue objects are keyed by both queue name and Object ID
if classKey.getClassName() == "queue":
queueName = record.name
if oid not in self.queueMap and not deleted:
self.queueMap[oid] = record
if queueName not in self.queueNameMap and not deleted:
self.queueNameMap[queueName] = record
# If the queue object has been deleted remove it from the map
if deleted:
if queueName in self.queueNameMap:
self.queueNameMap.pop(queueName)
if oid in self.queueMap:
self.queueMap.pop(oid)
if classKey.getClassName() == "exchange":
if oid not in self.exchangeMap and not deleted:
self.exchangeMap[oid] = record
# If the exchange object has been deleted remove it from the map
if deleted and oid in self.exchangeMap:
self.exchangeMap.pop(oid)
if classKey.getClassName() == "binding":
if oid not in self.bindingMap and not deleted:
self.bindingMap[oid] = record
# If the binding object has been deleted remove it from the map
if deleted and oid in self.bindingMap:
self.bindingMap.pop(oid)
if classKey.getClassName() == "connection":
if oid not in self.connectionMap and not deleted:
# Re-read the whitelist when a new connection occurs
self.readWhitelist()
self.connectionMap[oid] = record
# If the connection object has been deleted remove it from the map
if deleted and oid in self.connectionMap:
self.connectionMap.pop(oid)
if classKey.getClassName() == "session":
if oid not in self.sessionMap and not deleted:
self.sessionMap[oid] = record
# If the session object has been deleted remove it from the map
if deleted and oid in self.sessionMap:
self.sessionMap.pop(oid)
# Note that subscription objects are keyed by queue name NOT by Object
ID
if classKey.getClassName() == "subscription":
# Find the queue associated with the subscription by looking its
reference up in queueMap
queue = self.queueMap.get(record.queueRef)
if queue != None:
queueName = queue.name
if queueName not in self.queueNameSubscriptionMap and not
deleted:
self.queueNameSubscriptionMap[queueName] = record
self.validateSubscription(queueName)
# If the subscription object has been deleted remove it from
the map
if deleted and queueName in self.queueNameSubscriptionMap:
self.queueNameSubscriptionMap.pop(queueName)
# This method first checks if the whitelist file exists, if not it clears
the sets used as whitelists
# so that no whitelisting is applied. If the whitelist file does exist it
is parsed by an ElementTree
# we look for all exchange and queue elements and populate the respective
whitelist sets with their
# contents. Note that we check the whitelist file update time to avoid
reading it if it hasn't been changed
def readWhitelist(self):
if os.path.exists(self.whitelistFile):
mtime = os.path.getmtime(self.whitelistFile)
if mtime != self.whitelistLastModified:
self.whitelistLastModified = mtime
self.exchangeWhitelist.clear()
self.queueWhitelist.clear()
tree = ElementTree()
try:
tree.parse(self.whitelistFile)
exchanges = tree.findall("exchangeWhitelist/exchange")
for i in exchanges:
exchange = i.text
if exchange == None:
self.exchangeWhitelist.add("")
else:
self.exchangeWhitelist.add(exchange)
queues = tree.findall("queueWhitelist/queue")
for i in queues:
queue = i.text
self.queueWhitelist.add(queue)
except Exception, e: # Failed to parse correctly
print strftime("%c", localtime(time())), "WARN
connection-audit:readWhitelist failed: %s - %s" % (e.__class__.__name__, e)
else: # If whitelist file doesn't exist log a warning and clear the
whitelists
print strftime("%c", localtime(time())), "WARN
connection-audit:readWhitelist %s doesn't exist" % self.whitelistFile
sys.stdout.flush()
self.exchangeWhitelist.clear()
self.queueWhitelist.clear()
# Given a queue name this method looks up the associated subscription,
session and connection
# objects and returns the connection
def findConnection(self, queueName):
subscription = self.queueNameSubscriptionMap.get(queueName)
if subscription != None:
session = self.sessionMap.get(subscription.sessionRef)
if session != None:
connection = self.connectionMap.get(session.connectionRef)
if connection != None:
return connection
# This method finds bindings that match a given queue name, if one is found
the associated exchange is
# recovered and then a call to validateQueue() is made
def validateSubscription(self, queueName):
for binding in self.bindingMap.values():
queue = self.queueMap.get(binding.queueRef)
if queue != None and queue.name == queueName:
exchange = self.exchangeMap.get(binding.exchangeRef)
if exchange != None:
self.validateQueue(queueName, exchange, binding)
# This method validates the specified queue by comparing the queue name and
associated exchange name
# against the whitelist sets. If the exchange name or the queue name is in
the whitelist then we do
# nothing, but if not we generate an alert message.
def validateQueue(self, queueName, exchange, binding):
exchangeName = exchange.name
if exchangeName in self.exchangeWhitelist:
return
if queueName in self.queueWhitelist:
return
connection = self.findConnection(queueName)
address = connection.address
ctime = connection.getTimestamps()[1]/1000000000
if exchangeName == "":
exchangeName = "''"
if binding.arguments:
print strftime("%c", localtime(time())), "ALERT
connection-audit:validateQueue validation failed for queue: %s with binding[%s]
=> %s %s from address: %s with connection timestamp" % (queueName,
binding.bindingKey, exchangeName, binding.arguments, address), strftime("%c",
localtime(ctime))
else:
print strftime("%c", localtime(time())), "ALERT
connection-audit:validateQueue validation failed for queue: %s with binding[%s]
=> %s from address: %s with connection timestamp" % (queueName,
binding.bindingKey, exchangeName, address), strftime("%c", localtime(ctime))
sys.stdout.flush()
# Overridden method from Console
# def event(self, broker, event):
# if event.classKey.getClassName() == "subscribe":
# Overridden method from Console
def brokerConnected(self, broker):
# Log brokerConnected event
print strftime("%c", localtime(time())), "NOTIC
connection-audit:brokerConnected broker=%s" % broker.getUrl()
sys.stdout.flush()
# Overridden method from Console
def brokerConnectionFailed(self, broker):
# Log brokerConnectionFailed event
print strftime("%c", localtime(time())), "NOTIC
connection-audit:brokerConnectionFailed broker=%s %s" % (broker.getUrl(),
str(broker.conn_exc))
sys.stdout.flush()
# Overridden method from Console
def brokerDisconnected(self, broker):
# Clear state as it'll get resent when the broker reconnects
self.queueNameMap.clear()
self.queueNameSubscriptionMap.clear()
self.queueMap.clear()
self.exchangeMap.clear()
self.bindingMap.clear()
self.connectionMap.clear()
self.sessionMap.clear()
# Log brokerDisconnected event
print strftime("%c", localtime(time())), "NOTIC
connection-audit:brokerDisconnected broker=%s" % broker.getUrl()
sys.stdout.flush()
class JHelpFormatter(IndentedHelpFormatter):
"""Format usage and description without stripping newlines from usage
strings
"""
def format_usage(self, usage):
return usage
def format_description(self, description):
if description:
return description + "\n"
else:
return ""
_usage = "%prog [options] [broker-addr]..."
_description = \
"""
Audits connections to one or more Qpid message brokers.
Exchange and Queue names are checked against a whitelist and if no match is
found an alert is generated
If no broker-addr is supplied, %prog connects to 'localhost:5672'.
[broker-addr] syntax:
[username/password@] hostname
ip-address [:<port>]
Examples:
$ %prog localhost:5672
$ %prog 10.1.1.7:10000
$ %prog guest/guest@broker-host:10000
"""
def main(argv=None):
p = optparse.OptionParser(usage=_usage, description=_description,
formatter=JHelpFormatter())
p.add_option("--sasl-mechanism", action="store", type="string",
metavar="<mech>", help="SASL mechanism for authentication (e.g. EXTERNAL,
ANONYMOUS, PLAIN, CRAM-MD, DIGEST-MD5, GSSAPI). SASL automatically picks the
most secure available mechanism - use this option to override.")
p.add_option("--whitelist", action="store", type="string",
default="./whitelist.xml", metavar="<whitelist XML document>", help="The fully
qualified name of the whitelist XML file, default is ./whitelist.xml")
options, arguments = p.parse_args(args=argv)
if len(arguments) == 0:
arguments.append("localhost")
console = AuditConsole(options.whitelist)
session = Session(console, rcvObjects=True, rcvEvents=True,
userBindings=True, manageConnections=True)
# Register to receive updates for broker:queue objects.
session.bindClass("org.apache.qpid.broker", "queue")
# Register to receive updates for broker:exchange objects.
session.bindClass("org.apache.qpid.broker", "exchange")
# Register to receive updates for broker:binding objects.
session.bindClass("org.apache.qpid.broker", "binding")
# Register to receive updates for broker:connection objects.
session.bindClass("org.apache.qpid.broker", "connection")
# Register to receive updates for broker:session objects.
session.bindClass("org.apache.qpid.broker", "session")
# Register to receive updates for broker:subscription objects.
session.bindClass("org.apache.qpid.broker", "subscription")
# Register to receive updates for broker:subscribe event.
# session.bindEvent("org.apache.qpid.broker", "subscribe")
brokers = []
try:
try:
for host in arguments:
brokers.append(session.addBroker(host, None,
options.sasl_mechanism))
while (True):
sleep(10)
except KeyboardInterrupt:
print
return 0
except Exception, e:
print "Failed: %s - %s" % (e.__class__.__name__, e)
return 1
finally:
while len(brokers):
b = brokers.pop()
session.delBroker(b)
if __name__ == '__main__':
sys.exit(main())
<whitelist>
<exchangeWhitelist>
<exchange>qmf.default.topic</exchange>
<exchange>qmf.default.direct</exchange>
<exchange>qpid.management</exchange>
<exchange>amq.direct</exchange>
<exchange></exchange>
</exchangeWhitelist>
<queueWhitelist>
<queue>testqueue</queue>
</queueWhitelist>
</whitelist>
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
