diff --git a/base/server/healthcheck/pki/server/healthcheck/clones/connectivity_and_data.py b/base/server/healthcheck/pki/server/healthcheck/clones/connectivity_and_data.py
index ca5d6dae483..d9bb480f7f9 100644
--- a/base/server/healthcheck/pki/server/healthcheck/clones/connectivity_and_data.py
+++ b/base/server/healthcheck/pki/server/healthcheck/clones/connectivity_and_data.py
@@ -46,93 +46,83 @@ def check_ca_clones(self):
def check_kra_clones(self):
for host in self.clone_kras:
- cur_clone_msg = ' Host: ' + host.Hostname + ' Port: ' + host.SecurePort
- # Reach out and get some keys or requests , to serve as a data and connectivity check
+
+ url = 'https://' + host.Hostname + ':' + host.SecurePort
+
try:
- client_nick = self.security_domain.config.get('ca.connector.KRA.nickName')
-
- output = self.contact_subsystem_using_pki(
- host.SecurePort, host.Hostname, client_nick,
- self.passwd, self.db_dir, 'kra-key-show', ['0x01'])
-
- # check to see if we either got a key or a key not found exception
- # of which either will imply a successful connection
- if output is not None:
- key_found = output.find('Key ID:')
- key_not_found = output.find('KeyNotFoundException:')
- if key_found >= 0:
- logger.info('Key material found from kra clone.')
-
- if key_not_found >= 0:
- logger.info('key not found, possibly empty kra')
-
- if key_not_found == -1 and key_found == -1:
- logger.info('Failure to get key material from kra')
- raise BaseException('KRA clone problem detected ' + cur_clone_msg)
- else:
- raise BaseException('No data obtained from KRA clone.' + cur_clone_msg)
+ status = self.get_status(
+ host.Hostname,
+ host.SecurePort,
+ '/kra/admin/kra/getStatus')
- except BaseException as e:
- logger.error("Internal error testing KRA clone. %s", e)
- raise BaseException('Internal error testing KRA clone.' + cur_clone_msg)
+ logger.info('KRA at %s is %s', url, status)
- return
+ if status != 'running':
+ raise Exception('KRA at %s is %s' % (url, status))
+
+ except Exception as e:
+ logger.error('Unable to reach KRA at %s: %s', url, e)
+ raise Exception('Unable to reach KRA at %s: %s' % (url, e))
def check_ocsp_clones(self):
for host in self.clone_ocsps:
- cur_clone_msg = ' Host: ' + host.Hostname + ' Port: ' + host.SecurePort
- # Reach out to the ocsp clones
+
+ url = 'https://' + host.Hostname + ':' + host.SecurePort
+
try:
- output = self.contact_subsystem_using_sslget(
- host.SecurePort, host.Hostname, None,
- self.passwd, self.db_dir, None, '/ocsp/admin/ocsp/getStatus')
-
- good_status = output.find('1')
- if good_status == -1:
- raise BaseException('OCSP clone problem detected.' + cur_clone_msg)
- logger.info('good_status %s ', good_status)
- except BaseException as e:
- logger.error("Internal error testing OCSP clone. %s", e)
- raise BaseException('Internal error testing OCSP clone.' + cur_clone_msg)
+ status = self.get_status(
+ host.Hostname,
+ host.SecurePort,
+ '/ocsp/admin/ocsp/getStatus')
- return
+ logger.info('OCSP at %s is %s', url, status)
+
+ if status != 'running':
+ raise Exception('OCSP at %s is %s' % (url, status))
+
+ except Exception as e:
+ logger.error('Unable to reach OCSP at %s: %s', url, e)
+ raise Exception('Unable to reach OCSP at %s: %s' % (url, e))
def check_tks_clones(self):
for host in self.clone_tkss:
- cur_clone_msg = ' Host: ' + host.Hostname + ' Port: ' + host.SecurePort
- # Reach out to the tks clones
+
+ url = 'https://' + host.Hostname + ':' + host.SecurePort
+
try:
- output = self.contact_subsystem_using_sslget(
- host.SecurePort, host.Hostname, None,
- self.passwd, self.db_dir, None, '/tks/admin/tks/getStatus')
-
- good_status = output.find('1')
- if good_status == -1:
- raise BaseException('TKS clone problem detected.' + cur_clone_msg)
- logger.info('good_status %s ', good_status)
- except BaseException as e:
- logger.error("Internal error testing TKS clone. %s", e)
- raise BaseException('Internal error testing TKS clone.' + cur_clone_msg)
+ status = self.get_status(
+ host.Hostname,
+ host.SecurePort,
+ '/tks/admin/tks/getStatus')
- return
+ logger.info('TKS at %s is %s', url, status)
+
+ if status != 'running':
+ raise Exception('TKS at %s is %s' % (url, status))
+
+ except Exception as e:
+ logger.error('Unable to reach TKS at %s: %s', url, e)
+ raise Exception('Unable to reach TKS at %s: %s' % (url, e))
def check_tps_clones(self):
for host in self.clone_tpss:
- cur_clone_msg = ' Host: ' + host.Hostname + ' Port: ' + host.SecurePort
- # Reach out to the tps clones
+
+ url = 'https://' + host.Hostname + ':' + host.SecurePort
+
try:
- output = self.contact_subsystem_using_sslget(
- host.SecurePort, host.Hostname, None,
- self.passwd, self.db_dir, None, '/tps/admin/tps/getStatus')
-
- good_status = output.find('1')
- if good_status == -1:
- raise BaseException('TPS clone problem detected.' + cur_clone_msg)
- logger.info('good_status %s ', good_status)
- except BaseException as e:
- logger.error("Internal error testing TPS clone. %s", e)
- raise BaseException('Internal error testing TPS clone.' + cur_clone_msg)
- return
+ status = self.get_status(
+ host.Hostname,
+ host.SecurePort,
+ '/tps/admin/tps/getStatus')
+
+ logger.info('TPS at %s is %s', url, status)
+
+ if status != 'running':
+ raise Exception('TPS at %s is %s' % (url, status))
+
+ except Exception as e:
+ logger.error('Unable to reach TPS at %s: %s', url, e)
+ raise Exception('Unable to reach TPS at %s: %s' % (url, e))
@duration
def check(self):
diff --git a/base/server/healthcheck/pki/server/healthcheck/clones/plugin.py b/base/server/healthcheck/pki/server/healthcheck/clones/plugin.py
index 2472f35b5b5..4803e09ca51 100644
--- a/base/server/healthcheck/pki/server/healthcheck/clones/plugin.py
+++ b/base/server/healthcheck/pki/server/healthcheck/clones/plugin.py
@@ -6,6 +6,10 @@
# SPDX-License-Identifier: GPL-2.0-or-later
#
+import json
+import logging
+import xml.etree.ElementTree as ET
+
from ipahealthcheck.core.plugin import Plugin, Registry
from pki.server.instance import PKIInstance
from pki.client import PKIConnection
@@ -13,9 +17,6 @@
from pki.server.healthcheck.core.main import merge_dogtag_config
-import logging
-import subprocess
-
logger = logging.getLogger(__name__)
# Temporary workaround to skip VERBOSE data. Fix already pushed to upstream
@@ -46,60 +47,36 @@ def __init__(self, registry):
self.instance = PKIInstance(self.config.instance_name)
- def contact_subsystem_using_pki(
- self, subport, subhost, subsystemnick,
- token_pwd, db_path, cmd, exts=None):
- command = ["/usr/bin/pki",
- "-p", str(subport),
- "-h", subhost,
- "-n", subsystemnick,
- "-P", "https",
- "-d", db_path,
- "-c", token_pwd,
- cmd]
-
- if exts is not None:
- command.extend(exts)
-
- output = None
- try:
- output = subprocess.check_output(command, stderr=subprocess.STDOUT)
- except subprocess.CalledProcessError as e:
- output = e.output.decode('utf-8')
- return output
+ def get_status(self, host, port, path):
- output = output.decode('utf-8')
+ self.instance.export_ca_cert()
- return output
+ connection = PKIConnection(
+ protocol='https',
+ hostname=host,
+ port=port,
+ cert_paths=self.instance.ca_cert)
- def contact_subsystem_using_sslget(
- self, port, host, subsystemnick,
- token_pwd, db_path, params, url):
+ response = connection.get(path)
- command = ["/usr/bin/sslget"]
+ content_type = response.headers['Content-Type']
+ content = response.text
+ logger.info('Content:\n%s', content)
- if subsystemnick is not None:
- command.extend(["-n", subsystemnick])
+ # https://github.com/dogtagpki/pki/wiki/GetStatus-Service
+ if content_type == 'application/json':
+ json_response = json.loads(content)
+ status = json_response['Response']['Status']
- command.extend(["-p", token_pwd, "-d", db_path])
-
- if params is not None:
- command.extend(["-e", params])
-
- command.extend([
- "-r", url, host + ":" + port])
-
- logger.info(' command : %s ', command)
- output = None
- try:
- output = subprocess.check_output(command, stderr=subprocess.STDOUT)
- except subprocess.CalledProcessError as e:
- output = e.output.decode('utf-8')
- return output
+ elif content_type == 'application/xml':
+ root = ET.fromstring(response)
+ status = root.findtext('Status')
- output = output.decode('utf-8')
+ else:
+ raise Exception('Unsupported content-type: %s' % content_type)
- return output
+ logger.info('Status: %s', status)
+ return status
def get_security_domain_data(self, host, port):
domain_data = None
diff --git a/base/server/python/pki/server/__init__.py b/base/server/python/pki/server/__init__.py
index cbffc3ac0d1..bf02843f6ba 100644
--- a/base/server/python/pki/server/__init__.py
+++ b/base/server/python/pki/server/__init__.py
@@ -241,6 +241,10 @@ def nssdb_link(self):
def jss_conf(self):
return os.path.join(self.conf_dir, 'jss.conf')
+ @property
+ def ca_cert(self):
+ return os.path.join(self.nssdb_dir, 'ca.crt')
+
def is_valid(self):
return self.exists()
@@ -259,8 +263,6 @@ def is_active(self):
def export_ca_cert(self):
- ca_path = os.path.join(self.nssdb_dir, 'ca.crt')
-
token = pki.nssdb.INTERNAL_TOKEN_NAME
nickname = self.get_sslserver_cert_nickname()
@@ -275,7 +277,7 @@ def export_ca_cert(self):
nssdb = self.open_nssdb(token=token)
try:
- nssdb.extract_ca_cert(ca_path, nickname)
+ nssdb.extract_ca_cert(self.ca_cert, nickname)
finally:
nssdb.close()