Commit fd4acbc7 authored by Vladislav Yarmak's avatar Vladislav Yarmak Committed by Philipp Hörist

[pgp] Use full key fingerprints instead of short ids

parent 9da3d02b
......@@ -105,7 +105,7 @@ class PGP(gnupg.GPG, metaclass=Singleton):
)
result = super().verify(data.encode('utf8'))
if result.valid:
return result.key_id
return result.fingerprint
def get_key(self, key_id):
return super().list_keys(keys=[key_id])
......@@ -116,7 +116,7 @@ class PGP(gnupg.GPG, metaclass=Singleton):
for key in result:
# Take first not empty uid
keys[key['keyid'][8:]] = [uid for uid in key['uids'] if uid][0]
keys[key['fingerprint']] = next(uid for uid in key['uids'] if uid)
return keys
@staticmethod
......
......@@ -19,17 +19,19 @@ from pathlib import Path
from gajim.common import app
from gajim.common import configpaths
from gajim.common.helpers import delay_execution
CURRENT_STORE_VERSION = 3
class KeyResolveError(Exception):
pass
class KeyStore:
def __init__(self, account, own_jid, log):
def __init__(self, account, own_jid, log, list_keys_func):
self._list_keys_func = list_keys_func
self._log = log
self._account = account
self._store = {
'own_key_data': None,
'contact_key_data': {},
}
own_bare_jid = own_jid.getBare()
path = Path(configpaths.get('PLUGINS_DATA')) / 'pgplegacy' / own_bare_jid
......@@ -38,16 +40,41 @@ class KeyStore:
self._store_path = path / 'store'
if self._store_path.exists():
# having store v2 or higher
with self._store_path.open('r') as file:
try:
self._store = json.load(file)
except Exception:
log.exception('Could not load config')
self._store = self._empty_store()
ver = self._store.get('_version', 2)
if ver > CURRENT_STORE_VERSION:
raise Exception('Unknown store version! '
'Please upgrade pgp plugin.')
elif ver == 2:
self._migrate_v2_store()
self._save_store()
elif ver != CURRENT_STORE_VERSION:
# garbled version
self._store = self._empty_store()
log.warning('Bad pgp key store version. Initializing new.')
else:
# having store v1 or fresh install
self._store = self._empty_store()
self._migrate_v1_store()
self._migrate_v2_store()
self._save_store()
@staticmethod
def _empty_store():
return {
'_version': CURRENT_STORE_VERSION,
'own_key_data': None,
'contact_key_data': {},
}
if not self._store['contact_key_data']:
self._migrate()
def _migrate(self):
def _migrate_v1_store(self):
keys = {}
attached_keys = app.config.get_per(
'accounts', self._account, 'attached_gpg_keys')
......@@ -59,18 +86,42 @@ class KeyStore:
keys[attached_keys[2 * i]] = attached_keys[2 * i + 1]
for jid, key_id in keys.items():
self.set_contact_key_data(jid, (key_id, ''))
self._set_contact_key_data_nosync(jid, (key_id, ''))
own_key_id = app.config.get_per('accounts', self._account, 'keyid')
own_key_user = app.config.get_per('accounts', self._account, 'keyname')
if own_key_id:
self.set_own_key_data((own_key_id, own_key_user))
self._set_own_key_data_nosync((own_key_id, own_key_user))
attached_keys = app.config.set_per(
'accounts', self._account, 'attached_gpg_keys', '')
self._log.info('Migration successful')
self._log.info('Migration from store v1 was successful')
def _migrate_v2_store(self):
own_key_data = self.get_own_key_data()
if own_key_data is not None:
own_key_id, own_key_user = (own_key_data['key_id'],
own_key_data['key_user'])
try:
own_key_fp = self._resolve_short_id(own_key_id, has_secret=True)
self._set_own_key_data_nosync((own_key_fp, own_key_user))
except KeyResolveError:
self._set_own_key_data_nosync(None)
prune_list = []
for dict_key, key_data in self._store['contact_key_data'].items():
try:
key_data['key_id'] = self._resolve_short_id(key_data['key_id'])
except KeyResolveError:
prune_list.append[dict_key]
for dict_key in prune_list:
del self._store['contact_key_data'][dict_key]
self._store['_version'] = CURRENT_STORE_VERSION
self._log.info('Migration from store v2 was successful')
@delay_execution(500)
def _save_store(self):
with self._store_path.open('w') as file:
json.dump(self._store, file)
......@@ -78,7 +129,26 @@ class KeyStore:
def _get_dict_key(self, jid):
return '%s-%s' % (self._account, jid)
def _resolve_short_id(self, short_id, has_secret=False):
candidates = self._list_keys_func(
secret=has_secret, keys=(short_id,)).fingerprints
if len(candidates) == 1:
return candidates[0]
elif len(candidates) > 1:
self._log.critical('Key collision during migration. '
'Key ID is %s. Removing binding...',
repr(short_id))
else:
self._log.warning('Key %s was not found during migration. '
'Removing binding...',
repr(short_id))
raise KeyResolveError
def set_own_key_data(self, key_data):
self._set_own_key_data_nosync(key_data)
self._save_store()
def _set_own_key_data_nosync(self, key_data):
if key_data is None:
self._store['own_key_data'] = None
else:
......@@ -86,7 +156,6 @@ class KeyStore:
'key_id': key_data[0],
'key_user': key_data[1]
}
self._save_store()
def get_own_key_data(self):
return self._store['own_key_data']
......@@ -97,13 +166,16 @@ class KeyStore:
return key_ids.get(dict_key)
def set_contact_key_data(self, jid, key_data):
self._set_contact_key_data_nosync(jid, key_data)
self._save_store()
def _set_contact_key_data_nosync(self, jid, key_data):
key_ids = self._store['contact_key_data']
dict_key = self._get_dict_key(jid)
if key_data is None:
self._store['contact_key_data'][dict_key] = None
key_ids[dict_key] = None
else:
key_ids[dict_key] = {
'key_id': key_data[0],
'key_user': key_data[1]
}
self._save_store()
......@@ -74,8 +74,9 @@ class PGPLegacy(BaseModule):
self.own_jid = self._con.get_own_jid()
self._store = KeyStore(self._account, self.own_jid, self._log)
self._pgp = PGP()
self._store = KeyStore(self._account, self.own_jid, self._log,
self._pgp.list_keys)
self._always_trust = []
self._presence_key_id_store = {}
......@@ -118,7 +119,7 @@ class PGPLegacy(BaseModule):
if key_id is None:
return
self._presence_key_id_store[jid] = key_id[8:]
self._presence_key_id_store[jid] = key_id
key_data = self.get_contact_key_data(jid)
if key_data is not None:
......@@ -131,7 +132,7 @@ class PGPLegacy(BaseModule):
return
self._log.info('Assign key-id: %s to %s', key_id, jid)
self.set_contact_key_data(jid, (key_id[8:], key[0]['uids'][0]))
self.set_contact_key_data(jid, (key_id, key[0]['uids'][0]))
def _message_received(self, _con, stanza, properties):
if not properties.is_pgp_legacy or properties.from_muc:
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment