Add action param to validate challenge

authsignal/client.py

@@ -1,4 +1,5 @@
 import decimal
+import authsignal
 from authsignal.version import VERSION
 
 import humps
@@ -9,8 +10,7 @@
 
 _UNICODE_STRING = str
 
-API_BASE_URL = 'https://signal.authsignal.com'
-API_CHALLENGE_URL = 'https://api.authsignal.com/v1'
+API_BASE_URL = 'https://api.authsignal.com/v1'
 
 BLOCK = "BLOCK"
 ALLOW = "ALLOW"
@@ -52,8 +52,8 @@ def __init__(
         self.url = api_url
         self.timeout = timeout
         self.version = version
+        self.api_version = 'v1'
 
-
     def track(self, user_id, action, payload=None, path=None):
         """Tracks an action to authsignal, scoped to the user_id and action
         Returns the status of the action so that you can determine to whether to continue
@@ -152,8 +152,7 @@ def get_user(self, user_id, redirect_url=None,  path=None):
     def delete_user(self, user_id):
         _assert_non_empty_unicode(user_id, 'user_id')
 
-        user_id = urllib.parse.quote(user_id)
-        path = f'{self.url}/v1/users/{user_id}'
+        path = self._delete_user_url(user_id)
         headers = self._default_headers()
 
         try:
@@ -235,48 +234,71 @@ def enroll_verified_authenticator(self, user_id, authenticator_payload,  path=No
         except requests.exceptions.RequestException as e:
             raise ApiException(str(e), path) from e
 
-    def validate_challenge(self, token: str, user_id: Optional[str] = None) -> Dict[str, Any]:
-        path = f"{API_CHALLENGE_URL}/validate"
+    def validate_challenge(self, token: str, user_id: Optional[str] = None, action: Optional[str] = None) -> Dict[str, Any]:
+        path = self._validate_challenge_url()
         headers = {
             'Content-Type': 'application/json',
             'Accept': 'application/json'
         }
 
+        payload = {'token': token}
+        if user_id is not None:
+            payload['userId'] = user_id
+        if action is not None:
+            payload['action'] = action
+
         try:
             response = self.session.post(
                 path,
                 auth=requests.auth.HTTPBasicAuth(self.api_key, ''),
-                data=json.dumps({'token': token, 'userId': user_id}),
+                data=json.dumps(payload),
                 headers=headers,
                 timeout=self.timeout
             )
 
             response_data = humps.decamelize(response.json())
 
-            action = response_data.pop('action_code', None)
-
-            return {'action': action, **response_data}
+            return response_data
         except requests.exceptions.RequestException as e:
             raise ApiException(str(e), path) from e
 
     def _default_headers(self):
         return {'Content-type': 'application/json',
                 'Accept': '*/*',
                 'User-Agent': self._user_agent()}
+
     def _user_agent(self):
         return f'Authsignal Python v{self.version}'
 
     def _track_url(self, user_id, action):
-        return f'{self.url}/v1/users/{user_id}/actions/{action}'
-
+        path = self._ensure_versioned_path(f'/users/{user_id}/actions/{action}')
+        return f'{self.url}{path}'
+
     def _get_action_url(self, user_id, action, idempotency_key):
-        return f'{self.url}/v1/users/{user_id}/actions/{action}/{idempotency_key}'
-
+        path = self._ensure_versioned_path(f'/users/{user_id}/actions/{action}/{idempotency_key}')
+        return f'{self.url}{path}'
+
     def _get_user_url(self, user_id):
-        return f'{self.url}/v1/users/{user_id}'
+        path = self._ensure_versioned_path(f'/users/{user_id}')
+        return f'{self.url}{path}'
 
     def _post_enrollment_url(self, user_id):
-        return f'{self.url}/v1/users/{user_id}/authenticators'
+        path = self._ensure_versioned_path(f'/users/{user_id}/authenticators')
+        return f'{self.url}{path}'
+
+    def _validate_challenge_url(self):
+        path = self._ensure_versioned_path(f'/validate')
+        return f'{self.url}{path}'
+
+    def _delete_user_url(self, user_id):
+        user_id = urllib.parse.quote(user_id)
+        path = self._ensure_versioned_path(f'/users/{user_id}')
+        return f'{self.url}{path}'
+
+    def _ensure_versioned_path(self, path):
+        if not self.url.endswith(f'/{self.api_version}'):
+            return f'/{self.api_version}{path}'
+        return path
 
 class ApiException(Exception):
     def __init__(self, message, url, http_status_code=None, body=None, api_status=None,
@@ -307,4 +329,3 @@ def _assert_non_empty_dict(val, name):
         raise TypeError('{0} must be a non-empty dict'.format(name))
     elif not val:
         raise ValueError('{0} must be a non-empty dict'.format(name))
-

authsignal/client_tests.py

@@ -5,9 +5,7 @@
 
 import client
 
-base_url = "https://signal.authsignal.com/v1"
-
-base_challenge_url = 'https://api.authsignal.com/v1'
+base_url = "https://api.authsignal.com/v1"
 
 class Test(unittest.TestCase):
     def setUp(self):
@@ -133,7 +131,7 @@ def setUp(self):
 
     @responses.activate
     def test_it_returns_success_if_user_id_is_correct(self):
-        responses.add(responses.POST, f"{base_challenge_url}/validate",
+        responses.add(responses.POST, f"{base_url}/validate",
             json={
                 'isValid': True, 
                 'state': 'CHALLENGE_SUCCEEDED', 
@@ -151,12 +149,12 @@ def test_it_returns_success_if_user_id_is_correct(self):
         self.assertEqual(response["state"], "CHALLENGE_SUCCEEDED")
         self.assertTrue(response["is_valid"])
 
-
     @responses.activate
     def test_delete_user_authenticator(self):
         self.authsignal_client = client.Client(api_key='test_api_key')
         user_id = 'test_user'
         user_authenticator_id = 'test_authenticator'
+
         expected_url = f'{self.authsignal_client.url}/v1/users/{user_id}/authenticators/{user_authenticator_id}'
 
         responses.add(responses.DELETE, expected_url, json={"success": True}, status=200)
@@ -170,20 +168,34 @@ def test_delete_user_authenticator(self):
 
     @responses.activate
     def test_it_returns_success_false_if_user_id_is_incorrect(self):
-        responses.add(responses.POST, f"{base_challenge_url}/validate",
+        responses.add(responses.POST, f"{base_url}/validate",
             json={'isValid': False, 'error': 'User is invalid.'},
             status=400
         )
 
         response = self.authsignal_client.validate_challenge(user_id="spoofed_id", token=self.jwt_token)
 
-        self.assertIsNone(response['action'])
         self.assertFalse(response['is_valid'])
         self.assertEqual(response.get("error"), "User is invalid.")
 
+    @responses.activate
+    def test_it_returns_isValid_false_if_action_is_incorrect(self):
+        responses.add(responses.POST, f"{base_url}/validate",
+            json={
+                'isValid': False, 
+                'error': 'Action is invalid.',
+            },
+            status=200
+        )
+
+        response = self.authsignal_client.validate_challenge(action="malicious_action_id", token=self.jwt_token)
+
+        # self.assertEqual(response["error"], "CHALLENGE_SUCCEEDED")
+        self.assertFalse(response["is_valid"])
+
     @responses.activate
     def test_it_returns_success_true_if_no_user_id_is_provided(self):
-        responses.add(responses.POST, f"{base_challenge_url}/validate",
+        responses.add(responses.POST, f"{base_url}/validate",
             json={
                 'isValid': True, 
                 'state': 'CHALLENGE_SUCCEEDED', 

authsignal/version.py

@@ -1 +1 @@
-VERSION = '2.0.6'
+VERSION = '2.0.7'

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "authsignal"
-version = "2.0.6"
+version = "2.0.7"
 description = "Authsignal Python SDK for Passwordless Step Up Authentication"
 authors = ["justinsoong <[email protected]>"]
 license = "MIT"