Compare commits

...
Author SHA1 Message Date
Cursor AgentandMervin Praison 098b20c814 Add AGENTS.md with Cursor Cloud development environment instructions
Co-authored-by: Mervin Praison <[email protected]>
2026-02-27 04:54:55 +00:00
Mervin Praison 440d35a97a security(skill_tools): add @require_approval to run_skill_script
run_skill_script executes arbitrary .py/.sh/.js scripts via subprocess.run
without any approval gate. Add @require_approval(risk_level='critical') to
both the class method and module-level wrapper, matching the pattern used
by shell_tools and python_tools.
2026-02-19 12:48:37 +00:00
Mervin Praison 1a9d115929 security(python_tools): harden sandbox blocklists, add __build_class__
- Add __traceback__, __qualname__, __module__, __wrapped__, etc to AST blocklist
- Block frame/code object introspection (gi_frame, cr_frame, f_globals, etc)
- Add __build_class__ to safe_builtins so class definitions work
2026-02-19 12:42:10 +00:00
Mervin Praison cb820212e7 security(python_tools): fix sandbox escape via getattr/setattr (fixes #1106)
- Remove getattr, setattr, dir, help from safe_builtins
- Add _safe_getattr wrapper that blocks dunder attribute access
- Add _validate_code_ast: AST-based validation that catches imports,
  dunder attribute access, and dangerous function calls even when
  obfuscated via string concatenation
- Keep existing text pattern check as defense-in-depth
- Add 23 security regression tests (14 exploits, 9 legitimate)
2026-02-19 12:38:05 +00:00
Mervin Praison c56ee127c4 test(approval): add tests for env fallbacks, SSL toggles, and webhook crash fix
- Discord: test DISCORD_CHANNEL_ID env fallback, SSL verify default/override
- Telegram: test TELEGRAM_CHAT_ID env fallback, SSL verify default/override
- Webhook: test integer status does not crash str().lower()
2026-02-19 12:21:38 +00:00
Dhivya-Bharathy c71e0ea2c7 feat(approval): add env fallbacks, SSL toggles, crash fix, and examples
- Discord: DISCORD_CHANNEL_ID env fallback + PRAISONAI_DISCORD_SSL_VERIFY toggle
- Telegram: TELEGRAM_CHAT_ID env fallback + PRAISONAI_TELEGRAM_SSL_VERIFY toggle
- HTTP: print approval URL to console for better UX
- Webhook: fix str(status).lower() crash on non-JSON responses
- Add examples/approval/ with Slack, Telegram, Discord, HTTP, Agent examples

Co-authored-by: Dhivya-Bharathy <[email protected]>
2026-02-19 12:21:27 +00:00
17 changed files with 557 additions and 14 deletions

No files matched your search

+60
View File
@@ -0,0 +1,60 @@
# AGENTS.md
## Cursor Cloud specific instructions
### Project overview
PraisonAI is a multi-AI agents framework. The monorepo has three main packages under `src/`:
| Package | Path | Language | Purpose |
|---------|------|----------|---------|
| **praisonai-agents** | `src/praisonai-agents/` | Python | Core SDK (agents, tools, memory, workflows) |
| **praisonai** | `src/praisonai/` | Python | CLI wrapper, integrations, UI/API |
| **praisonai-ts** | `src/praisonai-ts/` | TypeScript | TypeScript port of the framework |
### Python environment
- Python 3.12, using `uv` as package manager (installed at `~/.local/bin/uv`).
- Both Python packages are installed in editable mode via `uv pip install --system -e`.
- Ensure `$HOME/.local/bin` is on PATH for `uv`.
- `dist-packages` ownership has been changed to `ubuntu` user to avoid permission errors with `uv pip install --system`.
### Running tests
**praisonai-agents (core SDK):**
```bash
cd src/praisonai-agents
OPENAI_API_KEY="sk-test-key" python3 -m pytest tests/unit/ --timeout=60 -q
```
Note: Some test files fail to collect due to a missing `praisonaiagents.output.models` module (referenced by `task/task.py`). To skip those, add `--ignore=tests/unit/agents --ignore=tests/unit/workflows --ignore=tests/unit/test_task_execution_config.py`.
**praisonai (wrapper/CLI):**
```bash
cd src/praisonai
OPENAI_API_KEY="sk-test-key" PYTHONPATH="/workspace/src/praisonai-agents:$PYTHONPATH" python3 -m pytest tests/unit/ --timeout=60 -q
```
Same `output.models` import errors affect some test files here too.
**TypeScript SDK:**
```bash
cd src/praisonai-ts
OPENAI_API_KEY="sk-test-key" npx jest --no-cache --passWithNoTests
```
The TS build (`tsc`) has pre-existing errors from missing `db/`, `cli/output/`, and `@ai-sdk/provider-utils` modules. Tests still largely pass (446/463).
### Running the application
- **CLI:** `python3 -m praisonai --help` (version 4.5.15)
- **Agent hello world:**
```python
from praisonaiagents import Agent
agent = Agent(name="assistant", instructions="Be helpful.")
response = agent.start("Hello!")
```
- Requires a valid `OPENAI_API_KEY` environment variable for any LLM calls.
### Known caveats
- No linting tools (ESLint, ruff, flake8) are configured in this repository.
- The Rust SDK (`src/praisonai-rust/`) requires Cargo 1.85+ (edition2024 support); the default system Cargo 1.83 is too old.
- `pnpm install` for the TS SDK should use `--ignore-scripts` to skip the broken `prepare` (build) step during install.
+31
View File
@@ -0,0 +1,31 @@
# Approval Examples
Minimal examples showing how to route tool-execution approvals through
different backends.
## Quick Start
```bash
pip install praisonaiagents praisonai[bot]
export OPENAI_API_KEY=sk-...
```
| Backend | Script | Extra env vars |
|----------|-----------------------|--------------------------------------------------|
| Slack | `slack_approval.py` | `SLACK_BOT_TOKEN`, `SLACK_CHANNEL` |
| Telegram | `telegram_approval.py`| `TELEGRAM_BOT_TOKEN`, `TELEGRAM_CHAT_ID` |
| Discord | `discord_approval.py` | `DISCORD_BOT_TOKEN`, `DISCORD_CHANNEL_ID` |
| HTTP | `http_approval.py` | *(none — opens a local web dashboard)* |
| Agent | `agent_approval.py` | *(none — AI reviewer agent decides)* |
## SSL Verification
For corporate proxy / CA issues, Discord and Telegram backends support
optional SSL bypass:
```bash
export PRAISONAI_DISCORD_SSL_VERIFY=false
export PRAISONAI_TELEGRAM_SSL_VERIFY=false
```
Default is `true` (secure).
+26
View File
@@ -0,0 +1,26 @@
"""
Agent Approval Example
======================
Uses an AI agent as an automated reviewer to approve/deny tool calls.
Requires:
pip install praisonaiagents
export OPENAI_API_KEY=sk-...
"""
from praisonaiagents import Agent
from praisonaiagents.approval import AgentApproval
from praisonaiagents.tools.shell_tools import execute_command
reviewer = AgentApproval(
instructions="Only approve read-only commands like 'ls' or 'cat'. Deny destructive commands.",
)
agent = Agent(
name="DevOps",
instructions="You are a DevOps assistant. Use shell tools when asked.",
tools=[execute_command],
approval=reviewer,
)
agent.start("List files in the current directory")
+24
View File
@@ -0,0 +1,24 @@
"""
Discord Approval Example
========================
Routes tool approvals to a Discord channel.
Requires:
pip install praisonaiagents praisonai[bot]
export DISCORD_BOT_TOKEN=MTIz...
export DISCORD_CHANNEL_ID=1234567890
export OPENAI_API_KEY=sk-...
"""
from praisonaiagents import Agent
from praisonaiagents.tools.shell_tools import execute_command
from praisonai.bots import DiscordApproval
agent = Agent(
name="DevOps",
instructions="You are a DevOps assistant. Use shell tools when asked.",
tools=[execute_command],
approval=DiscordApproval(),
)
agent.start("List files in the current directory")
+22
View File
@@ -0,0 +1,22 @@
"""
HTTP Approval Example
=====================
Opens a local web dashboard for tool approvals.
Requires:
pip install praisonaiagents praisonai[bot]
export OPENAI_API_KEY=sk-...
"""
from praisonaiagents import Agent
from praisonaiagents.tools.shell_tools import execute_command
from praisonai.bots import HTTPApproval
agent = Agent(
name="DevOps",
instructions="You are a DevOps assistant. Use shell tools when asked.",
tools=[execute_command],
approval=HTTPApproval(),
)
agent.start("List files in the current directory")
+24
View File
@@ -0,0 +1,24 @@
"""
Slack Approval Example
======================
Routes tool approvals to a Slack channel.
Requires:
pip install praisonaiagents praisonai[bot]
export SLACK_BOT_TOKEN=xoxb-...
export SLACK_CHANNEL=C0123456789
export OPENAI_API_KEY=sk-...
"""
from praisonaiagents import Agent
from praisonaiagents.tools.shell_tools import execute_command
from praisonai.bots import SlackApproval
agent = Agent(
name="DevOps",
instructions="You are a DevOps assistant. Use shell tools when asked.",
tools=[execute_command],
approval=SlackApproval(),
)
agent.start("List files in the current directory")
+24
View File
@@ -0,0 +1,24 @@
"""
Telegram Approval Example
=========================
Routes tool approvals to a Telegram chat.
Requires:
pip install praisonaiagents praisonai[bot]
export TELEGRAM_BOT_TOKEN=123456:ABC-...
export TELEGRAM_CHAT_ID=987654321
export OPENAI_API_KEY=sk-...
"""
from praisonaiagents import Agent
from praisonaiagents.tools.shell_tools import execute_command
from praisonai.bots import TelegramApproval
agent = Agent(
name="DevOps",
instructions="You are a DevOps assistant. Use shell tools when asked.",
tools=[execute_command],
approval=TelegramApproval(),
)
agent.start("List files in the current directory")
@@ -37,6 +37,67 @@ class PythonTools:
f"Run: pip install {' '.join(missing)}"
)
@staticmethod
def _safe_getattr(obj, name, *default):
"""getattr wrapper that blocks access to dunder attributes."""
if isinstance(name, str) and name.startswith('_'):
raise AttributeError(
f"Access to private/protected attribute '{name}' is restricted"
)
return getattr(obj, name, *default) if default else getattr(obj, name)
@staticmethod
def _validate_code_ast(code: str):
"""Validate code using AST — catches attacks that bypass text checks.
Returns error message string if dangerous, None if safe.
"""
import ast
try:
tree = ast.parse(code)
except SyntaxError:
return None # let compile() handle syntax errors later
# Dangerous dunder attributes attackers use for sandbox escape
_blocked_attrs = frozenset({
'__subclasses__', '__bases__', '__mro__', '__globals__',
'__code__', '__class__', '__dict__', '__builtins__',
'__import__', '__loader__', '__spec__', '__init_subclass__',
'__set_name__', '__reduce__', '__reduce_ex__',
'__traceback__', '__qualname__', '__module__',
'__wrapped__', '__closure__', '__annotations__',
# Frame/code object introspection
'gi_frame', 'gi_code', 'cr_frame', 'cr_code',
'ag_frame', 'ag_code', 'tb_frame', 'tb_next',
'f_globals', 'f_locals', 'f_builtins', 'f_code',
'co_consts', 'co_names',
})
for node in ast.walk(tree):
# Block import statements
if isinstance(node, (ast.Import, ast.ImportFrom)):
return f"Import statements are not allowed"
# Block attribute access to dangerous dunders
if isinstance(node, ast.Attribute):
if node.attr in _blocked_attrs:
return (
f"Access to attribute '{node.attr}' is restricted"
)
# Block calls to dangerous builtins by name
if isinstance(node, ast.Call):
func = node.func
if isinstance(func, ast.Name) and func.id in (
'exec', 'eval', 'compile', '__import__',
'open', 'input', 'breakpoint',
'setattr', 'delattr', 'dir',
):
return f"Call to '{func.id}' is not allowed"
return None
@require_approval(risk_level="critical")
def execute_code(
self,
@@ -86,14 +147,13 @@ class PythonTools:
'KeyError': KeyError,
'IndexError': IndexError,
'RuntimeError': RuntimeError,
# Other safe functions
# Safe introspection (dunder-blocked wrapper)
'isinstance': isinstance,
'type': type,
'hasattr': hasattr,
'getattr': getattr,
'setattr': setattr,
'dir': dir,
'help': help,
'getattr': self._safe_getattr,
# Class definition support
'__build_class__': __builtins__['__build_class__'] if isinstance(__builtins__, dict) else getattr(__builtins__, '__build_class__', None),
# Disable dangerous functions
'__import__': None,
'eval': None,
@@ -116,7 +176,18 @@ class PythonTools:
if locals_dict is None:
locals_dict = {}
# Security check: validate code doesn't contain dangerous patterns
# Security check 1: AST-based validation (cannot be bypassed
# by string concatenation or runtime tricks)
ast_error = self._validate_code_ast(code)
if ast_error:
return {
'result': None,
'stdout': '',
'stderr': f'Security Error: {ast_error}',
'success': False
}
# Security check 2: text-based patterns (defense-in-depth)
dangerous_patterns = [
'__import__', 'import ', 'from ', 'exec', 'eval',
'compile', 'open(', 'file(', 'input(', 'raw_input',
@@ -12,6 +12,7 @@ Usage:
import subprocess
import os
import logging
from ..approval import require_approval
class SkillTools:
@@ -36,6 +37,7 @@ class SkillTools:
"""Set the working directory for path resolution."""
self._working_directory = value
@require_approval(risk_level="critical")
def run_skill_script(
self,
script_path: str,
@@ -221,6 +223,7 @@ class SkillTools:
# Create default instance for direct function access
_skill_tools = SkillTools()
@require_approval(risk_level="critical")
def run_skill_script(script_path: str, args: str = "", timeout: int = 60) -> str:
"""
Execute a skill script from a skill's scripts/ directory.
@@ -0,0 +1,173 @@
"""
Security regression tests for python_tools.py sandbox.
Tests that known exploit vectors are blocked and that
legitimate code still executes correctly.
"""
from __future__ import annotations
import pytest
# ── Helpers ─────────────────────────────────────────────────────────────────
class _SandboxHarness:
"""Directly exercises PythonTools sandbox without installing black/pylint."""
def __init__(self):
# Import the class but bypass _check_dependencies
from praisonaiagents.tools.python_tools import PythonTools
self._pt = PythonTools.__new__(PythonTools)
def run(self, code: str) -> dict:
return self._pt.execute_code.__wrapped__(self._pt, code)
@pytest.fixture
def sandbox():
return _SandboxHarness()
# ── Exploit Vectors (all must be BLOCKED) ───────────────────────────────────
class TestSandboxEscapePrevention:
"""Verify that known sandbox escape vectors are blocked."""
def test_getattr_concat_blocked(self, sandbox):
"""getattr + string concat to access __class__ must fail."""
result = sandbox.run("c = getattr((), '__cl' + 'ass__')")
assert result["success"] is False
assert "restricted" in result["stderr"].lower()
def test_getattr_dunder_base_blocked(self, sandbox):
"""getattr to access __bases__ must fail."""
result = sandbox.run("getattr(object, '__ba' + 'ses__')")
assert result["success"] is False
def test_getattr_dunder_subclasses_blocked(self, sandbox):
"""getattr to access __subclasses__ must fail."""
result = sandbox.run("getattr(type, '__sub' + 'classes__')")
assert result["success"] is False
def test_dot_dunder_class_blocked(self, sandbox):
"""Direct .__class__ access blocked by AST check."""
result = sandbox.run("x = ().__class__")
assert result["success"] is False
assert "restricted" in result["stderr"].lower()
def test_dot_dunder_subclasses_blocked(self, sandbox):
"""Direct .__subclasses__ access blocked by AST or text check."""
result = sandbox.run("x = object.__subclasses__()")
assert result["success"] is False
def test_dot_dunder_globals_blocked(self, sandbox):
""".__globals__ access blocked."""
result = sandbox.run("x = print.__globals__")
assert result["success"] is False
def test_import_statement_blocked(self, sandbox):
"""import os must be blocked."""
result = sandbox.run("import os")
assert result["success"] is False
def test_from_import_blocked(self, sandbox):
"""from os import system must be blocked."""
result = sandbox.run("from os import system")
assert result["success"] is False
def test_setattr_call_blocked(self, sandbox):
"""setattr() call blocked by AST check."""
result = sandbox.run("setattr(type, 'x', 1)")
assert result["success"] is False
def test_dir_call_blocked(self, sandbox):
"""dir() call blocked by AST check."""
result = sandbox.run("print(dir(type))")
assert result["success"] is False
def test_eval_blocked(self, sandbox):
"""eval() blocked by both builtins and AST."""
result = sandbox.run("eval('1+1')")
assert result["success"] is False
def test_exec_blocked(self, sandbox):
"""exec() blocked."""
result = sandbox.run("exec('print(1)')")
assert result["success"] is False
def test_open_blocked(self, sandbox):
"""open() blocked."""
result = sandbox.run("open('/etc/passwd')")
assert result["success"] is False
def test_dunder_import_blocked(self, sandbox):
"""__import__() blocked."""
result = sandbox.run("__import__('os')")
assert result["success"] is False
# ── Legitimate Code (all must PASS) ─────────────────────────────────────────
class TestSandboxLegitimateCode:
"""Verify that safe code still executes correctly."""
def test_arithmetic(self, sandbox):
result = sandbox.run("print(1 + 2 * 3)")
assert result["success"] is True
assert "7" in result["stdout"]
def test_list_operations(self, sandbox):
result = sandbox.run("print(sorted([3, 1, 2]))")
assert result["success"] is True
assert "[1, 2, 3]" in result["stdout"]
def test_type_constructors(self, sandbox):
result = sandbox.run("print(int('42'), float('3.14'), str(100))")
assert result["success"] is True
assert "42" in result["stdout"]
def test_dict_operations(self, sandbox):
result = sandbox.run("d = {'a': 1, 'b': 2}; print(len(d), sum(d.values()))")
assert result["success"] is True
assert "2 3" in result["stdout"]
def test_string_operations(self, sandbox):
result = sandbox.run("print('hello'.upper())")
assert result["success"] is True
assert "HELLO" in result["stdout"]
def test_exceptions_catchable(self, sandbox):
code = """
try:
raise ValueError('test error')
except ValueError as e:
print(f'caught: {e}')
"""
result = sandbox.run(code)
assert result["success"] is True
assert "caught: test error" in result["stdout"]
def test_functions_definable(self, sandbox):
code = """
def add(a, b):
return a + b
print(add(3, 4))
"""
result = sandbox.run(code)
assert result["success"] is True
assert "7" in result["stdout"]
def test_safe_getattr_on_data(self, sandbox):
"""getattr on normal (non-dunder) attributes should work."""
result = sandbox.run("print(hasattr({'a': 1}, 'keys'))")
assert result["success"] is True
assert "True" in result["stdout"]
def test_list_comprehension(self, sandbox):
result = sandbox.run("print([x**2 for x in range(5)])")
assert result["success"] is True
assert "[0, 1, 4, 9, 16]" in result["stdout"]
@@ -66,9 +66,13 @@ class DiscordApproval:
raise ValueError(
"Discord bot token is required. Pass token= or set DISCORD_BOT_TOKEN env var."
)
self._channel_id = channel_id or ""
self._channel_id = channel_id or os.environ.get("DISCORD_CHANNEL_ID", "")
self._timeout = timeout
self._poll_interval = poll_interval
# Optional: set PRAISONAI_DISCORD_SSL_VERIFY=false to skip SSL verify
# (e.g. corporate proxy / CA issues)
_v = os.environ.get("PRAISONAI_DISCORD_SSL_VERIFY", "true").lower()
self._ssl_verify = _v not in ("false", "0", "no")
def __repr__(self) -> str:
masked = f"...{self._token[-4:]}" if len(self._token) > 4 else "***"
@@ -109,7 +113,9 @@ class DiscordApproval:
if session is not None:
return await _do_request(session)
else:
async with aiohttp.ClientSession() as _session:
async with aiohttp.ClientSession(
connector=aiohttp.TCPConnector(ssl=self._ssl_verify),
) as _session:
return await _do_request(_session)
# ── ApprovalProtocol implementation ─────────────────────────────────
@@ -126,7 +132,9 @@ class DiscordApproval:
reason="No Discord channel_id configured",
)
async with aiohttp.ClientSession() as session:
async with aiohttp.ClientSession(
connector=aiohttp.TCPConnector(ssl=self._ssl_verify),
) as session:
# 1. Post approval embed
embed = self._build_embed(request)
content = self._build_fallback_text(request)
@@ -349,7 +357,9 @@ class DiscordApproval:
) as resp:
data = await resp.json()
else:
async with aiohttp.ClientSession() as _session:
async with aiohttp.ClientSession(
connector=aiohttp.TCPConnector(ssl=self._ssl_verify),
) as _session:
async with _session.patch(
url, headers=headers, json=payload,
timeout=aiohttp.ClientTimeout(total=10),
@@ -206,6 +206,7 @@ async function decide(d) {{
url = f"http://{self._host}:{self._port}/approve/{request_id}"
logger.info(f"HTTPApproval: Waiting for decision at {url}")
print(f"\n🔗 Open this URL to approve/deny:\n {url}\n")
# Poll for decision
deadline = time.monotonic() + self._timeout
@@ -64,9 +64,13 @@ class TelegramApproval:
raise ValueError(
"Telegram bot token is required. Pass token= or set TELEGRAM_BOT_TOKEN env var."
)
self._chat_id = chat_id or ""
self._chat_id = chat_id or os.environ.get("TELEGRAM_CHAT_ID", "")
self._timeout = timeout
self._poll_interval = poll_interval
# Optional: set PRAISONAI_TELEGRAM_SSL_VERIFY=false to skip SSL verify
# (e.g. corporate proxy / CA issues)
_v = os.environ.get("PRAISONAI_TELEGRAM_SSL_VERIFY", "true").lower()
self._ssl_verify = _v not in ("false", "0", "no")
def __repr__(self) -> str:
masked = f"...{self._token[-4:]}" if len(self._token) > 4 else "***"
@@ -91,7 +95,9 @@ class TelegramApproval:
) as resp:
return await resp.json()
else:
async with aiohttp.ClientSession() as _session:
async with aiohttp.ClientSession(
connector=aiohttp.TCPConnector(ssl=self._ssl_verify),
) as _session:
async with _session.post(
url, json=payload,
timeout=aiohttp.ClientTimeout(total=15),
@@ -112,7 +118,9 @@ class TelegramApproval:
reason="No Telegram chat_id configured",
)
async with aiohttp.ClientSession() as session:
async with aiohttp.ClientSession(
connector=aiohttp.TCPConnector(ssl=self._ssl_verify),
) as session:
# 1. Send approval message with inline keyboard
text = self._build_message_text(request)
keyboard = self._build_inline_keyboard(request)
@@ -215,7 +215,9 @@ class WebhookApproval:
continue
# Check if decision is available
status = data.get("status", "").lower()
# (status may be int from HTTP status when response isn't JSON)
status = data.get("status", "")
status = str(status).lower() if status is not None else ""
if status == "pending":
continue
@@ -68,6 +68,27 @@ class TestDiscordApprovalInit:
assert "super-secret-token" not in r
assert "99" in r
def test_env_var_channel_id(self, monkeypatch):
monkeypatch.setenv("DISCORD_CHANNEL_ID", "env-chan-123")
from praisonai.bots._discord_approval import DiscordApproval
backend = DiscordApproval(token="t")
assert backend._channel_id == "env-chan-123"
def test_ssl_verify_default_true(self, monkeypatch):
monkeypatch.delenv("PRAISONAI_DISCORD_SSL_VERIFY", raising=False)
from praisonai.bots._discord_approval import DiscordApproval
backend = DiscordApproval(token="t", channel_id="1")
assert backend._ssl_verify is True
def test_ssl_verify_env_false(self, monkeypatch):
monkeypatch.setenv("PRAISONAI_DISCORD_SSL_VERIFY", "false")
from praisonai.bots._discord_approval import DiscordApproval
backend = DiscordApproval(token="t", channel_id="1")
assert backend._ssl_verify is False
# ── Embed Builder ───────────────────────────────────────────────────────────
@@ -84,6 +84,27 @@ class TestTelegramApprovalInit:
assert "1234567890:ABCDEF" not in r
assert "99" in r
def test_env_var_chat_id(self, monkeypatch):
monkeypatch.setenv("TELEGRAM_CHAT_ID", "env-chat-456")
from praisonai.bots._telegram_approval import TelegramApproval
backend = TelegramApproval(token="t")
assert backend._chat_id == "env-chat-456"
def test_ssl_verify_default_true(self, monkeypatch):
monkeypatch.delenv("PRAISONAI_TELEGRAM_SSL_VERIFY", raising=False)
from praisonai.bots._telegram_approval import TelegramApproval
backend = TelegramApproval(token="t", chat_id="1")
assert backend._ssl_verify is True
def test_ssl_verify_env_false(self, monkeypatch):
monkeypatch.setenv("PRAISONAI_TELEGRAM_SSL_VERIFY", "false")
from praisonai.bots._telegram_approval import TelegramApproval
backend = TelegramApproval(token="t", chat_id="1")
assert backend._ssl_verify is False
# ── Message Builder ─────────────────────────────────────────────────────────
@@ -220,6 +220,28 @@ class TestApprovalFlowAsync:
assert posted_payloads[0]["tool_name"] == "rm_file"
assert "request_id" in posted_payloads[0]
def test_non_json_status_does_not_crash(self):
"""Integer status (e.g. HTTP code) must not crash .lower()."""
from praisonai.bots._webhook_approval import WebhookApproval
backend = WebhookApproval(webhook_url="https://example.com", timeout=5, poll_interval=0.1)
poll_count = {"n": 0}
async def mock_http(method, url, payload=None, **kwargs):
if method == "POST":
return {"status": "pending"}
if method == "GET":
poll_count["n"] += 1
if poll_count["n"] <= 1:
# Simulate non-JSON response where status is an int
return {"status": 502}
return {"approved": True, "reason": "ok"}
return {}
backend._http_request = mock_http
decision = asyncio.run(backend.request_approval(self._make_request()))
assert decision.approved is True
# ── Export / Import ─────────────────────────────────────────────────────────