DGS-7866: Disallow properties starting with $ by default in JsonSchema

confluentinc · Sep 6, 2023 · e8fa743 · e8fa743
1 parent 992168b
commit e8fa743
Show file tree

Hide file tree

Showing 14 changed files with 126 additions and 25 deletions.
diff --git a/client/src/main/java/io/confluent/kafka/schemaregistry/ParsedSchema.java b/client/src/main/java/io/confluent/kafka/schemaregistry/ParsedSchema.java
@@ -180,7 +180,7 @@ default ParsedSchema normalize() {
    * Validates the schema and ensures all references are resolved properly.
    * Throws an exception if the schema is not valid.
    */
-  default void validate() {
+  default void validate(boolean strict) {
   }
 
   /**

diff --git a/client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/entities/Config.java b/client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/entities/Config.java
@@ -32,6 +32,8 @@ public class Config {
 
   private String alias;
   private Boolean normalize;
+
+  private Boolean validateFields;
   private String compatibilityLevel;
   private String compatibilityGroup;
   private Metadata defaultMetadata;
@@ -42,6 +44,7 @@ public class Config {
   @JsonCreator
   public Config(@JsonProperty("alias") String alias,
                 @JsonProperty("normalize") Boolean normalize,
+                @JsonProperty("validateFields") Boolean validateFields,
                 @JsonProperty("compatibilityLevel") String compatibilityLevel,
                 @JsonProperty("compatibilityGroup") String compatibilityGroup,
                 @JsonProperty("defaultMetadata") Metadata defaultMetadata,
@@ -50,6 +53,7 @@ public Config(@JsonProperty("alias") String alias,
                 @JsonProperty("overrideRuleSet") RuleSet overrideRuleSet) {
     this.alias = alias;
     this.normalize = normalize;
+    this.validateFields = validateFields;
     this.compatibilityLevel = compatibilityLevel;
     this.compatibilityGroup = compatibilityGroup;
     this.defaultMetadata = defaultMetadata;
@@ -68,6 +72,7 @@ public Config() {
   public Config(ConfigUpdateRequest request) {
     this.alias = request.getAlias();
     this.normalize = request.isNormalize();
+    this.validateFields = request.isValidateFields();
     this.compatibilityLevel = request.getCompatibilityLevel();
     this.compatibilityGroup = request.getCompatibilityGroup();
     this.defaultMetadata = request.getDefaultMetadata();
@@ -96,6 +101,16 @@ public void setNormalize(Boolean normalize) {
     this.normalize = normalize;
   }
 
+  @JsonProperty("validateFields")
+  public Boolean isValidateFields() {
+    return validateFields;
+  }
+
+  @JsonProperty("validateFields")
+  public void setValidateFields(Boolean validateFieldNames) {
+    this.validateFields = validateFieldNames;
+  }
+
   @Schema(description = "Compatibility Level",
       allowableValues = {"BACKWARD", "BACKWARD_TRANSITIVE", "FORWARD", "FORWARD_TRANSITIVE",
         "FULL", "FULL_TRANSITIVE", "NONE"},
@@ -179,6 +194,7 @@ public boolean equals(Object o) {
     Config config = (Config) o;
     return Objects.equals(alias, config.alias)
         && Objects.equals(normalize, config.normalize)
+        && Objects.equals(validateFields, config.validateFields)
         && Objects.equals(compatibilityLevel, config.compatibilityLevel)
         && Objects.equals(compatibilityGroup, config.compatibilityGroup)
         && Objects.equals(defaultMetadata, config.defaultMetadata)
@@ -189,7 +205,7 @@ public boolean equals(Object o) {
 
   @Override
   public int hashCode() {
-    return Objects.hash(alias, normalize, compatibilityLevel, compatibilityGroup,
+    return Objects.hash(alias, normalize, validateFields, compatibilityLevel, compatibilityGroup,
         defaultMetadata, overrideMetadata, defaultRuleSet, overrideRuleSet);
   }
 
@@ -198,6 +214,7 @@ public String toString() {
     return "Config{"
         + "alias='" + alias + '\''
         + ", normalize=" + normalize
+        + ", validateFields=" + validateFields
         + ", compatibilityLevel='" + compatibilityLevel + '\''
         + ", compatibilityGroup='" + compatibilityGroup + '\''
         + ", defaultMetadata=" + defaultMetadata

diff --git a/.../io/confluent/kafka/schemaregistry/client/rest/entities/requests/ConfigUpdateRequest.java b/.../io/confluent/kafka/schemaregistry/client/rest/entities/requests/ConfigUpdateRequest.java
@@ -35,6 +35,7 @@ public class ConfigUpdateRequest {
 
   private String alias;
   private Boolean normalize;
+  private Boolean validateFields;
   private String compatibilityLevel;
   private String compatibilityGroup;
   private Metadata defaultMetadata;
@@ -48,6 +49,7 @@ public ConfigUpdateRequest() {
   public ConfigUpdateRequest(Config config) {
     this.alias = config.getAlias();
     this.normalize = config.isNormalize();
+    this.validateFields = config.isValidateFields();
     this.compatibilityLevel = config.getCompatibilityLevel();
     this.compatibilityGroup = config.getCompatibilityGroup();
     this.defaultMetadata = config.getDefaultMetadata();
@@ -80,6 +82,16 @@ public void setNormalize(Boolean normalize) {
     this.normalize = normalize;
   }
 
+  @JsonProperty("validateFields")
+  public Boolean isValidateFields() {
+    return validateFields;
+  }
+
+  @JsonProperty("validateFields")
+  public void setValidateFields(Boolean validateFields) {
+    this.validateFields = validateFields;
+  }
+
   @Schema(description = "Compatibility Level",
       allowableValues = {"BACKWARD", "BACKWARD_TRANSITIVE", "FORWARD", "FORWARD_TRANSITIVE", "FULL",
         "FULL_TRANSITIVE", "NONE"},
@@ -159,6 +171,7 @@ public boolean equals(Object o) {
     ConfigUpdateRequest that = (ConfigUpdateRequest) o;
     return Objects.equals(alias, that.alias)
         && Objects.equals(normalize, that.normalize)
+        && Objects.equals(validateFields, that.validateFields)
         && Objects.equals(compatibilityLevel, that.compatibilityLevel)
         && Objects.equals(compatibilityGroup, that.compatibilityGroup)
         && Objects.equals(defaultMetadata, that.defaultMetadata)
@@ -169,7 +182,7 @@ public boolean equals(Object o) {
 
   @Override
   public int hashCode() {
-    return Objects.hash(alias, normalize, compatibilityLevel, compatibilityGroup,
+    return Objects.hash(alias, normalize, validateFields, compatibilityLevel, compatibilityGroup,
         defaultMetadata, overrideMetadata, defaultRuleSet, overrideRuleSet);
   }
 }
diff --git a/core/src/main/java/io/confluent/kafka/schemaregistry/storage/ConfigValue.java b/core/src/main/java/io/confluent/kafka/schemaregistry/storage/ConfigValue.java
@@ -30,6 +30,7 @@ public class ConfigValue extends SubjectValue {
 
   private String alias;
   private Boolean normalize;
+  private Boolean validateFields;
   private CompatibilityLevel compatibilityLevel;
   private String compatibilityGroup;
   private Metadata defaultMetadata;
@@ -40,6 +41,7 @@ public class ConfigValue extends SubjectValue {
   public ConfigValue(@JsonProperty("subject") String subject,
                      @JsonProperty("alias") String alias,
                      @JsonProperty("normalize") Boolean normalize,
+                     @JsonProperty("validateFields") Boolean validateFields,
                      @JsonProperty("compatibilityLevel") CompatibilityLevel compatibilityLevel,
                      @JsonProperty("compatibilityGroup") String compatibilityGroup,
                      @JsonProperty("defaultMetadata") Metadata defaultMetadata,
@@ -49,6 +51,7 @@ public ConfigValue(@JsonProperty("subject") String subject,
     super(subject);
     this.alias = alias;
     this.normalize = normalize;
+    this.validateFields = validateFields;
     this.compatibilityLevel = compatibilityLevel;
     this.compatibilityGroup = compatibilityGroup;
     this.defaultMetadata = defaultMetadata;
@@ -61,6 +64,7 @@ public ConfigValue(String subject, Config configEntity) {
     super(subject);
     this.alias = configEntity.getAlias();
     this.normalize = configEntity.isNormalize();
+    this.validateFields = configEntity.isValidateFields();
     this.compatibilityLevel = CompatibilityLevel.forName(configEntity.getCompatibilityLevel());
     this.compatibilityGroup = configEntity.getCompatibilityGroup();
     io.confluent.kafka.schemaregistry.client.rest.entities.Metadata defaultMetadata =
@@ -81,6 +85,7 @@ public ConfigValue(String subject, Config configEntity, RuleSetHandler ruleSetHa
     super(subject);
     this.alias = configEntity.getAlias();
     this.normalize = configEntity.isNormalize();
+    this.validateFields = configEntity.isValidateFields();
     this.compatibilityLevel = CompatibilityLevel.forName(configEntity.getCompatibilityLevel());
     this.compatibilityGroup = configEntity.getCompatibilityGroup();
     io.confluent.kafka.schemaregistry.client.rest.entities.Metadata defaultMetadata =
@@ -118,6 +123,16 @@ public void setNormalize(Boolean normalize) {
     this.normalize = normalize;
   }
 
+  @JsonProperty("validateFields")
+  public Boolean isValidateFields() {
+    return validateFields;
+  }
+
+  @JsonProperty("validateFields")
+  public void setValidateFields(Boolean validateFields) {
+    this.validateFields = validateFields;
+  }
+
   @JsonProperty("compatibilityLevel")
   public CompatibilityLevel getCompatibilityLevel() {
     return compatibilityLevel;
@@ -192,6 +207,7 @@ public boolean equals(Object o) {
     ConfigValue that = (ConfigValue) o;
     return Objects.equals(alias, that.alias)
         && Objects.equals(normalize, that.normalize)
+        && Objects.equals(validateFields, that.validateFields)
         && compatibilityLevel == that.compatibilityLevel
         && Objects.equals(compatibilityGroup, that.compatibilityGroup)
         && Objects.equals(defaultMetadata, that.defaultMetadata)
@@ -202,15 +218,17 @@ public boolean equals(Object o) {
 
   @Override
   public int hashCode() {
-    return Objects.hash(super.hashCode(), alias, normalize, compatibilityLevel, compatibilityGroup,
-        defaultMetadata, overrideMetadata, defaultRuleSet, overrideRuleSet);
+    return Objects.hash(super.hashCode(), alias, normalize, validateFields, compatibilityLevel,
+            compatibilityGroup, defaultMetadata, overrideMetadata, defaultRuleSet,
+            overrideRuleSet);
   }
 
   @Override
   public String toString() {
     return "ConfigValue{"
         + "alias='" + alias + '\''
         + ", normalize=" + normalize
+        + ", validateFields=" + validateFields
         + ", compatibilityLevel=" + compatibilityLevel
         + ", compatibilityGroup='" + compatibilityGroup + '\''
         + ", defaultMetadata=" + defaultMetadata
@@ -229,6 +247,7 @@ public Config toConfigEntity() {
     return new Config(
         alias,
         normalize,
+        validateFields,
         compatibilityLevel != null ? compatibilityLevel.name : null,
         compatibilityGroup,
         defaultMetadata != null ? defaultMetadata.toMetadataEntity() : null,
@@ -251,6 +270,8 @@ public static ConfigValue update(ConfigValue oldConfig, ConfigValue newConfig) {
               ? newConfig.getAlias() : oldConfig.getAlias(),
           newConfig.isNormalize() != null
               ? newConfig.isNormalize() : oldConfig.isNormalize(),
+          newConfig.isValidateFields() != null
+              ? newConfig.isValidateFields() : oldConfig.isValidateFields(),
           newConfig.getCompatibilityLevel() != null
               ? newConfig.getCompatibilityLevel() : oldConfig.getCompatibilityLevel(),
           newConfig.getCompatibilityGroup() != null

diff --git a/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java b/core/src/main/java/io/confluent/kafka/schemaregistry/storage/KafkaSchemaRegistry.java
@@ -629,7 +629,7 @@ public Schema register(String subject,
       }
 
       int schemaId = schema.getId();
-      ParsedSchema parsedSchema = canonicalizeSchema(schema, schemaId < 0, normalize);
+      ParsedSchema parsedSchema = canonicalizeSchema(schema, config, schemaId < 0, normalize);
 
       if (parsedSchema != null) {
         // see if the schema to be registered already exists
@@ -1104,7 +1104,8 @@ public Schema lookUpSchemaUnderSubject(
       // Pass a copy of the schema so the original is not modified during normalization
       // to ensure that invalid defaults are not dropped since default validation is disabled
       Schema newSchema = schema != null ? schema.copy() : null;
-      ParsedSchema parsedSchema = canonicalizeSchema(newSchema, false, normalize);
+      Config config = getConfigInScope(subject);
+      ParsedSchema parsedSchema = canonicalizeSchema(newSchema, config, false, normalize);
       if (parsedSchema != null) {
         SchemaIdAndSubjects schemaIdAndSubjects = this.lookupCache.schemaIdAndSubjects(newSchema);
         if (schemaIdAndSubjects != null) {
@@ -1348,23 +1349,29 @@ private void forwardDeleteSubjectModeRequestToLeader(
     }
   }
 
-  private ParsedSchema canonicalizeSchema(Schema schema, boolean isNew, boolean normalize)
-          throws InvalidSchemaException {
+  private ParsedSchema canonicalizeSchema(Schema schema,
+                                          Config config,
+                                          boolean isNew,
+                                          boolean normalize) throws InvalidSchemaException {
     if (schema == null
         || schema.getSchema() == null
         || schema.getSchema().trim().isEmpty()) {
       return null;
     }
     ParsedSchema parsedSchema = parseSchema(schema, isNew, normalize);
-    return maybeValidateAndNormalizeSchema(parsedSchema, schema, true, normalize);
+    return maybeValidateAndNormalizeSchema(parsedSchema, schema, config, true, normalize);
   }
 
-  private ParsedSchema maybeValidateAndNormalizeSchema(
-      ParsedSchema parsedSchema, Schema schema, boolean validate, boolean normalize)
+  private ParsedSchema maybeValidateAndNormalizeSchema(ParsedSchema parsedSchema,
+                                                       Schema schema,
+                                                       Config config,
+                                                       boolean validate,
+                                                       boolean normalize)
           throws InvalidSchemaException {
     try {
       if (validate) {
-        parsedSchema.validate();
+        boolean isStrict = config.isValidateFields() != null ? config.isValidateFields() : true;
+        parsedSchema.validate(isStrict);
       }
       if (normalize) {
         parsedSchema = parsedSchema.normalize();
@@ -1973,12 +1980,12 @@ public List<String> isCompatible(String subject,
       prevParsedSchemas.add(new LazyParsedSchemaHolder(this, previousSchema));
     }
 
-    ParsedSchema parsedSchema = canonicalizeSchema(newSchema, true, normalize);
+    Config config = getConfigInScope(subject);
+    ParsedSchema parsedSchema = canonicalizeSchema(newSchema, config, true, normalize);
     if (parsedSchema == null) {
       log.error("Empty schema");
       throw new InvalidSchemaException("Empty schema");
     }
-    Config config = getConfigInScope(subject);
     return isCompatibleWithPrevious(config, parsedSchema, prevParsedSchemas);
   }
 

diff --git a/json-schema-provider/src/main/java/io/confluent/kafka/schemaregistry/json/JsonSchema.java b/json-schema-provider/src/main/java/io/confluent/kafka/schemaregistry/json/JsonSchema.java
@@ -51,6 +51,7 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashSet;
+import java.util.Optional;
 import java.util.Set;
 import org.everit.json.schema.ArraySchema;
 import org.everit.json.schema.BooleanSchema;
@@ -399,9 +400,25 @@ public JsonSchema normalize() {
   }
 
   @Override
-  public void validate() {
+  public void validate(boolean strict) {
     // Access the raw schema since it is computed lazily
-    rawSchema();
+    Schema rawSchema = rawSchema();
+    if (strict) {
+      if (rawSchema instanceof ObjectSchema) {
+        ObjectSchema schema = (ObjectSchema) rawSchema;
+        Optional<String> restrictedField = schema.getPropertySchemas()
+                .keySet()
+                .stream()
+                .filter(field -> field.startsWith("$"))
+                .findAny();
+        if (restrictedField.isPresent()) {
+          throw new ValidationException(schema,
+                  "Field names cannot start with $ symbol",
+                  "properties",
+                  String.format("#/properties/%s", restrictedField.get()));
+        }
+      }
+    }
   }
 
   public void validate(Object value) throws JsonProcessingException, ValidationException {

diff --git a/...-schema-provider/src/test/java/io/confluent/kafka/schemaregistry/json/JsonSchemaTest.java b/...-schema-provider/src/test/java/io/confluent/kafka/schemaregistry/json/JsonSchemaTest.java
@@ -49,6 +49,7 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -936,6 +937,32 @@ public void testPropertyConflictingWithReservedProperty() {
             errorMessages.get(1));
   }
 
+  @Test
+  public void testRestrictedFields() {
+    String schema = "{\n"
+            + "  \"$schema\": \"http://json-schema.org/draft-07/schema#\",\n"
+            + "  \"$id\": \"task.schema.json\",\n"
+            + "  \"title\": \"Task\",\n"
+            + "  \"description\": \"A task\",\n"
+            + "  \"type\": \"object\",\n"
+            + "  \"properties\": {\n"
+            + "    \"$id\": {\n"
+            + "        \"type\": \"string\"\n"
+            + "    },    \n"
+            + "    \"$title\": {\n"
+            + "        \"description\": \"Task title\",\n"
+            + "        \"type\": \"string\"\n"
+            + "    },    \n"
+            + "    \"status\": {\n"
+            + "        \"type\": \"string\"\n"
+            + "    }\n"
+            + "  }\n"
+            + "}";
+    JsonSchema jsonSchema = new JsonSchema(schema);
+    jsonSchema.validate(false);
+    assertThrows(ValidationException.class, () -> jsonSchema.validate(true));
+  }
+
   private static Map<String, String> getJsonSchemaWithReferences() {
     Map<String, String> schemas = new HashMap<>();
     String reference = "{\"type\":\"object\",\"additionalProperties\":false,\"definitions\":"

diff --git a/...gin/src/main/java/io/confluent/kafka/schemaregistry/maven/ValidateSchemaRegistryMojo.java b/...gin/src/main/java/io/confluent/kafka/schemaregistry/maven/ValidateSchemaRegistryMojo.java
@@ -38,8 +38,7 @@ protected boolean processSchema(String subject,
           String.format("Calling validate('%s', '%s')", subject, schema)
       );
     }
-
-    schema.validate();
+    schema.validate(false);
     return true;
   }
 }