Skip to content

feat: adapt async enforce and batch-enforce APIs to latest format #104

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hsluoyz merged 1 commit into from
Jul 30, 2025
Merged

feat: adapt async enforce and batch-enforce APIs to latest format #104

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
156 changes: 99 additions & 57 deletions src/casdoor/async_main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
# limitations under the License.

import base64
import json
from typing import Dict, List, Optional

import aiohttp
Expand Down Expand Up @@ -263,73 +264,114 @@ def parse_jwt_token(self, token: str, **kwargs) -> Dict:

async def enforce(
self,
permission_model_name: str,
sub: str,
obj: str,
act: str,
v3: Optional[str] = None,
v4: Optional[str] = None,
v5: Optional[str] = None,
permission_id: str,
model_id: str,
resource_id: str,
enforce_id: str,
owner: str,
casbin_request: Optional[List[str]] = None,
) -> bool:
"""
Send data to Casdoor enforce API
# https://casdoor.org/docs/permission/exposed-casbin-apis#enforce

:param permission_model_name: Name permission model
:param sub: sub from Casbin
:param obj: obj from Casbin
:param act: act from Casbin
:param v3: v3 from Casbin
:param v4: v4 from Casbin
:param v5: v5 from Casbin
Send data to Casdoor enforce API asynchronously

:param permission_id: the permission id (i.e. organization name/permission name)
:param model_id: the model id
:param resource_id: the resource id
:param enforce_id: the enforce id
:param owner: the owner of the permission
:param casbin_request: a list containing the request data (i.e. sub, obj, act)
:return: a boolean value indicating whether the request is allowed
"""
path = "/api/enforce"
params = {
"id": permission_model_name,
"v0": sub,
"v1": obj,
"v2": act,
"v3": v3,
"v4": v4,
"v5": v5,
url = "/api/enforce"
params: Dict[str, str] = {
"permissionId": permission_id,
"modelId": model_id,
"resourceId": resource_id,
"enforceId": enforce_id,
"owner": owner,
}

async with self._session as session:
has_permission = await session.post(path, headers=self.headers, json=params)
if not isinstance(has_permission, bool):
raise ValueError(f"Casdoor response error: {has_permission}")
return has_permission
response = await session.post(
url,
params=params,
data=json.dumps(casbin_request),
auth=aiohttp.BasicAuth(self.client_id, self.client_secret),
headers={"Content-Type": "application/json"},
)

if isinstance(response, dict):
data = response.get("data")
if isinstance(data, list) and len(data) > 0:
has_permission = data[0]
else:
has_permission = response
else:
has_permission = response

async def batch_enforce(self, permission_model_name: str, permission_rules: List[List[str]]) -> List[bool]:
"""
Send data to Casdoor enforce API

:param permission_model_name: Name permission model
:param permission_rules: permission rules to enforce
[][0] -> sub: sub from Casbin
[][1] -> obj: obj from Casbin
[][2] -> act: act from Casbin
[][3] -> v3: v3 from Casbin (optional)
[][4] -> v4: v4 from Casbin (optional)
[][5] -> v5: v5 from Casbin (optional)
"""
path = "/api/batch-enforce"
if not isinstance(has_permission, bool):
error_str = f"Casdoor response error (invalid type {type(has_permission)}):\n{json.dumps(response)}"
raise ValueError(error_str)

def map_rule(rule: List[str], idx) -> Dict:
if len(rule) < 3:
raise ValueError(f"Invalid permission rule[{idx}]: {rule}")
result = {"id": permission_model_name}
for i in range(len(rule)):
result.update({f"v{i}": rule[i]})
return result
return has_permission

params = [map_rule(permission_rules[i], i) for i in range(len(permission_rules))]
async def batch_enforce(
self,
permission_id: str,
model_id: str,
enforce_id: str,
owner: str,
casbin_request: Optional[List[List[str]]] = None,
) -> List[bool]:
"""
Send data to Casdoor batch enforce API asynchronously

:param permission_id: the permission id (i.e. organization name/permission name)
:param model_id: the model id
:param enforce_id: the enforce id
:param owner: the owner of the permission
:param casbin_request: a list of lists containing the request data
:return: a list of boolean values indicating whether each request is allowed
"""
url = "/api/batch-enforce"
params = {
"permissionId": permission_id,
"modelId": model_id,
"enforceId": enforce_id,
"owner": owner,
}

async with self._session as session:
enforce_results = await session.post(path, headers=self.headers, json=params)
if not isinstance(enforce_results, bool):
raise ValueError(f"Casdoor response error:{enforce_results}")

return enforce_results
response = await session.post(
url,
params=params,
data=json.dumps(casbin_request),
auth=aiohttp.BasicAuth(self.client_id, self.client_secret),
headers={"Content-Type": "application/json"},
)

data = response.get("data")
if data is None:
error_str = "Casdoor response error: 'data' field is missing\n" + json.dumps(response)
raise ValueError(error_str)

if not isinstance(data, list):
error_str = f"Casdoor 'data' is not a list (got {type(data)}):\n{json.dumps(response)}"
raise ValueError(error_str)

enforce_results = data[0] if data else []

if (
not isinstance(enforce_results, list)
or len(enforce_results) > 0
and not isinstance(enforce_results[0], bool)
):
error_str = (
f"Casdoor response contains invalid results (got {type(enforce_results)}):\n{json.dumps(response)}"
)
raise ValueError(error_str)

return enforce_results

async def get_users(self) -> Dict:
"""
Expand Down
2 changes: 1 addition & 1 deletion src/casdoor/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,7 @@ def parse_jwt_token(self, token: str, **kwargs) -> Dict:
certificate = x509.load_pem_x509_certificate(self.certification, default_backend())

return_json = jwt.decode(
token.encode("utf-8"),
token,
certificate.public_key(),
algorithms=self.algorithms,
audience=self.client_id,
Expand Down
101 changes: 19 additions & 82 deletions src/tests/test_async_oauth.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from unittest import IsolatedAsyncioTestCase, mock
from unittest import IsolatedAsyncioTestCase

import src.tests.test_util as test_util
from src.casdoor.async_main import AsyncCasdoorSDK
Expand All @@ -28,7 +28,7 @@ class TestOAuth(IsolatedAsyncioTestCase):
"""

# server returned authorization code
code = "6d038ac60d4e1f17e742"
code = "21dc0ac806c27e6d7962"

# Casdoor user and password for auth with
# Resource Owner Password Credentials Grant.
Expand Down Expand Up @@ -135,92 +135,29 @@ async def test_parse_jwt_token(self):
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)

def mocked_enforce_requests_post(*args, **kwargs):
class MockResponse:
def __init__(
self,
json_data,
status_code=200,
):
self.json_data = json_data
self.status_code = status_code

def json(self):
return self.json_data

result = True
for i in range(0, 5):
if kwargs.get("json").get(f"v{i}") != f"v{i}":
result = False

return MockResponse(result)

@mock.patch("aiohttp.ClientSession.post", side_effect=mocked_enforce_requests_post)
async def test_enforce_parmas(self, mock_post):
sdk = self.get_sdk()
status = await sdk.enforce(
"built-in/permission-built-in",
"v0",
"v1",
"v2",
v3="v3",
v4="v4",
v5="v5",
permission_id="built-in/permission-built-in",
model_id="",
resource_id="",
enforce_id="",
owner="",
casbin_request=["alice", "data1", "read"],
)
self.assertEqual(status, True)

def mocked_batch_enforce_requests_post(*args, **kwargs):
class MockResponse:
def __init__(
self,
json_data,
status_code=200,
):
self.json_data = json_data
self.status_code = status_code

def json(self):
return self.json_data

json = kwargs.get("json")
result = [True for i in range(0, len(json))]
for k in range(0, len(json)):
for i in range(0, len(json[k]) - 1):
if json[k].get(f"v{i}") != f"v{i}":
result[k] = False

return MockResponse(result)

@mock.patch(
"aiohttp.ClientSession.post",
side_effect=mocked_batch_enforce_requests_post,
)
def test_batch_enforce(self, mock_post):
self.assertIsInstance(status, bool)

async def test_batch_enforce(self):
sdk = self.get_sdk()
status = sdk.batch_enforce(
"built-in/permission-built-in",
[
["v0", "v1", "v2", "v3", "v4", "v5"],
["v0", "v1", "v2", "v3", "v4", "v1"],
],
status = await sdk.batch_enforce(
permission_id="built-in/permission-built-in",
model_id="",
enforce_id="",
owner="",
casbin_request=[["alice", "data1", "read"], ["bob", "data2", "write"]],
)
self.assertEqual(len(status), 2)
self.assertEqual(status[0], True)
self.assertEqual(status[1], False)

@mock.patch(
"aiohttp.ClientSession.post",
side_effect=mocked_batch_enforce_requests_post,
)
def test_batch_enforce_raise(self, mock_post):
sdk = self.get_sdk()
with self.assertRaises(ValueError) as context:
sdk.batch_enforce("built-in/permission-built-in", [["v0", "v1"]])
self.assertEqual("Invalid permission rule[0]: ['v0', 'v1']", str(context.exception))
self.assertIsInstance(status[0], bool)
self.assertIsInstance(status[1], bool)

async def test_get_users(self):
sdk = self.get_sdk()
Expand Down