#!/usr/bin/env python3 # Libervia plugin to handle web frontend registration links with Ad-Hoc Commands # 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 . import time from typing import cast from uuid import uuid4 from twisted.internet import defer from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import D_, _ from libervia.backend.core.log import getLogger from libervia.backend.memory import persistent from libervia.backend.plugins.plugin_xep_0004 import DataForm, ListSingle, Option, Text from libervia.backend.plugins.plugin_xep_0050 import ( XEP_0050, AdHocCallbackData, AdHocCommand, AdHocError, Error, Note, PageData, Status, ) from libervia.backend.tools.common import data_format, date_utils log = getLogger(__name__) PLUGIN_INFO = { C.PI_NAME: "Ad-Hoc Commands - Registration", C.PI_IMPORT_NAME: "AD_HOC_REGISTRATION", C.PI_TYPE: "Misc", C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: ["XEP-0050"], C.PI_MAIN: "AdHocRegistration", C.PI_HANDLER: "no", C.PI_DESCRIPTION: _("""Add registration link handling Ad-Hoc commands"""), } class AdHocRegistration: def __init__(self, host): log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") self.host = host self._c = cast(XEP_0050, host.plugins["XEP-0050"]) self.ad_hoc_registration_data = persistent.LazyPersistentBinaryDict( "registration_links" ) host.bridge.add_method( "registration_link_get", ".plugin", in_sign="s", out_sign="s", method=self._get, async_=True, ) def _get(self, registration_link_id: str) -> defer.Deferred[str]: d = defer.ensureDeferred(self.get(registration_link_id)) d.addCallback(data_format.serialise) return d async def get(self, registration_link_id: str) -> dict: """Retrieve registration link from its ID @param registration_link_id: registration link data @return: registration link data @raise exceptions.NotFound: not registration link found with this ID. """ link_data = await self.ad_hoc_registration_data.get(registration_link_id) if not link_data: raise exceptions.NotFound expiration_timestamp = link_data.get("expiration_timestamp") if expiration_timestamp is not None and expiration_timestamp < time.time(): log.info(f"Deleting expiration link {registration_link_id}.") await self.ad_hoc_registration_data.adel(registration_link_id) raise exceptions.NotFound return link_data async def profile_connected(self, client): if client.is_admin: self._c.register_ad_hoc_command( client, AdHocCommand( callback=self.create_registration_link, label=D_("Create Registration Link"), node="https://libervia.org/registration/create", form=DataForm( title=D_("Create Registration Link"), fields=[ Text(var="recipient", label=D_("Recipient Name")), Text( var="expiration_time_pattern", label=D_("Expiration Date"), desc=D_( "Set the expiry duration for this link. Use the" " Libervia Time Pattern (e.g., '1 week'). The link" " will expire after this period." ), value="1 week", required=True, ), Text( var="registration_limit", label=D_("Maximum registrations limit"), desc=D_( "How many accounts can be registered using this link." " Set to 0 for unlimited registrations." ), value="1", required=True, ), Text( type="text-multi", var="welcome_message", label=D_("Welcome Message"), ), Text( type="text-multi", var="extra_details", label=D_("Additional Details"), ), ], ), ), ) self._c.register_ad_hoc_command( client, AdHocCommand( callback=self.list_registration_links, label=D_("List Registration Links"), node="https://libervia.org/registration/list", ), ) self._c.register_ad_hoc_command( client, AdHocCommand( callback=self.delete_registration_link, label=D_("Delete Registration Link"), node="https://libervia.org/registration/delete", ), ) async def create_registration_link( self, client: SatXMPPEntity, page_data: PageData ) -> AdHocCallbackData: """Ad-hoc command used to create a registration link. This method presents a form to the user for creating a registration link, and processes the submitted form to generate and store the link with its associated data. @param client: The XMPP client instance. @param page_data: Ad-Hoc page data. @return: Command results. @raise AdHocError: If the payload is invalid or internal error occurs. """ form = page_data.form try: recipient = form.get_field("recipient", Text).value expiration_time_pattern = form.get_field( "expiration_time_pattern", Text ).value if not expiration_time_pattern: raise KeyError if expiration_time_pattern == "never": expiration_timestamp = None else: expiration_timestamp = date_utils.date_parse_ext( expiration_time_pattern, default_tz=date_utils.TZ_LOCAL ) registration_limit = form.get_field("registration_limit", Text).value if registration_limit is None: raise KeyError registration_limit = int(registration_limit) welcome_message = form.get_field("welcome_message", Text).value or "" extra_details = form.get_field("extra_details", Text).value or "" link_id = str(uuid4()) link_data = { "recipient": recipient, "registration_limit": registration_limit, "welcome_message": welcome_message, "extra_details": extra_details, "creator_jid": page_data.requestor.full(), "use_count": 0, "creation_timestamp": int(time.time()), } if expiration_timestamp is not None: link_data["expiration_timestamp"] = expiration_timestamp await self.ad_hoc_registration_data.aset(link_id, link_data) notes = [ Note( text=D_("Registration link created successfully: {link_id}").format( link_id=link_id ), ) ] except (KeyError, ValueError): raise AdHocError(Error.BAD_PAYLOAD) return AdHocCallbackData(status=Status.COMPLETED, notes=notes) async def list_registration_links( self, client: SatXMPPEntity, page_data: PageData ) -> AdHocCallbackData: """Ad-hoc command used to list all registration links. This method retrieves all the registration links and presents them to the user. @param client: The XMPP client instance. @param page_data: Ad-Hoc page data. @return: Command results. @raise AdHocError: If internal error occurs. """ all_links = await self.ad_hoc_registration_data.all() form = DataForm(type="result", title=D_("Registered Links")) for link_id, link_data in all_links.items(): form.fields.append( Text( type="text-multi", var=f"link_{link_id}", label=D_("Link ID: {link_id}").format(link_id=link_id), value=str(link_data), ) ) return AdHocCallbackData(form=form, status=Status.COMPLETED) async def delete_registration_link( self, client: SatXMPPEntity, page_data: PageData ) -> AdHocCallbackData: """Ad-hoc command used to delete a registration link. This method presents a form to the user for selecting a registration link to delete, and processes the submitted form to delete the selected link. @param client: The XMPP client instance. @param page_data: Ad-Hoc page data. @return: Command results. @raise AdHocError: If the payload is invalid or internal error occurs. """ if page_data.idx == 0: all_links = await self.ad_hoc_registration_data.all() form = DataForm( title=D_("Delete Registration Link"), ) link_options = [Option(value=link_id, label=link_id) for link_id in all_links] form.fields.append( ListSingle( var="link_to_delete", label=D_("Select Link to Delete"), options=link_options, required=True, ) ) return AdHocCallbackData(form=form, status=Status.EXECUTING) else: link_to_delete = page_data.form.get_field("link_to_delete", ListSingle).value if not link_to_delete: raise AdHocError(Error.BAD_PAYLOAD) await self.ad_hoc_registration_data.adel(link_to_delete) return AdHocCallbackData( status=Status.COMPLETED, notes=[ Note( text=D_( "Registration link {link_id} deleted successfully." ).format(link_id=link_to_delete), ) ], )