test(security): close diff-coverage gaps in PR #4341 hybrid policy helpers

diff-cover surfaced five PR #4341 lines that no test exercised. All five are policy-critical paths in the new hybrid visibility helpers; without coverage, a regression that only broke one branch could ship undetected. Added 17 new tests across 4 files.

Coverage gaps closed:

  - server_service.py:1022-1044 (was 35.7% diff coverage). New TestServerAccessCheckMatrix in test_authorization_access.py with 7 tests mirroring the TestCheckToolAccess pattern in test_tool_service.py. Each test names the production line(s) it covers (no-user-email deny, public-only token deny, own-private allow, JWT team match, DB team lookup, fall-through deny).
  - gateway_service.py:2734,2739-2761 (was 32.1%). New TestGatewayAccessCheckMatrix with the same 7-test shape; gateway and server helpers are sibling implementations of the canonical hybrid policy.
  - base_service.py:216 (was 90%). New test_apply_visibility_scope_db_admin_includes_own_private_only — the existing test_apply_visibility_scope_admin_bypass_excludes_private only covered (None, None); the (email, None) DB-admin branch was unexecuted. Uses the same exact-OR-count predicate guard as the other DB-admin tests so a too-broad predicate fails.
  - a2a_service.py:1155 (was 91.7%). New test_get_agent_card_returns_none_when_visibility_denies covers the deny path of the cycle-2 S6-a in-service gate.
  - auth_context.py:215 (was 99.1%). The non-object payload guard in decode_internal_mcp_auth_context had a test, but the test was named 'testdecode_*' (missing underscore separator) so pytest's default 'test_*' collection pattern rejected it — the assertion never ran. This is the same bug class as cycle-1 B6. Renamed to test_decode_internal_mcp_auth_context_rejects_non_object_payload with a docstring naming the bug class.

Other testCASE-without-underscore names exist elsewhere (test_resource_service.py:2007/2024/2033, test_a2a_service.py:942, test_toolops_altk_service.py:59) but predate PR #4341 and are out of scope. Worth filing a separate cleanup issue.

Verified: 8443 tests pass across services/, utils/test_gateway_access.py, test_internal_a2a_endpoints.py, test_main.py, test_main_extended.py, test_admin.py (+17 from this commit; +1 of those is the previously-broken testdecode_* now actually running).
Signed-off-by: Jonathan Springer <jps@s390x.com>

Author: jonpspri

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "(?x)( package-lock\\.json$ |Cargo\\.lock$ |uv\\.lock$ |go\\.sum$ |mcpgateway/sri_hashes\\.json$ )|^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2026-04-26T08:27:41Z",
+  "generated_at": "2026-04-26T09:19:42Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"
@@ -7450,23 +7450,23 @@
         "hashed_secret": "d92342976d720ff38cf5dcb329be41959ab1ba6c",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 3268,
+        "line_number": 3292,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "a343c9b5b1abfcf385d5c6f6dc0282f4d88a416e",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 3429,
+        "line_number": 3453,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "c00dbbc9dadfbe1e232e93a729dd4752fade0abf",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 3461,
+        "line_number": 3485,
         "type": "Secret Keyword",
         "verified_result": null
       }
@@ -9262,15 +9262,15 @@
         "hashed_secret": "6745fa570d36be08400efa1cbc2f057bb001290e",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 2515,
+        "line_number": 2523,
         "type": "Hex High Entropy String",
         "verified_result": null
       },
       {
         "hashed_secret": "4f13f134744a2fadbbe2d624687246347d12fa63",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 2803,
+        "line_number": 2811,
         "type": "Hex High Entropy String",
         "verified_result": null
       }

tests/unit/mcpgateway/services/test_a2a_service.py

