3386784809

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #191 from cisagov/DJ_implement_views

Run Details

49 of 336 branches covered (14.58%)

Branch coverage included in aggregate %.

11 of 39 new or added lines in 2 files covered. (28.21%)

1 existing line in 1 file now uncovered.

395 of 1353 relevant lines covered (29.19%)

1.46 hits per line

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

30.58

/src/pe_reports/metrics.py

"""Class methods for report metrics."""

# Import query functions
# Standard Python Libraries
import datetime

# Third-Party Libraries
import pandas as pd

from .data.db_query import (
    query_breachdetails_view,
    query_creds_view,
    query_credsbyday_view,
    query_darkweb,
    query_darkweb_cves,
    query_domMasq,
    query_shodan,
)


class Credentials:
    """Credentials class."""

    def __init__(self, trending_start_date, start_date, end_date, org_uid):
        """Initialize credentials class."""
        self.trending_start_date = trending_start_date
        self.start_date = start_date
        self.end_date = end_date
        self.org_uid = org_uid
        self.trending_creds_view = query_creds_view(
            org_uid, trending_start_date, end_date
        )
        self.creds_view = query_creds_view(org_uid, start_date, end_date)
        self.creds_by_day = query_credsbyday_view(
            org_uid, trending_start_date, end_date
        )
        self.breach_details_view = query_breachdetails_view(
            org_uid, start_date, end_date
        )

    def by_week(self):
        """Return number of credentials by week."""
        df = self.creds_by_day
        idx = pd.date_range(self.trending_start_date, self.end_date)
        df = df.set_index("mod_date").reindex(idx).fillna(0.0).rename_axis("added_date")
        group_limit = self.end_date + datetime.timedelta(1)
        df = df.groupby(
            pd.Grouper(level="added_date", freq="7d", origin=group_limit)
        ).sum()
        df["modified_date"] = df.index
        df["modified_date"] = df["modified_date"].dt.strftime("%m/%d")
        df = df.set_index("modified_date")
        df = df.rename(
            columns={
                "password_included": "Passwords Included",
                "no_password": "No Password",
            }
        )
        if len(df.columns) == 0:
            df["Passwords Included"] = 0
        return df

    def breaches(self):
        """Return total number of breaches."""
        all_breaches = self.creds_view["breach_name"]
        return all_breaches.nunique()

    def breach_appendix(self):
        """Return breach name and description to be added to the appendix."""
        view_df = self.creds_view
        view_df = view_df[["breach_name", "description"]]

        view_df = view_df.drop_duplicates()
        return view_df[["breach_name", "description"]]

    def breach_details(self):
        """Return breach details."""
        breach_df = self.breach_details_view
        breach_det_df = breach_df.rename(columns={"modified_date": "update_date"})
        breach_det_df["update_date"] = pd.to_datetime(breach_det_df["update_date"])
        if len(breach_det_df) > 0:
            breach_det_df["update_date"] = breach_det_df["update_date"].dt.strftime(
                "%m/%d/%y"
            )
            breach_det_df["breach_date"] = pd.to_datetime(
                breach_det_df["breach_date"]
            ).dt.strftime("%m/%d/%y")
        breach_det_df = breach_det_df.rename(
            columns={
                "breach_name": "Breach Name",
                "breach_date": "Breach Date",
                "update_date": "Date Reported",
                "password_included": "Password Included",
                "number_of_creds": "Number of Creds",
            }
        )
        return breach_det_df

    def password(self):
        """Return total number of credentials with passwords."""
        return len(self.creds_view[self.creds_view["password_included"]])

    def total(self):
        """Return total number of credentials found in breaches."""
        return self.creds_view.shape[0]


class Domains_Masqs:
    """Domains Masquerading class."""

    def __init__(self, start_date, end_date, org_uid):
        """Initialize domains masquerading class."""
        self.start_date = start_date
        self.end_date = end_date
        self.org_uid = org_uid
        df = query_domMasq(org_uid, start_date, end_date)
        self.df_mal = df[df["malicious"]]

    def count(self):
        """Return total count of malicious domains."""
        df = self.df_mal
        return len(df.index)

    def summary(self):
        """Return domain masquerading summary information."""
        if len(self.df_mal) > 0:
            domain_sum = self.df_mal[
                [
                    "domain_permutation",
                    "ipv4",
                    "ipv6",
                    "mail_server",
                    "name_server",
                ]
            ]
            domain_sum = domain_sum[:25]
            domain_sum = domain_sum.rename(
                columns={
                    "domain_permutation": "Domain",
                    "ipv4": "IPv4",
                    "ipv6": "IPv6",
                    "mail_server": "Mail Server",
                    "name_server": "Name Server",
                }
            )
        else:
            domain_sum = pd.DataFrame(
                columns=[
                    "Domain",
                    "IPv4",
                    "IPv6",
                    "Mail Server",
                    "Name Server",
                ]
            )
        return domain_sum

    def utlds(self):
        """Return count of unique top level domains."""
        mal_df = self.df_mal

        if len(mal_df.index) > 0:
            mal_df["tld"] = (
                mal_df["domain_permutation"]
                .str.split(".")
                .str[-1]
                .str.split("/")
                .str[0]
            )
            utlds = len(mal_df["tld"].unique())
        else:
            utlds = 0

        return utlds


