Add Credential Issuer tutorial

Adds a code sample and a code walkthrough explaining how to build a
service that issues Credentials (XLS-70) on the XRP Ledger.

Credential issuer: Clarify/revise documents field

Issue credentials code sample: fix bugs

Apply suggestions from @oeggert review

Co-authored-by: oeggert <117319296+oeggert@users.noreply.github.com>

Credential Issuer: more edits for clarity

_code-samples/issue-credentials/README.md

@@ -0,0 +1,5 @@
+# Credential Issuing Service
+
+This sample code shows how to issue credentials to XRPL users using a basic API service.
+
+For a full walkthrough of the code, see the tutorial: https://xrpl.org/docs/tutorials/python/build-apps/credential-issuing-service

_code-samples/issue-credentials/py/README.md

@@ -0,0 +1,15 @@
+# Credential Issuing Service - Python sample code
+
+This code implements an HTTP API that issues credentials on the XRPL on request.
+
+Quick install & usage:
+
+```sh
+python -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+flask --app issuer_service run
+```
+
+For more detail, see the full tutorial for [How to build a service that issues credentials on the XRP Ledger](https://xrpl.org/docs/tutorials/python/build-apps/credential-issuing-service).
+

_code-samples/issue-credentials/py/accept_credential.py

@@ -0,0 +1,61 @@
+#!/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)

_code-samples/issue-credentials/py/credential_model.py

@@ -0,0 +1,181 @@
+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

_code-samples/issue-credentials/py/decode_hex.py

@@ -0,0 +1,14 @@
+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

_code-samples/issue-credentials/py/issuer_service.py

@@ -0,0 +1,153 @@
+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.

_code-samples/issue-credentials/py/look_up_credentials.py

@@ -0,0 +1,57 @@
+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

_code-samples/issue-credentials/py/requirements.txt

@@ -0,0 +1,2 @@
+Flask==3.0.3
+xrpl-py==4.0.0