debianmemberportfolio/ddportfolioservice/model/keyringanalyzer.py

# -*- python -*-
# -*- coding: utf-8 -*-
#
# DDPortfolio service application key ring analyzer tool
# Copyright (c) 2009 Jan Dittberner <jan@dittberner.info>
#
# This file is part of DDPortfolio service.
#
# DDPortfolio service is free software: you can redistribute it and/or
# modify it under the terms of the GNU Affero General Public License
# as published by the Free Software Foundation, either version 3 of
# the License, or (at your option) any later version.
#
# DDPortfolio service is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public
# License along with this program.  If not, see
# <http://www.gnu.org/licenses/>.
#
"""
This is a tool that analyzes GPG and PGP keyrings and stores the
retrieved data in a file database. The tool was inspired by Debian
qa's carnivore.
"""

import anydbm
import pkg_resources
import glob
import ConfigParser
import os


def _get_keyrings():
    """Gets the available keyring files from the keyring directory
    configured in ddportfolio.ini."""
    my_config = ConfigParser.ConfigParser()
    my_config.readfp(pkg_resources.resource_stream(__name__, 'ddportfolio.ini'))
    keyrings = glob.glob(my_config.get('DEFAULT', 'keyring.dir') +
                         '/*.gpg')
    keyrings.extend(glob.glob(my_config.get('DEFAULT', 'keyring.dir') +
                              '/*.pgp'))
    keyrings.sort()
    return keyrings


def _parse_uid(uid):
    """Parse a uid of the form 'Real Name <email@example.com>' into
    email and realname parts."""
    uid = uid.strip()
	# First, strip comment
    s = uid.find('(')
    e = uid.find(')')
    if s >= 0 and e >= 0:
        uid = uid[:s] + uid[e+1:]
    s = uid.find('<')
    e = uid.find('>')
    email = None
    if s >= 0 and e >= 0:
        email = uid[s+1:e]
        uid = uid[:s] + uid[e+1:]
    uid = uid.strip()
    if not email and uid.find('@') >= 0:
        email, uid = uid, email
    return (uid, email)

resultdict = {}

def _get_canonical(key):
    if not key in resultdict:
        resultdict[key] = []
    return key


def _add_to_result(key, newvalue):
    thekey = _get_canonical(key)
    if newvalue not in resultdict[thekey]:
        resultdict[thekey].append(newvalue)


def process_keyrings():
    """Process the keyrings and store the extracted data in an anydbm
    file."""
    for keyring in _get_keyrings():
        contents = os.popen("gpg --no-default-keyring \
                --no-expensive-trust-checks \
                --keyring %s --list-keys \
                --with-colons --fingerprint" % (keyring))
        fpr = None
        entry = None
        lastpub = None
        for line in contents.readlines():
            items = line.split(':')
            uid = None
            if items[0] == 'pub':
                fpr = entry = None
                lastpub = items[9].strip()
                continue
            elif items[0] == 'fpr':
                fpr = items[9].strip()
                uid = lastpub
            elif items[0] == 'uid':
                uid = items[9].strip()
            else:
                continue
            # Do stuff with 'uid'
            uid, email = _parse_uid(uid)
            if email:
                if email.endswith('@debian.org'):
                    login = email[0:-len('@debian.org')]
                    _add_to_result('login:email:%s' % email, login)
                    _add_to_result('login:fpr:%s' % fpr, login)
                _add_to_result('email:fpr:%s' % fpr, email)
            if uid:
                _add_to_result('name:fpr:%s' % fpr, uid)
                if email:
                    _add_to_result('name:email:%s' % email, uid)
        contents.close()
    db = anydbm.open(pkg_resources.resource_filename(__name__,
                                                     'keyringcache'), 'c')
    for key in resultdict:
        db[key] = ":".join(resultdict[key])
    db.close()


if __name__ == '__main__':
    process_keyrings()