#!/usr/bin/env python3 """ WinRM Script Executor A Python-based alternative to SSH for executing PowerShell scripts on Windows machines. """ import os import sys import argparse import json import time from datetime import datetime from pathlib import Path from winrm import Session import logging class WinRMExecutor: def __init__(self, host, username, password, port=5985, use_https=False): self.host = host self.username = username self.password = password self.port = port self.use_https = use_https self.session = None # Setup logging self.setup_logging() def setup_logging(self): """Setup logging configuration""" log_dir = Path(__file__).parent / "../../logs" log_dir.mkdir(exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") log_file = log_dir / f"winrm-{self.host}-{timestamp}.log" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) self.logger.info(f"Starting WinRM session to {self.host}") def connect(self): """Establish WinRM connection""" try: protocol = 'https' if self.use_https else 'http' endpoint = f'{protocol}://{self.host}:{self.port}/wsman' self.logger.info(f"Connecting to {endpoint}") self.session = Session( endpoint, auth=(self.username, self.password), transport='ntlm' if not self.use_https else 'ssl', server_cert_validation='ignore' if self.use_https else 'validate' ) # Test connection with a simple command result = self.session.run_cmd('echo Connection test') if result.status_code == 0: self.logger.info("✅ WinRM connection established successfully") return True else: self.logger.error(f"❌ Connection test failed: {result.std_err.decode()}") return False except Exception as e: self.logger.error(f"❌ Failed to connect: {e}") return False def upload_and_execute_script(self, local_script_path, remote_script_name=None): """Upload and execute a PowerShell script""" if not self.session: self.logger.error("No active WinRM session") return False try: # Read the local script with open(local_script_path, 'r', encoding='utf-8') as f: script_content = f.read() # Generate remote script name if not provided if not remote_script_name: script_name = Path(local_script_path).name remote_script_name = f"C:\\Users\\{self.username}\\{script_name}" self.logger.info(f"Uploading script: {local_script_path} -> {remote_script_name}") # Create the script on remote machine create_script = f""" $scriptContent = @' {script_content} '@ # Ensure UTF-8 encoding with BOM for Chinese Windows $utf8BOM = New-Object System.Text.UTF8Encoding($true) [System.IO.File]::WriteAllText('{remote_script_name}', $scriptContent, $utf8BOM) Write-Host "Script uploaded to {remote_script_name}" """ upload_result = self.session.run_ps(create_script) if upload_result.status_code != 0: self.logger.error(f"Failed to upload script: {upload_result.std_err.decode()}") return False self.logger.info("✅ Script uploaded successfully") # Execute the script self.logger.info(f"Executing PowerShell script: {remote_script_name}") execute_command = f"PowerShell -NoProfile -ExecutionPolicy Bypass -File '{remote_script_name}'" result = self.session.run_cmd(execute_command) # Output results if result.std_out: output = result.std_out.decode('utf-8', errors='replace') self.logger.info("Script Output:") for line in output.split('\n'): if line.strip(): print(line) self.logger.info(line) if result.std_err: error = result.std_err.decode('utf-8', errors='replace') if error.strip(): self.logger.warning("Script Errors:") for line in error.split('\n'): if line.strip(): print(f"ERROR: {line}") self.logger.warning(line) success = result.status_code == 0 status = "✅ SUCCESS" if success else f"❌ FAILED (exit code: {result.status_code})" self.logger.info(f"Script execution completed: {status}") return success except Exception as e: self.logger.error(f"❌ Script execution failed: {e}") return False def execute_command(self, command, use_powershell=False): """Execute a single command""" if not self.session: self.logger.error("No active WinRM session") return False try: self.logger.info(f"Executing command: {command}") if use_powershell: result = self.session.run_ps(command) else: result = self.session.run_cmd(command) # Output results if result.std_out: output = result.std_out.decode('utf-8', errors='replace') for line in output.split('\n'): if line.strip(): print(line) if result.std_err: error = result.std_err.decode('utf-8', errors='replace') if error.strip(): for line in error.split('\n'): if line.strip(): print(f"ERROR: {line}") return result.status_code == 0 except Exception as e: self.logger.error(f"Command execution failed: {e}") return False def close(self): """Close the WinRM session""" if self.session: self.logger.info("Closing WinRM session") # Note: pywinrm doesn't have an explicit close method self.session = None def main(): parser = argparse.ArgumentParser(description='Execute PowerShell scripts via WinRM') parser.add_argument('--host', default=os.getenv('HOST', '192.168.88.108'), help='Target host (default: from HOST env var or 192.168.88.108)') parser.add_argument('--username', default=os.getenv('USER_NAME', 'Admin'), help='Username (default: from USER_NAME env var or Admin)') parser.add_argument('--password', default=os.getenv('PASS', 'P@ssw0rd!'), help='Password (default: from PASS env var)') parser.add_argument('--port', type=int, default=5985, help='WinRM port (default: 5985)') parser.add_argument('--https', action='store_true', help='Use HTTPS instead of HTTP') parser.add_argument('--script', required=True, help='PowerShell script file to execute') parser.add_argument('--test-connection', action='store_true', help='Test connection only, do not execute script') args = parser.parse_args() # Create executor executor = WinRMExecutor( host=args.host, username=args.username, password=args.password, port=args.port, use_https=args.https ) try: # Connect if not executor.connect(): print("❌ Failed to establish WinRM connection") sys.exit(1) if args.test_connection: print("✅ WinRM connection test successful") return # Execute script script_path = Path(args.script) if not script_path.exists(): print(f"❌ Script file not found: {script_path}") sys.exit(1) success = executor.upload_and_execute_script(script_path) if success: print(f"\n🎉 Script execution completed successfully") else: print(f"\n❌ Script execution failed") sys.exit(1) finally: executor.close() if __name__ == "__main__": main()