# Copyright Mirage authors & contributors # SPDX-License-Identifier: LGPL-3.0-or-later import asyncio from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import parse_qs, quote, urlparse from . import __display_name__ _SUCCESS_HTML_PAGE = """ """ + __display_name__ + """
✓
""" class _SSORequestHandler(BaseHTTPRequestHandler): def do_GET(self) -> None: self.server: "SSOServer" redirect = "%s/_matrix/client/r0/login/sso/redirect?redirectUrl=%s" % ( self.server.for_homeserver, quote(self.server.url_to_open), ) parameters = parse_qs(urlparse(self.path).query) if "loginToken" in parameters: self.server._token = parameters["loginToken"][0] self.send_response(200) # OK self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(_SUCCESS_HTML_PAGE.encode()) else: self.send_response(308) # Permanent redirect, same method only self.send_header("Location", redirect) self.end_headers() self.close_connection = True class SSOServer(HTTPServer): """Local HTTP server to retrieve a SSO login token. Call `SSOServer.wait_for_token()` in a background task to start waiting for a SSO login token from the Matrix homeserver. Once the task is running, the user must open `SSOServer.url_to_open` in their browser, where they will be able to complete the login process. Once they are done, the homeserver will call us back with a login token and the `SSOServer.wait_for_token()` task will return. """ def __init__(self, for_homeserver: str) -> None: self.for_homeserver: str = for_homeserver self._token: str = "" # Pick the first available port super().__init__(("127.0.0.1", 0), _SSORequestHandler) @property def url_to_open(self) -> str: """URL for the user to open in their browser, to do the SSO process.""" return f"http://{self.server_address[0]}:{self.server_port}" async def wait_for_token(self) -> str: """Wait until the homeserver gives us a login token and return it.""" loop = asyncio.get_event_loop() while not self._token: await loop.run_in_executor(None, self.handle_request) return self._token