"""Scoped user sharing of individual items and folder descendants. User shares grant no space-level privileges; public links use cryptographically random bearer tokens and keep only SHA-256 digests in the database. """ import hashlib import secrets from django.contrib.auth import get_user_model from django.core.exceptions import ValidationError from django.http import FileResponse, Http404, HttpResponse, StreamingHttpResponse from django.db.models import Q from django.utils import timezone from rest_framework.decorators import api_view, permission_classes, throttle_classes from rest_framework.permissions import AllowAny from rest_framework.throttling import AnonRateThrottle from rest_framework.response import Response from .models import DriveItem, ItemGrant, PublicItemLink from .services import error, get_item, readable, writable, chain, item_info from .spaces import storage_path from .word_preview import render_word_preview from .document_formats import render_odf_preview from .permissions import item_rights, assert_item_delegation, available_item_share_rights, PRESETS from .views_content import safe_preview_type, read_bytes def shareable(item, user): return readable(item, user) and 'share' in item_rights(user,item) def share_info(grant): u=grant.user return {'id': grant.pk,'permissions':sorted((set(grant.permissions) if grant.permissions is not None else set(PRESETS['reader'])) & set(available_item_share_rights(grant.item)),key=('view','download','modify','add').index),'user_id':u.pk,'username':u.username, 'first_name':u.first_name,'last_name':u.last_name} @api_view(['GET']) def search_share_users(request): q=str(request.query_params.get('q','')).strip()[:100] if len(q)<2:return Response({'users':[]}) User=get_user_model() users=User.objects.filter(is_active=True).exclude(pk=request.user.pk).filter( Q(username__icontains=q) | Q(first_name__icontains=q) | Q(last_name__icontains=q) ).order_by('username')[:15] return Response({'users':[{'id':u.pk,'username':u.username, 'first_name':u.first_name,'last_name':u.last_name} for u in users]}) @api_view(['GET','POST']) def manage_item_shares(request,item_id): item=get_item(item_id) if not shareable(item,request.user):return error('Partage non autorisé.',403) if request.method=='GET': grants=ItemGrant.objects.filter(item=item).select_related('user','item').order_by('user__username') links=PublicItemLink.objects.filter(item=item,revoked_at__isnull=True).order_by('-created_at') return Response({'users':[share_info(g) for g in grants], 'links':[{'id':l.pk,'created_at':l.created_at.isoformat()} for l in links]}) if item.trashed_at:return error('Impossible de partager un élément supprimé.',400) try:user_id=int(request.data.get('user_id')) except (TypeError,ValueError):return error('Utilisateur invalide.') try:account=get_user_model().objects.get(pk=user_id,is_active=True) except get_user_model().DoesNotExist:return error('Utilisateur introuvable.',404) if account.pk==request.user.pk:return error('Cet élément t’appartient déjà.') try: rights=assert_item_delegation(item_rights(request.user,item),request.data.get('permissions',list(PRESETS['reader'])),item) except ValueError as exc:return error(str(exc),403) grant,created=ItemGrant.objects.get_or_create(item=item,user=account,defaults={'created_by':request.user,'permissions':rights}) if not created:return error('Cet élément est déjà partagé avec cet utilisateur.',409) return Response(share_info(grant),status=201) @api_view(['PATCH']) def update_item_share(request,item_id,grant_id): item=get_item(item_id) if not shareable(item,request.user):return error('Partage non autorisé.',403) try: rights=assert_item_delegation(item_rights(request.user,item), request.data.get('permissions'),item) except ValueError as exc:return error(str(exc),403) changed=ItemGrant.objects.filter(item=item,pk=grant_id).update(permissions=rights) if not changed:return error('Partage introuvable.',404) return Response(share_info(ItemGrant.objects.select_related('user','item').get(pk=grant_id))) @api_view(['DELETE']) def remove_item_share(request,item_id,grant_id): item=get_item(item_id) if not shareable(item,request.user):return error('Partage non autorisé.',403) deleted,_=ItemGrant.objects.filter(pk=grant_id,item=item).delete() if not deleted:return error('Partage introuvable.',404) return Response(status=204) @api_view(['DELETE']) def leave_item_share(request,item_id): """Remove only the current recipient's share, never the underlying item.""" item=get_item(item_id) for node in chain(item): grant=ItemGrant.objects.filter(item=node,user=request.user).first() if grant is not None: grant.delete() return Response(status=204) return error('Aucun partage à retirer.',404) @api_view(['POST']) def public_links(request,item_id): item=get_item(item_id) if not shareable(item,request.user) or not {'view','download','share'}.issubset(item_rights(request.user,item)): return error('Partage non autorisé.',403) token=secrets.token_urlsafe(32) PublicItemLink.objects.create(item=item,token_hash=hashlib.sha256(token.encode()).hexdigest(),created_by=request.user) return Response({'token':token},status=201) @api_view(['DELETE']) def revoke_public_link(request,item_id,link_id): item=get_item(item_id) if not shareable(item,request.user):return error('Partage non autorisé.',403) changed=PublicItemLink.objects.filter(pk=link_id,item=item,revoked_at__isnull=True).update(revoked_at=timezone.now()) if not changed:return error('Lien introuvable.',404) return Response(status=204) def get_public_root(token): if not token or len(token)>128:raise Http404 digest=hashlib.sha256(token.encode()).hexdigest() try:link=PublicItemLink.objects.select_related('item','item__parent','item__space').get( token_hash=digest,revoked_at__isnull=True) except PublicItemLink.DoesNotExist:raise Http404 if any(n.trashed_at for n in chain(link.item)):raise Http404 return link.item def in_public_tree(root, candidate): if candidate.space_id!=root.space_id:return False return any(ancestor.pk==root.pk for ancestor in chain(candidate)) and not any(n.trashed_at for n in chain(candidate)) @api_view(['GET']) @permission_classes([AllowAny]) @throttle_classes([AnonRateThrottle]) def public_share(request,token): root=get_public_root(token) parent_id=request.query_params.get('parent') if parent_id: try:folder=get_item(parent_id) except (Http404,ValueError,ValidationError):raise Http404 else:folder=root if not folder.is_folder or not in_public_tree(root,folder): if not parent_id and not root.is_folder: return Response({'root':{'id':str(root.pk),'name':root.name,'is_folder':False,'size':root.size,'mime_type':root.mime_type},'items':[]}) raise Http404 qs=DriveItem.objects.filter(parent=folder,trashed_at__isnull=True).order_by('-is_folder','name')[:501] items=[{'id':str(i.pk),'name':i.name,'is_folder':i.is_folder,'size':i.size,'mime_type':i.mime_type} for i in qs] # Never list ancestors outside the shared root. breadcrumb=[] for node in reversed(chain(folder)): if node.pk==root.pk or breadcrumb: breadcrumb.append({'id':str(node.pk),'name':node.name}) return Response({'root':{'id':str(root.pk),'name':root.name,'is_folder':True}, 'breadcrumb':breadcrumb,'items':items[:500],'limited':len(items)>500}) @api_view(['GET','HEAD']) @permission_classes([AllowAny]) @throttle_classes([AnonRateThrottle]) def public_file(request,token,item_id): """Bearer-link file access; inline previews only for safe MIME types.""" import re root=get_public_root(token) item=get_item(item_id) if item.is_folder or not in_public_tree(root,item):raise Http404 path=storage_path(item) if not path.is_file():raise Http404 inline=request.query_params.get('inline')=='1' preview=safe_preview_type(item) if inline and not preview:return error('Aperçu indisponible pour ce format.',415) content_type=preview if inline else 'application/octet-stream' size=path.stat().st_size if inline and item.name.lower().endswith(('.docx', '.odt', '.ods', '.odp')): try: html = render_word_preview(path) if item.name.lower().endswith('.docx') else render_odf_preview(path) except ValueError: raise Http404 response = HttpResponse(html if request.method=='GET' else '', content_type='text/html; charset=utf-8') response['Cache-Control'] = 'no-store' response['Content-Security-Policy'] = "sandbox; default-src 'none'; style-src 'unsafe-inline'; img-src data:" response['X-Content-Type-Options'] = 'nosniff' response['X-Frame-Options'] = 'SAMEORIGIN' response['Referrer-Policy'] = 'no-referrer' return response byte_range=request.headers.get('Range') if byte_range: match=re.fullmatch(r'bytes=(\d*)-(\d*)',byte_range.strip()) if not match or (not match[1] and not match[2]): response=HttpResponse(status=416) response['Content-Range']=f'bytes */{size}' return response if match[1]: start=int(match[1]);end=int(match[2]) if match[2] else size-1 else: start=max(0,size-int(match[2]));end=size-1 end=min(end,size-1) if start>=size or end