Skip to content
Back to top

# Source code for slixmpp.plugins.xep_0455.sos

    # Copyright © 2025 Mathieu Pasquet
    # This file is part of slixmpp
    # See the file LICENSE for copying permission.
    import json
    import logging
    from datetime import datetime
    from urllib.parse import urlparse

    from slixmpp import JID
    from slixmpp.plugins import BasePlugin, xep_0082
    from dataclasses import dataclass

    log = logging.getLogger(__name__)

    class AiohttpNotFound(Exception):
        pass

    @dataclass
    class ExternalStatus:
        planned: bool | None
        beginning: datetime
        expected_end: datetime | None
        message: dict[str, str] | None

    class XEP_0455(BasePlugin):
        """
        XEP-0455: Service Outage Status
        """
        name = 'xep_0455'
        description = 'XEP-0455: Service Outage Status'
        dependencies = {'xep_0128', 'xep_0030', 'xep_0082'}
        namespace = 'urn:xmpp:sos:0'

        async def get_external_status_addresses(self, domain: JID | None = None,
                                                **iqkwargs) -> list[str]:
            """Return the list of external status addresses for this domain.

            :param domain: Domain to disco to find a service.
            """
            if domain is None:
                domain = JID(self.xmpp.boundjid.host)

            uris = []
            results = await self.xmpp.plugin['xep_0030'].get_info(jid=domain,
                **iqkwargs
            )

            disco = results.get_plugin('disco_info', check=True)
            if disco is None or 'forms' not in disco:
                return uris
            forms = disco['forms']
            if not forms:
                return uris
            field = 'external-status-addresses'
            for form in forms:
                values = form.get_values()
                if values.get('FORM_TYPE') == [self.namespace]:
                    uris.extend(values.get(field, []))
            return uris

        @classmethod
        async def fetch_status(cls, addresses: str | list[str]) -> dict:
            """
            Get the external status from a list of addresses.
            Only works with http/https for now and stops on the first status
            fetched successfully.

            :param addresses: address or list of addresses to fetch the status from.
            """
            try:
                from aiohttp import ClientSession
            except ImportError:
                raise AiohttpNotFound("aiohttp was not found, unable to download statuses")

            if not isinstance(addresses, list):
                addresses = [addresses]

            async with ClientSession(headers={'User-Agent': 'slixmpp'}) as session:
                for address in addresses:
                    scheme, *_ = urlparse(address)
                    if scheme not in ('http', 'https'):
                        continue
                    response = await session.get(address, timeout=60)
                    if response.status >= 400:
                        log.debug(f'Server "{address}" answered with code {response.status}')
                        continue
                    text = await response.text()
                    status = cls._parse_status(text)
                    if status is not None:
                        return status

        @staticmethod
        def _parse_status(raw: str) -> ExternalStatus | None:
            """
            Parse a status json payload. Return None if the required (beginning)
            field is not found or not parseable.

            Returns a dataclass with the fields.
            """
            try:
                payload = json.loads(raw)
            except json.JSONDecodeError:
                log.error('Unable to parse the external status: {text}')
                return None
            if not payload:
                return None
            beginning_raw = payload.get('beginning')
            try:
                beginning = xep_0082.parse(beginning_raw)
            except (ValueError, TypeError):
                log.error(f'Bad value for beginning: "{beginning_raw}"')
                return None

            expected_end_raw = payload.get('expected_end')
            expected_end = None
            try:
                expected_end = xep_0082.parse(expected_end_raw)
            except (ValueError, TypeError):
                log.error(f'Bad value for expected end: "{expected_end_raw}"')

            planned_raw = payload.get('planned')
            planned = planned_raw if isinstance(planned_raw, bool) else None

            message_raw = payload.get('message')
            message = None
            if isinstance(message_raw, dict):
                message = {}
                for key, value in message_raw.items():
                    if isinstance(key, str) and isinstance(value, str):
                        message[key] = value
            return ExternalStatus(planned, beginning, expected_end, message)