Imported from digitaleflex/GitHub-Cyber-Scanner-Pro (.agents/skills/pywinrm/SKILL.md). Install upstream with npx skills add digitaleflex/GitHub-Cyber-Scanner-Pro --skill pywinrm. Copyright stays with the author.

PyWinRM: Windows Remote Management Skill

Enterprise Windows Remote Administration
Connect to and manage Windows servers remotely using Python's PyWinRM library for WinRM protocol access.

What I Do

I am a comprehensive Windows remote management skill that provides full access to Windows servers through WinRM (Windows Remote Management). I enable:

Remote Command Execution: Execute PowerShell, CMD, and batch commands on remote Windows servers
System Monitoring: CPU, memory, disk, network, and process monitoring
Service Management: Start, stop, restart, and configure Windows services
Active Directory: User, group, computer, and OU management
Event Log Analysis: Query and analyze Windows event logs
Registry Operations: Read, write, and modify registry keys
File Operations: Copy, move, delete files; read/write content remotely
Software Management: Install, update, remove applications via WinRM
Scheduled Tasks: Create, modify, and manage Windows scheduled tasks
Performance Monitoring: Real-time performance counter collection
Security Auditing: Security policy analysis and compliance checking

When to Use Me

Use this skill when you need to:

Execute commands on remote Windows servers without RDP
Perform bulk operations across multiple Windows machines
Automate Windows system administration tasks
Monitor Windows server health and performance
Manage Active Directory objects remotely
Deploy software or configurations to Windows servers
Collect logs and diagnostics from Windows systems
Audit security configurations on Windows machines

Prerequisites

Python Environment

# Install pywinrm with all extras using uv
uv pip install pywinrm[credssp,kerberos]

# Or create a virtual environment
uv venv winrm-env
source winrm-env/bin/activate  # Unix
# winrm-env\Scripts\activate   # Windows
uv pip install pywinrm requests-ntlm requests-credssp

Windows Server Requirements

The target Windows server must have WinRM enabled:

# On target Windows server (run as Administrator)
Enable-PSRemoting -Force
winrm quickconfig -q
winrm set winrm/config/service '@{AllowUnencrypted="true"}'
winrm set winrm/config/service/auth '@{Basic="true"}'

# For HTTPS (recommended for production)
$cert = New-SelfSignedCertificate -DnsName "server.domain.com" -CertStoreLocation Cert:\LocalMachine\My
winrm create winrm/config/Listener?Address=*+Transport=HTTPS "@{Hostname=`"server.domain.com`";CertificateThumbprint=`"$($cert.Thumbprint)`"}"

# Open firewall
New-NetFirewallRule -Name "WinRM-HTTP" -DisplayName "WinRM HTTP" -Enabled True -Direction Inbound -Protocol TCP -LocalPort 5985 -Action Allow
New-NetFirewallRule -Name "WinRM-HTTPS" -DisplayName "WinRM HTTPS" -Enabled True -Direction Inbound -Protocol TCP -LocalPort 5986 -Action Allow

Connection Configuration

Known Windows Servers

Server	IP	Username	Password	Port
AWS Windows 1	44.197.31.152	Administrator	&@RSp14muPii*SdILEISHr&BqEu6;j2t	5985
AWS Windows 2	52.3.242.251	Administrator	hcDfrOjKrvK9&;;CH1N.BxH@)AglPN2Dw	5985
21CTL Windows	80.248.0.66	Administrator	data_21ctl@123	5985
21CTL Windows	80.248.0.66	superagent	@@Igbosere186@@	5985

Connection Methods

# Basic Auth (HTTP)
from winrm.protocol import Protocol

p = Protocol(
    endpoint='http://IP_ADDRESS:5985/wsman',
    transport='basic',
    username='Administrator',
    password='PASSWORD'
)

# NTLM Auth (Recommended for domains)
from winrm import Session

session = Session(
    'IP_ADDRESS',
    auth=('DOMAIN\\username', 'password'),
    transport='ntlm'
)

# HTTPS with certificate verification disabled (testing)
session = Session(
    'IP_ADDRESS',
    auth=('Administrator', 'password'),
    transport='basic',
    server_cert_validation='ignore'
)

# CredSSP (for double-hop scenarios)
session = Session(
    'IP_ADDRESS',
    auth=('Administrator', 'password'),
    transport='credssp',
    server_cert_validation='ignore'
)

Core Operations

1. Execute PowerShell Commands

#!/usr/bin/env python3
"""Execute PowerShell commands on remote Windows server."""
import winrm

def execute_powershell(host, username, password, command, port=5985):
    """Execute PowerShell command and return output."""
    session = winrm.Session(
        f'http://{host}:{port}/wsman',
        auth=(username, password),
        transport='basic'
    )
    
    # Encode command for PowerShell
    ps_script = f'''
    $ErrorActionPreference = "Stop"
    try {{
        {command}
    }} catch {{
        Write-Error $_.Exception.Message
        exit 1
    }}
    '''
    
    result = session.run_ps(ps_script)
    
    return {
        'status_code': result.status_code,
        'stdout': result.std_out.decode('utf-8', errors='replace'),
        'stderr': result.std_err.decode('utf-8', errors='replace')
    }

# Example usage
if __name__ == '__main__':
    result = execute_powershell(
        host='44.197.31.152',
        username='Administrator',
        password='YOUR_PASSWORD',
        command='Get-ComputerInfo | Select-Object CsName, WindowsVersion, OsArchitecture'
    )
    print(result['stdout'])

