Build a Credential Issuing Service

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

flask --app issuer_service run

Issuer account seed: 
Starting credential issuer with XRPL address rJ6XzCCSapCaWZxExArkcBWLgJvT6bXCbV
 * Serving Flask app 'issuer_service'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on http://127.0.0.1:5000
Press CTRL+C to quit

{
  "credentials": [
    {
      "accepted": false,
      "credential": "TestCredential",
      "subject": "rGtnKx7veDhV9CgYenkiCV5HMLpgU2BfcQ"
    }
  ]
}

python accept_credential.py

Accept a credential?
    0) No, quit.
    1) 'TestCredential' issued by rJ6XzCCSapCaWZxExArkcBWLgJvT6bXCbV
    2) 'AnotherTestCredential' issued by rJ6XzCCSapCaWZxExArkcBWLgJvT6bXCbV
Select an option (0-2): 1

#!/usr/bin/env python

from binascii import unhexlify
from os import getenv
from getpass import getpass

from xrpl.clients import JsonRpcClient
from xrpl.models.requests import AccountObjects, AccountObjectType
from xrpl.models.transactions import CredentialAccept
from xrpl.transaction import submit_and_wait
from xrpl.utils import str_to_hex, datetime_to_ripple_time
from xrpl.wallet import Wallet, generate_faucet_wallet

from look_up_credentials import look_up_credentials
from decode_hex import decode_hex

XRPL_SERVER = "https://s.devnet.rippletest.net:51234/"

client = JsonRpcClient(XRPL_SERVER)

def init_wallet():
    seed = getenv("SUBJECT_ACCOUNT_SEED")
    if not seed:
        seed = getpass(prompt='Subject account seed: ',stream=None)
    if not seed:
        print("Please specify the subject's master seed")
        exit(1)
    return Wallet.from_seed(seed=seed)

wallet = init_wallet()

pending_credentials = look_up_credentials(
        client, 
        subject=wallet.address, 
        accepted="no"
)

prompt = """
Accept a credential?
    0) No, quit."""
for i, cred in enumerate(pending_credentials):
    credential_type_s = decode_hex(cred["CredentialType"])
    prompt += f"\n    {i+1}) '{credential_type_s}' issued by {cred['Issuer']}"

selection = None
options = [str(n) for n in range(len(pending_credentials)+1)]
while selection not in options:
    selection = input(prompt+f"\n Select an option (0-{len(options)-1}): ")

if selection == "0":
    exit(0)

chosen_cred = pending_credentials[int(selection)-1]
tx = CredentialAccept(
    account=wallet.address, 
    credential_type=chosen_cred["CredentialType"],
    issuer=chosen_cred["Issuer"]
)
print("Submitting transaction", tx)
response = submit_and_wait(tx, client=client, wallet=wallet, autofill=True)
print(response)

from os import getenv
from getpass import getpass

from flask import Flask, jsonify, request

from xrpl.clients import JsonRpcClient
from xrpl.models.exceptions import XRPLModelException
from xrpl.models.requests import LedgerEntry
from xrpl.models.transactions import CredentialCreate, CredentialDelete
from xrpl.transaction import sign_and_submit
from xrpl.wallet import Wallet

from look_up_credentials import look_up_credentials, XRPLLookupError
from credential_model import Credential, CredentialRequest

# Set up XRPL connection ------------------------------------------------------
def init_wallet():
    seed = getenv("ISSUER_ACCOUNT_SEED")
    if not seed:
        seed = getpass(prompt='Issuer account seed: ',stream=None)
    if not seed:
        print("Please specify the issuer's master seed")
        exit(1)
    return Wallet.from_seed(seed=seed)

wallet = init_wallet()
print("Starting credential issuer with XRPL address", wallet.address)

client = JsonRpcClient("https://s.devnet.rippletest.net:51234/")

# Define Flask app ------------------------------------------------------------
app = Flask(__name__)

