interactive_chat_shares.py
Rama main · réplica de lectura del repositorio original.
"""Durable reusable access tokens and browser capabilities for one exact chat.
This ledger grants no provider or native API authority. Raw invitation/browser
tokens are returned once and never persisted. Upload references stay private.
"""
from __future__ import annotations
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import secrets
import time
from public_chat_shares import (
PublicChatShareStore, PublicChatShareUnavailable, PublicChatShareValidationError,
TOKEN_PATTERN, _timestamp, _utf8_size,
)
Unavailable = PublicChatShareUnavailable
ValidationError = PublicChatShareValidationError
SHARE_ID = re.compile(r"interactive_[a-f0-9]{32}\Z")
UPLOAD_ID = re.compile(r"upload_[a-f0-9]{32}\Z")
MAX_UPLOAD_BYTES = 8 * 1024 * 1024
MAX_SHARE_UPLOAD_BYTES = 64 * 1024 * 1024
MAX_PROMPT_BYTES = 64 * 1024
class Conflict(ValueError):
"""A stable request ID already names different or indeterminate work."""
def token_hash(token):
if not isinstance(token, str) or TOKEN_PATTERN.fullmatch(token) is None:
raise Unavailable()
return hashlib.sha256(token.encode("ascii")).digest()
def csrf_token(browser_token):
token_hash(browser_token)
return hmac.new(browser_token.encode("ascii"), b"interactive-chat-csrf-v1", hashlib.sha256).hexdigest()
class InteractiveChatShareStore(PublicChatShareStore):
"""Reuse the existing owner-only SQLite connection checks, not snapshot data."""
def __init__(self, storage_root, *, now=time.time):
self.storage_root = Path(storage_root)
if not self.storage_root.is_absolute():
raise ValidationError("Share storage must be an absolute private directory")
self.storage_root.mkdir(mode=0o700, parents=True, exist_ok=True)
self.database_path = self.storage_root / "interactive.sqlite3"
self._now = now
self._check_directory()
try:
fd = os.open(self.database_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600)
except FileExistsError:
self._check_database_stat(self.database_path.lstat())
else:
os.close(fd)
with self._connection(write=True) as db:
db.execute("BEGIN IMMEDIATE")
if db.execute("PRAGMA user_version").fetchone()[0] not in (0, 1):
raise OSError("Unsupported interactive share database version")
db.execute("""CREATE TABLE IF NOT EXISTS interactive_shares(
id TEXT PRIMARY KEY,session_id TEXT NOT NULL,title TEXT NOT NULL,
created_at REAL NOT NULL,expires_at REAL,revoked_at REAL,redeemed_at REAL,
invite_hash BLOB NOT NULL UNIQUE,browser_hash BLOB UNIQUE)""")
db.execute("CREATE INDEX IF NOT EXISTS interactive_share_session ON interactive_shares(session_id,created_at DESC)")
# Additive migration: old grants keep their configured-origin behavior.
# Never run migrations from anonymous read/open_existing requests.
if "public_origin" not in {row[1] for row in db.execute("PRAGMA table_info(interactive_shares)")}:
db.execute("ALTER TABLE interactive_shares ADD COLUMN public_origin TEXT NOT NULL DEFAULT ''")
db.execute("""CREATE TABLE IF NOT EXISTS interactive_uploads(
id TEXT PRIMARY KEY,share_id TEXT NOT NULL REFERENCES interactive_shares(id),
name TEXT NOT NULL,media_type TEXT NOT NULL,byte_size INTEGER NOT NULL,
private_ref TEXT,created_at REAL NOT NULL)""")
db.execute("CREATE INDEX IF NOT EXISTS interactive_upload_share ON interactive_uploads(share_id)")
db.execute("""CREATE TABLE IF NOT EXISTS interactive_submissions(
share_id TEXT NOT NULL REFERENCES interactive_shares(id),request_id TEXT NOT NULL,
fingerprint BLOB NOT NULL,receipt_json TEXT,created_at REAL NOT NULL,
PRIMARY KEY(share_id,request_id))""")
db.execute("PRAGMA user_version=1")
@classmethod
def open_existing(cls, storage_root, *, now=time.time):
instance = cls.__new__(cls)
instance.storage_root = Path(storage_root)
instance.database_path = instance.storage_root / "interactive.sqlite3"
instance._now = now
with instance._connection() as db:
if db.execute("PRAGMA user_version").fetchone()[0] != 1:
raise OSError("Unsupported interactive share database version")
return instance
@staticmethod
def metadata(row):
return {key: row[key] for key in ("id", "title", "created_at", "expires_at", "revoked_at", "redeemed_at")}
def create_share(self, session_id, *, title=None, expires_at=None, public_origin=""):
if not isinstance(session_id, str) or re.fullmatch(r"[A-Za-z0-9_-]{1,128}", session_id) is None:
raise ValidationError("Invalid chat")
title = "Shared conversation" if title is None else title
if not isinstance(title, str) or not title.strip() or len(title) > 256:
raise ValidationError("Title must contain 1 to 256 characters")
_utf8_size(title, "Title", 1024)
now = _timestamp(self._now(), "current time")
if expires_at is not None and _timestamp(expires_at, "expires_at") <= now:
raise ValidationError("Expiry must be in the future")
invite = secrets.token_urlsafe(32)
share_id = "interactive_" + secrets.token_hex(16)
with self._connection(write=True) as db:
db.execute("INSERT INTO interactive_shares(id,session_id,title,created_at,expires_at,invite_hash,public_origin) VALUES(?,?,?,?,?,?,?)",
(share_id, session_id, title, now, expires_at, token_hash(invite), public_origin))
row = db.execute("SELECT * FROM interactive_shares WHERE id=?", (share_id,)).fetchone()
return {**self.metadata(row), "invitation_token": invite}
def share_origin(self, share_id):
with self._connection() as db:
row = db.execute("SELECT * FROM interactive_shares WHERE id=?", (share_id,)).fetchone()
if row is None:
raise Unavailable()
return row["public_origin"] if "public_origin" in row.keys() else ""
def list_shares(self, session_id):
with self._connection() as db:
return [self.metadata(row) for row in db.execute(
"SELECT * FROM interactive_shares WHERE session_id=? ORDER BY created_at DESC,id DESC LIMIT 100", (session_id,))]
def revoke_share(self, share_id, *, session_id):
with self._connection(write=True) as db:
return db.execute("UPDATE interactive_shares SET revoked_at=COALESCE(revoked_at,?) WHERE id=? AND session_id=?",
(self._now(), share_id, session_id)).rowcount == 1
def redeem(self, share_id, invite):
digest = token_hash(invite)
now = self._now()
with self._connection(write=True) as db:
db.execute("BEGIN IMMEDIATE")
# One reusable token admits multiple browsers. Preserve browser_hash
# so cookies issued by older releases remain valid after an upgrade.
updated = db.execute("""UPDATE interactive_shares SET redeemed_at=COALESCE(redeemed_at,?)
WHERE id=? AND invite_hash=? AND revoked_at IS NULL
AND (expires_at IS NULL OR expires_at>?)""", (now, share_id, digest, now))
if updated.rowcount != 1:
raise Unavailable()
return invite
def _authorized(self, db, share_id, browser_token):
digest = token_hash(browser_token)
row = db.execute("""SELECT * FROM interactive_shares WHERE id=? AND (browser_hash=? OR invite_hash=?)
AND redeemed_at IS NOT NULL AND revoked_at IS NULL AND (expires_at IS NULL OR expires_at>?)""",
(share_id, digest, digest, self._now())).fetchone()
if row is None:
raise Unavailable()
return row
def authenticate(self, share_id, browser_token):
with self._connection() as db:
row = self._authorized(db, share_id, browser_token)
return {**self.metadata(row), "session_id": row["session_id"]}
def reserve_upload(self, share_id, browser_token, *, name, media_type, byte_size):
if (not isinstance(name, str) or not 1 <= len(name) <= 160 or name in {".", ".."}
or any(ord(c) < 32 or c in "/\\" for c in name)):
raise ValidationError("Use a simple filename without directories")
_utf8_size(name, "Filename", 640)
if not isinstance(media_type, str) or re.fullmatch(r"[a-zA-Z0-9!#$&^_.+-]+/[a-zA-Z0-9!#$&^_.+-]+", media_type) is None:
raise ValidationError("Invalid upload media type")
if type(byte_size) is not int or not 1 <= byte_size <= MAX_UPLOAD_BYTES:
raise ValidationError("Files must contain 1 byte to 8 MiB")
upload_id = "upload_" + secrets.token_hex(16)
with self._connection(write=True) as db:
db.execute("BEGIN IMMEDIATE")
self._authorized(db, share_id, browser_token)
used = db.execute("SELECT COALESCE(SUM(byte_size),0) FROM interactive_uploads WHERE share_id=?", (share_id,)).fetchone()[0]
if used + byte_size > MAX_SHARE_UPLOAD_BYTES:
raise ValidationError("This share has reached its 64 MiB upload storage bound")
db.execute("INSERT INTO interactive_uploads VALUES(?,?,?,?,?,?,?)",
(upload_id, share_id, name, media_type, byte_size, None, self._now()))
return upload_id
def complete_upload(self, share_id, browser_token, upload_id, private_ref):
if not isinstance(private_ref, str) or not private_ref or len(private_ref) > 1024:
raise ValidationError("Upload was not accepted")
with self._connection(write=True) as db:
db.execute("BEGIN IMMEDIATE")
self._authorized(db, share_id, browser_token)
if db.execute("UPDATE interactive_uploads SET private_ref=? WHERE id=? AND share_id=? AND private_ref IS NULL",
(private_ref, upload_id, share_id)).rowcount != 1:
raise Unavailable()
def abandon_upload(self, share_id, upload_id):
with self._connection(write=True) as db:
db.execute("DELETE FROM interactive_uploads WHERE id=? AND share_id=? AND private_ref IS NULL", (upload_id, share_id))
def upload_refs(self, share_id, browser_token, upload_ids):
if (not isinstance(upload_ids, list) or len(upload_ids) > 4 or len(set(str(i) for i in upload_ids)) != len(upload_ids)
or any(not isinstance(i, str) or UPLOAD_ID.fullmatch(i) is None for i in upload_ids)):
raise ValidationError("Select at most four distinct uploads from this share")
with self._connection() as db:
self._authorized(db, share_id, browser_token)
refs = []
for upload_id in upload_ids:
row = db.execute("SELECT private_ref FROM interactive_uploads WHERE id=? AND share_id=? AND private_ref IS NOT NULL",
(upload_id, share_id)).fetchone()
if row is None:
raise Unavailable()
refs.append(row["private_ref"])
return refs
def reserve_submission(self, share_id, browser_token, request_id, prompt, refs, *, operation="prompt"):
fingerprint = hashlib.sha256(json.dumps([operation, prompt, refs], ensure_ascii=False, separators=(",", ":")).encode("utf-8")).digest()
with self._connection(write=True) as db:
db.execute("BEGIN IMMEDIATE")
self._authorized(db, share_id, browser_token)
row = db.execute("SELECT fingerprint,receipt_json FROM interactive_submissions WHERE share_id=? AND request_id=?",
(share_id, request_id)).fetchone()
if row is not None:
if not hmac.compare_digest(bytes(row["fingerprint"]), fingerprint):
raise Conflict("This request ID was already used for a different message")
if row["receipt_json"] is None:
raise Conflict("Message acceptance is indeterminate. Inspect the chat before sending another message; this request will not be retried")
return json.loads(row["receipt_json"])
db.execute("INSERT INTO interactive_submissions VALUES(?,?,?,?,?)",
(share_id, request_id, fingerprint, None, self._now()))
return None
def accept_submission(self, share_id, request_id, receipt):
# Persist an acceptance fact even if expiry occurred during the callback.
# A failure leaves the existing pending receipt indeterminate, not retryable.
encoded = json.dumps(receipt, separators=(",", ":"), allow_nan=False)
with self._connection(write=True) as db:
if db.execute("UPDATE interactive_submissions SET receipt_json=? WHERE share_id=? AND request_id=? AND receipt_json IS NULL",
(encoded, share_id, request_id)).rowcount != 1:
raise Conflict("Message receipt could not be recorded")