Add anoncreds upgrade via api endpoint

aries_cloudagent/admin/server.py

@@ -18,7 +18,6 @@
     setup_aiohttp_apispec,
     validation_middleware,
 )
-
 from marshmallow import fields
 
 from ..config.injection_context import InjectionContext
@@ -31,7 +30,9 @@
 from ..messaging.responder import BaseResponder
 from ..messaging.valid import UUIDFour
 from ..multitenant.base import BaseMultitenantManager, MultitenantManagerError
+from ..storage.base import BaseStorage
 from ..storage.error import StorageNotFoundError
+from ..storage.type import RECORD_TYPE_ACAPY_UPGRADING
 from ..transport.outbound.message import OutboundMessage
 from ..transport.outbound.status import OutboundSendStatus
 from ..transport.queue.basic import BasicMessageQueue
@@ -205,6 +206,22 @@ async def ready_middleware(request: web.BaseRequest, handler: Coroutine):
     raise web.HTTPServiceUnavailable(reason="Shutdown in progress")
 
 
+@web.middleware
+async def upgrade_middleware(request: web.BaseRequest, handler: Coroutine):
+    """Blocking middleware for upgrades."""
+    context: AdminRequestContext = request["context"]
+
+    is_upgrading = []
+    async with context.profile.session() as session:
+        storage = session.inject(BaseStorage)
+        is_upgrading = await storage.find_all_records(RECORD_TYPE_ACAPY_UPGRADING)
+
+    if not is_upgrading:
+        return await handler(request)
+
+    raise web.HTTPServiceUnavailable(reason="Upgrade in progress")
+
+
 @web.middleware
 async def debug_middleware(request: web.BaseRequest, handler: Coroutine):
     """Show request detail in debug log."""
@@ -351,6 +368,8 @@ async def check_multitenant_authorization(request: web.Request, handler):
 
                 is_multitenancy_path = path.startswith("/multitenancy")
                 is_server_path = path in self.server_paths or path == "/features"
+                # allow base wallets to trigger update through api
+                is_upgrade_path = path.startswith("/anoncreds/wallet/upgrade")
 
                 # subwallets are not allowed to access multitenancy routes
                 if authorization_header and is_multitenancy_path:
@@ -380,6 +399,7 @@ async def check_multitenant_authorization(request: web.Request, handler):
                     and not is_unprotected_path(path)
                     and not base_limited_access_path
                     and not (request.method == "OPTIONS")  # CORS fix
+                    and not is_upgrade_path
                 ):
                     raise web.HTTPUnauthorized()
 
@@ -453,6 +473,9 @@ async def setup_context(request: web.Request, handler):
 
         middlewares.append(setup_context)
 
+        # Upgrade middleware needs the context setup
+        middlewares.append(upgrade_middleware)
+
         # Register validation_middleware last avoiding unauthorized validations
         middlewares.append(validation_middleware)
 

aries_cloudagent/admin/tests/test_admin_server.py

@@ -1,22 +1,26 @@
 import gc
 import json
+from unittest import IsolatedAsyncioTestCase
 
 import pytest
-from aries_cloudagent.tests import mock
-from unittest import IsolatedAsyncioTestCase
 from aiohttp import ClientSession, DummyCookieJar, TCPConnector, web
 from aiohttp.test_utils import unused_port
 
+from aries_cloudagent.tests import mock
+
 from ...config.default_context import DefaultContextBuilder
 from ...config.injection_context import InjectionContext
 from ...core.event_bus import Event
+from ...core.goal_code_registry import GoalCodeRegistry
 from ...core.in_memory import InMemoryProfile
 from ...core.protocol_registry import ProtocolRegistry
-from ...core.goal_code_registry import GoalCodeRegistry
+from ...storage.base import BaseStorage
+from ...storage.record import StorageRecord
+from ...storage.type import RECORD_TYPE_ACAPY_UPGRADING
 from ...utils.stats import Collector
 from ...utils.task_queue import TaskQueue
-
 from .. import server as test_module
+from ..request_context import AdminRequestContext
 from ..server import AdminServer, AdminSetupError
 
 
@@ -477,6 +481,36 @@ async def test_server_health_state(self):
             assert response.status == 503
         await server.stop()
 
+    async def test_upgrade_middleware(self):
+        profile = InMemoryProfile.test_profile()
+        self.context = AdminRequestContext.test_context({}, profile)
+        self.request_dict = {
+            "context": self.context,
+        }
+        request = mock.MagicMock(
+            method="GET",
+            path_qs="/schemas/created",
+            match_info={},
+            __getitem__=lambda _, k: self.request_dict[k],
+        )
+        handler = mock.CoroutineMock()
+
+        await test_module.upgrade_middleware(request, handler)
+
+        async with profile.session() as session:
+            storage = session.inject(BaseStorage)
+            upgrading_record = StorageRecord(
+                RECORD_TYPE_ACAPY_UPGRADING,
+                "true",
+            )
+            await storage.add_record(upgrading_record)
+
+            with self.assertRaises(test_module.web.HTTPServiceUnavailable):
+                await test_module.upgrade_middleware(request, handler)
+
+            await storage.delete_record(upgrading_record)
+            await test_module.upgrade_middleware(request, handler)
+
 
 @pytest.fixture
 async def server():

aries_cloudagent/anoncreds/default/legacy_indy/registry.py

