Reject relative paths in SFTP storage backup location config flow (#164408)

homeassistant/components/sftp_storage/config_flow.py

@@ -124,6 +124,17 @@ class SFTPFlowHandler(ConfigFlow, domain=DOMAIN):
                 }
             )
 
+            if not user_input[CONF_BACKUP_LOCATION].startswith("/"):
+                errors[CONF_BACKUP_LOCATION] = "backup_location_relative"
+                return self.async_show_form(
+                    step_id=step_id,
+                    data_schema=self.add_suggested_values_to_schema(
+                        DATA_SCHEMA, user_input
+                    ),
+                    description_placeholders=placeholders,
+                    errors=errors,
+                )
+
             try:
                 # Validate auth input and save uploaded key file if provided
                 user_input = await self._validate_auth_and_save_keyfile(user_input)

homeassistant/components/sftp_storage/strings.json

@@ -4,6 +4,7 @@
       "already_configured": "Integration already configured. Host with same address, port and backup location already exists."
     },
     "error": {
+      "backup_location_relative": "The remote path must be an absolute path (starting with `/`).",
       "invalid_key": "Invalid key uploaded. Please make sure key corresponds to valid SSH key algorithm.",
       "key_or_password_needed": "Please configure password or private key file location for SFTP Storage.",
       "os_error": "{error_message}. Please check if host and/or port are correct.",

tests/components/sftp_storage/conftest.py

@@ -31,7 +31,7 @@ from tests.common import MockConfigEntry
 type ComponentSetup = Callable[[], Awaitable[None]]
 
 BACKUP_METADATA = {
-    "file_path": "backup_location/backup.tar",
+    "file_path": "/backup_location/backup.tar",
     "metadata": {
         "addons": [{"name": "Test", "slug": "test", "version": "1.0.0"}],
         "backup_id": "test-backup",
@@ -60,7 +60,7 @@ USER_INPUT = {
     CONF_USERNAME: "username",
     CONF_PASSWORD: "password",
     CONF_PRIVATE_KEY_FILE: PRIVATE_KEY_FILE_UUID,
-    CONF_BACKUP_LOCATION: "backup_location",
+    CONF_BACKUP_LOCATION: "/backup_location",
 }
 TEST_AGENT_ID = ulid()
 
@@ -118,7 +118,7 @@ def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
             CONF_USERNAME: "username",
             CONF_PASSWORD: "password",
             CONF_PRIVATE_KEY_FILE: str(private_key),
-            CONF_BACKUP_LOCATION: "backup_location",
+            CONF_BACKUP_LOCATION: "/backup_location",
         },
     )
 

tests/components/sftp_storage/test_backup.py

@@ -151,7 +151,7 @@ async def test_agents_list_backups_include_bad_metadata(
     # Called two times, one for bad backup metadata and once for good
     assert mock_ssh_connection._sftp._mock_open._mock_read.call_count == 2
     assert (
-        "Failed to load backup metadata from file: backup_location/invalid.metadata.json. Expecting value: line 1 column 1 (char 0)"
+        "Failed to load backup metadata from file: /backup_location/invalid.metadata.json. Expecting value: line 1 column 1 (char 0)"
         in caplog.messages
     )
 

tests/components/sftp_storage/test_config_flow.py

@@ -15,6 +15,7 @@ from homeassistant.components.sftp_storage.config_flow import (
     SFTPStorageMissingPasswordOrPkey,
 )
 from homeassistant.components.sftp_storage.const import (
+    CONF_BACKUP_LOCATION,
     CONF_HOST,
     CONF_PASSWORD,
     CONF_PRIVATE_KEY_FILE,
@@ -194,3 +195,35 @@ async def test_config_entry_error(hass: HomeAssistant) -> None:
         result["flow_id"], user_input
     )
     assert "errors" in result and result["errors"]["base"] == "key_or_password_needed"
+
+
+@pytest.mark.usefixtures("current_request_with_host")
+@pytest.mark.usefixtures("mock_process_uploaded_file")
+@pytest.mark.usefixtures("mock_ssh_connection")
+async def test_relative_backup_location_rejected(
+    hass: HomeAssistant,
+) -> None:
+    """Test that a relative backup location path is rejected."""
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+    assert result["step_id"] == "user"
+
+    user_input = USER_INPUT.copy()
+    user_input[CONF_BACKUP_LOCATION] = "backups"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["errors"] == {CONF_BACKUP_LOCATION: "backup_location_relative"}
+
+    # Fix the path and verify the flow succeeds
+    user_input[CONF_BACKUP_LOCATION] = "/backups"
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY