feat(storage): add local storage framework

backend/hermes/app/storage/local.py

@@ -0,0 +1,141 @@
+from pathlib import Path
+import hashlib
+
+from app.storage.base import StorageProvider
+from app.storage.exceptions import StorageFileNotFoundError, StorageValidationError
+from app.storage.schemas import FileMetadata, StoredFileMetadata
+
+
+class LocalDiskStorageProvider(StorageProvider):
+    def __init__(
+        self,
+        *,
+        base_path: str,
+        legacy_base_paths: list[str] | None = None,
+    ) -> None:
+        self.base_path = Path(base_path).expanduser().resolve()
+        self.legacy_base_paths = [
+            Path(path).expanduser().resolve()
+            for path in legacy_base_paths or []
+            if path
+        ]
+        self._ensure_layout()
+
+    def save_file(
+        self,
+        *,
+        namespace: str,
+        content: bytes,
+        original_filename: str,
+        stored_filename: str,
+        mime_type: str,
+    ) -> StoredFileMetadata:
+        if Path(stored_filename).name != stored_filename or ".." in Path(stored_filename).parts:
+            raise StorageValidationError("Ungültiger gespeicherter Dateiname")
+        namespace_path = self._safe_namespace_path(namespace)
+        namespace_path.mkdir(parents=True, exist_ok=True)
+        absolute_path = self._safe_child(namespace_path / stored_filename)
+        absolute_path.write_bytes(content)
+        relative_path = absolute_path.relative_to(self.base_path).as_posix()
+        checksum = self._checksum(absolute_path)
+        return StoredFileMetadata(
+            storage_key=relative_path,
+            original_filename=original_filename,
+            stored_filename=stored_filename,
+            relative_path=relative_path,
+            absolute_path=absolute_path,
+            mime_type=mime_type,
+            size=len(content),
+            checksum_sha256=checksum,
+        )
+
+    def open_file(self, storage_key: str) -> Path:
+        path = self._resolve_storage_key(storage_key)
+        if not path.exists() or not path.is_file():
+            raise StorageFileNotFoundError("Datei nicht gefunden")
+        return path
+
+    def delete_file(self, storage_key: str) -> None:
+        path = self._resolve_storage_key(storage_key)
+        if not path.exists():
+            return
+        if not path.is_file():
+            raise StorageValidationError("Storage-Key verweist nicht auf eine Datei")
+        path.unlink()
+
+    def file_exists(self, storage_key: str) -> bool:
+        try:
+            path = self._resolve_storage_key(storage_key)
+        except StorageValidationError:
+            return False
+        return path.exists() and path.is_file()
+
+    def get_file_metadata(self, storage_key: str) -> FileMetadata:
+        path = self.open_file(storage_key)
+        return FileMetadata(
+            storage_key=storage_key,
+            relative_path=self._relative_storage_key(path),
+            absolute_path=path,
+            size=path.stat().st_size,
+            checksum_sha256=self._checksum(path),
+        )
+
+    def _ensure_layout(self) -> None:
+        for relative_path in [
+            "knowledge/documents",
+            "knowledge/thumbnails",
+            "customers",
+            "projects",
+            "tickets",
+            "imports",
+            "temp",
+        ]:
+            (self.base_path / relative_path).mkdir(parents=True, exist_ok=True)
+
+    def _safe_namespace_path(self, namespace: str) -> Path:
+        if namespace.startswith("/") or ".." in Path(namespace).parts:
+            raise StorageValidationError("Ungültiger Storage-Namespace")
+        return self._safe_child(self.base_path / namespace)
+
+    def _safe_child(self, path: Path) -> Path:
+        resolved = path.resolve()
+        if self.base_path != resolved and self.base_path not in resolved.parents:
+            raise StorageValidationError("Ungültiger Storage-Pfad")
+        return resolved
+
+    def _resolve_storage_key(self, storage_key: str) -> Path:
+        raw_path = Path(storage_key)
+        candidate_paths: list[Path] = []
+
+        if raw_path.is_absolute():
+            candidate_paths.append(raw_path)
+        else:
+            candidate_paths.append(self.base_path / raw_path)
+            candidate_paths.extend(legacy_base / raw_path for legacy_base in self.legacy_base_paths)
+
+        for candidate in candidate_paths:
+            resolved = candidate.expanduser().resolve()
+            if self._is_allowed_path(resolved):
+                return resolved
+
+        raise StorageValidationError("Ungültiger Storage-Key")
+
+    def _is_allowed_path(self, path: Path) -> bool:
+        roots = [self.base_path, *self.legacy_base_paths]
+        return any(root == path or root in path.parents for root in roots)
+
+    def _relative_storage_key(self, path: Path) -> str:
+        for root in [self.base_path, *self.legacy_base_paths]:
+            try:
+                return path.relative_to(root).as_posix()
+            except ValueError:
+                continue
+        return path.name
+
+    @staticmethod
+    def _checksum(path: Path) -> str:
+        digest = hashlib.sha256()
+        with path.open("rb") as handle:
+            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+                digest.update(chunk)
+        return digest.hexdigest()