Refactor: Centralize command patterns in single source of truth

CRITICAL: Prevents inconsistent sudo/SSH patterns across codebase.

Created command_patterns.py with:
- Single source of truth for ALL command execution patterns
- SSH key path constant: /var/lib/macha/.ssh/id_ed25519
- Remote user constant: macha
- sudo prefix for all remote commands
- Helper functions: build_ssh_command(), transform_ssh_command()
- Self-validation tests

Updated files to use centralized patterns:
- tools.py: Uses transform_ssh_command()
- remote_monitor.py: Uses build_ssh_command()
- system_discovery.py: Uses build_ssh_command()
- DESIGN.md: Documents centralized approach

Benefits:
- Impossible to have inconsistent patterns
- Single place to update if needed
- Self-documenting with validation tests
- Prevents future refactoring errors

DO NOT duplicate these patterns in other files - always import.
This commit is contained in:
Lily Miller
2025-10-06 16:06:31 -06:00
parent ab72a98849
commit 2f367f7cdc
5 changed files with 236 additions and 33 deletions

View File

@@ -26,9 +26,12 @@ Macha is an AI-powered autonomous system administrator capable of monitoring, ma
**Macha CAN and SHOULD use SSH to manage other hosts.** **Macha CAN and SHOULD use SSH to manage other hosts.**
#### SSH Access #### SSH Access
- **CRITICAL**: Always uses explicit SSH key path: `-i /var/lib/macha/.ssh/id_ed25519` - **CRITICAL**: All command patterns defined in `command_patterns.py` (SINGLE SOURCE OF TRUTH)
- Always uses explicit SSH key path: `-i /var/lib/macha/.ssh/id_ed25519`
- All SSH commands automatically include the `-i` flag with absolute key path - All SSH commands automatically include the `-i` flag with absolute key path
- Remote commands always prefixed with `sudo`
- Runs as `macha` user (UID 2501) - Runs as `macha` user (UID 2501)
- **DO NOT DUPLICATE these patterns elsewhere** - import from `command_patterns.py`
- Has `NOPASSWD` sudo access for administrative commands - Has `NOPASSWD` sudo access for administrative commands
- Shares SSH keys with other hosts in the infrastructure - Shares SSH keys with other hosts in the infrastructure
- Can SSH to: `rhiannon`, `alexander`, `UCAR-Kinston`, and others in the flake - Can SSH to: `rhiannon`, `alexander`, `UCAR-Kinston`, and others in the flake

219
command_patterns.py Normal file
View File

