py-gva/libs/paylib.py

#!/usr/bin/env python3

import sys, re, os.path, json, ast
from termcolor import colored
from natools import fmt, sign, get_privkey
from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport

VERSION = "0.1.0"
PUBKEY_REGEX = "(?![OIl])[1-9A-Za-z]{42,45}"

class Transaction:

    def __init__(self, dunikey, node, recipient, amount, comment='', verbose=False):
        self.dunikey = dunikey
        self.recipient = recipient
        self.amount = amount
        self.comment = comment
        self.issuer = get_privkey(dunikey, "pubsec").pubkey
        self.verbose = verbose
        self.isChange = False

        if not re.match(PUBKEY_REGEX, recipient) or len(recipient) > 45:
            sys.stderr.write("La clé publique n'est pas au bon format.\n")
            sys.exit(1)

        if recipient == self.issuer:
            sys.stderr.write("Le destinataire ne peut pas être vous même.\n")
            sys.exit(1)

        # Define Duniter GVA node
        transport = AIOHTTPTransport(url=node)
        self.client = Client(transport=transport, fetch_schema_from_transport=True)

    def genDoc(self):
        # Build TX generation document
        queryBuild = gql(
            """
            query ($recipient: String!, $issuer: String!, $amount: Int!, $comment: String!, $isChange: Boolean!){ genTxs(
            amount: $amount
            comment: $comment
            issuer: $issuer
            recipient: $recipient
            useMempoolSources: $isChange
            )
        }
        """
        )
        paramsBuild = {
            "recipient": self.recipient,
            "issuer": self.issuer,
            "amount": self.amount,
            "comment": self.comment,
            "isChange": self.isChange
        }

        # Send TX document
        try:
            self.txDoc = str(self.client.execute(queryBuild, variable_values=paramsBuild))[13:-3].replace('\\n','\n')
            return self.txDoc
        except Exception as e:
            message = ast.literal_eval(str(e))["message"]
            sys.stderr.write("Echec de la génération du document:\n" + message + "\n")
            sys.exit(1)


    # Check document
    def checkTXDoc(self):
        issuerRaw=[];outAmount=[];outPubkey=[];commentRaw=[]
        self.splitDoc = self.txDoc.split("', '")
        for i, docs in enumerate(self.splitDoc):
            docList = docs.splitlines()
            for j, line in enumerate(docList):
                if re.search("Issuers:", line):
                    issuerRaw[i:] = [docList[(j + 1) % len(docList)]]
                if re.search("Outputs:", line):
                    outputRaw = docList[(j + 1) % len(docList)].split(":")
                    outAmount[i:] = [int(outputRaw[0])]
                    outPubkey[i:] = [outputRaw[2].split("SIG(")[1].replace(')','')]
                if re.search("Comment:", line):
                    commentRaw[i:] = [line.split(': ', 1)[1]]

        # Check if is only change transaction
        if all(i == self.issuer for i in outPubkey):
            print("Le document contient une transaction de change")
            self.isChange = True
        # Check validity of the document
        elif issuerRaw[0] != self.issuer or outAmount[0] != self.amount or outPubkey[0] != self.recipient or commentRaw[0] != self.comment:
            sys.stderr.write(colored("Le document généré est corrompu !\nNe fait plus confiance au noeud " + node + "\n", 'red'))
            sys.stderr.write(colored(issuerRaw[0] + " envoi " + str(outAmount[0]) + " vers " + outPubkey[0] + " with comment: " + commentRaw[0] + "\n", "yellow"))
            sys.exit(1)
        else:
            print("Le document généré est conforme.")
            self.isChange = False
            return self.txDoc

    def signDoc(self):
        # Sign TX documents
        signature=[]
        self.signedDoc=[]
        for i, docs in enumerate(self.splitDoc):
            signature.append(fmt["64"](sign(docs.encode(), get_privkey(self.dunikey, "pubsec"))[:-len(docs.encode())]))
            self.signedDoc.append(docs + signature[i].decode())
        return self.signedDoc


    def sendTXDoc(self):
        # Build TX documents
        txResult=[]
        for docs in self.signedDoc:
            querySign = gql(
                """
                mutation ($signedDoc: String!){ tx(
                rawTx: $signedDoc
                ) {
                    version
                    issuers
                    outputs
                }
            }
            """
            )
            paramsSign = {
                "signedDoc": docs
            }

            # Send TX Signed document
            try:
                txResult.append(str(self.client.execute(querySign, variable_values=paramsSign)))
            except Exception as e:
                message = ast.literal_eval(str(e))["message"]
                sys.stderr.write("Echec de la transaction:\n" + message + "\n")
                if self.verbose:
                    sys.stderr.write("Document final:\n" + docs)
                sys.exit(1)
            else:
                if self.isChange:
                    self.send()
                else:
                    print(colored("Transaction effectué avec succès !", "green"))
                    if self.verbose:
                        print(docs)
                    break

        return txResult

    def send(self):
        result = self.genDoc()
        result = self.checkTXDoc()
        result = self.signDoc()
        result = self.sendTXDoc()
        return result