206 lines
8.4 KiB
Python
206 lines
8.4 KiB
Python
"""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):
|
||
"""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):
|
||
# 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:
|
||
raise ValueError('L’espace privé doit être rattaché au quota attribué à 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:
|
||
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 hash_token(token):
|
||
return hashlib.sha256(token.encode('utf-8')).hexdigest()
|