250 lines
8.5 KiB
Python
Executable File
250 lines
8.5 KiB
Python
Executable File
#!/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() |