Production Hardening

# All policies in the Production Hardening bundle (26 resources)
# Save as bundle-production-hardening.yaml then: conduktor apply -f bundle-production-hardening.yaml
# Each ResourcePolicy must still be linked via Application(Instance).spec.policyRef
# or KafkaCluster.spec.policiesRef to take effect.
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: topic-name-convention
spec:
  targetKind: Topic
  description: Topic names must follow <env>.<domain>.<entity>.<version>
  rules:
    - condition: metadata.name.matches("^(dev|staging|prod)\\.[a-z0-9-]+\\.[a-z0-9-]+\\.v[0-9]+$")
      errorMessage: "Topic name must match <env>.<domain>.<entity>.<version> (e.g. prod.orders.placed.v1)"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: consumer-group-prefix
spec:
  targetKind: ApplicationGroup
  description: Consumer-group / ApplicationGroup names must be <team>.<app>.cg
  rules:
    - condition: metadata.name.matches("^[a-z][a-z0-9-]+\\.[a-z0-9-]+\\.cg$")
      errorMessage: "Consumer group must be <team>.<app>.cg (e.g. orders.fraud-detector.cg)"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: partition-count-bounds
spec:
  targetKind: Topic
  description: partitions must be in [1, 2, 3, 6, 12] (override via labels.partition-override-approved=true)
  rules:
    - condition: '(has(metadata.labels) && "partition-override-approved" in metadata.labels && metadata.labels["partition-override-approved"] == "true") || spec.partitions in [1, 2, 3, 6, 12]'
      errorMessage: "Partitions must be one of [1, 2, 3, 6, 12] (set label partition-override-approved=true to request an override)"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: min-replication-factor
spec:
  targetKind: Topic
  description: Production topics must have replication factor >= 3
  rules:
    - condition: '!metadata.name.startsWith("prod.") || spec.replicationFactor >= 3'
      errorMessage: "Production topics must have replication factor >= 3"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: max-retention-bound
spec:
  targetKind: Topic
  description: retention.ms must be <= 30d and not infinite, unless labels.retention-justified == "true"
  rules:
    - condition: '(has(metadata.labels) && "retention-justified" in metadata.labels && metadata.labels["retention-justified"] == "true") || ("retention.ms" in spec.configs && int(string(spec.configs["retention.ms"])) != -1 && int(string(spec.configs["retention.ms"])) <= 2592000000)'
      errorMessage: "retention.ms must be <= 30 days and not -1 (set label retention-justified=true to override)"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: topic-compression-passthrough
spec:
  targetKind: Topic
  description: compression.type on the topic must be "producer" (or unset). Compression is enforced producer-side via Gateway.
  rules:
    - condition: '!has(spec.configs) || !("compression.type" in spec.configs) || spec.configs["compression.type"] == "producer"'
      errorMessage: "compression.type must be \"producer\" (or unset) — broker recompression wastes CPU. Enforce lz4/zstd at the producer via the Gateway producer-compression-required interceptor."
---
apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: schema-id-required-on-produce
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.safeguard.SchemaPayloadValidationPolicyPlugin
  priority: 100
  config:
    topic: "^(?!__|_).+$"
    schemaIdRequired: true
    validateSchema: false
    action: BLOCK
    schemaRegistryConfig:
      host: "${SR_URL}"
      type: CONFLUENT
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: no-wildcard-acl-prod
spec:
  targetKind: ApplicationGroup
  description: ApplicationGroups cannot grant wildcard resource permissions (name="*") regardless of patternType
  rules:
    - condition: 'spec.permissions.all(p, p.name != "*")'
      errorMessage: "Wildcard resource name=\"*\" is not allowed (LITERAL or PREFIXED) — scope by an explicit prefix instead"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: max-message-bytes-bound
spec:
  targetKind: Topic
  description: max.message.bytes must be <= 10 MB
  rules:
    - condition: '!("max.message.bytes" in spec.configs) || int(string(spec.configs["max.message.bytes"])) <= 10485760'
      errorMessage: "max.message.bytes > 10 MB — consider claim-check pattern with S3"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: min-insync-replicas-bound
