codekingpro/portable-devtools

1# Copyright (C) 2005 John Arbash-Meinel <john@arbash-meinel.com>2# Modified up by: Todd Whiteman <ToddW@ActiveState.com>3#4# This file is part of paramiko.5#6# Paramiko is free software; you can redistribute it and/or modify it under the7# terms of the GNU Lesser General Public License as published by the Free8# Software Foundation; either version 2.1 of the License, or (at your option)9# any later version.10#11# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY12# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR13# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more14# details.15#16# You should have received a copy of the GNU Lesser General Public License17# along with Paramiko; if not, write to the Free Software Foundation, Inc.,18# 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.19 20"""21Functions for communicating with Pageant, the basic windows ssh agent program.22"""23 24import array25import ctypes.wintypes26import platform27import struct28from paramiko.common import zero_byte29from paramiko.util import b30 31import _thread as thread32 33from . import _winapi34 35 36_AGENT_COPYDATA_ID = 0x804E50BA37_AGENT_MAX_MSGLEN = 819238# Note: The WM_COPYDATA value is pulled from win32con, as a workaround39# so we do not have to import this huge library just for this one variable.40win32con_WM_COPYDATA = 7441 42 43def _get_pageant_window_object():44    return ctypes.windll.user32.FindWindowA(b"Pageant", b"Pageant")45 46 47def can_talk_to_agent():48    """49    Check to see if there is a "Pageant" agent we can talk to.50 51    This checks both if we have the required libraries (win32all or ctypes)52    and if there is a Pageant currently running.53    """54    return bool(_get_pageant_window_object())55 56 57if platform.architecture()[0] == "64bit":58    ULONG_PTR = ctypes.c_uint6459else:60    ULONG_PTR = ctypes.c_uint3261 62 63class COPYDATASTRUCT(ctypes.Structure):64    """65    ctypes implementation of66    http://msdn.microsoft.com/en-us/library/windows/desktop/ms649010%28v=vs.85%29.aspx67    """68 69    _fields_ = [70        ("num_data", ULONG_PTR),71        ("data_size", ctypes.wintypes.DWORD),72        ("data_loc", ctypes.c_void_p),73    ]74 75 76def _query_pageant(msg):77    """78    Communication with the Pageant process is done through a shared79    memory-mapped file.80    """81    hwnd = _get_pageant_window_object()82    if not hwnd:83        # Raise a failure to connect exception, pageant isn't running anymore!84        return None85 86    # create a name for the mmap87    map_name = f"PageantRequest{thread.get_ident():08x}"88 89    pymap = _winapi.MemoryMap(90        map_name, _AGENT_MAX_MSGLEN, _winapi.get_security_attributes_for_user()91    )92    with pymap:93        pymap.write(msg)94        # Create an array buffer containing the mapped filename95        char_buffer = array.array("b", b(map_name) + zero_byte)  # noqa96        char_buffer_address, char_buffer_size = char_buffer.buffer_info()97        # Create a string to use for the SendMessage function call98        cds = COPYDATASTRUCT(99            _AGENT_COPYDATA_ID, char_buffer_size, char_buffer_address100        )101 102        response = ctypes.windll.user32.SendMessageA(103            hwnd, win32con_WM_COPYDATA, ctypes.sizeof(cds), ctypes.byref(cds)104        )105 106        if response > 0:107            pymap.seek(0)108            datalen = pymap.read(4)109            retlen = struct.unpack(">I", datalen)[0]110            return datalen + pymap.read(retlen)111        return None112 113 114class PageantConnection:115    """116    Mock "connection" to an agent which roughly approximates the behavior of117    a unix local-domain socket (as used by Agent).  Requests are sent to the118    pageant daemon via special Windows magick, and responses are buffered back119    for subsequent reads.120    """121 122    def __init__(self):123        self._response = None124 125    def send(self, data):126        self._response = _query_pageant(data)127 128    def recv(self, n):129        if self._response is None:130            return ""131        ret = self._response[:n]132        self._response = self._response[n:]133        if self._response == "":134            self._response = None135        return ret136 137    def close(self):138        pass139