Merge pull request #157 from boeronline/main

Added user_agent setting

Author: SimOnPanw

prismacloud/api/cspm/cspm.py

@@ -20,6 +20,8 @@ def login(self, url=None):
             url = 'https://%s/login' % self.api
         action = 'POST'
         request_headers = {'Content-Type': 'application/json'}
+        # Add User-Agent to the headers
+        request_headers['User-Agent'] = self.user_agent
         body_params_json = json.dumps({'username': self.identity, 'password': self.secret})
         # try:
         #     api_response = requests.request(action, url, headers=request_headers, data=body_params_json, verify=self.verify, timeout=self.timeout)
@@ -48,6 +50,8 @@ def extend_login(self):
         url = 'https://%s/auth_token/extend' % self.api
         action = 'GET'
         request_headers = {'Content-Type': 'application/json', 'x-redlock-auth': self.token}
+        # Add User-Agent to the headers
+        request_headers['User-Agent'] = self.user_agent
         api_response = requests.request(action, url, headers=request_headers, verify=self.verify, timeout=self.timeout)
         if api_response.status_code in self.retry_status_codes:
             for exponential_wait in self.retry_waits:
@@ -91,6 +95,8 @@ def execute(self, action, endpoint, query_params=None, body_params=None, request
             self.debug_print('API Request Headers: (%s)' % request_headers)
             self.debug_print('API Query Params: %s' % query_params)
             self.debug_print('API Body Params: %s' % body_params_json)
+            # Add User-Agent to the headers
+            request_headers['User-Agent'] = self.user_agent
             api_response = requests.request(action, url, headers=request_headers, params=query_params, data=body_params_json, verify=self.verify, timeout=self.timeout)
             self.debug_print('API Response Status Code: %s' % api_response.status_code)
             self.debug_print('API Response Headers: (%s)' % api_response.headers)

prismacloud/api/cwpp/cwpp.py

@@ -35,6 +35,8 @@ def execute_compute(self, action, endpoint, query_params=None, body_params=None,
             body_params_json = json.dumps(body_params)
         else:
             body_params_json = None
+        # Set User Agent
+        request_headers['User-Agent'] = "W"
         # Endpoints that return large numbers of results use a 'Total-Count' response header.
         # Pagination is via query parameters for both GET and POST, and the limit has a maximum of 50.
         offset = 0
@@ -59,6 +61,8 @@ def execute_compute(self, action, endpoint, query_params=None, body_params=None,
             self.debug_print('API Request Headers: (%s)' % request_headers)
             self.debug_print('API Query Params: %s' % query_params)
             self.debug_print('API Body Params: %s' % body_params_json)
+            # Add User-Agent to the headers
+            request_headers['User-Agent'] = self.user_agent
             api_response = requests.request(action, url, headers=request_headers, params=query_params, data=body_params_json, verify=self.verify, timeout=self.timeout)
             self.debug_print('API Response Status Code: (%s)' % api_response.status_code)
             self.debug_print('API Response Headers: (%s)' % api_response.headers)

prismacloud/api/pc_lib_api.py

@@ -7,6 +7,7 @@
 from .pccs import PrismaCloudAPIPCCS
 
 from .pc_lib_utility import PrismaCloudUtility
+from .version import version  # Import version from your version.py
 
 # --Description-- #
 
@@ -46,6 +47,9 @@ def __init__(self):
         #
         self.error_log          = 'error.log'
         self.logger             = None
+        # Set User-Agent
+        default_user_agent = f"PrismaCloudAPI/{version}"  # Dynamically set default User-Agent
+        self.user_agent = default_user_agent
 
     def __repr__(self):
         return 'Prisma Cloud API:\n  API: (%s)\n  Compute API: (%s)\n  API Error Count: (%s)\n  API Token: (%s)' % (self.api, self.api_compute, self.logger.error.counter, self.token)
@@ -56,6 +60,8 @@ def configure(self, settings, use_meta_info=True):
         self.secret      = settings.get('secret')
         self.verify      = settings.get('verify', True)
         self.debug       = settings.get('debug', False)
+        self.user_agent  = settings.get('user_agent', self.user_agent)
+        print(self.user_agent)
         #
         # self.logger      = settings['logger']
         self.logger = logging.getLogger(__name__)

prismacloud/api/pccs/pccs.py

@@ -40,6 +40,8 @@ def execute_code_security(self, action, endpoint, query_params=None, body_params
             self.debug_print('API Headers: %s' % request_headers)
             self.debug_print('API Query Params: %s' % query_params)
             self.debug_print('API Body Params: %s' % body_params_json)
+            # Add User-Agent to the headers
+            request_headers['User-Agent'] = self.user_agent
             api_response = requests.request(action, url, headers=request_headers, params=query_params, data=body_params_json, verify=self.verify, timeout=self.timeout)
             self.debug_print('API Response Status Code: %s' % api_response.status_code)
             self.debug_print('API Response Headers: (%s)' % api_response.headers)

prismacloud/api/version.py

@@ -1,3 +1,3 @@
 """ version file """
 
-version = '5.2.5'
+version = '5.2.6'

scripts/pc_compute_cloud_discovery_example_standalone.py

@@ -40,6 +40,9 @@
 api_url = 'https://' + PRISMA_CLOUD_COMPUTE_CONSOLE_URL + PRISMA_CLOUD_COMPUTE_CONSOLE_API_VERSION + '/credentials'
 action  = 'POST'
 
+# Add User-Agent to the headers
+request_headers['User-Agent'] = self.user_agent
+
 # Set the POST
 data = {}
 data['_id']             = PRISMA_CLOUD_CREDENTIAL_FRIENDLY_NAME
@@ -67,6 +70,9 @@
 api_url = 'https://' + PRISMA_CLOUD_COMPUTE_CONSOLE_URL + PRISMA_CLOUD_COMPUTE_CONSOLE_API_VERSION + '/policies/cloud-platforms'
 action = 'GET'
 
+# Add User-Agent to the headers
+request_headers['User-Agent'] = self.user_agent
+
 # GET
 response_raw = requests.request(action, api_url, auth=HTTPBasicAuth(PRISMA_CLOUD_API_ACCESS_KEY, PRISMA_CLOUD_API_SECRET_KEY), headers=headers, timeout=16)
 response_data = response_raw.json()
@@ -78,6 +84,9 @@
 api_url = 'https://' + PRISMA_CLOUD_COMPUTE_CONSOLE_URL + PRISMA_CLOUD_COMPUTE_CONSOLE_API_VERSION + '/policies/cloud-platforms'
 action = 'PUT'
 
+# Add User-Agent to the headers
+request_headers['User-Agent'] = self.user_agent
+
 # Set the POST
 new_policy_object = {}
 new_policy_object['awsRegionType']          = 'regular'

scripts/pcs_compute_forward_to_siem.py

@@ -100,6 +100,9 @@ def outbound_api_call(data_type:str, data: Union[list, dict]):
         profile_log('OUTBOUND_API_CALL', 'FINISHED')
         return
     print(f'        OUTBOUND_API_CALL for {data_type} ...')
+    # Add User-Agent to the headers
+    req_headers['User-Agent'] = self.user_agent
+
     api_response = requests.request(req_method, req_url, headers=req_headers, params=req_query_params, data=json.dumps(req_body_params), timeout=req_timeout, verify=False)
     if api_response.status_code in retry_status_codes:
         for _ in range(1, retry_limit):