Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[MalwareBazaar] Add SHA256 Indicators and visual change #3311

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
+138 −21
Open

[MalwareBazaar] Add SHA256 Indicators and visual change #3311

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
159 changes: 138 additions & 21 deletions external-import/malwarebazaar-recent-additions/src/malwarebazaar-recent-additions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
import datetime
import io
import json
import os
import sys
import time
Expand All @@ -8,7 +10,7 @@
import requests
import stix2
import yaml
from pycti import OpenCTIConnectorHelper, get_config_variable
from pycti import OpenCTIApiClient, OpenCTIConnectorHelper, get_config_variable


class MalwareBazaarRecentAdditions:
Expand Down Expand Up @@ -50,6 +52,18 @@ def __init__(self):
config,
)

self.url_base = get_config_variable(
"OPENCTI_URL",
["malwarebazaar_recent_additions", "opencti_url"],
config,
)

self.api_key_opencti = get_config_variable(
"OPENCTI_TOKEN",
["malwarebazaar_recent_additions", "opencti_token"],
config,
)

self.include_tags = get_config_variable(
"MALWAREBAZAAR_RECENT_ADDITIONS_INCLUDE_TAGS",
["malwarebazaar_recent_additions", "include_tags"],
Expand Down Expand Up @@ -87,6 +101,7 @@ def run(self):
self.helper.log_info("Starting MalwareBazaar Recent Additions Connector")
while True:
try:
self.update_state()
recent_additions_list = self.get_recent_additions()
for recent_additions_dict in recent_additions_list:
self.helper.log_info(f"Processing: {recent_additions_dict}")
Expand Down Expand Up @@ -122,7 +137,13 @@ def run(self):
continue

# Download the artifact and unzip with default "infected" password
file_contents = self.download_unzip(sha256)
try:
file_contents = self.download_unzip(sha256)
except:
self.helper.log_error(
f"Error downloading and unzipping {sha256}."
)
continue

# Upload the artifact to OpenCTI
response = self.upload_artifact_opencti(
Expand Down Expand Up @@ -159,9 +180,16 @@ def run(self):
id=response["id"], label_id=label["id"]
)

self.helper.log_info(
f"Re-checking for new additions in {self.cooldown_seconds} seconds..."
)
# Creation of the STIX indicator and relationship
tags_indicator = tags
tags.insert(0, "malware-bazaar")
indicator_id = self.create_stix_indicator(sha256, tags_indicator)
self.relation_ship(
self.url_base,
self.api_key_opencti,
response["id"],
indicator_id,
)

except (KeyboardInterrupt, SystemExit):
self.helper.log_info("Connector stop")
Expand Down Expand Up @@ -193,23 +221,14 @@ def get_recent_additions(self):
# Handle the response data

recent_additions_list = resp.json()
return recent_additions_list["data"]

def download_unzip(self, sha256):
"""
Download and unzip a sample from MalwareBazaar.

sha256: a str representing the sample's sha256.
returns: a bytes object containing the contents of the file
"""
data = {"query": "get_file", "sha256_hash": sha256}
resp = requests.post(self.api_url, data=data)
zip_contents = resp.content
zip_obj = io.BytesIO(zip_contents)
zip_file = pyzipper.AESZipFile(zip_obj)
zip_file.setpassword(b"infected")
file_name = zip_file.namelist()[0]
return zip_file.read(file_name)
if "data" in recent_additions_list:
return recent_additions_list["data"]
else:
self.helper.log_error(
"Key 'data' not found in the response from MalwareBazaar API."
)
return []

def artifact_exists_opencti(self, sha256):
"""
Expand Down Expand Up @@ -259,6 +278,104 @@ def upload_artifact_opencti(self, file_name, file_contents, description):

return self.helper.api.stix_cyber_observable.upload_artifact(**kwargs)

def download_unzip(self, sha256):
"""
Download and unzip a sample from MalwareBazaar.

sha256: a str representing the sample's sha256.
returns: a bytes object containing the contents of the file
"""
data = {"query": "get_file", "sha256_hash": sha256}
resp = requests.post(self.api_url, data=data)
zip_contents = resp.content
zip_obj = io.BytesIO(zip_contents)
zip_file = pyzipper.AESZipFile(zip_obj)
zip_file.setpassword(b"infected")
file_name = zip_file.namelist()[0]
return zip_file.read(file_name)

def update_state(self):
timestamp = int(time.time())
interval = self.cooldown_seconds
new_state = {"last_run": timestamp}

current_state = self.helper.get_state()
self.helper.log_info("Actual state: " + json.dumps(current_state, indent=4))

if current_state is None:
self.helper.set_state(new_state)
self.helper.log_info("Initial state: " + json.dumps(new_state))
else:
if "last_run" in current_state:
last_run = current_state["last_run"]
if timestamp - last_run < interval:
self.helper.log_info("Waiting 300 seconds...")
return

current_state["last_run"] = timestamp
self.helper.set_state(current_state)
self.helper.log_info("State update: " + json.dumps(current_state))

def create_stix_indicator(self, sha256, tags="malware-bazaar"):
"""
Create a STIX indicator for the given sha256.

sha256: a str representing the sha256 of the artifact

returns: the created indicator
"""
url = self.url_base
token = self.api_key_opencti

api_client = OpenCTIApiClient(url, token)

pattern = f"[file:hashes.'SHA-256' = '{sha256}']"
valid_from = (
datetime.datetime.now(datetime.timezone.utc)
.isoformat(timespec="milliseconds")
.replace("+00:00", "Z")
)

indicator = api_client.indicator.create(
name=f"{sha256}",
description=f"Indicator for hash SHA256 {sha256}",
pattern=pattern,
pattern_type="stix",
valid_from=valid_from,
x_opencti_main_observable_type="Artifact",
objectLabel=tags,
)

if indicator:
self.helper.log_info(f"Indicator with {sha256} created in OpenCTI.")
indicator_id = indicator["id"]
return indicator_id
else:
self.helper.log_error("Error creating indicator in OpenCTI.")

def relation_ship(self, url, token, artifact_id, indicator_id):
api_client = OpenCTIApiClient(url, token)

relation = api_client.stix_core_relationship.create(
fromType="Artifact",
fromId=artifact_id,
toType="Indicator",
toId=indicator_id,
relationship_type="related-to",
description="Relationship between artifact and indicator.",
first_seen=datetime.datetime.now(datetime.timezone.utc)
.isoformat(timespec="milliseconds")
.replace("+00:00", "Z"),
last_seen=datetime.datetime.now(datetime.timezone.utc)
.isoformat(timespec="milliseconds")
.replace("+00:00", "Z"),
)

if relation:
self.helper.log_info(f"Relation {relation["id"]} created in OpenCTI.")
else:
self.helper.log_info("Error creating relation in OpenCTI.")


if __name__ == "__main__":
try:
Expand Down