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_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" 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()