app.py

# This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
# See the file 'LICENSE' for copying permission.

import json
import logging

# system imports
import os
import secrets
import shutil

# web imports
from flask import Flask
from flask_executor import Executor
from flask_executor.futures import Future
from flask_shell2http import Shell2HTTP
from werkzeug.utils import safe_join

# get flask-shell2http logger instance
logger = logging.getLogger("flask_shell2http")
# logger config
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
log_level = os.getenv("LOG_LEVEL", logging.INFO)
log_path = os.getenv("LOG_PATH", "/var/log/intel_owl/malware_tools_analyzers")
# create new file handlers, files are created if doesn't already exists
fh = logging.FileHandler(f"{log_path}/malware_tools_analyzers.log")
fh.setFormatter(formatter)
fh.setLevel(log_level)
fh_err = logging.FileHandler(f"{log_path}/malware_tools_analyzers_errors.log")
fh_err.setFormatter(formatter)
fh_err.setLevel(logging.ERROR)
# add the handlers to the logger
logger.addHandler(fh)
logger.addHandler(fh_err)
logger.setLevel(log_level)

# Globals
app = Flask(__name__)
app.config["SECRET_KEY"] = secrets.token_hex(16)
executor = Executor(app)
shell2http = Shell2HTTP(app, executor)

# we are changeing the directory for execution of
#  artifacts script as it requires us to be in the
#  same directory
os.chdir("/opt/deploy/artifacts/artifacts")


# Box-JS report
def read_files_and_make_report(dir_loc):
    report = {}
    files_to_read = [
        "IOC.json",
        "snippets.json",
        "resources.json",
        "analysis.log",
        "urls.json",
        "active_urls.json",
    ]
    # Read output from files one by one
    for fname in files_to_read:
        try:
            with open(safe_join(dir_loc, fname), encoding="utf-8") as fp:
                try:
                    report[fname] = json.load(fp)
                except json.JSONDecodeError:
                    report[fname] = fp.readlines()
        except FileNotFoundError:
            report[fname] = f"FileNotFoundError: {fname}"

    return report


def intercept_box_js_result(context, future: Future) -> None:
    """
    Box-JS doesn't output result to standard output but to a file,
    using this callback function,
    we intercept the future object and update its result attribute
    by reading the final analysis result from the saved result file
    before it is ready to be consumed.
    """
    # get current result
    res = future.result()
    fname = context.get("read_result_from", "")
    dir_loc = safe_join("/tmp/boxjs", fname + ".results")
    if not fname:
        if res.get("returncode", -1) == 0:
            res["returncode"] = -1
        raise Exception("No file specified to read result from")
    try:
        res["report"] = read_files_and_make_report(dir_loc)
    except Exception as e:
        res["error"] += str(e)
    finally:
        # set final result
        future._result = res  # skipcq PYL-W0212
        # Remove the directory
        shutil.rmtree(dir_loc, ignore_errors=True)


def intercept_droidlysis_result(context, future: Future) -> None:
    logger.info("Intercepting droidlysis result")
    res = future.result()

    dir_loc = "/opt/deploy/droidlysis/out/"
    subdir = os.listdir(dir_loc)[0]
    f_loc = safe_join(dir_loc, subdir, "report.json")

    logger.info(f"Reading result from {f_loc}")
    if not os.path.exists(f_loc):
        res["error"] += f", result file {f_loc} does not exists."
        if res.get("returncode", -1) == 0:
            res["returncode"] = -1
    else:
        with open(f_loc, "r", encoding="utf-8") as fp:
            try:
                res["report"] = json.load(fp)
            except json.JSONDecodeError:
                res["report"] = fp.read()
    # 4. set final result after modifications
    future._result = res  # skipcq PYL-W0212

    # 5. directory can be removed now
    if dir_loc:
        shutil.rmtree(dir_loc, ignore_errors=True)


def intercept_thug_result(context, future: Future) -> None:
    """
    Thug doesn't output result to standard output but to a file,
    using this callback function,
    we intercept the future object and update its result attribute
    by reading the final analysis result from the saved result file
    before it is ready to be consumed.
    """
    # 1. get current result object
    res = future.result()
    # 2. dir from which we will read final analysis result
    dir_loc = context.get("read_result_from", None)
    if not dir_loc:
        res["error"] += ", No specified file to read result from"
        if res.get("returncode", -1) == 0:
            res["returncode"] = -1
    else:
        # 3. read saved result file, if it exists
        f_loc = dir_loc + "/analysis/json/analysis.json"
        if not os.path.exists(f_loc):
            res["error"] += f", result file {f_loc} does not exists."
            if res.get("returncode", -1) == 0:
                res["returncode"] = -1
        else:
            with open(f_loc, "r", encoding="utf-8") as fp:
                try:
                    res["report"] = json.load(fp)
                except json.JSONDecodeError:
                    res["report"] = fp.read()

    # 4. set final result after modifications
    future._result = res  # skipcq PYL-W0212

    # 5. directory can be removed now
    if dir_loc:
        shutil.rmtree(dir_loc, ignore_errors=True)


# with this, we can make http calls to the endpoint: /capa
shell2http.register_command(endpoint="capa", command_name="/usr/local/bin/capa -q -j")

# with this, we can make http calls to the endpoint: /floss
shell2http.register_command(
    endpoint="floss",
    command_name="/usr/local/bin/floss -q -j",
)

# with this, we can make http calls to the endpoint: /peframe
shell2http.register_command(
    endpoint="peframe", command_name="/opt/deploy/peframe/venv/bin/peframe"
)

# with this, we can make http calls to the endpoint: /stringsifter
shell2http.register_command(
    endpoint="stringsifter",
    command_name="/opt/deploy/stringsifter/wrapper.py",
)

# with this, we can make http calls to the endpoint: /clamav
shell2http.register_command(
    endpoint="clamav", command_name="/usr/bin/clamdscan --allmatch --infected"
)

# with this, we can make http calls to the endpoint: /boxjs
shell2http.register_command(
    endpoint="boxjs",
    command_name="/usr/local/bin/box-js",
    callback_fn=intercept_box_js_result,
)

# with this, we can make http calls to the endpoint: /apkid
shell2http.register_command(
    endpoint="apkid", command_name="/opt/deploy/apkid/venv/bin/apkid"
)

shell2http.register_command(
    endpoint="qiling",
    command_name="/opt/deploy/qiling/venv/bin/python3 /opt/deploy/qiling/analyze.py",
)

# diec is the command for Detect It Easy
shell2http.register_command(
    endpoint="die",
    command_name="/opt/deploy/die/die_lin64_portable/base/diec",
)

# mobsfscan is the command for DroidLysis
shell2http.register_command(
    endpoint="mobsf",
    command_name="/opt/deploy/mobsf/venv/bin/mobsfscan",
)

# droidlysis is the command for DroidLysis
shell2http.register_command(
    endpoint="droidlysis",
    command_name="/opt/deploy/droidlysis/venv/bin/droidlysis",
    callback_fn=intercept_droidlysis_result,
)

# flake8: noqa
shell2http.register_command(
    endpoint="artifacts",
    command_name="/opt/deploy/artifacts/venv/bin/python3 /opt/deploy/artifacts/artifacts/artifacts.py",
)
# goresym is the command for GoReSym
shell2http.register_command(
    endpoint="goresym",
    command_name="/usr/local/bin/goresym",
)

# with this, we can make http calls to the endpoint: /thug
shell2http.register_command(
    endpoint="thug",
    command_name="/opt/deploy/thug/venv/bin/thug -qZF",
    callback_fn=intercept_thug_result,
)