"""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 from .permissions import has_space_right, space_rights, can_change_space_quota def is_admin(user): return user.is_authenticated and (user.is_superuser or user.is_staff) def ensure_space(user): """Return an accessible shared space, or None if access was revoked. Never recreate a deleted assigned space or grant access to the root when an administrator removes an account's last shared space. """ obj = DriveSpace.objects.filter(user=user, private_owner__isnull=True).first() if obj: return obj grant = SpaceGrant.objects.filter(user=user).select_related('space').order_by('created_at', 'pk').first() if grant: return grant.space if not is_admin(user): return None with transaction.atomic(): root = DriveSpace.objects.filter(parent__isnull=True, private_owner__isnull=True).order_by('id').first() if root is None: root = DriveSpace.objects.create(user=user, relative_path='', display_name='Racine') return root 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) if assigned is None: assigned = DriveSpace.objects.filter(parent__isnull=True, private_owner__isnull=True).first() if assigned is None: raise PermissionError('Stockage principal non initialisé.') # 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): return has_space_right(user, space, 'view') def can_write_space(user, space): # Kept for legacy callers: use explicit right checks on each operation. return has_space_right(user, space, 'add') or has_space_right(user, space, 'modify') def can_manage_space(user, space): return has_space_right(user, space, 'manage_access') 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 or user 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), 'permissions': sorted(space_rights(user, space)), 'can_edit': not space.private_owner_id and has_space_right(user, space, 'manage_space'), 'can_edit_quota': can_change_space_quota(user, space), 'can_delete': not space.private_owner_id and has_space_right(user, space, 'manage_space') and (space.parent_id is not None or is_admin(user)), '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, 'parent_permissions': space.parent_permissions if space.parent_permissions is not None else (['view','download'] + (['add','modify','delete','share'] if space.can_write_parent else []) + (['manage_access','create_spaces'] if space.can_invite else [])), '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()