Fix python3 compatibility issues

.travis.yml

@@ -1,6 +1,6 @@
 language: python
 python:
-- '2.7'
+- '3.6'
 install:
 - pip install -U pip pytest
 - pip install -e .[dev]

credsmash/api/put.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 import logging
+# from typing import Any, Optional
 
 from credsmash.crypto import seal_secret, open_secret, ALGO_AES_CTR
 
@@ -12,6 +13,7 @@ def put_secret(
         plaintext, version=None, compare=True,
         algorithm=ALGO_AES_CTR, **seal_kwargs
 ):
+    # type: (Any, Any, str, bytes, Optional[int], bool, str, ...) -> int
     if version is None:
         latest_secret = storage_service.get_one(secret_name)
         version = 1

credsmash/cli.py

@@ -10,6 +10,7 @@
 import boto3
 import click
 import pkg_resources
+import six
 
 import credsmash.api
 from credsmash.crypto import ALGO_AES_CTR
@@ -86,7 +87,7 @@ def __repr__(self):
               help="the KMS key-id of the master key "
                    "to use. See the README for more "
                    "information. Defaults to alias/credsmash")
-@click.option('--context', type=(unicode, unicode), multiple=True,
+@click.option('--context', type=(six.text_type, six.text_type), multiple=True,
               help="the KMS encryption context to use."
                    "Only works if --key-id is passed.")
 @click.pass_context
@@ -234,7 +235,7 @@ def cmd_prune_many(ctx, pattern):
 
 @main.command('get')
 @click.argument('secret_name')
-@click.argument('destination', type=click.File('wb'), required=False, default=sys.stdout)
+@click.argument('destination', type=click.File('wb'), required=False, default='-')
 @click.option('-f', '--format', 'fmt', default=None)
 @click.option('--version', '-v', default=None, type=click.INT)
 @click.pass_context
@@ -254,7 +255,7 @@ def cmd_get_one(ctx, secret_name, destination, fmt=None, version=None):
 
 
 @main.command('get-all')
-@click.argument('destination', type=click.File('wb'), required=False, default=sys.stdout)
+@click.argument('destination', type=click.File('wb'), required=False, default='-')
 @click.option('-f', '--format', 'fmt', default=None)
 @click.pass_context
 def cmd_get_all(ctx, destination, fmt):
@@ -279,7 +280,7 @@ def cmd_get_all(ctx, destination, fmt):
 
 @main.command('find-one')
 @click.argument('pattern')
-@click.argument('destination', type=click.File('wb'), required=False, default=sys.stdout)
+@click.argument('destination', type=click.File('wb'), required=False, default='-')
 @click.option('-f', '--format', 'fmt', default=None)
 @click.option('--version', '-v', default=None)
 @click.pass_context
@@ -310,7 +311,7 @@ def cmd_find_one(ctx, pattern, destination, fmt=None, version=None):
 
 @main.command('find-many')
 @click.argument('pattern')
-@click.argument('destination', type=click.File('wb'), required=False, default=sys.stdout)
+@click.argument('destination', type=click.File('wb'), required=False, default='-')
 @click.option('-f', '--format', 'fmt', default=None)
 @click.pass_context
 def cmd_find_many(ctx, pattern, destination, fmt=None):

credsmash/crypto/aes_ctr.py

@@ -1,7 +1,8 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
-from base64 import b64encode, b64decode
 import os
+import base64
+import binascii
 
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes
@@ -22,7 +23,7 @@
 }
 
 DEFAULT_DIGEST = 'SHA256'
-HASHING_ALGORITHMS = _hash_classes.keys()
+HASHING_ALGORITHMS = list(_hash_classes.keys())
 LEGACY_NONCE = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01'
 
 
@@ -73,13 +74,13 @@ def open_aes_ctr_legacy(key_service, material):
     """
     Decrypts secrets stored by `seal_aes_ctr_legacy`.
 
-    Assumes that the plaintext is unicode (non-binary).
+    Assumes that the plaintext is str (non-binary).
     """
-    key = key_service.decrypt(b64decode(material['key']))
+    key = key_service.decrypt(_from_b64(material['key']))
     digest_method = material.get('digest', DEFAULT_DIGEST)
