admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-27 21:05:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: enhance security scan with DNS check, JSON output, Rich UI, and API endpoint

Browse Source 
Incorporates the useful new features from PR #135 (by @gridworks) on top
of the existing PrivacyScanner implementation:

- Add DNS configuration check (macOS, via scutil --dns)
- Add --json flag to `jarvis scan` for machine-readable output
- Add --no-scan flag to `jarvis init` to skip the post-init audit
- Expand remote-access process list (ngrok, tailscaled, cloudflared, ZeroTier)
- Upgrade `jarvis scan` output from plain text to Rich table
- Add GET /v1/security/scan API endpoint
- Add tests for all new features (30 tests, all passing)

Closes #133

Co-Authored-By: gridworks <5502067+gridworks@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
krypticmouse
2026-03-27 01:00:38 +00:00
co-authored by gridworks Claude Opus 4.6
parent 7f336d7679
commit 4d05475ec4
4 changed files with 434 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openjarvis/cli/init_cmd.py
+10 -1
View File
@@ -236,6 +236,13 @@ def _do_download(engine: str, model: str, spec, console: Console) -> None:
@click.option(
"--no-download", is_flag=True, default=False, help="Skip the model download prompt."
)
@click.option(
"--no-scan",
"skip_scan",
is_flag=True,
default=False,
help="Skip the post-init security environment audit.",
)
@click.option(
"--host",
default=None,
@@ -247,6 +254,7 @@ def init(
full_config: bool = False,
engine: Optional[str] = None,
no_download: bool = False,
skip_scan: bool = False,
host: Optional[str] = None,
) -> None:
"""Detect hardware and generate ~/.openjarvis/config.toml."""
@@ -405,7 +413,8 @@ def init(
if click.confirm(prompt, default=True):
_do_download(selected_engine, model, spec, console)
_quick_privacy_check(console)
if not skip_scan:
_quick_privacy_check(console)
console.print()
console.print(
Panel(
src/openjarvis/cli/scan_cmd.py
+163 -27
View File
@@ -19,11 +19,24 @@ _CLOUD_SYNC_PROCS = ["Dropbox", "OneDrive", "Google Drive", "iCloudDrive"]
# Screen-recording / remote-access processes (macOS).
_SCREEN_RECORDING_PROCS = [
"TeamViewer", "AnyDesk", "ScreenConnect", "vncviewer", "Vine"
"TeamViewer",
"AnyDesk",
"ScreenConnect",
"vncviewer",
"Vine",
]
# Remote-access processes (Linux).
_REMOTE_ACCESS_PROCS = ["xrdp", "x11vnc", "vncserver", "AnyDesk"]
# Remote-access processes (cross-platform).
_REMOTE_ACCESS_PROCS = [
"xrdp",
"x11vnc",
"vncserver",
"AnyDesk",
"ngrok",
"tailscaled",
"cloudflared",
"ZeroTier",
]
# ---------------------------------------------------------------------------
@@ -295,12 +308,83 @@ class PrivacyScanner:
)
def check_remote_access(self) -> ScanResult:
"""Check for running remote-access processes (Linux)."""
"""Check for running remote-access / tunneling processes."""
return self._check_processes(
names=_REMOTE_ACCESS_PROCS,
check_name="Remote Access",
warn_msg="{name} is running — system may be accessible remotely.",
platform="linux",
platform="all",
)
def check_dns(self) -> ScanResult:
"""Check DNS configuration for encrypted resolvers (macOS)."""
if sys.platform != "darwin":
return ScanResult(
name="DNS Configuration",
status="skip",
message="DNS check not yet implemented for this platform.",
platform="darwin",
)
try:
proc = self._run(["scutil", "--dns"])
output = proc.stdout
except Exception:
return ScanResult(
name="DNS Configuration",
status="skip",
message="scutil command not available.",
platform="darwin",
)
_DOH_INDICATORS = [
"dns-over-https",
"dns-over-tls",
"encrypted",
"doh",
"dot",
]
if any(ind in output.lower() for ind in _DOH_INDICATORS):
return ScanResult(
name="DNS Configuration",
status="ok",
message="Encrypted DNS (DoH/DoT) appears to be active.",
platform="darwin",
)
# Extract nameserver IPs
nameservers: list[str] = []
for line in output.splitlines():
stripped = line.strip()
if stripped.startswith("nameserver["):
parts = stripped.split(":", 1)
if len(parts) == 2:
nameservers.append(parts[1].strip())
if not nameservers:
return ScanResult(
name="DNS Configuration",
status="ok",
message="Could not parse DNS nameservers from scutil output.",
platform="darwin",
)
_PLAIN_DNS = {"8.8.8.8", "8.8.4.4", "1.1.1.1", "1.0.0.1", "9.9.9.9"}
plain = [ns for ns in nameservers if ns in _PLAIN_DNS]
if plain:
return ScanResult(
name="DNS Configuration",
status="warn",
message=(
f"Plain DNS in use ({', '.join(plain)}). "
"Queries may be visible to your ISP or resolver."
),
platform="darwin",
)
return ScanResult(
name="DNS Configuration",
status="ok",
message=f"DNS resolvers: {', '.join(nameservers[:3])}.",
platform="darwin",
)
# -- Orchestration -------------------------------------------------------
@@ -315,6 +399,7 @@ class PrivacyScanner:
self.check_luks,
self.check_screen_recording,
self.check_remote_access,
self.check_dns,
]
def run_all(self) -> list[ScanResult]:
@@ -350,37 +435,88 @@ class PrivacyScanner:
# CLI command
# ---------------------------------------------------------------------------
_STATUS_ICONS = {"ok": "✓", "warn": "!", "fail": "✗", "skip": "-"}
_RICH_ICONS = {
"ok": "[green]\u2713[/green]",
"warn": "[yellow]![/yellow]",
"fail": "[red]\u2717[/red]",
"skip": "[dim]-[/dim]",
}
def _render_results(results: List[ScanResult]) -> None:
"""Render scan results as a Rich table."""
from rich.console import Console
from rich.table import Table
console = Console()
console.print()
console.print("[bold]OpenJarvis Security Scan[/bold]")
console.print()
table = Table(show_header=True, header_style="bold", show_lines=True)
table.add_column("", width=3, justify="center")
table.add_column("Check")
table.add_column("Finding")
for r in results:
icon = _RICH_ICONS.get(r.status, "?")
style = {"ok": "green", "warn": "yellow", "fail": "red"}.get(r.status, "white")
table.add_row(icon, r.name, f"[{style}]{r.message}[/{style}]")
console.print(table)
ok_count = sum(1 for r in results if r.status == "ok")
warn_count = sum(1 for r in results if r.status == "warn")
fail_count = sum(1 for r in results if r.status == "fail")
console.print()
parts = [f"[green]{ok_count} ok[/green]"]
if warn_count:
parts.append(f"[yellow]{warn_count} warning(s)[/yellow]")
if fail_count:
parts.append(f"[red]{fail_count} issue(s)[/red]")
console.print(" " + ", ".join(parts))
console.print()
if fail_count:
console.print(
"[red bold]Action required:[/red bold] address critical findings "
"before storing sensitive data with OpenJarvis."
)
console.print()
elif warn_count:
console.print(
"[yellow]Review warnings above[/yellow] — they may not block "
"usage but could affect your privacy posture."
)
console.print()
@click.command()
@click.option("--quick", is_flag=True, default=False, help="Run only critical checks.")
def scan(quick: bool) -> None:
@click.option("--json", "as_json", is_flag=True, default=False, help="Output as JSON.")
def scan(quick: bool, as_json: bool) -> None:
"""Audit your environment for privacy and security risks."""
scanner = PrivacyScanner()
results: List[ScanResult] = scanner.run_quick() if quick else scanner.run_all()
if as_json:
import json as json_mod
output = [
{
"name": r.name,
"status": r.status,
"message": r.message,
"platform": r.platform,
}
for r in results
]
click.echo(json_mod.dumps(output, indent=2))
return
if not results:
click.echo("No applicable checks for this platform.")
return
warnings = 0
failures = 0
for r in results:
icon = _STATUS_ICONS.get(r.status, "?")
click.echo(f" [{icon}] {r.name}: {r.message}")
if r.status == "warn":
warnings += 1
elif r.status == "fail":
failures += 1
click.echo("")
parts = []
if warnings:
parts.append(f"{warnings} warning(s)")
if failures:
parts.append(f"{failures} issue(s)")
if parts:
click.echo("Summary: " + ", ".join(parts) + ".")
else:
click.echo("Summary: all checks passed.")
_render_results(results)
src/openjarvis/server/routes.py
+127 -55
View File
@@ -32,12 +32,14 @@ def _to_messages(chat_messages) -> list[Message]:
messages = []
for m in chat_messages:
role = Role(m.role) if m.role in {r.value for r in Role} else Role.USER
messages.append(Message(
role=role,
content=m.content or "",
name=m.name,
tool_call_id=m.tool_call_id,
))
messages.append(
Message(
role=role,
content=m.content or "",
name=m.name,
tool_call_id=m.tool_call_id,
)
)
return messages
@@ -75,7 +77,10 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
max_context_tokens=config.memory.context_max_tokens,
)
enriched = inject_context(
query_text, messages, memory_backend, config=ctx_cfg,
query_text,
messages,
memory_backend,
config=ctx_cfg,
)
# Rebuild request messages from enriched Message objects
if len(enriched) > len(messages):
@@ -83,16 +88,19 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
new_msgs = []
for msg in enriched:
new_msgs.append(ChatMessage(
role=msg.role.value,
content=msg.content,
name=msg.name,
tool_call_id=getattr(msg, "tool_call_id", None),
))
new_msgs.append(
ChatMessage(
role=msg.role.value,
content=msg.content,
name=msg.name,
tool_call_id=getattr(msg, "tool_call_id", None),
)
)
request_body.messages = new_msgs
except Exception:
logging.getLogger("openjarvis.server").debug(
"Memory context injection failed", exc_info=True,
"Memory context injection failed",
exc_info=True,
)
# Run complexity analysis on the last user message
@@ -111,7 +119,8 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
cr = score_complexity(query_text_for_complexity)
suggested = adjust_tokens_for_model(
cr.suggested_max_tokens, model,
cr.suggested_max_tokens,
model,
)
complexity_info = ComplexityInfo(
score=cr.score,
@@ -124,7 +133,8 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
request_body.max_tokens = suggested
except Exception:
logging.getLogger("openjarvis.server").debug(
"Complexity analysis failed", exc_info=True,
"Complexity analysis failed",
exc_info=True,
)
if request_body.stream:
@@ -143,13 +153,19 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
bus = getattr(request.app.state, "bus", None)
return _handle_direct(
engine, model, request_body,
bus=bus, complexity_info=complexity_info,
engine,
model,
request_body,
bus=bus,
complexity_info=complexity_info,
)
def _handle_direct(
engine, model: str, req: ChatCompletionRequest, bus=None,
engine,
model: str,
req: ChatCompletionRequest,
bus=None,
complexity_info=None,
) -> ChatCompletionResponse:
"""Direct engine call without agent."""
@@ -161,8 +177,12 @@ def _handle_direct(
from openjarvis.telemetry.wrapper import instrumented_generate
result = instrumented_generate(
engine, messages, model=model, bus=bus,
temperature=req.temperature, max_tokens=req.max_tokens,
engine,
messages,
model=model,
bus=bus,
temperature=req.temperature,
max_tokens=req.max_tokens,
**kwargs,
)
else:
@@ -194,10 +214,12 @@ def _handle_direct(
return ChatCompletionResponse(
model=model,
choices=[Choice(
message=choice_msg,
finish_reason=result.get("finish_reason", "stop"),
)],
choices=[
Choice(
message=choice_msg,
finish_reason=result.get("finish_reason", "stop"),
)
],
usage=UsageInfo(
prompt_tokens=usage.get("prompt_tokens", 0),
completion_tokens=usage.get("completion_tokens", 0),
@@ -208,7 +230,9 @@ def _handle_direct(
def _handle_agent(
agent, model: str, req: ChatCompletionRequest,
agent,
model: str,
req: ChatCompletionRequest,
complexity_info=None,
) -> ChatCompletionResponse:
"""Run through agent."""
@@ -241,10 +265,12 @@ def _handle_agent(
return ChatCompletionResponse(
model=model,
choices=[Choice(
message=ChoiceMessage(role="assistant", content=result.content),
finish_reason="stop",
)],
choices=[
Choice(
message=ChoiceMessage(role="assistant", content=result.content),
finish_reason="stop",
)
],
usage=usage,
complexity=complexity_info,
)
@@ -258,7 +284,9 @@ async def _handle_agent_stream(agent, bus, model, req):
async def _handle_stream(
engine, model: str, req: ChatCompletionRequest,
engine,
model: str,
req: ChatCompletionRequest,
complexity_info=None,
):
"""Stream response using SSE format."""
@@ -270,9 +298,11 @@ async def _handle_stream(
first_chunk = ChatCompletionChunk(
id=chunk_id,
model=model,
choices=[StreamChoice(
delta=DeltaMessage(role="assistant"),
)],
choices=[
StreamChoice(
delta=DeltaMessage(role="assistant"),
)
],
)
yield f"data: {first_chunk.model_dump_json()}\n\n"
@@ -287,27 +317,34 @@ async def _handle_stream(
chunk = ChatCompletionChunk(
id=chunk_id,
model=model,
choices=[StreamChoice(
delta=DeltaMessage(content=token),
)],
choices=[
StreamChoice(
delta=DeltaMessage(content=token),
)
],
)
yield f"data: {chunk.model_dump_json()}\n\n"
except Exception as exc:
# Surface errors as a content chunk so the frontend can
# display them instead of silently failing.
import logging
logging.getLogger("openjarvis.server").error(
"Stream error: %s", exc, exc_info=True,
"Stream error: %s",
exc,
exc_info=True,
)
error_chunk = ChatCompletionChunk(
id=chunk_id,
model=model,
choices=[StreamChoice(
delta=DeltaMessage(
content=f"\n\nError during generation: {exc}",
),
finish_reason="stop",
)],
choices=[
StreamChoice(
delta=DeltaMessage(
content=f"\n\nError during generation: {exc}",
),
finish_reason="stop",
)
],
)
yield f"data: {error_chunk.model_dump_json()}\n\n"
yield "data: [DONE]\n\n"
@@ -315,13 +352,16 @@ async def _handle_stream(
# Send finish chunk with usage data if available
import json as _json
finish_data = ChatCompletionChunk(
id=chunk_id,
model=model,
choices=[StreamChoice(
delta=DeltaMessage(),
finish_reason="stop",
)],
choices=[
StreamChoice(
delta=DeltaMessage(),
finish_reason="stop",
)
],
)
finish_dict = _json.loads(finish_data.model_dump_json())
@@ -456,18 +496,22 @@ async def savings(request: Request):
# Exclude cloud model tokens from savings — only local
# inference counts toward cost savings.
_cloud_prefixes = (
"gpt-", "o1-", "o3-", "o4-",
"claude-", "gemini-", "openrouter/",
"gpt-",
"o1-",
"o3-",
"o4-",
"claude-",
"gemini-",
"openrouter/",
)
local_models = [
m for m in summary.per_model
m
for m in summary.per_model
if not any(m.model_id.startswith(p) for p in _cloud_prefixes)
]
result = compute_savings(
prompt_tokens=sum(m.prompt_tokens for m in local_models),
completion_tokens=sum(
m.completion_tokens for m in local_models
),
completion_tokens=sum(m.completion_tokens for m in local_models),
total_calls=sum(m.call_count for m in local_models),
session_start=session_start if session_start else 0.0,
prompt_tokens_evaluated=sum(
@@ -557,7 +601,8 @@ async def channel_send(request: Request):
if not channel_name or not content:
raise HTTPException(
status_code=400, detail="'channel' and 'content' are required",
status_code=400,
detail="'channel' and 'content' are required",
)
ok = bridge.send(channel_name, content, conversation_id=conversation_id)
@@ -575,4 +620,31 @@ async def channel_status(request: Request):
return {"status": bridge.status().value}
# ---------------------------------------------------------------------------
# Security scan endpoint
# ---------------------------------------------------------------------------
@router.get("/v1/security/scan")
async def security_scan():
"""Run a read-only security environment audit and return findings."""
from openjarvis.cli.scan_cmd import PrivacyScanner
scanner = PrivacyScanner()
results = scanner.run_all()
return {
"has_warnings": any(r.status == "warn" for r in results),
"has_failures": any(r.status == "fail" for r in results),
"findings": [
{
"name": r.name,
"status": r.status,
"message": r.message,
"platform": r.platform,
}
for r in results
],
}
__all__ = ["router"]
tests/cli/test_scan.py
+134 -21
View File
@@ -31,9 +31,7 @@ def _make_proc(
class TestScanResultDataclass:
def test_fields_exist(self) -> None:
r = ScanResult(
name="test", status="ok", message="all good", platform="all"
)
r = ScanResult(name="test", status="ok", message="all good", platform="all")
assert r.name == "test"
assert r.status == "ok"
assert r.message == "all good"
@@ -246,15 +244,9 @@ class TestPlatformFiltering:
other_plat = "linux" if current_plat == "darwin" else "darwin"
with patch.object(scanner, "_get_all_checks") as mock_get:
darwin_ck = MagicMock(
return_value=ScanResult("d", "ok", "msg", "darwin")
)
linux_ck = MagicMock(
return_value=ScanResult("l", "ok", "msg", "linux")
)
all_ck = MagicMock(
return_value=ScanResult("a", "ok", "msg", "all")
)
darwin_ck = MagicMock(return_value=ScanResult("d", "ok", "msg", "darwin"))
linux_ck = MagicMock(return_value=ScanResult("l", "ok", "msg", "linux"))
all_ck = MagicMock(return_value=ScanResult("a", "ok", "msg", "all"))
mock_get.return_value = [darwin_ck, linux_ck, all_ck]
results = scanner.run_all()
@@ -275,23 +267,19 @@ class TestRemoteAccess:
def test_no_remote_access(self) -> None:
scanner = PrivacyScanner()
with patch.object(scanner, "_run") as mock_run:
mock_run.return_value = CompletedProcess(
[], 1, stdout="", stderr=""
)
mock_run.return_value = CompletedProcess([], 1, stdout="", stderr="")
result = scanner.check_remote_access()
assert result.status == "ok"
def test_xrdp_running(self) -> None:
scanner = PrivacyScanner()
with patch.object(scanner, "_run") as mock_run:
def side_effect(cmd, **kw):
if any("xrdp" in str(c) for c in cmd):
return CompletedProcess(
cmd, 0, stdout="12345", stderr=""
)
return CompletedProcess(
cmd, 1, stdout="", stderr=""
)
return CompletedProcess(cmd, 0, stdout="12345", stderr="")
return CompletedProcess(cmd, 1, stdout="", stderr="")
mock_run.side_effect = side_effect
result = scanner.check_remote_access()
assert result.status == "warn"
@@ -348,3 +336,128 @@ class TestRunQuick:
names = {r.name for r in results}
assert "Network Exposure" not in names
assert "Screen Recording" not in names
# ---------------------------------------------------------------------------
# TestDNS
# ---------------------------------------------------------------------------
class TestDNS:
"""Tests for check_dns (macOS)."""
def test_encrypted_dns_detected(self) -> None:
scanner = PrivacyScanner()
scutil_out = (
"resolver #1\n nameserver[0] : 127.0.0.1\n flags : dns-over-https\n"
)
with (
patch("sys.platform", "darwin"),
patch("subprocess.run", return_value=_make_proc(stdout=scutil_out)),
):
result = scanner.check_dns()
assert result.status == "ok"
assert "Encrypted" in result.message or "DoH" in result.message
def test_plain_dns_detected(self) -> None:
scanner = PrivacyScanner()
scutil_out = (
"resolver #1\n nameserver[0] : 8.8.8.8\n nameserver[1] : 8.8.4.4\n"
)
with (
patch("sys.platform", "darwin"),
patch("subprocess.run", return_value=_make_proc(stdout=scutil_out)),
):
result = scanner.check_dns()
assert result.status == "warn"
assert "8.8.8.8" in result.message
def test_private_dns(self) -> None:
scanner = PrivacyScanner()
scutil_out = "resolver #1\n nameserver[0] : 192.168.1.1\n"
with (
patch("sys.platform", "darwin"),
patch("subprocess.run", return_value=_make_proc(stdout=scutil_out)),
):
result = scanner.check_dns()
assert result.status == "ok"
assert "192.168.1.1" in result.message
def test_skip_on_linux(self) -> None:
scanner = PrivacyScanner()
with patch("sys.platform", "linux"):
result = scanner.check_dns()
assert result.status == "skip"
def test_scutil_not_available(self) -> None:
scanner = PrivacyScanner()
with (
patch("sys.platform", "darwin"),
patch("subprocess.run", side_effect=FileNotFoundError),
):
result = scanner.check_dns()
assert result.status == "skip"
# ---------------------------------------------------------------------------
# TestExpandedRemoteAccess
# ---------------------------------------------------------------------------
class TestExpandedRemoteAccess:
"""Verify expanded remote-access process list."""
def test_ngrok_detected(self) -> None:
scanner = PrivacyScanner()
with patch.object(scanner, "_run") as mock_run:
def side_effect(cmd, **kw):
if any("ngrok" in str(c) for c in cmd):
return CompletedProcess(cmd, 0, stdout="99999", stderr="")
return CompletedProcess(cmd, 1, stdout="", stderr="")
mock_run.side_effect = side_effect
result = scanner.check_remote_access()
assert result.status == "warn"
assert "ngrok" in result.message
def test_tailscaled_detected(self) -> None:
scanner = PrivacyScanner()
with patch.object(scanner, "_run") as mock_run:
def side_effect(cmd, **kw):
if any("tailscaled" in str(c) for c in cmd):
return CompletedProcess(cmd, 0, stdout="88888", stderr="")
return CompletedProcess(cmd, 1, stdout="", stderr="")
mock_run.side_effect = side_effect
result = scanner.check_remote_access()
assert result.status == "warn"
# ---------------------------------------------------------------------------
# TestJsonOutput
# ---------------------------------------------------------------------------
class TestJsonOutput:
"""Test --json output flag."""
def test_json_output_structure(self) -> None:
from click.testing import CliRunner
from openjarvis.cli.scan_cmd import scan
runner = CliRunner()
with patch(
"openjarvis.cli.scan_cmd.PrivacyScanner.run_all",
return_value=[
ScanResult("Test", "ok", "all good", "all"),
],
):
result = runner.invoke(scan, ["--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert isinstance(data, list)
assert data[0]["name"] == "Test"
assert data[0]["status"] == "ok"