This is an automated email from the ASF dual-hosted git repository.
hsluoyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/casbin-pycasbin.git
The following commit(s) were added to refs/heads/master by this push:
new 12edbf9 fix: keep rm and cond_rm usable after policy reload
12edbf9 is described below
commit 12edbf95f7a7b24215be1eb9a7898854262e0cd0
Author: Yang Luo <[email protected]>
AuthorDate: Tue Aug 4 14:28:44 2026 +0800
fix: keep rm and cond_rm usable after policy reload
---
casbin/async_internal_enforcer.py | 7 ++++-
casbin/model/assertion.py | 15 +++++++---
tests/test_enforcer.py | 62 +++++++++++++++++++++++++++++++++++++++
3 files changed, 79 insertions(+), 5 deletions(-)
diff --git a/casbin/async_internal_enforcer.py
b/casbin/async_internal_enforcer.py
index 0c65728..275a0ae 100644
--- a/casbin/async_internal_enforcer.py
+++ b/casbin/async_internal_enforcer.py
@@ -62,8 +62,13 @@ class AsyncInternalEnforcer(CoreEnforcer):
need_to_rebuild = True
for rm in self.rm_map.values():
rm.clear()
+ if len(self.rm_map) != 0:
+ new_model.build_role_links(self.rm_map)
- new_model.build_role_links(self.rm_map)
+ for crm in self.cond_rm_map.values():
+ crm.clear()
+ if len(self.cond_rm_map) != 0:
+ new_model.build_conditional_role_links(self.cond_rm_map)
self.model = new_model
diff --git a/casbin/model/assertion.py b/casbin/model/assertion.py
index 88096a0..f92da8d 100644
--- a/casbin/model/assertion.py
+++ b/casbin/model/assertion.py
@@ -12,8 +12,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import logging
import copy
+import logging
+
from casbin.model.policy_op import PolicyOp
@@ -111,15 +112,21 @@ class Assertion:
self.cond_rm.set_domain_link_condition_func_params(rule[0],
rule[1], domain, *rule[len(self.tokens) :])
def __deepcopy__(self, memo):
- """Custom deepcopy implementation that excludes rm and cond_rm
attributes.
- This stems from an issue with the watcher concurrent reloading causing
an edge case of deepcopy error.
+ """Custom deepcopy implementation that copies policy data only,
+ sharing rm and cond_rm by reference instead of cloning them.
+
+ The role managers are live, shared, cyclic object graphs: deep-copying
them
+ races with concurrent add_link() calls from another thread (as happens
when a
+ watcher reloads the policy), and the copy is discarded by
build_role_links()
+ anyway. This matches casbin-go, whose Assertion.copy() also carries RM
and
+ CondRM over by reference.
"""
cls = self.__class__
result = cls.__new__(cls)
memo[id(self)] = result
for k, v in self.__dict__.items():
if k in ("rm", "cond_rm"):
- setattr(result, k, None)
+ setattr(result, k, v)
else:
setattr(result, k, copy.deepcopy(v, memo))
return result
diff --git a/tests/test_enforcer.py b/tests/test_enforcer.py
index 5ad0626..d1b5f32 100644
--- a/tests/test_enforcer.py
+++ b/tests/test_enforcer.py
@@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import copy
import os
import time
from unittest import TestCase, IsolatedAsyncioTestCase
@@ -1041,3 +1042,64 @@ class TestConfigAsync(IsolatedAsyncioTestCase):
self.assertTrue(e.enforce("alice", "data2", "read"))
self.assertTrue(e.enforce("alice", "data3", "scribble"))
self.assertFalse(e.enforce("alice", "data4", "scribble"))
+
+ async def test_load_policy_rebuilds_conditional_role_links(self):
+ e = self.get_enforcer(
+ get_examples("rbac_with_temporal_roles_model.conf"),
+ get_examples("rbac_with_temporal_roles_policy.csv"),
+ )
+ await e.load_policy()
+
+ ast = e.model.model["g"]["g"]
+ self.assertIsNotNone(ast.cond_rm)
+ self.assertIs(ast.cond_rm, e.cond_rm_map["g"])
+
+ e.add_named_link_condition_func("g", "alice", "data2_admin",
util.time_match_func)
+ self.assertTrue(e.enforce("alice", "data1", "read"))
+ self.assertFalse(e.enforce("alice", "data2", "read"))
+
+
+class TestModelDeepCopy(TestCase):
+ """Assertion.__deepcopy__ must not clone the live role managers.
+
+ They are shared, cyclic object graphs; deep-copying them races with
concurrent
+ add_link() calls and the copy would be thrown away by build_role_links()
anyway.
+ """
+
+ def test_deepcopy_shares_role_managers(self):
+ e = casbin.Enforcer(
+ get_examples("rbac_model.conf"),
+ get_examples("rbac_policy.csv"),
+ )
+ ast = e.model.model["g"]["g"]
+
+ new_model = copy.deepcopy(e.model)
+ new_ast = new_model.model["g"]["g"]
+
+ self.assertIs(new_ast.rm, ast.rm)
+ self.assertIs(new_ast.cond_rm, ast.cond_rm)
+ # policy data is still copied, not shared
+ self.assertIsNot(new_ast.policy, ast.policy)
+ self.assertEqual(new_ast.policy, ast.policy)
+
+ def test_load_policy_keeps_role_manager(self):
+ e = casbin.Enforcer(
+ get_examples("rbac_model.conf"),
+ get_examples("rbac_policy.csv"),
+ )
+ e.load_policy()
+
+ self.assertIs(e.model.model["g"]["g"].rm, e.rm_map["g"])
+ self.assertTrue(e.enforce("alice", "data2", "read"))
+
+ def test_load_policy_without_auto_build_role_links(self):
+ e = casbin.Enforcer(
+ get_examples("rbac_model.conf"),
+ get_examples("rbac_policy.csv"),
+ )
+ e.enable_auto_build_role_links(False)
+ e.load_policy()
+
+ # the role manager must still be usable even though nothing rebuilt
the links
+ self.assertIsNotNone(e.model.model["g"]["g"].rm)
+ self.assertTrue(e.enforce("alice", "data2", "read"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
