155 lines
4.8 KiB
Python
155 lines
4.8 KiB
Python
"""
|
|
tests/test_storage_backend.py
|
|
-----------------------------
|
|
Guards the one thing that makes S3 storage safe in a multi-tenant deployment:
|
|
the ``t<tenant_id>/`` object-key prefix (app/utils/storage.py).
|
|
|
|
Two failure modes are pinned here, both of which would be silent in production:
|
|
|
|
1. A prefix regression makes tenant A's presigned URL point at tenant B's
|
|
object — a cross-tenant data leak with no error anywhere.
|
|
2. An upload reaching S3Backend.save() with no tenant bound writes an
|
|
unprefixed key. Nothing would break at write time; the leak appears later
|
|
when a second tenant's DB happens to hold the same key string.
|
|
|
|
S3Backend is constructed with ``object.__new__`` and hand-wired attributes on
|
|
purpose: its ``__init__`` builds a real boto3 client, and these tests are about
|
|
key arithmetic, not the network. A stub client records the calls.
|
|
"""
|
|
|
|
import pytest
|
|
from flask import g
|
|
|
|
from app.utils import storage
|
|
|
|
|
|
class _StubClient:
|
|
"""Minimal S3 client double — records calls, answers HEAD from a key set."""
|
|
|
|
def __init__(self, existing=()):
|
|
self.existing = set(existing)
|
|
self.put_calls = []
|
|
self.presigned = []
|
|
self.deleted = []
|
|
|
|
def put_object(self, Bucket=None, Key=None, Body=None, ContentType=None):
|
|
self.put_calls.append(Key)
|
|
self.existing.add(Key)
|
|
return {}
|
|
|
|
def head_object(self, Bucket=None, Key=None):
|
|
if Key in self.existing:
|
|
return {'ETag': '"x"', 'ContentLength': 1}
|
|
raise RuntimeError('404')
|
|
|
|
def get_object(self, Bucket=None, Key=None):
|
|
raise RuntimeError('404')
|
|
|
|
def generate_presigned_url(self, _op, Params=None, ExpiresIn=None):
|
|
key = (Params or {}).get('Key')
|
|
self.presigned.append(key)
|
|
return f'https://r2.example/{key}'
|
|
|
|
def delete_object(self, Bucket=None, Key=None):
|
|
self.deleted.append(Key)
|
|
return {}
|
|
|
|
|
|
class _FakeTenant:
|
|
def __init__(self, tid):
|
|
self.id = tid
|
|
|
|
|
|
class _FakeUpload:
|
|
"""Stands in for a werkzeug FileStorage well enough for save()."""
|
|
|
|
def __init__(self, filename='photo.jpg', data=b'\xff\xd8\xffbytes'):
|
|
import io
|
|
self.filename = filename
|
|
self.stream = io.BytesIO(data)
|
|
|
|
|
|
def _make_s3_backend(stub):
|
|
backend = object.__new__(storage.S3Backend)
|
|
backend.bucket = 'test-bucket'
|
|
backend.ttl = 60
|
|
backend.fallback = False
|
|
backend._client = stub
|
|
backend._local = storage.LocalBackend()
|
|
return backend
|
|
|
|
|
|
@pytest.fixture
|
|
def mt_ctx(app):
|
|
"""App context with multi-tenancy on and tenant 7 bound."""
|
|
with app.test_request_context('/'):
|
|
app.config['MULTI_TENANT_ENABLED'] = True
|
|
g.tenant = _FakeTenant(7)
|
|
try:
|
|
yield app
|
|
finally:
|
|
app.config['MULTI_TENANT_ENABLED'] = False
|
|
|
|
|
|
def test_tenant_key_prefix_reflects_bound_tenant(mt_ctx):
|
|
assert storage.tenant_key_prefix() == 't7/'
|
|
|
|
|
|
def test_tenant_key_prefix_empty_when_mt_disabled(app):
|
|
with app.test_request_context('/'):
|
|
app.config['MULTI_TENANT_ENABLED'] = False
|
|
g.tenant = _FakeTenant(7)
|
|
assert storage.tenant_key_prefix() == ''
|
|
|
|
|
|
def test_save_writes_prefixed_object_but_returns_bare_key(mt_ctx):
|
|
stub = backend_stub = _StubClient()
|
|
backend = _make_s3_backend(backend_stub)
|
|
|
|
key = backend.save(_FakeUpload('evidence.JPG'), 'issue_photos')
|
|
|
|
# DB value stays unprefixed and portable.
|
|
assert key.startswith('uploads/issue_photos/')
|
|
assert key.endswith('.jpg')
|
|
assert not key.startswith('t7/')
|
|
# The object on the wire is tenant-scoped.
|
|
assert stub.put_calls == [f't7/{key}']
|
|
|
|
|
|
def test_save_refuses_when_no_tenant_bound(app):
|
|
"""An upload path outside tenant resolution must fail loudly, not silently
|
|
write a key another tenant could later collide with."""
|
|
with app.test_request_context('/'):
|
|
app.config['MULTI_TENANT_ENABLED'] = True
|
|
g.tenant = None
|
|
stub = _StubClient()
|
|
backend = _make_s3_backend(stub)
|
|
try:
|
|
with pytest.raises(RuntimeError):
|
|
backend.save(_FakeUpload(), 'issue_photos')
|
|
finally:
|
|
app.config['MULTI_TENANT_ENABLED'] = False
|
|
assert stub.put_calls == []
|
|
|
|
|
|
def test_media_url_presigns_the_prefixed_key(mt_ctx):
|
|
stub = _StubClient()
|
|
backend = _make_s3_backend(stub)
|
|
|
|
url = backend.media_url('uploads/issue_photos/abc.jpg')
|
|
|
|
assert stub.presigned == ['t7/uploads/issue_photos/abc.jpg']
|
|
assert url.endswith('t7/uploads/issue_photos/abc.jpg')
|
|
|
|
|
|
def test_exists_and_delete_cover_legacy_unprefixed_key(mt_ctx):
|
|
"""Objects written before the prefix existed must stay reachable."""
|
|
key = 'uploads/issue_photos/legacy.jpg'
|
|
stub = _StubClient(existing={key}) # only the unprefixed object
|
|
backend = _make_s3_backend(stub)
|
|
|
|
assert backend.exists(key) is True
|
|
|
|
backend.delete(key)
|
|
assert stub.deleted == [f't7/{key}', key]
|