# Method for users to request a credential from the service -------------------
@app.route("/credential", methods=['POST'])
def request_credential():
    # CredentialRequest throws if the request is not validly formatted
    cred_request = CredentialRequest(request.json)
    # verify_documents() throws if the provided documents don't pass inspection
    cred_request.verify_documents()
    cred_xrpl = cred_request.to_xrpl()

    cc_response = sign_and_submit(CredentialCreate(
        account=wallet.address,
        subject=cred_xrpl.subject,
        credential_type=cred_xrpl.credential,
        uri=cred_xrpl.uri,
        expiration=cred_xrpl.expiration
    ), client=client, wallet=wallet, autofill=True)

    if cc_response.status != "success":
        raise XRPLTxError(cc_response)
    elif cc_response.result["engine_result"] == "tecDUPLICATE":
        raise XRPLTxError(cc_response, status_code=409)
    elif cc_response.result["engine_result"] != "tesSUCCESS":
        raise XRPLTxError(cc_response)

    response = jsonify(cc_response.result)
    response.status_code = 201
    return response

# Method for admins to look up all credentials issued -------------------------
@app.route("/admin/credential")
def get_credentials():
    # ?accepted=yes|no|both query parameter - the default is "both"
    filter_accepted = request.args.get("accepted", "both").lower()

    credentials = look_up_credentials(
            client,
            issuer=wallet.address, 
            accepted=filter_accepted
    )
    response = {
        "credentials": [Credential.from_xrpl(c).to_dict() for c in credentials]
    }
    return response

# Method for admins to revoke an issued credential ----------------------------
@app.route("/admin/credential", methods=['DELETE'])
def delete_credential():
    del_request = Credential(request.json)

    # To save on transaction fees, check if the Credential
    # exists on ledger before attempting to delete it.
    xrpl_response = client.request(LedgerEntry(credential={
        "subject": del_request.subject,
        "issuer": wallet.address,
        "credential_type": del_request.to_xrpl().credential
    }))
    if (xrpl_response.status != "success" and 
            xrpl_response.result["error"] == "entryNotFound"):
        response = jsonify({
            "error": "entryNotFound",
            "error_message": (f"Credential doesn't exist for subject "
                              f"'{del_request.subject} and credential type "
                              f"'{del_request.credential}'")
        })
        response.status_code = 404
        return response

    cd_response = sign_and_submit(CredentialDelete(
        account=wallet.address,
        subject=del_request.subject,
        credential_type=del_request.to_xrpl().credential
    ), client=client, wallet=wallet, autofill=True)

    if cd_response.status != "success":
        raise XRPLTxError(cd_response)

    if cd_response.result["engine_result"] == "tecNO_ENTRY":
        # Usually this won't happen since we just checked for the credential,
        # but it's possible it got deleted since then.
        raise XRPLTxError(cd_response, status_code=404)
    elif cd_response.result["engine_result"] != "tesSUCCESS":
        raise XRPLTxError(cd_response)

    response = jsonify(cd_response.result)
    response.status_code = 200
    return response

# Error handling --------------------------------------------------------------
class XRPLTxError(Exception):
    def __init__(self, xrpl_response, status_code=400):
        self.body = xrpl_response.result
        self.status_code = status_code

@app.errorhandler(XRPLTxError)
def handle_tx_error(e):
    response = jsonify(e.body)
    response.status_code = e.status_code
    return response

@app.errorhandler(XRPLLookupError)
def handle_xrpl_error(e):
    response = jsonify(e.body)
    response.status_code = 400
    return response

@app.errorhandler(ValueError)
def handle_value_error(e):
    response = jsonify({
        "error": "badRequest",
        "error_message": str(e)
    })
    response.status_code = 400
    return response

# Reuse the same handler for xrpl-py's model exceptions
app.register_error_handler(XRPLModelException, handle_value_error)

# Tip: Some of Flask's built-in errors return HTML, not JSON, by default. 
# If you want to configure those, you can import error cases like BadRequest 
# from werkzeug.exceptions and implement custom handlers.

from xrpl.clients import JsonRpcClient
from xrpl.models.requests import AccountObjects, AccountObjectType

lsfAccepted = 0x00010000

