feat: make base wallet route access configurable

aries_cloudagent/admin/server.py

@@ -4,7 +4,7 @@
 from hmac import compare_digest
 import logging
 import re
-from typing import Callable, Coroutine
+from typing import Callable, Coroutine, Optional, Pattern, Sequence, cast
 import uuid
 import warnings
 import weakref
@@ -261,9 +261,32 @@ def __init__(
         self.websocket_queues = {}
         self.site = None
         self.multitenant_manager = context.inject_or(BaseMultitenantManager)
+        self._additional_route_pattern: Optional[Pattern] = None
 
         self.server_paths = []
 
+    @property
+    def additional_routes_pattern(self) -> Optional[Pattern]:
+        """Pattern for configured addtional routes to permit base wallet to access."""
+        if self._additional_route_pattern:
+            return self._additional_route_pattern
+
+        base_wallet_routes = self.context.settings.get("multitenant.base_wallet_routes")
+        base_wallet_routes = cast(Sequence[str], base_wallet_routes)
+        if base_wallet_routes:
+            self._additional_route_pattern = re.compile(
+                "^(?:" + "|".join(base_wallet_routes) + ")"
+            )
+        return None
+
+    def _matches_additional_routes(self, path: str) -> bool:
+        """Path matches additional_routes_pattern."""
+        pattern = self.additional_routes_pattern
+        if pattern:
+            return bool(pattern.match(path))
+
+        return False
+
     async def make_application(self) -> web.Application:
         """Get the aiohttp application instance."""
 
@@ -336,6 +359,7 @@ async def check_multitenant_authorization(request: web.Request, handler):
                         path,
                     )
                     or path.startswith("/mediation/default-mediator")
+                    or self._matches_additional_routes(path)
                 )
 
                 # base wallet is not allowed to perform ssi related actions.

aries_cloudagent/admin/tests/test_admin_server.py

@@ -195,7 +195,9 @@ async def test_import_routes(self):
 
     async def test_import_routes_multitenant_middleware(self):
         # imports all default admin routes
-        context = InjectionContext()
+        context = InjectionContext(
+            settings={"multitenant.base_wallet_routes": ["/test"]}
+        )
         context.injector.bind_instance(ProtocolRegistry, ProtocolRegistry())
         context.injector.bind_instance(GoalCodeRegistry, GoalCodeRegistry())
         context.injector.bind_instance(
@@ -249,6 +251,16 @@ async def test_import_routes_multitenant_middleware(self):
         await mt_authz_middle(mock_request, mock_handler)
         assert mock_handler.called_once_with(mock_request)
 
+        mock_request = async_mock.MagicMock(
+            method="GET",
+            headers={"Authorization": "Non-bearer ..."},
+            path="/test",
+            text=async_mock.CoroutineMock(return_value="abc123"),
+        )
+        mock_handler = async_mock.CoroutineMock()
+        await mt_authz_middle(mock_request, mock_handler)
+        assert mock_handler.called_once_with(mock_request)
+
         # multitenant setup context exception paths
         [setup_ctx_middle] = [m for m in app.middlewares if ".setup_context" in str(m)]
 

aries_cloudagent/config/argparse.py

@@ -1633,6 +1633,18 @@ def add_arguments(self, parser: ArgumentParser):
                 '"wallet_type" is "askar-profile"'
             ),
         )
+        parser.add_argument(
+            "--base-wallet-routes",
+            type=str,
+            nargs="+",
+            required=False,
+            metavar="<REGEX>",
+            help=(
+                "Patterns matching admin routes that should be permitted for "
+                "base wallet. The base wallet is preconfigured to have access to "
+                "essential endpoints. This argument should be used sparingly."
+            ),
+        )
 
     def get_settings(self, args: Namespace):
         """Extract multitenant settings."""
@@ -1683,6 +1695,9 @@ def get_settings(self, args: Namespace):
                         value = yaml.safe_load(value)
                         settings[f"multitenant.{key}"] = value
 
+            if args.base_wallet_routes:
+                settings["multitenant.base_wallet_routes"] = args.base_wallet_routes
+
         return settings
 
 

aries_cloudagent/config/tests/test_argparse.py

@@ -232,6 +232,8 @@ async def test_multitenancy_settings(self):
                 "secret",
                 "--multitenancy-config",
                 '{"wallet_type":"askar","wallet_name":"test", "cache_size": 10}',
+                "--base-wallet-routes",
+                "/my_route",
             ]
         )
 
@@ -241,6 +243,7 @@ async def test_multitenancy_settings(self):
         assert settings.get("multitenant.jwt_secret") == "secret"
         assert settings.get("multitenant.wallet_type") == "askar"
         assert settings.get("multitenant.wallet_name") == "test"
+        assert settings.get("multitenant.base_wallet_routes") == ["/my_route"]
 
         result = parser.parse_args(
             [
@@ -251,6 +254,8 @@ async def test_multitenancy_settings(self):
                 "wallet_type=askar",
                 "wallet_name=test",
                 "cache_size=10",
+                "--base-wallet-routes",
+                "/my_route",
             ]
         )
 
@@ -260,6 +265,7 @@ async def test_multitenancy_settings(self):
         assert settings.get("multitenant.jwt_secret") == "secret"
         assert settings.get("multitenant.wallet_type") == "askar"
         assert settings.get("multitenant.wallet_name") == "test"
+        assert settings.get("multitenant.base_wallet_routes") == ["/my_route"]
 
     async def test_endorser_settings(self):
         """Test required argument parsing."""