Loading src/hls/tests.py +25 −0 Original line number Diff line number Diff line Loading @@ -103,6 +103,31 @@ class HLSAccessTestCase(TestCase): response = file_passthru(self._request(self.user), self._path()) self.assertEqual(response.status_code, 200) @override_settings(HLS_URLS_NEED_AUTH=True) def test_hls_access_binds_authorization_to_resolved_file_path(self): other_project = Project.objects.create(label="private hls project") private_hash = "a" * 64 File.objects.create( title="private video", original_name="private.mp4", project=other_project, hash=private_hash, file_type=self.file.file_type, size=123, ) misleading_path = "{}/../../../{}/master.m3u8".format( hash_to_hls_path(self.file_hash), hash_to_hls_path(private_hash) ) with TemporaryDirectory() as temp_dir: hls_dir = Path(temp_dir) private_hls_file = hls_dir / hash_to_hls_path(private_hash) / "master.m3u8" private_hls_file.parent.mkdir(parents=True) private_hls_file.write_text("private HLS data", encoding="utf-8") with override_settings(HLS_DIR=str(hls_dir)): with self.assertRaises(PermissionDenied): file_passthru(self._request(self.user), misleading_path) @override_settings(HLS_URLS_NEED_AUTH=True) def test_hls_access_404s_when_path_has_no_hash(self): with self.assertRaises(Http404): Loading src/hls/views.py +43 −17 Original line number Diff line number Diff line Loading @@ -10,34 +10,62 @@ import re def _file_hash_from_hls_path(path: str) -> Union[str, None]: match = re.search(r"(?i)(?<![0-9a-f])[0-9a-f]{64}(?![0-9a-f])", path) if not match: resolved_path = _resolved_hls_path(path) return _file_hash_from_resolved_hls_path(resolved_path) def _file_hash_from_resolved_hls_path( resolved_path: Path, ) -> Union[str, None]: relative_parts = resolved_path.relative_to(Path(settings.HLS_DIR).resolve()).parts if len(relative_parts) < 4: return None first_shard, second_shard, file_hash = relative_parts[:3] if not re.fullmatch(r"(?i)[0-9a-f]{64}", file_hash): return None file_hash = file_hash.lower() if first_shard.lower() != file_hash[:2] or second_shard.lower() != file_hash[2:4]: return None return match.group(0).lower() return file_hash def _hls_file_path(path: str) -> Path: def _resolved_hls_path(path: str) -> Path: hls_dir = Path(settings.HLS_DIR).resolve() tested_path = Path(hls_dir, path).resolve() if not tested_path.is_relative_to(hls_dir): resolved_path = Path(hls_dir, path).resolve() if not resolved_path.is_relative_to(hls_dir): raise PermissionDenied() if not tested_path.exists(): return resolved_path def _hls_file_path(path: str) -> Path: resolved_path = _resolved_hls_path(path) if not resolved_path.is_file(): raise Http404() return tested_path return resolved_path def _hls_response(request: HttpRequest, path: str, resource_instance: Resource): def _hls_response( request: HttpRequest, hls_file_path: Path, resource_instance: Resource ): if not hls_file_path.is_file(): raise Http404() return RangedFileResponse( request, open(_hls_file_path(path), "rb"), open(hls_file_path, "rb"), content_type=resource_instance.file.file_type.mime, ) def file_passthru_noauth(request: HttpRequest, path: str) -> HttpResponseBase: file_hash = _file_hash_from_hls_path(path) def _hls_target(path: str) -> tuple[str, Path]: resolved_path = _resolved_hls_path(path) file_hash = _file_hash_from_resolved_hls_path(resolved_path) if not file_hash: raise Http404() return file_hash, resolved_path def file_passthru_noauth(request: HttpRequest, path: str) -> HttpResponseBase: file_hash, hls_file_path = _hls_target(path) resource_instance = ( Resource.objects.filter(file__hash=file_hash, deleted_at__isnull=True) .select_related("file__file_type") Loading @@ -45,16 +73,14 @@ def file_passthru_noauth(request: HttpRequest, path: str) -> HttpResponseBase: ) if not resource_instance: raise Http404() return _hls_response(request, path, resource_instance) return _hls_response(request, hls_file_path, resource_instance) def file_passthru_auth(request: HttpRequest, path: str) -> HttpResponseBase: if request.user.is_anonymous: raise PermissionDenied() file_hash = _file_hash_from_hls_path(path) if not file_hash: raise Http404() file_hash, hls_file_path = _hls_target(path) resources = Resource.objects.filter( file__hash=file_hash, Loading @@ -67,7 +93,7 @@ def file_passthru_auth(request: HttpRequest, path: str) -> HttpResponseBase: if UserAccess(request.user, resource_instance.ptr_project).can_read( resource_instance ): return _hls_response(request, path, resource_instance) return _hls_response(request, hls_file_path, resource_instance) raise PermissionDenied() Loading Loading
src/hls/tests.py +25 −0 Original line number Diff line number Diff line Loading @@ -103,6 +103,31 @@ class HLSAccessTestCase(TestCase): response = file_passthru(self._request(self.user), self._path()) self.assertEqual(response.status_code, 200) @override_settings(HLS_URLS_NEED_AUTH=True) def test_hls_access_binds_authorization_to_resolved_file_path(self): other_project = Project.objects.create(label="private hls project") private_hash = "a" * 64 File.objects.create( title="private video", original_name="private.mp4", project=other_project, hash=private_hash, file_type=self.file.file_type, size=123, ) misleading_path = "{}/../../../{}/master.m3u8".format( hash_to_hls_path(self.file_hash), hash_to_hls_path(private_hash) ) with TemporaryDirectory() as temp_dir: hls_dir = Path(temp_dir) private_hls_file = hls_dir / hash_to_hls_path(private_hash) / "master.m3u8" private_hls_file.parent.mkdir(parents=True) private_hls_file.write_text("private HLS data", encoding="utf-8") with override_settings(HLS_DIR=str(hls_dir)): with self.assertRaises(PermissionDenied): file_passthru(self._request(self.user), misleading_path) @override_settings(HLS_URLS_NEED_AUTH=True) def test_hls_access_404s_when_path_has_no_hash(self): with self.assertRaises(Http404): Loading
src/hls/views.py +43 −17 Original line number Diff line number Diff line Loading @@ -10,34 +10,62 @@ import re def _file_hash_from_hls_path(path: str) -> Union[str, None]: match = re.search(r"(?i)(?<![0-9a-f])[0-9a-f]{64}(?![0-9a-f])", path) if not match: resolved_path = _resolved_hls_path(path) return _file_hash_from_resolved_hls_path(resolved_path) def _file_hash_from_resolved_hls_path( resolved_path: Path, ) -> Union[str, None]: relative_parts = resolved_path.relative_to(Path(settings.HLS_DIR).resolve()).parts if len(relative_parts) < 4: return None first_shard, second_shard, file_hash = relative_parts[:3] if not re.fullmatch(r"(?i)[0-9a-f]{64}", file_hash): return None file_hash = file_hash.lower() if first_shard.lower() != file_hash[:2] or second_shard.lower() != file_hash[2:4]: return None return match.group(0).lower() return file_hash def _hls_file_path(path: str) -> Path: def _resolved_hls_path(path: str) -> Path: hls_dir = Path(settings.HLS_DIR).resolve() tested_path = Path(hls_dir, path).resolve() if not tested_path.is_relative_to(hls_dir): resolved_path = Path(hls_dir, path).resolve() if not resolved_path.is_relative_to(hls_dir): raise PermissionDenied() if not tested_path.exists(): return resolved_path def _hls_file_path(path: str) -> Path: resolved_path = _resolved_hls_path(path) if not resolved_path.is_file(): raise Http404() return tested_path return resolved_path def _hls_response(request: HttpRequest, path: str, resource_instance: Resource): def _hls_response( request: HttpRequest, hls_file_path: Path, resource_instance: Resource ): if not hls_file_path.is_file(): raise Http404() return RangedFileResponse( request, open(_hls_file_path(path), "rb"), open(hls_file_path, "rb"), content_type=resource_instance.file.file_type.mime, ) def file_passthru_noauth(request: HttpRequest, path: str) -> HttpResponseBase: file_hash = _file_hash_from_hls_path(path) def _hls_target(path: str) -> tuple[str, Path]: resolved_path = _resolved_hls_path(path) file_hash = _file_hash_from_resolved_hls_path(resolved_path) if not file_hash: raise Http404() return file_hash, resolved_path def file_passthru_noauth(request: HttpRequest, path: str) -> HttpResponseBase: file_hash, hls_file_path = _hls_target(path) resource_instance = ( Resource.objects.filter(file__hash=file_hash, deleted_at__isnull=True) .select_related("file__file_type") Loading @@ -45,16 +73,14 @@ def file_passthru_noauth(request: HttpRequest, path: str) -> HttpResponseBase: ) if not resource_instance: raise Http404() return _hls_response(request, path, resource_instance) return _hls_response(request, hls_file_path, resource_instance) def file_passthru_auth(request: HttpRequest, path: str) -> HttpResponseBase: if request.user.is_anonymous: raise PermissionDenied() file_hash = _file_hash_from_hls_path(path) if not file_hash: raise Http404() file_hash, hls_file_path = _hls_target(path) resources = Resource.objects.filter( file__hash=file_hash, Loading @@ -67,7 +93,7 @@ def file_passthru_auth(request: HttpRequest, path: str) -> HttpResponseBase: if UserAccess(request.user, resource_instance.ptr_project).can_read( resource_instance ): return _hls_response(request, path, resource_instance) return _hls_response(request, hls_file_path, resource_instance) raise PermissionDenied() Loading