extend audit trail for Hermes (#75)

* bump go-bits to support durability for dataplane.audit queue

On-behalf-of: SAP <filipp.akinfiev@clyso.com>
Signed-off-by: Filipp Akinfiev <filipp.akinfiev@clyso.com>

* feat(opslog): extend audit trail for Hermes (tenant/read/loop filters)

On-behalf-of: SAP <filipp.akinfiev@clyso.com>
Signed-off-by: Filipp Akinfiev <filipp.akinfiev@clyso.com>

---------

Signed-off-by: Filipp Akinfiev <filipp.akinfiev@clyso.com>

Author: fa-at-pulsit

docs/ops-log.md

@@ -365,10 +365,17 @@ Each S3 operation produces a CADF event with:
 | Variable | Description | Default |
 |----------|-------------|---------|
 | `AUDIT_ENABLED` | Publish to RabbitMQ | `false` |
-| `AUDIT_RABBITMQ_URL` | Connection URL (`amqp://user:pass@host:port`) | |
-| `AUDIT_QUEUE_NAME` | Queue name | `keystone.notifications.info` |
+| `AUDIT_RABBITMQ_URL` | Connection URL (`amqp://host:port`) | |
+| `AUDIT_RABBITMQ_USERNAME` | Username; overrides URL userinfo (e.g. from Vault) | |
+| `AUDIT_RABBITMQ_PASSWORD` | Password; overrides URL userinfo | |
+| `AUDIT_QUEUE_NAME` | Queue name (`dataplane.audit` → durable queue) | `keystone.notifications.info` |
 | `AUDIT_QUEUE_SIZE` | Internal event buffer size | `20` |
 | `AUDIT_DEBUG` | Log published events | `false` |
+| `AUDIT_REQUIRE_TENANT` | Drop events lacking `project_id`/`domain_id` (counted in `prysm_audit_events_dropped_total`) | `true` |
+| `AUDIT_OBSERVER_NAME` | CADF observer name (storage service) | `radosgw` |
+| `AUDIT_REGION` | Static region stamped on events (empty = off) | |
+| `AUDIT_INCLUDE_READS` | Audit reads (get/head/list) too; false = mutations-only | `true` |
+| `AUDIT_SKIP_BUCKETS` | Buckets excluded from audit (comma-list, loop prevention) | `hermes` |
 
 ### Metrics tracking
 

ops-log-k8s-mutating-wh/README.md

@@ -197,6 +197,14 @@ stringData:
 | `AUDIT_QUEUE_NAME`        | Target queue (see durability note below)             | `keystone.notifications.info`    |
 | `AUDIT_QUEUE_SIZE`        | Internal event buffer size                           | `20`                             |
 | `AUDIT_DEBUG`             | Log every published event (verbose)                  | `false`                          |
+| `AUDIT_REQUIRE_TENANT`    | Drop events lacking a project_id/domain_id (counted) | `true`                           |
+| `AUDIT_OBSERVER_NAME`     | CADF observer name (storage service)                 | `radosgw`                        |
+| `AUDIT_REGION`            | Static region stamped on events (empty = off)        | _empty_                          |
+| `AUDIT_INCLUDE_READS`     | Audit reads (get/head/list) too; false = mutations-only | `true`                        |
+| `AUDIT_SKIP_BUCKETS`      | Buckets excluded from audit (comma-list, loop prevention) | `hermes`                    |
+
+> These are non-sensitive — put them in the ConfigMap (`sidecarEnvConfig.config`),
+> not the Secret. Only the RabbitMQ credentials belong in the Secret.
 
 > If `AUDIT_ENABLED=true` but `AUDIT_RABBITMQ_URL` is empty, the sidecar logs a
 > warning and falls back to a no-op auditor — log processing is never blocked.

pkg/commands/producer_ops_log.go

@@ -38,6 +38,11 @@ var (
 	opsAuditQueueName         string
 	opsAuditInternalQueueSize int
 	opsAuditDebug             bool
+	opsAuditRequireTenant     bool
+	opsAuditRegion            string
+	opsAuditObserverName      string
+	opsAuditIncludeReads      bool
+	opsAuditSkipBuckets       string
 
 	// Shortcut config
 	opsTrackEverything bool
@@ -231,6 +236,11 @@ Following this configuration change, the RadosGW will log operations to the file
 				QueueName:         opsAuditQueueName,
 				InternalQueueSize: opsAuditInternalQueueSize,
 				Debug:             opsAuditDebug,
+				RequireTenant:     opsAuditRequireTenant,
+				Region:            opsAuditRegion,
+				ObserverName:      opsAuditObserverName,
+				IncludeReads:      opsAuditIncludeReads,
+				SkipBuckets:       opsAuditSkipBuckets,
 			},
 		}
 
