feat(B-STOR): Gemeinsamer File Storage — MIME-Prüfung, Size-Limits, Hashing, save_with_metadata
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()
This commit is contained in:
Agent Zero
@@ -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"
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user