# Copyright 2019-2025 Ping Identity Corporation. All Rights Reserved # # This code is to be used exclusively in connection with Ping Identity # Corporation software or services. Ping Identity Corporation only offers # such software or services to legal entities who have entered into a # binding license agreement with Ping Identity Corporation. # -*- coding: utf-8 -*- # load tasks used by configuration file # Framework imports from pyrock.lib.scheduler.tasks.StepTask import StepTask from pyrock.tasks.scenario.gatling import GatlingTask from shared.lib.utils.HttpCmd import HttpCmd from shared.lib.utils.exception import FailException from shared.lib.cloud_utils import kubectl from pyrock.lib.PyRockRun import get_pyrock_run pyrock_run = get_pyrock_run() class EnableDsRemotePasswordHashingTask(StepTask): microservice_name = "ds-password-hashing-microservice" def pre(self): """Checking task config""" if not self.source.instance_name.startswith("ds-idrepo"): raise FailException("Source must be ds-idrepo component") def step1(self): """Verify that the password hashing microserice is deployed""" kubectl(f"get service {self.microservice_name}", component=self.source) def step2(self): """Scale the password hashing microserice to the same replica number than ds-idrepo""" kubectl( f"scale deployment {self.microservice_name} --replicas={self.source.get_expected_replica_count()}", component=self.source, ) def step3(self): """Increase the number of PBKDF2 iterations first to see the gains (and to be close to production environment)""" self.update_pbkdf2_password_storage_cheme("--set pbkdf2-iterations:10000") def step4(self): """Configure the remote password hashing feature""" self.update_pbkdf2_password_storage_cheme( f" --set remote-password-hashing-base-uri:http://{self.microservice_name}:8080/" " --set remote-password-hashing-connection-timeout:20s" " --set remote-password-hashing-request-timeout:120s" ) def step5(self): """Enable the remote password hashing feature""" self.update_pbkdf2_password_storage_cheme("--set remote-password-hashing-enabled:true") def update_pbkdf2_password_storage_cheme(self, dsconfig_set_clauses): for pod in self.source.pods: self.source.dsconfig( pod=pod, command=f"set-password-storage-scheme-prop --scheme-name PBKDF2-HMAC-SHA256 {dsconfig_set_clauses}", ) class ConfigureCTSPurge(StepTask): def pre(self): """Pre method""" pass def step1(self): """Reduce the replication purge delay""" ds_cts_idp = pyrock_run.get_component("ds-cts") ds_cts_idp.set_replication_purge_delay(21600) class EnablePrometheusMonitoring(StepTask): http_cmd = None rest_url = None mon_response = None monitoring_service_url = None mon_payload_json = None prometheus_response = None prometheus_conf_url = None prometheus_payload_json = None headers_json = None headers = {"cache-control": "no-cache", "content-type": "application/json", "accept": "application/json"} def pre(self): """Checking task config""" self.http_cmd = HttpCmd() if self.source.component_type != "am": raise FailException("Source must be am component") def step1(self): """Obtain SSO Token for amadmin""" self.rest_url = f"{self.source.base_url}/json" auth_url = self.rest_url + "/authenticate" self.headers_json = self.headers self.headers_json.update({"accept-api-version": "protocol=2.0,resource=1.0"}) self.headers_json.update({"x-openam-username": self.source.admin_user}) self.headers_json.update({"x-openam-password": self.source.admin_password}) self.http_cmd.post(url=auth_url, headers=self.headers_json) def step2(self): """Fetch the pay load for enabling monitoring""" self.monitoring_service_url = self.rest_url + "/global-config/services/monitoring" self.headers_json = self.headers self.headers_json.update({"accept-api-version": "protocol=1.0,resource=1.0"}) self.mon_response = self.http_cmd.get(self.monitoring_service_url, headers=self.headers_json) def step3(self): """Enabling AM monitoring in payload""" self.mon_payload_json = self.mon_response.json() self.mon_payload_json["enabled"] = True self.mon_payload_json["httpEnabled"] = True self.mon_payload_json["rmiEnabled"] = True print(f"Updated payload : {self.mon_payload_json}") def step4(self): """Remove '_id' from the payload""" for name in ["_id"]: print(f" removing : {name}") del self.mon_payload_json[name] def step5(self): """Update the configuration in AM (send the payload to server)""" self.headers_json = self.headers self.headers_json.update({"accept-api-version": "protocol=1.0,resource=1.0"}) self.http_cmd.put( self.monitoring_service_url, headers=self.headers_json, json_data=self.mon_payload_json, expected_string='"enabled":true', ) def step6(self): """Fetch the pay load for prometheus configuration""" self.prometheus_conf_url = self.rest_url + "/global-config/services/monitoring/prometheus/prometheus" self.headers_json = self.headers self.headers_json.update({"accept-api-version": "protocol=1.0,resource=1.0"}) self.prometheus_response = self.http_cmd.get(self.prometheus_conf_url, headers=self.headers_json) def step7(self): """Enabling prometheus monitoring in payload""" self.prometheus_payload_json = self.prometheus_response.json() self.prometheus_payload_json["enabled"] = True def step8(self): """Remove '_id' from the payload""" for name in ["_id"]: print(f" removing : {name}") del self.prometheus_payload_json[name] def step9(self): """Update the configuration in AM (send the payload to server)""" self.headers_json = self.headers self.headers_json.update({"accept-api-version": "protocol=1.0,resource=1.0"}) self.http_cmd.put( self.prometheus_conf_url, headers=self.headers_json, json_data=self.prometheus_payload_json, expected_string='"enabled":true', ) def post(self): """Post method""" self.http_cmd.close() self.set_result_pass()