hybrid_api.py

from flask import Flask, Blueprint, abort, request, url_for, jsonify, session
import requests
import hybridanalysis
from os.path import join
from hybridanalysis import HybridAnalysis
import config
from auth import admin_required

ha = Blueprint('ha', __name__, url_prefix='/ha')

#### HYBRIDANALYSIS API ####

def getHARoutes(ha):
    res = []
    endpoints = ha.getEndpoints()
    for base in endpoints:
        for idx in endpoints[base]['endpoints']:
            res.append(idx)
    return res

@ha.route('/', methods=["GET"])
def ha_main():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.getEndpoints()})

@ha.route('/search')
@ha.route('/search/', methods=['GET'])
def ha_search():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.Search.getEndpoints()})

@ha.route('/search/hash', methods=['GET', 'POST'])
def ha_search_hash():
    ha = config.HA['HA']
    if request.method == 'GET':
        args = request.args
        hash = args.get('hash', None)
    elif request.method == 'POST':
        print('HERE')
        hash = None

    try:
        res = ha.Search.hash(hash)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/search/hashes', methods=['GET', 'POST'])
def ha_search_hashes():
    ha = config.HA['HA']
    if request.method == 'GET':
        args = request.args
        hashes = []
        for arg in args:
            hashes.append(args.get(arg))
    elif request.method == 'POST':
        print('HERE')
        hashes = []
    
    try:
        res = ha.Search.hashes(hashes)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/search/terms', methods=['GET', 'POST'])
def ha_search_terms():
    ha = config.HA['HA']
    if request.method == 'GET':
        args = request.args
        filename = args.get('filename', None)
        filetype = args.get('filetype', None)
        filetype_desc = args.get('filetype_desc', None)
        env_id = args.get('env_id', None)
        country = args.get('country', None)
        verdict = args.get('verdict', None)
        av_detect = args.get('av_detect', None)
        vx_family = args.get('vx_family', None)
        tag = args.get('tag', None)
        port = args.get('port', None)
        host = args.get('host', None)
        domain = args.get('domain', None)
        url = args.get('url', None)
        similar_to = args.get('similar_to', None)
        context = args.get('context', None)
        imp_hash = args.get('imp_hash', None)
        ssdeep = args.get('ssdeep', None)
        authentihash = args.get('authentihash', None)
    elif request.method == 'POST':
        print('HERE')
        query = None
    
    try:
        res = ha.Search.terms(filename=filename, filetype=filetype, filetype_desc=filetype_desc, env_id=env_id, country=country, verdict=verdict, av_detect=av_detect, vx_family=vx_family, tag=tag, port=port, host=host, domain=domain, url=url, similar_to=similar_to, context=context, imp_hash=imp_hash, ssdeep=ssdeep, authentihash=authentihash)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/quick-scan')
@ha.route('/quick-scan/', methods=['GET'])
def ha_quick_scan():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.QuickScan.getEndpoints()})

@ha.route('/quick-scan/state', methods=['GET'])
def ha_quick_scan_state():
    ha = config.HA['HA']

    try:
        res = ha.QuickScan.state()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

# TODO

@ha.route('/quick-scan/url-for-analysis', methods=['GET', 'POST'])
def ha_quick_scan_url_for_analysis():
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        scan_type = args.get('scan_type', None)
        url = args.get('url', None)
        no_share_third_party = args.get('no_share_third_party', None)
        allow_community_access = args.get('allow_community_access', None)
        comment = args.get('comment', None)
        submit_name = args.get('submit_name', None)
    elif request.method == 'POST':
        print('HERE')
        url = None

    try:
        res = ha.QuickScan.url_for_analysis(scan_type, url, no_share_third_party=no_share_third_party, allow_community_access=allow_community_access, comment=comment, submit_name=submit_name)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/quick-scan/<id>', methods=['GET'])
def ha_quick_scan_id(id):
    ha = config.HA['HA']

    try:
        res = ha.QuickScan.id(id)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/quick-scan/<id>/convert-to-full', methods=['GET', 'POST'])
