Commit f1173ea5 authored by Jérome Perrin's avatar Jérome Perrin

software/erp5/test: test how balancer validates certs and updates CRL from frontend caucases

parent 5a702aa3
......@@ -561,20 +561,25 @@ class CaucaseClientCertificate(ManagedResource):
# type: () -> None
shutil.rmtree(self.tmpdir)
def request(self, common_name, caucase):
# type: (str, CaucaseService) -> None
"""Generate certificate and request signature to the caucase service.
This overwrite any previously requested certificate for this instance.
@property
def _caucase_path(self):
# type: () -> str
"""path of caucase executable.
"""
software_release_root_path = os.path.join(
self._cls.slap._software_root,
hashlib.md5(self._cls.getSoftwareURL().encode()).hexdigest(),
)
caucase_path = os.path.join(software_release_root_path, 'bin', 'caucase')
return os.path.join(software_release_root_path, 'bin', 'caucase')
def request(self, common_name, caucase):
# type: (str, CaucaseService) -> None
"""Generate certificate and request signature to the caucase service.
This overwrite any previously requested certificate for this instance.
"""
cas_args = [
caucase_path,
self._caucase_path,
'--ca-url', caucase.url,
'--ca-crt', self.ca_crt_file,
'--crl', self.crl_file,
......@@ -630,6 +635,18 @@ class CaucaseClientCertificate(ManagedResource):
with open(self.cert_file) as f:
assert 'BEGIN CERTIFICATE' in f.read()
def revoke(self, caucase):
# type: (str, CaucaseService) -> None
"""Revoke the client certificate on this caucase instance.
"""
subprocess.check_call([
self._caucase_path,
'--ca-url', caucase.url,
'--ca-crt', self.ca_crt_file,
'--crl', self.crl_file,
'--revoke-crt', self.cert_file, self.key_file,
])
class TestFrontendXForwardedFor(BalancerTestCase):
__partition_reference__ = 'xff'
......@@ -682,3 +699,103 @@ class TestFrontendXForwardedFor(BalancerTestCase):
headers={'X-Forwarded-For': '1.2.3.4'},
verify=False,
)
class TestClientTLS(BalancerTestCase):
__partition_reference__ = 'c'
@classmethod
def _getInstanceParameterDict(cls):
# type: () -> Dict
frontend_caucase1 = cls.getManagedResource('frontend_caucase1', CaucaseService)
certificate1 = cls.getManagedResource('client_certificate1', CaucaseClientCertificate)
certificate1.request(u'client_certificate1', frontend_caucase1)
frontend_caucase2 = cls.getManagedResource('frontend_caucase2', CaucaseService)
certificate2 = cls.getManagedResource('client_certificate2', CaucaseClientCertificate)
certificate2.request(u'client_certificate2', frontend_caucase2)
parameter_dict = super(TestClientTLS, cls)._getInstanceParameterDict()
parameter_dict['ssl-authentication-dict'] = {
'default': True,
}
parameter_dict['ssl']['frontend-caucase-url-list'] = [
frontend_caucase1.url,
frontend_caucase2.url,
]
return parameter_dict
def test_refresh_crl(self):
# type: () -> None
logger = self.logger
class DebugLogFile:
def write(self, msg):
logger.info("output from caucase_updater: %s", msg)
def flush(self):
pass
for client_certificate_name, caucase_name in (
('client_certificate1', 'frontend_caucase1'),
('client_certificate2', 'frontend_caucase2'),
):
client_certificate = self.getManagedResource(client_certificate_name,
CaucaseClientCertificate)
# when client certificate can be authenticated, backend receive the CN of
# the client certificate in "remote-user" header
def _make_request():
return requests.get(
self.default_balancer_url,
cert=(client_certificate.cert_file, client_certificate.key_file),
verify=False,
).json()
self.assertEqual(_make_request()['Incoming Headers'].get('remote-user'),
client_certificate_name)
# when certificate is revoked, updater service should update the CRL
# used by balancer from the caucase service used for client certificates
# (ie. the one used by frontend).
caucase = self.getManagedResource(caucase_name, CaucaseService)
client_certificate.revoke(caucase)
# until the CRL is updated, the client certificate is still accepted.
self.assertEqual(_make_request()['Incoming Headers'].get('remote-user'),
client_certificate_name)
# We have two services, in charge of updating CRL and CA certificates for
# each frontend CA
caucase_updater_list = glob.glob(
os.path.join(
self.computer_partition_root_path,
'etc',
'service',
'caucase-updater-*',
))
self.assertEqual(len(caucase_updater_list), 2)
# find the one corresponding to this caucase
for caucase_updater_candidate in caucase_updater_list:
with open(caucase_updater_candidate) as f:
if caucase.url in f.read():
caucase_updater = caucase_updater_candidate
break
else:
self.fail("Could not find caucase updater script for %s" % caucase.url)
# simulate running updater service in the future, to confirm that it fetches
# the new CRL and make sure balancer uses that new CRL.
process = pexpect.spawnu(
"faketime +1day %s" % caucase_updater,
env=dict(os.environ, PYTHONPATH=''),
)
process.logfile = DebugLogFile()
process.expect(u"Got new CRL.*Next wake-up at.*")
process.terminate()
process.wait()
with self.assertRaisesRegexp(Exception, 'certificate revoked'):
_make_request()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment