SentryProxy.java

Go to the documentation of this file.

// Copyright 2014 Cloudera Inc.
//
// Licensed 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 com.cloudera.impala.util;

import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.apache.sentry.provider.db.service.thrift.TSentryGroup;
import org.apache.sentry.provider.db.service.thrift.TSentryPrivilege;
import org.apache.sentry.provider.db.service.thrift.TSentryRole;

import com.cloudera.impala.authorization.SentryConfig;
import com.cloudera.impala.authorization.User;
import com.cloudera.impala.catalog.AuthorizationException;
import com.cloudera.impala.catalog.CatalogException;
import com.cloudera.impala.catalog.CatalogServiceCatalog;
import com.cloudera.impala.catalog.Role;
import com.cloudera.impala.catalog.RolePrivilege;
import com.cloudera.impala.common.ImpalaException;
import com.cloudera.impala.common.ImpalaRuntimeException;
import com.cloudera.impala.thrift.TPrivilege;
import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;

public class SentryProxy {
  private static final Logger LOG = Logger.getLogger(SentryProxy.class);

  // Used to periodically poll the Sentry service and updates the catalog with any
  // changes.
  private final ScheduledExecutorService policyReader_ =
      Executors.newScheduledThreadPool(1);

  // The Catalog the SentryPolicyUpdater is associated with.
  private final CatalogServiceCatalog catalog_;

  // The interface to access the Sentry Policy Service to read policy metadata.
  private final SentryPolicyService sentryPolicyService_;

  // This is user that the Catalog Service is running as. This user should always be a
  // Sentry Service admin => have full rights to read/update the Sentry Service.
  private final User processUser_ = new User(System.getProperty("user.name"));

  public SentryProxy(SentryConfig sentryConfig, CatalogServiceCatalog catalog) {
    Preconditions.checkNotNull(catalog);
    Preconditions.checkNotNull(sentryConfig);
    catalog_ = catalog;
    sentryPolicyService_ = new SentryPolicyService(sentryConfig);
    // Sentry Service is enabled.
    // TODO: Make this configurable
    policyReader_.scheduleAtFixedRate(new PolicyReader(), 0, 60,
        TimeUnit.SECONDS);
  }

  private class PolicyReader implements Runnable {
    public void run() {
      synchronized (SentryProxy.this) {
        // Assume all roles should be removed. Then query the Policy Service and remove
        // roles from this set that actually exist.
        Set<String> rolesToRemove = catalog_.getAuthPolicy().getAllRoleNames();
        try {
          // Read the full policy, adding new/modified roles to "updatedRoles".
          for (TSentryRole sentryRole:
              sentryPolicyService_.listAllRoles(processUser_)) {
            // This role exists and should not be removed, delete it from the
            // rolesToRemove set.
            rolesToRemove.remove(sentryRole.getRoleName().toLowerCase());

            Set<String> grantGroups = Sets.newHashSet();
            for (TSentryGroup group: sentryRole.getGroups()) {
              grantGroups.add(group.getGroupName());
            }
            Role existingRole =
                catalog_.getAuthPolicy().getRole(sentryRole.getRoleName());
            Role role;
            // These roles are the same, use the current role.
            if (existingRole != null &&
                existingRole.getGrantGroups().equals(grantGroups)) {
              role = existingRole;
            } else {
              role = catalog_.addRole(sentryRole.getRoleName(), grantGroups);
            }

            // Assume all privileges should be removed. Privileges that still exist are
            // deleted from this set and we are left with the set of privileges that need
            // to be removed.
            Set<String> privilegesToRemove = role.getPrivilegeNames();

            // Check all the privileges that are part of this role.
            for (TSentryPrivilege sentryPriv:
                sentryPolicyService_.listRolePrivileges(processUser_, role.getName())) {
              TPrivilege thriftPriv =
                  SentryPolicyService.sentryPrivilegeToTPrivilege(sentryPriv);
              thriftPriv.setRole_id(role.getId());
              privilegesToRemove.remove(thriftPriv.getPrivilege_name().toLowerCase());

              RolePrivilege existingPriv =
                  role.getPrivilege(thriftPriv.getPrivilege_name());
              // We already know about this privilege (privileges cannot be modified).
              if (existingPriv != null &&
                  existingPriv.getCreateTimeMs() == sentryPriv.getCreateTime()) {
                continue;
              }
              catalog_.addRolePrivilege(role.getName(), thriftPriv);
            }

            // Remove the privileges that no longer exist.
            for (String privilegeName: privilegesToRemove) {
              TPrivilege privilege = new TPrivilege();
              privilege.setPrivilege_name(privilegeName);
              catalog_.removeRolePrivilege(role.getName(), privilege);
            }
          }
        } catch (Exception e) {
          LOG.error("Error refreshing Sentry policy: ", e);
          return;
        }

        // Remove all the roles, incrementing the catalog version to indicate
        // a change.
        for (String roleName: rolesToRemove) {
          catalog_.removeRole(roleName);
        }
      }
    }
  }