def ha_quick_scan_id_convert_to_full(id):
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        environment_id = args.get('environment_id', None)
        no_share_third_party = args.get('no_share_third_party', None)
        allow_community_access = args.get('allow_community_access', None)
        no_hash_lookup = args.get('no_hash_lookup', None)
        action_script = args.get('action_script', None)
        hybrid_analysis = args.get('hybrid_analysis', None)
        experimental_anti_evasion = args.get('experimental_anti_evasion', None)
        script_logging = args.get('script_logging', None)
        input_sample_tampering = args.get('input_sample_tampering', None)
        tor_enabled_analysis = args.get('tor_enabled_analysis', None)
        offline_analysis = args.get('offline_analysis', None)
        email = args.get('email', None)
        properties = args.get('properties', None)
        comment = args.get('comment', None)
        custom_date_time = args.get('custom_date_time', None)
        custom_cmd_line = args.get('custom_cmd_line', None)
        custom_run_time = args.get('custom_run_time', None)
        client = args.get('client', None)
        submit_name = args.get('submit_name', None)
        priority = args.get('priority', None)
        document_password = args.get('document_password', None)
        environment_variable = args.get('environment_variable', None)
    elif request.method == 'POST':
        print('HERE')
        environment_id = None

    try:
        res = ha.QuickScan.convert_to_full(id, environment_id, no_share_third_party=no_share_third_party, allow_community_access=allow_community_access, no_hash_lookup=no_hash_lookup, action_script=action_script, hybrid_analysis=hybrid_analysis, experimental_anti_evasion=experimental_anti_evasion, script_logging=script_logging, input_sample_tampering=input_sample_tampering, tor_enabled_analysis=tor_enabled_analysis, offline_analysis=offline_analysis, email=email, properties=properties, comment=comment, custom_date_time=custom_date_time, custom_cmd_line=custom_cmd_line, custom_run_time=custom_run_time, client=client, submit_name=submit_name, priority=priority, document_password=document_password, environment_variable=environment_variable)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/file-collection')
@ha.route('/file-collection/', methods=['GET'])
def ha_file_collection():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.FileCollection.getEndpoints()})

@ha.route('/file-collection/create', methods=['GET', 'POST'])
def ha_file_collection_create():
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        collection_name = args.get('collection_name', None)
        comment = args.get('comment', None)
        no_share_third_party = args.get('no_share_third_party', None)
        allow_community_access = args.get('allow_community_access', None)
    elif request.method == 'POST':
        print('HERE')
        url = None

    try:
        res = ha.FileCollection.create(collection_name=collection_name, comment=comment, no_share_third_party=no_share_third_party, allow_community_access=allow_community_access)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/file-collection/search', methods=['GET', 'POST'])
def ha_file_collection_search():
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        collection_name = args.get('collection_name', None)
        tag = args.get('tag', None)
    elif request.method == 'POST':
        print('HERE')
        url = None

    try:
        res = ha.FileCollection.search(collection_name=collection_name, tag=tag)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/file-collection/<id>', methods=['GET'])
def ha_file_collection_id(id):
    ha = config.HA['HA']

    try:
        res = ha.FileCollection.id(id)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

# TODO

@ha.route('/file-collection/<id>/files/download', methods=['GET', 'POST'])
def ha_file_collection_id_files_download(id):
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        if len(args) == 0:
            pass
        else:
            hashes = []
            for arg in args:
                hashes.append(args.get(arg))
    elif request.method == 'POST':
        print('HERE')
        url = None

    try:
        if len(args) == 0:
            res = ha.FileCollection.id_files_download_get(id)
        else:
            res = ha.FileCollection.id_files_download_post(id, hashes)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/file-collection/<id>/files/<hash>', methods=['DELETE'])
def ha_file_collection_id_files_download_delete(id, hash):
    ha = config.HA['HA']

    try:
        res = ha.FileCollection.id_files_download_delete(id, hash)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/file-collection/<id>', methods=['DELETE'])
def ha_file_collection_id_delete(id):
    ha = config.HA['HA']

    try:
        res = ha.FileCollection.delete(id)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/overview')
@ha.route('/overview/', methods=['GET'])
def ha_overview():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.Overview.getEndpoints()})

@ha.route('/overview/<sha256_hash>', methods=['GET'])
def ha_overview_hash(sha256_hash):
    ha = config.HA['HA']

    try:
        res = ha.Overview.hash(sha256_hash)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/overview/<sha256_hash>/refresh', methods=['GET'])
