2959129380

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #169 from cisagov/CD-pe-reports-logging

Run Details

22 of 250 branches covered (8.8%)

Branch coverage included in aggregate %.

18 of 74 new or added lines in 5 files covered. (24.32%)

1 existing line in 1 file now uncovered.

237 of 1114 relevant lines covered (21.27%)

1.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

19.87

/src/pe_reports/data/db_query.py

"""Query the PE PostgreSQL database."""

# Standard Python Libraries
import sys

# Third-Party Libraries
import numpy as np
import pandas as pd
import psycopg2
from psycopg2 import OperationalError
from psycopg2.extensions import AsIs

# cisagov Libraries
from pe_reports import app

from .config import config

# Setup logging to central file
LOGGER = app.config["LOGGER"]

CONN_PARAMS_DIC = config()


def show_psycopg2_exception(err):
    """Handle errors for PostgreSQL issues."""
    err_type, err_obj, traceback = sys.exc_info()
    LOGGER.error(
        "Database connection error: %s on line number: %s", err, traceback.tb_lineno
    )


def connect():
    """Connect to PostgreSQL database."""
    conn = None
    try:
        LOGGER.info("Connecting to the PostgreSQL......")
        conn = psycopg2.connect(**CONN_PARAMS_DIC)
        LOGGER.info("Connection successful......")
    except OperationalError as err:
        show_psycopg2_exception(err)
        conn = None
    return conn


def close(conn):
    """Close connection to PostgreSQL."""
    conn.close()
    return


def get_orgs(conn):
    """Query organizations table."""
    try:
        cur = conn.cursor()
        sql = """SELECT * FROM organizations"""
        cur.execute(sql)
        pe_orgs = cur.fetchall()
        cur.close()
        return pe_orgs
    except (Exception, psycopg2.DatabaseError) as error:
        LOGGER.error("There was a problem with your database query %s", error)
    finally:
        if conn is not None:
            close(conn)


def query_creds_view(org_uid, start_date, end_date):
    """Query credentials view."""
    conn = connect()
    try:
        sql = """SELECT * FROM vw_breachcomp
        WHERE organizations_uid = %(org_uid)s
        AND modified_date BETWEEN %(start_date)s AND %(end_date)s"""
        df = pd.read_sql(
            sql,
            conn,
            params={"org_uid": org_uid, "start_date": start_date, "end_date": end_date},
        )
        return df
    except (Exception, psycopg2.DatabaseError) as error:
        LOGGER.error("There was a problem with your database query %s", error)
    finally:
        if conn is not None:
            close(conn)


def query_domMasq(org_uid, start_date, end_date):
    """Query domain masquerading table."""
    conn = connect()
    try:
        sql = """SELECT * FROM dnstwist_domain_masq
        WHERE organizations_uid = %(org_uid)s
        AND date_observed BETWEEN %(start_date)s AND %(end_date)s"""
        df = pd.read_sql(
            sql,
            conn,
            params={
                "org_uid": org_uid,
                "start_date": start_date,
                "end_date": end_date,
            },
        )
        return df
    except (Exception, psycopg2.DatabaseError) as error:
        LOGGER.error("There was a problem with your database query %s", error)
    finally:
        if conn is not None:
            close(conn)


# The 'table' parameter is used in query_shodan, query_darkweb and
# query_darkweb_cves functions to call specific tables that relate to the
# function name.  The result of this implementation reduces the code base,
# the code reduction leads to an increase in efficiency by reusing the
# function by passing only a parameter to get the required information from
# the database.


def query_shodan(org_uid, start_date, end_date, table):
    """Query Shodan table."""
    conn = connect()
    try:
        sql = """SELECT * FROM %(table)s
        WHERE organizations_uid = %(org_uid)s
        AND timestamp BETWEEN %(start_date)s AND %(end_date)s"""
        df = pd.read_sql(
            sql,
            conn,
            params={
                "table": AsIs(table),
                "org_uid": org_uid,
                "start_date": start_date,
                "end_date": end_date,
            },
        )
        return df
    except (Exception, psycopg2.DatabaseError) as error:
        LOGGER.error("There was a problem with your database query %s", error)
    finally:
        if conn is not None:
            close(conn)


def query_darkweb(org_uid, start_date, end_date, table):
    """Query Dark Web table."""
    conn = connect()
    try:
        sql = """SELECT * FROM %(table)s
        WHERE organizations_uid = %(org_uid)s
        AND date BETWEEN %(start_date)s AND %(end_date)s"""
        df = pd.read_sql(
            sql,
            conn,
            params={
                "table": AsIs(table),
                "org_uid": org_uid,
                "start_date": start_date,
                "end_date": end_date,
            },
        )
        return df
    except (Exception, psycopg2.DatabaseError) as error:
        LOGGER.error("There was a problem with your database query %s", error)
    finally:
        if conn is not None:
            close(conn)


