summaryrefslogtreecommitdiff
path: root/src/tabs/rostertab.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/tabs/rostertab.py')
-rw-r--r--src/tabs/rostertab.py166
1 files changed, 166 insertions, 0 deletions
diff --git a/src/tabs/rostertab.py b/src/tabs/rostertab.py
index 52a3a88c..846ba327 100644
--- a/src/tabs/rostertab.py
+++ b/src/tabs/rostertab.py
@@ -10,9 +10,11 @@ from gettext import gettext as _
import logging
log = logging.getLogger(__name__)
+import base64
import curses
import difflib
import os
+import ssl
from os import getenv, path
from . import Tab
@@ -146,6 +148,170 @@ class RosterInfoTab(Tab):
self.core.xmpp.del_event_handler('session_start', self.check_blocking)
self.core.xmpp.add_event_handler('blocked_message', self.on_blocked_message)
+ def check_saslexternal(self, features):
+ if 'urn:xmpp:saslcert:1' in features:
+ self.register_command('certs', self.command_certs,
+ desc=_('List the fingerprints of certificates'
+ ' which can connect to your account.'),
+ shortdesc=_('List allowed client certs.'))
+ self.register_command('cert_add', self.command_cert_add,
+ desc=_('Add a client certificate to the authorized ones. '
+ 'It must have an unique name and be contained in '
+ 'a PEM file. [management] is a boolean indicating'
+ ' if a client connected using this certificate can'
+ ' manage the certificates itself.'),
+ shortdesc=_('Add a client certificate.'),
+ usage='<name> <certificate path> [management]')
+ self.register_command('cert_disable', self.command_cert_disable,
+ desc=_('Remove a certificate from the list '
+ 'of allowed ones. Clients currently '
+ 'using this certificate will not be '
+ 'forcefully disconnected.'),
+ shortdesc=_('Disable a certificate'),
+ usage='<name>')
+ self.register_command('cert_revoke', self.command_cert_revoke,
+ desc=_('Remove a certificate from the list '
+ 'of allowed ones. Clients currently '
+ 'using this certificate will be '
+ 'forcefully disconnected.'),
+ shortdesc=_('Revoke a certificate'),
+ usage='<name>')
+ self.register_command('cert_fetch', self.command_cert_fetch,
+ desc=_('Retrieve a certificate with its '
+ 'name. It will be stored in <path>.'),
+ shortdesc=_('Fetch a certificate'),
+ usage='<name> <path>')
+
+ @command_args_parser.ignored
+ def command_certs(self):
+ """
+ /certs
+ """
+ def cb(iq):
+ if iq['type'] == 'error':
+ self.core.information(_('Unable to retrieve the certificate list.'),
+ _('Error'))
+ return
+ certs = []
+ for item in iq['sasl_certs']['items']:
+ users = '\n'.join(item['users'])
+ certs.append((item['name'], users))
+
+ if not certs:
+ return self.core.information(_('No certificates found'), _('Info'))
+ msg = _('Certificates:\n')
+ msg += '\n'.join(((' %s%s' % (item[0] + (': ' if item[1] else ''), item[1])) for item in certs))
+ self.core.information(msg, 'Info')
+
+ self.core.xmpp.plugin['xep_0257'].get_certs(callback=cb, timeout=3)
+
+ @command_args_parser.quoted(2, 1)
+ def command_cert_add(self, args):
+ """
+ /cert_add <name> <certfile> [cert-management]
+ """
+ if not args or len(args) < 2:
+ return self.core.command_help('cert_add')
+ def cb(iq):
+ if iq['type'] == 'error':
+ self.core.information(_('Unable to add the certificate.'), _('Error'))
+ else:
+ self.core.information(_('Certificate added.'), _('Info'))
+
+ name = args[0]
+
+ try:
+ with open(args[1]) as fd:
+ crt = fd.read()
+ crt = crt.replace(ssl.PEM_FOOTER, '').replace(ssl.PEM_HEADER, '').replace(' ', '').replace('\n', '')
+ except Exception as e:
+ self.core.information('Unable to read the certificate: %s' % e, 'Error')
+ return
+
+ if len(args) > 2:
+ management = args[2]
+ if management:
+ management = management.lower()
+ if management not in ('false', '0'):
+ management = True
+ else:
+ management = False
+ else:
+ management = False
+ else:
+ management = True
+
+ self.core.xmpp.plugin['xep_0257'].add_cert(name, crt, callback=cb,
+ allow_management=management)
+
+ @command_args_parser.quoted(1)
+ def command_cert_disable(self, args):
+ """
+ /cert_disable <name>
+ """
+ if not args:
+ return self.core.command_help('cert_disable')
+ def cb(iq):
+ if iq['type'] == 'error':
+ self.core.information(_('Unable to disable the certificate.'), _('Error'))
+ else:
+ self.core.information(_('Certificate disabled.'), _('Info'))
+
+ name = args[0]
+
+ self.core.xmpp.plugin['xep_0257'].disable_cert(name, callback=cb)
+
+ @command_args_parser.quoted(1)
+ def command_cert_revoke(self, args):
+ """
+ /cert_revoke <name>
+ """
+ if not args:
+ return self.core.command_help('cert_revoke')
+ def cb(iq):
+ if iq['type'] == 'error':
+ self.core.information(_('Unable to revoke the certificate.'), _('Error'))
+ else:
+ self.core.information(_('Certificate revoked.'), _('Info'))
+
+ name = args[0]
+
+ self.core.xmpp.plugin['xep_0257'].revoke_cert(name, callback=cb)
+
+
+ @command_args_parser.quoted(2)
+ def command_cert_fetch(self, args):
+ """
+ /cert_fetch <name> <path>
+ """
+ if not args or len(args) < 2:
+ return self.core.command_help('cert_fetch')
+ def cb(iq):
+ if iq['type'] == 'error':
+ self.core.information(_('Unable to fetch the certificate.'),
+ _('Error'))
+ return
+
+ cert = None
+ for item in iq['sasl_certs']['items']:
+ if item['name'] == name:
+ cert = base64.b64decode(item['x509cert'])
+ break
+
+ if not cert:
+ return self.core.information(_('Certificate not found.'), _('Info'))
+
+ cert = ssl.DER_cert_to_PEM_cert(cert)
+ with open(path, 'w') as fd:
+ fd.write(cert)
+
+ self.core.information(_('File stored at %s') % path, 'Info')
+
+ name = args[0]
+ path = args[1]
+
+ self.core.xmpp.plugin['xep_0257'].get_certs(callback=cb)
+
def on_blocked_message(self, message):
"""
When we try to send a message to a blocked contact