Add support for Plex OAuth (#974)

* Add method to return Plex OAuth url

* Add helper utils function for Plex OAuth
This commit is contained in:
JonnyWong16 2022-07-20 19:48:52 -07:00 committed by GitHub
parent 3107e25a0e
commit 9f7f11ba9c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 67 additions and 4 deletions

View file

@ -1417,11 +1417,12 @@ class MyPlexPinLogin:
session (requests.Session, optional): Use your own session object if you want to
cache the http responses from PMS
requestTimeout (int): timeout in seconds on initial connect to plex.tv (default config.TIMEOUT).
headers (dict): A dict of X-Plex headers to send with requests.
oauth (bool): True to use Plex OAuth instead of PIN login.
Attributes:
PINS (str): 'https://plex.tv/api/v2/pins'
CHECKPINS (str): 'https://plex.tv/api/v2/pins/{pinid}'
LINK (str): 'https://plex.tv/api/v2/pins/link'
POLLINTERVAL (int): 1
finished (bool): Whether the pin login has finished or not.
expired (bool): Whether the pin login has expired or not.
@ -1432,12 +1433,13 @@ class MyPlexPinLogin:
CHECKPINS = 'https://plex.tv/api/v2/pins/{pinid}' # get
POLLINTERVAL = 1
def __init__(self, session=None, requestTimeout=None, headers=None):
def __init__(self, session=None, requestTimeout=None, headers=None, oauth=False):
super(MyPlexPinLogin, self).__init__()
self._session = session or requests.Session()
self._requestTimeout = requestTimeout or TIMEOUT
self.headers = headers
self._oauth = oauth
self._loginTimeout = None
self._callback = None
self._thread = None
@ -1452,8 +1454,36 @@ class MyPlexPinLogin:
@property
def pin(self):
""" Return the 4 character PIN used for linking a device at https://plex.tv/link. """
if self._oauth:
raise BadRequest('Cannot use PIN for Plex OAuth login')
return self._code
def oauthUrl(self, forwardUrl=None):
""" Return the Plex OAuth url for login.
Parameters:
forwardUrl (str, optional): The url to redirect the client to after login.
"""
if not self._oauth:
raise BadRequest('Must use "MyPlexPinLogin(oauth=True)" for Plex OAuth login.')
headers = self._headers()
params = {
'clientID': headers['X-Plex-Client-Identifier'],
'context[device][product]': headers['X-Plex-Product'],
'context[device][version]': headers['X-Plex-Version'],
'context[device][platform]': headers['X-Plex-Platform'],
'context[device][platformVersion]': headers['X-Plex-Platform-Version'],
'context[device][device]': headers['X-Plex-Device'],
'context[device][deviceName]': headers['X-Plex-Device-Name'],
'code': self._code
}
if forwardUrl:
params['forwardUrl'] = forwardUrl
return f'https://app.plex.tv/auth/#!?{urlencode(params)}'
def run(self, callback=None, timeout=None):
""" Starts the thread which monitors the PIN login state.
Parameters:
@ -1517,7 +1547,13 @@ class MyPlexPinLogin:
def _getCode(self):
url = self.PINS
response = self._query(url, self._session.post)
if self._oauth:
params = {'strong': True}
else:
params = None
response = self._query(url, self._session.post, params=params)
if not response:
return None

View file

@ -394,7 +394,7 @@ def getMyPlexAccount(opts=None): # pragma: no cover
def createMyPlexDevice(headers, account, timeout=10): # pragma: no cover
""" Helper function to create a new MyPlexDevice.
""" Helper function to create a new MyPlexDevice. Returns a new MyPlexDevice instance.
Parameters:
headers (dict): Provide the X-Plex- headers for the new device.
@ -417,6 +417,33 @@ def createMyPlexDevice(headers, account, timeout=10): # pragma: no cover
return account.device(clientId=clientIdentifier)
def plexOAuth(headers, forwardUrl=None, timeout=120): # pragma: no cover
""" Helper function for Plex OAuth login. Returns a new MyPlexAccount instance.
Parameters:
headers (dict): Provide the X-Plex- headers for the new device.
A unique X-Plex-Client-Identifier is required.
forwardUrl (str, optional): The url to redirect the client to after login.
timeout (int, optional): Timeout in seconds to wait for device login. Default 120 seconds.
"""
from plexapi.myplex import MyPlexAccount, MyPlexPinLogin
if 'X-Plex-Client-Identifier' not in headers:
raise BadRequest('The X-Plex-Client-Identifier header is required.')
pinlogin = MyPlexPinLogin(headers=headers, oauth=True)
print('Login to Plex at the following url:')
print(pinlogin.oauthUrl(forwardUrl))
pinlogin.run(timeout=timeout)
pinlogin.waitForLogin()
if pinlogin.token:
print('Login successful!')
return MyPlexAccount(token=pinlogin.token)
else:
print('Login failed.')
def choose(msg, items, attr): # pragma: no cover
""" Command line helper to display a list of choices, asking the
user to choose one of the options.