diff --git a/backend/drive/admin.py b/backend/drive/admin.py index 7ef82fc..c261fb9 100644 --- a/backend/drive/admin.py +++ b/backend/drive/admin.py @@ -10,6 +10,12 @@ class SpaceAdmin(admin.ModelAdmin): search_fields = ('display_name', 'user__username') readonly_fields = ('directory_name', 'relative_path') + def get_readonly_fields(self, request, obj=None): + base = super().get_readonly_fields(request, obj) + if obj and obj.private_owner_id: + return tuple(base) + ('quota_bytes', 'parent', 'private_owner') + return base + def get_queryset(self, request): # Private spaces from other users never appear, even for a superuser. return super().get_queryset(request).filter( diff --git a/backend/drive/migrations/0013_private_space_quota.py b/backend/drive/migrations/0013_private_space_quota.py new file mode 100644 index 0000000..e047e82 --- /dev/null +++ b/backend/drive/migrations/0013_private_space_quota.py @@ -0,0 +1,15 @@ +"""Set the fixed quota on existing private spaces; filesystem relocation is manual.""" +from django.db import migrations + + +QUOTA = 50 * 1024 ** 3 + + +def set_private_quotas(apps, schema_editor): + Space = apps.get_model('drive', 'DriveSpace') + Space.objects.filter(private_owner__isnull=False).update(quota_bytes=QUOTA) + + +class Migration(migrations.Migration): + dependencies = [('drive', '0012_merge_20261014_0129')] + operations = [migrations.RunPython(set_private_quotas, migrations.RunPython.noop)] diff --git a/backend/drive/quotas.py b/backend/drive/quotas.py index e87aa26..c5aff4b 100644 --- a/backend/drive/quotas.py +++ b/backend/drive/quotas.py @@ -9,6 +9,7 @@ from django.db.models import Q, Sum from .models import DriveItem, DriveSpace, UploadSession MAX_QUOTA_BYTES = 9_007_199_254_740_991 # exact integer in JS too +PRIVATE_SPACE_QUOTA_BYTES = 50 * 1024 ** 3 # 50 Gio per private account (shown as 50 Go) class QuotaExceeded(Exception): @@ -47,9 +48,10 @@ def ancestry(space): def subtree_space_ids(space): """Use the stable DB path rather than a filesystem traversal.""" - # The logical root has an empty relative_path and contains every space. + # A shared root contains only shared spaces: private spaces have their + # own quota and must never consume/reserve shared root capacity. if space.parent_id is None and space.relative_path == '': - return DriveSpace.objects.all().values_list('pk', flat=True) + return DriveSpace.objects.filter(private_owner__isnull=True).values_list('pk', flat=True) return DriveSpace.objects.filter( Q(relative_path=space.relative_path) | Q(relative_path__startswith=space.relative_path + '/') ).values_list('pk', flat=True) diff --git a/backend/drive/spaces.py b/backend/drive/spaces.py index 416921d..42aa2d9 100644 --- a/backend/drive/spaces.py +++ b/backend/drive/spaces.py @@ -5,7 +5,8 @@ from pathlib import Path from django.conf import settings from django.db import transaction from django.http import Http404 -from .models import DriveSpace, SpaceGrant +from .models import DriveSpace, SpaceGrant, DriveItem +from .quotas import PRIVATE_SPACE_QUOTA_BYTES def is_admin(user): @@ -33,13 +34,26 @@ def ensure_space(user): def _make_numbered_space(parent, **fields): - if len(parent.relative_path) + 1 + 24 > 700: + # Private directories live in a separate physical and quota hierarchy. + private = fields.get('private_owner') is not None + if private != (parent is None): + raise ValueError('Hiérarchie de stockage privée invalide.') + base = 'privates' if private else parent.relative_path + if len(base) + 1 + 24 > 700: raise ValueError('Hiérarchie trop profonde.') + root = Path(settings.DRIVE_STORAGE_ROOT).resolve() + directory = root / base + if directory.is_symlink() or not directory.resolve().is_relative_to(root): + raise ValueError('Chemin de stockage privé invalide.') + directory.mkdir(mode=0o700, parents=True, exist_ok=True) + if private: + main_root = DriveSpace.objects.filter(parent__isnull=True, private_owner__isnull=True, relative_path='').first() + if main_root and DriveItem.objects.filter(space=main_root, parent__isnull=True, name='privates').exists(): + raise ValueError('Le nom privates est déjà utilisé comme dossier à la racine du Drive.') + fields['quota_bytes'] = PRIVATE_SPACE_QUOTA_BYTES with transaction.atomic(): obj = DriveSpace.objects.create(parent=parent, relative_path=f'pending/{secrets.token_hex(10)}', **fields) number = obj.pk - 1 - directory = Path(settings.DRIVE_STORAGE_ROOT) / parent.relative_path - directory.mkdir(mode=0o700, parents=True, exist_ok=True) # With real file names, a user may already have created "0001" as a # folder/file. Do NOT reuse it for another person's space. for _ in range(100000): @@ -55,7 +69,7 @@ def _make_numbered_space(parent, **fields): continue try: obj.directory_name = candidate - obj.relative_path = '/'.join(part for part in (parent.relative_path, candidate) if part) + obj.relative_path = '/'.join(part for part in (base, candidate) if part) obj.save(update_fields=['directory_name', 'relative_path']) except Exception: target.rmdir() @@ -81,8 +95,8 @@ def ensure_private_space(user): obj = DriveSpace.objects.select_for_update().filter(private_owner=user).first() if obj: return obj - shared = ensure_space(user) - return _make_numbered_space(shared, private_owner=user, display_name='Mes fichiers') + ensure_space(user) # Account must still have its assigned/shared space. + return _make_numbered_space(None, private_owner=user, display_name='Mes fichiers') def my_space(user): diff --git a/backend/drive/test_private_layout.py b/backend/drive/test_private_layout.py new file mode 100644 index 0000000..8941659 --- /dev/null +++ b/backend/drive/test_private_layout.py @@ -0,0 +1,148 @@ +"""Focused regressions for separate private storage and its 50 Go quota. + +Run with: USE_SQLITE=true python manage.py test drive.test_private_layout +""" +import tempfile +import uuid +from pathlib import Path + +from django.contrib.auth import get_user_model +from django.core.files.uploadedfile import SimpleUploadedFile +from django.core.management import call_command +from django.core.management.base import CommandError +from django.db import transaction +from django.test import TestCase, override_settings +from rest_framework.test import APIClient + +from .models import DriveItem, DriveSpace +from .quotas import PRIVATE_SPACE_QUOTA_BYTES, QuotaExceeded, assert_space_capacity, locked_quota_tree, occupancy, quota_state +from .spaces import ensure_private_space, ensure_space, create_space, storage_path + + +class PrivateStorageTests(TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.rootdir = Path(self.temp.name) + overrides = override_settings( + DRIVE_STORAGE_ROOT=self.rootdir, + FILE_UPLOAD_TEMP_DIR=str(self.rootdir / '.uploads'), + ) + overrides.enable() + self.addCleanup(overrides.disable) + User = get_user_model() + self.admin = User.objects.create_superuser('root-admin', 'admin@example.org', 'TestPassword123!') + self.shared = ensure_space(self.admin) + self.private = ensure_private_space(self.admin) + self.client = APIClient() + self.client.force_authenticate(user=self.admin) + + def test_private_directories_are_separate_from_shared_spaces(self): + self.assertIsNone(self.private.parent_id) + self.assertEqual(self.private.relative_path, f'privates/{self.private.directory_name}') + self.assertEqual(self.private.quota_bytes, PRIVATE_SPACE_QUOTA_BYTES) + self.assertTrue((self.rootdir / self.private.relative_path).is_dir()) + person = get_user_model().objects.create_user('member', password='TestPassword123!') + shared = create_space(person, self.shared, 'Compte membre') + private = ensure_private_space(person) + self.assertTrue(shared.relative_path.startswith(f'{shared.directory_name}')) + self.assertEqual(private.relative_path, f'privates/{private.directory_name}') + self.assertIsNone(private.parent_id) + self.assertNotEqual(shared.directory_name, private.directory_name) + + def test_private_quota_does_not_consume_shared_parent(self): + self.shared.quota_bytes = 1 + self.shared.save(update_fields=['quota_bytes']) + created = self.client.post('/api/drive/uploads/', { + 'files': [SimpleUploadedFile('photo.jpg', b'abcdef')] + }, format='multipart') + self.assertEqual(created.status_code, 201, created.data) + node = DriveItem.objects.get(pk=created.data['items'][0]['id']) + self.assertEqual(node.storage_location, f'{self.private.relative_path}/photo.jpg') + self.assertEqual(storage_path(node).read_bytes(), b'abcdef') + self.assertEqual(occupancy(self.shared)[0], 0) + self.assertEqual(quota_state(self.private)['quota_bytes'], PRIVATE_SPACE_QUOTA_BYTES) + self.assertEqual(quota_state(self.private)['used_bytes'], 6) + rejected = self.client.patch(f'/api/drive/spaces/{self.private.pk}/quota/', { + 'quota_bytes': None + }, format='json') + self.assertEqual(rejected.status_code, 403) + + def test_private_50_go_upload_limit_is_enforced(self): + DriveItem.objects.create( + owner=self.admin, space=self.private, name='large.bin', + storage_key=str(uuid.uuid4()), + storage_location=f'{self.private.relative_path}/large.bin', + size=PRIVATE_SPACE_QUOTA_BYTES - 1, + ) + with transaction.atomic(): + with locked_quota_tree(self.private): + with self.assertRaises(QuotaExceeded): + assert_space_capacity(self.private, 2) + assert_space_capacity(self.private, 1) + + def simulate_legacy_private(self): + """Relocate the empty modern directory to the historical nested location.""" + new = self.rootdir / self.private.relative_path + new.rmdir() + old_name = self.private.directory_name + old = f'{old_name}' + src = self.rootdir / old + (src / 'Documents').mkdir(parents=True) + (src / 'Documents' / 'facture.pdf').write_bytes(b'PDF CONTENT') + self.private.parent = self.shared + self.private.relative_path = old + self.private.quota_bytes = None + self.private.save(update_fields=['parent', 'relative_path', 'quota_bytes']) + folder = DriveItem.objects.create( + owner=self.admin, space=self.private, name='Documents', is_folder=True, + ) + item = DriveItem.objects.create( + owner=self.admin, space=self.private, parent=folder, name='facture.pdf', + storage_key=str(uuid.uuid4()), storage_location=f'{old}/Documents/facture.pdf', + size=11, + ) + return old, item + + def test_migrate_private_storage_dry_run_and_apply(self): + old, item = self.simulate_legacy_private() + call_command('move_private_storage') + self.private.refresh_from_db() + self.assertEqual(self.private.relative_path, old) + self.assertTrue((self.rootdir / old / 'Documents' / 'facture.pdf').exists()) + call_command('move_private_storage', '--apply') + self.private.refresh_from_db() + item.refresh_from_db() + self.assertEqual(self.private.relative_path, f'privates/{self.private.directory_name}') + self.assertIsNone(self.private.parent_id) + self.assertEqual(self.private.quota_bytes, PRIVATE_SPACE_QUOTA_BYTES) + self.assertEqual(item.storage_location, f'{self.private.relative_path}/Documents/facture.pdf') + self.assertEqual(storage_path(item).read_bytes(), b'PDF CONTENT') + self.assertFalse((self.rootdir / old).exists()) + self.assertFalse((self.rootdir / '.gooddrive-move-privates.json').exists()) + call_command('move_private_storage', '--apply') # Idempotent. + + def test_interrupted_migration_can_resume_after_directory_move(self): + from .management.commands.move_private_storage import Command, write_journal + old, item = self.simulate_legacy_private() + plan = Command().plan() + journal = self.rootdir / '.gooddrive-move-privates.json' + write_journal(journal, plan) + new = self.rootdir / 'privates' / self.private.directory_name + (self.rootdir / old).rename(new) # Simulate a crash before SQL update. + call_command('move_private_storage', '--apply') + item.refresh_from_db() + self.assertEqual(storage_path(item).read_bytes(), b'PDF CONTENT') + self.assertFalse(journal.exists()) + + def test_migration_never_overwrites_existing_destination(self): + old, item = self.simulate_legacy_private() + target = self.rootdir / 'privates' / self.private.directory_name + target.mkdir(parents=True) + (target / 'unrelated.txt').write_text('KEEP ME') + with self.assertRaises(CommandError): + call_command('move_private_storage', '--apply') + self.assertEqual((target / 'unrelated.txt').read_text(), 'KEEP ME') + self.assertTrue((self.rootdir / old / 'Documents' / 'facture.pdf').exists()) + item.refresh_from_db() + self.assertEqual(item.storage_location, f'{old}/Documents/facture.pdf') diff --git a/backend/drive/views_spaces.py b/backend/drive/views_spaces.py index 807f623..2e3f7f1 100644 --- a/backend/drive/views_spaces.py +++ b/backend/drive/views_spaces.py @@ -86,6 +86,8 @@ def members(request): def _update_quota(space, requested): """Must run under transaction.atomic and after locking all ancestors.""" + if space.private_owner_id: + raise ValueError('Le quota des espaces privés est fixé à 50 Go.') with locked_quota_tree(space): quota = parse_quota(requested) assert_updated_quota(space, quota) @@ -126,8 +128,8 @@ def update_space_quota(request, space_id): with transaction.atomic(): try:space=DriveSpace.objects.select_related('parent','user').get(pk=space_id) except DriveSpace.DoesNotExist:return error('Espace introuvable.',404) - if space.private_owner_id and space.private_owner_id != request.user.pk: - return error('Cet espace personnel est privé.',403) + if space.private_owner_id: + return error('Le quota des espaces privés est fixé à 50 Go.',403) if not (is_admin(request.user) or (space.parent_id and not space.private_owner_id and manager(request.user,space.parent))): return error('Modification du quota interdite.',403) if 'quota_bytes' not in request.data:return error('quota_bytes est obligatoire.')