Add better OAuth Redirect handling to web adapters.

Author: EvieePy

twitchio/web/aio_adapter.py

@@ -316,10 +316,12 @@ async def fetch_token(self, request: web.Request) -> FetchTokenPayload:
         if "code" not in request.query:
             return FetchTokenPayload(400, response=web.Response(status=400, text="No 'code' parameter provided."))
 
+        redirect = self._find_redirect(request)
+
         try:
             resp: UserTokenPayload = await self.client._http.user_access_token(
                 request.query["code"],
-                redirect_uri=self.redirect_url,
+                redirect_uri=redirect,
             )
         except HTTPException as e:
             logger.error("Exception raised while fetching Token in <%s>: %s", self.__class__.__qualname__, e)
@@ -378,7 +380,22 @@ async def oauth_callback(self, request: web.Request) -> web.Response:
 
         return payload.response
 
+    def _find_redirect(self, request: web.Request) -> str:
+        stripped = self._domain.removeprefix(f"{self._proto}://")
+        local = f"{self._proto}://{self._host}"
+
+        if request.host.startswith((self._domain, stripped)):
+            redirect = self.redirect_url
+        elif request.host.startswith((self._host, local)):
+            redirect = f"{local}/oauth/callback"
+        else:
+            redirect = f"{request.scheme}://{request.host}/oauth/callback"
+
+        return redirect
+
     async def oauth_redirect(self, request: web.Request) -> web.Response:
+        redirect = self._find_redirect(request)
+
         scopes: str | None = request.query.get("scopes", request.query.get("scope", None))
         force_verify: bool = request.query.get("force_verify", "false").lower() == "true"
 
@@ -397,7 +414,7 @@ async def oauth_redirect(self, request: web.Request) -> web.Response:
         try:
             payload: AuthorizationURLPayload = self.client._http.get_authorization_url(
                 scopes=scopes_,
-                redirect_uri=self.redirect_url,
+                redirect_uri=redirect,
                 force_verify=force_verify,
             )
         except Exception as e:

twitchio/web/starlette_adapter.py

@@ -162,7 +162,7 @@ def __init__(
         if eventsub_secret and not 10 <= len(eventsub_secret) <= 100:
             raise ValueError("Eventsub Secret must be between 10 and 100 characters long.")
 
-        self._domain: str | None = None
+        self._domain: str
         self._proto = "https" if (ssl_keyfile or domain) else "http"
 
         if domain:
@@ -358,10 +358,12 @@ async def fetch_token(self, request: Request) -> FetchTokenPayload:
         if "code" not in request.query_params:
             return FetchTokenPayload(400, response=Response(status_code=400, content="No 'code' parameter provided."))
 
+        redirect = self._find_redirect(request)
+
         try:
             resp: UserTokenPayload = await self.client._http.user_access_token(
                 request.query_params["code"],
-                redirect_uri=self.redirect_url,
+                redirect_uri=redirect,
             )
         except HTTPException as e:
             logger.error("Exception raised while fetching Token in <%s>: %s", self.__class__.__qualname__, e)
@@ -380,6 +382,25 @@ async def fetch_token(self, request: Request) -> FetchTokenPayload:
             payload=resp,
         )
 
+    def _find_redirect(self, request: Request) -> str:
+        stripped = self._domain.removeprefix(f"{self._proto}://")
+        local = f"{self._proto}://{self._host}"
+
+        host = request.url.hostname
+        scheme = request.url.scheme
+
+        if not host:
+            return self.redirect_url
+
+        if host.startswith((self._domain, stripped)):
+            redirect = self.redirect_url
+        elif host.startswith((self._host, local)):
+            redirect = f"{local}:{self._port}/oauth/callback"
+        else:
+            redirect = f"{scheme}://{host}/oauth/callback"
+
+        return redirect
+
     async def oauth_callback(self, request: Request) -> Response:
         """Default route callback for the OAuth Authentication redirect URL.
 
@@ -423,6 +444,7 @@ async def oauth_callback(self, request: Request) -> Response:
     async def oauth_redirect(self, request: Request) -> Response:
         scopes: str | None = request.query_params.get("scopes", None)
         force_verify: bool = request.query_params.get("force_verify", "false").lower() == "true"
+        redirect = self._find_redirect(request)
 
         if not scopes:
             scopes = str(self.client._http.scopes) if self.client._http.scopes else None
@@ -439,7 +461,7 @@ async def oauth_redirect(self, request: Request) -> Response:
         try:
             payload: AuthorizationURLPayload = self.client._http.get_authorization_url(
                 scopes=scopes_,
-                redirect_uri=self.redirect_url,
+                redirect_uri=redirect,
                 force_verify=force_verify,
             )
         except Exception as e: