Source code for aries_cloudagent.ledger.routes

"""Ledger admin routes."""

from aiohttp import web
from aiohttp_apispec import docs, request_schema, response_schema

from marshmallow import fields, Schema

from .base import BaseLedger
from .error import LedgerTransactionError


[docs]class AMLRecordSchema(Schema): """Ledger AML record.""" version = fields.Str() aml = fields.Dict(fields.Str(), fields.Str()) amlContext = fields.Str()
[docs]class TAARecordSchema(Schema): """Ledger TAA record.""" version = fields.Str() text = fields.Str() digest = fields.Str()
[docs]class TAAAcceptanceSchema(Schema): """TAA acceptance record.""" mechanism = fields.Str() time = fields.Int()
[docs]class TAAInfoSchema(Schema): """Transaction author agreement info.""" aml_record = fields.Nested(AMLRecordSchema()) taa_record = fields.Nested(TAARecordSchema()) taa_required = fields.Bool() taa_accepted = fields.Nested(TAAAcceptanceSchema())
[docs]class TAAResultSchema(Schema): """Result schema for a transaction author agreement.""" result = fields.Nested(TAAInfoSchema())
[docs]class TAAAcceptSchema(Schema): """Input schema for accepting the TAA.""" version = fields.Str() text = fields.Str() mechanism = fields.Str()
[docs]@docs( tags=["ledger"], summary="Send a NYM registration to the ledger.", parameters=[ {"name": "did", "in": "query", "schema": {"type": "string"}, "required": True}, { "name": "verkey", "in": "query", "schema": {"type": "string"}, "required": True, }, { "name": "alias", "in": "query", "schema": {"type": "string"}, "required": False, }, { "name": "role", "in": "query", "schema": {"type": "string"}, "required": False, }, ], ) async def register_ledger_nym(request: web.BaseRequest): """ Request handler for registering a NYM with the ledger. Args: request: aiohttp request object """ context = request.app["request_context"] ledger = await context.inject(BaseLedger, required=False) if not ledger: raise web.HTTPForbidden() did = request.query.get("did") verkey = request.query.get("verkey") if not did or not verkey: raise web.HTTPBadRequest() alias, role = request.query.get("alias"), request.query.get("role") success = False async with ledger: try: await ledger.register_nym(did, verkey, alias, role) success = True except LedgerTransactionError as e: raise web.HTTPForbidden(text=e.message) return web.json_response({"success": success})
[docs]@docs( tags=["ledger"], summary="Get the verkey for a DID from the ledger.", parameters=[ {"name": "did", "in": "query", "schema": {"type": "string"}, "required": True} ], ) async def get_did_verkey(request: web.BaseRequest): """ Request handler for getting a verkey for a DID from the ledger. Args: request: aiohttp request object """ context = request.app["request_context"] ledger = await context.inject(BaseLedger, required=False) if not ledger: raise web.HTTPForbidden() did = request.query.get("did") if not did: raise web.HTTPBadRequest() async with ledger: r = await ledger.get_key_for_did(did) return web.json_response({"verkey": r})
[docs]@docs( tags=["ledger"], summary="Get the endpoint for a DID from the ledger.", parameters=[ {"name": "did", "in": "query", "schema": {"type": "string"}, "required": True} ], ) async def get_did_endpoint(request: web.BaseRequest): """ Request handler for getting a verkey for a DID from the ledger. Args: request: aiohttp request object """ context = request.app["request_context"] ledger = await context.inject(BaseLedger, required=False) if not ledger: raise web.HTTPForbidden() did = request.query.get("did") if not did: raise web.HTTPBadRequest() async with ledger: r = await ledger.get_endpoint_for_did(did) return web.json_response({"endpoint": r})
[docs]@docs(tags=["ledger"], summary="Fetch the current transaction author agreement, if any") @response_schema(TAAResultSchema, 200) async def ledger_get_taa(request: web.BaseRequest): """ Request handler for fetching the transaction author agreement. Args: request: aiohttp request object Returns: The TAA information including the AML """ context = request.app["request_context"] ledger: BaseLedger = await context.inject(BaseLedger, required=False) if not ledger or ledger.LEDGER_TYPE != "indy": raise web.HTTPForbidden() taa_info = await ledger.get_txn_author_agreement() accepted = None if taa_info["taa_required"]: accept_record = await ledger.get_latest_txn_author_acceptance() if accept_record: accepted = { "mechanism": accept_record["mechanism"], "time": accept_record["time"], } taa_info["taa_accepted"] = accepted return web.json_response({"result": taa_info})
[docs]@docs(tags=["ledger"], summary="Accept the transaction author agreement") @request_schema(TAAAcceptSchema) async def ledger_accept_taa(request: web.BaseRequest): """ Request handler for accepting the current transaction author agreement. Args: request: aiohttp request object Returns: The DID list response """ context = request.app["request_context"] ledger: BaseLedger = await context.inject(BaseLedger, required=False) if not ledger or ledger.LEDGER_TYPE != "indy": raise web.HTTPForbidden() accept_input = await request.json() taa_info = await ledger.get_txn_author_agreement() if not taa_info["taa_required"]: raise web.HTTPBadRequest() taa_record = { "version": accept_input["version"], "text": accept_input["text"], "digest": ledger.taa_digest(accept_input["version"], accept_input["text"]), } await ledger.accept_txn_author_agreement(taa_record, accept_input["mechanism"]) return web.json_response({})
[docs]async def register(app: web.Application): """Register routes.""" app.add_routes( [ web.post("/ledger/register-nym", register_ledger_nym), web.get("/ledger/did-verkey", get_did_verkey), web.get("/ledger/did-endpoint", get_did_endpoint), web.get("/ledger/taa", ledger_get_taa), web.post("/ledger/taa/accept", ledger_accept_taa), ] )