libervia-backend/libervia/backend/plugins/plugin_misc_account.py

#!/usr/bin/env python3

# Libervia plugin for account creation
# Copyright (C) 2009-2026 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 typing import cast
from pydantic import Field
from libervia.backend import G
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
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
from libervia.backend.tools.config import EmailSettings, PluginSettings

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: "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
}

# profiles which can't be used for accound creation. Must be lowercase.
RESERVED_LIST: list[str] = ["libervia"]


WELCOME_MSG = D_("""Welcome to Libervia.

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!

The Libervia team.
https://www.libervia.org
""")


class PluginsAccountSettings(PluginSettings, plugin_info=PLUGIN_INFO):
    # Use global email settings by default.
    email: EmailSettings | None = Field(
        default=None,
        description="Email settings, default to settings from ``email`` section, use "
        "this if you want different ones for account creation.",
    )


class MiscAccount(object):
    """Account plugin: create a Libervia + XMPP account."""

    # 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=lambda: G.config.common.new_account_domain,
            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="ssb",
            out_sign="b",
            method=self.credentials_xmpp_connect,
            async_=True,
        )

        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
        )

    def get_email_conf(self) -> EmailSettings:
        """Return settings to send an email.

        Will use settings of this plugins is set, otherwise global ones.
        """
        plugin_settings = G.config.get_settings(PluginsAccountSettings)
        if plugin_settings.email is not None:
            return plugin_settings.email
        else:
            return G.config.email

    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 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 = f"{profile}@{G.config.common.new_account_domain}"
            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 = G.config.common.new_account_domain

        # email to the administrators
        admins_emails = G.config.common.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,
                email_settings=self.get_email_conf(),
            )

        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,
            email_settings=self.get_email_conf(),
        )
        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 _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)
            return 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:
                return self.__delete_account(profile)
            return 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:
                    return self.__delete_blog_posts(posts, comments, profile)
                return 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)
                    return data
                else:
                    return error_ui(
                        D_("The values entered for the new password are not equal.")
                    )
            return error_ui()

        return {}

    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 Libervia 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}, including your {target} will be erased."
            ).format(server=G.config.common.new_account_domain, 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, send_email=True):
        """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
        @param send_email (bool): send the administrator notification email
        @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):
            if send_email:
                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