1#
2# Manually written part of python bindings for libvirt
3#
4
5# On cygwin, the DLL is called cygvirtmod.dll
6try:
7    import libvirtmod  # type: ignore
8except ImportError as lib_e:
9    try:
10        import cygvirtmod as libvirtmod  # type: ignore
11    except ImportError as cyg_e:
12        if "No module named" in str(cyg_e):
13            raise lib_e
14
15from types import TracebackType
16from typing import Any, Callable, Dict, List, Optional, overload, Tuple, Type, TypeVar, Union
17_T = TypeVar('_T')
18_EventCB = Callable[[int, int, int, _T], None]
19_EventAddHandleFunc = Callable[[int, int, _EventCB, _T], int]
20_EventUpdateHandleFunc = Callable[[int, int], None]
21_EventRemoveHandleFunc = Callable[[int], int]
22_TimerCB = Callable[[int, _T], None]
23_EventAddTimeoutFunc = Callable[[int, _TimerCB, _T], int]
24_EventUpdateTimeoutFunc = Callable[[int, int], None]
25_EventRemoveTimeoutFunc = Callable[[int], int]
26_DomainCB = Callable[['virConnect', 'virDomain', int, int, _T], Optional[int]]
27_BlkioParameter = Dict[str, Any]
28_MemoryParameter = Dict[str, Any]
29_SchedParameter = Dict[str, Any]
30_TypedParameter = Dict[str, Any]
31
32
33# The root of all libvirt errors.
34class libvirtError(Exception):
35    def __init__(self, defmsg: str) -> None:
36
37        # Never call virConnGetLastError().
38        # virGetLastError() is now thread local
39        err = libvirtmod.virGetLastError()  # type: Optional[Tuple[int, int, str, int, str, Optional[str], Optional[str], int, int]]
40        if err is None:
41            msg = defmsg
42        else:
43            msg = err[2]
44
45        Exception.__init__(self, msg)
46
47        self.err = err
48
49    def get_error_code(self) -> Optional[int]:
50        if self.err is None:
51            return None
52        return self.err[0]
53
54    def get_error_domain(self) -> Optional[int]:
55        if self.err is None:
56            return None
57        return self.err[1]
58
59    def get_error_message(self) -> Optional[str]:
60        if self.err is None:
61            return None
62        return self.err[2]
63
64    def get_error_level(self) -> Optional[int]:
65        if self.err is None:
66            return None
67        return self.err[3]
68
69    def get_str1(self) -> Optional[str]:
70        if self.err is None:
71            return None
72        return self.err[4]
73
74    def get_str2(self) -> Optional[str]:
75        if self.err is None:
76            return None
77        return self.err[5]
78
79    def get_str3(self) -> Optional[str]:
80        if self.err is None:
81            return None
82        return self.err[6]
83
84    def get_int1(self) -> Optional[int]:
85        if self.err is None:
86            return None
87        return self.err[7]
88
89    def get_int2(self) -> Optional[int]:
90        if self.err is None:
91            return None
92        return self.err[8]
93
94
95#
96# register the libvirt global error handler
97#
98def registerErrorHandler(f: Callable[[_T, List], None], ctx: _T) -> int:
99    """Register a Python function for error reporting.
100       The function is called back as f(ctx, error), with error
101       being a list of information about the error being raised.
102       Returns 1 in case of success."""
103    return libvirtmod.virRegisterErrorHandler(f, ctx)
104
105
106def openAuth(uri: str, auth: List, flags: int = 0) -> 'virConnect':
107    # TODO: The C code rquires a List and there is not *Mutable*Tuple for a better description such as
108    # auth: Tuple[List[int], Callable[[List[MutableTuple[int, str, str, str, Any]], _T], int], _T]
109    """
110    This function should be called first to get a connection to the
111    Hypervisor. If necessary, authentication will be performed fetching
112    credentials via the callback.
113
114    See :py:func:`open` for notes about environment variables which can
115    have an effect on opening drivers and freeing the connection resources.
116
117    :param str uri: (Optional) connection URI, see https://libvirt.org/uri.html
118    :param auth: a list that contains 3 items:
119        - a list of supported credential types
120        - a callable that takes 2 arguments (credentials, user-data) and returns 0 on succcess and -1 on errors.
121            The credentials argument is a list of credentials that libvirt (actually
122            the ESX driver) would like to request. An element of this list is itself a
123            list containing 5 items (4 inputs, 1 output):
124                - the credential type, e.g. :py:const:`libvirt.VIR_CRED_AUTHNAME`
125                - a prompt to be displayed to the user
126                - a challenge, the ESX driver sets this to the hostname to allow automatic
127                    distinction between requests for ESX and vCenter credentials
128                - a default result for the request
129                - a place to store the actual result for the request
130        - user data that will be passed to the callable as second argument
131    :param int flags: bitwise-OR of virConnectFlags
132    :returns: a :py:class:`virConnect` instance on success.
133    :raises libvirtError: on errors.
134    """
135    ret = libvirtmod.virConnectOpenAuth(uri, auth, flags)
136    if ret is None:
137        raise libvirtError('virConnectOpenAuth() failed')
138    return virConnect(_obj=ret)
139
140
141#
142# Return library version.
143#
144def getVersion(name: Optional[str] = None) -> int:
145    """If no name parameter is passed (or name is None) then the
146    version of the libvirt library is returned as an integer.
147
148    If a name is passed and it refers to a driver linked to the
149    libvirt library, then this returns a tuple of (library version,
150    driver version).
151
152    If the name passed refers to a non-existent driver, then you
153    will get the exception 'no support for hypervisor'.
154
155    Versions numbers are integers: 1000000*major + 1000*minor + release."""
156    if name is None:
157        ret = libvirtmod.virGetVersion()
158    else:
159        ret = libvirtmod.virGetVersion(name)
160    if ret is None:
161        raise libvirtError("virGetVersion() failed")
162    return ret
163
164
165#
166# Invoke an EventHandle callback
167#
168@overload
169def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: Tuple[_EventCB, _T], opaquecompat: None = None) -> None: ...  # noqa E704
170@overload  # noqa F811
171def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: _EventCB, opaquecompat: _T = None) -> None: ...  # noqa E704
172def _eventInvokeHandleCallback(watch: int, fd: int, event: int, opaque: Union[Tuple[_EventCB, _T], _EventCB], opaquecompat: Optional[_T] = None) -> None:  # noqa F811
173    """
174    Invoke the Event Impl Handle Callback in C
175    """
176    # libvirt 0.9.2 and earlier required custom event loops to know
177    # that opaque=(cb, original_opaque) and pass the values individually
178    # to this wrapper. This should handle the back compat case, and make
179    # future invocations match the virEventHandleCallback prototype
180    if opaquecompat:
181        callback = opaque
182        opaque_ = opaquecompat
183    else:
184        assert isinstance(opaque, tuple)
185        callback = opaque[0]
186        opaque_ = opaque[1]
187
188    libvirtmod.virEventInvokeHandleCallback(watch, fd, event, callback, opaque_)
189
190
191#
192# Invoke an EventTimeout callback
193#
194def _eventInvokeTimeoutCallback(timer: int, opaque: Union[Tuple[_TimerCB, _T], _TimerCB], opaquecompat: Optional[_T] = None) -> None:
195    """
196    Invoke the Event Impl Timeout Callback in C
197    """
198    # libvirt 0.9.2 and earlier required custom event loops to know
199    # that opaque=(cb, original_opaque) and pass the values individually
200    # to this wrapper. This should handle the back compat case, and make
201    # future invocations match the virEventTimeoutCallback prototype
202    if opaquecompat:
203        callback = opaque
204        opaque_ = opaquecompat
205    else:
206        assert isinstance(opaque, tuple)
207        callback = opaque[0]
208        opaque_ = opaque[1]
209
210    libvirtmod.virEventInvokeTimeoutCallback(timer, callback, opaque_)
211
212
213def _dispatchEventHandleCallback(watch: int, fd: int, events: int, cbData: Dict[str, Any]) -> int:
214    cb = cbData["cb"]
215    opaque = cbData["opaque"]
216
217    cb(watch, fd, events, opaque)
218    return 0
219
220
221def _dispatchEventTimeoutCallback(timer: int, cbData: Dict[str, Any]) -> int:
222    cb = cbData["cb"]
223    opaque = cbData["opaque"]
224
225    cb(timer, opaque)
226    return 0
227
228
229def virEventAddHandle(fd: int, events: int, cb: _EventCB, opaque: _T) -> int:
230    """
231    register a callback for monitoring file handle events
232
233    @fd: file handle to monitor for events
234    @events: bitset of events to watch from virEventHandleType constants
235    @cb: callback to invoke when an event occurs
236    @opaque: user data to pass to callback
237
238    Example callback prototype is:
239        def cb(watch,   # int id of the handle
240               fd,      # int file descriptor the event occurred on
241               events,  # int bitmap of events that have occurred
242               opaque): # opaque data passed to eventAddHandle
243    """
244    cbData = {"cb": cb, "opaque": opaque}
245    ret = libvirtmod.virEventAddHandle(fd, events, cbData)
246    if ret == -1:
247        raise libvirtError('virEventAddHandle() failed')
248    return ret
249
250
251def virEventAddTimeout(timeout: int, cb: _TimerCB, opaque: _T) -> int:
252    """
253    register a callback for a timer event
254
255    @timeout: time between events in milliseconds
256    @cb: callback to invoke when an event occurs
257    @opaque: user data to pass to callback
258
259    Setting timeout to -1 will disable the timer. Setting the timeout
260    to zero will cause it to fire on every event loop iteration.
261
262    Example callback prototype is:
263        def cb(timer,   # int id of the timer
264               opaque): # opaque data passed to eventAddTimeout
265    """
266    cbData = {"cb": cb, "opaque": opaque}
267    ret = libvirtmod.virEventAddTimeout(timeout, cbData)
268    if ret == -1:
269        raise libvirtError('virEventAddTimeout() failed')
270    return ret
271
272
273#
274# a caller for the ff callbacks for custom event loop implementations
275#
276
277def virEventInvokeFreeCallback(opaque: Any) -> None:
278    """
279    Execute callback which frees the opaque buffer
280
281    @opaque: the opaque object passed to addHandle or addTimeout
282
283    WARNING: This function should not be called from any call by libvirt's
284    core. It will most probably cause deadlock in C-level libvirt code.
285    Instead it should be scheduled and called from implementation's stack.
286
287    See https://libvirt.org/html/libvirt-libvirt-event.html#virEventAddHandleFunc
288    for more information.
289
290    This function is not dependent on any event loop implementation.
291    """
292
293    libvirtmod.virEventInvokeFreeCallback(opaque[2], opaque[1])
294