FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

New Wi-Fi handling for SMARTCAM devices by ZeliardM · Pull Request #1639 · python-kasa/python-kasa · GitHub

Repository navigation

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
c06c965
New wifi handling for smart devices
ZeliardM Jan 10, 2026
b6e9238
Scan, Join, and Test fixes for new Wi-Fi
ZeliardM Jan 10, 2026
8d92d0b
Fix new scan handling
ZeliardM Jan 10, 2026
7f3bd97
Add debug logging
ZeliardM Jan 10, 2026
3b06ede
Payload format changes
ZeliardM Jan 21, 2026
d8b2550
Force connectAp as single request
ZeliardM Jan 24, 2026
c9e4cf3
Update test coverage
ZeliardM Jan 25, 2026
47fb39f
Merge branch 'python-kasa:master' into feature/device-wifi-scan
ZeliardM Jan 26, 2026
e620066
Revert "Update test coverage"
ZeliardM Jan 26, 2026
43fcf86
Revert "Force connectAp as single request"
ZeliardM Jan 26, 2026
740bfbe
Revert "Payload format changes"
ZeliardM Jan 26, 2026
3a5cbec
Revert "Add debug logging"
ZeliardM Jan 26, 2026
b7b09b7
Revert "Fix new scan handling"
ZeliardM Jan 26, 2026
eb5cfad
Revert "Scan, Join, and Test fixes for new Wi-Fi"
ZeliardM Jan 26, 2026
64117af
Revert "New wifi handling for smart devices"
ZeliardM Jan 26, 2026
1d426c4
Revert smartdevice.py changes and implement into smartcamdevice.py
ZeliardM Jan 26, 2026
9a9af6a
Update test coverage
ZeliardM Jan 27, 2026
1d312cd
Remove channel from wifi_scan
ZeliardM Jan 27, 2026
1acec12
Fix wifi --help and make --keytype optional
ZeliardM Jan 27, 2026
c95a5c2
Fix default keytype and require for none smartcam devices
ZeliardM Jan 27, 2026
0da9c38
Fix fixture handling for tests
ZeliardM Jan 27, 2026
3d1c7db
Change keytype handling
ZeliardM Feb 21, 2026
84c359d
Fix merge conflicts
ZeliardM Feb 21, 2026
0314865
Merge branch 'python-kasa:master' into feature/device-wifi-scan
ZeliardM Feb 21, 2026
4ff99bf
Fix test coverage
ZeliardM Feb 21, 2026
File filter

Filter by extension

Filter by extension .py  (9) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
18 changes: 14 additions & 4 deletions kasa/cli/wifi.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

from kasa import (
Device,
KasaException,
)

from .common import (
Expand All @@ -15,8 +16,7 @@


@click.group()
@pass_dev
def wifi(dev) -> None:
def wifi() -> None:
"""Commands to control wifi settings."""


Expand All @@ -35,13 +35,23 @@ async def scan(dev):

@wifi.command()
@click.argument("ssid")
@click.option("--keytype", prompt=True)
@click.option(
"--keytype",
default="",
help="KeyType (Not needed for SmartCamDevice).",
)
@click.option("--password", prompt=True, hide_input=True)
@pass_dev
async def join(dev: Device, ssid: str, password: str, keytype: str):
"""Join the given wifi network."""
echo(f"Asking the device to connect to {ssid}..")
res = await dev.wifi_join(ssid, password, keytype=keytype)
try:
res = await dev.wifi_join(ssid, password, keytype=keytype)
except KasaException as e:
if type(e) is KasaException:
echo(str(e))
return
raise
echo(
f"Response: {res} - if the device is not able to join the network, "
f"it will revert back to its previous state."
Expand Down
11 changes: 7 additions & 4 deletions kasa/device.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -138,15 +138,18 @@ class WifiNetwork:
"""Wifi network container."""

ssid: str
key_type: int
# This is available on both netif and on softaponboarding
key_type: int | None = None
# These are available only on softaponboarding
cipher_type: int | None = None
bssid: str | None = None
channel: int | None = None
# These are available on softaponboarding, SMART, and SMARTCAM devices
bssid: str | None = None
rssi: int | None = None

# For SMART devices
# These are available on both SMART and SMARTCAM devices
signal_level: int | None = None
auth: int | None = None
encryption: int | None = None


_LOGGER = logging.getLogger(__name__)
Expand Down
3 changes: 3 additions & 0 deletions kasa/iot/iotdevice.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -688,6 +688,9 @@ async def wifi_join(self, ssid: str, password: str, keytype: str = "3") -> dict:
async def _join(target: str, payload: dict) -> dict:
return await self._query_helper(target, "set_stainfo", payload)

if not keytype:
raise KasaException("KeyType is required for this device.")

payload = {"ssid": ssid, "password": password, "key_type": int(keytype)}
try:
return await _join("netif", payload)
Expand Down
1 change: 1 addition & 0 deletions kasa/protocols/smartprotocol.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ def mask_area(area: dict[str, Any]) -> dict[str, Any]:
# Queries that are known not to work properly when sent as a
# multiRequest. They will not return the `method` key.
FORCE_SINGLE_REQUEST = {
"connectAp",
"getConnectStatus",
"scanApList",
}
Expand Down
3 changes: 3 additions & 0 deletions kasa/smart/smartdevice.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -769,6 +769,9 @@ async def wifi_join(
if not self.credentials:
raise AuthenticationError("Device requires authentication.")

if not keytype:
raise KasaException("KeyType is required for this device.")

payload = {
"account": {
"username": base64.b64encode(
Expand Down
104 changes: 103 additions & 1 deletion kasa/smartcam/smartcamdevice.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,20 @@

from __future__ import annotations

import base64
import logging
from typing import Any, cast

from ..device import DeviceInfo
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey

from ..device import DeviceInfo, WifiNetwork
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..exceptions import AuthenticationError, DeviceError, KasaException
from ..module import Module
from ..protocols import SmartProtocol
from ..protocols.smartcamprotocol import _ChildCameraProtocolWrapper
from ..smart import SmartChildDevice, SmartDevice
from ..smart.smartdevice import ComponentsRaw
Expand All @@ -23,6 +31,24 @@ class SmartCamDevice(SmartDevice):
# Modules that are called as part of the init procedure on first update
FIRST_UPDATE_MODULES = {DeviceModule, ChildDevice}

STATIC_PUBLIC_KEY_B64 = (
"MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQC4D6i0oD/Ga5qb//RfSe8MrPVI"
"rMIGecCxkcGWGj9kxxk74qQNq8XUuXoy2PczQ30BpiRHrlkbtBEPeWLpq85tfubT"
"UjhBz1NPNvWrC88uaYVGvzNpgzZOqDC35961uPTuvdUa8vztcUQjEZy16WbmetRj"
"URFIiWJgFCmemyYVbQIDAQAB"
)

def __init__(
self,
host: str,
*,
config: DeviceConfig | None = None,
protocol: SmartProtocol | None = None,
) -> None:
super().__init__(host, config=config, protocol=protocol)
self._public_key: str | None = None
self._networks: list[WifiNetwork] = []

@staticmethod
def _get_device_type_from_sysinfo(sysinfo: dict[str, Any]) -> DeviceType:
"""Find type to be displayed as a supported device category."""
Expand Down Expand Up @@ -288,3 +314,79 @@ def hw_info(self) -> dict:
def rssi(self) -> int | None:
"""Return the device id."""
return self.modules[SmartCamModule.SmartCamDeviceModule].rssi

async def wifi_scan(self) -> list[WifiNetwork]:
"""Scan for available wifi networks."""

def _net_for_scan_info(res: dict) -> WifiNetwork:
return WifiNetwork(
ssid=res["ssid"],
auth=res["auth"],
encryption=res["encryption"],
rssi=res["rssi"],
bssid=res["bssid"],
)

_LOGGER.debug("Querying networks")

resp = await self._query_helper("scanApList", {"onboarding": {"scan": {}}})
scan_data: dict = resp["scanApList"]["onboarding"]["scan"]
self._public_key = scan_data.get("publicKey", "")
self._networks = [_net_for_scan_info(net) for net in scan_data["ap_list"]]
return self._networks

async def wifi_join(
self, ssid: str, password: str, keytype: str = "wpa2_psk"
) -> dict:
"""Join the given wifi network.

This method returns nothing as the device tries to activate the new
settings immediately instead of responding to the request.

If joining the network fails, the device will return to the previous state
after some delay.
"""
if not self.credentials:
raise AuthenticationError("Device requires authentication.")

if not self._networks:
await self.wifi_scan()
net = next(
(n for n in self._networks if getattr(n, "ssid", None) == ssid), None
)
if net is None:
raise DeviceError(f"Network with SSID '{ssid}' not found.")

public_key_b64 = self._public_key or self.STATIC_PUBLIC_KEY_B64
key_bytes = base64.b64decode(public_key_b64)
public_key = serialization.load_der_public_key(key_bytes)
if not isinstance(public_key, RSAPublicKey):
raise TypeError("Loaded public key is not an RSA public key")
encrypted = public_key.encrypt(password.encode(), padding.PKCS1v15())
encrypted_password = base64.b64encode(encrypted).decode()

payload = {
"onboarding": {
"connect": {
"auth": net.auth,
"bssid": net.bssid,
"encryption": net.encryption,
"password": encrypted_password,
"rssi": net.rssi,
"ssid": net.ssid,
}
}
}

# The device does not respond to the request but changes the settings
# immediately which causes us to timeout.
# Thus, We limit retries and suppress the raised exception as useless.
try:
return await self.protocol.query({"connectAp": payload}, retry_count=0)
except DeviceError:
raise # Re-raise on device-reported errors
except KasaException:
_LOGGER.debug(
"Received a kasa exception for wifi join, but this is expected"
)
return {}
46 changes: 46 additions & 0 deletions tests/fakeprotocol_smartcam.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -256,6 +256,52 @@ async def _send_request(self, request_dict: dict):
method = request_dict["method"]

info = self.info
if method == "connectAp":
if self.verbatim:
return {"error_code": -1}
return {"result": {}, "error_code": 0}
if method == "scanApList":
if method in info:
result = self._get_method_from_info(method, request_dict.get("params"))
if not self.verbatim:
scan = (
result.get("result", {}).get("onboarding", {}).get("scan", {})
)
ap_list = scan.get("ap_list")
if isinstance(ap_list, list) and not any(
ap.get("ssid") == "FOOBAR" for ap in ap_list
):
ap_list.append(
{
"ssid": "FOOBAR",
"auth": 3,
"encryption": 3,
"rssi": -40,
"bssid": "00:00:00:00:00:00",
}
)
return result
if self.verbatim:
return {"error_code": -1}
return {
"result": {
"onboarding": {
"scan": {
"publicKey": "",
"ap_list": [
{
"ssid": "FOOBAR",
"auth": 3,
"encryption": 3,
"rssi": -40,
"bssid": "00:00:00:00:00:00",
}
],
}
}
},
"error_code": 0,
}
if method == "controlChild":
return await self._handle_control_child(
request_dict["params"]["childControl"]
Expand Down
Loading

Back | FazBrowse Home | New Git URL