-    ciphertext = b64decode(material['contents'])
-    hmac = material['hmac'].decode('hex')
-    return _open_aes_ctr(key, LEGACY_NONCE, ciphertext, hmac, digest_method).decode("utf-8")
+    ciphertext = _from_b64(material['contents'])
+    hmac = _from_hex(material['hmac'])
+    return _open_aes_ctr(key, LEGACY_NONCE, ciphertext, hmac, digest_method)
 
 
 def seal_aes_ctr_legacy(key_service, secret, digest_method=DEFAULT_DIGEST):
@@ -97,9 +98,9 @@ def seal_aes_ctr_legacy(key_service, secret, digest_method=DEFAULT_DIGEST):
         secret, key, LEGACY_NONCE, digest_method,
     )
     return {
-        'key': b64encode(encoded_key).decode('utf-8'),
-        'contents': b64encode(ciphertext).decode('utf-8'),
-        'hmac': hmac.encode('hex'),
+        'key': _to_b64(encoded_key),
+        'contents': _to_b64(ciphertext),
+        'hmac': _to_hex(hmac),
         'digest': digest_method,
     }
 
@@ -153,6 +154,24 @@ def get_digest(digest):
         raise ValueError("Could not find " + digest + " in cryptography.hazmat.primitives.hashes")
 
 
+def _to_hex(b):  # type (bytes): -> str
+    return binascii.hexlify(b).decode('ascii')
+
+
+def _from_hex(s):  # type: (str) -> bytes
+    return binascii.unhexlify(s)
+
+
+def _to_b64(b):  # type (bytes): -> str
+    b2 = base64.b64encode(b)
+    return b2.decode('ascii')
+
+
+def _from_b64(s):  # type (str): -> bytes
+    b = s.encode('ascii')
+    return base64.b64decode(b)
+
+
 class IntegrityError(Exception):
 
     def __init__(self, value=""):

credsmash/templates.py

@@ -6,25 +6,38 @@
 import logging
 import os
 import pwd
+# from typing import Dict, Union
 
 import click
 import jinja2.sandbox
 
 import credsmash.api
-from .util import read_many, minjson, envfile_quote, shell_quote, parse_manifest, detect_format, ItemNotFound
+from .util import read_many, read_many_str, minjson, envfile_quote, shell_quote, parse_manifest, detect_format, ItemNotFound
 from .cli import main
 
 logger = logging.getLogger(__name__)
 
 
 class CredsmashProxy(object):
-    def __init__(self, key_service, storage_service, key_fmt):
+    def __init__(self, key_service, storage_service, key_fmt, encoding='utf-8', errors='strict'):
         self._key_service = key_service
         self._storage_service = storage_service
         self._key_fmt = key_fmt
         self._data = {}
+        self._encoding = encoding
+        self._errors = errors
 
     def __getitem__(self, key):
+        # type: (str) -> str
+        return self.get_str(key)
+
+    def get_str(self, key):
+        # type: (str) -> str
+        b = self.get_bytes(key)
+        return b.decode(self._encoding, self._errors)
+
+    def get_bytes(self, key):
+        # type: (str) -> bytes
         if key in self._data:
             return self._data[key]
 
@@ -33,28 +46,43 @@ def __getitem__(self, key):
         else:
             lookup_key = self._key_fmt.format(key)
         logger.debug('key=%s lookup_key=%s', key, lookup_key)
-        res = credsmash.api.get_secret(
-            self._storage_service,
-            self._key_service,
-            key,
-        )
+        try:
+            res = credsmash.api.get_secret(
+                self._storage_service,
+                self._key_service,
+                key,
+            )
+        except ItemNotFound:
+            raise KeyError(repr(key))
         self._data[key] = res
         return res
 
     def __contains__(self, key):
         try:
-            self.__getitem__(key)
+            self.get_bytes(key)
             return True
-        except ItemNotFound:
+        except KeyError:
             return False
 
 
 class DictProxy(object):
-    def __init__(self, items, key_fmt):
-        self._items = items
+    def __init__(self, items, key_fmt, encoding='utf-8', errors='strict'):
+        self._items = items  # type: Dict[str,bytes]
         self._key_fmt = key_fmt
+        self._encoding = encoding
+        self._errors = errors
 
     def __getitem__(self, key):
+        # type: (str) -> str
+        return self.get_str(key)
+
+    def get_str(self, key):
+        # type: (str) -> str
+        b = self.get_bytes(key)
+        return b.decode(self._encoding, self._errors)
+
+    def get_bytes(self, key):
+        # type: (str) -> bytes
         if isinstance(key, tuple):
             lookup_key = self._key_fmt.format(*key)
         else:
