Implement P2P protocol for device control

.gitignore

@@ -6,3 +6,4 @@ coverage.xml
 poetry.lock
 tags
 __pycache__
+.vscode/

AUTHORS.md

@@ -3,3 +3,4 @@
 * Aaron Bach (https://github.com/bachya)
 * Alexander Bessonov (https://github.com/nonsleepr)
 * Fuzzy (https://github.com/FuzzyMistborn)
+* Keshav (https://github.com/keshavdv)

eufy_security/api.py

@@ -8,6 +8,8 @@
 
 from .camera import Camera
 from .errors import InvalidCredentialsError, RequestError, raise_error
+from .station import Station
+from .types import DeviceType
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
@@ -28,6 +30,7 @@ def __init__(self, email: str, password: str, websession: ClientSession) -> None
         self._token_expiration: Optional[datetime] = None
 
         self.cameras: Dict[str, Camera] = {}
+        self.stations: Dict[str, Station] = {}
 
     async def async_authenticate(self) -> None:
         """Authenticate and get an access token."""
@@ -54,17 +57,35 @@ async def async_get_history(self) -> dict:
 
     async def async_update_device_info(self) -> None:
         """Get the latest device info."""
+        # Cameras
         devices_resp = await self.request("post", "app/get_devs_list")
 
         if not devices_resp.get("data"):
             return
 
         for device_info in devices_resp["data"]:
-            if device_info["device_sn"] in self.cameras:
-                camera = self.cameras[device_info["device_sn"]]
-                camera.camera_info = device_info
-                continue
-            self.cameras[device_info["device_sn"]] = Camera(self, device_info)
+            if DeviceType(device_info["device_type"]).is_camera():
+                if device_info["device_sn"] in self.cameras:
+                    camera = self.cameras[device_info["device_sn"]]
+                    camera.camera_info = device_info
+                    continue
+                self.cameras[device_info["device_sn"]] = Camera.from_info(
+                    self, device_info
+                )
+
+        # Stations
+        stations_resp = await self.request("post", "app/get_hub_list")
+
+        if not stations_resp.get("data"):
+            return
+
+        for device_info in stations_resp["data"]:
+            if DeviceType(device_info["device_type"]).is_station():
+                if device_info["station_sn"] in self.stations:
+                    station = self.stations[device_info["station_sn"]]
+                    station.station_info = device_info
+                    continue
+                self.stations[device_info["station_sn"]] = Station(self, device_info)
 
     async def request(
         self,

eufy_security/camera.py

@@ -2,7 +2,12 @@
 import logging
 from typing import TYPE_CHECKING
 
-from .params import ParamType
+from async_generator import asynccontextmanager
+
+from .errors import EufySecurityP2PError
+from .p2p.session import P2PSession
+from .p2p.types import CommandType
+from .types import DeviceType, ParamType
 
 if TYPE_CHECKING:
     from .api import API  # pylint: disable=cyclic-import
@@ -18,6 +23,22 @@ def __init__(self, api: "API", camera_info: dict) -> None:
         self._api = api
         self.camera_info: dict = camera_info
 
+    @staticmethod
+    def from_info(api: "API", camera_info: dict) -> "Camera":
+        camera_type = DeviceType(camera_info["device_type"])
+        if camera_type == DeviceType.FLOODLIGHT:
+            klass = FloodlightCamera
+        elif camera_type == DeviceType.DOORBELL:
+            klass = DoorbellCamera
+        else:
+            klass = Camera
+        return klass(api, camera_info)
+
+    @property
+    def device_type(self) -> str:
+        """Return the station's device type."""
+        return DeviceType(self.camera_info["device_type"])
+
     @property
     def hardware_version(self) -> str:
         """Return the camera's hardware version."""
@@ -131,3 +152,43 @@ async def async_stop_stream(self) -> None:
     async def async_update(self) -> None:
         """Get the latest values for the camera's properties."""
         await self._api.async_update_device_info()
+
+    @asynccontextmanager
+    async def async_establish_session(self, session: P2PSession = None):
+        if session and session.valid_for(self.station_serial):
+            yield session
+            return
+
+        if self.station_serial in self._api.stations:
+            async with self._api.stations[self.station_serial].connect() as session:
+                yield session
+                return
+        else:
+            raise EufySecurityP2PError(f"Could not find station for {self.name}")
+
+
+class DoorbellCamera(Camera):
+    async def enable_osd(self, enable: bool, session: P2PSession = None) -> None:
+        async with self.async_establish_session(session) as session:
+            await session.async_send_command_with_int_string(
+                0, CommandType.CMD_SET_DEVS_OSD, 1 if enable else 0
+            )
+
+
+class FloodlightCamera(Camera):
+    async def enable_osd(self, enable: bool, session: P2PSession = None) -> None:
+        async with self.async_establish_session(session) as session:
+            # 0 - disables the timestamp
+            # 1 - enables timestamp but removes logo
+            # 2 - enables all OSD items
+            await session.async_send_command_with_int_string(
+                0, CommandType.CMD_SET_DEVS_OSD, 2 if enable else 1
+            )
+
+    async def enable_manual_light(
+        self, enable: bool, session: P2PSession = None
+    ) -> None:
+        async with self.async_establish_session(session) as session:
+            await session.async_send_command_with_int_string(
+                0, CommandType.CMD_SET_FLOODLIGHT_MANUAL_SWITCH, 1 if enable else 0
+            )

eufy_security/errors.py

@@ -20,6 +20,12 @@ class RequestError(EufySecurityError):
     pass
 
 
+class EufySecurityP2PError(Exception):
+    """Define a P2P related error."""
+
+    pass
+
+
 ERRORS: Dict[int, Type[EufySecurityError]] = {26006: InvalidCredentialsError}
 
 

eufy_security/p2p/connection_manager.py

@@ -0,0 +1,24 @@
+import asyncio
+
+
+class ConnectionManager(asyncio.DatagramProtocol):
+    def __init__(self):
+        self.connection_map = {}
+
+    def connect(self, target, protocol):
+        self.connection_map[target] = protocol
+        protocol.connection_made(self.transport, target)
+
+    def connection_made(self, transport):
+        self.transport = transport
+
+    def datagram_received(self, data, addr):
+        if addr in self.connection_map:
+            self.connection_map[addr].datagram_received(data, addr)
+
+    def connection_lost(self, exc):
+        for _, protocol in self.connection_map.items():
+            protocol.connection_lost(exc)
+
+    def close(self):
+        self.transport.close()

eufy_security/p2p/discovery.py

@@ -0,0 +1,107 @@
+import asyncio
+from typing import Tuple
+
+from .lib32100 import BaseP2PClientProtocol
+from .types import (
+    P2PClientProtocolRequestMessageType,
+    P2PClientProtocolResponseMessageType,
+)
+
+
+class DiscoveryP2PClientProtocol(BaseP2PClientProtocol):
+    def __init__(self, loop, p2p_did: str, key: str, on_lookup_complete):
+        self.loop = loop
+        self.p2p_did = p2p_did
+        self.key = key
+        self.on_lookup_complete = on_lookup_complete
+        self.addresses = []
+        self.response_count = 0
+
+    def connection_made(self, transport, addr):
+        # Build payload
+        p2p_did_components = self.p2p_did.split("-")
+        payload = bytearray(p2p_did_components[0].encode())
+        payload.extend(int(p2p_did_components[1]).to_bytes(5, byteorder="big"))
+        payload.extend(p2p_did_components[2].encode())
+        payload.extend([0x00, 0x00, 0x00, 0x00, 0x00])
+        ip, port = transport.get_extra_info("sockname")
+        payload.extend(port.to_bytes(2, byteorder="little"))
+        payload.extend([int(x) for x in ip.split(".")[::-1]])
+        payload.extend(
+            [0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x04, 0x00, 0x00]
+        )
+        payload.extend(self.key.encode())
+        payload.extend([0x00, 0x00, 0x00, 0x00])
+
+        transport.sendto(
+            self.create_message(
+                P2PClientProtocolRequestMessageType.LOOKUP_WITH_KEY, payload
+            ),
+            addr,
+        )
+        # Manually timeout if we don't get an answer
+        self.loop.create_task(self.timeout(1.5))
+
+    async def timeout(self, seconds: float):
+        await asyncio.sleep(seconds)
+        self.return_candidates()
+
+    def process_response(
+        self,
+        msg_type: P2PClientProtocolResponseMessageType,
+        payload: bytes,
+        addr: Tuple[str, int],
+    ):
+        msg = payload[2:]
+        if msg_type == P2PClientProtocolResponseMessageType.LOOKUP_ADDR:
+            port = payload[5] * 256 + payload[4]
+            ip = f"{payload[9]}.{payload[8]}.{payload[7]}.{payload[6]}"
+            self.addresses.append((ip, port))
+
+            # We expect at most two IP/port combos so we can bail early
+            # if we received both
+            self.response_count += 1
+            if self.response_count == 2:
+                self.return_candidates()
+
+    def return_candidates(self):
+        if not self.on_lookup_complete.done():
+            self.on_lookup_complete.set_result(self.addresses)
+
+
+class LocalDiscoveryP2PClientProtocol(BaseP2PClientProtocol):
+    def __init__(self, loop, target: str, on_lookup_complete):
+        self.loop = loop
+        self.target = target
+        self.addresses = []
+        self.on_lookup_complete = on_lookup_complete
+
+    def connection_made(self, transport):
+        # Build payload
+        payload = bytearray([0] * 2)
+        transport.sendto(
+            self.create_message(
+                P2PClientProtocolRequestMessageType.LOCAL_LOOKUP, payload
+            ),
+            addr=(self.target, 32108),
+        )
+        # Manually timeout if we don't get an answer
+        self.loop.create_task(self.timeout(1.5))
+
+    async def timeout(self, seconds: float):
+        await asyncio.sleep(seconds)
+        self.return_candidates()
+
+    def process_response(
+        self,
+        msg_type: P2PClientProtocolResponseMessageType,
+        payload: bytes,
+        addr: Tuple[str, int],
+    ):
+        if msg_type == P2PClientProtocolResponseMessageType.LOCAL_LOOKUP_RESP:
+            self.addresses.append(addr)
+            self.return_candidates()
+
+    def return_candidates(self):
+        if not self.on_lookup_complete.done():
+            self.on_lookup_complete.set_result(self.addresses)

eufy_security/p2p/lib32100.py

@@ -0,0 +1,34 @@
+import asyncio
+import math
+from typing import Tuple
+
+from .types import (
+    P2PClientProtocolRequestMessageType,
+    P2PClientProtocolResponseMessageType,
+)
+
+
+class BaseP2PClientProtocol(asyncio.DatagramProtocol):
+    def create_message(
+        self, msg_type: P2PClientProtocolRequestMessageType, payload=bytearray()
+    ):
+        msg = bytearray()
+        msg.extend(msg_type.value)
+        payload_size = len(payload)
+        msg.append(math.floor(payload_size / 256))
+        msg.append(payload_size % 256)
+        msg.extend(payload)
+        return msg
+
+    def datagram_received(self, data, addr):
+        msg_type = P2PClientProtocolResponseMessageType(bytes(data[0:2]))
+        payload = data[2:]
+        self.process_response(msg_type, payload, addr)
+
+    def process_response(
+        self,
+        msg_type: P2PClientProtocolResponseMessageType,
+        payload: bytes,
+        addr: Tuple[str, int],
+    ):
+        raise NotImplementedError()