@@ -0,0 +1,219 @@
#!/usr/bin/env python3
"""
Command Execution Patterns - SINGLE SOURCE OF TRUTH
DO NOT DUPLICATE THESE PATTERNS ELSEWHERE.
All command execution must use these functions to ensure consistency.
Pattern Rules:
1. SSH keys are ALWAYS explicit: -i /var/lib/macha/.ssh/id_ed25519
2. Remote commands ALWAYS use sudo: ssh user@host sudo command
3. Local commands run as the macha user (no sudo prefix needed when already macha)
"""
from typing import List, Dict, Any
import subprocess
# ============================================================================
# CONSTANTS - DO NOT MODIFY WITHOUT UPDATING DESIGN.MD
# ============================================================================
SSH_KEY_PATH = "/var/lib/macha/.ssh/id_ed25519"
SSH_OPTIONS = ["-o", "StrictHostKeyChecking=no"]
REMOTE_USER = "macha"
# ============================================================================
# SSH COMMAND CONSTRUCTION
# ============================================================================
def build_ssh_command(hostname: str, remote_command: str, timeout: int = 30) -> List[str]:
"""
Build SSH command with correct patterns.
Args:
hostname: Target hostname (e.g., 'rhiannon')
remote_command: Command to execute on remote host
timeout: Command timeout in seconds
Returns:
List of command arguments ready for subprocess
Example:
>>> build_ssh_command("rhiannon", "systemctl status ollama")
['ssh', '-i', '/var/lib/macha/.ssh/id_ed25519', '-o', 'StrictHostKeyChecking=no',
'-o', 'ConnectTimeout=10', 'macha@rhiannon', 'sudo systemctl status ollama']
"""
cmd = [
"ssh",
"-i", SSH_KEY_PATH,
*SSH_OPTIONS,
"-o", "ConnectTimeout=10",
f"{REMOTE_USER}@{hostname}",
f"sudo {remote_command}"
]
return cmd
def build_scp_command(hostname: str, source: str, dest: str, remote_to_local: bool = True) -> List[str]:
"""
Build SCP command with correct patterns.
Args:
hostname: Target hostname
source: Source path
dest: Destination path
remote_to_local: If True, copy from remote to local (default)
Returns:
List of command arguments ready for subprocess
"""
if remote_to_local:
source_spec = f"{REMOTE_USER}@{hostname}:{source}"
dest_spec = dest
else:
source_spec = source
dest_spec = f"{REMOTE_USER}@{hostname}:{dest}"
cmd = [
"scp",
"-i", SSH_KEY_PATH,
*SSH_OPTIONS,
source_spec,
dest_spec
]
return cmd
# ============================================================================
# COMMAND TRANSFORMATION (for tools.py)
# ============================================================================
def transform_ssh_command(command: str) -> str:
"""
Transform simplified SSH commands to full format.
Converts: "ssh hostname command args"
To: "ssh -i /path/to/key -o StrictHostKeyChecking=no macha@hostname sudo command args"
Args:
command: User-provided command string
Returns:
Transformed command string with proper SSH options
Note:
This is used by tools.py execute_command for string-based commands.
For new code, prefer build_ssh_command() which returns a list.
"""
if not command.strip().startswith('ssh '):
return command
parts = command.split(maxsplit=2)
if len(parts) < 2:
return command
# Check if already has @ (already transformed)
if '@' in parts[1]:
return command
hostname = parts[1]
remaining = parts[2] if len(parts) > 2 else ''
ssh_opts = f"-i {SSH_KEY_PATH} -o StrictHostKeyChecking=no"
if remaining:
return f"ssh {ssh_opts} {REMOTE_USER}@{hostname} sudo {remaining}"
else:
return f"ssh {ssh_opts} {REMOTE_USER}@{hostname}"
# ============================================================================
# EXECUTION HELPERS
# ============================================================================
def execute_ssh_command(
hostname: str,
command: str,
timeout: int = 30,
capture_output: bool = True
) -> Dict[str, Any]:
"""
Execute command on remote host via SSH.
Args:
hostname: Target hostname
command: Command to execute (will be prefixed with sudo automatically)
timeout: Command timeout in seconds
capture_output: Whether to capture stdout/stderr
Returns:
Dict with keys: success, stdout, stderr, exit_code
"""
ssh_cmd = build_ssh_command(hostname, command, timeout)
try:
result = subprocess.run(
ssh_cmd,
capture_output=capture_output,
text=True,
timeout=timeout
)
return {
"success": result.returncode == 0,
"stdout": result.stdout if capture_output else "",
"stderr": result.stderr if capture_output else "",
"exit_code": result.returncode
}
except subprocess.TimeoutExpired:
return {
"success": False,
"stdout": "",
"stderr": f"Command timed out after {timeout}s",
"exit_code": -1
}
except Exception as e:
return {
"success": False,
"stdout": "",
"stderr": str(e),
"exit_code": -1
}
# ============================================================================
# VALIDATION
# ============================================================================
def validate_patterns():
"""
Self-test to ensure patterns are correct.
Run this in tests to catch accidental modifications.
"""
# Test SSH command construction
cmd = build_ssh_command("testhost", "echo test")
assert "-i" in cmd, "SSH key flag missing"
assert SSH_KEY_PATH in cmd, "SSH key path missing"
assert "macha@testhost" in cmd, "Remote user@host missing"
assert "sudo echo test" in cmd, "sudo prefix missing"
# Test command transformation
transformed = transform_ssh_command("ssh rhiannon systemctl status ollama")
assert SSH_KEY_PATH in transformed, "Key path not added"
assert "macha@rhiannon" in transformed, "User not added"
assert "sudo systemctl" in transformed, "sudo not added"
print("✓ Command patterns validated")
if __name__ == "__main__":
# Run self-tests
validate_patterns()
# Show examples
print("\nExample SSH command:")
print(build_ssh_command("rhiannon", "systemctl status ollama"))
print("\nExample transformation:")
print(transform_ssh_command("ssh alexander df -h"))