@@ -654,6 +664,11 @@ func mergeOpsLogConfigWithEnv(cfg opslog.OpsLogConfig) opslog.OpsLogConfig {
 	cfg.AuditSink.RabbitMQUsername = getEnv("AUDIT_RABBITMQ_USERNAME", cfg.AuditSink.RabbitMQUsername)
 	cfg.AuditSink.RabbitMQPassword = getEnv("AUDIT_RABBITMQ_PASSWORD", cfg.AuditSink.RabbitMQPassword)
 	cfg.AuditSink.QueueName = getEnv("AUDIT_QUEUE_NAME", cfg.AuditSink.QueueName)
+	cfg.AuditSink.RequireTenant = getEnvBool("AUDIT_REQUIRE_TENANT", cfg.AuditSink.RequireTenant)
+	cfg.AuditSink.Region = getEnv("AUDIT_REGION", cfg.AuditSink.Region)
+	cfg.AuditSink.ObserverName = getEnv("AUDIT_OBSERVER_NAME", cfg.AuditSink.ObserverName)
+	cfg.AuditSink.IncludeReads = getEnvBool("AUDIT_INCLUDE_READS", cfg.AuditSink.IncludeReads)
+	cfg.AuditSink.SkipBuckets = getEnv("AUDIT_SKIP_BUCKETS", cfg.AuditSink.SkipBuckets)
 	cfg.AuditSink.InternalQueueSize = getEnvInt("AUDIT_QUEUE_SIZE", cfg.AuditSink.InternalQueueSize)
 	cfg.AuditSink.Debug = getEnvBool("AUDIT_DEBUG", cfg.AuditSink.Debug)
 
@@ -684,6 +699,11 @@ func init() {
 	opsLogCmd.Flags().StringVar(&opsAuditQueueName, "audit-queue-name", "keystone.notifications.info", "RabbitMQ queue name for audit events")
 	opsLogCmd.Flags().IntVar(&opsAuditInternalQueueSize, "audit-queue-size", 20, "Internal queue size for audit events")
 	opsLogCmd.Flags().BoolVar(&opsAuditDebug, "audit-debug", false, "Log published audit events for debugging")
+	opsLogCmd.Flags().BoolVar(&opsAuditRequireTenant, "audit-require-tenant", true, "Drop audit events that have neither a project_id nor a domain_id (the audit consumer rejects them)")
+	opsLogCmd.Flags().StringVar(&opsAuditRegion, "audit-region", "", "Static region stamped onto each audit event (the ops log has none); empty = not stamped")
+	opsLogCmd.Flags().StringVar(&opsAuditObserverName, "audit-observer-name", "radosgw", "CADF observer name identifying the storage service in audit events (e.g. radosgw/ceph/swift)")
+	opsLogCmd.Flags().BoolVar(&opsAuditIncludeReads, "audit-include-reads", true, "Audit read operations (get/head/list); default true for object-storage data-access auditing. Set false for mutations-only")
+	opsLogCmd.Flags().StringVar(&opsAuditSkipBuckets, "audit-skip-buckets", "hermes", "Comma-separated, case-insensitive bucket names excluded from audit (loop prevention for the Hermes audit bucket)")
 
 	// Shortcut flag
 	opsLogCmd.Flags().BoolVar(&opsTrackEverything, "track-everything", false, "Enable detailed tracking for all metric types (efficient mode)")

pkg/commands/producer_ops_log_test.go

@@ -23,6 +23,7 @@ func TestMergeOpsLogConfigWithEnv_AuditSink(t *testing.T) {
 			QueueName:         "keystone.notifications.info",
 			InternalQueueSize: 20,
 			Debug:             false,
+			RequireTenant:     true,
 		},
 	}
 
@@ -34,6 +35,11 @@ func TestMergeOpsLogConfigWithEnv_AuditSink(t *testing.T) {
 		t.Setenv("AUDIT_QUEUE_NAME", "custom.audit.queue")
 		t.Setenv("AUDIT_QUEUE_SIZE", "100")
 		t.Setenv("AUDIT_DEBUG", "true")
+		t.Setenv("AUDIT_REQUIRE_TENANT", "false")
+		t.Setenv("AUDIT_REGION", "qa-de-1")
+		t.Setenv("AUDIT_OBSERVER_NAME", "ceph")
+		t.Setenv("AUDIT_INCLUDE_READS", "true")
+		t.Setenv("AUDIT_SKIP_BUCKETS", "hermes,_default")
 
 		cfg := mergeOpsLogConfigWithEnv(base)
 
@@ -44,6 +50,11 @@ func TestMergeOpsLogConfigWithEnv_AuditSink(t *testing.T) {
 		assert.Equal(t, "custom.audit.queue", cfg.AuditSink.QueueName)
 		assert.Equal(t, 100, cfg.AuditSink.InternalQueueSize)
 		assert.True(t, cfg.AuditSink.Debug)
+		assert.False(t, cfg.AuditSink.RequireTenant)
+		assert.Equal(t, "qa-de-1", cfg.AuditSink.Region)
+		assert.Equal(t, "ceph", cfg.AuditSink.ObserverName)
+		assert.True(t, cfg.AuditSink.IncludeReads)
+		assert.Equal(t, "hermes,_default", cfg.AuditSink.SkipBuckets)
 	})
 
 	t.Run("unset env vars preserve flag defaults", func(t *testing.T) {
@@ -56,5 +67,10 @@ func TestMergeOpsLogConfigWithEnv_AuditSink(t *testing.T) {
 		assert.Equal(t, "keystone.notifications.info", cfg.AuditSink.QueueName)
 		assert.Equal(t, 20, cfg.AuditSink.InternalQueueSize)
 		assert.False(t, cfg.AuditSink.Debug)
+		assert.True(t, cfg.AuditSink.RequireTenant)
+		assert.Equal(t, "", cfg.AuditSink.Region)
+		assert.Equal(t, "", cfg.AuditSink.ObserverName)
+		assert.False(t, cfg.AuditSink.IncludeReads)
+		assert.Equal(t, "", cfg.AuditSink.SkipBuckets)
 	})
 }

pkg/producers/opslog/README.md

@@ -73,11 +73,29 @@ prysm local-producer ops-log [flags]
   connection, client, server).
 - `--audit-enabled` - Enable RabbitMQ audit trail publishing.
 - `--audit-rabbitmq-url` - RabbitMQ connection URL (e.g.,
-  `amqp://user:pass@host:port`).
+  `amqp://host:port`; credentials may be embedded or supplied separately).
+- `--audit-rabbitmq-username` - RabbitMQ username; overrides any userinfo in
+  the URL (e.g. sourced from a Vault entry).
+- `--audit-rabbitmq-password` - RabbitMQ password; overrides any userinfo in
+  the URL.
 - `--audit-queue-name` - RabbitMQ queue name for audit events (default:
-  `keystone.notifications.info`).
+  `keystone.notifications.info`). Set to `dataplane.audit` to get a durable
+  queue (see note below).
 - `--audit-queue-size` - Internal audit event queue size (default: 20).
 - `--audit-debug` - Enable debug logging for published audit events.
