Read the IPA CA certificate to obtain the serial number

The dogtag connectivity test contains the command
ipa cert-show to verify that the CA basically works
and we are allowed to use it. It had a hardcoded 1 as
the CA certificate since pre-random serial numbers it was
a predictable value.

Instead read the CA cert and pluck the serial number from
it and use that value instead. /etc/ipa/ca.crt may contain
multiple certificates in the case of an external CA install
or additional certs added by ipa-cacert-manage so sift
through the list to find the expected IPA subject.

#260

Signed-off-by: Rob Crittenden <[email protected]>

src/ipahealthcheck/dogtag/ca.py

@@ -9,11 +9,13 @@
 from ipahealthcheck.core.plugin import duration
 from ipahealthcheck.core import constants
 
-from ipalib import api, errors
+from ipalib import api, errors, x509
 from ipaplatform.paths import paths
 from ipaserver.install import certs
+from ipaserver.install import ca
 from ipaserver.install import krainstance
 from ipapython.directivesetter import get_directive
+from ipapython.dn import DN
 from cryptography.hazmat.primitives.serialization import Encoding
 
 logger = logging.getLogger()
@@ -102,22 +104,59 @@ def check(self):
  logger.debug('CA is not configured, skipping connectivity check')
  return
 
+ config = api.Command.config_show()
+
+ subject_base = config['result']['ipacertificatesubjectbase'][0]
+ ipa_subject = ca.lookup_ca_subject(api, subject_base)
+ try:
+ certs = x509.load_certificate_list_from_file(paths.IPA_CA_CRT)
+ except Exception as e:
+ yield Result(self, constants.ERROR,
+ key='ipa_ca_crt_file_missing',
+ path=paths.IPA_CA_CRT,
+ error=str(e),
+ msg='The IPA CA cert file {path} could not be '
+ 'opened: {error}')
+ return
+
+ found = False
+ for cert in certs:
+ if DN(cert.subject) == ipa_subject:
+ found = True
+ break
+
+ if not found:
+ yield Result(self, constants.ERROR,
+ key='ipa_ca_cert_not_found',
+ subject=str(ipa_subject),
+ path=paths.IPA_CA_CRT,
+ msg='The CA certificate with subject {subject} '
+ 'was not found in {path}')
+ return
+ # Load the IPA CA certificate to obtain its serial number. This
+ # was traditionally 1 prior to random serial number support.
  # There is nothing special about cert 1. Even if there is no cert
  # serial number 1 but the connection is ok it is considered passing.
  try:
- api.Command.cert_show(1, all=True)
+ api.Command.cert_show(cert.serial_number, all=True)
  except errors.CertificateOperationError as e:
- if 'not found' not in str(e):
+ if 'not found' in str(e):
  yield Result(self, constants.ERROR,
  key='cert_show_1',
- msg='Request for certificate failed, %s' %
- e)
+ error=str(e),
+ serial=str(cert.serial_number),
+ msg='Serial number not found: {error}')
  else:
- yield Result(self, constants.SUCCESS)
+ yield Result(self, constants.ERROR,
+ key='cert_show_1',
+ error=str(e),
+ serial=str(cert.serial_number),
+ msg='Request for certificate failed: {error}')
  except Exception as e:
  yield Result(self, constants.ERROR,
  key='cert_show_1',
- msg='Request for certificate failed, %s' %
- e)
+ error=str(e),
+ serial=str(cert.serial_number),
+ msg='Request for certificate failed: {error')
  else:
  yield Result(self, constants.SUCCESS)

tests/test_dogtag_connectivity.py

@@ -2,14 +2,46 @@
 # Copyright (C) 2019 FreeIPA Contributors see COPYING for license
 #
 
+from unittest.mock import Mock, patch
 from util import capture_results, CAInstance
 from util import m_api
+
 from base import BaseTest
 from ipahealthcheck.core import constants, config
 from ipahealthcheck.dogtag.plugin import registry
 from ipahealthcheck.dogtag.ca import DogtagCertsConnectivityCheck
-from unittest.mock import Mock
+
 from ipalib.errors import CertificateOperationError
+from ipaplatform.paths import paths
+from ipapython.dn import DN
+
+
+class IPACertificate:
+ def __init__(self, serial_number=1,
+ subject='CN=Certificate Authority, O=%s' % m_api.env.realm):
+ self.serial_number = serial_number
+ self.subject = subject
+
+ def __eq__(self, other):
+ return self.serial_number == other.serial_number
+
+ def __hash__(self):
+ return hash(self.serial_number)
+
+
+subject_base = [{
+ 'result':
+ {
+ 'ipacertificatesubjectbase': [f'O={m_api.env.realm}'],
+ },
+}]
+
+bad_subject_base = [{
+ 'result':
+ {
+ 'ipacertificatesubjectbase': ['O=BAD'],
+ },
+}]
 
 
 class TestCAConnectivity(BaseTest):
