shared_chat_videos.py
Rama main · réplica de lectura del repositorio original.
"""Read-only, signed references to videos already published in one chat.
No workspace paths, file discovery, registry writes, or video copies. Capabilities
pin the registered copy's metadata and inode revision; changed files fail closed.
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import stat
VIDEO_TYPES = frozenset({"video/mp4", "video/webm", "video/quicktime", "video/ogg"})
VIDEO_ID = re.compile(r"video_[A-Za-z0-9_-]+\.[a-f0-9]{64}\Z")
FILE_ID = re.compile(r"(?:file|art)_[a-f0-9]{16,32}\Z")
SESSION_ID = re.compile(r"[A-Za-z0-9_-]{1,128}\Z")
EXTENSIONS = {".mp4": "video/mp4", ".m4v": "video/mp4", ".webm": "video/webm",
".mov": "video/quicktime", ".ogv": "video/ogg", ".ogg": "video/ogg"}
class SharedVideoUnavailable(ValueError):
def __init__(self):
super().__init__("Shared video is unavailable")
def normalize_shared_chat_videos(value):
if not isinstance(value, list) or len(value) > 32:
raise ValueError("Invalid shared videos")
output, seen = [], set()
for item in value:
if not isinstance(item, dict) or set(item) != {"id", "filename", "content_type", "size"}:
raise ValueError("Invalid shared video")
identity, name = item["id"], item["filename"]
if (not isinstance(identity, str) or len(identity) > 1024 or VIDEO_ID.fullmatch(identity) is None
or identity in seen or not isinstance(name, str) or not 1 <= len(name) <= 255
or name in {".", ".."} or any(ord(c) < 32 or ord(c) == 127 or c in "/\\" for c in name)
or len(name.encode("utf-8")) > 1024 or not isinstance(item["content_type"], str) or item["content_type"] not in VIDEO_TYPES
or type(item["size"]) is not int or not 0 < item["size"] <= (1 << 53) - 1):
raise ValueError("Invalid shared video")
output.append(dict(item)); seen.add(identity)
return output
def _key(secret):
if not isinstance(secret, str) or not secret:
raise SharedVideoUnavailable()
return hmac.new(secret.encode("utf-8"), b"agentsdock-chat-video-v1", hashlib.sha256).digest()
def _regular(info, owner):
if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != owner:
raise SharedVideoUnavailable()
def _open_registered(root, session_id, file_id, *, legacy_owner=False):
"""Open through pinned directories; return an owned FD, never reopen a path."""
if (not isinstance(session_id, str) or SESSION_ID.fullmatch(session_id) is None
or not isinstance(file_id, str) or FILE_ID.fullmatch(file_id) is None):
raise SharedVideoUnavailable()
root = Path(root)
if not root.is_absolute() or not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"):
raise SharedVideoUnavailable()
directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
root_fd = directory_fd = meta_fd = video_fd = None
try:
root_fd = os.open(root, directory_flags)
owner = os.fstat(root_fd).st_uid
directory_fd = os.open(file_id, directory_flags, dir_fd=root_fd)
if os.fstat(directory_fd).st_uid != owner:
raise SharedVideoUnavailable()
meta_fd = os.open("meta.json", os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory_fd)
before = os.fstat(meta_fd)
_regular(before, owner)
if not 0 < before.st_size <= 64 * 1024:
raise SharedVideoUnavailable()
raw = os.read(meta_fd, 64 * 1024 + 1)
after = os.fstat(meta_fd)
if len(raw) != before.st_size or (before.st_mtime_ns, before.st_ctime_ns) != (after.st_mtime_ns, after.st_ctime_ns):
raise SharedVideoUnavailable()
meta = json.loads(raw)
if not isinstance(meta, dict) or meta.get("id") != file_id:
raise SharedVideoUnavailable()
if meta.get("session_id") != session_id and not (legacy_owner and meta.get("session_id") in (None, "")):
raise SharedVideoUnavailable()
name = meta.get("filename")
if (not isinstance(name, str) or not 1 <= len(name) <= 255 or name in {".", "..", "meta.json"}
or any(ord(c) < 32 or ord(c) == 127 or c in "/\\" for c in name)
or len(name.encode("utf-8")) > 1024 or meta.get("path") != str(root / file_id / name)):
raise SharedVideoUnavailable()
guessed = EXTENSIONS.get(Path(name).suffix.lower())
mime = str(meta.get("content_type") or "").split(";", 1)[0].strip().lower()
if mime in {"", "application/octet-stream", "binary/octet-stream"}:
mime = guessed
if mime not in VIDEO_TYPES or mime != guessed:
raise SharedVideoUnavailable()
video_fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory_fd)
info = os.fstat(video_fd)
_regular(info, owner)
if type(meta.get("size")) is not int or meta["size"] != info.st_size or not 0 < info.st_size <= (1 << 53) - 1:
raise SharedVideoUnavailable()
revision = hashlib.sha256(json.dumps([name, mime, meta.get("session_id"), info.st_dev, info.st_ino,
info.st_size, info.st_mtime_ns, info.st_ctime_ns, info.st_uid, info.st_mode],
ensure_ascii=True, separators=(",", ":")).encode("ascii")).hexdigest()
result = {"file_fd": video_fd, "filename": name, "content_type": mime, "size": info.st_size, "revision": revision,
"file_revision": (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)}
video_fd = None
return result
except (OSError, ValueError, TypeError, KeyError, UnicodeError, RecursionError):
raise SharedVideoUnavailable() from None
finally:
for fd in (video_fd, meta_fd, directory_fd, root_fd):
if fd is not None:
os.close(fd)
def shared_chat_video_descriptor(root, secret, session_id, file_id, *, legacy_owner=False):
key = _key(secret)
opened = _open_registered(root, session_id, file_id, legacy_owner=legacy_owner)
try:
payload = json.dumps([1, session_id, file_id, opened["revision"]], separators=(",", ":")).encode("ascii")
encoded = base64.urlsafe_b64encode(payload).rstrip(b"=").decode("ascii")
identity = "video_" + encoded + "." + hmac.new(key, encoded.encode("ascii"), hashlib.sha256).hexdigest()
return normalize_shared_chat_videos([{key: opened[key] for key in ("filename", "content_type", "size")} | {"id": identity}])[0]
finally:
os.close(opened["file_fd"])
def open_shared_chat_video(root, secret, session_id, identity):
opened = None
try:
if not isinstance(identity, str) or len(identity) > 1024 or VIDEO_ID.fullmatch(identity) is None:
raise SharedVideoUnavailable()
encoded, signature = identity[len("video_"):].split(".")
expected = hmac.new(_key(secret), encoded.encode("ascii"), hashlib.sha256).hexdigest()
if not hmac.compare_digest(signature, expected):
raise SharedVideoUnavailable()
payload = json.loads(base64.b64decode(encoded + "=" * (-len(encoded) % 4), altchars=b"-_", validate=True))
if (not isinstance(payload, list) or len(payload) != 4 or type(payload[0]) is not int or payload[0] != 1
or payload[1] != session_id or not isinstance(payload[3], str)):
raise SharedVideoUnavailable()
opened = _open_registered(root, session_id, payload[2], legacy_owner=True)
if not hmac.compare_digest(opened.pop("revision"), payload[3]):
raise SharedVideoUnavailable()
result, opened = opened, None
return result
except (ValueError, TypeError, KeyError, UnicodeError, RecursionError):
raise SharedVideoUnavailable() from None
finally:
if opened is not None:
os.close(opened["file_fd"])