le_util cleanup, le_util_test

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):

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

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,

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)))

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()