Use base64 instead of hex for user_pk

python-kasa · Andrei965 · Oct 18, 2025 · Oct 19, 2025 · Oct 19, 2025 · Oct 19, 2025
commit ade20d7aa1714f85fbeec864c817b0c139dd1f79
diff --git a/kasa/transports/tpaptransport.py b/kasa/transports/tpaptransport.py
@@ -4,7 +4,6 @@
 
 import asyncio
 import base64
-import binascii
 import contextlib
 import hashlib
 import hmac
@@ -341,13 +340,13 @@
        """Split intermediate and root certificates from a chain PEM."""
        inter_pem, root_pem = chain_pem.split("-----END CERTIFICATE-----", 1)
        inter_pem += "-----END CERTIFICATE-----"
        return inter_pem, root_pem

    def apply(self, username: str, password: str) -> TpapNOCData:
        """Apply for a new NOC and cache materials."""
        if self._key_pem and self._cert_pem and self._inter_pem and self._root_pem:
            return self.get()
        try:
            token, account_id = self._login(username, password)
            url = self._get_url(account_id, token, username)
            priv = ec.generate_private_key(ec.SECP256R1())
@@ -468,7 +467,7 @@
            raise KasaException(
                f"{self._transport._host} {step_name} bad status/body: "
                f"{status} {type(data)}"
            )
        resp = cast(dict[str, Any], data)
        self._authenticator._handle_response_error_code(
            resp, f"TPAP {step_name} failed"
@@ -488,7 +487,7 @@
             raise KasaException("NOC materials unavailable")
         self.noc_cert_pem = noc.nocCertificate
         self.noc_key_pem = noc.nocPrivateKey
-        self.user_icac_pem = noc.nocIntermediateCertificate or ""
+        self.user_icac_pem = noc.nocIntermediateCertificate
         self.device_root_pem = noc.nocRootCertificate
         self._ephemeral_priv: ec.EllipticCurvePrivateKey | None = None
         self._ephemeral_pub_bytes: bytes | None = None
@@ -499,21 +498,21 @@
         self._hkdf_hash = "SHA256"
 
     @staticmethod
-    def _hex(b: bytes) -> str:
-        return binascii.hexlify(b).decode()
+    def _base64(b: bytes) -> str:
+        return base64.b64encode(b).decode()
 
     @staticmethod
-    def _unhex(s: str) -> bytes:
-        return binascii.unhexlify(s.encode())
+    def _unbase64(s: str) -> bytes:
+        return base64.b64decode(s)
 
     def _gen_ephemeral(self) -> bytes:
-        if self._ephemeral_priv is None:
+        if self._ephemeral_pub_bytes is None:
             self._ephemeral_priv = ec.generate_private_key(ec.SECP256R1())
             self._ephemeral_pub_bytes = self._ephemeral_priv.public_key().public_bytes(
                 encoding=serialization.Encoding.X962,
                 format=serialization.PublicFormat.UncompressedPoint,
             )
-        return cast(bytes, self._ephemeral_pub_bytes)
+        return self._ephemeral_pub_bytes
 
     def _derive_shared_secret(self, dev_pub_uncompressed: bytes) -> bytes:
         if self._ephemeral_priv is None:
@@ -551,8 +550,8 @@
             dev_ica_pem = dev_proof_obj.get("dev_icac") or dev_proof_obj.get(
                 "devIcacCertificate"
             )
-            proof_hex = dev_proof_obj.get("proof")
-            if not dev_noc_pem or not proof_hex:
+            proof_base64 = dev_proof_obj.get("proof")
+            if not dev_noc_pem or not proof_base64:
                 raise KasaException("Device proof missing fields")
 
             dev_cert = x509.load_pem_x509_certificate(dev_noc_pem.encode())
@@ -571,7 +570,7 @@
                 + self._dev_pub_bytes
                 + self._ephemeral_pub_bytes
             )
-            signature = binascii.unhexlify(proof_hex)
+            signature = self._unbase64(proof_base64)
             dev_pub = cast(ec.EllipticCurvePublicKey, dev_cert.public_key())
             dev_pub.verify(signature, message, ec.ECDSA(hashes.SHA256()))
         except InvalidSignature as exc:
