Add command to view the safety number of another user

mausignald/signald.py

@@ -12,7 +12,7 @@ from mautrix.util.logging import TraceLogger
 from .rpc import SignaldRPCClient
 from .errors import UnexpectedError, UnexpectedResponse, make_linking_error
 from .types import (Address, Quote, Attachment, Reaction, Account, Message, Contact, Group,
-                    Profile, GroupID)
+                    Profile, GroupID, Identity, GetIdentitiesResponse)
 
 T = TypeVar('T')
 EventHandler = Callable[[T], Awaitable[None]]
@@ -143,5 +143,10 @@ class SignaldClient(SignaldRPCClient):
             raise
         return Profile.deserialize(resp)
 
+    async def get_identities(self, username: str, address: Address) -> GetIdentitiesResponse:
+        resp = await self.request("get_identities", "identities", username=username,
+                                  recipientAddress=address.serialize())
+        return GetIdentitiesResponse.deserialize(resp)
+
     async def set_profile(self, username: str, new_name: str) -> None:
         await self.request("set_profile", "profile_set", username=username, name=new_name)

mausignald/types.py

@@ -57,6 +57,26 @@ class Address(SerializableAttrs['Address']):
         return Address(number=value) if value.startswith("+") else Address(uuid=UUID(value))
 
 
+@dataclass
+class TrustLevel(SerializableEnum):
+    TRUSTED_UNVERIFIED = "TRUSTED_UNVERIFIED"
+
+
+@dataclass
+class Identity(SerializableAttrs['Identity']):
+    trust_level: TrustLevel
+    added: int
+    fingerprint: str
+    safety_number: str
+    qr_code_data: str
+    address: Address
+
+
+@dataclass
+class GetIdentitiesResponse(SerializableAttrs['GetIdentitiesResponse']):
+    identities: List[Identity]
+
+
 @dataclass
 class Contact(SerializableAttrs['Contact']):
     address: Address

mautrix_signal/commands/__init__.py

@@ -1,3 +1,3 @@
 from .auth import SECTION_AUTH
 from .conn import SECTION_CONNECTION
-from .signal import SECTION_CREATING_PORTALS
+from .signal import SECTION_SIGNAL

mautrix_signal/commands/auth.py

@@ -13,11 +13,13 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Union
 import io
 
 from mausignald.errors import UnexpectedResponse
 from mautrix.client import Client
 from mautrix.bridge import custom_puppet as cpu
+from mautrix.appservice import IntentAPI
 from mautrix.types import MediaMessageEventContent, MessageType, ImageInfo
 from mautrix.bridge.commands import HelpSection, command_handler
 
@@ -33,6 +35,20 @@ except ImportError:
 SECTION_AUTH = HelpSection("Authentication", 10, "")
 
 
+async def make_qr(intent: IntentAPI, data: Union[str, bytes], body: str = None
+                  ) -> MediaMessageEventContent:
+    # TODO always encrypt QR codes?
+    buffer = io.BytesIO()
+    image = qrcode.make(data)
+    size = image.pixel_size
+    image.save(buffer, "PNG")
+    qr = buffer.getvalue()
+    mxc = await intent.upload_media(qr, "image/png", "qr.png", len(qr))
+    return MediaMessageEventContent(body=body or data, url=mxc, msgtype=MessageType.IMAGE,
+                                    info=ImageInfo(mimetype="image/png", size=len(qr),
+                                                   width=size, height=size))
+
+
 @command_handler(needs_auth=False, management_only=True, help_section=SECTION_AUTH,
                  help_text="Link the bridge as a secondary device", help_args="[device name]")
 async def link(evt: CommandEvent) -> None:
@@ -43,15 +59,7 @@ async def link(evt: CommandEvent) -> None:
     device_name = " ".join(evt.args) or "Mautrix-Signal bridge"
 
     async def callback(uri: str) -> None:
-        buffer = io.BytesIO()
-        image = qrcode.make(uri)
-        size = image.pixel_size
-        image.save(buffer, "PNG")
-        qr = buffer.getvalue()
-        mxc = await evt.az.intent.upload_media(qr, "image/png", "link-qr.png", len(qr))
-        content = MediaMessageEventContent(body=uri, url=mxc, msgtype=MessageType.IMAGE,
-                                           info=ImageInfo(mimetype="image/png", size=len(qr),
-                                                          width=size, height=size))
+        content = await make_qr(evt.az.intent, uri)
         await evt.az.intent.send_message(evt.room_id, content)
 
     account = await evt.bridge.signal.link(callback, device_name=device_name)

mautrix_signal/commands/signal.py