2. Execute CMD Commands

#!/usr/bin/env python3
"""Execute CMD commands on remote Windows server."""
import winrm

def execute_cmd(host, username, password, command, port=5985):
    """Execute CMD command and return output."""
    session = winrm.Session(
        f'http://{host}:{port}/wsman',
        auth=(username, password),
        transport='basic'
    )
    
    result = session.run_cmd(command)
    
    return {
        'status_code': result.status_code,
        'stdout': result.std_out.decode('utf-8', errors='replace'),
        'stderr': result.std_err.decode('utf-8', errors='replace')
    }

# Example
result = execute_cmd('server', 'user', 'pass', 'systeminfo')
print(result['stdout'])

3. System Information Collection

#!/usr/bin/env python3
"""Collect comprehensive system information from Windows server."""
import winrm
import json

def get_system_info(host, username, password, port=5985):
    """Collect system information."""
    session = winrm.Session(
        f'http://{host}:{port}/wsman',
        auth=(username, password),
        transport='basic'
    )
    
    ps_script = '''
    $info = @{
        ComputerName = $env:COMPUTERNAME
        Domain = (Get-WmiObject Win32_ComputerSystem).Domain
        OS = (Get-WmiObject Win32_OperatingSystem).Caption
        Version = (Get-WmiObject Win32_OperatingSystem).Version
        Architecture = (Get-WmiObject Win32_OperatingSystem).OSArchitecture
        LastBootTime = (Get-WmiObject Win32_OperatingSystem).LastBootUpTime
        TotalMemoryGB = [math]::Round((Get-WmiObject Win32_ComputerSystem).TotalPhysicalMemory / 1GB, 2)
        Processors = (Get-WmiObject Win32_Processor | Measure-Object).Count
        ProcessorName = (Get-WmiObject Win32_Processor | Select-Object -First 1).Name
        IPAddresses = (Get-NetIPAddress -AddressFamily IPv4 | Where-Object {$_.IPAddress -ne '127.0.0.1'}).IPAddress
    }
    $info | ConvertTo-Json -Depth 3
    '''
    
    result = session.run_ps(ps_script)
    if result.status_code == 0:
        return json.loads(result.std_out.decode('utf-8'))
    else:
        raise Exception(result.std_err.decode('utf-8'))

# Example
info = get_system_info('server', 'user', 'pass')
print(json.dumps(info, indent=2))

4. Performance Monitoring

#!/usr/bin/env python3
"""Monitor Windows server performance metrics."""
import winrm
import json

def get_performance_metrics(host, username, password, port=5985):
    """Get current performance metrics."""
    session = winrm.Session(
        f'http://{host}:{port}/wsman',
        auth=(username, password),
        transport='basic'
    )
    
    ps_script = '''
    $metrics = @{
        Timestamp = Get-Date -Format "yyyy-MM-dd HH:mm:ss"
        CPU = @{
            UsagePercent = [math]::Round((Get-Counter '\Processor(_Total)\% Processor Time').CounterSamples.CookedValue, 2)
        }
        Memory = @{
            TotalGB = [math]::Round((Get-WmiObject Win32_ComputerSystem).TotalPhysicalMemory / 1GB, 2)
            AvailableGB = [math]::Round((Get-Counter '\Memory\Available MBytes').CounterSamples.CookedValue / 1024, 2)
            UsagePercent = [math]::Round(100 - ((Get-Counter '\Memory\Available MBytes').CounterSamples.CookedValue / ((Get-WmiObject Win32_ComputerSystem).TotalPhysicalMemory / 1MB) * 100), 2)
        }
        Disk = Get-WmiObject Win32_LogicalDisk -Filter "DriveType=3" | ForEach-Object {
            @{
                Drive = $_.DeviceID
                SizeGB = [math]::Round($_.Size / 1GB, 2)
                FreeGB = [math]::Round($_.FreeSpace / 1GB, 2)
                UsagePercent = [math]::Round(100 - ($_.FreeSpace / $_.Size * 100), 2)
            }
        }
        Network = Get-NetAdapter | Where-Object {$_.Status -eq 'Up'} | ForEach-Object {
            $stats = Get-NetAdapterStatistics -Name $_.Name
            @{
                Name = $_.Name
                LinkSpeedMbps = [math]::Round($_.LinkSpeed / 1000000, 0)
                BytesSent = $stats.SentBytes
                BytesReceived = $stats.ReceivedBytes
            }
        }
    }
    $metrics | ConvertTo-Json -Depth 4
    '''
    
    result = session.run_ps(ps_script)
    if result.status_code == 0:
        return json.loads(result.std_out.decode('utf-8'))
    else:
        raise Exception(result.std_err.decode('utf-8'))

# Example
metrics = get_performance_metrics('server', 'user', 'pass')
print(json.dumps(metrics, indent=2))

5. Service Management

#!/usr/bin/env python3
"""Manage Windows services remotely."""
import winrm
import json

