浏览代码

[raz] Skeletons of Raz Client and S3 FileSystem via RAZ

Focused on end to end
Goal is `list buckets` and `list keys` methods first

Tests mocking RAZ signed URL generation via Boto2 generate_url
Unmarshalling XML back from requests call to S3 via Boto2 (might require
a connection)
Unclear if all the FS calls we need are mappable to a signed url with
boto2
boto3 URL generation feels simpler and documents all the possible call
Live Test trying to talk to a RAZ for a list bucket S3 token
Live Test listing buckets via a self signed S3 URL and unmarshalling it

Good next steps:
- Polish RAZ Client to get a token for real
- Investigate how to implement the `list keys with prefix in a bucket`
- On top of bucket/key listing, download/upload/delete/new file as MVP
- Decide on the simplest way to unmarshall into the boto2 Python objects
currently used by the S3 Browser lib
Romain Rigaux 4 年之前
父节点
当前提交
dce5696d1a

+ 14 - 0
desktop/conf.dist/hue.ini

@@ -890,6 +890,20 @@
         # The JSON credentials to authenticate to Google Cloud e.g. '{ "type": "service_account", "project_id": .... }'
         # json_credentials=None
 
+
+  ## Configuration for RAZ service integration
+  # ------------------------------------------------------------------------
+  [[raz]]
+  ## Turns on the integration as ready to use
+  # is_enabled=false
+
+  ## Endpoint to contact
+  # api_url=https://localhost:8080
+
+  ## How to authenticate against: KERBEROS or JWT (not supported yet)
+  # api_authentication=KERBEROS
+
+
 ###########################################################################
 # Settings to configure the snippets available in the Notebook
 ###########################################################################

+ 14 - 0
desktop/conf/pseudo-distributed.ini.tmpl

@@ -874,6 +874,20 @@
         # The JSON credentials to authenticate to Google Cloud e.g. '{ "type": "service_account", "project_id": .... }'
         # json_credentials=None
 
+
+  ## Configuration for RAZ service integration
+  # ------------------------------------------------------------------------
+  [[raz]]
+  ## Turns on the integration as ready to use
+  # is_enabled=false
+
+  ## Endpoint to contact
+  # api_url=https://localhost:8080
+
+  ## How to authenticate against: KERBEROS or JWT (not supported yet)
+  # api_authentication=KERBEROS
+
+
 ###########################################################################
 # Settings to configure the snippets available in the Notebook
 ###########################################################################

+ 1 - 0
desktop/core/requirements.txt

@@ -3,6 +3,7 @@ asn1crypto==0.24.0
 avro-python3==1.8.2
 Babel==2.5.1
 boto==2.46.1
+# boto3==1.17.41  # Last Py2 is 1.17
 celery[redis]==4.4.5  # For Python 3.8
 cffi==1.13.2
 channels==3.0.3

+ 27 - 0
desktop/core/src/desktop/conf.py

@@ -2075,6 +2075,33 @@ CONNECTORS = UnspecifiedConfigSection(
 )
 
 
