from typing import Union
import nio
from nio.crypto import OlmDevice, TrustState
from slidge.command import Command, CommandAccess, Form, FormField, TableResult
from slidge.command.base import FormValues
from slixmpp.exceptions import XMPPError
from .group import MUC
from .session import Session
[docs]
class ListSpaces(Command):
CHAT_COMMAND = NODE = "spaces"
[docs]
HELP = "List the matrix spaces you're part of"
[docs]
ACCESS = CommandAccess.USER_LOGGED
[docs]
async def run(self, session: Session, _ifrom, *args: str) -> Form: # type:ignore
spaces = list[nio.MatrixRoom]()
for room in session.matrix.rooms.values():
if room.children:
spaces.append(room)
spaces = sorted(spaces, key=lambda r: r.name)
return Form(
title=self.NAME,
instructions="Choose a space to list its children rooms. "
"NB: as of now, you can also see rooms that you are a member of.",
handler=self.finish, # type:ignore
handler_args=(spaces,),
fields=[
FormField(
"space",
label="Matrix space",
type="list-single",
options=[
{"label": room.display_name, "value": str(i)}
for i, room in enumerate(spaces)
],
)
],
)
@staticmethod
[docs]
async def finish(
form_values: FormValues,
session: Session,
_ifrom,
rooms: list[nio.MatrixRoom],
):
space = rooms[int(form_values["space"])] # type:ignore
mucs = list[MUC]()
for room_id in space.children:
try:
mucs.append(await session.bookmarks.by_legacy_id(room_id))
except XMPPError:
continue
mucs = sorted(mucs, key=lambda muc: muc.name)
return TableResult(
fields=[FormField("name"), FormField("jid", type="jid-single")],
description=f"Rooms of '{space.display_name}'",
jids_are_mucs=True,
items=[{"name": muc.name, "jid": str(muc.jid)} for muc in mucs],
)
[docs]
class ManageTrust(Command):
CHAT_COMMAND = NODE = "verify"
[docs]
HELP = "Manage which OLM keys you trust or not."
[docs]
ACCESS = CommandAccess.USER_LOGGED
[docs]
HUMAN_STATES = {0: "unset", 1: "verified", 2: "blacklisted", 3: "ignored"}
[docs]
def __human_device(self, d: OlmDevice, state=True):
r = f"{d.ed25519} of {d.user_id}"
if state:
return r + f" ({self.HUMAN_STATES[d.trust_state.value]})"
return r
[docs]
async def run(
self,
session: Session, # type:ignore
_ifrom,
*args: str,
) -> Union[Form, str]:
devices = list[OlmDevice](session.matrix.olm.device_store)
device_dict = {d.id: d for d in devices}
# this part if for chat commands only
if args:
if args[0] == "all":
return await self.step2(
{
"device": list(device_dict.keys()), # type:ignore
"new_state": "verified",
},
session,
None,
device_dict,
)
else:
return await self.step2(
{"device": args[0].upper(), "new_state": "verified"},
session,
None,
device_dict,
)
return Form(
title=self.NAME,
instructions="Choose the session(s) which trust state you want to change",
handler=self.step2, # type:ignore
handler_args=(device_dict,),
fields=[
FormField(
"device",
label="Device(s)",
type="list-multi",
options=[
{"label": self.__human_device(d), "value": d.id}
for d in devices
],
),
FormField(
"new_state",
label="What new status do you want to give the selected devices?",
type="list-single",
options=[
{"label": v, "value": v} for v in self.HUMAN_STATES.values()
],
),
],
)
[docs]
async def step2(
self,
form_values: FormValues,
session: Session,
_ifrom,
devices: dict[str, OlmDevice],
):
new_state = form_values["new_state"]
matrix = session.matrix
result = ""
for device_name in form_values["device"]: # type:ignore
device = devices[device_name]
change = False
if new_state == "unset":
if device.trust_state == TrustState.verified:
change = matrix.unverify_device(device)
elif device.trust_state == TrustState.ignored:
change = matrix.unignore_device(device)
elif device.trust_state == TrustState.blacklisted:
change = matrix.unblacklist_device(device)
elif new_state == "verified":
change = session.matrix.verify_device(device)
elif new_state == "blacklisted":
change = session.matrix.blacklist_device(device)
elif new_state == "ignored":
change = session.matrix.ignore_device(device)
if change:
result += (
f"\nThe status of {self.__human_device(device, False)} "
f"is now {new_state}."
)
else:
result += (
f"\nThe status of {self.__human_device(device, False)} "
f"has not changed."
)
return result or "Nothing was changed."