Commit 5405eb20 authored by Vincent Pelletier's avatar Vincent Pelletier

test: Run client with a realistic stdout.

stdout is supposed to accept bytes, not unicode objects.
And byte-ify all printed values to satisfy python3.
parent 3e509b16
......@@ -95,7 +95,10 @@ class CLICaucaseClient(object):
csr_pem = utils.getCertRequest(csr_path)
# Quick sanity check
utils.load_certificate_request(csr_pem)
print(self._client.createCertificateSigningRequest(csr_pem), csr_path)
print(
self._client.createCertificateSigningRequest(csr_pem),
utils.toBytes(csr_path),
)
def getCSR(self, csr_id_path_list):
"""
......@@ -122,10 +125,10 @@ class CLICaucaseClient(object):
except CaucaseError as e:
if e.args[0] != httplib.NOT_FOUND:
raise
print(crt_id, 'not found - maybe CSR was rejected ?')
print(crt_id, b'not found - maybe CSR was rejected ?')
error = True
else:
print(crt_id, 'CSR still pending')
print(crt_id, b'CSR still pending')
warning = True
else:
print(crt_id, end=' ')
......@@ -134,15 +137,15 @@ class CLICaucaseClient(object):
ca_list,
None,
)):
print('was (originally) automatically approved')
print(b'was (originally) automatically approved')
else:
print('was (originally) manually approved')
print(b'was (originally) manually approved')
if os.path.exists(crt_path):
try:
key_pem = utils.getKey(crt_path)
except ValueError:
print(
'Expected to find exactly one privatekey key in %s, skipping' % (
b'Expected to find exactly one privatekey key in %s, skipping' % (
crt_path,
),
file=sys.stderr,
......@@ -153,7 +156,7 @@ class CLICaucaseClient(object):
utils.validateCertAndKey(crt_pem, key_pem)
except ValueError:
print(
'Key in %s does not match retrieved certificate, skipping',
b'Key in %s does not match retrieved certificate, skipping',
file=sys.stderr,
)
error = True
......@@ -171,7 +174,7 @@ class CLICaucaseClient(object):
crt, key, _ = utils.getKeyPair(crt_path, key_path)
except ValueError:
print(
'Could not find (exactly) one matching key pair in %s, skipping' % (
b'Could not find (exactly) one matching key pair in %s, skipping' % (
[x for x in set((crt_path, key_path)) if x],
),
file=sys.stderr,
......@@ -201,7 +204,7 @@ class CLICaucaseClient(object):
)
except ValueError:
print(
'Could not find (exactly) one matching key pair in %s, skipping' % (
b'Could not find (exactly) one matching key pair in %s, skipping' % (
[x for x in set((crt_path, key_path)) if x],
),
file=sys.stderr,
......@@ -217,11 +220,11 @@ class CLICaucaseClient(object):
except exceptions.CertificateVerificationError:
print(
crt_path,
'was not signed by this CA, revoked or otherwise invalid, skipping',
b'was not signed by this CA, revoked or otherwise invalid, skipping',
)
continue
if renewal_deadline < old_crt.not_valid_after:
print(crt_path, 'did not reach renew threshold, not renewing')
print(crt_path, b'did not reach renew threshold, not renewing')
continue
new_key_pem, new_crt_pem = self._client.renewCertificate(
old_crt=old_crt,
......@@ -249,22 +252,22 @@ class CLICaucaseClient(object):
"""
--list-csr
"""
print('-- pending', mode, 'CSRs --')
print(b'-- pending', mode, b'CSRs --')
print(
'%20s | %s' % (
'csr_id',
'subject preview (fetch csr and check full content !)',
b'%20s | %s' % (
b'csr_id',
b'subject preview (fetch csr and check full content !)',
),
)
for entry in self._client.getPendingCertificateRequestList():
csr = utils.load_certificate_request(utils.toBytes(entry['csr']))
print(
'%20s | %r' % (
entry['id'],
csr.subject,
b'%20s | %r' % (
utils.toBytes(entry['id']),
utils.toBytes(repr(csr.subject)),
),
)
print('-- end of pending', mode, 'CSRs --')
print(b'-- end of pending', mode, b'CSRs --')
def signCSR(self, csr_id_list):
"""
......@@ -308,7 +311,7 @@ class CLICaucaseClient(object):
crt_pem = utils.getCert(crt_path)
except ValueError:
print(
'Could not load a single certificate in %s, skipping' % (
b'Could not load a single certificate in %s, skipping' % (
crt_path,
),
file=sys.stderr,
......@@ -524,8 +527,8 @@ def main(argv=None):
sign_csr_id_set.intersection(sign_with_csr_id_set)
):
print(
'A given CSR_ID cannot be in more than one of --sign-csr, '
'--sign-csr-with and --reject-csr',
b'A given CSR_ID cannot be in more than one of --sign-csr, '
b'--sign-csr-with and --reject-csr',
file=sys.stderr,
)
raise SystemExit(STATUS_ERROR)
......@@ -751,12 +754,12 @@ def updater(argv=None, until=utils.until):
ca_crt_pem_list=utils.getCertList(args.cas_ca)
)
if args.crt and not utils.hasOneCert(args.crt):
print('Bootstraping...')
print(b'Bootstraping...')
csr_pem = utils.getCertRequest(args.csr)
# Quick sanity check before bothering server
utils.load_certificate_request(csr_pem)
csr_id = client.createCertificateSigningRequest(csr_pem)
print('Waiting for signature of', csr_id)
print(b'Waiting for signature of', csr_id)
while True:
try:
crt_pem = client.getCertificate(csr_id)
......@@ -774,12 +777,12 @@ def updater(argv=None, until=utils.until):
crt_file.write(crt_pem)
updated = True
break
print('Bootstrap done')
print(b'Bootstrap done')
next_deadline = datetime.datetime.utcnow()
while True:
print(
'Next wake-up at',
next_deadline.strftime('%Y-%m-%d %H:%M:%S +0000'),
b'Next wake-up at',
next_deadline.strftime(b'%Y-%m-%d %H:%M:%S +0000'),
)
now = until(next_deadline)
next_deadline = now + max_sleep
......@@ -792,7 +795,7 @@ def updater(argv=None, until=utils.until):
ca_crt_pem_list=utils.getCertList(args.cas_ca)
)
if RetryingCaucaseClient.updateCAFile(ca_url, args.ca):
print('Got new CA')
print(b'Got new CA')
updated = True
# Note: CRL expiration should happen several time during CA renewal
# period, so it should not be needed to keep track of CA expiration
......@@ -802,7 +805,7 @@ def updater(argv=None, until=utils.until):
for x in utils.getCertList(args.ca)
]
if RetryingCaucaseClient.updateCRLFile(ca_url, args.crl, ca_crt_list):
print('Got new CRL')
print(b'Got new CRL')
updated = True
with open(args.crl, 'rb') as crl_file:
next_deadline = min(
......@@ -813,7 +816,7 @@ def updater(argv=None, until=utils.until):
crt_pem, key_pem, key_path = utils.getKeyPair(args.crt, args.key)
crt = utils.load_certificate(crt_pem, ca_crt_list, None)
if crt.not_valid_after - threshold <= now:
print('Renewing', args.crt)
print(b'Renewing', args.crt)
new_key_pem, new_crt_pem = client.renewCertificate(
old_crt=crt,
old_key=utils.load_privatekey(key_pem),
......@@ -843,7 +846,7 @@ def updater(argv=None, until=utils.until):
if args.on_renew is not None:
status = os.system(args.on_renew)
if status:
print('Renewal hook exited with status:', status, file=sys.stderr)
print(b'Renewal hook exited with status:', status, file=sys.stderr)
raise SystemExit(STATUS_ERROR)
updated = False
except (utils.SleepInterrupt, SystemExit):
......@@ -954,4 +957,4 @@ def key_id(argv=None):
backup_file.read(struct.calcsize('<I')),
)
for key_entry in json.loads(backup_file.read(header_len))['key_list']:
print(' ', key_entry['id'])
print(b' ', key_entry['id'])
......@@ -182,7 +182,7 @@ class CaucaseSSLWSGIRequestHandler(CaucaseWSGIRequestHandler):
# Note: compared to BaseHTTPHandler, logs the client certificate serial as
# user name.
print(
"%s - %s [%s] %s" % (
'%s - %s [%s] %s' % (
self.client_address[0],
self.ssl_client_cert_serial,
self.log_date_time_string(),
......@@ -952,11 +952,11 @@ def manage(argv=None):
found_from = ', '.join(ca_pair['from'])
crt = ca_pair['crt']
if crt is None:
print('No certificate correspond to ' + found_from + ', skipping')
print(b'No certificate correspond to', found_from, b'- skipping')
continue
expiration = utils.datetime2timestamp(crt.not_valid_after)
if expiration < now:
print('Skipping expired certificate from ' + found_from)
print(b'Skipping expired certificate from', found_from)
del import_ca_dict[identifier]
continue
if not args.import_bad_ca:
......@@ -975,11 +975,11 @@ def manage(argv=None):
or not key_usage.key_cert_sign or not key_usage.crl_sign
)
if failed:
print('Skipping non-CA certificate from ' + found_from)
print(b'Skipping non-CA certificate from', found_from)
continue
key = ca_pair['key']
if key is None:
print('No private key correspond to ' + found_from + ', skipping')
print(b'No private key correspond to', found_from, b'- skipping')
continue
imported += 1
cas_db.appendCAKeyPair(
......@@ -991,7 +991,7 @@ def manage(argv=None):
)
if not imported:
raise ValueError('No CA certificate imported')
print('Imported %i CA certificates' % imported)
print(b'Imported %i CA certificates' % imported)
if args.import_crl:
db = SQLite3Storage(db_path, table_prefix='cas')
trusted_ca_crt_set = [
......@@ -1014,7 +1014,7 @@ def manage(argv=None):
already_revoked_count += 1
else:
revoked_count += 1
print('Revoked %i certificates (%i were already revoked)' % (
print(b'Revoked %i certificates (%i were already revoked)' % (
revoked_count,
already_revoked_count,
))
......
......@@ -398,7 +398,7 @@ class CaucaseTest(unittest.TestCase):
Returns stdout.
"""
orig_stdout = sys.stdout
sys.stdout = stdout = StringIO()
sys.stdout = stdout = BytesIO()
try:
cli.main(
argv=(
......@@ -2194,7 +2194,7 @@ class CaucaseTest(unittest.TestCase):
orig_stdout = sys.stdout
try:
caucase.http.getBytePass = lambda x: b'test'
sys.stdout = stdout = StringIO()
sys.stdout = stdout = BytesIO()
self.assertFalse(os.path.exists(exported_ca), exported_ca)
caucase.http.manage(
argv=(
......
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