This is an automated email from the ASF dual-hosted git repository. EmryZhang pushed a commit to branch prep/pycasbin-authorization-fixes in repository https://gitbox.apache.org/repos/asf/casbin-pycasbin.git
commit 43b7c2f4c8699836e855c9d36993701382acb85c Author: Zhangyanrui123 <[email protected]> AuthorDate: Sat Nov 7 15:53:02 2026 +0800 fix: preserve authorization results and conditional cache validity --- casbin/cached_enforcer.py | 11 ++-- casbin/core_enforcer.py | 2 +- tests/test_cached_enforcer.py | 15 +++++ tests/test_explanation_copy.py | 77 ++++++++++++++++++++++++ tests/test_release_regressions.py | 119 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 218 insertions(+), 6 deletions(-) diff --git a/casbin/cached_enforcer.py b/casbin/cached_enforcer.py index f7a06fb..58ed1bd 100644 --- a/casbin/cached_enforcer.py +++ b/casbin/cached_enforcer.py @@ -12,6 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +import math + from casbin.cache import DefaultCache from casbin.enforcer import Enforcer @@ -102,12 +104,12 @@ class CacheSupport: decision is not a pure function of the request and caching it would keep granting access after the role has expired. - Setting an expire time is treated as accepting a known, bounded amount of - staleness, so it re-enables caching. + Setting a positive, finite expire time accepts a bounded amount of + staleness and re-enables caching. An unbounded expiry must not do so. """ if not self.cond_rm_map: return False - if self._expire_time is not None: + if self._expire_time is not None and self._expire_time > 0 and math.isfinite(self._expire_time): return False if not self._warned_about_dynamic_model: @@ -134,8 +136,7 @@ class CacheSupport: cached = self._cache.get(key, _MISSING) if cached is not _MISSING: result, explain = cached - # hand out a copy: the explanation upstream is the policy row itself, and a - # caller mutating it would corrupt both the cache and the model. + # Hand out a fresh explanation so callers cannot alter later cache hits. return result, list(explain) result, explain = super().enforce_ex(*rvals) diff --git a/casbin/core_enforcer.py b/casbin/core_enforcer.py index e904483..15309cd 100644 --- a/casbin/core_enforcer.py +++ b/casbin/core_enforcer.py @@ -663,7 +663,7 @@ class CoreEnforcer: explain_rule = [] if explain_index != -1 and explain_index < policy_len: - explain_rule = self.model["p"][ptype].policy[explain_index] + explain_rule = list(self.model["p"][ptype].policy[explain_index]) return result, explain_rule diff --git a/tests/test_cached_enforcer.py b/tests/test_cached_enforcer.py index 004bbf6..dfcc137 100644 --- a/tests/test_cached_enforcer.py +++ b/tests/test_cached_enforcer.py @@ -629,6 +629,21 @@ class TestConditionalRolesAreNeverCached(TestCase): self.assertTrue(e.enforce("alice", "data2", "read")) self.assertEqual(len(e.get_cache()), 1) + def test_unbounded_expiry_does_not_cache_conditional_roles(self): + for expiry in (0, -1, float("inf"), float("nan")): + with self.subTest(expiry=expiry): + e = self.get_enforcer() + e.set_expire_time(expiry) + e.set_named_link_condition_func_params( + "g", "alice", "data2_admin", "2000-01-01 00:00:00", "2100-01-01 00:00:00" + ) + self.assertTrue(e.enforce("alice", "data2", "read")) + e.set_named_link_condition_func_params( + "g", "alice", "data2_admin", "2000-01-01 00:00:00", "2000-01-02 00:00:00" + ) + self.assertFalse(e.enforce("alice", "data2", "read")) + self.assertEqual(len(e.get_cache()), 0) + class TestSyncedCachedEnforcer(TestCase): def get_enforcer(self, model, policy): diff --git a/tests/test_explanation_copy.py b/tests/test_explanation_copy.py new file mode 100644 index 0000000..5e6c623 --- /dev/null +++ b/tests/test_explanation_copy.py @@ -0,0 +1,77 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""End-to-end checks for fixes included since the previous published version.""" + +from unittest import IsolatedAsyncioTestCase, TestCase + +import casbin + + +def model(matcher="g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act"): + result = casbin.Model() + result.load_model_from_text(""" +[request_definition] +r = sub, obj, act +[policy_definition] +p = sub, obj, act +[role_definition] +g = _, _ +[policy_effect] +e = some(where (p.eft == allow)) +[matchers] +m = """ + matcher) + return result + + +class TestReleaseRegressions(TestCase): + def test_explanations_do_not_change_policy_or_cached_decisions(self): + for cls in ( + casbin.Enforcer, + casbin.CachedEnforcer, + casbin.SyncedEnforcer, + casbin.SyncedCachedEnforcer, + ): + with self.subTest(enforcer=cls.__name__): + enforcer = cls(model()) + enforcer.enable_auto_save(False) + enforcer.add_policy("admin", "data1", "read") + enforcer.add_grouping_policy("alice", "admin") + for _ in range(2): + allowed, explanation = enforcer.enforce_ex("alice", "data1", "read") + self.assertTrue(allowed) + self.assertEqual(explanation, ["admin", "data1", "read"]) + explanation[2] = "write" + self.assertEqual(enforcer.get_policy(), [["admin", "data1", "read"]]) + self.assertFalse(enforcer.enforce("alice", "data1", "write")) + + +class TestAsyncReleaseRegressions(IsolatedAsyncioTestCase): + async def test_explanations_do_not_change_policy_or_cached_decisions(self): + for cls in (casbin.AsyncEnforcer, casbin.AsyncCachedEnforcer): + with self.subTest(enforcer=cls.__name__): + enforcer = cls(model()) + enforcer.enable_auto_save(False) + await enforcer.add_policy("admin", "data1", "read") + await enforcer.add_grouping_policy("alice", "admin") + for _ in range(2): + allowed, explanation = enforcer.enforce_ex("alice", "data1", "read") + self.assertTrue(allowed) + self.assertEqual(explanation, ["admin", "data1", "read"]) + explanation[2] = "write" + self.assertEqual(enforcer.get_policy(), [["admin", "data1", "read"]]) + self.assertFalse(enforcer.enforce("alice", "data1", "write")) diff --git a/tests/test_release_regressions.py b/tests/test_release_regressions.py new file mode 100644 index 0000000..2e27502 --- /dev/null +++ b/tests/test_release_regressions.py @@ -0,0 +1,119 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""End-to-end checks for fixes included since the previous published version.""" + +from unittest import IsolatedAsyncioTestCase, TestCase + +import casbin +from casbin.persist import Adapter +from casbin.util import ip_match, key_match2, key_match3, key_match5 + + +def model(matcher="g(r.sub, p.sub) && r.obj == p.obj && r.act == p.act"): + result = casbin.Model() + result.load_model_from_text(""" +[request_definition] +r = sub, obj, act +[policy_definition] +p = sub, obj, act +[role_definition] +g = _, _ +[policy_effect] +e = some(where (p.eft == allow)) +[matchers] +m = """ + matcher) + return result + + +def check_permissions(test, enforcer, permissions): + test.assertEqual(permissions, [["admin", "data1", "read"]]) + permissions[0][2] = "write" + permissions.append(["alice", "data2", "write"]) + test.assertEqual(enforcer.get_policy(), [["admin", "data1", "read"]]) + test.assertTrue(enforcer.enforce("alice", "data1", "read")) + test.assertFalse(enforcer.enforce("alice", "data1", "write")) + + +class TestReleaseRegressions(TestCase): + def test_implicit_permission_rows_do_not_change_policy(self): + for cls in ( + casbin.Enforcer, + casbin.CachedEnforcer, + casbin.SyncedEnforcer, + casbin.SyncedCachedEnforcer, + ): + with self.subTest(enforcer=cls.__name__): + enforcer = cls(model(), Adapter()) + enforcer.add_policy("admin", "data1", "read") + enforcer.add_grouping_policy("alice", "admin") + check_permissions(self, enforcer, enforcer.get_implicit_permissions_for_user("alice")) + + def test_quoted_matcher_values_keep_their_spelling(self): + for quote in ('"', "'"): + for prefix in ("r.", "p.", "r2.", "p2."): + literal = "role " + prefix + "owner" + for cls in ( + casbin.Enforcer, + casbin.CachedEnforcer, + casbin.SyncedCachedEnforcer, + ): + with self.subTest(quote=quote, prefix=prefix, enforcer=cls.__name__): + enforcer = cls(model("r.sub == " + quote + literal + quote), Adapter()) + self.assertTrue(enforcer.enforce(literal, "data1", "read")) + self.assertFalse(enforcer.enforce(literal.replace(".", "_"), "data1", "read")) + + def test_synced_accessors_return_delegated_values(self): + class FilteredAdapter(Adapter): + def is_filtered(self): + return True + + adapter = FilteredAdapter() + enforcer = casbin.SyncedEnforcer(model(), adapter) + self.assertIs(enforcer.get_adapter(), adapter) + self.assertIs(enforcer.is_filtered(), True) + + def test_invalid_ip_stays_an_explicit_error(self): + # Applications must reject malformed request IPs; a parsing exception is not an allow. + with self.assertRaises(ValueError): + ip_match("invalid-address", "192.0.2.0/24") + self.assertFalse(ip_match("198.51.100.1", "192.0.2.0/24")) + self.assertTrue(ip_match("192.0.2.1", "192.0.2.0/24")) + + def test_plain_wildcard_paths_avoid_regex_backtracking(self): + from unittest.mock import patch + + # Check the fast path structurally rather than executing an expensive regex benchmark. + with patch( + "casbin.util.builtin_operators.regex_match", + side_effect=AssertionError("unexpected regex fallback"), + ): + for matcher in (key_match2, key_match3, key_match5): + self.assertTrue(matcher("/a/b/end", "/*/*/end")) + self.assertFalse(matcher("/a/b/other", "/*/*/end")) + + +class TestAsyncReleaseRegressions(IsolatedAsyncioTestCase): + async def test_implicit_permission_rows_do_not_change_policy(self): + for cls in (casbin.AsyncEnforcer, casbin.AsyncCachedEnforcer): + with self.subTest(enforcer=cls.__name__): + enforcer = cls(model()) + enforcer.enable_auto_save(False) + await enforcer.add_policy("admin", "data1", "read") + await enforcer.add_grouping_policy("alice", "admin") + permissions = await enforcer.get_implicit_permissions_for_user("alice") + check_permissions(self, enforcer, permissions) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
