Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Save successfully login sessions, re-use them next time + small code cleanup #175

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Save successfully login sessions, re-use them next time + small code cleanup #175

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 26 additions & 7 deletions pyezviz/__main__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import pandas as pd

from .camera import EzvizCamera
from .client import EzvizClient
from .client import EzvizClient, EzvizSessionManager
from .constants import BatteryCameraWorkMode, DefenseModeType
from .exceptions import EzvizAuthVerificationCode
from .light_bulb import EzvizLightBulb
Expand All @@ -30,6 +30,16 @@ def main() -> Any:
parser.add_argument(
"--debug", "-d", action="store_true", help="Print debug messages to stderr"
)
parser.add_argument(
"--force-login",
action="store_true",
help="Force new login ignoring stored session"
)
parser.add_argument(
"--config-dir",
required=False,
help="Custom directory for storing the configuration file",
)

subparsers = parser.add_subparsers(dest="action")

Expand Down Expand Up @@ -185,21 +195,30 @@ def main() -> Any:
)

args = parser.parse_args()
session_manager = EzvizSessionManager(config_dir=args.config_dir)

if args.force_login:
session_manager.save_session({})

# print("--------------args")
# print("--------------args: %s",args)
# print("--------------args")

client = EzvizClient(args.username, args.password, args.region)
client = EzvizClient(
account=args.username,
password=args.password,
api_url=args.region,
session_manager=session_manager
)
try:
client.login()

client.start()
except EzvizAuthVerificationCode:
mfa_code = input("MFA code required, please input MFA code.\n")
client.login(sms_code=mfa_code)

client._login(smscode=mfa_code)
except Exception as exp: # pylint: disable=broad-except
print(exp)
print(f"Error: {exp}")
session_manager.save_session({})
return 1

if args.debug:
# You must initialize logging, otherwise you'll not see debug output.
Expand Down
202 changes: 158 additions & 44 deletions pyezviz/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,15 @@

from __future__ import annotations

import os
import platform
import time
from pathlib import Path
from datetime import datetime
import hashlib
import json
import logging
from typing import Any
from typing import Any, Optional
import urllib.parse
from uuid import uuid4

Expand Down Expand Up @@ -73,6 +77,71 @@

_LOGGER = logging.getLogger(__name__)

class EzvizSessionManager:
"""Manages persistent sessions for the Ezviz client with cross-platform support."""

def __init__(self, config_dir: Optional[str] = None):
"""Initialize the session manager with platform-specific paths."""
if config_dir is None:
if platform.system() == "Windows":
# Windows: C:\Users\<username>\AppData\Roaming\pyezviz
base_dir = os.path.join(os.environ.get("APPDATA") or str(Path.home()), "pyezviz")
elif platform.system() == "Darwin":
# macOS: ~/Library/Application Support/pyezviz
base_dir = os.path.join(str(Path.home()), "Library", "Application Support", "pyezviz")
else:
# Linux/Unix: ~/.config/pyezviz
base_dir = os.path.join(str(Path.home()), ".config", "pyezviz")

self.config_dir = base_dir
else:
self.config_dir = config_dir

# Use Path for reliable cross-platform path handling
self.config_path = Path(self.config_dir)
self.session_file = self.config_path / "session.json"
self._ensure_config_dir()

def _ensure_config_dir(self) -> None:
"""Create config directory if it doesn't exist with proper permissions."""
try:
# Create all parent directories if they don't exist
self.config_path.mkdir(parents=True, exist_ok=True)

if platform.system() != "Windows":
# 0o700 for Unix-like systems (user read/write/execute only)
self.config_path.chmod(0o700)

except Exception as e:
raise RuntimeError(f"Failed to create config directory: {e}")

def load_session(self) -> dict[str, Any]:
"""Load session data from file with proper error handling."""
try:
if self.session_file.exists():
with self.session_file.open('r', encoding='utf-8') as f:
return json.load(f)
except (json.JSONDecodeError, IOError, OSError) as e:
print(f"Warning: Could not load session file: {e}")
return {}

def save_session(self, session_data: dict[str, Any]) -> None:
"""Save session data to file with proper permissions."""
try:
with self.session_file.open('w', encoding='utf-8') as f:
json.dump(session_data, f, indent=2)

if platform.system() != "Windows":
# 0o600 for Unix-like systems (user read/write only)
self.session_file.chmod(0o600)

except (IOError, OSError) as e:
print(f"Warning: Could not save session file: {e}")

def get_session_path(self) -> str:
"""Return the full path to the session file."""
return str(self.session_file)


