Sign Up
Log In
Log In
or
Sign Up
Places
All Projects
Status Monitor
Collapse sidebar
SUSE:SLE-12-SP5:GA
rhnlib.24203
rhnlib-git-0.bd3cc6b.obscpio
Overview
Repositories
Revisions
Requests
Users
Attributes
Meta
File rhnlib-git-0.bd3cc6b.obscpio of Package rhnlib.24203
07070100000000000041FD000000000000000000000001625E86EB00000000000000000000000000000000000000000000000700000000rhnlib07070100000001000081B4000000000000000000000001625E86EB000046AC000000000000000000000000000000000000000F00000000rhnlib/COPYING GNU GENERAL PUBLIC LICENSE Version 2, June 1991 Copyright (C) 1989, 1991 Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA Everyone is permitted to copy and distribute verbatim copies of this license document, but changing it is not allowed. Preamble The licenses for most software are designed to take away your freedom to share and change it. By contrast, the GNU General Public License is intended to guarantee your freedom to share and change free software--to make sure the software is free for all its users. This General Public License applies to most of the Free Software Foundation's software and to any other program whose authors commit to using it. (Some other Free Software Foundation software is covered by the GNU Lesser General Public License instead.) You can apply it to your programs, too. When we speak of free software, we are referring to freedom, not price. Our General Public Licenses are designed to make sure that you have the freedom to distribute copies of free software (and charge for this service if you wish), that you receive source code or can get it if you want it, that you can change the software or use pieces of it in new free programs; and that you know you can do these things. To protect your rights, we need to make restrictions that forbid anyone to deny you these rights or to ask you to surrender the rights. These restrictions translate to certain responsibilities for you if you distribute copies of the software, or if you modify it. For example, if you distribute copies of such a program, whether gratis or for a fee, you must give the recipients all the rights that you have. You must make sure that they, too, receive or can get the source code. And you must show them these terms so they know their rights. We protect your rights with two steps: (1) copyright the software, and (2) offer you this license which gives you legal permission to copy, distribute and/or modify the software. Also, for each author's protection and ours, we want to make certain that everyone understands that there is no warranty for this free software. If the software is modified by someone else and passed on, we want its recipients to know that what they have is not the original, so that any problems introduced by others will not reflect on the original authors' reputations. Finally, any free program is threatened constantly by software patents. We wish to avoid the danger that redistributors of a free program will individually obtain patent licenses, in effect making the program proprietary. To prevent this, we have made it clear that any patent must be licensed for everyone's free use or not licensed at all. The precise terms and conditions for copying, distribution and modification follow. GNU GENERAL PUBLIC LICENSE TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION 0. This License applies to any program or other work which contains a notice placed by the copyright holder saying it may be distributed under the terms of this General Public License. The "Program", below, refers to any such program or work, and a "work based on the Program" means either the Program or any derivative work under copyright law: that is to say, a work containing the Program or a portion of it, either verbatim or with modifications and/or translated into another language. (Hereinafter, translation is included without limitation in the term "modification".) Each licensee is addressed as "you". Activities other than copying, distribution and modification are not covered by this License; they are outside its scope. The act of running the Program is not restricted, and the output from the Program is covered only if its contents constitute a work based on the Program (independent of having been made by running the Program). Whether that is true depends on what the Program does. 1. You may copy and distribute verbatim copies of the Program's source code as you receive it, in any medium, provided that you conspicuously and appropriately publish on each copy an appropriate copyright notice and disclaimer of warranty; keep intact all the notices that refer to this License and to the absence of any warranty; and give any other recipients of the Program a copy of this License along with the Program. You may charge a fee for the physical act of transferring a copy, and you may at your option offer warranty protection in exchange for a fee. 2. You may modify your copy or copies of the Program or any portion of it, thus forming a work based on the Program, and copy and distribute such modifications or work under the terms of Section 1 above, provided that you also meet all of these conditions: a) You must cause the modified files to carry prominent notices stating that you changed the files and the date of any change. b) You must cause any work that you distribute or publish, that in whole or in part contains or is derived from the Program or any part thereof, to be licensed as a whole at no charge to all third parties under the terms of this License. c) If the modified program normally reads commands interactively when run, you must cause it, when started running for such interactive use in the most ordinary way, to print or display an announcement including an appropriate copyright notice and a notice that there is no warranty (or else, saying that you provide a warranty) and that users may redistribute the program under these conditions, and telling the user how to view a copy of this License. (Exception: if the Program itself is interactive but does not normally print such an announcement, your work based on the Program is not required to print an announcement.) These requirements apply to the modified work as a whole. If identifiable sections of that work are not derived from the Program, and can be reasonably considered independent and separate works in themselves, then this License, and its terms, do not apply to those sections when you distribute them as separate works. But when you distribute the same sections as part of a whole which is a work based on the Program, the distribution of the whole must be on the terms of this License, whose permissions for other licensees extend to the entire whole, and thus to each and every part regardless of who wrote it. Thus, it is not the intent of this section to claim rights or contest your rights to work written entirely by you; rather, the intent is to exercise the right to control the distribution of derivative or collective works based on the Program. In addition, mere aggregation of another work not based on the Program with the Program (or with a work based on the Program) on a volume of a storage or distribution medium does not bring the other work under the scope of this License. 3. You may copy and distribute the Program (or a work based on it, under Section 2) in object code or executable form under the terms of Sections 1 and 2 above provided that you also do one of the following: a) Accompany it with the complete corresponding machine-readable source code, which must be distributed under the terms of Sections 1 and 2 above on a medium customarily used for software interchange; or, b) Accompany it with a written offer, valid for at least three years, to give any third party, for a charge no more than your cost of physically performing source distribution, a complete machine-readable copy of the corresponding source code, to be distributed under the terms of Sections 1 and 2 above on a medium customarily used for software interchange; or, c) Accompany it with the information you received as to the offer to distribute corresponding source code. (This alternative is allowed only for noncommercial distribution and only if you received the program in object code or executable form with such an offer, in accord with Subsection b above.) The source code for a work means the preferred form of the work for making modifications to it. For an executable work, complete source code means all the source code for all modules it contains, plus any associated interface definition files, plus the scripts used to control compilation and installation of the executable. However, as a special exception, the source code distributed need not include anything that is normally distributed (in either source or binary form) with the major components (compiler, kernel, and so on) of the operating system on which the executable runs, unless that component itself accompanies the executable. If distribution of executable or object code is made by offering access to copy from a designated place, then offering equivalent access to copy the source code from the same place counts as distribution of the source code, even though third parties are not compelled to copy the source along with the object code. 4. You may not copy, modify, sublicense, or distribute the Program except as expressly provided under this License. Any attempt otherwise to copy, modify, sublicense or distribute the Program is void, and will automatically terminate your rights under this License. However, parties who have received copies, or rights, from you under this License will not have their licenses terminated so long as such parties remain in full compliance. 5. You are not required to accept this License, since you have not signed it. However, nothing else grants you permission to modify or distribute the Program or its derivative works. These actions are prohibited by law if you do not accept this License. Therefore, by modifying or distributing the Program (or any work based on the Program), you indicate your acceptance of this License to do so, and all its terms and conditions for copying, distributing or modifying the Program or works based on it. 6. Each time you redistribute the Program (or any work based on the Program), the recipient automatically receives a license from the original licensor to copy, distribute or modify the Program subject to these terms and conditions. You may not impose any further restrictions on the recipients' exercise of the rights granted herein. You are not responsible for enforcing compliance by third parties to this License. 7. If, as a consequence of a court judgment or allegation of patent infringement or for any other reason (not limited to patent issues), conditions are imposed on you (whether by court order, agreement or otherwise) that contradict the conditions of this License, they do not excuse you from the conditions of this License. If you cannot distribute so as to satisfy simultaneously your obligations under this License and any other pertinent obligations, then as a consequence you may not distribute the Program at all. For example, if a patent license would not permit royalty-free redistribution of the Program by all those who receive copies directly or indirectly through you, then the only way you could satisfy both it and this License would be to refrain entirely from distribution of the Program. If any portion of this section is held invalid or unenforceable under any particular circumstance, the balance of the section is intended to apply and the section as a whole is intended to apply in other circumstances. It is not the purpose of this section to induce you to infringe any patents or other property right claims or to contest validity of any such claims; this section has the sole purpose of protecting the integrity of the free software distribution system, which is implemented by public license practices. Many people have made generous contributions to the wide range of software distributed through that system in reliance on consistent application of that system; it is up to the author/donor to decide if he or she is willing to distribute software through any other system and a licensee cannot impose that choice. This section is intended to make thoroughly clear what is believed to be a consequence of the rest of this License. 8. If the distribution and/or use of the Program is restricted in certain countries either by patents or by copyrighted interfaces, the original copyright holder who places the Program under this License may add an explicit geographical distribution limitation excluding those countries, so that distribution is permitted only in or among countries not thus excluded. In such case, this License incorporates the limitation as if written in the body of this License. 9. The Free Software Foundation may publish revised and/or new versions of the General Public License from time to time. Such new versions will be similar in spirit to the present version, but may differ in detail to address new problems or concerns. Each version is given a distinguishing version number. If the Program specifies a version number of this License which applies to it and "any later version", you have the option of following the terms and conditions either of that version or of any later version published by the Free Software Foundation. If the Program does not specify a version number of this License, you may choose any version ever published by the Free Software Foundation. 10. If you wish to incorporate parts of the Program into other free programs whose distribution conditions are different, write to the author to ask for permission. For software which is copyrighted by the Free Software Foundation, write to the Free Software Foundation; we sometimes make exceptions for this. Our decision will be guided by the two goals of preserving the free status of all derivatives of our free software and of promoting the sharing and reuse of software generally. NO WARRANTY 11. BECAUSE THE PROGRAM IS LICENSED FREE OF CHARGE, THERE IS NO WARRANTY FOR THE PROGRAM, TO THE EXTENT PERMITTED BY APPLICABLE LAW. EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT HOLDERS AND/OR OTHER PARTIES PROVIDE THE PROGRAM "AS IS" WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE PROGRAM IS WITH YOU. SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION. 12. IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MAY MODIFY AND/OR REDISTRIBUTE THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES, INCLUDING ANY GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED TO LOSS OF DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER PROGRAMS), EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGES. END OF TERMS AND CONDITIONS How to Apply These Terms to Your New Programs If you develop a new program, and you want it to be of the greatest possible use to the public, the best way to achieve this is to make it free software which everyone can redistribute and change under these terms. To do so, attach the following notices to the program. It is safest to attach them to the start of each source file to most effectively convey the exclusion of warranty; and each file should have at least the "copyright" line and a pointer to where the full notice is found. <one line to give the program's name and a brief idea of what it does.> Copyright (C) <year> <name of author> This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. Also add information on how to contact you by electronic and paper mail. If the program is interactive, make it output a short notice like this when it starts in an interactive mode: Gnomovision version 69, Copyright (C) year name of author Gnomovision comes with ABSOLUTELY NO WARRANTY; for details type `show w'. This is free software, and you are welcome to redistribute it under certain conditions; type `show c' for details. The hypothetical commands `show w' and `show c' should show the appropriate parts of the General Public License. Of course, the commands you use may be called something other than `show w' and `show c'; they could even be mouse-clicks or menu items--whatever suits your program. You should also get your employer (if you work as a programmer) or your school, if any, to sign a "copyright disclaimer" for the program, if necessary. Here is a sample; alter the names: Yoyodyne, Inc., hereby disclaims all copyright interest in the program `Gnomovision' (which makes passes at compilers) written by James Hacker. <signature of Ty Coon>, 1 April 1989 Ty Coon, President of Vice This General Public License does not permit incorporating your program into proprietary programs. If your program is a subroutine library, you may consider it more useful to permit linking proprietary applications with the library. If this is what you want to do, use the GNU Lesser General Public License instead of this License. 07070100000002000081B4000000000000000000000001625E86EB0000129C000000000000000000000000000000000000001100000000rhnlib/ChangeLogVersion 1.8 - Breaking transports.Transport.request() into smaller pieces - Added factories for the transport classes in the Server object - Fixed a bug in File.__del__ (if close is called twice, it tried to call None) - Fixed a bug in SSL.read() (on an open stream, read() would block instead of returning with the amount of data available). - The previous fix in SSL.read uncovered a nastier bug in httplib: http://python.org/sf/988120 Fixed it in our HTTPResponse subclass - The internal _httplib did not have a _read_chunked (http://bugzilla.redhat.com/128008) Version 1.7 - Renamed _xmlrpclib.py to _internal_xmlrpclib.py to avoid _xmlrpclib from self-importing (bug #119876) - Added a User-Agent header for CONNECTION tunnels. - Fixed lookupEncoding Version 1.6 - GET methods have empty bodies, so rpclib.GETServer._req_body should return empty string not None - Properly convert chunk sizes in python 1.5.2 Version 1.5 - Added iterator for the UserDictCase objects. Easier than subclassing IterableUserDict since the code has to work with python 1.5.2. - rhn/transports.py: one can pass to an Input object a stream that doesn't have a close() method. - rpclib exposes File now too - rhn/transports.py: the Output object now sets Content-Length (which will be cleared when sent upstream) Version 1.3 - work on HTTP/HTTPS redirection support at the Server class level in rhnlib.py: - new data in the instance to keep setting made to the underlying transport - new methods to control and check redirection: allow_redirect() and redirected() which returns the redirected URL or None - separated the initialization code to provide a default transport as an new method default_transport() - added redirection handling in _request() with some associated checking like https->http redirection being forbidden - at the transport level, request() checks redirection, position a redirection flag and returns None if redirected. - Added some flexibility on the way to specify proxy informations, the get_proxy_info() function of rhnlib.py is used to extract host, port, username and password information from the proxy URL or old host:port format. - Fixed an error related to decoding the tuples that XMLRPC returns back. Version 1.2 - Changed the ugly way of passing around InputStream as an exception - The transport flags are now more generic, allowing for future extensibility - Added a transport flag 'allow_partial_content' which will prevent the transport from raising an exception if a 206 status code is received - New Server methods: get_response_headers, get_response_status and get_response_reason; you can now inspect the headers even for successful connections - Changed the GETServer method calls to accept offset and amount - Added get_content_range, to parse the interesting values out of the HTTP headers - Added accept_ranges, a function that returns true if the server accepts ranges. Note that even though the server doesn't advertise itself as accepting ranges, it's still possible for the client to issue a range request; depending on the response status code, the client can see if the server supports ranges (206) or not (200) - set_header and add_header are now different and doing what they are supposed to do - added get_transport_flags() as a Server method - fixed an instance when Input.io was not rewound Version 1.1 - Minor bugfixes - Building the rpm with distutils is too painful; switching to a human-maintained spec file instead. Version 1.0 - More or less 0.9 for public consumption Version 0.9 - SSL.py: use load_verify_locations multiple times. This function seems to do more than my poor method of getting the cert store and adding certs by hand, probably because load_certificate from a buffer can only handle one certificate at the time. - With the addition of SmartIO, the memory usage should be limited to 16k, after which the objects will be dumped on the disk in temporary files. - Fixed yet another file descriptor leak (and still in SSL). Version 0.8 - User-Agent, X-Info and X-Client-Version were not present in the HTTP headers; fixed - Completely deprecating rhnHTTPlib: swallowed reportError - Fixed stupid typo introduced when fixing the previous bug. - Fixed #69518 (up2date seems to never properly reauthenticate after a auth timeout) - Fixed #69311 (leaking file descriptors over SSL connections). - Fixed #68911 (and some other bugs that were related to this one). - Fixed a proxy related bug. Thanks to Gil Chilton <Gil.Chilton@capitalone.com> for helping me discover and test it. Version 0.7 - More careful distutils packaging. 07070100000003000081B4000000000000000000000001625E86EB00000048000000000000000000000000000000000000001300000000rhnlib/MANIFEST.ininclude ChangeLog COPYING README TODO rhnlib.spec Makefile.rhnlib_build 07070100000004000081B4000000000000000000000001625E86EB00000067000000000000000000000000000000000000001000000000rhnlib/MakefileNAME := rhnlib SPECFILE = $(NAME).spec DO_TAR_GZ := git-tree-sha1 include ../../../rel-eng/Makefile 07070100000005000081B4000000000000000000000001625E86EB00000361000000000000000000000000000000000000001700000000rhnlib/Makefile.pythonTHIS_MAKEFILE := $(realpath $(lastword $(MAKEFILE_LIST))) CURRENT_DIR := $(dir $(THIS_MAKEFILE)) include $(CURRENT_DIR)../../rel-eng/Makefile.python # Docker tests variables DOCKER_CONTAINER_BASE = systemsmanagement/uyuni/master/docker/containers/uyuni-master DOCKER_REGISTRY = registry.opensuse.org DOCKER_RUN_EXPORT = "PYTHONPATH=$PYTHONPATH" DOCKER_VOLUMES = -v "$(CURDIR)/../../:/manager" __pylint :: $(call update_pip_env) cd ..; pylint --rcfile=rhn/pylintrc rhn | tee rhn/reports/pylint.log || true docker_pylint :: docker run --rm -e $(DOCKER_RUN_EXPORT) $(DOCKER_VOLUMES) $(DOCKER_REGISTRY)/$(DOCKER_CONTAINER_BASE)-pgsql /bin/sh -c "cd /manager/python/rhn; make -f Makefile.python __pylint" docker_shell :: docker run -t -i --rm -e $(DOCKER_RUN_EXPORT) $(DOCKER_VOLUMES) $(DOCKER_REGISTRY)/$(DOCKER_CONTAINER_BASE)-pgsql /bin/bash 07070100000006000081B4000000000000000000000001625E86EB0000061F000000000000000000000000000000000000001700000000rhnlib/Makefile.rhnlib# Makefile for the rhn modules # PYTHON = /usr/bin/python NAME = rhnlib VERSION = $(call _get_tag,VERSION) RELEASE = $(call _get_tag,RELEASE) _get_tag = $(shell rpm -q --queryformat "%{$1}" --specfile $(NAME).spec | head -1) WORKDIR = $(shell pwd) SOURCEDIR = $(WORKDIR) SRCRPMDIR = $(WORKDIR) BUILDDIR = $(WORKDIR) RPMDIR = $(WORKDIR) RPMBUILD = rpmbuild RPMBUILDCMD = $(RPMBUILD) --define "_sourcedir $(SOURCEDIR)" --define "_srcrpmdir $(SRCRPMDIR)" --define "_builddir $(BUILDDIR)" --define "_rpmdir $(RPMDIR)" --define "_build_name_fmt %{NAME}-%{VERSION}-%{RELEASE}.%{ARCH}.rpm" TARBALL = $(NAME)-$(VERSION).tar.gz .PHONY: build install rpm devel release clean all: build setup.py: setup.py.in $(VERFILE) @sed -e 's/@VERSION@/$(VERSION)/' -e 's/@NAME@/$(NAME)/' $< > $@ setup.cfg: setup.cfg.in $(VERFILE) @sed 's/@RELEASE@/$(RELEASE)/' $< > $@ build: setup.py @$(PYTHON) setup.py build for i in build/lib*/rhn/rpclib.py build/lib*/rhn/transports.py ; do sed -i 's!\$$Revision\$$!$(VERSION)-$(RELEASE)!' $$i ; done install: setup.py setup.cfg @$(PYTHON) setup.py install $(TARBALL): clean setup.py setup.cfg @$(PYTHON) setup.py sdist $(DISTUTILS_OPT) @mv dist/$(TARBALL) $(WORKDIR) @rm -rf dist tar: $(TARBALL) test-srpm: $(TARBALL) $(RPMBUILDCMD) --nodeps -ts $(TARBALL) test-rpm: $(TARBALL) $(RPMBUILDCMD) --nodeps -ta $(TARBALL) rpm: $(TARBALL) $(RPMBUILD) -ta $(TARBALL) release: tar test-srpm clean: @rm -rfv dist build MANIFEST setup.cfg setup.py *pyc *.tar.gz *.rpm $(NAME)-$(VERSION) 07070100000007000081B4000000000000000000000001625E86EB00000422000000000000000000000000000000000000000E00000000rhnlib/READMErhnlib - A collection of python modules used by the Spacewalk (http://spacewalk.redhat.com) software. Copyright (C) 2002--2015 Red Hat, Inc. This library is distributed under the terms of the GNU Public License (GPL). 0. HISTORY Traditionally, rhn used a patched version of Fredrik Lundh's xmlrpclib, shipped with Red Hat 6.2, 7.0, 7.1, 7.2 and 7.3 in a package called python-xmlrpc. The additions included improved transport objects (proxy support, a better SSL framework, encodings and transfers), most of them part of a module called cgiwrap. Since python 2.2 (and previous python 2 releases) ships Fredrik Lundh's xmlrpclib as a standard library, maintaining a patched version of xmlrpclib was no longer feasible. SSL support in python 2.2 is also hopelessly incomplete, so using an external module seems to be a better approach. We chose pyOpenSSL instead of m2crypto, partly because swig seems to change rather quickly. rhnlib tries to reuse as much as possible from the old cgiwrap, as well as from what the standard python libraries provide. 07070100000008000081B4000000000000000000000001625E86EB00002A7F000000000000000000000000000000000000000E00000000rhnlib/SSL.py# # Higher-level SSL objects used by rpclib # # Copyright (c) 2002--2017 Red Hat, Inc. # # Author: Mihai Ibanescu <misa@redhat.com> # # In addition, as a special exception, the copyright holders give # permission to link the code of portions of this program with the # OpenSSL library under certain conditions as described in each # individual source file, and distribute linked combinations # including the two. # You must obey the GNU General Public License in all respects # for all of the code used other than OpenSSL. If you modify # file(s) with this exception, you may extend this exception to your # version of the file(s), but you are not obligated to do so. If you # do not wish to do so, delete this exception statement from your # version. If you delete this exception statement from all source # files in the program, then also delete it here. """ rhn.SSL builds an abstraction on top of the objects provided by pyOpenSSL """ # SSL.crypto is provided to other modules from OpenSSL import crypto import ssl as SSL import os import socket import select from rhn.stringutils import bstr import sys DEFAULT_TIMEOUT = 120 if hasattr(socket, 'sslerror'): socket_error = socket.sslerror else: from ssl import socket_error try: from ssl import CertificateError except ImportError: # python 2.6 from backports.ssl_match_hostname import match_hostname, CertificateError class SSLSocket: """ Class that wraps a pyOpenSSL Connection object, adding more methods """ def __init__(self, socket, trusted_certs=None): # SSL.Context object self._ctx = None # SSL.Connection object self._connection = None self._sock = socket self._trusted_certs = [] # convert None to empty list trusted_certs = trusted_certs or [] for f in trusted_certs: self.add_trusted_cert(f) # SSL method to use if hasattr(SSL, 'PROTOCOL_TLS'): self._ssl_method = SSL.PROTOCOL_TLS else: self._ssl_method = SSL.PROTOCOL_SSLv23 # Buffer size for reads self._buffer_size = 8192 # Position, for tell() self._pos = 0 # Buffer self._buffer = bstr("") # Flag to show if makefile() was called self._makefile_called = 0 self._closed = None def add_trusted_cert(self, file): """ Adds a trusted certificate to the certificate store of the SSL context object. """ if not os.access(file, os.R_OK): raise ValueError("Unable to read certificate file %s" % file) self._trusted_certs.append(file.encode("utf-8")) def init_ssl(self, server_name=None): """ Initializes the SSL connection. """ self._check_closed() # Get a context if hasattr(SSL, 'SSLContext'): self._ctx = SSL.SSLContext(self._ssl_method) self._ctx.verify_mode = SSL.CERT_REQUIRED self._ctx.check_hostname = True self._ctx.load_default_certs(SSL.Purpose.SERVER_AUTH) if self._trusted_certs: # We have been supplied with trusted CA certs for f in self._trusted_certs: self._ctx.load_verify_locations(f) self._connection = self._ctx.wrap_socket(self._sock, server_hostname=server_name) else: # Python 2.6-2.7.8 cacert = None if self._trusted_certs: # seems python2.6 supports only 1 cacert = self._trusted_certs[0] self._connection = SSL.wrap_socket(self._sock, ssl_version=self._ssl_method, cert_reqs=SSL.CERT_REQUIRED, ca_certs=cacert) match_hostname(self._connection.getpeercert(), server_name) def makefile(self, mode, bufsize=None): """ Returns self, since we are a file-like object already """ if bufsize: self._buffer_size = bufsize # Increment the counter with the number of times we've called makefile # - we don't want close to actually close things until all the objects # that originally called makefile() are gone self._makefile_called = self._makefile_called + 1 return self def close(self): """ Closes the SSL connection """ # XXX Normally sock.makefile does a dup() on the socket file # descriptor; httplib relies on this, but there is no dup for an ssl # connection; so we have to count how may times makefile() was called if self._closed: # Nothing to do return if not self._makefile_called: self._really_close() return self._makefile_called = self._makefile_called - 1 # BZ 1464157 - Python 3 http attempts to call this method during close, # at least add it empty def flush(self): pass def _really_close(self): # No connection was established if self._connection is None: return self._connection.close() self._closed = 1 def _check_closed(self): if self._closed: raise ValueError("I/O operation on closed file") def __getattr__(self, name): if hasattr(self._connection, name): return getattr(self._connection, name) raise AttributeError(name) # File methods def isatty(self): """ Returns false always. """ return 0 def tell(self): return self._pos def seek(self, pos, mode=0): raise NotImplementedError("seek") def read(self, amt=None): """ Reads up to amt bytes from the SSL connection. """ self._check_closed() # Initially, the buffer size is the default buffer size. # Unfortunately, pending() does not return meaningful data until # recv() is called, so we only adjust the buffer size after the # first read buffer_size = self._buffer_size buffer_length = len(self._buffer) # Read only the specified amount of data while buffer_length < amt or amt is None: # if amt is None (read till the end), fills in self._buffer if amt is not None: buffer_size = min(amt - buffer_length, buffer_size) try: data = self._connection.recv(buffer_size) self._buffer = self._buffer + data buffer_length = len(self._buffer) # More bytes to read? pending = self._connection.pending() if pending == 0 and buffer_length == amt: # we're done here break except SSL.SSLError as err: if err.args[0] == SSL.SSL_ERROR_ZERO_RETURN: # Nothing more to be read break elif err.args[0] == SSL.SSL_ERROR_SYSCALL: e = sys.exc_info()[1] print("SSL exception", e.args) break elif err.args[0] == SSL.SSL_ERROR_WANT_WRITE: self._poll(select.POLLOUT, 'read') elif err.args[0] == SSL.SSL_ERROR_WANT_READ: self._poll(select.POLLIN, 'read') if amt: ret = self._buffer[:amt] self._buffer = self._buffer[amt:] else: ret = self._buffer self._buffer = bstr("") self._pos = self._pos + len(ret) return ret def readinto(self, buf): buf[:] = self.read(len(buf)) return len(buf) def _poll(self, filter_type, caller_name): poller = select.poll() poller.register(self._sock, filter_type) res = poller.poll(self._sock.gettimeout() * 1000) if res == []: raise TimeoutException("Connection timed out on %s" % caller_name) def write(self, data): """ Writes to the SSL connection. """ self._check_closed() # XXX Should use sendall # sent = self._connection.sendall(data) origlen = len(data) while True: try: sent = self._connection.send(data) if sent == len(data): break data = data[sent:] except SSL.SSLError as err: if err.args[0] == SSL.SSL_ERROR_WANT_WRITE: self._poll(select.POLLOUT, 'write') elif err.args[0] == SSL.SSL_ERROR_WANT_READ: self._poll(select.POLLIN, 'write') return origlen def recv(self, amt): return self.read(amt) send = write sendall = write def readline(self, length=None): """ Reads a single line (up to `length' characters long) from the SSL connection. """ self._check_closed() while True: # charcount contains the number of chars to be outputted (or None # if none to be outputted at this time) charcount = None i = self._buffer.find(bstr('\n')) if i >= 0: # Go one char past newline charcount = i + 1 elif length and len(self._buffer) >= length: charcount = length if charcount is not None: ret = self._buffer[:charcount] self._buffer = self._buffer[charcount:] self._pos = self._pos + len(ret) return ret # Determine the number of chars to be read next bufsize = self._buffer_size if length: # we know length > len(self._buffer) bufsize = min(self._buffer_size, length - len(self._buffer)) try: data = self._connection.recv(bufsize) self._buffer = self._buffer + data except SSL.SSLError as err: if err.args[0] == SSL.SSL_ERROR_ZERO_RETURN: # Nothing more to be read break elif err.args[0] == SSL.SSL_ERROR_WANT_WRITE: self._poll(select.POLLOUT, 'readline') elif err.args[0] == SSL.SSL_ERROR_WANT_READ: self._poll(select.POLLIN, 'readline') # We got here if we're done reading, so return everything ret = self._buffer self._buffer = "" self._pos = self._pos + len(ret) return ret class TimeoutException(SSL.SSLError, socket.timeout): def __init__(self, *args): self.args = args def __str__(self): return "Timeout Exception" 07070100000009000081B4000000000000000000000001625E86EB00000874000000000000000000000000000000000000001200000000rhnlib/SmartIO.py# # Smart IO class # # Copyright (c) 2002--2016 Red Hat, Inc. # # Author: Mihai Ibanescu <misa@redhat.com> """ This module implements the SmartIO class """ import os try: # python2 from cStringIO import StringIO except ImportError: # python3 from io import BytesIO as StringIO class SmartIO: """ The SmartIO class allows one to put a cap on the memory consumption. StringIO objects are very fast, because they are stored in memory, but if they are too big the memory footprint becomes noticeable. The write method of a SmartIO determines if the data that is to be added to the (initially) StrintIO object does not exceed a certain threshold; if it does, it switches the storage to a temporary disk file """ def __init__(self, max_mem_size=16384, force_mem=0): self._max_mem_size = max_mem_size self._io = StringIO() # self._fixed is a flag to show if we're supposed to consider moving # the StringIO object into a tempfile # Invariant: if self._fixed == 0, we have a StringIO (if self._fixed # is 1 and force_mem was 0, then we have a file) if force_mem: self._fixed = 1 else: self._fixed = 0 def set_max_mem_size(self, max_mem_size): self._max_mem_size = max_mem_size def get_max_mem_size(self): return self._max_mem_size def write(self, data): if not self._fixed: # let's consider moving it to a file if len(data) + self._io.tell() > self._max_mem_size: # We'll overflow, change to a tempfile tmpfile = _tempfile() tmpfile.write(self._io.getvalue()) self._fixed = 1 self._io = tmpfile self._io.write(data) def __getattr__(self, name): return getattr(self._io, name) # Creates a temporary file and passes back its file descriptor def _tempfile(): import tempfile (fd, fname) = tempfile.mkstemp(prefix="_rhn_transports-%d-" \ % os.getpid()) # tempfile, unlink it os.unlink(fname) return os.fdopen(fd, "wb+") 0707010000000A000081B4000000000000000000000001625E86EB00000293000000000000000000000000000000000000000C00000000rhnlib/TODO - Check nonblocking support for HTTPS as well. - Allow for multi-valued headers - Wrapper around the socket class, to allow for calling select() on the socket I/O - Split this module intelligently - Build an __init__ file to expose the desired API - Fix the Input and Output objects: setting the encoding from the input object to the output object is difficult (there is a patch somewhere for this, I'll have to look it up -- misa) - Clean up things - More documentation - Check compliance with python 1.5 - Figure out a better way to pass the client version - Figure out what the version we pass in the headers should look like. - Finish the TODO :-) 0707010000000B000081B4000000000000000000000001625E86EB00000BD8000000000000000000000000000000000000001700000000rhnlib/UserDictCase.py# # Copyright (c) 2001--2016 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # # # This file implements a case insensitive dictionary on top of the # UserDict standard python class # try: # python2 from UserDict import UserDict from types import StringType except ImportError: # python3 from collections import UserDict StringType = str from functools import reduce # A dictionary with case insensitive keys class UserDictCase(UserDict): def __init__(self, data = None): self.kcase = {} UserDict.__init__(self, data) def __lower_string(self, key): """ Return the lower() of key if it is a string. """ if isinstance(key, StringType): return key.lower() else: return key # some methods used to make the class work as a dictionary def __setitem__(self, key, value): lkey = self.__lower_string(key) self.data[lkey] = value self.kcase[lkey] = key def __getitem__(self, key): key = self.__lower_string(key) return self.data[key] get = __getitem__ def __delitem__(self, key): key = self.__lower_string(key) del self.data[key] del self.kcase[key] def __contains__(self, key): key = self.__lower_string(key) return key in self.data def keys(self): return self.kcase.values() def items(self): return self.get_hash().items() def has_key(self, key): # obsoleted, left for compatibility with older python return key in self def clear(self): self.data.clear() self.kcase.clear() # return this data as a real hash def get_hash(self): return reduce(lambda a, t, hc=self.kcase: a.update({ hc[t[0]] : t[1]}) or a, self.data.items(), {}) # return the data for marshalling def __getstate__(self): return self.get_hash() # we need a setstate because of the __getstate__ presence screws up deepcopy def __setstate__(self, state): self.__init__(state) # get a dictionary out of this instance ({}.update doesn't get instances) def dict(self): return self.get_hash() def update(self, dict): for (k, v) in dict.items(): self[k] = v # Expose an iterator. This would normally fail if there is no iter() # function defined - but __iter__ will never be called on python 1.5.2 def __iter__(self): return iter(self.data) 0707010000000C000081B4000000000000000000000001625E86EB00000084000000000000000000000000000000000000001300000000rhnlib/__init__.py# # __init__.py # # Copyright (c) 2011--2013 Red Hat, Inc. # """ rhn - A collection of modules used by Red Hat Network Classic """ 0707010000000D000081FD000000000000000000000001625E86EB00000CC3000000000000000000000000000000000000001700000000rhnlib/buildpackage.sh#!/usr/bin/bash # script to build a solaris package of up2date-list # # Adrian Likins <alikins@redhat.com> # NAME=`cat $1 | grep PKG | cut -f2 -d'='| cut -f2 -d\"` TMP=/tmp BUILDPREFIX=$TMP/build-$NAME # er, lame BUILDPREFIX_REL=tmp/build-$NAME DESTPREFIX_REL=opt/redhat/rhn/solaris DESTPREFIX=/${DESTPREFIX_REL} DATADIR=${TMP}/${NAME}-pkginfo/ USERNAME=`whoami` # path for solaris build stuff, will need to get tweaked for hpux/aix, but # theres some infof or that already un use_spec.sh #PATH=/opt/redhat/rhn/solaris/bin:/opt/redhat/gnupro-03r1/H-sparc-sun-solaris2.5/bin:/opt/redhat/gnupro-03r1/contrib:/opt/redhat/rpm/solaris/bin:/home/cygnus/release/bin:/es/cst/bin:/bsp/bin:/usr/progressive/bin:/usr/unsupported/bin:/bin:/usr/ucb:/usr/sbin:/usr/local/bin:/sbin:/usr/kerberos/bin:/usr/local/bin:/usr/bin:/usr/X11R6/bin PYTHONDIR=$DESTPREFIX/lib/python2.2/ PYTHONMODDIR=$PYTHONDIR/site-packages BANGPATH=$DESTPREFIX/bin/python # helper stuff #INSTALL=/opt/redhat/gnupro-03r1/contrib/H-sparc-sun-solaris2.6/bin/install #INSTALL=/es/unsupported/sparc-sun-solaris2.5/src/fileutils-4.1/src/ginstall INSTALL=/usr/ucb/install FILELIST=$TMP/filelist-$NAME mkdir -p $BUILDPREFIX # clean up the build root rm -rf $BUILDPREFIX/* # er, duh #make python setup.py build # install into the fake prefix # make PREFIX=$BUILDPREFIX INSTALL=$INSTALL PYTHONDIR=$PYTHONDIR PYTHONMOD_DIR=$PYTHONMODDIR BANGPATH=$BANGPATH install python setup.py install --prefix=$BUILDPREFIX # find the packages installed into the build root find $BUILDPREFIX -print > $FILELIST # create the package prototype mkdir -p $DATADIR cat $FILELIST | pkgproto > $DATADIR/prototype # add the info about pkginfo file echo "i pkginfo" > $DATADIR/prototype.tmp cat $DATADIR/prototype >> $DATADIR/prototype.tmp # write it back again cp $DATADIR/prototype.tmp $DATADIR/prototype PROTOTYPE=$DATADIR/prototype # cp our pkginfo file to the datafir cp pkginfo $DATADIR/pkginfo # okay, now the fun begins. The problem is that all the paths are # wrong. You can't simply change the prototype, cause it will # look for the new paths, where nothing exist. So you have to # build the package, then tweak the package info directly. # build the actual package, pre munge pkgmk -o -r / -d $TMP -f $PROTOTYPE ls -la $TMP # the package is in filesystem format, so we # need to go into that dir and start mucking # with stuff cd $TMP/${NAME}/root/ # since the package as is has a local # equilvient of $(BUILDPREFIX), we need to # create a dir structure representing what # the dest version is mkdir -p $DESTPREFIX_REL # move the stuff to the new dir that we actually # want... echo $BUILDPREFIX_REL $DESTPREFIX_REL echo pwd echo mv ${BUILDPREFIX_REL}/* ${DESTPREFIX_REL}/ rm -rf $BUILDPREFIX_REL # pkgmap is the manifest of files and what not # we need to blip it about to sub in the new file # paths cd .. # use perl cause I'm dumb and lazy perl -pi -e "s|${BUILDPREFIX}|${DESTPREFIX}|g" pkgmap # fix the file ownership as well perl -pi -e "s|${USERNAME}|root|g" pkgmap # now that we've got the file all munged up, lets # go ahead and convert the file to a datastream format pkgtrans -s $TMP $TMP/$NAME.package $NAME # probabaly want to compress the package at this # point 0707010000000E000081B4000000000000000000000001625E86EB00002806000000000000000000000000000000000000001600000000rhnlib/connections.py# # Connection objects # # Copyright (c) 2002--2016 Red Hat, Inc. # Copyright (c) 2022 SUSE, LLC # # Author: Mihai Ibanescu <misa@redhat.com> import base64 import encodings.idna import socket from platform import python_version from rhn.stringutils import bstr, ustr, sstr from rhn import SSL from rhn import nonblocking try: # python2 import httplib # Import into the local namespace some httplib-related names from httplib import _CS_REQ_SENT, _CS_IDLE, ResponseNotReady import xmlrpclib except ImportError: # python3 import http.client as httplib # Import into the local namespace some httplib-related names from http.client import _CS_REQ_SENT, _CS_IDLE, ResponseNotReady import xmlrpc.client as xmlrpclib class HTTPResponse(httplib.HTTPResponse): def set_callback(self, rs, ws, ex, user_data, callback): if not isinstance(self.fp, nonblocking.NonBlockingFile): self.fp = nonblocking.NonBlockingFile(self.fp) self.fp.set_callback(rs, ws, ex, user_data, callback) class HTTPConnection(httplib.HTTPConnection): response_class = HTTPResponse def __init__(self, host, port=None, timeout=SSL.DEFAULT_TIMEOUT): if python_version() >= '2.6.1': httplib.HTTPConnection.__init__(self, host, port, timeout=timeout) else: httplib.HTTPConnection.__init__(self, host, port) self._cb_rs = [] self._cb_ws = [] self._cb_ex = [] self._cb_user_data = None self._cb_callback = None self._user_agent = "rhn.connections $Revision$ (python)" self.timeout = timeout def set_callback(self, rs, ws, ex, user_data, callback): # XXX check the params self._cb_rs = rs self._cb_ws = ws self._cb_ex = ex self._cb_user_data = user_data self._cb_callback = callback def set_user_agent(self, user_agent): self._user_agent = user_agent # XXX Had to copy the function from httplib.py, because the nonblocking # framework had to be initialized def getresponse(self): "Get the response from the server." # check if a prior response has been completed if self.__response and self.__response.isclosed(): self.__response = None # # if a prior response exists, then it must be completed (otherwise, we # cannot read this response's header to determine the connection-close # behavior) # # note: if a prior response existed, but was connection-close, then the # socket and response were made independent of this HTTPConnection # object since a new request requires that we open a whole new # connection # # this means the prior response had one of two states: # 1) will_close: this connection was reset and the prior socket and # response operate independently # 2) persistent: the response was retained and we await its # isclosed() status to become true. # if self.__state != _CS_REQ_SENT or self.__response: raise ResponseNotReady() if self.debuglevel > 0: response = self.response_class(self.sock, self.debuglevel) else: response = self.response_class(self.sock) # The only modification compared to the stock HTTPConnection if self._cb_callback: response.set_callback(self._cb_rs, self._cb_ws, self._cb_ex, self._cb_user_data, self._cb_callback) response.begin() assert response.will_close != httplib._UNKNOWN self.__state = _CS_IDLE if response.will_close: # this effectively passes the connection to the response self.close() else: # remember this, so we can tell when it is complete self.__response = response return response def connect(self): httplib.HTTPConnection.connect(self) self.sock.settimeout(self.timeout) class HTTPProxyConnection(HTTPConnection): def __init__(self, proxy, host, port=None, username=None, password=None, timeout=SSL.DEFAULT_TIMEOUT): # The connection goes through the proxy HTTPConnection.__init__(self, proxy, timeout=timeout) # save the proxy values self.__proxy, self.__proxy_port = self.host, self.port # self.host and self.port will point to the real host if hasattr(self, '_get_hostport'): self.host, self.port = self._get_hostport(host, port) else: self._set_hostport(host, port) # save the host and port self._host, self._port = self.host, self.port # Authenticated proxies support self.__username = username self.__password = password def connect(self): # We are actually connecting to the proxy if hasattr(self, '_get_hostport'): self.host, self.port = self._get_hostport(self.__proxy, self.__proxy_port) else: self._set_hostport(self.__proxy, self.__proxy_port) HTTPConnection.connect(self) # Restore the real host and port if hasattr(self, '_get_hostport'): self.host, self.port = self._get_hostport(self._host, self._port) else: self._set_hostport(self._host, self._port) def putrequest(self, method, url, skip_host=0): # The URL has to include the real host hostname = self._host if self._port != self.default_port: hostname = hostname + ':' + str(self._port) newurl = "http://%s%s" % (hostname, url) # Piggyback on the parent class HTTPConnection.putrequest(self, method, newurl, skip_host=skip_host) # Add proxy-specific headers self._add_proxy_headers() def _add_proxy_headers(self): if not self.__username: return # Authenticated proxy userpass = "%s:%s" % (self.__username, self.__password) enc_userpass = base64.encodestring(bstr(userpass)).replace(bstr("\n"), bstr("")) self.putheader("Proxy-Authorization", "Basic %s" % sstr(enc_userpass)) def _set_hostport(self, host, port): (self.host, self.port) = self._get_hostport(host, port) class HTTPSConnection(HTTPConnection): response_class = HTTPResponse default_port = httplib.HTTPSConnection.default_port def __init__(self, host, port=None, trusted_certs=None, timeout=SSL.DEFAULT_TIMEOUT): HTTPConnection.__init__(self, host, port, timeout=timeout) trusted_certs = trusted_certs or [] self.trusted_certs = trusted_certs def connect(self): "Connect to a host on a given (SSL) port" results = socket.getaddrinfo(self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM) for r in results: af, socktype, proto, canonname, sa = r try: sock = socket.socket(af, socktype, proto) except socket.error: sock = None continue try: sock.connect((self.host, self.port)) sock.settimeout(self.timeout) except socket.error: sock.close() sock = None continue break if sock is None: raise socket.error("Unable to connect to the host and port specified") self.sock = SSL.SSLSocket(sock, self.trusted_certs) self.sock.init_ssl(self.host) class HTTPSProxyResponse(HTTPResponse): def begin(self): HTTPResponse.begin(self) self.will_close = 0 class HTTPSProxyConnection(HTTPProxyConnection): default_port = HTTPSConnection.default_port def __init__(self, proxy, host, port=None, username=None, password=None, trusted_certs=None, timeout=SSL.DEFAULT_TIMEOUT): HTTPProxyConnection.__init__(self, proxy, host, port, username, password, timeout=timeout) trusted_certs = trusted_certs or [] self.trusted_certs = trusted_certs def connect(self): # Set the connection with the proxy HTTPProxyConnection.connect(self) # Use the stock HTTPConnection putrequest host = "%s:%s" % (self._host, self._port) HTTPConnection.putrequest(self, "CONNECT", host) # Add proxy-specific stuff self._add_proxy_headers() # And send the request HTTPConnection.endheaders(self) # Save the response class response_class = self.response_class # And replace the response class with our own one, which does not # close the connection after self.response_class = HTTPSProxyResponse response = HTTPConnection.getresponse(self) # Restore the response class self.response_class = response_class # Close the response object manually response.close() if response.status != 200: # Close the connection manually self.close() raise xmlrpclib.ProtocolError(host, response.status, response.reason, response.msg) self.sock = SSL.SSLSocket(self.sock, self.trusted_certs) self.sock.init_ssl(self.host) def putrequest(self, method, url, skip_host=0): return HTTPConnection.putrequest(self, method, url, skip_host=skip_host) def _add_proxy_headers(self): HTTPProxyConnection._add_proxy_headers(self) # Add a User-Agent header self.putheader("User-Agent", self._user_agent) def idn_puny_to_unicode(hostname): """ Convert Internationalized domain name from Punycode (RFC3492) to Unicode """ if hostname is None: return None else: hostname = bstr(hostname) return hostname.decode('idna') def idn_ascii_to_puny(hostname): """ Convert domain name to Punycode (RFC3492). Hostname can be instance of string or Unicode """ if hostname is None: return None else: hostname = ustr(hostname) return ustr(hostname.encode('idna')) idn_pune_to_unicode = idn_puny_to_unicode idn_ascii_to_pune = idn_ascii_to_puny 0707010000000F000081B4000000000000000000000001625E86EB000003D1000000000000000000000000000000000000001100000000rhnlib/copyrightRhnlib is copyright 2010 by Red Hat Inc. Rhnlib is currently developed by Spacewalk Development Team <spacewalk-devel@redhat.com>. License: GPLv2 This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA. See /usr/share/common-licenses/GPL-2, or <http://www.gnu.org/copyleft/gpl.txt> for the terms of the latest version of the GNU General Public License. 07070100000010000081B4000000000000000000000001625E86EB0000097B000000000000000000000000000000000000001600000000rhnlib/nonblocking.py# # # import select import fcntl import os class NonBlockingFile: def __init__(self, fd): # Keep a copy of the file descriptor self.fd = fd fcntl.fcntl(self.fd.fileno(), fcntl.F_SETFL, os.O_NDELAY) # Set the callback-related stuff self.read_fd_set = [] self.write_fd_set = [] self.exc_fd_set = [] self.user_data = None self.callback = None def set_callback(self, read_fd_set, write_fd_set, exc_fd_set, user_data, callback): self.read_fd_set = read_fd_set # Make the objects non-blocking for f in self.read_fd_set: fcntl.fcntl(f.fileno(), fcntl.F_SETFL, os.O_NDELAY) self.write_fd_set = write_fd_set self.exc_fd_set = exc_fd_set self.user_data = user_data self.callback = callback def read(self, amt=0): while 1: status_changed = 0 readfds = self.read_fd_set + [self.fd] writefds = self.write_fd_set excfds = self.exc_fd_set print("Calling select", readfds) readfds, writefds, excfds = select.select(readfds, writefds, excfds) print("Select returned", readfds, writefds, excfds) if self.fd in readfds: # Our own file descriptor has changed status # Mark this, but also try to call the callback with the rest # of the file descriptors that changed status status_changed = 1 readfds.remove(self.fd) if self.callback and (readfds or writefds or excfds): self.callback(readfds, writefds, excfds, self.user_data) if status_changed: break print("Returning") return self.fd.read(amt) def write(self, data): return self.fd.write(data) def __getattr__(self, name): return getattr(self.fd, name) def callback(r, w, e, user_data): print("Callback called", r, w, e) print(r[0].read()) if __name__ == '__main__': import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect(("localhost", 5555)) f = s.makefile() ss = NonBlockingFile(f) s2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s2.connect(("localhost", 5556)) f = s2.makefile() ss.set_callback([f], [], [], None, callback) xx = ss.read() print(len(xx)) 07070100000011000081B4000000000000000000000001625E86EB00001394000000000000000000000000000000000000001000000000rhnlib/pylintrc# rhnlib package pylint configuration [MASTER] # Profiled execution. profile=no # Pickle collected data for later comparisons. persistent=no [MESSAGES CONTROL] # Disable the message(s) with the given id(s). disable=I0011, C0302, C0111, R0801, R0902, R0903, R0904, R0912, R0913, R0914, R0915, R0921, R0922, W0142, W0403, W0603, C1001, W0121, useless-else-on-loop, bad-whitespace, unpacking-non-sequence, superfluous-parens, cyclic-import, redefined-variable-type, no-else-return, # Uyuni disabled E0203, E0611, E1101, E1102 # list of disabled messages: #I0011: 62: Locally disabling R0201 #C0302: 1: Too many lines in module (2425) #C0111: 1: Missing docstring #R0902: 19:RequestedChannels: Too many instance attributes (9/7) #R0903: Too few public methods #R0904: 26:Transport: Too many public methods (22/20) #R0912:171:set_slots_from_cert: Too many branches (59/20) #R0913:101:GETServer.__init__: Too many arguments (11/10) #R0914:171:set_slots_from_cert: Too many local variables (38/20) #R0915:171:set_slots_from_cert: Too many statements (169/50) #W0142:228:MPM_Package.write: Used * or ** magic #W0403: 28: Relative import 'rhnLog', should be 'backend.common.rhnLog' #W0603: 72:initLOG: Using the global statement # for pylint-1.0 we also disable #C1001: 46, 0: Old-style class defined. (old-style-class) #W0121: 33,16: Use raise ErrorClass(args) instead of raise ErrorClass, args. (old-raise-syntax) #W:243, 8: Else clause on loop without a break statement (useless-else-on-loop) # pylint-1.1 checks #C:334, 0: No space allowed after bracket (bad-whitespace) #W:162, 8: Attempting to unpack a non-sequence defined at line 6 of (unpacking-non-sequence) #C: 37, 0: Unnecessary parens after 'not' keyword (superfluous-parens) #C:301, 0: Unnecessary parens after 'if' keyword (superfluous-parens) [REPORTS] # Set the output format. Available formats are text, parseable, colorized, msvs # (visual studio) and html output-format=parseable # Include message's id in output include-ids=yes # Tells whether to display a full report or only the messages reports=yes # Template used to display messages. This is a python new-style format string # used to format the message information. See doc for all details msg-template="{path}:{line}: [{msg_id}({symbol}), {obj}] {msg}" [VARIABLES] # A regular expression matching names used for dummy variables (i.e. not used). dummy-variables-rgx=_|dummy [BASIC] # Regular expression which should only match correct module names #module-rgx=(([a-z_][a-z0-9_]*)|([A-Z][a-zA-Z0-9]+))$ module-rgx=([a-zA-Z_][a-zA-Z0-9_]+)$ # Regular expression which should only match correct module level names const-rgx=(([a-zA-Z_][a-zA-Z0-9_]*)|(__.*__))$ # Regular expression which should only match correct class names class-rgx=[a-zA-Z_][a-zA-Z0-9_]+$ # Regular expression which should only match correct function names function-rgx=[a-z_][a-zA-Z0-9_]{,42}$ # Regular expression which should only match correct method names method-rgx=[a-z_][a-zA-Z0-9_]{,42}$ # Regular expression which should only match correct instance attribute names attr-rgx=[a-z_][a-zA-Z0-9_]{,30}$ # Regular expression which should only match correct argument names argument-rgx=[a-z_][a-zA-Z0-9_]{,30}$ # Regular expression which should only match correct variable names variable-rgx=[a-z_][a-zA-Z0-9_]{,30}$ # Regular expression which should only match correct list comprehension / # generator expression variable names inlinevar-rgx=[A-Za-z_][A-Za-z0-9_]*$ # Regular expression which should only match correct class sttribute names class-attribute-rgx=([A-Za-z_][A-Za-z0-9_]{2,42}|(__.*__))$ # Good variable names which should always be accepted, separated by a comma good-names=i,j,k,ex,Run,_ # Bad variable names which should always be refused, separated by a comma bad-names=foo,bar,baz,toto,tutu,tata # List of builtins function names that should not be used, separated by a comma bad-functions=apply,input [DESIGN] # Maximum number of arguments for function / method max-args=10 # Maximum number of locals for function / method body max-locals=20 # Maximum number of return / yield for function / method body max-returns=6 # Maximum number of branch for function / method body max-branchs=20 # Maximum number of statements in function / method body max-statements=50 # Maximum number of parents for a class (see R0901). max-parents=7 # Maximum number of attributes for a class (see R0902). max-attributes=7 # Minimum number of public methods for a class (see R0903). min-public-methods=1 # Maximum number of public methods for a class (see R0904). max-public-methods=20 [CLASSES] [FORMAT] # Maximum number of characters on a single line. max-line-length=120 # Maximum number of lines in a module max-module-lines=1000 # String used as indentation unit. This is usually " " (4 spaces) or "\t" (1 # tab). indent-string=' ' [MISCELLANEOUS] # List of note tags to take in consideration, separated by a comma. notes= 07070100000012000081B4000000000000000000000001625E86EB00000D6A000000000000000000000000000000000000001600000000rhnlib/rhnLockfile.py# # Copyright (c) 2008--2016 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # import os import sys import fcntl from errno import EWOULDBLOCK, EEXIST from rhn.stringutils import bstr import fcntl class LockfileLockedException(Exception): """thrown ONLY when pid file is locked.""" pass class Lockfile: """class that provides simple access to a PID-style lockfile. methods: __init__(lockfile), acquire(), and release() NOTE: currently acquires upon init The *.pid file will be acquired, or an LockfileLockedException is raised. """ def __init__(self, lockfile, pid=None): """create (if need be), and acquire lock on lockfile lockfile example: '/var/run/up2date.pid' """ # cleanup the path and assign it. self.lockfile = os.path.abspath( os.path.expanduser( os.path.expandvars(lockfile))) self.pid = pid if not self.pid: self.pid = os.getpid() # create the directory structure dirname = os.path.dirname(self.lockfile) if not os.path.exists(dirname): try: os.makedirs(dirname) except OSError: e = sys.exc_info()[1] if hasattr(e, 'errno') and e.errno == EEXIST: # race condition... dirname exists now. pass else: raise # open the file -- non-destructive read-write, unless it needs # to be created XXX: potential race condition upon create? self.f = os.open(self.lockfile, os.O_RDWR|os.O_CREAT|os.O_SYNC) self.acquire() def acquire(self): """acquire the lock; else raise LockfileLockedException.""" try: fcntl.flock(self.f, fcntl.LOCK_EX|fcntl.LOCK_NB) except IOError: if sys.exc_info()[1].errno == EWOULDBLOCK: raise LockfileLockedException( "cannot acquire lock on %s." % self.lockfile, None, sys.exc_info()[2]) else: raise # unlock upon exit fcntl.fcntl(self.f, fcntl.F_SETFD, 1) # truncate and write the pid os.ftruncate(self.f, 0) os.write(self.f, bstr(str(self.pid) + '\n')) def release(self): # Remove the lock file os.unlink(self.lockfile) fcntl.flock(self.f, fcntl.LOCK_UN) os.close(self.f) def main(): """test code""" try: L = Lockfile('./test.pid') except LockfileLockedException: sys.stderr.write("%s\n" % sys.exc_info()[1]) sys.exit(-1) else: print("lock acquired ") print("...sleeping for 10 seconds") import time time.sleep(10) L.release() print("lock released ") if __name__ == '__main__': # test code sys.exit(main() or 0) 07070100000013000081B4000000000000000000000001625E86EB000028A4000000000000000000000000000000000000001600000000rhnlib/rhnlib.changes------------------------------------------------------------------- Tue Apr 19 11:54:48 CEST 2022 - jgonzalez@suse.com - version 4.3.4-1 * Fix the condition for preventing building python 2 subpackage for SLE15 ------------------------------------------------------------------- Fri Mar 11 15:44:41 CET 2022 - jgonzalez@suse.com - version 4.3.3-1 * Reorganize python files ------------------------------------------------------------------- Tue Jan 18 13:36:55 CET 2022 - jgonzalez@suse.com - version 4.3.2-1 * do not build python 2 package for SLE15 ------------------------------------------------------------------- Mon Aug 09 10:56:00 CEST 2021 - jgonzalez@suse.com - version 4.3.1-1 - Bump version to 4.3.0 ------------------------------------------------------------------- Fri Apr 16 13:14:01 CEST 2021 - jgonzalez@suse.com - version 4.2.3-1 - require missing python-backports.ssl_match_hostname on SLE 11 (bsc#1183959) ------------------------------------------------------------------- Wed Feb 17 12:16:30 CET 2021 - jgonzalez@suse.com - version 4.2.2-1 - change SSL implementation to python ssl for better SAN and hostname matching support (bsc#1181807) ------------------------------------------------------------------- Fri Sep 18 11:29:26 CEST 2020 - jgonzalez@suse.com - version 4.2.1-1 - Update package version to 4.2.0 ------------------------------------------------------------------- Wed Mar 11 10:47:57 CET 2020 - jgonzalez@suse.com - version 4.1.2-1 - Fix building ------------------------------------------------------------------- Wed Nov 27 16:57:04 CET 2019 - jgonzalez@suse.com - version 4.1.1-1 - Fix malformed XML response when data contains non-ASCII chars (bsc#1154968) - Bump version to 4.1.0 (bsc#1154940) - fix initialize ssl connection (bsc#1144155) - Add SNI support for clients - Fix bootstrapping SLE11SP4 trad client with SSL enabled (bsc#1148177) ------------------------------------------------------------------- Wed May 15 15:04:15 CEST 2019 - jgonzalez@suse.com - version 4.0.8-1 - SPEC cleanup ------------------------------------------------------------------- Mon Apr 22 12:05:29 CEST 2019 - jgonzalez@suse.com - version 4.0.7-1 - Add makefile and pylint configuration for PyLint ------------------------------------------------------------------- Fri Mar 29 15:08:26 CET 2019 - jgonzalez@suse.com - version 4.0.6-1 - Add group to python*-rhnlib to fix building at SLE11 ------------------------------------------------------------------- Sat Mar 02 00:08:46 CET 2019 - jgonzalez@suse.com - version 4.0.5-1 - Read SSL decoded buffer completely when no pending bytes on the underlying connection. ------------------------------------------------------------------- Wed Feb 27 12:58:36 CET 2019 - jgonzalez@suse.com - version 4.0.4-1 - Fix encoding issues after porting to Python 3. ------------------------------------------------------------------- Thu Jan 31 09:38:49 CET 2019 - jgonzalez@suse.com - version 4.0.3-1 - Sync changes from Spacewalk - 1652859 - python3 http.client does not contain _set_hostport() ------------------------------------------------------------------- Fri Oct 26 10:02:33 CEST 2018 - jgonzalez@suse.com - version 4.0.2-1 - use rpm for debian packaging ------------------------------------------------------------------- Fri Aug 10 15:09:47 CEST 2018 - jgonzalez@suse.com - version 4.0.1-1 - Bump version to 4.0.0 (bsc#1104034) - Fix copyright for the package specfile (bsc#1103696) ------------------------------------------------------------------- Mon May 07 14:57:08 CEST 2018 - jgonzalez@suse.com - version 2.8.11.1-1 - Sync with upstream (bsc#1083294) ------------------------------------------------------------------- Mon Mar 26 08:38:18 CEST 2018 - jgonzalez@suse.com - version 2.8.10.1-1 - Sync with upstream (bsc#1083294) ------------------------------------------------------------------- Mon Mar 05 08:40:18 CET 2018 - jgonzalez@suse.com - version 2.8.7.2-1 - remove empty clean section from spec (bsc#1083294) ------------------------------------------------------------------- Fri Feb 23 10:47:01 CET 2018 - jgonzalez@suse.com - version 2.8.7.1-1 - Sync with upstream ------------------------------------------------------------------- Fri Feb 02 11:56:38 CET 2018 - jgonzalez@suse.com - version 2.8.5.3-1 - replace netstat with ss command ------------------------------------------------------------------- Wed Jan 17 11:13:43 CET 2018 - jgonzalez@suse.com - version 2.8.5.2-1 - Fix update mechanism when updating the updateservice (bsc#1073619) ------------------------------------------------------------------- Fri Nov 10 16:25:29 CET 2017 - mc@suse.de - version 2.8.5.1-1 - merge back upstream changes ------------------------------------------------------------------- Thu Oct 26 16:44:28 CEST 2017 - mc@suse.de - version 2.8.4.1-1 - build python3 package - check a state of handshake before shutdown SSL connection - Python's OpenSSL.SSL.Connection method for getting state was renamed. ------------------------------------------------------------------- Wed May 03 16:38:11 CEST 2017 - michele.bologna@suse.com - version 2.7.2.1-1 ------------------------------------------------------------------- Fri Mar 31 09:33:33 CEST 2017 - mc@suse.de - version 2.7.1.3-1 - support all TLS versions in rpclib (bsc#1025312) ------------------------------------------------------------------- Mon Mar 06 16:19:18 CET 2017 - mc@suse.de - version 2.7.1.2-1 - Updated links to github in spec files ------------------------------------------------------------------- Wed Jan 11 15:31:19 CET 2017 - michele.bologna@suse.com - version 2.7.1.1-1 - Version 2.7.1-1 ------------------------------------------------------------------- Thu Oct 06 15:08:47 CEST 2016 - mc@suse.de - version 2.5.84.3-1 - Add function aliases for backward compatibility (bsc#998185) ------------------------------------------------------------------- Mon Mar 21 16:41:39 CET 2016 - mc@suse.de - version 2.5.84.2-1 - use TLSv1_METHOD in SSL Context (bsc#970989) ------------------------------------------------------------------- Wed Mar 02 11:13:30 CET 2016 - mc@suse.de - version 2.5.84.1-1 - just set one required attribute to be compatible with all xmlrpclib versions - fixed SyntaxError: invalid syntax ' .. = b ' to work in python 2.4 ------------------------------------------------------------------- Tue Jan 26 13:53:44 CET 2016 - mc@suse.de - version 2.5.83.1-1 - change encoding for python 2 - os.write wants buffer - xmlrpc can't convert map objects - dict.iteritems() is not in python3 - perform decoding on byte string - modified exception raising to work in python2/3 - fixed string module calls to work in python3 ------------------------------------------------------------------- Sat Jan 16 11:31:43 CET 2016 - mc@suse.de - version 2.5.81.1-1 - make rhnlib compatible with python 2 and 3 ------------------------------------------------------------------- Mon Nov 30 10:48:24 CET 2015 - mc@suse.de - version 2.5.77.2-1 - python 2.7 httplib changed function _set_hostport to _get_hostport ------------------------------------------------------------------- Wed Oct 07 15:09:31 CEST 2015 - mc@suse.de - version 2.5.77.1-1 - python 3 compatibility - SmartIO: don't use tmpDir configuration from /etc/sysconfig/rhn/up2date ------------------------------------------------------------------- Tue Mar 31 14:30:45 CEST 2015 - mc@suse.de - version 2.5.69.7-1 - Documentation changes - fix name and refer to RFC. ------------------------------------------------------------------- Tue Feb 03 12:42:14 CET 2015 - mc@suse.de - version 2.5.69.6-1 - Getting rid of Tabs and trailing spaces ------------------------------------------------------------------- Thu Dec 18 13:50:31 CET 2014 - mc@suse.de - version 2.5.69.5-1 ------------------------------------------------------------------- Wed Sep 3 01:41:38 CEST 2014 - ro@suse.de - sanitize release line in specfile ------------------------------------------------------------------- Tue Jun 17 11:02:20 CEST 2014 - jrenner@suse.de - version 2.5.69.4-1 - Ensure bytes strings are sent to pyOpenSSL (bnc#880388) ------------------------------------------------------------------- Thu Apr 10 17:30:43 CEST 2014 - mc@suse.de - version 2.5.69.3-1 ------------------------------------------------------------------- Thu Apr 10 17:30:28 CEST 2014 - mc@suse.de - fix release in specfile for SLE12 (bnc#872970) ------------------------------------------------------------------- Mon Jan 20 15:16:29 CET 2014 - mc@suse.de - version 2.5.69.2-1 - change license to SPDX format ------------------------------------------------------------------- Mon Dec 09 16:05:46 CET 2013 - mc@suse.de - version 2.5.69.1-1 - switch to 2.1 ------------------------------------------------------------------- Thu Nov 28 16:27:53 CET 2013 - mc@suse.de - version 2.5.51.6-1 - fixed UserDictCase behaviour when key is not found ------------------------------------------------------------------- Wed Aug 21 15:37:30 CEST 2013 - mc@suse.de - version 2.5.51.5-1 - timeout fixes ------------------------------------------------------------------- Mon Apr 22 15:44:27 CEST 2013 - mc@suse.de - version 2.5.51.4-1 - Make timeout of yum-rhn-plugin calls through rhn-client-tools configurable (bnc#815460) - make Proxy timeouts configurable (bnc#815460) ------------------------------------------------------------------- Thu Jun 21 11:34:34 CEST 2012 - jrenner@suse.de - version 2.5.51.3-1 - allow linking against openssl ------------------------------------------------------------------- Fri Apr 27 16:56:23 CEST 2012 - mc@suse.de - version 2.5.51.2-1 - fixing typos ------------------------------------------------------------------- Wed Mar 21 17:06:43 CET 2012 - mc@suse.de - version 2.5.51.1-1 - Bumping package version ------------------------------------------------------------------- Mon Sep 19 15:36:29 CEST 2011 - mc@suse.de - require python-openssl on SUSE ------------------------------------------------------------------- Sat Jan 22 12:39:39 CET 2011 - mc@suse.de - fix macros ------------------------------------------------------------------- Tue Sep 14 17:56:57 CEST 2010 - mantel@suse.de - Initial release of rhnlib ------------------------------------------------------------------- 07070100000014000081B4000000000000000000000001625E86EB000000C6000000000000000000000000000000000000001600000000rhnlib/rhnlib.pkginfoPKG="RHATrhnl" NAME="Red Hat update agent" VERSION="1.0" ARCH="sparc" CLASSES="none" CATEGORY="utility" VENDOR="RHAT" EMAIL="alikins@redhat.com" ISTATES="S s 1 2 3" RSTATES="S s 1 2 3" BASEDIR="/" 07070100000015000081B4000000000000000000000001625E86EB0000176C000000000000000000000000000000000000001300000000rhnlib/rhnlib.spec# # spec file for package rhnlib # # Copyright (c) 2021 SUSE LLC # Copyright (c) 2008-2018 Red Hat, Inc. # # All modifications and additions to the file contributed by third parties # remain the property of their copyright owners, unless otherwise agreed # upon. The license for this file, and modifications and additions to the # file, is the same license as for the pristine package itself (unless the # license for the pristine package is not an Open Source License, in which # case the license is the MIT License). An "Open Source License" is a # license that conforms to the Open Source Definition (Version 1.9) # published by the Open Source Initiative. # Please submit bugfixes or comments via https://bugs.opensuse.org/ # %if 0%{?fedora} || 0%{?suse_version} > 1320 || 0%{?rhel} >= 8 %global build_py3 1 %{!?__python3:%global __python3 /usr/bin/python3} %if %{undefined python3_sitelib} %global python3_sitelib %(%{__python3} -c "from distutils.sysconfig import get_python_lib; print(get_python_lib())") %endif %endif %if !(0%{?rhel} >= 8 || 0%{?sle_version} >= 150000 ) %global build_py2 1 %{!?__python2:%global __python2 /usr/bin/python2} %if %{undefined python2_sitelib} %global python2_sitelib %(%{__python2} -c "from distutils.sysconfig import get_python_lib; print(get_python_lib())") %endif %endif %if "%{_vendor}" == "debbuild" # For making sure we can set the right args for deb distros %global is_deb 1 %endif Summary: Python libraries for the Spacewalk project License: GPL-2.0-only Name: rhnlib Version: 4.3.4 Release: 1 %if "%{_vendor}" == "debbuild" Group: python Packager: Uyuni Project <uyuni-devel@opensuse.org> %else Group: Development/Libraries %endif URL: https://github.com/uyuni-project/uyuni Source0: %{name}-%{version}.tar.gz BuildRoot: %{_tmppath}/%{name}-%{version}-build %if 0%{?fedora} || 0%{?rhel} || 0%{?suse_version} >= 1210 BuildArch: noarch %endif %description rhnlib is a collection of python modules used by the Spacewalk (http://spacewalk.redhat.com) software. %if 0%{?build_py2} %package -n python2-rhnlib Summary: Python libraries for the Spacewalk project Group: Development/Libraries %if "%{_vendor}" != "debbuild" %if 0%{?fedora} >= 28 || 0%{?rhel} >= 8 BuildRequires: python2-devel Requires: python2-pyOpenSSL %else BuildRequires: python-devel %if 0%{?suse_version} %if 0%{?suse_version} > 1200 Requires: python-pyOpenSSL %else Requires: python-backports.ssl_match_hostname Requires: python-openssl %endif # 0{?suse_version} > 1200 %else Requires: pyOpenSSL %endif # 0{?suse_version} %endif # 0{?fedora} >= 28 || 0{?rhel} >= 8 %endif # {_vendor} != "debbuild" %if "%{_vendor}" == "debbuild" BuildRequires: python-dev BuildRequires: rpm Requires(preun): python-minimal Requires(post): python-minimal Requires: python-openssl Obsoletes: python-rhn Conflicts: python-rhn %endif Conflicts: rhncfg < 5.10.45 Conflicts: spacewalk-proxy-installer < 1.3.2 Conflicts: rhn-client-tools < 1.3.3 Conflicts: rhn-custom-info < 5.4.7 Conflicts: rhnpush < 5.5.10 Conflicts: rhnclient < 0.10 Conflicts: spacewalk-proxy < 1.3.6 Provides: rhnlib = %{version}-%{release} Obsoletes: rhnlib < %{version}-%{release} %description -n python2-rhnlib rhnlib is a collection of python modules used by the Spacewalk software. %endif # 0%{?build_py2} %if 0%{?build_py3} %package -n python3-rhnlib Summary: Python libraries for the Spacewalk project Group: python %if "%{_vendor}" != "debbuild" BuildRequires: python3-devel %if 0%{?suse_version} BuildRequires: python-rpm-macros %endif %endif Requires: python3-pyOpenSSL %if "%{_vendor}" == "debbuild" BuildRequires: python3-dev BuildRequires: rpm Requires(preun): python3-minimal Requires(post): python3-minimal Requires: python3-openssl %endif Conflicts: rhncfg < 5.10.45 Conflicts: spacewalk-proxy-installer < 1.3.2 Conflicts: rhn-client-tools < 1.3.3 Conflicts: rhn-custom-info < 5.4.7 Conflicts: rhnpush < 5.5.10 Conflicts: rhnclient < 0.10 Conflicts: spacewalk-proxy < 1.3.6 %description -n python3-rhnlib rhnlib is a collection of python modules used by the Spacewalk software. %endif # 0%{?build_py2} %prep %setup -q # Recreate the rhn module mkdir rhn pushd rhn for pyfile in $(ls ../*.py) do ln -s $pyfile done popd if [ ! -e setup.py ]; then sed -e 's/@VERSION@/%{version}/' -e 's/@NAME@/%{name}/' setup.py.in > setup.py fi if [ ! -e setup.cfg ]; then sed 's/@RELEASE@/%{release}/' setup.cfg.in > setup.cfg fi %build %if 0%{?build_py2} make -f Makefile.rhnlib PYTHON=%{__python2} %endif %if 0%{?build_py3} make -f Makefile.rhnlib PYTHON=%{__python3} %endif %install %if 0%{?build_py2} %{__python2} setup.py install %{!?is_deb:-O1}%{?is_deb:--no-compile -O0} --skip-build --root $RPM_BUILD_ROOT %{?is_deb:--install-layout=deb} --prefix=%{_prefix} %endif %if 0%{?build_py3} %{__python3} setup.py install %{!?is_deb:-O1}%{?is_deb:--no-compile -O0} --skip-build --root $RPM_BUILD_ROOT %{?is_deb:--install-layout=deb} --prefix=%{_prefix} %endif %if 0%{?build_py2} %files -n python2-rhnlib %defattr(-,root,root) %doc ChangeLog COPYING README TODO %{python2_sitelib}/* %endif %if 0%{?build_py3} %files -n python3-rhnlib %doc ChangeLog COPYING README TODO %{python3_sitelib}/* %endif %if "%{_vendor}" == "debbuild" %post -n python2-rhnlib # Do late-stage bytecompilation, per debian policy pycompile -p python2-rhnlib -V -3.0 %preun -n python2-rhnlib # Ensure all *.py[co] files are deleted, per debian policy pyclean -p python2-rhnlib %if 0%{?build_py3} %post -n python3-rhnlib # Do late-stage bytecompilation, per debian policy py3compile -p python3-rhnlib -V -4.0 %preun -n python3-rhnlib # Ensure all *.py[co] files are deleted, per debian policy py3clean -p python3-rhnlib %endif %endif %changelog 07070100000016000081B4000000000000000000000001625E86EB00005E76000000000000000000000000000000000000001100000000rhnlib/rpclib.py# # This module contains all the RPC-related functions the RHN code uses # # Copyright (c) 2005--2018 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # __version__ = "$Revision$" import socket import re import sys from rhn import transports from rhn.stringutils import sstr from rhn.UserDictCase import UserDictCase try: # python2 import xmlrpclib from types import ListType, TupleType, StringType, UnicodeType, DictType, DictionaryType from urllib import splittype, splithost except ImportError: # python3 import xmlrpc.client as xmlrpclib ListType = list TupleType = tuple StringType = bytes UnicodeType = str DictType = dict DictionaryType = dict from urllib.parse import splittype, splithost # Redirection handling MAX_REDIRECTIONS = 5 def check_ipv6(n): """ Returns true if n is IPv6 address, false otherwise. """ try: socket.inet_pton(socket.AF_INET6, n) return True except: return False def split_host(hoststring): """ Function used to split host information in an URL per RFC 2396 handle full hostname like user:passwd@host:port """ l = hoststring.split('@', 1) host = None port = None user = None passwd = None if len(l) == 2: hostport = l[1] # userinfo present userinfo = l[0].split(':', 1) user = userinfo[0] if len(userinfo) == 2: passwd = userinfo[1] else: hostport = l[0] # Now parse hostport if hostport[0] == '[': # IPv6 with port host, port = re.split('(?<=\]):', hostport, 1) host = host.lstrip('[').rstrip(']') elif check_ipv6(hostport): # just IPv6 host = hostport else: # IPv4 arr = hostport.split(':', 1) host = arr[0] if len(arr) == 2: port = arr[1] return (host, port, user, passwd) def get_proxy_info(proxy): if proxy == None: raise ValueError("Host string cannot be null") arr = proxy.split('://', 1) if len(arr) == 2: # scheme found, strip it proxy = arr[1] return split_host(proxy) class MalformedURIError(IOError): pass # Originaly taken from xmlrpclib.ServerProxy, now changed most of the code class Server: """uri [,options] -> a logical connection to an XML-RPC server uri is the connection point on the server, given as scheme://host/target. The standard implementation always supports the "http" scheme. If SSL socket support is available (Python 2.0), it also supports "https". If the target part and the slash preceding it are both omitted, "/RPC2" is assumed. The following options can be given as keyword arguments: transport: a transport factory encoding: the request encoding (default is UTF-8) verbose: verbosity level proxy: use an HTTP proxy username: username for authenticated HTTP proxy password: password for authenticated HTTP proxy All 8-bit strings passed to the server proxy are assumed to use the given encoding. """ # Default factories _transport_class = transports.Transport _transport_class_https = transports.SafeTransport _transport_class_proxy = transports.ProxyTransport _transport_class_https_proxy = transports.SafeProxyTransport def __init__(self, uri, transport=None, encoding=None, verbose=0, proxy=None, username=None, password=None, refreshCallback=None, progressCallback=None, timeout=None): # establish a "logical" server connection # # First parse the proxy information if available # if proxy != None: (ph, pp, pu, pw) = get_proxy_info(proxy) if pp is not None: proxy = "%s:%s" % (ph, pp) else: proxy = ph # username and password will override whatever was passed in the # URL if pu is not None and username is None: username = pu if pw is not None and password is None: password = pw self._uri = sstr(uri) self._refreshCallback = None self._progressCallback = None self._bufferSize = None self._proxy = proxy self._username = username self._password = password self._timeout = timeout if len(__version__.split()) > 1: self.rpc_version = __version__.split()[1] else: self.rpc_version = __version__ self._reset_host_handler_and_type() if transport is None: self._allow_redirect = 1 transport = self.default_transport(self._type, proxy, username, password, timeout) else: # # dont allow redirect on unknow transports, that should be # set up independantly # self._allow_redirect = 0 self._redirected = None self.use_handler_path = 1 self._transport = transport self._trusted_cert_files = [] self._lang = None self._encoding = encoding self._verbose = verbose self.set_refresh_callback(refreshCallback) self.set_progress_callback(progressCallback) # referer, which redirect us to new handler self.send_handler=None self._headers = UserDictCase() def default_transport(self, type, proxy=None, username=None, password=None, timeout=None): if proxy: if type == 'https': transport = self._transport_class_https_proxy(proxy, proxyUsername=username, proxyPassword=password, timeout=timeout) else: transport = self._transport_class_proxy(proxy, proxyUsername=username, proxyPassword=password, timeout=timeout) else: if type == 'https': transport = self._transport_class_https(timeout=timeout) else: transport = self._transport_class(timeout=timeout) return transport def allow_redirect(self, allow): self._allow_redirect = allow def redirected(self): if not self._allow_redirect: return None return self._redirected def set_refresh_callback(self, refreshCallback): self._refreshCallback = refreshCallback self._transport.set_refresh_callback(refreshCallback) def set_buffer_size(self, bufferSize): self._bufferSize = bufferSize self._transport.set_buffer_size(bufferSize) def set_progress_callback(self, progressCallback, bufferSize=16384): self._progressCallback = progressCallback self._transport.set_progress_callback(progressCallback, bufferSize) def _req_body(self, params, methodname): return xmlrpclib.dumps(params, methodname, encoding=self._encoding) def get_response_headers(self): if self._transport: return self._transport.headers_in return None def get_response_status(self): if self._transport: return self._transport.response_status return None def get_response_reason(self): if self._transport: return self._transport.response_reason return None def get_content_range(self): """Returns a dictionary with three values: length: the total length of the entity-body (can be None) first_byte_pos: the position of the first byte (zero based) last_byte_pos: the position of the last byte (zero based) The range is inclusive; that is, a response 8-9/102 means two bytes """ headers = self.get_response_headers() if not headers: return None content_range = headers.get('Content-Range') if not content_range: return None arr = list(filter(None, content_range.split())) assert arr[0] == "bytes" assert len(arr) == 2 arr = arr[1].split('/') assert len(arr) == 2 brange, total_len = arr if total_len == '*': # Per RFC, the server is allowed to use * if the length of the # entity-body is unknown or difficult to determine total_len = None else: total_len = int(total_len) start, end = brange.split('-') result = { 'length' : total_len, 'first_byte_pos' : int(start), 'last_byte_pos' : int(end), } return result def accept_ranges(self): headers = self.get_response_headers() if not headers: return None if 'Accept-Ranges' in headers: return headers['Accept-Ranges'] return None def _reset_host_handler_and_type(self): """ Reset the attributes: self._host, self._handler, self._type according the value of self._uri. """ # get the url type, uri = splittype(self._uri) if type is None: raise MalformedURIError("missing protocol in uri") # with a real uri passed in, uri will now contain "//hostname..." so we # need at least 3 chars for it to maybe be ok... if len(uri) < 3 or uri[0:2] != "//": raise MalformedURIError self._type = type.lower() if self._type not in ("http", "https"): raise IOError("unsupported XML-RPC protocol") self._host, self._handler = splithost(uri) if not self._handler: self._handler = "/RPC2" def _strip_characters(self, *args): """ Strip characters, which are not allowed according: http://www.w3.org/TR/2006/REC-xml-20060816/#charsets From spec: Char ::= #x9 | #xA | #xD | [#x20-#xD7FF] | [#xE000-#xFFFD] | [#x10000-#x10FFFF] /* any Unicode character, excluding the surrogate blocks, FFFE, and FFFF. */ """ regexp = r'[\x00-\x09]|[\x0b-\x0c]|[\x0e-\x1f]' result=[] for item in args: item_type = type(item) if item_type == StringType or item_type == UnicodeType: item = re.sub(regexp, '', sstr(item)) elif item_type == TupleType: item = tuple(self._strip_characters(i) for i in item) elif item_type == ListType: item = [self._strip_characters(i) for i in item] elif item_type == DictType or item_type == DictionaryType: item = dict([(self._strip_characters(name, val)) for name, val in item.items()]) # else: some object - should take care of himself # numbers - are safe result.append(item) if len(result) == 1: return result[0] else: return tuple(result) def _request(self, methodname, params): """ Call a method on the remote server we can handle redirections. """ # the loop is used to handle redirections redirect_response = 0 retry = 0 self._reset_host_handler_and_type() while 1: if retry >= MAX_REDIRECTIONS: raise InvalidRedirectionError( "Unable to fetch requested Package") # Clear the transport headers first self._transport.clear_headers() for k, v in self._headers.items(): self._transport.set_header(k, v) self._transport.add_header("X-Info", 'RPC Processor (C) Red Hat, Inc (version %s)' % self.rpc_version) # identify the capability set of this client to the server self._transport.set_header("X-Client-Version", 1) if self._allow_redirect: # Advertise that we follow redirects #changing the version from 1 to 2 to support backward compatibility self._transport.add_header("X-RHN-Transport-Capability", "follow-redirects=3") if redirect_response: self._transport.add_header('X-RHN-Redirect', '0') if self.send_handler: self._transport.add_header('X-RHN-Path', self.send_handler) request = self._req_body(self._strip_characters(params), methodname) try: response = self._transport.request(self._host, \ self._handler, request, verbose=self._verbose) save_response = self._transport.response_status except xmlrpclib.ProtocolError: if self.use_handler_path: raise else: save_response = sys.exc_info()[1].errcode self._redirected = None retry += 1 if save_response == 200: # exit redirects loop and return response break elif save_response not in (301, 302): # Retry pkg fetch self.use_handler_path = 1 continue # rest of loop is run only if we are redirected (301, 302) self._redirected = self._transport.redirected() self.use_handler_path = 0 redirect_response = 1 if not self._allow_redirect: raise InvalidRedirectionError("Redirects not allowed") if self._verbose: print("%s redirected to %s" % (self._uri, self._redirected)) typ, uri = splittype(self._redirected) if typ != None: typ = typ.lower() if typ not in ("http", "https"): raise InvalidRedirectionError( "Redirected to unsupported protocol %s" % typ) # # We forbid HTTPS -> HTTP for security reasons # Note that HTTP -> HTTPS -> HTTP is allowed (because we compare # the protocol for the redirect with the original one) # if self._type == "https" and typ == "http": raise InvalidRedirectionError( "HTTPS redirected to HTTP is not supported") self._host, self._handler = splithost(uri) if not self._handler: self._handler = "/RPC2" # Create a new transport for the redirected service and # set up the parameters on the new transport del self._transport self._transport = self.default_transport(typ, self._proxy, self._username, self._password, self._timeout) self.set_progress_callback(self._progressCallback) self.set_refresh_callback(self._refreshCallback) self.set_buffer_size(self._bufferSize) self.setlang(self._lang) if self._trusted_cert_files != [] and \ hasattr(self._transport, "add_trusted_cert"): for certfile in self._trusted_cert_files: self._transport.add_trusted_cert(certfile) # Then restart the loop to try the new entry point. if isinstance(response, transports.File): # Just return the file return response # an XML-RPC encoded data structure if isinstance(response, TupleType) and len(response) == 1: response = response[0] return response def __repr__(self): return ( "<%s for %s%s>" % (self.__class__.__name__, self._host, self._handler) ) __str__ = __repr__ def __getattr__(self, name): # magic method dispatcher return _Method(self._request, name) # note: to call a remote object with an non-standard name, use # result getattr(server, "strange-python-name")(args) def set_transport_flags(self, transfer=0, encoding=0, **kwargs): if not self._transport: # Nothing to do return kwargs.update({ 'transfer' : transfer, 'encoding' : encoding, }) self._transport.set_transport_flags(**kwargs) def get_transport_flags(self): if not self._transport: # Nothing to do return {} return self._transport.get_transport_flags() def reset_transport_flags(self): # Does nothing pass # Allow user-defined additional headers. def set_header(self, name, arg): if type(arg) in [ type([]), type(()) ]: # Multivalued header self._headers[name] = [str(a) for a in arg] else: self._headers[name] = str(arg) def add_header(self, name, arg): if name in self._headers: vlist = self._headers[name] if not isinstance(vlist, ListType): vlist = [ vlist ] else: vlist = self._headers[name] = [] vlist.append(str(arg)) # Sets the i18n options def setlang(self, lang): self._lang = lang if self._transport and hasattr(self._transport, "setlang"): self._transport.setlang(lang) # Sets the CA chain to be used def use_CA_chain(self, ca_chain = None): raise NotImplementedError("This method is deprecated") def add_trusted_cert(self, certfile): self._trusted_cert_files.append(certfile) if self._transport and hasattr(self._transport, "add_trusted_cert"): self._transport.add_trusted_cert(certfile) def close(self): if self._transport: self._transport.close() self._transport = None # RHN GET server class GETServer(Server): def __init__(self, uri, transport=None, proxy=None, username=None, password=None, client_version=2, headers={}, refreshCallback=None, progressCallback=None, timeout=None): Server.__init__(self, uri, proxy=proxy, username=username, password=password, transport=transport, refreshCallback=refreshCallback, progressCallback=progressCallback, timeout=timeout) self._client_version = client_version self._headers = headers # Back up the original handler, since we mangle it self._orig_handler = self._handler # Download resumption self.set_range(offset=None, amount=None) def _req_body(self, params, methodname): if not params or len(params) < 1: raise Exception("Required parameter channel not found") # Strip the multiple / from the handler h_comps = list(filter(lambda x: x != '', self._orig_handler.split('/'))) # Set the handler we are going to request hndl = h_comps + ["$RHN", params[0], methodname] + list(params[1:]) self._handler = '/' + '/'.join(hndl) #save the constructed handler in case of redirect self.send_handler = self._handler # Add headers #override the handler to replace /XMLRPC with pkg path if self._redirected and not self.use_handler_path: self._handler = self._new_req_body() for h, v in self._headers.items(): self._transport.set_header(h, v) if self._offset is not None: if self._offset >= 0: brange = str(self._offset) + '-' if self._amount is not None: brange = brange + str(self._offset + self._amount - 1) else: # The last bytes # amount is ignored in this case brange = '-' + str(-self._offset) self._transport.set_header('Range', "bytes=" + brange) # Flag that we allow for partial content self._transport.set_transport_flags(allow_partial_content=1) # GET requests have empty body return "" def _new_req_body(self): type, tmpuri = splittype(self._redirected) site, handler = splithost(tmpuri) return handler def set_range(self, offset=None, amount=None): if offset is not None: try: offset = int(offset) except ValueError: # Error raise RangeError("Invalid value `%s' for offset" % offset, None, sys.exc_info()[2]) if amount is not None: try: amount = int(amount) except ValueError: # Error raise RangeError("Invalid value `%s' for amount" % amount, None, sys.exc_info()[2]) if amount <= 0: raise RangeError("Invalid value `%s' for amount" % amount) self._amount = amount self._offset = offset def reset_transport_flags(self): self._transport.set_transport_flags(allow_partial_content=0) def __getattr__(self, name): # magic method dispatcher return SlicingMethod(self._request, name) def default_transport(self, type, proxy=None, username=None, password=None, timeout=None): ret = Server.default_transport(self, type, proxy=proxy, username=username, password=password, timeout=timeout) ret.set_method("GET") return ret class RangeError(Exception): pass class InvalidRedirectionError(Exception): pass def getHeaderValues(headers, name): import mimetools if not isinstance(headers, mimetools.Message): if name in headers: return [headers[name]] return [] return [x.split(':', 1)[1].strip() for x in headers.getallmatchingheaders(name)] class _Method: """ some magic to bind an XML-RPC method to an RPC server. supports "nested" methods (e.g. examples.getStateName) """ def __init__(self, send, name): self._send = send self._name = name def __getattr__(self, name): return _Method(self._send, "%s.%s" % (self._name, name)) def __call__(self, *args): return self._send(self._name, args) def __repr__(self): return ( "<%s %s (%s)>" % (self.__class__.__name__, self._name, self._send) ) __str__ = __repr__ class SlicingMethod(_Method): """ A "slicing method" allows for byte range requests """ def __init__(self, send, name): _Method.__init__(self, send, name) self._offset = None def __getattr__(self, name): return SlicingMethod(self._send, "%s.%s" % (self._name, name)) def __call__(self, *args, **kwargs): self._offset = kwargs.get('offset') self._amount = kwargs.get('amount') # im_self is a pointer to self, so we can modify the class underneath try: self._send.im_self.set_range(offset=self._offset, amount=self._amount) except AttributeError: pass result = self._send(self._name, args) # Reset "sticky" transport flags try: self._send.im_self.reset_transport_flags() except AttributeError: pass return result def reportError(headers): """ Reports the error from the headers. """ errcode = 0 errmsg = "" s = "X-RHN-Fault-Code" if s in headers: errcode = int(headers[s]) s = "X-RHN-Fault-String" if s in headers: _sList = getHeaderValues(headers, s) if _sList: _s = ''.join(_sList) import base64 errmsg = "%s" % base64.decodestring(_s) return errcode, errmsg 07070100000017000081B4000000000000000000000001625E86EB00000101000000000000000000000000000000000000001400000000rhnlib/setup.cfg.in# Configuration file for distutils [bdist_rpm] release = @RELEASE@ packager = Red Hat, Inc. <http://bugzilla.redhat.com/bugzilla> requires = pyOpenSSL python doc-files = ChangeLog COPYING README TODO changelog = changelog.spec build-requires = python-devel 07070100000018000081B4000000000000000000000001625E86EB000001E1000000000000000000000000000000000000001300000000rhnlib/setup.py.in#!/usr/bin/python # # from distutils.core import setup setup(name = "@NAME@", version = "@VERSION@", description = "Python libraries for the Spacewalk project", long_description = """\ rhnlib is a collection of python modules used by the Spacewalk (http://spacewalk.redhat.com) software.""", author = 'Mihai Ibanescu', author_email = 'misa@redhat.com', url = 'http://rhn.redhat.com', packages = ["rhn"], license = "GPL", ) 07070100000019000081B4000000000000000000000001625E86EB0000070C000000000000000000000000000000000000001600000000rhnlib/stringutils.py# # This module contains all the RPC-related functions the RHN code uses # # Copyright (c) 2016 Red Hat, Inc. # Copyright (c) 2022 SUSE, LLC # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # from sys import version_info try: PY3 = version_info.major >= 3 except AttributeError: PY3 = False def ustr(obj): # converts object to unicode like object if PY3: # python3 if isinstance(obj, str): return obj else: return str(obj, 'utf8', errors='ignore') else: # python2 if isinstance(obj, unicode): return obj return unicode(obj, 'utf8', 'ignore') def bstr(obj): # converts object to bytes like object if PY3: # python3 if isinstance(obj, bytes): return obj else: return bytes(obj, 'utf8', errors='ignore') else: # python2 if isinstance(obj, str): return obj return str(obj.encode('utf8', 'ignore')) def sstr(obj): # converts object to string if PY3: # python3 if isinstance(obj, str): return obj else: return str(obj, 'utf8', errors='ignore') else: # python2 if isinstance(obj, str): return obj return str(obj.encode('utf8', 'ignore')) 0707010000001A000081B4000000000000000000000001625E86EB00000344000000000000000000000000000000000000000D00000000rhnlib/tb.py# # Copyright (c) 2016 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # import sys try: PY3 = sys.version_info.major >= 3 except AttributeError: PY3 = False if PY3: def raise_with_tb(e): raise e else: exec(""" def raise_with_tb(e): raise e, None, sys.exc_info()[2] """) 0707010000001B000041FD000000000000000000000001625E86EB00000000000000000000000000000000000000000000000C00000000rhnlib/test0707010000001C000081FD000000000000000000000001625E86EB00000588000000000000000000000000000000000000001D00000000rhnlib/test/00-connection.py#!/usr/bin/python # # # # # Usage: $0 SERVER PROXY [SYSTEMID] import sys sys.path.append('..') from rhn.rpclib import Server SERVER = "xmlrpc.rhn.redhat.com" HANDLER = "/XMLRPC" PROXY = "proxy.example.com:8080" system_id_file = '/etc/sysconfig/rhn/systemid' if len(sys.argv) < 3: print "Non efficient cmd-line arguments! Provide at least server & proxy!" sys.exit(1); try: SERVER = sys.argv[1] PROXY = sys.argv[2] system_id_file = sys.argv[3] except: pass print "SERVER = %s" % SERVER print "PROXY = %s" % PROXY print "system_id_file = %s" % system_id_file def get_test_server_proxy_http(): global SERVER, HANDLER, PROXY return Server("http://%s%s" % (SERVER, HANDLER), proxy=PROXY) def get_test_server_proxy_https(): global SERVER, HANDLER, PROXY return Server("https://%s%s" % (SERVER, HANDLER), proxy=PROXY) def get_test_server_https(): global SERVER, HANDLER return Server("https://%s%s" % (SERVER, HANDLER)) def get_test_server_http(): global SERVER, HANDLER return Server("http://%s%s" % (SERVER, HANDLER)) if __name__ == '__main__': systemid = open(system_id_file).read() tests = [ get_test_server_http, get_test_server_https, get_test_server_proxy_http, get_test_server_proxy_https, ] for gs in tests: s = gs() print "--- %s ---" % gs print s.up2date.login(systemid) 0707010000001D000081B4000000000000000000000001625E86EB0000041F000000000000000000000000000000000000001600000000rhnlib/test/01-get.py#!/usr/bin/python # # # import sys sys.path.append('..') from rhn.rpclib import Server, GETServer SERVER = "http://xmlrpc.rhn.redhat.com/XMLRPC" system_id_file = "/etc/sysconfig/rhn/systemid" try: SERVER = "http://%s/XMLRPC" % sys.argv[1] system_id_file = sys.argv[2] except: pass print "SERVER = %s" % SERVER print "system_id_file = %s" % system_id_file s = Server(SERVER) sysid = open(system_id_file).read() dict = s.up2date.login(sysid) print dict channels = dict['X-RHN-Auth-Channels'] channel_name, channel_version = channels[0][:2] sg = GETServer(SERVER, headers=dict) l = sg.listPackages(channel_name, channel_version) print l # Package download package = l[0] print "PACKAGE TO DOWNLOAD: %s %s %s %s" % (package[0], package[1], package[2], package[4]) filename = "%s-%s-%s.%s.rpm" % (package[0], package[1], package[2], package[4]) fd = sg.getPackage(channel_name, filename) f_name = "/tmp/test-get-%s" % filename f = open(f_name, "w+") f.write(fd.read()) f.close() print "PACKAGE DOWNLOADED AS: %s" % f_name 0707010000001E000081FD000000000000000000000001625E86EB0000064D000000000000000000000000000000000000001C00000000rhnlib/test/02-callbacks.py#!/usr/bin/python # # # import sys sys.path.append('..') from rhn.rpclib import Server, GETServer SERVER = "http://xmlrpc.rhn.redhat.com/XMLRPC" system_id_file = "/etc/sysconfig/rhn/systemid" try: SERVER = "http://%s/XMLRPC" % sys.argv[1] system_id_file = sys.argv[2] except: pass print "SERVER = %s" % SERVER print "system_id_file = %s" % system_id_file def refreshCallback(*args, **kwargs): print "Called refreshCallback, args %s, kwargs %s" % (args, kwargs) def progressCallback(*args, **kwargs): print "Called progressCallback, args %s, kwargs %s" % (args, kwargs) if __name__ == '__main__': sysid = open(system_id_file).read() s = Server(SERVER) s.set_refresh_callback(refreshCallback) s.set_progress_callback(progressCallback) dict = s.up2date.login(sysid) gs = GETServer(SERVER, headers=dict) gs.set_refresh_callback(refreshCallback) gs.set_progress_callback(progressCallback, 16384) channels = dict['X-RHN-Auth-Channels'] cn, cv = channels[0][:2] print "Calling listPackages" l = gs.listPackages(cn, cv) for p in l: if p[0] == 'kernel': package = p break else: raise Exception("Package not found") print "PACKAGE TO DOWNLOAD: %s %s %s %s" % (package[0], package[1], package[2], package[4]) filename = "%s-%s-%s.%s.rpm" % (package[0], package[1], package[2], package[4]) print "Calling getPackages" fd = gs.getPackage(cn, filename) data_name = "/tmp/foobar" data = open(data_name, "w+").write(fd.read()) print "PACKAGE DOWNLOADED AS: %s" % data_name 0707010000001F000081B4000000000000000000000001625E86EB000001D7000000000000000000000000000000000000001E00000000rhnlib/test/03-nonblocking.pyimport socket import sys sys.path.append('..') from rhn.connections import HTTPConnection def callback(r, w, x, u): print "Callback called" print r[0].read() s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect(("localhost", 5555)) f = s.makefile() h = HTTPConnection("roadrunner.devel.redhat.com", 8001) h.set_callback([f], [], [], None, callback) h.putrequest("GET", "/") h.endheaders() resp = h.getresponse() print resp.status print resp.read() 07070100000020000081FD000000000000000000000001625E86EB000004FE000000000000000000000000000000000000002600000000rhnlib/test/04-authenticated-proxy.py#!/usr/bin/python # # # # # Usage: $0 SERVER PROXY:PORT [SYSTEMID] [PROXY_USER] [PROXY_PASS] import sys import httplib sys.path.append('..') from rhn.rpclib import Server SERVER = "xmlrpc.rhn.redhat.com" HANDLER = "/XMLRPC" PROXY = "proxy.example.com:8080" PROXY_USERNAME = None PROXY_PASSWORD = None system_id_file = '/etc/sysconfig/rhn/systemid' if len(sys.argv) < 3: print "Non efficient cmd-line arguments! Provide at least server & proxy!" sys.exit(1); try: SERVER = sys.argv[1]; PROXY = sys.argv[2]; system_id_file = sys.argv[3] PROXY_USERNAME = sys.argv[4]; PROXY_PASSWORD = sys.argv[5]; except: pass def get_test_server_proxy_http(): global SERVER, HANDLER, PROXY return Server("http://%s%s" % (SERVER, HANDLER), proxy=PROXY, username=PROXY_USERNAME, password=PROXY_PASSWORD) def get_test_server_proxy_https(): global SERVER, HANDLER, PROXY return Server("https://%s%s" % (SERVER, HANDLER), proxy=PROXY, username=PROXY_USERNAME, password=PROXY_PASSWORD) if __name__ == '__main__': systemid = open(system_id_file).read() tests = [ get_test_server_proxy_http, get_test_server_proxy_https, ] for gs in tests: s = gs() print s.up2date.login(systemid) 07070100000021000081B4000000000000000000000001625E86EB00000333000000000000000000000000000000000000002200000000rhnlib/test/05-delete-packages.py#!/usr/bin/python # # tests uploads over SSL # # # USAGE: $0 SERVER SYSTEMID # OUTPUT: return code = 0 import sys sys.path.append('..') from rhn.rpclib import Server SERVER = "xmlrpc.rhn.redhat.com" HANDLER = "/XMLRPC" system_id_file = '/etc/sysconfig/rhn/systemid' try: SERVER = sys.argv[1] system_id_file = sys.argv[2] except: pass def get_test_server_https(): global SERVER, HANDLER return Server("https://%s%s" % (SERVER, HANDLER)) if __name__ == '__main__': systemid = open(system_id_file).read() s = get_test_server_https() # Generate a huge list of packages to "delete" packages = [] for i in range(3000): packages.append(["package-%d" % i, '1.1', '1', '']) result = s.registration.delete_packages(systemid, packages[:1000]); sys.exit(result); 07070100000022000081B4000000000000000000000001625E86EB000008C6000000000000000000000000000000000000002100000000rhnlib/test/06-leaking-ssl-fd.py#!/usr/bin/python # # tests leaking file descriptors # # # USAGE: $0 SERVER SYSTEMID import os import sys sys.path.append('..') from rhn.rpclib import Server from threading import Thread import time SERVER = "xmlrpc.rhn.redhat.com" HANDLER = "/XMLRPC" system_id_file = '/etc/sysconfig/rhn/systemid' try: SERVER = sys.argv[1] system_id_file = sys.argv[2] except: pass """Make few attempts to login""" class makeAttempts(Thread): def __init__(self): Thread.__init__(self); self.server = self.get_test_server_https() self.term = False; def run(self): global system_id_file systemid = open(system_id_file).read() i = 0; while i < 100 and not self.term: # Make few attempts try: dict = self.server.up2date.login(systemid); except Exception, e: if (str(e) == "(4, 'Interrupted system call')"): pass else: raise e; i = i + 1; return(0); def terminate(self): self.term = True; def get_test_server_https(self): global SERVER, HANDLER return Server("https://%s%s" % (SERVER, HANDLER)) if __name__ == '__main__': print "PID:", os.getpid() attempt = makeAttempts(); attempt.start(); time.sleep(3); # Try to catch netstat, while thread is still running #port443 = "netstat -tanp |grep :443 | grep %s/python" % os.getpid() #port80 = "netstat -tanp |grep :80 | grep %s/python" % os.getpid() port443 = "ss -tanp |grep :443 | grep python | grep pid=%s" % os.getpid() port80 = "ss -tanp |grep :80 | grep python | grep pid=%s" % os.getpid() while attempt.isAlive(): res80 = os.system(port80) if (res80 == 0): # Port 80 is used ERROR attemt.terminate(); attempt.join(); print "ERROR: Port 80 is used!" sys.exit(1); res443 = os.system(port443) if (res443 == 0): # Port 443 is used ok attempt.terminate(); attempt.join(); print "OK" sys.exit(0); attempt.join(); print("ERROR: Port 443 was not used!"); sys.exit(-1); 07070100000023000081B4000000000000000000000001625E86EB00000BFC000000000000000000000000000000000000002100000000rhnlib/test/07-leaking-ssl-fd.py#!/usr/bin/python # # tests leaking file descriptors # # # USAGE: $0 SERVER SYSTEMID import os import sys sys.path.append('..') from rhn.rpclib import Server, GETServer from threading import Thread SERVER = "xmlrpc.rhn.redhat.com" HANDLER = "/XMLRPC" system_id_file = '/etc/sysconfig/rhn/systemid' try: SERVER = sys.argv[1] system_id_file = sys.argv[2] except: pass """Make few attempts""" class makeAttempts(Thread): def __init__(self): Thread.__init__(self); self.server = self.get_test_server_https() self.term = False; def run(self): global system_id_file systemid = open(system_id_file).read() dict = self.server.up2date.login(systemid) channels = dict['X-RHN-Auth-Channels'] c = channels[0] gs = self.get_test_GET_server_https(dict) lp = gs.listPackages(c[0], c[1]) package_count = len(lp) i = 0 pi = 0 while i < 1000 and not self.term: if pi == package_count: # Wrap pi = 0 p = lp[pi] pn = self.get_package_name(p) try: fd = gs.getPackageHeader(c[0], pn) except Exception, e: if (str(e) == "(4, 'Interrupted system call')"): pass continue; else: raise e; buffer = fd.read() assert len(buffer) != 0 print "Called %4d; header length: %-6d for %s" % (i, len(buffer), pn) i = i + 1 pi = pi + 1 def terminate(self): self.term = True; def get_test_server_https(self): global SERVER, HANDLER return Server("https://%s%s" % (SERVER, HANDLER)) def get_test_GET_server_https(self, headers): global SERVER, HANDLER return GETServer("https://%s%s" % (SERVER, HANDLER), headers=headers) def get_package_name(self, p): return "%s-%s-%s.%s.rpm" % (p[0], p[1], p[2], p[4]) import time; if __name__ == '__main__': print "PID:", os.getpid() attempt = makeAttempts(); attempt.start(); time.sleep(3); # Try to catch netstat, while thread is still running #port443 = "netstat -tanp |grep :443 | grep %s/python" % os.getpid() #port80 = "netstat -tanp |grep :80 | grep %s/python" % os.getpid() port443 = "ss -tanp |grep :443 | grep python | grep pid=%s" % os.getpid() port80 = "ss -tanp |grep :80 | grep python | grep pid=%s" % os.getpid() while attempt.isAlive(): res80 = os.system(port80) if (res80 == 0): # Port 80 is used ERROR attemt.terminate(); attempt.join(); print "ERROR: Port 80 is used!" sys.exit(1); res443 = os.system(port443) if (res443 == 0): # Port 443 is used ok attempt.terminate(); attempt.join(); print "OK" sys.exit(0); attempt.join(); print("ERROR: Port 443 was not used!"); sys.exit(-1); 07070100000024000081B4000000000000000000000001625E86EB00000B0B000000000000000000000000000000000000001D00000000rhnlib/test/08-leaking-fd.py#!/usr/bin/python # # tests leaking file descriptors # # # USAGE: $0 SERVER SYSTEMID import os import sys sys.path.append('..') from rhn.rpclib import Server, GETServer from threading import Thread import time SERVER = "xmlrpc.rhn.redhat.com" HANDLER = "/XMLRPC" PROXY=None system_id_file = '/etc/sysconfig/rhn/systemid' try: SERVER = sys.argv[1] system_id_file = sys.argv[2] PROXY = sys.argv[3] except: pass """Make few attempts over ssl""" class makeAttempts(Thread): def __init__(self): Thread.__init__(self); self.server = self.get_test_server_https() self.term = False; def run(self): global system_id_file systemid = open(system_id_file).read() dict = self.server.up2date.login(systemid) channels = dict['X-RHN-Auth-Channels'] c = channels[0] gs = self.get_test_GET_server_https(dict) lp = gs.listPackages(c[0], c[1]) p = lp[0] pn = "%s-%s-%s.%s.rpm" % (p[0], p[1], p[2], p[4]) print pn i = 0 while i < 100 and not self.term: # Make few attempts try: fd = gs.getPackageHeader(c[0], pn) print "Called %-4d" % i except Exception, e: if (str(e) == "(4, 'Interrupted system call')"): pass else: raise e; i = i + 1 return 0; def terminate(self): self.term = True; def get_test_server_https(self): global SERVER, HANDLER, PROXY return Server("https://%s%s" % (SERVER, HANDLER), proxy=PROXY) def get_test_GET_server_https(self, headers): global SERVER, HANDLER, PROXY return GETServer("https://%s%s" % (SERVER, HANDLER), headers=headers,proxy=PROXY) if __name__ == '__main__': print "PID:", os.getpid() attempt = makeAttempts(); attempt.start(); time.sleep(3); # Try to catch netstat, while thread is still running #port443 = "netstat -tanp |grep :443 | grep %s/python" % os.getpid() #port80 = "netstat -tanp |grep :80 | grep %s/python" % os.getpid() port443 = "ss -tanp |grep :443 | grep python | grep pid=%s" % os.getpid() port80 = "ss -tanp |grep :80 | grep python | grep pid=%s" % os.getpid() while attempt.isAlive(): res80 = os.system(port80) if (res80 == 0): # Port 80 is used ERROR attempt.terminate(); attempt.join(); print "ERROR: Port 80 is used!" sys.exit(1); res443 = os.system(port443) if (res443 == 0): # Port 443 is used ok attempt.terminate(); attempt.join(); print "OK" sys.exit(0); attempt.join(); print("ERROR: Port 443 was not used!"); sys.exit(-1); 07070100000025000081FD000000000000000000000001625E86EB00000409000000000000000000000000000000000000002000000000rhnlib/test/09-test-errcodes.py#!/usr/bin/python # # munge the headers to produce an error message # # # USAGE: $0 SERVER SYSTEMID import sys sys.path.append('..') from rhn.rpclib import Server, GETServer, ProtocolError, reportError SERVER = "xmlrpc.rhn.redhat.com" HANDLER = "/XMLRPC" system_id_file = '/etc/sysconfig/rhn/systemid' try: SERVER = sys.argv[1] system_id_file = sys.argv[2] except: pass def get_test_server_https(): global SERVER, HANDLER return Server("https://%s%s" % (SERVER, HANDLER)) s = get_test_server_https() sysid = open(system_id_file).read() dict = s.up2date.login(sysid) print dict dict['X-RHN-Auth-Server-Time'] = 1324 channels = dict['X-RHN-Auth-Channels'] channel_name, channel_version = channels[0][:2] sg = GETServer("http://xmlrpc.rhn.redhat.com/XMLRPC", headers=dict) try: l = sg.listPackages(channel_name, channel_version) except ProtocolError, e: print reportError(e.headers) print("OK (error above expected)"); sys.exit(0); print("ERROR: Exception didn't occurred!"); sys.exit(-1); 07070100000026000081FD000000000000000000000001625E86EB00000509000000000000000000000000000000000000001B00000000rhnlib/test/10-tempfile.py#!/usr/bin/python # # Test for tempfile creation # Use lsof to see how many open files we have # import sys import os import glob sys.path.append('..') from rhn.SmartIO import _tempfile def t(): f = _tempfile() for i in range(1024): f.write(("%s" % (i % 10)) * 1023 + "\n") f.seek(0, 2) assert(f.tell() == 1048576) return f def openedFiles(): global pid path = '/proc/' + pid + '/fd/'; return len(glob.glob(os.path.join(path, '*'))); if __name__ == '__main__': global pid pid = str(os.getpid()); print "PID: ", pid; failed = False; print "Running and saving stream object references" ret = [] for i in range(100): print "Saving", i ret.append(t()) if openedFiles() != i + 5: print "FAIL: Opened files: ", openedFiles(), "but expected: ", str(i + 5); failed = True; del ret print "Running without saving object references" for i in range(1000): print "Running", i t() if openedFiles() not in [4, ]: print "FAIL: Opened files: ", openedFiles(), "but expected 4!"; failed = True; if failed: print "Test FAILS!" sys.exit(1); else: print "Test PASSES!" sys.exit(0); 07070100000027000081B4000000000000000000000001625E86EB000001D1000000000000000000000000000000000000001A00000000rhnlib/test/11-SmartIO.py#!/usr/bin/python # # Test for SmartIO objects # import sys sys.path.append('..') from rhn.SmartIO import SmartIO from cStringIO import OutputType if __name__ == '__main__': s = SmartIO(max_mem_size=16384) for i in range(20): s.write(("%d" % (i % 10)) * 1023 + '\n') if i < 16: assert(isinstance(s._io, OutputType)) else: assert(not isinstance(s._io, OutputType)) print i, type(s._io), s._io.tell() 07070100000028000081B4000000000000000000000001625E86EB000003A8000000000000000000000000000000000000001F00000000rhnlib/test/12-Input-Output.py#!/usr/bin/python # # Tests the encodings in Input and Output objects # import sys sys.path.append('..') from rhn import transports try: # python2 from cStringIO import StringIO except ImportError: # python3 from io import StringIO REFERENCE = "the quick brown fox jumps over the lazy dog" * 1024 def t(transfer, encoding): print "\n---> Testing transfer=%s, encoding=%s" % (transfer, encoding) o = transports.Output(transfer=transfer, encoding=encoding) o.process(REFERENCE) print "Output: data length: %s; headers: %s" % (len(o.data), o.headers) i = transports.Input(o.headers) i.read(StringIO(o.data)) io = i.decode() io.seek(0, 0) data = io.read() assert(REFERENCE == data) if __name__ == '__main__': tests = [] for transfer in range(3): for encoding in range(3): tests.append((transfer, encoding)) for test in tests: t(test[0], test[1]) 07070100000029000081FD000000000000000000000001625E86EB000005EF000000000000000000000000000000000000001C00000000rhnlib/test/13-safe_read.py#!/usr/bin/python # # Test _smart_read over a slow socket # import sys import os import glob sys.path.append('..') from rhn.rpclib import transports import time try: # python2 from cStringIO import StringIO except ImportError: # python3 from io import StringIO class SlowSocket: def __init__(self): self._buf = StringIO() def read(self, amt=None): time.sleep(.01) return self._buf.read(amt) def __getattr__(self, name): return getattr(self._buf, name) def t(): buf = SlowSocket() for i in range(1024): buf.write(("%s" % (i % 10)) * 1023 + "\n") buf.seek(0, 2) amt = buf.tell() buf.seek(0, 0) print "Using temp file" f = transports._smart_read(buf, amt) f.seek(0, 2) print "Read", f.tell(), type(f._io) buf.seek(0, 0) print "Reading in memory..." f = transports._smart_read(buf, amt, max_mem_size=amt+1) f.seek(0, 2) print "Read", f.tell(), type(f._io) def openedFiles(): global pid path = '/proc/' + pid + '/fd/'; return len(glob.glob(os.path.join(path, '*'))); if __name__ == '__main__': global pid pid = str(os.getpid()); failed = False for i in range(100): print "Running", i t() if openedFiles() != 4: print "FAIL: Opened files = ", openedFiles(), ", but expected: 4!" failed = True if failed: print "Test FAILS!" sys.exit(1); else: print "Test PASSES!" sys.exit(0); 0707010000002A000081B4000000000000000000000001625E86EB000007E2000000000000000000000000000000000000001F00000000rhnlib/test/14-Input-Output.py#!/usr/bin/python # # Tests the encodings in Input and Output objects # # # 2004-11-29: fails, but it looks like it never worked. Need to investigate import sys sys.path.append('..') from rhn import transports import xmlrpclib try: # python2 from cStringIO import StringIO except ImportError: # python3 from io import StringIO REFERENCE_XML = { 'a' : [1, 'b', '3'], '2' : [1, {'b' : 2}], } REFERENCE_BLOB = "a1b2c3d4e5" * 100 def test_xmlrpc(transfer, encoding): print "\n---> XML Testing transfer=%s, encoding=%s" % (transfer, encoding) data = xmlrpclib.dumps((REFERENCE_XML, ), methodresponse=1) o = transports.Output(transfer=transfer, encoding=encoding) o.set_header('Content-Type', 'text/xml') o.process(data) headers = o.headers # Added by the connection layer headers['Content-Length'] = len(o.data) print "Output: headers: %s" % headers.items() i = transports.Input(headers) io = i.decode(StringIO(o.data)) assert(i.type.lower() == 'text/xml') io.seek(0, 0) data = io.read() params, dummy = xmlrpclib.loads(data) assert(REFERENCE_XML == params[0]) def test_blob(transfer, encoding): print "\n---> BLOB Testing transfer=%s, encoding=%s" % (transfer, encoding) o = transports.Output(transfer=transfer, encoding=encoding) o.set_header('Content-Type', 'application/binary') o.process(REFERENCE_BLOB) headers = o.headers # Added by the connection layer headers['Content-Length'] = len(o.data) print "Output: headers: %s" % headers.items() i = transports.Input(headers) io = i.decode(StringIO(o.data)) assert(i.type.lower() == 'application/binary') io.seek(0, 0) data = io.read() assert(REFERENCE_BLOB == data) if __name__ == '__main__': tests = [] for transfer in range(3): for encoding in range(3): tests.append((transfer, encoding)) for test in tests: test_xmlrpc(test[0], test[1]) test_blob(test[0], test[1]) 0707010000002B000081B4000000000000000000000001625E86EB000001DF000000000000000000000000000000000000001900000000rhnlib/test/15-ranges.py#!/usr/bin/python # # Tests the encodings in Input and Output objects # import sys sys.path.append('..') from rhn import rpclib gs = rpclib.GETServer("http://coyote.devel.redhat.com/DOWNLOAD") gs.set_transport_flags(allow_partial_content=1) fd = gs.a.b('a', 'b', 'c', offset=9, amount=1) #fd = gs.a.b('a', 'b', 'c') print fd.read() print gs.get_response_headers() print "Status", gs.get_response_status() print "Reason", gs.get_response_reason() print gs.get_content_range() 0707010000002C000081B4000000000000000000000001625E86EB0000045C000000000000000000000000000000000000001E00000000rhnlib/test/16-test-ranges.py#!/usr/bin/python # # Tests the encodings in Input and Output objects # import sys sys.path.append('..') from rhn import rpclib if len(sys.argv) > 1: system_id_file = sys.argv[1] else: system_id_file = '/etc/sysconfig/rhn/systemid' systemid = open(system_id_file).read() #server_url = "http://coyote.devel.redhat.com/XMLRPC" server_url = "http://xmlrpc.rhn.webdev.redhat.com/XMLRPC" s = rpclib.Server(server_url) cookie = s.up2date.login(systemid) gs = rpclib.GETServer(server_url, headers=cookie) gs.set_transport_flags(allow_partial_content=1) channel_name, channel_version = cookie['X-RHN-Auth-Channels'][0][:2] package_list = gs.listPackages(channel_name, channel_version) for p in package_list: if p[0] == 'python': break pn, pv, pr, pe, pa = p[:5] package_name = "%s-%s-%s.%s.rpm" % (pn, pv, pr, pa) fd = gs.getPackage(channel_name, package_name, offset=1023) #, amount=10) print gs.get_response_headers() print "Status", gs.get_response_status() print "Reason", gs.get_response_reason() h = gs.get_content_range() print h assert(h['first_byte_pos'] == 1023, h['first_byte_pos']) 0707010000002D000081B4000000000000000000000001625E86EB000001A4000000000000000000000000000000000000002B00000000rhnlib/test/17-test-multivalued-headers.py#!/usr/bin/python # # # import sys sys.path.append('..') from rhn.transports import Output from rhn.connections import HTTPConnection if __name__ == '__main__': conn = HTTPConnection("localhost", 5555) o = Output(connection=conn) o.set_header("X-Test-1", "a") o.set_header("X-Test-2", ["a", "b", "c"]) data = "0123456789" o.process(data) headers, fd = o.send_http("fake.example.com") 0707010000002E000081FD000000000000000000000001625E86EB0000022D000000000000000000000000000000000000001D00000000rhnlib/test/18-split-host.py#!/usr/bin/python # # # import sys sys.path.append('..') from rhn.rpclib import get_proxy_info tests = [ ["http://user:pass@host:https", ('host', 'https', 'user', 'pass')], ["ftp://user@host", ('host', None, 'user', None)], ["http://user:@host:8080", ('host', '8080', 'user', '')], ["user:pass@host", ('host', None, 'user', 'pass')], ] fail=0 for url, result in tests: r = get_proxy_info(url) if result != r: print "Test failed", url, r, result fail += 1 if (not fail): print "Test PASSES" sys.exit(fail); 0707010000002F000081FD000000000000000000000001625E86EB000007AB000000000000000000000000000000000000001C00000000rhnlib/test/19-redirects.py#!/usr/bin/python # # # import sys sys.path.append('..') from rhn.rpclib import Server, InvalidRedirectionError _host = 'xmlrpc.rhn.webdev.redhat.com' tests = [ # Syntax: url, allow_redirect, should_fail, text_message ('http://%s/XMLRPC-REDIRECT' % _host, 1, None, "HTTP->HTTPS"), ('https://%s/XMLRPC-REDIRECT' % _host, 1, None, "HTTPS->HTTPS"), ('http://%s/XMLRPC-REDIRECT-NOSSL' % _host, 1, None, "HTTP->HTTP"), ('https://%s/XMLRPC-REDIRECT-NOSSL' % _host, 1, InvalidRedirectionError, "HTTPS->HTTP"), # These should fail ('http://%s/XMLRPC-REDIRECT' % _host, 0, InvalidRedirectionError, "HTTP->HTTPS"), ('https://%s/XMLRPC-REDIRECT' % _host, 0, InvalidRedirectionError, "HTTPS->HTTPS"), ('http://%s/XMLRPC-REDIRECT-NOSSL' % _host, 0, InvalidRedirectionError, "HTTP->HTTP"), ('https://%s/XMLRPC-REDIRECT-NOSSL' % _host, 0, InvalidRedirectionError, "HTTPS->HTTP"), ] def main(): if len(sys.argv) > 1: systemid_path = sys.argv[1] else: systemid_path = "/etc/sysconfig/rhn/systemid" print "Using %s as systemid (command line to override)" % systemid_path global SYSTEM_ID SYSTEM_ID = open(systemid_path).read() ret = 0 for t in tests: ret = run_test(*t) or ret return ret def run_test(url, allow_redirect, should_fail, text_message): global SYSTEM_ID message = "Running test: %s" % text_message print message, s = Server(url) s.allow_redirect(allow_redirect) try: s.up2date.login(SYSTEM_ID) except Exception, e: if should_fail and isinstance(e, should_fail): print "PASS" return 0 print "FAIL (exception: %s)" % (e.__class__.__name__) return 1 if should_fail: print "FAIL (no exception)" return 1 print "PASS" return 0 if __name__ == '__main__': sys.exit(main() or 0) 07070100000030000081FD000000000000000000000001625E86EB0000015E000000000000000000000000000000000000002300000000rhnlib/test/20-proxy-user-agent.py#!/usr/bin/python # # # # # Usage: $0 SERVER PROXY import sys sys.path.append('..') from rhn.connections import HTTPSProxyConnection try: SERVER = sys.argv[1]; PROXY = sys.argv[2]; except: print "Non efficient cmd-line arguments! Provide at least server & proxy!" sys.exit(1); h = HTTPSProxyConnection(PROXY, SERVER) h.connect() 07070100000031000081FD000000000000000000000001625E86EB000002CC000000000000000000000000000000000000002300000000rhnlib/test/21-proxy-user-agent.py#!/usr/bin/python # # tests rhn.rpclib.Server(), connection through proxy # # # USAGE: $0 SERVER PROXY [SYSTEMID] import sys sys.path.append('..') from rhn import rpclib SERVER = "xmlrpc.rhn.redhat.com" HANDLER = "/XMLRPC" PROXY = "proxy.example.com:8080" system_id_file = '/etc/sysconfig/rhn/systemid' if len(sys.argv) < 3: print "Non efficient cmd-line arguments! Provide at least server & proxy!" sys.exit(1); try: SERVER = sys.argv[1] PROXY = sys.argv[2] system_id_file = sys.argv[3] except: pass SERVER_URL = "https://" + SERVER + HANDLER systemid = open(system_id_file).read() s = rpclib.Server(SERVER_URL, proxy = PROXY) dict = s.up2date.login(systemid); print "Test PASSES" 07070100000032000081B4000000000000000000000001625E86EB000001AD000000000000000000000000000000000000001F00000000rhnlib/test/22-base16values.py#!/usr/bin/python # # Checking if the values encoded in hex (base 16) are properly decoded. # import os import sys filename = "base16values.txt" filename = os.path.join(os.path.dirname(sys.argv[0]), filename) f = open(filename) while 1: line = f.readline() if not line: break arr = line.split(" ", 1) if len(arr) != 2: break i = int(arr[0]) val = int(arr[1], 16) assert i == val, i 07070100000033000081FD000000000000000000000001625E86EB0000095F000000000000000000000000000000000000001E00000000rhnlib/test/23-digest-auth.py#!/usr/bin/python # # Test case for digest authentication # # # USAGE: (echo -n) | $0 PORT # # Few notes about what is done here: # - thread AUTH sends authentication digest # - thread NC uses netcat and grep to see results # - Little hack with (echo -n) is much more easier to use, # than some settrace machinary import sys import socket import os import httplib from threading import Thread sys.path.append('..') from rhn.rpclib import Server SERVER = "longusername0123456789:longpassword0123456789@localhost" PORT = "1234" HANDLER = "/XMLRPC" try: PORT = sys.argv[1] except: pass class killable(Thread): """Just Thread with a kill() method.""" def __init__(self, *args, **keywords): Thread.__init__(self, *args, **keywords) self.killed = False def start(self): self.__run_backup = self.run self.run = self.__run # Force the Thread to install our trace. Thread.start(self) def __run(self): sys.settrace(self.globaltrace) self.__run_backup() self.run = self.__run_backup def globaltrace(self, frame, why, arg): if why == 'call': return self.localtrace else: return None def localtrace(self, frame, why, arg): if self.killed: if why == 'line': raise SystemExit() return self.localtrace def kill(self): self.killed = True def authenticate(): global SERVER, PORT, HANDLER s = Server("http://" + SERVER + ":" + PORT + HANDLER); connected = False; while not connected: try: connected = True; print s.test.method() except socket.error, e: # nobody is listenning, try to authenticate again connected = False; pass; except httplib.BadStatusLine, e: # This is ok, netcat does not send apropriate response pass def netcat(): global auth cmd = "nc -l " + PORT + " | grep authorization\:\ Basic\ bG9uZ3VzZXJuYW1lMDEyMzQ1Njc4OTpsb25ncGFzc3dvcmQwMTIzNDU2Nzg5" result = os.system(cmd); if (result == 0): print "Tests PASSES" else: auth.kill(); print "Test FAILS" if __name__ == '__main__': global nc, auth nc = killable(target = netcat); auth = killable(target = authenticate); nc.start(); auth.start(); 07070100000034000081FD000000000000000000000001625E86EB000008ED000000000000000000000000000000000000001E00000000rhnlib/test/24-ssl-memleak.py#!/usr/bin/python # # # USAGE: $0 SERVER CERT """ Tests a memory leak in the applet """ import sys from rhn import rpclib def main(): server_name = "xmlrpc.rhn.redhat.com" ca_cert = "/usr/share/rhn/RHNS-CA-CERT" try: server_name = sys.argv[1] ca_cert = sys.argv[2] except: pass server_url = "https://" + server_name + "/APPLET" mem_usage = None mem_usage_VmSize_max = None mem_usage_VmSize_first = None mem_usage_VmSize_allowed_percent = 0.5 # [%] allowed gain of first -> max for i in range(1,10000): run_test(server_url, ca_cert) if i % 100 == 0: new_mem_usage = mem_usage_int() if mem_usage is not None: if new_mem_usage[1] > mem_usage_VmSize_max: mem_usage_VmSize_max = new_mem_usage[1] else: mem_usage_VmSize_max = new_mem_usage[1] mem_usage_VmSize_first = new_mem_usage[1] mem_usage = new_mem_usage print "memory usage: %s %s %s" % mem_usage[1:4] percent = float((mem_usage_VmSize_max - mem_usage_VmSize_first)) / (float(mem_usage_VmSize_first) / 100) if percent >= mem_usage_VmSize_allowed_percent: # Failure print "Test FAILS (%s %%)" % percent return 1 print "Test PASSES" return 0 def run_test(server_url, ca_cert): s = rpclib.Server(server_url) s.add_trusted_cert(ca_cert) status = s.applet.poll_status() def mem_usage(): f = open("/proc/self/status") dct = {} while 1: line = f.readline() if not line: break arr = [s.strip() for s in line.split(':', 1)] if len(arr) == 1: continue dct[arr[0]] = arr[1] return dct['Name'], dct['VmSize'], dct['VmRSS'], dct['VmData'] def mem_usage_int(): memusage = mem_usage() ret = [memusage[0]] for val in memusage[1:4]: # Split it arr = val.split() try: v = int(arr[0]) except ValueError: v = val ret.append(v) return tuple(ret) def _line_value(line): arr = line.split(':', 1) if len(arr) == 1: return None return arr[1].strip() if __name__ == '__main__': sys.exit(main() or 0) 07070100000035000081B4000000000000000000000001625E86EB000007B5000000000000000000000000000000000000001E00000000rhnlib/test/25-ssl-memleak.py#!/usr/bin/python # """ Tests a memory leak in the applet """ import sys import socket from rhn import SSL def main(): server_name = "xmlrpc.rhn.redhat.com" server_port = 443 ca_cert = "/usr/share/rhn/RHNS-CA-CERT" try: server_name = sys.argv[1] server_port = int(sys.argv[2]) ca_cert = sys.argv[3] except: pass diff_count = 0 mem_usage = None for i in range(10000): run_test(server_name, server_port, ca_cert) if i % 100 == 0: new_mem_usage = mem_usage_int() if mem_usage is not None: if mem_usage[1] != new_mem_usage[1]: diff_count = diff_count + 1 mem_usage = new_mem_usage print "memory usage: %s %s %s" % mem_usage[1:4] if diff_count > 4: # Failure print "Test FAILS" return diff_count print "Test PASSES" return 0 def run_test(server_name, server_port, ca_cert): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((server_name, server_port)) sslsock = SSL.SSLSocket(sock, [ca_cert]) sslsock.init_ssl() sslsock.close() def mem_usage(): f = open("/proc/self/status") dct = {} while 1: line = f.readline() if not line: break arr = [s.strip() for s in line.split(':', 1)] if len(arr) == 1: continue dct[arr[0]] = arr[1] return dct['Name'], dct['VmSize'], dct['VmRSS'], dct['VmData'] def mem_usage_int(): memusage = mem_usage() ret = [memusage[0]] for val in memusage[1:4]: # Split it arr = val.split() try: v = int(arr[0]) except ValueError: v = val ret.append(v) return tuple(ret) def _line_value(line): arr = line.split(':', 1) if len(arr) == 1: return None return arr[1].strip() if __name__ == '__main__': sys.exit(main() or 0) 07070100000036000081B4000000000000000000000001625E86EB0000039F000000000000000000000000000000000000001F00000000rhnlib/test/26-ssl-bad-cert.py#!/usr/bin/python # """ Negative test - make sure we raise an exception when the wrong SSL cert is used """ import sys import socket from rhn import SSL def main(): server_name = "www.redhat.com" server_port = 443 ca_cert = "/usr/share/rhn/RHNS-CA-CERT" try: run_test(server_name, server_port, ca_cert) except SSL.SSL.Error, e: if e[0][0][2] == 'certificate verify failed': print "test PASSES" return 0 print "Test failed for unknown reasons:", e return 1 print "Connection did not fail, test FAILS" return 1 def run_test(server_name, server_port, ca_cert): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((server_name, server_port)) sslsock = SSL.SSLSocket(sock, [ca_cert]) sslsock.init_ssl() sslsock.do_handshake() sslsock.close() if __name__ == '__main__': sys.exit(main() or 0) 07070100000037000081B4000000000000000000000001625E86EB00000123000000000000000000000000000000000000001500000000rhnlib/test/MakefilePY_VERSION = $(shell python -c "import sys; print sys.version.split()[0]") FILES = $(wildcard *.py) .PHONY: all test compile all: test: compile $(FILES) compile: @python /usr/lib/python$(PY_VERSION)/compileall.py . %.py: %.pyc @python $< || exit 1 clean: @rm -fv `cat .cvsignore` 07070100000038000081B4000000000000000000000001625E86EB00000000000000000000000000000000000000000000001800000000rhnlib/test/__init__.py07070100000039000081B4000000000000000000000001625E86EB00001E9A000000000000000000000000000000000000001D00000000rhnlib/test/base16values.txt0 0 1 1 2 2 3 3 4 4 5 5 6 6 7 7 8 8 9 9 10 a 11 b 12 c 13 d 14 e 15 f 16 10 17 11 18 12 19 13 20 14 21 15 22 16 23 17 24 18 25 19 26 1a 27 1b 28 1c 29 1d 30 1e 31 1f 32 20 33 21 34 22 35 23 36 24 37 25 38 26 39 27 40 28 41 29 42 2a 43 2b 44 2c 45 2d 46 2e 47 2f 48 30 49 31 50 32 51 33 52 34 53 35 54 36 55 37 56 38 57 39 58 3a 59 3b 60 3c 61 3d 62 3e 63 3f 64 40 65 41 66 42 67 43 68 44 69 45 70 46 71 47 72 48 73 49 74 4a 75 4b 76 4c 77 4d 78 4e 79 4f 80 50 81 51 82 52 83 53 84 54 85 55 86 56 87 57 88 58 89 59 90 5a 91 5b 92 5c 93 5d 94 5e 95 5f 96 60 97 61 98 62 99 63 100 64 101 65 102 66 103 67 104 68 105 69 106 6a 107 6b 108 6c 109 6d 110 6e 111 6f 112 70 113 71 114 72 115 73 116 74 117 75 118 76 119 77 120 78 121 79 122 7a 123 7b 124 7c 125 7d 126 7e 127 7f 128 80 129 81 130 82 131 83 132 84 133 85 134 86 135 87 136 88 137 89 138 8a 139 8b 140 8c 141 8d 142 8e 143 8f 144 90 145 91 146 92 147 93 148 94 149 95 150 96 151 97 152 98 153 99 154 9a 155 9b 156 9c 157 9d 158 9e 159 9f 160 a0 161 a1 162 a2 163 a3 164 a4 165 a5 166 a6 167 a7 168 a8 169 a9 170 aa 171 ab 172 ac 173 ad 174 ae 175 af 176 b0 177 b1 178 b2 179 b3 180 b4 181 b5 182 b6 183 b7 184 b8 185 b9 186 ba 187 bb 188 bc 189 bd 190 be 191 bf 192 c0 193 c1 194 c2 195 c3 196 c4 197 c5 198 c6 199 c7 200 c8 201 c9 202 ca 203 cb 204 cc 205 cd 206 ce 207 cf 208 d0 209 d1 210 d2 211 d3 212 d4 213 d5 214 d6 215 d7 216 d8 217 d9 218 da 219 db 220 dc 221 dd 222 de 223 df 224 e0 225 e1 226 e2 227 e3 228 e4 229 e5 230 e6 231 e7 232 e8 233 e9 234 ea 235 eb 236 ec 237 ed 238 ee 239 ef 240 f0 241 f1 242 f2 243 f3 244 f4 245 f5 246 f6 247 f7 248 f8 249 f9 250 fa 251 fb 252 fc 253 fd 254 fe 255 ff 256 100 257 101 258 102 259 103 260 104 261 105 262 106 263 107 264 108 265 109 266 10a 267 10b 268 10c 269 10d 270 10e 271 10f 272 110 273 111 274 112 275 113 276 114 277 115 278 116 279 117 280 118 281 119 282 11a 283 11b 284 11c 285 11d 286 11e 287 11f 288 120 289 121 290 122 291 123 292 124 293 125 294 126 295 127 296 128 297 129 298 12a 299 12b 300 12c 301 12d 302 12e 303 12f 304 130 305 131 306 132 307 133 308 134 309 135 310 136 311 137 312 138 313 139 314 13a 315 13b 316 13c 317 13d 318 13e 319 13f 320 140 321 141 322 142 323 143 324 144 325 145 326 146 327 147 328 148 329 149 330 14a 331 14b 332 14c 333 14d 334 14e 335 14f 336 150 337 151 338 152 339 153 340 154 341 155 342 156 343 157 344 158 345 159 346 15a 347 15b 348 15c 349 15d 350 15e 351 15f 352 160 353 161 354 162 355 163 356 164 357 165 358 166 359 167 360 168 361 169 362 16a 363 16b 364 16c 365 16d 366 16e 367 16f 368 170 369 171 370 172 371 173 372 174 373 175 374 176 375 177 376 178 377 179 378 17a 379 17b 380 17c 381 17d 382 17e 383 17f 384 180 385 181 386 182 387 183 388 184 389 185 390 186 391 187 392 188 393 189 394 18a 395 18b 396 18c 397 18d 398 18e 399 18f 400 190 401 191 402 192 403 193 404 194 405 195 406 196 407 197 408 198 409 199 410 19a 411 19b 412 19c 413 19d 414 19e 415 19f 416 1a0 417 1a1 418 1a2 419 1a3 420 1a4 421 1a5 422 1a6 423 1a7 424 1a8 425 1a9 426 1aa 427 1ab 428 1ac 429 1ad 430 1ae 431 1af 432 1b0 433 1b1 434 1b2 435 1b3 436 1b4 437 1b5 438 1b6 439 1b7 440 1b8 441 1b9 442 1ba 443 1bb 444 1bc 445 1bd 446 1be 447 1bf 448 1c0 449 1c1 450 1c2 451 1c3 452 1c4 453 1c5 454 1c6 455 1c7 456 1c8 457 1c9 458 1ca 459 1cb 460 1cc 461 1cd 462 1ce 463 1cf 464 1d0 465 1d1 466 1d2 467 1d3 468 1d4 469 1d5 470 1d6 471 1d7 472 1d8 473 1d9 474 1da 475 1db 476 1dc 477 1dd 478 1de 479 1df 480 1e0 481 1e1 482 1e2 483 1e3 484 1e4 485 1e5 486 1e6 487 1e7 488 1e8 489 1e9 490 1ea 491 1eb 492 1ec 493 1ed 494 1ee 495 1ef 496 1f0 497 1f1 498 1f2 499 1f3 500 1f4 501 1f5 502 1f6 503 1f7 504 1f8 505 1f9 506 1fa 507 1fb 508 1fc 509 1fd 510 1fe 511 1ff 512 200 513 201 514 202 515 203 516 204 517 205 518 206 519 207 520 208 521 209 522 20a 523 20b 524 20c 525 20d 526 20e 527 20f 528 210 529 211 530 212 531 213 532 214 533 215 534 216 535 217 536 218 537 219 538 21a 539 21b 540 21c 541 21d 542 21e 543 21f 544 220 545 221 546 222 547 223 548 224 549 225 550 226 551 227 552 228 553 229 554 22a 555 22b 556 22c 557 22d 558 22e 559 22f 560 230 561 231 562 232 563 233 564 234 565 235 566 236 567 237 568 238 569 239 570 23a 571 23b 572 23c 573 23d 574 23e 575 23f 576 240 577 241 578 242 579 243 580 244 581 245 582 246 583 247 584 248 585 249 586 24a 587 24b 588 24c 589 24d 590 24e 591 24f 592 250 593 251 594 252 595 253 596 254 597 255 598 256 599 257 600 258 601 259 602 25a 603 25b 604 25c 605 25d 606 25e 607 25f 608 260 609 261 610 262 611 263 612 264 613 265 614 266 615 267 616 268 617 269 618 26a 619 26b 620 26c 621 26d 622 26e 623 26f 624 270 625 271 626 272 627 273 628 274 629 275 630 276 631 277 632 278 633 279 634 27a 635 27b 636 27c 637 27d 638 27e 639 27f 640 280 641 281 642 282 643 283 644 284 645 285 646 286 647 287 648 288 649 289 650 28a 651 28b 652 28c 653 28d 654 28e 655 28f 656 290 657 291 658 292 659 293 660 294 661 295 662 296 663 297 664 298 665 299 666 29a 667 29b 668 29c 669 29d 670 29e 671 29f 672 2a0 673 2a1 674 2a2 675 2a3 676 2a4 677 2a5 678 2a6 679 2a7 680 2a8 681 2a9 682 2aa 683 2ab 684 2ac 685 2ad 686 2ae 687 2af 688 2b0 689 2b1 690 2b2 691 2b3 692 2b4 693 2b5 694 2b6 695 2b7 696 2b8 697 2b9 698 2ba 699 2bb 700 2bc 701 2bd 702 2be 703 2bf 704 2c0 705 2c1 706 2c2 707 2c3 708 2c4 709 2c5 710 2c6 711 2c7 712 2c8 713 2c9 714 2ca 715 2cb 716 2cc 717 2cd 718 2ce 719 2cf 720 2d0 721 2d1 722 2d2 723 2d3 724 2d4 725 2d5 726 2d6 727 2d7 728 2d8 729 2d9 730 2da 731 2db 732 2dc 733 2dd 734 2de 735 2df 736 2e0 737 2e1 738 2e2 739 2e3 740 2e4 741 2e5 742 2e6 743 2e7 744 2e8 745 2e9 746 2ea 747 2eb 748 2ec 749 2ed 750 2ee 751 2ef 752 2f0 753 2f1 754 2f2 755 2f3 756 2f4 757 2f5 758 2f6 759 2f7 760 2f8 761 2f9 762 2fa 763 2fb 764 2fc 765 2fd 766 2fe 767 2ff 768 300 769 301 770 302 771 303 772 304 773 305 774 306 775 307 776 308 777 309 778 30a 779 30b 780 30c 781 30d 782 30e 783 30f 784 310 785 311 786 312 787 313 788 314 789 315 790 316 791 317 792 318 793 319 794 31a 795 31b 796 31c 797 31d 798 31e 799 31f 800 320 801 321 802 322 803 323 804 324 805 325 806 326 807 327 808 328 809 329 810 32a 811 32b 812 32c 813 32d 814 32e 815 32f 816 330 817 331 818 332 819 333 820 334 821 335 822 336 823 337 824 338 825 339 826 33a 827 33b 828 33c 829 33d 830 33e 831 33f 832 340 833 341 834 342 835 343 836 344 837 345 838 346 839 347 840 348 841 349 842 34a 843 34b 844 34c 845 34d 846 34e 847 34f 848 350 849 351 850 352 851 353 852 354 853 355 854 356 855 357 856 358 857 359 858 35a 859 35b 860 35c 861 35d 862 35e 863 35f 864 360 865 361 866 362 867 363 868 364 869 365 870 366 871 367 872 368 873 369 874 36a 875 36b 876 36c 877 36d 878 36e 879 36f 880 370 881 371 882 372 883 373 884 374 885 375 886 376 887 377 888 378 889 379 890 37a 891 37b 892 37c 893 37d 894 37e 895 37f 896 380 897 381 898 382 899 383 900 384 901 385 902 386 903 387 904 388 905 389 906 38a 907 38b 908 38c 909 38d 910 38e 911 38f 912 390 913 391 914 392 915 393 916 394 917 395 918 396 919 397 920 398 921 399 922 39a 923 39b 924 39c 925 39d 926 39e 927 39f 928 3a0 929 3a1 930 3a2 931 3a3 932 3a4 933 3a5 934 3a6 935 3a7 936 3a8 937 3a9 938 3aa 939 3ab 940 3ac 941 3ad 942 3ae 943 3af 944 3b0 945 3b1 946 3b2 947 3b3 948 3b4 949 3b5 950 3b6 951 3b7 952 3b8 953 3b9 954 3ba 955 3bb 956 3bc 957 3bd 958 3be 959 3bf 960 3c0 961 3c1 962 3c2 963 3c3 964 3c4 965 3c5 966 3c6 967 3c7 968 3c8 969 3c9 970 3ca 971 3cb 972 3cc 973 3cd 974 3ce 975 3cf 976 3d0 977 3d1 978 3d2 979 3d3 980 3d4 981 3d5 982 3d6 983 3d7 984 3d8 985 3d9 986 3da 987 3db 988 3dc 989 3dd 990 3de 991 3df 992 3e0 993 3e1 994 3e2 995 3e3 996 3e4 997 3e5 998 3e6 999 3e7 1000 3e8 1001 3e9 1002 3ea 1003 3eb 1004 3ec 1005 3ed 1006 3ee 1007 3ef 1008 3f0 1009 3f1 1010 3f2 1011 3f3 1012 3f4 1013 3f5 1014 3f6 1015 3f7 1016 3f8 1017 3f9 1018 3fa 1019 3fb 1020 3fc 1021 3fd 1022 3fe 1023 3ff 0707010000003A000081B4000000000000000000000001625E86EB0000041B000000000000000000000000000000000000001B00000000rhnlib/test/test_server.py#!/usr/bin/python # # import unittest from rhn.rpclib import Server from rhn.rpclib import MalformedURIError class ServerTest(unittest.TestCase): # def setUp(self): # def tearDown(self): def testGoodURIWithHTTP(self): try: Server("http://localhost") except: assert False def testGoodURIWithHTTPS(self): try: Server("https://localhost") except: assert False def testURIMissingProtocol(self): self.assertRaises(MalformedURIError, Server, "localhost") def testURIMissingHost(self): self.assertRaises(MalformedURIError, Server, "http://") def testURIMissingHostAndSlashes(self): self.assertRaises(MalformedURIError, Server, "http:") def testURIWithGarbageInsteadOfSlashesAndHost(self): self.assertRaises(MalformedURIError, Server, "http:alsofh") def testURIMissingColon(self): self.assertRaises(MalformedURIError, Server, "http//localhost") if __name__ == "__main__": unittest.main() 0707010000003B000081B4000000000000000000000001625E86EB00007C4D000000000000000000000000000000000000001500000000rhnlib/transports.py# # Helper transport objects # # Copyright (c) 2002--2016 Red Hat, Inc. # # Author: Mihai Ibanescu <misa@redhat.com> # Based on what was previously shipped as cgiwrap: # - Cristian Gafton <gafton@redhat.com> # - Erik Troan <ewt@redhat.com> # Transport objects import os import sys import time from rhn import connections from rhn.stringutils import sstr, bstr from rhn.SmartIO import SmartIO from rhn.UserDictCase import UserDictCase try: # python2 import xmlrpclib from types import IntType, StringType, ListType except ImportError: # python3 import xmlrpc.client as xmlrpclib IntType = int StringType = bytes ListType = list __version__ = "$Revision$" # XXX COMPRESS_LEVEL = 6 # Exceptions class NotProcessed(Exception): pass class Transport(xmlrpclib.Transport): user_agent = "rhn.rpclib.py/%s" % __version__ def __init__(self, transfer=0, encoding=0, refreshCallback=None, progressCallback=None, use_datetime=None, timeout=None): self._use_builtin_types = False self._transport_flags = {'transfer' : 0, 'encoding' : 0} self.set_transport_flags(transfer=transfer, encoding=encoding) self._headers = UserDictCase() self.verbose = 0 self.connection = None self.method = "POST" self._lang = None self.refreshCallback = refreshCallback self.progressCallback = progressCallback self.bufferSize = 16384 self.headers_in = None self.response_status = None self.response_reason = None self._redirected = None self._use_datetime = use_datetime self.timeout = timeout # set the progress callback def set_progress_callback(self, progressCallback, bufferSize=16384): self.progressCallback = progressCallback self.bufferSize = bufferSize # set the refresh callback def set_refresh_callback(self, refreshCallback): self.refreshCallback = refreshCallback # set the buffer size # The bigger this is, the faster the read is, but the more seldom is the # progress callback called def set_buffer_size(self, bufferSize): if bufferSize is None: # No buffer size specified; go with 16k bufferSize = 16384 self.bufferSize = bufferSize # set the request method def set_method(self, method): if method not in ("GET", "POST"): raise IOError("Unknown request method %s" % method) self.method = method # reset the transport options def set_transport_flags(self, transfer=None, encoding=None, **kwargs): # For backwards compatibility, we keep transfer and encoding as # positional parameters (they could come in as kwargs easily) self._transport_flags.update(kwargs) if transfer is not None: self._transport_flags['transfer'] = transfer if encoding is not None: self._transport_flags['encoding'] = encoding self.validate_transport_flags() def get_transport_flags(self): return self._transport_flags.copy() def validate_transport_flags(self): # Transfer and encoding are guaranteed to be there transfer = self._transport_flags.get('transfer') transfer = lookupTransfer(transfer, strict=1) self._transport_flags['transfer'] = transfer encoding = self._transport_flags.get('encoding') encoding = lookupEncoding(encoding, strict=1) self._transport_flags['encoding'] = encoding # Add arbitrary additional headers. def set_header(self, name, arg): if type(arg) in [ type([]), type(()) ]: # Multivalued header self._headers[name] = [str(a) for a in arg] else: self._headers[name] = str(arg) def add_header(self, name, arg): if name in self._headers: vlist = self._headers[name] if not isinstance(vlist, ListType): vlist = [ vlist ] else: vlist = self._headers[name] = [] vlist.append(str(arg)) def clear_headers(self): self._headers.clear() def get_connection(self, host): if self.verbose: print("Connecting via http to %s" % (host, )) if self.timeout: return connections.HTTPConnection(host, timeout=self.timeout) else: return connections.HTTPConnection(host) def request(self, host, handler, request_body, verbose=0): # issue XML-RPC request # XXX: automatically compute how to send depending on how much data # you want to send # XXX Deal with HTTP/1.1 if necessary self.verbose = verbose # implement BASIC HTTP AUTHENTICATION host, extra_headers, x509 = self.get_host_info(host) if not extra_headers: extra_headers = [] # Establish the connection connection = self.get_connection(host) # Setting the user agent. Only interesting for SSL tunnels, in any # other case the general headers are good enough. connection.set_user_agent(self.user_agent) if self.verbose: connection.set_debuglevel(self.verbose - 1) # Get the output object to push data with req = Output(connection=connection, method=self.method) req.set_transport_flags(**self._transport_flags) # Add the extra headers req.set_header('User-Agent', self.user_agent) for header, value in list(self._headers.items()) + extra_headers: # Output.set_header correctly deals with multivalued headers now req.set_header(header, value) # Content-Type req.set_header("Content-Type", "text/xml") req.process(request_body) # Host and Content-Length are set by HTTP*Connection for h in ['Content-Length', 'Host']: req.clear_header(h) headers, fd = req.send_http(host, handler) if self.verbose: print("Incoming headers:") for header, value in headers.items(): print("\t%s : %s" % (header, value)) if fd.status in (301, 302): self._redirected = headers["Location"] self.response_status = fd.status return None # Save the headers self.headers_in = headers self.response_status = fd.status self.response_reason = fd.reason return self._process_response(fd, connection) def _process_response(self, fd, connection): # Now use the Input class in case we get an enhanced response resp = Input(self.headers_in, progressCallback=self.progressCallback, bufferSize=self.bufferSize) fd = resp.decode(fd) if isinstance(fd, InputStream): # When the File object goes out of scope, so will the InputStream; # that will eventually call the connection's close() method and # cleanly reap it f = File(fd.fd, fd.length, fd.name, bufferSize=self.bufferSize, progressCallback=self.progressCallback) # Set the File's close method to the connection's # Note that calling the HTTPResponse's close() is not enough, # since the main socket would remain open, and this is # particularily bad with SSL f.close = connection.close return f # We can safely close the connection now; if we had an # application/octet/stream (for which Input.read passes the original # socket object), Input.decode would return an InputStream, # so we wouldn't reach this point connection.close() return self.parse_response(fd) # Give back the new URL if redirected def redirected(self): return self._redirected # Rewrite parse_response to provide refresh callbacks def parse_response(self, f): # read response from input file, and parse it p, u = self.getparser() while 1: response = f.read(1024) if not response: break if self.refreshCallback: self.refreshCallback() if self.verbose: print("body:", repr(response)) p.feed(response) f.close() p.close() return u.close() def setlang(self, lang): self._lang = lang class SafeTransport(Transport): def __init__(self, transfer=0, encoding=0, refreshCallback=None, progressCallback=None, trusted_certs=None, timeout=None): Transport.__init__(self, transfer, encoding, refreshCallback=refreshCallback, progressCallback=progressCallback, timeout=timeout) self.trusted_certs = [] for certfile in (trusted_certs or []): self.add_trusted_cert(certfile) def add_trusted_cert(self, certfile): if not os.access(certfile, os.R_OK): raise ValueError("Certificate file %s is not accessible" % certfile) self.trusted_certs.append(certfile) def get_connection(self, host): # implement BASIC HTTP AUTHENTICATION host, extra_headers, x509 = self.get_host_info(host) if self.verbose: print("Connecting via https to %s" % (host, )) if self.timeout: return connections.HTTPSConnection(host, trusted_certs=self.trusted_certs, timeout=self.timeout) else: return connections.HTTPSConnection(host, trusted_certs=self.trusted_certs) class ProxyTransport(Transport): def __init__(self, proxy, proxyUsername=None, proxyPassword=None, transfer=0, encoding=0, refreshCallback=None, progressCallback=None, timeout=None): Transport.__init__(self, transfer, encoding, refreshCallback=refreshCallback, progressCallback=progressCallback, timeout=timeout) self._proxy = proxy self._proxy_username = proxyUsername self._proxy_password = proxyPassword def get_connection(self, host): if self.verbose: print("Connecting via http to %s proxy %s, username %s, pass %s" % ( host, self._proxy, self._proxy_username, self._proxy_password)) if self.timeout: return connections.HTTPProxyConnection(self._proxy, host, username=self._proxy_username, password=self._proxy_password, timeout=self.timeout) else: return connections.HTTPProxyConnection(self._proxy, host, username=self._proxy_username, password=self._proxy_password) class SafeProxyTransport(ProxyTransport): def __init__(self, proxy, proxyUsername=None, proxyPassword=None, transfer=0, encoding=0, refreshCallback=None, progressCallback=None, trusted_certs=None, timeout=None): ProxyTransport.__init__(self, proxy, proxyUsername=proxyUsername, proxyPassword=proxyPassword, transfer=transfer, encoding=encoding, refreshCallback=refreshCallback, progressCallback=progressCallback, timeout=timeout) self.trusted_certs = [] for certfile in (trusted_certs or []): self.add_trusted_cert(certfile) def add_trusted_cert(self, certfile): if not os.access(certfile, os.R_OK): raise ValueError("Certificate file %s is not accessible" % certfile) self.trusted_certs.append(certfile) def get_connection(self, host): if self.verbose: print("Connecting via https to %s proxy %s, username %s, pass %s" % ( host, self._proxy, self._proxy_username, self._proxy_password)) if self.timeout: return connections.HTTPSProxyConnection(self._proxy, host, username=self._proxy_username, password=self._proxy_password, trusted_certs=self.trusted_certs, timeout=self.timeout) else: return connections.HTTPSProxyConnection(self._proxy, host, username=self._proxy_username, password=self._proxy_password, trusted_certs=self.trusted_certs) # ============================================================================ # Extended capabilities for transport # # We allow for the following possible headers: # # Content-Transfer-Encoding: # This header tells us how the POST data is encoded in what we read. # If it is not set, we assume plain text that can be passed along # without any other modification. If set, valid values are: # - binary : straight binary data # - base64 : will pass through base64 decoder to get the binary data # # Content-Encoding: # This header tells us what should we do with the binary data obtained # after acting on the Content-Transfer-Encoding header. Valid values: # - x-gzip : will need to pass through GNU gunzip-like to get plain # text out # - x-zlib : this denotes the Python's own zlib bindings which are a # datastream based on gzip, but not quite # - x-gpg : will need to pass through GPG to get out the text we want # ============================================================================ # Input class to automate reading the posting from the network # Having to work with environment variables blows, though class Input: def __init__(self, headers=None, progressCallback=None, bufferSize=1024, max_mem_size=16384): self.transfer = None self.encoding = None self.type = None self.length = 0 self.lang = "C" self.name = "" self.progressCallback = progressCallback self.bufferSize = bufferSize self.max_mem_size = max_mem_size if not headers: # we need to get them from environment if "HTTP_CONTENT_TRANSFER_ENCODING" in os.environ: self.transfer = os.environ["HTTP_CONTENT_TRANSFER_ENCODING"].lower() if "HTTP_CONTENT_ENCODING" in os.environ: self.encoding = os.environ["HTTP_CONTENT_ENCODING"].lower() if "CONTENT-TYPE" in os.environ: self.type = os.environ["CONTENT-TYPE"].lower() if "CONTENT_LENGTH" in os.environ: self.length = int(os.environ["CONTENT_LENGTH"]) if "HTTP_ACCEPT_LANGUAGE" in os.environ: self.lang = os.environ["HTTP_ACCEPT_LANGUAGE"] if "HTTP_X_PACKAGE_FILENAME" in os.environ: self.name = os.environ["HTTP_X_PACKAGE_FILENAME"] else: # The stupid httplib screws up the headers from the HTTP repsonse # and converts them to lowercase. This means that we have to # convert to lowercase all the dictionary keys in case somebody calls # us with sane values --gaftonc (actually mimetools is the culprit) for header in headers.keys(): value = headers[header] h = header.lower() if h == "content-length": try: self.length = int(value) except ValueError: self.length = 0 elif h == "content-transfer-encoding": # RFC 2045 #6.1: case insensitive self.transfer = value.lower() elif h == "content-encoding": # RFC 2616 #3.5: case insensitive self.encoding = value.lower() elif h == "content-type": # RFC 2616 #3.7: case insensitive self.type = value.lower() elif h == "accept-language": # RFC 2616 #3.10: case insensitive self.lang = value.lower() elif h == "x-package-filename": self.name = value self.io = None def read(self, fd = sys.stdin): # The octet-streams are passed right back if self.type == "application/octet-stream": return if self.length: # Read exactly the amount of data we were told self.io = _smart_read(fd, self.length, bufferSize=self.bufferSize, progressCallback=self.progressCallback, max_mem_size=self.max_mem_size) else: # Oh well, no clue; read until EOF (hopefully) self.io = _smart_total_read(fd) if not self.transfer or self.transfer == "binary": return elif self.transfer == "base64": import base64 old_io = self.io old_io.seek(0, 0) self.io = SmartIO(max_mem_size=self.max_mem_size) base64.decode(old_io, self.io) else: raise NotImplementedError(self.transfer) def decode(self, fd = sys.stdin): # The octet-stream data are passed right back if self.type == "application/octet-stream": return InputStream(fd, self.length, self.name, close=fd.close) if not self.io: self.read(fd) # At this point self.io exists (the only case when self.read() does # not initialize self.io is when content-type is # "application/octet-stream" - and we already dealt with that case # We can now close the file descriptor if hasattr(fd, "close"): fd.close() # Now we have the binary goo if not self.encoding or self.encoding == "__plain": # all is fine. pass elif self.encoding in ("x-zlib", "deflate"): import zlib obj = zlib.decompressobj() self.io.seek(0, 0) data = obj.decompress(self.io.read()) + obj.flush() del obj self.length = len(data) self.io = SmartIO(max_mem_size=self.max_mem_size) self.io.write(data) elif self.encoding in ("x-gzip", "gzip"): import gzip self.io.seek(0, 0) gz = gzip.GzipFile(mode="rb", compresslevel = COMPRESS_LEVEL, fileobj=self.io) data = gz.read() self.length = len(data) self.io = SmartIO(max_mem_size=self.max_mem_size) self.io.write(data) elif self.encoding == "x-gpg": # XXX: should be written raise NotImplementedError(self.transfer, self.encoding) else: raise NotImplementedError(self.transfer, self.encoding) # Play nicely and rewind the file descriptor self.io.seek(0, 0) return self.io def getlang(self): return self.lang # Utility functions def _smart_total_read(fd, bufferSize=1024, max_mem_size=16384): """ Tries to read data from the supplied stream, and puts the results into a StmartIO object. The data will be in memory or in a temporary file, depending on how much it's been read Returns a SmartIO object """ io = SmartIO(max_mem_size=max_mem_size) while 1: chunk = fd.read(bufferSize) if not chunk: # EOF reached break io.write(chunk) return io def _smart_read(fd, amt, bufferSize=1024, progressCallback=None, max_mem_size=16384): # Reads amt bytes from fd, or until the end of file, whichever # occurs first # The function will read in memory if the amout to be read is smaller than # max_mem_size, or to a temporary file otherwise # # Unlike read(), _smart_read tries to return exactly the requested amount # (whereas read will return _up_to_ that amount). Reads from sockets will # usually reaturn less data, or the read can be interrupted # # Inspired by Greg Stein's httplib.py (the standard in python 2.x) # # support for progress callbacks added startTime = time.time() lastTime = startTime buf = SmartIO(max_mem_size=max_mem_size) origsize = amt while amt > 0: curTime = time.time() l = min(bufferSize, amt) chunk = fd.read(l) # read guarantees that len(chunk) <= l l = len(chunk) if not l: # Oops. Most likely EOF break # And since the original l was smaller than amt, we know amt >= 0 amt = amt - l buf.write(chunk) if progressCallback is None: # No progress callback, so don't do fancy computations continue # We update the progress callback if: # we haven't updated it for more than a secord, or # it's the last read (amt == 0) if curTime - lastTime >= 1 or amt == 0: lastTime = curTime # use float() so that we force float division in the next step bytesRead = float(origsize - amt) # if amt == 0, on a fast machine it is possible to have # curTime - lastTime == 0, so add an epsilon to prevent a division # by zero speed = bytesRead / ((curTime - startTime) + .000001) if origsize == 0: secs = 0 else: # speed != 0 because bytesRead > 0 # (if bytesRead == 0 then origsize == amt, which means a read # of 0 length; but that's impossible since we already checked # that l is non-null secs = amt / speed progressCallback(bytesRead, origsize, speed, secs) # Now rewind the SmartIO buf.seek(0, 0) return buf class InputStream: def __init__(self, fd, length, name = "<unknown>", close=None): self.fd = fd self.length = int(length) self.name = name # Close function self.close = close def __repr__(self): return "Input data is a stream of %d bytes for file %s.\n" % (self.length, self.name) # ============================================================================ # Output class that will be used to build the temporary output string class BaseOutput: # DEFINES for instances use # Content-Encoding ENCODE_NONE = 0 ENCODE_GZIP = 1 ENCODE_ZLIB = 2 ENCODE_GPG = 3 # Content-Transfer-Encoding TRANSFER_NONE = 0 TRANSFER_BINARY = 1 TRANSFER_BASE64 = 2 # Mappings to make things easy encodings = [ [None, "__plain"], # ENCODE_NONE ["x-gzip", "gzip"], # ENCODE_GZIP ["x-zlib", "deflate"], # ENCODE_ZLIB ["x-gpg"], # ENCODE_GPG ] transfers = [ None, # TRANSFER_NONE "binary", # TRANSFRE_BINARY "base64", # TRANSFER_BASE64 ] def __init__(self, transfer=0, encoding=0, connection=None, method="POST"): # Assumes connection is an instance of HTTPConnection if connection: if not isinstance(connection, connections.HTTPConnection): raise Exception("Expected an HTTPConnection type object") self.method = method # Store the connection self._connection = connection self.data = None self.headers = UserDictCase() self.encoding = 0 self.transfer = 0 self.transport_flags = {} # for authenticated proxies self.username = None self.password = None # Fields to keep the information about the server self._host = None self._handler = None self._http_type = None self._protocol = None # Initialize self.transfer and self.encoding self.set_transport_flags(transfer=transfer, encoding=encoding) # internal flags self.__processed = 0 def set_header(self, name, arg): if type(arg) in [ type([]), type(()) ]: # Multi-valued header # # Per RFC 2616, section 4.2 (Message Headers): # Multiple message-header fields with the same field-name MAY be # present in a message if and only if the entire field-value for # the header field is defined as a comma-separated list [i.e. # #(values)]. It MUST be possible to combine the multiple header # fields into one "field-name: field-value" pair, without # changing the semantics of the message, by appending each # subsequent field-value to the first, each separated by a comma. self.headers[name] = ','.join(map(str, arg)) else: self.headers[name] = str(arg) def clear_header(self, name): if name in self.headers: del self.headers[name] def process(self, data): # Assume straight text/xml self.data = data # Content-Encoding header if self.encoding == self.ENCODE_GZIP: import gzip encoding_name = self.encodings[self.ENCODE_GZIP][0] self.set_header("Content-Encoding", encoding_name) f = SmartIO(force_mem=1) gz = gzip.GzipFile(mode="wb", compresslevel=COMPRESS_LEVEL, fileobj = f) if sys.version_info[0] == 3: gz.write(bstr(data)) else: gz.write(sstr(data)) gz.close() self.data = f.getvalue() f.close() elif self.encoding == self.ENCODE_ZLIB: import zlib encoding_name = self.encodings[self.ENCODE_ZLIB][0] self.set_header("Content-Encoding", encoding_name) obj = zlib.compressobj(COMPRESS_LEVEL) self.data = obj.compress(data.encode()) + obj.flush() elif self.encoding == self.ENCODE_GPG: # XXX: fix me. raise NotImplementedError(self.transfer, self.encoding) # Content-Transfer-Encoding header if self.transfer == self.TRANSFER_BINARY: transfer_name = self.transfers[self.TRANSFER_BINARY] self.set_header("Content-Transfer-Encoding", transfer_name) self.set_header("Content-Type", "application/binary") elif self.transfer == self.TRANSFER_BASE64: import base64 transfer_name = self.transfers[self.TRANSFER_BASE64] self.set_header("Content-Transfer-Encoding", transfer_name) self.set_header("Content-Type", "text/base64") self.data = base64.encodestring(self.data).decode() self.set_header("Content-Length", len(bstr(self.data))) rpc_version = __version__ if len(__version__.split()) > 1: rpc_version = __version__.split()[1] # other headers self.set_header("X-Transport-Info", 'Extended Capabilities Transport (C) Red Hat, Inc (version %s)' % rpc_version) self.__processed = 1 # reset the transport options def set_transport_flags(self, transfer=0, encoding=0, **kwargs): self.transfer = transfer self.encoding = encoding self.transport_flags.update(kwargs) def send_http(self, host, handler="/RPC2"): if not self.__processed: raise NotProcessed self._host = host if self._connection is None: raise Exception("No connection object found") self._connection.connect() # wrap self data into binary object, otherwise HTTPConnection.request # will encode it as ISO-8859-1 https://docs.python.org/3/library/http.client.html#httpconnection-objects self._connection.request(self.method, handler, body=bstr(self.data), headers=self.headers) response = self._connection.getresponse() if not self.response_acceptable(response): raise xmlrpclib.ProtocolError("%s %s" % (self._host, handler), response.status, response.reason, response.msg) # A response object has read() and close() methods, so we can safely # pass the whole object back return response.msg, response def response_acceptable(self, response): """Returns true if the response is acceptable""" if response.status == 200: return 1 if response.status in (301, 302): return 1 if response.status != 206: return 0 # If the flag is not set, it's unacceptable if not self.transport_flags.get('allow_partial_content'): return 0 if response.msg['Content-Type'] != 'application/octet-stream': # Don't allow anything else to be requested as a range, it could # break the XML parser return 0 return 1 def close(self): if self._connection: self._connection.close() self._connection = None def lookupTransfer(transfer, strict=0): """Given a string or numeric representation of a transfer, return the transfer code""" if transfer is None: # Plain return 0 if isinstance(transfer, IntType) and 0 <= transfer < len(Output.transfers): return transfer if isinstance(transfer, StringType): for i in range(len(Output.transfers)): if Output.transfers[i] == transfer.lower(): return i if strict: raise ValueError("Unsupported transfer %s" % transfer) # Return default return 0 def lookupEncoding(encoding, strict=0): """Given a string or numeric representation of an encoding, return the encoding code""" if encoding is None: # Plain return 0 if isinstance(encoding, IntType) and 0 <= encoding < len(Output.encodings): return encoding if isinstance(encoding, StringType): for i in range(len(Output.encodings)): if encoding.lower() in Output.encodings[i]: return i if strict: raise ValueError("Unsupported encoding %s" % encoding) # Return default return 0 Output = BaseOutput # File object class File: def __init__(self, file_obj, length = 0, name = None, progressCallback=None, bufferSize=16384): self.length = length self.file_obj = file_obj self.close = file_obj.close self.bufferSize=bufferSize self.name = "" if name: self.name = name[name.rfind("/")+1:] self.progressCallback = progressCallback def __len__(self): return self.length def read(self, amt=None): # If they want to read everything, use _smart_read if amt is None: fd = self._get_file() return fd.read() return self.file_obj.read(amt) def read_to_file(self, file): """Copies the contents of this File object into another file object""" fd = self._get_file() while 1: buf = fd.read(self.bufferSize) if not buf: break if sys.version_info[0] == 3: file.write(bstr(buf)) else: file.write(sstr(buf)) return file def _get_file(self): """Read everything into a temporary file and call the progress callbacks if the file length is defined, or just reads till EOF""" if self.length: io = _smart_read(self.file_obj, self.length, bufferSize=self.bufferSize, progressCallback=self.progressCallback) io.seek(0, 0) else: # Read everuthing - no callbacks involved io = _smart_total_read(self.file_obj, bufferSize=self.bufferSize) io.seek(0, 0) return io def __del__(self): if self.close: self.close() self.close = None 07070100000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000B00000000TRAILER!!!
Locations
Projects
Search
Status Monitor
Help
OpenBuildService.org
Documentation
API Documentation
Code of Conduct
Contact
Support
@OBShq
Terms
openSUSE Build Service is sponsored by
The Open Build Service is an
openSUSE project
.
Sign Up
Log In
Places
Places
All Projects
Status Monitor