class ServiceManager:
    def __init__(self, host, username, password, port=5985):
        self.session = winrm.Session(
            f'http://{host}:{port}/wsman',
            auth=(username, password),
            transport='basic'
        )
    
    def list_services(self, filter_status=None):
        """List all services, optionally filtered by status."""
        filter_clause = ""
        if filter_status:
            filter_clause = f"| Where-Object {{$_.Status -eq '{filter_status}'}}"
        
        ps_script = f'''
        Get-Service {filter_clause} | Select-Object Name, DisplayName, Status, StartType | 
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_service(self, name):
        """Get details for a specific service."""
        ps_script = f'''
        $svc = Get-Service -Name "{name}" -ErrorAction SilentlyContinue
        if ($svc) {{
            $wmi = Get-WmiObject Win32_Service -Filter "Name='{name}'"
            @{{
                Name = $svc.Name
                DisplayName = $svc.DisplayName
                Status = $svc.Status.ToString()
                StartType = $svc.StartType.ToString()
                Path = $wmi.PathName
                Account = $wmi.StartName
                PID = $wmi.ProcessId
            }} | ConvertTo-Json
        }} else {{
            Write-Error "Service not found"
        }}
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def start_service(self, name):
        """Start a service."""
        ps_script = f'Start-Service -Name "{name}" -PassThru | Select-Object Name, Status | ConvertTo-Json'
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def stop_service(self, name):
        """Stop a service."""
        ps_script = f'Stop-Service -Name "{name}" -Force -PassThru | Select-Object Name, Status | ConvertTo-Json'
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def restart_service(self, name):
        """Restart a service."""
        ps_script = f'Restart-Service -Name "{name}" -PassThru | Select-Object Name, Status | ConvertTo-Json'
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def set_startup_type(self, name, startup_type):
        """Set service startup type (Automatic, Manual, Disabled)."""
        ps_script = f'Set-Service -Name "{name}" -StartupType {startup_type} -PassThru | Select-Object Name, StartType | ConvertTo-Json'
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))

# Example usage
mgr = ServiceManager('server', 'user', 'pass')
print(mgr.list_services('Running'))
print(mgr.get_service('Spooler'))

6. Event Log Analysis

#!/usr/bin/env python3
"""Query and analyze Windows event logs."""
import winrm
import json

def get_event_logs(host, username, password, log_name='System', 
                   entry_type=None, newest=50, port=5985):
    """Query Windows event logs."""
    session = winrm.Session(
        f'http://{host}:{port}/wsman',
        auth=(username, password),
        transport='basic'
    )
    
    filter_clause = ""
    if entry_type:
        filter_clause = f"-EntryType {entry_type}"
    
    ps_script = f'''
    Get-EventLog -LogName {log_name} {filter_clause} -Newest {newest} | 
    Select-Object TimeGenerated, EntryType, Source, EventID, Message |
    ConvertTo-Json -Depth 2
    '''
    
    result = session.run_ps(ps_script)
    if result.status_code == 0:
        return json.loads(result.std_out.decode('utf-8'))
    else:
        raise Exception(result.std_err.decode('utf-8'))

def get_security_events(host, username, password, hours=24, port=5985):
    """Get security events from the last N hours."""
    session = winrm.Session(
        f'http://{host}:{port}/wsman',
        auth=(username, password),
        transport='basic'
    )
    
    ps_script = f'''
    $startTime = (Get-Date).AddHours(-{hours})
    Get-WinEvent -FilterHashtable @{{
        LogName = 'Security'
        StartTime = $startTime
    }} -MaxEvents 100 -ErrorAction SilentlyContinue | 
    Select-Object TimeCreated, Id, LevelDisplayName, Message |
    ConvertTo-Json -Depth 2
    '''
    
    result = session.run_ps(ps_script)
    if result.status_code == 0 and result.std_out:
        return json.loads(result.std_out.decode('utf-8'))
    return []

# Example
events = get_event_logs('server', 'user', 'pass', 'Application', 'Error', 10)
print(json.dumps(events, indent=2))

7. Process Management

#!/usr/bin/env python3
"""Manage Windows processes remotely."""
import winrm
import json

