HIVE-29671: Harden Schema path authorization for avro serde

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -2182,6 +2182,17 @@ public static enum ConfVars {
         "Whether to use former Java date/time APIs to convert between timezones when writing timestamps in " +
         "Avro files. Once data are written to the file the effect is permanent (also reflected in the metadata)." +
         "Changing the value of this property affects only new data written to the file."),
+    HIVE_AVRO_SCHEMA_URL_ALLOWED_SCHEMES("hive.avro.schema.url.allowed.schemes",
+        "hdfs,s3,s3a,s3n,abfs,abfss,gs,wasb,wasbs,viewfs,o3fs,ofs",
+        "Comma-separated list of URI schemes permitted for avro.schema.url when loading schemas from a remote " +
+            "location. HTTP/HTTPS and file:// are never allowed via this setting. A URI with no scheme is " +
+            "resolved against the default filesystem."),
+    HIVE_AVRO_SCHEMA_URL_REMOTE_HTTP_ENABLED("hive.avro.schema.url.remote.http.enabled", false,
+        "Whether to allow avro.schema.url values that use http or https. When enabled, the host must also appear " +
+            "in hive.avro.schema.url.http.allowed.hosts. Disabled by default to prevent server-side request forgery."),
+    HIVE_AVRO_SCHEMA_URL_HTTP_ALLOWED_HOSTS("hive.avro.schema.url.http.allowed.hosts", "",
+        "Comma-separated list of hosts permitted for avro.schema.url when hive.avro.schema.url.remote.http.enabled " +
+            "is true. HTTP/HTTPS schema fetch is rejected when this list is empty."),
     HIVE_INT_TIMESTAMP_CONVERSION_IN_SECONDS("hive.int.timestamp.conversion.in.seconds", false,
         "Boolean/tinyint/smallint/int/bigint value is interpreted as milliseconds during the timestamp conversion.\n" +
         "Set this flag to true to interpret the value as seconds to be consistent with float/double." ),

ql/src/java/org/apache/hadoop/hive/ql/plan/PlanUtils.java

@@ -65,6 +65,7 @@
 import org.apache.hadoop.hive.ql.metadata.Partition;
 import org.apache.hadoop.hive.ql.parse.SemanticAnalyzer;
 import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.security.authorization.AuthorizationUtils;
 import org.apache.hadoop.hive.ql.parse.type.ExprNodeTypeCheck;
 import org.apache.hadoop.hive.ql.session.SessionState;
 import org.apache.hadoop.hive.ql.util.NullOrdering;
@@ -1002,6 +1003,11 @@ public static ReadEntity addInput(Set<ReadEntity> inputs, ReadEntity newInput, b
       assert false;
     } else {
       inputs.add(newInput);
+      try {
+        AuthorizationUtils.addAvroSchemaUrlInputForReadEntity(inputs, newInput);
+      } catch (SemanticException e) {
+        throw new RuntimeException("Failed to authorize avro.schema.url for " + newInput.getName(), e);
+      }
       return newInput;
     }
     // make compile happy

ql/src/java/org/apache/hadoop/hive/ql/security/authorization/AuthorizationUtils.java

@@ -17,11 +17,18 @@
  */
 package org.apache.hadoop.hive.ql.security.authorization;
 
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
+import java.util.Set;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.api.HiveObjectPrivilege;
 import org.apache.hadoop.hive.metastore.api.HiveObjectRef;
 import org.apache.hadoop.hive.metastore.api.HiveObjectType;
@@ -34,9 +41,13 @@
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.hooks.Entity;
 import org.apache.hadoop.hive.ql.hooks.Entity.Type;
+import org.apache.hadoop.hive.ql.hooks.ReadEntity;
 import org.apache.hadoop.hive.ql.hooks.WriteEntity;
 import org.apache.hadoop.hive.ql.hooks.WriteEntity.WriteType;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.PlanUtils;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HiveAuthorizationTranslator;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HivePrincipal;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HivePrincipal.HivePrincipalType;
@@ -46,6 +57,9 @@
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HivePrivilegeObject.HivePrivObjectActionType;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HivePrivilegeObject.HivePrivilegeObjectType;
 import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils;
