diff --git a/letsencrypt/client/apache_configurator.py b/letsencrypt/client/apache_configurator.py index 9612b5872..b95d5c8bb 100644 --- a/letsencrypt/client/apache_configurator.py +++ b/letsencrypt/client/apache_configurator.py @@ -1106,7 +1106,7 @@ LogLevel warn \n\ # Create all of the challenge certs for t in chall_dict["listSNITuple"]: # Need to decode from base64 - r = le_util.b64_url_dec(t[1]) + r = le_util.jose_b64decode(t[1]) ext = self.dvsni_gen_ext(r, s) self.dvsni_create_chall_cert(t[0], ext, t[2], chall_dict["dvsni_key"]) @@ -1116,7 +1116,7 @@ LogLevel warn \n\ self.save("SNI Challenge", True) self.restart(True) - s = le_util.b64_url_enc(s) + s = le_util.jose_b64encode(s) return {"type":"dvsni", "s":s} def cleanup(self): diff --git a/letsencrypt/client/client.py b/letsencrypt/client/client.py index 7516eee05..36ab435cc 100644 --- a/letsencrypt/client/client.py +++ b/letsencrypt/client/client.py @@ -244,7 +244,7 @@ class Client(object): def revocation_request(self, key_file, cert_der): return {"type":"revocationRequest", - "certificate":le_util.b64_url_enc(cert_der), + "certificate":le_util.jose_b64encode(cert_der), "signature":crypto_util.create_sig(cert_der, key_file)} @@ -306,7 +306,7 @@ class Client(object): def certificate_request(self, csr_der, key): logger.info("Preparing and sending CSR..") return {"type":"certificateRequest", - "csr":le_util.b64_url_enc(csr_der), + "csr":le_util.jose_b64encode(csr_der), "signature":crypto_util.create_sig(csr_der, self.key_file)} def cleanup_challenges(self, challenge_objs): @@ -351,7 +351,7 @@ class Client(object): "nonce":server_nonce} auth_req["signature"] = crypto_util.create_sig( - name + le_util.b64_url_dec(server_nonce), self.key_file) + name + le_util.jose_b64decode(server_nonce), self.key_file) auth_req["responses"] = responses return auth_req diff --git a/letsencrypt/client/crypto_util.py b/letsencrypt/client/crypto_util.py index 815e795cd..f09dfd51f 100644 --- a/letsencrypt/client/crypto_util.py +++ b/letsencrypt/client/crypto_util.py @@ -15,7 +15,7 @@ from letsencrypt.client import le_util def b64_cert_to_pem(b64_der_cert): return M2Crypto.X509.load_cert_der_string( - le_util.b64_url_dec(b64_der_cert)).as_pem() + le_util.jose_b64decode(b64_der_cert)).as_pem() def create_sig(msg, key_file, signer_nonce=None, @@ -35,10 +35,10 @@ def create_sig(msg, key_file, signer_nonce=None, #print "signature:", signature n_bytes = binascii.unhexlify(leading_zeros(hex(key.n)[2:].replace("L", ""))) e_bytes = binascii.unhexlify(leading_zeros(hex(key.e)[2:].replace("L", ""))) - n_encoded = le_util.b64_url_enc(n_bytes) - e_encoded = le_util.b64_url_enc(e_bytes) - signer_nonce_encoded = le_util.b64_url_enc(signer_nonce) - sig_encoded = le_util.b64_url_enc(signature) + n_encoded = le_util.jose_b64encode(n_bytes) + e_encoded = le_util.jose_b64encode(e_bytes) + signer_nonce_encoded = le_util.jose_b64encode(signer_nonce) + sig_encoded = le_util.jose_b64encode(signature) jwk = {"kty": "RSA", "n": n_encoded, "e": e_encoded} signature = { "nonce": signer_nonce_encoded, diff --git a/letsencrypt/client/le_util.py b/letsencrypt/client/le_util.py index f9afd8171..68a8c405d 100644 --- a/letsencrypt/client/le_util.py +++ b/letsencrypt/client/le_util.py @@ -1,41 +1,63 @@ """Utilities for all Let's Encrypt.""" import base64 -import grp import errno import os -import pwd import stat -import sys - -from letsencrypt.client import logger -def make_or_verify_dir(directory, permissions=0755, uid=0): +def make_or_verify_dir(directory, mode=0755, uid=0): + """Make sure directory exists with proper permissions. + + :param directory: Path to a directry. + :type directory: str + + :param mode: Diretory mode. + :type mode: int + + :param uid: Directory owner. + :type uid: int + + :raises: Exception -- TODO + + """ try: - os.makedirs(directory, permissions) + os.makedirs(directory, mode) except OSError as exception: if exception.errno == errno.EEXIST: - if not check_permissions(directory, permissions, uid): - logger.fatal("%s exists and does not contain the proper permissions or owner" % directory) - sys.exit(57) + if not check_permissions(directory, mode, uid): + raise Exception('%s exists and does not contain the proper ' + 'permissions or owner' % directory) else: raise -def check_permissions(filepath, mode, uid=0): - file_stat = os.stat(filepath) - if stat.S_IMODE(file_stat.st_mode) != mode: - return False - return file_stat.st_uid == uid -def unique_file(default_name, mode = 0777): - """ - Safely finds a unique file for writing only (by default) +def check_permissions(filepath, mode, uid=0): + """Check file or directory permissions. + + :param filepath: Path to the tested file (or directory). + :type filepath: str + + :param mode: Expected file mode. + :type mode: int + + :param uid: Expected file owner. + :type uid: int + + :returns: bool -- True if `mode` and `uid` match, False otherwise. + """ + file_stat = os.stat(filepath) + return stat.S_IMODE(file_stat.st_mode) == mode and file_stat.st_uid == uid + + +def unique_file(default_name, mode=0777): + """Safely finds a unique file for writing only (by default).""" count = 1 f_parsed = os.path.splitext(default_name) while 1: try: - fd = os.open(default_name, os.O_CREAT|os.O_EXCL|os.O_RDWR, mode) + fd = os.open( + default_name, os.O_CREAT | os.O_EXCL | os.O_RDWR, mode) return os.fdopen(fd, 'w'), default_name except OSError: pass @@ -43,31 +65,48 @@ def unique_file(default_name, mode = 0777): count += 1 +def _to_utf8(arg): + """Normalize to UTF-8 string.""" + return arg.encode('utf-8') if isinstance(arg, unicode) else arg -def drop_privs(): - nogroup = grp.getgrnam("nogroup").gr_gid - nobody = pwd.getpwnam("nobody").pw_uid - os.setgid(nogroup) - os.setgroups([]) - os.setuid(nobody) -# Quick implementations of b64 url safe encode/decode -# We will include a proper library in the future if the library -# doesn't conflict with our existing dependencies -def b64_url_enc(s): - try: - s = s.encode("utf8") - except: - pass +def jose_b64encode(arg): + """JOSE Base64 encode. - i = base64.urlsafe_b64encode(s) - return i.rstrip("=") + JOSE Base64: + - URL-safe Base64 + - padding stripped -def b64_url_dec(s): - try: - s = s.encode("utf8") - except: - pass + https://tools.ietf.org/html/draft-ietf-jose-json-web-signature-37#appendix-C - pad = '=' * (4 - (len(s) % 4)) - return base64.urlsafe_b64decode(s + pad) + :param arg: String to be encoded. Unicode input will be encoded + to UTF-8 before Base64 encoding. + :type arg: str or unicode + + :returns: JOSE Base64 string. + :rtype: str + + """ + return base64.urlsafe_b64encode(_to_utf8(arg)).rstrip('=') + + +def jose_b64decode(arg): + """JOSE Base64 decode. + + Jose Base64: + - URL-safe Base64 + - padding stripped + + https://tools.ietf.org/html/draft-ietf-jose-json-web-signature-37#appendix-C + + :param arg: Base64 string to be decoded. Unicode input will be + encoded to UTF-8 before Base64 decoding. + :type arg: str or unicode + + :returns: Decoded string. + :rtype: str + + """ + normalized = _to_utf8(arg) + return base64.urlsafe_b64decode( + normalized + '=' * (4 - (len(normalized) % 4))) diff --git a/letsencrypt/client/le_util_test.py b/letsencrypt/client/le_util_test.py new file mode 100644 index 000000000..66c21b473 --- /dev/null +++ b/letsencrypt/client/le_util_test.py @@ -0,0 +1,104 @@ +"""Tests for letsencrypt.client.le_util.""" +import os +import shutil +import tempfile +import unittest + + +class MakeOrVerifyDirTest(unittest.TestCase): + """Tests for letsencrypt.client.le_util.make_or_verify_dir. + + Note that it is not possible to test for a wrong directory owner, + as this testing script would have to be run as root. + + """ + + def setUp(self): + self.root_path = tempfile.mkdtemp() + self.path = os.path.join(self.root_path, 'foo') + os.mkdir(self.path, 0400) + + self.uid = os.getuid() + + def tearDown(self): + shutil.rmtree(self.root_path, ignore_errors=True) + + def _call(self, directory, mode): + from letsencrypt.client.le_util import make_or_verify_dir + return make_or_verify_dir(directory, mode, self.uid) + + def test_creates_dir_when_missing(self): + path = os.path.join(self.root_path, 'bar') + self._call(path, 0650) + self.assertTrue(os.path.isdir(path)) + # TODO: check mode + + def test_existing_correct_mode_does_not_fail(self): + self._call(self.path, 0400) + # TODO: check mode + + def test_existing_wrong_mode_fails(self): + self.assertRaises(Exception, self._call, self.path, 0600) + + +class CheckPermissionsTest(unittest.TestCase): + """Tests for letsencrypt.client.le_util.check_permissions. + + Note that it is not possible to test for a wrong file owner, + as this testing script would have to be run as root. + + """ + + def setUp(self): + _, self.path = tempfile.mkstemp() + self.uid = os.getuid() + + def tearDown(self): + os.remove(self.path) + + def _call(self, mode): + from letsencrypt.client.le_util import check_permissions + return check_permissions(self.path, mode, self.uid) + + def test_ok_mode(self): + os.chmod(self.path, 0600) + self.assertTrue(self._call(0600)) + + def test_wrong_mode(self): + os.chmod(self.path, 0400) + self.assertFalse(self._call(0600)) + + +class JOSEB64EncodeTest(unittest.TestCase): + """Tests for letsencrypt.client.le_util.jose_b64encode.""" + + def _call(self, arg): + from letsencrypt.client.le_util import jose_b64encode + return jose_b64encode(arg) + + def test_str(self): + self.assertEqual(self._call('foo'), 'Zm9v') + + def test_unicode(self): + self.assertEqual(self._call(u'\u0105'), 'xIU') + + +class JOSEB64DecodeTest(unittest.TestCase): + """Tests for letsencrypt.client.le_util.jose_b64decode.""" + + def _call(self, arg): + from letsencrypt.client.le_util import jose_b64decode + return jose_b64decode(arg) + + def test_str(self): + self.assertEqual(self._call('Zm9v='), 'foo') + + def test_unicode(self): + self.assertEqual(self._call(u'XIU='), '\\\x85') + + def test_fills_padding(self): + self.assertEqual(self._call('Zm9v'), 'foo') + + +if __name__ == '__main__': + unittest.main()