Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions api_app/chatbot_manager/agent/system_prompt.txt
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@ Answer with short, data-driven summaries. Always cite the tools you used at the
- Call the right tool instead of guessing. Say so when a tool returns no data.
- Jobs vs investigations: any question about the user's own jobs ("show/list/find my jobs", "what jobs do I have?") uses search_jobs. list_investigations is only for investigations — a named, grouped collection of analyses — so never answer a plain "jobs" question with list_investigations.
- When a tool needs a value from a previous tool's result (an id, an md5), pass that actual value — never a placeholder like <job_id>.
- Copy analyzer, playbook and job names verbatim from tool results. Never replace them with placeholders or bracketed labels like [Analyzer 1] or [Playbook A] — write the real names exactly as the tool returned them.
- analyze_observable only previews; never tell the user an analysis has started — they approve it themselves via the Confirm button.
- When analyze_observable returns a plan with a non-null `reason`, state that reason in your confirmation message so the user knows why that playbook was chosen.

[Response style]
- One paragraph unless the user asks for a list or breakdown.
Expand Down
63 changes: 59 additions & 4 deletions api_app/chatbot_manager/agent/tools/analyze_observable.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,17 @@
AnalyzeObservableResultSerializer,
flatten_errors,
)
from api_app.choices import TLP
from api_app.choices import TLP, Classification
from api_app.playbooks_manager.models import PlaybookConfig
from api_app.serializers.job import ObservableAnalysisSerializer

# IntelOwl actively curates this playbook for plugins that need no API key (many migrations add
# analyzers to it), so it is the safe default when the user names neither a playbook nor analyzers.
FREE_TO_USE_PLAYBOOK = "FREE_TO_USE_ANALYZERS"
# Cap the playbook list in the fallback error so the LLM-facing message stays compact and actionable
# (visible_for_user includes every public playbook, so a classification can match dozens).
_MAX_PLAYBOOKS_IN_ERROR = 20


