feat: modernize legacy PentestGPT with native multi-LLM support - #469
Merged
Merged
Conversation
Rebuild the classic USENIX-2024 interactive PentestGPT (reasoning / generation / parsing sessions + Pentesting Task Tree + REPL) as a standalone `pentestgpt_legacy` package on a native per-provider LLM layer that supports the latest 2026 models. - llm/: BaseProvider + OpenAI-compatible / Anthropic / Gemini connectors, a web-verified model registry (OpenAI, Anthropic, Gemini, DeepSeek, xAI, Qwen, Moonshot, local Ollama), a factory, and an LLMClient bridging async providers to the core's synchronous send_new_message/send_message session API. - CLI `pentestgpt-legacy`: --list-models and --smoke-test (live per-model round-trip matrix), plus --reasoning-model / --parsing-model / --base-url. - Tests: 25 unit tests (mocked, no network). Live smoke test verified 22/22 models with a configured key respond. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request restructures the repository to separate the autonomous agent (pentestgpt/) from the modernized legacy interactive mode (pentestgpt_legacy/), introducing a multi-stage pipeline orchestrator and updating the Docker environment to use uv. The legacy mode is rebuilt with a native per-provider LLM layer supporting modern models. The code review identified several critical issues, including an unawaited coroutine in the backend process cleanup, potential type errors when handling flag dictionaries in the pipeline and stage prompts, and unhandled exceptions in the synchronous runner thread. Additionally, the reviewer recommended optimizing LLM client instantiation across the Anthropic, Gemini, and OpenAI-compatible providers to prevent connection overhead.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
Comment on lines
+148
to
+150
| # Non-async wait — safe even during event loop teardown | ||
| with contextlib.suppress(Exception): | ||
| proc.wait() # type: ignore[unused-coroutine] |
There was a problem hiding this comment.
In _kill_process, proc.wait() is called without await. Since proc.wait() is an asynchronous coroutine in asyncio.subprocess.Process, calling it without await does not wait for or reap the process, and it will raise a RuntimeWarning: coroutine 'Process.wait' was never awaited.
To perform a non-blocking synchronous wait safely during event loop teardown, we should wait on the underlying subprocess.Popen object directly.
Suggested change
| # Non-async wait — safe even during event loop teardown | |
| with contextlib.suppress(Exception): | |
| proc.wait() # type: ignore[unused-coroutine] | |
| # Non-async wait — safe even during event loop teardown | |
| with contextlib.suppress(Exception): | |
| if proc._transport: | |
| popen = proc._transport.get_extra_info('subprocess') | |
| if popen: | |
| popen.wait() |
Comment on lines
+93
to
+97
| for f in r.flags_found: | ||
| if f not in seen: | ||
| seen.add(f) | ||
| flags.append(f) | ||
| return flags |
There was a problem hiding this comment.
If r.flags_found contains dictionary objects (e.g., {"flag": "...", "context": "..."}), iterating over them and adding them directly to the seen set will raise TypeError: unhashable type: 'dict'.
We should extract the flag string from f if it is a dictionary before checking or adding it to the seen set.
Suggested change
| for f in r.flags_found: | |
| if f not in seen: | |
| seen.add(f) | |
| flags.append(f) | |
| return flags | |
| for r in self.stage_results: | |
| for f in r.flags_found: | |
| flag_str = f.get("flag", f) if isinstance(f, dict) else f | |
| if flag_str not in seen: | |
| seen.add(flag_str) | |
| flags.append(flag_str) |
Comment on lines
+142
to
+143
| parts.append(f"Flags found: {', '.join(result.flags_found)}") | ||
|
|
There was a problem hiding this comment.
If result.flags_found contains dictionary objects, calling ', '.join(result.flags_found) will raise TypeError: sequence item 0: expected str instance, dict found.
We should extract the flag string from each element in result.flags_found before joining them.
Suggested change
| parts.append(f"Flags found: {', '.join(result.flags_found)}") | |
| if result.flags_found: | |
| flag_strs = [f.get("flag", f) if isinstance(f, dict) else f for f in result.flags_found] | |
| parts.append(f"Flags found: {', '.join(flag_strs)}") |
Comment on lines
+39
to
+47
| box: dict[str, Any] = {} | ||
|
|
||
| def _runner() -> None: | ||
| box["value"] = asyncio.run(coro) | ||
|
|
||
| thread = threading.Thread(target=_runner) | ||
| thread.start() | ||
| thread.join() | ||
| return box["value"] |
There was a problem hiding this comment.
In run_sync, the _runner thread function executes asyncio.run(coro) but does not capture or propagate any exceptions raised during execution. If an exception occurs (e.g., network timeout, authentication failure), the thread will terminate silently, and the main thread will raise a confusing KeyError: 'value' when trying to access box["value"].
We should catch exceptions in the thread, store them in box, and re-raise them in the main thread to preserve the original traceback and error message.
Suggested change
| box: dict[str, Any] = {} | |
| def _runner() -> None: | |
| box["value"] = asyncio.run(coro) | |
| thread = threading.Thread(target=_runner) | |
| thread.start() | |
| thread.join() | |
| return box["value"] | |
| box: dict[str, Any] = {} | |
| def _runner() -> None: | |
| try: | |
| box["value"] = asyncio.run(coro) | |
| except Exception as e: | |
| box["error"] = e | |
| thread = threading.Thread(target=_runner) | |
| thread.start() | |
| thread.join() | |
| if "error" in box: | |
| raise box["error"] | |
| return box["value"] |
Comment on lines
+17
to
+45
| from pentestgpt_legacy.llm.registry import ModelSpec | ||
|
|
||
|
|
||
| class OpenAICompatibleProvider(BaseProvider): | ||
| """Async OpenAI-compatible provider (Chat Completions, optional Responses API).""" | ||
|
|
||
| async def acomplete( | ||
| self, | ||
| messages: list[Message], | ||
| system: str | None, | ||
| spec: ModelSpec, | ||
| *, | ||
| max_output_tokens: int | None = None, | ||
| temperature: float | None = None, | ||
| ) -> str: | ||
| try: | ||
| from openai import AsyncOpenAI | ||
| except ImportError as exc: # pragma: no cover - install-time guard | ||
| raise RuntimeError( | ||
| "The 'openai' package is required. Install with: uv add openai" | ||
| ) from exc | ||
|
|
||
| # Ollama needs a non-empty key string but ignores its value. | ||
| api_key = self.api_key or "not-needed" | ||
| async with AsyncOpenAI(api_key=api_key, base_url=self.base_url) as client: | ||
| if spec.responses_api: | ||
| return await self._via_responses( | ||
| client, messages, system, spec, max_output_tokens, temperature | ||
| ) |
There was a problem hiding this comment.
Instantiating a new AsyncOpenAI client on every call to acomplete is highly inefficient. It creates a new httpx.AsyncClient and connection pool each time, requiring a new TCP and TLS handshake for every single request, which adds significant latency and prevents connection reuse.
We should instantiate the client once in __init__ and reuse it across requests.
class OpenAICompatibleProvider(BaseProvider):
"""Async OpenAI-compatible provider (Chat Completions, optional Responses API)."""
def __init__(self, info: ProviderInfo, api_key: str | None, base_url: str | None):
super().__init__(info, api_key, base_url)
from openai import AsyncOpenAI
self.client = AsyncOpenAI(api_key=api_key or "not-needed", base_url=base_url)
async def acomplete(
self,
messages: list[Message],
system: str | None,
spec: ModelSpec,
*,
max_output_tokens: int | None = None,
temperature: float | None = None,
) -> str:
if spec.responses_api:
return await self._via_responses(
self.client, messages, system, spec, max_output_tokens, temperature
)
return await self._via_chat(
self.client, messages, system, spec, max_output_tokens, temperature
)
Comment on lines
+13
to
+43
| from pentestgpt_legacy.llm.registry import ModelSpec | ||
|
|
||
| # Anthropic requires an explicit output cap; generous enough for detailed steps. | ||
| _DEFAULT_MAX_TOKENS = 8192 | ||
|
|
||
|
|
||
| class AnthropicProvider(BaseProvider): | ||
| """Async Anthropic Messages API provider.""" | ||
|
|
||
| async def acomplete( | ||
| self, | ||
| messages: list[Message], | ||
| system: str | None, | ||
| spec: ModelSpec, | ||
| *, | ||
| max_output_tokens: int | None = None, | ||
| temperature: float | None = None, | ||
| ) -> str: | ||
| try: | ||
| from anthropic import AsyncAnthropic | ||
| except ImportError as exc: # pragma: no cover - install-time guard | ||
| raise RuntimeError( | ||
| "The 'anthropic' package is required. Install with: uv add anthropic" | ||
| ) from exc | ||
|
|
||
| kwargs: dict[str, Any] = { | ||
| "model": spec.api_id, | ||
| "messages": list(messages), | ||
| "max_tokens": max_output_tokens or _DEFAULT_MAX_TOKENS, | ||
| } | ||
| if system: |
There was a problem hiding this comment.
Instantiating a new AsyncAnthropic client on every call to acomplete is highly inefficient. It creates a new connection pool each time, requiring a new TCP and TLS handshake for every single request, which adds significant latency and prevents connection reuse.
We should instantiate the client once in __init__ and reuse it across requests.
class AnthropicProvider(BaseProvider):
"""Async Anthropic Messages API provider."""
def __init__(self, info: ProviderInfo, api_key: str | None, base_url: str | None):
super().__init__(info, api_key, base_url)
from anthropic import AsyncAnthropic
self.client = AsyncAnthropic(api_key=api_key or "")
async def acomplete(
self,
messages: list[Message],
system: str | None,
spec: ModelSpec,
*,
max_output_tokens: int | None = None,
temperature: float | None = None,
) -> str:
kwargs: dict[str, Any] = {
"model": spec.api_id,
"messages": list(messages),
"max_tokens": max_output_tokens or _DEFAULT_MAX_TOKENS,
}
if system:
kwargs["system"] = system
if temperature is not None:
kwargs["temperature"] = temperature
response = await self.client.messages.create(**kwargs)
Comment on lines
+16
to
+64
|
|
||
| class GeminiProvider(BaseProvider): | ||
| """Async Google Gemini provider (google-genai).""" | ||
|
|
||
| async def acomplete( | ||
| self, | ||
| messages: list[Message], | ||
| system: str | None, | ||
| spec: ModelSpec, | ||
| *, | ||
| max_output_tokens: int | None = None, | ||
| temperature: float | None = None, | ||
| ) -> str: | ||
| try: | ||
| from google import genai | ||
| from google.genai import types | ||
| except ImportError as exc: # pragma: no cover - install-time guard | ||
| raise RuntimeError( | ||
| "The 'google-genai' package is required. Install with: uv add google-genai" | ||
| ) from exc | ||
|
|
||
| contents: list[Any] = [] | ||
| for message in messages: | ||
| role = "user" if message["role"] == "user" else "model" | ||
| contents.append( | ||
| types.Content(role=role, parts=[types.Part.from_text(text=message["content"])]) | ||
| ) | ||
|
|
||
| # Relax safety filters — security tooling output routinely trips defaults. | ||
| safety_categories = ( | ||
| types.HarmCategory.HARM_CATEGORY_HARASSMENT, | ||
| types.HarmCategory.HARM_CATEGORY_HATE_SPEECH, | ||
| types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT, | ||
| types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, | ||
| ) | ||
| safety = [ | ||
| types.SafetySetting(category=category, threshold=types.HarmBlockThreshold.BLOCK_NONE) | ||
| for category in safety_categories | ||
| ] | ||
| config = types.GenerateContentConfig( | ||
| system_instruction=system or None, | ||
| temperature=temperature, | ||
| max_output_tokens=max_output_tokens, | ||
| safety_settings=safety, | ||
| ) | ||
|
|
||
| client = genai.Client(api_key=self.api_key or "") | ||
| response = await client.aio.models.generate_content( | ||
| model=spec.api_id, contents=contents, config=config |
There was a problem hiding this comment.
Instantiating a new genai.Client on every call to acomplete is highly inefficient. It creates a new connection pool each time, requiring a new TCP and TLS handshake for every single request, which adds significant latency and prevents connection reuse.
We should instantiate the client once in __init__ and reuse it across requests.
class GeminiProvider(BaseProvider):
"""Async Google Gemini provider (google-genai)."""
def __init__(self, info: ProviderInfo, api_key: str | None, base_url: str | None):
super().__init__(info, api_key, base_url)
from google import genai
self.client = genai.Client(api_key=api_key or "")
async def acomplete(
self,
messages: list[Message],
system: str | None,
spec: ModelSpec,
*,
max_output_tokens: int | None = None,
temperature: float | None = None,
) -> str:
try:
from google.genai import types
except ImportError as exc: # pragma: no cover - install-time guard
raise RuntimeError(
"The 'google-genai' package is required. Install with: uv add google-genai"
) from exc
contents: list[Any] = []
for message in messages:
role = "user" if message["role"] == "user" else "model"
contents.append(
types.Content(role=role, parts=[types.Part.from_text(text=message["content"])])
)
# Relax safety filters — security tooling output routinely trips defaults.
safety_categories = (
types.HarmCategory.HARM_CATEGORY_HARASSMENT,
types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
)
safety = [
types.SafetySetting(category=category, threshold=types.HarmBlockThreshold.BLOCK_NONE)
for category in safety_categories
]
config = types.GenerateContentConfig(
system_instruction=system or None,
temperature=temperature,
max_output_tokens=max_output_tokens,
safety_settings=safety,
)
response = await self.client.aio.models.generate_content(
model=spec.api_id, contents=contents, config=config
)
|
Thank you
…On Sat 6. Jun 2026 at 8:33 PM, gemini-code-assist[bot] < ***@***.***> wrote:
***@***.***[bot]* commented on this pull request.
Code Review
This pull request restructures the repository to separate the autonomous
agent (pentestgpt/) from the modernized legacy interactive mode (
pentestgpt_legacy/), introducing a multi-stage pipeline orchestrator and
updating the Docker environment to use uv. The legacy mode is rebuilt
with a native per-provider LLM layer supporting modern models. The code
review identified several critical issues, including an unawaited coroutine
in the backend process cleanup, potential type errors when handling flag
dictionaries in the pipeline and stage prompts, and unhandled exceptions in
the synchronous runner thread. Additionally, the reviewer recommended
optimizing LLM client instantiation across the Anthropic, Gemini, and
OpenAI-compatible providers to prevent connection overhead.
Important
The consumer version of Gemini Code Assist on GitHub
<https://developers.google.com/gemini-code-assist/docs/review-repo-code>
is being sunset. Starting *June 18, 2026*, new organization installations
will be blocked, and all code review activity will officially cease on *July
17, 2026*.
For more details on the timeline and next steps, please review the Help
Documentation
<https://developers.google.com/gemini-code-assist/docs/deprecations/consumer-code-review>
.
------------------------------
In pentestgpt/core/backend.py
<#469 (comment)>:
> + # Non-async wait — safe even during event loop teardown
+ with contextlib.suppress(Exception):
+ proc.wait() # type: ignore[unused-coroutine]
[image: high]
<https://camo.githubusercontent.com/7559374fd248a2a146dfe7112beda558979c4f6d02dbe7c5161fd893fe834423/68747470733a2f2f7777772e677374617469632e636f6d2f636f64657265766965776167656e742f686967682d7072696f726974792e737667>
In _kill_process, proc.wait() is called without await. Since proc.wait()
is an asynchronous coroutine in asyncio.subprocess.Process, calling it
without await does not wait for or reap the process, and it will raise a RuntimeWarning:
coroutine 'Process.wait' was never awaited.
To perform a non-blocking synchronous wait safely during event loop
teardown, we should wait on the underlying subprocess.Popen object
directly.
⬇️ Suggested change
- # Non-async wait — safe even during event loop teardown
- with contextlib.suppress(Exception):
- proc.wait() # type: ignore[unused-coroutine]
+ # Non-async wait — safe even during event loop teardown
+ with contextlib.suppress(Exception):
+ if proc._transport:
+ popen = proc._transport.get_extra_info('subprocess')
+ if popen:
+ popen.wait()
------------------------------
In pentestgpt/core/pipeline.py
<#469 (comment)>:
> + for f in r.flags_found:
+ if f not in seen:
+ seen.add(f)
+ flags.append(f)
+ return flags
[image: high]
<https://camo.githubusercontent.com/7559374fd248a2a146dfe7112beda558979c4f6d02dbe7c5161fd893fe834423/68747470733a2f2f7777772e677374617469632e636f6d2f636f64657265766965776167656e742f686967682d7072696f726974792e737667>
If r.flags_found contains dictionary objects (e.g., {"flag": "...",
"context": "..."}), iterating over them and adding them directly to the
seen set will raise TypeError: unhashable type: 'dict'.
We should extract the flag string from f if it is a dictionary before
checking or adding it to the seen set.
⬇️ Suggested change
- for f in r.flags_found:
- if f not in seen:
- seen.add(f)
- flags.append(f)
- return flags
+ for r in self.stage_results:
+ for f in r.flags_found:
+ flag_str = f.get("flag", f) if isinstance(f, dict) else f
+ if flag_str not in seen:
+ seen.add(flag_str)
+ flags.append(flag_str)
------------------------------
In pentestgpt/prompts/stages.py
<#469 (comment)>:
> + parts.append(f"Flags found: {', '.join(result.flags_found)}")
+
[image: high]
<https://camo.githubusercontent.com/7559374fd248a2a146dfe7112beda558979c4f6d02dbe7c5161fd893fe834423/68747470733a2f2f7777772e677374617469632e636f6d2f636f64657265766965776167656e742f686967682d7072696f726974792e737667>
If result.flags_found contains dictionary objects, calling ',
'.join(result.flags_found) will raise TypeError: sequence item 0:
expected str instance, dict found.
We should extract the flag string from each element in result.flags_found
before joining them.
⬇️ Suggested change
- parts.append(f"Flags found: {', '.join(result.flags_found)}")
-
+ if result.flags_found:
+ flag_strs = [f.get("flag", f) if isinstance(f, dict) else f for f in result.flags_found]
+ parts.append(f"Flags found: {', '.join(flag_strs)}")
------------------------------
In pentestgpt_legacy/llm/client.py
<#469 (comment)>:
> + box: dict[str, Any] = {}
+
+ def _runner() -> None:
+ box["value"] = asyncio.run(coro)
+
+ thread = threading.Thread(target=_runner)
+ thread.start()
+ thread.join()
+ return box["value"]
[image: high]
<https://camo.githubusercontent.com/7559374fd248a2a146dfe7112beda558979c4f6d02dbe7c5161fd893fe834423/68747470733a2f2f7777772e677374617469632e636f6d2f636f64657265766965776167656e742f686967682d7072696f726974792e737667>
In run_sync, the _runner thread function executes asyncio.run(coro) but
does not capture or propagate any exceptions raised during execution. If an
exception occurs (e.g., network timeout, authentication failure), the
thread will terminate silently, and the main thread will raise a confusing KeyError:
'value' when trying to access box["value"].
We should catch exceptions in the thread, store them in box, and re-raise
them in the main thread to preserve the original traceback and error
message.
⬇️ Suggested change
- box: dict[str, Any] = {}
-
- def _runner() -> None:
- box["value"] = asyncio.run(coro)
-
- thread = threading.Thread(target=_runner)
- thread.start()
- thread.join()
- return box["value"]
+ box: dict[str, Any] = {}
+
+ def _runner() -> None:
+ try:
+ box["value"] = asyncio.run(coro)
+ except Exception as e:
+ box["error"] = e
+
+ thread = threading.Thread(target=_runner)
+ thread.start()
+ thread.join()
+ if "error" in box:
+ raise box["error"]
+ return box["value"]
------------------------------
In pentestgpt_legacy/llm/providers/openai_compatible.py
<#469 (comment)>:
> +from pentestgpt_legacy.llm.registry import ModelSpec
+
+
+class OpenAICompatibleProvider(BaseProvider):
+ """Async OpenAI-compatible provider (Chat Completions, optional Responses API)."""
+
+ async def acomplete(
+ self,
+ messages: list[Message],
+ system: str | None,
+ spec: ModelSpec,
+ *,
+ max_output_tokens: int | None = None,
+ temperature: float | None = None,
+ ) -> str:
+ try:
+ from openai import AsyncOpenAI
+ except ImportError as exc: # pragma: no cover - install-time guard
+ raise RuntimeError(
+ "The 'openai' package is required. Install with: uv add openai"
+ ) from exc
+
+ # Ollama needs a non-empty key string but ignores its value.
+ api_key = self.api_key or "not-needed"
+ async with AsyncOpenAI(api_key=api_key, base_url=self.base_url) as client:
+ if spec.responses_api:
+ return await self._via_responses(
+ client, messages, system, spec, max_output_tokens, temperature
+ )
[image: medium]
<https://camo.githubusercontent.com/32601710f6703a1d3cdbb05c7f9f05d1d8c88abc4d4e4d4e25ff218874a45279/68747470733a2f2f7777772e677374617469632e636f6d2f636f64657265766965776167656e742f6d656469756d2d7072696f726974792e737667>
Instantiating a new AsyncOpenAI client on every call to acomplete is
highly inefficient. It creates a new httpx.AsyncClient and connection
pool each time, requiring a new TCP and TLS handshake for every single
request, which adds significant latency and prevents connection reuse.
We should instantiate the client once in __init__ and reuse it across
requests.
class OpenAICompatibleProvider(BaseProvider):
"""Async OpenAI-compatible provider (Chat Completions, optional Responses API)."""
def __init__(self, info: ProviderInfo, api_key: str | None, base_url: str | None):
super().__init__(info, api_key, base_url)
from openai import AsyncOpenAI
self.client = AsyncOpenAI(api_key=api_key or "not-needed", base_url=base_url)
async def acomplete(
self,
messages: list[Message],
system: str | None,
spec: ModelSpec,
*,
max_output_tokens: int | None = None,
temperature: float | None = None,
) -> str:
if spec.responses_api:
return await self._via_responses(
self.client, messages, system, spec, max_output_tokens, temperature
)
return await self._via_chat(
self.client, messages, system, spec, max_output_tokens, temperature
)
------------------------------
In pentestgpt_legacy/llm/providers/anthropic_provider.py
<#469 (comment)>:
> +from pentestgpt_legacy.llm.registry import ModelSpec
+
+# Anthropic requires an explicit output cap; generous enough for detailed steps.
+_DEFAULT_MAX_TOKENS = 8192
+
+
+class AnthropicProvider(BaseProvider):
+ """Async Anthropic Messages API provider."""
+
+ async def acomplete(
+ self,
+ messages: list[Message],
+ system: str | None,
+ spec: ModelSpec,
+ *,
+ max_output_tokens: int | None = None,
+ temperature: float | None = None,
+ ) -> str:
+ try:
+ from anthropic import AsyncAnthropic
+ except ImportError as exc: # pragma: no cover - install-time guard
+ raise RuntimeError(
+ "The 'anthropic' package is required. Install with: uv add anthropic"
+ ) from exc
+
+ kwargs: dict[str, Any] = {
+ "model": spec.api_id,
+ "messages": list(messages),
+ "max_tokens": max_output_tokens or _DEFAULT_MAX_TOKENS,
+ }
+ if system:
[image: medium]
<https://camo.githubusercontent.com/32601710f6703a1d3cdbb05c7f9f05d1d8c88abc4d4e4d4e25ff218874a45279/68747470733a2f2f7777772e677374617469632e636f6d2f636f64657265766965776167656e742f6d656469756d2d7072696f726974792e737667>
Instantiating a new AsyncAnthropic client on every call to acomplete is
highly inefficient. It creates a new connection pool each time, requiring a
new TCP and TLS handshake for every single request, which adds significant
latency and prevents connection reuse.
We should instantiate the client once in __init__ and reuse it across
requests.
class AnthropicProvider(BaseProvider):
"""Async Anthropic Messages API provider."""
def __init__(self, info: ProviderInfo, api_key: str | None, base_url: str | None):
super().__init__(info, api_key, base_url)
from anthropic import AsyncAnthropic
self.client = AsyncAnthropic(api_key=api_key or "")
async def acomplete(
self,
messages: list[Message],
system: str | None,
spec: ModelSpec,
*,
max_output_tokens: int | None = None,
temperature: float | None = None,
) -> str:
kwargs: dict[str, Any] = {
"model": spec.api_id,
"messages": list(messages),
"max_tokens": max_output_tokens or _DEFAULT_MAX_TOKENS,
}
if system:
kwargs["system"] = system
if temperature is not None:
kwargs["temperature"] = temperature
response = await self.client.messages.create(**kwargs)
------------------------------
In pentestgpt_legacy/llm/providers/gemini_provider.py
<#469 (comment)>:
> +
+class GeminiProvider(BaseProvider):
+ """Async Google Gemini provider (google-genai)."""
+
+ async def acomplete(
+ self,
+ messages: list[Message],
+ system: str | None,
+ spec: ModelSpec,
+ *,
+ max_output_tokens: int | None = None,
+ temperature: float | None = None,
+ ) -> str:
+ try:
+ from google import genai
+ from google.genai import types
+ except ImportError as exc: # pragma: no cover - install-time guard
+ raise RuntimeError(
+ "The 'google-genai' package is required. Install with: uv add google-genai"
+ ) from exc
+
+ contents: list[Any] = []
+ for message in messages:
+ role = "user" if message["role"] == "user" else "model"
+ contents.append(
+ types.Content(role=role, parts=[types.Part.from_text(text=message["content"])])
+ )
+
+ # Relax safety filters — security tooling output routinely trips defaults.
+ safety_categories = (
+ types.HarmCategory.HARM_CATEGORY_HARASSMENT,
+ types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
+ types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
+ types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
+ )
+ safety = [
+ types.SafetySetting(category=category, threshold=types.HarmBlockThreshold.BLOCK_NONE)
+ for category in safety_categories
+ ]
+ config = types.GenerateContentConfig(
+ system_instruction=system or None,
+ temperature=temperature,
+ max_output_tokens=max_output_tokens,
+ safety_settings=safety,
+ )
+
+ client = genai.Client(api_key=self.api_key or "")
+ response = await client.aio.models.generate_content(
+ model=spec.api_id, contents=contents, config=config
[image: medium]
<https://camo.githubusercontent.com/32601710f6703a1d3cdbb05c7f9f05d1d8c88abc4d4e4d4e25ff218874a45279/68747470733a2f2f7777772e677374617469632e636f6d2f636f64657265766965776167656e742f6d656469756d2d7072696f726974792e737667>
Instantiating a new genai.Client on every call to acomplete is highly
inefficient. It creates a new connection pool each time, requiring a new
TCP and TLS handshake for every single request, which adds significant
latency and prevents connection reuse.
We should instantiate the client once in __init__ and reuse it across
requests.
class GeminiProvider(BaseProvider):
"""Async Google Gemini provider (google-genai)."""
def __init__(self, info: ProviderInfo, api_key: str | None, base_url: str | None):
super().__init__(info, api_key, base_url)
from google import genai
self.client = genai.Client(api_key=api_key or "")
async def acomplete(
self,
messages: list[Message],
system: str | None,
spec: ModelSpec,
*,
max_output_tokens: int | None = None,
temperature: float | None = None,
) -> str:
try:
from google.genai import types
except ImportError as exc: # pragma: no cover - install-time guard
raise RuntimeError(
"The 'google-genai' package is required. Install with: uv add google-genai"
) from exc
contents: list[Any] = []
for message in messages:
role = "user" if message["role"] == "user" else "model"
contents.append(
types.Content(role=role, parts=[types.Part.from_text(text=message["content"])])
)
# Relax safety filters — security tooling output routinely trips defaults.
safety_categories = (
types.HarmCategory.HARM_CATEGORY_HARASSMENT,
types.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
)
safety = [
types.SafetySetting(category=category, threshold=types.HarmBlockThreshold.BLOCK_NONE)
for category in safety_categories
]
config = types.GenerateContentConfig(
system_instruction=system or None,
temperature=temperature,
max_output_tokens=max_output_tokens,
safety_settings=safety,
)
response = await self.client.aio.models.generate_content(
model=spec.api_id, contents=contents, config=config
)
—
Reply to this email directly, view it on GitHub
<#469?email_source=notifications&email_token=BSWJ3DEFJI634YQV4JL6SMT46RIXRA5CNFSNUABKM5UWIORPF5TWS5BNNB2WEL2QOVWGYUTFOF2WK43UKJSXM2LFO4XTINBUGM2DOMJTGQ2KM4TFMFZW63VKON2WE43DOJUWEZLEUVSXMZLOOSWGM33PORSXEX3DNRUWG2Y#pullrequestreview-4443471344>,
or unsubscribe
<https://github.com/notifications/unsubscribe-auth/BSWJ3DB4APRBYNDW6PQI6HL46RIXRAVCNFSM6AAAAACZ5KCCAGVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHM2DINBTGQ3TCMZUGQ>
.
You are receiving this because you are subscribed to this thread.Message
ID: ***@***.***>
|
GreyDGL
added a commit
that referenced
this pull request
Jun 7, 2026
* fix: 🐛 minor typo and build process * feat: 🎸 [WIP] Pentest mode * feat: 🎸 code abstraction * feat: modernize legacy PentestGPT with native multi-LLM support (#469) Rebuild the classic USENIX-2024 interactive PentestGPT (reasoning / generation / parsing sessions + Pentesting Task Tree + REPL) as a standalone `pentestgpt_legacy` package on a native per-provider LLM layer that supports the latest 2026 models. - llm/: BaseProvider + OpenAI-compatible / Anthropic / Gemini connectors, a web-verified model registry (OpenAI, Anthropic, Gemini, DeepSeek, xAI, Qwen, Moonshot, local Ollama), a factory, and an LLMClient bridging async providers to the core's synchronous send_new_message/send_message session API. - CLI `pentestgpt-legacy`: --list-models and --smoke-test (live per-model round-trip matrix), plus --reasoning-model / --parsing-model / --base-url. - Tests: 25 unit tests (mocked, no network). Live smoke test verified 22/22 models with a configured key respond. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> * fix(backend): address review on ClaudeCodeBackend subprocess handling - _build_env: pop ANTHROPIC_API_KEY instead of setting it to "", so an empty value can't shadow the CLI's own auth fallback (e.g. subscription login). - _kill_process: reap the force-killed process with os.waitpid(.., WNOHANG) instead of calling the proc.wait() coroutine without awaiting it (removes the "coroutine was never awaited" warning). - query/_drain_stderr: drain subprocess stderr in a background task so its pipe buffer can't fill and deadlock the child. Also reformats backend.py, fixing the failing Lint (ruff format) check. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(docker-test): assert uv instead of Poetry in container health check The project migrated from Poetry to uv (the Dockerfile installs uv to /home/pentester/.local/bin, which is on PATH), so test_poetry_installed failed with exit 127. Replace it with test_uv_installed checking `uv --version`. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Rewrites and modernizes the legacy PentestGPT (the classic USENIX-2024 human-in-the-loop tool) so it works with the latest 2026 LLMs. It ships as a standalone
pentestgpt_legacypackage /pentestgpt-legacyCLI and does not touch the autonomous 3rd-gen agent.The classic design is preserved — three cooperating LLM sessions (reasoning / generation / parsing) maintaining a Pentesting Task Tree through an interactive REPL (
next,more,todo,discuss) — but the broken legacy LLM layer (mixed OpenAI SDK versions,tiktoken("gpt-4o")for every provider, stale hardcoded model ids, an unwiredapp.config) is replaced with a clean native per-provider layer.What's included
llm/registry.py— single source of truth for supported models (web-verified June 2026 ids); powers--list-modelsand the README table.llm/providers/—OpenAICompatibleProvider(OpenAI + DeepSeek/Ollama/xAI/Qwen/Moonshot viabase_url, with a Responses-API path for*-pro/*-codex),AnthropicProvider,GeminiProvider(newgoogle-genaiSDK).llm/client.py—LLMClientbridges async providers to the core's synchronoussend_new_message/send_message(drop-in for the oldLLMAPI), with context-window-aware history trimming.llm/factory.py/llm/config.py— model→provider construction and pydantic-settings credentials (per-provider keys + base-url overrides).pentestgpt-legacywith--list-models,--smoke-test,--reasoning-model,--parsing-model,--base-url.Supported providers/models (June 2026)
OpenAI (
gpt-5.5,gpt-5.5-pro,gpt-5.4-mini/nano,gpt-5.2,gpt-5.3-codex, + legacygpt-4o/o3/o4-mini) · Anthropic (claude-opus-4-8,claude-sonnet-4-6,claude-haiku-4-5) · Google Gemini (gemini-3.1-pro,gemini-3.5-flash,gemini-3-pro, + 2.5) · DeepSeek (deepseek-v4-flash/pro) · xAI (grok-4.3) · Qwen (qwen3.7-max,qwen3.5-flash) · Moonshot (kimi-k2.6) · local Ollama (ollama:<model>).Verification
pentestgpt-legacy --smoke-test) — every model with a configured key did a real round-trip; Gemini skipped (no key in this env). The first run surfaced 5 real issues, all fixed:gpt-5.5-pro/gpt-5.3-codexneeded the Responses API;claude-3-7-sonnet-latest/grok-4.1-fastwere retired/invalid ids (removed); Moonshot defaulted to the wrong region (now.cn, overridable).ruffclean;mypyclean on the typed core; existingmake typecheckunchanged; wheel builds with the new package.Notes for reviewers
main, which is a few commits ahead oforigin/main, so the diff also shows those prior commits.make typecheckstays scoped topentestgpt/; the new package is covered byruff(make lint) andtests/legacy/. Runuv run mypy pentestgpt_legacy/llm/for its typed core.🤖 Generated with Claude Code