test.py

import hashlib
import os
import sys
import time
import requests

# wait time when polling for results with data id
TIME_DELAY = 5

# api key for authorization
API_KEY_META_DEFENDER_CLOUD = "INSERT API KEY HERE"

# give program status stdout: options 1 = print all updates. 0 = print only import updates
VERBOSE = 0


# given a filepath, generate a hash
def computeHash(file_name):
    try:
        # declare hash type- options- sha1, md5, sha256
        file_hash = hashlib.sha1()

        # open file in read byte mode
        with open(file_name, 'rb') as file:
            data = file.read(file_hash.block_size)
            file_hash.update(data)

        # convert hash to expected format
        return file_hash.hexdigest().upper()

    except IOError as e:
        print("io error: ", e)
        return None


# check if hash is already registered in database
def validateHash(file_hash):
    # api path
    url = 'https://api.metadefender.com/v4/hash/' + str(file_hash)

    try:
        # run http get method
        response = requests.get(
            url,
            headers={
                'apikey': API_KEY_META_DEFENDER_CLOUD
            }
        )

        # raise any errors from http method
        response.raise_for_status()

        # convert to json for processing
        response_json = response.json()

        # return any data about the hash
        return response_json

    # url with hash likely doesnt exist- 404
    except requests.exceptions.HTTPError:
        # if url is not found- no hash endpoint on server. this is OK
        if response.status_code == 404:
            return None

        # any other http errors
        print("http error")
        sys.exit(-5)
    except requests.exceptions.ConnectionError:
        print("bad connection")
        sys.exit(-4)
    except requests.exceptions.Timeout:
        print("timeout")
        sys.exit(-3)
    except requests.exceptions.TooManyRedirects:
        print("too many redirects")
        sys.exit(-2)
    except requests.exceptions.RequestException as e:
        print("unexpected error: ", e)
        sys.exit(-1)


# send file to OPSWAT for processing
def uploadFile(filename):
    # api path
    url = 'https://api.metadefender.com/v4/file'

    try:
        # run http post method
        response = requests.post(
            url,
            headers={
                'content-type': 'application/octet-stream',  # for binary stream
                'apikey': API_KEY_META_DEFENDER_CLOUD
            },
            files={
                'file': open(filename, 'rb')  # mode: read and binary
            }
        )

        # raise any errors from http method
        response.raise_for_status()

        # convert to json for processing
        response_json = response.json()

        if VERBOSE:
            print("file hash: ", response_json["sha1"])

        return response_json["data_id"]

    # file could not get to OPSWAT
    except requests.exceptions.HTTPError:
        print("http error")
        sys.exit(-5)
    except requests.exceptions.ConnectionError:
        print("bad connection")
        sys.exit(-4)
    except requests.exceptions.Timeout:
        print("timeout")
        sys.exit(-3)
    except requests.exceptions.TooManyRedirects:
        print("too many redirects")
        sys.exit(-2)
    except requests.exceptions.RequestException as e:
        print("fatal error: ", e)
        sys.exit(-1)


# get scan results with data_id
def getScanResults(id):
    # api path
    url = 'https://api.metadefender.com/v4/file/' + str(id)

    # keep track of status polling
    counter = 1

    # loop until good response or error
    while True:
        try:
            if VERBOSE:
                print("Polling attempt :: ", counter)

            # run http get method
            response = requests.get(
                url,
                headers={
                    'apikey': API_KEY_META_DEFENDER_CLOUD
                }
            )

            # raise any errors from http method
            response.raise_for_status()

            # convert to json for processing
            response_json = response.json()

            # get server processing status
            scan_status = response_json["scan_results"]["scan_all_result_a"]

            # if scan complete, return results
            if scan_status == "No Threat Detected" or scan_status == "Infected":
                return response_json

        # something happened with the request
        except requests.exceptions.HTTPError:
            print("http error")
            sys.exit(-5)
        except requests.exceptions.ConnectionError:
            print("bad connection")
            sys.exit(-4)
        except requests.exceptions.Timeout:
            print("timeout")
            sys.exit(-3)
        except requests.exceptions.TooManyRedirects:
            print("too many redirects")
            sys.exit(-2)
        except requests.exceptions.RequestException as e:
            print("fatal error when getting results: ", e)
            sys.exit(-1)

        # wait before next polling request is made
        counter += 1
        time.sleep(TIME_DELAY)


# display results in required format
def displayResults(result):
    print("filename: ", result["file_info"]["display_name"])
    print("overall_status:", result["scan_results"]["scan_all_result_a"])

    for scan in result["scan_results"]["scan_details"]:
        print("engine: ", scan)
        print("threat_found: ", result["scan_results"]["scan_details"][scan]["threat_found"])
        print("scan_result: ", result["scan_results"]["scan_details"][scan]["scan_result_i"])
        print("def_time: ", result["scan_results"]["scan_details"][scan]["def_time"])

    print("END")


# main program thread
def start(arg):
    # make sure filename program input is valid
    if os.path.isfile(arg):

        # calculate hash
        h = computeHash(arg)

        if VERBOSE:
            print("Hash: ", h)

        if h is not None:
            # call API
            hash_result = validateHash(h)

            # if server has cached results about hash
            if hash_result is not None:
                if VERBOSE:
                    print("hash found")

                # display file data
                displayResults(hash_result)

            # if server does not have cached results about hash
            else:
                if VERBOSE:
                    print("hash not found. uploading file")

                # send file to server and get data_id to get results later
                data_id = uploadFile(arg)

                if VERBOSE:
                    print("data id: ", data_id)

                # continue polling server with data id until analysis is complete or error
                result = getScanResults(data_id)

                # display file data
                displayResults(result)


# program entry
if __name__ == "__main__":
    start(sys.argv[1])