@@ -13,31 +13,61 @@
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from typing import Optional
+import base64
+import io
+
+from mautrix.types import MediaMessageEventContent, MessageType, ImageInfo
 from mautrix.bridge.commands import HelpSection, command_handler
 from mausignald.types import Address
 
 from .. import puppet as pu, portal as po
+from .auth import make_qr
 from .typehint import CommandEvent
 
-SECTION_CREATING_PORTALS = HelpSection("Creating portals", 20, "")
+try:
+    import qrcode
+    import PIL as _
+except ImportError:
+    qrcode = None
+
+SECTION_SIGNAL = HelpSection("Signal actions", 20, "")
 
 remove_extra_chars = str.maketrans("", "", " .,-()")
 
 
-@command_handler(needs_auth=True, management_only=False, help_section=SECTION_CREATING_PORTALS,
-                 help_text="Open a private chat portal with a specific phone number",
-                 help_args="<_phone_>")
-async def pm(evt: CommandEvent) -> None:
+async def _get_puppet_from_cmd(evt: CommandEvent) -> Optional['pu.Puppet']:
     if len(evt.args) == 0 or not evt.args[0].startswith("+"):
         await evt.reply("**Usage:** `$cmdprefix+sp pm <phone>` "
                         "(enter phone number in international format)")
-        return
+        return None
     phone = "".join(evt.args).translate(remove_extra_chars)
     if not phone[1:].isdecimal():
         await evt.reply("**Usage:** `$cmdprefix+sp pm <phone>` "
                         "(enter phone number in international format)")
+        return None
+    return await pu.Puppet.get_by_address(Address(number=phone))
+
+
+def _format_safety_number(number: str) -> str:
+    line_size = 20
+    chunk_size = 5
+    return "\n".join(" ".join([number[chunk:chunk + chunk_size]
+                               for chunk in range(line, line + line_size, chunk_size)])
+                     for line in range(0, len(number), line_size))
+
+
+def _pill(puppet: 'pu.Puppet') -> str:
+    return f"[{puppet.name}](https://matrix.to/#/{puppet.mxid})"
+
+
+@command_handler(needs_auth=True, management_only=False, help_section=SECTION_SIGNAL,
+                 help_text="Open a private chat portal with a specific phone number",
+                 help_args="<_phone_>")
+async def pm(evt: CommandEvent) -> None:
+    puppet = await _get_puppet_from_cmd(evt)
+    if not puppet:
         return
-    puppet = await pu.Puppet.get_by_address(Address(number=phone))
     portal = await po.Portal.get_by_chat_id(puppet.address, receiver=evt.sender.username,
                                             create=True)
     if portal.mxid:
@@ -46,4 +76,39 @@ async def pm(evt: CommandEvent) -> None:
         await portal.main_intent.invite_user(portal.mxid, evt.sender.mxid)
         return
     await portal.create_matrix_room(evt.sender, puppet.address)
-    await evt.reply(f"Created a portal room with [{puppet.name}](https://matrix.to/#/{puppet.mxid}) and invited you to it")
+    await evt.reply(f"Created a portal room with {_pill(puppet)} and invited you to it")
+
+
+@command_handler(needs_auth=True, management_only=False, help_section=SECTION_SIGNAL,
+                 help_text="View the safety number of a specific user",
+                 help_args="[--qr] [_phone_]")
+async def safety_number(evt: CommandEvent) -> None:
+    show_qr = evt.args and evt.args[0].lower() == "--qr"
+    if show_qr:
+        if not qrcode:
+            await evt.reply("Can't generate QR code: qrcode and/or PIL not installed")
+            return
+        evt.args = evt.args[1:]
+
+    puppet = await _get_puppet_from_cmd(evt)
+    if not puppet:
+        return
+
+    resp = await evt.bridge.signal.get_identities(evt.sender.username, puppet.address)
+    if not resp.identities:
+        await evt.reply(f"No identities found for {_pill(puppet)}")
+        return
+    most_recent = resp.identities[0]
+    for identity in resp.identities:
+        if identity.added > most_recent.added:
+            most_recent = identity
+    uuid = most_recent.address.uuid or "unknown"
+    await evt.reply(f"### {puppet.name}\n\n"
+                    f"**UUID:** {uuid}  \n"
+                    f"**Trust level:** {most_recent.trust_level}  \n"
+                    f"**Safety number:**\n"
+                    f"```\n{_format_safety_number(most_recent.safety_number)}\n```")
+    if show_qr and most_recent.qr_code_data:
+        data = base64.b64decode(most_recent.qr_code_data)
+        content = await make_qr(evt.az.intent, data, "verification-qr.png")
+        await evt.az.intent.send_message(evt.room_id, content)