def query_darkweb_cves(table):
    """Query Dark Web CVE table."""
    conn = connect()
    try:
        sql = """SELECT * FROM %(table)s"""
        df = pd.read_sql(
            sql,
            conn,
            params={"table": AsIs(table)},
        )
        return df
    except (Exception, psycopg2.DatabaseError) as error:
        LOGGER.error("There was a problem with your database query %s", error)
    finally:
        if conn is not None:
            close(conn)


def query_cyberSix_creds(org_uid, start_date, end_date):
    """Query cybersix_exposed_credentials table."""
    conn = connect()
    try:
        sql = """SELECT * FROM public.cybersix_exposed_credentials as creds
        WHERE organizations_uid = %(org_uid)s
        AND breach_date BETWEEN %(start)s AND %(end)s"""
        df = pd.read_sql(
            sql,
            conn,
            params={"org_uid": org_uid, "start": start_date, "end": end_date},
        )
        df["breach_date_str"] = pd.to_datetime(df["breach_date"]).dt.strftime(
            "%m/%d/%Y"
        )
        df.loc[df["breach_name"] == "", "breach_name"] = (
            "Cyber_six_" + df["breach_date_str"]
        )
        df["description"] = (
            df["description"].str.split("Query to find the related").str[0]
        )
        df["password_included"] = np.where(df["password"] != "", True, False)
        return df
    except (Exception, psycopg2.DatabaseError) as error:
        LOGGER.error("There was a problem with your database query %s", error)
    finally:
        if conn is not None:
            close(conn)

