You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spot.apache.org by ev...@apache.org on 2017/03/29 16:51:32 UTC
[08/50] [abbrv] incubator-spot git commit: Adding backend methods for
GraphQL API
Adding backend methods for GraphQL API
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/5b75b418
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/5b75b418
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/5b75b418
Branch: refs/heads/SPOT-35_graphql_api
Commit: 5b75b418990f15b4b8dfdf04720e2edd9b5588f4
Parents: 415c7f5
Author: Everardo Lopez Sandoval (Intel) <el...@jmoren4x-mobl2.amr.corp.intel.com>
Authored: Sun Mar 5 18:14:45 2017 -0600
Committer: Diego Ortiz Huerta <di...@intel.com>
Committed: Wed Mar 15 11:49:47 2017 -0700
----------------------------------------------------------------------
spot-oa/__init__.py | 0
spot-oa/api/resources/__init__.py | 0
spot-oa/api/resources/configurator.py | 43 ++
spot-oa/api/resources/configurator.pyc | Bin 0 -> 1706 bytes
spot-oa/api/resources/dns.py | 342 +++++++++++
spot-oa/api/resources/flow.py | 906 ++++++++++++++++++++++++++++
spot-oa/api/resources/hdfs_client.py | 50 ++
spot-oa/api/resources/impala_engine.py | 34 ++
spot-oa/api/resources/proxy.py | 383 ++++++++++++
9 files changed, 1758 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/__init__.py
----------------------------------------------------------------------
diff --git a/spot-oa/__init__.py b/spot-oa/__init__.py
new file mode 100755
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/api/resources/__init__.py
----------------------------------------------------------------------
diff --git a/spot-oa/api/resources/__init__.py b/spot-oa/api/resources/__init__.py
new file mode 100755
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/api/resources/configurator.py
----------------------------------------------------------------------
diff --git a/spot-oa/api/resources/configurator.py b/spot-oa/api/resources/configurator.py
new file mode 100644
index 0000000..1cc470e
--- /dev/null
+++ b/spot-oa/api/resources/configurator.py
@@ -0,0 +1,43 @@
+import ConfigParser
+import os
+
+def configuration():
+
+ conf_file = "/etc/spot.conf"
+ config = ConfigParser.ConfigParser()
+ config.readfp(SecHead(open(conf_file)))
+ return config
+
+def db():
+ conf = configuration()
+ return conf.get('conf', 'DBNAME').replace("'","").replace('"','')
+
+def impala():
+ conf = configuration()
+ return conf.get('conf', 'IMPALA_DEM'),conf.get('conf', 'IMPALA_PORT')
+
+def hdfs():
+ conf = configuration()
+ name_node = conf.get('conf',"NAME_NODE")
+ web_port = conf.get('conf',"WEB_PORT")
+ hdfs_user = conf.get('conf',"HUSER")
+ hdfs_user = hdfs_user.split("/")[-1].replace("'","").replace('"','')
+ return name_node,web_port,hdfs_user
+
+def spot():
+ conf = configuration()
+ return conf.get('conf',"HUSER").replace("'","").replace('"','')
+
+class SecHead(object):
+ def __init__(self, fp):
+ self.fp = fp
+ self.sechead = '[conf]\n'
+
+ def readline(self):
+ if self.sechead:
+ try:
+ return self.sechead
+ finally:
+ self.sechead = None
+ else:
+ return self.fp.readline()
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/api/resources/configurator.pyc
----------------------------------------------------------------------
diff --git a/spot-oa/api/resources/configurator.pyc b/spot-oa/api/resources/configurator.pyc
new file mode 100644
index 0000000..04505e0
Binary files /dev/null and b/spot-oa/api/resources/configurator.pyc differ
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/api/resources/dns.py
----------------------------------------------------------------------
diff --git a/spot-oa/api/resources/dns.py b/spot-oa/api/resources/dns.py
new file mode 100644
index 0000000..ba26ae9
--- /dev/null
+++ b/spot-oa/api/resources/dns.py
@@ -0,0 +1,342 @@
+import api.resources.impala_engine as ImpalaEngine
+import api.resources.hdfs_client as HDFSClient
+import api.resources.configurator as Configuration
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all the suspicious dns connections in one day.
+--------------------------------------------------------------------------
+"""
+def suspicious_queries(date, ip=None, query=None,limit=250):
+
+ db = Configuration.db()
+ sq_query = ("""
+ SELECT
+ ds.unix_tstamp,frame_len,ds.ip_dst,ds.dns_qry_name,
+ dns_qry_class,dns_qry_type,dns_qry_rcode,ml_score,tld,
+ query_rep,hh,dns_qry_class_name,dns_qry_type_name,
+ dns_qry_rcode_name,network_context
+ FROM {0}.dns_scores ds
+ LEFT JOIN {0}.dns_threat_investigation dt
+ ON (ds.dns_qry_name = dt.dns_qry_name)
+ WHERE
+ ds.y={1} AND ds.m={2} AND ds.d={3}
+ AND (dt.dns_qry_name is NULL)
+ """).format(db,date.year,date.month,date.day)
+
+ sq_filter = ""
+ sq_filter += " AND ds.ip_dst = '{0}'".format(ip) if ip else ""
+ sq_filter += " AND ds.dns_qry_name LIKE '%{0}%'".format(query) if query else ""
+ sq_filter += " ORDER BY ds.ml_score limit {0}".format(limit)
+
+ sq_query = sq_query + sq_filter
+ return ImpalaEngine.execute_query_as_list(sq_query)
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all the connectios related to a query name in one hour
+--------------------------------------------------------------------------
+"""
+def details(frame_time, query):
+
+ db = Configuration.db()
+ details_query = ("""
+ SELECT
+ unix_tstamp,frame_len,ip_dst,ip_src,dns_qry_name,dns_qry_class,
+ dns_qry_type,dns_qry_rcode,dns_a,dns_qry_type_name,
+ dns_qry_rcode_name,dns_qry_class_name
+ FROM
+ {0}.dns_edge
+ WHERE
+ y={1} AND m={2} AND d={3} AND hh={4} AND dns_qry_name = '{5}'
+ """).format(db,frame_time.year,frame_time.month,frame_time.day,\
+ frame_time.hour,query)
+
+ return ImpalaEngine.execute_query_as_list(details_query)
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all the connections with a single client.
+--------------------------------------------------------------------------
+"""
+def client_details(date, ip):
+
+ db = Configuration.db()
+ client_query =("""
+ SELECT
+ ip_dst,dns_a,dns_qry_name,ip_dst
+ FROM
+ {0}.dns_dendro
+ WHERE
+ y={1} AND m={2} AND d={3}
+ AND ip_dst='{4}'
+ """).format(db,date.year,date.month,date.day,ip)
+
+ return ImpalaEngine.execute_query_as_list(client_query)
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all the detected suspicious connections in one day.
+--------------------------------------------------------------------------
+"""
+def incident_progression(date, query,ip):
+
+ if not ip and not query:
+ return None
+
+ db = Configuration.db()
+ return_value = "dns_qry_name" if ip else "ip_dst"
+ dns_threat_query = ("""
+ SELECT
+ anchor,total,{0}
+ FROM
+ {1}.dns_threat_dendro
+ WHERE
+ y={2} AND m={3} AND d={4}
+ AND anchor = '{5}'
+ """).format(return_value,db,date.year,date.month,date.day,\
+ query if query else ip)
+
+ return ImpalaEngine.execute_query_as_list(dns_threat_query)
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all queries saved in threat investigation.
+--------------------------------------------------------------------------
+"""
+def comments(date):
+
+ db = Configuration.db()
+ comments_query = ("""
+ SELECT
+ ip_threat,dns_threat,title,text
+ FROM
+ {0}.dns_storyboard
+ WHERE
+ y={1} AND m={2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+
+ results = ImpalaEngine.execute_query_as_list(comments_query)
+ for row in results:
+ row["text"] = row["text"].replace("\n","\\n")
+ return results
+
+"""
+--------------------------------------------------------------------------
+Score connections.
+--------------------------------------------------------------------------
+"""
+def score_connection(date,ip="", dns="", ip_sev=0, dns_sev=0):
+
+ if (not ip and not ip_sev) and (not dns and not dns_sev):
+ return False
+
+ db = Configuration.db()
+ sq_query = ("""
+ SELECT
+ frame_time,unix_tstamp,frame_len,ip_dst,dns_qry_name,dns_qry_class,
+ dns_qry_type,dns_qry_rcode,ml_score,tld,query_rep,
+ hh,dns_qry_class_name,dns_qry_type_name,dns_qry_rcode_name,
+ network_context
+ FROM
+ {0}.dns_scores
+ WHERE
+ y={1} and m={2} and d={3}
+ AND (
+ """).format(db,date.year,date.month,date.day)
+
+ connections_filter = ""
+ connections_filter += "ip_dst = '{0}' ".format(ip) if ip else ""
+ connections_filter += " OR " if ip and dns else ""
+ connections_filter += "dns_qry_name = '{0}' ".format(dns) if dns else ""
+ connections_filter += ")"
+ connections = ImpalaEngine.execute_query(sq_query + connections_filter)
+
+ # add score to connections
+ fb_data = []
+ for row in connections:
+ # insert into dns_threat_investigation.
+ threat_data = (row[1],row[3],row[4],ip_sev if ip == row[3] else 0,\
+ dns_sev if dns == row[4] else 0)
+
+ fb_data.append([row[0],row[2],row[3],row[4],row[5],row[6],row[7],\
+ row[8],row[9],row[10],row[11],ip_sev,dns_sev,row[12],row[13],row[14],\
+ row[15],row[1]])
+
+ insert_command = ("""
+ INSERT INTO {0}.dns_threat_investigation
+ PARTITION (y={1},m={2},d={3})
+ VALUES {4}
+ """).format(db,date.year,date.month,date.day,threat_data)
+
+ ImpalaEngine.execute_query(insert_command)
+
+ # create feedback file.
+ app_path = Configuration.spot()
+ feedback_path = "{0}/dns/scored_results/{1}{2}{3}/feedback"\
+ .format(app_path,date.year,str(date.month).zfill(2),str(date.day).zfill(2))
+ ap_file = True
+
+ if len(HDFSClient.list_dir(feedback_path)) == 0:
+ fb_data.insert(0,["frame_time","frame_len","ip_dst","dns_qry_name",\
+ "dns_qry_class","dns_qry_type","dns_qry_rcode","score","tld","query_rep",\
+ "hh","ip_sev","dns_sev","dns_qry_class_name","dns_qry_type_name",\
+ "dns_qry_rcode_name","network_context","unix_tstamp"])
+ ap_file = False
+
+ HDFSClient.put_file_csv(fb_data,feedback_path,"ml_feedback.csv",append_file=ap_file)
+ return True
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all the scored connections in one day.
+--------------------------------------------------------------------------
+"""
+def get_scored_connections(date):
+
+ db = Configuration.db()
+ sc_query = ("""
+ SELECT
+ unix_tstamp,ip_dst,dns_qry_name,ip_sev,dns_sev
+ FROM
+ {0}.dns_threat_investigation
+ WHERE
+ y={1} AND m={2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+
+ return ImpalaEngine.execute_query_as_list(sc_query)
+
+"""
+--------------------------------------------------------------------------
+Get expanded search from raw data table.
+--------------------------------------------------------------------------
+"""
+def expanded_search(date,query=None,ip=None,limit=20):
+
+ if not ip and not query:
+ return False
+
+ db = Configuration.db()
+ if ip:
+ count = "dns_qry_name"
+ filter_param = "ip_dst"
+ filter_value = ip
+ else:
+ count = "ip_dst"
+ filter_param = "dns_qry_name"
+ filter_value = query
+
+ expanded_query = ("""
+ SELECT
+ COUNT({0}) as total,dns_qry_name,ip_dst
+ FROM
+ {1}.dns
+ WHERE y={2} AND m={3} AND d={4}
+ AND {5} = '{6}'
+ GROUP BY {0},{5}
+ ORDER BY total DESC
+ LIMIT {7}
+ """).format(count,db,date.year,date.month,date.day,\
+ filter_param,filter_value,limit if limit else 20)
+
+ return ImpalaEngine.execute_query_as_list(expanded_query)
+
+"""
+--------------------------------------------------------------------------
+Create StoryBoard.
+--------------------------------------------------------------------------
+"""
+def create_storyboard(expanded_search,date,ip,query,title,text):
+
+ if not ip and not query:
+ return False
+
+ anchor = ip if ip else query
+ create_dendro(expanded_search,date,anchor)
+ save_comments(ip,query,title,text,date)
+
+"""
+--------------------------------------------------------------------------
+Create dendrogram for StoryBoard.
+--------------------------------------------------------------------------
+"""
+def create_dendro(expanded_search,date,anchor):
+
+ db = Configuration.db()
+ for row in expanded_search:
+ dendro_query = ("""
+ INSERT INTO {0}.dns_threat_dendro PARTITION (y={1}, m={2},d={3})
+ VALUES ( '{4}',{5},'{6}','{7}')
+ """)\
+ .format(db,date.year,date.month,date.day,anchor,\
+ row["total"],row["dnsQuery"],row["clientIp"])
+
+ ImpalaEngine.execute_query(dendro_query)
+
+"""
+--------------------------------------------------------------------------
+Create save comments for StoryBoard.
+--------------------------------------------------------------------------
+"""
+def save_comments(ip,query,title,text,date):
+
+ db = Configuration.db()
+ sb_query = ("""
+ SELECT
+ ip_threat,dns_threat,title,text
+ FROM
+ {0}.dns_storyboard
+ WHERE
+ y = {1} AND m= {2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+ sb_data = ImpalaEngine.execute_query_as_list(sb_query)
+
+ # find value if already exists.
+ saved = False
+ for item in sb_data:
+ if item["ip_threat"] == ip or item["dns_threat"]== query:
+ item["title"] = title
+ item["text"] = text
+ saved = True
+
+ if not saved:
+ sb_data.append({'text': text, 'ip_threat': str(ip), 'title': title,'dns_threat':query})
+
+
+ #remove old file.
+ app_path = Configuration.spot()
+ old_file = "{0}/dns/hive/oa/storyboard/y={1}/m={2}/d={3}/"\
+ .format(app_path,date.year,date.month,date.day)
+
+ HDFSClient.delete_folder(old_file,"impala")
+ ImpalaEngine.execute_query("invalidate metadata")
+
+ for item in sb_data:
+ insert_query = ("""
+ INSERT INTO {0}.dns_storyboard PARTITION(y={1} , m={2} ,d={3})
+ VALUES ( '{4}', '{5}', '{6}','{7}')
+ """)\
+ .format(db,date.year,date.month,date.day,\
+ item["ip_threat"],item["dns_threat"],item["title"],item["text"])
+ ImpalaEngine.execute_query(insert_query)
+
+ return True
+"""
+--------------------------------------------------------------------------
+Return a list(dict) with all the data ingested during the time frame provided.
+--------------------------------------------------------------------------
+"""
+def ingest_summary(start_date,end_date):
+
+ db = Configuration.db()
+ is_query = ("""
+ SELECT
+ tdate,total
+ FROM {0}.dns_ingest_summary
+ WHERE
+ ( y <= {1} and y >= {2}) AND
+ ( m <= {3} and m >= {4})
+ """)\
+ .format(start_date.year,end_date.year,start_date.month,end_date.month)
+
+ return ImpalaEngine.execute_query_as_list(is_query)
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/api/resources/flow.py
----------------------------------------------------------------------
diff --git a/spot-oa/api/resources/flow.py b/spot-oa/api/resources/flow.py
new file mode 100755
index 0000000..d904b61
--- /dev/null
+++ b/spot-oa/api/resources/flow.py
@@ -0,0 +1,906 @@
+import api.resources.impala_engine as ImpalaEngine
+import api.resources.hdfs_client as HDFSClient
+import api.resources.configurator as Configuration
+import os
+import struct, socket
+import json
+
+"""
+--------------------------------------------------------------------------
+Return a list (dict) of all the suspicious connections that happened
+during the specified timeframe.
+--------------------------------------------------------------------------
+"""
+def suspicious_connections(date,ip=None,limit=250):
+
+ db = Configuration.db()
+ sc_query = ("""
+ SELECT
+ fs.tstart,fs.srcip,fs.dstip,fs.sport,fs.dport,proto,
+ ipkt,ibyt,opkt,obyt,ml_score,rank,srcip_internal,
+ dstip_internal,src_geoloc,dst_geoloc,src_domain,
+ dst_domain,src_rep,dst_rep
+ FROM {0}.flow_scores fs
+ LEFT JOIN {0}.flow_threat_investigation ft
+ ON (( fs.srcip = ft.srcip) OR ( fs.dstip = ft.dstip))
+ WHERE fs.y={1} AND fs.m={2} and fs.d={3}
+ AND ( ft.srcip is NULL AND ft.dstip is NULL )
+ """).format(db,date.year,date.month,date.day)
+
+ sc_filter = ""
+ if ip:
+ sc_filter = " AND ( fs.srcip='{0}' OR fs.dstip='{0}')".format(ip)
+
+ sc_filter += " ORDER BY rank limit {0}".format(limit)
+ sc_query = sc_query + sc_filter
+ return ImpalaEngine.execute_query_as_list(sc_query)
+
+"""
+--------------------------------------------------------------------------
+Retuarn a list(dict) of all the connections that happened
+between 2 IPs in one minute.
+--------------------------------------------------------------------------
+"""
+def details(src_ip,dst_ip,date):
+
+ db = Configuration.db()
+ details_query = ("""
+ SELECT
+ tstart,srcip,dstip,sport,dport,proto,flags,
+ tos,ibyt,ipkt,input,output,rip,obyt,opkt
+ FROM {0}.flow_edge
+ WHERE
+ y={1} AND m={2} AND d={3} AND hh={4} AND mn={5}
+ AND ((srcip='{6}' AND dstip='{7}')
+ OR (srcip='{7}' AND dstip='{6}'))
+ ORDER BY tstart
+ """).format(db,date.year,date.month,date.day,date.hour, \
+ date.minute,src_ip,dst_ip)
+
+ return ImpalaEngine.execute_query_as_list(details_query)
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all the connections related with a single IP.
+--------------------------------------------------------------------------
+"""
+def chord_details(ip,date):
+
+ db = Configuration.db()
+ chord_query = ("""
+ SELECT
+ srcip,dstip,ibyt,ipkt
+ FROM {0}.flow_chords
+ WHERE y={1} AND m={2} AND d={3} AND ip_threat='{4}'
+ """).format(db,date.year,date.month,date.day,ip)
+
+ return ImpalaEngine.execute_query_as_list(chord_query)
+
+"""
+--------------------------------------------------------------------------
+Return a list(dict) with all the data ingested during
+the timeframe provided.
+--------------------------------------------------------------------------
+"""
+def ingest_summary(start_date,end_date):
+
+ db = Configuration.db()
+ is_query = ("""
+ SELECT
+ tdate,total
+ FROM {0}.flow_ingest_summary
+ WHERE
+ ( y >= {1} AND y <= {2})
+ AND
+ ( m >= {3} AND m <= {4})
+ ORDER BY tdate
+ """).format(db,start_date.year,end_date.year, \
+ start_date.month,end_date.month)
+
+ return ImpalaEngine.execute_query_as_list(is_query)
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of connecions that happened between 2 ip
+grouped by second.
+--------------------------------------------------------------------------
+"""
+def time_line(ip,date):
+
+ db = Configuration.db()
+ time_line_query = ("""
+ SELECT
+ ip_threat,tstart,tend,srcip,dstip,proto,
+ sport,dport,ipkt,ibyt
+ FROM {0}.flow_timeline
+ WHERE
+ y={1} AND m={2} AND d={3}
+ AND ip_threat = '{4}'
+ """).format(db,date.year,date.month,date.day,ip)
+
+ return ImpalaEngine.execute_query_as_list(time_line_query)
+
+"""
+--------------------------------------------------------------------------
+Return json file with all the geo localization information.
+--------------------------------------------------------------------------
+"""
+def sc_geo(ip,date):
+
+ app_path = Configuration.spot()
+ file_name = "globe-{0}.json".format(ip.replace(".","_"))
+ hdfs_path = "{0}/flow/oa/storyboard/{1}/{2}/{3}/{4}" \
+ .format(app_path,date.year,date.month,date.day,ip.replace(".","_"))
+
+ if HDFSClient.file_exists(hdfs_path,file_name):
+ return json.loads(HDFSClient.get_file("{0}/{1}" \
+ .format(hdfs_path,file_name)))
+ else:
+ return {}
+
+"""
+--------------------------------------------------------------------------
+Return a list(dict) with the ip threatn information captured by the
+security expert.
+--------------------------------------------------------------------------
+"""
+def story_board(date):
+
+ db = Configuration.db()
+ sb_query= ("""
+ SELECT
+ ip_threat,title,text
+ FROM
+ {0}.flow_storyboard
+ WHERE
+ y={1} AND m={2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+
+ results = ImpalaEngine.execute_query_as_list(sb_query)
+ for row in results:
+ row["text"] = row["text"].replace("\n","\\n")
+ return results
+
+"""
+--------------------------------------------------------------------------
+Return a json file with the impact analysis information.
+--------------------------------------------------------------------------
+"""
+def impact_analysis(ip,date):
+
+ app_path = Configuration.spot()
+ file_name = "stats-{0}.json".format(ip.replace(".","_"))
+ hdfs_path = "{0}/flow/oa/storyboard/{1}/{2}/{3}/{4}" \
+ .format(app_path,date.year,date.month,date.day,ip.replace(".","_"))
+
+ if HDFSClient.file_exists(hdfs_path,file_name):
+ return json.loads(HDFSClient.get_file("{0}/{1}" \
+ .format(hdfs_path,file_name)))
+ else:
+ return {}
+
+"""
+--------------------------------------------------------------------------
+Return a list(dict) with all the inbound, outbound and twoway connections.
+--------------------------------------------------------------------------
+"""
+def incident_progression(ip,date):
+
+ app_path = Configuration.spot()
+ file_name = "threat-dendro-{0}.json".format(ip.replace(".","_"))
+
+ hdfs_path = "{0}/flow/oa/storyboard/{1}/{2}/{3}/{4}" \
+ .format(app_path,date.year,date.month,date.day,ip.replace(".","_"))
+
+ if HDFSClient.file_exists(hdfs_path,file_name):
+ return json.loads(HDFSClient.get_file("{0}/{1}" \
+ .format(hdfs_path,file_name)))
+ else:
+ return {}
+
+"""
+--------------------------------------------------------------------------
+Save scored connection into Threat investigation table.
+--------------------------------------------------------------------------
+"""
+def score_connection(score,date,src_ip=None,dst_ip=None,src_port=None,dst_port=None):
+
+ if not src_ip and not dst_ip and not src_port and not dst_port:
+ return False
+
+ db = Configuration.db()
+ # get connections to score
+ connections_query = ("""
+ SELECT
+ tstart,srcip,dstip,sport,dport, ibyt,ipkt
+ FROM {0}.flow_scores
+ WHERE
+ y = {1} AND m={2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+
+ connections_filter = ""
+ connections_filter += " AND srcip = '{0}'".format(src_ip) if src_ip else ""
+ connections_filter += " AND dstip = '{0}'".format(dst_ip) if dst_ip else ""
+
+ connections_filter += " AND sport = {0}" \
+ .format(str(src_port)) if src_port else ""
+
+ connections_filter += " AND dport = {0}" \
+ .format(str(dst_port)) if dst_port else ""
+
+ connections = ImpalaEngine.execute_query(connections_query + connections_filter)
+
+
+ # add score to connections
+ fb_data = []
+ for row in connections:
+ # insert into flow_threat_investigation.
+ threat_data = (row[0],row[1],row[2],row[3],row[4],score)
+ fb_data.append([score,row[0],row[1],row[2],row[3],row[4],row[5],row[6]])
+
+ insert_command = ("""
+ INSERT INTO {0}.flow_threat_investigation
+ PARTITION (y={1},m={2},d={3})
+ VALUES {4}
+ """).format(db,date.year,date.month,date.day,threat_data)
+
+ ImpalaEngine.execute_query(insert_command)
+
+ # create feedback file.
+ app_path = Configuration.spot()
+ feedback_path = "{0}/flow/scored_results/{1}{2}{3}/feedback" \
+ .format(app_path,date.year,str(date.month).zfill(2),str(date.day).zfill(2))
+
+ append_file = True
+ if len(HDFSClient.list_dir(feedback_path)) == 0:
+ fb_data.insert(0,["sev","tstart","sip","dip","sport","dport","ipkt","ibyt"])
+ append_file = False
+
+ HDFSClient.put_file_csv(fb_data,feedback_path,"ml_feedback.csv",\
+ append_file=append_file)
+ return True
+
+"""
+--------------------------------------------------------------------------
+Save connections details to flow_storyboard table.
+--------------------------------------------------------------------------
+"""
+def save_comment(ip,title,text,date):
+
+ #Get current table info.
+ db = Configuration.db()
+ sb_query = ("""
+ SELECT
+ ip_threat,title,text
+ FROM
+ {0}.flow_storyboard
+ WHERE
+ y = {1} AND m= {2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+
+ sb_data = ImpalaEngine.execute_query_as_list(sb_query)
+
+ # find value if already exists.
+ saved = False
+ for item in sb_data:
+ if item["ip_threat"] == ip:
+ item["title"] = title
+ item["text"] = text
+ saved = True
+
+ if not saved:
+ sb_data.append({'text': text, 'ip_threat': str(ip), 'title': title})
+
+ #remove old file.
+ app_path = Configuration.spot()
+ old_file = "{0}/flow/hive/oa/storyboard/y={1}/m={2}/d={3}/" \
+ .format(app_path,date.year,date.month,date.day)
+
+ # remove file manually to allow the comments update.
+ HDFSClient.delete_folder(old_file,"impala")
+ ImpalaEngine.execute_query("invalidate metadata")
+
+ for item in sb_data:
+ insert_query = ("""
+ INSERT INTO {0}.flow_storyboard PARTITION(y={1} , m={2} ,d={3})
+ VALUES ( '{4}', '{5}','{6}')
+ """) \
+ .format(db,date.year,date.month,date.day, \
+ item["ip_threat"],item["title"],item["text"])
+
+ ImpalaEngine.execute_query(insert_query)
+ return True
+
+"""
+--------------------------------------------------------------------------
+Get scored connections from threat investigation table.
+--------------------------------------------------------------------------
+"""
+def get_scored_connections(date):
+
+
+ db = Configuration.db()
+ scored_query = ("""
+ SELECT
+ tstart,srcip,dstip,srcport,dstport,score
+ FROM
+ {0}.flow_threat_investigation
+ WHERE
+ y={1} AND m={2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+
+ return ImpalaEngine.execute_query_as_list(scored_query)
+
+"""
+--------------------------------------------------------------------------
+Get expanded search data from raw data table.
+--------------------------------------------------------------------------
+"""
+def expanded_search(date,ip):
+
+ db = Configuration.db()
+ expanded_query = ("""
+ SELECT
+ min(treceived) as firstseen, max(treceived) as lastseen,
+ sip as srcip, dip as dstip, sport as sport,
+ dport as dport, count(sip) as conns, max(ipkt) as maxpkts,
+ avg(ipkt) as avgpkts, max(ibyt) as maxbyts, avg(ibyt) as avgbyts
+ FROM
+ {0}.flow
+ WHERE
+ y={1} AND m={2} AND d={3}
+ AND (sip ='{4}' OR dip='{4}')
+ GROUP BY
+ sip, dip,sport,dport
+ """).format(db,date.year,date.month,date.day,ip)
+
+ return ImpalaEngine.execute_query_as_list(expanded_query)
+
+"""
+--------------------------------------------------------------------------
+Generates all the required data for StoryBoard.
+--------------------------------------------------------------------------
+"""
+def create_storyboard(expanded_search,date,ip,title,text,top_results=20):
+
+
+ cpath = "{0}/context/" \
+ .format(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+ iploc = "{0}/{1}".format(cpath,'iploc.csv')
+ nwloc = "{0}/{1}".format(cpath,'networkcontext_1.csv')
+
+ connections = get_in_out_and_twoway_conns(expanded_search,top_results)
+ inbound,outbound,twoway = add_network_context(nwloc,connections["inbound"] \
+ ,connections["outbound"],connections["twoway"])
+
+ inbound,outbound,twoway = add_geospatial_info(iploc,inbound,outbound,twoway)
+ create_impact_analysis(ip, inbound,outbound,twoway, "",date)
+ create_map_view(ip,inbound,outbound,twoway,date,iploc)
+ create_incident_progression(ip,inbound,outbound,twoway,date)
+ create_time_line(ip,inbound,outbound,twoway,date)
+ save_comment(ip,title,text,date)
+ return True
+
+"""
+--------------------------------------------------------------------------
+Calculate number of inbound only, two-way, and outbound only.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def get_in_out_and_twoway_conns(expanded_search,top_results=20):
+
+ inbound = {}
+ outbound = {}
+ twoway = {}
+ srcdict = {}
+ dstdict = {}
+ conns_dict= {}
+ rowct = 0
+
+ for row in expanded_search:
+ srcdict[row['srcIp']] = {
+ 'ip_int': struct.unpack("!L", socket.inet_aton(str(row['srcIp'])))[0],
+ 'dst_ip': row['dstIp'],
+ 'dst_ip_int': struct.unpack("!L", socket.inet_aton(str(row['dstIp'])))[0],
+ 'conns': int(row['connections']),
+ 'maxbytes': int(row['maxBytes'])
+ }
+ dstdict[row['dstIp']] = {
+ 'ip_int': struct.unpack("!L", socket.inet_aton(str(row['dstIp'])))[0],
+ 'src_ip': row['srcIp'],
+ 'src_ip_int': struct.unpack("!L", socket.inet_aton(str(row['srcIp'])))[0],
+ 'conns': int(row['connections']),
+ 'maxbytes': int(row['maxBytes'])
+ }
+ rowct +=1
+
+ if rowct > 0:
+ for result in srcdict:
+ if result in dstdict:
+ twoway[result] = srcdict[result]
+ else:
+ outbound[result] = srcdict[result]
+
+ for result in dstdict:
+ if result not in srcdict:
+ inbound[result] = dstdict[result]
+
+ top_inbound_b = {}
+ top_outbound_b = {}
+ top_twoway_b = {}
+
+ if len(inbound) > 0:
+ top_inbound_b = get_top_bytes(inbound,top_results)
+ top_inbound_conns = get_top_conns(inbound,top_results)
+ top_inbound_b.update(top_inbound_conns) # merge the two dictionaries
+ if len(outbound) > 0:
+ top_outbound_b = get_top_bytes(outbound,top_results)
+ top_outbound_conns = get_top_conns(outbound,top_results)
+ top_outbound_b.update(top_outbound_conns) # merge the two dictionaries
+ if len(twoway) > 0:
+ top_twoway_b = get_top_bytes(twoway,top_results)
+ top_twoway_conns = get_top_conns(twoway,top_results)
+ top_twoway_b.update(top_twoway_conns) # merge the two dictionaries
+
+
+ result = {}
+ result["inbound"] = top_inbound_b
+ result["outbound"] = top_outbound_b
+ result["twoway"] = top_twoway_b
+
+ return result
+
+"""
+--------------------------------------------------------------------------
+Create incident progression file.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def create_incident_progression(anchor, inbound, outbound, twoway, date):
+
+ dendro_fpath = 'threat-dendro-' + anchor.replace('.','_') + ".json"
+ obj = {
+ 'name':anchor,
+ 'children': [],
+ 'time': ""
+ }
+
+ #----- Add Inbound Connections-------#
+ if len(inbound) > 0:
+ obj["children"].append({'name': 'Inbound Only', 'children': [], 'impact': 0})
+ in_ctxs = {}
+ for ip in inbound:
+ if 'nwloc' in inbound[ip] and len(inbound[ip]['nwloc']) > 0:
+ ctx = inbound[ip]['nwloc'][2]
+ if ctx not in in_ctxs:
+ in_ctxs[ctx] = 1
+ else:
+ in_ctxs[ctx] += 1
+ for ctx in in_ctxs:
+ obj["children"][0]['children'].append({
+ 'name': ctx,
+ 'impact': in_ctxs[ctx]
+ })
+
+ #------ Add Outbound ----------------#
+ if len(outbound) > 0:
+ obj["children"].append({'name':'Outbound Only','children':[],'impact':0})
+ out_ctxs = {}
+ for ip in outbound:
+ if 'nwloc' in outbound[ip] and len(outbound[ip]['nwloc']) > 0:
+ ctx = outbound[ip]['nwloc'][2]
+ if ctx not in out_ctxs:
+ out_ctxs[ctx] = 1
+ else:
+ out_ctxs[ctx] += 1
+ for ctx in out_ctxs:
+ obj["children"][1]['children'].append({
+ 'name': ctx,
+ 'impact': out_ctxs[ctx]
+ })
+
+ #------ Add TwoWay ----------------#
+ if len(twoway) > 0:
+ obj["children"].append({'name':'two way','children': [], 'impact': 0})
+ tw_ctxs = {}
+ for ip in twoway:
+ if 'nwloc' in twoway[ip] and len(twoway[ip]['nwloc']) > 0:
+ ctx = twoway[ip]['nwloc'][2]
+ if ctx not in tw_ctxs:
+ tw_ctxs[ctx] = 1
+ else:
+ tw_ctxs[ctx] += 1
+
+ for ctx in tw_ctxs:
+ obj["children"][2]['children'].append({
+ 'name': ctx,
+ 'impact': tw_ctxs[ctx]
+ })
+
+ app_path = Configuration.spot()
+ hdfs_path = "{0}/flow/oa/storyboard/{1}/{2}/{3}/{4}" \
+ .format(app_path,date.year,date.month,date.day,anchor.replace(".","_"))
+
+ if HDFSClient.put_file_json(obj,hdfs_path,dendro_fpath,overwrite_file=True):
+ return "Incident progression successfully created \n"
+ else:
+ return "Incident progression couldn't be created \n"
+
+"""
+--------------------------------------------------------------------------
+Create map view for StoryBoard.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def create_map_view(ip, inbound, outbound, twoway,date,iploc):
+
+ iplist = ''
+ if os.path.isfile(iploc):
+ iplist = np.loadtxt(iploc,dtype=np.uint32,delimiter=',',usecols={0},\
+ converters={0: lambda s: np.uint32(s.replace('"',''))})
+ else:
+ print "No iploc.csv file was found, Map View map won't be created"
+
+ response = ""
+ if iplist != '':
+ globe_fpath = 'globe-' + ip.replace('.','_') + ".json"
+ globe_json = {}
+ globe_json['type'] = "FeatureCollection"
+ globe_json['sourceips'] = []
+ globe_json['destips'] = []
+ for srcip in twoway:
+ try:
+ row = twoway[srcip]['geo']
+ globe_json['destips'].append({
+ 'type': 'Feature',
+ 'properties': {
+ 'location':row[8],
+ 'ip':srcip,
+ 'type':1
+ },
+ 'geometry': {
+ 'type': 'Point',
+ 'coordinates': [float(row[7]), float(row[6])]
+ }
+ })
+ except ValueError:
+ pass
+ for dstip in outbound:
+ try:
+ row = outbound[dstip]['geo']
+ dst_geo = outbound[dstip]['geo_dst']
+ globe_json['sourceips'].append({
+ 'type': 'Feature',
+ 'properties': {
+ 'location':row[8],
+ 'ip':dstip,
+ 'type':3
+ },
+ 'geometry': {
+ 'type': 'Point',
+ 'coordinates': [float(row[7]), float(row[6])]
+ }
+ })
+ globe_json['destips'].append({
+ 'type': 'Feature',
+ 'properties': {
+ 'location':row[8],
+ 'ip':outbound[dstip]['dst_ip'],
+ 'type':3
+ },
+ 'geometry': {
+ 'type': 'Point',
+ 'coordinates': [float(dst_geo[7]), float(dst_geo[6])]
+ }
+ })
+ except ValueError:
+ pass
+ for dstip in inbound:
+ try:
+ row = inbound[dstip]['geo']
+ dst_geo = inbound[dstip]['geo_src']
+ globe_json['sourceips'].append({
+ 'type': 'Feature',
+ 'properties': {
+ 'location':row[8],
+ 'ip':dstip,
+ 'type':2
+ },
+ 'geometry': {
+ 'type': 'Point',
+ 'coordinates': [float(row[7]), float(row[6])]
+ }
+ })
+ globe_json['destips'].append({
+ 'type': 'Feature',
+ 'properties': {
+ 'location':row[8],
+ 'ip':inbound[dstip]['src_ip'],
+ 'type':2
+ },
+ 'geometry': {
+ 'type': 'Point',
+ 'coordinates': [float(dst_geo[7]), float(dst_geo[6])]
+ }
+ })
+ except ValueError:
+ pass
+ json_str = json.dumps(globe_json)
+ app_path = Configuration.spot()
+ hdfs_path = "{0}/flow/oa/storyboard/{1}/{2}/{3}/{4}" \
+ .format(app_path,date.year,date.month,date.day,ip.replace(".","_"))
+
+ if HDFSClient.put_file_json(globe_json,hdfs_path,stats_fpath,overwrite_file=True) :
+ response = "Geolocation map successfully created \n"
+ else:
+ response = "The map can't be created without an iploc file \n"
+
+ return response
+
+"""
+--------------------------------------------------------------------------
+Create timeline for storyboard.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def create_time_line(anchor,inbound, outbound, twoway,date):
+
+ top_keys = []
+ if len(twoway) > 0: top_keys.extend(twoway.keys())
+ if len(outbound) > 0: top_keys.extend(outbound.keys())
+ if len(inbound) > 0: top_keys.extend(inbound.keys())
+
+
+ db = Configuration.db()
+
+ imp_query =("""
+ INSERT INTO TABLE {0}.flow_timeline PARTITION (y={4}, m={5},d={6})
+ SELECT
+ '{7}' ,min(treceived) as tstart, max(treceived) as tend,
+ sip as srcIP,dip as dstip, proto as proto, sport as sport,
+ dport AS dport, ipkt as ipkt, ibyt as ibyt
+ FROM
+ {0}.flow
+ WHERE y={4} AND m={5} AND d={6}
+ AND ((dip IN({1}) AND sip ='{2}') OR (sip IN({1}) AND dip ='{2}'))
+ GROUP BY sip, dip, proto, sport, dport, ipkt, ibyt
+ ORDER BY tstart
+ LIMIT {3}
+ """)
+
+ ips = "'" + "','".join(top_keys) + "'"
+ imp_query = imp_query.format(db,ips,anchor,1000,date.year,date.month, date.day,anchor)
+
+ if ImpalaEngine.execute_query(imp_query):
+ return "Timeline successfully created \n"
+ else:
+ return "Timeline couldn't be created \n"
+
+"""
+--------------------------------------------------------------------------
+Create Impact Analysis for StoryBoard.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def create_impact_analysis(anchor, inbound, outbound, twoway, threat_name,date):
+
+ stats_fpath = 'stats-' + anchor.replace('.','_') + ".json"
+
+ obj = {
+ 'name':threat_name,
+ 'children': [],
+ 'size': len(inbound) + len(outbound) + len(twoway)
+ }
+
+ #----- Add Inbound Connections-------#
+ obj["children"].append({'name': 'Inbound Only', 'children': [], 'size': len(inbound)})
+ in_ctxs = {}
+ for ip in inbound:
+ full_ctx = ''
+ if 'nwloc' in inbound[ip] and len(inbound[ip]['nwloc']) > 0:
+ full_ctx = inbound[ip]['nwloc'][2].split('.')[0]
+ ctx = get_ctx_name(full_ctx)
+ if ctx not in in_ctxs:
+ in_ctxs[ctx] = 1
+ else:
+ in_ctxs[ctx] += 1
+ for ctx in in_ctxs:
+ obj["children"][0]['children'].append({
+ 'name': ctx,
+ 'size': in_ctxs[ctx]
+ })
+
+
+ #------ Add Outbound ----------------#
+ obj["children"].append({'name':'Outbound Only','children':[],'size':len(outbound)})
+ out_ctxs = {}
+ for ip in outbound:
+ full_ctx = ''
+ if 'nwloc' in outbound[ip] and len(outbound[ip]['nwloc']) > 0:
+ full_ctx = outbound[ip]['nwloc'][2].split('.')[0]
+ ctx = get_ctx_name(full_ctx)
+ if ctx not in out_ctxs:
+ out_ctxs[ctx] = 1
+ else:
+ out_ctxs[ctx] += 1
+ for ctx in out_ctxs:
+ obj["children"][1]['children'].append({
+ 'name': ctx,
+ 'size': out_ctxs[ctx]
+ })
+
+ #------ Add Twoway ----------------#
+ obj["children"].append({'name': 'two way', 'children': [], 'size': len(twoway)})
+ tw_ctxs = {}
+ for ip in twoway:
+ full_ctx = ''
+ if 'nwloc' in twoway[ip] and len(twoway[ip]['nwloc']) > 0:
+ full_ctx = twoway[ip]['nwloc'][2].split('.')[0]
+ ctx = get_ctx_name(full_ctx)
+ if ctx not in tw_ctxs:
+ tw_ctxs[ctx] = 1
+ else:
+ tw_ctxs[ctx] += 1
+
+ for ctx in tw_ctxs:
+ obj["children"][2]['children'].append({
+ 'name': ctx,
+ 'size': tw_ctxs[ctx]
+ })
+
+ app_path = Configuration.spot()
+ hdfs_path = "{0}/flow/oa/storyboard/{1}/{2}/{3}/{4}" \
+ .format(app_path,date.year,date.month,date.day,anchor.replace(".","_"))
+
+ data = json.dumps(obj)
+ if HDFSClient.put_file_json(obj,hdfs_path,stats_fpath,overwrite_file=True):
+ return "Stats file successfully created \n"
+ else:
+ return "Stats file couldn't be created \n"
+
+"""
+--------------------------------------------------------------------------
+Get topbytess list.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def get_top_bytes(conns_dict, top):
+ topbytes = sorted(conns_dict.iteritems(),key=lambda (x,y): y['maxbytes'],reverse=True)
+ topbytes = topbytes[0:top]
+ return dict(topbytes)
+
+"""
+--------------------------------------------------------------------------
+Get top connections.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def get_top_conns(conns_dict, top):
+ topconns = sorted(conns_dict.iteritems(), key=lambda (x,y): y['conns'], reverse=True)
+ topconns = topconns[0:top]
+ return dict(topconns)
+
+"""
+--------------------------------------------------------------------------
+Get network context - get start and end ranges.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def add_network_context(nwloc,inbound,outbound,twoway):
+ nwdict = {}
+ if os.path.isfile(nwloc) :
+ with open(nwloc, 'r') as f:
+ reader = csv.reader(f,delimiter=',')
+ reader.next()
+ #address range, description
+ for row in reader:
+
+ if '/' in row[0]:
+ #Range in subnet
+ iprange = row[0].split('/')
+ if len(iprange) < 2:
+ ipend = 0
+ else:
+ ipend = int(iprange[1])
+ nwdict[row[0]] = [struct.unpack("!L", \
+ socket.inet_aton(iprange[0]))[0],\
+ struct.unpack("!L",socket.inet_aton(iprange[0]))[0]+2**(32-ipend)-1, row[1]]
+ elif '-' in row[0]:
+ #IP Range
+ iprange = row[0].split('-')
+ nwdict[row[0]] = [struct.unpack("!L",\
+ socket.inet_aton(iprange[0].replace(" ", "")))[0],\
+ struct.unpack("!L", socket.inet_aton(iprange[1].replace(" ", "")))[0], row[1]]
+ else:
+ #Exact match
+ nwdict[row[0]] = [struct.unpack("!L",\
+ socket.inet_aton(row[0]))[0],struct.unpack("!L",\
+ socket.inet_aton(row[0]))[0], row[1]]
+
+ for srcip in outbound:
+ temp_ip = struct.unpack("!L", socket.inet_aton(srcip))[0]
+ if srcip in nwdict:
+ inbound[srcip]['nwloc'] = nwdict[srcip]
+ else:
+ matchingVals = [x for x in nwdict if nwdict[x][1] >= temp_ip and nwdict[x][0] <= temp_ip]
+ outbound[srcip]['nwloc'] = nwdict[matchingVals[0]] if len(matchingVals) > 0 else ''
+
+ for dstip in twoway:
+ temp_ip = struct.unpack("!L", socket.inet_aton(dstip))[0]
+ if dstip in nwdict:
+ twoway[dstip]['nwloc'] = nwdict[dstip]
+ else:
+ matchingVals = [x for x in nwdict if nwdict[x][1] >= temp_ip and nwdict[x][0] <= temp_ip]
+ twoway[dstip]['nwloc'] = nwdict[matchingVals[0]] if len(matchingVals) > 0 else ''
+
+ for srcip in inbound:
+ temp_ip = struct.unpack("!L", socket.inet_aton(srcip))[0]
+ if srcip in nwdict:
+ inbound[srcip]['nwloc'] = nwdict[srcip]
+ else:
+ matchingVals = [x for x in nwdict if nwdict[x][1] >= temp_ip and nwdict[x][0] <= temp_ip]
+ inbound[srcip]['nwloc'] = nwdict[matchingVals[0]] if len(matchingVals) > 0 else ''
+
+ return inbound,outbound,twoway
+
+
+"""
+--------------------------------------------------------------------------
+Add Geo spatial info
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def add_geospatial_info(iploc,inbound,outbound,twoway):
+ iplist = ''
+ if os.path.isfile(iploc):
+ iplist = np.loadtxt(iploc,dtype=np.uint32,delimiter=',',usecols={0},\
+ converters={0: lambda s: np.uint32(s.replace('"',''))})
+ else:
+ print "No iploc.csv file was found, Map View map won't be created"
+
+
+ # get geospatial info, only when iplocation file is available
+ if iplist != '':
+ for srcip in outbound:
+ reader = csv.reader([linecache.getline(\
+ iploc, bisect.bisect(iplist,outbound[srcip]['ip_int'])).replace('\n','')])
+
+ outbound[srcip]['geo'] = reader.next()
+ reader = csv.reader([linecache.getline(\
+ iploc, bisect.bisect(iplist,outbound[srcip]['dst_ip_int'])).replace('\n','')])
+ outbound[srcip]['geo_dst'] = reader.next()
+
+ for dstip in twoway:
+ reader = csv.reader([linecache.getline(\
+ iploc,bisect.bisect(iplist,twoway[dstip]['ip_int'])).replace('\n','')])
+ twoway[dstip]['geo'] = reader.next()
+
+ for srcip in inbound:
+ reader = csv.reader([linecache.getline(\
+ iploc, bisect.bisect(iplist,inbound[srcip]['ip_int'])).replace('\n','')])
+
+ inbound[srcip]['geo'] = reader.next()
+ reader = csv.reader([linecache.getline(\
+ iploc, bisect.bisect(iplist,inbound[srcip]['src_ip_int'])).replace('\n','')])
+ inbound[srcip]['geo_src'] = reader.next()
+
+ return inbound,outbound,twoway
+
+"""
+--------------------------------------------------------------------------
+Get context name.
+Migrated from IPython NoteBooks.
+--------------------------------------------------------------------------
+"""
+def get_ctx_name(full_context):
+ ctx= 'DMZ'
+ if "VPN" in full_context:
+ ctx = "VPN"
+ elif "DMZ" in full_context:
+ ctx = "DMZ"
+ elif "Proxy" in full_context:
+ ctx = "Proxy"
+ elif "FW" in full_context:
+ ctx = "FW"
+ return ctx
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/api/resources/hdfs_client.py
----------------------------------------------------------------------
diff --git a/spot-oa/api/resources/hdfs_client.py b/spot-oa/api/resources/hdfs_client.py
new file mode 100644
index 0000000..d47d16c
--- /dev/null
+++ b/spot-oa/api/resources/hdfs_client.py
@@ -0,0 +1,50 @@
+
+from hdfs import InsecureClient
+from json import dump
+import api.resources.configurator as Config
+
+
+def _get_client(user=None):
+ hdfs_nm,hdfs_port,hdfs_user = Config.hdfs()
+ client = InsecureClient('http://{0}:{1}'.format(hdfs_nm,hdfs_port), user= user if user else hdfs_user)
+ print hdfs_nm,hdfs_port,hdfs_user
+ return client
+
+def get_file(hdfs_file):
+ client = _get_client()
+ with client.read(hdfs_file) as reader:
+ results = reader.read()
+ return results
+
+def put_file_csv(hdfs_file_content,hdfs_path,hdfs_file_name,append_file=False,overwrite_file=False):
+ client = _get_client()
+ hdfs_full_name = "{0}/{1}".format(hdfs_path,hdfs_file_name)
+ with client.write(hdfs_full_name,append=append_file,overwrite=overwrite_file) as writer:
+ for item in hdfs_file_content:
+ data = ','.join(str(d) for d in item)
+ writer.write("{0}\n".format(data))
+
+ return True
+
+def put_file_json(hdfs_file_content,hdfs_path,hdfs_file_name,append_file=False,overwrite_file=False):
+ client = _get_client()
+ hdfs_full_name = "{0}/{1}".format(hdfs_path,hdfs_file_name)
+ with client.write(hdfs_full_name,append=append_file,overwrite=overwrite_file,encoding='utf-8') as writer:
+ dump(hdfs_file_content, writer)
+
+ return True
+
+def delete_folder(hdfs_file,user=None):
+ client = _get_client(user)
+ client.delete(hdfs_file,recursive=True)
+
+def list_dir(hdfs_path):
+ client = _get_client()
+ return client.list(hdfs_path)
+
+def file_exists(hdfs_path,file_name):
+ files = list_dir(hdfs_path)
+ if str(file_name) in files:
+ return True
+ else:
+ return False
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/api/resources/impala_engine.py
----------------------------------------------------------------------
diff --git a/spot-oa/api/resources/impala_engine.py b/spot-oa/api/resources/impala_engine.py
new file mode 100644
index 0000000..d6b4c55
--- /dev/null
+++ b/spot-oa/api/resources/impala_engine.py
@@ -0,0 +1,34 @@
+from impala.dbapi import connect
+import api.resources.configurator as Config
+
+def create_connection():
+
+ impala_host, impala_port = Config.impala()
+ db = Config.db()
+ conn = connect(host=impala_host, port=int(impala_port),database=db)
+ return conn.cursor()
+
+def execute_query(query,fetch=False):
+
+ impala_cursor = create_connection()
+ impala_cursor.execute(query)
+
+ return impala_cursor if not fetch else impala_cursor.fetchall()
+
+def execute_query_as_list(query):
+
+ query_results = execute_query(query)
+ row_result = {}
+ results = []
+
+ for row in query_results:
+ x=0
+ for header in query_results.description:
+ row_result[header[0]] = row[x]
+ x +=1
+ results.append(row_result)
+ row_result = {}
+
+ return results
+
+
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/5b75b418/spot-oa/api/resources/proxy.py
----------------------------------------------------------------------
diff --git a/spot-oa/api/resources/proxy.py b/spot-oa/api/resources/proxy.py
new file mode 100644
index 0000000..a2a8e46
--- /dev/null
+++ b/spot-oa/api/resources/proxy.py
@@ -0,0 +1,383 @@
+import md5
+import api.resources.impala_engine as ImpalaEngine
+import api.resources.hdfs_client as HDFSClient
+import api.resources.configurator as Configuration
+from collections import defaultdict
+import json
+import os
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all the connectios related to a request name in one hour
+--------------------------------------------------------------------------
+"""
+def suspicious_requests(date,uri=None,ip=None,limit=250):
+
+ db = Configuration.db()
+ proxy_query = ("""
+ SELECT
+ ps.tdate,ps.time,ps.clientip,ps.host,ps.reqmethod,ps.useragent,
+ ps.resconttype,ps.duration,ps.username,ps.webcat,ps.referer,
+ ps.respcode,ps.uriport,ps.uripath,ps.uriquery,ps.serverip,ps.scbytes,
+ ps.csbytes,ps.fulluri,ps.ml_score,ps.uri_rep,ps.respcode_name,
+ ps.network_context
+ FROM
+ {0}.proxy_scores ps
+ LEFT JOIN
+ {0}.proxy_threat_investigation pt
+ ON (ps.fulluri = pt.fulluri)
+ WHERE
+ ps.y={1} AND ps.m={2} AND ps.d={3}
+ AND (pt.fulluri is NULL)
+ """).format(db,date.year,date.month,date.day)
+
+
+ p_filter = ""
+ p_filter += " AND ps.fulluri LIKE '%{0}%'".format(uri) if uri else ""
+ p_filter += " AND ps.clientip = '{0}'".format(ip) if ip else ""
+ p_filter += " ORDER BY ps.ml_score limit {0}".format(limit)
+ proxy_query = proxy_query + p_filter
+ return ImpalaEngine.execute_query_as_list(proxy_query)
+
+"""
+--------------------------------------------------------------------------
+Return list(dict) of all the connectios details for one request.
+--------------------------------------------------------------------------
+"""
+def details(date,uri,ip):
+
+ if not uri and not ip:
+ return None
+
+ db = Configuration.db()
+ p_details = ("""
+ SELECT
+ tdate,time,clientIp,host,webcat,respcode,respcode_name
+ ,reqmethod,useragent,resconttype,referer,uriport,serverip
+ ,scbytes,csbytes,fulluri,hh
+ FROM
+ {0}.proxy_edge
+ WHERE
+ y={1} AND m={2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+ return ImpalaEngine.execute_query_as_list(p_details)
+
+"""
+--------------------------------------------------------------------------
+Score a request
+--------------------------------------------------------------------------
+"""
+def score_request(date,score,uri):
+
+ if not score and not uri:
+ return None
+
+ db = Configuration.db()
+ p_query = ("""
+ SELECT
+ tdate,time,clientip,host,reqmethod,useragent,resconttype
+ ,duration,username,webcat,referer,respcode,uriport
+ ,uripath,uriquery,serverip,scbytes,csbytes,fulluri
+ ,word,ml_score,uri_rep,respcode_name,network_context
+ FROM
+ {0}.proxy_scores
+ WHERE
+ y={1} and m={2} and d={3}
+ AND fulluri = '{4}'
+ """).format(db,date.year,date.month,date.day,uri)
+
+ connections = ImpalaEngine.execute_query(p_query)
+
+ # add score to connections
+ fb_data = []
+
+ for row in connections:
+ cip_index = row[2]
+ uri_index = row[18]
+ tme_index = row[2]
+ hash_field = [str( md5.new(str(cip_index) + str(uri_index)).hexdigest() \
+ + str((tme_index.split(":"))[0]) )]
+
+ threat_data = (row[0],row[18],score)
+ fb_data.append([row[0],row[1],row[2],row[3],row[4],row[5],row[6],row[7] \
+ ,row[8],row[9],row[10],row[11],row[12],row[13],row[14],row[15] \
+ ,row[16],row[17],row[18],row[19],score,row[20],row[21],row[22], \
+ row[23],hash_field])
+
+ insert_command = ("""
+ INSERT INTO {0}.proxy_threat_investigation PARTITION (y={1},m={2},d={3})
+ VALUES {4}
+ """).format(db,date.year,date.month,date.day,threat_data)
+ ImpalaEngine.execute_query(insert_command)
+
+ # create feedback file.
+ app_path = Configuration.spot()
+ feedback_path = "{0}/proxy/scored_results/{1}{2}{3}/feedback"\
+ .format(app_path,date.year,str(date.month).zfill(2),str(date.day).zfill(2))
+
+ ap_file = True
+ if len(HDFSClient.list_dir(feedback_path)) == 0:
+ fb_data.insert(0,["p_date","p_time","clientip","host","reqmethod",\
+ "useragent","resconttype","duration","username","webcat","referer",\
+ "respcode","uriport","uripath","uriquery","serverip","scbytes","csbytes",\
+ "fulluri","word","score","uri_rep","uri_sev","respcode_name",\
+ "network_context","hash"])
+ ap_file = False
+
+ HDFSClient.put_file_csv(fb_data,feedback_path,"ml_feedback.csv",append_file=ap_file)
+ return True
+
+"""
+--------------------------------------------------------------------------
+Get expanded search from raw table.
+--------------------------------------------------------------------------
+"""
+def expanded_search(date,uri):
+
+ db = Configuration.db()
+ expanded_query = ("""
+ SELECT p_date, p_time, clientip, username, duration, fulluri,\
+ webcat, respcode, reqmethod,useragent, resconttype,\
+ referer, uriport, serverip, scbytes, csbytes
+ FROM {0}.proxy
+ WHERE y='{1}' AND m='{2}' AND d='{3}'
+ AND (fulluri='{4}' OR referer ='{4}')
+ ORDER BY p_time
+ """)\
+ .format(db,date.year,str(date.month).zfill(2),str(date.day).zfill(2),uri)
+ return ImpalaEngine.execute_query_as_list(expanded_query)
+
+"""
+--------------------------------------------------------------------------
+Get scored request from threat investigation.
+--------------------------------------------------------------------------
+"""
+def get_scored_requests(date):
+
+ db = Configuration.db()
+ sc_query = ("""
+ SELECT
+ tdate,fulluri,uri_sev
+ FROM
+ {0}.proxy_threat_investigation
+ WHERE
+ y={1} AND m={2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+
+ return ImpalaEngine.execute_query_as_list(sc_query)
+
+"""
+--------------------------------------------------------------------------
+Create storyboard.
+Migrated from IPython Notebooks
+--------------------------------------------------------------------------
+"""
+def create_storyboard(uri,date,title,text,expanded_search,top_results):
+
+
+ clientips = defaultdict(int)
+ reqmethods = defaultdict(int)
+ rescontype = defaultdict(int)
+ referers = defaultdict(int)
+ refered = defaultdict(int)
+ requests = defaultdict(int)
+
+ for row in expanded_search:
+ clientips[row['clientIp']]+=1
+ reqmethods[row['requestMethod']]+=1
+ rescontype[row['responseContentType']]+=1
+
+ if row['uri'] != uri:
+ #Source URI's that refered the user to the threat
+ referers[row[10]]+=1
+ if({'clientip':row['clientIp'],'referer':row['referer'],\
+ 'reqmethod':row['requestMethod'],\
+ 'resconttype':row['responseContentType']}) not in requests:
+ requests.append({'clientip':row['clientIp'],\
+ 'referer':row['referer'],'reqmethod':row['requestMethod'],\
+ 'resconttype':row['responseContentType']})
+ else:
+ #Destination URI's refered by the threat
+ refered[row['uri']]+=1
+
+ create_incident_progression(uri,requests,referers,date)
+ create_timeline(uri,clientips,date,top_results)
+ save_comments(uri,title,text,date)
+
+ return True
+
+"""
+--------------------------------------------------------------------------
+Create timeline for storyboard
+--------------------------------------------------------------------------
+"""
+def create_timeline(anchor,clientips,date,top_results):
+
+ response = ""
+ susp_ips = []
+
+
+ if clientips:
+ srtlist = sorted(list(clientips.items()), key=lambda x: x[1], reverse=True)
+ for val in srtlist[:top_results]:
+ susp_ips.append(val[0])
+
+ if anchor != "":
+ db = Configuration.db()
+ imp_query = ("""
+ INSERT INTO TABLE {0}.proxy_timeline
+ PARTITION (y={2}, m={3},d={4})
+ SELECT
+ '{7}' as p_threat, concat(cast(p_date as string),
+ ' ', cast(MIN(p_time) as string)) AS tstart,
+ concat(cast(p_date as string), ' ',
+ cast(MAX(p_time) as string)) AS tend,
+ SUM(duration) AS duration,
+ clientip, respcode,"respCodeName" as respCodeName
+ FROM {0}.proxy
+ WHERE fulluri='{1}' AND clientip IN ({5})
+ AND y='{2}' AND m='{3}' AND d='{4}'
+ GROUP BY clientip, p_time, respcode, p_date
+ LIMIT {6}
+ """)\
+ .format(db,anchor,date.year,str(date.month).zfill(2),\
+ str(date.day).zfill(2),("'" + "','".join(susp_ips) + "'")\
+ ,top_results,anchor)
+
+ app_path = Configuration.spot()
+ old_file = "{0}/proxy/hive/oa/timeline/y={1}/m={2}/d={3}"\
+ .format(app_path,date.year,date.month,date.day)
+
+ HDFSClient.delete_folder(old_file,"impala")
+ ImpalaEngine.execute_query("invalidate metadata")
+ ImpalaEngine.execute_query(imp_query)
+ response = "Timeline successfully saved"
+ else:
+ response = "Timeline couldn't be created"
+
+"""
+--------------------------------------------------------------------------
+Create inciden progression for storyboard.
+--------------------------------------------------------------------------
+"""
+def create_incident_progression(anchor,requests,referers,date):
+
+ hash_name = md5.new(str(anchor)).hexdigest()
+ file_name = "incident-progression-{0}.json".format(hash_name)
+ app_path = Configuration.spot()
+ hdfs_path = "{0}/proxy/oa/storyboard/{1}/{2}/{3}"\
+ .format(app_path,date.year,date.month,date.day)
+
+ data = {'fulluri':anchor, 'requests':requests,'referer_for':referers.keys()}
+ if HDFSClient.put_file_json(data,hdfs_path,file_name,overwrite_file=True) :
+ response = "Incident progression successfuly created"
+ else:
+ return False
+
+"""
+--------------------------------------------------------------------------
+Save comments for storyboard.
+--------------------------------------------------------------------------
+"""
+def save_comments(uri,title,text,date):
+
+ db = Configuration.db()
+ sb_query = ("""
+ SELECT
+ p_threat,title,text
+ FROM
+ {0}.proxy_storyboard
+ WHERE
+ y = {1} AND m= {2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+ sb_data = ImpalaEngine.execute_query_as_list(sb_query)
+
+
+ # find value if already exists.
+ saved = False
+ for item in sb_data:
+ if item["p_threat"] == uri:
+ item["title"] = title
+ item["text"] = text
+ saved = True
+
+ if not saved:
+ sb_data.append({'text': text, 'p_threat': str(uri), 'title': title})
+
+ #remove old file.
+ app_path = Configuration.spot()
+ old_file = "{0}/proxy/hive/oa/storyboard/y={1}/m={2}/d={3}/"\
+ .format(app_path,date.year,date.month,date.day)
+ HDFSClient.delete_folder(old_file,"impala")
+ ImpalaEngine.execute_query("invalidate metadata")
+
+ for item in sb_data:
+ insert_query = ("""
+ INSERT INTO {0}.proxy_storyboard PARTITION(y={1} , m={2} ,d={3})
+ VALUES ( '{4}', '{5}', '{6}')
+ """)\
+ .format(db,date.year,date.month,date.day,\
+ item["p_threat"],item["title"],item["text"])
+
+ ImpalaEngine.execute_query(insert_query)
+
+"""
+--------------------------------------------------------------------------
+Get storyboard comments.
+--------------------------------------------------------------------------
+"""
+def story_board(date):
+
+ db = Configuration.db()
+ sb_query= ("""
+ SELECT
+ p_threat,title,text
+ FROM
+ {0}.proxy_storyboard
+ WHERE
+ y={1} AND m={2} AND d={3}
+ """).format(db,date.year,date.month,date.day)
+
+ results = ImpalaEngine.execute_query_as_list(sb_query)
+ for row in results:
+ row["text"] = row["text"].replace("\n","\\n")
+ return results
+
+"""
+--------------------------------------------------------------------------
+Get timeline for storyboard.
+--------------------------------------------------------------------------
+"""
+def time_line(date,uri):
+
+ db = Configuration.db()
+ time_line_query = ("""
+ SELECT
+ p_threat,tstart,tend,duration,clientip,respcode,respcodename
+ FROM {0}.proxy_timeline
+ WHERE
+ y={1} AND m={2} AND d={3}
+ AND p_threat = '{4}'
+ """).format(db,date.year,date.month,date.day,uri)
+
+ return ImpalaEngine.execute_query_as_list(time_line_query)
+
+"""
+--------------------------------------------------------------------------
+Get incident progression for storyboard.
+--------------------------------------------------------------------------
+"""
+def incident_progression(date,uri):
+
+ app_path = Configuration.spot()
+ hdfs_path = "{0}/proxy/oa/storyboard/{1}/{2}/{3}".format(app_path,\
+ date.year,date.month,date.day)
+
+ hash_name = md5.new(str(uri)).hexdigest()
+ file_name = "incident-progression-{0}.json".format(hash_name)
+
+ if HDFSClient.file_exists(hdfs_path,file_name):
+ return json.loads(HDFSClient.get_file("{0}/{1}"\
+ .format(hdfs_path,file_name)))
+ else:
+ return {}