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

Updated login method to allow for email verification and 2FA in future

Browse Source
This commit is contained in:
Kevin Fronczak
2020-05-05 15:28:38 -04:00
parent c51551afc5
commit 0c79620baa
5 changed files with 79 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
blinkpy/api.py
+1
View File
@@ -36,6 +36,7 @@ def request_login(
"client_name": "Computer",
"client_type": "android",
"device_identifier": "Blinkpy",
"device_name": "Blinkpy",
"os_version": "5.1.1",
"reauth": "true",
}
blinkpy/blinkpy.py
+36 -18
View File
@@ -57,6 +57,7 @@ class Blink:
motion_interval=DEFAULT_MOTION_INTERVAL,
legacy_subdomain=False,
no_prompt=False,
persist_key=None,
):
"""
Initialize Blink system.
@@ -78,9 +79,13 @@ class Blink:
api issues).
:param no_prompt: Set to TRUE if using an implementation that needs to
suppress command-line output.
:param persist_key: Location of persistant identifier.
"""
self.login_handler = LoginHandler(
username=username, password=password, cred_file=cred_file
username=username,
password=password,
cred_file=cred_file,
persist_key=persist_key,
)
self._token = None
self._auth_header = None
@@ -126,24 +131,29 @@ class Blink:
if self.key_required and not self.no_prompt:
email = self.login_handler.data["username"]
key = input("Enter code sent to {}: ".format(email))
self.login_handler.send_auth_key(self, key)
result = self.login_handler.send_auth_key(self, key)
self.key_required = not result
self.setup_post_verify()
elif not self.key_required:
self.setup_post_verify()
if self.available:
camera_list = self.get_cameras()
networks = self.get_ids()
for network_name, network_id in networks.items():
if network_id not in camera_list.keys():
camera_list[network_id] = {}
_LOGGER.warning("No cameras found for %s", network_name)
sync_module = BlinkSyncModule(
self, network_name, network_id, camera_list[network_id]
)
sync_module.start()
self.sync[network_name] = sync_module
def setup_post_verify(self):
"""Initialize blink system after verification."""
camera_list = self.get_cameras()
networks = self.get_ids()
for network_name, network_id in networks.items():
if network_id not in camera_list.keys():
camera_list[network_id] = {}
_LOGGER.warning("No cameras found for %s", network_name)
sync_module = BlinkSyncModule(
self, network_name, network_id, camera_list[network_id]
)
sync_module.start()
self.sync[network_name] = sync_module
self.cameras = self.merge_cameras()
def login(self):
"""Login method. DEPRECATED."""
"""Perform server login. DEPRECATED."""
_LOGGER.warning(
"Method is deprecated and will be removed in a future version. Please use the LoginHandler.login() method instead."
)
@@ -165,11 +175,19 @@ class Blink:
((self.region_id, self.region),) = response["region"].items()
self._host = "{}.{}".format(self.region_id, BLINK_URL)
self._token = response["authtoken"]["authtoken"]
self.networks = response["networks"]
self.client_id = response["client"]["id"]
self.account_id = response["account"]["id"]
self._auth_header = {"Host": self._host, "TOKEN_AUTH": self._token}
self.urls = BlinkURLHandler(self.region_id, legacy=self.legacy)
self.networks = self.get_networks()
self.client_id = response["client"]["id"]
self.account_id = response["account"]["id"]
def get_networks(self):
"""Get network information."""
response = api.request_networks(self)
try:
return response["summary"]
except KeyError:
return None
def get_ids(self):
"""Set the network ID and Account ID."""
blinkpy/helpers/constants.py
+1
View File
@@ -50,6 +50,7 @@ DEFAULT_URL = "{}.{}".format("rest-prod", BLINK_URL)
BASE_URL = "https://{}".format(DEFAULT_URL)
LOGIN_URLS = [
"{}/api/v4/login".format(BASE_URL),
"{}/api/v3/login".format(BASE_URL),
"{}/api/v2/login".format(BASE_URL),
]
blinkpy/login_handler.py
+25 -11
View File
@@ -13,26 +13,43 @@ _LOGGER = logging.getLogger(__name__)
class LoginHandler:
"""Class to handle login communication."""
def __init__(
self, username=None, password=None, cred_file=None,
):
def __init__(self, username=None, password=None, cred_file=None, persist_key=None):
"""
Initialize login handler.
:param username: Blink username
:param password: Blink password
:param cred_file: JSON formatted credential file.
:param persist_key: File location of persistant key.
"""
self.login_url = None
self.login_urls = const.LOGIN_URLS
self.cred_file = cred_file
self.persist_key = persist_key
self.data = {
"username": username,
"password": password,
"uid": util.gen_uid(const.SIZE_UID),
"notification_key": util.gen_uid(const.SIZE_NOTIFICATION_KEY),
"uid": None,
"notification_key": None,
}
self.check_keys()
def check_keys(self):
"""Check if uid exists, if not create."""
uid = util.gen_uid(const.SIZE_UID)
notification_key = util.gen_uid(const.SIZE_NOTIFICATION_KEY)
data = {"uid": uid, "notification_key": notification_key}
if self.persist_key is None:
return data
if not isfile(self.persist_key):
with open(self.persist_key, "w") as json_file:
json.dump(data, json_file)
else:
with open(self.persist_key, "r") as json_file:
data = json.load(json_file)
return data
def check_cred_file(self):
"""Check if credential file supplied and use if so."""
if isfile(self.cred_file):
@@ -113,18 +130,15 @@ class LoginHandler:
json_resp = response.json()
blink.available = json_resp["valid"]
except KeyError:
blink.available = False
return False
_LOGGER.error("Did not receive valid response from server.")
_LOGGER.error("Invalid key. Got %s", key)
return True
def check_key_required(self, blink):
"""Check if 2FA key is required."""
# No idea if this is the right end point. Placeholder for now.
try:
if blink.login_response["account"]["email_verification_required"]:
blink.available = False
if blink.login_response["client"]["verification_required"]:
return True
except KeyError:
pass
blink.available = True
return False
blinkpy/sync_module.py
+16 -3
View File
@@ -38,6 +38,7 @@ class BlinkSyncModule:
self.motion = {}
self.last_record = {}
self.camera_list = camera_list
self.available = False
@property
def attributes(self):
@@ -102,8 +103,7 @@ class BlinkSyncModule:
"Could not extract some sync module info: %s", response, exc_info=True
)
self.network_info = api.request_network_status(self.blink, self.network_id)
self.get_network_info()
self.check_new_videos()
try:
for camera_config in self.camera_list:
@@ -120,6 +120,7 @@ class BlinkSyncModule:
)
return False
self.available = True
return True
def get_events(self, **kwargs):
@@ -141,9 +142,21 @@ class BlinkSyncModule:
_LOGGER.error("Could not extract camera info: %s", response, exc_info=True)
return []
def get_network_info(self):
"""Retrieve network status."""
is_errored = False
self.network_info = api.request_network_status(self.blink, self.network_id)
try:
is_errored = self.network_info["network"]["sync_module_error"]
except KeyError:
is_errored = True
if is_errored:
self.available = False
def refresh(self, force_cache=False):
"""Get all blink cameras and pulls their most recent status."""
self.network_info = api.request_network_status(self.blink, self.network_id)
self.get_network_info()
self.check_new_videos()
for camera_name in self.cameras.keys():
camera_id = self.cameras[camera_name].camera_id