shared_chat_video_stream.py
Rama main · réplica de lectura del repositorio original.
"""Bounded HTTP video streaming from an already authorized, pinned descriptor.
This module does not resolve paths, read a share ledger, or open server state.
The caller securely opens and validates a regular file and transfers exclusive
ownership of its descriptor to SharedVideoResponse, including on init failure.
"""
from __future__ import annotations
import os
import re
import stat
import threading
import time
from urllib.parse import quote
from starlette.concurrency import run_in_threadpool
from starlette.responses import Response, StreamingResponse
VIDEO_CONTENT_TYPES = frozenset({"video/mp4", "video/webm", "video/quicktime", "video/ogg"})
VIDEO_CHUNK_BYTES = 64 * 1024
REAUTHORIZE_INTERVAL_SECONDS = 1.0
_RANGE = re.compile(r"bytes=([0-9]*)-([0-9]*)", re.IGNORECASE)
_SECURITY_HEADERS = {
"cache-control": "no-store",
"referrer-policy": "no-referrer",
"x-content-type-options": "nosniff",
"x-robots-tag": "noindex, nofollow, noarchive",
"cross-origin-resource-policy": "same-origin",
}
class SharedVideoUnavailable(RuntimeError):
"""Abort an already started response without exposing internal details."""
class VideoStreamAdmission:
"""A bounded, nonwaiting lease shared by both video route families."""
def __init__(self, limit=8):
if type(limit) is not int or limit < 1:
raise ValueError("Invalid video stream limit")
self.limit = limit
self._active = 0
self._lock = threading.Lock()
def acquire(self):
with self._lock:
if self._active >= self.limit:
return False
self._active += 1
return True
def release(self):
with self._lock:
self._active -= 1
_ADMISSION = VideoStreamAdmission()
def _byte_range(request, byte_size):
"""Return (status, start, length), including strict single-range rejection."""
values = request.headers.getlist("range")
if not values:
return 200, 0, byte_size
if len(values) != 1 or len(values[0]) > 256:
return 416, 0, 0
match = _RANGE.fullmatch(values[0].strip())
if match is None or not any(match.groups()) or byte_size == 0:
return 416, 0, 0
first, last = match.groups()
if first:
start = int(first)
end = min(int(last), byte_size - 1) if last else byte_size - 1
if start >= byte_size or end < start:
return 416, 0, 0
else:
suffix = int(last)
if suffix == 0:
return 416, 0, 0
start, end = max(0, byte_size - suffix), byte_size - 1
return 206, start, end - start + 1
def _file_revision(info):
return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns,
info.st_ctime_ns, info.st_uid, info.st_mode, info.st_nlink)
class SharedVideoResponse(StreamingResponse):
"""GET/HEAD response owning a descriptor for its complete ASGI lifecycle.
``request`` is a Starlette Request. ``reauthorize`` is an optional async
zero-argument callback: None/True grants access, False or an exception denies
it. It runs immediately before headers and, on demand while consuming the
response, at most once per second. Denial before headers produces an empty
404; denial after headers aborts the stream with SharedVideoUnavailable.
Admission is acquired only when ASGI invokes the response and released on
every exit, including header disconnects. ``close()`` explicitly disposes a
response that will never be invoked; finalization is an additional fallback.
Neither the caller nor any other object may close/reuse the transferred fd.
``file_revision`` may pin the resolver's fstat result as the tuple
(dev, ino, size, mtime_ns, ctime_ns, uid, mode, nlink). The revision is checked
at construction, before headers, and before/after every bounded read.
"""
def __init__(self, fd, *, byte_size, content_type, filename, request,
reauthorize=None, extra_headers=None, admission=None, file_revision=None):
self._fd = fd
self._fd_lock = threading.Lock()
self._called = False
try:
if (type(fd) is not int or fd < 0 or type(byte_size) is not int
or byte_size < 0 or content_type not in VIDEO_CONTENT_TYPES):
raise ValueError("Invalid shared video")
info = os.fstat(fd)
if not stat.S_ISREG(info.st_mode) or info.st_size != byte_size:
raise ValueError("Invalid shared video")
self._file_revision = _file_revision(info)
if file_revision is not None and file_revision != self._file_revision:
raise ValueError("Invalid shared video")
# The caller supplies a display filename, never a path. Fail closed
# on path/control characters even if that contract was violated.
if (not isinstance(filename, str) or not filename or len(filename) > 255
or filename in {".", ".."}
or re.search(r"[\x00-\x1f\x7f/\\]", filename)):
raise ValueError("Invalid shared video")
self._method = request.method.upper()
self._reauthorize = reauthorize
self._last_authorized = None
self._admission = admission if admission is not None else _ADMISSION
self._base_headers = {str(key).lower(): value for key, value in (extra_headers or {}).items()}
# Route CSP is retained; callers cannot replace the mandatory
# privacy headers or representation/framing headers below.
for key in ("content-type", "content-length", "content-range", "content-disposition",
"accept-ranges", "transfer-encoding", "etag", "last-modified"):
self._base_headers.pop(key, None)
self._base_headers.update(_SECURITY_HEADERS)
headers = dict(self._base_headers)
status, self._start, self._length = _byte_range(request, byte_size)
if self._method not in {"GET", "HEAD"}:
status, self._length = 405, 0
headers["allow"] = "GET, HEAD"
else:
headers["accept-ranges"] = "bytes"
if status == 416:
headers["content-range"] = f"bytes */{byte_size}"
else:
headers["content-type"] = content_type
headers["content-disposition"] = "inline; filename*=UTF-8''" + quote(filename, safe="")
if status == 206:
headers["content-range"] = f"bytes {self._start}-{self._start + self._length - 1}/{byte_size}"
headers["content-length"] = str(self._length)
# stream_response owns the body; no generator owns the descriptor.
super().__init__(iter(()), status_code=status, headers=headers)
except BaseException:
self.close()
raise
def close(self):
"""Idempotently close the owned descriptor, including unused responses."""
with self._fd_lock:
fd, self._fd = self._fd, None
if type(fd) is int and fd >= 0:
try:
os.close(fd)
except OSError:
pass
def __del__(self):
try:
self.close()
except Exception:
pass
def _read(self, size, offset):
# Cancellation may end an await while its worker still runs. Dup under
# the close lock so a delayed worker can never read a reused descriptor.
# The worker owns that duplicate until its one bounded read finishes.
with self._fd_lock:
if self._fd is None:
raise SharedVideoUnavailable("Shared video unavailable")
fd = os.dup(self._fd)
try:
if _file_revision(os.fstat(fd)) != self._file_revision:
raise SharedVideoUnavailable("Shared video unavailable")
chunk = os.pread(fd, size, offset)
if _file_revision(os.fstat(fd)) != self._file_revision:
raise SharedVideoUnavailable("Shared video unavailable")
return chunk
finally:
os.close(fd)
def _unchanged(self):
with self._fd_lock:
try:
return self._fd is not None and _file_revision(os.fstat(self._fd)) == self._file_revision
except OSError:
return False
async def _authorized(self, *, force=False):
if self._reauthorize is None:
return True
if (not force and self._last_authorized is not None
and time.monotonic() - self._last_authorized < REAUTHORIZE_INTERVAL_SECONDS):
return True
try:
granted = await self._reauthorize()
except Exception:
return False
self._last_authorized = time.monotonic()
return granted is None or granted is True
async def stream_response(self, send):
if not await self._authorized(force=True):
await self._empty_response(send, 404)
return
if not self._unchanged():
await self._empty_response(send, 404)
return
await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})
if self._method != "HEAD" and self.status_code in {200, 206}:
offset, remaining = self._start, self._length
while remaining:
if not await self._authorized():
raise SharedVideoUnavailable("Shared video unavailable")
try:
chunk = await run_in_threadpool(self._read, min(VIDEO_CHUNK_BYTES, remaining), offset)
except OSError:
raise SharedVideoUnavailable("Shared video unavailable") from None
if not chunk:
raise SharedVideoUnavailable("Shared video unavailable")
# A slow disk read can cross the authorization interval.
if not await self._authorized():
raise SharedVideoUnavailable("Shared video unavailable")
await send({"type": "http.response.body", "body": chunk, "more_body": True})
offset += len(chunk)
remaining -= len(chunk)
await send({"type": "http.response.body", "body": b"", "more_body": False})
async def _empty_response(self, send, status):
response = Response(status_code=status, headers=self._base_headers)
await send({"type": "http.response.start", "status": status, "headers": response.raw_headers})
await send({"type": "http.response.body", "body": b"", "more_body": False})
async def __call__(self, scope, receive, send):
admitted = False
try:
if self._called or self._fd is None:
raise SharedVideoUnavailable("Shared video unavailable")
self._called = True
admitted = self._admission.acquire()
if not admitted:
await self._empty_response(send, 503)
return
await super().__call__(scope, receive, send)
finally:
self.close()
if admitted:
self._admission.release()