From 3e0d1b89bf36c9c91b95b21697268ddc42eb3d16 Mon Sep 17 00:00:00 2001 From: Manish Kumar Date: Tue, 25 Aug 2026 22:45:35 -0500 Subject: [PATCH] test: expand policy engine security boundary coverage --- tests/test_policy_boundaries_extra.py | 175 ++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) create mode 100644 tests/test_policy_boundaries_extra.py diff --git a/tests/test_policy_boundaries_extra.py b/tests/test_policy_boundaries_extra.py new file mode 100644 index 0000000..a9d7cdf --- /dev/null +++ b/tests/test_policy_boundaries_extra.py @@ -0,0 +1,175 @@ +"""Additional deterministic boundary tests for the policy engine. + +These tests invoke pure policy functions and route/helper callables directly. +They intentionally avoid the environment's hanging FastAPI TestClient path. +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import MagicMock + +import pytest +from pydantic import ValidationError + +from app import main as main_module +from app.api import routes_policy +from app.core import abac, permissions, rules, tenancy +from app.main import custom_swagger_ui, health, swagger_static +from app.models.request import PolicyRequest +from app.services import policy_service + + +@pytest.mark.parametrize( + ("context", "roles", "allowed", "reason"), + [ + ({}, [], True, "abac passed"), + ({"gpu_required": True}, [], False, "GPU access denied"), + ({"gpu_required": True}, ["gpu_user"], True, "abac passed"), + ({"node": "hpc"}, [], False, "HPC access denied"), + ({"node": "hpc"}, ["hpc_user"], True, "abac passed"), + ({"gpu_required": True, "node": "hpc"}, ["gpu_user"], False, "HPC access denied"), + ], +) +def test_abac_boundary_combinations(context, roles, allowed, reason): + assert abac.evaluate_abac(context, roles) == (allowed, reason) + + +@pytest.mark.parametrize( + ("action", "resource", "expected"), + [ + ("tes.submit", "job", "workflow.execute"), + ("tes.submit.extra", "job", "workflow.execute"), + ("dataset.read", "dataset", "dataset.read"), + ("delete", "model_registry", "workflow.manage"), + ("read", "model_registry", None), + ], +) +def test_permission_resolution_covers_exact_prefix_resource_and_unknown_actions( + action, resource, expected +): + assert permissions.required_permission(action, resource) == expected + + +def test_permission_gate_admin_override_and_missing_permission_are_distinct(): + assert permissions.evaluate_permission(["admin"], [], "unknown", "resource") == ( + True, + "admin override", + ) + assert permissions.evaluate_permission([], ["dataset.read"], "delete", "model_registry") == ( + False, + "missing permission: workflow.manage", + ) + + +def test_permission_gate_unknown_action_is_currently_permissive(): + # This records the current fail-open behavior without disguising it as a + # secure contract; the pre-existing security-boundary test tracks the + # desired fail-closed behavior separately as an expected failure. + assert permissions.evaluate_permission([], ["unrelated"], "unknown", "resource") == ( + True, + "no permission required", + ) + + +@pytest.mark.parametrize( + ("org_id", "resource_org_id", "expected"), + [ + ("org-a", None, (True, "no tenancy scoping required")), + ("org-a", "org-a", (True, "tenancy check passed")), + ("org-a", "org-b", (False, "cross-tenant access denied")), + (None, "org-a", (False, "missing organization context")), + (7, "7", (True, "tenancy check passed")), + ], +) +def test_tenancy_boundary_values(org_id, resource_org_id, expected): + assert tenancy.evaluate_tenancy(org_id, {"resource_org_id": resource_org_id}) == expected + + +def test_rules_protect_both_human_genome_and_model_registry_resources(): + assert rules.evaluate_rules("dataset.delete", "human_genome_v1") == ( + False, + "protected dataset cannot be deleted", + ) + assert rules.evaluate_rules("delete", "model_registry") == ( + False, + "model registry is immutable", + ) + assert rules.evaluate_rules("dataset.read", "human_genome_v1") == (True, "rules passed") + + +def test_policy_request_defaults_are_independent_and_validate_required_fields(): + first = PolicyRequest(user_id="u1", action="read", resource="r") + second = PolicyRequest(user_id="u2", action="read", resource="r") + first.roles.append("viewer") + first.context["resource_org_id"] = "org-a" + + assert second.roles == [] + assert second.context == {} + with pytest.raises(ValidationError): + PolicyRequest(user_id="u1", action="read") + + +def test_policy_service_rejects_invalid_request_before_calling_engine(monkeypatch): + engine = MagicMock() + monkeypatch.setattr(policy_service, "engine", engine) + + with pytest.raises(ValidationError): + policy_service.evaluate_policy({"user_id": "u1", "action": "read"}) + engine.evaluate.assert_not_called() + + +def test_policy_route_delegates_without_testclient(monkeypatch): + evaluate = MagicMock(return_value={"allowed": True}) + monkeypatch.setattr(routes_policy, "evaluate_policy", evaluate) + payload = {"user_id": "u1", "action": "read", "resource": "dataset"} + + assert routes_policy.evaluate(payload) == {"allowed": True} + evaluate.assert_called_once_with(payload) + + +def test_health_and_swagger_helpers_are_deterministic(): + assert asyncio.run(health()) == {"status": "ok"} + missing = asyncio.run(swagger_static("does-not-exist.js")) + assert missing.status_code == 404 + css = asyncio.run(swagger_static("swagger-ui.css")) + assert css.status_code == 200 + + +def test_custom_swagger_ui_contains_openapi_metadata(): + response = asyncio.run(custom_swagger_ui()) + body = response.body.decode() + assert response.status_code == 200 + assert "OmniBioAI Policy Engine" in body + assert '"openapi"' in body + + +def test_invalidation_subscriber_restarts_after_pubsub_stream_closes(monkeypatch): + class PubSub: + def __init__(self): + self.listen_calls = 0 + + async def subscribe(self, channel): + assert channel == "policy:invalidate" + + async def listen(self): + self.listen_calls += 1 + if self.listen_calls == 1: + return + raise asyncio.CancelledError + yield # Keep this an async generator for the first empty stream. + + class Client: + def __init__(self): + self.pubsub_instance = PubSub() + + def pubsub(self): + return self.pubsub_instance + + client = Client() + monkeypatch.setattr(main_module.aioredis, "from_url", lambda *args, **kwargs: client) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(main_module._invalidation_subscriber()) + + assert client.pubsub_instance.listen_calls == 2