@@ -581,21 +580,21 @@
 
     async def start(self) -> TlaSession | None:
         """Run NOC KEX + proof exchange and return session."""
-        user_pk_hex = self._hex(self._gen_ephemeral())
+        user_pk_base64 = self._base64(self._gen_ephemeral())
         admin_md5 = self._md5_hex("admin")
         params = {
             "sub_method": "noc_kex",
             "username": admin_md5,
             "encryption": ["aes_128_ccm", "chacha20_poly1305", "aes_256_ccm"],
-            "user_pk": user_pk_hex,
+            "user_pk": user_pk_base64,
             "stok": None,
         }
         resp = await self._login(params, step_name="noc_kex")
         _LOGGER.debug("NOC KEX response: %r", resp)
-        dev_pk_hex = resp.get("dev_pk")
-        if not dev_pk_hex:
+        dev_pk_base64 = resp.get("dev_pk")
+        if not dev_pk_base64:
             raise KasaException(f"NOC KEX response missing dev_pk, got {resp!r}")
-        self._dev_pub_bytes = self._unhex(dev_pk_hex)
+        self._dev_pub_bytes = self._unbase64(dev_pk_base64)
         chosen = (resp.get("encryption") or "aes_128_ccm").lower().replace("-", "_")
         self._chosen_cipher = (
             cast(_CipherId, chosen)
@@ -626,7 +625,7 @@
             {
                 "user_noc": self.noc_cert_pem,
                 "user_icac": self.user_icac_pem,
-                "proof": self._hex(signature),
+                "proof": self._base64(signature),
             },
             separators=(",", ":"),
         ).encode("utf-8")
@@ -635,16 +634,16 @@
         )
         proof_params = {
             "sub_method": "noc_proof",
-            "user_proof_encrypt": self._hex(ct_body),
-            "tag": self._hex(tag),
+            "user_proof_encrypt": self._base64(ct_body),
+            "tag": self._base64(tag),
         }
         proof_res = await self._login(proof_params, step_name="noc_proof")
-        dev_proof_hex = proof_res.get("dev_proof_encrypt")
-        tag_hex = proof_res.get("tag")
-        if not dev_proof_hex:
+        dev_proof_base64 = proof_res.get("dev_proof_encrypt")
+        tag_base64 = proof_res.get("tag")
+        if not dev_proof_base64:
             raise KasaException("NOC proof response missing device proof")
-        dev_ct = self._unhex(dev_proof_hex)
-        dev_tag = self._unhex(tag_hex) if tag_hex else b""
+        dev_ct = self._unbase64(dev_proof_base64)
+        dev_tag = self._unbase64(tag_base64) if tag_base64 else b""
         dev_plain = _SessionCipher.sec_decrypt(
             self._chosen_cipher, key, base_nonce, dev_ct, dev_tag, seq=1
         )