def ha_overview_hash_refresh(sha256_hash):
    ha = config.HA['HA']

    try:
        res = ha.Overview.hash_refresh(sha256_hash)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/overview/<sha256_hash>/summary', methods=['GET'])
def ha_overview_hash_summary(sha256_hash):
    ha = config.HA['HA']

    try:
        res = ha.Overview.hash_summary(sha256_hash)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/overview/<sha256_hash>/sample', methods=['GET'])
def ha_overview_hash_sample(sha256_hash):
    ha = config.HA['HA']

    try:
        res = ha.Overview.hash_sample(sha256_hash)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/submit')
@ha.route('/submit/', methods=['GET'])
def ha_submit():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.SandboxSubmission.getEndpoints()})

# TODO

@ha.route('/submit/url-for-analysis', methods=['GET', 'POST'])
def ha_submit_url_for_analysis():
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        url = args.get('url', None)
        environment_id = args.get('environment_id', None)
        no_share_third_party = args.get('no_share_third_party', None)
        allow_community_access = args.get('allow_community_access', None)
        no_hash_lookup = args.get('no_hash_lookup', None)
        action_script = args.get('action_script', None)
        hybrid_analysis = args.get('hybrid_analysis', None)
        experimental_anti_evasion = args.get('experimental_anti_evasion', None)
        script_logging = args.get('script_logging', None)
        input_sample_tampering = args.get('input_sample_tampering', None)
        tor_enabled_analysis = args.get('tor_enabled_analysis', None)
        offline_analysis = args.get('offline_analysis', None)
        email = args.get('email', None)
        properties = args.get('properties', None)
        comment = args.get('comment', None)
        custom_date_time = args.get('custom_date_time', None)
        custom_cmd_line = args.get('custom_cmd_line', None)
        custom_run_time = args.get('custom_run_time', None)
        client = args.get('client', None)
        submit_name = args.get('submit_name', None)
        priority = args.get('priority', None)
        document_password = args.get('document_password', None)
        environment_variable = args.get('environment_variable', None)
    elif request.method == 'POST':
        print('HERE')
        url = None

    try:
        res = ha.SandboxSubmission.url_for_analysis(url, environment_id, no_share_third_party=no_share_third_party, allow_community_access=allow_community_access, no_hash_lookup=no_hash_lookup, action_script=action_script, hybrid_analysis=hybrid_analysis, experimental_anti_evasion=experimental_anti_evasion, script_logging=script_logging, input_sample_tampering=input_sample_tampering, tor_enabled_analysis=tor_enabled_analysis, offline_analysis=offline_analysis, email=email, properties=properties, comment=comment, custom_date_time=custom_date_time, custom_cmd_line=custom_cmd_line, custom_run_time=custom_run_time, client=client, submit_name=submit_name, priority=priority, document_password=document_password, environment_variable=environment_variable)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/submit/hash-for-url', methods=['GET', 'POST'])
def ha_submit_hash_for_url():
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        url = args.get('url', None)
    elif request.method == 'POST':
        print('HERE')
        url = None

    try:
        res = ha.SandboxSubmission.hash_for_url(url)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/submit/dropped-file', methods=['GET', 'POST'])
def ha_submit_dropped_file():
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        id = args.get('id', None)
        file_hash = args.get('file_hash', None)
        no_share_third_party = args.get('no_share_third_party', None)
    elif request.method == 'POST':
        print('HERE')
        url = None

    try:
        res = ha.SandboxSubmission.dropped_file(id, file_hash, no_share_third_party=no_share_third_party)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/report')
@ha.route('/report/', methods=['GET'])
def ha_report():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.SandboxReport.getEndpoints()})

@ha.route('/report/<id>/state', methods=['GET'])
def ha_report_state(id):
    ha = config.HA['HA']

    try:
        res = ha.SandboxReport.id_state(id)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/report/<id>/summary', methods=['GET'])
def ha_report_summary(id):
    ha = config.HA['HA']

    try:
        res = ha.SandboxReport.id_summary(id)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

# FIX THIS
@ha.route('/report/summary', methods=['GET', 'POST'])
def ha_report_summary_multiple():
    ha = config.HA['HA']

    try:
        res = ha.SandboxReport.id_summary_multiple()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/report/<id>/file/<type>', methods=['GET'])