@@ -1123,18 +1123,14 @@ async def fix_ledger_entry(
 
     async def txn_submit(
         self,
-        profile: Profile,
+        ledger: BaseLedger,
         ledger_transaction: str,
         sign: bool = None,
         taa_accept: bool = None,
         sign_did: DIDInfo = sentinel,
         write_ledger: bool = True,
     ) -> str:
         """Submit a transaction to the ledger."""
-        ledger = profile.inject(BaseLedger)
-
-        if not ledger:
-            raise LedgerError("No ledger available")
 
         try:
             async with ledger:

aries_cloudagent/anoncreds/default/legacy_indy/tests/test_registry.py

@@ -9,7 +9,6 @@
 from base58 import alphabet
 
 from .....anoncreds.base import (
-    AnonCredsRegistrationError,
     AnonCredsSchemaAlreadyExists,
 )
 from .....anoncreds.models.anoncreds_schema import (
@@ -21,7 +20,7 @@
 from .....connections.models.conn_record import ConnRecord
 from .....core.in_memory.profile import InMemoryProfile
 from .....ledger.base import BaseLedger
-from .....ledger.error import LedgerError, LedgerObjectAlreadyExistsError
+from .....ledger.error import LedgerObjectAlreadyExistsError
 from .....messaging.responder import BaseResponder
 from .....protocols.endorse_transaction.v1_0.manager import (
     TransactionManager,
@@ -728,27 +727,16 @@ async def test_register_revocation_registry_definition_with_create_transaction_a
         assert mock_create_record.called
 
     async def test_txn_submit(self):
-        self.profile.inject = mock.MagicMock(
-            side_effect=[
-                None,
-                mock.CoroutineMock(
-                    txn_submit=mock.CoroutineMock(side_effect=LedgerError("test error"))
-                ),
-                mock.CoroutineMock(
-                    txn_submit=mock.CoroutineMock(return_value="transaction response")
-                ),
-            ]
+        self.profile.context.injector.bind_instance(
+            BaseLedger,
+            mock.MagicMock(
+                txn_submit=mock.CoroutineMock(return_value="transaction_id")
+            ),
         )
-
-        # No ledger
-        with self.assertRaises(LedgerError):
-            await self.registry.txn_submit(self.profile, "test_txn")
-        # Write error
-        with self.assertRaises(AnonCredsRegistrationError):
-            await self.registry.txn_submit(self.profile, "test_txn")
-
-        result = await self.registry.txn_submit(self.profile, "test_txn")
-        assert result == "transaction response"
+        async with self.profile.session() as session:
+            ledger = session.inject(BaseLedger)
+            result = await self.registry.txn_submit(ledger, "test_txn")
+            assert result == "transaction_id"
 
     async def test_register_revocation_list_no_endorsement(self):
         self.profile.context.injector.bind_instance(

aries_cloudagent/core/conductor.py

@@ -7,6 +7,7 @@
 
 """
 
+import asyncio
 import hashlib
 import json
 import logging
@@ -17,11 +18,7 @@
 
 from ..admin.base_server import BaseAdminServer
 from ..admin.server import AdminResponder, AdminServer
-from ..commands.upgrade import (
-    add_version_record,
-    get_upgrade_version_list,
-    upgrade,
-)
+from ..commands.upgrade import add_version_record, get_upgrade_version_list, upgrade
 from ..config.default_context import ContextBuilder
 from ..config.injection_context import InjectionContext
 from ..config.ledger import (
@@ -71,10 +68,12 @@
 from ..transport.outbound.message import OutboundMessage
 from ..transport.outbound.status import OutboundSendStatus
 from ..transport.wire_format import BaseWireFormat
+from ..utils.profiles import get_subwallet_profiles_from_storage
 from ..utils.stats import Collector
 from ..utils.task_queue import CompletedTask, TaskQueue
 from ..vc.ld_proofs.document_loader import DocumentLoader
 from ..version import RECORD_TYPE_ACAPY_VERSION, __version__
+from ..wallet.anoncreds_upgrade import upgrade_wallet_to_anoncreds
 from ..wallet.did_info import DIDInfo
 from .dispatcher import Dispatcher
 from .error import StartupError
@@ -522,7 +521,9 @@ async def start(self) -> None:
             except Exception:
                 LOGGER.exception("Error accepting mediation invitation")
 
-        # notify protocols of startup status
+        await self.check_for_wallet_upgrades_in_progress()
+
+        # notify protcols of startup status
         await self.root_profile.notify(STARTUP_EVENT_TOPIC, {})
 
     async def stop(self, timeout=1.0):
@@ -823,3 +824,20 @@ async def check_for_valid_wallet_type(self, profile):
                 raise StartupError(
                     f"Wallet type config [{storage_type_from_config}] doesn't match with the wallet type in storage [{storage_type_record.value}]"  # noqa: E501
                 )
+
+    async def check_for_wallet_upgrades_in_progress(self):
+        """Check for upgrade and upgrade if needed."""
+        multitenant_mgr = self.context.inject_or(BaseMultitenantManager)
+        if multitenant_mgr:
+            subwallet_profiles = await get_subwallet_profiles_from_storage(
+                self.root_profile
+            )
+            await asyncio.gather(
+                *[
+                    upgrade_wallet_to_anoncreds(profile, True)
+                    for profile in subwallet_profiles
+                ]
+            )
+
+        else:
+            await upgrade_wallet_to_anoncreds(self.root_profile)