class ProcessManager:
    def __init__(self, host, username, password, port=5985):
        self.session = winrm.Session(
            f'http://{host}:{port}/wsman',
            auth=(username, password),
            transport='basic'
        )
    
    def list_processes(self, sort_by='CPU', top=20):
        """List top processes by CPU or Memory usage."""
        sort_prop = 'CPU' if sort_by == 'CPU' else 'WorkingSet64'
        
        ps_script = f'''
        Get-Process | Sort-Object {sort_prop} -Descending | Select-Object -First {top} |
        Select-Object Id, ProcessName, 
            @{{Name='CPU_Seconds';Expression={{[math]::Round($_.CPU, 2)}}}},
            @{{Name='MemoryMB';Expression={{[math]::Round($_.WorkingSet64 / 1MB, 2)}}}},
            @{{Name='Threads';Expression={{$_.Threads.Count}}}},
            Path |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_process(self, name_or_id):
        """Get details for a specific process."""
        if str(name_or_id).isdigit():
            filter_clause = f"-Id {name_or_id}"
        else:
            filter_clause = f"-Name '{name_or_id}'"
        
        ps_script = f'''
        Get-Process {filter_clause} -ErrorAction SilentlyContinue | 
        Select-Object Id, ProcessName, CPU, 
            @{{Name='MemoryMB';Expression={{[math]::Round($_.WorkingSet64 / 1MB, 2)}}}},
            StartTime, Path, MainWindowTitle |
        ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def stop_process(self, name_or_id, force=False):
        """Stop a process by name or ID."""
        force_flag = "-Force" if force else ""
        if str(name_or_id).isdigit():
            ps_script = f'Stop-Process -Id {name_or_id} {force_flag} -PassThru | Select-Object Id, ProcessName, HasExited | ConvertTo-Json'
        else:
            ps_script = f'Stop-Process -Name "{name_or_id}" {force_flag} -PassThru | Select-Object Id, ProcessName, HasExited | ConvertTo-Json'
        
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def start_process(self, path, arguments=None):
        """Start a new process."""
        args = f"-ArgumentList '{arguments}'" if arguments else ""
        ps_script = f'''
        $proc = Start-Process -FilePath "{path}" {args} -PassThru
        @{{
            Id = $proc.Id
            ProcessName = $proc.ProcessName
            StartTime = $proc.StartTime
        }} | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))

# Example
pm = ProcessManager('server', 'user', 'pass')
print(pm.list_processes('Memory', 10))

8. File Operations

#!/usr/bin/env python3
"""Remote file operations on Windows servers."""
import winrm
import json
import base64

class FileManager:
    def __init__(self, host, username, password, port=5985):
        self.session = winrm.Session(
            f'http://{host}:{port}/wsman',
            auth=(username, password),
            transport='basic'
        )
    
    def list_directory(self, path):
        """List directory contents."""
        ps_script = f'''
        Get-ChildItem -Path "{path}" -ErrorAction Stop | 
        Select-Object Name, 
            @{{Name='Type';Expression={{if($_.PSIsContainer){{'Directory'}}else{{'File'}}}}}},
            @{{Name='SizeBytes';Expression={{$_.Length}}}},
            LastWriteTime,
            Attributes |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def read_file(self, path, encoding='utf-8'):
        """Read file content."""
        ps_script = f'Get-Content -Path "{path}" -Raw -Encoding {encoding}'
        result = self.session.run_ps(ps_script)
        return result.std_out.decode('utf-8')
    
    def write_file(self, path, content, encoding='utf-8'):
        """Write content to file."""
        # Base64 encode to handle special characters
        encoded = base64.b64encode(content.encode('utf-8')).decode('ascii')
        ps_script = f'''
        $bytes = [Convert]::FromBase64String("{encoded}")
        $content = [System.Text.Encoding]::UTF8.GetString($bytes)
        Set-Content -Path "{path}" -Value $content -Encoding {encoding}
        "Success"
        '''
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def copy_file(self, source, destination):
        """Copy file or directory."""
        ps_script = f'Copy-Item -Path "{source}" -Destination "{destination}" -Recurse -Force; "Success"'
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def move_file(self, source, destination):
        """Move file or directory."""
        ps_script = f'Move-Item -Path "{source}" -Destination "{destination}" -Force; "Success"'
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def delete_file(self, path, recurse=False):
        """Delete file or directory."""
        recurse_flag = "-Recurse" if recurse else ""
        ps_script = f'Remove-Item -Path "{path}" {recurse_flag} -Force; "Success"'
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def file_exists(self, path):
        """Check if file or directory exists."""
        ps_script = f'Test-Path -Path "{path}"'
        result = self.session.run_ps(ps_script)
        return result.std_out.decode('utf-8').strip().lower() == 'true'
    
    def get_file_info(self, path):
        """Get detailed file information."""
        ps_script = f'''
        $item = Get-Item -Path "{path}" -ErrorAction Stop
        @{{
            FullName = $item.FullName
            Name = $item.Name
            Extension = $item.Extension
            IsDirectory = $item.PSIsContainer
            SizeBytes = $item.Length
            CreationTime = $item.CreationTime.ToString("yyyy-MM-dd HH:mm:ss")
            LastWriteTime = $item.LastWriteTime.ToString("yyyy-MM-dd HH:mm:ss")
            LastAccessTime = $item.LastAccessTime.ToString("yyyy-MM-dd HH:mm:ss")
            Attributes = $item.Attributes.ToString()
            Owner = (Get-Acl $item.FullName).Owner
        }} | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))

# Example
fm = FileManager('server', 'user', 'pass')
print(fm.list_directory('C:\\'))
print(fm.get_file_info('C:\\Windows\\System32\\drivers\\etc\\hosts'))

9. Registry Operations

#!/usr/bin/env python3
"""Windows Registry operations via WinRM."""
import winrm
import json

class RegistryManager:
    def __init__(self, host, username, password, port=5985):
        self.session = winrm.Session(
            f'http://{host}:{port}/wsman',
            auth=(username, password),
            transport='basic'
        )
    
    def get_value(self, path, name):
        """Get a registry value."""
        ps_script = f'''
        $value = Get-ItemProperty -Path "{path}" -Name "{name}" -ErrorAction Stop
        @{{
            Path = "{path}"
            Name = "{name}"
            Value = $value."{name}"
            Type = (Get-Item -Path "{path}").GetValueKind("{name}").ToString()
        }} | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def set_value(self, path, name, value, value_type='String'):
        """Set a registry value."""
        ps_script = f'''
        if (-not (Test-Path "{path}")) {{
            New-Item -Path "{path}" -Force | Out-Null
        }}
        Set-ItemProperty -Path "{path}" -Name "{name}" -Value "{value}" -Type {value_type}
        "Success"
        '''
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def delete_value(self, path, name):
        """Delete a registry value."""
        ps_script = f'Remove-ItemProperty -Path "{path}" -Name "{name}" -Force; "Success"'
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def list_values(self, path):
        """List all values in a registry key."""
        ps_script = f'''
        Get-Item -Path "{path}" -ErrorAction Stop | 
        Select-Object -ExpandProperty Property | ForEach-Object {{
            $name = $_
            $item = Get-Item -Path "{path}"
            @{{
                Name = $name
                Value = $item.GetValue($name)
                Type = $item.GetValueKind($name).ToString()
            }}
        }} | ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def list_subkeys(self, path):
        """List subkeys of a registry key."""
        ps_script = f'''
        Get-ChildItem -Path "{path}" -ErrorAction Stop | 
        Select-Object Name, PSChildName |
        ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))

