fix(datasources): bound connector reads and verify real query access

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Native observability connectors: connection tests verify bounded query access instead of public health endpoints. Explicit instance reads require ID-bound credentials; migrated defaults must be backfilled through the administrator save flow before rollout. Prometheus and Mimir default to a 10-second execution ceiling, configurable per datasource and tightened by Explore; provider response bytes and body reads are bounded. Valid short Jaeger trace IDs are retained, and local loss, search saturation and unknown results carry distinct completeness evidence.
 - Security scanning hardening: deployment workflows select only explicit preview secrets; chat history deletion hides internal errors and session IDs always use cryptographic randomness. The configurator escapes literal HCL strings, screenshot login requires exact configured origins and valid TLS, presenter titles retain text semantics, datasource hostname classification checks the AWS DNS suffix including root-dot and China-partition forms, and rotation diagnostics omit unrelated secret identifiers.
 - AgentCore Gateway inbound authentication: the provisioner creates every section gateway with IAM (SigV4) inbound authentication instead of none. Because a gateway's authorizer type cannot be changed after creation, `make agentcore` fails on any existing gateway that still accepts unauthenticated calls, and the AgentCore reference documents the operator replacement procedure.
 - AI chat grounding: the chat agent is told the current time, the account and region its tools read, and how to treat capped, unfinished or failed results. Capped AWS list results report how many items were returned and found and whether more exist, a list is called complete only when its tool says so, and a successful empty result is reported as empty. Unfinished or failed Logs Insights queries, AWS-capped route searches, ECS lists with failed descriptions, unreadable serverless-cache listings and Cost Explorer comparisons that stopped early are reported as incomplete or unknown rather than complete or empty. Synced-inventory answers state that they cover the host account and warn when another account was asked for; a requested account that is unregistered, disabled or unverifiable is named instead of silently falling back to the host account; an answer produced without live tools starts with a notice that it uses no real data; and long conversations send only the most recent turns that fit the request size. A follow-up question without a domain keyword ("why is that?", "more detail") stays with the section that answered the previous turn instead of falling back to the general Ops agent, while an explicit keyword, a pin or (with hybrid routing) a classifier choice still re-routes. On AgentCore agent answers, the agent is told which regions AWSops is configured to inventory for the target account (when that scope can be read; it is an AWSops setting, not which regions AWS has enabled), so it queries them and names the regions it checked rather than answering account-wide questions from one region. When a thread switches accounts, earlier turns are labelled with the account they were about (where that was recorded), so answers about one account are not presented as facts about another. Long answers get an explicit output budget instead of the model default, so root-cause answers are no longer cut mid-sentence, and an answer that still stops early says whether it hit the length limit or lost the connection. Synced-inventory rows returned by the inventory query tool carry their region and, for EC2, instance id, type, power state and last state change (the reader view allowlists those payload keys; the last-change time is empty for running instances), so region-scoped and running/stopped questions use recorded values instead of resource-name prefixes or an "unknown" answer.
