tessl/pypi-pywinrm

Python library for Windows Remote Management (WinRM) service enabling remote command execution and PowerShell scripts on Windows machines.

—

Pending

Overview

Eval results

Files

Protocol Interface

Name: tessl/pypi-pywinrm
Author: tessl

Low-level SOAP protocol implementation providing full control over WinRM operations. The Protocol class offers fine-grained management of shell lifecycle, command streaming, input/output handling, and advanced WS-Management features for complex automation scenarios.

Capabilities

Protocol Initialization

Creates a Protocol instance with comprehensive configuration options for authentication, transport, timeouts, and security settings.

class Protocol:
    """Low-level SOAP protocol implementation for WinRM operations."""
    
    # Default configuration constants
    DEFAULT_READ_TIMEOUT_SEC = 30
    DEFAULT_OPERATION_TIMEOUT_SEC = 20
    DEFAULT_MAX_ENV_SIZE = 153600
    DEFAULT_LOCALE = "en-US"
    
    def __init__(
        self,
        endpoint: str,
        transport: Literal["auto", "basic", "certificate", "ntlm", "kerberos", "credssp", "plaintext", "ssl"] = "plaintext",
        username: str | None = None,
        password: str | None = None,
        realm: None = None,
        service: str = "HTTP",
        keytab: None = None,
        ca_trust_path: Literal["legacy_requests"] | str = "legacy_requests",
        cert_pem: str | None = None,
        cert_key_pem: str | None = None,
        server_cert_validation: Literal["validate", "ignore"] | None = "validate",
        kerberos_delegation: bool = False,
        read_timeout_sec: str | int = DEFAULT_READ_TIMEOUT_SEC,
        operation_timeout_sec: str | int = DEFAULT_OPERATION_TIMEOUT_SEC,
        kerberos_hostname_override: str | None = None,
        message_encryption: Literal["auto", "always", "never"] = "auto",
        credssp_disable_tlsv1_2: bool = False,
        send_cbt: bool = True,
        proxy: Literal["legacy_requests"] | str | None = "legacy_requests"
    ):
        """
        Initialize Protocol with comprehensive WinRM configuration.
        
        Parameters:
        - endpoint: WinRM webservice endpoint URL
        - transport: authentication/transport method
        - username: authentication username
        - password: authentication password
        - service: service name for Kerberos (default: "HTTP")
        - ca_trust_path: CA trust path or "legacy_requests" for environment variables
        - cert_pem: client certificate file path (PEM format)
        - cert_key_pem: client certificate key file path (PEM format)
        - server_cert_validation: certificate validation ("validate" or "ignore")
        - kerberos_delegation: enable Kerberos ticket forwarding
        - read_timeout_sec: HTTP connection/read timeout (must exceed operation_timeout_sec)
        - operation_timeout_sec: WinRM operation timeout
        - kerberos_hostname_override: override hostname for Kerberos SPN
        - message_encryption: message-level encryption setting
        - credssp_disable_tlsv1_2: disable TLS 1.2 for CredSSP
        - send_cbt: send channel binding tokens
        - proxy: proxy configuration
        """

Shell Management

Create and manage remote shells with customizable environment, working directory, and shell options.

def open_shell(
    self,
    i_stream: str = "stdin",
    o_stream: str = "stdout stderr",
    working_directory: str | None = None,
    env_vars: dict[str, str] | None = None,
    noprofile: bool = False,
    codepage: int = 437,
    lifetime: None = None,
    idle_timeout: str | int | None = None
) -> str:
    """
    Open a remote shell for command execution.
    
    Parameters:
    - i_stream: input stream specification (default: "stdin")
    - o_stream: output stream specification (default: "stdout stderr")
    - working_directory: initial working directory
    - env_vars: environment variables dictionary
    - noprofile: skip loading user profile
    - codepage: shell code page (default: 437)
    - lifetime: maximum shell lifetime (default: None for server default)
    - idle_timeout: idle timeout before shell closure (default: None for server default)
    
    Returns:
    Shell ID string for subsequent operations
    """

def close_shell(self, shell_id: str, close_session: bool = True) -> None:
    """
    Close remote shell and optionally the transport session.
    
    Parameters:
    - shell_id: shell identifier from open_shell()
    - close_session: whether to close the underlying transport session
    """

Usage Example:

from winrm import Protocol

protocol = Protocol(
    'http://windows-host:5985/wsman',
    username='administrator',
    password='password',
    transport='ntlm'
)

# Open shell with custom environment
env_vars = {
    'PATH': 'C:\\Windows\\System32;C:\\Windows',
    'TEMP': 'C:\\Temp'
}

shell_id = protocol.open_shell(
    working_directory='C:\\Scripts',
    env_vars=env_vars,
    lifetime=1800
)

# Use shell for multiple commands...

