Merge pull request #198 from natekspencer/auth

Update auth flow to OAuth PKCE

README.md

@@ -6,7 +6,7 @@
 
 Python library for interacting with a Vivint security and smart home system.
 
-This was built to support the `Vivint` integration in [Home-Assistant](https://www.home-assistant.io/) but _should_ work outside of it too. Currently, it can be utilized via [HACS](https://hacs.xyz/) by adding the [hacs-vivint](https://github.com/natekspencer/hacs-vivint) custom repository.
+This was built to support the [`Vivint`](https://github.com/natekspencer/hacs-vivint) integration in [Home-Assistant](https://www.home-assistant.io/) but _should_ work outside of it too. Currently, it can be utilized via [HACS](https://hacs.xyz/) by adding the [hacs-vivint](https://github.com/natekspencer/hacs-vivint) custom repository.
 
 ## Credit
 

pyproject.toml

@@ -16,6 +16,7 @@ certifi = ">=2022.9.24,<2025.0.0"
 pubnub = ">=7,<9"
 grpcio = "^1.51.1"
 protobuf = "^4.23.1"
+pyjwt = "^2.9.0"
 
 [tool.poetry.group.dev.dependencies]
 tox = ">=3.28,<5.0"

vivintpy/account.py

@@ -32,8 +32,8 @@ class Account:
     def __init__(
         self,
         username: str,
-        password: str,
-        persist_session: bool = False,
+        password: str | None = None,
+        refresh_token: str | None = None,
         client_session: aiohttp.ClientSession | None = None,
     ):
         """Initialize an account."""
@@ -44,7 +44,7 @@ def __init__(
         self._api = VivintSkyApi(
             username=username,
             password=password,
-            persist_session=persist_session,
+            refresh_token=refresh_token,
             client_session=client_session,
         )
         self.systems: list[System] = []

vivintpy/const.py

@@ -4,6 +4,7 @@
 class AuthenticationResponse:
     """Authentication response constants."""
 
+    ERROR = "error"
     INVALID = "Invalid username and/or password"
     MESSAGE = "msg"
     MFA_REQUIRED = "Multi-factor authentication required"

vivintpy/devices/door_lock.py

@@ -4,10 +4,10 @@
 
 from ..const import ZWaveDeviceAttribute as Attribute
 from ..utils import send_deprecation_warning
-from . import BypassTamperDevice, VivintDevice
+from . import BypassTamperDevice
 
 
-class DoorLock(BypassTamperDevice, VivintDevice):
+class DoorLock(BypassTamperDevice):
     """Represents a vivint door lock device."""
 
     @property

vivintpy/entity.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import logging
 from collections.abc import Callable
 
+_LOGGER = logging.getLogger(__name__)
+
 UPDATE = "update"
 
 
@@ -31,6 +34,7 @@ def update_data(self, new_val: dict, override: bool = False) -> None:
 
     def handle_pubnub_message(self, message: dict) -> None:
         """Handle a pubnub message directed to this entity."""
+        _LOGGER.debug("Message received by %s: %s", self.name, message)
         self.update_data(message)
 
     def on(  # pylint: disable=invalid-name

vivintpy/utils.py

@@ -3,7 +3,11 @@
 from __future__ import annotations
 
 import asyncio
+import base64
+import hashlib
 import logging
+import os
+import re
 from typing import Any, Callable, Coroutine, Iterable, TypeVar
 from warnings import warn
 
@@ -44,3 +48,20 @@ def send_deprecation_warning(old_name: str, new_name: str) -> None:
         stacklevel=2,
     )
     _LOGGER.warning(message)
+
+
+def generate_code_challenge() -> tuple[str, str]:
+    """Generate PKCE code verifier and challenge for authentication."""
+    code_verifier = base64.urlsafe_b64encode(os.urandom(40)).decode("utf-8")
+    code_verifier = re.sub("[^a-zA-Z0-9]+", "", code_verifier)
+
+    code_challenge = hashlib.sha256(code_verifier.encode("utf-8")).digest()
+    code_challenge = base64.urlsafe_b64encode(code_challenge).decode("utf-8")
+    code_challenge = code_challenge.replace("=", "")
+
+    return (code_verifier, code_challenge)
+
+
+def generate_state() -> str:
+    """Generate state."""
+    return base64.urlsafe_b64encode(os.urandom(40)).decode("utf-8")