class EzvizClient:
"""Initialize api client object."""
Expand All @@ -81,30 +150,64 @@ def __init__(
self,
account: str | None = None,
password: str | None = None,
url: str = "apiieu.ezvizlife.com",
api_url: str = "apiieu.ezvizlife.com",
timeout: int = DEFAULT_TIMEOUT,
token: dict | None = None,
session_manager: Optional[EzvizSessionManager] = None,
) -> None:
"""Initialize the client object."""
self.api_url = api_url
self.account = account
self.password = (
hashlib.md5(password.encode("utf-8")).hexdigest() if password else None
) # Ezviz API sends md5 of password
)
self._session = requests.session()
self._session.headers.update(REQUEST_HEADER)
self._session.headers["sessionId"] = token["session_id"] if token else None
self._token = token or {
self._timeout = timeout
self._cameras: dict[str, Any] = {}
self._light_bulbs: dict[str, Any] = {}

self._session_manager = session_manager or EzvizSessionManager()
self._token = {
"session_id": None,
"rf_session_id": None,
"username": None,
"api_url": url,
"email": self.account,
"api_url": self.api_url,
}
self._timeout = timeout
self._cameras: dict[str, Any] = {}
self._light_bulbs: dict[str, Any] = {}

def start(self) -> None:
"""Start the client."""
self._token = self._load_or_initialize_token()
self._session.headers["sessionId"] = self._token["session_id"]
self._token["service_urls"] = self.get_service_urls()

def _load_or_initialize_token(self) -> dict[str, Any]:
"""Load existing token or initialize a new one, validating the account."""
stored_session = self._session_manager.load_session()

if self._is_valid_stored_session(stored_session):
if (
stored_session.get("email") == self.account
and stored_session.get("api_url") == self.api_url
):
print("Using stored session.")
return stored_session
else:
print("Warning: Account does not match the stored session.")

print("Logging in...")
return self._login()

def _is_valid_stored_session(self, stored_session: dict[str, Any]) -> bool:
"""Check if stored session is valid and not expired."""
if not stored_session:
return False

required_fields = ["session_id", "rf_session_id", "username", "email", "api_url"]
return all(stored_session.get(field) for field in required_fields)

def _login(self, smscode: int | None = None) -> dict[Any, Any]:
"""Login to Ezviz API."""
"""Login to Ezviz API with session persistence."""

# Region code to url.
if len(self._token["api_url"].split(".")) == 1:
Expand Down Expand Up @@ -146,45 +249,49 @@ def _login(self, smscode: int | None = None) -> dict[Any, Any]:
+ "\nResponse was: "
+ str(req.text)
) from err

match json_result["meta"]["code"]:
case 200:
self._session.headers["sessionId"] = json_result["loginSession"]["sessionId"]

self._token = {
"session_id": str(json_result["loginSession"]["sessionId"]),
"rf_session_id": str(json_result["loginSession"]["rfSessionId"]),
"username": str(json_result["loginUser"]["username"]),
"email": str(json_result["loginUser"]["email"]),
"api_url": str(json_result["loginArea"]["apiDomain"]),
}

# Save the session after successful login
self._session_manager.save_session(self._token)
return self._token

if json_result["meta"]["code"] == 200:
self._session.headers["sessionId"] = json_result["loginSession"][
"sessionId"
]
self._token = {
"session_id": str(json_result["loginSession"]["sessionId"]),
"rf_session_id": str(json_result["loginSession"]["rfSessionId"]),
"username": str(json_result["loginUser"]["username"]),
"api_url": str(json_result["loginArea"]["apiDomain"]),
}

self._token["service_urls"] = self.get_service_urls()

return self._token
case 1100:
self._token["api_url"] = json_result["loginArea"]["apiDomain"]
_LOGGER.warning("Region incorrect!")
_LOGGER.warning("Your region url: %s", self._token["api_url"])
return self.login()

if json_result["meta"]["code"] == 1100:
self._token["api_url"] = json_result["loginArea"]["apiDomain"]
_LOGGER.warning("Region incorrect!")
_LOGGER.warning("Your region url: %s", self._token["api_url"])
return self.login()
case 1012:
raise PyEzvizError("The MFA code is invalid, please try again.")

if json_result["meta"]["code"] == 1012:
raise PyEzvizError("The MFA code is invalid, please try again.")
case 1013:
raise PyEzvizError("Incorrect Username.")

if json_result["meta"]["code"] == 1013:
raise PyEzvizError("Incorrect Username.")
case 1014:
raise PyEzvizError("Incorrect Password.")

if json_result["meta"]["code"] == 1014:
raise PyEzvizError("Incorrect Password.")
case 1015:
raise PyEzvizError("The user is locked.")

if json_result["meta"]["code"] == 1015:
raise PyEzvizError("The user is locked.")
case 1226:
raise PyEzvizError("User does not exist or wrong password")

if json_result["meta"]["code"] == 6002:
self.send_mfa_code()
raise EzvizAuthVerificationCode(
"MFA enabled on account. Please retry with code."
)
case 6002:
self.send_mfa_code()
raise EzvizAuthVerificationCode(
"MFA enabled on account. Please retry with code."
)

raise PyEzvizError(f"Login error: {json_result['meta']}")

Expand Down Expand Up @@ -238,7 +345,10 @@ def get_service_urls(self) -> Any:
raise InvalidURL("A Invalid URL or Proxy error occurred") from err

except requests.HTTPError as err:
raise HTTPError from err
raise HTTPError(f"Failed to get service URLs: {req.status_code} {req.text}") from err

except Exception as err:
raise PyEzvizError(f"Error getting Service URLs: {err}") from err

try:
json_output = req.json()
Expand Down Expand Up @@ -1512,6 +1622,7 @@ def ptz_control_coordinates(

def login(self, sms_code: int | None = None) -> dict[Any, Any]:
"""Get or refresh ezviz login token."""

if self._token["session_id"] and self._token["rf_session_id"]:
try:
req = self._session.put(
Expand Down Expand Up @@ -1552,14 +1663,17 @@ def login(self, sms_code: int | None = None) -> dict[Any, Any]:
if not self._token.get("service_urls"):
self._token["service_urls"] = self.get_service_urls()

self._session_manager.save_session(self._token)
return self._token

if json_result["meta"]["code"] == 403:
if self.account and self.password:
self._session_manager.save_session({})
self._token = {
"session_id": None,
"rf_session_id": None,
"username": None,
"email": self.account,
"api_url": self._token["api_url"],
}
return self.login()
Expand Down