Import KeePassX 2 CSV into pass

keepassx2csv2pass.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Import KeePassX 2 CSV into pass.
"""

import argparse
import collections
import csv
import logging
import os.path
import subprocess
import sys

try:
    # Python 3
    from builtins import str
except ImportError:
    # Make code backwards compatible with Python 2.
    str = unicode

try:
    # Python 3
    from subprocess import DEVNULL
except ImportError:
    import os
    DEVNULL = open(os.devnull, 'wb')


logging.basicConfig(level=logging.WARNING, format='%(message)s')
logger = logging.getLogger(__name__)


# A KeepassX database record.
KeepassXRecord = collections.namedtuple(
    'KeepassXRecord',
    ['Group', 'Notes', 'Password', 'Title', 'URL', 'Username',],
)

# A pass database record.
BasePassRecord = collections.namedtuple(
    'BasePassRecord',
    ['path', 'username', 'password', 'url', 'notes'],
)

class PassRecord(BasePassRecord):

    # Use parent dict.
    __slots__ = ()

    @classmethod
    def from_keepassx(cls, record):
        """
        Convert a KeepassX record into a pass record.
        """
        path = os.path.join(record.Group, record.Title)
        username = record.Username
        password = record.Password
        url = record.URL
        notes = record.Notes
        return cls(path, username, password, url, notes)

    def __str__(self):
        return self.__unicode__()

    def __unicode__(self):
        """
        Returns a multi-line pass record.
        """
        template = [u'{password}']
        if self.username:
            template.append(u'Username: {username}')
        if self.url:
            template.append(u'URL: {url}')
        if self.notes:
            template.append(u'{notes}')
        return u'\n'.join(template).format(**self._asdict()) + u'\n'


def parse_records(data):
    """
    Parse CSV entries into pass records.

    Yields: PassRecord
    """
    def utf8_dict_reader(utf8_data, **kwargs):
        """
        Unicode-enabled DictReader.

        https://stackoverflow.com/a/5005573
        """
        csv_reader = csv.DictReader(utf8_data, **kwargs)
        for row in csv_reader:
            yield {key: str(value, 'utf-8') for key, value in list(row.items())}

    if sys.version_info > (3,):
        reader = csv.DictReader
    else:
        reader = utf8_dict_reader

    for row in reader(data):
        yield PassRecord.from_keepassx(KeepassXRecord(**row))


def import_into_pass(record, force=False):
    """
    Import a password into pass.

    Returns: None

    Raises: CalledProcessError if calling `pass` fails.
    """
    command = ['pass', 'insert', '-m']
    if force:
        command.append('-f')
    command.append(record.path)
    p = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=DEVNULL,
                         stderr=subprocess.PIPE)
    _, stderr = p.communicate(str(record).encode('utf-8'))
    if stderr:
        logger.error(stderr)


def parse_args(argv):
    """
    Parse command-line arguments.

    Returns: argparse.Namespace
    """
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        '-v', '--verbose',
        action='store_true', help='print verbose output',
    )
    parser.add_argument('infile', metavar='FILE', type=argparse.FileType('r'))
    return parser.parse_args(argv)


def main(argv=None):
    if not argv:
        argv = sys.argv[1:]

    args = parse_args(argv)

    if args.verbose:
        logger.setLevel(logging.INFO)

    count = 0

    for record in parse_records(args.infile):
        try:
            logger.info('importing %s into password store', record.path)
            import_into_pass(record)
        except Exception as e:
            logger.error(e)
            raise
            return 1
        count += 1

    logger.info('imported %s records into password store', count)


if __name__ == '__main__':
    sys.exit(main())