#!/usr/bin/env python # Licensed to Cloudera, Inc. under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. Cloudera, Inc. 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. import logging import json import random import threading import time from django.utils.translation import ugettext as _ from desktop.lib.exceptions_renderable import PopupException from libzookeeper.models import ZookeeperClient from libsentry.client2 import SentryClient from libsentry.conf import HOSTNAME, PORT from libsentry.sentry_site import get_sentry_server_ha_enabled, get_sentry_server_ha_zookeeper_quorum, get_sentry_server_ha_zookeeper_namespace LOG = logging.getLogger(__name__) _api_cache = None _api_cache_lock = threading.Lock() def ha_error_handler(func): def decorator(*args, **kwargs): retries = 15 while retries > 0: try: return func(*args, **kwargs) except SentryException, e: raise e except Exception, e: retries -= 1 if not get_sentry_server_ha_enabled() or retries == 0: raise e else: # Right now retries on any error and pull a fresh list of servers from ZooKeeper LOG.info('Retrying fetching an available client in ZooKeeper.') global _api_cache _api_cache = None time.sleep(1) args[0].client = _get_client(args[0].client.username, args[0].client.component) LOG.info('Picked %s' % args[0].client) return decorator def get_api(user, component): if component == 'solr': component = component.upper() client = _get_client(user.username, component) return SentryApi(client) class SentryApi(object): def __init__(self, client): self.client = client @ha_error_handler def create_sentry_role(self, roleName): response = self.client.create_sentry_role(roleName) if response.status.value == 0: return response else: raise SentryException(response) @ha_error_handler def drop_sentry_role(self, roleName): response = self.client.drop_sentry_role(roleName) if response.status.value == 0: return response else: raise SentryException(response) @ha_error_handler def alter_sentry_role_grant_privilege(self, roleName, tSentryPrivilege=None): response = self.client.alter_sentry_role_grant_privilege(roleName, tSentryPrivilege) if response.status.value == 0: return response else: raise SentryException(response) @ha_error_handler def alter_sentry_role_revoke_privilege(self, roleName, tSentryPrivilege=None): response = self.client.alter_sentry_role_revoke_privilege(roleName, tSentryPrivilege) if response.status.value == 0: return response else: raise SentryException(response) @ha_error_handler def alter_sentry_role_add_groups(self, roleName, groups): response = self.client.alter_sentry_role_add_groups(roleName, groups) if response.status.value == 0: return response else: raise SentryException(response) @ha_error_handler def alter_sentry_role_delete_groups(self, roleName, groups): response = self.client.alter_sentry_role_delete_groups(roleName, groups) if response.status.value == 0: return response else: raise SentryException(response) @ha_error_handler def list_sentry_roles_by_group(self, groupName=None): response = self.client.list_sentry_roles_by_group(groupName) if response.status.value == 0: roles = [] for role in response.roles: roles.append({ 'name': role.roleName, 'groups': list(role.groups) }) return roles else: raise SentryException(response) @ha_error_handler def list_sentry_privileges_by_role(self, serviceName, roleName, authorizables=None): response = self.client.list_sentry_privileges_by_role(serviceName, roleName, authorizables) if response.status.value == 0: return [self._massage_privilege(privilege) for privilege in response.privileges] else: raise SentryException(response) @ha_error_handler def list_sentry_privileges_for_provider(self, groups, roleSet=None, authorizableHierarchy=None): response = self.client.list_sentry_privileges_for_provider(groups, roleSet, authorizableHierarchy) if response.status.value == 0: return response else: raise SentryException(response) @ha_error_handler def list_sentry_privileges_by_authorizable(self, serviceName, authorizableSet, groups=None, roleSet=None): response = self.client.list_sentry_privileges_by_authorizable(serviceName, authorizableSet, groups, roleSet) if response.status.value != 0: raise SentryException(response) _privileges = [] for authorizable, roles in response.privilegesMapByAuth.iteritems(): _roles = {} for role, privileges in roles.privilegeMap.iteritems(): _roles[role] = [self._massage_privilege(privilege) for privilege in privileges] _privileges.append((self._massage_string_authorizable(authorizable), _roles)) return _privileges @ha_error_handler def drop_sentry_privileges(self, authorizableHierarchy): response = self.client.drop_sentry_privilege(authorizableHierarchy) if response.status.value == 0: return response else: raise SentryException(response) @ha_error_handler def rename_sentry_privileges(self, oldAuthorizable, newAuthorizable): response = self.client.rename_sentry_privilege(oldAuthorizable, newAuthorizable) if response.status.value == 0: return response else: raise SentryException(response) def _massage_privilege(self, privilege): return { 'component': privilege.component, 'serviceName': privilege.serviceName, 'authorizables': self._massage_authorizable(privilege.authorizables), 'action': 'ALL' if privilege.action == '*' else privilege.action.upper(), 'timestamp': privilege.createTime, 'grantorPrincipal': privilege.grantorPrincipal, 'grantOption': privilege.grantOption == 1, } def _massage_authorizable(self, authorizables): return [{'type': auth.type, 'name': auth.name} for auth in authorizables] def _massage_string_authorizable(self, authorizables): return [{'type': auth.split('=')[0], 'name': auth.split('=')[1]} for auth in authorizables.split('->')] class SentryException(Exception): def __init__(self, e): super(SentryException, self).__init__(e) self.message = e.status.message def __str__(self): return self.message def _get_client(username, component): if get_sentry_server_ha_enabled(): servers = _get_server_properties() if servers: server = random.choice(servers) else: raise PopupException(_('No Sentry servers are available.')) else: server = { 'hostname': HOSTNAME.get(), 'port': PORT.get() } return SentryClient(server['hostname'], server['port'], username, component) def _get_server_properties(): global _api_cache if not _api_cache: # If we need to refresh the list or if previously no servers were up _api_cache_lock.acquire() try: if not _api_cache: servers = [] with ZookeeperClient(hosts=get_sentry_server_ha_zookeeper_quorum()) as client: sentry_servers = client.get_children_data(namespace=get_sentry_server_ha_zookeeper_namespace()) for data in sentry_servers: server = json.loads(data.decode("utf-8")) servers.append({'hostname': server['address'], 'port': server['sslPort'] if server['sslPort'] else server['port']}) _api_cache = servers except Exception, e: raise PopupException(_('Error in retrieving Sentry server properties from Zookeeper.'), detail=e) finally: _api_cache_lock.release() return _api_cache