@@ -63,7 +91,7 @@ def __getitem__(self, key):
 
     def __contains__(self, key):
         try:
-            self.__getitem__(key)
+            self.get_bytes(key)
             return True
         except KeyError:
             return False
@@ -77,7 +105,7 @@ def __contains__(self, key):
 @click.option('--key-fmt', default='{0}',
               help='Re-use templates by tweaking which variable it maps to- '
                    'eg, "dev.{0}" converts {{secrets.potato}} to the secret "dev.potato"')
-@click.option('--template-vars', type=click.File(mode='rb'))
+@click.option('--template-vars', type=click.File(mode='r', encoding='utf-8'))
 @click.option('--template-vars-format', default=None)
 @click.option('--secrets-file', type=click.File(mode='rb'),
               help="Source from a local file instead of credential store "
@@ -105,11 +133,11 @@ def cmd_render_template(
             key_fmt,
         )
 
-    render_args = {}
+    render_args = {}  # type: Dict[str,Union[str,DictProxy,CredsmashProxy]]
     if template_vars:
         if not template_vars_format:
             template_vars_format = detect_format(template_vars, 'json')
-        render_args = read_many(template_vars, template_vars_format)
+        render_args = read_many_str(template_vars, template_vars_format)
     if obj_name in render_args:
         logger.warning('Overwrote %r from template vars with secrets var.', obj_name)
     render_args[obj_name] = secrets
@@ -127,7 +155,7 @@ def cmd_render_template(
 @click.option('--key-fmt', default='{0}',
               help='Re-use templates by tweaking which variable it maps to- '
                    'eg, "dev.{0}" converts {{secrets.potato}} to the secret "dev.potato"')
-@click.option('--template-vars', type=click.File(mode='rb'))
+@click.option('--template-vars', type=click.File(mode='r', encoding='utf-8'))
 @click.option('--template-vars-format', default=None)
 @click.option('--secrets-file', type=click.File(mode='rb'),
               help="Source from a local file instead of credential store "
@@ -155,11 +183,11 @@ def cmd_render_template(
             key_fmt,
         )
 
-    render_args = {}
+    render_args = {}  # type: Dict[str,Union[str,DictProxy,CredsmashProxy]]
     if template_vars:
         if not template_vars_format:
             template_vars_format = detect_format(template_vars, 'json')
-        render_args = read_many(template_vars, template_vars_format)
+        render_args = read_many_str(template_vars, template_vars_format)
     if obj_name in render_args:
         logger.warning('Overwrote %r from template vars with secrets var.', obj_name)
     render_args[obj_name] = secrets
@@ -168,31 +196,33 @@ def cmd_render_template(
     if not manifest_format:
         manifest_format = detect_format(manifest, 'json')
     for entry in parse_manifest(manifest, manifest_format):
+        destination_path = os.path.realpath(entry['destination'])
         if 'source' in entry:
-            with codecs.open(entry['source'], 'r', encoding='utf-8') as template:
+            source_path = os.path.realpath(entry['source'])
+            with codecs.open(source_path, 'r', encoding='utf-8') as template:
                 output = env.from_string(template.read()).render(render_args)
             # Only open the file after rendering the template
             #  as we truncate the file when opening.
-            with codecs.open(entry['destination'], 'w', encoding='utf-8') as destination:
+            with codecs.open(destination_path, 'w', encoding='utf-8') as destination:
                 destination.write(output)
             logger.info('Rendered template="%s" destination="%s"', entry['source'], entry['destination'])
         elif 'secret' in entry:
-            output = secrets[entry['secret']]
-            with open(entry['destination'], 'wb') as destination:
+            output = secrets.get_bytes(entry['secret'])
+            with open(destination_path, 'wb') as destination:
                 destination.write(output)
             logger.info('Wrote secret="%s" destination="%s"', entry['secret'], entry['destination'])
         else:
             raise RuntimeError('Manifest entry must contain a secret or source')
 
         if 'mode' in entry:
             os.chmod(
-                entry['destination'],
+                destination_path,
                 entry['mode']
             )
 
         if 'owner' in entry and 'group' in entry:
             os.chown(
-                entry['destination'],
+                destination_path,
                 pwd.getpwnam(entry['owner']).pw_uid,
                 grp.getgrnam(entry['group']).gr_gid
             )