1import pytest2from akos.gateway import ssrf345def test_ssrf_blocks_private_and_metadata_targets():6 for url in ("http://169.254.169.254/latest/meta-data/", "http://127.0.0.1:8520/", "http://localhost/", "http://10.0.0.1/", "http://192.168.2.1/", "http://metadata.google.internal/", "file:///etc/passwd", "http://user:pw@example.com/", "http://[::1]/", "http://0.0.0.0/", "http://100.100.100.100/"):7 with pytest.raises(ssrf.SSRFError):8 ssrf.validate(url, allow_local=False)9 t = ssrf.validate("https://api.worldbank.org/v2/", allow_local=False)10 assert t.host == "api.worldbank.org" and t.port == 443 and t.ip111213async def test_redirect_to_private_is_refused(app_client, mock_api, monkeypatch):14 """Even with loopback allowed for the mock, the redirect target (cloud metadata) is validated and refused."""15 c = app_client; h = c.auth_headers16 await c.post("/api/v1/projects", json={"slug": "ssrf", "name": "ssrf", "policy_mode": "discovery"}, headers=h)17 import akos.gateway.executor as ex18 real = ssrf.validate19 def strict(url, allow_local=None):20 return real(url, allow_local=("169.254" not in url))21 monkeypatch.setattr(ex.ssrf, "validate", strict)22 r = await c.post("/api/v1/execute", json={"project_id": "ssrf", "url": mock_api["base"] + "/redirect-private"}, headers=h)23 assert r.status_code == 200 and r.json()["state"] == "failed" and "ssrf" in (r.json()["error"] or "")242526async def test_prompt_injection_in_docs_is_inert(app_client, pool, mock_api):27 from tests.conftest import run_job_inline28 c = app_client; h = c.auth_headers29 res = await run_job_inline(pool, "discover", {"project_id": "default", "openapi_url": mock_api["base"] + "/openapi.json", "name": "Mock Data API", "provider_slug": "mock", "license": "CC0-1.0", "created_by": "t"})30 pkg = (await c.get(f"/api/v1/akps/{res['package_id']}", headers=h)).json()31 src = (await c.get(f"/api/v1/akps/{pkg['id']}/files/client.py", headers=h)).text32 assert pkg["manifest"]["static_scan"]["ok"]33 assert "IGNORE PREVIOUS INSTRUCTIONS" in src and "rm -rf" in src # kept verbatim as a string literal (data)34 import ast35 tree = ast.parse(src)36 assert not any(isinstance(n, ast.Call) and getattr(n.func, "id", "") in ("eval", "exec") for n in ast.walk(tree))37 readme = (await c.get(f"/api/v1/akps/{pkg['id']}/files/README.md", headers=h)).text38 assert "rm -rf" not in readme or "IGNORE" in readme # description is rendered as text in docs, never executed394041def test_rate_limit_token_bucket():42 from akos.api.deps import rate_limit, ApiError43 for _ in range(5):44 rate_limit("rl-test", limit=5, window=60)45 with pytest.raises(ApiError) as e:46 rate_limit("rl-test", limit=5, window=60)47 assert e.value.status_code == 429484950async def test_path_traversal_and_validation(app_client):51 c = app_client; h = c.auth_headers52 pk = (await c.get("/api/v1/akps", headers=h)).json()["packages"]53 if pk:54 r = await c.get(f"/api/v1/akps/{pk[0]['id']}/files/../../etc/passwd", headers=h); assert r.status_code in (400, 404)55 r = await c.get(f"/api/v1/akps/{pk[0]['id']}/files/%2e%2e%2f.env", headers=h); assert r.status_code in (400, 404)56 assert (await c.get("/api/v1/artifacts/not-a-sha", headers=h)).status_code == 40457 assert (await c.post("/api/v1/execute", json={"project_id": "default", "url": "x", "method": "TRACE"}, headers=h)).status_code == 42258 assert (await c.get("/docs/md/..%2F..%2F.env", headers=h)).status_code in (400, 404)59 assert (await c.get("/docs/md/nonexistent-doc", headers=h)).status_code == 40460 import httpx61 anon = httpx.AsyncClient(transport=c._transport, base_url="http://test")62 assert (await anon.get("/docs")).status_code == 401 and (await c.get("/docs", headers=h)).status_code == 20063 r = await c.get("/api/v1/health"); assert r.status_code == 200 and "X-Content-Type-Options" in r.headers and r.headers.get("Strict-Transport-Security")646566async def test_mcp_endpoint_auth_and_tools(app_client):67 c = app_client; h = c.auth_headers68 import httpx69 anon = httpx.AsyncClient(transport=c._transport, base_url="http://test")70 assert (await anon.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}})).status_code == 40171 r = await c.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18"}}, headers=h)72 assert r.status_code == 200 and r.json()["result"]["serverInfo"]["name"] == "akos"73 r = await c.post("/mcp", json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, headers=h)74 names = {t["name"] for t in r.json()["result"]["tools"]}75 assert names >= {"search_knowledge", "inspect_api", "discover_api", "execute_api", "generate_client", "verify_package", "list_knowledge_debt", "compose_recipe", "export_research_dataset", "get_execution"}76 r = await c.post("/mcp", json={"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "akos.status", "arguments": {}}}, headers=h)77 assert r.status_code == 200 and "counts" in r.json()["result"]["content"][0]["text"]78 r = await c.post("/mcp", json={"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "execute_api", "arguments": {"project_id": "default", "url": "http://169.254.169.254/"}}}, headers=h)79 assert r.json()["result"]["isError"] is True80