Add command to list and terminate sessions (ref #249)

mautrix_telegram/commands/telegram/account.py

@@ -16,8 +16,11 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from typing import Dict, Optional
 
-from telethon.errors import UsernameInvalidError, UsernameNotModifiedError, UsernameOccupiedError
-from telethon.tl.functions.account import UpdateUsernameRequest
+from telethon.errors import (UsernameInvalidError, UsernameNotModifiedError, UsernameOccupiedError,
+                             HashInvalidError)
+from telethon.tl.types import Authorization
+from telethon.tl.functions.account import (UpdateUsernameRequest, GetAuthorizationsRequest,
+                                           ResetAuthorizationRequest)
 
 from mautrix_telegram.commands import command_handler, CommandEvent, SECTION_AUTH
 
@@ -48,3 +51,52 @@ async def username(evt: CommandEvent) -> Optional[Dict]:
         await evt.reply("Username removed")
     else:
         await evt.reply(f"Username changed to {evt.sender.username}")
+
+
+def _format_session(sess: Authorization) -> str:
+    return (f"**{sess.app_name} {sess.app_version}**  \n"
+            f"  **Platform:** {sess.device_model} {sess.platform} {sess.system_version}  \n"
+            f"  **Active:** {sess.date_active} (created {sess.date_created})  \n"
+            f"  **From:** {sess.ip} - {sess.region}, {sess.country}")
+
+
+@command_handler(needs_auth=True,
+                 help_section=SECTION_AUTH,
+                 help_args="<`list`|`terminate`> [_hash_]",
+                 help_text="View or delete other Telegram sessions.")
+async def session(evt: CommandEvent) -> Optional[Dict]:
+    if len(evt.args) == 0:
+        return await evt.reply("**Usage:** `$cmdprefix+sp session <list|terminate> [hash]`")
+    elif evt.sender.is_bot:
+        return await evt.reply("Bots can't manage their sessions")
+    cmd = evt.args[0].lower()
+    if cmd == "list":
+        res = await evt.sender.client(GetAuthorizationsRequest())
+        session_list = res.authorizations
+        current = [s for s in session_list if s.current][0]
+        current_text = _format_session(current)
+        other_text = "\n".join(f"* {_format_session(sess)}  \n"
+                               f"  **Hash:** {sess.hash}"
+                               for sess in session_list if not sess.current)
+        return await evt.reply(f"### Current session\n"
+                               f"{current_text}\n"
+                               f"\n"
+                               f"### Other active sessions\n"
+                               f"{other_text}")
+    elif cmd == "terminate" and len(evt.args) > 1:
+        try:
+            session_hash = int(evt.args[1])
+        except ValueError:
+            return await evt.reply("Hash must be a positive integer")
+        if session_hash <= 0:
+            return await evt.reply("Hash must be a positive integer")
+        try:
+            ok = await evt.sender.client(ResetAuthorizationRequest(hash=session_hash))
+        except HashInvalidError:
+            return await evt.reply("Invalid session hash.")
+        if ok:
+            return await evt.reply("Session terminated successfully.")
+        else:
+            return await evt.reply("Session not found.")
+    else:
+        return await evt.reply("**Usage:** `$cmdprefix+sp session <list|terminate> [hash]`")