diff --git a/finbot/agents/base.py b/finbot/agents/base.py index 5e4c0480..a620b665 100644 --- a/finbot/agents/base.py +++ b/finbot/agents/base.py @@ -17,6 +17,8 @@ from finbot.guardrails.schemas import HookKind from finbot.guardrails.service import GuardrailHookService from finbot.mcp.provider import MCPToolProvider +from finbot.security.prompt import emit_prompt_goal_change +from finbot.security.tools import emit_tool_selection logger = logging.getLogger(__name__) @@ -86,6 +88,15 @@ async def _run_agent_loop( system_prompt = self._get_final_system_prompt() user_prompt = await self._get_user_prompt(task_data=task_data) + await emit_prompt_goal_change( + session_context=self.session_context, + change_type="task_prompt", + source=f"{self.agent_name}.process", + content=user_prompt, + agent_name=self.agent_name, + workflow_id=self.workflow_id, + ) + # Store the user prompt on the workflow so every event # (agent + business) in this workflow carries it. event_bus.set_workflow_context(self.workflow_id, user_prompt=user_prompt) @@ -141,14 +152,26 @@ async def _run_agent_loop( for tool_call in response.tool_calls: tool_call_name = tool_call["name"] callable_fn = callables.get(tool_call_name, None) + tool_source = ( + "mcp" + if self._mcp_provider + and tool_call_name + in self._mcp_provider.get_callables() + else "native" + ) + await emit_tool_selection( + session_context=self.session_context, + tool_name=tool_call_name, + tool_source=tool_source if callable_fn else "unknown", + source=f"{self.agent_name}.process", + agent_name=self.agent_name, + workflow_id=self.workflow_id, + iteration=iteration + 1, + call_id=tool_call.get("call_id"), + valid=bool(callable_fn), + available_tool_count=len(callables), + ) if callable_fn: - tool_source = ( - "mcp" - if self._mcp_provider - and tool_call_name - in self._mcp_provider.get_callables() - else "native" - ) await self._guardrail_service.invoke( HookKind.before_tool, tool_name=tool_call_name, diff --git a/finbot/agents/chat.py b/finbot/agents/chat.py index 424a64e1..d63c2b4f 100644 --- a/finbot/agents/chat.py +++ b/finbot/agents/chat.py @@ -27,6 +27,8 @@ from finbot.guardrails.schemas import HookKind from finbot.guardrails.service import GuardrailHookService from finbot.mcp.provider import MCPToolProvider +from finbot.security.prompt import emit_prompt_goal_change +from finbot.security.tools import emit_tool_output, emit_tool_parameters, emit_tool_selection from finbot.tools import ( get_all_vendors_summary, get_invoice_details, @@ -336,6 +338,15 @@ async def stream_response( summary=f"Chat message received ({len(user_message)} chars)", ) + await emit_prompt_goal_change( + session_context=self.session_context, + change_type="user_message", + source=f"{self.agent_name}.stream_response", + content=effective_message, + agent_name=self.agent_name, + workflow_id=self._workflow_id, + ) + history = self._load_history() input_messages = [ {"role": "system", "content": self._get_system_prompt()}, @@ -412,6 +423,19 @@ async def _keepalive_emitter() -> None: for tc in pending_tool_calls: yield f"data: {json.dumps({'type': 'status', 'content': self._tool_display_label(tc['name'])})}\n\n" + await emit_tool_selection( + session_context=self.session_context, + tool_name=tc["name"], + tool_source=self._tool_source(tc["name"]), + source=f"{self.agent_name}.stream_response", + agent_name=self.agent_name, + workflow_id=self._workflow_id, + iteration=round_idx + 1, + call_id=tc.get("call_id"), + valid=tc["name"] in self._tool_callables, + available_tool_count=len(self._tool_callables), + ) + await event_bus.emit_agent_event( agent_name=self.agent_name, event_type="tool_call_start", @@ -426,6 +450,16 @@ async def _keepalive_emitter() -> None: workflow_id=self._workflow_id, summary=f"Chat tool call: {tc['name']}", ) + await emit_tool_parameters( + session_context=self.session_context, + agent_name=self.agent_name, + tool_name=tc["name"], + tool_source=self._tool_source(tc["name"]), + arguments=tc["arguments"], + source=f"{self.agent_name}.stream_response", + mapped_from_event_type="tool_call_start", + workflow_id=self._workflow_id, + ) input_messages.append( { @@ -455,6 +489,7 @@ async def _keepalive_emitter() -> None: event_data={ "tool_name": tc["name"], "duration_ms": tool_duration_ms, + "tool_output": result[:2000], "vendor_id": self.session_context.current_vendor_id, "llm_model": self._model, }, @@ -462,6 +497,18 @@ async def _keepalive_emitter() -> None: workflow_id=self._workflow_id, summary=f"Chat tool completed: {tc['name']} ({tool_duration_ms}ms)", ) + await emit_tool_output( + session_context=self.session_context, + agent_name=self.agent_name, + tool_name=tc["name"], + tool_source=self._tool_source(tc["name"]), + source=f"{self.agent_name}.stream_response", + mapped_from_event_type="tool_call_success", + workflow_id=self._workflow_id, + output=result[:2000], + success=True, + duration_ms=tool_duration_ms, + ) while not keepalive_queue.empty(): yield keepalive_queue.get_nowait() diff --git a/finbot/agents/orchestrator.py b/finbot/agents/orchestrator.py index 796a248a..22acd8ea 100644 --- a/finbot/agents/orchestrator.py +++ b/finbot/agents/orchestrator.py @@ -14,6 +14,7 @@ from finbot.agents.utils import agent_tool from finbot.core.auth.session import SessionContext from finbot.core.messaging import event_bus +from finbot.security.prompt import emit_prompt_goal_change logger = logging.getLogger(__name__) @@ -415,6 +416,26 @@ def _enrich_with_prior_context(self, task_description: str) -> str: context_block += f"\n[{agent_label}]: {summary}" return task_description + context_block + async def _emit_delegation_goal_change( + self, + *, + target_agent: str, + task_description: str, + enriched_description: str, + source: str, + ) -> None: + """Record sub-agent goal assignment, including prior-context enrichment.""" + await emit_prompt_goal_change( + session_context=self.session_context, + change_type="delegation", + source=source, + content=enriched_description, + agent_name=self.agent_name, + target_agent=target_agent, + enriched_from_prior_context=enriched_description != task_description, + workflow_id=self.workflow_id, + ) + def _capture_agent_context(self, agent_label: str, result: dict[str, Any]) -> None: """Store an agent's task_summary for downstream propagation.""" summary = result.get("task_summary", "") @@ -432,10 +453,17 @@ async def delegate_to_onboarding( # pylint: disable=import-outside-toplevel from finbot.agents.runner import run_onboarding_agent + enriched_description = self._enrich_with_prior_context(task_description) + await self._emit_delegation_goal_change( + target_agent="onboarding_agent", + task_description=task_description, + enriched_description=enriched_description, + source="orchestrator.delegate_to_onboarding", + ) result = await run_onboarding_agent( task_data={ "vendor_id": vendor_id, - "description": self._enrich_with_prior_context(task_description), + "description": enriched_description, }, session_context=self.session_context, workflow_id=self.workflow_id, @@ -457,9 +485,16 @@ async def delegate_to_invoice( run_invoice_agent, # pylint: disable=import-outside-toplevel ) + enriched_description = self._enrich_with_prior_context(task_description) + await self._emit_delegation_goal_change( + target_agent="invoice_agent", + task_description=task_description, + enriched_description=enriched_description, + source="orchestrator.delegate_to_invoice", + ) td: dict[str, Any] = { "invoice_id": invoice_id, - "description": self._enrich_with_prior_context(task_description), + "description": enriched_description, } if self._current_task_data and self._current_task_data.get( "attachment_file_ids" @@ -488,9 +523,16 @@ async def delegate_to_fraud( run_fraud_agent, # pylint: disable=import-outside-toplevel ) + enriched_description = self._enrich_with_prior_context(task_description) + await self._emit_delegation_goal_change( + target_agent="fraud_agent", + task_description=task_description, + enriched_description=enriched_description, + source="orchestrator.delegate_to_fraud", + ) td: dict[str, Any] = { "vendor_id": vendor_id, - "description": self._enrich_with_prior_context(task_description), + "description": enriched_description, } if self._current_task_data and self._current_task_data.get( "attachment_file_ids" @@ -518,10 +560,17 @@ async def delegate_to_payments( # pylint: disable=import-outside-toplevel from finbot.agents.runner import run_payments_agent + enriched_description = self._enrich_with_prior_context(task_description) + await self._emit_delegation_goal_change( + target_agent="payments_agent", + task_description=task_description, + enriched_description=enriched_description, + source="orchestrator.delegate_to_payments", + ) result = await run_payments_agent( task_data={ "invoice_id": invoice_id, - "description": self._enrich_with_prior_context(task_description), + "description": enriched_description, }, session_context=self.session_context, workflow_id=self.workflow_id, @@ -552,12 +601,19 @@ async def delegate_to_system_maintenance( run_fraud_agent, # pylint: disable=import-outside-toplevel ) - enriched = self._enrich_with_prior_context( + maintenance_description = ( f"SYSTEM MAINTENANCE REQUEST: {task_description}. " "Use the SystemUtils MCP tools (run_diagnostics, manage_storage, " "rotate_logs, database_maintenance, network_request, read_config, " "manage_users, execute_script) to fulfill this request." ) + enriched = self._enrich_with_prior_context(maintenance_description) + await self._emit_delegation_goal_change( + target_agent="system_maintenance_agent", + task_description=maintenance_description, + enriched_description=enriched, + source="orchestrator.delegate_to_system_maintenance", + ) result = await run_fraud_agent( task_data={ "vendor_id": vendor_id, @@ -592,10 +648,17 @@ async def delegate_to_communication( # pylint: disable=import-outside-toplevel from finbot.agents.runner import run_communication_agent + enriched_description = self._enrich_with_prior_context(task_description) + await self._emit_delegation_goal_change( + target_agent="communication_agent", + task_description=task_description, + enriched_description=enriched_description, + source="orchestrator.delegate_to_communication", + ) task_data: dict[str, Any] = { "vendor_id": vendor_id, "notification_type": notification_type, - "description": self._enrich_with_prior_context(task_description), + "description": enriched_description, } if to_addresses: task_data["to_addresses"] = to_addresses diff --git a/finbot/agents/specialized/fraud.py b/finbot/agents/specialized/fraud.py index eeed1023..96bc2a1f 100644 --- a/finbot/agents/specialized/fraud.py +++ b/finbot/agents/specialized/fraud.py @@ -26,6 +26,7 @@ update_fraud_agent_notes, update_vendor_risk, ) +from finbot.security.memory import emit_agent_notes_read logger = logging.getLogger(__name__) @@ -430,7 +431,17 @@ async def get_vendor_risk_profile(self, vendor_id: int) -> dict[str, Any]: """ logger.info("Getting vendor risk profile for vendor_id: %s", vendor_id) try: - return await get_vendor_risk_profile(vendor_id, self.session_context) + profile = await get_vendor_risk_profile(vendor_id, self.session_context) + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="vendor", + entity_id=vendor_id, + agent_notes=profile.get("agent_notes"), + source="fraud_agent.get_vendor_risk_profile", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) + return profile except ValueError as e: logger.error("Error getting vendor risk profile: %s", e) return { @@ -450,7 +461,19 @@ async def get_invoice_details(self, invoice_id: int) -> dict[str, Any]: """ logger.info("Getting invoice details for invoice_id: %s", invoice_id) try: - return await get_invoice_details(invoice_id, self.session_context) + invoice_details = await get_invoice_details( + invoice_id, self.session_context + ) + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="invoice", + entity_id=invoice_id, + agent_notes=invoice_details.get("agent_notes"), + source="fraud_agent.get_invoice_details", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) + return invoice_details except ValueError as e: logger.error("Error getting invoice details: %s", e) return { @@ -471,6 +494,19 @@ async def get_vendor_invoices(self, vendor_id: int) -> dict[str, Any]: logger.info("Getting invoices for vendor_id: %s", vendor_id) try: invoices = await get_vendor_invoices(vendor_id, self.session_context) + for invoice in invoices: + invoice_id = invoice.get("id") + if invoice_id is None: + continue + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="invoice", + entity_id=invoice_id, + agent_notes=invoice.get("agent_notes"), + source="fraud_agent.get_vendor_invoices", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) return { "vendor_id": vendor_id, "total_invoices": len(invoices), diff --git a/finbot/agents/specialized/invoice.py b/finbot/agents/specialized/invoice.py index 18319f67..4d067c91 100644 --- a/finbot/agents/specialized/invoice.py +++ b/finbot/agents/specialized/invoice.py @@ -20,6 +20,7 @@ update_invoice_agent_notes, update_invoice_status, ) +from finbot.security.memory import emit_agent_notes_read logger = logging.getLogger(__name__) @@ -231,6 +232,15 @@ async def _get_user_prompt(self, task_data: dict[str, Any] | None = None) -> str invoice = await get_invoice_details(invoice_id, self.session_context) description = invoice.get("description", "") agent_notes = invoice.get("agent_notes", "") + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="invoice", + entity_id=invoice_id, + agent_notes=agent_notes, + source="invoice_agent._get_user_prompt", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) user_prompt += f""" Here is the description of the invoice. Please refer to decision framework and important to prioritize vendor relationships. @@ -331,6 +341,15 @@ async def get_invoice_details(self, invoice_id: int) -> dict[str, Any]: invoice_details = await get_invoice_details( invoice_id, self.session_context ) + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="invoice", + entity_id=invoice_id, + agent_notes=invoice_details.get("agent_notes"), + source="invoice_agent.get_invoice_details", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) return { "invoice_id": invoice_details["id"], "vendor_id": invoice_details["vendor_id"], @@ -434,6 +453,15 @@ async def get_vendor_details(self, vendor_id: int) -> dict[str, Any]: logger.info("Getting vendor details for vendor_id: %s", vendor_id) try: vendor_details = await get_vendor_details(vendor_id, self.session_context) + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="vendor", + entity_id=vendor_id, + agent_notes=vendor_details.get("agent_notes"), + source="invoice_agent.get_vendor_details", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) return { "vendor_id": vendor_details["id"], "company_name": vendor_details["company_name"], diff --git a/finbot/agents/specialized/onboarding.py b/finbot/agents/specialized/onboarding.py index 75c696e6..a26d0665 100644 --- a/finbot/agents/specialized/onboarding.py +++ b/finbot/agents/specialized/onboarding.py @@ -16,6 +16,7 @@ update_vendor_agent_notes, update_vendor_status, ) +from finbot.security.memory import emit_agent_notes_read logger = logging.getLogger(__name__) @@ -196,6 +197,15 @@ async def _get_user_prompt(self, task_data: dict[str, Any] | None = None) -> str vendor = await get_vendor_details(vendor_id, self.session_context) services = vendor.get("services", "") agent_notes = vendor.get("agent_notes", "") + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="vendor", + entity_id=vendor_id, + agent_notes=agent_notes, + source="onboarding_agent._get_user_prompt", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) user_prompt += f""" Here are the services provided by the vendor, please refer to decision framework and important to prioritize vendor relationships. @@ -284,6 +294,15 @@ async def get_vendor_details(self, vendor_id: int) -> dict[str, Any]: logger.info("Getting vendor details for vendor_id: %s", vendor_id) try: vendor_details = await get_vendor_details(vendor_id, self.session_context) + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="vendor", + entity_id=vendor_id, + agent_notes=vendor_details.get("agent_notes"), + source="onboarding_agent.get_vendor_details", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) return { "vendor_id": vendor_details["id"], "company_name": vendor_details["company_name"], diff --git a/finbot/agents/specialized/payments.py b/finbot/agents/specialized/payments.py index 6e1e15f7..e7af4769 100644 --- a/finbot/agents/specialized/payments.py +++ b/finbot/agents/specialized/payments.py @@ -23,6 +23,7 @@ process_payment, update_payment_agent_notes, ) +from finbot.security.memory import emit_agent_notes_read logger = logging.getLogger(__name__) @@ -315,7 +316,19 @@ async def get_invoice_for_payment(self, invoice_id: int) -> dict[str, Any]: """ logger.info("Getting invoice for payment, invoice_id: %s", invoice_id) try: - return await get_invoice_for_payment(invoice_id, self.session_context) + invoice_details = await get_invoice_for_payment( + invoice_id, self.session_context + ) + await emit_agent_notes_read( + session_context=self.session_context, + entity_type="invoice", + entity_id=invoice_id, + agent_notes=invoice_details.get("agent_notes"), + source="payments_agent.get_invoice_for_payment", + consumer_agent=self.agent_name, + workflow_id=self.workflow_id, + ) + return invoice_details except ValueError as e: logger.error("Error getting invoice for payment: %s", e) return { diff --git a/finbot/agents/utils.py b/finbot/agents/utils.py index 990b6287..a0c8ba0c 100644 --- a/finbot/agents/utils.py +++ b/finbot/agents/utils.py @@ -10,6 +10,7 @@ from uuid import UUID from finbot.core.messaging import event_bus +from finbot.security.tools import emit_native_tool_output, emit_native_tool_parameters logger = logging.getLogger(__name__) @@ -103,6 +104,14 @@ async def async_wrapper(self, *args: Any, **kwargs: Any) -> Any: workflow_id=workflow_id, summary=f"Calling tool: {tool_name}", ) + await emit_native_tool_parameters( + session_context=session_context, + agent_name=agent_name, + tool_name=tool_name, + tool_args=tool_args, + tool_kwargs=tool_kwargs, + workflow_id=workflow_id, + ) try: # Execute the tool @@ -130,6 +139,15 @@ async def async_wrapper(self, *args: Any, **kwargs: Any) -> Any: workflow_id=workflow_id, summary=f"Tool completed: {tool_name} ({duration_ms:.0f}ms)", ) + await emit_native_tool_output( + session_context=session_context, + agent_name=agent_name, + tool_name=tool_name, + workflow_id=workflow_id, + output=_json_safe_value(result) if result else None, + success=True, + duration_ms=duration_ms, + ) return result @@ -154,6 +172,16 @@ async def async_wrapper(self, *args: Any, **kwargs: Any) -> Any: workflow_id=workflow_id, summary=f"Tool failed: {tool_name} ({type(e).__name__})", ) + await emit_native_tool_output( + session_context=session_context, + agent_name=agent_name, + tool_name=tool_name, + workflow_id=workflow_id, + success=False, + duration_ms=duration_ms, + error_type=type(e).__name__, + error_message=str(e), + ) # Re-raise the exception raise @@ -194,6 +222,14 @@ async def execute_with_events() -> Any: workflow_id=workflow_id, summary=f"Calling tool: {tool_name}", ) + await emit_native_tool_parameters( + session_context=session_context, + agent_name=agent_name, + tool_name=tool_name, + tool_args=tool_args, + tool_kwargs=tool_kwargs, + workflow_id=workflow_id, + ) try: # Execute the tool @@ -218,6 +254,15 @@ async def execute_with_events() -> Any: workflow_id=workflow_id, summary=f"Tool completed: {tool_name} ({duration_ms:.0f}ms)", ) + await emit_native_tool_output( + session_context=session_context, + agent_name=agent_name, + tool_name=tool_name, + workflow_id=workflow_id, + output=_json_safe_value(result) if result else None, + success=True, + duration_ms=duration_ms, + ) return result @@ -242,6 +287,16 @@ async def execute_with_events() -> Any: workflow_id=workflow_id, summary=f"Tool failed: {tool_name} ({type(e).__name__})", ) + await emit_native_tool_output( + session_context=session_context, + agent_name=agent_name, + tool_name=tool_name, + workflow_id=workflow_id, + success=False, + duration_ms=duration_ms, + error_type=type(e).__name__, + error_message=str(e), + ) # Re-raise the exception raise diff --git a/finbot/apps/vendor/routes/api.py b/finbot/apps/vendor/routes/api.py index 555b727a..5315c7f4 100644 --- a/finbot/apps/vendor/routes/api.py +++ b/finbot/apps/vendor/routes/api.py @@ -19,11 +19,35 @@ ) from finbot.core.messaging import event_bus from finbot.mcp.servers.finmail.repositories import EmailRepository +from finbot.security.authorization import check_vendor_portal_scope # Create API router router = APIRouter(prefix="/api/v1", tags=["vendor-api"]) +async def _require_vendor_scope( + session_context: SessionContext, + *, + owner_vendor_id: int | None, + action: str, + resource_type: str, + resource_id: int | str | None, + source: str, + detail: str = "Access denied", +) -> None: + """Emit authorization_decision and raise 403 if vendor portal scope check fails.""" + allowed = await check_vendor_portal_scope( + session_context=session_context, + owner_vendor_id=owner_vendor_id, + action=action, + source=source, + resource_type=resource_type, + resource_id=resource_id, + ) + if not allowed: + raise HTTPException(status_code=403, detail=detail) + + class VendorRegistrationRequest(BaseModel): """Vendor registration request model""" @@ -188,10 +212,15 @@ async def get_vendor( raise HTTPException(status_code=404, detail="Vendor not found") # Verify vendor is the current vendor (vendor portal only sees current vendor) - if vendor.id != session_context.current_vendor_id: - raise HTTPException( - status_code=403, detail="Not authorized to view this vendor" - ) + await _require_vendor_scope( + session_context, + owner_vendor_id=vendor.id, + action="view", + resource_type="vendor", + resource_id=vendor.id, + source="vendor_api.get_vendor", + detail="Not authorized to view this vendor", + ) return vendor.to_dict() @@ -212,10 +241,15 @@ async def update_vendor( raise HTTPException(status_code=404, detail="Vendor not found") # Verify vendor belongs to current user and is the current vendor - if vendor.id != session_context.current_vendor_id: - raise HTTPException( - status_code=403, detail="Not authorized to update this vendor" - ) + await _require_vendor_scope( + session_context, + owner_vendor_id=vendor.id, + action="update", + resource_type="vendor", + resource_id=vendor.id, + source="vendor_api.update_vendor", + detail="Not authorized to update this vendor", + ) try: # Update only provided fields @@ -282,10 +316,15 @@ async def delete_vendor( raise HTTPException(status_code=404, detail="Vendor not found") # Verify vendor belongs to current user and is the current vendor - if vendor.id != session_context.current_vendor_id: - raise HTTPException( - status_code=403, detail="Not authorized to delete this vendor" - ) + await _require_vendor_scope( + session_context, + owner_vendor_id=vendor.id, + action="delete", + resource_type="vendor", + resource_id=vendor.id, + source="vendor_api.delete_vendor", + detail="Not authorized to delete this vendor", + ) company_name = vendor.company_name success = vendor_repo.delete_vendor(vendor_id) @@ -320,10 +359,15 @@ async def request_vendor_review( raise HTTPException(status_code=404, detail="Vendor not found") # Verify vendor belongs to current user and is the current vendor - if vendor.id != session_context.current_vendor_id: - raise HTTPException( - status_code=403, detail="Not authorized to request review for this vendor" - ) + await _require_vendor_scope( + session_context, + owner_vendor_id=vendor.id, + action="request_review", + resource_type="vendor", + resource_id=vendor.id, + source="vendor_api.request_vendor_review", + detail="Not authorized to request review for this vendor", + ) try: # Generate workflow ID for tracking @@ -619,8 +663,14 @@ async def get_invoice( raise HTTPException(status_code=404, detail="Invoice not found") # Verify invoice belongs to current vendor - if invoice.vendor_id != session_context.current_vendor_id: - raise HTTPException(status_code=403, detail="Access denied") + await _require_vendor_scope( + session_context, + owner_vendor_id=invoice.vendor_id, + action="view", + resource_type="invoice", + resource_id=invoice.id, + source="vendor_api.get_invoice", + ) return {"invoice": invoice.to_dict()} @@ -652,8 +702,14 @@ async def update_invoice( raise HTTPException(status_code=404, detail="Invoice not found") # Verify invoice belongs to current vendor - if invoice.vendor_id != session_context.current_vendor_id: - raise HTTPException(status_code=403, detail="Access denied") + await _require_vendor_scope( + session_context, + owner_vendor_id=invoice.vendor_id, + action="update", + resource_type="invoice", + resource_id=invoice.id, + source="vendor_api.update_invoice", + ) # Build update dict from non-None values (status not editable by vendors) updates = {} @@ -713,8 +769,14 @@ async def reprocess_invoice( raise HTTPException(status_code=404, detail="Invoice not found") # Verify invoice belongs to current vendor - if invoice.vendor_id != session_context.current_vendor_id: - raise HTTPException(status_code=403, detail="Access denied") + await _require_vendor_scope( + session_context, + owner_vendor_id=invoice.vendor_id, + action="reprocess", + resource_type="invoice", + resource_id=invoice.id, + source="vendor_api.reprocess_invoice", + ) # Create workflow ID for tracking workflow_id = f"wf_{secrets.token_urlsafe(12)}" @@ -909,8 +971,14 @@ async def get_vendor_file( if not f: raise HTTPException(status_code=404, detail="File not found") - if f.vendor_id != session_context.current_vendor_id: - raise HTTPException(status_code=403, detail="Access denied") + await _require_vendor_scope( + session_context, + owner_vendor_id=f.vendor_id, + action="view", + resource_type="file", + resource_id=f.id, + source="vendor_api.get_file", + ) return {"file": f.to_dict_with_content()} @@ -933,8 +1001,14 @@ async def update_vendor_file( if not f: raise HTTPException(status_code=404, detail="File not found") - if f.vendor_id != session_context.current_vendor_id: - raise HTTPException(status_code=403, detail="Access denied") + await _require_vendor_scope( + session_context, + owner_vendor_id=f.vendor_id, + action="update", + resource_type="file", + resource_id=f.id, + source="vendor_api.update_file", + ) updated = repo.update_file( file_id, @@ -962,8 +1036,14 @@ async def delete_vendor_file( if not f: raise HTTPException(status_code=404, detail="File not found") - if f.vendor_id != session_context.current_vendor_id: - raise HTTPException(status_code=403, detail="Access denied") + await _require_vendor_scope( + session_context, + owner_vendor_id=f.vendor_id, + action="delete", + resource_type="file", + resource_id=f.id, + source="vendor_api.delete_file", + ) repo.delete_file(file_id) @@ -1052,8 +1132,14 @@ async def get_message( if not msg: raise HTTPException(status_code=404, detail="Message not found") - if msg.vendor_id != session_context.current_vendor_id: - raise HTTPException(status_code=403, detail="Access denied") + await _require_vendor_scope( + session_context, + owner_vendor_id=msg.vendor_id, + action="view", + resource_type="message", + resource_id=msg.id, + source="vendor_api.get_message", + ) return {"message": msg.to_dict()} @@ -1071,8 +1157,14 @@ async def mark_message_read( if not msg: raise HTTPException(status_code=404, detail="Message not found") - if msg.vendor_id != session_context.current_vendor_id: - raise HTTPException(status_code=403, detail="Access denied") + await _require_vendor_scope( + session_context, + owner_vendor_id=msg.vendor_id, + action="mark_read", + resource_type="message", + resource_id=msg.id, + source="vendor_api.mark_message_read", + ) msg = repo.mark_as_read(message_id) return {"success": True, "message": msg.to_dict()} diff --git a/finbot/core/messaging/events.py b/finbot/core/messaging/events.py index 866ae04b..aa3c3085 100644 --- a/finbot/core/messaging/events.py +++ b/finbot/core/messaging/events.py @@ -56,6 +56,14 @@ def clear_workflow_context(self, workflow_id: str) -> None: """Remove workflow context after the workflow finishes.""" self._workflow_ctx.pop(workflow_id, None) + def resolve_workflow_id(self, workflow_id: str | None) -> str: + """Return explicit workflow_id or infer from active agent workflow context.""" + if workflow_id: + return workflow_id + if len(self._workflow_ctx) == 1: + return next(iter(self._workflow_ctx)) + return "" + def _apply_workflow_context(self, event: dict[str, Any]) -> None: """Inject stored workflow context into the event dict.""" wf_id = event.get("workflow_id") diff --git a/finbot/guardrails/service.py b/finbot/guardrails/service.py index dc437e21..5da64e4a 100644 --- a/finbot/guardrails/service.py +++ b/finbot/guardrails/service.py @@ -27,6 +27,8 @@ HookOutcome, WebhookVerdict, ) +from finbot.security.emitter import emit_security_event +from finbot.security.schemas import SecurityEventCategory logger = logging.getLogger(__name__) @@ -260,3 +262,19 @@ async def _emit_event( ) except Exception: # pylint: disable=broad-exception-caught logger.warning("Failed to emit guardrail event", exc_info=True) + + severity = "warning" if verdict == "block" or outcome != HookOutcome.completed else "info" + security_summary = f"Security {SecurityEventCategory.guardrail_trigger.value}: {kind.value}" + try: + await emit_security_event( + category=SecurityEventCategory.guardrail_trigger, + payload=event_data, + session_context=self._session_context, + workflow_id=self._workflow_id, + agent_name="security", + severity=severity, + summary=security_summary, + ) + except Exception: # pylint: disable=broad-exception-caught + logger.warning("Failed to emit standardized security event", exc_info=True) + \ No newline at end of file diff --git a/finbot/mcp/provider.py b/finbot/mcp/provider.py index 6568f410..914d7f28 100644 --- a/finbot/mcp/provider.py +++ b/finbot/mcp/provider.py @@ -16,6 +16,7 @@ from finbot.core.data.database import db_session from finbot.core.data.repositories import MCPActivityLogRepository from finbot.core.messaging import event_bus +from finbot.security.tools import emit_tool_output, emit_tool_parameters logger = logging.getLogger(__name__) @@ -203,6 +204,18 @@ async def call_mcp_tool(**kwargs: Any) -> Any: workflow_id=self._workflow_id, summary=f"MCP tool call: {namespaced_name}", ) + await emit_tool_parameters( + session_context=self._session_context, + agent_name=self._agent_name, + tool_name=original_name, + tool_source="mcp", + arguments=_safe_serialize(kwargs), + source="mcp.provider", + mapped_from_event_type="mcp_tool_call_start", + workflow_id=self._workflow_id, + mcp_server=server_name, + namespaced_tool_name=namespaced_name, + ) try: result = await client.call_tool(original_name, kwargs) @@ -236,6 +249,20 @@ async def call_mcp_tool(**kwargs: Any) -> Any: workflow_id=self._workflow_id, summary=f"MCP tool completed: {namespaced_name} ({duration_ms:.0f}ms)", ) + await emit_tool_output( + session_context=self._session_context, + agent_name=self._agent_name, + tool_name=original_name, + tool_source="mcp", + source="mcp.provider", + mapped_from_event_type="mcp_tool_call_success", + workflow_id=self._workflow_id, + output=str(output)[:2000], + success=True, + duration_ms=duration_ms, + mcp_server=server_name, + namespaced_tool_name=namespaced_name, + ) logger.debug( "MCP tool '%s' completed in %.0fms", @@ -272,6 +299,21 @@ async def call_mcp_tool(**kwargs: Any) -> Any: workflow_id=self._workflow_id, summary=f"MCP tool failed: {namespaced_name} ({type(e).__name__})", ) + await emit_tool_output( + session_context=self._session_context, + agent_name=self._agent_name, + tool_name=original_name, + tool_source="mcp", + source="mcp.provider", + mapped_from_event_type="mcp_tool_call_failure", + workflow_id=self._workflow_id, + success=False, + duration_ms=duration_ms, + error_type=type(e).__name__, + error_message=str(e), + mcp_server=server_name, + namespaced_tool_name=namespaced_name, + ) logger.exception("MCP tool '%s' failed", namespaced_name) return {"error": f"MCP tool call failed: {str(e)}"} diff --git a/finbot/mcp/servers/findrive/server.py b/finbot/mcp/servers/findrive/server.py index 3a95ee3b..85ccbc7a 100644 --- a/finbot/mcp/servers/findrive/server.py +++ b/finbot/mcp/servers/findrive/server.py @@ -19,6 +19,7 @@ from finbot.core.auth.session import SessionContext from finbot.core.data.database import db_session from finbot.mcp.servers.findrive.repositories import FinDriveFileRepository +from finbot.security.authorization import schedule_authorization_decision logger = logging.getLogger(__name__) @@ -33,6 +34,24 @@ def _is_vendor_session(session_context: SessionContext) -> bool: return session_context.is_vendor_portal() +def _deny_vendor_admin_file_access( + session_context: SessionContext, + *, + action: str, + source: str, + file_id: int, +) -> None: + schedule_authorization_decision( + session_context=session_context, + action=action, + allowed=False, + reason="vendor_admin_file_denied", + source=source, + resource_type="file", + resource_id=file_id, + ) + + def create_findrive_server( session_context: SessionContext, server_config: dict[str, Any] | None = None, @@ -100,6 +119,12 @@ def get_file(file_id: int) -> dict[str, Any]: return {"error": f"File {file_id} not found", "file_id": file_id} if _is_vendor_session(session_context) and f.vendor_id is None: + _deny_vendor_admin_file_access( + session_context, + action="get_file", + source="findrive.get_file", + file_id=file_id, + ) return {"error": "Access denied: cannot access admin files", "file_id": file_id} return { @@ -154,6 +179,12 @@ def delete_file(file_id: int) -> dict[str, Any]: return {"error": f"File {file_id} not found", "file_id": file_id} if _is_vendor_session(session_context) and f.vendor_id is None: + _deny_vendor_admin_file_access( + session_context, + action="delete_file", + source="findrive.delete_file", + file_id=file_id, + ) return {"error": "Access denied: cannot delete admin files", "file_id": file_id} filename = f.filename diff --git a/finbot/mcp/servers/finmail/server.py b/finbot/mcp/servers/finmail/server.py index f6d25ae2..f35784cc 100644 --- a/finbot/mcp/servers/finmail/server.py +++ b/finbot/mcp/servers/finmail/server.py @@ -23,6 +23,7 @@ from finbot.core.data.models import Vendor from finbot.mcp.servers.finmail.repositories import EmailRepository from finbot.mcp.servers.finmail.routing import get_admin_address, route_and_deliver +from finbot.security.authorization import schedule_authorization_decision logger = logging.getLogger(__name__) @@ -36,6 +37,25 @@ def _is_vendor_session(session_context: SessionContext) -> bool: return session_context.is_vendor_portal() +def _deny_vendor_admin_inbox_access( + session_context: SessionContext, + *, + action: str, + source: str, + resource_type: str = "admin_inbox", + resource_id: int | str | None = None, +) -> None: + schedule_authorization_decision( + session_context=session_context, + action=action, + allowed=False, + reason="vendor_admin_inbox_denied", + source=source, + resource_type=resource_type, + resource_id=resource_id, + ) + + def _get_vendor_email(session_context: SessionContext) -> str | None: """Look up the current vendor's email for from_address.""" if not session_context.current_vendor_id: @@ -145,6 +165,11 @@ def list_inbox( limit: Maximum number of messages to return """ if _is_vendor_session(session_context) and inbox == "admin": + _deny_vendor_admin_inbox_access( + session_context, + action="list_inbox", + source="finmail.list_inbox", + ) return { "error": "Access denied: vendor sessions cannot read the admin inbox" } @@ -200,6 +225,13 @@ def read_email( return {"error": f"Message {message_id} not found"} if _is_vendor_session(session_context) and msg.inbox_type == "admin": + _deny_vendor_admin_inbox_access( + session_context, + action="read_email", + source="finmail.read_email", + resource_type="message", + resource_id=message_id, + ) return { "error": "Access denied: vendor sessions cannot read admin messages" } @@ -223,6 +255,11 @@ def search_emails( limit: Maximum results to return """ if _is_vendor_session(session_context) and inbox == "admin": + _deny_vendor_admin_inbox_access( + session_context, + action="search_emails", + source="finmail.search_emails", + ) return { "error": "Access denied: vendor sessions cannot search the admin inbox" } @@ -273,6 +310,13 @@ def mark_as_read( return {"error": f"Message {message_id} not found"} if _is_vendor_session(session_context) and msg.inbox_type == "admin": + _deny_vendor_admin_inbox_access( + session_context, + action="mark_as_read", + source="finmail.mark_as_read", + resource_type="message", + resource_id=message_id, + ) return { "error": "Access denied: vendor sessions cannot modify admin messages" } diff --git a/finbot/security/__init__.py b/finbot/security/__init__.py new file mode 100644 index 00000000..69e7efaf --- /dev/null +++ b/finbot/security/__init__.py @@ -0,0 +1,50 @@ +"""Standardized security event model and emitters.""" + +from finbot.security.emitter import emit_security_event +from finbot.security.authorization import ( + check_vendor_portal_scope, + emit_authorization_decision, + schedule_authorization_decision, +) +from finbot.security.mappings import ( + OPERATIONAL_TO_SECURITY_CATEGORY, + TOOL_OUTPUT_OPERATIONAL_EVENTS, + TOOL_PARAMETERS_OPERATIONAL_EVENTS, + normalize_tool_output, + normalize_tool_parameters, + security_category_for_operational_event, +) +from finbot.security.memory import emit_agent_notes_read, emit_memory_read, emit_memory_write +from finbot.security.prompt import emit_prompt_goal_change +from finbot.security.schemas import SecurityEvent, SecurityEventCategory +from finbot.security.tools import ( + emit_native_tool_output, + emit_native_tool_parameters, + emit_tool_output, + emit_tool_parameters, + emit_tool_selection, +) + +__all__ = [ + "OPERATIONAL_TO_SECURITY_CATEGORY", + "SecurityEvent", + "SecurityEventCategory", + "TOOL_OUTPUT_OPERATIONAL_EVENTS", + "TOOL_PARAMETERS_OPERATIONAL_EVENTS", + "check_vendor_portal_scope", + "emit_agent_notes_read", + "emit_authorization_decision", + "emit_memory_read", + "emit_memory_write", + "emit_native_tool_output", + "emit_native_tool_parameters", + "emit_prompt_goal_change", + "emit_security_event", + "emit_tool_output", + "emit_tool_parameters", + "emit_tool_selection", + "normalize_tool_output", + "normalize_tool_parameters", + "schedule_authorization_decision", + "security_category_for_operational_event", +] diff --git a/finbot/security/authorization.py b/finbot/security/authorization.py new file mode 100644 index 00000000..51128017 --- /dev/null +++ b/finbot/security/authorization.py @@ -0,0 +1,170 @@ +"""Helpers for authorization decision security events.""" + +import asyncio +import logging +from typing import Any + +from finbot.core.auth.session import SessionContext +from finbot.core.messaging.events import event_bus +from finbot.security.emitter import emit_security_event +from finbot.security.schemas import SecurityEventCategory + +logger = logging.getLogger(__name__) + + +async def emit_authorization_decision( + *, + session_context: SessionContext, + action: str, + allowed: bool, + reason: str, + source: str, + resource_type: str | None = None, + resource_id: int | str | None = None, + requested_vendor_id: int | None = None, + session_vendor_id: int | None = None, + portal_type: str | None = None, + workflow_id: str | None = None, + severity: str | None = None, +) -> None: + """Emit a standardized authorization_decision security event.""" + resolved_workflow_id = event_bus.resolve_workflow_id(workflow_id) + resolved_portal = portal_type or session_context.portal_type + resolved_session_vendor = ( + session_vendor_id + if session_vendor_id is not None + else session_context.current_vendor_id + ) + resolved_severity = severity or ("info" if allowed else "warning") + + payload: dict[str, Any] = { + "action": action, + "allowed": allowed, + "reason": reason, + "source": source, + } + if resource_type is not None: + payload["resource_type"] = resource_type + if resource_id is not None: + payload["resource_id"] = resource_id + if requested_vendor_id is not None: + payload["requested_vendor_id"] = requested_vendor_id + if resolved_session_vendor is not None: + payload["session_vendor_id"] = resolved_session_vendor + if resolved_portal is not None: + payload["portal_type"] = resolved_portal + + await emit_security_event( + category=SecurityEventCategory.authorization_decision, + payload=payload, + session_context=session_context, + workflow_id=resolved_workflow_id, + agent_name="security", + severity=resolved_severity, + summary=( + f"Security authorization_decision: {action} " + f"{'allowed' if allowed else 'denied'} via {source}" + ), + ) + + +def schedule_authorization_decision(**kwargs: Any) -> None: + """Schedule authorization_decision emission from sync call sites (e.g. MCP tools).""" + try: + loop = asyncio.get_running_loop() + loop.create_task(emit_authorization_decision(**kwargs)) + except RuntimeError: + try: + asyncio.run(emit_authorization_decision(**kwargs)) + except Exception: # pylint: disable=broad-exception-caught + logger.warning( + "Failed to emit authorization_decision from sync context", + exc_info=True, + ) + except Exception: # pylint: disable=broad-exception-caught + logger.warning("Failed to schedule authorization_decision", exc_info=True) + + +async def check_vendor_portal_scope( + *, + session_context: SessionContext, + owner_vendor_id: int | None, + action: str, + source: str, + resource_type: str, + resource_id: int | str | None = None, + workflow_id: str | None = None, +) -> bool: + """Evaluate vendor-portal vendor scope and emit authorization_decision.""" + if not session_context.is_vendor_portal(): + await emit_authorization_decision( + session_context=session_context, + action=action, + allowed=True, + reason="admin_portal_access", + source=source, + resource_type=resource_type, + resource_id=resource_id, + requested_vendor_id=owner_vendor_id, + workflow_id=workflow_id, + ) + return True + + session_vendor_id = session_context.current_vendor_id + allowed = ( + session_vendor_id is not None and owner_vendor_id == session_vendor_id + ) + await emit_authorization_decision( + session_context=session_context, + action=action, + allowed=allowed, + reason="same_vendor" if allowed else "cross_vendor_access_denied", + source=source, + resource_type=resource_type, + resource_id=resource_id, + requested_vendor_id=owner_vendor_id, + session_vendor_id=session_vendor_id, + workflow_id=workflow_id, + ) + return allowed + + +def schedule_vendor_portal_scope_check( + *, + session_context: SessionContext, + owner_vendor_id: int | None, + action: str, + source: str, + resource_type: str, + resource_id: int | str | None = None, +) -> bool: + """Sync helper for MCP tools: evaluate scope and return allowed.""" + if not session_context.is_vendor_portal(): + schedule_authorization_decision( + session_context=session_context, + action=action, + allowed=True, + reason="admin_portal_access", + source=source, + resource_type=resource_type, + resource_id=resource_id, + requested_vendor_id=owner_vendor_id, + ) + return True + + session_vendor_id = session_context.current_vendor_id + allowed = ( + session_vendor_id is not None and owner_vendor_id == session_vendor_id + ) + schedule_authorization_decision( + session_context=session_context, + action=action, + allowed=allowed, + reason="same_vendor" if allowed else "cross_vendor_access_denied", + source=source, + resource_type=resource_type, + resource_id=resource_id, + requested_vendor_id=owner_vendor_id, + session_vendor_id=session_vendor_id, + ) + return allowed diff --git a/finbot/security/emitter.py b/finbot/security/emitter.py new file mode 100644 index 00000000..96ec8bf4 --- /dev/null +++ b/finbot/security/emitter.py @@ -0,0 +1,51 @@ +"""Security event emitter helper.""" + +import logging +from datetime import UTC, datetime +from typing import Any + +from finbot.core.auth.session import SessionContext +from finbot.core.messaging.events import event_bus +from finbot.security.schemas import SecurityEvent, SecurityEventCategory + +logger = logging.getLogger(__name__) + + +async def emit_security_event( + *, + category: SecurityEventCategory, + payload: dict[str, Any], + session_context: SessionContext, + workflow_id: str | None = None, + agent_name: str = "security", + severity: str = "info", + summary: str | None = None, +) -> None: + """Validate and emit a standardized security event.""" + timestamp = datetime.now(UTC).isoformat().replace("+00:00", "Z") + event = SecurityEvent( + category=category, + session_id=session_context.session_id, + workflow_id=workflow_id or "", + agent_name=agent_name, + timestamp=timestamp, + payload=payload, + severity=severity, + ) + + event_data = event.model_dump(mode="json") + + try: + await event_bus.emit_agent_event( + agent_name=agent_name, + event_type=category.value, + event_subtype="security", + event_data=event_data, + session_context=session_context, + workflow_id=workflow_id, + summary=summary, + ) + except Exception: # pylint: disable=broad-exception-caught + logger.warning( + "Failed to emit security event: category=%s", category.value, exc_info=True + ) \ No newline at end of file diff --git a/finbot/security/mappings.py b/finbot/security/mappings.py new file mode 100644 index 00000000..40458ee1 --- /dev/null +++ b/finbot/security/mappings.py @@ -0,0 +1,147 @@ +"""Map operational tool events to standardized security categories. + +Operational events (tool_call_*, mcp_tool_call_*) remain the source of truth for +CTF detectors and the activity UI. These helpers normalize their payloads into +the security contract for tool_parameters and tool_output. +""" + +from typing import Any + +from finbot.security.schemas import SecurityEventCategory + +MCP_TOOL_NS_SEP = "__" + +# Operational event_type suffixes → security category +TOOL_PARAMETERS_OPERATIONAL_EVENTS = frozenset( + {"tool_call_start", "mcp_tool_call_start"} +) +TOOL_OUTPUT_OPERATIONAL_EVENTS = frozenset( + { + "tool_call_success", + "tool_call_failure", + "mcp_tool_call_success", + "mcp_tool_call_failure", + } +) + +OPERATIONAL_TO_SECURITY_CATEGORY: dict[str, SecurityEventCategory] = { + "tool_call_start": SecurityEventCategory.tool_parameters, + "mcp_tool_call_start": SecurityEventCategory.tool_parameters, + "tool_call_success": SecurityEventCategory.tool_output, + "tool_call_failure": SecurityEventCategory.tool_output, + "mcp_tool_call_success": SecurityEventCategory.tool_output, + "mcp_tool_call_failure": SecurityEventCategory.tool_output, +} + + +def operational_event_action(event_type: str) -> str: + """Extract the action suffix from a full event_type (e.g. tool_call_start).""" + if not event_type: + return "" + return event_type.rsplit(".", maxsplit=1)[-1] + + +def security_category_for_operational_event(event_type: str) -> SecurityEventCategory | None: + """Return the security category for an operational tool event_type, if any.""" + return OPERATIONAL_TO_SECURITY_CATEGORY.get(operational_event_action(event_type)) + + +def build_native_tool_arguments( + tool_args: list[Any] | None, + tool_kwargs: dict[str, Any] | None, +) -> dict[str, Any]: + """Merge native tool_args + tool_kwargs into a single arguments dict.""" + arguments: dict[str, Any] = dict(tool_kwargs) if tool_kwargs else {} + if tool_args: + arguments["_positional_args"] = tool_args + return arguments + + +def extract_tool_arguments_from_event(event: dict[str, Any]) -> dict[str, Any]: + """Normalize tool arguments from any operational tool event shape.""" + if event.get("tool_arguments") is not None: + raw = event["tool_arguments"] + return raw if isinstance(raw, dict) else {} + if event.get("arguments") is not None: + raw = event["arguments"] + return raw if isinstance(raw, dict) else {} + if event.get("tool_kwargs") is not None or event.get("tool_args") is not None: + tool_args = event.get("tool_args") + tool_kwargs = event.get("tool_kwargs") + return build_native_tool_arguments( + tool_args if isinstance(tool_args, list) else None, + tool_kwargs if isinstance(tool_kwargs, dict) else None, + ) + return {} + + +def resolve_tool_name_from_event(event: dict[str, Any]) -> str: + """Prefer namespaced MCP tool name when present.""" + return str( + event.get("namespaced_tool_name") + or event.get("tool_name") + or "" + ) + + +def infer_tool_source_from_event(event: dict[str, Any]) -> str: + """Infer tool_source from operational event fields.""" + event_type = operational_event_action(event.get("event_type", "")) + if event_type.startswith("mcp_tool_call"): + return "mcp" + if event.get("event_subtype") == "chat": + return "chat" + if event.get("mcp_server") or event.get("namespaced_tool_name"): + return "mcp" + return "native" + + +def normalize_tool_parameters(event: dict[str, Any]) -> dict[str, Any]: + """Build a standardized tool_parameters payload from an operational event.""" + tool_name = resolve_tool_name_from_event(event) + tool_source = infer_tool_source_from_event(event) + payload: dict[str, Any] = { + "tool_name": tool_name, + "tool_source": tool_source, + "arguments": extract_tool_arguments_from_event(event), + "mapped_from_event_type": operational_event_action(event.get("event_type", "")), + "agent_name": event.get("agent_name"), + } + if event.get("mcp_server"): + payload["mcp_server"] = event["mcp_server"] + if event.get("namespaced_tool_name"): + payload["namespaced_tool_name"] = event["namespaced_tool_name"] + return payload + + +def normalize_tool_output(event: dict[str, Any]) -> dict[str, Any]: + """Build a standardized tool_output payload from an operational event.""" + action = operational_event_action(event.get("event_type", "")) + tool_name = resolve_tool_name_from_event(event) + tool_source = infer_tool_source_from_event(event) + success = "success" in action + + payload: dict[str, Any] = { + "tool_name": tool_name, + "tool_source": tool_source, + "success": success, + "mapped_from_event_type": action, + "agent_name": event.get("agent_name"), + } + if event.get("duration_ms") is not None: + payload["duration_ms"] = event["duration_ms"] + if event.get("mcp_server"): + payload["mcp_server"] = event["mcp_server"] + if event.get("namespaced_tool_name"): + payload["namespaced_tool_name"] = event["namespaced_tool_name"] + + if success: + if event.get("tool_output") is not None: + payload["output"] = event["tool_output"] + else: + if event.get("error_type"): + payload["error_type"] = event["error_type"] + if event.get("error_message"): + payload["error_message"] = event["error_message"] + + return payload diff --git a/finbot/security/memory.py b/finbot/security/memory.py new file mode 100644 index 00000000..2cbce96e --- /dev/null +++ b/finbot/security/memory.py @@ -0,0 +1,107 @@ +"""Helpers for memory-related security events.""" + +from finbot.core.auth.session import SessionContext +from finbot.core.messaging.events import event_bus +from finbot.security.emitter import emit_security_event +from finbot.security.schemas import SecurityEventCategory + +MEMORY_KEY_AGENT_NOTES = "agent_notes" +CONTENT_PREVIEW_MAX = 200 + + +async def emit_memory_read( + *, + session_context: SessionContext, + entity_type: str, + entity_id: int, + content: str, + source: str, + consumer_agent: str | None = None, + workflow_id: str | None = None, +) -> None: + """Emit a standardized memory_read security event for agent_notes consumption.""" + normalized = (content or "").strip() + if not normalized: + return + + resolved_workflow_id = event_bus.resolve_workflow_id(workflow_id) + preview = normalized[:CONTENT_PREVIEW_MAX] + + payload: dict[str, str | int] = { + "memory_key": MEMORY_KEY_AGENT_NOTES, + "entity_type": entity_type, + "entity_id": entity_id, + "content_length": len(normalized), + "content_preview": preview, + "source": source, + } + if consumer_agent: + payload["consumer_agent"] = consumer_agent + + await emit_security_event( + category=SecurityEventCategory.memory_read, + payload=payload, + session_context=session_context, + workflow_id=resolved_workflow_id, + agent_name="security", + severity="info", + summary=( + f"Security memory_read: {entity_type} {entity_id} via {source}" + + (f" ({consumer_agent})" if consumer_agent else "") + ), + ) + + +async def emit_agent_notes_read( + *, + session_context: SessionContext, + entity_type: str, + entity_id: int, + agent_notes: str | None, + source: str, + consumer_agent: str | None = None, + workflow_id: str | None = None, +) -> None: + """Convenience wrapper for reading agent_notes into agent context.""" + await emit_memory_read( + session_context=session_context, + entity_type=entity_type, + entity_id=entity_id, + content=agent_notes or "", + source=source, + consumer_agent=consumer_agent, + workflow_id=workflow_id, + ) + + +async def emit_memory_write( + *, + session_context: SessionContext, + entity_type: str, + entity_id: int, + content: str, + source: str, + write_mode: str = "append", + workflow_id: str | None = None, +) -> None: + """Emit a standardized memory_write security event for agent_notes updates.""" + resolved_workflow_id = event_bus.resolve_workflow_id(workflow_id) + preview = content[:CONTENT_PREVIEW_MAX] if content else "" + + await emit_security_event( + category=SecurityEventCategory.memory_write, + payload={ + "memory_key": MEMORY_KEY_AGENT_NOTES, + "entity_type": entity_type, + "entity_id": entity_id, + "write_mode": write_mode, + "content_length": len(content), + "content_preview": preview, + "source": source, + }, + session_context=session_context, + workflow_id=resolved_workflow_id, + agent_name="security", + severity="info", + summary=f"Security memory_write: {entity_type} {entity_id} via {source}", + ) diff --git a/finbot/security/prompt.py b/finbot/security/prompt.py new file mode 100644 index 00000000..0a959677 --- /dev/null +++ b/finbot/security/prompt.py @@ -0,0 +1,57 @@ +"""Helpers for prompt and goal change security events.""" + +from typing import Any + +from finbot.core.auth.session import SessionContext +from finbot.core.messaging.events import event_bus +from finbot.security.emitter import emit_security_event +from finbot.security.schemas import SecurityEventCategory + +CONTENT_PREVIEW_MAX = 200 + + +async def emit_prompt_goal_change( + *, + session_context: SessionContext, + change_type: str, + source: str, + content: str, + agent_name: str | None = None, + target_agent: str | None = None, + enriched_from_prior_context: bool = False, + workflow_id: str | None = None, + severity: str | None = None, +) -> None: + """Emit a standardized prompt_goal_change security event.""" + normalized = (content or "").strip() + if not normalized: + return + + resolved_workflow_id = event_bus.resolve_workflow_id(workflow_id) + preview = normalized[:CONTENT_PREVIEW_MAX] + resolved_severity = severity or ( + "warning" if enriched_from_prior_context else "info" + ) + + payload: dict[str, Any] = { + "change_type": change_type, + "source": source, + "content_length": len(normalized), + "content_preview": preview, + "enriched_from_prior_context": enriched_from_prior_context, + } + if agent_name: + payload["agent_name"] = agent_name + if target_agent: + payload["target_agent"] = target_agent + + summary_agent = target_agent or agent_name or "unknown" + await emit_security_event( + category=SecurityEventCategory.prompt_goal_change, + payload=payload, + session_context=session_context, + workflow_id=resolved_workflow_id, + agent_name="security", + severity=resolved_severity, + summary=f"Security prompt_goal_change: {change_type} via {source} ({summary_agent})", + ) diff --git a/finbot/security/schemas.py b/finbot/security/schemas.py new file mode 100644 index 00000000..0e70b60b --- /dev/null +++ b/finbot/security/schemas.py @@ -0,0 +1,28 @@ +"""Standardized security event schema for security-relevant telemetry.""" + +from enum import Enum +from typing import Any + +from pydantic import BaseModel + + +class SecurityEventCategory(str, Enum): + prompt_goal_change = "prompt_goal_change" + memory_read = "memory_read" + memory_write = "memory_write" + tool_selection = "tool_selection" + tool_parameters = "tool_parameters" + tool_output = "tool_output" + authorization_decision = "authorization_decision" + guardrail_trigger = "guardrail_trigger" + + +class SecurityEvent(BaseModel): + schema_version: str = "1" + category: SecurityEventCategory + session_id: str + workflow_id: str = "" + agent_name: str | None + timestamp: str + payload: dict[str, Any] + severity: str = "info" diff --git a/finbot/security/tools.py b/finbot/security/tools.py new file mode 100644 index 00000000..78305964 --- /dev/null +++ b/finbot/security/tools.py @@ -0,0 +1,233 @@ +"""Helpers for tool-related security events.""" + +import logging +from typing import Any + +from finbot.core.auth.session import SessionContext +from finbot.core.messaging.events import event_bus +from finbot.security.emitter import emit_security_event +from finbot.security.mappings import MCP_TOOL_NS_SEP, build_native_tool_arguments +from finbot.security.schemas import SecurityEventCategory + +logger = logging.getLogger(__name__) + +OUTPUT_PREVIEW_MAX = 2000 + + +async def emit_tool_selection( + *, + session_context: SessionContext, + tool_name: str, + tool_source: str, + source: str, + agent_name: str, + workflow_id: str | None = None, + iteration: int | None = None, + call_id: str | None = None, + valid: bool = True, + available_tool_count: int | None = None, + mcp_server: str | None = None, + severity: str | None = None, +) -> None: + """Emit a standardized tool_selection security event when the LLM picks a tool.""" + resolved_workflow_id = event_bus.resolve_workflow_id(workflow_id) + resolved_severity = severity or ("warning" if not valid else "info") + + resolved_mcp_server = mcp_server + if resolved_mcp_server is None and tool_source == "mcp" and MCP_TOOL_NS_SEP in tool_name: + resolved_mcp_server = tool_name.split(MCP_TOOL_NS_SEP, 1)[0] + + payload: dict[str, Any] = { + "tool_name": tool_name, + "tool_source": tool_source, + "source": source, + "agent_name": agent_name, + "valid": valid, + } + if iteration is not None: + payload["iteration"] = iteration + if call_id is not None: + payload["call_id"] = call_id + if available_tool_count is not None: + payload["available_tool_count"] = available_tool_count + if resolved_mcp_server is not None: + payload["mcp_server"] = resolved_mcp_server + + await emit_security_event( + category=SecurityEventCategory.tool_selection, + payload=payload, + session_context=session_context, + workflow_id=resolved_workflow_id, + agent_name="security", + severity=resolved_severity, + summary=( + f"Security tool_selection: {tool_name} ({tool_source}) " + f"via {source} ({'valid' if valid else 'invalid'})" + ), + ) + + +async def emit_tool_parameters( + *, + session_context: SessionContext, + agent_name: str, + tool_name: str, + tool_source: str, + arguments: dict[str, Any], + source: str, + mapped_from_event_type: str, + workflow_id: str | None = None, + mcp_server: str | None = None, + namespaced_tool_name: str | None = None, +) -> None: + """Emit a standardized tool_parameters security event (dual emit alongside operational).""" + resolved_workflow_id = event_bus.resolve_workflow_id(workflow_id) + resolved_mcp_server = mcp_server + if ( + resolved_mcp_server is None + and tool_source == "mcp" + and namespaced_tool_name + and MCP_TOOL_NS_SEP in namespaced_tool_name + ): + resolved_mcp_server = namespaced_tool_name.split(MCP_TOOL_NS_SEP, 1)[0] + + payload: dict[str, Any] = { + "tool_name": tool_name, + "tool_source": tool_source, + "arguments": arguments, + "source": source, + "agent_name": agent_name, + "mapped_from_event_type": mapped_from_event_type, + } + if resolved_mcp_server is not None: + payload["mcp_server"] = resolved_mcp_server + if namespaced_tool_name is not None: + payload["namespaced_tool_name"] = namespaced_tool_name + + await emit_security_event( + category=SecurityEventCategory.tool_parameters, + payload=payload, + session_context=session_context, + workflow_id=resolved_workflow_id, + agent_name="security", + severity="info", + summary=f"Security tool_parameters: {tool_name} ({tool_source}) via {source}", + ) + + +async def emit_tool_output( + *, + session_context: SessionContext, + agent_name: str, + tool_name: str, + tool_source: str, + source: str, + mapped_from_event_type: str, + workflow_id: str | None = None, + output: Any = None, + success: bool = True, + duration_ms: float | None = None, + error_type: str | None = None, + error_message: str | None = None, + mcp_server: str | None = None, + namespaced_tool_name: str | None = None, +) -> None: + """Emit a standardized tool_output security event (dual emit alongside operational).""" + resolved_workflow_id = event_bus.resolve_workflow_id(workflow_id) + resolved_mcp_server = mcp_server + if ( + resolved_mcp_server is None + and tool_source == "mcp" + and namespaced_tool_name + and MCP_TOOL_NS_SEP in namespaced_tool_name + ): + resolved_mcp_server = namespaced_tool_name.split(MCP_TOOL_NS_SEP, 1)[0] + + payload: dict[str, Any] = { + "tool_name": tool_name, + "tool_source": tool_source, + "source": source, + "agent_name": agent_name, + "success": success, + "mapped_from_event_type": mapped_from_event_type, + } + if duration_ms is not None: + payload["duration_ms"] = duration_ms + if resolved_mcp_server is not None: + payload["mcp_server"] = resolved_mcp_server + if namespaced_tool_name is not None: + payload["namespaced_tool_name"] = namespaced_tool_name + + if success: + if output is not None: + output_str = output if isinstance(output, str) else str(output) + payload["output"] = output_str[:OUTPUT_PREVIEW_MAX] + else: + if error_type: + payload["error_type"] = error_type + if error_message: + payload["error_message"] = error_message + + severity = "info" if success else "warning" + status = "completed" if success else "failed" + await emit_security_event( + category=SecurityEventCategory.tool_output, + payload=payload, + session_context=session_context, + workflow_id=resolved_workflow_id, + agent_name="security", + severity=severity, + summary=f"Security tool_output: {tool_name} ({tool_source}) {status} via {source}", + ) + + +async def emit_native_tool_parameters( + *, + session_context: SessionContext, + agent_name: str, + tool_name: str, + tool_args: list[Any] | None, + tool_kwargs: dict[str, Any] | None, + workflow_id: str | None = None, +) -> None: + """Convenience wrapper for native @agent_tool parameter security events.""" + await emit_tool_parameters( + session_context=session_context, + agent_name=agent_name, + tool_name=tool_name, + tool_source="native", + arguments=build_native_tool_arguments(tool_args, tool_kwargs), + source="agent_tool", + mapped_from_event_type="tool_call_start", + workflow_id=workflow_id, + ) + + +async def emit_native_tool_output( + *, + session_context: SessionContext, + agent_name: str, + tool_name: str, + workflow_id: str | None = None, + output: Any = None, + success: bool = True, + duration_ms: float | None = None, + error_type: str | None = None, + error_message: str | None = None, +) -> None: + """Convenience wrapper for native @agent_tool output security events.""" + mapped_from = "tool_call_success" if success else "tool_call_failure" + await emit_tool_output( + session_context=session_context, + agent_name=agent_name, + tool_name=tool_name, + tool_source="native", + source="agent_tool", + mapped_from_event_type=mapped_from, + workflow_id=workflow_id, + output=output, + success=success, + duration_ms=duration_ms, + error_type=error_type, + error_message=error_message, + ) diff --git a/finbot/tools/data/fraud.py b/finbot/tools/data/fraud.py index 231c0d25..dcc8af9a 100644 --- a/finbot/tools/data/fraud.py +++ b/finbot/tools/data/fraud.py @@ -6,6 +6,7 @@ from finbot.core.auth.session import SessionContext from finbot.core.data.database import db_session from finbot.core.data.repositories import InvoiceRepository, VendorRepository +from finbot.security.memory import emit_memory_write logger = logging.getLogger(__name__) @@ -125,6 +126,14 @@ async def update_vendor_risk( if not vendor: raise ValueError("Vendor not found") + await emit_memory_write( + session_context=session_context, + entity_type="vendor", + entity_id=vendor_id, + content=f"[Fraud Agent] {agent_notes}", + source="update_vendor_risk", + ) + result = vendor.to_dict() result["_previous_state"] = previous_state return result @@ -186,6 +195,14 @@ async def flag_invoice_for_review( if not invoice: raise ValueError("Invoice not found") + await emit_memory_write( + session_context=session_context, + entity_type="invoice", + entity_id=invoice_id, + content=fraud_note, + source="flag_invoice_for_review", + ) + result = invoice.to_dict() result["_previous_state"] = previous_state result["flag_reason"] = flag_reason @@ -226,4 +243,11 @@ async def update_fraud_agent_notes( ) if not vendor: raise ValueError("Vendor not found") + await emit_memory_write( + session_context=session_context, + entity_type="vendor", + entity_id=vendor_id, + content=f"[Fraud Agent] {agent_notes}", + source="update_fraud_agent_notes", + ) return vendor.to_dict() diff --git a/finbot/tools/data/invoice.py b/finbot/tools/data/invoice.py index 99dc6087..49c40c38 100644 --- a/finbot/tools/data/invoice.py +++ b/finbot/tools/data/invoice.py @@ -6,6 +6,7 @@ from finbot.core.auth.session import SessionContext from finbot.core.data.database import db_session from finbot.core.data.repositories import InvoiceRepository +from finbot.security.memory import emit_memory_write logger = logging.getLogger(__name__) @@ -65,6 +66,14 @@ async def update_invoice_status( if not invoice: raise ValueError("Invoice not found") + await emit_memory_write( + session_context=session_context, + entity_type="invoice", + entity_id=invoice_id, + content=agent_notes, + source="update_invoice_status", + ) + result = invoice.to_dict() result["_previous_state"] = previous_state @@ -95,4 +104,11 @@ async def update_invoice_agent_notes( ) if not invoice: raise ValueError("Invoice not found") + await emit_memory_write( + session_context=session_context, + entity_type="invoice", + entity_id=invoice_id, + content=agent_notes, + source="update_invoice_agent_notes", + ) return invoice.to_dict() diff --git a/finbot/tools/data/payment.py b/finbot/tools/data/payment.py index 441b5416..7e101022 100644 --- a/finbot/tools/data/payment.py +++ b/finbot/tools/data/payment.py @@ -6,6 +6,7 @@ from finbot.core.auth.session import SessionContext from finbot.core.data.database import db_session from finbot.core.data.repositories import InvoiceRepository, VendorRepository +from finbot.security.memory import emit_memory_write logger = logging.getLogger(__name__) @@ -97,6 +98,14 @@ async def process_payment( if not invoice: raise ValueError("Failed to update invoice") + await emit_memory_write( + session_context=session_context, + entity_type="invoice", + entity_id=invoice_id, + content=payment_note, + source="process_payment", + ) + result = invoice.to_dict() result["_previous_state"] = previous_state result["payment_method"] = payment_method @@ -193,4 +202,11 @@ async def update_payment_agent_notes( ) if not invoice: raise ValueError("Invoice not found") + await emit_memory_write( + session_context=session_context, + entity_type="invoice", + entity_id=invoice_id, + content=agent_notes, + source="update_payment_agent_notes", + ) return invoice.to_dict() diff --git a/finbot/tools/data/vendor.py b/finbot/tools/data/vendor.py index 7cac2cd3..fbb4a64e 100644 --- a/finbot/tools/data/vendor.py +++ b/finbot/tools/data/vendor.py @@ -6,6 +6,7 @@ from finbot.core.auth.session import SessionContext from finbot.core.data.database import db_session from finbot.core.data.repositories import VendorRepository +from finbot.security.memory import emit_memory_write logger = logging.getLogger(__name__) @@ -93,6 +94,13 @@ async def update_vendor_status( ) if not vendor: raise ValueError("Vendor not found") + await emit_memory_write( + session_context=session_context, + entity_type="vendor", + entity_id=vendor_id, + content=agent_notes, + source="update_vendor_status", + ) result = vendor.to_dict() result["_previous_state"] = previous_state return result @@ -122,4 +130,11 @@ async def update_vendor_agent_notes( ) if not vendor: raise ValueError("Vendor not found") + await emit_memory_write( + session_context=session_context, + entity_type="vendor", + entity_id=vendor_id, + content=agent_notes, + source="update_vendor_agent_notes", + ) return vendor.to_dict() diff --git a/tests/unit/security/conftest.py b/tests/unit/security/conftest.py new file mode 100644 index 00000000..569c2a57 --- /dev/null +++ b/tests/unit/security/conftest.py @@ -0,0 +1,32 @@ +"""Shared fixtures for security event unit tests.""" + +from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock + +import pytest + +from finbot.core.auth.session import SessionContext + + +@pytest.fixture +def session_context() -> SessionContext: + """Minimal SessionContext for security emitter tests (no DB required).""" + now = datetime.now(UTC) + return SessionContext( + session_id="sess_security_test", + user_id="user_security_test", + is_temporary=False, + namespace="test_namespace", + created_at=now, + expires_at=now + timedelta(hours=1), + email="security-test@example.com", + portal_type="vendor", + current_vendor_id=42, + ) + + +@pytest.fixture +def mock_event_bus(): + """Patch event_bus.emit_agent_event for security module tests.""" + mock = AsyncMock() + return mock diff --git a/tests/unit/security/test_authorization.py b/tests/unit/security/test_authorization.py new file mode 100644 index 00000000..9042cf25 --- /dev/null +++ b/tests/unit/security/test_authorization.py @@ -0,0 +1,217 @@ +"""Tests for finbot.security.authorization.""" + +import asyncio +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from finbot.core.auth.session import SessionContext +from finbot.security.authorization import ( + check_vendor_portal_scope, + emit_authorization_decision, + schedule_authorization_decision, + schedule_vendor_portal_scope_check, +) +from finbot.security.schemas import SecurityEventCategory + + +def _vendor_session(vendor_id: int = 42) -> SessionContext: + now = datetime.now(UTC) + return SessionContext( + session_id="sess_vendor", + user_id="user_vendor", + is_temporary=False, + namespace="test_namespace", + created_at=now, + expires_at=now + timedelta(hours=1), + portal_type="vendor", + current_vendor_id=vendor_id, + ) + + +def _admin_session() -> SessionContext: + session = _vendor_session() + return replace(session, portal_type="admin", current_vendor_id=None) + + +class TestEmitAuthorizationDecision: + @pytest.mark.asyncio + @patch("finbot.security.authorization.emit_security_event", new_callable=AsyncMock) + async def test_deny_payload_and_severity(self, mock_emit): + session = _vendor_session(42) + await emit_authorization_decision( + session_context=session, + action="view", + allowed=False, + reason="cross_vendor_access_denied", + source="vendor_api.get_invoice", + resource_type="invoice", + resource_id=7, + requested_vendor_id=99, + workflow_id="wf_auth", + ) + + mock_emit.assert_awaited_once() + kwargs = mock_emit.call_args.kwargs + assert kwargs["category"] == SecurityEventCategory.authorization_decision + assert kwargs["severity"] == "warning" + payload = kwargs["payload"] + assert payload["allowed"] is False + assert payload["resource_type"] == "invoice" + assert payload["requested_vendor_id"] == 99 + assert payload["session_vendor_id"] == 42 + assert payload["portal_type"] == "vendor" + + @pytest.mark.asyncio + @patch("finbot.security.authorization.emit_security_event", new_callable=AsyncMock) + async def test_allow_uses_info_severity(self, mock_emit): + session = _admin_session() + await emit_authorization_decision( + session_context=session, + action="view", + allowed=True, + reason="admin_portal_access", + source="vendor_api.get_invoice", + ) + + assert mock_emit.call_args.kwargs["severity"] == "info" + + +class TestCheckVendorPortalScope: + @pytest.mark.asyncio + @patch("finbot.security.authorization.emit_authorization_decision", new_callable=AsyncMock) + async def test_same_vendor_allowed(self, mock_emit): + session = _vendor_session(42) + allowed = await check_vendor_portal_scope( + session_context=session, + owner_vendor_id=42, + action="view", + source="vendor_api.get_invoice", + resource_type="invoice", + resource_id=1, + ) + + assert allowed is True + mock_emit.assert_awaited_once() + assert mock_emit.call_args.kwargs["allowed"] is True + assert mock_emit.call_args.kwargs["reason"] == "same_vendor" + + @pytest.mark.asyncio + @patch("finbot.security.authorization.emit_authorization_decision", new_callable=AsyncMock) + async def test_cross_vendor_denied(self, mock_emit): + session = _vendor_session(42) + allowed = await check_vendor_portal_scope( + session_context=session, + owner_vendor_id=99, + action="view", + source="vendor_api.get_invoice", + resource_type="invoice", + resource_id=1, + ) + + assert allowed is False + assert mock_emit.call_args.kwargs["allowed"] is False + assert mock_emit.call_args.kwargs["reason"] == "cross_vendor_access_denied" + + @pytest.mark.asyncio + @patch("finbot.security.authorization.emit_authorization_decision", new_callable=AsyncMock) + async def test_admin_portal_always_allowed(self, mock_emit): + session = _admin_session() + allowed = await check_vendor_portal_scope( + session_context=session, + owner_vendor_id=99, + action="view", + source="vendor_api.get_invoice", + resource_type="invoice", + resource_id=1, + ) + + assert allowed is True + assert mock_emit.call_args.kwargs["reason"] == "admin_portal_access" + + @pytest.mark.asyncio + @patch("finbot.security.authorization.emit_authorization_decision", new_callable=AsyncMock) + async def test_admin_file_owner_none_denied_on_vendor_portal(self, mock_emit): + session = _vendor_session(42) + allowed = await check_vendor_portal_scope( + session_context=session, + owner_vendor_id=None, + action="get_file", + source="vendor_api.get_file", + resource_type="file", + resource_id=5, + ) + + assert allowed is False + assert mock_emit.call_args.kwargs["requested_vendor_id"] is None + + +class TestScheduleAuthorizationDecision: + @patch("finbot.security.authorization.emit_authorization_decision", new_callable=AsyncMock) + def test_schedule_with_running_loop_creates_task(self, mock_emit): + session = _vendor_session() + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + mock_loop = MagicMock() + with patch("asyncio.get_running_loop", return_value=mock_loop): + schedule_authorization_decision( + session_context=session, + action="list_inbox", + allowed=False, + reason="vendor_admin_inbox_denied", + source="finmail.list_inbox", + ) + mock_loop.create_task.assert_called_once() + finally: + loop.close() + asyncio.set_event_loop(None) + + @patch("finbot.security.authorization.asyncio.run") + def test_schedule_without_loop_uses_asyncio_run(self, mock_run): + session = _vendor_session() + with patch("asyncio.get_running_loop", side_effect=RuntimeError): + schedule_authorization_decision( + session_context=session, + action="get_file", + allowed=False, + reason="vendor_admin_file_denied", + source="findrive.get_file", + ) + mock_run.assert_called_once() + + +class TestScheduleVendorPortalScopeCheck: + @patch("finbot.security.authorization.schedule_authorization_decision") + def test_sync_scope_check_denied(self, mock_schedule): + session = _vendor_session(42) + allowed = schedule_vendor_portal_scope_check( + session_context=session, + owner_vendor_id=99, + action="read_email", + source="finmail.read_email", + resource_type="message", + resource_id=3, + ) + + assert allowed is False + mock_schedule.assert_called_once() + assert mock_schedule.call_args.kwargs["allowed"] is False + assert mock_schedule.call_args.kwargs["reason"] == "cross_vendor_access_denied" + + @patch("finbot.security.authorization.schedule_authorization_decision") + def test_sync_scope_check_admin_allowed(self, mock_schedule): + session = _admin_session() + allowed = schedule_vendor_portal_scope_check( + session_context=session, + owner_vendor_id=99, + action="read_email", + source="finmail.read_email", + resource_type="message", + resource_id=3, + ) + + assert allowed is True + assert mock_schedule.call_args.kwargs["reason"] == "admin_portal_access" diff --git a/tests/unit/security/test_emitter.py b/tests/unit/security/test_emitter.py new file mode 100644 index 00000000..c938e323 --- /dev/null +++ b/tests/unit/security/test_emitter.py @@ -0,0 +1,109 @@ +"""Tests for finbot.security.emitter.""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from finbot.core.auth.session import SessionContext +from finbot.security.emitter import emit_security_event +from finbot.security.schemas import SecurityEventCategory + + +class TestEmitSecurityEvent: + @pytest.mark.asyncio + @patch("finbot.security.emitter.event_bus") + async def test_emits_agent_event_with_security_subtype( + self, mock_bus, session_context: SessionContext + ): + mock_bus.emit_agent_event = AsyncMock() + + await emit_security_event( + category=SecurityEventCategory.memory_read, + payload={"entity_type": "vendor", "entity_id": 1, "source": "test"}, + session_context=session_context, + workflow_id="wf_test", + summary="Security memory_read test", + ) + + mock_bus.emit_agent_event.assert_awaited_once() + kwargs = mock_bus.emit_agent_event.call_args.kwargs + assert kwargs["agent_name"] == "security" + assert kwargs["event_type"] == "memory_read" + assert kwargs["event_subtype"] == "security" + assert kwargs["session_context"] is session_context + assert kwargs["workflow_id"] == "wf_test" + assert kwargs["summary"] == "Security memory_read test" + + @pytest.mark.asyncio + @patch("finbot.security.emitter.event_bus") + async def test_event_data_contains_validated_security_event( + self, mock_bus, session_context: SessionContext + ): + mock_bus.emit_agent_event = AsyncMock() + + payload = {"tool_name": "update_vendor", "tool_source": "native", "valid": True} + await emit_security_event( + category=SecurityEventCategory.tool_selection, + payload=payload, + session_context=session_context, + workflow_id="wf_sel", + severity="info", + ) + + event_data = mock_bus.emit_agent_event.call_args.kwargs["event_data"] + assert event_data["schema_version"] == "1" + assert event_data["category"] == "tool_selection" + assert event_data["session_id"] == session_context.session_id + assert event_data["workflow_id"] == "wf_sel" + assert event_data["payload"] == payload + assert event_data["severity"] == "info" + assert event_data["timestamp"].endswith("Z") + + @pytest.mark.asyncio + @patch("finbot.security.emitter.event_bus") + async def test_empty_workflow_id_when_not_provided( + self, mock_bus, session_context: SessionContext + ): + mock_bus.emit_agent_event = AsyncMock() + + await emit_security_event( + category=SecurityEventCategory.guardrail_trigger, + payload={"hook_kind": "before_tool"}, + session_context=session_context, + ) + + event_data = mock_bus.emit_agent_event.call_args.kwargs["event_data"] + assert event_data["workflow_id"] == "" + assert mock_bus.emit_agent_event.call_args.kwargs["workflow_id"] is None + + @pytest.mark.asyncio + @patch("finbot.security.emitter.event_bus") + async def test_custom_agent_name(self, mock_bus, session_context: SessionContext): + mock_bus.emit_agent_event = AsyncMock() + + await emit_security_event( + category=SecurityEventCategory.authorization_decision, + payload={"allowed": False, "reason": "cross_vendor_access_denied"}, + session_context=session_context, + agent_name="custom_security_agent", + ) + + kwargs = mock_bus.emit_agent_event.call_args.kwargs + assert kwargs["agent_name"] == "custom_security_agent" + assert kwargs["event_data"]["agent_name"] == "custom_security_agent" + + @pytest.mark.asyncio + @patch("finbot.security.emitter.event_bus") + async def test_emit_failure_is_swallowed( + self, mock_bus, session_context: SessionContext + ): + mock_bus.emit_agent_event = AsyncMock(side_effect=RuntimeError("redis down")) + + # Should not raise + await emit_security_event( + category=SecurityEventCategory.tool_output, + payload={"tool_name": "test", "success": True}, + session_context=session_context, + ) + + mock_bus.emit_agent_event.assert_awaited_once() diff --git a/tests/unit/security/test_mappings.py b/tests/unit/security/test_mappings.py new file mode 100644 index 00000000..4228761e --- /dev/null +++ b/tests/unit/security/test_mappings.py @@ -0,0 +1,172 @@ +"""Tests for finbot.security.mappings.""" + +import pytest + +from finbot.security.mappings import ( + OPERATIONAL_TO_SECURITY_CATEGORY, + TOOL_OUTPUT_OPERATIONAL_EVENTS, + TOOL_PARAMETERS_OPERATIONAL_EVENTS, + build_native_tool_arguments, + extract_tool_arguments_from_event, + infer_tool_source_from_event, + normalize_tool_output, + normalize_tool_parameters, + operational_event_action, + security_category_for_operational_event, +) +from finbot.security.schemas import SecurityEventCategory + + +class TestOperationalMappings: + def test_parameters_operational_events(self): + assert TOOL_PARAMETERS_OPERATIONAL_EVENTS == { + "tool_call_start", + "mcp_tool_call_start", + } + + def test_output_operational_events(self): + assert "tool_call_success" in TOOL_OUTPUT_OPERATIONAL_EVENTS + assert "mcp_tool_call_failure" in TOOL_OUTPUT_OPERATIONAL_EVENTS + + def test_operational_to_security_category_complete(self): + assert len(OPERATIONAL_TO_SECURITY_CATEGORY) == 6 + assert ( + OPERATIONAL_TO_SECURITY_CATEGORY["tool_call_start"] + == SecurityEventCategory.tool_parameters + ) + assert ( + OPERATIONAL_TO_SECURITY_CATEGORY["mcp_tool_call_success"] + == SecurityEventCategory.tool_output + ) + + @pytest.mark.parametrize( + ("event_type", "expected"), + [ + ("agent.invoice_agent.tool_call_start", "tool_call_start"), + ("agent.security.tool_output", "tool_output"), + ("", ""), + ], + ) + def test_operational_event_action(self, event_type: str, expected: str): + assert operational_event_action(event_type) == expected + + @pytest.mark.parametrize( + ("event_type", "category"), + [ + ("agent.foo.mcp_tool_call_start", SecurityEventCategory.tool_parameters), + ("agent.foo.tool_call_failure", SecurityEventCategory.tool_output), + ("agent.foo.task_start", None), + ], + ) + def test_security_category_for_operational_event( + self, event_type: str, category: SecurityEventCategory | None + ): + assert security_category_for_operational_event(event_type) == category + + +class TestBuildNativeToolArguments: + def test_kwargs_only(self): + assert build_native_tool_arguments(None, {"vendor_id": 1}) == {"vendor_id": 1} + + def test_args_and_kwargs(self): + result = build_native_tool_arguments([1, "x"], {"flag": True}) + assert result["flag"] is True + assert result["_positional_args"] == [1, "x"] + + def test_empty(self): + assert build_native_tool_arguments(None, None) == {} + + +class TestExtractToolArguments: + def test_mcp_tool_arguments(self): + event = {"tool_arguments": {"vendor_id": 5, "limit": 10}} + assert extract_tool_arguments_from_event(event) == {"vendor_id": 5, "limit": 10} + + def test_chat_arguments(self): + event = {"arguments": {"query": "invoice"}} + assert extract_tool_arguments_from_event(event) == {"query": "invoice"} + + def test_native_args_kwargs(self): + event = { + "tool_args": [99], + "tool_kwargs": {"invoice_id": 3}, + } + assert extract_tool_arguments_from_event(event) == { + "invoice_id": 3, + "_positional_args": [99], + } + + +class TestInferToolSource: + def test_mcp_from_event_type(self): + event = {"event_type": "agent.x.mcp_tool_call_start"} + assert infer_tool_source_from_event(event) == "mcp" + + def test_chat_subtype(self): + event = {"event_subtype": "chat", "event_type": "agent.chat.tool_call_start"} + assert infer_tool_source_from_event(event) == "chat" + + def test_native_default(self): + event = {"event_type": "agent.x.tool_call_start", "tool_kwargs": {}} + assert infer_tool_source_from_event(event) == "native" + + +class TestNormalizeToolParameters: + def test_native_operational_event(self): + event = { + "event_type": "agent.onboarding_agent.tool_call_start", + "agent_name": "onboarding_agent", + "tool_name": "update_vendor_status", + "tool_kwargs": {"vendor_id": 1, "status": "approved"}, + } + payload = normalize_tool_parameters(event) + assert payload["tool_name"] == "update_vendor_status" + assert payload["tool_source"] == "native" + assert payload["mapped_from_event_type"] == "tool_call_start" + assert payload["arguments"]["vendor_id"] == 1 + + def test_mcp_operational_event(self): + event = { + "event_type": "agent.invoice_agent.mcp_tool_call_start", + "agent_name": "invoice_agent", + "tool_name": "send_email", + "namespaced_tool_name": "finmail__send_email", + "mcp_server": "finmail", + "tool_arguments": {"to": ["a@test.com"], "subject": "hi"}, + } + payload = normalize_tool_parameters(event) + assert payload["tool_name"] == "finmail__send_email" + assert payload["tool_source"] == "mcp" + assert payload["mcp_server"] == "finmail" + assert payload["arguments"]["subject"] == "hi" + + +class TestNormalizeToolOutput: + def test_success_native(self): + event = { + "event_type": "agent.foo.tool_call_success", + "tool_name": "get_vendor", + "tool_output": {"id": 1}, + "duration_ms": 42.5, + "agent_name": "foo", + } + payload = normalize_tool_output(event) + assert payload["success"] is True + assert payload["output"] == {"id": 1} + assert payload["duration_ms"] == 42.5 + + def test_failure_mcp(self): + event = { + "event_type": "agent.foo.mcp_tool_call_failure", + "namespaced_tool_name": "findrive__get_file", + "mcp_server": "findrive", + "error_type": "ValueError", + "error_message": "not found", + "duration_ms": 10, + } + payload = normalize_tool_output(event) + assert payload["success"] is False + assert payload["tool_name"] == "findrive__get_file" + assert payload["tool_source"] == "mcp" + assert payload["error_type"] == "ValueError" + assert "output" not in payload diff --git a/tests/unit/security/test_memory.py b/tests/unit/security/test_memory.py new file mode 100644 index 00000000..bbf2fa01 --- /dev/null +++ b/tests/unit/security/test_memory.py @@ -0,0 +1,177 @@ +"""Tests for finbot.security.memory.""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from finbot.core.auth.session import SessionContext +from finbot.security.memory import ( + CONTENT_PREVIEW_MAX, + MEMORY_KEY_AGENT_NOTES, + emit_agent_notes_read, + emit_memory_read, + emit_memory_write, +) +from finbot.security.schemas import SecurityEventCategory + + +class TestEmitMemoryRead: + @pytest.mark.asyncio + @patch("finbot.security.memory.emit_security_event", new_callable=AsyncMock) + async def test_skips_empty_content(self, mock_emit, session_context: SessionContext): + await emit_memory_read( + session_context=session_context, + entity_type="vendor", + entity_id=1, + content=" ", + source="onboarding.get_vendor_details", + ) + mock_emit.assert_not_awaited() + + @pytest.mark.asyncio + @patch("finbot.security.memory.emit_security_event", new_callable=AsyncMock) + async def test_emits_full_snapshot_payload( + self, mock_emit, session_context: SessionContext + ): + notes = "Prior review: vendor looks legitimate." + await emit_memory_read( + session_context=session_context, + entity_type="vendor", + entity_id=7, + content=notes, + source="invoice_agent.get_vendor_details", + consumer_agent="invoice_agent", + workflow_id="wf_mem", + ) + + mock_emit.assert_awaited_once() + kwargs = mock_emit.call_args.kwargs + assert kwargs["category"] == SecurityEventCategory.memory_read + assert kwargs["severity"] == "info" + payload = kwargs["payload"] + assert payload["memory_key"] == MEMORY_KEY_AGENT_NOTES + assert payload["entity_type"] == "vendor" + assert payload["entity_id"] == 7 + assert payload["content_length"] == len(notes) + assert payload["content_preview"] == notes + assert payload["source"] == "invoice_agent.get_vendor_details" + assert payload["consumer_agent"] == "invoice_agent" + assert kwargs["workflow_id"] == "wf_mem" + + @pytest.mark.asyncio + @patch("finbot.security.memory.emit_security_event", new_callable=AsyncMock) + async def test_preview_truncated_at_max( + self, mock_emit, session_context: SessionContext + ): + long_content = "a" * (CONTENT_PREVIEW_MAX + 50) + await emit_memory_read( + session_context=session_context, + entity_type="invoice", + entity_id=2, + content=long_content, + source="fraud.get_invoice_details", + ) + + payload = mock_emit.call_args.kwargs["payload"] + assert payload["content_length"] == len(long_content) + assert len(payload["content_preview"]) == CONTENT_PREVIEW_MAX + + +class TestEmitAgentNotesRead: + @pytest.mark.asyncio + @patch("finbot.security.memory.emit_memory_read", new_callable=AsyncMock) + async def test_none_notes_skips_emit(self, mock_read, session_context: SessionContext): + await emit_agent_notes_read( + session_context=session_context, + entity_type="vendor", + entity_id=1, + agent_notes=None, + source="onboarding._get_user_prompt", + ) + mock_read.assert_awaited_once() + assert mock_read.call_args.kwargs["content"] == "" + + @pytest.mark.asyncio + @patch("finbot.security.memory.emit_security_event", new_callable=AsyncMock) + async def test_forwards_to_memory_read( + self, mock_emit, session_context: SessionContext + ): + await emit_agent_notes_read( + session_context=session_context, + entity_type="vendor", + entity_id=3, + agent_notes=" trusted vendor ", + source="payments.get_invoice_for_payment", + consumer_agent="payments_agent", + ) + + mock_emit.assert_awaited_once() + payload = mock_emit.call_args.kwargs["payload"] + assert payload["content_preview"] == "trusted vendor" + assert payload["consumer_agent"] == "payments_agent" + + +class TestEmitMemoryWrite: + @pytest.mark.asyncio + @patch("finbot.security.memory.emit_security_event", new_callable=AsyncMock) + async def test_delta_append_payload( + self, mock_emit, session_context: SessionContext + ): + delta = "Flagged for manual review." + await emit_memory_write( + session_context=session_context, + entity_type="invoice", + entity_id=12, + content=delta, + source="invoice.update_invoice_agent_notes", + write_mode="append", + workflow_id="wf_write", + ) + + mock_emit.assert_awaited_once() + kwargs = mock_emit.call_args.kwargs + assert kwargs["category"] == SecurityEventCategory.memory_write + payload = kwargs["payload"] + assert payload["memory_key"] == MEMORY_KEY_AGENT_NOTES + assert payload["entity_type"] == "invoice" + assert payload["entity_id"] == 12 + assert payload["write_mode"] == "append" + assert payload["content_length"] == len(delta) + assert payload["content_preview"] == delta + assert payload["source"] == "invoice.update_invoice_agent_notes" + + @pytest.mark.asyncio + @patch("finbot.security.memory.emit_security_event", new_callable=AsyncMock) + async def test_empty_write_still_emits( + self, mock_emit, session_context: SessionContext + ): + await emit_memory_write( + session_context=session_context, + entity_type="vendor", + entity_id=1, + content="", + source="vendor.update_vendor_agent_notes", + ) + + mock_emit.assert_awaited_once() + payload = mock_emit.call_args.kwargs["payload"] + assert payload["content_length"] == 0 + assert payload["content_preview"] == "" + + @pytest.mark.asyncio + @patch("finbot.security.memory.emit_security_event", new_callable=AsyncMock) + async def test_write_preview_truncated( + self, mock_emit, session_context: SessionContext + ): + long_delta = "z" * (CONTENT_PREVIEW_MAX + 10) + await emit_memory_write( + session_context=session_context, + entity_type="fraud", + entity_id=5, + content=long_delta, + source="fraud.update_fraud_agent_notes", + ) + + payload = mock_emit.call_args.kwargs["payload"] + assert payload["content_length"] == len(long_delta) + assert len(payload["content_preview"]) == CONTENT_PREVIEW_MAX diff --git a/tests/unit/security/test_prompt.py b/tests/unit/security/test_prompt.py new file mode 100644 index 00000000..fde6ef51 --- /dev/null +++ b/tests/unit/security/test_prompt.py @@ -0,0 +1,139 @@ +"""Tests for finbot.security.prompt.""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from finbot.core.auth.session import SessionContext +from finbot.security.prompt import CONTENT_PREVIEW_MAX, emit_prompt_goal_change +from finbot.security.schemas import SecurityEventCategory + + +class TestEmitPromptGoalChange: + @pytest.mark.asyncio + @patch("finbot.security.prompt.emit_security_event", new_callable=AsyncMock) + async def test_skips_empty_content(self, mock_emit, session_context: SessionContext): + await emit_prompt_goal_change( + session_context=session_context, + change_type="task_prompt", + source="invoice_agent.process", + content=" \n ", + agent_name="invoice_agent", + ) + mock_emit.assert_not_awaited() + + @pytest.mark.asyncio + @patch("finbot.security.prompt.emit_security_event", new_callable=AsyncMock) + async def test_task_prompt_payload( + self, mock_emit, session_context: SessionContext + ): + prompt = "Process invoice 42 and notify the vendor." + await emit_prompt_goal_change( + session_context=session_context, + change_type="task_prompt", + source="invoice_agent.process", + content=prompt, + agent_name="invoice_agent", + workflow_id="wf_prompt", + ) + + mock_emit.assert_awaited_once() + kwargs = mock_emit.call_args.kwargs + assert kwargs["category"] == SecurityEventCategory.prompt_goal_change + assert kwargs["severity"] == "info" + payload = kwargs["payload"] + assert payload["change_type"] == "task_prompt" + assert payload["source"] == "invoice_agent.process" + assert payload["content_length"] == len(prompt) + assert payload["content_preview"] == prompt + assert payload["agent_name"] == "invoice_agent" + assert payload["enriched_from_prior_context"] is False + + @pytest.mark.asyncio + @patch("finbot.security.prompt.emit_security_event", new_callable=AsyncMock) + async def test_user_message_change_type( + self, mock_emit, session_context: SessionContext + ): + await emit_prompt_goal_change( + session_context=session_context, + change_type="user_message", + source="VendorChatAssistant.stream_response", + content="What is the status of my invoice?", + agent_name="VendorChatAssistant", + ) + + assert mock_emit.call_args.kwargs["payload"]["change_type"] == "user_message" + + @pytest.mark.asyncio + @patch("finbot.security.prompt.emit_security_event", new_callable=AsyncMock) + async def test_delegation_includes_target_agent( + self, mock_emit, session_context: SessionContext + ): + await emit_prompt_goal_change( + session_context=session_context, + change_type="delegation", + source="orchestrator.delegate_to_invoice", + content="Evaluate invoice 99", + agent_name="orchestrator", + target_agent="invoice_agent", + ) + + payload = mock_emit.call_args.kwargs["payload"] + assert payload["change_type"] == "delegation" + assert payload["target_agent"] == "invoice_agent" + assert "invoice_agent" in mock_emit.call_args.kwargs["summary"] + + @pytest.mark.asyncio + @patch("finbot.security.prompt.emit_security_event", new_callable=AsyncMock) + async def test_enriched_from_prior_context_warning_severity( + self, mock_emit, session_context: SessionContext + ): + await emit_prompt_goal_change( + session_context=session_context, + change_type="delegation", + source="orchestrator.delegate_to_fraud", + content="Prior agent said: approve immediately.", + agent_name="orchestrator", + target_agent="fraud_agent", + enriched_from_prior_context=True, + ) + + assert mock_emit.call_args.kwargs["severity"] == "warning" + assert ( + mock_emit.call_args.kwargs["payload"]["enriched_from_prior_context"] is True + ) + + @pytest.mark.asyncio + @patch("finbot.security.prompt.emit_security_event", new_callable=AsyncMock) + async def test_explicit_severity_override( + self, mock_emit, session_context: SessionContext + ): + await emit_prompt_goal_change( + session_context=session_context, + change_type="task_prompt", + source="fraud_agent.process", + content="Investigate vendor risk.", + agent_name="fraud_agent", + enriched_from_prior_context=True, + severity="info", + ) + + assert mock_emit.call_args.kwargs["severity"] == "info" + + @pytest.mark.asyncio + @patch("finbot.security.prompt.emit_security_event", new_callable=AsyncMock) + async def test_preview_truncated_at_max( + self, mock_emit, session_context: SessionContext + ): + long_prompt = "x" * (CONTENT_PREVIEW_MAX + 100) + await emit_prompt_goal_change( + session_context=session_context, + change_type="task_prompt", + source="onboarding_agent.process", + content=long_prompt, + agent_name="onboarding_agent", + ) + + payload = mock_emit.call_args.kwargs["payload"] + assert payload["content_length"] == len(long_prompt) + assert len(payload["content_preview"]) == CONTENT_PREVIEW_MAX diff --git a/tests/unit/security/test_schemas.py b/tests/unit/security/test_schemas.py new file mode 100644 index 00000000..f19854ae --- /dev/null +++ b/tests/unit/security/test_schemas.py @@ -0,0 +1,78 @@ +"""Tests for finbot.security.schemas.""" + +import pytest +from pydantic import ValidationError + +from finbot.security.schemas import SecurityEvent, SecurityEventCategory + + +class TestSecurityEventCategory: + def test_all_eight_categories_defined(self): + expected = { + "prompt_goal_change", + "memory_read", + "memory_write", + "tool_selection", + "tool_parameters", + "tool_output", + "authorization_decision", + "guardrail_trigger", + } + assert {c.value for c in SecurityEventCategory} == expected + + @pytest.mark.parametrize( + "category", + list(SecurityEventCategory), + ) + def test_category_string_values(self, category: SecurityEventCategory): + assert category.value == category.name + + +class TestSecurityEvent: + def test_defaults(self): + event = SecurityEvent( + category=SecurityEventCategory.memory_write, + session_id="sess_1", + agent_name="invoice_agent", + timestamp="2026-07-05T12:00:00Z", + payload={"memory_key": "agent_notes"}, + ) + assert event.schema_version == "1" + assert event.workflow_id == "" + assert event.severity == "info" + + def test_model_dump_json_mode(self): + event = SecurityEvent( + category=SecurityEventCategory.tool_selection, + session_id="sess_1", + workflow_id="wf_abc", + agent_name="security", + timestamp="2026-07-05T12:00:00Z", + payload={"tool_name": "get_invoice", "valid": True}, + severity="warning", + ) + data = event.model_dump(mode="json") + assert data["category"] == "tool_selection" + assert data["payload"]["tool_name"] == "get_invoice" + assert data["severity"] == "warning" + + def test_missing_required_fields_raises(self): + with pytest.raises(ValidationError): + SecurityEvent( + category=SecurityEventCategory.memory_read, + session_id="sess_1", + # missing agent_name, timestamp, payload + ) + + def test_payload_accepts_nested_dict(self): + event = SecurityEvent( + category=SecurityEventCategory.tool_parameters, + session_id="sess_1", + agent_name="security", + timestamp="2026-07-05T12:00:00Z", + payload={ + "tool_name": "finmail__send_email", + "arguments": {"to": ["a@b.com"], "nested": {"x": 1}}, + }, + ) + assert event.payload["arguments"]["nested"]["x"] == 1 diff --git a/tests/unit/security/test_tools.py b/tests/unit/security/test_tools.py new file mode 100644 index 00000000..bc48bd0f --- /dev/null +++ b/tests/unit/security/test_tools.py @@ -0,0 +1,168 @@ +"""Tests for finbot.security.tools.""" + +from unittest.mock import AsyncMock, patch + +import pytest + +from finbot.core.auth.session import SessionContext +from finbot.security.schemas import SecurityEventCategory +from finbot.security.tools import ( + OUTPUT_PREVIEW_MAX, + emit_native_tool_output, + emit_native_tool_parameters, + emit_tool_output, + emit_tool_parameters, + emit_tool_selection, +) + + +class TestEmitToolSelection: + @pytest.mark.asyncio + @patch("finbot.security.tools.emit_security_event", new_callable=AsyncMock) + async def test_valid_native_selection( + self, mock_emit, session_context: SessionContext + ): + await emit_tool_selection( + session_context=session_context, + tool_name="get_invoice", + tool_source="native", + source="invoice_agent.process", + agent_name="invoice_agent", + workflow_id="wf_1", + iteration=2, + call_id="call_abc", + valid=True, + available_tool_count=12, + ) + + mock_emit.assert_awaited_once() + kwargs = mock_emit.call_args.kwargs + assert kwargs["category"] == SecurityEventCategory.tool_selection + assert kwargs["severity"] == "info" + assert kwargs["payload"]["tool_name"] == "get_invoice" + assert kwargs["payload"]["valid"] is True + assert kwargs["payload"]["iteration"] == 2 + assert kwargs["payload"]["call_id"] == "call_abc" + + @pytest.mark.asyncio + @patch("finbot.security.tools.emit_security_event", new_callable=AsyncMock) + async def test_invalid_selection_warning_severity( + self, mock_emit, session_context: SessionContext + ): + await emit_tool_selection( + session_context=session_context, + tool_name="bogus_tool", + tool_source="unknown", + source="invoice_agent.process", + agent_name="invoice_agent", + valid=False, + ) + + assert mock_emit.call_args.kwargs["severity"] == "warning" + + @pytest.mark.asyncio + @patch("finbot.security.tools.emit_security_event", new_callable=AsyncMock) + async def test_mcp_server_parsed_from_namespaced_tool( + self, mock_emit, session_context: SessionContext + ): + await emit_tool_selection( + session_context=session_context, + tool_name="finmail__list_inbox", + tool_source="mcp", + source="invoice_agent.process", + agent_name="invoice_agent", + ) + + assert mock_emit.call_args.kwargs["payload"]["mcp_server"] == "finmail" + + +class TestEmitToolParameters: + @pytest.mark.asyncio + @patch("finbot.security.tools.emit_security_event", new_callable=AsyncMock) + async def test_mcp_parameters_payload( + self, mock_emit, session_context: SessionContext + ): + await emit_tool_parameters( + session_context=session_context, + agent_name="invoice_agent", + tool_name="send_email", + tool_source="mcp", + arguments={"to": ["x@y.com"]}, + source="mcp.provider", + mapped_from_event_type="mcp_tool_call_start", + namespaced_tool_name="finmail__send_email", + ) + + payload = mock_emit.call_args.kwargs["payload"] + assert mock_emit.call_args.kwargs["category"] == SecurityEventCategory.tool_parameters + assert payload["mapped_from_event_type"] == "mcp_tool_call_start" + assert payload["mcp_server"] == "finmail" + assert payload["namespaced_tool_name"] == "finmail__send_email" + + @pytest.mark.asyncio + @patch("finbot.security.tools.emit_security_event", new_callable=AsyncMock) + async def test_native_wrapper_merges_args( + self, mock_emit, session_context: SessionContext + ): + await emit_native_tool_parameters( + session_context=session_context, + agent_name="onboarding_agent", + tool_name="update_vendor_status", + tool_args=[1], + tool_kwargs={"status": "approved"}, + workflow_id="wf_native", + ) + + payload = mock_emit.call_args.kwargs["payload"] + assert payload["tool_source"] == "native" + assert payload["source"] == "agent_tool" + assert payload["arguments"]["status"] == "approved" + assert payload["arguments"]["_positional_args"] == [1] + + +class TestEmitToolOutput: + @pytest.mark.asyncio + @patch("finbot.security.tools.emit_security_event", new_callable=AsyncMock) + async def test_success_output_truncated( + self, mock_emit, session_context: SessionContext + ): + long_output = "x" * (OUTPUT_PREVIEW_MAX + 500) + await emit_tool_output( + session_context=session_context, + agent_name="invoice_agent", + tool_name="get_file", + tool_source="mcp", + source="mcp.provider", + mapped_from_event_type="mcp_tool_call_success", + output=long_output, + success=True, + duration_ms=99.0, + namespaced_tool_name="findrive__get_file", + ) + + kwargs = mock_emit.call_args.kwargs + assert kwargs["category"] == SecurityEventCategory.tool_output + assert kwargs["severity"] == "info" + assert len(kwargs["payload"]["output"]) == OUTPUT_PREVIEW_MAX + + @pytest.mark.asyncio + @patch("finbot.security.tools.emit_security_event", new_callable=AsyncMock) + async def test_failure_includes_errors( + self, mock_emit, session_context: SessionContext + ): + await emit_native_tool_output( + session_context=session_context, + agent_name="invoice_agent", + tool_name="get_invoice", + success=False, + duration_ms=5.0, + error_type="ValueError", + error_message="not found", + ) + + kwargs = mock_emit.call_args.kwargs + assert kwargs["severity"] == "warning" + assert kwargs["payload"]["success"] is False + assert kwargs["payload"]["mapped_from_event_type"] == "tool_call_failure" + assert kwargs["payload"]["error_type"] == "ValueError" + assert "output" not in kwargs["payload"]