def ha_report_file_type(id, type):
    ha = config.HA['HA']

    try:
        res = ha.SandboxReport.id_file_type(id, type)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/report/<id>/screenshots', methods=['GET'])
def ha_report_screenshots(id):
    ha = config.HA['HA']

    try:
        res = ha.SandboxReport.id_screenshots(id)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/report/<id>/dropped-file-raw/<hash>', methods=['GET'])
def ha_report_dropped_file_raw(id, hash):
    ha = config.HA['HA']

    try:
        res = ha.SandboxReport.id_dropped_file_raw(id, hash)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/report/<id>/dropped-files', methods=['GET'])
def ha_report_dropped_files(id):
    ha = config.HA['HA']

    try:
        res = ha.SandboxReport.id_dropped_files(id)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system')
@ha.route('/system/', methods=['GET'])
def ha_system():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.System.getEndpoints()})

@ha.route('/system/version', methods=['GET'])
def ha_system_version():
    ha = config.HA['HA']

    try:
        res = ha.System.version()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/environments', methods=['GET'])
def ha_system_environments():
    ha = config.HA['HA']

    try:
        res = ha.System.environments()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/stats', methods=['GET'])
def ha_system_stats():
    ha = config.HA['HA']

    try:
        res = ha.System.stats()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/state', methods=['GET'])
def ha_system_state():
    ha = config.HA['HA']

    try:
        res = ha.System.state()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/configuration', methods=['GET'])
def ha_system_configuration():
    ha = config.HA['HA']

    try:
        res = ha.System.configuration()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/backend', methods=['GET'])
def ha_system_backend():
    ha = config.HA['HA']

    try:
        res = ha.System.backend()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/queue-size', methods=['GET'])
def ha_system_queue_size():
    ha = config.HA['HA']

    try:
        res = ha.System.queue_size()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/in-progress', methods=['GET'])
def ha_system_in_progress():
    ha = config.HA['HA']

    try:
        res = ha.System.in_progress()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/total-submissions', methods=['GET'])
def ha_system_total_submissions():
    ha = config.HA['HA']

    try:
        res = ha.System.total_submissions()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/system/heartbeat', methods=['GET'])
def ha_system_heartbeat():
    ha = config.HA['HA']

    try:
        res = ha.System.heartbeat()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/key')
@ha.route('/key/', methods=['GET'])
@admin_required
def ha_key():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.Key.getEndpoints()})

@ha.route('/key/current', methods=['GET'])
@admin_required
def ha_key_current():
    if session.get('is_admin', False):
        ha = config.HA['HA']

        try:
            res = ha.Key.current()
        except Exception as msg:
            return jsonify({'error': str(msg)}), 500

        return jsonify({'results': res})
    else:
        return jsonify({'error': 'invalid permissions'}), 500

@ha.route('/feed')
@ha.route('/feed/', methods=['GET'])
def ha_feed():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.Feed.getEndpoints()})

@ha.route('/feed/latest', methods=['GET'])
def ha_feed_latest():
    ha = config.HA['HA']

    try:
        res = ha.Feed.latest()
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})

@ha.route('/abuse-reports')
@ha.route('/abuse-reports/', methods=['GET'])
def ha_abusereports():
    ha = config.HA['HA']
    return jsonify({'endpoints': ha.AbuseReports.getEndpoints()})

@ha.route('/abuse-reports/new', methods=['GET', 'POST'])
def ha_abusereports_new():
    ha = config.HA['HA']

    if request.method == 'GET':
        args = request.args
        sha256 = args.get('sha256', None)
        reason = args.get('reason', None)
    elif request.method == 'POST':
        # TODO
        sha256 = None
        reason = None

    try:
        res = ha.AbuseReports.new(sha256, reason)
    except Exception as msg:
        return jsonify({'error': str(msg)}), 500

    return jsonify({'results': res})


########################

def searchHA(query, offset=None):
    ha = config.HA['HA']

    try:
        res = ha.Search.terms(filename=query)
    except hybridanalysis.HA_Error as msg:
        print(msg)
        return {'error': str(msg)}
    return {"data": res}