1	"""Query the PE PostgreSQL database."""
2
3	# Standard Python Libraries
4	import sys	5✔
5
6	# Third-Party Libraries
7	import numpy as np	5✔
8	import pandas as pd	5✔
9	import psycopg2	5✔
10	from psycopg2 import OperationalError	5✔
11	from psycopg2.extensions import AsIs	5✔
12
13	# cisagov Libraries
14	from pe_reports import app	5✔
15
16	from .config import config	5✔
17
18	# Setup logging to central file
19	LOGGER = app.config["LOGGER"]	5✔
20
21	CONN_PARAMS_DIC = config()	5✔
22
23
24	def show_psycopg2_exception(err):	5✔
25	"""Handle errors for PostgreSQL issues."""
26	err_type, err_obj, traceback = sys.exc_info()	5✔
27	LOGGER.error(	5✔
28	"Database connection error: %s on line number: %s", err, traceback.tb_lineno
29	)
30
31
32	def connect():	5✔
33	"""Connect to PostgreSQL database."""
34	conn = None	5✔
35	try:	5✔
36	LOGGER.info("Connecting to the PostgreSQL......")	5✔
37	conn = psycopg2.connect(**CONN_PARAMS_DIC)	5✔
NEW 38	LOGGER.info("Connection successful......")	×
39	except OperationalError as err:	5✔
40	show_psycopg2_exception(err)	5✔
41	conn = None	5✔
42	return conn	5✔
43
44
45	def close(conn):	5✔
46	"""Close connection to PostgreSQL."""
47	conn.close()	×
48	return	×
49
50
51	def get_orgs(conn):	5✔
52	"""Query organizations table."""
53	try:	×
54	cur = conn.cursor()	×
55	sql = """SELECT * FROM organizations"""	×
56	cur.execute(sql)	×
57	pe_orgs = cur.fetchall()	×
58	cur.close()	×
59	return pe_orgs	×
60	except (Exception, psycopg2.DatabaseError) as error:	×
NEW 61	LOGGER.error("There was a problem with your database query %s", error)	×
62	finally:
63	if conn is not None:	×
64	close(conn)	×
65
66
67	def query_creds_view(org_uid, start_date, end_date):	5✔
68	"""Query credentials view."""
69	conn = connect()	×
70	try:	×
71	sql = """SELECT * FROM vw_breachcomp	×
72	WHERE organizations_uid = %(org_uid)s
73	AND modified_date BETWEEN %(start_date)s AND %(end_date)s"""
74	df = pd.read_sql(	×
75	sql,
76	conn,
77	params={"org_uid": org_uid, "start_date": start_date, "end_date": end_date},
78	)
79	return df	×
80	except (Exception, psycopg2.DatabaseError) as error:	×
NEW 81	LOGGER.error("There was a problem with your database query %s", error)	×
82	finally:
83	if conn is not None:	×
84	close(conn)	×
85
86
87	def query_domMasq(org_uid, start_date, end_date):	5✔
88	"""Query domain masquerading table."""
89	conn = connect()	×
90	try:	×
91	sql = """SELECT * FROM dnstwist_domain_masq	×
92	WHERE organizations_uid = %(org_uid)s
93	AND date_observed BETWEEN %(start_date)s AND %(end_date)s"""
94	df = pd.read_sql(	×
95	sql,
96	conn,
97	params={
98	"org_uid": org_uid,
99	"start_date": start_date,
100	"end_date": end_date,
101	},
102	)
103	return df	×
104	except (Exception, psycopg2.DatabaseError) as error:	×
NEW 105	LOGGER.error("There was a problem with your database query %s", error)	×
106	finally:
107	if conn is not None:	×
108	close(conn)	×
109
110
111	# The 'table' parameter is used in query_shodan, query_darkweb and
112	# query_darkweb_cves functions to call specific tables that relate to the
113	# function name. The result of this implementation reduces the code base,
114	# the code reduction leads to an increase in efficiency by reusing the
115	# function by passing only a parameter to get the required information from
116	# the database.
117
118
119	def query_shodan(org_uid, start_date, end_date, table):	5✔
120	"""Query Shodan table."""
121	conn = connect()	×
122	try:	×
123	sql = """SELECT * FROM %(table)s	×
124	WHERE organizations_uid = %(org_uid)s
125	AND timestamp BETWEEN %(start_date)s AND %(end_date)s"""
126	df = pd.read_sql(	×
127	sql,
128	conn,
129	params={
130	"table": AsIs(table),
131	"org_uid": org_uid,
132	"start_date": start_date,
133	"end_date": end_date,
134	},
135	)
136	return df	×
137	except (Exception, psycopg2.DatabaseError) as error:	×
NEW 138	LOGGER.error("There was a problem with your database query %s", error)	×
139	finally:
140	if conn is not None:	×
141	close(conn)	×
142
143
144	def query_darkweb(org_uid, start_date, end_date, table):	5✔
145	"""Query Dark Web table."""
146	conn = connect()	×
147	try:	×
148	sql = """SELECT * FROM %(table)s	×
149	WHERE organizations_uid = %(org_uid)s
150	AND date BETWEEN %(start_date)s AND %(end_date)s"""
151	df = pd.read_sql(	×
152	sql,
153	conn,
154	params={
155	"table": AsIs(table),
156	"org_uid": org_uid,
157	"start_date": start_date,
158	"end_date": end_date,
159	},
160	)
161	return df	×
162	except (Exception, psycopg2.DatabaseError) as error:	×
NEW 163	LOGGER.error("There was a problem with your database query %s", error)	×
164	finally:
165	if conn is not None:	×
166	close(conn)	×
167
168
169	def query_darkweb_cves(table):	5✔
170	"""Query Dark Web CVE table."""
171	conn = connect()	×
172	try:	×
173	sql = """SELECT * FROM %(table)s"""	×
174	df = pd.read_sql(	×
175	sql,
176	conn,
177	params={"table": AsIs(table)},
178	)
179	return df	×
180	except (Exception, psycopg2.DatabaseError) as error:	×
NEW 181	LOGGER.error("There was a problem with your database query %s", error)	×
182	finally:
183	if conn is not None:	×
184	close(conn)	×
185
186
187	def query_cyberSix_creds(org_uid, start_date, end_date):	5✔
188	"""Query cybersix_exposed_credentials table."""
189	conn = connect()	×
190	try:	×
191	sql = """SELECT * FROM public.cybersix_exposed_credentials as creds	×
192	WHERE organizations_uid = %(org_uid)s
193	AND breach_date BETWEEN %(start)s AND %(end)s"""
194	df = pd.read_sql(	×
195	sql,
196	conn,
197	params={"org_uid": org_uid, "start": start_date, "end": end_date},
198	)
199	df["breach_date_str"] = pd.to_datetime(df["breach_date"]).dt.strftime(	×
200	"%m/%d/%Y"
201	)
202	df.loc[df["breach_name"] == "", "breach_name"] = (	×
203	"Cyber_six_" + df["breach_date_str"]
204	)
205	df["description"] = (	×
206	df["description"].str.split("Query to find the related").str[0]
207	)
208	df["password_included"] = np.where(df["password"] != "", True, False)	×
209	return df	×
210	except (Exception, psycopg2.DatabaseError) as error:	×
NEW 211	LOGGER.error("There was a problem with your database query %s", error)	×
212	finally:
213	if conn is not None:	×
214	close(conn)	×

cisagov / pe-reports / 2959129380

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous