Commit d8bf566d authored by Philipp Hörist's avatar Philipp Hörist

[openpgp] Add GPGME backend

parent a2622aa6
......@@ -15,180 +15,169 @@
# along with OpenPGP Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.
import io
from collections import namedtuple
import logging
from nbxmpp.protocol import JID
import gpg
from gpg.results import ImportResult
from openpgp.modules.util import DecryptionFailed
log = logging.getLogger('gajim.p.openpgp.gpgme')
class KeyringItem:
def __init__(self, key):
self._key = key
self._uid = self._get_uid()
from gajim.common import app
def _get_uid(self):
for uid in self._key.uids:
if uid.uid.startswith('xmpp:'):
return uid.uid
KeyringItem = namedtuple('KeyringItem',
'type keyid userid fingerprint')
@property
def fingerprint(self):
return self._key.fpr
log = logging.getLogger('gajim.p.openpgp.pgpme')
@property
def uid(self):
if self._uid is not None:
return self._uid
@property
def jid(self):
if self._uid is not None:
return JID.from_string(self._uid[5:])
class PGPContext():
def __hash__(self):
return hash(self.fingerprint)
class GPGME:
def __init__(self, jid, gnuhome):
self.context = gpg.Context(home_dir=str(gnuhome))
# self.create_new_key()
# self.get_key_by_name()
# self.get_key_by_fingerprint()
self.export_public_key()
def create_new_key(self):
parms = """<GnupgKeyParms format="internal">
Key-Type: RSA
Key-Length: 2048
Subkey-Type: RSA
Subkey-Length: 2048
Name-Real: Joe Tester
Name-Comment: with stupid passphrase
Name-Email: test@example.org
Passphrase: Crypt0R0cks
Expire-Date: 2020-12-31
</GnupgKeyParms>
"""
with self.context as c:
c.set_engine_info(gpg.constants.PROTOCOL_OpenPGP, None, app.gajimpaths['MY_DATA'])
c.set_progress_cb(gpg.callbacks.progress_stdout)
c.op_genkey(parms, None, None)
print("Generated key with fingerprint {0}.".format(
c.op_genkey_result().fpr))
def get_all_keys(self):
c = gpg.Context()
for key in c.keylist():
user = key.uids[0]
print("Keys for %s (%s):" % (user.name, user.email))
self._jid = jid
self._context_args = {
'home_dir': str(gnuhome),
'offline': True,
'armor': False,
}
def generate_key(self):
with gpg.Context(**self._context_args) as context:
result = context.create_key(f'xmpp:{str(self._jid)}',
expires=False,
sign=True,
encrypt=True,
certify=False,
authenticate=False,
passphrase=None,
force=False)
log.info('Generated new key: %s', result.fpr)
def get_key(self, fingerprint):
with gpg.Context(**self._context_args) as context:
try:
key = context.get_key(fingerprint)
except gpg.errors.KeyNotFound as error:
log.warning('key not found: %s', error.keystr)
return
except Exception as error:
log.warning('get_key() error: %s', error)
return
return key
def get_own_key_details(self):
with gpg.Context(**self._context_args) as context:
keys = list(context.keylist(secret=True))
if not keys:
return None, None
key = keys[0]
for subkey in key.subkeys:
features = []
if subkey.can_authenticate:
features.append('auth')
if subkey.can_certify:
features.append('cert')
if subkey.can_encrypt:
features.append('encrypt')
if subkey.can_sign:
features.append('sign')
print(' %s %s' % (subkey.fpr, ','.join(features)))
def get_key_by_name(self):
c = gpg.Context()
for key in c.keylist('john'):
print(key.subkeys[0].fpr)
def get_key_by_fingerprint(self):
c = gpg.Context()
fingerprint = 'key fingerprint to search for'
try:
key = c.get_key(fingerprint)
print('%s (%s)' % (key.uids[0].name, key.uids[0].email))
except gpg.errors.KeyNotFound:
print("No key for fingerprint '%s'." % fingerprint)
def get_secret_key(self):
'''
Key(can_authenticate=1,
can_certify=1,
can_encrypt=1,
can_sign=1,
chain_id=None,
disabled=0,
expired=0,
fpr='7ECE1F88BAFCA37F168E1556A4DBDD1BA55FE3CE',
invalid=0,
is_qualified=0,
issuer_name=None,
issuer_serial=None,
keylist_mode=1,
last_update=0,
origin=0,
owner_trust=5,
protocol=0,
revoked=0,
secret=1,
subkeys=[
SubKey(can_authenticate=1,
can_certify=1,
can_encrypt=1,
can_sign=1,
card_number=None
curve=None,
disabled=0,
expired=0,
expires=0,
fpr='7ECE1F88BAFCA37F168E1556A4DBDD1BA55FE3CE',
invalid=0,
is_cardkey=0,
is_de_vs=1,
is_qualified=0,
keygrip='15BECB77301E4810ABB9CA6A9391158E575DABEC',
keyid='A4DBDD1BA55FE3CE',
length=2048,
pubkey_algo=1,
revoked=0,
secret=1,
timestamp=1525006759)],
uids=[
UID(address=None,
comment='',
email='',
invalid=0,
last_update=0,
name='xmpp:philw@jabber.at',
origin=0,
revoked=0,
signatures=[],
tofu=[],
uid='xmpp:philw@jabber.at',
validity=5)])
'''
for key in self.context.keylist(secret=True):
break
return key.fpr, key.fpr[-16:]
def get_keys(self, secret=False):
if subkey.fpr == key.fpr:
return subkey.fpr, subkey.timestamp
return None, None
def get_keys(self):
keys = []
for key in self.context.keylist():
for uid in key.uids:
if uid.uid.startswith('xmpp:'):
keys.append((key, uid.uid[5:]))
break
with gpg.Context(**self._context_args) as context:
for key in context.keylist():
keys.append(KeyringItem(key))
return keys
def export_public_key(self):
# print(dir(self.context))
result = self.context.key_export_minimal()
print(result)
def encrypt_decrypt_files(self):
c = gpg.Context()
recipient = c.get_key("fingerprint of recipient's key")
# Encrypt
with open('foo.txt', 'r') as input_file:
with open('foo.txt.gpg', 'wb') as output_file:
c.encrypt([recipient], 0, input_file, output_file)
# Decrypt
with open('foo.txt.gpg', 'rb') as input_file:
with open('foo2.txt', 'w') as output_file:
c.decrypt(input_file, output_file)
def encrypt(self):
c = gpg.Context()
recipient = c.get_key("fingerprint of recipient's key")
plaintext_string = u'plain text data'
plaintext_bytes = io.BytesIO(plaintext_string.encode('utf8'))
encrypted_bytes = io.BytesIO()
c.encrypt([recipient], 0, plaintext_bytes, encrypted_bytes)
def decrypt(self):
c = gpg.Context()
decrypted_bytes = io.BytesIO()
c.decrypt(encrypted_bytes, decrypted_bytes)
decrypted_string = decrypted_bytes.getvalue().decode('utf8')
def export_key(self, fingerprint):
with gpg.Context(**self._context_args) as context:
key = context.key_export_minimal(pattern=fingerprint)
return key
# def encrypt_decrypt_files(self):
# c = gpg.Context()
# recipient = c.get_key("fingerprint of recipient's key")
# # Encrypt
# with open('foo.txt', 'r') as input_file:
# with open('foo.txt.gpg', 'wb') as output_file:
# c.encrypt([recipient], 0, input_file, output_file)
# # Decrypt
# with open('foo.txt.gpg', 'rb') as input_file:
# with open('foo2.txt', 'w') as output_file:
# c.decrypt(input_file, output_file)
def encrypt(self, plaintext, keys):
recipients = []
with gpg.Context(**self._context_args) as context:
for key in keys:
key = context.get_key(key.fingerprint)
if key is not None:
recipients.append(key)
if not recipients:
return None, 'No keys found to encrypt to'
with gpg.Context(**self._context_args) as context:
result = context.encrypt(str(plaintext).encode(),
recipients,
always_trust=True)
ciphertext, result, _sign_result = result
return ciphertext, None
def decrypt(self, ciphertext):
with gpg.Context(**self._context_args) as context:
try:
result = context.decrypt(ciphertext)
except Exception as error:
raise DecryptionFailed('Decryption failed: %s' % error)
plaintext, result, verify_result = result
plaintext = plaintext.decode()
fingerprints = [sig.fpr for sig in verify_result.signatures]
if not fingerprints or len(fingerprints) > 1:
log.error(result)
log.error(verify_result)
raise DecryptionFailed('Verification failed')
return plaintext, fingerprints[0]
def import_key(self, data, jid):
log.info('Import key from %s', jid)
with gpg.Context(**self._context_args) as context:
result = context.key_import(data)
if not isinstance(result, ImportResult) or result.imported != 1:
log.error('Key import failed: %s', jid)
log.error(result)
return
fingerprint = result.imports[0].fpr
key = self.get_key(fingerprint)
return KeyringItem(key)
......@@ -31,7 +31,7 @@ if log.getEffectiveLevel() == logging.DEBUG:
KeyringItem = namedtuple('KeyringItem', 'jid keyid fingerprint')
class PGPContext(gnupg.GPG):
class PythonGnuPG(gnupg.GPG):
def __init__(self, jid, gnupghome):
gnupg.GPG.__init__(
self, gpgbinary='gpg', gnupghome=str(gnupghome))
......
......@@ -93,12 +93,6 @@ class KeyWizard(Gtk.Assistant):
elif self.get_current_page() == Page.SUCCESS:
self._activate_encryption()
def _on_error(self, error_text):
log.info('Show Error page')
page = self.get_nth_page(Page.ERROR)
page.set_text(error_text)
self.set_current_page(Page.ERROR)
def _on_cancel(self, widget):
self.destroy()
......@@ -168,22 +162,23 @@ class NewKeyPage(RequestPage):
thread.start()
def worker(self):
error = None
text = None
try:
self._con.get_module('OpenPGP').generate_key()
except Exception as e:
error = e
else:
self._con.get_module('OpenPGP').get_own_key_details()
self._con.get_module('OpenPGP').set_public_key()
self._con.get_module('OpenPGP').request_keylist()
GLib.idle_add(self.finished, error)
except Exception as error:
text = str(error)
GLib.idle_add(self.finished, text)
def finished(self, error):
if error is None:
self._assistant.set_current_page(Page.SUCCESS)
self._con.get_module('OpenPGP').get_own_key_details()
self._con.get_module('OpenPGP').set_public_key()
self._con.get_module('OpenPGP').request_keylist()
else:
log.error(error)
error_page = self._assistant.get_nth_page(Page.ERROR)
error_page.set_text(error)
self._assistant.set_current_page(Page.ERROR)
......@@ -229,7 +224,7 @@ class SuccessfulPage(Gtk.Box):
class ErrorPage(Gtk.Box):
type_ = Gtk.AssistantPageType.SUMMARY
title = _('Registration failed')
title = _('Setup failed')
complete = True
def __init__(self):
......
......@@ -14,6 +14,7 @@
# You should have received a copy of the GNU General Public License
# along with OpenPGP Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.
import sys
import time
import logging
from pathlib import Path
......@@ -44,7 +45,11 @@ from openpgp.modules.util import DecryptionFailed
from openpgp.modules.util import prepare_stanza
from openpgp.modules.key_store import PGPContacts
from openpgp.backend.sql import Storage
from openpgp.backend.pygpg import PGPContext
if sys.platform == 'win32':
from openpgp.backend.pygpg import PythonGnuPG as PGPBackend
else:
from openpgp.backend.gpgme import GPGME as PGPBackend
log = logging.getLogger('gajim.p.openpgp')
......@@ -84,9 +89,9 @@ class OpenPGP(BaseModule):
own_bare_jid = self.own_jid.getBare()
path = Path(configpaths.get('MY_DATA')) / 'openpgp' / own_bare_jid
if not path.exists():
path.mkdir(parents=True)
path.mkdir(mode=0o700, parents=True)
self._pgp = PGPContext(self.own_jid, path)
self._pgp = PGPBackend(self.own_jid, path)
self._storage = Storage(path)
self._contacts = PGPContacts(self._pgp, self._storage)
self._fingerprint, self._date = self.get_own_key_details()
......@@ -223,7 +228,14 @@ class OpenPGP(BaseModule):
if not any(map(self.own_jid.bareMatch, recipients)):
log.warning('to attr not valid')
log.warning(payload)
log.warning(str(payload))
return
keys = self._contacts.get_keys(properties.jid.bare)
fingerprints = [key.fingerprint for key in keys]
if fingerprint not in fingerprints:
log.warning('Invalid fingerprint on message: %s', fingerprint)
log.warning('Expected: %s', fingerprints)
return
log.info('Received OpenPGP message from: %s', properties.jid)
......
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