def make_analyze_observable_tool(user):
# Built per-request and closed over `user`. This is the only action-capable tool, but it now
Expand Down Expand Up @@ -63,6 +70,45 @@ def analyze_observable(
if analyzers_list:
data["analyzers_requested"] = analyzers_list

plan_reason = None
if not playbook and not analyzers_list:
# Neither a playbook nor analyzers were named -- the natural shape of "analyze X". Without a
# default this validates to zero plugins and the core raises "No Analyzers and Connectors can
# be run after filtering", which the model paraphrases into "no analyzers available" and the
# user reads as a broken deploy. Default to the curated FREE_TO_USE_ANALYZERS playbook when it
# is visible, applicable to the classification and enabled; otherwise return an actionable
# error naming the playbooks the user can pick from.
classification = Classification.calculate_observable(observable_name)
# The playbooks that WOULD qualify for this observable: visible, enabled, applicable. Both
# the default lookup and the fallback list derive from this single queryset so the error
# names exactly the playbooks that could have run.
applicable_playbooks = PlaybookConfig.objects.visible_for_user(user).filter(
disabled=False, type__contains=[classification]
)
default_playbook = applicable_playbooks.filter(name=FREE_TO_USE_PLAYBOOK).first()
if default_playbook is not None:
data["playbook_requested"] = default_playbook.name
plan_reason = (
f"No playbook or analyzers were specified, so IntelOwl's curated "
f"'{FREE_TO_USE_PLAYBOOK}' playbook (key-free plugins) was selected for this "
f"{classification} observable."
)
else:
names = list(applicable_playbooks.order_by("name").values_list("name", flat=True))
shown = ", ".join(names[:_MAX_PLAYBOOKS_IN_ERROR])
if len(names) > _MAX_PLAYBOOKS_IN_ERROR:
shown += f" (and {len(names) - _MAX_PLAYBOOKS_IN_ERROR} more)"
message = (
f"No playbook or analyzers were specified. Pick one of the playbooks available to you "
f"for {classification} observables: {shown}."
if names
else f"No playbook or analyzers were specified and no playbook is available to you for "
f"{classification} observables; specify analyzers explicitly."
)
return AnalyzeObservableResultSerializer(
{"errors": [message], "plan": None, "pending_id": None}
).to_json()

serializer = ObservableAnalysisSerializer(data=data, context={"request": shim})
if not serializer.is_valid(raise_exception=False):
return AnalyzeObservableResultSerializer(
Expand All @@ -78,12 +124,21 @@ def analyze_observable(
"analyzers": [analyzer.name for analyzer in validated["analyzers_to_execute"]],
"connectors": [connector.name for connector in validated["connectors_to_execute"]],
"skipped": list(validated.get("warnings", [])),
# Non-null only when the plan defaulted to FREE_TO_USE_ANALYZERS, so the model can tell
# the user WHY that playbook was chosen instead of silently picking one.
"reason": plan_reason,
}
# Store the RAW inputs (re-validated at confirm time); the model cannot launch -- only a
# user POST of this pending_id to the confirm endpoint can.
# Store the inputs re-validated at confirm time; the model cannot launch -- only a user POST of
# this pending_id can. Persist the RESOLVED playbook (may be the defaulted FREE_TO_USE_ANALYZERS)
# so the confirm endpoint re-validates and launches exactly the previewed plan.
pending_id = create_pending_analysis(
user.id,
{"observable_name": observable_name, "tlp": tlp, "playbook": playbook, "analyzers": analyzers},
{
"observable_name": observable_name,
"tlp": tlp,
"playbook": data.get("playbook_requested", ""),
"analyzers": analyzers,
},
)
return AnalyzeObservableResultSerializer(
{"errors": [], "plan": plan, "pending_id": pending_id}
Expand Down
3 changes: 3 additions & 0 deletions api_app/chatbot_manager/serializers/analyze_observable.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,9 @@ class AnalysisPlanSerializer(serializers.Serializer):
analyzers = serializers.ListField(child=serializers.CharField())
connectors = serializers.ListField(child=serializers.CharField())
skipped = serializers.ListField(child=serializers.CharField())
# Human-readable justification, set only when the plan defaulted to a playbook the user did not
# name (so the confirmation response can explain the choice); null for explicit requests.
reason = serializers.CharField(allow_null=True)


class AnalyzeObservableResultSerializer(ToolResultSerializer):
Expand Down
17 changes: 17 additions & 0 deletions tests/api_app/chatbot_manager/test_prompt.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,23 @@ def test_prompt_sections_are_present(self):
for section in ("[Role]", "[Tools", "[Rules]", "[Response style]"):
self.assertIn(section, _SYSTEM_PROMPT, f"Missing section: {section}")

def test_prompt_forbids_placeholder_names(self):
"""A2: the [Rules] section must tell the model to copy analyzer/playbook/job names verbatim
and never emit bracketed placeholders like [Analyzer 1] — the qwen2.5:3b failure @mlodic hit.
"""
lowered = _SYSTEM_PROMPT.lower()
self.assertIn("verbatim", lowered)
self.assertIn("placeholder", lowered)

def test_prompt_tells_model_to_surface_plan_reason(self):
"""F2: the plan carries a `reason` when analyze_observable defaults a playbook, but the model
only narrates it if the prompt says to. The [Rules] section must instruct it to report the
reason in the confirmation message.
"""
lowered = _SYSTEM_PROMPT.lower()
self.assertIn("reason", lowered)
self.assertIn("why that playbook was chosen", lowered)

def test_page_context_not_in_the_file(self):
"""The file must NOT contain {page_context} — interpolation is the prompt
template's job, not the static file's.
Expand Down
46 changes: 46 additions & 0 deletions tests/api_app/chatbot_manager/tools/test_analyze_observable.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,3 +125,49 @@ def test_visible_playbook_preview(self, mock_apply):
self.assertEqual(data["plan"]["playbook"], self.pb_owned.name)
self.assertTrue(data["pending_id"])
mock_apply.assert_not_called()

@patch(_APPLY_ASYNC)
def test_explicit_request_has_no_default_reason(self, mock_apply):
# `reason` is populated ONLY when the plan defaults a playbook the user did not name.
data = json.loads(
self.analyze_observable.invoke({"observable_name": "example.com", "analyzers": "Tranco"})
)
self.assertEqual(data["errors"], [])
self.assertIsNone(data["plan"]["reason"])
mock_apply.assert_not_called()

@patch(_APPLY_ASYNC)
def test_no_plugins_defaults_to_free_to_use_playbook(self, mock_apply):
# A1: neither playbook nor analyzers -> default to the curated FREE_TO_USE_ANALYZERS playbook
# instead of failing with "No Analyzers and Connectors can be run after filtering".
data = json.loads(self.analyze_observable.invoke({"observable_name": "example.com"}))
self.assertEqual(data["errors"], [])
self.assertEqual(data["plan"]["playbook"], "FREE_TO_USE_ANALYZERS")
self.assertTrue(data["plan"]["analyzers"]) # the default playbook contributes real plugins
self.assertIsNotNone(data["plan"]["reason"]) # explains WHY that playbook was chosen
self.assertIn("FREE_TO_USE_ANALYZERS", data["plan"]["reason"])
self.assertTrue(data["pending_id"])
mock_apply.assert_not_called()

@patch(_APPLY_ASYNC)
def test_no_plugins_pending_stores_resolved_playbook(self, mock_apply):
# The pending must persist the RESOLVED default, or the confirm endpoint would re-validate an
# empty request and dead-end on "No Analyzers ...".
from api_app.chatbot_manager.pending_action import consume_pending_analysis

data = json.loads(self.analyze_observable.invoke({"observable_name": "example.com"}))
payload = consume_pending_analysis(self.user.id, data["pending_id"])
self.assertEqual(payload["playbook"], "FREE_TO_USE_ANALYZERS")
mock_apply.assert_not_called()

@patch(_APPLY_ASYNC)
def test_no_plugins_without_default_returns_actionable_error(self, mock_apply):
# When FREE_TO_USE_ANALYZERS is not resolvable, the error must name the playbooks the user can
# pick from -- not the current dead end.
PlaybookConfig.objects.filter(name="FREE_TO_USE_ANALYZERS").update(disabled=True)
data = json.loads(self.analyze_observable.invoke({"observable_name": "example.com"}))
self.assertIsNone(data["plan"])
self.assertIsNone(data["pending_id"])
self.assertTrue(any("Pick one of the playbooks" in e for e in data["errors"]))
self.assertTrue(any(self.pb_owned.name in e for e in data["errors"]))
mock_apply.assert_not_called()
Loading