/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hcatalog.cli.SemanticAnalysis;

import java.io.Serializable;
import java.util.List;

import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.ql.exec.Task;
import org.apache.hadoop.hive.ql.metadata.AuthorizationException;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.InvalidTableException;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.parse.AbstractSemanticAnalyzerHook;
import org.apache.hadoop.hive.ql.parse.HiveSemanticAnalyzerHookContext;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.DDLWork;
import org.apache.hadoop.hive.ql.security.authorization.HiveAuthorizationProvider;
import org.apache.hadoop.hive.ql.security.authorization.Privilege;
import org.apache.hadoop.hive.ql.session.SessionState;

/**
 * Base class for HCatSemanticAnalyzer hooks.
 */
public class HCatSemanticAnalyzerBase extends AbstractSemanticAnalyzerHook {

    private HiveAuthorizationProvider authProvider;

    protected String getDbName(Hive hive, String dbName) {
        return dbName == null ? hive.getCurrentDatabase() : dbName;
    }

    public HiveAuthorizationProvider getAuthProvider() {
        if (authProvider == null) {
            authProvider = SessionState.get().getAuthorizer();
        }

        return authProvider;
    }

    @Override
    public void postAnalyze(HiveSemanticAnalyzerHookContext context,
                            List<Task<? extends Serializable>> rootTasks) throws SemanticException {
        super.postAnalyze(context, rootTasks);

        //Authorize the operation.
        authorizeDDL(context, rootTasks);
    }

    /**
     * Checks for the given rootTasks, and calls authorizeDDLWork() for each DDLWork to
     * be authorized. The hooks should override this, or authorizeDDLWork to perform the
     * actual authorization.
     */
    /*
    * Impl note: Hive provides authorization with it's own model, and calls the defined
    * HiveAuthorizationProvider from Driver.doAuthorization(). However, HCat has to
    * do additional calls to the auth provider to implement expected behavior for
    * StorageDelegationAuthorizationProvider. This means, that the defined auth provider
    * is called by both Hive and HCat. The following are missing from Hive's implementation,
    * and when they are fixed in Hive, we can remove the HCat-specific auth checks.
    * 1. CREATE DATABASE/TABLE, ADD PARTITION statements does not call
    * HiveAuthorizationProvider.authorize() with the candidate objects, which means that
    * we cannot do checks against defined LOCATION.
    * 2. HiveOperation does not define sufficient Privileges for most of the operations,
    * especially database operations.
    * 3. For some of the operations, Hive SemanticAnalyzer does not add the changed
    * object as a WriteEntity or ReadEntity.
    *
    * @see https://issues.apache.org/jira/browse/HCATALOG-244
    * @see https://issues.apache.org/jira/browse/HCATALOG-245
    */
    protected void authorizeDDL(HiveSemanticAnalyzerHookContext context,
                                List<Task<? extends Serializable>> rootTasks) throws SemanticException {

        if (!HiveConf.getBoolVar(context.getConf(),
            HiveConf.ConfVars.HIVE_AUTHORIZATION_ENABLED)) {
            return;
        }

        Hive hive;
        try {
            hive = context.getHive();

            for (Task<? extends Serializable> task : rootTasks) {
                if (task.getWork() instanceof DDLWork) {
                    DDLWork work = (DDLWork) task.getWork();
                    if (work != null) {
                        authorizeDDLWork(context, hive, work);
                    }
                }
            }
        } catch (SemanticException ex) {
            throw ex;
        } catch (AuthorizationException ex) {
            throw ex;
        } catch (Exception ex) {
            throw new SemanticException(ex);
        }
    }

    /**
     * Authorized the given DDLWork. Does nothing by default. Override this
     * and delegate to the relevant method in HiveAuthorizationProvider obtained by
     * getAuthProvider().
     */
    protected void authorizeDDLWork(HiveSemanticAnalyzerHookContext context,
                                    Hive hive, DDLWork work) throws HiveException {
    }

    protected void authorize(Privilege[] inputPrivs, Privilege[] outputPrivs)
        throws AuthorizationException, SemanticException {
        try {
            getAuthProvider().authorize(inputPrivs, outputPrivs);
        } catch (HiveException ex) {
            throw new SemanticException(ex);
        }
    }

    protected void authorize(Database db, Privilege priv)
        throws AuthorizationException, SemanticException {
        try {
            getAuthProvider().authorize(db, null, new Privilege[]{priv});
        } catch (HiveException ex) {
            throw new SemanticException(ex);
        }
    }

    protected void authorizeTable(Hive hive, String tableName, Privilege priv)
        throws AuthorizationException, HiveException {
        Table table;
        try {
            table = hive.getTable(tableName);
        } catch (InvalidTableException ite) {
            // Table itself doesn't exist in metastore, nothing to validate.
            return;
        }

        authorize(table, priv);
    }

    protected void authorize(Table table, Privilege priv)
        throws AuthorizationException, SemanticException {
        try {
            getAuthProvider().authorize(table, new Privilege[]{priv}, null);
        } catch (HiveException ex) {
            throw new SemanticException(ex);
        }
    }

    protected void authorize(Partition part, Privilege priv)
        throws AuthorizationException, SemanticException {
        try {
            getAuthProvider().authorize(part, new Privilege[]{priv}, null);
        } catch (HiveException ex) {
            throw new SemanticException(ex);
        }
    }
}