diff --git a/glocaltokens/client.py b/glocaltokens/client.py index abde458e..ba245a7c 100644 --- a/glocaltokens/client.py +++ b/glocaltokens/client.py @@ -22,7 +22,7 @@ ) from .google.internal.home.foyer.v1_pb2 import GetHomeGraphRequest, GetHomeGraphResponse from .google.internal.home.foyer.v1_pb2_grpc import StructuresServiceStub -from .scanner import GoogleDevice, discover_devices +from .scanner import NetworkDevice, discover_devices from .types import DeviceDict from .utils import network as net_utils, token as token_utils from .utils.logs import censor @@ -39,22 +39,18 @@ def __init__( device_id: str, device_name: str, local_auth_token: str, - google_device: GoogleDevice | None = None, - ip_address: str | None = None, - port: int | None = None, + network_device: NetworkDevice | None = None, hardware: str | None = None, ): """ - Initializes a Device. Can set or google_device or ip and port + Initializes a Device. """ log_prefix = f"[Device - {device_name}(id={device_id})]" LOGGER.debug("%s Initializing new Device instance", log_prefix) self.device_id = device_id self.device_name = device_name self.local_auth_token = None - self.ip_address = ip_address - self.port = port - self.google_device = google_device + self.network_device = network_device self.hardware = hardware # Token and name validations @@ -68,30 +64,15 @@ def __init__( return # Setting IP and PORT - if google_device: + if network_device: LOGGER.debug( - "%s google_device is provided, using it's IP and PORT", log_prefix + "%s network_device is provided, using its IP and PORT", log_prefix ) - self.ip_address = google_device.ip_address - self.port = google_device.port + self.ip_address: str | None = network_device.ip_address + self.port: int | None = network_device.port else: - LOGGER.debug( - "%s google_device is not provided, " - "using manually provided IP and PORT", - log_prefix, - ) - # If both ip_address and port are not set, this is fine. - if (ip_address and not port) or (not ip_address and port): - LOGGER.error( - "%s google_device is not provided, " - "both IP(%s) and PORT(%s) must be manually provided", - log_prefix, - ip_address, - port, - ) - return - self.ip_address = ip_address - self.port = port + self.ip_address = None + self.port = None # IP and PORT validation if ( @@ -127,7 +108,7 @@ def as_dict(self) -> DeviceDict: return { "device_id": self.device_id, "device_name": self.device_name, - "google_device": { + "network_device": { "ip": self.ip_address, "port": self.port, }, @@ -382,7 +363,7 @@ def get_google_devices( LOGGER.debug("Failed to fetch homegraph") return devices - network_devices: list[GoogleDevice] = [] + network_devices: list[NetworkDevice] = [] if disable_discovery is False: LOGGER.debug("Getting network devices...") network_devices = discover_devices( @@ -392,7 +373,7 @@ def get_google_devices( logging_level=self.logging_level, ) - def find_device(name: str) -> GoogleDevice | None: + def find_device(name: str) -> NetworkDevice | None: for device in network_devices: if device.name == name: return device @@ -409,16 +390,16 @@ def find_device(name: str) -> GoogleDevice | None: LOGGER.debug("%s not in models_list", item.hardware.model) continue - google_device = None + network_device = None if network_devices: LOGGER.debug("Looking for '%s' in local network", item.device_name) - google_device = find_device(item.device_name) + network_device = find_device(item.device_name) device = Device( device_id=item.device_info.device_id, device_name=item.device_name, local_auth_token=item.local_auth_token, - google_device=google_device, + network_device=network_device, hardware=item.hardware.model, ) if device.local_auth_token: diff --git a/glocaltokens/const.py b/glocaltokens/const.py index 872e6e7f..def8edf2 100644 --- a/glocaltokens/const.py +++ b/glocaltokens/const.py @@ -3,32 +3,32 @@ from typing import Final -ACCESS_TOKEN_APP_NAME: Final[str] = "com.google.android.apps.chromecast.app" -ACCESS_TOKEN_CLIENT_SIGNATURE: Final[str] = "24bb24c05e47e0aefa68a58a766179d9b613a600" -ACCESS_TOKEN_DURATION: Final[int] = 60 * 60 -ACCESS_TOKEN_SERVICE: Final[str] = "oauth2:https://www.google.com/accounts/OAuthLogin" +ACCESS_TOKEN_APP_NAME: Final = "com.google.android.apps.chromecast.app" +ACCESS_TOKEN_CLIENT_SIGNATURE: Final = "24bb24c05e47e0aefa68a58a766179d9b613a600" +ACCESS_TOKEN_DURATION: Final = 60 * 60 +ACCESS_TOKEN_SERVICE: Final = "oauth2:https://www.google.com/accounts/OAuthLogin" -ANDROID_ID_LENGTH: Final[int] = 16 -MASTER_TOKEN_LENGTH: Final[int] = 216 -ACCESS_TOKEN_LENGTH: Final[int] = 315 -LOCAL_AUTH_TOKEN_LENGTH: Final[int] = 108 +ANDROID_ID_LENGTH: Final = 16 +MASTER_TOKEN_LENGTH: Final = 216 +ACCESS_TOKEN_LENGTH: Final = 315 +LOCAL_AUTH_TOKEN_LENGTH: Final = 108 -GOOGLE_HOME_FOYER_API: Final[str] = "googlehomefoyer-pa.googleapis.com:443" +GOOGLE_HOME_FOYER_API: Final = "googlehomefoyer-pa.googleapis.com:443" -HOMEGRAPH_DURATION: Final[int] = 24 * 60 * 60 +HOMEGRAPH_DURATION: Final = 24 * 60 * 60 -DISCOVERY_TIMEOUT: Final[int] = 2 +DISCOVERY_TIMEOUT: Final = 2 -GOOGLE_HOME_MODELS: Final[list[str]] = [ +GOOGLE_HOME_MODELS: Final = [ "Google Home", "Google Home Mini", "Google Nest Mini", "Lenovo Smart Clock", ] -JSON_KEY_DEVICE_NAME: Final[str] = "device_name" -JSON_KEY_GOOGLE_DEVICE: Final[str] = "google_device" -JSON_KEY_HARDWARE: Final[str] = "hardware" -JSON_KEY_IP: Final[str] = "ip" -JSON_KEY_LOCAL_AUTH_TOKEN: Final[str] = "local_auth_token" -JSON_KEY_PORT: Final[str] = "port" +JSON_KEY_DEVICE_NAME: Final = "device_name" +JSON_KEY_NETWORK_DEVICE: Final = "network_device" +JSON_KEY_HARDWARE: Final = "hardware" +JSON_KEY_IP: Final = "ip" +JSON_KEY_LOCAL_AUTH_TOKEN: Final = "local_auth_token" +JSON_KEY_PORT: Final = "port" diff --git a/glocaltokens/scanner.py b/glocaltokens/scanner.py index d6147e42..18d1e47f 100644 --- a/glocaltokens/scanner.py +++ b/glocaltokens/scanner.py @@ -3,7 +3,7 @@ import logging from threading import Event -from typing import Callable +from typing import Callable, NamedTuple from zeroconf import ServiceBrowser, ServiceInfo, ServiceListener, Zeroconf @@ -13,6 +13,16 @@ LOGGER = logging.getLogger(__name__) +class NetworkDevice(NamedTuple): + """Discovered Google device representation""" + + name: str + ip_address: str + port: int + model: str + unique_id: str + + class CastListener(ServiceListener): """ Zeroconf Cast Services collection. @@ -26,7 +36,7 @@ def __init__( remove_callback: Callable[[], None] | None = None, update_callback: Callable[[], None] | None = None, ): - self.devices: list[GoogleDevice] = [] + self.devices: dict[str, NetworkDevice] = {} self.add_callback = add_callback self.remove_callback = remove_callback self.update_callback = update_callback @@ -49,6 +59,8 @@ def update_service(self, zc: Zeroconf, type_: str, name: str) -> None: def remove_service(self, _zc: Zeroconf, type_: str, name: str) -> None: """Called when a cast has beeen lost (mDNS info expired or host down).""" LOGGER.debug("remove_service %s, %s", type_, name) + if name in self.devices: + del self.devices[name] def _add_update_service( self, @@ -58,11 +70,11 @@ def _add_update_service( callback: Callable[[], None] | None, ) -> None: """ Add or update a service. """ - service = None - tries = 0 if name.endswith("_sub._googlecast._tcp.local."): LOGGER.debug("_add_update_service ignoring %s, %s", type_, name) return + service = None + tries = 0 while service is None and tries < 4: try: service = zc.get_service_info(type_, name) @@ -77,19 +89,38 @@ def _add_update_service( return addresses = service.parsed_addresses() - host = addresses[0] if addresses else service.server + ip_address = addresses[0] if addresses else service.server model_name = self.get_service_value(service, "md") friendly_name = self.get_service_value(service, "fn") + unique_id = self.get_service_value(service, "cd") - if not model_name or not friendly_name or not service.port: + if not model_name or not friendly_name or not service.port or not unique_id: LOGGER.debug( - "Device %s doesn't have friendly name, model name or port, skipping...", - host, + "Discovered device %s has incomplete service info, skipping...", + ip_address, + ) + return + + if not net_utils.is_valid_ipv4_address( + ip_address + ) and not net_utils.is_valid_ipv6_address(ip_address): + LOGGER.error("Discovered device has invalid IP address: %s", ip_address) + return + + if not 0 <= service.port <= 65535: + LOGGER.error( + "Port of discovered device is out of the valid range: [0,65535]" ) return - self.devices.append(GoogleDevice(friendly_name, host, service.port, model_name)) + self.devices[name] = NetworkDevice( + name=friendly_name, + ip_address=ip_address, + port=service.port, + model=model_name, + unique_id=unique_id, + ) if callback: callback() @@ -104,55 +135,20 @@ def get_service_value(service: ServiceInfo, key: str) -> str | None: return value.decode("utf-8") -class GoogleDevice: - """Discovered Google device representation""" - - def __init__(self, name: str, ip_address: str, port: int, model: str): - LOGGER.debug("Initializing GoogleDevice...") - if not net_utils.is_valid_ipv4_address( - ip_address - ) and not net_utils.is_valid_ipv6_address(ip_address): - LOGGER.error("IP must be a valid IP address") - return - - self.name = name - self.ip_address = ip_address - self.port = port - self.model = model - LOGGER.debug( - "Set self name to %s, IP to %s, PORT to %s and model to %s", - name, - ip_address, - port, - model, - ) - - if not 0 <= self.port <= 65535: - LOGGER.error("Port is out of the valid range: [0,65535]") - return - - def __str__(self) -> str: - """Serializes the class into a str""" - return ( - f"{{name:{self.name},ip:{self.ip_address}," - f"port:{self.port},model:{self.model}}}" - ) - - def discover_devices( models_list: list[str] | None = None, max_devices: int | None = None, timeout: int = DISCOVERY_TIMEOUT, zeroconf_instance: Zeroconf | None = None, logging_level: int = logging.ERROR, -) -> list[GoogleDevice]: +) -> list[NetworkDevice]: """Discover devices""" LOGGER.setLevel(logging_level) LOGGER.debug("Discovering devices...") def callback() -> None: - """Called when zeroconf has discovered a new chromecast.""" + """Called when zeroconf has discovered a new device.""" if max_devices is not None and listener.count >= max_devices: discovery_complete.set() @@ -167,20 +163,25 @@ def callback() -> None: LOGGER.debug("Using attribute Zeroconf instance") zc = zeroconf_instance LOGGER.debug("Creating zeroconf service browser for _googlecast._tcp.local.") - ServiceBrowser(zc, "_googlecast._tcp.local.", listener) + service_browser = ServiceBrowser(zc, "_googlecast._tcp.local.", listener) # Wait for the timeout or the maximum number of devices LOGGER.debug("Waiting for discovery completion...") discovery_complete.wait(timeout) - devices: list[GoogleDevice] = [] - LOGGER.debug("Got %s devices. Iterating...", len(listener.devices)) - for device in listener.devices: - if not models_list or device.model in models_list: - LOGGER.debug("Appending new device: %s", device) - devices.append(device) - else: + # Stop discovery + service_browser.cancel() + service_browser.zc.close() + + devices: list[NetworkDevice] = [] + LOGGER.debug("Got %d devices. Iterating...", listener.count) + for device in listener.devices.values(): + if models_list and device.model not in models_list: LOGGER.debug( - 'Won\'t add device since model "%s" is not in models_list', device.model + 'Skip discovered device since model "%s" is not in models_list', + device.model, ) + continue + LOGGER.debug("Add discovered device: %s", device) + devices.append(device) return devices diff --git a/glocaltokens/types.py b/glocaltokens/types.py index 191bd87d..8e85aed7 100644 --- a/glocaltokens/types.py +++ b/glocaltokens/types.py @@ -4,8 +4,8 @@ from typing import TypedDict -class GoogleDeviceDict(TypedDict): - """Typed dict for google_device field of DeviceDict.""" +class NetworkDeviceDict(TypedDict): + """Typed dict for network_device field of DeviceDict.""" ip: str | None port: int | None @@ -17,5 +17,5 @@ class DeviceDict(TypedDict): device_id: str device_name: str hardware: str | None - google_device: GoogleDeviceDict + network_device: NetworkDeviceDict local_auth_token: str | None diff --git a/tests/test_client.py b/tests/test_client.py index b94b3cdc..51a2a178 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -19,12 +19,13 @@ ANDROID_ID_LENGTH, HOMEGRAPH_DURATION, JSON_KEY_DEVICE_NAME, - JSON_KEY_GOOGLE_DEVICE, JSON_KEY_HARDWARE, JSON_KEY_IP, JSON_KEY_LOCAL_AUTH_TOKEN, + JSON_KEY_NETWORK_DEVICE, JSON_KEY_PORT, ) +from glocaltokens.scanner import NetworkDevice from tests.assertions import DeviceAssertions, TypeAssertions from tests.factory.providers import HomegraphProvider, TokenProvider @@ -338,12 +339,14 @@ def test_get_google_devices_json( ip_address = faker.ipv4() port = faker.port_number() hardware = faker.word() + unique_id = faker.word() google_device = Device( device_id=device_id, device_name=device_name, local_auth_token=local_auth_token, - ip_address=ip_address, - port=port, + network_device=NetworkDevice( + device_name, ip_address, port, hardware, unique_id + ), hardware=hardware, ) m_get_google_devices.return_value = [google_device] @@ -355,9 +358,9 @@ def test_get_google_devices_json( self.assertEqual(received_device[JSON_KEY_DEVICE_NAME], device_name) self.assertEqual(received_device[JSON_KEY_HARDWARE], hardware) self.assertEqual(received_device[JSON_KEY_LOCAL_AUTH_TOKEN], local_auth_token) - self.assertEqual(received_device[JSON_KEY_GOOGLE_DEVICE][JSON_KEY_PORT], port) + self.assertEqual(received_device[JSON_KEY_NETWORK_DEVICE][JSON_KEY_PORT], port) self.assertEqual( - received_device[JSON_KEY_GOOGLE_DEVICE][JSON_KEY_IP], ip_address + received_device[JSON_KEY_NETWORK_DEVICE][JSON_KEY_IP], ip_address ) @@ -372,8 +375,13 @@ def test_initialization__valid(self) -> None: device = Device( device_id=faker.uuid4(), device_name=faker.word(), - ip_address=faker.ipv4(), - port=faker.port_number(), + network_device=NetworkDevice( + faker.word(), + faker.ipv4(), + faker.port_number(), + faker.word(), + faker.word(), + ), local_auth_token=local_auth_token, ) @@ -382,25 +390,6 @@ def test_initialization__valid(self) -> None: @patch("glocaltokens.client.LOGGER.error") def test_initialization__invalid(self, m_log: NonCallableMock) -> None: """Test initialization that is invalid""" - # With only ip - device = Device( - device_id=faker.uuid4(), - device_name=faker.word(), - local_auth_token=faker.local_auth_token(), - ip_address=faker.ipv4_private(), - ) - self.assertEqual(m_log.call_count, 1) - self.assertIsNone(device.local_auth_token) - - # With only port - device = Device( - device_id=faker.uuid4(), - device_name=faker.word(), - local_auth_token=faker.local_auth_token(), - port=faker.port_number(), - ) - self.assertEqual(m_log.call_count, 2) - self.assertIsNone(device.local_auth_token) # Invalid local_auth_token device = Device( @@ -408,5 +397,5 @@ def test_initialization__invalid(self, m_log: NonCallableMock) -> None: device_name=faker.word(), local_auth_token=faker.word(), ) - self.assertEqual(m_log.call_count, 3) + self.assertEqual(m_log.call_count, 1) self.assertIsNone(device.local_auth_token) diff --git a/tests/test_scanner.py b/tests/test_scanner.py index d53f4312..7ab4ec2b 100644 --- a/tests/test_scanner.py +++ b/tests/test_scanner.py @@ -7,16 +7,16 @@ from faker import Faker from faker.providers import internet as internet_provider, python as python_provider -from glocaltokens.scanner import GoogleDevice +from glocaltokens.scanner import NetworkDevice faker = Faker() # type: ignore faker.add_provider(internet_provider) faker.add_provider(python_provider) -class GoogleDeviceTests(TestCase): +class NetworkDeviceTests(TestCase): """ - GoogleDevice specific tests + NetworkDevice specific tests """ def test_initialization(self) -> None: @@ -25,46 +25,29 @@ def test_initialization(self) -> None: ip_address = faker.ipv4_private() port = faker.port_number() model = faker.word() + unique_id = faker.word() - device = GoogleDevice(name, ip_address, port, model) + device = NetworkDevice(name, ip_address, port, model, unique_id) self.assertEqual(name, device.name) self.assertEqual(ip_address, device.ip_address) self.assertEqual(port, device.port) self.assertEqual(model, device.model) + self.assertEqual(unique_id, device.unique_id) self.assertEqual( - f"{{name:{name},ip:{ip_address},port:{port},model:{model}}}", str(device) + f"NetworkDevice(name='{name}', ip_address='{ip_address}', " + f"port={port}, model='{model}', unique_id='{unique_id}')", + str(device), ) @patch("glocaltokens.scanner.LOGGER.error") def test_initialization__valid(self, mock: NonCallableMock) -> None: """Valid initialization tests""" - GoogleDevice( - faker.word(), faker.ipv4_private(), faker.port_number(), faker.word() - ) - self.assertEqual(mock.call_count, 0) - - @patch("glocaltokens.scanner.LOGGER.error") - def test_initialization__invalid(self, mock: NonCallableMock) -> None: - """Invalid initialization tests""" - # With invalid IP - GoogleDevice(faker.word(), faker.word(), faker.port_number(), faker.word()) - self.assertEqual(mock.call_count, 1) - - # With negative port - GoogleDevice( + NetworkDevice( faker.word(), faker.ipv4_private(), - faker.pyint(min_value=-9999, max_value=-1), + faker.port_number(), faker.word(), - ) - self.assertEqual(mock.call_count, 2) - - # With greater port - GoogleDevice( - faker.word(), - faker.ipv4_private(), - faker.pyint(min_value=65535, max_value=999999), faker.word(), ) - self.assertEqual(mock.call_count, 3) + self.assertEqual(mock.call_count, 0)