spec:
  targetKind: Topic
  description: min.insync.replicas must be >= 2 and <= replicationFactor - 1
  rules:
    - condition: '"min.insync.replicas" in spec.configs && int(string(spec.configs["min.insync.replicas"])) >= 2 && int(string(spec.configs["min.insync.replicas"])) <= int(spec.replicationFactor) - 1'
      errorMessage: "min.insync.replicas must be >= 2 and <= replicationFactor - 1 (durability vs availability tradeoff)"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: partition-count-power-aligned
spec:
  targetKind: Topic
  description: partitions must be one of [1, 3, 6, 12, 24, 48, 96]
  rules:
    - condition: 'spec.partitions in [1, 3, 6, 12, 24, 48, 96]'
      errorMessage: "partitions must be one of [1, 3, 6, 12, 24, 48, 96] — odd or prime counts cause permanent consumer-group skew"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: retention-bytes-required
spec:
  targetKind: Topic
  description: delete-policy topics must declare retention.bytes > 0
  rules:
    - condition: '!("cleanup.policy" in spec.configs) || !spec.configs["cleanup.policy"].contains("delete") || ("retention.bytes" in spec.configs && int(string(spec.configs["retention.bytes"])) > 0)'
      errorMessage: "retention.bytes must be set (>0) on delete-policy topics — time-only retention is unbounded in disk"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: compact-topic-requires-tombstone-retention
spec:
  targetKind: Topic
  description: compacted topics must declare delete.retention.ms in [1h, 30d]
  rules:
    - condition: '!("cleanup.policy" in spec.configs) || !spec.configs["cleanup.policy"].contains("compact") || ("delete.retention.ms" in spec.configs && int(string(spec.configs["delete.retention.ms"])) >= 3600000 && int(string(spec.configs["delete.retention.ms"])) <= 2592000000)'
      errorMessage: "compacted topics must declare delete.retention.ms in [1h, 30d] — default 24h often breaks slow consumers"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: segment-bytes-sane-bound
spec:
  targetKind: Topic
  description: segment.bytes must be in [64 MiB, 2 GiB]
  rules:
    - condition: '!("segment.bytes" in spec.configs) || (int(string(spec.configs["segment.bytes"])) >= 67108864 && int(string(spec.configs["segment.bytes"])) <= 2147483648)'
      errorMessage: "segment.bytes must be between 64MiB and 2GiB — tiny segments blow up file count, huge segments delay retention"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: topic-owner-label-required
spec:
  targetKind: Topic
  description: every topic must carry owner (email) and data-criticality (C0..C3) labels
  rules:
    - condition: 'has(metadata.labels.owner) && metadata.labels.owner.matches("^[a-z0-9._-]+@[a-z0-9.-]+\\.[a-z]{2,}$") && "data-criticality" in metadata.labels && metadata.labels["data-criticality"] in ["C0", "C1", "C2", "C3"]'
      errorMessage: "metadata.labels.owner (email) and metadata.labels[\"data-criticality\"] (C0..C3) are required — unowned topics become stale and unpageable"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: topic-domain-prefix-matches-app
spec:
  targetKind: Topic
  description: topic name must start with "<domain>." matching metadata.labels.domain
  rules:
    - condition: 'has(metadata.labels.domain) && metadata.name.startsWith(metadata.labels.domain + ".")'
      errorMessage: "topic name must start with \"<domain>.\" matching metadata.labels.domain (keeps naming, ACL prefix and chargeback in sync)"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: connector-class-allowlist
spec:
  targetKind: Connector
  description: connector.class must be on the vetted allowlist
  rules:
    - condition: |
        spec.config["connector.class"] in [
          "io.confluent.connect.jdbc.JdbcSinkConnector",
          "io.confluent.connect.jdbc.JdbcSourceConnector",
          "io.debezium.connector.postgresql.PostgresConnector",
          "io.debezium.connector.mysql.MySqlConnector",
          "io.confluent.connect.s3.S3SinkConnector",
          "org.apache.kafka.connect.mirror.MirrorSourceConnector",
          "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector"
        ]
      errorMessage: "connector.class is not on the vetted allowlist — ask the platform team to certify the plugin first"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: connector-tasks-max-bound