@@ -862,11 +861,11 @@
            )
            .derive(ikm)
            .hex()
            .upper()
        )

    async def start(self) -> TlaSession | None:
        """Run SPAKE2+ register/share and return session."""
        params = {
            "sub_method": "pake_register",
            "username": self.username,

diff --git a/tests/transports/test_tpaptransport.py b/tests/transports/test_tpaptransport.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import base64
-import binascii
 import hashlib
 import logging
 import os
@@ -399,7 +398,7 @@ async def post(self, url, *, json=None, data=None, headers=None, ssl=None):
                 return 200, {
                     "error_code": 0,
                     "result": {
-                        "dev_pk": binascii.hexlify(b"\x04" + b"\x01" * 64).decode(),
+                        "dev_pk": base64.b64encode(b"\x04" + b"\x01" * 64).decode(),
                         "encryption": "aes_128_ccm",
                         "expired": 99,
                     },
@@ -408,8 +407,8 @@ async def post(self, url, *, json=None, data=None, headers=None, ssl=None):
                 return 200, {
                     "error_code": 0,
                     "result": {
-                        "dev_proof_encrypt": "00",
-                        "tag": "00" * 16,
+                        "dev_proof_encrypt": base64.b64encode(b"\x00").decode(),
+                        "tag": base64.b64encode(b"\x00" * 16).decode(),
                         "sessionId": "SID",
                         "start_seq": 5,
                         "expired": 123,
@@ -555,7 +554,11 @@ def bad_sec_dec(cls, *a, **k):  # noqa: ARG001
     ctx_verify._ephemeral_pub_bytes = b"\x04" + b"\x04" * 64
     with pytest.raises(KasaException, match="Invalid NOC device proof signature"):
         ctx_verify._verify_device_proof(
-            {"dev_noc": dev_cert_pem, "dev_icac": inter_pem, "proof": bad_sig.hex()}
+            {
+                "dev_noc": dev_cert_pem,
+                "dev_icac": inter_pem,
+                "proof": base64.b64encode(bad_sig).decode(),
+            }
         )
 
 
@@ -638,7 +641,7 @@ async def post(self, url, *, json=None, data=None, headers=None, ssl=None):
                 return 200, {
                     "error_code": 0,
                     "result": {
-                        "dev_pk": binascii.hexlify(b"\x04" + b"\x02" * 64).decode(),
+                        "dev_pk": base64.b64encode(b"\x04" + b"\x02" * 64).decode(),
                         "encryption": "unknown",
                         "expired": 7,
                     },
@@ -647,7 +650,7 @@ async def post(self, url, *, json=None, data=None, headers=None, ssl=None):
                 return 200, {
                     "error_code": 0,
                     "result": {
-                        "dev_proof_encrypt": "00",
+                        "dev_proof_encrypt": base64.b64encode(b"\x00").decode(),
                         "stok": "STK",
                         "startSeq": 3,
                         "sessionExpired": 321,
@@ -729,7 +732,7 @@ async def post(self, url, *, json=None, data=None, headers=None, ssl=None):
                 return 200, {
                     "error_code": 0,
                     "result": {
-                        "dev_pk": binascii.hexlify(b"\x04" + b"\x03" * 64).decode(),
+                        "dev_pk": base64.b64encode(b"\x04" + b"\x03" * 64).decode(),
                         "encryption": "aes_128_ccm",
                         "expired": 1,
                     },
@@ -738,7 +741,10 @@ async def post(self, url, *, json=None, data=None, headers=None, ssl=None):
                 return 200, {
                     "error_code": 0,
                     "result": {
-                        "dev_proof": "00",
+                        # This test covers the case where the response has a
+                        # non-encrypted `dev_proof` field; keep it base64 for
+                        # realism, but the transport expects `dev_proof_encrypt`.
+                        "dev_proof": base64.b64encode(b"\x00").decode(),
                         "sessionId": "SIDN",
                         "start_seq": 2,
                         "expired": 5,
@@ -816,7 +822,7 @@ async def post(self, url, *, json=None, data=None, headers=None, ssl=None):
                 return 200, {
                     "error_code": 0,
                     "result": {
-                        "dev_pk": binascii.hexlify(b"\x04" + b"\x05" * 64).decode(),
+                        "dev_pk": base64.b64encode(b"\x04" + b"\x05" * 64).decode(),
                         "encryption": "aes_128_ccm",
                         "expired": 42,
                     },
@@ -825,7 +831,7 @@ async def post(self, url, *, json=None, data=None, headers=None, ssl=None):
                 return 200, {
                     "error_code": 0,
                     "result": {
-                        "dev_proof_encrypt": "00",
+                        "dev_proof_encrypt": base64.b64encode(b"\x00").decode(),
                         "session_id": "SID_ALT",
                     },
                 }
@@ -982,7 +988,7 @@ def _ensure_noc(self):
 
     wrong_message = b"not-the-expected-message"
     bad_sig = priv.sign(wrong_message, ec.ECDSA(hashes.SHA256()))
-    dev_proof_obj = {"dev_noc": cert_pem, "proof": binascii.hexlify(bad_sig).decode()}
+    dev_proof_obj = {"dev_noc": cert_pem, "proof": base64.b64encode(bad_sig).decode()}
     with pytest.raises(KasaException, match="Invalid NOC device proof signature"):
         ctx._verify_device_proof(dev_proof_obj)