class XRPLLookupError(Exception):
    def __init__(self, xrpl_response):
        self.body = xrpl_response.result

def look_up_credentials(client:JsonRpcClient, 
                        issuer:str="", 
                        subject:str="", 
                        accepted:str="both"):
    """
    Looks up Credentials issued by/to a specified XRPL account, optionally
    filtering by accepted status. Handles pagination.
    """
    account = issuer or subject # Use whichever is specified, issuer if both
    if not account:
        raise ValueError("Must specify issuer or subject")
    
    accepted = accepted.lower()
    if accepted not in ("yes","no","both"):
        raise ValueError("accepted must be str 'yes', 'no', or 'both'")

    credentials = []
    has_more_pages = True
    marker = None
    while has_more_pages:
        xrpl_response = client.request(AccountObjects(
            account=account,
            type=AccountObjectType.CREDENTIAL,
            marker=marker
        ))
        if xrpl_response.status != "success":
            raise XRPLLookupError(xrpl_response)

        for obj in xrpl_response.result["account_objects"]:
            # Skip credentials that aren't issued to/by the requested address.
            if issuer and obj["Issuer"] != issuer:
                continue
            if subject and obj["Subject"] != subject:
                continue
            # Skip credentials that don't match the specified accepted status
            cred_accepted = obj["Flags"] & lsfAccepted
            
            if accepted == "yes" and not cred_accepted:
                continue
            if accepted == "no" and cred_accepted:
                continue
            credentials.append(obj)
        
        marker = xrpl_response.result.get("marker")
        if not marker:
            has_more_pages = False
    
    return credentials

from binascii import unhexlify

def decode_hex(s_hex):
    """
    Try decoding a hex string as ASCII; return the decoded string on success,
    or the un-decoded string prefixed by '(BIN) ' on failure.
    """
    try:
        s = unhexlify(s_hex).decode("ascii")
        # Could use utf-8 instead, but it has more edge cases.
        # Optionally, sanitize the string for display before returning
    except:
        s = "(BIN) "+s_hex
    return s

import re
from datetime import datetime

from xrpl.core.addresscodec import is_valid_classic_address
from xrpl.utils import ripple_time_to_datetime, datetime_to_ripple_time, str_to_hex

from decode_hex import decode_hex

def is_allowed_credential_type(credential_type: str):
    """
    Returns True if the specified credential type is one that this service
    issues, or False otherwise.
    
    XRPL credential types can be any binary data; this service issues
    any credential that can be encoded from the following ASCII chars:
    alphanumeric characters, underscore, period, and dash.
    (min length 1, max 64)

    You might want to further limit the credential types, depending on your 
    use case; for example, you might only issue one specific credential type.
    """
    CREDENTIAL_REGEX = re.compile(r'^[A-Za-z0-9_\.\-]{1,64}$')
    if CREDENTIAL_REGEX.match(credential_type):
        return True
    return False

def is_allowed_uri(uri):
    """
    Returns True if the specified URI is acceptable for this service, or
    False otherwise.

    XRPL Credentials' URI values can be any binary data; this service
    adds any user-requested URI to a Credential as long as the URI
    can be encoded from the characters usually allowed in URIs, namely
    the following ASCII chars:

    alphanumeric characters (upper and lower case)
    the following symbols: -._~:/?#[]@!$&'()*+,;=%
    (minimum length 1 and max length 256 chars)

    You might want to instead define your own URI and attach it to the
    Credential regardless of user input, or you might want to verify that the
    URI points to a valid Verifiable Credential document that matches the user.
    """
    URI_REGEX = re.compile(r"^[A-Za-z0-9\-\._~:/\?#\[\]@!$&'\(\)\*\+,;=%]{1,256}$")
    if URI_REGEX.match(uri):
        return True
    return False

