From 37910af4f83be4188103022b20c2663469a60379 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 2 Oct 2026 14:10:28 +0800 Subject: [PATCH] fix: inspect effective protected account selector configuration Co-Authored-By: Codex --- scripts/inspect_account_data_readiness.py | 102 ++++++++++++++ tests/test_account_data_readiness.py | 157 ++++++++++++++++++++++ 2 files changed, 259 insertions(+) diff --git a/scripts/inspect_account_data_readiness.py b/scripts/inspect_account_data_readiness.py index 5391422..b539d8b 100644 --- a/scripts/inspect_account_data_readiness.py +++ b/scripts/inspect_account_data_readiness.py @@ -194,6 +194,103 @@ def _selector_configuration(revision: Mapping[str, Any]) -> tuple[bool, str]: return False, "absent" +def _unique_env_entry(env: list[Any], name: str) -> Mapping[str, Any] | None: + matches = [ + entry + for entry in env + if isinstance(entry, Mapping) and entry.get("name") == name + ] + if len(matches) > 1: + raise DiagnosticFailure("runtime_target_configuration_ambiguous") + return matches[0] if matches else None + + +def _literal_env_value(entry: Mapping[str, Any]) -> str | None: + value_source = entry.get("valueSource") + secret_ref = ( + value_source.get("secretKeyRef") + if isinstance(value_source, Mapping) + else None + ) + has_secret_ref = isinstance(secret_ref, Mapping) + has_literal = "value" in entry + if has_secret_ref and has_literal: + raise DiagnosticFailure("runtime_target_configuration_ambiguous") + if has_secret_ref: + return None + value = entry.get("value") + if value is None: + return "" + if not isinstance(value, str): + raise DiagnosticFailure("runtime_target_configuration_invalid") + return value + + +def _reject_duplicate_json_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise DiagnosticFailure("runtime_target_json_ambiguous") + result[key] = value + return result + + +def _selector_is_nonempty(value: Any) -> bool: + if value is None: + return False + try: + if isinstance(value, str): + return bool(value.strip()) + return any(item is not None and str(item).strip() for item in value) + except TypeError: + raise DiagnosticFailure("runtime_target_selector_invalid") from None + + +def _effective_selector_configuration( + revision: Mapping[str, Any], direct_selector: tuple[bool, str] +) -> tuple[bool, str]: + containers = revision.get("containers") + if not isinstance(containers, list) or len(containers) != 1: + raise DiagnosticFailure("revision_container_invalid") + container = containers[0] + env = container.get("env") if isinstance(container, Mapping) else None + if env is None: + env = [] + if not isinstance(env, list): + raise DiagnosticFailure("revision_environment_invalid") + + # The deployed resolver prefers QSL_RUNTIME_TARGET_JSON, then + # RUNTIME_TARGET_JSON, and only uses FIRSTRADE_ACCOUNT when both are empty. + for name, source in ( + ("QSL_RUNTIME_TARGET_JSON", "qsl_runtime_target_literal"), + ("RUNTIME_TARGET_JSON", "runtime_target_literal"), + ): + entry = _unique_env_entry(env, name) + if entry is None: + continue + raw = _literal_env_value(entry) + if raw is None: + return False, "runtime_target_secret_unresolved" + if raw == "": + continue + if not raw.strip(): + raise DiagnosticFailure("runtime_target_json_invalid") + try: + payload = json.loads(raw, object_pairs_hook=_reject_duplicate_json_keys) + except DiagnosticFailure: + raise + except (json.JSONDecodeError, TypeError, ValueError): + raise DiagnosticFailure("runtime_target_json_invalid") from None + if not isinstance(payload, Mapping): + raise DiagnosticFailure("runtime_target_json_invalid") + return _selector_is_nonempty(payload.get("account_selector")), source + + direct_configured, direct_source = direct_selector + if direct_source == "secret_reference": + return False, "first_trade_account_secret_unresolved" + return direct_configured, f"first_trade_account_{direct_source}" + + def inspect_readiness( service_name: str, region: str, @@ -226,6 +323,9 @@ def inspect_readiness( if not isinstance(source_commit, str) or not COMMIT_RE.fullmatch(source_commit): raise DiagnosticFailure("source_commit_unavailable") selector_configured, selector_source = _selector_configuration(revision) + selector_nonempty, effective_selector_source = _effective_selector_configuration( + revision, (selector_configured, selector_source) + ) after = request_json(service_url, token) if after.get("name") != service_path: raise DiagnosticFailure("service_identity_mismatch") @@ -238,6 +338,8 @@ def inspect_readiness( "source_commit": source_commit.lower(), "selector_configured": selector_configured, "selector_source": selector_source, + "selector_nonempty": selector_nonempty, + "effective_selector_source": effective_selector_source, } diff --git a/tests/test_account_data_readiness.py b/tests/test_account_data_readiness.py index cd0a388..6b56054 100644 --- a/tests/test_account_data_readiness.py +++ b/tests/test_account_data_readiness.py @@ -77,6 +77,8 @@ def test_inspects_only_exact_project_region_service_and_serving_revision() -> No "source_commit": COMMIT, "selector_configured": True, "selector_source": "literal", + "selector_nonempty": True, + "effective_selector_source": "first_trade_account_literal", } assert urls == [ f"{readiness.API_ROOT}/{PROJECT}", @@ -108,6 +110,161 @@ def test_selector_reports_only_presence_and_source( assert SENTINEL not in json.dumps(result) +@pytest.mark.parametrize( + ("env", "nonempty", "source"), + [ + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":["private"]}'}], + True, + "runtime_target_literal", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":[]}'}], + False, + "runtime_target_literal", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":[null]}'}], + False, + "runtime_target_literal", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":[null," "]}'}], + False, + "runtime_target_literal", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":[null,"real-placeholder"]}'}], + True, + "runtime_target_literal", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":" "}'}], + False, + "runtime_target_literal", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"platform_id":"firstrade"}'}], + False, + "runtime_target_literal", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "valueSource": {"secretKeyRef": {"secret": SENTINEL}}}], + False, + "runtime_target_secret_unresolved", + ), + ( + [{"name": "FIRSTRADE_ACCOUNT", "value": "private"}], + True, + "first_trade_account_literal", + ), + ( + [{"name": "FIRSTRADE_ACCOUNT", "valueSource": {"secretKeyRef": {"secret": SENTINEL}}}], + False, + "first_trade_account_secret_unresolved", + ), + ([], False, "first_trade_account_absent"), + ], +) +def test_effective_selector_reports_runtime_target_without_exposing_values( + env: list[dict[str, Any]], nonempty: bool, source: str +) -> None: + request, _ = _runner(revision=_revision(env=env)) + + result = readiness.inspect_readiness( + "firstrade-platform", "us-central1", "token", request_json=request + ) + + assert result["selector_nonempty"] is nonempty + assert result["effective_selector_source"] == source + serialized = json.dumps(result) + assert SENTINEL not in serialized + assert "private" not in serialized + + +def test_qsl_runtime_target_literal_has_deployed_precedence() -> None: + request, _ = _runner( + revision=_revision( + env=[ + {"name": "QSL_RUNTIME_TARGET_JSON", "value": '{"account_selector":[]}'}, + {"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":["private"]}'}, + {"name": "FIRSTRADE_ACCOUNT", "value": "another-private-value"}, + ] + ) + ) + + result = readiness.inspect_readiness( + "firstrade-platform", "us-central1", "token", request_json=request + ) + + assert result["selector_nonempty"] is False + assert result["effective_selector_source"] == "qsl_runtime_target_literal" + assert "private" not in json.dumps(result) + + +def test_empty_qsl_runtime_target_falls_through_to_runtime_target() -> None: + request, _ = _runner( + revision=_revision( + env=[ + {"name": "QSL_RUNTIME_TARGET_JSON", "value": ""}, + {"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":["private"]}'}, + ] + ) + ) + + result = readiness.inspect_readiness( + "firstrade-platform", "us-central1", "token", request_json=request + ) + + assert result["selector_nonempty"] is True + assert result["effective_selector_source"] == "runtime_target_literal" + assert "private" not in json.dumps(result) + + +@pytest.mark.parametrize( + ("env", "reason"), + [ + ( + [ + {"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":[]}'}, + {"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":["x"]}'}, + ], + "runtime_target_configuration_ambiguous", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":'}], + "runtime_target_json_invalid", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":[],"account_selector":["x"]}'}], + "runtime_target_json_ambiguous", + ), + ( + [{"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":1}'}], + "runtime_target_selector_invalid", + ), + ( + [ + {"name": "QSL_RUNTIME_TARGET_JSON", "value": " "}, + {"name": "RUNTIME_TARGET_JSON", "value": '{"account_selector":["x"]}'}, + ], + "runtime_target_json_invalid", + ), + ], +) +def test_rejects_ambiguous_or_invalid_runtime_target_selector_safely( + env: list[dict[str, Any]], reason: str +) -> None: + request, _ = _runner(revision=_revision(env=env)) + + with pytest.raises(readiness.DiagnosticFailure, match=reason) as caught: + readiness.inspect_readiness( + "firstrade-platform", "us-central1", "token", request_json=request + ) + + assert SENTINEL not in str(caught.value) + + @pytest.mark.parametrize( "traffic", [