+- `--audit-require-tenant` - Drop audit events that have neither a `project_id`
+  nor a `domain_id` (default: true; the customer audit consumer rejects them).
+- `--audit-observer-name` - CADF observer name identifying the storage service
+  in audit events (default: `radosgw`).
+- `--audit-region` - Static region stamped onto each audit event (the ops log
+  has none; default: empty = not stamped).
+- `--audit-include-reads` - Audit read operations (get/head/list) in addition
+  to mutations (default: true, for object-storage data-access auditing). Set
+  false for mutations-only.
+- `--audit-skip-buckets` - Comma-separated, case-insensitive bucket names
+  excluded from audit (default: `hermes`). Breaks the Hermes loop: Hermes writes
+  audit events into this bucket, and auditing those writes would re-trigger
+  events. Empty disables the filter.
 
 ### Latency Tracking Examples:
 
@@ -146,9 +164,16 @@ prysm local-producer ops-log \
 | `TRACK_BUCKET_SLO`           | Enable low-cardinality bucket GET/LIST SLI metrics. |
 | `AUDIT_ENABLED`              | Enable RabbitMQ audit trail publishing.         |
 | `AUDIT_RABBITMQ_URL`         | RabbitMQ connection URL.                        |
-| `AUDIT_QUEUE_NAME`           | RabbitMQ queue name for audit events.           |
+| `AUDIT_RABBITMQ_USERNAME`    | RabbitMQ username; overrides URL userinfo.      |
+| `AUDIT_RABBITMQ_PASSWORD`    | RabbitMQ password; overrides URL userinfo.      |
+| `AUDIT_QUEUE_NAME`           | RabbitMQ queue name (`dataplane.audit` = durable). |
 | `AUDIT_QUEUE_SIZE`           | Internal audit event queue size.                |
 | `AUDIT_DEBUG`                | Enable debug logging for audit events.          |
