Mercurial > libervia-backend
view libervia/backend/plugins/plugin_misc_account.py @ 4334:111dce64dcb5
plugins XEP-0300, XEP-0446, XEP-0447, XEP0448 and others: Refactoring to use Pydantic:
Pydantic models are used more and more in Libervia, for the bridge API, and also to
convert `domish.Element` to internal representation.
Type hints have also been added in many places.
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:12:38 +0100 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia plugin for account creation # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program 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. # This program 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/>. from libervia.backend.core.i18n import _, D_ from libervia.backend.core.log import getLogger from libervia.backend.core import exceptions from libervia.backend.tools import xml_tools from libervia.backend.memory.memory import Sessions from libervia.backend.memory.crypto import PasswordHasher from libervia.backend.core.constants import Const as C import configparser from twisted.internet import defer from twisted.python.failure import Failure from twisted.words.protocols.jabber import jid from libervia.backend.tools.common import email as sat_email log = getLogger(__name__) # FIXME: this plugin code is old and need a cleaning # TODO: account deletion/password change need testing PLUGIN_INFO = { C.PI_NAME: "Account Plugin", C.PI_IMPORT_NAME: "MISC-ACCOUNT", C.PI_TYPE: "MISC", C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: ["XEP-0077"], C.PI_RECOMMENDATIONS: ["GROUPBLOG"], C.PI_MAIN: "MiscAccount", C.PI_HANDLER: "no", C.PI_DESCRIPTION: _("""Libervia account creation"""), } CONFIG_SECTION = "plugin account" # You need do adapt the following consts to your server # all theses values (key=option name, value=default) can (and should) be overriden # in libervia.conf in section CONFIG_SECTION default_conf = { "email_from": "NOREPLY@example.net", "email_server": "localhost", "email_sender_domain": "", "email_port": 25, "email_username": "", "email_password": "", "email_starttls": "false", "email_auth": "false", "email_admins_list": [], "admin_email": "", "new_account_server": "localhost", "new_account_domain": "", # use xmpp_domain if not found "reserved_list": ["libervia"], # profiles which can't be used } WELCOME_MSG = D_( """Welcome to Libervia, the web interface of Salut à Toi. Your account on {domain} has been successfully created. This is a demonstration version to show you the current status of the project. It is still under development, please keep it in mind! Here is your connection information: Login on {domain}: {profile} Jabber ID (JID): {jid} Your password has been chosen by yourself during registration. In the beginning, you have nobody to talk to. To find some contacts, you may use the users' directory: - make yourself visible in "Service / Directory subscription". - search for people with "Contacts" / Search directory". Any feedback welcome. Thank you! Salut à Toi association https://www.salut-a-toi.org """ ) DEFAULT_DOMAIN = "example.net" class MiscAccount(object): """Account plugin: create a SàT + XMPP account, used by Libervia""" # XXX: This plugin was initialy a Q&D one used for the demo. # TODO: cleaning, separate email handling, more configuration/tests, fixes def __init__(self, host): log.info(_("Plugin Account initialization")) self.host = host host.bridge.add_method( "libervia_account_register", ".plugin", in_sign="sss", out_sign="", method=self._register_account, async_=True, ) host.bridge.add_method( "account_domain_new_get", ".plugin", in_sign="", out_sign="s", method=self.account_domain_new_get, async_=False, ) host.bridge.add_method( "account_dialog_ui_get", ".plugin", in_sign="s", out_sign="s", method=self._get_account_dialog_ui, async_=False, ) host.bridge.add_method( "credentials_xmpp_connect", ".plugin", in_sign="ss", out_sign="b", method=self.credentials_xmpp_connect, async_=True, ) self.fix_email_admins() self._sessions = Sessions() self.__account_cb_id = host.register_callback( self._account_dialog_cb, with_data=True ) self.__change_password_id = host.register_callback( self.__change_password_cb, with_data=True ) def delete_blog_callback(posts, comments): return lambda data, profile: self.__delete_blog_posts_cb( posts, comments, data, profile ) self.__delete_posts_id = host.register_callback( delete_blog_callback(True, False), with_data=True ) self.__delete_comments_id = host.register_callback( delete_blog_callback(False, True), with_data=True ) self.__delete_posts_comments_id = host.register_callback( delete_blog_callback(True, True), with_data=True ) self.__delete_account_id = host.register_callback( self.__delete_account_cb, with_data=True ) # FIXME: remove this after some time, when the deprecated parameter is really abandoned def fix_email_admins(self): """Handle deprecated config option "admin_email" to fix the admin emails list""" admin_email = self.config_get("admin_email") if not admin_email: return log.warning( "admin_email parameter is deprecated, please use email_admins_list instead" ) param_name = "email_admins_list" try: section = "" value = self.host.memory.config_get(section, param_name, Exception) except (configparser.NoOptionError, configparser.NoSectionError): section = CONFIG_SECTION value = self.host.memory.config_get( section, param_name, default_conf[param_name] ) value = set(value) value.add(admin_email) self.host.memory.config.set(section, param_name, ",".join(value)) def config_get(self, name, section=CONFIG_SECTION): if name.startswith("email_"): # XXX: email_ parameters were first in [plugin account] section # but as it make more sense to have them in common with other plugins, # they can now be in [DEFAULT] section try: value = self.host.memory.config_get(None, name, Exception) except (configparser.NoOptionError, configparser.NoSectionError): pass else: return value if section == CONFIG_SECTION: default = default_conf[name] else: default = None return self.host.memory.config_get(section, name, default) def _register_account(self, email, password, profile): return self.registerAccount(email, password, None, profile) def registerAccount(self, email, password, jid_s, profile): """Register a new profile, its associated XMPP account, send the confirmation emails. @param email (unicode): where to send to confirmation email to @param password (unicode): password chosen by the user while be used for profile *and* XMPP account @param jid_s (unicode): JID to re-use or to register: - non empty value: bind this JID to the new sat profile - None or "": register a new JID on the local XMPP server @param profile @return Deferred """ d = self.create_profile(password, jid_s, profile) d.addCallback(lambda __: self.send_emails(email, profile)) return d def create_profile(self, password, jid_s, profile): """Register a new profile and its associated XMPP account. @param password (unicode): password chosen by the user while be used for profile *and* XMPP account @param jid_s (unicode): JID to re-use or to register: - non empty value: bind this JID to the new sat profile - None or "": register a new JID on the local XMPP server @param profile @return Deferred """ if not password or not profile: raise exceptions.DataError if profile.lower() in self.config_get("reserved_list"): return defer.fail(Failure(exceptions.ConflictError)) d = self.host.memory.create_profile(profile, password) d.addCallback(lambda __: self.profile_created(password, jid_s, profile)) return d def profile_created(self, password, jid_s, profile): """Create the XMPP account and set the profile connection parameters. @param password (unicode): password chosen by the user @param jid_s (unicode): JID to re-use or to register: - non empty value: bind this JID to the new sat profile - None or empty: register a new JID on the local XMPP server @param profile @return: Deferred """ if jid_s: d = defer.succeed(None) jid_ = jid.JID(jid_s) else: jid_s = profile + "@" + self.account_domain_new_get() jid_ = jid.JID(jid_s) d = self.host.plugins["XEP-0077"].register_new_account(jid_, password) def setParams(__): self.host.memory.param_set( "JabberID", jid_s, "Connection", profile_key=profile ) d = self.host.memory.param_set( "Password", password, "Connection", profile_key=profile ) return d def remove_profile(failure): self.host.memory.profile_delete_async(profile) return failure d.addCallback(lambda __: self.host.memory.start_session(password, profile)) d.addCallback(setParams) d.addCallback(lambda __: self.host.memory.stop_session(profile)) d.addErrback(remove_profile) return d def _send_email_eb(self, failure_, email): # TODO: return error code to user log.error( _("Failed to send account creation confirmation to {email}: {msg}").format( email=email, msg=failure_ ) ) def send_emails(self, email, profile): # time to send the email domain = self.account_domain_new_get() # email to the administrators admins_emails = self.config_get("email_admins_list") if not admins_emails: log.warning( "No known admin email, we can't send email to administrator(s).\n" "Please fill email_admins_list parameter" ) d_admin = defer.fail(exceptions.DataError("no admin email")) else: subject = _("New Libervia account created") # there is no email when an existing XMPP account is used body = f"New account created on {domain}: {profile} [{email or '<no email>'}]" d_admin = sat_email.send_email( self.host.memory.config, admins_emails, subject, body ) admins_emails_txt = ", ".join(["<" + addr + ">" for addr in admins_emails]) d_admin.addCallbacks( lambda __: log.debug( "Account creation notification sent to admin(s) {}".format( admins_emails_txt ) ), lambda __: log.error( "Failed to send account creation notification to admin {}".format( admins_emails_txt ) ), ) if not email: # TODO: if use register with an existing account, an XMPP message should be sent return d_admin jid_s = self.host.memory.param_get_a( "JabberID", "Connection", profile_key=profile ) subject = _("Your Libervia account has been created") body = _(WELCOME_MSG).format(profile=profile, jid=jid_s, domain=domain) # XXX: this will not fail when the email address doesn't exist # FIXME: check email reception to validate email given by the user # FIXME: delete the profile if the email could not been sent? d_user = sat_email.send_email(self.host.memory.config, [email], subject, body) d_user.addCallbacks( lambda __: log.debug( "Account creation confirmation sent to <{}>".format(email) ), self._send_email_eb, errbackArgs=[email], ) return defer.DeferredList([d_user, d_admin]) def account_domain_new_get(self): """get the domain that will be set to new account""" domain = self.config_get("new_account_domain") or self.config_get( "xmpp_domain", None ) if not domain: log.warning( _( 'xmpp_domain needs to be set in sat.conf. Using "{default}" meanwhile' ).format(default=DEFAULT_DOMAIN) ) return DEFAULT_DOMAIN return domain def _get_account_dialog_ui(self, profile): """Get the main dialog to manage your account @param menu_data @param profile: %(doc_profile)s @return: XML of the dialog """ form_ui = xml_tools.XMLUI( "form", "tabs", title=D_("Manage your account"), submit_id=self.__account_cb_id, ) tab_container = form_ui.current_container tab_container.add_tab( "update", D_("Change your password"), container=xml_tools.PairsContainer ) form_ui.addLabel(D_("Current profile password")) form_ui.addPassword("current_passwd", value="") form_ui.addLabel(D_("New password")) form_ui.addPassword("new_passwd1", value="") form_ui.addLabel(D_("New password (again)")) form_ui.addPassword("new_passwd2", value="") # FIXME: uncomment and fix these features """ if 'GROUPBLOG' in self.host.plugins: tab_container.add_tab("delete_posts", D_("Delete your posts"), container=xml_tools.PairsContainer) form_ui.addLabel(D_("Current profile password")) form_ui.addPassword("delete_posts_passwd", value="") form_ui.addLabel(D_("Delete all your posts and their comments")) form_ui.addBool("delete_posts_checkbox", "false") form_ui.addLabel(D_("Delete all your comments on other's posts")) form_ui.addBool("delete_comments_checkbox", "false") tab_container.add_tab("delete", D_("Delete your account"), container=xml_tools.PairsContainer) form_ui.addLabel(D_("Current profile password")) form_ui.addPassword("delete_passwd", value="") form_ui.addLabel(D_("Delete your account")) form_ui.addBool("delete_checkbox", "false") """ return form_ui.toXml() @defer.inlineCallbacks def _account_dialog_cb(self, data, profile): """Called when the user submits the main account dialog @param data @param profile """ sat_cipher = yield self.host.memory.param_get_a_async( C.PROFILE_PASS_PATH[1], C.PROFILE_PASS_PATH[0], profile_key=profile ) @defer.inlineCallbacks def verify(attempt): auth = yield PasswordHasher.verify(attempt, sat_cipher) defer.returnValue(auth) def error_ui(message=None): if not message: message = D_("The provided profile password doesn't match.") error_ui = xml_tools.XMLUI("popup", title=D_("Attempt failure")) error_ui.addText(message) return {"xmlui": error_ui.toXml()} # check for account deletion # FIXME: uncomment and fix these features """ delete_passwd = data[xml_tools.SAT_FORM_PREFIX + 'delete_passwd'] delete_checkbox = data[xml_tools.SAT_FORM_PREFIX + 'delete_checkbox'] if delete_checkbox == 'true': verified = yield verify(delete_passwd) assert isinstance(verified, bool) if verified: defer.returnValue(self.__delete_account(profile)) defer.returnValue(error_ui()) # check for blog posts deletion if 'GROUPBLOG' in self.host.plugins: delete_posts_passwd = data[xml_tools.SAT_FORM_PREFIX + 'delete_posts_passwd'] delete_posts_checkbox = data[xml_tools.SAT_FORM_PREFIX + 'delete_posts_checkbox'] delete_comments_checkbox = data[xml_tools.SAT_FORM_PREFIX + 'delete_comments_checkbox'] posts = delete_posts_checkbox == 'true' comments = delete_comments_checkbox == 'true' if posts or comments: verified = yield verify(delete_posts_passwd) assert isinstance(verified, bool) if verified: defer.returnValue(self.__delete_blog_posts(posts, comments, profile)) defer.returnValue(error_ui()) """ # check for password modification current_passwd = data[xml_tools.SAT_FORM_PREFIX + "current_passwd"] new_passwd1 = data[xml_tools.SAT_FORM_PREFIX + "new_passwd1"] new_passwd2 = data[xml_tools.SAT_FORM_PREFIX + "new_passwd2"] if new_passwd1 or new_passwd2: verified = yield verify(current_passwd) assert isinstance(verified, bool) if verified: if new_passwd1 == new_passwd2: data = yield self.__change_password(new_passwd1, profile=profile) defer.returnValue(data) else: defer.returnValue( error_ui( D_("The values entered for the new password are not equal.") ) ) defer.returnValue(error_ui()) defer.returnValue({}) def __change_password(self, password, profile): """Ask for a confirmation before changing the XMPP account and SàT profile passwords. @param password (str): the new password @param profile (str): %(doc_profile)s """ session_id, __ = self._sessions.new_session( {"new_password": password}, profile=profile ) form_ui = xml_tools.XMLUI( "form", title=D_("Change your password?"), submit_id=self.__change_password_id, session_id=session_id, ) form_ui.addText( D_( "Note for advanced users: this will actually change both your SàT profile password AND your XMPP account password." ) ) form_ui.addText(D_("Continue with changing the password?")) return {"xmlui": form_ui.toXml()} def __change_password_cb(self, data, profile): """Actually change the user XMPP account and SàT profile password @param data (dict) @profile (str): %(doc_profile)s """ client = self.host.get_client(profile) password = self._sessions.profile_get(data["session_id"], profile)["new_password"] del self._sessions[data["session_id"]] def password_changed(__): d = self.host.memory.param_set( C.PROFILE_PASS_PATH[1], password, C.PROFILE_PASS_PATH[0], profile_key=profile, ) d.addCallback( lambda __: self.host.memory.param_set( "Password", password, "Connection", profile_key=profile ) ) confirm_ui = xml_tools.XMLUI("popup", title=D_("Confirmation")) confirm_ui.addText(D_("Your password has been changed.")) return defer.succeed({"xmlui": confirm_ui.toXml()}) def errback(failure): error_ui = xml_tools.XMLUI("popup", title=D_("Error")) error_ui.addText( D_("Your password could not be changed: %s") % failure.getErrorMessage() ) return defer.succeed({"xmlui": error_ui.toXml()}) d = self.host.plugins["XEP-0077"].change_password(client, password) d.addCallbacks(password_changed, errback) return d def __delete_account(self, profile): """Ask for a confirmation before deleting the XMPP account and SàT profile @param profile """ form_ui = xml_tools.XMLUI( "form", title=D_("Delete your account?"), submit_id=self.__delete_account_id ) form_ui.addText( D_( "If you confirm this dialog, you will be disconnected and then your XMPP account AND your SàT profile will both be DELETED." ) ) target = D_( "contact list, messages history, blog posts and comments" if "GROUPBLOG" in self.host.plugins else D_("contact list and messages history") ) form_ui.addText( D_( "All your data stored on %(server)s, including your %(target)s will be erased." ) % {"server": self.account_domain_new_get(), "target": target} ) form_ui.addText( D_( "There is no other confirmation dialog, this is the very last one! Are you sure?" ) ) return {"xmlui": form_ui.toXml()} def __delete_account_cb(self, data, profile): """Actually delete the XMPP account and SàT profile @param data @param profile """ client = self.host.get_client(profile) def user_deleted(__): # FIXME: client should be disconnected at this point, so 2 next loop should be removed (to be confirmed) for jid_ in client.roster._jids: # empty roster client.presence.unsubscribe(jid_) for jid_ in self.host.memory.sub_waiting_get( profile ): # delete waiting subscriptions self.host.memory.del_waiting_sub(jid_) delete_profile = lambda: self.host.memory.profile_delete_async( profile, force=True ) if "GROUPBLOG" in self.host.plugins: d = self.host.plugins["GROUPBLOG"].deleteAllGroupBlogsAndComments( profile_key=profile ) d.addCallback(lambda __: delete_profile()) else: delete_profile() return defer.succeed({}) def errback(failure): error_ui = xml_tools.XMLUI("popup", title=D_("Error")) error_ui.addText( D_("Your XMPP account could not be deleted: %s") % failure.getErrorMessage() ) return defer.succeed({"xmlui": error_ui.toXml()}) d = self.host.plugins["XEP-0077"].unregister(client, jid.JID(client.jid.host)) d.addCallbacks(user_deleted, errback) return d def __delete_blog_posts(self, posts, comments, profile): """Ask for a confirmation before deleting the blog posts @param posts: delete all posts of the user (and their comments) @param comments: delete all the comments of the user on other's posts @param data @param profile """ if posts: if comments: # delete everything form_ui = xml_tools.XMLUI( "form", title=D_("Delete all your (micro-)blog posts and comments?"), submit_id=self.__delete_posts_comments_id, ) form_ui.addText( D_( "If you confirm this dialog, all the (micro-)blog data you submitted will be erased." ) ) form_ui.addText( D_( "These are the public and private posts and comments you sent to any group." ) ) form_ui.addText( D_( "There is no other confirmation dialog, this is the very last one! Are you sure?" ) ) else: # delete only the posts form_ui = xml_tools.XMLUI( "form", title=D_("Delete all your (micro-)blog posts?"), submit_id=self.__delete_posts_id, ) form_ui.addText( D_( "If you confirm this dialog, all the public and private posts you sent to any group will be erased." ) ) form_ui.addText( D_( "There is no other confirmation dialog, this is the very last one! Are you sure?" ) ) elif comments: # delete only the comments form_ui = xml_tools.XMLUI( "form", title=D_("Delete all your (micro-)blog comments?"), submit_id=self.__delete_comments_id, ) form_ui.addText( D_( "If you confirm this dialog, all the public and private comments you made on other people's posts will be erased." ) ) form_ui.addText( D_( "There is no other confirmation dialog, this is the very last one! Are you sure?" ) ) return {"xmlui": form_ui.toXml()} def __delete_blog_posts_cb(self, posts, comments, data, profile): """Actually delete the XMPP account and SàT profile @param posts: delete all posts of the user (and their comments) @param comments: delete all the comments of the user on other's posts @param profile """ if posts: if comments: target = D_("blog posts and comments") d = self.host.plugins["GROUPBLOG"].deleteAllGroupBlogsAndComments( profile_key=profile ) else: target = D_("blog posts") d = self.host.plugins["GROUPBLOG"].deleteAllGroupBlogs( profile_key=profile ) elif comments: target = D_("comments") d = self.host.plugins["GROUPBLOG"].deleteAllGroupBlogsComments( profile_key=profile ) def deleted(result): ui = xml_tools.XMLUI("popup", title=D_("Deletion confirmation")) # TODO: change the message when delete/retract notifications are done with XEP-0060 ui.addText(D_("Your %(target)s have been deleted.") % {"target": target}) ui.addText( D_( "Known issue of the demo version: you need to refresh the page to make the deleted posts actually disappear." ) ) return defer.succeed({"xmlui": ui.toXml()}) def errback(failure): error_ui = xml_tools.XMLUI("popup", title=D_("Error")) error_ui.addText( D_("Your %(target)s could not be deleted: %(message)s") % {"target": target, "message": failure.getErrorMessage()} ) return defer.succeed({"xmlui": error_ui.toXml()}) d.addCallbacks(deleted, errback) return d def credentials_xmpp_connect(self, jid_s, password): """Create and connect a new SàT profile using the given XMPP credentials. Re-use given JID and XMPP password for the profile name and profile password. @param jid_s (unicode): JID @param password (unicode): XMPP password @return Deferred(bool) @raise exceptions.PasswordError, exceptions.ConflictError """ try: # be sure that the profile doesn't exist yet self.host.memory.get_profile_name(jid_s) except exceptions.ProfileUnknownError: pass else: raise exceptions.ConflictError d = self.create_profile(password, jid_s, jid_s) d.addCallback( lambda __: self.host.memory.get_profile_name(jid_s) ) # checks if the profile has been successfuly created d.addCallback( lambda profile: defer.ensureDeferred( self.host.connect(profile, password, {}, 0) ) ) def connected(result): self.send_emails(None, profile=jid_s) return result def remove_profile( failure, ): # profile has been successfully created but the XMPP credentials are wrong! log.debug( "Removing previously auto-created profile: %s" % failure.getErrorMessage() ) self.host.memory.profile_delete_async(jid_s) raise failure # FIXME: we don't catch the case where the JID host is not an XMPP server, and the user # has to wait until the DBUS timeout ; as a consequence, emails are sent to the admins # and the profile is not deleted. When the host exists, remove_profile is well called. d.addCallbacks(connected, remove_profile) return d