(
state: WorkflowState,
*,
context: WorkflowContext,
sender_name: str = "Vuln_select",
)
| 112 | |
| 113 | |
| 114 | def vuln_select_node( |
| 115 | state: WorkflowState, |
| 116 | *, |
| 117 | context: WorkflowContext, |
| 118 | sender_name: str = "Vuln_select", |
| 119 | ) -> dict[str, Any]: |
| 120 | check_count = state["check_count"] |
| 121 | |
| 122 | if check_count == 0: |
| 123 | scan_res = message_content(state["message"][-1]) |
| 124 | vulns = parse_vulnerabilities(scan_res) |
| 125 | if vulns: |
| 126 | selected = vulns[0] |
| 127 | vuln_select_message = ( |
| 128 | "I think we can try this vulnerability. " |
| 129 | f"The vulnerability information is as follows {selected}" |
| 130 | ) |
| 131 | else: |
| 132 | vuln_select_message = "continue to select vulnerability" |
| 133 | else: |
| 134 | vulns = state["vulns"] |
| 135 | if len(vulns) > 1: |
| 136 | vulns.pop(0) |
| 137 | selected = vulns[0] if vulns else {} |
| 138 | vuln_select_message = ( |
| 139 | "I think we can try this vulnerability. " |
| 140 | f"The vulnerability information is as follows {selected}" |
| 141 | ) |
| 142 | |
| 143 | context.history.append(f"{sender_name}{check_count}_{vuln_select_message}") |
| 144 | |
| 145 | return { |
| 146 | "message": [_human_message(vuln_select_message)], |
| 147 | "sender": sender_name, |
| 148 | "vulns": vulns if check_count == 0 else state["vulns"], |
| 149 | "check_count": check_count, |
| 150 | } |
| 151 | |
| 152 | |
| 153 | def check_node( |
nothing calls this directly
no test coverage detected