From 44c52edfcdc419f5b59087f809a553b4bdf41887 Mon Sep 17 00:00:00 2001 From: Jon Saad-Falcon <41205309+jonsaadfalcon@users.noreply.github.com> Date: Sat, 28 Mar 2026 19:26:59 -0700 Subject: [PATCH] =?UTF-8?q?docs:=20security=20hardening=20implementation?= =?UTF-8?q?=20plan=20=E2=80=94=2015=20tasks=20with=20TDD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Detailed plan covering network exposure, boundary guard, webhook validation, file permissions, log sanitization, credential scoping, CORS hardening, and security profiles. Each task has failing tests first, then implementation, then verification. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../plans/2026-03-28-security-hardening.md | 2027 +++++++++++++++++ 1 file changed, 2027 insertions(+) create mode 100644 docs/superpowers/plans/2026-03-28-security-hardening.md diff --git a/docs/superpowers/plans/2026-03-28-security-hardening.md b/docs/superpowers/plans/2026-03-28-security-hardening.md new file mode 100644 index 00000000..30c039e2 --- /dev/null +++ b/docs/superpowers/plans/2026-03-28-security-hardening.md @@ -0,0 +1,2027 @@ +# Security Hardening Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Harden OpenJarvis against network exposure, data leakage to cloud providers, local data exposure, and webhook spoofing — using layered boundary enforcement at device exit points. + +**Architecture:** A `BoundaryGuard` wraps all exit points (cloud engines, external tools, webhooks). Config defaults change to secure values (`127.0.0.1` binding, `redact` mode, rate limiting on). File permissions enforced via shared helpers. Security profiles provide convenience shorthand. Each section is implemented and tested independently in severity order. + +**Tech Stack:** Python 3.10+, FastAPI/Starlette, SQLite, pytest, existing Rust-backed SecretScanner/PIIScanner. + +**Spec:** `docs/superpowers/specs/2026-03-28-security-hardening-design.md` + +--- + +## File Structure + +### New Files +| File | Responsibility | +|------|---------------| +| `src/openjarvis/security/boundary.py` | BoundaryGuard — scans content at device exit points | +| `src/openjarvis/security/file_utils.py` | `secure_mkdir()` and `secure_create()` helpers | +| `tests/security/test_network_defaults.py` | Tests for Section 1 (binding, CORS, auth enforcement) | +| `tests/security/test_boundary_guard.py` | Tests for Section 2 (scan/redact, engine/tool tagging) | +| `tests/security/test_webhook_validation.py` | Tests for Section 3 (fail-closed, secret enforcement) | +| `tests/security/test_file_permissions.py` | Tests for Section 4 (secure_mkdir, secure_create, DB paths) | +| `tests/security/test_log_sanitization.py` | Tests for Section 5 (SanitizingFormatter, scoped credentials) | +| `tests/security/test_security_profiles.py` | Tests for Section 7 (profile field expansion, overrides) | + +### Modified Files +| File | Change | +|------|--------| +| `src/openjarvis/core/config.py` | ServerConfig defaults, SecurityConfig defaults, new fields, profile expansion | +| `src/openjarvis/server/app.py` | CORS from config, startup guards | +| `src/openjarvis/server/auth_middleware.py` | Non-loopback auth enforcement | +| `src/openjarvis/server/webhook_routes.py` | Fail-closed validation | +| `src/openjarvis/server/middleware.py` | CSP header | +| `src/openjarvis/engine/_stubs.py` | `is_cloud` attribute on InferenceEngine | +| `src/openjarvis/engine/cloud.py` | `is_cloud = True` | +| `src/openjarvis/engine/litellm.py` | `is_cloud = True` | +| `src/openjarvis/tools/_stubs.py` | `is_local` attribute on BaseTool, BoundaryGuard in ToolExecutor | +| `src/openjarvis/tools/web_search.py` | `is_local = False` | +| `src/openjarvis/tools/http_request.py` | `is_local = False` | +| `src/openjarvis/tools/browser.py` | `is_local = False` on all browser tools | +| `src/openjarvis/tools/browser_axtree.py` | `is_local = False` | +| `src/openjarvis/tools/channel_tools.py` | `is_local = False` on ChannelSendTool | +| `src/openjarvis/tools/image_tool.py` | `is_local = False` | +| `src/openjarvis/tools/audio_tool.py` | `is_local = False` | +| `src/openjarvis/security/guardrails.py` | Delegate to BoundaryGuard | +| `src/openjarvis/tools/storage/sqlite.py` | secure_create for memory.db | +| `src/openjarvis/server/session_store.py` | secure_create for sessions.db | +| `src/openjarvis/traces/store.py` | secure_create for traces.db | +| `src/openjarvis/security/audit.py` | secure_create for audit.db | +| `src/openjarvis/connectors/store.py` | secure_create for knowledge.db | +| `src/openjarvis/connectors/attachment_store.py` | secure_mkdir/secure_create for blobs | +| `src/openjarvis/core/credentials.py` | `get_tool_credential()`, deprecate `inject_credentials()` | +| `src/openjarvis/cli/log_config.py` | SanitizingFormatter | +| `src/openjarvis/cli/serve.py` | Startup guards, credential audit log | +| `src/openjarvis/cli/doctor_cmd.py` | Security profile check | +| `src/openjarvis/system.py` | BoundaryGuard wiring | + +--- + +## Task 1: Network Exposure — Secure Server Defaults + +**Files:** +- Modify: `src/openjarvis/core/config.py:759-767` (ServerConfig) +- Modify: `src/openjarvis/core/config.py:968-986` (SecurityConfig) +- Test: `tests/security/test_network_defaults.py` + +- [ ] **Step 1: Write failing tests for secure defaults** + +```python +# tests/security/test_network_defaults.py +"""Tests for secure network defaults (Section 1 of security hardening).""" + +from __future__ import annotations + + +class TestServerConfigDefaults: + """ServerConfig should bind to loopback by default.""" + + def test_default_host_is_loopback(self) -> None: + from openjarvis.core.config import ServerConfig + + cfg = ServerConfig() + assert cfg.host == "127.0.0.1" + + def test_default_port_unchanged(self) -> None: + from openjarvis.core.config import ServerConfig + + cfg = ServerConfig() + assert cfg.port == 8000 + + def test_cors_origins_default(self) -> None: + from openjarvis.core.config import ServerConfig + + cfg = ServerConfig() + assert isinstance(cfg.cors_origins, list) + assert "http://localhost:3000" in cfg.cors_origins + assert "http://localhost:5173" in cfg.cors_origins + assert "tauri://localhost" in cfg.cors_origins + assert "*" not in cfg.cors_origins + + +class TestSecurityConfigDefaults: + """SecurityConfig should default to redact mode with rate limiting.""" + + def test_default_mode_is_redact(self) -> None: + from openjarvis.core.config import SecurityConfig + + cfg = SecurityConfig() + assert cfg.mode == "redact" + + def test_rate_limiting_enabled_by_default(self) -> None: + from openjarvis.core.config import SecurityConfig + + cfg = SecurityConfig() + assert cfg.rate_limit_enabled is True + + def test_bypass_defaults_conservative(self) -> None: + from openjarvis.core.config import SecurityConfig + + cfg = SecurityConfig() + assert cfg.local_engine_bypass is False + assert cfg.local_tool_bypass is False + + def test_profile_default_empty(self) -> None: + from openjarvis.core.config import SecurityConfig + + cfg = SecurityConfig() + assert cfg.profile == "" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_network_defaults.py -v` +Expected: FAIL — `ServerConfig` has `host="0.0.0.0"`, no `cors_origins` field, `SecurityConfig` has `mode="warn"`, `rate_limit_enabled=False`, no `local_engine_bypass`/`local_tool_bypass`/`profile` fields. + +- [ ] **Step 3: Update ServerConfig defaults** + +In `src/openjarvis/core/config.py`, replace the `ServerConfig` dataclass (lines 759-767): + +```python +@dataclass(slots=True) +class ServerConfig: + """API server settings.""" + + host: str = "127.0.0.1" + port: int = 8000 + agent: str = "orchestrator" + model: str = "" + workers: int = 1 + cors_origins: list = field( + default_factory=lambda: [ + "http://localhost:3000", + "http://localhost:5173", + "http://127.0.0.1:3000", + "http://127.0.0.1:5173", + "tauri://localhost", + ] + ) +``` + +- [ ] **Step 4: Update SecurityConfig defaults** + +In `src/openjarvis/core/config.py`, replace the `SecurityConfig` dataclass (lines 968-986): + +```python +@dataclass(slots=True) +class SecurityConfig: + """Security guardrails settings.""" + + enabled: bool = True + scan_input: bool = True + scan_output: bool = True + mode: str = "redact" # "redact" | "warn" | "block" + secret_scanner: bool = True + pii_scanner: bool = True + audit_log_path: str = str(DEFAULT_CONFIG_DIR / "audit.db") + enforce_tool_confirmation: bool = True + merkle_audit: bool = True + signing_key_path: str = "" + ssrf_protection: bool = True + rate_limit_enabled: bool = True + rate_limit_rpm: int = 60 + rate_limit_burst: int = 10 + local_engine_bypass: bool = False + local_tool_bypass: bool = False + profile: str = "" + vault_key_path: str = str(DEFAULT_CONFIG_DIR / ".vault_key") + capabilities: CapabilitiesConfig = field(default_factory=CapabilitiesConfig) +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_network_defaults.py -v` +Expected: All 7 tests PASS. + +- [ ] **Step 6: Commit** + +```bash +git add tests/security/test_network_defaults.py src/openjarvis/core/config.py +git commit -m "feat: secure server defaults — loopback binding, redact mode, rate limiting" +``` + +--- + +## Task 2: Network Exposure — Non-Loopback Auth Enforcement & CORS + +**Files:** +- Modify: `src/openjarvis/cli/serve.py:68-69` (bind resolution) and `337-352` (API key loading) +- Modify: `src/openjarvis/server/app.py:182-188` (CORS) +- Test: `tests/security/test_network_defaults.py` (add more tests) + +- [ ] **Step 1: Write failing tests for non-loopback auth and CORS** + +Append to `tests/security/test_network_defaults.py`: + +```python +import ipaddress + + +def _is_loopback(host: str) -> bool: + """Check if a host string is a loopback address.""" + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return host in ("localhost", "") + + +class TestNonLoopbackAuthEnforcement: + """Server must require API key when binding non-loopback.""" + + def test_loopback_allows_no_key(self) -> None: + """127.0.0.1 should not require an API key.""" + assert _is_loopback("127.0.0.1") + + def test_wildcard_is_not_loopback(self) -> None: + """0.0.0.0 should be treated as non-loopback.""" + assert not _is_loopback("0.0.0.0") + + def test_non_loopback_requires_key(self) -> None: + """Binding to 0.0.0.0 without API key should raise.""" + from openjarvis.server.auth_middleware import check_bind_safety + + try: + check_bind_safety("0.0.0.0", api_key="") + assert False, "Should have raised" + except SystemExit: + pass + + def test_non_loopback_with_key_ok(self) -> None: + """Binding to 0.0.0.0 with API key should succeed.""" + from openjarvis.server.auth_middleware import check_bind_safety + + # Should not raise + check_bind_safety("0.0.0.0", api_key="oj_sk_test123") + + +class TestCORSConfiguration: + """CORS should use configured origins, not wildcard.""" + + def test_create_app_uses_configured_origins(self) -> None: + """create_app should pass cors_origins from config, not '*'.""" + # This test verifies the integration — we check that the app + # responds with the correct Access-Control-Allow-Origin. + from unittest.mock import MagicMock + + from fastapi.testclient import TestClient + + from openjarvis.server.app import create_app + + mock_engine = MagicMock() + mock_engine.health.return_value = True + mock_engine.list_models.return_value = ["test-model"] + + app = create_app( + mock_engine, + "test-model", + cors_origins=["http://localhost:3000"], + ) + client = TestClient(app) + + # Request from allowed origin + resp = client.options( + "/health", + headers={ + "Origin": "http://localhost:3000", + "Access-Control-Request-Method": "GET", + }, + ) + assert resp.headers.get("access-control-allow-origin") == "http://localhost:3000" + + # Request from disallowed origin should not get CORS header + resp2 = client.options( + "/health", + headers={ + "Origin": "http://evil.com", + "Access-Control-Request-Method": "GET", + }, + ) + assert resp2.headers.get("access-control-allow-origin") != "http://evil.com" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_network_defaults.py::TestNonLoopbackAuthEnforcement -v` +Expected: FAIL — `check_bind_safety` does not exist yet. + +Run: `uv run pytest tests/security/test_network_defaults.py::TestCORSConfiguration -v` +Expected: FAIL — `create_app` does not accept `cors_origins` parameter. + +- [ ] **Step 3: Add `check_bind_safety` to auth_middleware.py** + +In `src/openjarvis/server/auth_middleware.py`, add after the `generate_api_key()` function (after line 58): + +```python +def check_bind_safety(host: str, *, api_key: str) -> None: + """Refuse to bind non-loopback without an API key. + + Raises ``SystemExit`` if *host* is not a loopback address and + *api_key* is empty. + """ + import ipaddress + import sys + + try: + is_loop = ipaddress.ip_address(host).is_loopback + except ValueError: + is_loop = host in ("localhost", "") + + if not is_loop and not api_key: + logger.error( + "Binding to %s requires OPENJARVIS_API_KEY to be set. " + "Run: jarvis auth generate-key", + host, + ) + sys.exit(1) +``` + +- [ ] **Step 4: Update `create_app` to accept and use `cors_origins`** + +In `src/openjarvis/server/app.py`, add `cors_origins: list[str] | None = None` to the `create_app()` signature (line 140). + +Replace the CORS middleware block (lines 182-188): + +```python + from fastapi.middleware.cors import CORSMiddleware + + _origins = cors_origins if cors_origins is not None else ["*"] + app.add_middleware( + CORSMiddleware, + allow_origins=_origins, + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + ) +``` + +- [ ] **Step 5: Wire `check_bind_safety` and `cors_origins` in `serve.py`** + +In `src/openjarvis/cli/serve.py`, after API key resolution (around line 352), add: + +```python + from openjarvis.server.auth_middleware import check_bind_safety + + check_bind_safety(bind_host, api_key=api_key) +``` + +In the `create_app()` call (around line 383), add the `cors_origins` kwarg: + +```python + app = create_app( + engine, + model_name, + # ... existing kwargs ... + cors_origins=config.server.cors_origins, + ) +``` + +- [ ] **Step 6: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_network_defaults.py -v` +Expected: All tests PASS. + +- [ ] **Step 7: Commit** + +```bash +git add src/openjarvis/server/auth_middleware.py src/openjarvis/server/app.py src/openjarvis/cli/serve.py tests/security/test_network_defaults.py +git commit -m "feat: enforce API key for non-loopback binding, restrict CORS origins" +``` + +--- + +## Task 3: Boundary Guard — Core Module + +**Files:** +- Create: `src/openjarvis/security/boundary.py` +- Test: `tests/security/test_boundary_guard.py` + +- [ ] **Step 1: Write failing tests for BoundaryGuard** + +```python +# tests/security/test_boundary_guard.py +"""Tests for BoundaryGuard — scanning at device exit points.""" + +from __future__ import annotations + +import pytest + +from openjarvis.core.types import ToolCall + + +class TestBoundaryGuardScanOutbound: + """scan_outbound should detect and redact secrets/PII.""" + + def test_redacts_openai_key(self) -> None: + from openjarvis.security.boundary import BoundaryGuard + + guard = BoundaryGuard(mode="redact") + text = "Use this key: sk-proj-abc123def456ghi789jkl012mno345pqr678stu" + result = guard.scan_outbound(text, destination="openai") + assert "sk-proj-" not in result + assert "[REDACTED" in result + + def test_redacts_aws_key(self) -> None: + from openjarvis.security.boundary import BoundaryGuard + + guard = BoundaryGuard(mode="redact") + text = "AWS key: AKIAIOSFODNN7EXAMPLE" + result = guard.scan_outbound(text, destination="openai") + assert "AKIAIOSFODNN7EXAMPLE" not in result + + def test_warn_mode_does_not_alter_text(self) -> None: + from openjarvis.security.boundary import BoundaryGuard + + guard = BoundaryGuard(mode="warn") + text = "Use this key: sk-proj-abc123def456ghi789jkl012mno345pqr678stu" + result = guard.scan_outbound(text, destination="openai") + assert result == text + + def test_block_mode_raises(self) -> None: + from openjarvis.security.boundary import BoundaryGuard, SecurityBlockError + + guard = BoundaryGuard(mode="block") + text = "Use this key: sk-proj-abc123def456ghi789jkl012mno345pqr678stu" + with pytest.raises(SecurityBlockError): + guard.scan_outbound(text, destination="openai") + + def test_clean_text_passes_through(self) -> None: + from openjarvis.security.boundary import BoundaryGuard + + guard = BoundaryGuard(mode="redact") + text = "Hello, how are you?" + result = guard.scan_outbound(text, destination="openai") + assert result == text + + +class TestBoundaryGuardCheckOutbound: + """check_outbound should redact secrets in tool call arguments.""" + + def test_redacts_tool_call_arguments(self) -> None: + from openjarvis.security.boundary import BoundaryGuard + + guard = BoundaryGuard(mode="redact") + tc = ToolCall( + id="test_1", + name="web_search", + arguments='{"query": "my key is sk-proj-abc123def456ghi789jkl012mno345pqr678stu"}', + ) + result = guard.check_outbound(tc) + assert "sk-proj-" not in result.arguments + assert result.id == "test_1" + assert result.name == "web_search" + + def test_clean_args_pass_through(self) -> None: + from openjarvis.security.boundary import BoundaryGuard + + guard = BoundaryGuard(mode="redact") + tc = ToolCall(id="test_2", name="web_search", arguments='{"query": "weather"}') + result = guard.check_outbound(tc) + assert result.arguments == tc.arguments + + def test_block_mode_raises_on_tool_call(self) -> None: + from openjarvis.security.boundary import BoundaryGuard, SecurityBlockError + + guard = BoundaryGuard(mode="block") + tc = ToolCall( + id="test_3", + name="web_search", + arguments='{"query": "AKIAIOSFODNN7EXAMPLE"}', + ) + with pytest.raises(SecurityBlockError): + guard.check_outbound(tc) + + +class TestBoundaryGuardDisabled: + """When disabled, BoundaryGuard should pass everything through.""" + + def test_disabled_passes_secrets_through(self) -> None: + from openjarvis.security.boundary import BoundaryGuard + + guard = BoundaryGuard(mode="redact", enabled=False) + text = "sk-proj-abc123def456ghi789jkl012mno345pqr678stu" + result = guard.scan_outbound(text, destination="openai") + assert result == text +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_boundary_guard.py -v` +Expected: FAIL — `openjarvis.security.boundary` does not exist. + +- [ ] **Step 3: Implement BoundaryGuard** + +```python +# src/openjarvis/security/boundary.py +"""BoundaryGuard — scans content at device exit points. + +Wraps SecretScanner and PIIScanner to redact, warn, or block +secrets and PII before data leaves the device via cloud engines +or external tool calls. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, replace +from typing import TYPE_CHECKING, List, Optional + +from openjarvis.core.types import ToolCall + +if TYPE_CHECKING: + from openjarvis.core.events import EventBus + from openjarvis.security.scanner import BaseScanner + +logger = logging.getLogger(__name__) + + +class SecurityBlockError(Exception): + """Raised when mode='block' and secrets/PII are detected.""" + + +@dataclass(slots=True) +class ScanFinding: + """A single finding from a boundary scan.""" + + pattern_name: str + destination: str + + +class BoundaryGuard: + """Scans outbound content for secrets and PII at device boundaries. + + Parameters + ---------- + mode: + Action on findings: ``"redact"`` replaces matches, + ``"warn"`` logs but passes through, ``"block"`` raises. + enabled: + Master switch. When ``False``, all content passes through. + bus: + Optional event bus for publishing SECURITY_ALERT events. + scanners: + Custom scanners. Defaults to SecretScanner + PIIScanner. + """ + + def __init__( + self, + mode: str = "redact", + *, + enabled: bool = True, + bus: Optional["EventBus"] = None, + scanners: Optional[List["BaseScanner"]] = None, + ) -> None: + self._mode = mode + self._enabled = enabled + self._bus = bus + if scanners is not None: + self._scanners = scanners + else: + self._scanners = self._default_scanners() + + @staticmethod + def _default_scanners() -> List["BaseScanner"]: + from openjarvis.security.scanner import PIIScanner, SecretScanner + + return [SecretScanner(), PIIScanner()] + + def scan_outbound(self, content: str, destination: str) -> str: + """Scan text before it leaves the device. + + Returns redacted text in ``"redact"`` mode, original text in + ``"warn"`` mode, or raises ``SecurityBlockError`` in ``"block"`` + mode when findings are detected. + """ + if not self._enabled or not content: + return content + + has_findings = False + redacted = content + for scanner in self._scanners: + result = scanner.scan(content) + if result.findings: + has_findings = True + if self._mode == "redact": + redacted = scanner.redact(redacted) + + if has_findings: + self._emit_alert(destination, content) + if self._mode == "block": + raise SecurityBlockError( + f"Secrets/PII detected in outbound content to {destination}" + ) + if self._mode == "warn": + logger.warning( + "Secrets/PII detected in outbound content to %s", destination + ) + return content + return redacted + + return content + + def check_outbound(self, tool_call: ToolCall) -> ToolCall: + """Scan tool call arguments before execution. + + Returns a new ToolCall with redacted arguments if needed. + """ + if not self._enabled or not tool_call.arguments: + return tool_call + + redacted_args = self.scan_outbound( + tool_call.arguments, destination=f"tool:{tool_call.name}" + ) + if redacted_args != tool_call.arguments: + return replace(tool_call, arguments=redacted_args) + return tool_call + + def _emit_alert(self, destination: str, content: str) -> None: + if self._bus is None: + return + try: + from openjarvis.core.events import EventType + + self._bus.publish( + EventType.SECURITY_ALERT, + { + "source": "boundary_guard", + "destination": destination, + "mode": self._mode, + "content_preview": content[:80], + }, + ) + except Exception: + logger.debug("Failed to emit security alert event", exc_info=True) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_boundary_guard.py -v` +Expected: All 9 tests PASS. + +- [ ] **Step 5: Commit** + +```bash +git add src/openjarvis/security/boundary.py tests/security/test_boundary_guard.py +git commit -m "feat: BoundaryGuard module — scan/redact/block at device exit points" +``` + +--- + +## Task 4: Boundary Guard — Engine & Tool Tagging + +**Files:** +- Modify: `src/openjarvis/engine/_stubs.py:48-120` (InferenceEngine ABC) +- Modify: `src/openjarvis/engine/cloud.py:211` (CloudEngine) +- Modify: `src/openjarvis/engine/litellm.py:16` (LiteLLMEngine) +- Modify: `src/openjarvis/tools/_stubs.py:46-74` (BaseTool ABC) +- Modify: `src/openjarvis/tools/web_search.py` (WebSearchTool) +- Modify: `src/openjarvis/tools/http_request.py` (HttpRequestTool) +- Modify: `src/openjarvis/tools/browser.py` (BrowserNavigateTool and others) +- Modify: `src/openjarvis/tools/browser_axtree.py` (BrowserAXTreeTool) +- Modify: `src/openjarvis/tools/channel_tools.py` (ChannelSendTool) +- Modify: `src/openjarvis/tools/image_tool.py` (ImageGenerateTool) +- Modify: `src/openjarvis/tools/audio_tool.py` (AudioTranscribeTool) +- Test: `tests/security/test_boundary_guard.py` (extend) + +- [ ] **Step 1: Write failing tests for engine and tool tagging** + +Append to `tests/security/test_boundary_guard.py`: + +```python +class TestEngineTagging: + """Cloud engines must have is_cloud=True, local engines is_cloud=False.""" + + def test_inference_engine_default_is_local(self) -> None: + from openjarvis.engine._stubs import InferenceEngine + + assert InferenceEngine.is_cloud is False + + def test_cloud_engine_is_cloud(self) -> None: + from openjarvis.engine.cloud import CloudEngine + + assert CloudEngine.is_cloud is True + + def test_litellm_engine_is_cloud(self) -> None: + from openjarvis.engine.litellm import LiteLLMEngine + + assert LiteLLMEngine.is_cloud is True + + def test_ollama_engine_is_local(self) -> None: + from openjarvis.engine.ollama import OllamaEngine + + assert OllamaEngine.is_cloud is False + + +class TestToolTagging: + """External tools must have is_local=False, local tools is_local=True.""" + + def test_base_tool_default_is_local(self) -> None: + from openjarvis.tools._stubs import BaseTool + + assert BaseTool.is_local is True + + def test_web_search_is_external(self) -> None: + from openjarvis.tools.web_search import WebSearchTool + + assert WebSearchTool.is_local is False + + def test_http_request_is_external(self) -> None: + from openjarvis.tools.http_request import HttpRequestTool + + assert HttpRequestTool.is_local is False + + def test_channel_send_is_external(self) -> None: + from openjarvis.tools.channel_tools import ChannelSendTool + + assert ChannelSendTool.is_local is False + + def test_think_tool_is_local(self) -> None: + from openjarvis.tools.think import ThinkTool + + assert ThinkTool.is_local is True + + def test_calculator_is_local(self) -> None: + from openjarvis.tools.calculator import CalculatorTool + + assert CalculatorTool.is_local is True +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_boundary_guard.py::TestEngineTagging -v` +Expected: FAIL — `InferenceEngine` has no `is_cloud` attribute. + +Run: `uv run pytest tests/security/test_boundary_guard.py::TestToolTagging -v` +Expected: FAIL — `BaseTool` has no `is_local` attribute. + +- [ ] **Step 3: Add `is_cloud` to InferenceEngine ABC** + +In `src/openjarvis/engine/_stubs.py`, add to the `InferenceEngine` class body (after `engine_id: str`, around line 55): + +```python + is_cloud: bool = False +``` + +- [ ] **Step 4: Set `is_cloud = True` on cloud engines** + +In `src/openjarvis/engine/cloud.py`, add to the `CloudEngine` class body (after `engine_id = "cloud"`, around line 216): + +```python + is_cloud = True +``` + +In `src/openjarvis/engine/litellm.py`, add to the `LiteLLMEngine` class body (after `engine_id = "litellm"`, around line 30): + +```python + is_cloud = True +``` + +- [ ] **Step 5: Add `is_local` to BaseTool ABC** + +In `src/openjarvis/tools/_stubs.py`, add to the `BaseTool` class body (after `tool_id: str`, around line 53): + +```python + is_local: bool = True +``` + +- [ ] **Step 6: Set `is_local = False` on external tools** + +In each of these files, add `is_local = False` to the class body (after `tool_id = "..."`): + +- `src/openjarvis/tools/web_search.py` — `WebSearchTool` +- `src/openjarvis/tools/http_request.py` — `HttpRequestTool` +- `src/openjarvis/tools/browser.py` — `BrowserNavigateTool`, `BrowserClickTool`, `BrowserTypeTool`, `BrowserScreenshotTool`, `BrowserExtractTool` +- `src/openjarvis/tools/browser_axtree.py` — `BrowserAXTreeTool` +- `src/openjarvis/tools/channel_tools.py` — `ChannelSendTool` +- `src/openjarvis/tools/image_tool.py` — `ImageGenerateTool` +- `src/openjarvis/tools/audio_tool.py` — `AudioTranscribeTool` + +Example for `WebSearchTool`: +```python +@ToolRegistry.register("web_search") +class WebSearchTool(BaseTool): + tool_id = "web_search" + is_local = False + # ... rest unchanged +``` + +- [ ] **Step 7: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_boundary_guard.py -v` +Expected: All tests PASS. + +- [ ] **Step 8: Commit** + +```bash +git add src/openjarvis/engine/_stubs.py src/openjarvis/engine/cloud.py src/openjarvis/engine/litellm.py src/openjarvis/tools/_stubs.py src/openjarvis/tools/web_search.py src/openjarvis/tools/http_request.py src/openjarvis/tools/browser.py src/openjarvis/tools/browser_axtree.py src/openjarvis/tools/channel_tools.py src/openjarvis/tools/image_tool.py src/openjarvis/tools/audio_tool.py tests/security/test_boundary_guard.py +git commit -m "feat: tag engines with is_cloud and tools with is_local for boundary scanning" +``` + +--- + +## Task 5: Boundary Guard — Wire into ToolExecutor and GuardrailsEngine + +**Files:** +- Modify: `src/openjarvis/tools/_stubs.py:112-266` (ToolExecutor.execute) +- Modify: `src/openjarvis/security/guardrails.py:20-70` (GuardrailsEngine) +- Modify: `src/openjarvis/system.py:18-52` (JarvisSystem) +- Test: `tests/security/test_boundary_guard.py` (extend) + +- [ ] **Step 1: Write failing tests for ToolExecutor integration** + +Append to `tests/security/test_boundary_guard.py`: + +```python +from unittest.mock import MagicMock + + +class TestToolExecutorBoundaryIntegration: + """ToolExecutor should use BoundaryGuard for external tool calls.""" + + def _make_executor(self, boundary_guard=None): + from openjarvis.tools._stubs import BaseTool, ToolExecutor, ToolSpec + + class FakeExternalTool(BaseTool): + tool_id = "fake_external" + is_local = False + + @property + def spec(self): + return ToolSpec( + name="fake_external", + description="test", + parameters={"type": "object", "properties": {"q": {"type": "string"}}}, + ) + + def execute(self, **params): + from openjarvis.core.types import ToolResult + + return ToolResult( + tool_name="fake_external", + content=f"result for {params.get('q', '')}", + success=True, + ) + + return ToolExecutor( + tools=[FakeExternalTool()], + boundary_guard=boundary_guard, + ) + + def test_external_tool_args_scanned(self) -> None: + from openjarvis.core.types import ToolCall + from openjarvis.security.boundary import BoundaryGuard + + guard = BoundaryGuard(mode="redact") + executor = self._make_executor(boundary_guard=guard) + + tc = ToolCall( + id="t1", + name="fake_external", + arguments='{"q": "my key is sk-proj-abc123def456ghi789jkl012mno345pqr678stu"}', + ) + result = executor.execute(tc) + # The tool should have received redacted args + assert "sk-proj-" not in result.content + + def test_no_guard_passes_through(self) -> None: + from openjarvis.core.types import ToolCall + + executor = self._make_executor(boundary_guard=None) + tc = ToolCall( + id="t2", + name="fake_external", + arguments='{"q": "sk-proj-abc123def456ghi789jkl012mno345pqr678stu"}', + ) + result = executor.execute(tc) + assert "sk-proj-" in result.content +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_boundary_guard.py::TestToolExecutorBoundaryIntegration -v` +Expected: FAIL — `ToolExecutor.__init__` does not accept `boundary_guard`. + +- [ ] **Step 3: Add BoundaryGuard to ToolExecutor** + +In `src/openjarvis/tools/_stubs.py`, modify the `ToolExecutor.__init__` method to accept a `boundary_guard` parameter. Add it as `self._boundary_guard = boundary_guard`. + +In the `execute()` method, after argument parsing succeeds (around where params are parsed from JSON, before the tool.execute call), add boundary checking for non-local tools: + +```python + # Boundary guard: scan external tool arguments + if ( + self._boundary_guard is not None + and not getattr(tool, "is_local", True) + ): + try: + tool_call = self._boundary_guard.check_outbound(tool_call) + # Re-parse arguments after potential redaction + params = json.loads(tool_call.arguments) if tool_call.arguments else {} + except Exception as exc: + return ToolResult( + tool_name=tool_call.name, + content=f"Security block: {exc}", + success=False, + ) +``` + +- [ ] **Step 4: Add BoundaryGuard to JarvisSystem** + +In `src/openjarvis/system.py`, add to the `JarvisSystem` dataclass (after `audit_logger`): + +```python + boundary_guard: Optional[Any] = None # BoundaryGuard +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_boundary_guard.py -v` +Expected: All tests PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/openjarvis/tools/_stubs.py src/openjarvis/system.py tests/security/test_boundary_guard.py +git commit -m "feat: wire BoundaryGuard into ToolExecutor for external tool scanning" +``` + +--- + +## Task 6: Webhook Fail-Closed Validation + +**Files:** +- Modify: `src/openjarvis/server/webhook_routes.py:31-45` (_validate_twilio_signature) +- Test: `tests/security/test_webhook_validation.py` + +- [ ] **Step 1: Write failing tests for fail-closed webhooks** + +```python +# tests/security/test_webhook_validation.py +"""Tests for webhook fail-closed validation (Section 3).""" + +from __future__ import annotations + +from unittest.mock import patch + + +class TestTwilioValidationFailClosed: + """Twilio validation must reject when SDK is unavailable.""" + + def test_missing_sdk_returns_false(self) -> None: + """When twilio is not installed, validation returns False.""" + from openjarvis.server.webhook_routes import _validate_twilio_signature + + with patch.dict("sys.modules", {"twilio": None, "twilio.request_validator": None}): + # Force re-import to hit the ImportError path + result = _validate_twilio_signature( + auth_token="test_token", + url="https://example.com/webhooks/twilio", + params={"Body": "hello"}, + signature="invalid", + ) + assert result is False + + def test_empty_auth_token_returns_false(self) -> None: + """When no auth token is configured, validation returns False.""" + from openjarvis.server.webhook_routes import _validate_twilio_signature + + result = _validate_twilio_signature( + auth_token="", + url="https://example.com/webhooks/twilio", + params={}, + signature="", + ) + assert result is False + + +class TestWebhookSecretEnforcement: + """Webhooks must return 503 when secrets are not configured.""" + + def test_twilio_webhook_503_without_token(self) -> None: + """POST /webhooks/twilio should return 503 if no auth_token configured.""" + from unittest.mock import MagicMock + + from fastapi.testclient import TestClient + + from openjarvis.server.app import create_app + + mock_engine = MagicMock() + mock_engine.health.return_value = True + mock_engine.list_models.return_value = ["test"] + + app = create_app( + mock_engine, + "test", + webhook_config={"twilio_auth_token": ""}, + ) + client = TestClient(app) + resp = client.post( + "/webhooks/twilio", + data={"From": "+1234567890", "Body": "hello"}, + ) + assert resp.status_code == 503 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_webhook_validation.py -v` +Expected: FAIL — `_validate_twilio_signature` returns `True` when SDK missing (not `False`). The 503 test may also fail since the current code doesn't enforce secret presence. + +- [ ] **Step 3: Fix `_validate_twilio_signature` to fail-closed** + +In `src/openjarvis/server/webhook_routes.py`, replace lines 31-45: + +```python +def _validate_twilio_signature( + auth_token: str, + url: str, + params: dict, + signature: str, +) -> bool: + """Validate Twilio webhook signature using the SDK. + + Fails closed: returns ``False`` if the SDK is not installed or + if no auth_token is configured. + """ + if not auth_token: + logger.error("Twilio auth token not configured — rejecting webhook") + return False + try: + from twilio.request_validator import RequestValidator + + validator = RequestValidator(auth_token) + return validator.validate(url, params, signature) + except ImportError: + logger.error( + "twilio SDK not installed — rejecting webhook. " + "Install it: pip install twilio" + ) + return False +``` + +- [ ] **Step 4: Add 503 response for unconfigured webhook secrets** + +In `src/openjarvis/server/webhook_routes.py`, in the Twilio webhook route handler, add an early check at the top of the handler (before signature validation): + +```python + if not twilio_auth_token: + return Response( + '{"detail": "Webhook not configured — set TWILIO_AUTH_TOKEN"}', + status_code=503, + media_type="application/json", + ) +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_webhook_validation.py -v` +Expected: All tests PASS. + +- [ ] **Step 6: Apply SendBlue HMAC enforcement** + +In `src/openjarvis/server/webhook_routes.py`, in the SendBlue webhook handler, update the secret validation block. The current code (around lines 234-238) only checks if `sb and sb.webhook_secret` — if no secret is configured, it accepts everything silently. Add a warning log when no secret is configured: + +```python + if sb and sb.webhook_secret: + header_secret = request.headers.get("x-sendblue-secret", "") + if header_secret != sb.webhook_secret: + return Response("Invalid secret", status_code=403) + elif sb: + logger.warning( + "SendBlue webhook received without secret verification. " + "Set webhook_secret for HMAC validation." + ) +``` + +- [ ] **Step 7: Commit** + +```bash +git add src/openjarvis/server/webhook_routes.py tests/security/test_webhook_validation.py +git commit -m "fix: fail-closed webhook validation — reject when SDK missing or secret unconfigured" +``` + +--- + +## Task 7: File Permissions — Secure Helpers + +**Files:** +- Create: `src/openjarvis/security/file_utils.py` +- Test: `tests/security/test_file_permissions.py` + +- [ ] **Step 1: Write failing tests for secure file helpers** + +```python +# tests/security/test_file_permissions.py +"""Tests for secure file creation helpers (Section 4).""" + +from __future__ import annotations + +import os +import stat +import tempfile +from pathlib import Path + + +class TestSecureMkdir: + """secure_mkdir should create directories with 0o700.""" + + def test_creates_directory_with_700(self) -> None: + from openjarvis.security.file_utils import secure_mkdir + + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "secure_dir" + result = secure_mkdir(target) + assert result.is_dir() + mode = stat.S_IMODE(os.stat(target).st_mode) + assert mode == 0o700 + + def test_creates_parent_directories(self) -> None: + from openjarvis.security.file_utils import secure_mkdir + + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "a" / "b" / "c" + result = secure_mkdir(target) + assert result.is_dir() + + def test_existing_directory_gets_permission_fix(self) -> None: + from openjarvis.security.file_utils import secure_mkdir + + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "existing" + target.mkdir(mode=0o755) + secure_mkdir(target) + mode = stat.S_IMODE(os.stat(target).st_mode) + assert mode == 0o700 + + +class TestSecureCreate: + """secure_create should create files with 0o600.""" + + def test_creates_file_with_600(self) -> None: + from openjarvis.security.file_utils import secure_create + + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "secure_file.db" + result = secure_create(target) + assert result.exists() + mode = stat.S_IMODE(os.stat(target).st_mode) + assert mode == 0o600 + + def test_existing_file_gets_permission_fix(self) -> None: + from openjarvis.security.file_utils import secure_create + + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "existing.db" + target.write_text("data") + os.chmod(target, 0o644) + secure_create(target) + mode = stat.S_IMODE(os.stat(target).st_mode) + assert mode == 0o600 + + def test_creates_parent_directory_with_700(self) -> None: + from openjarvis.security.file_utils import secure_create + + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "sub" / "file.db" + secure_create(target) + parent_mode = stat.S_IMODE(os.stat(target.parent).st_mode) + assert parent_mode == 0o700 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_file_permissions.py -v` +Expected: FAIL — `openjarvis.security.file_utils` does not exist. + +- [ ] **Step 3: Implement file_utils.py** + +```python +# src/openjarvis/security/file_utils.py +"""Secure file and directory creation helpers. + +All OpenJarvis data files under ``~/.openjarvis/`` should be created +through these helpers to ensure consistent, restrictive permissions. +""" + +from __future__ import annotations + +import os +from pathlib import Path + + +def secure_mkdir(path: Path, mode: int = 0o700) -> Path: + """Create a directory with restrictive permissions. + + Creates parent directories as needed, then sets *mode* on the + target directory (even if it already exists). + """ + path.mkdir(parents=True, exist_ok=True) + os.chmod(path, mode) + return path + + +def secure_create(path: Path, mode: int = 0o600) -> Path: + """Ensure a file exists with restrictive permissions. + + Creates the parent directory with ``0o700`` if needed, touches the + file if it doesn't exist, and sets *mode* on it. + """ + secure_mkdir(path.parent, mode=0o700) + if not path.exists(): + path.touch() + os.chmod(path, mode) + return path +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_file_permissions.py -v` +Expected: All 6 tests PASS. + +- [ ] **Step 5: Commit** + +```bash +git add src/openjarvis/security/file_utils.py tests/security/test_file_permissions.py +git commit -m "feat: secure_mkdir and secure_create helpers for restrictive file permissions" +``` + +--- + +## Task 8: File Permissions — Apply to All Database Paths + +**Files:** +- Modify: `src/openjarvis/tools/storage/sqlite.py:33-45` +- Modify: `src/openjarvis/server/session_store.py:23-29` +- Modify: `src/openjarvis/traces/store.py:82-94` +- Modify: `src/openjarvis/security/audit.py:33-56` +- Modify: `src/openjarvis/connectors/store.py:118-131` +- Modify: `src/openjarvis/connectors/attachment_store.py:50-72` +- Modify: `src/openjarvis/cli/log_config.py:57-59` + +- [ ] **Step 1: Protect ~/.openjarvis/ parent directory in config.py** + +In `src/openjarvis/core/config.py`, after the `DEFAULT_CONFIG_DIR` definition (line 28), add: + +```python +# Ensure the config directory exists with restrictive permissions on first access. +# This is the single biggest protection — even if individual files miss chmod, +# the 0o700 parent blocks other users. +def _ensure_config_dir() -> Path: + from openjarvis.security.file_utils import secure_mkdir + return secure_mkdir(DEFAULT_CONFIG_DIR) +``` + +Then call `_ensure_config_dir()` at the start of `load_config()` (around line 1310): + +```python + _ensure_config_dir() + hw = detect_hardware() +``` + +- [ ] **Step 2: Update tools/storage/sqlite.py** + +In `src/openjarvis/tools/storage/sqlite.py`, in the `__init__` method (around line 33), after resolving `db_path`, add: + +```python + if self._db_path != ":memory:": + from openjarvis.security.file_utils import secure_create + secure_create(Path(self._db_path)) +``` + +- [ ] **Step 3: Update session_store.py** + +In `src/openjarvis/server/session_store.py`, replace the directory creation (line 26): + +```python + # Before: + # Path(db_path).parent.mkdir(parents=True, exist_ok=True) + + # After: + from openjarvis.security.file_utils import secure_create + secure_create(Path(db_path)) +``` + +- [ ] **Step 4: Update traces/store.py** + +In `src/openjarvis/traces/store.py`, add before the `sqlite3.connect` call (around line 88): + +```python + from openjarvis.security.file_utils import secure_create + if self._db_path != ":memory:": + secure_create(Path(self._db_path)) +``` + +- [ ] **Step 5: Update security/audit.py** + +In `src/openjarvis/security/audit.py`, replace the directory creation (line 39): + +```python + # Before: + # self._db_path.parent.mkdir(parents=True, exist_ok=True) + + # After: + from openjarvis.security.file_utils import secure_create + secure_create(self._db_path) +``` + +- [ ] **Step 6: Update connectors/store.py** + +In `src/openjarvis/connectors/store.py`, replace the directory creation (lines 127-128): + +```python + # Before: + # if self._db_path != ":memory:": + # Path(self._db_path).parent.mkdir(parents=True, exist_ok=True) + + # After: + if self._db_path != ":memory:": + from openjarvis.security.file_utils import secure_create + secure_create(Path(self._db_path)) +``` + +- [ ] **Step 7: Update connectors/attachment_store.py** + +In `src/openjarvis/connectors/attachment_store.py`, replace directory creation (line 57-58): + +```python + # Before: + # self._base_dir.mkdir(parents=True, exist_ok=True) + + # After: + from openjarvis.security.file_utils import secure_mkdir + secure_mkdir(self._base_dir) +``` + +And in the `store()` method, after writing blob files (around line 97): + +```python + blob_path.write_bytes(content) + os.chmod(blob_path, 0o600) +``` + +- [ ] **Step 8: Update cli/log_config.py** + +In `src/openjarvis/cli/log_config.py`, replace directory creation (lines 57-58): + +```python + # Before: + # log_dir = Path.home() / ".openjarvis" + # log_dir.mkdir(parents=True, exist_ok=True) + + # After: + from openjarvis.security.file_utils import secure_mkdir + log_dir = Path.home() / ".openjarvis" + secure_mkdir(log_dir) +``` + +- [ ] **Step 9: Run existing tests to check for regressions** + +Run: `uv run pytest tests/ -v -m "not live and not cloud" --timeout=30 -x` +Expected: No regressions. File creation still works, just with tighter permissions. + +- [ ] **Step 10: Commit** + +```bash +git add src/openjarvis/core/config.py src/openjarvis/tools/storage/sqlite.py src/openjarvis/server/session_store.py src/openjarvis/traces/store.py src/openjarvis/security/audit.py src/openjarvis/connectors/store.py src/openjarvis/connectors/attachment_store.py src/openjarvis/cli/log_config.py +git commit -m "fix: enforce 0o600/0o700 permissions on all database and data files" +``` + +--- + +> **Deferred to follow-up:** Section 4.4 (Optional SQLCipher encryption) and Section 5.4 (vault promotion tip in auth setup wizard) are lower-priority items that can be implemented in a subsequent PR without blocking the core hardening. + +--- + +## Task 9: Log Sanitization — SanitizingFormatter + +**Files:** +- Modify: `src/openjarvis/cli/log_config.py` +- Test: `tests/security/test_log_sanitization.py` + +- [ ] **Step 1: Write failing tests for SanitizingFormatter** + +```python +# tests/security/test_log_sanitization.py +"""Tests for log sanitization (Section 5).""" + +from __future__ import annotations + +import logging + + +class TestSanitizingFormatter: + """SanitizingFormatter should redact secrets in log messages.""" + + def test_redacts_openai_key(self) -> None: + from openjarvis.cli.log_config import SanitizingFormatter + + fmt = SanitizingFormatter("%(message)s") + record = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="", + lineno=0, + msg="Key is sk-proj-abc123def456ghi789jkl012mno345pqr678stu", + args=(), + exc_info=None, + ) + result = fmt.format(record) + assert "sk-proj-" not in result + assert "[REDACTED" in result + + def test_redacts_aws_key(self) -> None: + from openjarvis.cli.log_config import SanitizingFormatter + + fmt = SanitizingFormatter("%(message)s") + record = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="", + lineno=0, + msg="AWS: AKIAIOSFODNN7EXAMPLE", + args=(), + exc_info=None, + ) + result = fmt.format(record) + assert "AKIAIOSFODNN7EXAMPLE" not in result + + def test_clean_message_unchanged(self) -> None: + from openjarvis.cli.log_config import SanitizingFormatter + + fmt = SanitizingFormatter("%(message)s") + record = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="", + lineno=0, + msg="Server started on port 8000", + args=(), + exc_info=None, + ) + result = fmt.format(record) + assert result == "Server started on port 8000" + + def test_redacts_slack_token(self) -> None: + from openjarvis.cli.log_config import SanitizingFormatter + + fmt = SanitizingFormatter("%(message)s") + record = logging.LogRecord( + name="test", + level=logging.INFO, + pathname="", + lineno=0, + msg="Token: xoxb-1234-5678-abcdefghij", + args=(), + exc_info=None, + ) + result = fmt.format(record) + assert "xoxb-" not in result +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_log_sanitization.py -v` +Expected: FAIL — `SanitizingFormatter` does not exist in `log_config.py`. + +- [ ] **Step 3: Implement SanitizingFormatter** + +In `src/openjarvis/cli/log_config.py`, add at the top of the file (after imports): + +```python +from openjarvis.security.credential_stripper import CredentialStripper + +_stripper = CredentialStripper() + + +class SanitizingFormatter(logging.Formatter): + """Formatter that redacts credentials from log messages.""" + + def format(self, record: logging.LogRecord) -> str: + msg = super().format(record) + return _stripper.strip(msg) +``` + +Then update `setup_logging()` to use `SanitizingFormatter` instead of `logging.Formatter` for both the console handler and file handler. + +Replace the console formatter line: +```python + # Before: + # fmt = logging.Formatter("%(levelname)s %(name)s: %(message)s") + + # After: + fmt = SanitizingFormatter("%(levelname)s %(name)s: %(message)s") +``` + +Replace the file formatter line: +```python + # Before: + # file_fmt = logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s") + + # After: + file_fmt = SanitizingFormatter("%(asctime)s %(levelname)s %(name)s: %(message)s") +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_log_sanitization.py -v` +Expected: All 4 tests PASS. + +- [ ] **Step 5: Commit** + +```bash +git add src/openjarvis/cli/log_config.py tests/security/test_log_sanitization.py +git commit -m "feat: SanitizingFormatter — auto-redact credentials in all log output" +``` + +--- + +## Task 10: Scoped Credential Access + +**Files:** +- Modify: `src/openjarvis/core/credentials.py` +- Test: `tests/security/test_log_sanitization.py` (extend) + +- [ ] **Step 1: Write failing tests for scoped credential access** + +Append to `tests/security/test_log_sanitization.py`: + +```python +import os +import tempfile +from pathlib import Path + + +class TestScopedCredentialAccess: + """get_tool_credential should return values without polluting os.environ.""" + + def test_returns_credential_value(self) -> None: + from openjarvis.core.credentials import get_tool_credential + + with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f: + f.write('[slack]\nSLACK_BOT_TOKEN = "xoxb-test-token"\n') + f.flush() + result = get_tool_credential( + "slack", "SLACK_BOT_TOKEN", path=Path(f.name) + ) + assert result == "xoxb-test-token" + # Must NOT have polluted os.environ + assert os.environ.get("SLACK_BOT_TOKEN") != "xoxb-test-token" + os.unlink(f.name) + + def test_returns_none_for_missing(self) -> None: + from openjarvis.core.credentials import get_tool_credential + + with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as f: + f.write("[slack]\n") + f.flush() + result = get_tool_credential( + "slack", "SLACK_BOT_TOKEN", path=Path(f.name) + ) + assert result is None + os.unlink(f.name) + + def test_returns_none_for_missing_file(self) -> None: + from openjarvis.core.credentials import get_tool_credential + + result = get_tool_credential( + "slack", "SLACK_BOT_TOKEN", path=Path("/nonexistent/file.toml") + ) + assert result is None +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_log_sanitization.py::TestScopedCredentialAccess -v` +Expected: FAIL — `get_tool_credential` does not exist. + +- [ ] **Step 3: Implement `get_tool_credential`** + +In `src/openjarvis/core/credentials.py`, add after `inject_credentials()`: + +```python +def get_tool_credential( + tool_name: str, + key: str, + *, + path: Path | None = None, +) -> str | None: + """Read a single credential without polluting ``os.environ``. + + Falls back to ``os.environ`` if the key is not in credentials.toml, + for backward compatibility with Docker env var workflows. + """ + creds = load_credentials(path=path) + tool_creds = creds.get(tool_name, {}) + value = tool_creds.get(key) + if value is not None: + return value + # Fallback to env var for backward compat + return os.environ.get(key) or None +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_log_sanitization.py -v` +Expected: All tests PASS. + +- [ ] **Step 5: Commit** + +```bash +git add src/openjarvis/core/credentials.py tests/security/test_log_sanitization.py +git commit -m "feat: get_tool_credential — scoped credential access without env pollution" +``` + +--- + +## Task 11: Startup Credential Audit & Non-Loopback CORS Warning + +**Files:** +- Modify: `src/openjarvis/cli/serve.py` + +- [ ] **Step 1: Add startup credential audit log** + +In `src/openjarvis/cli/serve.py`, after credential injection (after the `inject_credentials()` call if present, or after API key resolution around line 352), add: + +```python + # Log credential status at startup + from openjarvis.core.credentials import get_credential_status, TOOL_CREDENTIALS + + _cred_parts = [] + for _tool_name in sorted(TOOL_CREDENTIALS): + _status = get_credential_status(_tool_name) + _set = sum(1 for v in _status.values() if v) + _total = len(_status) + if _set > 0: + _cred_parts.append(f"{_tool_name}: {_set}/{_total} keys") + if _cred_parts: + logger.info("Credentials loaded — %s", ", ".join(_cred_parts)) +``` + +- [ ] **Step 2: Add non-loopback CORS wildcard warning** + +In `src/openjarvis/cli/serve.py`, before the `uvicorn.run()` call (around line 408), add: + +```python + # Warn about wildcard CORS on non-loopback + import ipaddress as _ipa + + try: + _is_loop = _ipa.ip_address(bind_host).is_loopback + except ValueError: + _is_loop = bind_host in ("localhost", "") + + if not _is_loop and "*" in config.server.cors_origins: + console.print( + "[yellow bold]WARNING:[/yellow bold] Wildcard CORS with credentials " + "enabled on non-loopback interface. This allows any website to make " + "authenticated requests to your instance." + ) +``` + +- [ ] **Step 3: Run full test suite to check for regressions** + +Run: `uv run pytest tests/ -v -m "not live and not cloud" --timeout=30 -x` +Expected: No regressions. + +- [ ] **Step 4: Commit** + +```bash +git add src/openjarvis/cli/serve.py +git commit -m "feat: startup credential audit log and CORS wildcard warning" +``` + +--- + +## Task 12: Security Headers — CSP for Docs + +**Files:** +- Modify: `src/openjarvis/server/middleware.py:33-51` + +- [ ] **Step 1: Add CSP header to SecurityHeadersMiddleware** + +In `src/openjarvis/server/middleware.py`, add the CSP header inside the `dispatch` method, after the existing headers (around line 50): + +```python + response.headers["Content-Security-Policy"] = "default-src 'self'" +``` + +Also add it to the `SECURITY_HEADERS` dict: + +```python +SECURITY_HEADERS = { + "X-Content-Type-Options": "nosniff", + "X-Frame-Options": "DENY", + "X-XSS-Protection": "1; mode=block", + "Strict-Transport-Security": "max-age=31536000; includeSubDomains", + "Referrer-Policy": "strict-origin-when-cross-origin", + "Permissions-Policy": "camera=(), microphone=(), geolocation=()", + "Content-Security-Policy": "default-src 'self'", +} +``` + +- [ ] **Step 2: Run existing middleware tests** + +Run: `uv run pytest tests/server/test_middleware.py -v` +Expected: May need to update `test_headers_dict` to include the new CSP header. If it fails, update the test's `expected_keys` set to include `"Content-Security-Policy"`. + +- [ ] **Step 3: Commit** + +```bash +git add src/openjarvis/server/middleware.py +git commit -m "feat: add Content-Security-Policy header to API responses" +``` + +--- + +## Task 13: Security Profiles + +**Files:** +- Modify: `src/openjarvis/core/config.py:1308-1368` (load_config) +- Test: `tests/security/test_security_profiles.py` + +- [ ] **Step 1: Write failing tests for security profiles** + +```python +# tests/security/test_security_profiles.py +"""Tests for security profile expansion (Section 7).""" + +from __future__ import annotations + + +# Profile definitions — must match the implementation +_PROFILES = { + "personal": { + "host": "127.0.0.1", + "mode": "redact", + "rate_limit_enabled": True, + "local_engine_bypass": False, + "local_tool_bypass": False, + }, + "shared": { + "host": "127.0.0.1", + "mode": "redact", + "rate_limit_enabled": True, + "local_engine_bypass": False, + "local_tool_bypass": False, + }, + "server": { + "host": "0.0.0.0", + "mode": "block", + "rate_limit_enabled": True, + "rate_limit_rpm": 30, + "rate_limit_burst": 5, + "local_engine_bypass": False, + "local_tool_bypass": False, + }, +} + + +class TestProfileExpansion: + """Profiles should pre-fill security and server config fields.""" + + def test_personal_profile_sets_redact(self) -> None: + from openjarvis.core.config import SecurityConfig, apply_security_profile + + cfg = SecurityConfig(profile="personal") + server_cfg = None + apply_security_profile(cfg, server_cfg) + assert cfg.mode == "redact" + assert cfg.rate_limit_enabled is True + + def test_server_profile_sets_block(self) -> None: + from openjarvis.core.config import SecurityConfig, ServerConfig, apply_security_profile + + cfg = SecurityConfig(profile="server") + server_cfg = ServerConfig() + apply_security_profile(cfg, server_cfg) + assert cfg.mode == "block" + assert cfg.rate_limit_rpm == 30 + assert cfg.rate_limit_burst == 5 + assert server_cfg.host == "0.0.0.0" + + def test_explicit_override_beats_profile(self) -> None: + """User-set values in config.toml should override profile defaults.""" + from openjarvis.core.config import SecurityConfig, apply_security_profile + + # Simulate: user set profile=server but also mode=warn + cfg = SecurityConfig(profile="server", mode="warn") + apply_security_profile(cfg, None, overrides={"mode"}) + # mode should stay "warn" because user explicitly set it + assert cfg.mode == "warn" + + def test_empty_profile_is_noop(self) -> None: + from openjarvis.core.config import SecurityConfig, apply_security_profile + + cfg = SecurityConfig() + original_mode = cfg.mode + apply_security_profile(cfg, None) + assert cfg.mode == original_mode + + def test_unknown_profile_raises(self) -> None: + import pytest + + from openjarvis.core.config import SecurityConfig, apply_security_profile + + cfg = SecurityConfig(profile="nonexistent") + with pytest.raises(ValueError, match="Unknown security profile"): + apply_security_profile(cfg, None) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/security/test_security_profiles.py -v` +Expected: FAIL — `apply_security_profile` does not exist. + +- [ ] **Step 3: Implement profile expansion** + +In `src/openjarvis/core/config.py`, add the profile expansion function (after the `SecurityConfig` dataclass): + +```python +_SECURITY_PROFILES: dict[str, dict[str, Any]] = { + "personal": { + "security": { + "mode": "redact", + "rate_limit_enabled": True, + "local_engine_bypass": False, + "local_tool_bypass": False, + }, + "server": { + "host": "127.0.0.1", + }, + }, + "shared": { + "security": { + "mode": "redact", + "rate_limit_enabled": True, + "local_engine_bypass": False, + "local_tool_bypass": False, + }, + "server": { + "host": "127.0.0.1", + }, + }, + "server": { + "security": { + "mode": "block", + "rate_limit_enabled": True, + "rate_limit_rpm": 30, + "rate_limit_burst": 5, + "local_engine_bypass": False, + "local_tool_bypass": False, + }, + "server": { + "host": "0.0.0.0", + }, + }, +} + + +def apply_security_profile( + security_cfg: SecurityConfig, + server_cfg: ServerConfig | None, + *, + overrides: set[str] | None = None, +) -> None: + """Expand a named security profile into config fields. + + Fields in *overrides* (explicitly set by the user in TOML) are + not overwritten by the profile. + """ + profile = security_cfg.profile + if not profile: + return + + if profile not in _SECURITY_PROFILES: + raise ValueError( + f"Unknown security profile '{profile}'. " + f"Valid profiles: {', '.join(_SECURITY_PROFILES)}" + ) + + _overrides = overrides or set() + pdef = _SECURITY_PROFILES[profile] + + for key, value in pdef.get("security", {}).items(): + if key not in _overrides and hasattr(security_cfg, key): + setattr(security_cfg, key, value) + + if server_cfg is not None: + for key, value in pdef.get("server", {}).items(): + if key not in _overrides and hasattr(server_cfg, key): + setattr(server_cfg, key, value) +``` + +- [ ] **Step 4: Hook profile expansion into load_config()** + +In `src/openjarvis/core/config.py`, in the `load_config()` function, after all TOML sections have been applied (after the `for section_name in top_sections` loop, around line 1362), add: + +```python + # Expand security profile (user TOML overrides take precedence) + _user_security_keys = set(data.get("security", {}).keys()) + apply_security_profile(cfg.security, cfg.server, overrides=_user_security_keys) +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/security/test_security_profiles.py -v` +Expected: All 5 tests PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/openjarvis/core/config.py tests/security/test_security_profiles.py +git commit -m "feat: security profiles — personal, shared, server presets with user overrides" +``` + +--- + +## Task 14: Doctor Security Check + +**Files:** +- Modify: `src/openjarvis/cli/doctor_cmd.py:267-278` (_run_all_checks) + +- [ ] **Step 1: Add security profile check to doctor** + +In `src/openjarvis/cli/doctor_cmd.py`, add a new check function: + +```python +def _check_security_profile() -> CheckResult: + """Check if a security profile is configured.""" + try: + from openjarvis.core.config import load_config + + config = load_config() + if config.security.profile: + return CheckResult( + name="Security profile", + status="ok", + message=f"Profile '{config.security.profile}' active", + ) + return CheckResult( + name="Security profile", + status="warn", + message="No security profile set", + details="Recommended: add security.profile = 'personal' to config.toml", + ) + except Exception as exc: + return CheckResult( + name="Security profile", + status="fail", + message=f"Could not check: {exc}", + ) +``` + +Add `checks.append(_check_security_profile())` to `_run_all_checks()`. + +- [ ] **Step 2: Run doctor to verify** + +Run: `uv run jarvis doctor` +Expected: Shows a "Security profile" row with a warning suggesting `security.profile = 'personal'`. + +- [ ] **Step 3: Commit** + +```bash +git add src/openjarvis/cli/doctor_cmd.py +git commit -m "feat: jarvis doctor checks for security profile configuration" +``` + +--- + +## Task 15: Integration Verification + +**Files:** +- All modified files from Tasks 1-14 + +- [ ] **Step 1: Run full test suite** + +Run: `uv run pytest tests/ -v -m "not live and not cloud" --timeout=60` +Expected: All tests pass. No regressions. + +- [ ] **Step 2: Run linting** + +Run: `uv run ruff check src/ tests/` +Run: `uv run ruff format --check src/ tests/` +Expected: No lint errors, no format issues. + +- [ ] **Step 3: Manual smoke test — server binding** + +Run: `uv run jarvis serve --host 127.0.0.1 --port 8000` +Expected: Server starts on `127.0.0.1:8000`. Not accessible from other machines on the network. + +- [ ] **Step 4: Manual smoke test — non-loopback rejection** + +Run: `uv run jarvis serve --host 0.0.0.0` +Expected: Server refuses to start with error: `Binding to 0.0.0.0 requires OPENJARVIS_API_KEY to be set.` + +- [ ] **Step 5: Manual smoke test — doctor** + +Run: `uv run jarvis doctor` +Expected: Security profile check appears with a warning or OK status. + +- [ ] **Step 6: Final commit (if any lint fixes needed)** + +```bash +git add -u +git commit -m "fix: lint and format fixes for security hardening" +```