From c3fde1e38c07e45f0c0be49cc1de1eede4fb7887 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Wed, 23 Sep 2026 04:05:11 +0000 Subject: [PATCH] feat(opax-mcp): add deployment status sanitizer --- agents/mcp_server/server.py | 46 ++++++++++++++++++++ agents/mcp_server/test_server.py | 73 ++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+) create mode 100644 agents/mcp_server/test_server.py diff --git a/agents/mcp_server/server.py b/agents/mcp_server/server.py index 0f8982e..073aa8a 100644 --- a/agents/mcp_server/server.py +++ b/agents/mcp_server/server.py @@ -18,6 +18,9 @@ CLOUD_RUN_SERVICE = os.environ.get("CLOUD_RUN_SERVICE", "osvauco-agent") MCP_SECRET = os.environ.get("MCP_SECRET", "") BUILD_TRIGGER_ID = os.environ.get("CLOUD_BUILD_TRIGGER_ID", "38423976-91ff-4ff4-859e-1f262344c609") +_OPAX_MCP_SERVICE = "opax-mcp" +_OPAX_MCP_REGION = "us-central1" + def verify_token(request: Request, x_mcp_key: str = Header(default="")): # Aksepter både X-MCP-Key og api-key (Perplexity MCP connector bruker api-key) @@ -70,6 +73,49 @@ def _parse_ts(ts): return None +def _sanitize_opax_deployment_status(service_data: dict) -> dict: + """ + Takes a decoded Cloud Run v2 service JSON object and returns a sanitized + dict conforming to the minimal v1 output contract. + """ + def _validate_generation(val): + if isinstance(val, bool): + return None + if isinstance(val, int) and val >= 0: + return val + if isinstance(val, str) and val.isascii() and val.isdecimal(): + return int(val) + return None + + def _validate_update_time(val): + from datetime import datetime, timezone + if ( + not isinstance(val, str) + or not val.endswith("Z") + or val.endswith("ZZ") + or len(val) < 20 + or val[10] != "T" + ): + return None + try: + normalized_val = val[:-1] + "+00:00" + dt = datetime.fromisoformat(normalized_val) + if dt.tzinfo is None: + return None # Reject timezone-naive + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + except ValueError: + return None + + return { + "service": _OPAX_MCP_SERVICE, + "region": _OPAX_MCP_REGION, + "generation": _validate_generation(service_data.get("generation")), + "observed_generation": _validate_generation(service_data.get("observedGeneration")), + "last_update_time": _validate_update_time(service_data.get("updateTime")), + "reason_code": None, + } + + class PushStaticRequest(BaseModel): file_path: str content: str diff --git a/agents/mcp_server/test_server.py b/agents/mcp_server/test_server.py new file mode 100644 index 0000000..852833d --- /dev/null +++ b/agents/mcp_server/test_server.py @@ -0,0 +1,73 @@ +import unittest +from agents.mcp_server.server import _sanitize_opax_deployment_status + +class TestSanitizeOpaxDeploymentStatus(unittest.TestCase): + + def test_valid_inputs(self): + """Tests valid integer/string generations and a valid UTC timestamp.""" + service_data = { + "generation": 10, + "observedGeneration": "9", + "updateTime": "2023-10-27T10:00:00Z" + } + result = _sanitize_opax_deployment_status(service_data) + self.assertEqual(result["generation"], 10) + self.assertEqual(result["observed_generation"], 9) + self.assertEqual(result["last_update_time"], "2023-10-27T10:00:00Z") + self.assertIsNone(result["reason_code"]) + + def test_zero_values(self): + """Tests handling of zero for generation values.""" + service_data = {"generation": 0, "observedGeneration": "0"} + result = _sanitize_opax_deployment_status(service_data) + self.assertEqual(result["generation"], 0) + self.assertEqual(result["observed_generation"], 0) + + def test_invalid_generation_values(self): + """Tests various invalid generation inputs.""" + invalid_inputs = [ + True, False, -1, 1.5, "-1", " 1 ", "abc", "١", None, {}, [] + ] + for val in invalid_inputs: + with self.subTest(val=val): + res1 = _sanitize_opax_deployment_status({"generation": val}) + res2 = _sanitize_opax_deployment_status({"observedGeneration": val}) + self.assertIsNone(res1["generation"]) + self.assertIsNone(res2["observed_generation"]) + + def test_fractional_second_timestamp_is_truncated(self): + """Tests that fractional seconds are correctly truncated.""" + service_data = {"updateTime": "2023-10-27T10:00:00.123456Z"} + result = _sanitize_opax_deployment_status(service_data) + self.assertEqual(result["last_update_time"], "2023-10-27T10:00:00Z") + + def test_invalid_and_non_z_timestamps(self): + """Tests invalid, tz-naive, missing, and non-Z timestamps.""" + invalid_timestamps = [ + "2023-10-27T10:00:00", # Missing Z (naive) + "2023-10-27T10:00:00ZZ", # Multiple Z + "2023-10-27T11:00:00+01:00", # Valid but not Z-suffix + "2023-10-27 10:00:00Z", # Malformed + "not-a-timestamp", # Invalid + 1698399600, # Not a string + None, # Missing + ] + for ts in invalid_timestamps: + with self.subTest(ts=ts): + result = _sanitize_opax_deployment_status({"updateTime": ts}) + self.assertIsNone(result["last_update_time"]) + + def test_exact_key_set_and_fixed_values(self): + """Tests for exact key set and fixed service/region/reason_code.""" + result = _sanitize_opax_deployment_status({}) + expected_keys = { + "service", "region", "generation", "observed_generation", + "last_update_time", "reason_code" + } + self.assertEqual(set(result.keys()), expected_keys) + self.assertEqual(result["service"], "opax-mcp") + self.assertEqual(result["region"], "us-central1") + self.assertIsNone(result["reason_code"]) + +if __name__ == '__main__': + unittest.main()