+import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils.AvroTableProperties;
+import org.apache.hadoop.hive.serde2.avro.AvroSerDe;
 
 /**
  * Utility code shared by hive internal code and sql standard authorization plugin implementation
@@ -307,4 +321,83 @@ public static HivePrivObjectActionType getActionType(Entity privObject) {
     return actionType;
   }
 
+  /**
+   * When a table or partition is read, add a DFS ReadEntity for its avro.schema.url if the
+   * schema would be fetched from a filesystem location at query time.
+   */
+  public static void addAvroSchemaUrlInputForReadEntity(Collection<ReadEntity> inputs,
+      ReadEntity readEntity) throws SemanticException {
+    if (readEntity == null || !readEntity.isDirect() || readEntity.isUpdateOrDelete()) {
+      return;
+    }
+    switch (readEntity.getTyp()) {
+    case TABLE:
+      if (readEntity.getTable() != null) {
+        addAvroSchemaUrlInputIfNeeded(inputs, readEntity.getTable());
+      }
+      break;
+    case PARTITION:
+    case DUMMYPARTITION:
+      if (readEntity.getPartition() != null) {
+        addAvroSchemaUrlInputIfNeeded(inputs, readEntity.getPartition().getTable());
+      }
+      break;
+    default:
+      break;
+    }
+  }
+
+  public static void addAvroSchemaUrlInputIfNeeded(Collection<ReadEntity> inputs, Table table)
+      throws SemanticException {
+    String schemaUrl = getFilesystemAvroSchemaUrlToAuthorize(table);
+    if (schemaUrl == null) {
+      return;
+    }
+    ReadEntity schemaUrlInput = toAvroSchemaUrlReadEntity(schemaUrl);
+    if (inputs instanceof Set) {
+      PlanUtils.addInput((Set<ReadEntity>) inputs, schemaUrlInput);
+    } else if (!inputs.contains(schemaUrlInput)) {
+      inputs.add(schemaUrlInput);
+    }
+  }
+
+  private static ReadEntity toAvroSchemaUrlReadEntity(String schemaUrl) {
+    Path path = new Path(schemaUrl);
+    return new ReadEntity(path, isLocalFilesystemSchemaUrl(schemaUrl));
+  }
+
+  private static boolean isLocalFilesystemSchemaUrl(String schemaUrl) {
+    try {
+      String scheme = new URI(schemaUrl).getScheme();
+      return scheme != null && "file".equalsIgnoreCase(scheme);
+    } catch (URISyntaxException e) {
+      return false;
+    }
+  }
+
+  /**
+   * Returns the avro.schema.url when it refers to a filesystem location that will be read to
+   * resolve the schema, or null when no schema-url authorization is required.
+   */
+  public static String getFilesystemAvroSchemaUrlToAuthorize(Table table) {
+    if (table == null || table.isTemporary()) {
+      return null;
+    }
+    if (!AvroSerDe.class.getName().equals(table.getSerializationLib())) {
+      return null;
+    }
+    String literal = table.getProperty(AvroTableProperties.SCHEMA_LITERAL.getPropName());
+    if (StringUtils.isNotEmpty(literal) && !AvroSerdeUtils.SCHEMA_NONE.equals(literal)) {
+      return null;
+    }
+    String schemaUrl = table.getProperty(AvroTableProperties.SCHEMA_URL.getPropName());
+    if (StringUtils.isEmpty(schemaUrl) || AvroSerdeUtils.SCHEMA_NONE.equals(schemaUrl)) {
+      return null;
+    }
+    if (!AvroSerdeUtils.isFilesystemSchemaUrl(schemaUrl)) {
+      return null;
+    }
+    return schemaUrl;
+  }
+
 }

ql/src/java/org/apache/hadoop/hive/ql/security/authorization/command/CommandAuthorizerV2.java

@@ -39,6 +39,7 @@
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.metadata.Table;
 import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
 import org.apache.hadoop.hive.ql.plan.HiveOperation;
 import org.apache.hadoop.hive.ql.security.authorization.AuthorizationUtils;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HiveAuthzContext;