+
+RAZ = ConfigSection(
+  key='raz',
+  help=_("""Configuration for RAZ service integration"""),
+  members=dict(
+    IS_ENABLED=Config(
+      key='is_enabled',
+      help=_('Turns on the integration as ready to use'),
+      type=coerce_bool,
+      default=False,
+    ),
+    API_URL=Config(
+        key='api_url',
+        help=_('Endpoint to contact'),
+        type=str,
+        default='https://localhost:8080',
+    ),
+    API_AUTHENTICATION=Config(
+        key='api_authentication',
+        help=_('How to authenticate against: KERBEROS or JWT (not supported yet)'),
+        type=coerce_str_lowercase,
+        default='kerberos',
+    ),
+  )
+)
+
+
 QUERY_DATABASE = ConfigSection(
   key='query_database',
   help=_("""Configuration options for specifying the Query History Database."""),

+ 36 - 0
desktop/core/src/desktop/lib/raz/README.md

@@ -0,0 +1,36 @@
+
+
+    [desktop]
+    [[raz]]
+    ## Turns on the integration as ready to use
+    is_enabled=true
+
+    ## Endpoint to contact
+    # api_url=https://localhost:8080
+
+    ## How to authenticate against: KERBEROS or JWT (not supported yet)
+    # api_authentication=KERBEROS
+
+
+## RAZ Hue Clients
+
+Located in [desktop/core/src/desktop/lib/raz/clients_test.py](desktop/core/src/desktop/lib/raz/clients_test.py).
+
+    ./build/env/bin/hue test specific desktop.lib.raz.clients_test
+
+
+## File Systems going via RAZ
+
+Located in [desktop/libs/aws/src/aws/s3/s3connection.py](desktop/libs/aws/src/aws/s3/s3connection.py).
+
+    TEST_S3_BUCKET=gethue-test ./build/env/bin/hue test specific aws.s3.s3connection_test
+
+  GET 'get_all_buckets'
+  GET 'gethue-test'
+  GET 'gethue-test/data/query-hive-weblogs.csv'
+
+
+## Boto 3
+
+https://boto3.amazonaws.com/v1/documentation/api/latest/guide/s3-presigned-urls.html#using-presigned-urls-to-perform-other-s3-operations
+https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.list_objects

+ 141 - 0
desktop/core/src/desktop/lib/raz/__init__.py

@@ -0,0 +1,141 @@
+# 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 __future__ import absolute_import
+
+from builtins import map
+from future.utils import raise_
+import calendar
+import errno
+import logging
+import posixpath
+import re
+import sys
+import time
+
+from functools import wraps
+
+from boto.exception import S3ResponseError
+from hadoop.fs import normpath as fs_normpath
+
+
+ERRNO_MAP = {
+  403: errno.EACCES,
+  404: errno.ENOENT
+}
+DEFAULT_ERRNO = errno.EINVAL
+
+S3_PATH_RE = re.compile('^/*[sS]3[aA]?://([^/]+)(/(.*?([^/]+)?/?))?$')
+S3_ROOT = 's3://'
+S3A_ROOT = 's3a://'
+
+
+def lookup_s3error(error):
+  err_no = ERRNO_MAP.get(error.status, DEFAULT_ERRNO)
+  return IOError(err_no, error.reason)
+
+
+def translate_s3_error(fn):
+  @wraps(fn)
+  def wrapped(*args, **kwargs):
+    try:
+      return fn(*args, **kwargs)
+    except S3ResponseError:
+      _, exc, tb = sys.exc_info()
+      logging.error('S3 error: %s' % exc)
+      lookup = lookup_s3error(exc)
+      raise_(lookup.__class__, lookup, tb)
+  return wrapped
+
+
+def parse_uri(uri):
+  """
+  Returns tuple (bucket_name, key_name, key_basename).
+  Raises ValueError if invalid S3 URI is passed.
+  """
+  match = S3_PATH_RE.match(uri)
+  if not match:
+    raise ValueError("Invalid S3 URI: %s" % uri)
+  key = match.group(3) or ''
+  basename = match.group(4) or ''
+  return match.group(1), key, basename
+
+
+def is_root(uri):
+  """
+  Check if URI is S3 root (S3A://)
+  """
+  return uri.lower() == S3A_ROOT
+
+
+def abspath(cd, uri):
+  """
+  Returns absolute URI, examples:
+
+  abspath('s3a://bucket/key', key2') == 's3a://bucket/key/key2'
+  abspath('s3a://bucket/key', 's3a://bucket2/key2') == 'sa://bucket2/key2'
+  """
+  if cd.lower().startswith(S3A_ROOT):
+    uri = join(cd, uri)
+  else:
+    uri = normpath(join(cd, uri))
+  return uri
+
+
+def join(*comp_list):
+  def _prep(uri):
+    try:
+      return '/%s/%s' % parse_uri(uri)[:2]
+    except ValueError:
+      return '/' if is_root(uri) else uri
+  joined = posixpath.join(*list(map(_prep, comp_list)))
+  if joined and joined[0] == '/':
+    joined = 's3a:/%s' % joined
+  return joined
+
+
+def normpath(path):
+  """
+  Return normalized path but ignore leading S3A_ROOT prefix if it exists
+  """
+  if path.lower().startswith(S3A_ROOT):
+    if is_root(path):
+      normalized = path
+    else:
+      normalized = '%s%s' % (S3A_ROOT, fs_normpath(path[len(S3A_ROOT):]))
+  else:
+    normalized = fs_normpath(path)
+  return normalized
+
+
+def s3datetime_to_timestamp(datetime):
+  """
+  Returns timestamp (seconds) by datetime string from S3 API responses.
+  S3 REST API returns two types of datetime strings:
+  * `Thu, 26 Feb 2015 20:42:07 GMT` for Object HEAD requests
+    (see http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectHEAD.html);
+  * `2015-02-26T20:42:07.000Z` for Bucket GET requests
+    (see http://docs.aws.amazon.com/AmazonS3/latest/API/RESTBucketGET.html).
+  """
+  # There is chance (depends on platform) to get
+  # `'z' is a bad directive in format ...` error (see https://bugs.python.org/issue6641),
+  # but S3 always returns time in GMT, so `GMT` and `.000Z` can be pruned.
+  try:
+    stripped = time.strptime(datetime[:-4], '%a, %d %b %Y %H:%M:%S')
+    assert datetime[-4:] == ' GMT', 'Time [%s] is not in GMT.' % datetime
+  except ValueError:
+    stripped = time.strptime(datetime[:-5], '%Y-%m-%dT%H:%M:%S')
+    assert datetime[-1:] == 'Z' and datetime[-5:-4] == '.', 'Time [%s] is not in GMT.' % datetime
+  return int(calendar.timegm(stripped))

+ 57 - 0
desktop/core/src/desktop/lib/raz/clients.py

@@ -0,0 +1,57 @@
+# 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
+
+from requests_kerberos import HTTPKerberosAuth
+
+from desktop.conf import RAZ
+from desktop.lib.raz.ranger.clients.ranger_raz_adls import RangerRazAdls
+from desktop.lib.raz.ranger.clients.ranger_raz_s3 import RangerRazS3
+
+
+LOG = logging.getLogger(__name__)
+
+
+class S3RazClient():
+
+  def __init__(self, **kwargs):
+    if RAZ.API_AUTHENTICATION.get() == 'kerberos':
+      auth = HTTPKerberosAuth()
+    else:
+      auth = None
+
+    self.ranger = RangerRazS3(RAZ.API_URL.get(), auth)
+
+  def get_url(self, bucket, path, perm='read'):
+    # No GET/POST spec?
+    # e.g. get_url('<storage_account?>', '<bucket>', '<relative_path>', 'read')
+    return self.ranger.get_dsas_token(bucket, path, perm)
+
+
+class AdlsRazClient():
+
+  def __init__(self, storage_account):
+    if RAZ.API_AUTHENTICATION.get() == 'kerberos':
+      auth = HTTPKerberosAuth()
+    else:
+      auth = None
+
+    self.ranger = RangerRazAdls(RAZ.API_URL(), auth)
+
+  def get_url(self, storage_account, container, relative_path, perm='read'):
+    # e.g. get_url('<storage_account>', '<container>', '<relative_path>', 'read')
+    return self.ranger.get_dsas_token(storage_account, container, relative_path, perm)

+ 0 - 59
desktop/core/src/desktop/lib/raz/clients/ranger_raz_adls.py

@@ -1,59 +0,0 @@
-#!/usr/bin/env python
-
-#
-# 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.
-
-import logging
-
-from desktop.lib.raz.clients.model.ranger_raz         import RangerRazRequest, RangerRazResult, ResourceAccess
-from desktop.lib.raz.clients.client.ranger_raz_client import RangerRazClient
-from apache_ranger.utils                    import *
-
-LOG = logging.getLogger(__name__)
-
-class RangerRazAdls:
-    def __init__(self, url, auth):
-        self.razClient = RangerRazClient(url, auth)
-
-    def get_raz_client(self):
-        return self.razClient
-
-    def get_dsas_token(self, storage_account, container, relative_path, action):
-        req = RangerRazRequest()
-
-        req.serviceType = 'adls'
-        req.operation   = ResourceAccess({'resource': {'storageaccount': storage_account, 'container': container, 'relativepath': relative_path}, 'action': action})
-
-        res = self.razClient.check_privilege(req)
-
-        return res.operResult.additionalInfo["ADLS_DSAS"]
-
-'''
-#
-# Sample usage
-#
-from apache_ranger.client.ranger_raz_adls import RangerRazAdls
-from requests_kerberos                    import HTTPKerberosAuth
-
-razAdls = RangerRazAdls('https://<raz_server_host>:<raz_port>', HTTPKerberosAuth())
-
-# disable HTTPS certificate validation; not recommended for production use
-razAdls.razClient.session.verify = False
-
-dsas = razAdls.get_dsas_token('<storage_account>', '<container>>', '<relative_path>', 'read')
-
-print(dsas)
-'''

+ 0 - 199
desktop/core/src/desktop/lib/raz/clients/ranger_raz_client.py

@@ -1,199 +0,0 @@
-#!/usr/bin/env python
-
-#
-# 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.
-
-import json
-import logging
-import os
-
-from requests                       import Session, Response
-from desktop.lib.raz.clients.model.ranger_raz import RangerRazResult
-from apache_ranger.utils            import *
-
-
-LOG = logging.getLogger(__name__)
-
-class RangerRazClient:
-    def __init__(self, url, auth):
-        self.url          = url
-        self.session      = Session()
-        self.session.auth = auth
-
-        logging.getLogger("requests").setLevel(logging.WARNING)
-
-    def get_delegation_token(self, renewer, dtServiceName=None, doAsUser=None):
-        resp = None
-
-        if self.__is_kerberos_authenticated():
-            resp = self.__call_api(RangerRazClient.GET_DELEGATION_TOKEN, self.__get_query_params({ RangerRazClient.PARAM_OP: RangerRazClient.GET_DELEGATION_TOKEN,
-                                                                                                   RangerRazClient.PARAM_RENEWER: renewer,
-                                                                                                   RangerRazClient.PARAM_DT_SERVICENAME: dtServiceName }, doAsUser))
-        else:
-            LOG.error("Kerberos Authentication is required to get RAZ delegation token")
-
-        return resp
-
-    def renew_delegation_token(self, delegation_token, doAsUser=None):
-        resp = None
-
-        if self.__is_kerberos_authenticated():
-            resp = self.__call_api(RangerRazClient.RENEW_DELEGATION_TOKEN, self.__get_query_params({ RangerRazClient.PARAM_OP: RangerRazClient.RENEW_DELEGATION_TOKEN,
-                                                                                                     RangerRazClient.PARAM_TOKEN: delegation_token }, doAsUser))
-        else:
-            LOG.error("Kerberos Authentication is required to renew RAZ delegation token")
-
-        return resp
-
-    def cancel_delegation_token(self, delegation_token, doAsUser=None):
-        resp = None
-
-        if self.__is_kerberos_authenticated():
-            resp = self.__call_api(RangerRazClient.CANCEL_DELEGATION_TOKEN, self.__get_query_params({ RangerRazClient.PARAM_OP: RangerRazClient.CANCEL_DELEGATION_TOKEN,
-                                                                                                      RangerRazClient.PARAM_TOKEN: delegation_token }, doAsUser))
-        else:
-            LOG.error("Kerberos Authentication is required to cancel RAZ delegation token")
-
-        return resp
-
-    def check_privilege(self, raz_request, doAsUser=None):
-        resp = self.__call_api(RangerRazClient.CHECK_PRIVILEGE.format_path({ 'serviceType': raz_request.serviceType }), query_params=self.__get_query_params(None, doAsUser), request_data=raz_request)
-
-        return type_coerce(resp, RangerRazResult)
-
-    def check_privileges(self, raz_requests, doAsUser=None):
-        resp = self.__call_api(RangerRazClient.CHECK_PRIVILEGES.format_path({ 'serviceType': raz_request.serviceType }), query_params=self.__get_query_params(None, doAsUser), request_data=raz_requests)
-
-        return type_coerce_list(resp, RangerRazResult)
-
-    def __is_kerberos_authenticated(self):
-        from requests_kerberos import HTTPKerberosAuth
-
-        return isinstance(self.session.auth, HTTPKerberosAuth)
-
-    def __get_query_params(self, query_params, doAsUser=None):
-        if doAsUser is not None:
-            query_params = query_params or {}
-
-            query_params[RangerRazClient.PARAM_DOAS] = doAsUser
-
-        return query_params
-
-    def __call_api(self, api, query_params=None, request_data=None):
-        ret    = None
-        params = { 'headers': { 'Accept': api.consumes, 'Content-type': api.produces } }
-
-        if query_params:
-            params['params'] = query_params
-
-        if request_data:
-            params['data'] = json.dumps(request_data)
-
-        path = os.path.join(self.url, api.path)
-
-        if LOG.isEnabledFor(logging.DEBUG):
-            LOG.debug("------------------------------------------------------")
-            LOG.debug("Call         : %s %s", api.method, path)
-            LOG.debug("Content-type : %s", api.consumes)
-            LOG.debug("Accept       : %s", api.produces)
-
-        response = None
-
-        if api.method == HttpMethod.GET:
-            response = self.session.get(path, **params)
-        elif api.method == HttpMethod.POST:
-            response = self.session.post(path, **params)
-        elif api.method == HttpMethod.PUT:
-            response = self.session.put(path, **params)
-        elif api.method == HttpMethod.DELETE:
-            response = self.session.delete(path, **params)
-
-        if LOG.isEnabledFor(logging.DEBUG):
-            LOG.debug("HTTP Status: %s", response.status_code if response else "None")
-
-        if response is None:
-            ret = None
-        elif response.status_code == api.expected_status:
-            try:
-                if response.content is not None:
-                    if LOG.isEnabledFor(logging.DEBUG):
-                        LOG.debug("<== __call_api(%s, %s, %s), result=%s", vars(api), params, request_data, response)
-
-                        LOG.debug(response.json())
-
-                    ret = response.json()
-                else:
-                    ret = None
-            except Exception as e:
-                print(e)
-
-                LOG.exception("Exception occurred while parsing response with msg: %s", e)
-
-                raise RangerRazException(api, response)
-        elif response.status_code == HTTPStatus.SERVICE_UNAVAILABLE:
-            LOG.error("Ranger Raz server unavailable. HTTP Status: %s", HTTPStatus.SERVICE_UNAVAILABLE)
-
-            ret = None
-        else:
-            raise RangerRazException(api, response)
-
-        return ret
-
-    # URIs
-    PARAM_OP                 = "op"
-    PARAM_RENEWER            = "renewer"
-    PARAM_TOKEN              = "token"
-    PARAM_DELEGATION         = "delegation"
-    PARAM_DOAS               = "doAs"
-    PARAM_DT_SERVICENAME     = "service"
-    OP_GETDELEGATIONTOKEN    = "GETDELEGATIONTOKEN"
-    OP_RENEWDELEGATIONTOKEN  = "RENEWDELEGATIONTOKEN"
-    OP_CANCELDELEGATIONTOKEN = "CANCELDELEGATIONTOKEN"
-    URI_DELEGATION_TOKEN     = ""
-    URI_CHECK_PRIVILEGE      = "api/authz/{serviceType}/access"
-    URI_CHECK_PRIVILEGES     = "api/authz/{serviceType}/accesses"
-
-    # APIs
-    GET_DELEGATION_TOKEN    = API(URI_DELEGATION_TOKEN, HttpMethod.GET, HTTPStatus.OK)
-    RENEW_DELEGATION_TOKEN  = API(URI_DELEGATION_TOKEN, HttpMethod.PUT, HTTPStatus.OK)
-    CANCEL_DELEGATION_TOKEN = API(URI_DELEGATION_TOKEN, HttpMethod.PUT, HTTPStatus.OK)
-    CHECK_PRIVILEGE         = API(URI_CHECK_PRIVILEGE, HttpMethod.POST, HTTPStatus.OK)
-    CHECK_PRIVILEGES        = API(URI_CHECK_PRIVILEGES, HttpMethod.POST, HTTPStatus.OK)
-
-class RangerRazException(Exception):
-    """Exception raised for errors in API calls.
-
-    Attributes:
-        api      -- api endpoint which caused the error
-        response -- response from the server
-    """
-
-    def __init__(self, api, response):
-        self.method          = api.method.name
-        self.path            = api.path
-        self.expected_status = api.expected_status
-        self.statusCode      = -1
-        self.msgDesc         = None
-        self.messageList     = None
-
-        print(response)
-
-        if api is not None and response is not None:
-            self.statusCode  = response.status_code
-            self.message     = response.content
-
-        Exception.__init__(self, "{} {} failed: expected_status={}, status={}, message={}".format(self.method, self.path, self.expected_status, self.statusCode, self.message))

+ 57 - 0
desktop/core/src/desktop/lib/raz/clients_test.py

@@ -0,0 +1,57 @@
+# 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 unittest
+
+from nose.plugins.skip import SkipTest
+from nose.tools import assert_equal, assert_false, assert_true, assert_raises
+
+from desktop.conf import RAZ
+from desktop.lib.raz.clients import S3RazClient
+
+
+class S3RazClientLiveTest(unittest.TestCase):
+
+  @classmethod
+  def setUpClass(cls):
+    if not RAZ.IS_ENABLED.get():
+      raise SkipTest
+
+  def test_check_access_s3_list_buckets(self):
+
+    url = S3RazClient().get_url()
+
+    assert_true('Expires=' in url)
+
+
+  def test_check_acccess_s3_list_file(self):
+    # e.g. 'https://gethue-test.s3.amazonaws.com/data/query-hive-weblogs.csv?AWSAccessKeyId=AKIA23E77ZX2HVY76YGL&Signature=3lhK%2BwtQ9Q2u5VDIqb4MEpoY3X4%3D&Expires=1617207304'
+
+    url = S3RazClient().get_url(bucket='gethue-test', path='/data/query-hive-weblogs.csv')
+
+    assert_true('data/query-hive-weblogs.csv' in url)
+    assert_true('AWSAccessKeyId=' in url)
+    assert_true('Signature=' in url)
+    assert_true('Expires=' in url)
+
+    url = S3RazClient().get_url(bucket='gethue-test', path='/data/query-hive-weblogs.csv', perm='read', action='write')
+
+    assert_true('data/query-hive-weblogs.csv' in url)
+    assert_true('AWSAccessKeyId=' in url)
+    assert_true('Signature=' in url)
+    assert_true('Expires=' in url)
+
+  def test_check_acccess_s3_list_file_no_access(self): pass

+ 0 - 154
desktop/core/src/desktop/lib/raz/model/ranger_raz.py

@@ -1,154 +0,0 @@
-#!/usr/bin/env python
-
-#
-# 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.
-
-from desktop.lib.raz.clients.model.ranger_base import *
-from apache_ranger.utils             import *
-
-class RangerRazRequestBase(RangerBase):
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerBase.__init__(self, attrs)
-
-        self.requestId       = attrs.get('requestId')
-        self.serviceType     = attrs.get('serviceType')
-        self.serviceName     = attrs.get('serviceName')
-        self.user            = attrs.get('user')
-        self.userGroups      = attrs.get('userGroups')
-        self.accessTime      = attrs.get('accessTime')
-        self.clientIpAddress = attrs.get('clientIpAddress')
-        self.clientType      = attrs.get('clientType')
-        self.clusterName     = attrs.get('clusterName')
-        self.clusterType     = attrs.get('clusterType')
-        self.sessionId       = attrs.get('sessionId')
-        self.context         = attrs.get('context')
-
-    def type_coerce_attrs(self):
-        super(RangerRazRequestBase, self).type_coerce_attrs()
-
-class ResourceAccess(RangerBase):
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerBase.__init__(self, attrs)
-
-        self.resource      = attrs.get('resource')
-        self.resourceOwner = attrs.get('resourceOwner')
-        self.action        = attrs.get('action')
-        self.accessTypes   = attrs.get('accessTypes')
-
-    def type_coerce_attrs(self):
-        super(ResourceAccess, self).type_coerce_attrs()
-
-class RangerRazRequest(RangerRazRequestBase):
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerRazRequestBase.__init__(self, attrs)
-
-        self.operation = attrs.get('operation')
-
-    def type_coerce_attrs(self):
-        super(RangerRazRequest, self).type_coerce_attrs()
-
-        self.operation = type_coerce(self.operation, ResourceAccess)
-
-class RangerRazMultiOperationRequest(RangerRazRequestBase):
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerRazRequestBase.__init__(self, attrs)
-
-        self.operations = attrs.get('operations')
-
-    def type_coerce_attrs(self):
-        super(RangerRazMultiOperationRequest, self).type_coerce_attrs()
-
-        self.operation = type_coerce_list(self.operation, ResourceAccess)
-
-class RangerRazResultBase(RangerBase):
-    ALLOWED        = 0
-    DENIED         = 1
-    NOT_DETERMINED = 2
-
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerBase.__init__(self, attrs)
-
-        self.requestId = attrs.get('requestId')
-
-    def type_coerce_attrs(self):
-        super(RangerRazResultBase, self).type_coerce_attrs()
-
-class AuditInfo(RangerBase):
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerBase.__init__(self, attrs)
-
-        self.auditId       = attrs.get('auditId')
-        self.accessType    = attrs.get('accessType')
-        self.result        = attrs.get('result')
-        self.policyId      = attrs.get('policyId')
-        self.policyVersion = attrs.get('policyVersion')
-
-    def type_coerce_attrs(self):
-        super(AuditInfo, self).type_coerce_attrs()
-
-class ResourceAccessResult(RangerBase):
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerBase.__init__(self, attrs)
-
-        self.result         = attrs.get('result')
-        self.isAudited      = attrs.get('isAudited')
-        self.auditLogs      = attrs.get('auditLogs')
-        self.additionalInfo = attrs.get('additionalInfo')
-
-    def type_coerce_attrs(self):
-        super(ResourceAccessResult, self).type_coerce_attrs()
-
-        self.auditLogs = type_coerce_list(self.operation, AuditInfo)
-
-class RangerRazResult(RangerRazResultBase):
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerRazResultBase.__init__(self, attrs)
-
-        self.operResult = attrs.get('operResult')
-
-    def type_coerce_attrs(self):
-        super(RangerRazResult, self).type_coerce_attrs()
-
-        self.operResult = type_coerce(self.operResult, ResourceAccessResult)
-
-class RangerRazMultiOperationResult(RangerRazResultBase):
-    def __init__(self, attrs=None):
-        attrs = attrs or {}
-
-        RangerRazRequestBase.__init__(self, attrs)
-
-        self.operResults = attrs.get('operResults')
-
-    def type_coerce_attrs(self):
-        super(RangerRazMultiOperationResult, self).type_coerce_attrs()
-
-        self.operResults = type_coerce_list(self.operResults, ResourceAccessResult)

+ 141 - 0
desktop/core/src/desktop/lib/raz/ranger/__init__.py

@@ -0,0 +1,141 @@
+# 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 __future__ import absolute_import
+
+from builtins import map
+from future.utils import raise_
+import calendar
+import errno
+import logging
+import posixpath
+import re
+import sys
+import time
+
+from functools import wraps
+
+from boto.exception import S3ResponseError
+from hadoop.fs import normpath as fs_normpath
+
+
+ERRNO_MAP = {
+  403: errno.EACCES,
+  404: errno.ENOENT
+}
+DEFAULT_ERRNO = errno.EINVAL
+
+S3_PATH_RE = re.compile('^/*[sS]3[aA]?://([^/]+)(/(.*?([^/]+)?/?))?$')
+S3_ROOT = 's3://'
+S3A_ROOT = 's3a://'
+
+
+def lookup_s3error(error):
+  err_no = ERRNO_MAP.get(error.status, DEFAULT_ERRNO)
+  return IOError(err_no, error.reason)
+
+
+def translate_s3_error(fn):
+  @wraps(fn)
+  def wrapped(*args, **kwargs):
+    try:
+      return fn(*args, **kwargs)
+    except S3ResponseError:
+      _, exc, tb = sys.exc_info()
+      logging.error('S3 error: %s' % exc)
+      lookup = lookup_s3error(exc)
+      raise_(lookup.__class__, lookup, tb)
+  return wrapped
+
+
+def parse_uri(uri):
+  """
+  Returns tuple (bucket_name, key_name, key_basename).
+  Raises ValueError if invalid S3 URI is passed.
+  """
+  match = S3_PATH_RE.match(uri)
+  if not match:
+    raise ValueError("Invalid S3 URI: %s" % uri)
+  key = match.group(3) or ''
+  basename = match.group(4) or ''
+  return match.group(1), key, basename
+
+
+def is_root(uri):
+  """
+  Check if URI is S3 root (S3A://)
+  """
+  return uri.lower() == S3A_ROOT
+
+
+def abspath(cd, uri):
+  """
+  Returns absolute URI, examples:
+
+  abspath('s3a://bucket/key', key2') == 's3a://bucket/key/key2'
+  abspath('s3a://bucket/key', 's3a://bucket2/key2') == 'sa://bucket2/key2'
+  """
+  if cd.lower().startswith(S3A_ROOT):
+    uri = join(cd, uri)
+  else:
+    uri = normpath(join(cd, uri))
+  return uri
+
+
+def join(*comp_list):
+  def _prep(uri):
+    try:
+      return '/%s/%s' % parse_uri(uri)[:2]
+    except ValueError:
+      return '/' if is_root(uri) else uri
+  joined = posixpath.join(*list(map(_prep, comp_list)))
+  if joined and joined[0] == '/':
+    joined = 's3a:/%s' % joined
+  return joined
+
+
+def normpath(path):
+  """
+  Return normalized path but ignore leading S3A_ROOT prefix if it exists
+  """
+  if path.lower().startswith(S3A_ROOT):
+    if is_root(path):
+      normalized = path
+    else:
+      normalized = '%s%s' % (S3A_ROOT, fs_normpath(path[len(S3A_ROOT):]))
+  else:
+    normalized = fs_normpath(path)
+  return normalized
+
+
+def s3datetime_to_timestamp(datetime):
+  """
+  Returns timestamp (seconds) by datetime string from S3 API responses.
+  S3 REST API returns two types of datetime strings:
+  * `Thu, 26 Feb 2015 20:42:07 GMT` for Object HEAD requests
+    (see http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectHEAD.html);
+  * `2015-02-26T20:42:07.000Z` for Bucket GET requests
+    (see http://docs.aws.amazon.com/AmazonS3/latest/API/RESTBucketGET.html).
+  """
+  # There is chance (depends on platform) to get
+  # `'z' is a bad directive in format ...` error (see https://bugs.python.org/issue6641),
+  # but S3 always returns time in GMT, so `GMT` and `.000Z` can be pruned.
+  try:
+    stripped = time.strptime(datetime[:-4], '%a, %d %b %Y %H:%M:%S')
+    assert datetime[-4:] == ' GMT', 'Time [%s] is not in GMT.' % datetime
+  except ValueError:
+    stripped = time.strptime(datetime[:-5], '%Y-%m-%dT%H:%M:%S')
+    assert datetime[-1:] == 'Z' and datetime[-5:-4] == '.', 'Time [%s] is not in GMT.' % datetime
+  return int(calendar.timegm(stripped))

+ 16 - 0
desktop/core/src/desktop/lib/raz/ranger/clients/__init__.py

@@ -0,0 +1,16 @@
+#!/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.

+ 55 - 0
desktop/core/src/desktop/lib/raz/ranger/clients/ranger_raz_adls.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+
+#
+# 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.
+
+import logging
+
+from desktop.lib.raz.ranger.model.ranger_raz import RangerRazRequest, ResourceAccess
+from desktop.lib.raz.ranger.clients.ranger_raz_client import RangerRazClient
+
+
+LOG = logging.getLogger(__name__)
+
+
+class RangerRazAdls:
+  def __init__(self, url, auth):
+    self.razClient = RangerRazClient(url, auth)
+    # move `storage_account` into constructor?
+
+  def get_raz_client(self):
+    return self.razClient
+
+  def get_dsas_token(self, storage_account, container, relative_path, action="read"):
+    req = RangerRazRequest()
+
+    req.serviceType = "adls"
+    req.operation = ResourceAccess(
+      {
+        "resource": {
+          "storageaccount": storage_account,
+          "container": container,
+          "relativepath": relative_path,
+        },
+        "action": action,
+      }
+    )
+
+    res = self.razClient.check_privilege(req)
+
+    # TODO: Check if no access inside RangerRazResult and raise exception?
+
+    return res.operResult.additionalInfo["ADLS_DSAS"]

+ 261 - 0
desktop/core/src/desktop/lib/raz/ranger/clients/ranger_raz_client.py

@@ -0,0 +1,261 @@
+#!/usr/bin/env python
+
+#
+# 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.
+
+import json
+import logging
+import os
+
+from requests import Session, Response
+from desktop.lib.raz.ranger.model.ranger_raz import RangerRazResult
+from apache_ranger.utils import *
+
+
+LOG = logging.getLogger(__name__)
+
+
+class RangerRazClient:
+  def __init__(self, url, auth):
+    self.url = url
+    self.session = Session()
+    self.session.auth = auth
+
+    logging.getLogger("requests").setLevel(logging.WARNING)
+
+  def get_delegation_token(self, renewer, dtServiceName=None, doAsUser=None):
+    resp = None
+
+    if self.__is_kerberos_authenticated():
+      resp = self.__call_api(
+        RangerRazClient.GET_DELEGATION_TOKEN,
+        self.__get_query_params(
+          {
+            RangerRazClient.PARAM_OP: RangerRazClient.GET_DELEGATION_TOKEN,
+            RangerRazClient.PARAM_RENEWER: renewer,
+            RangerRazClient.PARAM_DT_SERVICENAME: dtServiceName,
+          },
+          doAsUser,
+        ),
+      )
+    else:
+      LOG.error("Kerberos Authentication is required to get RAZ delegation token")
+
+    return resp
+
+  def renew_delegation_token(self, delegation_token, doAsUser=None):
+    resp = None
+
+    if self.__is_kerberos_authenticated():
+      resp = self.__call_api(
+        RangerRazClient.RENEW_DELEGATION_TOKEN,
+        self.__get_query_params(
+          {
+            RangerRazClient.PARAM_OP: RangerRazClient.RENEW_DELEGATION_TOKEN,
+            RangerRazClient.PARAM_TOKEN: delegation_token,
+          },
+          doAsUser,
+        ),
+      )
+    else:
+      LOG.error(
+        "Kerberos Authentication is required to renew RAZ delegation token"
+      )
+
+    return resp
+
+  def cancel_delegation_token(self, delegation_token, doAsUser=None):
+    resp = None
+
+    if self.__is_kerberos_authenticated():
+      resp = self.__call_api(
+        RangerRazClient.CANCEL_DELEGATION_TOKEN,
+        self.__get_query_params(
+          {
+            RangerRazClient.PARAM_OP: RangerRazClient.CANCEL_DELEGATION_TOKEN,
+            RangerRazClient.PARAM_TOKEN: delegation_token,
+          },
+          doAsUser,
+        ),
+      )
+    else:
+      LOG.error(
+        "Kerberos Authentication is required to cancel RAZ delegation token"
+      )
+
+    return resp
+
+  def check_privilege(self, raz_request, doAsUser=None):
+    resp = self.__call_api(
+      RangerRazClient.CHECK_PRIVILEGE.format_path(
+        {"serviceType": raz_request.serviceType}
+      ),
+      query_params=self.__get_query_params(None, doAsUser),
+      request_data=raz_request,
+    )
+
+    return type_coerce(resp, RangerRazResult)
+
+  def check_privileges(self, raz_requests, doAsUser=None):
+    resp = self.__call_api(
+      RangerRazClient.CHECK_PRIVILEGES.format_path(
+        {"serviceType": raz_request.serviceType}
+      ),
+      query_params=self.__get_query_params(None, doAsUser),
+      request_data=raz_requests,
+    )
+
+    return type_coerce_list(resp, RangerRazResult)
+
+  def __is_kerberos_authenticated(self):
+    from requests_kerberos import HTTPKerberosAuth
+
+    return isinstance(self.session.auth, HTTPKerberosAuth)
+
+  def __get_query_params(self, query_params, doAsUser=None):
+    if doAsUser is not None:
+      query_params = query_params or {}
+
+      query_params[RangerRazClient.PARAM_DOAS] = doAsUser
+
+    return query_params
+
+  def __call_api(self, api, query_params=None, request_data=None):
+    ret = None
+    params = {"headers": {"Accept": api.consumes, "Content-type": api.produces}}
+
+    if query_params:
+      params["params"] = query_params
+
+    if request_data:
+      params["data"] = json.dumps(request_data)
+
+    path = os.path.join(self.url, api.path)
+
+    if LOG.isEnabledFor(logging.DEBUG):
+      LOG.debug("------------------------------------------------------")
+      LOG.debug("Call         : %s %s", api.method, path)
+      LOG.debug("Content-type : %s", api.consumes)
+      LOG.debug("Accept       : %s", api.produces)
+
+    response = None
+
+    if api.method == HttpMethod.GET:
+      response = self.session.get(path, **params)
+    elif api.method == HttpMethod.POST:
+      response = self.session.post(path, **params)
+    elif api.method == HttpMethod.PUT:
+      response = self.session.put(path, **params)
+    elif api.method == HttpMethod.DELETE:
+      response = self.session.delete(path, **params)
+
+    if LOG.isEnabledFor(logging.DEBUG):
+      LOG.debug("HTTP Status: %s", response.status_code if response else "None")
+
+    if response is None:
+      ret = None
+    elif response.status_code == api.expected_status:
+      try:
+        if response.content is not None:
+          if LOG.isEnabledFor(logging.DEBUG):
+            LOG.debug(
+              "<== __call_api(%s, %s, %s), result=%s",
+              vars(api),
+              params,
+              request_data,
+              response,
+            )
+
+            LOG.debug(response.json())
+
+          ret = response.json()
+        else:
+          ret = None
+      except Exception as e:
+        print(e)
+
+        LOG.exception(
+          "Exception occurred while parsing response with msg: %s", e
+        )
+
+        raise RangerRazException(api, response)
+    elif response.status_code == HTTPStatus.SERVICE_UNAVAILABLE:
+      LOG.error(
+        "Ranger Raz server unavailable. HTTP Status: %s",
+        HTTPStatus.SERVICE_UNAVAILABLE,
+      )
+
+      ret = None
+    else:
+      raise RangerRazException(api, response)
+
+    return ret
+
+  # URIs
+  PARAM_OP = "op"
+  PARAM_RENEWER = "renewer"
+  PARAM_TOKEN = "token"
+  PARAM_DELEGATION = "delegation"
+  PARAM_DOAS = "doAs"
+  PARAM_DT_SERVICENAME = "service"
+  OP_GETDELEGATIONTOKEN = "GETDELEGATIONTOKEN"
+  OP_RENEWDELEGATIONTOKEN = "RENEWDELEGATIONTOKEN"
+  OP_CANCELDELEGATIONTOKEN = "CANCELDELEGATIONTOKEN"
+  URI_DELEGATION_TOKEN = ""
+  URI_CHECK_PRIVILEGE = "api/authz/{serviceType}/access"
+  URI_CHECK_PRIVILEGES = "api/authz/{serviceType}/accesses"
+
+  # APIs
+  GET_DELEGATION_TOKEN = API(URI_DELEGATION_TOKEN, HttpMethod.GET, HTTPStatus.OK)
+  RENEW_DELEGATION_TOKEN = API(URI_DELEGATION_TOKEN, HttpMethod.PUT, HTTPStatus.OK)
+  CANCEL_DELEGATION_TOKEN = API(URI_DELEGATION_TOKEN, HttpMethod.PUT, HTTPStatus.OK)
+  CHECK_PRIVILEGE = API(URI_CHECK_PRIVILEGE, HttpMethod.POST, HTTPStatus.OK)
+  CHECK_PRIVILEGES = API(URI_CHECK_PRIVILEGES, HttpMethod.POST, HTTPStatus.OK)
+
+
+class RangerRazException(Exception):
+  """Exception raised for errors in API calls.
+
+  Attributes:
+    api      -- api endpoint which caused the error
+    response -- response from the server
+  """
+
+  def __init__(self, api, response):
+    self.method = api.method.name
+    self.path = api.path
+    self.expected_status = api.expected_status
+    self.statusCode = -1
+    self.msgDesc = None
+    self.messageList = None
+
+    print(response)
+
+    if api is not None and response is not None:
+      self.statusCode = response.status_code
+      self.message = response.content
+
+    Exception.__init__(
+      self,
+      "{} {} failed: expected_status={}, status={}, message={}".format(
+        self.method,
+        self.path,
+        self.expected_status,
+        self.statusCode,
+        self.message,
+      ),
+    )

+ 53 - 0
desktop/core/src/desktop/lib/raz/ranger/clients/ranger_raz_s3.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+
+#
+# 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.
+
+import logging
+
+from desktop.lib.raz.ranger.model.ranger_raz import RangerRazRequest, ResourceAccess
+from desktop.lib.raz.ranger.clients.ranger_raz_client import RangerRazClient
+
+
+LOG = logging.getLogger(__name__)
+
+
+class RangerRazS3:
+  def __init__(self, url, auth):
+    self.razClient = RangerRazClient(url, auth)
+    # move `storage_account` into constructor?
+
+  def get_dsas_token(self, storage_account, container, relative_path, action="read"):
+    req = RangerRazRequest()
+
+    req.serviceType = "s3"
+    req.operation = ResourceAccess(
+      # TODO: parameters for S3
+      {
+        "resource": {
+          "storageaccount": storage_account,
+          "container": container,
+          "relativepath": relative_path,
+        },
+        "action": action,
+      }
+    )
+
+    res = self.razClient.check_privilege(req)
+
+    # TODO: Check if no access inside RangerRazResult and raise exception?
+
+    return res.operResult.additionalInfo["S3_DSAS"]

+ 16 - 0
desktop/core/src/desktop/lib/raz/ranger/model/__init__.py

@@ -0,0 +1,16 @@
+#!/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.

+ 163 - 0
desktop/core/src/desktop/lib/raz/ranger/model/ranger_raz.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python
+
+#
+# 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.
+
+from apache_ranger.model.ranger_base import *
+from apache_ranger.utils import *
+
+
+class RangerRazRequestBase(RangerBase):
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerBase.__init__(self, attrs)
+
+    self.requestId = attrs.get("requestId")
+    self.serviceType = attrs.get("serviceType")
+    self.serviceName = attrs.get("serviceName")
+    self.user = attrs.get("user")
+    self.userGroups = attrs.get("userGroups")
+    self.accessTime = attrs.get("accessTime")
+    self.clientIpAddress = attrs.get("clientIpAddress")
+    self.clientType = attrs.get("clientType")
+    self.clusterName = attrs.get("clusterName")
+    self.clusterType = attrs.get("clusterType")
+    self.sessionId = attrs.get("sessionId")
+    self.context = attrs.get("context")
+
+  def type_coerce_attrs(self):
+    super(RangerRazRequestBase, self).type_coerce_attrs()
+
+
+class ResourceAccess(RangerBase):
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerBase.__init__(self, attrs)
+
+    self.resource = attrs.get("resource")
+    self.resourceOwner = attrs.get("resourceOwner")
+    self.action = attrs.get("action")
+    self.accessTypes = attrs.get("accessTypes")
+
+  def type_coerce_attrs(self):
+    super(ResourceAccess, self).type_coerce_attrs()
+
+
+class RangerRazRequest(RangerRazRequestBase):
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerRazRequestBase.__init__(self, attrs)
+
+    self.operation = attrs.get("operation")
+
+  def type_coerce_attrs(self):
+    super(RangerRazRequest, self).type_coerce_attrs()
+
+    self.operation = type_coerce(self.operation, ResourceAccess)
+
+
+class RangerRazMultiOperationRequest(RangerRazRequestBase):
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerRazRequestBase.__init__(self, attrs)
+
+    self.operations = attrs.get("operations")
+
+  def type_coerce_attrs(self):
+    super(RangerRazMultiOperationRequest, self).type_coerce_attrs()
+
+    self.operation = type_coerce_list(self.operation, ResourceAccess)
+
+
+class RangerRazResultBase(RangerBase):
+  ALLOWED = 0
+  DENIED = 1
+  NOT_DETERMINED = 2
+
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerBase.__init__(self, attrs)
+
+    self.requestId = attrs.get("requestId")
+
+  def type_coerce_attrs(self):
+    super(RangerRazResultBase, self).type_coerce_attrs()
+
+
+class AuditInfo(RangerBase):
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerBase.__init__(self, attrs)
+
+    self.auditId = attrs.get("auditId")
+    self.accessType = attrs.get("accessType")
+    self.result = attrs.get("result")
+    self.policyId = attrs.get("policyId")
+    self.policyVersion = attrs.get("policyVersion")
+
+  def type_coerce_attrs(self):
+    super(AuditInfo, self).type_coerce_attrs()
+
+
+class ResourceAccessResult(RangerBase):
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerBase.__init__(self, attrs)
+
+    self.result = attrs.get("result")
+    self.isAudited = attrs.get("isAudited")
+    self.auditLogs = attrs.get("auditLogs")
+    self.additionalInfo = attrs.get("additionalInfo")
+
+  def type_coerce_attrs(self):
+    super(ResourceAccessResult, self).type_coerce_attrs()
+
+    self.auditLogs = type_coerce_list(self.operation, AuditInfo)
+
+
+class RangerRazResult(RangerRazResultBase):
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerRazResultBase.__init__(self, attrs)
+
+    self.operResult = attrs.get("operResult")
+
+  def type_coerce_attrs(self):
+    super(RangerRazResult, self).type_coerce_attrs()
+
+    self.operResult = type_coerce(self.operResult, ResourceAccessResult)
+
+
+class RangerRazMultiOperationResult(RangerRazResultBase):
+  def __init__(self, attrs=None):
+    attrs = attrs or {}
+
+    RangerRazRequestBase.__init__(self, attrs)
+
+    self.operResults = attrs.get("operResults")
+
+  def type_coerce_attrs(self):
+    super(RangerRazMultiOperationResult, self).type_coerce_attrs()
+
+    self.operResults = type_coerce_list(self.operResults, ResourceAccessResult)

+ 28 - 17
desktop/libs/aws/src/aws/client.py

@@ -15,44 +15,53 @@
 # limitations under the License.
 from __future__ import absolute_import
 
-from builtins import str
-from builtins import object
+from builtins import str, object
 import logging
 import os
-
-import boto.s3.connection
+import boto
 
 from aws import conf as aws_conf
-from aws.s3.s3fs import S3FileSystemException
-from aws.s3.s3fs import S3FileSystem
+from aws.s3.s3connection import RazUrlConnection, BotoUrlConnection
+from aws.s3.s3fs import S3FileSystem, S3FileSystemException
 
+from desktop.conf import RAZ
 from desktop.lib.idbroker import conf as conf_idbroker
 from desktop.lib.idbroker.client import IDBroker
 
 LOG = logging.getLogger(__name__)
 
+
 HTTP_SOCKET_TIMEOUT_S = 60
 
 
 def get_credential_provider(identifier, user):
   client_conf = aws_conf.AWS_ACCOUNTS[identifier] if identifier in aws_conf.AWS_ACCOUNTS else None
-  return CredentialProviderIDBroker(IDBroker.from_core_site('s3a', user)) if conf_idbroker.is_idbroker_enabled('s3a') else CredentialProviderConf(client_conf)
+  return CredentialProviderIDBroker(IDBroker.from_core_site('s3a', user)) if conf_idbroker.is_idbroker_enabled('s3a') \
+      else CredentialProviderConf(client_conf)
 
 
 def _make_client(identifier, user):
-  client_conf = aws_conf.AWS_ACCOUNTS[identifier] if identifier in aws_conf.AWS_ACCOUNTS else None
+  if RAZ.IS_ENABLED.get() and not aws_conf.IS_SELF_SIGNING_ENABLED.get():
+    s3_client = RazUrlConnection()  # Note: AWS configuration is fully skipped
+    s3_client_expiration = None
+  else:
+    client_conf = aws_conf.AWS_ACCOUNTS[identifier] if identifier in aws_conf.AWS_ACCOUNTS else None
 
-  client = Client.from_config(client_conf, get_credential_provider(identifier, user))
-  return S3FileSystem(client.get_s3_connection(), client.expiration) # It would be nice if the connection is lazy loaded
+    s3_client_builder = Client.from_config(client_conf, get_credential_provider(identifier, user))
+    s3_client = s3_client_builder.get_s3_connection()
+    s3_client_expiration = s3_client_builder.expiration
+
+  return S3FileSystem(s3_client, s3_client_expiration)
 
 
 class CredentialProviderConf(object):
   def __init__(self, conf):
-    self._conf=conf
+    self._conf = conf
 
   def validate(self):
     credentials = self.get_credentials()
-    if None in (credentials.get('AccessKeyId'), credentials.get('SecretAccessKey')) and not credentials.get('AllowEnvironmentCredentials') and not aws_conf.has_iam_metadata():
+    if None in (credentials.get('AccessKeyId'), credentials.get('SecretAccessKey')) and not credentials.get('AllowEnvironmentCredentials') \
+        and not aws_conf.has_iam_metadata():
       raise ValueError('Can\'t create AWS client, credential is not configured')
     return True
 
@@ -75,7 +84,7 @@ class CredentialProviderConf(object):
 
 class CredentialProviderIDBroker(object):
   def __init__(self, idbroker):
-    self.idbroker=idbroker
+    self.idbroker = idbroker
     self.credentials = None
 
   def validate(self):
@@ -139,7 +148,7 @@ class Client(object):
       )
 
   def get_s3_connection(self):
