"""Private files stored separately but counted against the assigned-space quota.""" import hashlib import secrets from pathlib import Path from django.conf import settings from django.db import transaction from django.http import Http404 from .models import DriveSpace, SpaceGrant, DriveItem from .quotas import PRIVATE_SPACE_QUOTA_BYTES def is_admin(user): return user.is_authenticated and (user.is_superuser or user.is_staff) def ensure_space(user): """Return an owned space, or the existing space granted to an invitee.""" obj = DriveSpace.objects.filter(user=user, private_owner__isnull=True).first() if obj: (Path(settings.DRIVE_STORAGE_ROOT) / obj.relative_path).mkdir(mode=0o700, parents=True, exist_ok=True) return obj # Invitation-only users access an existing space through a SpaceGrant. # They must never create another numbered shared directory implicitly. grant = SpaceGrant.objects.filter(user=user).select_related('space').order_by('created_at','pk').first() if grant: return grant.space with transaction.atomic(): obj = DriveSpace.objects.select_for_update().filter(user=user, private_owner__isnull=True).first() if obj: return obj root = DriveSpace.objects.filter(parent__isnull=True, private_owner__isnull=True).order_by('id').first() if root is None: if not is_admin(user): raise PermissionError('Un administrateur doit initialiser le Drive.') obj = DriveSpace.objects.create(user=user, relative_path='', display_name='Racine') else: obj = create_space(user, root, user.username) return obj def _make_numbered_space(parent, **fields): # Private directories are physically separate, but logically child spaces # of the account's assigned space so their *actual bytes* are included. private = fields.get('private_owner') is not None if parent is None or parent.private_owner_id: raise ValueError('Un espace doit avoir un parent partagé.') if private and parent.user_id != fields['private_owner'].pk and not SpaceGrant.objects.filter( user=fields['private_owner'], space=parent).exists(): raise ValueError('L’espace privé doit être rattaché à un espace accessible par son propriétaire.') 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 # 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): candidate = f'{number:04d}' target = directory / candidate if DriveSpace.objects.filter(directory_name=candidate).exists() or target.exists() or target.is_symlink(): number += 1 continue try: target.mkdir(mode=0o700) except FileExistsError: number += 1 continue try: obj.directory_name = candidate obj.relative_path = '/'.join(part for part in (base, candidate) if part) obj.save(update_fields=['directory_name', 'relative_path']) except Exception: target.rmdir() raise return obj raise ValueError('Aucun numéro de répertoire disponible pour ce nouvel espace.') def create_space(user, parent, display_name, **permissions): """Create a delegated (shared) space owned by a new user.""" if parent.private_owner_id: raise ValueError('Un espace privé ne peut pas accueillir de sous-utilisateur.') return _make_numbered_space(parent, user=user, display_name=display_name, **permissions) def ensure_private_space(user): """Every account has another, strictly private, numbered directory.""" obj = DriveSpace.objects.filter(private_owner=user).first() if obj: (Path(settings.DRIVE_STORAGE_ROOT) / obj.relative_path).mkdir(mode=0o700, parents=True, exist_ok=True) return obj with transaction.atomic(): obj = DriveSpace.objects.select_for_update().filter(private_owner=user).first() if obj: return obj assigned = ensure_space(user) # The 50-Go cap is only a limit on private uploads, not a reservation. return _make_numbered_space(assigned, private_owner=user, display_name='Mes fichiers') def my_space(user): return ensure_space(user) def _grant(user, space): if not user.is_authenticated or not user.is_active: return None return SpaceGrant.objects.filter(user=user, space=space).first() def can_read_space(user, space): if not user.is_authenticated or not user.is_active: return False # Personal files are never subject to administrative or inherited access. if space.private_owner_id: return space.private_owner_id == user.pk if is_admin(user) or space.managed_by_id == user.pk: return True own = ensure_space(user) if own.pk == space.pk and own.user_id == user.pk: return True if own.user_id == user.pk and own.parent_id == space.pk and own.can_view_parent: return True return _grant(user, space) is not None def can_write_space(user, space): if not user.is_authenticated or not user.is_active: return False # Personal files are never subject to administrative or inherited access. if space.private_owner_id: return space.private_owner_id == user.pk if is_admin(user) or space.managed_by_id == user.pk: return True own = ensure_space(user) if own.pk == space.pk and own.user_id == user.pk: return True if own.user_id == user.pk and own.parent_id == space.pk and own.can_view_parent and own.can_write_parent: return True grant = _grant(user, space) return bool(grant and grant.can_write) def can_manage_space(user, space): if not user.is_authenticated or not user.is_active or space.private_owner_id: return False if is_admin(user) or space.managed_by_id == user.pk: return True if space.user_id == user.pk and space.can_invite: return True grant = _grant(user, space) return bool(grant and grant.can_invite) def resolve_space(user, space_id=None, write=False): if not space_id: space = ensure_private_space(user) else: try: space = DriveSpace.objects.select_related('user', 'private_owner', 'parent').get(pk=int(space_id)) except (ValueError, TypeError, DriveSpace.DoesNotExist): raise Http404 permitted = can_write_space(user, space) if write else can_read_space(user, space) if not permitted: raise PermissionError('Vous n’avez pas accès à cet espace.') return space def space_info(space, user): from .quotas import quota_state owner = space.private_owner or space.user or space.managed_by return { 'id': space.pk, 'user_id': owner.pk, 'username': owner.username, 'first_name': owner.first_name, 'last_name': owner.last_name, 'parent_id': space.parent_id, 'display_name': space.display_name, 'is_private': bool(space.private_owner_id), 'can_read': can_read_space(user, space), 'can_write': can_write_space(user, space), 'can_invite': can_manage_space(user, space), 'can_edit': is_admin(user) or space.managed_by_id == user.pk or (space.user_id == user.pk and can_manage_space(user, space)), 'has_account': bool(space.user_id), 'is_active': owner.is_active if owner else True, **quota_state(space), 'can_view_parent': space.can_view_parent, 'can_write_parent': space.can_write_parent, 'created_at': space.created_at.isoformat(), } def storage_path(item): key = item.storage_key if not key or len(key) != 36: raise Http404 relative = item.storage_location or f'{key[:2]}/{key}' path = Path(settings.DRIVE_STORAGE_ROOT) / relative root = Path(settings.DRIVE_STORAGE_ROOT).resolve() resolved = path.resolve() if not resolved.is_relative_to(root): raise Http404 return resolved def hash_token(token): return hashlib.sha256(token.encode('utf-8')).hexdigest()