253 lines
8.8 KiB
Python
Executable File
253 lines
8.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Improved WinRM Script Executor
|
|
Handles large scripts by uploading them in chunks
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import argparse
|
|
import json
|
|
import time
|
|
import base64
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from winrm import Session
|
|
import logging
|
|
|
|
|
|
class WinRMExecutorV2:
|
|
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-v2-{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_script_chunked(self, local_script_path, remote_script_name):
|
|
"""Upload script in chunks to avoid command line length limits"""
|
|
try:
|
|
# Read the local script
|
|
with open(local_script_path, 'r', encoding='utf-8') as f:
|
|
script_content = f.read()
|
|
|
|
self.logger.info(f"Uploading script in chunks: {local_script_path} -> {remote_script_name}")
|
|
|
|
# Encode content as base64 to avoid special character issues
|
|
script_bytes = script_content.encode('utf-8')
|
|
script_b64 = base64.b64encode(script_bytes).decode('ascii')
|
|
|
|
# Split into chunks of 8000 chars (safe for command line)
|
|
chunk_size = 8000
|
|
chunks = [script_b64[i:i+chunk_size] for i in range(0, len(script_b64), chunk_size)]
|
|
|
|
self.logger.info(f"Uploading {len(chunks)} chunks...")
|
|
|
|
# Create the file and append chunks
|
|
for i, chunk in enumerate(chunks):
|
|
if i == 0:
|
|
# First chunk - create file
|
|
ps_command = f"""
|
|
$chunk = '{chunk}'
|
|
$chunk | Out-File -FilePath '{remote_script_name}.b64' -Encoding ASCII
|
|
"""
|
|
else:
|
|
# Subsequent chunks - append
|
|
ps_command = f"""
|
|
$chunk = '{chunk}'
|
|
$chunk | Out-File -FilePath '{remote_script_name}.b64' -Encoding ASCII -Append
|
|
"""
|
|
|
|
result = self.session.run_ps(ps_command)
|
|
|
|
if result.status_code != 0:
|
|
self.logger.error(f"Failed to upload chunk {i+1}: {result.std_err.decode()}")
|
|
return False
|
|
|
|
if (i + 1) % 10 == 0:
|
|
self.logger.info(f"Uploaded {i+1}/{len(chunks)} chunks...")
|
|
|
|
# Decode the base64 file to create the final script
|
|
decode_command = f"""
|
|
$b64Content = Get-Content '{remote_script_name}.b64' -Raw
|
|
$bytes = [System.Convert]::FromBase64String($b64Content)
|
|
|
|
# Write with UTF-8 BOM for Chinese Windows compatibility
|
|
$utf8BOM = New-Object System.Text.UTF8Encoding($true)
|
|
[System.IO.File]::WriteAllBytes('{remote_script_name}', $bytes)
|
|
|
|
# Clean up temporary file
|
|
Remove-Item '{remote_script_name}.b64' -ErrorAction SilentlyContinue
|
|
|
|
Write-Host "Script uploaded successfully to {remote_script_name}"
|
|
"""
|
|
|
|
result = self.session.run_ps(decode_command)
|
|
|
|
if result.status_code == 0:
|
|
self.logger.info("✅ Script uploaded and decoded successfully")
|
|
return True
|
|
else:
|
|
self.logger.error(f"Failed to decode script: {result.std_err.decode()}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"❌ Script upload failed: {e}")
|
|
return False
|
|
|
|
def execute_script(self, remote_script_name):
|
|
"""Execute the uploaded PowerShell script"""
|
|
try:
|
|
self.logger.info(f"Executing PowerShell script: {remote_script_name}")
|
|
|
|
# Execute the script
|
|
execute_command = f"PowerShell -NoProfile -ExecutionPolicy Bypass -File '{remote_script_name}'"
|
|
result = self.session.run_cmd(execute_command)
|
|
|
|
# Output results in real-time
|
|
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)
|
|
|
|
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}")
|
|
|
|
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 upload_and_execute_script(self, local_script_path, remote_script_name=None):
|
|
"""Upload and execute a PowerShell script using chunked upload"""
|
|
if not self.session:
|
|
self.logger.error("No active WinRM session")
|
|
return False
|
|
|
|
# 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}"
|
|
|
|
# Upload script
|
|
if not self.upload_script_chunked(local_script_path, remote_script_name):
|
|
return False
|
|
|
|
# Execute script
|
|
return self.execute_script(remote_script_name)
|
|
|
|
def close(self):
|
|
"""Close the WinRM session"""
|
|
if self.session:
|
|
self.logger.info("Closing WinRM session")
|
|
self.session = None
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Execute PowerShell scripts via WinRM (v2)')
|
|
parser.add_argument('--host', default=os.getenv('HOST', '192.168.88.108'),
|
|
help='Target host')
|
|
parser.add_argument('--username', default=os.getenv('USER_NAME', 'Admin'),
|
|
help='Username')
|
|
parser.add_argument('--password', default=os.getenv('PASS', 'P@ssw0rd!'),
|
|
help='Password')
|
|
parser.add_argument('--script', required=True,
|
|
help='PowerShell script file to execute')
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Create executor
|
|
executor = WinRMExecutorV2(
|
|
host=args.host,
|
|
username=args.username,
|
|
password=args.password
|
|
)
|
|
|
|
try:
|
|
# Connect
|
|
if not executor.connect():
|
|
sys.exit(1)
|
|
|
|
# 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:
|
|
sys.exit(1)
|
|
|
|
finally:
|
|
executor.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |