Merge pull request #59 from berylliumsec/skipping_updates_if_no_internet

Skipping updates if no internet

.github/workflows/publish.yml

@@ -131,7 +131,7 @@ jobs:
           prerelease: false
           body: |
             Release Notes:
-              - This release removes some unnecessary functions that were needed when the models werent performing so well but are now impacting thier performance.
+              - This release checks if the internet is avaiable, if it is not, it skips update checks
       - name: Upload Release Assets
         if: github.event.pull_request.merged == true
         run: |

src/nebula/nebula.py

@@ -667,6 +667,10 @@ def get_latest_pypi_version(package_name):
 
     def check_new_pypi_version(self, package_name="nebula-ai"):
         """Check if a newer version of the package is available on PyPI."""
+        if not self.is_internet_available():
+            cprint("No internet connection available. Skipping version check.", "red")
+            return
+
         installed_version = version(package_name)
         cprint(f"installed version: {installed_version}", "green")
         latest_version = self.get_latest_pypi_version(package_name)
@@ -737,7 +741,19 @@ def _determine_s3_url():
             else "https://nebula-models.s3.amazonaws.com/unified_models.zip"
         )
 
+    def is_internet_available(self, host="8.8.8.8", port=53, timeout=3):
+        """Check if there is an internet connection."""
+        try:
+            socket.setdefaulttimeout(timeout)
+            socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect((host, port))
+            return True
+        except Exception:
+            return False
+
     def get_s3_file_etag(self, s3_url):
+        if not self.is_internet_available():
+            cprint("No internet connection available. Skipping version check.", "red")
+            return False
         response = requests.head(s3_url)
         return response.headers.get("ETag")
 
@@ -757,11 +773,12 @@ def _ensure_model_folder_exists(self):
         metadata_file = "metadata.json"
         local_etag = self.get_local_metadata(metadata_file)
         s3_etag = self.get_s3_file_etag(self.s3_url)
-
+        if s3_etag is False:
+            return
         if not self.folder_exists_and_not_empty(self.args.model_dir) or (
             local_etag != s3_etag
         ):
-            if local_etag != s3_etag:
+            if local_etag and local_etag != s3_etag:
                 user_input = self.get_input_with_default(
                     "New versions of the models are available, would you like to download them? (y/n) "
                 )
@@ -788,7 +805,6 @@ def _ensure_model_folder_exists(self):
             )
 
     def _validate_model_dirs(self):
-
         while True:
             try:
                 model_dirs = [
@@ -912,7 +928,6 @@ def execute_command(command: Union[str, List[str]]) -> Tuple[int, str, str]:
             cprint("XML format detected, nothing to do", "green")
             return
         except Exception:
-
             # Conditions to decide if the results should be written to the file or not
             should_write_stderr = stderr and self.args.autonomous_mode is False
             if stdout.startswith("Starting Nmap"):
@@ -967,7 +982,6 @@ def ensure_space_between_letter_and_number(s: str) -> str:
                     s = ":".join(segments[i + 1 :])
                     break
 
-
             s = re.sub(r"\.$", "", s)
 
             return s.strip()
@@ -1009,7 +1023,6 @@ def process_string(
         replacement_urls: Optional[List[str]] = None,
         port_arg: Optional[int] = None,
     ) -> str:
-
         # Handle the default values
         if replacement_ips is None:
             replacement_ips = []
@@ -2135,7 +2148,6 @@ def generate_and_display_text(self, prompt: str) -> str:
             first_clean_up = self.ensure_space_between_letter_and_number(generated_text)
             second_clean_up = self.process_string(first_clean_up, prompt_ip, urls)
             if self.args.autonomous_mode is False:
-
                 try:
                     help = self.extract_and_match_flags(second_clean_up)
                     if help:
@@ -2220,7 +2232,6 @@ def user_search_interface(self):
         return True
 
     def get_suggestions(self):
-
         with open(self.suggestions_file, "r") as file:
             return [line.strip() for line in file if line.strip()]
 
@@ -2292,7 +2303,6 @@ def handle_result_selection_and_modification(self, other_lines, history):
                 )
 
                 if 1 <= selection <= len(other_lines):
-
                     self.modify_and_run_command(other_lines[selection - 1])
                     break
                 else:
@@ -2440,7 +2450,6 @@ def handle_generated_text(self, text):
 
             # Ask the user for their action choice
             if text.strip().endswith(".txt") and not text.strip().startswith("nmap"):
-
                 return
             else:
                 action_choice = self.get_action_choice()