@@ -72,6 +73,7 @@ static void doAuthorization(HiveOperation op, BaseSemanticAnalyzer sem, SessionS
     List<ReadEntity> inputList = new ArrayList<ReadEntity>(inputs);
     List<WriteEntity> outputList = new ArrayList<WriteEntity>(outputs);
     addPermanentFunctionEntities(ss, inputList);
+    enrichAvroSchemaUrlInputs(inputList);
 
     List<HivePrivilegeObject> inputsHObjs = getHivePrivObjects(inputList, selectTab2Cols, hiveOpType, sem);
     List<HivePrivilegeObject> outputHObjs = getHivePrivObjects(outputList, updateTab2Cols, hiveOpType, sem);
@@ -84,6 +86,17 @@ static void doAuthorization(HiveOperation op, BaseSemanticAnalyzer sem, SessionS
     ss.getAuthorizerV2().checkPrivileges(hiveOpType, inputsHObjs, outputHObjs, authzContextBuilder.build());
   }
 
+  private static void enrichAvroSchemaUrlInputs(List<ReadEntity> inputList) throws HiveException {
+    List<ReadEntity> snapshot = new ArrayList<>(inputList);
+    for (ReadEntity readEntity : snapshot) {
+      try {
+        AuthorizationUtils.addAvroSchemaUrlInputForReadEntity(inputList, readEntity);
+      } catch (SemanticException e) {
+        throw new HiveException("Failed to authorize avro.schema.url for " + readEntity.getName(), e);
+      }
+    }
+  }
+
   private static void addPermanentFunctionEntities(SessionState ss, List<ReadEntity> inputList) throws HiveException {
     for (Entry<String, FunctionInfo> function : ss.getCurrentFunctionsInUse().entrySet()) {
       if (function.getValue().getFunctionType() != FunctionType.PERSISTENT) {

ql/src/java/org/apache/hadoop/hive/ql/security/authorization/plugin/metastore/events/AlterTableEvent.java

@@ -32,6 +32,9 @@
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HivePrivilegeObject.HivePrivilegeObjectType;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.metastore.HiveMetaStoreAuthzInfo;
 import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
+import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils;
+import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils.AvroTableProperties;
+import org.apache.hadoop.hive.serde2.avro.AvroSerDe;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -78,6 +81,8 @@ private List<HivePrivilegeObject> getInputHObjs() {
 
     ret.add(getHivePrivilegeObject(oldTable));
 
+    addAvroSchemaUrlInputAuth(ret, event);
+
     COMMAND_STR = buildCommandString(COMMAND_STR, oldTable);
 
     LOG.debug("<== AlterTableEvent.getInputHObjs(): ret={}", ret);
@@ -122,6 +127,27 @@ private List<HivePrivilegeObject> getOutputHObjs() {
     return ret;
   }
 
+  private void addAvroSchemaUrlInputAuth(List<HivePrivilegeObject> ret, PreAlterTableEvent event) {
+    Table newTable = event.getNewTable();
+    Table oldTable = event.getOldTable();
+    if (!AvroSerDe.class.getName().equals(newTable.getSd().getSerdeInfo().getSerializationLib())) {
+      return;
+    }
+    String newSchemaUrl = newTable.getParameters().get(AvroTableProperties.SCHEMA_URL.getPropName());
+    if (StringUtils.isEmpty(newSchemaUrl) || AvroSerdeUtils.SCHEMA_NONE.equals(newSchemaUrl)) {
+      return;
+    }
+    String oldSchemaUrl = oldTable == null ? null
+        : oldTable.getParameters().get(AvroTableProperties.SCHEMA_URL.getPropName());
+    if (StringUtils.equals(oldSchemaUrl, newSchemaUrl)) {
+      return;
+    }
+    if (!AvroSerdeUtils.isFilesystemSchemaUrl(newSchemaUrl)) {
+      return;
+    }
+    ret.add(getHivePrivilegeObjectDfsUri(newSchemaUrl));
+  }
+
   private String buildCommandString(String cmdStr, Table tbl) {
     String ret = cmdStr;
     if (tbl != null) {

ql/src/java/org/apache/hadoop/hive/ql/security/authorization/plugin/metastore/events/CreateTableEvent.java

@@ -31,6 +31,9 @@
 import org.apache.hadoop.hive.ql.security.authorization.plugin.HivePrivilegeObject.HivePrivilegeObjectType;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.metastore.HiveMetaStoreAuthorizableEvent;
 import org.apache.hadoop.hive.ql.security.authorization.plugin.metastore.HiveMetaStoreAuthzInfo;
+import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils;
+import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils.AvroTableProperties;
+import org.apache.hadoop.hive.serde2.avro.AvroSerDe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,18 +68,35 @@ private List<HivePrivilegeObject> getInputHObjs() {
     Database                  database = event.getDatabase();
     String                    uri   = getSdLocation(table.getSd());
 
-    if (StringUtils.isEmpty(uri)) {
-      return ret;
+    if (StringUtils.isNotEmpty(uri)) {
+      // Skip DFS_URI only if table location is under default db path
+      if (this.needDFSUriAuth(uri, this.getDefaultTablePath(database, table))) {
+        ret.add(new HivePrivilegeObject(HivePrivilegeObjectType.DFS_URI, uri));
+      }
     }
 
-    // Skip DFS_URI only if table location is under default db path
-    if (this.needDFSUriAuth(uri, this.getDefaultTablePath(database, table))) {
-      ret.add(new HivePrivilegeObject(HivePrivilegeObjectType.DFS_URI, uri));
-    }
+    addAvroSchemaUrlInputAuth(ret, table, database);
 
     return ret;
   }
 
+  private void addAvroSchemaUrlInputAuth(List<HivePrivilegeObject> ret, Table table, Database database) {
+    if (!AvroSerDe.class.getName().equals(table.getSd().getSerdeInfo().getSerializationLib())) {
+      return;
+    }
+    String schemaUrl = table.getParameters().get(AvroTableProperties.SCHEMA_URL.getPropName());
+    if (StringUtils.isEmpty(schemaUrl) || AvroSerdeUtils.SCHEMA_NONE.equals(schemaUrl)) {
+      return;
+    }
+    if (!AvroSerdeUtils.isFilesystemSchemaUrl(schemaUrl)) {
+      return;
+    }
+    if (!needDFSUriAuth(schemaUrl, getDefaultTablePath(database, table))) {
+      return;
+    }
+    ret.add(getHivePrivilegeObjectDfsUri(schemaUrl));
+  }
+
   private List<HivePrivilegeObject> getOutputHObjs() {
     LOG.debug("==> CreateTableEvent.getOutputHObjs()");