如何在Django及DRF中安全处理私有媒体文件权限与视频流？

要解决媒体文件直接URL访问、权限控制和视频流式传输的问题，核心思路是不让Web服务器直接处理私有文件请求，而是通过Django视图接管所有访问请求，在视图层完成权限验证后再返回文件内容。以下是具体实现步骤：

1. 调整媒体文件存储结构与Web服务器配置 #

首先要把私有文件和公开媒体文件分开，禁止Web服务器直接提供私有文件的访问：

• 在MEDIA_ROOT下创建private子目录，专门存放需要权限验证的文件（比如数字产品的视频、文档）。
• 修改Web服务器（以Nginx为例）配置，将私有文件的请求转发给Django：

  server {
      # 公开媒体文件的配置（保持不变）
      location /media/ {
          alias /path/to/your/project/media/;
      }
  
      # 私有媒体文件的请求转发给Django处理
      location /private-media/ {
          internal; # 禁止外部直接访问这个location
          uwsgi_pass unix:/path/to/your/uwsgi.sock;
          include uwsgi_params;
          # 传递请求路径给Django
          uwsgi_param PATH_INFO /private-media$request_uri;
      }
  }
  

  这里的internal指令确保Nginx不会直接响应外部对/private-media/的请求，必须通过Django转发。

2. 实现权限验证的文件访问视图 #

2.1 基础权限判断逻辑 #

假设你的项目中有Product、Order模型，其中Order关联用户和产品，且有status字段标记订单是否完成：

# models.py
from django.db import models
from django.contrib.auth import get_user_model

User = get_user_model()

class Product(models.Model):
    name = models.CharField(max_length=255)
    # 私有文件字段，存储到private子目录
    media_file = models.FileField(upload_to='private/products/')
    video_file = models.FileField(upload_to='private/videos/')

class Order(models.Model):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    product = models.ForeignKey(Product, on_delete=models.CASCADE)
    status = models.CharField(max_length=20, choices=[('completed', '已完成'), ('pending', '待处理')])
    created_at = models.DateTimeField(auto_now_add=True)

判断用户是否有权访问某产品文件的逻辑：检查用户是否存在该产品的已完成订单。

2.2 普通文件下载视图 #

用DRF的APIView实现，先验证权限，再返回文件：

# views.py
from django.http import FileResponse
from rest_framework.views import APIView
from rest_framework.permissions import IsAuthenticated
from rest_framework.exceptions import PermissionDenied, NotFound
from .models import Product, Order

class PrivateFileDownloadView(APIView):
    permission_classes = [IsAuthenticated]

    def get(self, request, product_id):
        try:
            product = Product.objects.get(id=product_id)
        except Product.DoesNotExist:
            raise NotFound("产品不存在")
        
        # 验证用户是否已购买该产品
        has_purchased = Order.objects.filter(
            user=request.user,
            product=product,
            status='completed'
        ).exists()
        
        if not has_purchased:
            raise PermissionDenied("你未购买该产品，无法访问文件")
        
        # 返回文件，设置Content-Disposition让浏览器下载
        response = FileResponse(product.media_file.open('rb'))
        response['Content-Disposition'] = f'attachment; filename="{product.media_file.name.split("/")[-1]}"'
        return response

2.3 视频流式传输视图 #

视频需要支持断点续传和流式播放，要处理HTTP的Range请求，用StreamingHttpResponse实现：

# views.py
import os
from django.http import StreamingHttpResponse, Http404
from rest_framework.views import APIView
from rest_framework.permissions import IsAuthenticated
from rest_framework.exceptions import PermissionDenied
from .models import Product, Order

class PrivateVideoStreamView(APIView):
    permission_classes = [IsAuthenticated]

    def get(self, request, product_id):
        try:
            product = Product.objects.get(id=product_id)
        except Product.DoesNotExist:
            raise Http404("视频不存在")
        
        # 权限验证
        has_purchased = Order.objects.filter(
            user=request.user,
            product=product,
            status='completed'
        ).exists()
        
        if not has_purchased:
            raise PermissionDenied("你未购买该产品，无法播放视频")
        
        video_path = product.video_file.path
        file_size = os.path.getsize(video_path)
        content_type = product.video_file.content_type or 'video/mp4'

        # 处理Range请求
        range_header = request.META.get('HTTP_RANGE', '')
        if range_header:
            start, end = range_header.strip().split('=')[1].split('-')
            start = int(start) if start else 0
            end = int(end) if end else file_size - 1
            length = end - start + 1
            response = StreamingHttpResponse(
                self.stream_video(video_path, start, end),
                status=206,
                content_type=content_type
            )
            response['Content-Range'] = f'bytes {start}-{end}/{file_size}'
        else:
            response = StreamingHttpResponse(
                self.stream_video(video_path),
                content_type=content_type
            )
            response['Content-Length'] = str(file_size)
        
        response['Accept-Ranges'] = 'bytes'
        return response

    def stream_video(self, file_path, start=0, end=None):
        chunk_size = 8192
        with open(file_path, 'rb') as f:
            f.seek(start)
            while True:
                chunk = f.read(chunk_size)
                if not chunk or (end is not None and f.tell() > end + 1):
                    break
                yield chunk

3. 配置URL路由 #

将视图映射到URL，确保所有私有文件访问都经过这些视图：

# urls.py
from django.urls import path
from .views import PrivateFileDownloadView, PrivateVideoStreamView

urlpatterns = [
    path('private-file/<int:product_id>/', PrivateFileDownloadView.as_view(), name='private_file_download'),
    path('private-video/<int:product_id>/', PrivateVideoStreamView.as_view(), name='private_video_stream'),
]

4. 前端访问注意事项 #

• 前端不要直接使用媒体文件的原始URL，而是通过上述路由生成的URL请求文件。比如用DRF的reverse方法在序列化器中返回安全的访问链接：

  # serializers.py
  from rest_framework import serializers
  from django.urls import reverse
  from .models import Product
  
  class ProductSerializer(serializers.ModelSerializer):
      download_url = serializers.SerializerMethodField()
      video_stream_url = serializers.SerializerMethodField()
  
      class Meta:
          model = Product
          fields = ['id', 'name', 'download_url', 'video_stream_url']
  
      def get_download_url(self, obj):
          request = self.context.get('request')
          return request.build_absolute_uri(reverse('private_file_download', args=[obj.id]))
  
      def get_video_stream_url(self, obj):
          request = self.context.get('request')
          return request.build_absolute_uri(reverse('private_video_stream', args=[obj.id]))
  

• 视频播放可以用HTML5的<video>标签，直接将video_stream_url赋值给src属性即可，浏览器会自动处理流式播放和断点续传。

额外优化建议 #

• 可以给视图添加缓存控制头，比如Cache-Control: private，避免浏览器缓存文件后未授权用户访问。
• 对于高并发场景，可以考虑使用CDN的私有访问模式（比如签名URL），但需要和Django的权限系统结合，确保签名仅颁发给已授权用户。
• 定期清理无效的私有文件，避免存储冗余。

内容的提问来源于stack exchange，提问作者Nahidujjaman Hridoy