Merge pull request #3 from warrant-dev/v2Authz

akajla09 · web-flow · commit 174b7d7aac2f · 2022-07-09T19:39:51.000-07:00
Add RBAC methods and update to v2 authz endpoint
diff --git a/examples/example.py b/examples/example.py
@@ -1,25 +1,64 @@
 from warrant import *
 
 def make_warrant_requests(api_key):
-    client = Warrant(api_key)
+    client = WarrantClient(api_key)
 
-    # Create users and session tokens
-    print("Created tenant with provided id: " + client.create_tenant("custom_tenant_001"))
-    print("Created user with provided id: " + client.create_user("custom_user_001"))
-    new_user = client.create_user()
-    print("Created user with generated id: " + new_user)
-    new_subject = Subject("user", new_user)
-    print(client.create_warrant(object_type="tenant", object_id="custom_tenant_001", relation="member", subject=new_subject))
-    print("Created session token: " + client.create_session(new_user))
+    # Create users, tenants, roles, permissions
+    user1 = client.create_user()
+    print("Created user with generated id: " + user1)
+    provided_user_id = "custom_user_100"
+    user2 = client.create_user(provided_user_id)
+    print("Created user with provided id: " + user2)
+    print("Created session token for user " + user1 + ": " + client.create_session(user1))
+    print("Created session token for user " + user2 + ": " + client.create_session(user2))
+    tenant1 = client.create_tenant("custom_tenant_210")
+    print("Created tenant with provided id: " + tenant1)
+    admin_role = client.create_role("admin1")
+    print("Created role: " + admin_role)
+    permission1 = client.create_permission("create_report")
+    print("Created permission: " + permission1)
+    permission2 = client.create_permission("delete_report")
+    print("Created permission: " + permission2)
+    print("Assigned role " + client.assign_role_to_user(user1, admin_role) + " to user " + user1)
+    print("Assigned permission " + client.assign_permission_to_user(user1, permission1) + " to user " + user1)
+    print("Assigned permission " + client.assign_permission_to_role(admin_role, permission2) + " to role " + admin_role)
 
-    # Create and check warrants
-    print(client.create_warrant(object_type="store", object_id="store1", relation="owner", subject=new_subject))
-    subject_to_check = Subject("user", new_user)
-    is_authorized = client.is_authorized(object_type="store", object_id="store1", relation="owner", subject_to_check=subject_to_check)
-    print(f"New user authorization result: {is_authorized}")
-    print(f"All warrants: {client.list_warrants()}")
+    # Create and test warrants
+    user1_subject = Subject("user", user1)
+    print("--- Testing Warrants ---")
+    print(client.create_warrant(object_type="tenant", object_id=tenant1, relation="member", subject=user1_subject))
+    subject_to_check = Subject("user", user1)
+    warrants_to_check = [Warrant("tenant", tenant1, "member", subject_to_check)]
+    is_authorized = client.is_authorized(WarrantCheck(warrants_to_check, "allOf"))
+    print(f"Tenant check authorization result: {is_authorized}")
+    role_warrants_to_check = [Warrant("role", admin_role, "member", subject_to_check)]
+    role_check = client.is_authorized(WarrantCheck(role_warrants_to_check, "allOf"))
+    print(f"Admin role check authorization result: {role_check}")
+    permission_warrants_to_check = [Warrant("permission", permission1, "member", subject_to_check)]
+    permission_check = client.is_authorized(WarrantCheck(permission_warrants_to_check, "allOf"))
+    print(f"create_report permission check authorization result: {permission_check}")
+    role_subject = Subject("role", admin_role)
+    role_permission_warrants_to_check = [Warrant("permission", permission2, "parent", role_subject)]
+    role_permission_check = client.is_authorized(WarrantCheck(role_permission_warrants_to_check, "allOf"))
+    print(f"create_report role/permission check authorization result: {role_permission_check}")
+    print(f"List all warrants: {client.list_warrants()}")
+
+    # Delete users, tenants, roles, permissions
+    client.remove_permission_from_role(admin_role, permission2)
+    client.remove_permission_from_user(user1, permission1)
+    client.remove_role_from_user(user1, admin_role)
+    client.delete_user(user1)
+    print("Deleted user " + user1)
+    client.delete_user(user2)
+    print("Deleted user " + user2)
+    client.delete_tenant(tenant1)
+    print("Deleted tenant " + tenant1)
+    client.delete_role(admin_role)
+    print("Deleted role " + admin_role)
+    client.delete_permission(permission1)
+    print("Deleted permission " + permission1)
 
 if __name__ == '__main__':
     # Replace with your Warrant api key
     api_key = "API_KEY"
