caelebfi/blinkpy
Archived
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Move all login handling to auth class

Browse Source 
- Update quick start guide with new flow
- If token given at startup don't bother with logging in (token will be
  auto-refreshed on failed server request anyways)
- Update to v4 login by default (2FA endpoint)
This commit is contained in:
Kevin Fronczak
2020-05-25 01:51:04 +00:00
parent 7fc63e5f23
commit f7bc4c7f93
5 changed files with 91 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.rst
+51 -24
View File
@@ -45,59 +45,86 @@ This library was built with the intention of allowing easy communication with Bl
Quick Start
=============
The simplest way to use this package from a terminal is to call ``Blink.start()`` which will prompt for your Blink username and password and then log you in. Alternatively, you can instantiate the Blink class with a username and password, and call ``Blink.start()`` to login and setup without prompt, as shown below. In addition, http requests are throttled internally via use of the ``Blink.refresh_rate`` variable, which can be set at initialization and defaults to 30 seconds.
The simplest way to use this package from a terminal is to call ``Blink.start()`` which will prompt for your Blink username and password and then log you in. In addition, http requests are throttled internally via use of the ``Blink.refresh_rate`` variable, which can be set at initialization and defaults to 30 seconds.
.. code:: python
from blinkpy import blinkpy
blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD', refresh_rate=30)
from blinkpy.blinkpy import Blink
blink = Blink()
blink.start()
At startup, you may be prompted for a verification key. Just enter this in the command-line prompt. If you just receive a verification email asking to validate access for your device, enter nothing at this prompt. To avoid any command-line interaction, call the ``Blink`` class with the ``no_prompt=True`` flag. Instead, once you receive the verification email, call the following functions:
This flow will prompt you for your username and password. Once entered, if you likely will need to send a 2FA key to the blink servers (this pin is sent to your email address). When you receive this pin, enter at the prompt and the Blink library will proceed with setup.
Starting blink without a prompt
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
In some cases, having an interactive command-line session is not desired. In this case, you will need to set the ``Blink.auth.no_prompt`` value to ``True``. In addition, since you will not be prompted with a username and password, you must supply the login data to the blink authentication handler. This is best done by instantiating your own auth handler with a dictionary containing at least your username and password.
.. code:: python
blink.login_handler.send_auth_key(blink, VERIFICATION_KEY)
from blinkpy.blinkpy import Blink
from blinkpy.auth import Auth
blink = Blink()
# Can set no_prompt when initializing auth handler
auth = Auth({"username": <your username>, "password": <your password>}, no_prompt=True)
blink.auth = auth
blink.start()
Since you will not be prompted for any 2FA pin, you must call the ``blink.auth.send_auth_key`` function. There are two required parameters: the ``blink`` object as well as the ``key`` you received from Blink for 2FA:
.. code:: python
auth.send_auth_key(blink, <your key>)
blink.setup_post_verify()
In addition, you can also save your credentials in a json file and initialize Blink with the credential file as follows:
Supplying credentials from file
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Other use cases may involved loading credentials from a file. This file must be ``json`` formatted and contain a minimum of ``username`` and ``password``. A built in function in the ``blinkpy.helpers.util`` module can aid in loading this file. Note, if ``no_prompt`` is desired, a similar flow can be followed as above.
.. code:: python
from blinkpy import blinkpy
blink = blinkpy.Blink(cred_file="path/to/credentials.json")
from blinkpy.blinkpy import Blink
from blinkpy.auth import Auth
from blinkpy.helpers.util import json_load
blink = Blink()
auth = Auth(json_load("<File Location>"))
blink.auth = auth
blink.start()
The credential file must be json formatted with a ``username`` and ``password`` key like follows:
.. code:: json
Saving credentials
~~~~~~~~~~~~~~~~~~~
This library also allows you to save your credentials to use in future sessions. Saved information includes authentication tokens as well as unique ids which should allow for a more streamlined experience and limits the frequency of login requests. This data can be saved as follows (it can then be loaded by following the instructions above for supplying credentials from a file):
{
"username": "YOUR USER NAME",
"password": "YOUR PASSWORD"
}
.. code:: python
blink.save("<File location>")
Getting cameras
~~~~~~~~~~~~~~~~
Cameras are instantiated as individual ``BlinkCamera`` classes within a ``BlinkSyncModule`` instance. All of your sync modules are stored within the ``Blink.sync`` dictionary and can be accessed using the name of the sync module as the key (this is the name of your sync module in the Blink App).
The below code will display cameras and their available attributes:
.. code:: python
from blinkpy import blinkpy
blink = blinkpy.Blink(username='YOUR USER NAME', password='YOUR PASSWORD')
blink.start()
for name, camera in blink.cameras.items():
print(name) # Name of the camera
print(camera.attributes) # Print available attributes of camera
The most recent images and videos can be accessed as a bytes-object via internal variables. These can be updated with calls to ``Blink.refresh()`` but will only make a request if motion has been detected or other changes have been found. This can be overridden with the ``force_cache`` flag, but this should be used for debugging only since it overrides the internal request throttling.
The most recent images and videos can be accessed as a bytes-object via internal variables. These can be updated with calls to ``Blink.refresh()`` but will only make a request if motion has been detected or other changes have been found. This can be overridden with the ``force`` flag, but this should be used for debugging only since it overrides the internal request throttling.
.. code:: python
camera = blink.cameras['SOME CAMERA NAME']
blink.refresh(force_cache=True) # force a cache update USE WITH CAUTION
blink.refresh(force=True) # force a cache update USE WITH CAUTION
camera.image_from_cache.raw # bytes-like image object (jpg)
camera.video_from_cache.raw # bytes-like video object (mp4)
@@ -110,15 +137,15 @@ The ``blinkpy`` api also allows for saving images and videos to a file and snapp
blink.refresh() # Get new information from server
camera.image_to_file('/local/path/for/image.jpg')
camera.video_to_file('/local/path/for/video.mp4')
Download videos
~~~~~~~~~~~~~~~~
You can also use this library to download all videos from the server. In order to do this, you must specify a ``path``. You may also specifiy a how far back in time to go to retrieve videos via the ``since=`` variable (a simple string such as ``"2017/09/21"`` is sufficient), as well as how many pages to traverse via the ``page=`` variable. Note that by default, the library will search the first ten pages which is sufficient in most use cases. Additionally, you can specidy one or more cameras via the ``camera=`` property. This can be a single string indicating the name of the camera, or a list of camera names. By default, it is set to the string ``'all'`` to grab videos from all cameras.
Example usage, which downloads all videos recorded since July 4th, 2018 at 9:34am to the ``/home/blink`` directory:
.. code:: python
blink = blinkpy.Blink(username="YOUR USER NAME", password="YOUR PASSWORD")
blink.start()
blink.download_videos('/home/blink', since='2018/07/04 09:34')
blinkpy/auth.py
+27 -21
View File
@@ -18,33 +18,33 @@ class Auth:
Initialize auth handler.
:param login_data: dictionary for login data
can contain the following:
must contain the following:
- username
- password
- device_id
- uid
- notification_key
:param no_prompt: Should any user input prompts
be supressed? True/FALSE
"""
if login_data is None:
login_data = {}
self.data = login_data
self.token = None
self.host = None
self.region_id = None
self.token = login_data.get("token", None)
self.host = login_data.get("host", None)
self.region_id = login_data.get("region_id", None)
self.client_id = login_data.get("client_id", None)
self.account_id = login_data.get("account_id", None)
self.login_response = None
self.no_prompt = no_prompt
self.session = self.create_session()
self.validate_login()
@property
def login_attributes(self):
"""Return a dictionary of login attributes."""
attributes = {
"token": self.token,
"host": self.host,
"region_id": self.region_id,
}
return util.merge_dicts(attributes, self.data)
self.data["token"] = self.token
self.data["host"] = self.host
self.data["region_id"] = self.region_id
self.data["client_id"] = self.client_id
self.data["account_id"] = self.account_id
return self.data
@property
def header(self):
@@ -66,10 +66,8 @@ class Auth:
def validate_login(self):
"""Check login information and prompt if not available."""
if "username" not in self.data:
self.data["username"] = None
if "password" not in self.data:
self.data["password"] = None
self.data["username"] = self.data.get("username", None)
self.data["password"] = self.data.get("password", None)
if not self.no_prompt:
self.data = util.prompt_login_data(self.data)
@@ -77,6 +75,7 @@ class Auth:
def login(self, login_url=LOGIN_ENDPOINT):
"""Attempt login to blink servers."""
self.validate_login()
_LOGGER.info("Attempting login with %s", login_url)
response = api.request_login(self, login_url, self.data, is_retry=False,)
try:
@@ -91,15 +90,22 @@ class Auth:
try:
_LOGGER.info("Token expired, attempting automatic refresh.")
self.login_response = self.login()
region_ids = (*self.login_response["region"],)
self.region_id = region_ids[0]
self.region_id = self.login_response["region"]["tier"]
self.host = f"{self.region_id}.{BLINK_URL}"
self.token = self.login_response["authtoken"]["authtoken"]
self.client_id = self.login_response["client"]["id"]
self.account_id = self.login_response["account"]["id"]
except KeyError:
_LOGGER.error("Malformed login response: %s", self.login_response)
raise TokenRefreshFailed
return True
def startup(self):
"""Initialize tokens for communication."""
self.validate_login()
if None in self.login_attributes.values():
self.refresh_token()
def validate_response(self, response, json_resp):
"""Check for valid response."""
if not json_resp:
@@ -175,7 +181,7 @@ class Auth:
try:
if self.login_response["client"]["verification_required"]:
return True
except KeyError:
except (KeyError, TypeError):
pass
return False
blinkpy/blinkpy.py
+6 -52
View File
@@ -28,7 +28,6 @@ from blinkpy.helpers import util
from blinkpy.helpers.constants import (
DEFAULT_MOTION_INTERVAL,
DEFAULT_REFRESH,
DEVICE_ID,
MIN_THROTTLE_TIME,
)
from blinkpy.helpers.constants import __version__
@@ -42,45 +41,19 @@ class Blink:
"""Class to initialize communication."""
def __init__(
self,
data=None,
username=None,
password=None,
refresh_rate=DEFAULT_REFRESH,
motion_interval=DEFAULT_MOTION_INTERVAL,
no_prompt=False,
cred_file=None,
device_id=DEVICE_ID,
self, refresh_rate=DEFAULT_REFRESH, motion_interval=DEFAULT_MOTION_INTERVAL,
):
"""
Initialize Blink system.
:param data: Blink login data. Ignores username, password,
and device_id if supplied.
:param username: Blink username (usually email address)
:param password: Blink password
:param refresh_rate: Refresh rate of blink information.
Defaults to 15 (seconds)
:param motion_interval: How far back to register motion in minutes.
Defaults to last refresh time.
Useful for preventing motion_detected property
from de-asserting too quickly.
:param no_prompt: Set to TRUE if using an implementation that needs to
suppress command-line output.
:param device_id: Identifier for the application. Default is 'Blinkpy'.
This is used when logging in and should be changed to
fit the implementation (ie. "Home Assistant" in a
Home Assistant integration).
:param cred_file: JSON formatted credential file (containing username and password).
"""
if data is None:
data = {}
data["username"] = username
data["password"] = password
data["device_id"] = device_id
data = self.check_cred_file(cred_file, data)
self.auth = Auth(login_data=data, no_prompt=no_prompt)
self.auth = Auth()
self.account_id = None
self.client_id = None
self.network_ids = []
@@ -93,7 +66,6 @@ class Blink:
self.video_list = CaseInsensitiveDict({})
self.motion_interval = motion_interval
self.version = __version__
self.no_prompt = no_prompt
self.available = False
self.key_required = False
@@ -121,7 +93,7 @@ class Blink:
def start(self):
"""Perform full system setup."""
try:
self.auth.refresh_token()
self.auth.startup()
self.setup_login_ids()
self.setup_urls()
except (LoginError, TokenRefreshFailed, BlinkSetupError):
@@ -131,7 +103,7 @@ class Blink:
self.key_required = self.auth.check_key_required()
if self.key_required:
if self.no_prompt:
if self.auth.no_prompt:
return True
self.setup_prompt_2fa()
return self.setup_post_verify()
@@ -186,12 +158,8 @@ class Blink:
def setup_login_ids(self):
"""Retrieve login id numbers from login response."""
try:
self.client_id = self.auth.login_response["client"]["id"]
self.account_id = self.auth.login_response["account"]["id"]
except KeyError:
_LOGGER.error("Malformed login response of %s", self.auth.login_response)
raise BlinkSetupError
self.client_id = self.auth.client_id
self.account_id = self.auth.account_id
def setup_urls(self):
"""Create urls for api."""
@@ -246,20 +214,6 @@ class Blink:
combined = util.merge_dicts(combined, self.sync[sync].cameras)
return combined
def check_cred_file(self, cred_file, data):
"""Check if cred file is supplied and load data is so."""
if cred_file:
cred_data = util.json_load(cred_file)
try:
data["username"] = cred_data["username"]
data["password"] = cred_data["password"]
except KeyError:
_LOGGER.error(
"Supplied cred file must contain 'username' and 'password' keys."
)
return data
def save(self, file_name):
"""Save login data to file."""
util.json_save(self.auth.login_attributes, file_name)
blinkpy/helpers/constants.py
+2 -2
View File
@@ -48,9 +48,9 @@ URLS
BLINK_URL = "immedia-semi.com"
DEFAULT_URL = "{}.{}".format("rest-prod", BLINK_URL)
BASE_URL = "https://{}".format(DEFAULT_URL)
LOGIN_ENDPOINT = "{}/api/v3/login".format(BASE_URL)
LOGIN_ENDPOINT = "{}/api/v4/account/login".format(BASE_URL)
LOGIN_URLS = [
"{}/api/v4/login".format(BASE_URL),
"{}/api/v4/account/login".format(BASE_URL),
"{}/api/v3/login".format(BASE_URL),
]
blinkpy/helpers/util.py
+5 -8
View File
@@ -79,14 +79,11 @@ def prompt_login_data(data):
def validate_login_data(data):
"""Check for missing keys."""
valid_keys = {
"uid": gen_uid(const.SIZE_UID),
"notification_key": gen_uid(const.SIZE_NOTIFICATION_KEY),
"device_id": const.DEVICE_ID,
}
for key in valid_keys:
if key not in data:
data[key] = valid_keys[key]
data["uid"] = data.get("uid", gen_uid(const.SIZE_UID))
data["notification_key"] = data.get(
"notification_key", gen_uid(const.SIZE_NOTIFICATION_KEY)
)
data["device_id"] = data.get("device_id", const.DEVICE_ID)
return data