+| `AUDIT_REQUIRE_TENANT`       | Drop events without a project_id/domain_id (default true). |
+| `AUDIT_OBSERVER_NAME`        | CADF observer (storage service) name (default radosgw). |
+| `AUDIT_REGION`               | Static region stamped on events (empty = off).  |
+| `AUDIT_INCLUDE_READS`        | Audit reads (get/head/list) too (default true). |
+| `AUDIT_SKIP_BUCKETS`         | Buckets excluded from audit, comma-list (default `hermes`). |
 
 #### Request Tracking Environment Variables:
 
@@ -410,6 +435,49 @@ These events are consumed by Hermes and other audit processing systems.
   (development/testing)
 - **Buffered Channel**: 20-event internal queue with automatic retry on
   failures
+- **Tenant Enforcement**: With `AUDIT_REQUIRE_TENANT=true` (default), events
+  without a `project_id` or `domain_id` are dropped before publishing — the
+  customer audit consumer (Hermes) requires a tenant. Dropped events are
+  counted in `prysm_audit_events_dropped_total{reason="no_tenant"}`, not
+  silently discarded.
+
+### Event filtering
+
+Audit emission is gated before publishing. Each drop is counted (not silent) in
+`prysm_audit_events_dropped_total{reason}`:
+
+- **`skip_bucket`** — operations on a bucket listed in `AUDIT_SKIP_BUCKETS`
+  (default `hermes`) are excluded. This breaks the Hermes loop: Hermes writes
+  audit events into a per-customer (WORM) bucket, and auditing those writes
+  would re-trigger events indefinitely. Bucket-name matching is correct here
+  because the audit bucket lives inside each customer's project — only Hermes
+  can write to it (WORM), so nothing legitimate is lost.
+- **`no_tenant`** — events without a `project_id`/`domain_id` (see Tenant
+  Enforcement above).
+- **`read`** — read operations (get/head/list) when `AUDIT_INCLUDE_READS=false`.
+  Reads are audited by default (object-storage data-access events); set the flag
+  false for a mutations-only trail.
+
+### Durable queue
+
+The underlying `go-bits/audittools` library declares the RabbitMQ queue
+**durable** only when `AUDIT_QUEUE_NAME` is exactly `dataplane.audit`; any other
+name is transient. The dataplane audit log-router consumes `dataplane.audit` and
+requires a durable queue, so set `AUDIT_QUEUE_NAME=dataplane.audit` for it to
+connect. Note: the queue survives a broker restart, but messages are still
+published transient. If the queue already exists with a different durability
+flag, delete it first — RabbitMQ rejects a redeclare with `406
+PRECONDITION_FAILED`.
+
+### Observer and region
+
+- **Observer**: events identify the storage service via
+  `observer = { typeURI: "service/storage", name: <AUDIT_OBSERVER_NAME> }`
+  (default name `radosgw`).
+- **Region**: the ops log carries no region, so it can be supplied statically
+  via `AUDIT_REGION` and is stamped onto the target as a `region` attachment.
+  Leave empty to omit. (Placement may change pending audit-consumer
+  confirmation.)
 
 ### CADF Event Structure
 

