From 6abf89b4260a3f94c5ab5348700f44483ea9c1a3 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 06:13:35 +0000 Subject: [PATCH 1/2] test: cover access-policy RBAC deny-path and circuit-breaker probe guard Close the three untested branches on two security/resilience primitives to 100% line coverage: - access_policy.evaluate_access: the final `rbac_denied` deny-path, reached when a non-admin owns (or is delegated) a resource but their role and group both fail the permit set. Ownership is necessary, not sufficient. - access_policy._equivalent_roles: the fallback that maps an unrecognised role to only itself. - CircuitBreaker.call: the half-open concurrency guard that rejects a second probe while the lone half-open probe is still in flight, so a provider that may still be down is not hit twice at once. Tests only; no production behavior changes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01T47gHdkeM8H2Mpu4VwZT3c --- backend/tests/test_access_policy.py | 37 +++++++++++++++++++++++++++ backend/tests/test_circuit_breaker.py | 31 ++++++++++++++++++++++ 2 files changed, 68 insertions(+) diff --git a/backend/tests/test_access_policy.py b/backend/tests/test_access_policy.py index 5a3d8dcfa..4628cd32d 100644 --- a/backend/tests/test_access_policy.py +++ b/backend/tests/test_access_policy.py @@ -3,6 +3,7 @@ from services.access_policy import ( AccessRequest, ResourcePolicy, + _equivalent_roles, _is_system_admin_role, evaluate_access, ) @@ -316,6 +317,42 @@ def test_tenant_admin_satisfies_organization_admin_alias_deterministically(): assert decision.reason == "allowed" +def test_owner_without_role_or_group_permission_is_rbac_denied(): + """Owning (or being delegated) a resource is necessary but not sufficient: + when every ABAC gate clears and ownership holds, a non-admin whose role and + group both fail the permit set is still denied (the final RBAC deny-path).""" + decision = evaluate_access( + AccessRequest( + user_id="alice", + role="member", + organization_id="org-acme", + group_ids=("sales",), + data_region="eu", + consent_scopes=("mail.read",), + ), + ResourcePolicy( + owner_id="alice", # owns the resource -> ownership gate clears + organization_id="org-acme", + permitted_roles=("tenant_admin",), # member does not satisfy tenant_admin + permitted_group_ids=("exec",), # no overlap with ("sales",) + data_region="eu", + required_consent_scopes=("mail.read",), + ), + ) + + assert decision.allowed is False + assert decision.reason == "rbac_denied" + + +def test_equivalent_roles_maps_known_alias_and_falls_back_for_unknown(): + """Known roles expand to their alias set; an unrecognised role maps only to + itself (the fallback branch).""" + assert _equivalent_roles("system_admin") == frozenset( + {"system_admin", "platform_admin"} + ) + assert _equivalent_roles("unknown_role") == frozenset({"unknown_role"}) + + def test_abac_owner_denial_overrides_group_admin_rbac_allow_without_delegation(): decision = evaluate_access( AccessRequest( diff --git a/backend/tests/test_circuit_breaker.py b/backend/tests/test_circuit_breaker.py index a9cc61066..daa50e261 100644 --- a/backend/tests/test_circuit_breaker.py +++ b/backend/tests/test_circuit_breaker.py @@ -1,5 +1,7 @@ """Tests for the provider circuit breaker.""" +import asyncio + import pytest from services.circuit_breaker import CircuitBreaker, CircuitOpenError @@ -87,6 +89,35 @@ async def test_half_open_probe_failure_reopens(): await breaker.call("p", _succeeding()) +@pytest.mark.asyncio +async def test_half_open_rejects_second_probe_while_first_in_flight(): + """Only one half-open probe is admitted at a time: a concurrent call that + arrives while the single probe is still awaiting fails fast rather than + piling a second request onto a provider that may still be down.""" + clock = _Clock() + breaker = CircuitBreaker(failure_threshold=1, cooldown_seconds=10, clock=clock) + + with pytest.raises(RuntimeError): + await breaker.call("p", _failing()) + clock.now = 11.0 # past cooldown -> the next call is the lone half-open probe + + release = asyncio.Event() + + async def slow_probe(): + await release.wait() + return "recovered" + + probe = asyncio.create_task(breaker.call("p", slow_probe)) + await asyncio.sleep(0) # let the probe start and claim the in-flight slot + + # A second call while the probe is in flight is rejected fast. + with pytest.raises(CircuitOpenError): + await breaker.call("p", _succeeding()) + + release.set() + assert await probe == "recovered" + + @pytest.mark.asyncio async def test_keys_are_isolated(): clock = _Clock() From dbb5d2f7d3308d98213a20c140afea704a05edd1 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 09:04:10 +0900 Subject: [PATCH 2/2] test: cover delegated RBAC deny path --- .../test_access_policy_delegated_rbac.py | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 backend/tests/test_access_policy_delegated_rbac.py diff --git a/backend/tests/test_access_policy_delegated_rbac.py b/backend/tests/test_access_policy_delegated_rbac.py new file mode 100644 index 000000000..5a458415e --- /dev/null +++ b/backend/tests/test_access_policy_delegated_rbac.py @@ -0,0 +1,27 @@ +from services.access_policy import AccessRequest, ResourcePolicy, evaluate_access + + +def test_delegated_user_without_role_or_group_permission_is_rbac_denied(): + """Delegation clears ownership, but does not bypass the final RBAC gate.""" + decision = evaluate_access( + AccessRequest( + user_id="delegate", + role="member", + organization_id="org-acme", + group_ids=("sales",), + data_region="eu", + consent_scopes=("mail.read",), + ), + ResourcePolicy( + owner_id="alice", + organization_id="org-acme", + permitted_roles=("tenant_admin",), + permitted_group_ids=("exec",), + data_region="eu", + required_consent_scopes=("mail.read",), + delegated_user_ids=("delegate",), + ), + ) + + assert decision.allowed is False + assert decision.reason == "rbac_denied"