diff --git a/tools/typesafe-pareto/score_pareto.py b/tools/typesafe-pareto/score_pareto.py index e54339e..b1fe67e 100644 --- a/tools/typesafe-pareto/score_pareto.py +++ b/tools/typesafe-pareto/score_pareto.py @@ -15,7 +15,10 @@ python tools/typesafe-pareto/score_pareto.py --candidates path.json Live POST https://api.typesafe.ai/v1/systemone only when TYPESAFE_API_KEY -is set (and --force-mock is not). CI uses the in-process mock. +is set (and --force-mock is not). The API key is sent only to hosts in +ALLOWED_TYPESAFE_HOSTS, over HTTPS. Other --base-url / TYPESAFE_BASE_URL +values are rejected before a request, and cross-host redirects are refused. +CI uses the in-process mock. """ from __future__ import annotations @@ -29,7 +32,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Mapping, Sequence -from urllib.parse import urljoin +from urllib.parse import urljoin, urlsplit # Optional official SDK (https://docs.typesafe.ai/sdk/python.md). # CI does not install it; HTTP fallback matches the documented REST API. @@ -42,6 +45,9 @@ REPO = "AppSprout-dev/Unsga3" DEFAULT_MODEL = "jev-latest" DEFAULT_BASE_URL = "https://api.typesafe.ai" +# Hosts that may receive the TypeSafe API key. Only API origins already +# referenced in this repo belong here. +ALLOWED_TYPESAFE_HOSTS = frozenset({"api.typesafe.ai"}) SYSTEMONE_PATH = "/v1/systemone" API_KEY_ENV = "TYPESAFE_API_KEY" BASE_URL_ENV = "TYPESAFE_BASE_URL" @@ -83,6 +89,10 @@ class TypeSafeRequestError(RuntimeError): """Raised when the TypeSafe HTTP API returns an error or cannot be reached.""" +class TypeSafeConfigError(ValueError): + """Raised when a base URL would send the TypeSafe API key off its allowlist.""" + + def repo_root() -> Path: return REPO_ROOT @@ -372,11 +382,69 @@ def mock_system_one( } +def _allowlist_text() -> str: + return ", ".join(sorted(ALLOWED_TYPESAFE_HOSTS)) + + +def _typesafe_key_host(url: str) -> str | None: + """Return the hostname when ``url`` may receive the TypeSafe API key. + + The host must be an exact allowlist match, the scheme must be HTTPS, + and the URL must not carry userinfo. Anything else returns None. + """ + if not isinstance(url, str): + return None + parsed = urlsplit(url.strip()) + if parsed.scheme != "https" or parsed.username is not None or parsed.password is not None: + return None + host = parsed.hostname + if host is None: + return None + host = host.lower().rstrip(".") + if host not in ALLOWED_TYPESAFE_HOSTS: + return None + return host + + +def validate_typesafe_base_url(url: str) -> str: + """Return ``url`` when its host may receive the TypeSafe API key. + + Raises TypeSafeConfigError otherwise. This does not open a connection. + """ + cleaned = url.strip().rstrip("/") + if _typesafe_key_host(cleaned) is None: + raise TypeSafeConfigError( + f"Refusing TypeSafe base URL {url!r}: the API key is only sent " + f"over HTTPS to {_allowlist_text()}." + ) + return cleaned + + def _systemone_url(base_url: str) -> str: root = base_url.rstrip("/") + "/" return urljoin(root, SYSTEMONE_PATH.lstrip("/")) +class _AllowlistedRedirectHandler(urllib.request.HTTPRedirectHandler): + """Follow only HTTPS redirects that stay on the TypeSafe key allowlist. + + urllib keeps headers passed to Request(..., headers=...), including + Authorization, when HTTPRedirectHandler.redirect_request builds the next + request. A cross-host 3xx would send the API key to that second host. + Same-host HTTPS redirects on an allowlisted host are followed. Every + other redirect is refused before the next request is opened. + """ + + def redirect_request(self, req, fp, code, msg, headers, newurl): + if _typesafe_key_host(newurl) is None: + raise TypeSafeRequestError( + f"Refusing TypeSafe redirect to {newurl!r}: the API key is only " + f"sent over HTTPS to {_allowlist_text()}, and Authorization is " + "not forwarded to any other host." + ) + return super().redirect_request(req, fp, code, msg, headers, newurl) + + def _http_system_one( state: Mapping[str, Any], questions: Mapping[str, Mapping[str, Any]], @@ -386,9 +454,17 @@ def _http_system_one( base_url: str, timeout: float, ) -> dict[str, Any]: + # Check the origin before Authorization is attached to a request. + base_url = validate_typesafe_base_url(base_url) + target = _systemone_url(base_url) + if _typesafe_key_host(target) is None: + raise TypeSafeConfigError( + f"Refusing TypeSafe request URL {target!r}: the API key is only sent " + f"over HTTPS to {_allowlist_text()}." + ) body = json.dumps({"state": state, "model": model, "questions": questions}).encode("utf-8") request = urllib.request.Request( - _systemone_url(base_url), + target, data=body, method="POST", headers={ @@ -397,8 +473,9 @@ def _http_system_one( "Accept": "application/json", }, ) + opener = urllib.request.build_opener(_AllowlistedRedirectHandler()) try: - with urllib.request.urlopen(request, timeout=timeout) as resp: + with opener.open(request, timeout=timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") @@ -419,6 +496,7 @@ def _sdk_system_one( base_url: str, timeout: float, ) -> dict[str, Any]: + base_url = validate_typesafe_base_url(base_url) if TypeSafeClient is None: raise TypeSafeRequestError("typesafe-sdk is not installed.") with TypeSafeClient(api_key=api_key, model=model, base_url=base_url, timeout=timeout) as client: @@ -484,10 +562,18 @@ def resolve_model(explicit: str | None = None) -> str: def resolve_base_url(explicit: str | None = None) -> str: - if explicit: - return explicit.rstrip("/") - env = (os.environ.get(BASE_URL_ENV) or "").strip() - return (env or DEFAULT_BASE_URL).rstrip("/") + """Return the allowlisted TypeSafe base URL. + + Precedence is the explicit value (``--base-url`` / ``base_url=``), then + ``TYPESAFE_BASE_URL``, then ``DEFAULT_BASE_URL``. A host that is not on + ``ALLOWED_TYPESAFE_HOSTS`` raises TypeSafeConfigError before any request. + """ + chosen = explicit.strip() if isinstance(explicit, str) else "" + if not chosen: + chosen = (os.environ.get(BASE_URL_ENV) or "").strip() + if not chosen: + chosen = DEFAULT_BASE_URL + return validate_typesafe_base_url(chosen) def call_system_one( @@ -506,19 +592,24 @@ def call_system_one( if force_mock or not api_key: response = mock_system_one(state, questions, model=model) mode = "mock" - elif http_post is not None: - response = http_post(state, questions, model=model, api_key=api_key, base_url=base_url, timeout=timeout) - mode = "http" - elif TypeSafeClient is not None: - response = _sdk_system_one( - state, questions, model=model, api_key=api_key, base_url=base_url, timeout=timeout - ) - mode = "sdk" else: - response = _http_system_one( - state, questions, model=model, api_key=api_key, base_url=base_url, timeout=timeout - ) - mode = "http" + # Pin the key before the HTTP client, the SDK, or an injected post sees the URL. + base_url = validate_typesafe_base_url(base_url) + if http_post is not None: + response = http_post( + state, questions, model=model, api_key=api_key, base_url=base_url, timeout=timeout + ) + mode = "http" + elif TypeSafeClient is not None: + response = _sdk_system_one( + state, questions, model=model, api_key=api_key, base_url=base_url, timeout=timeout + ) + mode = "sdk" + else: + response = _http_system_one( + state, questions, model=model, api_key=api_key, base_url=base_url, timeout=timeout + ) + mode = "http" latency_ms = (time.perf_counter() - started) * 1000.0 return response, mode, latency_ms @@ -718,6 +809,15 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: p.add_argument("--metrics", type=Path, default=DEFAULT_METRICS) p.add_argument("--no-metrics", action="store_true") p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT_S) + p.add_argument( + "--base-url", + default=None, + help=( + "TypeSafe API origin " + f"(default: {DEFAULT_BASE_URL}, or {BASE_URL_ENV}). " + "The host must be on the API key allowlist." + ), + ) p.add_argument("--json", action="store_true", help="Print full result JSON") return p.parse_args(argv) @@ -737,15 +837,20 @@ def main(argv: Sequence[str] | None = None) -> int: payload = load_export(path) metrics_path = None if args.no_metrics else args.metrics - result = run_pass( - payload, - limit=args.limit, - model=args.model, - force_mock=args.force_mock, - metrics_path=metrics_path, - notes="smoke fixture" if args.smoke or path == DEFAULT_FIXTURE else f"export {path.name}", - timeout=args.timeout, - ) + try: + result = run_pass( + payload, + limit=args.limit, + model=args.model, + base_url=args.base_url, + force_mock=args.force_mock, + metrics_path=metrics_path, + notes="smoke fixture" if args.smoke or path == DEFAULT_FIXTURE else f"export {path.name}", + timeout=args.timeout, + ) + except TypeSafeConfigError as exc: + print(exc, file=sys.stderr) + return 2 print(f"TypeSafe Pareto pass | mode={result['mode']} model={result['metrics']['model']} " f"candidates={result['metrics']['candidate_count']} questions={len(result['questions'])} " diff --git a/tools/typesafe-pareto/test_score_pareto.py b/tools/typesafe-pareto/test_score_pareto.py index 3470dac..21549b1 100644 --- a/tools/typesafe-pareto/test_score_pareto.py +++ b/tools/typesafe-pareto/test_score_pareto.py @@ -2,13 +2,19 @@ """Mock tests for the TypeSafe Pareto helper. No network, no API keys.""" from __future__ import annotations +import http.client +import io import json import os +import socket import sys import tempfile import unittest +import urllib.parse +import urllib.request from pathlib import Path from unittest import mock +from urllib.response import addinfourl HERE = Path(__file__).resolve().parent if str(HERE) not in sys.path: @@ -204,6 +210,254 @@ def boom(*_a, **_k): self.assertEqual(result["mode"], "mock") +class _ScriptedHttpsHandler(urllib.request.HTTPSHandler): + """Canned HTTPS responses. https_open does not touch the network.""" + + def __init__( + self, + scripted: list[tuple[int, str, bytes]], + seen: list[dict[str, str | None]], + ) -> None: + super().__init__() + self._scripted = scripted + self._seen = seen + + def https_open(self, req: urllib.request.Request): + host = urllib.parse.urlsplit(req.full_url).hostname + self._seen.append( + { + "url": req.full_url, + "host": None if host is None else host.lower().rstrip("."), + "authorization": req.get_header("Authorization"), + } + ) + if self._scripted: + status, location, body = self._scripted.pop(0) + else: + status, location, body = 200, "", b'{"answers":{}}' + headers = http.client.HTTPMessage() + if location: + headers["Location"] = location + headers["Content-Type"] = "application/json" + response = addinfourl(io.BytesIO(body), headers, req.full_url, status) + response.msg = "Found" if status >= 300 else "OK" + return response + + +def _patched_opener(handler: urllib.request.BaseHandler): + """Install ``handler`` in place of the real HTTPS handler. No proxies.""" + real_build_opener = urllib.request.build_opener + + def build(*handlers: urllib.request.BaseHandler): + return real_build_opener(urllib.request.ProxyHandler({}), handler, *handlers) + + return mock.patch("urllib.request.build_opener", side_effect=build) + + +def _block_network(): + """Fail the test if urllib falls through to a real connection.""" + boom = mock.Mock(side_effect=AssertionError("network")) + return mock.patch.multiple(socket, create_connection=boom, getaddrinfo=boom) + + +class KeyAllowlistTests(unittest.TestCase): + def test_allowlisted_default_base_url_ok(self) -> None: + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop(sp.BASE_URL_ENV, None) + self.assertEqual(sp.resolve_base_url(), "https://api.typesafe.ai") + self.assertEqual(sp.resolve_base_url(None), "https://api.typesafe.ai") + self.assertEqual(sp.ALLOWED_TYPESAFE_HOSTS, frozenset({"api.typesafe.ai"})) + self.assertEqual( + sp.resolve_base_url("https://api.typesafe.ai/"), + "https://api.typesafe.ai", + ) + self.assertEqual( + sp.resolve_base_url("https://API.typesafe.ai"), + "https://API.typesafe.ai", + ) + seen: list[dict[str, str | None]] = [] + handler = _ScriptedHttpsHandler( + [(200, "", b'{"model":"jev-latest","answers":{"ok":{"type":"choice","choice":"keep"}}}')], + seen, + ) + with _block_network(), _patched_opener(handler): + payload = sp._http_system_one( + {"candidates": []}, + {}, + model="jev-latest", + api_key="test-key", + base_url=sp.resolve_base_url("https://api.typesafe.ai"), + timeout=1.0, + ) + self.assertEqual(payload["answers"]["ok"]["choice"], "keep") + self.assertEqual(len(seen), 1) + self.assertEqual(seen[0]["host"], "api.typesafe.ai") + self.assertEqual(seen[0]["url"], "https://api.typesafe.ai/v1/systemone") + self.assertEqual(seen[0]["authorization"], "Bearer test-key") + + def test_disallowed_base_url_rejected_without_request(self) -> None: + rejected = ( + "https://evil.example", + "http://api.typesafe.ai", + "https://api.typesafe.ai.evil.example", + "https://notapi.typesafe.ai", + "https://user:pass@api.typesafe.ai", + "https://api.typesafe.ai:443@evil.example", + ) + for url in rejected: + with self.subTest(url=url): + with self.assertRaises(sp.TypeSafeConfigError) as caught: + sp.resolve_base_url(url) + message = str(caught.exception) + self.assertIn("api.typesafe.ai", message) + self.assertNotIn("test-key", message) + + with mock.patch.dict(os.environ, {sp.BASE_URL_ENV: "https://evil.example/v1"}): + with self.assertRaises(sp.TypeSafeConfigError) as caught: + sp.resolve_base_url(None) + self.assertIn("evil.example", str(caught.exception)) + self.assertEqual( + sp.resolve_base_url("https://api.typesafe.ai"), + "https://api.typesafe.ai", + ) + + def boom(*_args, **_kwargs): + raise AssertionError("request must not be sent") + + with mock.patch("urllib.request.Request", side_effect=boom), mock.patch( + "urllib.request.build_opener", side_effect=boom + ): + with self.assertRaises(sp.TypeSafeConfigError): + sp._http_system_one( + {}, + {}, + model="jev-latest", + api_key="test-key", + base_url="https://evil.example", + timeout=1.0, + ) + + with self.assertRaises(sp.TypeSafeConfigError): + sp.call_system_one( + {"candidates": []}, + {}, + model="jev-latest", + api_key="test-key", + base_url="https://evil.example", + http_post=boom, + ) + with self.assertRaises(sp.TypeSafeConfigError): + sp._sdk_system_one( + {}, + {}, + model="jev-latest", + api_key="test-key", + base_url="https://evil.example", + timeout=1.0, + ) + payload = sp.load_export(FIXTURE) + with self.assertRaises(sp.TypeSafeConfigError): + sp.run_pass( + payload, + api_key="test-key", + base_url="https://evil.example", + force_mock=True, + metrics_path=None, + http_post=boom, + ) + + def test_cross_host_redirect_does_not_send_authorization(self) -> None: + seen: list[dict[str, str | None]] = [] + handler = _ScriptedHttpsHandler( + [(302, "https://evil.example/collect", b"")], + seen, + ) + with _block_network(), _patched_opener(handler): + with self.assertRaises(sp.TypeSafeRequestError) as caught: + sp._http_system_one( + {"candidates": []}, + {}, + model="jev-latest", + api_key="test-key", + base_url="https://api.typesafe.ai", + timeout=1.0, + ) + self.assertIn("evil.example", str(caught.exception)) + self.assertNotIn("test-key", str(caught.exception)) + self.assertEqual([row["host"] for row in seen], ["api.typesafe.ai"]) + self.assertEqual(seen[0]["authorization"], "Bearer test-key") + self.assertTrue(all(row["host"] != "evil.example" for row in seen)) + self.assertTrue(all("evil.example" not in (row["url"] or "") for row in seen)) + + def test_https_downgrade_redirect_is_refused(self) -> None: + seen: list[dict[str, str | None]] = [] + handler = _ScriptedHttpsHandler( + [(302, "http://api.typesafe.ai/v1/systemone", b"")], + seen, + ) + with _block_network(), _patched_opener(handler): + with self.assertRaises(sp.TypeSafeRequestError) as caught: + sp._http_system_one( + {"candidates": []}, + {}, + model="jev-latest", + api_key="test-key", + base_url="https://api.typesafe.ai", + timeout=1.0, + ) + self.assertIn("http://api.typesafe.ai", str(caught.exception)) + self.assertEqual([row["host"] for row in seen], ["api.typesafe.ai"]) + self.assertEqual(seen[0]["authorization"], "Bearer test-key") + self.assertTrue(all(row["url"].startswith("https://") for row in seen if row["url"])) + + def test_same_host_https_redirect_keeps_authorization(self) -> None: + seen: list[dict[str, str | None]] = [] + handler = _ScriptedHttpsHandler( + [ + (302, "https://api.typesafe.ai/v1/systemone/", b""), + (200, "", b'{"answers":{"ok":{"type":"choice","choice":"keep"}}}'), + ], + seen, + ) + with _block_network(), _patched_opener(handler): + payload = sp._http_system_one( + {"candidates": []}, + {}, + model="jev-latest", + api_key="test-key", + base_url="https://api.typesafe.ai", + timeout=1.0, + ) + self.assertEqual(payload["answers"]["ok"]["choice"], "keep") + self.assertEqual( + [row["host"] for row in seen], + ["api.typesafe.ai", "api.typesafe.ai"], + ) + self.assertEqual( + [row["authorization"] for row in seen], + ["Bearer test-key", "Bearer test-key"], + ) + + def test_cli_rejects_disallowed_base_url(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + metrics = Path(tmp) / "typesafe-runs.jsonl" + stderr = io.StringIO() + with mock.patch("sys.stderr", new=stderr), mock.patch("sys.stdout", new=io.StringIO()): + rc = sp.main( + [ + "--smoke", + "--force-mock", + "--metrics", + str(metrics), + "--base-url", + "https://evil.example", + ] + ) + self.assertEqual(rc, 2) + self.assertIn("evil.example", stderr.getvalue()) + self.assertFalse(metrics.exists()) + + class CliTests(unittest.TestCase): def test_smoke_cli_writes_metrics(self) -> None: with tempfile.TemporaryDirectory() as tmp: