170 lines
6.4 KiB
Python
170 lines
6.4 KiB
Python
"""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()
|