From 2477510fe8bd3af5039456bccd08383cfbdad6d2 Mon Sep 17 00:00:00 2001 From: mlugo-apx Date: Fri, 27 Feb 2026 13:43:15 -0500 Subject: [PATCH] Fix all 12 SME-reviewed issues with comprehensive TDD coverage Addresses bugs, reliability gaps, security hardening, and feature requests identified in SME review. All changes driven by strict Red-Green-Refactor TDD discipline (94 tests, up from 67). Bugs: deploy.sh and install_and_start.sh now delegate to install_service.sh instead of raw-copying the service template without placeholder substitution. Reliability: Added desktop alerting via notify-send in health checks, log rotation via RotatingFileHandler (10MB/5 backups), and remote disk space checking before rsync transfers. Security: Re-enabled SystemCallFilter in systemd service, added deprecation notice to bash monitor script. Features: Desktop notifications on sync success/failure, DELETE_AFTER_SYNC config option, multiple watch directory support (comma-separated WATCH_DIR). Code quality: Converted all 33 f-string logging calls to lazy % formatting, removed duplicate imports. Co-Authored-By: Claude Opus 4.6 --- check_gcode_monitor.sh | 13 ++ config.example | 5 +- deploy.sh | 8 +- gcode-monitor.service | 8 +- install_and_start.sh | 5 +- monitor_and_sync.py | 247 +++++++++++++++++--------- monitor_and_sync.sh | 7 + tests/integration/test_integration.py | 104 +++++++++++ tests/security/test_security.py | 36 +++- tests/unit/test_validation.py | 124 ++++++++++++- 10 files changed, 459 insertions(+), 98 deletions(-) diff --git a/check_gcode_monitor.sh b/check_gcode_monitor.sh index 0842aed..8fa9048 100644 --- a/check_gcode_monitor.sh +++ b/check_gcode_monitor.sh @@ -18,11 +18,23 @@ print_header() { echo "[$(timestamp)] $1" } +send_alert() { + local title="$1" + local body="$2" + local urgency="${3:-critical}" + # Desktop notification (works if DISPLAY/DBUS available) + if command -v notify-send &>/dev/null; then + notify-send --urgency="$urgency" "$title" "$body" 2>/dev/null || true + fi + print_header "ALERT: $title - $body" +} + check_service_status() { if systemctl is-active --quiet "${SERVICE_NAME}"; then print_header "Service ${SERVICE_NAME} is active." else print_header "ERROR: Service ${SERVICE_NAME} is not active." + send_alert "GCode Monitor Down" "Service ${SERVICE_NAME} is not active" systemctl status "${SERVICE_NAME}" --no-pager exit 1 fi @@ -52,6 +64,7 @@ check_recent_errors() { print_header "Scanning journal for recent errors..." if journalctl -u "${SERVICE_NAME}" --since "30 minutes ago" | grep -qi "error"; then print_header "ERROR: Recent errors detected in journal." + send_alert "GCode Monitor Errors" "Recent errors detected in journal for ${SERVICE_NAME}" journalctl -u "${SERVICE_NAME}" --since "30 minutes ago" exit 1 else diff --git a/config.example b/config.example index d4593c9..d7dd49c 100644 --- a/config.example +++ b/config.example @@ -3,7 +3,7 @@ # Note: config.local is git-ignored for security # Local machine settings -WATCH_DIR="$HOME/Desktop" # Directory to monitor for .gcode files +WATCH_DIR="$HOME/Desktop" # Directory to monitor for .gcode files (comma-separated for multiple dirs) LOG_FILE="$HOME/.gcode_sync.log" # Log file location # Raspberry Pi connection settings @@ -12,6 +12,9 @@ REMOTE_HOST="192.168.1.100" # IP address or hostname of your Raspberry P REMOTE_PORT="22" # SSH port (default: 22) REMOTE_PATH="/mnt/usb_share" # Path to USB gadget mount point on Pi +# Optional settings +DELETE_AFTER_SYNC="false" # Delete local file after successful sync (true/false) + # Example configurations: # Direct connection: REMOTE_HOST="192.168.1.6" REMOTE_PORT="22" # SSH port forwarding: REMOTE_HOST="localhost" REMOTE_PORT="9702" diff --git a/deploy.sh b/deploy.sh index fb62242..5c1a0ec 100755 --- a/deploy.sh +++ b/deploy.sh @@ -19,11 +19,11 @@ if [ "$EUID" -ne 0 ]; then exec sudo bash "$0" "$@" fi -# Step 1: Copy updated service file -echo -e "${YELLOW}[1/4] Copying updated service file...${NC}" +# Step 1: Install service file via install_service.sh (template substitution) +echo -e "${YELLOW}[1/5] Installing service file...${NC}" SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -cp "${SCRIPT_DIR}/gcode-monitor.service" /etc/systemd/system/ -echo -e "${GREEN}✓ Service file updated${NC}" +bash "${SCRIPT_DIR}/install_service.sh" +echo -e "${GREEN}✓ Service file installed${NC}" echo "" # Step 2: Reload systemd configuration diff --git a/gcode-monitor.service b/gcode-monitor.service index 1b28c56..97af6a1 100644 --- a/gcode-monitor.service +++ b/gcode-monitor.service @@ -63,10 +63,10 @@ ProtectClock=true CapabilityBoundingSet= AmbientCapabilities= -# System call filtering - relaxed for Python compatibility -# SystemCallFilter=@system-service -# SystemCallFilter=~@privileged @resources @obsolete -# SystemCallArchitectures=native +# System call filtering +SystemCallFilter=@system-service +SystemCallFilter=~@privileged @resources @obsolete +SystemCallArchitectures=native # Logging StandardOutput=journal diff --git a/install_and_start.sh b/install_and_start.sh index bb7b168..1e516d4 100755 --- a/install_and_start.sh +++ b/install_and_start.sh @@ -78,10 +78,7 @@ read -p "Install as systemd service (auto-start on boot)? [Y/n]: " install_servi if [[ ! "$install_service" =~ ^[Nn]$ ]]; then echo "Installing systemd service..." - sudo cp "$SCRIPT_DIR/gcode-monitor.service" /etc/systemd/system/ - sudo systemctl daemon-reload - sudo systemctl enable gcode-monitor.service - sudo systemctl start gcode-monitor.service + sudo bash "$SCRIPT_DIR/install_service.sh" echo -e "${GREEN}✓ Service installed and started${NC}" echo diff --git a/monitor_and_sync.py b/monitor_and_sync.py index 81dbc0b..a983d67 100755 --- a/monitor_and_sync.py +++ b/monitor_and_sync.py @@ -10,6 +10,7 @@ import subprocess import shlex import logging +from logging.handlers import RotatingFileHandler import threading import re import shutil @@ -39,6 +40,25 @@ RETRY_BACKOFF_MULTIPLIER = 2 # Exponential backoff multiplier + +def send_notification(title: str, body: str, urgency: str = "normal") -> None: + """Send desktop notification using notify-send with graceful fallback. + + Args: + title: Notification title + body: Notification body text + urgency: Notification urgency (low, normal, critical) + """ + try: + subprocess.run( + ["notify-send", "--urgency=%s" % urgency, title, body], + timeout=5, + capture_output=True + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + pass # notify-send not available or timed out — not critical + + def parse_rsync_stats(stdout: Optional[str]) -> Dict[str, Any]: """Parse rsync --stats output into a dictionary of values.""" stats: Dict[str, Any] = {} @@ -104,8 +124,8 @@ def wrapper(*args: Any, **kwargs: Any) -> Tuple[Any, int]: raise # Log retry attempt - logging.warning(f"Attempt {attempt}/{max_attempts} failed: {type(e).__name__}") - logging.warning(f"Retrying in {delay}s...") + logging.warning("Attempt %d/%d failed: %s", attempt, max_attempts, type(e).__name__) + logging.warning("Retrying in %ss...", delay) time.sleep(delay) delay *= backoff_multiplier @@ -153,8 +173,6 @@ def load_config() -> Dict[str, str]: sys.exit(1) # Validate config values to prevent injection attacks - import re - # Port must be numeric if not config['REMOTE_PORT'].isdigit(): print(f"ERROR: REMOTE_PORT must be numeric (got: {config['REMOTE_PORT']})", file=sys.stderr) @@ -181,28 +199,32 @@ def load_config() -> Dict[str, str]: sys.exit(1) # Validate WATCH_DIR and LOG_FILE paths (defense in depth) - from pathlib import Path - - # Validate WATCH_DIR is absolute and within user home - try: - watch_dir = Path(config['WATCH_DIR']).resolve() - except Exception as e: - print(f"ERROR: Invalid WATCH_DIR path: {config['WATCH_DIR']}: {e}", file=sys.stderr) + # Support comma-separated directories for multi-dir monitoring + watch_dir_entries = [d.strip() for d in config['WATCH_DIR'].split(',') if d.strip()] + if not watch_dir_entries: + print("ERROR: WATCH_DIR must not be empty", file=sys.stderr) sys.exit(1) user_home = Path.home() - if not watch_dir.is_absolute(): - print(f"ERROR: WATCH_DIR must be an absolute path (got: {config['WATCH_DIR']})", file=sys.stderr) - sys.exit(1) + for wd_entry in watch_dir_entries: + try: + watch_dir = Path(wd_entry).resolve() + except Exception as e: + print(f"ERROR: Invalid WATCH_DIR path: {wd_entry}: {e}", file=sys.stderr) + sys.exit(1) - # Ensure WATCH_DIR is within user's home directory - try: - watch_dir.relative_to(user_home) - except ValueError: - print(f"ERROR: WATCH_DIR must be within user home directory ({user_home})", file=sys.stderr) - print(f" Got: {watch_dir}", file=sys.stderr) - sys.exit(1) + if not watch_dir.is_absolute(): + print(f"ERROR: WATCH_DIR must be an absolute path (got: {wd_entry})", file=sys.stderr) + sys.exit(1) + + # Ensure WATCH_DIR is within user's home directory + try: + watch_dir.relative_to(user_home) + except ValueError: + print(f"ERROR: WATCH_DIR must be within user home directory ({user_home})", file=sys.stderr) + print(f" Got: {watch_dir}", file=sys.stderr) + sys.exit(1) # Validate LOG_FILE path try: @@ -233,9 +255,10 @@ def load_config() -> Dict[str, str]: # Check for path traversal sequences path_traversal = re.compile(r'\.\.') - if path_traversal.search(config['WATCH_DIR']): - print("ERROR: WATCH_DIR contains path traversal sequence (..)", file=sys.stderr) - sys.exit(1) + for wd_entry in watch_dir_entries: + if path_traversal.search(wd_entry): + print("ERROR: WATCH_DIR contains path traversal sequence (..)", file=sys.stderr) + sys.exit(1) if path_traversal.search(config['LOG_FILE']): print("ERROR: LOG_FILE contains path traversal sequence (..)", file=sys.stderr) @@ -245,12 +268,14 @@ def load_config() -> Dict[str, str]: # Load configuration config = load_config() -WATCH_DIR = config['WATCH_DIR'] +WATCH_DIRS = [d.strip() for d in config['WATCH_DIR'].split(',') if d.strip()] +WATCH_DIR = WATCH_DIRS[0] # Backward compatibility: first dir REMOTE_USER = config['REMOTE_USER'] REMOTE_HOST = config['REMOTE_HOST'] REMOTE_PORT = config['REMOTE_PORT'] REMOTE_PATH = config['REMOTE_PATH'] LOG_FILE = config['LOG_FILE'] +DELETE_AFTER_SYNC = config.get('DELETE_AFTER_SYNC', 'false').lower() == 'true' # Ensure log directory exists with secure permissions before configuring logging log_path = Path(LOG_FILE) @@ -263,7 +288,7 @@ def load_config() -> Dict[str, str]: level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ - logging.FileHandler(LOG_FILE), + RotatingFileHandler(LOG_FILE, maxBytes=10*1024*1024, backupCount=5), logging.StreamHandler(sys.stdout) ] ) @@ -304,73 +329,81 @@ def sync_file(self, file_path: str) -> None: # Security: Validate file path is within watch directory abs_file_path = os.path.abspath(file_path) - abs_watch_dir = os.path.abspath(WATCH_DIR) - - if not abs_file_path.startswith(abs_watch_dir + os.sep): - logging.error(f"Security: File outside watch directory: {file_path}") - logging.error(f" File path: {abs_file_path}") - logging.error(f" Watch dir: {abs_watch_dir}") + # Check file is within any configured watch directory + in_watch_dir = any( + abs_file_path.startswith(os.path.abspath(wd) + os.sep) + for wd in WATCH_DIRS + ) + + if not in_watch_dir: + logging.error("Security: File outside watch directory: %s", file_path) + logging.error(" File path: %s", abs_file_path) + logging.error(" Watch dirs: %s", WATCH_DIRS) return # Security: Check it's a regular file (not symlink, directory, device, etc.) if not os.path.exists(abs_file_path): - logging.warning(f"File no longer exists: {file_path}") + logging.warning("File no longer exists: %s", file_path) return if os.path.islink(abs_file_path): - logging.error(f"Security: Refusing to sync symlink: {file_path}") + logging.error("Security: Refusing to sync symlink: %s", file_path) return if not os.path.isfile(abs_file_path): - logging.warning(f"Skipping non-regular file: {file_path}") + logging.warning("Skipping non-regular file: %s", file_path) return # Security: Validate file extension (defense in depth) if not abs_file_path.endswith('.gcode'): - logging.warning(f"Skipping non-gcode file: {file_path}") + logging.warning("Skipping non-gcode file: %s", file_path) return # Validate file size to prevent DoS try: file_size = os.path.getsize(abs_file_path) except OSError as e: - logging.error(f"Cannot determine file size: {file_path}: {e}") + logging.error("Cannot determine file size: %s: %s", file_path, e) return if file_size < MIN_FILE_SIZE: - logging.warning(f"Skipping empty file: {file_path}") + logging.warning("Skipping empty file: %s", file_path) return if file_size > MAX_FILE_SIZE: - logging.error(f"File too large: {file_path} ({file_size / (1024*1024):.2f} MB)") - logging.error(f"Maximum allowed size: {MAX_FILE_SIZE / (1024*1024):.2f} MB") + logging.error("File too large: %s (%.2f MB)", file_path, file_size / (1024*1024)) + logging.error("Maximum allowed size: %.2f MB", MAX_FILE_SIZE / (1024*1024)) return if file_size > WARN_FILE_SIZE: - logging.warning(f"Large file detected: {file_path} ({file_size / (1024*1024):.2f} MB)") + logging.warning("Large file detected: %s (%.2f MB)", file_path, file_size / (1024*1024)) logging.warning("This may take several minutes to sync") - logging.info(f"Syncing file: {abs_file_path} ({file_size / (1024*1024):.2f} MB)") + logging.info("Syncing file: %s (%.2f MB)", abs_file_path, file_size / (1024*1024)) # SECURITY: Re-validate immediately before rsync to prevent TOCTOU race condition # This closes the window where file could be replaced with symlink after validation if os.path.islink(abs_file_path): - logging.error(f"Security: File became symlink after validation: {file_path}") + logging.error("Security: File became symlink after validation: %s", file_path) return if not os.path.isfile(abs_file_path): - logging.error(f"Security: File changed type after validation: {file_path}") + logging.error("Security: File changed type after validation: %s", file_path) return if not abs_file_path.endswith('.gcode'): - logging.error(f"Security: File extension changed after validation: {file_path}") + logging.error("Security: File extension changed after validation: %s", file_path) return # Calculate dynamic timeout based on file size # Baseline: 2 minutes for small files, add 1 minute per 100 MB for large files timeout_seconds = max(RSYNC_TOTAL_TIMEOUT, int((file_size / (100 * 1024 * 1024)) * 60)) - logging.debug(f"Using timeout: {timeout_seconds}s for {file_size / (1024*1024):.2f} MB file") + logging.debug("Using timeout: %ds for %.2f MB file", timeout_seconds, file_size / (1024*1024)) + + # Check remote disk space before syncing + if not self.check_remote_disk_space(file_size): + return # Build rsync command with timeouts rsync_cmd = [ @@ -378,10 +411,10 @@ def sync_file(self, file_path: str) -> None: "--stats", "--protect-args", "-avz", - f"--timeout={RSYNC_TIMEOUT}", - "-e", f"ssh -p {REMOTE_PORT} -o StrictHostKeyChecking=yes -o ConnectTimeout=10 -o ServerAliveInterval=5 -o ServerAliveCountMax=3", + "--timeout=%d" % RSYNC_TIMEOUT, + "-e", "ssh -p %s -o StrictHostKeyChecking=yes -o ConnectTimeout=10 -o ServerAliveInterval=5 -o ServerAliveCountMax=3" % REMOTE_PORT, abs_file_path, - f"{REMOTE_USER}@{REMOTE_HOST}:{shlex.quote(REMOTE_PATH if REMOTE_PATH.endswith('/') else f'{REMOTE_PATH}/')}" + "%s@%s:%s" % (REMOTE_USER, REMOTE_HOST, shlex.quote(REMOTE_PATH if REMOTE_PATH.endswith('/') else REMOTE_PATH + '/')) ] # Execute rsync with retry logic (handles transient network failures) @@ -392,7 +425,16 @@ def sync_file(self, file_path: str) -> None: mb_size = file_size / (1024 * 1024) transfer_rate = mb_size / duration if duration > 0 else float('inf') - logging.info(f"Successfully synced: {os.path.basename(abs_file_path)}") + logging.info("Successfully synced: %s", os.path.basename(abs_file_path)) + send_notification("GCode Synced", "Successfully synced: %s" % os.path.basename(abs_file_path)) + + # Delete local file after successful sync if configured + if DELETE_AFTER_SYNC: + try: + os.remove(abs_file_path) + logging.info("Deleted local file after sync: %s", os.path.basename(abs_file_path)) + except OSError as e: + logging.warning("Failed to delete local file after sync: %s: %s", abs_file_path, e) # Trigger USB gadget refresh usb_refresh_success = self.refresh_usb_gadget() @@ -405,41 +447,81 @@ def sync_file(self, file_path: str) -> None: speedup_value = stats.get("speedup") refresh_status = "ok" if usb_refresh_success else "failed" - rate_display = "inf" if transfer_rate == float('inf') else f"{transfer_rate:.2f}" + rate_display = "inf" if transfer_rate == float('inf') else "%.2f" % transfer_rate bytes_sent_display = total_bytes_sent if total_bytes_sent is not None else "n/a" bytes_received_display = total_bytes_received if total_bytes_received is not None else "n/a" literal_display = literal_data if literal_data is not None else "n/a" matched_display = matched_data if matched_data is not None else "n/a" - speedup_display = f"{speedup_value:.2f}" if speedup_value is not None else "n/a" + speedup_display = "%.2f" % speedup_value if speedup_value is not None else "n/a" summary_block = "\n".join([ "==================== Sync Summary ====================", - f" File : {os.path.basename(abs_file_path)}", - f" Size : {mb_size:.2f} MB", - f" Duration : {duration:.2f} s", - f" Average Rate : {rate_display} MB/s", - f" Attempts : {attempts_used}", - f" USB Refresh : {refresh_status}", - f" Bytes Sent : {bytes_sent_display}", - f" Bytes Received : {bytes_received_display}", - f" Literal Data : {literal_display}", - f" Matched Data : {matched_display}", - f" Speedup : {speedup_display}", + " File : %s" % os.path.basename(abs_file_path), + " Size : %.2f MB" % mb_size, + " Duration : %.2f s" % duration, + " Average Rate : %s MB/s" % rate_display, + " Attempts : %s" % attempts_used, + " USB Refresh : %s" % refresh_status, + " Bytes Sent : %s" % bytes_sent_display, + " Bytes Received : %s" % bytes_received_display, + " Literal Data : %s" % literal_display, + " Matched Data : %s" % matched_display, + " Speedup : %s" % speedup_display, "======================================================" ]) logging.info(summary_block) except subprocess.TimeoutExpired: - logging.error(f"Timeout syncing {file_path} - transfer took longer than 2 minutes") + logging.error("Timeout syncing %s - transfer took longer than 2 minutes", file_path) + send_notification("GCode Sync Failed", "Timeout syncing %s" % os.path.basename(file_path), "critical") except subprocess.CalledProcessError as e: - logging.error(f"Failed to sync {file_path}: {e}") - logging.error(f"STDERR: {e.stderr}") + logging.error("Failed to sync %s: %s", file_path, e) + logging.error("STDERR: %s", e.stderr) + send_notification("GCode Sync Failed", "Failed to sync %s" % os.path.basename(file_path), "critical") except Exception as e: - logging.error(f"Unexpected error syncing {file_path}: {e}") + logging.error("Unexpected error syncing %s: %s", file_path, e) + send_notification("GCode Sync Error", "Unexpected error syncing %s" % os.path.basename(file_path), "critical") finally: with self.syncing_lock: self.syncing.discard(file_path) + + def check_remote_disk_space(self, file_size: int) -> bool: + """Check if remote Pi has enough disk space for the file. + + Args: + file_size: Size of the file to sync in bytes + + Returns: + True if enough space (or check fails — non-blocking), False if insufficient + """ + try: + ssh_cmd = [ + "ssh", + "-p", REMOTE_PORT, + "-o", "StrictHostKeyChecking=yes", + "-o", "ConnectTimeout=10", + "%s@%s" % (REMOTE_USER, REMOTE_HOST), + "df -B1 --output=avail %s | tail -1" % shlex.quote(REMOTE_PATH) + ] + result = subprocess.run( + ssh_cmd, + capture_output=True, + text=True, + check=True, + timeout=15 + ) + avail_bytes = int(result.stdout.strip()) + if avail_bytes < file_size * 2: # Require 2x file size as buffer + logging.error("Insufficient disk space on Pi: %d bytes available, need %d", + avail_bytes, file_size * 2) + return False + logging.debug("Remote disk space OK: %d bytes available", avail_bytes) + return True + except Exception as e: + logging.warning("Could not check remote disk space: %s (proceeding anyway)", e) + return True # Non-blocking: proceed on check failure + @retry_on_failure() def _execute_rsync_with_retry(self, rsync_cmd: List[str], timeout_seconds: int) -> subprocess.CompletedProcess: """Execute rsync command with retry logic for transient failures. @@ -482,7 +564,7 @@ def _execute_usb_refresh_with_retry(self) -> bool: "-o", "ConnectTimeout=10", "-o", "ServerAliveInterval=5", "-o", "ServerAliveCountMax=3", - f"{REMOTE_USER}@{REMOTE_HOST}", + "%s@%s" % (REMOTE_USER, REMOTE_HOST), "sudo /usr/local/bin/refresh_usb_gadget.sh" ] @@ -495,7 +577,7 @@ def _execute_usb_refresh_with_retry(self) -> bool: ) logging.info("USB gadget refreshed successfully") if result.stdout: - logging.debug(f"Refresh output: {result.stdout.strip()}") + logging.debug("Refresh output: %s", result.stdout.strip()) return True def refresh_usb_gadget(self) -> bool: @@ -507,7 +589,7 @@ def refresh_usb_gadget(self) -> bool: try: result, attempts = self._execute_usb_refresh_with_retry() if attempts > 1: - logging.info(f"USB refresh succeeded after {attempts} attempts") + logging.info("USB refresh succeeded after %d attempts", attempts) return result except subprocess.TimeoutExpired: @@ -516,15 +598,15 @@ def refresh_usb_gadget(self) -> bool: return False except subprocess.CalledProcessError as e: - logging.error(f"USB gadget refresh failed after all retries (exit code {e.returncode})") + logging.error("USB gadget refresh failed after all retries (exit code %d)", e.returncode) if e.stderr: - logging.error(f"Error details: {e.stderr.strip()}") + logging.error("Error details: %s", e.stderr.strip()) logging.warning("File was synced but printer may not see it until Pi reboot") - logging.info(f"To manually refresh: ssh {REMOTE_USER}@{REMOTE_HOST} 'sudo /usr/local/bin/refresh_usb_gadget.sh'") + logging.info("To manually refresh: ssh %s@%s 'sudo /usr/local/bin/refresh_usb_gadget.sh'", REMOTE_USER, REMOTE_HOST) return False except Exception as e: - logging.error(f"Unexpected error during USB gadget refresh: {type(e).__name__}: {e}") + logging.error("Unexpected error during USB gadget refresh: %s: %s", type(e).__name__, e) logging.warning("File was synced but printer may not see it") return False @@ -557,20 +639,23 @@ def main() -> None: import watchdog logging.info("watchdog module loaded successfully. Continuing...") except subprocess.CalledProcessError as e: - logging.error(f"Failed to install watchdog: {e}") + logging.error("Failed to install watchdog: %s", e) logging.error("Please install manually: pip install -r requirements.txt") sys.exit(1) - # Create watch directory if it doesn't exist - os.makedirs(WATCH_DIR, exist_ok=True) + # Create watch directories if they don't exist + for watch_dir in WATCH_DIRS: + os.makedirs(watch_dir, exist_ok=True) - logging.info(f"Starting gcode file monitor on {WATCH_DIR}") - logging.info(f"Will sync to {REMOTE_USER}@{REMOTE_HOST}:{REMOTE_PORT}:{REMOTE_PATH}") + logging.info("Starting gcode file monitor on %s", WATCH_DIRS) + logging.info("Will sync to %s@%s:%s:%s", REMOTE_USER, REMOTE_HOST, REMOTE_PORT, REMOTE_PATH) - # Setup file system observer + # Setup file system observer for each watch directory event_handler = GCodeHandler() observer = Observer() - observer.schedule(event_handler, WATCH_DIR, recursive=False) + for watch_dir in WATCH_DIRS: + observer.schedule(event_handler, watch_dir, recursive=False) + logging.info("Watching directory: %s", watch_dir) # Start monitoring observer.start() diff --git a/monitor_and_sync.sh b/monitor_and_sync.sh index a92eeb0..3729b2f 100755 --- a/monitor_and_sync.sh +++ b/monitor_and_sync.sh @@ -1,4 +1,11 @@ #!/bin/bash +# =========================================================================== +# DEPRECATED: This bash script is deprecated. Use monitor_and_sync.py instead. +# The Python version provides better security controls, retry logic, and +# file validation. This script is kept for reference only. +# =========================================================================== + +echo "WARNING: monitor_and_sync.sh is DEPRECATED. Use monitor_and_sync.py instead." >&2 # Determine script directory SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" diff --git a/tests/integration/test_integration.py b/tests/integration/test_integration.py index 5f08062..19d6abe 100644 --- a/tests/integration/test_integration.py +++ b/tests/integration/test_integration.py @@ -352,5 +352,109 @@ def test_shell_script_syntax_valid(self): self.assertTrue(os.path.isfile(script)) +class TestDeployScriptUsesInstallService(unittest.TestCase): + """Test that deploy.sh delegates to install_service.sh instead of raw cp""" + + def test_deploy_does_not_raw_copy_service_file(self): + """deploy.sh must NOT contain cp of gcode-monitor.service to /etc/systemd/system/""" + import re + with open('deploy.sh', 'r') as f: + content = f.read() + raw_cp_pattern = re.compile( + r'^\s*cp\s+.*gcode-monitor\.service.*(/etc/systemd/system/|/etc/systemd/system)', + re.MULTILINE + ) + match = raw_cp_pattern.search(content) + self.assertIsNone(match, "deploy.sh must not raw-copy the template service file") + + def test_deploy_calls_install_service(self): + """deploy.sh must delegate to install_service.sh""" + with open('deploy.sh', 'r') as f: + content = f.read() + self.assertIn('install_service.sh', content) + + def test_deploy_step_numbering_consistent(self): + """deploy.sh step numbering should be sequential [1/5] through [5/5]""" + import re + with open('deploy.sh', 'r') as f: + content = f.read() + steps = re.findall(r'\[(\d+/\d+)\]', content) + self.assertTrue(len(steps) >= 5, f"Expected at least 5 steps, got {len(steps)}") + for i, step in enumerate(steps, 1): + self.assertEqual(step, f"{i}/5", f"Step {i} should be [{i}/5], got [{step}]") + + +class TestHealthCheckAlerting(unittest.TestCase): + """Test that check_gcode_monitor.sh sends alerts on failure""" + + def test_health_check_has_send_alert_function(self): + """check_gcode_monitor.sh must define a send_alert function""" + with open('check_gcode_monitor.sh', 'r') as f: + content = f.read() + self.assertIn('send_alert', content) + + def test_health_check_uses_notify_send(self): + """check_gcode_monitor.sh must use notify-send for desktop alerts""" + with open('check_gcode_monitor.sh', 'r') as f: + content = f.read() + self.assertIn('notify-send', content) + + def test_health_check_alerts_on_service_down(self): + """send_alert must be called in check_service_status error path""" + with open('check_gcode_monitor.sh', 'r') as f: + content = f.read() + lines = content.split('\n') + in_func = False + found = False + for line in lines: + if 'check_service_status()' in line: + in_func = True + if in_func and 'send_alert' in line: + found = True + break + if in_func and line.strip() == '}': + in_func = False + self.assertTrue(found, "send_alert must be called in check_service_status error path") + + def test_health_check_alerts_on_errors(self): + """send_alert must be called in check_recent_errors error path""" + with open('check_gcode_monitor.sh', 'r') as f: + content = f.read() + lines = content.split('\n') + in_func = False + found = False + for line in lines: + if 'check_recent_errors()' in line: + in_func = True + if in_func and 'send_alert' in line: + found = True + break + if in_func and line.strip() == '}': + in_func = False + self.assertTrue(found, "send_alert must be called in check_recent_errors error path") + + +class TestInstallAndStartUsesInstallService(unittest.TestCase): + """Test that install_and_start.sh delegates to install_service.sh""" + + def test_install_and_start_does_not_raw_copy_service_file(self): + """install_and_start.sh must NOT contain cp of gcode-monitor.service to /etc/systemd/system/""" + import re + with open('install_and_start.sh', 'r') as f: + content = f.read() + raw_cp_pattern = re.compile( + r'^\s*(sudo\s+)?cp\s+.*gcode-monitor\.service.*(/etc/systemd/system/|/etc/systemd/system)', + re.MULTILINE + ) + match = raw_cp_pattern.search(content) + self.assertIsNone(match, "install_and_start.sh must not raw-copy the template service file") + + def test_install_and_start_calls_install_service(self): + """install_and_start.sh must delegate to install_service.sh""" + with open('install_and_start.sh', 'r') as f: + content = f.read() + self.assertIn('install_service.sh', content) + + if __name__ == '__main__': unittest.main() diff --git a/tests/security/test_security.py b/tests/security/test_security.py index eccf17c..e532380 100644 --- a/tests/security/test_security.py +++ b/tests/security/test_security.py @@ -268,7 +268,7 @@ def test_network_restrictions(self): self.assertIn('RestrictAddressFamilies=AF_INET AF_INET6', content) self.assertIn('IPAddressDeny=any', content) self.assertIn('IPAddressAllow=localhost', content) - self.assertIn('IPAddressAllow=192.168.1.0/24', content) + self.assertIn('IPAddressAllow=%LOCAL_SUBNET%', content) def test_filesystem_restrictions(self): """Test that filesystem access is restricted""" @@ -299,6 +299,24 @@ def test_syscall_filtering(self): self.assertIn('SystemCallFilter=@system-service', content) self.assertIn('SystemCallFilter=~@privileged', content) + def test_syscall_filter_uncommented(self): + """SystemCallFilter lines must be uncommented (active)""" + with open('gcode-monitor.service', 'r') as f: + lines = f.readlines() + uncommented_scf = [l for l in lines + if l.strip().startswith('SystemCallFilter=')] + self.assertTrue(len(uncommented_scf) >= 2, + "Expected at least 2 uncommented SystemCallFilter= lines") + + def test_syscall_architectures_uncommented(self): + """SystemCallArchitectures=native must be uncommented""" + with open('gcode-monitor.service', 'r') as f: + lines = f.readlines() + uncommented = [l for l in lines + if l.strip().startswith('SystemCallArchitectures=native')] + self.assertEqual(len(uncommented), 1, + "Expected 1 uncommented SystemCallArchitectures=native line") + class TestInputValidation(unittest.TestCase): """Test input validation for configuration""" @@ -334,5 +352,21 @@ def test_config_path_validation(self): self.assertTrue(Path(absolute_path).is_absolute()) +class TestBashScriptDeprecation(unittest.TestCase): + """Test that monitor_and_sync.sh is deprecated""" + + def test_bash_script_has_deprecation_notice(self): + """monitor_and_sync.sh must contain DEPRECATED notice""" + with open('monitor_and_sync.sh', 'r') as f: + content = f.read() + self.assertIn('DEPRECATED', content.upper()) + + def test_bash_script_references_python_version(self): + """monitor_and_sync.sh must reference monitor_and_sync.py""" + with open('monitor_and_sync.sh', 'r') as f: + content = f.read() + self.assertIn('monitor_and_sync.py', content) + + if __name__ == '__main__': unittest.main() diff --git a/tests/unit/test_validation.py b/tests/unit/test_validation.py index b20b6a5..0cb3e01 100644 --- a/tests/unit/test_validation.py +++ b/tests/unit/test_validation.py @@ -249,7 +249,7 @@ def test_on_created_does_not_deadlock(self): """on_created should return promptly without holding the lock""" # Import here to ensure configuration is loaded as in production sys.modules.pop("monitor_and_sync", None) - with patch("logging.FileHandler", return_value=logging.NullHandler()): + with patch("logging.handlers.RotatingFileHandler", return_value=logging.NullHandler()): monitor_and_sync = importlib.import_module("monitor_and_sync") handler = monitor_and_sync.GCodeHandler() @@ -269,15 +269,17 @@ class TestRsyncDestinationQuoting(unittest.TestCase): """Ensure rsync destination is safely quoted for remote paths.""" def setUp(self): - self.filehandler_patch = patch("logging.FileHandler", return_value=logging.NullHandler()) + self.filehandler_patch = patch("logging.handlers.RotatingFileHandler", return_value=logging.NullHandler()) self.filehandler_patch.start() self.addCleanup(self.filehandler_patch.stop) self.module = importlib.import_module("monitor_and_sync") self.temp_dir = tempfile.mkdtemp() self.original_watch_dir = self.module.WATCH_DIR + self.original_watch_dirs = self.module.WATCH_DIRS self.original_remote_path = self.module.REMOTE_PATH self.module.WATCH_DIR = self.temp_dir + self.module.WATCH_DIRS = [self.temp_dir] self.stats_output = textwrap.dedent(""" Number of files: 1 (reg: 1) Number of created files: 1 (reg: 1) @@ -296,6 +298,7 @@ def setUp(self): def tearDown(self): self.module.WATCH_DIR = self.original_watch_dir + self.module.WATCH_DIRS = self.original_watch_dirs self.module.REMOTE_PATH = self.original_remote_path if self.file_path.exists(): self.file_path.unlink() @@ -386,7 +389,7 @@ def fake_filehandler(path, *args, **kwargs): self.assertTrue(Path(path).parent.exists(), "Log directory should exist before FileHandler initialization") return logging.NullHandler() - with patch("logging.FileHandler", side_effect=fake_filehandler): + with patch("logging.handlers.RotatingFileHandler", side_effect=fake_filehandler): importlib.import_module("monitor_and_sync") finally: config_path.write_text(original_config, encoding="utf-8") @@ -394,5 +397,120 @@ def fake_filehandler(path, *args, **kwargs): shutil.rmtree(temp_root.parent, ignore_errors=True) +class TestCodeQuality(unittest.TestCase): + """Test code quality standards in monitor_and_sync.py""" + + def test_no_duplicate_import_re(self): + """import re should appear exactly once""" + with open('monitor_and_sync.py', 'r') as f: + lines = f.readlines() + import_re_lines = [l for l in lines if l.strip() == 'import re'] + self.assertEqual(len(import_re_lines), 1, + f"Expected 1 'import re', found {len(import_re_lines)}") + + def test_no_duplicate_from_pathlib_import_path(self): + """from pathlib import Path should appear exactly once""" + with open('monitor_and_sync.py', 'r') as f: + lines = f.readlines() + pathlib_lines = [l for l in lines if l.strip() == 'from pathlib import Path'] + self.assertEqual(len(pathlib_lines), 1, + f"Expected 1 'from pathlib import Path', found {len(pathlib_lines)}") + + def test_send_notification_function_exists(self): + """monitor_and_sync.py must have send_notification function""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + self.assertIn('def send_notification', content) + + def test_disk_space_check_exists(self): + """monitor_and_sync.py must have check_remote_disk_space method""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + self.assertIn('check_remote_disk_space', content) + + def test_log_rotation_configured(self): + """monitor_and_sync.py must use RotatingFileHandler for log rotation""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + self.assertIn('RotatingFileHandler', content) + self.assertIn('maxBytes', content) + self.assertIn('backupCount', content) + + def test_send_notification_called_on_success(self): + """send_notification must be called after 'Successfully synced'""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + # After the "Successfully synced" log line, send_notification should appear + idx_synced = content.index('Successfully synced') + idx_notify = content.index('send_notification', idx_synced) + self.assertGreater(idx_notify, idx_synced) + + def test_send_notification_called_multiple_times(self): + """send_notification must be called in both success and failure paths""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + count = content.count('send_notification(') + # At least 4: function def + success + 2 failure except blocks + self.assertGreaterEqual(count, 4, + f"Expected send_notification called at least 4 times (def + success + failure paths), found {count}") + + def test_delete_after_sync_in_config_example(self): + """config.example must document DELETE_AFTER_SYNC option""" + with open('config.example', 'r') as f: + content = f.read() + self.assertIn('DELETE_AFTER_SYNC', content) + + def test_delete_after_sync_in_monitor(self): + """monitor_and_sync.py must read DELETE_AFTER_SYNC config with default""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + self.assertIn('DELETE_AFTER_SYNC', content) + # Must use config.get() with a default value (not config['DELETE_AFTER_SYNC']) + self.assertIn(".get('DELETE_AFTER_SYNC'", content) + + def test_delete_after_sync_logic_exists(self): + """monitor_and_sync.py must have os.remove logic gated by DELETE_AFTER_SYNC""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + self.assertIn('DELETE_AFTER_SYNC', content) + self.assertIn('os.remove(', content) + + def test_config_example_documents_multiple_dirs(self): + """config.example must document comma-separated WATCH_DIR support""" + with open('config.example', 'r') as f: + content = f.read() + # Must mention comma-separated directories + self.assertTrue('comma' in content.lower() or 'multiple' in content.lower(), + "config.example must document comma-separated directory support") + + def test_watch_dir_split_in_monitor(self): + """monitor_and_sync.py must split WATCH_DIR for multiple directory support""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + # Must call .split(',') or .split(', ') on WATCH_DIR value + self.assertIn(".split(',')", content, + "monitor_and_sync.py must split WATCH_DIR by comma for multi-dir support") + + def test_observer_scheduled_in_loop(self): + """observer.schedule must be called in a loop over watch directories""" + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + # Must have a loop scheduling observers for each WATCH_DIR + self.assertIn('for watch_dir in WATCH_DIRS', content) + self.assertIn('observer.schedule(event_handler, watch_dir', content) + + def test_no_fstrings_in_logging_calls(self): + """Logging calls must use lazy %s formatting, not f-strings""" + import re + with open('monitor_and_sync.py', 'r') as f: + content = f.read() + fstring_log_pattern = re.compile( + r'logging\.(debug|info|warning|error|critical)\(f["\']' + ) + matches = fstring_log_pattern.findall(content) + self.assertEqual(len(matches), 0, + f"Found {len(matches)} f-string logging calls; use %s lazy formatting instead") + + if __name__ == '__main__': unittest.main()