项目文件夹

文件
2024-04-21 01:09:21 +08:00

779 行
33 KiB
Python

# an automated penetration testing parser empowered by GPT
import json
import os
import sys
import textwrap
import time
import traceback
import loguru
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.shortcuts import confirm
from rich.console import Console
from rich.spinner import Spinner
from pentestgpt.config.chat_config import ChatGPTConfig
from pentestgpt.prompts.prompt_class import PentestGPTPrompt
from pentestgpt.utils.APIs.module_import import dynamic_import
from pentestgpt.utils.chatgpt import ChatGPT
from pentestgpt.utils.prompt_select import prompt_ask, prompt_select
from pentestgpt.utils.task_handler import (
local_task_entry,
localTaskCompleter,
main_task_entry,
mainTaskCompleter,
)
from pentestgpt.utils.web_parser import google_search
logger = loguru.logger
def prompt_continuation(width, line_number, wrap_count):
"""
The continuation: display line numbers and '->' before soft wraps.
Notice that we can return any kind of formatted text from here.
The prompt continuation doesn't have to be the same width as the prompt
which is displayed before the first line, but in this example we choose to
align them. The `width` input that we receive here represents the width of
the prompt.
"""
if wrap_count > 0:
return " " * (width - 3) + "-> "
text = ("- %i - " % (line_number + 1)).rjust(width)
return HTML("<strong>%s</strong>") % text
class pentestGPT:
postfix_options = {
"tool": "The input content is from a security testing tool. You need to list down all the points that are interesting to you; you should summarize it as if you are reporting to a senior penetration tester for further guidance.\n",
"user-comments": "The input content is from user comments.\n",
"web": "The input content is from web pages. You need to summarize the readable-contents, and list down all the points that can be interesting for penetration testing.\n",
"default": "The user did not specify the input source. You need to summarize based on the contents.\n",
}
options_desc = {
"tool": " Paste the output of the security test tool used",
"user-comments": "",
"web": " Paste the relevant content of a web page",
"default": " Write whatever you want, the tool will handle it",
}
def __init__(
self,
log_dir="logs",
reasoning_model="gpt-4-turbo",
parsing_model="gpt-4-turbo",
useAPI=True,
azure=False,
use_langfuse_logging=False,
):
self.log_dir = log_dir
logger.add(sink=os.path.join(log_dir, "pentestGPT.log"))
self.save_dir = "test_history"
self.task_log = (
{}
) # the information that can be saved to continue in the next session
self.useAPI = useAPI
self.parsing_char_window = 16000 # the chunk size for parsing in # of chars
# TODO: link the parsing_char_window to the model used
# load the module
reasoning_model_object = dynamic_import(
reasoning_model, self.log_dir, use_langfuse_logging=use_langfuse_logging
)
generation_model_object = dynamic_import(
reasoning_model, self.log_dir, use_langfuse_logging=use_langfuse_logging
)
parsing_model_object = dynamic_import(
parsing_model, self.log_dir, use_langfuse_logging=use_langfuse_logging
)
if useAPI is False: # deprecated usage of cookie
self.parsingAgent = ChatGPT(ChatGPTConfig(log_dir=self.log_dir))
self.reasoningAgent = ChatGPT(
ChatGPTConfig(model=reasoning_model, log_dir=self.log_dir)
)
else:
self.parsingAgent = parsing_model_object
self.generationAgent = generation_model_object
self.reasoningAgent = reasoning_model_object
self.prompts = PentestGPTPrompt
self.console = Console()
self.spinner = Spinner("line", "Processing")
self.test_generation_session_id = None
self.test_reasoning_session_id = None
self.input_parsing_session_id = None
self.chat_count = 0
self.step_reasoning = (
None # the response from the reasoning session for the current step
)
self.history = {
"user": [],
"pentestGPT": [],
"reasoning": [],
"input_parsing": [],
"generation": [],
"exception": [],
} # the history of the current conversation
# print the initialization message on the current implementation.
self.console.print(
"Welcome to pentestGPT, an automated penetration testing parser empowered by GPT.",
style="bold green",
)
self.console.print("The settings are: ")
self.console.print(
f" - parsing model: {parsing_model_object.name}", style="bold green"
)
self.console.print(
f" - reasoning model: {reasoning_model_object.name}", style="bold green"
)
self.console.print(f" - use API: {useAPI}", style="bold green")
self.console.print(f" - log directory: {log_dir}", style="bold green")
def log_conversation(self, source, text):
"""
append the conversation into the history
Parameters:
----------
source: str
the source of the conversation
text: str
the content of the conversation
"""
# append the conversation into the history
timestamp = time.time()
if source not in self.history.keys():
# an exception
source = "exception"
self.history[source].append((timestamp, text))
def refresh_session(self):
if self.useAPI:
self.console.print(
"You're using API mode, so no need to refresh the session."
)
self.log_conversation(
"pentestGPT",
"You're using API mode, so no need to refresh the session.",
)
else:
self.console.print(
"Please ensure that you put the curl command into `config/chatgpt_config_curl.txt`",
style="bold green",
)
self.log_conversation(
"pentestGPT",
"Please ensure that you put the curl command into `config/chatgpt_config_curl.txt`",
)
input("Press Enter to continue...")
self.parsingAgent.refresh()
self.reasoningAgent.refresh()
self.console.print(
"Session refreshed. If you receive the same session refresh request, please refresh the ChatGPT page and paste the new curl request again.",
style="bold green",
)
self.log_conversation("pentestGPT", "Session refreshed.")
return "Session refreshed."
def _feed_init_prompts(self):
# 1. User firstly provide basic information of the task
init_description = prompt_ask(
"Please describe the penetration testing task in one line, including the target IP, task type, etc.\n> ",
multiline=False,
)
self.log_conversation("user", init_description)
self.task_log["task description"] = init_description
# 2. Provide the information to the reasoning session for the task initialization.
# Note that this information is not parsed by the three-step process in reasoning.
# It is directly used to initialize the task.
prefixed_init_description = self.prompts.task_description + init_description
with self.console.status(
"[bold green] Constructing Initial Penetration Testing Tree..."
) as status:
_reasoning_response = self.reasoningAgent.send_message(
prefixed_init_description, self.test_reasoning_session_id
)
# 3. Pass to generation session for more details.
# Note that the generation session is not used for the task initialization.
with self.console.status("[bold green] Generating Initial Task") as status:
_generation_response = self.generationAgent.send_message(
self.prompts.todo_to_command + _reasoning_response,
self.test_generation_session_id,
)
# Display the initial generation result
response = _reasoning_response + "\n" + _generation_response
self.console.print("PentestGPT output: ", style="bold green")
self.console.print(response)
self.log_conversation("PentestGPT", "PentestGPT output:" + response)
def initialize(self, previous_session_ids=None):
# initialize the backbone sessions and test the connection to chatGPT
# define three sessions: testGenerationSession, testReasoningSession, and InputParsingSession
if previous_session_ids is not None and self.useAPI is False:
self.test_generation_session_id = previous_session_ids.get(
"test_generation", None
)
self.test_reasoning_session_id = previous_session_ids.get("reasoning", None)
self.input_parsing_session_id = previous_session_ids.get("parsing", None)
# debug the three sessions
print(f"Previous session ids: {str(previous_session_ids)}")
print(f"Test generation session id: {str(self.test_generation_session_id)}")
print(f"Test reasoning session id: {str(self.test_reasoning_session_id)}")
print(f"Input parsing session id: {str(self.input_parsing_session_id)}")
print("-----------------")
self.task_log = previous_session_ids.get("task_log", {})
self.console.print(f"Task log: {str(self.task_log)}", style="bold green")
print("You may use discussion function to remind yourself of the task.")
## verify that all the sessions are not None
if (
self.test_generation_session_id is None
or self.test_reasoning_session_id is None
or self.input_parsing_session_id is None
):
self.console.print(
"[bold red] Error: the previous session ids are not valid. Loading new sessions"
)
self.initialize()
else:
with self.console.status(
"[bold green] Initialize ChatGPT Sessions..."
) as status:
try:
(
text_0,
self.test_generation_session_id,
) = self.generationAgent.send_new_message(
self.prompts.generation_session_init,
)
(
text_1,
self.test_reasoning_session_id,
) = self.reasoningAgent.send_new_message(
self.prompts.reasoning_session_init
)
(
text_2,
self.input_parsing_session_id,
) = self.parsingAgent.send_new_message(
self.prompts.input_parsing_init
)
except Exception as e:
logger.error(e)
self.console.print("- ChatGPT Sessions Initialized.", style="bold green")
self._feed_init_prompts()
def reasoning_handler(self, text) -> str:
# summarize the contents if necessary.
if len(text) > self.parsing_char_window:
text = self.input_parsing_handler(text)
"""
# pass the information to reasoning_handler and obtain the results
response = self.reasoningAgent.send_message(
self.prompts.process_results + text, self.test_reasoning_session_id
)
# log the conversation
"""
# PentestGPT Reasoning Logic
## 1. Given the information, update the PTT
_updated_ptt_response = self.reasoningAgent.send_message(
self.prompts.process_results + text, self.test_reasoning_session_id
)
## 2. Validate if the PTT is correct
# TODO
## 3. If the PTT is correct, select all the to-dos
_task_selection_response = self.reasoningAgent.send_message(
self.prompts.process_results_task_selection, self.test_reasoning_session_id
)
# get the complete output:
response = _updated_ptt_response + _task_selection_response
self.log_conversation("reasoning", response)
return response
def input_parsing_handler(self, text, source=None) -> str:
prefix = "Please summarize the following input. "
# do some engineering trick here. Add postfix to the input to make it more understandable by LLMs.
if source is not None and source in self.postfix_options.keys():
prefix += self.postfix_options[source]
# The default token-size limit is 4096 (web UI even shorter). 1 token ~= 4 chars in English
# Use textwrap to split inputs. Limit to 2000 token (8000 chars) for each input
# (1) replace all the newlines with spaces
text = text.replace("\r", " ").replace("\n", " ")
# (2) wrap the text
wrapped_text = textwrap.fill(text, 8000)
wrapped_inputs = wrapped_text.split("\n")
# (3) send the inputs to chatGPT input_parsing_session and obtain the results
summarized_content = ""
for wrapped_input in wrapped_inputs:
word_limit = f"Please ensure that the input is less than {8000 / len(wrapped_inputs)} words.\n"
summarized_content += self.parsingAgent.send_message(
prefix + word_limit + wrapped_input, self.input_parsing_session_id
)
# log the conversation
self.log_conversation("input_parsing", summarized_content)
return summarized_content
def test_generation_handler(self, text):
# send the contents to chatGPT test_generation_session and obtain the results
response = self.generationAgent.send_message(
text, self.test_generation_session_id
)
# log the conversation
self.log_conversation("generation", response)
return response
def local_input_handler(self) -> str:
"""
Request for user's input to handle the local task
"""
local_task_response = ""
self.chat_count += 1
local_request_option = local_task_entry()
self.log_conversation("user", local_request_option)
if local_request_option == "help":
print(localTaskCompleter().task_details)
elif local_request_option == "discuss":
## (1) Request for user multi-line input
self.console.print(
"Please share your findings and questions with PentestGPT."
)
self.log_conversation(
"pentestGPT",
"Please share your findings and questions with PentestGPT. (End with <shift + right-arrow>)",
)
user_input = prompt_ask("Your input: ", multiline=True)
self.log_conversation("user", user_input)
## (2) pass the information to the reasoning session.
with self.console.status("[bold green] PentestGPT Thinking...") as status:
local_task_response = self.test_generation_handler(
self.prompts.local_task_prefix + user_input
)
## (3) print the results
self.console.print("PentestGPT:\n", style="bold green")
self.console.print(local_task_response + "\n", style="yellow")
self.log_conversation("pentestGPT", local_task_response)
elif local_request_option == "brainstorm":
## (1) Request for user multi-line input
self.console.print(
"Please share your concerns and questions with PentestGPT."
)
self.log_conversation(
"pentestGPT",
"Please share your concerns and questions with PentestGPT. End with <shift + right-arrow>)",
)
user_input = prompt_ask("Your input: ", multiline=True)
self.log_conversation("user", user_input)
## (2) pass the information to the reasoning session.
with self.console.status("[bold green] PentestGPT Thinking...") as status:
local_task_response = self.test_generation_handler(
self.prompts.local_task_brainstorm + user_input
)
## (3) print the results
self.console.print("PentestGPT:\n", style="bold green")
self.console.print(local_task_response + "\n", style="yellow")
self.log_conversation("pentestGPT", local_task_response)
elif local_request_option == "google":
# get the users input
self.console.print(
"Please enter your search query. PentestGPT will summarize the info from google. (End with <shift + right-arrow>) ",
style="bold green",
)
self.log_conversation(
"pentestGPT",
"Please enter your search query. PentestGPT will summarize the info from google.",
)
user_input = prompt_ask("Your input: ", multiline=False)
self.log_conversation("user", user_input)
with self.console.status("[bold green] PentestGPT Thinking...") as status:
# query the question
result: dict = google_search(user_input, 5) # 5 results by default
# summarize the results
# TODO
local_task_response = (
"Google search results:\n" + "still under development."
)
self.console.print(local_task_response + "\n", style="yellow")
self.log_conversation("pentestGPT", local_task_response)
return local_task_response
elif local_request_option == "continue":
self.console.print("Exit the local task and continue the main task.")
self.log_conversation(
"pentestGPT", "Exit the local task and continue the main task."
)
local_task_response = "continue"
return local_task_response
def input_handler(self) -> str:
"""
Request for user's input to:
(1) input test results,
(2) ask for todos,
(3) input other information (discuss),
(4) google.
(4) end.
The design details are based on PentestGPT_design.md
Return
-----
response: str
The response from the chatGPT model.
"""
self.chat_count += 1
request_option = main_task_entry()
self.log_conversation("user", request_option)
# always check if the session expires.
# check if session expires
if not self.useAPI:
conversation_history = self.parsingAgent.get_conversation_history()
while conversation_history is None:
self.refresh_session()
conversation_history = self.parsingAgent.get_conversation_history()
if request_option == "help":
print(mainTaskCompleter().task_details)
if request_option == "next":
## (1) pass the information to input_parsing session.
## Give an option list for user to choose from
options = list(self.postfix_options.keys())
opt_desc = list(self.options_desc.values())
value_list = [
(
i,
HTML(
f'<style fg="cyan">{options[i]}</style><style fg="LightSeaGreen">{opt_desc[i]}</style>'
),
)
for i in range(len(options))
]
source = prompt_select(
title="Please choose the source of the information.", values=value_list
)
self.console.print(
"Your input: (End with <shift + right-arrow>)", style="bold green"
)
user_input = prompt_ask("> ", multiline=True)
self.log_conversation(
"user", f"Source: {options[int(source)]}" + "\n" + user_input
)
with self.console.status("[bold green] PentestGPT Thinking...") as status:
parsed_input = self.input_parsing_handler(
user_input, source=options[int(source)]
)
## (2) pass the summarized information to the reasoning session.
reasoning_response = self.reasoning_handler(parsed_input)
self.step_reasoning_response = reasoning_response
## (3) print the results
self.console.print(
"Based on the analysis, the following tasks are recommended:",
style="bold green",
)
self.console.print(reasoning_response + "\n")
self.log_conversation(
"pentestGPT",
"Based on the analysis, the following tasks are recommended:"
+ reasoning_response,
)
response = reasoning_response
elif request_option == "more":
self.log_conversation("user", "more")
## (1) check if reasoning session is initialized
if not hasattr(self, "step_reasoning_response"):
self.console.print(
"You have not initialized the task yet. Please perform the basic testing following `next` option.",
style="bold red",
)
response = "You have not initialized the task yet. Please perform the basic testing following `next` option."
self.log_conversation("pentestGPT", response)
return response
## (2) start local task generation.
### (2.1) ask the reasoning session to analyze the current situation, and explain the task
self.console.print(
"PentestGPT will generate more test details, and enter the sub-task generation mode. (Pressing Enter to continue)",
style="bold green",
)
self.log_conversation(
"pentestGPT",
"PentestGPT will generate more test details, and enter the sub-task generation mode.",
)
input()
### (2.2) pass the sub-tasks to the test generation session
with self.console.status("[bold green] PentestGPT Thinking...") as status:
generation_response = self.test_generation_handler(
self.step_reasoning_response
)
_local_init_response = self.test_generation_handler(
self.prompts.local_task_init
)
self.console.print(
"Below are the further details.",
style="bold green",
)
self.console.print(generation_response + "\n")
response = generation_response
self.log_conversation("pentestGPT", response)
### (2.3) local task handler
while True:
local_task_response = self.local_input_handler()
if local_task_response == "continue":
# break the local task handler
break
elif request_option == "todo":
## log that user is asking for todo list
self.log_conversation("user", "todo")
## (1) ask the reasoning session to analyze the current situation, and list the top sub-tasks
with self.console.status("[bold green] PentestGPT Thinking...") as status:
reasoning_response = self.reasoning_handler(self.prompts.ask_todo)
## (2) pass the sub-tasks to the test_generation session.
message = self.prompts.todo_to_command + "\n" + reasoning_response
generation_response = self.test_generation_handler(message)
## (3) print the results
self.console.print(
"Based on the analysis, the following tasks are recommended:",
style="bold green",
)
self.console.print(reasoning_response + "\n")
self.console.print(
"You can follow the instructions below to complete the tasks.",
style="bold green",
)
self.console.print(generation_response + "\n")
response = reasoning_response
self.log_conversation(
"pentestGPT",
(
(
(
(
"Based on the analysis, the following tasks are recommended:"
+ response
)
+ "\n"
)
+ "You can follow the instructions below to complete the tasks."
)
+ generation_response
),
)
elif request_option == "discuss":
## (1) Request for user multi-line input
self.console.print(
"Please share your thoughts/questions with PentestGPT. (End with <shift + right-arrow>) "
)
self.log_conversation(
"pentestGPT", "Please share your thoughts/questions with PentestGPT."
)
user_input = prompt_ask("Your input: ", multiline=True)
self.log_conversation("user", user_input)
## (2) pass the information to the reasoning session.
with self.console.status("[bold green] PentestGPT Thinking...") as status:
response = self.reasoning_handler(self.prompts.discussion + user_input)
## (3) print the results
self.console.print("PentestGPT:\n", style="bold green")
self.console.print(response + "\n", style="yellow")
self.log_conversation("pentestGPT", response)
elif request_option == "google":
# get the users input
self.console.print(
"Please enter your search query. PentestGPT will summarize the info from google. (End with <shift + right-arrow>) ",
style="bold green",
)
self.log_conversation(
"pentestGPT",
"Please enter your search query. PentestGPT will summarize the info from google.",
)
user_input = prompt_ask("Your input: ", multiline=False)
self.log_conversation("user", user_input)
with self.console.status("[bold green] PentestGPT Thinking...") as status:
# query the question
result: dict = google_search(user_input, 5) # 5 results by default
# summarize the results
# TODO
response = "Google search results:\n" + "still under development."
self.console.print(response + "\n", style="yellow")
self.log_conversation("pentestGPT", response)
return response
elif request_option == "quit":
response = False
self.console.print("Thank you for using PentestGPT!", style="bold green")
self.log_conversation("pentestGPT", "Thank you for using PentestGPT!")
else:
self.console.print("Please key in the correct options.", style="bold red")
self.log_conversation("pentestGPT", "Please key in the correct options.")
response = "Please key in the correct options."
return response
def save_session(self):
"""
Save the current session for next round of usage.
The test information is saved in the directory `./test_history`
"""
self.console.print(
"Before you quit, you may want to save the current session.",
style="bold green",
)
# 1. Require a save name from the user. If not, use the current time as the save name.
save_name = prompt_ask(
"Please enter the name of the current session. (Default with current timestamp)\n> ",
multiline=False,
)
if save_name == "":
save_name = str(time.time())
# 2. Save the current session
with open(
os.path.join(
os.path.realpath(os.path.dirname(__file__)),
os.pardir,
os.pardir,
self.save_dir,
save_name,
),
"w",
) as f:
# store the three ids and task_log
session_ids = {
"reasoning": self.test_reasoning_session_id,
"test_generation": self.test_generation_session_id,
"parsing": self.input_parsing_session_id,
"task_log": self.task_log,
}
json.dump(session_ids, f)
self.console.print(
f"The current session is saved as {save_name}", style="bold green"
)
return
def _preload_session(self) -> dict:
"""
Preload the session from the save directory.
Returns:
dict: the session ids for the three sessions.
None if no previous session is found.
"""
if continue_from_previous := confirm(
"Do you want to continue from previous session?"
):
# load the filenames from the save directory
filenames = os.listdir(
os.path.join(
os.path.realpath(os.path.dirname(__file__)),
os.pardir,
os.pardir,
self.save_dir,
)
)
if len(filenames) == 0:
print("No previous session found. Please start a new session.")
return None
else: # print all the files
print("Please select the previous session by its index (integer):")
for i, filename in enumerate(filenames):
print(f"{str(i)}. {filename}")
# ask for the user input
try:
previous_testing_name = filenames[
int(input("Please key in your option (integer): "))
]
print(f"You selected: {previous_testing_name}")
except ValueError as e:
print("You input an invalid option. Will start a new session.")
return None
elif continue_from_previous is False:
return None
else:
print("You input an invalid option. Will start a new session.")
return None
# 2. load the previous session information
if previous_testing_name is not None:
# try to load the file content with json
try:
with open(
os.path.join(
os.path.realpath(os.path.dirname(__file__)),
os.pardir,
os.pardir,
self.save_dir,
previous_testing_name,
),
"r",
) as f:
return json.load(f)
except Exception as e:
print(
"Error when loading the previous session. The file name is not correct"
)
print(e)
previous_testing_name = None
return None
def main(self):
"""
The main function of pentestGPT. The design is based on PentestGPT_design.md
"""
# 0. initialize the backbone sessions and test the connection to chatGPT
loaded_ids = self._preload_session()
self.initialize(previous_session_ids=loaded_ids)
# enter the main loop.
while True:
try:
result = self.input_handler()
self.console.print(
"-----------------------------------------", style="bold white"
)
if not result: # end the session
break
except Exception as e: # catch all general exception.
# log the exception
self.log_conversation("exception", str(e))
# print the exception
self.console.print(f"Exception: {str(e)}", style="bold red")
# add a more detailed debugging
exc_type, exc_obj, exc_tb = sys.exc_info()
fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
self.console.print(
"Exception details are below. You may submit an issue on github and paste the error trace",
style="bold green",
)
# self.console.print(exc_type, fname, exc_tb.tb_lineno)
print(traceback.format_exc())
# safely quit the session
break
# log the session. Save self.history into a txt file based on timestamp
timestamp = time.time()
log_name = f"pentestGPT_log_{str(timestamp)}.txt"
# save it in the logs folder
log_path = os.path.join(self.log_dir, log_name)
with open(log_path, "w") as f:
json.dump(self.history, f)
# save the sessions; continue from previous testing
self.save_session()
if __name__ == "__main__":
pentestGPT = pentestGPT()
pentestGPT.main()