Spaces:
Running
Running
| import importlib.util | |
| import json | |
| import subprocess | |
| import threading | |
| from pathlib import Path | |
| import httpx | |
| import pytest | |
| from flask import Flask, Response | |
| from helpers import api, cache, fasta2a_server, runtime, settings | |
| ROOT = Path(__file__).resolve().parents[1] | |
| def isolate(monkeypatch): | |
| for key in ( | |
| "SPACE_ID", | |
| "HF_SPACE", | |
| "SPYNEL_AGENT_ZERO_API_KEY", | |
| "A0_SET_MCP_SERVER_TOKEN", | |
| ): | |
| monkeypatch.delenv(key, raising=False) | |
| monkeypatch.setattr(settings, "create_auth_token", lambda: "generated-local") | |
| def test_real_hf_detection(monkeypatch, space, legacy, expected): | |
| if space is not None: | |
| monkeypatch.setenv("SPACE_ID", space) | |
| if legacy is not None: | |
| monkeypatch.setenv("HF_SPACE", legacy) | |
| assert runtime.is_huggingface_space() is expected | |
| def test_token_authority(monkeypatch, env, configured, expected, source): | |
| for key, value in env.items(): | |
| monkeypatch.setenv(key, value) | |
| assert settings.resolve_api_token_with_source(configured) == (expected, source) | |
| def test_hf_flags_mask_and_safe_diagnostics(monkeypatch): | |
| monkeypatch.setenv("SPACE_ID", "owner/space") | |
| current = settings.normalize_settings(settings.get_default_settings()) | |
| assert not current["mcp_server_enabled"] and not current["a2a_server_enabled"] | |
| assert ( | |
| settings.external_service_diagnostics(current)["incoming_api_token_source"] | |
| == "missing" | |
| ) | |
| monkeypatch.setenv("SPYNEL_AGENT_ZERO_API_KEY", "canonical-test-secret") | |
| current = settings.normalize_settings(current) | |
| assert current["mcp_server_enabled"] and current["a2a_server_enabled"] | |
| output = settings.convert_out(current) | |
| assert output["settings"]["mcp_server_token"] == "************" | |
| assert "canonical-test-secret" not in json.dumps(output) | |
| assert ( | |
| output["additional"]["external_services"]["incoming_api_token_source"] | |
| == "deployment_secret" | |
| ) | |
| def test_real_http_dispatch_preserves_browser_and_external_security(monkeypatch): | |
| monkeypatch.setenv("SPACE_ID", "owner/space") | |
| monkeypatch.setenv("SPYNEL_AGENT_ZERO_API_KEY", "canonical-test-secret") | |
| monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "login-hash") | |
| current = settings.normalize_settings(settings.get_default_settings()) | |
| monkeypatch.setattr(settings, "get_settings", lambda: current) | |
| from api.api_message import ApiMessage | |
| async def harmless(self, input, request): | |
| return {"accepted": True} | |
| monkeypatch.setattr(ApiMessage, "process", harmless) | |
| from helpers import modules | |
| original_loader = modules.load_classes_from_file | |
| monkeypatch.setattr( | |
| modules, | |
| "load_classes_from_file", | |
| lambda path, base: ( | |
| [ApiMessage] | |
| if str(path).endswith("/api_message.py") | |
| else original_loader(path, base) | |
| ), | |
| ) | |
| app = Flask("hf-auth") | |
| app.secret_key = "unit-only" | |
| app.add_url_rule("/", "serve_index", lambda: Response("index")) | |
| app.add_url_rule("/login", "login_handler", lambda: Response("login")) | |
| cache.clear(api.CACHE_AREA) | |
| api.register_api_route(app, threading.RLock()) | |
| client = app.test_client() | |
| try: | |
| for path in ("settings_get", "message"): | |
| assert client.post("/api/" + path, json={}).status_code == 302 | |
| for path in ("settings_probe", "api_message"): | |
| url = "/api/" + path | |
| assert ( | |
| client.post(url, json={}, headers={"X-API-KEY": "wrong"}).status_code | |
| == 401 | |
| ) | |
| response = client.post( | |
| url, json={}, headers={"X-API-KEY": "canonical-test-secret"} | |
| ) | |
| assert response.status_code == 200 | |
| assert "canonical-test-secret" not in response.get_data(as_text=True) | |
| with client.session_transaction() as session: | |
| session["authentication"] = "login-hash" | |
| session["csrf_token"] = "csrf-test" | |
| assert client.post("/api/settings_get", json={}).status_code == 403 | |
| assert ( | |
| client.post( | |
| "/api/settings_get", json={}, headers={"X-CSRF-Token": "csrf-test"} | |
| ).status_code | |
| == 200 | |
| ) | |
| monkeypatch.delenv("SPYNEL_AGENT_ZERO_API_KEY") | |
| assert ( | |
| client.post( | |
| "/api/settings_probe", json={}, headers={"X-API-KEY": "generated-local"} | |
| ).status_code | |
| == 503 | |
| ) | |
| finally: | |
| cache.clear(api.CACHE_AREA) | |
| async def test_a2a_authentication_routes(monkeypatch, mode): | |
| monkeypatch.setenv("SPACE_ID", "owner/space") | |
| if mode != "missing": | |
| monkeypatch.setenv("SPYNEL_AGENT_ZERO_API_KEY", "canonical-test-secret") | |
| monkeypatch.setattr( | |
| settings, | |
| "get_settings", | |
| lambda: {"mcp_server_token": "old", "a2a_server_enabled": True}, | |
| ) | |
| monkeypatch.setattr(fasta2a_server, "FASTA2A_AVAILABLE", True) | |
| proxy = object.__new__(fasta2a_server.DynamicA2AProxy) | |
| proxy._startup_done = True | |
| proxy._reconfigure_needed = False | |
| proxy._lock = threading.Lock() | |
| async def target(scope, receive, send): | |
| await send({"type": "http.response.start", "status": 200, "headers": []}) | |
| await send({"type": "http.response.body", "body": b'{"accepted":true}'}) | |
| proxy.app = target | |
| headers = ( | |
| {"Authorization": "Bearer canonical-test-secret"} | |
| if mode == "bearer" | |
| else {"X-API-KEY": "canonical-test-secret" if mode == "header" else "wrong"} | |
| ) | |
| path = ( | |
| "/a2a/t-canonical-test-secret/.well-known/agent.json" | |
| if mode == "path" | |
| else "/a2a/.well-known/agent.json" | |
| ) | |
| async with httpx.AsyncClient( | |
| transport=httpx.ASGITransport(app=proxy), base_url="https://test" | |
| ) as client: | |
| response = await client.get(path, headers=headers) | |
| assert response.status_code == ( | |
| 503 if mode == "missing" else 401 if mode == "wrong" else 200 | |
| ) | |
| assert "canonical-test-secret" not in response.text | |
| def load_deployer(): | |
| spec = importlib.util.spec_from_file_location( | |
| "hf_deployer", ROOT / "scripts/deploy_hf_space.py" | |
| ) | |
| mod = importlib.util.module_from_spec(spec) | |
| spec.loader.exec_module(mod) | |
| return mod | |
| def test_snapshot_refuses_dirty_and_archives_exact_commit(tmp_path, capsys): | |
| def git(*args): | |
| return subprocess.check_output(["git", *args], cwd=tmp_path, text=True).strip() | |
| git("init", "-q") | |
| git("config", "user.email", "test@example.test") | |
| git("config", "user.name", "Test") | |
| (tmp_path / "README.md").write_text("first") | |
| git("add", ".") | |
| git("commit", "-qm", "first") | |
| first = git("rev-parse", "HEAD") | |
| (tmp_path / "README.md").write_text("second") | |
| deploy = load_deployer() | |
| with pytest.raises(RuntimeError, match="uncommitted"): | |
| deploy.deployment_identity(tmp_path, "", "test/space") | |
| git("commit", "-qam", "second") | |
| untracked = tmp_path / "untracked.txt" | |
| untracked.write_text("not committed") | |
| with pytest.raises(RuntimeError, match="uncommitted"): | |
| deploy.deployment_identity(tmp_path, first, "test/space") | |
| untracked.unlink() | |
| assert deploy.deployment_identity(tmp_path, first, "test/space") == first | |
| assert "historical snapshot" in capsys.readouterr().out | |
| output = tmp_path / "snapshot" | |
| output.mkdir() | |
| deploy.archive_snapshot(tmp_path, first, output) | |
| assert (output / "README.md").read_text() == "first" | |
| assert ( | |
| json.loads((output / "openoperator-build.json").read_text())["source_sha"] | |
| == first | |
| ) | |
| def test_connection_examples_never_use_settings_mask_as_token(): | |
| for file in ( | |
| "mcp/server/example.html", | |
| "a2a/a2a-connection.html", | |
| "external/api-examples.html", | |
| ): | |
| source = (ROOT / "webui/components/settings" / file).read_text() | |
| assert 'const token = "YOUR_SPYNEL_AGENT_ZERO_API_KEY"' in source | |
| assert "token = response?.settings?.mcp_server_token" not in source | |
| assert "t-************" not in source | |
| async def test_mcp_token_path_transport_auth(monkeypatch, transport): | |
| from helpers import mcp_server | |
| proxy = object.__new__(mcp_server.DynamicMcpProxy) | |
| proxy._lock = threading.Lock() | |
| proxy.token = "canonical-test-secret" | |
| called = [] | |
| async def target(scope, receive, send): | |
| called.append(scope["path"]) | |
| proxy.sse_app = proxy.http_app = target | |
| async def unused(*args): | |
| pass | |
| scope = { | |
| "type": "http", | |
| "path": f"/t-canonical-test-secret/{transport}", | |
| "headers": [], | |
| } | |
| await proxy(scope, unused, unused) | |
| assert len(called) == 1 | |
| with pytest.raises(mcp_server.StarletteHTTPException) as caught: | |
| await proxy({**scope, "path": f"/t-wrong/{transport}"}, unused, unused) | |
| assert caught.value.status_code == 403 and len(called) == 1 | |
| async def test_mcp_missing_canonical_secret_fails_closed(monkeypatch): | |
| from helpers import mcp_server | |
| monkeypatch.setenv("SPACE_ID", "owner/space") | |
| monkeypatch.setattr( | |
| settings, | |
| "get_settings", | |
| lambda: {"mcp_server_enabled": True, "mcp_server_token": "old"}, | |
| ) | |
| async def must_not_run(request): | |
| raise AssertionError("unauthorized MCP dispatch") | |
| with pytest.raises(mcp_server.StarletteHTTPException) as caught: | |
| await mcp_server.mcp_middleware(None, must_not_run) | |
| assert caught.value.status_code == 503 | |
| def test_actual_entrypoint_settings_block_requires_deployment_secret(tmp_path, present): | |
| import os | |
| source = (ROOT / "docker/run/fs/exe/huggingface-entrypoint.sh").read_text() | |
| code = source.split("python3 - <<'PY'\n", 1)[1].split("\nPY", 1)[0] | |
| path = tmp_path / "settings.json" | |
| path.write_text( | |
| json.dumps( | |
| { | |
| "sentinel": "preserve", | |
| "a2a_server_enabled": not present, | |
| "mcp_server_enabled": not present, | |
| } | |
| ) | |
| ) | |
| code = code.replace('Path("/a0/usr/settings.json")', f"Path({str(path)!r})") | |
| env = dict(os.environ) | |
| if present: | |
| env["SPYNEL_AGENT_ZERO_API_KEY"] = "entrypoint-test-secret" | |
| else: | |
| env.pop("SPYNEL_AGENT_ZERO_API_KEY", None) | |
| result = subprocess.run( | |
| ["python3", "-c", code], env=env, capture_output=True, text=True, check=True | |
| ) | |
| state = json.loads(path.read_text()) | |
| assert ( | |
| state["a2a_server_enabled"] is present | |
| and state["mcp_server_enabled"] is present | |
| ) | |
| assert state["sentinel"] == "preserve" | |
| assert ( | |
| "entrypoint-test-secret" not in result.stdout + result.stderr + path.read_text() | |
| ) | |
| if not present: | |
| assert "External integrations disabled" in result.stdout | |
| def test_deployment_health_requires_exact_sha(monkeypatch): | |
| deploy = load_deployer() | |
| class Health: | |
| status = 200 | |
| def __enter__(self): | |
| return self | |
| def __exit__(self, *args): | |
| pass | |
| def read(self, *args): | |
| return b'{"sha":"expected"}' | |
| monkeypatch.setattr( | |
| deploy.urllib.request, "urlopen", lambda *args, **kwargs: Health() | |
| ) | |
| assert deploy.verify_health("https://space.example/health", "expected") | |
| assert not deploy.verify_health("https://space.example/health", "different") | |