Spaces:
Runtime error
Runtime error
| """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/")) | |