protocol.close_shell(shell_id)

Command Execution

Execute commands within shells with fine-grained control over execution mode and command arguments.

def run_command(
    self,
    shell_id: str,
    command: str,
    arguments: collections.abc.Iterable[str | bytes] = (),
    console_mode_stdin: bool = True,
    skip_cmd_shell: bool = False
) -> str:
    """
    Execute command in specified shell.
    
    Parameters:
    - shell_id: target shell identifier
    - command: command to execute
    - arguments: command arguments
    - console_mode_stdin: enable console mode for stdin
    - skip_cmd_shell: bypass cmd.exe shell wrapper
    
    Returns:
    Command ID for output retrieval and cleanup
    """

def cleanup_command(self, shell_id: str, command_id: str) -> None:
    """
    Clean up command resources after execution.
    
    Parameters:
    - shell_id: shell containing the command
    - command_id: command identifier from run_command()
    """

Input/Output Operations

Handle command input and retrieve output with support for streaming and partial results.

def send_command_input(
    self,
    shell_id: str,
    command_id: str,
    stdin_input: str | bytes,
    end: bool = False
) -> None:
    """
    Send input to running command.
    
    Parameters:
    - shell_id: shell identifier
    - command_id: command identifier
    - stdin_input: input data to send
    - end: whether this is the final input (triggers EOF)
    """

def get_command_output(
    self,
    shell_id: str,
    command_id: str
) -> tuple[bytes, bytes, int]:
    """
    Retrieve command output, blocking until completion.
    
    Parameters:
    - shell_id: shell identifier
    - command_id: command identifier
    
    Returns:
    Tuple of (stdout, stderr, status_code)
    """

def get_command_output_raw(
    self,
    shell_id: str,
    command_id: str
) -> tuple[bytes, bytes, int, bool]:
    """
    Retrieve command output with completion status.
    
    Returns:
    Tuple of (stdout, stderr, status_code, done)
    where done indicates if command execution is complete
    """

# Legacy alias for backward compatibility
_raw_get_command_output = get_command_output_raw

Advanced Command Execution Example:

# Execute long-running command with streaming
shell_id = protocol.open_shell()
command_id = protocol.run_command(shell_id, 'ping', ['google.com', '-t'])

# Stream output until completion
while True:
    stdout, stderr, status_code, done = protocol.get_command_output_raw(shell_id, command_id)
    
    if stdout:
        print(f"Output: {stdout.decode('utf-8')}")
    if stderr:
        print(f"Error: {stderr.decode('utf-8')}")
    
    if done:
        print(f"Command completed with status: {status_code}")
        break
    
    time.sleep(1)

protocol.cleanup_command(shell_id, command_id)
protocol.close_shell(shell_id)

Interactive Command Example

# Interactive command with input
shell_id = protocol.open_shell()
command_id = protocol.run_command(shell_id, 'python', ['-i'])

# Send Python commands
protocol.send_command_input(shell_id, command_id, "print('Hello, World!')\n")
protocol.send_command_input(shell_id, command_id, "import os\n")
protocol.send_command_input(shell_id, command_id, "print(os.getcwd())\n")
protocol.send_command_input(shell_id, command_id, "exit()\n", end=True)

# Get final output
stdout, stderr, status_code = protocol.get_command_output(shell_id, command_id)
print(f"Python session output:\n{stdout.decode('utf-8')}")

protocol.cleanup_command(shell_id, command_id)
protocol.close_shell(shell_id)

Message Operations

Low-level SOAP message construction and transmission for advanced WS-Management operations.

def build_wsman_header(
    self,
    action: str,
    resource_uri: str,
    shell_id: str | None = None,
    message_id: str | uuid.UUID | None = None
) -> dict[str, Any]:
    """
    Build WS-Management SOAP header for custom operations.
    
    Parameters:
    - action: WS-Management action URI
    - resource_uri: resource URI for the operation
    - shell_id: shell identifier for shell-specific operations
    - message_id: custom message ID (auto-generated if None)
    
    Returns:
    SOAP header dictionary structure
    """

# Legacy alias for Ansible compatibility
_get_soap_header = build_wsman_header

def send_message(self, message: str) -> bytes:
    """
    Send raw SOAP message and return response.
    
    Parameters:
    - message: complete SOAP message XML
    
    Returns:
    Raw response bytes from WinRM service
    """

Timeout and Error Handling

The Protocol class provides sophisticated timeout handling:

read_timeout_sec: HTTP connection and read timeout (must be > operation_timeout_sec)
operation_timeout_sec: WinRM operation timeout for individual requests
Automatic retry logic for operation timeouts during output retrieval
Proper cleanup of resources on timeout or error conditions

Timeouts are validated during initialization to ensure proper operation order and prevent deadlock conditions.

Install with Tessl CLI