-    make_warrant_requests(api_key)
+    make_warrant_requests(api_key)
diff --git a/warrant/__init__.py b/warrant/__init__.py
@@ -4,7 +4,6 @@
 __version__ = "0.2.1"
 
 API_ENDPOINT = "https://api.warrant.dev"
-API_VERSION = "/v1"
 
 class WarrantException(Exception):
     def __init__(self, msg, status_code=-1):
@@ -21,45 +20,134 @@ def __init__(self, object_type, object_id, relation=""):
         self.relation = relation
 
 class Warrant(object):
+    def __init__(self, object_type, object_id, relation, subject):
+        self.objectType = object_type
+        self.objectId = object_id
+        self.relation = relation
+        self.subject = subject
+
+class WarrantCheck(object):
+    def __init__(self, warrants, op):
+        self.warrants = warrants
+        self.op = op
+
+class WarrantClient(object):
     def __init__(self, api_key):
         self._apiKey = api_key
 
     def _make_post_request(self, uri, json={}):
         headers = { "Authorization": "ApiKey " + self._apiKey }
-        resp = requests.post(url = API_ENDPOINT+API_VERSION+uri, headers = headers, json = json)
+        resp = requests.post(url = API_ENDPOINT+uri, headers = headers, json = json)
         if resp.status_code == 200:
             return resp.json()
         else:
             raise WarrantException(msg=resp.text, status_code=resp.status_code)
 
     def _make_get_request(self, uri, params={}):
         headers = { "Authorization": "ApiKey " + self._apiKey }
-        resp = requests.get(url = API_ENDPOINT+API_VERSION+uri, headers = headers, params = params)
+        resp = requests.get(url = API_ENDPOINT+uri, headers = headers, params = params)
         if resp.status_code == 200:
             return resp.json()
         else:
             raise WarrantException(msg=resp.text, status_code=resp.status_code)
 
+    def _make_delete_request(self, uri, params={}):
+        headers = { "Authorization": "ApiKey " + self._apiKey }
+        resp = requests.delete(url = API_ENDPOINT+uri, headers = headers, params = params)
+        if resp.status_code != 200:
+            raise WarrantException(msg=resp.text, status_code=resp.status_code)
+
     def create_user(self, user_id=""):
         if user_id == "":
             payload = {}
         else:
             payload = { "userId": user_id }
-        json = self._make_post_request(uri="/users", json=payload)
+        json = self._make_post_request(uri="/v1/users", json=payload)
         return json['userId']
 
+    def delete_user(self, user_id):
+        if user_id == "":
+            raise WarrantException(msg="Must include a userId")
+        self._make_delete_request(uri="/v1/users/"+user_id)
+
     def create_tenant(self, tenant_id=""):
         if tenant_id == "":
             payload = {}
         else:
             payload = { "tenantId": tenant_id }
-        json = self._make_post_request(uri="/tenants", json=payload)
+        json = self._make_post_request(uri="/v1/tenants", json=payload)
         return json['tenantId']
 
