Merge branch 'organize_tools' into 'main'

Organize tools in killchain @lidia9

See merge request aliasrobotics/alias_research/cai!3
This commit is contained in:
Víctor Mayoral Vilches 2025-01-10 14:06:41 +00:00
commit f9a39bfa87
22 changed files with 178 additions and 0 deletions

View File

View File

@ -0,0 +1,2 @@
#MsfModuleSelectionTool
#MsfModuleExecTool

View File

@ -0,0 +1,6 @@
# ReverseShellTool
# ReverseShellSession
# ReverseShellManager
# ShellCommandExecutor
# SessionManager
# UploadReverseShellTool

View File

@ -0,0 +1,2 @@
#SSHTestCredentialTool
#SSHRunCommandTool

View File

View File

View File

@ -0,0 +1,45 @@
"""
Here are crypto tools
"""
import base64
# # URLDecodeTool
# # HexDumpTool
# # Base64DecodeTool
# # ROT13DecodeTool
# # BinaryAnalysisTool
# def strings_command(file_path: str, ctf=None) -> str:
# """
# Extract printable strings from a binary file.
# Args:
# args: Additional arguments to pass to the strings command
# file_path: Path to the binary file to extract strings from
# Returns:
# str: The output of running the strings command
# """
# command = f'strings {file_path} '
# return run_command(ctf, command)
def decode64(input_data: str) -> str:
"""
Decode a base64-encoded string.
Args:
input_data: The base64-encoded string to decode
args: Additional arguments (not used in this function)
Returns:
str: The decoded string
"""
try:
decoded_bytes = base64.b64decode(input_data)
decoded_str = decoded_bytes.decode('utf-8')
return decoded_str
except UnicodeDecodeError as e:
print(f"Error decoding bytes to string: {e}")
return f"Error decoding bytes to string: {str(e)}"

View File

@ -0,0 +1 @@
#curl

View File

@ -0,0 +1 @@
# DirbScanTool

View File

@ -33,3 +33,23 @@ def cat_file(file_path: str, args: str = "", ctf=None) -> str:
"""
command = f'cat {args} {file_path} '
return run_command(command, ctf=ctf)
# FileSearchTool
# ListDirTool
# TextSearchTool
# FileAnalysisTool
# StringExtractionTool
# ReadFileTool
# FilePermissionsTool
# FileCompressionTool
def pwd_command(ctf=None) -> str:
"""
Retrieve the current working directory.
Returns:
str: The absolute path of the current working directory
"""
command = 'pwd'
return run_command(command, ctf=ctf)

View File

@ -0,0 +1,24 @@
"""
Here are the tools for netcat command
"""
from cai.tools.common import run_command # pylint: disable=import-error
def netcat(host: str, port: int, data: str = '',
args: str = '', ctf=None) -> str:
"""
A simple netcat tool to connect to a specified host and port.
Args:
args: Additional arguments to pass to the netcat command
host: The target host to connect to
port: The target port to connect to
data: Data to send to the host (optional)
Returns:
str: The output of running the netcat command
"""
if data:
command = f'echo "{data}" | nc -w 3 {host} {port} {args}'
else:
command = f'nc -w 3 {host} {port} {args}'
return run_command(command, ctf=ctf)

View File

@ -0,0 +1 @@
#NetworkConnectionstool in exploitFlow

View File

@ -0,0 +1,18 @@
from cai.tools.common import run_command
"""
Here are the nmap tools.
"""
def nmap(args: str, target: str, ctf=None) -> str:
"""
A simple nmap tool to scan a specified target.
Args:
args: Additional arguments to pass to the nmap command
target: The target host or IP address to scan
Returns:
str: The output of running the nmap command
"""
command = f'nmap {args} {target}'
return run_command(command, ctf=ctf)

View File

@ -0,0 +1,3 @@
# ProcessListTool
# EnvironmentVariablesTool
# SystemInformationTool

View File

@ -0,0 +1 @@
#FileDownloadTool in exploitFlow

View File

@ -0,0 +1,54 @@
"""
This is a short example of how we can solve ctf scenarios programmatically. With CAI framework
"""
# 1. Instance CAI, pentestperf and tools
from cai import CAI, Agent
from cai.tools.reconnaissance.crypto_tools import *
from cai.tools.reconnaissance.nmap import *
from cai.tools.reconnaissance.netcat import *
import pentestperf as ptt
from wasabi import color
import os
# 2. Init the CTF environment
ctf = ptt.ctf(
"picoctf_reversing_python",
subnet="192.168.9.0/24",
container_name="ancient",
ip_address="192.168.9.52"
)
ctf.start_ctf()
challenges = list(ctf.get_challenges().keys())
challenge = challenges[0]
# 3. Instance the cybersecurity AI Framework client
client = CAI(ctf=ctf)
# 4. Instance the agent/agents to solve the CTF
ctf_agent = Agent(
model="qwen2.5:14b",
name="CTF agent",
instructions="You are an CTF solver agent",
functions=[netcat, nmap, decode64], # This code can execute by LLM
# parallel_tool_calls=True, Enable multi function calling
)
messages = [{
"role": "user",
"content": "Instructions: " + ctf.get_instructions() +
"\nChallenge: " + ctf.get_challenges()[challenge] +
"\nTechniques: " + ctf.get_techniques() +
"\nExtract and return only the flag"
"\nYOU MUST RETURN THE FLAG IN PLAIN TEXT"
}]
# 5. Run the CAI
response = client.run(
agent=ctf_agent,
messages=messages,
debug=True,
brief=True)
print(response.messages[-1]["content"])
ctf.stop_ctf()