Mercurial > libervia-web
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))