From a3a26d1f661992ad1f36f9c7b2bd2d8e4fbd5728 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Thu, 13 Aug 2026 16:32:24 +0200 Subject: [PATCH] =?UTF-8?q?feat(B-STOR):=20Gemeinsamer=20File=20Storage=20?= =?UTF-8?q?=E2=80=94=20MIME-Pr=C3=BCfung,=20Size-Limits,=20Hashing,=20save?= =?UTF-8?q?=5Fwith=5Fmetadata?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B-STOR: storage.py um 5 Funktionen erweitert - validate_mime(): MIME-Erkennung (python-magic/mimetypes) + Allowlist-Prüfung - validate_size(): Dateigrößen-Prüfung (Default 50MB, konfigurierbar) - compute_hash(): SHA256/MD5/SHA1 Hash-Berechnung - save_with_metadata(): save + validate + hash in einem Call - get_file_metadata(): File-Stat ohne Content zu laden - config.py: storage_max_file_size_mb, storage_allowed_mimes Settings - Backward compatible: save/read/delete/exists unverändert B-STOR-TEST: 27 Tests in test_storage.py — alle grün - Path-Traversal, MIME-Validation, Size-Limits, Hashing, save_with_metadata, LocalStorage, Factory B-STOR-MIG: Bereits erledigt — DMS, Mail, Report-Generator, Attachments nutzen bereits get_storage_backend() --- app/config.py | 2 + app/core/storage.py | 237 ++++++++++++++++++++++++++++++++++ tests/test_storage.py | 286 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 525 insertions(+) create mode 100644 tests/test_storage.py diff --git a/app/config.py b/app/config.py index b75a6af..7546b86 100644 --- a/app/config.py +++ b/app/config.py @@ -48,6 +48,8 @@ class Settings(BaseSettings): # Storage storage_path: str = "/data/storage" + storage_max_file_size_mb: int = 50 + storage_allowed_mimes: str = "" # comma-separated, empty = all allowed # SMTP smtp_host: str = "localhost" diff --git a/app/core/storage.py b/app/core/storage.py index 759f087..c54a692 100644 --- a/app/core/storage.py +++ b/app/core/storage.py @@ -15,8 +15,10 @@ Configuration via environment variables: from __future__ import annotations import asyncio +import hashlib import io import logging +import mimetypes import os import tempfile from abc import ABC, abstractmethod @@ -26,6 +28,41 @@ import aiofiles logger = logging.getLogger(__name__) +# Try to import python-magic for content-based MIME detection +try: + import magic # type: ignore + + _HAS_MAGIC = True +except ImportError: + _HAS_MAGIC = False + logger.debug("python-magic not installed, falling back to mimetypes") + + +# Default MIME allowlist — common document, image, and archive types +_DEFAULT_ALLOWED_MIMES: list[str] = [ + # Documents + "text/plain", "text/html", "text/csv", "text/markdown", + "application/pdf", "application/msword", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + "application/vnd.ms-excel", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "application/vnd.ms-powerpoint", + "application/vnd.openxmlformats-officedocument.presentationml.presentation", + "application/vnd.oasis.opendocument.text", + "application/vnd.oasis.opendocument.spreadsheet", + "application/rtf", "application/json", "application/xml", + # Images + "image/jpeg", "image/png", "image/gif", "image/webp", + "image/svg+xml", "image/tiff", "image/bmp", "image/x-icon", + # Archives + "application/zip", "application/x-tar", "application/gzip", + "application/x-7z-compressed", "application/x-rar-compressed", + "application/x-bzip2", + # Other + "application/octet-stream", "message/rfc822", "application/x-yaml", + "text/x-yaml", +] + class StorageBackend(ABC): """Abstract storage backend for file operations.""" @@ -303,3 +340,203 @@ def reset_storage_backend() -> None: """Reset the storage backend singleton (for testing).""" global _storage_backend _storage_backend = None + + +# ─── Validation & Metadata Helpers ─── + + +def validate_mime( + path: str, + data: bytes, + allowed_mimes: list[str] | None = None, +) -> str: + """Detect the MIME type of *data* and validate it against an allowlist. + + Uses ``python-magic`` for content-based detection when available, + falling back to ``mimetypes`` (extension-based) otherwise. + + Parameters + ---------- + path: + Filename or relative path — used for extension-based fallback. + data: + File content bytes used for content-based detection. + allowed_mimes: + Allowlist of MIME types. ``None`` uses the built-in default + allowlist. An empty list means *all* MIME types are allowed. + + Returns + ------- + str + The detected MIME type. + + Raises + ------ + ValueError + If the detected MIME type is not in *allowed_mimes*. + """ + # --- detect MIME type ------------------------------------------------- + if _HAS_MAGIC: + try: + mime_type = magic.from_buffer(data, mime=True) + except Exception: + mime_type, _ = mimetypes.guess_type(path) + mime_type = mime_type or "application/octet-stream" + else: + mime_type, _ = mimetypes.guess_type(path) + mime_type = mime_type or "application/octet-stream" + + # --- validate against allowlist -------------------------------------- + if allowed_mimes is None: + from app.config import get_settings + + config_mimes = get_settings().storage_allowed_mimes + if config_mimes: + allowed = [m.strip() for m in config_mimes.split(",") if m.strip()] + else: + allowed = _DEFAULT_ALLOWED_MIMES + else: + allowed = allowed_mimes + + if allowed and mime_type not in allowed: + raise ValueError( + f"MIME type '{mime_type}' is not allowed. " + f"Allowed types: {', '.join(allowed[:10])}{'...' if len(allowed) > 10 else ''}" + ) + + return mime_type + + +def validate_size(data: bytes, max_size_mb: int | None = None) -> None: + """Validate that *data* does not exceed the configured size limit. + + Parameters + ---------- + data: + File content bytes. + max_size_mb: + Maximum allowed size in megabytes. ``None`` reads the value + from the ``STORAGE_MAX_FILE_SIZE_MB`` environment variable + (default: 50). + + Raises + ------ + ValueError + If ``len(data)`` exceeds ``max_size_mb * 1024 * 1024``. + """ + if max_size_mb is None: + from app.config import get_settings + + max_size_mb = get_settings().storage_max_file_size_mb + + max_bytes = max_size_mb * 1024 * 1024 + if len(data) > max_bytes: + raise ValueError( + f"File size {len(data)} bytes exceeds limit of {max_size_mb} MB ({max_bytes} bytes)" + ) + + +def compute_hash(data: bytes, algorithm: str = "sha256") -> str: + """Compute a cryptographic hash of *data*. + + Parameters + ---------- + data: + Content to hash. + algorithm: + Hash algorithm name (e.g. ``"sha256"``, ``"md5"``, ``"sha1"``). + + Returns + ------- + str + Hexadecimal digest string. + """ + h = hashlib.new(algorithm) + h.update(data) + return h.hexdigest() + + +async def save_with_metadata( + path: str, + data: bytes, + allowed_mimes: list[str] | None = None, + max_size_mb: int | None = None, +) -> dict[str, Any]: + """Save *data* to storage with full validation and metadata extraction. + + Combines :func:`validate_size`, :func:`validate_mime`, + :func:`compute_hash` and :meth:`StorageBackend.save` into a single + call. + + Parameters + ---------- + path: + Relative storage path. + data: + File content bytes. + allowed_mimes: + MIME allowlist — ``None`` uses the default allowlist. + max_size_mb: + Size limit in MB — ``None`` reads from config. + + Returns + ------- + dict + ``{path, mime_type, size, hash, storage_path}`` + """ + validate_size(data, max_size_mb) + mime_type = validate_mime(path, data, allowed_mimes) + file_hash = compute_hash(data) + + backend = get_storage_backend() + storage_path = await backend.save(path, data) + + return { + "path": path, + "mime_type": mime_type, + "size": len(data), + "hash": file_hash, + "storage_path": storage_path, + } + + +def get_file_metadata(path: str) -> dict[str, Any]: + """Read metadata of a stored file without loading its content. + + Works with the *local* storage backend. For S3, use the S3 client + ``stat_object`` API directly. + + Parameters + ---------- + path: + Relative storage path. + + Returns + ------- + dict + ``{size, modified, exists}`` — ``exists`` is ``False`` when the + file is not found, in which case ``size`` and ``modified`` are + ``None``. + """ + backend = get_storage_backend() + if isinstance(backend, LocalStorage): + full_path = backend._full_path(path) + if not os.path.exists(full_path): + return {"size": None, "modified": None, "exists": False} + stat = os.stat(full_path) + return { + "size": stat.st_size, + "modified": stat.st_mtime, + "exists": True, + } + # S3 or other backends — fall back to exists() check + import asyncio as _asyncio + + loop = _asyncio.new_event_loop() + try: + exists = loop.run_until_complete(backend.exists(path)) + if not exists: + return {"size": None, "modified": None, "exists": False} + return {"size": None, "modified": None, "exists": True} + finally: + loop.close() diff --git a/tests/test_storage.py b/tests/test_storage.py new file mode 100644 index 0000000..20810d6 --- /dev/null +++ b/tests/test_storage.py @@ -0,0 +1,286 @@ +"""Tests for app/core/storage.py — validation, hashing, metadata, LocalStorage, factory.""" + +from __future__ import annotations + +import asyncio +import hashlib +import os +import tempfile + +import pytest + +# Set test env BEFORE importing app modules +os.environ.setdefault("SECRET_KEY", "test-secret-key-with-at-least-32-characters-for-testing-only!!") +os.environ.setdefault("ENVIRONMENT", "testing") +os.environ.setdefault("SESSION_COOKIE_SECURE", "false") + +from app.core.storage import ( # noqa: E402 + LocalStorage, + compute_hash, + get_file_metadata, + get_storage_backend, + reset_storage_backend, + save_with_metadata, + validate_mime, + validate_size, +) + + +# ─── Fixtures ─── + + +@pytest.fixture +def tmp_storage(tmp_path, monkeypatch): + """Provide a LocalStorage backed by a temp directory.""" + monkeypatch.setenv("STORAGE_PATH", str(tmp_path)) + monkeypatch.setenv("STORAGE_BACKEND", "local") + reset_storage_backend() + backend = LocalStorage(base_path=str(tmp_path)) + yield backend + reset_storage_backend() + + +# ─── 1. Path-Traversal Tests ─── + + +class TestPathTraversal: + @pytest.mark.asyncio + async def test_save_traversal_relative_parent(self, tmp_storage): + with pytest.raises(ValueError, match="[Pp]ath traversal"): + await tmp_storage.save("../etc/passwd", b"data") + + @pytest.mark.asyncio + async def test_save_traversal_double_parent(self, tmp_storage): + with pytest.raises(ValueError, match="[Pp]ath traversal"): + await tmp_storage.save("../../secret", b"data") + + @pytest.mark.asyncio + async def test_save_normal_path_ok(self, tmp_storage): + result = await tmp_storage.save("normal/path/file.txt", b"hello") + assert result == "normal/path/file.txt" + + +# ─── 2. MIME Tests ─── + + +class TestValidateMime: + def test_text_plain_allowed(self): + mime = validate_mime("file.txt", b"hello", ["text/plain"]) + assert mime == "text/plain" + + def test_exe_rejected(self): + with pytest.raises(ValueError, match="not allowed"): + validate_mime("file.exe", b"MZ\x90\x00", ["text/plain"]) + + def test_pdf_default_allowlist(self): + # PDF magic bytes + mime = validate_mime("file.pdf", b"%PDF-1.4 test", None) + assert mime is not None + + def test_empty_allowlist_all_allowed(self): + mime = validate_mime("file.xyz", b"data", []) + assert mime is not None + + def test_custom_allowlist_reject(self): + with pytest.raises(ValueError, match="not allowed"): + validate_mime("file.txt", b"hello", ["application/pdf"]) + + +# ─── 3. Size-Limit Tests ─── + + +class TestValidateSize: + def test_small_file_ok(self): + validate_size(b"x" * 100, max_size_mb=1) + + def test_oversized_file_raises(self): + with pytest.raises(ValueError, match="exceeds limit"): + validate_size(b"x" * (60 * 1024 * 1024), max_size_mb=50) + + def test_exact_limit_ok(self): + # Exactly at the limit should pass (<=) + validate_size(b"x" * (1024 * 1024), max_size_mb=1) + + def test_default_from_config(self): + # Should use config default (50 MB) — small data passes + validate_size(b"x" * 100) + + +# ─── 4. Hash Tests ─── + + +class TestComputeHash: + def test_sha256_known_value(self): + result = compute_hash(b"hello") + expected = hashlib.sha256(b"hello").hexdigest() + assert result == expected + assert result == "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" + + def test_md5_known_value(self): + result = compute_hash(b"hello", algorithm="md5") + expected = hashlib.md5(b"hello").hexdigest() + assert result == expected + assert result == "5d41402abc4b2a76b9719d911017c592" + + def test_empty_data(self): + result = compute_hash(b"") + assert result == hashlib.sha256(b"").hexdigest() + + def test_sha1(self): + result = compute_hash(b"test", algorithm="sha1") + assert result == hashlib.sha1(b"test").hexdigest() + + +# ─── 5. save_with_metadata Tests ─── + + +class TestSaveWithMetadata: + @pytest.mark.asyncio + async def test_save_returns_full_metadata(self, tmp_storage, monkeypatch): + monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path)) + monkeypatch.setenv("STORAGE_BACKEND", "local") + reset_storage_backend() + # Patch the singleton to use our tmp backend + import app.core.storage as storage_mod + storage_mod._storage_backend = tmp_storage + + data = b"hello world" + result = await save_with_metadata("test/file.txt", data) + + assert result["path"] == "test/file.txt" + assert result["size"] == len(data) + assert result["hash"] == compute_hash(data) + assert "mime_type" in result + assert result["storage_path"] == "test/file.txt" + reset_storage_backend() + + @pytest.mark.asyncio + async def test_save_oversized_raises(self, tmp_storage, monkeypatch): + monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path)) + monkeypatch.setenv("STORAGE_BACKEND", "local") + reset_storage_backend() + import app.core.storage as storage_mod + storage_mod._storage_backend = tmp_storage + + big_data = b"x" * (60 * 1024 * 1024) + with pytest.raises(ValueError, match="exceeds limit"): + await save_with_metadata("big.txt", big_data, max_size_mb=50) + reset_storage_backend() + + @pytest.mark.asyncio + async def test_save_invalid_mime_raises(self, tmp_storage, monkeypatch): + monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path)) + monkeypatch.setenv("STORAGE_BACKEND", "local") + reset_storage_backend() + import app.core.storage as storage_mod + storage_mod._storage_backend = tmp_storage + + with pytest.raises(ValueError, match="not allowed"): + await save_with_metadata("file.xyz", b"data", allowed_mimes=["application/pdf"]) + reset_storage_backend() + + +# ─── 6. get_file_metadata Tests ─── + + +class TestGetFileMetadata: + def test_existing_file(self, tmp_storage, monkeypatch): + monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path)) + monkeypatch.setenv("STORAGE_BACKEND", "local") + reset_storage_backend() + import app.core.storage as storage_mod + storage_mod._storage_backend = tmp_storage + + # Save a file first + asyncio.run(tmp_storage.save("meta/test.txt", b"metadata test")) + + meta = get_file_metadata("meta/test.txt") + assert meta["exists"] is True + assert meta["size"] == len(b"metadata test") + assert meta["modified"] is not None + reset_storage_backend() + + def test_non_existing_file(self, tmp_storage, monkeypatch): + monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path)) + monkeypatch.setenv("STORAGE_BACKEND", "local") + reset_storage_backend() + import app.core.storage as storage_mod + storage_mod._storage_backend = tmp_storage + + meta = get_file_metadata("nonexistent/file.txt") + assert meta["exists"] is False + assert meta["size"] is None + assert meta["modified"] is None + reset_storage_backend() + + +# ─── 7. LocalStorage Tests ─── + + +class TestLocalStorage: + @pytest.mark.asyncio + async def test_save_read_delete_cycle(self, tmp_storage): + data = b"cycle test data" + path = await tmp_storage.save("cycle/test.txt", data) + assert path == "cycle/test.txt" + + read_data = await tmp_storage.read("cycle/test.txt") + assert read_data == data + + deleted = await tmp_storage.delete("cycle/test.txt") + assert deleted is True + + deleted_again = await tmp_storage.delete("cycle/test.txt") + assert deleted_again is False + + @pytest.mark.asyncio + async def test_save_stream_read_cycle(self, tmp_storage): + async def chunk_gen(): + yield b"chunk1-" + yield b"chunk2-" + yield b"chunk3" + + total = await tmp_storage.save_stream("stream/test.bin", chunk_gen()) + assert total == len(b"chunk1-chunk2-chunk3") + + read_data = await tmp_storage.read("stream/test.bin") + assert read_data == b"chunk1-chunk2-chunk3" + + @pytest.mark.asyncio + async def test_exists_check(self, tmp_storage): + assert await tmp_storage.exists("exists/no.txt") is False + await tmp_storage.save("exists/yes.txt", b"yes") + assert await tmp_storage.exists("exists/yes.txt") is True + + @pytest.mark.asyncio + async def test_list_files(self, tmp_storage): + await tmp_storage.save("list/a.txt", b"a") + await tmp_storage.save("list/sub/b.txt", b"b") + files = await tmp_storage.list_files("list") + assert len(files) == 2 + # Paths are relative to base_path + assert any("a.txt" in f for f in files) + assert any("b.txt" in f for f in files) + + +# ─── 8. Backend Factory Tests ─── + + +class TestBackendFactory: + def test_get_storage_backend_local_default(self, monkeypatch): + monkeypatch.setenv("STORAGE_BACKEND", "local") + monkeypatch.setenv("STORAGE_PATH", tempfile.mkdtemp()) + reset_storage_backend() + backend = get_storage_backend() + assert isinstance(backend, LocalStorage) + reset_storage_backend() + + def test_reset_storage_backend(self, monkeypatch): + monkeypatch.setenv("STORAGE_BACKEND", "local") + monkeypatch.setenv("STORAGE_PATH", tempfile.mkdtemp()) + reset_storage_backend() + backend1 = get_storage_backend() + reset_storage_backend() + backend2 = get_storage_backend() + assert backend1 is not backend2 + reset_storage_backend()