  public void checkUserSentryAdmin(User requestingUser)
      throws AuthorizationException {
    // Check if the user has access by issuing a read-only RPC.
    // TODO: This is not an elegant way to verify whether the user has privileges to
    // access Sentry. This should be modified in the future when Sentry has
    // a more robust mechanism to perform these checks.
    try {
      sentryPolicyService_.listAllRoles(requestingUser);
    } catch (ImpalaException e) {
      throw new AuthorizationException(String.format("User '%s' does not have " +
          "privileges to access the requested policy metadata or Sentry Service is " +
          "unavailable.", requestingUser.getName()));
    }
  }

  public synchronized Role createRole(User user, String roleName)
      throws ImpalaException {
    Role role = null;
    if (catalog_.getAuthPolicy().getRole(roleName) != null) {
      throw new CatalogException("Role already exists: " + roleName);
    }
    sentryPolicyService_.createRole(user, roleName, false);
    // Initially the role has no grant groups (empty set).
    role = catalog_.addRole(roleName, Sets.<String>newHashSet());
    return role;
  }

  public synchronized Role dropRole(User user, String roleName) throws ImpalaException {
    sentryPolicyService_.dropRole(user, roleName, false);
    return catalog_.removeRole(roleName);
  }

  public synchronized Role grantRoleGroup(User user, String roleName, String groupName)
      throws ImpalaException {
    sentryPolicyService_.grantRoleToGroup(user, roleName, groupName);
    return catalog_.addRoleGrantGroup(roleName, groupName);
  }

  public synchronized Role revokeRoleGroup(User user, String roleName, String groupName)
      throws ImpalaException {
    sentryPolicyService_.revokeRoleFromGroup(user, roleName, groupName);
    return catalog_.removeRoleGrantGroup(roleName, groupName);
  }

  public synchronized RolePrivilege grantRolePrivilege(User user, String roleName,
      TPrivilege privilege) throws ImpalaException {
    sentryPolicyService_.grantRolePrivilege(user, roleName, privilege);
    return catalog_.addRolePrivilege(roleName, privilege);
  }

  public synchronized RolePrivilege revokeRolePrivilege(User user, String roleName,
      TPrivilege privilege) throws ImpalaException {
    if (!privilege.isHas_grant_opt()) {
      sentryPolicyService_.revokeRolePrivilege(user, roleName, privilege);
      return catalog_.removeRolePrivilege(roleName, privilege);
    } else {
      // If the REVOKE GRANT OPTION has been specified the privilege should not be
      // removed, it should just be updated to clear the GRANT OPTION flag.
      RolePrivilege existingPriv = catalog_.getRolePrivilege(roleName, privilege);
      if (existingPriv == null || !existingPriv.toThrift().isHas_grant_opt()) {
        // Nothing to do. The privilege doesn't exist or the grant option flag is not set
        return existingPriv;
      }

      // Sentry does not yet provide an "alter privilege" API so we need to remove the
      // privilege and re-add it.
      sentryPolicyService_.revokeRolePrivilege(user, roleName, privilege);
      TPrivilege updatedPriv = existingPriv.toThrift();
      updatedPriv.setHas_grant_opt(false);
      sentryPolicyService_.grantRolePrivilege(user, roleName, updatedPriv);
      return catalog_.addRolePrivilege(roleName, updatedPriv);
    }
  }

  public void refresh() throws ImpalaRuntimeException {
    try {
      policyReader_.submit(new PolicyReader()).get();
    } catch (Exception e) {
      // We shouldn't make it here. It means an exception leaked from the
      // AuthorizationPolicyReader.
      throw new ImpalaRuntimeException("Error refreshing authorization policy, " +
          "current policy state may be inconsistent. Running 'invalidate metadata' " +
          "may resolve this problem: ", e);
    }
  }
}