feat(agent): validate ssh diagnosis receipt

apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py

@@ -383,6 +383,16 @@ def _harbor_recovery_receipt_inputs() -> list[dict[str, Any]]:
             ),
             **metadata_boundary,
         },
+        {
+            "input_id": "ssh_publickey_diagnosis_output",
+            "source": "diagnose-110-ssh-publickey-auth.sh",
+            "required_when": "diagnose_ssh_publickey_phase",
+            "purpose": (
+                "classify TCP banner, node load, and publickey auth timeout "
+                "without reading authorized_keys or asking for passwords"
+            ),
+            **metadata_boundary,
+        },
         {
             "input_id": "ssh_local_repair_output",
             "source": "recover-110-control-path-and-harbor-local.sh --check/apply",

apps/api/src/services/harbor_registry_controlled_recovery_receipt.py

@@ -21,10 +21,14 @@ def validate_harbor_registry_controlled_recovery_receipt(
     if not isinstance(receipt_payload, dict):
         raise ValueError("receipt payload must be a JSON object")
 
+    ssh_diagnosis_output = _text(
+        receipt_payload.get("ssh_publickey_diagnosis_output")
+    )
     ssh_local_output = _text(receipt_payload.get("ssh_local_repair_output"))
     watchdog_check_output = _text(receipt_payload.get("watchdog_check_output"))
     watchdog_repair_output = _text(receipt_payload.get("watchdog_repair_output"))
 
+    ssh_diagnosis = _parse_ssh_publickey_diagnosis_output(ssh_diagnosis_output)
     ssh_local = _parse_ssh_local_repair_output(ssh_local_output)
     watchdog_check = _parse_watchdog_output(watchdog_check_output)
     watchdog_repair = _parse_watchdog_output(watchdog_repair_output)
@@ -37,6 +41,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
     )
 
     active_blockers = _active_blockers(
+        ssh_diagnosis=ssh_diagnosis,
         ssh_local=ssh_local,
         watchdog_check=watchdog_check,
         watchdog_repair=watchdog_repair,
@@ -45,6 +50,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
         deploy_marker=deploy_marker,
     )
     status = _status(
+        ssh_diagnosis=ssh_diagnosis,
         ssh_local=ssh_local,
         watchdog_check=watchdog_check,
         watchdog_repair=watchdog_repair,
@@ -54,6 +60,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
         active_blockers=active_blockers,
     )
     local_console_phase_readback = _local_console_phase_readback(
+        ssh_diagnosis=ssh_diagnosis,
         ssh_local=ssh_local,
         watchdog_check=watchdog_check,
         watchdog_repair=watchdog_repair,
@@ -71,6 +78,9 @@ def validate_harbor_registry_controlled_recovery_receipt(
         "active_blocker_count": len(active_blockers),
         "input_redaction": {
             "raw_output_returned": False,
+            "ssh_publickey_diagnosis_output": _text_stats(
+                ssh_diagnosis_output
+            ),
             "ssh_local_repair_output": _text_stats(ssh_local_output),
             "watchdog_check_output": _text_stats(watchdog_check_output),
             "watchdog_repair_output": _text_stats(watchdog_repair_output),
@@ -87,6 +97,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
             },
         },
         "readback": {
+            "ssh_publickey_diagnosis": ssh_diagnosis,
             "ssh_local_repair": ssh_local,
             "watchdog_check": watchdog_check,
             "watchdog_repair": watchdog_repair,
@@ -102,6 +113,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
             "current_apply_allowed": False,
             "current_apply_blocker": _current_apply_blocker(status=status),
             "allowed_scope": [
+                "validate_non_secret_ssh_publickey_diagnosis_receipt",
                 "validate_non_secret_ssh_local_repair_receipt",
                 "validate_harbor_watchdog_check_receipt",
                 "validate_harbor_watchdog_repair_once_receipt",
@@ -123,6 +135,18 @@ def validate_harbor_registry_controlled_recovery_receipt(
         },
         "learning_writeback_contracts": _learning_writeback_contracts(status=status),
         "rollups": {
+            "ssh_publickey_diagnosis_receipt_seen": ssh_diagnosis[
+                "receipt_seen"
+            ],
+            "ssh_publickey_diagnosis_ready": ssh_diagnosis["diagnosis_ready"],
+            "ssh_publickey_wooo_publickey_classification": ssh_diagnosis[
+                "wooo_publickey_classification"
+            ],
+            "ssh_publickey_offer_timeout_seen": ssh_diagnosis[
+                "publickey_offer_timeout_seen"
+            ],
+            "ssh_publickey_node_exporter_ok": ssh_diagnosis["node_exporter_ok"],
+            "ssh_publickey_port_tcp_open": ssh_diagnosis["ssh_port_tcp_open"],
             "ssh_local_repair_receipt_seen": ssh_local["receipt_seen"],
             "ssh_local_repair_control_channel_metadata_ready": ssh_local[
                 "control_channel_metadata_ready"
@@ -193,6 +217,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
 
 def _local_console_phase_readback(
     *,
+    ssh_diagnosis: dict[str, Any],
     ssh_local: dict[str, Any],
     watchdog_check: dict[str, Any],
     watchdog_repair: dict[str, Any],
@@ -202,10 +227,10 @@ def _local_console_phase_readback(
         _phase(
             "diagnose_ssh_publickey",
             _phase_status(
-                ready=ssh_local["receipt_seen"],
+                ready=ssh_diagnosis["receipt_seen"] or ssh_local["receipt_seen"],
                 blocked_status="blocked_waiting_ssh_publickey_diagnosis_receipt",
             ),
-            "ssh_local_repair",
+            "ssh_publickey_diagnosis",
         ),
         _phase(
             "preflight_control_path_and_harbor",
@@ -217,7 +242,11 @@ def _local_console_phase_readback(
         ),
         _phase(
             "repair_ssh_metadata_if_check_confirms_metadata_drift",
-            _ssh_metadata_phase_status(ssh_local=ssh_local, watchdog_check=watchdog_check),
+            _ssh_metadata_phase_status(
+                ssh_diagnosis=ssh_diagnosis,
+                ssh_local=ssh_local,
+                watchdog_check=watchdog_check,
+            ),
             "ssh_local_repair",
         ),
         _phase(
@@ -268,6 +297,7 @@ def _phase_status(*, ready: bool, blocked_status: str) -> str:
 
 def _ssh_metadata_phase_status(
     *,
+    ssh_diagnosis: dict[str, Any],
     ssh_local: dict[str, Any],
     watchdog_check: dict[str, Any],
 ) -> str:
@@ -275,6 +305,8 @@ def _ssh_metadata_phase_status(
         if ssh_local["control_channel_metadata_ready"]:
             return "ready"
         return "blocked_ssh_metadata_repair_receipt_not_ready"
+    if ssh_diagnosis["publickey_offer_timeout_seen"]:
+        return "blocked_waiting_ssh_metadata_repair_receipt_after_publickey_timeout"
     if watchdog_check["receipt_seen"]:
         return "skipped_not_required"
     return "blocked_waiting_ssh_metadata_or_harbor_preflight_receipt"
@@ -296,6 +328,77 @@ def _harbor_repair_once_phase_status(
     return "blocked_waiting_harbor_watchdog_check_receipt"
 
 
+def _parse_ssh_publickey_diagnosis_output(output: str) -> dict[str, Any]:
+    fields = _parse_key_values(output)
+    marker_seen = "AWOOOI_110_SSH_PUBLICKEY_AUTH_DIAGNOSIS" in output
+    auth_attempts = _ssh_auth_attempts(output)
+    wooo_publickey = _auth_classification(
+        auth_attempts,
+        user="wooo",
+        mode="publickey",
+    )
+    publickey_offer_timeout_seen = any(
+        item["mode"] == "publickey"
+        and item["classification"] == "publickey_offer_timeout"
+        for item in auth_attempts
+    )
+    preauth_timeout_count = sum(
+        1 for item in auth_attempts if item["classification"] == "preauth_timeout"
+    )
+    permission_denied_count = sum(
+        1 for item in auth_attempts if item["classification"] == "permission_denied"
+    )
+    ssh_port_tcp_open = str(fields.get("SSH_PORT") or "") == "tcp_open"
+    node_exporter_ok = str(fields.get("NODE_EXPORTER") or "") == "ok"
+    return {
+        "receipt_seen": marker_seen,
+        "target": str(fields.get("TARGET") or ""),
+        "node_exporter_ok": node_exporter_ok,
+        "node_load_classifier": str(fields.get("NODE_LOAD_CLASSIFIER") or ""),
+        "node_load1_per_cpu": _float_or_none(fields.get("NODE_LOAD1_PER_CPU")),
+        "ssh_port_tcp_open": ssh_port_tcp_open,
+        "ssh_banner_seen": "SSH_BANNER=SSH-" in output,
+        "auth_attempt_count": len(auth_attempts),
+        "auth_classifications": auth_attempts,
+        "wooo_publickey_classification": wooo_publickey,
+        "publickey_offer_timeout_seen": publickey_offer_timeout_seen,
+        "preauth_timeout_count": preauth_timeout_count,
+        "permission_denied_count": permission_denied_count,
+        "diagnosis_ready": bool(marker_seen and ssh_port_tcp_open and auth_attempts),
+        "metadata_only": True,
+        "raw_output_returned": False,
+    }
+
+
+def _ssh_auth_attempts(output: str) -> list[dict[str, Any]]:
+    attempts: list[dict[str, Any]] = []
+    for line in output.splitlines():
+        if not line.startswith("SSH_AUTH "):
+            continue
+        fields = _parse_key_values(line)
+        attempts.append(
+            {
+                "user": str(fields.get("user") or ""),
+                "mode": str(fields.get("mode") or ""),
+                "rc": _int_or_none(fields.get("rc")),
+                "classification": str(fields.get("classification") or ""),
+            }
+        )
+    return attempts
+
+
+def _auth_classification(
+    attempts: list[dict[str, Any]],
+    *,
+    user: str,
+    mode: str,
+) -> str:
+    for item in attempts:
+        if item["user"] == user and item["mode"] == mode:
+            return str(item["classification"] or "")
+    return ""
+
+
 def _parse_ssh_local_repair_output(output: str) -> dict[str, Any]:
     fields = _parse_key_values(output)
     marker_seen = "AWOOOI_110_SSH_PUBLICKEY_AUTH_LOCAL_REPAIR" in output
@@ -655,6 +758,7 @@ def _deploy_marker_blockers(
 
 def _active_blockers(
     *,
+    ssh_diagnosis: dict[str, Any],
     ssh_local: dict[str, Any],
     watchdog_check: dict[str, Any],
     watchdog_repair: dict[str, Any],
@@ -663,6 +767,13 @@ def _active_blockers(
     deploy_marker: dict[str, Any],
 ) -> list[str]:
     blockers: list[str] = []
+    if ssh_diagnosis["receipt_seen"] and not ssh_diagnosis["diagnosis_ready"]:
+        blockers.append("ssh_publickey_diagnosis_receipt_not_ready")
+    if (
+        ssh_diagnosis["publickey_offer_timeout_seen"]
+        and not ssh_local["control_channel_metadata_ready"]
+    ):
+        blockers.append("ssh_publickey_offer_timeout_on_wooo")
     if ssh_local["receipt_seen"] and not ssh_local["control_channel_metadata_ready"]:
         blockers.append("ssh_local_repair_receipt_metadata_not_ready")
     if not watchdog_check["receipt_seen"]:
@@ -686,6 +797,7 @@ def _active_blockers(
 
 def _status(
     *,
+    ssh_diagnosis: dict[str, Any],
     ssh_local: dict[str, Any],
     watchdog_check: dict[str, Any],
     watchdog_repair: dict[str, Any],
@@ -716,6 +828,8 @@ def _status(
         return "harbor_watchdog_check_unhealthy_waiting_repair_once_receipt"
     if ssh_local["receipt_seen"]:
         return "ssh_local_repair_receipt_waiting_harbor_watchdog_check"
+    if ssh_diagnosis["receipt_seen"]:
+        return "ssh_publickey_diagnosis_receipt_waiting_110_local_repair_or_harbor_watchdog_check"
     return "blocked_waiting_harbor_controlled_recovery_receipt"
 
 
@@ -734,6 +848,11 @@ def _safe_next_step(*, status: str) -> str:
         return "run_harbor_watchdog_repair_once_then_submit_receipt_with_verifier"
     if status == "ssh_local_repair_receipt_waiting_harbor_watchdog_check":
         return "rerun_ssh_control_channel_probe_then_harbor_watchdog_check_mode"
+    if (
+        status
+        == "ssh_publickey_diagnosis_receipt_waiting_110_local_repair_or_harbor_watchdog_check"
+    ):
+        return "run_110_local_ssh_metadata_check_then_harbor_watchdog_check_mode"
     return "submit_non_secret_ssh_local_or_harbor_watchdog_receipt"
 
 
@@ -748,6 +867,11 @@ def _current_apply_blocker(*, status: str) -> str:
         return "repair_once_receipt_required_after_unhealthy_check"
     if status == "ssh_local_repair_receipt_waiting_harbor_watchdog_check":
         return "harbor_watchdog_check_receipt_required_after_ssh_local_repair"
+    if (
+        status
+        == "ssh_publickey_diagnosis_receipt_waiting_110_local_repair_or_harbor_watchdog_check"
+    ):
+        return "110_local_ssh_metadata_or_harbor_watchdog_check_receipt_required"
     return "registry_v2_verifier_or_controlled_receipt_required"
 
 
@@ -816,6 +940,13 @@ def _int_or_none(value: Any) -> int | None:
         return None
 
 
+def _float_or_none(value: Any) -> float | None:
+    try:
+        return float(str(value))
+    except (TypeError, ValueError):
+        return None
+
+
 def _dict(value: Any) -> dict[str, Any]:
     return value if isinstance(value, dict) else {}