Dev

[mikey0000]

Description

• Revamped device integration with new configurations and limits.
• Implemented encryption for secure communication.
• Enhanced MQTT handling for device status updates.
• Refactored device management structure for better organization.

---

Changes walkthrough 📝

Relevant files

Enhancement

9 files device_config.py Device Configuration and Limits Implementation                      pymammotion/utility/device_config.py Added device limits configuration for various models. Implemented methods to retrieve device configurations and working parameters. Introduced default configurations for Luba and Yuka devices. +754/-0  h2client.py HTTP/2 Client for File Uploads                                                      pymammotion/mqtt/linkkit/h2client.py Added HTTP/2 client for handling file uploads. Implemented stream handling for file uploads. Introduced error handling for file upload processes. +585/-0  device.py MowingDevice Class Update                                                                pymammotion/data/model/device.py Updated MowingDevice class to include device firmware information. Removed unused properties and methods related to device limits. +31/-222 encryption.py Encryption Utilities Implementation                                            pymammotion/http/encryption.py Implemented encryption utilities for secure data handling. Added methods for AES and RSA encryption. +220/-0  device_limits.py Device Limits Management                                                                  pymammotion/data/model/device_limits.py Created DeviceLimits class for managing device operational limits. Added validation methods for device limits. +49/-0    http.py HTTP Integration with Encryption                                                  pymammotion/http/http.py Integrated encryption utilities into HTTP requests. Updated login method to use encrypted credentials. +92/-39  mammotion.py Device Management Refactor                                                              pymammotion/mammotion/devices/mammotion.py Refactored device management to use MammotionDeviceManager. Updated login method to utilize new HTTP client. +15/-12  mammotion_cloud.py Cloud Device MQTT Enhancements                                                      pymammotion/mammotion/devices/mammotion_cloud.py Enhanced MQTT message handling to include device status updates. Added event handling for device status messages. +17/-9    state_manager.py State Management Improvements                                                        pymammotion/data/state_manager.py Updated state management to handle device status changes. Improved device online/offline tracking. +29/-11

---

💡 Penify usage:
Comment /help on the PR to get a list of all available Penify tools and their descriptions

[penify-dev]

PR Review 🔍

⏱️ Estimated effort to review [1-5]	4, because the PR introduces significant changes across multiple files, including new classes and methods for device configuration, encryption, and MQTT handling. The complexity of the changes and the number of files affected will require a thorough review to ensure everything integrates correctly.
🧪 Relevant tests	No
⚡ Possible issues	Possible Bug: The get_working_parameters method in DeviceConfig class returns None if the configuration is not found, which may lead to unexpected behavior if not handled properly in the calling code.
	Possible Bug: The encrypt_by_public_key method in EncryptionUtils may raise an error if the public key is not initialized, which could lead to crashes if not properly managed.
🔒 Security concerns	- Sensitive information exposure: The private key is hardcoded in the EncryptionUtils class, which poses a security risk. It should be securely managed and not exposed in the codebase.

[penify-dev]

PR Code Suggestions ✨

Category	Suggestion	Score
Possible bug	Define the id variable before using it to create an H2FileTask In the upload_file_async method, ensure that the id variable is defined before it is used to create an H2FileTask instance. pymammotion/mqtt/linkkit/h2client.py [364] -return H2FileTask(id, file_info, future_result) +task_id = self.__create_stream_id() +return H2FileTask(task_id, file_info, future_result) Suggestion importance[1-10]: 9 Why: This suggestion addresses a critical bug where the id variable is used without being defined, which would lead to a runtime error. Defining it before use is essential for proper functionality.	9
	Add a safeguard against division by zero for orientation calculation Ensure that the division operation for orientation is safe from potential division by zero errors. pymammotion/data/model/device.py [91] -self.location.orientation = int(location.real_toward / 10000) +self.location.orientation = int(location.real_toward / 10000) if location.real_toward != 0 else 0 Suggestion importance[1-10]: 8 Why: The suggestion addresses a potential division by zero issue, which could lead to runtime errors. Adding a safeguard improves the robustness of the code.	8
