Bug fixes

Author: QuinnDamerell

homeway/homeway/commandhandler.py

@@ -241,13 +241,14 @@ def ProcessCommand(self, commandPath:str, jsonObj_CanBeNone:Optional[Dict[str, A
                 domainsFilterRaw = jsonObj_CanBeNone.get("FilterDomains", None)
                 if domainsFilterRaw is not None:
                     if isinstance(domainsFilterRaw, list) and all(isinstance(d, str) for d in domainsFilterRaw):
-                        domainsFilter = domainsFilterRaw
+                        if len(domainsFilterRaw) > 0:
+                            domainsFilter = domainsFilterRaw
                     else:
-                        return CommandResponse.Error(CommandHandler.c_CommandError_ArgParseFailure, "'DomainsFilter' must be a list of strings.")
+                        return CommandResponse.Error(CommandHandler.c_CommandError_ArgParseFailure, "'FilterDomains' must be a list of strings.")
 
-            allEntities, allStates = self.HomeContext.GetFullDeviceAndEntityTree(forceRefresh, includeStates, domainsFilter)
+            allEntities, allStates, labels = self.HomeContext.GetFullDeviceAndEntityTree(forceRefresh, includeStates, domainsFilter)
             successful = allEntities is not None
-            return CommandResponse.Success({"Success": successful, "Floors": allEntities, "States": allStates})
+            return CommandResponse.Success({"Success": successful, "Floors": allEntities, "States": allStates, "Labels": labels})
 
         # Used by Sage and MCP to get the current home context and live state in a single round trip.
         if commandPathLower.startswith("get-live-context"):
@@ -503,7 +504,7 @@ def HandleReadTextFileCommand(self, jsonArgs:Dict[str, Any]) -> CommandResponse:
             return textEncodingError
 
         rawStartLine = self._GetCommandArg(jsonArgs, "StartLine")
-        startLine, startLineError = self._ParseOptionalNonNegativeInt(rawStartLine, "StartLine")
+        startLine, startLineError = self._ParseOptionalPositiveInt(rawStartLine, "StartLine")
         if startLineError is not None:
             return startLineError
 
@@ -543,6 +544,8 @@ def HandleWriteFileCommand(self, jsonArgs:Dict[str, Any]) -> CommandResponse:
         base64Data, base64DataError = self._ParseOptionalString(rawBase64Data, "Base64Data")
         if base64DataError is not None:
             return base64DataError
+        if text is not None and base64Data is not None:
+            return CommandResponse.Error(CommandHandler.c_CommandError_ArgParseFailure, "Only one of 'Text' or 'Base64Data' can be provided.")
 
         rawTextEncoding = self._GetCommandArg(jsonArgs, "TextEncoding")
         textEncoding, textEncodingError = self._ParseOptionalString(rawTextEncoding, "TextEncoding")
@@ -581,8 +584,16 @@ def HandleMoveFileCommand(self, jsonArgs:Dict[str, Any]) -> CommandResponse:
         rawCopy = self._GetCommandArg(jsonArgs, "Copy")
         copy = self._ParseOptionalBool(rawCopy, False)
 
+        rawOverride = self._GetCommandArg(jsonArgs, "Override")
+        override = self._ParseOptionalBool(rawOverride, False)
+
+        rawExpectedSha256 = self._GetCommandArg(jsonArgs, "ExpectedSha256")
+        expectedSha256, expectedSha256Error = self._ParseOptionalString(rawExpectedSha256, "ExpectedSha256")
+        if expectedSha256Error is not None:
+            return expectedSha256Error
+
         try:
-            return CommandResponse.Success(self.HomeAssistantFileSystem.MoveFile(rawPath, rawNewPath, copy))
+            return CommandResponse.Success(self.HomeAssistantFileSystem.MoveFile(rawPath, rawNewPath, copy, override, expectedSha256))
         except Exception as e:
             return self._FileSystemExceptionToCommandResponse(e)
 
@@ -618,8 +629,11 @@ def HandleDeleteFileCommand(self, jsonArgs:Dict[str, Any]) -> CommandResponse:
         if not isinstance(rawPath, str):
             return CommandResponse.Error(CommandHandler.c_CommandError_ArgParseFailure, "'Path' must be a string.")
 
+        rawRecursive = self._GetCommandArg(jsonArgs, "Recursive")
+        recursive = self._ParseOptionalBool(rawRecursive, False)
+
         try:
-            return CommandResponse.Success(self.HomeAssistantFileSystem.DeleteFile(rawPath))
+            return CommandResponse.Success(self.HomeAssistantFileSystem.DeleteFile(rawPath, recursive))
         except Exception as e:
             return self._FileSystemExceptionToCommandResponse(e)
 

homeway/homeway/interfaces.py

@@ -105,15 +105,15 @@ def WriteFile(self, path:str, text:Optional[str], base64Data:Optional[str], text
         pass
 
     @abstractmethod
-    def MoveFile(self, path:str, newPath:str, copy:bool) -> Dict[str, Any]:
+    def MoveFile(self, path:str, newPath:str, copy:bool, override:bool, expectedSha256:Optional[str]) -> Dict[str, Any]:
         pass
 
     @abstractmethod
     def PatchFile(self, path:str, unifiedDiffPatch:str, expectedSha256:Optional[str]) -> Dict[str, Any]:
         pass
 
     @abstractmethod
-    def DeleteFile(self, path:str) -> Dict[str, Any]:
+    def DeleteFile(self, path:str, recursive:bool) -> Dict[str, Any]:
         pass
 
 
@@ -139,7 +139,7 @@ def GetStatesAndLiveContext(self) -> Tuple[Optional["CompressionResult"], Option
     # Gets the full device and entity tree stored in our cache.
     # Optionally also gets the states and can be filtered by domains.
     @abstractmethod
-    def GetFullDeviceAndEntityTree(self, forceRefresh: bool, includeStates:bool=False, domainsFilter:Optional[List[str]]=None) -> Tuple[Optional[List[Dict[str, Any]]], Optional[List[Dict[str, Any]]]]:
+    def GetFullDeviceAndEntityTree(self, forceRefresh: bool, includeStates:bool=False, domainsFilter:Optional[List[str]]=None) -> Tuple[Optional[List[Dict[str, Any]]], Optional[List[Dict[str, Any]]], Optional[List[Dict[str, Any]]]]:
         pass
 
     # Looks up a full entity dict by its entity ID, or None if not found.

homeway/homeway_linuxhost/ha/filesystem.py

@@ -68,8 +68,8 @@ def ListFiles(self, path:str, recursive:bool) -> Dict[str, Any]:
                     self._AddFileListEntry(files, rootPath, entryPath, entryName)
 
         return {
-            "Path": self._ToResponseRelativePath(normalizedPath),
-            "Files": files,
+            "path": self._ToResponseRelativePath(normalizedPath),
+            "files": files,
         }
 
 
@@ -106,7 +106,11 @@ def ReadTextFile(self, path:str, textEncoding:Optional[str], startLine:Optional[
         fileSize = os.path.getsize(targetPath)
         lines, fullLineCount, readStartLine = self._ReadTextLines(targetPath, textEncoding, startLine, maxLines, tailLines)
         linesRead = len(lines)
-        readEndLine = readStartLine + linesRead - 1
+        if linesRead == 0:
+            readStartLine = 0
+            readEndLine = 0
+        else:
+            readEndLine = readStartLine + linesRead - 1
         isPartialRead = fullLineCount > 0 and (readStartLine > 1 or readEndLine < fullLineCount)
 
         # Note! These properties are used by the MCP server and explicitly deserialized by the server, so they must stay in sync!
@@ -147,7 +151,7 @@ def WriteFile(self, path:str, text:Optional[str], base64Data:Optional[str], text
         }
 
 
-    def MoveFile(self, path:str, newPath:str, copy:bool) -> Dict[str, Any]:
+    def MoveFile(self, path:str, newPath:str, copy:bool, override:bool, expectedSha256:Optional[str]) -> Dict[str, Any]:
         rootPath = self._GetRootPath(True)
         sourceOperationName = "copy" if copy else "move"
         _, targetPath = self._ResolveExistingFilePath(rootPath, path, sourceOperationName)
@@ -157,6 +161,11 @@ def MoveFile(self, path:str, newPath:str, copy:bool) -> Dict[str, Any]:
         if os.path.realpath(targetPath) == os.path.realpath(newTargetPath):
             raise ValueError("'NewPath' must be different from 'Path'.")
 
+        expectedSha256 = self._NormalizeExpectedSha256(expectedSha256)
+        self._ValidateExpectedSha256(targetPath, expectedSha256)
+        if override is False and os.path.exists(newTargetPath):
+            raise FileExistsError("Destination file already exists and 'Override' is false.")
+
         fileSize = os.path.getsize(targetPath)
         if copy:
             shutil.copy2(targetPath, newTargetPath)
@@ -186,7 +195,7 @@ def PatchFile(self, path:str, unifiedDiffPatch:str, expectedSha256:Optional[str]
         try:
             applied = patchSet.apply()
             if applied is False:
-                raise RuntimeError("Patch could not be applied.")
+                raise RuntimeError("Patch context did not match the current file.")
             if os.path.exists(targetPath) is False:
                 raise RuntimeError("Patch removed file unexpectedly.")
             self._ReadUtf8TextFileBytesForPatch(targetPath, "Patched file")
@@ -203,15 +212,26 @@ def PatchFile(self, path:str, unifiedDiffPatch:str, expectedSha256:Optional[str]
         }
 
 
-    def DeleteFile(self, path:str) -> Dict[str, Any]:
+    def DeleteFile(self, path:str, recursive:bool) -> Dict[str, Any]:
         rootPath = self._GetRootPath(True)
-        _, targetPath = self._ResolveExistingFilePath(rootPath, path, "remove")
-
-        os.remove(targetPath)
+        _, targetPath = self._ResolveExistingPath(rootPath, path, "remove")
+
+        isDirectory = os.path.isdir(targetPath)
+        if isDirectory:
+            if recursive:
+                self._ValidateDirectoryTreeForDelete(rootPath, targetPath)
+                shutil.rmtree(targetPath)
+            else:
+                if len(os.listdir(targetPath)) > 0:
+                    raise OSError("Directory is not empty. Set 'Recursive' to true to delete it.")
+                os.rmdir(targetPath)
+        else:
+            os.remove(targetPath)
 
         # Note! These properties are used by the MCP server and explicitly deserialized by the server, so they must stay in sync!
         return {
             "deleted": True,
+            "is_directory": isDirectory,
         }
 
 
@@ -234,15 +254,20 @@ def _ResolvePath(self, rootPath:str, rawPath:str, allowRoot:bool) -> Tuple[str,
 
 
     def _ResolveExistingFilePath(self, rootPath:str, rawPath:str, operationName:str) -> Tuple[str, str]:
+        normalizedPath, targetPath = self._ResolveExistingPath(rootPath, rawPath, operationName)
+        if os.path.isfile(targetPath) is False:
+            raise ValueError("'Path' must reference a file.")
+        return normalizedPath, targetPath
+
+
+    def _ResolveExistingPath(self, rootPath:str, rawPath:str, operationName:str) -> Tuple[str, str]:
         normalizedPath, targetPath = self._ResolvePath(rootPath, rawPath, False)
         if self._IsDeniedFileName(normalizedPath):
             raise PermissionError("Access to this file is denied.")
         if os.path.exists(targetPath) is False:
             raise FileNotFoundError("File does not exist.")
         if os.path.islink(targetPath):
             raise PermissionError(f"Refusing to {operationName} a symbolic link.")
-        if os.path.isfile(targetPath) is False:
-            raise ValueError("'Path' must reference a file.")
         return normalizedPath, targetPath
 
 
@@ -291,9 +316,11 @@ def _GetTextEncoding(self, textEncoding:Optional[str]) -> str:
 
 
     def _GetTextStartLine(self, startLine:Optional[int]) -> int:
-        # StartLine is 1-based for callers, but accept 0 as the beginning like the byte offset API did.
-        if startLine is None or startLine == 0:
+        # StartLine is 1-based for callers.
+        if startLine is None:
             return 1
+        if startLine <= 0:
+            raise ValueError("'StartLine' must be greater than zero.")
         return startLine
 
 
@@ -339,15 +366,17 @@ def _ReadTextLineRange(self, targetPath:str, textEncoding:str, startLine:int, ma
 
 
     def _GetWriteContentBytes(self, text:Optional[str], base64Data:Optional[str], textEncoding:Optional[str]) -> bytes:
+        if text is not None and base64Data is not None:
+            raise ValueError("Only one of 'Text' or 'Base64Data' can be provided.")
+        if text is None and base64Data is None:
+            raise ValueError("One of 'Text' or 'Base64Data' must be provided.")
         if text is not None:
             if textEncoding is None:
                 textEncoding = "utf-8"
             try:
                 return text.encode(textEncoding)
             except LookupError as e:
                 raise ValueError(f"Unknown text encoding '{textEncoding}'.") from e
-        if base64Data is None:
-            raise ValueError("'Base64Data' must be provided when 'Format' is 'data'.")
         try:
             return base64.b64decode(base64Data, validate=True)
         except Exception as e:
@@ -423,6 +452,23 @@ def _RestoreFileBytesAfterFailedPatch(self, targetPath:str, fileBytes:bytes) ->
             f.write(fileBytes)
 
 
+    def _ValidateDirectoryTreeForDelete(self, rootPath:str, targetPath:str) -> None:
+        for currentRoot, dirNames, fileNames in os.walk(targetPath):
+            if self._IsPathInsideRoot(rootPath, currentRoot) is False:
+                raise PermissionError("Directory contains a path outside of the Home Assistant config directory.")
+            for dirName in dirNames:
+                entryPath = os.path.join(currentRoot, dirName)
+                if os.path.islink(entryPath):
+                    raise PermissionError("Refusing to delete a directory containing a symbolic link.")
+            for fileName in fileNames:
+                entryPath = os.path.join(currentRoot, fileName)
+                normalizedPath = os.path.relpath(entryPath, rootPath)
+                if os.path.islink(entryPath):
+                    raise PermissionError("Refusing to delete a directory containing a symbolic link.")
+                if self._IsDeniedFileName(normalizedPath):
+                    raise PermissionError("Refusing to delete a directory containing a denied file.")
+
+
     def _HashFile(self, targetPath:str) -> str:
         sha256 = hashlib.sha256()
         with open(targetPath, "rb") as f:

homeway/homeway_linuxhost/ha/homecontext.py

@@ -71,6 +71,7 @@ def __init__(self, logger:logging.Logger, haConnection:Connection, eventHandler:
 
         # These cache the full device and entity tree for use by the server and other components.
         self.FullDeviceAndEntityTree:Optional[List[Dict[str, Any]]] = None
+        self.FullLabels:Optional[List[Dict[str, Any]]] = None
         # This is the same entity that are in the full tree, but just dumped into a map for fast lookups.
         self.FullEntityMap:Optional[Dict[str, Dict[str, Any]]] = None
 
@@ -107,9 +108,11 @@ def GetSageHomeContext(self) -> Optional[CompressionResult]:
 
 
     # Returns the full floor -> area -> device -> entity tree.
-    def GetFullDeviceAndEntityTree(self, forceRefresh: bool, includeStates:bool=False, domainsFilter:Optional[List[str]]=None) -> Tuple[Optional[List[Dict[str, Any]]], Optional[List[Dict[str, Any]]]]:
+    def GetFullDeviceAndEntityTree(self, forceRefresh: bool, includeStates:bool=False, domainsFilter:Optional[List[str]]=None) -> Tuple[Optional[List[Dict[str, Any]]], Optional[List[Dict[str, Any]]], Optional[List[Dict[str, Any]]]]:
         # Always start with the full entity tree.
         allEntities = self._GetFullDeviceAndEntityTree(forceRefresh)
+        with self.CacheLock:
+            labels = self.FullLabels
 
         # If we got a result, filter if needed.
         if allEntities is not None:
@@ -123,7 +126,7 @@ def GetFullDeviceAndEntityTree(self, forceRefresh: bool, includeStates:bool=Fals
             (states, _) = self._FilterStateList(states, useSageFiltering=False, domainsFilter=domainsFilter)
 
         # Done!
-        return (allEntities, states)
+        return (allEntities, states, labels)
 
 
     # Returns the full floor -> area -> device -> entity tree.
@@ -704,6 +707,7 @@ def _HandleAllObjectsResult(self, result:"HomeContextQueryResult") -> Tuple[int,
             self.AssistantDeviceContexts = assistDeviceContexts
             # This is the full device and entity tree for other components that need it.
             self.FullDeviceAndEntityTree = fullDeviceAndEntityTree
+            self.FullLabels = labelsList
             # This is the full entity map for fast lookups.
             self.FullEntityMap = fullEntityMap
             self.CacheUpdatedEvent.set()