Spaces:
Runtime error
Runtime error
File size: 3,063 Bytes
148af80 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 | """A small, self-contained gate for a protected Space share link."""
from __future__ import annotations
import hmac
import os
from hashlib import sha256
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import PlainTextResponse, RedirectResponse, Response
COOKIE_NAME = "jev_share"
COOKIE_MAX_AGE = 30 * 24 * 60 * 60
class ShareAccessMiddleware(BaseHTTPMiddleware):
"""Require a secret URL once, then a non-reversible signed cookie."""
def __init__(self, app: object) -> None:
super().__init__(app)
value = os.environ.get("JEV_SHARE_KEY", "")
self._secret = value.encode("utf-8")
self._enabled = bool(value)
# A configured but weak secret is never treated as an open Space.
self._misconfigured = self._enabled and len(self._secret) < 32
self._hf_missing_secret = bool(os.environ.get("SPACE_ID")) and not self._enabled
self._cookie_value = hmac.new(self._secret, b"mocovoice-audio-jev/share-cookie/v1", sha256).hexdigest() if self._enabled else ""
def _headers(self, response: Response, *, cacheable_model: bool = False) -> Response:
response.headers["X-Robots-Tag"] = "noindex, nofollow"
response.headers["Referrer-Policy"] = "no-referrer"
response.headers["Cache-Control"] = "private, max-age=86400" if cacheable_model else "private, no-store"
return response
def _denied(self, status_code: int = 404) -> Response:
# This deliberately has no application, asset, or API information.
return self._headers(PlainTextResponse("共有リンクが必要です。", status_code=status_code))
async def dispatch(self, request: Request, call_next: object) -> Response:
path = request.url.path
# Crawlers may receive only this harmless declaration without access.
if path == "/robots.txt":
return self._headers(PlainTextResponse("User-agent: *\nDisallow: /\n", media_type="text/plain"))
if self._hf_missing_secret or self._misconfigured:
return self._headers(PlainTextResponse("Service unavailable.", status_code=503))
if self._enabled:
supplied = request.query_params.get("share")
if request.method == "GET" and path == "/" and supplied is not None:
if hmac.compare_digest(supplied.encode("utf-8"), self._secret):
response = RedirectResponse(url="/", status_code=303)
response.set_cookie(
COOKIE_NAME, self._cookie_value, max_age=COOKIE_MAX_AGE,
secure=True, httponly=True, samesite="lax", path="/",
)
return self._headers(response)
return self._denied()
if not hmac.compare_digest(request.cookies.get(COOKIE_NAME, ""), self._cookie_value):
return self._denied()
response = await call_next(request)
return self._headers(response, cacheable_model=path.startswith("/models/"))
|