@@ -18,12 +50,18 @@ class TestCAConnectivity(BaseTest):
  Mock(return_value=CAInstance()),
  }
 
- def test_ca_connection_ok(self):
+ @patch('ipaserver.install.ca.lookup_ca_subject')
+ @patch('ipalib.x509.load_certificate_list_from_file')
+ def test_ca_connection_ok(self, mock_load_cert, mock_ca_subject):
  """CA connectivity check when cert_show returns a valid value"""
  m_api.Command.cert_show.side_effect = None
+ m_api.Command.config_show.side_effect = subject_base
  m_api.Command.cert_show.return_value = {
  u'result': {u'revoked': False}
  }
+ mock_load_cert.return_value = [IPACertificate(12345)]
+ mock_ca_subject.return_value = DN(('cn', 'Certificate Authority'),
+ f'O={m_api.env.realm}')
 
  framework = object()
  registry.initialize(framework, config.Config)
@@ -38,13 +76,20 @@ def test_ca_connection_ok(self):
  assert result.source == 'ipahealthcheck.dogtag.ca'
  assert result.check == 'DogtagCertsConnectivityCheck'
 
- def test_ca_connection_cert_not_found(self):
+ @patch('ipaserver.install.ca.lookup_ca_subject')
+ @patch('ipalib.x509.load_certificate_list_from_file')
+ def test_ca_connection_cert_not_found(self, mock_load_cert,
+ mock_ca_subject):
  """CA connectivity check for a cert that doesn't exist"""
  m_api.Command.cert_show.reset_mock()
+ m_api.Command.config_show.side_effect = subject_base
  m_api.Command.cert_show.side_effect = CertificateOperationError(
  message='Certificate operation cannot be completed: '
  'EXCEPTION (Certificate serial number 0x0 not found)'
  )
+ mock_load_cert.return_value = [IPACertificate()]
+ mock_ca_subject.return_value = DN(('cn', 'Certificate Authority'),
+ f'O={m_api.env.realm}')
 
  framework = object()
  registry.initialize(framework, config.Config)
@@ -55,16 +100,82 @@ def test_ca_connection_cert_not_found(self):
  assert len(self.results) == 1
 
  result = self.results.results[0]
- assert result.result == constants.SUCCESS
+ assert result.result == constants.ERROR
+ assert result.source == 'ipahealthcheck.dogtag.ca'
+ assert result.check == 'DogtagCertsConnectivityCheck'
+ assert result.kw.get('key') == 'cert_show_1'
+ assert result.kw.get('serial') == '1'
+ assert result.kw.get('msg') == 'Serial number not found: {error}'
+
+ @patch('ipaserver.install.ca.lookup_ca_subject')
+ @patch('ipalib.x509.load_certificate_list_from_file')
+ def test_ca_connection_cert_file_not_found(self, mock_load_cert,
+ mock_ca_subject):
+ """CA connectivity check for a cert that doesn't exist"""
+ m_api.Command.cert_show.reset_mock()
+ m_api.Command.config_show.side_effect = subject_base
+ mock_load_cert.side_effect = FileNotFoundError()
+ mock_ca_subject.return_value = DN(('cn', 'Certificate Authority'),
+ f'O={m_api.env.realm}')
+
+ framework = object()
+ registry.initialize(framework, config.Config)
+ f = DogtagCertsConnectivityCheck(registry)
+
+ self.results = capture_results(f)
+
+ assert len(self.results) == 1
+
+ result = self.results.results[0]
+ assert result.result == constants.ERROR
  assert result.source == 'ipahealthcheck.dogtag.ca'
  assert result.check == 'DogtagCertsConnectivityCheck'
+ assert result.kw.get('key') == 'ipa_ca_crt_file_missing'
+ assert result.kw.get('path') == paths.IPA_CA_CRT
+
+ @patch('ipaserver.install.ca.lookup_ca_subject')
+ @patch('ipalib.x509.load_certificate_list_from_file')
+ def test_ca_connection_cert_not_in_file_list(self, mock_load_cert,
+ mock_ca_subject):
+ """CA connectivity check for a cert that isn't in IPA_CA_CRT"""
+ m_api.Command.cert_show.reset_mock()
+ m_api.Command.config_show.side_effect = bad_subject_base
+ mock_load_cert.return_value = [IPACertificate()]
+ mock_ca_subject.return_value = DN(('cn', 'Certificate Authority'),
+ 'O=BAD')
 
