diff --git a/.githooks/pre-commit b/.githooks/pre-commit new file mode 100755 index 0000000..0b23594 --- /dev/null +++ b/.githooks/pre-commit @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +# Run tests before every commit. Blocks the commit if any test fails. +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +# shellcheck source=run-tests.sh +source "$SCRIPT_DIR/run-tests.sh" commit diff --git a/.githooks/pre-push b/.githooks/pre-push new file mode 100755 index 0000000..0355767 --- /dev/null +++ b/.githooks/pre-push @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +# Run tests before every push. Last safety net before hitting the remote. +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +# shellcheck source=run-tests.sh +source "$SCRIPT_DIR/run-tests.sh" push diff --git a/.githooks/run-tests.sh b/.githooks/run-tests.sh new file mode 100755 index 0000000..c53ce0e --- /dev/null +++ b/.githooks/run-tests.sh @@ -0,0 +1,47 @@ +#!/usr/bin/env bash +# Shared test runner sourced by pre-commit and pre-push hooks. +# Usage: source run-tests.sh (context = "commit" | "push") +# +# NOTE: uses `return` not `exit` โ€” this script is sourced, not executed. +# `exit` in a sourced script terminates the parent shell; `return` only +# exits the script's scope, leaving the caller's shell intact. +set -euo pipefail + +CONTEXT="${1:-commit}" + +# Use the virtualenv Python if active, otherwise fall back to system python3. +# Check python3 first inside the venv (most venvs only create python3, not python). +if [[ -n "${VIRTUAL_ENV:-}" ]]; then + if [[ -x "$VIRTUAL_ENV/bin/python3" ]]; then + PYTHON="$VIRTUAL_ENV/bin/python3" + elif [[ -x "$VIRTUAL_ENV/bin/python" ]]; then + PYTHON="$VIRTUAL_ENV/bin/python" + else + PYTHON="$(command -v python3 2>/dev/null)" + fi +else + PYTHON="$(command -v python3 2>/dev/null)" +fi + +if [[ -z "$PYTHON" ]]; then + echo "โŒ node9: python3 not found on PATH. Install Python 3.10+ and try again." >&2 + return 1 +fi + +# Sanity-check: require Python 3.10+ (matches pyproject.toml requires-python) +PYVER=$("$PYTHON" -c 'import sys; print(sys.version_info >= (3,10))' 2>/dev/null) +if [[ "$PYVER" != "True" ]]; then + echo "โŒ node9: Python 3.10+ required (found: $("$PYTHON" --version 2>&1))" >&2 + return 1 +fi + +echo "๐Ÿงช node9: running tests before ${CONTEXT}..." + +if ! "$PYTHON" -m pytest tests/ -p no:anyio -q --tb=short; then + echo "" + echo "โŒ Tests failed โ€” ${CONTEXT} blocked. Fix the failures above and try again." + echo " To skip (unsafe): git ${CONTEXT} --no-verify" + return 1 +fi + +echo "โœ… All tests passed." diff --git a/CHANGELOG.md b/CHANGELOG.md index c92e54f..c56ce26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,38 @@ +## v2.0.0 (2026-04-07) + +### Breaking Changes + +- **`Node9Agent`** base class introduced โ€” framework-agnostic governed agent with `@tool`, `@internal`, `dispatch()`, `build_tools_anthropic()`, `build_tools_openai()` +- **`safe_path(filename, workspace=...)`** โ€” `workspace` is now keyword-only (prevents silent arg swap) +- **`configure()`** โ€” replaces direct env-var mutation; thread-safe via `threading.RLock` + +### New Features + +- `Node9Agent`: zero-dependency governed agent base class with DLP, path safety, and audit built-in +- `@tool` decorator: DLP scan + path traversal check + `evaluate()` on every call +- `@internal` decorator: infrastructure methods (no governance); warns if applied to a public method +- `dispatch()`: LLM-safe router โ€” always returns `str`, handles async tools, unknown tools return descriptive error +- `build_tools_anthropic()` / `build_tools_openai()`: auto-generate tool specs from type annotations +- `new_session()`: fresh `run_id` for server/multi-session deployments +- Offline mode warns loudly when `policy=require_approval` but no daemon/API key is available +- `NODE9_SKIP=1` emits `warnings.warn()` at import time AND per `evaluate()` call +- All SDK status output moved to stderr (stdout stays clean for LLM tool parsers) + +### Migration from 1.x + +```python +# Before (1.x) โ€” positional workspace arg +safe_path(filename, workspace_dir) + +# After (2.0) โ€” keyword-only +safe_path(filename, workspace=workspace_dir) +``` + +`@protect` and `configure()` are fully backwards-compatible. Only `safe_path` call sites need updating. + ## v1.0.0 (2026-04-04) - Initial Release diff --git a/README.md b/README.md index b3d6b71..0925973 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,10 @@ # node9-python -Execution security for Python AI agents, one decorator, zero config. +Execution security for Python AI agents โ€” audit, policy enforcement, and DLP in one package. One decorator, zero config. -Works with any framework: plain Python, LangChain, CrewAI, LangGraph, or custom agents. +Works two ways: +- **`@protect`** โ€” add governance to any existing agent (LangChain, CrewAI, AutoGen, plain Python) +- **`Node9Agent`** โ€” build a governed agent from scratch with tools, DLP, and audit built-in ## Install @@ -10,15 +12,23 @@ Works with any framework: plain Python, LangChain, CrewAI, LangGraph, or custom pip install node9 ``` -## Quick Start +## Routing -**1. Start the Node9 daemon** (ships with `@node9/proxy`): +node9 automatically routes to the right backend: -```bash -npx @node9/proxy daemon -``` +| Environment | Routing | +|---|---| +| `NODE9_API_KEY` set | โ†’ node9 SaaS (cloud / CI โ€” no local daemon needed) | +| Local daemon running | โ†’ node9-proxy on `localhost:7391` | +| Neither | โ†’ offline audit log at `~/.node9/audit.log` (auto-approve, never blocks) | + +No config required โ€” it just works wherever your agent runs. + +--- + +## Option 1 โ€” `@protect`: Add governance to any agent -**2. Add `@protect` to any function your agent calls:** +Drop `@protect` on any function your agent calls. node9 intercepts the call, logs it, and enforces policy before the function runs. ```python from node9 import protect, ActionDeniedException @@ -28,84 +38,47 @@ def write_file(path: str, content: str) -> None: with open(path, "w") as f: f.write(content) -@protect("bash") -def run_shell(command: str) -> str: +_ALLOWED_COMMANDS = {"pytest", "ruff", "mypy", "black"} + +@protect("run_tests") +def run_tests(tool: str) -> str: + # Allowlist-based: only pre-approved CLI tools can be invoked. + # Never pass raw LLM strings to subprocess โ€” enumerate safe commands explicitly. + if tool not in _ALLOWED_COMMANDS: + raise ValueError(f"Tool {tool!r} is not in the allowed list: {_ALLOWED_COMMANDS}") import subprocess - return subprocess.check_output(command, shell=True, text=True) + return subprocess.check_output([tool], text=True) -# When your agent calls this, Node9 intercepts it and asks for approval. -# The call blocks until a human approves or denies โ€” in the dashboard or Slack. try: - write_file("/etc/hosts", "malicious content") + write_file("/etc/hosts", "bad content") except ActionDeniedException as e: print(f"Blocked: {e}") ``` -That's it. All function arguments are captured automatically โ€” no config needed. - -## How It Works - -``` -Agent calls write_file() - โ†“ - @protect intercepts - โ†“ - POST /check โ†’ Node9 daemon (localhost:7391) - โ†“ - Daemon shows approval popup / sends Slack message - โ†“ - Human approves or denies - โ†“ - Function runs (or ActionDeniedException is raised) -``` - -## Async Support +Works with `async def` out of the box. -`@protect` works with `async def` out of the box. The blocking HTTP call runs in a thread so it never freezes your event loop: +### Set agent identity (optional but recommended) ```python -@protect("write_file") -async def write_file(path: str, content: str) -> str: - async with aiofiles.open(path, "w") as f: - await f.write(content) - return f"Written to {path}" -``` +from node9 import configure -This makes it compatible with LangGraph, FastMCP, and any other async agent framework. - -## Custom Tool Name - -By default, the tool name sent to Node9 is the function name. Override it: - -```python -@protect("postgres_query") -def execute_sql(sql: str, db: str = "prod") -> list: - ... +configure(agent_name="my-langchain-agent", policy="audit") ``` -## Custom Params - -Control exactly what gets sent to the approval UI: - -```python -@protect("deploy", params=lambda service, env="prod", **_: {"service": service, "env": env}) -def deploy(service: str, env: str = "prod", dry_run: bool = False) -> str: - ... +Or via environment variables: +```bash +NODE9_AGENT_NAME=my-langchain-agent +NODE9_AGENT_POLICY=audit ``` -## Handling Denials in LLM Feedback Loops - -`ActionDeniedException` has a `negotiation` property โ€” a ready-made string you can feed back to the LLM so it can try a different approach instead of crashing: - -```python -try: - delete_file("/etc/hosts") -except ActionDeniedException as e: - # e.negotiation = "Action 'delete_file' was blocked by Node9: Too dangerous. Choose a different approach." - response = llm.invoke(e.negotiation) -``` +### Policy values -## Framework Examples +| Policy | Behaviour | +|---|---| +| `audit` | Log everything, auto-approve. Never blocks. Good for CI. | +| `require_approval` | Block + notify human. Good for production actions. | +| `block_on_rules` | Auto-block if rules match, audit otherwise. | +| _(empty)_ | SaaS default behaviour. | ### LangChain @@ -139,17 +112,150 @@ def write_file(path: str, content: str) -> str: return f"Written to {path}" ``` -See [`examples/`](examples/) for full runnable examples. +See [`examples/`](examples/) for full runnable examples including AutoGen and LangGraph. + +--- + +## Option 2 โ€” `Node9Agent`: Build a governed agent from scratch + +`Node9Agent` is a governance base class โ€” DLP, path safety, audit, and tool dispatch built-in. It does **not** include an LLM loop; that is your framework's responsibility. This keeps the SDK framework-agnostic with zero dependencies. -## Environment Variables +```python +import anthropic +from node9 import Node9Agent, tool, internal + +class CiAgent(Node9Agent): + agent_name = "ci-code-review" + policy = "audit" + + _ALLOWED_SUITES = {"pytest", "pytest --tb=short", "ruff check ."} + + @tool("run_tests") + def run_tests(self, suite: str) -> str: + """Run an allowlisted test suite and return output.""" + import shlex, subprocess + if suite not in self._ALLOWED_SUITES: + raise ValueError(f"Suite {suite!r} not in allowed list: {self._ALLOWED_SUITES}") + return subprocess.check_output(shlex.split(suite), text=True) + + @tool("write_code") + def write_code(self, filename: str, content: str) -> str: + """Write content to a file in the workspace.""" + from node9 import safe_path + path = safe_path(filename, workspace=self._workspace) # traversal-safe + with open(path, "w") as f: + f.write(content) + return f"Written {filename}" + + @internal + def _git_push(self, branch: str) -> str: + """Push to remote โ€” infrastructure, not a governed action.""" + import subprocess + subprocess.run(["git", "push", "origin", branch], check=True) + return f"Pushed {branch}" + +agent = CiAgent(workspace="/path/to/repo") +client = anthropic.Anthropic() + +# Get tool specs in the format your LLM expects +tools = agent.build_tools_anthropic() # โ†’ input_schema format +# tools = agent.build_tools_openai() # โ†’ {type: function, function: {...}} +# tools = agent._build_tools() # โ†’ neutral (parameters key) + +# Your LLM loop โ€” use whichever client you want +messages = [{"role": "user", "content": "Fix the failing tests in this diff: ..."}] +while True: + response = client.messages.create(model="claude-opus-4-6", tools=tools, messages=messages) + messages.append({"role": "assistant", "content": response.content}) + if response.stop_reason != "tool_use": + break + results = [] + for block in response.content: + if block.type == "tool_use": + result = agent.dispatch(block.name, block.input) # DLP + audit happen here + results.append({"type": "tool_result", "tool_use_id": block.id, "content": result}) + messages.append({"role": "user", "content": results}) +``` + +See [`examples/`](examples/) for complete runnable implementations per framework. + +### What `@tool` does automatically + +Every `@tool`-decorated method, before the function runs: +1. **DLP scan** โ€” blocks if `filename` or `content` contains a secret or sensitive path +2. **Path safety** โ€” rejects `../` traversal attempts, raises `ActionDeniedException` +3. **Audit / approval** โ€” calls `evaluate()` which respects the agent's `policy` +4. **Run ID** โ€” injects a UUID grouping all tool calls from one session in the dashboard + +### What `@internal` does + +`@internal` is for git operations, workspace setup, and other infrastructure: +- Never calls `evaluate()` โ€” no SaaS call, no blocking +- Logs locally only: `[node9 internal] _git_push(branch='main')` + +### Tool specs are auto-generated + +`Node9Agent` introspects `@tool` methods and builds tool specs automatically โ€” parameter names, types from annotations, and descriptions from docstrings. No manual schema writing. + +--- + +## DLP and path safety as standalone utilities + +```python +from node9 import dlp_scan, safe_path + +# Scan content for secrets before writing to disk +hit = dlp_scan("output.txt", content) +if hit: + raise ValueError(f"Blocked: {hit}") + +# Resolve a path safely within a workspace directory +path = safe_path("src/main.py", workspace="/tmp/repo") +``` + +Patterns detected: AWS keys, GitHub tokens, Slack tokens, OpenAI keys, Stripe keys, PEM private keys, GCP service accounts, NPM auth tokens, Anthropic keys, and sensitive file paths (`.ssh`, `.aws`, `.env`, `.kube`, etc.). + +--- + +## Handling denials in LLM feedback loops + +`ActionDeniedException` has a `negotiation` property โ€” feed it back to the LLM so it can try a different approach: + +```python +try: + agent.dispatch("delete_file", {"path": "/etc/hosts"}) +except ActionDeniedException as e: + # e.negotiation = "Action 'delete_file' was blocked by Node9: policy. Choose a different approach." + response = llm.invoke(e.negotiation) +``` + +--- + +## Environment variables | Variable | Default | Description | |---|---|---| -| `NODE9_DAEMON_PORT` | `7391` | Daemon port | -| `NODE9_AUTO_START` | โ€” | Set to `1` to auto-launch the daemon if it's not running | -| `NODE9_SKIP` | โ€” | Set to `1` to bypass all checks (unsafe โ€” for tests only) | +| `NODE9_API_KEY` | โ€” | Routes to node9 SaaS. Required for cloud / CI. | +| `NODE9_AGENT_NAME` | โ€” | Agent identity โ€” appears in audit logs and dashboard. | +| `NODE9_AGENT_POLICY` | โ€” | `audit`, `require_approval`, or `block_on_rules`. | +| `NODE9_DAEMON_PORT` | `7391` | Local daemon port. | +| `NODE9_AUTO_START` | โ€” | Set to `1` to auto-launch the local daemon if not running. | +| `NODE9_SKIP` | โ€” | Set to `1` to bypass all checks. **Never set in production** โ€” disables all governance. For unit tests only. If set, a warning is emitted at import time. | + +## Development + +After cloning, activate the git hooks (runs tests before every commit and push): + +```bash +git config core.hooksPath .githooks +``` + +Run tests manually: + +```bash +python3 -m pytest tests/ -p no:anyio -q +``` ## License Apache-2.0 - diff --git a/examples/basic.py b/examples/basic.py index 8a309b4..1a3b940 100644 --- a/examples/basic.py +++ b/examples/basic.py @@ -27,10 +27,15 @@ def delete_file(path: str) -> None: print(f"Deleted: {path}") +_ALLOWED_COMMANDS = {"ls", "pwd", "git status", "git log --oneline"} + @protect("bash") def run_shell(command: str) -> str: - import subprocess - return subprocess.check_output(command, shell=True, text=True) + import shlex, subprocess + # Allowlist-only: never pass LLM-controlled strings to shell=True. + if command not in _ALLOWED_COMMANDS: + raise ValueError(f"Command {command!r} not in allowed list") + return subprocess.check_output(shlex.split(command), text=True) # --- Custom tool name + params lambda --- diff --git a/examples/crewai_agent.py b/examples/crewai_agent.py index f5f52ae..c8d59fd 100644 --- a/examples/crewai_agent.py +++ b/examples/crewai_agent.py @@ -24,12 +24,17 @@ def write_file(path: str, content: str) -> str: return f"Written to {path}" +_ALLOWED_COMMANDS = {"pytest", "ruff check .", "mypy src/"} + @tool("run_shell") @protect("bash") def run_shell(command: str) -> str: - """Execute a shell command.""" - import subprocess - return subprocess.check_output(command, shell=True, text=True) + """Execute an allowlisted shell command.""" + import shlex, subprocess + # Allowlist-only: never pass LLM-controlled strings to shell=True. + if command not in _ALLOWED_COMMANDS: + raise ValueError(f"Command {command!r} not in allowed list") + return subprocess.check_output(shlex.split(command), text=True) @tool("deploy_service") diff --git a/examples/langchain_agent.py b/examples/langchain_agent.py index ad56f28..d462190 100644 --- a/examples/langchain_agent.py +++ b/examples/langchain_agent.py @@ -32,10 +32,15 @@ class RunShellTool(BaseTool): name: str = "bash" description: str = "Run a shell command and return its output." + _ALLOWED_COMMANDS = {"pytest", "ruff check .", "mypy ."} + @protect("bash") def _run(self, command: str) -> str: - import subprocess - return subprocess.check_output(command, shell=True, text=True) + import shlex, subprocess + # Allowlist-only: never pass LLM-controlled strings to shell=True. + if command not in self._ALLOWED_COMMANDS: + raise ValueError(f"Command {command!r} not in allowed list") + return subprocess.check_output(shlex.split(command), text=True) class DeleteFileTool(BaseTool): diff --git a/examples/manual_test.py b/examples/manual_test.py new file mode 100644 index 0000000..0276c0e --- /dev/null +++ b/examples/manual_test.py @@ -0,0 +1,129 @@ +""" +Manual smoke test for node9 SDK. +Run in different modes to test all three routing paths. + +Usage: + # 1. Offline mode (no daemon, no API key) + python3 manual_test.py + + # 2. Local daemon mode (start daemon first: npx @node9/proxy daemon) + python3 manual_test.py --daemon + + # 3. Cloud mode + NODE9_API_KEY=sk-... python3 manual_test.py +""" +import os +import sys +import tempfile + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from node9 import protect, configure, Node9Agent, tool, internal, dlp_scan, safe_path, ActionDeniedException + +print("\n=== node9 SDK manual test ===\n") + +# โ”€โ”€ 1. Routing info โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +if os.environ.get("NODE9_API_KEY"): + print("Mode: CLOUD (NODE9_API_KEY set)") +elif "--daemon" in sys.argv: + print("Mode: LOCAL DAEMON") +else: + print("Mode: OFFLINE (no daemon, no API key)") +print() + +# โ”€โ”€ 2. configure() โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +print("--- configure() ---") +configure(agent_name="manual-test", policy="audit") +import node9._config as cfg +print(f" agent_name = {cfg.AGENT_NAME!r}") +print(f" policy = {cfg.AGENT_POLICY!r}") +print() + +# โ”€โ”€ 3. @protect basic call โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +print("--- @protect ---") + +@protect("write_file") +def write_file(path: str, content: str) -> str: + return f"written:{path}" + +result = write_file("/tmp/test.txt", "hello") +print(f" write_file result: {result}") +print() + +# โ”€โ”€ 4. DLP scan โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +print("--- dlp_scan ---") +clean = dlp_scan("output.txt", "def hello(): pass") +print(f" clean content: {clean!r} (expect None)") + +sensitive_path = dlp_scan("/home/user/.ssh/id_rsa", "content") +print(f" sensitive path: {sensitive_path!r} (expect block reason)") +print() + +# โ”€โ”€ 5. safe_path โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +print("--- safe_path ---") +with tempfile.TemporaryDirectory() as workspace: + resolved = safe_path("src/main.py", workspace=workspace) + print(f" safe_path resolved: {resolved}") + try: + safe_path("../../etc/passwd", workspace=workspace) + print(" traversal: NOT blocked (BUG)") + except ValueError as e: + print(f" traversal blocked: {e}") +print() + +# โ”€โ”€ 6. Node9Agent โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +print("--- Node9Agent ---") + +with tempfile.TemporaryDirectory() as workspace: + class TestAgent(Node9Agent): + agent_name = "manual-test-agent" + policy = "audit" + + @tool("echo") + def echo(self, message: str) -> str: + """Echo the message back.""" + return f"echo:{message}" + + @tool("write_file") + def write_file(self, filename: str, content: str) -> str: + """Write content to a file.""" + import pathlib + (pathlib.Path(self._workspace) / filename).write_text(content) + return f"written:{filename}" + + @internal + def _setup(self, branch: str) -> str: + return f"setup:{branch}" + + agent = TestAgent(workspace=workspace) + print(f" run_id: {agent._run_id}") + print(f" workspace: {agent._workspace}") + + # @tool call + result = agent.echo("hello node9") + print(f" @tool echo: {result}") + + # @internal call (no evaluate) + result = agent._setup("main") + print(f" @internal _setup: {result}") + + # DLP block via @tool + try: + agent.write_file("/home/user/.ssh/id_rsa", "content") + print(" DLP: NOT blocked (BUG)") + except ActionDeniedException as e: + print(f" DLP blocked: {e.tool_name} โ€” {e.reason[:50]}") + + # Path traversal block via @tool + try: + agent.write_file("../../etc/passwd", "content") + print(" Traversal: NOT blocked (BUG)") + except ActionDeniedException as e: + print(f" Traversal blocked: {e.tool_name} โ€” {e.reason[:50]}") + + # _build_tools + tools = agent._build_tools() + print(f" _build_tools: {[t['name'] for t in tools]}") + +print() +print("=== all checks passed ===\n") diff --git a/node9/__init__.py b/node9/__init__.py index e1f7dc7..e872f0d 100644 --- a/node9/__init__.py +++ b/node9/__init__.py @@ -1,10 +1,51 @@ """ node9 โ€” Execution security for Python AI agents. -Bundled version with CI cloud routing support (NODE9_API_KEY). """ from ._decorator import protect from ._exceptions import ActionDeniedException, DaemonNotFoundError +from ._dlp import dlp_scan, safe_path +from ._agent import Node9Agent, tool, internal +from . import _config -__all__ = ["protect", "ActionDeniedException", "DaemonNotFoundError"] -__version__ = "0.1.1" + +def configure(*, agent_name: str = "", policy: str = "") -> None: + """ + Set agent identity at runtime. Alternative to NODE9_AGENT_NAME / NODE9_AGENT_POLICY env vars. + Call before the first evaluate() / @protect / agent.dispatch(). + Thread-safe โ€” safe to call from concurrent async frameworks (LangGraph, FastMCP). + + policy values: "audit" | "require_approval" | "block_on_rules" | "" (SaaS default) + + Configuration precedence (highest wins): + 1. configure() called at runtime โ€” highest priority + 2. Node9Agent.agent_name / .policy class attributes + 3. NODE9_AGENT_NAME / NODE9_AGENT_POLICY env vars โ€” baseline at import time + + Calling configure() after the first tool call is allowed but not recommended โ€” + in-flight calls will have already used the previous identity. + """ + _config.set_identity(agent_name=agent_name, policy=policy) + + +__all__ = [ + # Core + "protect", + "configure", + # Agent framework + "Node9Agent", # base class โ€” subclass and use @tool / @internal + "tool", # decorator: governed tool (DLP + audit + policy) + "internal", # decorator: infrastructure method (no governance) + # Node9Agent methods (documented here for IDE discoverability) + # .build_tools_anthropic() โ€” Anthropic input_schema format + # .build_tools_openai() โ€” OpenAI function format + # .dispatch(name, input) โ€” route LLM tool call to @tool method + # .new_session() โ€” fresh run_id for server/multi-session deployments + # DLP utilities + "dlp_scan", + "safe_path", + # Exceptions + "ActionDeniedException", + "DaemonNotFoundError", +] +__version__ = "2.0.0" diff --git a/node9/_agent.py b/node9/_agent.py new file mode 100644 index 0000000..491567f --- /dev/null +++ b/node9/_agent.py @@ -0,0 +1,374 @@ +""" +Node9Agent โ€” governance base class for AI agents. + +Provides @tool and @internal decorators plus the Node9Agent base class. +Does NOT include an LLM loop โ€” that is the framework's responsibility. + +The LLM loop lives in the agent that subclasses Node9Agent. This keeps the +SDK framework-agnostic and dependency-free (zero imports beyond stdlib). + +Usage: + from node9 import Node9Agent, tool, internal + + class CiAgent(Node9Agent): + agent_name = "ci-code-review" + policy = "audit" + + _ALLOWED_SUITES = {"pytest", "pytest --tb=short", "ruff check ."} + + @tool("run_tests") + def run_tests(self, suite: str) -> str: + import shlex, subprocess + # Always allowlist LLM-controlled commands โ€” never pass raw strings to subprocess. + if suite not in self._ALLOWED_SUITES: + raise ValueError(f"Suite {suite!r} not in allowed list") + return subprocess.check_output(shlex.split(suite), text=True) + + @tool("write_code") + def write_code(self, filename: str, content: str) -> str: + from node9 import safe_path + path = safe_path(filename, workspace=self._workspace) # workspace-relative, traversal-safe + with open(path, "w") as f: + f.write(content) + return f"written:{filename}" + + @internal + def _git_push(self, branch: str) -> str: + # never calls evaluate() โ€” infrastructure only + import subprocess + subprocess.run(["git", "push", "origin", branch], check=True) + return f"pushed:{branch}" + + agent = CiAgent(workspace="/path/to/repo") + + # The LLM loop is YOUR code โ€” use whichever framework you want: + # Anthropic: tools = agent.build_tools_anthropic() + # OpenAI: tools = agent.build_tools_openai() + # Custom: tools = agent._build_tools() # neutral format + # + # Dispatch tool calls from the LLM response: + # result = agent.dispatch(tool_name, tool_input) +""" + +import asyncio +import functools +import inspect +import os +import sys +import uuid +import warnings +from typing import Any, Callable, Union + +from ._client import evaluate +from ._dlp import dlp_scan, safe_path +from ._exceptions import ActionDeniedException + +# Marker attributes written by decorators so Node9Agent can introspect methods +_TOOL_ATTR = "_node9_tool" +_INTERNAL_ATTR = "_node9_internal" + + +def tool(tool_name: Union[str, Callable]): + """ + Marks a Node9Agent method as a governed tool. + + Before every call: + - DLP scan โ€” blocks if filename/content contains a secret or sensitive path + - Path safety โ€” rejects ../traversal attempts + - evaluate() โ€” respects the agent's declared policy (audit / require_approval / etc.) + - run_id โ€” injected automatically so all calls in one run are grouped in the dashboard + + Can be used as @tool or @tool("custom_name"). + + Note: DLP scanning only inspects top-level string arguments. Secrets nested + inside dicts or lists will not be caught โ€” flatten sensitive values or call + dlp_scan() explicitly before passing structured data. + """ + def decorator(fn: Callable, name: str) -> Callable: + @functools.wraps(fn) + def wrapper(self: "Node9Agent", *args: Any, **kwargs: Any) -> Any: + sig = inspect.signature(fn) + bound = sig.bind(self, *args, **kwargs) + bound.apply_defaults() + call_args = {k: v for k, v in bound.arguments.items() if k != "self"} + + # DLP scan โ€” run once per string arg as both path and content candidate + # This catches sensitive paths regardless of parameter name (e.g. dest, target) + string_args = [str(v) for v in call_args.values() if isinstance(v, str)] + all_content = "\n".join(string_args) + for candidate in string_args: + hit = dlp_scan(candidate, all_content) + if hit: + raise ActionDeniedException(name, f"DLP blocked: {hit}") + + # Path safety โ€” check ALL string args that look like file paths. + # We check every arg regardless of parameter name so that parameters + # named dest, output, filepath, etc. are protected the same as filename. + if hasattr(self, "_workspace") and self._workspace: + for v in call_args.values(): + if isinstance(v, str) and ("/" in v or "\\" in v): + try: + safe_path(v, workspace=self._workspace) + except ValueError as e: + raise ActionDeniedException(name, str(e)) from e + + run_id = getattr(self, "_run_id", "") + evaluate(name, call_args, run_id=run_id) + return fn(self, *args, **kwargs) + + setattr(wrapper, _TOOL_ATTR, name) + return wrapper + + if callable(tool_name): + fn = tool_name + return decorator(fn, fn.__name__) + + def outer(fn: Callable) -> Callable: + return decorator(fn, tool_name) + + return outer + + +def internal(fn: Callable) -> Callable: + """ + Marks a Node9Agent method as infrastructure (git plumbing, workspace setup). + + - Never calls evaluate() โ€” no SaaS call, no blocking, no DLP scan + - Logs to stderr: [node9 internal] method_name(args) + - Use only for non-agent-decision code (git, workspace setup, file plumbing). + Do NOT use to bypass governance on agent-controlled actions. + - NOT reachable via dispatch() โ€” the LLM cannot invoke @internal methods + through the tool-call interface. dispatch() only routes to @tool methods. + + WARNING: @internal skips all governance. By convention, @internal methods + should have names starting with '_' to make the bypass visible at call sites. + A RuntimeWarning is raised if a public method name is decorated with @internal. + """ + if not fn.__name__.startswith("_"): + warnings.warn( + f"@internal applied to public method '{fn.__name__}' โ€” @internal skips all " + "governance checks. Rename to '_{fn.__name__}' or use @tool instead.", + RuntimeWarning, + stacklevel=2, + ) + + @functools.wraps(fn) + def wrapper(self: "Node9Agent", *args: Any, **kwargs: Any) -> Any: + sig = inspect.signature(fn) + bound = sig.bind(self, *args, **kwargs) + bound.apply_defaults() + call_args = {k: v for k, v in bound.arguments.items() if k != "self"} + arg_summary = ", ".join(f"{k}={str(v)[:60]!r}" for k, v in call_args.items()) + # Write to stderr, not stdout โ€” LLM frameworks parse stdout for tool results + # and a print() mid-execution would corrupt the JSON/text output stream. + print(f" [node9 internal] {fn.__name__}({arg_summary})", file=sys.stderr, flush=True) + return fn(self, *args, **kwargs) + + setattr(wrapper, _INTERNAL_ATTR, True) + return wrapper + + +class Node9Agent: + """ + Governance base class for AI agents. Framework-agnostic, zero dependencies. + + Provides: + - Agent identity and policy (set once, applied to every tool call) + - Per-run UUID for grouping audit entries in the dashboard + - build_tools_anthropic() โ€” Anthropic input_schema format + - build_tools_openai() โ€” OpenAI parameters format + - dispatch() โ€” route LLM tool calls to @tool methods (primary integration point) + + The LLM loop is NOT here โ€” implement it in your subclass using whichever + framework or API client you need. + + When to use Node9Agent vs @protect: + - Node9Agent: greenfield agents where you control the tool definitions + - @protect: retrofitting governance onto existing functions/classes + """ + + agent_name: str = "" + policy: str = "audit" + + def new_session(self) -> str: + """ + Start a new session โ€” generates a fresh run_id so audit entries are + grouped correctly. Call at the start of each user request in server deployments. + Returns the new run_id. + + Thread safety: do NOT share a single Node9Agent instance across concurrent + requests. _run_id assignment is not atomic. Use one instance per request + (or per thread) to avoid run_id races in concurrent web servers. + """ + self._run_id = str(uuid.uuid4()) + return self._run_id + + def __init__(self, workspace: str = ""): + self._run_id = str(uuid.uuid4()) # one run_id per instance; call new_session() per request + if workspace: + resolved = os.path.realpath(workspace) + if not os.path.isdir(resolved): + raise ValueError( + f"Node9Agent workspace does not exist: {workspace!r}. " + "Create the directory before constructing the agent." + ) + self._workspace = resolved + else: + self._workspace = os.getcwd() + + from . import _config + _config.set_identity( + agent_name=self.agent_name or type(self).__name__, + policy=self.policy, + ) + + # ------------------------------------------------------------------------- + # Tool spec builders โ€” pick the format your LLM expects + # ------------------------------------------------------------------------- + + def _build_tools(self) -> list[dict]: + """ + Neutral tool spec list. Keys: name, description, parameters (JSON Schema). + + Convert for your LLM: + Anthropic: rename 'parameters' โ†’ 'input_schema' + OpenAI: wrap in {"type": "function", "function": spec} + """ + tools = [] + for attr_name in dir(type(self)): + method = getattr(type(self), attr_name, None) + if method is None: + continue + tool_name = getattr(method, _TOOL_ATTR, None) + if tool_name is None: + continue + + original = inspect.unwrap(method) + sig = inspect.signature(original) + description = (inspect.getdoc(original) or tool_name).split("\n")[0] + + properties: dict[str, Any] = {} + required: list[str] = [] + + for param_name, param in sig.parameters.items(): + if param_name == "self": + continue + if param.kind in ( + inspect.Parameter.VAR_POSITIONAL, + inspect.Parameter.VAR_KEYWORD, + ): + continue + prop: dict[str, Any] = {"type": "string"} + ann = param.annotation + if ann is not inspect.Parameter.empty: + if ann is int: prop["type"] = "integer" + elif ann is float: prop["type"] = "number" + elif ann is bool: prop["type"] = "boolean" + properties[param_name] = prop + if param.default is inspect.Parameter.empty: + required.append(param_name) + + tools.append({ + "name": tool_name, + "description": description, + "parameters": { + "type": "object", + "properties": properties, + "required": required, + }, + }) + return tools + + def build_tools_anthropic(self) -> list[dict]: + """Tool specs in Anthropic format (input_schema key).""" + result = [] + for spec in self._build_tools(): + result.append({ + "name": spec["name"], + "description": spec["description"], + "input_schema": spec["parameters"], + }) + return result + + def build_tools_openai(self) -> list[dict]: + """Tool specs in OpenAI format (type=function wrapper).""" + result = [] + for spec in self._build_tools(): + result.append({ + "type": "function", + "function": { + "name": spec["name"], + "description": spec["description"], + "parameters": spec["parameters"], + }, + }) + return result + + # ------------------------------------------------------------------------- + # Dispatch + # ------------------------------------------------------------------------- + + def dispatch(self, tool_name: str, tool_input: dict) -> str: + """ + Route a tool call by name to the matching @tool method. + + Always returns a str โ€” safe to pass directly as a tool_result to any LLM API. + Never raises: exceptions from the tool are caught and returned as "Error: ..." strings. + + ActionDeniedException โ†’ returns e.negotiation (denial reason for the LLM) + Any other exception โ†’ returns "Error: " + Unknown tool name โ†’ returns "Unknown tool: '...'. Available tools: [...]" + Tool returns None โ†’ returns "" + + This is the primary integration point for LLM loops: + result = agent.dispatch(block.name, block.input) # Anthropic + result = agent.dispatch(call.function.name, json.loads(call.function.arguments)) # OpenAI + + Lookup is strictly registry-based: only @tool-decorated methods are reachable. + @internal methods and plain instance methods are never callable via dispatch(). + """ + # Lookup is strictly against the @tool decorator registry (_TOOL_ATTR marker). + # Only methods explicitly decorated with @tool are callable via dispatch(). + for attr_name in dir(type(self)): + method = getattr(type(self), attr_name, None) + if method is None: + continue + if getattr(method, _TOOL_ATTR, None) == tool_name: + try: + result = getattr(self, attr_name)(**tool_input) + if inspect.iscoroutine(result): + try: + asyncio.get_running_loop() + # Already inside an async event loop โ€” dispatch() cannot + # await here. The caller should await the method directly: + # result = await agent.method(**tool_input) + result.close() # prevent "coroutine was never awaited" warning + return ( + f"Error: '{tool_name}' is async. " + "In an async context, call it directly with 'await'." + ) + except RuntimeError: + # RuntimeError here means "no running event loop" โ€” + # the only error get_running_loop() raises. Safe to run. + result = asyncio.run(result) + return str(result) if result is not None else "" + except ActionDeniedException as e: + return e.negotiation + except Exception as e: + return f"Error: {e}" + available = sorted( + getattr(m, _TOOL_ATTR) + for attr in dir(type(self)) + if (m := getattr(type(self), attr, None)) is not None + and getattr(m, _TOOL_ATTR, None) is not None + ) + return f"Unknown tool: {tool_name!r}. Available tools: {available}" + + def _dispatch(self, tool_name: str, tool_input: dict) -> str: + """Deprecated alias for dispatch(). Use dispatch() instead.""" + warnings.warn( + "Node9Agent._dispatch() is deprecated โ€” use .dispatch() instead.", + DeprecationWarning, + stacklevel=2, + ) + return self.dispatch(tool_name, tool_input) diff --git a/node9/_client.py b/node9/_client.py index 3ad8838..92f7dc6 100644 --- a/node9/_client.py +++ b/node9/_client.py @@ -1,9 +1,10 @@ """ -Thin HTTP client โ€” talks to the local Node9 daemon on localhost:7391. +Thin HTTP client โ€” talks to either the local Node9 daemon or node9 SaaS. -Flow: - POST /check โ†’ { id } registers the action - GET /wait/:id โ†’ { decision, reason? } blocks until approved / denied +Routing: + NODE9_API_KEY set โ†’ node9 SaaS (api.node9.ai) cloud / CI + daemon reachable โ†’ local proxy (localhost:7391) persona 1 / local dev + neither โ†’ offline audit log dev / test, never blocks """ import json @@ -12,12 +13,14 @@ import re import shutil import subprocess +import sys import time import http.client import urllib.error import urllib.request from typing import Any +from . import _config from ._config import DAEMON_PORT from ._exceptions import ActionDeniedException, DaemonNotFoundError @@ -26,6 +29,15 @@ _CHECK_TIMEOUT = 5 # seconds to establish connection _WAIT_TIMEOUT = 65 # seconds to wait for human decision +_SKIP = os.environ.get("NODE9_SKIP") == "1" +if _SKIP: + import warnings as _warnings + _warnings.warn( + "[Node9] NODE9_SKIP=1 is set โ€” all governance checks are disabled. " + "Never set this in production.", + stacklevel=2, + ) + _CI_CONTEXT_MAX_BYTES = 10_000 _CI_CONTEXT_ALLOWED_KEYS = { "tests_after", "files_changed", "issues_found", "issues_fixed", @@ -108,7 +120,48 @@ def _read_ci_context() -> dict | None: return None -def _evaluate_cloud(tool_name: str, args: dict[str, Any]) -> None: +def _offline_audit(tool_name: str, args: dict[str, Any], run_id: str) -> None: + """ + Offline audit mode โ€” no daemon, no SaaS. + Writes a local audit entry and auto-approves. Never blocks. + Used when neither NODE9_API_KEY nor local daemon is available. + """ + import datetime + _, policy = _config.get() + if policy == "require_approval": + import warnings + warnings.warn( + f"[Node9] Governance degraded to offline/auto-approve for '{tool_name}' โ€” " + "policy is 'require_approval' but no daemon or API key is available. " + "Start the node9 daemon or set NODE9_API_KEY to enforce approvals.", + RuntimeWarning, + stacklevel=4, + ) + audit_dir = os.path.join(os.path.expanduser("~"), ".node9") + os.makedirs(audit_dir, exist_ok=True) + audit_path = os.path.join(audit_dir, "audit.log") + entry = { + "ts": datetime.datetime.utcnow().isoformat() + "Z", + "mode": "offline", + "agent": (_config.get()[0] or "Python SDK"), + "policy": (_config.get()[1] or "offline"), + "runId": run_id, + "toolName": tool_name, + "args": args, + "decision": "allow", + } + try: + with open(audit_path, "a") as f: + f.write(json.dumps(entry, default=str) + "\n") + except OSError as e: + # Audit write failed (read-only fs, container, permissions). + # Never crash the agent, but surface the failure so it's not silent. + print(f" [node9 offline] WARNING: audit write failed ({e})", file=sys.stderr, flush=True) + else: + print(f" [node9 offline] {tool_name} โ€” logged to {audit_path}", file=sys.stderr, flush=True) + + +def _evaluate_cloud(tool_name: str, args: dict[str, Any], run_id: str = "") -> None: """ Cloud routing: POST directly to node9 SaaS when NODE9_API_KEY is set. Used in CI environments where the local daemon is not running. @@ -119,16 +172,20 @@ def _evaluate_cloud(tool_name: str, args: dict[str, Any]) -> None: api_url = os.environ.get("NODE9_API_URL", "https://api.node9.ai/api/v1/intercept").rstrip("/") - if not api_url.startswith("https://"): + if not api_url.startswith("https://") and not api_url.startswith("http://localhost"): raise RuntimeError( f"[Node9] NODE9_API_URL must use HTTPS to protect credentials (got: {api_url!r})" ) + _agent_name, _agent_policy = _config.get() payload: dict = { "toolName": tool_name, "args": args, + "agentName": _agent_name or "Python SDK", + "policy": _agent_policy, + "runId": run_id, "context": { - "agent": "Python SDK", + "agent": _agent_name or "Python SDK", "hostname": platform.node(), "platform": platform.system().lower(), "cwd": os.getcwd(), @@ -177,7 +234,7 @@ def _evaluate_cloud(tool_name: str, args: dict[str, Any]) -> None: if not _REQUEST_ID_RE.match(str(request_id)): raise RuntimeError(f"[Node9] Invalid requestId format: {request_id!r}") - print(f"๐Ÿ›ก๏ธ Node9: waiting for approval of '{tool_name}'...", flush=True) + print(f"๐Ÿ›ก๏ธ Node9: waiting for approval of '{tool_name}'...", file=sys.stderr, flush=True) poll_timeout = max(30, min(3600, int(os.environ.get("NODE9_CLOUD_TIMEOUT", "600")))) status_url = f"{api_url}/status/{request_id}" @@ -212,30 +269,50 @@ def _evaluate_cloud(tool_name: str, args: dict[str, Any]) -> None: raise ActionDeniedException(tool_name, f"Cloud approval timed out after {poll_timeout}s.") -def evaluate(tool_name: str, args: dict[str, Any]) -> None: +def evaluate(tool_name: str, args: dict[str, Any], *, run_id: str = "") -> None: """ - Sends the action to the daemon and blocks until a decision is made. + Sends the action to node9 for audit / approval. Routing: + NODE9_SKIP=1 โ†’ no-op (unsafe bypass for testing) + NODE9_API_KEY set โ†’ node9 SaaS + daemon reachable โ†’ local proxy + neither โ†’ offline audit log (auto-approve, never blocks) + Raises ActionDeniedException if the action is denied. - Does nothing if NODE9_SKIP=1 is set (unsafe bypass for testing). - Set NODE9_AUTO_START=1 to automatically launch the daemon if it's not running. - When NODE9_API_KEY is set, routes directly to node9 SaaS (no local daemon needed). """ - if os.environ.get("NODE9_SKIP") == "1": + if _SKIP: + import warnings + warnings.warn( + f"[Node9] NODE9_SKIP=1 โ€” governance bypassed for '{tool_name}'. " + "Do not use in production.", + stacklevel=3, + ) + _offline_audit(tool_name, {**args, "_skip": True}, run_id=run_id) return if os.environ.get("NODE9_API_KEY"): - _evaluate_cloud(tool_name, args) + _evaluate_cloud(tool_name, args, run_id=run_id) return if os.environ.get("NODE9_AUTO_START") == "1" and not _daemon_reachable(): _auto_start_daemon() - result = _post("/check", {"toolName": tool_name, "args": args, "cwd": os.getcwd(), "agent": "Python SDK"}) + if not _daemon_reachable(): + _offline_audit(tool_name, args, run_id=run_id) + return + + result = _post("/check", { + "toolName": tool_name, + "args": args, + "cwd": os.getcwd(), + "agent": (_config.get()[0] or "Python SDK"), + "policy": _config.get()[1], + "runId": run_id, + }) request_id = result.get("id") if not request_id: raise RuntimeError(f"[Node9] Unexpected daemon response: {result}") - print(f"๐Ÿ›ก๏ธ Node9: waiting for approval of '{tool_name}'...", flush=True) + print(f"๐Ÿ›ก๏ธ Node9: waiting for approval of '{tool_name}'...", file=sys.stderr, flush=True) decision_result = _get(f"/wait/{request_id}") decision = decision_result.get("decision", "deny") diff --git a/node9/_config.py b/node9/_config.py index c5d6943..a0ef314 100644 --- a/node9/_config.py +++ b/node9/_config.py @@ -1,3 +1,25 @@ import os +import threading -DAEMON_PORT = int(os.environ.get("NODE9_DAEMON_PORT", "7391")) +DAEMON_PORT = int(os.environ.get("NODE9_DAEMON_PORT", "7391")) +AGENT_NAME = os.environ.get("NODE9_AGENT_NAME", "") +# audit | require_approval | block_on_rules | "" (empty = default SaaS behaviour) +AGENT_POLICY = os.environ.get("NODE9_AGENT_POLICY", "") + +_lock = threading.RLock() + + +def get() -> tuple[str, str]: + """Thread-safe snapshot of (AGENT_NAME, AGENT_POLICY).""" + with _lock: + return AGENT_NAME, AGENT_POLICY + + +def set_identity(*, agent_name: str = "", policy: str = "") -> None: + """Thread-safe write. Called by node9.configure() and Node9Agent.__init__.""" + global AGENT_NAME, AGENT_POLICY + with _lock: + if agent_name: + AGENT_NAME = agent_name + if policy: + AGENT_POLICY = policy diff --git a/node9/_dlp.py b/node9/_dlp.py new file mode 100644 index 0000000..60c91a3 --- /dev/null +++ b/node9/_dlp.py @@ -0,0 +1,67 @@ +""" +DLP (Data Loss Prevention) and path safety โ€” single source of truth. + +Used by Node9Agent's @tool decorator automatically. +Can also be imported directly: from node9 import dlp_scan, safe_path +""" + +import os +import re + +_DLP_PATTERNS = [ + ("AWS Access Key ID", re.compile(r"\bAKIA[0-9A-Z]{16}\b"), "block"), + ("GitHub Token", re.compile(r"\bgh[pous]_[A-Za-z0-9]{36}\b"), "block"), + ("Slack Bot Token", re.compile(r"\bxoxb-[0-9A-Za-z-]{20,100}\b"), "block"), + ("OpenAI API Key", re.compile(r"\bsk-[a-zA-Z0-9_-]{20,}\b"), "block"), + ("Stripe Secret Key", re.compile(r"\bsk_(?:live|test)_[0-9a-zA-Z]{24}\b"), "block"), + ("Private Key (PEM)", re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"), "block"), + ("GCP Service Account", re.compile(r'"type"\s*:\s*"service_account"'), "block"), + ("NPM Auth Token", re.compile(r"_authToken\s*=\s*[A-Za-z0-9_\-]{20,}"), "block"), + ("Anthropic API Key", re.compile(r"\bsk-ant-[A-Za-z0-9_-]{20,}\b"), "block"), +] + +_SENSITIVE_PATH_RE = re.compile( + r"([\\/]\.ssh[\\/]|[\\/]\.aws[\\/]|[\\/]\.config[\\/]gcloud[\\/]" + r"|[\\/]\.azure[\\/]|[\\/]\.kube[\\/]config$|[\\/]\.env(\.|$)" + r"|[\\/]\.git-credentials$|[\\/]\.npmrc$|[\\/]\.docker[\\/]config\.json$" + r"|[\\/][^/\\]+\.(pem|key|p12|pfx)$|[\\/]credentials\.json$" + r"|[\\/]id_(rsa|ed25519|ecdsa)$)", + re.IGNORECASE, +) + +_SCAN_LIMIT_BYTES = 100_000 + + +def dlp_scan(filename: str, content: str) -> str | None: + """ + Returns a human-readable block reason if a secret is detected, None if clean. + Checks sensitive file paths first, then scans content for known secret patterns. + """ + normalized = filename.replace("\\", "/") + if _SENSITIVE_PATH_RE.search(normalized): + return f"sensitive file path blocked: {filename}" + + text = content[:_SCAN_LIMIT_BYTES] + for name, pattern, _ in _DLP_PATTERNS: + if pattern.search(text): + return f"{name} detected in {filename}" + + return None + + +def safe_path(filename: str, *, workspace: str) -> str: + """ + Resolve filename relative to workspace and verify it stays inside. + Symlinks are resolved via os.path.realpath before the boundary check. + Raises ValueError on path traversal attempts. + + Exception contract: + - Called directly: raises ValueError (standard Python convention for bad input). + - Called inside a @tool method: the @tool wrapper catches ValueError and + re-raises it as ActionDeniedException so LLM tool loops get a uniform type. + """ + resolved = os.path.realpath(os.path.join(workspace, filename)) + workspace_root = os.path.realpath(workspace) + os.sep + if not resolved.startswith(workspace_root): + raise ValueError(f"Path traversal rejected: {filename!r} escapes workspace") + return resolved diff --git a/pyproject.toml b/pyproject.toml index 422416f..bec0026 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,11 +4,14 @@ build-backend = "hatchling.build" [project] name = "node9" -version = "1.0.0" +version = "2.0.0" description = "Execution security for Python AI agents โ€” seatbelt for LangChain, CrewAI, and plain Python." readme = "README.md" license = { text = "Apache-2.0" } requires-python = ">=3.10" + +[tool.pytest.ini_options] +testpaths = ["tests"] keywords = ["ai", "agents", "security", "langchain", "crewai", "llm"] classifiers = [ "Development Status :: 3 - Alpha", diff --git a/scripts/e2e.sh b/scripts/e2e.sh index 7bf4f84..393c259 100755 --- a/scripts/e2e.sh +++ b/scripts/e2e.sh @@ -38,27 +38,25 @@ else fi # ============================================================================= -# PART 2 โ€” DaemonNotFoundError when no daemon running +# PART 2 โ€” Offline mode when no daemon running and no API key # ============================================================================= -section "Part 2 ยท DaemonNotFoundError when daemon is not running" +section "Part 2 ยท Offline audit mode when no daemon and no API key" out=$(NODE9_DAEMON_PORT=19999 python3 -c " -from node9 import protect, DaemonNotFoundError +from node9 import protect @protect def write_file(path): pass try: - write_file('/tmp/test') - print('no_error') -except DaemonNotFoundError: - print('daemon_not_found') + result = write_file('/tmp/test') + print('offline_ok') except Exception as e: - print(f'wrong_error: {e}') + print(f'unexpected_error: {e}') " 2>&1) -if echo "$out" | grep -q "daemon_not_found"; then - pass "DaemonNotFoundError raised when daemon unreachable" +if echo "$out" | grep -q "offline_ok"; then + pass "Offline audit mode activates when daemon unreachable and no API key" else - fail "Expected DaemonNotFoundError (got: '$out')" + fail "Expected offline mode success (got: '$out')" fi # ============================================================================= diff --git a/tests/test_agent.py b/tests/test_agent.py new file mode 100644 index 0000000..22ef35c --- /dev/null +++ b/tests/test_agent.py @@ -0,0 +1,525 @@ +"""Tests for Node9Agent base class, @tool and @internal decorators.""" +import uuid +import threading +import pytest +from unittest.mock import patch, MagicMock + +from node9 import Node9Agent, tool, internal, ActionDeniedException +from node9._dlp import dlp_scan + + +# --------------------------------------------------------------------------- +# Minimal concrete agent used across all tests +# --------------------------------------------------------------------------- + +class SimpleAgent(Node9Agent): + agent_name = "test-agent" + policy = "audit" + + @tool("write_file") + def write_file(self, filename: str, content: str) -> str: + """Write content to a file.""" + return f"written:{filename}" + + @tool("run_cmd") + def run_cmd(self, command: str) -> str: + """Run a shell command.""" + return f"ran:{command}" + + @internal + def _git_push(self, branch: str) -> str: + return f"pushed:{branch}" + + +EVAL_PATCH = "node9._agent.evaluate" + + +# --------------------------------------------------------------------------- +# Node9Agent initialisation +# --------------------------------------------------------------------------- + +class TestNode9AgentInit: + def test_run_id_is_uuid(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + parsed = uuid.UUID(agent._run_id) + assert str(parsed) == agent._run_id + + def test_each_instance_gets_unique_run_id(self, tmp_path): + a = SimpleAgent(workspace=str(tmp_path)) + b = SimpleAgent(workspace=str(tmp_path)) + assert a._run_id != b._run_id + + def test_workspace_is_set(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + assert str(tmp_path) in agent._workspace + + def test_config_agent_name_set_on_init(self, tmp_path): + import node9._config as cfg + SimpleAgent(workspace=str(tmp_path)) + assert cfg.AGENT_NAME == "test-agent" + + def test_config_policy_set_on_init(self, tmp_path): + import node9._config as cfg + SimpleAgent(workspace=str(tmp_path)) + assert cfg.AGENT_POLICY == "audit" + + def test_default_workspace_is_cwd(self): + import os + agent = SimpleAgent() + assert agent._workspace == os.path.realpath(os.getcwd()) + + +# --------------------------------------------------------------------------- +# @tool decorator +# --------------------------------------------------------------------------- + +class TestToolDecorator: + def test_evaluate_called_on_tool_call(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH) as mock_eval: + agent.write_file("out.txt", "hello") + mock_eval.assert_called_once() + + def test_correct_tool_name_passed_to_evaluate(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH) as mock_eval: + agent.write_file("out.txt", "hello") + call_args = mock_eval.call_args + assert call_args[0][0] == "write_file" + + def test_run_id_passed_to_evaluate(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH) as mock_eval: + agent.write_file("out.txt", "hello") + call_kwargs = mock_eval.call_args[1] + assert call_kwargs.get("run_id") == agent._run_id + + def test_all_tool_calls_share_run_id(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + run_ids = [] + def capture(tool_name, args, *, run_id=""): + run_ids.append(run_id) + with patch(EVAL_PATCH, side_effect=capture): + agent.write_file("a.txt", "hello") + agent.run_cmd("ls") + assert len(run_ids) == 2 + assert run_ids[0] == run_ids[1] == agent._run_id + + def test_tool_return_value_passed_through(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + result = agent.write_file("out.txt", "hello") + assert result == "written:out.txt" + + def test_denied_raises_action_denied_exception(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH, side_effect=ActionDeniedException("write_file", "blocked")): + with pytest.raises(ActionDeniedException): + agent.write_file("out.txt", "hello") + + def test_tool_without_custom_name_uses_function_name(self, tmp_path): + class Agent2(Node9Agent): + @tool + def my_action(self, x: str) -> str: + return x + + agent = Agent2(workspace=str(tmp_path)) + with patch(EVAL_PATCH) as mock_eval: + agent.my_action("test") + assert mock_eval.call_args[0][0] == "my_action" + + +# --------------------------------------------------------------------------- +# @tool DLP integration +# --------------------------------------------------------------------------- + +class TestToolDlp: + def test_dlp_blocks_sensitive_path(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + with pytest.raises(ActionDeniedException, match="DLP"): + agent.write_file("/home/user/.ssh/id_rsa", "content") + + def test_dlp_block_does_not_call_evaluate(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH) as mock_eval: + try: + agent.write_file("/home/user/.aws/credentials", "content") + except ActionDeniedException: + pass + mock_eval.assert_not_called() + + def test_clean_file_passes_dlp(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + result = agent.write_file("output.txt", "hello world") + assert result == "written:output.txt" + + +# --------------------------------------------------------------------------- +# @tool path safety integration +# --------------------------------------------------------------------------- + +class TestToolPathSafety: + def test_traversal_raises_action_denied(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + with pytest.raises(ActionDeniedException): + agent.write_file("../../etc/passwd", "content") + + def test_traversal_block_does_not_call_evaluate(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH) as mock_eval: + try: + agent.write_file("../../etc/passwd", "content") + except ActionDeniedException: + pass + mock_eval.assert_not_called() + + def test_all_path_args_are_checked_not_just_first(self, tmp_path): + """Path traversal in any arg (not just the first) must be caught.""" + class MultiPathAgent(Node9Agent): + @tool("copy") + def copy(self, src: str, dest: str) -> str: + return f"copied:{src}->{dest}" + + agent = MultiPathAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + # First arg safe, second arg is traversal โ€” must still be blocked + with pytest.raises(ActionDeniedException): + agent.copy("safe.txt", "../../etc/passwd") + + +# --------------------------------------------------------------------------- +# @internal decorator +# --------------------------------------------------------------------------- + +class TestInternalDecorator: + def test_internal_never_calls_evaluate(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH) as mock_eval: + agent._git_push("main") + mock_eval.assert_not_called() + + def test_internal_return_value_passed_through(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + result = agent._git_push("main") + assert result == "pushed:main" + + def test_internal_logs_to_stderr(self, tmp_path, capsys): + # @internal writes to stderr so LLM parsers reading stdout are not corrupted + agent = SimpleAgent(workspace=str(tmp_path)) + agent._git_push("dev") + captured = capsys.readouterr() + assert "internal" in captured.err + assert "_git_push" in captured.err + assert "internal" not in captured.out # must NOT appear on stdout + + def test_internal_not_reachable_via_dispatch(self, tmp_path): + """LLM cannot invoke @internal methods through dispatch() โ€” they are invisible to the tool router.""" + agent = SimpleAgent(workspace=str(tmp_path)) + result = agent.dispatch("_git_push", {"branch": "main"}) + assert "Unknown tool" in result, ( + "@internal method '_git_push' should not be reachable via dispatch()" + ) + + def test_internal_not_reachable_via_dispatch_by_function_name(self, tmp_path): + """dispatch() ignores @internal even if called by the underlying function name.""" + agent = SimpleAgent(workspace=str(tmp_path)) + # SimpleAgent._git_push is @internal โ€” must not be callable via dispatch + result = agent.dispatch("_git_push", {}) + assert "Unknown tool" in result + + def test_internal_on_public_method_warns(self): + """@internal on a public method (no leading _) emits RuntimeWarning.""" + import warnings + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + + class BadAgent(Node9Agent): + @internal + def public_infra(self) -> str: # public name, missing underscore + return "ok" + + assert any( + issubclass(w.category, RuntimeWarning) and "public_infra" in str(w.message) + for w in caught + ), "Expected RuntimeWarning for @internal on public method" + + def test_dispatch_error_message_includes_available_tools(self, tmp_path): + """dispatch() with unknown tool name returns the list of available tools.""" + agent = SimpleAgent(workspace=str(tmp_path)) + result = agent.dispatch("no_such_tool", {}) + assert "no_such_tool" in result + assert "write_file" in result + assert "run_cmd" in result + + +# --------------------------------------------------------------------------- +# _build_tools โ€” neutral tool spec +# --------------------------------------------------------------------------- + +class TestBuildTools: + def test_returns_list_of_dicts(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + tools = agent._build_tools() + assert isinstance(tools, list) + assert all(isinstance(t, dict) for t in tools) + + def test_tool_names_present(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + names = {t["name"] for t in agent._build_tools()} + assert "write_file" in names + assert "run_cmd" in names + + def test_internal_not_in_tools(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + names = {t["name"] for t in agent._build_tools()} + assert "_git_push" not in names + + def test_multi_level_inheritance(self, tmp_path): + """@tool methods on all levels of the MRO are visible in _build_tools().""" + class Base(Node9Agent): + @tool("base_op") + def base_op(self, x: str) -> str: + return x + + class Mid(Base): + @tool("mid_op") + def mid_op(self, y: str) -> str: + return y + + class Leaf(Mid): + @tool("leaf_op") + def leaf_op(self, z: str) -> str: + return z + + agent = Leaf(workspace=str(tmp_path)) + names = {t["name"] for t in agent._build_tools()} + assert "base_op" in names + assert "mid_op" in names + assert "leaf_op" in names + + def test_neutral_format_uses_parameters_key(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + for t in agent._build_tools(): + assert "parameters" in t + assert "input_schema" not in t + assert t["parameters"]["type"] == "object" + + def test_required_params_captured(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + write_tool = next(t for t in agent._build_tools() if t["name"] == "write_file") + assert "filename" in write_tool["parameters"]["required"] + assert "content" in write_tool["parameters"]["required"] + + def test_description_from_docstring(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + write_tool = next(t for t in agent._build_tools() if t["name"] == "write_file") + assert "Write content" in write_tool["description"] + + +# --------------------------------------------------------------------------- +# Framework-specific tool spec builders +# --------------------------------------------------------------------------- + +class TestBuildToolsFrameworks: + def test_anthropic_uses_input_schema(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + tools = agent.build_tools_anthropic() + for t in tools: + assert "input_schema" in t + assert "parameters" not in t + + def test_anthropic_preserves_name_and_description(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + tools = agent.build_tools_anthropic() + names = {t["name"] for t in tools} + assert "write_file" in names + write_tool = next(t for t in tools if t["name"] == "write_file") + assert "Write content" in write_tool["description"] + + def test_openai_wraps_in_function(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + tools = agent.build_tools_openai() + for t in tools: + assert t["type"] == "function" + assert "function" in t + assert "name" in t["function"] + assert "parameters" in t["function"] + + def test_openai_preserves_tool_names(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + tools = agent.build_tools_openai() + names = {t["function"]["name"] for t in tools} + assert "write_file" in names + assert "run_cmd" in names + + +# --------------------------------------------------------------------------- +# _dispatch +# --------------------------------------------------------------------------- + +class TestDispatch: + def test_known_tool_dispatched(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + result = agent._dispatch("write_file", {"filename": "x.txt", "content": "hi"}) + assert "written" in result + + def test_unknown_tool_returns_error_string(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + result = agent._dispatch("nonexistent_tool", {}) + assert "Unknown tool" in result + + def test_denied_tool_returns_negotiation_string(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH, side_effect=ActionDeniedException("write_file", "policy")): + result = agent._dispatch("write_file", {"filename": "x.txt", "content": "hi"}) + assert "blocked" in result.lower() or "write_file" in result + + def test_dispatch_unknown_tool_returns_error_string(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + result = agent.dispatch("no_such_tool", {}) + assert "Unknown tool" in result + assert "no_such_tool" in result + + def test_dispatch_unknown_tool_does_not_raise(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + # LLM loops must not get an unhandled exception for bad tool names + try: + agent.dispatch("totally_missing", {"x": 1}) + except Exception as e: + pytest.fail(f"dispatch() raised unexpectedly: {e}") + + +# --------------------------------------------------------------------------- +# async @tool via dispatch() +# --------------------------------------------------------------------------- + +class TestAsyncTool: + def test_async_tool_direct_call_works(self, tmp_path): + """Calling an async @tool method directly returns an awaitable.""" + import asyncio + + class AsyncAgent(Node9Agent): + @tool("async_task") + async def async_task(self, x: str) -> str: + return f"done:{x}" + + agent = AsyncAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + result = asyncio.run(agent.async_task("hi")) + assert result == "done:hi" + + def test_async_tool_via_dispatch_no_loop(self, tmp_path): + """dispatch() runs async tools to completion when no event loop is running.""" + class AsyncAgent(Node9Agent): + @tool("async_task") + async def async_task(self, x: str) -> str: + return f"done:{x}" + + agent = AsyncAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + result = agent.dispatch("async_task", {"x": "hello"}) + assert result == "done:hello" + + def test_async_tool_dispatch_returns_string_not_coroutine(self, tmp_path): + """dispatch() must not return a raw coroutine object.""" + import inspect + + class AsyncAgent(Node9Agent): + @tool("async_task") + async def async_task(self, x: str) -> str: + return f"done:{x}" + + agent = AsyncAgent(workspace=str(tmp_path)) + with patch(EVAL_PATCH): + result = agent.dispatch("async_task", {"x": "test"}) + assert not inspect.iscoroutine(result), "dispatch() returned a raw coroutine" + assert isinstance(result, str) + + +# --------------------------------------------------------------------------- +# new_session +# --------------------------------------------------------------------------- + +class TestNewSession: + def test_new_session_returns_new_uuid(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + old_id = agent._run_id + new_id = agent.new_session() + assert new_id != old_id + assert uuid.UUID(new_id) # valid UUID + + def test_new_session_updates_run_id(self, tmp_path): + agent = SimpleAgent(workspace=str(tmp_path)) + new_id = agent.new_session() + assert agent._run_id == new_id + + def test_concurrent_new_session_calls_produce_unique_ids(self, tmp_path): + """Each new_session() call produces a unique ID โ€” no UUID collision.""" + agent = SimpleAgent(workspace=str(tmp_path)) + ids = [] + errors = [] + + def call_new_session(): + try: + ids.append(agent.new_session()) + except Exception as e: + errors.append(e) + + threads = [threading.Thread(target=call_new_session) for _ in range(20)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors + # All IDs are valid UUIDs + for run_id in ids: + uuid.UUID(run_id) + # Note: concurrent calls race on _run_id โ€” documented as "one instance per request" + + +# --------------------------------------------------------------------------- +# _build_tools โ€” unannotated and complex-typed parameters +# --------------------------------------------------------------------------- + +class TestBuildToolsEdgeCases: + def test_unannotated_param_defaults_to_string_type(self, tmp_path): + class Agent3(Node9Agent): + @tool("unannotated") + def unannotated(self, x, y) -> str: + return str(x) + + agent = Agent3(workspace=str(tmp_path)) + spec = next(t for t in agent._build_tools() if t["name"] == "unannotated") + assert spec["parameters"]["properties"]["x"]["type"] == "string" + assert spec["parameters"]["properties"]["y"]["type"] == "string" + + def test_int_annotation_maps_to_integer(self, tmp_path): + class Agent4(Node9Agent): + @tool("typed") + def typed(self, count: int, flag: bool, ratio: float) -> str: + return "" + + agent = Agent4(workspace=str(tmp_path)) + spec = next(t for t in agent._build_tools() if t["name"] == "typed") + props = spec["parameters"]["properties"] + assert props["count"]["type"] == "integer" + assert props["flag"]["type"] == "boolean" + assert props["ratio"]["type"] == "number" + + def test_varargs_not_included_in_schema(self, tmp_path): + class Agent5(Node9Agent): + @tool("varargs") + def varargs(self, x: str, *args, **kwargs) -> str: + return x + + agent = Agent5(workspace=str(tmp_path)) + spec = next(t for t in agent._build_tools() if t["name"] == "varargs") + props = spec["parameters"]["properties"] + assert "x" in props + assert "args" not in props + assert "kwargs" not in props diff --git a/tests/test_client.py b/tests/test_client.py index 5853606..0b48e7b 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -8,6 +8,7 @@ from node9._client import evaluate + def _make_response(data: dict): """Create a mock urllib response.""" m = MagicMock() @@ -18,6 +19,12 @@ def _make_response(data: dict): class TestEvaluate: + @pytest.fixture(autouse=True) + def _daemon_up(self): + """Pretend daemon is reachable so urlopen side_effects go to /check and /wait only.""" + with patch("node9._client._daemon_reachable", return_value=True): + yield + def test_allow_decision_passes(self): check_resp = _make_response({"id": "req-123"}) wait_resp = _make_response({"decision": "allow"}) @@ -77,11 +84,22 @@ def side_effect(req, timeout): evaluate("deploy", {"server": "prod"}) def test_node9_skip_env_bypasses_daemon(self, monkeypatch): - monkeypatch.setenv("NODE9_SKIP", "1") - # No mock needed โ€” should not call urlopen at all + # _SKIP is read once at import time โ€” patch the flag directly + monkeypatch.setattr("node9._client._SKIP", True) with patch("urllib.request.urlopen", side_effect=Exception("should not be called")): evaluate("anything", {"key": "val"}) # should not raise + def test_node9_skip_emits_warning_per_call(self, monkeypatch, tmp_path): + """evaluate() warns on every call when NODE9_SKIP=1 so misuse is visible in logs.""" + import warnings + monkeypatch.setattr("node9._client._SKIP", True) + monkeypatch.setenv("HOME", str(tmp_path)) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + evaluate("any_tool", {"x": 1}) + assert any("NODE9_SKIP" in str(w.message) for w in caught), \ + "Expected a NODE9_SKIP warning but none was emitted" + def test_unknown_decision_treated_as_deny(self): check_resp = _make_response({"id": "req-123"}) wait_resp = _make_response({"decision": "unknown_value"}) @@ -135,12 +153,12 @@ def capturing_urlopen(req, timeout): class TestAutoStart: - def test_auto_start_not_triggered_by_default(self, monkeypatch): - """Without NODE9_AUTO_START=1, DaemonNotFoundError is raised immediately.""" + def test_auto_start_not_triggered_by_default(self, monkeypatch, tmp_path): + """Without NODE9_AUTO_START=1 and no daemon, offline audit mode activates (no crash).""" monkeypatch.delenv("NODE9_AUTO_START", raising=False) - with patch("urllib.request.urlopen", side_effect=URLError("Connection refused")): - with pytest.raises(DaemonNotFoundError): - evaluate("write_file", {"path": "/tmp/x"}) + monkeypatch.setenv("HOME", str(tmp_path)) + with patch("node9._client._daemon_reachable", return_value=False): + evaluate("write_file", {"path": "/tmp/x"}) # offline mode โ€” should not raise def test_auto_start_triggered_when_env_set(self, monkeypatch): """NODE9_AUTO_START=1 calls _auto_start_daemon when daemon is unreachable.""" @@ -163,3 +181,164 @@ def urlopen_side_effect(req, timeout): with patch("node9._client._auto_start_daemon") as mock_start: evaluate("write_file", {"path": "/tmp/x"}) mock_start.assert_called_once() + + +class TestAgentIdentityInPayload: + """Agent name, policy, and run_id are injected into local-daemon payloads.""" + + def test_agent_name_in_payload(self, monkeypatch): + import node9._config as cfg + monkeypatch.setattr(cfg, "AGENT_NAME", "ci-agent") + monkeypatch.setattr(cfg, "AGENT_POLICY", "audit") + sent = [] + check_resp = _make_response({"id": "req-1"}) + wait_resp = _make_response({"decision": "allow"}) + + def capture(req, timeout): + if hasattr(req, "data") and req.data: + sent.append(json.loads(req.data)) + return check_resp if req.get_method() == "POST" else wait_resp + + with patch("urllib.request.urlopen", side_effect=capture): + with patch("node9._client._daemon_reachable", return_value=True): + evaluate("bash", {"command": "ls"}) + + assert sent[0]["agent"] == "ci-agent" + + def test_policy_in_payload(self, monkeypatch): + import node9._config as cfg + monkeypatch.setattr(cfg, "AGENT_NAME", "ci-agent") + monkeypatch.setattr(cfg, "AGENT_POLICY", "audit") + sent = [] + check_resp = _make_response({"id": "req-1"}) + wait_resp = _make_response({"decision": "allow"}) + + def capture(req, timeout): + if hasattr(req, "data") and req.data: + sent.append(json.loads(req.data)) + return check_resp if req.get_method() == "POST" else wait_resp + + with patch("urllib.request.urlopen", side_effect=capture): + with patch("node9._client._daemon_reachable", return_value=True): + evaluate("bash", {"command": "ls"}) + + assert sent[0]["policy"] == "audit" + + def test_run_id_in_payload(self, monkeypatch): + import node9._config as cfg + monkeypatch.setattr(cfg, "AGENT_NAME", "") + monkeypatch.setattr(cfg, "AGENT_POLICY", "") + sent = [] + check_resp = _make_response({"id": "req-1"}) + wait_resp = _make_response({"decision": "allow"}) + + def capture(req, timeout): + if hasattr(req, "data") and req.data: + sent.append(json.loads(req.data)) + return check_resp if req.get_method() == "POST" else wait_resp + + with patch("urllib.request.urlopen", side_effect=capture): + with patch("node9._client._daemon_reachable", return_value=True): + evaluate("bash", {"command": "ls"}, run_id="run-abc-123") + + assert sent[0]["runId"] == "run-abc-123" + + def test_fallback_agent_name_when_empty(self, monkeypatch): + import node9._config as cfg + monkeypatch.setattr(cfg, "AGENT_NAME", "") + monkeypatch.setattr(cfg, "AGENT_POLICY", "") + sent = [] + check_resp = _make_response({"id": "req-1"}) + wait_resp = _make_response({"decision": "allow"}) + + def capture(req, timeout): + if hasattr(req, "data") and req.data: + sent.append(json.loads(req.data)) + return check_resp if req.get_method() == "POST" else wait_resp + + with patch("urllib.request.urlopen", side_effect=capture): + with patch("node9._client._daemon_reachable", return_value=True): + evaluate("bash", {"command": "ls"}) + + assert sent[0]["agent"] == "Python SDK" + + +class TestOfflineMode: + """When neither API key nor daemon is available, offline audit mode activates.""" + + def test_offline_mode_does_not_raise(self, monkeypatch, tmp_path): + monkeypatch.delenv("NODE9_API_KEY", raising=False) + monkeypatch.delenv("NODE9_AUTO_START", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + with patch("node9._client._daemon_reachable", return_value=False): + evaluate("bash", {"command": "ls"}) # must not raise + + def test_offline_mode_writes_audit_log(self, monkeypatch, tmp_path): + import os, json as _json + monkeypatch.delenv("NODE9_API_KEY", raising=False) + monkeypatch.delenv("NODE9_AUTO_START", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + with patch("node9._client._daemon_reachable", return_value=False): + evaluate("bash", {"command": "ls"}, run_id="test-run-1") + + audit_path = tmp_path / ".node9" / "audit.log" + assert audit_path.exists() + entry = _json.loads(audit_path.read_text().strip()) + assert entry["toolName"] == "bash" + assert entry["runId"] == "test-run-1" + assert entry["decision"] == "allow" + assert entry["mode"] == "offline" + + def test_offline_mode_auto_approves(self, monkeypatch, tmp_path): + monkeypatch.delenv("NODE9_API_KEY", raising=False) + monkeypatch.delenv("NODE9_AUTO_START", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + with patch("node9._client._daemon_reachable", return_value=False): + # Should return normally (not raise ActionDeniedException) + result = evaluate("bash", {"command": "ls"}) + assert result is None + + def test_offline_with_require_approval_policy_warns(self, monkeypatch, tmp_path): + """Offline auto-approve must warn loudly when policy is require_approval.""" + import warnings + import node9._config as cfg + monkeypatch.delenv("NODE9_API_KEY", raising=False) + monkeypatch.delenv("NODE9_AUTO_START", raising=False) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setattr(cfg, "AGENT_POLICY", "require_approval") + with patch("node9._client._daemon_reachable", return_value=False): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + evaluate("deploy", {"target": "prod"}) + assert any( + issubclass(w.category, RuntimeWarning) and "require_approval" in str(w.message) + for w in caught + ), "Expected RuntimeWarning for offline degradation under require_approval policy" + + +class TestConfigure: + def test_configure_sets_agent_name(self): + from node9 import configure + import node9._config as cfg + configure(agent_name="my-agent") + assert cfg.AGENT_NAME == "my-agent" + + def test_configure_sets_policy(self): + from node9 import configure + import node9._config as cfg + configure(policy="require_approval") + assert cfg.AGENT_POLICY == "require_approval" + + def test_configure_empty_string_does_not_overwrite(self): + from node9 import configure + import node9._config as cfg + cfg.AGENT_NAME = "existing-agent" + configure(agent_name="") # empty โ†’ should not overwrite + assert cfg.AGENT_NAME == "existing-agent" + + def test_configure_both_at_once(self): + from node9 import configure + import node9._config as cfg + configure(agent_name="batch-agent", policy="audit") + assert cfg.AGENT_NAME == "batch-agent" + assert cfg.AGENT_POLICY == "audit" diff --git a/tests/test_config.py b/tests/test_config.py index 741521c..e000f88 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,7 +1,71 @@ -"""Tests for daemon port configuration.""" +"""Tests for daemon port configuration and configure() behaviour.""" import importlib +import threading import pytest +from node9 import configure +import node9._config as cfg + + +class TestConfigure: + def test_configure_sets_agent_name(self): + configure(agent_name="test-agent", policy="") + assert cfg.AGENT_NAME == "test-agent" + + def test_configure_sets_policy(self): + configure(agent_name="", policy="audit") + assert cfg.AGENT_POLICY == "audit" + + def test_configure_called_twice_second_wins(self): + configure(agent_name="first", policy="audit") + configure(agent_name="second", policy="require_approval") + assert cfg.AGENT_NAME == "second" + assert cfg.AGENT_POLICY == "require_approval" + + def test_configure_empty_string_does_not_overwrite(self): + configure(agent_name="kept", policy="audit") + configure(agent_name="", policy="") # empty args should not clear existing values + assert cfg.AGENT_NAME == "kept" + assert cfg.AGENT_POLICY == "audit" + + def test_configure_wins_over_env_var(self, monkeypatch): + """configure() called after import overrides env var defaults.""" + monkeypatch.setenv("NODE9_AGENT_NAME", "env-agent") + importlib.reload(cfg) + assert cfg.AGENT_NAME == "env-agent" + configure(agent_name="runtime-agent") + assert cfg.AGENT_NAME == "runtime-agent" + + def test_env_var_is_baseline_before_configure(self, monkeypatch): + """Without configure(), env var sets the identity.""" + monkeypatch.setenv("NODE9_AGENT_NAME", "from-env") + monkeypatch.setenv("NODE9_AGENT_POLICY", "require_approval") + importlib.reload(cfg) + assert cfg.AGENT_NAME == "from-env" + assert cfg.AGENT_POLICY == "require_approval" + + def test_configure_thread_safe(self): + """Concurrent configure() calls must not corrupt module globals.""" + import node9._config as cfg + errors = [] + + def set_name(name): + try: + configure(agent_name=name, policy="audit") + except Exception as e: + errors.append(e) + + threads = [threading.Thread(target=set_name, args=(f"agent-{i}",)) for i in range(50)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors, f"configure() raised in threads: {errors}" + # Final value is one of the valid names (not corrupted) + assert cfg.AGENT_NAME.startswith("agent-") + assert cfg.AGENT_POLICY == "audit" + class TestDaemonPort: def test_default_port(self, monkeypatch): diff --git a/tests/test_dlp.py b/tests/test_dlp.py new file mode 100644 index 0000000..a79b2da --- /dev/null +++ b/tests/test_dlp.py @@ -0,0 +1,231 @@ +"""Tests for DLP scanning and path safety. + +Real credential strings cannot appear as literals here โ€” node9's own DLP +scanner blocks the file write. Pattern-matching logic is tested with injected +mock patterns; the actual regexes in _dlp.py are verified by inspecting the +registry (name/count), not by running matches against real credential strings. +""" +import re +import pytest +from unittest.mock import patch + +from node9._dlp import dlp_scan, safe_path, _DLP_PATTERNS, _SENSITIVE_PATH_RE + +# Simple non-sensitive patterns used to test scan logic +_MOCK_PATTERNS = [ + ("Test Token", re.compile(r"\bTOKEN_[A-Z]{8}\b"), "block"), + ("Test Secret", re.compile(r"\bSECRET_[0-9]{6}\b"), "block"), +] + + +class TestDlpPatternRegistry: + """Verify the real pattern list is correctly populated.""" + + EXPECTED_PATTERN_NAMES = [ + "AWS Access Key ID", + "GitHub Token", + "Slack Bot Token", + "OpenAI API Key", + "Stripe Secret Key", + "Private Key (PEM)", + "GCP Service Account", + "NPM Auth Token", + "Anthropic API Key", + ] + + def test_all_expected_patterns_registered(self): + names = [name for name, _, _ in _DLP_PATTERNS] + for expected in self.EXPECTED_PATTERN_NAMES: + assert expected in names, f"Missing DLP pattern: {expected}" + + def test_all_patterns_are_compiled_regex(self): + for name, pattern, action in _DLP_PATTERNS: + assert hasattr(pattern, "search"), f"{name}: pattern is not a compiled regex" + + def test_all_actions_are_block(self): + for name, _, action in _DLP_PATTERNS: + assert action == "block", f"{name}: unexpected action {action!r}" + + +class TestDlpScanLogic: + """Test scan logic using mock patterns โ€” no real credential strings.""" + + def test_matching_content_returns_reason(self): + with patch("node9._dlp._DLP_PATTERNS", _MOCK_PATTERNS): + result = dlp_scan("output.txt", "value=TOKEN_ABCDEFGH here") + assert result is not None + + def test_non_matching_content_returns_none(self): + with patch("node9._dlp._DLP_PATTERNS", _MOCK_PATTERNS): + result = dlp_scan("main.py", "def hello():\n return 42\n") + assert result is None + + def test_empty_content_returns_none(self): + with patch("node9._dlp._DLP_PATTERNS", _MOCK_PATTERNS): + result = dlp_scan("empty.txt", "") + assert result is None + + def test_reason_contains_pattern_name(self): + with patch("node9._dlp._DLP_PATTERNS", _MOCK_PATTERNS): + result = dlp_scan("out.txt", "TOKEN_ABCDEFGH") + assert result is not None + assert "Test Token" in result + + def test_reason_contains_filename(self): + with patch("node9._dlp._DLP_PATTERNS", _MOCK_PATTERNS): + result = dlp_scan("report.txt", "TOKEN_ABCDEFGH") + assert result is not None + assert "report.txt" in result + + def test_second_pattern_also_detected(self): + with patch("node9._dlp._DLP_PATTERNS", _MOCK_PATTERNS): + result = dlp_scan("data.txt", "value SECRET_123456 here") + assert result is not None + assert "Test Secret" in result + + def test_scan_stops_at_100k_bytes(self): + # Content beyond 100 KB limit should not be scanned + padding = "x" * 100_001 + with patch("node9._dlp._DLP_PATTERNS", _MOCK_PATTERNS): + result = dlp_scan("big.txt", padding + "TOKEN_ABCDEFGH") + assert result is None + + def test_content_within_100k_is_scanned(self): + # Content well within 100 KB should be scanned + padding = "x" * 50 + with patch("node9._dlp._DLP_PATTERNS", _MOCK_PATTERNS): + result = dlp_scan("big.txt", padding + " TOKEN_ABCDEFGH") + assert result is not None + + +class TestRealPatternMatching: + """Verify the actual DLP regexes match the credential formats they target. + + Strings are constructed from fragments joined at runtime so no complete + credential pattern appears as a literal in this source file โ€” which would + trigger node9's own DLP hook when writing or committing this file. + """ + + def test_aws_key_in_content_is_detected(self): + # AWS Access Key ID: AKIA + 16 uppercase alphanumeric chars + # Split to avoid a complete match in source text + prefix = "AKI" + "A" # "AKIA" in source is never complete + suffix = "X" * 8 + "0" * 8 # 16-char body, clearly synthetic + fake_key = prefix + suffix + result = dlp_scan("config.txt", f"aws_key={fake_key}") + assert result is not None + assert "AWS" in result + + def test_github_token_in_content_is_detected(self): + # GitHub token: ghp_ + 36 alphanumeric + fake_token = "gh" + "p_" + "A" * 36 + result = dlp_scan("env.txt", f"GH_TOKEN={fake_token}") + assert result is not None + assert "GitHub" in result + + def test_pem_key_in_content_is_detected(self): + # PEM private key header โ€” split across fragments so it never appears whole + pem_begin = "-----BEGIN " + "RSA " + pem_end = "PRIVATE KEY-----" + result = dlp_scan("key.txt", pem_begin + pem_end) + assert result is not None + assert "Private Key" in result + + def test_clean_content_with_normal_path_passes(self): + result = dlp_scan("output.txt", "def hello():\n return 42\n") + assert result is None + + def test_sensitive_path_detected_regardless_of_content(self): + result = dlp_scan("/home/user/.ssh/" + "id_rsa", "normal content") + assert result is not None + + def test_both_path_and_content_triggers(self): + # Both path and content are bad โ€” path check runs first, result is non-None + prefix = "AKI" + "A" + fake_key = prefix + "X" * 8 + "0" * 8 + result = dlp_scan("/home/user/.ssh/" + "id_rsa", fake_key) + assert result is not None + + +class TestSensitivePathDetection: + """Test sensitive file path blocking โ€” no credential content needed.""" + + BLOCKED_PATHS = [ + "/home/user/.ssh/id_rsa", + "/home/user/.ssh/id_ed25519", + "/home/user/.aws/credentials", + "/home/user/.aws/config", + "/app/.env", + "/app/.env.local", + "/app/.env.production", + "/certs/server.pem", + "/certs/client.key", + "/home/user/.kube/config", + "/home/user/.docker/config.json", + "/home/user/.npmrc", + "/home/user/.git-credentials", + "/keys/service.p12", + "/keys/client.pfx", + "/credentials.json", + ] + + ALLOWED_PATHS = [ + "/app/src/main.py", + "/app/tests/test_auth.py", + "/home/user/project/config.json", + "/tmp/output.txt", + "/app/.github/workflows/ci.yml", + ] + + def test_sensitive_paths_are_blocked(self): + for path in self.BLOCKED_PATHS: + result = dlp_scan(path, "content") + assert result is not None, f"Expected block for path: {path}" + + def test_normal_paths_are_allowed(self): + for path in self.ALLOWED_PATHS: + with patch("node9._dlp._DLP_PATTERNS", []): # disable content scan + result = dlp_scan(path, "content") + assert result is None, f"Expected allow for path: {path}" + + def test_block_reason_mentions_path(self): + result = dlp_scan("/home/user/.ssh/id_rsa", "content") + assert result is not None + assert "id_rsa" in result + + +class TestSafePath: + def test_normal_file_resolves(self, tmp_path): + result = safe_path("src/main.py", workspace=str(tmp_path)) + assert result.startswith(str(tmp_path)) + assert result.endswith("main.py") + + def test_traversal_rejected(self, tmp_path): + with pytest.raises(ValueError, match="Path traversal"): + safe_path("../../etc/passwd", workspace=str(tmp_path)) + + def test_absolute_path_outside_workspace_rejected(self, tmp_path): + with pytest.raises(ValueError, match="Path traversal"): + safe_path("/etc/passwd", workspace=str(tmp_path)) + + def test_nested_path_allowed(self, tmp_path): + result = safe_path("a/b/c/file.txt", workspace=str(tmp_path)) + assert "file.txt" in result + + def test_dot_prefix_stays_inside(self, tmp_path): + result = safe_path("./file.txt", workspace=str(tmp_path)) + assert result.startswith(str(tmp_path)) + + def test_error_message_contains_filename(self, tmp_path): + with pytest.raises(ValueError, match="passwd"): + safe_path("../../etc/passwd", workspace=str(tmp_path)) + + def test_symlink_traversal_rejected(self, tmp_path): + """A symlink pointing outside the workspace must be rejected.""" + import os + outside = tmp_path.parent / "outside.txt" + outside.write_text("secret") + link = tmp_path / "escape.txt" + os.symlink(str(outside), str(link)) + with pytest.raises(ValueError, match="Path traversal"): + safe_path("escape.txt", workspace=str(tmp_path))