2302119809

Build # 2302119809

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #192 from cisagov/DJ_add_trending

Run Details

18 of 143 branches covered (12.59%)

Branch coverage included in aggregate %.

10 of 100 new or added lines in 5 files covered. (10.0%)

9 existing lines in 2 files now uncovered.

142 of 778 relevant lines covered (18.25%)

0.91 hits per line

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

17.09

/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_creds_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)

    def by_days(self):
        """Return number of credentials by day."""
        df = self.trending_creds_view
        df = df[["modified_date", "password_included", "email"]].copy()
        df.loc[:, "modified_date"] = pd.to_datetime(df["modified_date"]).dt.date
        df = df.groupby(["modified_date", "password_included"], as_index=False).agg(
            {"email": ["count"]}
        )
        idx = pd.date_range(self.trending_start_date, self.end_date)
        df.columns = df.columns.droplevel(1)
        df = (
            df.pivot(index="modified_date", columns="password_included", values="email")
            .fillna(0)
            .reset_index()
            .rename_axis(None)
        )
        df.columns.name = None
        df = (
            df.set_index("modified_date")
            .reindex(idx)
            .fillna(0.0)
            .rename_axis("added_date")
        )
        group_limit = self.end_date + datetime.timedelta(1)
        df = df.groupby(
            pd.Grouper(  # lgtm [py/call/wrong-named-class-argument]
                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={True: "Passwords Included", False: "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."""
        view_df = self.creds_view
        view_df = view_df[
            [
                "breach_name",
                "modified_date",
                "description",
                "breach_date",
                "password_included",
                "email",
            ]
        ]
        breach_df = view_df.groupby(
            [
                "breach_name",
                "modified_date",
                "description",
                "breach_date",
                "password_included",
            ],
            as_index=False,
        ).agg({"email": ["count"]})
        breach_df.columns = breach_df.columns.droplevel(1)
        breach_df = breach_df.rename(columns={"email": "number_of_creds"})
        breach_df = breach_df[
            [
                "breach_name",
                "breach_date",
                "modified_date",
                "password_included",
                "number_of_creds",
            ]
        ]
        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,
            "shodan_insecure_protocols_unverified_vulns",
        )
        self.insecure_df = insecure_df

        vulns_df = query_shodan(org_uid, start_date, end_date, "shodan_verified_vulns")
        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")
        ]
        return insecure[["ip", "protocol"]].drop_duplicates(keep="first")

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

cisagov / pe-reports / 2302119809

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