Check for common permissions issues with scratch area

docs/how-to/edit-live.md

@@ -11,6 +11,8 @@ Blueapi can be configured to install editable Python packages from a chosen dire
 
 scratch:
     root: /path/to/my/scratch/directory
+    # Required GID for the scratch area
+    required_gid: 12345
     repositories:
         # Repository for DLS devices
         - name: dodal
@@ -21,6 +23,9 @@ scratch:
           remote_url: https://github.com/DiamondLightSource/mx-bluesky.git
 ```
 
+Note the `required_gid` field, which is useful for stopping blueapi from locking the files it clones
+to a particular owner.
+
 ## Synchronization
 
 Blueapi will synchronize reality with the configuration if you run

src/blueapi/cli/cli.py

@@ -1,5 +1,7 @@
 import json
 import logging
+import os
+import stat
 import sys
 from functools import wraps
 from pathlib import Path
@@ -39,6 +41,9 @@
 def main(ctx: click.Context, config: Path | None | tuple[Path, ...]) -> None:
     # if no command is supplied, run with the options passed
 
+    # Set umask to DLS standard
+    os.umask(stat.S_IWOTH)
+
     config_loader = ConfigLoader(ApplicationConfig)
     if config is not None:
         configs = (config,) if isinstance(config, Path) else config

src/blueapi/cli/scratch.py

@@ -1,12 +1,14 @@
 import logging
 import os
 import stat
+import textwrap
 from pathlib import Path
 from subprocess import Popen
 
 from git import Repo
 
 from blueapi.config import ScratchConfig
+from blueapi.utils import get_owner_gid, is_sgid_set
 
 _DEFAULT_INSTALL_TIMEOUT: float = 300.0
 
@@ -23,7 +25,7 @@ def setup_scratch(
         install_timeout: Timeout for installing packages
     """
 
-    _validate_directory(config.root)
+    _validate_root_directory(config.root, config.required_gid)
 
     logging.info(f"Setting up scratch area: {config.root}")
 
@@ -74,9 +76,6 @@ def scratch_install(path: Path, timeout: float = _DEFAULT_INSTALL_TIMEOUT) -> No
 
     _validate_directory(path)
 
-    # Set umask to DLS standard
-    os.umask(stat.S_IWOTH)
-
     logging.info(f"Installing {path}")
     process = Popen(
         [
@@ -94,6 +93,37 @@ def scratch_install(path: Path, timeout: float = _DEFAULT_INSTALL_TIMEOUT) -> No
         raise RuntimeError(f"Failed to install {path}: Exit Code: {process.returncode}")
 
 
+def _validate_root_directory(root_path: Path, required_gid: int | None) -> None:
+    _validate_directory(root_path)
+
+    if not is_sgid_set(root_path):
+        raise PermissionError(
+            textwrap.dedent(f"""
+        The scratch area root directory ({root_path}) needs to have the
+        SGID permission bit enabled. This allows blueapi to clone
+        repositories into it while retaining the ability for
+        other users in an approved group to edit/delete them.
+
+        See https://www.redhat.com/en/blog/suid-sgid-sticky-bit for how to
+        enable the SGID bit.
+        """)
+        )
+    elif required_gid is not None and get_owner_gid(root_path) != required_gid:
+        raise PermissionError(
+            textwrap.dedent(f"""
+        The configuration requires that {root_path} be owned by the group with
+        ID {required_gid}.
+        You may be able to find this group's name by running the following
+        in the terminal.
+
+        getent group 1000 | cut -d: -f1
+
+        You can transfer ownership, if you have sufficient permissions, with the chgrp
+        command.
+        """)
+        )
+
+
 def _validate_directory(path: Path) -> None:
     if not path.exists():
         raise KeyError(f"{path}: No such file or directory")

src/blueapi/config.py

@@ -1,3 +1,4 @@
+import textwrap
 from collections.abc import Mapping
 from enum import Enum
 from functools import cached_property
@@ -75,13 +76,34 @@ class RestConfig(BlueapiBaseModel):
 
 
 class ScratchRepository(BlueapiBaseModel):
-    name: str = "example"
-    remote_url: str = "https://github.com/example/example.git"
+    name: str = Field(
+        description="Unique name for this repository in the scratch directory",
+        default="example",
+    )
+    remote_url: str = Field(
+        description="URL to clone from",
+        default="https://github.com/example/example.git",
+    )
 
 
 class ScratchConfig(BlueapiBaseModel):
-    root: Path = Path("/tmp/scratch/blueapi")
-    repositories: list[ScratchRepository] = Field(default_factory=list)
+    root: Path = Field(
+        description="The root directory of the scratch area, all repositories will "
+        "be cloned under this directory.",
+        default=Path("/tmp/scratch/blueapi"),
+    )
+    required_gid: int | None = Field(
+        description=textwrap.dedent("""
+    Required owner GID for the scratch directory. If supplied the setup-scratch
+    command will check the scratch area ownership and raise an error if it is
+    not owned by <GID>.
+    """),
+        default=None,
+    )
+    repositories: list[ScratchRepository] = Field(
+        description="Details of repositories to be cloned and imported into blueapi",
+        default_factory=list,
+    )
 
 
 class OIDCConfig(BlueapiBaseModel):

src/blueapi/utils/__init__.py

@@ -1,5 +1,6 @@
 from .base_model import BlueapiBaseModel, BlueapiModelConfig, BlueapiPlanModelConfig
 from .connect_devices import connect_devices
+from .file_permissions import get_owner_gid, is_sgid_set
 from .invalid_config_error import InvalidConfigError
 from .modules import load_module_all
 from .serialization import serialize
@@ -14,4 +15,6 @@
     "BlueapiPlanModelConfig",
     "InvalidConfigError",
     "connect_devices",
+    "is_sgid_set",
+    "get_owner_gid",
 ]

src/blueapi/utils/file_permissions.py

@@ -0,0 +1,32 @@
+import stat
+from pathlib import Path
+
+
+def is_sgid_set(path: Path) -> bool:
+    """Check if the SGID bit is set so that new files created
+    under a directory owned by a group are owned by that same group.
+
+    See https://www.redhat.com/en/blog/suid-sgid-sticky-bit
+
+    Args:
+        path: Path to the file to check
+
+    Returns:
+        bool: True if the SGID bit is set
+    """
+
+    mask = path.stat().st_mode
+    return bool(mask & stat.S_ISGID)
+
+
+def get_owner_gid(path: Path) -> int:
+    """Get the GID of the owner of a file
+
+    Args:
+        path: Path to the file to check
+
+    Returns:
+        bool: The GID of the file owner
+    """
+
+    return path.stat().st_gid

tests/unit_tests/cli/test_scratch.py

@@ -10,15 +10,25 @@
 
 from blueapi.cli.scratch import ensure_repo, scratch_install, setup_scratch
 from blueapi.config import ScratchConfig, ScratchRepository
+from blueapi.utils import get_owner_gid
 
 
 @pytest.fixture
-def directory_path() -> Generator[Path]:
+def directory_path_no_permissions() -> Generator[Path]:
     temporary_directory = TemporaryDirectory()
     yield Path(temporary_directory.name)
     temporary_directory.cleanup()
 
 
+@pytest.fixture
+def directory_path(directory_path_no_permissions: Path) -> Path:
+    os.chmod(
+        directory_path_no_permissions,
+        os.stat(directory_path_no_permissions).st_mode + stat.S_ISGID,
+    )
+    return directory_path_no_permissions
+
+
 @pytest.fixture
 def file_path(directory_path: Path) -> Generator[Path]:
     file_path = directory_path / str(uuid.uuid4())
@@ -149,6 +159,40 @@ def test_setup_scratch_fails_on_non_directory_root(
         setup_scratch(config)
 
 
+def test_setup_scratch_fails_on_non_sgid_root(
+    directory_path_no_permissions: Path,
+):
+    config = ScratchConfig(root=directory_path_no_permissions, repositories=[])
+    with pytest.raises(PermissionError):
+        setup_scratch(config)
+
+
+def test_setup_scratch_fails_on_wrong_gid(
+    directory_path: Path,
+):
+    config = ScratchConfig(
+        root=directory_path,
+        required_gid=12345,
+        repositories=[],
+    )
+    assert get_owner_gid(directory_path) != 12345
+    with pytest.raises(PermissionError):
+        setup_scratch(config)
+
+
+def test_setup_scratch_succeeds_on_required_gid(
+    directory_path: Path,
+):
+    os.chown(directory_path, uid=12345, gid=12345)
+    config = ScratchConfig(
+        root=directory_path,
+        required_gid=12345,
+        repositories=[],
+    )
+    assert get_owner_gid(directory_path) == 12345
+    setup_scratch(config)
+
+
 @patch("blueapi.cli.scratch.ensure_repo")
 @patch("blueapi.cli.scratch.scratch_install")
 def test_setup_scratch_iterates_repos(

tests/unit_tests/test_cli.py

@@ -60,6 +60,21 @@ def test_main_no_params():
     assert result.stdout == expected
 
 
+@patch("blueapi.service.main.start")
+@patch("blueapi.cli.scratch.setup_scratch")
+@patch("blueapi.cli.cli.os.umask")
+@pytest.mark.parametrize("subcommand", ["serve", "setup-scratch"])
+def test_runs_with_umask_002(
+    mock_umask: Mock,
+    mock_setup_scratch: Mock,
+    mock_start: Mock,
+    runner: CliRunner,
+    subcommand: str,
+):
+    runner.invoke(main, [subcommand])
+    mock_umask.assert_called_once_with(0o002)
+
+
 @patch("requests.request")
 def test_connection_error_caught_by_wrapper_func(
     mock_requests: Mock, runner: CliRunner

tests/unit_tests/test_config.py

@@ -277,6 +277,7 @@ def test_config_yaml_parsed(temp_yaml_config_file):
             },
             "scratch": {
                 "root": "/tmp/scratch/blueapi",
+                "required_gid": None,
                 "repositories": [
                     {
                         "name": "dodal",
@@ -309,6 +310,7 @@ def test_config_yaml_parsed(temp_yaml_config_file):
             },
             "scratch": {
                 "root": "/tmp/scratch/blueapi",
+                "required_gid": None,
                 "repositories": [
                     {
                         "name": "dodal",

tests/unit_tests/utils/test_file_permissions.py

@@ -0,0 +1,65 @@
+import stat
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+
+from blueapi.utils import get_owner_gid, is_sgid_set
+
+
+@pytest.mark.parametrize(
+    "bits",
+    [
+        # Files
+        0o10_0600,  # -rw-------.
+        0o10_0777,  # -rwxrwxrwx.
+        0o10_0000,  # ----------.
+        0o10_0644,  # -rw-r--r--.
+        0o10_0400,  # -r--------.
+        0o10_0666,  # -rw-rw-rw-.
+        0o10_0444,  # -r--r--r--.
+        # Directories
+        0o04_0777,  # drwxrwxrwx.
+        0o04_0000,  # d---------.
+        0o04_0600,  # drw-------.
+    ],
+    ids=lambda p: f"{p:06o} ({stat.filemode(p)})",
+)
+def test_is_sgid_set_should_be_disabled(bits: int):
+    assert not _mocked_is_sgid_set(bits)
+
+
+@pytest.mark.parametrize(
+    "bits",
+    [
+        # Files
+        0o10_2777,  # -rwxrwsrwx.
+        0o10_2000,  # ------S---.
+        0o10_2644,  # -rw-r-Sr--.
+        0o10_2600,  # -rw---S---.
+        0o10_2400,  # -r----S---.
+        0o10_2666,  # -rw-rwSrw-.
+        0o10_2444,  # -r--r-Sr--.
+        # Directories
+        0o04_2777,  # drwxrwsrwx.
+        0o04_2000,  # d-----S---.
+        0o04_2600,  # drw---S---.
+    ],
+    ids=lambda p: f"{p:06o} ({stat.filemode(p)})",
+)
+def test_is_sgid_set_should_be_enabled(bits: int):
+    assert _mocked_is_sgid_set(bits)
+
+
+def _mocked_is_sgid_set(bits: int) -> bool:
+    path = Mock(spec=Path)
+    path.stat().st_mode = bits
+
+    return is_sgid_set(path)
+
+
+def test_get_owner_gid():
+    path = Mock(spec=Path)
+    path.stat().st_gid = 12345
+
+    assert get_owner_gid(path) == 12345