View File

@@ -7,6 +7,7 @@ import json
import subprocess import subprocess
from typing import Dict, Any, Optional from typing import Dict, Any, Optional
from pathlib import Path from pathlib import Path
from command_patterns import build_ssh_command
class RemoteMonitor: class RemoteMonitor:
@@ -36,16 +37,8 @@ class RemoteMonitor:
(success, stdout, stderr) (success, stdout, stderr)
""" """
try: try:
# Use explicit SSH key path from macha user's home directory # Use centralized command pattern - see command_patterns.py
ssh_key = "/var/lib/macha/.ssh/id_ed25519" ssh_cmd = build_ssh_command(self.hostname, command, timeout)
ssh_cmd = [
"ssh",
"-i", ssh_key,
"-o", "StrictHostKeyChecking=no",
"-o", "ConnectTimeout=10",
self.ssh_target,
command
]
result = subprocess.run( result = subprocess.run(
ssh_cmd, ssh_cmd,

View File

@@ -9,6 +9,7 @@ import re
from typing import Dict, List, Set, Optional, Any from typing import Dict, List, Set, Optional, Any
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from command_patterns import build_ssh_command
class SystemDiscovery: class SystemDiscovery:
@@ -65,12 +66,10 @@ class SystemDiscovery:
def detect_os_type(self, hostname: str) -> str: def detect_os_type(self, hostname: str) -> str:
"""Detect the operating system of a remote host via SSH""" """Detect the operating system of a remote host via SSH"""
try: try:
# Use explicit SSH key path # Use centralized command pattern - see command_patterns.py
ssh_key = "/var/lib/macha/.ssh/id_ed25519" ssh_cmd = build_ssh_command(hostname, "cat /etc/os-release", timeout=10)
# Try to detect OS via SSH
result = subprocess.run( result = subprocess.run(
["ssh", "-i", ssh_key, "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=no", ssh_cmd,
hostname, "cat /etc/os-release"],
capture_output=True, capture_output=True,
text=True, text=True,
timeout=10 timeout=10
@@ -96,9 +95,9 @@ class SystemDiscovery:
return 'alpine' return 'alpine'
# Try uname for other systems # Try uname for other systems
ssh_cmd = build_ssh_command(hostname, "uname -s", timeout=10)
result = subprocess.run( result = subprocess.run(
["ssh", "-i", ssh_key, "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=no", ssh_cmd,
hostname, "uname -s"],
capture_output=True, capture_output=True,
text=True, text=True,
timeout=10 timeout=10

View File

@@ -8,6 +8,7 @@ import json
import os import os
from typing import Dict, Any, List, Optional from typing import Dict, Any, List, Optional
from pathlib import Path from pathlib import Path
from command_patterns import transform_ssh_command
class SysadminTools: class SysadminTools:
@@ -268,21 +269,9 @@ class SysadminTools:
"allowed_commands": self.allowed_commands "allowed_commands": self.allowed_commands
} }
# Automatically configure SSH commands to use macha user on remote systems # Automatically configure SSH commands using centralized command_patterns
# Transform: ssh hostname cmd -> ssh -i /var/lib/macha/.ssh/id_ed25519 macha@hostname sudo cmd # See command_patterns.py for the single source of truth
if command.strip().startswith('ssh ') and '@' not in command.split()[1]: command = transform_ssh_command(command)
parts = command.split(maxsplit=2)
if len(parts) >= 2:
hostname = parts[1]
remaining = ' '.join(parts[2:]) if len(parts) > 2 else ''
# Always use explicit SSH key path
ssh_key = "/var/lib/macha/.ssh/id_ed25519"
ssh_opts = f"-i {ssh_key} -o StrictHostKeyChecking=no"
# If there's a command to run remotely, prefix it with sudo
if remaining:
command = f"ssh {ssh_opts} macha@{hostname} sudo {remaining}".strip()
else:
command = f"ssh {ssh_opts} macha@{hostname}".strip()
try: try:
result = subprocess.run( result = subprocess.run(