plex: Add auth v2

Closes #356, #245

Author: iamkroot

trakt_scrobbler/cli/plex.py

@@ -4,43 +4,11 @@
 from rich.prompt import Prompt
 
 from trakt_scrobbler import logger
-from trakt_scrobbler.utils import safe_request
-
 from .console import console
 
 app = typer.Typer(help="Runs the authentication flow for plex.")
 
 
-def plex_token_auth(login, password):
-    auth_params = {
-        "url": "https://plex.tv/users/sign_in.json",
-        "data": {"user[login]": login, "user[password]": password},
-        "headers": {
-            "X-Plex-Client-Identifier": "com.iamkroot.trakt_scrobbler",
-            "X-Plex-Product": "Trakt Scrobbler",
-            "Accept": "application/json",
-        },
-    }
-    return safe_request("post", auth_params)
-
-
-def get_token():
-    logger.info("Retrieving plex token")
-    login = Prompt.ask("Plex login ID", console=console)
-    pwd = Prompt.ask("Plex password", password=True, console=console)
-    resp = plex_token_auth(login, pwd)
-    if resp:
-        return resp.json()["user"]["authToken"]
-    elif resp is not None:
-        err_msg = resp.json().get("error", resp.text)
-        console.print(err_msg, style="error")
-        logger.error(err_msg)
-        return None
-    else:
-        logger.error("Unable to get access token")
-        return None
-
-
 @app.command(name="auth", help="Runs the authentication flow for Plex")
 def auth(
     force: Annotated[
@@ -60,6 +28,7 @@ def auth(
         ),
     ] = False,
 ):
+    from trakt_scrobbler.player_monitors.plex import PlexAuth
     from trakt_scrobbler.player_monitors.plex import token as token_store
 
     if force or token:  # token implies force
@@ -69,14 +38,14 @@ def auth(
     if token:
         # TODO: Verify that token is valid
         token_store.data = Prompt.ask("[question]Enter token[/]", console=console)
+        console.print("Plex token is saved.", style="info")
     elif not token_store:
-        token_data = get_token()
-        if token_data:
-            token_store.data = token_data
+        plex_auth = PlexAuth()
+        if plex_auth.device_auth():
             logger.info("Saved plex token")
-
-    if token_store:
-        console.print("Plex token is saved.", style="info")
+            console.print("Plex token is saved.", style="info")
+        else:
+            console.print("Failed to retrieve plex token.", style="error")
+            raise typer.Exit(1)
     else:
-        console.print("Failed to retrieve plex token.", style="error")
-        raise typer.Exit(1)
+        console.print("Plex token is saved.", style="info")

trakt_scrobbler/player_monitors/plex.py

@@ -1,3 +1,6 @@
+import os
+import time
+import webbrowser
 from json.decoder import JSONDecodeError
 import confuse
 from requests import HTTPError
@@ -15,6 +18,7 @@ class PlexToken:
 
     TODO: Use some form of OS-provided encrypted storage
     """
+
     PATH = DATA_DIR / "plex_token.txt"
     OLD_PATH = DATA_DIR / "plex_token.json"
 
@@ -52,6 +56,84 @@ def __bool__(self):
 token = PlexToken()
 
 
+class PlexAuth:
+    # https://forums.plex.tv/t/authenticating-with-plex/609370
+    API_URL = "https://plex.tv/api/v2"
+    CLIENT_ID = (
+        "com.iamkroot.trakt_scrobbler"  # Reuse from CLI or consistently use this
+    )
+    PRODUCT = "Trakt Scrobbler"
+
+    def __init__(self):
+        self._token_store = token
+
+    def get_pin(self):
+        params = {
+            "url": f"{self.API_URL}/pins",
+            "headers": {"Accept": "application/json"},
+            "data": {
+                "strong": "true",
+                "X-Plex-Product": self.PRODUCT,
+                "X-Plex-Client-Identifier": self.CLIENT_ID,
+            },
+        }
+        resp = safe_request("post", params)
+        return resp.json() if resp else None
+
+    def device_auth(self):
+        pin_data = self.get_pin()
+        if not pin_data:
+            logger.error("Could not get Plex PIN.")
+            return False
+
+        auth_url = (
+            f"https://app.plex.tv/auth#?clientID={self.CLIENT_ID}&code={pin_data['code']}"
+            f"&context[device][product]={self.PRODUCT}"
+        )
+
+        logger.info(f"Verification URL: {auth_url}")
+        notify(
+            "Opening browser for Plex authentication...", stdout=True, category="plex"
+        )
+
+        term_bak = os.environ.pop("TERM", None)
+        webbrowser.open(auth_url)
+        if term_bak is not None:
+            os.environ["TERM"] = term_bak
+
+        # Poll for token
+        pin_id = pin_data["id"]
+        check_url = f"{self.API_URL}/pins/{pin_id}"
+
+        start_time = time.time()
+        poll_interval = 2
+        # at max 2 minutes
+        expires_in = min(pin_data.get("expiresIn", 120), 120)
+
+        while time.time() - start_time < expires_in:
+            params = {
+                "url": check_url,
+                "headers": {"Accept": "application/json"},
+                "params": {"X-Plex-Client-Identifier": self.CLIENT_ID},
+            }
+            resp = safe_request("get", params)
+            if resp and resp.status_code == 200:
+                data = resp.json()
+                if data.get("authToken"):
+                    self._token_store.data = data["authToken"]
+                    logger.info("Plex authenticated successfully.")
+                    notify(
+                        "Plex authenticated successfully.", stdout=True, category="plex"
+                    )
+                    return True
+
+            time.sleep(poll_interval)
+
+        logger.error("Plex authentication timed out.")
+        notify("Plex authentication timed out.", stdout=True, category="plex")
+        return False
+
+
 class PlexMon(WebInterfaceMon):
     name = "plex"
     exclude_import = False
@@ -62,7 +144,7 @@ class PlexMon(WebInterfaceMon):
         "ip": confuse.String(default="localhost"),
         "port": confuse.String(default="32400"),
         "poll_interval": confuse.Number(default=10),
-        "scrobble_user": confuse.String(default="")
+        "scrobble_user": confuse.String(default=""),
     }
 
     def __init__(self, scrobble_queue):
@@ -74,8 +156,9 @@ def __init__(self, scrobble_queue):
         self.token = token.data
         if not self.token:
             logger.error("Unable to retrieve plex token.")
-            notify("Unable to retrieve plex token. Rerun plex auth.",
-                   category="exception")
+            notify(
+                "Unable to retrieve plex token. Rerun plex auth.", category="exception"
+            )
             return
         super().__init__(scrobble_queue)
         self.sess.headers["Accept"] = "application/json"