view libervia/web/pages/chat/page_meta.py @ 1598:86c7a3a625d5

server: always start a new session on connection: The session was kept when a user was connecting from service profile (but not from other profiles), this was leading to session fixation vulnerability (an attacker on the same machine could get service profile session cookie, and use it when a victim would log-in). This patch fixes it by always starting a new session on connection. fix 443
author Goffi <goffi@goffi.org>
date Fri, 23 Feb 2024 13:35:24 +0100
parents 7941444c1671
children
line wrap: on
line source

#!/usr/bin/env python3

from libervia.backend.core.i18n import _
from twisted.internet import defer
from libervia.backend.core.log import getLogger
from libervia.backend.tools.common import data_objects
from libervia.backend.tools.common import data_format
from libervia.frontends.tools import jid
from libervia.web.server.constants import Const as C
from libervia.web.server import session_iface


log = getLogger(__name__)

name = "chat"
access = C.PAGES_ACCESS_PROFILE
template = "chat/chat.html"
dynamic = True


def parse_url(self, request):
    rdata = self.get_r_data(request)

    try:
        target_jid_s = self.next_path(request)
    except IndexError:
        # not chat jid, we redirect to jid selection page
        self.page_redirect("chat_select", request)
        return

    try:
        target_jid = jid.JID(target_jid_s)
        if not target_jid.local:
            raise ValueError(_("invalid jid for chat (no local part)"))
    except Exception as e:
        log.warning(
            _("bad chat jid entered: {jid} ({msg})").format(jid=target_jid_s, msg=e)
        )
        self.page_error(request, C.HTTP_BAD_REQUEST)
    else:
        rdata["target"] = target_jid


async def prepare_render(self, request):
    #  FIXME: bug on room filtering (currently display messages from all rooms)
    session = self.host.get_session_data(request, session_iface.IWebSession)
    template_data = request.template_data
    rdata = self.get_r_data(request)
    target_jid = rdata["target"]
    profile = session.profile
    profile_jid = session.jid

    disco = await self.host.bridge_call(
        "disco_infos", target_jid.domain, "", True, profile
    )
    if "conference" in [i[0] for i in disco[1]]:
        chat_type = C.CHAT_GROUP
        join_ret = await self.host.bridge_call(
            "muc_join", target_jid.bare, "", "", profile
        )
        (already_joined,
         room_jid_s,
         occupants,
         user_nick,
         room_subject,
         room_statuses,
         __) = join_ret
        template_data["subject"] = room_subject
        template_data["room_statuses"] = room_statuses
        own_local_jid = jid.JID(room_jid_s)
        own_local_jid = own_local_jid.change_resource(user_nick)
    else:
        room_subject = None
        chat_type = C.CHAT_ONE2ONE
        own_local_jid = profile_jid
    rdata["chat_type"] = chat_type
    template_data["own_local_jid"] = own_local_jid

    history = await self.host.bridge_call(
        "history_get",
        profile_jid.userhost(),
        target_jid.bare,
        20,
        True,
        {},
        profile,
    )

    authors = {m[2] for m in history}
    identities = session.identities
    for author in authors:
        id_raw = await self.host.bridge_call(
            "identity_get", author, [], True, profile)
        identities[author] = data_format.deserialise(id_raw)

    template_data["messages"] = data_objects.Messages(history)
    rdata['identities'] = identities
    template_data["target_jid"] = target_jid
    template_data["chat_type"] = chat_type
    self.expose_to_scripts(
        request,
        room_subject=room_subject,
        own_local_jid=str(own_local_jid),
        target_jid=target_jid,
        chat_type=chat_type,
    )


def on_data(self, request, data):
    session = self.host.get_session_data(request, session_iface.IWebSession)
    rdata = self.get_r_data(request)
    target = rdata["target"]
    data_type = data.get("type", "")
    if data_type == "msg":
        message = data["body"]
        mess_type = (
            C.MESS_TYPE_GROUPCHAT
            if rdata["chat_type"] == C.CHAT_GROUP
            else C.MESS_TYPE_CHAT
        )
        log.debug("message received: {}".format(message))
        self.host.bridge_call(
            "message_send",
            target.full(),
            {"": message},
            {},
            mess_type,
            "",
            session.profile,
        )
    else:
        log.warning("unknown message type: {type}".format(type=data_type))