@@ -2680,6 +2680,30 @@ async def test_get_agent_card_returns_none_when_agent_missing(self, service, moc
 
         assert await service.get_agent_card(mock_db, "missing") is None
 
+    async def test_get_agent_card_returns_none_when_visibility_denies(self, service, mock_db):
+        """PR #4341 (S6-a) coverage: in-service gate returns None on visibility deny.
+
+        Exercises a2a_service.py:1155 — the deny path of the gate ``get_agent_card``
+        adopted in cycle 2. A private agent owned by a different user with the
+        anonymous-bypass shape must return None (not raise, not return the card).
+        """
+        agent = SimpleNamespace(
+            name="ag",
+            description="desc",
+            endpoint_url="https://x.com",
+            version=1,
+            protocol_version="1.0",
+            capabilities={},
+            visibility="private",
+            team_id=None,
+            owner_email="other@example.com",
+        )
+        mock_db.execute.return_value.scalar_one_or_none.return_value = agent
+
+        result = await service.get_agent_card(mock_db, "ag", user_email=None, token_teams=None)
+
+        assert result is None
+
     async def test_get_agent_card_builds_capabilities(self, service, mock_db):
         agent = SimpleNamespace(
             name="ag",

tests/unit/mcpgateway/services/test_authorization_access.py

@@ -14,6 +14,7 @@
 """
 
 # Standard
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 # Third-Party
@@ -1401,3 +1402,152 @@ async def test_tag_apply_visibility_scope_admin_bypass_excludes_private(self):
         assert "visibility" in compiled
         assert "private" in compiled
         assert "!=" in compiled or "<>" in compiled
+
+
+class TestServerAccessCheckMatrix:
+    """Branch coverage for ServerService._check_server_access (server_service.py:1015-1044).
+
+    Mirrors the existing TestCheckToolAccess matrix in test_tool_service.py so that
+    every documented policy outcome (public allow, anonymous bypass deny private,
+    public-only token deny, own-private allow, team membership via JWT, team
+    membership via DB lookup, final deny) has at least one focused regression.
+    """
+
+    @pytest.fixture
+    def service(self):
+        from mcpgateway.services.server_service import ServerService
+
+        return ServerService()
+
+    @staticmethod
+    def _server(visibility: str, owner_email=None, team_id=None) -> MagicMock:
+        s = MagicMock()
+        s.visibility = visibility
+        s.owner_email = owner_email
+        s.team_id = team_id
+        return s
+
+    @pytest.mark.asyncio
+    async def test_public_always_allowed(self, service):
+        """Public visibility short-circuits before any other check."""
+        s = self._server("public")
+        assert await service._check_server_access(MagicMock(), s, user_email=None, token_teams=[]) is True
+        assert await service._check_server_access(MagicMock(), s, user_email="any@test.com", token_teams=["team-x"]) is True
+
+    @pytest.mark.asyncio
+    async def test_anonymous_admin_bypass_denies_private(self, service):
+        """(None, None) anonymous admin bypass: team allowed, private denied."""
+        assert await service._check_server_access(MagicMock(), self._server("private", owner_email="other@test.com"), user_email=None, token_teams=None) is False
+        assert await service._check_server_access(MagicMock(), self._server("team", team_id="team-x"), user_email=None, token_teams=None) is True
+
+    @pytest.mark.asyncio
+    async def test_no_user_email_with_token_teams_denied(self, service):
+        """(None, [team]) shape: server_service.py line 1022-1023 — without user_email, denied."""
+        assert await service._check_server_access(MagicMock(), self._server("team", team_id="team-x"), user_email=None, token_teams=["team-x"]) is False
+
+    @pytest.mark.asyncio
+    async def test_public_only_token_denies_non_public(self, service):
+        """(email, []) public-only token: covers server_service.py line 1025-1027."""
+        s_team = self._server("team", team_id="team-x")
+        s_private = self._server("private", owner_email="user@test.com")
+        assert await service._check_server_access(MagicMock(), s_team, user_email="user@test.com", token_teams=[]) is False
+        assert await service._check_server_access(MagicMock(), s_private, user_email="user@test.com", token_teams=[]) is False
+
+    @pytest.mark.asyncio
+    async def test_own_private_allowed(self, service):
+        """Owner of a private server can read it (covers line 1029-1031)."""
+        own = self._server("private", owner_email="user@test.com")
+        assert await service._check_server_access(MagicMock(), own, user_email="user@test.com", token_teams=["team-x"]) is True
+
+    @pytest.mark.asyncio
+    async def test_team_member_via_jwt_token_teams(self, service):
+        """token_teams from JWT carries team membership (covers line 1033-1036)."""
+        s = self._server("team", team_id="team-x")
+        assert await service._check_server_access(MagicMock(), s, user_email="user@test.com", token_teams=["team-x"]) is True
+        # Non-matching team in JWT → denied (covers line 1044 fall-through).
+        assert await service._check_server_access(MagicMock(), s, user_email="user@test.com", token_teams=["team-y"]) is False
+
+    @pytest.mark.asyncio
+    async def test_team_member_via_db_lookup(self, service):
+        """token_teams=None with email + team server: falls back to TeamManagementService (covers line 1038-1042)."""
+        s = self._server("team", team_id="team-x")
+
+        with patch("mcpgateway.services.server_service.TeamManagementService") as mock_tms_cls:
+            mock_tms_cls.return_value.get_user_teams = AsyncMock(return_value=[SimpleNamespace(id="team-x")])
+            allowed = await service._check_server_access(MagicMock(), s, user_email="user@test.com", token_teams=None)
+
+        assert allowed is True
+        mock_tms_cls.return_value.get_user_teams.assert_awaited_once_with("user@test.com")
+
+
+class TestGatewayAccessCheckMatrix:
+    """Branch coverage for GatewayService._check_gateway_access (gateway_service.py:2732-2761).
+
+    Same shape as TestServerAccessCheckMatrix above — both helpers were added by
+    PR #4341 and both implement the canonical hybrid visibility policy.
+    """
+
+    @pytest.fixture
+    def service(self):
+        from mcpgateway.services.gateway_service import GatewayService
+
+        return GatewayService()
+
+    @staticmethod
+    def _gw(visibility: str, owner_email=None, team_id=None) -> MagicMock:
+        g = MagicMock()
+        g.visibility = visibility
+        g.owner_email = owner_email
+        g.team_id = team_id
+        return g
+
+    @pytest.mark.asyncio
+    async def test_public_always_allowed(self, service):
+        """Covers gateway_service.py line 2734."""
+        g = self._gw("public")
+        assert await service._check_gateway_access(MagicMock(), g, user_email=None, token_teams=[]) is True
+
+    @pytest.mark.asyncio
+    async def test_anonymous_admin_bypass_denies_private(self, service):
+        """(None, None) anonymous admin bypass: team allowed, private denied."""
+        assert await service._check_gateway_access(MagicMock(), self._gw("private", owner_email="other@test.com"), user_email=None, token_teams=None) is False
+        assert await service._check_gateway_access(MagicMock(), self._gw("team", team_id="team-x"), user_email=None, token_teams=None) is True
+
+    @pytest.mark.asyncio
+    async def test_no_user_email_with_token_teams_denied(self, service):
+        """(None, [team]) shape: covers gateway_service.py line 2739-2740."""
+        assert await service._check_gateway_access(MagicMock(), self._gw("team", team_id="team-x"), user_email=None, token_teams=["team-x"]) is False
+
+    @pytest.mark.asyncio
+    async def test_public_only_token_denies_non_public(self, service):
+        """(email, []) public-only token: covers line 2742-2744."""
+        g_team = self._gw("team", team_id="team-x")
+        g_private = self._gw("private", owner_email="user@test.com")
+        assert await service._check_gateway_access(MagicMock(), g_team, user_email="user@test.com", token_teams=[]) is False
+        assert await service._check_gateway_access(MagicMock(), g_private, user_email="user@test.com", token_teams=[]) is False
+
+    @pytest.mark.asyncio
+    async def test_own_private_allowed(self, service):
+        """Owner of a private gateway can read it (covers line 2746-2748)."""
+        own = self._gw("private", owner_email="user@test.com")
+        assert await service._check_gateway_access(MagicMock(), own, user_email="user@test.com", token_teams=["team-x"]) is True
+
+    @pytest.mark.asyncio
+    async def test_team_member_via_jwt_token_teams(self, service):
+        """token_teams from JWT carries team membership (covers line 2750-2753)."""
+        g = self._gw("team", team_id="team-x")
+        assert await service._check_gateway_access(MagicMock(), g, user_email="user@test.com", token_teams=["team-x"]) is True
+        # Non-matching team in JWT → denied (covers line 2761 fall-through).
+        assert await service._check_gateway_access(MagicMock(), g, user_email="user@test.com", token_teams=["team-y"]) is False
+
+    @pytest.mark.asyncio
+    async def test_team_member_via_db_lookup(self, service):
+        """token_teams=None with email + team gateway: falls back to TeamManagementService (covers line 2755-2759)."""
+        g = self._gw("team", team_id="team-x")
+
+        with patch("mcpgateway.services.gateway_service.TeamManagementService") as mock_tms_cls:
+            mock_tms_cls.return_value.get_user_teams = AsyncMock(return_value=[SimpleNamespace(id="team-x")])
+            allowed = await service._check_gateway_access(MagicMock(), g, user_email="user@test.com", token_teams=None)
+
+        assert allowed is True
+        mock_tms_cls.return_value.get_user_teams.assert_awaited_once_with("user@test.com")

tests/unit/mcpgateway/services/test_base_service.py

@@ -187,6 +187,45 @@ async def test_db_lookup_fallback_no_user_email(self, service, mock_db, query):
         assert result == "filtered"
         query.where.assert_called_once()
 
+    def test_apply_visibility_scope_db_admin_includes_own_private_only(self):
+        """PR #4341 carve-out coverage for the static ``_apply_visibility_scope`` helper.
+
+        ``_apply_visibility_scope`` is the sibling of ``_apply_access_control`` used by
+        completion/tag enumeration. Without this test the (email, None) DB-admin
+        branch in base_service.py:215-221 was unexecuted by the suite; the existing
+        ``test_apply_visibility_scope_admin_bypass_excludes_private`` coverage in
+        test_authorization_access.py only exercises the ``(None, None)`` shape.
+        """
+        from mcpgateway.services.base_service import BaseService
+        from mcpgateway.services.tag_service import TagService
+
+        service = TagService()
+        stmt = sa.select(_FakeItem)
+        mock_db = MagicMock()
+
+        with patch("mcpgateway.services.base_service.is_user_admin", return_value=True):
+            scoped = BaseService._apply_visibility_scope(
+                stmt,
+                _FakeItem,
+                user_email="dba@test.com",
+                token_teams=None,
+                team_ids=[],
+                db=mock_db,
+            )
+
+        compiled = _compile_where(scoped)
+        assert "visibility != 'private'" in compiled, f"public/team carve-out missing: {compiled}"
+        assert "visibility = 'private'" in compiled, f"own-private allowance missing: {compiled}"
+        assert "owner_email = 'dba@test.com'" in compiled, f"owner clause must bind caller: {compiled}"
+        # Same OR-count guard as test_db_admin_bypass_includes_own_private_only.
+        or_count = compiled.upper().count(" OR ")
+        assert or_count == 1, f"expected exactly 1 OR in WHERE clause, got {or_count}: {compiled}"
+        # Sanity: TagService._apply_visibility_scope is unused here (the static
+        # helper on BaseService is what we exercise) but importing it ensures the
+        # module-level binding exists so this test fails loudly if a refactor
+        # removes the indirection.
+        assert callable(getattr(service, "_apply_visibility_scope", None))
+
     @pytest.mark.asyncio
     async def test_db_admin_bypass_includes_own_private_only(self, service, mock_db):
         """PR #4341 carve-out: DB-admin (email, None) shape compiles a WHERE that allows own private but not others'.

tests/unit/mcpgateway/test_main_extended.py

@@ -1166,8 +1166,16 @@ def test_serialize_legacy_tool_payloads_preserves_dicts_and_unknowns(self):
 class TestInternalMcpHelperCoverage:
     """Target helper branches added for trusted Rust MCP forwarding."""
 
-    def testdecode_internal_mcp_auth_context_rejects_non_object_payload(self):
-        """Non-object JSON payloads should be rejected."""
+    def test_decode_internal_mcp_auth_context_rejects_non_object_payload(self):
+        """Non-object JSON payloads (lists, scalars) must raise ValueError.
+
+        The pre-fix name ``testdecode_*`` was missing the underscore separator
+        and silently failed pytest's default ``test_*`` collection pattern, so
+        the assertion below — guarding ``auth_context.decode_internal_mcp_auth_context``
+        line 215 — was never executed. A non-object payload smuggled past this
+        gate would let the helper return a list or scalar that the trust-header
+        plumbing assumes is a dict.
+        """
         header_value = base64.urlsafe_b64encode(orjson.dumps(["not-an-object"])).decode().rstrip("=")
         with pytest.raises(ValueError, match="must be an object"):
             decode_internal_mcp_auth_context(header_value)