classBaseModule:"""The base class for all BBOT modules. Attributes:
@@ -2957,7 +2976,7 @@
try:awaitself.ping()self.hugesuccess(f"API is ready")
- returnTrue
+ returnTrue,""exceptExceptionase:self.trace(traceback.format_exc())returnNone,f"Error with API ({str(e).strip()})"
@@ -2976,7 +2995,7 @@
defapi_failure_abort_threshold(self):return(self.api_retries*self._api_failure_abort_threshold)+1
- asyncdefping(self):
+ asyncdefping(self,url=None):"""Asynchronously checks the health of the configured API.
- This method is used in conjunction with require_api_key() to verify that the API is not just configured, but also responsive. This method should include an assert statement to validate the API's health, typically by making a test request to a known endpoint.
+ This method is used in conjunction with require_api_key() to verify that the API is not just configured, but also responsive. It makes a test request to a known endpoint to validate the API's health.
- Example Usage:
- In your implementation, if the API has a "/ping" endpoint:
- async def ping(self):
- r = await self.api_request(f"{self.base_url}/ping")
- resp_content = getattr(r, "text", "")
- assert getattr(r, "status_code", 0) == 200, resp_content
+ The method uses the `ping_url` attribute if defined, or falls back to a provided URL. If neither is available, no request is made.
+
+ Args:
+ url (str, optional): A specific URL to use for the ping request. If not provided, the method will use the `ping_url` attribute. Returns: None Raises:
- AssertionError: If the API does not respond as expected.
+ ValueError: If the API response is not successful (status code != 200).
+
+ Example Usage:
+ To use this method, simply define the `ping_url` attribute in your module:
+
+ class MyModule(BaseModule):
+ ping_url = "https://api.example.com/ping"
+
+ Alternatively, you can override this method for more complex health checks:
+
+ async def ping(self):
+ r = await self.api_request(f"{self.base_url}/complex-health-check")
+ if r.status_code != 200 or r.json().get('status') != 'healthy':
+ raise ValueError(f"API unhealthy: {r.text}") """
- return
+ ifurlisNone:
+ url=getattr(self,"ping_url","")
+ ifurl:
+ r=awaitself.api_request(url)
+ ifgetattr(r,"status_code",0)!=200:
+ response_text=getattr(r,"text","no response from server")
+ raiseValueError(response_text)@propertydefbatch_size(self):
@@ -3779,6 +3815,8 @@
self._api_request_failures=0else:status_code=getattr(r,"status_code",0)
+ response_text=getattr(r,"text","")
+ self.trace(f"API response to {url} failed with status code {status_code}: {response_text}")self._api_request_failures+=1ifself._api_request_failures>=self.api_failure_abort_threshold:self.set_error_state(
@@ -4600,26 +4638,7 @@
asyncdefapi_page_iter(self,url,page_size=100,json=True,next_key=None,**requests_kwargs):""" An asynchronous generator function for iterating through paginated API data.
@@ -4741,24 +4779,7 @@
defcritical(self,*args,trace=True,**kwargs):"""Logs a whole message in emboldened red text, and optionally the stack trace of the most recent exception. Args:
@@ -4997,21 +5039,21 @@
defdebug(self,*args,trace=False,**kwargs):"""Logs debug messages and optionally the stack trace of the most recent exception. Args:
@@ -5092,24 +5134,7 @@
asyncdefemit_event(self,*args,**kwargs):"""Emit an event to the event queue and distribute it to interested modules. This is how modules "return" data.
@@ -5214,21 +5256,21 @@
deferror(self,*args,trace=True,**kwargs):"""Logs an error message, and optionally the stack trace of the most recent exception. Args:
@@ -5403,24 +5445,24 @@
defget_per_domain_hash(self,event):""" Computes a per-domain hash value for a given event. This method may be optionally overridden in subclasses.
@@ -5477,23 +5519,23 @@
defget_per_host_hash(self,event):""" Computes a per-host hash value for a given event. This method may be optionally overridden in subclasses.
@@ -5550,29 +5592,29 @@
defget_per_hostport_hash(self,event):""" Computes a per-host:port hash value for a given event. This method may be optionally overridden in subclasses.
@@ -5619,17 +5661,17 @@
defget_watched_events(self):"""Retrieve the set of events that the module is interested in observing. Override this method if the set of events the module should watch needs to be determined dynamically, e.g., based on configuration options or other runtime conditions.
@@ -5829,21 +5871,21 @@
defhugeinfo(self,*args,trace=False,**kwargs):"""Logs a whole message in emboldened blue text, and optionally the stack trace of the most recent exception. Args:
@@ -5903,21 +5945,21 @@
defhugesuccess(self,*args,trace=False,**kwargs):"""Logs a whole message in emboldened green text, and optionally the stack trace of the most recent exception. Args:
@@ -5977,21 +6019,21 @@
defhugeverbose(self,*args,trace=False,**kwargs):"""Logs a whole message in emboldened white text, and optionally the stack trace of the most recent exception. Args:
@@ -6051,21 +6093,21 @@
defhugewarning(self,*args,trace=True,**kwargs):"""Logs a whole message in emboldened orange text, and optionally the stack trace of the most recent exception. Args:
@@ -6125,21 +6167,21 @@
definfo(self,*args,trace=False,**kwargs):"""Logs informational messages and optionally the stack trace of the most recent exception. Args:
@@ -6199,26 +6241,7 @@
deflog_table(self,*args,**kwargs):"""Logs a table to the console and optionally writes it to a file. This function generates a table using `self.helpers.make_table`, then logs each line
@@ -6325,24 +6367,7 @@
defmake_event(self,*args,**kwargs):"""Create an event for the scan. Raises a validation error if the event could not be created, unless raise_error is set to False.
@@ -6401,19 +6443,25 @@
async
-
ping()
+
ping(url=None)
Asynchronously checks the health of the configured API.
-
This method is used in conjunction with require_api_key() to verify that the API is not just configured, but also responsive. This method should include an assert statement to validate the API's health, typically by making a test request to a known endpoint.
-
-Example Usage
-
In your implementation, if the API has a "/ping" endpoint:
-async def ping(self):
- r = await self.api_request(f"{self.base_url}/ping")
- resp_content = getattr(r, "text", "")
- assert getattr(r, "status_code", 0) == 200, resp_content
-
+
This method is used in conjunction with require_api_key() to verify that the API is not just configured, but also responsive. It makes a test request to a known endpoint to validate the API's health.
+
The method uses the ping_url attribute if defined, or falls back to a provided URL. If neither is available, no request is made.
asyncdefping(self,url=None):"""Asynchronously checks the health of the configured API.
- This method is used in conjunction with require_api_key() to verify that the API is not just configured, but also responsive. This method should include an assert statement to validate the API's health, typically by making a test request to a known endpoint.
+ This method is used in conjunction with require_api_key() to verify that the API is not just configured, but also responsive. It makes a test request to a known endpoint to validate the API's health.
- Example Usage:
- In your implementation, if the API has a "/ping" endpoint:
- async def ping(self):
- r = await self.api_request(f"{self.base_url}/ping")
- resp_content = getattr(r, "text", "")
- assert getattr(r, "status_code", 0) == 200, resp_content
+ The method uses the `ping_url` attribute if defined, or falls back to a provided URL. If neither is available, no request is made.
+
+ Args:
+ url (str, optional): A specific URL to use for the ping request. If not provided, the method will use the `ping_url` attribute. Returns: None Raises:
- AssertionError: If the API does not respond as expected.
+ ValueError: If the API response is not successful (status code != 200).
+
+ Example Usage:
+ To use this method, simply define the `ping_url` attribute in your module:
+
+ class MyModule(BaseModule):
+ ping_url = "https://api.example.com/ping"
+
+ Alternatively, you can override this method for more complex health checks:
+
+ async def ping(self):
+ r = await self.api_request(f"{self.base_url}/complex-health-check")
+ if r.status_code != 200 or r.json().get('status') != 'healthy':
+ raise ValueError(f"API unhealthy: {r.text}") """
- return
+ ifurlisNone:
+ url=getattr(self,"ping_url","")
+ ifurl:
+ r=awaitself.api_request(url)
+ ifgetattr(r,"status_code",0)!=200:
+ response_text=getattr(r,"text","no response from server")
+ raiseValueError(response_text)
defprepare_api_request(self,url,kwargs):""" Prepare an API request by adding the necessary authentication - header, bearer token, etc. """
@@ -6556,24 +6649,7 @@