#!/usr/bin/env python # -*- coding: utf-8 -*- # 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. from builtins import object import json import re from nose.tools import assert_true, assert_false, assert_equal, assert_not_equal, assert_raises from beeswax.conf import HIVE_SERVER_HOST from useradmin.models import get_default_user_group, User from desktop.conf import ENABLE_GIST_PREVIEW from desktop.lib.django_test_util import make_logged_in_client from desktop.models import Document2, Directory class TestApi2(object): def setUp(self): self.client = make_logged_in_client(username="api2_user", groupname="default", recreate=True, is_superuser=False) self.user = User.objects.get(username="api2_user") def test_search_entities_interactive_xss(self): query = Document2.objects.create( name='', description='', type='query-hive', owner=self.user ) try: response = self.client.post('/desktop/api/search/entities_interactive/', data={ 'sources': json.dumps(['documents']), 'query_s': json.dumps('alert') }) results = json.loads(response.content)['results'] assert_true(results) result_json = json.dumps(results) assert_false(re.match('<(?!em)', result_json), result_json) assert_false(re.match('(?!em)>', result_json), result_json) assert_false('' in result_json, result_json) assert_true('<' in result_json, result_json) assert_true('>' in result_json, result_json) finally: query.delete() def test_get_hue_config(self): client = make_logged_in_client(username="api2_superuser", groupname="default", recreate=True, is_superuser=True) user = User.objects.get(username="api2_superuser") response = client.get('/desktop/api2/get_hue_config', data={}) # It should have multiple config sections in json config = json.loads(response.content)['config'] assert_true(len(config) > 1) # It should only allow superusers client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test') response = client_not_me.get('/desktop/api2/get_hue_config', data={}) assert_true(b"You must be a superuser" in response.content, response.content) # It should contain a config parameter CANARY = b"abracadabra" clear = HIVE_SERVER_HOST.set_for_testing(CANARY) try: response = client.get('/desktop/api2/get_hue_config', data={}) assert_true(CANARY in response.content, response.content) finally: clear() def test_get_hue_config_private(self): client = make_logged_in_client(username="api2_superuser", groupname="default", recreate=True, is_superuser=True) user = User.objects.get(username="api2_superuser") # Not showing private if not asked for response = client.get('/desktop/api2/get_hue_config', data={}) assert_false(b'bind_password' in response.content) # Masking passwords if private private_response = client.get('/desktop/api2/get_hue_config', data={'private': True}) assert_true(b'bind_password' in private_response.content) config_json = json.loads(private_response.content) desktop_config = [conf for conf in config_json['config'] if conf['key'] == 'desktop'] ldap_desktop_config = [val for conf in desktop_config for val in conf['values'] if val['key'] == 'ldap'] assert_true( # Note: level 1 might not be hidden, e.g. secret_key_script any( val['value'] == '**********' for conf in ldap_desktop_config for val in conf['values'] if val['key'] == 'bind_password' ), ldap_desktop_config ) # There should be more private than non-private assert_true(len(response.content) < len(private_response.content)) def test_get_config(self): response = self.client.get('/desktop/api2/get_config') assert_equal(200, response.status_code) config = json.loads(response.content) assert_true('types' in config['documents']) assert_false('query-TestApi2.test_get_config' in config['documents']['types'], config) doc = Document2.objects.create( name='Query xxx', type='query-TestApi2.test_get_config', owner=self.user ) doc2 = Document2.objects.create( name='Query xxx 2', type='query-TestApi2.test_get_config', owner=self.user ) try: response = self.client.get('/desktop/api2/get_config') assert_equal(200, response.status_code) config = json.loads(response.content) assert_true('query-TestApi2.test_get_config' in config['documents']['types'], config) assert_equal(1, len([t for t in config['documents']['types'] if t == 'query-TestApi2.test_get_config'])) finally: doc.delete() class TestDocumentApiSharingPermissions(object): def setUp(self): self.client = make_logged_in_client(username="perm_user", groupname="default", recreate=True, is_superuser=False) self.client_not_me = make_logged_in_client(username="not_perm_user", groupname="default", recreate=True, is_superuser=False) self.user = User.objects.get(username="perm_user") self.user_not_me = User.objects.get(username="not_perm_user") def _add_doc(self, name): return Document2.objects.create( name=name, type='query-hive', owner=self.user ) def share_doc(self, doc, permissions, client=None): if client is None: client = self.client return client.post("/desktop/api2/doc/share", { 'uuid': json.dumps(doc.uuid), 'data': json.dumps(permissions) }) def share_link_doc(self, doc, perm, client=None): if client is None: client = self.client return client.post("/desktop/api2/doc/share/link", { 'uuid': json.dumps(doc.uuid), 'perm': json.dumps(perm) }) def test_update_permissions(self): doc = self._add_doc('test_update_permissions') response = self.share_doc( doc, { 'read': { 'user_ids': [self.user_not_me.id], 'group_ids': [] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) def test_share_document_permissions(self): # No doc response = self.client.get('/desktop/api2/docs/') assert_false(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_false(json.loads(response.content)['documents']) # Add doc doc = self._add_doc('test_update_permissions') doc_id = '%s' % doc.id response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_false(json.loads(response.content)['documents']) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_false(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) # Share by user response = self.share_doc(doc, { 'read': { 'user_ids': [ self.user_not_me.id ], 'group_ids': [] }, 'write': { 'user_ids': [], 'group_ids': [] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) # Un-share response = self.share_doc(doc, { 'read': { 'user_ids': [ ], 'group_ids': [] }, 'write': { 'user_ids': [], 'group_ids': [] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_false(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_false(json.loads(response.content)['documents']) # Share by group default_group = get_default_user_group() response = self.share_doc(doc, { 'read': { 'user_ids': [ ], 'group_ids': [] }, 'write': { 'user_ids': [], 'group_ids': [default_group.id] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_true(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) # Un-share response = self.share_doc(doc, { 'read': { 'user_ids': [ ], 'group_ids': [] }, 'write': { 'user_ids': [], 'group_ids': [] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_false(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_false(json.loads(response.content)['documents']) # Modify by other user response = self.share_doc(doc, { 'read': { 'user_ids': [ ], 'group_ids': [] }, 'write': { 'user_ids': [self.user_not_me.id], 'group_ids': [] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_true(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) # Un-share response = self.share_doc(doc, { 'read': { 'user_ids': [ ], 'group_ids': [] }, 'write': { 'user_ids': [], 'group_ids': [] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_false(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_false(json.loads(response.content)['documents']) # Modify by group response = self.share_doc(doc, { 'read': { 'user_ids': [ ], 'group_ids': [] }, 'write': { 'user_ids': [], 'group_ids': [default_group.id] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_true(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) # Un-share response = self.share_doc(doc, { 'read': { 'user_ids': [ ], 'group_ids': [] }, 'write': { 'user_ids': [], 'group_ids': [] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_false(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/') assert_false(json.loads(response.content)['documents']) def test_update_permissions_cannot_escalate_privileges(self): doc = self._add_doc('test_update_permissions_cannot_escape_privileges') # Share read permissions response = self.share_doc(doc, { 'read': { 'user_ids': [ self.user_not_me.id ], 'group_ids': [] }, 'write': { 'user_ids': [], 'group_ids': [] } } ) assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) # Try, and fail to escalate privileges. response = self.share_doc(doc, { 'read': { 'user_ids': [ self.user_not_me.id ], 'group_ids': [] }, 'write': { 'user_ids': [ self.user_not_me.id, ], 'group_ids': [] } }, self.client_not_me ) content = json.loads(response.content) assert_equal(content['status'], -1) assert_true("Document does not exist or you don\'t have the permission to access it." in content['message'], content['message']) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) def test_link_sharing_permissions(self): # Add doc doc = self._add_doc('test_link_sharing_permissions') doc_id = '%s' % doc.id response = self.client.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_false(json.loads(response.content)['documents']) response = self.client.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) response = self.client_not_me.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(-1, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_false(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) # Share by read link response = self.share_link_doc(doc, perm='read') assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_false(json.loads(response.content)['documents']) # Link sharing does not list docs in Home, only provides direct access response = self.client.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) response = self.client_not_me.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) # Un-share response = self.share_link_doc(doc, perm='off') assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_false(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_false(json.loads(response.content)['documents']) response = self.client.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) response = self.client_not_me.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(-1, json.loads(response.content)['status'], response.content) # Share by write link response = self.share_link_doc(doc, perm='write') assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_true(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_false(json.loads(response.content)['documents']) response = self.client.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) response = self.client_not_me.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) # Demote to read link response = self.share_link_doc(doc, perm='read') assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_true(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) # Back to false response = self.client.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_false(json.loads(response.content)['documents']) # Link sharing does not list docs in Home, only provides direct access response = self.client.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) response = self.client_not_me.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) # Un-share response = self.share_link_doc(doc, perm='off') assert_equal(0, json.loads(response.content)['status'], response.content) assert_true(doc.can_read(self.user)) assert_true(doc.can_write(self.user)) assert_false(doc.can_read(self.user_not_me)) assert_false(doc.can_write(self.user_not_me)) response = self.client.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_true(json.loads(response.content)['documents']) response = self.client_not_me.get('/desktop/api2/docs/?text=test_link_sharing_permissions') assert_false(json.loads(response.content)['documents']) response = self.client.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(0, json.loads(response.content)['status'], response.content) response = self.client_not_me.get('/desktop/api2/doc/?uuid=%s' % doc_id) assert_equal(-1, json.loads(response.content)['status'], response.content) class TestDocumentGist(object): def setUp(self): self.client = make_logged_in_client(username="gist_user", groupname="default", recreate=True, is_superuser=False) self.client_not_me = make_logged_in_client(username="other_gist_user", groupname="default", recreate=True, is_superuser=False) self.user = User.objects.get(username="gist_user") self.user_not_me = User.objects.get(username="other_gist_user") def _create_gist(self, statement, doc_type, name='', description='', client=None): if client is None: client = self.client return client.post("/desktop/api2/gist/create", { 'statement': statement, 'doc_type': doc_type, 'name': name, 'description': description, }, ) def _get_gist(self, uuid, client=None, is_crawler_bot=False): if client is None: client = self.client if is_crawler_bot: headers = {'HTTP_USER_AGENT': 'Slackbot-LinkExpanding 1.0 (+https://api.slack.com/robots)'} else: headers = {} return client.get("/desktop/api2/gist/open", { 'uuid': uuid, }, **headers ) def test_create(self): assert_false(Document2.objects.filter(type='gist', name='test_gist_create')) response = self._create_gist( statement='SELECT 1', doc_type='hive-query', name='test_gist_create', ) gist = json.loads(response.content) assert_true(Document2.objects.filter(type='gist', name='test_gist_create')) assert_true(Document2.objects.filter(type='gist', uuid=gist['uuid'])) assert_equal( 'SELECT 1', json.loads(Document2.objects.get(type='gist', uuid=gist['uuid']).data)['statement_raw'] ) response2 = self._create_gist( statement='SELECT 2', doc_type='hive-query', name='test_gist_create2', ) gist2 = json.loads(response2.content) assert_true(Document2.objects.filter(type='gist', name='test_gist_create2')) assert_true(Document2.objects.filter(type='gist', uuid=gist2['uuid'])) assert_equal( 'SELECT 2', json.loads(Document2.objects.get(type='gist', uuid=gist2['uuid']).data)['statement_raw'] ) # Create multiple gist directories and then make a new gist home_dir = Directory.objects.get_home_directory(self.user) gist_dir2 = Directory.objects.create(name='Gist', owner=self.user, parent_directory=home_dir) assert_equal(2, Directory.objects.filter(name='Gist', type='directory').count()) response3 = self._create_gist( statement='SELECT 3', doc_type='hive-query', name='test_gist_create3', ) gist3 = json.loads(response3.content) # Gist directories merged into one assert_equal(1, Directory.objects.filter(name='Gist', type='directory').count()) assert_true(Document2.objects.filter(type='gist', name='test_gist_create3')) assert_true(Document2.objects.filter(type='gist', uuid=gist3['uuid'])) assert_equal( 'SELECT 3', json.loads(Document2.objects.get(type='gist', uuid=gist3['uuid']).data)['statement_raw'] ) def test_get(self): response = self._create_gist( statement='SELECT 1', doc_type='hive-query', name='test_gist_get', ) gist = json.loads(response.content) response = self._get_gist(uuid=gist['uuid']) assert_equal(302, response.status_code) assert_equal('/hue/editor?gist=%(uuid)s&type=hive-query' % gist, response.url) response = self._get_gist(uuid=gist['uuid'], client=self.client_not_me) assert_equal(302, response.status_code) assert_equal('/hue/editor?gist=%(uuid)s&type=hive-query' % gist, response.url) def test_gist_directory_creation(self): home_dir = Directory.objects.get_home_directory(self.user) assert_false(home_dir.children.filter(name='Gist').exists()) Document2.objects.get_gist_directory(self.user) assert_true(home_dir.children.filter(name='Gist').exists()) def test_get_unfurl(self): # Unfurling on f = ENABLE_GIST_PREVIEW.set_for_testing(True) try: response = self._create_gist( statement='SELECT 1', doc_type='hive-query', name='test_gist_get', ) gist = json.loads(response.content) response = self._get_gist( uuid=gist['uuid'], is_crawler_bot=True ) assert_equal(200, response.status_code) assert_true(b'' in response.content, response.content) assert_true(b'' in response.content, response.content) finally: f() # Unfurling off f = ENABLE_GIST_PREVIEW.set_for_testing(False) try: response = self._get_gist( uuid=gist['uuid'], is_crawler_bot=True ) assert_equal(302, response.status_code) assert_equal('/hue/editor?gist=%(uuid)s&type=hive-query' % gist, response.url) finally: f()