141 lines
3.4 KiB

import jbxapi
import base64
import io
import json
import logging
import sys
import zipfile
import re
from urllib.parse import urljoin
log = logging.getLogger(__name__)
sh = logging.StreamHandler(sys.stdout)
fmt = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
moduleinfo = {
"version": "1.0",
"author": "Joe Security LLC",
"description": "Submit files and URLs to Joe Sandbox",
"module-type": ["expansion", "hover"]
moduleconfig = [
mispattributes = {
"input": ["attachment", "malware-sample", "url", "domain"],
"output": ["link"],
def handler(q=False):
if q is False:
return False
request = json.loads(q)
apiurl = request["config"].get("apiurl") or "https://jbxcloud.joesecurity.org/api"
apikey = request["config"].get("apikey")
# systems
systems = request["config"].get("systems") or ""
systems = [s.strip() for s in re.split(r"[\s,;]", systems) if s.strip()]
accept_tac = _parse_bool(request["config"].get("accept-tac"), "accept-tac")
report_cache = _parse_bool(request["config"].get("report-cache"), "report-cache")
except _ParseError as e:
return {"error": str(e)}
params = {
"report-cache": report_cache,
"systems": systems,
if not apikey:
return {"error": "No API key provided"}
joe = jbxapi.JoeSandbox(apiurl=apiurl, apikey=apikey, user_agent="MISP joesandbox_submit", accept_tac=accept_tac)
is_url_submission = "url" in request or "domain" in request
if is_url_submission:
url = request.get("url") or request.get("domain")
log.info("Submitting URL: %s", url)
result = joe.submit_url(url, params=params)
if "malware-sample" in request:
filename = request.get("malware-sample").split("|", 1)[0]
data = _decode_malware(request["data"], True)
elif "attachment" in request:
filename = request["attachment"]
data = _decode_malware(request["data"], False)
data_fp = io.BytesIO(data)
log.info("Submitting sample: %s", filename)
result = joe.submit_sample((filename, data_fp), params=params)
assert "submission_id" in result
except jbxapi.JoeException as e:
return {"error": str(e)}
link_to_analysis = urljoin(apiurl, "../submissions/{}".format(result["submission_id"]))
return {
"results": [{
"types": "link",
"categories": "External analysis",
"values": link_to_analysis,
def introspection():
return mispattributes
def version():
moduleinfo["config"] = moduleconfig
return moduleinfo
def _decode_malware(data, is_encrypted):
data = base64.b64decode(data)
if is_encrypted:
with zipfile.ZipFile(io.BytesIO(data)) as zipf:
data = zipf.read(zipf.namelist()[0], pwd=b"infected")
return data
class _ParseError(Exception):
def _parse_bool(value, name="bool"):
if value is None or value == "":
return None
if value == "true":
return True
if value == "false":
return False
raise _ParseError("Cannot parse {}. Must be 'true' or 'false'".format(name))