# Example
reg = RegistryManager('server', 'user', 'pass')
print(reg.list_values('HKLM:\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion'))

10. Active Directory Management

#!/usr/bin/env python3
"""Active Directory management via WinRM."""
import winrm
import json

class ADManager:
    def __init__(self, host, username, password, port=5985):
        self.session = winrm.Session(
            f'http://{host}:{port}/wsman',
            auth=(username, password),
            transport='basic'
        )
    
    def get_domain_info(self):
        """Get domain information."""
        ps_script = '''
        Import-Module ActiveDirectory
        $domain = Get-ADDomain
        @{
            Name = $domain.Name
            DNSRoot = $domain.DNSRoot
            Forest = $domain.Forest
            DomainMode = $domain.DomainMode.ToString()
            PDCEmulator = $domain.PDCEmulator
            InfrastructureMaster = $domain.InfrastructureMaster
        } | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def list_users(self, filter_str=None, properties=None):
        """List AD users."""
        props = properties or ['SamAccountName', 'DisplayName', 'EmailAddress', 'Enabled', 'LastLogonDate']
        props_str = ','.join(props)
        filter_clause = f"-Filter \"{filter_str}\"" if filter_str else ""
        
        ps_script = f'''
        Import-Module ActiveDirectory
        Get-ADUser {filter_clause} -Properties {props_str} | 
        Select-Object {props_str} |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_user(self, identity):
        """Get detailed user information."""
        ps_script = f'''
        Import-Module ActiveDirectory
        Get-ADUser -Identity "{identity}" -Properties * | 
        Select-Object SamAccountName, DisplayName, EmailAddress, 
            Department, Title, Manager, MemberOf, 
            Enabled, LockedOut, PasswordExpired, 
            LastLogonDate, Created, Modified |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def create_user(self, sam_account_name, name, password, ou, enabled=True):
        """Create a new AD user."""
        ps_script = f'''
        Import-Module ActiveDirectory
        $securePass = ConvertTo-SecureString "{password}" -AsPlainText -Force
        New-ADUser -SamAccountName "{sam_account_name}" -Name "{name}" `
            -AccountPassword $securePass -Enabled ${str(enabled).lower()} `
            -Path "{ou}" -PassThru | 
        Select-Object SamAccountName, Name, Enabled, DistinguishedName |
        ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def disable_user(self, identity):
        """Disable an AD user."""
        ps_script = f'''
        Import-Module ActiveDirectory
        Disable-ADAccount -Identity "{identity}"
        Get-ADUser -Identity "{identity}" | Select-Object SamAccountName, Enabled | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def unlock_user(self, identity):
        """Unlock a locked AD user."""
        ps_script = f'''
        Import-Module ActiveDirectory
        Unlock-ADAccount -Identity "{identity}"
        Get-ADUser -Identity "{identity}" -Properties LockedOut | 
        Select-Object SamAccountName, LockedOut | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def reset_password(self, identity, new_password):
        """Reset user password."""
        ps_script = f'''
        Import-Module ActiveDirectory
        $securePass = ConvertTo-SecureString "{new_password}" -AsPlainText -Force
        Set-ADAccountPassword -Identity "{identity}" -NewPassword $securePass -Reset
        "Password reset successful"
        '''
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def list_groups(self, filter_str=None):
        """List AD groups."""
        filter_clause = f"-Filter \"{filter_str}\"" if filter_str else ""
        ps_script = f'''
        Import-Module ActiveDirectory
        Get-ADGroup {filter_clause} -Properties Description, Members | 
        Select-Object Name, SamAccountName, GroupScope, GroupCategory, 
            Description, @{{Name='MemberCount';Expression={{$_.Members.Count}}}} |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_group_members(self, identity):
        """Get members of an AD group."""
        ps_script = f'''
        Import-Module ActiveDirectory
        Get-ADGroupMember -Identity "{identity}" | 
        Select-Object Name, SamAccountName, objectClass |
        ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def add_group_member(self, group, member):
        """Add a member to an AD group."""
        ps_script = f'''
        Import-Module ActiveDirectory
        Add-ADGroupMember -Identity "{group}" -Members "{member}"
        "Member added successfully"
        '''
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def remove_group_member(self, group, member):
        """Remove a member from an AD group."""
        ps_script = f'''
        Import-Module ActiveDirectory
        Remove-ADGroupMember -Identity "{group}" -Members "{member}" -Confirm:$false
        "Member removed successfully"
        '''
        result = self.session.run_ps(ps_script)
        return result.status_code == 0

# Example
ad = ADManager('domain-controller', 'admin', 'pass')
print(ad.list_users())
print(ad.get_user('jsmith'))

11. Scheduled Task Management

#!/usr/bin/env python3
"""Manage Windows Scheduled Tasks via WinRM."""
import winrm
import json

