From 05d8a9c77598ea73abc64d4a6d1a3b6749c0c6ed Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 1 Nov 2025 18:14:22 +0000 Subject: [PATCH 1/2] Initial plan From ce3785dadc8da913790cbe7ed74a57f504e9a932 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 1 Nov 2025 18:24:55 +0000 Subject: [PATCH 2/2] feat(v2): add FmdClient/Device package, port core fmd_api behaviors, and add tests Co-authored-by: devinslick <1762071+devinslick@users.noreply.github.com> --- docs/MIGRATE_FROM_V1.md | 266 ++++++++++++++++ examples/async_example.py | 60 ++++ fmd_api/__init__.py | 43 +++ fmd_api/_version.py | 3 + fmd_api/client.py | 616 ++++++++++++++++++++++++++++++++++++++ fmd_api/device.py | 205 +++++++++++++ fmd_api/exceptions.py | 26 ++ fmd_api/helpers.py | 42 +++ fmd_api/types.py | 70 +++++ pyproject.toml | 8 +- tests/__init__.py | 1 + tests/test_client.py | 236 +++++++++++++++ tests/test_device.py | 242 +++++++++++++++ 13 files changed, 1814 insertions(+), 4 deletions(-) create mode 100644 docs/MIGRATE_FROM_V1.md create mode 100644 examples/async_example.py create mode 100644 fmd_api/__init__.py create mode 100644 fmd_api/_version.py create mode 100644 fmd_api/client.py create mode 100644 fmd_api/device.py create mode 100644 fmd_api/exceptions.py create mode 100644 fmd_api/helpers.py create mode 100644 fmd_api/types.py create mode 100644 tests/__init__.py create mode 100644 tests/test_client.py create mode 100644 tests/test_device.py diff --git a/docs/MIGRATE_FROM_V1.md b/docs/MIGRATE_FROM_V1.md new file mode 100644 index 0000000..e8ced1b --- /dev/null +++ b/docs/MIGRATE_FROM_V1.md @@ -0,0 +1,266 @@ +# Migration Guide: V1 to V2 + +This guide helps you migrate from the old `fmd_api.py` module to the new v2 package structure with `FmdClient` and `Device` classes. + +## Overview of Changes + +V2 introduces: +- Structured package (`fmd_api/`) instead of single module +- Separate `FmdClient` (low-level) and `Device` (high-level) classes +- Typed exceptions for better error handling +- Type-annotated dataclasses for locations and photos +- More Pythonic API design + +## Quick Migration Reference + +### Imports + +**V1:** +```python +from fmd_api import FmdApi, FmdCommands +``` + +**V2:** +```python +from fmd_api import FmdClient, Device, FmdCommands +``` + +### Creating a Client + +**V1:** +```python +api = await FmdApi.create('https://fmd.example.com', 'device-id', 'password') +``` + +**V2 (Low-level client):** +```python +client = await FmdClient.create('https://fmd.example.com', 'device-id', 'password') +``` + +**V2 (High-level device wrapper):** +```python +device = await Device.create('https://fmd.example.com', 'device-id', 'password') +``` + +## API Mapping + +### Location Operations + +| V1 Method | V2 FmdClient Method | V2 Device Method | +|-----------|---------------------|------------------| +| `api.get_all_locations(num_to_get=10)` | `client.get_locations(num_to_get=10)` | `device.get_history(count=10)` | +| `api.decrypt_data_blob(blob)` | `client.decrypt_data_blob(blob)` | *(automatic in Device)* | +| `api.request_location('gps')` | `client.request_location('gps')` | `device.refresh('gps')` | + +**V1 Example:** +```python +api = await FmdApi.create('https://fmd.example.com', 'device-id', 'password') +location_blobs = await api.get_all_locations(num_to_get=10) + +for blob in location_blobs: + decrypted_bytes = api.decrypt_data_blob(blob) + location = json.loads(decrypted_bytes) + print(f"Location: {location['lat']}, {location['lon']}") +``` + +**V2 FmdClient Example:** +```python +from fmd_api import FmdClient +import json + +client = await FmdClient.create('https://fmd.example.com', 'device-id', 'password') +location_blobs = await client.get_locations(num_to_get=10) + +for blob in location_blobs: + decrypted_bytes = client.decrypt_data_blob(blob) + location = json.loads(decrypted_bytes) + print(f"Location: {location['lat']}, {location['lon']}") +``` + +**V2 Device Example (Recommended):** +```python +from fmd_api import Device + +device = await Device.create('https://fmd.example.com', 'device-id', 'password') +locations = await device.get_history(count=10) + +for location in locations: + print(f"Location: {location.latitude}, {location.longitude}") + print(f" Battery: {location.battery}%") + print(f" Provider: {location.provider}") + if location.speed: + print(f" Speed: {location.speed} m/s") +``` + +### Command Operations + +| V1 Method | V2 FmdClient Method | V2 Device Method | +|-----------|---------------------|------------------| +| `api.send_command('ring')` | `client.send_command('ring')` | `device.play_sound()` | +| `api.send_command('lock')` | `client.send_command('lock')` | `device.lock()` | +| `api.send_command('delete')` | `client.send_command('delete')` | `device.wipe()` | +| `api.take_picture('back')` | `client.take_picture('back')` | `device.take_photo('back')` | +| `api.toggle_bluetooth(True)` | `client.toggle_bluetooth(True)` | *(use client)* | +| `api.toggle_do_not_disturb(True)` | `client.toggle_do_not_disturb(True)` | *(use client)* | +| `api.set_ringer_mode('vibrate')` | `client.set_ringer_mode('vibrate')` | *(use client)* | +| `api.get_device_stats()` | `client.get_device_stats()` | *(use client)* | + +**V1 Example:** +```python +await api.send_command('ring') +await api.take_picture('front') +await api.toggle_bluetooth(True) +``` + +**V2 Device Example:** +```python +await device.play_sound() +await device.take_photo('front') +await device.client.toggle_bluetooth(True) # Access client for advanced commands +``` + +### Picture Operations + +| V1 Method | V2 FmdClient Method | V2 Device Method | +|-----------|---------------------|------------------| +| `api.get_pictures(num_to_get=5)` | `client.get_pictures(num_to_get=5)` | `device.fetch_pictures(count=5)` | + +**V1 Example:** +```python +pictures = await api.get_pictures(num_to_get=5) +for pic in pictures: + decrypted = api.decrypt_data_blob(pic['data']) + photo_b64 = decrypted.decode('utf-8') + photo_bytes = base64.b64decode(photo_b64) + with open(f'photo_{pic["timestamp"]}.jpg', 'wb') as f: + f.write(photo_bytes) +``` + +**V2 Device Example:** +```python +pictures = await device.fetch_pictures(count=5) +for pic in pictures: + with open(f'photo_{pic.timestamp}.jpg', 'wb') as f: + await device.download_photo(pic, f) +``` + +### Export Operations + +| V1 Method | V2 FmdClient Method | V2 Device Method | +|-----------|---------------------|------------------| +| `api.export_data_zip('export.zip')` | `client.export_data_zip('export.zip')` | `device.client.export_data_zip('export.zip')` | + +**No changes needed - same API.** + +## Exception Handling + +**V1:** +```python +from fmd_api import FmdApiException + +try: + api = await FmdApi.create(url, fmd_id, password) +except FmdApiException as e: + print(f"Error: {e}") +``` + +**V2:** +```python +from fmd_api import ( + FmdApiException, + FmdAuthenticationError, + FmdDecryptionError, + FmdApiRequestError, + FmdInvalidDataError, +) + +try: + client = await FmdClient.create(url, fmd_id, password) +except FmdAuthenticationError as e: + print(f"Authentication failed: {e}") +except FmdApiException as e: + print(f"General error: {e}") +``` + +## Data Types + +### Location Data + +**V1:** +```python +location = json.loads(decrypted_bytes) +lat = location['lat'] +lon = location['lon'] +speed = location.get('speed') # Optional field +``` + +**V2:** +```python +from fmd_api import Location + +location = await device.get_location() +lat = location.latitude +lon = location.longitude +speed = location.speed # None if not available +``` + +### Photo Data + +**V1:** +```python +pic = pictures[0] # Dictionary +timestamp = pic.get('timestamp', 0) +camera = pic.get('camera', 'unknown') +data = pic.get('data', '') +``` + +**V2:** +```python +from fmd_api import PhotoResult + +pic = pictures[0] # PhotoResult object +timestamp = pic.timestamp +camera = pic.camera +data = pic.encrypted_data +``` + +## Choosing Between FmdClient and Device + +### Use `FmdClient` when: +- You need low-level control over API requests +- You want to handle decryption and parsing manually +- You're building custom tooling or integrations +- You need access to all API endpoints + +### Use `Device` when: +- You want simple, high-level device operations +- You prefer automatic decryption and parsing +- You're building end-user applications +- You want type-safe data structures + +### Combining Both + +The `Device` class wraps `FmdClient`, so you can use both: + +```python +device = await Device.create(url, fmd_id, password) + +# High-level operations +location = await device.get_location() +await device.play_sound() + +# Access client for advanced features +await device.client.toggle_bluetooth(True) +await device.client.set_ringer_mode('vibrate') +``` + +## Summary + +The v2 API maintains backward compatibility at the client level while providing: +- Better structure and organization +- Typed data classes for improved IDE support +- More granular exception types +- High-level `Device` wrapper for common operations +- Same underlying protocol and encryption + +For most applications, we recommend using the `Device` class for its convenience and type safety. diff --git a/examples/async_example.py b/examples/async_example.py new file mode 100644 index 0000000..6305abd --- /dev/null +++ b/examples/async_example.py @@ -0,0 +1,60 @@ +"""Minimal async example of using fmd_api v2.""" +import asyncio +import logging +from fmd_api import Device + +# Configure logging to see what's happening +logging.basicConfig(level=logging.INFO) + + +async def main(): + # Replace with your FMD server details + SERVER_URL = "https://fmd.example.com" + DEVICE_ID = "your-device-id" + PASSWORD = "your-password" + + # Create and authenticate device + print("Authenticating...") + device = await Device.create(SERVER_URL, DEVICE_ID, PASSWORD) + print("✓ Authenticated successfully") + + # Get current location + print("\nFetching current location...") + location = await device.get_location() + if location: + print(f"✓ Location: ({location.latitude}, {location.longitude})") + print(f" Battery: {location.battery}%") + print(f" Provider: {location.provider}") + print(f" Timestamp: {location.time}") + if location.speed: + print(f" Speed: {location.speed:.2f} m/s") + else: + print("✗ No location data available") + + # Get location history + print("\nFetching location history (last 5)...") + history = await device.get_history(count=5) + print(f"✓ Retrieved {len(history)} location(s)") + for i, loc in enumerate(history, 1): + print(f" {i}. {loc.time} - ({loc.latitude}, {loc.longitude})") + + # Request a new location update + print("\nRequesting GPS location update...") + await device.refresh(provider="gps") + print("✓ Location update requested (device will update when online)") + + # Send ring command + print("\nMaking device ring...") + await device.play_sound() + print("✓ Ring command sent") + + # For more advanced operations, access the client directly + print("\nEnabling Bluetooth...") + await device.client.toggle_bluetooth(True) + print("✓ Bluetooth enable command sent") + + print("\n✓ All operations completed successfully") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/fmd_api/__init__.py b/fmd_api/__init__.py new file mode 100644 index 0000000..b34d709 --- /dev/null +++ b/fmd_api/__init__.py @@ -0,0 +1,43 @@ +"""FMD API v2 - Python client for FMD (Find My Device) servers. + +This package provides a client implementation for the FMD server API, +supporting authentication, encrypted communication, and device control. + +Basic Usage: + from fmd_api import FmdClient, Device + + # Using FmdClient directly + client = await FmdClient.create('https://fmd.example.com', 'device-id', 'password') + locations = await client.get_locations(10) + + # Using Device wrapper + device = await Device.create('https://fmd.example.com', 'device-id', 'password') + location = await device.get_location() + await device.play_sound() +""" + +from ._version import __version__ +from .client import FmdClient, FmdCommands +from .device import Device +from .types import Location, PhotoResult +from .exceptions import ( + FmdApiException, + FmdAuthenticationError, + FmdDecryptionError, + FmdApiRequestError, + FmdInvalidDataError, +) + +__all__ = [ + "__version__", + "FmdClient", + "FmdCommands", + "Device", + "Location", + "PhotoResult", + "FmdApiException", + "FmdAuthenticationError", + "FmdDecryptionError", + "FmdApiRequestError", + "FmdInvalidDataError", +] diff --git a/fmd_api/_version.py b/fmd_api/_version.py new file mode 100644 index 0000000..85e7f8c --- /dev/null +++ b/fmd_api/_version.py @@ -0,0 +1,3 @@ +"""Version information for fmd_api package.""" + +__version__ = "2.0.0" diff --git a/fmd_api/client.py b/fmd_api/client.py new file mode 100644 index 0000000..66b0deb --- /dev/null +++ b/fmd_api/client.py @@ -0,0 +1,616 @@ +"""FMD API Client implementation. + +This module provides the FmdClient class that handles authentication, +encryption/decryption, and communication with the FMD server. +""" +import base64 +import json +import logging +import time +from typing import List, Optional, Any + +import aiohttp +from argon2.low_level import hash_secret_raw, Type +from cryptography.hazmat.primitives import serialization, hashes +from cryptography.hazmat.primitives.asymmetric import padding +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +from .exceptions import ( + FmdApiException, + FmdAuthenticationError, + FmdDecryptionError, + FmdApiRequestError, + FmdInvalidDataError, +) +from .helpers import decode_base64, encode_base64 +from .types import Location, PhotoResult + +# --- Constants --- +CONTEXT_STRING_LOGIN = "context:loginAuthentication" +CONTEXT_STRING_ASYM_KEY_WRAP = "context:asymmetricKeyWrap" +ARGON2_SALT_LENGTH = 16 +AES_GCM_IV_SIZE_BYTES = 12 +RSA_KEY_SIZE_BYTES = 384 # 3072 bits / 8 + +log = logging.getLogger(__name__) + + +class FmdCommands: + """Constants for available FMD device commands.""" + + # Location requests + LOCATE_ALL = "locate" + LOCATE_GPS = "locate gps" + LOCATE_CELL = "locate cell" + LOCATE_LAST = "locate last" + + # Device control + RING = "ring" + LOCK = "lock" + DELETE = "delete" + + # Camera + CAMERA_FRONT = "camera front" + CAMERA_BACK = "camera back" + + # Bluetooth + BLUETOOTH_ON = "bluetooth on" + BLUETOOTH_OFF = "bluetooth off" + + # Do Not Disturb + NODISTURB_ON = "nodisturb on" + NODISTURB_OFF = "nodisturb off" + + # Ringer Mode + RINGERMODE_NORMAL = "ringermode normal" + RINGERMODE_VIBRATE = "ringermode vibrate" + RINGERMODE_SILENT = "ringermode silent" + + # Information/Status + STATS = "stats" + GPS = "gps" + + +class FmdClient: + """Client for the FMD server API. + + This class handles authentication, key management, and encrypted + communication with an FMD server. + """ + + def __init__(self, base_url: str, session_duration: int = 3600): + """Initialize FMD client. + + Args: + base_url: Base URL of the FMD server + session_duration: Session duration in seconds (default: 3600) + """ + self.base_url = base_url.rstrip('/') + self.session_duration = session_duration + self.access_token: Optional[str] = None + self.private_key = None + self._fmd_id: Optional[str] = None + self._password: Optional[str] = None + + @classmethod + async def create(cls, base_url: str, fmd_id: str, password: str, + session_duration: int = 3600) -> "FmdClient": + """Create and authenticate an FmdClient instance. + + Args: + base_url: Base URL of the FMD server + fmd_id: Device ID + password: Device password + session_duration: Session duration in seconds + + Returns: + Authenticated FmdClient instance + + Raises: + FmdAuthenticationError: If authentication fails + """ + instance = cls(base_url, session_duration) + instance._fmd_id = fmd_id + instance._password = password + await instance.authenticate(fmd_id, password, session_duration) + return instance + + async def authenticate(self, fmd_id: str, password: str, session_duration: int): + """Perform full authentication and key retrieval workflow. + + Args: + fmd_id: Device ID + password: Device password + session_duration: Session duration in seconds + + Raises: + FmdAuthenticationError: If authentication fails + """ + try: + log.info("[1] Requesting salt...") + salt = await self._get_salt(fmd_id) + + log.info("[2] Hashing password with salt...") + password_hash = self._hash_password(password, salt) + + log.info("[3] Requesting access token...") + self._fmd_id = fmd_id + self.access_token = await self._get_access_token(fmd_id, password_hash, session_duration) + + log.info("[3a] Retrieving encrypted private key...") + privkey_blob = await self._get_private_key_blob() + + log.info("[3b] Decrypting private key...") + privkey_bytes = self._decrypt_private_key_blob(privkey_blob, password) + self.private_key = self._load_private_key_from_bytes(privkey_bytes) + except Exception as e: + log.error(f"Authentication failed: {e}") + raise FmdAuthenticationError(f"Authentication failed: {e}") from e + + def _hash_password(self, password: str, salt: str) -> str: + """Hash password using Argon2id. + + Args: + password: Plain text password + salt: Base64-encoded salt + + Returns: + Argon2id hash string + """ + salt_bytes = decode_base64(salt) + password_bytes = (CONTEXT_STRING_LOGIN + password).encode('utf-8') + hash_bytes = hash_secret_raw( + secret=password_bytes, + salt=salt_bytes, + time_cost=1, + memory_cost=131072, + parallelism=4, + hash_len=32, + type=Type.ID + ) + hash_b64 = encode_base64(hash_bytes) + return f"$argon2id$v=19$m=131072,t=1,p=4${salt}${hash_b64}" + + async def _get_salt(self, fmd_id: str) -> str: + """Get salt for password hashing.""" + return await self._make_api_request("PUT", "/api/v1/salt", {"IDT": fmd_id, "Data": ""}) + + async def _get_access_token(self, fmd_id: str, password_hash: str, session_duration: int) -> str: + """Get access token.""" + payload = { + "IDT": fmd_id, + "Data": password_hash, + "SessionDurationSeconds": session_duration + } + return await self._make_api_request("PUT", "/api/v1/requestAccess", payload) + + async def _get_private_key_blob(self) -> str: + """Get encrypted private key blob.""" + return await self._make_api_request("PUT", "/api/v1/key", {"IDT": self.access_token, "Data": "unused"}) + + def _decrypt_private_key_blob(self, key_b64: str, password: str) -> bytes: + """Decrypt private key blob using password-derived key. + + Args: + key_b64: Base64-encoded encrypted key blob + password: Device password + + Returns: + Decrypted private key bytes + + Raises: + FmdDecryptionError: If decryption fails + """ + try: + key_bytes = decode_base64(key_b64) + salt = key_bytes[:ARGON2_SALT_LENGTH] + iv = key_bytes[ARGON2_SALT_LENGTH:ARGON2_SALT_LENGTH + AES_GCM_IV_SIZE_BYTES] + ciphertext = key_bytes[ARGON2_SALT_LENGTH + AES_GCM_IV_SIZE_BYTES:] + + password_bytes = (CONTEXT_STRING_ASYM_KEY_WRAP + password).encode('utf-8') + aes_key = hash_secret_raw( + secret=password_bytes, + salt=salt, + time_cost=1, + memory_cost=131072, + parallelism=4, + hash_len=32, + type=Type.ID + ) + + aesgcm = AESGCM(aes_key) + return aesgcm.decrypt(iv, ciphertext, None) + except Exception as e: + raise FmdDecryptionError(f"Failed to decrypt private key: {e}") from e + + def _load_private_key_from_bytes(self, privkey_bytes: bytes): + """Load private key from bytes (PEM or DER format). + + Args: + privkey_bytes: Private key bytes + + Returns: + Private key object + """ + try: + return serialization.load_pem_private_key(privkey_bytes, password=None) + except ValueError: + return serialization.load_der_private_key(privkey_bytes, password=None) + + def decrypt_data_blob(self, data_b64: str) -> bytes: + """Decrypt a data blob using the instance's private key. + + Args: + data_b64: Base64-encoded encrypted blob + + Returns: + Decrypted data bytes + + Raises: + FmdDecryptionError: If decryption fails + FmdInvalidDataError: If blob is too small + """ + try: + blob = decode_base64(data_b64) + + # Check if blob is large enough + min_size = RSA_KEY_SIZE_BYTES + AES_GCM_IV_SIZE_BYTES + if len(blob) < min_size: + raise FmdInvalidDataError( + f"Blob too small for decryption: {len(blob)} bytes " + f"(expected at least {min_size} bytes)" + ) + + session_key_packet = blob[:RSA_KEY_SIZE_BYTES] + iv = blob[RSA_KEY_SIZE_BYTES:RSA_KEY_SIZE_BYTES + AES_GCM_IV_SIZE_BYTES] + ciphertext = blob[RSA_KEY_SIZE_BYTES + AES_GCM_IV_SIZE_BYTES:] + + # Decrypt session key with RSA + session_key = self.private_key.decrypt( + session_key_packet, + padding.OAEP( + mgf=padding.MGF1(algorithm=hashes.SHA256()), + algorithm=hashes.SHA256(), + label=None + ) + ) + + # Decrypt data with AES-GCM + aesgcm = AESGCM(session_key) + return aesgcm.decrypt(iv, ciphertext, None) + except (FmdInvalidDataError, FmdDecryptionError): + raise + except Exception as e: + raise FmdDecryptionError(f"Failed to decrypt data blob: {e}") from e + + async def _make_api_request(self, method: str, endpoint: str, payload: dict, + stream: bool = False, expect_json: bool = True, + retry_auth: bool = True) -> Any: + """Make an API request with automatic retry on 401. + + Args: + method: HTTP method + endpoint: API endpoint + payload: Request payload + stream: If True, return response object for streaming + expect_json: If True, parse response as JSON + retry_auth: If True, retry once with new auth on 401 + + Returns: + Response data or response object if streaming + + Raises: + FmdApiRequestError: If request fails + """ + url = self.base_url + endpoint + try: + async with aiohttp.ClientSession() as session: + async with session.request(method, url, json=payload) as resp: + # Handle 401 Unauthorized by re-authenticating + if resp.status == 401 and retry_auth and self._fmd_id and self._password: + log.info("Received 401 Unauthorized, re-authenticating...") + await self.authenticate(self._fmd_id, self._password, self.session_duration) + # Retry with new token + payload["IDT"] = self.access_token + return await self._make_api_request(method, endpoint, payload, stream, expect_json, retry_auth=False) + + resp.raise_for_status() + + log.debug(f"{endpoint} response - status: {resp.status}, content-type: {resp.content_type}") + + if not stream: + if expect_json: + # FMD server sometimes returns wrong content-type + # Use content_type=None to force JSON parsing + try: + json_data = await resp.json(content_type=None) + log.debug(f"{endpoint} JSON response: {json_data}") + return json_data["Data"] + except (KeyError, ValueError, json.JSONDecodeError) as e: + # Fall back to text + log.debug(f"{endpoint} JSON parsing failed ({e}), trying as text") + text_data = await resp.text() + log.debug(f"{endpoint} returned text length: {len(text_data)}") + if not text_data: + log.warning(f"{endpoint} returned EMPTY response body") + return text_data + else: + text_data = await resp.text() + log.debug(f"{endpoint} text response length: {len(text_data)}") + return text_data + else: + return resp + except aiohttp.ClientError as e: + log.error(f"API request failed for {endpoint}: {e}") + raise FmdApiRequestError(f"API request failed for {endpoint}: {e}") from e + except (KeyError, ValueError) as e: + log.error(f"Failed to parse server response for {endpoint}: {e}") + raise FmdApiRequestError(f"Failed to parse server response for {endpoint}: {e}") from e + + async def get_locations(self, num_to_get: int = -1, skip_empty: bool = True, + max_attempts: int = 10) -> List[str]: + """Fetch location blobs from the server. + + Args: + num_to_get: Number of locations to get (-1 for all) + skip_empty: If True, skip empty blobs and search backwards + max_attempts: Maximum indices to try when skip_empty is True + + Returns: + List of encrypted location blobs + """ + log.debug(f"Getting locations, num_to_get={num_to_get}, skip_empty={skip_empty}") + size_str = await self._make_api_request("PUT", "/api/v1/locationDataSize", + {"IDT": self.access_token, "Data": "unused"}) + size = int(size_str) + log.debug(f"Server reports {size} locations available") + + if size == 0: + log.info("No locations found to download.") + return [] + + locations = [] + if num_to_get == -1: # Download all + log.info(f"Found {size} locations to download.") + for i in range(size): + log.info(f" - Downloading location at index {i}...") + blob = await self._make_api_request("PUT", "/api/v1/location", + {"IDT": self.access_token, "Data": str(i)}) + locations.append(blob) + return locations + else: # Download N most recent + num_to_download = min(num_to_get, size) + log.info(f"Found {size} locations. Downloading the {num_to_download} most recent.") + start_index = size - 1 + + if skip_empty: + indices = range(start_index, max(0, start_index - max_attempts), -1) + log.info(f"Will search for {num_to_download} non-empty location(s) starting from index {start_index}") + else: + end_index = size - num_to_download + indices = range(start_index, end_index - 1, -1) + log.info(f"Will fetch indices: {list(indices)}") + + for i in indices: + log.info(f" - Downloading location at index {i}...") + blob = await self._make_api_request("PUT", "/api/v1/location", + {"IDT": self.access_token, "Data": str(i)}) + log.debug(f"Received blob type: {type(blob)}, length: {len(blob) if blob else 0}") + + if blob and blob.strip(): + log.debug(f"First 100 chars: {blob[:100]}") + locations.append(blob) + log.info(f"Found valid location at index {i}") + if len(locations) >= num_to_get and num_to_get != -1: + break + else: + log.warning(f"Empty blob received for location index {i}") + + if not locations and num_to_get != -1: + log.warning(f"No valid locations found after checking {min(max_attempts, size)} indices") + + return locations + + async def get_pictures(self, num_to_get: int = -1) -> List[dict]: + """Fetch picture metadata from the server. + + Args: + num_to_get: Number of pictures to get (-1 for all) + + Returns: + List of picture metadata dictionaries + """ + try: + async with aiohttp.ClientSession() as session: + async with session.put(f"{self.base_url}/api/v1/pictures", + json={"IDT": self.access_token, "Data": ""}) as resp: + resp.raise_for_status() + all_pictures = await resp.json() + except aiohttp.ClientError as e: + log.warning(f"Failed to get pictures: {e}") + return [] + + if num_to_get == -1: + log.info(f"Found {len(all_pictures)} pictures to download.") + return all_pictures + else: + num_to_download = min(num_to_get, len(all_pictures)) + log.info(f"Found {len(all_pictures)} pictures. Selecting the {num_to_download} most recent.") + return all_pictures[-num_to_download:][::-1] + + async def export_data_zip(self, output_file: str): + """Download the pre-packaged export data zip file. + + Args: + output_file: Path to save the zip file + + Raises: + FmdApiRequestError: If export fails + """ + try: + async with aiohttp.ClientSession() as session: + async with session.post(f"{self.base_url}/api/v1/exportData", + json={"IDT": self.access_token, "Data": "unused"}) as resp: + resp.raise_for_status() + with open(output_file, 'wb') as f: + while True: + chunk = await resp.content.read(8192) + if not chunk: + break + f.write(chunk) + log.info(f"Exported data saved to {output_file}") + except aiohttp.ClientError as e: + log.error(f"Failed to export data: {e}") + raise FmdApiRequestError(f"Failed to export data: {e}") from e + + async def send_command(self, command: str) -> bool: + """Send a command to the device. + + Args: + command: Command string (use FmdCommands constants) + + Returns: + True if command was sent successfully + + Raises: + FmdApiRequestError: If command sending fails + """ + log.info(f"Sending command to device: {command}") + + # Get current Unix time in milliseconds + unix_time_ms = int(time.time() * 1000) + + # Sign the command using RSA-PSS + # IMPORTANT: Sign "timestamp:command", not just the command + message_to_sign = f"{unix_time_ms}:{command}" + message_bytes = message_to_sign.encode('utf-8') + signature = self.private_key.sign( + message_bytes, + padding.PSS( + mgf=padding.MGF1(hashes.SHA256()), + salt_length=32 + ), + hashes.SHA256() + ) + signature_b64 = encode_base64(signature) + + try: + result = await self._make_api_request( + "POST", + "/api/v1/command", + { + "IDT": self.access_token, + "Data": command, + "UnixTime": unix_time_ms, + "CmdSig": signature_b64 + }, + expect_json=False + ) + log.info(f"Command sent successfully: {command}") + return True + except Exception as e: + log.error(f"Failed to send command '{command}': {e}") + raise FmdApiRequestError(f"Failed to send command '{command}': {e}") from e + + # Convenience methods + + async def request_location(self, provider: str = "all") -> bool: + """Request a new location update from the device. + + Args: + provider: Location provider ("all", "gps", "cell", "last") + + Returns: + True if request was sent successfully + """ + provider_map = { + "all": FmdCommands.LOCATE_ALL, + "gps": FmdCommands.LOCATE_GPS, + "cell": FmdCommands.LOCATE_CELL, + "network": FmdCommands.LOCATE_CELL, + "last": FmdCommands.LOCATE_LAST + } + command = provider_map.get(provider.lower(), FmdCommands.LOCATE_ALL) + log.info(f"Requesting location update with provider: {provider} (command: {command})") + return await self.send_command(command) + + async def take_picture(self, camera: str = "back") -> bool: + """Request the device to take a picture. + + Args: + camera: Which camera to use ("front" or "back") + + Returns: + True if command was sent successfully + + Raises: + ValueError: If camera is not "front" or "back" + """ + camera = camera.lower() + if camera not in ["front", "back"]: + raise ValueError(f"Invalid camera '{camera}'. Must be 'front' or 'back'") + + command = FmdCommands.CAMERA_FRONT if camera == "front" else FmdCommands.CAMERA_BACK + log.info(f"Requesting picture from {camera} camera") + return await self.send_command(command) + + async def toggle_bluetooth(self, enable: bool) -> bool: + """Enable or disable Bluetooth on the device. + + Args: + enable: True to enable, False to disable + + Returns: + True if command was sent successfully + """ + command = FmdCommands.BLUETOOTH_ON if enable else FmdCommands.BLUETOOTH_OFF + log.info(f"{'Enabling' if enable else 'Disabling'} Bluetooth") + return await self.send_command(command) + + async def toggle_do_not_disturb(self, enable: bool) -> bool: + """Enable or disable Do Not Disturb mode. + + Args: + enable: True to enable, False to disable + + Returns: + True if command was sent successfully + """ + command = FmdCommands.NODISTURB_ON if enable else FmdCommands.NODISTURB_OFF + log.info(f"{'Enabling' if enable else 'Disabling'} Do Not Disturb mode") + return await self.send_command(command) + + async def set_ringer_mode(self, mode: str) -> bool: + """Set the device ringer mode. + + Args: + mode: Ringer mode ("normal", "vibrate", "silent") + + Returns: + True if command was sent successfully + + Raises: + ValueError: If mode is invalid + """ + mode = mode.lower() + mode_map = { + "normal": FmdCommands.RINGERMODE_NORMAL, + "vibrate": FmdCommands.RINGERMODE_VIBRATE, + "silent": FmdCommands.RINGERMODE_SILENT + } + + if mode not in mode_map: + raise ValueError(f"Invalid ringer mode '{mode}'. Must be 'normal', 'vibrate', or 'silent'") + + command = mode_map[mode] + log.info(f"Setting ringer mode to: {mode}") + return await self.send_command(command) + + async def get_device_stats(self) -> bool: + """Request device network statistics. + + Returns: + True if command was sent successfully + """ + log.info("Requesting device network statistics") + return await self.send_command(FmdCommands.STATS) diff --git a/fmd_api/device.py b/fmd_api/device.py new file mode 100644 index 0000000..720a215 --- /dev/null +++ b/fmd_api/device.py @@ -0,0 +1,205 @@ +"""Device class for FMD API. + +This module provides the Device class which wraps FmdClient +and provides higher-level device-oriented operations. +""" +import json +import logging +from typing import List, Optional, BinaryIO + +from .client import FmdClient +from .types import Location, PhotoResult +from .helpers import decode_base64 + +log = logging.getLogger(__name__) + + +class Device: + """Represents an FMD device with high-level operations. + + This class wraps FmdClient to provide device-centric methods + that automatically handle decryption and data parsing. + """ + + def __init__(self, client: FmdClient): + """Initialize Device with an FmdClient. + + Args: + client: Authenticated FmdClient instance + """ + self.client = client + self._cached_location: Optional[Location] = None + + @classmethod + async def create(cls, base_url: str, fmd_id: str, password: str, + session_duration: int = 3600) -> "Device": + """Create a Device instance with authentication. + + Args: + base_url: Base URL of the FMD server + fmd_id: Device ID + password: Device password + session_duration: Session duration in seconds + + Returns: + Device instance with authenticated client + """ + client = await FmdClient.create(base_url, fmd_id, password, session_duration) + return cls(client) + + async def refresh(self, provider: str = "all") -> bool: + """Request a new location update from the device. + + Args: + provider: Location provider ("all", "gps", "cell", "last") + + Returns: + True if request was sent successfully + """ + log.info(f"Refreshing device location with provider: {provider}") + return await self.client.request_location(provider) + + async def get_location(self, use_cached: bool = False) -> Optional[Location]: + """Get the most recent location from the device. + + Args: + use_cached: If True and cache exists, return cached location + + Returns: + Location object or None if no location available + """ + if use_cached and self._cached_location: + log.debug("Returning cached location") + return self._cached_location + + log.info("Fetching latest location from server") + blobs = await self.client.get_locations(num_to_get=1) + + if not blobs: + log.warning("No location data available") + return None + + try: + decrypted = self.client.decrypt_data_blob(blobs[0]) + location_data = json.loads(decrypted) + location = Location.from_dict(location_data) + self._cached_location = location + log.info(f"Retrieved location: ({location.latitude}, {location.longitude})") + return location + except Exception as e: + log.error(f"Failed to parse location: {e}") + return None + + async def get_history(self, count: int = 10) -> List[Location]: + """Get location history. + + Args: + count: Number of historical locations to retrieve (-1 for all) + + Returns: + List of Location objects + """ + log.info(f"Fetching {count if count != -1 else 'all'} historical locations") + blobs = await self.client.get_locations(num_to_get=count) + + locations = [] + for i, blob in enumerate(blobs): + try: + decrypted = self.client.decrypt_data_blob(blob) + location_data = json.loads(decrypted) + location = Location.from_dict(location_data) + locations.append(location) + except Exception as e: + log.warning(f"Failed to parse location {i}: {e}") + + log.info(f"Retrieved {len(locations)} location(s) from history") + return locations + + async def play_sound(self) -> bool: + """Make the device ring. + + Returns: + True if command was sent successfully + """ + log.info("Sending ring command to device") + return await self.client.send_command("ring") + + async def take_photo(self, camera: str = "back") -> bool: + """Take a photo with the device camera. + + Args: + camera: Which camera to use ("front" or "back") + + Returns: + True if command was sent successfully + """ + log.info(f"Requesting photo from {camera} camera") + return await self.client.take_picture(camera) + + async def fetch_pictures(self, count: int = -1) -> List[PhotoResult]: + """Fetch picture metadata from the server. + + Args: + count: Number of pictures to fetch (-1 for all) + + Returns: + List of PhotoResult objects + """ + log.info(f"Fetching {count if count != -1 else 'all'} picture(s)") + pictures = await self.client.get_pictures(num_to_get=count) + + results = [] + for pic_data in pictures: + try: + result = PhotoResult.from_dict(pic_data) + results.append(result) + except Exception as e: + log.warning(f"Failed to parse picture metadata: {e}") + + log.info(f"Retrieved {len(results)} picture(s)") + return results + + async def download_photo(self, photo: PhotoResult, output: BinaryIO) -> bool: + """Download and decrypt a photo. + + Args: + photo: PhotoResult object with encrypted data + output: Binary file-like object to write decrypted photo + + Returns: + True if download and decryption successful + """ + try: + log.info(f"Decrypting photo from {photo.camera} camera") + # Decrypt the photo data blob + decrypted = self.client.decrypt_data_blob(photo.encrypted_data) + + # Photo data is base64 encoded after decryption + photo_bytes = decode_base64(decrypted.decode('utf-8')) + + output.write(photo_bytes) + log.info(f"Photo downloaded successfully ({len(photo_bytes)} bytes)") + return True + except Exception as e: + log.error(f"Failed to download photo: {e}") + return False + + async def lock(self) -> bool: + """Lock the device screen. + + Returns: + True if command was sent successfully + """ + log.info("Sending lock command to device") + return await self.client.send_command("lock") + + async def wipe(self) -> bool: + """Wipe device data (factory reset). + + WARNING: This is a destructive operation that will erase all data! + + Returns: + True if command was sent successfully + """ + log.warning("Sending WIPE command to device - this will erase all data!") + return await self.client.send_command("delete") diff --git a/fmd_api/exceptions.py b/fmd_api/exceptions.py new file mode 100644 index 0000000..0f1889b --- /dev/null +++ b/fmd_api/exceptions.py @@ -0,0 +1,26 @@ +"""Exception types for FMD API.""" + + +class FmdApiException(Exception): + """Base exception for FMD API errors.""" + pass + + +class FmdAuthenticationError(FmdApiException): + """Raised when authentication fails.""" + pass + + +class FmdDecryptionError(FmdApiException): + """Raised when data decryption fails.""" + pass + + +class FmdApiRequestError(FmdApiException): + """Raised when an API request fails.""" + pass + + +class FmdInvalidDataError(FmdApiException): + """Raised when received data is invalid or corrupted.""" + pass diff --git a/fmd_api/helpers.py b/fmd_api/helpers.py new file mode 100644 index 0000000..712d59d --- /dev/null +++ b/fmd_api/helpers.py @@ -0,0 +1,42 @@ +"""Helper utilities for FMD API.""" +import base64 + + +def pad_base64(s: str) -> str: + """Add padding to base64 string if needed. + + Args: + s: Base64 string that may be missing padding + + Returns: + Properly padded base64 string + """ + return s + '=' * (-len(s) % 4) + + +def decode_base64(s: str) -> bytes: + """Decode base64 string, adding padding if needed. + + Args: + s: Base64 string + + Returns: + Decoded bytes + """ + return base64.b64decode(pad_base64(s)) + + +def encode_base64(data: bytes, strip_padding: bool = True) -> str: + """Encode bytes to base64 string. + + Args: + data: Bytes to encode + strip_padding: If True, remove padding characters + + Returns: + Base64 encoded string + """ + encoded = base64.b64encode(data).decode('utf-8') + if strip_padding: + return encoded.rstrip('=') + return encoded diff --git a/fmd_api/types.py b/fmd_api/types.py new file mode 100644 index 0000000..2a2778f --- /dev/null +++ b/fmd_api/types.py @@ -0,0 +1,70 @@ +"""Data types for FMD API.""" +from dataclasses import dataclass +from typing import Optional + + +@dataclass +class Location: + """Represents a device location.""" + + # Always present fields + time: str # Human-readable timestamp + date: int # Unix timestamp in milliseconds + provider: str # "gps", "network", "fused", or "BeaconDB" + battery: int # Battery percentage (0-100) + latitude: float # Latitude in degrees + longitude: float # Longitude in degrees + + # Optional fields (GPS/movement dependent) + accuracy: Optional[float] = None # GPS accuracy in meters + altitude: Optional[float] = None # Altitude in meters + speed: Optional[float] = None # Speed in meters/second + heading: Optional[float] = None # Direction in degrees 0-360 + + @classmethod + def from_dict(cls, data: dict) -> "Location": + """Create Location from dictionary. + + Args: + data: Location data dictionary from decrypted blob + + Returns: + Location instance + """ + return cls( + time=data['time'], + date=data['date'], + provider=data['provider'], + battery=data['bat'], + latitude=data['lat'], + longitude=data['lon'], + accuracy=data.get('accuracy'), + altitude=data.get('altitude'), + speed=data.get('speed'), + heading=data.get('heading') + ) + + +@dataclass +class PhotoResult: + """Represents a device photo result.""" + + timestamp: int # Unix timestamp in milliseconds + camera: str # "front" or "back" + encrypted_data: str # Base64 encrypted photo data + + @classmethod + def from_dict(cls, data: dict) -> "PhotoResult": + """Create PhotoResult from dictionary. + + Args: + data: Photo data dictionary + + Returns: + PhotoResult instance + """ + return cls( + timestamp=data.get('timestamp', 0), + camera=data.get('camera', 'unknown'), + encrypted_data=data.get('data', '') + ) diff --git a/pyproject.toml b/pyproject.toml index 4c64160..0cf58fa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "fmd_api" -version = "0.1.0" +version = "2.0.0" authors = [ {name = "Devin Slick", email = "fmd_client_github@devinslick.com"}, ] @@ -28,7 +28,6 @@ classifiers = [ ] keywords = ["fmd", "find-my-device", "location", "tracking", "device-tracking", "api-client"] dependencies = [ - "requests", "argon2-cffi", "cryptography", "aiohttp", @@ -44,13 +43,14 @@ Documentation = "https://github.com/devinslick/fmd_api#readme" dev = [ "pytest>=7.0", "pytest-asyncio", + "aioresponses", "black", "flake8", "mypy", ] -[tool.setuptools] -py-modules = ["fmd_api"] +[tool.setuptools.packages.find] +include = ["fmd_api*"] [tool.black] line-length = 120 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..b8a39b2 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests package for fmd_api.""" diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..4e7e9c8 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,236 @@ +"""Tests for FmdClient class.""" +import pytest +from unittest.mock import Mock, patch +from aioresponses import aioresponses + +from fmd_api import FmdClient, FmdCommands +from fmd_api.exceptions import FmdAuthenticationError, FmdInvalidDataError +from fmd_api.helpers import encode_base64 + + +@pytest.fixture +def mock_private_key(): + """Mock RSA private key.""" + mock_key = Mock() + mock_key.decrypt.return_value = b'0' * 32 # 32-byte AES key + mock_key.sign.return_value = b'0' * 384 # Mock signature + return mock_key + + +@pytest.mark.asyncio +async def test_create_and_authenticate(): + """Test FmdClient creation and authentication.""" + with aioresponses() as m: + # Mock salt request + m.put('https://fmd.test/api/v1/salt', payload={'Data': 'dGVzdHNhbHQxMjM0NTY='}) + + # Mock access token request + m.put('https://fmd.test/api/v1/requestAccess', payload={'Data': 'test-token-123'}) + + # Mock private key blob request + m.put('https://fmd.test/api/v1/key', payload={'Data': encode_base64(b'0' * 500)}) + + with patch('fmd_api.client.FmdClient._decrypt_private_key_blob') as mock_decrypt, \ + patch('fmd_api.client.FmdClient._load_private_key_from_bytes') as mock_load: + mock_decrypt.return_value = b'fake_key_bytes' + mock_load.return_value = Mock() + + client = await FmdClient.create('https://fmd.test', 'test-device', 'test-pass') + + assert client.access_token == 'test-token-123' + assert client._fmd_id == 'test-device' + assert client.private_key is not None + + +@pytest.mark.asyncio +async def test_get_locations_all(): + """Test getting all locations.""" + client = FmdClient('https://fmd.test') + client.access_token = 'test-token' + + with aioresponses() as m: + # Mock location size request + m.put('https://fmd.test/api/v1/locationDataSize', payload={'Data': '3'}) + + # Mock location requests + m.put('https://fmd.test/api/v1/location', payload={'Data': 'blob1'}) + m.put('https://fmd.test/api/v1/location', payload={'Data': 'blob2'}) + m.put('https://fmd.test/api/v1/location', payload={'Data': 'blob3'}) + + locations = await client.get_locations(num_to_get=-1) + + assert len(locations) == 3 + assert locations[0] == 'blob1' + assert locations[1] == 'blob2' + assert locations[2] == 'blob3' + + +@pytest.mark.asyncio +async def test_get_locations_recent(): + """Test getting N most recent locations.""" + client = FmdClient('https://fmd.test') + client.access_token = 'test-token' + + with aioresponses() as m: + # Mock location size request + m.put('https://fmd.test/api/v1/locationDataSize', payload={'Data': '10'}) + + # Mock location requests for last 2 + m.put('https://fmd.test/api/v1/location', payload={'Data': 'blob9'}) + m.put('https://fmd.test/api/v1/location', payload={'Data': 'blob8'}) + + locations = await client.get_locations(num_to_get=2, skip_empty=False) + + assert len(locations) == 2 + + +@pytest.mark.asyncio +async def test_decrypt_data_blob(mock_private_key): + """Test decrypting a data blob.""" + client = FmdClient('https://fmd.test') + client.private_key = mock_private_key + + # Create a minimal valid blob (RSA key packet + IV + ciphertext) + rsa_packet = b'R' * 384 # 384 bytes RSA packet + iv = b'I' * 12 # 12 bytes IV + ciphertext = b'C' * 50 # Some ciphertext + + blob = rsa_packet + iv + ciphertext + blob_b64 = encode_base64(blob, strip_padding=False) + + with patch('fmd_api.client.AESGCM') as mock_aesgcm: + mock_aesgcm.return_value.decrypt.return_value = b'{"test": "data"}' + + result = client.decrypt_data_blob(blob_b64) + + assert result == b'{"test": "data"}' + mock_private_key.decrypt.assert_called_once() + + +@pytest.mark.asyncio +async def test_decrypt_data_blob_too_small(mock_private_key): + """Test decrypting a blob that is too small.""" + client = FmdClient('https://fmd.test') + client.private_key = mock_private_key + + # Create a blob that's too small + small_blob = encode_base64(b'x' * 100, strip_padding=False) + + with pytest.raises(FmdInvalidDataError): + client.decrypt_data_blob(small_blob) + + +@pytest.mark.asyncio +async def test_send_command(mock_private_key): + """Test sending a command.""" + client = FmdClient('https://fmd.test') + client.access_token = 'test-token' + client.private_key = mock_private_key + + with aioresponses() as m: + m.post('https://fmd.test/api/v1/command', status=200, body='OK') + + result = await client.send_command(FmdCommands.RING) + + assert result is True + + +@pytest.mark.asyncio +async def test_reauth_on_401(): + """Test automatic re-authentication on 401.""" + client = FmdClient('https://fmd.test') + client.access_token = 'old-token' + client._fmd_id = 'test-device' + client._password = 'test-pass' + + with aioresponses() as m: + # First request returns 401 + m.put('https://fmd.test/api/v1/locationDataSize', status=401) + + # Re-auth sequence + m.put('https://fmd.test/api/v1/salt', payload={'Data': 'dGVzdHNhbHQxMjM0NTY='}) + m.put('https://fmd.test/api/v1/requestAccess', payload={'Data': 'new-token-456'}) + m.put('https://fmd.test/api/v1/key', payload={'Data': encode_base64(b'0' * 500)}) + + # Retry with new token succeeds + m.put('https://fmd.test/api/v1/locationDataSize', payload={'Data': '5'}) + + with patch('fmd_api.client.FmdClient._decrypt_private_key_blob') as mock_decrypt, \ + patch('fmd_api.client.FmdClient._load_private_key_from_bytes') as mock_load: + mock_decrypt.return_value = b'fake_key_bytes' + mock_load.return_value = Mock() + + result = await client._make_api_request( + 'PUT', '/api/v1/locationDataSize', + {'IDT': client.access_token, 'Data': 'unused'} + ) + + assert result == '5' + assert client.access_token == 'new-token-456' + + +@pytest.mark.asyncio +async def test_export_data_zip(tmp_path): + """Test exporting data to zip file.""" + client = FmdClient('https://fmd.test') + client.access_token = 'test-token' + + output_file = tmp_path / 'export.zip' + + with aioresponses() as m: + # Mock zip file download + m.post('https://fmd.test/api/v1/exportData', status=200, body=b'ZIPDATA123') + + await client.export_data_zip(str(output_file)) + + assert output_file.exists() + assert output_file.read_bytes() == b'ZIPDATA123' + + +@pytest.mark.asyncio +async def test_convenience_methods(mock_private_key): + """Test convenience wrapper methods.""" + client = FmdClient('https://fmd.test') + client.access_token = 'test-token' + client.private_key = mock_private_key + + with aioresponses() as m: + m.post('https://fmd.test/api/v1/command', status=200, body='OK') + m.post('https://fmd.test/api/v1/command', status=200, body='OK') + m.post('https://fmd.test/api/v1/command', status=200, body='OK') + m.post('https://fmd.test/api/v1/command', status=200, body='OK') + m.post('https://fmd.test/api/v1/command', status=200, body='OK') + + # Test request_location + assert await client.request_location('gps') is True + + # Test take_picture + assert await client.take_picture('front') is True + + # Test toggle_bluetooth + assert await client.toggle_bluetooth(True) is True + + # Test toggle_do_not_disturb + assert await client.toggle_do_not_disturb(False) is True + + # Test set_ringer_mode + assert await client.set_ringer_mode('vibrate') is True + + +@pytest.mark.asyncio +async def test_get_pictures(): + """Test getting pictures.""" + client = FmdClient('https://fmd.test') + client.access_token = 'test-token' + + with aioresponses() as m: + m.put('https://fmd.test/api/v1/pictures', payload=[ + {'timestamp': 1000, 'camera': 'back', 'data': 'blob1'}, + {'timestamp': 2000, 'camera': 'front', 'data': 'blob2'}, + ]) + + pictures = await client.get_pictures(num_to_get=-1) + + assert len(pictures) == 2 + assert pictures[0]['timestamp'] == 1000 + assert pictures[1]['camera'] == 'front' diff --git a/tests/test_device.py b/tests/test_device.py new file mode 100644 index 0000000..f4d8de1 --- /dev/null +++ b/tests/test_device.py @@ -0,0 +1,242 @@ +"""Tests for Device class.""" +import json +import pytest +from unittest.mock import Mock, AsyncMock, patch +from io import BytesIO + +from fmd_api import Device, Location, PhotoResult +from fmd_api.client import FmdClient +from fmd_api.helpers import encode_base64 + + +@pytest.fixture +def mock_client(): + """Mock FmdClient.""" + client = Mock(spec=FmdClient) + client.get_locations = AsyncMock() + client.get_pictures = AsyncMock() + client.send_command = AsyncMock(return_value=True) + client.request_location = AsyncMock(return_value=True) + client.take_picture = AsyncMock(return_value=True) + client.decrypt_data_blob = Mock() + return client + + +@pytest.mark.asyncio +async def test_device_get_location(mock_client): + """Test getting current location.""" + device = Device(mock_client) + + # Mock location data + location_data = { + 'time': 'Mon Jan 1 12:00:00 UTC 2024', + 'date': 1704110400000, + 'provider': 'gps', + 'bat': 75, + 'lat': 37.7749, + 'lon': -122.4194, + 'accuracy': 10.5, + 'speed': 1.2 + } + + mock_client.get_locations.return_value = ['encrypted_blob'] + mock_client.decrypt_data_blob.return_value = json.dumps(location_data).encode() + + location = await device.get_location() + + assert location is not None + assert location.latitude == 37.7749 + assert location.longitude == -122.4194 + assert location.battery == 75 + assert location.provider == 'gps' + assert location.accuracy == 10.5 + assert location.speed == 1.2 + + mock_client.get_locations.assert_called_once_with(num_to_get=1) + + +@pytest.mark.asyncio +async def test_device_get_location_no_data(mock_client): + """Test getting location when no data available.""" + device = Device(mock_client) + + mock_client.get_locations.return_value = [] + + location = await device.get_location() + + assert location is None + + +@pytest.mark.asyncio +async def test_device_get_location_cached(mock_client): + """Test getting cached location.""" + device = Device(mock_client) + + # Set a cached location + cached_location = Location( + time='Mon Jan 1 12:00:00 UTC 2024', + date=1704110400000, + provider='gps', + battery=75, + latitude=37.7749, + longitude=-122.4194 + ) + device._cached_location = cached_location + + location = await device.get_location(use_cached=True) + + assert location == cached_location + mock_client.get_locations.assert_not_called() + + +@pytest.mark.asyncio +async def test_device_get_history(mock_client): + """Test getting location history.""" + device = Device(mock_client) + + location_data_1 = { + 'time': 'Mon Jan 1 12:00:00 UTC 2024', + 'date': 1704110400000, + 'provider': 'gps', + 'bat': 75, + 'lat': 37.7749, + 'lon': -122.4194 + } + + location_data_2 = { + 'time': 'Mon Jan 1 13:00:00 UTC 2024', + 'date': 1704114000000, + 'provider': 'network', + 'bat': 70, + 'lat': 37.7750, + 'lon': -122.4195 + } + + mock_client.get_locations.return_value = ['blob1', 'blob2'] + mock_client.decrypt_data_blob.side_effect = [ + json.dumps(location_data_1).encode(), + json.dumps(location_data_2).encode() + ] + + history = await device.get_history(count=2) + + assert len(history) == 2 + assert history[0].latitude == 37.7749 + assert history[1].latitude == 37.7750 + assert history[0].battery == 75 + assert history[1].battery == 70 + + mock_client.get_locations.assert_called_once_with(num_to_get=2) + + +@pytest.mark.asyncio +async def test_device_refresh(mock_client): + """Test refreshing device location.""" + device = Device(mock_client) + + result = await device.refresh(provider='gps') + + assert result is True + mock_client.request_location.assert_called_once_with('gps') + + +@pytest.mark.asyncio +async def test_device_play_sound(mock_client): + """Test making device ring.""" + device = Device(mock_client) + + result = await device.play_sound() + + assert result is True + mock_client.send_command.assert_called_once_with('ring') + + +@pytest.mark.asyncio +async def test_device_take_photo(mock_client): + """Test taking a photo.""" + device = Device(mock_client) + + result = await device.take_photo(camera='front') + + assert result is True + mock_client.take_picture.assert_called_once_with('front') + + +@pytest.mark.asyncio +async def test_device_fetch_pictures(mock_client): + """Test fetching pictures.""" + device = Device(mock_client) + + mock_client.get_pictures.return_value = [ + {'timestamp': 1000, 'camera': 'back', 'data': 'blob1'}, + {'timestamp': 2000, 'camera': 'front', 'data': 'blob2'} + ] + + pictures = await device.fetch_pictures(count=2) + + assert len(pictures) == 2 + assert pictures[0].timestamp == 1000 + assert pictures[0].camera == 'back' + assert pictures[1].timestamp == 2000 + assert pictures[1].camera == 'front' + + mock_client.get_pictures.assert_called_once_with(num_to_get=2) + + +@pytest.mark.asyncio +async def test_device_download_photo(mock_client): + """Test downloading and decrypting a photo.""" + device = Device(mock_client) + + photo = PhotoResult( + timestamp=1000, + camera='back', + encrypted_data='encrypted_blob' + ) + + # Mock decryption to return base64-encoded image data + photo_bytes = b'FAKE_IMAGE_DATA' + photo_b64 = encode_base64(photo_bytes, strip_padding=False) + mock_client.decrypt_data_blob.return_value = photo_b64.encode() + + output = BytesIO() + result = await device.download_photo(photo, output) + + assert result is True + assert output.getvalue() == photo_bytes + mock_client.decrypt_data_blob.assert_called_once_with('encrypted_blob') + + +@pytest.mark.asyncio +async def test_device_lock(mock_client): + """Test locking device.""" + device = Device(mock_client) + + result = await device.lock() + + assert result is True + mock_client.send_command.assert_called_once_with('lock') + + +@pytest.mark.asyncio +async def test_device_wipe(mock_client): + """Test wiping device.""" + device = Device(mock_client) + + result = await device.wipe() + + assert result is True + mock_client.send_command.assert_called_once_with('delete') + + +@pytest.mark.asyncio +async def test_device_create(): + """Test Device.create factory method.""" + with patch('fmd_api.device.FmdClient.create') as mock_create: + mock_client = Mock(spec=FmdClient) + mock_create.return_value = mock_client + + device = await Device.create('https://fmd.test', 'device-id', 'password') + + assert device.client == mock_client + mock_create.assert_called_once_with('https://fmd.test', 'device-id', 'password', 3600)