diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py index 4146921ed..909e2a1d5 100644 --- a/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py +++ b/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", diff --git a/apps/api/src/services/harbor_registry_controlled_recovery_receipt.py b/apps/api/src/services/harbor_registry_controlled_recovery_receipt.py index 13fcbfd6d..7c4275423 100644 --- a/apps/api/src/services/harbor_registry_controlled_recovery_receipt.py +++ b/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 {} diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py index 713371d37..4192632e4 100644 --- a/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py +++ b/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py @@ -68,7 +68,7 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False): assert payload["rollups"]["current_blocker_local_recovery_package_count"] == 1 assert ( payload["rollups"]["current_blocker_harbor_recovery_receipt_input_count"] - == 7 + == 8 ) assert payload["rollups"]["runtime_dispatch_performed"] is False @@ -127,12 +127,13 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False): assert current_queue[0]["harbor_recovery_receipt_endpoint"] == ( "/api/v1/agents/harbor-registry-controlled-recovery-receipt" ) - assert current_queue[0]["harbor_recovery_receipt_input_count"] == 7 + assert current_queue[0]["harbor_recovery_receipt_input_count"] == 8 assert [ item["input_id"] for item in current_queue[0]["harbor_recovery_receipt_inputs"] ] == [ "gitea_actions_queue_readback", + "ssh_publickey_diagnosis_output", "ssh_local_repair_output", "watchdog_check_output", "watchdog_repair_output", diff --git a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py index b9c69c3dc..208488b95 100644 --- a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py +++ b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py @@ -321,9 +321,10 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu assert evidence["ai_loop_current_blocker_harbor_recovery_receipt_endpoint"] == ( "/api/v1/agents/harbor-registry-controlled-recovery-receipt" ) - assert evidence["ai_loop_current_blocker_harbor_recovery_receipt_input_count"] == 7 + assert evidence["ai_loop_current_blocker_harbor_recovery_receipt_input_count"] == 8 assert evidence["ai_loop_current_blocker_harbor_recovery_receipt_input_ids"] == [ "gitea_actions_queue_readback", + "ssh_publickey_diagnosis_output", "ssh_local_repair_output", "watchdog_check_output", "watchdog_repair_output", @@ -357,7 +358,7 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu payload["summary"][ "ai_loop_current_blocker_harbor_recovery_receipt_input_count" ] - == 7 + == 8 ) assert payload["summary"][ "ai_loop_current_blocker_harbor_recovery_receipt_input_ids" diff --git a/apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py b/apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py index 4cc4b2217..5743d0c38 100644 --- a/apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py +++ b/apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py @@ -103,6 +103,45 @@ def test_harbor_recovery_receipt_routes_unhealthy_check_to_repair_once() -> None assert payload["controlled_apply_policy"]["manual_end_state"] is False +def test_harbor_recovery_receipt_accepts_ssh_publickey_diagnosis() -> None: + payload = validate_harbor_registry_controlled_recovery_receipt( + { + "ssh_publickey_diagnosis_output": _ssh_publickey_diagnosis_output(), + } + ) + + assert payload["status"] == ( + "ssh_publickey_diagnosis_receipt_waiting_110_local_repair_or_harbor_watchdog_check" + ) + assert payload["safe_next_step"] == ( + "run_110_local_ssh_metadata_check_then_harbor_watchdog_check_mode" + ) + assert "ssh_publickey_offer_timeout_on_wooo" in payload["active_blockers"] + diagnosis = payload["readback"]["ssh_publickey_diagnosis"] + assert diagnosis["diagnosis_ready"] is True + assert diagnosis["node_exporter_ok"] is True + assert diagnosis["ssh_port_tcp_open"] is True + assert diagnosis["ssh_banner_seen"] is True + assert diagnosis["wooo_publickey_classification"] == "publickey_offer_timeout" + assert diagnosis["publickey_offer_timeout_seen"] is True + assert diagnosis["raw_output_returned"] is False + phases = { + phase["phase_id"]: phase + for phase in payload["local_console_phase_readback"]["phases"] + } + assert phases["diagnose_ssh_publickey"]["status"] == "ready" + assert phases["repair_ssh_metadata_if_check_confirms_metadata_drift"][ + "status" + ] == "blocked_waiting_ssh_metadata_repair_receipt_after_publickey_timeout" + assert payload["rollups"]["ssh_publickey_diagnosis_receipt_seen"] is True + assert payload["rollups"]["ssh_publickey_wooo_publickey_classification"] == ( + "publickey_offer_timeout" + ) + assert payload["input_redaction"]["ssh_publickey_diagnosis_output"][ + "line_count" + ] > 0 + + def test_harbor_recovery_receipt_surfaces_gitea_queue_blockers() -> None: payload = validate_harbor_registry_controlled_recovery_receipt( { @@ -303,6 +342,24 @@ SSH_RELOAD=skipped """ +def _ssh_publickey_diagnosis_output() -> str: + return """ +AWOOOI_110_SSH_PUBLICKEY_AUTH_DIAGNOSIS +TARGET=192.168.0.110:22 +NODE_EXPORTER=ok +NODE_LOAD1_PER_CPU=0.93 +NODE_LOAD_CLASSIFIER=load_not_high +SSH_PORT=tcp_open +SSH_BANNER=SSH-2.0-OpenSSH_8.9p1 +SSH_AUTH user=wooo mode=publickey rc=255 classification=publickey_offer_timeout +SSH_AUTH user=root mode=publickey rc=255 classification=permission_denied +SSH_AUTH user=git mode=publickey rc=255 classification=preauth_timeout +SSH_AUTH user=ollama mode=publickey rc=255 classification=preauth_timeout +SSH_AUTH user=wooo mode=password_disabled rc=255 classification=permission_denied +INTERPRETATION=publickey_offer_timeout_on_wooo_means_check_110_authorized_keys_permissions_pam_or_account_lookup_path +""" + + def _watchdog_check_output(*, ready: bool, status: int) -> str: ready_text = "true" if ready else "false" return f"""