• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

Gallopsled / pwntools / 1

28 Jan 2022 10:58AM UTC coverage: 0.0% (-73.8%) from 73.823%
1

push

github

web-flow
Fix CI after Groovy Gorilla went away for libc unstrip test (#2025)

* Fix CI after Groovy Gorilla went away for libc unstrip test

Build elfutils 0.181 from source since we can't use builds
from a newer ubuntu version anymore.

* Install python wheels in CI

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

13713 existing lines in 142 files now uncovered.

0 of 16559 relevant lines covered (0.0%)

0.0 hits per line

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

0.0
/pwnlib/timeout.py
1
# -*- coding: utf-8 -*-
2
"""
3
Timeout encapsulation, complete with countdowns and scope managers.
4
"""
UNCOV
5
from __future__ import division
×
6

UNCOV
7
import time
×
8

UNCOV
9
import pwnlib
×
10

11

UNCOV
12
class _DummyContextClass(object):
×
UNCOV
13
    def __enter__(self):   pass
×
UNCOV
14
    def __exit__(self,*a): pass
×
15

UNCOV
16
_DummyContext = _DummyContextClass()
×
17

UNCOV
18
class _countdown_handler(object):
×
UNCOV
19
    def __init__(self, obj, timeout):
×
UNCOV
20
        self.obj     = obj
×
UNCOV
21
        self.timeout = timeout
×
22

UNCOV
23
    def __enter__(self):
×
UNCOV
24
        self.old_timeout  = self.obj._timeout
×
UNCOV
25
        self.old_stop     = self.obj._stop
×
26

UNCOV
27
        self.obj._stop    = time.time() + self.timeout
×
28

UNCOV
29
        if self.old_stop:
×
UNCOV
30
            self.obj._stop = min(self.obj._stop, self.old_stop)
×
31

UNCOV
32
        self.obj._timeout = self.timeout
×
UNCOV
33
    def __exit__(self, *a):
×
UNCOV
34
        self.obj._timeout = self.old_timeout
×
UNCOV
35
        self.obj._stop    = self.old_stop
×
36

UNCOV
37
class _local_handler(object):
×
UNCOV
38
    def __init__(self, obj, timeout):
×
UNCOV
39
        self.obj     = obj
×
UNCOV
40
        self.timeout = timeout
×
UNCOV
41
    def __enter__(self):
×
UNCOV
42
        self.old_timeout  = self.obj._timeout
×
UNCOV
43
        self.old_stop     = self.obj._stop
×
44

UNCOV
45
        self.obj._stop    = 0
×
UNCOV
46
        self.obj._timeout = self.timeout # leverage validation
×
UNCOV
47
        self.obj.timeout_change()
×
48

UNCOV
49
    def __exit__(self, *a):
×
UNCOV
50
        self.obj._timeout = self.old_timeout
×
UNCOV
51
        self.obj._stop    = self.old_stop
×
UNCOV
52
        self.obj.timeout_change()
×
53

UNCOV
54
class TimeoutDefault(object):
×
UNCOV
55
    def __repr__(self): return "pwnlib.timeout.Timeout.default"
×
UNCOV
56
    def __str__(self): return "<default timeout>"
×
57

UNCOV
58
class Maximum(float):
×
UNCOV
59
    def __repr__(self):
×
UNCOV
60
        return 'pwnlib.timeout.maximum'
×
UNCOV
61
maximum = Maximum(2**20)
×
62

UNCOV
63
class Timeout(object):
×
64
    """
65
    Implements a basic class which has a timeout, and support for
66
    scoped timeout countdowns.
67

68
    Valid timeout values are:
69

70
    - ``Timeout.default`` use the global default value (``context.default``)
71
    - ``Timeout.forever`` or :const:`None` never time out
72
    - Any positive float, indicates timeouts in seconds
73

74
    Example:
75

76
        >>> context.timeout = 30
77
        >>> t = Timeout()
78
        >>> t.timeout == 30
79
        True
80
        >>> t = Timeout(5)
81
        >>> t.timeout == 5
82
        True
83
        >>> i = 0
84
        >>> with t.countdown():
85
        ...     print(4 <= t.timeout and t.timeout <= 5)
86
        ...
87
        True
88
        >>> with t.countdown(0.5): # doctest: +ELLIPSIS
89
        ...     while t.timeout:
90
        ...         print(round(t.timeout,1))
91
        ...         time.sleep(0.1)
92
        0.5
93
        0.4
94
        0.3
95
        0.2
96
        0.1
97
        >>> print(t.timeout)
98
        5.0
99
        >>> with t.local(0.5):# doctest: +ELLIPSIS
100
        ...     for i in range(5):
101
        ...         print(round(t.timeout,1))
102
        ...         time.sleep(0.1)
103
        0.5
104
        0.5
105
        0.5
106
        0.5
107
        ...
108
        >>> print(t.timeout)
109
        5.0
110
    """
111

112

113
    #: Value indicating that the timeout should not be changed
UNCOV
114
    default = TimeoutDefault()
×
115

116
    #: Value indicating that a timeout should not ever occur
UNCOV
117
    forever = None
×
118

119
    #: Maximum value for a timeout.  Used to get around platform issues
120
    #: with very large timeouts.
121
    #:
122
    #: OSX does not permit setting socket timeouts to 2**22.
123
    #: Assume that if we receive a timeout of 2**21 or greater,
124
    #: that the value is effectively infinite.
UNCOV
125
    maximum = maximum
×
126

UNCOV
127
    def __init__(self, timeout=default):
×
UNCOV
128
        self._stop    = 0
×
UNCOV
129
        self.timeout = self._get_timeout_seconds(timeout)
×
130

UNCOV
131
    @property
×
UNCOV
132
    def timeout(self):
×
133
        """
134
        Timeout for obj operations.  By default, uses ``context.timeout``.
135
        """
UNCOV
136
        timeout = self._timeout
×
UNCOV
137
        stop    = self._stop
×
138

UNCOV
139
        if not stop:
×
UNCOV
140
            return timeout
×
141

UNCOV
142
        return max(stop-time.time(), 0)
×
143

UNCOV
144
    @timeout.setter
×
UNCOV
145
    def timeout(self, value):
×
UNCOV
146
        assert not self._stop
×
UNCOV
147
        self._timeout = self._get_timeout_seconds(value)
×
UNCOV
148
        self.timeout_change()
×
149

UNCOV
150
    def _get_timeout_seconds(self, value):
×
UNCOV
151
        if value is self.default:
×
UNCOV
152
            value = pwnlib.context.context.timeout
×
153

UNCOV
154
        elif value is self.forever:
×
UNCOV
155
            value = self.maximum
×
156

157
        else:
UNCOV
158
            value = float(value)
×
159

UNCOV
160
            if value is value < 0:
×
UNCOV
161
                raise AttributeError("timeout: Timeout cannot be negative")
×
162

163
            if value > self.maximum:
×
UNCOV
164
                value = self.maximum
×
UNCOV
165
        return value
×
166

UNCOV
167
    def countdown_active(self):
×
UNCOV
168
        return (self._stop == 0) or (self._stop > time.time())
×
169

UNCOV
170
    def timeout_change(self):
×
171
        """
172
        Callback for subclasses to hook a timeout change.
173
        """
UNCOV
174
        pass
×
175

UNCOV
176
    def countdown(self, timeout = default):
×
177
        """
178
        Scoped timeout setter.  Sets the timeout within the scope,
179
        and restores it when leaving the scope.
180

181
        When accessing :attr:`timeout` within the scope, it will be
182
        calculated against the time when the scope was entered, in a
183
        countdown fashion.
184

185
        If :const:`None` is specified for ``timeout``, then the current
186
        timeout is used is made.  This allows :const:`None` to be specified
187
        as a default argument with less complexity.
188
        """
189
        # Don't count down from infinity
UNCOV
190
        if timeout is self.maximum:
×
UNCOV
191
            return _DummyContext
×
192

193
        if timeout is self.default and self.timeout is self.maximum:
×
UNCOV
194
            return _DummyContext
×
195

196
        if timeout is self.default:
×
UNCOV
197
            timeout = self._timeout
×
198

UNCOV
199
        return _countdown_handler(self, timeout)
×
200

UNCOV
201
    def local(self, timeout):
×
202
        """
203
        Scoped timeout setter.  Sets the timeout within the scope,
204
        and restores it when leaving the scope.
205
        """
UNCOV
206
        if timeout is self.default or timeout == self.timeout:
×
UNCOV
207
            return _DummyContext
×
208

UNCOV
209
        return _local_handler(self, timeout)
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc