Fixed Linting Issues

msticnb/nb/azsent/account/account_summary.py

@@ -55,8 +55,12 @@ class AccountType(Flag):
     Office365 = auto()
     Windows = auto()
     Linux = auto()
-    Azure = AzureActiveDirectory | AzureActivity | Office365
-    All = Azure | Windows | Linux
+    Azure = (
+        AzureActiveDirectory
+        | AzureActivity
+        | Office365  # pylint:disable=unsupported-binary-operation
+    )
+    All = Azure | Windows | Linux  # pylint:disable=unsupported-binary-operation
 
     def in_list(self, acct_types: Iterable[Union["AccountType", str]]):
         """Is the current value in the `acct_types` list."""
@@ -137,20 +141,20 @@ def __init__(
         super().__init__(description, timespan, notebooklet)
         self.description: str = "Account Activity Summary"
         self.account_entity: entities.Account = None
-        self.account_activity: pd.DataFrame = None
+        self.account_activity: Optional[pd.DataFrame] = None
         self.account_selector: nbwidgets.SelectItem = None
-        self.related_alerts: pd.DataFrame = None
+        self.related_alerts: Optional[pd.DataFrame] = None
         self.alert_timeline: LayoutDOM = None
-        self.related_bookmarks: pd.DataFrame = None
-        self.host_logons: pd.DataFrame = None
-        self.host_logon_summary: pd.DataFrame = None
-        self.azure_activity: pd.DataFrame = None
-        self.azure_activity_summary: pd.DataFrame = None
+        self.related_bookmarks: Optional[pd.DataFrame] = None
+        self.host_logons: Optional[pd.DataFrame] = None
+        self.host_logon_summary: Optional[pd.DataFrame] = None
+        self.azure_activity: Optional[pd.DataFrame] = None
+        self.azure_activity_summary: Optional[pd.DataFrame] = None
         self.azure_timeline_by_provider: LayoutDOM = None
         self.account_timeline_by_ip: LayoutDOM = None
         self.azure_timeline_by_operation: LayoutDOM = None
-        self.ip_summary: pd.DataFrame = None
-        self.ip_all_data: pd.DataFrame = None
+        self.ip_summary: Optional[pd.DataFrame] = None
+        self.ip_all_data: Optional[pd.DataFrame] = None
 
 
 # pylint: enable=too-few-public-methods
@@ -418,22 +422,14 @@ def get_geoip_map(self):
         return None
 
     @set_text(docs=_CELL_DOCS, key="find_additional_data")
-    def get_additional_data(self) -> pd.DataFrame:
-        """
-        Find additional data for the selected account.
-
-        Returns
-        -------
-        pd.DataFrame
-            Results with expanded columns.
-
-        """
+    def get_additional_data(self):
+        """Find additional data for the selected account."""
         if not self.check_valid_result_data():
-            return
+            return None
         acct, source = self._get_selected_account()
         if not acct or not source:
             print("Please use select an account before using this method.")
-            return
+            return None
         self._last_result.host_logons = None
         self._last_result.host_logon_summary = None
         self._last_result.account_timeline_by_ip = None
@@ -464,6 +460,7 @@ def get_additional_data(self) -> pd.DataFrame:
                 geoip=self._geo_lookup,
             )
             nb_display(self._last_result.ip_summary)
+            return None
         if acct_type == AccountType.Windows:
             self._last_result.host_logons = _get_windows_add_activity(
                 self.query_provider, acct, self.timespan
@@ -483,6 +480,7 @@ def get_additional_data(self) -> pd.DataFrame:
                 geoip=self._geo_lookup,
             )
             nb_display(self._last_result.ip_summary)
+            return None
         if acct_type in [
             AccountType.AzureActiveDirectory,
             AccountType.AzureActivity,
@@ -508,6 +506,8 @@ def get_additional_data(self) -> pd.DataFrame:
                 geoip=self._geo_lookup,
             )
             nb_display(self._last_result.ip_summary)
