48
# Show warnings by default
49
if not sys.warnoptions:
51
warnings.simplefilter("default")
53
log = logging.getLogger(sys.argv[0])
54
logging.basicConfig(level="INFO", # Show info level messages
55
format="%(message)s") # Show basic log messages
57
logging.captureWarnings(True) # Show warnings via the logging system
46
59
if sys.version_info.major == 2:
103
116
datetime.timedelta(0, 60)
104
117
>>> rfc3339_duration_to_delta("PT60M")
105
118
datetime.timedelta(0, 3600)
119
>>> rfc3339_duration_to_delta("P60M")
120
datetime.timedelta(1680)
106
121
>>> rfc3339_duration_to_delta("PT24H")
107
122
datetime.timedelta(1)
108
123
>>> rfc3339_duration_to_delta("P1W")
111
126
datetime.timedelta(0, 330)
112
127
>>> rfc3339_duration_to_delta("P1DT3M20S")
113
128
datetime.timedelta(1, 200)
129
>>> # Can not be empty:
130
>>> rfc3339_duration_to_delta("")
131
Traceback (most recent call last):
133
ValueError: Invalid RFC 3339 duration: u''
134
>>> # Must start with "P":
135
>>> rfc3339_duration_to_delta("1D")
136
Traceback (most recent call last):
138
ValueError: Invalid RFC 3339 duration: u'1D'
139
>>> # Must use correct order
140
>>> rfc3339_duration_to_delta("PT1S2M")
141
Traceback (most recent call last):
143
ValueError: Invalid RFC 3339 duration: u'PT1S2M'
144
>>> # Time needs time marker
145
>>> rfc3339_duration_to_delta("P1H2S")
146
Traceback (most recent call last):
148
ValueError: Invalid RFC 3339 duration: u'P1H2S'
149
>>> # Weeks can not be combined with anything else
150
>>> rfc3339_duration_to_delta("P1D2W")
151
Traceback (most recent call last):
153
ValueError: Invalid RFC 3339 duration: u'P1D2W'
154
>>> rfc3339_duration_to_delta("P2W2H")
155
Traceback (most recent call last):
157
ValueError: Invalid RFC 3339 duration: u'P2W2H'
116
160
# Parsing an RFC 3339 duration with regular expressions is not
196
240
def string_to_delta(interval):
197
241
"""Parse a string and return a datetime.timedelta
199
>>> string_to_delta('7d')
200
datetime.timedelta(7)
201
>>> string_to_delta('60s')
202
datetime.timedelta(0, 60)
203
>>> string_to_delta('60m')
204
datetime.timedelta(0, 3600)
205
>>> string_to_delta('24h')
206
datetime.timedelta(1)
207
>>> string_to_delta('1w')
208
datetime.timedelta(7)
209
>>> string_to_delta('5m 30s')
210
datetime.timedelta(0, 330)
214
245
return rfc3339_duration_to_delta(interval)
246
except ValueError as e:
247
log.warning("%s - Parsing as pre-1.6.1 interval instead",
249
return parse_pre_1_6_1_interval(interval)
252
def parse_pre_1_6_1_interval(interval):
253
"""Parse an interval string as documented by Mandos before 1.6.1, and
254
return a datetime.timedelta
255
>>> parse_pre_1_6_1_interval('7d')
256
datetime.timedelta(7)
257
>>> parse_pre_1_6_1_interval('60s')
258
datetime.timedelta(0, 60)
259
>>> parse_pre_1_6_1_interval('60m')
260
datetime.timedelta(0, 3600)
261
>>> parse_pre_1_6_1_interval('24h')
262
datetime.timedelta(1)
263
>>> parse_pre_1_6_1_interval('1w')
264
datetime.timedelta(7)
265
>>> parse_pre_1_6_1_interval('5m 30s')
266
datetime.timedelta(0, 330)
267
>>> parse_pre_1_6_1_interval('')
268
datetime.timedelta(0)
269
>>> # Ignore unknown characters, allow any order and repetitions
270
>>> parse_pre_1_6_1_interval('2dxy7zz11y3m5m')
271
datetime.timedelta(2, 480, 18000)
218
275
value = datetime.timedelta(0)
219
276
regexp = re.compile(r"(\d+)([dsmhw]?)")
237
294
def print_clients(clients, keywords):
238
295
def valuetostring(value, keyword):
239
if type(value) is dbus.Boolean:
296
if isinstance(value, dbus.Boolean):
240
297
return "Yes" if value else "No"
241
298
if keyword in ("Timeout", "Interval", "ApprovalDelay",
242
299
"ApprovalDuration", "ExtendedTimeout"):
345
402
if options.all and not has_actions(options):
346
403
parser.error("--all requires an action.")
350
fail_count, test_count = doctest.testmod()
351
sys.exit(os.EX_OK if fail_count == 0 else 1)
354
406
bus = dbus.SystemBus()
355
407
mandos_dbus_objc = bus.get_object(busname, server_path)
356
408
except dbus.exceptions.DBusException:
357
print("Could not connect to Mandos server", file=sys.stderr)
409
log.critical("Could not connect to Mandos server")
360
412
mandos_serv = dbus.Interface(mandos_dbus_objc,
379
431
os.dup2(stderrcopy, sys.stderr.fileno())
380
432
os.close(stderrcopy)
381
433
except dbus.exceptions.DBusException as e:
382
print("Access denied: "
383
"Accessing mandos server through D-Bus: {}".format(e),
434
log.critical("Failed to access Mandos server through D-Bus:"
387
438
# Compile dict of (clients: properties) to process
398
449
clients[client_objc] = client
401
print("Client not found on server: {!r}"
402
.format(name), file=sys.stderr)
452
log.critical("Client not found on server: %r", name)
405
455
if not has_actions(options) and clients:
493
543
client.Approve(dbus.Boolean(False),
494
544
dbus_interface=client_interface)
547
class Test_milliseconds_to_string(unittest.TestCase):
549
self.assertEqual(milliseconds_to_string(93785000),
551
def test_no_days(self):
552
self.assertEqual(milliseconds_to_string(7385000), "02:03:05")
553
def test_all_zero(self):
554
self.assertEqual(milliseconds_to_string(0), "00:00:00")
555
def test_no_fractional_seconds(self):
556
self.assertEqual(milliseconds_to_string(400), "00:00:00")
557
self.assertEqual(milliseconds_to_string(900), "00:00:00")
558
self.assertEqual(milliseconds_to_string(1900), "00:00:01")
560
class Test_string_to_delta(unittest.TestCase):
561
def test_handles_basic_rfc3339(self):
562
self.assertEqual(string_to_delta("PT2H"),
563
datetime.timedelta(0, 7200))
564
def test_falls_back_to_pre_1_6_1_with_warning(self):
565
# assertLogs only exists in Python 3.4
566
if hasattr(self, "assertLogs"):
567
with self.assertLogs(log, logging.WARNING):
568
value = string_to_delta("2h")
570
value = string_to_delta("2h")
571
self.assertEqual(value, datetime.timedelta(0, 7200))
574
def should_only_run_tests():
575
parser = argparse.ArgumentParser(add_help=False)
576
parser.add_argument("--check", action='store_true')
577
args, unknown_args = parser.parse_known_args()
578
run_tests = args.check
580
# Remove --check argument from sys.argv
581
sys.argv[1:] = unknown_args
584
# Add all tests from doctest strings
585
def load_tests(loader, tests, none):
587
tests.addTests(doctest.DocTestSuite())
497
590
if __name__ == "__main__":
591
if should_only_run_tests():
592
# Call using ./tdd-python-script --check [--verbose]