feat: Append enforce function to sdk

.github/workflows/build.yml

@@ -19,7 +19,7 @@ jobs:
       - name: Setup Node.js
         uses: actions/setup-node@v2
         with:
-          node-version: '16'
+          node-version: '18'
 
       - name: Setup
         run: npm install -g semantic-release @semantic-release/github @semantic-release/changelog @semantic-release/commit-analyzer @semantic-release/git @semantic-release/release-notes-generator semantic-release-pypi

README.md

@@ -113,3 +113,4 @@ casdoor-python-sdk support basic user operations, like:
 - `get_users()`, get all users.
 - `modify_user(method: str, user: User)/add_user(user: User)/update_user(user: User)/delete_user(user: User)`, write user to database.
 - `refresh_token_request(refresh_token: str, scope: str)`, refresh access token
+- `enforce(self, permission_model_name: str, sub: str, obj: str, act: str)`, check permission from model

src/casdoor/async_main.py

@@ -59,8 +59,8 @@ def __del__(self):
     @property
     def certification(self) -> bytes:
         if type(self.certificate) is not str:
-            raise TypeError('certificate field must be str type')
-        return self.certificate.encode('utf-8')
+            raise TypeError("certificate field must be str type")
+        return self.certificate.encode("utf-8")
 
     async def get_auth_link(
             self,
@@ -165,6 +165,49 @@ def parse_jwt_token(self, token: str) -> dict:
         )
         return return_json
 
+    async def enforce(
+            self,
+            permission_model_name: str,
+            sub: str,
+            obj: str,
+            act: str
+    ) -> bool:
+        """
+        Send data to Casdoor enforce API
+        :param permission_model_name: Name permission model
+        :param sub: sub from Casbin
+        :param obj: obj from Casbin
+        :param act: act from Casbin
+        """
+        url = self.endpoint + "/api/enforce"
+        query_params = {
+            "clientId": self.client_id,
+            "clientSecret": self.client_secret
+        }
+        params = {
+            "id": permission_model_name,
+            "v0": sub,
+            "v1": obj,
+            "v2": act,
+        }
+        async with self._session.post(
+                url, params=query_params, json=params
+        ) as response:
+            if (
+                    response.status != 200 or
+                    "json" not in response.headers["content-type"]
+            ):
+                error_str = "Casdoor response error:\n" + str(response.text)
+                raise ValueError(error_str)
+
+            has_permission = await response.json()
+
+            if not isinstance(has_permission, bool):
+                error_str = "Casdoor response error:\n" + await response.text()
+                raise ValueError(error_str)
+
+            return has_permission
+
     async def get_users(self) -> List[dict]:
         """
         Get the users from Casdoor.

src/casdoor/main.py

@@ -53,8 +53,8 @@ def __init__(
     @property
     def certification(self) -> bytes:
         if type(self.certificate) is not str:
-            raise TypeError('certificate field must be str type')
-        return self.certificate.encode('utf-8')
+            raise TypeError("certificate field must be str type")
+        return self.certificate.encode("utf-8")
 
     def get_auth_link(
             self,
@@ -154,6 +154,44 @@ def parse_jwt_token(self, token: str) -> dict:
         )
         return return_json
 
+    def enforce(
+            self,
+            permission_model_name: str,
+            sub: str,
+            obj: str,
+            act: str
+    ) -> bool:
+        """
+        Send data to Casdoor enforce API
+        :param permission_model_name: Name permission model
+        :param sub: sub from Casbin
+        :param obj: obj from Casbin
+        :param act: act from Casbin
+        """
+        url = self.endpoint + "/api/enforce"
+        query_params = {
+            "clientId": self.client_id,
+            "clientSecret": self.client_secret
+        }
+        params = {
+            "id": permission_model_name,
+            "v0": sub,
+            "v1": obj,
+            "v2": act,
+        }
+        r = requests.post(url, json=params, params=query_params)
+        if r.status_code != 200 or "json" not in r.headers["content-type"]:
+            error_str = "Casdoor response error:\n" + str(r.text)
+            raise ValueError(error_str)
+
+        has_permission = r.json()
+
+        if not isinstance(has_permission, bool):
+            error_str = "Casdoor response error:\n" + r.text
+            raise ValueError(error_str)
+
+        return has_permission
+
     def get_users(self) -> List[dict]:
         """
         Get the users from Casdoor.

src/tests/test_async_oauth.py

@@ -75,6 +75,13 @@ async def test_parse_jwt_token(self):
         decoded_msg = sdk.parse_jwt_token(access_token)
         self.assertIsInstance(decoded_msg, dict)
 
+    async def test_enforce(self):
+        sdk = self.get_sdk()
+        status = await sdk.enforce(
+            "built-in/permission-built-in", "admin", "a", "ac"
+        )
+        self.assertIsInstance(status, bool)
+
     async def test_get_users(self):
         sdk = self.get_sdk()
         users = await sdk.get_users()

src/tests/test_oauth.py

@@ -75,6 +75,13 @@ def test_parse_jwt_token(self):
         decoded_msg = sdk.parse_jwt_token(access_token)
         self.assertIsInstance(decoded_msg, dict)
 
+    def test_enforce(self):
+        sdk = self.get_sdk()
+        status = sdk.enforce(
+            "built-in/permission-built-in", "admin", "a", "ac"
+        )
+        self.assertIsInstance(status, bool)
+
     def test_get_users(self):
         sdk = self.get_sdk()
         users = sdk.get_users()