14667842019

Committed 25 Apr 2025 03:14PM UTC coverage: 73.861% (-0.03%) from 73.889%

Build # 14667842019

Build Type

push

github

Committed by

tesuji

Commit Message

use log.exception inside except block

Run Details

3827 of 6452 branches covered (59.31%)

0 of 1 new or added line in 1 file covered. (0.0%)

5 existing lines in 2 files now uncovered.

13377 of 18111 relevant lines covered (73.86%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.31

/pwnlib/tubes/remote.py

from __future__ import absolute_import
from __future__ import division

import socket
import socks

from pwnlib.log import getLogger
from pwnlib.timeout import Timeout
from pwnlib.tubes.sock import sock

log = getLogger(__name__)

class remote(sock):
    r"""Creates a TCP or UDP-connection to a remote host. It supports
    both IPv4 and IPv6.

    The returned object supports all the methods from
    :class:`pwnlib.tubes.sock` and :class:`pwnlib.tubes.tube`.

    Arguments:
        host(str): The host to connect to.
        port(int): The port to connect to.
        fam: The string "any", "ipv4" or "ipv6" or an integer to pass to :func:`socket.getaddrinfo`.
        typ: The string "tcp" or "udp" or an integer to pass to :func:`socket.getaddrinfo`.
        timeout: A positive number, None or the string "default".
        sock(:class:`socket.socket`): Socket to inherit, rather than connecting
        ssl(bool): Wrap the socket with SSL
        ssl_context(ssl.SSLContext): Specify SSLContext used to wrap the socket.
        ssl_args(dict): Pass :func:`ssl.wrap_socket` named arguments in a dictionary.
        sni(str,bool): Set 'server_hostname' in ssl_args. Set to True to set it based on the host argument. Set to False to not provide any value. Default is True.

    Examples:

        >>> r = remote('google.com', 443, ssl=True)
        >>> r.send(b'GET /\r\n\r\n')
        >>> r.recvn(4)
        b'HTTP'

        If a connection cannot be made, an exception is raised.

        >>> r = remote('127.0.0.1', 1)
        Traceback (most recent call last):
        ...
        PwnlibException: Could not connect to 127.0.0.1 on port 1

        You can also use :meth:`.remote.fromsocket` to wrap an existing socket.

        >>> import socket
        >>> s = socket.socket()
        >>> s.connect(('google.com', 80))
        >>> s.send(b'GET /' + b'\r\n'*2)
        9
        >>> r = remote.fromsocket(s)
        >>> r.recvn(4)
        b'HTTP'
        >>> s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) #doctest: +SKIP
        >>> s.connect(('2606:4700:4700::1111', 80)) #doctest: +SKIP
        >>> s.send(b'GET ' + b'\r\n'*2) #doctest: +SKIP
        8
        >>> r = remote.fromsocket(s) #doctest: +SKIP
        >>> r.recvn(4) #doctest: +SKIP
        b'HTTP'
    """

    def __init__(self, host, port,
                 fam = "any", typ = "tcp",
                 sock=None, ssl=False, ssl_context=None, ssl_args=None, sni=True,
                 *args, **kwargs):
        super(remote, self).__init__(*args, **kwargs)

        # convert port to string for sagemath support
        self.rport  = str(port)
        self.rhost  = host

        if sock:
            self.family = sock.family
            self.type   = sock.type
            self.proto  = sock.proto
            self.sock   = sock

        else:
            typ = self._get_type(typ)
            fam = self._get_family(fam)
            try:
                self.sock   = self._connect(fam, typ)
            except socket.gaierror as e:
                if e.errno != socket.EAI_NONAME:
                    raise
                self.exception('Could not resolve hostname: %r', host)
        if self.sock:
            self.settimeout(self.timeout)
            self.lhost, self.lport = self.sock.getsockname()[:2]

            if ssl:
                # Deferred import to save startup time
                import ssl as _ssl

                ssl_args = ssl_args or {}
                if "server_hostname" in ssl_args and sni:
                    log.error("sni and server_hostname cannot be set at the same time")
                ssl_context = ssl_context or _ssl.SSLContext(_ssl.PROTOCOL_TLSv1_2)
                if isinstance(sni, str):
                    ssl_args["server_hostname"] = sni
                elif sni:
                    ssl_args["server_hostname"] = host
                self.sock = ssl_context.wrap_socket(self.sock,**ssl_args)

    def _connect(self, fam, typ):
        sock    = None
        timeout = self.timeout

        with self.waitfor('Opening connection to %s on port %s' % (self.rhost, self.rport)) as h:
            for res in socket.getaddrinfo(self.rhost, self.rport, fam, typ, 0, socket.AI_PASSIVE):
                self.family, self.type, self.proto, _canonname, sockaddr = res

                if self.type not in [socket.SOCK_STREAM, socket.SOCK_DGRAM]:
                    continue

                h.status("Trying %s", sockaddr[0])

                sock = socket.socket(self.family, self.type, self.proto)

                if timeout is not None and timeout <= 0:
                    sock.setblocking(0)
                else:
                    sock.setblocking(1)
                    sock.settimeout(timeout)

                try:
                    sock.connect(sockaddr)
                    return sock
                except socks.ProxyError:
                    raise
                except socket.error:
                    pass
            self.error("Could not connect to %s on port %s", self.rhost, self.rport)

    @classmethod
    def fromsocket(cls, socket):
        """
        Helper method to wrap a standard python socket.socket with the
        tube APIs.

        Arguments:
            socket: Instance of socket.socket

        Returns:
            Instance of pwnlib.tubes.remote.remote.
        """
        s = socket
        host, port = s.getpeername()[:2]
        return remote(host, port, fam=s.family, typ=s.type, sock=s)

class tcp(remote):
    __doc__ = remote.__doc__
    def __init__(self, host, port, *a, **kw):
        return super(tcp, self).__init__(host, port, typ="tcp", *a, **kw)

class udp(remote):
    __doc__ = remote.__doc__
    def __init__(self, host, port, *a, **kw):
        return super(udp, self).__init__(host, port, typ="udp", *a, **kw)

class connect(remote):
    __doc__ = remote.__doc__

1	from __future__ import absolute_import	1✔
2	from __future__ import division	1✔
3
4	import socket	1✔
5	import socks	1✔
6
7	from pwnlib.log import getLogger	1✔
8	from pwnlib.timeout import Timeout	1✔
9	from pwnlib.tubes.sock import sock	1✔
10
11	log = getLogger(__name__)	1✔
12
13	class remote(sock):	1✔
14	r"""Creates a TCP or UDP-connection to a remote host. It supports
15	both IPv4 and IPv6.
16
17	The returned object supports all the methods from
18	:class:`pwnlib.tubes.sock` and :class:`pwnlib.tubes.tube`.
19
20	Arguments:
21	host(str): The host to connect to.
22	port(int): The port to connect to.
23	fam: The string "any", "ipv4" or "ipv6" or an integer to pass to :func:`socket.getaddrinfo`.
24	typ: The string "tcp" or "udp" or an integer to pass to :func:`socket.getaddrinfo`.
25	timeout: A positive number, None or the string "default".
26	sock(:class:`socket.socket`): Socket to inherit, rather than connecting
27	ssl(bool): Wrap the socket with SSL
28	ssl_context(ssl.SSLContext): Specify SSLContext used to wrap the socket.
29	ssl_args(dict): Pass :func:`ssl.wrap_socket` named arguments in a dictionary.
30	sni(str,bool): Set 'server_hostname' in ssl_args. Set to True to set it based on the host argument. Set to False to not provide any value. Default is True.
31
32	Examples:
33
34	>>> r = remote('google.com', 443, ssl=True)
35	>>> r.send(b'GET /\r\n\r\n')
36	>>> r.recvn(4)
37	b'HTTP'
38
39	If a connection cannot be made, an exception is raised.
40
41	>>> r = remote('127.0.0.1', 1)
42	Traceback (most recent call last):
43	...
44	PwnlibException: Could not connect to 127.0.0.1 on port 1
45
46	You can also use :meth:`.remote.fromsocket` to wrap an existing socket.
47
48	>>> import socket
49	>>> s = socket.socket()
50	>>> s.connect(('google.com', 80))
51	>>> s.send(b'GET /' + b'\r\n'*2)
52	9
53	>>> r = remote.fromsocket(s)
54	>>> r.recvn(4)
55	b'HTTP'
56	>>> s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) #doctest: +SKIP
57	>>> s.connect(('2606:4700:4700::1111', 80)) #doctest: +SKIP
58	>>> s.send(b'GET ' + b'\r\n'*2) #doctest: +SKIP
59	8
60	>>> r = remote.fromsocket(s) #doctest: +SKIP
61	>>> r.recvn(4) #doctest: +SKIP
62	b'HTTP'
63	"""
64
65	def __init__(self, host, port,	1✔
66	fam = "any", typ = "tcp",
67	sock=None, ssl=False, ssl_context=None, ssl_args=None, sni=True,
68	args, *kwargs):
69	super(remote, self).__init__(args, *kwargs)	1✔
70
71	# convert port to string for sagemath support
72	self.rport = str(port)	1✔
73	self.rhost = host	1✔
74
75	if sock:	1✔
76	self.family = sock.family	1✔
77	self.type = sock.type	1✔
78	self.proto = sock.proto	1✔
79	self.sock = sock	1✔
80
81	else:
82	typ = self._get_type(typ)	1✔
83	fam = self._get_family(fam)	1✔
84	try:	1✔
85	self.sock = self._connect(fam, typ)	1✔
86	except socket.gaierror as e:	1✔
87	if e.errno != socket.EAI_NONAME:	×
88	raise	×
NEW 89	self.exception('Could not resolve hostname: %r', host)	×
90	if self.sock:	1!
91	self.settimeout(self.timeout)	1✔
92	self.lhost, self.lport = self.sock.getsockname()[:2]	1✔
93
94	if ssl:	1✔
95	# Deferred import to save startup time
96	import ssl as _ssl	1✔
97
98	ssl_args = ssl_args or {}	1✔
99	if "server_hostname" in ssl_args and sni:	1!
100	log.error("sni and server_hostname cannot be set at the same time")	×
101	ssl_context = ssl_context or _ssl.SSLContext(_ssl.PROTOCOL_TLSv1_2)	1✔
102	if isinstance(sni, str):	1!
103	ssl_args["server_hostname"] = sni	×
104	elif sni:	1!
105	ssl_args["server_hostname"] = host	1✔
106	self.sock = ssl_context.wrap_socket(self.sock,**ssl_args)	1✔
107
108	def _connect(self, fam, typ):	1✔
109	sock = None	1✔
110	timeout = self.timeout	1✔
111
112	with self.waitfor('Opening connection to %s on port %s' % (self.rhost, self.rport)) as h:	1✔
113	for res in socket.getaddrinfo(self.rhost, self.rport, fam, typ, 0, socket.AI_PASSIVE):	1✔
114	self.family, self.type, self.proto, _canonname, sockaddr = res	1✔
115
116	if self.type not in [socket.SOCK_STREAM, socket.SOCK_DGRAM]:	1!
117	continue	×
118
119	h.status("Trying %s", sockaddr[0])	1✔
120
121	sock = socket.socket(self.family, self.type, self.proto)	1✔
122
123	if timeout is not None and timeout <= 0:	1!
124	sock.setblocking(0)	×
125	else:
126	sock.setblocking(1)	1✔
127	sock.settimeout(timeout)	1✔
128
129	try:	1✔
130	sock.connect(sockaddr)	1✔
131	return sock	1✔
132	except socks.ProxyError:	1✔
133	raise	1✔
134	except socket.error:	1✔
135	pass	1✔
136	self.error("Could not connect to %s on port %s", self.rhost, self.rport)	1✔
137
138	@classmethod	1✔
139	def fromsocket(cls, socket):	1✔
140	"""
141	Helper method to wrap a standard python socket.socket with the
142	tube APIs.
143
144	Arguments:
145	socket: Instance of socket.socket
146
147	Returns:
148	Instance of pwnlib.tubes.remote.remote.
149	"""
150	s = socket	1✔
151	host, port = s.getpeername()[:2]	1✔
152	return remote(host, port, fam=s.family, typ=s.type, sock=s)	1✔
153
154	class tcp(remote):	1✔
155	__doc__ = remote.__doc__	1✔
156	def __init__(self, host, port, a, *kw):	1✔
157	return super(tcp, self).__init__(host, port, typ="tcp", a, *kw)	×
158
159	class udp(remote):	1✔
160	__doc__ = remote.__doc__	1✔
161	def __init__(self, host, port, a, *kw):	1✔
162	return super(udp, self).__init__(host, port, typ="udp", a, *kw)	×
163
164	class connect(remote):	1✔
165	__doc__ = remote.__doc__	1✔

Gallopsled / pwntools / 14667842019

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous