Allow specifying CIFS version (#4395)

* Allow specifying cifs_version

* cifs_version -> version

supervisor/mounts/const.py

@@ -27,3 +27,10 @@ class MountUsage(str, Enum):
     BACKUP = "backup"
     MEDIA = "media"
     SHARE = "share"
+
+
+class MountCifsVersion(str, Enum):
+    """Mount CIFS version."""
+
+    LEGACY_1_0 = "1.0"
+    LEGACY_2_0 = "2.0"

supervisor/mounts/mount.py

@@ -8,7 +8,14 @@ from pathlib import Path, PurePath
 from dbus_fast import Variant
 from voluptuous import Coerce
 
-from ..const import ATTR_NAME, ATTR_PASSWORD, ATTR_PORT, ATTR_TYPE, ATTR_USERNAME
+from ..const import (
+    ATTR_NAME,
+    ATTR_PASSWORD,
+    ATTR_PORT,
+    ATTR_TYPE,
+    ATTR_USERNAME,
+    ATTR_VERSION,
+)
 from ..coresys import CoreSys, CoreSysAttributes
 from ..dbus.const import (
     DBUS_ATTR_DESCRIPTION,
@@ -30,7 +37,15 @@ from ..exceptions import (
 from ..resolution.const import ContextType, IssueType
 from ..resolution.data import Issue
 from ..utils.sentry import capture_exception
-from .const import ATTR_PATH, ATTR_SERVER, ATTR_SHARE, ATTR_USAGE, MountType, MountUsage
+from .const import (
+    ATTR_PATH,
+    ATTR_SERVER,
+    ATTR_SHARE,
+    ATTR_USAGE,
+    MountCifsVersion,
+    MountType,
+    MountUsage,
+)
 from .validate import MountData
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -332,6 +347,7 @@ class CIFSMount(NetworkMount):
         if not skip_secrets and self.username is not None:
             out[ATTR_USERNAME] = self.username
             out[ATTR_PASSWORD] = self.password
+        out[ATTR_VERSION] = self.version
         return out
 
     @property
@@ -349,6 +365,16 @@ class CIFSMount(NetworkMount):
         """Get password, returns none if auth is not used."""
         return self._data.get(ATTR_PASSWORD)
 
+    @property
+    def version(self) -> str | None:
+        """Get password, returns none if auth is not used."""
+        version = self._data.get(ATTR_VERSION)
+        if version == MountCifsVersion.LEGACY_1_0:
+            return "1.0"
+        if version == MountCifsVersion.LEGACY_2_0:
+            return "2.0"
+        return None
+
     @property
     def what(self) -> str:
         """What to mount."""
@@ -357,11 +383,16 @@ class CIFSMount(NetworkMount):
     @property
     def options(self) -> list[str]:
         """Options to use to mount."""
-        return (
+        options = super().options
-            super().options + [f"username={self.username}", f"password={self.password}"]
+        if self.version:
-            if self.username and self.password
+            options.append(f"vers={self.version}")
-            else ["guest"]
+
-        )
+        if self.username and self.password:
+            options.extend([f"username={self.username}", f"password={self.password}"])
+        else:
+            options.append("guest")
+
+        return options
 
 
 class NFSMount(NetworkMount):

supervisor/mounts/validate.py

@@ -6,7 +6,14 @@ from typing import TypedDict
 from typing_extensions import NotRequired
 import voluptuous as vol
 
-from ..const import ATTR_NAME, ATTR_PASSWORD, ATTR_PORT, ATTR_TYPE, ATTR_USERNAME
+from ..const import (
+    ATTR_NAME,
+    ATTR_PASSWORD,
+    ATTR_PORT,
+    ATTR_TYPE,
+    ATTR_USERNAME,
+    ATTR_VERSION,
+)
 from ..validate import network_port
 from .const import (
     ATTR_DEFAULT_BACKUP_MOUNT,
@@ -15,6 +22,7 @@ from .const import (
     ATTR_SERVER,
     ATTR_SHARE,
     ATTR_USAGE,
+    MountCifsVersion,
     MountType,
     MountUsage,
 )
@@ -51,6 +59,9 @@ SCHEMA_MOUNT_CIFS = _SCHEMA_MOUNT_NETWORK.extend(
         vol.Required(ATTR_SHARE): VALIDATE_SHARE,
         vol.Inclusive(ATTR_USERNAME, "basic_auth"): VALIDATE_USERNAME,
         vol.Inclusive(ATTR_PASSWORD, "basic_auth"): VALIDATE_PASSWORD,
+        vol.Optional(ATTR_VERSION, default=None): vol.Maybe(
+            vol.Coerce(MountCifsVersion)
+        ),
     }
 )
 

tests/api/test_mounts.py

@@ -61,6 +61,7 @@ async def test_api_create_mount(
             "usage": "backup",
             "server": "backup.local",
             "share": "backups",
+            "version": "2.0",
         },
     )
     result = await resp.json()
@@ -71,6 +72,7 @@ async def test_api_create_mount(
 
     assert result["data"]["mounts"] == [
         {
+            "version": "2.0",
             "name": "backup_test",
             "type": "cifs",
             "usage": "backup",
@@ -236,6 +238,7 @@ async def test_api_update_mount(api_client: TestClient, coresys: CoreSys, mount)
 
     assert result["data"]["mounts"] == [
         {
+            "version": None,
             "name": "backup_test",
             "type": "cifs",
             "usage": "backup",
@@ -301,6 +304,7 @@ async def test_api_update_dbus_error_mount_remains(
     result = await resp.json()
     assert result["data"]["mounts"] == [
         {
+            "version": None,
             "name": "backup_test",
             "type": "cifs",
             "usage": "backup",
@@ -340,6 +344,7 @@ async def test_api_update_dbus_error_mount_remains(
     result = await resp.json()
     assert result["data"]["mounts"] == [
         {
+            "version": None,
             "name": "backup_test",
             "type": "cifs",
             "usage": "backup",

tests/mounts/test_manager.py

@@ -486,6 +486,7 @@ async def test_save_data(
         config = json.load(file)
         assert config["mounts"] == [
             {
+                "version": None,
                 "name": "auth_test",
                 "type": "cifs",
                 "usage": "backup",

tests/mounts/test_mount.py

@@ -1,7 +1,9 @@
 """Tests for mounts."""
+from __future__ import annotations
 
 import os
 from pathlib import Path
+from typing import Any
 from unittest.mock import patch
 
 from dbus_fast import DBusError, ErrorType, Variant
@@ -10,7 +12,7 @@ import pytest
 from supervisor.coresys import CoreSys
 from supervisor.dbus.const import UnitActiveState
 from supervisor.exceptions import MountError, MountInvalidError
-from supervisor.mounts.const import MountType, MountUsage
+from supervisor.mounts.const import MountCifsVersion, MountType, MountUsage
 from supervisor.mounts.mount import CIFSMount, Mount, NFSMount
 from supervisor.resolution.const import ContextType, IssueType, SuggestionType
 
@@ -22,11 +24,30 @@ ERROR_FAILURE = DBusError(ErrorType.FAILED, "error")
 ERROR_NO_UNIT = DBusError("org.freedesktop.systemd1.NoSuchUnit", "error")
 
 
+@pytest.mark.parametrize(
+    "additional_data,expected_options",
+    (
+        (
+            {"version": MountCifsVersion.LEGACY_1_0},
+            ["vers=1.0"],
+        ),
+        (
+            {"version": MountCifsVersion.LEGACY_2_0},
+            ["vers=2.0"],
+        ),
+        (
+            {"version": None},
+            [],
+        ),
+    ),
+)
 async def test_cifs_mount(
     coresys: CoreSys,
     all_dbus_services: dict[str, DBusServiceMock],
     tmp_supervisor_data: Path,
     path_extern,
+    additional_data: dict[str, Any],
+    expected_options: list[str],
 ):
     """Test CIFS mount."""
     systemd_service: SystemdService = all_dbus_services["systemd"]
@@ -38,8 +59,10 @@ async def test_cifs_mount(
         "type": "cifs",
         "server": "test.local",
         "share": "camera",
+        "version": None,
         "username": "admin",
         "password": "password",
+        **additional_data,
     }
     mount: CIFSMount = Mount.from_dict(coresys, mount_data)
 
@@ -54,7 +77,10 @@ async def test_cifs_mount(
     assert mount.what == "//test.local/camera"
     assert mount.where == Path("/mnt/data/supervisor/mounts/test")
     assert mount.local_where == tmp_supervisor_data / "mounts" / "test"
-    assert mount.options == ["username=admin", "password=password"]
+    assert mount.options == expected_options + [
+        f"username={mount_data['username']}",
+        f"password={mount_data['password']}",
+    ]
 
     assert not mount.local_where.exists()
     assert mount.to_dict(skip_secrets=False) == mount_data
@@ -73,7 +99,19 @@ async def test_cifs_mount(
             "mnt-data-supervisor-mounts-test.mount",
             "fail",
             [
-                ["Options", Variant("s", "username=admin,password=password")],
+                [
+                    "Options",
+                    Variant(
+                        "s",
+                        ",".join(
+                            expected_options
+                            + [
+                                f"username={mount_data['username']}",
+                                f"password={mount_data['password']}",
+                            ]
+                        ),
+                    ),
+                ],
                 ["Type", Variant("s", "cifs")],
                 ["Description", Variant("s", "Supervisor cifs mount: test")],
                 ["What", Variant("s", "//test.local/camera")],