deepsource

api_app/analyzers_manager/file_analyzers/floss.py

@@ -24,6 +24,10 @@ class Floss(FileAnalyzer, DockerBasedAnalyzer):
     max_no_of_strings: dict
     rank_strings: dict
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         # get binary
         binary = self.read_file_bytes()

api_app/analyzers_manager/file_analyzers/xlm_macro_deobfuscator.py

@@ -14,6 +14,10 @@
 class XlmMacroDeobfuscator(FileAnalyzer):
     passwords_to_check: list
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         results = {}
         try:

api_app/analyzers_manager/observable_analyzers/checkphish.py

@@ -19,6 +19,10 @@ class CheckPhish(classes.ObservableAnalyzer):
 
     _api_key_name: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         json_data = {
             "apiKey": self._api_key_name,

api_app/analyzers_manager/observable_analyzers/crowdsec.py

@@ -12,6 +12,10 @@ class Crowdsec(ObservableAnalyzer):
     _api_key_name: str
     url: str = "https://cti.api.crowdsec.net"
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         headers = {
             "x-api-key": self._api_key_name,

api_app/analyzers_manager/observable_analyzers/emailrep.py

@@ -13,6 +13,10 @@ class EmailRep(classes.ObservableAnalyzer):
 
     _api_key_name: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         """
         API key is not mandatory, emailrep supports requests with no key:

api_app/analyzers_manager/observable_analyzers/greynoiseintel.py

@@ -19,6 +19,10 @@ class GreyNoiseAnalyzer(classes.ObservableAnalyzer):
 
     _api_key_name: str = None
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     @property
     def integration_name(self):
         if self.greynoise_api_version == "v2":

api_app/analyzers_manager/observable_analyzers/ha_get.py

@@ -15,6 +15,10 @@ class HybridAnalysisGet(ObservableAnalyzer):
 
     _api_key_name: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         headers = {
             "api-key": self._api_key_name,

api_app/analyzers_manager/observable_analyzers/haveibeenpwned.py

@@ -15,6 +15,10 @@ class HaveIBeenPwned(classes.ObservableAnalyzer):
     domain: str
     _api_key_name: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         params = {
             "truncateResponse": self.truncate_response,

api_app/analyzers_manager/observable_analyzers/inquest.py

@@ -23,6 +23,10 @@ class InQuest(ObservableAnalyzer):
     _api_key_name: str
     inquest_analysis: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def config(self, runtime_configuration: Dict):
         super().config(runtime_configuration)
         self.generic_identifier_mode = "user-defined"  # Or auto

api_app/analyzers_manager/observable_analyzers/ip2location.py

@@ -13,6 +13,10 @@ class Ip2location(classes.ObservableAnalyzer):
     _api_key_name: str
     api_version: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def get_response(self, payload):
         return requests.get(self.url, params=payload)
 

api_app/analyzers_manager/observable_analyzers/koodous.py

@@ -13,6 +13,10 @@ class Koodous(classes.ObservableAnalyzer):
 
     _api_key_name: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def get_response(self, url):
         return requests.get(
             url, headers={"Authorization": f"Token {self._api_key_name}"}

api_app/analyzers_manager/observable_analyzers/mnemonic_pdns.py

@@ -15,6 +15,10 @@ class MnemonicPassiveDNS(classes.ObservableAnalyzer):
     cof_format: bool
     limit: int
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         if self.cof_format:
             self.url += "cof/"

api_app/analyzers_manager/observable_analyzers/phishstats.py

@@ -18,6 +18,10 @@ class PhishStats(ObservableAnalyzer):
 
     url: str = "https://phishstats.info:2096/api"
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def __build_phishstats_url(self) -> str:
         to_analyze_observable_classification = self.observable_classification
         to_analyze_observable_name = self.observable_name

api_app/analyzers_manager/observable_analyzers/robtex.py

@@ -14,6 +14,10 @@
 class Robtex(classes.ObservableAnalyzer):
     url = "https://freeapi.robtex.com/"
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         if self.observable_classification == self.ObservableTypes.IP:
             uris = [

api_app/analyzers_manager/observable_analyzers/spyse.py

@@ -16,6 +16,10 @@ class Spyse(classes.ObservableAnalyzer):
 
     _api_key_name: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def __build_spyse_api_uri(self) -> str:
         if self.observable_classification == self.ObservableTypes.DOMAIN:
             endpoint = "domain"

api_app/analyzers_manager/observable_analyzers/urlscan.py

@@ -21,6 +21,10 @@ class UrlScan(ObservableAnalyzer):
     search_size: int
     _api_key_name: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         headers = {"Content-Type": "application/json", "User-Agent": "IntelOwl/v1.x"}
         if not hasattr(self, "_api_key_name") and self.urlscan_analysis == "search":

api_app/analyzers_manager/observable_analyzers/xforce.py

@@ -20,6 +20,10 @@ class XForce(classes.ObservableAnalyzer):
     malware_only: bool
     timeout: int = 5
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def run(self):
         auth = HTTPBasicAuth(self._api_key_name, self._api_password_name)
         headers = {"Accept": "application/json"}

api_app/pivots_manager/pivots/compare.py

@@ -6,6 +6,10 @@
 class Compare(Pivot):
     field_to_compare: str
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     def _get_value(self, field: str) -> Any:
         report = self.related_reports.filter(
             status=self.report_model.Status.SUCCESS.value

api_app/visualizers_manager/visualizers/domain_reputation_services.py

@@ -15,6 +15,10 @@
 
 
 class DomainReputationServices(Visualizer):
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     @visualizable_error_handler_with_params("VirusTotal")
     def _vt3(self):
         try:

api_app/visualizers_manager/visualizers/yara.py

@@ -15,6 +15,10 @@
 
 
 class Yara(Visualizer):
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
     @visualizable_error_handler_with_params("Analyzer")
     def _yara_analyzer(self):
         return self.Title(

tests/api_app/analyzers_manager/test_classes.py

@@ -15,6 +15,10 @@ class MockUpObservableAnalyzer(ObservableAnalyzer):
     def run(self) -> dict:
         return {}
 
+    @classmethod
+    def update(cls) -> bool:
+        pass
+
 
 class FileAnalyzerTestCase(CustomTestCase):
     fixtures = [

tests/test_crons.py

@@ -1,6 +1,5 @@
 # This file is a part of IntelOwl https://github.com/intelowlproject/IntelOwl
 # See the file 'LICENSE' for copying permission.
-import datetime
 import os
 
 from django.conf import settings
@@ -30,6 +29,8 @@
 
 class CronTests(CustomTestCase):
     def test_check_stuck_analysis(self):
+        import datetime
+
         _job = Job.objects.create(
             user=self.user,
             status=Job.Status.RUNNING.value,
@@ -54,6 +55,8 @@ def test_check_stuck_analysis(self):
         _job.delete()
 
     def test_remove_old_jobs(self):
+        import datetime
+
         _job = Job.objects.create(
             user=self.user,
             status=Job.Status.FAILED.value,