Initial commit
This commit is contained in:
250
scripts/winrm/winrm_executor.py
Executable file
250
scripts/winrm/winrm_executor.py
Executable file
@@ -0,0 +1,250 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user