Fusillade Authz helper function

dcplib/security/__init__.py

@@ -21,6 +21,7 @@ class Config:
  _openid_provider = "humancellatlas.auth0.com"
  _auth_provider = "https://auth.data.humancellatlas.org"
  _trusted_google_projects = None
+ _oidc_email_claim = 'https://auth.data.humancellatlas.org/email'
 
  @staticmethod
  def setup(trusted_google_projects: List[str], *, openid_provider: str = None, auth_url: str = None):
@@ -52,3 +53,7 @@ def get_trusted_google_projects():
  if Config._trusted_google_projects is None:
  raise SecurityException(security_config_not_set_error.format('trusted_google_projects'))
  return Config._trusted_google_projects
+
+ @staticmethod
+ def get_oidc_email_claim():
+ return Config._oidc_email_claim

dcplib/security/authz.py

@@ -0,0 +1,67 @@
+import functools
+import typing
+
+import requests
+from requests import HTTPError
+
+from dcplib.errors import AuthorizationException
+from dcplib.security import Config, DCPServiceAccountManager
+
+# recycling the same session for all requests.
+session = requests.Session()
+
+
+def assert_authorized(principal: str,
+ actions: typing.List[str],
+ resources: typing.List[str],
+ authorization_header: typing.Dict[str, str]):
+ resp = session.post("{}/v1/policies/evaluate".format(Config.get_auth_url()),
+ headers=authorization_header,
+ json={"action": actions,
+ "resource": resources,
+ "principal": principal})
+ try:
+ resp.raise_for_status()
+ except HTTPError as ex:
+ raise AuthorizationException(resp) from ex
+ if not resp.json()['result']:
+ raise AuthorizationException()
+
+
+def get_email_claim(token_info: dict):
+ email = token_info.get(Config.get_oidc_email_claim()) or token_info.get('email')
+ if email:
+ return email
+ else:
+ raise AuthorizationException("Email claim {} is missing from token.".format(Config.get_oidc_email_claim()))
+
+
+def authorize(
+ service_account: DCPServiceAccountManager,
+ actions: typing.List[str],
+ resources: typing.List[str],
+):
+ """
+ A decorator for assert_authorized. The Decorated function must take the variable `token_info`, which is a decoded
+ JWT with the appropriate OIDC email claim.
+
+ :param service_account: A DCPServiceAccountManager with a service account that has permission to make request to the
+ /policy/evaluation endpoint.
+ :param actions: The actions passed to assert_authorized
+ :param resources: The resources passed to assert_authorized
+ :return:
+ """
+
+ def decorate(func):
+ @functools.wraps(func)
+ def call(*args, **kwargs):
+ assert_authorized(get_email_claim(kwargs['token_info']),
+ actions,
+ resources,
+ service_account.get_authorization_header()
+ )
+ return func(*args, **kwargs)
+
+ return call
+
+ return decorate

tests/security/test_authz.py

@@ -0,0 +1,52 @@
+import json
+import unittest
+from unittest import mock
+
+from requests import Response
+
+from dcplib import security
+from dcplib.errors import AuthorizationException
+from dcplib.security import authz, DCPServiceAccountManager
+
+
+class TestAuthn(unittest.TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ security.Config.setup(['dummy.iam.gserviceaccount.com'],
+ openid_provider='https://openid_provider.test',
+ auth_url='https://auth_url.test',
+ )
+ cls.gcp_creds = DCPServiceAccountManager.from_secrets_manager(
+ DCPServiceAccountManager.format_secret_id(
+ 'gcp_credentials.json',
+ deployment='test',
+ service='dcplib'
+ ),
+ audience=["https://data.humancellatlas.org/"])
+
+ @mock.patch('dcplib.security.authz.session.post')
+ def test_authorized(self, mocked_class):
+ test_token_info = {'https://auth.data.humancellatlas.org/email': 'fake_email@somewhere.somewhere'}
+ resp = Response()
+ resp.encoding = 'utf-8'
+ resp.status_code = 200
+ mocked_class.return_value = resp
+
+ @authz.authorize(self.gcp_creds, ['test:Action'], ['aws:test:resource:place:21351345134'])
+ def _test(token_info: dict):
+ pass
+
+ resp._content = json.dumps({'result': True}).encode('utf-8')
+ with self.subTest("result = True"):
+ _test(token_info=test_token_info)
+
+ resp._content = json.dumps({'result': False}).encode('utf-8')
+ with self.subTest("result = False"):
+ with self.assertRaises(AuthorizationException):
+ _test(token_info=test_token_info)
+
+ resp.status_code = 500
+ with self.subTest("Request failed"):
+ with self.assertRaises(AuthorizationException):
+ _test(token_info=test_token_info)