PolicyBasedCheckOtpController (#31)

OTP support is implemented via PolicyBasedCheckOtpController and check_otp method in PolicyBasedController.
The flag of connected OTP support is reading from token_info.otp (filled in genesis core)

---------

Co-authored-by: electis <electis@tokens.team>

Author: electis

gcl_iam/controllers.py

@@ -16,6 +16,7 @@
 
 import uuid
 
+from restalchemy.api import constants
 from restalchemy.api import controllers
 from restalchemy.common import contexts
 from restalchemy.dm import filters
@@ -29,6 +30,7 @@ class PolicyBasedControllerMixin(object):
 
     __policy_service_name__ = ""
     __policy_name__ = None
+    _otp_mandatory = set()
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -76,6 +78,14 @@ def _enforce_and_override_project_id_in_kwargs(self, method, kwargs):
                 self._ctx_project_id
             )
 
+    def _check_otp(self, method):
+        if (
+            self._introspection.get("otp_enabled")
+            or method in self._otp_mandatory
+        ):
+            if not self._introspection.get("otp_verified"):
+                raise exceptions.OTPInvalidCodeError()
+
 
 class PolicyBasedController(
     PolicyBasedControllerMixin, controllers.BaseResourceController
@@ -174,3 +184,27 @@ def update(self, uuid, **kwargs):
         dm.update_dm(values=kwargs)
         dm.update()
         return dm
+
+
+class PolicyBasedCheckOtpController(PolicyBasedController):
+    _otp_mandatory = {constants.CREATE, constants.UPDATE, constants.DELETE}
+
+    def create(self, **kwargs):
+        self._check_otp(constants.CREATE)
+        return PolicyBasedController(self).create(**kwargs)
+
+    def get(self, **kwargs):
+        self._check_otp(constants.GET)
+        return PolicyBasedController(self).get(**kwargs)
+
+    def filter(self, filters, order_by=None):
+        self._check_otp(constants.FILTER)
+        return PolicyBasedController(self).filter(filters, order_by=order_by)
+
+    def delete(self, uuid):
+        self._check_otp(constants.DELETE)
+        PolicyBasedController(self).delete(uuid)
+
+    def update(self, uuid, **kwargs):
+        self._check_otp(constants.UPDATE)
+        return PolicyBasedController(self).update(uuid, **kwargs)

gcl_iam/drivers.py

@@ -16,7 +16,7 @@
 
 import abc
 
-import bazooka
+import bazooka.exceptions
 
 from gcl_iam import exceptions
 
@@ -76,5 +76,5 @@ def get_introspection_info(self, token_info, otp_code=None):
                 introspection_url,
                 headers=headers,
             ).json()
-        except bazooka.exceptions.RequestError:
+        except bazooka.exceptions.BadRequestError:
             raise exceptions.InvalidAuthTokenError()

gcl_iam/engines.py

@@ -95,6 +95,8 @@ def __init__(
             self._introspection_info["permissions"]
         )
 
+        self._introspection_info["otp_enabled"] = self._token_info.otp_enabled
+
     @property
     def token_info(self):
         return self._token_info

gcl_iam/tests/unit/test_controllers.py

@@ -20,6 +20,7 @@
 
 from unittest import mock
 
+from restalchemy.api import constants
 from restalchemy.common import contexts
 
 from gcl_iam import controllers
@@ -41,8 +42,7 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
 
-@pytest.fixture
-def user_context():
+def ctx_storage_context(**kwargs):
     ctx_storage = mock.Mock()
     contexts.ContextWithStorage._store_context_session(ctx_storage)
     ctx_storage.iam_context.introspection_info.return_value = {
@@ -56,31 +56,45 @@ def user_context():
             "genesis_core.vm.admin",
         ],
     }
-
-    yield ctx_storage.iam_context.introspection_info
-
-    contexts.ContextWithStorage._clear_context()
+    ctx_storage.iam_context.introspection_info.return_value.update(kwargs)
+    return ctx_storage.iam_context.introspection_info
 
 
 @pytest.fixture
 def unscoped_context():
-    ctx_storage = mock.Mock()
-    contexts.ContextWithStorage._store_context_session(ctx_storage)
-    ctx_storage.iam_context.introspection_info.return_value = {
-        "user_info": {},
-        "project_id": None,
-        "otp_verified": True,
-        "permission_hash": "xxxx",
-        "permissions": [
+    yield ctx_storage_context(
+        project_id=None,
+        introspection_infopermissions=[
             "service.resource.action",
             "genesis_core.vm.create",
             "genesis_core.vm.admin",
             "*.*.*",
         ],
-    }
+    )
+    contexts.ContextWithStorage._clear_context()
+
+
+@pytest.fixture
+def user_context():
+    yield ctx_storage_context(project_id=FAKE_PROJECT_ID)
+    contexts.ContextWithStorage._clear_context()
 
-    yield ctx_storage.iam_context.introspection_info
 
+@pytest.fixture
+def otp_enabled_context():
+    yield ctx_storage_context(otp_enabled=True)
+    contexts.ContextWithStorage._clear_context()
+
+
+@pytest.fixture
+def otp_not_verified_context():
+    yield ctx_storage_context(otp_enabled=True, otp_verified=False)
+    contexts.ContextWithStorage._clear_context()
+
+
+@pytest.fixture
+def otp_not_enabled_context():
+    yield ctx_storage_context(otp_verified=False)
     contexts.ContextWithStorage._clear_context()
 
 
@@ -164,3 +178,24 @@ def test_override_project_id_from_ctx(self, user_context):
         pc._enforce_and_override_project_id_in_kwargs("create", kwargs)
 
         assert kwargs == {"project_id": FAKE_PROJECT_ID}
+
+
+class TestPolicyBasedCheckOtpController:
+
+    def test_check_otp_verified_true(self, otp_enabled_context):
+        pc = controllers.PolicyBasedCheckOtpController(request=mock.Mock())
+        pc._check_otp(constants.GET)
+
+    def test_check_otp_verified_false(self, otp_not_verified_context):
+        pc = controllers.PolicyBasedCheckOtpController(request=mock.Mock())
+        with pytest.raises(exceptions.OTPInvalidCodeError):
+            pc._check_otp(constants.GET)
+
+    def test_check_otp_mandatory_on(self, otp_not_enabled_context):
+        pc = controllers.PolicyBasedCheckOtpController(request=mock.Mock())
+        with pytest.raises(exceptions.OTPInvalidCodeError):
+            pc._check_otp(constants.CREATE)
+
+    def test_check_otp_mandatory_off(self, otp_not_enabled_context):
+        pc = controllers.PolicyBasedCheckOtpController(request=mock.Mock())
+        pc._check_otp(constants.FILTER)

gcl_iam/tokens.py

@@ -102,6 +102,10 @@ def autenticated_at(self):
     def token_type(self):
         return self._token_info["typ"]
 
+    @property
+    def otp_enabled(self):
+        return self._token_info.get("otp")
+
 
 class IdToken(BaseToken):