Add tenant ids and remove olso

auditmiddleware/__init__.py

@@ -44,7 +44,9 @@
     cfg.StrOpt('driver',
                help='The Driver to handle sending notifications. Possible '
                     'values are messaging, messagingv2, routing, log, test, '
-                    'noop. If not specified, then value from '
+                    'noop, raw_amqp. Use raw_amqp to publish events as plain '
+                    'JSON directly to AMQP without any oslo envelope. If not '
+                    'specified, then value from '
                     'oslo_messaging_notifications conf section is used.'),
     cfg.ListOpt('topics',
                 help='List of AMQP topics used for OpenStack notifications. If'
@@ -96,6 +98,26 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
+def _tenant_ids(event):
+    """Build the tenant_ids list for a CADF event.
+
+    Collects distinct project IDs from initiator and target so that
+    downstream consumers (log-router, OpenSearch) can route and filter
+    events per tenant without having to know which field carries the
+    relevant project ID.
+    """
+    ids = []
+    initiator_project = getattr(getattr(event, 'initiator', None),
+                                'project_id', None)
+    if initiator_project:
+        ids.append(initiator_project)
+    target_project = getattr(getattr(event, 'target', None),
+                             'project_id', None)
+    if target_project and target_project not in ids:
+        ids.append(target_project)
+    return ids if ids else ['Default']
+
+
 class ConfigError(BaseException):
     """Exception for configuration errors."""
 
@@ -146,7 +168,9 @@ def _process_request(self, request, response=None):
             request.environ['audit.context'] = {}
             for e in events:
                 ctx = request.environ['audit.context']
-                self._notifier.notify(ctx, e.as_dict())
+                event_dict = e.as_dict()
+                event_dict['tenant_ids'] = _tenant_ids(e)
+                self._notifier.notify(ctx, event_dict)
 
     @webob.dec.wsgify
     def __call__(self, req):

auditmiddleware/_notifier.py

@@ -10,18 +10,25 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
-"""Provides a oslo-messaging and a log-based messaging connector."""
+"""Provides a oslo-messaging, log-based, and raw-AMQP messaging connector."""
 
+import json
 import os
 import queue
 import sys
 from threading import Thread
+from urllib.parse import urlparse
 
 try:
     import oslo_messaging
 except ImportError:
     oslo_messaging = None
 
+try:
+    import kombu
+except ImportError:
+    kombu = None
+
 
 class _LogNotifier(object):
     """A notifier that dumps the event content into the log.
@@ -34,10 +41,7 @@ def __init__(self, log):
         self._log = log
 
     def notify(self, context, payload):
-        self._log.info('Event type: audit.cadf, Context: %(context)s, '
-                       'Payload: %(payload)s',
-                       {'context': context, 'event_type': 'audit.cadf',
-                        'payload': payload})
+        self._log.info(json.dumps(payload))
 
 
 class _MessagingNotifier(Thread):
@@ -158,8 +162,116 @@ def flush_to_log(self):
             pass
 
 
+class _RawAmqpNotifier(Thread):
+    """Publishes CADF events as plain JSON directly to an AMQP exchange.
+
+    Unlike _MessagingNotifier, this bypasses oslo_messaging entirely so the
+    message body on the wire is just the CADF dict serialised as JSON — no
+    oslo envelope, no double-wrapped payload field.
+
+    Wire format:
+        routing_key : <topic>.info   (e.g. "notifications.info")
+        content_type: application/json
+        body        : json.dumps(cadf_event)
+    """
+
+    def __init__(self, transport_url, topics, log, mem_queue_size,
+                 metrics_enabled):
+        super(_RawAmqpNotifier, self).__init__(
+            name='async auditmiddleware raw-amqp notifications')
+        self._log = log
+        self._transport_url = transport_url
+        self._topics = topics or ['notifications']
+        self._queue_capacity = mem_queue_size
+        self._queue = queue.Queue(mem_queue_size)
+        self._statsd = None
+        if metrics_enabled:
+            try:
+                import datadog
+                self._statsd = datadog.dogstatsd.DogStatsd(
+                    host=os.getenv('STATSD_HOST', 'localhost'),
+                    port=int(os.getenv('STATSD_PORT', '8125')),
+                    namespace='openstack_audit_messaging')
+            except ImportError:
+                self._log.warning("Python datadog package not installed. "
+                                  "No openstack_audit_* metrics will be "
+                                  "produced.")
+
+    def notify(self, context, payload):
+        try:
+            self._queue.put((payload, context), timeout=1)
+        except queue.Full:
+            self._log.error("Audit events could not be delivered (buffer "
+                            "full). Payload follows ...")
+            self._flush_to_log()
+            self._log_event(context, payload)
+
+    def _log_event(self, context, payload):
+        self._log.info('Event type: audit.cadf, Context: %(context)s, '
+                       'Payload: %(payload)s',
+                       {'context': context, 'event_type': 'audit.cadf',
+                        'payload': payload})
+
+    def _flush_to_log(self):
+        try:
+            while True:
+                payload, context = self._queue.get_nowait()
+                self._log_event(context, payload)
+        except queue.Empty:
+            pass
+
+    def _make_connection(self):
+        url = self._transport_url
+        if url and url.startswith('rabbit://'):
+            url = 'amqp://' + url[len('rabbit://'):]
+        return kombu.Connection(url)
+
+    def run(self):
+        while True:
+            try:
+                payload, context = self._queue.get()
+                self._publish(payload)
+                self._log.debug("Push event (raw): %s", payload.get("id"))
+            except Exception:
+                self._log.error("Cannot push raw audit event to AMQP: %s",
+                                str(sys.exc_info()[0]))
+                if 'payload' in dir():
+                    self._log_event(context, payload)
+
+    def _publish(self, payload):
+        body = json.dumps(payload, ensure_ascii=False).encode('utf-8')
+        with self._make_connection() as conn:
+            conn.ensure_connection()
+            for topic in self._topics:
+                routing_key = '%s.info' % topic
+                exchange = kombu.Exchange(
+                    topic, type='topic', durable=False, auto_delete=False)
+                with kombu.Producer(conn.channel(), exchange=exchange,
+                                    routing_key=routing_key) as producer:
+                    producer.publish(
+                        body,
+                        content_type='application/json',
+                        content_encoding='utf-8',
+                    )
+
+
 def create_notifier(conf, log, metrics_enabled):
     """Create a new notifier."""
+    driver = getattr(conf.audit_middleware_notifications, 'driver', None)
+
+    if driver == 'raw_amqp':
+        if kombu is None:
+            raise ImportError(
+                "kombu is required for driver=raw_amqp but is not installed")
+        transport_url = conf.audit_middleware_notifications.transport_url
+        topics = conf.audit_middleware_notifications.topics or ['notifications']
+        mqs = conf.audit_middleware_notifications.mem_queue_size or 10000
+        notf = _RawAmqpNotifier(transport_url, topics, log, mqs,
+                                metrics_enabled)
+        notf.daemon = True
+        notf.start()
+        return notf
+
     if oslo_messaging and conf.audit_middleware_notifications.get(
         'use_oslo_messaging'):
         transport = oslo_messaging.get_notification_transport(

auditmiddleware/tests/unit/test_audit_middleware.py

@@ -169,6 +169,42 @@ def test_no_response(self):
         payload = self.notifier.notify.call_args_list[0][0][1]
         self.assertEqual(payload['outcome'], 'unknown')
         self.assertNotIn('reason', payload)
+
+    def test_tenant_ids_in_event(self):
+        """Test that tenant_ids is injected into every published event.
+
+        tenant_ids must be present so that log-router and OpenSearch can
+        route and filter events per tenant without knowing which field
+        carries the relevant project ID.
+        """
+        path = '/v2/' + self.project_id + '/servers'
+        self.create_simple_app().get(path,
+                                     extra_environ=self.get_environ_header())
+
+        self.assertEqual(1, self.notifier.notify.call_count)
+        payload = self.notifier.notify.call_args_list[0][0][1]
+
+        self.assertIn('tenant_ids', payload)
+        self.assertIsInstance(payload['tenant_ids'], list)
+        self.assertGreater(len(payload['tenant_ids']), 0)
+        # the initiator project_id from the request headers must be present
+        self.assertIn(self.project_id, payload['tenant_ids'])
+
+    def test_tenant_ids_default_when_no_project(self):
+        """Test that tenant_ids falls back to ['Default'] with no project."""
+        path = '/v2/' + self.project_id + '/servers'
+        # omit HTTP_X_PROJECT_ID so initiator.project_id is absent
+        environ = {'HTTP_X_USER_ID': self.user_id,
+                   'HTTP_X_USER_NAME': self.username,
+                   'HTTP_X_AUTH_TOKEN': 'token',
+                   'HTTP_X_IDENTITY_STATUS': 'Confirmed'}
+        self.create_simple_app().get(path, extra_environ=environ)
+
+        self.assertEqual(1, self.notifier.notify.call_count)
+        payload = self.notifier.notify.call_args_list[0][0][1]
+
+        self.assertIn('tenant_ids', payload)
+        self.assertEqual(['Default'], payload['tenant_ids'])
         # self.assertEqual(len(payload['reporterchain']), 1)
         # self.assertEqual(payload['reporterchain'][0]['role'], 'modifier')
         # self.assertEqual(payload['reporterchain'][0]['reporter']['id'],

auditmiddleware/tests/unit/test_audit_oslo_messaging.py

@@ -163,3 +163,51 @@ def test_do_not_use_oslo_messaging(self):
         # make sure it is using a local notifier instead of oslo_messaging
         self.assertIsInstance(audit_middleware._notifier,
                               _notifier._LogNotifier)
+
+
+class RawAmqpNotifierTest(base.BaseAuditMiddlewareTest):
+    """Tests for _RawAmqpNotifier — plain JSON published without oslo envelope."""
+
+    def test_raw_amqp_driver_selected(self):
+        """driver=raw_amqp must create a _RawAmqpNotifier."""
+        transport_url = 'memory://'
+        self.cfg.config(driver='raw_amqp',
+                        transport_url=transport_url,
+                        group='audit_middleware_notifications')
+        middleware = self.create_simple_middleware()
+        self.assertIsInstance(middleware._notifier, _notifier._RawAmqpNotifier)
+
+    def test_raw_amqp_publishes_plain_json(self):
+        """Events must land on the broker as plain JSON with no oslo wrapper."""
+        import json as _json
+        transport_url = 'memory://'
+        topic = 'notifications'
+        self.cfg.config(driver='raw_amqp',
+                        transport_url=transport_url,
+                        topics=[topic],
+                        group='audit_middleware_notifications')
+
+        published = []
+
+        def _capture_publish(body, content_type, content_encoding, **kw):
+            published.append((body, content_type))
+
+        with mock.patch(
+                'kombu.Producer.publish',
+                side_effect=_capture_publish):
+            app = self.create_simple_app()
+            path = '/v2/' + self.project_id + '/servers'
+            app.get(path, extra_environ=self.get_environ_header())
+            import time; time.sleep(0.5)
+
+        self.assertEqual(1, len(published), 'expected exactly one publish call')
+        body, content_type = published[0]
+        self.assertEqual('application/json', content_type)
+
+        # body must be plain JSON — no oslo.message or payload wrapper
+        event = _json.loads(body)
+        self.assertNotIn('oslo.message', event)
+        self.assertNotIn('oslo.version', event)
+        self.assertNotIn('payload', event)
+        self.assertIn('typeURI', event)
+        self.assertIn('id', event)

auditmiddleware/tests/unit/test_logging_notifier.py

@@ -29,7 +29,8 @@ def setUp(self):
         super(TestLoggingNotifier, self).setUp()
 
     def test_api_request_no_messaging(self):
-        """Test that logging works and event_type is correct."""
+        """Test that logging works and event is plain JSON."""
+        import json as _json
         self.cfg.config(use_oslo_messaging=False,
                         group='audit_middleware_notifications')
         app = self.create_simple_app()
@@ -38,10 +39,12 @@ def test_api_request_no_messaging(self):
             path = '/v2/' + self.project_id + '/servers'
             app.get(path, extra_environ=self.get_environ_header())
 
-            # Ensure that log.info was called
             self.assertGreater(len(log.call_args_list), 0,
                                 "log.info was not called")
 
-            # Check notification'
-            call_args = log.call_args_list[0][0]
-            self.assertEqual('audit.cadf', call_args[1]['event_type'])
+            # body is plain JSON — no oslo wrapper
+            body = log.call_args_list[0][0][0]
+            event = _json.loads(body)
+            self.assertIn('typeURI', event)
+            self.assertNotIn('event_type', event)
+            self.assertNotIn('oslo.message', event)

docker-demo/Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install dependencies first (better layer caching)
+COPY requirements.txt /tmp/requirements.txt
+RUN pip install --no-cache-dir -r /tmp/requirements.txt gunicorn
+
+# Copy the middleware source — install as a directory package (no setup.py needed)
+COPY auditmiddleware /app/auditmiddleware
+COPY etc /app/etc
+
+# Copy the demo app
+COPY docker-demo/demo_app.py /app/demo_app.py
+
+EXPOSE 8080
+
+CMD ["gunicorn", "--bind", "0.0.0.0:8080", "--workers", "1", "--access-logfile", "-", "demo_app:app"]