- def test_ca_connection_down(self):
+ framework = object()
+ registry.initialize(framework, config.Config)
+ f = DogtagCertsConnectivityCheck(registry)
+
+ self.results = capture_results(f)
+
+ assert len(self.results) == 1
+
+ result = self.results.results[0]
+ assert result.result == constants.ERROR
+ assert result.source == 'ipahealthcheck.dogtag.ca'
+ assert result.check == 'DogtagCertsConnectivityCheck'
+ bad = bad_subject_base[0]['result']['ipacertificatesubjectbase'][0]
+ bad_subject = DN(f'CN=Certificate Authority,{bad}')
+ assert DN(result.kw['subject']) == bad_subject
+ assert result.kw['path'] == paths.IPA_CA_CRT
+ assert result.kw['msg'] == (
+ 'The CA certificate with subject {subject} was not found in {path}'
+ )
+
+ @patch('ipaserver.install.ca.lookup_ca_subject')
+ @patch('ipalib.x509.load_certificate_list_from_file')
+ def test_ca_connection_down(self, mock_load_cert, mock_ca_subject):
  """CA connectivity check with the CA down"""
  m_api.Command.cert_show.side_effect = CertificateOperationError(
  message='Certificate operation cannot be completed: '
  'Unable to communicate with CMS (503)'
  )
+ m_api.Command.config_show.side_effect = subject_base
+ mock_load_cert.return_value = [IPACertificate()]
+ mock_ca_subject.return_value = DN(('cn', 'Certificate Authority'),
+ f'O={m_api.env.realm}')
 
  framework = object()
  registry.initialize(framework, config.Config)
@@ -78,4 +189,93 @@ def test_ca_connection_down(self):
  assert result.result == constants.ERROR
  assert result.source == 'ipahealthcheck.dogtag.ca'
  assert result.check == 'DogtagCertsConnectivityCheck'
- assert 'Unable to communicate' in result.kw.get('msg')
+ assert result.kw.get('msg') == (
+ 'Request for certificate failed: {error}'
+ )
+
+ @patch('ipaserver.install.ca.lookup_ca_subject')
+ @patch('ipalib.x509.load_certificate_list_from_file')
+ def test_ca_connection_multiple_ok(self, mock_load_cert, mock_ca_subject):
+ """CA connectivity check when cert_show returns a valid value"""
+ m_api.Command.cert_show.side_effect = None
+ m_api.Command.config_show.side_effect = subject_base
+ m_api.Command.cert_show.return_value = {
+ u'result': {u'revoked': False}
+ }
+ mock_load_cert.return_value = [
+ IPACertificate(1, 'CN=something'),
+ IPACertificate(12345),
+ ]
+ mock_ca_subject.return_value = DN(('cn', 'Certificate Authority'),
+ f'O={m_api.env.realm}')
+
+ framework = object()
+ registry.initialize(framework, config.Config)
+ f = DogtagCertsConnectivityCheck(registry)
+
+ self.results = capture_results(f)
+
+ assert len(self.results) == 1
+
+ result = self.results.results[0]
+ assert result.result == constants.SUCCESS
+ assert result.source == 'ipahealthcheck.dogtag.ca'
+
+ @patch('ipaserver.install.ca.lookup_ca_subject')
+ @patch('ipalib.x509.load_certificate_list_from_file')
+ def test_ca_connection_multiple_ok_reverse(self, mock_load_cert,
+ mock_ca_subject):
+ """CA connectivity check when cert_show returns a valid value"""
+ m_api.Command.cert_show.side_effect = None
+ m_api.Command.config_show.side_effect = subject_base
+ m_api.Command.cert_show.return_value = {
+ u'result': {u'revoked': False}
+ }
+ mock_load_cert.return_value = [
+ IPACertificate(12345),
+ IPACertificate(1, 'CN=something'),
+ ]
+ mock_ca_subject.return_value = DN(('cn', 'Certificate Authority'),
+ f'O={m_api.env.realm}')
+
+ framework = object()
+ registry.initialize(framework, config.Config)
+ f = DogtagCertsConnectivityCheck(registry)
+
+ self.results = capture_results(f)
+
+ assert len(self.results) == 1
+
+ result = self.results.results[0]
+ assert result.result == constants.SUCCESS
+ assert result.source == 'ipahealthcheck.dogtag.ca'
+
+ @patch('ipaserver.install.ca.lookup_ca_subject')
+ @patch('ipalib.x509.load_certificate_list_from_file')
+ def test_ca_connection_not_found(self, mock_load_cert, mock_ca_subject):
+ """CA connectivity check when cert_show returns a valid value"""
+ m_api.Command.cert_show.side_effect = None
+ m_api.Command.config_show.side_effect = subject_base
+ m_api.Command.cert_show.return_value = {
+ u'result': {u'revoked': False}
+ }
+ mock_load_cert.return_value = [
+ IPACertificate(1, 'CN=something'),
+ ]
+ mock_ca_subject.return_value = DN(('cn', 'Certificate Authority'),
+ f'O={m_api.env.realm}')
+
+ framework = object()
+ registry.initialize(framework, config.Config)
+ f = DogtagCertsConnectivityCheck(registry)
+
+ self.results = capture_results(f)
+
+ assert len(self.results) == 1
+
+ result = self.results.results[0]
+ assert result.result == constants.ERROR
+ assert result.source == 'ipahealthcheck.dogtag.ca'
+ assert result.kw['msg'] == (
+ 'The CA certificate with subject {subject} was not found in {path}'
+ )