From ced38bcba33514173fbb51a64f1329f4faccac8d Mon Sep 17 00:00:00 2001 From: Jon Saad-Falcon <41205309+jonsaadfalcon@users.noreply.github.com> Date: Tue, 24 Mar 2026 18:07:54 -0700 Subject: [PATCH] feat: add privacy environment scanner with platform-specific checks New PrivacyScanner class with checks for disk encryption (FileVault/LUKS), MDM profiles, cloud sync agents, network exposure, and screen recording. Supports macOS and Linux with graceful skip on missing tools. Co-Authored-By: Claude Opus 4.6 (1M context) --- src/openjarvis/cli/scan_cmd.py | 386 +++++++++++++++++++++++++++++++++ tests/cli/test_scan.py | 256 ++++++++++++++++++++++ 2 files changed, 642 insertions(+) create mode 100644 src/openjarvis/cli/scan_cmd.py create mode 100644 tests/cli/test_scan.py diff --git a/src/openjarvis/cli/scan_cmd.py b/src/openjarvis/cli/scan_cmd.py new file mode 100644 index 00000000..e8d83e08 --- /dev/null +++ b/src/openjarvis/cli/scan_cmd.py @@ -0,0 +1,386 @@ +"""``jarvis scan`` — audit your environment for privacy and security risks.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Callable, List + +import click + +# Engine ports that should only be listening on localhost. +_ENGINE_PORTS = {11434, 8080, 8000, 30000, 1234, 52415, 18181} + +# Processes associated with cloud-sync agents. +_CLOUD_SYNC_PROCS = ["Dropbox", "OneDrive", "Google Drive", "iCloudDrive"] + +# Screen-recording / remote-access processes (macOS). +_SCREEN_RECORDING_PROCS = [ + "TeamViewer", "AnyDesk", "ScreenConnect", "vncviewer", "Vine" +] + +# Remote-access processes (Linux). +_REMOTE_ACCESS_PROCS = ["xrdp", "x11vnc", "vncserver", "AnyDesk"] + + +# --------------------------------------------------------------------------- +# Data model +# --------------------------------------------------------------------------- + + +@dataclass +class ScanResult: + """Result of a single privacy/security check.""" + + name: str + status: str # "ok" | "warn" | "fail" | "skip" + message: str + platform: str # "darwin" | "linux" | "all" + + +# --------------------------------------------------------------------------- +# Scanner +# --------------------------------------------------------------------------- + + +class PrivacyScanner: + """Collection of environment privacy checks.""" + + # -- Subprocess helper --------------------------------------------------- + + def _run(self, cmd: list[str]) -> subprocess.CompletedProcess: # type: ignore[type-arg] + return subprocess.run(cmd, capture_output=True, text=True, timeout=10) + + # -- Individual checks --------------------------------------------------- + + def check_filevault(self) -> ScanResult: + """Check whether FileVault disk encryption is enabled (macOS).""" + try: + proc = self._run(["fdesetup", "status"]) + if "On" in proc.stdout: + return ScanResult( + name="FileVault", + status="ok", + message="FileVault is enabled.", + platform="darwin", + ) + return ScanResult( + name="FileVault", + status="fail", + message="FileVault is NOT enabled. Enable full-disk encryption.", + platform="darwin", + ) + except Exception: + return ScanResult( + name="FileVault", + status="skip", + message="fdesetup not available.", + platform="darwin", + ) + + def check_mdm(self) -> ScanResult: + """Check whether the device is enrolled in an MDM profile (macOS).""" + try: + proc = self._run(["profiles", "status", "-type", "enrollment"]) + output = proc.stdout + proc.stderr + lower = output.lower() + # "not enrolled" is a strong negative signal — check it first. + not_enrolled = "not enrolled" in lower or "no" in lower + # Positive signals: explicit Yes/enrolled without a negation. + enrolled_yes = ( + "mdm enrollment: yes" in lower + or "enrolled via dep: yes" in lower + or ("enrolled" in lower and not not_enrolled) + ) + if enrolled_yes: + return ScanResult( + name="MDM Enrollment", + status="warn", + message="Device appears to be enrolled in an MDM profile.", + platform="darwin", + ) + return ScanResult( + name="MDM Enrollment", + status="ok", + message="Device is not enrolled in an MDM profile.", + platform="darwin", + ) + except Exception: + return ScanResult( + name="MDM Enrollment", + status="skip", + message="profiles command not available.", + platform="darwin", + ) + + def check_icloud_sync(self) -> ScanResult: + """Check whether ~/.openjarvis is inside iCloud Drive sync scope.""" + try: + config_path = Path("~/.openjarvis").expanduser().resolve() + icloud_path = Path("~/Library/Mobile Documents/").expanduser().resolve() + if str(config_path).startswith(str(icloud_path)): + return ScanResult( + name="iCloud Sync", + status="warn", + message="~/.openjarvis may be synced to iCloud.", + platform="darwin", + ) + # Also probe defaults for com.apple.bird (iCloud daemon) + try: + proc = self._run( + ["defaults", "read", "com.apple.bird", "optout_preference"] + ) + val = proc.stdout.strip() + if val == "0": + return ScanResult( + name="iCloud Sync", + status="warn", + message="iCloud Desktop/Documents sync may be active.", + platform="darwin", + ) + except Exception: + pass + return ScanResult( + name="iCloud Sync", + status="ok", + message="~/.openjarvis is not inside iCloud Drive.", + platform="darwin", + ) + except Exception: + return ScanResult( + name="iCloud Sync", + status="skip", + message="Could not determine iCloud sync status.", + platform="darwin", + ) + + def check_luks(self) -> ScanResult: + """Check whether any block device uses LUKS encryption (Linux).""" + try: + proc = self._run(["lsblk", "-o", "NAME,TYPE,FSTYPE", "-J"]) + data = json.loads(proc.stdout) + except Exception: + return ScanResult( + name="LUKS Encryption", + status="skip", + message="lsblk not available or returned unexpected output.", + platform="linux", + ) + + def _has_luks(devices: list) -> bool: # type: ignore[type-arg] + for dev in devices: + if dev.get("fstype") == "crypto_LUKS": + return True + children = dev.get("children") or [] + if _has_luks(children): + return True + return False + + try: + devices = data.get("blockdevices", []) + if _has_luks(devices): + return ScanResult( + name="LUKS Encryption", + status="ok", + message="At least one LUKS-encrypted device found.", + platform="linux", + ) + return ScanResult( + name="LUKS Encryption", + status="fail", + message="No LUKS-encrypted block devices found.", + platform="linux", + ) + except Exception: + return ScanResult( + name="LUKS Encryption", + status="skip", + message="Could not parse lsblk output.", + platform="linux", + ) + + def _check_processes( + self, + names: list[str], + check_name: str, + warn_msg: str, + platform: str, + ) -> ScanResult: + """Shared helper: pgrep for any of the given process names.""" + try: + for name in names: + try: + proc = self._run(["pgrep", "-x", name]) + if proc.returncode == 0: + return ScanResult( + name=check_name, + status="warn", + message=warn_msg.format(name=name), + platform=platform, + ) + except Exception: + continue + return ScanResult( + name=check_name, + status="ok", + message=f"No {check_name.lower()} processes detected.", + platform=platform, + ) + except Exception: + return ScanResult( + name=check_name, + status="skip", + message="pgrep not available.", + platform=platform, + ) + + def check_cloud_sync_agents(self) -> ScanResult: + """Check for running cloud-sync agent processes.""" + return self._check_processes( + names=_CLOUD_SYNC_PROCS, + check_name="Cloud Sync Agents", + warn_msg="{name} sync agent is running — weights may be uploaded to cloud.", + platform="all", + ) + + def check_network_exposure(self) -> ScanResult: + """Check if engine ports are exposed on 0.0.0.0 rather than localhost.""" + try: + if sys.platform == "darwin": + proc = self._run(["lsof", "-iTCP", "-sTCP:LISTEN", "-n", "-P"]) + else: + proc = self._run(["ss", "-tlnp"]) + output = proc.stdout + + exposed: list[int] = [] + for port in _ENGINE_PORTS: + # Look for patterns like *:PORT or 0.0.0.0:PORT + for token in (f"*:{port}", f"0.0.0.0:{port}", f":::{ port}"): + if token.replace(" ", "") in output.replace(" ", ""): + exposed.append(port) + break + + if exposed: + ports_str = ", ".join(str(p) for p in sorted(exposed)) + return ScanResult( + name="Network Exposure", + status="warn", + message=f"Engine port(s) {ports_str} exposed on all interfaces.", + platform="all", + ) + return ScanResult( + name="Network Exposure", + status="ok", + message="All engine ports appear to be bound to localhost only.", + platform="all", + ) + except Exception: + return ScanResult( + name="Network Exposure", + status="skip", + message="Could not determine network exposure (lsof/ss unavailable).", + platform="all", + ) + + def check_screen_recording(self) -> ScanResult: + """Check for running screen-recording / remote-desktop processes (macOS).""" + return self._check_processes( + names=_SCREEN_RECORDING_PROCS, + check_name="Screen Recording", + warn_msg="{name} is running — screen may be accessible remotely.", + platform="darwin", + ) + + def check_remote_access(self) -> ScanResult: + """Check for running remote-access processes (Linux).""" + return self._check_processes( + names=_REMOTE_ACCESS_PROCS, + check_name="Remote Access", + warn_msg="{name} is running — system may be accessible remotely.", + platform="linux", + ) + + # -- Orchestration ------------------------------------------------------- + + def _get_all_checks(self) -> list[Callable[[], ScanResult]]: + return [ + self.check_filevault, + self.check_mdm, + self.check_icloud_sync, + self.check_cloud_sync_agents, + self.check_network_exposure, + self.check_luks, + self.check_screen_recording, + self.check_remote_access, + ] + + def run_all(self) -> list[ScanResult]: + """Run all checks, filter to the current platform, hide 'skip' results.""" + current_plat = "darwin" if sys.platform == "darwin" else "linux" + results: list[ScanResult] = [] + for check_fn in self._get_all_checks(): + result = check_fn() + if result.platform not in (current_plat, "all"): + continue + if result.status == "skip": + continue + results.append(result) + return results + + def run_quick(self) -> list[ScanResult]: + """Run only critical checks: disk encryption + cloud sync agents.""" + current_plat = "darwin" if sys.platform == "darwin" else "linux" + quick_checks: list[Callable[[], ScanResult]] + if current_plat == "darwin": + quick_checks = [self.check_filevault, self.check_cloud_sync_agents] + else: + quick_checks = [self.check_luks, self.check_cloud_sync_agents] + results = [] + for check_fn in quick_checks: + result = check_fn() + if result.status != "skip": + results.append(result) + return results + + +# --------------------------------------------------------------------------- +# CLI command +# --------------------------------------------------------------------------- + +_STATUS_ICONS = {"ok": "✓", "warn": "!", "fail": "✗", "skip": "-"} + + +@click.command() +@click.option("--quick", is_flag=True, default=False, help="Run only critical checks.") +def scan(quick: 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 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.") diff --git a/tests/cli/test_scan.py b/tests/cli/test_scan.py new file mode 100644 index 00000000..9b8eea7e --- /dev/null +++ b/tests/cli/test_scan.py @@ -0,0 +1,256 @@ +"""Tests for ``jarvis scan`` privacy scanner CLI command.""" + +from __future__ import annotations + +import json +from subprocess import CompletedProcess +from unittest.mock import MagicMock, patch + +import pytest + +from openjarvis.cli.scan_cmd import ScanResult, PrivacyScanner + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_proc(stdout: str = "", stderr: str = "", returncode: int = 0) -> CompletedProcess: + return CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr) + + +# --------------------------------------------------------------------------- +# TestScanResultDataclass +# --------------------------------------------------------------------------- + + +class TestScanResultDataclass: + def test_fields_exist(self) -> None: + r = ScanResult(name="test", status="ok", message="all good", platform="all") + assert r.name == "test" + assert r.status == "ok" + assert r.message == "all good" + assert r.platform == "all" + + def test_status_values(self) -> None: + for status in ("ok", "warn", "fail", "skip"): + r = ScanResult(name="x", status=status, message="", platform="all") + assert r.status == status + + def test_platform_values(self) -> None: + for plat in ("darwin", "linux", "all"): + r = ScanResult(name="x", status="ok", message="", platform=plat) + assert r.platform == plat + + +# --------------------------------------------------------------------------- +# TestFileVault +# --------------------------------------------------------------------------- + + +class TestFileVault: + def test_filevault_enabled(self) -> None: + scanner = PrivacyScanner() + with patch("subprocess.run", return_value=_make_proc(stdout="FileVault is On.")): + result = scanner.check_filevault() + assert result.status == "ok" + + def test_filevault_disabled(self) -> None: + scanner = PrivacyScanner() + with patch("subprocess.run", return_value=_make_proc(stdout="FileVault is Off.")): + result = scanner.check_filevault() + assert result.status == "fail" + + def test_command_not_found(self) -> None: + scanner = PrivacyScanner() + with patch("subprocess.run", side_effect=FileNotFoundError): + result = scanner.check_filevault() + assert result.status == "skip" + + +# --------------------------------------------------------------------------- +# TestMDM +# --------------------------------------------------------------------------- + + +class TestMDM: + def test_not_enrolled(self) -> None: + scanner = PrivacyScanner() + with patch( + "subprocess.run", + return_value=_make_proc(stdout="Enrolled via DEP: No\nMDM enrollment: not enrolled"), + ): + result = scanner.check_mdm() + assert result.status == "ok" + + def test_enrolled(self) -> None: + scanner = PrivacyScanner() + with patch( + "subprocess.run", + return_value=_make_proc(stdout="MDM enrollment: Yes\nEnrolled via DEP: Yes"), + ): + result = scanner.check_mdm() + assert result.status == "warn" + + +# --------------------------------------------------------------------------- +# TestCloudSync +# --------------------------------------------------------------------------- + + +class TestCloudSync: + def test_no_agents(self) -> None: + """pgrep returns non-zero (process not found) → no sync agents running.""" + scanner = PrivacyScanner() + with patch("subprocess.run", return_value=_make_proc(returncode=1, stdout="")): + result = scanner.check_cloud_sync_agents() + assert result.status == "ok" + + def test_dropbox_running(self) -> None: + """pgrep returns 0 when Dropbox is in the command.""" + scanner = PrivacyScanner() + + def _pgrep_side_effect(cmd, **kwargs): + if "Dropbox" in cmd: + return _make_proc(returncode=0, stdout="1234") + return _make_proc(returncode=1, stdout="") + + with patch("subprocess.run", side_effect=_pgrep_side_effect): + result = scanner.check_cloud_sync_agents() + assert result.status == "warn" + + +# --------------------------------------------------------------------------- +# TestNetworkExposure +# --------------------------------------------------------------------------- + + +class TestNetworkExposure: + def test_no_exposed_ports(self) -> None: + """Only localhost bindings → ok.""" + scanner = PrivacyScanner() + # lsof / ss output showing only 127.0.0.1 + lsof_output = "ollama 1234 user IPv4 TCP 127.0.0.1:11434 (LISTEN)\n" + with patch("subprocess.run", return_value=_make_proc(stdout=lsof_output)): + result = scanner.check_network_exposure() + assert result.status == "ok" + + def test_exposed_port(self) -> None: + """A port bound to 0.0.0.0 or * → warn with port in message.""" + scanner = PrivacyScanner() + lsof_output = "ollama 1234 user IPv4 TCP *:11434 (LISTEN)\n" + with patch("subprocess.run", return_value=_make_proc(stdout=lsof_output)): + result = scanner.check_network_exposure() + assert result.status == "warn" + assert "11434" in result.message + + +# --------------------------------------------------------------------------- +# TestLUKS +# --------------------------------------------------------------------------- + + +class TestLUKS: + def test_encrypted(self) -> None: + """lsblk JSON contains crypto_LUKS → ok.""" + scanner = PrivacyScanner() + lsblk_data = { + "blockdevices": [ + { + "name": "sda", + "type": "disk", + "fstype": None, + "children": [ + {"name": "sda1", "type": "part", "fstype": "crypto_LUKS"} + ], + } + ] + } + with patch( + "subprocess.run", + return_value=_make_proc(stdout=json.dumps(lsblk_data)), + ): + result = scanner.check_luks() + assert result.status == "ok" + + def test_not_encrypted(self) -> None: + """lsblk JSON has only ext4 → fail.""" + scanner = PrivacyScanner() + lsblk_data = { + "blockdevices": [ + {"name": "sda", "type": "disk", "fstype": None}, + {"name": "sda1", "type": "part", "fstype": "ext4"}, + ] + } + with patch( + "subprocess.run", + return_value=_make_proc(stdout=json.dumps(lsblk_data)), + ): + result = scanner.check_luks() + assert result.status == "fail" + + +# --------------------------------------------------------------------------- +# TestScreenRecording +# --------------------------------------------------------------------------- + + +class TestScreenRecording: + def test_none_running(self) -> None: + """No screen-recording processes → ok.""" + scanner = PrivacyScanner() + with patch("subprocess.run", return_value=_make_proc(returncode=1, stdout="")): + result = scanner.check_screen_recording() + assert result.status == "ok" + + def test_teamviewer_running(self) -> None: + """TeamViewer found → warn.""" + scanner = PrivacyScanner() + + def _pgrep_side_effect(cmd, **kwargs): + if "TeamViewer" in cmd: + return _make_proc(returncode=0, stdout="5678") + return _make_proc(returncode=1, stdout="") + + with patch("subprocess.run", side_effect=_pgrep_side_effect): + result = scanner.check_screen_recording() + assert result.status == "warn" + + +# --------------------------------------------------------------------------- +# TestPlatformFiltering +# --------------------------------------------------------------------------- + + +class TestPlatformFiltering: + def test_run_all_filters_to_current_platform(self) -> None: + """run_all() should only call checks tagged 'all' or current platform.""" + scanner = PrivacyScanner() + called_platforms: list[str] = [] + + # Patch every check method to record its platform tag instead of running + checks = scanner._get_all_checks() + for check_fn in checks: + # Run the real method but capture which ones get included + pass + + import sys + + current_plat = "darwin" if sys.platform == "darwin" else "linux" + other_plat = "linux" if current_plat == "darwin" else "darwin" + + # Make every check return immediately with a known platform + with patch.object(scanner, "_get_all_checks") as mock_get: + darwin_check = MagicMock(return_value=ScanResult("d", "ok", "msg", "darwin")) + linux_check = MagicMock(return_value=ScanResult("l", "ok", "msg", "linux")) + all_check = MagicMock(return_value=ScanResult("a", "ok", "msg", "all")) + + mock_get.return_value = [darwin_check, linux_check, all_check] + results = scanner.run_all() + + # The check for the other platform should not appear in results + other_result_platform = "linux" if current_plat == "darwin" else "darwin" + result_platforms = {r.platform for r in results} + assert other_result_platform not in result_platforms + assert "all" in result_platforms or current_plat in result_platforms