# # This file is licensed under the Affero General Public License (AGPL) version 3. # # Copyright 2015, 2016 OpenMarket Ltd # Copyright (C) 2023 New Vector, Ltd # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as # published by the Free Software Foundation, either version 3 of the # License, or (at your option) any later version. # # See the GNU Affero General Public License for more details: # . # # Originally licensed under the Apache License, Version 2.0: # . # # [This file includes modifications made by New Vector Limited] # # import logging from typing import TYPE_CHECKING, cast from twisted.web.server import Request from synapse.api.constants import LoginType from synapse.api.errors import LoginError, SynapseError from synapse.api.urls import CLIENT_API_PREFIX from synapse.http.server import HttpServer, respond_with_html, respond_with_redirect from synapse.http.servlet import RestServlet, parse_string from synapse.http.site import SynapseRequest from ._base import client_patterns if TYPE_CHECKING: from synapse.server import HomeServer logger = logging.getLogger(__name__) class AuthRestServlet(RestServlet): """ Handles Client / Server API authentication in any situations where it cannot be handled in the normal flow (with requests to the same endpoint). Current use is for web fallback auth. """ PATTERNS = client_patterns(r"/auth/(?P[\w\.]*)/fallback/web") def __init__(self, hs: "HomeServer"): super().__init__() self.hs = hs self.auth = hs.get_auth() self.auth_handler = hs.get_auth_handler() self.registration_handler = hs.get_registration_handler() self.recaptcha_template = hs.config.captcha.recaptcha_template self.terms_template = hs.config.consent.terms_template self.registration_token_template = ( hs.config.registration.registration_token_template ) self.success_template = hs.config.registration.fallback_success_template async def on_GET(self, request: SynapseRequest, stagetype: str) -> None: session = parse_string(request, "session") if not session: raise SynapseError(400, "No session supplied") if ( self.hs.config.experimental.msc3861.enabled and stagetype == "org.matrix.cross_signing_reset" ): # If MSC3861 is enabled, we can assume self._auth is an instance of MSC3861DelegatedAuth # We import lazily here because of the authlib requirement from synapse.api.auth.msc3861_delegated import MSC3861DelegatedAuth auth = cast(MSC3861DelegatedAuth, self.auth) url = await auth.account_management_url() if url is not None: url = f"{url}?action=org.matrix.cross_signing_reset" else: url = await auth.issuer() respond_with_redirect(request, str.encode(url)) if stagetype == LoginType.RECAPTCHA: html = self.recaptcha_template.render( session=session, myurl="%s/v3/auth/%s/fallback/web" % (CLIENT_API_PREFIX, LoginType.RECAPTCHA), sitekey=self.hs.config.captcha.recaptcha_public_key, ) elif stagetype == LoginType.TERMS: html = self.terms_template.render( session=session, terms_url="%s_matrix/consent?v=%s" % ( self.hs.config.server.public_baseurl, self.hs.config.consent.user_consent_version, ), myurl="%s/v3/auth/%s/fallback/web" % (CLIENT_API_PREFIX, LoginType.TERMS), ) elif stagetype == LoginType.SSO: # Display a confirmation page which prompts the user to # re-authenticate with their SSO provider. html = await self.auth_handler.start_sso_ui_auth(request, session) elif stagetype == LoginType.REGISTRATION_TOKEN: html = self.registration_token_template.render( session=session, myurl=f"{CLIENT_API_PREFIX}/r0/auth/{LoginType.REGISTRATION_TOKEN}/fallback/web", ) else: raise SynapseError(404, "Unknown auth stage type") # Render the HTML and return. respond_with_html(request, 200, html) return None async def on_POST(self, request: Request, stagetype: str) -> None: session = parse_string(request, "session") if not session: raise SynapseError(400, "No session supplied") if stagetype == LoginType.RECAPTCHA: response = parse_string(request, "g-recaptcha-response") if not response: raise SynapseError(400, "No captcha response supplied") authdict = {"response": response, "session": session} try: await self.auth_handler.add_oob_auth( LoginType.RECAPTCHA, authdict, request.getClientAddress().host ) except LoginError as e: # Authentication failed, let user try again html = self.recaptcha_template.render( session=session, myurl="%s/v3/auth/%s/fallback/web" % (CLIENT_API_PREFIX, LoginType.RECAPTCHA), sitekey=self.hs.config.captcha.recaptcha_public_key, error=e.msg, ) else: # No LoginError was raised, so authentication was successful html = self.success_template.render() elif stagetype == LoginType.TERMS: authdict = {"session": session} try: await self.auth_handler.add_oob_auth( LoginType.TERMS, authdict, request.getClientAddress().host ) except LoginError as e: # Authentication failed, let user try again html = self.terms_template.render( session=session, terms_url="%s_matrix/consent?v=%s" % ( self.hs.config.server.public_baseurl, self.hs.config.consent.user_consent_version, ), myurl="%s/v3/auth/%s/fallback/web" % (CLIENT_API_PREFIX, LoginType.TERMS), error=e.msg, ) else: # No LoginError was raised, so authentication was successful html = self.success_template.render() elif stagetype == LoginType.SSO: # The SSO fallback workflow should not post here, raise SynapseError(404, "Fallback SSO auth does not support POST requests.") elif stagetype == LoginType.REGISTRATION_TOKEN: token = parse_string(request, "token", required=True) authdict = {"session": session, "token": token} try: await self.auth_handler.add_oob_auth( LoginType.REGISTRATION_TOKEN, authdict, request.getClientAddress().host, ) except LoginError as e: html = self.registration_token_template.render( session=session, myurl=f"{CLIENT_API_PREFIX}/r0/auth/{LoginType.REGISTRATION_TOKEN}/fallback/web", error=e.msg, ) else: html = self.success_template.render() else: raise SynapseError(404, "Unknown auth stage type") # Render the HTML and return. respond_with_html(request, 200, html) return None def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: AuthRestServlet(hs).register(http_server)