"""Exercise seeded clients against a local proxy, without internal credentials.""" import importlib.util import json import os from pathlib import Path import subprocess import sys import tempfile import threading import unittest from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer ROOT = Path(__file__).parent / "project-seeds/ticket-manager/.agents/skills/builtin" class ProxySkillsTest(unittest.TestCase): def setUp(self): self.requests = [] self.status = 200 test = self class Handler(BaseHTTPRequestHandler): def do_GET(self): body = self.rfile.read(int(self.headers.get("Content-Length", "0"))) test.requests.append((self.command, self.path, dict(self.headers), body)) self.send_response(test.status) self.send_header("Content-Type", "application/json") if test.status == 302: self.send_header("Location", test.proxy_url + "/bypassed") self.end_headers() self.wfile.write(b'{}') do_POST = do_GET do_PUT = do_GET def log_message(self, *args): pass self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.proxy_url = f"http://127.0.0.1:{self.server.server_port}" self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() self.addCleanup(self.stop_server) self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.env = { **os.environ, "DS_PROXY_URL": self.proxy_url + "/", "DEVOPS_ENV": "staging", # Removed settings must neither override proxy routing nor add auth. "DEVOPS_API_TOKEN": "unused-test-token", "DEVOPS_BASE_DOMAIN": "http://direct.invalid", "JSM_SYNCER_BASE_URL": "http://direct.invalid", "JIRA_URL": "http://direct.invalid", "JIRA_USERNAME": "unused-test-user", "JIRA_API_TOKEN": "unused-test-token", "HTTP_PROXY": "http://unreachable.invalid:1", "http_proxy": "http://unreachable.invalid:1", "ALL_PROXY": "http://unreachable.invalid:1", "NO_PROXY": "", "no_proxy": "", "PYTHONDONTWRITEBYTECODE": "1", } # Neither source .env nor read cached credentials from it. Path(self.temp.name, ".env").write_text("DS_PROXY_URL=http://direct.invalid\n") def stop_server(self): self.server.shutdown() self.server.server_close() self.thread.join() def run_client(self, client, *args, env=None): script = ROOT / client return subprocess.run( ["bash" if script.suffix == ".sh" else sys.executable, str(script.resolve()), *args], env=self.env if env is None else env, cwd=self.temp.name, capture_output=True, text=True, timeout=10, ) def assert_no_auth(self): for _, _, headers, _ in self.requests: names = {name.lower() for name in headers} self.assertFalse(names & {"authorization", "proxy-authorization", "cookie", "x-api-token", "x-api-key"}) def test_devops_proxy_environment_query_and_body(self): result = self.run_client( "devops-api/scripts/devops_http.sh", "--path", "/cloudconfig/cluster/changes", "--env", "dev", "--method", "POST", "--query", "cluster_id=123", "--body", '{"cluster_id":"123"}', ) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual(("POST", "/devops/dev/api/v1/devops/cloudconfig/cluster/changes?cluster_id=123"), self.requests[0][:2]) self.assertEqual({"cluster_id": "123"}, json.loads(self.requests[0][3])) self.assert_no_auth() def test_jira_proxy_get(self): result = self.run_client("jira-search/scripts/jira_search.py", "issue", "ST-1") self.assertEqual(0, result.returncode, result.stderr) self.assertTrue(self.requests[0][1].startswith("/jira/rest/api/2/issue/ST-1?")) self.assert_no_auth() def test_jira_proxy_post(self): spec = importlib.util.spec_from_file_location("jira_search_proxy_test", ROOT / "jira-search/scripts/jira_search.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) module.jira_post(module.JiraConfig(self.proxy_url, 5), "/rest/api/2/search/approximate-count", {"jql": "project = ST"}) self.assertEqual(("POST", "/jira/rest/api/2/search/approximate-count"), self.requests[0][:2]) self.assertEqual({"jql": "project = ST"}, json.loads(self.requests[0][3])) self.assert_no_auth() def test_jsm_doctor_uses_both_routes_and_reports_proxy_failure(self): self.status = 503 result = self.run_client("jsm-ticket-access/scripts/check_ticket_access.py", "--doctor", "--env", "dev", "--retries", "0") self.assertEqual(2, result.returncode, result.stderr) self.assertFalse(json.loads(result.stdout)["ok"]) paths = [item[1] for item in self.requests] self.assertTrue(any(path.startswith("/devops/dev/api/v1/jsm-syncer/") for path in paths)) self.assertTrue(any(path.startswith("/jira/rest/") for path in paths)) self.assert_no_auth() def clients(self): return ( ("devops-api/scripts/devops_http.sh", "--path", "/test"), ("jira-search/scripts/jira_search.py", "issue", "ST-1"), ("jsm-ticket-access/scripts/check_ticket_access.py", "--doctor", "--retries", "0"), ) def test_missing_proxy_fails_before_requests_even_with_legacy_credentials(self): env = {key: value for key, value in self.env.items() if key != "DS_PROXY_URL"} for client in self.clients(): with self.subTest(client=client[0]): result = self.run_client(*client, env=env) self.assertNotEqual(0, result.returncode) self.assertIn("DS_PROXY_URL", result.stderr) self.assertEqual([], self.requests) def test_redirects_and_denials_never_bypass_proxy(self): for status in (302, 403): self.status = status for client in self.clients(): with self.subTest(status=status, client=client[0]): result = self.run_client(*client) self.assertNotEqual(0, result.returncode) self.assertFalse(any(path == "/bypassed" for _, path, _, _ in self.requests)) self.assert_no_auth() def test_devops_rejects_auth_headers(self): for header in ("AUTHORIZATION: test", "X-Api-Key: test", "Cookie: test"): result = self.run_client("devops-api/scripts/devops_http.sh", "--path", "/test", "--header", header) self.assertNotEqual(0, result.returncode) self.assertEqual([], self.requests) if __name__ == "__main__": unittest.main()