Security	Use a cryptographic random generator for AES key generation to enhance security Ensure that the AES key and IV are securely generated and not predictable; consider using a cryptographic random generator. pymammotion/http/encryption.py [40] -self.AES_PASW = self.get_aes_key() # Get from previous implementation +self.AES_PASW = secrets.token_bytes(16) # Securely generate a random AES key Suggestion importance[1-10]: 9 Why: This suggestion addresses a critical security concern by recommending the use of a cryptographic random generator, which is essential for secure key generation.	9
	Add validation for key and IV lengths to ensure they meet AES requirements Validate the length of the key and IV before using them in encryption to prevent potential errors or vulnerabilities. pymammotion/http/encryption.py [96] +if len(key_bytes) != 16 or len(iv_bytes) != 16: + raise ValueError("Key and IV must be 16 bytes long.") cipher = Cipher(algorithms.AES(key_bytes), modes.CBC(iv_bytes), backend=default_backend()) Suggestion importance[1-10]: 7 Why: Adding validation for key and IV lengths is a good practice to ensure compliance with AES requirements, though it may not address a major vulnerability.	7
Error handling	Add exception handling to the oauth_check method for better error management Ensure that the oauth_check method handles potential exceptions when making the HTTP request to avoid unhandled errors. pymammotion/http/http.py [42-50] async def oauth_check(self) -> Response: ... - async with session.post("/user-server/v1/user/oauth/check", headers=self._headers) as resp: + try: + async with session.post("/user-server/v1/user/oauth/check", headers=self._headers) as resp: + data = await resp.json() + return Response.from_dict(data) + except Exception as e: + print(f"Error during OAuth check: {e}") + return Response.from_dict({"status": 500, "msg": "Internal Server Error"}) Suggestion importance[1-10]: 9 Why: Adding exception handling in the oauth_check method is crucial for managing potential errors during HTTP requests, which can lead to unhandled exceptions.	9
	Improve error handling in the pair_devices_mqtt method by checking the response status In the pair_devices_mqtt method, check the response status before attempting to access the data to avoid potential key errors. pymammotion/http/http.py [52-64] async def pair_devices_mqtt(self, mower_name: str, rtk_name: str) -> Response: ... if data.get("status") == 200: print(data) return Response.from_dict(data) + else: + print(f"Error pairing devices: {data.get('msg', 'Unknown error')}") + return Response.from_dict({"status": data.get("status"), "msg": "Pairing failed"}) Suggestion importance[1-10]: 8 Why: This suggestion improves error handling by ensuring that the response status is checked, which prevents potential key errors when accessing the response data.	8
	Validate the response structure in the login method to avoid accessing undefined properties In the login method, ensure that the response is checked for a valid structure before accessing its properties to prevent runtime errors. pymammotion/http/http.py [109-133] async def login(self, username: str, password: str) -> Response[LoginResponseData]: ... - login_response = Response[LoginResponseData].from_dict(data) - self.login_info = LoginResponseData.from_dict(login_response.data) + if 'data' in data: + login_response = Response[LoginResponseData].from_dict(data) + self.login_info = LoginResponseData.from_dict(login_response.data) + else: + print("Invalid login response structure") + return Response.from_dict({"status": 400, "msg": "Invalid response"}) Suggestion importance[1-10]: 8 Why: Validating the response structure in the login method is important to avoid runtime errors when accessing properties, enhancing the robustness of the code.	8
	Enhance error handling in the net_rtk_enable method by checking the response status In the net_rtk_enable method, ensure that the response is checked for a valid status before proceeding to process the data. pymammotion/http/http.py [80-90] async def net_rtk_enable(self, device_id: str) -> Response: ... if data.get("status") == 200: print(data) return Response.from_dict(data) + else: + print(f"Error enabling net RTK: {data.get('msg', 'Unknown error')}") + return Response.from_dict({"status": data.get("status"), "msg": "Enable failed"}) Suggestion importance[1-10]: 8 Why: This suggestion enhances error handling by ensuring the response status is validated, which helps prevent issues when processing the response data.	8
Best practice	Replace generic exceptions with more specific ones for better error handling Consider using a more specific exception type instead of the generic Exception when loading keys, to provide clearer error handling. pymammotion/http/encryption.py [52] -raise Exception(f"Failed to load private key: {e!s}") +raise ValueError(f"Failed to load private key: {e!s}") Suggestion importance[1-10]: 8 Why: Using a more specific exception type like ValueError improves error handling and makes it easier to catch specific errors related to key loading.	8
Possible issue	Add a check to ensure streams are closed only if a connection exists Ensure that the __close_all_streams method is called even if the connection is not established, to avoid potential resource leaks. pymammotion/mqtt/linkkit/h2client.py [351] -self.__close_all_streams() +if self.__conn: + self.__close_all_streams() Suggestion importance[1-10]: 7 Why: The suggestion correctly identifies a potential resource leak by ensuring that streams are closed only if a connection exists, which is a good practice for resource management.	7
	Add handling for unexpected firmware types in the update method Ensure that the update_device_firmwares method handles unexpected firmware types gracefully. pymammotion/data/model/device.py [127-142] match mod.type: + case 1 | 3 | 4 | 5 | 101 | 102 | 103: + # existing assignments + case _: + # Handle unexpected firmware types Suggestion importance[1-10]: 7 Why: Adding handling for unexpected firmware types is a good practice to prevent potential runtime errors, making the code more resilient.	7
	Validate fw_info before processing to prevent errors Consider validating fw_info before processing it in update_device_firmwares to ensure it contains expected data. pymammotion/data/model/device.py [124] def update_device_firmwares(self, fw_info: DeviceFwInfo) -> None: + if not fw_info or not hasattr(fw_info, 'mod'): + return # or raise an exception Suggestion importance[1-10]: 6 Why: Validating fw_info before processing is a prudent measure to prevent errors, although the current implementation may still function correctly without it.	6
	Ensure the header dictionary is initialized to avoid potential errors In the __fill_auth_header method, ensure that the header dictionary is initialized before being filled to avoid potential NoneType errors. pymammotion/mqtt/linkkit/h2client.py [531] -header = {} +header = header or {} Suggestion importance[1-10]: 5 Why: The suggestion is valid but the existing code already initializes the header dictionary, making this suggestion unnecessary. It does not significantly improve the code.	5
Enhancement	Improve the error message for better clarity when an exception is raised In the __check_response method, consider raising a more specific exception or providing a more detailed message for better debugging. pymammotion/mqtt/linkkit/h2client.py [289] -raise H2Exception(response.status, msg if msg else "fail to request http/2, code:%d" % (response.status)) +raise H2Exception(response.status, msg if msg else f"HTTP/2 request failed with status code: {response.status}") Suggestion importance[1-10]: 6 Why: While the suggestion improves clarity in error messages, it is more of a code style enhancement rather than a critical issue. It helps with debugging but does not address a major bug.	6
Maintainability	Clarify the type of err_code_list_time for better type safety Consider using a more explicit type for err_code_list_time to avoid potential issues with type inference. pymammotion/data/model/device.py [41] -err_code_list_time: list | None = field(default_factory=list) +err_code_list_time: Optional[list] = field(default_factory=list) Suggestion importance[1-10]: 5 Why: While clarifying the type of err_code_list_time can enhance readability and maintainability, the current type is still functional. This is a minor improvement.	5
	Enhance logging messages to provide more context during encryption failures Consider using a more descriptive logging message when encryption fails to aid in debugging. pymammotion/http/encryption.py [158] -_LOGGER.error("Encryption failed: %s", str(err)) +_LOGGER.error("Encryption failed during RSA encryption: %s", str(err)) Suggestion importance[1-10]: 5 Why: While enhancing logging messages is beneficial for maintainability, this suggestion does not address a critical issue and is more of a minor improvement.	5