class Malware_Vulns:
    """Malware and Vulnerabilities Class."""

    def __init__(self, start_date, end_date, org_uid):
        """Initialize Shodan vulns and malware class."""
        self.start_date = start_date
        self.end_date = end_date
        self.org_uid = org_uid
        insecure_df = query_shodan(
            org_uid,
            start_date,
            end_date,
            "vw_shodanvulns_suspected",
        )
        self.insecure_df = insecure_df

        vulns_df = query_shodan(
            org_uid, start_date, end_date, "vw_shodanvulns_verified"
        )
        vulns_df["port"] = vulns_df["port"].astype(str)
        self.vulns_df = vulns_df

        assets_df = query_shodan(org_uid, start_date, end_date, "shodan_assets")
        self.assets_df = assets_df

    @staticmethod
    def isolate_risky_assets(df):
        """Return risky assets from the insecure_df dataframe."""
        insecure = df[df["type"] == "Insecure Protocol"]
        insecure = insecure[
            (insecure["protocol"] != "http") & (insecure["protocol"] != "smtp")
        ]
        insecure["port"] = insecure["port"].astype(str)
        return insecure[["protocol", "ip", "port"]].drop_duplicates(keep="first")

    def insecure_protocols(self):
        """Get risky assets grouped by protocol."""
        risky_assets = self.isolate_risky_assets(self.insecure_df)
        risky_assets = (
            risky_assets.groupby("protocol")
            .agg(lambda x: "  ".join(set(x)))
            .reset_index()
        )
        # Limit the IP column to 32 characters so the table isn't too big.
        # 30 characters is the max length of 2 IPs, plus the 2 spaces.
        if len(risky_assets.index) > 0:
            risky_assets["ip"] = risky_assets["ip"].str[:32]
            risky_assets.loc[risky_assets["ip"].str.len() == 32, "ip"] = (
                risky_assets["ip"] + "  ..."
            )

        return risky_assets

    def protocol_count(self):
        """Return a count for each insecure protocol."""
        risky_assets = self.isolate_risky_assets(self.insecure_df)
        # Horizontal bar: insecure protocol count
        pro_count = risky_assets.groupby(["protocol"], as_index=False)["protocol"].agg(
            {"id_count": "count"}
        )
        return pro_count

    def risky_ports_count(self):
        """Return total count of insecure protocols."""
        risky_assets = self.isolate_risky_assets(self.insecure_df)

        pro_count = risky_assets.groupby(["protocol"], as_index=False)["protocol"].agg(
            {"id_count": "count"}
        )

        # Total Open Ports with Insecure protocols
        return pro_count["id_count"].sum()

    def total_verif_vulns(self):
        """Return total count of verified vulns."""
        vulns_df = self.vulns_df
        verif_vulns = (
            vulns_df[["cve", "ip", "port"]]
            .groupby("cve")
            .agg(lambda x: "  ".join(set(x)))
            .reset_index()
        )

        if len(verif_vulns) > 0:
            verif_vulns["count"] = verif_vulns["ip"].str.split("  ").str.len()
            verifVulns = verif_vulns["count"].sum()

        else:
            verifVulns = 0

        return verifVulns

    def unverified_cve(self):
        """Return top 15 unverified CVEs and their counts."""
        insecure_df = self.insecure_df
        unverif_df = insecure_df[insecure_df["type"] != "Insecure Protocol"]
        unverif_df = unverif_df.copy()
        unverif_df["potential_vulns"] = (
            unverif_df["potential_vulns"].sort_values().apply(lambda x: sorted(x))
        )
        unverif_df["potential_vulns"] = unverif_df["potential_vulns"].astype("str")
        unverif_df = (
            unverif_df[["potential_vulns", "ip"]]
            .drop_duplicates(keep="first")
            .reset_index(drop=True)
        )
        unverif_df["count"] = unverif_df["potential_vulns"].str.split(",").str.len()
        unverif_df = unverif_df[["ip", "count"]]
        unverif_df = unverif_df.sort_values(by=["count"], ascending=False)
        unverif_df = unverif_df[:15].reset_index(drop=True)
        return unverif_df

    def unverified_vuln_count(self):
        """Return the count of IP addresses with unverified vulnerabilities."""
        insecure_df = self.insecure_df
        unverif_df = insecure_df[insecure_df["type"] != "Insecure Protocol"]
        unverif_df = unverif_df.copy()
        unverif_df["potential_vulns"] = (
            unverif_df["potential_vulns"].sort_values().apply(lambda x: sorted(x))
        )
        unverif_df["potential_vulns"] = unverif_df["potential_vulns"].astype("str")
        unverif_df = (
            unverif_df[["potential_vulns", "ip"]]
            .drop_duplicates(keep="first")
            .reset_index(drop=True)
        )

        return len(unverif_df.index)

    def verif_vulns(self):
        """Return a dataframe with each CVE, the associated IPs and the affected ports."""
        vulns_df = self.vulns_df
        verif_vulns = (
            vulns_df[["cve", "ip", "port"]]
            .groupby("cve")
            .agg(lambda x: "  ".join(set(x)))
            .reset_index()
        )
        return verif_vulns

    def verif_vulns_summary(self):
        """Return summary dataframe for verified vulns."""
        vulns_df = self.vulns_df
        verif_vulns_summary = (
            vulns_df[["cve", "ip", "port", "summary"]]
            .groupby("cve")
            .agg(lambda x: "  ".join(set(x)))
            .reset_index()
        )

        verif_vulns_summary = verif_vulns_summary.rename(
            columns={
                "cve": "CVE",
                "ip": "IP",
                "port": "Port",
                "summary": "Summary",
            }
        )
        return verif_vulns_summary


class Cyber_Six:
    """Dark web and Cyber Six data class."""

    def __init__(self, trending_start_date, start_date, end_date, org_uid):
        """Initialize Cybersixgill vulns and malware class."""
        self.trending_start_date = trending_start_date
        self.start_date = start_date
        self.end_date = end_date
        self.org_uid = org_uid

        trending_dark_web_mentions = query_darkweb(
            org_uid,
            trending_start_date,
            end_date,
            "mentions",
        )
        trending_dark_web_mentions = trending_dark_web_mentions.drop(
            columns=["organizations_uid", "mentions_uid"],
            errors="ignore",
        )
        self.trending_dark_web_mentions = trending_dark_web_mentions

        dark_web_mentions = query_darkweb(
            org_uid,
            start_date,
            end_date,
            "mentions",
        )
        dark_web_mentions = dark_web_mentions.drop(
            columns=["organizations_uid", "mentions_uid"],
            errors="ignore",
        )
        self.dark_web_mentions = dark_web_mentions

        alerts = query_darkweb(
            org_uid,
            start_date,
            end_date,
            "alerts",
        )
        alerts = alerts.drop(
            columns=["organizations_uid", "alerts_uid"],
            errors="ignore",
        )
        self.alerts = alerts

        top_cves = query_darkweb_cves(
            "top_cves",
        )
        top_cves = top_cves[top_cves["date"] == top_cves["date"].max()]
        self.top_cves = top_cves

    def alerts_exec(self):
        """Get top executive mentions."""
        alerts = self.alerts
        alerts_exec = alerts[["site", "title"]]
        alerts_exec = alerts_exec[alerts_exec["site"] != "NaN"]
        alerts_exec = alerts_exec[alerts_exec["site"] != ""]
        alerts_exec = (
            alerts_exec.groupby(["site", "title"])["title"]
            .count()
            .nlargest(10)
            .reset_index(name="Events")
        )
        alerts_exec = alerts_exec.rename(columns={"site": "Site", "title": "Title"})
        return alerts_exec

    def alerts_threats(self):
        """Get threat alerts."""
        alerts = self.alerts
        alerts_threats = alerts[["site", "threats"]]
        alerts_threats = alerts_threats[alerts_threats["site"] != "NaN"]
        alerts_threats = alerts_threats[alerts_threats["site"] != ""]
        alerts_threats = (
            alerts_threats.groupby(["site", "threats"])["threats"]
            .count()
            .nlargest(5)
            .reset_index(name="Events")
        )
        alerts_threats["threats"] = alerts_threats["threats"].str.strip("{}")
        alerts_threats["threats"] = alerts_threats["threats"].str[:50]
        alerts_threats = alerts_threats.rename(
            columns={"site": "Site", "threats": "Threats"}
        )
        return alerts_threats

    def dark_web_bad_actors(self):
        """Get dark web bad actors."""
        dark_web_mentions = self.dark_web_mentions
        dark_web_bad_actors = dark_web_mentions[["creator", "rep_grade"]]
        dark_web_bad_actors = dark_web_bad_actors.groupby(
            "creator", as_index=False
        ).max()
        dark_web_bad_actors = dark_web_bad_actors.sort_values(
            by=["rep_grade"], ascending=False
        )[:10]
        dark_web_bad_actors["rep_grade"] = (
            dark_web_bad_actors["rep_grade"].astype(float).round(decimals=3)
        )
        dark_web_bad_actors = dark_web_bad_actors.rename(
            columns={"creator": "Creator", "rep_grade": "Grade"}
        )
        return dark_web_bad_actors

    def dark_web_content(self):
        """Get dark web categories."""
        dark_web_mentions = self.dark_web_mentions
        dark_web_content = dark_web_mentions[["category"]]
        dark_web_content = (
            dark_web_content.groupby(["category"])["category"]
            .count()
            .nlargest(10)
            .reset_index(name="count")
        )
        return dark_web_content

    def dark_web_count(self):
        """Get total number of dark web mentions."""
        return len(self.dark_web_mentions.index)

    def dark_web_date(self):
        """Get dark web mentions by date."""
        dark_web_mentions = self.trending_dark_web_mentions
        dark_web_date = dark_web_mentions[["date"]]
        dark_web_date = (
            dark_web_date.groupby(["date"])["date"].count().reset_index(name="Count")
        )
        dark_web_date["date"] = pd.to_datetime(dark_web_date["date"])
        idx = pd.date_range(self.trending_start_date, self.end_date)
        dark_web_date = (
            dark_web_date.set_index("date").reindex(idx).fillna(0.0).rename_axis("date")
        )

        group_limit = self.end_date + datetime.timedelta(1)
        dark_web_date = dark_web_date.groupby(
            pd.Grouper(  # lgtm [py/call/wrong-named-class-argument]
                level="date", freq="7d", origin=group_limit
            )
        ).sum()
        dark_web_date["date"] = dark_web_date.index
        dark_web_date["date"] = dark_web_date["date"].dt.strftime("%m/%d")
        dark_web_date = dark_web_date.set_index("date")
        dark_web_date = dark_web_date[["Count"]]
        return dark_web_date

    def dark_web_most_act(self):
        """Get most active posts."""
        dark_web_mentions = self.dark_web_mentions
        dark_web_most_act = dark_web_mentions[["title", "comments_count"]]
        dark_web_most_act = dark_web_most_act[
            dark_web_most_act["comments_count"] != "NaN"
        ]
        dark_web_most_act = dark_web_most_act.rename(
            columns={"comments_count": "Comments Count"}
        )
        dark_web_most_act["Comments Count"] = (
            dark_web_most_act["Comments Count"].astype(float).astype(int)
        )
        dark_web_most_act = dark_web_most_act.sort_values(
            by="Comments Count", ascending=False
        )
        dark_web_most_act = dark_web_most_act[:5]
        dark_web_most_act = dark_web_most_act.rename(columns={"title": "Title"})
        dark_web_most_act["Title"] = dark_web_most_act["Title"].str[:100]
        dark_web_most_act = dark_web_most_act.replace(r"^\s*$", "Untitled", regex=True)
        return dark_web_most_act

    def dark_web_sites(self):
        """Get mentions by dark web sites (top 10)."""
        dark_web_mentions = self.dark_web_mentions
        dark_web_sites = dark_web_mentions[["site"]]
        dark_web_sites = (
            dark_web_sites.groupby(["site"])["site"]
            .count()
            .nlargest(10)
            .reset_index(name="count")
        )
        dark_web_sites = dark_web_sites.rename(
            columns={"site": "Site", "count": "Count"}
        )
        return dark_web_sites

    def dark_web_tags(self):
        """Get dark web notable tags."""
        dark_web_mentions = self.dark_web_mentions
        dark_web_tags = dark_web_mentions[["tags"]]
        dark_web_tags = dark_web_tags[dark_web_tags["tags"] != "NaN"]
        dark_web_tags = (
            dark_web_tags.groupby(["tags"])["tags"]
            .count()
            .nlargest(8)
            .reset_index(name="Events")
        )
        dark_web_tags["tags"] = dark_web_tags["tags"].str.strip("{}")
        dark_web_tags["tags"] = dark_web_tags["tags"].str.replace(
            ",", ", ", regex=False
        )
        dark_web_tags = dark_web_tags.rename(columns={"tags": "Tags"})
        return dark_web_tags

    def alerts_site(self):
        """Get alerts in invite-only markets."""
        alerts_site = self.alerts[["site"]]
        alerts_site = alerts_site[alerts_site["site"] != "NaN"]
        alerts_site = alerts_site[alerts_site["site"] != ""]
        alerts_site = alerts_site[alerts_site["site"].str.startswith("market")]
        alerts_site = (
            alerts_site.groupby(["site"])["site"]
            .count()
            .nlargest(10)
            .reset_index(name="Alerts")
        )
        alerts_site = alerts_site.rename(columns={"site": "Site"})
        return alerts_site

    def top_cve_table(self):
        """Get top CVEs."""
        top_cves = self.top_cves
        top_cves["summary_short"] = top_cves["summary"].str[:400]
        top_cve_table = top_cves[["cve_id", "summary_short"]]
        top_cve_table = top_cve_table.rename(
            columns={"cve_id": "CVE", "summary_short": "Description"}
        )
        return top_cve_table

1	"""Class methods for report metrics."""
2
3	# Import query functions
4	# Standard Python Libraries
5	import datetime	5✔
6
7	# Third-Party Libraries
8	import pandas as pd	5✔
9
10	from .data.db_query import (	5✔
11	query_breachdetails_view,
12	query_creds_view,
13	query_credsbyday_view,
14	query_darkweb,
15	query_darkweb_cves,
16	query_domMasq,
17	query_shodan,
18	)
19
20
21	class Credentials:	5✔
22	"""Credentials class."""
23
24	def __init__(self, trending_start_date, start_date, end_date, org_uid):	5✔
25	"""Initialize credentials class."""
26	self.trending_start_date = trending_start_date	5✔
27	self.start_date = start_date	5✔
28	self.end_date = end_date	5✔
29	self.org_uid = org_uid	5✔
30	self.trending_creds_view = query_creds_view(	5✔
31	org_uid, trending_start_date, end_date
32	)
33	self.creds_view = query_creds_view(org_uid, start_date, end_date)	5✔
34	self.creds_by_day = query_credsbyday_view(	5✔
35	org_uid, trending_start_date, end_date
36	)
37	self.breach_details_view = query_breachdetails_view(	5✔
38	org_uid, start_date, end_date
39	)
40
41	def by_week(self):	5✔
42	"""Return number of credentials by week."""
43	df = self.creds_by_day	5✔
44	idx = pd.date_range(self.trending_start_date, self.end_date)	5✔
45	df = df.set_index("mod_date").reindex(idx).fillna(0.0).rename_axis("added_date")	5✔
46	group_limit = self.end_date + datetime.timedelta(1)	5✔
47	df = df.groupby(	5✔
48	pd.Grouper(level="added_date", freq="7d", origin=group_limit)
49	).sum()
50	df["modified_date"] = df.index	5✔
51	df["modified_date"] = df["modified_date"].dt.strftime("%m/%d")	5✔
52	df = df.set_index("modified_date")	5✔
53	df = df.rename(	5✔
54	columns={
55	"password_included": "Passwords Included",
56	"no_password": "No Password",
57	}
58	)
59	if len(df.columns) == 0:	5!
60	df["Passwords Included"] = 0	×
61	return df	5✔
62
63	def breaches(self):	5✔
64	"""Return total number of breaches."""
65	all_breaches = self.creds_view["breach_name"]	5✔
66	return all_breaches.nunique()	5✔
67
68	def breach_appendix(self):	5✔
69	"""Return breach name and description to be added to the appendix."""
70	view_df = self.creds_view	5✔
71	view_df = view_df[["breach_name", "description"]]	5✔
72
73	view_df = view_df.drop_duplicates()	5✔
74	return view_df[["breach_name", "description"]]	5✔
75
76	def breach_details(self):	5✔
77	"""Return breach details."""
78	breach_df = self.breach_details_view	5✔
79	breach_det_df = breach_df.rename(columns={"modified_date": "update_date"})	5✔
80	breach_det_df["update_date"] = pd.to_datetime(breach_det_df["update_date"])	5✔
81	if len(breach_det_df) > 0:	5!
82	breach_det_df["update_date"] = breach_det_df["update_date"].dt.strftime(	5✔
83	"%m/%d/%y"
84	)
85	breach_det_df["breach_date"] = pd.to_datetime(	5✔
86	breach_det_df["breach_date"]
87	).dt.strftime("%m/%d/%y")
88	breach_det_df = breach_det_df.rename(	5✔
89	columns={
90	"breach_name": "Breach Name",
91	"breach_date": "Breach Date",
92	"update_date": "Date Reported",
93	"password_included": "Password Included",
94	"number_of_creds": "Number of Creds",
95	}
96	)
97	return breach_det_df	5✔
98
99	def password(self):	5✔
100	"""Return total number of credentials with passwords."""
101	return len(self.creds_view[self.creds_view["password_included"]])	5✔
102
103	def total(self):	5✔
104	"""Return total number of credentials found in breaches."""
105	return self.creds_view.shape[0]	5✔
106
107
108	class Domains_Masqs:	5✔
109	"""Domains Masquerading class."""
110
111	def __init__(self, start_date, end_date, org_uid):	5✔
112	"""Initialize domains masquerading class."""
113	self.start_date = start_date	×
114	self.end_date = end_date	×
115	self.org_uid = org_uid	×
116	df = query_domMasq(org_uid, start_date, end_date)	×
117	self.df_mal = df[df["malicious"]]	×
118
119	def count(self):	5✔
120	"""Return total count of malicious domains."""
121	df = self.df_mal	×
122	return len(df.index)	×
123
124	def summary(self):	5✔
125	"""Return domain masquerading summary information."""
126	if len(self.df_mal) > 0:	×
127	domain_sum = self.df_mal[	×
128	[
129	"domain_permutation",
130	"ipv4",
131	"ipv6",
132	"mail_server",
133	"name_server",
134	]
135	]
136	domain_sum = domain_sum[:25]	×
137	domain_sum = domain_sum.rename(	×
138	columns={
139	"domain_permutation": "Domain",
140	"ipv4": "IPv4",
141	"ipv6": "IPv6",
142	"mail_server": "Mail Server",
143	"name_server": "Name Server",
144	}
145	)
146	else:
147	domain_sum = pd.DataFrame(	×
148	columns=[
149	"Domain",
150	"IPv4",
151	"IPv6",
152	"Mail Server",
153	"Name Server",
154	]
155	)
156	return domain_sum	×
157
158	def utlds(self):	5✔
159	"""Return count of unique top level domains."""
160	mal_df = self.df_mal	×
161
162	if len(mal_df.index) > 0:	×
163	mal_df["tld"] = (	×
164	mal_df["domain_permutation"]
165	.str.split(".")
166	.str[-1]
167	.str.split("/")
168	.str[0]
169	)
170	utlds = len(mal_df["tld"].unique())	×
171	else:
172	utlds = 0	×
173
174	return utlds	×
175
176
177	class Malware_Vulns:	5✔
178	"""Malware and Vulnerabilities Class."""
179
180	def __init__(self, start_date, end_date, org_uid):	5✔
181	"""Initialize Shodan vulns and malware class."""
182	self.start_date = start_date	×
183	self.end_date = end_date	×
184	self.org_uid = org_uid	×
185	insecure_df = query_shodan(	×
186	org_uid,
187	start_date,
188	end_date,
189	"vw_shodanvulns_suspected",
190	)
191	self.insecure_df = insecure_df	×
192
NEW 193	vulns_df = query_shodan(	×
194	org_uid, start_date, end_date, "vw_shodanvulns_verified"
195	)
196	vulns_df["port"] = vulns_df["port"].astype(str)	×
197	self.vulns_df = vulns_df	×
198
199	assets_df = query_shodan(org_uid, start_date, end_date, "shodan_assets")	×
200	self.assets_df = assets_df	×
201
202	@staticmethod	5✔
203	def isolate_risky_assets(df):	3✔
204	"""Return risky assets from the insecure_df dataframe."""
205	insecure = df[df["type"] == "Insecure Protocol"]	×
206	insecure = insecure[	×
207	(insecure["protocol"] != "http") & (insecure["protocol"] != "smtp")
208	]
NEW 209	insecure["port"] = insecure["port"].astype(str)	×
NEW 210	return insecure[["protocol", "ip", "port"]].drop_duplicates(keep="first")	×
211
212	def insecure_protocols(self):	5✔
213	"""Get risky assets grouped by protocol."""
NEW 214	risky_assets = self.isolate_risky_assets(self.insecure_df)	×
NEW 215	risky_assets = (	×
216	risky_assets.groupby("protocol")
217	.agg(lambda x: " ".join(set(x)))
218	.reset_index()
219	)
220	# Limit the IP column to 32 characters so the table isn't too big.
221	# 30 characters is the max length of 2 IPs, plus the 2 spaces.
NEW 222	if len(risky_assets.index) > 0:	×
NEW 223	risky_assets["ip"] = risky_assets["ip"].str[:32]	×
NEW 224	risky_assets.loc[risky_assets["ip"].str.len() == 32, "ip"] = (	×
225	risky_assets["ip"] + " ..."
226	)
227
NEW 228	return risky_assets	×
229
230	def protocol_count(self):	5✔
231	"""Return a count for each insecure protocol."""
232	risky_assets = self.isolate_risky_assets(self.insecure_df)	×
233	# Horizontal bar: insecure protocol count
234	pro_count = risky_assets.groupby(["protocol"], as_index=False)["protocol"].agg(	×
235	{"id_count": "count"}
236	)
237	return pro_count	×
238
239	def risky_ports_count(self):	5✔
240	"""Return total count of insecure protocols."""
241	risky_assets = self.isolate_risky_assets(self.insecure_df)	×
242
243	pro_count = risky_assets.groupby(["protocol"], as_index=False)["protocol"].agg(	×
244	{"id_count": "count"}
245	)
246
247	# Total Open Ports with Insecure protocols
248	return pro_count["id_count"].sum()	×
249
250	def total_verif_vulns(self):	5✔
251	"""Return total count of verified vulns."""
252	vulns_df = self.vulns_df	×
253	verif_vulns = (	×
254	vulns_df[["cve", "ip", "port"]]
255	.groupby("cve")
256	.agg(lambda x: " ".join(set(x)))
257	.reset_index()
258	)
259
260	if len(verif_vulns) > 0:	×
261	verif_vulns["count"] = verif_vulns["ip"].str.split(" ").str.len()	×
262	verifVulns = verif_vulns["count"].sum()	×
263
264	else:
265	verifVulns = 0	×
266
267	return verifVulns	×
268
269	def unverified_cve(self):	5✔
270	"""Return top 15 unverified CVEs and their counts."""
271	insecure_df = self.insecure_df	×
272	unverif_df = insecure_df[insecure_df["type"] != "Insecure Protocol"]	×
273	unverif_df = unverif_df.copy()	×
274	unverif_df["potential_vulns"] = (	×
275	unverif_df["potential_vulns"].sort_values().apply(lambda x: sorted(x))
276	)
277	unverif_df["potential_vulns"] = unverif_df["potential_vulns"].astype("str")	×
278	unverif_df = (	×
279	unverif_df[["potential_vulns", "ip"]]
280	.drop_duplicates(keep="first")
281	.reset_index(drop=True)
282	)
283	unverif_df["count"] = unverif_df["potential_vulns"].str.split(",").str.len()	×
284	unverif_df = unverif_df[["ip", "count"]]	×
285	unverif_df = unverif_df.sort_values(by=["count"], ascending=False)	×
286	unverif_df = unverif_df[:15].reset_index(drop=True)	×
287	return unverif_df	×
288
289	def unverified_vuln_count(self):	5✔
290	"""Return the count of IP addresses with unverified vulnerabilities."""
291	insecure_df = self.insecure_df	×
292	unverif_df = insecure_df[insecure_df["type"] != "Insecure Protocol"]	×
293	unverif_df = unverif_df.copy()	×
294	unverif_df["potential_vulns"] = (	×
295	unverif_df["potential_vulns"].sort_values().apply(lambda x: sorted(x))
296	)
297	unverif_df["potential_vulns"] = unverif_df["potential_vulns"].astype("str")	×
298	unverif_df = (	×
299	unverif_df[["potential_vulns", "ip"]]
300	.drop_duplicates(keep="first")
301	.reset_index(drop=True)
302	)
303
304	return len(unverif_df.index)	×
305
306	def verif_vulns(self):	5✔
307	"""Return a dataframe with each CVE, the associated IPs and the affected ports."""
308	vulns_df = self.vulns_df	×
309	verif_vulns = (	×
310	vulns_df[["cve", "ip", "port"]]
311	.groupby("cve")
312	.agg(lambda x: " ".join(set(x)))
313	.reset_index()
314	)
315	return verif_vulns	×
316
317	def verif_vulns_summary(self):	5✔
318	"""Return summary dataframe for verified vulns."""
319	vulns_df = self.vulns_df	×
320	verif_vulns_summary = (	×
321	vulns_df[["cve", "ip", "port", "summary"]]
322	.groupby("cve")
323	.agg(lambda x: " ".join(set(x)))
324	.reset_index()
325	)
326
327	verif_vulns_summary = verif_vulns_summary.rename(	×
328	columns={
329	"cve": "CVE",
330	"ip": "IP",
331	"port": "Port",
332	"summary": "Summary",
333	}
334	)
335	return verif_vulns_summary	×
336
337
338	class Cyber_Six:	5✔
339	"""Dark web and Cyber Six data class."""
340
341	def __init__(self, trending_start_date, start_date, end_date, org_uid):	5✔
342	"""Initialize Cybersixgill vulns and malware class."""
343	self.trending_start_date = trending_start_date	×
344	self.start_date = start_date	×
345	self.end_date = end_date	×
346	self.org_uid = org_uid	×
347
348	trending_dark_web_mentions = query_darkweb(	×
349	org_uid,
350	trending_start_date,
351	end_date,
352	"mentions",
353	)
354	trending_dark_web_mentions = trending_dark_web_mentions.drop(	×
355	columns=["organizations_uid", "mentions_uid"],
356	errors="ignore",
357	)
358	self.trending_dark_web_mentions = trending_dark_web_mentions	×
359
360	dark_web_mentions = query_darkweb(	×
361	org_uid,
362	start_date,
363	end_date,
364	"mentions",
365	)
366	dark_web_mentions = dark_web_mentions.drop(	×
367	columns=["organizations_uid", "mentions_uid"],
368	errors="ignore",
369	)
370	self.dark_web_mentions = dark_web_mentions	×
371
372	alerts = query_darkweb(	×
373	org_uid,
374	start_date,
375	end_date,
376	"alerts",
377	)
378	alerts = alerts.drop(	×
379	columns=["organizations_uid", "alerts_uid"],
380	errors="ignore",
381	)
382	self.alerts = alerts	×
383
384	top_cves = query_darkweb_cves(	×
385	"top_cves",
386	)
387	top_cves = top_cves[top_cves["date"] == top_cves["date"].max()]	×
388	self.top_cves = top_cves	×
389
390	def alerts_exec(self):	5✔
391	"""Get top executive mentions."""
392	alerts = self.alerts	×
393	alerts_exec = alerts[["site", "title"]]	×
394	alerts_exec = alerts_exec[alerts_exec["site"] != "NaN"]	×
395	alerts_exec = alerts_exec[alerts_exec["site"] != ""]	×
396	alerts_exec = (	×
397	alerts_exec.groupby(["site", "title"])["title"]
398	.count()
399	.nlargest(10)
400	.reset_index(name="Events")
401	)
402	alerts_exec = alerts_exec.rename(columns={"site": "Site", "title": "Title"})	×
403	return alerts_exec	×
404
405	def alerts_threats(self):	5✔
406	"""Get threat alerts."""
407	alerts = self.alerts	×
408	alerts_threats = alerts[["site", "threats"]]	×
409	alerts_threats = alerts_threats[alerts_threats["site"] != "NaN"]	×
410	alerts_threats = alerts_threats[alerts_threats["site"] != ""]	×
411	alerts_threats = (	×
412	alerts_threats.groupby(["site", "threats"])["threats"]
413	.count()
414	.nlargest(5)
415	.reset_index(name="Events")
416	)
417	alerts_threats["threats"] = alerts_threats["threats"].str.strip("{}")	×
418	alerts_threats["threats"] = alerts_threats["threats"].str[:50]	×
419	alerts_threats = alerts_threats.rename(	×
420	columns={"site": "Site", "threats": "Threats"}
421	)
422	return alerts_threats	×
423
424	def dark_web_bad_actors(self):	5✔
425	"""Get dark web bad actors."""
426	dark_web_mentions = self.dark_web_mentions	×
427	dark_web_bad_actors = dark_web_mentions[["creator", "rep_grade"]]	×
428	dark_web_bad_actors = dark_web_bad_actors.groupby(	×
429	"creator", as_index=False
430	).max()
431	dark_web_bad_actors = dark_web_bad_actors.sort_values(	×
432	by=["rep_grade"], ascending=False
433	)[:10]
434	dark_web_bad_actors["rep_grade"] = (	×
435	dark_web_bad_actors["rep_grade"].astype(float).round(decimals=3)
436	)
437	dark_web_bad_actors = dark_web_bad_actors.rename(	×
438	columns={"creator": "Creator", "rep_grade": "Grade"}
439	)
440	return dark_web_bad_actors	×
441
442	def dark_web_content(self):	5✔
443	"""Get dark web categories."""
444	dark_web_mentions = self.dark_web_mentions	×
445	dark_web_content = dark_web_mentions[["category"]]	×
446	dark_web_content = (	×
447	dark_web_content.groupby(["category"])["category"]
448	.count()
449	.nlargest(10)
450	.reset_index(name="count")
451	)
452	return dark_web_content	×
453
454	def dark_web_count(self):	5✔
455	"""Get total number of dark web mentions."""
456	return len(self.dark_web_mentions.index)	×
457
458	def dark_web_date(self):	5✔
459	"""Get dark web mentions by date."""
460	dark_web_mentions = self.trending_dark_web_mentions	×
461	dark_web_date = dark_web_mentions[["date"]]	×
462	dark_web_date = (	×
463	dark_web_date.groupby(["date"])["date"].count().reset_index(name="Count")
464	)
465	dark_web_date["date"] = pd.to_datetime(dark_web_date["date"])	×
466	idx = pd.date_range(self.trending_start_date, self.end_date)	×
467	dark_web_date = (	×
468	dark_web_date.set_index("date").reindex(idx).fillna(0.0).rename_axis("date")
469	)
470
471	group_limit = self.end_date + datetime.timedelta(1)	×
472	dark_web_date = dark_web_date.groupby(	×
473	pd.Grouper( # lgtm [py/call/wrong-named-class-argument]
474	level="date", freq="7d", origin=group_limit
475	)
476	).sum()
477	dark_web_date["date"] = dark_web_date.index	×
478	dark_web_date["date"] = dark_web_date["date"].dt.strftime("%m/%d")	×
479	dark_web_date = dark_web_date.set_index("date")	×
480	dark_web_date = dark_web_date[["Count"]]	×
481	return dark_web_date	×
482
483	def dark_web_most_act(self):	5✔
484	"""Get most active posts."""
485	dark_web_mentions = self.dark_web_mentions	×
486	dark_web_most_act = dark_web_mentions[["title", "comments_count"]]	×
487	dark_web_most_act = dark_web_most_act[	×
488	dark_web_most_act["comments_count"] != "NaN"
489	]
490	dark_web_most_act = dark_web_most_act.rename(	×
491	columns={"comments_count": "Comments Count"}
492	)
493	dark_web_most_act["Comments Count"] = (	×
494	dark_web_most_act["Comments Count"].astype(float).astype(int)
495	)
496	dark_web_most_act = dark_web_most_act.sort_values(	×
497	by="Comments Count", ascending=False
498	)
499	dark_web_most_act = dark_web_most_act[:5]	×
500	dark_web_most_act = dark_web_most_act.rename(columns={"title": "Title"})	×
501	dark_web_most_act["Title"] = dark_web_most_act["Title"].str[:100]	×
502	dark_web_most_act = dark_web_most_act.replace(r"^\s*$", "Untitled", regex=True)	×
503	return dark_web_most_act	×
504
505	def dark_web_sites(self):	5✔
506	"""Get mentions by dark web sites (top 10)."""
507	dark_web_mentions = self.dark_web_mentions	×
508	dark_web_sites = dark_web_mentions[["site"]]	×
509	dark_web_sites = (	×
510	dark_web_sites.groupby(["site"])["site"]
511	.count()
512	.nlargest(10)
513	.reset_index(name="count")
514	)
515	dark_web_sites = dark_web_sites.rename(	×
516	columns={"site": "Site", "count": "Count"}
517	)
518	return dark_web_sites	×
519
520	def dark_web_tags(self):	5✔
521	"""Get dark web notable tags."""
522	dark_web_mentions = self.dark_web_mentions	×
523	dark_web_tags = dark_web_mentions[["tags"]]	×
524	dark_web_tags = dark_web_tags[dark_web_tags["tags"] != "NaN"]	×
525	dark_web_tags = (	×
526	dark_web_tags.groupby(["tags"])["tags"]
527	.count()
528	.nlargest(8)
529	.reset_index(name="Events")
530	)
531	dark_web_tags["tags"] = dark_web_tags["tags"].str.strip("{}")	×
532	dark_web_tags["tags"] = dark_web_tags["tags"].str.replace(	×
533	",", ", ", regex=False
534	)
535	dark_web_tags = dark_web_tags.rename(columns={"tags": "Tags"})	×
536	return dark_web_tags	×
537
538	def alerts_site(self):	5✔
539	"""Get alerts in invite-only markets."""
540	alerts_site = self.alerts[["site"]]	×
541	alerts_site = alerts_site[alerts_site["site"] != "NaN"]	×
542	alerts_site = alerts_site[alerts_site["site"] != ""]	×
543	alerts_site = alerts_site[alerts_site["site"].str.startswith("market")]	×
544	alerts_site = (	×
545	alerts_site.groupby(["site"])["site"]
546	.count()
547	.nlargest(10)
548	.reset_index(name="Alerts")
549	)
550	alerts_site = alerts_site.rename(columns={"site": "Site"})	×
551	return alerts_site	×
552
553	def top_cve_table(self):	5✔
554	"""Get top CVEs."""
555	top_cves = self.top_cves	×
556	top_cves["summary_short"] = top_cves["summary"].str[:400]	×
557	top_cve_table = top_cves[["cve_id", "summary_short"]]	×
558	top_cve_table = top_cve_table.rename(	×
559	columns={"cve_id": "CVE", "summary_short": "Description"}
560	)
561	return top_cve_table	×

cisagov / pe-reports / 3386784809

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