Chapter 1. (Preview) Setting up the Record Validation filter

Procedure

1. Configure a filter of type RecordValidation.

   • For an OpenShift deployment, configure a KafkaProtocolFilter resource. See Section 1.1, “Example KafkaProtocolFilter resource”.

2. Replace the <rule definition> token in the YAML configuration with a validation rule that matches your requirements.

   Choose one of the following rule types:

   • Schema validation, to validate record keys or values against a schema stored in an Apicurio Registry.
   • JSON syntax validation, to validate that record keys or values contain syntactically valid JSON.

   • JWS Signature validation, to validate a JSON Web Signature (JWS) provided in a Kafka record header.

     The following examples show how to configure each rule type.

     Example schema validation rule:

     schemaValidationConfig:
       apicurioId: 1001
       apicurioRegistryUrl: http://registry.local:8080
       tls:
         trust:
           storeFile: /opt/proxy/trust/ca.p12
           storePassword:
             passwordFile: /opt/proxy/trust/store.password
           storeType: PKCS12
       wireFormatVersion: V3
     allowNulls: true
     allowEmpty: true

     • apicurioId identifies the schema that the filter enforces. By default, this ID refers to the Apicurio Registry’s contentId. If the record key or value does not conform to the schema identified by this ID, the record is rejected. If a Kafka producer embeds a schema ID in the record (either in a header or magic bytes at the start of the record’s key or value), the filter validates it against this value. The record is rejected if the IDs do not match.
     • apicurioRegistryUrl specifies the endpoint URL for the Apicurio Registry.

     • tls specifies the optional TLS configuration used when connecting securely to the Apicurio Registry.

       • storeFile path to the truststore file that contains the trust anchors used to validate the server connection.
       • storePassword (Optional) password used to protect the truststore.
       • storeType truststore type. Supported values include PKCS12, JKS, and PEM.
     • wireFormatVersion controls whether the Apicurio client operates in V3 (default) or V2 (deprecated). In V3 mode, the apicurioId property identifies schema by content ID, and the filter expects the Kafka producer to use content IDs to identify schema. In V2 mode, the apicurioId property identifies schema by global ID, and the filter expects the Kafka producer to use global IDs to identify schema. V2 mode exists only for compatibility with older Apicurio configurations and is deprecated. For new applications, use V3.
     • allowNulls specifies whether the validator allows record keys or values to be null. The default is false.
     • allowEmpty specifies whether the validator allows record keys or values to be empty. The default is false.

     For more information about embedding the ID in a record, see the Apicurio documentation.

     Note

     Schema validation currently supports enforcing JSON schemas only (issue).

     Example JSON syntax validation rule:

     syntacticallyCorrectJson:
       validateObjectKeysUnique: true
     allowNulls: true
     allowEmpty: true

     • syntacticallyCorrectJson.validateObjectKeysUnique specifies whether JSON object keys must be unique. The default is false.
     • allowNulls specifies whether the validator allows record keys or values to be null. The default is false.
     • allowEmpty specifies whether the validator allows record keys or values to be empty. The default is false.

     This rule rejects records that contain syntactically invalid JSON in the record key or value.

     Example JWS Signature validation rule:

     jwsSignatureValidation:
       trustedJsonWebKeySet: |
         {
           "keys": [
             {
               "kty": "RSA",
               "kid": "RSA JWK",
               "key_ops": [ "verify" ],
               "alg": "RS256",
               "n": "lx-5h_lkVJGc8-NxgKGkfxobBlM7DDCjqAloEieCf8c9KbPJ12V0Bm8arOX-lm0wRSLxmWzQ3NTM6S9pDbSc7fQt77THlMD1zEDMwIAJxfoMt3U_VrMHVdiOvO6YpU3jWBp7BfQNYHJjCSiaxIHqDPuDCh2GTflkU32Nfim7W3iLBuH4_K8ADYHz3QdeX29vzl49PDbUiiJEsjnEconsfuYjrFaN3uGn41mGzjedc7oxlcID8fDxq5bvZvOFBAIqdrxs5qPYw1QlVupB3LT0AMU1qKWOB_vkM1n54eDIxP9Xd__WnvX4wagYqA1OZ9LPBenhBX7_Rbnrap2QNQ58-Q",
               "e": "AQAB"
             },
             {
               "kty": "EC",
               "kid": "EC JWK",
               "key_ops": [ "verify" ],
               "crv": "P-256",
               "alg": "ES256",
               "x": "f83OJ3D2xF4h2Jm2VvZd7uKQ6GFDp4zQpN9qzZPJxG8",
               "y": "x_FEzRu9PZxC5H5KJ5-3HqWQX5J8V4a5j8nY1JvG9r0"
             }
           ]
         }
       algorithms:
         allowed:
           - ES256
           - RS256
       recordHeader:
         key: kroxylicious.io/jws
         required: true
       content:
         detached: false

     • jwsSignatureValidation.trustedJsonWebKeySet specifies the set of JSON Web Keys (JWK) used to verify JSON Web Signature (JWS) signatures.
     • The JWS Signature validation rule verifies the JWS only and does not perform JSON Web Token (JWT) claim validation. Token claims such as expiration and issuer are not checked.

     • jwsSignatureValidation.algorithms.allowed and jwsSignatureValidation.algorithms.denied specify which signature algorithms are permitted.

       • If both allowed and denied are omitted or empty, all algorithms are blocked and JWS Signature verification fails.
       • If only allowed is specified, only the algorithms listed in allowed are permitted.
       • If only denied is specified, all algorithms except those listed in denied are permitted.
       • If both allowed and denied are specified, only the algorithms listed in allowed are permitted.
     • jwsSignatureValidation.recordHeader.key specifies the Kafka record header key that contains the JWS. The default value is kroxylicious.io/jws.
     • jwsSignatureValidation.recordHeader.required specifies whether the JWS record header must be present. If set to false, validation is skipped when the header is missing.
     • jwsSignatureValidation.content.detached specifies whether the record buffer is used as the JWS payload (detached content, as defined in RFC 7515 Appendix F).