feat: bind api downloads to verified principal scope
Hermes Agent
parent
04fdad3053
commit
04b69b3d7c
|
|
@ -724,7 +724,11 @@ def _resolve_media_to_data_urls(text: str) -> str:
|
|||
return text
|
||||
|
||||
|
||||
def _rewrite_local_artifact_references_for_remote_client(text: str) -> str:
|
||||
def _rewrite_local_artifact_references_for_remote_client(
|
||||
text: str,
|
||||
*,
|
||||
owner_metadata: Optional[Dict[str, Any]] = None,
|
||||
) -> str:
|
||||
"""Turn host-local file references into managed HTTPS download URLs.
|
||||
|
||||
Open WebUI / API-server clients cannot open ``file://`` URIs or host-local
|
||||
|
|
@ -759,7 +763,7 @@ def _rewrite_local_artifact_references_for_remote_client(text: str) -> str:
|
|||
rewritten[key] = ""
|
||||
return None
|
||||
try:
|
||||
url = str(build_public_download_url(safe_path) or "").strip()
|
||||
url = str(build_public_download_url(safe_path, owner_metadata=owner_metadata) or "").strip()
|
||||
except Exception:
|
||||
url = ""
|
||||
rewritten[key] = url
|
||||
|
|
@ -2362,12 +2366,29 @@ class APIServerAdapter(BasePlatformAdapter):
|
|||
live_email = self._normalize_api_identity_text(store.get_verified_email_for_source(source) or "").lower()
|
||||
if live_email != email:
|
||||
raise PermissionError("verified_email_mismatch")
|
||||
principal_context = store.get_principal_context(source) or {}
|
||||
principal_id = self._normalize_api_identity_text(principal_context.get("principal_id") or "")
|
||||
if principal_id:
|
||||
from gateway.principal_profiles import ensure_principal_profile
|
||||
|
||||
principal_profile = ensure_principal_profile(principal_id)
|
||||
requested_profile = str(_api_request_profile.get() or "").strip()
|
||||
if requested_profile and requested_profile != principal_profile:
|
||||
raise PermissionError("principal_profile_mismatch")
|
||||
identity_context["principal_id"] = principal_id
|
||||
identity_context["profile"] = principal_profile
|
||||
except ValueError as exc:
|
||||
code = "email_domain_not_allowed" if str(exc) == "invalid_domain" else "verified_identity_conflict"
|
||||
return identity_context, web.json_response(
|
||||
_openai_error("Forbidden: verified identity required", code=code),
|
||||
status=403,
|
||||
)
|
||||
except PermissionError as exc:
|
||||
code = str(exc) or "verified_identity_conflict"
|
||||
return identity_context, web.json_response(
|
||||
_openai_error("Forbidden: verified identity required", code=code),
|
||||
status=403,
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.warning("api_server verified identity gate failed for user_id=%s email=%s: %s", user_id, email, exc)
|
||||
return identity_context, web.json_response(
|
||||
|
|
@ -2377,6 +2398,32 @@ class APIServerAdapter(BasePlatformAdapter):
|
|||
|
||||
return identity_context, None
|
||||
|
||||
@staticmethod
|
||||
def _effective_identity_profile(identity_context: Optional[Dict[str, Optional[str]]] = None) -> Optional[str]:
|
||||
identity_context = identity_context or {}
|
||||
profile = str(identity_context.get("profile") or "").strip()
|
||||
if profile:
|
||||
return profile
|
||||
fallback = str(_api_request_profile.get() or "").strip()
|
||||
return fallback or None
|
||||
|
||||
@staticmethod
|
||||
def _managed_download_owner_metadata(
|
||||
identity_context: Optional[Dict[str, Optional[str]]] = None,
|
||||
) -> Dict[str, str]:
|
||||
identity_context = identity_context or {}
|
||||
metadata: Dict[str, str] = {}
|
||||
for src_key, out_key in (
|
||||
("email", "email"),
|
||||
("principal_id", "principal_id"),
|
||||
("user_id", "external_user_id"),
|
||||
("profile", "profile"),
|
||||
):
|
||||
value = str(identity_context.get(src_key) or "").strip()
|
||||
if value:
|
||||
metadata[out_key] = value
|
||||
return metadata
|
||||
|
||||
@staticmethod
|
||||
def _session_owner_from_request(request: "web.Request") -> Optional[str]:
|
||||
_ = request
|
||||
|
|
@ -2672,7 +2719,8 @@ class APIServerAdapter(BasePlatformAdapter):
|
|||
)
|
||||
effective_session_id = result.get("session_id") if isinstance(result, dict) else session_id
|
||||
final_response = _rewrite_local_artifact_references_for_remote_client(
|
||||
_resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else "")
|
||||
_resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else ""),
|
||||
owner_metadata=self._managed_download_owner_metadata(identity_context),
|
||||
)
|
||||
headers = {"X-Hermes-Session-Id": effective_session_id or session_id}
|
||||
if gateway_session_key:
|
||||
|
|
@ -2769,7 +2817,8 @@ class APIServerAdapter(BasePlatformAdapter):
|
|||
identity_context=identity_context,
|
||||
)
|
||||
final_response = _rewrite_local_artifact_references_for_remote_client(
|
||||
_resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else "")
|
||||
_resolve_media_to_data_urls(result.get("final_response", "") if isinstance(result, dict) else ""),
|
||||
owner_metadata=self._managed_download_owner_metadata(identity_context),
|
||||
)
|
||||
effective_session_id = result.get("session_id", session_id) if isinstance(result, dict) else session_id
|
||||
turn_messages = self._turn_transcript_messages(history, user_message, result) if isinstance(result, dict) else []
|
||||
|
|
@ -3104,7 +3153,8 @@ class APIServerAdapter(BasePlatformAdapter):
|
|||
)
|
||||
|
||||
final_response = _rewrite_local_artifact_references_for_remote_client(
|
||||
_resolve_media_to_data_urls(result.get("final_response") or "")
|
||||
_resolve_media_to_data_urls(result.get("final_response") or ""),
|
||||
owner_metadata=self._managed_download_owner_metadata(identity_context),
|
||||
)
|
||||
is_partial = bool(result.get("partial"))
|
||||
is_failed = bool(result.get("failed"))
|
||||
|
|
@ -4214,7 +4264,8 @@ class APIServerAdapter(BasePlatformAdapter):
|
|||
)
|
||||
|
||||
final_response = _rewrite_local_artifact_references_for_remote_client(
|
||||
_resolve_media_to_data_urls(result.get("final_response", ""))
|
||||
_resolve_media_to_data_urls(result.get("final_response", "")),
|
||||
owner_metadata=self._managed_download_owner_metadata(identity_context),
|
||||
)
|
||||
if not final_response:
|
||||
final_response = _redact_api_error_text(result.get("error", "(No response generated)"))
|
||||
|
|
@ -4870,7 +4921,7 @@ class APIServerAdapter(BasePlatformAdapter):
|
|||
# Capture before hopping to the executor — ContextVars do not follow
|
||||
# run_in_executor threads, so the profile scope must be re-entered
|
||||
# inside _run() from this explicit value.
|
||||
request_profile = _api_request_profile.get()
|
||||
request_profile = self._effective_identity_profile(identity_context)
|
||||
identity_context = identity_context or {}
|
||||
|
||||
def _run():
|
||||
|
|
@ -5125,7 +5176,7 @@ class APIServerAdapter(BasePlatformAdapter):
|
|||
route = self._resolve_route(body.get("model"))
|
||||
# Background task outlives the HTTP response (and thus the middleware
|
||||
# profile scope). Capture now and re-enter inside the task/executor.
|
||||
request_profile = _api_request_profile.get()
|
||||
request_profile = self._effective_identity_profile(identity_context)
|
||||
|
||||
async def _run_and_close():
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -101,11 +101,16 @@ def _connect_short_links_db() -> sqlite3.Connection:
|
|||
short_id TEXT PRIMARY KEY,
|
||||
path TEXT NOT NULL,
|
||||
exp INTEGER NOT NULL,
|
||||
created_at INTEGER NOT NULL
|
||||
created_at INTEGER NOT NULL,
|
||||
metadata TEXT NOT NULL DEFAULT '{}'
|
||||
)
|
||||
'''
|
||||
)
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_short_links_exp ON short_links(exp)")
|
||||
cols = {str(row[1]) for row in conn.execute("PRAGMA table_info(short_links)").fetchall()}
|
||||
if "metadata" not in cols:
|
||||
conn.execute("ALTER TABLE short_links ADD COLUMN metadata TEXT NOT NULL DEFAULT '{}' ")
|
||||
conn.commit()
|
||||
return conn
|
||||
|
||||
|
||||
|
|
@ -165,7 +170,23 @@ def _b64url_decode(text: str) -> bytes:
|
|||
return base64.urlsafe_b64decode(padded.encode("ascii"))
|
||||
|
||||
|
||||
def create_signed_download_token(file_path: str, *, expiry_seconds: int = _DEFAULT_EXPIRY_SECONDS) -> str:
|
||||
def _normalize_owner_metadata(owner_metadata: dict | None) -> dict[str, str]:
|
||||
if not isinstance(owner_metadata, dict):
|
||||
return {}
|
||||
allowed = {}
|
||||
for key in ("email", "principal_id", "external_user_id", "profile"):
|
||||
value = str(owner_metadata.get(key) or "").strip()
|
||||
if value:
|
||||
allowed[key] = value
|
||||
return allowed
|
||||
|
||||
|
||||
def create_signed_download_token(
|
||||
file_path: str,
|
||||
*,
|
||||
expiry_seconds: int = _DEFAULT_EXPIRY_SECONDS,
|
||||
owner_metadata: dict | None = None,
|
||||
) -> str:
|
||||
staged = stage_public_download(file_path)
|
||||
secret = _download_secret()
|
||||
if staged is None or not secret:
|
||||
|
|
@ -174,6 +195,7 @@ def create_signed_download_token(file_path: str, *, expiry_seconds: int = _DEFAU
|
|||
payload = {
|
||||
"path": str(staged),
|
||||
"exp": int(time.time()) + max(60, int(expiry_seconds or _DEFAULT_EXPIRY_SECONDS)),
|
||||
"owner": _normalize_owner_metadata(owner_metadata),
|
||||
}
|
||||
payload_text = json.dumps(payload, separators=(",", ":"), sort_keys=True)
|
||||
payload_b64 = _b64url_encode(payload_text.encode("utf-8"))
|
||||
|
|
@ -181,7 +203,12 @@ def create_signed_download_token(file_path: str, *, expiry_seconds: int = _DEFAU
|
|||
return f"{payload_b64}.{_b64url_encode(signature)}"
|
||||
|
||||
|
||||
def create_short_download_id(file_path: str, *, expiry_seconds: int = _DEFAULT_EXPIRY_SECONDS) -> str:
|
||||
def create_short_download_id(
|
||||
file_path: str,
|
||||
*,
|
||||
expiry_seconds: int = _DEFAULT_EXPIRY_SECONDS,
|
||||
owner_metadata: dict | None = None,
|
||||
) -> str:
|
||||
staged = stage_public_download(file_path)
|
||||
if staged is None:
|
||||
return ""
|
||||
|
|
@ -201,8 +228,8 @@ def create_short_download_id(file_path: str, *, expiry_seconds: int = _DEFAULT_E
|
|||
if existing is not None:
|
||||
continue
|
||||
conn.execute(
|
||||
"INSERT INTO short_links (short_id, path, exp, created_at) VALUES (?, ?, ?, ?)",
|
||||
(short_id, str(staged), expiry, int(time.time())),
|
||||
"INSERT INTO short_links (short_id, path, exp, created_at, metadata) VALUES (?, ?, ?, ?, ?)",
|
||||
(short_id, str(staged), expiry, int(time.time()), json.dumps(_normalize_owner_metadata(owner_metadata), separators=(",", ":"), sort_keys=True)),
|
||||
)
|
||||
conn.commit()
|
||||
return short_id
|
||||
|
|
@ -211,11 +238,15 @@ def create_short_download_id(file_path: str, *, expiry_seconds: int = _DEFAULT_E
|
|||
return ""
|
||||
|
||||
|
||||
def resolve_signed_download_token(token: str) -> Path:
|
||||
def _coerce_record_payload(path: Path, metadata: dict | None = None) -> dict:
|
||||
return {"path": path, "metadata": _normalize_owner_metadata(metadata), "owner": _normalize_owner_metadata(metadata)}
|
||||
|
||||
|
||||
def resolve_signed_download_record(token: str) -> dict:
|
||||
raw = str(token or "").strip()
|
||||
if not raw or "." not in raw:
|
||||
if raw.startswith(_SHORT_DOWNLOADS_TOKEN_PREFIX):
|
||||
return resolve_short_download_id(raw[len(_SHORT_DOWNLOADS_TOKEN_PREFIX) :])
|
||||
return resolve_short_download_record(raw[len(_SHORT_DOWNLOADS_TOKEN_PREFIX) :])
|
||||
raise ValueError("invalid_download_token")
|
||||
payload_b64, signature_b64 = raw.split(".", 1)
|
||||
secret = _download_secret()
|
||||
|
|
@ -237,10 +268,14 @@ def resolve_signed_download_token(token: str) -> Path:
|
|||
raise ValueError("download_target_outside_stage_dir")
|
||||
if not target.exists() or not target.is_file():
|
||||
raise FileNotFoundError(str(target))
|
||||
return target
|
||||
return _coerce_record_payload(target, payload.get("owner"))
|
||||
|
||||
|
||||
def resolve_short_download_id(short_id: str) -> Path:
|
||||
def resolve_signed_download_token(token: str) -> Path:
|
||||
return Path(resolve_signed_download_record(token)["path"])
|
||||
|
||||
|
||||
def resolve_short_download_record(short_id: str) -> dict:
|
||||
raw = str(short_id or "").strip()
|
||||
if not raw:
|
||||
raise ValueError("invalid_short_download_id")
|
||||
|
|
@ -248,7 +283,7 @@ def resolve_short_download_id(short_id: str) -> Path:
|
|||
conn = _connect_short_links_db()
|
||||
try:
|
||||
row = conn.execute(
|
||||
"SELECT path, exp FROM short_links WHERE short_id = ?",
|
||||
"SELECT path, exp, metadata FROM short_links WHERE short_id = ?",
|
||||
(raw,),
|
||||
).fetchone()
|
||||
finally:
|
||||
|
|
@ -266,16 +301,25 @@ def resolve_short_download_id(short_id: str) -> Path:
|
|||
raise ValueError("download_target_outside_stage_dir")
|
||||
if not target.exists() or not target.is_file():
|
||||
raise FileNotFoundError(str(target))
|
||||
return target
|
||||
metadata = {}
|
||||
try:
|
||||
metadata = json.loads(str(row[2] or "{}")) if row[2] else {}
|
||||
except Exception:
|
||||
metadata = {}
|
||||
return _coerce_record_payload(target, metadata)
|
||||
|
||||
|
||||
def build_public_download_url(file_path: str) -> str:
|
||||
def resolve_short_download_id(short_id: str) -> Path:
|
||||
return Path(resolve_short_download_record(short_id)["path"])
|
||||
|
||||
|
||||
def build_public_download_url(file_path: str, *, owner_metadata: dict | None = None) -> str:
|
||||
base_url = str(resolve_download_public_url() or "").rstrip("/")
|
||||
if not base_url:
|
||||
return ""
|
||||
short_id = create_short_download_id(file_path)
|
||||
short_id = create_short_download_id(file_path, owner_metadata=owner_metadata)
|
||||
if not short_id:
|
||||
return ""
|
||||
# Preflight: prove the short link resolves before we hand it out.
|
||||
resolve_short_download_id(short_id)
|
||||
resolve_short_download_record(short_id)
|
||||
return f"{base_url}{_SHORT_DOWNLOADS_PREFIX}/{quote(f'{_SHORT_DOWNLOADS_TOKEN_PREFIX}{short_id}', safe='')}"
|
||||
|
|
|
|||
|
|
@ -750,6 +750,35 @@ def _deidentify_export_payload(value: Any) -> Any:
|
|||
return value
|
||||
|
||||
|
||||
def _managed_download_visible_to_scope(metadata: dict[str, Any] | None, scope: dict[str, Any]) -> bool:
|
||||
metadata = metadata or {}
|
||||
if not metadata:
|
||||
return True
|
||||
if scope.get("admin"):
|
||||
return True
|
||||
email = str(metadata.get("email") or "").strip().lower()
|
||||
principal_id = str(metadata.get("principal_id") or "").strip()
|
||||
external_user_id = str(metadata.get("external_user_id") or "").strip()
|
||||
profile = str(metadata.get("profile") or "").strip()
|
||||
if email and email == str(scope.get("email") or "").strip().lower():
|
||||
return True
|
||||
if principal_id and principal_id in set(scope.get("principal_ids") or []):
|
||||
return True
|
||||
if external_user_id and external_user_id in set(scope.get("external_user_ids") or []):
|
||||
return True
|
||||
if profile and profile in _allowed_profile_names_for_scope(scope):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _enforce_managed_download_access(request: Request, metadata: dict[str, Any] | None) -> None:
|
||||
if not metadata:
|
||||
return
|
||||
scope = _dashboard_identity_scope(request)
|
||||
if not _managed_download_visible_to_scope(metadata, scope):
|
||||
raise HTTPException(status_code=403, detail="forbidden")
|
||||
|
||||
|
||||
def _verification_result_html(success: bool, message: str) -> str:
|
||||
title = "驗證成功" if success else "驗證失敗"
|
||||
accent = "#16a34a" if success else "#dc2626"
|
||||
|
|
@ -16897,16 +16926,19 @@ def mount_spa(application: FastAPI):
|
|||
|
||||
@application.get("/downloads/{token}")
|
||||
async def serve_managed_download(token: str, request: Request):
|
||||
from hermes_cli.managed_downloads import resolve_signed_download_token
|
||||
from hermes_cli.managed_downloads import resolve_signed_download_record
|
||||
|
||||
raw_token = str(token or "").strip()
|
||||
accept = str(request.headers.get("accept", "") or "").lower()
|
||||
if raw_token.startswith("s-") and "text/html" in accept:
|
||||
short_id = raw_token[2:]
|
||||
try:
|
||||
resolve_signed_download_token(raw_token)
|
||||
record = resolve_signed_download_record(raw_token)
|
||||
_enforce_managed_download_access(request, record.get("metadata"))
|
||||
except FileNotFoundError:
|
||||
return Response(content="找不到可下載的檔案。", media_type="text/plain; charset=utf-8", status_code=404)
|
||||
except HTTPException as exc:
|
||||
return Response(content="你沒有權限下載這個檔案。", media_type="text/plain; charset=utf-8", status_code=exc.status_code)
|
||||
except ValueError as exc:
|
||||
detail = str(exc)
|
||||
status = 410 if detail == "expired_download_token" else 404
|
||||
|
|
@ -16945,9 +16977,13 @@ def mount_spa(application: FastAPI):
|
|||
)
|
||||
|
||||
try:
|
||||
file_path = resolve_signed_download_token(token)
|
||||
record = resolve_signed_download_record(token)
|
||||
_enforce_managed_download_access(request, record.get("metadata"))
|
||||
file_path = record["path"]
|
||||
except FileNotFoundError:
|
||||
return Response(content="找不到可下載的檔案。", media_type="text/plain; charset=utf-8", status_code=404)
|
||||
except HTTPException as exc:
|
||||
return Response(content="你沒有權限下載這個檔案。", media_type="text/plain; charset=utf-8", status_code=exc.status_code)
|
||||
except ValueError as exc:
|
||||
detail = str(exc)
|
||||
status = 410 if detail == "expired_download_token" else 404
|
||||
|
|
@ -16961,13 +16997,17 @@ def mount_spa(application: FastAPI):
|
|||
)
|
||||
|
||||
@application.get("/downloads/s/{short_id}")
|
||||
async def serve_short_managed_download(short_id: str):
|
||||
from hermes_cli.managed_downloads import resolve_short_download_id
|
||||
async def serve_short_managed_download(short_id: str, request: Request):
|
||||
from hermes_cli.managed_downloads import resolve_short_download_record
|
||||
|
||||
try:
|
||||
file_path = resolve_short_download_id(short_id)
|
||||
record = resolve_short_download_record(short_id)
|
||||
_enforce_managed_download_access(request, record.get("metadata"))
|
||||
file_path = record["path"]
|
||||
except FileNotFoundError:
|
||||
return Response(content="找不到可下載的檔案。", media_type="text/plain; charset=utf-8", status_code=404)
|
||||
except HTTPException as exc:
|
||||
return Response(content="你沒有權限下載這個檔案。", media_type="text/plain; charset=utf-8", status_code=exc.status_code)
|
||||
except ValueError as exc:
|
||||
detail = str(exc)
|
||||
status = 410 if detail == "expired_download_token" else 404
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ from gateway.platforms.api_server import (
|
|||
_PROFILE_REJECTED,
|
||||
_api_request_profile,
|
||||
)
|
||||
import types
|
||||
|
||||
|
||||
def _make_adapter(multiplex: bool = True) -> APIServerAdapter:
|
||||
|
|
@ -84,3 +85,56 @@ class TestApiServerModelsUnderProfile:
|
|||
assert adapter._resolve_model_name("") == "coder"
|
||||
finally:
|
||||
_api_request_profile.reset(token_prof)
|
||||
|
||||
|
||||
class TestVerifiedIdentityPrincipalRouting:
|
||||
def test_verified_identity_attaches_principal_profile(self, monkeypatch):
|
||||
adapter = _make_adapter(multiplex=True)
|
||||
monkeypatch.setenv("API_SERVER_REQUIRE_VERIFIED_IDENTITY", "1")
|
||||
|
||||
class _FakeStore:
|
||||
def force_bind_identity(self, **kwargs):
|
||||
return None
|
||||
|
||||
def get_verified_email_for_source(self, source):
|
||||
return "user@bremen.com.tw"
|
||||
|
||||
def get_principal_context(self, source):
|
||||
return {"principal_id": "p-123"}
|
||||
|
||||
monkeypatch.setattr("gateway.platforms.api_server.GatewayUserStore", lambda: _FakeStore())
|
||||
monkeypatch.setattr("gateway.principal_profiles.ensure_principal_profile", lambda principal_id: "principal_p123")
|
||||
req = types.SimpleNamespace(headers={"X-Hermes-User-Id": "owui-42", "X-Hermes-User-Email": "user@bremen.com.tw"})
|
||||
|
||||
identity, err = adapter._verified_identity_context_or_error(req, {"metadata": {}})
|
||||
|
||||
assert err is None
|
||||
assert identity["principal_id"] == "p-123"
|
||||
assert identity["profile"] == "principal_p123"
|
||||
|
||||
def test_verified_identity_rejects_mismatched_prefixed_profile(self, monkeypatch):
|
||||
adapter = _make_adapter(multiplex=True)
|
||||
monkeypatch.setenv("API_SERVER_REQUIRE_VERIFIED_IDENTITY", "1")
|
||||
|
||||
class _FakeStore:
|
||||
def force_bind_identity(self, **kwargs):
|
||||
return None
|
||||
|
||||
def get_verified_email_for_source(self, source):
|
||||
return "user@bremen.com.tw"
|
||||
|
||||
def get_principal_context(self, source):
|
||||
return {"principal_id": "p-123"}
|
||||
|
||||
monkeypatch.setattr("gateway.platforms.api_server.GatewayUserStore", lambda: _FakeStore())
|
||||
monkeypatch.setattr("gateway.principal_profiles.ensure_principal_profile", lambda principal_id: "principal_p123")
|
||||
req = types.SimpleNamespace(headers={"X-Hermes-User-Id": "owui-42", "X-Hermes-User-Email": "user@bremen.com.tw"})
|
||||
token_prof = _api_request_profile.set("other_profile")
|
||||
try:
|
||||
_identity, err = adapter._verified_identity_context_or_error(req, {"metadata": {}})
|
||||
finally:
|
||||
_api_request_profile.reset(token_prof)
|
||||
|
||||
assert err is not None
|
||||
assert err.status == 403
|
||||
assert b"principal_profile_mismatch" in err.body
|
||||
|
|
|
|||
|
|
@ -107,3 +107,58 @@ def test_short_download_browser_route_shows_launch_page(_isolate_hermes_home):
|
|||
assert "text/html" in response.headers.get("content-type", "")
|
||||
assert "下載已開始" in response.text
|
||||
assert f'/downloads/s/{short_id.removeprefix("s-")}' in response.text
|
||||
|
||||
|
||||
def test_owner_bound_short_download_rejects_other_scope(_isolate_hermes_home, monkeypatch):
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.managed_downloads import build_public_download_url
|
||||
from hermes_cli import web_server
|
||||
from hermes_cli.web_server import app
|
||||
|
||||
home = get_hermes_home()
|
||||
(home / "config.yaml").write_text(
|
||||
"dashboard:\n"
|
||||
" public_url: https://hermesadmin.bremen.com.tw\n"
|
||||
" download_public_url: https://hermesadmin.bremen.com.tw\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
source = home / "owner-bound.pdf"
|
||||
source.write_bytes(b"%PDF-1.4 owner-bound")
|
||||
|
||||
url = build_public_download_url(
|
||||
str(source),
|
||||
owner_metadata={"email": "owner@bremen.com.tw", "principal_id": "p-123"},
|
||||
)
|
||||
short_id = Path(urlparse(url).path).name
|
||||
monkeypatch.setattr(
|
||||
web_server,
|
||||
"_dashboard_identity_scope",
|
||||
lambda request: {"admin": False, "email": "other@bremen.com.tw", "principal_ids": ["p-999"], "external_user_ids": [], "platforms": []},
|
||||
)
|
||||
|
||||
client = TestClient(app)
|
||||
response = client.get(f"/downloads/{short_id}")
|
||||
|
||||
assert response.status_code == 403
|
||||
assert "沒有權限" in response.text
|
||||
|
||||
|
||||
def test_owner_metadata_roundtrips_in_short_download_record(_isolate_hermes_home):
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.managed_downloads import create_short_download_id, resolve_short_download_record
|
||||
|
||||
home = get_hermes_home()
|
||||
source = home / "owner-meta.txt"
|
||||
source.write_text("hello", encoding="utf-8")
|
||||
|
||||
short_id = create_short_download_id(
|
||||
str(source),
|
||||
owner_metadata={"email": "owner@bremen.com.tw", "principal_id": "p-123", "external_user_id": "openwebui-42", "ignored": "x"},
|
||||
)
|
||||
record = resolve_short_download_record(short_id)
|
||||
|
||||
assert record["metadata"] == {
|
||||
"email": "owner@bremen.com.tw",
|
||||
"principal_id": "p-123",
|
||||
"external_user_id": "openwebui-42",
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue