"""Physical numbered directories and strict permissions between personal/shared spaces.""" 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 def is_admin(user): return user.is_authenticated and (user.is_superuser or user.is_staff) def ensure_space(user): """One shared/assigned space for every user; the administrator owns root.""" 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 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): if len(parent.relative_path) + 1 + 24 > 700: raise ValueError('Hiérarchie trop profonde.') obj = DriveSpace.objects.create(parent=parent, relative_path=f'pending/{secrets.token_hex(10)}', **fields) obj.directory_name = f'{obj.pk - 1:04d}' obj.relative_path = '/'.join(part for part in (parent.relative_path, obj.directory_name) if part) obj.save(update_fields=['directory_name', 'relative_path']) (Path(settings.DRIVE_STORAGE_ROOT) / obj.relative_path).mkdir(mode=0o700, parents=True, exist_ok=True) return obj 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 shared = ensure_space(user) return _make_numbered_space(shared, 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: return True if 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: return True if 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), '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 new_storage_location(space, key): return '/'.join(part for part in (space.relative_path, key[:2], key) if part) def hash_token(token): return hashlib.sha256(token.encode('utf-8')).hexdigest()