class ScheduledTaskManager:
    def __init__(self, host, username, password, port=5985):
        self.session = winrm.Session(
            f'http://{host}:{port}/wsman',
            auth=(username, password),
            transport='basic'
        )
    
    def list_tasks(self, path='\\'):
        """List scheduled tasks."""
        ps_script = f'''
        Get-ScheduledTask -TaskPath "{path}*" | 
        Select-Object TaskName, TaskPath, State, 
            @{{Name='LastRunTime';Expression={{$_.LastRunTime}}}},
            @{{Name='NextRunTime';Expression={{$_.NextRunTime}}}} |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_task(self, name, path='\\'):
        """Get detailed task information."""
        ps_script = f'''
        $task = Get-ScheduledTask -TaskName "{name}" -TaskPath "{path}" -ErrorAction Stop
        $info = Get-ScheduledTaskInfo -TaskName "{name}" -TaskPath "{path}"
        @{{
            TaskName = $task.TaskName
            TaskPath = $task.TaskPath
            State = $task.State.ToString()
            Description = $task.Description
            Author = $task.Author
            Actions = $task.Actions | ForEach-Object {{
                @{{
                    Type = $_.CimClass.CimClassName
                    Execute = $_.Execute
                    Arguments = $_.Arguments
                    WorkingDirectory = $_.WorkingDirectory
                }}
            }}
            Triggers = $task.Triggers | ForEach-Object {{
                @{{
                    Type = $_.CimClass.CimClassName
                    Enabled = $_.Enabled
                }}
            }}
            LastRunTime = $info.LastRunTime
            LastTaskResult = $info.LastTaskResult
            NextRunTime = $info.NextRunTime
            NumberOfMissedRuns = $info.NumberOfMissedRuns
        }} | ConvertTo-Json -Depth 3
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def run_task(self, name, path='\\'):
        """Run a scheduled task immediately."""
        ps_script = f'''
        Start-ScheduledTask -TaskName "{name}" -TaskPath "{path}"
        Start-Sleep -Seconds 2
        $info = Get-ScheduledTaskInfo -TaskName "{name}" -TaskPath "{path}"
        @{{
            TaskName = "{name}"
            State = (Get-ScheduledTask -TaskName "{name}" -TaskPath "{path}").State.ToString()
            LastRunTime = $info.LastRunTime
        }} | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def stop_task(self, name, path='\\'):
        """Stop a running scheduled task."""
        ps_script = f'Stop-ScheduledTask -TaskName "{name}" -TaskPath "{path}"; "Stopped"'
        result = self.session.run_ps(ps_script)
        return result.status_code == 0
    
    def enable_task(self, name, path='\\'):
        """Enable a scheduled task."""
        ps_script = f'Enable-ScheduledTask -TaskName "{name}" -TaskPath "{path}" | Select-Object TaskName, State | ConvertTo-Json'
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def disable_task(self, name, path='\\'):
        """Disable a scheduled task."""
        ps_script = f'Disable-ScheduledTask -TaskName "{name}" -TaskPath "{path}" | Select-Object TaskName, State | ConvertTo-Json'
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def create_task(self, name, execute, arguments=None, trigger_type='Daily', 
                    time='09:00', description='', run_as='SYSTEM'):
        """Create a new scheduled task."""
        args_param = f'-Argument "{arguments}"' if arguments else ''
        
        ps_script = f'''
        $action = New-ScheduledTaskAction -Execute "{execute}" {args_param}
        $trigger = New-ScheduledTaskTrigger -{trigger_type} -At "{time}"
        $principal = New-ScheduledTaskPrincipal -UserId "{run_as}" -LogonType ServiceAccount -RunLevel Highest
        $settings = New-ScheduledTaskSettingsSet -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries
        
        Register-ScheduledTask -TaskName "{name}" -Action $action -Trigger $trigger `
            -Principal $principal -Settings $settings -Description "{description}" |
        Select-Object TaskName, State | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def delete_task(self, name, path='\\'):
        """Delete a scheduled task."""
        ps_script = f'Unregister-ScheduledTask -TaskName "{name}" -TaskPath "{path}" -Confirm:$false; "Deleted"'
        result = self.session.run_ps(ps_script)
        return result.status_code == 0

# Example
stm = ScheduledTaskManager('server', 'user', 'pass')
print(stm.list_tasks())

12. Windows Update Management

#!/usr/bin/env python3
"""Manage Windows Updates via WinRM."""
import winrm
import json

class WindowsUpdateManager:
    def __init__(self, host, username, password, port=5985):
        self.session = winrm.Session(
            f'http://{host}:{port}/wsman',
            auth=(username, password),
            transport='basic'
        )
    
    def get_update_history(self, count=20):
        """Get Windows Update history."""
        ps_script = f'''
        $session = New-Object -ComObject Microsoft.Update.Session
        $searcher = $session.CreateUpdateSearcher()
        $history = $searcher.QueryHistory(0, {count})
        $history | ForEach-Object {{
            @{{
                Title = $_.Title
                Date = $_.Date
                Operation = switch($_.Operation) {{
                    1 {{"Installation"}}
                    2 {{"Uninstallation"}}
                    3 {{"Other"}}
                }}
                ResultCode = switch($_.ResultCode) {{
                    0 {{"Not Started"}}
                    1 {{"In Progress"}}
                    2 {{"Succeeded"}}
                    3 {{"Succeeded With Errors"}}
                    4 {{"Failed"}}
                    5 {{"Aborted"}}
                }}
            }}
        }} | ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def check_for_updates(self):
        """Check for available updates."""
        ps_script = '''
        $session = New-Object -ComObject Microsoft.Update.Session
        $searcher = $session.CreateUpdateSearcher()
        $results = $searcher.Search("IsInstalled=0")
        $results.Updates | ForEach-Object {
            @{
                Title = $_.Title
                Description = $_.Description
                KBArticleIDs = $_.KBArticleIDs -join ","
                IsMandatory = $_.IsMandatory
                IsDownloaded = $_.IsDownloaded
                MaxDownloadSize = $_.MaxDownloadSize
            }
        } | ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        if result.std_out:
            return json.loads(result.std_out.decode('utf-8'))
        return []
    
    def get_installed_updates(self):
        """Get list of installed updates."""
        ps_script = '''
        Get-HotFix | Select-Object HotFixID, Description, InstalledBy, InstalledOn |
        ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_pending_reboot(self):
        """Check if system has pending reboot."""
        ps_script = '''
        $pendingReboot = $false
        $reasons = @()
        
        # Check Windows Update
        if (Test-Path "HKLM:\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\WindowsUpdate\\Auto Update\\RebootRequired") {
            $pendingReboot = $true
            $reasons += "Windows Update"
        }
        
        # Check Component Based Servicing
        if (Test-Path "HKLM:\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\Component Based Servicing\\RebootPending") {
            $pendingReboot = $true
            $reasons += "Component Based Servicing"
        }
        
        # Check PendingFileRenameOperations
        $pfro = Get-ItemProperty "HKLM:\\SYSTEM\\CurrentControlSet\\Control\\Session Manager" -Name PendingFileRenameOperations -EA SilentlyContinue
        if ($pfro) {
            $pendingReboot = $true
            $reasons += "Pending File Rename"
        }
        
        @{
            PendingReboot = $pendingReboot
            Reasons = $reasons
        } | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))

# Example
wum = WindowsUpdateManager('server', 'user', 'pass')
print(wum.get_installed_updates())
print(wum.get_pending_reboot())

13. Network Configuration

#!/usr/bin/env python3
"""Network configuration and diagnostics via WinRM."""
import winrm
import json

class NetworkManager:
    def __init__(self, host, username, password, port=5985):
        self.session = winrm.Session(
            f'http://{host}:{port}/wsman',
            auth=(username, password),
            transport='basic'
        )
    
    def get_network_adapters(self):
        """Get network adapter information."""
        ps_script = '''
        Get-NetAdapter | Select-Object Name, InterfaceDescription, Status,
            MacAddress, LinkSpeed, 
            @{Name='IPv4';Expression={(Get-NetIPAddress -InterfaceIndex $_.ifIndex -AddressFamily IPv4 -ErrorAction SilentlyContinue).IPAddress}},
            @{Name='IPv6';Expression={(Get-NetIPAddress -InterfaceIndex $_.ifIndex -AddressFamily IPv6 -ErrorAction SilentlyContinue | Where-Object {$_.PrefixOrigin -ne 'WellKnown'}).IPAddress}} |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_ip_configuration(self):
        """Get detailed IP configuration."""
        ps_script = '''
        Get-NetIPConfiguration | ForEach-Object {
            @{
                InterfaceAlias = $_.InterfaceAlias
                InterfaceIndex = $_.InterfaceIndex
                IPv4Address = $_.IPv4Address.IPAddress
                IPv4DefaultGateway = $_.IPv4DefaultGateway.NextHop
                DNSServer = $_.DNSServer.ServerAddresses
            }
        } | ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_tcp_connections(self, state=None):
        """Get TCP connections."""
        filter_clause = f"| Where-Object {{$_.State -eq '{state}'}}" if state else ""
        ps_script = f'''
        Get-NetTCPConnection {filter_clause} | 
        Select-Object LocalAddress, LocalPort, RemoteAddress, RemotePort, 
            State, OwningProcess,
            @{{Name='ProcessName';Expression={{(Get-Process -Id $_.OwningProcess -ErrorAction SilentlyContinue).ProcessName}}}} |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_listening_ports(self):
        """Get all listening ports."""
        return self.get_tcp_connections('Listen')
    
    def test_connection(self, target, port=None):
        """Test network connectivity."""
        port_param = f"-Port {port}" if port else ""
        ps_script = f'''
        $result = Test-NetConnection -ComputerName "{target}" {port_param}
        @{{
            ComputerName = $result.ComputerName
            RemoteAddress = $result.RemoteAddress.IPAddressToString
            PingSucceeded = $result.PingSucceeded
            TcpTestSucceeded = $result.TcpTestSucceeded
            RoundtripTime = $result.PingReplyDetails.RoundtripTime
        }} | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def get_firewall_rules(self, enabled_only=True):
        """Get firewall rules."""
        filter_clause = "| Where-Object {$_.Enabled -eq $true}" if enabled_only else ""
        ps_script = f'''
        Get-NetFirewallRule {filter_clause} | 
        Select-Object -First 50 Name, DisplayName, Enabled, Direction, Action, Profile |
        ConvertTo-Json -Depth 2
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def create_firewall_rule(self, name, direction, action, protocol, port, description=''):
        """Create a firewall rule."""
        ps_script = f'''
        New-NetFirewallRule -Name "{name}" -DisplayName "{name}" `
            -Direction {direction} -Action {action} `
            -Protocol {protocol} -LocalPort {port} `
            -Description "{description}" |
        Select-Object Name, Enabled, Direction, Action | ConvertTo-Json
        '''
        result = self.session.run_ps(ps_script)
        return json.loads(result.std_out.decode('utf-8'))
    
    def remove_firewall_rule(self, name):
        """Remove a firewall rule."""
        ps_script = f'Remove-NetFirewallRule -Name "{name}"; "Removed"'
        result = self.session.run_ps(ps_script)
        return result.status_code == 0

# Example
nm = NetworkManager('server', 'user', 'pass')
print(nm.get_network_adapters())
print(nm.get_listening_ports())

Quick Reference Commands

One-Liner Examples

# Execute PowerShell command
uv run python -c "
import winrm
s = winrm.Session('http://IP:5985/wsman', auth=('Administrator', 'PASS'), transport='basic')
r = s.run_ps('Get-ComputerInfo | Select-Object CsName, WindowsVersion')
print(r.std_out.decode())
"

# Get system info
uv run python -c "
import winrm, json
s = winrm.Session('http://IP:5985/wsman', auth=('Administrator', 'PASS'), transport='basic')
r = s.run_ps('Get-ComputerInfo | ConvertTo-Json')
print(json.dumps(json.loads(r.std_out.decode()), indent=2))
"

# List running services
uv run python -c "
import winrm
s = winrm.Session('http://IP:5985/wsman', auth=('Administrator', 'PASS'), transport='basic')
r = s.run_ps('Get-Service | Where-Object {\$_.Status -eq \"Running\"} | Select-Object Name, DisplayName')
print(r.std_out.decode())
"

# Check disk space
uv run python -c "
import winrm
s = winrm.Session('http://IP:5985/wsman', auth=('Administrator', 'PASS'), transport='basic')
r = s.run_ps('Get-WmiObject Win32_LogicalDisk | Select-Object DeviceID, @{n=\"SizeGB\";e={\$_.Size/1GB}}, @{n=\"FreeGB\";e={\$_.FreeSpace/1GB}}')
print(r.std_out.decode())
"

Multi-Server Execution

#!/usr/bin/env python3
"""Execute commands across multiple Windows servers."""
import winrm
from concurrent.futures import ThreadPoolExecutor, as_completed

SERVERS = [
    {'host': '44.197.31.152', 'user': 'Administrator', 'pass': 'PASS1'},
    {'host': '52.3.242.251', 'user': 'Administrator', 'pass': 'PASS2'},
    {'host': '80.248.0.66', 'user': 'Administrator', 'pass': 'PASS3'},
]

def execute_on_server(server, command):
    """Execute command on a single server."""
    try:
        session = winrm.Session(
            f"http://{server['host']}:5985/wsman",
            auth=(server['user'], server['pass']),
            transport='basic'
        )
        result = session.run_ps(command)
        return {
            'host': server['host'],
            'success': True,
            'output': result.std_out.decode('utf-8'),
            'error': result.std_err.decode('utf-8') if result.std_err else None
        }
    except Exception as e:
        return {
            'host': server['host'],
            'success': False,
            'error': str(e)
        }

def execute_parallel(servers, command, max_workers=5):
    """Execute command on multiple servers in parallel."""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(execute_on_server, s, command): s for s in servers}
        for future in as_completed(futures):
            results.append(future.result())
    return results

# Example: Get uptime from all servers
if __name__ == '__main__':
    command = '(Get-Date) - (Get-CimInstance Win32_OperatingSystem).LastBootUpTime | Select-Object Days, Hours, Minutes'
    results = execute_parallel(SERVERS, command)
    for r in results:
        print(f"\n=== {r['host']} ===")
        if r['success']:
            print(r['output'])
        else:
            print(f"ERROR: {r['error']}")

Security Best Practices

Credential Management

# Use environment variables
import os
password = os.environ.get('WINRM_PASSWORD')

# Or use a secrets manager
from getpass import getpass
password = getpass("Enter password: ")

HTTPS Configuration

# Use HTTPS for production
session = winrm.Session(
    'https://server.domain.com:5986/wsman',
    auth=('Administrator', 'password'),
    transport='ntlm',
    server_cert_validation='validate',  # Use 'ignore' only for testing
    ca_trust_path='/path/to/ca-bundle.crt'
)

Kerberos Authentication

# For domain environments
session = winrm.Session(
    'server.domain.com',
    auth=('user@DOMAIN.COM', 'password'),
    transport='kerberos'
)

Troubleshooting

Common Issues

Connection Refused
- Verify WinRM is enabled: winrm quickconfig
- Check firewall: netsh advfirewall firewall show rule name=all | find "5985"
- Test locally: winrm id -r:http://localhost:5985
Authentication Failed
- Check credentials
- Verify Basic auth is enabled: winrm get winrm/config/service/auth
- For NTLM, use DOMAIN\\username format
Access Denied
- User must be in Administrators group or WinRMRemoteWMIUsers__
- Check UAC settings for remote connections

Timeout Issues

session = winrm.Session(
    'server',
    auth=('user', 'pass'),
    transport='basic',
    read_timeout_sec=60,
    operation_timeout_sec=30
)

Debug Mode

import logging
logging.basicConfig(level=logging.DEBUG)

# Now WinRM operations will show detailed logs

Version Information

Version: 1.0
Last Updated: 2026-01-28
Compatibility: OpenCode Agent Skills Framework
Stack: Python + PyWinRM
Platforms: Windows Server 2012+, Windows 10/11

This skill provides comprehensive Windows remote management capabilities via WinRM. Always ensure proper authorization before connecting to remote systems. Use HTTPS and strong authentication in production environments.

pywinrm

About