Added S2Pairing implementation

Author: MauriceHendrix

dev-requirements.txt

@@ -239,6 +239,17 @@ wheel==0.44.0
     # via pip-tools
 zipp==3.20.1
     # via importlib-metadata
+jwskate==0.11.1
+binapy==0.8.0
+    # via jwskate
+cffi==1.17.1
+    # via jwskate
+cryptography==44.0.2
+    # via jwskate
+pycparser==2.22
+    # via jwskate
+requests==2.32.3
+
 
 # The following packages are considered to be unsafe in a requirements file:
 # pip

setup.cfg

@@ -43,6 +43,8 @@ install_requires =
     pytz
     click
     websockets~=13.1
+    jwskate~=0.11
+    requests~=2.32.3
 
 [options.packages.find]
 where = src

src/s2python/generated/gen_s2_pairing.py

@@ -0,0 +1,80 @@
+# generated by datamodel-codegen:
+#   filename:  s2-over-ip-pairing
+#   timestamp: 2025-02-28T14:52:45+00:00
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import List
+
+from pydantic import BaseModel, ConfigDict, Field
+from s2python.common import EnergyManagementRole as S2Role
+
+
+
+class Deployment(Enum):
+    WAN = 'WAN'
+    LAN = 'LAN'
+
+
+class Protocols(Enum):
+    WebSocketSecure = 'WebSocketSecure'
+
+    
+class PairingInfo(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    pairingUri: str
+    token: str
+    validUntil: str
+
+
+class S2NodeDescription(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    brand: str
+    logoUri: str
+    type: str
+    modelName: str
+    userDefinedName: str
+    role: S2Role
+    deployment: Deployment
+
+
+class PairingRequest(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    token: str
+    publicKey: str
+    s2ClientNodeId: str
+    s2ClientNodeDescription: str
+    supportedProtocols: List[Protocols]
+
+
+class PairingResponse(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    s2ServerNodeId: str
+    serverNodeDescription: str
+    requestConnectionUri: str
+
+
+class ConnectionRequest(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    s2ClientNodeId: str
+    supportedProtocols: List[Protocols]
+
+
+class ConnectionDetails(BaseModel):
+    model_config = ConfigDict(
+        extra='forbid',
+    )
+    selectedProtocol: Protocols
+    challenge: str
+    connectionUri: str

src/s2python/s2_pairing.py

@@ -0,0 +1,86 @@
+import logging
+import uuid
+from typing import Tuple
+import requests
+
+from jwskate import JweCompact
+from jwskate.jwk.rsa import RSAJwk
+from binapy.binapy import BinaPy
+
+from s2python.generated.gen_s2_pairing import (Protocols,
+                                               PairingRequest,
+                                               S2NodeDescription,
+                                               PairingResponse,
+                                               ConnectionRequest,
+                                               ConnectionDetails)
+
+
+logger = logging.getLogger("s2python")
+
+class S2Pairing:  # pylint: disable=too-many-instance-attributes
+    paired: bool
+    s2_server_node_id: str
+    server_node_description: str
+    selected_protocol: Protocols
+    connection_uri: str
+    challenge: BinaPy
+
+    _request_pairing_endpoint: str
+    _token: str
+    _s2_client_node_description: S2NodeDescription
+    _verify_certificate: bool | str  # pylint: disable=E1131
+    _client_node_id: str
+    _supported_protocols: Tuple[Protocols]
+    _rsa_key_pair: RSAJwk
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        request_pairing_endpoint: str,
+        token: str,
+        s2_client_node_description: S2NodeDescription,
+        verify_certificate: bool | str = False,  # pylint: disable=E1131
+        client_node_id: str = str(uuid.uuid4()),
+        supported_protocols: Tuple[Protocols] = (Protocols.WebSocketSecure, )
+    ) -> None:
+        self.paired = False
+
+        self._request_pairing_endpoint = request_pairing_endpoint
+        self._token = token
+        self._s2_client_node_description = s2_client_node_description
+        self._verify_certificate = verify_certificate
+        self._client_node_id = client_node_id
+        self._supported_protocols = supported_protocols
+        self._rsa_key_pair = RSAJwk(self._rsa_key_pair)
+
+    def pair(self) -> bool:
+        self.paired = False
+        pairing_request: PairingRequest = PairingRequest(token=self._token,
+                                                        publicKey=self._rsa_key_pair.public_jwk().to_pem(),
+                                                        s2ClientNodeId=self._client_node_id,
+                                                        s2ClientNodeDescription=self._s2_client_node_description,
+                                                        supportedProtocols=self._supported_protocols)
+
+        response = requests.post(self._request_pairing_endpoint,
+                                 json=pairing_request.model_dump_json(),
+                                 timeout=10,
+                                 verify = self._verify_certificate)
+        response.raise_for_status()
+        pairing_response: PairingResponse = PairingResponse.parse_raw(response.json())
+        self.s2_server_node_id = pairing_response.s2ServerNodeId
+        self.server_node_description = pairing_response.serverNodeDescription
+
+        connection_request: ConnectionRequest = ConnectionRequest(s2ClientNodeId=self._client_node_id,
+                                                                   supportedProtocols=self._supported_protocols)
+
+        response = requests.post(pairing_response.requestConnectionUri,
+                                 json=connection_request.model_dump_json(),
+                                 timeout=10,
+                                 verify = self._verify_certificate)
+        response.raise_for_status()
+        connection_details: ConnectionDetails = ConnectionDetails.parse_raw(response.json())
+
+        self.selected_protocol = connection_details.selectedProtocol
+        self.connection_uri = connection_details.connectionUri
+        self.challenge = JweCompact(connection_details.challenge).decrypt(self._rsa_key_pair)
+
+        self.paired = True
+        return self.paired