@@ -691,6 +692,7 @@ First release of the **v2 line** (versioned independently from the v1 1.x line,
 
 ### 수정
 
+- 외부 관측 데이터소스: 연결 테스트가 공개 상태 엔드포인트 대신 제한된 실제 조회 권한을 검증합니다. 인스턴스 ID 조회는 해당 ID의 자격 증명만 허용하며, 이전 기본 연결은 배포 전에 관리자 저장 흐름으로 자격 증명을 보완해야 합니다. Prometheus·Mimir의 기본 실행 제한은 10초이며 데이터소스별 설정과 Explore의 더 짧은 제한을 적용합니다. 응답 크기와 본문 읽기 시간을 제한하고, 짧은 Jaeger trace ID를 보존하며 실제 누락·조회 한도 도달·판별 불가 결과를 구분합니다.
 - 보안 검사 개선: 배포 워크플로가 명시된 미리보기 시크릿만 선택하며, 채팅 기록 삭제 오류는 내부 정보를 숨기고 세션 ID는 암호학적 난수만 사용합니다. 설정 도구는 HCL 문자열을 안전하게 인코딩하고, 스크린샷 로그인은 지정한 정확한 출처와 유효한 TLS를 요구합니다. 발표자 제목은 텍스트로 보존하며, 데이터소스 호스트 분류는 DNS 루트 점과 중국 파티션을 포함한 AWS 도메인 접미사를 확인하고, 회전 진단에서는 무관한 시크릿 식별자를 생략합니다.
 - AgentCore Gateway 유입 인증: 프로비저너가 모든 섹션 Gateway를 인증 없음 대신 IAM(SigV4) 유입 인증으로 생성합니다. Gateway의 인증 방식은 생성 후 바꿀 수 없으므로, 아직 인증 없는 호출을 받는 기존 Gateway가 있으면 `make agentcore`가 실패하며, 운영자 교체 절차는 AgentCore 참조 문서에 안내합니다.
 - AI 채팅 근거 강화: 채팅 에이전트가 현재 시각, 도구가 조회하는 계정·리전, 잘리거나 끝나지 않았거나 실패한 결과를 다루는 방법을 전달받습니다. 상한이 걸린 AWS 목록 결과는 반환·조회 개수와 추가 항목 여부를 알려 주고, 도구가 완전하다고 표시한 목록만 완전하다고 말하며, 성공했지만 비어 있는 결과는 비어 있다고 답합니다. 끝나지 않았거나 실패한 Logs Insights 쿼리, AWS가 상한을 둔 경로 검색, 상세 조회가 실패한 ECS 목록, 읽을 수 없는 서버리스 캐시 목록, 중간에 멈춘 Cost Explorer 비교는 완전하거나 비어 있다고 하지 않고 불완전하거나 알 수 없다고 보고합니다. 동기화 인벤토리 답변은 호스트 계정 기준임을 밝히며 다른 계정을 물었을 때 경고하고, 요청한 계정이 등록되지 않았거나 비활성화되었거나 확인할 수 없으면 조용히 호스트 계정으로 바꾸지 않고 그 사실을 밝히며, 실시간 도구 없이 만든 답변은 실제 데이터가 아니라는 안내로 시작하고, 긴 대화는 요청 크기에 맞는 최근 대화만 보냅니다. 도메인 키워드가 없는 후속 질문("그건 왜 그래?", "더 자세히")은 일반 Ops 에이전트로 넘어가지 않고 직전 답변을 한 섹션이 이어서 답합니다. 명시적 키워드, 섹션 고정, (하이브리드 라우팅에서는) 분류기의 선택은 이전처럼 다시 라우팅합니다. AgentCore 에이전트 답변에서는 에이전트가 AWSops가 대상 계정에 대해 수집하도록 설정된 리전 목록을(조회할 수 있을 때. AWS에서 활성화된 리전이 아니라 AWSops 설정입니다) 받아, 계정 전체 질문을 한 리전만 보고 답하지 않고 해당 리전들을 조회한 뒤 확인한 리전을 밝힙니다. 한 스레드에서 계정을 바꾸면 이전 턴에 (기록된 경우) 해당 계정이 표시되어, 다른 계정에 대한 답변을 현재 계정의 사실로 제시하지 않습니다. 긴 답변에는 모델 기본값 대신 명시적인 출력 한도를 적용해 원인 분석 답변이 문장 중간에 끊기지 않으며, 그래도 일찍 끝난 답변은 출력 길이 한도에 도달했는지 연결이 끊겼는지를 알려 줍니다. 인벤토리 조회 도구가 반환하는 동기화 행에 리전이 포함되고 EC2는 인스턴스 ID·유형·전원 상태·마지막 상태 변경 시각도 포함되어(리더 뷰가 이 페이로드 키를 허용하며 실행 중 인스턴스의 상태 변경 시각은 비어 있음), 리전 한정 질문과 실행/중지 질문이 이름 접두사 추정이나 "알 수 없음" 대신 기록된 값으로 답변됩니다.

agent/lambda/clickhouse_mcp.py

@@ -17,13 +17,14 @@
 
 from cross_account import resolve_tool_name
 from datasource_http import (
+    MAX_TRACE_RESPONSE_BYTES,
     NotConnected,
     SsrfBlocked,
     assert_host_allowed,
     auth_headers,
-    health,
     http_json,
     load_datasource,
+    query_probe,
     set_request_conn,
 )
 from sql_readonly_guard import assert_read_only as _shared_assert_read_only
@@ -161,7 +162,9 @@ def _run_sql(sql, max_rows, trusted=False, max_execution_time=None):
     # HTTP timeout ALIGNED ABOVE the execution bound (+3s margin): the shared default of 12s
     # would client-abort a legitimately long query while the server kept scanning under
     # timeout_overflow_mode=throw — the server-side bound must fire first.
-    status, data = http_json("POST", url, headers=headers, body=body, timeout=max_execution_time + 3)
+    # Wide max_result_rows-bounded results can exceed the default cap.
+    status, data = http_json("POST", url, headers=headers, body=body, timeout=max_execution_time + 3,
+                             max_bytes=MAX_TRACE_RESPONSE_BYTES)
     if status >= 400:
         snippet = (data.get("raw") or data.get("exception") or data) if isinstance(data, dict) else "non-object error response"
         return err(f"ClickHouse query failed ({status}): {str(snippet)[:300]}")
@@ -273,8 +276,12 @@ def clickhouse_schema(args):
 
 
 def clickhouse_health(args):
-    """Connectivity probe for the pre-save Test / status badge: GET /ping."""
-    return ok(health(load_datasource(SLUG), "/ping"))
+    """Verify read-only SQL access using one constant row and a one-second ceiling."""
+    def run():
+        result = _run_sql("SELECT 1 AS probe", 1, max_execution_time=1)
+        return json.loads(result["body"]).get("collectionStatus") if result["statusCode"] == 200 else None
+    # "empty" is impossible for SELECT 1; "partial" means the result cap applied, which still proves access.
+    return ok(query_probe(run, "ClickHouse query access could not be verified", accepted=("ok", "partial")))
 
 
 _TOOLS = {

agent/lambda/datadog_mcp.py

@@ -16,7 +16,7 @@
 from cross_account import resolve_tool_name
 from datasource_http import (
     NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, http_json, load_datasource,
-    set_request_conn,
+    query_probe, set_request_conn, request_timeout, finite_number, observed_payload, HTTP_TIMEOUT,
 )
 
 SLUG = "datadog"
@@ -43,11 +43,13 @@ def _parse_time_s(v, default_delta_s=None):
 
 def _headers(creds):
     """DD dual-key headers: dedicated apiKey/appKey fields win; else the generic custom-header pair
-    (headerName/headerValue [+2]) from auth_headers covers users who configured them directly."""
+    (headerName/headerValue [+2]) from auth_headers covers users who configured them directly.
+    Dedicated keys apply only to a blob with no authType; any explicit type (including custom_header
+    and none) uses its own headers."""
     h = dict(auth_headers(creds))
-    if creds.get("apiKey"):
+    if creds.get("authType") is None and creds.get("apiKey"):
         h["DD-API-KEY"] = str(creds["apiKey"])
-    if creds.get("appKey"):
+    if creds.get("authType") is None and creds.get("appKey"):
         h["DD-APPLICATION-KEY"] = str(creds["appKey"])
     return h
 
@@ -58,13 +60,14 @@ def _ds():
     return creds
 
 
-def _get(creds, path, params=None):
+def _get(creds, path, params=None, *, timeout=None):
     url = creds["endpoint"].rstrip("/") + path + ("?" + urlencode(params, doseq=True) if params else "")
-    status, data = http_json("GET", url, headers=_headers(creds))
+    status, data = http_json("GET", url, headers=_headers(creds),
+                             timeout=timeout if timeout is not None else request_timeout(creds))
     if status >= 400:
         errs = data.get("errors") if isinstance(data, dict) else None
         raise _ApiError(f"Datadog HTTP {status}: {str(errs or (data.get('raw') if isinstance(data, dict) else data))[:300]}")
-    return data
+    return observed_payload(data, status)
 
 
 def datadog_query(args):
@@ -77,38 +80,76 @@ def datadog_query(args):
         "to": _parse_time_s(args.get("end")),
     }
     data = _get(_ds(), "/api/v1/query", params)
-    series = data.get("series", []) if isinstance(data, dict) else []
+    if isinstance(data, dict) and (data.get("status") == "error" or data.get("errors") or data.get("error")):
+        return ok({"series": [], "truncated": False, "collectionStatus": "error"})
+    series = data.get("series") if isinstance(data, dict) else None
+    if not isinstance(data, dict) or data.get("status") != "ok" or not isinstance(series, list):
+        return ok({"series": [], "truncated": False, "collectionStatus": "unknown"})
     out = []
+    truncated = len(series) > MAX_SERIES
+    invalid = False
     for s in series[:MAX_SERIES]:
-        if not isinstance(s, dict):
+        if not isinstance(s, dict) or not isinstance(s.get("metric"), str) or not isinstance(s.get("pointlist"), list):
+            invalid = True
             continue
-        pts = s.get("pointlist") or []
+        pts = s["pointlist"]
+        truncated = truncated or len(pts) > MAX_POINTS
+        points = []
+        for point in pts[:MAX_POINTS]:
+            if (not isinstance(point, list) or len(point) != 2
+                    or not finite_number(point[0], 8640000000000000)
+                    or point[1] is not None and not finite_number(point[1])):
+                invalid = True
+                continue
+            points.append(point)
+        units = s.get("unit")
+        unit = units[0].get("name") if isinstance(units, list) and units and isinstance(units[0], dict) else None
         out.append({
             "metric": s.get("metric", ""), "scope": s.get("scope", ""),
-            "unit": (s.get("unit") or [{}])[0].get("name") if isinstance(s.get("unit"), list) and s.get("unit") else None,
-            "pointlist": pts[:MAX_POINTS],
+            "unit": unit, "pointlist": points,
         })
-    return ok({"series": out, "truncated": len(series) > MAX_SERIES,
+    http_status = getattr(data, "http_status", 200)
+    state = "unknown" if invalid or http_status not in (200, 206) else "partial" if truncated or http_status == 206 else "ok" if out else "empty"
+    return ok({"series": out, "truncated": truncated,
+               "collectionStatus": state,
                "status": data.get("status") if isinstance(data, dict) else None})
 
 
 def datadog_schema(args):
     """Actively-reporting metric names from the last hour (query-gen vocabulary)."""
-    data = _get(_ds(), "/api/v1/metrics", {"from": _parse_time_s(None, 3600)})
+    data = _get(_ds(), "/api/v1/metrics", {"from": _parse_time_s(None, 3600)}, timeout=HTTP_TIMEOUT)
     metrics = data.get("metrics", []) if isinstance(data, dict) else []
     metrics = metrics if isinstance(metrics, list) else []
     return ok({"metrics": metrics[:200], "truncated": len(metrics) > 200})
 
 
 def datadog_health(args):
-    """Key validation probe: GET /api/v1/validate (API key), then a 1-metric list (app key scope)."""
-    creds = load_datasource(SLUG)
-    assert_host_allowed(creds["endpoint"])
-    try:
-        v = _get(creds, "/api/v1/validate")
-        return ok({"ok": bool(isinstance(v, dict) and v.get("valid"))})
-    except _ApiError as e:
-        return ok({"ok": False, "error": str(e)})
+    """Validate configured API keys, then metric-query access (key pair or bearer) under one deadline."""
+    def run():
+        creds = load_datasource(SLUG)
+        assert_host_allowed(creds["endpoint"])
+        deadline = time.monotonic() + request_timeout(creds)
+
+        def remaining():
+            left = deadline - time.monotonic()
+            if left <= 0:
+                raise TimeoutError("Datadog connection test deadline exceeded")
+            return left
+        # The API-key validator does not validate bearer access tokens. Those use
+        # the supported metric-query operation itself as the authentication proof.
+        if any(name.lower() == "dd-api-key" for name in _headers(creds)):
+            v = _get(creds, "/api/v1/validate", timeout=remaining())
+            if not isinstance(v, dict) or v.get("valid") is not True:
+                return None
+        end = int(time.time())
+        # A one-minute, single-metric query with a probe-only host filter verifies
+        # metric-query permission without fetching the fleet's real metric series.
+        probe = _get(creds, "/api/v1/query", {
+            "query": "avg:system.cpu.user{host:awsops-connection-probe}",
+            "from": str(end - 60), "to": str(end),
+        }, timeout=remaining())
+        return "ok" if isinstance(probe, dict) and probe.get("status") == "ok" and isinstance(probe.get("series"), list) else None
+    return ok(query_probe(run, "Datadog credentials or metric-query access could not be verified"))
 
 
 _TOOLS = {