+            return None
+        return None
 
     def _get_selected_account(self):
         if (

msticnb/nb/azsent/alert/ti_enrich.py

@@ -83,8 +83,8 @@ def __init__(
         """
         super().__init__(description, timespan, notebooklet)
         self.description: str = "Enriched Alerts"
-        self.enriched_results: pd.DataFrame = None
-        self.picker: SelectAlert = None
+        self.enriched_results: Optional[pd.DataFrame] = None
+        self.picker: Optional[SelectAlert] = None
 
 
 # pylint: enable=too-few-public-methods

msticnb/nb/azsent/host/host_logons_summary.py

@@ -164,7 +164,7 @@ def run(  # noqa:MC0001
             raise MsticnbMissingParameterError("data, or a hostname and timespan.")
 
         # If data is not provided use host_name and timespan to get data
-        if data is None and timespan is not None:
+        if not isinstance(data, pd.DataFrame) or data.empty and timespan:
             nb_data_wait(f"{value}")
             host_verif = verify_host_name(
                 qry_prov=self.query_provider, timespan=timespan, host_name=value
@@ -184,11 +184,11 @@ def run(  # noqa:MC0001
             if host_type == "Windows" or not host_type == "Linux":
                 # If no known data type try Windows
                 data = self.query_provider.WindowsSecurity.list_all_logons_by_host(  # type: ignore
-                    host_name=host_name, start=timespan.start, end=timespan.end
+                    host_name=host_name, start=timespan.start, end=timespan.end  # type: ignore
                 )
             else:
                 data = self.query_provider.LinuxSyslog.list_logons_for_host(  # type: ignore
-                    host_name=host_name, start=timespan.start, end=timespan.end
+                    host_name=host_name, start=timespan.start, end=timespan.end  # type: ignore
                 )
         else:
             # If data is provided do some required formatting
@@ -266,7 +266,7 @@ def _gen_timeline(data: pd.DataFrame, silent: bool):
 @set_text(docs=_CELL_DOCS, key="show_map")
 def _map_logons(data: pd.DataFrame, silent: bool) -> FoliumMap:
     """Produce a map of source IP logon locations."""
-    map_data = data[data["IpAddress"].isin(["-", "::1", "", "NaN"]) is False]
+    map_data = data[data["IpAddress"].isin(["-", "::1", "", "NaN"]) == False]  # noqa: E712
     if not isinstance(map_data, pd.DataFrame) or map_data.empty:
         if not silent:
             md("No plotable logins avaliable")
@@ -365,7 +365,7 @@ def _process_stack_bar(data: pd.DataFrame, silent: bool) -> figure:
     results = ["Success", "Failure"]
     colors = ["#536d4c", "#832828"]
 
-    data = {"processes": procs, "Success": s_data, "Failure": f_data}
+    graph_data = {"processes": procs, "Success": s_data, "Failure": f_data}
 
     viz = figure(
         x_range=processes,
@@ -381,7 +381,7 @@ def _process_stack_bar(data: pd.DataFrame, silent: bool) -> figure:
         x="processes",
         width=0.75,
         color=colors,
-        source=data,
+        source=graph_data,
         legend_label=results,
     )
 
@@ -407,8 +407,9 @@ def _process_stack_bar(data: pd.DataFrame, silent: bool) -> figure:
 @set_text(docs=_CELL_DOCS, key="logon_matrix")
 def _logon_matrix(data: pd.DataFrame, silent: bool) -> pd.DataFrame:
     """Produce DataFrame showing logons grouped by user and process."""
+    print(data.columns)
     logon_by_type = (
-        data[(data["Account"] != "") & (data["LogonResult"] != "Unknown")][
+        data[(data["Account"] != "") & (data["LogonResult"] != "Unknown")][  # type: ignore
             ["Account", "LogonTypeName", "LogonResult", "TimeGenerated"]
         ]
         .groupby(["Account", "LogonTypeName", "LogonResult"])
@@ -467,7 +468,7 @@ def _format_raw_data(data: pd.DataFrame) -> pd.DataFrame:
 def _get_logon_result_lx(row: pd.Series) -> str:
     """Identify if a Linux syslog event is for a sucessful or failed logon."""
     failure_events = row.str.contains(
-        "failure|failed|invalid|unable to negotiate|authentication failures|did not receive identification|bad protocol version identification|^Connection closed .* [preauth]",
+        """failure|failed|invalid|unable to negotiate|authentication failures|did not receive identification|bad protocol version identification|^Connection closed .* [preauth]""",  # pylint: disable=line-too-long
         regex=True,
     )
 

msticnb/nb/azsent/host/host_network_summary.py

@@ -30,7 +30,6 @@
     set_text,
 )
 from ....nb_metadata import read_mod_metadata, update_class_doc
-from ....nblib.iptools import map_ips
 from ....nblib.ti import get_ti_results
 from ....notebooklet import NBMetadata, Notebooklet, NotebookletResult
 

msticnb/nb/azsent/host/host_summary.py

@@ -94,16 +94,16 @@ def __init__(
 
         """
         super().__init__(description, timespan, notebooklet)
-        self.host_entity: entities.Host = None  # type: ignore
-        self.related_alerts: pd.DataFrame = None  # type: ignore
-        self.alert_timeline: Union[LayoutDOM, Figure] = None  # type: ignore
-        self.related_bookmarks: pd.DataFrame = None  # type: ignore
-        self.summary: pd.DataFrame = None  # type: ignore
-        self.scheduled_tasks: pd.DataFrame = None  # type: ignore
-        self.account_actions: pd.DataFrame = None  # type: ignore
-        self.notable_events: pd.DataFrame = None  # type: ignore
-        self.processes: pd.DataFrame = None  # type: ignore
-        self.process_ti: pd.DataFrame = None  # type: ignore
+        self.host_entity: entities.Host = None
+        self.related_alerts: Optional[pd.DataFrame] = None
+        self.alert_timeline: Union[LayoutDOM, Figure] = None
+        self.related_bookmarks: Optional[pd.DataFrame] = None
+        self.summary: Optional[pd.DataFrame] = None
+        self.scheduled_tasks: Optional[pd.DataFrame] = None
+        self.account_actions: Optional[pd.DataFrame] = None
+        self.notable_events: Optional[pd.DataFrame] = None
+        self.processes: Optional[pd.DataFrame] = None
+        self.process_ti: Optional[pd.DataFrame] = None
 
 
 # pylint: disable=too-few-public-methods
@@ -325,15 +325,15 @@ def display_alert_timeline(self):
         return None
 
 
-def _process_ti(data, col, ti_prov) -> pd.DataFrame:
+def _process_ti(data, col, ti_prov) -> Optional[pd.DataFrame]:
     extracted_iocs = extract_iocs(data, col, True)
     _, ti_merged_df = get_ti_results(ti_lookup=ti_prov, data=extracted_iocs, col="IoC")
     return ti_merged_df
 
 
 @lru_cache()
 def _get_process_events(qry_prov, timespan, host_name, os_family) -> pd.DataFrame:
-    process_events = pd.DataFrame
+    process_events = pd.DataFrame()
     if os_family.name == "Windows":
         nb_data_wait("Process Events")
         process_events = qry_prov.WindowsSecurity.list_host_processes(
@@ -353,7 +353,7 @@ def _get_process_events(qry_prov, timespan, host_name, os_family) -> pd.DataFram
 
 @lru_cache()
 def _get_host_event_summary(qry_prov, timespan, host_name, os_family) -> pd.DataFrame:
-    host_events = pd.DataFrame
+    host_events = pd.DataFrame()
     if os_family.name == "Windows":
         nb_data_wait("Events")
         host_events = qry_prov.WindowsSecurity.summarize_events(

msticnb/nb/azsent/host/logon_session_rarity.py

@@ -83,9 +83,9 @@ def __init__(
         # Add attributes as needed here.
         # Make sure they are documented in the Attributes section
         # above.
-        self.process_clusters: pd.DataFrame = None
-        self.processes_with_cluster: pd.DataFrame = None
-        self.session_rarity: pd.DataFrame = None
+        self.process_clusters: Optional[pd.DataFrame] = None
+        self.processes_with_cluster: Optional[pd.DataFrame] = None
+        self.session_rarity: Optional[pd.DataFrame] = None
 
 
 # pylint: enable=too-few-public-methods

msticnb/nb/azsent/host/win_host_events.py

@@ -91,12 +91,12 @@ def __init__(
         """
         super().__init__(description, timespan, notebooklet)
         self.description: str = "Windows Host Security Events"
-        self.all_events: pd.DataFrame = None
-        self.event_pivot: pd.DataFrame = None
-        self.account_events: pd.DataFrame = None
-        self.account_pivot: pd.DataFrame = None
+        self.all_events: Optional[pd.DataFrame] = None
+        self.event_pivot: Optional[pd.DataFrame] = None
+        self.account_events: Optional[pd.DataFrame] = None
+        self.account_pivot: Optional[pd.DataFrame] = None
         self.account_timeline: Union[Figure, LayoutDOM] = None
-        self.expanded_events: pd.DataFrame = None
+        self.expanded_events: Optional[pd.DataFrame] = None
 
 
 class WinHostEvents(Notebooklet):
@@ -209,7 +209,7 @@ def run(
 
     def expand_events(
         self, event_ids: Optional[Union[int, Iterable[int]]] = None
-    ) -> pd.DataFrame:
+    ) -> Optional[pd.DataFrame]:
         """
         Expand `EventData` for `event_ids` into separate columns.
 

msticnb/nb/azsent/network/ip_summary.py

@@ -157,29 +157,29 @@ def __init__(
         self.host_entities: List[Host] = []
         self.geoip: Optional[Dict[str, Any]] = None
         self.location: Optional[GeoLocation] = None
-        self.whois: pd.DataFrame = None
-        self.whois_nets: pd.DataFrame = None
-        self.heartbeat: pd.DataFrame = None
-        self.az_network_if: pd.DataFrame = None
-        self.vmcomputer: pd.DataFrame = None
-        self.az_network_flows: pd.DataFrame = None
-        self.az_network_flow_summary: pd.DataFrame = None
+        self.whois: Optional[pd.DataFrame] = None
+        self.whois_nets: Optional[pd.DataFrame] = None
+        self.heartbeat: Optional[pd.DataFrame] = None
+        self.az_network_if: Optional[pd.DataFrame] = None
+        self.vmcomputer: Optional[pd.DataFrame] = None
+        self.az_network_flows: Optional[pd.DataFrame] = None
+        self.az_network_flow_summary: Optional[pd.DataFrame] = None
         self.az_network_flows_timeline: Figure = None
-        self.aad_signins: pd.DataFrame = None
-        self.azure_activity: pd.DataFrame = None
-        self.azure_activity_summary: pd.DataFrame = None
-        self.office_activity: pd.DataFrame = None
-        self.common_security: pd.DataFrame = None
-        self.related_alerts: pd.DataFrame = None
-        self.related_bookmarks: pd.DataFrame = None
+        self.aad_signins: Optional[pd.DataFrame] = None
+        self.azure_activity: Optional[pd.DataFrame] = None
+        self.azure_activity_summary: Optional[pd.DataFrame] = None
+        self.office_activity: Optional[pd.DataFrame] = None
+        self.common_security: Optional[pd.DataFrame] = None
+        self.related_alerts: Optional[pd.DataFrame] = None
+        self.related_bookmarks: Optional[pd.DataFrame] = None
         self.alert_timeline: Figure = None
-        self.ti_results: pd.DataFrame = None
-        self.passive_dns: pd.DataFrame = None
-        self.host_logons: pd.DataFrame = None
-        self.related_accounts: pd.DataFrame = None
-        self.associated_hosts: pd.DataFrame = None
-        self.device_info: pd.DataFrame = None
-        self.network_connections: pd.DataFrame = None
+        self.ti_results: Optional[pd.DataFrame] = None
+        self.passive_dns: Optional[pd.DataFrame] = None
+        self.host_logons: Optional[pd.DataFrame] = None
+        self.related_accounts: Optional[pd.DataFrame] = None
+        self.associated_hosts: Optional[pd.DataFrame] = None
+        self.device_info: Optional[pd.DataFrame] = None
+        self.network_connections: Optional[pd.DataFrame] = None
 
 
 # pylint: enable=too-few-public-methods, too-many-instance-attributes
@@ -298,7 +298,9 @@ def run(  # noqa: MC0001
         if "alerts" in self.options:
             self._get_related_alerts(src_ip=value, result=result, timespan=timespan)
         if "bookmarks" in self.options:
-            self._get_related_bookmarks(src_ip=value, result=result, timespan=timespan)
+            result = self._get_related_bookmarks(
+                src_ip=value, result=result, timespan=timespan
+            )
         # Azure NSG netflow
         if "az_netflow" in self.options:
             self._get_azure_netflow(src_ip=value, result=result, timespan=timespan)
@@ -501,10 +503,13 @@ def _get_related_bookmarks(
         self, src_ip, result, timespan: TimeSpan
     ) -> pd.DataFrame:
         nb_data_wait("Bookmarks")
-        result.related_bookmarks = self.query_provider.AzureSentinel.list_bookmarks_for_entity(  # type: ignore
-            timespan, entity_id=src_ip
+        result.related_bookmarks = (
+            self.query_provider.AzureSentinel.list_bookmarks_for_entity(  # type: ignore
+                timespan, entity_id=src_ip
+            )
         )
         _display_df_summary(result.related_bookmarks, "related bookmarks")
+        return result
 
 
 # %%