+    def delete_tenant(self, tenant_id):
+        if tenant_id == "":
+            raise WarrantException(msg="Must include a tenantId")
+        self._make_delete_request(uri="/v1/tenants/"+tenant_id)
+
+    def create_role(self, role_id):
+        if role_id == "":
+            raise WarrantException(msg="Must include a roleId")
+        payload = { "roleId": role_id }
+        json = self._make_post_request(uri="/v1/roles", json=payload)
+        return json['roleId']
+
+    def delete_role(self, role_id):
+        if role_id == "":
+            raise WarrantException(msg="Must include a roleId")
+        self._make_delete_request(uri="/v1/roles/"+role_id)
+
+    def create_permission(self, permission_id):
+        if permission_id == "":
+            raise WarrantException(msg="Must include a permissionId")
+        payload = { "permissionId": permission_id }
+        json = self._make_post_request(uri="/v1/permissions", json=payload)
+        return json['permissionId']
+
+    def delete_permission(self, permission_id):
+        if permission_id == "":
+            raise WarrantException(msg="Must include a permissionId")
+        self._make_delete_request(uri="/v1/permissions/"+permission_id)
+
+    def assign_role_to_user(self, user_id, role_id):
+        if user_id == "" or role_id == "":
+            raise WarrantException(msg="Must include a userId and roleId")
+        json = self._make_post_request(uri="/v1/users/" + user_id + "/roles/" + role_id)
+        return json['roleId']
+
+    def remove_role_from_user(self, user_id, role_id):
+        if user_id == "" or role_id == "":
+            raise WarrantException(msg="Must include a userId and roleId")
+        self._make_delete_request(uri="/v1/users/"+user_id+"/roles/"+role_id)
+
+    def assign_permission_to_user(self, user_id, permission_id):
+        if user_id == "" or permission_id == "":
+            raise WarrantException(msg="Must include a userId and permissionId")
+        json = self._make_post_request(uri="/v1/users/" + user_id + "/permissions/" + permission_id)
+        return json['permissionId']
+
+    def remove_permission_from_user(self, user_id, permission_id):
+        if user_id == "" or permission_id == "":
+            raise WarrantException(msg="Must include a userId and permissionId")
+        self._make_delete_request(uri="/v1/users/"+user_id+"/permissions/"+permission_id)
+
+    def assign_permission_to_role(self, role_id, permission_id):
+        if role_id == "" or permission_id == "":
+            raise WarrantException(msg="Must include a roleId and permissionId")
+        json = self._make_post_request(uri="/v1/roles/" + role_id + "/permissions/" + permission_id)
+        return json['permissionId']
+
+    def remove_permission_from_role(self, role_id, permission_id):
+        if role_id == "" or permission_id == "":
+            raise WarrantException(msg="Must include a roleId and permissionId")
+        self._make_delete_request(uri="/v1/roles/"+role_id+"/permissions/"+permission_id)
+
     def create_session(self, user_id):
         if user_id == "":
             raise WarrantException(msg="Invalid userId provided")
-        json = self._make_post_request(uri="/users/"+user_id+"/sessions")
+        payload = {
+            "type": "sess",
+            "userId": user_id
+        }
+        json = self._make_post_request(uri="/v1/sessions", json=payload)
         return json['token']
 
     def create_warrant(self, object_type, object_id, relation, subject):
@@ -74,7 +162,7 @@ def create_warrant(self, object_type, object_id, relation, subject):
             payload["subject"] = subject.__dict__
         else:
             raise WarrantException(msg="Invalid type for \'subject\'. Must be of type Subject")
-        resp = self._make_post_request(uri="/warrants", json=payload)
+        resp = self._make_post_request(uri="/v1/warrants", json=payload)
         return resp['id']
 
     def list_warrants(self, object_type="", object_id="", relation="", user_id=""):
@@ -84,26 +172,20 @@ def list_warrants(self, object_type="", object_id="", relation="", user_id=""):
             "relation": relation,
             "userId": user_id,
         }
-        resp = self._make_get_request(uri="/warrants", params=filters)
+        resp = self._make_get_request(uri="/v1/warrants", params=filters)
         return resp
 
-    def is_authorized(self, object_type, object_id, relation, subject_to_check):
-        if object_type == "" or object_id == "" or relation == "":
-            raise WarrantException(msg="Invalid object_type, object_id and/or relation")
-        payload = {
-            "objectType": object_type,
-            "objectId": object_id,
-            "relation": relation
-        }
-        if isinstance(subject_to_check, Subject):
-            payload["subject"] = subject_to_check.__dict__
-        else:
-            raise WarrantException(msg="Invalid type for \'subject_to_check\'. Must be of type Subject")
+    def is_authorized(self, warrant_check):
+        if not isinstance(warrant_check.warrants, list):
+            raise WarrantException(msg="Invalid list of warrants to check")
+        payload = json.dumps(warrant_check, default = lambda x: x.__dict__)
         headers = { "Authorization": "ApiKey " + self._apiKey }
-        resp = requests.post(url = API_ENDPOINT+API_VERSION+"/authorize", headers = headers, json=payload)
-        if resp.status_code == 200:
+        resp = requests.post(url = API_ENDPOINT+"/v2/authorize", headers = headers, data=payload)
+        if resp.status_code != 200:
+            raise WarrantException(msg=resp.text, status_code=resp.status_code)
+        response_payload = resp.json()
+        result = response_payload['code']
+        if result == 200:
             return True
-        elif resp.status_code == 401:
-            return False
         else:
-            raise WarrantException(msg=resp.text, status_code=resp.status_code)
+            return False
