Disroot/gpg-lacre
8
15
Fork 4
Code Issues 36 Pull requests 1 Releases 3 Wiki Activity

Split webgate-cron.py into small functions

Browse source
This commit is contained in:
Piotr F. Mieszkowski 2024-01-21 11:28:46 +01:00
parent bfd3541b18
commit be615df6e4
1 changed files with 53 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

88
webgate-cron.py
View file

@ -42,6 +42,49 @@ def _validate_config():
exit(lacre.EX_CONFIG)
def import_key(key_dir, armored_key, key_id, email, key_queue, identities):
# import the key to gpg
(fingerprint, _) = GnuPG.add_key(key_dir, armored_key)
key_queue.mark_accepted(key_id)
identities.register_or_update(email, fingerprint)
LOG.info('Imported key from: %s', email)
if conf.flag_enabled('cron', 'send_email'):
notify("PGP key registration successful", "registrationSuccess.md", email)
def import_failed(key_id, email, key_queue):
key_queue.delete_keys(key_id)
LOG.warning('Import confirmation failed: %s', email)
if conf.flag_enabled('cron', 'send_email'):
notify("PGP key registration failed", "registrationError.md", email)
def delete_key(key_id, email, key_queue):
# delete key so we don't continue processing it
LOG.debug('Empty key received, just deleting')
key_queue.delete_keys(row_id)
if conf.flag_enabled('cron', 'send_email'):
notify("PGP key deleted", "keyDeleted.md", email)
def cleanup(key_dir, key_queue):
"""Delete keys and queue entries."""
LOG.info('Cleaning up after a round of key confirmation')
for email, row_id in key_queue.fetch_keys_to_delete():
LOG.debug('Removing key from keyring: %s', email)
GnuPG.delete_key(key_dir, email)
LOG.debug('Removing key from identity store: %s', row_id)
key_queue.delete_keys(row_id)
LOG.info('Deleted key for: %s', email)
_validate_config()
if not (conf.flag_enabled('database', 'enabled') and conf.config_item_set('database', 'url')):
@ -59,48 +102,23 @@ try:
key_dir = conf.get_item('gpg', 'keyhome')
LOG.debug('Using GnuPG with home directory in %s', key_dir)
result_set = key_queue.fetch_keys()
for armored_key, row_id, email in result_set:
for armored_key, row_id, email in key_queue.fetch_keys():
# delete any other public keys associated with this confirmed email address
key_queue.delete_keys(row_id, email=email)
identities.delete(email)
GnuPG.delete_key(key_dir, email)
LOG.info('Deleted key for <%s> via import request', email)
LOG.info('Deleted key via import request for: %s', email)
if armored_key.strip(): # we have this so that user can submit blank key to remove any encryption
if GnuPG.confirm_key(armored_key, email):
# import the key to gpg
(fingerprint, _) = GnuPG.add_key(key_dir, armored_key)
key_queue.mark_accepted(row_id)
identities.register_or_update(email, fingerprint)
LOG.info('Imported key from <%s>', email)
if conf.flag_enabled('cron', 'send_email'):
notify("PGP key registration successful", "registrationSuccess.md", email)
else:
key_queue.delete_keys(row_id)
LOG.warning('Import confirmation failed for <%s>', email)
if conf.flag_enabled('cron', 'send_email'):
notify("PGP key registration failed", "registrationError.md", email)
else:
if not armored_key.strip(): # we have this so that user can submit blank key to remove any encryption
# delete key so we don't continue processing it
LOG.debug('Empty key received, just deleting')
delete_key(row_id, email, key_queue)
continue
key_queue.delete_keys(row_id)
if conf.flag_enabled('cron', 'send_email'):
notify("PGP key deleted", "keyDeleted.md", email)
if GnuPG.confirm_key(armored_key, email):
import_key(key_dir, armored_key, row_id, email, key_queue, identities)
else:
import_failed(row_id, email, key_queue)
LOG.info('Cleaning up after a round of key confirmation')
for email, row_id in key_queue.fetch_keys_to_delete():
LOG.debug('Removing key from keyring: %s', email)
GnuPG.delete_key(key_dir, email)
LOG.debug('Removing key from identity store: %s', row_id)
key_queue.delete_keys(row_id)
LOG.info('Deleted key for <%s>', email)
cleanup(key_dir, key_queue)
except:
LOG.exception('Unexpected issue during key confirmation')