From a5fb4ad3fae079113ca68c4105c837a92e1c0e0d Mon Sep 17 00:00:00 2001 From: Louis Mazin Date: Wed, 14 Oct 2026 18:47:43 +0200 Subject: [PATCH] nice --- backend/drive/models.py | 2 +- backend/drive/services.py | 2 +- backend/drive/storage_layout.py | 93 +++++++++++-- backend/drive/test_trash_isolation.py | 87 ++++++++++++ backend/drive/test_word_preview.py | 22 ++++ backend/drive/trash_cleanup.py | 33 +++-- backend/drive/views_content.py | 13 ++ backend/drive/views_items.py | 41 +++++- backend/drive/views_shares.py | 13 ++ backend/drive/word_preview.py | 124 ++++++++++++++++++ .../drive/dialogs/PreviewDialog.jsx | 1 + frontend/src/pages/AdminUsersPage.jsx | 9 +- frontend/src/pages/DrivePage.jsx | 2 +- frontend/src/pages/PublicSharePage.jsx | 2 + 14 files changed, 409 insertions(+), 35 deletions(-) create mode 100644 backend/drive/test_trash_isolation.py create mode 100644 backend/drive/test_word_preview.py create mode 100644 backend/drive/word_preview.py diff --git a/backend/drive/models.py b/backend/drive/models.py index 511f0f5..8589f15 100644 --- a/backend/drive/models.py +++ b/backend/drive/models.py @@ -82,7 +82,7 @@ class DriveItem(models.Model): indexes = [models.Index(fields=['owner','parent']), models.Index(fields=['space','trashed_at'])] def clean(self): - if not self.name or self.name in {'.', '..'} or any(c in self.name for c in '/\\\x00\r\n'): + if not self.name or self.name in {'.', '..', '.gooddrive-trash'} or any(c in self.name for c in '/\\\x00\r\n'): raise ValidationError('Nom de fichier ou dossier invalide.') if len(self.name) > 255: raise ValidationError('Nom trop long.') diff --git a/backend/drive/services.py b/backend/drive/services.py index afc1dd6..9bcfe61 100644 --- a/backend/drive/services.py +++ b/backend/drive/services.py @@ -60,7 +60,7 @@ def valid_name(name): if not isinstance(name,str) or not name.strip() or len(name)>255: raise ValueError('Nom invalide (1 à 255 caractères).') name=name.strip() - if name in ('.','..') or any(c in name for c in ('/','\\','\x00','\r','\n')): + if name in ('.','..', '.gooddrive-trash') or any(c in name for c in ('/','\\','\x00','\r','\n')): raise ValueError('Ce nom contient des caractères interdits.') return name diff --git a/backend/drive/storage_layout.py b/backend/drive/storage_layout.py index 7fe1a07..fd02b98 100644 --- a/backend/drive/storage_layout.py +++ b/backend/drive/storage_layout.py @@ -48,7 +48,8 @@ def disk_path(location): def assert_available(space, parent, name, exclude=None): - siblings = DriveItem.objects.filter(space=space, parent=parent, name=name) + # Trashed entries have their own isolated physical location. Names can be reused. + siblings = DriveItem.objects.filter(space=space, parent=parent, name=name, trashed_at__isnull=True) if exclude is not None: siblings = siblings.exclude(pk=exclude) if siblings.exists(): @@ -72,8 +73,8 @@ def relocate_item(item, new_parent, new_name): The on-disk move is reversed if the database transaction fails. The caller must ensure authorization and that new_parent is not a descendant of item. """ - old = (file_location(item.space, item.parent, item.name) if item.is_folder - else item.storage_location or f'{item.storage_key[:2]}/{item.storage_key}') + old = item.storage_location or (file_location(item.space, item.parent, item.name) + if item.is_folder else f'{item.storage_key[:2]}/{item.storage_key}') new = file_location(item.space, new_parent, new_name) source = disk_path(old) target = assert_available(item.space, new_parent, new_name, exclude=item.pk) @@ -92,16 +93,8 @@ def relocate_item(item, new_parent, new_name): source.rename(target) moved = True if item.is_folder: - prefix = old.rstrip('/') + '/' - affected = list(DriveItem.objects.filter( - space=item.space, storage_location__startswith=prefix - ).only('pk', 'storage_location')) - for child in affected: - child.storage_location = new + '/' + child.storage_location[len(prefix):] - if affected: - DriveItem.objects.bulk_update(affected, ['storage_location'], batch_size=500) - else: - item.storage_location = new + _update_descendant_locations(item, old, new) + item.storage_location = new item.name = new_name item.parent = new_parent item.full_clean() @@ -112,3 +105,77 @@ def relocate_item(item, new_parent, new_name): if target.exists() and not source.exists(): target.rename(source) raise + + +def _update_descendant_locations(item, old, new): + """Keep physical paths in sync, including legacy folders without a stored path. + + Entries already moved independently to trash keep their own location. + """ + old_prefix = old.rstrip('/') + '/' + new_prefix = new.rstrip('/') + '/' + pending = [item.pk] + while pending: + batch, pending = pending[:300], pending[300:] + children = list(DriveItem.objects.filter(parent_id__in=batch).select_related('parent', 'space')) + pending.extend(child.pk for child in children if child.is_folder) + changed = [] + for child in children: + location = child.storage_location + if not location and child.is_folder: + # Historical folders used only their logical name as a disk path. + location = file_location(child.space, child.parent, child.name) + if location and location.startswith(old_prefix): + child.storage_location = new_prefix + location[len(old_prefix):] + changed.append(child) + if changed: + DriveItem.objects.bulk_update(changed, ['storage_location'], batch_size=300) + + +def trash_location(item): + """Each deleted object gets a separate directory, regardless of display name.""" + return f'.gooddrive-trash/{item.pk}/{item.name}' + + +def move_to_trash(item, *, migrate_existing=False): + """Atomic disk/SQL move. Preserve the original name/parent for restoration.""" + from django.utils import timezone + from .trash_cleanup import has_active_uploads_in_tree + if (item.trashed_at is not None) != migrate_existing: + raise ValueError('État de corbeille inattendu.') + if migrate_existing and item.storage_location == trash_location(item): + return item + if has_active_uploads_in_tree(item): + raise ValueError('Un transfert est encore en cours dans ce dossier.') + old = item.storage_location or (file_location(item.space, item.parent, item.name) + if item.is_folder else f'{item.storage_key[:2]}/{item.storage_key}') + new = trash_location(item) + source, target = disk_path(old), disk_path(new) + if not source.exists() or (item.is_folder != source.is_dir()): + raise FileNotFoundError('Élément absent du stockage, déplacement impossible.') + if target.exists() or target.is_symlink(): + raise NameConflict('La destination de la corbeille est déjà utilisée.') + target.parent.mkdir(parents=True, exist_ok=False) + moved = False + try: + with transaction.atomic(): + source.rename(target) + moved = True + if item.is_folder: + _update_descendant_locations(item, old, new) + item.storage_location = new + if migrate_existing: + item.save(update_fields=['storage_location']) + else: + item.trashed_at = timezone.now() + item.save(update_fields=['storage_location', 'trashed_at', 'updated_at']) + except Exception: + if moved and target.exists() and not source.exists(): + target.rename(source) + raise + finally: + # Leave the isolation folder only while it contains the deleted element. + if not target.exists(): + try: target.parent.rmdir() + except OSError: pass + return item diff --git a/backend/drive/test_trash_isolation.py b/backend/drive/test_trash_isolation.py new file mode 100644 index 0000000..e070265 --- /dev/null +++ b/backend/drive/test_trash_isolation.py @@ -0,0 +1,87 @@ +"""Run: USE_SQLITE=true python manage.py test drive.test_trash_isolation""" +import tempfile +from pathlib import Path +from datetime import timedelta +from django.contrib.auth import get_user_model +from django.core.management import call_command +from django.test import TestCase, override_settings +from django.utils import timezone +from rest_framework.test import APIClient +from .models import DriveItem +from .spaces import ensure_space +from .storage_layout import disk_path + + +class TrashIsolationTests(TestCase): + def setUp(self): + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.root = Path(temp.name) + config = override_settings(DRIVE_STORAGE_ROOT=self.root, FILE_UPLOAD_TEMP_DIR=str(self.root / '.tmp')) + config.enable() + self.addCleanup(config.disable) + self.user = get_user_model().objects.create_superuser('trash-tester', 'trash@example.org', 'password') + self.space = ensure_space(self.user) + self.client = APIClient() + self.client.force_authenticate(self.user) + + def new_file(self, name='rapport.txt', content=b'original'): + from uuid import uuid4 + location = f'{self.space.relative_path}/{name}' if self.space.relative_path else name + path = disk_path(location) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + return DriveItem.objects.create(space=self.space, owner=self.user, name=name, size=len(content), + mime_type='text/plain', storage_key=str(uuid4()), storage_location=location) + + def test_reuse_same_name_multiple_times_and_restore(self): + first = self.new_file(content=b'first') + self.assertEqual(self.client.post(f'/api/drive/items/{first.pk}/trash/').status_code, 200) + first.refresh_from_db() + self.assertEqual(disk_path(first.storage_location).read_bytes(), b'first') + second = self.new_file(content=b'second') + self.assertEqual(self.client.post(f'/api/drive/items/{second.pk}/trash/').status_code, 200) + second.refresh_from_db() + self.assertNotEqual(first.storage_location, second.storage_location) + active = self.new_file(content=b'third') + response = self.client.post(f'/api/drive/items/{first.pk}/restore/') + self.assertEqual(response.status_code, 200, response.data) + self.assertEqual(response.data['name'], 'rapport (restauré 2).txt') + first.refresh_from_db() + self.assertEqual(disk_path(first.storage_location).read_bytes(), b'first') + self.assertEqual(disk_path(second.storage_location).read_bytes(), b'second') + self.assertEqual(disk_path(active.storage_location).read_bytes(), b'third') + self.assertEqual(self.client.delete(f'/api/drive/items/{second.pk}/delete/').status_code, 204) + self.assertTrue(disk_path(active.storage_location).exists()) + + def test_trashed_folder_frees_name_and_restores_contents(self): + folder = self.client.post('/api/drive/folders/', {'name': 'Cours', 'space':self.space.pk}, format='json') + self.assertEqual(folder.status_code, 201, folder.data) + original = DriveItem.objects.get(pk=folder.data['id']) + location = '/'.join(part for part in (self.space.relative_path, 'Cours', 'file.txt') if part) + target = disk_path(location) + target.write_bytes(b'present') + from uuid import uuid4 + child = DriveItem.objects.create(space=self.space, owner=self.user, parent=original, + name='file.txt', storage_key=str(uuid4()), storage_location=location) + self.assertEqual(self.client.post(f'/api/drive/items/{original.pk}/trash/').status_code, 200) + child.refresh_from_db() + self.assertTrue(disk_path(child.storage_location).is_file()) + replacement = self.client.post('/api/drive/folders/', {'name': 'Cours', 'space':self.space.pk}, format='json') + self.assertEqual(replacement.status_code, 201, replacement.data) + restored = self.client.post(f'/api/drive/items/{original.pk}/restore/') + self.assertEqual(restored.status_code, 200, restored.data) + self.assertEqual(restored.data['name'], 'Cours (restauré 2)') + child.refresh_from_db() + self.assertEqual(disk_path(child.storage_location).read_bytes(), b'present') + + def test_migrate_existing_trashed_without_resetting_retention(self): + item = self.new_file() + when = timezone.now() - timedelta(days=4) + DriveItem.objects.filter(pk=item.pk).update(trashed_at=when) + call_command('migrate_trash_storage', '--apply', verbosity=0) + item.refresh_from_db() + self.assertEqual(item.trashed_at, when) + self.assertEqual(disk_path(item.storage_location).read_bytes(), b'original') + self.new_file(content=b'new copy') + call_command('migrate_trash_storage', '--apply', verbosity=0) diff --git a/backend/drive/test_word_preview.py b/backend/drive/test_word_preview.py new file mode 100644 index 0000000..b3086c2 --- /dev/null +++ b/backend/drive/test_word_preview.py @@ -0,0 +1,22 @@ +import tempfile +from pathlib import Path +from zipfile import ZipFile +from django.test import SimpleTestCase +from .word_preview import render_word_preview + + +class WordPreviewTests(SimpleTestCase): + def test_docx_text_escaped_and_tables(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / 'note.docx' + document = b''' + <script>alert(1)</script> + Tableau + ''' + with ZipFile(path, 'w') as archive: + archive.writestr('word/document.xml', document) + html = render_word_preview(path) + self.assertIn('<script>', html) + self.assertNotIn('