-
+    """S3 connection can actually be seen as a S3Client. A true new client would be a Boto3Client."""
     kwargs = {
       'aws_access_key_id': self._access_key_id,
       'aws_secret_access_key': self._secret_access_key,
@@ -166,8 +175,7 @@ class Client(object):
         kwargs.update({'host': self._host})
         connection = boto.s3.connection.S3Connection(**kwargs)
       elif self._region:
-        connection = boto.s3.connect_to_region(self._region,
-                                             **kwargs)
+        connection = boto.s3.connect_to_region(self._region, **kwargs)
       else:
         kwargs.update({'host': 's3.amazonaws.com'})
         connection = boto.s3.connection.S3Connection(**kwargs)
@@ -176,10 +184,13 @@ class Client(object):
       raise S3FileSystemException('Failed to construct S3 Connection, check configurations for aws.')
 
     if connection is None:
-      # If no connection, attemt to fallback to IAM instance metadata
+      # If no connection, attempt to fallback to IAM instance metadata
       connection = boto.connect_s3()
 
       if connection is None:
         raise S3FileSystemException('Can not construct S3 Connection for region %s' % self._region)
 
+    if aws_conf.IS_SELF_SIGNING_ENABLED.get():
+      connection = BotoUrlConnection(connection)
+
     return connection

+ 8 - 1
desktop/libs/aws/src/aws/conf.py

@@ -20,7 +20,6 @@ import os
 import re
 import sys
 
-
 import requests
 
 from desktop.lib.conf import Config, UnspecifiedConfigSection, ConfigSection, coerce_bool, coerce_password_from_script
@@ -162,6 +161,14 @@ HAS_IAM_DETECTION = Config(
   type=coerce_bool
 )
 
+IS_SELF_SIGNING_ENABLED = Config(
+  key='is_self_signing_enabled',
+  help=_('Skip boto and use self signed URL and requests to make the calls to S3. Used for testing the RAZ integration.'),
+  type=coerce_bool,
+  private=True,
+  default=False,
+)
+
 AWS_ACCOUNTS = UnspecifiedConfigSection(
   'aws_accounts',
   help=_('One entry for each AWS account'),

+ 200 - 0
desktop/libs/aws/src/aws/s3/s3connection.py

@@ -0,0 +1,200 @@
+# 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 boto
+import logging
+import requests
+import xml.sax
+
+from boto.exception import BotoClientError
+from boto.resultset import ResultSet
+from boto.s3.bucket import Bucket, Key
+from boto.s3.bucketlistresultset import BucketListResultSet
+from boto.s3.prefix import Prefix
+
+from desktop.lib.raz.clients import S3RazClient
+
+
+LOG = logging.getLogger(__name__)
+
+
+# Note: Connection means more "Client" but we follow boto2 terminology
+
+
+class UrlConnection():
+
+  def get_all_buckets(self, response):
+    LOG.debug('get_all_buckets')
+    LOG.debug(response)
+    LOG.debug(response.content)
+
+    rs = ResultSet([('Bucket', self.connection.bucket_class)])
+    h = boto.handler.XmlHandler(rs, self.connection)
+    xml.sax.parseString(response.content, h)
+    LOG.debug(rs)
+
+
+class RazUrlConnection():
+
+  def __init__(self):
+    self.raz = S3RazClient()
+
+  def _generate_url(self, bucket_name=None, object_name=None, expiration=3600):
+    self.raz.get_url(bucket_name, object_name)
+
+  def get_all_buckets(self, headers=None):
+    url = self._generate_url()
+
+  def get_bucket(self, bucket_name, validate=True, headers=None):
+    pass
+
+  def get_key(self, key_name, headers=None, version_id=None, response_headers=None, validate=True):
+    pass
+
+  def get_all_keys(self, headers=None, **params):
+    pass
+
+
+class UrlKey(Key):
+
+  def _generate_url(self, action='GET', **kwargs):
+    LOG.debug(kwargs)
+    try:
+      # http://boto.cloudhackers.com/en/latest/ref/s3.html#boto.s3.key.Key.generate_url
+      tmp_url = self.generate_url(self.expiration, action, **kwargs)
+    except BotoClientError as e:
+      LOG.error(e)
+      return None
+
+    return tmp_url
+
+
+class UrlBucket(Bucket):
+
+  def list(self, prefix='', delimiter='', marker='', headers=None, encoding_type=None):
+    LOG.debug('prefix')
+    LOG.debug(prefix)
+    return self.get_all_keys()  # TODO: unsure yet how to generate URL with key prefix filtering
+
+
+  def get_key(self, key_name, headers=None, version_id=None, response_headers=None, validate=True):
+    LOG.debug('key name: %s' % key_name)
+    kwargs = {'bucket': self.name, 'key': key_name}
+
+    tmp_url = self.connection.generate_url(3000, 'GET', **kwargs)
+    # tmp_url = self.generate_url(1000, 'GET', self.name, key_name, headers, version_id, response_headers, validate)
+
+    response = requests.get(tmp_url)
+    LOG.debug(response)
+    LOG.debug(response.content)
+
+    k = self.key_class(self, key_name)
+
+    return k
+
+
+  def get_all_keys(self, headers=None, **params):
+    # delimiter=/&prefix=data/
+    kwargs = {'bucket': self.name, 'key': ''} # data/
+
+    tmp_url = self.connection.generate_url(3000, 'GET', **kwargs)
+
+    response = requests.get(tmp_url)
+
+    LOG.debug('get_all_keys %s' % kwargs)
+    LOG.debug(params)
+    LOG.debug(response)
+    LOG.debug(response.content)
+
+    rs = ResultSet([('Contents', Key), ('CommonPrefixes', Prefix)])  # Or BucketListResultSet?
+    h = boto.handler.XmlHandler(rs, self)
+    xml.sax.parseString(response.content, h)
+    LOG.debug(rs)
+
+    return rs
+
+
+  def _generate_url(self, action='GET', **kwargs):
+    LOG.debug(kwargs)
+    try:
+      # http://boto.cloudhackers.com/en/latest/ref/s3.html#boto.s3.bucket.Bucket.generate_url
+      tmp_url = self.generate_url(self.expiration, action, **kwargs)
+    except BotoClientError as e:
+      LOG.error(e)
+      return None
+
+    return tmp_url
+
+
+class BotoUrlConnection():
+
+  def __init__(self, connection):
+    self.connection = connection
+    self.expiration = 3600
+
+    self.connection.make_request = None  # We make sure we never call directly
+    self.connection.set_bucket_class(UrlBucket)  # We use our bucket class to override any direct call to S3
+
+
+  def get_all_buckets(self, headers=None):
+    kwargs = {'action': 'GET'}
+    try:
+      tmp_url = self._generate_url(**kwargs)
+    except BotoClientError as e:
+      LOG.error(e)
+      return None
+
+    response = requests.get(tmp_url)
+
+    LOG.debug('get_all_buckets')
+    LOG.debug(response)
+    LOG.debug(response.content)
+
+    rs = ResultSet([('Bucket', self.connection.bucket_class)])
+    h = boto.handler.XmlHandler(rs, self.connection)
+    xml.sax.parseString(response.content, h)
+    LOG.debug(rs)
+
+    return rs
+
+
+  def get_bucket(self, bucket_name, validate=True, headers=None):
+    kwargs = {'action': 'GET', 'bucket': bucket_name}
+
+    tmp_url = self._generate_url(**kwargs)
+
+    response = requests.get(tmp_url)
+
+    LOG.debug('get_bucket')
+    LOG.debug(response)
+    LOG.debug(response.content)
+
+    rs = self.connection.bucket_class(self.connection, bucket_name, key_class=UrlKey)  # Using content?
+    LOG.debug(rs)
+
+    return rs
+
+
+  def _generate_url(self, action='GET', **kwargs):
+    LOG.debug(kwargs)
+    try:
+      # http://boto.cloudhackers.com/en/latest/ref/s3.html#boto.s3.connection.S3Connection.generate_url
+      tmp_url = self.connection.generate_url(self.expiration, action, **kwargs)
+    except BotoClientError as e:
+      LOG.error(e)
+      return None
+
+    return tmp_url

+ 59 - 0
desktop/libs/aws/src/aws/s3/s3connection_test.py

@@ -0,0 +1,59 @@
+# 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 requests
+
+from nose.tools import assert_equal, assert_false, assert_true, assert_raises
+
+from aws.client import _make_client
+from aws.s3.s3connection import BotoUrlConnection
+from aws.s3.s3test_utils import S3TestBase
+
+# TEST_S3_BUCKET=gethue-test ./build/env/bin/hue test specific aws.s3.s3connection_test
+
+
+class BotoUrlConnectionIntegrationTest(S3TestBase):
+
+  @classmethod
+  def setUpClass(cls):
+    S3TestBase.setUpClass()
+
+
+  def setUp(self):
+    super(BotoUrlConnectionIntegrationTest, self).setUp()
+
+    self.c = _make_client(identifier='default', user=None)
+    self.connection = self.c._s3_connection.connection
+
+
+  def test_list_buckets(self):
+    buckets = BotoUrlConnection(self.connection).get_all_buckets()
+
+    assert_equal('[<Bucket: demo-gethue>, <Bucket: gethue-test>]', str(buckets))
+
+
+  def test_list_file(self):
+    kwargs = {'action': 'GET', 'bucket': 'gethue-test', 'key': 'data/query-hive-weblogs.csv'}
+    url = BotoUrlConnection(self.connection).generate_url(**kwargs)
+
+    url = 'https://gethue-test.s3.amazonaws.com/data/query-hive-weblogs.csv?AWSAccessKeyId=AKIA23E77ZX2HVY76YGL&Signature=3lhK%2BwtQ9Q2u5VDIqb4MEpoY3X4%3D&Expires=1617207304'
+
+    assert_true('data/query-hive-weblogs.csv' in url)
+    assert_true('AWSAccessKeyId=' in url)
+    assert_true('Signature=' in url)
+    assert_true('Expires=' in url)
+
+    response = requests.get(url)

+ 7 - 1
desktop/libs/aws/src/aws/s3/s3fs.py

@@ -168,6 +168,8 @@ class S3FileSystem(object):
     bucket_name, key_name = s3.parse_uri(path)[:2]
     bucket = self._get_bucket(bucket_name)
     try:
+      print('_get_key')
+      print(key_name)
       return bucket.get_key(key_name, validate=validate)
     except BotoClientError as e:
       raise S3FileSystemException(_('Failed to access path at "%s": %s') % (path, e.reason))
@@ -191,6 +193,8 @@ class S3FileSystem(object):
       return S3Stat.for_s3_root()
 
     try:
+      print('path:')
+      print(path)
       key = self._get_key(path, validate=True)
     except BotoClientError as e:
       raise S3FileSystemException(_('Failed to access path "%s": %s') % (path, e.reason))
@@ -202,6 +206,8 @@ class S3FileSystem(object):
       else:
         raise S3FileSystemException(_('Failed to access path "%s": %s') % (path, e.reason))
     except Exception as e: # SSL errors show up here, because they've been remapped in boto
+      import traceback
+      print(traceback.print_exc())
       raise S3FileSystemException(_('Failed to access path "%s": %s') % (path, str(e)))
     if key is None:
       key = self._get_key(path, validate=False)
@@ -214,7 +220,7 @@ class S3FileSystem(object):
       return S3Stat.from_key(key, is_dir=is_directory_name, fs=fs)
     else:
       key.name = S3FileSystem._append_separator(key.name)
-      ls = key.bucket.get_all_keys(prefix=key.name, max_keys=1)
+      ls = key.bucket.get_all_keys(prefix=key.name, max_keys=1)  # Not sure possible via signed request
       if len(ls) > 0:
         return S3Stat.from_key(key, is_dir=True, fs=fs)
     return None

+ 8 - 3
desktop/libs/aws/src/aws/s3/s3test_utils.py

@@ -16,7 +16,6 @@
 from __future__ import absolute_import
 
 from builtins import range
-import logging
 import os
 import random
 import string
@@ -26,8 +25,10 @@ from nose.plugins.skip import SkipTest
 
 import aws
 
-from contextlib import contextmanager
+from aws import conf as aws_conf
 from aws.s3 import parse_uri, join
+from contextlib import contextmanager
+from desktop.lib.fsmanager import get_client
 
 
 def get_test_bucket():
@@ -52,7 +53,11 @@ class S3TestBase(unittest.TestCase):
       return
 
     cls.path_prefix = 'test-hue/%s' % generate_id(size=16)
-    cls.s3_connection = aws.get_client('default').get_s3_connection()
+
+    if aws_conf.IS_SELF_SIGNING_ENABLED.get():
+      cls.s3_connection = get_client(name='default', fs='s3a', user='hue')._s3_connection
+    else:
+      cls.s3_connection = aws.get_client('default').get_s3_connection()  # Probably broken nowadays
     cls.bucket = cls.s3_connection.get_bucket(cls.bucket_name, validate=True)
 
   @classmethod