Merge branch 'main' into allow_self_signed_cert

Author: MauriceHendrix

dev-requirements.txt

@@ -239,16 +239,6 @@ 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
-
 
 # The following packages are considered to be unsafe in a requirements file:
 # pip

setup.cfg

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

src/s2python/generated/gen_s2_pairing.py

@@ -51,7 +51,7 @@ class PairingRequest(BaseModel):
     publicKey: str
     s2ClientNodeId: str
     s2ClientNodeDescription: str
-    supportedProtocols: List[Protocols]
+    supportedProtocols: Protocols
 
 
 class PairingResponse(BaseModel):
@@ -68,7 +68,7 @@ class ConnectionRequest(BaseModel):
         extra='forbid',
     )
     s2ClientNodeId: str
-    supportedProtocols: List[Protocols]
+    supportedProtocols: Protocols
 
 
 class ConnectionDetails(BaseModel):

src/s2python/s2_connection.py

@@ -7,7 +7,6 @@
 from dataclasses import dataclass
 from typing import Optional, List, Type, Dict, Callable, Awaitable, Union
 
-import ssl
 import websockets
 from websockets.asyncio.client import ClientConnection as WSConnection, connect as ws_connect
 
@@ -330,6 +329,7 @@ async def _connect_ws(self) -> None:
                 ssl_context.verify_mode = ssl.CERT_NONE
 
             self.ws = await ws_connect(uri=self.url, ssl=ssl_context)
+
         except (EOFError, OSError) as e:
             logger.info("Could not connect due to: %s", str(e))
 

src/s2python/s2_pairing.py

@@ -1,91 +1,49 @@
+import asyncio
+import json
 import logging
-import uuid
-from typing import List
-import requests
+from dataclasses import dataclass
 
-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)
+from s2python.generated.gen_s2_pairing import Protocols, S2NodeDescription
 
 
 logger = logging.getLogger("s2python")
 
+@dataclass
+class S2PairingEndpoint:
+    request_pairing_endpoint: str
+    request_connection_endpoint: str
+    
+
 class S2Pairing:  # pylint: disable=too-many-instance-attributes
+    publicKey: str
+    privateKey: str
+    pairing_endpoints: S2PairingEndpoint
+    token: str
+    s2ClientNodeId: str
+    s2ClientNodeDescription: S2NodeDescription
+    selectedProtocol: Protocols
+    challenge: str
     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
-    _client_node_id: str
-    _supported_protocols: List[Protocols]
-    _rsa_key_pair: RSAJwk
     def __init__(  # pylint: disable=too-many-arguments
         self,
-        request_pairing_endpoint: str,
+        publicKey: str,
+        privateKey: str,
+        pairing_endpoints: S2PairingEndpoint,
         token: str,
-        s2_client_node_description: S2NodeDescription,
-        verify_certificate: bool | str = False,
-        client_node_id: str = str(uuid.uuid4()),
-        supported_protocols: Protocols = (Protocols.WebSocketSecure, )
+        s2ClientNodeId: str,
+        s2ClientNodeDescription: S2NodeDescription,
+        selectedProtocol: Protocols = Protocols.WebSocketSecure
     ) -> None:
-        self.paired = False
-        self.s2_server_node_id = None
-        self.server_node_description = None
-        self.selected_protocol = None
-        self.connection_uri = None
-        self.challenge = None
-
-        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
+      self.publicKey = publicKey
+      self.privateKey = privateKey
+      self.pairing_endpoints = pairing_endpoints
+      self.token = token
+      self.s2ClientNodeId = s2ClientNodeId
+      self.s2ClientNodeDescription = s2ClientNodeDescription
+      self.selectedProtocol = selectedProtocol
+      
+      self.challenge = None
+      self.paired = false
+
+    def pair() -> bool:
+        return False