-
Notifications
You must be signed in to change notification settings - Fork 280
Expand file tree
/
Copy pathgeoip_asn.py
More file actions
84 lines (70 loc) · 2.73 KB
/
Copy pathgeoip_asn.py
File metadata and controls
84 lines (70 loc) · 2.73 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
import json
import logging
import sys
import geoip2.database
log = logging.getLogger("geoip_asn")
log.setLevel(logging.DEBUG)
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
ch.setFormatter(formatter)
log.addHandler(ch)
misperrors = {"error": "Error"}
mispattributes = {"input": ["ip-src", "ip-dst", "domain|ip"], "output": ["freetext"]}
moduleconfig = ["local_geolite_db"]
# possible module-types: 'expansion', 'hover' or both
moduleinfo = {
"version": "0.1",
"author": "GlennHD",
"description": "Query a local copy of the Maxmind Geolite ASN database (MMDB format)",
"module-type": ["expansion", "hover"],
"name": "GeoIP ASN Lookup",
"logo": "maxmind.png",
"requirements": ["A local copy of Maxmind's Geolite database"],
"features": (
"The module takes an IP address attribute as input and queries a local copy of the Maxmind's Geolite database"
" to get information about the related AS number."
),
"references": ["https://www.maxmind.com/en/home"],
"input": "An IP address MISP attribute.",
"output": "Text containing information about the AS number of the IP address.",
"descrption": (
"An expansion module to query a local copy of Maxmind's Geolite database with an IP address, in order to get"
" information about its related AS number."
),
}
def handler(q=False):
if q is False:
return False
request = json.loads(q)
if not request.get("config") or not request["config"].get("local_geolite_db"):
return {"error": "Please specify the path of your local copy of the Maxmind Geolite ASN database"}
path_to_geolite = request["config"]["local_geolite_db"]
if request.get("ip-dst"):
toquery = request["ip-dst"]
elif request.get("ip-src"):
toquery = request["ip-src"]
elif request.get("domain|ip"):
toquery = request["domain|ip"].split("|")[1]
else:
return False
try:
reader = geoip2.database.Reader(path_to_geolite)
except FileNotFoundError:
return {"error": f"Unable to locate the GeoLite database you specified ({path_to_geolite})."}
log.debug(toquery)
try:
answer = reader.asn(toquery)
stringmap = (
"ASN=" + str(answer.autonomous_system_number) + ", AS Org=" + str(answer.autonomous_system_organization)
)
except Exception as e:
misperrors["error"] = f"GeoIP resolving error: {e}"
return misperrors
r = {"results": [{"types": mispattributes["output"], "values": stringmap}]}
return r
def introspection():
return mispattributes
def version():
moduleinfo["config"] = moduleconfig
return moduleinfo