Jan Dittberner
9af353765d
- add logging to ddportfolioservice/model/keyfinder.py and
ddportfolioservice/model/keyringanalyzer.py
- use ~ instead of ${HOME} in ddportfolioservice/model/ddportfolio.ini
- use subprocess.Popen instead of os.popen
140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
# -*- python -*-
|
|
# -*- coding: utf-8 -*-
|
|
#
|
|
# DDPortfolio service application key ring analyzer tool
|
|
# Copyright © 2009, 2010 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
|
|
import os.path
|
|
import logging
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
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'))
|
|
keyringdir = os.path.expanduser(my_config.get('DEFAULT', 'keyring.dir'))
|
|
logging.debug("keyring dir is %s", keyringdir)
|
|
keyrings = glob.glob(os.path.join(keyringdir, '*.gpg'))
|
|
keyrings.extend(glob.glob(os.path.join(keyringdir, '*.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():
|
|
logging.debug("get data from %s", keyring)
|
|
proc = subprocess.Popen(["gpg", "--no-default-keyring",
|
|
"--no-expensive-trust-checks",
|
|
"--keyring", keyring, "--list-keys",
|
|
"--with-colons", "--fingerprint"],
|
|
stdout=subprocess.PIPE)
|
|
fpr = None
|
|
entry = None
|
|
lastpub = None
|
|
for line in proc.stdout.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('fpr:login:%s' % login, fpr)
|
|
_add_to_result('fpr:email:%s' % email, fpr)
|
|
_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)
|
|
retcode = proc.wait()
|
|
if retcode != 0:
|
|
logging.error("subprocess ended with return code %d", retcode)
|
|
db = anydbm.open(pkg_resources.resource_filename(__name__,
|
|
'keyringcache'), 'c')
|
|
for key in resultdict:
|
|
db[key] = ":".join(resultdict[key])
|
|
db.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
logging.basicConfig(stream=sys.stderr, level=logging.WARNING)
|
|
process_keyrings()
|