/**
 * 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.sentry.provider.db.service.thrift;

import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.sentry.SentryUserException;
import org.apache.sentry.core.model.db.AccessConstants;
import org.apache.sentry.hdfs.ExtendedMetastoreClient;
import org.apache.sentry.hdfs.HMSPaths;
import org.apache.sentry.hdfs.MetastoreClient;
import org.apache.sentry.hdfs.PathsUpdate;
import org.apache.sentry.hdfs.PermissionsUpdate;
import org.apache.sentry.hdfs.UpdateableAuthzPaths;
import org.apache.sentry.provider.common.GroupMappingService;
import org.apache.sentry.provider.db.SentryAccessDeniedException;
import org.apache.sentry.provider.db.SentryAlreadyExistsException;
import org.apache.sentry.provider.db.SentryInvalidInputException;
import org.apache.sentry.provider.db.SentryNoSuchObjectException;
import org.apache.sentry.provider.db.log.entity.JsonLogEntityFactory;
import org.apache.sentry.provider.db.log.util.Constants;
import org.apache.sentry.provider.db.service.UpdateForwarder;
import org.apache.sentry.provider.db.service.UpdateForwarder.ExternalImageRetriever;
import org.apache.sentry.provider.db.service.UpdateablePermissions;
import org.apache.sentry.provider.db.service.persistent.CommitContext;
import org.apache.sentry.provider.db.service.persistent.SentryStore;
import org.apache.sentry.provider.db.service.thrift.PolicyStoreConstants.PolicyStoreServerConfig;
import org.apache.sentry.service.thrift.ServiceConstants.ServerConfig;
import org.apache.sentry.service.thrift.Status;
import org.apache.sentry.service.thrift.TSentryResponseStatus;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

@SuppressWarnings("unused")
public class SentryPolicyStoreProcessor implements SentryPolicyService.Iface {
  private static final Logger LOGGER = LoggerFactory.getLogger(SentryPolicyStoreProcessor.class);
  private static final Logger AUDIT_LOGGER = LoggerFactory.getLogger(Constants.AUDIT_LOGGER_NAME);

  public static final String SENTRY_POLICY_SERVICE_NAME = "SentryPolicyService";

  private final String name;
  private final Configuration conf;
  private final SentryStore sentryStore;
  private final NotificationHandlerInvoker notificationHandlerInvoker;
  private final ImmutableSet<String> adminGroups;
  private boolean isReady;

  private final UpdateForwarder<PathsUpdate> pathsUpdater;
  private final UpdateForwarder<PermissionsUpdate> permsUpdater;
  
  // Initialized to some value > 1 so that the first update notification
  // will trigger a full Image fetch
  private final AtomicLong permSeqNum = new AtomicLong(5);

  public SentryPolicyStoreProcessor(String name, Configuration conf) throws Exception {
    super();
    this.name = name;
    this.conf = conf;
    this.notificationHandlerInvoker = new NotificationHandlerInvoker(conf,
        createHandlers(conf));
    isReady = false;
    sentryStore = new SentryStore(conf);
    isReady = true;
    adminGroups = ImmutableSet.copyOf(toTrimedLower(Sets.newHashSet(conf.getStrings(
        ServerConfig.ADMIN_GROUPS, new String[]{}))));
    HiveConf hiveConf = new HiveConf(conf, Configuration.class);
    if (conf.getBoolean(ServerConfig.SENTRY_HDFS_INTEGRATION_ENABLE, true)) {
      final MetastoreClient hmsClient = new ExtendedMetastoreClient(hiveConf);
      final String[] pathPrefixes = conf
          .getStrings(ServerConfig.SENTRY_HDFS_INTEGRATION_PATH_PREFIXES, new String[]{"/"});
      pathsUpdater = new UpdateForwarder<PathsUpdate>(new UpdateableAuthzPaths(
          pathPrefixes), createHMSImageRetriever(pathPrefixes, hmsClient), 100);
      permsUpdater = new UpdateForwarder<PermissionsUpdate>(
          new UpdateablePermissions(sentryStore), sentryStore, 100);
    } else {
      pathsUpdater = null;
      permsUpdater = null;
    }
  }

  private ExternalImageRetriever<PathsUpdate> createHMSImageRetriever(
      final String[] pathPrefixes, final MetastoreClient hmsClient) {
    return new ExternalImageRetriever<PathsUpdate>() {
      @Override
      public PathsUpdate retrieveFullImage(long currSeqNum) {
        PathsUpdate tempUpdate = new PathsUpdate(currSeqNum, false);
        List<Database> allDatabases = hmsClient.getAllDatabases();
        for (Database db : allDatabases) {
          tempUpdate.newPathChange(db.getName()).addToAddPaths(
              PathsUpdate.cleanPath(db.getLocationUri()));
          List<Table> allTables = hmsClient.getAllTablesOfDatabase(db);
          for (Table tbl : allTables) {
            TPathChanges tblPathChange = tempUpdate.newPathChange(tbl
                .getDbName() + "." + tbl.getTableName());
            List<Partition> tblParts = hmsClient.listAllPartitions(db, tbl);
            tblPathChange.addToAddPaths(PathsUpdate.cleanPath(tbl.getSd()
                    .getLocation() == null ? db.getLocationUri() : tbl
                    .getSd().getLocation()));
            for (Partition part : tblParts) {
              tblPathChange.addToAddPaths(PathsUpdate.cleanPath(part.getSd()
                  .getLocation()));
            }
          }
        }
        UpdateableAuthzPaths tmpAuthzPaths = new UpdateableAuthzPaths(
            pathPrefixes);
        tmpAuthzPaths.updatePartial(Lists.newArrayList(tempUpdate),
            new ReentrantReadWriteLock());
        PathsUpdate retUpdate = new PathsUpdate(currSeqNum, true);
        retUpdate.getThriftObject().setPathsDump(
            tmpAuthzPaths.getPathsDump().createPathsDump());
        return retUpdate;
      }
    };
  }

  public void stop() {
    if (isReady) {
      sentryStore.stop();
    }
  }

  @VisibleForTesting
  static List<NotificationHandler> createHandlers(Configuration conf)
  throws SentryConfigurationException {
    List<NotificationHandler> handlers = Lists.newArrayList();
    Iterable<String> notificationHandlers = Splitter.onPattern("[\\s,]").trimResults()
                                            .omitEmptyStrings().split(conf.get(PolicyStoreServerConfig.NOTIFICATION_HANDLERS, ""));
    for (String notificationHandler : notificationHandlers) {
      Class<?> clazz = null;
      try {
        clazz = Class.forName(notificationHandler);
        if (!NotificationHandler.class.isAssignableFrom(clazz)) {
          throw new SentryConfigurationException("Class " + notificationHandler + " is not a " +
                                                 NotificationHandler.class.getName());
        }
      } catch (ClassNotFoundException e) {
        throw new SentryConfigurationException("Value " + notificationHandler +
                                               " is not a class", e);
      }
      Preconditions.checkNotNull(clazz, "Error class cannot be null");
      try {
        Constructor<?> constructor = clazz.getConstructor(Configuration.class);
        handlers.add((NotificationHandler)constructor.newInstance(conf));
      } catch (Exception e) {
        throw new SentryConfigurationException("Error attempting to create " + notificationHandler, e);
      }
    }
    return handlers;
  }

  @VisibleForTesting
  public Configuration getSentryStoreConf() {
    return conf;
  }

  private static Set<String> toTrimedLower(Set<String> s) {
    Set<String> result = Sets.newHashSet();
    for (String v : s) {
      result.add(v.trim().toLowerCase());
    }
    return result;
  }

  private boolean inAdminGroups(Set<String> requestorGroups) {
    requestorGroups = toTrimedLower(requestorGroups);
    if (Sets.intersection(adminGroups, requestorGroups).isEmpty()) {
      return false;
    } else return true;
  }
  private void authorize(String requestorUser, Set<String> requestorGroups)
  throws SentryAccessDeniedException {
    if (!inAdminGroups(requestorGroups)) {
      String msg = "User: " + requestorUser + " is part of " + requestorGroups +
          " which does not, intersect admin groups " + adminGroups;
      LOGGER.warn(msg);
      throw new SentryAccessDeniedException("Access denied to " + requestorUser);
    }
  }

  @Override
  public TCreateSentryRoleResponse create_sentry_role(
    TCreateSentryRoleRequest request) throws TException {
    TCreateSentryRoleResponse response = new TCreateSentryRoleResponse();
    try {
      authorize(request.getRequestorUserName(),
          getRequestorGroups(request.getRequestorUserName()));
      CommitContext commitContext = sentryStore.createSentryRole(request.getRoleName(),
          request.getRequestorUserName());
      response.setStatus(Status.OK());
      notificationHandlerInvoker.create_sentry_role(commitContext,
          request, response);
    } catch (SentryAlreadyExistsException e) {
      String msg = "Role: " + request + " already exists.";
      LOGGER.error(msg, e);
      response.setStatus(Status.AlreadyExists(msg, e));
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: " + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }

    AUDIT_LOGGER.info(JsonLogEntityFactory.getInstance().createJsonLogEntity(
    		request, response, conf).toJsonFormatLog());
    return response;
  }

  @Override
  public TAlterSentryRoleGrantPrivilegeResponse alter_sentry_role_grant_privilege
  (TAlterSentryRoleGrantPrivilegeRequest request) throws TException {

    TAlterSentryRoleGrantPrivilegeResponse response = new TAlterSentryRoleGrantPrivilegeResponse();
    try {
      CommitContext commitContext = sentryStore.alterSentryRoleGrantPrivilege(request.getRoleName(),
                                    request.getPrivilege());
      response.setStatus(Status.OK());
      notificationHandlerInvoker.alter_sentry_role_grant_privilege(commitContext,
          request, response);
      String authzObj = getAuthzObj(request.getPrivilege());
      if (authzObj != null) {
        PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
        update.addPrivilegeUpdate(authzObj).putToAddPrivileges(
            request.getRoleName(),
            SentryStore.ACTION_MAPPING.get(request.getPrivilege().getAction().toUpperCase())
                .SYMBOL);
        permsUpdater.handleUpdateNotification(update);
        LOGGER.info("Authz Perm preUpdate [" + update.getSeqNum() + "]..");
      }
    } catch (SentryNoSuchObjectException e) {
      String msg = "Role: " + request.getRoleName() + " doesn't exist.";
      LOGGER.error(msg, e);
      response.setStatus(Status.NoSuchObject(msg, e));
    } catch (SentryInvalidInputException e) {
      String msg = "Invalid input privilege object";
      LOGGER.error(msg, e);
      response.setStatus(Status.InvalidInput(msg, e));
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: " + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }

    AUDIT_LOGGER.info(JsonLogEntityFactory.getInstance().createJsonLogEntity(
    		request, response, conf).toJsonFormatLog());
    return response;
  }

  @Override
  public TAlterSentryRoleRevokePrivilegeResponse alter_sentry_role_revoke_privilege
  (TAlterSentryRoleRevokePrivilegeRequest request) throws TException {
    TAlterSentryRoleRevokePrivilegeResponse response = new TAlterSentryRoleRevokePrivilegeResponse();
    try {
      CommitContext commitContext = sentryStore.alterSentryRoleRevokePrivilege(request.getRoleName(),
                                    request.getPrivilege());
      response.setStatus(Status.OK());
      notificationHandlerInvoker.alter_sentry_role_revoke_privilege(commitContext,
          request, response);
      String authzObj = getAuthzObj(request.getPrivilege());
      if (authzObj != null) {
        PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
        update.addPrivilegeUpdate(authzObj).putToDelPrivileges(
            request.getRoleName(),
            SentryStore.ACTION_MAPPING.get(request.getPrivilege().getAction().toUpperCase())
                .SYMBOL);
        permsUpdater.handleUpdateNotification(update);
        LOGGER.info("Authz Perm preUpdate [" + update.getSeqNum() + ", " + authzObj + "]..");
      }
    } catch (SentryNoSuchObjectException e) {
      String msg = "Privilege: [server=" + request.getPrivilege().getServerName() +
    		  ",db=" + request.getPrivilege().getDbName() +
    		  ",table=" + request.getPrivilege().getTableName() +
    		  ",URI=" + request.getPrivilege().getURI() +
    		  ",action=" + request.getPrivilege().getAction() + "] doesn't exist.";
      LOGGER.error(msg, e);
      response.setStatus(Status.NoSuchObject(msg, e));
    } catch (SentryInvalidInputException e) {
      String msg = "Invalid input privilege object";
      LOGGER.error(msg, e);
      response.setStatus(Status.InvalidInput(msg, e));
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: " + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }

    AUDIT_LOGGER.info(JsonLogEntityFactory.getInstance().createJsonLogEntity(
    		request, response, conf).toJsonFormatLog());
    return response;
  }

  @Override
  public TDropSentryRoleResponse drop_sentry_role(
    TDropSentryRoleRequest request)  throws TException {
    TDropSentryRoleResponse response = new TDropSentryRoleResponse();
    TSentryResponseStatus status;
    try {
      authorize(request.getRequestorUserName(),
          getRequestorGroups(request.getRequestorUserName()));
      CommitContext commitContext = sentryStore.dropSentryRole(request.getRoleName());
      response.setStatus(Status.OK());
      notificationHandlerInvoker.drop_sentry_role(commitContext,
          request, response);
      PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
      update.addPrivilegeUpdate(PermissionsUpdate.ALL_AUTHZ_OBJ).putToDelPrivileges(
          request.getRoleName(), PermissionsUpdate.ALL_AUTHZ_OBJ);
      update.addRoleUpdate(request.getRoleName()).addToDelGroups(PermissionsUpdate.ALL_GROUPS);
      permsUpdater.handleUpdateNotification(update);
      LOGGER.info("Authz Perm preUpdate [" + update.getSeqNum() + ", " + request.getRoleName() + "]..");
    } catch (SentryNoSuchObjectException e) {
      String msg = "Role :" + request + " does not exist.";
      LOGGER.error(msg, e);
      response.setStatus(Status.NoSuchObject(msg, e));
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: " + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }

    AUDIT_LOGGER.info(JsonLogEntityFactory.getInstance().createJsonLogEntity(
    		request, response, conf).toJsonFormatLog());
    return response;
  }

  @Override
  public TAlterSentryRoleAddGroupsResponse alter_sentry_role_add_groups(
    TAlterSentryRoleAddGroupsRequest request) throws TException {
    TAlterSentryRoleAddGroupsResponse response = new TAlterSentryRoleAddGroupsResponse();
    try {
      authorize(request.getRequestorUserName(),
          getRequestorGroups(request.getRequestorUserName()));
      CommitContext commitContext = sentryStore.alterSentryRoleAddGroups(request.getRequestorUserName(),
                                    request.getRoleName(), request.getGroups());
      response.setStatus(Status.OK());
      notificationHandlerInvoker.alter_sentry_role_add_groups(commitContext,
          request, response);
      PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
      TRoleChanges rUpdate = update.addRoleUpdate(request.getRoleName());
      for (TSentryGroup group : request.getGroups()) {
        rUpdate.addToAddGroups(group.getGroupName());
      }
      permsUpdater.handleUpdateNotification(update);
      LOGGER.info("Authz Perm preUpdate [" + update.getSeqNum() + ", " + request.getRoleName() + "]..");
    } catch (SentryNoSuchObjectException e) {
      String msg = "Role: " + request + " does not exist.";
      LOGGER.error(msg, e);
      response.setStatus(Status.NoSuchObject(msg, e));
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: " + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }

    AUDIT_LOGGER.info(JsonLogEntityFactory.getInstance().createJsonLogEntity(
    		request, response, conf).toJsonFormatLog());
    return response;
  }

  @Override
  public TAlterSentryRoleDeleteGroupsResponse alter_sentry_role_delete_groups(
    TAlterSentryRoleDeleteGroupsRequest request) throws TException {
    TAlterSentryRoleDeleteGroupsResponse response = new TAlterSentryRoleDeleteGroupsResponse();
    try {
      authorize(request.getRequestorUserName(),
          getRequestorGroups(request.getRequestorUserName()));
      CommitContext commitContext = sentryStore.alterSentryRoleDeleteGroups(request.getRoleName(),
          request.getGroups());
      response.setStatus(Status.OK());
      notificationHandlerInvoker.alter_sentry_role_delete_groups(commitContext,
          request, response);
      PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
      TRoleChanges rUpdate = update.addRoleUpdate(request.getRoleName());
      for (TSentryGroup group : request.getGroups()) {
        rUpdate.addToDelGroups(group.getGroupName());
      }
      permsUpdater.handleUpdateNotification(update);
      LOGGER.info("Authz Perm preUpdate [" + update.getSeqNum() + ", " + request.getRoleName() + "]..");
    } catch (SentryNoSuchObjectException e) {
      String msg = "Role: " + request + " does not exist.";
      LOGGER.error(msg, e);
      response.setStatus(Status.NoSuchObject(msg, e));
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error adding groups to role: " + request;
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }

    AUDIT_LOGGER.info(JsonLogEntityFactory.getInstance().createJsonLogEntity(
    		request, response, conf).toJsonFormatLog());
    return response;
  }

  @Override
  public TListSentryRolesResponse list_sentry_roles_by_group(
    TListSentryRolesRequest request) throws TException {
    TListSentryRolesResponse response = new TListSentryRolesResponse();
    TSentryResponseStatus status;
    Set<TSentryRole> roleSet = new HashSet<TSentryRole>();
    Set<String> groups = new HashSet<String>();
    boolean checkAllGroups = false;
    try {
      // Don't check admin permissions for listing requestor's own roles
      if (AccessConstants.ALL.equalsIgnoreCase(request.getGroupName())) {
        groups = getRequestorGroups(request.getRequestorUserName());
        checkAllGroups = true;
      } else {
        authorize(request.getRequestorUserName(),
          getRequestorGroups(request.getRequestorUserName()));
        groups.add(request.getGroupName());
      }
      roleSet = sentryStore.getTSentryRolesByGroupName(groups, checkAllGroups);
      response.setRoles(roleSet);
      response.setStatus(Status.OK());
    } catch (SentryNoSuchObjectException e) {
      response.setRoles(roleSet);
      String msg = "Role: " + request + " couldn't be retrieved.";
      LOGGER.error(msg, e);
      response.setStatus(Status.NoSuchObject(msg, e));
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: " + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }
    return response;
  }

  @Override
  public TListSentryPrivilegesResponse list_sentry_privileges_by_role(
      TListSentryPrivilegesRequest request) throws TException {
    TListSentryPrivilegesResponse response = new TListSentryPrivilegesResponse();
    TSentryResponseStatus status;
    Set<TSentryPrivilege> privilegeSet = new HashSet<TSentryPrivilege>();
    String subject = request.getRequestorUserName();
    try {
      Set<String> groups = getRequestorGroups(subject);
      Boolean admin = inAdminGroups(groups);
      if(!admin) {
        Set<String> roleNamesForGroups = toTrimedLower(sentryStore.getRoleNamesForGroups(groups));
        if(!roleNamesForGroups.contains(request.getRoleName().trim().toLowerCase())) {
          throw new SentryAccessDeniedException("Access denied to " + subject);
        }
      }
      if (request.isSetAuthorizableHierarchy()) {
        TSentryAuthorizable authorizableHierarchy = request.getAuthorizableHierarchy();
        privilegeSet = sentryStore.getTSentryPrivileges(Sets.newHashSet(request.getRoleName()), authorizableHierarchy);
      } else {
        privilegeSet = sentryStore.getAllTSentryPrivilegesByRoleName(request.getRoleName());
      }
      response.setPrivileges(privilegeSet);
      response.setStatus(Status.OK());
    } catch (SentryNoSuchObjectException e) {
      response.setPrivileges(privilegeSet);
      String msg = "Privilege: " + request + " couldn't be retrieved.";
      LOGGER.error(msg, e);
      response.setStatus(Status.NoSuchObject(msg, e));
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: " + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }
    return response;
  }

  /**
   * This method was created specifically for ProviderBackend.getPrivileges() and is not meant
   * to be used for general privilege retrieval. More details in the .thrift file.
   */
  @Override
  public TListSentryPrivilegesForProviderResponse list_sentry_privileges_for_provider(
      TListSentryPrivilegesForProviderRequest request) throws TException {
    TListSentryPrivilegesForProviderResponse response = new TListSentryPrivilegesForProviderResponse();
    response.setPrivileges(new HashSet<String>());
    try {
      Set<String> privilegesForProvider = sentryStore.listSentryPrivilegesForProvider(
          request.getGroups(), request.getRoleSet(), request.getAuthorizableHierarchy());
      response.setPrivileges(privilegesForProvider);
      if (((privilegesForProvider == null)||(privilegesForProvider.size() == 0))&&(request.getAuthorizableHierarchy() != null)) {
        if (sentryStore.hasAnyServerPrivileges(
            request.getGroups(), request.getRoleSet(), request.getAuthorizableHierarchy().getServer())) {

          // REQUIRED for ensuring 'default' Db is accessible by any user
          // with privileges to atleast 1 object with the specific server as root

          // Need some way to specify that even though user has no privilege
          // For the specific AuthorizableHierarchy.. he has privilege on
          // atleast 1 object in the server hierarchy
          HashSet<String> serverPriv = Sets.newHashSet("server=+");
          response.setPrivileges(serverPriv);
        }
      }
      response.setStatus(Status.OK());
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: " + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }
    return response;
  }

  // retrieve the group mapping for the given user name
  private Set<String> getRequestorGroups(String userName)
      throws SentryUserException {
    return getGroupsFromUserName(this.conf, userName);
  }

  public static Set<String> getGroupsFromUserName(Configuration conf,
      String userName) throws SentryUserException {
    String groupMapping = conf.get(ServerConfig.SENTRY_STORE_GROUP_MAPPING,
        ServerConfig.SENTRY_STORE_GROUP_MAPPING_DEFAULT);
    String authResoruce = conf
        .get(ServerConfig.SENTRY_STORE_GROUP_MAPPING_RESOURCE);

    // load the group mapping provider class
    GroupMappingService groupMappingService;
    try {
      Constructor<?> constrctor = Class.forName(groupMapping)
          .getDeclaredConstructor(Configuration.class, String.class);
      constrctor.setAccessible(true);
      groupMappingService = (GroupMappingService) constrctor
          .newInstance(new Object[] { conf, authResoruce });
    } catch (NoSuchMethodException e) {
      throw new SentryUserException("Unable to instantiate group mapping", e);
    } catch (SecurityException e) {
      throw new SentryUserException("Unable to instantiate group mapping", e);
    } catch (ClassNotFoundException e) {
      throw new SentryUserException("Unable to instantiate group mapping", e);
    } catch (InstantiationException e) {
      throw new SentryUserException("Unable to instantiate group mapping", e);
    } catch (IllegalAccessException e) {
      throw new SentryUserException("Unable to instantiate group mapping", e);
    } catch (IllegalArgumentException e) {
      throw new SentryUserException("Unable to instantiate group mapping", e);
    } catch (InvocationTargetException e) {
      throw new SentryUserException("Unable to instantiate group mapping", e);
    }
    return groupMappingService.getGroups(userName);
  }

  @Override
  public TDropPrivilegesResponse drop_sentry_privilege(
      TDropPrivilegesRequest request) throws TException {
    TDropPrivilegesResponse response = new TDropPrivilegesResponse();
    try {
      authorize(request.getRequestorUserName(), adminGroups);
      sentryStore.dropPrivilege(request.getAuthorizable());
      response.setStatus(Status.OK());
      // TODO : Sentry - HDFS : Have to handle this 
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: "
          + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }
    return response;
  }

  @Override
  public TRenamePrivilegesResponse rename_sentry_privilege(
      TRenamePrivilegesRequest request) throws TException {
    TRenamePrivilegesResponse response = new TRenamePrivilegesResponse();
    try {
      authorize(request.getRequestorUserName(), adminGroups);
      sentryStore.renamePrivilege(request.getOldAuthorizable(),
          request.getNewAuthorizable(), request.getRequestorUserName());
      response.setStatus(Status.OK());
      // TODO : Sentry - HDFS : Have to handle this
    } catch (SentryAccessDeniedException e) {
      LOGGER.error(e.getMessage(), e);
      response.setStatus(Status.AccessDenied(e.getMessage(), e));
    } catch (Exception e) {
      String msg = "Unknown error for request: " + request + ", message: "
          + e.getMessage();
      LOGGER.error(msg, e);
      response.setStatus(Status.RuntimeError(msg, e));
    }
    return response;
  }

  @Override
  public void handle_hms_notification(TPathsUpdate update) throws TException {
    if (pathsUpdater == null) {
      throw new TException("HiveMetastore Path Cache not enabled !!");
    }
    try {
      PathsUpdate hmsUpdate = new PathsUpdate(update);
      pathsUpdater.handleUpdateNotification(hmsUpdate);
      LOGGER.info("Authz Path preUpdate [" + hmsUpdate.getSeqNum() + "]..");
    } catch (Exception e) {
      LOGGER.error("Error handling notification from HMS", e);
      throw new TException(e);
    }
  }

  @Override
  public TAuthzUpdateResponse get_all_authz_updates_from(long permSeqNum, long pathSeqNum) throws TException {
    if (pathsUpdater == null) {
      throw new TException("HiveMetastore Path Cache not enabled !!");
    }
    List<PathsUpdate> pathUpdates = pathsUpdater.getAllUpdatesFrom(pathSeqNum);
    List<PermissionsUpdate> permUpdates = permsUpdater.getAllUpdatesFrom(permSeqNum);
    TAuthzUpdateResponse retVal = new TAuthzUpdateResponse();
    retVal.setAuthzPathUpdate(new LinkedList<TPathsUpdate>());
    retVal.setAuthzPermUpdate(new LinkedList<TPermissionsUpdate>());
    try {
      for (PathsUpdate update : pathUpdates) {
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("### Sending PATH preUpdate seq [" + update.getSeqNum() + "] ###");
          LOGGER.debug("### Sending PATH preUpdate [" + update.getThriftObject() + "] ###");
        }
        retVal.getAuthzPathUpdate().add(update.getThriftObject());
      }
      for (PermissionsUpdate update : permUpdates) {
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("### Sending PERM preUpdate seq [" + update.getSeqNum() + "] ###");
          LOGGER.debug("### Sending PERM preUpdate [" + update.getThriftObject() + "] ###");
        }
        retVal.getAuthzPermUpdate().add(update.getThriftObject());
      }
    } catch (Exception e) {
      LOGGER.error("Error Sending updates to downstream Cache", e);
      throw new TException(e);
    }
    return retVal;
  }

  @Override
  public Map<String, List<String>> get_all_related_paths(String path,
      boolean exactMatch) throws TException {
    if (pathsUpdater == null) {
      throw new TException("HiveMetastore Path Cache not enabled !!");
    }
//    Map<String, LinkedList<String>> relatedPaths = hmsPathCache
//        .getAllRelatedPaths(path, exactMatch);
    return new HashMap<String, List<String>>();
  }

  private String getAuthzObj(TSentryPrivilege privilege) {
    String authzObj = null;
    if (!SentryStore.isNULL(privilege.getDbName())) {
      String dbName = privilege.getDbName();
      String tblName = privilege.getTableName();
      if (tblName == null) {
        authzObj = dbName;
      } else {
        authzObj = dbName + "." + tblName;
      }
    }
    return authzObj;
  }

}