class Credential:
    """
    A credential object, in a simplified format for our API.
    The constructor performs parameter validation. Attributes:
    subject (str): the subject of the credential, as a classic address
    credential (str): the credential type, in human-readable (ASCII) chars
    uri (str, optional): URI of the credential in human-readable (ASCII) chars
    expiration (datetime, optional): time when the credential expires
                            (displayed as an ISO 8601 format string in JSON)
    accepted (bool, optional): true if this credential has been accepted
                               on the XRPL by the subject account.
                               False if not accepted. 
                               Omitted for credentials that haven't been 
                               issued yet.
    """
    

    def __init__(self, d: dict):
        self.subject = d.get("subject")
        if type(self.subject) != str:
            raise ValueError("Must provide a string 'subject' field")
        if not is_valid_classic_address(self.subject):
            raise ValueError(f"subject not valid address: '{self.subject}'")

        self.credential = d.get("credential")
        if type(self.credential) != str:
            raise ValueError("Must provide a string 'credential' field")

        if not is_allowed_credential_type(self.credential):
            raise ValueError(f"credential not allowed: '{self.credential}'.")

        self.uri = d.get("uri")
        if self.uri is not None and (
                type(self.uri) != str or not is_allowed_uri(self.uri)):
            raise ValueError(f"URI isn't valid: {self.uri}")

        exp = d.get("expiration")
        if exp:
            if type(exp) == str:
                self.expiration = datetime.fromisoformat(exp)
            elif type(exp) == datetime:
                self.expiration = exp
            else:
                raise ValueError(f"Unsupported expiration format: {type(exp)}")
        else:
            self.expiration = None
        
        self.accepted = d.get("accepted")
    
    

    @classmethod
    def from_xrpl(cls, xrpl_d: dict):
        """
        Instantiate from a Credential ledger entry in the XRPL format.
        """
        d = {
            "subject": xrpl_d["Subject"],
            "credential": decode_hex(xrpl_d["CredentialType"]),
            "accepted": bool(xrpl_d["Flags"] & 0x00010000) # lsfAccepted
        }
        if xrpl_d.get("URI"):
            d["uri"] = decode_hex(xrpl_d["URI"])
        if xrpl_d.get("Expiration"):
            d["expiration"] = ripple_time_to_datetime(xrpl_d["Expiration"])
        return cls(d)

    

    def to_dict(self):
        d = {
            "subject": self.subject,
            "credential": self.credential,
        }
        if self.expiration is not None:
            d["expiration"] = self.expiration.isoformat()
        if self.uri:
            d["uri"] = self.uri
        if self.accepted is not None:
            d["accepted"] = self.accepted
        return d
    
    

    def to_xrpl(self):
        """
        Return an object with parameters formatted for the XRPL
        """
        return XrplCredential(self)

class XrplCredential:
    """
    A Credential object, in a format closer to the XRP Ledger representation.
    Credential type and URI are hexadecimal;
    Expiration, if present, is in seconds since the Ripple Epoch.
    """
    def __init__(self, c:Credential):
        self.subject = c.subject
        self.credential = str_to_hex(c.credential)
        if c.expiration:
            self.expiration = datetime_to_ripple_time(c.expiration)
        else:
            self.expiration = None

        if c.uri:
            self.uri = str_to_hex(c.uri)
        else:
            self.uri = None

class CredentialRequest(Credential):
    """
    Request from user to issue a credential on ledger.
    The constructor performs parameter validation.
    """
    def __init__(self, cred_request):
        super().__init__(cred_request)
        # As a credential issuer, you typically need to verify some information
        # about someone before you issue them a credential. For this example,
        # the user passes relevant information in a documents field of the API
        # request. The documents are kept confidential, off-chain.
        self.documents = cred_request.get("documents")
    
    def verify_documents(self):
        # This is where you would check the user's documents to see if you
        # should issue the requested Credential to them.
        # Depending on the type of credentials your service needs, you might
        # need to implement different types of checks here.
        if not self.documents:
            raise ValueError(f"you must provide a non-empty 'documents' field")
        
        # As a placeholder, this example checks that the documents field
        # contains a string field named "reason" containing the word "please"
        if type(self.documents.get("reason")) != str:
            raise ValueError(f"documents must contain a 'reason' string")
        if "please" not in self.documents["reason"].lower():
            raise ValueError(f"reason must include 'please'")
        
        return True