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

Merge pull request #241 from fronzbot/refactor-login-logic

Browse Source 
Refactor login logic in preperation for 2FA
This commit is contained in:
Kevin Fronczak
2020-05-05 13:56:39 -04:00
committed by GitHub
parent 853c596167 f80a93f8e1
commit 9bc20cc82e
11 changed files with 243 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1 -1
View File
@@ -3,7 +3,7 @@
.tox/*
__pycache__/*
htmlcov/*
.coverage
.coverage.*
coverage.xml
*.pyc
*.egg*/*
blinkpy/api.py
+19 -2
View File
@@ -11,7 +11,9 @@ _LOGGER = logging.getLogger(__name__)
MIN_THROTTLE_TIME = 2
def request_login(blink, url, username, password, is_retry=False):
def request_login(
blink, url, username, password, notification_key, uid, is_retry=False
):
"""
Login request.
@@ -26,7 +28,14 @@ def request_login(blink, url, username, password, is_retry=False):
{
"email": username,
"password": password,
"client_specifier": "iPhone 9.2 | 2.2 | 222",
"notification_key": notification_key,
"unique_id": uid,
"app_version": "6.0.7 (520300) #afb0be72a",
"client_name": "Computer",
"client_type": "android",
"device_identifier": "Blinkpy",
"os_version": "5.1.1",
"reauth": "true",
}
)
return http_req(
@@ -40,6 +49,14 @@ def request_login(blink, url, username, password, is_retry=False):
)
def request_verify(blink, verify_key):
"""Send verification key to blink servers."""
url = "api/v4/account/{}/client/{}/pin/verify".format(
blink.account_id, blink.client_id
)
return http_post(blink, url)
def request_networks(blink):
"""Request all networks information."""
url = "{}/networks".format(blink.urls.base_url)
blinkpy/blinkpy.py
+19 -88
View File
@@ -15,9 +15,7 @@ blinkpy is in no way affiliated with Blink, nor Immedia Inc.
import os.path
import time
import getpass
import logging
import json
from shutil import copyfileobj
from requests.structures import CaseInsensitiveDict
@@ -26,25 +24,22 @@ from slugify import slugify
from blinkpy import api
from blinkpy.sync_module import BlinkSyncModule
from blinkpy.helpers import errors as ERROR
from blinkpy.helpers.util import (
create_session,
merge_dicts,
get_time,
BlinkURLHandler,
BlinkAuthenticationException,
Throttle,
)
from blinkpy.helpers.constants import (
BLINK_URL,
LOGIN_URL,
OLD_LOGIN_URL,
LOGIN_BACKUP_URL,
DEFAULT_MOTION_INTERVAL,
DEFAULT_REFRESH,
MIN_THROTTLE_TIME,
LOGIN_URLS,
)
from blinkpy.helpers.constants import __version__
from blinkpy.login_handler import LoginHandler
_LOGGER = logging.getLogger(__name__)
@@ -81,13 +76,14 @@ class Blink:
endpoints (only use if you are having
api issues).
"""
self._username = username
self._password = password
self._cred_file = cred_file
self.login_handler = LoginHandler(
username=username, password=password, cred_file=cred_file
)
self._token = None
self._auth_header = None
self._host = None
self.account_id = None
self.client_id = None
self.network_ids = []
self.urls = None
self.sync = CaseInsensitiveDict({})
@@ -99,11 +95,12 @@ class Blink:
self.networks = []
self.cameras = CaseInsensitiveDict({})
self.video_list = CaseInsensitiveDict({})
self._login_url = LOGIN_URL
self.login_url = LOGIN_URLS[0]
self.login_urls = []
self.motion_interval = motion_interval
self.version = __version__
self.legacy = legacy_subdomain
self.available = False
@property
def auth_header(self):
@@ -117,12 +114,7 @@ class Blink:
Method logs in and sets auth token, urls, and ids for future requests.
Essentially this is just a wrapper function for ease of use.
"""
if self._username is None or self._password is None:
if not self.login():
return
elif not self.get_auth_token():
return
self.get_auth_token()
camera_list = self.get_cameras()
networks = self.get_ids()
for network_name, network_id in networks.items():
@@ -137,88 +129,33 @@ class Blink:
self.cameras = self.merge_cameras()
def login(self):
"""Prompt user for username and password."""
if self._cred_file is not None and os.path.isfile(self._cred_file):
try:
with open(self._cred_file, "r") as json_file:
creds = json.load(json_file)
self._username = creds["username"]
self._password = creds["password"]
except ValueError:
_LOGGER.error(
"Improperly formated json file %s.", self._cred_file, exc_info=True
)
return False
except KeyError:
_LOGGER.error("JSON file information incomplete %s.", exc_info=True)
return False
else:
self._username = input("Username:")
self._password = getpass.getpass("Password:")
if self.get_auth_token():
_LOGGER.debug("Login successful!")
return True
_LOGGER.warning("Unable to login with %s.", self._username)
return False
"""Login method. DEPRECATED."""
_LOGGER.warning(
"Method is deprecated and will be removed in a future version. Please use the LoginHandler.login() method instead."
)
return self.login_handler.login(self)
def get_auth_token(self, is_retry=False):
"""Retrieve the authentication token from Blink."""
if not isinstance(self._username, str):
raise BlinkAuthenticationException(ERROR.USERNAME)
if not isinstance(self._password, str):
raise BlinkAuthenticationException(ERROR.PASSWORD)
self.login_urls = [LOGIN_URL, OLD_LOGIN_URL, LOGIN_BACKUP_URL]
response = self.login_request(is_retry=is_retry)
response = self.login_handler.login(self)
if not response:
return False
self.login_url = self.login_handler.login_url
((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)
return self._auth_header
def login_request(self, is_retry=False):
"""Make a login request."""
try:
login_url = self.login_urls.pop(0)
except IndexError:
_LOGGER.error("Could not login to blink servers.")
return False
_LOGGER.info("Attempting login with %s", login_url)
response = api.request_login(
self, login_url, self._username, self._password, is_retry=is_retry
)
try:
if response.status_code != 200:
response = self.login_request(is_retry=True)
response = response.json()
((self.region_id, self.region),) = response["region"].items()
except AttributeError:
_LOGGER.error("Login API endpoint failed with response %s", response)
return False
except KeyError:
_LOGGER.warning("Could not extract region info.")
self.region_id = "piri"
self.region = "UNKNOWN"
self._login_url = login_url
return response
def get_ids(self):
"""Set the network ID and Account ID."""
response = api.request_networks(self)
all_networks = []
network_dict = {}
for network, status in self.networks.items():
@@ -226,12 +163,6 @@ class Blink:
all_networks.append("{}".format(network))
network_dict[status["name"]] = network
# For the first onboarded network we find, grab the account id
for resp in response["networks"]:
if str(resp["id"]) in all_networks:
self.account_id = resp["account_id"]
break
self.network_ids = all_networks
return network_dict
blinkpy/helpers/constants.py
+6 -4
View File
@@ -48,9 +48,10 @@ URLS
BLINK_URL = "immedia-semi.com"
DEFAULT_URL = "{}.{}".format("rest-prod", BLINK_URL)
BASE_URL = "https://{}".format(DEFAULT_URL)
LOGIN_URL = "{}/api/v2/login".format(BASE_URL)
OLD_LOGIN_URL = "{}/login".format(BASE_URL)
LOGIN_BACKUP_URL = "https://{}.{}/login".format("rest-piri", BLINK_URL)
LOGIN_URLS = [
"{}/api/v4/login".format(BASE_URL),
"{}/api/v2/login".format(BASE_URL),
]
"""
Dictionaries
@@ -61,7 +62,8 @@ ONLINE = {"online": True, "offline": False}
OTHER
"""
TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S%z"
DEFAULT_MOTION_INTERVAL = 1
DEFAULT_REFRESH = 30
MIN_THROTTLE_TIME = 2
SIZE_NOTIFICATION_KEY = 152
SIZE_UID = 16
blinkpy/helpers/util.py
+7
View File
@@ -2,6 +2,7 @@
import logging
import time
import secrets
from calendar import timegm
from functools import partial, wraps
from requests import Request, Session, exceptions
@@ -13,6 +14,12 @@ import blinkpy.helpers.errors as ERROR
_LOGGER = logging.getLogger(__name__)
def gen_uid(size):
"""Create a random sring."""
full_token = secrets.token_hex(size)
return full_token[0:size]
def time_to_seconds(timestamp):
"""Convert TIMESTAMP_FORMAT time to seconds."""
try:
blinkpy/login_handler.py
+106
View File
@@ -0,0 +1,106 @@
"""Login handler for blink."""
import json
import logging
from os.path import isfile
from getpass import getpass
from blinkpy import api
from blinkpy.helpers import util
from blinkpy.helpers import constants as const
_LOGGER = logging.getLogger(__name__)
class LoginHandler:
"""Class to handle login communication."""
def __init__(
self, username=None, password=None, cred_file=None,
):
"""
Initialize login handler.
:param username: Blink username
:param password: Blink password
:param cred_file: JSON formatted credential file.
"""
self.login_url = None
self.login_urls = const.LOGIN_URLS
self.cred_file = cred_file
self.data = {
"username": username,
"password": password,
"uid": util.gen_uid(const.SIZE_UID),
"notification_key": util.gen_uid(const.SIZE_NOTIFICATION_KEY),
}
def check_cred_file(self):
"""Check if credential file supplied and use if so."""
if isfile(self.cred_file):
try:
with open(self.cred_file, "r") as json_file:
creds = json.load(json_file)
self.data["username"] = creds["username"]
self.data["password"] = creds["password"]
except ValueError:
_LOGGER.error(
"Improperly formatted json file %s.", self.cred_file, exc_info=True
)
return False
except KeyError:
_LOGGER.error("JSON file information incomplete %s.", exc_info=True)
return False
return True
return False
def check_login(self):
"""Check login information and prompt if not available."""
if self.data["username"] is None:
self.data["username"] = input("Username:")
if self.data["password"] is None:
self.data["password"] = getpass("Password:")
if self.data["username"] and self.data["password"]:
return True
return False
def validate_response(self, url, response):
"""Validate response from login endpoint."""
try:
if response.status_code != 200:
return False
except AttributeError:
_LOGGER.error(
"Response for %s did not return a status code. Deprecated endpoint?",
url,
)
return False
return True
def login(self, blink):
"""Attempt login to blink servers."""
if self.cred_file is not None:
self.check_cred_file()
if not self.check_login():
_LOGGER.error("Cannot login with username %s", self.data["username"])
return False
for url in self.login_urls:
_LOGGER.info("Attempting login with %s", url)
response = api.request_login(
blink,
url,
self.data["username"],
self.data["password"],
self.data["notification_key"],
self.data["uid"],
is_retry=False,
)
if self.validate_response(url, response):
self.login_url = url
return response.json()
_LOGGER.error("Failed to login to Blink servers. Last response: %s", response)
return False
tests/mock_responses.py
+3 -1
View File
@@ -7,6 +7,8 @@ LOGIN_RESPONSE = {
"region": {"mock": "Test"},
"networks": {"1234": {"name": "test", "onboarded": True}},
"authtoken": {"authtoken": "foobar123", "message": "auth"},
"client": {"id": "5678"},
"account": {"id": "1337"},
}
@@ -50,7 +52,7 @@ def mocked_session_send(*args, **kwargs):
response = {"test": "foo"}
status = 200
elif method == "POST":
if url in (const.LOGIN_URL, const.LOGIN_BACKUP_URL):
if url in const.LOGIN_URLS:
response = LOGIN_RESPONSE
status = 200
elif url == "http://wrong.url/" or url is None:
tests/test_api.py
-46
View File
@@ -1,46 +0,0 @@
"""Test various api functions."""
import unittest
from unittest import mock
from blinkpy import api
from blinkpy.blinkpy import Blink
from blinkpy.helpers.util import create_session
class TestBlinkAPI(unittest.TestCase):
"""Test the blink api module."""
def setUp(self):
"""Initialize the blink module."""
self.blink = Blink()
self.blink.session = create_session()
# pylint: disable=protected-access
self.blink._auth_header = {}
def tearDown(self):
"""Tear down blink module."""
self.blink = None
@mock.patch("blinkpy.blinkpy.Blink.get_auth_token")
def test_http_req_connect_error(self, mock_auth):
"""Test http_get error condition."""
mock_auth.return_value = {"foo": "bar"}
firstlog = (
"INFO:blinkpy.helpers.util:" "Cannot connect to server with url {}."
).format("http://notreal.fake")
nextlog = (
"INFO:blinkpy.helpers.util:"
"Auth token expired, attempting reauthorization."
)
lastlog = (
"ERROR:blinkpy.helpers.util:"
"Endpoint {} failed. Possible issue with "
"Blink servers."
).format("http://notreal.fake")
expected = [firstlog, nextlog, firstlog, lastlog]
with self.assertLogs() as getlog:
api.http_get(self.blink, "http://notreal.fake")
with self.assertLogs() as postlog:
api.http_post(self.blink, "http://notreal.fake")
self.assertEqual(getlog.output, expected)
self.assertEqual(postlog.output, expected)
tests/test_blink_functions.py
+7 -9
View File
@@ -50,7 +50,7 @@ class TestBlinkFunctions(unittest.TestCase):
"""Clean up after test."""
self.blink = None
@mock.patch("blinkpy.blinkpy.api.request_login")
@mock.patch("blinkpy.login_handler.api.request_login")
def test_backup_url(self, req, mock_sess):
"""Test backup login method."""
json_resp = {
@@ -60,16 +60,14 @@ class TestBlinkFunctions(unittest.TestCase):
bad_req = mresp.MockResponse({}, 404)
new_req = mresp.MockResponse(json_resp, 200)
req.side_effect = [bad_req, bad_req, new_req]
self.blink.login_urls = ["test1", "test2", "test3"]
self.blink.login_request()
# pylint: disable=protected-access
self.assertEqual(self.blink._login_url, "test3")
self.blink.login_handler.login_urls = ["test1", "test2", "test3"]
self.blink.login_handler.login(self.blink)
self.assertEqual(self.blink.login_handler.login_url, "test3")
req.side_effect = [bad_req, new_req, bad_req]
self.blink.login_urls = ["test1", "test2", "test3"]
self.blink.login_request()
# pylint: disable=protected-access
self.assertEqual(self.blink._login_url, "test2")
self.blink.login_handler.login_urls = ["test1", "test2", "test3"]
self.blink.login_handler.login(self.blink)
self.assertEqual(self.blink.login_handler.login_url, "test2")
def test_merge_cameras(self, mock_sess):
"""Test merge camera functionality."""
tests/test_blinkpy.py
+5 -94
View File
@@ -14,7 +14,6 @@ from blinkpy.sync_module import BlinkSyncModule
from blinkpy.helpers.util import (
http_req,
create_session,
BlinkAuthenticationException,
BlinkException,
BlinkURLHandler,
)
@@ -31,7 +30,6 @@ class TestBlinkSetup(unittest.TestCase):
def setUp(self):
"""Set up Blink module."""
self.blink_no_cred = Blink()
self.blink = Blink(username=USERNAME, password=PASSWORD)
self.blink.sync["test"] = BlinkSyncModule(self.blink, "test", "1234", [])
self.blink.urls = BlinkURLHandler("test")
@@ -40,84 +38,12 @@ class TestBlinkSetup(unittest.TestCase):
def tearDown(self):
"""Clean up after test."""
self.blink = None
self.blink_no_cred = None
def test_initialization(self, mock_sess):
"""Verify we can initialize blink."""
self.assertEqual(self.blink.version, __version__)
# pylint: disable=protected-access
self.assertEqual(self.blink._username, USERNAME)
# pylint: disable=protected-access
self.assertEqual(self.blink._password, PASSWORD)
def test_no_credentials(self, mock_sess):
"""Check that we throw an exception when no username/password."""
with self.assertRaises(BlinkAuthenticationException):
self.blink_no_cred.get_auth_token()
# pylint: disable=protected-access
self.blink_no_cred._username = USERNAME
with self.assertRaises(BlinkAuthenticationException):
self.blink_no_cred.get_auth_token()
def test_no_auth_header(self, mock_sess):
"""Check that we throw an exception when no auth header given."""
# pylint: disable=unused-variable
((region_id, region),) = mresp.LOGIN_RESPONSE["region"].items()
self.blink.urls = BlinkURLHandler(region_id)
with self.assertRaises(BlinkException):
self.blink.get_ids()
@mock.patch("blinkpy.blinkpy.getpass.getpass")
def test_manual_login(self, getpwd, mock_sess):
"""Check that we can manually use the login() function."""
getpwd.return_value = PASSWORD
with mock.patch("builtins.input", return_value=USERNAME):
self.assertTrue(self.blink_no_cred.login())
# pylint: disable=protected-access
self.assertEqual(self.blink_no_cred._username, USERNAME)
# pylint: disable=protected-access
self.assertEqual(self.blink_no_cred._password, PASSWORD)
@mock.patch("blinkpy.blinkpy.getpass.getpass")
@mock.patch("blinkpy.blinkpy.Blink.get_auth_token")
def test_no_cred_file(self, getpwd, getauth, mock_sess):
"""Check that normal login occurs when cred file doesn't exist."""
# pylint: disable=protected-access
self.blink._cred_file = "/tmp/fake.file"
getpwd.return_value = PASSWORD
getauth.return_value = True
with mock.patch("builtins.input", return_value=USERNAME):
self.assertTrue(self.blink.login())
def test_exit_on_missing_json(self, mock_sess):
"""Test that we fail on missing json data."""
# pylint: disable=protected-access
self.blink._cred_file = "/tmp/fake.file"
with mock.patch("os.path.isfile", return_value=True):
with mock.patch("builtins.open", mock.mock_open(read_data="{}")):
self.assertFalse(self.blink.login())
def test_exit_on_bad_json(self, mock_sess):
"""Test that we fail on bad json format."""
# pylint: disable=protected-access
self.blink._cred_file = "/tmp/fake.file"
with mock.patch("os.path.isfile", return_value=True):
with mock.patch("builtins.open", mock.mock_open(read_data="{]")):
self.assertFalse(self.blink.login())
@mock.patch("blinkpy.blinkpy.json.load")
def test_cred_file(self, mockjson, mock_sess):
"""Test that loading credential file works."""
# pylint: disable=protected-access
self.blink_no_cred._cred_file = "/tmp/fake.file"
mockjson.return_value = {"username": "foo", "password": "bar"}
with mock.patch("os.path.isfile", return_value=True):
with mock.patch("builtins.open", mock.mock_open(read_data="")):
self.assertTrue(self.blink_no_cred.login())
# pylint: disable=protected-access
self.assertEqual(self.blink_no_cred._username, "foo")
# pylint: disable=protected-access
self.assertEqual(self.blink_no_cred._password, "bar")
self.assertEqual(self.blink.login_handler.data["username"], USERNAME)
self.assertEqual(self.blink.login_handler.data["password"], PASSWORD)
def test_bad_request(self, mock_sess):
"""Check that we raise an Exception with a bad request."""
@@ -147,15 +73,8 @@ class TestBlinkSetup(unittest.TestCase):
api.request_homescreen(self.blink)
self.assertEqual(self.blink.auth_header, original_header)
@mock.patch("blinkpy.api.request_networks")
def test_multiple_networks(self, mock_net, mock_sess):
def test_multiple_networks(self, mock_sess):
"""Check that we handle multiple networks appropriately."""
mock_net.return_value = {
"networks": [
{"id": 1234, "account_id": 1111},
{"id": 5678, "account_id": 2222},
]
}
self.blink.networks = {
"0000": {"onboarded": False, "name": "foo"},
"5678": {"onboarded": True, "name": "bar"},
@@ -163,26 +82,18 @@ class TestBlinkSetup(unittest.TestCase):
}
self.blink.get_ids()
self.assertTrue("5678" in self.blink.network_ids)
self.assertEqual(self.blink.account_id, 2222)
@mock.patch("blinkpy.api.request_networks")
def test_multiple_onboarded_networks(self, mock_net, mock_sess):
def test_multiple_onboarded_networks(self, mock_sess):
"""Check that we handle multiple networks appropriately."""
mock_net.return_value = {
"networks": [
{"id": 0000, "account_id": 2222},
{"id": 5678, "account_id": 1111},
]
}
self.blink.networks = {
"0000": {"onboarded": False, "name": "foo"},
"5678": {"onboarded": True, "name": "bar"},
"1234": {"onboarded": True, "name": "test"},
}
self.blink.get_ids()
self.assertTrue("0000" not in self.blink.network_ids)
self.assertTrue("5678" in self.blink.network_ids)
self.assertTrue("1234" in self.blink.network_ids)
self.assertEqual(self.blink.account_id, 1111)
@mock.patch("blinkpy.blinkpy.time.time")
def test_throttle(self, mock_time, mock_sess):
tests/test_login_handler.py
+70
View File
@@ -0,0 +1,70 @@
"""Test login handler."""
import unittest
from unittest import mock
from blinkpy.login_handler import LoginHandler
import tests.mock_responses as mresp
USERNAME = "foobar"
PASSWORD = "deadbeef"
@mock.patch("blinkpy.helpers.util.Session.send", side_effect=mresp.mocked_session_send)
class TestLoginHandler(unittest.TestCase):
"""Test the LoginHandler class in blinkpy."""
def setUp(self):
"""Set up Login Handler."""
self.login_handler = LoginHandler()
def tearDown(self):
"""Clean up after test."""
self.login_handler = None
@mock.patch("blinkpy.login_handler.getpass")
def test_manual_login(self, getpwd, mock_sess):
"""Check that we can manually use the login() function."""
getpwd.return_value = PASSWORD
with mock.patch("builtins.input", return_value=USERNAME):
self.assertTrue(self.login_handler.check_login())
self.assertEqual(self.login_handler.data["username"], USERNAME)
self.assertEqual(self.login_handler.data["password"], PASSWORD)
def test_no_cred_file(self, mock_sess):
"""Check that we return false when cred file doesn't exist."""
self.login_handler.cred_file = "/tmp/fake.file"
self.assertFalse(self.login_handler.check_cred_file())
@mock.patch("blinkpy.login_handler.isfile")
def test_exit_on_missing_json(self, mockisfile, mock_sess):
"""Test that we fail on missing json data."""
self.login_handler.cred_file = "/tmp/fake.file"
mockisfile.return_value = True
with mock.patch("builtins.open", mock.mock_open(read_data="{}")):
self.assertFalse(self.login_handler.check_cred_file())
@mock.patch("blinkpy.login_handler.json.load")
@mock.patch("blinkpy.login_handler.isfile")
def test_cred_file(self, mockisfile, mockjson, mock_sess):
"""Test that loading credential file works."""
self.login_handler.cred_file = "/tmp/fake.file"
mockjson.return_value = {"username": "foo", "password": "bar"}
mockisfile.return_value = True
with mock.patch("builtins.open", mock.mock_open(read_data="")):
self.assertTrue(self.login_handler.check_cred_file())
self.assertEqual(self.login_handler.data["username"], "foo")
self.assertEqual(self.login_handler.data["password"], "bar")
def test_bad_response(self, mock_sess):
"""Check bad response from server."""
self.assertFalse(self.login_handler.validate_response(None, None))
def test_bad_response_code(self, mock_sess):
"""Check bad response code from server."""
fake_resp = mresp.MockResponse(None, 404)
self.assertFalse(self.login_handler.validate_response(None, fake_resp))
def test_good_response_code(self, mock_sess):
"""Check good response code from server."""
fake_resp = mresp.MockResponse(None, 200)
self.assertTrue(self.login_handler.validate_response(None, fake_resp))