lib.winrm

  1#!/usr/bin/env python3
  2# -*- coding: utf-8; py-indent-offset: 4 -*-
  3#
  4# Author:  Linuxfabrik GmbH, Zurich, Switzerland
  5# Contact: info (at) linuxfabrik (dot) ch
  6#          https://www.linuxfabrik.ch/
  7# License: The Unlicense, see LICENSE file.
  8
  9# https://github.com/Linuxfabrik/monitoring-plugins/blob/main/CONTRIBUTING.rst
 10
 11"""This library collects some Microsoft WinRM related functions."""
 12
 13__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
 14__version__ = '2026030301'
 15
 16try:
 17    import winrm
 18
 19    HAVE_WINRM = True
 20except ImportError:
 21    HAVE_WINRM = False
 22
 23try:
 24    from pypsrp.client import Client
 25
 26    HAVE_JEA = True
 27except ImportError:
 28    HAVE_JEA = False
 29
 30from . import txt
 31
 32_AUTH_MAP = {
 33    'kerberos': 'kerberos',
 34    'negotiate': 'negotiate',
 35    'ntlm': 'ntlm',
 36    'credssp': 'credssp',
 37    'basic': 'basic',
 38    'plaintext': 'basic',
 39    'ssl': 'basic',
 40}
 41
 42
 43def _build_auth(args):
 44    """
 45    Build a `(username, password)` tuple from `args`.
 46
 47    For Kerberos/negotiate transports with missing credentials,
 48    returns `(None, None)` so the library falls back to the
 49    Kerberos credential cache (`kinit`). Otherwise prepends
 50    `WINRM_DOMAIN` to the username when set.
 51
 52    ### Parameters
 53    - **args**: Object with `WINRM_USERNAME`,
 54      `WINRM_PASSWORD`, `WINRM_TRANSPORT`, and optionally
 55      `WINRM_DOMAIN`.
 56
 57    ### Returns
 58    - **tuple**: `(username, password)` suitable for
 59      pypsrp or pywinrm.
 60    """
 61    username = getattr(args, 'WINRM_USERNAME', None)
 62    password = getattr(args, 'WINRM_PASSWORD', None)
 63    transport = (getattr(args, 'WINRM_TRANSPORT', None) or '').lower()
 64    if transport in ('kerberos', 'negotiate') and (not username or not password):
 65        return (None, None)
 66    if getattr(args, 'WINRM_DOMAIN', None):
 67        return (f'{username}@{args.WINRM_DOMAIN}', password)
 68    return (username, password)
 69
 70
 71def _map_transport(args):
 72    """
 73    Derive PSRP auth method, SSL flag, and port from `args`.
 74
 75    ### Parameters
 76    - **args**: Object with `WINRM_TRANSPORT`.
 77
 78    ### Returns
 79    - **tuple**: `(psrp_auth, use_ssl, port)`.
 80    """
 81    transport = (getattr(args, 'WINRM_TRANSPORT', None) or '').lower()
 82    psrp_auth = _AUTH_MAP.get(transport, 'negotiate')
 83    use_ssl = transport == 'ssl'
 84    port = 5986 if use_ssl else 5985
 85    return (psrp_auth, use_ssl, port)
 86
 87
 88def run_cmd(args, cmd, params=None):
 89    """
 90    Run a native command on a remote Windows host via WinRM/PSRP and return a
 91    normalized result dictionary.
 92
 93    Prefers **pypsrp (PSRP)** if available (for JEA/PowerShell Remoting
 94    compatibility); otherwise falls back to **pywinrm**. Authentication,
 95    transport and SSL/port selection are derived from the provided `args`.
 96
 97    ### Parameters
 98    - **args**: An object (e.g., `argparse.Namespace`) that provides at least:
 99        - `WINRM_HOSTNAME` (`str`): Target host or IP.
100        - `WINRM_USERNAME` (`str`, optional): Username. If `None` or empty when using
101          Kerberos transport, will use existing Kerberos credentials from credential cache
102          (e.g., obtained via `kinit`).
103        - `WINRM_PASSWORD` (`str`, optional): Password. If `None` or empty when using
104          Kerberos transport, will use existing Kerberos credentials from credential cache.
105        - `WINRM_TRANSPORT` (`str`, optional): Transport (e.g., `'negotiate'`, `'kerberos'`,
106          `'ntlm'`, `'credssp'`, `'basic'`, `'ssl'`). Defaults to `'negotiate'` if unset.
107        - `WINRM_DOMAIN` (`str`, optional): If set, username is sent as `user@domain`.
108      (Additional fields may be honored by the underlying libraries if present.)
109    - **cmd** (`str`): The executable/command to run remotely (native command, not a PowerShell
110       script block).
111    - **params** (`list[str]`, optional): Positional arguments passed to the command. Defaults
112       to `[]`.
113
114    ### Returns
115    - **dict**: A normalized result with:
116        - `retc` (`int`): Process return code (`0` on success).
117        - `stdout` (`str`): Captured standard output (text).
118        - `stderr` (`str`): Captured standard error (text).
119
120    ### Behavior
121    - If **pypsrp** is available, maps `WINRM_TRANSPORT` to an appropriate PSRP auth
122      and chooses SSL/port (5986 for SSL, 5985 otherwise), then executes the command
123      via `Client.execute_cmd()`.
124    - If pypsrp is unavailable but **pywinrm** is installed, executes via
125      `Session.run_cmd()`.
126    - For Kerberos authentication: if `WINRM_USERNAME` and `WINRM_PASSWORD` are not provided
127      (or are empty/None), the function will attempt to use existing Kerberos credentials
128      from the credential cache (obtained via `kinit`).
129    - On any exception, returns `{'retc': 1, 'stdout': '', 'stderr': <exception text>}`.
130    - If neither backend is present, returns an error indicating that no compatible
131      remoting library is available.
132
133    ### Example
134    >>> # With explicit credentials:
135    >>> run_cmd(args, 'ipconfig', ['/all'])
136    {'retc': 0, 'stdout': 'Windows IP Configuration\\r\\n...','stderr': ''}
137    >>> # With Kerberos using kinit credentials (username/password can be None):
138    >>> run_cmd(args, 'ipconfig', ['/all'])
139    {'retc': 0, 'stdout': 'Windows IP Configuration\\r\\n...','stderr': ''}
140    """
141    auth = _build_auth(args)
142    if params is None:
143        params = []
144
145    if HAVE_JEA:
146        try:
147            _psrp_auth, _use_ssl, _port = _map_transport(args)
148            session = Client(
149                server=args.WINRM_HOSTNAME,
150                username=auth[0],
151                password=auth[1],
152                auth=_psrp_auth,
153                ssl=_use_ssl,
154                port=_port,
155                cert_validation=True,
156            )
157
158            stdout, stderr, rc = session.execute_cmd(
159                cmd,
160                args=params,
161            )
162            return {
163                'retc': rc,
164                'stdout': txt.to_text(stdout),
165                'stderr': txt.to_text(stderr),
166            }
167        except Exception as e:
168            return {
169                'retc': 1,
170                'stdout': '',
171                'stderr': txt.exception2text(e),
172            }
173
174    if HAVE_WINRM:
175        try:
176            session = winrm.Session(
177                args.WINRM_HOSTNAME,
178                auth=auth,
179                transport=args.WINRM_TRANSPORT,
180            )
181
182            result = session.run_cmd(cmd, params)
183            return {
184                'retc': result.status_code,
185                'stdout': txt.to_text(result.std_out),
186                'stderr': txt.to_text(result.std_err),
187            }
188        except Exception as e:
189            return {
190                'retc': 1,
191                'stdout': '',
192                'stderr': txt.exception2text(e),
193            }
194
195    return {
196        'retc': 1,
197        'stdout': '',
198        'stderr': 'No compatible remoting library available (pypsrp or pywinrm).',
199    }
200
201
202def _quote_ps_value(value):
203    """Escape a value for use in a PowerShell command string.
204
205    Single-quotes the value, doubling any embedded single
206    quotes (`'` becomes `''`).
207
208    ### Parameters
209    - **value**: The value to quote (converted to `str`).
210
211    ### Returns
212    - **str**: A safely quoted PowerShell string literal.
213    """
214    return "'{}'".format(str(value).replace("'", "''"))
215
216
217def run_ps(args, cmd, params=None):
218    """
219    Run PowerShell on a remote Windows host via WinRM/PSRP
220    and return a normalized result dictionary.
221
222    Prefers **pypsrp (PSRP)** if available (best for
223    JEA/PowerShell Remoting); otherwise falls back to
224    **pywinrm**.
225
226    ### Parameters
227    - **args**: Object (e.g. `argparse.Namespace`) with:
228        - `WINRM_HOSTNAME` (`str`): Target host or IP.
229        - `WINRM_USERNAME` (`str`, optional): Username.
230          If empty with Kerberos transport, the credential
231          cache (`kinit`) is used.
232        - `WINRM_PASSWORD` (`str`, optional): Password.
233          Same Kerberos fallback as username.
234        - `WINRM_TRANSPORT` (`str`, optional): Transport
235          (e.g. `'negotiate'`, `'kerberos'`, `'ntlm'`,
236          `'credssp'`, `'basic'`, `'ssl'`).
237          Defaults to `'negotiate'`.
238        - `WINRM_DOMAIN` (`str`, optional): If set,
239          username is sent as `user@domain`.
240        - `WINRM_CONFIGURATION_NAME` (`str`, optional):
241          JEA endpoint name. Defaults to
242          `'Microsoft.PowerShell'`. Only with **pypsrp**.
243    - **cmd** (`str`): What to execute remotely. Meaning
244      depends on `params`:
245        - `params is None` — `cmd` is an arbitrary
246          PowerShell script (pipelines, expressions, etc.)
247          executed via `add_script()`.
248        - `params` given (`list` or `dict`) — `cmd` is a
249          single cmdlet name executed via `add_cmdlet()`
250          (optimal for JEA allow/deny).
251    - **params** (`list[str]`, `dict`, or `None`):
252        - `None` (default) — no params; `cmd` is run as a
253          script.
254        - `list[str]` — positional arguments added via
255          `add_argument()`.
256        - `dict` — named parameters added via
257          `add_parameter(name, value)`.
258
259    ### Returns
260    - **dict**: Normalized result with:
261        - `retc` (`int`): `0` if no errors.
262        - `stdout` (`str`): Captured output.
263        - `stderr` (`str`): Error/diagnostic output.
264          For **pywinrm**: CLIXML progress noise is
265          suppressed when `retc == 0`.
266
267    ### Example
268    Pipeline (params=None, uses add_script):
269    >>> run_ps(args, 'Get-Process | Select -First 1')
270
271    Positional params (uses add_cmdlet + add_argument):
272    >>> run_ps(args, 'Get-Service', ['WinRM'])
273
274    Named params (uses add_cmdlet + add_parameter):
275    >>> run_ps(
276    ...     args,
277    ...     'Get-WmiObject',
278    ...     {'Class': 'Win32_OperatingSystem'},
279    ... )
280    """
281    auth = _build_auth(args)
282
283    configuration_name = getattr(
284        args,
285        'WINRM_CONFIGURATION_NAME',
286        None,
287    )
288    if configuration_name and not HAVE_JEA:
289        return {
290            'retc': 1,
291            'stdout': '',
292            'stderr': 'WINRM_CONFIGURATION_NAME requires '
293            'pypsrp (JEA). Install pypsrp or '
294            'unset '
295            '--winrm-configuration-name.',
296        }
297
298    if HAVE_JEA:
299        try:
300            from pypsrp.powershell import (
301                PowerShell,
302                RunspacePool,
303            )
304            from pypsrp.wsman import WSMan
305
306            _psrp_auth, _use_ssl, _port = _map_transport(args)
307
308            wsman = WSMan(
309                server=args.WINRM_HOSTNAME,
310                username=auth[0],
311                password=auth[1],
312                auth=_psrp_auth,
313                ssl=_use_ssl,
314                port=_port,
315                cert_validation=True,
316            )
317
318            with RunspacePool(
319                wsman,
320                configuration_name=(configuration_name or 'Microsoft.PowerShell'),
321            ) as pool:
322                ps = PowerShell(pool)
323                if params is not None:
324                    ps.add_cmdlet(cmd)
325                    if isinstance(params, dict):
326                        for name, value in params.items():
327                            ps.add_parameter(name, value)
328                    else:
329                        for param in params:
330                            ps.add_argument(param)
331                else:
332                    ps.add_script(cmd)
333                output = ps.invoke()
334
335            stdout = '\n'.join(str(o) for o in output)
336
337            stderr_lines = []
338            for err in ps.streams.error:
339                try:
340                    stderr_lines.append(
341                        err.to_string(),
342                    )
343                except Exception:
344                    msg = getattr(err, 'message', None) or str(err)
345                    stderr_lines.append(str(msg))
346            stderr = '\n'.join(stderr_lines)
347
348            return {
349                'retc': 0 if not ps.had_errors else 1,
350                'stdout': txt.to_text(stdout),
351                'stderr': txt.to_text(stderr),
352            }
353        except Exception as e:
354            return {
355                'retc': 1,
356                'stdout': '',
357                'stderr': txt.exception2text(e),
358            }
359
360    if HAVE_WINRM:
361        try:
362            session = winrm.Session(
363                args.WINRM_HOSTNAME,
364                auth=auth,
365                transport=args.WINRM_TRANSPORT,
366            )
367
368            if params is not None:
369                if isinstance(params, dict):
370                    param_str = ' '.join(
371                        f'-{k} {_quote_ps_value(v)}' for k, v in params.items()
372                    )
373                    ps_cmd = f'{cmd} {param_str}'
374                else:
375                    ps_cmd = (
376                        '{} {}'.format(
377                            cmd,
378                            ' '.join(str(p) for p in params),
379                        )
380                        if params
381                        else cmd
382                    )
383            else:
384                ps_cmd = cmd
385
386            result = session.run_ps(ps_cmd)
387
388            result = {
389                'retc': result.status_code,
390                'stdout': txt.to_text(result.std_out),
391                'stderr': txt.to_text(result.std_err),
392            }
393            if result['retc'] == 0 and result['stderr'].startswith(
394                '#< CLIXML',
395            ):
396                result['stderr'] = ''
397            return result
398        except Exception as e:
399            return {
400                'retc': 1,
401                'stdout': '',
402                'stderr': txt.exception2text(e),
403            }
404
405    return {
406        'retc': 1,
407        'stdout': '',
408        'stderr': 'No compatible remoting library available (pypsrp or pywinrm).',
409    }