Add ownership checks for storage directories

sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md

@@ -3,6 +3,8 @@
 ## 1.0.0b52 (Unreleased)
 
 ### Features Added
+- Add ownership checks for storage directories
+  ([#46725](https://github.com/Azure/azure-sdk-for-python/pull/46725))
 - Add logger name to custom dimensions for Message, Exception and Event telemetry
   ([#46096](https://github.com/Azure/azure-sdk-for-python/pull/46096))
 - Add support for populating SDK version from distro and Microsoft OpenTelemetry distro environment variables

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_storage.py

@@ -66,7 +66,15 @@ def get(self) -> Optional[Tuple[Any, ...]]:
     def put(self, data: List[Any], lease_period: int = 0) -> Union[StorageExportResult, str]:
         try:
             fullpath = self.fullpath + ".tmp"
-            with open(fullpath, "w", encoding="utf-8") as file:
+            # Use O_CREAT | O_EXCL | O_WRONLY to atomically create the file  # cspell:disable-line
+            # and fail if it already exists, preventing race conditions.
+            fd = os.open(fullpath, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)  # cspell:disable-line
+            try:
+                file = os.fdopen(fd, "w", encoding="utf-8")
+            except Exception:
+                os.close(fd)
+                raise
+            with file:
                 for item in data:
                     file.write(json.dumps(item))
                     # The official Python doc: Do not use os.linesep as a line
@@ -255,7 +263,27 @@ def _check_and_set_folder_permissions(self) -> bool:
                     return True
             # Unix
             else:
-                os.chmod(self._path, 0o700)
+                open_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW  # pylint: disable=no-member  # cspell:disable-line
+                dir_fd = os.open(self._path, open_flags)
+                try:
+                    dir_stat = os.fstat(dir_fd)
+                    owner_uid = dir_stat.st_uid
+                    current_uid = os.getuid()  # pylint: disable=no-member
+                    if owner_uid not in (current_uid, 0):
+                        logger.error(
+                            "Storage directory %s is owned by uid %d, not the current user (%d) or admin (uid 0). "
+                            "Refusing to use this directory.",
+                            self._path,
+                            owner_uid,
+                            current_uid,
+                        )
+                        set_local_storage_setup_state_exception(
+                            f"Directory owned by uid {owner_uid}, expected {current_uid} or 0"
+                        )
+                        return False
+                    os.fchmod(dir_fd, 0o700)  # pylint: disable=no-member  # cspell:disable-line
+                finally:
+                    os.close(dir_fd)
                 return True
         except OSError as error:
             if getattr(error, "errno", None) == errno.EROFS:  # cspell:disable-line