306 lines
11 KiB
Python
306 lines
11 KiB
Python
"""
|
|
tests/test_storage_seam_guard.py
|
|
--------------------------------
|
|
MT-22: a static guard against re-opening the storage seam.
|
|
|
|
Every uploaded file must go through ``app/utils/storage.py``. Any code that
|
|
writes bytes to disk directly survives the R2 cutover as a local-only file:
|
|
the tenant's other media moves to the bucket, that one file does not, and on
|
|
the local backend it lands in a directory shared by every tenant with no
|
|
``t<id>/`` prefix.
|
|
|
|
That is exactly how ``tenant_settings._save_logo`` stayed broken — it was
|
|
written before the seam existed and nothing failed when the seam arrived. A
|
|
runtime test cannot catch this class of bug, because a direct write *works*;
|
|
it just works in the wrong place. So this scans the source instead.
|
|
|
|
Adding a new writer is a deliberate act: if one is genuinely needed (a cache,
|
|
a temp file), mark the line ``# storage-seam-exempt: <reason>``.
|
|
"""
|
|
|
|
import ast
|
|
import pathlib
|
|
|
|
import pytest
|
|
|
|
APP_ROOT = pathlib.Path(__file__).resolve().parent.parent / 'app'
|
|
|
|
EXEMPT_MARKER = 'storage-seam-exempt'
|
|
|
|
# storage.py is the seam itself; photo_stamp works on already-materialized
|
|
# temp files handed to it by the seam's materialize_to_dir().
|
|
EXEMPT_FILES = {
|
|
'utils/storage.py',
|
|
'utils/photo_stamp.py',
|
|
}
|
|
|
|
|
|
def _python_files():
|
|
for path in sorted(APP_ROOT.rglob('*.py')):
|
|
rel = path.relative_to(APP_ROOT).as_posix()
|
|
if rel in EXEMPT_FILES:
|
|
continue
|
|
yield rel, path
|
|
|
|
|
|
def _is_exempt_line(source_lines, lineno):
|
|
"""True when the flagged line (or the one above it) carries the marker."""
|
|
for idx in (lineno - 1, lineno - 2):
|
|
if 0 <= idx < len(source_lines) and EXEMPT_MARKER in source_lines[idx]:
|
|
return True
|
|
return False
|
|
|
|
|
|
def test_no_direct_uploaded_file_saves():
|
|
"""No route or util may call ``<werkzeug FileStorage>.save(...)``.
|
|
|
|
Matches any ``*.save(path)`` where the receiver name looks like an upload
|
|
(``file``, ``file_obj``, ``photo``, ``logo``, ``upload``...). ORM
|
|
``db.session.save`` and friends take no positional path, so they don't
|
|
match the shape here.
|
|
"""
|
|
offenders = []
|
|
upload_names = ('file', 'photo', 'image', 'logo', 'upload', 'attachment')
|
|
|
|
for rel, path in _python_files():
|
|
source = path.read_text(encoding='utf-8-sig')
|
|
lines = source.splitlines()
|
|
try:
|
|
tree = ast.parse(source)
|
|
except SyntaxError as exc: # pragma: no cover
|
|
pytest.fail(f'{rel} does not parse: {exc}')
|
|
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
func = node.func
|
|
if not (isinstance(func, ast.Attribute) and func.attr == 'save'):
|
|
continue
|
|
if not node.args: # .save() with no path
|
|
continue
|
|
receiver = func.value
|
|
name = getattr(receiver, 'id', None) or getattr(receiver, 'attr', '')
|
|
if not any(token in name.lower() for token in upload_names):
|
|
continue
|
|
if _is_exempt_line(lines, node.lineno):
|
|
continue
|
|
offenders.append(f'{rel}:{node.lineno} — {name}.save(...)')
|
|
|
|
assert not offenders, (
|
|
'Direct file writes bypass the storage seam (app/utils/storage.py). '
|
|
'Use storage.save(file_obj, subfolder) instead, or mark the line '
|
|
f'"# {EXEMPT_MARKER}: <reason>".\n ' + '\n '.join(offenders)
|
|
)
|
|
|
|
|
|
def test_no_direct_writes_under_upload_folder():
|
|
"""Nothing may build a path from ``UPLOAD_FOLDER`` and write to it.
|
|
|
|
The seam owns that directory. Reading config for display or for a
|
|
migration script is fine; opening a file under it for writing is not.
|
|
"""
|
|
offenders = []
|
|
|
|
for rel, path in _python_files():
|
|
lines = path.read_text(encoding='utf-8-sig').splitlines()
|
|
for i, line in enumerate(lines, start=1):
|
|
if 'UPLOAD_FOLDER' not in line:
|
|
continue
|
|
if _is_exempt_line(lines, i):
|
|
continue
|
|
# Creating the uploads root itself at startup is the seam's own
|
|
# precondition. What matters is code reaching *into* it — that
|
|
# always joins a subfolder or opens a file.
|
|
if 'os.path.join' in line or 'open(' in line:
|
|
offenders.append(f'{rel}:{i} — {line.strip()}')
|
|
|
|
assert not offenders, (
|
|
'Direct writes under UPLOAD_FOLDER bypass the storage seam.\n '
|
|
+ '\n '.join(offenders)
|
|
)
|
|
|
|
|
|
def test_logo_helpers_use_the_seam():
|
|
"""Pin the MT-22 fix itself so it cannot silently regress."""
|
|
source = (APP_ROOT / 'routes' / 'tenant_settings.py').read_text(encoding='utf-8-sig')
|
|
assert 'from app.utils import storage' in source, \
|
|
'tenant_settings.py no longer imports the storage seam'
|
|
assert "storage.save(file_obj, 'logos')" in source, \
|
|
'_save_logo() no longer writes through storage.save()'
|
|
assert 'storage.delete(' in source, \
|
|
'_delete_logo() no longer deletes through the storage seam'
|
|
|
|
|
|
def test_logo_templates_use_media_url():
|
|
"""Logos must render through media_url(), not url_for('static', ...).
|
|
|
|
On an R2-backed tenant a static URL points at a file that isn't there.
|
|
"""
|
|
templates = [
|
|
APP_ROOT / 'templates' / 'layouts' / 'modern.html',
|
|
APP_ROOT / 'templates' / 'layouts' / 'classic.html',
|
|
APP_ROOT / 'templates' / 'tenant_settings' / 'branding.html',
|
|
]
|
|
offenders = []
|
|
for tpl in templates:
|
|
for i, line in enumerate(tpl.read_text(encoding='utf-8-sig').splitlines(), start=1):
|
|
if 'logo_url' in line and 'url_for(' in line and 'static' in line:
|
|
offenders.append(f'{tpl.name}:{i} — {line.strip()}')
|
|
|
|
assert not offenders, (
|
|
'Logo rendered via url_for(static) instead of media_url().\n '
|
|
+ '\n '.join(offenders)
|
|
)
|
|
|
|
|
|
# ── Functional: the logo path now behaves like every other upload ────────────
|
|
#
|
|
# The static guards above prove no direct writer remains. These prove the
|
|
# replacement is correct: tenant-prefixed on the wire, bare key in the DB,
|
|
# and rejected outright when no tenant is bound.
|
|
|
|
import io
|
|
|
|
from flask import g
|
|
|
|
from app.utils import storage
|
|
from app.routes.tenant_settings import _save_logo, _delete_logo
|
|
|
|
|
|
class _FakeTenant:
|
|
def __init__(self, tid):
|
|
self.id = tid
|
|
|
|
|
|
class _FakeUpload:
|
|
"""Stands in for a werkzeug FileStorage well enough for _save_logo()."""
|
|
|
|
def __init__(self, filename='logo.png', data=b'\x89PNG\r\n\x1a\n' + b'x' * 32):
|
|
self.filename = filename
|
|
self.stream = io.BytesIO(data)
|
|
|
|
# _save_logo reads magic bytes off the object itself before handing it on.
|
|
def read(self, n=-1):
|
|
return self.stream.read(n)
|
|
|
|
def seek(self, pos):
|
|
return self.stream.seek(pos)
|
|
|
|
def save(self, dest): # storage-seam-exempt: test double
|
|
raise AssertionError(
|
|
'_save_logo() wrote directly to disk instead of using storage.save()'
|
|
)
|
|
|
|
|
|
class _StubClient:
|
|
def __init__(self):
|
|
self.put_calls = []
|
|
self.deleted = []
|
|
|
|
def put_object(self, Bucket=None, Key=None, Body=None, ContentType=None):
|
|
self.put_calls.append((Key, ContentType))
|
|
return {}
|
|
|
|
def head_object(self, Bucket=None, Key=None):
|
|
raise RuntimeError('404')
|
|
|
|
def delete_object(self, Bucket=None, Key=None):
|
|
self.deleted.append(Key)
|
|
return {}
|
|
|
|
|
|
def _install_s3(app, stub):
|
|
backend = object.__new__(storage.S3Backend)
|
|
backend.bucket = 'test-bucket'
|
|
backend.ttl = 60
|
|
backend.fallback = False
|
|
backend._client = stub
|
|
backend._local = storage.LocalBackend()
|
|
app.extensions['_storage_backend_s3'] = backend
|
|
app.config['STORAGE_BACKEND'] = 's3'
|
|
return backend
|
|
|
|
|
|
@pytest.fixture
|
|
def mt_s3(app):
|
|
"""App context: multi-tenancy on, tenant 7 bound, stubbed S3 backend."""
|
|
stub = _StubClient()
|
|
with app.test_request_context('/'):
|
|
app.config['MULTI_TENANT_ENABLED'] = True
|
|
g.tenant = _FakeTenant(7)
|
|
_install_s3(app, stub)
|
|
try:
|
|
yield stub
|
|
finally:
|
|
app.config['MULTI_TENANT_ENABLED'] = False
|
|
app.config['STORAGE_BACKEND'] = 'local'
|
|
app.extensions.pop('_storage_backend_s3', None)
|
|
|
|
|
|
def test_logo_save_is_tenant_prefixed_on_the_wire(mt_s3):
|
|
key = _save_logo(_FakeUpload('Company Logo.PNG'))
|
|
|
|
# DB value keeps the historic shape — no migration, no data rewrite.
|
|
assert key.startswith('uploads/logos/')
|
|
assert key.endswith('.png')
|
|
assert not key.startswith('t7/')
|
|
# The object itself is scoped to the tenant.
|
|
assert mt_s3.put_calls == [(f't7/{key}', 'image/png')]
|
|
|
|
|
|
def test_logo_delete_targets_the_prefixed_object(mt_s3):
|
|
_delete_logo('uploads/logos/abc123.png')
|
|
assert 't7/uploads/logos/abc123.png' in mt_s3.deleted
|
|
|
|
|
|
def test_logo_delete_ignores_empty_value(mt_s3):
|
|
_delete_logo(None)
|
|
_delete_logo('')
|
|
assert mt_s3.deleted == []
|
|
|
|
|
|
def test_logo_delete_cannot_escape_the_logos_folder(mt_s3):
|
|
"""A tampered column value must not reach another subfolder's objects.
|
|
|
|
S3Backend.delete() probes both the tenant-prefixed key and the legacy
|
|
unprefixed one (its pre-existing transition behaviour, shared by all
|
|
media), so assert containment in uploads/logos/ rather than the prefix.
|
|
"""
|
|
_delete_logo('uploads/issue_photos/../../secret.jpg')
|
|
assert mt_s3.deleted, 'expected a delete attempt'
|
|
for key in mt_s3.deleted:
|
|
assert 'issue_photos' not in key
|
|
assert key.endswith('uploads/logos/secret.jpg')
|
|
|
|
|
|
def test_logo_save_refuses_with_no_tenant_bound(app):
|
|
"""Same loud failure as any other upload — never an unprefixed key."""
|
|
stub = _StubClient()
|
|
with app.test_request_context('/'):
|
|
app.config['MULTI_TENANT_ENABLED'] = True
|
|
g.tenant = None
|
|
_install_s3(app, stub)
|
|
try:
|
|
with pytest.raises(RuntimeError):
|
|
_save_logo(_FakeUpload())
|
|
finally:
|
|
app.config['MULTI_TENANT_ENABLED'] = False
|
|
app.config['STORAGE_BACKEND'] = 'local'
|
|
app.extensions.pop('_storage_backend_s3', None)
|
|
assert stub.put_calls == []
|
|
|
|
|
|
def test_logo_validation_still_rejects_non_images(mt_s3):
|
|
"""Validation stays in the caller — the seam only moves bytes."""
|
|
assert _save_logo(_FakeUpload('payload.exe', b'MZ\x90\x00')) is None
|
|
assert _save_logo(_FakeUpload('fake.png', b'not-an-image')) is None
|
|
assert _save_logo(None) is None
|
|
assert mt_s3.put_calls == []
|
|
|
|
|
|
def test_svg_logo_gets_a_renderable_content_type(mt_s3):
|
|
"""SVG skips the magic-byte check; it must still upload as an image type."""
|
|
key = _save_logo(_FakeUpload('brand.svg', b'<svg xmlns="..."></svg>'))
|
|
assert key.endswith('.svg')
|
|
assert mt_s3.put_calls == [(f't7/{key}', 'image/svg+xml')]
|