spec:
  targetKind: Connector
  description: tasks.max must be in [1, 16]
  rules:
    - condition: '"tasks.max" in spec.config && int(string(spec.config["tasks.max"])) >= 1 && int(string(spec.config["tasks.max"])) <= 16'
      errorMessage: "tasks.max must be in [1, 16] — higher values starve other tenants on the Connect cluster"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: connector-error-tolerance-bounded
spec:
  targetKind: Connector
  description: errors.tolerance=all requires errors.deadletterqueue.topic.name to be set
  rules:
    - condition: '!("errors.tolerance" in spec.config) || spec.config["errors.tolerance"] != "all" || ("errors.deadletterqueue.topic.name" in spec.config && spec.config["errors.deadletterqueue.topic.name"] != "")'
      errorMessage: "errors.tolerance=all requires errors.deadletterqueue.topic.name — otherwise poison records are silently dropped"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: subject-naming-strategy-topic-value
spec:
  targetKind: Subject
  description: subject name must end with -key or -value (TopicNameStrategy)
  rules:
    - condition: 'metadata.name.endsWith("-key") || metadata.name.endsWith("-value")'
      errorMessage: "subject name must end with \"-key\" or \"-value\" (TopicNameStrategy) — other suffixes break auto-resolution in standard SR clients"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: subject-compatibility-not-none
spec:
  targetKind: Subject
  description: spec.compatibility must be set and not NONE
  rules:
    - condition: 'has(spec.compatibility) && spec.compatibility in ["BACKWARD", "BACKWARD_TRANSITIVE", "FORWARD", "FORWARD_TRANSITIVE", "FULL", "FULL_TRANSITIVE"]'
      errorMessage: "spec.compatibility must be explicitly set and not NONE — NONE allows arbitrary schema changes that break downstream consumers"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: applicationgroup-no-wildcard-write
spec:
  targetKind: ApplicationGroup
  description: wildcard (name="*") permissions cannot include write/create/delete on topics or connectors
  rules:
    - condition: 'spec.permissions.all(p, p.name != "*" || !p.permissions.exists(perm, perm in ["topicProduce", "topicCreate", "topicDelete", "kafkaConnectorCreate", "kafkaConnectorUpdate", "kafkaConnectorDelete"]))'
      errorMessage: "wildcard (name=\"*\") permissions cannot include write/create/delete on topics or connectors — scope by prefix instead"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: applicationgroup-prod-requires-external-group
spec:
  targetKind: ApplicationGroup
  description: production ApplicationGroups must use externalGroups/externalGroupRegex (SSO sync)
  rules:
    - condition: '!spec.permissions.exists(p, p.appInstance.matches(".*-(prod|prd)$")) || (size(spec.members) == 0 && ((has(spec.externalGroups) && size(spec.externalGroups) > 0) || (has(spec.externalGroupRegex) && size(spec.externalGroupRegex) > 0)))'
      errorMessage: "production ApplicationGroups must use externalGroups/externalGroupRegex (SSO sync) — manual spec.members lists rot when staff leave"
---
apiVersion: self-serve/v1
kind: ResourcePolicy
metadata:
  name: applicationgroup-no-subject-wildcard-read
spec:
  targetKind: ApplicationGroup
  description: SUBJECT permissions must be prefix-scoped (patternType=PREFIXED, name!="*")
  rules:
    - condition: 'spec.permissions.all(p, p.resourceType != "SUBJECT" || (p.name != "*" && p.patternType == "PREFIXED"))'
      errorMessage: "SUBJECT permissions must be prefix-scoped (patternType=PREFIXED, name!=\"*\") — schema field names leak PII structure"
---
apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: producer-idempotence-acks-all
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.safeguard.ProducePolicyPlugin
  priority: 100
  config:
    topic: ".*"
    acks:
      value:
        - -1
      action: BLOCK
    idempotenceRequired:
      value: true
      action: BLOCK
---
apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: producer-compression-required
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.safeguard.ProducePolicyPlugin
  priority: 100
  config:
    topic: ".*"
    compressions:
      value:
        - LZ4
        - ZSTD
      action: BLOCK