-
Notifications
You must be signed in to change notification settings - Fork 281
Expand file tree
/
Copy pathtriage_submit.py
More file actions
120 lines (95 loc) · 3.36 KB
/
Copy pathtriage_submit.py
File metadata and controls
120 lines (95 loc) · 3.36 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
import base64
import io
import json
import zipfile
import requests
misperrors = {"error": "Error"}
mispattributes = {"input": ["attachment", "malware-sample", "url"], "output": ["link"]}
moduleinfo = {
"version": "1",
"author": "Karen Yousefi",
"description": "Module to submit samples to tria.ge",
"module-type": ["expansion", "hover"],
"name": "Triage Submit",
"logo": "",
}
moduleconfig = ["apikey", "url_mode"]
def handler(q=False):
if q is False:
return False
request = json.loads(q)
if request.get("config", {}).get("apikey") is None:
misperrors["error"] = "tria.ge API key is missing"
return misperrors
api_key = request["config"]["apikey"]
url_mode = request["config"].get("url_mode", "submit") # 'submit' or 'fetch'
base_url = "https://tria.ge/api/v0/samples"
headers = {"Authorization": f"Bearer {api_key}"}
if "attachment" in request:
data = request["data"]
filename = request["attachment"]
return submit_file(headers, base_url, data, filename)
elif "malware-sample" in request:
data = request["data"]
filename = request["malware-sample"].split("|")[0]
return submit_file(headers, base_url, data, filename, is_malware_sample=True)
elif "url" in request:
url = request["url"]
return submit_url(headers, base_url, url, url_mode)
else:
misperrors["error"] = "Unsupported input type"
return misperrors
def submit_file(headers, base_url, data, filename, is_malware_sample=False):
try:
if is_malware_sample:
file_data = base64.b64decode(data)
zip_file = zipfile.ZipFile(io.BytesIO(file_data))
file_data = zip_file.read(zip_file.namelist()[0], pwd=b"infected")
else:
file_data = base64.b64decode(data)
files = {"file": (filename, file_data)}
response = requests.post(base_url, headers=headers, files=files)
response.raise_for_status()
result = response.json()
sample_id = result["id"]
sample_url = f"https://tria.ge/{sample_id}"
return {
"results": [
{
"types": "link",
"values": sample_url,
"comment": "Link to tria.ge analysis",
}
]
}
except Exception as e:
misperrors["error"] = f"Error submitting to tria.ge: {str(e)}"
return misperrors
def submit_url(headers, base_url, url, mode):
try:
if mode == "fetch":
data = {"kind": "fetch", "url": url}
else: # submit
data = {"kind": "url", "url": url}
response = requests.post(base_url, headers=headers, json=data)
response.raise_for_status()
result = response.json()
sample_id = result["id"]
sample_url = f"https://tria.ge/{sample_id}"
return {
"results": [
{
"types": "link",
"values": sample_url,
"comment": f"Link to tria.ge analysis ({mode} mode)",
}
]
}
except Exception as e:
misperrors["error"] = f"Error submitting to tria.ge: {str(e)}"
return misperrors
def introspection():
return mispattributes
def version():
moduleinfo["config"] = moduleconfig
return moduleinfo