Merge pull request #219 from timvlaer/set_wlan_guide_settings

wlan_guide_settings api

examples/initial_setup_e5576-320.py

@@ -0,0 +1,69 @@
+from argparse import ArgumentParser
+
+from huawei_lte_api.Client import Client
+from huawei_lte_api.Connection import Connection
+from time import sleep
+
+parser = ArgumentParser()
+parser.add_argument('--password', type=str)
+parser.add_argument('--newpassword', type=str)
+parser.add_argument('--ssid', type=str)
+parser.add_argument('--wpapassword', type=str)
+args = parser.parse_args()
+
+original_password = args.password
+new_password = args.newpassword
+
+wifi_ssid = args.ssid
+wifi_password = args.wpapassword
+
+url = 'http://192.168.8.1/'
+with Connection(url, password=original_password) as connection:
+    client = Client(connection)
+
+    locale = "en-us"
+    print("Set language to " + locale)
+    print(client.language.set_current_language(locale))
+
+    print("Accept privacy policy")
+    print(client.app.accept_privacypolicy(approve=True))
+
+    print("Set autoupdate config")
+    print(client.online_update.set_autoupdate_config(autoupdate=True))
+
+    print("Set basic information")
+    print(client.device.set_basic_information())
+
+    print(f"Set wlan ({wifi_ssid}/{wifi_password}) and account settings (admin/{new_password})")
+    resp = client.wlan.set_wlan_guide_settings(
+        ssid=wifi_ssid, wpa_psk=wifi_password, current_password=original_password, new_password=new_password
+    )
+    print(resp)
+
+print("Admin password changed, reconnect...")
+sleep(10)
+failing = True
+while failing:
+    try:
+        with Connection(url, password=new_password) as connection:
+            print("Get basic information")
+            status = client.monitoring.status()
+            if status["ConnectionStatus"] == "901":
+                failing = False
+            else:
+                sleep(60)
+    except Exception as e:
+        print("Failed with exception: " + str(e) + ", sleeping 60s")
+        sleep(60)
+
+with Connection(url, password=new_password) as connection:
+    client = Client(connection)
+
+    print("Set basic information")
+    print(client.device.set_basic_information())
+
+    # restart wifi to see effect of the new ssid
+    # see toggle_wifi.py
+
+
+

huawei_lte_api/Session.py

@@ -36,6 +36,7 @@ def wrapped(*args: Any, **kw: Any) -> T:
             args[0].reload()
             return fn(*args, **kw)
 
+
     return wrapped
 
 
@@ -253,8 +254,6 @@ def _post(self,
         )
         response.raise_for_status()
 
-        response_data = cast(str, self._check_response_status(self._process_response_data(response)))
-
         if refresh_csrf:
             self.request_verification_tokens = []
 
@@ -267,6 +266,8 @@ def _post(self,
         else:
             _LOGGER.debug('Failed to get CSRF from POST response headers')
 
+        response_data = cast(str, self._check_response_status(self._process_response_data(response)))
+
         return response_data
 
     def post_file(self,

huawei_lte_api/api/App.py

@@ -1,5 +1,6 @@
 from huawei_lte_api.ApiGroup import ApiGroup
-from huawei_lte_api.Session import GetResponseType
+from huawei_lte_api.Session import GetResponseType, SetResponseType
+from huawei_lte_api.exceptions import ResponseErrorException
 
 
 class App(ApiGroup):
@@ -8,3 +9,21 @@ def operatorinfo(self, lang: str = 'en_us') -> GetResponseType:
 
     def privacypolicy(self, lang: str = 'en_us') -> GetResponseType:
         return self._session.get('app/privacypolicy', {'lang': lang})
+
+    def accept_privacypolicy(self, approve: bool = False) -> SetResponseType:
+        response = self._session.post_get('app/privacypolicy',
+                                      {
+                                          "data": {
+                                              "Approve": "2" if approve else "0",
+                                              "Liscence": "0"  # deliberate typo
+                                          }
+                                      },
+                                      is_json=True)
+        error_code = response["errcode"]
+        if error_code == 0:
+            return "OK"
+
+        raise ResponseErrorException(
+            message="Unexpected response: " + str(response),
+            code=int(error_code)
+        )

huawei_lte_api/api/Device.py

@@ -17,6 +17,12 @@ def device_feature_switch(self) -> GetResponseType:
     def basic_information(self) -> GetResponseType:
         return self._session.get('device/basic_information')
 
+    def set_basic_information(self, restore_default_status: bool = False) -> SetResponseType:
+        return self._session.post_set('device/basic_information',
+                                      {
+                                          "restore_default_status": 1 if restore_default_status else 0
+                                      })
+
     def basicinformation(self) -> GetResponseType:
         return self._session.get('device/basicinformation')
 

huawei_lte_api/api/OnlineUpdate.py

@@ -46,5 +46,9 @@ def configuration(self) -> GetResponseType:
     def autoupdate_config(self) -> GetResponseType:
         return self._session.get('online-update/autoupdate-config')
 
+    def set_autoupdate_config(self, autoupdate: bool) -> SetResponseType:
+        return self._session.post_set('online-update/autoupdate-config',
+                                      {"auto_update": int(autoupdate is True), "ui_download": 0})
+
     def redirect_cancel(self) -> GetResponseType:
         return self._session.get('online-update/redirect_cancel')

huawei_lte_api/api/User.py

@@ -48,20 +48,39 @@ class User(ApiGroup):
     def state_login(self) -> GetResponseType:
         return self._session.get('user/state-login')
 
-    def _login(self, username: str, password: Optional[str], password_type: PasswordTypeEnum = PasswordTypeEnum.BASE_64) -> bool:
+    def _state_login_with_retry(self) -> GetResponseType:
+        tries = 5
+        for i in range(tries):
+            try:
+                state_login = self.state_login()
+                return state_login
+            except requests.exceptions.ConnectionError:
+                # Some models reportedly close the connection if we attempt to access login state too soon after
+                # setting up the session etc. In that case, retry a few times. The error is reported to be
+                # ConnectionError: (
+                #     'Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))
+                if i == tries - 1:
+                    raise
+                time.sleep((i + 1) / 10)
+
+        raise ResponseErrorException(message="Tries exhausted", code=0)
+
+    def _encode_password(self, username: str, password: Optional[str], password_type: PasswordTypeEnum = PasswordTypeEnum.BASE_64) -> bytes:
         if not password:
-            password_encoded = b''
-        else:
-            if password_type == PasswordTypeEnum.SHA256:
-                concentrated = b''.join([
-                    username.encode('UTF-8'),
-                    base64.b64encode(hashlib.sha256(password.encode('UTF-8')).hexdigest().encode('ascii')),
-                    self._session.request_verification_tokens[0].encode('UTF-8')
-                ])
-                password_encoded = base64.b64encode(hashlib.sha256(concentrated).hexdigest().encode('ascii'))
-            else:
-                password_encoded = base64.b64encode(password.encode('UTF-8'))
+            return b''
+
+        if password_type == PasswordTypeEnum.SHA256:
+            concentrated = b''.join([
+                username.encode('UTF-8'),
+                base64.b64encode(hashlib.sha256(password.encode('UTF-8')).hexdigest().encode('ascii')),
+                self._session.request_verification_tokens[0].encode('UTF-8')
+            ])
+            return base64.b64encode(hashlib.sha256(concentrated).hexdigest().encode('ascii'))
 
+        return base64.b64encode(password.encode('UTF-8'))
+
+    def _login(self, username: str, password: Optional[str], password_type: PasswordTypeEnum = PasswordTypeEnum.BASE_64) -> bool:
+        password_encoded = self._encode_password(username, password, password_type)
         try:
             result = self._session.post_set('user/login', {
                 'Username': username,
@@ -96,20 +115,11 @@ def _login(self, username: str, password: Optional[str], password_type: Password
     def login(self, username: str = DEFAULT_USERNAME, password: Optional[str] = None, force_new_login: bool = False) -> bool:
         if username == '':  # <= 1.6.4 backwards compatibility
             username = DEFAULT_USERNAME
-        tries = 5
-        for i in range(tries):
-            try:
-                state_login = self.state_login()
-            except requests.exceptions.ConnectionError:
-                # Some models reportedly close the connection if we attempt to access login state too soon after
-                # setting up the session etc. In that case, retry a few times. The error is reported to be
-                # ConnectionError: (
-                #     'Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))
-                if i == tries - 1:
-                    raise
-                time.sleep((i + 1) / 10)
-            except ResponseErrorNotSupportedException:
-                return True
+
+        try:
+            state_login = self._state_login_with_retry()
+        except ResponseErrorNotSupportedException:
+            return True
 
         if LoginStateEnum(int(state_login['State'])) == LoginStateEnum.LOGGED_IN and not force_new_login:
             return True
@@ -130,6 +140,12 @@ def password(self) -> GetResponseType:
     def pwd(self) -> GetResponseType:
         return self._session.get('user/pwd')
 
+    def set_pwd(self) -> SetResponseType:
+        return self._session.post_set('user/pwd', {
+            "module": "wlan",
+            "nonce": "aaaaaaa"
+        })
+
     def set_remind(self, remind_state: str) -> SetResponseType:
         return self._session.post_set('user/remind', {
             'remindstate': remind_state

huawei_lte_api/api/WLan.py

@@ -257,5 +257,33 @@ def wifi_network_switch(self, status: bool, criteria: Optional[dict] = None) ->
     def wlandbho(self) -> GetResponseType:
         return self._session.get('wlan/wlandbho')
 
+    def wlan_guide_settings(self) -> GetResponseType:
+        return self._session.get('wlan/wlan-guide-settings')
+
+    def set_wlan_guide_settings(self, ssid: str, wpa_psk: str, current_password: str, new_password: str) -> SetResponseType:
+        ssids = self.wlan_guide_settings()["Ssids"]["Ssid"]
+        new_ssid = ssids[0] if len(ssids) > 0 else {"Index": "0"}
+        new_ssid["WifiSsid"] = ssid
+        new_ssid["WifiWpapsk"] = wpa_psk
+
+        self._session.reload()
+
+        data = {
+            "Ssids": {
+                "Ssid": [new_ssid]
+            },
+            "rebootInfo": {
+                "isReboot": 0
+            },
+            "accountInfo": {
+                "currentpassword": current_password,
+                "newpassword": new_password,
+                "confirmpwd": new_password}
+        }
+        return self._session.post_set('wlan/wlan-guide-settings',
+                                      data,
+                                      refresh_csrf=True,
+                                      is_encrypted=True)
+
     def wlanintelligent(self) -> GetResponseType:
         return self._session.get('wlan/wlanintelligent')

huawei_lte_api/enums/client.py

@@ -16,4 +16,4 @@ class ResponseCodeEnum(enum.IntEnum):
     ERROR_VOICE_BUSY = 120001  # Unused
     ERROR_WRONG_TOKEN = 125001  # Unused
     ERROR_SYSTEM_CSRF = 125002
-    ERROR_WRONG_SESSION_TOKEN = 125003  # Unused
+    ERROR_WRONG_SESSION_TOKEN = 125003