Adds ability to login via access token. Adds whoami cli method to pri…

…nt who you are logged in as currently

fractal/matrix/async_client.py

@@ -1,6 +1,6 @@
 import logging
 import os
-from typing import Any, Callable, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 import aiohttp
 from aiofiles import open as aiofiles_open
@@ -21,7 +21,11 @@
 )
 from nio.responses import RegisterErrorResponse
 
-from .exceptions import GetLatestSyncTokenError
+from .exceptions import (
+    GetLatestSyncTokenError,
+    UnknownDiscoveryInfoException,
+    WellKnownNotFoundException,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -292,7 +296,7 @@ def __init__(
 
         if not self.homeserver_url and not self.matrix_id:
             raise KeyError(
-                "Environment variable MATRIX_HOMESERVER_URL or MATRIX_ID must be set if\
+                "Environment variable MATRIX_HOMESERVER_URL and MATRIX_ID must be set if\
     not passed explicitly to the MatrixClient context manager decorator."
             )
 
@@ -320,7 +324,7 @@ async def __aexit__(self, exc_type, exc_value, traceback):
         await self.client.close()
 
 
-async def get_homeserver_for_matrix_id(matrix_id: str):
+async def get_homeserver_for_matrix_id(matrix_id: str) -> str:
     """Lookup the homeserver url associated with a Matrix ID"""
     # FIXME: just because matrix_id has localhost, doesn't necessarily mean
     # that the homeserver is running on localhost. Could be synapse:8008, etc.
@@ -333,5 +337,5 @@ async def get_homeserver_for_matrix_id(matrix_id: str):
     if not res.transport_response.ok:  # type: ignore
         if res.transport_response.reason == "Not Found":  # type: ignore
             raise WellKnownNotFoundException()  # type: ignore
-        raise UnknownDiscoveryInfoException(f"Failed to get homeserver for MatrixID: {res.transport_response.reason}")  # type: ignore
+        raise UnknownDiscoveryInfoException(res.transport_response.reason)  # type: ignore
     return res.homeserver_url  # type: ignore

fractal/matrix/controllers/auth.py

@@ -7,7 +7,7 @@
 from fractal.cli.utils import read_user_data, write_user_data
 from fractal.matrix import MatrixClient, get_homeserver_for_matrix_id
 from fractal.matrix.utils import prompt_matrix_password
-from nio import LoginError
+from nio import LoginError, WhoamiError
 
 
 class MatrixLoginError(Exception):
@@ -19,19 +19,35 @@ class AuthController:
     TOKEN_FILE = "matrix.creds.yaml"
 
     @cli_method
-    def login(self, matrix_id: str, homeserver_url: Optional[str] = None):
+    def login(
+        self,
+        matrix_id: Optional[str] = None,
+        homeserver_url: Optional[str] = None,
+        access_token: Optional[str] = None,
+    ):
         """
         Login to a Matrix homeserver.
         ---
         Args:
             matrix_id: Matrix ID of user to login as
             homeserver_url: Homeserver to login to
+            access_token: Access token to use for login.
 
         """
-
-        homeserver_url, access_token = async_to_sync(self._login_with_password)(
-            matrix_id, homeserver_url=homeserver_url
-        )
+        if not access_token:
+            if not matrix_id:
+                print("Please provide a matrix ID.")
+                exit(1)
+            homeserver_url, access_token = async_to_sync(self._login_with_password)(
+                matrix_id, homeserver_url=homeserver_url
+            )
+        else:
+            if not homeserver_url:
+                print("Please provide a --homeserver-url if logging in with an access token.")
+                exit(1)
+            matrix_id, homeserver_url, access_token = async_to_sync(
+                self._login_with_access_token
+            )(access_token, homeserver_url=homeserver_url)
 
         # save access token to token file
         write_user_data(
@@ -47,6 +63,28 @@ def login(self, matrix_id: str, homeserver_url: Optional[str] = None):
 
     login.clicz_aliases = ["login"]
 
+    @cli_method
+    def whoami(self):
+        """
+        Get information about the current logged in user.
+        ---
+        Args:
+        """
+        try:
+            data, _ = read_user_data(self.TOKEN_FILE)
+        except (KeyError, FileNotFoundError):
+            print("You are not logged in.")
+            exit(1)
+
+        try:
+            homeserver_url = data["homeserver_url"]
+            matrix_id = data["matrix_id"]
+        except KeyError:
+            print("You are not logged in.")
+            exit(1)
+
+        print(f"You are logged in as {matrix_id} on {homeserver_url}")
+
     @cli_method
     def logout(self):
         """
@@ -75,6 +113,16 @@ async def _logout():
 
     logout.clicz_aliases = ["logout"]
 
+    async def _login_with_access_token(
+        self, access_token: str, homeserver_url: str
+    ) -> Tuple[str, str, str]:
+        async with MatrixClient(homeserver_url, access_token) as client:
+            res = await client.whoami()
+            if isinstance(res, WhoamiError):
+                raise MatrixLoginError(res.message)
+            matrix_id = res.user_id
+        return matrix_id, homeserver_url, access_token
+
     async def _login_with_password(
         self, matrix_id: str, password: Optional[str] = None, homeserver_url: Optional[str] = None
     ) -> Tuple[str, str]: