Anoncreds endorsement revision

Signed-off-by: jamshale <[email protected]>

aries_cloudagent/anoncreds/default/legacy_indy/registry.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import re
+import uuid
 from asyncio import shield
 from typing import List, Optional, Pattern, Sequence, Tuple
 
@@ -22,6 +23,7 @@
     GET_CRED_DEF,
     IndyLedgerRequestsExecutor,
 )
+from ....messaging.responder import BaseResponder
 from ....multitenant.base import BaseMultitenantManager
 from ....protocols.endorse_transaction.v1_0.manager import (
     TransactionManager,
@@ -196,7 +198,6 @@ async def register_schema(
         profile: Profile,
         schema: AnonCredsSchema,
         options: Optional[dict] = None,
-        outbound_handler: Optional[any] = None,
     ) -> SchemaResult:
         """Register a schema on the registry."""
 
@@ -261,12 +262,11 @@ async def register_schema(
             )
 
         # Need endorsement, so execute transaction flow
-        meta_data = {
-            "context": {"schema_id": schema_id, "schema": schema.serialize()},
-            "processing": {},
-        }
-
         (schema_id, schema_def) = result
+
+        job_id = uuid.uuid4().hex
+        meta_data = {"context": {"job_id": job_id, "schema_id": schema_id}}
+
         transaction_manager = TransactionManager(profile)
         try:
             transaction = await transaction_manager.create_record(
@@ -288,13 +288,23 @@ async def register_schema(
                     "Transaction manager failed to create request: " + err.roll_up
                 ) from err
 
-            await outbound_handler(
-                transaction_request, connection_id=endorser_connection_id
+            responder = profile.inject(BaseResponder)
+            await responder.send(
+                message=transaction_request,
+                connection_id=endorser_connection_id,
             )
-        return {
-            "sent": {"schema_id": schema_id, "schema": schema_def},
-            "txn": transaction.serialize(),
-        }
+
+        return SchemaResult(
+            job_id=job_id,
+            schema_state=SchemaState(
+                state=SchemaState.STATE_TRANSACTION_REQUESTED,
+                schema_id=schema_id,
+                schema=schema,
+            ),
+            registration_metadata={
+                "txn": transaction.serialize(),
+            },
+        )
 
     async def get_credential_definition(
         self, profile: Profile, cred_def_id: str

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

@@ -19,6 +19,7 @@
 from aries_cloudagent.connections.models.conn_record import ConnRecord
 from aries_cloudagent.core.in_memory.profile import InMemoryProfile
 from aries_cloudagent.ledger.error import LedgerError, LedgerObjectAlreadyExistsError
+from aries_cloudagent.messaging.responder import BaseResponder
 from aries_cloudagent.protocols.endorse_transaction.v1_0.manager import (
     TransactionManager,
 )
@@ -121,8 +122,7 @@ async def test_register_schema_with_author_role(self, mock_endorser_conn_record)
             self.profile, mock_schema, {"endorser_connection_id": "test_connection_id"}
         )
 
-        assert result["sent"] is not None
-        assert result["txn"] is not None
+        assert result.registration_metadata["txn"] is not None
         assert mock_endorser_conn_record.called
 
     @mock.patch.object(
@@ -183,8 +183,7 @@ async def test_register_schema_with_create_trasaction_param(
             },
         )
 
-        assert result["sent"] is not None
-        assert result["txn"] is not None
+        assert result.registration_metadata["txn"] is not None
         assert mock_endorser_conn_record.called
 
     @mock.patch.object(
@@ -228,20 +227,19 @@ def serialize(self):
 
         self.profile.settings.set_value("endorser.author", True)
         self.profile.settings.set_value("endorser.auto_request", True)
-
-        mock_outbound_handler = mock.CoroutineMock()
+        self.profile.context.injector.bind_instance(
+            BaseResponder, mock.MagicMock(send=mock.CoroutineMock(return_value=None))
+        )
 
         result = await self.registry.register_schema(
-            self.profile,
-            mock_schema,
-            {"endorser_connection_id": "test_connection_id"},
-            mock_outbound_handler,
+            self.profile, mock_schema, {"endorser_connection_id": "test_connection_id"}
         )
 
-        assert result["sent"] is not None
-        assert result["txn"] is not None
+        assert result.registration_metadata["txn"] is not None
         assert mock_create_request.called
-        assert mock_outbound_handler.called
+        assert self.profile.context.injector.get_provider(
+            BaseResponder
+        )._instance.send.called
 
     async def test_txn_submit(self):
         self.profile.inject = mock.MagicMock(

aries_cloudagent/anoncreds/events.py

@@ -3,22 +3,18 @@
 import re
 from typing import NamedTuple
 
-from aries_cloudagent.core.profile import Profile
-
 from ..core.event_bus import Event
 from .models.anoncreds_revocation import RevRegDef
 
 CRED_DEF_FINISHED_EVENT = "anoncreds::credential-definition::finished"
 REV_REG_DEF_FINISHED_EVENT = "anoncreds::revocation-registry-definition::finished"
 REV_LIST_FINISHED_EVENT = "anoncreds::revocation-list::finished"
-SCHEMA_EVENT = "acapy::schema-anoncreds::"
+SCHEMA_REGISTRATION_FINISHED_EVENT = "anoncreds::schema::registration::finished"
 
-SCHEMA_ANONCREDS_LISTENER_PATTERN = re.compile(f"^{SCHEMA_EVENT}(.*)?$")
-CRED_DEF_FINISHED_PATTERN = re.compile("anoncreds::credential-definition::finished")
-REV_REG_DEF_FINISHED_PATTERN = re.compile(
-    "anoncreds::revocation-registry-definition::finished"
-)
-REV_LIST_FINISHED_PATTERN = re.compile("anoncreds::revocation-list::finished")
+CRED_DEF_FINISHED_PATTERN = re.compile(CRED_DEF_FINISHED_EVENT)
+REV_REG_DEF_FINISHED_PATTERN = re.compile(REV_REG_DEF_FINISHED_EVENT)
+REV_LIST_FINISHED_PATTERN = re.compile(REV_LIST_FINISHED_EVENT)
+SCHEMA_REGISTRATION_FINISHED_PATTERN = re.compile(SCHEMA_REGISTRATION_FINISHED_EVENT)
 
 
 class CredDefFinishedPayload(NamedTuple):
@@ -133,11 +129,32 @@ def payload(self) -> RevListFinishedPayload:
         return self._payload
 
 
-async def notify_anoncreds_schema_event(
-    profile: Profile, schema_id: str, meta_data: dict
-):
-    """Send notification for a schema post-process event."""
-    await profile.notify(
-        SCHEMA_EVENT + schema_id,
-        meta_data,
-    )
+class SchemaRegistraionFinishedPayload(NamedTuple):
+    """Payload of schema transaction event."""
+
+    meta_data: dict
+
+
+class SchemaRegistrationFinishedEvent(Event):
+    """Event for schema post-process."""
+
+    def __init__(self, payload: SchemaRegistraionFinishedPayload):
+        """Initialize an instance.
+
+        Args:
+            schema_id: schema id
+            meta_data: meta data
+        """
+        self._topic = SCHEMA_REGISTRATION_FINISHED_EVENT
+        self._payload = payload
+
+    @classmethod
+    def with_payload(cls, meta_data: dict):
+        """With payload."""
+        payload = SchemaRegistraionFinishedPayload(meta_data)
+        return cls(payload)
+
+    @property
+    def payload(self) -> SchemaRegistraionFinishedPayload:
+        """Return payload."""
+        return self._payload

aries_cloudagent/anoncreds/issuer.py

@@ -135,15 +135,15 @@ async def store_schema(
         result: SchemaResult,
     ):
         """Store schema after reaching finished state."""
-        ident = result.schema_state.schema_id or result.job_id
-        if not ident:
+        identifier = result.job_id or result.schema_state.schema_id
+        if not identifier:
             raise ValueError("Schema id or job id must be set")
 
         try:
             async with self.profile.session() as session:
                 await session.handle.insert(
                     CATEGORY_SCHEMA,
-                    ident,
+                    identifier,
                     result.schema_state.schema.to_json(),
                     {
                         "name": result.schema_state.schema.name,
@@ -162,7 +162,6 @@ async def create_and_register_schema(
         version: str,
         attr_names: Sequence[str],
         options: Optional[dict] = None,
-        outbound_handler: Optional[any] = None,
     ) -> SchemaResult:
         """Create a new credential schema and store it in the wallet.
 
@@ -202,15 +201,9 @@ async def create_and_register_schema(
                 self.profile,
                 AnonCredsSchema.from_native(schema),
                 options,
-                outbound_handler,
             )
 
-            # Endorsement transaction - don't store in wallet
-            if not isinstance(schema_result, SchemaResult):
-                return schema_result
-
             await self.store_schema(schema_result)
-
             return schema_result
 
         except AnonCredsSchemaAlreadyExists as err:

aries_cloudagent/anoncreds/models/anoncreds_schema.py

@@ -6,13 +6,12 @@
 from marshmallow import EXCLUDE, fields
 from marshmallow.validate import OneOf
 
-from aries_cloudagent.messaging.valid import (
+from ...messaging.models.base import BaseModel, BaseModelSchema
+from ...messaging.valid import (
     INDY_OR_KEY_DID_EXAMPLE,
     INDY_SCHEMA_ID_EXAMPLE,
 )
 
-from ...messaging.models.base import BaseModel, BaseModelSchema
-
 
 class AnonCredsSchema(BaseModel):
     """An AnonCreds Schema object."""
@@ -144,6 +143,7 @@ class SchemaState(BaseModel):
     STATE_FAILED = "failed"
     STATE_ACTION = "action"
     STATE_WAIT = "wait"
+    STATE_TRANSACTION_REQUESTED = "transaction_requested"
 
     class Meta:
         """SchemaState metadata."""

aries_cloudagent/anoncreds/registry.py

@@ -91,13 +91,10 @@ async def register_schema(
         profile: Profile,
         schema: AnonCredsSchema,
         options: Optional[dict] = None,
-        outbound_handler: Optional[any] = None,
     ) -> SchemaResult:
         """Register a schema on the registry."""
         registrar = await self._registrar_for_identifier(schema.issuer_id)
-        return await registrar.register_schema(
-            profile, schema, options, outbound_handler
-        )
+        return await registrar.register_schema(profile, schema, options)
 
     async def get_credential_definition(
         self, profile: Profile, credential_definition_id: str

aries_cloudagent/anoncreds/routes.py

@@ -12,13 +12,11 @@
 )
 from marshmallow import fields
 
-from aries_cloudagent.anoncreds.events import SCHEMA_ANONCREDS_LISTENER_PATTERN
-from aries_cloudagent.core.profile import Profile
-from aries_cloudagent.ledger.error import LedgerError
-
 from ..admin.request_context import AdminRequestContext
 from ..askar.profile import AskarProfile
 from ..core.event_bus import Event, EventBus
+from ..core.profile import Profile
+from ..ledger.error import LedgerError
 from ..messaging.models.openapi import OpenAPISchema
 from ..messaging.valid import (
     INDY_CRED_DEF_ID_EXAMPLE,
@@ -42,16 +40,15 @@
     AnonCredsRegistrationError,
     AnonCredsResolutionError,
 )
+from .events import SCHEMA_REGISTRATION_FINISHED_PATTERN
 from .issuer import AnonCredsIssuer, AnonCredsIssuerError
 from .models.anoncreds_cred_def import CredDefResultSchema, GetCredDefResultSchema
 from .models.anoncreds_revocation import RevListResultSchema, RevRegDefResultSchema
 from .models.anoncreds_schema import (
-    AnonCredsSchema,
     AnonCredsSchemaSchema,
     GetSchemaResultSchema,
     SchemaResult,
     SchemaResultSchema,
-    SchemaState,
 )
 from .registry import AnonCredsRegistry
 from .revocation import AnonCredsRevocation, AnonCredsRevocationError
@@ -216,7 +213,6 @@ async def schemas_post(request: web.BaseRequest):
             version,
             attr_names,
             options,
-            request["outbound_message_router"],
         )
         if isinstance(result, SchemaResult):
             return web.json_response(result.serialize())
@@ -674,27 +670,14 @@ def register_events(event_bus: EventBus):
     # TODO Make this pluggable?
     setup_manager = DefaultRevocationSetup()
     setup_manager.register_events(event_bus)
-    event_bus.subscribe(SCHEMA_ANONCREDS_LISTENER_PATTERN, on_schema_event)
+    event_bus.subscribe(SCHEMA_REGISTRATION_FINISHED_PATTERN, on_schema_event)
 
 
 async def on_schema_event(profile: Profile, event: Event):
-    """Store the schema in wallet."""
-
-    await AnonCredsIssuer(profile).store_schema(
-        SchemaResult(
-            job_id=None,
-            schema_state=SchemaState(
-                state=SchemaState.STATE_FINISHED,
-                schema_id=event.payload["context"]["schema_id"],
-                schema=AnonCredsSchema(
-                    attr_names=event.payload["context"]["schema"]["attrNames"],
-                    name=event.payload["context"]["schema"]["name"],
-                    version=event.payload["context"]["schema"]["version"],
-                    issuer_id=event.payload["context"]["schema"]["issuerId"],
-                ),
-            ),
-            registration_metadata={},
-        )
+    """Schema post processing."""
+    await AnonCredsIssuer(profile).finish_schema(
+        event.payload.meta_data["context"]["job_id"],
+        event.payload.meta_data["context"]["schema_id"],
     )
 
 

aries_cloudagent/anoncreds/tests/test_issuer.py

@@ -372,7 +372,21 @@ async def test_create_and_register_schema_with_endorsed_transaction_response_doe
         mock_session_handle.insert = mock.CoroutineMock(return_value=None)
         self.profile.inject = mock.Mock(
             return_value=mock.MagicMock(
-                register_schema=mock.CoroutineMock(return_value={"txn": "transaction"})
+                register_schema=mock.CoroutineMock(
+                    return_value=SchemaResult(
+                        job_id="job-id",
+                        schema_state=SchemaState(
+                            state="finished",
+                            schema_id="schema-id",
+                            schema=AnonCredsSchema(
+                                issuer_id="issuer-id",
+                                name="schema-name",
+                                version="1.0",
+                                attr_names=["attr1", "attr2"],
+                            ),
+                        ),
+                    )
+                )
             )
         )
         result = await self.issuer.create_and_register_schema(
@@ -382,8 +396,8 @@ async def test_create_and_register_schema_with_endorsed_transaction_response_doe
             attr_names=["attr1", "attr2"],
         )
 
-        assert result is not None
-        assert not mock_store_schema.called
+        assert isinstance(result, SchemaResult)
+        assert mock_store_schema.called
 
     async def test_finish_schema(self):
         self.profile.transaction = mock.Mock(