pkg/producers/opslog/audit.go

@@ -19,6 +19,21 @@ import (
 	"github.com/sapcc/go-bits/audittools"
 )
 
+// buildObserver constructs the CADF observer that identifies the storage
+// service reporting the event (the service, not the resource or the tool).
+// The name is configurable and defaults to "radosgw".
+func buildObserver(cfg AuditSinkConfig) audittools.Observer {
+	name := cfg.ObserverName
+	if name == "" {
+		name = "radosgw"
+	}
+	return audittools.Observer{
+		TypeURI: "service/storage",
+		Name:    name,
+		ID:      audittools.GenerateUUID(),
+	}
+}
+
 // InitAuditor initializes the audit trail based on configuration.
 // Returns a NullAuditor if audit sink is not configured or disabled.
 func InitAuditor(ctx context.Context, cfg AuditSinkConfig, registry prometheus.Registerer) audittools.Auditor {
@@ -46,12 +61,8 @@ func InitAuditor(ctx context.Context, cfg AuditSinkConfig, registry prometheus.R
 	auditor, err := audittools.NewAuditor(ctx, audittools.AuditorOpts{
 		ConnectionURL: connectionURL,
 		QueueName:     cfg.QueueName,
-		Observer: audittools.Observer{
-			TypeURI: "service/storage/object",
-			Name:    "prysm-ops-log",
-			ID:      audittools.GenerateUUID(),
-		},
-		Registry: registry,
+		Observer:      buildObserver(cfg),
+		Registry:      registry,
 	})
 
 	if err != nil {
@@ -97,8 +108,37 @@ func buildRabbitMQConnectionURL(rawURL, username, password string) (string, erro
 	return u.String(), nil
 }
 
-// ToAuditEvent converts an S3OperationLog to an audittools.Event.
-func (opLog *S3OperationLog) ToAuditEvent() (audittools.Event, error) {
+// regionTarget decorates a Target with a static region attachment. Region is
+// a per-cluster deployment fact (the ops log has none), so it is supplied via
+// config rather than derived per request. Implemented as a decorator so the
+// placement can be changed easily if the audit consumer expects it elsewhere.
+type regionTarget struct {
+	inner  audittools.Target
+	region string
+}
+
+func (t regionTarget) Render() cadf.Resource {
+	resource := t.inner.Render()
+	resource.Attachments = append(resource.Attachments, cadf.Attachment{
+		Name:    "region",
+		TypeURI: "xs:string",
+		Content: t.region,
+	})
+	return resource
+}
+
+// withRegion wraps a Target so its rendered resource carries the region. An
+// empty region returns the target unchanged (no attachment added).
+func withRegion(target audittools.Target, region string) audittools.Target {
+	if region == "" {
+		return target
+	}
+	return regionTarget{inner: target, region: region}
+}
+
+// ToAuditEvent converts an S3OperationLog to an audittools.Event. The region is
+// a static per-cluster value stamped onto the target (empty = not stamped).
+func (opLog *S3OperationLog) ToAuditEvent(region string) (audittools.Event, error) {
 	// Parse timestamp
 	eventTime, err := time.Parse("2006-01-02T15:04:05.999999Z", opLog.Time)
 	if err != nil {
@@ -126,7 +166,7 @@ func (opLog *S3OperationLog) ToAuditEvent() (audittools.Event, error) {
 		User:       buildUserInfo(opLog),
 		ReasonCode: reasonCode,
 		Action:     mapOperationToAction(opLog.Operation),
-		Target:     buildTarget(opLog),
+		Target:     withRegion(buildTarget(opLog), region),
 	}, nil
 }
 
@@ -170,6 +210,34 @@ func buildHTTPRequest(opLog *S3OperationLog) (*http.Request, error) {
 	return req, nil
 }
 
+// isSkippedBucket reports whether operations on the given bucket must be
+// excluded from audit. This breaks the Hermes loop: Hermes writes audit events
+// into a per-customer WORM bucket, and those writes would otherwise generate
+// new audit events. Matching is case-insensitive over a comma-separated list;
+// an empty list disables the filter.
+func isSkippedBucket(bucket, skipBuckets string) bool {
+	if bucket == "" || skipBuckets == "" {
+		return false
+	}
+	b := strings.ToLower(strings.TrimSpace(bucket))
+	for _, name := range strings.Split(skipBuckets, ",") {
+		if b == strings.ToLower(strings.TrimSpace(name)) {
+			return true
+		}
+	}
+	return false
+}
+
+// isReadOperation reports whether an RGW operation is a read (get/head/list).
+// Read classification is by operation name and is independent of the CADF
+// action mapping, so it is robust regardless of how actions are finalized.
+// Reads are excluded from the customer audit trail by default (mutations-only).
+func isReadOperation(operation string) bool {
+	return strings.HasPrefix(operation, "get_") ||
+		strings.HasPrefix(operation, "head_") ||
+		strings.HasPrefix(operation, "list_")
+}
+
 // mapOperationToAction converts RadosGW operation names to CADF actions.
 func mapOperationToAction(operation string) cadf.Action {
 	switch operation {
@@ -206,14 +274,38 @@ func buildTarget(opLog *S3OperationLog) audittools.Target {
 			Bucket: opLog.Bucket,
 		}
 	} else {
-		// Account-level operation (e.g., list_buckets)
-		projectID := extractProjectID(opLog.User)
+		// Account-level operation (e.g., list_buckets). Prefer the Keystone
+		// project ID (same source as the initiator) so target and initiator
+		// agree; fall back to the parsed user when no Keystone scope exists.
+		projectID, _ := resolveTenant(opLog)
 		return &AccountTarget{
 			ProjectID: projectID,
 		}
 	}
 }
 
+// resolveTenant returns the project ID and domain ID that would populate the
+// CADF initiator for this entry, using the same precedence as buildUserInfo:
+// the Keystone scope when present, otherwise the project parsed from the user.
+func resolveTenant(opLog *S3OperationLog) (projectID, domainID string) {
+	if opLog.KeystoneScope != nil {
+		return opLog.KeystoneScope.Project.ID, opLog.KeystoneScope.Project.Domain.ID
+	}
+	return extractProjectID(opLog.User), ""
+}
+
+// hasUsableTenant reports whether the entry yields a project_id or domain_id
+// for the CADF initiator. The audit consumer rejects events that carry
+// neither, so AUDIT_REQUIRE_TENANT uses this to drop them before publishing.
+// The anonymous sentinel is treated as no tenant.
+func hasUsableTenant(opLog *S3OperationLog) bool {
+	projectID, domainID := resolveTenant(opLog)
+	if domainID != "" {
+		return true
+	}
+	return projectID != "" && projectID != "anonymous"
+}
+
 // extractProjectID extracts the project ID from the user field.
 // User format is typically: "projectID$projectID" or just "projectID"
 func extractProjectID(user string) string {